Cosmos DB Expired Document Retention Gate with Kestra

id: cosmosdb-expired-document-retention-gate
namespace: company.team

description: |
  Query a Cosmos DB container for documents that should have expired by their own expiry_field
  but are still present, classify the run COMPLIANT, WARNING, CRITICAL or PROBE_FAILED from the
  real count returned, alert Slack with that count, and optionally delete the lingering
  documents one by one behind a dry_run gate, then re-query to verify they are gone.

inputs:
  - id: database_id
    type: STRING
    displayName: Cosmos database ID
    description: Target Cosmos database name. Maps to the Query and Delete tasks'
      databaseId field.
    defaults: "compliance_db"

  - id: container_id
    type: STRING
    displayName: Cosmos container ID
    description: Target container (collection) name inside the database. Maps to the
      Query and Delete tasks' containerId field.
    defaults: "customer_records"

  - id: partition_key_field
    type: STRING
    displayName: Partition key field name
    description: Document property used as the partition key in the target container
      (e.g. partitionKey, id, or tenantId).
    defaults: "partitionKey"

  - id: expiry_field
    type: STRING
    displayName: Expiry field name
    description: Document field holding an ISO-8601 timestamp of when the document
      should have been purged, for example by application-level retention policy
      or a TTL that did not fire.
    defaults: "expiresAt"

  - id: grace_period_minutes
    type: INT
    displayName: Grace period in minutes
    description: Minutes past expiry_field a document is still tolerated before
      counting as a violation, to absorb normal purge latency.
    defaults: 60

  - id: violation_count_critical_threshold
    type: INT
    displayName: Critical violation count threshold
    description: Violation count at or above which the run is classified CRITICAL
      instead of WARNING.
    defaults: 50

  - id: remediation_action
    type: SELECT
    displayName: Remediation action
    description: What remediate_gate does with the violating documents when it runs.
      log_only never deletes anything, delete removes each violating document
      one by one.
    values:
      - "log_only"
      - "delete"
    defaults: "log_only"

  - id: auto_remediate
    type: BOOL
    displayName: Auto remediate
    description: When true, together with dry_run false and remediation_action
      delete, remove every violating document found.
    defaults: false

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

