MySQL Table Fragmentation and Optimize Advisor

id: mysql-table-fragmentation-and-optimize-advisor
namespace: company.team
description: |
  Automated MySQL InnoDB storage fragmentation auditor. Inspects DATA_FREE across schema tables,
  calculates reclaimable disk capacity, and prepares off-peak OPTIMIZE TABLE maintenance advisories.

concurrency:
  limit: 1

inputs:
  - id: mysql_url
    type: STRING
    displayName: MySQL JDBC URL
    defaults: "jdbc:mysql://mysql.internal:3306/app_db"
    description: "JDBC connection URL for the target MySQL database server."

  - id: mysql_username
    type: STRING
    displayName: Database Username
    defaults: "db_monitor"
    description: "Database user with SELECT privileges on information_schema.TABLES."

  - id: target_schema
    type: STRING
    displayName: Database Schema
    defaults: "app_db"
    description: "Target database schema name to inspect for table fragmentation."

  - id: fragmentation_ratio_threshold
    type: FLOAT
    displayName: Fragmentation Threshold
    defaults: 0.30
    description: "Minimum ratio of DATA_FREE to total table size to qualify as
      fragmented (0.30 = 30%)."

  - id: min_wasted_mb_threshold
    type: INT
    displayName: Minimum Wasted MB
    defaults: 500
    description: "Minimum wasted free storage in megabytes before recommending an
      OPTIMIZE TABLE command."

  - id: slack_channel
    type: STRING
    displayName: Slack Channel
    defaults: "#dba-maintenance"
    description: "Slack channel destination for DBA storage defragmentation advisories."

triggers:
  - id: weekly_fragmentation_audit
    type: io.kestra.plugin.core.trigger.Schedule
    description: "Executes storage fragmentation audit weekly on Sunday at 04:00 UTC."
    cron: "0 4 * * 0"

