OceanBase Major Compaction Stall & Auto-Resume Sentinel

id: oceanbase-major-compaction-stall-sentinel
namespace: company.data

inputs:
  - id: oceanbase_host
    type: STRING
    displayName: OceanBase Host
    defaults: "oceanbase.internal"
    description: Hostname or IP address of the OBServer or ODP proxy exposing port 2881.

  - id: oceanbase_port
    type: INT
    displayName: OceanBase Port
    defaults: 2881
    description: Port number for the MySQL wire protocol interface on OceanBase
      (default 2881).

  - id: oceanbase_database
    type: STRING
    displayName: Database Name
    defaults: "oceanbase"
    description: Database name to establish the session against (default oceanbase).

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

  - id: auto_resume
    type: BOOL
    displayName: Auto Resume Merge
    defaults: true
    description: Automatically dispatch ALTER SYSTEM RESUME MERGE when suspended
      compaction is detected.

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

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

tasks:
  - id: probe_major_compaction_health
    type: io.kestra.plugin.jdbc.mysql.Query
    url: "jdbc:mysql://{{ inputs.oceanbase_host }}:{{ inputs.oceanbase_port }}/{{
      inputs.oceanbase_database }}"
    username: "{{ secret('OCEANBASE_USERNAME') }}"
    password: "{{ secret('OCEANBASE_PASSWORD') }}"
    fetchType: FETCH
    allowFailure: true
    sql: |
      SELECT
        tenant_id,
        status,
        is_suspended,
        is_error,
        COALESCE(info, 'None') AS error_info
      FROM oceanbase.CDB_OB_MAJOR_COMPACTION
      WHERE is_suspended = 'YES' OR is_error = 'YES' OR status = 'SUSPENDED'
      ORDER BY tenant_id ASC;

  - id: route_by_compaction_health
    type: io.kestra.plugin.core.flow.Switch
    value: >-
      {% if outputs.probe_major_compaction_health.rows is null %} PROBE_FAILED
      {% else %}
        {% set error_count = 0 %}
        {% set suspended_count = 0 %}
        {% for row in outputs.probe_major_compaction_health.rows %}
          {% if row.is_error == 'YES' %}
            {% set error_count = error_count + 1 %}
          {% elseif row.is_suspended == 'YES' or row.status == 'SUSPENDED' %}
            {% set suspended_count = suspended_count + 1 %}
          {% endif %}
        {% endfor %}
        {% if error_count > 0 or suspended_count > 0 %}
        CRITICAL
        {% 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] OceanBase Major Compaction Suspended or Error Detected - MemTable Exhaustion Risk",
              "attachments": [
                {
                  "color": "#D32F2F",
                  "title": "OceanBase Major Compaction Failure Incident",
                  "fields": [
                    {
                      "title": "Cluster Host",
                      "value": "{{ inputs.oceanbase_host }}:{{ inputs.oceanbase_port }}",
                      "short": true
                    },
                    {
                      "title": "Degraded Tenants Count",
                      "value": "{{ outputs.probe_major_compaction_health.size }}",
                      "short": true
                    },
                    {
                      "title": "Severity",
                      "value": "CRITICAL - LSM Compaction Halted",
                      "short": true
                    },
                    {
                      "title": "Remediation Status",
                      "value": "{{ (inputs.auto_resume and not inputs.dry_run) ? 'Dispatching ALTER SYSTEM RESUME MERGE' : 'Remediation Skipped (Dry Run or Disabled)' }}",
                      "short": true
                    }
                  ],
                  "footer": "Kestra OceanBase Sentinel | Execution ID: {{ execution.id }}"
                }
              ]
            }

        - id: check_remediation
          type: io.kestra.plugin.core.flow.If
          condition: "{{ inputs.auto_resume == true and inputs.dry_run == false }}"
          then:
            - id: execute_resume_merge
              type: io.kestra.plugin.jdbc.mysql.Query
              url: "jdbc:mysql://{{ inputs.oceanbase_host }}:{{ inputs.oceanbase_port }}/{{
                inputs.oceanbase_database }}"
              username: "{{ secret('OCEANBASE_USERNAME') }}"
              password: "{{ secret('OCEANBASE_PASSWORD') }}"
              sql: "ALTER SYSTEM RESUME MERGE TENANT = all_user;"

            - id: notify_remediation_slack
              type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
              url: "{{ secret('SLACK_WEBHOOK_URL') }}"
              payload: |
                {
                  "channel": "{{ inputs.slack_channel }}",
                  "text": "[REMEDIATION] OceanBase ALTER SYSTEM RESUME MERGE TENANT = all_user command executed successfully."
                }

      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] OceanBase Sentinel Probe Failed - Unable to query oceanbase.CDB_OB_MAJOR_COMPACTION. Verify port 2881 routing and user privileges."
            }

      HEALTHY:
        - id: log_compaction_healthy
          type: io.kestra.plugin.core.log.Log
          message: "OceanBase major compaction health is normal. No suspended or failed
            compactions detected across user tenants."

  - id: log_audit_summary
    type: io.kestra.plugin.core.log.Log
    message: "OceanBase major compaction audit completed. Status: {{
      outputs.route_by_compaction_health.value ?? 'PROCESSED' }}. Evaluated
      degraded compaction count: {{ outputs.probe_major_compaction_health.size
      ?? 0 }}."

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