feat(orchestrator): add database integration support

cmd/harness/main.go

@@ -1,12 +1,16 @@
 package main
 
 import (
+	"context"
+	"errors"
 	"fmt"
 	"os"
 	"os/exec"
+	"os/signal"
 	"path/filepath"
 	"sort"
 	"strings"
+	"syscall"
 	"text/tabwriter"
 	"time"
 
@@ -52,6 +56,23 @@ var (
 )
 
 func main() {
+	// The root context is cancelled only by SIGINT/SIGTERM — never by a
+	// timeout — so context.Canceled unambiguously means "user interrupt"
+	// everywhere downstream. On the first signal the run loop winds down
+	// through its normal error path (deferred teardown still runs); the
+	// handler then unregisters itself so a second signal kills the process
+	// immediately via the default disposition.
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	sig := make(chan os.Signal, 1)
+	signal.Notify(sig, os.Interrupt, syscall.SIGTERM)
+	go func() {
+		<-sig
+		fmt.Fprintln(os.Stderr, "\ninterrupt — tearing down current run; press Ctrl+C again to exit immediately (then run `harness clean` to remove leftovers)")
+		cancel()
+		signal.Stop(sig)
+	}()
+
 	root := &cobra.Command{
 		Use:   "harness",
 		Short: "PipeBench — containerized data pipeline benchmarking",
@@ -73,7 +94,7 @@ No cloud account, Terraform, Ansible, or SSH required.`,
 	root.AddCommand(serveCmd())
 	root.AddCommand(versionCmd())
 
-	if err := root.Execute(); err != nil {
+	if err := root.ExecuteContext(ctx); err != nil {
 		os.Exit(1)
 	}
 }
@@ -240,7 +261,7 @@ func testCmd() *cobra.Command {
 				Drain:            drain,
 			}
 
-			r := runner.New(opts)
+			r := runner.New(cmd.Context(), opts)
 
 			// Stub out per-(hardware, subject) result files up front so a
 			// subject that fails every test still appears in the index.
@@ -284,7 +305,11 @@ func testCmd() *cobra.Command {
 				subjectOutcomes = subjectOutcomes[:0]
 			}
 
-			for _, p := range pairs {
+			for i, p := range pairs {
+				if cmd.Context().Err() != nil {
+					fmt.Fprintf(os.Stderr, "interrupted — skipping remaining %d run(s)\n", len(pairs)-i)
+					break
+				}
 				if perSubjectSummaries && p.subject.Name != currentSubject {
 					flushSubjectSummary()
 					currentSubject = p.subject.Name
@@ -307,7 +332,11 @@ func testCmd() *cobra.Command {
 
 				res, err := r.Run(p.tc, p.subject)
 				if err != nil {
-					fmt.Fprintf(os.Stderr, "ERROR running %s/%s: %v\n", p.tc.Name, p.subject.Name, err)
+					if errors.Is(err, context.Canceled) {
+						fmt.Fprintf(os.Stderr, "interrupted %s/%s\n", p.tc.Name, p.subject.Name)
+					} else {
+						fmt.Fprintf(os.Stderr, "ERROR running %s/%s: %v\n", p.tc.Name, p.subject.Name, err)
+					}
 					failed = append(failed, p.tc.Name+"/"+p.subject.Name)
 				}
 				o := runOutcome{
@@ -328,6 +357,9 @@ func testCmd() *cobra.Command {
 			}
 			printRunSummary(outcomes)
 
+			if cmd.Context().Err() != nil {
+				return fmt.Errorf("interrupted by signal (%d of %d runs attempted); leftover containers, if any: `harness clean`", len(outcomes), len(pairs))
+			}
 			if len(failed) > 0 {
 				return fmt.Errorf("%d run(s) failed: %v", len(failed), failed)
 			}
@@ -566,6 +598,16 @@ func cleanCmd() *cobra.Command {
 		Short: "Remove all bench containers and networks",
 		RunE: func(cmd *cobra.Command, args []string) error {
 			fmt.Println("Removing bench containers…")
+			// Prefix sweep catches every bench-* container regardless of
+			// name (localstack, azurite, redpanda, vault, verifier, plural
+			// generators/receivers, …) — the fixed list below is only a
+			// fallback for when `docker ps` itself fails.
+			if out, err := exec.Command("docker", "ps", "-aq", "--filter", "name=^bench-").Output(); err == nil {
+				if ids := strings.Fields(string(out)); len(ids) > 0 {
+					rmArgs := append([]string{"rm", "-f"}, ids...)
+					_ = exec.Command("docker", rmArgs...).Run()
+				}
+			}
 			containers := []string{
 				"bench-generator",
 				"bench-receiver",
@@ -577,6 +619,16 @@ func cleanCmd() *cobra.Command {
 			for _, c := range containers {
 				_ = exec.Command("docker", "rm", "-f", c).Run()
 			}
+			// Compose networks are named <tmpdir-project>_bench; an
+			// interrupted run that never reached `compose down` leaves them
+			// behind. Removal fails harmlessly while a container still uses
+			// the network, so containers go first above.
+			fmt.Println("Removing bench networks…")
+			if out, err := exec.Command("docker", "network", "ls", "-q", "--filter", "name=bench").Output(); err == nil {
+				for _, id := range strings.Fields(string(out)) {
+					_ = exec.Command("docker", "network", "rm", id).Run()
+				}
+			}
 			fmt.Println("Done.")
 			return nil
 		},

containers/collector/main.go

@@ -215,15 +215,8 @@ func dockerStatsToRow(cur *dockerStats, prev *dockerStats) MetricsRow {
 		cpuPct = 100 * numCPU
 	}
 
-	cache := cur.MemoryStats.Stats["cache"]
-	memUsed := int64(cur.MemoryStats.Usage) - int64(cache)
-	if memUsed < 0 {
-		memUsed = int64(cur.MemoryStats.Usage)
-	}
-	memFree := int64(cur.MemoryStats.Limit) - int64(cur.MemoryStats.Usage)
-	if memFree < 0 {
-		memFree = 0
-	}
+	memUsed, cache := memUsage(cur.MemoryStats.Usage, cur.MemoryStats.Stats)
+	memFree := max(int64(cur.MemoryStats.Limit)-int64(cur.MemoryStats.Usage), 0)
 
 	var netRecv, netSend int64
 	for _, n := range cur.Networks {
@@ -280,7 +273,7 @@ func dockerStatsToRow(cur *dockerStats, prev *dockerStats) MetricsRow {
 		CpuUsr:  cpuPct,
 		CpuIdl:  100.0 - cpuPct,
 		MemUsed: memUsed,
-		MemCach: int64(cache),
+		MemCach: cache,
 		MemFree: memFree,
 		NetRecv: netRecv,
 		NetSend: netSend,
@@ -289,6 +282,34 @@ func dockerStatsToRow(cur *dockerStats, prev *dockerStats) MetricsRow {
 	}
 }
 
+// memUsage converts Docker stats memory counters into used/cache bytes.
+//
+// All page cache is excluded, except shmem: cgroups charge shared image-layer
+// and binary pages to whichever container faults them first, and pages
+// touched twice migrate to the active list — so both raw usage and the
+// `docker stats` formula (usage - inactive_file) swing tens of MB between
+// otherwise identical runs. A benchmark needs a reproducible figure, so
+// mem_used is anon + shmem + kernel. shmem stays counted because tmpfs/shm
+// segments are swap-backed real memory that cgroups fold into the page
+// cache number.
+func memUsage(usage uint64, stats map[string]uint64) (used, cache int64) {
+	// cgroup v1 keys the page cache as "total_cache"/"cache"; v2 as "file".
+	pageCache, ok := stats["total_cache"]
+	shmem := stats["total_shmem"]
+	if !ok {
+		if pageCache, ok = stats["cache"]; !ok {
+			pageCache = stats["file"]
+		}
+		shmem = stats["shmem"]
+	}
+	cache = int64(pageCache)
+
+	if pageCache >= shmem && usage >= pageCache-shmem {
+		return int64(usage - (pageCache - shmem)), cache
+	}
+	return int64(usage), cache
+}
+
 // --- Common ---
 
 func mustEnv(key string) string {

containers/collector/main_test.go

@@ -0,0 +1,102 @@
+package main
+
+import "testing"
+
+func TestMemUsage(t *testing.T) {
+	tests := []struct {
+		name      string
+		usage     uint64
+		stats     map[string]uint64
+		wantUsed  int64
+		wantCache int64
+	}{
+		{
+			name:  "cgroup v2 excludes file cache",
+			usage: 200 << 20,
+			stats: map[string]uint64{
+				"file":          130 << 20,
+				"inactive_file": 120 << 20,
+				"active_file":   10 << 20,
+				"anon":          60 << 20,
+			},
+			wantUsed:  70 << 20,
+			wantCache: 130 << 20,
+		},
+		{
+			name:  "cgroup v2 keeps shmem counted",
+			usage: 200 << 20,
+			stats: map[string]uint64{
+				"file":  130 << 20,
+				"shmem": 30 << 20,
+			},
+			wantUsed:  100 << 20,
+			wantCache: 130 << 20,
+		},
+		{
+			name:  "cgroup v1 uses total_cache and total_shmem",
+			usage: 200 << 20,
+			stats: map[string]uint64{
+				"total_cache": 110 << 20,
+				"total_shmem": 20 << 20,
+				"cache":       90 << 20,
+			},
+			wantUsed:  110 << 20,
+			wantCache: 110 << 20,
+		},
+		{
+			name:  "cgroup v1 plain cache key fallback",
+			usage: 200 << 20,
+			stats: map[string]uint64{
+				"cache": 90 << 20,
+				"shmem": 10 << 20,
+			},
+			wantUsed:  120 << 20,
+			wantCache: 90 << 20,
+		},
+		{
+			name:  "cache >= usage falls back to raw usage",
+			usage: 50 << 20,
+			stats: map[string]uint64{
+				"file": 60 << 20,
+			},
+			wantUsed:  50 << 20,
+			wantCache: 60 << 20,
+		},
+		{
+			name:  "shmem > cache falls back to raw usage",
+			usage: 100 << 20,
+			stats: map[string]uint64{
+				"file":  10 << 20,
+				"shmem": 20 << 20,
+			},
+			wantUsed:  100 << 20,
+			wantCache: 10 << 20,
+		},
+		{
+			name:      "empty stats map reports raw usage",
+			usage:     70 << 20,
+			stats:     map[string]uint64{},
+			wantUsed:  70 << 20,
+			wantCache: 0,
+		},
+		{
+			name:      "nil stats map reports raw usage",
+			usage:     70 << 20,
+			stats:     nil,
+			wantUsed:  70 << 20,
+			wantCache: 0,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			used, cache := memUsage(tt.usage, tt.stats)
+			if used != tt.wantUsed {
+				t.Errorf("used = %d, want %d", used, tt.wantUsed)
+			}
+			if cache != tt.wantCache {
+				t.Errorf("cache = %d, want %d", cache, tt.wantCache)
+			}
+		})
+	}
+}