Cloud Spanner Schema Governance Guard

id: gcp-spanner-schema-governance-guard
namespace: company.team

description: |
  Find Cloud Spanner base tables that keep timestamped data but carry no row deletion policy, so
  rows pile up with no retention. The guard reads INFORMATION_SCHEMA, reports every table that has
  the retention timestamp column and a null ROW_DELETION_POLICY_EXPRESSION, alerts Slack and can
  add the policy with ALTER TABLE ADD ROW DELETION POLICY one table at a time, then reads
  INFORMATION_SCHEMA again to prove each policy is present. Runs against the Cloud Spanner emulator
  for a safe demo and against managed Spanner in production.

inputs:
  - id: project_id
    type: STRING
    displayName: GCP project id
    defaults: "test-project"

  - id: instance_id
    type: STRING
    displayName: Spanner instance id
    defaults: "w26-inst"

  - id: database_id
    type: STRING
    displayName: Spanner database id
    defaults: "governance"

  - id: emulator_host
    type: STRING
    displayName: Spanner emulator host and port
    description: Set to the emulator address like localhost:9010. For managed
      Spanner, delete the emulatorHost lines from the tasks so the service
      account is used instead.
    defaults: "localhost:9010"

  - id: retention_column
    type: STRING
    displayName: Retention timestamp column
    description: A base table with a TIMESTAMP column of this name is expected to
      carry a row deletion policy.
    defaults: "created_at"

  - id: retention_days
    type: INT
    displayName: Retention window in days
    description: The OLDER_THAN interval used when a missing policy is added.
    defaults: 90

  - 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 the ALTER to run.
    defaults: true

variables:
  detect_sql: |
    SELECT t.TABLE_NAME AS table_name,
           c.COLUMN_NAME AS retention_column,
           CONCAT('ALTER TABLE ', t.TABLE_NAME,
                  ' ADD ROW DELETION POLICY (OLDER_THAN(', c.COLUMN_NAME,
                  ', INTERVAL {{ inputs.retention_days }} DAY))') AS fix_sql
      FROM INFORMATION_SCHEMA.TABLES t
      JOIN INFORMATION_SCHEMA.COLUMNS c
        ON c.TABLE_SCHEMA = t.TABLE_SCHEMA AND c.TABLE_NAME = t.TABLE_NAME
     WHERE t.TABLE_SCHEMA = ''
       AND t.TABLE_TYPE = 'BASE TABLE'
       AND c.COLUMN_NAME = '{{ inputs.retention_column }}'
       AND c.SPANNER_TYPE = 'TIMESTAMP'
       AND t.ROW_DELETION_POLICY_EXPRESSION IS NULL
     ORDER BY t.TABLE_NAME

tasks:
  - id: detect
    type: io.kestra.plugin.gcp.spanner.Query
    description: >-
      Read-only. Lists base tables in the user schema that have the retention
      timestamp column but no row deletion policy. Each row carries a ready
      ALTER statement.
    projectId: "{{ inputs.project_id }}"
    instanceId: "{{ inputs.instance_id }}"
    databaseId: "{{ inputs.database_id }}"
    emulatorHost: "{{ inputs.emulator_host }}"
    serviceAccount: "{{ inputs.emulator_host != '' ? null : secret('GCP_SERVICE_ACCOUNT') }}"
    fetchType: FETCH
    sql: "{{ render(vars.detect_sql) }}"

  - id: summary
    type: io.kestra.plugin.core.log.Log
    message: >-
      Spanner tables without a row deletion policy on {{ inputs.database_id }}:
      {{ outputs.detect.size }}. {% for r in outputs.detect.rows %}{{
      r.table_name }}({{ r.retention_column }}). {% endfor %}

  - id: route
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.detect.size > 0 }}"
    then:
      - id: alert
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {{ {'text': 'Cloud Spanner ' ~ inputs.database_id ~ ': ' ~ outputs.detect.size ~ ' tables keep timestamped data with no row deletion policy: ' ~ (outputs.detect.rows | jq('map("\(.table_name)(\(.retention_column))") | join("; ")') | first) ~ '.' ~ (inputs.auto_remediate and not inputs.dry_run ? ' Adding the policies 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
            concurrencyLimit: 1
            description: One ALTER TABLE ADD ROW DELETION POLICY at a time, so a single
              failure stays isolated.
            values: "{{ outputs.detect.rows | jq('map(.fix_sql)') | first | toJson }}"
            tasks:
              - id: apply_policy
                type: io.kestra.plugin.gcp.spanner.Execute
                allowFailure: true
                projectId: "{{ inputs.project_id }}"
                instanceId: "{{ inputs.instance_id }}"
                databaseId: "{{ inputs.database_id }}"
                emulatorHost: "{{ inputs.emulator_host }}"
                serviceAccount: "{{ inputs.emulator_host != '' ? null :
                  secret('GCP_SERVICE_ACCOUNT') }}"
                isDdl: true
                sql: "{{ item.value }}"

          - id: reverify
            type: io.kestra.plugin.gcp.spanner.Query
            description: Read INFORMATION_SCHEMA again. A table still here was not covered
              by a valid policy.
            projectId: "{{ inputs.project_id }}"
            instanceId: "{{ inputs.instance_id }}"
            databaseId: "{{ inputs.database_id }}"
            emulatorHost: "{{ inputs.emulator_host }}"
            serviceAccount: "{{ inputs.emulator_host != '' ? null :
              secret('GCP_SERVICE_ACCOUNT') }}"
            fetchType: FETCH
            sql: "{{ render(vars.detect_sql) }}"

          - id: verify_result
            type: io.kestra.plugin.core.flow.If
            description: The run fails if any in-scope table still has no row deletion
              policy after the ALTER.
            condition: "{{ outputs.reverify.size > 0 }}"
            then:
              - id: remediation_failed
                type: io.kestra.plugin.core.execution.Fail
                errorMessage: "Some tables still have no row deletion policy: {{
                  outputs.reverify.rows |
                  jq('map(\"\\(.table_name)(\\(.retention_column))\") | join(\",
                  \")') | first }}."
            else:
              - id: remediated
                type: io.kestra.plugin.core.log.Log
                message: >-
                  Row deletion policy added and verified on: {{
                  outputs.detect.rows |
                  jq('map("\(.table_name)(\(.retention_column))") | join(", ")')
                  | first }}.
    else:
      - id: clean
        type: io.kestra.plugin.core.log.Log
        message: "Every scoped Spanner table has a row deletion policy."

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {{ {'text': 'Cloud Spanner schema governance guard failed on ' ~ inputs.database_id ~ ' (execution ' ~ execution.id ~ '). Check the logs.'} | toJson }}

triggers:
  - id: weekly
    type: io.kestra.plugin.core.trigger.Schedule
    description: Shipped disabled and always in safe mode.
    cron: "0 4 * * 1"
    disabled: true
    inputs:
      auto_remediate: false
      dry_run: true