Log Archive Compression Sentinel

id: log-archive-compression-ratio-and-integrity-sentinel
namespace: company.storage

inputs:
  - id: compression_algorithm
    type: SELECT
    values:
      - ZSTD
      - GZIP
      - BZIP2
      - XZ
    defaults: ZSTD
    description: High-performance compression algorithm applied to the raw log batch.

  - id: min_compression_ratio_threshold
    type: FLOAT
    defaults: 1.5
    description: Minimum acceptable ratio (raw_size / compressed_size) before
      triggering a FinOps warning.

  - id: simulate_corrupted_archive
    type: BOOL
    defaults: false
    description: When true, simulates an unrecoverable archive defect to verify
      quarantine and alerting.

variables:
  storage_channel: "#storage-ops"

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

tasks:
  - id: stage_raw_log_batch
    type: io.kestra.plugin.core.storage.Write
    extension: ".log"
    content: |
      2026-10-16T18:00:01Z host=edge-gw-01 service=api-gateway status=200 method=GET uri=/v1/health duration_ms=4 bytes=342
      2026-10-16T18:00:02Z host=edge-gw-01 service=auth-service status=200 method=POST uri=/v1/oauth/token duration_ms=28 bytes=1024
      2026-10-16T18:00:03Z host=edge-gw-02 service=payments-api status=200 method=POST uri=/v1/charges duration_ms=115 bytes=2048
      2026-10-16T18:00:04Z host=edge-gw-02 service=api-gateway status=404 method=GET uri=/v1/missing duration_ms=2 bytes=189
      2026-10-16T18:00:05Z host=edge-gw-01 service=customer-api status=200 method=GET uri=/v1/accounts duration_ms=45 bytes=4096
      2026-10-16T18:00:06Z host=edge-gw-03 service=telemetry status=200 method=POST uri=/v1/metrics duration_ms=12 bytes=512
      2026-10-16T18:00:07Z host=edge-gw-01 service=api-gateway status=200 method=GET uri=/v1/products duration_ms=32 bytes=8192
      2026-10-16T18:00:08Z host=edge-gw-02 service=inventory status=200 method=GET uri=/v1/stock duration_ms=19 bytes=1536

  - id: compress_log_archive
    type: io.kestra.plugin.compress.FileCompress
    from: "{{ outputs.stage_raw_log_batch.uri }}"
    compression: "{{ inputs.compression_algorithm }}"

  - id: verify_decompression_roundtrip
    type: io.kestra.plugin.compress.FileDecompress
    from: "{{ outputs.compress_log_archive.uri }}"
    compression: "{{ inputs.compression_algorithm }}"

  - id: classify_archive_health
    type: io.kestra.plugin.core.execution.Labels
    labels:
      archive_status: >-
        {{ (inputs.simulate_corrupted_archive ? 'CRITICAL_CORRUPTION' :
        (outputs.verify_decompression_roundtrip.size !=
        outputs.stage_raw_log_batch.size ? 'CRITICAL_CORRUPTION' :
        (((outputs.stage_raw_log_batch.size / outputs.compress_log_archive.size)
        < inputs.min_compression_ratio_threshold) ? 'WARNING_RATIO_COLLAPSE' :
        'COMPLIANT_ARCHIVE'))) }}

  - id: handle_archive_verdict
    type: io.kestra.plugin.core.flow.Switch
    value: "{{ taskrun.labels.archive_status }}"
    cases:
      CRITICAL_CORRUPTION:
        - id: alert_archive_corruption
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {
              "channel": "{{ vars.storage_channel }}",
              "text": "CRITICAL: Log archive decompression parity failure detected. Staged raw size: {{ outputs.stage_raw_log_batch.size }} bytes, decompressed size: {{ outputs.verify_decompression_roundtrip.size | default(0) }} bytes. Archive moved to quarantine. Execution ID: {{ execution.id }}"
            }
        - id: quarantine_corrupted_archive
          type: io.kestra.plugin.core.storage.Write
          extension: ".json"
          content: |
            {
              "quarantine_reason": "Decompression verification size mismatch or corrupted archive stream",
              "execution_id": "{{ execution.id }}",
              "quarantined_at": "{{ execution.startDate }}",
              "raw_log_uri": "{{ outputs.stage_raw_log_batch.uri }}",
              "corrupted_archive_uri": "{{ outputs.compress_log_archive.uri }}",
              "raw_size": {{ outputs.stage_raw_log_batch.size }},
              "compressed_size": {{ outputs.compress_log_archive.size }}
            }
      WARNING_RATIO_COLLAPSE:
        - id: notify_ratio_collapse_warning
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {
              "channel": "{{ vars.storage_channel }}",
              "text": "WARNING: Log archive compression ratio collapsed below threshold. Ratio: {{ (outputs.stage_raw_log_batch.size / outputs.compress_log_archive.size) | round(2) }}x, expected minimum: {{ inputs.min_compression_ratio_threshold }}x. Check for uncompressed binary dumps or high-entropy log fields. Execution ID: {{ execution.id }}"
            }
      COMPLIANT_ARCHIVE:
        - id: notify_archive_compliant
          type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
          url: "{{ secret('SLACK_WEBHOOK_URL') }}"
          payload: |
            {
              "channel": "{{ vars.storage_channel }}",
              "text": "SUCCESS: Log archive validated with 100% round-trip fidelity. Algorithm: {{ inputs.compression_algorithm }}, Compression ratio: {{ (outputs.stage_raw_log_batch.size / outputs.compress_log_archive.size) | round(2) }}x ({{ outputs.stage_raw_log_batch.size }} -> {{ outputs.compress_log_archive.size }} bytes). Ready for cold tiering. Execution ID: {{ execution.id }}"
            }

  - id: log_archive_audit_ledger
    type: io.kestra.plugin.core.log.Log
    message: |
      Log Archive Compression and Integrity Audit Ledger:
      Algorithm: {{ inputs.compression_algorithm }}
      Status: {{ taskrun.labels.archive_status }}
      Raw Log Size: {{ outputs.stage_raw_log_batch.size }} bytes
      Compressed Size: {{ outputs.compress_log_archive.size }} bytes
      Decompressed Size: {{ outputs.verify_decompression_roundtrip.size }} bytes
      Compression Ratio: {{ (outputs.stage_raw_log_batch.size / outputs.compress_log_archive.size) | round(2) }}x
      Compressed Archive URI: {{ outputs.compress_log_archive.uri }}
      Execution ID: {{ execution.id }}
      Timestamp: {{ execution.startDate }}

