SNS S3 Archiver

A Lambda function that archives all messages from the Hummingbird Events SNS topic to S3. Events are stored as clean JSON with decoded payloads, enabling easy browsing with zcat | jq and replay to handlers.

Features

• S3 Storage: Events stored with time-hierarchical key structure for efficient time-range queries
• Clean JSON: Decodes gzip+base64 Message payload for human-readable storage
• Replay Support: Preserves all MessageAttributes with updated content_encoding for replay fidelity
• Easy Access: Download via aws s3 sync, browse with zcat | jq

Prerequisites

• AWS CLI configured with appropriate credentials (IAM permissions for Lambda, S3, SNS, CloudFormation)
• Podman or Docker (for containerized SAM build/deploy)
• Python 3.11 or later (for development)

Deployment

Build and deploy using containerized AWS SAM CLI:

cd sns-s3-archiver
make build     # Build SAM application
make deploy    # First deployment (interactive/guided)
make redeploy  # Subsequent deployments (non-interactive)

Deployment outputs:

• BucketName - S3 bucket name for archived events
• BucketArn - S3 bucket ARN
• FunctionArn - Lambda function ARN

Parameters

Resource naming: Bucket and Lambda names are derived from ResourcePrefix:

• S3 bucket: {ResourcePrefix}-s3-events
• Lambda function: {ResourcePrefix}-lambda-s3-archiver

Prerequisites: Requires an existing SNS topic. Deploy hummingbird-events-topic first to create the topic, then use its ARN for the SnsTopicArn parameter.

S3 Key Structure

Events are stored with a time-hierarchical key structure optimized for time-range queries:

sns/YYYY/MM/DD/HH/MM/TIMESTAMP#MSGID::source::kind::name.json.gz

Examples:

sns/2025/12/13/12/34/2025-12-13T12:34:56.789Z#abc12345::kubernetes::Snapshot::konflux-rh03--tenant--my-snapshot.json.gz
sns/2025/12/13/12/34/2025-12-13T12:34:56.789Z#def67890::gitlab::push::abc123def456.json.gz
sns/2025/12/13/12/34/2025-12-13T12:34:56.789Z#0123abcd::gitlab::pipeline::org-project-42.json.gz
sns/2025/12/13/12/34/2025-12-13T12:34:56.789Z#4567efab::gitlab::build::org-project-99.json.gz
sns/2025/12/13/12/34/2025-12-13T12:34:56.789Z#89abcdef::kubernetes::TaskRun::konflux-rh03--tenant--task-1.json.gz

The sns/ prefix separates the new format from legacy data, enabling incremental migration. The :: delimiter separates metadata fields (avoiding conflicts with K8s names that contain --). The resource name in the key enables efficient “latest state per resource” queries without downloading all files.

The minute-level subdirectory (/MM/) provides write distribution for high-throughput scenarios.

Resource identities

The timestamp and first eight characters of the SNS MessageId make each key deterministic for one SNS delivery. The archiver never uses an SQS message ID. Repeated archiver delivery of the same SNS record writes the same S3 object key.

The logical cluster name comes from the Kubernetes event forwarder’s selected kubeconfig context, not from the API server URL. URL-valued cluster attributes from an older forwarder retain the name-only identity used by existing archives.

When a Pipeline or Job Hook omits its numeric ID, the archiver retains the project-only GitLab identity.

sns-s3-archiver/tests/fixtures/archive-key-contract.json is the versioned, machine-readable contract for these identities. Archiver and GitLab forwarder tests load it directly. Hummingbird Status tests will load the same repository file as part of HUM-7529. It is test data, not a runtime dependency.

Storage Format

Each S3 object contains a gzip-compressed JSON record:

{
  "MessageId": "abc12345-1234-5678-9abc-def012345678",
  "Timestamp": "2025-12-13T12:34:56.789Z",
  "Message": {
    "apiVersion": "appstudio.redhat.com/v1alpha1",
    "kind": "Snapshot",
    "metadata": { ... },
    "spec": { ... }
  },
  "MessageAttributes": {
    "source": { "Type": "String", "Value": "kubernetes" },
    "kind": { "Type": "String", "Value": "Snapshot" },
    "content_encoding": { "Type": "String", "Value": "json" }
  }
}

Key points:

• Message is a decoded JSON object (not the original base64+gzip string from SNS)
• content_encoding is set to "json" to indicate the decoded format

Usage

Download Events Locally

# Sync entire bucket
aws s3 sync s3://bucket-name/ ./local-events/

# Sync specific time range
aws s3 sync s3://bucket-name/2025/12/13/ ./local-events/2025/12/13/

# Sync with filtering
aws s3 sync s3://bucket-name/ ./local-events/ --exclude "*" --include "*kubernetes*Snapshot*"

Browse Events

# List events for a specific hour
aws s3 ls s3://bucket-name/2025/12/13/12/ --recursive

# View a single event
aws s3 cp s3://bucket-name/2025/12/13/12/34/event.json.gz - | zcat | jq

# Local browsing
zcat ./local-events/2025/12/13/12/34/event.json.gz | jq

Replay Events

Consumer handlers should support both live SNS format and archived format:

def _decode_message(sns_record: dict) -> dict:
    """Decode Message - handles live SNS and archived formats."""
    message = sns_record.get("Message", "")
    encoding = sns_record.get("MessageAttributes", {}).get("content_encoding", {}).get("Value")

    # Archived format: Message is already decoded
    if encoding == "json" or isinstance(message, dict):
        return message if isinstance(message, dict) else json.loads(message)

    # Live SNS format: Message is gzip+base64
    if encoding == "gzip+base64":
        return json.loads(gzip.decompress(base64.b64decode(message)))

    # Fallback: plain JSON string
    return json.loads(message)

Compression Flow

Forwarder → SNS → Archiver → S3

1. Forwarder: gzip(payload) → base64 → SNS Message
2. SNS: Passes through with content_encoding="gzip+base64"
3. Archiver: Decodes Message, sets content_encoding="json", gzip(record) → S3
4. S3: Stores clean JSON record, gzip-compressed at object level

Development

See the main README for development workflows.

make setup     # Install dependencies
make check     # Lint code (ruff)
make fmt       # Format code
make test      # Run unit tests
make coverage  # Run tests with coverage

S3 Lifecycle Policies

The bucket has two lifecycle rules:

• AbortIncompleteUploads: Automatically cleans up incomplete multipart uploads after 1 day (prevents storage cost from failed uploads)
• ExpireOldEvents: When RetentionDays > 0, automatically deletes objects older than the specified retention period. Disabled when RetentionDays = 0 (indefinite retention, the default).

Security & Limitations

Security:

• S3 bucket blocks all public access by default
• S3 objects encrypted at rest (AES256)
• Lambda follows least privilege principle (PutObject only)
• CloudWatch logs capture all archive operations (7-day retention)
• Sentry integration for error tracking

Limitations:

• Lambda timeout: 30 seconds
• Lambda memory: 256 MB
• S3 object key length: max 1024 bytes
• Single-region deployment (follows SNS topic region)

License

This project is licensed under the GNU General Public License v3.0 or later - see the LICENSE file for details.