Kafka Topic Config Drift Guard

id: kafka-topic-config-drift-guard
namespace: company.team

description: |
  Compare the live configuration of Kafka topics with the policy you declare for them
  (retention, cleanup policy, min.insync.replicas, max message size...), report every
  drifted key and every topic missing from the cluster, alert Slack, and optionally put the
  drifted keys back with TopicUpdate, then describe the topics again to prove the fix landed.
  Remediation is off by default and runs only when auto_remediate is true and dry_run is false.

inputs:
  - id: bootstrap_servers
    type: STRING
    displayName: Kafka bootstrap servers
    defaults: "localhost:9092"

  - id: policy
    type: JSON
    displayName: Topic policy
    description: >-
      Desired configs per topic, as Kafka config keys and string values. Only
      the keys listed are checked, every other config is left alone.
    defaults: >-
      {"orders": {"retention.ms": "604800000", "cleanup.policy": "delete",
      "min.insync.replicas": "1"},
       "payments": {"retention.ms": "2592000000", "cleanup.policy": "delete"},
       "audit-log": {"retention.ms": "31536000000"},
       "customers-compacted": {"cleanup.policy": "compact"},
       "invoices": {"retention.ms": "2592000000"}}

  - id: auto_remediate
    type: BOOL
    displayName: Auto remediate
    description: Set the drifted keys back to the policy value with TopicUpdate.
      Missing topics are never created.
    defaults: false

  - id: dry_run
    type: BOOL
    displayName: Dry run
    description: When true, the changes are only logged. Must be false, together
      with auto_remediate, for TopicUpdate to run.
    defaults: true

