feat: support Go input to Cpp pipeline #1715
Conversation
core/go_pipeline/LogtailPlugin.cpp
Outdated
| } | ||
| } | ||
| break; | ||
| case sls_logs::PipelineEventGroup::EventType::PipelineEventGroup_EventType_METRIC: |
There was a problem hiding this comment.
这个PR会把Metric也处理吗，还是只是给出定义，后续补齐？
There was a problem hiding this comment.
Metric和Trace只要能转成PipelineEventGroup，后续都可以用现有流水线处理了
There was a problem hiding this comment.
cpp这边的metric event暂时只支持untyped single value，go的multivalue以及typed value都无法转换
core/go_pipeline/LogtailPlugin.cpp
Outdated
| } | ||
| } | ||
| break; | ||
| case sls_logs::PipelineEventGroup::EventType::PipelineEventGroup_EventType_METRIC: |
There was a problem hiding this comment.
Metric和Trace只要能转成PipelineEventGroup，后续都可以用现有流水线处理了
core/go_pipeline/LogtailPlugin.cpp
Outdated
| } | ||
| } | ||
| break; | ||
| case sls_logs::PipelineEventGroup::EventType::PipelineEventGroup_EventType_METRIC: |
Abingcbc
changed the title
Abingcbc
changed the title
| @@ -0,0 +1,85 @@ | |||
| syntax = "proto2"; | |||
There was a problem hiding this comment.
这块有测试pb 序列化的开销和反序列化开销吗？另外有没有对比一些其他协议的性能测试，比如flatbuffer，arrow。
There was a problem hiding this comment.
这个之前有测试过了，pb序列化和大小都是比较有优势的，但是反序列稍微慢些
| state: input, | ||
| interval: wrapper.Interval * time.Millisecond, | ||
| state: &wrapper, | ||
| interval: wrapper.Interval, |
There was a problem hiding this comment.
这里wrapper.Interval本来就是乘过time.Millisecond的，这里再乘一个Millisecond运行过程中不对，应该是之前写错了
There was a problem hiding this comment.
这个确定是写错了吗，我们应该有业务用MetricInputV2 的，好像没反馈 interval 不对呀？
There was a problem hiding this comment.
这里wrapper.Interval是由metricWrapperV2.Init初始化的，这个Init函数输入参数有一个int类型的inputInterval，在函数中对inputInterval乘time.Millisecond赋给了wrapper.Interval
There was a problem hiding this comment.
好的，这个好像是近期引入的 bug，我们内部会延后一点时间和社区 main 同步，所以上一个版本还没问题
| @@ -59,6 +59,11 @@ struct containerMeta{ | |||
| char** envsVal; | |||
| }; | |||
|
|
|||
| struct loadGoPipelineResp { | |||
There was a problem hiding this comment.
func GetPipelineMetrics() *C.PluginMetrics也要调整吗？改成 pb协议到core。
| @@ -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; | |||
There was a problem hiding this comment.
这里是因为当流水线配置为go input -> cpp processor -> go processor/aggregator/flusher时，在pipeline::Init()中将go的配置全部添加到了mGoPipelineWithInput，此时mGoPipelineWithoutInput为空，因此这里会返回false，导致cpp无法向go发送数据
已实现
待实现