Copy & Run

Logstash로 nginx 로그를 Grok으로 파싱하고 ECS로 정규화하기

nginx 접근 로그를 Logstash grok으로 파싱해 ECS 필드 이름으로 정리하고, 파싱에 실패한 로그를 Elasticsearch 쿼리로 세어 별도 인덱스로 보냅니다. Docker로 Elasticsearch 8.19와 Logstash 8.19를 띄워 로컬에서 끝까지 돌립니다.

실습 노트 · 2026-10-16 발행 · 16분 · 2026-10 확인

nginx 접근 로그를 Logstash의 grok 필터로 파싱해 ECS(Elastic Common Schema) 필드 이름으로 정리하고, 파싱에 실패한 로그를 _grokparsefailure 태그로 찾아 개수를 세고 별도 인덱스로 보냅니다. Docker로 Elasticsearch 8.19와 Logstash 8.19를 띄워 로컬에서 끝까지 돌립니다.

이 글의 Elasticsearch는 인증과 암호화를 끈 구성이며 내 PC에서 하는 실습 전용입니다. 포트는 127.0.0.1에만 열고, 이 구성을 인터넷이나 공용 네트워크에 노출하지 않습니다. Kibana까지 붙이고 싶다면 Kibana 설치하고 Elasticsearch에 연결하기를 이어서 보시면 되고, 이 글은 그 글 없이도 처음부터 끝까지 돌아가도록 썼습니다. 샘플 로그는 직접 만든 가짜 로그이며 주소는 비공개 대역(10.x, 192.168.x)만 씁니다.

준비물

항목 내용
Docker Docker Desktop 또는 Docker Engine (확인한 버전 29.5.3)
Elasticsearch 8.19.0 (docker.elastic.co/elasticsearch/elasticsearch:8.19.0)
Logstash 8.19.0 (docker.elastic.co/logstash/logstash:8.19.0)
메모리 컨테이너 두 개에 약 2GB (Elasticsearch 힙 1GB, Logstash 힙 512MB)
호스트 포트 19400 (Elasticsearch, 127.0.0.1에만 바인딩)
예상 소요 시간 약 30분 (이미지 내려받기 제외)

명령은 bash 기준입니다. Windows에서는 Git Bash를 씁니다. 이미지는 두 개 합쳐 1GB가 넘으니 미리 받아 두면 편합니다.

docker pull docker.elastic.co/elasticsearch/elasticsearch:8.19.0
docker pull docker.elastic.co/logstash/logstash:8.19.0

Logstash 8 계열은 별도 설정이 없으면 모든 플러그인이 ECS v8 모드로 동작합니다. 이 글에서는 차이가 눈에 보이도록 grok 필터에 ecs_compatibility => v8을 명시해서 씁니다.

1. Elasticsearch 띄우고 샘플 로그 만들기

네트워크와 볼륨을 만들고 Elasticsearch를 띄웁니다. -p 127.0.0.1:19400:9200이 내 PC 안에서만 접속되게 하는 부분입니다.

docker network create cr-logstash-net
docker volume create cr-logstash-esdata

docker run -d --name cr-logstash-es --network cr-logstash-net \
  -p 127.0.0.1:19400:9200 \
  -e discovery.type=single-node \
  -e xpack.security.enabled=false \
  -e ES_JAVA_OPTS="-Xms1g -Xmx1g" \
  -v cr-logstash-esdata:/usr/share/elasticsearch/data \
  docker.elastic.co/elasticsearch/elasticsearch:8.19.0

until curl -sf http://127.0.0.1:19400 >/dev/null; do sleep 3; done
curl -s 'http://127.0.0.1:19400/_cluster/health?filter_path=status,number_of_nodes'

기동에는 20~40초가 걸립니다. 응답은 {"status":"green","number_of_nodes":1} 로 나오기도 하고 처음부터 yellow로 나오기도 하며, 환경마다 다를 수 있습니다. 인덱스를 만들면 복제본을 둘 노드가 없어서 yellow가 되는데, 단일 노드 실습에서는 정상입니다.

작업 폴더를 만들고 샘플 로그를 저장합니다. 45줄 가운데 40줄은 nginx 기본 combined 형식이고, 5줄은 일부러 형식에서 벗어나게 섞은 줄입니다.

