[ElasticStack] Logstash 에서 Elasticsearch 로 데이터 전송하기

지난번에 Logstash 를 설치하고 실행하여 입력한대로 출력이 되는것까지 해봤습니다.

이번엔 파일비트에서 받아오는 로그 데이터를 파싱하는 방법에 대해 설명해보겠습니다.

 

filebeat => Logstash 연동

filebeat 설치글은 https://ksr930.tistory.com/96 를 참고하세요.

 

Logstash 에서 입력을 받는 input plugin 에는 여러종류가 있는데 그중 filebeat 에서

입력받을수 있는 Beats input 은 기본으로 설치되있습니다.

filebeat 에서 Logstash 로 전송할 로그데이터는 엘라스틱서치 사이트에서 제공하는

샘플데이터를 이용합니다.

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 설정 변경

먼저 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 부분이

주석이 해제되있다면 해당부분은 주석처리 하면됩니다. 지금은 엘라스틱서치로

바로 보내는게 아니라 Logstash 로 전송해야 하기 때문입니다.

 

filebeat.yml 파일의 변경 내용을 적용해야 하므로 filebeat 데몬이 실행중이라면

재실행 해야 합니다.

$ sudo systemctl restart filebeat

 

Logstash 설정 변경

이제 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

• -f : 작성한 conf 파일을 지정
• --config.reload.automatic : Logstash가 실행중일때 conf 파일을 변경시 변경된 내용 자동 적용

 

{
       "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 에서 받아서 출력한 정보들입니다. 각 정보는 필드단위로 구분됩니다.

• message : 로그파일에서 읽은 데이터
• agent : Logstash 정보
• host : Logstash 가 실행되고있는 시스템의 정보
• @timestamp : 처리된 시간

 

ElasticStack 을 사용하는 목적은 로그를 분석하여 모니터링 하는것인데 저렇게

"message" 필드에 들어간 내용을 그대로 엘라스틱 서치에 넣는것은 의미가 없습니다.

의미 있는 결과를 만들기 위해서는 읽어온 로그 데이터를 정보의 종류에 따라 분류해야 합니다.

 

Logstash 필터 / grok

위에서 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 필터 플러그인은 내용이 많기때문에 따로 포스팅을 올리도록 하겠습니다.

 

Elasticsearch 저장

이제 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

여러단계를 거친만큼 이해가 잘 안된다면 천천히 다시 읽으면서 확인해보시기 바랍니다.

 

참고 사이트 : https://www.elastic.co/guide/en/logstash/current/advanced-pipeline.html