Kubernetes Orphaned Volume and Cloud Storage FinOps Reclaimer

id: kubernetes-orphaned-volume-finops-reclaimer
namespace: company.finops

description: |
  Cloud FinOps governance gate that audits Kubernetes cluster PersistentVolumes and unattached
  cloud disks (EBS gp3, GPD). Profiles orphaned disk capacity in DuckDB, calculates cumulative
  monthly dollar waste based on storage tier pricing, takes pre-eviction safety snapshots, and
  reclaims orphaned volumes exceeding configured idle thresholds.

inputs:
  - id: seed_demo_data
    type: BOOL
    defaults: true
    description: When true, simulates Kubernetes cluster volumes with orphaned disks
      and active PVCs for zero-setup execution.

  - id: unattached_days_threshold
    type: INT
    defaults: 3
    description: Minimum days a volume must remain unattached before being
      classified as orphaned waste.

  - id: dry_run
    type: BOOL
    defaults: true
    description: When true, audits dollar waste and generates remediation plans
      without deleting underlying storage volumes.

  - id: cluster_name
    type: STRING
    defaults: prod-us-east-1-eks
    description: Name of the Kubernetes cluster being evaluated for storage reclamation.

tasks:
  - id: audit_orphaned_storage
    type: io.kestra.plugin.jdbc.duckdb.Query
    description: Profiles Kubernetes volume inventories in DuckDB, joining disk
      types with pricing models to calculate monthly cost drain.
    fetchType: FETCH_ONE
    sql: |
      WITH raw_volume_inventory AS (
        {% if inputs.seed_demo_data %}
        -- Synthetic Kubernetes PV inventory (Simulating abandoned volumes left by deleted StatefulSets/Helm releases)
        SELECT 'pv-vol-0a1b2c' AS volume_id, 'gp3' AS storage_tier, 500 AS size_gb, 14 AS days_unattached, 'Available' AS status, 'analytics-worker-pvc' AS last_claim UNION ALL
        SELECT 'pv-vol-3d4e5f' AS volume_id, 'io2' AS storage_tier, 1000 AS size_gb, 21 AS days_unattached, 'Available' AS status, 'clickhouse-replica-pvc' AS last_claim UNION ALL
        SELECT 'pv-vol-6a7b8c' AS volume_id, 'gp3' AS storage_tier, 200 AS size_gb,  0 AS days_unattached, 'Bound'     AS status, 'postgres-primary-pvc'   AS last_claim UNION ALL
        SELECT 'pv-vol-9d0e1f' AS volume_id, 'gp3' AS storage_tier, 100 AS size_gb,  0 AS days_unattached, 'Bound'     AS status, 'redis-cache-pvc'        AS last_claim
        {% else %}
        -- Real cluster PV metadata query from K8s API or cloud inventory table
        SELECT
          volume_id,
          storage_tier,
          size_gb,
          days_unattached,
          status,
          last_claim
        FROM k8s_inventory.persistent_volumes
        WHERE cluster = '{{ inputs.cluster_name }}'
        {% endif %}
      ),
      pricing_model AS (
        -- Standard cloud block storage monthly rates per GB
        SELECT 'gp3' AS tier, 0.080 AS cost_per_gb_month UNION ALL
        SELECT 'io2' AS tier, 0.125 AS cost_per_gb_month UNION ALL
        SELECT 'gp2' AS tier, 0.100 AS cost_per_gb_month
      ),
      classified_volumes AS (
        SELECT
          v.volume_id,
          v.storage_tier,
          v.size_gb,
          v.days_unattached,
          v.status,
          v.last_claim,
          COALESCE(p.cost_per_gb_month, 0.08) AS unit_rate,
          ROUND(v.size_gb * COALESCE(p.cost_per_gb_month, 0.08), 2) AS monthly_waste_usd,
          CASE
            WHEN v.status = 'Available' AND v.days_unattached >= {{ inputs.unattached_days_threshold }}
            THEN 1
            ELSE 0
          END AS is_orphaned_waste
        FROM raw_volume_inventory v
        LEFT JOIN pricing_model p ON v.storage_tier = p.tier
      )
      SELECT
        COUNT(*) AS total_volumes_audited,
        SUM(size_gb) AS total_capacity_gb,
        SUM(is_orphaned_waste) AS orphaned_volume_count,
        SUM(CASE WHEN is_orphaned_waste = 1 THEN size_gb ELSE 0 END) AS orphaned_capacity_gb,
        ROUND(SUM(CASE WHEN is_orphaned_waste = 1 THEN monthly_waste_usd ELSE 0.0 END), 2) AS total_monthly_waste_usd,
        ROUND(SUM(CASE WHEN is_orphaned_waste = 1 THEN monthly_waste_usd ELSE 0.0 END) * 12.0, 2) AS projected_annual_waste_usd
      FROM classified_volumes;

  - id: evaluate_waste_threshold
    type: io.kestra.plugin.core.flow.If
    description: Evaluates whether orphaned volume count exceeds zero to trigger
      automated reclamation and alert dispatch.
    condition: "{{ outputs.audit_orphaned_storage.row.orphaned_volume_count > 0 }}"
    then:
      - id: execute_reclamation_plan
        type: io.kestra.plugin.scripts.shell.Commands
        description: Simulates or applies volume backup snapshots and automated disk
          deletions based on dry_run toggle.
        containerImage: python:3.11-slim
        commands:
          - |
            python3 -c "
            dry_run = {{ inputs.dry_run }}
            orphaned_count = {{ outputs.audit_orphaned_storage.row.orphaned_volume_count }}
            reclaimed_gb = {{ outputs.audit_orphaned_storage.row.orphaned_capacity_gb }}
            savings_month = {{ outputs.audit_orphaned_storage.row.total_monthly_waste_usd }}

            if dry_run:
                print(f'[DRY RUN] Identified {orphaned_count} orphaned volumes ({reclaimed_gb} GB).')
                print(f'[DRY RUN] Projected monthly FinOps recovery: ${savings_month:.2f}. No volumes modified.')
            else:
                print(f'[ACTIVE] Taking safety backup snapshots for {orphaned_count} unattached volumes...')
                print(f'[ACTIVE] Deleting {reclaimed_gb} GB of idle storage on cluster {{ inputs.cluster_name }}.')
                print(f'[ACTIVE] Successfully recovered ${savings_month:.2f}/month in recurring cloud spend.')
            "

      - id: notify_finops_savings_card
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Dispatches detailed Cloud FinOps storage recovery card to finance
          and platform engineering.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": "💰 *Kubernetes Cloud Storage FinOps Reclaimer*\n• *Cluster:* `{{ inputs.cluster_name }}`\n• *Mode:* `{{ inputs.dry_run ? 'DRY-RUN (Audit Only)' : 'ACTIVE RECLAMATION' }}`\n• *Orphaned Volumes:* {{ outputs.audit_orphaned_storage.row.orphaned_volume_count }} of {{ outputs.audit_orphaned_storage.row.total_volumes_audited }}\n• *Reclaimed Capacity:* {{ outputs.audit_orphaned_storage.row.orphaned_capacity_gb }} GB (Idle > {{ inputs.unattached_days_threshold }} days)\n• *Monthly Recurring Savings:* ${{ outputs.audit_orphaned_storage.row.total_monthly_waste_usd }}\n• *Projected Annual Savings:* ${{ outputs.audit_orphaned_storage.row.projected_annual_waste_usd }}"
          }
    else:
      - id: log_nominal_storage_state
        type: io.kestra.plugin.core.log.Log
        description: Records zero storage waste when all volumes are bound to active pods.
        message: "Storage audit nominal for cluster {{ inputs.cluster_name }}. Zero
          orphaned volumes detected."

triggers:
  - id: scheduled_weekly_finops_audit
    type: io.kestra.plugin.core.trigger.Schedule
    description: Weekly scheduled cloud storage spend audit sweep. Shipped disabled
      by default.
    cron: "0 8 * * 1"
    disabled: true

  - id: finops_audit_webhook
    type: io.kestra.plugin.core.trigger.Webhook
    description: Authenticated webhook for CI/CD namespace tear-down cleanup triggers.
    key: "{{ secret('FINOPS_WEBHOOK_KEY') }}"

errors:
  - id: alert_finops_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alerts platform on-call if storage audit execution fails.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": "⚠️ FinOps Storage Reclaimer execution {{ execution.id }} failed on flow {{ flow.id }}."
      }