mkdir -p ~/grok-demo/logs ~/grok-demo/pipeline && cd ~/grok-demo
cat > logs/access.log <<'LOG'
10.0.1.15 - - [16/Oct/2026:09:00:00 +0900] "GET /login HTTP/1.1" 302 911 "http://app.example.test/login" "Mozilla/5.0 (X11; Linux x86_64) Chrome/126.0"
10.0.1.22 - - [16/Oct/2026:09:00:07 +0900] "GET /api/search?q=test HTTP/1.1" 200 8433 "-" "Mozilla/5.0 (X11; Linux x86_64) Chrome/126.0"
10.0.1.15 - - [16/Oct/2026:09:01:14 +0900] "POST /static/style.css HTTP/1.1" 200 4063 "http://app.example.test/login" "python-requests/2.32.3"
192.168.50.3 - - [16/Oct/2026:09:01:21 +0900] "GET /api/search?q=test HTTP/1.1" 200 3777 "http://app.example.test/login" "Mozilla/5.0 (X11; Linux x86_64) Chrome/126.0"
10.0.1.22 - - [16/Oct/2026:09:02:28 +0900] "PUT /api/search?q=test HTTP/1.1" 302 932 "-" "curl/8.5.0"
10.0.1.15 - - [16/Oct/2026:09:02:35 +0900] "GET /static/style.css HTTP/1.1" 200 8978 "http://app.example.test/login" "Mozilla/5.0 (Windows NT 10.0; Win64; x64) Firefox/128.0"
this line is not an nginx access log
192.168.10.5 - - [16/Oct/2026:09:03:42 +0900] "PUT /login HTTP/1.1" 200 3198 "-" "Mozilla/5.0 (X11; Linux x86_64) Chrome/126.0"
192.168.50.3 - - [16/Oct/2026:09:03:49 +0900] "PUT / HTTP/1.1" 200 8253 "http://app.example.test/login" "python-requests/2.32.3"
192.168.10.5 - - [16/Oct/2026:09:04:56 +0900] "GET /admin HTTP/1.1" 500 6044 "-" "curl/8.5.0"
10.20.0.8 - - [16/Oct/2026:09:04:03 +0900] "GET /index.html HTTP/1.1" 304 0 "http://app.example.test/" "Mozilla/5.0 (Windows NT 10.0; Win64; x64) Firefox/128.0"
10.20.0.8 - - [16/Oct/2026:09:05:10 +0900] "POST /api/items/42 HTTP/1.1" 200 2054 "http://app.example.test/" "curl/8.5.0"
10.0.1.15 - - [16/Oct/2026:09:05:17 +0900] "GET /login HTTP/1.1" 500 7029 "http://app.example.test/login" "Mozilla/5.0 (X11; Linux x86_64) Chrome/126.0"
192.168.50.3 - - [16/Oct/2026:09:06:24 +0900] "PUT /api/search?q=test HTTP/1.1" 404 5692 "http://app.example.test/" "python-requests/2.32.3"
192.168.10.5 - - [16/Oct/2026:09:06:31 +0900] "PUT /admin HTTP/1.1" 200 1653 "http://app.example.test/" "Mozilla/5.0 (X11; Linux x86_64) Chrome/126.0"
192.168.50.3 - - [16/Oct/2026:09:07:38 +0900] "GET /api/items/42 HTTP/1.1" 500 4782 "http://app.example.test/" "Mozilla/5.0 (Windows NT 10.0; Win64; x64) Firefox/128.0"
10.0.1.15 - - [16/Oct/2026:09:21:05 +0900] "GET /broken HTTP/1.1" ok 512 "-" "curl/8.5.0"
10.20.0.8 - - [16/Oct/2026:09:07:45 +0900] "GET /admin HTTP/1.1" 404 2873 "-" "python-requests/2.32.3"
10.0.1.22 - - [16/Oct/2026:09:08:52 +0900] "GET /api/items?page=1 HTTP/1.1" 304 0 "http://app.example.test/login" "curl/8.5.0"
10.0.1.22 - - [16/Oct/2026:09:08:59 +0900] "POST /static/style.css HTTP/1.1" 500 1440 "http://app.example.test/" "python-requests/2.32.3"
10.20.0.8 - - [16/Oct/2026:09:09:06 +0900] "PUT /api/items/42 HTTP/1.1" 200 7173 "http://app.example.test/" "python-requests/2.32.3"
10.0.1.15 - - [16/Oct/2026:09:09:13 +0900] "GET /static/style.css HTTP/1.1" 200 2592 "-" "curl/8.5.0"
10.20.0.8 - - [16/Oct/2026:09:10:20 +0900] "GET /api/items?page=1 HTTP/1.1" 200 8065 "-" "Mozilla/5.0 (Windows NT 10.0; Win64; x64) Firefox/128.0"
10.20.0.8 - - [16/Oct/2026:09:10:27 +0900] "GET / HTTP/1.1" 200 6984 "http://app.example.test/" "Mozilla/5.0 (Windows NT 10.0; Win64; x64) Firefox/128.0"
192.168.50.3 - - [16/Oct/2026:09:11:34 +0900] "GET /health HTTP/1.1" 200 7601 "http://app.example.test/login" "python-requests/2.32.3"
192.168.10.77 - - [16/Oct/2026:09:11:41 +0900] "POST /static/style.css HTTP/1.1" 302 1816 "http://app.example.test/login" "python-requests/2.32.3"
2026/10/16 09:21:40 [error] 31#31: *7 open() "/usr/share/nginx/html/x" failed (2: No such file or directory)
192.168.10.77 - - [16/Oct/2026:09:12:48 +0900] "GET /api/items?page=1 HTTP/1.1" 200 3540 "-" "Mozilla/5.0 (X11; Linux x86_64) Chrome/126.0"
10.0.1.15 - - [16/Oct/2026:09:12:55 +0900] "GET /api/search?q=test HTTP/1.1" 200 1797 "http://app.example.test/login" "curl/8.5.0"
10.0.1.15 - - [16/Oct/2026:09:13:02 +0900] "PUT /index.html HTTP/1.1" 404 537 "-" "python-requests/2.32.3"
192.168.10.77 - - [16/Oct/2026:09:13:09 +0900] "GET /api/items/42 HTTP/1.1" 404 6086 "-" "Mozilla/5.0 (X11; Linux x86_64) Chrome/126.0"
192.168.10.5 - - [16/Oct/2026:09:14:16 +0900] "POST /admin HTTP/1.1" 500 8047 "-" "curl/8.5.0"
192.168.10.77 - - [16/Oct/2026:09:14:23 +0900] "GET /static/app.js HTTP/1.1" 304 0 "http://app.example.test/login" "curl/8.5.0"
192.168.10.5 - - [16/Oct/2026:09:15:30 +0900] "PUT / HTTP/1.1" 200 8774 "-" "Mozilla/5.0 (X11; Linux x86_64) Chrome/126.0"
192.168.10.5 - - 16/Oct/2026:09:22:10 "GET / HTTP/1.1" 200 612
10.20.0.8 - - [16/Oct/2026:09:15:37 +0900] "PUT /api/items/42 HTTP/1.1" 200 4398 "http://app.example.test/" "curl/8.5.0"
10.20.0.8 - - [16/Oct/2026:09:16:44 +0900] "GET /api/items?page=1 HTTP/1.1" 403 8993 "http://app.example.test/" "curl/8.5.0"
192.168.50.3 - - [16/Oct/2026:09:16:51 +0900] "PUT /api/items?page=1 HTTP/1.1" 200 6684 "-" "curl/8.5.0"
10.0.1.15 - - [16/Oct/2026:09:17:58 +0900] "PUT /admin HTTP/1.1" 404 594 "http://app.example.test/" "python-requests/2.32.3"
192.168.50.3 - - [16/Oct/2026:09:17:05 +0900] "GET /api/items?page=1 HTTP/1.1" 404 7447 "http://app.example.test/" "Mozilla/5.0 (Windows NT 10.0; Win64; x64) Firefox/128.0"
192.168.10.77 - - [16/Oct/2026:09:18:12 +0900] "GET /api/items?page=1 HTTP/1.1" 200 3836 "-" "Mozilla/5.0 (Windows NT 10.0; Win64; x64) Firefox/128.0"
192.168.50.3 - - [16/Oct/2026:09:18:19 +0900] "GET /admin HTTP/1.1" 200 7975 "http://app.example.test/" "Mozilla/5.0 (X11; Linux x86_64) Chrome/126.0"
--- logrotate marker ---
10.0.1.22 - - [16/Oct/2026:09:19:26 +0900] "GET /static/style.css HTTP/1.1" 200 7952 "http://app.example.test/" "Mozilla/5.0 (Windows NT 10.0; Win64; x64) Firefox/128.0"
192.168.50.3 - - [16/Oct/2026:09:19:33 +0900] "GET /static/style.css HTTP/1.1" 500 6696 "-" "curl/8.5.0"
LOG

