ELK Stack: 로그 수집과 분석 플랫폼 구축하기

복잡한 전문 검색이 필요한가? (예: 에러 메시지 패턴 분석)
├── YES → ELK/EFK
│         ├── Kubernetes 환경? → EFK (Fluentd = CNCF 표준)
│         └── 그 외? → ELK
└── NO  → Grafana Loki (비용 효율 최고)
          ├── 이미 Prometheus + Grafana 사용 중? → Loki 강추
          └── 로그 볼륨이 매우 큰가? → Loki (스토리지 비용 10배↓)

애플리케이션
    │
    ▼
┌───────────┐     ┌───────────┐     ┌───────────────┐     ┌────────┐
│ Filebeat  │────→│ Logstash  │────→│Elasticsearch  │────→│ Kibana │
│ (수집)    │     │ (변환)    │     │ (저장/검색)   │     │(시각화)│
└───────────┘     └───────────┘     └───────────────┘     └────────┘
    경량 Agent       파싱/필터/         역인덱스            대시보드
    Back-pressure    라우팅             분산 검색           알람

┌─ Filebeat ──────────────────────┐
│                                  │
│  Registry (파일별 읽기 위치 기록) │
│  ┌─────────────────────────┐    │
│  │ /var/log/app.log → 8240 │    │
│  │ /var/log/err.log → 1024 │    │
│  └─────────────────────────┘    │
│                                  │
│  Harvester (파일별 1개씩 읽기)   │
│  ┌──────┐  ┌──────┐            │
│  │app.log│  │err.log│           │
│  └──────┘  └──────┘            │
│                                  │
│  Spooler (배치 + Back-pressure) │
│  → Output (Logstash/ES)         │
└──────────────────────────────────┘

