chore: master pull

rudderlabs · Sep 19, 2023 · 6ef6694 · 6ef6694
2 parents 3e6feac + 4bf9ce7
commit 6ef6694
Show file tree

Hide file tree

Showing 119 changed files with 3,724 additions and 1,249 deletions.
diff --git a/.golangci.yml b/.golangci.yml
@@ -59,6 +59,10 @@ issues:
         - errcheck
         - unparam
 
+    - path: 'cmd/rudder-cli/status/status.go'
+      linters:
+        - bodyclose
+
 linters-settings:
   depguard:
     rules:

diff --git a/Dockerfile b/Dockerfile
@@ -27,7 +27,7 @@ RUN BUILD_DATE=$(date "+%F,%T") \
     make build
 
 RUN go build -o devtool ./cmd/devtool/
-
+RUN go build -o rudder-cli ./cmd/rudder-cli/
 
 FROM alpine:${ALPINE_VERSION}
 
@@ -39,10 +39,10 @@ COPY --from=builder rudder-server/rudder-server .
 COPY --from=builder rudder-server/build/wait-for-go/wait-for-go .
 COPY --from=builder rudder-server/build/regulation-worker .
 COPY --from=builder rudder-server/devtool .
+COPY --from=builder rudder-server/rudder-cli /usr/bin/rudder-cli
 
 COPY build/docker-entrypoint.sh /
 COPY build/wait-for /
-COPY ./rudder-cli/rudder-cli.linux.x86_64 /usr/bin/rudder-cli
 COPY scripts/generate-event /scripts/generate-event
 COPY scripts/batch.json /scripts/batch.json
 

diff --git a/app/apphandlers/embeddedAppHandler.go b/app/apphandlers/embeddedAppHandler.go
@@ -46,24 +46,19 @@ type embeddedApp struct {
 	log            logger.Logger
 	config         struct {
 		enableReplay       bool
-		processorDSLimit   int
-		routerDSLimit      int
-		batchRouterDSLimit int
-		gatewayDSLimit     int
+		processorDSLimit   misc.ValueLoader[int]
+		routerDSLimit      misc.ValueLoader[int]
+		batchRouterDSLimit misc.ValueLoader[int]
+		gatewayDSLimit     misc.ValueLoader[int]
 	}
 }
 
-// nolint:staticcheck // SA1019: config Register reloadable functions are deprecated
-func (a *embeddedApp) loadConfiguration() {
-	a.config.enableReplay = config.GetBoolVar(types.DefaultReplayEnabled, "Replay.enabled")
-	config.RegisterIntConfigVariable(0, &a.config.processorDSLimit, true, 1, "Processor.jobsDB.dsLimit", "JobsDB.dsLimit")
-	config.RegisterIntConfigVariable(0, &a.config.gatewayDSLimit, true, 1, "Gateway.jobsDB.dsLimit", "JobsDB.dsLimit")
-	config.RegisterIntConfigVariable(0, &a.config.routerDSLimit, true, 1, "Router.jobsDB.dsLimit", "JobsDB.dsLimit")
-	config.RegisterIntConfigVariable(0, &a.config.batchRouterDSLimit, true, 1, "BatchRouter.jobsDB.dsLimit", "JobsDB.dsLimit")
-}
-
 func (a *embeddedApp) Setup(options *app.Options) error {
-	a.loadConfiguration()
+	a.config.enableReplay = config.GetBoolVar(types.DefaultReplayEnabled, "Replay.enabled")
+	a.config.processorDSLimit = config.GetReloadableIntVar(0, 1, "Processor.jobsDB.dsLimit", "JobsDB.dsLimit")
+	a.config.gatewayDSLimit = config.GetReloadableIntVar(0, 1, "Gateway.jobsDB.dsLimit", "JobsDB.dsLimit")
+	a.config.routerDSLimit = config.GetReloadableIntVar(0, 1, "Router.jobsDB.dsLimit", "JobsDB.dsLimit")
+	a.config.batchRouterDSLimit = config.GetReloadableIntVar(0, 1, "BatchRouter.jobsDB.dsLimit", "JobsDB.dsLimit")
 
 	if err := db.HandleEmbeddedRecovery(options.NormalMode, options.DegradedMode, misc.AppStartTime, app.EMBEDDED); err != nil {
 		return err
@@ -153,7 +148,7 @@ func (a *embeddedApp) StartRudderCore(ctx context.Context, options *app.Options)
 		"gw",
 		jobsdb.WithClearDB(options.ClearDB),
 		jobsdb.WithPreBackupHandlers(prebackupHandlers),
-		jobsdb.WithDSLimit(&a.config.gatewayDSLimit),
+		jobsdb.WithDSLimit(a.config.gatewayDSLimit),
 		jobsdb.WithFileUploaderProvider(fileUploaderProvider),
 		jobsdb.WithSkipMaintenanceErr(config.GetBool("Gateway.jobsDB.skipMaintenanceError", true)),
 	)
@@ -162,7 +157,7 @@ func (a *embeddedApp) StartRudderCore(ctx context.Context, options *app.Options)
 		"rt",
 		jobsdb.WithClearDB(options.ClearDB),
 		jobsdb.WithPreBackupHandlers(prebackupHandlers),
-		jobsdb.WithDSLimit(&a.config.routerDSLimit),
+		jobsdb.WithDSLimit(a.config.routerDSLimit),
 		jobsdb.WithFileUploaderProvider(fileUploaderProvider),
 		jobsdb.WithSkipMaintenanceErr(config.GetBool("Router.jobsDB.skipMaintenanceError", false)),
 	)
