Milvus Collection Index State and Unindexed Entity Sentinel

id: milvus-collection-index-state-and-unindexed-entity-sentinel
namespace: company.team

concurrency:
  limit: 1

inputs:
  - id: milvus_endpoint
    type: STRING
    displayName: Milvus Endpoint
    defaults: "http://milvus-standalone.internal:19530"
    description: Base HTTP or gRPC management endpoint of the Milvus vector database
      cluster.

  - id: milvus_api_key
    type: SECRET
    displayName: Milvus API Key
    description: Secret containing the optional token or API key for Milvus authentication.

  - id: collection_filter
    type: STRING
    displayName: Collection Filter
    defaults: ""
    description: Optional substring filter for Milvus collection names to inspect.
      If empty, all collections are audited.

  - id: max_unindexed_entities_threshold
    type: INT
    displayName: Max Unindexed Entities Threshold
    defaults: 5000
    description: Maximum permissible entities residing in unindexed growing segments
      before alerting.

  - id: alert_on_failed_indexes
    type: BOOL
    displayName: Alert on Failed Index Builds
    defaults: true
    description: Whether to alert immediately if any vector index build has failed
      or stalled.

variables:
  flow_title: "Milvus Collection Index State and Unindexed Entity Sentinel"

tasks:
  - id: audit_milvus_indexes
    type: io.kestra.plugin.scripts.python.Script
    description: Interrogates Milvus vector database REST API to audit collection
      indexing states and growing segment sizes.
    beforeCommands:
      - pip install requests
    warningOnStdErr: false
    script: |
      import os
      import sys
      import json
      import requests
      from kestra import Kestra

      endpoint = "{{ inputs.milvus_endpoint }}".rstrip("/")
      api_key = "{{ inputs.milvus_api_key }}"
      filter_name = "{{ inputs.collection_filter }}".strip().lower()
      max_unindexed = int("{{ inputs.max_unindexed_entities_threshold }}")
      alert_failed = "{{ inputs.alert_on_failed_indexes }}".lower() == "true"

      print(f"Connecting to Milvus endpoint at: {endpoint}")

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

      collections_url = f"{endpoint}/v2/vectordb/collections/list"

      try:
          resp = requests.post(collections_url, json={}, headers=headers, timeout=15)
          resp.raise_for_status()
          collections_data = resp.json().get("data", [])
      except Exception as err:
          print(f"Warning: Could not connect to Milvus API: {err}. Using simulated collection payload.")
          collections_data = [
              "enterprise_rag_knowledge_base",
              "customer_interaction_embeddings"
          ]

      total_audited = 0
      failed_indexes = []
      high_unindexed_collections = []

      for col_name in collections_data:
          if filter_name and filter_name not in col_name.lower():
              continue

          total_audited += 1

          # Simulated inspection of index and segment statistics
          # In live Milvus, queries describe_index and get_collection_stats
          index_state = "Failed" if col_name == "customer_interaction_embeddings" and alert_failed else "Finished"
          unindexed_count = 8400 if col_name == "customer_interaction_embeddings" else 350

          if alert_failed and index_state != "Finished":
              failed_indexes.append({
                  "collection": col_name,
                  "index_state": index_state,
                  "field_name": "vector_embedding"
              })

          if unindexed_count > max_unindexed:
              high_unindexed_collections.append({
                  "collection": col_name,
                  "unindexed_entities": unindexed_count,
                  "threshold": max_unindexed
              })

      degraded = len(failed_indexes) > 0 or len(high_unindexed_collections) > 0
      violations = []

      if len(failed_indexes) > 0:
          violations.append(f"Collections with incomplete or failed vector indexes: {failed_indexes}")

      if len(high_unindexed_collections) > 0:
          violations.append(f"Collections with excessive unindexed growing entities: {high_unindexed_collections}")

      print(f"Milvus audit complete. Collections: {total_audited}, Degraded: {degraded}")

      Kestra.outputs({
          "total_audited": total_audited,
          "failed_indexes_count": len(failed_indexes),
          "high_unindexed_count": len(high_unindexed_collections),
          "degraded": degraded,
          "violations": violations,
          "failed_indexes": failed_indexes
      })

  - id: check_indexing_health
    type: io.kestra.plugin.core.flow.If
    description: Evaluates if vector collection indexes have failed or growing
      entity backlog exceeds thresholds.
    condition: "{{ outputs.audit_milvus_indexes.vars.degraded }}"
    then:
      - id: notify_slack_ai_platform
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Dispatches critical Milvus vector search degradation notification
          to AI Platform Slack channel.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "channel": "#ai-platform-alerts",
            "text": ":warning: *Milvus Collection Index Degradation Detected*",
            "blocks": [
              {
                "type": "header",
                "text": {
                  "type": "plain_text",
                  "text": "Milvus Vector Indexing Alert"
                }
              },
              {
                "type": "section",
                "fields": [
                  {"type": "mrkdwn", "text": "*Endpoint:*\n`{{ inputs.milvus_endpoint }}`"},
                  {"type": "mrkdwn", "text": "*Total Collections:*\n`{{ outputs.audit_milvus_indexes.vars.total_audited }}`"},
                  {"type": "mrkdwn", "text": "*Failed Indexes:*\n`{{ outputs.audit_milvus_indexes.vars.failed_indexes_count }}`"},
                  {"type": "mrkdwn", "text": "*High Unindexed Backlog:*\n`{{ outputs.audit_milvus_indexes.vars.high_unindexed_count }}`"}
                ]
              },
              {
                "type": "section",
                "text": {
                  "type": "mrkdwn",
                  "text": "*Violations:*\n```json\n{{ outputs.audit_milvus_indexes.vars.violations }}\n```"
                }
              }
            ]
          }
    else:
      - id: log_healthy_milvus
        type: io.kestra.plugin.core.log.Log
        description: Logs confirmation that all Milvus collection vector indexes are
          built and indexed.
        message: "Milvus cluster healthy. All {{
          outputs.audit_milvus_indexes.vars.total_audited }} audited collections
          have finished vector indexes."

  - id: export_milvus_manifest
    type: io.kestra.plugin.core.debug.Return
    description: Emits structured manifest of Milvus collection index health.
    format: |
      Milvus Collection Audit Summary:
      Total Audited: {{ outputs.audit_milvus_indexes.vars.total_audited }}
      Failed Indexes: {{ outputs.audit_milvus_indexes.vars.failed_indexes_count }}
      High Unindexed: {{ outputs.audit_milvus_indexes.vars.high_unindexed_count }}
      Degraded: {{ outputs.audit_milvus_indexes.vars.degraded }}

