Meilisearch Indexing Queue Staleness Sentinel

id: meilisearch-indexing-queue-staleness-sentinel
namespace: company.search

description: |
  Audits Meilisearch fast search engines to detect task queue accumulation, batch indexing stalls,
  and document ingestion bottlenecks before search query freshness is compromised.

concurrency:
  limit: 1

inputs:
  - id: meilisearch_host
    type: STRING
    displayName: Meilisearch Host URI
    defaults: "https://search.internal.company.com:7700"
    description: Base HTTP endpoint of the Meilisearch cluster or cloud instance.

  - id: max_enqueued_tasks_threshold
    type: INT
    displayName: Max Enqueued Tasks Threshold
    defaults: 100
    description: Maximum allowable pending and processing indexing tasks before
      flagging an ingestion stall.

  - id: max_task_age_seconds_threshold
    type: INT
    displayName: Max Task Age Threshold (Seconds)
    defaults: 300
    description: Maximum allowable duration in seconds for the oldest processing
      task before declaring a stuck pipeline.

  - id: target_index_uid
    type: STRING
    displayName: Target Index UID
    defaults: "all"
    description: Specific index UID to monitor, or all to audit all active indexes
      across the cluster.

  - id: alert_on_staleness
    type: BOOL
    displayName: Alert on Staleness
    defaults: true
    description: Whether to dispatch automated Slack notifications when indexing
      stalls or backlogs are detected.

tasks:
  - id: evaluate_indexing_queue
    type: io.kestra.plugin.scripts.python.Script
    description: Queries Meilisearch Tasks API for enqueued and processing document
      tasks to measure backlog and latency.
    containerImage: python:3.11-slim
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    beforeCommands:
      - pip install requests kestra
    env:
      MEILISEARCH_MASTER_KEY: "{{ secret('MEILISEARCH_MASTER_KEY') }}"
    script: |
      import os
      import json
      import requests
      from datetime import datetime, timezone
      from kestra import Kestra

      host = "{{ inputs.meilisearch_host }}".rstrip('/')
      api_key = os.environ.get("MEILISEARCH_MASTER_KEY", "")
      max_tasks_threshold = int("{{ inputs.max_enqueued_tasks_threshold }}")
      max_age_threshold = int("{{ inputs.max_task_age_seconds_threshold }}")
      target_index = "{{ inputs.target_index_uid }}".strip().lower()

      headers = {
          "Authorization": f"Bearer {api_key}",
          "Content-Type": "application/json"
      }

      params = {
          "statuses": "enqueued,processing",
          "limit": 100
      }
      if target_index != "all":
          params["indexUids"] = target_index

      now = datetime.now(timezone.utc)
      pending_tasks = []
      oldest_task_age = 0
      is_stalled = False

      try:
          resp = requests.get(f"{host}/tasks", headers=headers, params=params, timeout=15)
          resp.raise_for_status()
          data = resp.json()
          results = data.get("results", [])
          total_enqueued = data.get("total", len(results))

          for task in results:
              enqueued_at_str = task.get("enqueuedAt")
              task_age = 0
              if enqueued_at_str:
                  enqueued_dt = datetime.fromisoformat(enqueued_at_str.replace("Z", "+00:00"))
                  task_age = int((now - enqueued_dt).total_seconds())
                  if task_age > oldest_task_age:
                      oldest_task_age = task_age

              pending_tasks.append({
                  "uid": task.get("uid"),
                  "index_uid": task.get("indexUid"),
                  "type": task.get("type"),
                  "status": task.get("status"),
                  "age_seconds": task_age,
                  "started_at": task.get("startedAt")
              })

          if total_enqueued >= max_tasks_threshold or oldest_task_age >= max_age_threshold:
              is_stalled = True

          Kestra.outputs({
              "total_pending_tasks": total_enqueued,
              "oldest_task_age_seconds": oldest_task_age,
              "is_stalled": is_stalled,
              "cluster_host": host,
              "tasks_sample": json.dumps(pending_tasks[:10])
          })

      except Exception as e:
          print(f"Error querying Meilisearch cluster {host}: {e}")
          Kestra.outputs({
              "total_pending_tasks": 0,
              "oldest_task_age_seconds": 0,
              "is_stalled": True,
              "cluster_host": host,
              "error_message": str(e),
              "tasks_sample": "[]"
          })

  - id: check_indexing_staleness
    type: io.kestra.plugin.core.flow.If
    description: Evaluates whether Meilisearch task queue exceeds pending limits or
      task age thresholds.
    condition: "{{ outputs.evaluate_indexing_queue.vars.is_stalled and
      inputs.alert_on_staleness }}"
    then:
      - id: notify_slack_indexing_stall
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Dispatches high-priority incident notifications to Slack when
          search task backpressure occurs.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "channel": "#search-platform-alerts",
            "username": "Meilisearch-Sentinel",
            "text": "CRITICAL: Meilisearch indexing queue stall detected on host `{{ outputs.evaluate_indexing_queue.vars.cluster_host }}`. Total pending tasks: `{{ outputs.evaluate_indexing_queue.vars.total_pending_tasks }}` (Threshold: {{ inputs.max_enqueued_tasks_threshold }}), Oldest task age: `{{ outputs.evaluate_indexing_queue.vars.oldest_task_age_seconds }}s` (Threshold: {{ inputs.max_task_age_seconds_threshold }}s)."
          }

  - id: log_indexing_status
    type: io.kestra.plugin.core.log.Log
    description: Records an operational audit trail of Meilisearch task queue
      evaluation results.
    message: "Meilisearch audit completed for '{{ inputs.meilisearch_host }}'.
      Pending: {{ outputs.evaluate_indexing_queue.vars.total_pending_tasks }},
      Oldest Age: {{
      outputs.evaluate_indexing_queue.vars.oldest_task_age_seconds }}s, Stalled:
      {{ outputs.evaluate_indexing_queue.vars.is_stalled }}."

  - id: export_indexing_manifest
    type: io.kestra.plugin.core.debug.Return
    description: Emits structured manifest of Meilisearch indexing queue diagnostics
      for monitoring and data lineage.
    format:
      host: "{{ outputs.evaluate_indexing_queue.vars.cluster_host }}"
      target_index: "{{ inputs.target_index_uid }}"
      pending_tasks: "{{ outputs.evaluate_indexing_queue.vars.total_pending_tasks }}"
      oldest_task_age_seconds: "{{ outputs.evaluate_indexing_queue.vars.oldest_task_age_seconds }}"
      is_stalled: "{{ outputs.evaluate_indexing_queue.vars.is_stalled }}"
      sample_tasks: "{{ outputs.evaluate_indexing_queue.vars.tasks_sample }}"

triggers:
  - id: ten_minute_meilisearch_audit
    type: io.kestra.plugin.core.trigger.Schedule
    description: Automated schedule trigger auditing Meilisearch indexing queue
      every ten minutes.
    cron: "*/10 * * * *"