Quickwit Index Source Compliance Gate with Kestra

id: quickwit-index-source-compliance-gate
namespace: company.team

description: |
  Read a Quickwit index's attached sources, classify the index HEALTHY, NO_SOURCES,
  NON_COMPLIANT_SOURCE_TYPE or PROBE_FAILED from the real source types found, alert Slack,
  and optionally pause exactly one named non-compliant source without deleting it.

inputs:
  - id: quickwit_url
    type: STRING
    displayName: Quickwit base URL
    description: Base URL of the Quickwit REST API. Maps to every task's url field.
      A local node listens on port 7280.
    defaults: "http://localhost:7280"

  - id: index_id
    type: STRING
    displayName: Index ID
    description: ID of the Quickwit index to audit. Maps to every task's index field.
    defaults: "app-logs"

  - id: approved_source_type
    type: SELECT
    displayName: Approved source type
    description: The only source_type this index's sources are allowed to be. Any
      attached source of a different type is classified
      NON_COMPLIANT_SOURCE_TYPE.
    values:
      - "kafka"
      - "kinesis"
      - "pulsar"
    defaults: "kafka"

  - id: target_source_id
    type: STRING
    displayName: Target source ID for remediation
    description: Leave empty to never pause anything. When set, together with
      auto_remediate true and dry_run false, pauses exactly this one source, the
      one an operator already identified from a prior alert on this gate, never
      all non-compliant sources at once.
    defaults: ""

  - id: auto_remediate
    type: BOOL
    displayName: Auto remediate
    description: When true, together with dry_run false and target_source_id set,
      pauses that one source via Toggle and checks it still exists afterward.
    defaults: false

  - id: dry_run
    type: BOOL
    displayName: Dry run
    description: Must be false, together with auto_remediate true, for the pause to
      actually run.
    defaults: true

