Merge branch 'master' into chore.cleanup_router_oauthV2

Author: ItsSudip

jobsdb/integration_test.go

@@ -584,7 +584,7 @@ func TestJobsDB(t *testing.T) {
 		}
 		prefix := strings.ToLower(rand.String(5))
 		c.Set("JobsDB.jobDoneMigrateThreshold", 0.7)
-		c.Set("JobsDB.jobMinRowsMigrateThreshold", 0.6)
+		c.Set("JobsDB.jobMinRowsLeftMigrateThreshold", 0.41)
 		err := jobDB.Setup(ReadWrite, true, prefix)
 		require.NoError(t, err)
 		defer jobDB.TearDown()
@@ -622,7 +622,7 @@ func TestJobsDB(t *testing.T) {
 		})
 		require.NoError(t, err)
 
-		// process some jobs
+		// process first 15 jobs
 		for _, job := range jobsResult.Jobs[:15] {
 			status := JobStatusT{
 				JobID:         job.JobID,
@@ -642,20 +642,36 @@ func TestJobsDB(t *testing.T) {
 
 		dsList = getDSList()
 		require.Lenf(t, dsList, 5, "dsList length is not 5, got %+v", dsList)
-		require.Equal(t, prefix+"_jobs_1_1", dsList[0].JobTable)
-		require.Equal(t, prefix+"_jobs_2", dsList[1].JobTable)
-		require.Equal(t, prefix+"_jobs_3", dsList[2].JobTable)
-		require.Equal(t, prefix+"_jobs_4", dsList[3].JobTable)
-		require.Equal(t, prefix+"_jobs_5", dsList[4].JobTable)
+		require.Equal(t, prefix+"_jobs_1_1", dsList[0].JobTable) // 5 jobs
+		require.Equal(t, prefix+"_jobs_2", dsList[1].JobTable)   // 20 jobs
+		require.Equal(t, prefix+"_jobs_3", dsList[2].JobTable)   // 20 jobs
+		require.Equal(t, prefix+"_jobs_4", dsList[3].JobTable)   // 20 jobs
+		require.Equal(t, prefix+"_jobs_5", dsList[4].JobTable)   // 20 jobs
+
+		// process 1 more job
+		for _, job := range jobsResult.Jobs[15:16] {
+			status := JobStatusT{
+				JobID:         job.JobID,
+				JobState:      "succeeded",
+				AttemptNum:    1,
+				ExecTime:      time.Now(),
+				RetryTime:     time.Now(),
+				ErrorCode:     "202",
+				ErrorResponse: []byte(`{"success":"OK"}`),
+				Parameters:    []byte(`{}`),
+			}
+			err := jobDB.UpdateJobStatus(context.Background(), []*JobStatusT{&status}, []string{customVal}, []ParameterFilterT{})
+			require.NoError(t, err)
+		}
 
-		trigger() // jobs_1_1 will remain as is even though it is now a small table (5 < 10*0.6)
+		trigger() // jobs_1_1 will remain as is even though it is now a small table 4/10 = 0.4 < 0.41
 		dsList = getDSList()
 		require.Lenf(t, dsList, 5, "dsList length is not 5, got %+v", dsList)
-		require.Equal(t, prefix+"_jobs_1_1", dsList[0].JobTable)
-		require.Equal(t, prefix+"_jobs_2", dsList[1].JobTable)
-		require.Equal(t, prefix+"_jobs_3", dsList[2].JobTable)
-		require.Equal(t, prefix+"_jobs_4", dsList[3].JobTable)
-		require.Equal(t, prefix+"_jobs_5", dsList[4].JobTable)
+		require.Equal(t, prefix+"_jobs_1_1", dsList[0].JobTable) // 4 jobs
+		require.Equal(t, prefix+"_jobs_2", dsList[1].JobTable)   // 20 jobs
+		require.Equal(t, prefix+"_jobs_3", dsList[2].JobTable)   // 20 jobs
+		require.Equal(t, prefix+"_jobs_4", dsList[3].JobTable)   // 20 jobs
+		require.Equal(t, prefix+"_jobs_5", dsList[4].JobTable)   // 20 jobs
 
 		// process some jobs
 		jobsResult, err = jobDB.GetUnprocessed(context.Background(), GetQueryParams{
@@ -664,7 +680,7 @@ func TestJobsDB(t *testing.T) {
 			ParameterFilters: []ParameterFilterT{},
 		})
 		require.NoError(t, err)
-		for _, job := range jobsResult.Jobs[5:20] {
+		for _, job := range jobsResult.Jobs[4:20] { // process 16 jobs from jobs_2
 			status := JobStatusT{
 				JobID:         job.JobID,
 				JobState:      Succeeded.State,
@@ -682,10 +698,10 @@ func TestJobsDB(t *testing.T) {
 		trigger() // both jobs_1_1 and jobs_2 would be migrated to jobs_2_1
 		dsList = getDSList()
 		require.Lenf(t, dsList, 4, "dsList length is not 4, got %+v", dsList)
-		require.Equal(t, prefix+"_jobs_2_1", dsList[0].JobTable)
-		require.Equal(t, prefix+"_jobs_3", dsList[1].JobTable)
-		require.Equal(t, prefix+"_jobs_4", dsList[2].JobTable)
-		require.Equal(t, prefix+"_jobs_5", dsList[3].JobTable)
+		require.Equal(t, prefix+"_jobs_2_1", dsList[0].JobTable) // 8 jobs
+		require.Equal(t, prefix+"_jobs_3", dsList[1].JobTable)   // 20 jobs
+		require.Equal(t, prefix+"_jobs_4", dsList[2].JobTable)   // 20 jobs
+		require.Equal(t, prefix+"_jobs_5", dsList[3].JobTable)   // 20 jobs
 	})
 
 	t.Run(`migrates only moves non-terminal jobs to a new DS`, func(t *testing.T) {

jobsdb/jobsdb.go

@@ -527,8 +527,7 @@ type Handle struct {
 			vacuumFullStatusTableThreshold             config.ValueLoader[int64]
 			vacuumAnalyzeStatusTableThreshold          config.ValueLoader[int64]
 			jobDoneMigrateThres, jobStatusMigrateThres config.ValueLoader[float64]
-			jobMinRowsMigrateThres                     config.ValueLoader[float64]
-			jobMinRowsLeftMigrateThres                 config.ValueLoader[float64]
+			jobMinRowsLeftMigrateThreshold             config.ValueLoader[float64]
 			migrateDSLoopSleepDuration                 config.ValueLoader[time.Duration]
 			migrateDSTimeout                           config.ValueLoader[time.Duration]
 		}
@@ -933,14 +932,10 @@ func (jd *Handle) loadConfig() {
 	jd.conf.migration.jobDoneMigrateThres = jd.config.GetReloadableFloat64Var(0.8, jd.configKeys("jobDoneMigrateThreshold")...)
 	// jobStatusMigrateThres: A DS is migrated if the job_status exceeds this (* no_of_jobs)
 	jd.conf.migration.jobStatusMigrateThres = jd.config.GetReloadableFloat64Var(3, jd.configKeys("jobStatusMigrateThreshold")...)
-	// jobMinRowsMigrateThres: A DS with a low number of total rows should be eligible for migration if the number of total rows are
-	// less than jobMinRowsMigrateThres percent of maxDSSize (e.g. if jobMinRowsMigrateThres is 0.2
-	// then DSs that have less than 20% of maxDSSize total rows are eligible for migration)
-	jd.conf.migration.jobMinRowsMigrateThres = jd.config.GetReloadableFloat64Var(0.2, jd.configKeys("jobMinRowsMigrateThreshold")...)
-	// jobMinRowsMigrateThres: A DS with a low number of pending rows should be eligible for migration if the number of pending rows are
-	// less than jobMinRowsLeftMigrateThres percent of maxDSSize (e.g. if jobMinRowsLeftMigrateThres is 0.5
+	// jobMinRowsLeftMigrateThreshold: A DS with a low number of pending rows should be eligible for migration if the number of pending rows are
+	// less than jobMinRowsLeftMigrateThreshold percent of maxDSSize (e.g. if jobMinRowsLeftMigrateThreshold is 0.5
 	// then DSs that have less than 50% of maxDSSize pending rows are eligible for migration)
-	jd.conf.migration.jobMinRowsLeftMigrateThres = jd.config.GetReloadableFloat64Var(0.4, jd.configKeys("jobMinRowsLeftMigrateThres")...)
+	jd.conf.migration.jobMinRowsLeftMigrateThreshold = jd.config.GetReloadableFloat64Var(0.4, jd.configKeys("jobMinRowsLeftMigrateThreshold")...)
 	// maxMigrateOnce: Maximum number of DSs that are migrated together into one destination
 	jd.conf.migration.maxMigrateOnce = jd.config.GetReloadableIntVar(10, 1, jd.configKeys("maxMigrateOnce")...)
 	// maxMigrateDSProbe: Maximum number of DSs that are checked from left to right if they are eligible for migration

jobsdb/migration.go

@@ -657,18 +657,10 @@ func (jd *Handle) checkIfMigrateDS(ds dataSetT) (
 		return true, false, recordsLeft, nil
 	}
 
-	isSmallDS := float64(totalCount) < jd.conf.migration.jobMinRowsMigrateThres.Load()*float64(jd.conf.MaxDSSize.Load())
+	needsPair = float64(recordsLeft) < jd.conf.migration.jobMinRowsLeftMigrateThreshold.Load()*float64(jd.conf.MaxDSSize.Load())
 
-	if float64(delCount)/float64(totalCount) > jd.conf.migration.jobDoneMigrateThres.Load() {
-		return true, isSmallDS, recordsLeft, nil
-	}
-
-	if isSmallDS {
-		return true, true, recordsLeft, nil
-	}
-
-	if float64(recordsLeft) < jd.conf.migration.jobMinRowsLeftMigrateThres.Load()*float64(jd.conf.MaxDSSize.Load()) {
-		return true, true, recordsLeft, nil
+	if needsPair || float64(delCount)/float64(totalCount) >= jd.conf.migration.jobDoneMigrateThres.Load() {
+		return true, needsPair, recordsLeft, nil
 	}
 
 	return false, false, recordsLeft, nil

jobsdb/migration_test.go

@@ -106,7 +106,7 @@ func TestMigration(t *testing.T) {
 		// should have enough statuses for a cleanup to be triggered
 		// all non-terminal
 		require.NoError(t, jobDB.Store(context.Background(), jobs[20:30]))
-		for i := 0; i < 10; i++ {
+		for range 10 {
 			require.NoError(
 				t,
 				jobDB.UpdateJobStatus(
@@ -230,7 +230,7 @@ func TestMigration(t *testing.T) {
 		c.Set("JobsDB."+tablePrefix+"."+"maxDSRetention", "1ms")
 
 		// 3 datasets with 10 jobs each, 1 dataset with 0 jobs
-		for i := 0; i < 3; i++ {
+		for i := range 3 {
 			require.NoError(t, jobDB.Store(context.Background(), genJobs(defaultWorkspaceID, "test", 10, 1)))
 			triggerAddNewDS <- time.Now() // trigger addNewDSLoop to run
 			triggerAddNewDS <- time.Now() // waits for last loop to finish
@@ -239,10 +239,10 @@ func TestMigration(t *testing.T) {
 
 		// 1st ds 5 statuses each
 		var jobs []*JobT
-		for i := 0; i < 10; i++ {
+		for i := range 10 {
 			jobs = append(jobs, &JobT{JobID: int64(i + 1)})
 		}
-		for i := 0; i < 5; i++ {
+		for range 5 {
 			require.NoError(t, jobDB.UpdateJobStatus(context.Background(), genJobStatuses(jobs, "failed"), []string{"test"}, []ParameterFilterT{}))
 		}
 		var count int
@@ -251,10 +251,10 @@ func TestMigration(t *testing.T) {
 
 		// 2nd ds 10 statuses each
 		jobs = nil
-		for i := 0; i < 10; i++ {
+		for i := range 10 {
 			jobs = append(jobs, &JobT{JobID: int64(i + 11)})
 		}
-		for i := 0; i < 10; i++ {
+		for range 10 {
 			require.NoError(t, jobDB.UpdateJobStatus(context.Background(), genJobStatuses(jobs, "failed"), []string{"test"}, []ParameterFilterT{}))
 		}
 		require.NoError(t, jobDB.dbHandle.QueryRow(fmt.Sprintf(`SELECT COUNT(*) FROM %[1]s_job_status_2 `, tablePrefix)).Scan(&count))
@@ -317,10 +317,10 @@ func TestMigration(t *testing.T) {
 		require.Less(t, newTableSizes[fmt.Sprintf("%s_job_status_2", tablePrefix)], originalTableSizes[fmt.Sprintf("%s_job_status_2", tablePrefix)])
 
 		// after adding some statuses to the 1st DS its table size shouldn't increase
-		for i := 0; i < 10; i++ {
+		for i := range 10 {
 			jobs = append(jobs, &JobT{JobID: int64(i + 1)})
 		}
-		for i := 0; i < 4; i++ {
+		for range 4 {
 			require.NoError(t, jobDB.UpdateJobStatus(context.Background(), genJobStatuses(jobs, "failed"), []string{"test"}, []ParameterFilterT{}))
 		}
 
@@ -332,6 +332,7 @@ func TestMigration(t *testing.T) {
 		config.Reset()
 		c := config.New()
 		c.Set("JobsDB.maxDSSize", 1)
+		c.Set("JobsDB.jobMinRowsLeftMigrateThres", 0.2)
 
 		_ = startPostgres(t)
 
@@ -418,13 +419,6 @@ func TestMigration(t *testing.T) {
 		// add some more jobs to the new DS
 		require.NoError(t, jobDB.Store(context.Background(), jobs[10:20]))
 
-		// triggerMigrateDS <- time.Now()
-		// triggerMigrateDS <- time.Now()
-		// var payloadType_1_1 string
-		// err = jobDB.dbHandle.QueryRowContext(context.Background(), fmt.Sprintf(`select data_type from information_schema.columns where table_name='%s' and column_name='event_payload';`, tablePrefix+"_jobs_1_1")).Scan(&payloadType_1_1)
-		// require.NoError(t, err)
-		// require.EqualValues(t, "bytea", payloadType_1_1)
-
 		triggerAddNewDS <- time.Now() // trigger addNewDSLoop to run
 		triggerAddNewDS <- time.Now() // Second time, waits for the first loop to finish
 		require.EqualValues(t, 3, jobDB.GetMaxDSIndex())
@@ -437,7 +431,7 @@ func TestMigration(t *testing.T) {
 		// should have enough statuses for a cleanup to be triggered
 		// all non-terminal
 		require.NoError(t, jobDB.Store(context.Background(), jobs[20:30]))
-		for i := 0; i < 10; i++ {
+		for range 10 {
 			require.NoError(
 				t,
 				jobDB.UpdateJobStatus(

processor/internal/transformer/destination_transformer/destination_transformer.go

@@ -82,7 +82,7 @@ func New(conf *config.Config, log logger.Logger, stat stats.Stats, opts ...Opt)
 	handle.loggedEventsMu = sync.Mutex{}
 	handle.loggedFileName = generateLogFileName()
 
-	handle.config.compactionEnabled = conf.GetReloadableBoolVar(true, "Processor.DestinationTransformer.compactionEnabled", "Transformer.compactionEnabled")
+	handle.config.compactionEnabled = conf.GetReloadableBoolVar(false, "Processor.DestinationTransformer.compactionEnabled", "Transformer.compactionEnabled")
 
 	for _, opt := range opts {
 		opt(handle)

router/transformer/transformer.go

@@ -576,7 +576,7 @@ func (trans *handle) setup(destinationTimeout, transformTimeout time.Duration, c
 	trans.proxyClientOAuth = oauthv2httpclient.NewOAuthHttpClient(&http.Client{Transport: trans.tr, Timeout: trans.destinationTimeout + trans.transformTimeout}, common.RudderFlowDelivery, cache, backendConfig, GetAuthErrorCategoryFromTransformProxyResponse, proxyClientOptionalArgs)
 	trans.stats = stats.Default
 	trans.transformRequestTimerStat = stats.Default.NewStat("router.transformer_request_time", stats.TimerType)
-	trans.compactionEnabled = config.GetReloadableBoolVar(true, "Router.DestinationTransformer.compactionEnabled", "Transformer.compactionEnabled")
+	trans.compactionEnabled = config.GetReloadableBoolVar(false, "Router.DestinationTransformer.compactionEnabled", "Transformer.compactionEnabled")
 	if featuresService != nil {
 		go func() {
 			<-featuresService.Wait()