0. Summary

Elastic Stack 을 구축을 하고 Kibana에서 모니터링을 할 때 서비스, 서버 구분을 위해 Logstash에서 Multi Topic 설정과 Filebeat Fields를 추가하는 방법을 기록하려고 한다.

1, Architecture

• Elastic Stack (ELK + Filebeat)
  • elastic version 8.5
• kafka
  • 2.13-3.3.1
• AWS EC2 4대
  • t1 서비스 - t1-dev1 (Spring Boot)
  • t1 서비스 - t1-dev2 (Spring Boot)
  • t2 서비스 - t2-dev1 (Spring Boot)
  • Elastic Stack + kafka

2, Kafka Topics으로 서비스 구분

먼저 서비스를 구분하려고 한다.
서비스 별로 topic을 만들들어서 구분할것이다.

• t1, t2 서비스 모두 partition 3, replication-factor 1 으로 세팅

  • 운영에서는 replication-factor 는 3 이상으로 설정하는 것이 좋음(홀수로 설정)

• t1 서비스 topic 만들기

  ./kafka-topics.sh \
  > --bootstrap-server localhost:9092 \
  > --topic t1-logs \
  > --create \
  > --partitions 3  \
  > --replication-factor 1 \

• t2 서비스 topic 만들기

  ./kafka-topics.sh \
  > --bootstrap-server localhost:9092 \
  > --topic t2-logs \
  > --create \
  > --partitions 3  \
  > --replication-factor 1 \

• logstash 설정

  • decorate_events => true 로 설정해야 @metadata의 값을 사용할 수 있다.
  • output에서 topic별로 index를 따로 만들어서 관리함

  input {
      kafka {
          bootstrap_servers => "localhost:9092"
          group_id => "logstash"
          topics => ["t1-logs", "t2-logs"]
          consumer_threads => 2
          decorate_events => true
          codec => json { }
      }
  }
  
  output {
      if [@metadata][kafka][topic] == "ray-dev-logs" {
          elasticsearch {
              hosts => ["localhost:9200"]
              index => "t1-log-%{+YYYY.MM.dd}"
              user => "elastic"
              password => "changeme"
          }
      }
  
      else if [@metadata][kafka][topic] == "agent-logs" {
          elasticsearch {
              hosts => ["localhost:9200"]
              index => "t1-log-%{+YYYY.MM.dd}"
              user => "elastic"
              password => "changeme"
          }
      }
  }

3, 서버 구분

서버를 구분하기 위해서 생각한 방법은 2가지
I) logstash 에서 server.name 필드 추가
II) filebeat 에서 server.name 필드 추가

I) logstash으로 서버 구분

• logstash 에서 구분하기 위해서는 host.ip에 해당 아이피가 들어 있는지 여부를 확인해서 필드를 추가한다.
• /etc/logstash/conf.d/logstash.conf 의 filter 부분 (ec2: Elastic Stack)

  if "172.x.x.11" in [host][ip] {
      mutate {
          add_field => ["server.name", "dev1"]
      }
  }
  else if "172.x.x.12" in [host][ip] {
      mutate {
          add_field => ["server.name", "dev2"]
      }
  }

II) filebeat으로 서버 구분

• 위 I) logstash에서 서버 구분 에서 filter 설정 제거

• ec2

  • t1-dev1
  • t1-dev2
  • t2-dev1

• /etc/filebeat/filebeat.yml

  filebeat.inputs:
  - type: log
    enabled: true
    paths:
      - /home/ec2-user/.../*.log # 로그파일 위치
  
    multiline.pattern: '^\d{4}-\d{2}-\d{2}'
    multiline.negate: true
    multiline.match: after
  
  # ...
  
  output.kafka:
    hosts: ["kafka ip:9092"]
    topic: "t1-logs" # t2-dev1
    compression: gzip
  
  # ...
  
  processors:
    # ...
    - add_fields:
        target: server
        fields:
          name: "t1-dev1" # or t1-dev2, t2-dev1

• /etc/filebeat/fields.yml

  • fields에 추가하지 않으면 kibana 에서 type이 Unknown으로 나타남

  # ...
  - key: beat
    anchor: beat-common
    title: Beat
    description: >
      Contains common beat fields available in all event types.
    fields:
      - name: server.name
        type: text
      #...

III) logstash vs filebeat

logstash나 filebeat로 서버 명 필드를 추가를 진행해 보았다.
두 설정 모두 장단점이 존재한다. 나는 이 중 filebeat에서 설정하는 것을 선택했다.

• 차이점

  	Logstash	Filebeat
  서버명 필드 추가 방법	logstash.conf 파일에서 서버 IP로 구분하여 서버명 필드를 추가	filebeat.yml 파일에서 processors 에 필드 추가
  문법 오류 발생	모든 서버 로그 수집 불가	오류난 해당 서버 로그 수집 불가

• 결론: 서버 IP로 구분해서 조건문을 타는 것 보다 log를 보낼때 필드를 추가하는 것이 더 좋은 방법 인것 같고, 오류난 서버만 로그 수집이 안되는 것이 더 좋은 방법 일것 같아서 filebeat 에서 설정하는 것을 택함

5, Reference

• [로그수집] 서비스 별로 다른 저장소에 로그를 적재하고 싶다면 (링크)
• logstash와 kafka 연동시 Multiple Topic 사용하기 (링크)