Log Retention and Cost

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37

Log Volume: 1TB/day (compressed)
Retention Requirements:
  - Hot (searchable): 7 days
  - Warm (searchable): 14 days
  - Cold (archived): 60 days
  - Glacier: 1 year
 
Cost Breakdown (AWS-based estimates):
 
HOT TIER (7 days = 7TB):
  Storage: 7TB × $0.20/GB = $1,400/month
  Compute: 6 × r5.2xlarge = $2,700/month
  Subtotal: $4,100/month
 
WARM TIER (14 days = 14TB):
  Storage: 14TB × $0.07/GB = $1,000/month
  Compute: 3 × r5.xlarge = $675/month
  Subtotal: $1,675/month
 
COLD/FROZEN (60 days = 60TB):
  Storage: 60TB × $0.025/GB = $1,500/month
  Compute: 2 × r5.xlarge = $450/month
  Subtotal: $1,950/month
 
GLACIER ARCHIVE (365 days = 365TB):
  Storage: 365TB × $0.004/GB = $1,460/month
  No active compute needed
  Subtotal: $1,460/month
 
TOTAL MONTHLY: ~$9,185/month
 
WITHOUT TIERING (all hot, 365 days):
  Storage alone: 365TB × $0.20/GB = $73,000/month
  Compute scaled proportionally: ~$40,000/month
  TOTAL: ~$113,000/month
 
SAVINGS FROM TIERING: ~$103,000/month (92% reduction)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55

# retention-policy.yaml
# Different log types have different retention needs
 
policies:
  # Application debug logs - high volume, low value over time
  application-debug:
    description: "DEBUG and TRACE level logs"
    hot_days: 1
    warm_days: 0
    archive_days: 0
    total_retention_days: 1
    sampling_after_hot: 0.01  # Keep 1% sample
 
  # Application operational logs  
  application-info:
    description: "INFO and WARN level logs"
    hot_days: 7
    warm_days: 14
    archive_days: 69
    total_retention_days: 90
 
  # Application error logs - need longer for investigation
  application-error:
    description: "ERROR and FATAL level logs"
    hot_days: 14
    warm_days: 76
    archive_days: 275
    total_retention_days: 365
 
  # Security audit logs - compliance requires long retention
  security-audit:
    description: "Authentication, authorization, admin actions"
    hot_days: 30
    warm_days: 335
    archive_days: 2555  # 7 years
    total_retention_days: 2920  # 8 years
    immutable: true
    legal_hold_eligible: true
 
  # Infrastructure logs - medium retention
  infrastructure:
    description: "Kubernetes, network, database logs"
    hot_days: 7
    warm_days: 23
    archive_days: 60
    total_retention_days: 90
 
  # Access logs - high volume, summarize instead of retain
  access-logs:
    description: "HTTP request/response logs"
    hot_days: 3
    warm_days: 4
    archive_days: 0
    total_retention_days: 7
    aggregate_after_hot: true  # Convert to metrics

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

Time →                                                                
                                                                       
  Day 0-7          Day 7-30         Day 30-365        Day 365+         
  ────────────     ────────────     ────────────      ────────────     
  ┌──────────┐     ┌──────────┐     ┌──────────┐      ┌──────────┐     
  │   HOT    │ ──▶ │   WARM   │ ──▶ │   COLD   │ ──▶  │  ARCHIVE │     
  │          │     │          │     │          │      │          │     
  │ NVMe SSD │     │ Standard │     │ Frozen   │      │ Object   │     
  │ Full RAM │     │ SSD      │     │ Index    │      │ Store    │     
  │ Replicas │     │ Less RAM │     │ No RAM   │      │ Snapshots│     
  └──────────┘     └──────────┘     └──────────┘      └──────────┘     
                                                                       
  Query: <100ms    Query: <1s       Query: 10-60s     Query: Minutes+  
  Cost: $$$$$      Cost: $$$        Cost: $$          Cost: $          
                                                                       
  Usage: Active    Usage: Recent    Usage: Rare       Usage: Compliance
  debugging        investigation    investigation     only             

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

