Fix bug in scheduler processor (#6573)

* scheduler processor is creating grpc connections to frontend to send result. It needs to use frontend client config. Signed-off-by: Peter Štibraný <[email protected]> * Test. Signed-off-by: Peter Štibraný <[email protected]> * CHANGELOG.md Signed-off-by: Peter Štibraný <[email protected]> --------- Signed-off-by: Peter Štibraný <[email protected]>
grafana · Nov 6, 2023 · 0f6d6e5 · 0f6d6e5
1 parent d85a7bf
commit 0f6d6e5
Show file tree

Hide file tree

Showing 3 changed files with 22 additions and 3 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -4,7 +4,7 @@
 
 ### Grafana Mimir
 
-* [CHANGE] Querier: Split worker GRPC config into separate client configs for the frontend and scheduler to allow TLS to be configured correctly when specifying the `tls_server_name`. The GRPC config specified under `-querier.frontend-client.*` will no longer apply to the scheduler client, and will need to be set explicitly under `-querier.scheduler-client.*`. #6445
+* [CHANGE] Querier: Split worker GRPC config into separate client configs for the frontend and scheduler to allow TLS to be configured correctly when specifying the `tls_server_name`. The GRPC config specified under `-querier.frontend-client.*` will no longer apply to the scheduler client, and will need to be set explicitly under `-querier.scheduler-client.*`. #6445 #6573
 * [CHANGE] Store-gateway: enable sparse index headers by default. Sparse index headers reduce the time to load an index header up to 90%. #6005
 * [CHANGE] Store-gateway: lazy-loading concurrency limit default value is now 4. #6004
 * [CHANGE] General: enabled `-log.buffered` by default. The `-log.buffered` has been deprecated and will be removed in Mimir 2.13. #6131

diff --git a/pkg/querier/worker/scheduler_processor.go b/pkg/querier/worker/scheduler_processor.go
@@ -43,9 +43,9 @@ func newSchedulerProcessor(cfg Config, handler RequestHandler, log log.Logger, r
 	p := &schedulerProcessor{
 		log:            log,
 		handler:        handler,
-		maxMessageSize: cfg.QuerySchedulerGRPCClientConfig.MaxSendMsgSize,
+		maxMessageSize: cfg.QueryFrontendGRPCClientConfig.MaxSendMsgSize,
 		querierID:      cfg.QuerierID,
-		grpcConfig:     cfg.QuerySchedulerGRPCClientConfig,
+		grpcConfig:     cfg.QueryFrontendGRPCClientConfig,
 
 		schedulerClientFactory: func(conn *grpc.ClientConn) schedulerpb.SchedulerForQuerierClient {
 			return schedulerpb.NewSchedulerForQuerierClient(conn)

diff --git a/pkg/querier/worker/scheduler_processor_test.go b/pkg/querier/worker/scheduler_processor_test.go
@@ -10,6 +10,8 @@ import (
 	"github.com/go-kit/log"
 	"github.com/gogo/status"
 	"github.com/grafana/dskit/concurrency"
+	"github.com/grafana/dskit/flagext"
+	"github.com/grafana/dskit/grpcclient"
 	"github.com/grafana/dskit/httpgrpc"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
@@ -179,6 +181,23 @@ func TestSchedulerProcessor_QueryTime(t *testing.T) {
 	})
 }
 
+func TestCreateSchedulerProcessor(t *testing.T) {
+	conf := grpcclient.Config{}
+	flagext.DefaultValues(&conf)
+	conf.MaxSendMsgSize = 1 * 1024 * 1024
+
+	sp, _ := newSchedulerProcessor(Config{
+		SchedulerAddress:               "sched:12345",
+		QuerierID:                      "test",
+		QueryFrontendGRPCClientConfig:  conf,
+		QuerySchedulerGRPCClientConfig: grpcclient.Config{MaxSendMsgSize: 5 * 1024}, // schedulerProcessor should ignore this.
+		MaxConcurrentRequests:          5,
+	}, nil, nil, nil)
+
+	assert.Equal(t, 1*1024*1024, sp.maxMessageSize)
+	assert.Equal(t, conf, sp.grpcConfig)
+}
+
 func prepareSchedulerProcessor() (*schedulerProcessor, *querierLoopClientMock, *requestHandlerMock) {
 	var querierLoopCtx context.Context