How to Implement 2026 Data Retention Policies for Logs with Grafana Loki 3.0 and S3 2026

# s3-retention-bucket.tf
# Terraform >= 1.7 required for Loki 3.0 S3 object lock integration
terraform {
  required_providers {
    aws = {
      source  = "hashicorp/aws"
      version = "~> 5.0"
    }
  }
}

provider "aws" {
  region = "us-east-1" # Loki 3.0 recommends single-region S3 for retention consistency
}

# S3 bucket for Loki 3.0 log storage with 2026 compliance features
resource "aws_s3_bucket" "loki_logs" {
  bucket = "loki-3-0-retention-demo-2026" # Must be globally unique
  force_destroy = false # Prevent accidental deletion of compliance logs

  tags = {
    Purpose = "Loki 3.0 Log Retention 2026"
    Compliance = "GDPR-CCPA-2026"
  }
}

# Enable versioning (required for Loki 3.0 S3 retention point-in-time recovery)
resource "aws_s3_bucket_versioning" "loki_versioning" {
  bucket = aws_s3_bucket.loki_logs.id
  versioning_configuration {
    status = "Enabled"
  }
}

# Enable object lock for 2026 write-once-read-many (WORM) compliance requirements
resource "aws_s3_bucket_object_lock_configuration" "loki_object_lock" {
  bucket = aws_s3_bucket.loki_logs.id

  rule {
    default_retention {
      mode = "COMPLIANCE" # Enforce retention even if bucket owner tries to delete
      days = 2555 # 7 years default retention for 2026 audit requirements
    }
  }
}

# IAM role for Loki 3.0 to access S3 with least privilege
resource "aws_iam_role" "loki_s3_role" {
  name = "loki-3-0-s3-retention-role"

  assume_role_policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Action = "sts:AssumeRoleWithWebIdentity"
        Effect = "Allow"
        Principal = {
          Federated = "arn:aws:iam::123456789012:oidc-provider/loki-3-0-oidc.example.com" # Replace with your OIDC provider
        }
        Condition = {
          StringEquals = {
            "loki-3-0-oidc.example.com:sub" = "system:serviceaccount:loki:loki-sa"
          }
        }
      }
    ]
  })

  tags = {
    Purpose = "Loki 3.0 S3 Access"
  }
}

# IAM policy for Loki S3 access
resource "aws_iam_role_policy" "loki_s3_policy" {
  name = "loki-s3-retention-policy"
  role = aws_iam_role.loki_s3_role.id

  policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Effect = "Allow"
        Action = [
          "s3:PutObject",
          "s3:GetObject",
          "s3:DeleteObject",
          "s3:ListBucket",
          "s3:GetBucketLifecycleConfiguration",
          "s3:PutBucketLifecycleConfiguration"
        ]
        Resource = [
          aws_s3_bucket.loki_logs.arn,
          "${aws_s3_bucket.loki_logs.arn}/*"
        ]
      }
    ]
  })
}

# S3 lifecycle rule to tier cold logs to Glacier after 90 days (cost optimization)
resource "aws_s3_bucket_lifecycle_configuration" "loki_lifecycle" {
  bucket = aws_s3_bucket.loki_logs.id

  rule {
    id     = "tier-to-glacier"
    status = "Enabled"

    transition {
      days          = 90
      storage_class = "GLACIER"
    }

    # Expire logs after 7 years (2555 days) per 2026 retention requirements
    expiration {
      days = 2555
    }
  }
}

// validate-s3-bucket.go
// Go 1.22+ required, validates Loki 3.0 S3 bucket compliance for 2026 retention
package main

import (
    "context"
    "fmt"
    "log"
    "os"
    "time"

    "github.com/aws/aws-sdk-go-v2/aws"
    "github.com/aws/aws-sdk-go-v2/config"
    "github.com/aws/aws-sdk-go-v2/service/s3"
    "github.com/aws/aws-sdk-go-v2/service/s3/types"
)

const (
    requiredBucketName    = "loki-3-0-retention-demo-2026"
    requiredRegion        = "us-east-1"
    minRetentionDays      = 2555 // 7 years per 2026 audit rules
    requiredObjectLockMode = types.ObjectLockRetentionModeCompliance
)

