로그 파일을 키바나와 연동하여 모니터링하기2 - Logstash 수집된 에러로그 데이터 파싱하기

이 글은 아래의 문서와 이어집니다.
로그 파일을 키바나와 연동하여 모니터링하기1 - Filebeat 설치 및 Logstash, Kibana 연동

데이터 파싱, 왜 해야하죠?

Logstash는 ES에 데이터를 저장하기 전에 원하는 형태로 가공하는 역할을 합니다. 그럼 원하는 형태로 가공을 해야하는 이유는 뭘까요?
Logstash는 JSON 형태로 데이터를 출력하는데 Logstash가 만든 필드는 사용자가 만든 필드가 충돌날 상황을 대비해 필드명 앞에 @ 기호가 붙어 있습니다.
즉, @ 기호가 붙지 않은 필드는 수집을 통해 얻어진 정보이고 @ 기호가 붙은 필드는 Logstash에 의해 생성된 필드입니다. Logstash가 기본적으로 만들어 주는 필드는 상황에 따라 필요할 수 있지만, 앞으로 진행하려는 에러 로그 예시의 경우엔 오히려 이러한 기본 정보들이 에러를 한눈에 파악하는 데에 어려움을 주고 있어서 Logstash가 제공하는 기본 필드를 삭제할 예정입니다.
또한, 파싱을 하지 않으면 Logstash는 수집된 데이터를 message 필드에 통째로 저장하기 때문에 후에 Kibana에서 상세하게 분석하기 어렵습니다. 가령, 특정 메서드에서 얼마나 많은 에러 로그가 발생했는지 분석해야 하는데 로그가 통째로 들어있다면 이를 분석하기가 어렵겠죠? 다음은 앞으로 파싱을 진행할 에러 로그의 포맷입니다.

파싱하려는 로그는 API 성공 여부 - 걸린 시간 - API 시작 시간- 리퀘스트 ID - 유저ID - IP - 클래스명 - 메소드명 이 한 줄에 있고 파라미터 정보와 응답값을 차례대로 개행하여 출력한 뒤 에러 메시지를 출력하는 형식으로 되어있습니다.

#F# 13#2023-02-09 16:19:09:476# Req1TVK4NdA2QZ# 1SXvq2et04y# admin# 111.111.111.11#BoardServiceImpl#get
  [PARAM 01] : 	{String}{... 파라미터 정보}
  [RETURN]:	{-}
org.mybatis.spring.MyBatisSystemException: nested exception is org.apache.ibatis.exceptions.TooManyResultsException: Expected one result (or null) to be returned by selectOne(), but found: 136

첫 번째 줄에 있는 속성들을 # 기호를 기준으로 개별적인 필드로 추출하고, exception이라는 필드를 생성하여 에러 메시지만 확인하도록 파싱할 예정입니다. 전체 에러 로그도 확인할 수 있도록 message 필드는 삭제하지 않겠습니다.

작업은 크게 1) 필요한 필드만 남겨두고 나머지 필드는 삭제한다., 2) message 필드의 값을 세부적으로 분리한다. 로 나눠볼 수 있겠습니다.

데이터를 의미있는 구조로 파싱하기

Logstash의 이벤트 처리 파이프라인에는 input(입력) -> filter(필터) -> output(출력) 세 단계가 있습니다. 여기서 필터는 데이터를 정형화하고 의미있는 데이터 형태로 가공하는 데 핵심적인 역할을 합니다.
다음에 나오는 코드들은 저에게 필요한 것들만 정제하여 사용하는 예시이니, 각자의 상황에 맞게 변형하여 적용하시면 됩니다.

1) 필요한 필드만 추리기

agent나 host로 시작하는 기본 제공 필드는 메타데이터 성격을 띕니다. 실제로 파악하고자 하는 내용은 에러 메시지인데 기본 필드의 양이 많아 한눈에 빨리 로그를 파악하기 어렵게 만드므로 remove_field를 사용하여 agent나 host로 시작하는 필드들만 일단 제거하겠습니다.

정규식을 사용하면 특정 문자를 prefix로 하는 필드를 제거할 수 있습니다.
다음은 엘라스틱서치 문서에서 제공하는 샘플 코드입니다.

# You can also remove multiple fields at once:
filter {
    mutate {
      remove_field => [ "foo_%{somefield}", "my_extraneous_field" ]
    }

filter {    
    mutate {  
        remove_field => [ "/.*agent.*/", "/.*host.*/" ]   
    }
}

remove_field 는 Common Option이기 때문에 mutate 뿐만 아니라 모든 필터 플러그인에서 사용할 수 있는 옵션입니다.

2) grok 패턴을 사용하여 message에서 필요한 필드 추출하기

위에서 설명한 에러 로그의 패턴을 다시 보겠습니다.

#F#    13#2023-02-09 16:19:09:476#      0000000000#    1SXvq2et04y#          userId# 106.245.152.12#BoardServiceImpl#get
	[PARAM 01] : 	{String}{... 파라미터 정보}
	[RETURN]:	{-}
