Sevity Blog

kafka란

RabbitMQ와 같은 메시지 미들웨어.

LinkedIn에서 개발되어 현재는 Apache Software Foundation의 일부인 오픈 소스 메시지 스트리밍 플랫폼.
대량의 실시간 데이터 스트리밍을 처리하는 데 초점을 맞추고 있고, 높은 처리량, 데이터 복제, 분산 처리 등을 지원

이를 통해 대규모 데이터를 실시간으로 처리할 수 있다.

MSA간 통신에도 자주 쓰임

브로커: Kafka에서 메시지를 처리하는 서버

• 카프카에서는 이벤트를 구분하기 위한 단위로 '토픽'을 사용하며 파일시스템의 디렉토리와 비슷한 개념으로 이해 가능
  • 토픽은 게시/구독 메시징 시스템에서 흔하게 볼 수 있다.
    • 카프카 외적으로는 비슷한 개념에 대해 채널, 큐, 스트림이라는 용어를 사용하기도 함
      • 개인적으로는 채널이 더 직관적이네
  • 프로듀서는 데이터를 '토픽'에 넣고, 컨슈머는 '토픽'에서 데이터를 읽는다.
  • 이러한 토픽은 카프카 클러스터내에서 데이터를 분류하고 구독자가 관심있는 메시지만 구독할 수 있도록 해주는 중요한 역할
  • 카프카 클러스터 전체가 하나의 토픽만 사용한다면 생략도 가능한가? > NO.
• 토픽의 데이터는 파티션에 분산 저장되며 각 파티션은 순서 유지가 되는 메시지의 로그로 구성
• 하나의 브로커는 여러 파티션의 데이터를 저장할 수 있으며, 반대로 하나의 파티션도 여러 브로커에 복제될 수 있음에 주의
• 파티션의 개수는 토픽의 병렬처리 능력을 결정

리더 브로커: 특정 메시지그룹(파티션)의 처리를 담당하는 서버. Kafka에서 각 데이터 파티션에는 하나의 리더 브로커가 있음

• 리더 브로커는 파티션과 1:1의 관계를 가지며 하나의 파티션을 여러 브로커가 처리할경우 하나만 리더이고 나머지는 팔로워로 동작

ZooKeeper

• 일종의 비서역할로 서버들의 상태관리나 역할분배를 Kafka를 위해서 수행함. Kakfa설치시 같이 설치됨.
• 또는 일종의 '중앙 데이터 저장소' 역할. Kafka 시스템 내의 여러 서버들이 ZooKeeper를 통해 필요한 정보 공유하고 동기화.
• 특정 서버에 문제가 생기면 이 정보가 ZooKeeper에 기록되고 다른 서버들이 이를 확인하여 적절히 대응

kafka설치과정

Java설치

카프카는 Java기반이므로 java jdk가 안깔렸으면 깔아준다.

sudo apt update
sudo apt install default-jdk

사용자계정에 종속된 설치보다는 여러명이 쓸 수 있도록 아래 처럼 전용 유저를 생성해주는게 좋다.

sudo adduser kafka
sudo adduser kafka sudo

카프카는 자바 기반이므로 jar파일을 받아서 설정해준다. apt등의 패키지 매니저를 사용하면 예전버전일수 있으므로 다운로드해서 해보자.

mkdir ~/Downloads
curl "https://downloads.apache.org/kafka/3.5.1/kafka_2.13-3.5.1.tgz" -o ~/Downloads/kafka.tgz
mkdir /opt/kafka && cd /opt/kafka
tar -xvzf ~/Downloads/kafka.tgz --strip 1


#kafka디렉토리 소유권을 kafka사용자로 변경
sudo useradd kafka -m
sudo chown -R kafka:kafka /opt/kafka

/opt/kafka에 설치했으며, /opt 디렉토리는 선택적인(add-on) 애플리케이션 SW패키지를 위한 곳으로 여러사용자가 이용하는 라이브러리인 경우 선호되는 경로다.

zookeeper설정

zookeeper는 kafka와 독립적으로 실행되며, 비슷한 설정 매커니즘을 갖는다.

zookeeper설정파일 수정

sevity@sevityubuntu:/opt/kafka/config$ cat zookeeper.properties
# the directory where the snapshot is stored.
dataDir=/var/cache/zookeeper
# the port at which the clients will connect
clientPort=2181
# disable the per-ip limit on the number of connections since this is a non-production config
maxClientCnxns=0
# Disable the adminserver by default to avoid port conflicts.
# Set the port to something non-conflicting if choosing to enable this
admin.enableServer=false
# admin.serverPort=8080

