ELK 스택에서 데이터 집계/전달을 담당하는 Logstash에 관한 포스트입니다. 본 게시글에서는 설치과정부터 기본 사용법, 활용례 등을 소개하고자 합니다.
1. Logstash 설치
https://www.elastic.co/kr/downloads/logstash
Download Logstash Free | Get Started Now
Download Logstash or the complete Elastic Stack (formerly ELK stack) for free and start collecting, searching, and analyzing your data with Elastic in minutes.
www.elastic.co
가장 최신 버전인 8.8.2 버전(2023년 6월 기준)으로 설치를 진행합니다. 바이너리 버전 압축파일을 아래와 같이 다운받아 압축 해제합니다.
$ wget https://artifacts.elastic.co/downloads/logstash/logstash-8.8.2-linux-x86_64.tar.gz
$ tar -xvzf logstash-8.8.2-linux-x86_64.tar.gz
생성된 logstash 디렉토리 안에 실행도구부터 각종 유틸리티가 있으나, 사용자 계정 어느 Path에서든지 사용할 수 있도록 환경변수를 추가해줍니다.
$ vi .bashrc
# .bashrc 마지막 줄에 아래 내용 추가
LS_HOME="/home/pyro/logstash-8.8.2"
PATH="$LS_HOME/bin:$PATH"
수정 내역을 업데이트하고, logstash -V 명령어로 logstash가 잘 동작하는지 확인합니다.
$ source .bashrc
$ logstash -V
2. Logstash 기본 사용법
ELK 스택에서 각 요소는 저마다 맡는 역할이 있습니다. Elasticsearch가 데이터베이스나 다름 없으므로, Elasticsearch보다 앞단에서 동작하는 Logstash는 DB에 데이터를 밀어 넣어주는 작업을 맡게됩니다.
Logstash는 자체적으로 데이터를 생산해내지 않습니다. 주변(앱, 시스템 등)에서 만들어진 데이터를 Logstash가 한번에 모아서 전달하는 것입니다.
택배를 받는 과정에 빗대면, 대규모 물류센터에서 화물을 분류해내고 이를 지정된 목적지까지 배송하는 작업이 Logstash의 역할에 딱 들어맞는 설명이라고 할 수 있죠. 각 택배는 송신지(ex. 쇼핑몰, 매장), 목적지, 그리고 포장 형태가 다 다르며 이러한 정보는 필히 관리되어야 올바르게 운송이 이뤄질 수 있습니다. 이와 유사하게 Logstash의 파이프라인은 Input / Filter / Output 이라는 설정 정보를 갖추고 있습니다.
- Input: 데이터를 생성하여 제공하는 대상입니다. (ex. stdin, file, syslog, 그 외에 beats와 같은 외부 플러그인)
- Filter: Input에서 수집한 데이터를 가공/변형(ex. 구문 분석 및 변환)합니다. 다시 택배의 예시를 들면, 깨지기 쉬운 물건은 상자 안에 뽁뽁이(?)를 채우고 냉동식품은 아이스박스 안에 넣는 등의 포장 작업을 떠올려보시면 좋습니다.
- Output: 수집된 데이터가 가야할 목적지를 명시합니다. 주로 Elasticsearch, Graphite와 같은 DB로 보낼 수도 있고, 다른 file에 output을 직접 기록할 수도 있습니다.
앞서 설명한 Input / Filter / Output 3가지 요소를 conf 파일에 작성하여, logstash를 실행하면서 해당 conf 파일을 읽어 데이터 처리 작업을 이행하게 됩니다.
3. Input/Output 설정 예시
CLI Input/Output
커맨드라인 입력을 데이터 소스로 씀과 동시에, 이를 커맨드라인에 그대로 출력하는 경우입니다. 아래와 같이 1줄의 명령어로도 실행이 가능하며,
$ logstash -e 'input { stdin { } } output { stdout {} }'
직접 conf 파일을 만들어서 logstash 실행 시 -f 옵션을 주어 conf 파일 내 설정을 적용할 수 있습니다.
$ vi logstash-keyboardIO.conf
# logstash-keyboardIO.conf
input {
stdin { }
}
output {
stdout { }
}$ logstash -f ~/logstash-keyboardIO.conf
File Read/Write
특정 경로에 있는 파일을 읽고 그 내용을 다른 파일에 기록하는 예시입니다. input.log라는 파일에 메시지를 미리 담아놓고 이를 output.log 파일로 전달합니다.
# logstash-fileIO.conf
input {
file {
path => "/home/pyro/input.log" # 읽어들일 파일의 경로
start_position => "beginning" # 파일의 어디부터 읽을지
sincedb_path => "/dev/null" # 처리 내역 로그를 내보내는 곳
}
}
output {
file {
path => "/home/pyro/output.log"
}
}
# input.log
Hello there!
아래 명령어를 수행하고나서, 홈 디렉토리에 생성된 output.log 파일을 열어보면 input.log 에 있던 내용이 그대로 담긴 것을 확인할 수 있습니다.
$ logstash -f ~/logstash-fileIO.conf
ElasticSearch
결과물을 이제 로컬 파일시스템말고 DB나 서버에 전송해봅니다. Logstash와 궁합이 맞는 ElasticSearch 서비스를 띄워보고 해당 엔진에 데이터를 전달합니다.
Input 데이터로는 10줄 가량의 시스템 로그를 긁어모아 활용해보았습니다.
# input.log
2023-01-19 01:34:48,575 - main.py[DEBUG]: No kernel command line url found.
2023-01-19 01:34:48,575 - main.py[DEBUG]: Closing stdin.
2023-01-19 01:34:48,576 - util.py[DEBUG]: Writing to /var/log/cloud-init.log - ab: [644] 0 bytes
2023-01-19 01:34:48,577 - util.py[DEBUG]: Changing the ownership of /var/log/cloud-init.log to 102:4
2023-01-19 01:34:48,577 - util.py[DEBUG]: Attempting to remove /var/lib/cloud/instance/boot-finished
2023-01-19 01:34:48,577 - util.py[DEBUG]: Attempting to remove /var/lib/cloud/data/no-net
2023-01-19 01:34:48,577 - handlers.py[DEBUG]: start: init-local/check-cache: attempting to read from cache [check]
2023-01-19 01:34:48,577 - util.py[DEBUG]: Reading from /var/lib/cloud/instance/obj.pkl (quiet=False)
2023-01-19 01:34:48,578 - util.py[DEBUG]: Read 12128 bytes from /var/lib/cloud/instance/obj.pkl
데이터가 저장되려면 우선 elasticsearch 서비스를 활성화시켜야 합니다. 본 포스트를 포함하여 이후 글에서도 elasticsearch가 기본적으로 세팅되어 있다고 가정하고 내용을 진행할 예정이며, 만약 elasticsearch 설치가 되지 않으신 분은 https://citizen.tistory.com/35 을 참고해주세요.
$ elasticsearch -d
conf 파일 내용은 아래와 같습니다. 현재 elasticsearch는 logstash와 동일한 로컬 환경에서 동작하고 있고, input.log 파일의 결과물을 mylog라는 인덱스에 저장하고자 합니다.
# logstash-elastic.conf
input {
file {
path => "/home/pyro/input.log"
start_position => "beginning"
sincedb_path => "/dev/null"
}
}
output {
elasticsearch {
hosts => "localhost"
index => "mylog"
}
}
Logstash를 실행한 뒤, elasticsearch에 API로 mylog 인덱스를 조회해보면 input.log 안에 있던 시스템 로그 내용이 기록되어 있음을 확인 가능합니다.
$ logstash -f ~/logstash-elastic.conf
# 터미널 하나를 더 띄워서
$ curl -XGET http://localhost:9200/mylog/_search?pretty
기타 여러 특징들
앞서 예시에서는 Input, Output 요소를 각각 하나씩만 배정하였습니다. Input/Output은 하나 이상이 될 수 있으며, 아래와 같이 복수의 Output 요소를 추가하면 데이터가 병렬적으로 전달됩니다.
$ logstash -f ~/logstash-fileIO.conf
또한, Logstash는 실시간으로 동작하므로 데이터 Source에 새로운 내용이 들어오면 그걸 바로바로 처리해줍니다.
$ logstash -f ~/logstash-fileToStdout.conf
$ echo "new log detected" >> input.log
4. Filter
데이터를 내보내기 직전, 가공 단계를 거칠 수 있습니다. 데이터 형변환, 문자열 처리, 날짜 표시 형식 변경 등이 가능합니다. 실제 있는 데이터를 얻어 예제로 활용하였습니다.
https://www.kaggle.com/datasets/prasoonkottarathil/btcinusd
샘플로 사용하는 csv 데이터는 총 9개의 colume이 존재하며, 주어진 데이터에서 아래와 같은 사항들을 적용해볼 것입니다.
- 필요없는 column 제거:
unix,symbol,volume USD - 실수값 → 정수값:
open,high,low,close - 필드이름 변경:
volume BTC→volume - 날짜에서 시/분/초 제거:
date
# logstash-filter.conf
input {
file {
path => "/home/pyro/BTC-Daily.csv"
start_position => "beginning"
sincedb_path => "/dev/null"
}
}
filter {
# csv 파일 읽기
csv {
separator => ","
columns => ["unix", "date", "symbol", "open", "high", "low", "close", "volume BTC", "volume USD"]
}
mutate {gsub => ["date", " 00:00:00", '']} # 시/분/초 제거
mutate {convert => ["open", "integer"]} # 실수 -> 정수
mutate {convert => ["high", "integer"]}
mutate {convert => ["low", "integer"]}
mutate {convert => ["close", "integer"]}
mutate {remove_field => ["unix", "symbol", "volume USD"]} # 필드 제거
mutate {rename => ["volume BTC", "volume"]} # 필드명 변경
}
output {
stdout {}
}
Logstash를 실행하면 군데군데 필드 정보들이 변경된 것을 볼 수 있습니다.
$ logstash -f ~/logstash-filter.conf