feat(router): support for isolation modes using limiters (#3379)

.github/workflows/tests.yaml

@@ -18,7 +18,7 @@ jobs:
         uses: actions/checkout@v3
       - uses: actions/setup-go@v4
         with:
-          go-version: '~1.20.3'
+          go-version: '~1.20.4'
           check-latest: true
       - run: go version
       - run: go mod download # Not required, used to segregate module download vs test times
@@ -75,7 +75,7 @@ jobs:
       - name: Setup Go
         uses: actions/setup-go@v4
         with:
-          go-version: '~1.20.3'
+          go-version: '~1.20.4'
           check-latest: true
       - run: go version
       - run: go mod download # Not required, used to segregate module download vs test times
@@ -107,7 +107,7 @@ jobs:
       - uses: actions/checkout@v3
       - uses: actions/setup-go@v4
         with:
-          go-version: '~1.20.3'
+          go-version: '~1.20.4'
           check-latest: true
 
       - run: go version
@@ -156,7 +156,7 @@ jobs:
       - name: Setup Go
         uses: actions/setup-go@v4
         with:
-          go-version: '~1.20.3'
+          go-version: '~1.20.4'
           check-latest: true
       - name: Download coverage reports
         uses: actions/download-artifact@v3

.github/workflows/verify.yml

@@ -16,7 +16,7 @@ jobs:
       - uses: actions/setup-go@v4
         with:
           check-latest: true
-          go-version: '~1.20.3'
+          go-version: '~1.20.4'
       - run: go version
 
       - run: go mod tidy
@@ -55,7 +55,7 @@ jobs:
       - uses: actions/checkout@v3
       - uses: actions/setup-go@v4
         with:
-          go-version: '~1.20.3'
+          go-version: '~1.20.4'
           check-latest: true
       - name: golangci-lint
         uses: golangci/golangci-lint-action@v3

Makefile

@@ -79,7 +79,7 @@ install-tools:
 	go install mvdan.cc/gofumpt@latest
 	go install google.golang.org/protobuf/cmd/protoc-gen-go@v1.28.1
 	go install google.golang.org/grpc/cmd/protoc-gen-go-grpc@v1.2.0
-	go install gotest.tools/gotestsum@v1.8.2
+	go install gotest.tools/gotestsum@v1.10.0
 
 .PHONY: lint
 lint: fmt ## Run linters on all go files

admin/admin.go

@@ -45,8 +45,8 @@ import (
 	"github.com/spf13/viper"
 
 	"github.com/rudderlabs/rudder-go-kit/config"
+	kithttputil "github.com/rudderlabs/rudder-go-kit/httputil"
 	"github.com/rudderlabs/rudder-go-kit/logger"
-	"github.com/rudderlabs/rudder-server/utils/httputil"
 	"github.com/rudderlabs/rudder-server/utils/misc"
 )
 
@@ -170,5 +170,5 @@ func StartServer(ctx context.Context) error {
 
 	srv := &http.Server{Handler: srvMux, ReadHeaderTimeout: 3 * time.Second}
 
-	return httputil.Serve(ctx, srv, l, time.Second)
+	return kithttputil.Serve(ctx, srv, l, time.Second)
 }

app/apphandlers/embeddedAppHandler.go

@@ -227,6 +227,7 @@ func (a *embeddedApp) StartRudderCore(ctx context.Context, options *app.Options)
 		return fmt.Errorf("failed to create rt throttler factory: %w", err)
 	}
 	rtFactory := &router.Factory{
+		Logger:           logger.NewLogger().Child("router"),
 		Reporting:        reportingI,
 		Multitenant:      multitenantStats,
 		BackendConfig:    backendconfig.DefaultBackendConfig,
@@ -249,7 +250,7 @@ func (a *embeddedApp) StartRudderCore(ctx context.Context, options *app.Options)
 		Debugger:         destinationHandle,
 		AdaptiveLimit:    adaptiveLimit,
 	}
