Centralised Logging at Scale: Architecture, Tools, and Best Practices

// Bad: Unstructured log// Output: User 123 purchased item 456 for £99.99// Output: User 123 purchased item 456 for £99.99// Good: Structured JSON logGood: Structured JSON log
logger.info({
  event: "purchase_completed",
  user_id: "123",
  item_id: "456",
  amount: 99.99,
  currency: "GBP",
  trace_id: "abc123def456",
  span_id: "789xyz",
  timestamp: "2025-12-19T10:30:00Z",
  level: "info",
  service: "checkout-service",
  version: "1.2.3"
});

// Node.js middleware for correlation IDs// Logger that automatically includes correlation context// Logger that automatically includes correlation context// Logger that automatically includes correlation context// Logger that automatically includes correlation context// Logger that automatically includes correlation context// Logger that automatically includes correlation context// Logger that automatically includes correlation context// Logger that automatically includes correlation contextogger that automatically includes correlation context
function log(level, message, data = {}) {
  const context = asyncLocalStorage.getStore() || {};
  console.log(JSON.stringify({
    timestamp: new Date().toISOString(),
    level,
    message,
    correlation_id: context.correlationId,
    trace_id: context.traceId,
    ...data
  }));
}

// PII redaction example
function redactPII(data) {
  const sensitiveFields = ['password', 'credit_card', 'ssn', 'email'];
  const redacted = { ...data };
  
  for (const field of sensitiveFields) {
    if (redacted[field]) {
      redacted[field] = '[REDACTED]';
    }
  }
  
  // Mask email partially
  if (redacted.user_email) {
    const [local, domain] = redacted.user_email.split('@');
    redacted.user_email = `${local.slice(0, 2)}***@${domain}`;
  }
  
  return redacted;
}

# logstash.conf
input {
  beats {
    port => 5044
  }
}

filter {
  # Parse JSON logs
  json {
    source => "message"
    target => "parsed"
  }
  
  # Add geolocation for IP addresses
  geoip {
    source => "[parsed][client_ip]"
    target => "geoip"
  }
  
  # Parse timestamps
  date {
    match => ["[parsed][timestamp]", "ISO8601"]
    target => "@timestamp"
  }
  
  # Remove sensitive fields
  mutate {
    remove_field => ["[parsed][password]", "[parsed][credit_card]"]
  }
}

output {
  elasticsearch {
    hosts => ["elasticsearch:9200"]
    index => "logs-%{[parsed][service]}-%{+YYYY.MM.dd}"
    user => "elastic"
    password => "${ELASTIC_PASSWORD}"
  }
}

# Create ILM policy
PUT _ilm/policy/logs-policy
{
  "policy": {
    "phases": {
      "hot": {
        "min_age": "0ms",
        "actions": {
          "rollover": {
            "max_size": "50gb",
            "max_age": "1d"
          }
        }
      },
      "warm": {
        "min_age": "7d",
        "actions": {
          "shrink": {
            "number_of_shards": 1
          },
          "forcemerge": {
            "max_num_segments": 1
          }
        }
      },
      "cold": {
        "min_age": "30d",
        "actions": {
          "freeze": {}
        }
      },
      "delete": {
        "min_age": "90d",
        "actions": {
          "delete": {}
        }
      }
    }
  }
}

# promtail-config.yaml
server:
  http_listen_port: 9080

positions:
  filename: /tmp/positions.yaml

clients:
  - url: http://loki:3100/loki/api/v1/push

scrape_configs:
  - job_name: kubernetes-pods
    kubernetes_sd_configs:
      - role: pod
    
    relabel_configs:
      # Keep logs from pods with logging enabled
      - source_labels: [__meta_kubernetes_pod_annotation_logging_enabled]
        action: keep
        regex: "true"
      
      # Set namespace label
      - source_labels: [__meta_kubernetes_namespace]
        target_label: namespace
      
      # Set pod name label
      - source_labels: [__meta_kubernetes_pod_name]
        target_label: pod
      
      # Set container name label
      - source_labels: [__meta_kubernetes_container_name]
        target_label: container
    
    pipeline_stages:
      # Parse JSON logs
      - json:
          expressions:
            level: level
            trace_id: trace_id
      
      # Add level as label
      - labels:
          level:

# Basic label filter# Filter by log content# Filter by log content# JSON parsing and filtering# JSON parsing and filtering# JSON parsing and filtering# Regex matching# Regex matching# Regex matching# Regex matching# Aggregation - error count per service# Aggregation - error count per service# Rate of errors# Rate of errors# Rate of errors# Rate of errors# Rate of errors# Rate of errors# Rate of errorsof errors
sum(rate({namespace="production"} |= "error" [1m]))

