Skip to content

feat(#508b-2): scalar CDM slack-countdown monitor + gRPC transport #512

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
zzylol wants to merge 2 commits into from
Closed

feat(#508b-2): scalar CDM slack-countdown monitor + gRPC transport #512

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
ed8cf2e
feat(#508b-2): scalar CDM slack-countdown monitor + gRPC transport
zzylol Jul 9, 2026
3647ffb
remove SetSampleGrantHook (its only caller was the otlpfilter grant w…
zzylol Jul 16, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 3 additions & 2 deletions asap-precompute-go/monitor/engine.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -147,9 +147,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
Expand All @@ -159,6 +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
e.mu.Unlock()
}

// OnPoll answers a poll with the current local value and advances the baseline
Expand Down
5 changes: 3 additions & 2 deletions asap-precompute-go/monitor/grpcclient/monitorpb/monitor.pb.go
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

5 changes: 3 additions & 2 deletions asap-precompute-go/monitor/grpcclient/monitorpb/monitor.proto
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
Expand Down
8 changes: 5 additions & 3 deletions asap-precompute-go/monitor/types.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
Expand Down