OpenSearch PPL Missing Field Compliance Gate

A required field that silently stops being written - a broken ingest pipeline, a client library update that dropped a field, a migration that missed a column - does not throw an error. It just accumulates as documents with a hole where that field should be. This blueprint uses PPL (Piped Processing Language), OpenSearch's SQL-like query language, to count exactly how many documents in an index are missing a required field, alerts on breach, and can flag the offending documents through OpenSearch's own _update_by_query API - adding one marker field, never deleting or overwriting anything.

How it works

1. hourly_compliance_check (io.kestra.plugin.core.trigger.Schedule) runs hourly, always forcing auto_remediate: "false" and dry_run: "true" through its own inputs: override regardless of this flow's defaults. Shipped disabled so you can validate a manual run first.
2. audit_missing_field (io.kestra.plugin.opensearch.Ppl) runs source={{ index_name }} | where isnull({{ required_field }}) | stats count() as missing_count with fetchType: FETCH_ONE, populating row.missing_count.
3. evaluate_missing_field_gate (io.kestra.plugin.core.flow.If) flags a breach whenever missing_count > max_missing_count.
   • then (breach): route_remediation (io.kestra.plugin.core.flow.Switch on auto_remediate) either considers remediation ("true", gated again by dry_run through check_dry_run) or only logs that remediation is disabled ("false"). When both gates allow it, remediate_missing_field (io.kestra.plugin.opensearch.Request) calls OpenSearch's own _update_by_query endpoint with an exists-based query and a Painless script that sets flagged_for_review. alert_missing_field_breach always fires to Slack regardless of which path ran.
   • else (compliant): log_compliance_healthy records a clean check.
4. log_audit_complete always runs last, printing the headline numbers regardless of which branch fired.
5. The errors block alerts Slack separately if the flow itself fails - an unreachable cluster, or a cluster missing the optional opensearch-sql plugin that PPL requires, must not read as "no breach".

What you get

• A precise count of documents missing a required field, computed by OpenSearch itself through PPL, not estimated or sampled.
• Remediation that only ever adds a flagged_for_review marker - it cannot delete documents or overwrite required_field or any other existing field.
• Two independent gates (auto_remediate, dry_run) between "breach detected" and "documents actually flagged".
• A final status log every run, breach or not, so the execution history doubles as a data-quality trend for that field.

Who it's for

• Platform and data engineering teams running OpenSearch who need an early signal when an ingest pipeline silently stops populating a field.
• Teams evaluating PPL as an alternative to the native Query DSL for simple aggregate checks.
• Anyone who has found out about a broken field weeks later, from a downstream report instead of from the cluster itself.

Why orchestrate this with Kestra

A PPL query run by hand answers one moment in time; it does not decide a cadence, branch on a threshold, flag the affected documents, or notify anyone. Kestra supplies the schedule, the threshold comparison as a first-class branch, two independent authorization gates before anything is written back to the cluster, and an execution history that shows exactly when the field started going missing and whether it was flagged.

Prerequisites

• A reachable OpenSearch cluster with the optional opensearch-sql plugin installed - Ppl fails if it is missing.
• An index (index_name) with documents that may or may not carry required_field.
• A Slack incoming webhook for alerts.

