Quickwit Error Surge and Incident Triage Sentinel

id: quickwit-error-surge-and-incident-triage-sentinel
namespace: company.platform

inputs:
  - id: quickwit_url
    type: STRING
    displayName: Quickwit Cluster URL
    description: Base HTTP URL of the Quickwit cluster (e.g., http://quickwit:7280).
    defaults: "http://localhost:7280"

  - id: index_id
    type: STRING
    displayName: Log Index ID
    description: Target Quickwit index storing application logs and distributed traces.
    defaults: "app-logs"

  - id: surge_threshold
    type: INT
    displayName: Error Surge Threshold
    description: Delivered error count in a single polling interval required to
      classify the burst as CRITICAL_SURGE.
    defaults: 10

  - id: enforce_incident_gate
    type: BOOL
    displayName: Enforce Incident Gate
    description: When true, fails the execution on critical error surge to trigger
      on-call alert escalation and SLA tracking.
    defaults: true

triggers:
  - id: on_error_log_burst
    type: io.kestra.plugin.quickwit.search.Trigger
    url: "{{ secret('QUICKWIT_URL') }}"
    index: "app-logs"
    query: "severity:ERROR OR status:[500 TO 599]"
    timestampField: "timestamp"
    interval: "PT1M"

tasks:
  - id: sample_and_aggregate_errors
    type: io.kestra.plugin.core.output.OutputValues
    description: Extracts error event volume, sample messages, and impacted service
      metadata from the delivered trigger batch.
    values:
      error_count: "{{ trigger.numHits is defined ? trigger.numHits :
        (trigger.documents is defined ? (trigger.documents | length) : 1) }}"
      elapsed_time_micros: "{{ trigger.elapsedTimeMicros ?? 0 }}"
      watermark: "{{ trigger.watermark ?? 0 }}"
      sample_error_message: "{{ trigger.documents is defined and (trigger.documents |
        length) > 0 ? (trigger.documents[0].message ?? 'Unspecified error
        payload') : 'No payload details' }}"

  - id: query_cluster_context
    type: io.kestra.plugin.quickwit.search.Search
    description: Queries the Quickwit cluster for baseline error volume over the
      past 15 minutes to calculate surge magnitude.
    url: "{{ secret('QUICKWIT_URL') }}"
    index: "{{ inputs.index_id }}"
    query: "severity:ERROR"
    startTimestamp: "{{ now() | dateAdd(-15, 'MINUTES') | timestamp }}"
    maxHits: 100
    fetchType: FETCH
    allowFailure: true

  - id: evaluate_surge_severity
    type: io.kestra.plugin.core.output.OutputValues
    description: Evaluates error volume against surge thresholds to classify
      incident severity.
    values:
      baseline_15m_errors: "{{ outputs.query_cluster_context.size ?? 0 }}"
      severity: "{{ (outputs.sample_and_aggregate_errors.values.error_count | int) >=
        inputs.surge_threshold ? 'CRITICAL_SURGE' : 'WARNING_ANOMALY' }}"

  - id: route_incident_triage
    type: io.kestra.plugin.core.flow.If
    description: Branches on severity classification to dispatch high-priority
      incident alerts or record audit log.
    condition: "{{ outputs.evaluate_surge_severity.values.severity == 'CRITICAL_SURGE' }}"
    then:
      - id: alert_error_surge
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Alerts platform SREs and on-call engineers of an active critical
          error burst.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "attachments": [
              {
                "color": "#E53935",
                "title": "[ALERT] Quickwit Log Error Surge Detected",
                "text": "High-volume application error surge detected in Quickwit index `{{ inputs.index_id }}`.\n*Recent Errors in Window:* {{ outputs.sample_and_aggregate_errors.values.error_count }}\n*Surge Threshold:* {{ inputs.surge_threshold }}\n*15-Minute Baseline:* {{ outputs.evaluate_surge_severity.values.baseline_15m_errors }}\n*Severity:* `{{ outputs.evaluate_surge_severity.values.severity }}`\n*Sample Message:* `{{ outputs.sample_and_aggregate_errors.values.sample_error_message }}`\n*Watermark:* {{ outputs.sample_and_aggregate_errors.values.watermark }}\nInvestigate upstream service deployments, database connection pools, and downstream dependency health.",
                "footer": "Kestra Sentinel | Quickwit Incident Sentinel"
              }
            ]
          }

      - id: gate_enforcement
        type: io.kestra.plugin.core.flow.If
        description: Fails the execution when enforce_incident_gate is enabled to update
          on-call SLA trackers.
        condition: "{{ inputs.enforce_incident_gate }}"
        then:
          - id: fail_on_surge
            type: io.kestra.plugin.core.execution.Fail
            description: Marks execution failed to record incident breach in deployment
              metrics.

    else:
      - id: log_moderate_anomaly
        type: io.kestra.plugin.core.log.Log
        description: Records an operational audit log for sub-critical error anomalies.
        message: "[INFO] Quickwit error anomaly evaluated. Delivered errors: {{
          outputs.sample_and_aggregate_errors.values.error_count }}, 15m
          baseline: {{
          outputs.evaluate_surge_severity.values.baseline_15m_errors }}.
          Sub-threshold anomaly; no escalation required."