test(byoo): enforce collector startup health

tests/perf/byoo-otel-collector/README.md

@@ -8,8 +8,8 @@ library (`icms-translate`) rather than hand-written collector manifests.
 > Status: `render` (translate + validate, no cluster), `run` (provision a
 > managed k3d cluster or use a remote one, deploy an in-cluster OTLP sink + the
 > authentic collector pointed at it, drive telemetrygen load, and measure a
-> baseline), and `cleanup` are implemented. There are no pass/fail thresholds
-> yet; `run` establishes a reproducible baseline.
+> baseline), and `cleanup` are implemented. Collector startup health has a
+> target and maximum duration. Throughput has no pass/fail threshold yet.
 
 ## Why translation-driven
 
@@ -37,8 +37,8 @@ what the suite measures.
   the profile's rates.
 - `pkg/deploy`: applies the rendered workload and the sink to a cluster (k3d or
   remote), fronts the collector with a harness OTLP Service, backs its secrets
-  volume with export credentials, waits for readiness, runs the load, and tears
-  everything down.
+  volume with export credentials, measures collector startup health, waits for
+  readiness, runs the load, and tears everything down.
 - `pkg/labels`: the shared labels every object carries so cleanup is scoped.
 - `pkg/k3d`: provisions/tears down the managed local k3d cluster (`k3d` mode).
 - `pkg/report`: parses collector/sink metric scrapes over the measurement
@@ -103,8 +103,9 @@ multi-document stream (`---`-separated) and `json` emits an array, so
 2. renders the authentic collector with its export redirected at the sink
    (provider `OTEL_COLLECTOR`, endpoints pointed at the sink Service) and backs
    its secrets volume with dummy export credentials so the exporter can start;
-3. deploys the collector (fronted by a harness ClusterIP OTLP Service) and waits
-   for it to become ready;
+3. deploys the collector (fronted by a harness ClusterIP OTLP Service), polls
+   its `/health` endpoint, records startup timing, and waits for it to become
+   ready;
 4. repeats the following load+measure cycle once per profile repetition
    (`dev` runs once, `baseline` runs three times):
    1. drives telemetrygen load at the profile's rates, waiting for the generator
@@ -126,6 +127,17 @@ multi-document stream (`---`-separated) and `json` emits an array, so
    `<shape>.json` for a single repetition, `<shape>-run<N>.json` for several),
    then cleans up unless `--retain` is set.
 
+Startup health is the first successful HTTP response from the collector's
+`/health` endpoint on port `13133`. Reports record both Pod start to health and
+collector-container start to health. The first duration includes Pod startup
+and image pull. The second isolates collector initialization and is enforced.
+
+The default target is 15 seconds. The command emits a warning above the target
+and fails above the 30-second maximum. Adjust either bound with
+`--startup-target` or `--startup-max` when characterizing a known slower
+environment. These bounds do not alter the telemetry warmup or throughput
+measurement window.
+
 The target cluster depends on `--mode`:
 
 - `--mode k3d` (default): the suite provisions a dedicated local k3d cluster
@@ -159,13 +171,13 @@ GOWORK=off go run ./cmd/perf run --shape both --results-dir ./results
 
 Flags: `--shape`, `--profile`, `--mode` (`k3d`/`remote`), `--collector-image`,
 `--sink-image`, `--loadgen-image`, `--namespace`, `--kubeconfig`, `--context`,
-`--ready-timeout` (`3m`), `--retain`, `--skip-load`, `--k3d-cluster`,
-`--import-images`, `--results-dir`.
+`--ready-timeout` (`3m`), `--startup-target` (`15s`), `--startup-max` (`30s`),
+`--retain`, `--skip-load`, `--k3d-cluster`, `--import-images`, `--results-dir`.
 
