Refactoring after review

Signed-off-by: Marco Pracucci <[email protected]>

Author: pracucci

pkg/compactor/compactor.go

@@ -44,40 +44,12 @@ func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
 	cfg.retryMinBackoff = 10 * time.Second
 	cfg.retryMaxBackoff = time.Minute
 
-	f.Var(
-		&cfg.BlockRanges,
-		"compactor.block-ranges",
-		"Comma separated list of compaction ranges expressed in the time duration format")
-
-	f.DurationVar(
-		&cfg.ConsistencyDelay,
-		"compactor.consistency-delay",
-		30*time.Minute,
-		fmt.Sprintf("Minimum age of fresh (non-compacted) blocks before they are being processed. Malformed blocks older than the maximum of consistency-delay and %s will be removed.", compact.MinimumAgeForRemoval))
-
-	f.IntVar(
-		&cfg.BlockSyncConcurrency,
-		"compactor.block-sync-concurrency",
-		20,
-		"Number of goroutines to use when syncing block metadata from object storage")
-
-	f.StringVar(
-		&cfg.DataDir,
-		"compactor.data-dir",
-		"./data",
-		"Data directory in which to cache blocks and process compactions")
-
-	f.DurationVar(
-		&cfg.CompactionInterval,
-		"compactor.compaction-interval",
-		time.Hour,
-		"The frequency at which the compaction runs")
-
-	f.IntVar(
-		&cfg.CompactionRetries,
-		"compactor.compaction-retries",
-		3,
-		"How many times to retry a failed compaction during a single compaction interval")
+	f.Var(&cfg.BlockRanges, "compactor.block-ranges", "Comma separated list of compaction ranges expressed in the time duration format")
+	f.DurationVar(&cfg.ConsistencyDelay, "compactor.consistency-delay", 30*time.Minute, fmt.Sprintf("Minimum age of fresh (non-compacted) blocks before they are being processed. Malformed blocks older than the maximum of consistency-delay and %s will be removed.", compact.MinimumAgeForRemoval))
+	f.IntVar(&cfg.BlockSyncConcurrency, "compactor.block-sync-concurrency", 20, "Number of goroutines to use when syncing block metadata from object storage")
+	f.StringVar(&cfg.DataDir, "compactor.data-dir", "./data", "Data directory in which to cache blocks and process compactions")
+	f.DurationVar(&cfg.CompactionInterval, "compactor.compaction-interval", time.Hour, "The frequency at which the compaction runs")
+	f.IntVar(&cfg.CompactionRetries, "compactor.compaction-retries", 3, "How many times to retry a failed compaction during a single compaction interval")
 }
 
 // Compactor is a multi-tenant TSDB blocks compactor based on Thanos.
@@ -116,7 +88,7 @@ func NewCompactor(compactorCfg Config, storageCfg cortex_tsdb.Config, logger log
 		return nil, errors.Wrap(err, "failed to create the bucket client")
 	}
 
-	tsdbCompactor, err := tsdb.NewLeveledCompactor(ctx, registerer, logger, compactorCfg.BlockRanges.ToMillisecond(), downsample.NewPool())
+	tsdbCompactor, err := tsdb.NewLeveledCompactor(ctx, registerer, logger, compactorCfg.BlockRanges.ToMilliseconds(), downsample.NewPool())
 	if err != nil {
 		cancelCtx()
 		return nil, errors.Wrap(err, "failed to create TSDB compactor")
@@ -205,7 +177,7 @@ func (c *Compactor) compactUsersWithRetries(ctx context.Context) {
 	c.compactionRunsStarted.Inc()
 
 	for retries.Ongoing() {
-		if err := c.compactUsers(ctx); err == nil {
+		if success := c.compactUsers(ctx); success {
 			c.compactionRunsCompleted.Inc()
 			return
 		}
@@ -216,20 +188,20 @@ func (c *Compactor) compactUsersWithRetries(ctx context.Context) {
 	c.compactionRunsFailed.Inc()
 }
 
-func (c *Compactor) compactUsers(ctx context.Context) error {
+func (c *Compactor) compactUsers(ctx context.Context) bool {
 	level.Info(c.logger).Log("msg", "discovering users from bucket")
 	users, err := c.discoverUsers(ctx)
 	if err != nil {
 		level.Error(c.logger).Log("msg", "failed to discover users from bucket", "err", err)
-		return err
+		return false
 	}
 	level.Info(c.logger).Log("msg", "discovered users from bucket", "users", len(users))
 
 	for _, userID := range users {
 		// Ensure the context has not been canceled (ie. compactor shutdown has been triggered).
 		if ctx.Err() != nil {
 			level.Info(c.logger).Log("msg", "interrupting compaction of user blocks", "err", err)
-			return ctx.Err()
+			return false
 		}
 
 		level.Info(c.logger).Log("msg", "starting compaction of user blocks", "user", userID)
@@ -242,7 +214,7 @@ func (c *Compactor) compactUsers(ctx context.Context) error {
 		level.Info(c.logger).Log("msg", "successfully compacted user blocks", "user", userID)
 	}
 
-	return nil
+	return true
 }
 
 func (c *Compactor) compactUser(ctx context.Context, userID string) error {

pkg/ingester/ingester_v2.go

@@ -426,7 +426,7 @@ func (i *Ingester) createTSDB(userID string) (*tsdb.DB, error) {
 	// Create a new user database
 	db, err := tsdb.Open(udir, util.Logger, tsdbPromReg, &tsdb.Options{
 		RetentionDuration: uint64(i.cfg.TSDBConfig.Retention / time.Millisecond),
-		BlockRanges:       i.cfg.TSDBConfig.BlockRanges.ToMillisecond(),
+		BlockRanges:       i.cfg.TSDBConfig.BlockRanges.ToMilliseconds(),
 		NoLockfile:        true,
 	})
 	if err != nil {

pkg/storage/tsdb/config.go

@@ -72,11 +72,11 @@ func (d *DurationList) Set(s string) error {
 	return nil
 }
 
-// ToMillisecond returns the duration list in milliseconds
-func (d *DurationList) ToMillisecond() []int64 {
+// ToMilliseconds returns the duration list in milliseconds
+func (d *DurationList) ToMilliseconds() []int64 {
 	values := make([]int64, 0, len(*d))
 	for _, t := range *d {
-		values = append(values, int64(t/time.Millisecond))
+		values = append(values, t.Milliseconds())
 	}
 
 	return values

pkg/storage/tsdb/config_test.go

@@ -86,7 +86,7 @@ func TestConfig_DurationList(t *testing.T) {
 
 		t.Run(name, func(t *testing.T) {
 			testdata.f(&testdata.cfg)
-			assert.Equal(t, testdata.expectedRanges, testdata.cfg.BlockRanges.ToMillisecond())
+			assert.Equal(t, testdata.expectedRanges, testdata.cfg.BlockRanges.ToMilliseconds())
 		})
 	}
 }