func main() {
    // Load AWS config from default credentials chain (env vars, ~/.aws/credentials, IAM role)
    cfg, err := config.LoadDefaultConfig(context.Background(), config.WithRegion(requiredRegion))
    if err != nil {
        log.Fatalf("failed to load AWS config: %v", err) // Fatal error handling for config failure
    }

    client := s3.NewFromConfig(cfg)

    // Check 1: Bucket exists and is in correct region
    locOutput, err := client.GetBucketLocation(context.Background(), &s3.GetBucketLocationInput{
        Bucket: aws.String(requiredBucketName),
    })
    if err != nil {
        log.Fatalf("bucket %s not found or inaccessible: %v", requiredBucketName, err)
    }
    // GetBucketLocation returns empty string for us-east-1
    bucketRegion := string(locOutput.LocationConstraint)
    if bucketRegion == "" {
        bucketRegion = "us-east-1"
    }
    if bucketRegion != requiredRegion {
        log.Fatalf("bucket %s is in %s, expected %s", requiredBucketName, bucketRegion, requiredRegion)
    }
    fmt.Printf("✅ Bucket %s exists in %s\n", requiredBucketName, requiredRegion)

    // Check 2: Versioning is enabled
    versioningOutput, err := client.GetBucketVersioning(context.Background(), &s3.GetBucketVersioningInput{
        Bucket: aws.String(requiredBucketName),
    })
    if err != nil {
        log.Fatalf("failed to get bucket versioning: %v", err)
    }
    if versioningOutput.Status != types.BucketVersioningStatusEnabled {
        log.Fatalf("bucket versioning is %s, expected ENABLED", versioningOutput.Status)
    }
    fmt.Println("✅ Bucket versioning is enabled")

    // Check 3: Object lock is configured with COMPLIANCE mode
    objectLockOutput, err := client.GetObjectLockConfiguration(context.Background(), &s3.GetObjectLockConfigurationInput{
        Bucket: aws.String(requiredBucketName),
    })
    if err != nil {
        log.Fatalf("failed to get object lock config: %v", err)
    }
    if objectLockOutput.ObjectLockConfiguration == nil {
        log.Fatal("object lock is not configured on bucket")
    }
    defaultRetention := objectLockOutput.ObjectLockConfiguration.Rule.DefaultRetention
    if defaultRetention == nil {
        log.Fatal("no default retention set in object lock rule")
    }
    if defaultRetention.Mode != requiredObjectLockMode {
        log.Fatalf("object lock mode is %s, expected %s", defaultRetention.Mode, requiredObjectLockMode)
    }
    if defaultRetention.Days == nil || *defaultRetention.Days < minRetentionDays {
        log.Fatalf("default retention is %v days, expected at least %d", defaultRetention.Days, minRetentionDays)
    }
    fmt.Printf("✅ Object lock configured with COMPLIANCE mode, %d days retention\n", *defaultRetention.Days)

    // Check 4: Lifecycle rules are set
    lifecycleOutput, err := client.GetBucketLifecycleConfiguration(context.Background(), &s3.GetBucketLifecycleConfigurationInput{
        Bucket: aws.String(requiredBucketName),
    })
    if err != nil {
        log.Fatalf("failed to get lifecycle config: %v", err)
    }
    if len(lifecycleOutput.Rules) == 0 {
        log.Fatal("no lifecycle rules configured")
    }
    fmt.Printf("✅ Found %d lifecycle rules\n", len(lifecycleOutput.Rules))

    // Check 5: IAM role has correct permissions (test PutObject)
    testKey := fmt.Sprintf("loki-validation-%d.txt", time.Now().Unix())
    _, err = client.PutObject(context.Background(), &s3.PutObjectInput{
        Bucket: aws.String(requiredBucketName),
        Key:    aws.String(testKey),
        Body:   nil,
    })
    if err != nil {
        log.Fatalf("failed to put test object (check IAM permissions): %v", err)
    }
    fmt.Printf("✅ Successfully put test object %s\n", testKey)

    // Clean up test object
    _, err = client.DeleteObject(context.Background(), &s3.DeleteObjectInput{
        Bucket: aws.String(requiredBucketName),
        Key:    aws.String(testKey),
    })
    if err != nil {
        log.Printf("warning: failed to delete test object %s: %v", testKey, err)
    } else {
        fmt.Printf("✅ Cleaned up test object %s\n", testKey)
    }

    fmt.Println("\n🎉 All S3 bucket compliance checks passed for 2026 retention")
}

# docker-compose-loki-3-0.yml
# Docker Compose >= 2.20 required to run Loki 3.0 with S3 retention
version: "3.8"

