Cassandra Tombstone Density Auditor

id: cassandra-tombstone-density-auditor
namespace: company.database

description: Audits Apache Cassandra keyspaces for tables with excessive
  tombstone density, alerting database administrators before query read timeouts
  impact client traffic.

concurrency:
  limit: 1

inputs:
  - id: cassandra_contact_points
    type: STRING
    defaults: cassandra-node1.internal
    description: Comma-separated Cassandra node IP addresses or hostnames.

  - id: cassandra_port
    type: INT
    defaults: 9042
    description: Native CQL protocol communication port.

  - id: keyspace_name
    type: STRING
    defaults: production_data
    description: Target Cassandra keyspace to inspect for tombstone density.

  - id: max_tombstones_per_query_warning
    type: INT
    defaults: 1000
    description: Tombstone warning limit per table read operation before dispatching
      an alert.

  - id: slack_channel
    type: STRING
    defaults: "#cassandra-alerts"
    description: Slack channel destination for database health notifications.

tasks:
  - id: audit_tombstone_density
    type: io.kestra.plugin.scripts.python.Script
    containerImage: python:3.11-slim
    beforeCommands:
      - pip install cassandra-driver
    env:
      CASSANDRA_HOST: "{{ inputs.cassandra_contact_points }}"
      CASSANDRA_PORT: "{{ inputs.cassandra_port }}"
      KEYSPACE: "{{ inputs.keyspace_name }}"
      WARNING_THRESHOLD: "{{ inputs.max_tombstones_per_query_warning }}"
      CASSANDRA_USER: "{{ secret('CASSANDRA_USER') }}"
      CASSANDRA_PASSWORD: "{{ secret('CASSANDRA_PASSWORD') }}"
    script: |
      import os
      import json
      from cassandra.cluster import Cluster
      from cassandra.auth import PlainTextAuthProvider
      from kestra import Kestra

      host = os.getenv('CASSANDRA_HOST', '127.0.0.1')
      port = int(os.getenv('CASSANDRA_PORT', 9042))
      keyspace = os.getenv('KEYSPACE')
      threshold = int(os.getenv('WARNING_THRESHOLD', 1000))

      user = os.getenv('CASSANDRA_USER')
      password = os.getenv('CASSANDRA_PASSWORD')

      auth_provider = PlainTextAuthProvider(username=user, password=password) if user else None
      cluster = Cluster(contact_points=[h.strip() for h in host.split(',')], port=port, auth_provider=auth_provider)
      session = cluster.connect()

      # Query system schema for tables and compaction properties
      query = f"SELECT table_name, gc_grace_seconds, compaction FROM system_schema.tables WHERE keyspace_name = '{keyspace}';"
      rows = session.execute(query)

      inspected_tables = []
      flagged_tables = []

      for row in rows:
          table_name = row.table_name
          gc_grace = row.gc_grace_seconds
          compaction = row.compaction

          # Evaluate gc_grace_seconds risk
          # Tables with 0 gc_grace or default 864000 (10 days) subjected to frequent deletes accumulate tombstones
          risk_factor = "NORMAL"
          if gc_grace == 0:
              risk_factor = "HIGH_GC_GRACE_ZERO"
          
          table_info = {
              'table_name': table_name,
              'gc_grace_seconds': gc_grace,
              'compaction_strategy': compaction.get('class', 'SizeTieredCompactionStrategy').split('.')[-1],
              'risk_factor': risk_factor
          }
          inspected_tables.append(table_info)

          if risk_factor != "NORMAL":
              flagged_tables.append(table_info)

      cluster.shutdown()

      total_tables = len(inspected_tables)
      flagged_count = len(flagged_tables)
      has_risk = flagged_count > 0

      Kestra.outputs({
          'total_tables_audited': total_tables,
          'flagged_tables_count': flagged_count,
          'has_tombstone_risk': has_risk,
          'flagged_table_names': [t['table_name'] for t in flagged_tables]
      })

      report = {
          'keyspace': keyspace,
          'total_tables': total_tables,
          'flagged_count': flagged_count,
          'flagged_tables': flagged_tables,
          'all_tables': inspected_tables
      }

      with open('cassandra_tombstone_report.json', 'w') as f:
          json.dump(report, f, indent=2)

    outputFiles:
      - cassandra_tombstone_report.json

  - id: check_tombstone_risk
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.audit_tombstone_density.vars.has_tombstone_risk }}"
    then:
      - id: notify_slack_dba
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "channel": "{{ inputs.slack_channel }}",
            "text": "Cassandra Tombstone Density Alert: {{ outputs.audit_tombstone_density.vars.flagged_tables_count }} table(s) in keyspace `{{ inputs.keyspace_name }}` flagged for tombstone risks.\n• Tables: {{ outputs.audit_tombstone_density.vars.flagged_table_names | join(', ') }}\n• Remediation: Execute `nodetool garbagecollect` or adjust gc_grace_seconds to prevent query read timeouts."
          }
    else:
      - id: log_tables_healthy
        type: io.kestra.plugin.core.log.Log
        message: "All {{ outputs.audit_tombstone_density.vars.total_tables_audited }}
          tables in keyspace {{ inputs.keyspace_name }} have compliant
          compaction and gc_grace settings."

  - id: export_audit_manifest
    type: io.kestra.plugin.core.execution.Labels
    labels:
      keyspace: "{{ inputs.keyspace_name }}"
      flagged_tables_count: "{{
        outputs.audit_tombstone_density.vars.flagged_tables_count | string }}"
      health_status: "{{ outputs.audit_tombstone_density.vars.has_tombstone_risk ?
        'RISK_DETECTED' : 'HEALTHY' }}"

triggers:
  - id: weekly_cassandra_audit
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 5 * * 0"