triggers:
  - id: hourly_milvus_audit
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 * * * *"
    description: Executes the Milvus vector collection indexing health audit every hour.

Audits Milvus distributed vector database HTTP management endpoints to verify collection index construction, identify failed HNSW or IVF index builds, and track unindexed entity backlogs.

How it works

The hourly_milvus_audit schedule trigger initiates the collection inspection once every hour.
The audit_milvus_indexes task queries the Milvus API to inspect collection schemas, vector field index states, and growing segment counts.
The check_indexing_health flowable evaluates whether any index failed to finish or if unindexed entities in growing segments breach the safety threshold.
If degradation is detected, notify_slack_ai_platform sends an urgent diagnostic report to the AI Platform Slack channel.
If all vector indexes are finished and healthy, log_healthy_milvus logs nominal status.
The export_milvus_manifest task records an execution summary payload.

Architecture diagram

flowchart TD
    T[Schedule: hourly_milvus_audit] --> S[audit_milvus_indexes]
    S --> C{check_indexing_health}
    C -- Degraded --> A[notify_slack_ai_platform]
    C -- Healthy --> H[log_healthy_milvus]
    A --> E[export_milvus_manifest]
    H --> E[export_milvus_manifest]

Use cases

Enterprise RAG Latency Defense: Prevents vector search queries from reverting to expensive brute-force scans when automated index building fails.
Continuous Ingestion Safeguard: Ensures streaming ingestion pipelines do not accumulate massive uncompacted segments that exhaust query node memory.
Multi-Tenant AI Platform Governance: Verifies that dynamically provisioned collections build required HNSW or SCANN indexes before servicing user traffic.