services:
  loki:
    image: grafana/loki:3.0.0 # Loki 3.0.0 stable release with 2026 retention primitives
    container_name: loki-3-0-retention
    ports:
      - "3100:3100" # Loki HTTP API port
    volumes:
      - ./loki-config.yaml:/etc/loki/config.yaml # Mount Loki config
      - ./rules:/etc/loki/rules # Mount retention rules
    environment:
      - AWS_ACCESS_KEY_ID=${AWS_ACCESS_KEY_ID} # From env vars
      - AWS_SECRET_ACCESS_KEY=${AWS_SECRET_ACCESS_KEY}
      - AWS_REGION=us-east-1
    command: -config.file=/etc/loki/config.yaml -target=all
    healthcheck:
      test: ["CMD", "wget", "-q", "--spider", "http://localhost:3100/ready"]
      interval: 10s
      timeout: 5s
      retries: 5
    restart: unless-stopped

  promtail:
    image: grafana/promtail:2.9.0 # Compatible with Loki 3.0
    container_name: promtail-loki-3-0
    volumes:
      - /var/log:/var/log:ro # Read host logs
      - ./promtail-config.yaml:/etc/promtail/config.yaml
    command: -config.file=/etc/promtail/config.yaml
    depends_on:
      - loki
    restart: unless-stopped

  grafana:
    image: grafana/grafana:10.2.0
    container_name: grafana-loki-3-0
    ports:
      - "3000:3000"
    environment:
      - GF_AUTH_ANONYMOUS_ENABLED=true
      - GF_AUTH_ANONYMOUS_ORG_ROLE=Admin
      - GF_INSTALL_PLUGINS=grafana-loki-datasource
    volumes:
      - grafana-storage:/var/lib/grafana
    depends_on:
      - loki
    restart: unless-stopped

volumes:
  grafana-storage:

# test-retention-enforcement.py
# Python 3.11+ required, uses Loki 3.0 HTTP API to test retention policies
import os
import time
import json
import logging
from datetime import datetime, timedelta
import requests
from requests.exceptions import RequestException

# Configure logging for error handling
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)

# Configuration
LOKI_API_URL = os.getenv("LOKI_API_URL", "http://localhost:3100")
S3_BUCKET = os.getenv("S3_BUCKET", "loki-3-0-retention-demo-2026")
TEST_LOG_STREAM = "test_retention_stream"
RETENTION_DAYS = 7 # Test short retention for validation (override for 2026 compliance)
EXPECTED_DELETE_DELAY = 300 # Loki 3.0 retention controller runs every 5 minutes

def push_test_logs():
    """Push test logs to Loki 3.0 with a custom retention label"""
    url = f"{LOKI_API_URL}/loki/api/v1/push"
    now = int(time.time() * 1e9) # Nanosecond timestamp for Loki
    # Log with retention_policy label set to test_7d (matches Loki retention rule)
    payload = {
        "streams": [
            {
                "stream": {
                    "job": "retention_test",
                    "retention_policy": "test_7d",
                    "app": "loki-3-0-retention-demo"
                },
                "values": [
                    [str(now), f"Test log entry for retention validation {datetime.now().isoformat()}"]
                ]
            }
        ]
    }
    try:
        response = requests.post(url, json=payload)
        response.raise_for_status() # Raise HTTPError for bad responses (4xx, 5xx)
        logger.info(f"Successfully pushed test log to Loki: {response.status_code}")
        return True
    except RequestException as e:
        logger.error(f"Failed to push test log to Loki: {e}")
        if response := getattr(e, 'response', None):
            logger.error(f"Loki response: {response.text}")
        return False

def query_test_logs():
    """Query Loki 3.0 for test logs to confirm they exist"""
    url = f"{LOKI_API_URL}/loki/api/v1/query_range"
    params = {
        "query": f'{job="retention_test", retention_policy="test_7d"}',
        "start": int((datetime.now() - timedelta(hours=1)).timestamp()),
        "end": int(datetime.now().timestamp()),
        "limit": 10
    }
    try:
        response = requests.get(url, params=params)
        response.raise_for_status()
        data = response.json()
        if data["status"] == "success":
            result_count = len(data["data"]["result"][0]["values"]) if data["data"]["result"] else 0
            logger.info(f"Found {result_count} test log entries in Loki")
            return result_count
        else:
            logger.error(f"Loki query failed: {data.get('error', 'unknown')}")
            return 0
    except RequestException as e:
        logger.error(f"Failed to query Loki: {e}")
        return 0

def trigger_retention_enforcement():
    """Trigger Loki 3.0 retention controller manually (for testing)"""
    url = f"{LOKI_API_URL}/loki/api/v1/retention/trigger"
    try:
        response = requests.post(url)
        response.raise_for_status()
        logger.info("Triggered Loki retention enforcement manually")
        return True
    except RequestException as e:
        logger.error(f"Failed to trigger retention: {e}")
        return False

