采集配置Pipeline演化路线

[Takuka0311]

整体演进顺序为 A → B → C，支线 D 与 B/C 交叉推进。

总览：三条主线 + 一条支线

线	终态目标
A. C++ Input → Go	任意 Native Input 在含 Go Processor/Flusher 时均可经 PB 桥接进 Go
B. 插件全事件类型 + v2 就绪	Processor/Flusher 实现或透传 Log/Metric/Span；显式 StructureType: v2 可安全运行
C. 废弃 LogGroup	全链路 PipelineEventGroup；删 v1 Runner；Go→C++ 用 PB 替代 SendPb(LogGroup)
D. 统一 C++/Go 自监控（支线）	Go 指标/告警经 PipelineEventGroup PB 回 C++；与 C++ 同一套 category/指标名/采集规则

分阶段依赖总览

flowchart TB

  subgraph P1["阶段 1 · 短期"]
    A12["A1–A2\n扩 Input 白名单"]
    B12["B1–B2\n透传契约 + 插件矩阵盘点"]
  end

  subgraph P2["阶段 2 · 中期"]
    B34["B3–B4\nFlusher Export / Processor Process 迁移\nflusher_sls v2 包装或双接口"]
    D12["D1–D3\n自监控 PB 推送 + 指标目录对齐"]
    A3["A3\n移除 Input 白名单"]
    A4["A4\n多 Input E2E"]
    B56["B5–B6\n插件 lint + 混合 Log/Metric/Span E2E"]
  end

  subgraph P3["阶段 3 · 长期"]
    C123["C1–C3\nSendPipelineEventGroup\nflusher_sls Export + C++ 接 PB"]
    D45["D4–D6\n废弃 GetGoMetrics/Alarms CGO\nWindows 全量采集"]
    C4["C4\n删 pluginv1Runner / Legacy 转换"]
    C5["C5\n默认 StructureType: v2"]
    C6["C6\n清理 LogGroup 热路径"]
  end

  subgraph Target["终态"]
    T["全部 Native Input → Go\n仅 v2 Runner + 全插件多事件\n双向 PipelineEventGroup PB"]
  end

  A12 --> A3
  A12 --> A4
  B12 --> B34
  B12 --> D12
  B34 --> B56
  B34 --> C123
  D12 --> D45
  B34 --> D45
  D45 --> C4
  A3 --> T
  A4 --> T
  B56 --> C123
  C123 --> C4
  B34 --> C4
  C4 --> C5
  C5 --> C6
  C6 --> T
  D45 --> T

Loading

阶段	主线 / 支线步骤	依赖	备注
1	A1–A2 ∥ B1–B2	-	不改默认 StructureType
2	B3–B6，A3–A4，D1–D3	阶段 1	D：PB 推送替代 CGO 拉取；指标目录与 Go 插件 category 对齐
3	C1–C6，D4–D6	B3–B4；D3	D4 废弃 CGO 与 C4 删 v1 可合并规划；C5 默认 v2 仍最后

---

主线 A：所有 C++ Input 均可进入 Go Pipeline

现状（代码锚点）

• 配置闸门：core/config/CollectionConfig.cpp 中 hasFileInput / hasSecurityInput / hasSelfMonitorInput 决定是否允许「Native Input + Go Processor/Flusher」。未命中白名单则报错 extended * plugins coexist with native input plugins other than ...（约 L365、L495）。
• 仍走 Native Send、不进 Go 的配置：仅含 Native Flusher、且 IsFlushingThroughGoPipeline()==false 的 Pipeline（如 input_internal_metrics → flusher_file）。

目标

任意注册在 C++ PluginRegistry 的 Continuous Native Input（input_forward、input_prometheus、input_container_stdio 等），只要 Pipeline 含 Go Processor 或 Go Flusher，处理完成后均可进入 Go，不再维护 Input 类型白名单。

分阶段实施

