納尼?自建K8s集群日志收集還能通過(guò)JMQ保存到JES

作者：京東科技 劉恩浩

一、背景

基于K8s集群的私有化交付方案中，日志收集采用了ilogtail+logstash+kafka+es方案，其中ilogtail負(fù)責(zé)日志收集，logstash負(fù)責(zé)對(duì)數(shù)據(jù)轉(zhuǎn)換，kafka負(fù)責(zé)對(duì)日志傳遞中的消峰進(jìn)而減少es的寫(xiě)入壓力，es用來(lái)保存日志數(shù)據(jù)。在私有化交付中本方案中涉及的中間件一般需要單獨(dú)部署，但是在京東內(nèi)網(wǎng)環(huán)境的部署考慮到kafka和es的高可用，則不推薦采用單獨(dú)部署的方案。

二、新方案實(shí)踐

1.新方案簡(jiǎn)介

在京東內(nèi)網(wǎng)環(huán)境部署K8S并收集日志， kafka+es的替代方案考慮使用JMQ+JES，由于JMQ的底層是基于kafaka、JES的底層基于ES，所以該替換方案理論上是可行的

2.主要架構(gòu)

數(shù)據(jù)流向大致如下
應(yīng)用日志 -> ilogtail -> JMQ -> logstash -> JES

3.如何使用

核心改造點(diǎn)匯總

ilogtail nameservers配置
增加解析JMQ域名的nameserver（京東云主機(jī)上無(wú)法直接解析.local域名）

spec:
    spec:
      dnsPolicy: "None"
      dnsConfig:
        nameservers:
          - x.x.x.x # 可以解析jmq域名的nameserver

ilogtail flushers配置
調(diào)整發(fā)送到JMQ到配置

apiVersion: v1
kind: ConfigMap
metadata:
  name: ilogtail-user-cm
  namespace: elastic-system
data:
  app_stdout.yaml: |
    flushers:
      - Type: flusher_stdout
        OnlyStdout: true
      - Type: flusher_kafka_v2
        Brokers:
          - nameserver.jmq.jd.local:80 # jmq元數(shù)據(jù)地址
        Topic: ai-middle-k8s-log-prod # jmq topic 
        ClientID: ai4middle4log # Kafka的用戶ID（識(shí)別客戶端并設(shè)置其唯一性），對(duì)應(yīng)jmq的Group名稱，重要?? （https://ilogtail.gitbook.io/ilogtail-docs/plugins/input/service-kafka#cai-ji-pei-zhi-v2）   

logstash kafka&es配置

apiVersion: v1
kind: ConfigMap
metadata:
  name: logstash-config
  namespace: elastic-system
  labels:
    elastic-app: logstash
data:
  logstash.conf: |-
    input {
        kafka {
                bootstrap_servers => ["nameserver.jmq.jd.local:80"] #jmq的元數(shù)據(jù)地址
                group_id => "ai4middle4log" # jmq的Group的名稱
                client_id => "ai4middle4log" # jmq的Group的名稱，即jmq的省略了kafka中的client_id概念，用Group名稱代替
                consumer_threads => 2
                decorate_events => true
                topics => ["ai-middle-k8s-log-prod"] # jmp的topic
                auto_offset_reset => "latest"
                codec => json { charset => "UTF-8" }
        }
    }
    output {
        elasticsearch {
                hosts => ["http://x.x.x.x:40000","http://x.x.x.x:40000","http://x.x.x.x:40000"] # es地址
                index =>  "%{[@metadata][kafka][topic]}-%{+YYYY-MM-dd}" # 索引規(guī)則
                user => "XXXXXX" #jes的用戶名
                password => "xxxxx" #jes的密碼
                ssl => "false"
                ssl_certificate_verification => "false"
        }
    }

ilogtail 的配置如下

# ilogtail-daemonset.yaml
apiVersion: apps/v1
kind: DaemonSet
metadata:
  name: ilogtail-ds
  namespace: elastic-system
  labels:
    k8s-app: logtail-ds
