Skip to content

Latest commit

 

History

History
469 lines (376 loc) · 16.1 KB

005.Log_Data_Practice.md

File metadata and controls

469 lines (376 loc) · 16.1 KB

005. Log Data Practice 1 - Log Data 가공 후 Elasticsearch 저장하기

해당 실습에서는 Elastic Stack 환경 구축에 대한 내용은 포함하고 있지 않습니다.

00. 실습환경

  • Windows 10
  • Elasticsearch 7.10.2
  • Logstash 7.10.2
  • Kibana 7.10.2
  • Filebeat 7.10.2
  • Git Bash

Elasticsearch와 Kibana가 실행된 상태로 아래 실습을 진행해주세요.

01. 데이터 준비하기

실습을 진행하기 위해 임의로 생성한 로그 파일을 활용하도록 하겠습니다. 로그 파일은 아래 세 가지 파일을 사용할 예정이며, 로그 내용은 임의로 구성하였습니다.

02. filebeat 설정

먼저, filebeat 설정을 하겠습니다. access.log, error.log, rest.log 각각에 대해 설정을 추가하였습니다.

filebeat.inputs:
- type: log
  enabled: true
  paths:
     - E:/personal/Workspace/TIL/practice/elastic-stack-posting/post005/access.log
  multiline:
    pattern: '^(?:[0-9]{1,3}\.){3}[0-9]{1,3}$|^(?:[0-9a-fA-F]{1,4}:){7}[0-9a-fA-F]{1,4}$'
    negate: true
    match: after
  fields:
    log_file_name: access.log
    server_name:  practice01

- type: log
  enabled: true
  paths:
     - E:/personal/Workspace/TIL/practice/elastic-stack-posting/post005/error.log
  multiline.pattern: '^\d{4}-\d{2}-\d{2}'
  multiline.negate: true
  multiline.match: after
  fields:
    log_file_name: error.log
    server_name:  practice01

- type: log
  enabled: true
  paths:
     - E:/personal/Workspace/TIL/practice/elastic-stack-posting/post005/rest.log
  multiline.pattern: '^\d{4}-\d{2}-\d{2}'
  multiline.negate: true
  multiline.match: after
  fields:
    log_file_name: rest.log
    server_name:  practice01

output.logstash:
  hosts: ["localhost:5044"]

설정 파일에서 사용한 필드에 대해 정리하면 다음과 같습니다.

  • filebeat.inputs : input 설정
  • type : 수집 파일의 input 타입
  • enabled : 연결 여부
  • paths : log 위치
  • multiline : 여러 줄로 인식하는 방법에 대한 설정. 여기서는 "날짜로 시작하지 않는 줄"을 "날짜로 시작하는 줄" 바로 뒤에 이어서 오도록 설정함.
    • multiline.pattern : 정규식을 이용해 패턴 지정
    • multiline.negate : true일 때 패턴 일치 조건을 반전시킴
    • multiline.match : 멀티라인을 처리하는 방식으로, before와 after를 지정할 수 있음
  • fields : 필드 정보 설정
    • log_file_name : 로그 파일 이름 설정
    • server_name : 서버 이름 설정
  • output.logstash : output으로 logstash 설정
    • hosts : logstash host

설정 파일 내에 multiline 설정, server_name도 동일하므로 anchor(앵커)를 사용하여 중복을 제거해주도록 하겠습니다. 참고로, anchor(앵커)는 YAML 문서에서 재사용 가능한 값 또는 블록을 정의하는 기능입니다.

설정 파일 내에서 중복이 되는 내용을 anchor(앵커)를 사용하여 제거해주도록 하겠습니다. 참고로, anchor(앵커)는 YAML 문서에서 재사용 가능한 값 또는 블록을 정의하는 기능입니다.

server_name_defaults: &server_name_defaults
  server_name: practice01
multiline_defaults: &multiline_defaults
  multiline:
    pattern: '^\d{4}-\d{2}-\d{2}'
    negate: true
    match: after
filebeat.inputs:
- type: log
  enabled: true
  paths:
     - E:/personal/Workspace/TIL/practice/elastic-stack-posting/post005/access.log
  multiline:
    pattern: '^(?:[0-9]{1,3}\.){3}[0-9]{1,3}$|^(?:[0-9a-fA-F]{1,4}:){7}[0-9a-fA-F]{1,4}$'
    negate: false
    match: after
  fields:
    log_file_name: access
    <<: *server_name_defaults

