MongoDB Expired Document Retention Guard with Kestra

id: mongodb-expired-document-retention-guard
namespace: company.team

description: |
  Find the documents in a MongoDB collection that are past a retention window and should no
  longer be stored. A collection with no TTL index on its timestamp field keeps every document
  forever. A dropped TTL index does the same. Storage grows without bound and data lives past
  the policy that was meant to delete it. This flow reads the collection with a date
  filter, counts the documents older than the window, deletes them behind a safe mode gate and
  an approval, then reads the collection again and fails the run if any expired document is
  still there.

inputs:
  - id: connection_uri
    type: STRING
    displayName: MongoDB connection string
    description: >-
      Leave empty in production so the flow reads the MONGODB_URI secret. Set it
      to a plain connection string only for a local test run.
    defaults: ""

  - id: database
    type: STRING
    displayName: Database
    defaults: appdata

  - id: collection
    type: STRING
    displayName: Collection
    defaults: events

  - id: timestamp_field
    type: STRING
    displayName: Timestamp field
    description: The date field each document is aged on, for example createdAt. It
      must hold a BSON date.
    defaults: createdAt

  - id: retention_days
    type: INT
    displayName: Retention window in days
    description: A document whose timestamp is older than this many days is expired.
    defaults: 30

  - id: scan_limit
    type: INT
    displayName: Documents read per check
    description: >-
      Ceiling on the detection read and on each re-check so a huge expired set
      cannot exhaust the worker. The delete itself is not capped. If the count
      reaches this number the report says the result is partial.
    defaults: 1000

  - id: approval_timeout
    type: STRING
    displayName: Approval timeout
    description: ISO-8601 duration to wait for an answer before the pause expires.
    defaults: "PT2H"

  - id: auto_remediate
    type: BOOL
    displayName: Auto remediate
    description: Allow the delete. Still needs dry_run false and an approval.
    defaults: false

  - id: dry_run
    type: BOOL
    displayName: Dry run
    description: Must be false, together with auto_remediate, before any document is
      deleted.
    defaults: true

