feat: support Go input to Cpp pipeline #1715

Assassin718 · 2024-08-26T02:18:58Z

已完成

增加加载Go流水线的返回值信息input mode（Go input plugin类型），根据input mode创建对应的process queue（Bounded/Circular Process Queue）
支持v1流水线LogEvent的转发：
- 支持对多个LogEvent缓存后打包成PipelineEventGroup，异步发送给Cpp processor
- 支持对缓存参数的设置：在流水线配置文件global中设置PushNativeMaxCachedSize、PushNativeTimeoutMs，分别表示最大一次性发送日志数、超时发送时间
- input mode为PUSH情况下可根据process queue水位线阻塞input，降低输入速率
支持v2流水线PipelineEventGroup转发：
- 对每个PipelineEventGroup进行同步转发
- 暂不支持缓存多个PipelineEventGroup、调节输入速率

待完成

单元测试
e2e测试

shunjiazhu · 2024-08-27T08:13:00Z

core/go_pipeline/LogtailPlugin.cpp

+            }
+        }
+        break;
+    case sls_logs::PipelineEventGroup::EventType::PipelineEventGroup_EventType_METRIC:


这个PR会把Metric也处理吗，还是只是给出定义，后续补齐？

Metric和Trace只要能转成PipelineEventGroup，后续都可以用现有流水线处理了

转换也在这个PR里补齐吧

cpp这边的metric event暂时只支持untyped single value，go的multivalue以及typed value都无法转换

core/config/PipelineConfig.cpp

Abingcbc · 2024-08-27T08:21:16Z

core/go_pipeline/LogtailPlugin.cpp

+            }
+        }
+        break;
+    case sls_logs::PipelineEventGroup::EventType::PipelineEventGroup_EventType_METRIC:


Metric和Trace只要能转成PipelineEventGroup，后续都可以用现有流水线处理了

core/go_pipeline/LogtailPlugin.cpp

Abingcbc · 2024-08-27T08:24:56Z

core/go_pipeline/LogtailPlugin.cpp

+            }
+        }
+        break;
+    case sls_logs::PipelineEventGroup::EventType::PipelineEventGroup_EventType_METRIC:


转换也在这个PR里补齐吧

pluginmanager/plugin_wrapper_metric_v1.go

pluginmanager/plugin_wrapper_service_v1.go

shunjiazhu · 2024-09-02T03:16:04Z

pkg/protocol/proto/pipeline_event.proto

@@ -0,0 +1,85 @@
+syntax = "proto2";


这块有测试pb 序列化的开销和反序列化开销吗？另外有没有对比一些其他协议的性能测试，比如flatbuffer，arrow。

这个之前有测试过了，pb序列化和大小都是比较有优势的，但是反序列稍微慢些

pkg/pipeline/input.go

Abingcbc · 2024-09-02T08:19:42Z

pluginmanager/plugin_runner_v2.go

-		state:         input,
-		interval:      wrapper.Interval * time.Millisecond,
+		state:         &wrapper,
+		interval:      wrapper.Interval,


这里为什么少了time.Millisecond

这里wrapper.Interval本来就是乘过time.Millisecond的，这里再乘一个Millisecond运行过程中不对，应该是之前写错了

这个确定是写错了吗，我们应该有业务用MetricInputV2 的，好像没反馈 interval 不对呀？

这里wrapper.Interval是由metricWrapperV2.Init初始化的，这个Init函数输入参数有一个int类型的inputInterval，在函数中对inputInterval乘time.Millisecond赋给了wrapper.Interval

好的，这个好像是近期引入的 bug，我们内部会延后一点时间和社区 main 同步，所以上一个版本还没问题

pluginmanager/plugin_runner_v2.go

pluginmanager/plugin_wrapper_metric_v1.go

pluginmanager/plugin_wrapper_service_v1.go

shunjiazhu · 2024-09-11T06:31:23Z

plugin_main/plugin_export.go

@@ -59,6 +59,11 @@ struct containerMeta{
 	char** envsVal;
 };

