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

    今日已办

    watermill

    将 key 设置到 message 中

    修改 watermill-kafka 源码 将 key 设置到 message.metadata中

    image-20230818224812801

    image-20230818224728773

    接入 otel-sdk

    1. 添加 middleware resolveUpstreamCtx 解析上游上下文，开启根Span
    2. 添加 middleware middleware.InstantAck - 马上ACK，使得多条消息可以平行处理（走middleware 和 handler 的逻辑）
    // Package pubsub
// @Author xzx 2023/8/12 10:01:00
package pubsub

import (
	"context"
	"encoding/json"
	"github.com/ThreeDotsLabs/watermill/message"
	"go.opentelemetry.io/otel"
	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/propagation"
	semconv "go.opentelemetry.io/otel/semconv/v1.17.0"
	"go.opentelemetry.io/otel/trace"
	"go.uber.org/zap"
	"profile/internal/connector"
	"profile/internal/log"
	"profile/internal/otelclient"
	"profile/internal/schema"
	"profile/internal/schema/performance"
	"profile/internal/state"
	"profile/internal/watermill/watermillkafka"
)

// consumeCtxData
// @Description: Data collection of a message processing context
// @Author xzx 2023-08-17 13:36:12
type consumeCtxData struct {
	Status        int
	Event         schema.Event
	RootSpan      trace.Span
	RootSpanCtx   context.Context
	AppID         string // API 上报
	FetchScenario string // API 上报
}

// resolveUpstreamCtx
// @Description
// @Author xzx 2023-08-18 11:15:09
// @Param h
// @Return message.HandlerFunc
func resolveUpstreamCtx(h message.HandlerFunc) message.HandlerFunc {
	return func(msg *message.Message) ([]*message.Message, error) {
		var data consumeCtxData
		// get upstream producer's W3C trace context via propagation
		headerCarrier := make(propagation.HeaderCarrier)

		headerCarrier.Set("Traceparent", msg.Metadata.Get("Traceparent"))
		upstreamProducerCtx := otel.GetTextMapPropagator().Extract(msg.Context(), headerCarrier)
		// set traceID to consumer context
		consumerCtx := trace.ContextWithRemoteSpanContext(msg.Context(),
			trace.NewSpanContext(trace.SpanContextConfig{
				TraceID: trace.SpanContextFromContext(upstreamProducerCtx).TraceID(),
			}))
		//start tracing
		data.RootSpanCtx, data.RootSpan = otelclient.ConsumerTracer.Start(consumerCtx, "Profile-Consumer",
			trace.WithSpanKind(trace.SpanKindConsumer),
			trace.WithLinks(trace.LinkFromContext(upstreamProducerCtx, semconv.OpentracingRefTypeFollowsFrom)))

		msg.SetContext(context.WithValue(msg.Context(), "data", &data))
		return h(msg)
	}
}

// unpackKafkaMessage
// @Description
// @Author xzx 2023-08-12 12:27:30
// @Param hmsg.SetContext(context.WithValue(msg.Context(), "data", data))
// @Return message.HandlerFunc
func unpackKafkaMessage(h message.HandlerFunc) message.HandlerFunc {
	return func(msg *message.Message) ([]*message.Message, error) {
		data := msg.Context().Value("data").(*consumeCtxData)

		unpackKafkaMessageCtx, span := otelclient.ConsumerTracer.Start(data.RootSpanCtx, "unpackKafkaMessage",
			trace.WithSpanKind(trace.SpanKindClient))
		// 反序列化，存入通用结构体
		if contextErr := json.Unmarshal(msg.Payload, &data.Event); contextErr != nil {
			data.Status = state.StatusUnmarshalError
			handlerErr(unpackKafkaMessageCtx, "unmarshal error", contextErr)
			span.End()
			return nil, contextErr
		}
		log.Logger.InfoContext(unpackKafkaMessageCtx, "[UnpackKafkaItem] unpack kafka item success",
			zap.Any("event", data.Event),
			zap.String("profile_root_span_id", data.RootSpan.SpanContext().SpanID().String()))

		setSpanAttributes(span, data)

		msg.SetContext(context.WithValue(msg.Context(), "data", data))
		span.End()
		return h(msg)
	}
}

