vlinsert: improve opentelemetry decode perfomance (#720)

Co-authored-by: Aliaksandr Valialkin <[email protected]>

Author: vadimalekseev

app/vlinsert/opentelemetry/fmt_buffer.go

@@ -0,0 +1,78 @@
+package opentelemetry
+
+import (
+	"encoding/base64"
+	"encoding/hex"
+	"strconv"
+	"sync"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/bytesutil"
+	"github.com/valyala/fastjson"
+)
+
+type fmtBuffer struct {
+	buf []byte
+}
+
+var fmtBufferPool sync.Pool
+
+func getFmtBuffer() *fmtBuffer {
+	v := fmtBufferPool.Get()
+	if v == nil {
+		return &fmtBuffer{}
+	}
+	return v.(*fmtBuffer)
+}
+
+func putFmtBuffer(fb *fmtBuffer) {
+	fb.reset()
+	fmtBufferPool.Put(fb)
+}
+
+func (fb *fmtBuffer) reset() {
+	fb.buf = fb.buf[:0]
+}
+
+func (fb *fmtBuffer) formatInt(v int64) string {
+	n := len(fb.buf)
+	fb.buf = strconv.AppendInt(fb.buf, v, 10)
+	return bytesutil.ToUnsafeString(fb.buf[n:])
+}
+
+func (fb *fmtBuffer) formatFloat(v float64) string {
+	n := len(fb.buf)
+	fb.buf = strconv.AppendFloat(fb.buf, v, 'f', -1, 64)
+	return bytesutil.ToUnsafeString(fb.buf[n:])
+}
+
+func (fb *fmtBuffer) formatSubFieldName(prefix string, suffix []byte) string {
+	if prefix == "" {
+		// There is no prefix, so just return the suffix as is.
+		return bytesutil.ToUnsafeString(suffix)
+	}
+
+	n := len(fb.buf)
+	fb.buf = append(fb.buf, prefix...)
+	fb.buf = append(fb.buf, '.')
+	fb.buf = append(fb.buf, suffix...)
+
+	return bytesutil.ToUnsafeString(fb.buf[n:])
+}
+
+func (fb *fmtBuffer) formatHex(src []byte) string {
+	n := len(fb.buf)
+	fb.buf = hex.AppendEncode(fb.buf, src)
+	return bytesutil.ToUnsafeString(fb.buf[n:])
+}
+
+func (fb *fmtBuffer) formatBase64(src []byte) string {
+	n := len(fb.buf)
+	fb.buf = base64.StdEncoding.AppendEncode(fb.buf, src)
+	return bytesutil.ToUnsafeString(fb.buf[n:])
+}
+
+func (fb *fmtBuffer) encodeJSONValue(v *fastjson.Value) string {
+	n := len(fb.buf)
+	fb.buf = v.MarshalTo(fb.buf)
+	return bytesutil.ToUnsafeString(fb.buf[n:])
+}

app/vlinsert/opentelemetry/opentelemetry.go

@@ -7,7 +7,6 @@ import (
 
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/flagutil"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/httpserver"
-	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/opentelemetry/pb"
 	"github.com/VictoriaMetrics/VictoriaMetrics/lib/protoparser/protoparserutil"
 	"github.com/VictoriaMetrics/metrics"
 
@@ -76,80 +75,20 @@ var (
 )
 
 func pushProtobufRequest(data []byte, lmp insertutil.LogMessageProcessor, msgFields []string, useDefaultStreamFields bool) error {
-	var req pb.ExportLogsServiceRequest
-	if err := req.UnmarshalProtobuf(data); err != nil {
-		errorsTotal.Inc()
-		return fmt.Errorf("cannot unmarshal request from %d bytes: %w", len(data), err)
-	}
-
-	var commonFields []logstorage.Field
-	for _, rl := range req.ResourceLogs {
-		commonFields = commonFields[:0]
-		commonFields = appendKeyValues(commonFields, rl.Resource.Attributes, "")
-		commonFieldsLen := len(commonFields)
-		for _, sc := range rl.ScopeLogs {
-			commonFields = pushFieldsFromScopeLogs(&sc, commonFields[:commonFieldsLen], lmp, msgFields, useDefaultStreamFields)
-		}
-	}
-
-	return nil
-}
-
-func pushFieldsFromScopeLogs(sc *pb.ScopeLogs, commonFields []logstorage.Field, lmp insertutil.LogMessageProcessor, msgFields []string, useDefaultStreamFields bool) []logstorage.Field {
-	fields := commonFields
-	for _, lr := range sc.LogRecords {
-		fields = fields[:len(commonFields)]
-		if lr.Body.KeyValueList != nil {
-			fields = appendKeyValues(fields, lr.Body.KeyValueList.Values, "")
-			logstorage.RenameField(fields[len(commonFields):], msgFields, "_msg")
-		} else {
-			fields = append(fields, logstorage.Field{
-				Name:  "_msg",
-				Value: lr.Body.FormatString(true),
-			})
-		}
-		fields = appendKeyValues(fields, lr.Attributes, "")
-		if len(lr.TraceID) > 0 {
-			fields = append(fields, logstorage.Field{
-				Name:  "trace_id",
-				Value: lr.TraceID,
-			})
-		}
-		if len(lr.SpanID) > 0 {
-			fields = append(fields, logstorage.Field{
-				Name:  "span_id",
-				Value: lr.SpanID,
-			})
-		}
-		fields = append(fields, logstorage.Field{
-			Name:  "severity",
-			Value: lr.FormatSeverity(),
-		})
+	pushLogs := func(timestamp int64, fields []logstorage.Field, resourceFieldsLen int) {
+		logstorage.RenameField(fields[resourceFieldsLen:], msgFields, "_msg")
 
 		var streamFields []logstorage.Field
 		if useDefaultStreamFields {
-			streamFields = commonFields
+			streamFields = fields[:resourceFieldsLen]
 		}
-		lmp.AddRow(lr.ExtractTimestampNano(), fields, streamFields)
-	}
-	return fields
-}
 
-func appendKeyValues(fields []logstorage.Field, kvs []*pb.KeyValue, parentField string) []logstorage.Field {
-	for _, attr := range kvs {
-		fieldName := attr.Key
-		if parentField != "" {
-			fieldName = parentField + "." + fieldName
-		}
+		lmp.AddRow(timestamp, fields, streamFields)
+	}
 
-		if attr.Value.KeyValueList != nil {
-			fields = appendKeyValues(fields, attr.Value.KeyValueList.Values, fieldName)
-		} else {
-			fields = append(fields, logstorage.Field{
-				Name:  fieldName,
-				Value: attr.Value.FormatString(true),
-			})
-		}
+	if err := decodeLogsData(data, pushLogs); err != nil {
+		errorsTotal.Inc()
+		return fmt.Errorf("cannot decode LogsData request from %d bytes: %w", len(data), err)
 	}
-	return fields
+	return nil
 }