Cassandra Missing Default TTL Growth Guard with Kestra

id: cassandra-missing-default-ttl-growth-guard
namespace: company.team

description: >-
  Find base tables in keyspaces that are meant to expire their data but were
  created with default_time_to_live = 0, so every write stays forever and the
  keyspace grows without bound. You pass the keyspaces where expiry is expected.
  The guard reads system_schema.tables for those keyspaces and flags every base
  table whose default TTL is still 0. It alerts Slack. When you allow it, it
  sets a bounded default TTL with ALTER TABLE one table at a time, then reads
  the schema again and fails the run if any flagged table still reads 0.
  default_time_to_live applies to NEW writes only. This is a schema setting
  check. It does not expire or reclaim rows already on disk.

inputs:
  - id: cassandra_host
    type: STRING
    defaults: "cassandra"
    displayName: Cassandra Host
    description: Hostname or IP of a Cassandra contact point reachable from the
      Kestra Worker.

  - id: cassandra_port
    type: INT
    defaults: 9042
    displayName: Cassandra CQL Port
    description: Native CQL port the contact point listens on.

  - id: cassandra_datacenter
    type: STRING
    defaults: "datacenter1"
    displayName: Local Datacenter
    description: Datacenter the driver treats as local. The official Cassandra image
      uses datacenter1.

  - id: target_keyspaces
    type: ARRAY
    itemType: STRING
    displayName: Target keyspaces
    description: Keyspaces whose base tables are expected to expire data, for
      example time series, event or audit keyspaces. System keyspaces are never
      scanned.
    defaults:
      - events

  - id: default_ttl_seconds
    type: INT
    displayName: Default TTL to apply (seconds)
    description: Bounded default_time_to_live set on a flagged table during
      remediation. 2592000 is 30 days.
    defaults: 2592000

  - id: auto_remediate
    type: BOOL
    displayName: Auto remediate
    defaults: false

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