// initPerformanceEvent
// @Description
// @Author xzx 2023-08-12 12:27:35
// @Param h
// @Return message.HandlerFunc
func initPerformanceEvent(h message.HandlerFunc) message.HandlerFunc {
	return func(msg *message.Message) ([]*message.Message, error) {
		data := msg.Context().Value("data").(*consumeCtxData)

		initPerformanceEventCtx, span := otelclient.ConsumerTracer.Start(data.RootSpanCtx, "initPerformanceEvent",
			trace.WithSpanKind(trace.SpanKindInternal))
		event, contextErr := performance.EventFactory(data.Event.Category, data.Event.Dimensions, data.Event.Values)
		if contextErr != nil {
			data.Status = state.StatusEventFactoryError
			handlerErr(initPerformanceEventCtx, "event factory error", contextErr)
			span.End()
			return nil, contextErr
		}
		log.Logger.InfoContext(initPerformanceEventCtx, "[initPerformanceEvent] init performance event success",
			zap.Any("event", event),
			zap.String("profile_root_span_id", data.RootSpan.SpanContext().SpanID().String()))
		data.Event.ProfileData = event

		setSpanAttributes(span, data)

		msg.SetContext(context.WithValue(msg.Context(), "data", data))
		span.End()
		return h(msg)
	}
}

// analyzeEvent
// @Description
// @Author xzx 2023-08-12 12:27:38
// @Param h
// @Return message.HandlerFunc
func analyzeEvent(h message.HandlerFunc) message.HandlerFunc {
	return func(msg *message.Message) ([]*message.Message, error) {
		data := msg.Context().Value("data").(*consumeCtxData)

		analyzeEventCtx, span := otelclient.ConsumerTracer.Start(data.RootSpanCtx, "analyzeEvent",
			trace.WithSpanKind(trace.SpanKindInternal))

		contextErr := data.Event.ProfileData.Analyze()
		if contextErr != nil {
			data.Status = state.StatusAnalyzeError
			handlerErr(analyzeEventCtx, "analyze event error", contextErr)
			span.End()
			return nil, contextErr
		}
		log.Logger.InfoContext(analyzeEventCtx, "[analyzeEvent] analyze performance event success",
			zap.Any("event", data.Event),
			zap.String("profile_root_span_id", data.RootSpan.SpanContext().SpanID().String()))
		// clear dimensions and values
		data.Event.Dimensions = nil
		data.Event.Values = nil

		setSpanAttributes(span, data)

		msg.SetContext(context.WithValue(msg.Context(), "data", data))
		span.End()
		return h(msg)
	}
}

// crashHandler
// @Description
// @Author xzx 2023-08-12 15:09:15
// @Param msg
// @Return []*message.Message
// @Return error
func crashHandler(msg *message.Message) ([]*message.Message, error) {
	data := msg.Context().Value("data").(*consumeCtxData)

	writeKafkaCtx, span := otelclient.ConsumerTracer.Start(data.RootSpanCtx, "crashHandler",
		trace.WithSpanKind(trace.SpanKindProducer))
	toWriteBytes, contextErr := json.Marshal(data.Event)
	if contextErr != nil {
		data.Status = state.StatusUnmarshalError
		handlerErr(writeKafkaCtx, "marshal error", contextErr)
		span.End()
		return nil, contextErr
	}

	msg = message.NewMessage(data.Event.BackendID, toWriteBytes)
	msg.Metadata.Set(watermillkafka.HeaderKey, data.Event.ID)
	log.Logger.Info("[4-crashHandler] write kafka ...", zap.String("topic", connector.GetTopic(data.Event.Category)), zap.String("id", data.Event.ID), zap.String("msg", string(toWriteBytes)))

	setSpanAttributes(span, data)
	span.End()
	data.RootSpan.End()
	return message.Messages{msg}, nil
}

