filebeat對接kafka

工作中遇到了，filebeat對接kafka，記下來，分享一下，也為了防止自己忘記

對於filebeat是運行在客戶端的一個收集日志的agent，filebeat是一個耳朵進一個耳朵出，進的意思是監聽哪個日志文件，出的意思是監聽的日志內容輸出到哪里去，當然，這里我們輸出到kafka消息隊列中，而kafka就是一個消息隊列，為什么要用kafka？因為現在用的很多，而且工作中也確實遇到filebeat對接kafka了。具體的可以自行百度查詢，廢話不多說，開始做

 

第一步，安裝helm3

helm3的使用方法和安裝，博客里有，在哪下載？https://github.com/helm/helm/releases/tag/v3.5.2，目前是最新的

第二步， 加載helm倉庫，本來是需要加載官網的倉庫地址的，可惜，翻不了牆，用阿里的代理一下吧，也能用，親測

 

 第三步，下載helm文件，解壓出來就是filebeat目錄

helm pull apphub/kafka
helm pull apphub/filebeat

 

 

 

 第四步，應用kafka文件內容，紅色字體非常有用，接下來給你們解釋一下

[root@VM-0-15-centos ~]# helm install kafka2 ./kafka
NAME: kafka2
LAST DEPLOYED: Fri Feb  5 22:57:45 2021
NAMESPACE: default
STATUS: deployed
REVISION: 1
TEST SUITE: None
NOTES:
** Please be patient while the chart is being deployed **

Kafka can be accessed via port 9092 on the following DNS name from within your cluster:

    kafka2.default.svc.cluster.local #在k8s這個內容可以當做域名來解析出來kafka的ip地址

To create a topic run the following command:#創建主題

    export POD_NAME=$(kubectl get pods --namespace default -l "app.kubernetes.io/name=kafka,app.kubernetes.io/instance=kafka2,app.kubernetes.io/component=kafka" -o jsonpath="{.items[0].metadata.name}")
    kubectl --namespace default exec -it $POD_NAME -- kafka-topics.sh --create --zookeeper kafka2-zookeeper:2181 --replication-factor 1 --partitions 1 --topic test

To list all the topics run the following command:#查看所有的主題

    export POD_NAME=$(kubectl get pods --namespace default -l "app.kubernetes.io/name=kafka,app.kubernetes.io/instance=kafka2,app.kubernetes.io/component=kafka" -o jsonpath="{.items[0].metadata.name}")
    kubectl --namespace default exec -it $POD_NAME -- kafka-topics.sh --list --zookeeper kafka2-zookeeper:2181

To start a kafka producer run the following command:#進入kafka生產者命令行可以給主題添加消息

    export POD_NAME=$(kubectl get pods --namespace default -l "app.kubernetes.io/name=kafka,app.kubernetes.io/instance=kafka2,app.kubernetes.io/component=kafka" -o jsonpath="{.items[0].metadata.name}")
    kubectl --namespace default exec -it $POD_NAME -- kafka-console-producer.sh --broker-list localhost:9092 --topic test

To start a kafka consumer run the following command:#消費者窗口，可以查看到生產者發出的信息

    export POD_NAME=$(kubectl get pods --namespace default -l "app.kubernetes.io/name=kafka,app.kubernetes.io/instance=kafka2,app.kubernetes.io/component=kafka" -o jsonpath="{.items[0].metadata.name}")
    kubectl --namespace default exec -it $POD_NAME -- kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning
    #以下兩行是在容器內部運行的命令，一個是生產者，一個是消費者
    PRODUCER:
        kafka-console-producer.sh --broker-list 127.0.0.1:9092 --topic test
    CONSUMER:
        kafka-console-consumer.sh --bootstrap-server 127.0.0.1:9092 --topic test --from-beginning

 

 

接下來我們來安裝filebeat

filebeat有兩種方式：

　　1.以sidecar方式去手機容器日志，也就是說，一個pod中必須要運行一個filebeat容器，這樣的話，如果有1000個pod，每一個pod跑一個應用，一個filebeat，那么就是2000個，果斷放棄

　　2.以daemonSet方式，以節點運行，那么只需要有幾個node就運行幾個filebeat就可以了，所以我么選擇第二種

第一步，修改values.yml文件，如下，紅色字體需要注意，filebeat.input是監聽的文件，output.kafka是輸出到哪里去，我們這里配的是域名，coredns會自動解析成ip，具體規則是

pod名稱.名稱空間.svc.cluster.local，topic為我們創建的主題名稱

