fix add raw log

alibaba · Aug 27, 2024 · b44d250 · b44d250
1 parent a93d0fb
commit b44d250
Show file tree

Hide file tree

Showing 3 changed files with 29 additions and 0 deletions.
diff --git a/pkg/helper/pipeline_event_helper.go b/pkg/helper/pipeline_event_helper.go
@@ -45,6 +45,21 @@ func CreateLogEventByArray(t time.Time, enableTimestampNano bool, columns []stri
 	return &logEvent, nil
 }
 
+func CreateLogEventByRawLog(log *protocol.Log) (*protocol.LogEvent, error) {
+	var logEvent protocol.LogEvent
+	logEvent.Contents = make([]*protocol.LogEvent_Content, 0, len(log.Contents))
+	for _, logC := range log.Contents {
+		cont := &protocol.LogEvent_Content{
+			Key:   logC.Key,
+			Value: logC.Value,
+		}
+		logEvent.Contents = append(logEvent.Contents, cont)
+	}
+	logEvent.Time = uint32(log.Time)
+	logEvent.TimeNs = *log.TimeNs
+	return &logEvent, nil
+}
+
 func CreatePipelineEventGroup(metadata map[string]string, configTag map[string]string, logTags map[string]string, logEvents []*protocol.LogEvent) (*protocol.PipelineEventGroup, error) {
 	var pipelineEventGroup protocol.PipelineEventGroup
 	pipelineEventGroup.Metadata = metadata

diff --git a/pluginmanager/plugin_wrapper_metric_v1.go b/pluginmanager/plugin_wrapper_metric_v1.go
@@ -119,6 +119,8 @@ func (p *MetricWrapperV1) AddDataArrayWithContext(tags map[string]string,
 	}
 	if p.Config.GlobalConfig.GoInputToNativeProcessor {
 		logEvent, _ := helper.CreateLogEventByArray(logTime, p.Config.GlobalConfig.EnableTimestampNanosecond, columns, values)
+		p.inputRecordsTotal.Add(1)
+		p.inputRecordsSizeBytes.Add(int64(logEvent.Size()))
 		p.LogsCachedChan <- &pipeline.LogEventWithContext{LogEvent: logEvent, Context: tags}
 		return
 	}
@@ -131,6 +133,11 @@ func (p *MetricWrapperV1) AddDataArrayWithContext(tags map[string]string,
 func (p *MetricWrapperV1) AddRawLogWithContext(log *protocol.Log, ctx map[string]interface{}) {
 	p.inputRecordsTotal.Add(1)
 	p.inputRecordsSizeBytes.Add(int64(log.Size()))
+	if p.Config.GlobalConfig.GoInputToNativeProcessor {
+		logEvent, _ := helper.CreateLogEventByRawLog(log)
+		p.LogsCachedChan <- &pipeline.LogEventWithContext{LogEvent: logEvent, Context: nil}
+		return
+	}
 	p.LogsChan <- &pipeline.LogWithContext{Log: log, Context: ctx}
 }
 

diff --git a/pluginmanager/plugin_wrapper_service_v1.go b/pluginmanager/plugin_wrapper_service_v1.go
@@ -118,6 +118,8 @@ func (p *ServiceWrapperV1) AddDataArrayWithContext(tags map[string]string,
 	// need push to native processor
 	if p.Config.GlobalConfig.GoInputToNativeProcessor {
 		logEvent, _ := helper.CreateLogEventByArray(logTime, p.Config.GlobalConfig.EnableTimestampNanosecond, columns, values)
+		p.inputRecordsTotal.Add(1)
+		p.inputRecordsSizeBytes.Add(int64(logEvent.Size()))
 		p.LogsCachedChan <- &pipeline.LogEventWithContext{LogEvent: logEvent, Context: tags}
 		return
 	}
@@ -130,6 +132,11 @@ func (p *ServiceWrapperV1) AddDataArrayWithContext(tags map[string]string,
 func (p *ServiceWrapperV1) AddRawLogWithContext(log *protocol.Log, ctx map[string]interface{}) {
 	p.inputRecordsTotal.Add(1)
 	p.inputRecordsSizeBytes.Add(int64(log.Size()))
+	if p.Config.GlobalConfig.GoInputToNativeProcessor {
+		logEvent, _ := helper.CreateLogEventByRawLog(log)
+		p.LogsCachedChan <- &pipeline.LogEventWithContext{LogEvent: logEvent, Context: nil}
+		return
+	}
 	p.LogsChan <- &pipeline.LogWithContext{Log: log, Context: ctx}
 }