logstash 필터 dissect 성능 테스트

2019-09-05

방화벽처럼 key-value 구조가 반복되는 로그의 필드 분류에는 kv 필터가 딱이다. 특히 구분기호로 '='을 사용하는 key-value 구조는 kv {} 구문만으로도 완벽한 테이블 구조를 만들어 줌.

1
2
3
4
5
6
7
8
9
10
sip = 1.1.1.1 dip = 2.2.2.2 sport = 100 dport = 200 action = permit
sip = 1.1.1.1 dip = 2.2.2.2 sport = 100 dport = 200 action = permit
sip = 1.1.1.1 dip = 2.2.2.2 sport = 100 dport = 200 attack = 10.10.10.10
sip = 1.1.1.1 dip = 2.2.2.2 attack = 10.10.10.10
sip = 1.1.1.1 dip = 2.2.2.2 sport = 100 dport = 200 action = permit
sip = 1.1.1.1 dip = 2.2.2.2 sport = 100 dport = 200 attack = 10.10.10.10
sip = 1.1.1.1 dip = 2.2.2.2 attack = 10.10.10.10
sip = 1.1.1.1 dip = 2.2.2.2 sport = 100 dport = 200 action = permit
sip = 1.1.1.1 dip = 2.2.2.2 sport = 100 dport = 200 attack = 10.10.10.10
sip = 1.1.1.1 dip = 2.2.2.2 attack = 10.10.10.10

문제는 kv 필터가 내부적으로 정규표현식을 사용한다는 것. 성능 저하가 발생할 수 있다는 얘기. 이때 dissect 필터가 대안이 될 수 있다. dissect는 데이터 전체에 반영된 공통된 구조를 그대로 이용해서 원하는 부분을 잘라내는 기능을 제공하며, 로그는 보통 사람 읽기 좋으라고 띄어쓰기를 하기 때문에 '공백' 구조를 많이 활용한다. 로그 구조가 다음과 같을 때

1
sip = 1.1.1.1 dip = 2.2.2.2 attack = 10.10.10.10

다음 필터식을 이용하면

1
2
3
4
5
dissect {
 mapping => {
  "message" => "%{} = %{sip} %{} = %{dip} %{} = %{attack}"
 }
}

다음과 같은 테이블 구조를 만들 수 있다. %{} 구문은 데이터 검사만 하고, %{sip} 구문은 검사 결과를 sip 필드에 저장.

1
2
3
4
5
{
 "sip":"1.1.1.1"
 "dip": 2.2.2.2"
 "attack":"10.10.10.10"
}

데이터가 key-value 구조라면 더 효율적으로 테이블 구조를 만들 수도 있다. %{?same_name}구분기호%{&same_name} 구문을 사용하면 '구분기호'를 기준으로 앞에 있는 %{?same_name} 구문은 key, 뒤에 있는 %{&same_name} 구문은 value로 저장.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
input {
 file {
  path => "D:/test.log"
  start_position => "beginning"
  sincedb_path => "nul"
 }
}

filter {
 #kv {}

 ruby {
  code => "event.set('field_count', event.get('message').split('=').count() - 1)"
}

 if [field_count] == 5 {
  dissect {
   mapping => {
    "message" => "%{?1} = %{&1} %{?2} = %{&2} %{?3} = %{&3} %{?4} = %{&4} %{?5} = %{&5}"
   }
  }
 } else {
   dissect {
    mapping => {
     "message" => "%{?1} = %{&1} %{?2} = %{&2} %{?3} = %{&3}"
   }
  }
 }
}

output {
 file {
  path => "d:/test-dissect.log"
 }
}