Rebind invalid Db2 packages after a migration, with approval

id: db2-invalid-package-rebind-guard
namespace: company.team

description: |
  Find Db2 packages the catalog marks invalid or inoperative and rebind the ones you allow. A package
  is the compiled access plan for static SQL. DDL on anything the plan touches, dropping an index,
  rebuilding a table, revoking a privilege, clears the VALID flag in SYSCAT.PACKAGES. Db2 then
  recompiles on first use, which puts the cost on whichever request arrives first. An inoperative
  package it refuses to run at all. This flow reads SYSCAT.PACKAGES, joins each bad package to the
  routine that owns it, calls SYSPROC.REBIND_ROUTINE_PACKAGE behind an approval, then reads the
  catalog again and fails the run if a target is still not valid. Read-only by default.

labels:
  system: db2
  area: static-sql-packages

inputs:
  - id: jdbc_url
    type: STRING
    displayName: JDBC URL
    description: Db2 JDBC URL for the database whose catalog is read.
    defaults: "jdbc:db2://localhost:50000/testdb"

  - id: jdbc_username
    type: STRING
    displayName: Database user
    description: >-
      Needs SELECT on SYSCAT.PACKAGES, SYSCAT.PACKAGEDEP, SYSCAT.ROUTINEDEP,
      SYSCAT.ROUTINES to detect. Rebinding also needs ALTERIN on the package
      schema or BINDADD plus EXECUTE on SYSPROC.REBIND_ROUTINE_PACKAGE. A
      detect-only account is the safe choice.
    defaults: "db2inst1"

  - id: jdbc_password
    type: STRING
    displayName: Database password
    description: Defaults to the DB2_PASSWORD secret. Override per run only for
      local testing.
    defaults: "{{ secret('DB2_PASSWORD') }}"

  - id: package_schema
    type: STRING
    displayName: Package schema
    description: >-
      PKGSCHEMA to scan. Folded to upper case before the query runs. One schema
      per run, so a migration that touched two schemas gets two runs with their
      own allowlists.
    defaults: "DB2INST1"

  - id: routine_allowlist
    type: STRING
    displayName: Routine allowlist
    description: >-
      Comma separated SQL LIKE patterns matched against the owning routine, as
      SCHEMA.NAME or NAME. A package is only rebound when a pattern matches.
      Empty means nothing is eligible, which is the default, so a first run
      always reports without touching anything. Example "PAYMENTS.%,ORDERS_%".
    defaults: ""

  - id: max_rebinds
    type: INT
    displayName: Max rebinds per run
    description: >-
      Ceiling on how many packages one run rebinds. Eligible packages past the
      ceiling are reported as deferred so the next run picks them up. A rebind
      recompiles the plan, so a bounded batch keeps the blast radius of a bad
      plan small.
    defaults: 5

  - id: rebind_options
    type: SELECT
    displayName: Rebind options
    description: >-
      Options string passed to SYSPROC.REBIND_ROUTINE_PACKAGE. APREUSE YES asks
      Db2 to reuse the existing plan where it still can, which is the safe
      default because it limits plan drift. RESOLVE CONSERVATIVE resolves names
      only against objects that existed at the original bind.
    values:
      - "APREUSE YES RESOLVE ANY"
      - "RESOLVE ANY"
      - "APREUSE YES RESOLVE CONSERVATIVE"
      - "RESOLVE CONSERVATIVE"
    defaults: "APREUSE YES RESOLVE ANY"

  - id: auto_remediate
    type: BOOL
    displayName: Auto remediate
    description: Must be true, with dry_run false, before any rebind is proposed.
    defaults: false

  - id: dry_run
    type: BOOL
    displayName: Dry run
    description: True reports the batch it would rebind, names every package,
      changes nothing.
    defaults: true

