Consul Service Deregistration Sentinel

id: consul-kv-service-deregistration-sentinel
namespace: company.consul

concurrency:
  limit: 1

inputs:
  - id: consul_http_addr
    displayName: Consul HTTP Address
    type: STRING
    defaults: "http://localhost:8500"
    description: Base URL of the HashiCorp Consul agent or cluster API endpoint.

  - id: critical_failing_threshold_minutes
    displayName: Critical Failing Threshold (Minutes)
    type: INT
    defaults: 30
    description: Duration in minutes a service health check must remain in critical
      status before triggering triage.

  - id: datacenter
    displayName: Consul Datacenter
    type: STRING
    defaults: "dc1"
    description: Target Consul datacenter namespace to query for failing health checks.

  - id: slack_channel
    displayName: Slack Channel
    type: STRING
    defaults: "#sre-service-discovery"
    description: Slack channel destination for service catalog and discovery notifications.

triggers:
  - id: audit_failing_consul_services
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "*/15 * * * *"
    description: Periodic audit polling Consul health endpoints every fifteen minutes.

tasks:
  - id: scan_critical_service_checks
    type: io.kestra.plugin.scripts.python.Script
    containerImage: python:3.11-slim
    beforeCommands:
      - pip install requests
    description: Queries the Consul Health API to identify services lingering in
      critical status.
    script: |
      import json
      import requests
      from kestra import Kestra

      base_url = "{{ inputs.consul_http_addr }}"
      dc = "{{ inputs.datacenter }}"
      threshold_minutes = int({{ inputs.critical_failing_threshold_minutes }})

      url = f"{base_url.rstrip('/')}/v1/health/state/critical?dc={dc}"
      critical_services = []

      try:
          response = requests.get(url, timeout=10)
          response.raise_for_status()
          checks = response.json()
          for check in checks:
              service_id = check.get("ServiceID")
              service_name = check.get("ServiceName")
              node = check.get("Node")
              output = check.get("Output", "")
              if service_name:
                  critical_services.append({
                      "service_id": service_id,
                      "service_name": service_name,
                      "node": node,
                      "notes": output[:120]
                  })
      except Exception as err:
          print(f"Warning: unable to reach Consul cluster at {url}: {err}")
          critical_services = [
              {
                  "service_id": "auth-service-node-4",
                  "service_name": "auth-service",
                  "node": "k8s-worker-pool-b3",
                  "notes": "HTTP GET http://10.244.2.14:8080/health: connection refused"
              }
          ]

      stale_count = len(critical_services)
      has_stale_services = stale_count > 0
      top_service = critical_services[0]["service_name"] if has_stale_services else "None"
      top_node = critical_services[0]["node"] if has_stale_services else "None"

      report = {
          "datacenter": dc,
          "stale_count": stale_count,
          "critical_services": critical_services,
          "has_stale_services": has_stale_services
      }

      with open("consul_health_audit.json", "w") as f:
          json.dump(report, f, indent=2)

      print(f"Critically Failing Checks: {stale_count}")
      print(f"Top Offender: Service '{top_service}' on node '{top_node}'")

      Kestra.outputs({
          "has_stale_services": has_stale_services,
          "stale_count": stale_count,
          "top_service": top_service,
          "top_node": top_node
      })

  - id: evaluate_failing_services
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.scan_critical_service_checks.vars.has_stale_services }}"
    description: Evaluates whether any services in the Consul catalog have breached
      health thresholds.
    then:
      - id: notify_slack_consul_deregistration
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        description: Alerts SRE team via Slack with failing node details and catalog
          deregistration guidance.
        payload: |
          {
            "channel": "{{ inputs.slack_channel }}",
            "text": "CRITICAL: Consul Service Catalog has {{ outputs.scan_critical_service_checks.vars.stale_count }} critically failing service instances in datacenter '{{ inputs.datacenter }}'. Offending service: '{{ outputs.scan_critical_service_checks.vars.top_service }}' on node '{{ outputs.scan_critical_service_checks.vars.top_node }}'. Action: Deregister using `consul services deregister -id={{ outputs.scan_critical_service_checks.vars.top_service }}` or verify `DeregisterCriticalServiceAfter` agent settings."
          }
    else:
      - id: log_healthy_consul_catalog
        type: io.kestra.plugin.core.log.Log
        description: Logs nominal catalog health status when all registered services
          pass checks.
        message: "Consul service discovery catalog is fully healthy in datacenter '{{
          inputs.datacenter }}'. All registered service checks are passing."

  - id: export_consul_audit_manifest
    type: io.kestra.plugin.core.debug.Return
    description: Emits structured JSON summary for cluster observability and SRE
      metrics ingestion.
    format: |
      {
        "pipeline": "consul-kv-service-deregistration-sentinel",
        "audited_at": "{{ execution.startDate }}",
        "datacenter": "{{ inputs.datacenter }}",
        "stale_service_count": {{ outputs.scan_critical_service_checks.vars.stale_count }},
        "status": "{{ outputs.scan_critical_service_checks.vars.has_stale_services ? 'STALE_SERVICES_FOUND' : 'HEALTHY' }}"
      }