섞은 5줄은 각각 형식이 아닌 한 줄, 상태 코드 자리에 ok가 들어간 줄, nginx 에러 로그 줄, 대괄호가 빠진 줄, 로그 회전 표시줄입니다.

Logstash를 실행할 스크립트도 하나 만듭니다. 파이프라인 폴더와 로그 폴더를 읽기 전용으로 마운트하고, Elasticsearch와 같은 네트워크에 붙입니다. 실행 로그는 error 수준만 보이게 했습니다. 마지막 줄의 exit "${PIPESTATUS[0]}"는 파이프 앞쪽인 Logstash의 종료 코드를 그대로 돌려줍니다. 뒤쪽 grep -v는 걸러 낼 줄이 없어 출력이 비면 1을 반환하는데, 그 값이 섞이지 않게 하려는 것입니다. PIPELINE_DIR 환경 변수를 주면 다른 파이프라인 폴더를 쓸 수 있고, 5단계에서 씁니다.

cat > run.sh <<'SH'
#!/bin/bash
# 사용법: ./run.sh [logstash 추가 인자]   (입력은 표준입력 또는 file input)
MSYS_NO_PATHCONV=1 docker run --rm -i --name cr-logstash-run --network cr-logstash-net \
  -v "${PIPELINE_DIR:-$PWD/pipeline}:/usr/share/logstash/pipeline:ro" \
  -v "$PWD/logs:/data/logs:ro" \
  -e XPACK_MONITORING_ENABLED=false -e LS_JAVA_OPTS="-Xms512m -Xmx512m" \
  docker.elastic.co/logstash/logstash:8.19.0 --log.level error "$@" 2>&1 \
  | grep -v -E "^(Using bundled|Sending Logstash|[0-9]{4}/[0-9]{2}/[0-9]{2} )"
