MySQL Missing Primary Key Guard

id: mysql-missing-primary-key-guard
namespace: company.team

description: |
  Find MySQL tables without a primary key, and non-InnoDB tables. Without a primary key,
  row-based replication scans the whole table on the replica for every changed row, and Group
  Replication and most managed MySQL services refuse writes. Alert Slack, and optionally promote
  a unique index on NOT NULL columns to the primary key online, or add an invisible
  auto-increment id on tables up to a size limit, then audit again to verify. MyISAM tables are
  reported, never converted.

inputs:
  - id: jdbc_url
    type: STRING
    displayName: JDBC URL
    defaults: "jdbc:mysql://localhost:3306/mysql"

  - id: schemas
    type: ARRAY
    itemType: STRING
    displayName: Schemas to check
    defaults:
      - shop

  - id: add_invisible_id
    type: BOOL
    displayName: Add an invisible id when no unique key fits
    description: Adds the invisible `my_row_id` column that
      sql_generate_invisible_primary_key creates. It is not returned by SELECT
      *.
    defaults: true

  - id: max_rows_for_id
    type: INT
    displayName: Largest table for the invisible id (rows)
    description: Adding an auto-increment column needs LOCK=SHARED, which blocks
      writes for the length of the rebuild. Larger tables are reported, to be
      handled with an online schema change tool.
    defaults: 1000000

  - 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:
  audit_sql: |
    SELECT CONVERT(t.table_schema USING utf8mb4) AS schema_name, CONVERT(t.table_name USING utf8mb4) AS table_name,
           CONVERT(t.engine USING utf8mb4) AS engine, t.table_rows AS approx_rows,
           CONVERT(CASE WHEN t.engine <> 'InnoDB' THEN 'NON_INNODB_TABLE' ELSE 'NO_PRIMARY_KEY' END USING utf8mb4) AS finding,
           CONVERT(CASE WHEN t.engine <> 'InnoDB' THEN 'MEDIUM' ELSE 'HIGH' END USING utf8mb4) AS severity,
           CONVERT(u.index_name USING utf8mb4) AS candidate_key, CONVERT(u.cols USING utf8mb4) AS candidate_columns,
           CONVERT(CASE
             WHEN t.engine <> 'InnoDB' THEN NULL
             WHEN u.index_name IS NOT NULL THEN CONCAT('ALTER TABLE ', QUOTE_ID_SCHEMA, '.', QUOTE_ID_TABLE, ' ADD PRIMARY KEY (', u.quoted_cols, '), DROP INDEX ', CONCAT('`', REPLACE(u.index_name, '`', '``'), '`'), ', ALGORITHM=INPLACE, LOCK=NONE')
             WHEN {{ inputs.add_invisible_id }} AND t.table_rows <= {{ inputs.max_rows_for_id }} THEN CONCAT('ALTER TABLE ', QUOTE_ID_SCHEMA, '.', QUOTE_ID_TABLE, ' ADD COLUMN my_row_id BIGINT UNSIGNED NOT NULL AUTO_INCREMENT INVISIBLE PRIMARY KEY FIRST, ALGORITHM=INPLACE, LOCK=SHARED')
           END USING utf8mb4) AS fix_sql
      FROM (SELECT tt.*, CONCAT('`', REPLACE(tt.table_schema, '`', '``'), '`') AS QUOTE_ID_SCHEMA, CONCAT('`', REPLACE(tt.table_name, '`', '``'), '`') AS QUOTE_ID_TABLE
              FROM information_schema.tables tt
             WHERE tt.table_type = 'BASE TABLE'
               AND tt.table_schema IN ({{ inputs.schemas | jq('map(gsub("\\\\"; "\\\\") | gsub("\u0027"; "\u0027\u0027") | "\u0027" + . + "\u0027") | join(",")') | first }})) t
      LEFT JOIN (
        SELECT s.table_schema, s.table_name, s.index_name,
               GROUP_CONCAT(s.column_name ORDER BY s.seq_in_index) AS cols,
               GROUP_CONCAT(CONCAT('`', REPLACE(s.column_name, '`', '``'), '`') ORDER BY s.seq_in_index) AS quoted_cols,
               ROW_NUMBER() OVER (PARTITION BY s.table_schema, s.table_name ORDER BY COUNT(*), s.index_name) AS rk
          FROM information_schema.statistics s
          LEFT JOIN information_schema.columns c ON c.table_schema = s.table_schema AND c.table_name = s.table_name AND c.column_name = s.column_name
         WHERE s.non_unique = 0 AND s.index_name <> 'PRIMARY'
         GROUP BY s.table_schema, s.table_name, s.index_name
        HAVING SUM(c.is_nullable = 'YES') = 0 AND SUM(s.expression IS NOT NULL) = 0 AND SUM(s.sub_part IS NOT NULL) = 0
      ) u ON u.table_schema = t.table_schema AND u.table_name = t.table_name AND u.rk = 1
     WHERE t.engine <> 'InnoDB'
        OR NOT EXISTS (SELECT 1 FROM information_schema.table_constraints k
                        WHERE k.table_schema = t.table_schema AND k.table_name = t.table_name AND k.constraint_type = 'PRIMARY KEY')
     ORDER BY severity, schema_name, table_name

