From e140ee36892efd1c013ad1eac04bf2219a7aecc8 Mon Sep 17 00:00:00 2001 From: zzylol Date: Thu, 16 Jul 2026 21:47:07 -0600 Subject: [PATCH] =?UTF-8?q?feat(gos):=20split=20Discipline=20B=20=E2=80=94?= =?UTF-8?q?=20retire=20alerting,=20keep=20rate=20reporting=20(edge=20side)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Discipline B's global-threshold alerting (Engine.Observe's value-baseline>= slack trigger -> sendReportLocked) retires per design-gos-unified-edge- telemetry.md §11: "the edge no longer makes alerting decisions itself." This edge never decides to fire an alert. obsCount/rate-tracking (feeding the coordinator's SampleP grant negotiation) is a genuinely separate concern and is preserved, but no longer piggybacks on the alerting trigger: Engine.Observe now reports on its own periodic cadence (ReportEveryN observations, decoupled from any value/slack threshold) instead of gating on a coordinator-granted slack budget. Removed the now-dead roundBaseline/grantedSlack state; Grant.LocalSlack/Spec.Tau stay on the wire (vestigial) for compatibility with the coordinator side. Companion to the ASAPQuery-backend coordinator-side split (data_plane::monitor retires Action::Alert and the CMY slack-countdown entirely, answering each report with a per-edge coordinated-sampling grant computed straight from that edge's own rate). Verified with a real cross-language run (Go e2edriver <-> Rust monitor_coordinator_harness over live gRPC): the coordinator answers the edge's periodic rate report with a genuine computed sample_p grant. Part of the GOS per-family stack (base: split/pr-retire-cms-point-query / #531). 🤖 Generated with Claude Code Co-Authored-By: Claude Sonnet 5 --- asap-precompute-go/monitor/engine.go | 104 ++++++----- asap-precompute-go/monitor/engine_test.go | 164 +++++++++--------- .../monitor/grpcclient/cmd/e2edriver/main.go | 19 +- asap-precompute-go/monitor/types.go | 88 +++++----- asap-precompute-go/monitor_linear_e2e_test.go | 99 +++++++---- deploy/mvp-multinode/scripts/monitor_e2e.sh | 33 ++-- 6 files changed, 274 insertions(+), 233 deletions(-) diff --git a/asap-precompute-go/monitor/engine.go b/asap-precompute-go/monitor/engine.go index 1030149f..b80a2de9 100644 --- a/asap-precompute-go/monitor/engine.go +++ b/asap-precompute-go/monitor/engine.go @@ -4,6 +4,15 @@ import ( "sync" ) +// ReportEveryN bounds how often a monitor's rate report fires: once on the +// very first observation of an epoch (so the coordinator gets an early, if +// noisy, rate signal), then every ReportEveryN observations after that. This +// is the "own periodic cadence" the 2026-07 Discipline B split moved rate +// reporting to (see the package doc comment) — an observation-count cadence +// rather than a wall-clock one, since Observe has no clock input and this +// keeps report volume O(obsCount/ReportEveryN) regardless of traffic shape. +const ReportEveryN = 64 + // monitorState is the per-(AggID,key) edge state for one monitoring epoch. type monitorState struct { aggID uint64 @@ -13,34 +22,28 @@ type monitorState struct { // for a different epoch triggers an epoch reset + re-register. windowStart uint64 - // roundBaseline is the value the edge most recently REPORTED to the - // coordinator (0 at epoch start). It is the coordinator's known value for - // this edge, so resetting it only on report — never on grant — keeps the - // edge and coordinator in lock-step and loses no mass when slack shrinks. - // The edge reports whenever lastValue - roundBaseline >= grantedSlack, then - // advances roundBaseline to the reported value. - roundBaseline float64 - // grantedSlack is the coordinator-allocated local budget for the current - // round. Zero means "no grant yet" — the edge stays silent until the first - // grant. A grant only updates this (and the round number); it never resets - // the baseline, so a shrinking slack simply lowers the bar for the next - // report. - grantedSlack float64 - // lastValue is the most recent local additive value observed. + // lastValue is the most recent local additive value observed. Still sent + // on each Report (informational — the coordinator no longer acts on it, + // alerting having retired), so a future backend consumer keeps a cheap + // per-report snapshot without a wire change. lastValue float64 // obsCount is the number of observations admitted for this monitor since // the current epoch began. It is the edge's observed items/window (rate) // reported to the coordinator so it can size this edge's sampling // probability via the whole-sketch ε-floor. Reset to 0 at each epoch boundary. obsCount uint64 + // reportedAt is the obsCount value at the last report, driving the + // ReportEveryN cadence below. + reportedAt uint64 // grantedSampleP is the coordinator-allocated distributed-NitroSketch // update-sampling probability for this monitor's agg. 0 (unset) ⇒ no // sampling grant (treated as p=1). It is stored on grant and read by the // precompute at the NEXT epoch boundary, where it is applied via WithSampleP - // to the new window's sketch wrapper (never mid-window). Unlike the slack - // fields it survives a round re-grant and is NOT cleared on epoch reset — - // the edge keeps sampling at the last granted p until told otherwise (a new - // grant or coordinator restart, which ForceReregister clears). + // to the new window's sketch wrapper (never mid-window). Unlike the + // per-epoch fields it survives a round re-grant and is NOT cleared on + // epoch reset — the edge keeps sampling at the last granted p until told + // otherwise (a new grant or coordinator restart, which ForceReregister + // clears). grantedSampleP float64 round uint64 @@ -92,13 +95,15 @@ func (e *Engine) SetReporter(r Reporter) { // Observe is the hot-path entry point: called once per admitted observation on // a monitored series, with the series' CURRENT additive value already read // cheaply by the caller (SumWrapper.Sum / CMSWrapper.EstimateCount / linear -// readout). value MUST be monotone non-decreasing within an epoch. +// readout). // -// On the first observation of an epoch it registers the monitor and stays -// silent (no slack yet). Once a grant has arrived, it reports whenever the -// increase since the last reported value reaches the granted slack, advancing -// the baseline to the reported value each time (so a single large jump yields -// one report and a shrinking slack lowers the bar for the next). +// On the first observation of an epoch it registers the monitor. Reporting +// (which carries obsCount as the edge's rate, feeding the coordinator's +// SampleP grant) fires on its own periodic cadence (ReportEveryN) — this +// edge never decides to alert on value, and never gates reporting on a +// coordinator-granted budget: alerting/query-time decisions belong entirely to +// the backend against its synced sketch state (design-gos-unified-edge- +// telemetry.md §11), not to this streaming protocol. func (e *Engine) Observe(aggID uint64, key []byte, value float64, windowStart uint64) { e.mu.Lock() defer e.mu.Unlock() @@ -129,22 +134,19 @@ func (e *Engine) Observe(aggID uint64, key []byte, value float64, windowStart ui } } - // No grant yet → stay silent (the coordinator drives the first round). - if st.grantedSlack <= 0 { - return - } - if value-st.roundBaseline >= st.grantedSlack { + // obsCount==1 (the very first observation of the epoch) always reports — + // see the doc comment above — everything after that follows the plain + // ReportEveryN cadence. + if st.obsCount == 1 || st.obsCount-st.reportedAt >= ReportEveryN { + st.reportedAt = st.obsCount st.seq++ - st.roundBaseline = value // advance baseline to the reported value e.sendReportLocked(st) } } -// OnGrant installs the current round's slack budget. Stale-epoch grants are -// dropped. It deliberately does NOT touch the baseline: the baseline tracks the -// last reported value (the coordinator's known value for this edge), so a -// shrinking slack simply lowers the bar for the next report without losing the -// unreported-but-below-slack mass. +// OnGrant installs the coordinator's latest coordinated-sampling grant. +// Stale-epoch grants are dropped. g.LocalSlack is no longer consulted +// (alerting retired) — only g.SampleP matters here. func (e *Engine) OnGrant(g Grant) { e.mu.Lock() defer e.mu.Unlock() @@ -153,7 +155,6 @@ func (e *Engine) OnGrant(g Grant) { return // unknown monitor or stale epoch } st.round = g.Round - st.grantedSlack = g.LocalSlack // Store the granted sampling probability for the precompute to read and // apply at the next epoch boundary. A grant always carries the coordinator's // current decision, so 0 means "no sampling this round" and is recorded as @@ -161,10 +162,11 @@ func (e *Engine) OnGrant(g Grant) { st.grantedSampleP = g.SampleP } -// OnPoll answers a poll with the current local value and advances the baseline -// to it (the coordinator now knows this value authoritatively). Stale-epoch -// polls are dropped. The single-phase v1 coordinator does not emit polls; this -// path exists for protocol completeness and future poll-based variants. +// OnPoll answers a poll with the current local value immediately, outside the +// normal ReportEveryN cadence. Stale-epoch polls are dropped. The v1 +// coordinator does not emit polls; this path exists for protocol completeness +// and future poll-based variants (e.g. a coordinator wanting an out-of-cadence +// rate refresh from one edge). func (e *Engine) OnPoll(p Poll) { e.mu.Lock() defer e.mu.Unlock() @@ -174,13 +176,12 @@ func (e *Engine) OnPoll(p Poll) { } st.round = p.Round st.seq++ - st.roundBaseline = st.lastValue + st.reportedAt = st.obsCount e.sendReportLocked(st) } -// OnClose just records the round number. Baselines advance on report, not on -// close, so there is nothing else to do. Stale-epoch closes are dropped. The -// v1 coordinator does not emit closes (a re-grant subsumes them). +// OnClose just records the round number. Stale-epoch closes are dropped. The +// v1 coordinator does not emit closes. func (e *Engine) OnClose(c Close) { e.mu.Lock() defer e.mu.Unlock() @@ -192,8 +193,8 @@ func (e *Engine) OnClose(c Close) { } // EpochReset is called by the runtime at every tumbling-window rotation. It -// resets every monitor to a fresh epoch (baseline/slack/round cleared) so the -// next window is an independent monitoring instance, and clears the registered +// resets every monitor to a fresh epoch (obsCount/round cleared) so the next +// window is an independent monitoring instance, and clears the registered // flag so the next Observe re-registers with the coordinator. func (e *Engine) EpochReset(newWindowStart uint64) { e.mu.Lock() @@ -206,17 +207,13 @@ func (e *Engine) EpochReset(newWindowStart uint64) { // ForceReregister clears the registered flag and round state for every monitor // (keeping the current epoch window and last observed value) so the next // Observe re-announces each monitor to the coordinator. The transport calls -// this on (re)connect: a coordinator that restarted has no memory of this edge, -// so re-registering with a zeroed baseline makes the edge re-report its full -// current value to the fresh coordinator. lastValue is preserved so the very -// next Observe can immediately re-report if it already exceeds a new grant. +// this on (re)connect: a coordinator that restarted has no memory of this +// edge, so re-registering lets it start granting this edge again. func (e *Engine) ForceReregister() { e.mu.Lock() defer e.mu.Unlock() for _, st := range e.states { st.registered = false - st.roundBaseline = 0 - st.grantedSlack = 0 st.grantedSampleP = 0 // a restarted coordinator has no sampling allocation for this edge st.round = 0 } @@ -242,10 +239,9 @@ func (e *Engine) GrantedSampleP(aggID uint64) float64 { func (e *Engine) resetStateLocked(st *monitorState, windowStart uint64) { st.windowStart = windowStart - st.roundBaseline = 0 - st.grantedSlack = 0 st.lastValue = 0 st.obsCount = 0 + st.reportedAt = 0 st.round = 0 st.registered = false // grantedSampleP is deliberately NOT reset: the coordinator's sampling diff --git a/asap-precompute-go/monitor/engine_test.go b/asap-precompute-go/monitor/engine_test.go index 7ab7568a..1793a665 100644 --- a/asap-precompute-go/monitor/engine_test.go +++ b/asap-precompute-go/monitor/engine_test.go @@ -38,76 +38,82 @@ func newTestEngine() (*Engine, *fakeReporter) { return NewEngine("edge-test", win, f), f } -func TestSilentBeforeGrant(t *testing.T) { +// TestFirstObservationRegistersAndReports checks that the very first +// observation of an epoch both registers the monitor AND fires an immediate +// report (an early, if noisy, rate signal) — no grant is needed first: +// alerting retired, so reporting is no longer gated on a coordinator-granted +// budget. +func TestFirstObservationRegistersAndReports(t *testing.T) { e, f := newTestEngine() - // Many observations, no grant yet → must stay silent (but register once). - for v := 1.0; v <= 100; v++ { - e.Observe(7, nil, v, win) - } - if len(f.reports) != 0 { - t.Fatalf("expected zero reports before any grant, got %d", len(f.reports)) - } + e.Observe(7, nil, 5, win) if len(f.regs) != 1 { t.Fatalf("expected exactly one registration, got %d", len(f.regs)) } + if len(f.reports) != 1 { + t.Fatalf("expected an immediate report on the first observation, got %d", len(f.reports)) + } if f.regs[0].AggID != 7 || f.regs[0].EpochWindowMs != win { t.Fatalf("registration fields wrong: %+v", f.regs[0]) } } -func TestReportsWhenSlackCrossed(t *testing.T) { +// TestReportsOnFixedCadence checks the ReportEveryN cadence: after the +// obsCount==1 report, the next report fires only once obsCount reaches +// ReportEveryN observations — reporting no longer depends on the observed +// VALUE at all (unlike the retired slack-crossing trigger). +func TestReportsOnFixedCadence(t *testing.T) { e, f := newTestEngine() - e.Observe(7, nil, 5, win) // register; baseline=0 (epoch start), no grant yet - e.OnGrant(Grant{AggID: 7, Round: 1, LocalSlack: 10, WindowStartMs: win}) - // baseline stays 0; report fires when value-0 >= 10. - e.Observe(7, nil, 8, win) // 8 < 10 → silent - if len(f.reports) != 0 { - t.Fatalf("reported too early: %d reports", len(f.reports)) - } - e.Observe(7, nil, 12, win) // 12 >= 10 → report; baseline advances to 12 - e.Observe(7, nil, 18, win) // 18-12=6 < 10 → silent + e.Observe(7, nil, 0, win) // obs #1 → report #1 + if len(f.reports) != 1 { + t.Fatalf("expected one report after the first observation, got %d", len(f.reports)) + } + for i := 2; i < ReportEveryN+1; i++ { + e.Observe(7, nil, float64(i), win) + } if len(f.reports) != 1 { - t.Fatalf("expected one report after first crossing, got %d", len(f.reports)) + t.Fatalf("expected still one report short of the cadence, got %d", len(f.reports)) } - e.Observe(7, nil, 23, win) // 23-12=11 >= 10 → second report; baseline=23 + e.Observe(7, nil, 999, win) // obsCount reaches ReportEveryN+1 → report #2 if len(f.reports) != 2 { - t.Fatalf("expected a second report once baseline+slack crossed again, got %d", len(f.reports)) + t.Fatalf("expected a second report once the cadence was reached, got %d", len(f.reports)) } r, _ := f.lastReport() - if r.LocalValue != 23 || r.Round != 1 || r.Seq != 2 { - t.Fatalf("report fields wrong: %+v", r) + if r.Rate != float64(ReportEveryN+1) { + t.Fatalf("second report Rate = %v, want %v", r.Rate, ReportEveryN+1) } } -func TestShrinkingSlackLowersBar(t *testing.T) { +// TestGrantOnlyUpdatesSampleP checks that OnGrant no longer perturbs the +// report cadence at all — it only records SampleP (LocalSlack is vestigial, +// kept solely for SlackGrant wire compatibility). +func TestGrantOnlyUpdatesSampleP(t *testing.T) { e, f := newTestEngine() - e.Observe(7, nil, 0, win) - e.OnGrant(Grant{AggID: 7, Round: 1, LocalSlack: 10, WindowStartMs: win}) - e.Observe(7, nil, 12, win) // report #1; baseline=12 - // Coordinator re-grants a SMALLER slack (no baseline reset on the edge). - e.OnGrant(Grant{AggID: 7, Round: 2, LocalSlack: 5, WindowStartMs: win}) - e.Observe(7, nil, 14, win) // 14-12=2 < 5 → silent - e.Observe(7, nil, 18, win) // 18-12=6 >= 5 → report #2; baseline=18 - if len(f.reports) != 2 { - t.Fatalf("expected 2 reports, got %d", len(f.reports)) + e.Observe(7, nil, 5, win) // report #1 + before := len(f.reports) + e.OnGrant(Grant{AggID: 7, Round: 1, LocalSlack: 999, WindowStartMs: win, SampleP: 0.4}) + if len(f.reports) != before { + t.Fatalf("OnGrant must not itself trigger a report, got %d reports", len(f.reports)) } - r, _ := f.lastReport() - if r.Round != 2 || r.LocalValue != 18 { - t.Fatalf("second report wrong: %+v", r) + if got := e.GrantedSampleP(7); got != 0.4 { + t.Fatalf("GrantedSampleP after grant = %v, want 0.4", got) } } -func TestPollProducesAuthoritativeReport(t *testing.T) { +// TestPollProducesImmediateReport checks that a Poll forces a report right +// away, independent of the ReportEveryN cadence. +func TestPollProducesImmediateReport(t *testing.T) { e, f := newTestEngine() - e.Observe(7, nil, 0, win) - e.OnGrant(Grant{AggID: 7, Round: 1, LocalSlack: 100, WindowStartMs: win}) - e.Observe(7, nil, 33, win) // below slack → no spontaneous report - if len(f.reports) != 0 { - t.Fatalf("unexpected spontaneous report") + e.Observe(7, nil, 0, win) // obs #1 → report #1 + if len(f.reports) != 1 { + t.Fatalf("expected one report after the first observation, got %d", len(f.reports)) } - e.OnPoll(Poll{AggID: 7, Round: 1, WindowStartMs: win}) + e.Observe(7, nil, 33, win) // obs #2, short of cadence → no new report if len(f.reports) != 1 { - t.Fatalf("poll did not produce a report") + t.Fatalf("unexpected spontaneous report before the cadence, got %d", len(f.reports)) + } + e.OnPoll(Poll{AggID: 7, Round: 1, WindowStartMs: win}) + if len(f.reports) != 2 { + t.Fatalf("poll did not produce an immediate report, got %d", len(f.reports)) } r, _ := f.lastReport() if r.LocalValue != 33 { @@ -117,35 +123,29 @@ func TestPollProducesAuthoritativeReport(t *testing.T) { func TestEpochResetReRegisters(t *testing.T) { e, f := newTestEngine() - e.Observe(7, nil, 0, win) - e.OnGrant(Grant{AggID: 7, Round: 1, LocalSlack: 5, WindowStartMs: win}) - e.Observe(7, nil, 10, win) // report in epoch 1 + e.Observe(7, nil, 0, win) // report in epoch 1 // New epoch via explicit reset (mirrors rotateLocked). next := win + win e.EpochReset(next) if len(f.regs) != 1 { t.Fatalf("re-register should be lazy (on next Observe), regs=%d", len(f.regs)) } - e.Observe(7, nil, 1, next) // re-register; baseline cleared; no grant yet → silent + e.Observe(7, nil, 1, next) // re-register; fresh epoch → immediate report if len(f.regs) != 2 { t.Fatalf("expected re-registration in new epoch, regs=%d", len(f.regs)) } if f.regs[1].EpochWindowMs != win { t.Fatalf("epoch window size wrong on re-register: %+v", f.regs[1]) } - // A grant from the OLD epoch must be ignored. - before := len(f.reports) - e.OnGrant(Grant{AggID: 7, Round: 1, LocalSlack: 1, WindowStartMs: win}) - e.Observe(7, nil, 100, next) - if len(f.reports) != before { - t.Fatalf("stale-epoch grant should not enable reporting") + if len(f.reports) != 2 { + t.Fatalf("expected a report in the new epoch too, got %d", len(f.reports)) } } func TestEpochChangeViaObserveResets(t *testing.T) { e, _ := newTestEngine() e.Observe(7, nil, 0, win) - e.OnGrant(Grant{AggID: 7, Round: 1, LocalSlack: 5, WindowStartMs: win}) + e.OnGrant(Grant{AggID: 7, Round: 1, WindowStartMs: win, SampleP: 0.5}) e.Observe(7, nil, 50, win) // Observe with a new windowStart triggers an in-line epoch reset. next := win + win @@ -153,30 +153,32 @@ func TestEpochChangeViaObserveResets(t *testing.T) { e.mu.Lock() st := e.states[mapKey{7, ""}] e.mu.Unlock() - if st.windowStart != next || st.grantedSlack != 0 || st.registered != true { + if st.windowStart != next || st.obsCount != 1 || st.registered != true { t.Fatalf("epoch change did not reset state: %+v", st) } } -func TestCMSPointKeyedMonitors(t *testing.T) { +// TestKeyedMonitorsAreIndependent checks that two monitors under the same +// AggID but different keys (e.g. two CMSPoint keys) get fully independent +// registration/report state. +func TestKeyedMonitorsAreIndependent(t *testing.T) { e, f := newTestEngine() ka, kb := []byte("svc=a"), []byte("svc=b") - e.Observe(9, ka, 0, win) - e.Observe(9, kb, 0, win) - e.OnGrant(Grant{AggID: 9, Key: ka, Round: 1, LocalSlack: 10, WindowStartMs: win}) - e.OnGrant(Grant{AggID: 9, Key: kb, Round: 1, LocalSlack: 10, WindowStartMs: win}) - e.Observe(9, ka, 20, win) // a crosses - e.Observe(9, kb, 5, win) // b does not - if len(f.reports) != 1 { - t.Fatalf("expected one report (only key a crossed), got %d", len(f.reports)) - } - r, _ := f.lastReport() - if string(r.Key) != "svc=a" { - t.Fatalf("report should be for key a, got %q", r.Key) + e.Observe(9, ka, 20, win) // report #1 for a + e.Observe(9, kb, 5, win) // report #1 for b + if len(f.reports) != 2 { + t.Fatalf("expected one report per key on first observation, got %d", len(f.reports)) } if len(f.regs) != 2 { t.Fatalf("expected two registrations (one per key), got %d", len(f.regs)) } + keys := map[string]bool{} + for _, r := range f.reports { + keys[string(r.Key)] = true + } + if !keys["svc=a"] || !keys["svc=b"] { + t.Fatalf("expected reports for both keys, got %+v", f.reports) + } } // TestOnGrantStoresSampleP checks that a grant's SampleP is recorded on the @@ -195,26 +197,26 @@ func TestOnGrantStoresSampleP(t *testing.T) { t.Fatalf("GrantedSampleP(unknown) = %v, want 1.0", got) } - e.OnGrant(Grant{AggID: 7, Round: 1, LocalSlack: 10, WindowStartMs: win, SampleP: 0.3}) + e.OnGrant(Grant{AggID: 7, Round: 1, WindowStartMs: win, SampleP: 0.3}) if got := e.GrantedSampleP(7); got != 0.3 { t.Fatalf("GrantedSampleP after grant = %v, want 0.3", got) } // A grant carrying SampleP=0 means "no sampling this round"; it is recorded // as such and reads back as the unsampled default. - e.OnGrant(Grant{AggID: 7, Round: 2, LocalSlack: 10, WindowStartMs: win, SampleP: 0}) + e.OnGrant(Grant{AggID: 7, Round: 2, WindowStartMs: win, SampleP: 0}) if got := e.GrantedSampleP(7); got != 1.0 { t.Fatalf("GrantedSampleP after SampleP=0 grant = %v, want 1.0 (treat-as-unset)", got) } } -// TestSampchPSurvivesEpochReset checks that the granted sampling probability is +// TestSamplePSurvivesEpochReset checks that the granted sampling probability is // preserved across an epoch boundary (so the new window keeps sampling at the // last granted p) but is cleared by ForceReregister (coordinator restart). func TestSamplePSurvivesEpochReset(t *testing.T) { e, _ := newTestEngine() e.Observe(7, nil, 1, win) - e.OnGrant(Grant{AggID: 7, Round: 1, LocalSlack: 10, WindowStartMs: win, SampleP: 0.25}) + e.OnGrant(Grant{AggID: 7, Round: 1, WindowStartMs: win, SampleP: 0.25}) e.EpochReset(win + win) // rotate to the next epoch if got := e.GrantedSampleP(7); got != 0.25 { @@ -226,19 +228,17 @@ func TestSamplePSurvivesEpochReset(t *testing.T) { } } -// TestReportCarriesRate checks that a report emitted on a slack crossing carries -// the edge's observed per-epoch item count as Report.Rate. +// TestReportCarriesRate checks that a report carries the edge's observed +// per-epoch item count as Report.Rate. func TestReportCarriesRate(t *testing.T) { e, f := newTestEngine() - e.Observe(7, nil, 5, win) // obs #1, register - e.OnGrant(Grant{AggID: 7, Round: 1, LocalSlack: 10, WindowStartMs: win}) - e.Observe(7, nil, 8, win) // obs #2, below slack → silent - e.Observe(7, nil, 12, win) // obs #3, crosses slack → report + e.Observe(7, nil, 5, win) // obs #1 → report #1, Rate=1 + e.Observe(7, nil, 8, win) // obs #2, short of cadence → no new report r, ok := f.lastReport() if !ok { - t.Fatalf("expected a report after crossing the slack") + t.Fatalf("expected a report after the first observation") } - if r.Rate != 3 { - t.Fatalf("report Rate = %v, want 3 (observed items this epoch)", r.Rate) + if r.Rate != 1 { + t.Fatalf("report Rate = %v, want 1 (observed items at report time)", r.Rate) } } diff --git a/asap-precompute-go/monitor/grpcclient/cmd/e2edriver/main.go b/asap-precompute-go/monitor/grpcclient/cmd/e2edriver/main.go index 4cf0d626..3865d994 100644 --- a/asap-precompute-go/monitor/grpcclient/cmd/e2edriver/main.go +++ b/asap-precompute-go/monitor/grpcclient/cmd/e2edriver/main.go @@ -1,12 +1,15 @@ -// Command e2edriver is the EDGE side of the cross-language CDM e2e -// (deploy/mvp-multinode/scripts/monitor_e2e.sh). It wires the REAL edge -// runtime — precompute.Precompute (Sum) + monitor.Engine + the gRPC +// Command e2edriver is the EDGE side of the cross-language coordinated- +// sampling e2e (deploy/mvp-multinode/scripts/monitor_e2e.sh). It wires the +// REAL edge runtime — precompute.Precompute (Sum) + monitor.Engine + the gRPC // grpcclient transport — against a running Rust monitor-coordinator harness, -// then feeds a monotone stream of observations whose running window-sum climbs -// past τ. The coordinator should grant slack, receive the edge's reports over -// the live bidi stream, and fire the global-threshold alert. +// then feeds a stream of observations. The engine reports its observed rate +// on its own periodic cadence (monitor.ReportEveryN observations — alerting +// retired, see the monitor package docs), and the coordinator should answer +// with a computed coordinated-sampling grant (SlackGrant.sample_p) over the +// live bidi stream. // // Usage: e2edriver [agg_id] [tau] [per_obs] [count] +// (tau is accepted for CLI/back-compat but unused — alerting retired.) package main import ( @@ -96,8 +99,8 @@ func main() { }) time.Sleep(30 * time.Millisecond) } - // Grace for the final report → alert round-trip. + // Grace for the final report → grant round-trip. time.Sleep(1 * time.Second) - fmt.Fprintf(os.Stderr, "e2edriver: fed %d observations of %g (sum=%g, tau=%g), dropped_reports=%d\n", + fmt.Fprintf(os.Stderr, "e2edriver: fed %d observations of %g (sum=%g, tau=%g unused), dropped_reports=%d\n", count, perObs, float64(count)*perObs, tau, client.DroppedReports()) } diff --git a/asap-precompute-go/monitor/types.go b/asap-precompute-go/monitor/types.go index 9de5ea4d..78c807b4 100644 --- a/asap-precompute-go/monitor/types.go +++ b/asap-precompute-go/monitor/types.go @@ -1,11 +1,20 @@ -// Package monitor implements the edge side of the continuous distributed -// monitoring (CDM) model — Discipline B from ASAPCollector -// docs/continuous-monitoring-tumbling-cost-analysis.md. The runtime already -// emits one sketch per tumbling window (Discipline A); this package adds the -// intra-window early-alert protocol: the edge holds a per-monitor slack budget -// granted by the coordinator and sends a report only when its LOCAL additive -// value climbs past that slack, staying silent otherwise. State resets at the -// tumbling boundary (one window = one independent monitoring epoch). +// Package monitor implements the edge side of coordinated update-sampling — +// what used to be called "Discipline B" in ASAPCollector +// docs/continuous-monitoring-tumbling-cost-analysis.md. +// +// Global-threshold ALERTING (the CMY slack-countdown: register → grant +// (slack, sample_p) → countdown → report → alert) is RETIRED as of the +// 2026-07 insert-time-GOS redesign (docs/design-gos-unified-edge-telemetry.md +// §11): "Alerting and any other query-time decision is made entirely at the +// backend against [the reconstructed sketch state]; the edge no longer makes +// alerting decisions itself." This edge never decides to fire an alert. +// +// What's left: the edge reports its per-epoch observation RATE (obsCount) on +// its own periodic cadence (Engine.Observe, decoupled from any value/slack +// threshold — see reportEveryN), and the coordinator answers with a +// coordinated-sampling grant (Grant.SampleP), the whole-sketch ε-floor +// p_i = 1/(1+ε²·rate_i). State resets at the tumbling boundary (one window = +// one independent monitoring epoch). // // The package is intentionally gRPC-free: the engine talks to the coordinator // through the Reporter / Inbound interfaces, which the nested @@ -15,27 +24,25 @@ package monitor import "fmt" -// Functional selects which additive readout of a series' sketch the monitor -// thresholds. v1 covers only ADDITIVE functionals, which are monotone -// non-decreasing within a tumbling window — the property the slack countdown -// relies on. +// Functional selects which additive readout of a series' sketch a monitor +// reports as Report.LocalValue — informational only since alerting retired +// (see the package doc comment); it no longer drives any threshold decision. +// Identity is still (AggID, Key): CmsPoint carries a point-frequency key, +// Sum/LinearBuckets are whole-stream (key=""). type Functional uint8 const ( - // FunctionalSum thresholds the running window-sum (SumWrapper.Sum). O(1). + // FunctionalSum reports the running window-sum (SumWrapper.Sum). O(1). FunctionalSum Functional = iota - // FunctionalCMSPoint thresholds a Count-Min point-frequency f(x) for a + // FunctionalCMSPoint reports a Count-Min point-frequency f(x) for a // fixed key x (CMSWrapper.EstimateCount). O(rows). FunctionalCMSPoint - // FunctionalLinearBuckets thresholds a non-negative linear functional over an + // FunctionalLinearBuckets reports a non-negative linear functional over an // additive sketch's buckets. In v1 the only sketch that implements it is - // DDSketch, where the monotone realization is a VALUE-RANGE COUNT: Coeffs - // carries the value bounds — Coeffs[0]=lo, Coeffs[1]=hi (optional, default - // +Inf) — and the readout is the count of samples whose bucket value is in - // [lo, hi] (e.g. "number of requests slower than 500ms"). This is an - // additive non-negative aggregate, hence monotone within a window. The - // signed/quantile-threshold variants the cost-analysis doc also describes are - // non-monotone and out of scope; Validate rejects negative coefficients. + // DDSketch, where the realization is a VALUE-RANGE COUNT: Coeffs carries + // the value bounds — Coeffs[0]=lo, Coeffs[1]=hi (optional, default +Inf) — + // and the readout is the count of samples whose bucket value is in + // [lo, hi] (e.g. "number of requests slower than 500ms"). FunctionalLinearBuckets ) @@ -52,23 +59,22 @@ func (f Functional) String() string { } } -// Spec is the per-AggID threshold-monitor configuration. It rides inside +// Spec is the per-AggID monitor configuration. It rides inside // PrecomputeConfig and is delivered to edges through the existing control -// channel. τ is advisory at the edge (used for diagnostics); the authoritative -// τ lives at the coordinator. +// channel. Tau is vestigial (alerting retired — see the package doc comment); +// Epsilon still feeds the coordinator's whole-sketch ε-floor sampling law. type Spec struct { Enabled bool Functional Functional Key []byte // CMS point-frequency key x (FunctionalCMSPoint) Coeffs []float64 // linear-functional coefficients (FunctionalLinearBuckets); non-negative CoordinatorURL string // edge→coordinator dial target - Tau float64 // advisory threshold (authoritative copy at coordinator) - Epsilon float64 // advisory relative tolerance + Tau float64 // unused (alerting retired); kept for wire/config-schema compatibility + Epsilon float64 // feeds the coordinator's whole-sketch ε-floor sampling law } -// Validate rejects specs whose functional would violate the within-window -// monotonicity the slack countdown requires (negative linear coefficients), or -// that are missing required fields. +// Validate rejects specs with malformed per-functional fields (e.g. a missing +// CMSPoint key, or negative LinearBuckets coefficients). func (s Spec) Validate() error { if !s.Enabled { return nil @@ -86,7 +92,7 @@ func (s Spec) Validate() error { } for i, c := range s.Coeffs { if c < 0 { - return fmt.Errorf("monitor: linear_buckets coeff[%d]=%g is negative; only non-negative (monotone) functionals are supported in v1", i, c) + return fmt.Errorf("monitor: linear_buckets coeff[%d]=%g is negative; only non-negative value-range bounds are supported", i, c) } } default: @@ -109,9 +115,10 @@ type Registration struct { WindowStartMs uint64 } -// Report is one outbound edge→coordinator message: the current local additive -// value for a monitor, sent when the local increase since round start crossed -// the granted slack, or in answer to a Poll at round close. +// Report is one outbound edge→coordinator message, sent on the engine's own +// periodic cadence (reportEveryN observations, decoupled from any value/slack +// threshold — see the package doc comment). LocalValue is informational only +// (alerting retired); Rate is what the coordinator actually acts on. type Report struct { EdgeID string AggID uint64 @@ -129,11 +136,14 @@ type Report struct { Rate float64 } -// Grant is the coordinator→edge per-round slack budget. +// Grant is the coordinator's reply to one Report: this edge's freshly +// computed coordinated-sampling grant. type Grant struct { AggID uint64 Key []byte Round uint64 + // LocalSlack is unused (alerting retired); kept for SlackGrant wire + // compatibility. The edge no longer gates anything on it. LocalSlack float64 WindowStartMs uint64 // SampleP is the distributed-NitroSketch update-sampling probability the @@ -141,13 +151,12 @@ type Grant struct { // (p_i = 1/(1+ε²·rate_i); see data_plane allocate_p / epsilon_sample_floor). // 0 (unset) ⇒ no sampling grant (p=1). The edge applies it via WithSampleP on // the metric's sketch wrapper at the next EpochReset (never mid-window, so - // both merge operands share one p). Orthogonal to LocalSlack (which governs - // emission/bandwidth; this governs update CPU). See + // both merge operands share one p). See // docs/distributed-nitrosketch-coordinated-sampling.md. SampleP float64 } -// Poll is the coordinator→edge demand for the current local value (round close). +// Poll is the coordinator→edge demand for an immediate out-of-cadence report. type Poll struct { AggID uint64 Key []byte @@ -155,8 +164,7 @@ type Poll struct { WindowStartMs uint64 } -// Close is the coordinator→edge round-closed notice; the edge advances its -// round baseline. +// Close is the coordinator→edge round-closed notice. type Close struct { AggID uint64 Round uint64 diff --git a/asap-precompute-go/monitor_linear_e2e_test.go b/asap-precompute-go/monitor_linear_e2e_test.go index 1f79718f..0b467902 100644 --- a/asap-precompute-go/monitor_linear_e2e_test.go +++ b/asap-precompute-go/monitor_linear_e2e_test.go @@ -4,10 +4,11 @@ package precompute_test // End-to-end (in-process) proof that a FunctionalLinearBuckets monitor on a -// DDSketch series actually drives the slack countdown: Observe → window hook → -// monitorValue → DDSketchWrapper.LinearReadout (value-range count) → -// engine.Observe → report. Sum + CMS-point are covered elsewhere; this closes -// the loop for the linear functional specifically. +// DDSketch series actually drives the report pipeline: Observe → window hook +// → monitorValue → DDSketchWrapper.LinearReadout (value-range count) → +// engine.Observe → report, on the engine's own reportEveryN cadence (no +// slack/grant involved — alerting retired). Sum + CMS-point are covered +// elsewhere; this closes the loop for the linear functional specifically. import ( "testing" @@ -42,7 +43,6 @@ func TestMonitor_LinearBuckets_DDSketch_RangeCountDrivesReports(t *testing.T) { Functional: monitor.FunctionalLinearBuckets, Coeffs: []float64{50}, // count of samples with value >= 50 CoordinatorURL: "passthrough:///test", - Tau: 100, Epsilon: 0.05, }, } @@ -65,35 +65,44 @@ func TestMonitor_LinearBuckets_DDSketch_RangeCountDrivesReports(t *testing.T) { } } - // First observation registers the monitor (no grant yet → silent). - obs(1.0) // below 50 → range-count stays 0 - if len(rep.reports) != 0 { - t.Fatalf("reported before any grant") + // First observation registers the monitor AND fires an immediate report + // (obsCount==1), carrying the range-count at that point (0: 1.0 < 50). + obs(1.0) + if len(rep.reports) != 1 { + t.Fatalf("expected an immediate report on the first observation, got %d", len(rep.reports)) + } + if got := rep.reports[0].LocalValue; got != 0 { + t.Fatalf("first report should carry range-count 0, got %v", got) } - // Coordinator grants slack 3 for this round. + // Coordinator grants a sampling probability for this group — no longer + // gates reporting at all (that was the retired slack-crossing trigger). // Sum/LinearBuckets monitors key by the series GROUP key (the canonical // AggregateBy/label tuple), so the grant must target that same group. groupKey := []byte("svc=checkout") - eng.OnGrant(monitor.Grant{AggID: uint64(aggID), Key: groupKey, Round: 1, LocalSlack: 3, WindowStartMs: windowStart}) + eng.OnGrant(monitor.Grant{AggID: uint64(aggID), Key: groupKey, Round: 1, WindowStartMs: windowStart, SampleP: 0.5}) - // Two in-range samples: range-count = 2 < slack 3 → still silent. + // More in-range samples, short of the ReportEveryN cadence → no new report. obs(100.0) obs(200.0) - // An out-of-range sample does NOT advance the count. - obs(2.0) - if len(rep.reports) != 0 { - t.Fatalf("reported too early: range-count below slack; got %d reports", len(rep.reports)) + obs(2.0) // out-of-range, does not advance the range-count + obsCount := 4 + if len(rep.reports) != 1 { + t.Fatalf("reported too early, before the cadence: got %d reports", len(rep.reports)) } - // Third in-range sample: range-count = 3 >= slack 3 → exactly one report, - // carrying the range-count (3), not the raw observed value. - obs(100.0) - if len(rep.reports) != 1 { - t.Fatalf("expected exactly one report once range-count crossed slack, got %d", len(rep.reports)) + // Drive obsCount up to the ReportEveryN cadence to trigger report #2. + for obsCount < monitor.ReportEveryN+1 { + obs(1.0) // out-of-range; keeps range-count fixed while advancing obsCount + obsCount++ + } + if len(rep.reports) != 2 { + t.Fatalf("expected a second report once the cadence was reached, got %d", len(rep.reports)) } - if got := rep.reports[0].LocalValue; got != 3 { - t.Fatalf("report should carry the value-range count (3), got %v", got) + // range-count at this point: two in-range samples (100, 200) from above; + // every filler observation was out-of-range, so the count stays 2. + if got := rep.reports[1].LocalValue; got != 2 { + t.Fatalf("second report should carry the value-range count (2), got %v", got) } } @@ -125,23 +134,37 @@ func TestMonitor_Sum_PerGroup_NoCollision(t *testing.T) { Value: precompute.FloatValue(v), }) } - obs("a", 0) // registers group "zone=a" - obs("b", 0) // registers group "zone=b" - // Grant each group its own slack (keyed by the group bytes). - eng.OnGrant(monitor.Grant{AggID: uint64(aggID), Key: []byte("zone=a"), Round: 1, LocalSlack: 5, WindowStartMs: windowStart}) - eng.OnGrant(monitor.Grant{AggID: uint64(aggID), Key: []byte("zone=b"), Round: 1, LocalSlack: 5, WindowStartMs: windowStart}) + // Each group's first observation registers AND immediately reports (its + // own obsCount==1) — independent of the other group. + obs("a", 10) // registers + reports group "zone=a" + obs("b", 2) // registers + reports group "zone=b" + if len(rep.reports) != 2 { + t.Fatalf("expected one immediate report per group, got %d", len(rep.reports)) + } + byKey := map[string]monitor.Report{} + for _, r := range rep.reports { + byKey[string(r.Key)] = r + } + if byKey["zone=a"].LocalValue != 10 || byKey["zone=b"].LocalValue != 2 { + t.Fatalf("per-group first report should carry that group's own sum, got %+v", rep.reports) + } - obs("a", 10) // zone=a sum=10 >= 5 → reports group a - obs("b", 2) // zone=b sum=2 < 5 → silent (no collision with a) - if len(rep.reports) != 1 { - t.Fatalf("expected exactly 1 report (only zone=a crossed), got %d", len(rep.reports)) + // Grant each group its own sampling probability (keyed by the group bytes) + // — no longer gates reporting, only SampleP. + eng.OnGrant(monitor.Grant{AggID: uint64(aggID), Key: []byte("zone=a"), Round: 1, WindowStartMs: windowStart, SampleP: 0.5}) + eng.OnGrant(monitor.Grant{AggID: uint64(aggID), Key: []byte("zone=b"), Round: 1, WindowStartMs: windowStart, SampleP: 0.9}) + + // Drive ONLY zone=a to the ReportEveryN cadence (obsCount 1 → ReportEveryN+1); + // zone=b stays far short of it. Only zone=a should produce a second + // report — no collision. + for i := 0; i < monitor.ReportEveryN; i++ { + obs("a", 1) } - if string(rep.reports[0].Key) != "zone=a" { - t.Fatalf("report should be keyed by group zone=a, got %q", rep.reports[0].Key) + obs("b", 1) // zone=b's second observation — nowhere near its own cadence + if len(rep.reports) != 3 { + t.Fatalf("expected exactly one new report (zone=a's cadence), got %d total", len(rep.reports)) } - obs("b", 10) // zone=b sum=10 >= 5 → now reports group b independently - if len(rep.reports) != 2 || string(rep.reports[1].Key) != "zone=b" { - t.Fatalf("expected an independent zone=b report; got %d reports, last key %q", - len(rep.reports), rep.reports[len(rep.reports)-1].Key) + if string(rep.reports[2].Key) != "zone=a" { + t.Fatalf("the new report should be zone=a's, got key %q", rep.reports[2].Key) } } diff --git a/deploy/mvp-multinode/scripts/monitor_e2e.sh b/deploy/mvp-multinode/scripts/monitor_e2e.sh index a5d991d7..72a07c5a 100755 --- a/deploy/mvp-multinode/scripts/monitor_e2e.sh +++ b/deploy/mvp-multinode/scripts/monitor_e2e.sh @@ -1,7 +1,12 @@ #!/usr/bin/env bash -# Cross-language CDM e2e: a REAL Go edge (precompute + monitor.Engine + -# grpcclient) drives the REAL Rust monitor coordinator (tonic MonitorService) -# over a live bidi gRPC stream. Asserts the global-threshold alert fires. +# Cross-language coordinated-sampling e2e: a REAL Go edge (precompute + +# monitor.Engine + grpcclient) drives the REAL Rust monitor coordinator +# (tonic MonitorService) over a live bidi gRPC stream. Asserts a real +# coordinated-sampling grant (SlackGrant.sample_p) comes back. +# +# Global-threshold alerting is retired (see asap-precompute-go/monitor and +# ASAPQuery-backend data_plane::monitor package docs) — this script no longer +# waits for or asserts an alert. # # This validates the one seam unit tests can't: the Go-generated client and the # Rust-generated server interoperating over an actual stream (not a byte @@ -12,7 +17,7 @@ set -euo pipefail PORT="${1:-45319}" AGG_ID=1 -TAU=100 +TAU=100 # accepted by the harness/driver for CLI back-compat; unused WINDOW_MS=3600000 # Repo roots (this script lives in ASAPCollector/deploy/mvp-multinode/scripts). @@ -21,6 +26,12 @@ COLLECTOR_ROOT="$(cd "$SCRIPT_DIR/../../.." && pwd)" BACKEND_ROOT="$(cd "$COLLECTOR_ROOT/../ASAPQuery-backend" && pwd)" GRPCCLIENT_DIR="$COLLECTOR_ROOT/asap-precompute-go/monitor/grpcclient" +# The e2edriver observes a Sum series labeled svc=checkout; Sum monitors key +# by the series GROUP key (asap-precompute-go precompute.go groupKeyBytes), so +# the harness's monitor config must register under that same key or the +# driver's registration is rejected as unconfigured. +MON_KEY="svc=checkout" + HARNESS_LOG="$(mktemp)" HARNESS_PID="" cleanup() { @@ -36,8 +47,8 @@ HARNESS_BIN="$BACKEND_ROOT/target/debug/monitor_coordinator_harness" echo "==> Building Go edge driver" ( cd "$GRPCCLIENT_DIR" && GOFLAGS=-mod=mod go build -o /tmp/e2edriver ./cmd/e2edriver ) -echo "==> Starting coordinator on :$PORT (agg_id=$AGG_ID tau=$TAU window_ms=$WINDOW_MS)" -"$HARNESS_BIN" "$PORT" "$AGG_ID" "$TAU" "$WINDOW_MS" 30 >"$HARNESS_LOG" 2>&1 & +echo "==> Starting coordinator on :$PORT (agg_id=$AGG_ID window_ms=$WINDOW_MS key=$MON_KEY)" +"$HARNESS_BIN" "$PORT" "$AGG_ID" "$TAU" "$WINDOW_MS" 30 "$MON_KEY" >"$HARNESS_LOG" 2>&1 & HARNESS_PID=$! # Wait for the harness to report readiness. @@ -49,20 +60,20 @@ if ! grep -q "HARNESS_READY" "$HARNESS_LOG"; then echo "FAIL: coordinator did not become ready"; cat "$HARNESS_LOG"; exit 1 fi -echo "==> Running edge driver (sum climbs past tau)" +echo "==> Running edge driver (periodic rate reports)" /tmp/e2edriver "localhost:$PORT" "$AGG_ID" "$TAU" 5 200 || true -# The harness exits 0 on alert; give it a moment to flush + exit. +# The harness exits 0 once it observes a real grant; give it a moment to flush + exit. wait "$HARNESS_PID" 2>/dev/null || true HARNESS_PID="" echo "----- coordinator output -----" cat "$HARNESS_LOG" echo "------------------------------" -if grep -q "MONITOR_ALERT" "$HARNESS_LOG"; then - echo "PASS: global-threshold alert fired over live Go↔Rust gRPC stream" +if grep -q "MONITOR_GRANT" "$HARNESS_LOG"; then + echo "PASS: coordinated-sampling grant observed over live Go↔Rust gRPC stream" exit 0 else - echo "FAIL: no alert fired" + echo "FAIL: no grant observed" exit 1 fi