org.mybatis.spring.MyBatisSystemException: nested exception is org.apache.ibatis.exceptions.TooManyResultsException: Expected one result (or null) to be returned by selectOne(), but found: 136
	
	... 중략

	... 129 common frames omitted

첫번째 줄에는 호출된 API에 대한 정보가 담겨있습니다.

#F# 13#2023-02-09 16:19:09:476# Req1TVK4NdA2QZ# 1SXvq2et04y# admin# 106.245.152.12#BoardServiceImpl#get

#를 기준으로 해당 데이터는 다음을 의미합니다.

#F(성공여부:api_result)# 13(걸린시간:api_elapse_time)#2023-02-09 16:19:09:476(api시작시간:api_start_time)# Req1TVK4NdA2QZ(요청ID:request_id)# 1SXvq2et04y(로그인유저의 고유ID:user_oid)# admin(로그인유저ID:user_id)# 106.245.152.12(ip:client_ip)#BoardServiceImpl(api 클래스명:api_class_name)#get(api 메서드명:api_method_name)

첫번째 줄은 정형화된 형식을 가지고 있기 때문에 grok 패턴 정규식을 사용하여 파싱해보겠습니다.

참고로 파라미터 정보가 담겨있는 [PARAM 01] : ~~~ 부분은 파라미터 갯수에 따라 동적으로 라인이 변하기 때문에 따로 빼지 않고 에러로그와 함께 통째로 exception 이라는 필드에 넣었습니다. 혹시 해당 부분에 대해 해결책을 아시는 분이 있으시면 댓글 부탁드립니다,, 흑흑

grok 패턴이란?

gork은 자주 사용하는 정규 표현식들을 패턴화 해두었습니다. 이러한 패턴을 이용해 %{패턴:필드명} 형태로 데이터에서 특정 필드를 파싱할 수 있습니다. grok에서 기본으로 지원하는 120개의 패턴들은 https://github.com/elastic/logstash/blob/v1.4.0/patterns/grok-patterns 에서 확인할 수 있습니다. 사람들이 자주 사용하는 패턴들은 이미 만들어져 있기 때문에 가져다 쓰면 되고, 원하는 패턴이 없는 경우 직접 패턴을 만들어 사용하면 됩니다. 패턴 대신 사용하려면 (?<필드명>정규표현식)형태로 작성합니다.

dissect와 grok
dissect와 grok 플러그인은 패턴을 이용해 구문 분석을 한다는 공통점이 있지만 성능상에 차이가 있습니다.
dissect는 패턴을 해석하지 않아 속도가 빠르기 때문에 로그 형식이 일정하고 패턴이 변하지 않는다면 dissect를 사용하는 편이 좋습니다.
grok은 예외 처리나 패턴이 자유롭기 때문에 로그 형태가 일정하지 않다면 grok을 사용하는 편이 좋습니다.
결론적으로, 성능상의 차이가 존재하므로 grok은 dissect와 기타 필터 조합으로 해결되지 않는 진짜 정규 표현식이 필요한 경우에만 사용하는 것이 좋습니다.

먼저, 최종으로 완성된 grok 정규식은 다음과 같습니다.

(?m)\#(?<api_result>S|F?)\#%{SPACE}(%{NUMBER:api_elapse_time:int})?\#%{DATA:api_start_time}\#%{SPACE}(%{DATA:request_id})?\#%{SPACE}%{DATA:user_oid}\#%{SPACE}%{DATA:user_id}\#%{SPACE}%{IP:client_ip}\#%{DATA:api_class_name}\#%{DATA:api_method_name}%{GREEDYDATA:exception}

