feat(downsampling): persist decimated blocks and interpolate rate-like functions over them

cmd/prometheus/main.go

@@ -60,10 +60,12 @@ import (
 
 	"github.com/prometheus/prometheus/pp-pkg/blocks/block"
 	"github.com/prometheus/prometheus/pp-pkg/blocks/expirationpolicy"
+	"github.com/prometheus/prometheus/pp-pkg/blocks/fanout"
 	"github.com/prometheus/prometheus/pp-pkg/blocks/lcompactor"
-	"github.com/prometheus/prometheus/pp-pkg/blocks/manager"
-	"github.com/prometheus/prometheus/pp-pkg/blocks/tcompactor" // PP_CHANGES.md: rebuild on cpp
+	"github.com/prometheus/prometheus/pp-pkg/blocks/manager" // PP_CHANGES.md: rebuild on cpp
+	"github.com/prometheus/prometheus/pp-pkg/blocks/tcompactor"
 	"github.com/prometheus/prometheus/pp-pkg/featuresflags"
+	v1 "github.com/prometheus/prometheus/web/api/v1"
 
 	// PP_CHANGES.md: rebuild on cpp
 	"github.com/prometheus/prometheus/pp-pkg/localstorageobserver"
@@ -75,10 +77,8 @@ import (
 	pp_pkg_remote "github.com/prometheus/prometheus/pp-pkg/storage/remote" // PP_CHANGES.md: rebuild on cpp
 	pp_pkg_tsdb "github.com/prometheus/prometheus/pp-pkg/tsdb"             // PP_CHANGES.md: rebuild on cpp
 
-	pp_storage "github.com/prometheus/prometheus/pp/go/storage" // PP_CHANGES.md: rebuild on cpp
-	// PP_CHANGES.md: rebuild on cpp
-	"github.com/prometheus/prometheus/pp/go/storage/catalog" // PP_CHANGES.md: rebuild on cpp
-	// PP_CHANGES.md: rebuild on cpp
+	pp_storage "github.com/prometheus/prometheus/pp/go/storage"   // PP_CHANGES.md: rebuild on cpp
+	"github.com/prometheus/prometheus/pp/go/storage/catalog"      // PP_CHANGES.md: rebuild on cpp
 	"github.com/prometheus/prometheus/pp/go/storage/querier"      // PP_CHANGES.md: rebuild on cpp
 	"github.com/prometheus/prometheus/pp/go/storage/ready"        // PP_CHANGES.md: rebuild on cpp
 	"github.com/prometheus/prometheus/pp/go/storage/remotewriter" // PP_CHANGES.md: rebuild on cpp
@@ -183,6 +183,8 @@ type flagConfig struct {
 	HeadRetentionTimeout    model.Duration
 	UseBlockManagerStorage  bool
 
+	Downsampling model.Duration
+
 	featureList   []string
 	memlimitRatio float64
 	// These options are extracted from featureList
@@ -293,6 +295,11 @@ func (c *flagConfig) DisableBlockManagerStorage() {
 	c.UseBlockManagerStorage = false
 }
 
+// SetDownsampling sets the downsampling duration for the flagConfig.
+func (c *flagConfig) SetDownsampling(downsampling model.Duration) {
+	c.Downsampling = downsampling
+}
+
 func main() {
 	if os.Getenv("DEBUG") != "" {
 		runtime.SetBlockProfileRate(20)
@@ -657,6 +664,17 @@ func main() {
 		cfg.tsdb.OutOfOrderTimeWindow = cfgFile.StorageConfig.TSDBConfig.OutOfOrderTimeWindow
 	}
 
+	// check if the downsampling value is greater than the scrape interval
+	if cfg.Downsampling != 0 && cfg.Downsampling < cfgFile.GlobalConfig.ScrapeInterval {
+		_ = level.Error(logger).Log(
+			"msg", "The downsampling value must be greater than the scrape interval",
+			"scrape_interval", cfgFile.GlobalConfig.ScrapeInterval,
+			"downsampling", cfg.Downsampling,
+		)
+
+		os.Exit(2)
+	}
+
 	// Now that the validity of the config is established, set the config
 	// success metrics accordingly, although the config isn't really loaded
 	// yet. This will happen later (including setting these metrics again),
@@ -803,6 +821,7 @@ func main() {
 		&pp_storage.Options{
 			Seed:                cfgFile.GlobalConfig.ExternalLabels.Hash(),
 			BlockDuration:       time.Duration(cfg.tsdb.MinBlockDuration),
+			Downsampling:        time.Duration(cfg.Downsampling),
 			CommitInterval:      time.Duration(cfg.WalCommitInterval),
 			MaxRetentionPeriod:  ppRetentionPeriod,
 			HeadRetentionPeriod: time.Duration(cfg.HeadRetentionTimeout),
@@ -832,10 +851,16 @@ func main() {
 		prometheus.DefaultRegisterer,
 	)
 
+	retentionMS := int64(time.Duration(cfg.tsdb.RetentionDuration) / time.Millisecond)
+	downsamplingMS := int64(time.Duration(cfg.Downsampling) / time.Millisecond)
 	adapter := pp_pkg_storage.NewAdapter(
 		clock,
 		hManager.Proxy(),
 		hManager.Builder(),
+		&pp_pkg_storage.AdapterOptions{
+			RetentionMS:    retentionMS,
+			DownsamplingMS: downsamplingMS,
+		},
 		hManager.MergeOutOfOrderChunks,
 		prometheus.DefaultRegisterer,
 	)
@@ -861,6 +886,7 @@ func main() {
 		tsdbHistorical   *tsdbHistoricalStorage
 		persistedStorage storage.Storage = localStorage
 		startTimeFn                      = localStorage.StartTime
+		minTimeBlocks    v1.MinTimeBlocks
 	)
 	if !agentMode {
 		// Storage is constructed eagerly here for both schemes. The historical
@@ -892,7 +918,6 @@ func main() {
 				"CorruptedRetentionDuration", cfg.tsdb.CorruptedRetentionDuration,
 				"EnableOverlappingCompaction", cfg.tsdb.EnableOverlappingCompaction,
 			)
-			retentionMS := int64(time.Duration(cfg.tsdb.RetentionDuration) / time.Millisecond)
 
 			chunkPool := chunkenc.NewPool()
 			compactCtx, compactCancel := context.WithCancel(context.Background())
@@ -928,6 +953,7 @@ func main() {
 				localStoragePath,
 				&manager.Options{
 					RetentionDuration:           retentionMS,
+					DownsamplingMS:              downsamplingMS,
 					CorruptedRetentionDuration:  time.Duration(cfg.tsdb.CorruptedRetentionDuration),
 					EnableOverlappingCompaction: cfg.tsdb.EnableOverlappingCompaction,
 				},
@@ -948,6 +974,8 @@ func main() {
 				os.Exit(1)
 			}
 
+			minTimeBlocks = blockManager.MinTimeBlocks
+
 			bs := &blockStorage{m: blockManager, onClose: func() error {
 				// Cancel any in-flight leveled compaction first so the manager
 				// loop can return promptly, then stop the loop and close blocks.
@@ -984,7 +1012,7 @@ func main() {
 		log.With(logger, "component", "remote"),
 		startTimeFn,
 	)
-	fanoutStorage := storage.NewFanout(
+	fanoutStorage := fanout.New(
 		logger,
 		adapter,
 		persistedStorage,
@@ -1174,6 +1202,9 @@ func main() {
 	cfg.web.RuleManager = ruleManager
 	cfg.web.Notifier = notifierManager
 	cfg.web.LookbackDelta = time.Duration(cfg.lookbackDelta)
+	// x2 because the needed minimum lookback delta is 2*downsampling
+	cfg.web.DownsamplingLookbackDelta = time.Duration(cfg.Downsampling * 2)
+	cfg.web.MinTimeBlocks = minTimeBlocks
 	cfg.web.IsAgent = agentMode
 	cfg.web.AppName = modeAppName
 

cmd/prompptool/blocks.go

@@ -0,0 +1,137 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"os"
+	"slices"
+	"strconv"
+	"text/tabwriter"
+	"time"
+
+	"github.com/alecthomas/kingpin/v2"
+	"github.com/alecthomas/units"
+	"github.com/go-kit/log"
+	"github.com/prometheus/client_golang/prometheus"
+
+	"github.com/prometheus/prometheus/pp-pkg/blocks/block"
+)
+
+// cmdBlocks is the command for listing blocks.
+type cmdBlocks struct {
+	humanReadable bool
+}
+
+// registerCmdBlocks registers the blocks command.
+func registerCmdBlocks(cmd *cmdBlocks, clause *kingpin.CmdClause) {
+	clause.Flag(
+		"human-readable",
+		"Print human readable values. Default is false.",
+	).Default("false").Short('r').BoolVar(&cmd.humanReadable)
+}
+
+// Do lists the blocks in the directory.
+func (cmd *cmdBlocks) Do(
+	ctx context.Context,
+	workingDir string,
+	logger log.Logger,
+	registerer prometheus.Registerer,
+) error {
+	return listBlocks(logger, workingDir, cmd.humanReadable)
+}
+
+// listBlocks lists the blocks in the directory.
+func listBlocks(logger log.Logger, path string, humanReadable bool) error {
+	blocks, _, err := block.OpenBlocks(logger, path, nil, nil)
+	if err != nil {
+		return err
+	}
+	defer func() {
+		err = errors.Join(err, block.CloseAll(blocks))
+	}()
+
+	slices.SortFunc(blocks, func(a, b *block.Block) int {
+		switch {
+		case a.Meta().MinTime < b.Meta().MinTime:
+			return -1
+		case a.Meta().MinTime > b.Meta().MinTime:
+			return 1
+		default:
+			return 0
+		}
+	})
+
+	printBlocks(blocks, true, humanReadable)
+
+	return nil
+}
+
+// printBlocks prints the blocks to the console.
+func printBlocks(blocks []*block.Block, writeHeader, humanReadable bool) {
+	tw := tabwriter.NewWriter(os.Stdout, 13, 0, 2, ' ', 0)
+	defer tw.Flush()
+
+	if writeHeader {
+		fmt.Fprintln(
+			tw,
+			"BLOCK ULID\tMIN TIME\tMAX TIME\tDURATION\tNUM SAMPLES\tNUM CHUNKS\tNUM SERIES\tSIZE\tRESOLUTION\tLABELS",
+		)
+	}
+
+	for _, b := range blocks {
+		meta := b.Metadata()
+
+		fmt.Fprintf(tw,
+			"%v\t%v\t%v\t%v\t%v\t%v\t%v\t%v\t%v\t%v\n",
+			meta.ULID,
+			getFormatedTime(meta.MinTime, humanReadable),
+			getFormatedTime(meta.MaxTime, humanReadable),
+			time.Duration(meta.MaxTime-meta.MinTime)*time.Millisecond,
+			meta.Stats.NumSamples,
+			meta.Stats.NumChunks,
+			meta.Stats.NumSeries,
+			getFormatedBytes(b.Size(), humanReadable),
+			getFormatedDuration(meta.Thanos.Downsample.Resolution, humanReadable),
+			labelsToString(meta.Thanos.Labels),
+		)
+	}
+}
+
+// getFormatedTime converts a timestamp to a human readable string.
+func getFormatedTime(timestamp int64, humanReadable bool) string {
+	if humanReadable {
+		return time.Unix(timestamp/1000, 0).UTC().String()
+	}
+
+	return strconv.FormatInt(timestamp, 10)
+}
+
+// getFormatedBytes converts a number of bytes to a human readable string.
+func getFormatedBytes(bytes int64, humanReadable bool) string {
+	if humanReadable {
+		return units.Base2Bytes(bytes).String()
+	}
+
+	return strconv.FormatInt(bytes, 10)
+}
+
+// getFormatedDuration converts a duration to a human readable string.
+func getFormatedDuration(duration int64, humanReadable bool) string {
+	if humanReadable {
+		return (time.Duration(duration) * time.Millisecond).String()
+	}
+
+	return strconv.FormatInt(duration, 10)
+}
+
+// labelsToString converts a map[string]string to a JSON string.
+func labelsToString(ls map[string]string) string {
+	data, err := json.Marshal(ls)
+	if err != nil {
+		return err.Error()
+	}
+
+	return string(data)
+}

cmd/prompptool/main.go

@@ -6,6 +6,7 @@ import (
 	"os/signal"
 	"path/filepath"
 
+	"github.com/prometheus/common/model"
 	"github.com/prometheus/common/promlog"
 	"github.com/prometheus/common/version"
 
@@ -47,6 +48,10 @@ func main() {
 	var persistHeadCmd cmdPersistHead
 	registerCmdPersistHead(&persistHeadCmd, persistHeadClause)
 
+	blocksClause := app.Command("blocks", "Converting prometheus wal to tsdb-blocks.")
+	var blocksCmd cmdBlocks
+	registerCmdBlocks(&blocksCmd, blocksClause)
+
 	cmd := kingpin.MustParse(app.Parse(os.Args[1:]))
 	logger := initLogger(*verbose)
 	logger = log.With(logger, "cmd", cmd)
@@ -70,6 +75,11 @@ func main() {
 			level.Error(logger).Log("msg", "fail to persist head", "error", err)
 			os.Exit(1)
 		}
+	case blocksClause.FullCommand():
+		if err := blocksCmd.Do(ctx, workingDir, logger, nil); err != nil {
+			level.Error(logger).Log("msg", "fail to list blocks", "error", err)
+			os.Exit(1)
+		}
 	}
 }
 
@@ -103,3 +113,6 @@ type noopFlagConfig struct{}
 
 // DisableBlockManagerStorage is a no-op implementation of the FlagConfig interface, used when no feature flags are set.
 func (noopFlagConfig) DisableBlockManagerStorage() {}
+
+// SetDownsampling sets the downsampling duration for the flagConfig.
+func (noopFlagConfig) SetDownsampling(model.Duration) {}

cmd/prompptool/persisthead.go

@@ -130,6 +130,7 @@ func (cmd *cmdPersistHead) Do(
 	bw := block.NewWriter[*shard.Shard](
 		outputDir,
 		block.DefaultChunkSegmentSize,
+		cppbridge.NoDownsampling,
 		time.Duration(cmd.blockDuration),
 		0, // no retention filtering: persist all heads regardless of age
 		clockwork.NewRealClock(),

cmd/prompptool/walpp.go

@@ -86,6 +86,7 @@ func (cmd *cmdWALPPToBlock) Do(
 	bw := block.NewWriter[*shard.Shard](
 		workingDir,
 		block.DefaultChunkSegmentSize,
+		cppbridge.NoDownsampling,
 		time.Duration(cmd.blockDuration),
 		0, // no retention filtering: persist all heads regardless of age
 		clock,

pp-pkg/blocks/block/block.go

@@ -270,6 +270,13 @@ func (pb *Block) Metadata() metadata.Meta {
 	return pb.meta
 }
 
+// MinTime returns the minimum time of the block.
+func (pb *Block) MinTime() int64 {
+	pb.mtxMeta.RLock()
+	defer pb.mtxMeta.RUnlock()
+	return pb.meta.MinTime
+}
+
 // OverlapsClosedInterval returns true if the block overlaps [mint, maxt].
 func (pb *Block) OverlapsClosedInterval(mint, maxt int64) bool {
 	// The block itself is a half-open interval
@@ -558,14 +565,16 @@ func (o Overlaps) String() string {
 			))
 		}
 
-		res = append(res, fmt.Sprintf(
-			"[key: %s, mint: %d, maxt: %d, range: %s, blocks: %d]: %s",
-			r.Key,
-			r.Min, r.Max,
-			(time.Duration((r.Max-r.Min)/1000)*time.Second).String(),
-			len(overlaps),
-			strings.Join(groups, ", "),
-		))
+		res = append(
+			res, fmt.Sprintf(
+				"[key: %s, mint: %d, maxt: %d, range: %s, blocks: %d]: %s",
+				r.Key,
+				r.Min, r.Max,
+				(time.Duration((r.Max-r.Min)/1000)*time.Second).String(),
+				len(overlaps),
+				strings.Join(groups, ", "),
+			),
+		)
 	}
 
 	return strings.Join(res, "\n")

pp-pkg/blocks/block/functions.go

@@ -24,7 +24,7 @@ const checkContextEveryNIterations = 100
 //
 
 // CloseAll closes all given closers.
-func CloseAll(cs []io.Closer) error {
+func CloseAll[TCloser io.Closer](cs []TCloser) error {
 	errs := make([]error, 0, len(cs))
 	for _, c := range cs {
 		errs = append(errs, c.Close())