Log Aggregation (ELK, OpenSearch)

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

┌─────────────────────────────────────────────────────────────────────────┐
│                        APPLICATION LAYER                                  │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐  │
│  │ Service A│  │ Service B│  │ Service C│  │ Service D│  │ Service E│  │
│  │ (stdout) │  │ (stdout) │  │ (stdout) │  │ (stdout) │  │ (stdout) │  │
│  └────┬─────┘  └────┬─────┘  └────┬─────┘  └────┬─────┘  └────┬─────┘  │
└───────┼─────────────┼─────────────┼─────────────┼─────────────┼────────┘
        │             │             │             │             │
        ▼             ▼             ▼             ▼             ▼
┌─────────────────────────────────────────────────────────────────────────┐
│                        COLLECTION LAYER                                   │
│  ┌──────────────────────────────────────────────────────────────────┐  │
│  │  Log Collectors (Filebeat/Fluent Bit per node)                    │  │
│  │  - Tail container stdout/stderr                                   │  │
│  │  - Add metadata (pod, namespace, container ID)                    │  │
│  │  - Buffer locally, forward with backpressure handling             │  │
│  └──────────────────────────────────────────────────────────────────┘  │
└────────────────────────────────┬────────────────────────────────────────┘
                                 │
                                 ▼
┌─────────────────────────────────────────────────────────────────────────┐
│                        BUFFERING LAYER (Optional)                        │
│  ┌──────────────────────────────────────────────────────────────────┐  │
│  │  Kafka / Redis / Message Queue                                    │  │
│  │  - Decouples collection from processing                           │  │
│  │  - Handles traffic spikes                                         │  │
│  │  - Enables replay for reprocessing                                │  │
│  └──────────────────────────────────────────────────────────────────┘  │
└────────────────────────────────┬────────────────────────────────────────┘
                                 │
                                 ▼
┌─────────────────────────────────────────────────────────────────────────┐
│                        PROCESSING LAYER                                   │
│  ┌──────────────────────────────────────────────────────────────────┐  │
│  │  Logstash / Vector / Stream Processor                             │  │
│  │  - Parse and validate JSON                                        │  │
│  │  - Enrich with additional context                                 │  │
│  │  - Route to appropriate indexes                                   │  │
│  │  - Filter sensitive data                                          │  │
│  └──────────────────────────────────────────────────────────────────┘  │
└────────────────────────────────┬────────────────────────────────────────┘
                                 │
                                 ▼
┌─────────────────────────────────────────────────────────────────────────┐
│                        STORAGE LAYER                                      │
│  ┌──────────────────────────────────────────────────────────────────┐  │
│  │  Elasticsearch / OpenSearch Cluster                               │  │
│  │  - Sharded for horizontal scale                                   │  │
│  │  - Replicated for durability                                      │  │
│  │  - Time-based indexes (logs-2024.01.15)                          │  │
│  │  - Hot/warm/cold tiering for cost optimization                    │  │
│  └──────────────────────────────────────────────────────────────────┘  │
└────────────────────────────────┬────────────────────────────────────────┘
                                 │
                                 ▼
┌─────────────────────────────────────────────────────────────────────────┐
│                        VISUALIZATION LAYER                                │
│  ┌──────────────────────────────────────────────────────────────────┐  │
│  │  Kibana / Grafana / OpenSearch Dashboards                         │  │
│  │  - Full-text search                                               │  │
│  │  - Saved queries and dashboards                                   │  │
│  │  - Alert definition                                               │  │
│  │  - Access control                                                 │  │
│  └──────────────────────────────────────────────────────────────────┘  │
└─────────────────────────────────────────────────────────────────────────┘

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

# elasticsearch.yml - Data node configuration
cluster.name: production-logs
node.name: es-data-01
 
# Node roles (Elasticsearch 7.9+)
node.roles: [ data, data_content, data_hot ]
 
# Network
network.host: 0.0.0.0
http.port: 9200
transport.port: 9300
 
# Discovery
discovery.seed_hosts:
  - es-master-01.internal:9300
  - es-master-02.internal:9300
  - es-master-03.internal:9300
 
# Memory: Heap should be ~50% of RAM, max 32GB
# Set via ES_JAVA_OPTS: -Xms16g -Xmx16g
 
# Disk space thresholds
cluster.routing.allocation.disk.watermark.low: 85%
cluster.routing.allocation.disk.watermark.high: 90%
cluster.routing.allocation.disk.watermark.flood_stage: 95%
 
# Index lifecycle management
action.destructive_requires_name: true
 
# Security (X-Pack)
xpack.security.enabled: true
xpack.security.transport.ssl.enabled: true

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
87
88
89
90
91
92

# /etc/logstash/conf.d/logs-pipeline.conf
 
