OpenTelemetry GenAI & Trace Telemetry Pipeline to BigQuery via DuckDB

id: opentelemetry-lakehouse-analytics-bigquery
namespace: company.team
description: |
  Ingest OpenTelemetry GenAI trace spans, compact raw JSON batches into
  columnar Snappy Parquet using in-memory DuckDB, load into a day-partitioned
  and clustered BigQuery lakehouse table, and dispatch Slack alerts on token
  consumption anomalies or elevated error counts.

triggers:
  - id: batch_telemetry_webhook
    type: io.kestra.plugin.core.trigger.Webhook
    description: Webhook endpoint for event-driven invocation from OpenTelemetry
      collectors or telemetry forwarders.
    key: otel-trace-ingest

  - id: hourly_compaction_schedule
    type: io.kestra.plugin.core.trigger.Schedule
    description: Scheduled batch ingestion and compaction run hourly. Shipped
      disabled by default.
    cron: "0 * * * *"
    disabled: true

inputs:
  - id: gcs_bucket
    type: STRING
    displayName: Telemetry GCS Bucket
    description: Google Cloud Storage bucket containing raw OpenTelemetry JSON trace
      batches.
    defaults: company-telemetry-lake

  - id: gcs_prefix
    type: STRING
    displayName: GCS Trace Prefix
    description: Object prefix path for incoming raw OpenTelemetry trace files.
    defaults: traces/raw/

  - id: bq_project
    type: STRING
    displayName: BigQuery Project ID
    description: Google Cloud project ID hosting the destination analytics dataset.
    defaults: company-analytics

  - id: bq_dataset
    type: STRING
    displayName: BigQuery Dataset
    description: Destination BigQuery dataset for telemetry lakehouse tables.
    defaults: observability

  - id: bq_table
    type: STRING
    displayName: BigQuery Table
    description: Destination BigQuery table name for compacted OpenTelemetry GenAI spans.
    defaults: otel_genai_spans

  - id: token_anomaly_threshold
    type: INT
    displayName: Token Anomaly Threshold
    description: Threshold for alerting on excessive GenAI token consumption.
    defaults: 50000

  - id: send_slack_alert
    type: BOOLEAN
    displayName: Send Slack Alert
    description: Toggle to dispatch Slack notifications for detected anomalies or failures.
    defaults: false