spec:
  selector:
    matchLabels:
      k8s-app: logtail-ds
  template:
    metadata:
      labels:
        k8s-app: logtail-ds
    spec:
      dnsPolicy: "None"
      dnsConfig:
        nameservers:
          - x.x.x.x # （京東云主機(jī)上）可以解析jmq域名的nameserver
      tolerations:
        - operator: Exists                    # deploy on all nodes
      containers:
        - name: logtail
          env:
            - name: ALIYUN_LOG_ENV_TAGS       # add log tags from env
              value: _node_name_|_node_ip_
            - name: _node_name_
              valueFrom:
                fieldRef:
                  apiVersion: v1
                  fieldPath: spec.nodeName
            - name: _node_ip_
              valueFrom:
                fieldRef:
                  apiVersion: v1
                  fieldPath: status.hostIP
            - name: cpu_usage_limit           # iLogtail's self monitor cpu limit
              value: "1"
            - name: mem_usage_limit           # iLogtail's self monitor mem limit
              value: "512"
          image: dockerhub.ai.jd.local/ai-middleware/ilogtail-community-edition/ilogtail:1.3.1
          imagePullPolicy: IfNotPresent
          resources:
            limits:
              cpu: 1000m
              memory: 1Gi
            requests:
              cpu: 400m
              memory: 384Mi
          volumeMounts:
            - mountPath: /var/run                       # for container runtime socket
              name: run
            - mountPath: /logtail_host                  # for log access on the node
              mountPropagation: HostToContainer
              name: root
              readOnly: true
            - mountPath: /usr/local/ilogtail/checkpoint # for checkpoint between container restart
              name: checkpoint
            - mountPath: /usr/local/ilogtail/user_yaml_config.d # mount config dir
              name: user-config
              readOnly: true
            - mountPath: /usr/local/ilogtail/apsara_log_conf.json
              name: apsara-log-config
              readOnly: true
              subPath: apsara_log_conf.json
      dnsPolicy: ClusterFirst
      hostNetwork: true
      volumes:
        - hostPath:
            path: /var/run
            type: Directory
          name: run
        - hostPath:
            path: /
            type: Directory
          name: root
        - hostPath:
            path: /etc/ilogtail-ilogtail-ds/checkpoint
            type: DirectoryOrCreate
          name: checkpoint
        - configMap:
            defaultMode: 420
            name: ilogtail-user-cm
          name: user-config
        - configMap:
            defaultMode: 420
            name: ilogtail-apsara-log-config-cm
          name: apsara-log-config

# ilogtail-user-configmap.yaml
apiVersion: v1
kind: ConfigMap
metadata:
  name: ilogtail-user-cm
  namespace: elastic-system