tasks:
  - id: probe_index_metadata
    type: io.kestra.plugin.quickwit.index.Get
    description: Reads the index's effective configuration and the list of sources
      attached to it.
    allowFailure: true
    url: "{{ inputs.quickwit_url }}"
    index: "{{ inputs.index_id }}"

  - id: classify_index_source_compliance
    type: io.kestra.plugin.core.debug.Return
    description: PROBE_FAILED when the probe did not return at all, NO_SOURCES when
      the index has nothing feeding it, NON_COMPLIANT_SOURCE_TYPE when any
      attached source's real source_type does not match approved_source_type,
      HEALTHY otherwise.
    format: "{{ outputs.probe_index_metadata is not defined ? 'PROBE_FAILED' : (
      (outputs.probe_index_metadata.metadata.sources | length) == 0 ?
      'NO_SOURCES' : ( (outputs.probe_index_metadata.metadata.sources |
      jq(inputs.approved_source_type == 'kafka' ? 'any(.[]; .source_type !=
      \"kafka\")' : (inputs.approved_source_type == 'kinesis' ? 'any(.[];
      .source_type != \"kinesis\")' : 'any(.[]; .source_type != \"pulsar\")')) |
      first) ? 'NON_COMPLIANT_SOURCE_TYPE' : 'HEALTHY' ) ) }}"

  - id: route_by_index_source_compliance
    type: io.kestra.plugin.core.flow.Switch
    description: Branch A logs a fully compliant index. Branch B alerts Slack with
      the real source types found, never an invented list.
    value: "{{ outputs.classify_index_source_compliance.value }}"
    cases:
      "HEALTHY":
        - id: log_healthy
          type: io.kestra.plugin.core.log.Log
          message: "Quickwit index {{ inputs.index_id }} compliant. All attached sources
            are {{ inputs.approved_source_type }}."
      "NO_SOURCES":
        - id: alert_no_sources
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK') }}"
          payload: |
            {{ {"text": "Quickwit index " ~ inputs.index_id ~ " has NO_SOURCES attached. Nothing is feeding this index."} | toJson }}
      "NON_COMPLIANT_SOURCE_TYPE":
        - id: alert_non_compliant_source_type
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK') }}"
          payload: |
            {{ {"text": "Quickwit index " ~ inputs.index_id ~ " NON_COMPLIANT_SOURCE_TYPE. Approved type is " ~ inputs.approved_source_type ~ ". Attached source types found: " ~ (outputs.probe_index_metadata.metadata.sources | jq('[.[].source_type] | join(", ")') | first) ~ "."} | toJson }}
      "PROBE_FAILED":
        - id: alert_probe_failed
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK') }}"
          payload: |
            {{ {"text": "Quickwit index metadata probe for " ~ inputs.index_id ~ " did not return a result. Check quickwit_url, index_id, and that the index exists."} | toJson }}
    defaults:
      - id: log_unexpected_status
        type: io.kestra.plugin.core.log.Log
        message: "Unexpected classify_index_source_compliance value: {{
          outputs.classify_index_source_compliance.value }}"

  - id: remediate_gate
    type: io.kestra.plugin.core.flow.If
    description: Runs only when the index was NON_COMPLIANT_SOURCE_TYPE,
      target_source_id is set, auto_remediate is true, and dry_run is false.
    condition: "{{ outputs.classify_index_source_compliance.value ==
      'NON_COMPLIANT_SOURCE_TYPE' and inputs.target_source_id != '' and
      inputs.auto_remediate and not inputs.dry_run }}"
    then:
      - id: toggle_disable_source
        type: io.kestra.plugin.quickwit.source.Toggle
        description: Pauses indexing on this one source without deleting it or its
          checkpoint, never a bulk operation across every non-compliant source.
        allowFailure: true
        url: "{{ inputs.quickwit_url }}"
        index: "{{ inputs.index_id }}"
        source: "{{ inputs.target_source_id }}"
        enable: false

      - id: verify_source_still_exists
        type: io.kestra.plugin.quickwit.index.Get
        description: Re-reads the index to confirm target_source_id is still attached
          after the toggle. Quickwit's REST API exposes no enabled/disabled flag
          on a source, so this confirms the source was not unexpectedly removed,
          not that indexing actually paused.
        allowFailure: true
        url: "{{ inputs.quickwit_url }}"
        index: "{{ inputs.index_id }}"

      - id: report_remediation_result
        type: io.kestra.plugin.core.flow.If
        description: Fails the flow when the verify probe did not return at all or
          target_source_id no longer appears in the index's source list.
        condition: "{{ outputs.verify_source_still_exists is not defined or not
          (outputs.verify_source_still_exists.metadata.sources | jq('any(.[];
          .source_id == \"' ~ inputs.target_source_id ~ '\")') | first) }}"
        then:
          - id: remediation_not_confirmed
            type: io.kestra.plugin.core.execution.Fail
            errorMessage: "Toggle ran for source {{ inputs.target_source_id }} on index {{
              inputs.index_id }}, but it no longer appears in the index's source
              list."
        else:
          - id: remediation_confirmed
            type: io.kestra.plugin.core.log.Log
            message: "Source {{ inputs.target_source_id }} on index {{ inputs.index_id }}
              toggled to enable=false and still exists. Quickwit exposes no
              independent enabled/disabled flag, so this confirms the request
              was accepted and the source was not removed, not a verified paused
              state."

  - id: log_audit_summary
    type: io.kestra.plugin.core.log.Log
    description: Final audit log confirming the gate ran, regardless of the branch
      taken above.
    message: "Quickwit index source compliance gate completed. index={{
      inputs.index_id }} status={{
      outputs.classify_index_source_compliance.value }} auto_remediate={{
      inputs.auto_remediate }} dry_run={{ inputs.dry_run }}"

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK') }}"
    payload: |
      {{ {"text": "Quickwit index source compliance gate flow failed (execution " ~ execution.id ~ "). Check the execution logs."} | toJson }}

triggers:
  - id: every_15_minutes
    type: io.kestra.plugin.core.trigger.Schedule
    description: Shipped disabled, and always in safe mode (dry_run true) by default.
    cron: "*/15 * * * *"
    disabled: true
    inputs:
      quickwit_url: "http://localhost:7280"
      index_id: "app-logs"
      approved_source_type: "kafka"
      target_source_id: ""
      auto_remediate: false
      dry_run: true