tasks:
  - id: audit
    type: io.kestra.plugin.jdbc.mysql.Query
    description: Read-only query on information_schema. Identifiers are quoted with
      backticks, embedded backticks doubled.
    url: "{{ inputs.jdbc_url }}"
    username: "{{ secret('MYSQL_USERNAME') }}"
    password: "{{ secret('MYSQL_PASSWORD') }}"
    fetchType: FETCH
    sql: "{{ render(vars.audit_sql) }}"

  - id: summary
    type: io.kestra.plugin.core.log.Log
    message: >-
      MySQL primary key audit on {{ inputs.schemas | join(', ') }}: {{
      outputs.audit.size }} findings. {% for r in outputs.audit.rows %}{{
      r.severity }} {{ r.finding }} {{ r.schema_name }}.{{ r.table_name }} ({{
      r.engine }}, ~{{ r.approx_rows }} rows): {{ (r.candidate_key ?? '') != ''
      ? 'promote unique key ' ~ r.candidate_key ~ ' (' ~ r.candidate_columns ~
      '), online' : ((r.fix_sql ?? '') != '' ? 'add invisible my_row_id, blocks
      writes during the rebuild' : 'report only') }}. {% endfor %}

  - id: route
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.audit.size > 0 }}"
    then:
      - id: alert
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {{ {'text': 'MySQL ' ~ inputs.jdbc_url ~ ': ' ~ (outputs.audit.rows | jq('map("\(.severity) \(.finding) \(.schema_name).\(.table_name)") | join(", ")') | first) ~ '. Tables without a primary key slow row-based replication and are rejected by Group Replication.' ~ (inputs.auto_remediate and not inputs.dry_run ? ' Adding primary keys 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 and
          outputs.audit.rows | jq('map(select(.fix_sql != null)) | length') |
          first > 0 }}"
        then:
          - id: remediate
            type: io.kestra.plugin.core.flow.Loop
            description: One online ALTER TABLE at a time.
            concurrencyLimit: 1
            values: "{{ outputs.audit.rows | jq('map(select(.fix_sql != null) | .fix_sql)')
              | first | toJson }}"
            tasks:
              - id: fix
                type: io.kestra.plugin.jdbc.mysql.Query
                allowFailure: true
                url: "{{ inputs.jdbc_url }}"
                username: "{{ secret('MYSQL_USERNAME') }}"
                password: "{{ secret('MYSQL_PASSWORD') }}"
                fetchType: NONE
                sql: "{{ item.value }}"

          - id: reaudit
            type: io.kestra.plugin.jdbc.mysql.Query
            url: "{{ inputs.jdbc_url }}"
            username: "{{ secret('MYSQL_USERNAME') }}"
            password: "{{ secret('MYSQL_PASSWORD') }}"
            fetchType: FETCH
            sql: "{{ render(vars.audit_sql) }}"

          - id: verify_result
            type: io.kestra.plugin.core.flow.If
            condition: >-
              {{ {'b': outputs.audit.rows, 'a': outputs.reaudit.rows ?? []} |
              jq('([.a[] | .schema_name + "." + .table_name]) as $a
                 | [.b[] | select(.fix_sql != null) | .schema_name + "." + .table_name | select(. as $x | $a | index($x))] | length') | first > 0 }}
            then:
              - id: remediation_failed
                type: io.kestra.plugin.core.execution.Fail
                errorMessage: "Some tables still have no primary key: {{ outputs.reaudit.rows |
                  jq('map(select(.fix_sql != null) |
                  \"\\(.schema_name).\\(.table_name)\") | join(\", \")') | first
                  }}. Check the remediate logs (duplicates, NULLs or missing
                  ALTER privilege)."
            else:
              - id: remediated
                type: io.kestra.plugin.core.log.Log
                message: >-
                  Primary keys added and verified: {{ outputs.audit.rows |
                  jq('map(select(.fix_sql != null) |
                  "\(.schema_name).\(.table_name)") | join(", ")') | first }}.
                  Left for review: {{ outputs.reaudit.rows | jq('if length == 0
                  then "none" else map("\(.finding)
                  \(.schema_name).\(.table_name)") | join(", ") end') | first
                  }}.
    else:
      - id: clean
        type: io.kestra.plugin.core.log.Log
        message: "Every table in {{ inputs.schemas | join(', ') }} is InnoDB with a
          primary key."

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {{ {'text': 'MySQL missing primary key guard failed on ' ~ inputs.jdbc_url ~ ' (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 * * 2"
    disabled: true
    inputs:
      auto_remediate: false
      dry_run: true