- type: log
  enabled: true
  paths:
     - E:/personal/Workspace/TIL/practice/elastic-stack-posting/post005/error.log
  <<: *multiline_defaults
  fields:
    log_file_name: error
    <<: *server_name_defaults

- type: log
  enabled: true
  paths:
     - E:/personal/Workspace/TIL/practice/elastic-stack-posting/post005/rest.log
  <<: *multiline_defaults
  fields:
    log_file_name: rest
    <<: *server_name_defaults

output.logstash:
  hosts: ["localhost:5044"]

그 다음 filebeat가 설치되어 있는 디렉터리에서 아래 명령어를 사용해 filebeat를 실행합니다.

./filebeat.exe -e

03. logstash 설정

이번에는 logstash를 설정하도록 하겠습니다. 먼저 pipelines.yml을 설정합니다.

- pipeline.id: practicepipe
  path.config: "../config/practice.conf"

그 다음 practice.conf 파일을 생성하여 다음과 같이 전달되는 로그를 잘라서 구분된 필드로 저장되도록 설정하였습니다.

input {
	beats {
		port => 5044
	}
}

filter {
	if [fields][server_name] == "practice01" {
		if [fields][log_file_name] == "access" {
			dissect {
				mapping => {
					"message" => "%{ip} - - [%{log_datetime} %{}] %{message}"
				}
			}
		}
		else if [fields][log_file_name] == "error" {
			dissect {
				mapping => {
					"message" => "%{log_datetime} %{+log_datetime} %{level} %{} -- %{message}"
				}
			}
		}
		else { # rest
			dissect {
				mapping => {
					"message" => "%{log_datetime} %{+log_datetime} %{} %{level} [%{code_path}] %{message}"
				}
			}
		}
	}
}

output {
	if [fields][server_name] == "practice01" {
		elasticsearch {
			hosts => ["localhost:9200"]
			index => "practice_01"
		}
	}
}

설정을 완료하면, logstash를 실행합니다.

./logstash.bat

04. kibana DevTools에서 데이터 확인하기

이제 Kibana에서 정상적으로 데이터를 불러오는지 확인해보겠습니다. Kibana의 DevTools에서 아래 명령어를 입력합니다.

GET _cat/indices/practice_*?v

그러면 다음과 같이 정상적으로 index가 생성된 것을 확인할 수 있습니다.

image

docs.count가 총 43개로 올바르게 생성된 것도 확인할 수 있습니다. 만약, 각 log_file_name에 대해서 보고 싶다면 다음과 같이 검색하면 됩니다. 검색 결과는 생략하도록 하겠습니다.

GET practice_01/_search
{
  "size": 5,
  "query": {
    "match": {
      "fields.log_file_name.keyword": "error"
    }
  }
}

05. log 날짜/시간 문자열 분석

현재 log_datetime은 18/Jul/2023:13:31:44 또는 2023-07-18 13:32:44.070과 같은 형태의 text 타입을 값으로 가집니다. 그렇기에 Kibana 내에서 활용하기 위해서는 다음과 같이 date 플러그인을 이용해 기본 날짜/시간 포맷으로 변경해야 합니다.

# ...

filter {
	if [fields][server_name] == "practice01" {
		# ... if - else if - elas 문
		
		# common =============
		mutate {
			strip => ["log_datetime"] # 공백 제거
		}
		
		date { # date 타입으로 변환
			match => ["log_datetime", "YYYY-MM-dd HH:mm:ss.SSS", "dd/MMM/YYYY:HH:mm:ss"]
			target => "datetime"
			timezone => "Asia/Seoul"
		}
	}
}

# ...

log_datetime 필드 중에서 YYYY-MM-dd HH:mm:ss.SSS 포맷이거나 dd/MMM/YYYY:HH:mm:ss 포맷인 경우에 date 플러그인을 통해 매칭할 수있습니다. target은 매핑된 필드가 저장될 새로운 필드를 의미하며, 여기서는 datetime 필드를 생성합니다. 만약, datetime 대신 log_datetime를 사용하면 기존 필드의 내용을 덮어쓰게 됩니다. 로그스태시에서 사용하는 날짜/시간 포맷은 Joda Time 라이브러리를 사용하며, 자세한 내용은 공식 문서를 참고하길 바랍니다.