-The baseline has no pass/fail thresholds yet; the numbers establish a
-reproducible baseline. Metric-name suffixes vary across collector-contrib
-versions, so `pkg/report` matches a list of candidate names per concept and
-notes any that were missing from the scrape rather than failing.
+The startup-health bound is the only pass/fail threshold. The remaining numbers
+establish a reproducible baseline. Metric-name suffixes vary across
+collector-contrib versions, so `pkg/report` matches a list of candidate names
+per concept and notes any that were missing from the scrape rather than failing.
 
 ### `cleanup`
 

tests/perf/byoo-otel-collector/cmd/perf/main.go

@@ -120,6 +120,8 @@ type runConfig struct {
 	kubeconfig     string
 	kubeContext    string
 	readyTimeout   time.Duration
+	startupTarget  time.Duration
+	startupMax     time.Duration
 	retain         bool
 	skipLoad       bool
 	k3dCluster     string
@@ -131,11 +133,12 @@ func newRunCmd() *cobra.Command {
 	var cfg runConfig
 	cmd := &cobra.Command{
 		Use:   "run",
-		Short: "Deploy the collector + OTLP sink, drive load, and wait until it is ready",
+		Short: "Deploy the collector + OTLP sink, check startup health, and drive load",
 		Long: `run renders the production workload shape via icms-translate, validates it,
 deploys an in-cluster OTLP sink, deploys the authentic BYOO collector pointed at
-that sink, waits for both to become ready, and drives telemetrygen load at the
-selected profile's rates. It cleans up afterward unless --retain is set.
+that sink, measures collector startup health, waits for both to become ready,
+and drives telemetrygen load at the selected profile's rates. It cleans up
+afterward unless --retain is set.
 
 With --mode k3d (the default) it provisions a dedicated local k3d cluster, runs
 against it, and deletes it afterward (unless --retain). With --mode remote it
@@ -156,6 +159,8 @@ reporting land in a later milestone.`,
 	cmd.Flags().StringVar(&cfg.kubeconfig, "kubeconfig", "", "path to kubeconfig (remote mode; defaults to in-cluster or $KUBECONFIG)")
 	cmd.Flags().StringVar(&cfg.kubeContext, "context", "", "kubeconfig context to use (remote mode)")
 	cmd.Flags().DurationVar(&cfg.readyTimeout, "ready-timeout", 3*time.Minute, "how long to wait for the collector and sink to become ready")
+	cmd.Flags().DurationVar(&cfg.startupTarget, "startup-target", 15*time.Second, "target collector-container start-to-health duration")
+	cmd.Flags().DurationVar(&cfg.startupMax, "startup-max", 30*time.Second, "maximum collector-container start-to-health duration")
 	cmd.Flags().BoolVar(&cfg.retain, "retain", false, "retain deployed resources (and the managed k3d cluster) instead of cleaning up")
 	cmd.Flags().BoolVar(&cfg.skipLoad, "skip-load", false, "deploy the collector and sink but do not drive load")
 	cmd.Flags().StringVar(&cfg.k3dCluster, "k3d-cluster", "byoo-perf", "name of the managed k3d cluster (k3d mode)")
@@ -195,6 +200,9 @@ func runRun(stdout io.Writer, cfg runConfig) error {
 	if cfg.mode != "k3d" && cfg.mode != "remote" {
 		return fmt.Errorf("unknown mode %q (want \"k3d\" or \"remote\")", cfg.mode)
 	}
+	if err := validateStartupThresholds(cfg.startupTarget, cfg.startupMax); err != nil {
+		return err
+	}
 	prof, err := profile.Lookup(cfg.profile)
 	if err != nil {
 		return err
@@ -234,10 +242,19 @@ func runRun(stdout io.Writer, cfg runConfig) error {
 		}
 	}
 
-	if cfg.skipLoad {
-		fmt.Fprintln(stdout, "note: --skip-load set; the collector and sink were deployed but no load was driven and no baseline was measured.")
-	} else {
-		fmt.Fprintln(stdout, "note: load was driven end-to-end and a baseline was measured. There are no pass/fail thresholds yet; these numbers establish the reproducible baseline.")
+	if err := writeRunCompletion(stdout, cfg.skipLoad); err != nil {
+		return err
+	}
+	return nil
+}
+
+func writeRunCompletion(w io.Writer, skipLoad bool) error {
+	message := "note: load was driven end-to-end and a baseline was measured. Startup health has a pass/fail bound; throughput and delivery numbers establish the reproducible baseline."
+	if skipLoad {
+		message = "note: --skip-load set; the collector and sink were deployed but no load was driven and no baseline was measured."
+	}
+	if _, err := fmt.Fprintln(w, message); err != nil {
+		return fmt.Errorf("write run completion message: %w", err)
 	}
 	return nil
 }
@@ -334,6 +351,16 @@ func runShape(ctx context.Context, stdout io.Writer, client *deploy.Client, cfg
 		return cleanupAfterErr(ctx, client, cfg, ns, fmt.Errorf("deploy %s: %w", shape, err))
 	}
 
+	fmt.Fprintf(stdout, "[%s] waiting up to %s for collector pod %q to start and up to %s after container start for /health ...\n", shape, cfg.readyTimeout, dep.PodName, cfg.startupMax)
+	startup, err := client.WaitCollectorHealth(ctx, ns, dep.PodName, render.CollectorContainerName, cfg.readyTimeout, cfg.startupMax)
+	if err != nil {
+		return cleanupAfterErr(ctx, client, cfg, ns, fmt.Errorf("collector did not report healthy for %s shape: %w", shape, err))
+	}
+	printStartupHealth(stdout, shape, startup, cfg.startupTarget, cfg.startupMax)
+	if err := checkStartupHealth(startup, cfg.startupMax); err != nil {
+		return cleanupAfterErr(ctx, client, cfg, ns, fmt.Errorf("collector startup failed for %s shape: %w", shape, err))
+	}
+
 	fmt.Fprintf(stdout, "[%s] waiting up to %s for collector pod %q to become ready ...\n", shape, cfg.readyTimeout, dep.PodName)
 	if err := client.WaitPodReady(ctx, ns, dep.PodName, cfg.readyTimeout); err != nil {
 		return cleanupAfterErr(ctx, client, cfg, ns, fmt.Errorf("collector did not become ready for %s shape: %w", shape, err))
@@ -386,7 +413,7 @@ func runShape(ctx context.Context, stdout io.Writer, client *deploy.Client, cfg
 				return nil
 			},
 			func(run int) report.ShapeReport {
-				return measure(ctx, stdout, client, cfg, prof, shape, ns, dep.PodName, collectorMetricsPort)
+				return measure(ctx, stdout, client, cfg, prof, shape, ns, dep.PodName, collectorMetricsPort, startup)
 			},
 			func(run int) error {
 				return client.WaitLoad(ctx, ns, loadgen.Jobs(ns, dep.PodName, lgOpts), cfg.readyTimeout)
@@ -422,7 +449,7 @@ func runShape(ctx context.Context, stdout io.Writer, client *deploy.Client, cfg
 // measurement window (after a warmup) and computes the baseline. Scrapes are
 // best-effort: a failed scrape yields empty samples, which Build records as
 // missing rather than failing the run.
-func measure(ctx context.Context, stdout io.Writer, client *deploy.Client, cfg runConfig, prof profile.Profile, shape spec.Shape, ns, collectorPod, collectorMetricsPort string) report.ShapeReport {
+func measure(ctx context.Context, stdout io.Writer, client *deploy.Client, cfg runConfig, prof profile.Profile, shape spec.Shape, ns, collectorPod, collectorMetricsPort string, startup report.StartupHealth) report.ShapeReport {
 	snap := func(label string) report.Snapshot {
 		s, collErr, sinkErr := takeSnapshot(
 			func() (report.Samples, error) {
@@ -461,9 +488,38 @@ func measure(ctx context.Context, stdout io.Writer, client *deploy.Client, cfg r
 		Window:        report.Window{Start: start, End: end},
 		Health:        health,
 		HealthErr:     healthErr,
+		StartupHealth: &startup,
 	})
 }
 
+func validateStartupThresholds(target, max time.Duration) error {
+	if target <= 0 {
+		return fmt.Errorf("startup target must be positive, got %s", target)
+	}
+	if max < target {
+		return fmt.Errorf("startup maximum %s must be at least the target %s", max, target)
+	}
+	return nil
+}
+
+func printStartupHealth(w io.Writer, shape spec.Shape, startup report.StartupHealth, target, max time.Duration) {
+	podToHealth := time.Duration(startup.PodToHealthSeconds * float64(time.Second)).Round(time.Millisecond)
+	collectorToHealth := time.Duration(startup.CollectorToHealthSeconds * float64(time.Second))
+	collectorToHealthDisplay := collectorToHealth.Round(time.Millisecond)
+	fmt.Fprintf(w, "[%s] startup health: pod_to_health=%s collector_to_health=%s (target <= %s, maximum <= %s)\n", shape, podToHealth, collectorToHealthDisplay, target, max)
+	if collectorToHealth > target && collectorToHealth <= max {
+		fmt.Fprintf(w, "[%s] warning: collector startup exceeded the %s target\n", shape, target)
+	}
+}
+
+func checkStartupHealth(startup report.StartupHealth, max time.Duration) error {
+	collectorToHealth := time.Duration(startup.CollectorToHealthSeconds * float64(time.Second))
+	if collectorToHealth > max {
+		return fmt.Errorf("collector start-to-health duration %s exceeded the %s maximum", collectorToHealth.Round(time.Millisecond), max)
+	}
+	return nil
+}
+
 // loadStartupMargin extends the generator run beyond warmup+window. The
 // generators start when their pods reach Running; measurement warmup only
 // begins after that, so without a margin the measurement window would extend

tests/perf/byoo-otel-collector/cmd/perf/main_test.go

@@ -21,6 +21,7 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"strings"
@@ -174,18 +175,20 @@ func TestRenderCmdSummary(t *testing.T) {
 func TestRunCmdDefaults(t *testing.T) {
 	cmd := newRunCmd()
 	defaults := map[string]string{
-		"shape":         "both",
-		"profile":       "dev",
-		"mode":          "k3d",
-		"namespace":     "byoo-perf",
-		"ready-timeout": "3m0s",
-		"retain":        "false",
-		"skip-load":     "false",
-		"sink-image":    sink.DefaultImage,
-		"loadgen-image": loadgen.DefaultImage,
-		"k3d-cluster":   "byoo-perf",
-		"import-images": "false",
-		"results-dir":   "",
+		"shape":          "both",
+		"profile":        "dev",
+		"mode":           "k3d",
+		"namespace":      "byoo-perf",
+		"ready-timeout":  "3m0s",
+		"startup-target": "15s",
+		"startup-max":    "30s",
+		"retain":         "false",
+		"skip-load":      "false",
+		"sink-image":     sink.DefaultImage,
+		"loadgen-image":  loadgen.DefaultImage,
+		"k3d-cluster":    "byoo-perf",
+		"import-images":  "false",
+		"results-dir":    "",
 	}
 	for name, want := range defaults {
 		f := cmd.Flags().Lookup(name)
@@ -205,6 +208,8 @@ func TestRunCmdInvalidSelectors(t *testing.T) {
 		{"--mode", "nope"},
 		{"--profile", "nope"},
 		{"--shape", "nope"},
+		{"--startup-target", "0s"},
+		{"--startup-target", "31s", "--startup-max", "30s"},
 	} {
 		cmd := newRunCmd()
 		cmd.SetArgs(args)
@@ -265,6 +270,8 @@ func TestRunCleansUpPodWhenServiceCreateFails(t *testing.T) {
 		collectorImage: spec.DefaultCollectorImage,
 		namespace:      "byoo-perf",
 		readyTimeout:   time.Second,
+		startupTarget:  15 * time.Second,
+		startupMax:     30 * time.Second,
 	}
 	if err := runRun(io.Discard, cfg); err == nil {
 		t.Fatal("expected run to fail when service creation is rejected")
@@ -427,6 +434,79 @@ func TestLoadGenDurationExceedsWindowByMargin(t *testing.T) {
 	}
 }
 
+func TestStartupThresholds(t *testing.T) {
+	if err := validateStartupThresholds(15*time.Second, 30*time.Second); err != nil {
+		t.Fatalf("valid thresholds: %v", err)
+	}
+	for _, tt := range []struct {
+		target time.Duration
+		max    time.Duration
+	}{
+		{target: 0, max: 30 * time.Second},
+		{target: 31 * time.Second, max: 30 * time.Second},
+	} {
+		if err := validateStartupThresholds(tt.target, tt.max); err == nil {
+			t.Errorf("validateStartupThresholds(%s, %s): expected error", tt.target, tt.max)
+		}
+	}
+}
+
+func TestStartupHealthThresholdOutput(t *testing.T) {
+	startup := report.NewStartupHealth(
+		time.Unix(0, 0),
+		time.Unix(2, 0),
+		time.Unix(22, 0),
+	)
+	var out bytes.Buffer
+	printStartupHealth(&out, spec.ShapeContainer, startup, 15*time.Second, 30*time.Second)
+	if !strings.Contains(out.String(), "collector_to_health=20s") || !strings.Contains(out.String(), "exceeded the 15s target") {
+		t.Errorf("startup output missing duration or warning:\n%s", out.String())
+	}
+	if err := checkStartupHealth(startup, 30*time.Second); err != nil {
+		t.Fatalf("20s startup should meet 30s maximum: %v", err)
+	}
+	if err := checkStartupHealth(startup, 15*time.Second); err == nil {
+		t.Fatal("20s startup should exceed 15s maximum")
+	}
+}
+
+func TestPrintStartupHealthUsesUnroundedThresholds(t *testing.T) {
+	for _, tt := range []struct {
+		name        string
+		collectorTo time.Duration
+		wantWarning bool
+	}{
+		{name: "over target", collectorTo: 15*time.Second + 400*time.Microsecond, wantWarning: true},
+		{name: "over maximum", collectorTo: 30*time.Second + 400*time.Microsecond, wantWarning: false},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			startup := report.NewStartupHealth(time.Unix(0, 0), time.Unix(0, 0), time.Unix(0, 0).Add(tt.collectorTo))
+			var out bytes.Buffer
+			printStartupHealth(&out, spec.ShapeContainer, startup, 15*time.Second, 30*time.Second)
+			gotWarning := strings.Contains(out.String(), "warning: collector startup exceeded")
+			if gotWarning != tt.wantWarning {
+				t.Errorf("warning = %t, want %t; output:\n%s", gotWarning, tt.wantWarning, out.String())
+			}
+		})
+	}
+}
+
+type failingWriter struct{ err error }
+
+func (w failingWriter) Write([]byte) (int, error) {
+	return 0, w.err
+}
+
+func TestWriteRunCompletionReturnsWriteError(t *testing.T) {
+	writeErr := errors.New("write failed")
+	for _, skipLoad := range []bool{false, true} {
+		err := writeRunCompletion(failingWriter{err: writeErr}, skipLoad)
+		if !errors.Is(err, writeErr) {
+			t.Errorf("writeRunCompletion(skipLoad=%t) error = %v, want wrapped write error", skipLoad, err)
+		}
+	}
+}
+
 // takeSnapshot must stamp Snapshot.At after both scrapes return, so a slow
 // (but successful) scrape does not produce a timestamp earlier than when the
 // samples were actually captured. A window built from such snapshots would