Find Slow Jaeger Services and Escalate the Latency Digest

id: jaeger-slow-span-digest
namespace: company.team
description: |
  Turn Jaeger trace data into a latency digest that escalates itself instead
  of waiting for someone to notice a slow service in the UI. A keyed webhook
  runs the digest on demand and a disabled hourly schedule holds the cadence
  slot. The flow pulls recent traces from the Jaeger query API, falls back to
  a bundled six span fixture when no collector answers, computes per service
  p50 and p99 latency from the span durations, and compares each p99 against
  a millisecond budget. Services over the budget are noted one by one,
  tracked in a GitHub issue and announced on Slack, while a clean window is
  only logged, and the verdict, slow services, worst p99 and trace count are
  published as outputs.

triggers:
  - id: on_demand_digest
    type: io.kestra.plugin.core.trigger.Webhook
    description: POST here to digest recent traces on demand, for example right
      after a deploy or an incident review.
    key: "{{ secret('JAEGER_DIGEST_WEBHOOK_KEY') }}"
  - id: hourly_digest
    type: io.kestra.plugin.core.trigger.Schedule
    description: Hourly latency digest net, shipped disabled so nothing pages until
      an operator reviews the threshold and switches it on.
    cron: "0 * * * *"
    disabled: true

inputs:
  - id: jaeger_base_url
    type: STRING
    defaults: https://jaeger.example.test:16686
    description: Base URL of the Jaeger query API, the fetch step appends
      /api/traces with the limit parameter.
  - id: p99_threshold_ms
    type: INT
    defaults: 1000
    description: Per service p99 latency budget in milliseconds, any service whose
      p99 exceeds this value counts as slow and is escalated.
  - id: trace_limit
    type: INT
    defaults: 25
    description: How many recent traces the Jaeger query API should return for the
      digest, the bundled fixture ignores this value.