@@ -171,7 +166,7 @@ func (a *embeddedApp) StartRudderCore(ctx context.Context, options *app.Options)
 		"batch_rt",
 		jobsdb.WithClearDB(options.ClearDB),
 		jobsdb.WithPreBackupHandlers(prebackupHandlers),
-		jobsdb.WithDSLimit(&a.config.batchRouterDSLimit),
+		jobsdb.WithDSLimit(a.config.batchRouterDSLimit),
 		jobsdb.WithFileUploaderProvider(fileUploaderProvider),
 		jobsdb.WithSkipMaintenanceErr(config.GetBool("BatchRouter.jobsDB.skipMaintenanceError", false)),
 	)
@@ -182,7 +177,7 @@ func (a *embeddedApp) StartRudderCore(ctx context.Context, options *app.Options)
 		"proc_error",
 		jobsdb.WithClearDB(options.ClearDB),
 		jobsdb.WithPreBackupHandlers(prebackupHandlers),
-		jobsdb.WithDSLimit(&a.config.processorDSLimit),
+		jobsdb.WithDSLimit(a.config.processorDSLimit),
 		jobsdb.WithFileUploaderProvider(fileUploaderProvider),
 		jobsdb.WithSkipMaintenanceErr(config.GetBool("Processor.jobsDB.skipMaintenanceError", false)),
 	)
