Apache Doris Tablet Health and Backend Node Sentinel

id: apache-doris-tablet-health-and-backend-sentinel
namespace: company.data

inputs:
  - id: doris_fe_endpoint
    type: STRING
    displayName: Apache Doris Frontend HTTP Endpoint
    description: Base URL of the Apache Doris Frontend (FE) HTTP and Prometheus
      metrics endpoint.
    defaults: "http://localhost:8030"

  - id: max_allowed_unhealthy_tablets
    type: INT
    displayName: Maximum Allowed Unhealthy Tablets
    description: Threshold of unhealthy or under-replicated tablets allowed before
      raising a critical incident alert.
    defaults: 0

  - id: send_slack_alert
    type: BOOL
    displayName: Dispatch Slack Incident Alert
    description: Whether to transmit real-time Slack notifications upon tablet
      corruption or offline backend nodes.
    defaults: true

triggers:
  - id: doris_audit_schedule
    type: io.kestra.plugin.core.trigger.Schedule
    description: Recurring audit polling Apache Doris cluster health every 15 minutes.
    cron: "*/15 * * * *"
    disabled: true

tasks:
  - id: probe_doris_metrics
    type: io.kestra.plugin.core.http.Request
    description: Query Apache Doris Frontend Prometheus metrics endpoint for tablet
      and backend health counters.
    uri: "{{ inputs.doris_fe_endpoint }}/metrics"
    method: GET
    headers:
      Accept: "text/plain"
    options:
      allowFailed: true
      timeout:
        connectTimeout: PT10S
        readIdleTimeout: PT30S
    retry:
      type: constant
      interval: PT5S
      maxAttempts: 3
    allowFailure: true

  - id: evaluate_cluster_health
    type: io.kestra.plugin.scripts.python.Script
    description: Interrogate Doris Prometheus metrics text, extract unhealthy
      tablets and offline backends, and classify status.
    taskRunner:
      type: io.kestra.plugin.core.runner.Process
    inputFiles:
      metrics.txt: "{{ outputs.probe_doris_metrics.body }}"
    script: |
      import re
      import json

      max_unhealthy = int({{ inputs.max_allowed_unhealthy_tablets }})

      try:
          with open("metrics.txt", "r") as f:
              raw = f.read()
      except Exception:
          raw = ""

      def extract_metric(pattern):
          matches = re.findall(pattern, raw)
          total = 0.0
          for m in matches:
              try:
                  total += float(m)
              except ValueError:
                  pass
          return total

      unhealthy_tablets = extract_metric(r'doris_fe_tablet_unhealthy_count(?:\{[^}]*\})?\s+([0-9\.]+)')
      backends_down = extract_metric(r'doris_fe_backends_down_count(?:\{[^}]*\})?\s+([0-9\.]+)')
      active_connections = extract_metric(r'doris_fe_connection_total(?:\{[^}]*\})?\s+([0-9\.]+)')

      if unhealthy_tablets == 0 and backends_down == 0:
          classification = "CLUSTER_OPTIMAL"
      elif unhealthy_tablets <= max_unhealthy and backends_down == 0:
          classification = "BACKEND_STRESS_WARNING"
      else:
          classification = "CRITICAL_TABLET_CORRUPTION"

      output_payload = {
          "outputs": {
              "unhealthy_tablets_count": int(unhealthy_tablets),
              "backends_down_count": int(backends_down),
              "active_connections": int(active_connections),
              "classification": classification
          }
      }
      print(f"::{json.dumps(output_payload)}::")

  - id: classify_cluster_status
    type: io.kestra.plugin.core.log.Log
    description: Log evaluated Apache Doris cluster health metrics and status
      classification.
    message: |
      Doris FE Endpoint: {{ inputs.doris_fe_endpoint }}
      Unhealthy Tablets: {{ outputs.evaluate_cluster_health.vars.unhealthy_tablets_count }}
      Backends Down: {{ outputs.evaluate_cluster_health.vars.backends_down_count }}
      Active Connections: {{ outputs.evaluate_cluster_health.vars.active_connections }}
      Classification: {{ outputs.evaluate_cluster_health.vars.classification }}

  - id: route_cluster_gate
    type: io.kestra.plugin.core.flow.Switch
    description: Route flow execution based on Apache Doris cluster health classification.
    value: "{{ outputs.evaluate_cluster_health.vars.classification }}"
    cases:
      CLUSTER_OPTIMAL:
        - id: log_optimal_state
          type: io.kestra.plugin.core.log.Log
          description: Log nominal verification confirming all Doris tablets and backends
            are healthy.
          message: "OPTIMAL: Apache Doris cluster reports zero unhealthy tablets and zero
            offline backends."

      BACKEND_STRESS_WARNING:
        - id: log_stress_advisory
          type: io.kestra.plugin.core.log.Log
          description: Log advisory warning when tablet replication stress falls within
            allowable limits.
          message: "WARNING: Elevated tablet replication latency detected on Doris
            cluster. Unhealthy tablets: {{
            outputs.evaluate_cluster_health.vars.unhealthy_tablets_count }}."

      CRITICAL_TABLET_CORRUPTION:
        - id: alert_slack_tablet_corruption
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          description: Dispatch urgent incident alert to data platform engineering channel.
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {
              "text": ":rotating_light: *CRITICAL Apache Doris Data Warehouse Tablet Alert*\n*Endpoint:* `{{ inputs.doris_fe_endpoint }}`\n*Unhealthy Tablets:* `{{ outputs.evaluate_cluster_health.vars.unhealthy_tablets_count }}`\n*Offline Backends:* `{{ outputs.evaluate_cluster_health.vars.backends_down_count }}`\n*Active Connections:* `{{ outputs.evaluate_cluster_health.vars.active_connections }}`\n*Execution:* <{{ flow.url }}|{{ execution.id }}>"
            }

        - id: remediation_fail_gate
          type: io.kestra.plugin.core.execution.Fail
          description: Fail execution to halt downstream ELT pipelines and prevent corrupt
            reporting consumption.

  - id: final_audit_log
    type: io.kestra.plugin.core.log.Log
    description: Record immutable audit log entry documenting Apache Doris cluster
      health verification.
    message: |
      [AUDIT] Apache Doris Tablet Health and Backend Sentinel completed.
      Endpoint: {{ inputs.doris_fe_endpoint }}
      Unhealthy Tablets: {{ outputs.evaluate_cluster_health.vars.unhealthy_tablets_count }}
      Backends Down: {{ outputs.evaluate_cluster_health.vars.backends_down_count }}
      Verdict: {{ outputs.evaluate_cluster_health.vars.classification }}