exit "${PIPESTATUS[0]}"
SH
chmod +x run.sh

컨테이너와 클러스터 상태, 샘플 로그 줄 수는 아래 명령으로 확인합니다.

docker ps --filter name=cr-logstash --format 'table {{.Names}}\t{{.Status}}\t{{.Ports}}'
curl -s 'http://127.0.0.1:19400/_cluster/health?filter_path=status,number_of_nodes'; echo
wc -l logs/access.log

Elasticsearch 컨테이너가 127.0.0.1:19400 으로만 열려 있고 클러스터 상태가 응답하며, 샘플 로그가 45줄인지 확인합니다.
Elasticsearch 컨테이너가 127.0.0.1:19400 으로만 열려 있고 클러스터 상태가 응답하며, 샘플 로그가 45줄인지 확인합니다.

2. COMBINEDAPACHELOG로 파싱하고 ECS 필드 확인하기

nginx의 combined 형식은 Apache의 combined 형식과 같아서 grok 기본 패턴 COMBINEDAPACHELOG로 한 번에 읽습니다. 먼저 한 줄만 표준입력으로 넣어 결과를 눈으로 확인합니다. date 필터는 로그에 찍힌 시각을 @timestamp로 옮깁니다. 이 필터가 없으면 @timestamp는 Logstash가 줄을 읽은 시각이 됩니다.

cat > pipeline/nginx.conf <<'EOF'
input { stdin { } }

filter {
  grok {
    match => { "message" => "%{COMBINEDAPACHELOG}" }
    ecs_compatibility => v8
  }
  date {
    match => [ "timestamp", "dd/MMM/yyyy:HH:mm:ss Z" ]
    remove_field => [ "timestamp" ]
  }
}

output {
  stdout { codec => rubydebug }
}
EOF

sed -n 2p logs/access.log | ./run.sh

grok 이 http.request.method, source.address, url.original, http.response.status_code 같은 ECS 필드를 만들고, @timestamp 가 로그 시각을 UTC 로 바꾼 값(00:00:07)인지 확인합니다.
grok 이 http.request.method, source.address, url.original, http.response.status_code 같은 ECS 필드를 만들고, @timestamp 가 로그 시각을 UTC 로 바꾼 값(00:00:07)인지 확인합니다.

로그의 09:00:07 +0900이 @timestamp에는 UTC인 00:00:07로 저장됩니다. 상태 코드와 바이트 수는 숫자형입니다. 참조 URL이 -인 줄은 http.request.referrer 필드가 아예 만들어지지 않습니다. 원문은 event.original에 남습니다.

같은 줄을 ecs_compatibility => disabled로 파싱하면 필드 이름이 옛 방식으로 나옵니다. 같은 형식의 로그 첫 줄로 확인한 결과입니다.

