[PLAT-105856] fix grpc retry strategy (#31)

Author: jnyi

cmd/thanos/receive.go

@@ -157,6 +157,9 @@ func runReceive(
 	if conf.compression != compressionNone {
 		dialOpts = append(dialOpts, grpc.WithDefaultCallOptions(grpc.UseCompressor(conf.compression)))
 	}
+	if receiveMode == receive.RouterOnly {
+		dialOpts = append(dialOpts, extgrpc.EndpointGroupGRPCOpts()...)
+	}
 
 	var bkt objstore.Bucket
 	confContentYaml, err := conf.objStoreConfig.Content()

pkg/pool/worker_pool.go

@@ -28,26 +28,28 @@ type WorkerPool interface {
 
 type workerPool struct {
 	sync.Once
+	ctx    context.Context
 	workCh chan Work
 	cancel context.CancelFunc
 }
 
 func NewWorkerPool(workers uint) WorkerPool {
+	ctx, cancel := context.WithCancel(context.Background())
 	return &workerPool{
+		ctx:    ctx,
+		cancel: cancel,
 		workCh: make(chan Work, workers),
 	}
 }
 
 func (p *workerPool) Init() {
 	p.Do(func() {
-		ctx, cancel := context.WithCancel(context.Background())
-		p.cancel = cancel
-
 		for i := 0; i < cap(p.workCh); i++ {
 			go func() {
 				for {
 					select {
-					case <-ctx.Done():
+					case <-p.ctx.Done():
+						// TODO: exhaust workCh before exit
 						return
 					case work := <-p.workCh:
 						work()

pkg/receive/handler.go

@@ -829,6 +829,8 @@ func (h *Handler) sendLocalWrite(
 	defer span.Finish()
 	span.SetTag("endpoint", writeDestination.endpoint)
 	span.SetTag("replica", writeDestination.replica)
+	span.SetTag("tenant", tenant)
+	span.SetTag("samples", len(trackedSeries.timeSeries))
 	err := h.writer.Write(tracingCtx, tenant, &prompb.WriteRequest{
 		Timeseries: trackedSeries.timeSeries,
 	})

pkg/receive/writer.go

@@ -19,6 +19,7 @@ import (
 
 	"github.com/thanos-io/thanos/pkg/store/labelpb"
 	"github.com/thanos-io/thanos/pkg/store/storepb/prompb"
+	"github.com/thanos-io/thanos/pkg/tracing"
 )
 
 // Appendable returns an Appender.
@@ -277,7 +278,8 @@ func (r *Writer) Write(ctx context.Context, tenantID string, wreq *prompb.WriteR
 		level.Info(tLogger).Log("msg", "Error on ingesting exemplars with label length exceeding maximum limit", "numDropped", numExemplarsLabelLength)
 		errs.Add(errors.Wrapf(storage.ErrExemplarLabelLength, "add %d exemplars", numExemplarsLabelLength))
 	}
-
+	span, _ := tracing.StartSpan(ctx, "receive_commit")
+	defer span.Finish()
 	if err := app.Commit(); err != nil {
 		errs.Add(errors.Wrap(err, "commit samples"))
 	}