[root@VM-0-15-centos filebeat]# cat values.yaml | grep -v "#" | grep -v "^$"
image:
  repository: docker.elastic.co/beats/filebeat-oss
  tag: 7.4.0
  pullPolicy: IfNotPresent
config:
  filebeat.config:
    modules:
      path: ${path.config}/modules.d/*.yml
      reload.enabled: false
  filebeat.inputs: - type: log enabled: true paths: - /var/a.log output.kafka: enabled: true hosts: ["kafka.default.svc.cluster.local:9092"] topic: test1111
  http.enabled: true
  http.port: 5066
overrideConfig: {}
data:
  hostPath: /var/lib/filebeat
indexTemplateLoad: []
plugins: []
command: []
args: []
extraVars: []
extraVolumes: []
extraVolumeMounts: []
extraSecrets: {}
extraInitContainers: []
resources: {}
priorityClassName: ""
nodeSelector: {}
annotations: {}
tolerations: []
affinity: {}
rbac:
  create: true
serviceAccount:
  create: true
  name:
podSecurityPolicy:
  enabled: False
  annotations: {}
privileged: false
monitoring:
  enabled: true
  serviceMonitor:
    enabled: true
  image:
    repository: trustpilot/beat-exporter
    tag: 0.1.1
    pullPolicy: IfNotPresent
  resources: {}
  args: []
  exporterPort: 9479
  targetPort: 9479

第二步，應用filebeat

[root@VM-0-15-centos ~]# helm install filebeat2 ./filebeat
NAME: filebeat2
LAST DEPLOYED: Fri Feb  5 23:09:51 2021
NAMESPACE: default
STATUS: deployed
REVISION: 1
TEST SUITE: None
NOTES:
To verify that Filebeat has started, run:

  kubectl --namespace=default get pods -l "app=filebeat,release=filebeat2"

第三步，我們查看一下所有的pod，zookeeper是kafka集群必帶的

 

 第四步，測試

進入到容器，然后在/var/a.log這個文件內輸入點東西

 

 第五步，查看另一邊監聽的

 

 

 補充

如果要監控另外容器日志，那么，我們可以把這個應用的容器日志輸出到宿主機的目錄下面，然后再在filebeat容器跟這個宿主機目錄做映射，在配置filebeat.yml文件，來完成filebeat對應用容器產生的日志做監控。最主要就是配置values.yml這個文件，此文件內容如下：

[root@iZ8vb1m9mvb3ev1tqgrldwZ shell]# cat filebeat/values.yaml | grep -v "#" | grep -v "^$"
image:
  repository: docker.elastic.co/beats/filebeat-oss
  tag: 7.4.0
  pullPolicy: IfNotPresent
config:
  filebeat.config:
    modules:
      path: ${path.config}/modules.d/*.yml
      reload.enabled: false
  processors:
    - add_cloud_metadata:
 filebeat.inputs: - type: log enabled: true paths: - /host/var/melocal/logs/*.log - /host/var/geo/logs/*.log - /host/var/rgeo/log/*.log output.kafka: enabled: true hosts: ["kafka.default.svc.cluster.local:9092"] topic: test_topic
  http.enabled: true
  http.port: 5066
overrideConfig: {}
data:
  hostPath: /var/lib/filebeat
indexTemplateLoad: []
plugins: []
command: []
args: []
extraVars: []
extraVolumes: - hostPath: path: /root/jiaohang/amap-melocal/logs name: melocal-log - hostPath: path: /root/jiaohang/amap-geo/data/geocoding/log name: geo-log - hostPath: path: /root/jiaohang/amap-rgeo/data/reverse_geocoding/log name: rgeo-log extraVolumeMounts: - name: melocal-log mountPath: /host/var/melocal/logs readOnly: true - name: geo-log mountPath: /host/var/geo/log readOnly: true - name: rgeo-log mountPath: /host/var/rgeo/log readOnly: true
extraSecrets: {}
extraInitContainers: []
resources: {}
priorityClassName: ""
nodeSelector: {}
annotations: {}
tolerations: []
affinity: {}
rbac:
  create: true
serviceAccount:
  create: true
  name:
podSecurityPolicy:
  enabled: False
  annotations: {}
privileged: false
monitoring:
  enabled: true
  serviceMonitor:
    enabled: true
  image:
    repository: trustpilot/beat-exporter
    tag: 0.1.1
    pullPolicy: IfNotPresent
  resources: {}
  args: []
  exporterPort: 9479
  targetPort: 9479

我在這里監聽了3個目錄下面的日志動態，配置成功后在kafka下面就會有日志輸出了