# Archive logs to S3 with lifecycle policy
resource "aws_s3_bucket_lifecycle_configuration" "logs_archive" {
  bucket = aws_s3_bucket.logs_archive.id

  rule {
    id     = "archive-old-logs"
    status = "Enabled"

    transition {
      days          = 30
      storage_class = "STANDARD_IA"
    }

    transition {
      days          = 90
      storage_class = "GLACIER"
    }

    transition {
      days          = 365
      storage_class = "DEEP_ARCHIVE"
    }

    expiration {
      days = 2555  # 7 years
    }
  }
}

# fluent-bit-daemonset.yaml
apiVersion: apps/v1
kind: DaemonSet
metadata:
  name: fluent-bit
  namespace: logging
  labels:
    app: fluent-bit
spec:
  selector:
    matchLabels:
      app: fluent-bit
  template:
    metadata:
      labels:
        app: fluent-bit
    spec:
      serviceAccountName: fluent-bit
      tolerations:
        - key: node-role.kubernetes.io/master
          effect: NoSchedule
      containers:
        - name: fluent-bit
          image: fluent/fluent-bit:2.2
          resources:
            limits:
              memory: 200Mi
              cpu: 200m
            requests:
              memory: 100Mi
              cpu: 100m
          volumeMounts:
            - name: varlog
              mountPath: /var/log
              readOnly: true
            - name: varlibdockercontainers
              mountPath: /var/lib/docker/containers
              readOnly: true
            - name: config
              mountPath: /fluent-bit/etc/
      volumes:
        - name: varlog
          hostPath:
            path: /var/log
        - name: varlibdockercontainers
          hostPath:
            path: /var/lib/docker/containers
        - name: config
          configMap:
            name: fluent-bit-config

# fluent-bit.conf
[SERVICE]
    Flush         1
    Log_Level     info
    Daemon        off
    Parsers_File  parsers.conf

[INPUT]
    Name              tail
    Path              /var/log/containers/*.log
    Parser            docker
    Tag               kube.*
    Refresh_Interval  5
    Mem_Buf_Limit     50MB
    Skip_Long_Lines   On

[FILTER]
    Name                kubernetes
    Match               kube.*
    Kube_URL            https://kubernetes.default.svc:443
    Kube_CA_File        /var/run/secrets/kubernetes.io/serviceaccount/ca.crt
    Kube_Token_File     /var/run/secrets/kubernetes.io/serviceaccount/token
    Merge_Log           On
    K8S-Logging.Parser  On
    K8S-Logging.Exclude On

[FILTER]
    Name    modify
    Match   kube.*
    Remove  stream
    Remove  logtag

[OUTPUT]
    Name            loki
    Match           kube.*
    Host            loki.logging.svc.cluster.local
    Port            3100
    Labels          job=fluent-bit, cluster=production
    Auto_Kubernetes_Labels on

# Prometheus alerting rules for log-based metrics
groups:
  - name: log-alerts
    rules:
      # Alert on high error rate
      - alert: HighErrorRate
        expr: |
          sum(rate(
            log_messages_total{level="error"}[5m]
          )) by (service) > 10
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "High error rate in {{ $labels.service }}"
          description: "Error rate is {{ $value }} errors/sec"

      # Alert on specific error patterns
      - alert: DatabaseConnectionErrors
        expr: |
          sum(rate(
            log_messages_total{message=~".*connection refused.*database.*"}[5m]
          )) > 0
        for: 2m
        labels:
          severity: critical
        annotations:
          summary: "Database connection errors detected"

      # Alert on authentication failures
      - alert: AuthenticationFailureSpike
        expr: |
          sum(rate(
            log_messages_total{event="auth_failed"}[5m]
          )) > 100
        for: 5m
        labels:
          severity: warning
        annotations:
          summary: "Spike in authentication failures"

# CloudWatch Logs metric filter
resource "aws_cloudwatch_log_metric_filter" "errors" {
  name           = "error-count"
  pattern        = "[timestamp, level=ERROR, ...]"
  log_group_name = aws_cloudwatch_log_group.app.name

  metric_transformation {
    name      = "ErrorCount"
    namespace = "Application"
    value     = "1"
    dimensions = {
      Service = "$.service"
    }
  }
}

resource "aws_cloudwatch_metric_alarm" "high_errors" {
  alarm_name          = "high-error-rate"
  comparison_operator = "GreaterThanThreshold"
  evaluation_periods  = 2
  metric_name         = "ErrorCount"
  namespace           = "Application"
  period              = 300
  statistic           = "Sum"
  threshold           = 50
  alarm_description   = "Error count exceeded threshold"
  alarm_actions       = [aws_sns_topic.alerts.arn]
}