feat: transfer 优化日志处理效率 --story=117618448 (#351)

TencentBlueKing · May 27, 2024 · 8a5df6a · 8a5df6a
1 parent da9a18c
commit 8a5df6a
Show file tree

Hide file tree

Showing 4 changed files with 31 additions and 5 deletions.
diff --git a/pkg/transfer/define/interface.go b/pkg/transfer/define/interface.go
@@ -48,6 +48,9 @@ type Payload interface {
 	// Meta info
 	Meta() PayloadMeta
 
+	SetETLRecord(*ETLRecord)
+	GetETLRecord() *ETLRecord
+
 	// SetTime sets the time received
 	SetTime(t time.Time)
 	// GetTime gets the time received

diff --git a/pkg/transfer/define/payload.go b/pkg/transfer/define/payload.go
@@ -32,6 +32,8 @@ type BasePayload struct {
 	Data []byte
 	t    time.Time
 	flag PayloadFlag
+
+	r *ETLRecord
 }
 
 func (p BasePayload) copy() *BasePayload {
@@ -109,6 +111,14 @@ func (p *BasePayload) Flag() PayloadFlag {
 	return p.flag
 }
 
+func (p *BasePayload) SetETLRecord(r *ETLRecord) {
+	p.r = r
+}
+
+func (p *BasePayload) GetETLRecord() *ETLRecord {
+	return p.r
+}
+
 // NewBasePayloadFrom :
 func NewBasePayloadFrom(data []byte, sn int) *BasePayload {
 	return &BasePayload{

diff --git a/pkg/transfer/elasticsearch/backend.go b/pkg/transfer/elasticsearch/backend.go
@@ -88,10 +88,15 @@ func (b *BulkHandler) asRecord(etlRecord *define.ETLRecord) (*Record, error) {
 // Product
 func (b *BulkHandler) Handle(ctx context.Context, payload define.Payload, killChan chan<- error) (result interface{}, at time.Time, ok bool) {
 	var etlRecord define.ETLRecord
-	err := payload.To(&etlRecord)
-	if err != nil {
-		logging.Warnf("%v error %v dropped payload %+v", b, err, payload)
-		return nil, time.Time{}, false
+	r := payload.GetETLRecord()
+	if r != nil {
+		etlRecord = *r
+	} else {
+		err := payload.To(&etlRecord)
+		if err != nil {
+			logging.Warnf("%v error %v dropped payload %+v", b, err, payload)
+			return nil, time.Time{}, false
+		}
 	}
 
 	return &etlRecord, utils.ParseTimeStamp(*etlRecord.Time), true

diff --git a/pkg/transfer/template/etl/formatter/processor.go b/pkg/transfer/template/etl/formatter/processor.go
@@ -42,6 +42,7 @@ type Processor struct {
 	*define.ProcessorMonitor
 	pipelineConfig *config.PipelineConfig
 	handlers       RecordHandlers
+	etlConfig      string
 }
 
 // Process : process json data
@@ -55,12 +56,18 @@ func (p *Processor) Process(d define.Payload, outputChan chan<- define.Payload,
 	}
 
 	err = p.handlers.Handle(record, func(record *define.ETLRecord) error {
+		// 只对日志数据进行处理
+		if p.etlConfig == "bk_flat_batch" {
+			d.SetETLRecord(record)
+			outputChan <- d
+			return nil
+		}
+
 		payload, err := define.DerivePayload(d, record)
 		if err != nil {
 			logging.Warnf("%v handle payload %#v failed: %v", p, d, err)
 			return err
 		}
-
 		outputChan <- payload
 		return nil
 	})
@@ -81,6 +88,7 @@ func NewProcessor(ctx context.Context, name string, handlers RecordHandlers) *Pr
 		BaseDataProcessor: define.NewBaseDataProcessor(pipeConf.FormatName(name)),
 		ProcessorMonitor:  pipeline.NewDataProcessorMonitor(name, config.PipelineConfigFromContext(ctx)),
 		handlers:          handlers,
+		etlConfig:         pipeConf.ETLConfig,
 	}
 
 	return p