variables:
  scan_cql: >-
    SELECT keyspace_name, table_name, default_time_to_live FROM
    system_schema.tables WHERE keyspace_name IN ({{ inputs.target_keyspaces |
    jq('if length == 0 then "\u0027\u0027" else map("\u0027" + gsub("\u0027";
    "\u0027\u0027") + "\u0027") | join(",") end') | first }})
  fix_jq: >-
    [.[] | select(((.default_time_to_live // 0) | tonumber) == 0) | "ALTER TABLE
    \"" + (.keyspace_name | gsub("\"";"\"\"")) + "\".\"" + (.table_name |
    gsub("\"";"\"\"")) + "\" WITH default_time_to_live = {{
    inputs.default_ttl_seconds }}"]
tasks:
  - id: scan_tables
    type: io.kestra.plugin.cassandra.standard.Query
    description: Read only. Lists base tables and their default_time_to_live for the
      target keyspaces from system_schema.tables, restricted by the
      keyspace_name partition key so no ALLOW FILTERING and no system scan is
      needed.
    session:
      endpoints:
        - hostname: "{{ inputs.cassandra_host }}"
          port: "{{ inputs.cassandra_port }}"
      localDatacenter: "{{ inputs.cassandra_datacenter }}"
    cql: "{{ render(vars.scan_cql) }}"
    fetchType: FETCH

  - id: flagged
    type: io.kestra.plugin.core.output.OutputValues
    description: Build the list of ALTER statements and a readable name list from
      the base tables whose default TTL is 0.
    values:
      fix_sql: "{{ outputs.scan_tables.rows | jq(render(vars.fix_jq)) | first | toJson
        }}"
      names: "{{ outputs.scan_tables.rows | jq('[.[] | select(((.default_time_to_live
        // 0) | tonumber) == 0) | .keyspace_name + \".\" + .table_name] |
        join(\", \")') | first }}"

  - id: summary
    type: io.kestra.plugin.core.log.Log
    message: >-
      TTL growth guard on {{ inputs.cassandra_host }}: scanned {{
      outputs.scan_tables.size }} base tables in {{ inputs.target_keyspaces |
      jq('join(", ")') | first }}, {{ outputs.scan_tables.rows | jq('[.[] |
      select(((.default_time_to_live // 0) | tonumber) == 0)] | length') | first
      }} still at default_time_to_live = 0.

  - id: route
    type: io.kestra.plugin.core.flow.If
    condition: "{{ (outputs.scan_tables.rows | jq('[.[] |
      select(((.default_time_to_live // 0) | tonumber) == 0)] | length') |
      first) > 0 }}"
    then:
      - id: alert
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {{ {'text': 'Cassandra ' ~ inputs.cassandra_host ~ ': base tables in expiry keyspaces still at default_time_to_live = 0, so they grow without bound: ' ~ outputs.flagged.values.names ~ '. ' ~ (inputs.auto_remediate and not inputs.dry_run ? 'Setting default_time_to_live = ' ~ inputs.default_ttl_seconds ~ 's now.' : 'No change applied (auto_remediate=' ~ inputs.auto_remediate ~ ' dry_run=' ~ inputs.dry_run ~ ').')} | toJson }}

      - id: remediate_gate
        type: io.kestra.plugin.core.flow.If
        condition: "{{ inputs.auto_remediate and not inputs.dry_run }}"
        then:
          - id: remediate
            type: io.kestra.plugin.core.flow.Loop
            description: Sequential remediation on purpose. Concurrent ALTER TABLE in one
              keyspace race Cassandra schema agreement and can leave nodes
              disagreeing on the schema.
            concurrencyLimit: 1
            values: "{{ outputs.flagged.values.fix_sql }}"
            tasks:
              - id: apply_ttl
                type: io.kestra.plugin.cassandra.standard.Query
                description: One ALTER TABLE ... WITH default_time_to_live per flagged base
                  table.
                allowFailure: true
                session:
                  endpoints:
                    - hostname: "{{ inputs.cassandra_host }}"
                      port: "{{ inputs.cassandra_port }}"
                  localDatacenter: "{{ inputs.cassandra_datacenter }}"
                cql: "{{ item.value }}"

          - id: reverify
            type: io.kestra.plugin.cassandra.standard.Query
            description: Read system_schema.tables again for the same keyspaces to confirm
              the default TTL landed in the schema.
            session:
              endpoints:
                - hostname: "{{ inputs.cassandra_host }}"
                  port: "{{ inputs.cassandra_port }}"
              localDatacenter: "{{ inputs.cassandra_datacenter }}"
            cql: "{{ render(vars.scan_cql) }}"
            fetchType: FETCH

          - id: verify_result
            type: io.kestra.plugin.core.flow.If
            condition: "{{ (outputs.reverify.rows | jq('[.[] |
              select(((.default_time_to_live // 0) | tonumber) == 0)] | length')
              | first) > 0 }}"
            then:
              - id: remediation_failed
                type: io.kestra.plugin.core.execution.Fail
                errorMessage: "Base tables still read default_time_to_live = 0 after ALTER
                  TABLE: {{ outputs.reverify.rows | jq('[.[] |
                  select(((.default_time_to_live // 0) | tonumber) == 0) |
                  .keyspace_name + \".\" + .table_name] | join(\", \")') | first
                  }}. The connecting role needs ALTER on these tables."
            else:
              - id: remediated
                type: io.kestra.plugin.core.log.Log
                message: "Set and verified default_time_to_live = {{ inputs.default_ttl_seconds
                  }}s on {{ outputs.flagged.values.names }}. This changes NEW
                  writes only. Rows written before the change keep their own
                  TTL, so existing data is not expired by this run."
    else:
      - id: clean
        type: io.kestra.plugin.core.log.Log
        message: "Every base table in {{ inputs.target_keyspaces | jq('join(\", \")') |
          first }} has a non-zero default_time_to_live."
errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alert Slack when the guard itself fails, for example an unreachable
      cluster or a role without ALTER rights, so a broken run never reads as a
      clean one.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {{ {'text': 'Cassandra default TTL growth guard failed on ' ~ inputs.cassandra_host ~ ' (execution ' ~ execution.id ~ '). The TTL posture is now unknown. Check the task logs.'} | toJson }}

triggers:
  - id: daily
    type: io.kestra.plugin.core.trigger.Schedule
    description: Shipped disabled and always in safe mode (auto_remediate false,
      dry_run true) regardless of this flow's own defaults.
    cron: "15 4 * * *"
    disabled: true
    inputs:
      auto_remediate: false
      dry_run: true

A Cassandra table created without default_time_to_live keeps that setting at 0, which means no expiry. For a reference table that is correct. For a time series, event log or cache keyspace it is a slow outage: every write lives forever, SSTables keep growing, so compaction and repair cost climb until a node runs out of disk. Nothing in the cluster warns you, because 0 is a legal and common value.

This blueprint reads system_schema.tables for the keyspaces you name as expiry-expected, flags every base table whose default_time_to_live is still 0 and alerts Slack. When you allow it, it sets a bounded default TTL with ALTER TABLE ... WITH default_time_to_live, one table at a time, then reads the schema again and fails the run if any flagged table still reads 0.

This blueprint was created by zkasuran.

Safety

default_time_to_live applies to NEW writes only. Setting it does not expire, delete or reclaim rows already on disk. It does not change the per-row TTL of existing data. The re-verify step is a schema check: it confirms system_schema.tables now shows the bounded value, not that any disk was freed. To age out data already written you still need a per-row UPDATE ... USING TTL or a delete, which this guard deliberately does not do.

The remediation loop runs one ALTER TABLE at a time (concurrencyLimit: 1). Concurrent schema changes in one keyspace race Cassandra's schema agreement and can leave nodes disagreeing on the schema. Both write gates (auto_remediate and dry_run) must be set before any ALTER TABLE runs.

How it works

scan_tables (io.kestra.plugin.cassandra.standard.Query, read only) runs SELECT keyspace_name, table_name, default_time_to_live FROM system_schema.tables WHERE keyspace_name IN (...) for the target_keyspaces you pass. The restriction is on the keyspace_name partition key, so the read needs no ALLOW FILTERING and never touches a system keyspace. system_schema.tables lists base tables only, so views and indexes are out of scope by construction.
flagged (io.kestra.plugin.core.output.OutputValues) keeps the rows at default_time_to_live = 0 and builds both an ALTER TABLE statement per table and a readable name list. summary logs the counts.
route (io.kestra.plugin.core.flow.If) branches on whether any table is flagged:
- clean: clean logs that every base table in the target keyspaces has a non-zero default TTL.
- flagged: alert posts the table list to Slack.
  - remediate_gate runs only when auto_remediate is true and dry_run is false.
  - remediate (io.kestra.plugin.core.flow.Loop, concurrencyLimit: 1) runs each ALTER TABLE ... WITH default_time_to_live in turn.
  - reverify re-reads system_schema.tables. verify_result fails the run through io.kestra.plugin.core.execution.Fail if any flagged table still reads 0.
The errors block alerts Slack when the flow itself fails, for example an unreachable cluster or a role without ALTER rights.
The daily Schedule trigger is shipped disabled: true and always forces safe mode.

What you get

A standing answer to which base tables in your expiry keyspaces have no default TTL, read from Cassandra's own schema rather than guessed.
A bounded default TTL applied only behind two gates, one table at a time, with a re-read of the schema as proof.
A Slack page the moment an unbounded table appears, plus a separate failure channel when the guard cannot run.
A clear line between a schema fix (what this does) and reclaiming existing rows (what it does not).

Who it's for

Platform and database reliability teams running self-managed Cassandra who treat unbounded disk growth as an outage waiting to happen.
Teams onboarding new keyspaces where a CREATE TABLE without a default TTL is an easy and expensive miss.
Anyone evaluating the Cassandra plugin who wants a worked system_schema.tables audit that also drives ALTER TABLE.

Why orchestrate this with Kestra

A one-off DESCRIBE in cqlsh answers the question for one table at one moment and fixes nothing. Kestra reads the whole set on a schedule, turns it into a Slack page and an execution history, applies the fix only behind auto_remediate and dry_run, one statement at a time, then verifies the result by reading the schema back. The guard becomes a recurring check instead of a thing someone remembers to run.

Tested end to end

On Kestra 2.0.5 OSS against Cassandra 5.0 (cassandra:5) on the same Docker network, with a local HTTP sink standing in for Slack. Seeded schema:

keyspace metrics_clean (expiry-expected): table rollup_daily created WITH default_time_to_live = 86400;
keyspace metrics (expiry-expected): table readings created without a TTL, so default_time_to_live = 0;
keyspace events (expiry-expected): table audit_log created without a TTL, so default_time_to_live = 0.

Run	Inputs	Result
1	`target_keyspaces: [metrics_clean]`	Clean. Its one base table has a non-zero default TTL. No alert
2	`target_keyspaces: [metrics, events]`	2 base tables flagged (`metrics.readings`, `events.audit_log`). Alert sent. `system_schema.tables` still shows 0
3	run 2 keyspaces, `auto_remediate: true`, `dry_run: true`	Same report. cqlsh confirms both still read 0. No ALTER ran
4	run 2 keyspaces, `auto_remediate: true`, `dry_run: false`, `default_ttl_seconds: 2592000`	Both `ALTER TABLE` ran one at a time. `reverify` found 0 still at 0. cqlsh confirms both now read `2592000`
5	`cassandra_host` pointed at an unreachable name	`scan_tables` failed, the `errors` block posted to the sink, execution FAILED

Before and after values from the cluster's own CLI are in the PR description.

Prerequisites

A reachable Cassandra cluster with its native CQL port open to the Kestra Worker. For the audit, any authenticated role can read system_schema.tables. For remediation, a role with ALTER permission on the target tables.
Local testing: docker run -d --name cassandra -e MAX_HEAP_SIZE=1G -e HEAP_NEWSIZE=256M cassandra:5, connect it to the Kestra network, then create a keyspace and a table without a TTL. Allow a minute or two for the single node to accept CQL.

Secrets

SLACK_WEBHOOK_URL: Slack incoming webhook used by alert and the errors block.
This flow connects without a username or password, matching this repo's existing Cassandra blueprints against an unauthenticated local instance. For an authenticated cluster, add username and password to each task's session from {{ secret(...) }} values rather than hardcoding them.
In Kestra OSS, provide a secret as a base64-encoded environment variable prefixed SECRET_, for example SECRET_SLACK_WEBHOOK_URL.

Inputs

Input	Default	Purpose
`cassandra_host`	`cassandra`	Contact point hostname or IP
`cassandra_port`	`9042`	Native CQL port
`cassandra_datacenter`	`datacenter1`	Local datacenter for the driver
`target_keyspaces`	`[events]`	Keyspaces whose base tables are expected to expire data
`default_ttl_seconds`	`2592000`	Bounded default TTL applied on remediation (30 days)
`auto_remediate`	`false`	Allow `ALTER TABLE`
`dry_run`	`true`	Must be `false` for a statement to run

Quick start

Add the SLACK_WEBHOOK_URL secret.
Import the flow and set target_keyspaces to the keyspaces whose tables are meant to expire data.
Run it once with the defaults and read which base tables have no default TTL.
Agree a bounded default_ttl_seconds with the owners, then run with auto_remediate: true and dry_run: false.
Enable the daily trigger, which always runs in safe mode.

How to extend

Add a Pause before the loop for a human approval step on production clusters.
Set a different TTL per keyspace from a map, for example longer for audit tables than for metrics.
Also flag tables whose gc_grace_seconds is 0, which risks resurrecting deleted rows.
Pair with a separate flow that runs UPDATE ... USING TTL to age out rows written before the default TTL was set.

Cassandra Missing Default TTL Growth Guard

Safety

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Tested end to end

Prerequisites

Secrets

Inputs

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Cassandra Missing Default TTL Growth Guard

Safety

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Tested end to end

Prerequisites

Secrets

Inputs

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Infrastructure Blueprints

Clusterify

Fetch logs from all pods in a Kubernetes namespace

Kafka ACL Overly Broad Grant Audit

Kubernetes Deployment Restart Remediation from Alerts

Kubernetes Pod Lifecycle Management: Create, Inspect, and Cleanup

Kubernetes OOMKilled Watchdog with Approval-Gated Memory Right-Sizing

More Data Blueprints

Cache a computationally expensive task in an ETL pipeline using the taskCache property

Apache Cassandra Compaction Backlog Sentinel

Cassandra Superuser Login Exposure Sentinel

Cassandra Hinted Handoff Storm Guard

Cassandra Tombstone Density Auditor

Ceph Cluster Health Gate

New to Kestra?