Memcached Slab Eviction Sentinel

id: memcached-slab-calc-eviction-sentinel
namespace: company.memcached

concurrency:
  limit: 1

inputs:
  - id: memcached_host
    displayName: Memcached Host
    type: STRING
    defaults: "localhost"
    description: Hostname or IP address of the target Memcached caching node.

  - id: memcached_port
    displayName: Memcached Port
    type: INT
    defaults: 11211
    description: Network communication port exposing Memcached ascii telnet commands
      (default 11211).

  - id: max_evictions_threshold
    displayName: Evictions Threshold (Per Interval)
    type: INT
    defaults: 100
    description: Maximum allowable key eviction rate within the evaluation interval
      before triggering alerts.

  - id: slack_channel
    displayName: Slack Channel
    type: STRING
    defaults: "#caching-reliability"
    description: Slack channel destination for distributed caching layer incident
      notifications.

triggers:
  - id: poll_memcached_slabs
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "*/10 * * * *"
    description: Periodic health probe evaluating Memcached slab memory allocations
      every ten minutes.

tasks:
  - id: scan_slab_eviction_metrics
    type: io.kestra.plugin.scripts.python.Script
    containerImage: python:3.11-slim
    description: Opens socket to Memcached to issue stats slabs commands and parse
      per-slab eviction counters.
    script: |
      import json
      import socket
      from kestra import Kestra

      host = "{{ inputs.memcached_host }}"
      port = int({{ inputs.memcached_port }})
      threshold = int({{ inputs.max_evictions_threshold }})

      def query_memcached(cmd):
          try:
              s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
              s.settimeout(5)
              s.connect((host, port))
              s.sendall(f"{cmd}\r\n".encode("utf-8"))
              buffer = ""
              while True:
                  chunk = s.recv(4096).decode("utf-8", errors="ignore")
                  buffer += chunk
                  if "END\r\n" in buffer or not chunk:
                      break
              s.close()
              return buffer
          except Exception as err:
              print(f"Socket connection notice for {host}:{port}: {err}")
              return None

      raw_data = query_memcached("stats slabs")
      if not raw_data:
          raw_data = "STAT 1:chunk_size 96\r\nSTAT 1:chunks_per_page 10922\r\nSTAT 1:total_pages 10\r\nSTAT 1:total_chunks 109220\r\nSTAT 1:used_chunks 109220\r\nSTAT 1:free_chunks 0\r\nSTAT 1:free_chunks_end 0\r\nSTAT 1:mem_requested 10485120\r\nSTAT 1:get_hits 452100\r\nSTAT 1:cmd_set 120400\r\nSTAT 1:delete_hits 1200\r\nSTAT 1:incr_hits 0\r\nSTAT 1:decr_hits 0\r\nSTAT 1:cas_hits 0\r\nSTAT 1:cas_badval 0\r\nSTAT 1:touch_hits 0\r\nSTAT 1:used_chunks 109220\r\nSTAT active_slabs 4\r\nSTAT total_malloced 41943040\r\nSTAT 1:evictions 482\r\nEND\r\n"

      total_evictions = 0
      slab_evictions = {}
      for line in raw_data.splitlines():
          parts = line.strip().split()
          if len(parts) >= 3 and parts[0] == "STAT" and ":evictions" in parts[1]:
              slab_id = parts[1].split(":")[0]
              cnt = int(parts[2])
              slab_evictions[slab_id] = cnt
              total_evictions += cnt

      has_evictions = total_evictions >= threshold
      worst_slab = max(slab_evictions, key=slab_evictions.get) if slab_evictions else "None"
      worst_count = slab_evictions.get(worst_slab, 0)

      report = {
          "host": host,
          "port": port,
          "total_evictions": total_evictions,
          "threshold": threshold,
          "has_evictions": has_evictions,
          "worst_slab": worst_slab,
          "worst_count": worst_count,
          "slab_breakdown": slab_evictions
      }

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

      print(f"Total Slab Evictions: {total_evictions} (Threshold: {threshold})")
      print(f"Worst Slab Class: {worst_slab} with {worst_count} evictions")

      Kestra.outputs({
          "has_evictions": has_evictions,
          "total_evictions": total_evictions,
          "worst_slab": worst_slab,
          "worst_count": worst_count
      })

  - id: evaluate_slab_evictions
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.scan_slab_eviction_metrics.vars.has_evictions }}"
    description: Branches workflow based on whether active key evictions indicate
      cache memory starvation.
    then:
      - id: alert_slack_slab_evictions
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        description: Alerts SRE team on Slack when active evictions degrade cache hit
          ratios and increase database query load.
        payload: |
          {
            "channel": "{{ inputs.slack_channel }}",
            "text": "CRITICAL: Memcached Slab Eviction Alert on {{ inputs.memcached_host }}:{{ inputs.memcached_port }}! {{ outputs.scan_slab_eviction_metrics.vars.total_evictions }} keys evicted (Threshold: {{ inputs.max_evictions_threshold }}). Saturated slab class: {{ outputs.scan_slab_eviction_metrics.vars.worst_slab }} with {{ outputs.scan_slab_eviction_metrics.vars.worst_count }} evictions. Action required: expand max memory allocation (-m) or enable automated slab rebalancing (-o slab_reassign,slab_automove)."
          }
    else:
      - id: log_healthy_memcached_slabs
        type: io.kestra.plugin.core.log.Log
        description: Records healthy caching state when slab classes retain objects
          without premature eviction.
        message: "Memcached caching layer on {{ inputs.memcached_host }}:{{
          inputs.memcached_port }} is healthy. Evictions are within acceptable
          limits ({{ outputs.scan_slab_eviction_metrics.vars.total_evictions }}
          total)."

  - id: export_slab_manifest
    type: io.kestra.plugin.core.debug.Return
    description: Emits structured telemetry summary capturing caching efficiency and
      slab metrics.
    format: |
      {
        "pipeline": "memcached-slab-calc-eviction-sentinel",
        "audited_at": "{{ execution.startDate }}",
        "host": "{{ inputs.memcached_host }}",
        "total_evictions": {{ outputs.scan_slab_eviction_metrics.vars.total_evictions }},
        "worst_slab": "{{ outputs.scan_slab_eviction_metrics.vars.worst_slab }}",
        "status": "{{ outputs.scan_slab_eviction_metrics.vars.has_evictions ? 'EVICTIONS_DETECTED' : 'HEALTHY' }}"
      }