A Quickwit source can only be set to kafka, kinesis, or pulsar once, that choice cannot be changed after creation, only replaced by deleting and recreating the source. That makes source type an easy thing to standardize on for a given environment, and an easy thing to drift from silently, someone attaches a pulsar source to an index that is supposed to be kafka-only, and nothing in Quickwit itself flags it. This blueprint reads an index's real attached sources on a schedule, alerts Slack the moment one does not match the approved type, or the index has no sources at all, and can pause one specifically named non-compliant source without deleting it or its indexing checkpoint.

How it works

probe_index_metadata (io.kestra.plugin.quickwit.index.Get) reads the index's configuration and the list of sources attached to it, with allowFailure: true so a bad URL or a renamed index does not stop the flow before it can be classified as PROBE_FAILED below.
classify_index_source_compliance (io.kestra.plugin.core.debug.Return) returns PROBE_FAILED when the probe did not return at all, NO_SOURCES when the index has zero attached sources, NON_COMPLIANT_SOURCE_TYPE when any attached source's real source_type does not match approved_source_type, HEALTHY otherwise.
route_by_index_source_compliance (io.kestra.plugin.core.flow.Switch) branches on that status: HEALTHY logs it, the other three each alert Slack, the NON_COMPLIANT_SOURCE_TYPE alert lists the real source types actually found, never an invented list. An unexpected value falls through to a defaults branch that logs it rather than being silently dropped.
remediate_gate (io.kestra.plugin.core.flow.If) runs only when the status was NON_COMPLIANT_SOURCE_TYPE, target_source_id is set, auto_remediate is true, and dry_run is false. It calls io.kestra.plugin.quickwit.source.Toggle with enable: false on that one source, then re-reads the index to confirm the source is still attached, failing the flow with io.kestra.plugin.core.execution.Fail if it is not, rather than silently reporting success.
log_audit_summary always runs last and logs the final classification.
The errors block posts to Slack if the flow itself fails outside the gate logic above.

What you get

A real source-type inventory for an index, not a guess from how busy the index looks.
A Slack alert listing the actual source types found, never an invented list.
A pause, not a delete: Toggle stops a non-compliant source from indexing further while keeping its checkpoint, so re-enabling it later resumes exactly where it left off.
A fully local test path against the official Quickwit Docker image, no managed Quickwit cluster needed.

Who it's for

Data platform and search infrastructure teams running Quickwit who standardize which streaming source types are approved per index or per environment.
Teams evaluating the Quickwit plugin who want a working example of index.Get and source.Toggle.

Why orchestrate this with Kestra

Quickwit's own REST API will happily accept a source of any supported type on any index, it has no concept of an organization's own approved-type policy. Kestra adds that policy as a scheduled, declarative check: the probe, the classification against your approved type, the Slack alert naming the real sources found, and the optional scoped pause are all one auditable execution history.

Prerequisites

For local testing, the official Quickwit Docker image:
```
mkdir qwdata
docker run --rm -v $(pwd)/qwdata:/quickwit/qwdata -p 127.0.0.1:7280:7280 \\
  quickwit/quickwit:0.9.0 run
```
Quickwit has no authentication layer of its own, so no credentials are needed against a local node. Only add one in front of a shared or production cluster, typically a reverse proxy or API gateway, not something this blueprint configures.
An index to audit. Create one from the stackoverflow tutorial config, or any index config of your own, via POST /api/v1/indexes or Kestra's own io.kestra.plugin.quickwit.index.Create.
At least one source attached, to see NON_COMPLIANT_SOURCE_TYPE rather than NO_SOURCES. io.kestra.plugin.quickwit.source.Create can attach one without a live broker behind it, this blueprint only reads source configuration metadata, it does not check whether the source is actually receiving data.
A Slack incoming webhook for alerts.

Secrets

SLACK_WEBHOOK: Slack incoming webhook URL used by the alert tasks and the errors block.