{
  "policy": {
    "phases": {
      "hot": {
        "min_age": "0ms",
        "actions": {
          "rollover": {
            "max_primary_shard_size": "50gb",
            "max_age": "1d"
          },
          "set_priority": { "priority": 100 }
        }
      },
      "warm": {
        "min_age": "7d",
        "actions": {
          "shrink": { "number_of_shards": 1 },
          "forcemerge": { "max_num_segments": 1 },
          "allocate": {
            "number_of_replicas": 0,
            "require": { "data": "warm" }
          },
          "set_priority": { "priority": 50 }
        }
      },
      "cold": {
        "min_age": "30d",
        "actions": {
          "allocate": {
            "require": { "data": "cold" }
          },
          "freeze": {}
        }
      },
      "frozen": {
        "min_age": "90d",
        "actions": {
          "searchable_snapshot": {
            "snapshot_repository": "logs-archive-repo"
          }
        }
      },
      "delete": {
        "min_age": "365d",
        "actions": {
          "delete": {}
        }
      }
    }
  }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64

import structlog
import time
import random
from collections import defaultdict
 
class SampledLogger:
    """Logger that samples high-frequency events."""
    
    def __init__(self, base_logger, sample_rate=0.01):
        self.logger = base_logger
        self.sample_rate = sample_rate
        
    def info(self, event, **kwargs):
        # Always log if sampled or if marked critical
        if kwargs.pop('always_log', False) or random.random() < self.sample_rate:
            self.logger.info(event, sampled=True, sample_rate=self.sample_rate, **kwargs)
 
 
class RateLimitedLogger:
    """Logger that rate-limits identical messages."""
    
    def __init__(self, base_logger, max_per_minute=10):
        self.logger = base_logger
        self.max_per_minute = max_per_minute
        self.event_counts = defaultdict(lambda: {'count': 0, 'last_reset': time.time()})
    
    def error(self, event, **kwargs):
        # Create key from event and critical identifying fields
        key = f"{event}:{kwargs.get('error_type', '')}:{kwargs.get('service', '')}"
        
        now = time.time()
        entry = self.event_counts[key]
        
        # Reset counter if minute passed
        if now - entry['last_reset'] > 60:
            entry['count'] = 0
            entry['last_reset'] = now
        
        entry['count'] += 1
        
        # Log if under limit, or on round numbers for suppressed events
        if entry['count'] <= self.max_per_minute:
            self.logger.error(event, **kwargs)
        elif entry['count'] in [100, 1000, 10000]:
            self.logger.error(
                event, 
                suppressed_count=entry['count'],
                note="Rate limited - showing count milestone",
                **kwargs
            )
 
 
# Example usage
base_logger = structlog.get_logger()
access_logger = SampledLogger(base_logger, sample_rate=0.01)  # 1% sample
error_logger = RateLimitedLogger(base_logger, max_per_minute=10)
 
# High-frequency access logging
access_logger.info("http_request", path="/api/users", status=200)
 
# Rate-limited error logging
for i in range(1000):
    error_logger.error("database_connection_failed", error_type="TimeoutError")
# Only logs 10 times + milestones at 100, 1000

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58

# Compliance-driven retention configuration
 
data_classification:
  security_audit:
    description: "Authentication, authorization, admin actions"
    contains_pii: true
    regulatory_frameworks: [SOX, PCI-DSS, HIPAA]
    minimum_retention_days: 2555  # 7 years (SOX requirement)
    
    storage:
      hot_days: 30
      warm_days: 335
      archive_days: 2190
      archive_tier: GLACIER
      
    immutability:
      enabled: true
      method: WORM_STORAGE
      
    deletion:
      requires_approval: true
      audit_trail: true
      
  access_logs_with_pii:
    description: "HTTP access logs containing user IDs"
    contains_pii: true
    regulatory_frameworks: [GDPR, CCPA]
    maximum_retention_days: 365  # GDPR data minimization
    right_to_deletion: true
    
    anonymization:
      enabled: true
      fields: [user_id, client_ip, user_agent]
      anonymize_after_days: 30
    
    storage:
      hot_days: 7
      warm_days: 23
      archive_days: 0  # Anonymize or delete after warm
      
  operational_logs:
    description: "System logs without PII"
    contains_pii: false
    regulatory_frameworks: []
    
    storage:
      hot_days: 7
      warm_days: 23
      archive_days: 60
      total_retention_days: 90
 
legal_hold:
  enabled: true
  trigger: MANUAL  # Legal team initiates
  scope: BY_TAG  # Hold specific indices by tag
  notification:
    - legal@company.com
    - sre@company.com

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

# Vector (log collector) exposes metrics for cost attribution
 
groups:
- name: logging_cost
  rules:
  # Bytes ingested by service
  - record: logging:bytes_ingested_by_service:rate5m
    expr: sum by (service) (rate(vector_component_sent_bytes_total{component_type="sink"}[5m]))
 
  # Estimate monthly cost by service ($0.10 per GB ingested, simplified)
  - record: logging:estimated_monthly_cost_by_service
    expr: |
      (logging:bytes_ingested_by_service:rate5m * 2592000)  # bytes/sec × seconds/month
      / 1073741824  # to GB
      * 0.10  # cost per GB
 
  # Top cost-generating services
  - alert: LoggingCostAnomaly
    expr: |
      logging:estimated_monthly_cost_by_service 
      > 1.5 * avg_over_time(logging:estimated_monthly_cost_by_service[7d])
    for: 1h
    labels:
      severity: warning
    annotations:
      summary: "{{ $labels.service }} logging costs increased 50%+"
      description: "Service {{ $labels.service }} is generating significantly more logs than usual."
 
  # Detect debug logging enabled
  - alert: DebugLoggingInProduction
    expr: |
      sum by (service) (rate(log_messages_total{level="debug",environment="production"}[5m])) 
      > 0
    for: 30m
    labels:
      severity: info
    annotations:
      summary: "DEBUG logging enabled for {{ $labels.service }} in production"
      description: "Consider disabling DEBUG logging to reduce costs."

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86

#!/usr/bin/env python3
"""
Log retention enforcement script.
Run daily via cron or Kubernetes CronJob.
"""
 
import boto3
from elasticsearch import Elasticsearch
from datetime import datetime, timedelta
import logging
 
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
 
# Configuration
ES_HOST = "elasticsearch.internal:9200"
S3_BUCKET = "logs-archive"
 
RETENTION_POLICIES = {
    "logs-debug-*": {"max_age_days": 1, "delete": True},
    "logs-info-*": {"max_age_days": 30, "archive_after_days": 7, "delete_after_archive": True},
    "logs-error-*": {"max_age_days": 365, "archive_after_days": 30},
    "logs-audit-*": {"max_age_days": 2920, "archive_after_days": 90, "immutable": True},
}
 
def enforce_retention():
    es = Elasticsearch([ES_HOST])
    s3 = boto3.client('s3')
    
    for pattern, policy in RETENTION_POLICIES.items():
        indices = es.cat.indices(index=pattern, format='json')
        
        for idx in indices:
            idx_name = idx['index']
            idx_date = extract_date(idx_name)  # e.g., logs-info-2024.01.15 → datetime
            age_days = (datetime.now() - idx_date).days
            
            # Check immutability / legal hold
            if policy.get('immutable') or is_under_legal_hold(idx_name):
                logger.info(f"Skipping {idx_name} (immutable/legal hold)")
                continue
            
            # Archive if past threshold
            if policy.get('archive_after_days') and age_days >= policy['archive_after_days']:
                if not is_archived(idx_name, s3):
                    logger.info(f"Archiving {idx_name}")
                    archive_index(es, idx_name, s3)
            
            # Delete if past max age
            if age_days >= policy['max_age_days']:
                if policy.get('delete_after_archive') and not is_archived(idx_name, s3):
                    logger.warning(f"Cannot delete {idx_name}: not archived")
                    continue
                    
                logger.info(f"Deleting {idx_name} (age: {age_days} days)")
                delete_index(es, idx_name)
 
def archive_index(es, idx_name, s3):
    """Snapshot index to S3 before deletion."""
    snapshot_name = f"{idx_name}-{datetime.now().strftime('%Y%m%d')}"
    es.snapshot.create(
        repository="logs-archive-repo",
        snapshot=snapshot_name,
        body={"indices": idx_name},
        wait_for_completion=True
    )
    logger.info(f"Created snapshot {snapshot_name}")
 
def delete_index(es, idx_name):
    """Delete index with safety checks."""
    # Double-check age before deletion
    es.indices.delete(index=idx_name)
    logger.info(f"Deleted index {idx_name}")
 
def is_under_legal_hold(idx_name):
    """Check if index is under legal hold."""
    # Implementation: Check metadata or external hold registry
    return False
 
def is_archived(idx_name, s3):
    """Check if index snapshot exists."""
    # Implementation: Check snapshot repository
    return False
 
if __name__ == "__main__":
    enforce_retention()