variables:
  cutoff: "{{ execution.startDate | dateAdd(-1 * inputs.retention_days, 'DAYS') |
    date(\"yyyy-MM-dd'T'HH:mm:ss.SSS'Z'\", 'UTC') }}"

tasks:
  - id: detect
    type: io.kestra.plugin.mongodb.Find
    description: >-
      Read-only. Select every document whose timestamp field is before the
      retention cutoff, oldest first, up to scan_limit. size is the count the
      rest of the flow acts on.
    connection:
      uri: "{{ inputs.connection_uri != '' ? inputs.connection_uri :
        secret('MONGODB_URI') }}"
    database: "{{ inputs.database }}"
    collection: "{{ inputs.collection }}"
    filter:
      "{{ inputs.timestamp_field }}":
        $lt:
          $date: "{{ render(vars.cutoff) }}"
    projection:
      "{{ inputs.timestamp_field }}": 1
    sort:
      "{{ inputs.timestamp_field }}": 1
    limit: "{{ inputs.scan_limit }}"
    store: false

  - id: report
    type: io.kestra.plugin.core.log.Log
    description: Log the count. Task outputs are not in the execution JSON, so the
      count the gate uses is logged here.
    message: >-
      Checked {{ inputs.database }}.{{ inputs.collection }} for documents with
      {{ inputs.timestamp_field }} before {{ render(vars.cutoff) }} ({{
      inputs.retention_days }} day retention). Found {{ outputs.detect.size ?? 0
      }} expired document(s){{ (outputs.detect.size ?? 0) >= inputs.scan_limit ?
      ', scan_limit reached so there may be more' : '' }}. Oldest sampled
      document: {{ outputs.detect.rows[0] ?? 'none' }}.

  - id: route
    type: io.kestra.plugin.core.flow.If
    description: Expired documents present means alert and consider remediation. A
      clean collection just logs and stops.
    condition: "{{ (outputs.detect.size ?? 0) > 0 }}"
    then:
      - id: alert
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {{ {'text': ('MongoDB retention guard on ' ~ inputs.database ~ '.' ~ inputs.collection ~ ': ' ~ (outputs.detect.size ?? 0) ~ ' document(s) older than ' ~ inputs.retention_days ~ ' days (before ' ~ render(vars.cutoff) ~ '). Execution ' ~ execution.id ~ '.')} | toJson }}

      - id: planned
        type: io.kestra.plugin.core.log.Log
        description: What a remediation run would do, printed on every run including a
          dry one.
        message: >-
          {{ outputs.detect.size ?? 0 }} document(s) in {{ inputs.database }}.{{
          inputs.collection }} are past the {{ inputs.retention_days }} day
          window. dry_run={{ inputs.dry_run }} auto_remediate={{
          inputs.auto_remediate }}. A delete runs only when auto_remediate is
          true and dry_run is false and a reviewer approves the pause.

      - id: remediate_gate
        type: io.kestra.plugin.core.flow.If
        description: The delete opens only when auto_remediate is true and dry_run is false.
        condition: "{{ inputs.auto_remediate and not inputs.dry_run }}"
        then:
          - id: hold
            type: io.kestra.plugin.core.flow.Pause
            description: >-
              The delete is destructive, so a human answers first. With no
              answer nothing is deleted and the findings stay open.
            pauseDuration: "{{ inputs.approval_timeout }}"
            onResume:
              - id: approved
                type: BOOL
                displayName: Delete the expired documents?
                defaults: false
              - id: note
                type: STRING
                displayName: Decision note
                defaults: ""

          - id: apply
            type: io.kestra.plugin.core.flow.If
            condition: "{{ outputs.hold.onResume.approved ?? false }}"
            then:
              - id: delete
                type: io.kestra.plugin.mongodb.Delete
                description: >-
                  Remove every document older than the cutoff. Same filter as
                  detect, so it touches only the documents the detection step
                  counted.
                connection:
                  uri: "{{ inputs.connection_uri != '' ? inputs.connection_uri :
                    secret('MONGODB_URI') }}"
                database: "{{ inputs.database }}"
                collection: "{{ inputs.collection }}"
                operation: DELETE_MANY
                filter:
                  "{{ inputs.timestamp_field }}":
                    $lt:
                      $date: "{{ render(vars.cutoff) }}"

              - id: log_deleted
                type: io.kestra.plugin.core.log.Log
                message: "Deleted {{ outputs.delete.deletedCount }} document(s). Acknowledged:
                  {{ outputs.delete.wasAcknowledged }}."

              - id: reverify
                type: io.kestra.plugin.mongodb.Find
                description: Read-only. The same age filter again. If the delete landed this
                  returns 0.
                connection:
                  uri: "{{ inputs.connection_uri != '' ? inputs.connection_uri :
                    secret('MONGODB_URI') }}"
                database: "{{ inputs.database }}"
                collection: "{{ inputs.collection }}"
                filter:
                  "{{ inputs.timestamp_field }}":
                    $lt:
                      $date: "{{ render(vars.cutoff) }}"
                limit: "{{ inputs.scan_limit }}"
                store: false

              - id: log_reverify
                type: io.kestra.plugin.core.log.Log
                message: "Re-check after delete found {{ outputs.reverify.size ?? 0 }}
                  document(s) still older than {{ inputs.retention_days }}
                  days."

              - id: verify_gate
                type: io.kestra.plugin.core.flow.If
                description: Fail the run if any expired document is still there after the
                  delete.
                condition: "{{ (outputs.reverify.size ?? 0) > 0 }}"
                then:
                  - id: remediation_failed
                    type: io.kestra.plugin.core.execution.Fail
                    errorMessage: >-
                      Delete ran but the re-check still finds {{
                      outputs.reverify.size ?? 0 }} document(s) older than {{
                      inputs.retention_days }} days in {{ inputs.database }}.{{
                      inputs.collection }}. Something is still writing documents
                      with an old {{ inputs.timestamp_field }}. Or the delete
                      did not cover them. The retention policy is not met.
                else:
                  - id: remediated
                    type: io.kestra.plugin.core.log.Log
                    message: >-
                      Removed {{ outputs.delete.deletedCount }} expired
                      document(s) from {{ inputs.database }}.{{
                      inputs.collection }}. The re-check with the same filter
                      returned {{ outputs.reverify.size ?? 0 }} remaining, so
                      the retention window is now clean.
            else:
              - id: declined
                type: io.kestra.plugin.core.log.Log
                message: >-
                  Approval not granted{{ (outputs.hold.onResume.note ?? '') ==
                  '' ? '' : (', note: ' ~ outputs.hold.onResume.note) }}. No
                  document was deleted. The expired documents stay in place.
    else:
      - id: clean
        type: io.kestra.plugin.core.log.Log
        message: >-
          No expired documents. {{ inputs.database }}.{{ inputs.collection }}
          holds 0 documents with {{ inputs.timestamp_field }} before {{
          render(vars.cutoff) }} ({{ inputs.retention_days }} day retention).

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {{ {'text': ('MongoDB retention guard failed on ' ~ inputs.database ~ '.' ~ inputs.collection ~ ' (execution ' ~ execution.id ~ '). Check the logs.')} | toJson }}