값 disabled (옛 이름) v8 (ECS)
클라이언트 주소 clientip source.address
요청 메서드 verb http.request.method
요청 경로 request url.original
HTTP 버전 httpversion http.version
상태 코드 response (문자열) http.response.status_code (숫자)
응답 바이트 bytes (문자열) http.response.body.bytes (숫자)
참조 URL referrer (따옴표 포함) http.request.referrer
User-Agent agent (따옴표 포함) user_agent.original
로그 시각 timestamp timestamp (v8에서도 그대로, date로 처리)

disabled로 바꿔 같은 줄을 다시 돌리면 아래처럼 나옵니다(host와 @version 등 나머지 필드는 줄였습니다).

{
       "clientip" => "10.0.1.22",
           "verb" => "GET",
        "request" => "/api/search?q=test",
    "httpversion" => "1.1",
       "response" => "200",
          "bytes" => "8433",
       "referrer" => "\"-\"",
          "agent" => "\"Mozilla/5.0 (X11; Linux x86_64) Chrome/126.0\"",
     "@timestamp" => 2026-10-16T00:00:07.000Z
}

disabled일 때는 따옴표가 값에 그대로 들어가고 숫자도 문자열이라, 상태 코드 범위 검색이나 바이트 합계를 내려면 따로 변환해야 합니다. run.sh에서 --log.level error를 빼고 실행하면 Pipeline main is configured with pipeline.ecs_compatibility: v8 라는 정보 로그가 나오는데, 파이프라인 전체가 v8 기본값으로 도는 중이라는 뜻입니다. 같은 조건에서 grok이 ECS v8 support is a preview of the unreleased ECS v8, and uses the v1 patterns 경고도 냅니다. 두 메시지 모두 8.19.0에서 다시 실행해 확인했고 동작에는 영향이 없었습니다.

3. 직접 쓴 패턴과 비교하기

기본 패턴을 쓰지 않고 pattern_definitions에 패턴을 직접 정의할 수도 있습니다. 필드 이름을 대괄호 표기([http][request][method])로 직접 지정하고 숫자는 :int로 변환합니다. 이 방식에서는 ecs_compatibility 설정과 상관없이 내가 쓴 이름이 그대로 쓰입니다.

cat > pipeline/nginx.conf <<'EOF'
input { stdin { } }

filter {
  grok {
    pattern_definitions => {
      "NGINX_ACCESS" => '%{IPORHOST:[source][address]} - %{DATA:[user][name]} \[%{HTTPDATE:log_time}\] "%{WORD:[http][request][method]} %{NOTSPACE:[url][original]} HTTP/%{NUMBER:[http][version]}" %{NUMBER:[http][response][status_code]:int} %{NUMBER:[http][response][body][bytes]:int} "%{DATA:[http][request][referrer]}" "%{DATA:[user_agent][original]}"'
    }
    match => { "message" => "%{NGINX_ACCESS}" }
  }
  date {
    match => [ "log_time", "dd/MMM/yyyy:HH:mm:ss Z" ]
    remove_field => [ "log_time" ]
  }
}

output {
  stdout { codec => rubydebug }
}
EOF

sed -n 2p logs/access.log | ./run.sh

직접 쓴 NGINX_ACCESS 패턴도 같은 ECS 필드를 만들고, 값이 없는 자리는 user.name 과 http.request.referrer 에 - 문자가 그대로 남는지 확인합니다.
직접 쓴 NGINX_ACCESS 패턴도 같은 ECS 필드를 만들고, 값이 없는 자리는 user.name 과 http.request.referrer 에 - 문자가 그대로 남는지 확인합니다.

두 방식의 차이는 두 가지입니다. 직접 쓴 패턴은 -를 값으로 그대로 저장하고(user.name, http.request.referrer), 기본 패턴은 해당 필드를 만들지 않습니다. 또 기본 패턴은 요청 줄 형식이 GET /path HTTP/1.1로 정해져 있지만, 직접 쓴 패턴은 내 로그에 맞춰 얼마든지 바꿀 수 있습니다. 로그 형식이 combined에서 벗어나면 기본 패턴은 쓸 수 없으므로 그때는 직접 쓴 패턴으로 바꿉니다. 이후 단계는 기본 패턴 COMBINEDAPACHELOG로 진행합니다.

4. 전체 로그를 읽어 Elasticsearch로 보내기

이제 file 입력으로 샘플 로그 전체를 읽어 nginx-all 인덱스에 넣습니다. mode => "read"와 exit_after_read => true를 쓰면 파일을 끝까지 읽은 뒤 Logstash가 스스로 종료해서 실습에 편합니다. sincedb_path => "/dev/null"은 Linux 컨테이너에서 읽은 위치를 저장하지 않게 하려고 흔히 쓰는 방법으로, 몇 번을 다시 돌려도 처음부터 읽습니다. 그만큼 다시 돌릴 때마다 같은 문서가 인덱스에 또 들어갑니다. file_completed_action의 기본값은 delete인데 로그 폴더를 :ro로 마운트했으므로, 다 읽은 파일을 지우지 않고 log로 기록만 남기게 했습니다.

