TL;DR

以下のアプリケーションをMinikube環境にデプロイし、
サンプルアプリケーションのPodに出力されるログをKafkaに転送してみました。

Kafka
kafka-ui (Kafkaの管理画面ツール)
Fluent Bit
適当にログを出すサンプルアプリケーション

この記事では、
Kafkaにログを転送するためのFluent Bitのデプロイ方法について解説します。

また、デプロイのための設定ファイルやマニフェストファイルは以下のリポジトリから入手可能です。
https://github.com/cacapouh/fluentbit-kafka-example/

Kafka周りのセットアップ

まず、Minikubeを使っている場合、メモリを4GB以上にして起動します。¹

Minikubeの起動

minikube start --memory=4096

次に以下のコマンドでKafka周りのツールをデプロイします。
Kafka自体のセットアップについては説明を省きますが、詳細はこちらの記事で解説しています。

Kafka周りのアプリ起動

# ネームスペース作成
kubectl create namespace kafka

# Strimzi(Kafkaのデプロイツール)をデプロイ
kubectl create -f 'https://strimzi.io/install/latest?namespace=kafka' -n kafka

# Kafkaをデプロイ
kubectl apply -f https://strimzi.io/examples/latest/kafka/kafka-persistent-single.yaml -n kafka

# 起動するまで待つ 
kubectl wait kafka/my-cluster --for=condition=Ready --timeout=300s -n kafka

# kafka-ui(Kafkaの画面ツール)のデプロイ
kubectl apply -n kafka -f https://raw.githubusercontent.com/cacapouh/kafka-ui-k8s/main/kafka-ui.yml

デプロイ後、
kafka-ui(Kafkaの画面ツール)にて諸々の設定が上手くいっているか確認できます。
Minikube環境の場合、以下のコマンドでWeb UIへのアクセスリンクが手に入ります。

minikube service kafka-ui-service -n kafka

サンプルアプリのデプロイ

サンプルアプリとして、1秒に1回Count ${カウント値}を標準出力するものをデプロイします。
以下はそのためのPythonスクリプト, Dockerfile, Deploymentのマニフェストファイルです。

main.py

import time

if __name__ == '__main__':
    count = 0
    while True:
        count += 1
        print("Count {}".format(count))
        time.sleep(1)

Dockerfile

FROM python:3.9.18-slim

RUN mkdir /work
WORKDIR /work

COPY main.py /work
ENTRYPOINT python -u main.py

deployment.yml

apiVersion: apps/v1
kind: Deployment
metadata:
  name: simple-app-deployment
  labels:
    app: simple-app
spec:
  replicas: 1
  selector:
    matchLabels:
      app: simple-app
  template:
    metadata:
      labels:
        app: simple-app
    spec:
      containers:
        - name: simple-app
          image: "simple-app:latest"
          imagePullPolicy: IfNotPresent

こちらのアプリは、
docker buildして、マニフェストを適用するだけデプロイできます。

デプロイ方法

docker build -t simple-app .
minikube image load simple-app:latest # Minikube環境の場合
kubectl apply -f deployment.yml

以下はデプロイ後の挙動です。
1秒に1回カウント値がPodのログに出力されています。

$ kubectl get pods
NAME                                    READY   STATUS    RESTARTS   AGE
simple-app-deployment-dd44b5ff4-58nrb   1/1     Running   0          2m45s

$ kubectl logs simple-app-deployment-dd44b5ff4-58nrb | tail
Count 183
Count 184
Count 185
Count 186
Count 187
Count 188
Count 189
Count 190
Count 191
Count 192

FluentBitのデプロイ

以下はFluent BitをKafkaに転送するための設定ファイルの一部です。
この設定ファイルをConfigMapとして扱い、他のマニフェストファイルと一緒に適用すれば、
KafkaにサンプルアプリのPodのログが転送されるようになります。

INPUTセクションでは、入力プラグインの1種であるtailを用いて、
Kubernetesクラスター内のコンテナログが格納される/var/log/containers/*.logをtailしています
FILTERセクションでは、Pythonアプリが標準出力した内容だけを取得するために、ログにCountという文字列が含まれるかどうかでフィルタリングしています
OUTPUTセクションでは、出力先のKafkaクラスタとトピックを指定しています
- ブローカーに指定するアドレスとポートは、${サービス名}.${ネームスペース名}.svc.cluster.localという形式にしています²
- サービス名はkubectl get services -n kafkaで判明します

fluent-bit.confの一部

[INPUT]
    Name tail
    Path /var/log/containers/*.log
    multiline.parser docker, cri
    Tag kube.*
    Mem_Buf_Limit 5MB
    Skip_Long_Lines On

[FILTER]
    Name    grep
    Match   *
    Regex   log Count

[OUTPUT]
    Name kafka
    Match kube.*
    Brokers my-cluster-kafka-bootstrap.kafka.svc.cluster.local:9092
    Topics my-topic

Fluent Bitの詳細はなデプロイ方法については以下の記事で解説しているため、
ここでは説明を割愛します。