chore: no full vacuum for various reports tables

enterprise/reporting/error_reporting.go

@@ -21,6 +21,7 @@
 	"go.uber.org/atomic"
 	"golang.org/x/sync/errgroup"
 
+	"github.com/rudderlabs/rudder-go-kit/bytesize"
 	"github.com/rudderlabs/rudder-go-kit/config"
 	"github.com/rudderlabs/rudder-go-kit/logger"
 	"github.com/rudderlabs/rudder-go-kit/stats"
@@ -158,12 +159,12 @@
 	if !edr.config.GetBool("Reporting.errorReporting.syncer.enabled", true) {
 		return func() {}
 	}
 	if _, err := dbHandle.ExecContext(
 		context.Background(),
 		fmt.Sprintf("vacuum full analyze %s", pq.QuoteIdentifier(ErrorDetailReportsTable)),
 	); err != nil {
 		edr.log.Errorn("error full vacuuming", logger.NewStringField("table", ErrorDetailReportsTable), obskit.Error(err))
 		panic(err)
 	}
 
 	return func() {
@@ -362,21 +363,26 @@
 		// Aggregate
 		// Send in a separate go-routine
 		// Delete in a separate go-routine
 		var (
 			deletedRows               int
 			vacuumDeletedRowThreshold = edr.config.GetReloadableIntVar(
 				100000, 1,
 				"Reporting.errorReporting.vacuumThresholdDeletedRows",
 				"Reporting.vacuumThresholdDeletedRows",
 			)
 			lastVacuum     time.Time
 			vacuumInterval = edr.config.GetReloadableDurationVar(
 				15,
 				time.Minute,
 				"Reporting.errorReporting.vacuumInterval",
 				"Reporting.vacuumInterval",
 			)
+			vacuumThresholdBytes = config.GetReloadableInt64Var(
+				10*bytesize.GB, 1,
+				"Reporting.errorReporting.vacuumThresholdBytes",
+				"Reporting.vacuumThresholdBytes",
+			)
 		)
 		for {
 			if ctx.Err() != nil {
 				edr.log.Infof("stopping mainLoop for syncer %s : %s", c.Label, ctx.Err())
@@ -449,15 +455,25 @@
 				errorDetailReportsDeleteQueryTimer.Since(deleteReportsStart)
 				if err != nil {
 					edr.log.Errorf("[ Error Detail Reporting ]: Error deleting local reports from %s: %v", ErrorDetailReportsTable, err)
 				} else {
 					deletedRows += len(reports)
 				}
 				// vacuum error_reports_details table
+				var sizeEstimate int64
+				if err := dbHandle.QueryRowContext(
+					ctx,
+					`SELECT pg_table_size(oid) from pg_class where relname = $1`, ErrorDetailReportsTable,
+				).Scan(&sizeEstimate); err != nil {
+					edr.log.Errorn(
+						fmt.Sprintf(`Error getting %s table size estimate`, ErrorDetailReportsTable),
+						logger.NewErrorField(err),
+					)
+				}
 				if deletedRows >= vacuumDeletedRowThreshold.Load() ||
-					time.Since(lastVacuum) >= vacuumInterval.Load() {
+					(sizeEstimate >= vacuumThresholdBytes.Load() && time.Since(lastVacuum) >= vacuumInterval.Load()) {
 					if err := edr.vacuum(ctx, dbHandle, tags); err == nil {
 						deletedRows = 0
 						lastVacuum = time.Now()
 					}
 				}
 			}
@@ -476,28 +492,28 @@
 	}
 }
 
 func (edr *ErrorDetailReporter) vacuum(ctx context.Context, dbHandle *sql.DB, tags stats.Tags) error {
 	defer edr.stats.NewTaggedStat(StatReportingVacuumDuration, stats.TimerType, tags).RecordDuration()()
 	var query string
 	var full bool
 	if edr.vacuumFull.Load() {
 		query = fmt.Sprintf("vacuum full analyze %s", pq.QuoteIdentifier(ErrorDetailReportsTable))
 		full = true
 	} else {
 		query = fmt.Sprintf("vacuum analyze %s", pq.QuoteIdentifier(ErrorDetailReportsTable))
 	}
 	_, err := dbHandle.ExecContext(ctx, query)
 	if err != nil {
 		edr.log.Errorn(
 			"error vacuuming",
 			logger.NewStringField("table", ErrorDetailReportsTable),
 			obskit.Error(err),
 			logger.NewBoolField("full", full),
 		)
 		return err
 	}
 
 	return nil
 }
 
 func (edr *ErrorDetailReporter) getReports(ctx context.Context, currentMs int64, syncerKey string) ([]*types.EDReportsDB, int64) {

enterprise/reporting/flusher/flusher.go

@@ -15,6 +15,7 @@
 	"github.com/cenkalti/backoff"
 	"golang.org/x/sync/errgroup"
 
+	"github.com/rudderlabs/rudder-go-kit/bytesize"
 	"github.com/rudderlabs/rudder-go-kit/config"
 	"github.com/rudderlabs/rudder-go-kit/logger"
 	"github.com/rudderlabs/rudder-go-kit/stats"
@@ -43,6 +44,7 @@
 	aggressiveFlushEnabled              config.ValueLoader[bool]
 	lagThresholdForAggresiveFlushInMins config.ValueLoader[time.Duration]
 	vacuumThresholdDeletedRows          config.ValueLoader[int]
+	vacuumThresholdBytes                config.ValueLoader[int64]
 	deletedRows                         int
 	lastVacuum                          time.Time
 	vacuumFull                          config.ValueLoader[bool]
@@ -81,6 +83,7 @@
 	lagThresholdForAggresiveFlushInMins := conf.GetReloadableDurationVar(5, time.Minute, "Reporting.flusher.lagThresholdForAggresiveFlushInMins")
 	vacuumThresholdDeletedRows := conf.GetReloadableIntVar(100000, 1, "Reporting.flusher.vacuumThresholdDeletedRows")
 	vacuumInterval := conf.GetReloadableDurationVar(15, time.Minute, "Reporting.flusher.vacuumInterval", "Reporting.vacuumInterval")
+	vacuumThresholdBytes := conf.GetReloadableInt64Var(10*bytesize.GB, 1, "Reporting.flusher.vacuumThresholdBytes", "Reporting.vacuumThresholdBytes")
 
 	tr := &http.Transport{}
 	client := &http.Client{Transport: tr, Timeout: config.GetDuration("HttpClient.reporting.timeout", 60, time.Second)}
@@ -100,6 +103,7 @@
 		vacuumThresholdDeletedRows: vacuumThresholdDeletedRows,
 		vacuumFull:                 conf.GetReloadableBoolVar(false, "Reporting.flusher.vacuumFull", "Reporting.vacuumFull"),
 		vacuumInterval:             vacuumInterval,
+		vacuumThresholdBytes:       vacuumThresholdBytes,
 
 		table:                               table,
 		aggregator:                          aggregator,
@@ -116,8 +120,8 @@
 	if _, err := db.Exec(
 		fmt.Sprintf("vacuum full analyze %s", pq.QuoteIdentifier(table)),
 	); err != nil {
 		log.Errorn("error full vacuuming", logger.NewStringField("table", table), obskit.Error(err))
 		return nil, fmt.Errorf("error full vacuuming %s table %w", table, err)
 	}
 	return &f, nil
 }
@@ -336,26 +340,33 @@
 func (f *Flusher) vacuum(ctx context.Context) error {
 	var query string
 	var full bool
+	var sizeEstimate int64
+	if err := f.db.QueryRowContext(
+		ctx, `SELECT pg_table_size(oid) from pg_class where relname = $1`, f.table,
+	).Scan(&sizeEstimate); err != nil {
+		return fmt.Errorf("error getting table size %w", err)
+	}
 	if f.deletedRows >= f.vacuumThresholdDeletedRows.Load() ||
-		time.Since(f.lastVacuum) >= f.vacuumInterval.Load() {
+		(sizeEstimate >= f.vacuumThresholdBytes.Load() && time.Since(f.lastVacuum) >= f.vacuumInterval.Load()) {
 		vacuumStart := time.Now()
 		defer f.vacuumReportsTimer.Since(vacuumStart)
 		if f.vacuumFull.Load() {
 			full = true
 			query = fmt.Sprintf("vacuum full analyze %s", pq.QuoteIdentifier(f.table))
 		} else {
 			query = fmt.Sprintf("vacuum analyze %s", pq.QuoteIdentifier(f.table))
 		}
 		if _, err := f.db.ExecContext(ctx, query); err != nil {
 			f.log.Errorn(
 				"error vacuuming",
 				logger.NewStringField("table", f.table),
 				obskit.Error(err),
 				logger.NewBoolField("full", full),
 			)
 			return fmt.Errorf("error vacuuming table %w", err)
 		}
 		f.lastVacuum = time.Now()
+		f.deletedRows = 0
 	}
 	return nil
 }

enterprise/reporting/reporting.go

@@ -21,6 +21,7 @@
 	"github.com/cenkalti/backoff/v4"
 	"github.com/lib/pq"
 
+	"github.com/rudderlabs/rudder-go-kit/bytesize"
 	"github.com/rudderlabs/rudder-go-kit/config"
 	"github.com/rudderlabs/rudder-go-kit/logger"
 	"github.com/rudderlabs/rudder-go-kit/stats"
@@ -161,8 +162,8 @@
 		return func() {}
 	}
 	if _, err := dbHandle.ExecContext(context.Background(), `vacuum full analyze reports;`); err != nil {
 		r.log.Errorn(`[ Reporting ]: Error full vacuuming reports table`, logger.NewErrorField(err))
 		panic(err)
 	}
 	return func() {
 		r.g.Go(func() error {
@@ -385,8 +386,9 @@
 		var (
 			deletedRows                int
 			vacuumDeletedRowsThreshold = config.GetReloadableIntVar(100000, 1, "Reporting.vacuumThresholdDeletedRows")
-			lastVaccum                 time.Time
+			lastVacuum                 time.Time
 			vacuumInterval             = config.GetReloadableDurationVar(15, time.Minute, "Reporting.vacuumInterval")
+			vacuumThresholdBytes       = config.GetReloadableInt64Var(10*bytesize.GB, 1, "Reporting.vacuumThresholdBytes")
 		)
 		for {
 			if ctx.Err() != nil {
@@ -464,11 +466,21 @@
 					deletedRows += len(reports)
 				}
 
+				var sizeEstimate int64
+				if err := dbHandle.QueryRowContext(
+					ctx,
+					fmt.Sprintf(`SELECT pg_table_size(oid) from pg_class where relname='%s';`, ReportsTable),
+				).Scan(&sizeEstimate); err != nil {
+					r.log.Errorn(
+						`[ Reporting ]: Error getting table size estimate`,
+						logger.NewErrorField(err),
+					)
+				}
 				if deletedRows >= vacuumDeletedRowsThreshold.Load() ||
-					time.Since(lastVaccum) > vacuumInterval.Load() {
+					(sizeEstimate >= vacuumThresholdBytes.Load() && time.Since(lastVacuum) > vacuumInterval.Load()) {
 					if err := r.vacuum(ctx, dbHandle, tags); err == nil {
 						deletedRows = 0
-						lastVaccum = time.Now()
+						lastVacuum = time.Now()
 					}
 				}
 			}
@@ -488,27 +500,27 @@
 	}
 }
 
 func (r *DefaultReporter) vacuum(ctx context.Context, db *sql.DB, tags stats.Tags) error {
 	defer r.stats.NewTaggedStat(StatReportingVacuumDuration, stats.TimerType, tags).RecordDuration()()
 	var (
 		query string
 		full  bool
 	)
 	if r.vacuumFull.Load() {
 		full = true
 		query = `vacuum full analyze reports;`
 	} else {
 		query = `vacuum analyze reports;`
 	}
 	_, err := db.ExecContext(ctx, query)
 	if err != nil {
 		r.log.Errorn(
 			`[ Reporting ]: Error vacuuming reports table`,
 			obskit.Error(err),
 			logger.NewBoolField("full", full),
 		)
 	}
 	return nil
 }
 
 func (r *DefaultReporter) sendMetric(ctx context.Context, netClient *http.Client, label string, metric *types.Metric) error {