[1928-b4] Processor 迁移 + flusher_sls v2#2664
Open
Takuka0311 wants to merge 18 commits into
Open
Conversation
Migrate flusher_sls and a first batch of field-transform processors to the v2 pipeline as part of Epic alibaba#2595 step B4 (delivered as one PR per maintainer directive; further processor subdirs land on this PR later). - flusher_sls: implement FlusherV2.Export, converting each PipelineGroupEvents to a legacy LogGroup via B3's PipelineGroupEventsToLogGroup and reusing the SendPb cgo path. Flush and Export now share a sendLogGroup helper. Adds an injectable send seam so the path is unit-testable without a live Logtail adaptor. - processors add_fields / drop / drop_last_key / md5 / packjson: implement ProcessorV2.Process on models.Log, passing Metric and Span events through unchanged via pipeline.ProcessLogEventsOnly. - Makefile: stage libGoPluginAdapter.so for the flusher_sls test binary (mirrors the plugin_main / pluginmanager copies) and clean it up. - docs: refresh the B2 plugin-event matrix rows for the migrated processors (v1&v2, Metric/Span passthrough). Does not change the default StructureType. make unittest_plugin passes locally. Fixes alibaba#2603 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
3 tasks
Takuka0311
commented
Jul 20, 2026
Every extended processor now implements the v2 ProcessorV2 interface so none is v1-only or has an undefined v2 pass-through status. Real v2 log processing (Metric/Span passed through) is added for the in-place value/field transforms: appender, dict_map, string_replace, encrypt, desensitize, base64 encoding/decoding. encrypt/desensitize are implemented (not passed through) so v2 pipelines never leak sensitive data silently. Parsers, converters, filters, enrichers and time-parsers gain an explicit full pass-through Process (Log/Metric/Span emitted unchanged) so they load in v2 Runners with a documented status instead of "pass-through unknown". overview.md is updated: no processor row is v1-only or "透传未知"; missing rows (anchor, base64, csv, drop_last_key, filter_key_regex, geoip, md5, pick_key, split_log_string) are added. Adds v2 unit tests for every real port plus representative pass-through tests. Verified locally with go build/vet and go test ./plugins/processor/... Fixes alibaba#2603
Takuka0311
commented
Jul 20, 2026
Takuka0311
commented
Jul 20, 2026
Takuka0311
commented
Jul 20, 2026
Takuka0311
commented
Jul 20, 2026
Address review r3612298784: undocumented plugins need docs, and docs must stay consistent with the overview event-capability matrix. Cover every processor that already has a real v2 Process in this PR: - create docs for base64_decoding/base64_encoding/drop_last_key/md5/ split_log_string (previously 暂无文档) - sync 事件能力 tables of add_fields/appender/desensitize/dict_map/ drop/encrypt/packjson/string_replace to v1&v2 + Metric/Span 透传, matching overview.md - wire new docs into SUMMARY.md TOC and overview.md links Passthrough-stub processors (anchor/csv/regex/grok/split_*/strptime/ otel/log_to_sls_metric/geoip/... ) still need real v2 before their docs/v2 samples can be written; tracked for follow-up batches. Refs alibaba#2603
Per maintainer review, the v1/v2 processor split only differs by Process (v2) vs ProcessLogs (v1); a separate *_v2.go file per plugin is unnecessary. Move each v2 Process method into its main plugin file, mirroring flusher_sls (Flush + Export in one file), and drop the 19 *_v2.go implementation files. Test files keep the *_v2_test.go name. No behavior change: the merged methods are unchanged passthrough implementations. Fixes alibaba#2603
processor_split_log_string previously dropped non-Log events in the v2 Process path. Collect Metric/Span events unchanged so they pass through instead of being silently discarded, and sync docs event-capability tables accordingly. Fixes alibaba#2603
Migrate the remaining pass-through-stub extended processors to real v2 Process implementations, reusing v1 logic on the models.Log path so each plugin actually processes v2 pipeline events instead of only forwarding them. - Log->Log (in-place): anchor, csv, geoip, grok, regex, strptime, gotime, split_char/string/key_value, pick_key, fields_with_condition. - Drop semantics kept: filter_key_regex, filter_regex, rate_limit, pick_key (emptied log) and fields_with_condition (DropIfNotMatchCondition) drop only Log events; Metric/Span pass through unchanged. - Type-change: log_to_sls_metric and otel_metric emit models.Metric, otel_trace emits models.Span; cloud_meta enriches group Tags for all event types. - split_log_regex splits one Log into many (1->N). Add v2 unit tests per plugin. Add 5 missing plugin docs (anchor, csv, filter_key_regex, geoip, pick_key), add v2 event-capability sections to existing docs, and update overview.md annotations and SUMMARY.md. Refs alibaba#2603 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Takuka0311
commented
Jul 20, 2026
Address PR review r3613003617 (Option B + e2e simplification): - Add processor_v2 pure-v2 log pipeline case: metric_debug_file (Log input) -> processors -> flusher_grpc, asserting converted contents so migrated v2 Process paths are really exercised (split_key_value/add_fields/string_replace/appender, regex/ base64/md5, split_char/dict_map, csv, grok). - Upgrade flusher_http case to a pure v2 pipeline (FlusherV2.Export). - Delete input_mock_log/input_mock_metric (redundant with the v2 metric cases) and aggregator_context (no aggregators in v2). - Keep only the latest Kafka broker; consolidate the plaintext compression and headers features into it as scenarios. Refs alibaba#2603 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Each extended processor/flusher doc carried a second event-capability table (LogGroup/EventType) that duplicated the v1接口/v2接口/Log/Metric/ Span table above it, and in the pre-existing default/json/rename/stdout docs the two tables even contradicted each other. Keep the single table that matches overview.md column semantics. Fixes alibaba#2603
Takuka0311
commented
Jul 20, 2026
Takuka0311
commented
Jul 20, 2026
Takuka0311
commented
Jul 20, 2026
Address review r3613904556: drop the separate *_v2_test.go files and fold their PipelineEvent/SendPb Process-path cases into each processor's existing _test.go. Where a package had no real tests before (e.g. geoip), the v2 cases now live in the plain-named _test.go with an inline comment instead of a v2 file. No production code or test logic changed. Refs alibaba#2603 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
The flusher_http v2 e2e case wrote to influxdb with Query
db="%{tag.db}", but placed db on the per-metric Tags. In v2 the
query templating resolves via findTargetFieldsInGroup(group.GetTags()),
so a per-metric tag leaves %{tag.db} empty, the write misses database
"mydb", and the subscriber finds 0 records (expect at least 10, got 0).
Move db into GroupTags so the group-tag query routing resolves to
"mydb", matching the v1 case that grouped by __tag__:db. Verified
locally: TestFlusherHTTPExportV2 passes (7/7 steps).
Fixes alibaba#2603
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
v2 log bodies are stored under "content" as []byte (Log.SetBody and the
C++->Go bridge convertToPipelineEvent), but the migrated processors read
them with fmt.Sprintf("%v", ...), which renders byte numbers instead of
the text, so csv/grok/regex/split_* parse nothing on a real v2 pipeline.
Add pipeline.GetStringValue to convert string/[]byte content and use it
at all v2 content-read sites. Unit tests missed this because they set
content as strings; the processor_v2 e2e caught it.
Also init Tags in metric_debug_file's emitted group: a bare
&models.GroupInfo{} left Tags nil, panicking ProcessorTag.ProcessV2
(Group.Tags.Add) and killing the v2 processor goroutine, so no events
reached the flusher (0 logs).
Verified locally: processor_v2 e2e suite 5/5 scenarios pass.
Refs alibaba#2603
A v2 input may emit a group whose Tags is a nil interface or the shared
Nil sentinel (e.g. a bare &models.GroupInfo{}). ProcessV2 called
in.Group.Tags.Add directly, which panics on a nil interface and silently
drops on the sentinel, taking down the processor goroutine. Guard nil
in/Group and materialize a writable Tags before adding, mirroring the
nil-Context defense already present in ProcessV1.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Separate third-party imports from local github.com/alibaba/ilogtail imports per the goimports local-prefixes config, resolving the 33 "File is not properly formatted (goimports)" lint errors. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Takuka0311
changed the title
Takuka0311
marked this pull request as ready for review
July 21, 2026 06:29
Contributor
There was a problem hiding this comment.
Pull request overview
该 PR 在 LoongCollector 的 Go 插件体系中推进 v2 Pipeline（PipelineEvent / PipelineGroupEvents）迁移：为大量 processor 补齐 v2 Process 并对非 Log 事件做显式透传；同时补强 v2 场景的通用辅助能力（如 GetStringValue、Group/Tags nil 防护），并同步更新/整理部分 e2e 用例与中文文档。
Changes:
- 为多类 processor 增加 v2
Process路径（Log 处理 + Metric/Span 透传），并补齐对应单测。 - 增强 v2 pipeline 基础设施（
pipeline.GetStringValue、ProcessorTag.ProcessV2nil 防护、input_debug_file发出可写 Group/Tags）。 - 调整 e2e 用例（Kafka broker 版本矩阵/场景合并、部分用例移除）并更新/新增多篇中文插件文档与目录。
Reviewed changes
Copilot reviewed 141 out of 151 changed files in this pull request and generated 4 comments.
Show a summary per file
| File | Description |
|---|---|
| test/e2e/test_cases/processor_v2/split_pipe.log | 新增 processor_v2 e2e 输入样例 |
| test/e2e/test_cases/processor_v2/regex_hash.log | 新增 processor_v2 e2e 输入样例 |
| test/e2e/test_cases/processor_v2/grok_line.log | 新增 processor_v2 e2e 输入样例 |
| test/e2e/test_cases/processor_v2/csv_line.log | 新增 processor_v2 e2e 输入样例 |
| test/e2e/test_cases/processor_v2/content_kv.log | 新增 processor_v2 e2e 输入样例 |
| test/e2e/test_cases/input_mock_metric/case.feature | 移除 e2e 用例 |
| test/e2e/test_cases/input_mock_log/case.feature | 移除 e2e 用例 |
| test/e2e/test_cases/flusher_kafka_native_headers/run.sh | 移除旧 Kafka headers e2e 目录内容 |
| test/e2e/test_cases/flusher_kafka_native_headers/flusher_headers.log | e2e 目录内容调整（diff 未展示） |
| test/e2e/test_cases/flusher_kafka_native_headers/Dockerfile | 移除旧 Kafka headers e2e 目录内容 |
| test/e2e/test_cases/flusher_kafka_native_headers/docker-compose.yaml | 移除旧 Kafka headers e2e 目录内容 |
| test/e2e/test_cases/flusher_kafka_native_headers/case.feature | 移除旧 Kafka headers e2e 目录内容 |
| test/e2e/test_cases/flusher_kafka_native_compression/run.sh | 移除旧 Kafka compression e2e 目录内容 |
| test/e2e/test_cases/flusher_kafka_native_compression/flusher_test_compression.log | e2e 目录内容调整（diff 未展示） |
| test/e2e/test_cases/flusher_kafka_native_compression/Dockerfile | 移除旧 Kafka compression e2e 目录内容 |
| test/e2e/test_cases/flusher_kafka_native_compression/docker-compose.yaml | 移除旧 Kafka compression e2e 目录内容 |
| test/e2e/test_cases/flusher_kafka_native_compression/case.feature | 移除旧 Kafka compression e2e 目录内容 |
| test/e2e/test_cases/flusher_kafka_native_3.x.x/case.feature | 合并 Kafka plaintext 场景到 3.x broker e2e |
| test/e2e/test_cases/flusher_kafka_native_2.x.x/run.sh | 移除旧 Kafka 2.x e2e 目录内容 |
| test/e2e/test_cases/flusher_kafka_native_2.x.x/flusher_test_2.x.x.log | e2e 目录内容调整（diff 未展示） |
| test/e2e/test_cases/flusher_kafka_native_2.x.x/Dockerfile | 移除旧 Kafka 2.x e2e 目录内容 |
| test/e2e/test_cases/flusher_kafka_native_2.x.x/docker-compose.yaml | 移除旧 Kafka 2.x e2e 目录内容 |
| test/e2e/test_cases/flusher_kafka_native_2.x.x/case.feature | 移除旧 Kafka 2.x e2e 目录内容 |
| test/e2e/test_cases/flusher_kafka_native_1.x.x/run.sh | 移除旧 Kafka 1.x e2e 目录内容 |
| test/e2e/test_cases/flusher_kafka_native_1.x.x/flusher_test_1.x.x.log | e2e 目录内容调整（diff 未展示） |
| test/e2e/test_cases/flusher_kafka_native_1.x.x/Dockerfile | 移除旧 Kafka 1.x e2e 目录内容 |
| test/e2e/test_cases/flusher_kafka_native_1.x.x/docker-compose.yaml | 移除旧 Kafka 1.x e2e 目录内容 |
| test/e2e/test_cases/flusher_kafka_native_1.x.x/case.feature | 移除旧 Kafka 1.x e2e 目录内容 |
| test/e2e/test_cases/flusher_kafka_native_0.10.x/run.sh | 移除旧 Kafka 0.10 e2e 目录内容 |
| test/e2e/test_cases/flusher_kafka_native_0.10.x/flusher_test_0.10.x.log | e2e 目录内容调整（diff 未展示） |
| test/e2e/test_cases/flusher_kafka_native_0.10.x/Dockerfile | 移除旧 Kafka 0.10 e2e 目录内容 |
| test/e2e/test_cases/flusher_kafka_native_0.10.x/docker-compose.yaml | 移除旧 Kafka 0.10 e2e 目录内容 |
| test/e2e/test_cases/flusher_kafka_native_0.10.x/case.feature | 移除旧 Kafka 0.10 e2e 目录内容 |
| test/e2e/test_cases/flusher_http/case.feature | 将 flusher_http e2e 改为验证 v2 Export 路径 |
| test/e2e/test_cases/aggregator_context/run.sh | 移除 e2e 用例目录内容 |
| test/e2e/test_cases/aggregator_context/Dockerfile_2 | 移除 e2e 用例目录内容 |
| test/e2e/test_cases/aggregator_context/Dockerfile_1 | 移除 e2e 用例目录内容 |
| test/e2e/test_cases/aggregator_context/docker-compose.yaml | 移除 e2e 用例目录内容 |
| test/e2e/test_cases/aggregator_context/case.feature | 移除 e2e 用例目录内容 |
| plugins/processor/strptime/strptime.go | strptime 增加 v2 Process（Log 处理 + 透传） |
| plugins/processor/strptime/strptime_test.go | strptime 增加 v2 路径单测 |
| plugins/processor/stringreplace/processor_string_replace.go | stringreplace 抽共享 replaceValue + 增加 v2 Process |
| plugins/processor/stringreplace/processor_string_replace_test.go | stringreplace 增加 v2 路径单测 |
| plugins/processor/split/string/split_string.go | split_string 增加 v2 Process |
| plugins/processor/split/logstring/split_log_string.go | split_log_string v2 透传非 Log 事件并整理逻辑 |
| plugins/processor/split/logstring/split_log_string_test.go | split_log_string 增加 v2 路径单测 |
| plugins/processor/split/keyvalue/key_value_splitter.go | split_key_value 抽共享解析函数 + 增加 v2 Process |
| plugins/processor/split/keyvalue/key_value_splitter_test.go | split_key_value 增加 v2 路径单测 |
| plugins/processor/split/char/split_char.go | split_char 增加 v2 Process |
| plugins/processor/ratelimit/processor_rate_limit.go | rate_limit 增加 v2 Process 与 makeKeyV2 |
| plugins/processor/ratelimit/processor_rate_limit_test.go | rate_limit 增加 v2 路径单测 |
| plugins/processor/pickkey/processor_pick_key.go | pick_key 增加 v2 Process（字段过滤/可丢弃 Log） |
| plugins/processor/packjson/processor_packjson.go | packjson 增加 v2 Process |
| plugins/processor/packjson/processor_packjson_test.go | packjson 增加 v2 路径单测 |
| plugins/processor/otel/processor_otel_trace_test.go | otel trace 增加 v2 路径单测（Log→Span/透传） |
| plugins/processor/otel/processor_otel_metric.go | otel metric 增加 v2 Process（Log→Metric/透传） |
| plugins/processor/otel/processor_otel_metric_test.go | otel metric 增加 v2 路径单测 |
| plugins/processor/md5/processor_md5.go | md5 增加 v2 Process |
| plugins/processor/md5/processor_md5_test.go | md5 增加 v2 路径单测 |
| plugins/processor/gotime/processor_gotime.go | gotime 增加 v2 Process |
| plugins/processor/gotime/processor_gotime_test.go | gotime 增加 v2 路径单测 |
| plugins/processor/geoip/geoip_test.go | geoip 增加 v2 行为单测（nil db no-op + 透传） |
| plugins/processor/filter/regex/processor_filter_regex.go | filter_regex 增加 v2 Process（Log 过滤 + 透传） |
| plugins/processor/filter/regex/filter_regex_test.go | filter_regex 增加 v2 路径单测 |
| plugins/processor/filter/keyregex/processor_filter_key_regex.go | filter_key_regex 增加 v2 Process |
| plugins/processor/filter/keyregex/filter_key_regex_test.go | filter_key_regex 增加 v2 路径单测 |
| plugins/processor/fieldswithcondition/processor_fields_with_condition.go | fields_with_condition 增加 v2 Process（条件动作 + 透传） |
| plugins/processor/fieldswithcondition/processor_fields_with_condition_test.go | fields_with_condition 增加 v2 路径单测 |
| plugins/processor/encrypt/processor_encrypt.go | encrypt 增加 v2 Process（key 缺失时整体透传） |
| plugins/processor/encrypt/processor_encrypt_test.go | encrypt 增加 v2 路径单测 |
| plugins/processor/droplastkey/processor_drop_last_key.go | drop_last_key 增加 v2 Process |
| plugins/processor/droplastkey/processor_drop_last_key_test.go | drop_last_key 增加 v2 路径单测 |
| plugins/processor/drop/processor_drop.go | drop 增加 v2 Process |
| plugins/processor/drop/processor_drop_test.go | drop 增加 v2 路径单测 |
| plugins/processor/dictmap/processor_dict_map.go | dict_map 增加 v2 Process |
| plugins/processor/dictmap/processor_dict_map_test.go | dict_map 增加 v2 路径单测 |
| plugins/processor/desensitize/processor_desensitize.go | desensitize 增加 v2 Process |
| plugins/processor/desensitize/processor_desensitize_test.go | desensitize 增加 v2 路径单测 |
| plugins/processor/csv/processor_csv.go | csv 抽共享 CSV 解码 + 增加 v2 Process |
| plugins/processor/csv/processor_csv_test.go | csv 增加 v2 路径单测 |
| plugins/processor/cloudmeta/cloud_meta.go | cloudmeta 增加 v2 Process（写 Group Tags） |
| plugins/processor/base64/encoding/processor_base64_encoding.go | base64 encoding 增加 v2 Process |
| plugins/processor/base64/encoding/processor_base64_encoding_test.go | base64 encoding 增加 v2 路径单测 |
| plugins/processor/base64/decoding/processor_base64_decoding.go | base64 decoding 增加 v2 Process |
| plugins/processor/base64/decoding/processor_base64_decoding_test.go | base64 decoding 增加 v2 路径单测 |
| plugins/processor/appender/processor_appender.go | appender 抽共享 appendValue + 增加 v2 Process |
| plugins/processor/appender/processor_appender_test.go | appender 增加 v2 路径单测 |
| plugins/processor/anchor/anchor_test.go | anchor 增加 v2 路径单测 |
| plugins/processor/addfields/processor_add_fields.go | add_fields 增加 v2 Process |
| plugins/processor/addfields/processor_add_fields_test.go | add_fields 增加 v2 路径单测 |
| plugins/input/debugfile/input_debug_file.go | v2 input 发出初始化 Tags 的 Group，避免后续 processor panic |
| pluginmanager/processor_tag.go | ProcessorTag v2 增加 Group/Tags nil/sentinel 防护 |
| pluginmanager/processor_tag_test.go | 增加 ProcessorTag v2 nil 防护测试 |
| pkg/pipeline/events.go | 新增 GetStringValue 以统一处理 v2 contents 值类型 |
| Makefile | unittest_plugin 增加拷贝/清理 libGoPluginAdapter.so 到 flusher_sls 目录 |
| docs/cn/SUMMARY.md | 文档目录新增/补齐多篇 processor 文档链接 |
| docs/cn/plugins/processor/extended/processor-strptime.md | 文档补齐 v2 能力与透传说明 |
| docs/cn/plugins/processor/extended/processor-string-replace.md | 文档补齐 v2 能力与透传说明 |
| docs/cn/plugins/processor/extended/processor-split-log-string.md | 新增 split_log_string 文档 |
| docs/cn/plugins/processor/extended/processor-split-log-regex.md | 文档补齐 v2 能力与透传说明 |
| docs/cn/plugins/processor/extended/processor-split-key-value.md | 文档补齐 v2 能力与透传说明 |
| docs/cn/plugins/processor/extended/processor-rename.md | 清理/调整能力说明段落 |
| docs/cn/plugins/processor/extended/processor-regex.md | 文档补齐 v2 能力与透传说明 |
| docs/cn/plugins/processor/extended/processor-rate-limit.md | 文档补齐 v2 能力与透传说明 |
| docs/cn/plugins/processor/extended/processor-pick-key.md | 新增 pick_key 文档 |
| docs/cn/plugins/processor/extended/processor-packjson.md | 文档补齐 v2 能力与透传说明 |
| docs/cn/plugins/processor/extended/processor-otel-trace.md | 文档补齐 v2 Log→Span 行为说明 |
| docs/cn/plugins/processor/extended/processor-otel-metric.md | 文档补齐 v2 Log→Metric 行为说明 |
| docs/cn/plugins/processor/extended/processor-md5.md | 新增 md5 文档 |
| docs/cn/plugins/processor/extended/processor-log-to-sls-metric.md | 文档补齐 v2 能力说明 |
| docs/cn/plugins/processor/extended/processor-json.md | 清理/调整能力说明段落 |
| docs/cn/plugins/processor/extended/processor-grok.md | 文档补齐 v2 能力与透传说明 |
| docs/cn/plugins/processor/extended/processor-gotime.md | 文档补齐 v2 能力与透传说明 |
| docs/cn/plugins/processor/extended/processor-geoip.md | 新增 geoip 文档 |
| docs/cn/plugins/processor/extended/processor-filter-regex.md | 文档补齐 v2 能力与透传说明 |
| docs/cn/plugins/processor/extended/processor-filter-key-regex.md | 新增 filter_key_regex 文档 |
| docs/cn/plugins/processor/extended/processor-fields-with-condition.md | 文档补齐 v2 能力与透传说明 |
| docs/cn/plugins/processor/extended/processor-encrypy.md | 文档补齐 v2 能力与透传说明 |
| docs/cn/plugins/processor/extended/processor-drop.md | 文档补齐 v2 能力与透传说明 |
| docs/cn/plugins/processor/extended/processor-drop-last-key.md | 新增 drop_last_key 文档 |
| docs/cn/plugins/processor/extended/processor-dict-map.md | 文档补齐 v2 能力与透传说明 |
| docs/cn/plugins/processor/extended/processor-desensitize.md | 文档补齐 v2 能力与透传说明 |
| docs/cn/plugins/processor/extended/processor-delimiter.md | 文档补齐 v2 能力与透传说明 |
| docs/cn/plugins/processor/extended/processor-default.md | 清理/调整能力说明段落 |
| docs/cn/plugins/processor/extended/processor-csv.md | 新增 csv 文档 |
| docs/cn/plugins/processor/extended/processor-cloudmeta.md | 文档补齐 v2（写 Group Tags）说明 |
| docs/cn/plugins/processor/extended/processor-base64-encoding.md | 新增 base64 encoding 文档 |
| docs/cn/plugins/processor/extended/processor-base64-decoding.md | 新增 base64 decoding 文档 |
| docs/cn/plugins/processor/extended/processor-appender.md | 文档补齐 v2 能力与透传说明 |
| docs/cn/plugins/processor/extended/processor-add-fields.md | 文档补齐 v2 能力与透传说明 |
| docs/cn/plugins/flusher/extended/flusher-stdout.md | 清理/调整能力说明段落 |
makeKeyV2 (and its v1 mirror makeKey) preallocated values with len(Fields) then appended, so every limit key carried len(Fields) leading empty segments (e.g. "___a_b"), muddying key granularity. Preallocate with len=0 so the key has exactly one segment per field. Also hoist sort.Strings(Fields) into Init: Fields is fixed config, so sorting per call wasted hot-path work and raced on the shared slice across concurrent Process/ProcessLogs calls. Fixes alibaba#2603 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
CloudMeta's v2 Process only initialized Tags when Group != nil &&
Group.Tags == nil. When an input emits Group == nil, Group.Tags == nil,
or Group.Tags set to the immutable NilStringValues sentinel (e.g. a bare
&models.GroupInfo{} or a ByteArray event), GetTags() returns the sentinel
whose Add is a silent no-op, so cloud metadata tags were dropped.
Now create the GroupInfo when absent and replace nil or sentinel Tags
with a fresh mutable Tags before adding cloud metadata. Add regression
tests for all three cases.
Fixes alibaba#2603
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
v2 (Process) and v1 (ProcessLogs) OTEL metric parse errors were reported under ProcessorOTELTraceParserAlarm, misclassifying metric failures as trace alarms and hurting alarm aggregation. Add ProcessorOTELMetricParserAlarm and use it in both paths; also fix the v1 log message that said "trace" instead of "metric". Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
The token-bucket GC goroutine reset its call counter by reassigning the
whole atomic.Int32 value (numCalls = atomic.Int32{}), a plain struct
overwrite that races with the atomic Add/Load done by IsAllowed on the
same field. go test -race flags this as a data race under the GC path.
Reset via the atomic Store(0) instead, and add a concurrent regression
test that crosses the GC threshold from multiple goroutines so the race
is caught under -race.
Fixes alibaba#2603
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
背景
Epic #2595 · Discussion #1928 步骤 B4。依赖 B2（#2599，插件事件能力矩阵）与 B3（#2623，
FlusherV2.Export）均已合入。按维护者指令（Issue #2603 评论
合成一个pr吧），B4 以单个 PR 交付：flusher_slsv2 与在范围内的 processor 子目录迁移合并在本 PR。本 PR 为首批改造，后续 processor 子目录将追加提交到本 PR（不再另开 PR）。改动
flusher_sls v2（SendPb 过渡）
plugins/flusher/sls/flusher_sls.goExport([]*models.PipelineGroupEvents, PipelineContext) error，实现FlusherV2，使flusher_sls可加载于 v2 Runner。converter.PipelineGroupEventsToLogGroup（Log/Metric/Span 均结构化转换、不静默丢弃）将 v2 事件转为LogGroup，再走既有 SendPb cgo 路径 —— 过渡期保持面向 C++ Logtail 队列的单一 wire 格式。sendLogGrouphelper，Flush(v1) 与Export(v2) 共用同一 shardHash /SendPb/SendPbV2逻辑（DRY）。sendPb/sendPbV2函数字段（默认指向真实logtail符号），使发送路径可在无 live Logtail adaptor 时单测。FlusherV1与FlusherV2均实现。Processor 迁移（Process + Metric/Span 透传）
以下 5 个 field-transform processor 新增 v2
Process（作用于models.Log的GetIndices()），并通过 B2 helperpipeline.ProcessLogEventsOnly透传 Metric/Span 事件（不处理但不丢弃，满足 B1/B2 透传契约）：processor_add_fields、processor_drop、processor_drop_last_key、processor_md5、processor_packjson配套
Makefile：unittest_plugin为flusher_sls测试二进制拷贝libGoPluginAdapter.so（flusher_sls依赖 cgo，此前无测试文件故未触发；沿用plugin_main/pluginmanager既有做法），并在 clean 目标清理。docs/cn/plugins/overview.md：更新 B2 事件能力矩阵中已迁移 3 行（add_fields/drop/packjson）为v1&v2+ Metric/Span透传，并补充图例中透传语义。e2e 测试套件调整（回应 review r3613003617 / r3619947609）
本 PR 在 commit
1fa63359中按上一轮维护者意见（r3613003617「Option B + e2e 简化」）调整了 e2e 用例集：新增 1 个真实 v2 流水线断言用例，并移除/合并若干与 B4 迁移面无关或在 v2 模型下无法运行的旧用例。逐条说明移除原因与覆盖去向如下：flusher_kafka_native_0.10.x/1.x.x/2.x.xflusher_kafka_native_3.x.x。功能行为覆盖不变，仅不再对旧 broker 版本做矩阵回归flusher_kafka_native_compressionflusher_kafka_native_3.x.x的Scenario: TestFlusherKafkaNative_Compression_Gzip（gzip 压缩仍覆盖）flusher_kafka_native_headersflusher_kafka_native_3.x.x的Scenario: TestFlusherKafkaNative_Headers（静态 headers 仍覆盖）input_mock_log/input_mock_metricprocessor_v2用例以真实 Log 流水线（metric_debug_file→ 迁移后的 v2Process→flusher_grpc）断言转换后内容，实际执行了本 PR 迁移的 processor 路径；内部指标摄入另由既有input_internal_metrics_prometheus覆盖aggregator_context新增替代用例
test/e2e/test_cases/processor_v2/case.feature（5 个 Scenario）覆盖迁移后的 v2Process链路：TestProcessorV2SplitKeyValueAddFieldsReplaceAppend/TestProcessorV2RegexBase64MD5/TestProcessorV2SplitCharDictMap/TestProcessorV2Csv/TestProcessorV2Grok。不改默认 StructureType。 仅新增 v2 处理路径，v1 行为不变。
Test plan（本地实跑，非「待 CI」）
make unittest_plugin全量通过（MAKE_EXIT=0，无 FAIL / 无 so 加载错误）。新增 / 相关 UT：
本地命令：
验收标准
Process+ Metric/Span 透传flusher_sls可加载于 v2 Runner（SendPb 过渡）make unittest_plugin通过（本地实跑）Closes #2603
🤖 Generated with Claude Code