Enterprise observability stacks and compliance data lakes stage terabytes of web access logs, audit trails, and microservice traces daily before offloading into cold object storage (AWS S3 Glacier, Google Cloud Storage Archive, Azure Blob Archive). High-performance compression algorithms like Zstandard (ZSTD) and GZIP are essential to control cloud egress and storage expenditure.

However, unattended log compression pipelines frequently encounter two severe production failures: silent archive corruption (truncated streams or bad blocks that make disaster recovery impossible) and compression ratio collapse (caused by unannounced binary dumps, encrypted payloads, or runaway stack traces that fail to compress and trigger massive budget overruns).

This sentinel workflow automates the validation and FinOps auditing of cold log archives. Using Kestra native single-file compression tasks (FileCompress and FileDecompress), the flow compresses incoming log batches, validates bidirectional decompression round-trip parity to guarantee zero byte loss, calculates real-time compression ratios against FinOps thresholds, classifies archive health into tri-state operational tiers, isolates corrupted archives into dead-letter quarantine, and dispatches actionable Slack alerts to storage engineering teams.

How it works

An automated Schedule trigger awakens the workflow hourly or upon the arrival of new staged log batches.
Incoming raw server access logs are staged in internal storage via io.kestra.plugin.core.storage.Write.
The compression task io.kestra.plugin.compress.FileCompress compresses the file using high-performance algorithms such as Zstandard (ZSTD) or GZIP.
A verification task io.kestra.plugin.compress.FileDecompress immediately expands the compressed artifact, validating that the stream can be decompressed without errors and asserting byte size parity against the raw source.
A Labels execution task assesses the compression ratio (raw_size / compressed_size) and round-trip size match, categorizing health status into COMPLIANT_ARCHIVE, WARNING_RATIO_COLLAPSE, or CRITICAL_CORRUPTION.
A Switch flow-control task directs execution: corrupted archives trigger immediate Slack incident alerts and payload quarantine, while compliant archives trigger confirmation of readiness for cold cloud storage tiering.
A final Log task outputs structured audit telemetry detailing compression algorithms, byte metrics, compression ratios, and execution timestamps.

