diff --git a/README.md b/README.md index 2dfa068f6..96ca94a02 100644 --- a/README.md +++ b/README.md @@ -49,6 +49,33 @@ validation, worker-exit, and queue-overflow failures log `duplicate-mist-test failed` with connection identifiers. Protocol error details that could contain stream keys are redacted. +Prometheus metrics are available on the existing internal `/metrics` endpoint +(`http://127.0.0.1:39090/metrics` by default; configure `--http-internal-addr` +or `SP_HTTP_INTERNAL_ADDR`). Keep this admin listener private. + +| Metric | Meaning | +| -------------------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------- | +| `streamplace_duplicate_mist_verified_segments_total` | Successfully signed and verified segments, reported live even if the session later fails. | +| `streamplace_duplicate_mist_workers` | Active shadow workers, including workers draining after publisher EOF. | +| `streamplace_duplicate_mist_sessions_total{result="success"}` | Publisher EOF followed by a clean worker exit with at least one verified segment. | +| `streamplace_duplicate_mist_sessions_total{result="failure"}` | A running shadow failed, including parser/ingest/validation errors, overflow, unexpected exit, or drain timeout. | +| `streamplace_duplicate_mist_sessions_total{result="canceled"}` | Node shutdown canceled the shadow without an earlier failure. | +| `streamplace_duplicate_mist_sessions_total{result="skipped"}` | The worker cap or a startup error prevented shadowing; Mist forwarding continued. | + +For rollout confidence, watch live verification and completed outcomes: + +```promql +sum(rate(streamplace_duplicate_mist_verified_segments_total[5m])) +sum(increase(streamplace_duplicate_mist_sessions_total{result="success"}[1h])) +sum(increase(streamplace_duplicate_mist_sessions_total{result=~"failure|skipped"}[1h])) +``` + +Long-running streams only increment the success counter after disconnect. +An active worker alone is not evidence of successful ingest, and skipped +sessions are coverage gaps. All four outcome series exist at zero even when +shadowing is disabled; counters reset on node restart. Labels contain no +publisher identifiers or stream keys; use the failure logs for details. + The shadow has an 8 MiB per-connection buffer and a cap of 16 workers per addon listener, including draining workers. At capacity, new connections forward only to Mist and log a shadow failure. Shadow input idle for 10 diff --git a/pkg/api/duplicate_mist.go b/pkg/api/duplicate_mist.go index f5fa997dc..04f7d5b16 100644 --- a/pkg/api/duplicate_mist.go +++ b/pkg/api/duplicate_mist.go @@ -46,7 +46,9 @@ func (r shadowIdleReader) Read(p []byte) (int, error) { // the shadow's S1, so the publisher's response to Mist's handshake is valid. // All shadow replies are discarded; only Mist negotiates with the publisher. // This runs ONLY in a child process: no auth, database, storage, or live state. -func RunDuplicateMistWorker(parent context.Context, input io.Reader) (retErr error) { +// progress is the private machine channel back to the parent (stdout in the +// child): exactly one byte of value 1 per validated segment, nothing else. +func RunDuplicateMistWorker(parent context.Context, input io.Reader, progress io.Writer) (retErr error) { ctx, cancel := context.WithCancel(parent) defer cancel() ctx = log.WithLogValues(ctx, "component", "duplicate-mist-test") @@ -158,11 +160,21 @@ func RunDuplicateMistWorker(parent context.Context, input io.Reader) (retErr err }) var verified int var lastProgress time.Time + report := [1]byte{1} g.Go(func() error { err := media.RTMPIngestUnpublished(groupCtx, &config.CLI{}, "rtmp://"+ln.Addr().String()+"/live/"+ms.Streamer(), ms, func(c context.Context, segment []byte) error { if _, err := media.ValidateMP4Media(c, segment); err != nil { return fmt.Errorf("shadow signed segment validation: %w", err) } + // The parent counts only reported segments; a report it never + // received must not end as a clean worker exit. + n, err := progress.Write(report[:]) + if err == nil && n != len(report) { + err = io.ErrShortWrite + } + if err != nil { + return fmt.Errorf("shadow progress report: %w", err) + } verified++ if time.Since(lastProgress) >= 30*time.Second { log.Log(c, "duplicate-mist-test segment verified", "segments_verified", verified) diff --git a/pkg/api/duplicate_mist_test.go b/pkg/api/duplicate_mist_test.go index 9525a5dfe..df228d27b 100644 --- a/pkg/api/duplicate_mist_test.go +++ b/pkg/api/duplicate_mist_test.go @@ -15,7 +15,7 @@ func TestDuplicateMistIdleHandshakeClosesOnlyShadowInput(t *testing.T) { defer input.Close() defer publisher.Close() result := make(chan error, 1) - go func() { result <- RunDuplicateMistWorker(t.Context(), input) }() + go func() { result <- RunDuplicateMistWorker(t.Context(), input, io.Discard) }() select { case err := <-result: require.ErrorIs(t, err, errShadowIdleTimeout) @@ -33,7 +33,7 @@ func TestDuplicateMistRejectsTruncatedHandshake(t *testing.T) { "truncated plain handshake": {[]byte{3, 0, 0}, io.ErrUnexpectedEOF}, } { t.Run(name, func(t *testing.T) { - err := RunDuplicateMistWorker(t.Context(), bytes.NewReader(tc.input)) + err := RunDuplicateMistWorker(t.Context(), bytes.NewReader(tc.input), io.Discard) require.ErrorIs(t, err, tc.want) }) } diff --git a/pkg/cmd/duplicate_mist_test.go b/pkg/cmd/duplicate_mist_test.go index 0df3d203a..ebc2725b0 100644 --- a/pkg/cmd/duplicate_mist_test.go +++ b/pkg/cmd/duplicate_mist_test.go @@ -10,17 +10,19 @@ import ( "io" "log/slog" "net" + "net/http/httptest" "os" "os/exec" - "regexp" - "strconv" "sync" + "sync/atomic" "testing" "time" "github.com/bluenviron/gortmplib" "github.com/bluenviron/gortsplib/v5/pkg/format" "github.com/go-gst/go-gst/gst" + "github.com/prometheus/client_golang/prometheus/promhttp" + scraper "github.com/starttoaster/prometheus-exporter-scraper" "github.com/stretchr/testify/require" "stream.place/streamplace/pkg/config" "stream.place/streamplace/pkg/gstinit" @@ -53,13 +55,84 @@ func (w *shadowExitWriter) Write(p []byte) (int, error) { return w.Writer.Write(p) } +// zeroFillWriter forwards bytes intact until armed, then replaces every byte +// with zero. A zeroed chunk header is an RTMP message of invalid type 0, so +// the shadow parser fails deterministically; the byte count the primary +// receives is unchanged. +type zeroFillWriter struct { + io.Writer + armed atomic.Bool + intact atomic.Int64 // bytes forwarded before the first zeroed byte +} + +func (w *zeroFillWriter) Write(p []byte) (int, error) { + if w.armed.Load() { + return w.Writer.Write(make([]byte, len(p))) + } + w.intact.Add(int64(len(p))) + return w.Writer.Write(p) +} + +// shadowMetrics is the parent's view of the shadow, scraped over HTTP. +type shadowMetrics struct { + sessions map[string]int + verified int + workers float64 +} + +func scrapeShadowMetrics(t *testing.T, scrp *scraper.WebScraper) shadowMetrics { + t.Helper() + data, err := scrp.ScrapeWeb() + require.NoError(t, err) + m := shadowMetrics{sessions: map[string]int{}} + for _, c := range data.Counters { + switch c.Key { + case "streamplace_duplicate_mist_sessions_total": + m.sessions[c.Labels["result"]] = c.Value + case "streamplace_duplicate_mist_verified_segments_total": + m.verified = c.Value + } + } + for _, g := range data.Gauges { + if g.Key == "streamplace_duplicate_mist_workers" { + m.workers = g.Value + } + } + return m +} + +func waitForVerifiedSegment(ctx context.Context, t *testing.T, scrp *scraper.WebScraper) shadowMetrics { + t.Helper() + for { + m := scrapeShadowMetrics(t, scrp) + if m.verified > 0 { + return m + } + select { + case <-ctx.Done(): + t.Fatalf("shadow never reported a verified segment: %+v", m) + case <-time.After(100 * time.Millisecond): + } + } +} + +// Every outcome series must exist, so a zero is a zero and not a missing one. +func sessionCounts(success, failure int) map[string]int { + return map[string]int{"success": success, "failure": failure, "canceled": 0, "skipped": 0} +} + // An independently negotiated primary RTMP session receives all media while // the client-only replay passes the native relay, mp4mux, signer and verifier. -// Run the scenario in its own process so logging and GStreamer state remain -// isolated from the rest of the package suite. +// Run the scenario in its own process so logging, GStreamer state and the +// global metrics remain isolated from the rest of the package suite. +// +// paced streams in real time, burst sends everything at once to exercise the +// downstream drain, and corrupt zero-fills the publisher's bytes once the +// shadow has verified a segment: the shadow parser fails while the primary +// keeps receiving the stream. func TestDuplicateMistEndToEnd(t *testing.T) { if os.Getenv("DUPLICATE_MIST_E2E_SCENARIO") != "1" { - for _, mode := range []string{"paced", "burst"} { + for _, mode := range []string{"paced", "burst", "corrupt"} { t.Run(mode, func(t *testing.T) { ctx, cancel := context.WithTimeout(t.Context(), 60*time.Second) defer cancel() @@ -69,29 +142,33 @@ func TestDuplicateMistEndToEnd(t *testing.T) { cmd.Env = append(os.Environ(), "DUPLICATE_MIST_E2E_SCENARIO=1", "DUPLICATE_MIST_E2E_MODE="+mode) out, err := cmd.CombinedOutput() require.NoError(t, err, "%s", out) - // Every GoP, including buffered media at publisher EOF. The - // unpaced burst specifically exercises downstream drain. - match := regexp.MustCompile(`duplicate-mist-test shadow finished.*segments_verified=(\d+)`).FindSubmatch(out) - require.Len(t, match, 2, "%s", out) - verified, err := strconv.Atoi(string(match[1])) - require.NoError(t, err) - require.Equal(t, 10, verified, "%s", out) - require.NotContains(t, string(out), "duplicate-mist-test failed", "%s", out) t.Logf("%s", out) }) } return } + mode := os.Getenv("DUPLICATE_MIST_E2E_MODE") + corrupt := mode == "corrupt" + paced := mode != "burst" _ = flag.Set("v", "3") shadowExit := &shadowExitWriter{Writer: os.Stderr, done: make(chan struct{})} slog.SetDefault(slog.New(slog.NewTextHandler(shadowExit, nil))) ctx, cancel := context.WithTimeout(t.Context(), 40*time.Second) defer cancel() + // The parent's collectors are in this process's default registry; scrape + // them the way production is scraped, through promhttp over HTTP. + metricsSrv := httptest.NewServer(promhttp.Handler()) + defer metricsSrv.Close() + scrp, err := scraper.NewWebScraper(metricsSrv.URL) + require.NoError(t, err) + require.Equal(t, shadowMetrics{sessions: sessionCounts(0, 0)}, scrapeShadowMetrics(t, scrp)) + primary, err := net.Listen("tcp", "127.0.0.1:0") require.NoError(t, err) defer primary.Close() primaryResult := make(chan error, 1) + var primaryBytes atomic.Int64 go func() { conn, err := primary.Accept() if err != nil { @@ -109,6 +186,14 @@ func TestDuplicateMistEndToEnd(t *testing.T) { primaryResult <- err return } + if corrupt { + // Mist is only a byte sink here: it must keep receiving the + // publisher after the shadow's copy has turned to garbage. + n, err := io.Copy(io.Discard, conn) + primaryBytes.Store(n) + primaryResult <- err + return + } r := &gortmplib.Reader{Conn: sc} if err := r.Initialize(); err != nil { primaryResult <- err @@ -153,6 +238,7 @@ func TestDuplicateMistEndToEnd(t *testing.T) { require.NoError(t, err) defer bridge.Close() bridgeDone := make(chan error, 1) + zeroFill := &zeroFillWriter{} go func() { client, err := bridge.Accept() if err != nil { @@ -178,7 +264,8 @@ func TestDuplicateMistEndToEnd(t *testing.T) { defer stop() replies := make(chan struct{}) go func() { _, _ = io.Copy(client, server); client.Close(); close(replies) }() - _, err = io.Copy(server, client) + zeroFill.Writer = server + _, err = io.Copy(zeroFill, client) server.Close() <-replies bridgeDone <- err @@ -188,12 +275,27 @@ func TestDuplicateMistEndToEnd(t *testing.T) { pipeline, err := gst.NewPipelineFromString(fmt.Sprintf( "flvmux name=mux streamable=true ! rtmp2sink sync=%t location=rtmp://%s/live/shadow-e2e "+ "videotestsrc num-buffers=300 ! video/x-raw,width=320,height=240,framerate=30/1 ! x264enc key-int-max=30 bframes=0 tune=zerolatency ! h264parse ! queue ! mux.video "+ - "audiotestsrc num-buffers=470 samplesperbuffer=1024 ! audio/x-raw,rate=48000 ! audioconvert ! fdkaacenc ! aacparse ! queue ! mux.audio", os.Getenv("DUPLICATE_MIST_E2E_MODE") != "burst", bridge.Addr())) + "audiotestsrc num-buffers=470 samplesperbuffer=1024 ! audio/x-raw,rate=48000 ! audioconvert ! fdkaacenc ! aacparse ! queue ! mux.audio", paced, bridge.Addr())) require.NoError(t, err) defer pipeline.SetState(gst.StateNull) //nolint:errcheck busDone := make(chan error, 1) go func() { busDone <- media.HandleBusMessages(ctx, pipeline) }() require.NoError(t, pipeline.SetState(gst.StatePlaying)) + if paced { + // Progress must be visible while the publisher is still sending, + // not only once the worker has exited. + live := waitForVerifiedSegment(ctx, t, scrp) + select { + case err := <-busDone: + t.Fatalf("publisher finished before the shadow reported progress: %v", err) + default: + } + require.Equal(t, float64(1), live.workers) + require.Equal(t, sessionCounts(0, 0), live.sessions) + if corrupt { + zeroFill.armed.Store(true) + } + } select { case err := <-busDone: require.NoError(t, err) @@ -220,6 +322,21 @@ func TestDuplicateMistEndToEnd(t *testing.T) { case <-ctx.Done(): t.Fatal("shadow failed to exit after publisher EOF") } + // The session's one terminal outcome is recorded before the exit log. + final := scrapeShadowMetrics(t, scrp) + require.Equal(t, float64(0), final.workers) + if corrupt { + require.Equal(t, sessionCounts(0, 1), final.sessions) + require.Positive(t, final.verified) + // primaryBytes excludes the few KB consumed by its own handshake and + // publish exchange; seconds of media follow the zero-fill point. + intact := zeroFill.intact.Load() + require.Positive(t, intact) + require.Greater(t, primaryBytes.Load(), intact, "primary stopped receiving once the shadow failed") + } else { + require.Equal(t, sessionCounts(1, 0), final.sessions) + require.Equal(t, 10, final.verified) + } cancel() require.NoError(t, <-addonDone) } diff --git a/pkg/cmd/streamplace.go b/pkg/cmd/streamplace.go index fe0c9d311..fd9302a39 100644 --- a/pkg/cmd/streamplace.go +++ b/pkg/cmd/streamplace.go @@ -1025,6 +1025,7 @@ func makeStreamCommand(build *config.BuildFlags) *urfavecli.Command { } // The shadow uses an ephemeral signer and has no node, database, or storage. +// stdout carries only the per-segment progress bytes; all logs go to stderr. func makeDuplicateMistWorkerCommand() *urfavecli.Command { return &urfavecli.Command{ Name: "duplicate-mist-worker", @@ -1033,7 +1034,7 @@ func makeDuplicateMistWorkerCommand() *urfavecli.Command { Action: func(ctx context.Context, cmd *urfavecli.Command) error { _ = flag.Set("logtostderr", "true") _ = flag.Set("v", "3") - return api.RunDuplicateMistWorker(ctx, os.Stdin) + return api.RunDuplicateMistWorker(ctx, os.Stdin, os.Stdout) }, } } diff --git a/pkg/rtmps/shadow.go b/pkg/rtmps/shadow.go index 0bf805bd9..3e3b08e93 100644 --- a/pkg/rtmps/shadow.go +++ b/pkg/rtmps/shadow.go @@ -3,6 +3,7 @@ package rtmps import ( "bytes" "context" + "errors" "io" "os" "os/exec" @@ -13,8 +14,19 @@ import ( "time" "stream.place/streamplace/pkg/log" + "stream.place/streamplace/pkg/spmetrics" ) +// Cached so per-segment and per-session accounting never looks up labels. +var ( + shadowSuccess = spmetrics.DuplicateMistSessionsTotal.WithLabelValues("success") + shadowFailure = spmetrics.DuplicateMistSessionsTotal.WithLabelValues("failure") + shadowCanceled = spmetrics.DuplicateMistSessionsTotal.WithLabelValues("canceled") + shadowSkipped = spmetrics.DuplicateMistSessionsTotal.WithLabelValues("skipped") +) + +var errShadowProgress = errors.New("shadow worker wrote invalid progress output") + const ( shadowTag = "duplicate-mist-test" // Hidden subcommand that runs the shadow parser; reads the raw client @@ -33,6 +45,15 @@ const ( shadowWriteChunk = 64 << 10 shadowMaxLogLine = 64 << 10 + + // The only byte a worker may write to stdout, once per validated segment. + shadowProgressByte = 1 +) + +const ( + shadowRunning uint32 = iota + shadowFailed + shadowReaped ) type shadowOptions struct { @@ -60,8 +81,9 @@ type shadow struct { exitGrace time.Duration teed atomic.Int64 + verified atomic.Int64 finished atomic.Bool - aborted atomic.Bool + state atomic.Uint32 done chan struct{} } @@ -82,6 +104,7 @@ func startShadow(ctx context.Context, opts shadowOptions, connID uint64, remote parentRelease() } default: + shadowSkipped.Inc() log.Error(ctx, shadowTag+" failed: shadow worker limit reached, forwarding only to Mist", "max_workers", cap(opts.slots)) release() return nil @@ -92,6 +115,7 @@ func startShadow(ctx context.Context, opts shadowOptions, connID uint64, remote var err error exe, err = os.Executable() if err != nil { + shadowSkipped.Inc() log.Error(ctx, shadowTag+" failed: cannot locate executable for shadow worker, shadow disabled for this connection", "error", err) release() return nil @@ -111,33 +135,36 @@ func startShadow(ctx context.Context, opts shadowOptions, connID uint64, remote } wctx, cancel := context.WithCancel(ctx) + s := &shadow{ + ctx: wctx, + cancel: cancel, + exitGrace: grace, + done: make(chan struct{}), + } cmd := exec.CommandContext(wctx, exe, args...) cmd.Env = workerEnv(os.Environ()) cmd.WaitDelay = shadowWaitDelay - stdout := &lineLogger{ctx: ctx, stream: "stdout"} stderr := &lineLogger{ctx: ctx, stream: "stderr"} - cmd.Stdout = stdout + // Worker stdout is not a log: it is the private progress channel. + cmd.Stdout = &shadowProgress{ctx: ctx, s: s} cmd.Stderr = stderr stdin, err := cmd.StdinPipe() if err != nil { cancel() + shadowSkipped.Inc() log.Error(ctx, shadowTag+" failed: cannot create shadow worker stdin, shadow disabled for this connection", "error", err) release() return nil } if err := cmd.Start(); err != nil { cancel() + shadowSkipped.Inc() log.Error(ctx, shadowTag+" failed: cannot start shadow worker, shadow disabled for this connection", "error", err) release() return nil } - s := &shadow{ - ctx: wctx, - cancel: cancel, - ring: newRing(ringSize), - exitGrace: grace, - done: make(chan struct{}), - } + s.ring = newRing(ringSize) + spmetrics.DuplicateMistWorkers.Inc() log.Log(ctx, shadowTag+": shadow worker started", "pid", cmd.Process.Pid) writerDone := make(chan struct{}) @@ -148,15 +175,32 @@ func startShadow(ctx context.Context, opts shadowOptions, connID uint64, remote go func() { defer close(s.done) defer release() + // Wait also drains the progress pipe, so verified is final after it. err := cmd.Wait() - stdout.flush() stderr.flush() if err != nil || !s.finished.Load() { s.fail(ctx, "shadow worker exited unexpectedly", "error", err) + } else if s.verified.Load() == 0 { + s.fail(ctx, "shadow worker exited without validating any segment") } // Process is gone; unblock the pump if it is stuck in a write. s.cancel() <-writerDone + // The one terminal outcome of this session. Every other failure path + // (overflow, pump error, drain timeout, bad progress bytes) only + // aborts the shadow; it is counted here, once, as the worker is reaped. + // Freeze the outcome before recording it. A late grace timer must not + // log a failure after this session has already counted as successful. + failed := s.state.Swap(shadowReaped) == shadowFailed + switch { + case failed: + shadowFailure.Inc() + case ctx.Err() != nil: + shadowCanceled.Inc() + default: + shadowSuccess.Inc() + } + spmetrics.DuplicateMistWorkers.Dec() // The primary connection and grace timer may outlive this worker. // Release its large FIFO before admitting another worker. s.ring.mu.Lock() @@ -164,7 +208,7 @@ func startShadow(ctx context.Context, opts shadowOptions, connID uint64, remote s.ring.buf = nil s.ring.r, s.ring.n = 0, 0 s.ring.mu.Unlock() - log.Log(ctx, shadowTag+": shadow worker exited", "error", err, "aborted", s.aborted.Load(), "bytes_teed", s.teed.Load()) + log.Log(ctx, shadowTag+": shadow worker exited", "error", err, "aborted", failed, "bytes_teed", s.teed.Load(), "segments_verified", s.verified.Load()) }() return s } @@ -198,7 +242,7 @@ func (s *shadow) pump(ctx context.Context, stdin io.WriteCloser) { // is too far behind, the whole shadow is aborted; the stream is never // continued after a gap. func (s *shadow) tee(ctx context.Context, p []byte) { - if s == nil || s.aborted.Load() { + if s == nil || s.state.Load() != shadowRunning { return } if !s.ring.write(p) { @@ -230,12 +274,40 @@ func (s *shadow) fail(ctx context.Context, msg string, args ...any) { s.cancel() return } - if s.aborted.CompareAndSwap(false, true) { + if s.state.CompareAndSwap(shadowRunning, shadowFailed) { log.Error(ctx, shadowTag+" failed: "+msg, append(args, "bytes_teed", s.teed.Load())...) s.cancel() } } +// shadowProgress is the worker's stdout, a private machine channel: exactly +// one shadowProgressByte per segment the worker validated. Anything else +// means the channel can't be trusted, so it aborts the shadow rather than +// letting a session count as a clean success. +type shadowProgress struct { + ctx context.Context + s *shadow +} + +func (p *shadowProgress) Write(b []byte) (int, error) { + for i, c := range b { + if c != shadowProgressByte { + p.report(i) + p.s.fail(p.ctx, "shadow worker wrote invalid progress output", "byte", c) + return i, errShadowProgress + } + } + p.report(len(b)) + return len(b), nil +} + +func (p *shadowProgress) report(n int) { + if n > 0 { + p.s.verified.Add(int64(n)) + spmetrics.DuplicateMistVerifiedSegmentsTotal.Add(float64(n)) + } +} + // forwardClientToMist is io.Copy from the client to Mist that also tees every // byte, handshake included, to sh (which may be nil). func forwardClientToMist(ctx context.Context, mist io.Writer, client io.Reader, sh *shadow) (int64, error) { diff --git a/pkg/rtmps/shadow_test.go b/pkg/rtmps/shadow_test.go index ff5fb5334..1594768c8 100644 --- a/pkg/rtmps/shadow_test.go +++ b/pkg/rtmps/shadow_test.go @@ -16,7 +16,11 @@ import ( "testing" "time" + "github.com/prometheus/client_golang/prometheus" + dto "github.com/prometheus/client_model/go" "github.com/stretchr/testify/require" + + "stream.place/streamplace/pkg/spmetrics" ) const mistReply = "REPLY-FROM-MIST" @@ -44,6 +48,20 @@ func TestShadowHelperProcess(t *testing.T) { _ = os.WriteFile(out, []byte(strings.Join(os.Environ(), "\n")), 0o644) _, _ = io.Copy(io.Discard, os.Stdin) os.Exit(0) + case "silent": + _, _ = io.Copy(io.Discard, os.Stdin) + os.Exit(0) + case "report": + // Stdout is the progress channel: one 0x01 per validated segment, the + // first while the client is still streaming. + _, _ = os.Stdout.Write([]byte{1}) + _, _ = io.Copy(io.Discard, os.Stdin) + _, _ = os.Stdout.Write([]byte{1, 1}) + os.Exit(0) + case "garbage": + _, _ = os.Stdout.Write([]byte{1, 1, 'x', 1}) + _, _ = io.Copy(io.Discard, os.Stdin) + os.Exit(0) } os.Exit(2) } @@ -61,6 +79,41 @@ func randomBytes(n int) []byte { return b } +// The duplicate-mist collectors are process-global and shared with every +// other test, so assertions are deltas against a baseline. +type shadowMetrics struct { + success, failure, canceled, skipped, verified, workers float64 +} + +func readShadowMetrics(t *testing.T) shadowMetrics { + t.Helper() + val := func(m prometheus.Metric) float64 { + var d dto.Metric + require.NoError(t, m.Write(&d)) + return d.GetCounter().GetValue() + d.GetGauge().GetValue() + } + sessions := spmetrics.DuplicateMistSessionsTotal + return shadowMetrics{ + success: val(sessions.WithLabelValues("success")), + failure: val(sessions.WithLabelValues("failure")), + canceled: val(sessions.WithLabelValues("canceled")), + skipped: val(sessions.WithLabelValues("skipped")), + verified: val(spmetrics.DuplicateMistVerifiedSegmentsTotal), + workers: val(spmetrics.DuplicateMistWorkers), + } +} + +func (m shadowMetrics) since(base shadowMetrics) shadowMetrics { + return shadowMetrics{ + success: m.success - base.success, + failure: m.failure - base.failure, + canceled: m.canceled - base.canceled, + skipped: m.skipped - base.skipped, + verified: m.verified - base.verified, + workers: m.workers - base.workers, + } +} + func waitDone(t *testing.T, sh *shadow) { t.Helper() select { @@ -123,6 +176,7 @@ func runProxy(t *testing.T, ctx context.Context, sh *shadow, payload []byte) (mi func TestShadowAdmissionLimitLeavesMistRunningAndReusesSlot(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() + base := readShadowMetrics(t) opts, _ := helperOpts(t, "stall") opts.slots = make(chan struct{}, 1) opts.ringSize = 1024 @@ -130,6 +184,7 @@ func TestShadowAdmissionLimitLeavesMistRunningAndReusesSlot(t *testing.T) { require.NotNil(t, first) rejected := startShadow(ctx, opts, 2, "127.0.0.1:2") require.Nil(t, rejected, "a saturated addon must not spawn another worker") + require.Equal(t, shadowMetrics{skipped: 1, workers: 1}, readShadowMetrics(t).since(base)) payload := randomBytes(64 << 10) mistGot, clientGot := runProxy(t, ctx, rejected, payload) require.Equal(t, payload, mistGot) @@ -143,6 +198,7 @@ func TestShadowAdmissionLimitLeavesMistRunningAndReusesSlot(t *testing.T) { require.NotNil(t, next, "a reaped worker must release admission") nextCancel() waitDone(t, next) + require.Equal(t, shadowMetrics{canceled: 2, skipped: 1}, readShadowMetrics(t).since(base)) } func TestStalledShadowDoesNotAffectPrimary(t *testing.T) { @@ -150,6 +206,7 @@ func TestStalledShadowDoesNotAffectPrimary(t *testing.T) { defer cancel() opts, out := helperOpts(t, "stall") opts.ringSize = 256 << 10 + base := readShadowMetrics(t) sh := startShadow(ctx, opts, 1, "127.0.0.1:1") require.NotNil(t, sh) @@ -160,7 +217,7 @@ func TestStalledShadowDoesNotAffectPrimary(t *testing.T) { require.Equal(t, payload, mistGot) require.Equal(t, mistReply, string(clientGot)) - require.True(t, sh.aborted.Load(), "overflow must abort the shadow") + require.Equal(t, shadowMetrics{failure: 1}, readShadowMetrics(t).since(base)) pid := readPid(t, out) require.ErrorIs(t, syscall.Kill(pid, 0), syscall.ESRCH, "stalled worker must be killed and reaped") } @@ -169,6 +226,7 @@ func TestCrashedShadowDoesNotAffectPrimary(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() opts, _ := helperOpts(t, "crash") + base := readShadowMetrics(t) sh := startShadow(ctx, opts, 1, "127.0.0.1:1") require.NotNil(t, sh) @@ -178,7 +236,7 @@ func TestCrashedShadowDoesNotAffectPrimary(t *testing.T) { require.Equal(t, payload, mistGot) require.Equal(t, mistReply, string(clientGot)) - require.True(t, sh.aborted.Load()) + require.Equal(t, shadowMetrics{failure: 1}, readShadowMetrics(t).since(base), "crash, broken stdin pipe and bad exit are one failure") } func TestShadowHungAtStreamEndIsKilled(t *testing.T) { @@ -186,6 +244,7 @@ func TestShadowHungAtStreamEndIsKilled(t *testing.T) { defer cancel() opts, _ := helperOpts(t, "hang-on-eof") opts.exitGrace = 300 * time.Millisecond + base := readShadowMetrics(t) sh := startShadow(ctx, opts, 1, "127.0.0.1:1") require.NotNil(t, sh) @@ -193,7 +252,7 @@ func TestShadowHungAtStreamEndIsKilled(t *testing.T) { mistGot, _ := runProxy(t, ctx, sh, payload) require.Equal(t, payload, mistGot) waitDone(t, sh) - require.True(t, sh.aborted.Load(), "a worker that outlives the grace period must be killed") + require.Equal(t, shadowMetrics{failure: 1}, readShadowMetrics(t).since(base)) } func TestServerCancelReapsShadow(t *testing.T) { @@ -202,6 +261,7 @@ func TestServerCancelReapsShadow(t *testing.T) { opts, out := helperOpts(t, "stall") var released atomic.Int32 opts.release = func() { released.Add(1) } + base := readShadowMetrics(t) sh := startShadow(ctx, opts, 1, "127.0.0.1:1") require.NotNil(t, sh) sh.tee(ctx, []byte("some client bytes")) @@ -210,6 +270,7 @@ func TestServerCancelReapsShadow(t *testing.T) { cancel() waitDone(t, sh) require.Eventually(t, func() bool { return released.Load() == 1 }, 5*time.Second, 10*time.Millisecond) + require.Equal(t, shadowMetrics{canceled: 1}, readShadowMetrics(t).since(base), "server shutdown is neither a success nor a failure") require.ErrorIs(t, syscall.Kill(readPid(t, out), 0), syscall.ESRCH) } @@ -233,10 +294,12 @@ func TestShadowWorkerEnvStripsSPVars(t *testing.T) { func TestStartFailureLeavesPrimaryAlone(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() + base := readShadowMetrics(t) var released atomic.Int32 sh := startShadow(ctx, shadowOptions{exe: "/nonexistent/streamplace", release: func() { released.Add(1) }}, 1, "127.0.0.1:1") require.Nil(t, sh) require.Equal(t, int32(1), released.Load()) + require.Equal(t, shadowMetrics{skipped: 1}, readShadowMetrics(t).since(base)) payload := randomBytes(64 << 10) mistGot, clientGot := runProxy(t, ctx, sh, payload) @@ -244,6 +307,54 @@ func TestStartFailureLeavesPrimaryAlone(t *testing.T) { require.Equal(t, mistReply, string(clientGot)) } +// The worker's progress stream is what makes a session a success: the live +// reports must reach the parent while the client is still connected, and the +// session is counted exactly once, after the publisher's EOF. +func TestShadowCountsProgressLiveAndSuccessAtEOF(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + opts, _ := helperOpts(t, "report") + base := readShadowMetrics(t) + sh := startShadow(ctx, opts, 1, "127.0.0.1:1") + require.NotNil(t, sh) + + require.Eventually(t, func() bool { + return readShadowMetrics(t).since(base) == shadowMetrics{verified: 1, workers: 1} + }, 10*time.Second, 10*time.Millisecond, "a report must be counted before the publisher ends") + + sh.tee(ctx, randomBytes(1024)) + sh.finish(ctx) + waitDone(t, sh) + require.Equal(t, shadowMetrics{success: 1, verified: 3}, readShadowMetrics(t).since(base)) +} + +// A clean exit that never reported a validated segment proved nothing. +func TestShadowCleanExitWithoutProgressIsFailure(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + opts, _ := helperOpts(t, "silent") + base := readShadowMetrics(t) + sh := startShadow(ctx, opts, 1, "127.0.0.1:1") + require.NotNil(t, sh) + sh.finish(ctx) + waitDone(t, sh) + require.Equal(t, shadowMetrics{failure: 1}, readShadowMetrics(t).since(base)) +} + +// Anything but progress bytes on stdout makes the channel untrustworthy; only +// the valid prefix counts and the session can never be a success. +func TestShadowInvalidProgressOutputIsFailure(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + opts, _ := helperOpts(t, "garbage") + base := readShadowMetrics(t) + sh := startShadow(ctx, opts, 1, "127.0.0.1:1") + require.NotNil(t, sh) + sh.finish(ctx) + waitDone(t, sh) + require.Equal(t, shadowMetrics{failure: 1, verified: 2}, readShadowMetrics(t).since(base)) +} + func TestRingWrapsAndRejectsOverflow(t *testing.T) { r := newRing(1000) rng := rand.New(rand.NewSource(3)) diff --git a/pkg/spmetrics/spmetrics.go b/pkg/spmetrics/spmetrics.go index 358a2ba2b..a19883108 100644 --- a/pkg/spmetrics/spmetrics.go +++ b/pkg/spmetrics/spmetrics.go @@ -299,6 +299,28 @@ var S3MultipartBytesUploadedTotal = promauto.NewCounter(prometheus.CounterOpts{ Help: "total bytes written via pkg/s3.MultipartWriter", }) +// Duplicate-mist test: the shadow native ingest that runs beside Mist. One +// session is one duplication attempt, counted once by how it ended: +// success (publisher EOF, worker exited cleanly having validated at least +// one segment), failure, canceled (the server shut down first), or skipped +// (never started: addon at its worker limit, or the spawn failed). +var DuplicateMistSessionsTotal = promauto.NewCounterVec(prometheus.CounterOpts{ + Name: "streamplace_duplicate_mist_sessions_total", + Help: "duplicate-mist shadow sessions, by result (success|failure|canceled|skipped)", +}, []string{"result"}) + +// Counted in the parent as each validated segment is reported by the worker, +// so it climbs while the publisher is still streaming. +var DuplicateMistVerifiedSegmentsTotal = promauto.NewCounter(prometheus.CounterOpts{ + Name: "streamplace_duplicate_mist_verified_segments_total", + Help: "segments the duplicate-mist shadow workers parsed, signed and validated", +}) + +var DuplicateMistWorkers = promauto.NewGauge(prometheus.GaugeOpts{ + Name: "streamplace_duplicate_mist_workers", + Help: "duplicate-mist shadow worker subprocesses currently running (including ones draining after publisher EOF)", +}) + func ViewerInc(user string, protocol string) { go func() { viewersLock.Lock() @@ -362,6 +384,9 @@ func init() { for _, outcome := range []string{"clean", "crash"} { IngestWorkerExits.WithLabelValues("resumed", outcome) } + for _, result := range []string{"success", "failure", "canceled", "skipped"} { + DuplicateMistSessionsTotal.WithLabelValues(result) + } for _, backend := range []string{"file", "s3"} { VODProcessAttemptsTotal.WithLabelValues(backend) VODProcessSuccessesTotal.WithLabelValues(backend)