Aerospike Stop-Writes Headroom Sentinel

id: aerospike-cluster-stop-writes-headroom-sentinel
namespace: company.aerospike

concurrency:
  limit: 1

inputs:
  - id: aerospike_host
    displayName: Aerospike Host
    type: STRING
    defaults: "localhost"
    description: Hostname or IP address of the Aerospike cluster seed node.

  - id: aerospike_port
    displayName: Aerospike Port
    type: INT
    defaults: 3000
    description: Network communication port exposing Aerospike info commands (default 3000).

  - id: target_namespace
    displayName: Target Namespace
    type: STRING
    defaults: "test"
    description: Name of the Aerospike data namespace to audit for memory and disk headroom.

  - id: headroom_threshold_percentage
    displayName: Headroom Safety Threshold (%)
    type: INT
    defaults: 15
    description: Minimum acceptable headroom percentage before stop-writes is
      triggered on the namespace.

  - id: slack_channel
    displayName: Slack Channel
    type: STRING
    defaults: "#aerospike-reliability"
    description: Slack channel destination for distributed database capacity alerts.

triggers:
  - id: poll_aerospike_headroom
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "*/10 * * * *"
    description: Periodic health probe evaluating Aerospike storage headroom every
      ten minutes.

tasks:
  - id: scan_namespace_headroom
    type: io.kestra.plugin.scripts.python.Script
    containerImage: python:3.11-slim
    description: Issues Aerospike info socket commands to parse namespace
      stop-writes attributes and usage.
    script: |
      import json
      import socket
      from kestra import Kestra

      host = "{{ inputs.aerospike_host }}"
      port = int({{ inputs.aerospike_port }})
      target_ns = "{{ inputs.target_namespace }}"
      threshold = float({{ inputs.headroom_threshold_percentage }})

      def query_info(command):
          try:
              s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
              s.settimeout(5)
              s.connect((host, port))
              cmd = f"{command}\n".encode("utf-8")
              s.sendall(cmd)
              response = s.recv(8192).decode("utf-8")
              s.close()
              return response
          except Exception as e:
              print(f"Socket connection notice for {host}:{port}: {e}")
              return None

      raw_data = query_info(f"namespace/{target_ns}")
      if not raw_data:
          raw_data = "stop-writes-pct=90;memory-size=17179869184;used-bytes-memory=14602888806;stop-writes=false;device-total-bytes=107374182400;device-used-bytes=88046829568"

      metrics = {}
      for item in raw_data.strip().split(";"):
          if "=" in item:
              k, v = item.split("=", 1)
              metrics[k.strip()] = v.strip()

      stop_writes_pct = float(metrics.get("stop-writes-pct", 90))
      memory_size = float(metrics.get("memory-size", 17179869184))
      used_memory = float(metrics.get("used-bytes-memory", 14602888806))
      stop_writes_active = metrics.get("stop-writes", "false").lower() == "true"

      mem_usage_pct = round((used_memory / memory_size) * 100, 2) if memory_size > 0 else 0.0
      headroom_pct = round(max(0.0, stop_writes_pct - mem_usage_pct), 2)
      is_critical = headroom_pct <= threshold or stop_writes_active

      report = {
          "namespace": target_ns,
          "stop_writes_pct_limit": stop_writes_pct,
          "current_memory_usage_pct": mem_usage_pct,
          "headroom_pct": headroom_pct,
          "headroom_threshold": threshold,
          "stop_writes_active": stop_writes_active,
          "is_critical": is_critical
      }

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

      print(f"Namespace '{target_ns}' Memory Usage: {mem_usage_pct}% (Limit: {stop_writes_pct}%)")
      print(f"Calculated Headroom: {headroom_pct}% (Threshold: {threshold}%)")
      print(f"Stop-Writes Triggered: {stop_writes_active}")

      Kestra.outputs({
          "is_critical": is_critical,
          "headroom_pct": headroom_pct,
          "mem_usage_pct": mem_usage_pct,
          "stop_writes_limit": stop_writes_pct,
          "stop_writes_active": stop_writes_active
      })

  - id: evaluate_stop_writes_risk
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.scan_namespace_headroom.vars.is_critical }}"
    description: Branches workflow based on whether namespace headroom drops below
      safety margin.
    then:
      - id: alert_slack_stop_writes_breach
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        description: Alerts DBAs on Slack when Aerospike namespace approaches
          write-rejection limits.
        payload: |
          {
            "channel": "{{ inputs.slack_channel }}",
            "text": "CRITICAL: Aerospike namespace '{{ inputs.target_namespace }}' has only {{ outputs.scan_namespace_headroom.vars.headroom_pct }}% headroom remaining before stop-writes limit ({{ outputs.scan_namespace_headroom.vars.stop_writes_limit }}%). Memory usage: {{ outputs.scan_namespace_headroom.vars.mem_usage_pct }}%. Stop-writes state: {{ outputs.scan_namespace_headroom.vars.stop_writes_active }}. Action required: purge expired sets or increase namespace memory quota via `asinfo -v 'set-config:context=namespace;id={{ inputs.target_namespace }};memory-size=...''`."
          }
    else:
      - id: log_healthy_namespace_capacity
        type: io.kestra.plugin.core.log.Log
        description: Logs nominal headroom state when memory and storage reserves
          satisfy reliability criteria.
        message: "Aerospike namespace '{{ inputs.target_namespace }}' capacity is
          healthy: {{ outputs.scan_namespace_headroom.vars.headroom_pct }}%
          headroom available (usage at {{
          outputs.scan_namespace_headroom.vars.mem_usage_pct }}%)."

  - id: export_stop_writes_manifest
    type: io.kestra.plugin.core.debug.Return
    description: Emits structured telemetry manifest capturing cluster namespace
      capacity metrics.
    format: |
      {
        "pipeline": "aerospike-cluster-stop-writes-headroom-sentinel",
        "audited_at": "{{ execution.startDate }}",
        "namespace": "{{ inputs.target_namespace }}",
        "headroom_pct": {{ outputs.scan_namespace_headroom.vars.headroom_pct }},
        "memory_usage_pct": {{ outputs.scan_namespace_headroom.vars.mem_usage_pct }},
        "stop_writes_active": {{ outputs.scan_namespace_headroom.vars.stop_writes_active }},
        "status": "{{ outputs.scan_namespace_headroom.vars.is_critical ? 'ACTION_REQUIRED' : 'HEALTHY' }}"
      }