cat > pipeline/nginx.conf <<'EOF'
input {
  file {
    path => "/data/logs/access.log"
    mode => "read"
    sincedb_path => "/dev/null"
    exit_after_read => true
    file_completed_action => "log"
    file_completed_log_path => "/tmp/completed.log"
  }
}

filter {
  grok {
    match => { "message" => "%{COMBINEDAPACHELOG}" }
    ecs_compatibility => v8
  }
  date {
    match => [ "timestamp", "dd/MMM/yyyy:HH:mm:ss Z" ]
    remove_field => [ "timestamp" ]
  }
}

output {
  elasticsearch {
    hosts => ["http://cr-logstash-es:9200"]
    index => "nginx-all"
  }
}
EOF

./run.sh </dev/null

15초 안팎으로 끝나며 화면에는 아무것도 출력되지 않는 것이 정상이고, 이어서 echo $?를 실행하면 0이 나옵니다. 이어서 파싱에 실패한 문서 수를 셉니다. grok이 맞지 않으면 tags 필드에 _grokparsefailure가 붙습니다.

# 전체 문서 수
curl -s 'http://127.0.0.1:19400/nginx-all/_count?filter_path=count'; echo
# 파싱 실패 문서 수
curl -s 'http://127.0.0.1:19400/nginx-all/_count?q=tags:_grokparsefailure&filter_path=count'; echo
# 한 번에 두 값
curl -s 'http://127.0.0.1:19400/nginx-all/_search?size=0&filter_path=aggregations.*.buckets' \
  -H 'Content-Type: application/json' \
  -d '{"aggs":{"parse":{"filters":{"filters":{"failed":{"match":{"tags":"_grokparsefailure"}},"parsed":{"bool":{"must_not":{"match":{"tags":"_grokparsefailure"}}}}}}}}}'; echo

기대 출력입니다.

{"count":45}
{"count":5}
{"aggregations":{"parse":{"buckets":{"failed":{"doc_count":5},"parsed":{"doc_count":40}}}}}

45줄 가운데 5줄이 실패했으므로 실패 비율은 약 11.1%입니다. 운영 로그라면 이 비율을 주기적으로 확인하다가, 평소보다 오르면 앱이나 nginx의 로그 형식이 바뀌었는지 의심합니다.

전체 45건 가운데 _grokparsefailure 태그가 붙은 문서가 5건, 정상 파싱이 40건으로 나오는지 확인합니다.
전체 45건 가운데 _grokparsefailure 태그가 붙은 문서가 5건, 정상 파싱이 40건으로 나오는지 확인합니다.

5. 파싱 실패 로그를 별도 인덱스로 보내기

실패한 줄이 정상 로그와 같은 인덱스에 섞이면 대시보드 집계가 흐려집니다. output에서 tags에 _grokparsefailure가 있는지 조건으로 가르면 인덱스를 나눌 수 있습니다. 위 4단계에서 만든 nginx-all은 그대로 두고, 이번에는 nginx-access와 nginx-failed로 나눠 보냅니다.

cat > pipeline/nginx.conf <<'EOF'
input {
  file {
    path => "/data/logs/access.log"
    mode => "read"
    sincedb_path => "/dev/null"
    exit_after_read => true
    file_completed_action => "log"
    file_completed_log_path => "/tmp/completed.log"
  }
}

filter {
  grok {
    match => { "message" => "%{COMBINEDAPACHELOG}" }
    ecs_compatibility => v8
  }
  date {
    match => [ "timestamp", "dd/MMM/yyyy:HH:mm:ss Z" ]
    remove_field => [ "timestamp" ]
  }
}

output {
  if "_grokparsefailure" in [tags] {
    elasticsearch {
      hosts => ["http://cr-logstash-es:9200"]
      index => "nginx-failed"
    }
  } else {
    elasticsearch {
      hosts => ["http://cr-logstash-es:9200"]
      index => "nginx-access"
    }
  }
}
EOF

바로 돌리지 말고 문법부터 점검합니다. --config.test_and_exit는 설정 파일을 읽어 문법만 검사하고 종료합니다. 파이프라인은 실행하지 않으므로 인덱스가 생기지 않습니다.

