EFK(ELK) 구축 해보기!! (서버 로그 수집) - 5. Fluentd에서 필터링 & OpenSearch로 전송하기

Fluentd에서 필터링 후 전송하기!!

다시 서버1로 돌아와 Fluentd 필터링과 전송을 위한 세팅을 진행한다.

우선 Opensearch서버로 로그를 보내기 위해서 plugin 설치가 필요하다.

sudo fluent-gem install fluent-plugin-opensearch

그리고 
다시 fluent.conf 파일을 vi로 연다.

인풋에 대한 태그인 source 밑에

output 설정으로 위한 태그인 match를 아래처럼 추가한다.

Opensearch의 index lifecycle을 시간으로 가져가기 때문에
log에 담긴 시간 값에 맞는 index로 전송할 수 있다.

<match log.json.**>
  @type opensearch
  hosts $opensearch_server_ip:9200
  logstash_format true
  logstash_prefix json-timelog
  include_timestamp true
  time_key datetime
  time_key_format %d/%b/%Y:%H:%M:%S %z
</match>

위는 json 형식의 로그이며
log.json.** -> log.apache.**로 바꿔주고 index_name도 변경해주면
Apache 형식의 로그도 보낼 수 있다.

@type 뒤의 내용은 플러그인이고
hosts 뒤에는 자신의 2번째 서버의 public ip 주소를 넣는다.

logstash에서 주로 쓰이던 패턴이기 때문에에 logstash_format을 true로 설정한다.

자동으로 prefix에 logstash가 붙기에 이를 json으로 변경한다.

include_timestamp를 true로 설정해서 kibana(openDashboard)의 @timestamp로 매핑한다.

time_key를 datetime으로 설정하고 포맷을 알려줘야 파싱이 가능하다.

필터링하기

같은 파일 안에서 match태그 위에 filter 태그를 넣는다. (output 만들어지기 전에 필터링 먼저!)

예를 들어 잘된 200번대 status만 제외하고 수집하려면 아래와 같은 patter을 설정하면 된다.

<filter log.**>
  @type grep
  <exclude>
    key status
    pattern /^[2][0-9][0-9]/
  </exclude>
</filter>

exclude가 아닌 원래대로 적으려면

filter log.**>
  @type grep
  <regexp>
    key status
    pattern /^[1345][01235][0-9]/
  </regexp>
</filter>

이렇게 적을 수도 있다.

위에서 둘 중 하나를 넣어주고

record_transformer 플러그인으로 필드를 추가하거나 삭제할 수 있다.

<filter log.**>
  @type record_transformer
  <record>
    sent_by fluentd
    ftag ${tag}
  </record>
  remove_keys host
</filter>

remove_keys는 필드 이름을 찾아 삭제하며 위에서는 host 정보를 삭제해주는 것과 비슷한 역할을 한다.

여기서 중요한 것은 순서이다.
필터가 반드시 match 위에 와야하며
필터들도 무엇이 먼저왔냐에 따라 다른 값을 낼 수도 있다.

이전에 실행해서 positions~ 파일들이 있다면
삭제하고 실행시켜주는 것이 안전하다.

rm positions-*

fluentd -c ./fluent.conf -vv

 

---

OpenSearch 과정들은 이전 포스팅에서 진행했으니

바로 Open Dashboard로 넘어간다.

이전 포스팅에서 접속한
대시보드가 떠있는 http 웹페이지로 넘어간다.

거기서 메뉴의
Management > Stack Management 에서

Index Pattern > Create Index Patterns 로 들어간다.

*(asterisk)를 이용해서 인덱스들을 묶는 인덱스 패턴을 생성하고,
discover, visualize에서 해당 인덱스 패턴을 이용해서 검색하고 규칙을 정한다.

json-timelog* 을 우선 넣어주고 Next

그리고 timestamp를 가지고 있기 때문에 step2에서 @timestamp를 선택한다.
> create index patter 클릭

이후 새로고침 후 표를 보면
각 필드를 보고 타입을 추론한 뒤에 보여준다.

이후 Dashboard 와 Visualize에서 이것저것 눌러보며 완성해보았다.

완성!!