https://download.elastic.co/demos/logstash/gettingstarted/logstash-tutorial.log.gz
$ wget https://download.elastic.co/demos/logstash/gettingstarted/logstash-tutorial.log.gz
$ gzip -d logstash-tutorial.log.gz
먼저 filebeat 가 샘플로그를 바라볼수있도록 설정을 변경합니다.
$ sudo vi /etc/filebeat/filebeat.yml
# ============================== Filebeat inputs ===============================
filebeat.inputs:
- type: log
enabled: true
paths:
- /home/elktest/log/
....
....
....
# ------------------------------ Logstash Output -------------------------------
output.logstash:
# The Logstash hosts
hosts: ["localhost:5044"]
/home/elktest/log 경로는 제가 샘플로그 파일을 저장한 경로이므로 본인이 로그파일을 받은곳으로 위치를 지정하면 됩니다.
만약 이전에 filebeat 를 설치후 실행하다가 Output 옵션에 output.elasticsearch 부분이 주석이 해제되있다면 해당부분은 주석처리 하면됩니다. 지금은 elasticsearch로 바로 보내는게 아니라 Logstash 로 전송해야 하기 때문입니다.
filebeat.yml 파일의 변경 내용을 적용해야 하므로 filebeat 데몬이 실행중이라면 재실행 해야 합니다.
$ sudo systemctl restart filebeat
이제 filebeat 에서 Logstash 로 데이터 전송할 준비가 다 되었으니 Logstash 에서도 데이터를 받을 작업을 해야합니다.
이전에 Logstash 설치 및 실행 포스팅에서는 실행과 동시에 input / output 에 대한 내용을 입력하여 테스트를 했었습니다.
그런데 더 구체적인 필터와 여러가지 input / output 플러그인을 사용하려면 파일로 작성하여 실행할때 파일을 불러오는방법이 더 효과적일 것입니다.
$ vi beat-pipeline.conf
input {
beats {
port => "5044"
}
}
filter {
}
output {
stdout {
codec => rubydebug
}
}
input / output 영역을 가지고 있는 형태가 Logstash 에 불러올수있는 conf 파일의 기본 구조입니다. filter 영역은 선택적인 부분인데 외부에서 받아온 로그 데이터를 일정 단위로 구분해서 분석해야 한다면 작성해야되는 부분입니다.
Logstash 의 5044 포트에서 beats 방식으로 로그 데이터를 입력받고 출력은 rubydebug 형식으로 하고 있습니다. conf 파일 작성도 끝났으면 이제 Logstash 를 실행해보겠습니다.
여기서!! 실행하기전에 매번 설치경로 들어가서 실행하기 귀찮으니까 심볼링 링크먼저 걸어주고 하겠습니다.
$ ln -s logstash-7.12.0/bin/logstash logstash
$ ./logstash -f beat-pipeline.conf --config.reload.automatic
{
"message" => "83.149.9.216 - - [04/Jan/2015:05:13:42 +0000] \"GET /presentations/logstash-monitorama-2013/images/kibana-search.png HTTP/1.1\" 200 203023 \"http://semicomplete.com/presentations/logstash-monitorama-2013/\" \"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/32.0.1700.77 Safari/537.36\"",
"log" => {
"file" => {
"path" => "/home/elktest/log/logstash-tutorial.log"
},
"offset" => 75075
},
"input" => {
"type" => "log"
},
"tags" => [
[0] "beats_input_codec_plain_applied"
],
"@version" => "1",
"host" => {
"id" => "xxxxxxxxxxxxxxxx",
"containerized" => false,
"name" => "localhost.localdomain",
"mac" => [
[0] "xxxxxxxxxxxxxxxx"
],
"hostname" => "localhost.localdomain",
"architecture" => "x86_64",
"os" => {
"platform" => "centos",
"version" => "7 (Core)",
"name" => "CentOS Linux",
"type" => "linux",
"kernel" => "3.10.0-1160.el7.x86_64",
"family" => "redhat",
"codename" => "Core"
},
"ip" => [
[0] "192.168.92.129",
[1] "fe80::c1cc:e7e1:1449:f1b2"
]
},
"@timestamp" => 2021-04-21T04:57:53.854Z,
"agent" => {
"id" => "xxxxxxxxxxxxxxxxxxx",
"version" => "7.12.0",
"ephemeral_id" => "xxxxxxxxxxxxxxxxxxx",
"hostname" => "localhost.localdomain",
"name" => "localhost.localdomain",
"type" => "filebeat"
},
"ecs" => {
"version" => "1.8.0"
}
}
Logstash 에서 받아서 출력한 정보들입니다. 각 정보는 필드단위로 구분됩니다.
ElasticStack 을 사용하는 목적은 로그를 분석하여 모니터링 하는것인데 저렇게 "message" 필드에 들어간 내용을 그대로 엘라스틱 서치에 넣는것은 의미가 없습니다.
의미 있는 결과를 만들기 위해서는 읽어온 로그 데이터를 정보의 종류에 따라 분류해야 합니다.
위에서 Logstash 를 실행할때 작성한 conf 파일에 filter 영역이 있었습니다. filter 영역에서는 여러가지 플러그인을 사용할수 있는데 그중 grok 필터 플러그인을 사용하여 "message"필드를 분리해 여러개의 의미있는 필드를 만들수 있습니다.
작성했던 conf 파일의 filter 영역에 다음과같이 내용을 추가합니다.
filter {
grok {
match => { "message" => "%{COMBINEDAPACHELOG}"}
}
}
Logstash 를 실행할때 conf 변경시 자동으로 적용하는 옵션을 주었기때문에 conf 수정후 바로 업데이트해서 읽기 시작할것입니다.
하지만 filebeat 에서 이전에 한번 로그파일을 다읽어서 Logstash 로 전송했었기 때문에 같은 내용은 다시 읽지 않을것입니다. 오프셋 정보를 초기화 하기위해 작업이 필요합니다.
## filebeat 종료시킨후
$ cd /var/lib/filebeat
$ sudo rm -rf registry
$ sudo filebeat -e -c filebeat.yml -d "publish"
오프셋 정보를 초기화 시킨후 filebeat 를 시작하면 다시 읽어서 Logstash 로 전송합니다.
{
"message" => "86.1.76.62 - - [04/Jan/2015:05:30:37 +0000] \"GET /style2.css HTTP/1.1\" 200 4877 \"http://www.semicomplete.com/projects/xdotool/\" \"Mozilla/5.0 (X11; Linux x86_64; rv:24.0) Gecko/20140205 Firefox/24.0 Iceweasel/24.3.0\"",
"clientip" => "86.1.76.62",
"verb" => "GET",
"request" => "/style2.css",
"referrer" => "\"http://www.semicomplete.com/projects/xdotool/\"",
"ident" => "-",
"response" => "200",
"httpversion" => "1.1",
"bytes" => "4877",
"auth" => "-"
"agent" => {
"id" => "516d9258-175e-46fe-8195-1373e935fb9b",
"name" => "localhost.localdomain",
"version" => "7.12.0",
"hostname" => "localhost.localdomain",
"ephemeral_id" => "511aa1d3-d765-4a22-91d4-83bd275db324",
"type" => "filebeat"
},
....
....
....
....
}
이번엔 "message" 필드 외에 몇가지 필드들이 더 확인됩니다.
필드 명 | 정보 |
---|---|
clientip | IP Address |
ident | User ID |
auth | User Authentication |
timestamp | timestamp |
verb | HTTP Verb |
request | Request body |
httpversion | HTTP Version |
response | HTTP Status Code |
bytes | Bytes served |
referrer | Referrer URL |
agent | User agent |
이 예제의 경우는 아파치 로그의 포맷을 grok 플러그인에서 제공하기때문에 쉽게 해결했지만 우리가 분석해야 하는 로그가 아파치 로그만 있는것도 아니고 개발 하는 프로그램마다 로그포맷도 전부 다를것입니다.
grok 필터 플러그인은 내용이 많기때문에 따로 포스팅을 올리도록 하겠습니다.
이제 Logstash 에서 로그 데이터에 대한 필터작업도 끝났으니 터미널에 출력만하지말고 Elasticsearch 로 데이터를 저장해보겠습니다.
### Logstash conf 파일 수정
input {
beats {
port => "5044"
}
}
filter {
grok {
match => { "message" => "%{COMBINEDAPACHELOG}"}
}
}
output {
stdout {
codec => rubydebug
}
elasticsearch {
hosts => ["localhost:9200"]
}
}
output 영역에 elasticsearch 플러그인을 추가했습니다. 로그를 다시 읽어야 하니 filebeat 오프셋정보를 초기화하고 재시작해야 합니다.
$ sudo rm -rf /var/lib/filebeat/registry
$ sudo filebeat -e -c filebeat.yml -d "publish"
다시 로그를 읽기 시작하고 Elasticsearch 를 켜놓은 터미널에서도 로그가 찍히기 시작할것입니다.
우선 인덱스가 생성되었는지 확인해봅시다.
$ curl localhost:9200/_cat/indices?v
인덱스가 여러개 있는데 우리가 지금 테스트하면서 만들어진 인덱스는 logstash-{DATE} 형태의 인덱스입니다. logstash-2021.04.21-000001 인덱스를 확인해봅시다.
$ curl -XGET localhost:9200/logstash-2021.04.21-000001/_search?pretty
{
"took" : 1,
"timed_out" : false,
"_shards" : {
"total" : 1,
"successful" : 1,
"skipped" : 0,
"failed" : 0
},
"hits" : {
"total" : {
"value" : 98,
"relation" : "eq"
},
"max_score" : 0.02506397,
"hits" : [
{
"_index" : "logstash-2021.04.21-000001",
"_type" : "_doc",
"_id" : "RBJN83gBG_KGZIMpYKqv",
"_score" : 0.02506397,
"_source" : {
"clientip" : "83.149.9.216",
"log" : {
"file" : {
"path" : "/home/elktest/log/logstash-tutorial.log"
},
"offset" : 0
},
"ident" : "-",
"tags" : [
"beats_input_codec_plain_applied"
],
...
...
...
...
이러한 형식으로 터미널에 출력이 될것입니다. Kibana 도 같이 실행중이니까 웹에서도 확인할수 있습니다.
이번 포스팅은 내용이 조금 길었는데 filebeat => Logstash => Elasticsearch => Kibana 여러단계를 거친만큼 이해가 잘 안된다면 천천히 다시 읽으면서 확인해보시기 바랍니다.