./run.sh --config.test_and_exit </dev/null

Configuration OK가 나오면 통과입니다. 틀리면 줄과 열 위치가 나오고 종료 코드가 1입니다. 예를 들어 filter 뒤의 {를 지운 사본을 만들어 점검하면 아래처럼 보입니다. 사본은 bad 폴더에 만들고 PIPELINE_DIR로 지정합니다.

mkdir -p bad && sed 's/^filter {/filter/' pipeline/nginx.conf > bad/nginx.conf
PIPELINE_DIR=$PWD/bad ./run.sh --config.test_and_exit </dev/null; echo "exit=$?"

설정이 맞으면 Configuration OK 와 exit=0 이, filter 뒤의 중괄호를 지운 사본은 line 13 위치를 알려 주는 오류와 exit=1 이 나오는지 확인합니다.
설정이 맞으면 Configuration OK 와 exit=0 이, filter 뒤의 중괄호를 지운 사본은 line 13 위치를 알려 주는 오류와 exit=1 이 나오는지 확인합니다.

문법을 통과했으니 실행하고 인덱스별 문서 수와 실패한 원문을 확인합니다.

./run.sh </dev/null
curl -s 'http://127.0.0.1:19400/_cat/indices/nginx-*?v&h=index,docs.count&s=index'
curl -s 'http://127.0.0.1:19400/nginx-failed/_search?size=5&filter_path=hits.hits._source.message' | sed 's/},{/}\n{/g'; echo

nginx-access에 40건, nginx-failed에 5건이 들어가고, 실패 인덱스에는 일부러 섞은 5줄의 원문이 그대로 남습니다. nginx-all의 45건은 4단계에서 만든 것입니다.

nginx-access 에 40건, nginx-failed 에 5건이 들어가고 실패 인덱스에 섞어 넣은 5줄의 원문이 보이는지 확인합니다.
nginx-access 에 40건, nginx-failed 에 5건이 들어가고 실패 인덱스에 섞어 넣은 5줄의 원문이 보이는지 확인합니다.

정리하기

실습이 끝나면 컨테이너, 네트워크, 볼륨, 작업 폴더를 지웁니다. 이미지는 다른 용도로 쓸 수 있어 남겨 둡니다.

docker rm -f cr-logstash-es
docker network rm cr-logstash-net
docker volume rm cr-logstash-esdata
cd ~
rm -r ~/grok-demo

자주 막히는 곳

Windows Git Bash에서 컨테이너 안쪽 경로가 C:/Program Files/Git/... 로 바뀝니다. Git Bash가 /data/logs 같은 슬래시로 시작하는 인자를 Windows 경로로 바꾸기 때문입니다. 컨테이너에서 ls /data/logs를 실행하면 C:/Program Files/Git/data/logs: No such file or directory가 나옵니다. 명령 앞에 MSYS_NO_PATHCONV=1을 붙이면 막을 수 있고, 이 글의 run.sh에는 이미 들어 있습니다. 볼륨 마운트는 -v "$PWD/pipeline:/usr/share/logstash/pipeline:ro"처럼 호스트 쪽을 $PWD로 쓰면 동작했습니다.

필드 이름이 clientip, verb, response 로 나옵니다. grok에 ecs_compatibility => disabled가 지정됐거나 파이프라인이 옛 설정으로 도는 경우입니다. 글의 비교표처럼 이 값에서는 숫자도 문자열로 저장됩니다. 필터에 ecs_compatibility => v8을 명시하거나 pipeline.ecs_compatibility 설정을 확인합니다.

@timestamp가 로그 시각이 아니라 지금 시각입니다. date 필터가 없거나 형식이 맞지 않으면 @timestamp는 Logstash가 줄을 읽은 시각입니다. nginx 시각은 dd/MMM/yyyy:HH:mm:ss Z 형식이고 MMM은 영어 월 이름(Oct)이어야 합니다.

The given configuration is invalid 가 나옵니다. 메시지의 line과 column이 가리키는 위치 바로 앞을 봅니다. 중괄호나 =>가 빠진 경우가 대부분이며, 위 단계의 오류 화면이 그 예입니다. run.sh는 Logstash의 종료 코드를 그대로 돌려주므로, 오류이면 1, 정상이면 출력이 없어도 0이어서 스크립트나 CI에서 성공과 실패를 구분하는 데 쓸 수 있습니다.

mode와 exit_after_read 없이 file 입력을 쓰면 Logstash가 끝나지 않습니다. 기본값은 파일 끝에서 새 줄을 계속 기다리는 방식입니다. 이 글처럼 한 번 읽고 끝내려면 mode => "read"와 exit_after_read => true를 함께 지정하고, 계속 감시하려면 두 옵션을 빼고 docker rm -f cr-logstash-run으로 멈춥니다. 이때 mode의 기본값은 tail이고 start_position의 기본값은 end라서, tail 모드에서는 start_position => "beginning"을 함께 지정해야 기존 줄을 읽습니다. 지정하지 않으면 이미 있는 45줄은 읽지 않고 새로 추가되는 줄만 읽습니다. 직접 실행해 보니 지정하지 않았을 때는 인덱스가 만들어지지 않았고(0건), 지정했을 때는 45건이 들어갔습니다.

이어서 해 볼 것

  • geoip와 useragent 필터 추가: 비공개 대역은 위치가 나오지 않으니 테스트용 공개 주소로 source.geo가 채워지는지 확인합니다. user_agent.original을 useragent 필터로 분해하면 브라우저와 OS 필드를 얻습니다.
  • dissect와 비교: 형식이 고정된 로그는 정규식을 쓰는 grok보다 dissect가 가볍습니다. 같은 로그를 dissect로 파싱해 보고 처리 속도와 실패 처리 방식을 비교합니다.
  • 인덱스 템플릿 지정: 지금은 Elasticsearch가 필드 타입을 자동으로 추정합니다. source.address를 ip 타입으로 고정하는 템플릿을 만들어 두고 같은 로그를 다시 넣어 봅니다.
  • Kibana에서 보기: Kibana 설치하고 Elasticsearch에 연결하기를 참고해 Kibana를 붙이고 nginx-access 데이터 뷰를 만들어 상태 코드별 건수를 시각화합니다. 이 글의 Elasticsearch는 보안을 끈 구성이므로 Kibana도 로컬 전용으로만 띄웁니다.

확인 범위

직접 실행한 부분은 이 글의 1단계부터 5단계까지 전부이고, 침묵 실행과 오류 실행의 종료 코드(0과 1), 계속 감시 방식의 start_position 동작도 확인했습니다. file 입력의 mode, start_position, file_completed_action 기본값은 Logstash 8.19 file 입력 문서에서 확인했습니다. sincedb_path를 /dev/null로 두는 방법은 공식 문서에 없는 관용입니다. 환경은 Windows 11의 Docker Desktop(Docker 29.5.3)과 Git Bash, Elasticsearch 8.19.0, Logstash 8.19.0이며, 본문의 출력과 개수는 그 환경에서 나온 값입니다. 컨테이너 ID처럼 실행마다 달라지는 값은 다를 수 있습니다.

Linux나 macOS 호스트, Elasticsearch 클러스터 구성, 보안을 켠 환경(TLS와 인증)에서는 실행하지 못했습니다. Kibana 연동과 이어서 해 볼 것의 항목도 실행 전이므로 적용하기 전에 직접 돌려 확인하시기 바랍니다. grok의 ecs_compatibility 기본 동작은 Logstash 8.19 문서의 grok 필터와 ECS compatibility 설명을 기준으로 확인했고, COMBINEDAPACHELOG가 만드는 v8 필드 이름은 문서에 목록이 없어 실제 출력으로 확인했습니다.

이 글의 다음 단계

  • Kibana 대시보드와 GeoIP 지도로 접속 현황 보기Docker로 Elasticsearch 8.19와 Kibana 8.19를 띄우고, ingest pipeline의 geoip 프로세서로 접근 로그의 IP에 위치 정보를 붙입니다. 상위 IP, 상태 코드, 시간대별 요청 수, 지도 패널을 대시보드로 저장하고 저장 객체를 내보내 다시 가져옵니다.
  • Sigma 룰을 작성하고 Elasticsearch 쿼리로 변환해 탐지하기웹 서버 접근 로그용 Sigma 룰 3개를 작성해 sigma-cli로 점검하고 Lucene 쿼리로 변환합니다. 직접 만든 pipeline으로 필드명을 ECS 이름으로 바꾼 뒤 Elasticsearch 8.19와 Kibana Discover에서 탐지되는지 확인하고, 오탐을 줄이는 조건도 더합니다.

SIEM 관제 실습 (ELK) 과정의 일부입니다

이 실습은 교육에서 참가자마다 독립된 환경으로 진행합니다. 원하는 일정을 고르면 문의 양식에 미리 채워집니다.

과정 보기

교육 진행 순서와 자주 묻는 질문 보기

같은 카테고리의 글