Logstash 7.16.2 버전에서 로그 파싱하기

{"svctype":"test-svc-type","instance_name":"test-instance_ip-0.0.0.0", "@timestamp":"2021-12-30T12:30:06.731Z","environment":"test","role":"api","filename":"test.log","@version":"1","message":"* 2021-12-30 21:30:06,690 test log message","host":"ip-0.0.0.0.ap-northeast-2.compute.internal","team":"test"}

위와 같은 로그가 있다.

각 서버에서 위와 같은 로그를 받아와, 로그스태시에 input 형태로 넣고, SQS로 아웃풋을 보낸다.

중요한 것은 설정 파일에서 

input {
  file {
    path => "/usr/share/logstash/data/*.log"
    start_position => "beginning"
    codec => "json"
  }
}
output {
  stdout { }
  sqs {
    access_key_id => "***"
    secret_access_key => "****"
    codec => "json"
    queue => "test_queue"
    region => "ap-northeast-2"
    }
}

codec => "json" 을 해야 위의 로그가 다음과 같은 형태로 잘 파싱된다는 점에 있다. 

이걸 모르고 로그 자체가 제대로 파싱이 안되어서 한참 삽질했다..

{
         "filename" => "test.log",
          "message" => "* 2021-12-30 21:30:06,690 test log message",
          "svctype" => "test-svc-type",
         "@version" => "1",
       "@timestamp" => 2021-12-30T12:30:06.731Z,
             "path" => "/usr/share/logstash/data/test.log",
             "team" => "test",
             "host" => "ip-0.0.0.0.ap-northeast-2.compute.internal",
             "role" => "api",
      "environment" => "test",
    "instance_name" => "test-instance_ip-0.0.0.0"
}

 

첫번째 로그스태시를 거치면 로그가 이렇게 아름답게 파싱되지만, 단계가 더 남았다.

여러 서버를 가동할 경우 각 서버의 로그를 다 모아서 편히 볼 수 있도록,
콜렉터 역할을 하는 로그스태시 하나를 더 실행해 SQS에서 로그를 받아온다.

input {
    sqs {
        codec => "json"
        queue => "test_queue"
        region => "ap-northeast-2"
        threads => 8
        polling_frequency => 1
        access_key_id => "****"
        secret_access_key => "****"
    }
}
filter {
    grok {
      patterns_dir => ["/etc/logstash/patterns"]
      match => { "message" => "%{TEST}" }
      keep_empty_captures => true
    }
    mutate {
      remove_field => ["message"]
    }
  }
}
output {
    elasticsearch {
        index => "test-log-%{+YYYY-MM-dd}"
        hosts => ["127.0.0.1:9200"]
        }
     }
 }

그리고 로그 메시지도 예쁘게 한번 더 파싱해줘야 한다.

두번째 로그스태시의 설정 파일에서

          "message" => "* 2021-12-30 21:30:06,690 test log message",

이 메시지를 grok으로 파싱하도록 filter 설정을 해준다.

patterns_dir 파일 밑에 test 파일을 생성해

APPDATE %{YEAR}-%{MONTHNUM}-%{MONTHDAY} %{TIME}
TEST ^\* %{APPDATE:time_local} %{GREEDYDATA:log_message}

을 만들어준다. 패턴은 로그 메시지 형태에 따라 바꾸면 된다.

output을 엘라스틱서치나 S3으로 쏴주면 로그를 키바나에서 간편하게 확인할 수 있다.