-	rt := routerManager.New(rtFactory, brtFactory, backendconfig.DefaultBackendConfig)
+	rt := routerManager.New(rtFactory, brtFactory, backendconfig.DefaultBackendConfig, logger.NewLogger())
 
 	dm := cluster.Dynamic{
 		Provider:        modeProvider,

app/apphandlers/processorAppHandler.go

@@ -13,14 +13,14 @@ import (
 	"github.com/rudderlabs/rudder-server/internal/pulsar"
 	"github.com/rudderlabs/rudder-server/router/throttler"
 	schema_forwarder "github.com/rudderlabs/rudder-server/schema-forwarder"
-	"github.com/rudderlabs/rudder-server/utils/httputil"
 	"github.com/rudderlabs/rudder-server/utils/payload"
 	"github.com/rudderlabs/rudder-server/utils/types/deployment"
 
 	"golang.org/x/sync/errgroup"
 
 	"github.com/bugsnag/bugsnag-go/v2"
 
+	kithttputil "github.com/rudderlabs/rudder-go-kit/httputil"
 	"github.com/rudderlabs/rudder-go-kit/stats"
 	"github.com/rudderlabs/rudder-server/app"
 	"github.com/rudderlabs/rudder-server/app/cluster"
@@ -235,6 +235,7 @@ func (a *processorApp) StartRudderCore(ctx context.Context, options *app.Options
 		return fmt.Errorf("failed to create throttler factory: %w", err)
 	}
 	rtFactory := &router.Factory{
+		Logger:           logger.NewLogger().Child("router"),
 		Reporting:        reportingI,
 		Multitenant:      multitenantStats,
 		BackendConfig:    backendconfig.DefaultBackendConfig,
@@ -257,7 +258,7 @@ func (a *processorApp) StartRudderCore(ctx context.Context, options *app.Options
 		Debugger:         destinationHandle,
 		AdaptiveLimit:    adaptiveLimit,
 	}
-	rt := routerManager.New(rtFactory, brtFactory, backendconfig.DefaultBackendConfig)
+	rt := routerManager.New(rtFactory, brtFactory, backendconfig.DefaultBackendConfig, logger.NewLogger())
 
 	dm := cluster.Dynamic{
 		Provider:         modeProvider,
@@ -314,5 +315,5 @@ func (a *processorApp) startHealthWebHandler(ctx context.Context, db *jobsdb.Han
 		MaxHeaderBytes:    a.config.http.MaxHeaderBytes,
 	}
 
-	return httputil.ListenAndServe(ctx, srv)
+	return kithttputil.ListenAndServe(ctx, srv)
 }

app/cluster/integration_test.go

@@ -167,7 +167,6 @@ func initJobsDB() {
 	jobsdb.Init()
 	jobsdb.Init2()
 	archiver.Init()
-	router.Init()
 	Init()
 }
 
@@ -223,6 +222,7 @@ func TestDynamicClusterManager(t *testing.T) {
 
 	tDb := &jobsdb.MultiTenantHandleT{HandleT: rtDB}
 	rtFactory := &router.Factory{
+		Logger:           logger.NOP,
 		Reporting:        &reporting.NOOP{},
 		Multitenant:      mockMTI,
 		BackendConfig:    mockBackendConfig,
@@ -240,7 +240,7 @@ func TestDynamicClusterManager(t *testing.T) {
 		TransientSources: transientsource.NewEmptyService(),
 		RsourcesService:  mockRsourcesService,
 	}
-	router := routermanager.New(rtFactory, brtFactory, mockBackendConfig)
+	router := routermanager.New(rtFactory, brtFactory, mockBackendConfig, logger.NewLogger())
 
 	mockBackendConfig.EXPECT().Subscribe(gomock.Any(), gomock.Any()).DoAndReturn(func(
 		ctx context.Context, topic backendConfig.Topic,

backend-config/namespace_config.go

@@ -13,9 +13,9 @@ import (
 	jsoniter "github.com/json-iterator/go"
 
 	"github.com/rudderlabs/rudder-go-kit/config"
+	kithttputil "github.com/rudderlabs/rudder-go-kit/httputil"
 	"github.com/rudderlabs/rudder-go-kit/logger"
 	"github.com/rudderlabs/rudder-server/services/controlplane/identity"
-	"github.com/rudderlabs/rudder-server/utils/httputil"
 	"github.com/rudderlabs/rudder-server/utils/types"
 )
 
@@ -183,7 +183,7 @@ func (nc *namespaceConfig) makeHTTPRequest(req *http.Request) ([]byte, error) {
 		return nil, err
 	}
 
-	defer func() { httputil.CloseResponse(resp) }()
+	defer func() { kithttputil.CloseResponse(resp) }()
 	respBody, err := io.ReadAll(resp.Body)
 	if err != nil {
 		return nil, err

backend-config/single_workspace.go

@@ -13,8 +13,8 @@ import (
 	"github.com/cenkalti/backoff"
 
 	"github.com/rudderlabs/rudder-go-kit/config"
+	kithttputil "github.com/rudderlabs/rudder-go-kit/httputil"
 	"github.com/rudderlabs/rudder-server/services/controlplane/identity"
-	"github.com/rudderlabs/rudder-server/utils/httputil"
 	"github.com/rudderlabs/rudder-server/utils/types"
 )
 
@@ -151,7 +151,7 @@ func (wc *singleWorkspaceConfig) makeHTTPRequest(ctx context.Context, url string
 		return nil, err
 	}
 
-	defer func() { httputil.CloseResponse(resp) }()
+	defer func() { kithttputil.CloseResponse(resp) }()
 	respBody, err := io.ReadAll(resp.Body)
 	if err != nil {
 		return nil, err

gateway/gateway.go

@@ -31,6 +31,7 @@ import (
 
 	"github.com/rudderlabs/rudder-go-kit/chiware"
 	"github.com/rudderlabs/rudder-go-kit/config"
+	kithttputil "github.com/rudderlabs/rudder-go-kit/httputil"
 	"github.com/rudderlabs/rudder-go-kit/logger"
 	"github.com/rudderlabs/rudder-go-kit/stats"
 	"github.com/rudderlabs/rudder-server/app"
@@ -47,7 +48,6 @@ import (
 	"github.com/rudderlabs/rudder-server/services/diagnostics"
 	"github.com/rudderlabs/rudder-server/services/rsources"
 	rsources_http "github.com/rudderlabs/rudder-server/services/rsources/http"
-	rs_httputil "github.com/rudderlabs/rudder-server/utils/httputil"
 	"github.com/rudderlabs/rudder-server/utils/misc"
 	"github.com/rudderlabs/rudder-server/utils/types"
 )
@@ -1384,7 +1384,7 @@ func (gateway *HandleT) StartWebHandler(ctx context.Context) error {
 		MaxHeaderBytes:    maxHeaderBytes,
 	}
 
-	return rs_httputil.ListenAndServe(ctx, gateway.httpWebServer)
+	return kithttputil.ListenAndServe(ctx, gateway.httpWebServer)
 }
 
 // StartAdminHandler for Admin Operations
@@ -1404,7 +1404,7 @@ func (gateway *HandleT) StartAdminHandler(ctx context.Context) error {
 		ReadHeaderTimeout: ReadHeaderTimeout,
 	}
 
-	return rs_httputil.ListenAndServe(ctx, srv)
+	return kithttputil.ListenAndServe(ctx, srv)
 }
 
 // Gets the config from config backend and extracts enabled writekeys

gateway/integration_test.go

@@ -22,7 +22,7 @@ import (
 	"github.com/ory/dockertest/v3"
 	"github.com/stretchr/testify/require"
 
-	kitHelper "github.com/rudderlabs/rudder-go-kit/testhelper"
+	kithelper "github.com/rudderlabs/rudder-go-kit/testhelper"
 	"github.com/rudderlabs/rudder-go-kit/testhelper/docker/resource"
 	"github.com/rudderlabs/rudder-go-kit/testhelper/rand"
 	"github.com/rudderlabs/rudder-server/app"
@@ -117,11 +117,11 @@ func testGatewayByAppType(t *testing.T, appType string) {
 	t.Logf("BackendConfig server listening on: %s", backendConfigSrv.URL)
 	t.Cleanup(backendConfigSrv.Close)
 
-	httpPort, err := kitHelper.GetFreePort()
+	httpPort, err := kithelper.GetFreePort()
 	require.NoError(t, err)
-	httpAdminPort, err := kitHelper.GetFreePort()
+	httpAdminPort, err := kithelper.GetFreePort()
 	require.NoError(t, err)
-	debugPort, err := kitHelper.GetFreePort()
+	debugPort, err := kithelper.GetFreePort()
 	require.NoError(t, err)
 
 	rudderTmpDir, err := os.MkdirTemp("", "rudder_server_*_test")