Local testing: docker run -d --name opensearch-gate -p 9201:9200 -p 9601:9600 -e "discovery.type=single-node" -e "DISABLE_SECURITY_PLUGIN=true" opensearchproject/opensearch:2.11.0 docker network connect YOUR_KESTRA_NETWORK opensearch-gate (only needed if the Kestra Worker runs in a separate Docker network than this container; set opensearch_host to http://opensearch-gate:9200 from inside that network, or http://localhost:9201 from the host).

Secrets

• SLACK_WEBHOOK_URL: Slack incoming webhook used by alert_missing_field_breach and the errors block.
• This blueprint's connection.hosts assumes an unauthenticated local cluster (DISABLE_SECURITY_PLUGIN=true), matching the plugin's own documentation examples. A secured cluster needs connection-level auth headers or credentials on OpensearchConnection, which were not exercised here - add them from {{ secret(...) }} values rather than hardcoding them if you enable security.
• In Kestra OSS (no Enterprise secrets backend), secrets are supplied as environment variables prefixed SECRET_, base64-encoded, and read back in flows with {{ secret('NAME') }} - for example SECRET_SLACK_WEBHOOK_URL=$(echo -n 'https://hooks.slack.com/...' | base64). This keeps credentials out of the flow YAML but, per Kestra's own documentation, offers no encryption at rest or access control beyond the host environment; use the Enterprise secrets backend for stronger guarantees.

Inputs

• opensearch_host (STRING, default http://localhost:9200): connection.hosts entry.
• index_name (STRING, default orders): index audited and, if remediated, updated.
• required_field (STRING, default tracking_id): field whose absence is counted.
• max_missing_count (INT, default 0): breach threshold; 0 means any missing field at all breaches.
• auto_remediate (SELECT: "false", "true"; default "false"): must be "true" for remediation to even be considered.
• dry_run (SELECT: "true", "false"; default "true"): must be explicitly "false", together with auto_remediate: "true", for _update_by_query to actually run.

Outputs

• outputs.audit_missing_field.row.missing_count: the PPL-computed count for every run.
• outputs.remediate_missing_field.status / .response: the _update_by_query HTTP status and response body, present only when remediation actually ran.

Quick start

1. Start OpenSearch locally with the command above and index a few documents into orders, some with tracking_id and some without.
2. Add the SLACK_WEBHOOK_URL secret.
3. Run the flow manually with the defaults and confirm audit_missing_field.row.missing_count matches how many documents you left without tracking_id, and that alert_missing_field_breach fires when it is above 0.
4. Run once with auto_remediate: "true" and dry_run: "true" and confirm log_dry_run_remediation describes the update without running it.
5. Run once with auto_remediate: "true" and dry_run: "false" and confirm the matching documents now carry flagged_for_review: true.
6. Enable hourly_compliance_check once you trust the check; it always runs in the safe auto_remediate: "false" / dry_run: "true" mode regardless of what you leave the flow's own defaults set to.

How to extend

• Swap isnull({{ required_field }}) for a different PPL condition (an invalid value, a stale timestamp) to audit a different kind of data-quality gap with the same gate structure.
• Loop over several index_name/required_field pairs with io.kestra.plugin.core.flow.Loop to run the same audit across multiple indices in one flow.
• Replace the Painless flagged_for_review marker with a richer audit trail - a timestamp and the execution id - by extending the script source.
• Use format: CSV or format: RAW on Ppl to export the raw breach list instead of only a count, for a fuller incident report.

Pitfalls

• PPL requires the optional opensearch-sql plugin. Verified from the plugin's own description: "Requires the optional opensearch-sql plugin on the cluster. The task fails if the plugin is missing." A cluster without it will fail audit_missing_field, which is caught by the errors block, not silently read as "zero missing".
• There is no structured "flag documents" task in this plugin. Remediation goes through io.kestra.plugin.opensearch.Request calling OpenSearch's native _update_by_query REST endpoint directly - the Kestra plugin only transports the HTTP call, it does not know what _update_by_query does.
• The remediation script only ever adds a field. ctx._source.flagged_for_review = true is additive; it does not touch required_field or any other existing field, so remediation cannot corrupt the data it is auditing.
• auto_remediate and dry_run are independent gates, not a single boolean. Both must be "true"/"false" respectively for _update_by_query to run; setting only one leaves the other still blocking.
• Switch case keys "true"/"false" are quoted strings. auto_remediate renders as the literal string "true" or "false"; unquoted true:/false: YAML map keys would parse as booleans instead and would not match.

Links

• plugin-opensearch documentation
• OpenSearch PPL language reference
• OpenSearch stats command
• OpenSearch Update By Query API
• Kestra secrets (OSS vs Enterprise)