阶段	工作项	涉及代码 / 验收
A1	扩展白名单或改为「Native Input 默认允许 + 黑名单」	CollectionConfig.cpp；UT：CollectionConfigUnittest 增例
A2	按 Input 类型补齐 C++ 事件产出（部分 Input 仅 Log，需确认 Metric/Span 语义）	各 core/plugin/input/*；ProtocolConversion 单测
A3	删除 hasFileInput 等分支及 input_file 仅允许单 Input 等过渡约束（代码内 TODO）	CollectionConfig.cpp、CollectionPipeline.cpp（DefaultLogQueueSize 特例）
A4	E2E：至少覆盖 input_forward / input_prometheus + flusher_prometheus（v2）	test/e2e/test_cases/*

flowchart LR
  subgraph CppIn["C++ Native Input（目标：全部）"]
    IF["input_file"]
    IP["input_prometheus"]
    II["input_internal_metrics"]
    IF2["input_forward / ..."]
  end

  subgraph Gate["配置层（逐步拆除）"]
    CC["CollectionConfig.Parse\n白名单 → 默认允许"]
  end

  subgraph Bridge["数据层（已具备）"]
    PR["ProcessorRunner\nTransferPipelineEventGroupToPB"]
    PEG["ProcessPipelineEventGroup"]
  end

  CppIn --> CC --> PR --> PEG

Loading

风险与约束： 多 Input 同 Pipeline、Native+Go 多 Flusher 共存规则（nativeFlusherCnt、flusher_sls）需在 A3 一并梳理，避免仅拆白名单导致配置组合爆炸。

---

主线 B：Processor / Flusher 支持全部 PipelineGroupEvents

现状（接口与实现差距）

• 配置：fetchPluginVersion 在未写 global.StructureType 时默认 v1 → pluginv1Runner（logstore_config.go）。
• 契约（pkg/pipeline/processor.go / flusher.go）：
  • v1：ProcessLogs / Flush(LogGroup) — 仅 Log。
  • v2：Process(PipelineGroupEvents) / Export([]*PipelineGroupEvents) — Log / Metric / Span。
• 实现差距：
  • Processor：plugins/processor/* 大量仅 ProcessLogs（约 50+ 处）；v2 需 Process，且多数假定 models.Log。
  • Flusher：flusher_prometheus 等仅 v2 Export；flusher_http 双接口；flusher_sls 仅 FlusherV1.Flush，pluginv2Runner.addFlusher 无法直接加载。
  • Aggregator：v2 Record 已对齐；v1 路径仍聚合为 LogGroup 供 sls Flush。

目标

任一 v2 Processor/Flusher 对 PipelineGroupEvents：支持（按语义处理子集）、透传（不静默丢弃）、声明 SupportedEventKinds（或等价机制）。

分阶段实施

阶段	工作项	涉及代码	依赖
B1	事件透传约定与 helper（PassThroughEvents）	pkg/pipeline / pkg/helper	-
B2	插件矩阵：Processor/Flusher × {Log, Metric, Span}	扫描 Process / Export / ProcessLogs	B1
B3	Extended Flusher 补齐 FlusherV2.Export；Log-only 对 Metric/Span 透传	plugins/flusher/*	B2
B4	Processor 迁移：ProcessLogs → Process（Log），Metric/Span 透传；flusher_sls v2 包装或双接口（过渡期仍可用 SendPb）	plugins/processor/*、plugins/flusher/sls	B2–B3
B5	Lint/Review：禁止 Process 仅按 Events[0] 假定 Log	CI 规则	B4
B6	E2E：混合 Log + Metric + Span → flusher_prometheus / flusher_http	test/e2e/*	B3–B4

flowchart TB
  subgraph Input["PipelineGroupEvents"]
    L["Log"]
    M["Metric"]
    S["Span"]
  end

  subgraph Proc["Processor 链（目标）"]
    P1["Processor A\n处理 Log"]
    P2["Processor B\n处理 Metric"]
    P3["Processor C\n透传 Span"]
  end

  subgraph Flush["Flusher（目标）"]
    FP["flusher_prometheus\nMetric"]
    FH["flusher_http\nLog / 告警"]
  end

  L --> P1 --> FH
  M --> P2 --> FP
  S --> P3 --> FH

Loading

插件类型	典型现状	目标行为
文本解析类 Processor（regex、json、grok）	仅 ProcessLogs	v2 Process 只改 Log；Metric/Span 透传
指标类 Processor（processor_log_to_sls_metric）	部分 v2	Log→Metric；其它透传
观测导出 Flusher（prometheus、otlp）	Metric/Trace 为主	明确处理或拒绝 Log
存储 Flusher（kafka、http）	多为 Flush	B3：Export + Log
flusher_sls	仅 Flush → SendPb	B4：v2 包装；主线 C 再换 SendPipelineEventGroup

---

主线 C：彻底废弃 LogGroup，统一 Pipeline 事件模型

依赖主线 B：C4（删 v1 Runner）、C5（默认 v2）须在 B3–B4 插件迁移达到约定覆盖率之后，避免未写 StructureType 的存量配置误走 pluginv2Runner 而 pluginUnImplementError 或丢 Metric/Span。

现状：两套模型并存

层级	v1 / LogGroup 路径	v2 / PipelineGroupEvents 路径
配置	默认 v1	显式 StructureType: v2
C++ → Go	均已 ProcessPipelineEventGroup	ReceivePipelineEventGroup
Go 入口	TransferPBToLogGroupForV1（Metric/Span 丢弃）	TransferPBToPipelineGroupEvents
Go → C++	flusher_sls：Flush → SendPb	prometheus/http 等 Export，不回 C++
C++ 原生刷盘	pipeline->Send() → FlusherSLS::Send	不经 Go

flusher_sls 与 Export 型 Flusher

场景	走 Go Pipeline	回写 C++
仅 Native + 仅 flusher_sls	否	否（全程 C++ FlusherSLS::Send）
混合 Pipeline + flusher_sls	是（GenerateGoPlugin 注入 Go sls）	是（LogGroup → SendPb → C++）
混合 + flusher_prometheus（v2）	是	否

C++ ProcessPipelineEventGroup → Go Aggregator/Flusher
  → SlsFlusher.Flush(LogGroup) → SendPb/V2 → C++ FlusherSLS::Send → SLS

目标

1. Go→C++ 改为 SendPipelineEventGroup（PB），对称于 ProcessPipelineEventGroup。
2. 删除 pluginv1Runner、TransferPBToLogGroupForV1、ReceiveLogGroup 等。
3. C5：新配置默认 StructureType: v2（在 B4 完成后）。
4. 清理 sls_logs.LogGroup 热路径。

分阶段实施

阶段	工作项	涉及代码	依赖
C1	CGO SendPipelineEventGroup；与 SendPb 并存	plugin_export.go、LogtailPlugin.cpp	B4（sls v2 方案）
C2	flusher_sls 实现 FlusherV2.Export → SendPipelineEventGroup	plugins/flusher/sls/flusher_sls.go	B4、C1
C3	C++ FlusherSLS::Send 接 PipelineEventGroup PB	core/plugin/flusher/sls/*	C1–C2
C4	删 pluginv1Runner、Legacy PB 转换	pluginmanager/*、pipeline_event_helper.go	B4 + C2
C5	默认 StructureType: v2；文档/示例切换	logstore_config.go	B2 矩阵 + B4 + C4
C6	清理 LogGroup 热路径；协议保留兼容层	pkg/protocol/sls_logs.*	C4–C5

flowchart TB
  subgraph Today["Go → C++ 今日"]
    F1["flusher_sls.Flush"]
    LG["LogGroup.Marshal"]
    SP["SendPb / SendPbV2"]
    FS["FlusherSLS::Send"]
    F1 --> LG --> SP --> FS
  end

  subgraph Future["Go → C++ 目标（C1–C3）"]
    F2["flusher_sls.Export"]
    PB2["PipelineEventGroup PB"]
    SP2["SendPipelineEventGroup"]
    FS2["FlusherSLS 接 PB"]
    F2 --> PB2 --> SP2 --> FS2
  end

  Today -.-> Future

Loading

与 #2568 相关配置在 C5 完成前 须继续 显式 StructureType: v2。

---

支线 D：统一 C++ 与 Go 自监控（指标 + 告警）

问题研判（与代码对照）

你提出的两点 成立，且代码里已有明确佐证；另有几项关联问题值得一并纳入支线范围。

（1）CGO 自定义结构体不稳定 —— 成立，Windows 上更严重

维度	现状	结论
指标拉取	C++ 定时调用 GetGoMetrics → Go malloc 构造 PluginMetrics / InnerPluginMetric / InnerKeyValue → C++ 侧逐个 free（LogtailPlugin.cpp L476–505）	双向手工内存管理，无 PB 契约，易 double-free/泄漏
告警拉取	GetGoAlarms → InnerGoAlarms / InnerGoAlarm 同理（L508–575）→ 写入 AlarmManager::SendExternalAlarm	与导出自监控用的 PipelineEventGroup 不是同一路径
Windows	ReadMetrics::UpdateMetrics() 在 #ifndef _MSC_VER 下才调用 GetGoMetrics；注释写明 「windows 上的 cgo 内存有问题…可能会 crash」（MetricManager.cpp L168–184）	Windows 上 Go 插件指标根本不会进入自监控（仅 C++ 指标 + Agent 进程级部分）

业务数据 C++→Go 的 ProcessPipelineEventGroup（PipelineEventGroup PB）已打通；自监控 反向 仍走上述 CGO 结构体，并未统一。

（2）Go 与 C++ 监控能力未对齐 —— 成立，有多层表现

维度	C++	Go	差距
采集模型	WriteMetrics 快照 → MetricsRecord 链表 → SelfMonitorMetricEvent	轮询 GetGoMetrics → map 内嵌 JSON 字符串（labels/counters/gauges）再解析	模型不同；Go 侧丢字段会落 METRIC_CATEGORY_UNKNOWN
category 划分	agent / runner / pipeline / plugin / plugin_source / component	Go 插件统一 metric_category=plugin（GetPluginCommonLabels）	无 plugin_source；与 C++ Input 级指标（文件偏移、采集延迟等）无法同维度对比
指标类型	Counter / TimeCounter / IntGauge / DoubleGauge 等	ExportMetricRecords 仅导出 Counter + Gauge（metrics_record.go L68–73）	Average/Latency 等在 Go 侧未进入自监控
插件覆盖	Native 插件 + Runner 均有 MetricsRecord	依赖各 Go 插件 RegisterMetricRecord；未注册则无数据	部分 Go 插件/扩展 统计不全
告警	AlarmManager::FlushAllRegionAlarm → PipelineEventGroup → 自监控 Pipeline	Go 告警经 CGO 结构体 同步到 C++ AlarmManager，再与 C++ 告警合并 flush	入队形态不一致；Go 未直接产出 PB Log 事件

支线目标

1. Go → C++ 自监控数据统一为 PipelineEventGroup PB（Metric 用多值 UntypedMultiDoubleValues，告警用 LogEvent + metadata），废弃 GetGoMetrics / GetGoAlarms 及 InnerPlugin* / InnerGoAlarm 结构体。
2. Go 与 C++ 共用 指标目录（category、label 键、指标名常量、采集间隔规则），plugin / plugin_source / runner 语义一致，Windows/Linux 行为一致。
3. 自监控 采集 与 导出 对称：导出已支持 input_internal_* → Go Flusher；采集改为 Go 主动 Push（或 C++ 拉 PB 字节流），与 ProcessPipelineEventGroup 对称。

分阶段实施

阶段	工作项	涉及代码	依赖
D1	编写《自监控指标/告警字段对照表》（C++ MetricConstants ↔ Go pkg/selfmonitor）；明确 plugin vs plugin_source 映射规则	文档 + 单测契约	-
D2	定义 Go→C++ 自监控 PB 契约：Metric 事件（category + labels + 多值 map）、Log 告警（对齐 AlarmManager 字段）；复用 TransferPipelineEventGroupToPB / Go TransferMetricEventToPB	protobuf_public、pkg/helper	-
D3	Go 侧 PushSelfMonitorToCore(pb []byte)（或定时调用 C++ 注册的 ReceiveSelfMonitorEventGroup）；C++ 解析 PB 后直接 SelfMonitorServer::PushSelfMonitorMetricEvents / 告警入队	plugin_export.go、SelfMonitorServer.cpp、MetricManager.cpp	D2
D4	Go：plugin_source 级 MetricsRecord（Input 插件）；补齐 Counter/Gauge 以外类型的导出或等价聚合	plugin_wrapper*.go、pkg/selfmonitor	D1、B4
D5	删除 GetGoMetrics/GetGoAlarms CGO；移除 MetricManager.cpp 的 _MSC_VER 屏蔽；Windows 回归	LogtailPlugin.*、plugin_export.go	D3、D4
D6	E2E/UT：Go 插件造 counter → 自监控 Pipeline → flusher_file/Prometheus 可查到；Windows CI 覆盖	test/e2e、core/unittest/monitor	D5

flowchart TB
  subgraph Today["采集 · 今日（拉模式 + CGO 结构体）"]
    TMR["SelfMonitorServer::SendMetrics"]
    RM["ReadMetrics::UpdateMetrics"]
    GM["GetGoMetrics / GetGoAlarms\nInnerPlugin* / InnerGoAlarm"]
    GO1["Go: ExportMetricRecords → map+JSON"]
    PARSE["C++: SelfMonitorMetricEvent(map)\nAlarmManager::SendExternalAlarm"]
    TMR --> RM --> GM --> GO1 --> PARSE
    WIN["Windows: GetGoMetrics 整段禁用"]
    GM -.-> WIN
  end

  subgraph Target["采集 · 目标（推模式 + PB）"]
    TMR2["SelfMonitorServer::SendMetrics"]
    PUSH["Go PushSelfMonitor PB\n或 C++ Pull PB bytes"]
    PB["PipelineEventGroup\nMetrics + Logs"]
    MERGE["C++ 直接 Merge 到\nSelfMonitorMetricEventMap / 告警队列"]
    TMR2 --> MERGE
    PUSH --> PB --> MERGE
  end

  subgraph Export["导出 · 已打通"]
    EX["input_internal_* → PB → Go Flusher"]
  end

  Today -.->|"D2–D5"| Target
  MERGE --> Export

Loading

flowchart LR
  subgraph Align["D1/D4 · 能力对齐"]
    CAT["category 统一\nplugin / plugin_source / runner"]
    NAME["指标名常量统一"]
    TYPE["Go Export 覆盖 C++ 指标类型"]
  end

  subgraph Transport["D2/D3/D5 · 传输统一"]
    PB2["PipelineEventGroup PB"]
    NOP["移除 CGO 结构体"]
  end

  Align --> Transport

Loading

与主线的关系	说明
A	无直接依赖；自监控 Pipeline 仍可用 input_internal_* 导出
B	D4 与 B4 共享 Go 插件 MetricsRecord / plugin_source 改造
C	D3 可复用 SendPipelineEventGroup 基础设施；D5 与 C4 同期删除 Legacy 拉取/转换

---

里程碑 / Progress

步骤	Issue	PR	状态
A1	#2596	#2627	完成
A2	#2597	#2621	完成
A3	#2600	#2642	阻塞
A4	#2601	#—	阻塞
B1	#2598	#2626	完成
B2	#2599	#2619	完成
B3	#2602	#2623	完成
B4	#2603	#—	待开始
B5	#2604	#—	阻塞
B6	#2605	#—	阻塞
C1	#2609	#—	阻塞
C2	#2610	#—	阻塞
C3	#2611	#—	阻塞
C4	#2612	#—	阻塞
C5	#2613	#—	阻塞
C6	#2614	#—	阻塞
D1	#2606	#2624	完成
D2	#2607	#2625	完成
D3	#2608	#2641	完成
D4	#2615	#—	阻塞
D5	#2616	#—	阻塞
D6	#2617	#—	阻塞

快照更新于 2026-07-17T09:08:29.123Z（来源：Epic Console board）