Argo CD Out-of-Sync Application Automated Rollback Guard

id: argocd-application-out-of-sync-guard
namespace: company.gitops
description: "Guard against prolonged Argo CD application drift and degraded
  health by enforcing SLA windows, triggering automated sync refresh or
  rollbacks, and alerting Slack with diff summaries."

inputs:
  - id: argocd_server
    type: STRING
    defaults: "https://argocd.example.com"
    displayName: Argo CD Server URL
    description: "Base URL of the Argo CD API server (e.g. https://argocd.example.com)."

  - id: out_of_sync_sla_minutes
    type: INT
    defaults: 30
    displayName: Out-of-Sync SLA Window (Minutes)
    description: "Maximum allowable elapsed minutes an application may stay in
      OutOfSync or Degraded state before triggering automated remediation."

  - id: remediation_action
    type: SELECT
    values:
      - "dry-run"
      - "sync"
      - "rollback"
      - "notify-only"
    defaults: "dry-run"
    displayName: Remediation Action Mode
    description: "Action to perform for SLA breaches: dry-run (safe preview), sync
      (re-trigger synchronization), rollback (revert to previous stable
      revision), or notify-only."

  - id: auto_rollback_on_degraded
    type: BOOL
    defaults: true
    displayName: Auto-Rollback on Degraded Health
    description: "Prioritize rolling back to the previous stable revision if the
      application health is Degraded, even when remediation mode is set to
      sync."

  - id: prune_resources
    type: BOOL
    defaults: false
    displayName: Prune Resources
    description: "Whether sync or rollback operations should prune orphaned
      resources no longer defined in Git."

  - id: project_filter
    type: STRING
    defaults: "all"
    displayName: Argo CD Project Filter
    description: "Filter applications by Argo CD project name (e.g. 'production').
      Set to 'all' to audit all registered projects."

  - id: owner_tag_key
    type: STRING
    defaults: "owner"
    displayName: Owner Tag Key
    description: "Label or annotation key used to extract the application owning
      team for incident routing (falls back to team or
      app.kubernetes.io/part-of)."

  - id: slack_channel
    type: STRING
    defaults: "#gitops-alerts"
    displayName: Slack Notification Channel
    description: "Slack channel name or ID for routing SLA breach alerts."

