Add reference-aware cleanup for runnable-image staging trees

.github/workflows/e2e.yml

@@ -102,6 +102,11 @@ jobs:
             java: true
             cluster: true
             helm: false
+          - name: "Tier 17: Runnable-stage GC"
+            tiers: --tier 17
+            java: true
+            cluster: true
+            helm: true
     steps:
       - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
 

core/cmd/brewlet-metrics-exporter/README.md

@@ -3,7 +3,34 @@
 The node-local exporter receives bounded, best-effort telemetry from the
 containerd shim over `/opt/brewlet/metrics/telemetry.sock` and exposes
 Prometheus metrics for sandbox launches, artifact resolution, AppCDS decisions,
-and installed JDK and launcher inventory.
+installed JDK and launcher inventory, and runnable-image staging usage.
+
+`brewlet_runnable_stage_bytes` reports logical regular-file bytes under
+`--stage-root` (default: `BREWLET_RUNNABLE_STAGE`, or `/tmp/brewlet-runnable`
+on Linux regardless of `TMPDIR`). The directory must be the host's actual staging
+root; a containerized exporter needs a read-only host mount. The operator
+provides that mount for the default location. The gauge includes legacy and
+pending stages, does not follow symlinks, and refreshes on every scrape.
+Inspection errors fail the scrape instead of reporting zero.
+
+On Helm-managed nodes, the provisioner already schedules stage GC by default,
+independently of whether this exporter is enabled. The exporter never deletes
+stages. Do not add a second timer: it would run outside the operator's
+`stageGC.enabled` setting and migration gate.
+
+To inspect eligibility on a managed Linux node, run the installed helper as
+root in the host namespaces:
+
+```sh
+sudo /usr/local/bin/brewlet-stage-gc stage-gc --dry-run
+```
+
+Manual invocations retain the reaper's reference and mount checks but do not
+consult the provisioner's compatibility record or upgrade acknowledgment.
+Outside the provisioner, install the standalone `brewlet` CLI and arrange a
+host timer only after completing the same migration prerequisites.
+See [stage cleanup](../../../docs/runnable-image.md#reclaiming-unused-stages)
+for reference checks and rollout precautions.
 
 ## Screenshots
 

core/cmd/brewlet-metrics-exporter/main.go

@@ -19,6 +19,7 @@ import (
 	"syscall"
 	"time"
 
+	"github.com/microsoft/brewlet/internal/runnablestage"
 	"github.com/microsoft/brewlet/internal/telemetry"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
@@ -245,10 +246,12 @@ func main() {
 		root       string
 		listenAddr string
 		socketPath string
+		stageRoot  string
 	)
 	flag.StringVar(&root, "root", "/opt/brewlet", "Brewlet host-state root")
 	flag.StringVar(&listenAddr, "listen-address", ":9090", "HTTP metrics listen address")
 	flag.StringVar(&socketPath, "socket-path", "", "shim telemetry Unix datagram socket")
+	flag.StringVar(&stageRoot, "stage-root", runnablestage.Root(), "runnable staging root (BREWLET_RUNNABLE_STAGE)")
 	flag.Parse()
 	if socketPath == "" {
 		socketPath = filepath.Join(root, "metrics", "telemetry.sock")
@@ -261,6 +264,7 @@ func main() {
 	registry.MustRegister(prometheus.NewGoCollector(), prometheus.NewProcessCollector(prometheus.ProcessCollectorOpts{}))
 	events := newEventMetrics(registry)
 	registry.MustRegister(newInventoryCollector(root))
+	registry.MustRegister(newStageCollector(stageRoot))
 
 	server := &http.Server{
 		Addr:              listenAddr,

core/cmd/brewlet-metrics-exporter/stages.go

@@ -0,0 +1,38 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+package main
+
+import (
+	"github.com/microsoft/brewlet/internal/runnablestage"
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+type stageCollector struct {
+	root  string
+	bytes *prometheus.Desc
+}
+
+func newStageCollector(root string) *stageCollector {
+	return &stageCollector{
+		root: root,
+		bytes: prometheus.NewDesc(
+			"brewlet_runnable_stage_bytes",
+			"Logical bytes remaining in runnable staging trees, including legacy and pending stages.",
+			nil, nil,
+		),
+	}
+}
+
+func (c *stageCollector) Describe(ch chan<- *prometheus.Desc) {
+	ch <- c.bytes
+}
+
+func (c *stageCollector) Collect(ch chan<- prometheus.Metric) {
+	size, err := runnablestage.Bytes(c.root)
+	if err != nil {
+		ch <- prometheus.NewInvalidMetric(c.bytes, err)
+		return
+	}
+	ch <- prometheus.MustNewConstMetric(c.bytes, prometheus.GaugeValue, float64(size))
+}

core/cmd/brewlet-metrics-exporter/stages_test.go

@@ -0,0 +1,56 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+package main
+
+import (
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/client_golang/prometheus/testutil"
+)
+
+func TestStageCollector(t *testing.T) {
+	root := filepath.Join(t.TempDir(), "stages")
+	reg := prometheus.NewPedanticRegistry()
+	reg.MustRegister(newStageCollector(root))
+	check := func(value string) {
+		t.Helper()
+		expected := `
+# HELP brewlet_runnable_stage_bytes Logical bytes remaining in runnable staging trees, including legacy and pending stages.
+# TYPE brewlet_runnable_stage_bytes gauge
+brewlet_runnable_stage_bytes ` + value + "\n"
+		if err := testutil.GatherAndCompare(reg, strings.NewReader(expected)); err != nil {
+			t.Fatal(err)
+		}
+	}
+	check("0")
+	dir := filepath.Join(root, "immutable-v2", strings.Repeat("a", 64))
+	if err := os.MkdirAll(dir, 0o755); err != nil {
+		t.Fatal(err)
+	}
+	payload := filepath.Join(dir, "payload")
+	if err := os.WriteFile(payload, []byte("12345"), 0o644); err != nil {
+		t.Fatal(err)
+	}
+	check("5")
+	if err := os.RemoveAll(dir); err != nil {
+		t.Fatal(err)
+	}
+	check("0")
+}
+
+func TestStageCollectorReportsReadErrors(t *testing.T) {
+	root := filepath.Join(t.TempDir(), "not-a-directory")
+	if err := os.WriteFile(root, nil, 0o644); err != nil {
+		t.Fatal(err)
+	}
+	reg := prometheus.NewRegistry()
+	reg.MustRegister(newStageCollector(root))
+	if _, err := reg.Gather(); err == nil {
+		t.Fatal("expected a scrape error, not a misleading zero")
+	}
+}

core/cmd/brewlet/main.go

@@ -27,6 +27,7 @@ import (
 
 	"github.com/microsoft/brewlet/internal/artifact"
 	"github.com/microsoft/brewlet/internal/kube"
+	"github.com/microsoft/brewlet/internal/runnablestage"
 	"github.com/microsoft/brewlet/internal/runtime"
 )
 
@@ -55,6 +56,8 @@ func main() {
 		err = cmdJDKs(os.Args[2:])
 	case "doctor":
 		err = cmdDoctor(os.Args[2:])
+	case "stage-gc":
+		err = cmdStageGC(os.Args[2:])
 	case "k8s":
 		err = kube.Run(context.Background(), os.Args[2:], os.Stdout, os.Stderr)
 	case "version", "--version":
@@ -86,6 +89,7 @@ USAGE:
   brewlet bundle  <ref>       [--store DIR] [--cpu N] [--memory M] [--uid UID] [--gid GID] [--jdk-root DIR] [--launcher NAME] [--launcher-root DIR] [--out DIR]
   brewlet jdks                [--output table|wide|json] [--kubeconfig FILE] [--context CTX] [--selector SEL]
   brewlet doctor              [--namespace NS] [--output table|json] [--kubeconfig FILE] [--context CTX]
+  brewlet stage-gc            [--stage-root DIR] [--address SOCKET] [--min-age 24h] [--dry-run]
   brewlet k8s <command>        inventory, status, inspection, installation and profile updates (see k8s --help)
   brewlet version
 
@@ -674,6 +678,11 @@ func cmdRun(args []string) error {
 	}
 	ref := pos[0]
 
+	guard, err := runnablestage.Acquire(runnablestage.Root())
+	if err != nil {
+		return err
+	}
+	defer guard.Close()
 	s := artifact.Store{Root: *store}
 	blobs, err := s.ResolveBlobs(ref)
 	if err != nil {
@@ -736,6 +745,8 @@ func cmdRun(args []string) error {
 	fmt.Printf("[brewlet] sandbox  : %s\n", sandbox)
 	fmt.Printf("[brewlet] launch   : %s\n", plan.CommandLine())
 	fmt.Printf("[brewlet] ----- JVM output below -----\n")
+	// The sandbox and optional CDS cache now own all payloads.
+	_ = guard.Close()
 	return plan.Run()
 }
 
@@ -766,21 +777,35 @@ func cmdBundle(args []string) error {
 	if err != nil {
 		return err
 	}
+	guard, err := runnablestage.Acquire(runnablestage.Root())
+	if err != nil {
+		return err
+	}
+	defer guard.Close()
 	s := artifact.Store{Root: *store}
 	blobs, err := s.ResolveBlobs(pos[0])
 	if err != nil {
 		return err
 	}
+	res := runtime.Resources{CPULimit: *cpu, MemoryLimit: *mem}
+	if err := res.Validate(); err != nil {
+		return err
+	}
+	cleanup, err := retainBundlePayloads(&blobs, *out)
+	if err != nil {
+		return err
+	}
 	cfg := blobs.Config
 	cdsSrc := blobs.CDSHostPath
 	if err := runtime.GenerateBundleWithIdentityAndRegen(cfg, *jdkRoot, *launcherRoot, *launcher, blobs.JarHostPath, blobs.ClasspathHostPaths, blobs.ModulepathHostPaths, cdsSrc, *out,
-		runtime.Resources{CPULimit: *cpu, MemoryLimit: *mem}, nil, runtime.ProcessIdentity{UID: uid, GID: gid},
+		res, nil, runtime.ProcessIdentity{UID: uid, GID: gid},
 		runtime.CDSRegenOptions{
 			Regenerate:     *appcdsRegen,
 			CacheScope:     "local",
 			ArtifactDigest: blobs.ManifestDigest,
 			CacheDir:       os.Getenv("BREWLET_CDS_CACHE"),
 		}); err != nil {
+		cleanup()
 		return err
 	}
 	fmt.Printf("wrote OCI runtime bundle to %s/config.json\n", *out)

core/cmd/brewlet/stage_gc.go

@@ -0,0 +1,80 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+package main
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"os"
+	"os/signal"
+	"path/filepath"
+	"syscall"
+	"time"
+
+	"github.com/microsoft/brewlet/internal/artifact"
+	"github.com/microsoft/brewlet/internal/runnablestage"
+)
+
+func cmdStageGC(args []string) error {
+	fs := flag.NewFlagSet("stage-gc", flag.ContinueOnError)
+	root := fs.String("stage-root", runnablestage.Root(), "runnable staging root (BREWLET_RUNNABLE_STAGE)")
+	address := fs.String("address", "/run/containerd/containerd.sock", "containerd socket; all namespaces are checked")
+	age := fs.Duration("min-age", runnablestage.DefaultMinAge, "minimum stage age (must be positive)")
+	dryRun := fs.Bool("dry-run", false, "report eligible trees without deleting them")
+	if err := fs.Parse(args); err != nil {
+		if err == flag.ErrHelp {
+			return nil
+		}
+		return err
+	}
+	if fs.NArg() != 0 || *age <= 0 {
+		return fmt.Errorf("usage: stage-gc [--stage-root DIR] [--address SOCKET] [--min-age positive-duration] [--dry-run]")
+	}
+	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
+	defer stop()
+	ctx, cancel := context.WithTimeout(ctx, 5*time.Minute)
+	defer cancel()
+	result, err := runnablestage.Reap(ctx, *address, runnablestage.Options{
+		Root: *root, MinAge: *age, DryRun: *dryRun,
+	})
+	if err != nil {
+		return err
+	}
+	action := "removed"
+	if *dryRun {
+		action = "would remove"
+	}
+	fmt.Printf("%s %d stages (%d bytes); remaining stage bytes: %d\n", action, result.Removed, result.ReclaimedBytes, result.RemainingBytes)
+	return nil
+}
+
+// An exported bundle may be started long after this command exits. Give it
+// independent payloads rather than dangling references into the evictable cache.
+func retainBundlePayloads(blobs *artifact.ResolvedBlobs, out string) (func(), error) {
+	if blobs.Format != "image" {
+		return func() {}, nil
+	}
+	out, err := filepath.Abs(out)
+	if err != nil {
+		return nil, err
+	}
+	if err := os.MkdirAll(out, 0o755); err != nil {
+		return nil, err
+	}
+	dir, err := os.MkdirTemp(out, "brewlet-image-")
+	if err != nil {
+		return nil, err
+	}
+	cleanup := func() { _ = os.RemoveAll(dir) }
+	if err := os.CopyFS(dir, os.DirFS(filepath.Dir(blobs.JarHostPath))); err != nil {
+		cleanup()
+		return nil, fmt.Errorf("retain bundle payloads: %w", err)
+	}
+	blobs.JarHostPath = filepath.Join(dir, filepath.Base(blobs.JarHostPath))
+	if blobs.CDSHostPath != "" {
+		blobs.CDSHostPath = filepath.Join(dir, filepath.Base(blobs.CDSHostPath))
+	}
+	return cleanup, nil
+}