Repository: prometheus/prometheus
File: cmd/prometheus/main.go (Go)

Code before:
	}
	return strings.Join(ss, ",")
}

func (p *rwProtoMsgFlagParser) Set(opt string) error {
	t := remoteapi.WriteMessageType(opt)
	if err := t.Validate(); err != nil {
		return err
	}
	if slices.Contains(*p.msgs, t) {
		return fmt.Errorf("duplicated %v flag value, got %v already", t, *p.msgs)
	}
	*p.msgs = append(*p.msgs, t)
	return nil
}

type UploadMetaFile struct {
	Uploaded []string `json:"uploaded"`
}

// isBlockUploaded returns true if given TSDB block is marked as uploaded in the JSON file
// passed to Prometheus using the --storage.tsdb.upload-meta-file=... flag.
// If the file does not exist or there's an error while reading it then we assume it is uploaded
// so we don't block compactions of TSDB blocks by Prometheus.
func isBlockUploaded(logger *slog.Logger, uploadMetaPath string, meta *tsdb.BlockMeta) bool {
	data, err := os.ReadFile(uploadMetaPath)
	if err != nil {
		logger.Warn("cannot open TSDB upload meta file", slog.String("path", uploadMetaPath), slog.Any("err", err))
		return true
	}

	var shipper UploadMetaFile
	err = json.Unmarshal(data, &shipper)
	if err != nil {
		logger.Warn("cannot parse TSDB upload meta file", slog.String("path", uploadMetaPath), slog.Any("err", err))
		return true
	}

	return slices.Contains(shipper.Uploaded, meta.ULID.String())
}

func exludeBlocksPendingUpload(logger *slog.Logger, uploadMetaPath string) tsdb.BlockExcludeFilterFunc {
	return func(meta *tsdb.BlockMeta) bool {
		if meta.Compaction.Level > 1 {
			// Blocks with level > 1 are compacted, sidecars like Thanos don't upload these normally,
			// so we never exlude them.
			return false
		}
		uploaded := isBlockUploaded(logger, uploadMetaPath, meta)
		if !uploaded {
			logger.Info("Block still pending upload, excluding it from compactions", slog.String("ulid", meta.ULID.String()))

Review comment (performance):
Hm. Is there a downside for this vs watching changes and caching some version of the file instead of read for every single block queued? Or at least read once a filtering pass? It's slow and likely nothing will be updated for the 100ms for the filtering loop, so we just waste disk reads here, wdyt?

Code after:

func (p *rwProtoMsgFlagParser) Set(opt string) error {
	t := remoteapi.WriteMessageType(opt)
	if err := t.Validate(); err != nil {
		return err
	}
	if slices.Contains(*p.msgs, t) {
		return fmt.Errorf("duplicated %v flag value, got %v already", t, *p.msgs)
	}
	*p.msgs = append(*p.msgs, t)
	return nil
}

type UploadMeta struct {
	Uploaded []string `json:"uploaded"`
}

// Cache the last read UploadMeta.
var (
	tsdbDelayCompactLastMeta     *UploadMeta // The content of uploadMetaPath from the last time we've opened it.
	tsdbDelayCompactLastMetaTime time.Time   // The timestamp at which we stored tsdbDelayCompactLastMeta last time.
)

func exludeBlocksPendingUpload(logger *slog.Logger, uploadMetaPath string) tsdb.BlockExcludeFilterFunc {
	return func(meta *tsdb.BlockMeta) bool {
		if meta.Compaction.Level > 1 {
			// Blocks with level > 1 are assumed to be not uploaded, thus no need to delay those.
			// See `storage.tsdb.delay-compact-file.path` flag for detail.
			return false
		}

		// If we have cached uploadMetaPath content that was stored in the last minute the use it.
		if tsdbDelayCompactLastMeta != nil &&
			tsdbDelayCompactLastMetaTime.After(time.Now().UTC().Add(time.Minute*-1)) {
			return !slices.Contains(tsdbDelayCompactLastMeta.Uploaded, meta.ULID.String())
		}

		// We don't have anything cached or it's older than a minute. Try to open and parse the uploadMetaPath path.
		data, err := os.ReadFile(uploadMetaPath)
		if err != nil {
			logger.Warn("cannot open TSDB upload meta file", slog.String("path", uploadMetaPath), slog.Any("err", err))
			return false
		}

		var uploadMeta UploadMeta
		if err = json.Unmarshal(data, &uploadMeta); err != nil {
			logger.Warn("cannot parse TSDB upload meta file", slog.String("path", uploadMetaPath), slog.Any("err", err))
			return false
		}

		// We have parsed the uploadMetaPath file, cache it.