Cassandra Unbounded Row Cache Guard with Kestra

id: cassandra-row-cache-unbounded-guard
namespace: company.team

description: >-
  Find Cassandra tables in the keyspaces you name whose caching option sets
  rows_per_partition to ALL or to a number above a ceiling you set. ALL tells
  Cassandra to cache every row of every partition it reads, which is unbounded
  on a wide partition table, so once the row cache is actually turned on the
  cache churns and the heap takes the cost. The guard reads system_schema.tables
  for those keyspaces, flags every table above the ceiling and alerts Slack.
  When you allow it, it rewrites the caching option with ALTER TABLE one table
  at a time behind a human approval, then reads the schema again and fails the
  run if any flagged table is still unbounded. It keeps each table's existing
  keys cache setting.

inputs:
  - id: cassandra_host
    type: STRING
    defaults: "cassandra"
    displayName: Cassandra Host
    description: Hostname or IP of one 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 for its load balancing
      policy. A default Cassandra install uses datacenter1.

  - id: target_keyspaces
    type: ARRAY
    itemType: STRING
    displayName: Target keyspaces
    description: Keyspaces to audit. Any name starting with system is dropped from
      the list before the query runs, so the built in keyspaces are never in
      scope.
    defaults:
      - app

  - id: rows_per_partition_ceiling
    type: INT
    displayName: rows_per_partition ceiling
    description: Highest numeric rows_per_partition a table may carry. ALL is always
      flagged because it has no bound at all. NONE is always allowed. Keep the
      remediation value at or below this number.
    defaults: 1000

  - id: remediation_rows_per_partition
    type: SELECT
    displayName: Remediation rows_per_partition
    description: Value written into the caching option for each flagged table. NONE
      turns row caching off for that table. A number caps how many rows per
      partition may be cached.
    values:
      - "NONE"
      - "100"
      - "1000"
      - "10000"
    defaults: "NONE"

  - 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 true, before the
      approval step is even reached.
    defaults: true

