OpenObserve Log Stream and Error Surge Sentinel

id: openobserve-log-stream-and-error-surge-sentinel
namespace: company.sre

inputs:
  - id: openobserve_url
    type: STRING
    defaults: http://localhost:5080
    description: Base URL of the OpenObserve instance.

  - id: organization_id
    type: STRING
    defaults: default
    description: Target OpenObserve organization identifier.

  - id: stream_name
    type: STRING
    defaults: default
    description: Target log stream to inspect for error surges.

  - id: error_threshold
    type: INT
    defaults: 25
    description: Maximum allowable error log entries in the 15-minute evaluation
      window before triggering an alert.

  - id: slack_channel
    type: STRING
    defaults: "#ops-alerts"
    description: Destination Slack channel for operational warnings and critical outages.

tasks:
  - id: probe_openobserve_health
    type: io.kestra.plugin.core.http.Request
    method: GET
    uri: "{{ inputs.openobserve_url }}/healthz"
    description: Validates OpenObserve cluster node availability and liveness status.

  - id: query_error_stream
    type: io.kestra.plugin.core.http.Request
    method: POST
    uri: "{{ inputs.openobserve_url }}/api/{{ inputs.organization_id }}/{{
      inputs.stream_name }}/_search"
    headers:
      Authorization: "Basic {{ secret('OPENOBSERVE_AUTH_TOKEN') }}"
      Content-Type: application/json
    contentType: application/json
    body: |
      {
        "query": {
          "sql": "SELECT COUNT(*) as error_count FROM {{ inputs.stream_name }} WHERE log_level = 'error' AND _timestamp > NOW() - INTERVAL 15 MINUTE"
        }
      }
    description: Executes a SQL query against OpenObserve to measure recent error
      log frequency.

  - id: evaluate_ingestion_health
    type: io.kestra.plugin.core.flow.Switch
    value: "{{ (outputs.probe_openobserve_health.code != 200) ? 'CRITICAL' :
      ((outputs.query_error_stream.body.total ?? 0) > inputs.error_threshold ?
      'SURGE' : 'HEALTHY') }}"
    description: Evaluates HTTP probe status and error counts against operational
      thresholds.
    cases:
      CRITICAL:
        - id: alert_slack_critical
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {
              "channel": "{{ inputs.slack_channel }}",
              "text": "[CRITICAL] OpenObserve cluster node health probe failed on endpoint {{ inputs.openobserve_url }}. HTTP status code: {{ outputs.probe_openobserve_health.code }}."
            }
          description: Dispatches high-severity outage notification to Slack when node
            probe fails.

      SURGE:
        - id: alert_slack_surge
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {
              "channel": "{{ inputs.slack_channel }}",
              "text": "[WARNING] OpenObserve error surge detected on stream '{{ inputs.stream_name }}'. Observed errors in last 15 minutes: {{ outputs.query_error_stream.body.total ?? 0 }} (Threshold: {{ inputs.error_threshold }})."
            }
          description: Sends warning notification to Slack when error volume exceeds
            threshold.

      HEALTHY:
        - id: log_healthy_state
          type: io.kestra.plugin.core.log.Log
          message: "OpenObserve instance is operational. Stream '{{ inputs.stream_name }}'
            error volume is within nominal limits."
          description: Emits normal status confirmation when checks pass.

  - id: audit_sentinel_run
    type: io.kestra.plugin.core.log.Log
    message: "OpenObserve Sentinel complete. Probe Status: {{
      outputs.probe_openobserve_health.code }}, Stream: {{ inputs.stream_name
      }}, Errors: {{ outputs.query_error_stream.body.total ?? 0 }}, Assessment:
      {{ (outputs.probe_openobserve_health.code != 200) ? 'CRITICAL' :
      ((outputs.query_error_stream.body.total ?? 0) > inputs.error_threshold ?
      'SURGE' : 'HEALTHY') }}."
    description: Emits an immutable structured audit log entry for historical
      telemetry and compliance.

triggers:
  - id: hourly_sentinel_schedule
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 * * * *"
    description: Executes periodic continuous monitoring probe on an hourly cadence.