Logstach - dissect, grok 필터 예제

Logstash 개요

Logstash는 플러그인 기반의 데이터 수집 및 처리 엔진으로서, 

광범위한 플러그인이 구비되어 다양한 아키텍처에서 데이터를 수집, 처리, 전달할 수 있게 해줍니다.

 

크게 input, filter, output으로 구성되어 있으며,

프로세싱 스레드는 대기열에서 데이터를 마이크로 배치로 읽은 다음,

구성된 필터 플러그인을 통해 순서대로 처리합니다.

 

 

다운로드

# 아래 주소에서 다운로드 가능
https://www.elastic.co/kr/downloads/logstash

# 다운로드
wget https://artifacts.elastic.co/downloads/logstash/logstash-8.16.0-darwin-aarch64.tar.gz

# 압축풀기
tar xfz logstash-8.16.0-darwin-aarch64.tar.gz

 

 

첫번째 예제

Logstash 구성에는 적어도 하나의 입력 플러그인과 하나의 출력 플러그인이 있어야 합니다. 

필터는 선택 사항입니다. 

 

파일에서 테스트 데이터 집합을 읽고 이를 구조화된 형식으로 콘솔에 출력하겠습니다.

구성 파일이 test.conf라고 하고 테스트 데이터가 들어있는 파일이 “/home/logstash” 디렉토리에 저장됩니다.

[ test.conf ]

input {
 file {
   path => ["/opt/homebrew/tarLib/logstash-8.16.0/data/testdata.log"]
   sincedb_path => "/dev/null"
   start_position => "beginning"
  }
}
filter {
}
output {
  stdout {
    codec => rubydebug
  }
}

file {...} : 입력 섹션에서 파일 입력 플러그인을 지정

path => : 경로 지시문을 통해 테스트 데이터 파일의 경로를 제공

start_position => “beginning” : 새 파일이 발견될 때마다 처음부터 파일을 읽도록 플러그인에 지시

stdout : 표준 출력

codec=>rubydebug : rubydebug 형식은 각 필드를 JSON과 유사한 구조

 

[ testdata.log 파일 ]

cat testdata.log
>>> Hellow Logstash !!?

 

[ Logstash를 실행 ]

bin/logstash -r -f data/test.conf

-r (reload):

파일 변경 사항 자동 재로드 옵션으로, 지정된 설정 파일(여기서는 test.conf)을 주기적으로 모니터링합니다.

파일이 수정되면 Logstash는 재시작하지 않고, 변경된 설정을 반영하여 작업을 재로드합니다.

이 옵션은 구성 파일을 자주 수정하거나 실시간으로 변경 사항을 반영해야 하는 경우 유용합니다.

-f (config file):

설정 파일 경로를 지정하는 옵션입니다.

 

아래는 Logstash의 출력 값 입니다.

[ 출력 예제 ]

# 출력 값
{
    "log" => {
        "file" => {
            "path" => "/opt/homebrew/tarLib/logstash-8.16.0/data/testdata.log"
        }
    },
    "@timestamp" => 2024-11-13T13:18:07.959188Z,
    "event" => {
        "original" => "Hellow Logstash !!?"
    },
    "@version" => "1",
    "message" => "Hellow Logstash !!?",
    "host" => {
        "name" => "AL123123.local"
    }
}

event.original : 원본 로그 메시지를 나타냅니다.

@version : Logstash에서 자동으로 추가된 필드로, 이벤트의 버전 정보를 의미합니다.

message : 수집된 로그의 실제 내용입니다.

 

 

dissect(해부)로 로그 분석

Logstash에서 dissect 필터는 간단한 문자열 분해를 통해 로그 데이터를 파싱할 때 사용됩니다. 

dissect 필터는 특정 구분 문자를 기준으로 로그 메시지를 나누어 각 필드를 추출합니다.

 

예를 들어, 다음과 같은 로그 데이터가 있습니다.

[ 예제 로그 데이터 ]

2024-11-13 15:30:01 INFO user1 Login successful

위 로그 메시지를 날짜, 시간, 로그 수준, 사용자 ID, 메시지로 나누어 분석해보겠습니다.

 

다음은 dissect 필터를 사용하여 위 로그 메시지를 파싱하는 Logstash 구성 파일입니다.

[ test.conf ]

