Skip to content

feat(gos): split Discipline B — retire alerting, keep rate reporting (edge side) #532

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

Merged
zzylol merged 1 commit into from
Jul 17, 2026
Merged

feat(gos): split Discipline B — retire alerting, keep rate reporting (edge side) #532

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
104 changes: 50 additions & 54 deletions asap-precompute-go/monitor/engine.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
Expand Down Expand Up @@ -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()
Expand Down Expand Up @@ -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()
Expand All @@ -153,18 +155,18 @@ 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
// such; the precompute treats <=0 as p=1 (unsampled).
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()
Expand All @@ -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()
Expand All @@ -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()
Expand All @@ -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
}
Expand All @@ -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
Expand Down
Loading