feat:Add KafkaSerializer with unit tests #2297

ChaoEcho · 2025-07-10T04:09:25Z

Add KafkaSerializer with unit tests to support serializing events to JSON format. Implement conversion for logs, metrics, spans, and raw events, including label transformation logic for Kubernetes environments.

CLAassistant · 2025-07-10T04:09:32Z

All committers have signed the CLA.

…s to JSON format. Implement conversion for logs, metrics, spans, and raw events, including label transformation logic for Kubernetes environments.

Abingcbc · 2025-07-14T02:04:41Z

core/collection_pipeline/serializer/KafkaSerializer.cpp

+
+const string TAG_PREFIX = "__tag__:";
+
+const unordered_map<string, string> TAG_CONVERSION_MAP = {{"__path__", "log.file.path"},


C++的tag转换在其他地方，这里不需要考虑这些

Abingcbc · 2025-07-14T02:15:43Z

core/collection_pipeline/serializer/KafkaSerializer.cpp

+        const bool isK8s = getenv("KUBERNETES_SERVICE_HOST") != nullptr;
+        for (const auto& tag : group.mTags.mInner) {
+            const string key = tag.first.to_string();
+            if (key.empty()) {
+                continue;
+            }
+
+            string convertedKey = key;
+            bool converted = false;
+
+            // Try K8s-specific conversion first
+            if (isK8s) {
+                auto it = SPECIAL_TAG_CONVERSION_MAP.find(key);
+                if (it != SPECIAL_TAG_CONVERSION_MAP.end()) {
+                    convertedKey = it->second;
+                    converted = true;
+                }
+            }
+
+            // Try general conversion if not already converted
+            if (!converted) {
+                auto it = TAG_CONVERSION_MAP.find(key);
+                if (it != TAG_CONVERSION_MAP.end()) {
+                    convertedKey = it->second;
+                }
+            }
+
+            if (!convertedKey.empty()) {
+                item.tags[convertedKey] = tag.second.to_string();
+            }
+        }


C++这边不需要

Abingcbc · 2025-07-14T02:17:23Z

core/collection_pipeline/serializer/KafkaSerializer.cpp

+    }
+
+    vector<CustomSingleLogItem> items;
+    ConvertToCustomSingleItems(group, items);


PipelineEventGroup可以直接序列化成json，为什么还需要中间这个结果？

Abingcbc · 2025-07-14T02:18:40Z

core/collection_pipeline/serializer/KafkaSerializer.cpp