timezone의 경우, Asia/Seoul이라고 설정해야 입력받은 log_datetime을 다른 시간대가 아닌 Asia/Seoul 시간대로 인식합니다. 설정을 적용하기 위해 logstash와 filebeat를 재실행합니다. 이 때, logstash에서 생성한 인덱스와 filebeat data 디렉터리를 제거하고 진행해주세요!

logstash와 filebeat가 재실행되고, practice_01 인덱스가 다시 생성되었다면 DevTools에서 다음 명령어를 통해 조회해봅시다.

GET practice_01/_search
{
  "size": 12,
  "query": {
    "match": {
      "fields.log_file_name.keyword": "access"
    }
  }
}

실제로 확인해보면 log_datetime과 datetime이 다른 것을 확인할 수 있습니다.

// ...
{
// ...
  "log_datetime" : "18/Jul/2023:13:31:44",
  "datetime" : "2023-07-18T04:31:44.000Z",
  // ...
}
// ...

이렇게 다르게 저장된 이유는 Elasticsearch에서 사용하는 표준 시간대가 UTC이기 때문입니다. 그렇기에 여기서 설정한대로 타임존을 Asia/Seoul 지정한 경우, UTC로 변환되어 저장됩니다. 이와 관련하여 공식 문서에서 다음과 같이 설명합니다.

Internally, dates are converted to UTC (if the time-zone is specified) and stored as a long number representing milliseconds-since-the-epoch.

그렇다면 Kibana에서 볼 때, 'UTC 기준으로 표출되는 것일까' 하는 의문이 들 수도 있습니다. Kibana의 경우, 브라우저 시간대를 기준으로 날짜/시간을 표출하기 때문에, 현재 설정된 브라우저 시간대에 맞게 데이터 값이 변경됩니다.

실제로 Kibana > Discover에 접속해서 확인해보면, 다음과 같이 현재 설정된 브라우저 시간대인 Asia/Seoul에 맞게 변환된 날짜와 시간이 표시되는 것을 확인할 수 있습니다.

image

6. 문자열 분리하여 저장하기

이번에는 다음과 같이 A=[B]와 같은 구조로 들어오는 로그를 A라는 필드명에 B라는 값을 가지는 형태로 분리하여 저장하는 방법을 알아보도록 하겠습니다. 여기서는 rest.log의 여러 로그 중 status=[REQUEST]로 message가 시작하는 경우에 대해서만 적용할 예정입니다.

2023-07-18 13:30:02.283 1 DEBUG [DefaultGlobalFilter.java:lambda$filter$3():68] status=[REQUEST] key=[api-key] request_id=[request-id-1] host=[api.test.com] client_ip=[123.456.789.11] method=[GET] user_agent=[Mozilla/5.0 (Linux; Android 13; SM-S908N Build/TP1A.220624.014; wv) AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/104.0.5112.97 Mobile Safari/537.36Morpheus/NT] accept=[*/*] content_type=[*/*] service=[API_GET] url=[/api/get] http_status=[100] data=[{"key":"api-key"}]
2023-07-18 13:30:12.362 1 DEBUG [DefaultGlobalFilter.java:lambda$filter$3():68] status=[REQUEST] key=[api-master-key] request_id=[request-id-2] host=[api.test.com] client_ip=[123.456.789.12] method=[GET] user_agent=[Mozilla/5.0 (Linux; Android 13; SM-G991N Build/TP1A.220624.014;) AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/111.0.5563.58 Mobile Safari/537.36 android/GA_Android/_ga_cid=null;_ga_adid=null;; android/GA_Android/_ga_adid=00000000-0000-0000-0000-000000000000;_ga_cid=gacid;;] accept=[*/*] content_type=[*/*] service=[API_POST] url=[/api/get-post] http_status=[100] data=[{"key":"api-master-key","parameter":"value","location":"37.5557651,126.9458323"}]

