로그스태시 mutate filter plugin 파헤치기

raw data를 Elasticsearch로 밀어넣기 전에, 사용하기 편하도록 가공이 필요하다.

이는 Logstash의 filter로 해결할 수 있는데, 특정 필드를 가공하는 방법 중에 mutate filter plugin 가 있다.

해당 플러그인은 공식문서에 설명이 잘 되어있다.

( 공식문서에서의 mutate filter plugin 의 gsub 옵션에 대한 설명이다. )

공식문서에는 위와 같은 식으로 여러 옵션들이 친절하게 설명되어 있다.

하지만 예시가 없어서 개인적으로 한 눈에 와닿지가 않았었는데, 공부도 할 겸 하나씩 사용해보며 예시를 남기고자 한다.

이 글이 mutate filter plugin 를 오랫만에 접할 미래의 나와 **`mutate filter plugin 를 처음 학습하는 사용자에게 조금이나마 도움이 되었으면 한다.

---

Mutate filter plugin

mutate filter plugin 의 옵션으로는 아래와 같다.

1. convert
2. copy
3. gsub
4. join
5. split
6. lowercase, uppercase
7. capitalize
8. merge
9. coerce
10. update
11. replace
12. strip
13. rename

 

1. convert

해당 필드를 형변환한다.

 

logstash.conf

input {
  stdin{
    codec => json
  }
}

filter {
  mutate {
    convert => {
      "field1" => "integer"
      "field2" => "integer_eu"
      "field3" => "float"
      "field4" => "float_eu"
      "field5" => "string"
      "field6" => "boolean"  
    }
  }
}

output {
  stdout{}
}

 

1) “field1” => “integer”

 

2) “field2” => “integer_eu”

3) “field3” => “float”

4) “field4” => “float_eu”

5) “field5” => “string”

6) “field6” => “boolean”

• false: 0, 0.0, "0", "0.0", "false", "f", "no", "n"

• true: 1, 1.0, "1", "1.0", “true”, "t", "yes", "y"

 

( convert의 경우에는 예시가 공식문서에 아주 잘 나와 있었다. )

 

 

2. copy

필드의 값을 복제한다.

 

logstash.conf

input { ... }

filter {
  mutate {
    copy => {
      "field1" => "field1_copied"
    }
  }
}

output { ... }

 

input / output

 

 

3. gsub

정규식으로 일치하는 항목을 다른 문자열로 대체한다.

 

logstash.conf

input { ... }

filter {
  mutate {
    gsub => [
      "field1", "/", "_",
      "field2", "[\\?#-]", "."
    ]
  }
}

output { ... }

 

input / output

• field1: / => _

• field2: ?, #, - => .

 

 

4. join

배열을 하나의 문자로 합친다.

 

logstash.conf

input { ... }

filter {
   mutate {
     join => { "field1" => "," }
   }
 }

output { ... }

 

input / output

 

 

5. split

하나의 문자를 배열로 쪼갠다.

 

logstash.conf

input { ... }

filter {
  mutate {
    split => { "field1" => "," }
  }
}

output { ... }

 

input / output

구분자가 해당 필드에 존재하지 않는 경우에도 필드는 배열로 바뀌는 것을 알 수 있다.

join 옵션과 반대 개념이다.

 

 

6. lowercase, uppercase

소문자(대문자) 로 변경

 

logstash.conf

input { ... }

filter {
  mutate {
    lowercase => [ "field1", "field3" ]
    uppercase => [ "field4" ]
  }
}

output { ... }

 

input / output

 

 

7. capitalize

첫 글자를 대문자화, 이외는 소문자화 한다.

 

logstash.conf

input { ... }

filter {
  mutate {
    capitalize => [ "field1" ]
  }
}

output { ... }

 

input / output

 

 

8. merge

해당 필드 값들을 다른 필드에 포함시킨다.

 

logstash.conf

input { ... }

filter {
  mutate {
    merge => { "field2" => "field1" }
  }
}

output { ... }

 

input / output

• "banana" + ["apple", "melon"] => ["apple", "melon", "banana"]

• ["banana", "mango"] + ["apple", "melon"] => ["apple", "melon", "banana", "mango"]
• "banana" + "apple" => ["banana", "apple"]
• ["banana", "mango"] + “apple” => ["apple", "banana", "mango"]

 

 

9. coerce

필드 값이 null인 경우에 기본값을 넣어준다.

 

logstash.conf

input { ... }

filter {
  mutate {
    coerce => { "field2" => "empty" }
  }
}

output { ... }

 

input / output

단, input에 해당 필드가 key로 존재하지 않는 경우에는 기본값을 넣지 않는다. ( null 값이 있는 경우에는 대체 함.)

ex. input이 { "username": "bactoria" } 인 경우 field1 필드가 생성되지 않음.

 

 

10. update

해당 필드의 값을 특정값으로 대체한다.

 

logstash.conf

input { ... }

filter {
  mutate {
    update => { "message" => "%{source_host}: My new message" }
  }
}

output { ... }

 

input / output

input에서 message 필드가 들어오지 않는 경우 message 필드를 생성하지 않는다. (replace옵션은 필드를 생성함)

ex) input이 { } , { "source_host": "54.17.150.184" } 인 경우 field1 필드가 생성되지 않았다.

 

 

11. replace

해당 필드의 값을 특정값으로 대체한다.

 

logstash.conf

input { ... }

filter {
  mutate {
    replace => { "message" => "%{source_host}: My new message" }
  }
}

output { ... }

 

input / output

input에서 message 필드가 들어오지 않는 경우 message 필드를 생성하여 대체한다. (update옵션은 생성하지 않음)

ex) input이 { } , { "source_host": "54.17.150.184" }인 경우 message 필드가 대체되었다.

 

 

12. strip

좌우 공백을 제거한다.

 

logstash.conf

input { ... }

filter {
  mutate {
      strip => ["searchKeyword", "username"]
  }
}

output { ... }

 

input / output

trim() 함수와 같은 기능이다.

참고로 단어 사이의 공백은 제거되지 않는다. ex) 박토리 아

 

 

13. rename

해당 필드의 이름을 변경한다.

 

logstash.conf

input { ... }

filter {
  mutate {
    rename => { "HOSTORIP" => "client_ip" }
  }
}

output { ... }

 

input / output

 

 

---

P.S) mutate 옵션 실행 순서

mutate의 옵션에는 실행순서가 있다.

 

logstash.conf

input { ... }

filter {
  mutate {
    copy => {
      "field" => "field_copied"
    }
    uppercase => [ "field" ]
  }
}

output { ... }

input / output

 

위의 결과는 순차적으로 처리된다면 { "field": "ABCD", "field_copied": "abcd" } 일 것이다.

공교롭게도 결과는 { "field": "ABCD", "field_copied": "ABCD" } 이다.

 

그 이유는 옵션 별로 실행 순서가 정해져 있어 uppercase 옵션 실행 이후에 copy 옵션이 실행되기 때문이다.

실행 순서에 대해서는 공식문서에 나와있다.

 

만약 copy 옵션을 먼저 실행하고자 한다면 어떻게 해야 할까?

아래와 같이 mutation을 따로 선언해주면 copy 옵션 실행 이후에 uppercase 옵션이 실행된다.

 

logstash.conf

input { ... }

filter {
  mutate {
    copy => {
      "field" => "field_copied"
    }
  }
  mutate {
    uppercase => [ "field" ]
  }
}

output { ... }

 

input / output