tasks:
  - id: query_argocd_applications
    type: io.kestra.plugin.core.http.Request
    description: "Query Argo CD REST API /api/v1/applications to list all managed
      applications with full sync and health state."
    uri: "{{ inputs.argocd_server }}/api/v1/applications"
    method: GET
    headers:
      Authorization: "Bearer {{ secret('ARGOCD_TOKEN') }}"
      Content-Type: application/json
    retry:
      type: constant
      interval: PT5S
      maxAttempts: 3

  - id: inspect_and_classify_applications
    type: io.kestra.plugin.scripts.python.Script
    description: "Inspect application sync and health status, calculate drift
      duration against SLA, collect failing manifests, and plan remediation
      actions."
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    containerImage: python:3.11-slim
    inputFiles:
      applications.json: "{{ outputs.query_argocd_applications.body ?? '{}' }}"
    env:
      SLA_MINUTES: "{{ inputs.out_of_sync_sla_minutes }}"
      REMEDIATION_ACTION: "{{ inputs.remediation_action }}"
      AUTO_ROLLBACK_ON_DEGRADED: "{{ inputs.auto_rollback_on_degraded }}"
      PROJECT_FILTER: "{{ inputs.project_filter }}"
      OWNER_TAG_KEY: "{{ inputs.owner_tag_key }}"
    script: |
      import json
      import os
      import sys
      from datetime import datetime, timezone

      def parse_int(val, default):
          try:
              return int(val)
          except (TypeError, ValueError):
              return default

      def parse_bool(val, default=True):
          if isinstance(val, bool):
              return val
          if isinstance(val, str):
              return val.lower() in ("true", "1", "yes")
          return default

      sla_minutes = parse_int(os.environ.get("SLA_MINUTES"), 30)
      remediation_action = os.environ.get("REMEDIATION_ACTION", "dry-run").lower()
      auto_rollback_on_degraded = parse_bool(os.environ.get("AUTO_ROLLBACK_ON_DEGRADED"), True)
      project_filter = os.environ.get("PROJECT_FILTER", "all").strip()
      owner_tag_key = os.environ.get("OWNER_TAG_KEY", "owner").strip()

      now = datetime.now(timezone.utc)

      try:
          with open("applications.json", "r", encoding="utf-8") as f:
              raw_content = f.read().strip()
              data = json.loads(raw_content) if raw_content else {}
      except Exception as exc:
          print(f"Warning: could not parse applications.json: {exc}", file=sys.stderr)
          data = {}

      items = data.get("items", []) if isinstance(data, dict) else (data if isinstance(data, list) else [])

      total_count = 0
      healthy_count = 0
      out_of_sync_count = 0
      degraded_count = 0
      breached_apps = []
      remediation_actions = []

      for item in items:
          if not isinstance(item, dict):
              continue

          metadata = item.get("metadata", {})
          spec = item.get("spec", {})
          status = item.get("status", {})

          name = metadata.get("name", "unnamed")
          project = spec.get("project") or metadata.get("project", "default")

          if project_filter and project_filter.lower() != "all" and project.lower() != project_filter.lower():
              continue

          total_count += 1

          labels = metadata.get("labels") or {}
          annotations = metadata.get("annotations") or {}

          owner = (
              labels.get(owner_tag_key)
              or annotations.get(owner_tag_key)
              or labels.get("owner")
              or labels.get("team")
              or labels.get("app.kubernetes.io/part-of")
              or annotations.get("owner")
              or "unassigned"
          )

          sync_status = status.get("sync", {}).get("status", "Unknown")
          current_revision = status.get("sync", {}).get("revision") or spec.get("source", {}).get("targetRevision", "HEAD")
          health_status = status.get("health", {}).get("status", "Unknown")
          health_message = status.get("health", {}).get("message", "")

          if sync_status == "OutOfSync":
              out_of_sync_count += 1
          if health_status == "Degraded":
              degraded_count += 1

          if sync_status == "Synced" and health_status == "Healthy":
              healthy_count += 1
              continue

          op_state = status.get("operationState", {})
          finished_at_str = (
              op_state.get("finishedAt")
              or op_state.get("startedAt")
              or status.get("reconciledAt")
              or metadata.get("creationTimestamp")
          )

          elapsed_minutes = 0
          if finished_at_str:
              try:
                  dt = datetime.fromisoformat(finished_at_str.replace("Z", "+00:00"))
                  elapsed_minutes = max(0, int((now - dt).total_seconds() / 60))
              except Exception:
                  elapsed_minutes = 0

          is_breached = (
              (elapsed_minutes >= sla_minutes)
              or (health_status == "Degraded" and elapsed_minutes >= max(1, sla_minutes // 2))
          )

          failing_manifests = []
          for res in status.get("resources", []):
              r_kind = res.get("kind", "Resource")
              r_name = res.get("name", "unnamed")
              r_ns = res.get("namespace", project)
              r_sync = res.get("status", "")
              r_health = res.get("health", {}).get("status", "")
              r_msg = res.get("health", {}).get("message", "")

              if r_sync == "OutOfSync" or r_health in ["Degraded", "Missing", "Unknown"]:
                  reasons = []
                  if r_sync == "OutOfSync":
                      reasons.append("OutOfSync")
                  if r_health and r_health != "Healthy":
                      reasons.append(f"Health:{r_health}" + (f" ({r_msg})" if r_msg else ""))
                  failing_manifests.append({
                      "kind": r_kind,
                      "name": r_name,
                      "namespace": r_ns,
                      "status": ", ".join(reasons)
                  })

          for sr in op_state.get("syncResult", {}).get("resources", []):
              if sr.get("status") in ["SyncFailed", "Failed"]:
                  failing_manifests.append({
                      "kind": sr.get("kind", "Resource"),
                      "name": sr.get("name", "unnamed"),
                      "namespace": sr.get("namespace", project),
                      "status": f"SyncFailed: {sr.get('message', 'operation error')}"
                  })

          diff_summary = (
              op_state.get("message")
              or health_message
              or (failing_manifests[0]["status"] if failing_manifests else "Manifest diff detected with Git source-of-truth")
          )

          history = status.get("history", [])
          previous_revision = None
          previous_history_id = None
          if len(history) >= 2:
              prev_entry = history[-2]
              previous_revision = prev_entry.get("revision")
              previous_history_id = prev_entry.get("id")

          is_dry_run = (remediation_action == "dry-run")
          chosen_action = "NOTIFY_ONLY"
          endpoint_type = "sync"
          target_rev = current_revision
          hist_id = previous_history_id
          action_desc = "Notify only (no mutation)"

          if remediation_action != "notify-only" and is_breached:
              if (remediation_action == "rollback") or (auto_rollback_on_degraded and health_status == "Degraded"):
                  if previous_history_id is not None:
                      endpoint_type = "rollback"
                      chosen_action = "DRY_RUN_ROLLBACK" if is_dry_run else "ROLLBACK"
                      action_desc = f"{'[DRY-RUN] ' if is_dry_run else ''}Rollback to history #{previous_history_id} (commit {previous_revision[:8] if previous_revision else 'prev'})"
                  else:
                      endpoint_type = "sync"
                      chosen_action = "DRY_RUN_SYNC" if is_dry_run else "SYNC_REFRESH"
                      action_desc = f"{'[DRY-RUN] ' if is_dry_run else ''}Sync refresh (Rollback unavailable: single history entry)"
              else:
                  endpoint_type = "sync"
                  chosen_action = "DRY_RUN_SYNC" if is_dry_run else "SYNC_REFRESH"
                  action_desc = f"{'[DRY-RUN] ' if is_dry_run else ''}Sync refresh for revision {current_revision[:8] if current_revision else 'HEAD'}"

              remediation_actions.append({
                  "name": name,
                  "action": chosen_action,
                  "endpoint_type": endpoint_type,
                  "target_revision": target_rev,
                  "history_id": hist_id,
                  "dry_run": is_dry_run,
                  "owner": owner,
                  "project": project,
                  "reason": f"Health={health_status}, Sync={sync_status}, Elapsed={elapsed_minutes}m (SLA={sla_minutes}m)"
              })

          if is_breached:
              breached_apps.append({
                  "name": name,
                  "project": project,
                  "owner": owner,
                  "health": health_status,
                  "sync": sync_status,
                  "elapsed_minutes": elapsed_minutes,
                  "current_revision": current_revision,
                  "previous_revision": previous_revision,
                  "failing_manifests": failing_manifests,
                  "diff_summary": diff_summary,
                  "action_description": action_desc
              })

      slack_lines = [
          "*Fleet Audit Summary:*",
          f"• Total Inspected: {total_count} | OutOfSync: {out_of_sync_count} | Degraded: {degraded_count} | SLA Breaches: {len(breached_apps)}",
          f"• SLA Threshold: {sla_minutes} minutes | Remediation Mode: `{remediation_action}`",
          ""
      ]
      if breached_apps:
          slack_lines.append("*Applications Exceeding SLA Window:*")
          for idx, app in enumerate(breached_apps, 1):
              slack_lines.append(f"*{idx}. {app['name']}* (Project: `{app['project']}`, Owner: `{app['owner']}`)")
              slack_lines.append(f"• *Status:* Health=`{app['health']}` | Sync=`{app['sync']}` (Elapsed: `{app['elapsed_minutes']}m`)")
              c_rev = app['current_revision'][:8] if app['current_revision'] else "unknown"
              p_rev = app['previous_revision'][:8] if app['previous_revision'] else "none"
              slack_lines.append(f"• *Revisions:* Current=`{c_rev}` | Previous Stable=`{p_rev}`")
              if app['failing_manifests']:
                  slack_lines.append("• *Failing Manifests:*")
                  for m in app['failing_manifests'][:4]:
                      slack_lines.append(f"  - `{m['kind']}/{m['name']}` ({m['namespace']}): {m['status']}")
              if app['diff_summary']:
                  slack_lines.append(f"• *Diff Summary:* {app['diff_summary']}")
              slack_lines.append(f"• *Remediation Action:* `{app['action_description']}`")
              slack_lines.append("")

      slack_text = "\n".join(slack_lines)

      output_data = {
          "total_count": total_count,
          "healthy_count": healthy_count,
          "out_of_sync_count": out_of_sync_count,
          "degraded_count": degraded_count,
          "breached_count": len(breached_apps),
          "breached_apps": breached_apps,
          "remediation_actions": remediation_actions,
          "slack_text": slack_text
      }

      try:
          from kestra import Kestra
          Kestra.outputs(output_data)
      except ImportError:
          pass

      print("::" + json.dumps({"outputs": output_data}) + "::")

  - id: check_remediation_needed
    type: io.kestra.plugin.core.flow.If
    description: "Execute dry-run or automated sync / rollback when one or more
      applications exceed the SLA window and mode is not notify-only."
    condition: "{{
      (outputs.inspect_and_classify_applications.vars.remediation_actions |
      length > 0) and (inputs.remediation_action != 'notify-only') }}"
    then:
      - id: remediate_applications
        type: io.kestra.plugin.core.flow.Loop
        description: "Iterate over breached applications to execute sync refresh or
          rollback via Argo CD REST API."
        values: "{{ outputs.inspect_and_classify_applications.vars.remediation_actions
          }}"
        concurrencyLimit: 1
        tasks:
          - id: route_remediation
            type: io.kestra.plugin.core.flow.Switch
            value: "{{ item.value.endpoint_type }}"
            cases:
              "sync":
                - id: trigger_sync_refresh
                  type: io.kestra.plugin.core.http.Request
                  description: "POST to /api/v1/applications/{name}/sync to initiate dry-run or
                    automated synchronization."
                  uri: "{{ inputs.argocd_server }}/api/v1/applications/{{ item.value.name }}/sync"
                  method: POST
                  headers:
                    Authorization: "Bearer {{ secret('ARGOCD_TOKEN') }}"
                    Content-Type: application/json
                  body: |
                    {
                      "revision": "{{ item.value.target_revision }}",
                      "dryRun": {{ item.value.dry_run }},
                      "prune": {{ inputs.prune_resources }},
                      "strategy": {
                        "hook": {}
                      }
                    }
                  retry:
                    type: constant
                    interval: PT5S
                    maxAttempts: 2

              "rollback":
                - id: trigger_application_rollback
                  type: io.kestra.plugin.core.http.Request
                  description: "POST to /api/v1/applications/{name}/rollback to revert to previous
                    stable deployment revision."
                  uri: "{{ inputs.argocd_server }}/api/v1/applications/{{ item.value.name
                    }}/rollback"
                  method: POST
                  headers:
                    Authorization: "Bearer {{ secret('ARGOCD_TOKEN') }}"
                    Content-Type: application/json
                  body: |
                    {
                      "id": {{ item.value.history_id }},
                      "dryRun": {{ item.value.dry_run }},
                      "prune": {{ inputs.prune_resources }}
                    }
                  retry:
                    type: constant
                    interval: PT5S
                    maxAttempts: 2
            defaults:
              - id: log_skipped_remediation
                type: io.kestra.plugin.core.log.Log
                message: "Skipping automated mutation for {{ item.value.name }}: mode={{
                  item.value.action }}."

  - id: evaluate_notification_gate
    type: io.kestra.plugin.core.flow.If
    description: "Send detailed Slack alert if any application breached SLA;
      otherwise log healthy status."
    condition: "{{ outputs.inspect_and_classify_applications.vars.breached_count > 0 }}"
    then:
      - id: send_slack_alert
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: "Post detailed incident alert with failing manifests, diff summary,
          and owner tags."
        url: "{{ secret('SLACK_WEBHOOK') }}"
        payload: |
          {
            "channel": "{{ inputs.slack_channel }}",
            "attachments": [
              {
                "color": "{{ outputs.inspect_and_classify_applications.vars.degraded_count > 0 ? 'danger' : 'warning' }}",
                "title": "🚨 Argo CD Out-of-Sync & Rollback Guard Alert",
                "text": {{ outputs.inspect_and_classify_applications.vars.slack_text | toJson }},
                "footer": "Kestra GitOps Guard • Argo CD"
              }
            ]
          }
    else:
      - id: log_fleet_healthy
        type: io.kestra.plugin.core.log.Log
        message: "Argo CD Guard: All {{
          outputs.inspect_and_classify_applications.vars.total_count }}
          applications in project '{{ inputs.project_filter }}' are Synced,
          Healthy, and within SLA."

  - id: log_audit_summary
    type: io.kestra.plugin.core.log.Log
    description: "Record structured operational audit trail for compliance."
    message: "Argo CD Guard completed: execution={{ execution.id }}, project={{
      inputs.project_filter }}, inspected={{
      outputs.inspect_and_classify_applications.vars.total_count }},
      out_of_sync={{
      outputs.inspect_and_classify_applications.vars.out_of_sync_count }},
      degraded={{ outputs.inspect_and_classify_applications.vars.degraded_count
      }}, breached={{
      outputs.inspect_and_classify_applications.vars.breached_count }}, mode={{
      inputs.remediation_action }}."

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: "Alert Slack when the guard flow fails unexpectedly."
    url: "{{ secret('SLACK_WEBHOOK') }}"
    payload: |
      {
        "channel": "{{ inputs.slack_channel }}",
        "attachments": [
          {
            "color": "danger",
            "title": "❌ Argo CD Rollback Guard Execution Failure",
            "text": "The Argo CD Out-of-Sync Guard flow failed during execution `{{ execution.id }}` against server `{{ inputs.argocd_server }}`. Inspect Kestra task logs to verify network reachability, TLS trust, and API token validity.",
            "footer": "Kestra Failure Sentinel"
          }
        ]
      }