def check_s3_deletion():
    """Check if logs are deleted from S3 after retention period (requires AWS CLI)"""
    # Note: In production, use AWS SDK for Go/Python instead of subprocess
    import subprocess
    cmd = [
        "aws", "s3api", "list-objects-v2",
        "--bucket", S3_BUCKET,
        "--prefix", "fake/loki/chunk", # Loki S3 chunk prefix
        "--query", "length(Contents)"
    ]
    try:
        result = subprocess.run(cmd, capture_output=True, text=True, check=True)
        object_count = int(result.stdout.strip()) if result.stdout.strip() else 0
        logger.info(f"S3 bucket {S3_BUCKET} has {object_count} Loki chunks remaining")
        return object_count
    except subprocess.CalledProcessError as e:
        logger.error(f"Failed to list S3 objects: {e.stderr}")
        return -1

def main():
    logger.info("Starting Loki 3.0 retention enforcement test")
    # Step 1: Push test logs
    if not push_test_logs():
        logger.error("Failed to push test logs, exiting")
        return
    # Step 2: Confirm logs exist in Loki
    initial_count = query_test_logs()
    if initial_count == 0:
        logger.error("No test logs found in Loki, exiting")
        return
    # Step 3: Trigger retention (in real world, wait for RETENTION_DAYS, but we use short retention)
    logger.info(f"Waiting {EXPECTED_DELETE_DELAY} seconds for retention controller to run...")
    time.sleep(EXPECTED_DELETE_DELAY)
    trigger_retention_enforcement()
    # Step 4: Check if logs are deleted
    post_retention_count = query_test_logs()
    if post_retention_count < initial_count:
        logger.info("✅ Retention policy enforced: log count decreased")
    else:
        logger.warning("⚠️ Retention policy not yet enforced, may need more time")
    # Step 5: Check S3 for deleted chunks
    s3_count = check_s3_deletion()
    if s3_count == 0:
        logger.info("✅ All Loki chunks deleted from S3 per retention policy")
    else:
        logger.warning(f"⚠️ {s3_count} chunks remaining in S3")

if __name__ == "__main__":
    main()

# retention-rule-7yr.yaml
apiVersion: loki.grafana.com/v1beta1
kind: RetentionRule
metadata:
  name: audit-logs-7yr-retention
  namespace: loki
spec:
  tenantID: "1" # Single-tenant Loki 3.0 setup
  periods:
    - retentionPeriod: 2555d # 7 years per 2026 GDPR requirements
      objectStore: s3
      selector: '{job="audit", retention_policy="7yr"}' # Matches audit log streams
  priority: 10 # Higher priority rules are evaluated first

# Run Loki retention controller in dry-run mode
./loki --target=retention-controller \
  --config.file=/etc/loki/config.yaml \
  --retention.dry-run=true \
  --log.level=info

# Sample dry-run log output
# level=info ts=2026-04-05T12:34:56Z caller=retention.go:123 msg="would delete object" s3_key=fake/loki/chunk/abc123 retention_rule=audit-7yr stream_labels="{job=\"audit\"}"

# S3 lifecycle rule for Glacier Instant Retrieval tiering
resource "aws_s3_bucket_lifecycle_configuration" "loki_glacier_tiering" {
  bucket = aws_s3_bucket.loki_logs.id

  rule {
    id     = "tier-to-glacier-instant"
    status = "Enabled"

    transition {
      days          = 90
      storage_class = "GLACIER_IR" # Glacier Instant Retrieval
    }

    expiration {
      days = 2555 # 7 years retention
    }
  }
}

loki-s3-retention-2026/
├── terraform/ # S3 bucket and IAM configuration
│   ├── s3-retention-bucket.tf
│   ├── variables.tf
│   └── outputs.tf
├── loki/ # Loki 3.0 configuration files
│   ├── loki-config.yaml
│   ├── rules/ # RetentionRule CRD samples
│   │   └── audit-7yr-retention.yaml
│   └── docker-compose.yml
├── go/ # Go validation scripts
│   └── validate-s3-bucket.go
├── python/ # Python test scripts
│   └── test-retention-enforcement.py
├── benchmarks/ # Retention latency and cost benchmarks
│   ├── loki-2.9-vs-3.0.json
│   └── cost-calculator.xlsx
└── README.md # Setup instructions and troubleshooting