분리할 조건을 정리하면 다음과 같습니다.

  1. rest.log의 message 필드 내용 중 status=[REQUEST]로 시작하는 경우에 대해서만 수행
  2. A=[B]와 같은 구조일 때, A는 필드명, B는 필드 값으로 저장
  3. data의 경우, [] 내 내용이 2번 조건에 의해 분리되지 않도록 예외 처리 필요

각 조건을 만족하기 위해서는 다음과 같이 logstash config 파일를 작성해야 합니다.

# ...

filter {
	if [fields][server_name] == "practice01" {
		# ...
		else { # rest
			dissect {
				mapping => {
					"message" => "%{log_datetime} %{+log_datetime} %{} %{level} [%{code_path}] %{message}"
				}
			}
			
			mutate {
				strip => ["message"] # 공백 제거
			}

			# 1번 조건 만족
			if [message] =~ /^status=\[REQUEST\]/ {
				# 3번 조건 만족
				grok {
					match => { "message" => "^(?m)%{GREEDYDATA:request_info} data=\[%{GREEDYDATA:data}\]" }
				}
				# 2번 조건 만족
				kv {
					source => "request_info"
					field_split => " "
					value_split => "="
					target => "request_info"
				}
				mutate {
					add_field => {"[request_info][data]" => "%{data}"}
					remove_field => ["data", "message"] # 제거
				}
			}
		}
		
		# ...
	}
}

# ...

grok 플러그인을 이용해서 data= 라는 내용 앞까지 모두 request_info로 저장합니다. 이 때, ^(?m)을 적용하여 여러 문장에 대해서 적용되도록 설정하였습니다.그 다음, kv 플러그인을 이용해 request_info 내용을 분리하여 저장하도록 설정했습니다. 추가로, mutate 플러그인을 사용해 data 필드를 request_info 필드 내에 속하도록 설정하였습니다.

설정 적용을 위해 logstash와 filebeat를 재실행한 다음, DevTools에서 아래 명령어를 입력합니다. 이 때, 이전과 마찬가지로 logstash에서 생성한 인덱스와 filebeat data 디렉터리를 제거하고 진행해주세요.

GET practice_01/_search
{
  "_source": "request_info", 
  "query": {
    "bool": {
      "must": [
        {
          "term": {
            "fields.log_file_name.keyword": "rest"
          }
        },
        {
          "match": {
            "request_info.status": "REQUEST"
          }
        }
      ]
    }
  }
}

이번에는 request_info에 대해서만 출력하도록 설정하였습니다. 위 명령어를 실행하면 다음과 같이 정상적으로 request_info가 생성된 것을 확인할 수 있습니다.

image

7. json 플러그인으로 object 타입 생성하기

request_info.data 필드의 값은 json 형태로 들어옵니다. 이러한 점을 활용하여, 해당 필드를 text 타입이 아닌 object 타입으로 저장할 수 있게 json 플로그인을 사용해보겠습니다.

먼저, logstash config 파일 내에 아래와 같이 json 관련 설정을 추가합니다. (두 줄!)

# ...

filter {
	if [fields][server_name] == "practice01" {
		# ...
		else { # rest
			# ...

			if [message] =~ /^status=\[REQUEST\]/ {
				# ...
        
        # json 관련 설정 추가
				json {
					source => "data"
					target => "[request_info][data]"
				}

				mutate {
					remove_field => ["data", "message"]
				}
			}
		}
		
		# common =============
		date {
			match => ["log_datetime", "YYYY-MM-dd HH:mm:ss.SSS", "dd/MMM/YYYY:HH:mm:ss"]
			target => "datetime"
			timezone => "Asia/Seoul"
		}
	}
}

# ...

설정 적용을 위해 logstash와 filebeat를 재실행한 다음, DevTools에서 이전에 사용했던 아래 명령어를 실행합니다. 이 때, 이전과 마찬가지로 logstash에서 생성한 인덱스와 filebeat data 디렉터리를 제거하고 진행해주세요.

GET practice_01/_search
{
  "_source": "request_info", 
  "query": {
    "bool": {
      "must": [
        {
          "term": {
            "fields.log_file_name.keyword": "rest"
          }
        },
        {
          "match": {
            "request_info.status": "REQUEST"
          }
        }
      ]
    }
  }
}

그러면 다음과 같이, data 필드의 값이 object 타입으로 생성된 것을 확인할 수 있습니다.

image

참고