ELK 스택을 활용한 기업용 통합 로깅 시스템 설계 및 구현

목표 정의 및 요구사항 분석

중앙 집중형 로그 관리 시스템을 구축하기 위해서는 다음과 같은 핵심 기능이 지원되어야 합니다:

  • 실시간 모니터링: 경고 (Warning) 이상의 에러 로그는 즉시 포착되어야 합니다.
  • 보안 감사: 서버 내 모든 사용자의 명령어 실행 이력을 추적할 수 있어야 합니다.
  • Web 트래픽 분석: Nginx 접근 로그를 시각화 도구로 연동하여 분석합니다.
  • 애플리케이션 진단: Tomcat 및 Log4j 를 통한 Java 앱 런타임 로그를 수집합니다.
  • 성능 최적화: MySQL의 느린 쿼리 로그를 식별하여 DB 부하를 관리합니다.

1. 시스템 커널 및 보안 로그 수집 (Rsyslog)

CentOS 계열 OS에는 기본으로 포함된 Rsyslog 도구를 활용하여 시스템 내부 이벤트를 외부 전송합니다. 이 도구는 멀티쓰레드 처리가 가능하며, TCP/SSL 등 다양한 프로토콜과 관계형 DB 저장소 연동을 지원합니다.

시스템 장애나 비정상 상태를 사전에 감지하기 위해 로그 레벨은 정보 (info) 보다 높은 등급인 경고 (warning, 레벨 3) 이상으로 필터링하여 전파하도록 설정합니다.

로그 라우팅 설정

/etc/rsyslog.conf 파일을 편집하여 특정 로그 유형을 별도 파일로 분리하고, 리모스트서버로 전송하는 규칙을 추가합니다.

# 경고级以上的 시스템 로그 전용 저장
local3.* /var/log/boot.log
*.warning /var/log/warning_log

# 전체 로그를 수집 서버 로 포트 5514 로 전송
*.* @@log-server:5514

수정 후 서비스 재시작을 적용합니다.

service rsyslog restart

Logstash 수신기 구성

전송된 로그 스트림을 받아들이기 위한 logstash-system.conf 파일을 작성합니다. TCP 소켓을 통해 들어오는 데이터를 파싱하여 ElasticSearch 에 인덱싱합니다.

input {
  tcp {
    port => 5514
    type => "sys_monitor"
    tags => ["server"]
  }
}

filter {
  if [type] == "sys_monitor" {
    grok {
      match => { 
        "message" => "%{SYSLOGTIMESTAMP:sys_ts} %{SYSLOGHOST:src_host} %{DATA:proc_name}(?:\[%{POSINT:pid}\])?: %{GREEDYDATA:msg_body}" 
      }
    }
    date {
      match => ["sys_ts", "MMM  d HH:mm:ss", "MMM dd HH:mm:ss"]
      target => "@timestamp"
    }
  }
}

output {
  elasticsearch {
    hosts => ["log-server:9200"]
    index => "sys-log-%{+YYYY.MM.dd}"
  }
}

테스트를 위해 Telnet 을 사용하여 가상의 syslog 메시지를 주입하고 Kibana 에서 인덱스가 생성되는지 검증합니다.

2. 사용자 명령어 이력 추적

SSH 접속 후 사용자가 수행한 명령어는 기본적으로 .bash_history 에 기록되지만, 실시간 모니터링을 위해 별도의 로그 파일로 집계해야 합니다. 이를 위해 환경 변수 설정을 변경하여 명령어가 입력되는 순간 로그에 기록되도록 조정합니다.

Shell 환경 커스터마이징

/etc/bashrc 파일에 다음 내용을 추가하여 JSON 형식의 로그 파일을 생성하게 합니다.

HISTSIZE=4000
HISTFILESIZE=4000
HISTTIMEFORMAT='%F %T '

# 사용자별 명령어 로그 경로 지정
CMD_LOG_PATH="/var/log/cmd_trace.log"

export PROMPT_COMMAND='history 1 | tail -1 | awk "{sub(/^[ ]+[0-9]+/, \"\", \$0); print \$0}" >> '$CMD_LOG_PATH'

설정 반영을 위해 터미널을 다시 시작하거나 source /etc/bashrc를 실행합니다.

로그 파이프라인 연결

생성된 텍스트 로그 파일을 읽어서 ES 에 전송하는 Logstash 설정입니다.

input {
  file {
    path => "/var/log/cmd_trace.log"
    type => "user_cmd"
  }
}

output {
  elasticsearch {
    hosts => ["log-server:9200"]
    index => "audit-trail-%{+YYYY.MM.dd}"
  }
}

3. 고 부하 환경에서의 웹 로그 처리 (Nginx + Buffer)

직접적인 File -> Logstash -> ES 구조는 트래픽 폭주 시 ES 의 Write 압력으로 인해 데이터 손실이 발생할 수 있습니다. 이를 방지하기 위해 중간 버퍼 큐 (Redis, Kafka) 를 도입하여 백프레스 (Backpressure) 를 관리합니다.

