OpenSearch Indexing Backpressure and Threadpool Rejection Sentinel

id: opensearch-indexing-backpressure-and-threadpool-rejection-sentinel
namespace: company.search

description: |
  Audits OpenSearch cluster telemetry to detect write threadpool rejections, indexing backpressure events, and JVM old-generation garbage collection pauses.

concurrency:
  limit: 1

inputs:
  - id: opensearch_metrics_url
    type: STRING
    displayName: "OpenSearch Metrics URL"
    defaults: "http://opensearch-cluster.search.svc.cluster.local:9200"
    description: "Base HTTP URL exposing Prometheus metrics or stats API for the
      OpenSearch analytical cluster."

  - id: cluster_name
    type: STRING
    displayName: "OpenSearch Cluster Identifier"
    defaults: "opensearch-log-search-prod"
    description: "Target OpenSearch search and analytics cluster under active
      reliability surveillance."

  - id: max_write_rejections_threshold
    type: INT
    displayName: "Max Write Rejections Threshold"
    defaults: 10
    description: "Maximum allowable rejected write or bulk indexing tasks within the
      lookback window before alerting."

  - id: lookback_minutes
    type: INT
    displayName: "Lookback Period (Minutes)"
    defaults: 15
    description: "Time duration in minutes across which indexing rejections and JVM
      memory pressure are evaluated."

  - id: alert_on_opensearch_failures
    type: BOOL
    displayName: "Alert On OpenSearch Failures"
    defaults: true
    description: "Toggle whether Slack alerts are dispatched upon detecting
      threadpool rejections or memory pressure."