In Kestra OSS, set this as a base64-encoded environment variable, for example SECRET_SLACK_WEBHOOK.

Inputs

Input	Type	Default	Allowed values
`quickwit_url`	STRING	`http://localhost:7280`	any reachable Quickwit base URL
`index_id`	STRING	`app-logs`	any existing index ID
`approved_source_type`	SELECT	`kafka`	`kafka`, `kinesis`, `pulsar`
`target_source_id`	STRING	`` (empty)	empty, or an existing source ID
`auto_remediate`	BOOL	`false`	`true`, `false`
`dry_run`	BOOL	`true`	`true`, `false`

Outputs

Task	Output path	Description
`probe_index_metadata`	`{{ outputs.probe_index_metadata.metadata.sources }}`	Sources attached to the index, each with `source_id` and `source_type`.
`classify_index_source_compliance`	`{{ outputs.classify_index_source_compliance.value }}`	One of `HEALTHY`, `NO_SOURCES`, `NON_COMPLIANT_SOURCE_TYPE`, `PROBE_FAILED`.
`toggle_disable_source`	`{{ outputs.toggle_disable_source.enable }}`	`false` once the remediation toggle call completes.
`verify_source_still_exists`	`{{ outputs.verify_source_still_exists.metadata.sources }}`	Source list re-read after the toggle, used to confirm `target_source_id` is still attached.

Quick start

Start the Quickwit container above.
Create an index, then attach a source with source_type: pulsar (or any type other than your approved_source_type) using io.kestra.plugin.quickwit.source.Create.
Add the SLACK_WEBHOOK secret.
Run this flow manually with the defaults, it should report NON_COMPLIANT_SOURCE_TYPE and list pulsar as the source type found.
Set target_source_id to that source's ID, auto_remediate: true, and dry_run: false, then run it again, it should pause the source and log the verified result.
Enable the every_15_minutes trigger.

How to extend

Loop this flow over a list of index IDs with io.kestra.plugin.core.flow.Loop to audit every index in a cluster from one flow.
Add a check on num_pipelines against an approved maximum, a source with far more pipelines than expected can be a sign of a misconfigured deployment.
Chain io.kestra.plugin.quickwit.source.Toggle with enable: true in a separate, explicitly triggered flow once an operator has reviewed and approved the non-compliant source, rather than auto-re-enabling here.

Pitfalls

Quickwit's index.Get response exposes no enabled or disabled flag for a source, only its configuration. report_remediation_result can confirm the toggled source still exists after the call, it cannot independently confirm from this plugin that indexing actually paused.
source_type cannot be changed after a source is created, only replaced by deleting and recreating it. This gate flags a non-compliant type, it cannot fix it by changing the type in place.
NO_SOURCES and NON_COMPLIANT_SOURCE_TYPE are mutually exclusive in this classification, an index with zero sources is always reported NO_SOURCES even if that is itself the intended, static-ingestion state for that index.
This blueprint only reads source configuration, not ingestion throughput, a compliant kafka source that stopped actually consuming from its topic reports HEALTHY here.

Quickwit Index Source Compliance Gate

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Inputs

Outputs

Quick start

How to extend

Pitfalls

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Quickwit Index Source Compliance Gate

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Inputs

Outputs

Quick start

How to extend

Pitfalls

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Data Blueprints

Kafka Consumer Lag Sentinel

Kafka Schema Registry Compatibility and Mode Drift Guard

Real-Time Kafka Record Processing with Per-Record Executions

Kafka Orphaned Topic Reclaim Guard

Kafka DLQ Quarantine, Classification, and Safe Replay Gate

Kafka Producer Compression Ratio Sentinel

More Infrastructure Blueprints

Kafka Schema Registry Orphaned Subject Cleanup Guard

Block a breaking Avro schema before a Kafka producer deploys

Kafka Orphaned Topic Reclaim Guard

Kafka Idle Consumer Partition Guard

Kafka Stale Consumer Group Cleanup Guard

Kafka Schema Registry Subject Version Pruning Guard

New to Kestra?