Nginx 로그 포맷 정의

위치 정보가 추출 가능하도록 IP, User-Agent, Referer 등을 포함하는 포맷을 설정합니다.

http {
  log_format detailed '$remote_addr - $remote_user [$time_local] "$request" '
                      '$status $body_bytes_sent "$http_referer" '
                      '"$http_user_agent"';
  access_log /var/log/nginx/access.log detailed;
}

Agent: 로그 수집 및 버퍼링

로그 파일을 읽어서 Redis 리스트에 푸시하는 수집 에이전트입니다.

input {
  file {
    path => "/var/log/nginx/access.log"
    type => "web_req"
  }
}

output {
  redis {
    host => "log-server"
    port => 6379
    data_type => "list"
    key => "nginx_stream"
  }
}

Indexer: 버퍼消费 및 변환

Redis 에서 데이터를 받아地理位置 정보를 추론하고 최종적으로 저장합니다.

input {
  redis {
    host => "log-server"
    data_type => "list"
    key => "nginx_stream"
  }
}

filter {
  grok {
    match => {
      "message" => "%{IP:client_ip} .* \[%{HTTPDATE:http_time}\] \"%{WORD:method} %{PATH:request_path}\" %{NUMBER:response_status}"
    }
  }
  geoip {
    source => "client_ip"
    target => "geo_info"
  }
  mutate {
    convert => { "[geo_info][location]" => "geo_point" }
  }
}

output {
  elasticsearch {
    hosts => ["log-server:9200"]
    index => "web-access-%{+YYYY.MM.dd}"
  }
}

4. 애플리케이션 예외 로그 분석 (Tomcat)

Java 기반 애플리케이션의 스택 트레이스는 여러 줄에 걸쳐 출력되는 특징이 있어, 표준 로그 수집기로는 단위로 인식되지 않을 수 있습니다. 두 가지 방식을 제안합니다.

방식 A: Multiline 플러그인 활용

단일 로그 라인 패턴을 기준으로 앞뒤 줄을 병합합니다. 먼저 logstash-filter-multiline 플러그인을 설치해야 합니다.

input {
  file {
    path => "/var/logs/tomcat/catalina.out"
  }
}

filter {
  multiline {
    pattern => "^\d{4}-\d{2}-\d{2}"
    negate => true
    what => "previous"
  }
  grok {
    match => { "message" => "%{DATETIME:timestamp} \[%{LOGLEVEL:level}\]: %{GREEDYDATA:text}" }
  }
}

output { elasticsearch { ... } }

방식 B: JSON Event Layout 강제 (추천)

로그 소스 자체를 구조화된 JSON 형태로 변경하는 것이 가장 안정적입니다. tomcat-juli 와 log4j-jsonevent-layout 라이브러리를 라이브러리 폴더에 추가하고, log4j.properties를 아래와 같이 수정합니다.

log4j.rootCategory=INFO, CONSOLE
log4j.appender.CONSOLE=net.logstash.log4j.JSONEventLayoutV1
log4j.appender.CONSOLE.File=/var/logs/tomcat/app.json.log
log4j.appender.CONSOLE.DatePattern=.yyyy-MM-dd

이 방식은 Logstash 에서 단순 JSON 코디코만으로도 구조화된 데이터를 얻을 수 있어 성능과 정확도가 우수합니다.

5. 데이터베이스 성능 저조 원인 파악 (MySQL Slow Log)

서버 부하의 상당部分是 inefficient SQL 문의 발생 때문입니다. 이를 해결하기 위해 MySQL 에서 1 초 이상 소요되는 쿼리를 캡처해야 합니다.

Slow Log 활성화

/etc/my.cnf 설정 파일에 다음 파라미터를 명시합니다.

[mysqld]
slow_query_log = ON
slow_query_log_file = /var/log/mysql/slow.log
long_query_time = 1

또는 세션 동안만 활성화하려면 SET GLOBAL long_query_time=1; 명령어를 사용합니다.

로그 인사이트 도출

MySQL 은 각 쿼리에 대해 실행 시간, 잠금 시간, 스캔 행수 등의 메타데이터를 함께 남깁니다. 이러한 텍스트 블록을 하나의 이벤트로 묶어서 파싱합니다.

input {
  file {
    path => "/var/log/mysql/slow.log"
    codec => multiline {
      pattern => "^# Time:"
      what => "next"
    }
  }
}

filter {
  grok {
    match => {
      "message" => "(?m)^# User@Host: (?<user>\S+)\[\S+\] @ (?<host>.*?).*# Query_time: (?<duration>\S+).*Query: (?<sql_text>.*)"
    }
  }
}

output { elasticsearch { index => "db-slowquery-%{+YYYY.MM.dd}" } }

태그: elasticsearch Logstash Kibana rsyslog nginx

7월 20일 19:43에 게시됨

인기 태그

©2026 괴물 클럽