tasks:
  - id: list_topics
    type: io.kestra.plugin.kafka.TopicList
    description: Read-only AdminClient call listing the topics on the cluster, so
      topics in the policy that do not exist are reported instead of failing the
      describe.
    properties:
      bootstrap.servers: "{{ inputs.bootstrap_servers }}"

  - id: present
    type: io.kestra.plugin.core.output.OutputValues
    values:
      topics: "{{ inputs.policy | keys | jq('[.[] | select(. as $t | ' ~
        (outputs.list_topics.topics | toJson) ~ ' | index($t))]') | first |
        toJson }}"
      missing: "{{ inputs.policy | keys | jq('[.[] | select(. as $t | ' ~
        (outputs.list_topics.topics | toJson) ~ ' | index($t) | not)]') | first
        | toJson }}"

  - id: describe_topics
    type: io.kestra.plugin.core.flow.Loop
    description: Describe every policy topic present on the cluster, read-only.
    values: "{{ outputs.present.values.topics }}"
    concurrencyLimit: 4
    outputs:
      - id: live
        type: STRING
        value: "{{ {'topic': item.value, 'configs': outputs.describe.configs} | toJson
          }}"
    tasks:
      - id: describe
        type: io.kestra.plugin.kafka.TopicDescribe
        topic: "{{ item.value }}"
        properties:
          bootstrap.servers: "{{ inputs.bootstrap_servers }}"

  - id: drift
    type: io.kestra.plugin.core.output.OutputValues
    description: One entry per drifted key, with the live and the desired value.
    values:
      items: >-
        {{ {'policy': inputs.policy, 'actual': ((outputs.describe_topics.outputs
        ?? []) | jq('map(.outputs.live | fromjson | {(.topic): .configs}) | add
        // {}') | first)} | jq('. as $r | [$r.policy | to_entries[] as $p |
        ($p.value | to_entries[]) as $k
           | select($r.actual[$p.key] != null) | {topic: $p.key, key: $k.key, desired: $k.value, actual: ($r.actual[$p.key][$k.key] // "unset")}
           | select(.desired != .actual)]') | first | toJson }}

  - id: summary
    type: io.kestra.plugin.core.log.Log
    message: >-
      Kafka topic policy on {{ inputs.bootstrap_servers }}: {{ inputs.policy |
      keys | length }} topics in the policy, {{
      fromJson(outputs.drift.values.items) | length }} drifted keys{{
      fromJson(outputs.drift.values.items) | jq('if length == 0 then "" else ":
      " + (map("\(.topic) \(.key)=\(.actual) (want \(.desired))") | join(", "))
      end') | first }}. Missing from the cluster: {{
      fromJson(outputs.present.values.missing) | length == 0 ? 'none' :
      fromJson(outputs.present.values.missing) | join(', ') }}.

  - id: route
    type: io.kestra.plugin.core.flow.If
    condition: "{{ fromJson(outputs.drift.values.items) | length > 0 or
      fromJson(outputs.present.values.missing) | length > 0 }}"
    then:
      - id: alert
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {{ {'text': 'Kafka topic config drift on ' ~ inputs.bootstrap_servers ~ ': ' ~ (fromJson(outputs.drift.values.items) | length) ~ ' drifted keys, ' ~ (fromJson(outputs.present.values.missing) | length) ~ ' missing topics. ' ~ (fromJson(outputs.drift.values.items) | jq('map("\(.topic) \(.key)=\(.actual), policy \(.desired)") | join("; ")') | first) ~ '.' ~ (inputs.auto_remediate and not inputs.dry_run ? ' Remediating now.' : ' No change applied (auto_remediate=' ~ inputs.auto_remediate ~ ', dry_run=' ~ inputs.dry_run ~ ').')} | toJson }}

      - id: remediate_gate
        type: io.kestra.plugin.core.flow.If
        condition: "{{ inputs.auto_remediate and not inputs.dry_run and
          fromJson(outputs.drift.values.items) | length > 0 }}"
        then:
          - id: remediate
            type: io.kestra.plugin.core.flow.Loop
            description: One TopicUpdate per drifted topic, with only its drifted keys.
            values: "{{ fromJson(outputs.drift.values.items) | jq('[group_by(.topic)[] |
              {topic: .[0].topic, configs: (map({(.key): .desired}) | add)}]') |
              first | toJson }}"
            tasks:
              - id: update
                type: io.kestra.plugin.kafka.TopicUpdate
                topic: "{{ fromJson(item.value).topic }}"
                configs: "{{ fromJson(item.value).configs }}"
                properties:
                  bootstrap.servers: "{{ inputs.bootstrap_servers }}"

          - id: verify
            type: io.kestra.plugin.core.flow.Loop
            description: Describe the remediated topics again, to check the broker applied
              the change.
            values: "{{ fromJson(outputs.drift.values.items) | jq('[.[].topic] | unique') |
              first | toJson }}"
            outputs:
              - id: live
                type: STRING
                value: "{{ {'topic': item.value, 'configs': outputs.recheck.configs} | toJson
                  }}"
            tasks:
              - id: recheck
                type: io.kestra.plugin.kafka.TopicDescribe
                topic: "{{ item.value }}"
                properties:
                  bootstrap.servers: "{{ inputs.bootstrap_servers }}"

          - id: verify_result
            type: io.kestra.plugin.core.flow.If
            condition: >-
              {{ {'d': fromJson(outputs.drift.values.items), 'a':
              outputs.verify.outputs} | jq('(.a | map(.outputs.live | fromjson |
              {(.topic): .configs}) | add) as $live | [.d[] as $x |
              select($live[$x.topic][$x.key] != $x.desired)] | length') | first
              > 0 }}
            then:
              - id: remediation_failed
                type: io.kestra.plugin.core.execution.Fail
                errorMessage: "TopicUpdate ran, but some keys still differ from the policy after
                  the re-describe."
            else:
              - id: remediated
                type: io.kestra.plugin.core.log.Log
                message: "Remediated and verified: {{ fromJson(outputs.drift.values.items) |
                  length }} keys now match the policy."

      - id: missing_gate
        type: io.kestra.plugin.core.flow.If
        condition: "{{ fromJson(outputs.present.values.missing) | length > 0 }}"
        then:
          - id: missing_topics
            type: io.kestra.plugin.core.log.Log
            level: WARN
            message: "Topics in the policy but not on the cluster: {{
              fromJson(outputs.present.values.missing) | join(', ') }}. Create
              them through your usual process, this flow never creates topics."
    else:
      - id: compliant
        type: io.kestra.plugin.core.log.Log
        message: "All {{ inputs.policy | keys | length }} policy topics match their
          declared configs."

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: A broker that cannot be reached or a failed remediation must not
      read as a compliant cluster.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {{ {'text': 'Kafka topic config drift guard failed on ' ~ inputs.bootstrap_servers ~ ' (execution ' ~ execution.id ~ '). Check the execution logs: the cluster state is unknown or a change did not apply.'} | toJson }}

triggers:
  - id: hourly
    type: io.kestra.plugin.core.trigger.Schedule
    description: Shipped disabled, and always in safe mode. Enable it once the
      policy and bootstrap servers point at your cluster.
    cron: "0 * * * *"
    disabled: true
    inputs:
      auto_remediate: false
      dry_run: true