From b9e433a76960687ab1c0ff5c9583017734f676d9 Mon Sep 17 00:00:00 2001
From: "zhangkun9038@dingtalk.com" <zhangkun9038@dingtalk.com>
Date: Mon, 10 Mar 2025 11:20:33 +0800
Subject: [PATCH] up

---
 efk/fluentd-configMap2.yaml | 72 ++++++++++++++++++-------------------
 efk/fluentd.conf            | 41 +++++++++++++++------
 efk/fluentd1.conf           | 60 +++++++++++++++++++++++++++++++
 efk/myFluentd.Dockerfile    |  2 ++
 4 files changed, 128 insertions(+), 47 deletions(-)
 create mode 100644 efk/fluentd1.conf
 create mode 100644 efk/myFluentd.Dockerfile

diff --git a/efk/fluentd-configMap2.yaml b/efk/fluentd-configMap2.yaml
index 759b2e7..1183c77 100644
--- a/efk/fluentd-configMap2.yaml
+++ b/efk/fluentd-configMap2.yaml
@@ -9,46 +9,12 @@ data:
       @type http
       @id input_http
       port 8888
-      tag sardine.log  # 初始 tag，客户端需指定完整 tag 如 sardine.log.candle.BTC-USDT.15M
       @label @main
-      <parse>
-        @type json
-      </parse>
     </source>
 
     <label @main>
-      <filter sardine.log.candle.**>
-        @type record_transformer
-        enable_ruby true
-        <record>
-          # 注入调试信息
-          debug_tag "${tag}"
-          debug_record "#{record.to_json rescue 'record is nil'}"
-          # 处理 tag 和 timestamp
-          coin_pair "${tag.split('.')[3] || 'UNKNOWN'}"
-          timeframe "${tag.split('.')[4] || 'UNKNOWN'}"
-          year "#{Time.parse(record['timestamp']).strftime('%Y') rescue 'invalid timestamp'}"
-          index_name "logstash_candle_${tag.split('.')[3]}_#{Time.parse(record['timestamp']).strftime('%Y') rescue 'invalid timestamp'}_${tag.split('.')[4]}"
-        </record>
-      </filter>
-
-      <match sardine.log.candle.**>
+      <match sardine.log.**>
         @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 %{index_name}
-          id_key _id
-          flush_interval 5s
-          @log_level debug
-          remove_keys _id, coin_pair, timeframe, year, index_name, debug_tag, debug_record
-        </store>
         <store>
           @type elasticsearch
           @id output_elasticsearch
@@ -60,9 +26,9 @@ data:
           logstash_format true
           logstash_prefix logstash
           logstash_dateformat %Y.%m.%d
-          id_key _id
           flush_interval 5s
           @log_level debug
+          id_key _id
           remove_keys _id
         </store>
         <store>
@@ -70,6 +36,40 @@ data:
           @id output_stdout
         </store>
       </match>
+
+      <match tanya.**>
+        @type rewrite_tag_filter
+        @id rewrite_tag_for_tanya
+        <rule>
+          key @log_name
+          pattern ^tanya\.candle\.([^\.]+)\.([^\.]+)\.([^\.]+)$
+          tag candle.$1.$2.$3
+        </rule>
+      </match>
+
+      <match candle.**>
+        @type copy
+        <store>
+          @type elasticsearch
+          @id output_elasticsearch_tanya
+          host elasticsearch
+          port 9200
+          scheme http
+          user fluentd_user
+          password fluentd_password
+          index_name candle_${tag_parts[1]}_${tag_parts[2]}_${tag_parts[3]}
+          flush_interval 5s
+          @log_level debug
+          id_key _id
+          remove_keys _id
+          include_tag_key true
+          tag_key @log_name
+        </store>
+        <store>
+          @type stdout
+          @id output_stdout_tanya
+        </store>
+      </match>
     </label>
 
     <match **>
diff --git a/efk/fluentd.conf b/efk/fluentd.conf
index 5f8fd5e..0fdb084 100644
--- a/efk/fluentd.conf
+++ b/efk/fluentd.conf
@@ -3,18 +3,37 @@
   @id input_http
   port 8888
   tag sardine.log
+  @label @main
 </source>
 
-<match sardine.log>
-  @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
+<label @main>
+  <match sardine.log.**>
+    @type copy
+    <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
+      flush_interval 5s  # 缩短批量写入间隔
+      @log_level debug
+      id_key _id
+      remove_keys _id
+    </store>
+    <store>
+      @type stdout
+      @id output_stdout
+    </store>
+  </match>
+</label>
+
+<match **>
+  @type stdout
+  @id output_stdout_all
 </match>
 
diff --git a/efk/fluentd1.conf b/efk/fluentd1.conf
new file mode 100644
index 0000000..c5fb850
--- /dev/null
+++ b/efk/fluentd1.conf
@@ -0,0 +1,60 @@
+  <source>
+    @type http
+    @id input_http
+    port 8888
+    @label @main
+  </source>
+  
+  <label @main>
+    <match sardine.log.**>
+      @type copy
+      <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
+        flush_interval 5s
+        @log_level debug
+        id_key _id
+        remove_keys _id
+      </store>
+      <store>
+        @type stdout
+        @id output_stdout
+      </store>
+    </match>
+  
+    <match tanya.**>
+      @type copy
+      <store>
+        @type elasticsearch
+        @id output_elasticsearch_tanya
+        host elasticsearch
+        port 9200
+        scheme http
+        user fluentd_user
+        password fluentd_password
+        logstash_format false
+        index_name logstash_candle_${tag_parts[1]}_${tag_parts[2]}_${tag_parts[3]}
+        flush_interval 5s
+        @log_level debug
+        id_key _id
+        remove_keys _id
+      </store>
+      <store>
+        @type stdout
+        @id output_stdout_tanya
+      </store>
+    </match>
+  </label>
+  
+  <match **>
+    @type stdout
+    @id output_stdout_all
+  </match>
diff --git a/efk/myFluentd.Dockerfile b/efk/myFluentd.Dockerfile
new file mode 100644
index 0000000..22ea2ba
--- /dev/null
+++ b/efk/myFluentd.Dockerfile
@@ -0,0 +1,2 @@
+FROM fluent/fluentd-kubernetes-daemonset:v1.16-debian-elasticsearch8-2
+RUN fluent-gem install fluent-plugin-rewrite-tag-filter