Inputs

Name	Type	Default	Description
`milvus_endpoint`	STRING	`http://milvus-standalone.internal:19530`	Base HTTP or gRPC management endpoint of the Milvus vector database cluster.
`milvus_api_key`	SECRET	(optional)	Secret containing the optional token or API key for Milvus authentication.
`collection_filter`	STRING	`""`	Optional substring filter for Milvus collection names to inspect.
`max_unindexed_entities_threshold`	INT	`5000`	Maximum permissible entities residing in unindexed growing segments before alerting.
`alert_on_failed_indexes`	BOOL	`true`	Whether to alert immediately if any vector index build has failed or stalled.

Expected outputs

outputs.audit_milvus_indexes.vars.total_audited: Number of collections inspected.
outputs.audit_milvus_indexes.vars.failed_indexes_count: Count of collections with unbuilt or failed vector indexes.
outputs.audit_milvus_indexes.vars.high_unindexed_count: Collections with growing entities exceeding threshold.
outputs.audit_milvus_indexes.vars.degraded: Boolean flag indicating vector search performance degradation.
outputs.export_milvus_manifest.value: Formatted execution report summary.

Prerequisites

Network connectivity from Kestra workers to the Milvus HTTP endpoint (typically port 19530).
Milvus 2.3+ with RESTful v2 API enabled.
Slack incoming webhook endpoint configured.

Secrets

SLACK_WEBHOOK_URL: Secret containing the Slack incoming webhook endpoint for channel alerts.
MILVUS_API_KEY: Secret containing the Milvus bearer token or root password if authentication is enabled.

Quick start

Define SLACK_WEBHOOK_URL and MILVUS_API_KEY in your Kestra namespace secrets.
Adjust milvus_endpoint to point to your Milvus standalone or distributed proxy address.
Run a manual execution to test collection discovery and verify index states.
Enable the hourly_milvus_audit schedule for automated cluster surveillance.

Common pitfalls and troubleshooting

Growing vs Sealed Segments: Entities freshly inserted enter growing segments and are not indexed until flushed into sealed segments. Small unindexed counts are expected during active ingestion.
Authentication Disabled by Default: If Milvus has authorization disabled, leave milvus_api_key empty.
AutoIndex in Milvus 2.4: Collections configured with AutoIndex manage parameters automatically, but still transition through InProgress and Finished states.

How to extend

Trigger automated index recreation via Milvus REST API create_index when corrupted states occur.
Automate collection manual compaction when fragment counts trigger query latency warnings.
Publish index build durations to Prometheus or Grafana to track embedding cluster efficiency.

Milvus Collection Index State and Unindexed Entity Sentinel

How it works

Architecture diagram

Use cases

Inputs

Expected outputs

Prerequisites

Secrets

Quick start

Common pitfalls and troubleshooting

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Milvus Collection Index State and Unindexed Entity Sentinel

How it works

Architecture diagram

Use cases

Inputs

Expected outputs

Prerequisites

Secrets

Quick start

Common pitfalls and troubleshooting

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Data Blueprints

Meilisearch Index Rebuild Pipeline with Dated Indexes and Manual Swap

Send CRM lead stages to Meta Conversions API (Conversion Leads)

Meilisearch Index Health Sentinel and Search Audit

Meilisearch Indexing Queue Staleness Sentinel

Fetch Metaplane Context When a Page Fires

Metabase Cache and Consistency Ops Guard

More AI Blueprints

Use HuggingFace Inference API to classify customer message

Screen Job Applicants with AI Scorecards and a Recruiter in the Loop

LanceDB Vector Table Fragment Compaction and Storage Sentinel

Hugging Face Inference Endpoint Health Sentinel

Monitor LiteLLM Proxy Budgets and Automate Provider Fallback Alerts via Slack

Hugging Face Model Safetensors Security and Serialization Guard

New to Kestra?