add rocketmq flusher #850

wlazjr · 2023-05-16T05:02:50Z

add rocketmq flusher

CLAassistant · 2023-05-16T05:02:57Z

Thank you for your submission! We really appreciate it. Like many open source projects, we ask that you sign our Contributor License Agreement before we can accept your contribution.

伴农 seems not to be a GitHub user. You need a GitHub account to be able to sign the CLA. If you have already a GitHub account, please add the email address used for this commit to your account.
_{You have signed the CLA already but the status is still pending? Let us recheck it.}

messixukejia · 2023-05-16T05:27:32Z

很高效👍
贵公司如果已经在使用ilogtail，欢迎在这里登记下，多谢。#693

messixukejia · 2023-05-16T05:39:15Z

CI的问题需要处理下

shalousun · 2023-05-16T06:16:34Z

plugins/flusher/rocketmq/flusher_rocketmq.go

+}
+
+type FlusherRocketmq struct {
+	context     pipeline.Context


FlusherRocketmq这个结构体属性建议把面向用户配置和内部逻辑使用的分开，不要混合在一起，大写和小写命名的可以参考其它flusher 分割下

shalousun · 2023-05-16T06:24:10Z

plugins/flusher/rocketmq/flusher_rocketmq.go

+		}
+		for index, log := range logs.([][]byte) {
+			valueMap := values[index]
+			topic, err := fmtstr.FormatTopic(valueMap, r.Topic)


这里参考当前flusher pulsar的逻辑优化下，大部分场景不会触发动态topic，FormatTopic有一定的性能损耗，如果静态topic就不要调用FormatTopic

shalousun · 2023-05-16T06:34:23Z

https://github.com/alibaba/ilogtail/blob/main/licenses/LICENSE_OF_ILOGTAIL_DEPENDENCIES.md
https://github.com/alibaba/ilogtail/blob/main/licenses/LICENSE_OF_TESTENGINE_DEPENDENCIES.md
上面两个文件里添加新增依赖库的license，可以看ci static check里的报错，

https://github.com/alibaba/ilogtail/blob/main/CHANGELOG.md Unreleased里添加下rocketmq flusher的新增日志

https://github.com/alibaba/ilogtail/tree/main/docs/cn/data-pipeline/flusher 下新增一个rocketmq flusher的文档
https://github.com/alibaba/ilogtail/blob/main/docs/cn/data-pipeline/overview.md 这里面添加rocketmq flusher

shalousun · 2023-05-16T06:42:40Z

plugins/flusher/rocketmq/flusher_rocketmq.go

+	"context"
+	"errors"
+	"fmt"
+	"github.com/alibaba/ilogtail/pkg/fmtstr"


import这块按照golang的规范格式化下

all right, thanks bro, first time writing golang, I will fix above issues later.

henryzhx8

需要加一下e2e测试，详情见doc

henryzhx8 · 2023-05-18T09:12:32Z

docs/cn/data-pipeline/flusher/flusher-rocketmq.md

+|----------------------------|--------|------|-------------------------------------------------------------|
+| NameServers                | String | 是    | name server address                                         |
+| Topic                      | String | 是    | rocketmq Topic,支持动态topic, 例如: `test_%{content.appname}`     |
+| Sync                       | bool   | 否    | 默认为异常发送                                                     |


异常发送？异步？

henryzhx8 · 2023-05-18T09:12:47Z

docs/cn/data-pipeline/flusher/flusher-rocketmq.md

+
+| 参数                         | 类型     | 是否必选 | 说明                                                          |
+|----------------------------|--------|------|-------------------------------------------------------------|
+| NameServers                | String | 是    | name server address                                         |


名字和其它插件统一一下吧

你是说要改成Brokers么，kafka配置里写的是brokers的地址，rocketmq是需要填namesever地址，然后通过nameserver找到broker

这个保留和中间件本身的术语应该OK

henryzhx8 · 2023-05-18T09:16:15Z

plugins/flusher/rocketmq/flusher_rocketmq.go

+	r.context = context
+	if r.NameServers == nil || len(r.NameServers) == 0 {
+		var err = errors.New("name server is nil")
+		logger.Error(r.context.GetRuntimeContext(), "FLUSHER_INIT_ALARM", "init rocketmq flusher fail, eerror", err)


typo，以及可以在外层打印错误日志

henryzhx8 · 2023-05-18T09:29:02Z

plugins/flusher/rocketmq/flusher_rocketmq.go

+	return nil
+}
+
+type FlusherFunc func(projectName string, logstoreName string, configName string, logGroupList []*protocol.LogGroup) error


这个除了这种flush模式外，还有其它模式可能吗？

还有单条的，这次没加，等后续再加

shalousun · 2023-05-18T12:17:59Z

licenses/LICENCE-rocketmq

@@ -0,0 +1,201 @@
+ Apache License
+                           Version 2.0, January 2004
+                        http://www.apache.org/licenses/


不是让添加rocketmq的license文件，可以看下LICENSE_OF_ILOGTAIL_DEPENDENCIES.md这个文件里配置，把本次pr新添加的依赖库的license加上

shalousun · 2023-05-18T12:25:23Z

plugins/flusher/rocketmq/flusher_rocketmq.go

+		for index, log := range logs.([][]byte) {
+			valueMap := values[index]
+			topic := r.Topic
+			if len(r.topicKeys) > 0 {


topicKeys这个定义了并未看到实际有用的逻辑？是不是可以参靠flusher pulsar和kafka v2实现下

messixukejia · 2023-06-05T05:17:42Z

麻烦处理下这几个问题

shalousun · 2023-06-06T15:58:37Z

plugins/flusher/rocketmq/flusher_rocketmq.go

+		logger.Debug(r.context.GetRuntimeContext(), "[LogGroup] topic", logGroup.Topic, "logstore", logGroup.Category, "logcount", len(logGroup.Logs), "tags", logGroup.LogTags)
+		logs, values, err := r.converter.ToByteStreamWithSelectedFields(logGroup, r.topicKeys)
+		if err != nil {
+			logger.Error(r.context.GetRuntimeContext(), "FLUSHER_FLUSH_ALARM", "flush kafka convert log fail, error", err)


flush kafka 这个错误提示message需要改下

shalousun · 2023-06-06T16:04:11Z

plugins/flusher/rocketmq/flusher_rocketmq.go

+				logger.Error(r.context.GetRuntimeContext(), "FLUSHER_FLUSH_ALARM", "flush rocketmq error, error", err)
+				return err
+			}
+			logger.Debug(r.context.GetRuntimeContext(), "success flush 2 rocketmq with [LogGroup] projectName", projectName, "logstoreName", logstoreName, "logcount", len(msgs), "res", res.String())


It might be better to use 'to' instead of 2

shalousun · 2023-06-06T16:15:30Z

If "conversation" has been resolved, please mark it as resolved

wlazjr mentioned this pull request May 16, 2023

[FEATURE]: iLogtail support rocketmq Flusher #842

Open

messixukejia requested review from shalousun and henryzhx8 May 16, 2023 05:26

messixukejia linked an issue May 16, 2023 that may be closed by this pull request

[FEATURE]: iLogtail support rocketmq Flusher #842

Open

shalousun reviewed May 16, 2023

View reviewed changes

henryzhx8 requested changes May 18, 2023

View reviewed changes

shalousun requested changes May 18, 2023

View reviewed changes

wlazjr closed this Jun 5, 2023

wlazjr force-pushed the main branch 2 times, most recently from 86f76a6 to 00d6e7e Compare June 5, 2023 07:03

add rocketmq flusher

8df11f5

wlazjr reopened this Jun 5, 2023

add rocketmq flusher

6e27672

shalousun reviewed Jun 6, 2023

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

add rocketmq flusher #850

add rocketmq flusher #850

wlazjr commented May 16, 2023

CLAassistant commented May 16, 2023 •

edited

messixukejia commented May 16, 2023

messixukejia commented May 16, 2023

shalousun May 16, 2023

shalousun May 16, 2023

shalousun commented May 16, 2023 •

edited

shalousun May 16, 2023

wlazjr May 16, 2023

henryzhx8 left a comment

henryzhx8 May 18, 2023

wlazjr May 18, 2023

henryzhx8 May 18, 2023

wlazjr May 18, 2023

shalousun May 18, 2023

henryzhx8 May 18, 2023

henryzhx8 May 18, 2023

wlazjr May 18, 2023

shalousun May 18, 2023

shalousun May 18, 2023

messixukejia commented Jun 5, 2023

shalousun Jun 6, 2023

shalousun Jun 6, 2023

shalousun commented Jun 6, 2023

add rocketmq flusher #850

Are you sure you want to change the base?

add rocketmq flusher #850

Conversation

wlazjr commented May 16, 2023

CLAassistant commented May 16, 2023 • edited

messixukejia commented May 16, 2023

messixukejia commented May 16, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

shalousun commented May 16, 2023 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

henryzhx8 left a 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

messixukejia commented Jun 5, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

shalousun commented Jun 6, 2023

CLAassistant commented May 16, 2023 •

edited

shalousun commented May 16, 2023 •

edited