Add support for Kafka

.github/workflows/docker.yaml

@@ -8,8 +8,8 @@ on:
       - v*
 
 jobs:
-  docker:
-    name: Docker
+  fluent-bit-clickhouse:
+    name: Fluent Bit -> Clickhouse
     runs-on: ubuntu-20.04
     steps:
       - name: Checkout
@@ -47,3 +47,43 @@ jobs:
           file: ./cmd/out_clickhouse/Dockerfile
           platforms: linux/amd64,linux/arm,linux/arm64
           tags: kobsio/fluent-bit-clickhouse:${{ steps.tag.outputs.tag }}
+
+  kafka-clickhouse:
+    name: Kafka -> Clickhouse
+    runs-on: ubuntu-20.04
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v2
+        with:
+          fetch-depth: 0
+
+      - name: Set Docker Tag
+        id: tag
+        run: |
+          if [[ $GITHUB_REF == refs/tags/* ]]; then
+            DOCKER_TAG="${GITHUB_REF:10}"
+          else
+            DOCKER_TAG="main"
+          fi
+          echo ::set-output name=tag::${DOCKER_TAG}
+      - name: Set up QEMU
+        uses: docker/setup-qemu-action@v1
+
+      - name: Set up Docker Buildx
+        uses: docker/setup-buildx-action@v1
+
+      - name: Login to DockerHub
+        uses: docker/login-action@v1
+        with:
+          username: ${{ secrets.DOCKER_USERNAME }}
+          password: ${{ secrets.DOCKER_PASSWORD }}
+
+      - name: Build and Push Docker Image
+        id: docker_build
+        uses: docker/build-push-action@v2
+        with:
+          push: true
+          context: .
+          file: ./cmd/out_clickhouse/Dockerfile
+          platforms: linux/amd64,linux/arm,linux/arm64
+          tags: kobsio/fluent-bit-clickhouse:kafka-${{ steps.tag.outputs.tag }}

.gitignore

@@ -1,2 +1,3 @@
 out_clickhouse.h
 out_clickhouse.so
+kafka-clickhouse

CHANGELOG.md

@@ -9,6 +9,7 @@ NOTE: As semantic versioning states all 0.y.z releases can contain breaking chan
 ## Unreleased
 
 - [#4](https://github.com/kobsio/fluent-bit-clickhouse/pull/4): Use logrus for logging and remove conf directory.
+- [#5](https://github.com/kobsio/fluent-bit-clickhouse/pull/5): Add support for Kafka, so that Fluent Bit writes all logs to Kafka and we then write the logs from Kafka to ClickHouse.
 
 ## [v0.4.0](https://github.com/kobsio/fluent-bit-clickhouse/releases/tag/v0.4.0) (2021-09-08)
 

Makefile

@@ -6,14 +6,22 @@ REVISION  ?= $(shell git rev-parse HEAD)
 VERSION   ?= $(shell git describe --tags)
 
 .PHONY: build
-build:
+build-fluent-bit-clickhouse:
 	@go build -ldflags "-X ${REPO}/pkg/version.Version=${VERSION} \
 		-X ${REPO}/pkg/version.Revision=${REVISION} \
 		-X ${REPO}/pkg/version.Branch=${BRANCH} \
 		-X ${REPO}/pkg/version.BuildUser=${BUILDUSER} \
 		-X ${REPO}/pkg/version.BuildDate=${BUILDTIME}" \
 		-buildmode=c-shared -o out_clickhouse.so ./cmd/out_clickhouse;
 
+build-kafka-clickhouse:
+	@go build -ldflags "-X ${REPO}/pkg/version.Version=${VERSION} \
+		-X ${REPO}/pkg/version.Revision=${REVISION} \
+		-X ${REPO}/pkg/version.Branch=${BRANCH} \
+		-X ${REPO}/pkg/version.BuildUser=${BUILDUSER} \
+		-X ${REPO}/pkg/version.BuildDate=${BUILDTIME}" \
+		-o kafka-clickhouse ./cmd/kafka;
+
 .PHONY: release-major
 release-major:
 	$(eval MAJORVERSION=$(shell git describe --tags --abbrev=0 | sed s/v// | awk -F. '{print "v"$$1+1".0.0"}'))

README.md

@@ -85,60 +85,3 @@ CREATE TABLE IF NOT EXISTS logs.logs_local(
 
 CREATE TABLE IF NOT EXISTS logs.logs AS logs.logs_local ENGINE = Distributed('{cluster}', logs, logs_local, cityHash64(cluster, namespace, app, pod_name, container_name, host));
 ```
-
-## Development
-
-We are using [kind](https://kind.sigs.k8s.io/docs/user/quick-start/) for local development. To create a new Kubernetes cluster using kind you can run the `cluster/cluster.sh` script, which will create such a cluster with a Docker registry:
-
-```sh
-./cluster/cluster.sh
-```
-
-Once the cluster is running we can build and push the Docker image for Fluent Bit:
-
-```sh
-docker build -f cmd/out_clickhouse/Dockerfile -t localhost:5000/fluent-bit-clickhouse:latest .
-docker push localhost:5000/fluent-bit-clickhouse:latest
-
-# To run the Docker image locally, the following command can be used:
-docker run -it --rm localhost:5000/fluent-bit-clickhouse:latest
-```
-
-In the next step we have to create our ClickHouse cluster via the [ClickHouse Operator](https://github.com/Altinity/clickhouse-operator). To do that we can deploy all the files from the `cluster/clickhouse-operator` and `cluster/clickhouse` folder:
-
-```sh
-k apply -f cluster/clickhouse-operator
-k apply -f cluster/clickhouse
-```
-
-Once ClickHouse is running we have to connect to the two ClickHouse nodes to create our SQL schema. The schema can be found in the `schema.sql` file, just execute each SQL command one by one on both ClickHouse nodes:
-
-```sh
-k exec -n clickhouse -it chi-clickhouse-sharded-0-0-0 -c clickhouse -- clickhouse-client
-k exec -n clickhouse -it chi-clickhouse-sharded-1-0-0 -c clickhouse -- clickhouse-client
-```
-
-Now we can deploy Fluent Bit to ingest all logs into ClickHouse:
-
-```sh
-k apply -f cluster/fluent-bit
-k logs -n fluent-bit -l app=fluent-bit -f
-```
-
-To check if the logs are arriving in ClickHouse you can use the following SQL commands:
-
-```sql
-SELECT count(*) FROM logs.logs;
-SELECT * FROM logs.logs LIMIT 10;
-
-SELECT count(*) FROM logs.logs_local;
-SELECT * FROM logs.logs_local LIMIT 10;
-```
-
-To clean up all the created resources run the following commands:
-
-```sh
-kind delete cluster --name fluent-bit-clickhouse
-docker stop kind-registry
-docker rm kind-registry
-```

cluster/fluent-bit/kafka/fluent-bit-clusterrole.yaml

@@ -0,0 +1,14 @@
+---
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRole
+metadata:
+  name: fluent-bit
+  labels:
+    app: fluent-bit
+rules:
+  - apiGroups:
+      - ""
+    resources:
+      - pods
+    verbs:
+      - get

cluster/fluent-bit/kafka/fluent-bit-clusterrolebinding.yaml

@@ -0,0 +1,15 @@
+---
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRoleBinding
+metadata:
+  name: fluent-bit
+  labels:
+    app: fluent-bit
+roleRef:
+  apiGroup: rbac.authorization.k8s.io
+  kind: ClusterRole
+  name: fluent-bit
+subjects:
+  - kind: ServiceAccount
+    name: fluent-bit
+    namespace: fluent-bit

cluster/fluent-bit/kafka/fluent-bit-cm.yaml

@@ -0,0 +1,61 @@
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: fluent-bit
+  namespace: fluent-bit
+  labels:
+    app: fluent-bit
+data:
+  fluent-bit.conf: |
+    [SERVICE]
+        Flush        5
+        Daemon       Off
+        Log_Level    ${LOG_LEVEL}
+        HTTP_Server  On
+        HTTP_Listen  0.0.0.0
+        HTTP_Port    2020
+        Parsers_File parsers.conf
+        Parsers_File parsers_custom.conf
+
+    [INPUT]
+        Name              tail
+        Path              /var/log/containers/*.log
+        Parser            docker
+        Tag               kube.*
+        Refresh_Interval  5
+        Mem_Buf_Limit     20MB
+        Skip_Long_Lines   On
+        DB                /tail-db/tail-containers-state.db
+        DB.Sync           Normal
+
+    [FILTER]
+        Name                kubernetes
+        Match               kube.*
+        Kube_Tag_Prefix     kube.var.log.containers.
+        Kube_URL            https://kubernetes.default.svc:443
+        Kube_CA_File        /var/run/secrets/kubernetes.io/serviceaccount/ca.crt
+        Kube_Token_File     /var/run/secrets/kubernetes.io/serviceaccount/token
+        Merge_Log           On
+        Merge_Log_Key       content
+        K8S-Logging.Parser  On
+        K8S-Logging.Exclude On
+
+    [FILTER]
+        Name  modify
+        Match *
+        Add   cluster fluent-bit-kafka-clickhouse
+
+    [OUTPUT]
+        Name                          kafka
+        Match                         *
+        Brokers                       kafka-kafka-0.kafka-kafka-brokers.kafka.svc.cluster.local:9092,kafka-kafka-1.kafka-kafka-brokers.kafka.svc.cluster.local:9092,kafka-kafka-2.kafka-kafka-brokers.kafka.svc.cluster.local:9092
+        Topics                        fluent-bit
+        Queue_Full_Retries            10
+        rdkafka.log.connection.close  false
+        rdkafka.request.required.acks 1
+
+  parsers.conf: |
+    [PARSER]
+        Name        logfmt
+        Format      logfmt

cluster/fluent-bit/kafka/fluent-bit-ds.yaml

@@ -0,0 +1,72 @@
+---
+apiVersion: apps/v1
+kind: DaemonSet
+metadata:
+  name: fluent-bit
+  namespace: fluent-bit
+  labels:
+    app: fluent-bit
+spec:
+  selector:
+    matchLabels:
+      app: fluent-bit
+  template:
+    metadata:
+      labels:
+        app: fluent-bit
+    spec:
+      serviceAccountName: fluent-bit
+      serviceAccount: fluent-bit
+      containers:
+        - image: fluent/fluent-bit:1.8.4
+          imagePullPolicy: IfNotPresent
+          name: fluent-bit
+          ports:
+            - containerPort: 2020
+              name: http-metrics
+              protocol: TCP
+          env:
+            - name: LOG_LEVEL
+              value: "info"
+          volumeMounts:
+            - mountPath: /var/log
+              name: varlog
+            - mountPath: /var/lib/docker/containers
+              name: varlibdockercontainers
+              readOnly: true
+            - mountPath: /fluent-bit/etc/fluent-bit.conf
+              name: config
+              subPath: fluent-bit.conf
+            - mountPath: /fluent-bit/etc/parsers_custom.conf
+              name: config
+              subPath: parsers.conf
+            - mountPath: /tail-db
+              name: tail-db
+          resources:
+            limits:
+              cpu: 100m
+              memory: 256Mi
+            requests:
+              cpu: 10m
+              memory: 64Mi
+      dnsPolicy: ClusterFirst
+      tolerations:
+        - effect: NoSchedule
+          key: node-role.kubernetes.io/master
+        - effect: NoExecute
+          operator: Exists
+      volumes:
+        - hostPath:
+            path: /var/log
+          name: varlog
+        - hostPath:
+            path: /var/lib/docker/containers
+          name: varlibdockercontainers
+        - hostPath:
+            path: /var/lib/fluent-bit
+            type: DirectoryOrCreate
+          name: tail-db
+        - configMap:
+            defaultMode: 420
+            name: fluent-bit
+          name: config

cluster/fluent-bit/kafka/fluent-bit-ns.yaml

@@ -0,0 +1,7 @@
+---
+apiVersion: v1
+kind: Namespace
+metadata:
+  name: fluent-bit
+  labels:
+    app: fluent-bit

cluster/fluent-bit/kafka/fluent-bit-sa.yaml

@@ -0,0 +1,8 @@
+---
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: fluent-bit
+  namespace: fluent-bit
+  labels:
+    app: fluent-bit

cluster/fluent-bit/kafka/fluent-bit-svc.yaml

@@ -0,0 +1,17 @@
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: fluent-bit
+  namespace: fluent-bit
+  labels:
+    app: fluent-bit
+spec:
+  ports:
+    - name: http-metrics
+      port: 2020
+      protocol: TCP
+      targetPort: http-metrics
+  selector:
+    app: fluent-bit
+  type: ClusterIP