Self-heal Kafka Connect connectors with a restart budget and escalation

id: kafka-connect-self-healing-monitor
namespace: company.team

variables:
  connect_url: http://connect:8083
  connector_name: orders_jdbc_sink
  max_restarts: 3
  flap_window: PT30M

concurrency:
  limit: 1
  behavior: CANCEL

triggers:
  - id: connector_failed
    type: io.kestra.plugin.kafka.ConnectorStatusTrigger
    description: >
      Poll the Connect REST API every minute and start an execution only when
      the connector or any of its tasks is FAILED. Trigger properties are
      rendered before an execution exists, so they read flow variables, not
      inputs.
    connectUrl: "{{ vars.connect_url }}"
    connectorName: "{{ vars.connector_name }}"
    targetState: FAILED
    interval: PT1M
    when: "{{ kv('connect_paged_' ~ vars.connector_name, errorOnMissing=false) is
      null }}"

tasks:
  - id: get_status
    type: io.kestra.plugin.kafka.ConnectorGetStatus
    description: >
      Read the connector state and every task state, including the stack trace
      of failed tasks. A connector often stays RUNNING while one of its tasks is
      FAILED, so both levels are checked below.
    connectUrl: "{{ vars.connect_url }}"
    connectorName: "{{ vars.connector_name }}"
    retry:
      type: constant
      interval: PT10S
      maxAttempts: 3

  - id: read_restart_count
    type: io.kestra.plugin.core.output.OutputValues
    description: >
      Snapshot how many auto-restarts already happened in the current flap
      window. The kv() function returns null for a missing or expired key, so a
      counter whose TTL just ran out reads as 0.
    values:
      restarts: "{{ kv('connect_restarts_' ~ vars.connector_name,
        errorOnMissing=false) ?? 0 }}"

  - id: route
    type: io.kestra.plugin.core.flow.Switch
    description: >
      HEALTHY when nothing is FAILED any more, RESTART while under the restart
      budget, ESCALATE once the budget for the flap window is spent.
    value: >-
      {%- set failed_tasks = outputs.get_status.tasks | jq('map(select(.state ==
      "FAILED")) | length') | first -%} {%- set restarts =
      outputs.read_restart_count.values.restarts | number -%} {%- if
      outputs.get_status.connectorState != 'FAILED' and failed_tasks == 0
      -%}HEALTHY {%- elseif restarts < vars.max_restarts -%}RESTART {%- else
      -%}ESCALATE{%- endif -%}
    cases:
      RESTART:
        - id: bump_restart_count
          type: io.kestra.plugin.core.kv.Set
          description: >
            Count this restart before acting. The TTL equals the flap window, so
            the budget resets once the connector stays healthy for a full
            window.
          key: "connect_restarts_{{ vars.connector_name }}"
          kvType: NUMBER
          value: "{{ (outputs.read_restart_count.values.restarts | number) + 1 }}"
          ttl: "{{ vars.flap_window }}"

        - id: restart_connector
          type: io.kestra.plugin.kafka.ConnectorRestart
          description: >
            Restart the connector and only its FAILED tasks, so healthy tasks
            keep processing while the broken one comes back.
          connectUrl: "{{ vars.connect_url }}"
          connectorName: "{{ vars.connector_name }}"
          includeTasks: true
          onlyFailed: true
          retry:
            type: constant
            interval: PT10S
            maxAttempts: 3

        - id: wait_for_recovery
          type: io.kestra.plugin.core.flow.LoopUntil
          description: >
            Restarts are asynchronous. Re-read the status every 10 seconds until
            the connector and all of its tasks are RUNNING, for at most 2
            minutes.
          condition: "{{ outputs.recheck_status.connectorState == 'RUNNING' and
            (outputs.recheck_status.tasks | jq('map(select(.state !=
            \"RUNNING\")) | length') | first) == 0 }}"
          failOnMaxReached: false
          checkFrequency:
            interval: PT10S
            maxIterations: 12
          tasks:
            - id: recheck_status
              type: io.kestra.plugin.kafka.ConnectorGetStatus
              connectUrl: "{{ vars.connect_url }}"
              connectorName: "{{ vars.connector_name }}"

        - id: final_status
          type: io.kestra.plugin.kafka.ConnectorGetStatus
          description: The state the connector settled in after the restart.
          connectUrl: "{{ vars.connect_url }}"
          connectorName: "{{ vars.connector_name }}"

        - id: report_restart
          type: io.kestra.plugin.core.flow.If
          description: Report a real recovery or a restart that did not stick.
          condition: "{{ outputs.final_status.connectorState == 'RUNNING' and
            (outputs.final_status.tasks | jq('map(select(.state != \"RUNNING\"))
            | length') | first) == 0 }}"
          then:
            - id: notify_recovered
              type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
              url: "{{ secret('SLACK_WEBHOOK_URL') }}"
              payload: |
                {
                  "text": ":white_check_mark: Kafka Connect connector `{{ vars.connector_name }}` self-healed. Restart {{ (outputs.read_restart_count.values.restarts | number) + 1 }} of {{ vars.max_restarts }} in the {{ vars.flap_window }} window. All tasks RUNNING. Execution {{ execution.id }}."
                }
          else:
            - id: notify_not_recovered
              type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
              url: "{{ secret('SLACK_WEBHOOK_URL') }}"
              payload: |
                {
                  "text": ":warning: Kafka Connect connector `{{ vars.connector_name }}` is still unhealthy after restart {{ (outputs.read_restart_count.values.restarts | number) + 1 }} of {{ vars.max_restarts }} (connector {{ outputs.final_status.connectorState }}). The next failed poll retries until the budget is spent. Execution {{ execution.id }}."
                }

      ESCALATE:
        - id: page_on_call
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          description: >
            The restart budget is spent. Stop restarting and send a human the
            failed task trace, so the page carries the root cause and not just a
            state.
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {
              "text": {{ (':rotating_light: <!channel> Kafka Connect connector `' ~ vars.connector_name ~ '` failed again after ' ~ vars.max_restarts ~ ' auto-restarts in ' ~ vars.flap_window ~ '. Auto-restart is muted until the window ends. Execution ' ~ execution.id ~ '. First failed task trace: ```' ~ (outputs.get_status.tasks | jq('map(select(.state == "FAILED")) | (.[0].trace // "no task trace, the connector itself failed") | .[0:700]') | first) ~ '```') | toJson }}
            }

        - id: mute_until_window_ends
          type: io.kestra.plugin.core.kv.Set
          description: >
            Record that on-call was paged. The trigger skips polling while this
            key exists, so the page is sent once per flap window and not on
            every poll. Both keys expire after the window, then the restart
            budget starts again.
          key: "connect_paged_{{ vars.connector_name }}"
          value: "{{ execution.id }}"
          ttl: "{{ vars.flap_window }}"

    defaults:
      - id: already_healthy
        type: io.kestra.plugin.core.log.Log
        description: The connector recovered before this run acted. Nothing to do.
        message: "Connector {{ vars.connector_name }} has no FAILED connector or task
          any more. No restart needed."

errors:
  - id: alert_selfheal_failed
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: A Connect API call itself failed, so the connector may still be
      down. Page loudly.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": ":rotating_light: <!channel> Self-heal for Kafka Connect connector `{{ vars.connector_name }}` errored in {{ flow.namespace }}.{{ flow.id }}. The connector may still be down. Execution {{ execution.id }}."
      }