data:
  app_stdout.yaml: |
    enable: true
    inputs:
      - Type: service_docker_stdout
        Stderr: true
        Stdout: true
        K8sNamespaceRegex: ai-train
        ExternalK8sLabelTag:
          platform/resource-name: k8s_label_resource-name
          platform/task-identify: k8s_label_task-identify
          task-id: k8s_label_task-id
          run-id: k8s_label_run-id
          request-id: k8s_label_request-id
    processors:
      - Type: processor_rename
        SourceKeys:
          - k8s_label_resource-name
          - k8s_label_task-identify
          - k8s_label_task-id
          - k8s_label_run-id
          - k8s_label_request-id
          - _namespace_
          - _image_name_
          - _pod_uid_
          - _pod_name_
          - _container_name_
          - _container_ip_
          - __path__
          - _source_
        DestKeys:
          - resource_name
          - task_identify
          - task_id
          - run_id
          - request_id
          - namespace
          - image_name
          - pod_uid
          - pod_name
          - container_name
          - container_ip
          - path
          - source
    flushers:
      - Type: flusher_stdout
        OnlyStdout: true
      - Type: flusher_kafka_v2
        Brokers:
          - nameserver.jmq.jd.local:80 # jmq元數(shù)據(jù)地址
        Topic: ai-middle-k8s-log-prod # jmq topic 
        ClientID: ai4middle4log # Kafka的用戶ID（識(shí)別客戶端并設(shè)置其唯一性），對(duì)應(yīng)jmq的Group名稱，重要?? （https://ilogtail.gitbook.io/ilogtail-docs/plugins/input/service-kafka#cai-ji-pei-zhi-v2）

  app_file_log.yaml: |
    enable: true
    inputs:
      - Type: file_log
        LogPath: /export/Logs/ai-dt-algorithm-tools
        FilePattern: "*.log"
        ContainerInfo:
          K8sNamespaceRegex: ai-train
          ExternalK8sLabelTag:
            platform/resource-name: k8s_label_resource-name
            platform/task-identify: k8s_label_task-identify
            task-id: k8s_label_task-id
            run-id: k8s_label_run-id
            request-id: k8s_label_request-id

    processors:
      - Type: processor_add_fields
        Fields:
          source: file
      - Type: processor_rename
        SourceKeys:
          - __tag__:k8s_label_resource-name
          - __tag__:k8s_label_task-identify
          - __tag__:k8s_label_task-id
          - __tag__:k8s_label_run-id
          - __tag__:k8s_label_request-id
          - __tag__:_namespace_
          - __tag__:_image_name_
          - __tag__:_pod_uid_
          - __tag__:_pod_name_
          - __tag__:_container_name_
          - __tag__:_container_ip_
          - __tag__:__path__
        DestKeys:
          - resource_name
          - task_identify
          - task_id
          - run_id
          - request_id
          - namespace
          - image_name
          - pod_uid
          - pod_name
          - container_name
          - container_ip
          - path

    flushers:
      - Type: flusher_stdout
        OnlyStdout: true
      - Type: flusher_kafka_v2
        Brokers:
          - nameserver.jmq.jd.local:80
        Topic: ai-middle-k8s-log-prod
        ClientID: ai4middle4log

logstash 的配置如下

# logstash-configmap.yaml
---
apiVersion: v1
kind: ConfigMap
metadata:
  name: logstash-config
  namespace: elastic-system
  labels:
    elastic-app: logstash
data:
  logstash.conf: |-
    input {
        kafka {
                bootstrap_servers => ["nameserver.jmq.jd.local:80"] #jmq的元數(shù)據(jù)地址
                #group_id => "services"
                group_id => "ai4middle4log" # jmq的Group的名稱
                client_id => "ai4middle4log" # jmq的Group的名稱，即jmq的省略了kafka中的client_id概念，用Group名稱代替
                consumer_threads => 2
                decorate_events => true
                #topics_pattern => ".*"
                topics => ["ai-middle-k8s-log-prod"] # jmp的topic
                auto_offset_reset => "latest"
                codec => json { charset => "UTF-8" }
        }
    }


    filter {
      ruby {
          code => "event.set('index_date', event.get('@timestamp').time.localtime + 8*60*60)"
      }
      ruby {
          code => "event.set('message',event.get('contents'))"
      }
      #ruby {
      #    code => "event.set('@timestamp',event.get('time').time.localtime)"
      #}

      mutate {
          remove_field => ["contents"]
          convert => ["index_date", "string"]
          #convert => ["@timestamp", "string"]
          gsub => ["index_date", "T.*Z",""]
          #gsub => ["@timestamp", "T.*Z",""]
      }
    }


    output {
        elasticsearch {
                #hosts => ["https://ai-middle-cluster-es-http:9200"]
                hosts => ["http://x.x.x.x:40000","http://x.x.x.x:40000","http://x.x.x.x:40000"] # es地址
                index =>  "%{[@metadata][kafka][topic]}-%{+YYYY-MM-dd}" # 索引規(guī)則
                user => "XXXXXX" #jes的用戶名
                password => "xxxxx" #jes的密碼
                ssl => "false"
                ssl_certificate_verification => "false"
                #cacert => "/usr/share/logstash/cert/ca_logstash.cer"
        }
        stdout {
            codec => rubydebug
        }
    }

4.核心價(jià)值

在私有化部署的基礎(chǔ)上通過(guò)簡(jiǎn)單改造實(shí)現(xiàn)了與京東內(nèi)部中間件的完美融合，使得系統(tǒng)在高可用性上適應(yīng)性更強(qiáng)、可用范圍更廣。

審核編輯 黃宇