diff --git a/tests/perf/byoo-otel-collector/README.md b/tests/perf/byoo-otel-collector/README.md index 053fff0b9..4190b00c3 100644 --- a/tests/perf/byoo-otel-collector/README.md +++ b/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 `.json` for a single repetition, `-run.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` diff --git a/tests/perf/byoo-otel-collector/cmd/perf/main.go b/tests/perf/byoo-otel-collector/cmd/perf/main.go index 8d27133a2..8fe8dd8f0 100644 --- a/tests/perf/byoo-otel-collector/cmd/perf/main.go +++ b/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 diff --git a/tests/perf/byoo-otel-collector/cmd/perf/main_test.go b/tests/perf/byoo-otel-collector/cmd/perf/main_test.go index 7b543ca53..781694f58 100644 --- a/tests/perf/byoo-otel-collector/cmd/perf/main_test.go +++ b/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 diff --git a/tests/perf/byoo-otel-collector/pkg/deploy/deploy.go b/tests/perf/byoo-otel-collector/pkg/deploy/deploy.go index 6e2d84b64..2c2c2b775 100644 --- a/tests/perf/byoo-otel-collector/pkg/deploy/deploy.go +++ b/tests/perf/byoo-otel-collector/pkg/deploy/deploy.go @@ -65,6 +65,9 @@ const ( accountsSecretsMountPath = "/var/secrets" // accountsSecretsFile is the input file name the extractor waits for. accountsSecretsFile = "accounts-secrets.json" + + collectorHealthPort = "13133" + collectorHealthPath = "/health" ) // Client wraps a Kubernetes clientset with the operations the suite needs. @@ -462,26 +465,36 @@ func (c *Client) RunLoad(ctx context.Context, namespace string, jobs []*batchv1. // as a missing (best-effort) sample. It is a variable so tests can shorten it. var scrapeTimeout = 15 * time.Second +// healthPollInterval is the cadence for observing collector startup health. It +// is a variable so tests can shorten it. +var healthPollInterval = time.Second + // proxyGet performs the low-level API-proxy fetch. It is a seam so tests can // exercise the timeout path without a live cluster. var proxyGet = func(ctx context.Context, cs kubernetes.Interface, namespace, pod, port, path string) ([]byte, error) { return cs.CoreV1().Pods(namespace).ProxyGet("http", pod, port, path, nil).DoRaw(ctx) } -// ScrapePodMetrics fetches a pod's Prometheus endpoint through the API server -// proxy. This works without a metrics-server, an ingress, or port-forwarding, -// and is the only cross-namespace-safe way to read in-cluster endpoints from -// outside the cluster. Each scrape is bounded by scrapeTimeout. -func (c *Client) ScrapePodMetrics(ctx context.Context, namespace, pod, port, path string) ([]byte, error) { +// FetchPodEndpoint fetches an HTTP pod endpoint through the API server proxy. +// This works without a metrics-server, an ingress, or port-forwarding, and is +// the only cross-namespace-safe way to read in-cluster endpoints from outside +// the cluster. Each fetch is bounded by scrapeTimeout. +func (c *Client) FetchPodEndpoint(ctx context.Context, namespace, pod, port, path string) ([]byte, error) { ctx, cancel := context.WithTimeout(ctx, scrapeTimeout) defer cancel() raw, err := proxyGet(ctx, c.cs, namespace, pod, port, path) if err != nil { - return nil, fmt.Errorf("scrape %s/%s:%s%s: %w", namespace, pod, port, path, err) + return nil, fmt.Errorf("fetch %s/%s:%s%s: %w", namespace, pod, port, path, err) } return raw, nil } +// ScrapePodMetrics fetches a pod's Prometheus endpoint through the API server +// proxy. It is a metrics-specific wrapper around FetchPodEndpoint. +func (c *Client) ScrapePodMetrics(ctx context.Context, namespace, pod, port, path string) ([]byte, error) { + return c.FetchPodEndpoint(ctx, namespace, pod, port, path) +} + // PodHealth reports the collector pod's phase, aggregate restart count, and // whether any container was OOM killed. func (c *Client) PodHealth(ctx context.Context, namespace, pod string) (report.PodHealth, error) { @@ -562,6 +575,65 @@ func (c *Client) WaitPodReady(ctx context.Context, namespace, podName string, ti }) } +// WaitCollectorHealth waits up to timeout for the BYOO collector container to +// start. Once started, it waits no longer than startupMax for the health +// endpoint to return successfully. The returned timestamps separate pod +// scheduling/image-pull delay from collector initialization delay. +func (c *Client) WaitCollectorHealth(ctx context.Context, namespace, podName, collectorContainer string, timeout, startupMax time.Duration) (report.StartupHealth, error) { + var startup report.StartupHealth + err := wait.PollUntilContextTimeout(ctx, healthPollInterval, timeout, true, func(ctx context.Context) (bool, error) { + pod, err := c.cs.CoreV1().Pods(namespace).Get(ctx, podName, metav1.GetOptions{}) + if err != nil { + if k8serrors.IsNotFound(err) { + return false, nil + } + return false, err + } + if pod.Status.Phase == corev1.PodFailed { + return false, fmt.Errorf("pod %q failed: %s", podName, pod.Status.Reason) + } + if reason, ok := terminalContainerFailure(pod); ok { + return false, fmt.Errorf("pod %q not schedulable/healthy: %s", podName, reason) + } + if pod.Status.StartTime == nil { + return false, nil + } + collectorStartedAt, ok := containerStartedAt(pod, collectorContainer) + if !ok { + return false, nil + } + startupDeadline := collectorStartedAt.Add(startupMax) + if time.Now().After(startupDeadline) { + return false, fmt.Errorf("collector container %q exceeded startup maximum %s without reporting healthy", collectorContainer, startupMax) + } + healthCtx, cancel := context.WithDeadline(ctx, startupDeadline) + _, healthErr := c.FetchPodEndpoint(healthCtx, namespace, podName, collectorHealthPort, collectorHealthPath) + cancel() + healthyAt := time.Now().UTC() + if healthyAt.After(startupDeadline) { + return false, fmt.Errorf("collector container %q exceeded startup maximum %s without reporting healthy", collectorContainer, startupMax) + } + if healthErr == nil { + startup = report.NewStartupHealth(pod.Status.StartTime.Time, collectorStartedAt, healthyAt) + return true, nil + } + return false, nil + }) + if err != nil { + return report.StartupHealth{}, fmt.Errorf("wait for collector health endpoint %s:%s%s: %w", podName, collectorHealthPort, collectorHealthPath, err) + } + return startup, nil +} + +func containerStartedAt(pod *corev1.Pod, name string) (time.Time, bool) { + for _, status := range pod.Status.ContainerStatuses { + if status.Name == name && status.State.Running != nil { + return status.State.Running.StartedAt.Time, !status.State.Running.StartedAt.IsZero() + } + } + return time.Time{}, false +} + func (c *Client) waitPodDeleted(ctx context.Context, namespace, podName string, timeout time.Duration) error { return wait.PollUntilContextTimeout(ctx, time.Second, timeout, true, func(ctx context.Context) (bool, error) { _, err := c.cs.CoreV1().Pods(namespace).Get(ctx, podName, metav1.GetOptions{}) diff --git a/tests/perf/byoo-otel-collector/pkg/deploy/deploy_test.go b/tests/perf/byoo-otel-collector/pkg/deploy/deploy_test.go index 758d7184f..fe873cf42 100644 --- a/tests/perf/byoo-otel-collector/pkg/deploy/deploy_test.go +++ b/tests/perf/byoo-otel-collector/pkg/deploy/deploy_test.go @@ -20,6 +20,8 @@ package deploy import ( "context" "errors" + "fmt" + "strings" "testing" "time" @@ -156,6 +158,174 @@ func TestWaitPodReadyFailsFastOnCrashLoop(t *testing.T) { } } +func TestWaitCollectorHealthRecordsStartTimes(t *testing.T) { + origPoll, origProxy := healthPollInterval, proxyGet + t.Cleanup(func() { + healthPollInterval = origPoll + proxyGet = origProxy + }) + healthPollInterval = time.Millisecond + + podStartedAt := time.Now().Add(-8 * time.Second).UTC() + collectorStartedAt := podStartedAt.Add(2 * time.Second) + pod := &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{Name: "perf-collector", Namespace: "byoo-perf"}, + Status: corev1.PodStatus{ + Phase: corev1.PodRunning, + StartTime: &metav1.Time{Time: podStartedAt}, + ContainerStatuses: []corev1.ContainerStatus{{ + Name: "byoo-otel-collector", + State: corev1.ContainerState{Running: &corev1.ContainerStateRunning{ + StartedAt: metav1.Time{Time: collectorStartedAt}, + }}, + }}, + }, + } + var requests int + proxyGet = func(_ context.Context, _ kubernetes.Interface, namespace, name, port, path string) ([]byte, error) { + requests++ + if namespace != "byoo-perf" || name != "perf-collector" || port != collectorHealthPort || path != collectorHealthPath { + t.Fatalf("health request = %s/%s:%s%s", namespace, name, port, path) + } + if requests == 1 { + return nil, fmt.Errorf("collector is starting") + } + return []byte("ok"), nil + } + + c := NewClientForClientset(fake.NewSimpleClientset(pod)) + before := time.Now() + startup, err := c.WaitCollectorHealth(context.Background(), "byoo-perf", "perf-collector", "byoo-otel-collector", time.Second, time.Minute) + if err != nil { + t.Fatalf("WaitCollectorHealth: %v", err) + } + if requests != 2 { + t.Errorf("health requests = %d, want 2", requests) + } + if !startup.PodStartedAt.Equal(podStartedAt) || !startup.CollectorStartedAt.Equal(collectorStartedAt) { + t.Errorf("startup timestamps = %+v, want pod=%s collector=%s", startup, podStartedAt, collectorStartedAt) + } + if startup.HealthyAt.Before(before) { + t.Errorf("healthy at %s before wait began %s", startup.HealthyAt, before) + } + if startup.PodToHealthSeconds < 8 || startup.CollectorToHealthSeconds < 6 { + t.Errorf("startup durations = %+v, want at least pod=8s collector=6s", startup) + } +} + +func TestWaitCollectorHealthTimesOutWithoutContainerStart(t *testing.T) { + origPoll := healthPollInterval + t.Cleanup(func() { healthPollInterval = origPoll }) + healthPollInterval = time.Millisecond + + pod := &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{Name: "perf-collector", Namespace: "byoo-perf"}, + Status: corev1.PodStatus{ + Phase: corev1.PodPending, + StartTime: &metav1.Time{Time: time.Now()}, + }, + } + c := NewClientForClientset(fake.NewSimpleClientset(pod)) + _, err := c.WaitCollectorHealth(context.Background(), "byoo-perf", "perf-collector", "byoo-otel-collector", 10*time.Millisecond, time.Second) + if err == nil { + t.Fatal("expected WaitCollectorHealth to time out without a collector start time") + } + if !strings.Contains(err.Error(), "collector health endpoint") { + t.Errorf("timeout error = %v, want collector health endpoint context", err) + } +} + +func TestWaitCollectorHealthStopsAtStartupMaximum(t *testing.T) { + origPoll, origProxy := healthPollInterval, proxyGet + t.Cleanup(func() { + healthPollInterval = origPoll + proxyGet = origProxy + }) + healthPollInterval = time.Millisecond + + collectorStartedAt := time.Now().UTC() + pod := &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{Name: "perf-collector", Namespace: "byoo-perf"}, + Status: corev1.PodStatus{ + Phase: corev1.PodRunning, + StartTime: &metav1.Time{Time: collectorStartedAt.Add(-time.Second)}, + ContainerStatuses: []corev1.ContainerStatus{{ + Name: "byoo-otel-collector", + State: corev1.ContainerState{Running: &corev1.ContainerStateRunning{ + StartedAt: metav1.Time{Time: collectorStartedAt}, + }}, + }}, + }, + } + proxyGet = func(context.Context, kubernetes.Interface, string, string, string, string) ([]byte, error) { + return nil, errors.New("collector is still starting") + } + + c := NewClientForClientset(fake.NewSimpleClientset(pod)) + startedWaiting := time.Now() + _, err := c.WaitCollectorHealth(context.Background(), "byoo-perf", "perf-collector", "byoo-otel-collector", 250*time.Millisecond, 10*time.Millisecond) + if err == nil { + t.Fatal("expected WaitCollectorHealth to stop at the startup maximum") + } + if !strings.Contains(err.Error(), "startup maximum") { + t.Errorf("error = %v, want startup maximum context", err) + } + if elapsed := time.Since(startedWaiting); elapsed > 100*time.Millisecond { + t.Errorf("WaitCollectorHealth returned after %s, want it to stop near the 10ms startup maximum", elapsed) + } +} + +func TestWaitCollectorHealthRejectsLateHealthResponse(t *testing.T) { + origPoll, origProxy := healthPollInterval, proxyGet + t.Cleanup(func() { + healthPollInterval = origPoll + proxyGet = origProxy + }) + healthPollInterval = time.Millisecond + + const startupMax = 100 * time.Millisecond + collectorStartedAt := time.Now().UTC() + pod := &corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{Name: "perf-collector", Namespace: "byoo-perf"}, + Status: corev1.PodStatus{ + Phase: corev1.PodRunning, + StartTime: &metav1.Time{Time: collectorStartedAt.Add(-time.Second)}, + ContainerStatuses: []corev1.ContainerStatus{{ + Name: "byoo-otel-collector", + State: corev1.ContainerState{Running: &corev1.ContainerStateRunning{ + StartedAt: metav1.Time{Time: collectorStartedAt}, + }}, + }}, + }, + } + called := false + proxyGet = func(ctx context.Context, _ kubernetes.Interface, _, _, _, _ string) ([]byte, error) { + called = true + deadline, ok := ctx.Deadline() + if !ok { + t.Error("health request has no startup deadline") + } else if want := collectorStartedAt.Add(startupMax); !deadline.Equal(want) { + t.Errorf("health request deadline = %s, want %s", deadline, want) + } + if delay := time.Until(deadline) + time.Millisecond; delay > 0 { + time.Sleep(delay) + } + return []byte("ok"), nil + } + + c := NewClientForClientset(fake.NewSimpleClientset(pod)) + _, err := c.WaitCollectorHealth(context.Background(), "byoo-perf", "perf-collector", "byoo-otel-collector", time.Second, startupMax) + if !called { + t.Fatal("expected a health request before the startup deadline") + } + if err == nil { + t.Fatal("expected a late successful health response to fail") + } + if !strings.Contains(err.Error(), "startup maximum") { + t.Errorf("error = %v, want startup maximum context", err) + } +} + func TestCleanupScopedToLabel(t *testing.T) { c := NewClientForClientset(fake.NewSimpleClientset()) ctx := context.Background() diff --git a/tests/perf/byoo-otel-collector/pkg/report/report.go b/tests/perf/byoo-otel-collector/pkg/report/report.go index e23275479..bfff201a4 100644 --- a/tests/perf/byoo-otel-collector/pkg/report/report.go +++ b/tests/perf/byoo-otel-collector/pkg/report/report.go @@ -17,9 +17,8 @@ limitations under the License. // Package report turns collector and sink metric scrapes taken across a // measurement window into a performance baseline: per-signal throughput, drops, -// end-to-end delivery, collector resource usage, and pod health. It emits both a -// human-readable summary and structured JSON. There are no pass/fail thresholds -// yet; the goal is a reproducible baseline. +// end-to-end delivery, collector resource usage, pod health, and startup +// health. It emits both a human-readable summary and structured JSON. package report import ( @@ -54,6 +53,30 @@ type PodHealth struct { OOMKilled bool `json:"oom_killed"` } +// StartupHealth records when the pod and collector container started, and when +// the collector first returned a successful response from its health endpoint. +// PodToHealthSeconds includes pod startup and image-pull time. +// CollectorToHealthSeconds isolates collector initialization after its +// container started. +type StartupHealth struct { + PodStartedAt time.Time `json:"pod_started_at"` + CollectorStartedAt time.Time `json:"collector_started_at"` + HealthyAt time.Time `json:"healthy_at"` + PodToHealthSeconds float64 `json:"pod_to_health_seconds"` + CollectorToHealthSeconds float64 `json:"collector_to_health_seconds"` +} + +// NewStartupHealth constructs startup-health timestamps and derived durations. +func NewStartupHealth(podStartedAt, collectorStartedAt, healthyAt time.Time) StartupHealth { + return StartupHealth{ + PodStartedAt: podStartedAt, + CollectorStartedAt: collectorStartedAt, + HealthyAt: healthyAt, + PodToHealthSeconds: durationSeconds(podStartedAt, healthyAt), + CollectorToHealthSeconds: durationSeconds(collectorStartedAt, healthyAt), + } +} + // Snapshot is a set of metric scrapes taken at one instant. type Snapshot struct { At time.Time @@ -104,18 +127,19 @@ const ( // ShapeReport is the full baseline for one workload shape. type ShapeReport struct { - Shape string `json:"shape"` - Profile string `json:"profile"` - Run int `json:"run,omitempty"` - Repetitions int `json:"repetitions,omitempty"` - Status string `json:"status"` - FailureReason string `json:"failure_reason,omitempty"` - WindowSeconds float64 `json:"window_seconds"` - Logs SignalStat `json:"logs"` - Metrics SignalStat `json:"metrics"` - Resources ResourceStat `json:"resources"` - Health PodHealth `json:"health"` - Notes []string `json:"notes,omitempty"` + Shape string `json:"shape"` + Profile string `json:"profile"` + Run int `json:"run,omitempty"` + Repetitions int `json:"repetitions,omitempty"` + Status string `json:"status"` + FailureReason string `json:"failure_reason,omitempty"` + WindowSeconds float64 `json:"window_seconds"` + Logs SignalStat `json:"logs"` + Metrics SignalStat `json:"metrics"` + Resources ResourceStat `json:"resources"` + Health PodHealth `json:"health"` + StartupHealth *StartupHealth `json:"startup_health,omitempty"` + Notes []string `json:"notes,omitempty"` } // MarkInvalid flags the report as an invalid measurement with a reason, so @@ -133,6 +157,7 @@ type Inputs struct { MetricsPerSec int Window Window Health PodHealth + StartupHealth *StartupHealth // HealthErr, when non-nil, means pod health could not be observed. The // zero-value Health is then recorded as missing (note + partial) so a // report cannot read as healthy when health was never collected. @@ -148,6 +173,7 @@ func Build(in Inputs) ShapeReport { Profile: in.Profile, WindowSeconds: in.Window.Seconds(), Health: in.Health, + StartupHealth: in.StartupHealth, } win := r.WindowSeconds @@ -266,6 +292,12 @@ func (r ShapeReport) WriteSummary(w io.Writer) { writeSignal(w, "metrics", r.Metrics) fmt.Fprintf(w, " resources : cpu=%.3f cores (avg) mem_rss=%s\n", r.Resources.CPUCoresAvg, humanBytes(r.Resources.MemRSSBytes)) fmt.Fprintf(w, " health : phase=%s restarts=%d oom_killed=%t\n", r.Health.Phase, r.Health.Restarts, r.Health.OOMKilled) + if r.StartupHealth != nil { + fmt.Fprintf(w, " startup : pod_to_health=%s collector_to_health=%s\n", + humanDuration(r.StartupHealth.PodToHealthSeconds), + humanDuration(r.StartupHealth.CollectorToHealthSeconds), + ) + } if len(r.Notes) > 0 { fmt.Fprintf(w, " notes : missing metrics: ") for i, n := range r.Notes { @@ -298,3 +330,15 @@ func humanBytes(b float64) string { } return fmt.Sprintf("%.1f%s", val, units[i]) } + +func durationSeconds(start, end time.Time) float64 { + d := end.Sub(start).Seconds() + if d <= 0 { + return 0 + } + return d +} + +func humanDuration(seconds float64) string { + return time.Duration(seconds * float64(time.Second)).Round(time.Millisecond).String() +} diff --git a/tests/perf/byoo-otel-collector/pkg/report/report_test.go b/tests/perf/byoo-otel-collector/pkg/report/report_test.go index 919efa280..0d1532881 100644 --- a/tests/perf/byoo-otel-collector/pkg/report/report_test.go +++ b/tests/perf/byoo-otel-collector/pkg/report/report_test.go @@ -149,6 +149,47 @@ func TestBuildStatusReflectsCompleteness(t *testing.T) { } } +func TestStartupHealthIsReportedAndSerialized(t *testing.T) { + podStartedAt := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC) + collectorStartedAt := podStartedAt.Add(2 * time.Second) + healthyAt := collectorStartedAt.Add(15 * time.Second) + startup := NewStartupHealth(podStartedAt, collectorStartedAt, healthyAt) + + r := Build(Inputs{ + Shape: "container", + Profile: "dev", + LogsPerSec: 100, + MetricsPerSec: 120, + Window: window(), + Health: PodHealth{Phase: "Running"}, + StartupHealth: &startup, + }) + if r.StartupHealth == nil { + t.Fatal("startup health was not added to the report") + } + if r.StartupHealth.PodToHealthSeconds != 17 || r.StartupHealth.CollectorToHealthSeconds != 15 { + t.Errorf("startup durations = %+v, want pod=17s collector=15s", r.StartupHealth) + } + + data, err := r.JSON() + if err != nil { + t.Fatalf("JSON: %v", err) + } + var decoded ShapeReport + if err := json.Unmarshal(data, &decoded); err != nil { + t.Fatalf("unmarshal: %v", err) + } + if decoded.StartupHealth == nil || !decoded.StartupHealth.HealthyAt.Equal(healthyAt) { + t.Errorf("serialized startup health = %+v, want healthy at %s", decoded.StartupHealth, healthyAt) + } + + var buf bytes.Buffer + r.WriteSummary(&buf) + if !bytes.Contains(buf.Bytes(), []byte("collector_to_health=15s")) { + t.Errorf("summary missing startup duration:\n%s", buf.String()) + } +} + // A series present only in the end scrape cannot yield a window delta: it must // be reported as missing (zero + note + partial status), not as a full // process-lifetime counter.