input {
  beats {
    port => 5044
    ssl => true
    ssl_certificate => "/etc/logstash/ssl/logstash.crt"
    ssl_key => "/etc/logstash/ssl/logstash.key"
  }
  
  # Alternative: Direct Kafka consumption
  kafka {
    bootstrap_servers => "kafka-1:9092,kafka-2:9092,kafka-3:9092"
    topics => ["application-logs"]
    group_id => "logstash-consumers"
    codec => json
  }
}
 
filter {
  # Parse JSON body if not already parsed
  if [message] =~ /^\{/ {
    json {
      source => "message"
      target => "parsed"
    }
    
    # Move parsed fields to top level if successful
    if "_jsonparsefailure" not in [tags] {
      mutate {
        rename => { "[parsed][timestamp]" => "timestamp" }
        rename => { "[parsed][level]" => "level" }
        rename => { "[parsed][service]" => "service" }
        remove_field => [ "message", "parsed" ]
      }
    }
  }
  
  # Parse timestamp
  date {
    match => [ "timestamp", "ISO8601" ]
    target => "@timestamp"
    remove_field => [ "timestamp" ]
  }
  
  # Add processing metadata
  mutate {
    add_field => {
      "processed_at" => "%{+ISO8601}"
      "pipeline_version" => "2.1.0"
    }
  }
  
  # GeoIP enrichment for client IPs
  if [client_ip] {
    geoip {
      source => "client_ip"
      target => "geoip"
    }
  }
  
  # Drop DEBUG logs in production (optional)
  if [level] == "DEBUG" {
    drop { }
  }
  
  # Redact sensitive fields
  mutate {
    gsub => [
      "password", ".+", "[REDACTED]",
      "credit_card", "\d{12}(\d{4})", "************\1"
    ]
  }
}
 
output {
  elasticsearch {
    hosts => ["https://es-data-01:9200", "https://es-data-02:9200"]
    index => "logs-%{[service]}-%{+YYYY.MM.dd}"
    user => "logstash_writer"
    password => "${LOGSTASH_ES_PASSWORD}"
    ssl => true
    cacert => "/etc/logstash/ssl/ca.crt"
  }
  
  # Dead letter queue for failed documents
  if "_jsonparsefailure" in [tags] {
    file {
      path => "/var/log/logstash/dlq/%{+YYYY-MM-dd}.json"
    }
  }
}

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

{
  "index_patterns": ["logs-*"],
  "priority": 100,
  "template": {
    "settings": {
      "number_of_shards": 3,
      "number_of_replicas": 1,
      "codec": "best_compression",
      "refresh_interval": "30s",
      
      "index.lifecycle.name": "logs-lifecycle-policy",
      "index.lifecycle.rollover_alias": "logs-write",
      
      "index.translog.durability": "async",
      "index.translog.sync_interval": "5s"
    },
    "mappings": {
      "properties": {
        "@timestamp": { "type": "date" },
        "level": { "type": "keyword" },
        "logger": { "type": "keyword" },
        "service": { "type": "keyword" },
        "version": { "type": "keyword" },
        "environment": { "type": "keyword" },
        "trace_id": { "type": "keyword" },
        "span_id": { "type": "keyword" },
        "message": { 
          "type": "text",
          "fields": {
            "keyword": { "type": "keyword", "ignore_above": 256 }
          }
        },
        "error": {
          "properties": {
            "type": { "type": "keyword" },
            "message": { "type": "text" },
            "stack_trace": { "type": "text", "index": false }
          }
        },
        "http": {
          "properties": {
            "method": { "type": "keyword" },
            "status_code": { "type": "short" },
            "path": { "type": "keyword" },
            "latency_ms": { "type": "integer" }
          }
        },
        "user_id": { "type": "keyword" },
        "request_id": { "type": "keyword" }
      },
      "dynamic_templates": [
        {
          "strings_as_keywords": {
            "match_mapping_type": "string",
            "mapping": {
              "type": "keyword",
              "ignore_above": 256
            }
          }
        }
      ]
    }
  }
}

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

{
  "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": {
            "require": {
              "data": "warm"
            }
          },
          "set_priority": {
            "priority": 50
          }
        }
      },
      "cold": {
        "min_age": "30d",
        "actions": {
          "allocate": {
            "require": {
              "data": "cold"
            }
          },
          "freeze": {},
          "set_priority": {
            "priority": 0
          }
        }
      },
      "delete": {
        "min_age": "90d",
        "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

# AWS CDK TypeScript - OpenSearch Domain
const domain = new opensearch.Domain(this, 'LogsDomain', {
  version: opensearch.EngineVersion.OPENSEARCH_2_11,
  
  capacity: {
    dataNodes: 3,
    dataNodeInstanceType: 'r6g.large.search',
    masterNodes: 3,
    masterNodeInstanceType: 'm6g.large.search',
  },
  
  ebs: {
    volumeSize: 500,
    volumeType: ec2.EbsDeviceVolumeType.GP3,
  },
  
  nodeToNodeEncryption: true,
  encryptionAtRest: {
    enabled: true,
  },
  
  vpc: vpc,
  vpcSubnets: [{ subnetType: ec2.SubnetType.PRIVATE_ISOLATED }],
  
  zoneAwareness: {
    enabled: true,
    availabilityZoneCount: 3,
  },
  
  logging: {
    slowSearchLogEnabled: true,
    appLogEnabled: true,
    slowIndexLogEnabled: true,
  },
  
  fineGrainedAccessControl: {
    masterUserName: 'admin',
    masterUserPassword: secretsManager.Secret.fromSecretNameV2(...),
  },
});

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

# Basic label selector (like Prometheus)
{namespace="production", app="payment-service"}
 
# Filter by log level using label
{namespace="production", level="error"}
 
# Regex filter on log content (after label selection)
{namespace="production", app="payment-service"} |= "payment_failed"
 
# Negative filter (exclude lines)
{namespace="production"} != "healthcheck"
 
# JSON parsing for structured logs
{namespace="production", app="order-service"} 
  | json 
  | user_id="usr_a1b2c3"
 
# Aggregation (like Prometheus)
sum(rate({namespace="production", level="error"}[5m])) by (app)
 
# Pattern extraction for metrics
{namespace="production"} 
  | pattern "<timestamp> <level> <_> latency=<latency>ms"
  | latency > 500

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

groups:
- name: elasticsearch
  rules:
  # Cluster health not green
  - alert: ElasticsearchClusterHealthYellow
    expr: elasticsearch_cluster_health_status{color="yellow"} == 1
    for: 10m
    labels:
      severity: warning
    annotations:
      summary: "Elasticsearch cluster health is yellow"
      description: "Some replicas are not allocated. Check node health."
 
  - alert: ElasticsearchClusterHealthRed
    expr: elasticsearch_cluster_health_status{color="red"} == 1
    for: 5m
    labels:
      severity: critical
    annotations:
      summary: "Elasticsearch cluster health is RED"
      description: "Primary shards are unassigned. Data may be unavailable."
 
  # Disk space
  - alert: ElasticsearchDiskSpaceLow
    expr: elasticsearch_filesystem_data_available_bytes / elasticsearch_filesystem_data_size_bytes < 0.15
    for: 10m
    labels:
      severity: warning
    annotations:
      summary: "Elasticsearch disk space low on {{ $labels.node }}"
 
  # Ingest rate drop (may indicate collector failure)
  - alert: ElasticsearchIngestRateDrop
    expr: rate(elasticsearch_indices_indexing_index_total[5m]) < 0.5 * avg_over_time(rate(elasticsearch_indices_indexing_index_total[5m])[1h:5m])
    for: 10m
    labels:
      severity: warning
    annotations:
      summary: "Elasticsearch ingest rate dropped significantly"
      description: "May indicate log collector failures or network issues."
 
  # JVM heap pressure
  - alert: ElasticsearchJVMHeapHigh
    expr: elasticsearch_jvm_memory_used_bytes{area="heap"} / elasticsearch_jvm_memory_max_bytes{area="heap"} > 0.85
    for: 15m
    labels:
      severity: warning
    annotations:
      summary: "Elasticsearch JVM heap usage high on {{ $labels.node }}"

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

// ❌ SLOW: Full text search without filters
{
  "query": {
    "match": { "message": "payment failed" }
  }
}
 
// ✅ FAST: Time + keyword filters narrow scope first
{
  "query": {
    "bool": {
      "must": [
        { "match": { "message": "payment failed" } }
      ],
      "filter": [
        { "range": { "@timestamp": { "gte": "now-1h" } } },
        { "term": { "level": "error" } },
        { "term": { "service": "payment-service" } }
      ]
    }
  },
  "size": 100,
  "sort": [{ "@timestamp": "desc" }]
}
 
// Aggregation for error breakdown (no document fetch)
{
  "size": 0,
  "query": {
    "bool": {
      "filter": [
        { "range": { "@timestamp": { "gte": "now-24h" } } },
        { "term": { "level": "error" } }
      ]
    }
  },
  "aggs": {
    "errors_by_service": {
      "terms": { "field": "service", "size": 20 },
      "aggs": {
        "error_types": {
          "terms": { "field": "error.type", "size": 10 }
        }
      }
    }
  }
}
 
// Get specific trace (fast: keyword exact match)
{
  "query": {
    "term": { "trace_id": "4bf92f3577b34da6a3ce929d0e0e4736" }
  },
  "sort": [{ "@timestamp": "asc" }]
}