+struct loadGoPipelineResp {


func GetPipelineMetrics() *C.PluginMetrics也要调整吗？改成 pb协议到core。

Assassin718 · 2024-09-19T09:00:04Z

core/pipeline/Pipeline.h

@@ -51,11 +52,11 @@ class Pipeline {
    PipelineContext& GetContext() const { return mContext; }
    const Json::Value& GetConfig() const { return *mConfig; }
    const std::vector<std::unique_ptr<FlusherInstance>>& GetFlushers() const { return mFlushers; }
-    bool IsFlushingThroughGoPipeline() const { return !mGoPipelineWithoutInput.isNull(); }
+    bool IsFlushingThroughGoPipeline() const;


这里是因为当流水线配置为go input -> cpp processor -> go processor/aggregator/flusher时，在pipeline::Init()中将go的配置全部添加到了mGoPipelineWithInput，此时mGoPipelineWithoutInput为空，因此这里会返回false，导致cpp无法向go发送数据

core/go_pipeline/LogtailPlugin.cpp

core/pipeline/Pipeline.cpp

Abingcbc · 2024-10-29T08:35:54Z

pluginmanager/plugin_runner_v2.go

@@ -71,6 +73,7 @@ func (p *pluginv2Runner) Init(inputQueueSize int, flushQueueSize int) error {
 	p.ProcessPipeContext = helper.NewGroupedPipelineConext()
 	p.AggregatePipeContext = helper.NewObservePipelineConext(flushQueueSize)
 	p.FlushPipeContext = helper.NewNoopPipelineConext()
+	p.NativeInputPipeContext = nil


空指针很危险，这里还是初始化一个默认的

NativeProcessorPipeContext

这里InputMode只有后面具体分析每个Input的类型时才能确定下来，Init在这个过程之前，但是这个Context又必须传入InputMode，不太好改

pluginmanager/plugin_wrapper_metric_v1.go

Abingcbc · 2024-10-29T08:42:11Z

pluginmanager/plugin_wrapper_metric_v1.go

+	for {
+		if isValidToPushNativeProcessQueue {
+			select {
+			case <-wrapper.timer.C:


不用timer，直接 <-time.After

benchmark时直接timer.After会有比较大的定时器创建开销

Abingcbc · 2024-10-29T08:44:30Z

pluginmanager/plugin_wrapper_metric_v1.go

+	defer close(wrapper.ShutdownCachedChan)
+
+	for {
+		if isValidToPushNativeProcessQueue {


这块逻辑有点乱
pushNativeProcessQueue是否可以是阻塞的，无限重试的，这样外面不需要复杂的处理了

这里因为是直接调用cpp的PushQueue接口，cpp的ProcessQueueManager::PushQueue不是阻塞的

这里主要逻辑是：检测到ProcessQueue水位高时会停止监听LogCachedChan，以阻塞Input；直到检测到ProcessQueue水位低后才重新监听LogCachedChan，使Input运行

isValidToPushNativeProcessQueue这个其实是表示的上次push是否成功？

Abingcbc · 2024-10-29T08:46:03Z

pluginmanager/plugin_wrapper_metric_v1.go

+}
+
+func (wrapper *MetricWrapperV1) pushNativeProcessQueue(retryCnt int) bool {
+	if len(wrapper.pbBuffer) == 0 && len(wrapper.eventCached) == 0 {


这里为什么有两个buffer？
我理解应该只需要有event的buffer？每次发送前转成pb之后，就直接发送了

pbBuffer是用来放pb反序列化后的结果，这里是尽量希望内存复用

并且如果向cpp发送失败，需要保存上次已经反序列化后的结果，否则会重新反序列化

应该只有pb的buffer就够了？从channel里弹出event时，就序列化放入到pbbuffer中，似乎不需要再对event、tag和context再做一层buffer

因为序列化的对象是PipelineEventGroup，如果要将每次event序列化后的结果正确拼接到PipelineEventGroup序列化结果之后实现比较复杂

如果不保留event、tag、context的buffer的话，在出现tag/context不一致再加上发送失败时需要检查更多的边界条件，这里暂时是比较简单的实现

这里优化的话能减少一小部分内存占用，之后可以尝试

shunjiazhu reviewed Aug 27, 2024

View reviewed changes

Assassin718 force-pushed the go2cpp branch from b44d250 to 96c3056 Compare August 27, 2024 08:23

Abingcbc requested changes Aug 27, 2024

View reviewed changes

Assassin718 force-pushed the go2cpp branch from 9d41cfe to f7dd3bd Compare September 1, 2024 14:13

Abingcbc changed the title ~~[WIP] feat: add Go input to Cpp processor~~ feat: add Go input to Cpp processor Sep 2, 2024

Abingcbc changed the title ~~feat: add Go input to Cpp processor~~ feat: support Go input to Cpp pipeline Sep 2, 2024

shunjiazhu reviewed Sep 2, 2024

View reviewed changes

Abingcbc requested changes Sep 2, 2024

View reviewed changes

shunjiazhu reviewed Sep 11, 2024

View reviewed changes

Assassin718 requested a review from Abingcbc September 12, 2024 01:41

Assassin718 force-pushed the go2cpp branch from f070af9 to df280e8 Compare September 18, 2024 08:49

Assassin718 commented Sep 19, 2024

View reviewed changes

restore PipelineConfig to unsupport go2pp

da07ff6

Assassin718 force-pushed the go2cpp branch from 072e0d6 to da07ff6 Compare October 28, 2024 04:38

resolve PushQueue bug

6827847

Abingcbc requested changes Oct 29, 2024

View reviewed changes

Assassin718 force-pushed the go2cpp branch from deafe0a to 6827847 Compare October 29, 2024 10:03

polish

6b28bdb

Abingcbc changed the base branch from main to ospp_dev October 30, 2024 03:40

support go2cpp config check

12419f1

Abingcbc merged commit 9fefc58 into alibaba:ospp_dev Oct 31, 2024
15 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: support Go input to Cpp pipeline #1715

feat: support Go input to Cpp pipeline #1715

Assassin718 commented Aug 26, 2024 •

edited

Loading

shunjiazhu Aug 27, 2024

Abingcbc Aug 27, 2024

Abingcbc Aug 27, 2024

Assassin718 Sep 2, 2024

Abingcbc Aug 27, 2024

Abingcbc Aug 27, 2024

shunjiazhu Sep 2, 2024

Abingcbc Sep 2, 2024

Abingcbc Sep 2, 2024

Assassin718 Sep 2, 2024

shunjiazhu Sep 3, 2024

Assassin718 Sep 3, 2024

shunjiazhu Sep 4, 2024

shunjiazhu Sep 11, 2024

Assassin718 Sep 19, 2024

Abingcbc Oct 29, 2024

Abingcbc Oct 29, 2024

Assassin718 Oct 29, 2024 •

edited

Loading

Abingcbc Oct 29, 2024

Assassin718 Oct 29, 2024 •

edited

Loading

Abingcbc Oct 29, 2024

Assassin718 Oct 29, 2024 •

edited

Loading

Assassin718 Oct 29, 2024

Abingcbc Oct 29, 2024

Assassin718 Oct 29, 2024

Abingcbc Oct 29, 2024

Assassin718 Oct 29, 2024

Assassin718 Oct 29, 2024

Abingcbc Oct 29, 2024

Assassin718 Oct 29, 2024 •

edited

Loading

feat: support Go input to Cpp pipeline #1715

feat: support Go input to Cpp pipeline #1715

Conversation

Assassin718 commented Aug 26, 2024 • edited Loading

已完成

待完成

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Assassin718 Oct 29, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Assassin718 Oct 29, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Assassin718 Oct 29, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Assassin718 Oct 29, 2024 • edited Loading

Choose a reason for hiding this comment

Assassin718 commented Aug 26, 2024 •

edited

Loading

Assassin718 Oct 29, 2024 •

edited

Loading

Assassin718 Oct 29, 2024 •

edited

Loading

Assassin718 Oct 29, 2024 •

edited

Loading

Assassin718 Oct 29, 2024 •

edited

Loading