input {
  file {
    path => ["/opt/homebrew/tarLib/logstash-8.16.0/data/testdata.log"]
    sincedb_path => "/dev/null"
    start_position => "beginning"
   }
}

filter {
  dissect {
    # 파싱할 필드 시퀀스와 구분 문자 지정
    mapping => { 
      "message" => "%{timestamp} %{time} %{log_level} %{user} %{message_content}" 
    }
  }
}

output {
  stdout { codec => rubydebug }
}

mapping 옵션: dissect 필터의 핵심으로, 추출할 필드 시퀀스와 구분 문자를 정의합니다.

%{timestamp}: 첫 번째 필드로, 날짜 값을 추출합니다.

%{time}: 두 번째 필드로, 시간 값을 추출합니다.

%{log_level}: 세 번째 필드로, 로그 수준을 추출합니다.

%{user}: 네 번째 필드로, 사용자 ID를 추출합니다.

%{message_content}: 마지막 필드로, 로그 메시지 내용을 추출합니다.

[ 출력 값 ]

bin/logstash -r -f data/test.conf

# 출력 값
{
  "host" => {
    	"name" => "AL0123123.local"
  },
  "log" => {
        "file" => {
            "path" => "/opt/homebrew/tarLib/logstash-8.16.0/data/testdata.log"
        }
    },
    "log_level" => "INFO",
    "message" => "2024-11-13 15:30:01 INFO user1 Login successful",
    "@version" => "1",
    "@timestamp" => 2024-11-13T13:43:26.185107Z,
    "event" => {
        "original" => "2024-11-13 15:30:01 INFO user1 Login successful"
    },
    "user" => "user1",
    "timestamp" => "2024-11-13",
    "time" => "15:30:01",
    "message_content" => "Login successful"
}

 

 

grok으로 로그 분석

Grok filter는 Logstash에서 로그 데이터를 구조화된 형태로 파싱할 때 자주 사용됩니다.

Grok 필터를 사용하여 로그 데이터를 정규 표현식 패턴으로 나누고, 필요한 필드를 추출할 수 있습니다.

 

아래는 Apache 액세스 로그를 Grok 필터로 파싱하는 예제입니다.

[ 예제 로그 데이터 ]

127.0.0.1 - frank [10/Oct/2024:13:55:36 -0700] "GET /index.html HTTP/1.1" 200 1043

 

Logstash Grok 필터 설정 예제는 아래와 같습니다.

[ logstash.conf ]

input {
  stdin {...}
}

filter {
  grok {
    match => { 
      "message" => "%{IP:client_ip} - %{WORD:user} \[%{HTTPDATE:timestamp}\] \"%{WORD:method} %{DATA:request} HTTP/%{NUMBER:http_version}\" %{NUMBER:status} %{NUMBER:bytes}" 
    }
  }
}

output {
  stdout { codec => rubydebug }
}

%{IP:client_ip}: IP 주소를 client_ip 필드로 추출합니다.

%{WORD:user}: 사용자를 user 필드로 추출합니다.

%{HTTPDATE:timestamp}: 날짜와 시간을 timestamp 필드로 추출합니다.

%{WORD:method}: HTTP 메서드(GET, POST 등)를 method 필드로 추출합니다.

%{DATA:request}: 요청된 리소스를 request 필드로 추출합니다.

%{NUMBER:http_version}: HTTP 버전을 http_version 필드로 추출합니다.

%{NUMBER:status}: 응답 상태 코드를 status 필드로 추출합니다.

%{NUMBER:bytes}: 응답 바이트 수를 bytes 필드로 추출합니다.

 

[ 예제 실행 결과 ]

{
    "client_ip": "127.0.0.1",
    "user": "frank",
    "timestamp": "10/Oct/2024:13:55:36 -0700",
    "method": "GET",
    "request": "/index.html",
    "http_version": "1.1",
    "status": "200",
    "bytes": "1043",
    "message": "127.0.0.1 - frank [10/Oct/2024:13:55:36 -0700] \"GET /index.html HTTP/1.1\" 200 1043",
    "@version": "1",
    "@timestamp": "2024-10-10T20:55:36.000Z",
    "host": "your_host_name"
}

 

참조

공식 문서 : https://www.elastic.co/guide/en/logstash/current/introduction.html

간단 실습 : https://www.elastic.co/kr/blog/a-practical-introduction-to-logstash