@@ -200,15 +195,15 @@ func (a *embeddedApp) StartRudderCore(ctx context.Context, options *app.Options)
 	schemaDB := jobsdb.NewForReadWrite(
 		"esch",
 		jobsdb.WithClearDB(options.ClearDB),
-		jobsdb.WithDSLimit(&a.config.processorDSLimit),
+		jobsdb.WithDSLimit(a.config.processorDSLimit),
 		jobsdb.WithSkipMaintenanceErr(config.GetBool("Processor.jobsDB.skipMaintenanceError", false)),
 	)
 	defer schemaDB.Close()
 
 	archivalDB := jobsdb.NewForReadWrite(
 		"arc",
 		jobsdb.WithClearDB(options.ClearDB),
-		jobsdb.WithDSLimit(&a.config.processorDSLimit),
+		jobsdb.WithDSLimit(a.config.processorDSLimit),
 		jobsdb.WithSkipMaintenanceErr(config.GetBool("Processor.jobsDB.skipMaintenanceError", false)),
 		jobsdb.WithJobMaxAge(
 			func() time.Duration {

diff --git a/app/apphandlers/gatewayAppHandler.go b/app/apphandlers/gatewayAppHandler.go
@@ -30,17 +30,12 @@ type gatewayApp struct {
 	versionHandler func(w http.ResponseWriter, r *http.Request)
 	log            logger.Logger
 	config         struct {
-		gatewayDSLimit int
+		gatewayDSLimit misc.ValueLoader[int]
 	}
 }
 
-// nolint:staticcheck // SA1019: config Register reloadable functions are deprecated
-func (a *gatewayApp) loadConfiguration() {
-	config.RegisterIntConfigVariable(0, &a.config.gatewayDSLimit, true, 1, "Gateway.jobsDB.dsLimit", "JobsDB.dsLimit")
-}
-
 func (a *gatewayApp) Setup(options *app.Options) error {
-	a.loadConfiguration()
+	a.config.gatewayDSLimit = config.GetReloadableIntVar(0, 1, "Gateway.jobsDB.dsLimit", "JobsDB.dsLimit")
 	if err := db.HandleNullRecovery(options.NormalMode, options.DegradedMode, misc.AppStartTime, app.GATEWAY); err != nil {
 		return err
 	}
@@ -80,7 +75,7 @@ func (a *gatewayApp) StartRudderCore(ctx context.Context, options *app.Options)
 	gatewayDB := jobsdb.NewForWrite(
 		"gw",
 		jobsdb.WithClearDB(options.ClearDB),
-		jobsdb.WithDSLimit(&a.config.gatewayDSLimit),
+		jobsdb.WithDSLimit(a.config.gatewayDSLimit),
 		jobsdb.WithSkipMaintenanceErr(config.GetBool("Gateway.jobsDB.skipMaintenanceError", true)),
 		jobsdb.WithFileUploaderProvider(fileUploaderProvider),
 	)

diff --git a/app/apphandlers/processorAppHandler.go b/app/apphandlers/processorAppHandler.go
@@ -49,10 +49,10 @@ type processorApp struct {
 	versionHandler func(w http.ResponseWriter, r *http.Request)
 	log            logger.Logger
 	config         struct {
-		processorDSLimit   int
-		routerDSLimit      int
-		batchRouterDSLimit int
-		gatewayDSLimit     int
+		processorDSLimit   misc.ValueLoader[int]
+		routerDSLimit      misc.ValueLoader[int]
+		batchRouterDSLimit misc.ValueLoader[int]
+		gatewayDSLimit     misc.ValueLoader[int]
 		http               struct {
 			ReadTimeout       time.Duration
 			ReadHeaderTimeout time.Duration
@@ -64,22 +64,17 @@ type processorApp struct {
 	}
 }
 
-// nolint:staticcheck // SA1019: config Register reloadable functions are deprecated
-func (a *processorApp) loadConfiguration() {
+func (a *processorApp) Setup(options *app.Options) error {
 	a.config.http.ReadTimeout = config.GetDurationVar(0, time.Second, []string{"ReadTimeout", "ReadTimeOutInSec"}...)
 	a.config.http.ReadHeaderTimeout = config.GetDurationVar(0, time.Second, []string{"ReadHeaderTimeout", "ReadHeaderTimeoutInSec"}...)
 	a.config.http.WriteTimeout = config.GetDurationVar(10, time.Second, []string{"WriteTimeout", "WriteTimeOutInSec"}...)
 	a.config.http.IdleTimeout = config.GetDurationVar(720, time.Second, []string{"IdleTimeout", "IdleTimeoutInSec"}...)
 	a.config.http.webPort = config.GetIntVar(8086, 1, "Processor.webPort")
 	a.config.http.MaxHeaderBytes = config.GetIntVar(524288, 1, "MaxHeaderBytes")
-	config.RegisterIntConfigVariable(0, &a.config.processorDSLimit, true, 1, "Processor.jobsDB.dsLimit", "JobsDB.dsLimit")
-	config.RegisterIntConfigVariable(0, &a.config.gatewayDSLimit, true, 1, "Gateway.jobsDB.dsLimit", "JobsDB.dsLimit")
-	config.RegisterIntConfigVariable(0, &a.config.routerDSLimit, true, 1, "Router.jobsDB.dsLimit", "JobsDB.dsLimit")
-	config.RegisterIntConfigVariable(0, &a.config.batchRouterDSLimit, true, 1, "BatchRouter.jobsDB.dsLimit", "JobsDB.dsLimit")
-}
-
-func (a *processorApp) Setup(options *app.Options) error {
-	a.loadConfiguration()
+	a.config.processorDSLimit = config.GetReloadableIntVar(0, 1, "Processor.jobsDB.dsLimit", "JobsDB.dsLimit")
+	a.config.gatewayDSLimit = config.GetReloadableIntVar(0, 1, "Gateway.jobsDB.dsLimit", "JobsDB.dsLimit")
+	a.config.routerDSLimit = config.GetReloadableIntVar(0, 1, "Router.jobsDB.dsLimit", "JobsDB.dsLimit")
+	a.config.batchRouterDSLimit = config.GetReloadableIntVar(0, 1, "BatchRouter.jobsDB.dsLimit", "JobsDB.dsLimit")
 	if err := db.HandleNullRecovery(options.NormalMode, options.DegradedMode, misc.AppStartTime, app.PROCESSOR); err != nil {
 		return err
 	}
@@ -148,7 +143,7 @@ func (a *processorApp) StartRudderCore(ctx context.Context, options *app.Options
 		"gw",
 		jobsdb.WithClearDB(options.ClearDB),
 		jobsdb.WithPreBackupHandlers(prebackupHandlers),
-		jobsdb.WithDSLimit(&a.config.gatewayDSLimit),
+		jobsdb.WithDSLimit(a.config.gatewayDSLimit),
 		jobsdb.WithFileUploaderProvider(fileUploaderProvider),
 		jobsdb.WithSkipMaintenanceErr(config.GetBool("Gateway.jobsDB.skipMaintenanceError", true)),
 	)
@@ -157,7 +152,7 @@ func (a *processorApp) StartRudderCore(ctx context.Context, options *app.Options
 		"rt",
 		jobsdb.WithClearDB(options.ClearDB),
 		jobsdb.WithPreBackupHandlers(prebackupHandlers),
-		jobsdb.WithDSLimit(&a.config.routerDSLimit),
+		jobsdb.WithDSLimit(a.config.routerDSLimit),
 		jobsdb.WithFileUploaderProvider(fileUploaderProvider),
 		jobsdb.WithSkipMaintenanceErr(config.GetBool("Router.jobsDB.skipMaintenanceError", false)),
 	)
@@ -166,7 +161,7 @@ func (a *processorApp) StartRudderCore(ctx context.Context, options *app.Options
 		"batch_rt",
 		jobsdb.WithClearDB(options.ClearDB),
 		jobsdb.WithPreBackupHandlers(prebackupHandlers),
-		jobsdb.WithDSLimit(&a.config.batchRouterDSLimit),
+		jobsdb.WithDSLimit(a.config.batchRouterDSLimit),
 		jobsdb.WithFileUploaderProvider(fileUploaderProvider),
 		jobsdb.WithSkipMaintenanceErr(config.GetBool("BatchRouter.jobsDB.skipMaintenanceError", false)),
 	)
@@ -175,7 +170,7 @@ func (a *processorApp) StartRudderCore(ctx context.Context, options *app.Options
 		"proc_error",
 		jobsdb.WithClearDB(options.ClearDB),
 		jobsdb.WithPreBackupHandlers(prebackupHandlers),
-		jobsdb.WithDSLimit(&a.config.processorDSLimit),
+		jobsdb.WithDSLimit(a.config.processorDSLimit),
 		jobsdb.WithFileUploaderProvider(fileUploaderProvider),
 		jobsdb.WithSkipMaintenanceErr(config.GetBool("Processor.jobsDB.skipMaintenanceError", false)),
 	)
@@ -192,15 +187,15 @@ func (a *processorApp) StartRudderCore(ctx context.Context, options *app.Options
 	schemaDB := jobsdb.NewForReadWrite(
 		"esch",
 		jobsdb.WithClearDB(options.ClearDB),
-		jobsdb.WithDSLimit(&a.config.processorDSLimit),
+		jobsdb.WithDSLimit(a.config.processorDSLimit),
 		jobsdb.WithFileUploaderProvider(fileUploaderProvider),
 	)
 	defer schemaDB.Close()
 
 	archivalDB := jobsdb.NewForReadWrite(
 		"arc",
 		jobsdb.WithClearDB(options.ClearDB),
-		jobsdb.WithDSLimit(&a.config.processorDSLimit),
+		jobsdb.WithDSLimit(a.config.processorDSLimit),
 		jobsdb.WithSkipMaintenanceErr(config.GetBool("Processor.jobsDB.skipMaintenanceError", false)),
 		jobsdb.WithJobMaxAge(
 			func() time.Duration {

diff --git a/app/cluster/integration_test.go b/app/cluster/integration_test.go
@@ -38,7 +38,6 @@ import (
 	mocksBackendConfig "github.com/rudderlabs/rudder-server/mocks/backend-config"
 	mocksTransformer "github.com/rudderlabs/rudder-server/mocks/processor/transformer"
 	"github.com/rudderlabs/rudder-server/processor"
-	"github.com/rudderlabs/rudder-server/processor/stash"
 	"github.com/rudderlabs/rudder-server/router"
 	"github.com/rudderlabs/rudder-server/router/batchrouter"
 	routermanager "github.com/rudderlabs/rudder-server/router/manager"
@@ -161,7 +160,6 @@ var (
 func initJobsDB() {
 	config.Reset()
 	logger.Reset()
-	stash.Init()
 	admin.Init()
 	Init()
 }

diff --git a/backend-config/backend-config.go b/backend-config/backend-config.go
@@ -23,6 +23,7 @@ import (
 	"github.com/rudderlabs/rudder-server/rruntime"
 	"github.com/rudderlabs/rudder-server/services/controlplane/identity"
 	"github.com/rudderlabs/rudder-server/services/diagnostics"
+	"github.com/rudderlabs/rudder-server/utils/misc"
 	"github.com/rudderlabs/rudder-server/utils/pubsub"
 	"github.com/rudderlabs/rudder-server/utils/sysUtils"
 	"github.com/rudderlabs/rudder-server/utils/types"
@@ -31,14 +32,13 @@ import (
 
 var (
 	// environment variables
-	configBackendURL                      string
-	cpRouterURL                           string
-	pollInterval, regulationsPollInterval time.Duration
-	configJSONPath                        string
-	configFromFile                        bool
-	maxRegulationsPerRequest              int
-	configEnvReplacementEnabled           bool
-	dbCacheEnabled                        bool
+	configBackendURL            string
+	cpRouterURL                 string
+	pollInterval                misc.ValueLoader[time.Duration]
+	configJSONPath              string
+	configFromFile              bool
+	configEnvReplacementEnabled bool
+	dbCacheEnabled              bool
 
 	LastSync           string
 	LastRegulationSync string
@@ -92,13 +92,10 @@ type backendConfigImpl struct {
 	cache             cache.Cache
 }
 
-// nolint:staticcheck // SA1019: config Register reloadable functions are deprecated
 func loadConfig() {
 	configBackendURL = config.GetString("CONFIG_BACKEND_URL", "https://api.rudderstack.com")
 	cpRouterURL = config.GetString("CP_ROUTER_URL", "https://cp-router.rudderlabs.com")
-	config.RegisterDurationConfigVariable(5, &pollInterval, true, time.Second, "BackendConfig.pollInterval", "BackendConfig.pollIntervalInS")
-	config.RegisterDurationConfigVariable(300, &regulationsPollInterval, true, time.Second, "BackendConfig.regulationsPollInterval", "BackendConfig.regulationsPollIntervalInS")
-	config.RegisterIntConfigVariable(1000, &maxRegulationsPerRequest, true, 1, "BackendConfig.maxRegulationsPerRequest")
+	pollInterval = config.GetReloadableDurationVar(5, time.Second, "BackendConfig.pollInterval", "BackendConfig.pollIntervalInS")
 	configJSONPath = config.GetStringVar("/etc/rudderstack/workspaceConfig.json", "BackendConfig.configJSONPath")
 	configFromFile = config.GetBoolVar(false, "BackendConfig.configFromFile")
 	configEnvReplacementEnabled = config.GetBoolVar(true, "BackendConfig.envReplacementEnabled")
@@ -246,7 +243,7 @@ func (bc *backendConfigImpl) pollConfigUpdate(ctx context.Context) {
 		select {
 		case <-ctx.Done():
 			return
-		case <-time.After(pollInterval):
+		case <-time.After(pollInterval.Load()):
 		}
 	}
 }
@@ -384,7 +381,7 @@ func (bc *backendConfigImpl) WaitForConfig(ctx context.Context) {
 		select {
 		case <-ctx.Done():
 			return
-		case <-time.After(pollInterval):
+		case <-time.After(pollInterval.Load()):
 		}
 	}
 }

diff --git a/backend-config/backend_config_test.go b/backend-config/backend_config_test.go
@@ -25,6 +25,7 @@ import (
 	adminpkg "github.com/rudderlabs/rudder-server/admin"
 	"github.com/rudderlabs/rudder-server/backend-config/internal/cache"
 	"github.com/rudderlabs/rudder-server/services/diagnostics"
+	"github.com/rudderlabs/rudder-server/utils/misc"
 	"github.com/rudderlabs/rudder-server/utils/pubsub"
 	"github.com/rudderlabs/rudder-server/utils/types/deployment"
 )
@@ -324,7 +325,7 @@ func TestWaitForConfig(t *testing.T) {
 		defer ctrl.Finish()
 
 		pkgLogger = logger.NOP
-		pollInterval = time.Millisecond
+		pollInterval = misc.SingleValueLoader(time.Millisecond)
 		bc := &backendConfigImpl{initialized: false}
 
 		var done int32