CouchDB Active Task and Continuous Replication Sentinel

id: couchdb-replication-and-active-task-sentinel
namespace: company.data

inputs:
  - id: couchdb_host
    type: STRING
    displayName: CouchDB Host
    defaults: "couchdb.internal"
    description: Hostname or IP address of the Apache CouchDB node exposing port 5984.

  - id: couchdb_port
    type: INT
    displayName: CouchDB Port
    defaults: 5984
    description: Port number for the HTTP REST API on CouchDB (default 5984).

  - id: target_database
    type: STRING
    displayName: Target Database
    defaults: "production_documents"
    description: Database name monitored for active background compaction and replication.

  - id: max_pending_changes
    type: INT
    displayName: Max Pending Changes Threshold
    defaults: 500
    description: Maximum tolerated pending replication changes before triggering
      critical alerts.

  - id: slack_channel
    type: STRING
    displayName: Slack Channel
    defaults: "#data-platform-alerts"
    description: Slack channel destination for operational CouchDB incident cards.

  - id: auto_compact
    type: BOOL
    displayName: Auto Trigger Compaction
    defaults: true
    description: Automatically trigger database compaction via POST /db/_compact
      when pending revisions accumulate.

  - id: dry_run
    type: BOOL
    displayName: Dry Run Mode
    defaults: false
    description: When true, audits active tasks and dispatches notifications but
      skips remediation calls.

triggers:
  - id: every_fifteen_minutes
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "*/15 * * * *"

tasks:
  - id: probe_active_tasks
    type: io.kestra.plugin.core.http.Request
    uri: "http://{{ inputs.couchdb_host }}:{{ inputs.couchdb_port }}/_active_tasks"
    method: GET
    headers:
      Accept: application/json
      Authorization: "Basic {{ (secret('COUCHDB_USERNAME') ~ ':' ~
        secret('COUCHDB_PASSWORD')) | base64Encode }}"
    options:
      allowFailed: true
    allowFailure: true

  - id: route_by_task_health
    type: io.kestra.plugin.core.flow.Switch
    value: >-
      {% if outputs.probe_active_tasks.status != 200 or
      outputs.probe_active_tasks.body is null %} PROBE_FAILED {% else %}
        {% set tasks = outputs.probe_active_tasks.body %}
        {% set critical_count = 0 %}
        {% set warning_count = 0 %}
        {% for t in tasks %}
          {% if t.type == 'replication' and t.changes_pending is defined and t.changes_pending >= inputs.max_pending_changes %}
            {% set critical_count = critical_count + 1 %}
          {% elseif t.type == 'replication' and t.changes_pending is defined and t.changes_pending > 0 %}
            {% set warning_count = warning_count + 1 %}
          {% endif %}
        {% endfor %}
        {% if critical_count > 0 %}
        CRITICAL
        {% elseif warning_count > 0 %}
        WARNING
        {% else %}
        HEALTHY
        {% endif %}
      {% endif %}
    cases:
      CRITICAL:
        - id: alert_critical_slack
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {
              "channel": "{{ inputs.slack_channel }}",
              "text": "[CRITICAL] CouchDB Continuous Replication Lag Exceeded Threshold - Data Desynchronization Risk",
              "attachments": [
                {
                  "color": "#D32F2F",
                  "title": "CouchDB Replication Lag Incident",
                  "fields": [
                    {
                      "title": "Cluster Host",
                      "value": "{{ inputs.couchdb_host }}:{{ inputs.couchdb_port }}",
                      "short": true
                    },
                    {
                      "title": "Threshold",
                      "value": "{{ inputs.max_pending_changes }} pending changes",
                      "short": true
                    },
                    {
                      "title": "Severity",
                      "value": "CRITICAL - Replication Falling Behind",
                      "short": true
                    },
                    {
                      "title": "Remediation Status",
                      "value": "{{ (inputs.auto_compact and not inputs.dry_run) ? 'Triggering Compaction POST' : 'Remediation Skipped (Dry Run or Disabled)' }}",
                      "short": true
                    }
                  ],
                  "footer": "Kestra CouchDB Sentinel | Execution ID: {{ execution.id }}"
                }
              ]
            }

        - id: check_remediation
          type: io.kestra.plugin.core.flow.If
          condition: "{{ inputs.auto_compact == true and inputs.dry_run == false }}"
          then:
            - id: trigger_database_compaction
              type: io.kestra.plugin.core.http.Request
              uri: "http://{{ inputs.couchdb_host }}:{{ inputs.couchdb_port }}/{{
                inputs.target_database }}/_compact"
              method: POST
              headers:
                Content-Type: application/json
                Authorization: "Basic {{ (secret('COUCHDB_USERNAME') ~ ':' ~
                  secret('COUCHDB_PASSWORD')) | base64Encode }}"
              options:
                allowFailed: true

            - id: notify_remediation_slack
              type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
              url: "{{ secret('SLACK_WEBHOOK_URL') }}"
              payload: |
                {
                  "channel": "{{ inputs.slack_channel }}",
                  "text": "[REMEDIATION] CouchDB database compaction triggered successfully for {{ inputs.target_database }}."
                }

      WARNING:
        - id: alert_warning_slack
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {
              "channel": "{{ inputs.slack_channel }}",
              "text": "[WARNING] CouchDB Active Replication Tasks Processing Changes",
              "attachments": [
                {
                  "color": "#F57C00",
                  "title": "CouchDB Replication Activity Warning",
                  "fields": [
                    {
                      "title": "Cluster Host",
                      "value": "{{ inputs.couchdb_host }}:{{ inputs.couchdb_port }}",
                      "short": true
                    },
                    {
                      "title": "Monitored Database",
                      "value": "{{ inputs.target_database }}",
                      "short": true
                    },
                    {
                      "title": "Severity",
                      "value": "WARNING - Active Changes Pending",
                      "short": true
                    },
                    {
                      "title": "Auto Compact",
                      "value": "{{ inputs.auto_compact ? 'Enabled' : 'Disabled' }}",
                      "short": true
                    }
                  ],
                  "footer": "Kestra CouchDB Sentinel | Execution ID: {{ execution.id }}"
                }
              ]
            }

      PROBE_FAILED:
        - id: alert_probe_failure
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {
              "channel": "{{ inputs.slack_channel }}",
              "text": "[CRITICAL] CouchDB Sentinel Probe Failed - Unable to query /_active_tasks endpoint. Verify network routing and credentials."
            }

      HEALTHY:
        - id: log_tasks_healthy
          type: io.kestra.plugin.core.log.Log
          message: "CouchDB active tasks health is normal. All replication and compaction
            tasks operating within thresholds."

  - id: log_audit_summary
    type: io.kestra.plugin.core.log.Log
    message: "CouchDB active task health audit completed. Status: {{
      outputs.route_by_task_health.value ?? 'PROCESSED' }}."

errors:
  - id: on_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "channel": "{{ inputs.slack_channel }}",
        "text": "[ERROR] CouchDB Replication Sentinel flow execution aborted unexpectedly. Execution ID: {{ execution.id }}."
      }