다른 부분은 그냥 두었고, dataDir은 /var/cache 밑에 두어 휘발되지 않도록 설정했다. (기본은 /tmp 밑에 있었던듯)

이를 위해 다음과 같은 권한 설정을 했다.

sudo mkdir /var/cache/zookeeper
sudo chown kafka:kafka /var/cache/zookeeper
sudo chmod 700 /var/cache/zookeeper

zookeeper에 대해 systemctl에 등록해서 부팅시 마다 실행되도록 하기위해 다음 파일 작성

sudo vi /etc/systemd/system/zookeeper.service

[Unit]
Description=Apache Zookeeper server
Documentation=http://zookeeper.apache.org
Requires=network.target remote-fs.target
After=network.target remote-fs.target

[Service]
Type=simple
User=kafka
ExecStart=/opt/kafka/bin/zookeeper-server-start.sh /opt/kafka/config/zookeeper.properties
ExecStop=/opt/kafka/bin/zookeeper-server-stop.sh
Restart=on-abnormal

[Install]
WantedBy=multi-user.target

# 그 다음 systemctl을 통해 시작하고 재부팅 시에도 항상 실행되도록 설정
sudo systemctl start zookeeper
sudo systemctl enable zookeeper

kafka에 대해서도 마찬가지로 작성

sudo vi /etc/systemd/system/kafka.service

[Unit]
Description=Apache Kafka server
Documentation=http://kafka.apache.org
Requires=zookeeper.service
After=zookeeper.service

[Service]
Type=simple
User=kafka
ExecStart=/home/kafka/kafka/bin/kafka-server-start.sh /home/kafka/kafka/config/server.properties
ExecStop=/home/kafka/kafka/bin/kafka-server-stop.sh
Restart=on-abnormal

[Install]
WantedBy=multi-user.target

# 그 다음 systemctl을 통해 시작하고 재부팅 시에도 항상 실행되도록 설정
sudo systemctl start kafka
sudo systemctl enable kafka

트러블 슈팅

서버로그 실시간 모니터링

tail -f /opt/kafka/logs/server.log

jps로 kafka가 잘 실행되고 있는지 확인(QuorumPeerMain과 Kafka가 떠있으면OK)

su - kafka
jps
109092 Jps
108186 QuorumPeerMain(Zookeeper의 시작점)
108573 Kafka

토픽 메시지가 잘 수신되는지 확인하는 방법

가장 기본적으로는 9092포트를 리슨하고 있는지 확인

sevity@sevityubuntu:/opt/kafka/config$ lsof -i :9092
COMMAND    PID   USER   FD   TYPE  DEVICE SIZE/OFF NODE NAME
java    168958 sevity  116u  IPv6 1324110      0t0  TCP sevityubuntu:46330->sevityubuntu:9092 (ESTABLISHED)
java    168958 sevity  117u  IPv6 1320714      0t0  TCP sevityubuntu:46336->sevityubuntu:9092 (ESTABLISHED)

GetOffsetSell을 통해 특정 토픽에 게시된 메지시수를 확인

sevity@sevityubuntu:/opt/kafka$ bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list localhost:9092 --topic iis_log --time -1
iis_log:0:4043038

kafka-console-consumer.sh를 사용하여 특정 토픽의 메시지 수신상황을 실시간으로 확인

$ bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic iis_log --from-beginning
{"@timestamp":"2023-08-01T01:18:34.407Z","@metadata":{"beat":"filebeat","type":"_doc","version":"8.9.0"},"agent":{"version":"8.9.0","ephemeral_id":"98df9137-96e3-4653-b353-2821eebde875","id":"cf839823-29c8-4619-b9c4-85e90804e50e","name":"SEVITY-PC","type":"filebeat"},"ecs":{"version":"8.0.0"},"log":{"file":{"path":"c:\\inetpub\\logs\\LogFiles\\W3SVC1\\u_ex230801.log"},"offset":16076},"message":"2023-08-01 01:17:24 192.168.0.6 GET /wiki/doku.php id=Advice%20To%20Relieve%20Your%20Acid%20Reflux%20Signs%20and%20symptoms 80 - 196.245.181.117 Mozilla/5.0+(X11;+Ubuntu;+Linux+x86_64;+rv:114.0)+Gecko/20100101+Firefox/114.0 http://sevity.com/ 200 0 0 1182","input":{"type":"log"},"host":{"mac":["00-15-83-EA-2C-EE","00-23-24-63-E7-E3","02-50-41-00-00-01","88-36-6C-F7-D9-04","88-36-6C-F7-D9-06","88-36-6C-F7-D9-07"],"hostname":"sevity-pc","name":"sevity-pc","architecture":"x86_64","os":{"family":"windows","name":"Windows 10 Enterprise","kernel":"10.0.19041.3208 (WinBuild.160101.0800)","build":"19045.3208","type":"windows","platform":"windows","version":"10.0"},"id":"dbb18b3d-fb42-49ec-b731-f430dd2f3fd5","ip":["fe80::a30b:d99a:f7ed:2bac","192.168.100.15","fe80::a4ce:c09a:ea91:8a9","169.254.142.88","fe80::9c93:77c8:66bd:b6c5","169.254.55.217","fe80::dd77:ec13:69c4:6922","169.254.56.104","fe80::310a:1b3d:e9a3:431c","192.168.0.6","fe80::2c27:d03a:a322:765b","169.254.236.36"]}}