triggers:
  - id: daily
    type: io.kestra.plugin.core.trigger.Schedule
    description: >-
      Shipped disabled. Its inputs force safe mode, so even an enabled schedule
      only ever reports and never deletes on its own.
    cron: "0 3 * * *"
    disabled: true
    inputs:
      dry_run: true
      auto_remediate: false

MongoDB has one built-in way to drop old documents: a TTL index on a date field. When that index was never created or someone dropped it, nothing deletes the old documents. The collection grows without bound and data lives well past the policy that was supposed to remove it. A storage alert only fires once the disk is already a problem.

This blueprint reads the collection with a date filter, counts the documents older than the retention window and reports them. Behind a safe mode gate and an approval it deletes them with a single DELETE_MANY, then reads the collection a second time with the same filter and fails the run if any expired document is still there.

This blueprint was created by zkasuran.

Safety

Read-only by default. The first run with the defaults only counts and reports. A delete needs auto_remediate true, dry_run false, then a human answering the Pause. Three conditions, all of them, before a document is removed.
The detect read and every re-check are capped at scan_limit, so a huge expired set cannot exhaust the worker. The delete is a server-side DELETE_MANY, so it is not capped and removes every matching document in one call.
The delete uses the same age filter as the detection step, so it touches only the documents that were counted. It never issues an unfiltered delete.
The re-read is the proof, not the delete count. If any document older than the window is still present after the delete, the run fails rather than reporting success.

How it works

detect (io.kestra.plugin.mongodb.Find) selects documents whose timestamp_field is before the retention cutoff, oldest first, up to scan_limit. The cutoff is execution.startDate minus retention_days days, frozen for the whole run so detect, delete and the re-check all use one value.
report (io.kestra.plugin.core.log.Log) logs the count. Task outputs are not in the execution JSON, so the number the gate acts on is written to the log.
route (io.kestra.plugin.core.flow.If): zero expired documents logs clean and stops. One or more sends alert to Slack and prints planned, which states what a remediation run would do on every run including a dry one.
remediate_gate (io.kestra.plugin.core.flow.If) opens only with auto_remediate true and dry_run false. hold (io.kestra.plugin.core.flow.Pause) waits for an answer.
On approval delete (io.kestra.plugin.mongodb.Delete, DELETE_MANY) removes the documents, reverify (io.kestra.plugin.mongodb.Find) runs the same filter again and verify_gate fails the run with io.kestra.plugin.core.execution.Fail if the re-check still finds any. Without approval declined records the decision and nothing changes.
errors posts to Slack when any task fails, for example when MongoDB is unreachable.

What you get

A count of the documents that are past the retention window, with the oldest one sampled in the log.
A Slack alert the moment expired documents exist, before storage becomes the symptom.
A gated, approved delete of only the expired documents, with the deleted count recorded.
A re-read that fails the run if the retention window is still not clean, so a half-done delete or a writer still inserting old-dated documents is caught rather than hidden.

Who it's for

Teams whose collection lost its TTL index or never had one, so it keeps growing.
SREs who want the retention policy enforced on a schedule with an audit trail per run.
Anyone who needs the delete to be provable, not assumed, after it runs.

Why orchestrate this with Kestra

A TTL index deletes documents quietly and gives no report, no approval step and no proof the policy held. This flow makes each of those explicit. The schedule decides when the policy is enforced, the detection count is logged, the destructive half sits behind auto_remediate, dry_run and a Pause with the decision recorded on the execution. The re-read turns the delete into a checked outcome. The Slack alert and the error path are the same ones the rest of your flows use.

Tested end to end

On Kestra 2.0.5 OSS against a MongoDB 7.0 container, on the final file. The collection was seeded with 5 documents dated 60 days old and 3 dated 5 days old, timestamp field createdAt. Every before and after number was read back with mongosh, not only from the Kestra log.