variables:
  detect_sql: |
    WITH PATS (P, REST, D) AS (
      VALUES (CAST(NULL AS VARCHAR(256)), CAST(:routine_allowlist AS VARCHAR(4000)) || ',', 0)
      UNION ALL
      SELECT CAST(SUBSTR(REST, 1, POSSTR(REST, ',') - 1) AS VARCHAR(256)),
             SUBSTR(REST, POSSTR(REST, ',') + 1), D + 1
      FROM PATS WHERE POSSTR(REST, ',') > 0 AND D < 64
    ),
    PATLIST AS (
      SELECT DISTINCT UPPER(TRIM(P)) AS PAT FROM PATS WHERE P IS NOT NULL AND TRIM(P) <> ''
    ),
    BAD AS (
      SELECT PKGSCHEMA, PKGNAME, VALID, LAST_BIND_TIME
      FROM SYSCAT.PACKAGES
      WHERE PKGSCHEMA = UPPER(CAST(:package_schema AS VARCHAR(128))) AND VALID <> 'Y'
    ),
    OWNED AS (
      SELECT D.BSCHEMA AS PKGSCHEMA, D.BNAME AS PKGNAME, R.ROUTINESCHEMA, R.ROUTINENAME,
             R.SPECIFICNAME, R.ROUTINETYPE
      FROM SYSCAT.ROUTINEDEP D
      JOIN SYSCAT.ROUTINES R
        ON R.ROUTINESCHEMA = D.ROUTINESCHEMA AND R.SPECIFICNAME = D.SPECIFICNAME
      WHERE D.BTYPE = 'K'
    ),
    J AS (
      SELECT RTRIM(B.PKGSCHEMA) AS PKGSCHEMA, RTRIM(B.PKGNAME) AS PKGNAME,
             '''' || RTRIM(B.PKGNAME) || '''' AS QUOTED_NAME, B.VALID AS VALID_FLAG,
             CASE B.VALID WHEN 'N' THEN 'invalid' WHEN 'X' THEN 'inoperative' ELSE 'other' END AS STATE,
             CAST(B.LAST_BIND_TIME AS VARCHAR(26)) AS LAST_BIND_TIME,
             COALESCE(RTRIM(O.ROUTINESCHEMA), '') AS ROUTINESCHEMA,
             COALESCE(RTRIM(O.ROUTINENAME), '') AS ROUTINENAME,
             COALESCE(RTRIM(O.ROUTINESCHEMA) || '.' || RTRIM(O.SPECIFICNAME), '') AS SPECIFIC_REF,
             CASE O.ROUTINETYPE WHEN 'P' THEN 'SP' WHEN 'F' THEN 'SF' ELSE '' END AS REBIND_TYPE,
             COALESCE((SELECT LISTAGG(RTRIM(DP.BTYPE) || ':' || RTRIM(DP.BNAME), ' ')
                       FROM SYSCAT.PACKAGEDEP DP
                       WHERE DP.PKGSCHEMA = B.PKGSCHEMA AND DP.PKGNAME = B.PKGNAME), '') AS DEPENDS_ON,
             CASE WHEN O.ROUTINENAME IS NOT NULL
                   AND EXISTS (SELECT 1 FROM PATLIST L
                               WHERE RTRIM(O.ROUTINESCHEMA) || '.' || RTRIM(O.ROUTINENAME) LIKE L.PAT
                                  OR RTRIM(O.ROUTINENAME) LIKE L.PAT)
                  THEN 'Y' ELSE 'N' END AS ALLOWLISTED
      FROM BAD B
      LEFT JOIN OWNED O ON O.PKGSCHEMA = B.PKGSCHEMA AND O.PKGNAME = B.PKGNAME
    ),
    E AS (
      SELECT J.*, CASE WHEN ALLOWLISTED = 'Y' AND REBIND_TYPE <> '' THEN 'Y' ELSE 'N' END AS ELIGIBLE
      FROM J
    )
    SELECT E.*,
           CASE WHEN ELIGIBLE = 'Y'
                 AND ROW_NUMBER() OVER (PARTITION BY ELIGIBLE ORDER BY PKGNAME)
                     <= CAST(:max_rebinds AS INTEGER)
                THEN 'Y' ELSE 'N' END AS IN_BATCH
    FROM E
    ORDER BY PKGNAME

  verify_sql: |
    SELECT RTRIM(PKGNAME) AS PKGNAME, VALID AS VALID_FLAG,
           CAST(LAST_BIND_TIME AS VARCHAR(26)) AS LAST_BIND_TIME
    FROM SYSCAT.PACKAGES
    WHERE PKGSCHEMA = UPPER(CAST(:package_schema AS VARCHAR(128)))
      AND PKGNAME IN (__NAMES__)
    ORDER BY PKGNAME

