Fluentd 개념 정리

Fluentd 란 ?

• 로그를 수집할 수 있는 로그 수집기 입니다.
• Fluentd에서는 서버에 쌓이고 있는 log 파일을 지정하여 로그를 수집하도록 할 수 있습니다. 또는 http, tcp 통신하여 Fluentd로 직접 전송하는 방식도 사용할 수 있습니다.
• 이렇게 Fluentd에서 수집한 로그를 가공(로그에 값 추가 혹은 삭제, 필터 등) 할 수 있으며, 이를 파일이나 데이터베이스로 저장하거나 전송할 수 있습니다.

---

Fluentd를 이용한 로그 수집 아키텍처

1. 기동되고 있는 서버에서 로그를 수집하고 중앙 로그 저장소로 전송

2. 각 서버에서 Fluentd가 수집한 로그를 다른 Fluentd로 보내서 이 Fluentd가 최종적으로 로그 저장소에 저장

• fluent가 앞에서 들어오는 로그들을 수집해서 로그 저장소에 넣기 전에 로그 트래픽을 Throttling(속도 조절)을 해서 로그 저장소의 용량에 맞게 트래픽을 조정 가능하도록 하기 위해

3. 로그를 여러 개의 저장소에 복제해서 저장하고나 로그의 종류에 따라서 각각 다른 로그 저장소로 라우팅

---

Fluentd 내부 구조

Input, Parser, Engine, Filter, Buffer, Output, Formatter 7개의 컴포넌트로 구성

• 일반적인 흐름은 Input → Engine → Output이고,
• Parser, Buffer, Filter, Formatter 등을 설정에 따라 선택적으로 추가 또는 삭제 가능
  
  

1. Input

• 수집할 log를 내부에서 input plugin(tail, forward, http등)을 사용해서 가져옵니다.
• 다양한 서버나 애플리케이션으로 부터 다양한 포맷의 데이터를 수집할 수 있습니다.

@tail

• path 에 지정한 log를 읽어서 사용합니다. 리눅스 tail 명령어와 같이, 파일의 끝부분을 지속적으로 읽습니다.

<source>
  @type tail
  path /var/log/httpd-access.log
  pos_file /var/log/td-agent/httpd-access.log.pos
  tag apache.access
  <parse>
    @type apache2
  </parse>
</source>

@forward

• 데이터를 전달받기 위해 사용하며, 서버간에 TCP로 로그 데이터를 수신할 수 있습니다.
• forward로 전달되는 데이터는 json 이나 messagepack 형식으로 되어 있습니다.

<source>
  @type forward
  port 24224
  bind 0.0.0.0
</source>

@http

• curl 같은 명령어로 로그데이터를 http 요청으로 받을 수 있습니다.

<source>
  @type http
  port 9880
  bind 0.0.0.0
  body_size_limit 32m
  keepalive_timeout 10s
</source>    

2. filter (Optional)

• Filter 플러그인을 읽어들인 데이터를 Output으로 보내기 전에, 다음과 같은 3가지 기능을 합니다.

  • 필터링
  • 데이터 필드 추가
  • 데이터 필드 삭제 또는 특정 필드 마스킹

• <filter></filter> 내부에서 filter plugin(record_transformer, grep 등)을 사용합니다.

@record_transformer

• output으로 보내기 전에 source 의 log에 hostname을 key 로, #{Socket.gethostname} 를 value 로 추가합니다.

<filter foo.bar>
  @type record_transformer
  <record>
    hostname "#{Socket.gethostname}"
  </record>
</filter>

@grep

• <source> 에서 tar가 foo.bar 로 설정되어 읽어온 log 중에,
  • message filed의 값에 cool 이 있는 경우 만 grep 해서 output으로 보냅니다.
  • hostname field 값이 web.example.com 과 일치하는 경우에만 output으로 보냅니다.
  • message field 값에 uncool이 없는 경우에만 output으로 보냅니다.

<filter foo.bar>
  @type grep

  <regexp>
    key message
    pattern /cool/
  </regexp>

  <regexp>
    key hostname
    pattern /^web\d+\.example\.com$/
  </regexp>

  <exclude>
    key message
    pattern /uncool/
  </exclude>
</filter>

@parser

• expression 의 정규식에 일치하는 log가 있는 경우 오른쪽 정규식 내용으로 변경해서 output 으로 보냅니다.

<filter foo.bar>
  @type parser
  key_name log
  <parse>
    @type regexp
    expression /^(?<host>[^ ]*) [^ ]* (?<user>[^ ]*) \[(?<time>[^\]]*)\] "(?<method>\S+)(?: +(?<path>[^ ]*) +\S*)?" (?<code>[^ ]*) (?<size>[^ ]*)$/
    time_format %d/%b/%Y:%H:%M:%S %z
  </parse>
    </filter>

3. buffer (Optional)

• Input에서 들어온 데이터를 바로 Output으로 보내서 쓰는 것이 아니라 중간에 선택적으로 Buffer를 둬서 Throttling을 할 수 있습니다.
• 버퍼는 File과 Memory 2가지를 사용할 수 있습니다.
• <match></match> 에서 <buffer></buffer>를 사용하여 input 에서 들어온 log를 특정데이터 크기 제한까지 도달하면 output으로 보내도록 할 수 있습니다.

