Filebeat의 processors

filebeat의 system 모듈은 윈도우에서 동작하지 않는다. system 모듈이 처리하는 messages나 secure 로그는 윈도우랑 관계가 없기 때문인 듯. 쓸 데 없이 엄격하네

모듈 처리를 못하니 로그스태시로 보내든가, 직접 처리해야 한다. 로그스태시는 실컷 써봤으니 로그스태시 필터 역할을 하는 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