Compact: Display Planned and Running Compactions

.bingo/gotesplit.sum

@@ -1,4 +1,6 @@
 github.com/Songmu/gotesplit v0.2.1 h1:qJFvR75nJpeKyMQFwyDtFrcc6zDWhrHAkks7DvM8oLo=
 github.com/Songmu/gotesplit v0.2.1/go.mod h1:sVBfmLT26b1H5VhUpq8cRhCVK75GAmW9c8r2NiK0gzk=
 github.com/jstemmer/go-junit-report v1.0.0 h1:8X1gzZpR+nVQLAht+L/foqOeX2l9DTZoaIPbEQHxsds=
+github.com/jstemmer/go-junit-report v1.0.0/go.mod h1:Brl9GWCQeLvo8nXZwPNNblvFj/XSXhF0NWZEnDohbsk=
 golang.org/x/sync v0.0.0-20220513210516-0976fa681c29 h1:w8s32wxx3sY+OjLlv9qltkLU5yvJzxjjgiHWLjdIcw4=
+golang.org/x/sync v0.0.0-20220513210516-0976fa681c29/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=

cmd/thanos/compact.go

@@ -239,6 +239,13 @@ func runCompact(
 	consistencyDelayMetaFilter := block.NewConsistencyDelayMetaFilter(logger, conf.consistencyDelay, extprom.WrapRegistererWithPrefix("thanos_", reg))
 	timePartitionMetaFilter := block.NewTimePartitionMetaFilter(conf.filterConf.MinTime, conf.filterConf.MaxTime)
 
+	levels, err := compactions.levels(conf.maxCompactionLevel)
+	if err != nil {
+		return errors.Wrap(err, "get compaction levels")
+	}
+
+	tsdbPlanner := compact.NewPlanner(logger, levels, noCompactMarkerFilter)
+
 	var blockLister block.Lister
 	switch syncStrategy(conf.blockListStrategy) {
 	case concurrentDiscovery:
@@ -288,6 +295,7 @@ func runCompact(
 		cf.UpdateOnChange(func(blocks []metadata.Meta, err error) {
 			api.SetLoaded(blocks, err)
 		})
+
 		sy, err = compact.NewMetaSyncer(
 			logger,
 			reg,
@@ -303,11 +311,6 @@ func runCompact(
 		}
 	}
 
-	levels, err := compactions.levels(conf.maxCompactionLevel)
-	if err != nil {
-		return errors.Wrap(err, "get compaction levels")
-	}
-
 	if conf.maxCompactionLevel < compactions.maxLevel() {
 		level.Warn(logger).Log("msg", "Max compaction level is lower than should be", "current", conf.maxCompactionLevel, "default", compactions.maxLevel())
 	}
@@ -371,7 +374,6 @@ func runCompact(
 	)
 	var planner compact.Planner
 
-	tsdbPlanner := compact.NewPlanner(logger, levels, noCompactMarkerFilter)
 	largeIndexFilterPlanner := compact.WithLargeTotalIndexSizeFilter(
 		tsdbPlanner,
 		insBkt,
@@ -394,6 +396,8 @@ func runCompact(
 		insBkt,
 		conf.compactionConcurrency,
 		conf.skipBlockWithOutOfOrderChunks,
+		api,
+		nil,
 	)
 	if err != nil {
 		return errors.Wrap(err, "create bucket compactor")

pkg/api/blocks/v1.go

@@ -26,15 +26,16 @@ import (
 
 // BlocksAPI is a very simple API used by Thanos Block Viewer.
 type BlocksAPI struct {
-	baseAPI          *api.BaseAPI
-	logger           log.Logger
-	globalBlocksInfo *BlocksInfo
-	loadedBlocksInfo *BlocksInfo
+	baseAPI           *api.BaseAPI
+	logger            log.Logger
+	globalBlocksInfo  *BlocksInfo
+	loadedBlocksInfo  *BlocksInfo
+	plannedBlocksInfo *BlocksInfo
 
 	globalLock, loadedLock sync.Mutex
-	disableCORS            bool
-	bkt                    objstore.Bucket
-	disableAdminOperations bool
+	disableCORS                         bool
+	bkt                                 objstore.Bucket
+	disableAdminOperations              bool
 }
 
 type BlocksInfo struct {
@@ -77,6 +78,10 @@ func NewBlocksAPI(logger log.Logger, disableCORS bool, label string, flagsMap ma
 			Blocks: []metadata.Meta{},
 			Label:  label,
 		},
+		plannedBlocksInfo: &BlocksInfo{
+			Blocks: []metadata.Meta{},
+			Label:  label,
+		},
 		disableCORS:            disableCORS,
 		bkt:                    bkt,
 		disableAdminOperations: disableAdminOperations,
@@ -90,6 +95,7 @@ func (bapi *BlocksAPI) Register(r *route.Router, tracer opentracing.Tracer, logg
 
 	r.Get("/blocks", instr("blocks", bapi.blocks))
 	r.Post("/blocks/mark", instr("blocks_mark", bapi.markBlock))
+	r.Get("/blocks/plan", instr("blocks_plan", bapi.plannedBlocks))
 }
 
 func (bapi *BlocksAPI) markBlock(r *http.Request) (interface{}, []error, *api.ApiError, func()) {
@@ -146,6 +152,10 @@ func (bapi *BlocksAPI) blocks(r *http.Request) (interface{}, []error, *api.ApiEr
 	return bapi.globalBlocksInfo, nil, nil, func() {}
 }
 
+func (bapi *BlocksAPI) plannedBlocks(r *http.Request) (interface{}, []error, *api.ApiError, func()) {
+    return bapi.plannedBlocksInfo, nil, nil, func() {}
+}
+
 func (b *BlocksInfo) set(blocks []metadata.Meta, err error) {
 	if err != nil {
 		// Last view is maintained.
@@ -174,3 +184,8 @@ func (bapi *BlocksAPI) SetLoaded(blocks []metadata.Meta, err error) {
 
 	bapi.loadedBlocksInfo.set(blocks, err)
 }
+
+// SetPlanned updates the plan blocks' metadata in the API.
+func (bapi *BlocksAPI) SetPlanned(blocks []metadata.Meta, err error) {
+	bapi.plannedBlocksInfo.set(blocks, err)
+}

pkg/compact/compact.go

@@ -65,6 +65,15 @@ type Syncer struct {
 	g singleflight.Group
 }
 
+func (s *Syncer) SetPlanned(metas []metadata.Meta) error {
+	s.mtx.Lock()
+	defer s.mtx.Unlock()
+	for _, meta := range metas {
+		s.blocks[meta.ULID] = &meta
+	}
+	return nil
+}
+
 // SyncerMetrics holds metrics tracked by the syncer. This struct and its fields are exported
 // to allow depending projects (eg. Cortex) to implement their own custom syncer while tracking
 // compatible metrics.
@@ -411,6 +420,12 @@ type Group struct {
 	extensions                    any
 }
 
+func (g *Group) Metas() []*metadata.Meta {
+	g.mtx.Lock()
+	defer g.mtx.Unlock()
+	return g.metasByMinTime
+}
+
 // NewGroup returns a new compaction group.
 func NewGroup(
 	logger log.Logger,
@@ -795,6 +810,13 @@ type Planner interface {
 	// Plan returns a list of blocks that should be compacted into single one.
 	// The blocks can be overlapping. The provided metadata has to be ordered by minTime.
 	Plan(ctx context.Context, metasByMinTime []*metadata.Meta, errChan chan error, extensions any) ([]*metadata.Meta, error)
+	// Delete before merge: UpdateOnPlanned allows to update the planner with a function that will be called with the planned blocks and an error.
+	UpdateOnPlanned(f func([]metadata.Meta, error))
+}
+
+// Delete before merge: UpdateOnPlanned allows to update the planner with a function that will be called with the planned blocks and an error.
+func (p *tsdbBasedPlanner) UpdateOnPlanned(f func([]metadata.Meta, error)) {
+	p.updateOnPlanned = f
 }
 
 type BlockDeletableChecker interface {
@@ -1364,6 +1386,8 @@ type BucketCompactor struct {
 	bkt                            objstore.Bucket
 	concurrency                    int
 	skipBlocksWithOutOfOrderChunks bool
+	blocksAPI                      BlocksAPI
+	extensions                     any
 }
 
 // NewBucketCompactor creates a new bucket compactor.
@@ -1377,6 +1401,8 @@ func NewBucketCompactor(
 	bkt objstore.Bucket,
 	concurrency int,
 	skipBlocksWithOutOfOrderChunks bool,
+	blocksAPI BlocksAPI,
+	extensions any,
 ) (*BucketCompactor, error) {
 	if concurrency <= 0 {
 		return nil, errors.Errorf("invalid concurrency level (%d), concurrency level must be > 0", concurrency)
@@ -1393,9 +1419,15 @@ func NewBucketCompactor(
 		bkt,
 		concurrency,
 		skipBlocksWithOutOfOrderChunks,
+		blocksAPI,
+		extensions,
 	)
 }
 
+type BlocksAPI interface {
+	SetPlanned([]metadata.Meta, error)
+}
+
 func NewBucketCompactorWithCheckerAndCallback(
 	logger log.Logger,
 	sy *Syncer,
@@ -1408,6 +1440,8 @@ func NewBucketCompactorWithCheckerAndCallback(
 	bkt objstore.Bucket,
 	concurrency int,
 	skipBlocksWithOutOfOrderChunks bool,
+	blocksAPI BlocksAPI,
+	extensions any,
 ) (*BucketCompactor, error) {
 	if concurrency <= 0 {
 		return nil, errors.Errorf("invalid concurrency level (%d), concurrency level must be > 0", concurrency)
@@ -1424,6 +1458,8 @@ func NewBucketCompactorWithCheckerAndCallback(
 		bkt:                            bkt,
 		concurrency:                    concurrency,
 		skipBlocksWithOutOfOrderChunks: skipBlocksWithOutOfOrderChunks,
+		blocksAPI:                      blocksAPI,
+		extensions:                     extensions,
 	}, nil
 }
 
@@ -1455,12 +1491,46 @@ func (c *BucketCompactor) Compact(ctx context.Context) (rerr error) {
 
 		// Set up workers who will compact the groups when the groups are ready.
 		// They will compact available groups until they encounter an error, after which they will stop.
+		var allPlannedMetas []metadata.Meta
+		var plannedMetasMutex sync.Mutex
+
 		for i := 0; i < c.concurrency; i++ {
 			wg.Add(1)
 			go func() {
 				defer wg.Done()
 				for g := range groupChan {
-					shouldRerunGroup, _, err := g.Compact(workCtx, c.compactDir, c.planner, c.comp, c.blockDeletableChecker, c.compactionLifecycleCallback)
+					groupMetas := g.Metas()
+					plannedMetas := make([]metadata.Meta, len(groupMetas))
+					for i, meta := range groupMetas {
+						plannedMetas[i] = *meta
+					}
+
+					plannedMetasMutex.Lock()
+					allPlannedMetas = append(allPlannedMetas, plannedMetas...)
+					plannedMetasMutex.Unlock()
+
+					// Delete before merge: Call Syncer.SetPlanned before compaction
+					if err := c.sy.SetPlanned(plannedMetas); err != nil {
+						errChan <- errors.Wrapf(err, "set planned for group %s", g.Key())
+						return
+					}
+
+					// Delete before merge: Call BlocksAPI.SetPlanned to update the API
+					c.blocksAPI.SetPlanned(plannedMetas, nil)
+
+					shouldRerunGroup, compIDs, err := g.Compact(workCtx, c.compactDir, c.planner, c.comp, c.blockDeletableChecker, c.compactionLifecycleCallback)
+					if err == nil {
+						plannedMetas := make([]metadata.Meta, 0, len(compIDs))
+						for _, id := range compIDs {
+							if meta, ok := c.sy.Metas()[id]; ok {
+								plannedMetas = append(plannedMetas, *meta)
+							}
+						}
+						if err := c.sy.SetPlanned(plannedMetas); err != nil {
+							level.Warn(c.logger).Log("msg", "failed to set planned metas after compaction", "err", err)
+						}
+						c.blocksAPI.SetPlanned(plannedMetas, nil)
+					}
 					if err == nil {
 						if shouldRerunGroup {
 							mtx.Lock()
@@ -1531,7 +1601,33 @@ func (c *BucketCompactor) Compact(ctx context.Context) (rerr error) {
 
 		level.Info(c.logger).Log("msg", "start of compactions")
 
-		// Send all groups found during this pass to the compaction workers.
+		// Delete before merge: Convert map to slice
+		metas := make([]*metadata.Meta, 0, len(c.sy.Metas()))
+		for _, meta := range c.sy.Metas() {
+			metas = append(metas, meta)
+		}
+
+		// Delete before merge: Plan compaction outside of the group loop
+		plan, err := c.planner.Plan(ctx, metas, errChan, c.extensions)
+		if err != nil {
+			return errors.Wrap(err, "plan compaction")
+		}
+
+		// Delete before merge: Convert []*metadata.Meta to []metadata.Meta
+		plannedMetas := make([]metadata.Meta, len(plan))
+		for i, meta := range plan {
+			plannedMetas[i] = *meta
+		}
+
+		// Delete before merge: Call Syncer.SetPlanned before compaction
+		if err := c.sy.SetPlanned(plannedMetas); err != nil {
+			return errors.Wrap(err, "set planned metas")
+		}
+
+		// Delete before merge: Call BlocksAPI.SetPlanned to update the API
+		c.blocksAPI.SetPlanned(plannedMetas, nil)
+
+		// Now proceed with the group loop
 		var groupErrs errutil.MultiError
 	groupLoop:
 		for _, g := range groups {
@@ -1549,6 +1645,11 @@ func (c *BucketCompactor) Compact(ctx context.Context) (rerr error) {
 		close(groupChan)
 		wg.Wait()
 
+		if err := c.sy.SetPlanned(allPlannedMetas); err != nil {
+			return errors.Wrap(err, "set planned metas")
+		}
+		c.blocksAPI.SetPlanned(allPlannedMetas, nil)
+
 		// Collect any other error reported by the workers, or any error reported
 		// while we were waiting for the last batch of groups to run the compaction.
 		close(errChan)

pkg/compact/compact_e2e_test.go

@@ -217,7 +217,8 @@ func testGroupCompactE2e(t *testing.T, mergeFunc storage.VerticalChunkSeriesMerg
 
 		planner := NewPlanner(logger, []int64{1000, 3000}, noCompactMarkerFilter)
 		grouper := NewDefaultGrouper(logger, bkt, false, false, reg, blocksMarkedForDeletion, garbageCollectedBlocks, blocksMaredForNoCompact, metadata.NoneFunc, 10, 10)
-		bComp, err := NewBucketCompactor(logger, sy, grouper, planner, comp, dir, bkt, 2, true)
+		blocksAPI := &metaFetcherWrapper{metaFetcher}
+		bComp, err := NewBucketCompactor(logger, sy, grouper, planner, comp, dir, bkt, 2, true, blocksAPI, mergeFunc)
 		testutil.Ok(t, err)
 
 		// Compaction on empty should not fail.
@@ -542,3 +543,11 @@ func listBlocksMarkedForDeletion(ctx context.Context, bkt objstore.Bucket) ([]ul
 	})
 	return rem, err
 }
+
+type metaFetcherWrapper struct {
+    *block.MetaFetcher
+}
+
+func (m *metaFetcherWrapper) SetPlanned(metas []metadata.Meta, err error) {
+    // Implement the logic as needed
+}

pkg/compact/planner.go

@@ -25,6 +25,7 @@ type tsdbBasedPlanner struct {
 	ranges []int64
 
 	noCompBlocksFunc func() map[ulid.ULID]*metadata.NoCompactMark
+	updateOnPlanned  func([]metadata.Meta, error)
 }
 
 var _ Planner = &tsdbBasedPlanner{}

pkg/ui/react-app/src/App.tsx

@@ -5,6 +5,7 @@ import { QueryParamProvider } from 'use-query-params';
 import useMedia from 'use-media';
 
 import { Alerts, Config, Flags, Rules, ServiceDiscovery, Status, Targets, TSDBStatus, PanelList, NotFound } from './pages';
+import { GlobalBlocks, PlannedBlocks } from './thanos/pages/blocks/Blocks';
 import PathPrefixProps from './types/PathPrefixProps';
 import ThanosComponentProps from './thanos/types/ThanosComponentProps';
 import Navigation from './thanos/Navbar';
@@ -17,7 +18,7 @@ const defaultRouteConfig: { [component: string]: string } = {
   query: '/graph',
   rule: '/alerts',
   bucket: '/blocks',
-  compact: '/loaded',
+  compact: '/global-blocks',
   store: '/loaded',
 };
 
@@ -61,6 +62,8 @@ const App: FC<PathPrefixProps & ThanosComponentProps> = ({ pathPrefix, thanosCom
               <Stores path="/stores" pathPrefix={pathPrefix} />
               <Blocks path="/blocks" pathPrefix={pathPrefix} />
               <Blocks path="/loaded" pathPrefix={pathPrefix} view="loaded" />
+              <GlobalBlocks path="/global-blocks" pathPrefix={pathPrefix} />
+              <PlannedBlocks path="/planned-blocks" pathPrefix={pathPrefix} />
               <NotFound pathPrefix={pathPrefix} default defaultRoute={defaultRouteConfig[thanosComponent]} />
             </Router>
           </QueryParamProvider>

pkg/ui/react-app/src/thanos/Navbar.tsx

@@ -63,7 +63,8 @@ const navConfig: { [component: string]: (NavConfig | NavDropDown)[] } = {
     },
   ],
   compact: [
-    { name: 'Global Blocks', uri: '/blocks' },
+    { name: 'Global Blocks', uri: '/global-blocks' },
+    { name: 'Planned Blocks', uri: '/planned-blocks' },
     { name: 'Loaded Blocks', uri: '/loaded' },
     {
       name: 'Status',

pkg/ui/react-app/src/thanos/pages/blocks/Blocks.tsx

@@ -207,7 +207,7 @@ interface BlocksProps {
 
 export const Blocks: FC<RouteComponentProps & PathPrefixProps & BlocksProps> = ({ pathPrefix = '', view = 'global' }) => {
   const { response, error, isLoading } = useFetch<BlockListProps>(
-    `${pathPrefix}/api/v1/blocks${view ? '?view=' + view : ''}`
+    `${pathPrefix}/api/v1/blocks${view === 'planned' ? '/plan' : view ? '?view=' + view : ''}`
   );
   const { status: responseStatus } = response;
   const badResponse = responseStatus !== 'success' && responseStatus !== 'start fetching';
@@ -221,5 +221,12 @@ export const Blocks: FC<RouteComponentProps & PathPrefixProps & BlocksProps> = (
     />
   );
 };
+export const GlobalBlocks: React.FC<RouteComponentProps & PathPrefixProps> = ({ pathPrefix }) => {
+  return <Blocks view="global" pathPrefix={pathPrefix} />;
+};
+
+export const PlannedBlocks: React.FC<RouteComponentProps & PathPrefixProps> = ({ pathPrefix }) => {
+  return <Blocks view="planned" pathPrefix={pathPrefix} />;
+};
 
 export default Blocks;