[Buffer 구조와 동작 원리]

• Buffer에는 로그 데이터를 분리하는 Tag 단위로 Chunk가 생성이 됩니다.

• Chunk에 데이터가 쌓여서 buffer_chunk_limit만큼 Chunk가 쌓여서 full이 되거나, 설정값에 의해 정의된 flush_interval 주기가 되면 로그 저장소로 로그를 쓰기 위해서 Queue에 전달이 됩니다.

• 다음 Queue에서는 데이터를 읽어서 로그 저장소에 데이터를 쓰는데, 로그 저장소에 문제가 없다면 바로 로그가 써지겠지만 네트워크 에러나 로그 저장소 에러로 로그를 쓰지 못할 경우에는 retry_wait 시간 만큼 대기를 한 후에 다시 쓰기를 시도합니다.
• 만약에 Queue가 차버렸을 때 처리에 대한 정책을 설정할 수 있는데, “exception”과 “block” 모드 2가지가 있습니다.
• Exception 모드일 경우에는 BufferQueueLimitError를 내도록 하고,
• Block 모드의 경우 BufferQueueLimitError가 해결될 때까지 Input Plugin을 중지 시킵니다. (로그 수집 X)

• Queue가 차버렸을 때 다른 처리 방법으로는 큐가 다 찼을 때, Secondary output을 지정해서, 다른 로그 저장소에 로그를 저장하는 방법이 있을 수 있습니다.

  ex) secondary output을 AWS S3로 지정해 놓고, S3로 로그를 일단 저장하게 하고 나중에 mongodb가 복구된 후에, S3에서 다시 mongodb로 로그를 수집하는 방식을 취할 수 있습니다.

4. output

• 출력할 log를 <match></match> 내부에서 output plugin(file, stdout, forward등)을 사용해서 출력합니다.

@file

• log를 지정한 path 에 gzip 으로 압축하여 출력합니다.

<match pattern>
  @type file
  path /var/log/fluent/myapp
  compress gzip
  <buffer>
    timekey 1d
    timekey_use_utc true
    timekey_wait 10m
  </buffer>
</match>

@stdout

• Fluentd 에서 표준출력(stdout)으로 출력합니다.

<match **>
    @type stdout
</match>

@forward

• 다른 Fluentd 노드로 로그를 전달할 때 사용되며, 반드시 1개 이상을 포함해야 합니다.

<match **>
  @type forward

  <server>
    name another.fluentd1
    host 127.0.0.1
    port 24224
    weight 60   # 로드밸런싱 가중치 설정 
  </server>
  <server>
    name another.fluentd2
    host 127.0.0.1
    port 24225
    weight 40
  </server>
</match>

5. Parser (Optional)

• Input 플러그인을 통해서 들어온 데이터를 읽어도 데이터 포맷이 Fluentd에서 지원하지 않는 데이터 포맷인 경우가 있기 때문에 이를 파싱하기 위해 Parser 플러그인을 선택적으로 사용할 수 있습니다.
  (Regular Expression, Apache, Nginx, Syslog, ..)

• <source><source>, <match><match>, <filter></filter> 안에서 <pasre></parse> 형식으로 사용합니다.

• 전달받은 데이터를 파싱하기 위해서 @type 파라미터로 regexp, apache2, apache_error, nginx, syslog, csv, tsv, ltsv, json, multiline, none 과 같은 Parser plugin을 사용합니다.

6. Formatter (Optional)

• Output 플러그인을 통해 데이터를 저장소에 쓸 때, Formatter를 이용하면 데이터의 포맷을 정의할 수 있습니다.
  ex) Input의 parser가 포맷에 맞게 읽는 플러그인이라면, Formatter는 Output을 위한 포맷을 지정하는 플러그인)

• <match></match> 안에서 <format></format> 형식으로 사용합니다.

• 출력 형식을 변경 할 수 있습니다. @type 파라미터로 out_file, json, ltsv, csv, msgpack, hash, single_value, tsv 와 같은 Formatter plugin 을 사용합니다.

---

데이터 구조

Fluentd가 내부적으로 어떻게 로그 데이터를 핸들링하는지 데이터 구조를 확인해보겠습니다.

데이터는 크게 3가지 파트로 구성이 됩니다.

• Time : 로그데이터의 생성 시간
• Record : 로그 데이터의 내용으로 JSON 형태로 정의
• Tag : 데이터의 분류로 로그 레코드는 tag를 통해 로그의 종류가 정해지는데, 이 tag에 따라서 로그에 대한 필터링, 라우팅과 같은 플러그인이 적용이 된다.

Example

/etc/td-agent/td-agent.conf

<ROOT>
  <match td.*.*>
    type tdlog
    apikey xxxxxx
    auto_create_table
    buffer_type file
    buffer_path /var/log/td-agent/buffer/td
    <secondary>
      type file
      path /var/log/td-agent/failed_records
      buffer_path /var/log/td-agent/failed_records.*
    </secondary>
  </match>
  <match debug.**>
    type stdout
  </match>
  <source>
    type forward
  </source>
  <source>
    type http
    port 8888
  </source>
  <source>
    type debug_agent
    bind 127.0.0.1
    port 24230
  </source>
</ROOT>

참고

• https://minimilab.tistory.com/61
• https://nyyang.tistory.com/120