Merge pull request #8547 from thanos-io/pull_fix

Pull extgrpc fix and release 0.40.1

Author: GiedriusS

CHANGELOG.md

@@ -8,6 +8,16 @@ NOTE: As semantic versioning states all 0.y.z releases can contain breaking chan
 
 We use *breaking :warning:* to mark changes that are not backward compatible (relates only to v0.y.z releases.)
 
+## [v0.40.1](https://github.com/thanos-io/thanos/tree/release-0.40) - 2025 10 28
+
+### Fixed
+
+- [#8545](https://github.com/thanos-io/thanos/pull/8545) extgrpc: call gogo methods from the codec
+
+### Added
+
+### Changed
+
 ## [v0.40.0](https://github.com/thanos-io/thanos/tree/release-0.40) - 2025 10 27
 
 ### Fixed

VERSION

@@ -1 +1 @@
-0.40.0
+0.40.1

pkg/extgrpc/client.go

@@ -32,7 +32,13 @@ type nonPoolingCodec struct {
 }
 
 func init() {
-	encoding.RegisterCodecV2(&nonPoolingCodec{CodecV2: encoding.GetCodecV2("proto")})
+	encoding.RegisterCodecV2(&nonPoolingCodec{
+		CodecV2: encoding.GetCodecV2("proto"),
+	})
+}
+
+func (n *nonPoolingCodec) Name() string {
+	return "proto"
 }
 
 func messageV2Of(v any) protobufproto.Message {
@@ -57,8 +63,29 @@ func (n *nonPoolingCodec) Unmarshal(data mem.BufferSlice, v any) error {
 	return nil
 }
 
-func (n *nonPoolingCodec) Marshal(v any) (out mem.BufferSlice, err error) {
-	return n.CodecV2.Marshal(v)
+type gogoMsg interface {
+	Size() int
+	MarshalToSizedBuffer([]byte) (int, error)
+}
+
+func (c *nonPoolingCodec) Marshal(v any) (mem.BufferSlice, error) {
+	gmsg, ok := v.(gogoMsg)
+	if !ok {
+		return c.CodecV2.Marshal(v)
+	}
+	size := gmsg.Size()
+	pool := mem.DefaultBufferPool()
+	buf := pool.Get(size)
+
+	n, err := gmsg.MarshalToSizedBuffer((*buf)[:size])
+	if err != nil {
+		pool.Put(buf)
+		return mem.BufferSlice{}, err
+	}
+
+	bufExact := (*buf)[:n]
+
+	return mem.BufferSlice{mem.NewBuffer(&bufExact, pool)}, nil
 }
 
 // EndpointGroupGRPCOpts creates gRPC dial options for connecting to endpoint groups.

pkg/extgrpc/client_test.go

@@ -0,0 +1,101 @@
+// Copyright (c) The Thanos Authors.
+// Licensed under the Apache License 2.0.
+
+package extgrpc
+
+import (
+	"testing"
+
+	"github.com/prometheus/prometheus/model/labels"
+	"github.com/stretchr/testify/require"
+	"github.com/thanos-io/thanos/pkg/info/infopb"
+	"github.com/thanos-io/thanos/pkg/store/labelpb"
+	"google.golang.org/grpc/encoding"
+)
+
+func BenchmarkVanillaProtoMarshal(b *testing.B) {
+	ls := []labels.Labels{}
+
+	for i := range 1000 {
+		ls = append(ls, labels.FromStrings("label_"+string(rune(i)), "value_"+string(rune(i))))
+	}
+	i := &infopb.InfoResponse{
+		LabelSets: labelpb.ZLabelSetsFromPromLabels(
+			ls...,
+		),
+		ComponentType: "asdsadsadsa",
+		Store: &infopb.StoreInfo{
+			MinTime:          123,
+			MaxTime:          456,
+			SupportsSharding: true,
+			TsdbInfos: []infopb.TSDBInfo{
+				{
+					Labels:  labelpb.ZLabelSet{Labels: labelpb.ZLabelsFromPromLabels(labels.FromStrings("replica", "1"))},
+					MinTime: 100,
+					MaxTime: 200,
+				},
+			},
+		},
+		Rules:          &infopb.RulesInfo{},
+		MetricMetadata: &infopb.MetricMetadataInfo{},
+		Exemplars: &infopb.ExemplarsInfo{
+			MinTime: 100,
+			MaxTime: 200,
+		},
+	}
+
+	c := &nonPoolingCodec{}
+
+	ogCodec := encoding.GetCodecV2("proto").(*nonPoolingCodec).CodecV2
+
+	b.Run("vanilla proto marshal", func(b *testing.B) {
+		b.ReportAllocs()
+		b.ResetTimer()
+
+		for b.Loop() {
+			_, err := ogCodec.Marshal(i)
+			require.NoError(b, err)
+		}
+	})
+
+	b.Run("non pooling codec marshal", func(b *testing.B) {
+		for b.Loop() {
+			_, err := c.Marshal(i)
+			require.NoError(b, err)
+		}
+	})
+}
+
+func TestNonPoolingCodecMarshal(t *testing.T) {
+	i := &infopb.InfoResponse{
+		LabelSets: labelpb.ZLabelSetsFromPromLabels(
+			labels.FromStrings("foo", "bar"),
+			labels.FromStrings("baz", "qux"),
+			labels.FromStrings("aaa", "bbb"),
+		),
+		ComponentType: "asdsadsadsa",
+		Store: &infopb.StoreInfo{
+			MinTime:          123,
+			MaxTime:          456,
+			SupportsSharding: true,
+			TsdbInfos: []infopb.TSDBInfo{
+				{
+					Labels:  labelpb.ZLabelSet{Labels: labelpb.ZLabelsFromPromLabels(labels.FromStrings("replica", "1"))},
+					MinTime: 100,
+					MaxTime: 200,
+				},
+			},
+		},
+		Rules:          &infopb.RulesInfo{},
+		MetricMetadata: &infopb.MetricMetadataInfo{},
+		Exemplars: &infopb.ExemplarsInfo{
+			MinTime: 100,
+			MaxTime: 200,
+		},
+	}
+
+	c := &nonPoolingCodec{}
+
+	_, err := c.Marshal(i)
+	require.NoError(t, err)
+}