tasks:
  - id: detect
    type: io.kestra.plugin.jdbc.db2.Query
    description: >-
      Read-only. Every package in the schema whose VALID flag is not Y, with the
      routine that owns it, the objects it still depends on, whether the
      allowlist covers it, whether it fits this run's ceiling.
    url: "{{ inputs.jdbc_url }}"
    username: "{{ inputs.jdbc_username }}"
    password: "{{ inputs.jdbc_password }}"
    fetchType: FETCH
    sql: "{{ render(vars.detect_sql) }}"
    parameters:
      routine_allowlist: "{{ inputs.routine_allowlist }}"
      package_schema: "{{ inputs.package_schema }}"
      max_rebinds: "{{ inputs.max_rebinds }}"

  - id: plan
    type: io.kestra.plugin.core.output.OutputValues
    description: >-
      Split the findings into what this run will rebind, what the allowlist held
      back, what has no SQL rebind route. Flags are strings so the gates never
      depend on numeric rendering.
    values:
      batch: >-
        {{ outputs.detect.rows | jq('[ .[] | select(.IN_BATCH == "Y") | {t:
        .REBIND_TYPE, n: .SPECIFIC_REF, p: .PKGNAME, r: .ROUTINENAME} | tostring
        ]') | first | toJson }}
      in_list: >-
        {{ outputs.detect.rows | jq('[ .[] | select(.IN_BATCH == "Y") |
        .QUOTED_NAME ] | join(",")') | first }}
      batch_names: >-
        {{ outputs.detect.rows | jq('[ .[] | select(.IN_BATCH == "Y") |
        .ROUTINENAME + " (" + .PKGNAME + ")" ] | join(", ")') | first }}
      finding_names: >-
        {{ outputs.detect.rows | jq('[ .[] | (if .ROUTINENAME == "" then
        "application package " + .PKGNAME else .ROUTINENAME + " (" + .PKGNAME +
        ")" end) + " " + .STATE ] | join("; ")') | first }}
      counts: >-
        {{ outputs.detect.rows | jq('"total=" + ([ .[] ] | length | tostring) +
        " invalid=" + ([ .[] | select(.STATE == "invalid") ] | length |
        tostring) + " inoperative=" + ([ .[] | select(.STATE == "inoperative") ]
        | length | tostring) + " eligible=" + ([ .[] | select(.ELIGIBLE == "Y")
        ] | length | tostring) + " batch=" + ([ .[] | select(.IN_BATCH == "Y") ]
        | length | tostring) + " deferred_by_ceiling=" + ([ .[] |
        select(.ELIGIBLE == "Y" and .IN_BATCH == "N") ] | length | tostring) + "
        held_by_allowlist=" + ([ .[] | select(.ALLOWLISTED == "N" and
        .REBIND_TYPE != "") ] | length | tostring) + " no_sql_rebind_route=" +
        ([ .[] | select(.REBIND_TYPE == "") ] | length | tostring)') | first }}
      has_batch: >-
        {{ outputs.detect.rows | jq('if ([ .[] | select(.IN_BATCH == "Y") ] |
        length) > 0 then "yes" else "no" end') | first }}
      has_no_route: >-
        {{ outputs.detect.rows | jq('if ([ .[] | select(.REBIND_TYPE == "") ] |
        length) > 0 then "yes" else "no" end') | first }}

  - id: summary
    type: io.kestra.plugin.core.log.Log
    message: >-
      Db2 package scan on {{ inputs.jdbc_url }} schema {{ inputs.package_schema
      }}: {{ outputs.detect.size }} package(s) not valid. {{
      outputs.plan.values.counts }}. {{ outputs.detect.size > 0 ? 'Findings: ' ~
      outputs.plan.values.finding_names ~ '.' : 'Nothing to do.' }}

  - id: route
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.detect.size > 0 }}"
    then:
      - id: alert
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: One message carrying the counts, the batch, what was held back.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: >-
          {{ {'text': 'Db2 ' ~ inputs.jdbc_url ~ ' schema ' ~
          inputs.package_schema ~ ': ' ~ outputs.detect.size ~ ' package(s) not
          valid after DDL. ' ~ outputs.plan.values.counts ~ '. ' ~
          outputs.plan.values.finding_names ~ '.' ~ (inputs.auto_remediate and
          not inputs.dry_run ? (outputs.plan.values.has_batch == 'yes' ? '
          Waiting for approval to rebind: ' ~ outputs.plan.values.batch_names ~
          '.' : ' Nothing is eligible, so no rebind is proposed.') : ' No change
          applied (auto_remediate=' ~ inputs.auto_remediate ~ ', dry_run=' ~
          inputs.dry_run ~ ').')} | toJson }}

      - id: no_route_note
        type: io.kestra.plugin.core.log.Log
        level: WARN
        message: >-
          {{ outputs.plan.values.has_no_route == 'yes' ? 'Some packages have no
          routine behind them, so they were bound from a bind file by an
          application. SYSPROC.ADMIN_CMD refuses REBIND, so this flow never
          touches them. Run "db2 rebind package <schema>.<name>" from a CLP
          step. Otherwise the application drives the implicit rebind itself.' :
          'Every finding has a routine behind it, so every one of them has a SQL
          rebind route.' }}

      - id: remediate_gate
        type: io.kestra.plugin.core.flow.If
        condition: "{{ inputs.auto_remediate and not inputs.dry_run and
          outputs.plan.values.has_batch == 'yes' }}"
        then:
          - id: approval
            type: io.kestra.plugin.core.flow.Pause
            description: >-
              A rebind recompiles the access plan against today's statistics, so
              a package that was performing on an old plan can come back on a
              different one. Approve with the batch in front of you.
            onResume:
              - id: approved
                type: BOOL
                displayName: Rebind this batch
                description: >-
                  True rebinds {{ outputs.plan.values.batch_names }} with
                  options {{ inputs.rebind_options }}. False leaves the catalog
                  as it is.

          - id: decision
            type: io.kestra.plugin.core.flow.If
            condition: "{{ outputs.approval.onResume.approved ?? false }}"
            then:
              - id: rebind
                type: io.kestra.plugin.core.flow.Loop
                description: >-
                  Sequential remediation (concurrencyLimit 1). One package at a
                  time, because two rebinds against the same catalog rows race.
                concurrencyLimit: 1
                values: "{{ outputs.plan.values.batch }}"
                tasks:
                  - id: rebind_one
                    type: io.kestra.plugin.jdbc.db2.Query
                    description: >-
                      SYSPROC.REBIND_ROUTINE_PACKAGE recompiles the package
                      behind the routine. SYSPROC.ADMIN_CMD rejects the CLP
                      REBIND command, so this is the SQL route.
                    url: "{{ inputs.jdbc_url }}"
                    username: "{{ inputs.jdbc_username }}"
                    password: "{{ inputs.jdbc_password }}"
                    fetchType: NONE
                    sql: "CALL SYSPROC.REBIND_ROUTINE_PACKAGE(:rtype, :rname, :ropts)"
                    parameters:
                      rtype: "{{ item.value | jq('.t') | first }}"
                      rname: "{{ item.value | jq('.n') | first }}"
                      ropts: "{{ inputs.rebind_options }}"

                  - id: rebind_log
                    type: io.kestra.plugin.core.log.Log
                    message: >-
                      Rebound {{ item.value | jq('.r') | first }} package {{
                      item.value | jq('.p') | first }} with options {{
                      inputs.rebind_options }}. The catalog read decides whether
                      it landed.

              - id: verify
                type: io.kestra.plugin.jdbc.db2.Query
                description: >-
                  Read the catalog again for exactly the packages this run
                  targeted. The rebind task going green proves nothing on its
                  own.
                url: "{{ inputs.jdbc_url }}"
                username: "{{ inputs.jdbc_username }}"
                password: "{{ inputs.jdbc_password }}"
                fetchType: FETCH
                sql: "{{ render(vars.verify_sql) | replace({'__NAMES__':
                  outputs.plan.values.in_list}) }}"
                parameters:
                  package_schema: "{{ inputs.package_schema }}"

              - id: verify_log
                type: io.kestra.plugin.core.log.Log
                message: >-
                  Re-read of {{ outputs.verify.size }} targeted package(s): {{
                  outputs.verify.rows | jq('[ .[] | .PKGNAME + " VALID=" +
                  .VALID_FLAG + " bound " + .LAST_BIND_TIME ] | join("; ")') |
                  first }}.

              - id: verify_gate
                type: io.kestra.plugin.core.flow.If
                condition: "{{ (outputs.verify.rows | jq('if ([ .[] | select(.VALID_FLAG !=
                  \"Y\") ] | length) > 0 then \"bad\" else \"ok\" end') | first)
                  == 'bad' }}"
                then:
                  - id: still_invalid
                    type: io.kestra.plugin.core.execution.Fail
                    errorMessage: >-
                      Rebind did not land. {{ outputs.verify.rows | jq('[ .[] |
                      select(.VALID_FLAG != "Y") | .PKGNAME + " VALID=" +
                      .VALID_FLAG ] | join(", ")') | first }} still not valid
                      after SYSPROC.REBIND_ROUTINE_PACKAGE reported success.
                      Check that the account holds ALTERIN on {{
                      inputs.package_schema }} and that every object the plan
                      needs exists.
                else:
                  - id: verified
                    type: io.kestra.plugin.core.log.Log
                    message: >-
                      All {{ outputs.verify.size }} targeted package(s) read
                      back VALID=Y from SYSCAT.PACKAGES.
            else:
              - id: declined
                type: io.kestra.plugin.core.log.Log
                level: WARN
                message: >-
                  Approval declined or left unanswered, so nothing was rebound.
                  {{ outputs.plan.values.batch_names }} stay as the catalog has
                  them.
        else:
          - id: no_change
            type: io.kestra.plugin.core.log.Log
            message: >-
              Reporting only. {{ outputs.plan.values.has_batch == 'yes' ? 'Would
              rebind ' ~ outputs.plan.values.batch_names ~ ' with options ' ~
              inputs.rebind_options ~ '.' : 'Nothing is eligible: widen
              routine_allowlist or raise max_rebinds.' }} auto_remediate={{
              inputs.auto_remediate }} dry_run={{ inputs.dry_run }}.

errors:
  - id: failure_alert
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: >-
      {{ {'text': 'db2-invalid-package-rebind-guard failed on ' ~
      inputs.jdbc_url ~ ' schema ' ~ inputs.package_schema ~ '. Execution ' ~
      execution.id ~ '. Either the catalog read failed or a rebind did not land,
      so treat the packages as still not valid.'} | toJson }}

triggers:
  - id: daily_scan
    type: io.kestra.plugin.core.trigger.Schedule
    disabled: true
    cron: "0 6 * * *"
    inputs:
      dry_run: true
      auto_remediate: false