云上Kubernetes日志采集Linux架构演进：从Fluentd到Filebeat+轻量化Elasticsearch

apiVersion: apps/v1
kind: DaemonSet
metadata:
  name: filebeat
spec:
  selector:
    matchLabels:
      app: filebeat
  template:
    spec:
      containers:
      - name: filebeat
        image: docker.elastic.co/beats/filebeat:7.17.5
        args: ["-c", "/etc/filebeat/filebeat.yml", "-e"]
        env:
        - name: NODE_NAME
          valueFrom:
            fieldRef:
              fieldPath: spec.nodeName
        volumeMounts:
        - name: varlog
          mountPath: /var/log
        - name: dockercontainers
          mountPath: /var/lib/docker/containers
          readOnly: true
        - name: config
          mountPath: /etc/filebeat
      volumes:
      - name: varlog
        hostPath:
          path: /var/log
      - name: dockercontainers
        hostPath:
          path: /var/lib/docker/containers
      - name: config
        configMap:
          name: filebeat-config

filebeat.inputs:
- type: container
  paths:
    - /var/log/containers/*.log
  processors:
    - add_kubernetes_metadata:
        host: ${NODE_NAME}
        matchers:
        - logs_path:
            logs_path: "/var/log/containers/"
    - drop_fields:
        fields: ["kubernetes.labels", "kubernetes.annotations", "host"]  # 裁剪无用字段
    - rename:
        fields:
          - from: "kubernetes.pod.name"
            to: "pod_name"
    - decode_json_fields:
        fields: ["message"]
        target: "json"
        overwrite_keys: true

output.kafka:
  hosts: ["kafka-broker:9092"]
  topic: "k8s-logs"
  partition.round_robin:
    reachable_only: false
  required_acks: 1
  compression: gzip
  max_message_bytes: 1000000

queue.mem:
  events: 4096
  flush.min_events: 2048

PUT _ilm/policy/k8s-logs-policy
{
  "policy": {
    "phases": {
      "hot": {
        "actions": {
          "rollover": {
            "max_primary_shard_size": "30GB",
            "max_age": "1d"
          }
        }
      },
      "warm": {
        "min_age": "2d",
        "actions": {
          "shrink": {
            "number_of_shards": 1
          },
          "forcemerge": {
            "max_num_segments": 1
          }
        }
      },
      "cold": {
        "min_age": "4d",
        "actions": {
          "freeze": {}
        }
      },
      "delete": {
        "min_age": "7d",
        "actions": {
          "delete": {}
        }
      }
    }
  }
}

PUT _index_template/k8s-logs-template
{
  "index_patterns": ["k8s-logs-*"],
  "template": {
    "settings": {
      "number_of_shards": 3,
      "number_of_replicas": 1,
      "index.mapping.total_fields.limit": 1000,
      "index.query.default_field": "message",
      "index.mapping.doc_values": false,  // 全局默认关闭
      "index.mapping.all": false
    },
    "mappings": {
      "properties": {
        "@timestamp": { "type": "date" },
        "level": { "type": "keyword" },
        "message": { "type": "text" },
        "service": { "type": "keyword" },
        "pod_name": { "type": "keyword" },
        "namespace": { "type": "keyword" }
        // 其他字段关闭doc_values
      }
    }
  }
}

// 伪代码逻辑
consumer := sarama.NewConsumer(brokers, config)
for msg := range consumer.Partitions() {
    batch := []elastic.BulkRequest{}
    for _, record := range records {
        req := elastic.NewBulkIndexRequest().Index("k8s-logs-2026-08-07").Doc(record)
        batch = append(batch, req)
    }
    bulkService := client.Bulk().Add(batch...)
    bulkService.Do(ctx)
}