tasks:
  - id: probe_violations
    type: io.kestra.plugin.azure.storage.cosmosdb.Query
    description: Read-only. Finds every document whose expiry_field is in the past
      by more than grace_period_minutes. allowFailure keeps the flow going so an
      unreachable account can still be classified as PROBE_FAILED below.
    allowFailure: true
    tenantId: "{{ secret('AZURE_TENANT_ID') }}"
    connectionString: "{{ secret('COSMOS_CONNECTION_STRING') }}"
    databaseId: "{{ inputs.database_id }}"
    containerId: "{{ inputs.container_id }}"
    query: |
      SELECT c.id, c.{{ inputs.partition_key_field }}, c.{{ inputs.expiry_field }} FROM c WHERE IS_DEFINED(c.{{ inputs.expiry_field }}) AND c.{{ inputs.expiry_field }} < "{{ now() | dateAdd(-inputs.grace_period_minutes, 'MINUTES') | date("yyyy-MM-dd'T'HH:mm:ss'Z'") }}"

  - id: classify_retention_status
    type: io.kestra.plugin.core.debug.Return
    description: PROBE_FAILED when the probe produced no output at all, CRITICAL or
      WARNING when it found violations past or under the threshold, COMPLIANT
      when it found none.
    format: "{{ outputs.probe_violations is not defined ? 'PROBE_FAILED' :
      ((outputs.probe_violations.queryResults | length) == 0 ? 'COMPLIANT' :
      ((outputs.probe_violations.queryResults | length) >=
      inputs.violation_count_critical_threshold ? 'CRITICAL' : 'WARNING')) }}"

  - id: route_by_retention_status
    type: io.kestra.plugin.core.flow.Switch
    description: Branch A logs a compliant container. Branch B alerts Slack when
      documents lingered past their own expiry or the probe itself failed.
    value: "{{ outputs.classify_retention_status.value }}"
    cases:
      "COMPLIANT":
        - id: log_compliant
          type: io.kestra.plugin.core.log.Log
          message: "No expired documents found in {{ inputs.database_id }}/{{
            inputs.container_id }} past the {{ inputs.grace_period_minutes }}
            minute grace period."
      "WARNING":
        - id: alert_warning
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK') }}"
          payload: |
            {{ {"text": "Cosmos retention WARNING on " ~ inputs.database_id ~ "/" ~ inputs.container_id ~ ". " ~ (outputs.probe_violations.queryResults | length) ~ " document(s) are past their " ~ inputs.expiry_field ~ " plus a " ~ inputs.grace_period_minutes ~ " minute grace period and are still present."} | toJson }}
      "CRITICAL":
        - id: alert_critical
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK') }}"
          payload: |
            {{ {"text": "Cosmos retention CRITICAL on " ~ inputs.database_id ~ "/" ~ inputs.container_id ~ ". " ~ (outputs.probe_violations.queryResults | length) ~ " document(s) are past their " ~ inputs.expiry_field ~ " plus grace period, at or above the " ~ inputs.violation_count_critical_threshold ~ " threshold."} | toJson }}
      "PROBE_FAILED":
        - id: alert_probe_failed
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK') }}"
          payload: |
            {{ {"text": "Cosmos retention probe on " ~ inputs.database_id ~ "/" ~ inputs.container_id ~ " produced no result. Check COSMOS_CONNECTION_STRING and that the database and container exist."} | toJson }}
    defaults:
      - id: log_unhandled_status
        type: io.kestra.plugin.core.log.Log
        message: "Unhandled retention status: {{ outputs.classify_retention_status.value
          }}"

  - id: remediate_gate
    type: io.kestra.plugin.core.flow.If
    description: Runs only when the probe found violations, auto_remediate is true,
      dry_run is false, and remediation_action is delete.
    condition: "{{ (outputs.classify_retention_status.value == 'WARNING' or
      outputs.classify_retention_status.value == 'CRITICAL') and
      inputs.auto_remediate and not inputs.dry_run and inputs.remediation_action
      == 'delete' }}"
    then:
      - id: delete_expired_documents
        type: io.kestra.plugin.core.flow.Loop
        description: Deletes each violating document one at a time, supplying both id
          and the container partition key.
        concurrencyLimit: 1
        values: "{{ outputs.probe_violations.queryResults | toJson }}"
        tasks:
          - id: delete_expired_document
            type: io.kestra.plugin.azure.storage.cosmosdb.Delete
            allowFailure: true
            tenantId: "{{ secret('AZURE_TENANT_ID') }}"
            connectionString: "{{ secret('COSMOS_CONNECTION_STRING') }}"
            databaseId: "{{ inputs.database_id }}"
            containerId: "{{ inputs.container_id }}"
            item:
              id: "{{ fromJson(item.value).id }}"
              partitionKey: "{{ fromJson(item.value).partitionKey ??
                fromJson(item.value)[inputs.partition_key_field] ??
                fromJson(item.value).id }}"
              "{{ inputs.partition_key_field }}": "{{
                fromJson(item.value)[inputs.partition_key_field] ??
                fromJson(item.value).id }}"

      - id: reprobe_violations
        type: io.kestra.plugin.azure.storage.cosmosdb.Query
        allowFailure: true
        tenantId: "{{ secret('AZURE_TENANT_ID') }}"
        connectionString: "{{ secret('COSMOS_CONNECTION_STRING') }}"
        databaseId: "{{ inputs.database_id }}"
        containerId: "{{ inputs.container_id }}"
        query: |
          SELECT c.id, c.{{ inputs.partition_key_field }}, c.{{ inputs.expiry_field }} FROM c WHERE IS_DEFINED(c.{{ inputs.expiry_field }}) AND c.{{ inputs.expiry_field }} < "{{ now() | dateAdd(-inputs.grace_period_minutes, 'MINUTES') | date("yyyy-MM-dd'T'HH:mm:ss'Z'") }}"

      - id: reclassify_retention_status
        type: io.kestra.plugin.core.debug.Return
        format: "{{ outputs.reprobe_violations is not defined ? 'PROBE_FAILED' :
          ((outputs.reprobe_violations.queryResults | length) == 0 ? 'COMPLIANT'
          : 'WARNING') }}"

      - id: verify_remediation
        type: io.kestra.plugin.core.flow.If
        description: Fails the flow when violations still remain after the delete loop,
          so a partial delete is never silently reported as a fix.
        condition: "{{ outputs.reclassify_retention_status.value != 'COMPLIANT' }}"
        then:
          - id: remediation_failed
            type: io.kestra.plugin.core.execution.Fail
            errorMessage: "Delete loop ran, but {{ inputs.database_id }}/{{
              inputs.container_id }} still has expired documents."
        else:
          - id: remediated
            type: io.kestra.plugin.core.log.Log
            message: "Deleted and verified. 0 expired documents remain in {{
              inputs.database_id }}/{{ inputs.container_id }}."

  - 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: "Cosmos retention gate completed. database={{ inputs.database_id }}
      container={{ inputs.container_id }} status={{
      outputs.classify_retention_status.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": "Cosmos retention gate flow failed (execution " ~ execution.id ~ "). Check the execution logs."} | toJson }}

triggers:
  - id: six_hourly
    type: io.kestra.plugin.core.trigger.Schedule
    description: Shipped disabled, and always in safe mode by default.
    cron: "0 */6 * * *"
    disabled: true
    inputs:
      database_id: "compliance_db"
      container_id: "customer_records"
      partition_key_field: "partitionKey"
      expiry_field: "expiresAt"
      grace_period_minutes: 60
      violation_count_critical_threshold: 50
      remediation_action: "log_only"
      auto_remediate: false
      dry_run: true

Cosmos DB's container-level TTL deletes documents in the background, on its own schedule, and only if it is configured at all. An application that stamps every document with its own expiresAt field and assumes TTL will clean it up has no guarantee that actually happens, TTL can be disabled on the container, misconfigured, or simply lag under load, and the document, its PII, its stale state, just stays there. This blueprint turns that assumption into a Day 2 governance gate: it queries for documents whose own expiresAt field has passed, classifies the real count it finds, alerts Slack, and can delete the stragglers itself, then proves they are actually gone.

How it works

probe_violations (io.kestra.plugin.azure.storage.cosmosdb.Query) runs a cross-partition SQL query selecting id, partition_key_field, and expiry_field for documents where expiresAt is defined and older than the current UTC timestamp minus grace_period_minutes.
classify_retention_status (io.kestra.plugin.core.debug.Return) maps the count of returned documents to COMPLIANT (0), WARNING (1 to threshold - 1), CRITICAL (at or above violation_count_critical_threshold), or PROBE_FAILED if the query task failed to produce outputs.
route_by_retention_status (io.kestra.plugin.core.flow.Switch) routes based on that status: COMPLIANT logs an audit line, while WARNING, CRITICAL, and PROBE_FAILED post structured alerts to Slack.
remediate_gate (io.kestra.plugin.core.flow.If) activates only when violations were found, auto_remediate is true, dry_run is false, and remediation_action is delete. It calls io.kestra.plugin.azure.storage.cosmosdb.Delete once per violating document, supplying both id and the partition key value, runs reprobe_violations to ensure zero lingering documents remain, and fails the execution via io.kestra.plugin.core.execution.Fail if any expired documents are still detected.
log_audit_summary (io.kestra.plugin.core.log.Log) records the final retention gate outcome across all paths.
alert_flow_failure (io.kestra.plugin.slack.notifications.SlackIncomingWebhook) in the errors block captures unhandled flow-level errors.

What you get

Active enforcement of document retention SLAs rather than passive assumption that container TTL purged records.
Tiered severity classification (WARNING vs CRITICAL) based on violation volume.
Safe-by-default remediation: shipping in dry_run: true and log_only mode with explicit post-delete verification.
Zero external scripts: leverages official plugin-azure tasks directly.

Who it's for

Data compliance and security teams enforcing GDPR/CCPA data retention requirements.
SRE and platform engineers operating Azure Cosmos DB workloads who need assurance that data past its retention window is actually gone.
Teams evaluating io.kestra.plugin.azure.storage.cosmosdb.Query and Delete together.

Why orchestrate this with Kestra

Cosmos DB's own TTL has no concept of an application-level expiry field, a violation threshold, or an alert explaining what it found. Kestra adds the schedule, the WARNING versus CRITICAL classification built from the query's own row count, the Slack alert, the gated delete, and the verification that the delete worked, all as one declarative YAML file you can version and review like code.

Prerequisites

A Cosmos DB account (or the local emulator, see below) with a database and container whose documents carry an expiry_field such as expiresAt.
Local testing with the official Linux Cosmos DB emulator, no Azure account needed:
```
docker run --detach --name cosmos-emulator \\
  --publish 8081:8081 --publish 8082:8080 --publish 1234:1234 \\
  mcr.microsoft.com/cosmosdb/linux/azure-cosmos-emulator:vnext-latest --protocol https
```
This remaps the emulator's health probe from its default host port 8080, which otherwise collides with Kestra's own default UI port, to 8082. The real Cosmos gateway stays on 8081, and the Data Explorer stays on 1234.
This plugin calls the Cosmos Java SDK directly from the Kestra worker process. If Kestra itself runs in a container, point COSMOS_CONNECTION_STRING at https://host.docker.internal:8081/.
The Cosmos Java SDK requires HTTPS against the emulator (--protocol https above). Export and trust the emulator's self-signed certificate in the JVM running your Kestra worker.
A Slack incoming webhook for alerts.

Secrets

AZURE_TENANT_ID: Azure AD Tenant ID (GUID), required by Kestra's Azure identity task model.
COSMOS_CONNECTION_STRING: AccountEndpoint=<endpoint>;AccountKey=<key>; format, used by probe_violations, reprobe_violations, and delete_expired_document.
SLACK_WEBHOOK: Slack incoming webhook URL used by the alert tasks and the errors block.

In Kestra OSS, set these as base64-encoded environment variables: SECRET_AZURE_TENANT_ID, SECRET_COSMOS_CONNECTION_STRING, SECRET_SLACK_WEBHOOK.

Inputs

Input	Type	Default	Allowed values
`database_id`	STRING	`compliance_db`	any Cosmos database name
`container_id`	STRING	`customer_records`	any Cosmos container name
`partition_key_field`	STRING	`partitionKey`	document property used as partition key
`expiry_field`	STRING	`expiresAt`	document field name holding an ISO-8601 timestamp
`grace_period_minutes`	INT	`60`	any non-negative integer
`violation_count_critical_threshold`	INT	`50`	any non-negative integer
`remediation_action`	SELECT	`log_only`	`log_only`, `delete`
`auto_remediate`	BOOL	`false`	`true`, `false`
`dry_run`	BOOL	`true`	`true`, `false`

Quick start

Start the Cosmos emulator (or point at a real account) and seed a container with a few documents carrying a past expiresAt timestamp.
Add the AZURE_TENANT_ID, COSMOS_CONNECTION_STRING, and SLACK_WEBHOOK secrets.
Run this flow manually with the defaults and read log_audit_summary for the result.
Once you trust the query, set remediation_action: delete, auto_remediate: true, and dry_run: false and run it again to clean up and verify.
Enable the six_hourly trigger, which always ships in safe mode.

How to extend

Point this at a container partitioned on any property by configuring partition_key_field (e.g., tenantId, region, or id).
Add a second expiry_field and a second probe/classify/switch chain to audit more than one retention policy per container in one flow.
Replace the fixed violation_count_critical_threshold with a comparison against yesterday's count stored in a Kestra KV entry, to catch a retention backlog that is growing rather than a static size.

Pitfalls

Cosmos DB tasks inherit from AbstractAzureIdentityConnection, which marks tenantId with @NotNull. Even when connecting via COSMOS_CONNECTION_STRING, tenantId: "{{ secret('AZURE_TENANT_ID') }}" must be specified on each task to satisfy Kestra 2.0.1 schema validation.
io.kestra.plugin.azure.storage.cosmosdb.Delete requires both id and the partition key value in item. The probe query explicitly selects partition_key_field so delete_expired_document can pass both fields to the Cosmos DB SDK.
The Cosmos Java SDK does not support HTTP against the emulator, start it with --protocol https and trust its certificate, or every probe reports PROBE_FAILED with a TLS error that looks identical to a real outage.
IS_DEFINED(c.<expiry_field>) matters: without it, a document that never had the field set at all compares undefined < "<timestamp>", which Cosmos SQL does not treat as true, so those documents are silently excluded rather than counted.
grace_period_minutes should be at least as large as your container's configured TTL polling interval, otherwise this gate reports violations for documents Cosmos's own TTL was always going to delete on schedule.

Cosmos DB Expired Document Retention Gate

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Inputs

Quick start

How to extend

Pitfalls

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Cosmos DB Expired Document Retention Gate

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Inputs

Quick start

How to extend

Pitfalls

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Data Blueprints

More Infrastructure Blueprints

New to Kestra?