# 安裝 Fluentd on K8s(Kubeadm) use DaemonSet & 串接 Elasticsearch

## 部署 Fluentd

$ mkdir -p wulin/fluentd

* 建立 fluentd RBAC

$ echo 'apiVersion: v1
kind: ServiceAccount
metadata:
  name: fluentd
  namespace: kube-system

---
apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
  name: fluentd
rules:
- apiGroups:
  - ""
  resources:
  - pods
  - namespaces
  verbs:
  - get
  - list
  - watch

---
kind: ClusterRoleBinding
apiVersion: rbac.authorization.k8s.io/v1
metadata:
  name: fluentd
roleRef:
  kind: ClusterRole
  name: fluentd
  apiGroup: rbac.authorization.k8s.io
subjects:
- kind: ServiceAccount
  name: fluentd
  namespace: kube-system' > wulin/fluentd/rbac.yaml
  
$ kubectl apply -f wulin/fluentd/rbac.yaml

* 建立 ConfigMap
- `path /var/log/containers/*.log` 設定 fluentd 要收集哪裡的 log，這裡是 fluentd pod 肚子里的目錄，所以要改變目錄要去設定 fluentd DaemonSet 的 hostpath。

$ echo 'apiVersion: v1
kind: ConfigMap
metadata:
  name: fluentd-cm
  namespace: kube-system