filebeat.inputs:
  # 1) 애플리케이션 로그 (JSON 형식)
  - type: log
    enabled: true
    paths:
      - /var/log/myapp/*.log
    json.keys_under_root: true
    json.add_error_key: true
    json.overwrite_keys: true
    
    # 멀티라인 (Java Stack Trace)
    multiline.pattern: '^\d{4}-\d{2}-\d{2}'
    multiline.negate: true
    multiline.match: after
    multiline.max_lines: 50
    multiline.timeout: 5s
    
    # 메타데이터
    fields:
      app: myapp
      env: production
      team: backend
    fields_under_root: true
    
    # 성능: 최대 5000개 이벤트를 메모리에 버퍼
    harvester_buffer_size: 65536
    max_bytes: 10485760
    
  # 2) Nginx 액세스 로그
  - type: log
    enabled: true
    paths:
      - /var/log/nginx/access.log
    fields:
      app: nginx
      type: access
    fields_under_root: true

# 출력 설정
output.logstash:
  hosts: ["logstash:5044"]
  # Back-pressure: Logstash가 느리면 자동으로 전송 속도 조절
  bulk_max_size: 2048
  worker: 2
  loadbalance: true  # 여러 Logstash 인스턴스에 분산

# 내부 큐 설정
queue.mem:
  events: 4096
  flush.min_events: 512
  flush.timeout: 1s

# 프로세서 (Logstash 전에 경량 변환)
processors:
  - drop_fields:
      fields: ["agent.ephemeral_id", "agent.hostname", "agent.id"]
  - add_host_metadata: ~
  - add_cloud_metadata: ~

input {
  beats {
    port => 5044
    ssl => true
    ssl_certificate => "/etc/logstash/certs/logstash.crt"
    ssl_key => "/etc/logstash/certs/logstash.key"
  }
}

filter {
  # 1) 앱 타입별 분기
  if [app] == "myapp" {
    # JSON 구조화 로그 파싱
    if [message] =~ /^\{/ {
      json {
        source => "message"
        target => "log"
        remove_field => ["message"]
      }
    }
    
    # 날짜 파싱
    date {
      match => ["[log][timestamp]", "ISO8601", "yyyy-MM-dd HH:mm:ss.SSS"]
      target => "@timestamp"
    }
    
    # 로그 레벨 정규화
    mutate {
      uppercase => ["[log][level]"]
    }
  }
  
  if [app] == "nginx" {
    # Nginx 로그 파싱 (Grok)
    grok {
      match => {
        "message" => '%{IPORHOST:client_ip} - %{DATA:user} \[%{HTTPDATE:timestamp}\] "%{WORD:method} %{URIPATHPARAM:request} HTTP/%{NUMBER:http_version}" %{NUMBER:status:int} %{NUMBER:bytes:int} "%{DATA:referrer}" "%{DATA:user_agent}" %{NUMBER:request_time:float}'
      }
    }
    
    date {
      match => ["timestamp", "dd/MMM/yyyy:HH:mm:ss Z"]
      target => "@timestamp"
      remove_field => ["timestamp"]
    }
    
    # GeoIP (클라이언트 IP → 위치)
    geoip {
      source => "client_ip"
      target => "geo"
    }
    
    # User-Agent 파싱
    useragent {
      source => "user_agent"
      target => "ua"
    }
  }
  
  # 2) 공통 처리
  # 민감 정보 마스킹
  mutate {
    gsub => [
      "message", '\b\d{3}-\d{2}-\d{4}\b', '***-**-****',  # SSN
      "message", '\b[\w.+-]+@[\w-]+\.[\w.]+\b', '****@****.***'  # 이메일
    ]
  }
  
  # 불필요 이벤트 드롭
  if [log][level] == "DEBUG" and [env] == "production" {
    drop {}
  }
  
  # 3) 메트릭 추출
  metrics {
    meter => "events"
    add_tag => "metric"
  }
}

output {
  # 앱별 인덱스 분리
  if [app] == "myapp" {
    elasticsearch {
      hosts => ["https://es-node1:9200", "https://es-node2:9200"]
      index => "myapp-logs-%{+YYYY.MM.dd}"
      user => "logstash_writer"
      password => "${LOGSTASH_ES_PASSWORD}"
      ssl => true
      cacert => "/etc/logstash/certs/ca.crt"
      
      # 인덱스 템플릿 자동 적용
      template_name => "myapp-logs"
      template => "/etc/logstash/templates/myapp-logs.json"
      template_overwrite => true
    }
  }
  
  if [app] == "nginx" {
    elasticsearch {
      hosts => ["https://es-node1:9200"]
      index => "nginx-access-%{+YYYY.MM.dd}"
      user => "logstash_writer"
      password => "${LOGSTASH_ES_PASSWORD}"
      ssl => true
      cacert => "/etc/logstash/certs/ca.crt"
    }
  }
  
  # 에러 로그는 별도 인덱스 + 알림
  if [log][level] == "ERROR" {
    elasticsearch {
      hosts => ["https://es-node1:9200"]
      index => "errors-%{+YYYY.MM.dd}"
      user => "logstash_writer"
      password => "${LOGSTASH_ES_PASSWORD}"
      ssl => true
      cacert => "/etc/logstash/certs/ca.crt"
    }
  }
}

version: '3.8'

services:
  # ── Elasticsearch ──
  elasticsearch:
    image: docker.elastic.co/elasticsearch/elasticsearch:8.12.0
    container_name: es01
    environment:
      - node.name=es01
      - cluster.name=logging-cluster
      - discovery.type=single-node
      - xpack.security.enabled=true
      - xpack.security.http.ssl.enabled=false  # 개발용. 운영은 true
      - ELASTIC_PASSWORD=${ELASTIC_PASSWORD:-changeme}
      - "ES_JAVA_OPTS=-Xms2g -Xmx2g"
      # JVM 힙은 전체 메모리의 50% 이하, 최대 32GB (CompressedOops)
    ulimits:
      memlock:
        soft: -1
        hard: -1
      nofile:
        soft: 65536
        hard: 65536
    volumes:
      - es-data:/usr/share/elasticsearch/data
    ports:
      - "9200:9200"
    healthcheck:
      test: ["CMD-SHELL", "curl -s -u elastic:${ELASTIC_PASSWORD:-changeme} http://localhost:9200/_cluster/health | grep -q '\"status\":\"green\\|yellow\"'"]
      interval: 30s
      timeout: 10s
      retries: 5

  # ── Logstash ──
  logstash:
    image: docker.elastic.co/logstash/logstash:8.12.0
    container_name: logstash
    volumes:
      - ./logstash/pipeline:/usr/share/logstash/pipeline:ro
      - ./logstash/config/logstash.yml:/usr/share/logstash/config/logstash.yml:ro
    environment:
      - "LS_JAVA_OPTS=-Xms1g -Xmx1g"
      - LOGSTASH_ES_PASSWORD=${ELASTIC_PASSWORD:-changeme}
    ports:
      - "5044:5044"   # Beats input
      - "9600:9600"   # Monitoring API
    depends_on:
      elasticsearch:
        condition: service_healthy

  # ── Kibana ──
  kibana:
    image: docker.elastic.co/kibana/kibana:8.12.0
    container_name: kibana
    environment:
      - ELASTICSEARCH_HOSTS=http://elasticsearch:9200
      - ELASTICSEARCH_USERNAME=kibana_system
      - ELASTICSEARCH_PASSWORD=${ELASTIC_PASSWORD:-changeme}
      - SERVER_NAME=kibana
      - SERVER_BASEPATH=""
    ports:
      - "5601:5601"
    depends_on:
      elasticsearch:
        condition: service_healthy

  # ── Filebeat ──
  filebeat:
    image: docker.elastic.co/beats/filebeat:8.12.0
    container_name: filebeat
    user: root
    command: filebeat -e -strict.perms=false
    volumes:
      - ./filebeat/filebeat.yml:/usr/share/filebeat/filebeat.yml:ro
      - /var/log:/var/log:ro
      - /var/lib/docker/containers:/var/lib/docker/containers:ro
      - filebeat-data:/usr/share/filebeat/data  # Registry 영속
    depends_on:
      logstash:
        condition: service_started

volumes:
  es-data:
    driver: local
  filebeat-data:
    driver: local

http.host: "0.0.0.0"
xpack.monitoring.elasticsearch.hosts: ["http://elasticsearch:9200"]
xpack.monitoring.elasticsearch.username: "logstash_system"
xpack.monitoring.elasticsearch.password: "${LOGSTASH_ES_PASSWORD}"

# 파이프라인 성능 튜닝
pipeline.workers: 4           # CPU 코어 수에 맞춤
pipeline.batch.size: 500      # 한 번에 처리할 이벤트 수
pipeline.batch.delay: 50      # ms, 배치 채우기 대기 시간

# 영속 큐 (Logstash 재시작 시 데이터 유실 방지)
queue.type: persisted
queue.max_bytes: 1gb
queue.checkpoint.writes: 1024

# Dead Letter Queue (파싱 실패 이벤트 보존)
dead_letter_queue.enable: true
dead_letter_queue.max_bytes: 512mb

┌─────────────────────────────────────────────────────────────────┐
│                     Index Lifecycle                               │
│                                                                   │
│  Hot (SSD)          Warm (HDD)         Cold             Delete   │
│  ─────────          ─────────         ──────            ──────   │
│  0~7일              7~30일            30~90일            90일+    │
│  활발한 쓰기/읽기    읽기 위주          거의 접근 X       자동 삭제 │
│                                                                   │
│  Rollover 조건:     Shrink:           Freeze:                    │
│  - 50GB 초과        1 Shard로 병합     읽기 전용                   │
│  - 7일 경과         Force Merge:      Searchable                  │
│  - 1억 문서          1 Segment로 병합  Snapshot 가능               │
└─────────────────────────────────────────────────────────────────┘

PUT _ilm/policy/logs_lifecycle
{
  "policy": {
    "phases": {
      "hot": {
        "min_age": "0ms",
        "actions": {
          "rollover": {
            "max_age": "7d",
            "max_primary_shard_size": "50gb",
            "max_docs": 100000000
          },
          "set_priority": { "priority": 100 },
          "readonly": {}
        }
      },
      "warm": {
        "min_age": "7d",
        "actions": {
          "shrink": { "number_of_shards": 1 },
          "forcemerge": { "max_num_segments": 1 },
          "set_priority": { "priority": 50 },
          "allocate": {
            "number_of_replicas": 1,
            "require": { "data": "warm" }
          }
        }
      },
      "cold": {
        "min_age": "30d",
        "actions": {
          "set_priority": { "priority": 0 },
          "allocate": {
            "number_of_replicas": 0,
            "require": { "data": "cold" }
          },
          "freeze": {}
        }
      },
      "delete": {
        "min_age": "90d",
        "actions": {
          "delete": {}
        }
      }
    }
  }
}

PUT _index_template/myapp-logs
{
  "index_patterns": ["myapp-logs-*"],
  "template": {
    "settings": {
      "number_of_shards": 2,
      "number_of_replicas": 1,
      "index.lifecycle.name": "logs_lifecycle",
      "index.lifecycle.rollover_alias": "myapp-logs",
      "index.routing.allocation.require.data": "hot",
      "index.refresh_interval": "5s"
    },
    "mappings": {
      "dynamic": "strict",
      "properties": {
        "@timestamp": { "type": "date" },
        "level": { "type": "keyword" },
        "logger": { "type": "keyword" },
        "thread": { "type": "keyword" },
        "message": { "type": "text" },
        "traceId": { "type": "keyword" },
        "spanId": { "type": "keyword" },
        "app": { "type": "keyword" },
        "env": { "type": "keyword" },
        "host": { "type": "keyword" },
        "exception": {
          "properties": {
            "class": { "type": "keyword" },
            "message": { "type": "text" },
            "stacktrace": { "type": "text", "index": false }
          }
        }
      }
    }
  },
  "priority": 200
}

// 초기 인덱스 + alias 생성
PUT myapp-logs-000001
{
  "aliases": {
    "myapp-logs": {
      "is_write_index": true
    }
  }
}

implementation 'net.logstash.logback:logstash-logback-encoder:7.4'

<?xml version="1.0" encoding="UTF-8"?>
<configuration>
    <springProperty scope="context" name="APP_NAME" source="spring.application.name"/>
    <springProperty scope="context" name="APP_ENV" source="spring.profiles.active"/>
    
    <!-- 콘솔 출력 (개발용) -->
    <appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
        <encoder>
            <pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern>
        </encoder>
    </appender>
    
    <!-- Logstash TCP 전송 -->
    <appender name="LOGSTASH" class="net.logstash.logback.appender.LogstashTcpSocketAppender">
        <destination>logstash:5000</destination>
        <encoder class="net.logstash.logback.encoder.LogstashEncoder">
            <!-- 커스텀 필드 추가 -->
            <customFields>{"app":"${APP_NAME}","env":"${APP_ENV}"}</customFields>
            <!-- 스택 트레이스 해시 (동일 에러 그룹핑용) -->
            <throwableConverter class="net.logstash.logback.stacktrace.ShortenedThrowableConverter">
                <maxDepthPerThrowable>20</maxDepthPerThrowable>
                <maxLength>2048</maxLength>
                <shortenedClassNameLength>20</shortenedClassNameLength>
                <rootCauseFirst>true</rootCauseFirst>
            </throwableConverter>
            <!-- MDC (traceId, spanId 등) 자동 포함 -->
            <includeMdcKeyName>traceId</includeMdcKeyName>
            <includeMdcKeyName>spanId</includeMdcKeyName>
            <includeMdcKeyName>userId</includeMdcKeyName>
        </encoder>
        
        <!-- 재연결 설정 -->
        <reconnectionDelay>5 seconds</reconnectionDelay>
        <keepAliveDuration>5 minutes</keepAliveDuration>
        
        <!-- 비동기 + 큐 설정 -->
        <ringBufferSize>8192</ringBufferSize>
    </appender>
    
    <!-- 파일 출력 (Filebeat가 수집) -->
    <appender name="FILE" class="ch.qos.logback.core.rolling.RollingFileAppender">
        <file>/var/log/myapp/app.log</file>
        <rollingPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy">
            <fileNamePattern>/var/log/myapp/app.%d{yyyy-MM-dd}.%i.log</fileNamePattern>
            <maxFileSize>100MB</maxFileSize>
            <maxHistory>7</maxHistory>
            <totalSizeCap>2GB</totalSizeCap>
        </rollingPolicy>
        <encoder class="net.logstash.logback.encoder.LogstashEncoder"/>
    </appender>
    
    <!-- 프로파일별 설정 -->
    <springProfile name="local">
        <root level="DEBUG">
            <appender-ref ref="CONSOLE"/>
        </root>
    </springProfile>
    
    <springProfile name="production">
        <root level="INFO">
            <appender-ref ref="FILE"/>
            <appender-ref ref="LOGSTASH"/>
        </root>
    </springProfile>
</configuration>

@Component
public class LoggingFilter implements Filter {
    @Override
    public void doFilter(ServletRequest request, ServletResponse response,
                         FilterChain chain) throws IOException, ServletException {
        HttpServletRequest req = (HttpServletRequest) request;
        
        // MDC에 요청 컨텍스트 저장 → 모든 로그에 자동 포함
        MDC.put("requestId", req.getHeader("X-Request-ID") != null
            ? req.getHeader("X-Request-ID")
            : UUID.randomUUID().toString());
        MDC.put("clientIp", req.getRemoteAddr());
        MDC.put("method", req.getMethod());
        MDC.put("uri", req.getRequestURI());
        
        try {
            chain.doFilter(request, response);
        } finally {
            MDC.clear();
        }
    }
}

// 사용 예시
@Service
public class OrderService {
    private static final Logger log = LoggerFactory.getLogger(OrderService.class);
    
    public Order createOrder(OrderRequest request) {
        MDC.put("userId", request.getUserId());
        MDC.put("orderId", request.getOrderId());
        
        log.info("주문 생성 시작");
        // → {"@timestamp":"...","level":"INFO","message":"주문 생성 시작",
        //    "requestId":"abc-123","userId":"user-456","orderId":"order-789",...}
        
        // ... 비즈니스 로직
        
        log.info("주문 생성 완료 amount={}", request.getAmount());
        return order;
    }
}

Kibana → Stack Management → Data Views → Create

Name:      myapp-logs
Index:     myapp-logs-*
Timestamp: @timestamp

// 5분간 에러 100건 초과 시 Slack 알림
PUT _watcher/watch/error_spike_alert
{
  "trigger": {
    "schedule": { "interval": "5m" }
  },
  "input": {
    "search": {
      "request": {
        "indices": ["myapp-logs-*"],
        "body": {
          "query": {
            "bool": {
              "filter": [
                { "term": { "level": "ERROR" } },
                { "range": { "@timestamp": { "gte": "now-5m" } } }
              ]
            }
          }
        }
      }
    }
  },
  "condition": {
    "compare": { "ctx.payload.hits.total.value": { "gt": 100 } }
  },
  "actions": {
    "notify_slack": {
      "webhook": {
        "method": "POST",
        "url": "https://hooks.slack.com/services/xxx",
        "body": "{\"text\":\"🚨 에러 급증! 최근 5분간 {{ctx.payload.hits.total.value}}건 발생\"}"
      }
    }
  }
}

// 읽기 전용 역할
POST _security/role/log_reader
{
  "indices": [{
    "names": ["myapp-logs-*"],
    "privileges": ["read", "view_index_metadata"]
  }]
}

// Logstash 쓰기 역할
POST _security/role/logstash_writer
{
  "cluster": ["manage_index_templates", "monitor"],
  "indices": [{
    "names": ["myapp-logs-*", "nginx-access-*", "errors-*"],
    "privileges": ["create_index", "write", "manage"]
  }]
}

// 사용자 생성
POST _security/user/logstash_internal
{
  "password": "strong-password-here",
  "roles": ["logstash_writer"],
  "full_name": "Logstash Internal"
}

# Logstash 모니터링 API
curl -s localhost:9600/_node/stats/pipeline | jq '
  .pipelines.main.events | {
    in: .in,
    filtered: .filtered,
    out: .out,
    queue_push_duration_ms: .queue_push_duration_in_millis,
    duration_ms: .duration_in_millis
  }
'

# 파이프라인 지연 확인
curl -s localhost:9600/_node/stats/pipeline | jq '
  .pipelines.main.plugins.filters[] | {
    name: .name,
    events_in: .events.in,
    duration_ms: .events.duration_in_millis,
    avg_ms: (.events.duration_in_millis / (.events.in + 1))
  }
'

# jvm.options
-Xms4g          # 전체 메모리의 50%
-Xmx4g          # Xms = Xmx (동일하게!)
# 최대 32GB (CompressedOops 한계)
# 나머지 50%는 OS File Cache가 사용 (Lucene 성능에 중요!)