[Elasticsearch] Logstash Grok

Grok을 사용한 Logstash 구문 분석 및 필터링

logstash는 CSV, JSON 파일을 쉽게 분석할 수 있다. 그러나 이 데이터들은 Elasticsearch가 분석할 수 있는 형식에 맞춰 이미 정리가 되어있기 때문이다. 하지만 가끔은 비구조화 데이터로 작업을 해야한다. 이런 경우는 구문 분석을 해서 구조화 데이터로 만들어주어야 한다. 이번에는 logstash의 grok filter를 사용하여 비구조화 데이터를 처리하는 방법에 대해서 알아보자.
많은 종류의 로그에서는 로그 메세지의 형태도 다양하며, 마지막에 에러 코드가 나타날수도, ip주소가 나타날 수도 있다. 이를 명확히 분석하기 위해서는 각 필드가 무엇을 나타내는지 감지할 수 있어야 한다. 이러한 작업을 grok 필터는 어떻게 수행할까?

• Regular Expression
  Grok filter는 각 텍스트를 분석하고 지시한 패턴과 일치하는지 확인함으로 구문 분석을 수행할 수 있다. 다행히도 logstash의 grok 필터에 다양한 일반적인 정규표현식이 미리 정의되어 있어 이를 적절히 활용하면 된다. elasticsearch grok filter 에서 보다 자세한 내용을 찾아볼 수 있다.
• Generic Grok Syntax

%{PATTERN:identifier}

일반적인 grok 구문은 위와 같이 생겼다. 어떤 패턴을 찾으라고 지시하고, 패턴과 일치하는 문자열에 어떻게 label을 붙여야하는지 알려준다. 또 다른 예로는 email grok pattern이 있다. 이는 %{EMAILADDRESS:client_email} 과 같은 형태로 생겼다.

• Example
  • 2020-07-16T19:20:30.45+01:00 DEBUG This is a sample log
  • 위와 같은 log가 있다고 했을 때, 이 로그는 Timestamp, Loglevel, Message 필터로 총 세 부분으로 나누어 추출할 수 있다.
  • %{TIMESTAMP_ISO8601:time} %{LOGLEVEL:logLevel} %{GREEDYDATA:logMessage}
• Debugger Tool
  • Grok Debug Tool 를 통해서 쉽게 grok filter debugging을 수행할 수 있다.
• 다음과 같은 log file이 있다고 하자.

2020-10-11T09:49:35Z INFO variable server value is tomcat
2020-03-14T22:50:34Z ERROR cannot find the requested resource
2020-01-02T14:58:40Z INFO initializing the bootup
2020-06-04T06:56:04Z DEBUG initializing checksum
2020-05-07T03:07:11Z INFO variable server value is tomcat
55.12.32.134 GET /user/id/properties

• 그리고 conf 파일은 다음과 같다.

input {
  file {
    path => "/home/student/03-grok-examples/sample.log"
    start_position => "beginning"
    sincedb_path => "/dev/null"
  }
}
filter {
  grok {
    match => { "message" => ['%{TIMESTAMP_ISO8601:time} %{LOGLEVEL:logLevel} %{GREEDYDATA:logMessage}'] }
  }
}
output {
   elasticsearch {
     hosts => "http://localhost:9200"
     index => "demo-grok"
  }

stdout {}

}

• 정의한 패턴과 일치하지 않는 텍스트가 주어진다면 grok 필터는 어떻게 텍스트를 다룰까?

{
        "_index" : "demo-grok",
        "_type" : "_doc",
        "_id" : "QfERpYQBL_Ulg_9BkWpv",
        "_score" : 1.0,
        "_source" : {
          "time" : "2020-05-07T03:07:11Z",
          "path" : "/home/student/03-grok-examples/sample.log",
          "@version" : "1",
          "message" : "2020-05-07T03:07:11Z INFO variable server value is tomcat",
          "logLevel" : "INFO",
          "logMessage" : "variable server value is tomcat",
          "host" : "es7",
          "@timestamp" : "2022-11-23T15:18:22.087Z"
        }
      },
-------------------
      {
        "_index" : "demo-grok",
        "_type" : "_doc",
        "_id" : "QvERpYQBL_Ulg_9BkWpv",
        "_score" : 1.0,
        "_source" : {
          "path" : "/home/student/03-grok-examples/sample.log",
          "@version" : "1",
          "message" : "55:12.32.134 GET /user/id/properties",
          "tags" : [
            "_grokparsefailure"
          ],
          "host" : "es7",
          "@timestamp" : "2022-11-23T15:18:22.090Z"
        }

경계선을 기준으로 위의 document는 정상적으로 정의한 grok 필터대로 index화가 된 모습이지만 아래의 document는 _grokparsefailure라는 tag가 달렸다.

• 복잡판 파일에 여러 grok 필터를 사용해 구문 분석도 해보자.

input {
  file {
    path => "/home/student/03-grok-examples/sample.log"
    start_position => "beginning"
    sincedb_path => "/dev/null"
  }
}
filter {
  grok {
    match => { "message" => [
              '%{TIMESTAMP_ISO8601:time} %{LOGLEVEL:logLevel} %{GREEDYDATA:logMessage}',
              '%{IP:clientIP} %{WORD:httpMethod} %{URIPATH:url}'
              ] }
}
}
output {
   elasticsearch {
     hosts => "http://localhost:9200"
     index => "demo-grok-multiple"
  }

stdout {}

}

이렇게 grok 필터 안에 따옴표로 묶어 필터를 지정하면 된다. 이렇게까지 하는 이유는 여러 프로그램이 동일한 파일에 로깅하는 복잡한 로그를 처리해야할 때 유용하다. 위와 같은 방식으로 필터를 지정한 후 다시 logstash를 동일하게 실행시켰을 경우 동일한 로그에 대해서 다음과 같이 fail tag 없이 정상적으인 document로 저장된다.

{
        "_index" : "demo-grok-multiple",
        "_type" : "_doc",
        "_id" : "VPEopYQBL_Ulg_9By2q-",
        "_score" : 1.0,
        "_source" : {
          "@timestamp" : "2022-11-23T15:43:44.405Z",
          "message" : "55.12.32.134 GET /user/id/properties",
          "clientIP" : "55.12.32.134",
          "url" : "/user/id/properties",
          "host" : "es7",
          "@version" : "1",
          "path" : "/home/student/03-grok-examples/sample.log",
          "httpMethod" : "GET"
        }
      }