data:
  kubernetes.conf: |-
    # 要收集的資料來源
    <source>
      @id audit_proxy_buffer_logs
      @type tail
      path /var/log/containers/*.log
      pos_file /var/log/fluentd-containers.pos
      tag kubernetes.*
      read_from_head true
      <parse>
        @type cri
      </parse>
    </source>
    
    # 避免 Fluentd 重複收集自己的 log，所以要丟棄 Fluentd 自己的 log
    <match kubernetes.var.log.containers.**fluentd-ds**.log>
      @type null
    </match>
    
    # kubernetes_metadata 插件會把 Kubernetes 的 metadata（namespace、pod、container、labels、annotations、pod_ip、pod_id 等）加回到每一筆 record，後面才可以篩選 k8s 的這些資源名稱
    <filter kubernetes.**>
      @type kubernetes_metadata
    </filter>
    
    # 用 rewrite_tag_filter 把 tag 改成 namespace，意思就是後面可以透過指定 namespace 名稱來篩選 log
    <match kubernetes.**>
      @type rewrite_tag_filter
      <rule>
        key $.kubernetes.namespace_name
        pattern ^(.+)$
        tag $1
      </rule>
    </match>
    
    # <match demo> — 指定只收集 demo namespace 的所有 pod log，把 demo namespace 的 logs 拋轉到 Elasticsearch
    <match demo>
      @id elasticsearch
      @type elasticsearch
      @log_level info
      include_tag_key true
      host "#{ENV['FLUENT_ELASTICSEARCH_HOST']}"
      port "#{ENV['FLUENT_ELASTICSEARCH_PORT']}"
      scheme "#{ENV['FLUENT_ELASTICSEARCH_SCHEME']}"
      user "#{ENV['FLUENT_ELASTICSEARCH_USER']}"
      password "#{ENV['FLUENT_ELASTICSEARCH_PASSWORD']}"
      logstash_format true
      logstash_prefix ${tag}
      ssl_verify false
      reload_connections false
      reload_on_failure false
      request_timeout 30s
      <buffer>
        @type file
        path /var/log/fluentd-buffers/container.buffer
        flush_mode interval
        retry_type exponential_backoff
        flush_thread_count 2
        flush_interval 5s
        retry_forever true
        retry_max_interval 30
        queue_limit_length 8
        overflow_action block
        chunk_limit_size 20M
        chunk_limit_records 300
        retry_max_times 3
      </buffer>
    </match>' > wulin/fluentd/fluentd-cm.yaml
    
$ kubectl apply -f wulin/fluentd/fluentd-cm.yaml

* 建立 fluentd DaemonSet，他負責收集 log 以及拋轉到 Elasticsearch

$ echo 'apiVersion: apps/v1
kind: DaemonSet
metadata:
  name: fluentd
  namespace: kube-system
  labels:
    k8s-app: fluentd-logging
    version: v1
spec:
  selector:
    matchLabels:
      k8s-app: fluentd-logging
      version: v1
  template:
    metadata:
      labels:
        k8s-app: fluentd-logging
        version: v1
    spec:
      serviceAccountName: fluentd
      tolerations:
      - key: node-role.kubernetes.io/control-plane
        effect: NoSchedule
      - key: node-role.kubernetes.io/master
        effect: NoSchedule
      containers:
      - name: fluentd
        image: fluent/fluentd-kubernetes-daemonset:v1-debian-elasticsearch
        env:
          - name: K8S_NODE_NAME
            valueFrom:
              fieldRef:
                fieldPath: spec.nodeName
          # 指定 elasticsearch 位置
          - name:  FLUENT_ELASTICSEARCH_HOST
            value: "elasticsearch-es-http.elastic.svc.cluster.local"
          - name:  FLUENT_ELASTICSEARCH_PORT
            value: "9200"
          - name: FLUENT_ELASTICSEARCH_SCHEME
            value: "https"
          # 指定 elasticsearch 帳號及密碼
          - name: FLUENT_ELASTICSEARCH_USER
            value: "elastic"
          - name: FLUENT_ELASTICSEARCH_PASSWORD
            value: "5CCh5Nv659yDkk935bYCU17I"
          - name: FLUENT_UID
            value: "0"
          - name: FLUENTD_CONF
            value: "kubernetes.conf"
        resources:
          limits:
            memory: 500Mi
          requests:
            cpu: 100m
            memory: 200Mi
        volumeMounts:
        - name: varlog
          mountPath: /var/log
        - name: dockercontainerlogdirectory
          mountPath: /var/log/pods
          readOnly: true
        - name: config
          mountPath: /fluentd/etc/kubernetes.conf
          subPath: kubernetes.conf
      terminationGracePeriodSeconds: 30
      volumes:
      - name: varlog
        hostPath:
          path: /var/log
      - name: dockercontainerlogdirectory
        hostPath:
          path: /var/log/pods
      - name: config
        configMap:
          name: fluentd-cm' > wulin/fluentd/fluentd-ds.yaml
          
$ kubectl apply -f wulin/fluentd/fluentd-ds.yaml

## 測試

$ kubectl create ns demo

$ echo 'apiVersion: v1
kind: Pod
metadata:
  name: my-logging-pod
  namespace: demo
spec:
  restartPolicy: Always
  containers:
  - name: sak
    image: rancherlabs/swiss-army-knife:latest
    command: ["/bin/sh"]
    args:
    - -c
    - |
      i=0
      while true; do
        ts=$(date -u +"%Y-%m-%dT%H:%M:%SZ")
        echo "{\"@timestamp\":\"${ts}\",\"level\":\"INFO\",\"message\":\"hello-demo\",\"seq\":${i}}"
        i=$((i+1))
        sleep 1
      done' | kubectl apply -f -

$ kubectl -n demo logs my-logging-pod --tail 10
{"@timestamp":"2025-09-11T01:58:05Z","level":"INFO","message":"hello-demo","seq":61713}
{"@timestamp":"2025-09-11T01:58:06Z","level":"INFO","message":"hello-demo","seq":61714}
{"@timestamp":"2025-09-11T01:58:07Z","level":"INFO","message":"hello-demo","seq":61715}
{"@timestamp":"2025-09-11T01:58:08Z","level":"INFO","message":"hello-demo","seq":61716}
{"@timestamp":"2025-09-11T01:58:09Z","level":"INFO","message":"hello-demo","seq":61717}
{"@timestamp":"2025-09-11T01:58:10Z","level":"INFO","message":"hello-demo","seq":61718}
{"@timestamp":"2025-09-11T01:58:11Z","level":"INFO","message":"hello-demo","seq":61719}
{"@timestamp":"2025-09-11T01:58:12Z","level":"INFO","message":"hello-demo","seq":61720}
{"@timestamp":"2025-09-11T01:58:13Z","level":"INFO","message":"hello-demo","seq":61721}
{"@timestamp":"2025-09-11T01:58:14Z","level":"INFO","message":"hello-demo","seq":61722}

## 進入 kibana

### 建立 kibana view data

* Management -> Stack Management -> Data Views


* Kibana -> Data Views


* 建立 Data Views


* index pattern 填寫 `demo*`，設定好後點選 Save data view to Kibana


### 查詢 log
* Dev Tools -> Console
* 查詢 demo index 下的所有資料

POST /demo*/_search
{
  "query": {
    "match_all": {}
  }
}


* 使用字串查詢，這會查所有 index 的 log

GET _search
{
  "query": { "query_string": { "query": "my-logging-pod" } },
  "size": 10
}

* 使用 k8s namespace 名稱查詢 pod log

GET _search
{
  "query": {
    "match": {
      "kubernetes.namespace_name": "demo"
    }
  },
  "size": 50
}

* 查詢所有 index

GET _cat/indices?v&s=index

* 指定查詢 log，只要看到 `message`、`pod 名稱`、`namespace 名稱`、`k8s 節點名稱`、`container 名稱`、`時間`

GET /demo*/_search
{
  "_source": [
    "message",
    "kubernetes.pod_name",
    "kubernetes.namespace_name",
    "kubernetes.host",
    "kubernetes.container_name",
    "@timestamp"
  ],
  "query": {
    "match": {
      "kubernetes.namespace_name": "demo"
    }
  },
  "size": 50
}

## 參考

https://docs.fluentd.org/container-deployment/kubernetes