receive: try to fix pool race

Signed-off-by: Michael Hoffmann <mhoffm@posteo.de>

.golangci.yml

@@ -17,6 +17,7 @@ run:
   skip-dirs:
     - vendor
     - internal/cortex
+    - pkg/store/storepb/remotewritepb
 
 
 # output configuration options

pkg/receive/handler.go

@@ -825,11 +825,7 @@ func (h *Handler) sendLocalWrite(
 	span.SetTag("endpoint", writeDestination.endpoint)
 	span.SetTag("replica", writeDestination.replica)
 
-	wreq := remotewritepb.WriteRequestFromVTPool()
-	wreq.Timeseries = trackedSeries.timeSeries
-	defer wreq.ReturnToVTPool()
-
-	err := h.writer.Write(tracingCtx, tenant, wreq)
+	err := h.writer.Write(tracingCtx, tenant, trackedSeries.timeSeries)
 	if err != nil {
 		span.SetTag("error", true)
 		span.SetTag("error.msg", err.Error())
@@ -864,15 +860,15 @@ func (h *Handler) sendRemoteWrite(
 	// This is called "real" because it's 1-indexed.
 	realReplicationIndex := int64(endpointReplica.replica + 1)
 
-	wreq := remotewritepb.StoreWriteRequestFromVTPool()
-	wreq.Timeseries = trackedSeries.timeSeries
-	wreq.Tenant = tenant
-	// Increment replica since on-the-wire format is 1-indexed and 0 indicates un-replicated.
-	wreq.Replica = realReplicationIndex
+	wreq := &remotewritepb.StoreWriteRequest{
+		Timeseries: trackedSeries.timeSeries,
+		Tenant:     tenant,
+		// Increment replica since on-the-wire format is 1-indexed and 0 indicates un-replicated.
+		Replica: realReplicationIndex,
+	}
 
 	// Actually make the request against the endpoint we determined should handle these time series.
 	cl.RemoteWriteAsync(ctx, wreq, endpointReplica, trackedSeries.seriesIDs, responses, func(err error) {
-		wreq.ReturnToVTPool()
 		if err == nil {
 			h.forwardRequests.WithLabelValues(labelSuccess).Inc()
 			if !alreadyReplicated {
@@ -908,8 +904,6 @@ func quorumReached(successes []int, successThreshold int) bool {
 
 // RemoteWrite implements the gRPC remote write handler for storepb.WriteableStore.
 func (h *Handler) RemoteWrite(ctx context.Context, r *remotewritepb.StoreWriteRequest) (*remotewritepb.StoreWriteResponse, error) {
-	defer r.ReturnToVTPool()
-
 	span, ctx := tracing.StartSpan(ctx, "receive_grpc")
 	defer span.Finish()
 

pkg/receive/writer.go

@@ -72,7 +72,7 @@ func NewWriter(logger log.Logger, multiTSDB TenantStorage, opts *WriterOptions)
 	}
 }
 
-func (r *Writer) Write(ctx context.Context, tenantID string, wreq *remotewritepb.WriteRequest) error {
+func (r *Writer) Write(ctx context.Context, tenantID string, timeseries []*remotewritepb.TimeSeries) error {
 	tLogger := log.With(r.logger, "tenant", tenantID)
 
 	var (
@@ -112,7 +112,7 @@ func (r *Writer) Write(ctx context.Context, tenantID string, wreq *remotewritepb
 		tooFarInFuture: r.opts.TooFarInFutureTimeWindow,
 		Appender:       app,
 	}
-	for _, t := range wreq.Timeseries {
+	for _, t := range timeseries {
 		// Check if time series labels are valid. If not, skip the time series
 		// and report the error.
 		lset := remotewritepb.LabelsToPromLabels(t.Labels)

scripts/genproto.sh

@@ -62,9 +62,11 @@ for dir in ${DIRS}; do
   ${PROTOC_BIN} \
     --go_out=paths=source_relative:. \
     --plugin protoc-gen-go=${PROTOC_GEN_GO_BIN} \
+    --go-grpc_out=paths=source_relative:. \
+    --plugin protoc-gen-go-grpc=${PROTOC_GEN_GO_GRPC_BIN} \
     --go-vtproto_out=paths=source_relative:. \
     --plugin protoc-gen-go-vtproto=${PROTOC_GEN_GO_VTPROTO_BIN} \
-    --go-vtproto_opt=features=grpc+marshal+unmarshal+size+pool+clone \
+    --go-vtproto_opt=features=marshal+unmarshal+size+pool+clone \
     -I=. \
     -I="${VTPROTO_PATH}" \
     ${dir}/*.proto