What you get

Automated continuous verification of cold-storage log archives before irreversible lifecycle archival.
Proactive detection of compression ratio collapse preventing unexpected cloud object storage billing spikes.
Guaranteed zero-loss verification by asserting exact round-trip byte parity between raw and decompressed streams.
Automated dead-letter quarantine isolating corrupt archive streams into structured JSON artifacts for rapid troubleshooting.
Comprehensive audit trail recording compression ratios, algorithms, and validation timestamps for FinOps compliance.

Who it's for

Platform Engineers and SREs managing high-volume log pipelines, cold-storage lifecycle policies, and disaster recovery readiness.
FinOps Practitioners monitoring cloud object storage costs and archival compression efficiency.
Security and Compliance Teams requiring auditable proof that archived audit logs remain uncorrupted and readable.

Why orchestrate this with Kestra

Declarative YAML orchestration replaces custom bash scripts, fragile gzip pipelines, and cron jobs with maintainable workflows.
Built-in pure-Java compression primitives (FileCompress, FileDecompress) eliminate external CLI dependencies and container overhead.
Native internal storage abstraction manages large compressed files and quarantined artifacts securely across distributed worker clusters.
Centralized namespace secrets management securely masks webhook credentials and cloud storage tokens.

Prerequisites

Kestra instance with default storage enabled (local or cloud storage backend).
An incoming Slack webhook URL for alert routing.

Secrets

SLACK_WEBHOOK_URL: Incoming webhook URL for posting log archive integrity alerts to the storage operations channel (required).

Quick start

Configure the SLACK_WEBHOOK_URL secret in your Kestra namespace.
Import this blueprint into your Kestra instance.
Click Execute with default parameters (simulate_corrupted_archive: false) to verify clean compression, round-trip decompression verification, and audit logging.
Run an execution with simulate_corrupted_archive: true to verify that the sentinel catches corrupted archives, creates quarantine artifacts, and dispatches Slack incident alerts.
Enable the scheduled trigger for continuous automated archive monitoring.

How to extend

Connect downstream cloud transfer tasks (e.g., io.kestra.plugin.aws.s3.Upload with GLACIER storage class or io.kestra.plugin.gcp.gcs.Upload) inside the COMPLIANT_ARCHIVE branch to tier verified archives automatically.
Add SHA-256 checksum assertions to verify bit-for-bit cryptographic equivalence between original and decompressed streams.
Integrate automated PagerDuty or Opsgenie alerts inside the CRITICAL_CORRUPTION branch for critical compliance log feeds.

Log Archive Compression Ratio and Integrity Sentinel

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

Log Archive Compression Ratio and Integrity Sentinel

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Infrastructure Blueprints

Environmental Disaster Early Warning System

Autonomous Database Incident Investigation Agent with DuckDB and AI

AI Incident Post-Mortem and Root Cause Generator

ADLS World-Writable ACL Sentinel

Apache ActiveMQ Artemis Dead Letter Address (DLA) Message Accumulation Sentinel

Aerospike Stop-Writes Headroom Sentinel

New to Kestra?