Enhance the performance of the frontend JSON codec

CHANGELOG.md

@@ -42,6 +42,7 @@
 * [ENHANCEMENT] Parquet Storage: Add some metrics for parquet blocks and converter. #6809 #6821
 * [ENHANCEMENT] Compactor: Optimize cleaner run time. #6815
 * [ENHANCEMENT] Parquet Storage: Allow percentage based dynamic shard size for Parquet Converter. #6817
+* [ENHANCEMENT] Query Frontend: Enhance the performance of the JSON codec. #6816
 * [BUGFIX] Ingester: Avoid error or early throttling when READONLY ingesters are present in the ring #6517
 * [BUGFIX] Ingester: Fix labelset data race condition. #6573
 * [BUGFIX] Compactor: Cleaner should not put deletion marker for blocks with no-compact marker. #6576

pkg/chunk/json_helpers.go

@@ -10,14 +10,14 @@ import (
 )
 
 func init() {
-	jsoniter.RegisterTypeDecoderFunc("labels.Labels", decodeLabels)
-	jsoniter.RegisterTypeEncoderFunc("labels.Labels", encodeLabels, labelsIsEmpty)
+	jsoniter.RegisterTypeDecoderFunc("labels.Labels", DecodeLabels)
+	jsoniter.RegisterTypeEncoderFunc("labels.Labels", EncodeLabels, labelsIsEmpty)
 	jsoniter.RegisterTypeDecoderFunc("model.Time", decodeModelTime)
 	jsoniter.RegisterTypeEncoderFunc("model.Time", encodeModelTime, modelTimeIsEmpty)
 }
 
 // Override Prometheus' labels.Labels decoder which goes via a map
-func decodeLabels(ptr unsafe.Pointer, iter *jsoniter.Iterator) {
+func DecodeLabels(ptr unsafe.Pointer, iter *jsoniter.Iterator) {
 	labelsPtr := (*labels.Labels)(ptr)
 	*labelsPtr = make(labels.Labels, 0, 10)
 	iter.ReadMapCB(func(iter *jsoniter.Iterator, key string) bool {
@@ -31,7 +31,7 @@ func decodeLabels(ptr unsafe.Pointer, iter *jsoniter.Iterator) {
 }
 
 // Override Prometheus' labels.Labels encoder which goes via a map
-func encodeLabels(ptr unsafe.Pointer, stream *jsoniter.Stream) {
+func EncodeLabels(ptr unsafe.Pointer, stream *jsoniter.Stream) {
 	labelsPtr := (*labels.Labels)(ptr)
 	stream.WriteObjectStart()
 	for i, v := range *labelsPtr {

pkg/querier/tripperware/query.go

@@ -25,6 +25,7 @@ import (
 	"github.com/prometheus/prometheus/util/jsonutil"
 	"github.com/weaveworks/common/httpgrpc"
 
+	"github.com/cortexproject/cortex/pkg/chunk"
 	"github.com/cortexproject/cortex/pkg/cortexpb"
 	"github.com/cortexproject/cortex/pkg/util/limiter"
 	"github.com/cortexproject/cortex/pkg/util/runutil"
@@ -113,12 +114,8 @@ func decodeSampleStream(ptr unsafe.Pointer, iter *jsoniter.Iterator) {
 	for field := iter.ReadObject(); field != ""; field = iter.ReadObject() {
 		switch field {
 		case "metric":
-			metricString := iter.ReadAny().ToString()
 			lbls := labels.Labels{}
-			if err := json.UnmarshalFromString(metricString, &lbls); err != nil {
-				iter.ReportError("unmarshal SampleStream", err.Error())
-				return
-			}
+			chunk.DecodeLabels(unsafe.Pointer(&lbls), iter)
 			ss.Labels = cortexpb.FromLabelsToLabelAdapters(lbls)
 		case "values":
 			for iter.ReadArray() {
@@ -302,12 +299,8 @@ func encodeSampleStream(ptr unsafe.Pointer, stream *jsoniter.Stream) {
 	stream.WriteObjectStart()
 
 	stream.WriteObjectField(`metric`)
-	lbls, err := cortexpb.FromLabelAdaptersToLabels(ss.Labels).MarshalJSON()
-	if err != nil {
-		stream.Error = err
-		return
-	}
-	stream.SetBuffer(append(stream.Buffer(), lbls...))
+	metric := cortexpb.FromLabelAdaptersToLabels(ss.Labels)
+	chunk.EncodeLabels(unsafe.Pointer(&metric), stream)
 
 	if len(ss.Samples) > 0 {
 		stream.WriteMore()
@@ -343,12 +336,8 @@ func decodeSample(ptr unsafe.Pointer, iter *jsoniter.Iterator) {
 	for field := iter.ReadObject(); field != ""; field = iter.ReadObject() {
 		switch field {
 		case "metric":
-			metricString := iter.ReadAny().ToString()
 			lbls := labels.Labels{}
-			if err := json.UnmarshalFromString(metricString, &lbls); err != nil {
-				iter.ReportError("unmarshal Sample", err.Error())
-				return
-			}
+			chunk.DecodeLabels(unsafe.Pointer(&lbls), iter)
 			ss.Labels = cortexpb.FromLabelsToLabelAdapters(lbls)
 		case "value":
 			ss.Sample = &cortexpb.Sample{}
@@ -368,12 +357,8 @@ func encodeSample(ptr unsafe.Pointer, stream *jsoniter.Stream) {
 	stream.WriteObjectStart()
 
 	stream.WriteObjectField(`metric`)
-	lbls, err := cortexpb.FromLabelAdaptersToLabels(ss.Labels).MarshalJSON()
-	if err != nil {
-		stream.Error = err
-		return
-	}
-	stream.SetBuffer(append(stream.Buffer(), lbls...))
+	metric := cortexpb.FromLabelAdaptersToLabels(ss.Labels)
+	chunk.EncodeLabels(unsafe.Pointer(&metric), stream)
 
 	if ss.Sample != nil {
 		stream.WriteMore()