tasks:
  - id: query_table_free_space
    type: io.kestra.plugin.jdbc.mysql.Query
    description: "Queries MySQL information_schema.TABLES for InnoDB data and free
      space allocation."
    url: "{{ inputs.mysql_url }}"
    username: "{{ inputs.mysql_username }}"
    password: "{{ secret('MYSQL_PASSWORD') }}"
    fetchType: FETCH
    sql: |
      SELECT
        TABLE_SCHEMA,
        TABLE_NAME,
        ENGINE,
        ROUND(DATA_LENGTH / 1024 / 1024, 2) AS data_size_mb,
        ROUND(INDEX_LENGTH / 1024 / 1024, 2) AS index_size_mb,
        ROUND(DATA_FREE / 1024 / 1024, 2) AS free_space_mb,
        ROUND(DATA_FREE / NULLIF(DATA_LENGTH + INDEX_LENGTH + DATA_FREE, 0), 2) AS fragmentation_ratio
      FROM information_schema.TABLES
      WHERE TABLE_SCHEMA = '{{ inputs.target_schema }}'
        AND ENGINE = 'InnoDB'
        AND DATA_FREE > 0
      ORDER BY DATA_FREE DESC;

  - id: evaluate_table_fragmentation
    type: io.kestra.plugin.scripts.python.Script
    description: "Calculates total reclaimable storage and generates executable
      OPTIMIZE TABLE statements."
    containerImage: python:3.11-slim
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
    dependencies:
      - kestra
    inputFiles:
      tables.json: "{{ outputs.query_table_free_space.rows | default('[]') | toJson }}"
    outputFiles:
      - fragmentation_report.json
    script: |
      import json
      from kestra import Kestra

      with open("tables.json", "r", encoding="utf-8") as f:
          tables = json.load(f)

      ratio_thresh = float("{{ inputs.fragmentation_ratio_threshold }}")
      min_mb_thresh = int("{{ inputs.min_wasted_mb_threshold }}")

      fragmented_tables = []
      total_reclaimable_mb = 0.0

      for t in tables:
          name = t.get("TABLE_NAME", "unknown")
          free_mb = float(t.get("free_space_mb", 0.0))
          data_mb = float(t.get("data_size_mb", 0.0))
          ratio = float(t.get("fragmentation_ratio", 0.0))

          if free_mb >= min_mb_thresh and ratio >= ratio_thresh:
              fragmented_tables.append({
                  "table": name,
                  "free_space_mb": free_mb,
                  "data_size_mb": data_mb,
                  "fragmentation_ratio": ratio,
                  "recommendation": f"OPTIMIZE TABLE `{name}`;"
              })
              total_reclaimable_mb += free_mb

      action_recommended = len(fragmented_tables) > 0

      report = {
          "schema": "{{ inputs.target_schema }}",
          "total_tables_checked": len(tables),
          "fragmented_tables_count": len(fragmented_tables),
          "total_reclaimable_mb": round(total_reclaimable_mb, 2),
          "candidate_tables": fragmented_tables
      }

      with open("fragmentation_report.json", "w", encoding="utf-8") as f:
          json.dump(report, f, indent=2)

      Kestra.outputs({
          "action_recommended": action_recommended,
          "fragmented_count": len(fragmented_tables),
          "total_reclaimable_mb": round(total_reclaimable_mb, 2),
          "top_table": fragmented_tables[0]["table"] if fragmented_tables else "None",
          "top_table_waste_mb": fragmented_tables[0]["free_space_mb"] if fragmented_tables else 0
      })

  - id: evaluate_advisory_trigger
    type: io.kestra.plugin.core.flow.If
    description: "Evaluates whether any candidate tables crossed the fragmentation
      reclaim thresholds."
    condition: "{{ outputs.evaluate_table_fragmentation.vars.action_recommended }}"
    then:
      - id: alert_dba_fragmentation_advisor
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: "Dispatches storage defragmentation advisory card to DBA channel
          via Slack."
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "channel": "{{ inputs.slack_channel }}",
            "text": "*MySQL Storage Optimization and Defragmentation Advisory*\n• *Target Schema*: `{{ inputs.target_schema }}`\n• *Fragmented Tables Count*: {{ outputs.evaluate_table_fragmentation.vars.fragmented_count }}\n• *Total Reclaimable Storage*: ~{{ outputs.evaluate_table_fragmentation.vars.total_reclaimable_mb }} MB\n• *Top Waste Candidate*: `{{ outputs.evaluate_table_fragmentation.vars.top_table }}` ({{ outputs.evaluate_table_fragmentation.vars.top_table_waste_mb }} MB free space)\n• *Operational Impact*: Heavily fragmented InnoDB tables waste cloud disk storage and degrade buffer pool cache hit ratios during range scans.\n• *Recommended Remediation*: Schedule an `OPTIMIZE TABLE {{ outputs.evaluate_table_fragmentation.vars.top_table }};` statement during the upcoming off-peak maintenance window."
          }

    else:
      - id: log_compact_storage_status
        type: io.kestra.plugin.core.log.Log
        description: "Logs confirmation of healthy table storage allocation."
        message: "MySQL storage audit compliant. Schema {{ inputs.target_schema }}
          tables are compact with zero excessive fragmentation."

  - id: export_storage_manifest
    type: io.kestra.plugin.core.debug.Return
    description: "Outputs structured JSON storage audit manifest for maintenance planning."
    format: |
      {
        "pipeline": "mysql-table-fragmentation-and-optimize-advisor",
        "audited_at": "{{ execution.startDate }}",
        "schema": "{{ inputs.target_schema }}",
        "reclaimable_mb": {{ outputs.evaluate_table_fragmentation.vars.total_reclaimable_mb }},
        "status": "{{ outputs.evaluate_table_fragmentation.vars.action_recommended ? 'ADVISORY_PENDING' : 'OPTIMIZED' }}"
      }