Audit Qdrant Vector Collection Indexing Health and Status

id: qdrant-collection-indexing-health-sentinel
namespace: company.ai

description: |
  Audit Qdrant vector collection health, operational status, and HNSW indexing lag
  on a recurring schedule to prevent vector search latency spikes and unindexed point accumulation.

inputs:
  - id: qdrant_host
    type: STRING
    defaults: "localhost"
    displayName: "Qdrant Host"
    description: "Hostname or container IP address of the Qdrant gRPC service endpoint."

  - id: qdrant_port
    type: INT
    defaults: 6334
    displayName: "Qdrant gRPC Port"
    description: "gRPC service port for Qdrant client communication (default 6334)."

  - id: collection_name
    type: STRING
    defaults: "enterprise-knowledge-base"
    displayName: "Target Collection Name"
    description: "Name of the Qdrant vector collection to inspect and audit for
      indexing health."

  - id: max_unindexed_lag
    type: INT
    defaults: 100
    displayName: "Max Unindexed Vectors Lag"
    description: "Maximum allowable difference between pointsCount and
      indexedVectorsCount before flagging degradation."

  - id: remediation_mode
    type: SELECT
    values:
      - "ALERT_ONLY"
      - "OPTIMIZE_LOG"
    defaults: "ALERT_ONLY"
    displayName: "Remediation Mode"
    description: "Operational response mode when collection health degrades:
      dispatch Slack alert only or record optimization guidance."

  - id: dry_run
    type: SELECT
    values:
      - "true"
      - "false"
    defaults: "true"
    displayName: "Dry Run Mode"
    description: "When set to true, safe remediation actions only simulate actions
      without mutating collection configurations."

triggers:
  - id: schedule
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 */6 * * *"
    disabled: true
    inputs:
      qdrant_host: "localhost"
      qdrant_port: 6334
      collection_name: "enterprise-knowledge-base"
      max_unindexed_lag: 100
      remediation_mode: "ALERT_ONLY"
      dry_run: "true"

tasks:
  - id: probe_collection
    type: io.kestra.plugin.qdrant.CollectionInfo
    description: "Query collection operational status, point counts, and indexed
      vector metrics via Qdrant gRPC."
    host: "{{ inputs.qdrant_host }}"
    port: "{{ inputs.qdrant_port }}"
    apiKey: "{{ secret('QDRANT_API_KEY') }}"
    collectionName: "{{ inputs.collection_name }}"
    tlsEnabled: false

  - id: evaluate_health
    type: io.kestra.plugin.core.flow.Switch
    description: "Classify collection health into HEALTHY or DEGRADED based on
      operational status and indexing lag."
    value: "{{ (outputs.probe_collection is defined and
      outputs.probe_collection.status is defined and
      (outputs.probe_collection.status != 'Green' or
      (outputs.probe_collection.pointsCount is defined and
      outputs.probe_collection.indexedVectorsCount is defined and
      (outputs.probe_collection.pointsCount -
      outputs.probe_collection.indexedVectorsCount) >
      inputs.max_unindexed_lag))) ? 'DEGRADED' : 'HEALTHY' }}"
    cases:
      "HEALTHY":
        - id: log_healthy_status
          type: io.kestra.plugin.core.log.Log
          description: "Record healthy audit result when collection status is Green and
            vector indexing lag is within threshold."
          message: "Qdrant collection {{ inputs.collection_name }} is healthy: status={{
            outputs.probe_collection.status }}, points={{
            outputs.probe_collection.pointsCount }}, indexed={{
            outputs.probe_collection.indexedVectorsCount }}."
      "DEGRADED":
        - id: alert_collection_degraded
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          description: "Dispatch urgent Slack notification when Qdrant collection status
            degrades or vector indexing lags."
          url: "{{ secret('SLACK_WEBHOOK') }}"
          payload: |
            {"text": "Qdrant Collection Degradation Alert: Collection {{ inputs.collection_name }} reported status {{ outputs.probe_collection.status }}. Total points: {{ outputs.probe_collection.pointsCount }}, Indexed vectors: {{ outputs.probe_collection.indexedVectorsCount }} (Lag threshold: {{ inputs.max_unindexed_lag }}). Remediation: {{ inputs.remediation_mode }}, DryRun: {{ inputs.dry_run }}."}
        - id: log_remediation_guidance
          type: io.kestra.plugin.core.log.Log
          description: "Record remediation guidance telemetry and next steps when
            collection degradation is detected."
          message: "Remediation action recorded: mode={{ inputs.remediation_mode }},
            dry_run={{ inputs.dry_run }}. Investigate Qdrant segment optimizer
            thresholds and background indexing resources."
    defaults:
      - id: log_default_status
        type: io.kestra.plugin.core.log.Log
        description: "Fallback logger for unexpected collection health evaluation states."
        message: "Qdrant collection health audit encountered unclassified status for {{
          inputs.collection_name }}."

  - id: log_audit_completion
    type: io.kestra.plugin.core.log.Log
    description: "Post-audit summary confirming completion of the Qdrant vector
      collection health evaluation pass."
    message: "Qdrant collection health audit completed for {{ inputs.collection_name
      }}. Final status: {{ outputs.probe_collection.status ?? 'UNKNOWN' }}."

errors:
  - id: alert_on_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: "Dispatch urgent Slack notification if flow execution encounters an
      unexpected error."
    url: "{{ secret('SLACK_WEBHOOK') }}"
    payload: |
      {"text": "Flow execution failed: qdrant-collection-indexing-health-sentinel failed during execution {{ execution.id }}."}