add offset and partition id for log record (#6382)

erda-project · Jul 2, 2024 · e8c05f5 · e8c05f5
1 parent 0cc5cc8
commit e8c05f5
Show file tree

Hide file tree

Showing 2 changed files with 14 additions and 9 deletions.
diff --git a/internal/tools/monitor/oap/collector/lib/protoparser/spotlog/spotlog.go b/internal/tools/monitor/oap/collector/lib/protoparser/spotlog/spotlog.go
@@ -45,8 +45,8 @@ var (
 	logTerminusKeyTag = "terminus_log_key"
 )
 
-func ParseSpotLog(buf []byte, callback func(m *log.Log) error) error {
-	uw := newUnmarshalWork(buf, callback)
+func ParseSpotLog(buf []byte, offset int64, partition int32, callback func(m *log.Log) error) error {
+	uw := newUnmarshalWork(buf, offset, partition, callback)
 	uw.wg.Add(1)
 	unmarshalwork.Schedule(uw)
 	uw.wg.Wait()
@@ -57,14 +57,16 @@ func ParseSpotLog(buf []byte, callback func(m *log.Log) error) error {
 }
 
 type unmarshalWork struct {
-	buf      []byte
-	err      error
-	wg       sync.WaitGroup
-	callback func(m *log.Log) error
+	buf       []byte
+	partition int32
+	offset    int64
+	err       error
+	wg        sync.WaitGroup
+	callback  func(m *log.Log) error
 }
 
-func newUnmarshalWork(buf []byte, callback func(m *log.Log) error) *unmarshalWork {
-	return &unmarshalWork{buf: buf, callback: callback}
+func newUnmarshalWork(buf []byte, offset int64, partition int32, callback func(m *log.Log) error) *unmarshalWork {
+	return &unmarshalWork{buf: buf, offset: offset, partition: partition, callback: callback}
 }
 
 func (uw *unmarshalWork) Unmarshal() {
@@ -79,6 +81,9 @@ func (uw *unmarshalWork) Unmarshal() {
 		uw.err = err
 		return
 	}
+	data.Offset = uw.offset
+	// set partition as group id
+	data.GroupId = fmt.Sprintf("%d", uw.partition)
 
 	if err := uw.callback(&data.Log); err != nil {
 		uw.err = err

diff --git a/internal/tools/monitor/oap/collector/plugins/receivers/kafka/provider.go b/internal/tools/monitor/oap/collector/plugins/receivers/kafka/provider.go
@@ -187,7 +187,7 @@ func (p *provider) parseSpotProfile() kafkaInf.ConsumerFuncV2 {
 
 func (p *provider) parseSpotLog() kafkaInf.ConsumerFuncV2 {
 	return func(msg *sarama.ConsumerMessage) error {
-		return spotlog.ParseSpotLog(msg.Value, func(log *log.Log) error {
+		return spotlog.ParseSpotLog(msg.Value, msg.Offset, msg.Partition, func(log *log.Log) error {
 			return p.consumeData(log)
 		})
 	}