// lagHandler
// @Description
// @Author xzx 2023-08-12 15:09:15
// @Param msg
// @Return []*message.Message
// @Return error
func lagHandler(msg *message.Message) ([]*message.Message, error) {
	data := msg.Context().Value("data").(*consumeCtxData)

	writeKafkaCtx, span := otelclient.ConsumerTracer.Start(data.RootSpanCtx, "crashHandler",
		trace.WithSpanKind(trace.SpanKindProducer))
	toWriteBytes, contextErr := json.Marshal(data.Event)
	if contextErr != nil {
		data.Status = state.StatusUnmarshalError
		handlerErr(writeKafkaCtx, "marshal error", contextErr)
		span.End()
		return nil, contextErr
	}

	msg = message.NewMessage(data.Event.BackendID, toWriteBytes)
	log.Logger.Info("[4-lagHandler] write kafka ...", zap.String("topic", connector.GetTopic(data.Event.Category)), zap.String("id", data.Event.ID), zap.String("msg", string(toWriteBytes)))
	setSpanAttributes(span, data)
	span.End()
	data.RootSpan.End()
	return message.Messages{msg}, nil
}

// setSpanAttributes
// @Description  setSpanAttributes
// @Author xzx 2023-08-03 23:19:17
// @Param span
// @Param profileCtx
func setSpanAttributes(span trace.Span, data *consumeCtxData) {
	if span.IsRecording() {
		span.SetAttributes(
			attribute.String("event.category", data.Event.Category),
			attribute.String("event.backend_id", data.Event.BackendID),
		)
	}
}

// handlerErr
// @Description
// @Author xzx 2023-07-20 15:36:46
// @Param span
// @Param ctx
// @Param msg
// @Param err
func handlerErr(ctx context.Context, msg string, err error) {
	log.Logger.ErrorContext(ctx, msg, zap.Error(err))
}


    • 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
    • 150
    • 151
    • 152
    • 153
    • 154
    • 155
    • 156
    • 157
    • 158
    • 159
    • 160
    • 161
    • 162
    • 163
    • 164
    • 165
    • 166
    • 167
    • 168
    • 169
    • 170
    • 171
    • 172
    • 173
    • 174
    • 175
    • 176
    • 177
    • 178
    • 179
    • 180
    • 181
    • 182
    • 183
    • 184
    • 185
    • 186
    • 187
    • 188
    • 189
    • 190
    • 191
    • 192
    • 193
    • 194
    • 195
    • 196
    • 197
    • 198
    • 199
    • 200
    • 201
    • 202
    • 203
    • 204
    • 205
    • 206
    • 207
    • 208
    • 209
    • 210
    • 211
    • 212
    • 213
    • 214
    • 215
    • 216
    • 217
    • 218
    • 219
    • 220
    • 221
    • 222
    • 223
    • 224
    • 225
    • 226
    • 227
    • 228
    • 229
    • 230
    • 231
    • 232
    • 233
    • 234
    • 235
    • 236
    • 237
    • 238
    • 239

    会议纪要

    进度

    1. venus 的 metrics 独立分支开发
    2. venus 的 trace 修复了一些bug
      1. 返回 error 主动调用 span.end()
    3. profile 的 watemill pub/sub 和 trace 上报还原原本功能
    4. profile 的 hyperscan 的继续调研中

    待办

    1. 调研如何关闭otel，设置开关配置
      1. 找到了 sdktrace.WithSampler(sdktrace.NeverSample()) - 可以不上报 trace
    2. 性能benchmark，单topic、多topic的场景下，对比原profile代码和现有watermill代码性能
    3. 调研 watermill 的 otelEnabled 的功能，其他集成 otel 的第三方库等
    4. hyperscan 的 benchmark

    明日待办

    1. 性能benchmark，单topic、多topic的场景下，对比原profile代码和现有watermill代码性能
    2. 在 watermill-kafka 的源码基础上对 publisher 的 写回 kafka 添加 otel - trace & log 的逻辑
  • 相关阅读:
    【JAVA】-- 简易超市管理系统（二）（实现思路+每步代码）
    python设计模式_Python六大原则，23种设计模式
    最新面试详情
    三天快速搭建一个属于自己的管理系统，包括前端、后端、部署上线，超级详细的介绍。
    蓝桥杯-粘木棍-DFS
    Java泛型的总结
    JAVA反射机制详解
    C++实现enum反射，类似magic_enum，支持enum classes
    JSP中的cookie对象和Date时间类
    一个简单易用的m3u8下载器，支持下载m3u8链接或文件为mp4或ts格式
  • 原文地址：https://blog.csdn.net/xzx18822942899/article/details/132913909