tasks:
  - id: ingest_otel_traces
    type: io.kestra.plugin.scripts.python.Script
    description: Ingest or synthesize OpenTelemetry GenAI trace spans conforming to
      OpenTelemetry semantic conventions.
    taskRunner:
      type: io.kestra.plugin.core.runner.Process
    outputFiles:
      - otel_traces.json
    script: |
      import json
      import uuid
      from datetime import datetime, timezone
      import random

      models = [
          "gpt-4o",
          "claude-3-5-sonnet",
          "gemini-1.5-pro",
          "mistral-large"
      ]
      services = [
          "customer-support-agent",
          "code-review-assistant",
          "rag-search-pipeline",
          "document-summarizer"
      ]
      finish_reasons_list = [["stop"], ["stop"], ["stop"], ["length"]]

      now = datetime.now(timezone.utc)
      records = []

      # Generate 50 realistic trace spans matching OTel GenAI semantic conventions
      for i in range(50):
          model = random.choice(models)
          service = random.choice(services)

          # Inject an occasional high-token span or error to exercise anomaly thresholds
          if i == 7:
              input_tokens = 42000
              output_tokens = 12500
              status_code = "OK"
              error_msg = None
          elif i == 19:
              input_tokens = 1800
              output_tokens = 250
              status_code = "ERROR"
              error_msg = "Upstream model quota exceeded (HTTP 429)"
          else:
              input_tokens = random.randint(350, 4200)
              output_tokens = random.randint(80, 1100)
              status_code = "OK"
              error_msg = None

          duration_ms = float(random.randint(220, 3800))
          total_tokens = input_tokens + output_tokens

          span = {
              "trace_id": uuid.uuid4().hex,
              "span_id": uuid.uuid4().hex[:16],
              "timestamp": now.isoformat(),
              "service_name": service,
              "duration_ms": duration_ms,
              "status_code": status_code,
              "error_message": error_msg,
              "gen_ai": {
                  "system": model.split("-")[0] if "-" in model else "openai",
                  "request": {
                      "model": model
                  },
                  "usage": {
                      "input_tokens": input_tokens,
                      "output_tokens": output_tokens
                  },
                  "response": {
                      "finish_reasons": random.choice(finish_reasons_list)
                  }
              }
          }
          records.append(span)

      with open("otel_traces.json", "w") as f:
          for r in records:
              f.write(json.dumps(r) + "\n")

      max_tokens = max(
          r["gen_ai"]["usage"]["input_tokens"] + r["gen_ai"]["usage"]["output_tokens"]
          for r in records
      )
      error_count = sum(1 for r in records if r["status_code"] != "OK")
      threshold = {{ inputs.token_anomaly_threshold }}

      print("::" + json.dumps({
          "outputs": {
              "total_spans": len(records),
              "max_tokens": max_tokens,
              "error_count": error_count,
              "token_anomaly_detected": max_tokens >= threshold,
              "high_error_count_detected": error_count > 0
          }
      }) + "::")

  - id: compact_traces_to_parquet
    type: io.kestra.plugin.duckdb.Query
    description: In-memory compaction and flattening from raw OTel JSON traces to
      Snappy Parquet.
    inputFiles:
      otel_traces.json: "{{ outputs.ingest_otel_traces.outputFiles['otel_traces.json'] }}"
    outputFiles:
      - otel_genai_spans.parquet
    sql: |
      COPY (
        SELECT
          trace_id::VARCHAR AS trace_id,
          span_id::VARCHAR AS span_id,
          timestamp::TIMESTAMP AS timestamp,
          service_name::VARCHAR AS service_name,
          gen_ai.system::VARCHAR AS gen_ai_system,
          gen_ai.request.model::VARCHAR AS model,
          gen_ai.usage.input_tokens::BIGINT AS input_tokens,
          gen_ai.usage.output_tokens::BIGINT AS output_tokens,
          (gen_ai.usage.input_tokens + gen_ai.usage.output_tokens)::BIGINT AS total_tokens,
          CAST(gen_ai.response.finish_reasons AS VARCHAR) AS finish_reasons,
          duration_ms::DOUBLE AS duration_ms,
          status_code::VARCHAR AS status_code,
          error_message::VARCHAR AS error_message
        FROM read_json_auto('{{ workingDir }}/otel_traces.json')
      ) TO '{{ workingDir }}/otel_genai_spans.parquet' (FORMAT PARQUET, CODEC 'SNAPPY');

  - id: load_parquet_to_bigquery
    type: io.kestra.plugin.gcp.bigquery.Load
    description: Load compacted Snappy Parquet into BigQuery with day-partitioning
      on timestamp and clustering on service_name and model.
    projectId: "{{ inputs.bq_project }}"
    serviceAccount: "{{ secret('GCP_SERVICE_ACCOUNT_JSON') }}"
    destinationTable: "{{ inputs.bq_project }}.{{ inputs.bq_dataset }}.{{ inputs.bq_table }}"
    from: "{{
      outputs.compact_traces_to_parquet.outputFiles['otel_genai_spans.parquet']
      }}"
    format: PARQUET
    timePartitioningField: "timestamp"
    timePartitioningType: "DAY"
    clusteringFields:
      - "service_name"
      - "model"
    autodetect: true
    createDisposition: CREATE_IF_NEEDED
    writeDisposition: WRITE_APPEND

  - id: evaluate_anomalies
    type: io.kestra.plugin.core.flow.If
    description: Check if token consumption exceeded the anomaly threshold or high
      error count was detected.
    condition: "{{ outputs.ingest_otel_traces.token_anomaly_detected or
      outputs.ingest_otel_traces.high_error_count_detected }}"
    then:
      - id: notify_slack_alert
        type: io.kestra.plugin.notifications.slack.SlackIncomingWebhook
        description: Dispatch Slack notification with markdown block layout detailing
          the anomaly.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        runIf: "{{ inputs.send_slack_alert }}"
        payload: |
          {
            "text": "ALERT: OpenTelemetry GenAI span anomaly detected in execution {{ execution.id }}",
            "blocks": [
              {
                "type": "header",
                "text": {
                  "type": "plain_text",
                  "text": "OpenTelemetry GenAI Anomaly Alert",
                  "emoji": false
                }
              },
              {
                "type": "section",
                "fields": [
                  {
                    "type": "mrkdwn",
                    "text": "*Flow ID:*\n`{{ flow.id }}`"
                  },
                  {
                    "type": "mrkdwn",
                    "text": "*Execution ID:*\n`{{ execution.id }}`"
                  },
                  {
                    "type": "mrkdwn",
                    "text": "*Total Spans Processed:*\n{{ outputs.ingest_otel_traces.total_spans }}"
                  },
                  {
                    "type": "mrkdwn",
                    "text": "*Peak Span Tokens:*\n{{ outputs.ingest_otel_traces.max_tokens }} (Threshold: {{ inputs.token_anomaly_threshold }})"
                  },
                  {
                    "type": "mrkdwn",
                    "text": "*Error Spans Count:*\n{{ outputs.ingest_otel_traces.error_count }}"
                  },
                  {
                    "type": "mrkdwn",
                    "text": "*Destination Table:*\n`{{ inputs.bq_project }}.{{ inputs.bq_dataset }}.{{ inputs.bq_table }}`"
                  }
                ]
              },
              {
                "type": "section",
                "text": {
                  "type": "mrkdwn",
                  "text": "*Diagnostic Assessment:*\nOne or more OpenTelemetry GenAI spans exceeded the consumption threshold of {{ inputs.token_anomaly_threshold }} tokens or recorded non-OK status codes during trace batch processing. Compacted Parquet traces have been loaded into BigQuery for forensic analytics."
                }
              }
            ]
          }

errors:
  - id: log_pipeline_failure
    type: io.kestra.plugin.core.log.Log
    description: Record pipeline failure context and execution metadata.
    message: "Pipeline execution {{ execution.id }} failed in flow {{ flow.id }}
      under namespace {{ flow.namespace }}."

  - id: alert_pipeline_failure
    type: io.kestra.plugin.notifications.slack.SlackIncomingWebhook
    description: Dispatch critical alert to Slack when pipeline fails.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    runIf: "{{ inputs.send_slack_alert }}"
    payload: |
      {
        "text": "CRITICAL: OpenTelemetry Lakehouse Pipeline failed in flow {{ flow.id }} (execution {{ execution.id }}). Please inspect task logs in Kestra console."
      }