variables:
  scoped_keyspaces: >-
    {{ inputs.target_keyspaces | jq('map(select(startswith("system") | not)) |
    if length == 0 then "none" else join(", ") end') | first }}
  scan_cql: >-
    SELECT keyspace_name, table_name, caching FROM system_schema.tables WHERE
    keyspace_name IN ({{ inputs.target_keyspaces |
    jq('map(select(startswith("system") | not)) | if length == 0 then "\u0027
    \u0027" else map("\u0027" + gsub("\u0027"; "\u0027\u0027") + "\u0027") |
    join(",") end') | first }})
  flag_count_jq: >-
    [.[] | select((.caching.rows_per_partition // "NONE") as $r | ($r == "ALL")
    or (($r != "NONE") and (($r | tonumber) > {{
    inputs.rows_per_partition_ceiling }})))] | length
  flag_names_jq: >-
    [.[] | select((.caching.rows_per_partition // "NONE") as $r | ($r == "ALL")
    or (($r != "NONE") and (($r | tonumber) > {{
    inputs.rows_per_partition_ceiling }}))) | .keyspace_name + "." + .table_name
    + " (rows_per_partition=" + (.caching.rows_per_partition // "NONE") + ")"] |
    join(", ")
  fix_jq: >-
    [.[] | select((.caching.rows_per_partition // "NONE") as $r | ($r == "ALL")
    or (($r != "NONE") and (($r | tonumber) > {{
    inputs.rows_per_partition_ceiling }}))) | "ALTER TABLE \"" + (.keyspace_name
    | gsub("\"";"\"\"")) + "\".\"" + (.table_name | gsub("\"";"\"\"")) + "\"
    WITH caching = {'keys': '" + (.caching.keys // "ALL") + "',
    'rows_per_partition': '{{ inputs.remediation_rows_per_partition }}'}"]

tasks:
  - id: scan_tables
    type: io.kestra.plugin.cassandra.standard.Query
    description: Read only. Lists each table's caching option for the target
      keyspaces from system_schema.tables, restricted on the keyspace_name
      partition key so no ALLOW FILTERING 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: Keep the tables above the ceiling, build one ALTER TABLE per table
      and a readable name list.
    values:
      fix_sql: "{{ outputs.scan_tables.rows | jq(render(vars.fix_jq)) | first | toJson
        }}"
      names: "{{ outputs.scan_tables.rows | jq(render(vars.flag_names_jq)) | first }}"

  - id: summary
    type: io.kestra.plugin.core.log.Log
    message: >-
      Row cache guard on {{ inputs.cassandra_host }}: scanned {{
      outputs.scan_tables.size }} tables in {{ render(vars.scoped_keyspaces) }},
      {{ outputs.scan_tables.rows | jq(render(vars.flag_count_jq)) | first }}
      above the rows_per_partition ceiling of {{
      inputs.rows_per_partition_ceiling }}.

  - id: route
    type: io.kestra.plugin.core.flow.If
    condition: "{{ (outputs.scan_tables.rows | jq(render(vars.flag_count_jq)) |
      first) > 0 }}"
    then:
      - id: alert
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {{ {'text': 'Cassandra ' ~ inputs.cassandra_host ~ ': tables caching more rows per partition than the ceiling of ' ~ inputs.rows_per_partition_ceiling ~ ', so an enabled row cache has no bound on them: ' ~ outputs.flagged.values.names ~ '. ' ~ (inputs.auto_remediate and not inputs.dry_run ? 'Asking for approval to set rows_per_partition to ' ~ inputs.remediation_rows_per_partition ~ '.' : '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: hold
            type: io.kestra.plugin.core.flow.Pause
            description: Human approval before the schema changes. Row caching is sometimes
              set to ALL on purpose for a tiny static table, so a person
              confirms the list first.
            onResume:
              - id: approved
                type: BOOL
                required: false
                description: Set true to rewrite the caching option on the listed tables.
                  Anything else leaves the schema alone.

          - id: approval_gate
            type: io.kestra.plugin.core.flow.If
            condition: "{{ outputs.hold.onResume.approved ?? false }}"
            then:
              - id: remediate
                type: io.kestra.plugin.core.flow.Loop
                description: Sequential remediation on purpose. Concurrent schema changes race
                  Cassandra schema agreement and can leave nodes disagreeing on
                  the schema.
                concurrencyLimit: 1
                values: "{{ outputs.flagged.values.fix_sql }}"
                tasks:
                  - id: apply_caching
                    type: io.kestra.plugin.cassandra.standard.Query
                    description: One ALTER TABLE ... WITH caching per flagged table. The statement
                      carries that table's existing keys setting, so only
                      rows_per_partition changes.
                    allowFailure: true
                    session:
                      endpoints:
                        - hostname: "{{ inputs.cassandra_host }}"
                          port: "{{ inputs.cassandra_port }}"
                      localDatacenter: "{{ inputs.cassandra_datacenter }}"
                    cql: "{{ item.value }}"

              - id: settle
                type: io.kestra.plugin.core.flow.Sleep
                description: Give the cluster a moment to reach schema agreement before the
                  re-read, so a slow propagating ALTER does not read back as a
                  false failure.
                duration: PT5S

              - id: reverify
                type: io.kestra.plugin.cassandra.standard.Query
                description: Read system_schema.tables again for the same keyspaces to confirm
                  no table is left above the ceiling.
                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(render(vars.flag_count_jq)) | first)
                  > 0 }}"
                then:
                  - id: remediation_failed
                    type: io.kestra.plugin.core.execution.Fail
                    errorMessage: "Tables still cache more rows per partition than the ceiling after
                      ALTER TABLE: {{ outputs.reverify.rows |
                      jq(render(vars.flag_names_jq)) | first }}. Possible
                      causes: the role lacks ALTER on these tables; a table was
                      created above the ceiling while the run waited for
                      approval; remediation_rows_per_partition is set above
                      rows_per_partition_ceiling."
                else:
                  - id: remediated
                    type: io.kestra.plugin.core.log.Log
                    message: "Set and verified rows_per_partition = {{
                      inputs.remediation_rows_per_partition }} on {{
                      outputs.flagged.values.names }}. This changes the schema
                      option only."
            else:
              - id: declined
                type: io.kestra.plugin.core.log.Log
                message: "Approval was not given, so the caching option was left as it is on {{
                  outputs.flagged.values.names }}."
    else:
      - id: clean
        type: io.kestra.plugin.core.log.Log
        message: "Every table in {{ render(vars.scoped_keyspaces) }} caches at most {{
          inputs.rows_per_partition_ceiling }} rows per partition."

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 row cache guard failed on ' ~ inputs.cassandra_host ~ ' (execution ' ~ execution.id ~ '). The row cache 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: "40 4 * * *"
    disabled: true
    inputs:
      auto_remediate: false
      dry_run: true

CREATE TABLE ... WITH caching = {'keys': 'ALL', 'rows_per_partition': 'ALL'} is one line in a schema file that nobody reads again. It tells Cassandra that when the row cache holds anything for this table, it may hold every row of every partition it touches. On a wide partition table that is not a cache size, it is an instruction with no bound. Turn the row cache on later and that table can fill it on its own, so entries for every other table get evicted, the cache churns instead of serving reads and the heap pays for all of it.

This blueprint reads system_schema.tables for the keyspaces you name, flags every table whose caching['rows_per_partition'] is ALL or a number above a ceiling you set and alerts Slack. When you allow it, it rewrites the caching option with ALTER TABLE ... WITH caching, one table at a time, behind a human approval, then reads the schema again and fails the run if any flagged table is still unbounded. Each statement carries that table's existing keys value, so the key cache setting is preserved.

This blueprint was created by zkasuran.

Findings

Be clear about the size of this problem, because the honest version is still worth fixing. Cassandra ships with row_cache_size_in_mb: 0 in cassandra.yaml, which means the row cache is off. On a cluster at that default, rows_per_partition: 'ALL' costs nothing today. It is a landmine, not a fire: the day somebody raises row_cache_size_in_mb to make one hot lookup faster, every table carrying ALL starts competing for that memory with no bound. The operator who turns the cache on is rarely the person who wrote the schema. Finding the setting is cheap now. Finding it from a GC log during an incident is not.

So this flow does not claim to measure heap pressure or to prove an OOM. What it does is exact, reads back from the schema and is checkable with cqlsh: it tells you which tables carry an unbounded row cache instruction, then caps them.

How it works

scan_tables (io.kestra.plugin.cassandra.standard.Query, read only) runs SELECT keyspace_name, table_name, caching FROM system_schema.tables WHERE keyspace_name IN (...) for the target_keyspaces you pass. Any name starting with system is dropped from the list first, so the built in keyspaces are never in scope. The restriction is on the keyspace_name partition key, so the read needs no ALLOW FILTERING.
flagged (io.kestra.plugin.core.output.OutputValues) keeps the rows whose rows_per_partition is ALL or a number above rows_per_partition_ceiling, then builds one ALTER TABLE per table plus a readable name list. summary logs the counts.
route (io.kestra.plugin.core.flow.If) branches on whether any table is above the ceiling:
- clean: clean logs that every table in the target keyspaces is bounded.
- flagged: alert posts the table list to Slack.
  - remediate_gate runs only when auto_remediate is true with dry_run false.
  - hold (io.kestra.plugin.core.flow.Pause) waits for a person to approve the list.
  - remediate (io.kestra.plugin.core.flow.Loop, concurrencyLimit: 1) runs each ALTER TABLE ... WITH caching through io.kestra.plugin.cassandra.standard.Query in turn.
  - settle waits a few seconds for schema agreement, then reverify re-reads system_schema.tables. verify_result fails the run through io.kestra.plugin.core.execution.Fail if any table still reads above the ceiling.
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 ships disabled: true and always forces safe mode.

What you get

A standing answer to which tables carry an unbounded row cache instruction, read from Cassandra's own schema rather than from the schema file somebody committed two years ago.
A cap applied only behind two input gates plus a human approval, one table at a time, with a re-read of the schema as proof.
A Slack page listing the tables, plus a separate failure channel when the guard cannot run.
keys left exactly as it was on every table, because the key cache is on by default and is not the problem here.

Who it's for

Platform and database reliability teams who want the row cache to be a tuning option rather than a surprise.
Teams about to raise row_cache_size_in_mb who need to know what will compete for it before they do.
Anyone reviewing inherited keyspaces where CREATE TABLE statements were copied between services.

Why orchestrate this with Kestra

A DESCRIBE TABLE in cqlsh answers this for one table at one moment and changes nothing. Kestra reads the whole set on a schedule, turns it into a Slack page and an execution history, applies the cap only behind auto_remediate with dry_run plus a human approval, one statement at a time, then verifies by reading the schema back. A table created with ALL next month is caught by the next scheduled run instead of by an incident.

Tested end to end

On Kestra 2.0.5 against cassandra:5 (5.0.9) on the same Docker network, with a local HTTP sink standing in for Slack. Seeded keyspace app: sessions altered to rows_per_partition: 'ALL', events created with {'keys':'NONE','rows_per_partition':'5000'}, profiles at '100', lookup_small at the default 'NONE'. Keyspace app_clean holds one table at the default.

Run	Inputs	Result
1	`target_keyspaces: [app_clean]`, ceiling 1000	Clean. 1 table scanned, 0 flagged. No alert, no change
2	`target_keyspaces: [app]`	4 scanned, 2 flagged (`app.sessions` at ALL, `app.events` at 5000). Alert sent. `profiles` at 100 not flagged. Schema unchanged
3	run 2 plus `auto_remediate: true`, `dry_run: true`	Same report. cqlsh confirms all 4 unchanged. No ALTER ran, the Pause was never reached
4	`auto_remediate: true`, `dry_run: false`, resumed with no answer	Pause reached, `declined` branch taken, schema untouched
5	same, approved	2 ALTER TABLE ran one at a time, `reverify` found 0 above the ceiling. cqlsh: `sessions` ALL to NONE with `keys: ALL` kept, `events` 5000 to NONE with `keys: NONE` kept, `profiles` still 100, `lookup_small` still NONE
6	same, with `app.hotpath` created at ALL while the run waited for approval	The 2 known tables were fixed, `reverify` found `app.hotpath`, the run FAILED and the `errors` block posted
7	`target_keyspaces: [system, system_schema, app_clean]`	1 table scanned. The system names are dropped before the query runs
8	`target_keyspaces: [system, system_auth]`	0 tables scanned, clean, no failure
9	`cassandra_host` pointed at an unreachable name	`scan_tables` failed, the `errors` block posted to the sink, execution FAILED

Across the whole cluster, cqlsh reported 0 tables carrying rows_per_partition: 'ALL' after the last run, including every system* keyspace, which ships none. Before and after values from cqlsh are in the PR description.

Prerequisites

A reachable Cassandra cluster with its native CQL port open to the Kestra Worker. For the audit, a role that can read system_schema.tables. For remediation, a role with ALTER on the target tables.
The bundled Cassandra plugin (io.kestra.plugin.cassandra.standard.Query), which ships with Kestra.
Local testing: docker run -d --name cassandra --network <kestra-network> -e MAX_HEAP_SIZE=1G -e HEAP_NEWSIZE=256M cassandra:5, then create a keyspace and a table WITH caching = {'keys':'ALL','rows_per_partition':'ALL'}. Allow a minute or two before the node accepts CQL.

Secrets

SLACK_WEBHOOK_URL: Slack incoming webhook used by alert and the errors block.
This flow connects without a username or password, which matches a local single node Cassandra. For an authenticated cluster, add a username plus a password: {{ secret('CASSANDRA_PASSWORD') }} to each task's session rather than hardcoding them. The driver needs both or neither.
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`	`[app]`	Keyspaces to audit. Names starting with `system` are dropped
`rows_per_partition_ceiling`	`1000`	Highest numeric value allowed. `ALL` is always flagged
`remediation_rows_per_partition`	`NONE`	Value written by the fix. Keep it at or below the ceiling
`auto_remediate`	`false`	Allow `ALTER TABLE`
`dry_run`	`true`	Must be `false` before the approval step is reached

Quick start

Add the SLACK_WEBHOOK_URL secret. The Cassandra plugin ships with Kestra, so there is nothing extra to install.
Import the flow and set target_keyspaces to your application keyspaces.
Run it once with the defaults and read which tables carry an unbounded row cache instruction.
Agree the list with the table owners, then run with auto_remediate: true and dry_run: false and approve the Pause.
Enable the daily trigger, which always runs in safe mode.

How to extend

Set a different ceiling per keyspace from a map, for example a higher allowance for a small reference keyspace.
Keep an allowlist of tables that are meant to carry ALL, for example a tiny static lookup table, then skip them in flagged.
Flag the reverse case too: a table at keys: 'NONE' in a keyspace where the key cache is expected to help.
Pair with a flow that reads row_cache_size_in_mb per node and refuses to raise it while any table still carries ALL.

Cassandra Unbounded Row Cache Guard

Findings

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 Unbounded Row Cache Guard

Findings

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

More Data Blueprints

New to Kestra?