Cloudflare R2 Lifecycle Compaction and Cost Guard

id: cloudflare-r2-lifecycle-compaction-and-cost-guard
namespace: company.storage

tasks:
  - id: list_transient_partitions
    type: io.kestra.plugin.aws.s3.List
    endpointOverride: "https://{{ secret('R2_ACCOUNT_ID') }}.r2.cloudflarestorage.com"
    accessKeyId: "{{ secret('R2_ACCESS_KEY_ID') }}"
    secretKeyId: "{{ secret('R2_SECRET_ACCESS_KEY') }}"
    region: "auto"
    compatibilityMode: true
    forcePathStyle: true
    bucket: "{{ secret('R2_BUCKET_NAME') }}"
    prefix: "transient-logs/"

  - id: evaluate_compaction
    type: io.kestra.plugin.scripts.python.Script
    containerImage: python:3.12-slim
    script: |
      import sys
      from datetime import datetime, timezone, timedelta
      from kestra import Kestra

      # Direct ingestion of S3 listing output without synthetic fallbacks
      objects = {{ outputs.list_transient_partitions.objects }}

      if objects is None or not isinstance(objects, list):
          raise ValueError(f"Failed to retrieve valid S3 object list from Cloudflare R2 endpoint. Received: {type(objects)}")

      total_count = len(objects)
      total_bytes = sum(obj.get("size", 0) for obj in objects)

      # Objects older than 7 days qualify for compaction
      retention_cutoff = datetime.now(timezone.utc) - timedelta(days=7)
      stale_count = 0
      stale_bytes = 0

      for obj in objects:
          last_modified_str = obj.get("lastModified")
          if last_modified_str:
              try:
                  last_mod = datetime.fromisoformat(str(last_modified_str).replace("Z", "+00:00"))
                  if last_mod < retention_cutoff:
                      stale_count += 1
                      stale_bytes += obj.get("size", 0)
              except Exception:
                  pass

      compaction_needed = stale_count > 0 or total_count > 500

      print(f"Cloudflare R2 Partition Scan: {total_count} total objects ({total_bytes} bytes).")
      print(f"Compaction candidates: {stale_count} stale objects ({stale_bytes} bytes).")
      print(f"Execution required: {compaction_needed}")

      Kestra.outputs({
          "total_objects": total_count,
          "total_gigabytes": round(total_bytes / (1024 ** 3), 3),
          "stale_count": stale_count,
          "stale_gigabytes": round(stale_bytes / (1024 ** 3), 3),
          "compaction_required": compaction_needed
      })

  - id: compaction_execution_gate
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.evaluate_compaction.vars.compaction_required }}"
    then:
      - id: prune_stale_partitions
        type: io.kestra.plugin.aws.s3.DeleteList
        endpointOverride: "https://{{ secret('R2_ACCOUNT_ID') }}.r2.cloudflarestorage.com"
        accessKeyId: "{{ secret('R2_ACCESS_KEY_ID') }}"
        secretKeyId: "{{ secret('R2_SECRET_ACCESS_KEY') }}"
        region: "auto"
        compatibilityMode: true
        forcePathStyle: true
        bucket: "{{ secret('R2_BUCKET_NAME') }}"
        prefix: "transient-logs/"

      - id: notify_slack_finops
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": ":recycle: *Cloudflare R2 Storage Compaction Executed*\nBucket: `{{ secret('R2_BUCKET_NAME') }}`\nPruned Objects: `{{ outputs.evaluate_compaction.vars.stale_count }}`\nReclaimed Space: `{{ outputs.evaluate_compaction.vars.stale_gigabytes }} GB`\nPartition lifecycle compaction completed successfully."
          }

  - id: log_operational_summary
    type: io.kestra.plugin.core.log.Log
    message: "Cloudflare R2 lifecycle audit completed for bucket '{{
      secret('R2_BUCKET_NAME') }}'. Total Objects: {{
      outputs.evaluate_compaction.vars.total_objects }}, Compaction Executed: {{
      outputs.evaluate_compaction.vars.compaction_required }}"

triggers:
  - id: daily_compaction_schedule
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 2 * * *"