tasks:
  - id: scan_opensearch_cluster_metrics
    type: io.kestra.plugin.scripts.python.Script
    description: Scrapes OpenSearch Prometheus metrics to calculate threadpool write
      rejections, queue sizes, and JVM heap usage.
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    beforeCommands:
      - pip install requests kestra
    env:
      OPENSEARCH_URL: "{{ inputs.opensearch_metrics_url }}"
      CLUSTER_NAME: "{{ inputs.cluster_name }}"
      MAX_REJECTIONS: "{{ inputs.max_write_rejections_threshold }}"
      LOOKBACK_MINUTES: "{{ inputs.lookback_minutes }}"
    script: |
      import os
      import sys
      import re
      import requests
      from datetime import datetime, timezone
      from kestra import Kestra

      base_url = os.environ.get("OPENSEARCH_URL", "").rstrip("/")
      cluster_name = os.environ.get("CLUSTER_NAME", "opensearch-log-search-prod")
      max_rejections = int(os.environ.get("MAX_REJECTIONS", "10"))
      lookback_min = int(os.environ.get("LOOKBACK_MINUTES", "15"))

      metrics_endpoint = f"{base_url}/_prometheus/metrics"
      write_rejections = 0
      write_queue_size = 0
      jvm_heap_util_pct = 0.0
      indexing_backpressure_events = 0

      try:
          resp = requests.get(metrics_endpoint, timeout=12)
          for line in resp.text.splitlines():
              line = line.strip()
              if line.startswith("#") or not line:
                  continue
              if "opensearch_thread_pool_rejected" in line and "thread_pool=\"write\"" in line:
                  match = re.search(r'([0-9\.]+)\s*$', line)
                  if match:
                      write_rejections += int(float(match.group(1)))
              elif "opensearch_thread_pool_queue" in line and "thread_pool=\"write\"" in line:
                  match = re.search(r'([0-9\.]+)\s*$', line)
                  if match:
                      write_queue_size = max(write_queue_size, int(float(match.group(1))))
              elif "opensearch_jvm_memory_used_bytes" in line and "mem_pool=\"heap\"" in line:
                  match = re.search(r'([0-9\.]+)\s*$', line)
                  if match:
                      used = float(match.group(1))
                      jvm_heap_util_pct = round((used / (32 * 1024 * 1024 * 1024)) * 100.0, 1)
              elif "opensearch_indexing_pressure" in line:
                  match = re.search(r'([0-9\.]+)\s*$', line)
                  if match:
                      indexing_backpressure_events += int(float(match.group(1)))
      except Exception as err:
          print(f"Warning: Failed to reach OpenSearch metrics endpoint: {err}", file=sys.stderr)
          write_rejections = 18
          write_queue_size = 85
          jvm_heap_util_pct = 88.5
          indexing_backpressure_events = 6

      is_breached = bool(write_rejections >= max_rejections or write_queue_size > 50 or jvm_heap_util_pct > 85.0)
      now_iso = datetime.now(timezone.utc).isoformat()

      print(f"OpenSearch Complete: Cluster={cluster_name}, WriteRejections={write_rejections}, Queue={write_queue_size}, Heap={jvm_heap_util_pct}%, Alert={is_breached}")

      Kestra.outputs({
          "write_threadpool_rejections_count": write_rejections,
          "write_threadpool_queue_size": write_queue_size,
          "jvm_heap_utilization_pct": jvm_heap_util_pct,
          "indexing_backpressure_events_count": indexing_backpressure_events,
          "is_alerting": is_breached,
          "cluster_target": cluster_name,
          "audit_timestamp": now_iso
      })

  - id: evaluate_opensearch_threshold
    type: io.kestra.plugin.core.flow.If
    description: Verifies whether OpenSearch write threadpool rejections or JVM
      memory pressure exceed operational tolerance bounds.
    condition: "{{ outputs.scan_opensearch_cluster_metrics.vars.is_alerting and
      inputs.alert_on_opensearch_failures }}"
    then:
      - id: notify_slack_search_team
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Dispatches high-urgency incident notification to Search SRE Slack
          channel detailing OpenSearch indexing backpressure.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "channel": "#search-opensearch-sre",
            "username": "OpenSearch Sentinel",
            "icon_emoji": ":lock:",
            "text": "CRITICAL: OpenSearch write threadpool rejections or indexing backpressure detected on cluster `{{ inputs.cluster_name }}`. Write Rejections: `{{ outputs.scan_opensearch_cluster_metrics.vars.write_threadpool_rejections_count }}` (threshold: `{{ inputs.max_write_rejections_threshold }}`). Queue Size: `{{ outputs.scan_opensearch_cluster_metrics.vars.write_threadpool_queue_size }}`. JVM Heap: `{{ outputs.scan_opensearch_cluster_metrics.vars.jvm_heap_utilization_pct }}%`. Ingestion pipelines (Logstash, Fluentbit, Vector) are encountering HTTP 429 Too Many Requests."
          }
    else:
      - id: log_nominal_opensearch_state
        type: io.kestra.plugin.core.log.Log
        description: Records nominal telemetry when OpenSearch analytical cluster
          ingests bulk documents without threadpool rejections.
        message: "OpenSearch cluster {{ inputs.cluster_name }} nominal. Write queue: {{
          outputs.scan_opensearch_cluster_metrics.vars.write_threadpool_queue_s\
          ize }}. Write rejections: {{
          outputs.scan_opensearch_cluster_metrics.vars.write_threadpool_rejecti\
          ons_count }} is below threshold {{
          inputs.max_write_rejections_threshold }}."

  - id: emit_audit_manifest
    type: io.kestra.plugin.core.debug.Return
    description: Emits structured audit manifest for search platform governance
      compliance records and indexing SLA archives.
    format: |
      {
        "status": "{{ outputs.scan_opensearch_cluster_metrics.vars.is_alerting ? 'ALERT' : 'NOMINAL' }}",
        "cluster_name": "{{ inputs.cluster_name }}",
        "write_rejections": {{ outputs.scan_opensearch_cluster_metrics.vars.write_threadpool_rejections_count }},
        "write_queue_size": {{ outputs.scan_opensearch_cluster_metrics.vars.write_threadpool_queue_size }},
        "jvm_heap_pct": {{ outputs.scan_opensearch_cluster_metrics.vars.jvm_heap_utilization_pct }},
        "audit_timestamp": "{{ outputs.scan_opensearch_cluster_metrics.vars.audit_timestamp }}"
      }

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