fluentd daemon update

efk/fluentd-daemonset.yaml

@@ -1,56 +1,75 @@
-apiVersion: apps/v1
+apiVersion: v1
-kind: DaemonSet
+kind: ConfigMap
 metadata:
-  name: fluentd
+  name: fluentd-config
   namespace: efk
-  labels:
+data:
-    k8s-app: fluentd-logging
+  fluent.conf: |
-    version: v1
+    <source>
-spec:
+      @type http
-  selector:
+      @id input_http
-    matchLabels:
+      port 8888
-      k8s-app: fluentd-logging
+      tag sardine.log  # 初始 tag，客户端需指定完整 tag 如 sardine.log.candle.BTC-USDT.15M
-      version: v1
+      @label @main
-  template:
+      <parse>
-    metadata:
+        @type json
-      labels:
+      </parse>
-        k8s-app: fluentd-logging
+    </source>
-        version: v1
-    spec:
-      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.17.1-debian-elasticsearch8-1.0
-        env:
-          - name: K8S_NODE_NAME
-            valueFrom:
-              fieldRef:
-                fieldPath: spec.nodeName
-          - name: TZ
-            value: Asia/Shanghai  # 设置时区为东八区
-        resources:
-          limits:
-            memory: 200Mi
-          requests:
-            cpu: 100m
-            memory: 200Mi
-        volumeMounts:
-        - name: fluentd-config-volume
-          mountPath: /fluentd/etc/fluent.conf
-          subPath: fluent.conf
-        - name: timezone
-          mountPath: /etc/localtime
-          readOnly: true
-      terminationGracePeriodSeconds: 30
-      volumes:
-      - name: fluentd-config-volume
-        configMap:
-          name: fluentd-config
-      - name: timezone
-        hostPath:
-          path: /usr/share/zoneinfo/Asia/Shanghai  # 挂载东八区时区文件
 
+    <label @main>
+      <filter sardine.log.candle.**>
+        @type record_transformer
+        enable_ruby true
+        <record>
+          coin_pair ${tag_parts[3] || "UNKNOWN"}  # 从 tag 第 4 部分提取，如 BTC-USDT
+          timeframe ${tag_parts[4] || "UNKNOWN"}  # 从 tag 第 5 部分提取，如 15M
+          year ${time.strftime("%Y")}             # 从 timestamp 提取年份
+          index_name "logstash_candle_${tag_parts[3]}_${time.strftime('%Y')}_${tag_parts[4]}"
+        </record>
+      </filter>
+
+      <match sardine.log.candle.**>
+        @type copy
+        <store>
+          @type elasticsearch
+          @id output_elasticsearch_custom
+          host elasticsearch
+          port 9200
+          scheme http
+          user fluentd_user
+          password fluentd_password
+          logstash_format false
+          index_name ${record["index_name"]}  # 动态索引，如 logstash_candle_BTC-USDT_2024_15M
+          id_key _id                          # 从 URL 参数获取 _id，与原始逻辑一致
+          flush_interval 5s
+          @log_level debug
+          remove_keys _id, coin_pair, timeframe, year, index_name
+        </store>
+        # 保留原有按日期切分的输出
+        <store>
+          @type elasticsearch
+          @id output_elasticsearch
+          host elasticsearch
+          port 9200
+          scheme http
+          user fluentd_user
+          password fluentd_password
+          logstash_format true
+          logstash_prefix logstash
+          logstash_dateformat %Y.%m.%d
+          id_key _id                          # 从 URL 参数获取 _id
+          flush_interval 5s
+          @log_level debug
+          remove_keys _id
+        </store>
+        <store>
+          @type stdout
+          @id output_stdout
+        </store>
+      </match>
+    </label>
+
+    <match **>
+      @type stdout
+      @id output_stdout_all
+    </match>