使用fluent bit+ClickHouse 实现K8s日志采集

[INPUT]
 Name              tail #这里使用tail插件，具有与 tail -f shell 命令类似的行为，下面的配置项会因为这里选择的插件不同而不同
 Tag               kube.*  #读取的行的标记
 Path              /var/log/containers/*.log #通过使用通用通配符指定特定日志文件或多个日志文件的路径。
 Parser            docker #指定解析器的名称 
 DB                /var/log/flb_kube.db  #指定数据库文件以跟踪受监视的文件和偏移量
 Mem_Buf_Limit     5MB    #设置 Tail 插件可以使用的内存限制。
 Skip_Long_Lines   On     #当被监控的文件由于一行很长而达到缓冲容量时，默认停止监控
 Refresh_Interval  10 //刷新被监控的文件列表的时间间隔，以秒为单位

[OUTPUT]
 Name            clickhouse #插件名
 Match           * #匹配范围

package main

import (
   "C"
   "database/sql"
   "fmt"
   "os"
   "strconv"
   "sync"
   "time"
   "unsafe"
   "github.com/fluent/fluent-bit-go/output"
   "github.com/kshvakov/clickhouse"
   klog "k8s.io/klog"
)

var (
   client *sql.DB

   database  string
   table     string
   batchSize int

   insertSQL = "INSERT INTO %s.%s(date, cluster, namespace, app, pod_name, container_name, host, log, ts) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)"

   rw     sync.RWMutex
   buffer = make([]Log, 0)
)

const (
   DefaultWriteTimeout string = "20"
   DefaultReadTimeout  string = "10"

   DefaultBatchSize int = 1024
)

type Log struct {
   Cluster   string
   Namespace string
   App       string
   Pod       string
   Container string
   Host      string
   Log       string
   Ts        time.Time
}

//export FLBPluginRegister
func FLBPluginRegister(ctx unsafe.Pointer) int {
   return output.FLBPluginRegister(ctx, "clickhouse", "Clickhouse Output Plugin.!")
}

//export FLBPluginInit
// ctx (context) pointer to fluentbit context (state/ c code)
func FLBPluginInit(ctx unsafe.Pointer) int {

   var host string
   if v := os.Getenv("CLICKHOUSE_HOST"); v != "" {
      host = v
   } else {
      klog.Error("you must set host of clickhouse!")
      return output.FLB_ERROR
   }

   var user string
   if v := os.Getenv("CLICKHOUSE_USER"); v != "" {
      user = v
   } else {
      klog.Error("you must set user of clickhouse!")
      return output.FLB_ERROR
   }

   var password string
   if v := os.Getenv("CLICKHOUSE_PASSWORD"); v != "" {
      password = v
   } else {
      klog.Error("you must set password of clickhouse!")
      return output.FLB_ERROR
   }

   if v := os.Getenv("CLICKHOUSE_DATABASE"); v != "" {
      database = v
   } else {
      klog.Error("you must set database of clickhouse!")
      return output.FLB_ERROR
   }

   if v := os.Getenv("CLICKHOUSE_TABLE"); v != "" {
      table = v
   } else {
      klog.Error("you must set table of clickhouse!")
      return output.FLB_ERROR
   }

   if v := os.Getenv("CLICKHOUSE_BATCH_SIZE"); v != "" {
      size, err := strconv.Atoi(v)
      if err != nil {
         klog.Infof("you set the default bacth_size: %d", DefaultBatchSize)
         batchSize = DefaultBatchSize
      }
      batchSize = size
   } else {
      klog.Infof("you set the default bacth_size: %d", DefaultBatchSize)
      batchSize = DefaultBatchSize
   }

   var writeTimeout string
   if v := os.Getenv("CLICKHOUSE_WRITE_TIMEOUT"); v != "" {
      writeTimeout = v
   } else {
      klog.Infof("you set the default write_timeout: %s", DefaultWriteTimeout)
      writeTimeout = DefaultWriteTimeout
   }

   var readTimeout string
   if v := os.Getenv("CLICKHOUSE_READ_TIMEOUT"); v != "" {
      readTimeout = v
   } else {
      klog.Infof("you set the default read_timeout: %s", DefaultReadTimeout)
      readTimeout = DefaultReadTimeout
   }

   dsn := "tcp://" + host + "?username=" + user + "&password=" + password + "&database=" + database + "&write_timeout=" + writeTimeout + "&read_timeout=" + readTimeout

   db, err := sql.Open("clickhouse", dsn)
   if err != nil {
      klog.Error("connecting to clickhouse: %v", err)
      return output.FLB_ERROR
   }

   if err := db.Ping(); err != nil {
      if exception, ok := err.(*clickhouse.Exception); ok {
         klog.Errorf("[%d] %s \n%s\n", exception.Code, exception.Message, exception.StackTrace)
      } else {
         klog.Errorf("Failed to ping clickhouse: %v", err)
      }
      return output.FLB_ERROR
   }
   // ==
   client = db

   return output.FLB_OK
}

//export FLBPluginFlush
// FLBPluginFlush is called from fluent-bit when data need to be sent. is called from fluent-bit when data need to be sent.
func FLBPluginFlush(data unsafe.Pointer, length C.int, tag *C.char) int {
   rw.Lock()
   defer rw.Unlock()
   if err := client.Ping(); err != nil {
      if exception, ok := err.(*clickhouse.Exception); ok {
         klog.Errorf("[%d] %s \n%s\n", exception.Code, exception.Message, exception.StackTrace)
      } else {
         klog.Errorf("Failed to ping clickhouse: %v", err)
      }
      return output.FLB_ERROR
   }
   var ret int
   var timestampData interface{}
   var mapData map[interface{}]interface{}
   dec := output.NewDecoder(data, int(length))

   for {
      ret, timestampData, mapData = output.GetRecord(dec)
      if ret != 0 {
         break
      }
      var timestamp time.Time
      switch t := timestampData.(type) {
      case output.FLBTime:
         timestamp = timestampData.(output.FLBTime).Time
      case uint64:
         timestamp = time.Unix(int64(t), 0)
      default:
         timestamp = time.Now()
      }

      flattenData, err := Flatten(mapData, "", UnderscoreStyle)
      if err != nil {
         break
      }

      log := Log{}
      for k, v := range flattenData {
         value := ""
         switch t := v.(type) {
         case string:
            value = t
         case []byte:
            value = string(t)
         default:
            value = fmt.Sprintf("%v", v)
         }

         switch k {
         case "cluster":
            log.Cluster = value
         case "kubernetes_namespace_name":
            log.Namespace = value
         case "kubernetes_labels_app":
            log.App = value
         case "kubernetes_labels_k8s-app":
            log.App = value
         case "kubernetes_pod_name":
            log.Pod = value
         case "kubernetes_container_name":
            log.Container = value
         case "kubernetes_host":
            log.Host = value
         case "log":
            log.Log = value
         }

      }

      log.Ts = timestamp
      buffer = append(buffer, log)
   }

   // sink data
   if len(buffer) < batchSize {
      return output.FLB_OK
   }

   sql := fmt.Sprintf(insertSQL, database, table)

   tx, err := client.Begin()
   if err != nil {
      klog.Errorf("begin transaction failure: %s", err.Error())
      return output.FLB_ERROR
   }

   // build statements
   smt, err := tx.Prepare(sql)
   if err != nil {
      klog.Errorf("prepare statement failure: %s", err.Error())
      return output.FLB_ERROR
   }
   for _, l := range buffer {
      _, err = smt.Exec(l.Ts, l.Cluster, l.Namespace, l.App, l.Pod, l.Container, l.Host,
         l.Log, l.Ts)

      if err != nil {
         klog.Errorf("statement exec failure: %s", err.Error())
         return output.FLB_ERROR
      }
   }

   // commit and record metrics
   if err = tx.Commit(); err != nil {
      klog.Errorf("commit failed failure: %s", err.Error())
      return output.FLB_ERROR
   }

   buffer = make([]Log, 0)

   return output.FLB_OK
}

//export FLBPluginExit
func FLBPluginExit() int {
   return output.FLB_OK
}

func main() {
}

CREATE DATABASE IF NOT EXISTS scmp;
CREATE TABLE IF NOT EXISTS scmp.logs(
    date Date DEFAULT toDate(0),
    cluster String,namespace String,
    app String,
    pod_name String,
    container_name String,
    host String,log String,
    ts DateTime) 
ENGINE = MergeTree() PARTITION BY toYYYYMMDD(date) 
ORDER BY (cluster, namespace, app, pod_name, container_name, host,ts);

apiVersion: v1
kind: ConfigMap
metadata:
  name: k8s-log-agent-config
  namespace: kube
  labels:
    k8s-app: k8s-log-agent
data:
  # Configuration files: server, input, filters and output
  # ======================================================
  fluent-bit.conf: |
    [SERVICE]
     Flush         1
     Log_Level     error
     Daemon        off
     Parsers_File  parsers.conf
     HTTP_Server   On
     HTTP_Listen   0.0.0.0
     HTTP_Port     2020

    @INCLUDE input-kubernetes.conf
    @INCLUDE filter-kubernetes.conf
    @INCLUDE output-kubernetes.conf

  input-kubernetes.conf: |
    [INPUT]
        Name              tail
        Tag               kube.*
        Path              /var/log/containers/*.log
        Parser            docker
        DB                /var/log/flb_kube.db
        Mem_Buf_Limit     5MB
        Skip_Long_Lines   On
        Refresh_Interval  10

  filter-kubernetes.conf: |
    [FILTER]
     Name                kubernetes
     Match               *
     Kube_URL            https://kubernetes.default.svc.cluster.local:443
     Merge_Log           On
     Annotations         Off
     Kube_Tag_Prefix     kube.var.log.containers.
     Merge_Log_Key       log_processed

    [FILTER]
     Name                modify
     Match               *
     Set  cluster  ${CLUSTER_NAME}
  output-kubernetes.conf: |
    [OUTPUT]
     Name            clickhouse
     Match           *
  parsers.conf: |
    [PARSER]
     Name   apache
     Format regex
     Regex  ^(?<host>[^ ]*) [^ ]* (?<user>[^ ]*) \[(?<time>[^\]]*)\] "(?<method>\S+)(?: +(?<path>[^\"]*?)(?: +\S*)?)?" (?<code>[^ ]*) (?<size>[^ ]*)(?: "(?<referer>[^\"]*)" "(?<agent>[^\"]*)")?$
     Time_Key time
     Time_Format %d/%b/%Y:%H:%M:%S %z

    [PARSER]
     Name   apache2
     Format regex
     Regex  ^(?<host>[^ ]*) [^ ]* (?<user>[^ ]*) \[(?<time>[^\]]*)\] "(?<method>\S+)(?: +(?<path>[^ ]*) +\S*)?" (?<code>[^ ]*) (?<size>[^ ]*)(?: "(?<referer>[^\"]*)" "(?<agent>[^\"]*)")?$
     Time_Key time
     Time_Format %d/%b/%Y:%H:%M:%S %z

    [PARSER]
     Name   apache_error
     Format regex
     Regex  ^\[[^ ]* (?<time>[^\]]*)\] \[(?<level>[^\]]*)\](?: \[pid (?<pid>[^\]]*)\])?( \[client (?<client>[^\]]*)\])? (?<message>.*)$

    [PARSER]
     Name   nginx
     Format regex
     Regex ^(?<remote>[^ ]*) (?<host>[^ ]*) (?<user>[^ ]*) \[(?<time>[^\]]*)\] "(?<method>\S+)(?: +(?<path>[^\"]*?)(?: +\S*)?)?" (?<code>[^ ]*) (?<size>[^ ]*)(?: "(?<referer>[^\"]*)" "(?<agent>[^\"]*)")?$
     Time_Key time
     Time_Format %d/%b/%Y:%H:%M:%S %z

    [PARSER]
     Name   json
     Format json
     Time_Key time
     Time_Format %d/%b/%Y:%H:%M:%S %z

    [PARSER]
     Name         docker
     Format       json
     Time_Key     time
     Time_Format  %Y-%m-%dT%H:%M:%S.%L
     Time_Keep    On

    [PARSER]
     Name        syslog
     Format      regex
     Regex       ^\<(?<pri>[0-9]+)\>(?<time>[^ ]* {1,2}[^ ]* [^ ]*) (?<host>[^ ]*) (?<ident>[a-zA-Z0-9_\/\.\-]*)(?:\[(?<pid>[0-9]+)\])?(?:[^\:]*\:)? *(?<message>.*)$
     Time_Key    time
     Time_Format %b %d %H:%M:%S

apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRoleBinding
metadata:
    name: k8s-log-agent-read
roleRef:
  apiGroup: rbac.authorization.k8s.io
  kind: ClusterRole
  name: k8s-log-agent-read
subjects:
  - kind: ServiceAccount
    name: k8s-log-agent
    namespace: kube-system

---

apiVersion: rbac.authorization.k8s.io/v1
kind: ClusterRole
metadata:
  name: k8s-log-agent-read
rules:
  - apiGroups: [""]
    resources:
      - namespaces
      - pods
    verbs: ["get", "list", "watch"]

---


apiVersion: v1
kind: ServiceAccount
metadata:
  name:k8s-log-agent
  namespace: kube

apiVersion: apps/v1
kind: DaemonSet
metadata:
  name: k8s-log-agent
  namespace: kube
  labels:
    k8s-app: k8s-log-agent
    version: v1
    kubernetes.io/cluster-service: "true"
spec:
  selector:
    matchLabels:
      k8s-app: k8s-log-agent
  template:
    metadata:
      labels:
        k8s-app: k8s-log-agent
        version: v1
        kubernetes.io/cluster-service: "true"
      annotations:
        prometheus.io/scrape: "true"
        prometheus.io/port: "2020"
        prometheus.io/path: /api/v1/metrics/prometheus
    spec:
      containers:
        - name: fluent-bit
          image: iyacontrol/fluent-bit-ck:1.2.2
          imagePullPolicy: Always
          ports:
            - containerPort: 2020
          resources:
            limits:
              cpu: 200m
              memory: 200Mi
            requests:
              cpu: 200m
              memory: 200Mi
          env:
            - name: CLUSTER_NAME
              value: "clickhouse"
            - name: CLICKHOUSE_HOST
              value: "10.1.62.62:9150"
            - name: CLICKHOUSE_USER
              value: "oms"
            - name: CLICKHOUSE_PASSWORD
              value: "EBupt123"
            - name: CLICKHOUSE_DATABASE
              value: "scmp"
            - name: CLICKHOUSE_TABLE
              value: "logs"
            - name: NODENAME
              valueFrom:
                fieldRef:
                  fieldPath: spec.nodeName
          volumeMounts:
            - name: varlog
              mountPath: /var/log
            - name: varlibdockercontainers
              mountPath: /var/lib/docker/containers
              readOnly: true
            - name: k8s-log-agent-config
              mountPath: /fluent-bit/etc/
      terminationGracePeriodSeconds: 10
      volumes:
      - name: varlog
        hostPath:
          path: /var/log
      - name: varlibdockercontainers
        hostPath:
          path: /var/lib/docker/containers
      - name: k8s-log-agent-config
        configMap:
          name: k8s-log-agent-config
      serviceAccountName: k8s-log-agent
      tolerations:
      - key: node-role.kubernetes.io/master
        operator: Exists
        effect: NoSchedule

Name              Tail  #使用tail模式
Buffer_Chunk_Size 32k   #设置初始缓冲区大小以读取文件数据 默认为32
Buffer_Max_Siz    32k   #设置每个被监控文件的缓冲区大小限制 默认为32
Mem_Buf_Limit     5MB   #设置 Tail 插件在将数据附加到引擎时可以使用的内存限制。如果达到限制，则暂停；当数据被刷新时，它会恢复。 无默认值
Skip_Long_Lines   On    #当一行日志长度超过Buffer_Max_Size的设置时否停止监控该文件，设置为on为跳过长行并继续处理适合缓冲区大小的其他行 默认off为停止监控该文件
Refresh_Interval  10    #重新获取一次日志的时间间隔 默认60
Read_from_Head    False #对于开始时新发现的文件（没有数据库偏移/位置），从文件的头部读取内容，而不是尾部 默认False
Rotate_Wait       5      #指定在刷新某些挂起数据时监控文件一次的额外时间 默认5
DB                /var/log/flb_kube.db  #指定数据库文件以记录受监控的文件和偏移量 
DB.sync           normal    #设置默认同步 (I/O) 方法 默认normal
DB.locking        false     #指定数据库只能由 Fluent Bit 访问 默认false
DB.journal_mode   WAL       #设置数据库的日志模式 默认wal
exit_on_eof       false     #当读取文件时，它会在到达文件末尾时立即退出 默认false

Name                kubernetes #使用tail模式
Match               * #匹配规则
Kube_URL            https://kubernetes.default.svc.cluster.local:443 #k8s API 服务器端点
Merge_Log           On   #启用后，它会检查日志字段内容是否为 JSON 字符串映射，如果是，则将映射字段附加为日志结构的一部分。
Annotations         On   #在额外的元数据中包含 Kubernetes 资源注释 默认为on
Kube_Tag_Prefix     kube.var.log.containers. #当源记录来自 Tail 输入插件时，此选项指定 Tail 配置中使用的前缀
Merge_Log_Key       log_processed #启用 Merge_Log 后，过滤器会尝试假定传入消息中的日志字段是 JSON 字符串消息，并在映射中日志字段的同一级别对其进行结构化表示，设置key后从原始日志内容中提取的所有新结构化字段都将插入到新键下