Run	Inputs	Result
Clean	`retention_days: 90`	0 expired. `clean` logged, no Slack alert, SUCCESS. `mongosh` count still 8
Detect	`retention_days: 30`	5 expired. `alert` to Slack, SUCCESS. `mongosh` confirms 5 older than the cutoff
Dry run	`retention_days: 30`, defaults	Reports 5, `planned` says it would delete, nothing deleted. `mongosh` count still 8
Armed but dry	`auto_remediate: true`, `dry_run: true`	Still no delete. `mongosh` count still 8
Declined	`auto_remediate: true`, `dry_run: false`, resumed `approved: false`	Paused, then `declined`. `mongosh` count still 8
Remediate	`auto_remediate: true`, `dry_run: false`, resumed `approved: true`	Deleted 5, re-check 0, SUCCESS. `mongosh` count 3, all recent
Still dirty	the same, with a writer inserting old-dated documents during the run	Delete ran, re-check found documents still expired, `verify_gate` failed the run and `errors` alerted
Unreachable	`connection_uri` pointed at a dead host	`detect` failed, `errors` posted to Slack, execution FAILED

Does the detector actually fail on a buildup

The clean run and the detect run isolate that. Nothing on the collection changed between them except the retention window: 90 days returned 0 expired, 30 days returned 5, on the same 8 documents. mongosh confirmed it directly with a $lt count against each cutoff. The remediate run then removed those 5 and the re-read returned 0. The still-dirty run proved the gate the other way, failing when documents older than the window remained.

Prerequisites

A reachable MongoDB instance and a user that can read the collection and run a delete on it.
A collection with a date field to age documents on. A TTL index is not required and is in fact the thing this flow stands in for when one is missing.
Local testing: docker run -d --name mdb --network mynet mongo:7.0, seed a mix of old and recent documents, then run the flow with connection_uri: mongodb://mdb:27017.

Secrets

MONGODB_URI: the MongoDB connection string. Read when connection_uri is left empty, which is how the flow should run in production. In Kestra OSS, provide it as the base64-encoded environment variable SECRET_MONGODB_URI.
SLACK_WEBHOOK_URL: used by alert and the errors block. Provide it as SECRET_SLACK_WEBHOOK_URL.

Inputs

Input	Default	Purpose
`connection_uri`	empty	Plain connection string for a local test. Empty reads `MONGODB_URI`
`database`, `collection`	`appdata`, `events`	The collection to guard
`timestamp_field`	`createdAt`	The BSON date field documents are aged on
`retention_days`	`30`	Documents older than this many days are expired
`scan_limit`	`1000`	Ceiling on the detect read and each re-check
`approval_timeout`	`PT2H`	How long the pause waits for an answer
`auto_remediate`	`false`	Allow the delete
`dry_run`	`true`	Must be `false` for a delete to apply

Quick start

Add the MONGODB_URI and SLACK_WEBHOOK_URL secrets.
Set database, collection and timestamp_field, then run the flow. It reports the expired count and stops.
To clear a buildup, set auto_remediate: true and dry_run: false, run it again, then answer the pause. The delete runs and the re-read proves the window is clean.
Enable the daily schedule, which always runs in safe mode.

How to extend

Guard several collections by wrapping the tasks in a io.kestra.plugin.core.flow.Loop over a list, with concurrencyLimit: 1 on the part that deletes.
Replace the Pause with a Slack approval. Or drop it for a collection you are sure about.
Alert only when the expired count crosses a threshold, so a steady trickle stays quiet and a real buildup pages someone.
Once the collection is clean, create the TTL index the flow stands in for, so MongoDB keeps it clean between runs.

MongoDB Expired Document Retention Guard

Safety

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Tested end to end

Does the detector actually fail on a buildup

Prerequisites

Secrets

Inputs

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

MongoDB Expired Document Retention Guard

Safety

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Tested end to end

Does the detector actually fail on a buildup

Prerequisites

Secrets

Inputs

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Infrastructure Blueprints

Incremental, idempotent lake ingestion from MinIO with DuckDB

Catch MinIO Replication Drift Before Objects Go Missing on the Target

Meilisearch Indexing Queue Staleness Sentinel

Meilisearch Index Health Sentinel and Search Audit

Enforce a MinIO bucket lifecycle expiry and noncurrent version policy

Scheduled MinIO Bucket Retention Cleanup

More Data Blueprints

Automate Apache Iceberg Table Maintenance and Snapshot Expiry via REST Catalog

Sync Hubspot CRM data to BigQuery on a schedule

Huawei GeminiDB TTL Hygiene Sentinel

Sync new users from Postgres to HubSpot

Huawei DLI Data Quality Assertion Gate

Load time series data into InfluxDB

New to Kestra?