실습내용

IIS에서 웹서버 로그를 읽어서 분석서버(같은 랜 네트워크 우분투기반)의 kafka로 전송한다.

filebeat는?

filebeat는 ELK로 유명한 Elastic에서 제공하는 경량 로그 수집기이다.

같은 회사 Logstash와의 차이는 후자는 수집뿐 아니라 처리와 변환을 수행가능하다.

filebeat는 그러한 처리 변환 기능을 Flink등 별도의 실시간 로그처리 시스템에 맡기는 구조.

Java가 아닌 Go언어 기반으로 작성됨(Logstash는 Java)

로그수집방식은?

기본적으로 모니터링할 폴더를 지정하고, 그안에 생기는 로그들을 라인단위로 읽어서 kafka나 ElasticSearch쪽에 보내는 구조.

inputs방식 vs modules 방식

filebeat에는 2가지 설정법이 존재한다.

inputs방식: 일반적인 로그에 대해서 지정하는 방식. 어떤 로그도 가능하여 범용성이 좋다.

modules방식: Apache, Nginx, MySQL등 널리 알려진 특정한 로그에 대해서 수집하는 방식. 해당 로그형식에 대한 사전 구성된 설정을 제공하여 좀더 적합도가 높다.

이번 프로젝트에서는 전자를 사용했다. (후자는 시도했는데 뭔가 잘 안됨)

IIS가 돌 고 있는 윈도우 서버에 filebeat 설치하기

여기에 가서 window zip을 고른후 c:\program files\filebeat폴더에 풀어준다.

filebeat.yml을 아래와 같이 수정하여, IIS로 부터 로그를 분석하고, kafaka로 전송하도록 설정한다.

filebeat.inputs:
- type: log
  enabled: true
  paths:
    - c:\inetpub\logs\LogFiles\W3SVC1\u_ex23*
  harvester_limit: 1
  
output.kafka:
  enabled: true
  hosts: ["192.168.0.20:9092"]
  topic: "iis_log"

관리자권한으로 PowerShell 프롬프트를 열고, 다음 명령어를 실행해서 윈도우 서비스로 등록하여 재부팅시에도 실행되도록 만들자.

PS > cd 'C:\Program Files\Filebeat'
PS C:\Program Files\Filebeat> .\install-service-filebeat.ps1

# 다음 커맨드까지 해야 서비스에 등록된다(리눅스에서 systemctl start 와 비슷한 역할)
Start-Service filebeat

# 상태 확인은 다음으로..
Get-Service filebeat

트러블슈팅

PowerShell에서 스크립트 실행이 안되는 경우 아래처럼 정책설정을 해준다.

Set-ExecutionPolicy UnRestricted -File .\install-service-filebeat.ps1
# 또는
Set-ExecutionPolicy Unrestricted -Scope Process
# 후자는 모든 프로세스를 허용하는 구조로 돌려놓으려면 아래 커맨드 실행

Set-ExecutionPolicy Default -Scope Process

kafka로 로그전달이 안될때,

가장 먼저 확인할 것은 아래처럼 telnet등으로 kafka쪽 서버 9092포트로 접속되는지 확인

telnet 192.168.0.20 9092

다음처럼 host등록이 필요한 경우가 있었다.

(설정파일에 IP로 설정했지만 어느순간 hostname으로 변경된 다음 못찾는다고 나옴;)

Windows에서의 호스트 파일 경로: C:\Windows\System32\drivers\etc\hosts
Linux에서의 호스트 파일 경로: /etc/hosts

127.0.0.1   localhost
::1         localhost

# 추가된 매핑 정보
192.168.0.20 sevityubuntu

위처럼 192.168.0.20과 sevityubuntu라는 호스트명을 매핑

filebeat를 콘솔에서 실행하고 로그를 보려면 다음처럼 하면 된다.

PS C:\Program Files\Filebeat> .\filebeat -e

먼저 IIS서버에 Filebeat를 깔아서 우분투 분석 서버에 kafka전송을 한다.