Merge pull request #6804 from dmcgowan/metadata-collectible-resources

Add collectible resources to metadata gc

metadata/db.go

@@ -94,6 +94,9 @@ type DB struct {
 	// set indicating whether any dirty flags are set
 	mutationCallbacks []func(bool)
 
+	// collectible resources
+	collectors map[gc.ResourceType]Collector
+
 	dbopts dbOptions
 }
 
@@ -265,6 +268,37 @@ func (m *DB) RegisterMutationCallback(fn func(bool)) {
 	m.wlock.Unlock()
 }
 
+// RegisterCollectibleResource registers a resource type which can be
+// referenced by metadata resources and garbage collected.
+// Collectible Resources are useful ephemeral resources which need to
+// to be tracked by go away after reboot or process restart.
+//
+// A few limitations to consider:
+// - Collectible Resources cannot reference other resources.
+// - A failure to complete collection will not fail the garbage collection,
+//   however, the resources can be collected in a later run.
+// - Collectible Resources must track whether the resource is active and/or
+//   lease membership.
+func (m *DB) RegisterCollectibleResource(t gc.ResourceType, c Collector) {
+	if t < resourceEnd {
+		panic("cannot re-register metadata resource")
+	} else if t >= gc.ResourceMax {
+		panic("resource type greater than max")
+	}
+
+	m.wlock.Lock()
+	defer m.wlock.Unlock()
+
+	if m.collectors == nil {
+		m.collectors = map[gc.ResourceType]Collector{}
+	}
+
+	if _, ok := m.collectors[t]; ok {
+		panic("cannot register collectible type twice")
+	}
+	m.collectors[t] = c
+}
+
 // GCStats holds the duration for the different phases of the garbage collector
 type GCStats struct {
 	MetaD     time.Duration
@@ -281,8 +315,9 @@ func (s GCStats) Elapsed() time.Duration {
 func (m *DB) GarbageCollect(ctx context.Context) (gc.Stats, error) {
 	m.wlock.Lock()
 	t1 := time.Now()
+	c := startGCContext(ctx, m.collectors)
 
-	marked, err := m.getMarked(ctx)
+	marked, err := m.getMarked(ctx, c) // Pass in gc context
 	if err != nil {
 		m.wlock.Unlock()
 		return nil, err
@@ -304,16 +339,17 @@ func (m *DB) GarbageCollect(ctx context.Context) (gc.Stats, error) {
 			} else if n.Type == ResourceContent || n.Type == ResourceIngest {
 				m.dirtyCS = true
 			}
-			return remove(ctx, tx, n)
+			return c.remove(ctx, tx, n) // From gc context
 		}
 
-		if err := scanAll(ctx, tx, rm); err != nil {
+		if err := c.scanAll(ctx, tx, rm); err != nil { // From gc context
 			return fmt.Errorf("failed to scan and remove: %w", err)
 		}
 
 		return nil
 	}); err != nil {
 		m.wlock.Unlock()
+		c.cancel(ctx)
 		return nil, err
 	}
 
@@ -358,13 +394,15 @@ func (m *DB) GarbageCollect(ctx context.Context) (gc.Stats, error) {
 	stats.MetaD = time.Since(t1)
 	m.wlock.Unlock()
 
+	c.finish(ctx)
+
 	wg.Wait()
 
 	return stats, err
 }
 
 // getMarked returns all resources that are used.
-func (m *DB) getMarked(ctx context.Context) (map[gc.Node]struct{}, error) {
+func (m *DB) getMarked(ctx context.Context, c *gcContext) (map[gc.Node]struct{}, error) {
 	var marked map[gc.Node]struct{}
 	if err := m.db.View(func(tx *bolt.Tx) error {
 		ctx, cancel := context.WithCancel(ctx)
@@ -383,7 +421,7 @@ func (m *DB) getMarked(ctx context.Context) (map[gc.Node]struct{}, error) {
 			}
 		}()
 		// Call roots
-		if err := scanRoots(ctx, tx, roots); err != nil {
+		if err := c.scanRoots(ctx, tx, roots); err != nil { // From gc context
 			cancel()
 			return err
 		}
@@ -392,7 +430,7 @@ func (m *DB) getMarked(ctx context.Context) (map[gc.Node]struct{}, error) {
 
 		refs := func(n gc.Node) ([]gc.Node, error) {
 			var sn []gc.Node
-			if err := references(ctx, tx, n, func(nn gc.Node) {
+			if err := c.references(ctx, tx, n, func(nn gc.Node) { // From gc context
 				sn = append(sn, nn)
 			}); err != nil {
 				return nil, err