Filebeat → Kafka 설치 및 사용 가이드

목차
1. Filebeat 설치
2. Kafka 설치
3. Filebeat 실행
4. Kafka 실행
5. 테스트

 

개요

• 서비스가 커질수록 다양한 서버에서 발생하는 로그를 효과적으로 수집하고 처리하는 것이 중요하다.
• 이번 포스팅에서는 경량 로그 수집기인 Filebeat를 활용해 Kafka로 로그를 전송하는 방법을 소개하겠다.
• Filebeat → Kafka → (Logstash or Elasticsearch) 으로 이어지는 아키텍처 구조가 일반적이다.
  ※ Kafka 이후에 Logstash, Elasticsearch dashboard로 표현하는 방법은 추후에 포스팅 하겠음. 
• Filebeat는 로그를 수집하고, Kafka는 이를 버퍼링하여 후속 처리 시스템으로 안전하게 데이터를 전달한다.

사전작업

※ 이번 포스팅은 전통적인 방식(Zookeeper 사용)의 Kafka가 아닌 신규 버전의 Kafka 구동 방식임을 명시

 

Filebeat 설치

• https://www.elastic.co/ 접속

Elastic — The Search AI Company

 

• [Search] 클릭 → filebeat 입력 후 → quick start 클릭

 

• [Install Filebeat] 부분에서 [Linux] 클릭 후 → 아래 가이드 복사 후 리눅스 서버에 입력

 

• 복사한 내용 입력 : curl -L -O https://artifacts.elastic.co/downloads/beats/filebeat/filebeat-8.13.4-linux-x86_64.tar.gz

※ 위 홈페이지는 9.0.0 버전이지만 현재 9.0.0 버전은 버그가 있으므로 이번 포스팅은 8.13.4 버전으로 하겠음.

※ curl 명령어가 없으면 : [apt install curl] 로 설치

 

• 다운받은 tar.gz 해제 후 binary로 설치된 filebeat 확인

 

Kafka 설치

※ Kafka는 Java 기반이므로  JDK가 반드시 필요함.

• [apt update] → [apt install -y default-jdk]

 

• [java --version] 으로 java 설치 확인

 

• [wget https://downloads.apache.org/kafka/4.0.0/kafka_2.13-4.0.0.tgz] 이후 tar.gz 파일을 tar -zxvf 로 해제

※ 버전 4.0 은 zookeeper가 없는 최신 버전임.

 

Filebeat, Kafka 실행

• 우리는 log성 데이터를 filebeat 가 읽게 하고 그 데이터를 kafka topic으로 보낼 것이다.
• 이와 같은 상황을 이해 하고 아래 설정을 이해해보자.

Filebeat 실행

• [vi filebeat.yml] → input과 output.kafka 를 아래와 같이 수정

filebeat.inputs:
  - type: filestream
    id: minimal-filestream
    enabled: true
    paths:
      - /data/logs/test250602.log
    tail_files: true

output.kafka:
  enabled: true
  hosts: ["localhost:9092"]
  topic: "test"
  required_acks: 1
  compression: gzip
  max_message_bytes: 1000000

logging.metrics.enabled: false
filebeat.registry.path: /data/filebeat/data/registry

 

• [./filebeat -e -c filebeat.yml] 로 filebeat 실행

 

Kafka 실행

※ kafka 4.0은 KRaft 모드 전용이므로 Zookeeper 가 완전히 제거되어 있음.

• 설치한 kafka 경로로 이동

 

• [mkdir -p /data/kafka/config/kraft] → [vi /data/kafka/config/kraft/server.properties] 에서 아래와 같이 입력 후 저장

process.roles=broker,controller
node.id=1
controller.quorum.voters=1@localhost:9093
controller.listener.names=CONTROLLER
listeners=PLAINTEXT://:9092,CONTROLLER://:9093
listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
log.dirs=/data/kafka/kraft-logs

 

• [uuidgen] 입력 이후 출력화면 보관 : ${UUID}

• [/data/kafka/bin/kafka-storage.sh format -t ${UUID} -c /data/kafka/config/kraft/server.properties] 입력

 

• [/data/kafka/bin/kafka-server-start.sh /data/kafka/config/kraft/server.properties] 으로 kafka 실행

 

테스트

• Kafka 에서 topic 을 생성한 후 Filebeat에서 생성한 topic 으로 데이터를 보내주는 방식으로 테스트를 진행하겠다.

Kafka topic 생성 후 Consumer 실행

※ kafka 실행 sh의 기본 경로는 /data/kafka/bin

※ kafka를 실행하고 있는 상태에서 진행.

• Kafka 정상작동을 위한 기본 topic 생성 : [__consumer_offsets]

   ∘ [./kafka-topics.sh --create --topic __consumer_offsets --bootstrap-server localhost:9092 --replication-factor 1 --partitions 50 --config cleanup.policy=compact]

 

• 테스트 topic 생성

   ∘ [./kafka-topics.sh --create --topic test --bootstrap-server localhost:9092 --partitions 1 --replication-factor 1]

※ [test] 라는 topic 생성.

 

• [./kafka-topics.sh --list --bootstrap-server localhost:9092] 로 [test] 토픽이 생성되었는지 확인

※ topic 삭제 방법 : ./kafka-topics.sh --delete --topic ${생성시킨 topic명} --bootstrap-server localhost:9092

 

 

Filebeat 설정

• [vi /data/filebeat/filebeat.yml] 에서 테스트할 log 위치 확인

 

• [./filebeat -e -c filebeat.yml] 로 filebeat 실행

 

Filebeat → Kafka 데이터 전송 테스트

• Kafka의 Consumer 실행

※ Procuder 명령어 : ./kafka-console-producer.sh --topic ${생성시킨 topic명} --bootstrap-server localhost:9092

   ∘ [./kafka-console-consumer.sh --topic test --from-beginning --bootstrap-server localhost:9092]

test 토픽으로 데이터가 들어오는 것을 볼 수 있는 실시간 모니터링 상태

 

• 위에 kafka consumer 를 걸어둔 채로 새로운 터미널을 열어 아래 사항 진행

• [filebeat.yml] 에서 봤던 log 경로로 이동

• [echo "박상규 박상규 테스트 25년 6월 2일!!!!!!!" >> test250602.log] 입력

 

• 좀 전에 걸어뒀던 kafka consumer 화면 확인

 

 

 

끝.