triggers:
  - id: scheduled_sla_sweep
    type: io.kestra.plugin.core.trigger.Schedule
    description: "Audit Argo CD applications every 15 minutes. Disabled by default
      until credentials and SLA parameters are validated."
    cron: "*/15 * * * *"
    disabled: true

  - id: on_demand_webhook
    type: io.kestra.plugin.core.trigger.Webhook
    description: "Trigger on-demand audit from external webhooks, post-sync Git
      hooks, or incident response runbooks."
    key: "{{ secret('ARGOCD_GUARD_WEBHOOK_KEY') }}"

outputs:
  - id: total_applications_inspected
    type: INT
    value: "{{ outputs.inspect_and_classify_applications.vars.total_count }}"
  - id: out_of_sync_applications_count
    type: INT
    value: "{{ outputs.inspect_and_classify_applications.vars.out_of_sync_count }}"
  - id: degraded_applications_count
    type: INT
    value: "{{ outputs.inspect_and_classify_applications.vars.degraded_count }}"
  - id: breached_sla_count
    type: INT
    value: "{{ outputs.inspect_and_classify_applications.vars.breached_count }}"
  - id: remediation_actions_count
    type: INT
    value: "{{ outputs.inspect_and_classify_applications.vars.remediation_actions |
      length }}"