refactor storage & tests

Author: v.kavlakan

cmd/prometheus/main.go

@@ -148,25 +148,26 @@ func agentOnlyFlag(app *kingpin.Application, name, help string) *kingpin.FlagCla
 type flagConfig struct {
 	configFile string
 
-	agentStoragePath     string
-	serverStoragePath    string
-	notifier             notifier.Options
-	forGracePeriod       model.Duration
-	outageTolerance      model.Duration
-	resendDelay          model.Duration
-	maxConcurrentEvals   int64
-	web                  web.Options
-	scrape               scrape.Options
-	tsdb                 tsdbOptions
-	agent                agentOptions
-	lookbackDelta        model.Duration
-	webTimeout           model.Duration
-	queryTimeout         model.Duration
-	queryConcurrency     int
-	queryMaxSamples      int
-	RemoteFlushDeadline  model.Duration
-	WalCommitInterval    model.Duration
-	HeadRetentionTimeout model.Duration
+	agentStoragePath          string
+	serverStoragePath         string
+	notifier                  notifier.Options
+	forGracePeriod            model.Duration
+	outageTolerance           model.Duration
+	resendDelay               model.Duration
+	maxConcurrentEvals        int64
+	web                       web.Options
+	scrape                    scrape.Options
+	tsdb                      tsdbOptions
+	agent                     agentOptions
+	lookbackDelta             model.Duration
+	webTimeout                model.Duration
+	queryTimeout              model.Duration
+	queryConcurrency          int
+	queryMaxSamples           int
+	RemoteFlushDeadline       model.Duration
+	WalCommitInterval         model.Duration
+	HeadRetentionDuration     model.Duration
+	StorageProcessingInterval model.Duration
 
 	featureList   []string
 	memlimitRatio float64
@@ -390,8 +391,10 @@ func main() {
 	serverOnlyFlag(a, "storage.wal-commit-interval", "Interval between force commits.").
 		Default("5000ms").SetValue(&cfg.WalCommitInterval)
 
-	serverOnlyFlag(a, "storage.head-retention-timeout", "Timeout before inactive heads are shrieked.").
-		Default("5m").SetValue(&cfg.HeadRetentionTimeout)
+	serverOnlyFlag(a, "storage.head-retention-duration", "Timeout before inactive heads are shrieked.").
+		Default("5m").SetValue(&cfg.HeadRetentionDuration)
+	serverOnlyFlag(a, "storage.processing-interval", "Interval before head conversion iterations.").
+		Default("1m").SetValue(&cfg.StorageProcessingInterval)
 
 	// TODO: Remove in Prometheus 3.0.
 	var b bool
@@ -724,10 +727,9 @@ func main() {
 		reloadBlocksTriggerNotifier,
 		receiverReadyNotifier,
 		time.Duration(cfg.WalCommitInterval),
-		time.Duration(cfg.tsdb.RetentionDuration),
-		time.Duration(cfg.HeadRetentionTimeout),
-		// x3 ScrapeInterval timeout for write block
-		time.Duration(cfgFile.GlobalConfig.ScrapeInterval*3),
+		time.Duration(cfg.tsdb.MinBlockDuration),
+		time.Duration(cfg.HeadRetentionDuration),
+		time.Duration(cfg.StorageProcessingInterval),
 	)
 	if err != nil {
 		level.Error(logger).Log("msg", "failed to create a receiver", "err", err)

pp-pkg/receiver/receiver.go

@@ -123,9 +123,9 @@ func NewReceiver(
 	triggerNotifier *ReloadBlocksTriggerNotifier,
 	readyNotifier ready.Notifier,
 	commitInterval time.Duration,
-	maxRetentionDuration time.Duration,
-	headRetentionTimeout time.Duration,
-	writeTimeout time.Duration,
+	retentionDuration time.Duration,
+	afterConversionRetentionDuration time.Duration,
+	processingInterval time.Duration,
 ) (*Receiver, error) {
 	if logger == nil {
 		logger = log.NewNopLogger()
@@ -197,9 +197,9 @@ func NewReceiver(
 		triggerNotifier,
 		clock,
 		appender.DefaultInitialDelay,
-		appender.DefaultWriteInterval,
-		maxRetentionDuration,
-		headRetentionTimeout,
+		processingInterval,
+		retentionDuration,
+		afterConversionRetentionDuration,
 		appender.DefaultQueueSize,
 		rotatedHeads...,
 	)