Filebeat의 processors

filebeat의 system 모듈은 윈도우에서 동작하지 않는다. system 모듈이 처리하는 messages나 secure 로그는 윈도우랑 관계가 없기 때문인 듯. (8.13 버전에서는 됨)

모듈 처리를 못하니 로그스태시로 보내든가, 직접 처리해야 한다. 로그스태시는 실컷 써봤으니 로그스태시 필터 역할을 하는 filebeat의 processors 옵션을 사용해보자. 다음은 input 설정.

filebeat.inputs:
- type: log
  enabled: true
  paths:
    - d:\test.log

일단 processors 옵션을 사용하지 않은 연동 결과는 이렇다. message 필드가 제일 중요.

D:\ELK\filebeat-7.12.0-windows-x86_64>filebeat.exe
{
  "@timestamp": "2021-04-18T13:17:54.794Z",
  "@metadata": {
    "beat": "filebeat",
    "type": "_doc",
    "version": "7.12.0"
  },
  "ecs": {
    "version": "1.8.0"
  },
  "host": {
    "name": "MHKANG"
  },
  "agent": {
    "ephemeral_id": "f541a72b-7f2d-4787-9a11-7a89ac4e2f8e",
    "id": "f3733808-8148-48d9-8272-090481cef05c",
    "name": "MHKANG",
    "type": "filebeat",
    "version": "7.12.0",
    "hostname": "MHKANG"
  },
  "log": {
    "file": {
      "path": "d:\\test.log"
    },
    "offset": 0
  },
  "message": "Jan  9 17:31:21 Sensor sshd[1854]: Accepted password for root from 192.168.56.1 port 1939 ssh2",
  "input": {
    "type": "log"
  }
}

다음은 시계열 분석을 위해 필수인 timestamp 추출을 위한 processor 설정. dissect processor의 동작 방식은 로그스태시 dissect 필터와 똑같음.

processors:
  - drop_fields:
      fields: ["log", "input", "ecs", "host", "agent"]
  - dissect:
      tokenizer: "%{timestamp->} %{+timestamp} %{+timestamp} %{}"
      target_prefix: ""

실행 결과는 다음과 같다.

{
  "@timestamp": "2021-04-18T13:26:52.068Z",
  "@metadata": {
    "beat": "filebeat",
    "type": "_doc",
    "version": "7.12.0"
  },
  "timestamp": "Jan 9 17:31:21",
  "message": "Jan  9 17:31:21 Sensor sshd[1854]: Accepted password for root from 192.168.56.1 port 1939 ssh2"
}

timestamp를 추출했으니 해당 필드 정보를 엘라스틱이 시간 필드로 인식하게끔 date 유형으로 바꿔줘야 한다. 제일 간단한 방법은 @timestamp에 덮어씌우는 것. 로그스태시 date 필터 역할을 하는 timestamp processor 추가.

processors:
  - drop_fields:
      fields: ["log", "input", "ecs", "host", "agent"]
  - dissect:
      tokenizer: "%{timestamp->} %{+timestamp} %{+timestamp} %{}"
      target_prefix: ""
  - timestamp:
      field: timestamp
      layouts:
        - '2006 Jan 2 15:04:05'

그런데 변화가 없다. secure 로그는 연도 정보를 기록하지 않을 때가 있는데, 없는 정보를 알아서 채워주는 로그스태시 date 필터와는 달리 filebeat의 timestamp processor는 처리를 못한다.

{
  "@timestamp": "2021-04-18T13:39:51.851Z",
  "@metadata": {
    "beat": "filebeat",
    "type": "_doc",
    "version": "7.12.0"
  },
  "message": "Jan  9 17:31:21 Sensor sshd[1854]: Accepted password for root from 192.168.56.1 port 1939 ssh2",
  "timestamp": "Jan  9 17:31:21"
}

엘라스틱이 다 좋은데 오픈소스 연합(?) 성격이라 그런지 데이터 전처리 과정이 노드, 로그스태시, beat 다 다름. 다음은 샘플 로그에 연도 정보를 추가한 후, 바뀐 데이터에 맞게 dissect processor 수정 및 재실행 결과.

processors:
  - drop_fields:
      fields: ["log", "input", "ecs", "host", "agent"]
  - dissect:
      tokenizer: "%{timestamp} %{timestamp->} %{+timestamp} %{+timestamp} %{}"
      target_prefix: ""
  - timestamp:
      field: timestamp
      layouts:
        - '2006 Jan 2 15:04:05'

{
  "@timestamp": "2010-01-09T17:31:21.000Z",
  "@metadata": {
    "beat": "filebeat",
    "type": "_doc",
    "version": "7.12.0"
  },
  "message": "2010 Jan  9 17:31:21 Sensor sshd[1854]: Accepted password for root from 192.168.56.1 port 1939 ssh2",
  "timestamp": "2010 Jan 9 17:31:21"
}

주의사항

timestamp processor의 layouts 옵션값은 반드시 '2006-01-02 15:04:05' 시간대를 연동 대상 데이터의 timestamp 포맷으로 입력해야 한다. 1초만 틀려도 제대로 동작하지 않음. 초단위만 살짝 바꿔도,

processors:
  - drop_fields:
      fields: ["log", "input", "ecs", "host", "agent"]
  - dissect:
      tokenizer: "%{timestamp} %{timestamp->} %{+timestamp} %{+timestamp} %{}"
      target_prefix: ""
  - timestamp:
      field: timestamp
      layouts:
        - '2006 Jan 2 15:04:04'

이렇게 된다.

{
  "@timestamp": "2010-01-09T17:21:00.000Z",
  "@metadata": {
    "beat": "filebeat",
    "type": "_doc",
    "version": "7.12.0"
  },
  "timestamp": "2010 Jan 9 17:31:21",
  "message": "2010 Jan  9 17:31:21 Sensor sshd[1854]: Accepted password for root from 192.168.56.1 port 1939 ssh2"
}

beat가 GO 언어로 개발됐다는데, 해당 시간대가 GO 내부적으로 시간 정보를 처리하는 기준? 뭐 그런 거인듯. 잘 모르겠음

관련 글

• Filebeat의 processors - 2nd