Automated API SLA and Data Drift Quarantine Pipeline

id: api-sla-drift-quarantine-pipeline
namespace: company.platform

inputs:
  - id: target_service
    type: STRING
    defaults: "orders-ingestion-v2"
  - id: observed_p95_ms
    type: INT
    defaults: 340
  - id: max_latency_ms
    type: INT
    defaults: 450

triggers:
  - id: fifteen_min_sla_audit
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "*/15 * * * *"

  - id: upstream_deploy_webhook
    type: io.kestra.plugin.core.trigger.Webhook
    key: "deploy-drift-audit-hook"

tasks:
  - id: init_audit_context
    type: io.kestra.plugin.core.log.Log
    level: INFO
    message: "1/12 Starting SLA & schema drift audit for {{ inputs.target_service }}"

  - id: parallel_preflight_probes
    type: io.kestra.plugin.core.flow.Parallel
    description: "Flowable 1: Run concurrent endpoint and regional health checks"
    tasks:
      - id: probe_primary_gateway
        type: io.kestra.plugin.core.http.Request
        uri: "https://jsonplaceholder.typicode.com/todos/1"
        method: GET

      - id: probe_replica_gateway
        type: io.kestra.plugin.core.http.Request
        uri: "https://jsonplaceholder.typicode.com/todos/2"
        method: GET

      - id: log_probe_heartbeat
        type: io.kestra.plugin.core.log.Log
        level: INFO
        message: "Pre-flight network probes dispatched for {{ inputs.target_service }}"

  - id: query_staging_partitions
    type: io.kestra.plugin.jdbc.duckdb.Query
    description: "Extract active ingestion micro-batches using embedded DuckDB SQL"
    sql: |
      SELECT 'batch_101' AS batch_id, 1250 AS rows_ingested, 'VALID' AS schema_state
      UNION ALL
      SELECT 'batch_102' AS batch_id, 1420 AS rows_ingested, 'VALID' AS schema_state
      UNION ALL
      SELECT 'batch_103' AS batch_id, 980  AS rows_ingested, 'VALID' AS schema_state;
    fetchType: FETCH

  - id: evaluate_drift_and_sla
    type: io.kestra.plugin.scripts.python.Script
    description: "Compare batch telemetry against SLA thresholds and build
      remediation manifest"
    beforeCommands:
      - pip install kestra
    script: |
      from kestra import Kestra

      batch_count = {{ outputs.query_staging_partitions.size }}
      peak_latency = {{ inputs.observed_p95_ms }}
      max_latency = {{ inputs.max_latency_ms }}

      is_healthy = peak_latency <= max_latency
      status_label = "HEALTHY_COMMIT" if is_healthy else "QUARANTINE_REQUIRED"
      summary = f"Verified {batch_count} batches (p95: {peak_latency}ms / max: {max_latency}ms)"

      print(f"[{status_label}] {summary}")

      Kestra.outputs({
          'is_healthy': is_healthy,
          'status_label': status_label,
          'summary': summary,
          'batch_count': batch_count
      })

  - id: sync_downstream_consumers
    type: io.kestra.plugin.core.flow.Sequential
    description: "Flowable 2: Sequentially update watermarks across dependent
      consumer services"
    tasks:
      - id: sync_billing_ledger
        type: io.kestra.plugin.core.log.Log
        level: INFO
        message: "Synced watermark for [billing-ledger-worker] -> {{
          outputs.evaluate_drift_and_sla.vars.status_label }}"

      - id: sync_fraud_engine
        type: io.kestra.plugin.core.log.Log
        level: INFO
        message: "Synced watermark for [fraud-scoring-engine] -> {{
          outputs.evaluate_drift_and_sla.vars.status_label }}"

      - id: sync_warehouse_loader
        type: io.kestra.plugin.core.log.Log
        level: INFO
        message: "Synced watermark for [warehouse-hourly-sync] -> {{
          outputs.evaluate_drift_and_sla.vars.status_label }}"

  - id: route_batch_disposition
    type: io.kestra.plugin.core.flow.If
    description: "Flowable 3: Conditional promotion vs quarantine"
    condition: "{{ outputs.evaluate_drift_and_sla.vars.is_healthy == true }}"
    then:
      - id: promote_staging_partition
        type: io.kestra.plugin.core.log.Log
        level: INFO
        message: "Promoting {{ outputs.evaluate_drift_and_sla.vars.batch_count }}
          validated micro-batches for {{ inputs.target_service }} to
          production."

      - id: record_sla_compliance
        type: io.kestra.plugin.core.log.Log
        level: INFO
        message: "[SLA VERIFIED] {{ outputs.evaluate_drift_and_sla.vars.summary }}"
    else:
      - id: isolate_corrupt_partition
        type: io.kestra.plugin.core.log.Log
        level: WARN
        message: "[QUARANTINE TRIGGERED] Holding back staging partition for {{
          inputs.target_service }}."

  - id: finalize_audit_ledger
    type: io.kestra.plugin.core.log.Log
    level: INFO
    message: "Completed 12-stage audit run {{ execution.id }} with status: {{
      outputs.evaluate_drift_and_sla.vars.status_label }}"