tasks:
  # --- 1. trace workspace ---
  - id: trace_workspace
    type: io.kestra.plugin.core.flow.WorkingDirectory
    description: One shared directory so the trace fixture, the Jaeger response and
      the span parser all read the same files for the whole run.
    tasks:
      # --- 2. offline trace fixture ---
      - id: write_traces_fixture
        type: io.kestra.plugin.scripts.shell.Commands
        description: Write a deterministic traces.json fixture with a heredoc, three
          traces and six spans where four spans stay fast and two spans blow
          past the one second budget, so the digest stays reproducible offline.
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
          entryPoint: []
        containerImage: alpine:3.20
        outputFiles:
          - traces.json
        commands:
          - |
            set -u
            cat > traces.json <<'JSON'
            {
              "data": [
                {
                  "traceID": "aaa111bbb222",
                  "spans": [
                    {"spanID": "s1", "operationName": "GET /checkout", "process": {"serviceName": "checkout-api"}, "duration": 38000},
                    {"spanID": "s2", "operationName": "GET /cart", "process": {"serviceName": "checkout-api"}, "duration": 42000},
                    {"spanID": "s3", "operationName": "GET /search", "process": {"serviceName": "search-api"}, "duration": 29000}
                  ]
                },
                {
                  "traceID": "bbb222ccc333",
                  "spans": [
                    {"spanID": "s4", "operationName": "POST /checkout", "process": {"serviceName": "checkout-api"}, "duration": 55000}
                  ]
                },
                {
                  "traceID": "ccc333ddd444",
                  "spans": [
                    {"spanID": "s5", "operationName": "POST /payments/charge", "process": {"serviceName": "checkout-api"}, "duration": 3200000},
                    {"spanID": "s6", "operationName": "GET /search/reindex", "process": {"serviceName": "search-api"}, "duration": 2100000}
                  ]
                }
              ],
              "total": 3,
              "limit": 25,
              "offset": 0,
              "errors": null
            }
            JSON
            echo "wrote $(wc -c < traces.json) bytes of trace fixture"

      # --- 3. live trace fetch ---
      - id: fetch_traces
        type: io.kestra.plugin.core.http.Request
        description: GET the Jaeger trace search endpoint with the configured limit and
          absorb a non 2xx answer, so an unreachable collector hands the digest
          to the fixture instead of failing the run.
        method: GET
        uri: "{{ inputs.jaeger_base_url }}/api/traces?service=&limit={{
          inputs.trace_limit }}"
        headers:
          Accept: application/json
        options:
          allowFailed: true
          followRedirects: true
        timeout: PT30S

      # --- 4. latency verdict ---
      - id: analyze_traces
        type: io.kestra.plugin.scripts.python.Script
        description: Read the Jaeger response or the traces.json fixture, convert every
          span duration to milliseconds, compute per service p50 and p99, flag
          services whose p99 exceeds the budget, and emit the verdict through
          the outputs protocol.
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
        containerImage: python:3.12-slim
        env:
          JAEGER_BODY: "{{ (outputs.fetch_traces.body ?? '') | json }}"
          P99_THRESHOLD_MS: "{{ inputs.p99_threshold_ms }}"
        script: |
          import json
          import math
          import os

          FIXTURE = "traces.json"


          def as_int(value, fallback):
              try:
                  return int(str(value).strip())
              except (TypeError, ValueError):
                  return fallback


          def loads_payload(raw):
              text = str(raw or "").strip()
              if not text:
                  return None
              try:
                  parsed = json.loads(text)
              except ValueError:
                  return None
              if isinstance(parsed, str):
                  try:
                      parsed = json.loads(parsed)
                  except ValueError:
                      return None
              return parsed if isinstance(parsed, (dict, list)) else None


          def span_service(span):
              process = span.get("process")
              if isinstance(process, dict):
                  name = str(process.get("serviceName") or "").strip()
                  if name:
                      return name
              for key in ("serviceName", "service", "service_name"):
                  name = str(span.get(key) or "").strip()
                  if name:
                      return name
              return "unknown-service"


          def span_duration_ms(span):
              if "duration_ms" in span:
                  try:
                      return float(span["duration_ms"])
                  except (TypeError, ValueError):
                      return None
              try:
                  return float(span.get("duration")) / 1000.0
              except (TypeError, ValueError):
                  return None


          def extract(document):
              if isinstance(document, dict):
                  traces = document.get("data")
                  if not isinstance(traces, list):
                      traces = [document] if document.get("spans") else []
              elif isinstance(document, list):
                  traces = document
              else:
                  return 0, []
              spans = []
              for trace in traces:
                  if not isinstance(trace, dict):
                      continue
                  for span in trace.get("spans") or []:
                      if not isinstance(span, dict):
                          continue
                      duration = span_duration_ms(span)
                      if duration is None or duration < 0:
                          continue
                      spans.append((span_service(span), duration))
              return len(traces), spans


          def percentile(values, pct):
              ordered = sorted(values)
              if not ordered:
                  return 0.0
              rank = math.ceil(pct / 100.0 * len(ordered))
              rank = max(1, min(rank, len(ordered)))
              return float(ordered[rank - 1])


          threshold = as_int(os.environ.get("P99_THRESHOLD_MS"), 1000)
          if threshold < 1:
              threshold = 1

          document = loads_payload(os.environ.get("JAEGER_BODY"))
          traces_scanned, spans = extract(document)
          source = "jaeger_api" if spans else "fixture"

          if not spans:
              fixture_text = ""
              try:
                  with open(FIXTURE, "r", encoding="utf-8") as handle:
                      fixture_text = handle.read()
              except OSError:
                  pass
              traces_scanned, spans = extract(loads_payload(fixture_text))
              source = "fixture" if spans else "none"

          by_service = {}
          for service, duration in spans:
              by_service.setdefault(service, []).append(duration)

          per_service = []
          for service in sorted(by_service):
              durations = by_service[service]
              per_service.append({
                  "service": service,
                  "span_count": len(durations),
                  "p50_ms": round(percentile(durations, 50), 2),
                  "p99_ms": round(percentile(durations, 99), 2),
              })

          slow_services = sorted(
              [row for row in per_service if row["p99_ms"] > threshold],
              key=lambda row: -row["p99_ms"],
          )

          if not spans:
              verdict = "NO_TRACE_DATA"
              p99_ms = 0.0
          elif slow_services:
              verdict = "SLOW_SPANS"
              p99_ms = float(max(row["p99_ms"] for row in per_service))
          else:
              verdict = "WITHIN_BUDGET"
              p99_ms = float(max(row["p99_ms"] for row in per_service))

          print("source=%s traces=%d spans=%d slow=%d p99=%.2fms threshold=%dms verdict=%s" % (
              source, traces_scanned, len(spans), len(slow_services), p99_ms, threshold, verdict))
          print("::" + json.dumps({"outputs": {
              "verdict": verdict,
              "slow_services": slow_services,
              "p99_ms": round(p99_ms, 2),
              "traces_scanned": int(traces_scanned),
          }}) + "::")

  # --- 5. slow span gate ---
  - id: slow_span_gate
    type: io.kestra.plugin.core.flow.If
    description: Branch on whether any service blew the p99 budget, so slow services
      escalate while a healthy trace window is only recorded.
    condition: "{{ outputs.analyze_traces.vars.slow_services | length > 0 }}"
    then:
      # --- 6. per service notes ---
      - id: note_slow_services
        type: io.kestra.plugin.core.flow.Loop
        description: Walk the slow services one at a time and write one execution log
          line per service with its p50, p99 and span count, so the digest
          leaves a readable trail before the escalation lands.
        values: "{{ outputs.analyze_traces.vars.slow_services }}"
        concurrencyLimit: 1
        tasks:
          - id: note_slow_service
            type: io.kestra.plugin.core.log.Log
            description: Record the current slow service with its p50, p99 and span count
              against the configured budget.
            message: "Slow service {{ item.value.service }} - p99 {{ item.value.p99_ms }}ms
              over the {{ inputs.p99_threshold_ms }}ms budget, p50 {{
              item.value.p50_ms }}ms across {{ item.value.span_count }}
              span(s)."

      # --- 7. track the slow services ---
      - id: open_slow_span_issue
        type: io.kestra.plugin.github.issues.Create
        description: Open a labeled issue on the sre runbooks repository listing every
          slow service with its p50 and p99, so the latency work lands in the
          backlog instead of a chat scrollback.
        oauthToken: "{{ secret('GITHUB_TOKEN') }}"
        repository: your-org/sre-runbooks
        title: "Slow services in Jaeger digest - {{
          outputs.analyze_traces.vars.slow_services | length }} over {{
          inputs.p99_threshold_ms }}ms p99"
        body: |
          ## Jaeger slow span digest

          Verdict **{{ outputs.analyze_traces.vars.verdict }}** after scanning
          **{{ outputs.analyze_traces.vars.traces_scanned }}** trace(s) against a
          **{{ inputs.p99_threshold_ms }}ms** p99 budget. Worst service p99 is
          **{{ outputs.analyze_traces.vars.p99_ms }}ms**.

          ### Slow services

          {% for row in outputs.analyze_traces.vars.slow_services %}
          - `{{ row.service }}` - p99 {{ row.p99_ms }}ms, p50 {{ row.p50_ms }}ms, {{ row.span_count }} span(s)
          {% endfor %}

          Execution {{ execution.id }}

          Profile the listed services in Jaeger, confirm whether the slow window lines up
          with a deploy, then close this issue once a digest run reports WITHIN_BUDGET.
        labels:
          - jaeger
          - latency

      # --- 8. channel alert ---
      - id: alert_slow_spans
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Tell the ops channel how many services missed the p99 budget and
          which one is worst, right after the notes and the issue exist.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": "Jaeger digest {{ outputs.analyze_traces.vars.verdict }} - {{ outputs.analyze_traces.vars.slow_services | length }} service(s) over the {{ inputs.p99_threshold_ms }}ms p99 budget across {{ outputs.analyze_traces.vars.traces_scanned }} trace(s). Worst p99 {{ outputs.analyze_traces.vars.p99_ms }}ms. Issue opened in your-org/sre-runbooks. Execution {{ execution.id }}."
          }
    else:
      # --- 9. all clear ---
      - id: log_within_budget
        type: io.kestra.plugin.core.log.Log
        description: Record the healthy trace window so the execution history still
          proves which digest ran and what the worst p99 was.
        message: "Jaeger digest {{ outputs.analyze_traces.vars.verdict }} - worst p99 {{
          outputs.analyze_traces.vars.p99_ms }}ms inside the {{
          inputs.p99_threshold_ms }}ms budget across {{
          outputs.analyze_traces.vars.traces_scanned }} trace(s), no service
          escalated."

errors:
  - id: notify_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Escalate any failure to the ops channel with the execution url.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {"text": "FAILED: {{ flow.id }} {{ execution.id }}"}

outputs:
  - id: verdict
    type: STRING
    description: SLOW_SPANS when at least one service exceeded the p99 budget,
      WITHIN_BUDGET when every service stayed under it, otherwise NO_TRACE_DATA.
    value: "{{ outputs.analyze_traces.vars.verdict }}"
  - id: slow_services
    type: ARRAY
    description: 'Services whose p99 exceeded the budget as objects, for example
      [{"service": "checkout-api", "span_count": 4, "p50_ms": 42.0, "p99_ms":
      3200.0}], sorted worst p99 first.'
    value: "{{ outputs.analyze_traces.vars.slow_services | toJson }}"
  - id: p99_ms
    type: FLOAT
    description: Worst per service p99 latency in milliseconds, 3200.0 against the
      bundled fixture and 0.0 when no spans could be read.
    value: "{{ outputs.analyze_traces.vars.p99_ms }}"
  - id: traces_scanned
    type: INT
    description: Number of Jaeger traces the digest read from the API response or
      the fixture, 3 for the bundled six span fixture.
    value: "{{ outputs.analyze_traces.vars.traces_scanned }}"