From ed8cf2eea684eba353ad604bdfb0dad5bfebc34f Mon Sep 17 00:00:00 2001 From: zzylol Date: Wed, 8 Jul 2026 22:47:54 -0600 Subject: [PATCH 1/2] feat(#508b-2): scalar CDM slack-countdown monitor + gRPC transport MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Where the sample probability p (and the alert threshold) come from: the Cormode-Muthukrishnan-Yi distributed functional monitor. Per (agg_id,key) the coordinator runs rounds, grants each edge local slack + a sample_p via the ε-floor p=1/(1+ε²·rate), and fires on τ. Self-contained monitor package + regenerated grpc proto. Co-Authored-By: Claude Opus 4.8 (1M context) --- asap-precompute-go/monitor/engine.go | 29 +++++++++++- asap-precompute-go/monitor/grant_hook_test.go | 47 +++++++++++++++++++ .../grpcclient/monitorpb/monitor.pb.go | 5 +- .../grpcclient/monitorpb/monitor.proto | 5 +- asap-precompute-go/monitor/types.go | 8 ++-- 5 files changed, 85 insertions(+), 9 deletions(-) create mode 100644 asap-precompute-go/monitor/grant_hook_test.go diff --git a/asap-precompute-go/monitor/engine.go b/asap-precompute-go/monitor/engine.go index a28b9cb7..ba492242 100644 --- a/asap-precompute-go/monitor/engine.go +++ b/asap-precompute-go/monitor/engine.go @@ -31,7 +31,7 @@ type monitorState struct { // 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 (AllocateSampleRates). Reset to 0 at each epoch boundary. + // probability via the whole-sketch ε-floor. Reset to 0 at each epoch boundary. obsCount uint64 // grantedSampleP is the coordinator-allocated distributed-NitroSketch // update-sampling probability for this monitor's agg. 0 (unset) ⇒ no @@ -68,6 +68,14 @@ type Engine struct { reporter Reporter edgeID string epochWindowMs uint64 + + // onSampleGrant, when non-nil, is invoked (outside the engine mutex, on + // the transport read goroutine) for every ACCEPTED grant with the grant's + // aggID and sampling probability — the production hook that forwards the + // coordinator's sampling decision to the wire-level otlpfilter + // (OnGrant → SampleState.Upsert, design §3.1.1). Stale-epoch / unknown + // grants do not fire it, mirroring grantedSampleP storage. + onSampleGrant func(aggID uint64, sampleP float64) } // NewEngine builds an Engine for the given edge identity and tumbling window @@ -89,6 +97,16 @@ func (e *Engine) SetReporter(r Reporter) { e.mu.Unlock() } +// SetSampleGrantHook installs the callback fired for every accepted grant +// with (aggID, Grant.SampleP). The hook runs on the transport read goroutine +// OUTSIDE the engine mutex, so it may safely call back into shared state +// (e.g. otlpfilter.SampleState.Upsert). Safe to call concurrently. +func (e *Engine) SetSampleGrantHook(fn func(aggID uint64, sampleP float64)) { + e.mu.Lock() + e.onSampleGrant = fn + e.mu.Unlock() +} + // 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 @@ -147,9 +165,9 @@ func (e *Engine) Observe(aggID uint64, key []byte, value float64, windowStart ui // unreported-but-below-slack mass. func (e *Engine) OnGrant(g Grant) { e.mu.Lock() - defer e.mu.Unlock() st := e.states[mapKey{g.AggID, string(g.Key)}] if st == nil || st.windowStart != g.WindowStartMs { + e.mu.Unlock() return // unknown monitor or stale epoch } st.round = g.Round @@ -159,6 +177,13 @@ func (e *Engine) OnGrant(g Grant) { // current decision, so 0 means "no sampling this round" and is recorded as // such; the precompute treats <=0 as p=1 (unsampled). st.grantedSampleP = g.SampleP + hook := e.onSampleGrant + e.mu.Unlock() + // Forward the accepted grant's sampling decision to the wire filter + // (outside the mutex — the hook touches shared filter state). + if hook != nil { + hook(g.AggID, g.SampleP) + } } // OnPoll answers a poll with the current local value and advances the baseline diff --git a/asap-precompute-go/monitor/grant_hook_test.go b/asap-precompute-go/monitor/grant_hook_test.go new file mode 100644 index 00000000..3b2aee5b --- /dev/null +++ b/asap-precompute-go/monitor/grant_hook_test.go @@ -0,0 +1,47 @@ +package monitor + +import "testing" + +// TestSampleGrantHook_FiresOnAcceptedGrant: the hook receives (aggID, SampleP) +// exactly when a grant is accepted (known monitor, fresh epoch), mirroring +// grantedSampleP storage — the OnGrant → wire-filter forwarding contract. +func TestSampleGrantHook_FiresOnAcceptedGrant(t *testing.T) { + const ( + aggID = uint64(42) + window = uint64(60_000) + ) + eng := NewEngine("edge-1", window, nil) + + var gotAgg uint64 + var gotP float64 + fired := 0 + eng.SetSampleGrantHook(func(a uint64, p float64) { + gotAgg, gotP = a, p + fired++ + }) + + // Grant before any Observe: unknown monitor → dropped, hook silent. + eng.OnGrant(Grant{AggID: aggID, WindowStartMs: window, Round: 1, LocalSlack: 10, SampleP: 0.25}) + if fired != 0 { + t.Fatalf("hook fired for unknown monitor") + } + + // Register the monitor state via the hot path, then grant. + eng.Observe(aggID, nil, 1.0, window) + eng.OnGrant(Grant{AggID: aggID, WindowStartMs: window, Round: 1, LocalSlack: 10, SampleP: 0.25}) + if fired != 1 || gotAgg != aggID || gotP != 0.25 { + t.Fatalf("accepted grant: fired=%d agg=%d p=%v, want 1/%d/0.25", fired, gotAgg, gotP, aggID) + } + + // Stale epoch → dropped, hook silent. + eng.OnGrant(Grant{AggID: aggID, WindowStartMs: window - 60_000, Round: 2, SampleP: 0.5}) + if fired != 1 { + t.Fatalf("hook fired for stale-epoch grant") + } + + // Re-grant with a new p on the fresh epoch → fires again (p updates). + eng.OnGrant(Grant{AggID: aggID, WindowStartMs: window, Round: 2, LocalSlack: 10, SampleP: 0.5}) + if fired != 2 || gotP != 0.5 { + t.Fatalf("re-grant: fired=%d p=%v, want 2/0.5", fired, gotP) + } +} diff --git a/asap-precompute-go/monitor/grpcclient/monitorpb/monitor.pb.go b/asap-precompute-go/monitor/grpcclient/monitorpb/monitor.pb.go index baa16367..19e6740d 100644 --- a/asap-precompute-go/monitor/grpcclient/monitorpb/monitor.pb.go +++ b/asap-precompute-go/monitor/grpcclient/monitorpb/monitor.pb.go @@ -136,7 +136,7 @@ type MonitorReport struct { LocalValue float64 `protobuf:"fixed64,5,opt,name=local_value,json=localValue,proto3" json:"local_value,omitempty"` // current additive local value at report time Round uint64 `protobuf:"varint,6,opt,name=round,proto3" json:"round,omitempty"` // round this report answers Seq uint64 `protobuf:"varint,7,opt,name=seq,proto3" json:"seq,omitempty"` // per-edge monotonic counter; idempotent-retransmit dedup - Rate float64 `protobuf:"fixed64,8,opt,name=rate,proto3" json:"rate,omitempty"` // edge's observed items/window for this agg+key — feeds the coordinator's sample-rate allocation (p_i ~ sqrt(f_i/rate_i)) + Rate float64 `protobuf:"fixed64,8,opt,name=rate,proto3" json:"rate,omitempty"` // edge's observed items/window for this agg+key — feeds the coordinator's whole-sketch sampling floor (p_i = 1/(1+eps^2*rate_i)) unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache } @@ -237,7 +237,8 @@ type SlackGrant struct { LocalSlack float64 `protobuf:"fixed64,4,opt,name=local_slack,json=localSlack,proto3" json:"local_slack,omitempty"` WindowStartMs uint64 `protobuf:"varint,5,opt,name=window_start_ms,json=windowStartMs,proto3" json:"window_start_ms,omitempty"` // sample_p is the distributed-NitroSketch update-sampling probability the - // coordinator allocates this edge (AllocateSampleRates: p_i ~ sqrt(f_i/rate_i)). + // coordinator allocates this edge via the whole-sketch epsilon-floor + // (p_i = 1/(1+eps^2*rate_i); see data_plane allocate_p / epsilon_sample_floor). // 0 (unset) => no sampling grant (p=1). The edge applies it via WithSampleP on // sampling-capable sketch wrappers (CMS/CountSketch/DDSketch) at the next // EpochReset; other families ignore it. Orthogonal to local_slack (CPU vs diff --git a/asap-precompute-go/monitor/grpcclient/monitorpb/monitor.proto b/asap-precompute-go/monitor/grpcclient/monitorpb/monitor.proto index 82948b6a..068c1954 100644 --- a/asap-precompute-go/monitor/grpcclient/monitorpb/monitor.proto +++ b/asap-precompute-go/monitor/grpcclient/monitorpb/monitor.proto @@ -53,7 +53,7 @@ message MonitorReport { double local_value = 5; // current additive local value at report time uint64 round = 6; // round this report answers uint64 seq = 7; // per-edge monotonic counter; idempotent-retransmit dedup - double rate = 8; // edge's observed items/window for this agg+key — feeds the coordinator's sample-rate allocation (p_i ~ sqrt(f_i/rate_i)) + double rate = 8; // edge's observed items/window for this agg+key — feeds the coordinator's whole-sketch sampling floor (p_i = 1/(1+eps^2*rate_i)) } // coordinator → edge: this round's per-edge slack budget. The edge reports once @@ -65,7 +65,8 @@ message SlackGrant { double local_slack = 4; uint64 window_start_ms = 5; // sample_p is the distributed-NitroSketch update-sampling probability the - // coordinator allocates this edge (AllocateSampleRates: p_i ~ sqrt(f_i/rate_i)). + // coordinator allocates this edge via the whole-sketch epsilon-floor + // (p_i = 1/(1+eps^2*rate_i); see data_plane allocate_p / epsilon_sample_floor). // 0 (unset) => no sampling grant (p=1). The edge applies it via WithSampleP on // sampling-capable sketch wrappers (CMS/CountSketch/DDSketch) at the next // EpochReset; other families ignore it. Orthogonal to local_slack (CPU vs diff --git a/asap-precompute-go/monitor/types.go b/asap-precompute-go/monitor/types.go index dae843bb..9de5ea4d 100644 --- a/asap-precompute-go/monitor/types.go +++ b/asap-precompute-go/monitor/types.go @@ -121,8 +121,9 @@ type Report struct { Round uint64 Seq uint64 // Rate is the edge's observed item count for this monitor over the current - // epoch (items/window). The coordinator feeds it into AllocateSampleRates - // (p_i ∝ √(f_i/rate_i)) to size this edge's distributed-NitroSketch sampling + // epoch (items/window). The coordinator feeds it into the whole-sketch + // ε-floor (p_i = 1/(1+ε²·rate_i); see data_plane allocate_p / + // epsilon_sample_floor) to size this edge's distributed-NitroSketch sampling // probability. 0 (unset) ⇒ the coordinator falls back to an unsampled // allocation for this edge. Rate float64 @@ -136,7 +137,8 @@ type Grant struct { LocalSlack float64 WindowStartMs uint64 // SampleP is the distributed-NitroSketch update-sampling probability the - // coordinator allocates this edge (AllocateSampleRates: p_i ∝ √(f_i/rate_i)). + // coordinator allocates this edge via the whole-sketch ε-floor + // (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 From 3647ffba8d00fce1d9973afc37034e21d4cc14b9 Mon Sep 17 00:00:00 2001 From: zzylol Date: Wed, 15 Jul 2026 21:48:07 -0600 Subject: [PATCH 2/2] remove SetSampleGrantHook (its only caller was the otlpfilter grant wiring) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Superseded by SDK-side single-decision sampling — the coordinator's grant no longer needs forwarding to a collector-side wire filter. grantedSampleP storage (feeding the wrapper's own native sampler via applyGrantedSampleP) is untouched and remains the fallback for traffic without SDK-side sampling. Co-Authored-By: Claude Opus 4.8 (1M context) --- asap-precompute-go/monitor/engine.go | 24 ---------- asap-precompute-go/monitor/grant_hook_test.go | 47 ------------------- 2 files changed, 71 deletions(-) delete mode 100644 asap-precompute-go/monitor/grant_hook_test.go diff --git a/asap-precompute-go/monitor/engine.go b/asap-precompute-go/monitor/engine.go index ba492242..7bb14f4a 100644 --- a/asap-precompute-go/monitor/engine.go +++ b/asap-precompute-go/monitor/engine.go @@ -68,14 +68,6 @@ type Engine struct { reporter Reporter edgeID string epochWindowMs uint64 - - // onSampleGrant, when non-nil, is invoked (outside the engine mutex, on - // the transport read goroutine) for every ACCEPTED grant with the grant's - // aggID and sampling probability — the production hook that forwards the - // coordinator's sampling decision to the wire-level otlpfilter - // (OnGrant → SampleState.Upsert, design §3.1.1). Stale-epoch / unknown - // grants do not fire it, mirroring grantedSampleP storage. - onSampleGrant func(aggID uint64, sampleP float64) } // NewEngine builds an Engine for the given edge identity and tumbling window @@ -97,16 +89,6 @@ func (e *Engine) SetReporter(r Reporter) { e.mu.Unlock() } -// SetSampleGrantHook installs the callback fired for every accepted grant -// with (aggID, Grant.SampleP). The hook runs on the transport read goroutine -// OUTSIDE the engine mutex, so it may safely call back into shared state -// (e.g. otlpfilter.SampleState.Upsert). Safe to call concurrently. -func (e *Engine) SetSampleGrantHook(fn func(aggID uint64, sampleP float64)) { - e.mu.Lock() - e.onSampleGrant = fn - e.mu.Unlock() -} - // 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 @@ -177,13 +159,7 @@ func (e *Engine) OnGrant(g Grant) { // current decision, so 0 means "no sampling this round" and is recorded as // such; the precompute treats <=0 as p=1 (unsampled). st.grantedSampleP = g.SampleP - hook := e.onSampleGrant e.mu.Unlock() - // Forward the accepted grant's sampling decision to the wire filter - // (outside the mutex — the hook touches shared filter state). - if hook != nil { - hook(g.AggID, g.SampleP) - } } // OnPoll answers a poll with the current local value and advances the baseline diff --git a/asap-precompute-go/monitor/grant_hook_test.go b/asap-precompute-go/monitor/grant_hook_test.go deleted file mode 100644 index 3b2aee5b..00000000 --- a/asap-precompute-go/monitor/grant_hook_test.go +++ /dev/null @@ -1,47 +0,0 @@ -package monitor - -import "testing" - -// TestSampleGrantHook_FiresOnAcceptedGrant: the hook receives (aggID, SampleP) -// exactly when a grant is accepted (known monitor, fresh epoch), mirroring -// grantedSampleP storage — the OnGrant → wire-filter forwarding contract. -func TestSampleGrantHook_FiresOnAcceptedGrant(t *testing.T) { - const ( - aggID = uint64(42) - window = uint64(60_000) - ) - eng := NewEngine("edge-1", window, nil) - - var gotAgg uint64 - var gotP float64 - fired := 0 - eng.SetSampleGrantHook(func(a uint64, p float64) { - gotAgg, gotP = a, p - fired++ - }) - - // Grant before any Observe: unknown monitor → dropped, hook silent. - eng.OnGrant(Grant{AggID: aggID, WindowStartMs: window, Round: 1, LocalSlack: 10, SampleP: 0.25}) - if fired != 0 { - t.Fatalf("hook fired for unknown monitor") - } - - // Register the monitor state via the hot path, then grant. - eng.Observe(aggID, nil, 1.0, window) - eng.OnGrant(Grant{AggID: aggID, WindowStartMs: window, Round: 1, LocalSlack: 10, SampleP: 0.25}) - if fired != 1 || gotAgg != aggID || gotP != 0.25 { - t.Fatalf("accepted grant: fired=%d agg=%d p=%v, want 1/%d/0.25", fired, gotAgg, gotP, aggID) - } - - // Stale epoch → dropped, hook silent. - eng.OnGrant(Grant{AggID: aggID, WindowStartMs: window - 60_000, Round: 2, SampleP: 0.5}) - if fired != 1 { - t.Fatalf("hook fired for stale-epoch grant") - } - - // Re-grant with a new p on the fresh epoch → fires again (p updates). - eng.OnGrant(Grant{AggID: aggID, WindowStartMs: window, Round: 2, LocalSlack: 10, SampleP: 0.5}) - if fired != 2 || gotP != 0.5 { - t.Fatalf("re-grant: fired=%d p=%v, want 2/0.5", fired, gotP) - } -}