腾讯mini项目-【指标监控服务重构】2023-08-13

今日已办

使用watermill框架替代当前的base_runner框架
a. 参考官方提供的sarama kafka Pub/Sub(https://github.com/ThreeDotsLabs/watermill-kafka/)实现kafka-go(https://github.com/segmentio/kafka-go)的Pub/Sub（sarama需要cgo，会导致一些额外的镜像依赖）
b. 参考https://watermill.io/docs/messages-router/实现不同topic（不同事件）走不同的Handler处理逻辑，相同处理逻辑则可以使用MiddleWare(https://watermill.io/docs/middlewares/)

watermill-kafka

1. 构造 publisher, subscriber qiulin/watermill-kafkago: Kafka Pub/Sub for the Watermill project, based on segmentio/kafka-go (github.com)
2. 定义 router
3. 获取符合正则表达式的主题
4. 为每一个主题添加 middleware、handler

// Package consumer
// @Author xzx 2023/8/11 18:53:00
package consumer

import (
	"context"
	kc "github.com/Kevinello/kafka-client"
	"github.com/ThreeDotsLabs/watermill"
	"github.com/ThreeDotsLabs/watermill/message"
	"github.com/ThreeDotsLabs/watermill/message/router/middleware"
	"github.com/ThreeDotsLabs/watermill/message/router/plugin"
	"github.com/qiulin/watermill-kafkago/pkg/kafkago"
	"go.uber.org/zap"
	"profile/internal/config"
	"profile/internal/connector"
	"profile/internal/log"
	"profile/internal/schema"
	"profile/internal/schema/performance"
	"strings"
	"time"
)

// ProfileContext
// @Description:
// @Author xzx 2023-08-11 22:21:41
type ProfileContext struct {
	Status int
	Ctx    context.Context

	Router *message.Router
	Event  schema.Event

	AppID         string // API 上报
	FetchScenario string // API 上报
}

// NewProfileContext
// @Description
// @Author xzx 2023-08-11 22:49:00
// @Return *ProfileContext
func NewProfileContext() *ProfileContext {
	profileCtx := &ProfileContext{
		Ctx: context.Background(),
	}
	profileCtx.init()
	return profileCtx
}

// init
// @Description 初始化
// @Author xzx 2023-08-11 22:22:01
func (profileCtx *ProfileContext) init() {
	logger := watermill.NewStdLogger(false, false)
	publisher, subscriber := newPubSub()
	router, err := message.NewRouter(message.RouterConfig{}, logger)
	if err != nil {
		log.Logger.Fatal("creates a new Router with given configuration error", zap.Error(err))
	}

	router.AddPlugin(plugin.SignalsHandler)
	router.AddMiddleware(
		middleware.Retry{
			MaxRetries:      3,
			InitialInterval: time.Millisecond * 100,
			Logger:          logger,
		}.Middleware,
		middleware.Recoverer,
	)

	getTopics := kc.GetTopicReMatch(strings.Split(config.Profile.GetString("kafka.topicRE"), ","))
	topics, err := getTopics(config.Profile.GetString("kafka.bootstrap"))
	if err != nil {
		log.Logger.Fatal("get topics failed", zap.Error(err))
		return
	}

	for _, topic := range topics {
		var category string
		if strings.Contains(topic, performance.CategoryCrash) {
			category = performance.CategoryCrash
		} else if strings.Contains(topic, performance.CategoryLag) {
			category = performance.CategoryLag
		} else {
			continue
		}
		router.AddHandler("profile-consume", topic, subscriber, connector.GetTopic(category), publisher, profileCtx.WriteKafka).
			AddMiddleware(
				profileCtx.UnpackKafkaMessage,
				profileCtx.InitPerformanceEvent,
				profileCtx.AnalyzeEvent,
			)
	}
	profileCtx.Router = router
}

// Run
// @Description
// @Author xzx 2023-08-12 13:52:53
func (profileCtx *ProfileContext) Run() {
	// router.Run contains defer cancel()
	if err := profileCtx.Router.Run(profileCtx.Ctx); err != nil {
		log.Logger.Error("runs all plugins and handlers and starts subscribing to provided topics error", zap.Error(err))
	}
}

func (profileCtx *ProfileContext) Done() <-chan struct{} {
	return profileCtx.Ctx.Done()
}

func (profileCtx *ProfileContext) Err() error {
	return profileCtx.Ctx.Err()
}

func (profileCtx *ProfileContext) Deadline() (deadline time.Time, ok bool) {
	return profileCtx.Ctx.Deadline()
}

func (profileCtx *ProfileContext) Value(key any) any {
	return profileCtx.Ctx.Value(key)
}

// newPubSub
// @Description
// @Author xzx 2023-08-13 16:00:26
// @Return message.Publisher
// @Return message.Subscriber
func newPubSub() (message.Publisher, message.Subscriber) {
	logger := watermill.NewStdLogger(false, false)
	marshaler := kafkago.DefaultMarshaler{}
	publisher := kafkago.NewPublisher(kafkago.PublisherConfig{
		Brokers:     []string{config.Profile.GetString("kafka.bootstrap")},
		Async:       false,
		Marshaler:   marshaler,
		OTELEnabled: false,
		Ipv4Only:    true,
		Timeout:     100 * time.Second,
	}, logger)

	subscriber, err := kafkago.NewSubscriber(kafkago.SubscriberConfig{
		Brokers:       []string{config.Profile.GetString("kafka.bootstrap")},
		Unmarshaler:   marshaler,
		ConsumerGroup: config.Profile.GetString("kafka.group"),
		OTELEnabled:   false,
	}, logger)
	if err != nil {
		log.Logger.Fatal("Unable to create subscriber", zap.Error(err))
	}
	return publisher, subscriber
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149

组内会议

已完成：

1. 找到了基于 kafka-go 的 watermill 的 pub/sub 模型：https://github.com/qiulin/watermill-kafkago
2. 将 baserunner 替换为 watermill 的 pub/sub 中可以大致还原原本的功能
3. grafana 展示 trace 的SQL比较复杂，因为表结构的定义有出入，需要进行重命名，时间范围也需要重新修改SQL，而且组员反馈 grafana 的可视化并没有signoz 来得直观友好

待办：

1. 调研 HyperScan golang implement 和 benchmark
2. 优化 watermill-pub/sub 的逻辑

问题：

1. watermill-pub/sub : 目前的做法有将4个阶段整合为一个 Handler 来处理，还有将其抽离为Middleware 的做法，原有 baserunner 的逻辑是所有 topic 的都走 4 个阶段的 Handler
2. 是否可以停止 grafana 展示 trace 的工作

明日待办