diff --git a/cmd/thanos/receive.go b/cmd/thanos/receive.go index 031b82abc6..cd890ce71d 100644 --- a/cmd/thanos/receive.go +++ b/cmd/thanos/receive.go @@ -255,6 +255,8 @@ func runReceive( MaxBackoff: time.Duration(*conf.maxBackoff), TSDBStats: dbs, Limiter: limiter, + + AsyncForwardWorkerCount: conf.asyncForwardWorkerCount, }) grpcProbe := prober.NewGRPC() diff --git a/pkg/receive/handler.go b/pkg/receive/handler.go index c1e76212fb..7125f9d478 100644 --- a/pkg/receive/handler.go +++ b/pkg/receive/handler.go @@ -142,6 +142,7 @@ func NewHandler(logger log.Logger, o *Options) *Handler { if workers == 0 { workers = 1 } + level.Info(logger).Log("msg", "Starting receive handler with async forward workers", "workers", workers) h := &Handler{ logger: logger, @@ -163,7 +164,6 @@ func NewHandler(logger log.Logger, o *Options) *Handler { }, ), workers, - logger, o.DialOpts...), receiverMode: o.ReceiverMode, Limiter: o.Limiter, @@ -777,8 +777,6 @@ func (h *Handler) distributeTimeseriesToReplicas( var writeDestination = remoteWrites if endpoint == h.options.Endpoint { writeDestination = localWrites - } else { - endpointReplica.replica = 0 } writeableSeries, ok := writeDestination[endpointReplica] if !ok { @@ -1212,7 +1210,7 @@ type peerWorker struct { forwardDelay prometheus.Histogram } -func newPeerGroup(backoff backoff.Backoff, forwardDelay prometheus.Histogram, asyncForwardWorkersCount uint, log log.Logger, dialOpts ...grpc.DialOption) peersContainer { +func newPeerGroup(backoff backoff.Backoff, forwardDelay prometheus.Histogram, asyncForwardWorkersCount uint, dialOpts ...grpc.DialOption) peersContainer { return &peerGroup{ dialOpts: dialOpts, connections: map[string]*peerWorker{}, @@ -1222,7 +1220,6 @@ func newPeerGroup(backoff backoff.Backoff, forwardDelay prometheus.Histogram, as expBackoff: backoff, forwardDelay: forwardDelay, asyncForwardWorkersCount: asyncForwardWorkersCount, - log: log, } } @@ -1266,7 +1263,6 @@ type peerGroup struct { expBackoff backoff.Backoff forwardDelay prometheus.Histogram asyncForwardWorkersCount uint - log log.Logger m sync.RWMutex @@ -1322,8 +1318,6 @@ func (p *peerGroup) getConnection(ctx context.Context, addr string) (WriteableSt } p.connections[addr] = newPeerWorker(conn, p.forwardDelay, p.asyncForwardWorkersCount) - log.With(p.log).Log("msg", "established connection to peer", "peer", addr, - "asyncForwardWorkersCount", p.asyncForwardWorkersCount, "poolSize", p.connections[addr].wp.Size()) return p.connections[addr], nil }