(?m) : 정규식 시작부분에 선언하면 multiline pattern임을 알려주어 줄바꿈이 일어나도 정규식을 멈추지 않고 여러 줄의 데이터를 읽습니다. 저의 경우 에러의 stack trace까지 정규식에 포함시켰기 때문에 multiline으로 지정하였습니다.
\# : ‘[’, ‘-‘, ‘.’과 같은 특수문자는 백슬래시(\)와 함께 사용하여 escape합니다.
(?<api_result>S|F?)
- ? : 사용자가 지정한 문자열의 마지막 글자가 있거나 없을때에 대해 매칭합니다.
- S|F? : S 또는 F가 없는 데이터가 들어와도 에러가 나지 않고 매칭이 되게 하기 위해 ?를 붙여주었습니다. |는 OR 연산자입니다.
%{SPACE} : 스페이스, 탭 등 하나 이상의 공백을 인식합니다.
(%{NUMBER:api_elapse_time:int})?
- NUMBER:api_elapse_time:int : NUMBER는 십진수를 인식하며 부호와 소수점을 포함할 수 있습니다. 여기서 변수명 뒤에 :int를 추가하면 들어온 문자열을 숫자로 변경시 정수로 타입으로 지정합니다. 나중에 키바나에서 걸린시간에 대한 통계에 사용할 것이므로 NUMBER 타입을 지정했습니다.
- (%{NUMBER:api_elapse_time:int})? : ()? 를 사용하면 괄호 내부의 데이터를 부분집합처럼 사용할 수 있습니다. 해당 데이터가 들어오지 않아도 에러가 나지 않게 하기 위해 사용합니다.
%{DATA:api_start_time} : DATA는 이 패턴의 직전 패턴부터 다음 패턴 사이를 모두 인식합니다. 특별히 인식하고자 하는 값의 유형을 신경 쓸 필요가 없을 때 주로 사용합니다.
%{IP:client_ip} : IP는 IP 주소를 인식합니다. IPv4나 IPv6를 모두 인식할 수 있습니다.
%{GREEDYDATA:exception} : DATA 타입과 동일하나 표현식의 가장 뒤에 위치시킬 경우 해당 위치부터 이벤트의 끝까지를 값으로 인식합니다.
- 첫번째 줄의 마지막 데이터인 메서드 명이 끝난 후의 모든 데이터를 exception이라는 필드명에 담을 것이기 때문에 GREEDYDATA를 사용합니다.

정규식 적용 방법은 여기 를 참고했습니다.

Kibana를 실행중이라면, grok 패턴을 적용하면 데이터가 어떻게 재구성 되는지 Kibana의 Grok Debugger에서 확인할 수 있습니다.

적용된 Logstash.yml

input {
    beats {
        port => 6022
        include_codec_tag => false
    }
}
filter {
  mutate {
    gsub => [ "message", "\r", "" ]
  }
  grok {
    match => [ "message", "(?m)\#(?<api_result>S|F?)\#%{SPACE}(%{NUMBER:api_elapse_time:int})?\#%{DATA:api_start_time}\#%{SPACE}(%{DATA:request_id})?\#%{SPACE}%{DATA:user_oid}\#%{SPACE}%{DATA:user_id}\#%{SPACE}%{IP:client_ip}\#%{DATA:api_class_name}\#%{DATA:api_method_name}%{GREEDYDATA:exception}" ]
		remove_field => [ "[agent][ephemeral_id]", 
							"[agent][hostname]", 
							"[agent][id]", 
							"[agent][name]", 
							"[agent][type]", 
							"[agent][version]", 
							"[ecs][version]", 
							"[host][architecture]", 
							"[host][containerized]", 
							"[host][hostname]", 
							"[host][id]", 
							"[host][name]", 
							"[host][mac]",
							"[host][os][codename]",
							"[host][os][family]",
							"[host][os][kernel]",
							"[host][os][name]",
							"[host][os][platform]",
							"[host][os][type]",
							"[host][os][version]"
						]
  }
  date {
    match => [ "api_start_time" , "yyyy-MM-dd HH:mm:ss,SSS" ]
  }
}

output {
		
        elasticsearch {
            hosts => ["localhost:6001"]
            index => "filebeat-err"
        }
			 
}

include_codec_tag => false
- 이것을 설정하지 않으면 beats_input_codec_plain_applied 라는 tag가 생성됩니다. Beats가 생성하는 메시지는 모두 저 항목을 포함하고 있으므로 무시해도 되지만 input 단계에서 설정을 추가하여 제거합니다.
gsub => [ “message”, “\r”, “” ]
- \r이 로그마다 다르게 위치하여 grok 정규식에 매칭이 제대로 되지 않아 “”으로 치환합니다.
date { match => [ “api_start_time” , “yyyy-MM-dd HH:mm:ss,SSS” ] }
- date 플러그인은 날짜/시간 데이터를 ISO8601 타입으로 일괄 변경합니다.

Kibana에서 확인하기

기존에는 아래와 같이 에러 로그의 모든 정보가 message 필드 하나에 뭉쳐있었습니다.

설정 적용 후 확인해보면 다음과 같이 기본 정보 필드들이 제거되었고, error stacktrace를 exception에서 확인할 수 있으며 API 호출정보는 필드가 따로 분리가 되었습니다.

Twitter Facebook LinkedIn

로그 파일을 키바나와 연동하여 모니터링하기2 - Logstash 수집된 에러로그 데이터 파싱하기

Grey

데이터 파싱, 왜 해야하죠?

데이터를 의미있는 구조로 파싱하기

1) 필요한 필드만 추리기

2) grok 패턴을 사용하여 message에서 필요한 필드 추출하기

grok 패턴이란?

적용된 Logstash.yml

Kibana에서 확인하기

공유하기

댓글남기기

참고

이력서 기반으로 ChatGPT와 나의 강점과 약점 분석하고 계획하기

기업이 원하는 PM의 조건은 뭘까

[항해99 취업 리부트 코스 학습일지][day21] 질문 대폭발의 날

[항해99 취업 리부트 코스 학습일지][day20] 어제보단 나은 하루