Logstash로 로그 분석2 (필터)

 Grok 필터 플러그인을 통한 웹 로그 분석

 전단계에서 Filebeat를 통해 로그를 읽어들이는 파이프라인을 만들었다. 허나, 출력 로그 포맷이 그닥 인상적이진 않다는 것을 안다. 좀더 원하는 필드로 구분지어 특정한 형태로 로그를 출력시키길 원한다면 어떨까. 이를 위해 grok 필터 플러그인이 존재한다.

  이 grok 필터 플러그인은 Logstash에 기본으로 내장된 몇 플러그인 중 하나일 뿐이다. 상세는 역시 플러그인 관리자의 공식 문서를 보길 바란다.

 이 플러그인은 비구조화된 데이터를 어떤 질의가능한 구조화된 데이터로 변환하는 작업을 진행할 때 사용한다.

 이 grok 필터 플러그인이 들어오는 데이터의 패턴을 찾는 기능을 제공해 주기 때문에 어떤 패턴을 이용하여 데이터를 식별할지를 결정하는 것은 당신의 목적에 달려 있다. 대표적으로 웹 로그의 내용은 아래와 같다.

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"

 

IP 주소로 시작하며, 이 IP를 식별하는 건 쉽다. 괄호 안에 시간정보가 있기 때문이다. 이 데이터를 뽑아내려면 %{COMBINEDAPACHELOG} 란 grok 패턴을 사용하면 되는데, 이는 아래의 스키마로 들어오는 아파치 로그를 추출할 때 쓸 수 있는 미리 정의된 패턴이다.

Information	Field Name
IP Address	clientip
User ID	ident
User Authentication	auth
timestamp	timestamp
HTTP Verb	verb
Request body	request
HTTP Version	httpversion
HTTP Status Code	response
Bytes served	bytes
Referrer URL	referrer
User agent	agent

 ※ 만일 grok 패턴을 구축하는데 도움이 필요하다면 Grok Debugger를 사용해보라. 무료로 사용가능한 라이센스다.

first-pipeline.conf 를 수정하자. filter 구역을 밑의 글로 변경한다.

filter {
    grok {
        match => { "message" => "%{COMBINEDAPACHELOG}"}
    }
}

 

모두 완료한 뒤의 first-pipeline.conf의 내용은 이하와 같다.

input {
    beats {
        port => "5043"
    }
}
filter {
    grok {
        match => { "message" => "%{COMBINEDAPACHELOG}"}
    }
}
output {
    stdout { codec => rubydebug }
}

 

저장하라. 이전에 자동 설정 감지 후 재로딩되도록 옵션을 준 상태로 logstash를 구동했으므로 직접 다시 구동명령을 날릴 필요가 없다. 단, Filebeat는 로그 파일을 다시 읽도록 해야 하는데 이러기 위해선 Filebeat가 구동되는 터미널에서 ^C 를 눌러 Filebeat를 종료해야 한다. Filebeat의 레지스트리 파일을 지우고 다시 돌린다.

sudo rm data/registry

Filebeat가 각각의 파일의 읽은 정도를 레지스트리 파일에 저장하기 때문에 직접 지우면 처음부터 읽게 되는 것이다.

이제 Filebeat를 재시작 해보자.

sudo ./filebeat -e -c filebeat.yml -d "publish"

grok 패턴으로 처리가 진행될 것이며 결과는 아래와 같이 나오게 된다.

{
        "request" => "/presentations/logstash-monitorama-2013/images/kibana-search.png",
          "agent" => "\"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\"",
         "offset" => 325,
           "auth" => "-",
          "ident" => "-",
     "input_type" => "log",
           "verb" => "GET",
         "source" => "/path/to/file/logstash-tutorial.log",
        "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\"",
           "type" => "log",
           "tags" => [
        [0] "beats_input_codec_plain_applied"
    ],
       "referrer" => "\"http://semicomplete.com/presentations/logstash-monitorama-2013/\"",
     "@timestamp" => 2016-10-11T21:04:36.167Z,
       "response" => "200",
          "bytes" => "203023",
       "clientip" => "83.149.9.216",
       "@version" => "1",
           "beat" => {
        "hostname" => "My-MacBook-Pro.local",
            "name" => "My-MacBook-Pro.local"
    },
           "host" => "My-MacBook-Pro.local",
    "httpversion" => "1.1",
      "timestamp" => "04/Jan/2015:05:13:42 +0000"
}

원본 메시지를 포함하고 있지만 각각이 또하나의 별도 필드들로 구분되어 나타남을 잘 살펴보길 바란다.

Geoip 필터 플러그인으로 데이터 강화하기

로그를 분석하여 더 좋은 형태로 만드는 것도 가능하지만 추가 보충 정보를 기존 데이터로부터 추려내는 것도 가능하다. 예를 들어, geoip는 IP를 찾아 주소로부터 위치 정보를 알아낼 수도 있고 이를 통해 로그 출력에 위치 정보를 넣을 수도 있다.

geoip 필터 플러그인을 first-pipeline.conf의 필터 영역에 아래 구문을 추가하여 적용시켜 보자.

    geoip {
        source => "clientip"
    }

geoip 플러그인 설정은 데이터가 있는 곳에 대한 아이피 주소를 설정 필드로 가져가는데 여기서 clientip는 그 아이피를 포함하고 있다.

필터는 표기된 순서대로 평가되고 실행되기 때문에, geoip가 grok 보다 뒤에 오도록 확실히 하자. 패턴으로 특별한 데이터를 빼낸 뒤에 각 정보에 대해 추가 정보를 찾는 것이 자연스럽기 때문이다. 그리고 둘다 같은 필터 영역 내에 기술한다.

input {
    beats {
        port => "5043"
    }
}
 filter {
    grok {
        match => { "message" => "%{COMBINEDAPACHELOG}"}
    }
    geoip {
        source => "clientip"
    }
}
output {
    stdout { codec => rubydebug }
}

 

저장해라. 역시 처음부터 로그를 읽기 위해 Filebeat의 레지스터 파일을 지워라. 물론, Filebeat부터 ^C로 종료한 뒤에 말이다. 지웠으면 재시작한다.

sudo ./filebeat -e -c filebeat.yml -d "publish"

이제 결과에 지리 정보가 출력된다.

{
        "request" => "/presentations/logstash-monitorama-2013/images/kibana-search.png",
          "agent" => "\"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\"",
          "geoip" => {
              "timezone" => "Europe/Moscow",
                    "ip" => "83.149.9.216",
              "latitude" => 55.7522,
        "continent_code" => "EU",
             "city_name" => "Moscow",
         "country_code2" => "RU",
          "country_name" => "Russia",
              "dma_code" => nil,
         "country_code3" => "RU",
           "region_name" => "Moscow",
              "location" => [
            [0] 37.6156,
            [1] 55.7522
        ],
           "postal_code" => "101194",
             "longitude" => 37.6156,
           "region_code" => "MOW"
    },
    ...

(이하 생략)