+}
+
+
+bool KafkaEventGroupSerializer::SerializeToJSON(const vector<CustomSingleLogItem>& items, string& output) {


可以参考一下PipelineEventGroup::ToJson，这个是单测里用的。这两种序列化哪种性能更好？

shalousun · 2025-07-14T03:24:55Z

很多通用处理不应该定义的时候和kafka绑定，go里面的这块协议转换实际上kafka、pulsar、es可能都在用

- Implemented serialization for logs, metrics, spans, and raw events - Added corresponding unit tests - Removed KafkaSerializer-related code to simplify serialization logic

Abingcbc · 2025-07-17T02:44:48Z

core/collection_pipeline/serializer/CustomSingleSerializer.cpp

+const string PROTOCOL_KEY_TIME = "time";
+const string PROTOCOL_KEY_CONTENT = "contents";
+const string PROTOCOL_KEY_TAG = "tags";
+
+const string TAG_PREFIX = "__tag__:";


kXxxx这种命名规则

Abingcbc · 2025-07-17T02:47:38Z

core/collection_pipeline/serializer/CustomSingleSerializer.cpp

+                    if (kv.first.size() >= TAG_PREFIX.length()
+                        && strncmp(kv.first.data(), TAG_PREFIX.c_str(), TAG_PREFIX.length()) == 0) {
+                        continue;
+                    }


C++这边的tag一定在tag里面，不用这段特殊处理

Abingcbc · 2025-07-17T02:48:19Z

core/collection_pipeline/serializer/CustomSingleSerializer.cpp

+                if (kv.first.size() > TAG_PREFIX.length()
+                    && strncmp(kv.first.data(), TAG_PREFIX.c_str(), TAG_PREFIX.length()) == 0) {
+                    StringView tagName(kv.first.data() + TAG_PREFIX.length(), kv.first.size() - TAG_PREFIX.length());
+                    if (tagName != "__user_defined_id__" && !tagName.empty()) {
+                        writer.Key(tagName.data(), tagName.size());
+                        writer.String(kv.second.data(), kv.second.size());
+                    }
+                }


同上，不用特殊处理

Abingcbc · 2025-07-17T02:51:30Z

core/collection_pipeline/serializer/CustomSingleSerializer.cpp

+            const auto& metricEvent = event.Cast<MetricEvent>();
+            for (auto it = metricEvent.TagsBegin(); it != metricEvent.TagsEnd(); ++it) {
+                writer.Key(it->first.data(), it->first.size());
+                writer.String(it->second.data(), it->second.size());
+            }


这里不太对。你上面加的tag是group级别的，metric这里的tag是event级别的。不能加到一起。

Abingcbc · 2025-07-17T02:53:29Z

core/unittest/serializer/CustomSingleSerializerUnittest.cpp

+#ifdef _WIN32
+#include <windows.h>
+#endif


这个是干什么的？

- rename protocol key constants - add support for metric_tag

ChaoEcho · 2025-07-17T14:28:34Z

JSON序列化规范

1. 通用JSON结构

无论内部事件类型如何，所有事件在序列化后都遵循统一的顶层JSON结构。这个结构包含三个固定的键：time, contents, 和 tags。

{
  "time": 1678886400,
  "contents": {
    // ...
  },
  "tags": {
    // ...
  }
}

time: 事件时间戳（Unix秒），number类型。
contents: 一个JSON object，包含了事件的核心数据。其内部结构因事件类型而异。
tags: 一个JSON object，包含了整个事件组的元数据标签。

2. 事件类型序列化详解

2.1 LogEvent (日志事件)

序列化前:

核心数据是一个map<string, string>，包含多个键值对。

序列化后:

{
  "time": 1678886400,
  "contents": {
    "message": "user login successful",
    "user_id": "1001",
    "level": "info"
  },
  "tags": {
    "source": "/var/log/app.log",
    "host": "web-server-01"
  }
}

2.2 MetricEvent (指标事件)

序列化前:

指标名 name (string)
指标值 value (double 或 map<string, double>)
事件级标签 tags (map<string, string>)

序列化后:

{
  "time": 1678886401,
  "contents": {
    "__name__": "cpu.usage.percent",
    "__value__": "85.5",
    "metric_tags": {
      "cpu": "0"
    }
  },
  "tags": {
    "region": "cn-hangzhou",
    "host": "metric-server-02"
  }
}

2.3 SpanEvent (链路追踪事件)

序列化前:

traceId, spanId, parentSpanId, name (string)
startTimeNs, endTimeNs (uint64_t, 纳秒)
事件级标签 attributes (map<string, string>)
范围标签 scope (map<string, string>)

序列化后:

{
  "time": 1678886402,
  "contents": {
    "trace_id": "7a3f1c2b4d5e6f7a",
    "span_id": "9b8c7d6e5f4a3b2c",
    "parent_span_id": "1a2b3c4d5e6f7a8b",
    "name": "GET /api/users",
    "start_time": "1234567890123456789", 
    "end_time": "1234567890123457890",  
    "duration": "1101",                 
    "attributes": {
      "http.method": "GET",
      "http.status_code": "200"
    },
    "scope": {
      "service.name": "user-service"
    }
  },
  "tags": {
    "env": "production",
    "region": "us-west-2"
  }
}

2.4 RawEvent (原始事件)

序列化前:

核心数据是一个 content (string)。

序列化后:

{
  "time": 1678886403,
  "contents": {
    "content": "this is a raw log line"
  },
  "tags": {
    "source": "/var/log/raw.log",
    "host": "server-03"
  }
}

- Remove unused TestHostTags tests - Simplify event creation function parameters - Ensure events always include required tags

feat:Add KafkaSerializer with unit tests to support serializing event…

37872bb

…s to JSON format. Implement conversion for logs, metrics, spans, and raw events, including label transformation logic for Kubernetes environments.

ChaoEcho force-pushed the feat/task-1-serializer branch from 572870a to 37872bb Compare July 10, 2025 05:08

Abingcbc requested changes Jul 14, 2025

View reviewed changes

feat: Add CustomSingleSerializer to support JSON serialization of events

730391a

- Implemented serialization for logs, metrics, spans, and raw events - Added corresponding unit tests - Removed KafkaSerializer-related code to simplify serialization logic

Abingcbc requested changes Jul 17, 2025

View reviewed changes

feat:Optimize CustomSingleSerializer

8a94d91

- rename protocol key constants - add support for metric_tag

feat:Optimize CustomSingleSerializer unit tests

f584af9

- Remove unused TestHostTags tests - Simplify event creation function parameters - Ensure events always include required tags

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

feat:Add KafkaSerializer with unit tests #2297

feat:Add KafkaSerializer with unit tests #2297

Uh oh!

ChaoEcho commented Jul 10, 2025

Uh oh!

CLAassistant commented Jul 10, 2025 •

edited

Loading

Uh oh!

Abingcbc Jul 14, 2025

Uh oh!

Abingcbc Jul 14, 2025

Uh oh!

Abingcbc Jul 14, 2025

Uh oh!

Abingcbc Jul 14, 2025

Uh oh!

shalousun commented Jul 14, 2025

Uh oh!

Abingcbc Jul 17, 2025

Uh oh!

Abingcbc Jul 17, 2025

Uh oh!

Abingcbc Jul 17, 2025

Uh oh!

Abingcbc Jul 17, 2025

Uh oh!

Abingcbc Jul 17, 2025

Uh oh!

ChaoEcho commented Jul 17, 2025

Uh oh!

Uh oh!


		const string TAG_PREFIX = "__tag__:";

		const unordered_map<string, string> TAG_CONVERSION_MAP = {{"__path__", "log.file.path"},

		}


		bool KafkaEventGroupSerializer::SerializeToJSON(const vector<CustomSingleLogItem>& items, string& output) {

feat:Add KafkaSerializer with unit tests #2297

Are you sure you want to change the base?

feat:Add KafkaSerializer with unit tests #2297

Uh oh!

Conversation

ChaoEcho commented Jul 10, 2025

Uh oh!

CLAassistant commented Jul 10, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

shalousun commented Jul 14, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ChaoEcho commented Jul 17, 2025

JSON序列化规范

1. 通用JSON结构

2. 事件类型序列化详解

2.1 LogEvent (日志事件)

2.2 MetricEvent (指标事件)

2.3 SpanEvent (链路追踪事件)

2.4 RawEvent (原始事件)

Uh oh!

Uh oh!

CLAassistant commented Jul 10, 2025 •

edited

Loading