From 8a2d2e1a2d06385a3f7ee9112f953f2c4b711ec9 Mon Sep 17 00:00:00 2001 From: Ian Oberst Date: Wed, 21 Oct 2026 12:32:40 -0700 Subject: [PATCH 01/12] Stop the level collector without waiting for the next tick The level collector checked for stop only after each ticker tick, so stopping a monitor waited up to one tick (1s), 0.5s on average. With removals processed one reload at a time, that wait added up to about 0.5s per removed monitor. Run now waits for either the next tick or stop. It still checks for stop after a tick, because select picks randomly when both are ready. Dropped tick counting is unchanged. A collection in progress still finishes before the collector stops; it's bounded by the engine max runtime. Tests drive Run by sending values on a buffered stopChan, one per tick, then closing it. Once Run receives a value, it keeps that one-per-tick behavior, so those tests see the close on the next tick as before. With the 1s tick, a stop that took 0.3s to 0.9s depending on when in the tick it arrived now takes well under a millisecond in the new test. Co-Authored-By: Claude Opus 5.5 --- monitor/level_collector.go | 63 ++++++++++++++++++++++++++++--------- monitor/loader_stop_test.go | 25 +++++++++++++++ 2 files changed, 73 insertions(+), 15 deletions(-) diff --git a/monitor/level_collector.go b/monitor/level_collector.go index 76d3aef..1ffccf5 100644 --- a/monitor/level_collector.go +++ b/monitor/level_collector.go @@ -231,8 +231,44 @@ func (c *lco) Run(stopChan, doneChan chan struct{}) error { interval := uint(0) ticker := time.NewTicker(td) defer ticker.Stop() + + // stop stops changePlan goroutine (if any) and prevents new ones in the + // pathological case that the LCH calls ChangePlan while the LCO is + // terminating, then stops the engine + stop := func() { + blip.Debug("stopChan closed s=%d interval=%d", s, interval) + c.stopChangePlan() + c.engine.Stop() // stop all collectors and run their cleanup func + } + var lastTick time.Time - for startTime := range ticker.C { + countTicks := false // see below + for { + // Wait for the next tick or stop, whichever is first, so stopping + // doesn't wait up to one tick. Stop closes stopChan. + // + // Tests also send values on a buffered stopChan to run a number of + // ticks: each tick receives one value, and the close after the last + // value stops Run on the next tick. Once a value is received, Run + // keeps that one-per-tick behavior: the value counts for the next + // tick, and later values and the close are received after each tick. + var startTime time.Time + gotValue := false + if countTicks { + startTime = <-ticker.C + } else { + select { + case _, ok := <-stopChan: + if !ok { + stop() + return nil + } + countTicks = true + gotValue = true + startTime = <-ticker.C + case startTime = <-ticker.C: + } + } s = s + te // Count ticks dropped while the previous iteration was blocked, usually @@ -243,20 +279,18 @@ func (c *lco) Run(stopChan, doneChan chan struct{}) error { } lastTick = startTime - // Was monitor stopped? - select { - case _, ok := <-stopChan: // yes, return immediately - // Stop changePlan goroutine (if any) and prevent new ones in the - // pathological case that the LCH calls ChangePlan while the LCO - // is terminating - if ok { - break + // Was monitor stopped? Check again because select picks randomly + // when both the tick and stop are ready. + if !gotValue { + select { + case _, ok := <-stopChan: // yes, return immediately + if ok { + break + } + stop() + return nil + default: // no } - blip.Debug("stopChan closed at %s s=%d interval=%d", startTime, s, interval) - c.stopChangePlan() - c.engine.Stop() // stop all collectors and run their cleanup func - return nil - default: // no } // Paused because no plan is set or it's being changed? @@ -286,7 +320,6 @@ func (c *lco) Run(stopChan, doneChan chan struct{}) error { c.stateMux.Unlock() // -- UNLOCK -- } - return nil } // stopChangePlan prevents new plan changes, cancels the current plan change, diff --git a/monitor/loader_stop_test.go b/monitor/loader_stop_test.go index 02074ca..1c13325 100644 --- a/monitor/loader_stop_test.go +++ b/monitor/loader_stop_test.go @@ -1062,3 +1062,28 @@ func TestLoaderStartClearsHeld(t *testing.T) { setIds(st, &ids) require.NoError(t, loader.Load(context.Background())) } + +func TestLoaderStopDoesNotWaitForTick(t *testing.T) { + // With the real 1s LCO tick, a stop must not wait for the next tick. + // Before the LCO selected on stop, stops took 0 to 1s (0.5s on average), + // depending on when in the tick they arrived. + ids := []string{"tick1"} + st, loader := setupStopTest(t, "5s", &ids, nil) + defer status.RemoveMonitor("tick1") + monitor.TickerDuration(time.Second, time.Second) // setupStopTest restores the test tick + + require.NoError(t, loader.Load(context.Background())) + loader.StartMonitors() + waitPrepared(t, st, 1, "tick1") + + for i := 0; i < 4; i++ { + time.Sleep(time.Duration(100+200*i) * time.Millisecond) // vary the offset into the tick + t0 := time.Now() + require.NoError(t, loader.Stop("tick1", true)) + d := time.Since(t0) + t.Logf("stop %d: %s", i, d) + assert.Less(t, d, 250*time.Millisecond, "stop %d waited for the LCO tick", i) + require.NoError(t, loader.Start("tick1", true)) + waitPrepared(t, st, i+2, "tick1") + } +} From ef258a69fbc6834a604b18aadc78b55ac14ab532 Mon Sep 17 00:00:00 2001 From: Ian Oberst Date: Mon, 26 Oct 2026 10:10:59 -0700 Subject: [PATCH 02/12] Cancel the level collector's collection on stop Review fixes for stopping the level collector: - Run waits for the next tick or stop in one select and checks for stop again after the tick, instead of the test-only protocol that sent tick counts and values on stopChan. Tests stop the LCO after N ticks with a tick hook (StopAfterTicks in export_test.go). - A stop cancels a collection in progress instead of waiting for it, which could take up to the engine max runtime. The LCO drops the canceled collection's metrics, and the engine doesn't count a canceled collection as an EMR timeout. - New LCO tests measure a stop during the tick wait and during a blocked collection directly, with bounds relative to the tick and the max runtime. TestLoaderStopDoesNotWaitForTick uses a 4s tick and a bound of a quarter tick instead of a fixed 250ms. Co-Authored-By: Claude Opus 5.5 --- monitor/engine.go | 6 +- monitor/export_test.go | 21 ++++++ monitor/fault_counters_test.go | 11 ++- monitor/level_collector.go | 79 +++++++++++---------- monitor/level_collector_stop_test.go | 102 +++++++++++++++++++++++++++ monitor/level_collector_test.go | 30 +++----- monitor/loader_stop_test.go | 20 +++--- 7 files changed, 196 insertions(+), 73 deletions(-) create mode 100644 monitor/export_test.go create mode 100644 monitor/level_collector_stop_test.go diff --git a/monitor/engine.go b/monitor/engine.go index cde473a..e1fe071 100644 --- a/monitor/engine.go +++ b/monitor/engine.go @@ -319,7 +319,8 @@ func (e *Engine) Prepare(ctx context.Context, plan blip.Plan, before, after func // - []metrics is sorted ascending by Interval // // Collect returns when all collectors it starts return, or when emrCtx -// (engine max runtime) expires. The former is the normal case. +// (engine max runtime) expires or is canceled. The former is the normal case. +// A canceled emrCtx isn't counted as an EMR timeout. // // Both metrics and an error can be returned in the case of partially success: // some collectors work but others fail. Caller should check returned metrics @@ -444,6 +445,9 @@ SWEEP: // are missing from this interval. Not an error event because, when the // host is CPU-starved, this can happen every interval for every monitor. notReturned := len(running) + if emrTimeout && emrCtx.Err() == context.Canceled { + emrTimeout = false // canceled (stopping), not timed out + } if emrTimeout { e.counters.emrTimeouts.Add(1) e.counters.domainsNotStarted.Add(uint64(notStarted)) diff --git a/monitor/export_test.go b/monitor/export_test.go new file mode 100644 index 0000000..d80b585 --- /dev/null +++ b/monitor/export_test.go @@ -0,0 +1,21 @@ +// Copyright 2026 Block, Inc. + +package monitor + +// StopAfterTicks makes every LCO Run return at tick n+1, before it collects at +// that tick, by closing stopChan. Use it for a test that runs one LCO, and call +// the returned func to remove it. +func StopAfterTicks(n uint, stopChan chan struct{}) func() { + tickerMux.Lock() + tickHook = func(tick uint) { + if tick == n+1 { + close(stopChan) + } + } + tickerMux.Unlock() + return func() { + tickerMux.Lock() + tickHook = nil + tickerMux.Unlock() + } +} diff --git a/monitor/fault_counters_test.go b/monitor/fault_counters_test.go index 00cb3b4..e471575 100644 --- a/monitor/fault_counters_test.go +++ b/monitor/fault_counters_test.go @@ -219,12 +219,9 @@ func TestLevelCollectorCountsDroppedTicks(t *testing.T) { c.emr = 2 * time.Second // longer than the slow collection, so the EMR doesn't expire c.paused = false - // Run returns on the 7th tick: each tick reads one value, then stopChan is closed - stopChan := make(chan struct{}, 6) - for i := 0; i < 6; i++ { - stopChan <- struct{}{} - } - close(stopChan) + // Run returns on the 7th tick, after 6 ticks + stopChan := make(chan struct{}) + defer StopAfterTicks(6, stopChan)() doneChan := make(chan struct{}) go c.Run(stopChan, doneChan) select { @@ -260,7 +257,7 @@ func TestLevelCollectorCountsDroppedMetrics(t *testing.T) { for i := 0; i < cap(c.metricsChan); i++ { c.metricsChan <- nil } - c.collect(1, level, time.Now()) + c.collect(context.Background(), 1, level, time.Now()) if v := counters(t, monitorId)[status.LEVEL_METRICS_DROPPED]; v != 1 { t.Errorf("%s = %d, expected 1", status.LEVEL_METRICS_DROPPED, v) diff --git a/monitor/level_collector.go b/monitor/level_collector.go index 1ffccf5..20ded6d 100644 --- a/monitor/level_collector.go +++ b/monitor/level_collector.go @@ -140,6 +140,11 @@ var ( tickerMux = &sync.Mutex{} // make go test -race happy tickerDuration = 1 * time.Second // used for testing timeElapsed = 1 * time.Second // used for testing + + // tickHook, if set, is called by Run after each tick with the tick number + // (1 for the first tick). Tests use it to stop Run after a number of ticks + // by closing stopChan, which Run sees before it collects at that tick. + tickHook func(tick uint) // used for testing ) // recvMetrics receives metrics on metricsChan and send them to all the sinks. @@ -225,7 +230,7 @@ func (c *lco) Run(stopChan, doneChan chan struct{}) error { status.Monitor(c.monitorId, status.LEVEL_COLLECTOR, "started at %s (paused until plan change)", blip.FormatTime(time.Now())) tickerMux.Lock() // make go test -race happy - td, te := tickerDuration, timeElapsed + td, te, onTick := tickerDuration, timeElapsed, tickHook tickerMux.Unlock() s := -1 * te // -1 so first tick=0 and all levels collected interval := uint(0) @@ -241,35 +246,35 @@ func (c *lco) Run(stopChan, doneChan chan struct{}) error { c.engine.Stop() // stop all collectors and run their cleanup func } + // collectCtx is canceled when stopChan is closed, so a stop doesn't wait for + // a collection in progress + collectCtx, cancelCollect := context.WithCancel(context.Background()) + defer cancelCollect() + go func() { + select { + case <-stopChan: + cancelCollect() + case <-collectCtx.Done(): + } + }() + var lastTick time.Time - countTicks := false // see below + tick := uint(0) for { // Wait for the next tick or stop, whichever is first, so stopping // doesn't wait up to one tick. Stop closes stopChan. - // - // Tests also send values on a buffered stopChan to run a number of - // ticks: each tick receives one value, and the close after the last - // value stops Run on the next tick. Once a value is received, Run - // keeps that one-per-tick behavior: the value counts for the next - // tick, and later values and the close are received after each tick. var startTime time.Time - gotValue := false - if countTicks { - startTime = <-ticker.C - } else { - select { - case _, ok := <-stopChan: - if !ok { - stop() - return nil - } - countTicks = true - gotValue = true - startTime = <-ticker.C - case startTime = <-ticker.C: - } + select { + case <-stopChan: + stop() + return nil + case startTime = <-ticker.C: } s = s + te + tick++ + if onTick != nil { + onTick(tick) + } // Count ticks dropped while the previous iteration was blocked, usually // by a slow collection. Run doesn't catch up on dropped ticks, so each one @@ -281,16 +286,11 @@ func (c *lco) Run(stopChan, doneChan chan struct{}) error { // Was monitor stopped? Check again because select picks randomly // when both the tick and stop are ready. - if !gotValue { - select { - case _, ok := <-stopChan: // yes, return immediately - if ok { - break - } - stop() - return nil - default: // no - } + select { + case <-stopChan: // yes, return immediately + stop() + return nil + default: // no } // Paused because no plan is set or it's being changed? @@ -316,7 +316,7 @@ func (c *lco) Run(stopChan, doneChan chan struct{}) error { // Collect metrics at this level interval += 1 - c.collect(interval, c.levels[level].Name, startTime) + c.collect(collectCtx, interval, c.levels[level].Name, startTime) c.stateMux.Unlock() // -- UNLOCK -- } @@ -339,7 +339,10 @@ func (c *lco) stopChangePlan() { c.changePlanDoneChan = nil } -func (c *lco) collect(interval uint, levelName string, startTime time.Time) { +// collect collects metrics at one level and sends them to the sinks. If ctx +// is canceled (the LCO is stopping), the collection is canceled, and its +// metrics are dropped. +func (c *lco) collect(ctx context.Context, interval uint, levelName string, startTime time.Time) { status.Monitor(c.monitorId, status.LEVEL_COLLECT, "%s/%s: collecting", c.plan.Name, levelName) defer func() { if err := recover(); err != nil { // catch panic in engine and TransformMetrics @@ -355,10 +358,14 @@ func (c *lco) collect(interval uint, levelName string, startTime time.Time) { // // Collect all metrics at this level. This is where metrics // collection begins. Then Engine.Collect does the real work. - emrCtx, emrCancel := context.WithDeadline(context.Background(), startTime.Add(c.emr)) + emrCtx, emrCancel := context.WithDeadline(ctx, startTime.Add(c.emr)) defer emrCancel() metrics, err := c.engine.Collect(emrCtx, interval, levelName, startTime) blip.Debug("%s: level %s: done in %s", c.monitorId, levelName, metrics[0].End.Sub(metrics[0].Begin)) + if ctx.Err() != nil { + blip.Debug("%s: level %s: stopped, dropping metrics", c.monitorId, levelName) + return + } if err != nil { status.Monitor(c.monitorId, "error:collect", "%s", err.Error()) diff --git a/monitor/level_collector_stop_test.go b/monitor/level_collector_stop_test.go new file mode 100644 index 0000000..8348c03 --- /dev/null +++ b/monitor/level_collector_stop_test.go @@ -0,0 +1,102 @@ +// Copyright 2026 Block, Inc. + +package monitor + +import ( + "context" + "testing" + "time" + + "github.com/cashapp/blip/v2" + "github.com/cashapp/blip/v2/plan" + "github.com/cashapp/blip/v2/status" + "github.com/cashapp/blip/v2/test/mock" +) + +// runStopLCO runs an LCO with one collector, calls ready to wait until it's +// running, then stops it. It returns how long Run took to return after stop. +func runStopLCO(t *testing.T, c *lco, ready func()) time.Duration { + t.Helper() + stopChan := make(chan struct{}) + doneChan := make(chan struct{}) + go c.Run(stopChan, doneChan) + ready() + t0 := time.Now() + close(stopChan) + select { + case <-doneChan: + case <-time.After(2 * time.Minute): + t.Fatal("LCO did not stop") + } + return time.Since(t0) +} + +func newStopTestLCO(monitorId, level string, collector blip.Collector) *lco { + c := newLevelCollectorWithDBProvider(LevelCollectorArgs{Config: blip.ConfigMonitor{MonitorId: monitorId}}, nil) + c.engine = faultTestEngine(monitorId, level, collector) + c.plan = c.engine.plan + c.levels = []plan.SortedLevel{{Freq: time.Second, Name: level}} + c.emr = time.Second + c.paused = false + return c +} + +func TestLevelCollectorStopDoesNotWaitForTick(t *testing.T) { + // Run waits for the next tick or stop, so a stop doesn't wait for the + // tick. Use a long tick: before, a stop right after Run started waited + // for the whole tick. + const monitorId = "lco-stop-tick" + defer status.RemoveMonitor(monitorId) + const tick = 10 * time.Second + TickerDuration(tick, time.Second) + defer TickerDuration(time.Second, time.Second) + + c := newStopTestLCO(monitorId, "every", mock.MetricsCollector{DomainFunc: func() string { return "test.lco-stop-tick" }}) + d := runStopLCO(t, c, func() {}) + if d > tick/10 { + t.Errorf("stop took %s, expected well under the %s tick", d, tick) + } +} + +func TestLevelCollectorStopCancelsCollection(t *testing.T) { + // A stop cancels a collection in progress instead of waiting for it to + // finish, which can take up to the engine max runtime (EMR). The canceled + // collection isn't counted as an EMR timeout. + const monitorId = "lco-stop-collect" + defer status.RemoveMonitor(monitorId) + TickerDuration(10*time.Millisecond, time.Second) + defer TickerDuration(time.Second, time.Second) + + started := make(chan struct{}, 1) + slow := mock.MetricsCollector{ + DomainFunc: func() string { return "test.lco-stop-collect" }, + CollectFunc: func(ctx context.Context, _ string) ([]blip.MetricValue, error) { + select { + case started <- struct{}{}: + default: + } + <-ctx.Done() // until canceled or the collector max runtime + return nil, ctx.Err() + }, + } + const maxRuntime = time.Minute + c := newStopTestLCO(monitorId, "every", slow) + c.emr = maxRuntime + for _, cl := range c.engine.collectors { + cl.cmr = maxRuntime + } + + d := runStopLCO(t, c, func() { + select { + case <-started: + case <-time.After(5 * time.Second): + t.Fatal("timeout waiting for collection to start") + } + }) + if d > maxRuntime/10 { + t.Errorf("stop took %s, expected well under the %s max runtime", d, maxRuntime) + } + if v := counters(t, monitorId)[status.ENGINE_EMR_TIMEOUTS]; v != 0 { + t.Errorf("%s = %d, expected 0", status.ENGINE_EMR_TIMEOUTS, v) + } +} diff --git a/monitor/level_collector_test.go b/monitor/level_collector_test.go index 8bdcbf7..11ae565 100644 --- a/monitor/level_collector_test.go +++ b/monitor/level_collector_test.go @@ -567,19 +567,16 @@ func TestLevelCollector_RGB_DomainPrioirty(t *testing.T) { monitor.TickerDuration(100*time.Millisecond, 100*time.Millisecond) defer monitor.TickerDuration(time.Second, time.Second) - // Create LCO and prime its stopChan for 5 iterations + // Create LCO and stop it after 5 ticks lco := monitor.NewLevelCollector(monitor.LevelCollectorArgs{ Config: moncfg, DB: db, PlanLoader: pl, Sinks: []blip.Sink{}, }) - stopChan := make(chan struct{}, 5) + stopChan := make(chan struct{}) doneChan := make(chan struct{}) - for i := 0; i < 5; i++ { - stopChan <- struct{}{} - } - close(stopChan) + defer monitor.StopAfterTicks(5, stopChan)() // readyChan is closed by a fake event receiver when lco.ChangePlan sends // event.CHANGE_PLAN_SUCCESS, which means the plan is loaded and ready. @@ -692,12 +689,9 @@ func TestLevelCollector_RGB_SlowBlue(t *testing.T) { Sinks: []blip.Sink{}, TransformMetrics: xf, }) - stopChan := make(chan struct{}, 4) + stopChan := make(chan struct{}) doneChan := make(chan struct{}) - for i := 0; i < 4; i++ { - stopChan <- struct{}{} - } - close(stopChan) + defer monitor.StopAfterTicks(4, stopChan)() readyChan := planSet(nil) defer event.RemoveSubscribers() @@ -854,12 +848,9 @@ func TestLevelCollector_RGB_ProgressiveBlue(t *testing.T) { Sinks: []blip.Sink{}, TransformMetrics: xf, }) - stopChan := make(chan struct{}, 5) + stopChan := make(chan struct{}) doneChan := make(chan struct{}) - for i := 0; i < 5; i++ { - stopChan <- struct{}{} - } - close(stopChan) + defer monitor.StopAfterTicks(5, stopChan)() readyChan := planSet(nil) defer event.RemoveSubscribers() @@ -1002,12 +993,9 @@ func TestLevelCollector_RGB_Fault(t *testing.T) { Sinks: []blip.Sink{}, TransformMetrics: xf, }) - stopChan := make(chan struct{}, 3) + stopChan := make(chan struct{}) doneChan := make(chan struct{}) - for i := 0; i < 3; i++ { - stopChan <- struct{}{} - } - close(stopChan) + defer monitor.StopAfterTicks(3, stopChan)() // Record events to check that 2 event.COLLECTOR_FAULT events are sent events := []string{} diff --git a/monitor/loader_stop_test.go b/monitor/loader_stop_test.go index 1c13325..266e1f0 100644 --- a/monitor/loader_stop_test.go +++ b/monitor/loader_stop_test.go @@ -1064,25 +1064,29 @@ func TestLoaderStartClearsHeld(t *testing.T) { } func TestLoaderStopDoesNotWaitForTick(t *testing.T) { - // With the real 1s LCO tick, a stop must not wait for the next tick. - // Before the LCO selected on stop, stops took 0 to 1s (0.5s on average), - // depending on when in the tick they arrived. + // A stop must not wait for the next LCO tick. Before the LCO selected on + // stop, a stop took up to one tick, depending on when in the tick it + // arrived. Use a long tick so the bound is relative to it and not to how + // fast a stop is under -race. TestLevelCollectorStop*, + // TestPlanChangerStopsDuringOfflineWait, and TestWriterStopsDuringWaits + // cover the other waits directly. + const tick = 4 * time.Second ids := []string{"tick1"} - st, loader := setupStopTest(t, "5s", &ids, nil) + st, loader := setupStopTest(t, "10s", &ids, nil) // stop timeout > tick defer status.RemoveMonitor("tick1") - monitor.TickerDuration(time.Second, time.Second) // setupStopTest restores the test tick + monitor.TickerDuration(tick, time.Second) // setupStopTest restores the test tick require.NoError(t, loader.Load(context.Background())) loader.StartMonitors() waitPrepared(t, st, 1, "tick1") - for i := 0; i < 4; i++ { - time.Sleep(time.Duration(100+200*i) * time.Millisecond) // vary the offset into the tick + for i, offset := range []time.Duration{tick / 20, tick / 2} { // vary the offset into the tick + time.Sleep(offset) t0 := time.Now() require.NoError(t, loader.Stop("tick1", true)) d := time.Since(t0) t.Logf("stop %d: %s", i, d) - assert.Less(t, d, 250*time.Millisecond, "stop %d waited for the LCO tick", i) + assert.Less(t, d, tick/4, "stop %d waited for the LCO tick", i) require.NoError(t, loader.Start("tick1", true)) waitPrepared(t, st, i+2, "tick1") } From a11f6be644d753b11a45fe4002b35948ec703674 Mon Sep 17 00:00:00 2001 From: Ian Oberst Date: Mon, 26 Oct 2026 10:11:01 -0700 Subject: [PATCH 03/12] Wake the plan changer and heartbeat writer on stop The plan changer slept 1s, or its offline retry wait, and the heartbeat writer slept heartbeat.freq, or 10 or 20s while it couldn't write, without watching stopChan. Both now sleep with sleep(stopChan, d), so a stop doesn't wait for them. The heartbeat package has its own copy of sleep because it can't import monitor. The stop-timeout docs now say that a healthy monitor stops in milliseconds, and list what can still make a stop slow. Co-Authored-By: Claude Opus 5.5 --- docs/content/config/config-file.md | 8 ++- heartbeat/writer.go | 43 +++++++----- heartbeat/writer_stop_test.go | 106 +++++++++++++++++++++++++++++ monitor/plan_changer.go | 10 +-- monitor/plan_changer_stop_test.go | 68 ++++++++++++++++++ 5 files changed, 212 insertions(+), 23 deletions(-) create mode 100644 heartbeat/writer_stop_test.go create mode 100644 monitor/plan_changer_stop_test.go diff --git a/docs/content/config/config-file.md b/docs/content/config/config-file.md index d97e157..2de0c98 100644 --- a/docs/content/config/config-file.md +++ b/docs/content/config/config-file.md @@ -194,7 +194,13 @@ A monitor that hasn't stopped by then keeps stopping in the background, and Blip Blip reports it again every 10 times the stop timeout until it stops. If the monitor is reloaded with the same ID, the new monitor doesn't start until the old one has stopped. -A healthy monitor usually stops in about 1 second, but a slow collection or a heartbeat on a read-only server can take longer than the default stop timeout. +A stop doesn't wait for the next collection, plan change check, or heartbeat: + +* The level collector stops right away. It cancels a collection in progress and drops its metrics; a canceled collection isn't counted as an engine max runtime timeout. +* The plan changer and the heartbeat writer stop right away while they wait, including while they wait to retry when MySQL is offline or read-only. + +A healthy monitor usually stops in milliseconds. +A stop can take longer while a subsystem waits on MySQL: a collector that doesn't return when its collection is canceled, a heartbeat write (at most 5 seconds), or the plan changer's read-only check (at most 10 seconds). --- diff --git a/heartbeat/writer.go b/heartbeat/writer.go index 1c8f01d..408338b 100644 --- a/heartbeat/writer.go +++ b/heartbeat/writer.go @@ -105,19 +105,15 @@ func (w *Writer) Write(stopChan, doneChan chan struct{}) error { // Error -- blip.Debug("%s: first heartbeat failed: %s", w.monitorId, err) + wait := InitErrorWait if sqlutil.ReadOnly(err) { - status.Monitor(w.monitorId, status.HEARTBEAT_WRITER, "init: MySQL is read-only, sleeping %s", ReadOnlyWait) - time.Sleep(ReadOnlyWait) + wait = ReadOnlyWait + status.Monitor(w.monitorId, status.HEARTBEAT_WRITER, "init: MySQL is read-only, sleeping %s", wait) } else { - status.Monitor(w.monitorId, status.HEARTBEAT_WRITER, "init: error: %s (sleeping %s)", err, InitErrorWait) - time.Sleep(InitErrorWait) + status.Monitor(w.monitorId, status.HEARTBEAT_WRITER, "init: error: %s (sleeping %s)", err, wait) } - - // Was Stop called? - select { - case <-stopChan: // yes, return immediately - return nil - default: // no + if !sleep(stopChan, wait) { + return nil // stopped } } @@ -131,7 +127,9 @@ func (w *Writer) Write(stopChan, doneChan chan struct{}) error { ping = fmt.Sprintf("UPDATE %s SET ts=NOW(3) WHERE src_id=?", w.table) blip.Debug("%s: heartbeat: %s", w.monitorId, ping) for { - time.Sleep(w.freq) + if !sleep(stopChan, w.freq) { + return nil // stopped + } status.Monitor(w.monitorId, status.HEARTBEAT_WRITER, "write") ctx, cancel = context.WithTimeout(context.Background(), WriteTimeout) @@ -141,7 +139,9 @@ func (w *Writer) Write(stopChan, doneChan chan struct{}) error { blip.Debug("%s: %s", w.monitorId, err.Error()) if sqlutil.ReadOnly(err) { status.Monitor(w.monitorId, status.HEARTBEAT_WRITER, "MySQL is read-only, sleeping %s", ReadOnlyWait) - time.Sleep(ReadOnlyWait) + if !sleep(stopChan, ReadOnlyWait) { + return nil // stopped + } } else { status.Monitor(w.monitorId, status.HEARTBEAT_WRITER, "write error: %s", err) // No special sleep on random errors; keep trying to write at freq @@ -151,12 +151,19 @@ func (w *Writer) Write(stopChan, doneChan chan struct{}) error { // doesn't overwrite status set on Exec error; "sleep" = "write OK" status.Monitor(w.monitorId, status.HEARTBEAT_WRITER, "sleep") } + } +} - // Was Stop called? - select { - case <-stopChan: // yes, return immediately - return nil - default: // no - } +// sleep waits for d or until stopChan is closed. It returns false if stopChan +// was closed. It's the same as the monitor package sleep, which heartbeat +// can't import. +func sleep(stopChan chan struct{}, d time.Duration) bool { + t := time.NewTimer(d) + defer t.Stop() + select { + case <-stopChan: + return false + case <-t.C: + return true } } diff --git a/heartbeat/writer_stop_test.go b/heartbeat/writer_stop_test.go new file mode 100644 index 0000000..3c0b2a4 --- /dev/null +++ b/heartbeat/writer_stop_test.go @@ -0,0 +1,106 @@ +// Copyright 2026 Block, Inc. + +package heartbeat_test + +import ( + "context" + "database/sql" + "database/sql/driver" + "errors" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/go-sql-driver/mysql" + + "github.com/cashapp/blip/v2" + "github.com/cashapp/blip/v2/heartbeat" + "github.com/cashapp/blip/v2/status" +) + +// writeTestDriver is a fake SQL driver for the heartbeat writer: every write +// succeeds, or fails with a MySQL read-only error if writeTestReadOnly is set. +const writeTestDriver = "blip-heartbeat-write-test" + +var writeTestReadOnly atomic.Bool + +func init() { + sql.Register(writeTestDriver, writeTestSQLDriver{}) +} + +type writeTestSQLDriver struct{} + +func (writeTestSQLDriver) Open(string) (driver.Conn, error) { return writeTestConn{}, nil } + +type writeTestConn struct{} + +func (writeTestConn) Prepare(string) (driver.Stmt, error) { + return nil, errors.New("no prepared statements") +} +func (writeTestConn) Close() error { return nil } +func (writeTestConn) Begin() (driver.Tx, error) { return nil, errors.New("no transactions") } +func (writeTestConn) ExecContext(context.Context, string, []driver.NamedValue) (driver.Result, error) { + if writeTestReadOnly.Load() { + return nil, &mysql.MySQLError{Number: 1290, Message: "read-only"} + } + return driver.RowsAffected(1), nil +} + +// stopWriter starts w, waits until its status contains want, then stops it. +// It returns how long Write took to return after stop. +func stopWriter(t *testing.T, w *heartbeat.Writer, monitorId, want string) time.Duration { + t.Helper() + stopChan := make(chan struct{}) + doneChan := make(chan struct{}) + go w.Write(stopChan, doneChan) + + deadline := time.Now().Add(5 * time.Second) + for !strings.Contains(status.ReportMonitors(monitorId)[monitorId][status.HEARTBEAT_WRITER], want) { + if time.Now().After(deadline) { + close(stopChan) + t.Fatalf("timeout waiting for heartbeat writer status %q", want) + } + time.Sleep(5 * time.Millisecond) + } + + t0 := time.Now() + close(stopChan) + select { + case <-doneChan: + case <-time.After(time.Minute): + t.Fatal("heartbeat writer did not stop") + } + return time.Since(t0) +} + +func TestWriterStopsDuringWaits(t *testing.T) { + // Stopping the writer must not wait for its sleep: the heartbeat + // frequency, or the read-only wait (20s by default). + const monitorId = "hb-write-stop" + defer status.RemoveMonitor(monitorId) + db, err := sql.Open(writeTestDriver, "") + if err != nil { + t.Fatal(err) + } + defer db.Close() + + const wait = 30 * time.Second + defer func(d time.Duration) { heartbeat.ReadOnlyWait = d }(heartbeat.ReadOnlyWait) + heartbeat.ReadOnlyWait = wait + + // Read-only: the first write fails, and the writer sleeps ReadOnlyWait + writeTestReadOnly.Store(true) + defer writeTestReadOnly.Store(false) + w := heartbeat.NewWriter(monitorId, db, blip.ConfigHeartbeat{Freq: "1s", Table: "heartbeat"}) + if d := stopWriter(t, w, monitorId, "read-only"); d > wait/10 { + t.Errorf("stop during read-only wait took %s, expected well under %s", d, wait) + } + + // Writable: the writer sleeps the heartbeat frequency between writes + writeTestReadOnly.Store(false) + w = heartbeat.NewWriter(monitorId, db, blip.ConfigHeartbeat{Freq: wait.String(), Table: "heartbeat"}) + if d := stopWriter(t, w, monitorId, "sleep"); d > wait/10 { + t.Errorf("stop during heartbeat sleep took %s, expected well under %s", d, wait) + } +} diff --git a/monitor/plan_changer.go b/monitor/plan_changer.go index 9d3b9c4..3391557 100644 --- a/monitor/plan_changer.go +++ b/monitor/plan_changer.go @@ -141,10 +141,12 @@ func (pch *planChanger) Run(stopChan, doneChan chan struct{}) error { pch.CheckState() - if pch.curr.state != blip.STATE_OFFLINE { - time.Sleep(1 * time.Second) - } else { - time.Sleep(pch.retry.NextBackOff()) + wait := 1 * time.Second + if pch.curr.state == blip.STATE_OFFLINE { + wait = pch.retry.NextBackOff() + } + if !sleep(stopChan, wait) { + return nil // stopped } } } diff --git a/monitor/plan_changer_stop_test.go b/monitor/plan_changer_stop_test.go new file mode 100644 index 0000000..b2898f1 --- /dev/null +++ b/monitor/plan_changer_stop_test.go @@ -0,0 +1,68 @@ +// Copyright 2026 Block, Inc. + +package monitor + +import ( + "database/sql" + "database/sql/driver" + "errors" + "testing" + "time" + + "github.com/cashapp/blip/v2/ha" + "github.com/cashapp/blip/v2/status" +) + +// offlineTestDriver is a fake SQL driver that can't connect, so the PCH sees +// MySQL offline. +const offlineTestDriver = "blip-monitor-offline-test" + +func init() { + sql.Register(offlineTestDriver, offlineTestSQLDriver{}) +} + +type offlineTestSQLDriver struct{} + +func (offlineTestSQLDriver) Open(string) (driver.Conn, error) { return nil, errors.New("offline") } + +func TestPlanChangerStopsDuringOfflineWait(t *testing.T) { + // While MySQL is offline, the PCH waits its retry backoff (up to 60s) + // between state checks. Stopping it must not wait for that. + const monitorId = "pch-stop" + defer status.RemoveMonitor(monitorId) + db, err := sql.Open(offlineTestDriver, "") + if err != nil { + t.Fatal(err) + } + defer db.Close() + + const wait = time.Minute + pch := NewPlanChanger(PlanChangerArgs{MonitorId: monitorId, DB: db, HA: ha.Disabled}) + pch.retry.InitialInterval = wait + pch.retry.MaxInterval = wait + pch.retry.Reset() + + stopChan := make(chan struct{}) + doneChan := make(chan struct{}) + go pch.Run(stopChan, doneChan) + + // Wait for the first state check (offline), then stop during the wait + deadline := time.Now().Add(5 * time.Second) + for status.ReportMonitors(monitorId)[monitorId]["error:"+status.PLAN_CHANGER] == "" { + if time.Now().After(deadline) { + close(stopChan) + t.Fatal("timeout waiting for PCH state check") + } + time.Sleep(5 * time.Millisecond) + } + t0 := time.Now() + close(stopChan) + select { + case <-doneChan: + case <-time.After(2 * wait): + t.Fatal("PCH did not stop") + } + if d := time.Since(t0); d > wait/10 { + t.Errorf("stop during offline wait took %s, expected well under %s", d, wait) + } +} From 02af430bca8ce3ccf18cee4f8e351678d97915c9 Mon Sep 17 00:00:00 2001 From: Ian Oberst Date: Mon, 26 Oct 2026 11:57:35 -0700 Subject: [PATCH 04/12] Cancel plan changer and heartbeat queries on stop - The plan changer's read-only query (10s timeout) and heartbeat writes (WriteTimeout, 5s) used contexts from context.Background, so on a hung or partitioned MySQL a stop waited for them, longer than the 3s default stop timeout. Both now derive from a context that's canceled when stopChan closes, like the level collector's collection context. A canceled state check doesn't change the PCH state or set its error. The read-only query timeout is now the readOnlyTimeout var so a test can raise it. - New package internal/stopchan has Sleep (moved from monitor, and the copy in heartbeat removed) and Context, used by the LCO, PCH, and heartbeat writer. It's internal so that it doesn't add public API. - StopAfterTicks closes stopChan once, even if more than one LCO reaches the tick. - New tests TestPlanChangerStopsDuringStateQuery and TestWriterStopsDuringWrite use a fake driver that blocks until its context is canceled. - The stop-timeout docs no longer list these as slow cases. Co-Authored-By: Claude Opus 5.5 --- docs/content/config/config-file.md | 6 +-- heartbeat/writer.go | 36 +++++++------- heartbeat/writer_stop_test.go | 77 ++++++++++++++++++++++++++++-- internal/stopchan/stopchan.go | 39 +++++++++++++++ internal/stopchan/stopchan_test.go | 47 ++++++++++++++++++ monitor/export_test.go | 8 +++- monitor/level_collector.go | 10 +--- monitor/monitor.go | 22 ++------- monitor/plan_changer.go | 35 ++++++++++---- monitor/plan_changer_stop_test.go | 67 ++++++++++++++++++++++++++ 10 files changed, 286 insertions(+), 61 deletions(-) create mode 100644 internal/stopchan/stopchan.go create mode 100644 internal/stopchan/stopchan_test.go diff --git a/docs/content/config/config-file.md b/docs/content/config/config-file.md index 2de0c98..00e5fd0 100644 --- a/docs/content/config/config-file.md +++ b/docs/content/config/config-file.md @@ -197,10 +197,10 @@ If the monitor is reloaded with the same ID, the new monitor doesn't start until A stop doesn't wait for the next collection, plan change check, or heartbeat: * The level collector stops right away. It cancels a collection in progress and drops its metrics; a canceled collection isn't counted as an engine max runtime timeout. -* The plan changer and the heartbeat writer stop right away while they wait, including while they wait to retry when MySQL is offline or read-only. +* The plan changer and the heartbeat writer stop right away while they wait, including while they wait to retry when MySQL is offline or read-only. A stop cancels the plan changer's read-only check and a heartbeat write in progress, so it doesn't wait for a hung or unreachable MySQL. -A healthy monitor usually stops in milliseconds. -A stop can take longer while a subsystem waits on MySQL: a collector that doesn't return when its collection is canceled, a heartbeat write (at most 5 seconds), or the plan changer's read-only check (at most 10 seconds). +A monitor usually stops in milliseconds, even when MySQL is hung or unreachable. +A stop can take longer only if a collector doesn't return when its collection is canceled. --- diff --git a/heartbeat/writer.go b/heartbeat/writer.go index 408338b..d816d3d 100644 --- a/heartbeat/writer.go +++ b/heartbeat/writer.go @@ -9,6 +9,7 @@ import ( "time" "github.com/cashapp/blip/v2" + "github.com/cashapp/blip/v2/internal/stopchan" "github.com/cashapp/blip/v2/sqlutil" "github.com/cashapp/blip/v2/status" ) @@ -79,6 +80,11 @@ func (w *Writer) Write(stopChan, doneChan chan struct{}) error { cancel context.CancelFunc ) + // Writes use a context that's canceled on stop, so a stop doesn't wait + // for a write to a hung or unreachable MySQL (up to WriteTimeout) + stopCtx, stopCancel := stopchan.Context(stopChan) + defer stopCancel() + // First INSERT: either creates row if it doesn't exist for this monitor ID, // or it updates an existing row with the current timestamp and frequency. // This must be done else the simpler UPDATE statements below, which is the @@ -95,9 +101,12 @@ func (w *Writer) Write(stopChan, doneChan chan struct{}) error { blip.Debug("%s: first heartbeat: %s", w.monitorId, ping) for { status.Monitor(w.monitorId, status.HEARTBEAT_WRITER, "first insert") - ctx, cancel = context.WithTimeout(context.Background(), WriteTimeout) + ctx, cancel = context.WithTimeout(stopCtx, WriteTimeout) _, err = w.db.ExecContext(ctx, ping, params...) cancel() + if stopCtx.Err() != nil { + return nil // stopped + } if err == nil { // success status.Monitor(w.monitorId, status.HEARTBEAT_WRITER, "sleep") break @@ -112,7 +121,7 @@ func (w *Writer) Write(stopChan, doneChan chan struct{}) error { } else { status.Monitor(w.monitorId, status.HEARTBEAT_WRITER, "init: error: %s (sleeping %s)", err, wait) } - if !sleep(stopChan, wait) { + if !stopchan.Sleep(stopChan, wait) { return nil // stopped } } @@ -127,19 +136,22 @@ func (w *Writer) Write(stopChan, doneChan chan struct{}) error { ping = fmt.Sprintf("UPDATE %s SET ts=NOW(3) WHERE src_id=?", w.table) blip.Debug("%s: heartbeat: %s", w.monitorId, ping) for { - if !sleep(stopChan, w.freq) { + if !stopchan.Sleep(stopChan, w.freq) { return nil // stopped } status.Monitor(w.monitorId, status.HEARTBEAT_WRITER, "write") - ctx, cancel = context.WithTimeout(context.Background(), WriteTimeout) + ctx, cancel = context.WithTimeout(stopCtx, WriteTimeout) _, err = w.db.ExecContext(ctx, ping, w.srcId) cancel() + if stopCtx.Err() != nil { + return nil // stopped + } if err != nil { blip.Debug("%s: %s", w.monitorId, err.Error()) if sqlutil.ReadOnly(err) { status.Monitor(w.monitorId, status.HEARTBEAT_WRITER, "MySQL is read-only, sleeping %s", ReadOnlyWait) - if !sleep(stopChan, ReadOnlyWait) { + if !stopchan.Sleep(stopChan, ReadOnlyWait) { return nil // stopped } } else { @@ -153,17 +165,3 @@ func (w *Writer) Write(stopChan, doneChan chan struct{}) error { } } } - -// sleep waits for d or until stopChan is closed. It returns false if stopChan -// was closed. It's the same as the monitor package sleep, which heartbeat -// can't import. -func sleep(stopChan chan struct{}, d time.Duration) bool { - t := time.NewTimer(d) - defer t.Stop() - select { - case <-stopChan: - return false - case <-t.C: - return true - } -} diff --git a/heartbeat/writer_stop_test.go b/heartbeat/writer_stop_test.go index 3c0b2a4..2cecfb1 100644 --- a/heartbeat/writer_stop_test.go +++ b/heartbeat/writer_stop_test.go @@ -20,10 +20,15 @@ import ( ) // writeTestDriver is a fake SQL driver for the heartbeat writer: every write -// succeeds, or fails with a MySQL read-only error if writeTestReadOnly is set. +// succeeds, or fails with a MySQL read-only error if writeTestReadOnly is set, +// or blocks until its context is canceled if writeTestHung is set. const writeTestDriver = "blip-heartbeat-write-test" -var writeTestReadOnly atomic.Bool +var ( + writeTestReadOnly atomic.Bool + writeTestHung atomic.Bool + writeTestHungExec = make(chan struct{}, 10) // a hung write has started +) func init() { sql.Register(writeTestDriver, writeTestSQLDriver{}) @@ -40,7 +45,12 @@ func (writeTestConn) Prepare(string) (driver.Stmt, error) { } func (writeTestConn) Close() error { return nil } func (writeTestConn) Begin() (driver.Tx, error) { return nil, errors.New("no transactions") } -func (writeTestConn) ExecContext(context.Context, string, []driver.NamedValue) (driver.Result, error) { +func (writeTestConn) ExecContext(ctx context.Context, _ string, _ []driver.NamedValue) (driver.Result, error) { + if writeTestHung.Load() { + writeTestHungExec <- struct{}{} + <-ctx.Done() + return nil, ctx.Err() + } if writeTestReadOnly.Load() { return nil, &mysql.MySQLError{Number: 1290, Message: "read-only"} } @@ -104,3 +114,64 @@ func TestWriterStopsDuringWaits(t *testing.T) { t.Errorf("stop during heartbeat sleep took %s, expected well under %s", d, wait) } } + +func TestWriterStopsDuringWrite(t *testing.T) { + // A write can wait up to WriteTimeout on a hung or partitioned MySQL. + // Stopping the writer cancels it instead of waiting. + const monitorId = "hb-write-stop-hung" + defer status.RemoveMonitor(monitorId) + db, err := sql.Open(writeTestDriver, "") + if err != nil { + t.Fatal(err) + } + defer db.Close() + + const timeout = 30 * time.Second + defer func(d time.Duration) { heartbeat.WriteTimeout = d }(heartbeat.WriteTimeout) + heartbeat.WriteTimeout = timeout + + writeTestHung.Store(true) + defer writeTestHung.Store(false) + + // First write (INSERT) hangs. The status is set before the write, so the + // stop can come just before the write starts, which tests stopping it too. + w := heartbeat.NewWriter(monitorId, db, blip.ConfigHeartbeat{Freq: "1s", Table: "heartbeat"}) + if d := stopWriter(t, w, monitorId, "first insert"); d > timeout/10 { + t.Errorf("stop during first write took %s, expected well under %s", d, timeout) + } + for len(writeTestHungExec) > 0 { + <-writeTestHungExec + } + + // A heartbeat (UPDATE) hangs + writeTestHung.Store(false) + w = heartbeat.NewWriter(monitorId, db, blip.ConfigHeartbeat{Freq: "10ms", Table: "heartbeat"}) + stopChan := make(chan struct{}) + doneChan := make(chan struct{}) + go w.Write(stopChan, doneChan) + deadline := time.Now().Add(5 * time.Second) + for status.ReportMonitors(monitorId)[monitorId][status.HEARTBEAT_WRITER] != "sleep" { + if time.Now().After(deadline) { + close(stopChan) + t.Fatal("timeout waiting for first write") + } + time.Sleep(time.Millisecond) + } + writeTestHung.Store(true) + select { + case <-writeTestHungExec: + case <-time.After(5 * time.Second): + close(stopChan) + t.Fatal("timeout waiting for hung heartbeat") + } + t0 := time.Now() + close(stopChan) + select { + case <-doneChan: + case <-time.After(2 * timeout): + t.Fatal("heartbeat writer did not stop") + } + if d := time.Since(t0); d > timeout/10 { + t.Errorf("stop during heartbeat write took %s, expected well under %s", d, timeout) + } +} diff --git a/internal/stopchan/stopchan.go b/internal/stopchan/stopchan.go new file mode 100644 index 0000000..d7b915c --- /dev/null +++ b/internal/stopchan/stopchan.go @@ -0,0 +1,39 @@ +// Copyright 2026 Block, Inc. + +// Package stopchan has helpers for goroutines that stop when a stop channel is +// closed, like the monitor subsystems. It's internal so that both the monitor +// and heartbeat packages can use it without adding to Blip's public API. +package stopchan + +import ( + "context" + "time" +) + +// Sleep waits for d or until stopChan is closed. It returns false if stopChan +// was closed. +func Sleep(stopChan <-chan struct{}, d time.Duration) bool { + t := time.NewTimer(d) + defer t.Stop() + select { + case <-stopChan: + return false + case <-t.C: + return true + } +} + +// Context returns a context that's canceled when stopChan is closed, so a +// query or other call that takes the context returns on stop. Call cancel when +// done with the context to release its resources. +func Context(stopChan <-chan struct{}) (ctx context.Context, cancel context.CancelFunc) { + ctx, cancel = context.WithCancel(context.Background()) + go func() { + select { + case <-stopChan: + cancel() + case <-ctx.Done(): + } + }() + return ctx, cancel +} diff --git a/internal/stopchan/stopchan_test.go b/internal/stopchan/stopchan_test.go new file mode 100644 index 0000000..300e407 --- /dev/null +++ b/internal/stopchan/stopchan_test.go @@ -0,0 +1,47 @@ +// Copyright 2026 Block, Inc. + +package stopchan_test + +import ( + "testing" + "time" + + "github.com/cashapp/blip/v2/internal/stopchan" +) + +func TestSleep(t *testing.T) { + stopChan := make(chan struct{}) + if !stopchan.Sleep(stopChan, time.Millisecond) { + t.Error("Sleep returned false before stop") + } + close(stopChan) + t0 := time.Now() + if stopchan.Sleep(stopChan, time.Minute) { + t.Error("Sleep returned true after stop") + } + if d := time.Since(t0); d > time.Second { + t.Errorf("Sleep after stop took %s", d) + } +} + +func TestContext(t *testing.T) { + stopChan := make(chan struct{}) + ctx, cancel := stopchan.Context(stopChan) + defer cancel() + if ctx.Err() != nil { + t.Fatalf("context done before stop: %s", ctx.Err()) + } + close(stopChan) + select { + case <-ctx.Done(): + case <-time.After(5 * time.Second): + t.Fatal("context not canceled on stop") + } + + // cancel without stop + ctx, cancel = stopchan.Context(make(chan struct{})) + cancel() + if ctx.Err() == nil { + t.Error("context not canceled by cancel") + } +} diff --git a/monitor/export_test.go b/monitor/export_test.go index d80b585..480ed8a 100644 --- a/monitor/export_test.go +++ b/monitor/export_test.go @@ -2,14 +2,18 @@ package monitor +import "sync" + // StopAfterTicks makes every LCO Run return at tick n+1, before it collects at // that tick, by closing stopChan. Use it for a test that runs one LCO, and call -// the returned func to remove it. +// the returned func to remove it. If more than one LCO reaches tick n+1, +// stopChan is closed only once. func StopAfterTicks(n uint, stopChan chan struct{}) func() { + var once sync.Once tickerMux.Lock() tickHook = func(tick uint) { if tick == n+1 { - close(stopChan) + once.Do(func() { close(stopChan) }) } } tickerMux.Unlock() diff --git a/monitor/level_collector.go b/monitor/level_collector.go index 20ded6d..3c2b712 100644 --- a/monitor/level_collector.go +++ b/monitor/level_collector.go @@ -15,6 +15,7 @@ import ( "github.com/cashapp/blip/v2" "github.com/cashapp/blip/v2/event" + "github.com/cashapp/blip/v2/internal/stopchan" "github.com/cashapp/blip/v2/plan" "github.com/cashapp/blip/v2/status" ) @@ -248,15 +249,8 @@ func (c *lco) Run(stopChan, doneChan chan struct{}) error { // collectCtx is canceled when stopChan is closed, so a stop doesn't wait for // a collection in progress - collectCtx, cancelCollect := context.WithCancel(context.Background()) + collectCtx, cancelCollect := stopchan.Context(stopChan) defer cancelCollect() - go func() { - select { - case <-stopChan: - cancelCollect() - case <-collectCtx.Done(): - } - }() var lastTick time.Time tick := uint(0) diff --git a/monitor/monitor.go b/monitor/monitor.go index 11dd544..9011841 100644 --- a/monitor/monitor.go +++ b/monitor/monitor.go @@ -21,6 +21,7 @@ import ( "github.com/cashapp/blip/v2/event" "github.com/cashapp/blip/v2/ha" "github.com/cashapp/blip/v2/heartbeat" + "github.com/cashapp/blip/v2/internal/stopchan" "github.com/cashapp/blip/v2/plan" "github.com/cashapp/blip/v2/prom" "github.com/cashapp/blip/v2/sink" @@ -236,7 +237,7 @@ func (m *Monitor) runLoop(stopChan, doneChan chan struct{}) { m.setErr(err, false) if err != nil { - if !sleep(stopChan, m.retry.NextBackOff()) { + if !stopchan.Sleep(stopChan, m.retry.NextBackOff()) { return } continue @@ -254,26 +255,13 @@ func (m *Monitor) runLoop(stopChan, doneChan chan struct{}) { return case <-runChan: // internal failure blip.Debug("%s: runChan closed; restarting", m.monitorId) - if !sleep(stopChan, 1*time.Second) { // between monitor restarts + if !stopchan.Sleep(stopChan, 1*time.Second) { // between monitor restarts return } } } } -// sleep waits for d or until stopChan is closed. It returns false if stopChan -// was closed. -func sleep(stopChan chan struct{}, d time.Duration) bool { - t := time.NewTimer(d) - defer t.Stop() - select { - case <-stopChan: - return false - case <-t.C: - return true - } -} - // startup starts the monitor subsystems, which are optional depending on config: // heartbeat writer, exporter API (Prometheus emulation), PCH, and LCO. // The monitor is running once these have started. If any subsystem crashes @@ -321,7 +309,7 @@ func (m *Monitor) startup(stopChan chan struct{}) (err error) { break } status.Monitor(m.monitorId, status.MONITOR, "error making DB/DSN, sleep and retry: %s", err) - if !sleep(stopChan, m.retry.NextBackOff()) { + if !stopchan.Sleep(stopChan, m.retry.NextBackOff()) { return nil // runLoop stopped (Stop called) } } @@ -337,7 +325,7 @@ func (m *Monitor) startup(stopChan chan struct{}) (err error) { break } status.Monitor(m.monitorId, status.MONITOR, "error loading plans, sleep and retry: %s", err) - if !sleep(stopChan, m.retry.NextBackOff()) { + if !stopchan.Sleep(stopChan, m.retry.NextBackOff()) { return nil // runLoop stopped (Stop called) } } diff --git a/monitor/plan_changer.go b/monitor/plan_changer.go index 3391557..59fc67f 100644 --- a/monitor/plan_changer.go +++ b/monitor/plan_changer.go @@ -13,6 +13,7 @@ import ( "github.com/cashapp/blip/v2" "github.com/cashapp/blip/v2/event" "github.com/cashapp/blip/v2/ha" + "github.com/cashapp/blip/v2/internal/stopchan" "github.com/cashapp/blip/v2/status" ) @@ -132,6 +133,11 @@ func (pch *planChanger) Run(stopChan, doneChan chan struct{}) error { status.Monitor(pch.monitorId, status.PLAN_CHANGER, "running") + // The state check uses a context that's canceled on stop, so a stop + // doesn't wait for the read-only query to a hung or unreachable MySQL + ctx, cancel := stopchan.Context(stopChan) + defer cancel() + for { select { case <-stopChan: @@ -139,23 +145,27 @@ func (pch *planChanger) Run(stopChan, doneChan chan struct{}) error { default: } - pch.CheckState() + pch.CheckState(ctx) wait := 1 * time.Second if pch.curr.state == blip.STATE_OFFLINE { wait = pch.retry.NextBackOff() } - if !sleep(stopChan, wait) { + if !stopchan.Sleep(stopChan, wait) { return nil // stopped } } } // CheckState checks the current state and changes state when necessary. -// This is the main logic, called periodically by Run. -func (pch *planChanger) CheckState() { +// This is the main logic, called periodically by Run. It does nothing if ctx +// is canceled (the PCH is stopping) while it checks the state. +func (pch *planChanger) CheckState(ctx context.Context) { now := Now() - obsv := pch.state() + obsv := pch.state(ctx) + if ctx.Err() != nil { + return // stopped + } pch.Lock() defer pch.Unlock() @@ -239,8 +249,12 @@ func (pch *planChanger) lcoChangePlan(state, planName string) error { const readOnlyQuery = "SELECT @@read_only, @@super_read_only" -// state queries MySQL to ascertain the HA and read-only state. -func (pch *planChanger) state() string { +// readOnlyTimeout is how long the PCH waits for readOnlyQuery +var readOnlyTimeout = 10 * time.Second + +// state queries MySQL to ascertain the HA and read-only state. If ctx is +// canceled, it returns STATE_NONE, which the caller must ignore. +func (pch *planChanger) state(ctx context.Context) string { status.Monitor(pch.monitorId, status.PLAN_CHANGER, "checking HA standby") if pch.ha.Standby() { return blip.STATE_STANDBY @@ -250,9 +264,12 @@ func (pch *planChanger) state() string { status.Monitor(pch.monitorId, status.PLAN_CHANGER, "checking MySQL read-only") var ro, sro int - ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) - err := pch.db.QueryRowContext(ctx, readOnlyQuery).Scan(&ro, &sro) + qctx, cancel := context.WithTimeout(ctx, readOnlyTimeout) + err := pch.db.QueryRowContext(qctx, readOnlyQuery).Scan(&ro, &sro) cancel() + if ctx.Err() != nil { + return blip.STATE_NONE // stopped + } pch.setErr(err) if err != nil { blip.Debug(err.Error()) diff --git a/monitor/plan_changer_stop_test.go b/monitor/plan_changer_stop_test.go index b2898f1..af024b2 100644 --- a/monitor/plan_changer_stop_test.go +++ b/monitor/plan_changer_stop_test.go @@ -3,6 +3,7 @@ package monitor import ( + "context" "database/sql" "database/sql/driver" "errors" @@ -17,9 +18,33 @@ import ( // MySQL offline. const offlineTestDriver = "blip-monitor-offline-test" +// hungTestDriver is a fake SQL driver whose queries block until their context +// is canceled, like a hung or partitioned MySQL. +const hungTestDriver = "blip-monitor-hung-test" + func init() { sql.Register(offlineTestDriver, offlineTestSQLDriver{}) + sql.Register(hungTestDriver, hungTestSQLDriver{}) +} + +type hungTestSQLDriver struct{} + +func (hungTestSQLDriver) Open(string) (driver.Conn, error) { return hungTestConn{}, nil } + +type hungTestConn struct{} + +func (hungTestConn) Prepare(string) (driver.Stmt, error) { + return nil, errors.New("no prepared statements") } +func (hungTestConn) Close() error { return nil } +func (hungTestConn) Begin() (driver.Tx, error) { return nil, errors.New("no transactions") } +func (hungTestConn) QueryContext(ctx context.Context, _ string, _ []driver.NamedValue) (driver.Rows, error) { + hungTestQueries <- struct{}{} + <-ctx.Done() + return nil, ctx.Err() +} + +var hungTestQueries = make(chan struct{}, 10) // a query has started type offlineTestSQLDriver struct{} @@ -66,3 +91,45 @@ func TestPlanChangerStopsDuringOfflineWait(t *testing.T) { t.Errorf("stop during offline wait took %s, expected well under %s", d, wait) } } + +func TestPlanChangerStopsDuringStateQuery(t *testing.T) { + // The read-only query can wait up to readOnlyTimeout on a hung or + // partitioned MySQL. Stopping the PCH cancels it instead of waiting. + const monitorId = "pch-stop-query" + defer status.RemoveMonitor(monitorId) + db, err := sql.Open(hungTestDriver, "") + if err != nil { + t.Fatal(err) + } + defer db.Close() + + const timeout = time.Minute + defer func(d time.Duration) { readOnlyTimeout = d }(readOnlyTimeout) + readOnlyTimeout = timeout + + pch := NewPlanChanger(PlanChangerArgs{MonitorId: monitorId, DB: db, HA: ha.Disabled}) + stopChan := make(chan struct{}) + doneChan := make(chan struct{}) + go pch.Run(stopChan, doneChan) + select { + case <-hungTestQueries: + case <-time.After(5 * time.Second): + close(stopChan) + t.Fatal("timeout waiting for read-only query") + } + + t0 := time.Now() + close(stopChan) + select { + case <-doneChan: + case <-time.After(2 * timeout): + t.Fatal("PCH did not stop") + } + if d := time.Since(t0); d > timeout/10 { + t.Errorf("stop during read-only query took %s, expected well under %s", d, timeout) + } + // A canceled query isn't an error: the PCH didn't see MySQL offline + if e := status.ReportMonitors(monitorId)[monitorId]["error:"+status.PLAN_CHANGER]; e != "" { + t.Errorf("PCH error status %q after stop, expected none", e) + } +} From dde2b3bef8e74c85583f480973b5751fc1f65983 Mon Sep 17 00:00:00 2001 From: Ian Oberst Date: Mon, 26 Oct 2026 12:56:38 -0700 Subject: [PATCH 05/12] Use one stop context per loop and fix the PCH test race - The read-only query timeout is now a planChanger field, readOnlyTimeout, instead of a package var that the test wrote while PCH goroutines read it. - The plan changer and heartbeat writer loops use only their stop context: they sleep with stopchan.Sleep(ctx.Done(), d) and check ctx.Err(), instead of mixing stopChan and the context. - The stop-timeout docs name the remaining slow case for a monitor that's starting: a stop waits for it to connect (up to timeout-connect) and load its plan. Co-Authored-By: Claude Opus 5.5 --- docs/content/config/config-file.md | 7 +++++-- heartbeat/writer.go | 10 +++++----- internal/stopchan/stopchan.go | 2 +- monitor/plan_changer.go | 21 ++++++++++----------- monitor/plan_changer_stop_test.go | 6 ++---- 5 files changed, 23 insertions(+), 23 deletions(-) diff --git a/docs/content/config/config-file.md b/docs/content/config/config-file.md index 00e5fd0..0ec030b 100644 --- a/docs/content/config/config-file.md +++ b/docs/content/config/config-file.md @@ -199,8 +199,11 @@ A stop doesn't wait for the next collection, plan change check, or heartbeat: * The level collector stops right away. It cancels a collection in progress and drops its metrics; a canceled collection isn't counted as an engine max runtime timeout. * The plan changer and the heartbeat writer stop right away while they wait, including while they wait to retry when MySQL is offline or read-only. A stop cancels the plan changer's read-only check and a heartbeat write in progress, so it doesn't wait for a hung or unreachable MySQL. -A monitor usually stops in milliseconds, even when MySQL is hung or unreachable. -A stop can take longer only if a collector doesn't return when its collection is canceled. +A running monitor usually stops in milliseconds, even when MySQL is hung or unreachable. +A stop can take longer in two cases: + +* A collector doesn't return when its collection is canceled. +* The monitor is starting. A stop waits for it to connect to MySQL, which can take up to [`timeout-connect`](#timeout-connect) (default 10 seconds) when MySQL is unreachable, and to load its plan, which can wait on a hung MySQL if the plan is in a table. --- diff --git a/heartbeat/writer.go b/heartbeat/writer.go index d816d3d..ae976cc 100644 --- a/heartbeat/writer.go +++ b/heartbeat/writer.go @@ -80,8 +80,8 @@ func (w *Writer) Write(stopChan, doneChan chan struct{}) error { cancel context.CancelFunc ) - // Writes use a context that's canceled on stop, so a stop doesn't wait - // for a write to a hung or unreachable MySQL (up to WriteTimeout) + // stopCtx is canceled on stop, so a stop doesn't wait for a write to a + // hung or unreachable MySQL (up to WriteTimeout), or for a sleep stopCtx, stopCancel := stopchan.Context(stopChan) defer stopCancel() @@ -121,7 +121,7 @@ func (w *Writer) Write(stopChan, doneChan chan struct{}) error { } else { status.Monitor(w.monitorId, status.HEARTBEAT_WRITER, "init: error: %s (sleeping %s)", err, wait) } - if !stopchan.Sleep(stopChan, wait) { + if !stopchan.Sleep(stopCtx.Done(), wait) { return nil // stopped } } @@ -136,7 +136,7 @@ func (w *Writer) Write(stopChan, doneChan chan struct{}) error { ping = fmt.Sprintf("UPDATE %s SET ts=NOW(3) WHERE src_id=?", w.table) blip.Debug("%s: heartbeat: %s", w.monitorId, ping) for { - if !stopchan.Sleep(stopChan, w.freq) { + if !stopchan.Sleep(stopCtx.Done(), w.freq) { return nil // stopped } @@ -151,7 +151,7 @@ func (w *Writer) Write(stopChan, doneChan chan struct{}) error { blip.Debug("%s: %s", w.monitorId, err.Error()) if sqlutil.ReadOnly(err) { status.Monitor(w.monitorId, status.HEARTBEAT_WRITER, "MySQL is read-only, sleeping %s", ReadOnlyWait) - if !stopchan.Sleep(stopChan, ReadOnlyWait) { + if !stopchan.Sleep(stopCtx.Done(), ReadOnlyWait) { return nil // stopped } } else { diff --git a/internal/stopchan/stopchan.go b/internal/stopchan/stopchan.go index d7b915c..dd57897 100644 --- a/internal/stopchan/stopchan.go +++ b/internal/stopchan/stopchan.go @@ -11,7 +11,7 @@ import ( ) // Sleep waits for d or until stopChan is closed. It returns false if stopChan -// was closed. +// was closed. To sleep until a context is done, pass ctx.Done() as stopChan. func Sleep(stopChan <-chan struct{}, d time.Duration) bool { t := time.NewTimer(d) defer t.Stop() diff --git a/monitor/plan_changer.go b/monitor/plan_changer.go index 59fc67f..a5ba11c 100644 --- a/monitor/plan_changer.go +++ b/monitor/plan_changer.go @@ -67,6 +67,8 @@ type planChanger struct { event event.MonitorReceiver retry *backoff.ExponentialBackOff lerr error + + readOnlyTimeout time.Duration // how long to wait for readOnlyQuery } func NewPlanChanger(args PlanChangerArgs) *planChanger { @@ -110,6 +112,8 @@ func NewPlanChanger(args PlanChangerArgs) *planChanger { first: true, event: event.MonitorReceiver{MonitorId: args.MonitorId}, retry: retry, + + readOnlyTimeout: 10 * time.Second, } } @@ -133,16 +137,14 @@ func (pch *planChanger) Run(stopChan, doneChan chan struct{}) error { status.Monitor(pch.monitorId, status.PLAN_CHANGER, "running") - // The state check uses a context that's canceled on stop, so a stop - // doesn't wait for the read-only query to a hung or unreachable MySQL + // ctx is canceled on stop, so a stop doesn't wait for the read-only query + // to a hung or unreachable MySQL, or for the sleep between checks ctx, cancel := stopchan.Context(stopChan) defer cancel() for { - select { - case <-stopChan: - return nil - default: + if ctx.Err() != nil { + return nil // stopped } pch.CheckState(ctx) @@ -151,7 +153,7 @@ func (pch *planChanger) Run(stopChan, doneChan chan struct{}) error { if pch.curr.state == blip.STATE_OFFLINE { wait = pch.retry.NextBackOff() } - if !stopchan.Sleep(stopChan, wait) { + if !stopchan.Sleep(ctx.Done(), wait) { return nil // stopped } } @@ -249,9 +251,6 @@ func (pch *planChanger) lcoChangePlan(state, planName string) error { const readOnlyQuery = "SELECT @@read_only, @@super_read_only" -// readOnlyTimeout is how long the PCH waits for readOnlyQuery -var readOnlyTimeout = 10 * time.Second - // state queries MySQL to ascertain the HA and read-only state. If ctx is // canceled, it returns STATE_NONE, which the caller must ignore. func (pch *planChanger) state(ctx context.Context) string { @@ -264,7 +263,7 @@ func (pch *planChanger) state(ctx context.Context) string { status.Monitor(pch.monitorId, status.PLAN_CHANGER, "checking MySQL read-only") var ro, sro int - qctx, cancel := context.WithTimeout(ctx, readOnlyTimeout) + qctx, cancel := context.WithTimeout(ctx, pch.readOnlyTimeout) err := pch.db.QueryRowContext(qctx, readOnlyQuery).Scan(&ro, &sro) cancel() if ctx.Err() != nil { diff --git a/monitor/plan_changer_stop_test.go b/monitor/plan_changer_stop_test.go index af024b2..0c5f8ad 100644 --- a/monitor/plan_changer_stop_test.go +++ b/monitor/plan_changer_stop_test.go @@ -93,7 +93,7 @@ func TestPlanChangerStopsDuringOfflineWait(t *testing.T) { } func TestPlanChangerStopsDuringStateQuery(t *testing.T) { - // The read-only query can wait up to readOnlyTimeout on a hung or + // The read-only query can wait up to the read-only timeout on a hung or // partitioned MySQL. Stopping the PCH cancels it instead of waiting. const monitorId = "pch-stop-query" defer status.RemoveMonitor(monitorId) @@ -104,10 +104,8 @@ func TestPlanChangerStopsDuringStateQuery(t *testing.T) { defer db.Close() const timeout = time.Minute - defer func(d time.Duration) { readOnlyTimeout = d }(readOnlyTimeout) - readOnlyTimeout = timeout - pch := NewPlanChanger(PlanChangerArgs{MonitorId: monitorId, DB: db, HA: ha.Disabled}) + pch.readOnlyTimeout = timeout stopChan := make(chan struct{}) doneChan := make(chan struct{}) go pch.Run(stopChan, doneChan) From e23ec32de376416c07f48bcaf5eae7931337b9ee Mon Sep 17 00:00:00 2001 From: Ian Oberst Date: Mon, 26 Oct 2026 16:10:34 -0700 Subject: [PATCH 06/12] Report dropped collections and use stopchan.Sleep for plan retries - When a stop cancels a collection, the level-collect status says the metrics were dropped instead of "collected in". - Replace waitForChangePlanRetry with stopchan.Sleep, which is the same. - Name each subsystem's waits in the stop-timeout docs, and include collector cleanup functions in the slow stop case. Co-Authored-By: Claude Opus 5.5 --- docs/content/config/config-file.md | 4 ++-- monitor/level_collector.go | 22 ++++++++-------------- monitor/level_collector_stop_test.go | 5 +++++ 3 files changed, 15 insertions(+), 16 deletions(-) diff --git a/docs/content/config/config-file.md b/docs/content/config/config-file.md index 0ec030b..fc34724 100644 --- a/docs/content/config/config-file.md +++ b/docs/content/config/config-file.md @@ -197,12 +197,12 @@ If the monitor is reloaded with the same ID, the new monitor doesn't start until A stop doesn't wait for the next collection, plan change check, or heartbeat: * The level collector stops right away. It cancels a collection in progress and drops its metrics; a canceled collection isn't counted as an engine max runtime timeout. -* The plan changer and the heartbeat writer stop right away while they wait, including while they wait to retry when MySQL is offline or read-only. A stop cancels the plan changer's read-only check and a heartbeat write in progress, so it doesn't wait for a hung or unreachable MySQL. +* The plan changer and the heartbeat writer stop right away while they wait. The plan changer waits between checks, and backs off up to 60 seconds while MySQL is offline. The heartbeat writer waits between heartbeats, before it retries a first heartbeat that failed, and while MySQL is read-only. A stop cancels the plan changer's read-only check and a heartbeat write in progress, so it doesn't wait for a hung or unreachable MySQL. A running monitor usually stops in milliseconds, even when MySQL is hung or unreachable. A stop can take longer in two cases: -* A collector doesn't return when its collection is canceled. +* A collector, or its cleanup function, doesn't return promptly when its collection is canceled. A stop waits for every collector to return, then runs their cleanup functions. * The monitor is starting. A stop waits for it to connect to MySQL, which can take up to [`timeout-connect`](#timeout-connect) (default 10 seconds) when MySQL is unreachable, and to load its plan, which can wait on a hung MySQL if the plan is in a table. --- diff --git a/monitor/level_collector.go b/monitor/level_collector.go index 3c2b712..dc56749 100644 --- a/monitor/level_collector.go +++ b/monitor/level_collector.go @@ -338,12 +338,17 @@ func (c *lco) stopChangePlan() { // metrics are dropped. func (c *lco) collect(ctx context.Context, interval uint, levelName string, startTime time.Time) { status.Monitor(c.monitorId, status.LEVEL_COLLECT, "%s/%s: collecting", c.plan.Name, levelName) + dropped := false // stopped during collection defer func() { if err := recover(); err != nil { // catch panic in engine and TransformMetrics b := make([]byte, 4096) n := runtime.Stack(b, false) c.event.Errorf(event.LCO_COLLECT_PANIC, "PANIC: %s: %s\n%s", c.monitorId, err, string(b[0:n])) } + if dropped { + status.Monitor(c.monitorId, status.LEVEL_COLLECT, "%s/%s/%d: stopped; metrics dropped", c.plan.Name, levelName, interval) + return + } status.Monitor(c.monitorId, status.LEVEL_COLLECT, "%s/%s/%d: collected in %s", c.plan.Name, levelName, interval, time.Now().Sub(startTime)) }() @@ -358,6 +363,7 @@ func (c *lco) collect(ctx context.Context, interval uint, levelName string, star blip.Debug("%s: level %s: done in %s", c.monitorId, levelName, metrics[0].End.Sub(metrics[0].Begin)) if ctx.Err() != nil { blip.Debug("%s: level %s: stopped, dropping metrics", c.monitorId, levelName) + dropped = true return } @@ -471,7 +477,7 @@ func (c *lco) changePlan(ctx context.Context, doneChan chan struct{}, newState, errMsg := fmt.Sprintf("%s: error loading new plan %s: %s (retrying)", change, newPlanName, err) status.Monitor(c.monitorId, status.LEVEL_CHANGE_PLAN, "%s", errMsg) c.event.Sendf(event.CHANGE_PLAN_ERROR, "%s", errMsg) - if !waitForChangePlanRetry(ctx, 2*time.Second) { + if !stopchan.Sleep(ctx.Done(), 2*time.Second) { blip.Debug("changePlan canceled while loading plan") return } @@ -549,7 +555,7 @@ func (c *lco) changePlan(ctx context.Context, doneChan chan struct{}, newState, return // changePlan goroutine has been cancelled } status.Monitor(c.monitorId, status.LEVEL_CHANGE_PLAN, "%s: error preparing new plan %s: %s (retrying)", change, newPlan.Name, err) - if !waitForChangePlanRetry(ctx, retry.NextBackOff()) { + if !stopchan.Sleep(ctx.Done(), retry.NextBackOff()) { blip.Debug("changePlan canceled while waiting to retry plan preparation") return } @@ -559,18 +565,6 @@ func (c *lco) changePlan(ctx context.Context, doneChan chan struct{}, newState, c.event.Sendf(event.CHANGE_PLAN_SUCCESS, "%s", change) } -func waitForChangePlanRetry(ctx context.Context, delay time.Duration) bool { - timer := time.NewTimer(delay) - defer timer.Stop() - - select { - case <-ctx.Done(): - return false - case <-timer.C: - return true - } -} - // Pause pauses metrics collection until ChangePlan is called. Run still runs, // but it doesn't collect when paused. The only way to resume after pausing is // to call ChangePlan again. diff --git a/monitor/level_collector_stop_test.go b/monitor/level_collector_stop_test.go index 8348c03..c700a68 100644 --- a/monitor/level_collector_stop_test.go +++ b/monitor/level_collector_stop_test.go @@ -4,6 +4,7 @@ package monitor import ( "context" + "strings" "testing" "time" @@ -99,4 +100,8 @@ func TestLevelCollectorStopCancelsCollection(t *testing.T) { if v := counters(t, monitorId)[status.ENGINE_EMR_TIMEOUTS]; v != 0 { t.Errorf("%s = %d, expected 0", status.ENGINE_EMR_TIMEOUTS, v) } + // The canceled collection's metrics are dropped, not reported collected + if s := status.ReportMonitors(monitorId)[monitorId][status.LEVEL_COLLECT]; !strings.HasSuffix(s, ": stopped; metrics dropped") { + t.Errorf("%s status = %q, expected stopped; metrics dropped", status.LEVEL_COLLECT, s) + } } From 1681fff6fff9bb0bae8d95fc016d181f895df7da Mon Sep 17 00:00:00 2001 From: Ian Oberst Date: Tue, 27 Oct 2026 06:55:16 -0700 Subject: [PATCH 07/12] Drop unsent metrics on stop and fix stop docs - A stop drops metrics that the level collector queued for the sinks but hasn't sent yet. Run stops the sinks first when it stops, and recvMetrics checks for stop after it receives metrics and before each sink Send, so it no longer picks at random between stop and metrics. - The plan changer's offline backoff reaches about 90s (60s plus jitter), not 60s. - A stop during startup waits for plans to load from a table, the only MySQL query while starting, which takes up to 5s. - Add a release note for these stop changes. Co-Authored-By: Claude Opus 5.5 --- docs/content/about/release-notes.md | 1 + docs/content/config/config-file.md | 6 ++-- monitor/level_collector.go | 29 +++++++++++++++++-- monitor/level_collector_stop_test.go | 43 ++++++++++++++++++++++++++++ monitor/plan_changer.go | 9 +++--- monitor/plan_changer_stop_test.go | 2 +- 6 files changed, 79 insertions(+), 11 deletions(-) diff --git a/docs/content/about/release-notes.md b/docs/content/about/release-notes.md index 12c3735..dbb24dd 100644 --- a/docs/content/about/release-notes.md +++ b/docs/content/about/release-notes.md @@ -47,6 +47,7 @@ Default sink HTTP clients now have a 10-second whole-request timeout plus bounde * Changed the loader to send the `monitor-stopped` and `monitor-stop-timeout` events from its stop goroutine, so an event receiver that panics on either event crashes Blip. Previously, when the API stopped a monitor, `net/http` recovered the panic. Event receivers must not panic. * **Behavior change:** `monitor.Loader.Stop` now keeps the monitor stopped across `Load` and `StartMonitors` until `Start`, unless its config changes or it's removed and loaded again. Go callers that relied on the next reload to restart a stopped monitor must call `Start`. `POST /monitors/stop` keeps a monitor stopped across reloads, as before. * **Behavior change:** `POST /monitors/start` now calls the `StartMonitor` plugin, so it returns 409 for a monitor that the plugin declines. Previously the API started the monitor directly and skipped the plugin, so it could start a monitor that the plugin filtered out. Integrations that start filtered monitors through the API must allow them in the plugin. +* A monitor stop no longer waits for the next collection tick, plan change check, or heartbeat. It cancels a collection in progress and drops its metrics and any metrics not yet sent to sinks, and it cancels the plan changer's read-only check and a heartbeat write in progress. * **Behavior change:** Applied `timeout-connect` (default `10s`) to MySQL connections. Previously it was ignored, so connecting to an unreachable host waited for the operating system TCP connect timeout. Set it to `0` to retain the old behavior. An invalid `timeout-connect` value, which used to be ignored, now fails config validation, so Blip doesn't start, or a monitor in a reload doesn't load. For example, a bare number such as `timeout-connect: 5` has no unit and is invalid; use `5s`. * Added per-monitor fault counters, such as collector panics and timeouts, engine max runtime (EMR) timeouts, and dropped metrics, reported by the new [`GET /status/monitors/counters`]({{< ref "/api/status#get-statusmonitorscounters" >}}) API endpoint. Integrations can read them with the new `status.ReportCounters`, and register their own with `status.Counter`; the counter names are `status` constants like `status.COLLECTOR_PANICS`. * Added the optional `sink.CounterSink` interface for sinks that report monitor status counters; the monitor calls `RegisterCounters` each time it starts. A custom sink that wraps `sink.Retry` must forward `RegisterCounters` to it, else the retry buffer's `sink-retry-dropped` counter isn't reported. diff --git a/docs/content/config/config-file.md b/docs/content/config/config-file.md index fc34724..45903d1 100644 --- a/docs/content/config/config-file.md +++ b/docs/content/config/config-file.md @@ -196,14 +196,14 @@ If the monitor is reloaded with the same ID, the new monitor doesn't start until A stop doesn't wait for the next collection, plan change check, or heartbeat: -* The level collector stops right away. It cancels a collection in progress and drops its metrics; a canceled collection isn't counted as an engine max runtime timeout. -* The plan changer and the heartbeat writer stop right away while they wait. The plan changer waits between checks, and backs off up to 60 seconds while MySQL is offline. The heartbeat writer waits between heartbeats, before it retries a first heartbeat that failed, and while MySQL is read-only. A stop cancels the plan changer's read-only check and a heartbeat write in progress, so it doesn't wait for a hung or unreachable MySQL. +* The level collector stops right away. It cancels a collection in progress and drops its metrics, and it drops metrics that haven't been sent to the sinks yet (a send in progress isn't canceled). A canceled collection isn't counted as an engine max runtime timeout. +* The plan changer and the heartbeat writer stop right away while they wait. The plan changer waits between checks, and backs off while MySQL is offline (up to about 90 seconds: 60 seconds plus jitter). The heartbeat writer waits between heartbeats, before it retries a first heartbeat that failed, and while MySQL is read-only. A stop cancels the plan changer's read-only check and a heartbeat write in progress, so it doesn't wait for a hung or unreachable MySQL. A running monitor usually stops in milliseconds, even when MySQL is hung or unreachable. A stop can take longer in two cases: * A collector, or its cleanup function, doesn't return promptly when its collection is canceled. A stop waits for every collector to return, then runs their cleanup functions. -* The monitor is starting. A stop waits for it to connect to MySQL, which can take up to [`timeout-connect`](#timeout-connect) (default 10 seconds) when MySQL is unreachable, and to load its plan, which can wait on a hung MySQL if the plan is in a table. +* The monitor is starting. A stop waits for it to load its plans. If they're in a table ([`plans.table`](#table-1)), that's the only time a starting monitor queries MySQL, and it can take up to 5 seconds (connecting included) when MySQL is unreachable or hung. A custom `DbProviderFactory` that blocks while making the monitor's connections also delays the stop. --- diff --git a/monitor/level_collector.go b/monitor/level_collector.go index dc56749..6275dc6 100644 --- a/monitor/level_collector.go +++ b/monitor/level_collector.go @@ -152,6 +152,10 @@ var ( // This is a goroutine run by keepRecvMetrics and restarted by keepRecvMetrics // if a sink (or the transformMetrics) panics. It stops when stopSinksChan is // closed in Run, and it closes doneChan when stopped. +// +// When stopSinksChan is closed, metrics not yet sent are dropped, so a stopped +// or removed monitor doesn't send metrics. recvMetrics checks after it receives +// metrics and before each sink Send. A Send in progress isn't canceled. func (c *lco) recvMetrics(stopSinksChan, doneChan chan struct{}) { defer func() { close(doneChan) @@ -162,6 +166,15 @@ func (c *lco) recvMetrics(stopSinksChan, doneChan chan struct{}) { c.event.Error(event.LCO_RECEIVER_PANIC, perr.Error()) } }() + stopped := func() bool { + select { + case <-stopSinksChan: + blip.Debug("%s: stopped, dropping metrics not sent", c.monitorId) + return true + default: + return false + } + } RECV: for { status.Monitor(c.monitorId, status.LEVEL_SINKS, "idle") @@ -169,6 +182,9 @@ RECV: case <-stopSinksChan: return case metrics := <-c.metricsChan: + if stopped() { // select picks randomly when both are ready + return + } if c.transformMetrics != nil { blip.Debug("%s: transform metrics", c.monitorId) status.Monitor(c.monitorId, status.LEVEL_SINKS, "TransformMetrics") @@ -181,6 +197,9 @@ RECV: for _, m := range metrics { coId := fmt.Sprintf("%s/%s/%d", m.Plan, m.Level, m.Interval) for _, sink := range c.sinks { + if stopped() { + return + } sinkName := sink.Name() status.Monitor(c.monitorId, status.LEVEL_SINKS, "%s", coId+": sending to "+sinkName) err := sink.Send(context.Background(), m) // @todo ctx with timeout @@ -223,8 +242,9 @@ func (c *lco) Run(stopChan, doneChan chan struct{}) error { // Keep receiving metrics for as long as the LCO is running stopSinksChan := make(chan struct{}) + stopSinks := sync.OnceFunc(func() { close(stopSinksChan) }) go c.keepRecvMetrics(stopSinksChan) - defer close(stopSinksChan) + defer stopSinks() // ----------------------------------------------------------------------- // LCO main loop: collect metrics every configured minimum level interval @@ -238,11 +258,13 @@ func (c *lco) Run(stopChan, doneChan chan struct{}) error { ticker := time.NewTicker(td) defer ticker.Stop() - // stop stops changePlan goroutine (if any) and prevents new ones in the + // stop stops sending metrics to sinks, which drops metrics not sent yet, + // stops changePlan goroutine (if any) and prevents new ones in the // pathological case that the LCH calls ChangePlan while the LCO is // terminating, then stops the engine stop := func() { blip.Debug("stopChan closed s=%d interval=%d", s, interval) + stopSinks() c.stopChangePlan() c.engine.Stop() // stop all collectors and run their cleanup func } @@ -335,7 +357,8 @@ func (c *lco) stopChangePlan() { // collect collects metrics at one level and sends them to the sinks. If ctx // is canceled (the LCO is stopping), the collection is canceled, and its -// metrics are dropped. +// metrics are dropped. Metrics that collect has queued for the sinks but +// recvMetrics hasn't sent yet are dropped too. func (c *lco) collect(ctx context.Context, interval uint, levelName string, startTime time.Time) { status.Monitor(c.monitorId, status.LEVEL_COLLECT, "%s/%s: collecting", c.plan.Name, levelName) dropped := false // stopped during collection diff --git a/monitor/level_collector_stop_test.go b/monitor/level_collector_stop_test.go index c700a68..c608bc3 100644 --- a/monitor/level_collector_stop_test.go +++ b/monitor/level_collector_stop_test.go @@ -5,6 +5,8 @@ package monitor import ( "context" "strings" + "sync" + "sync/atomic" "testing" "time" @@ -105,3 +107,44 @@ func TestLevelCollectorStopCancelsCollection(t *testing.T) { t.Errorf("%s status = %q, expected stopped; metrics dropped", status.LEVEL_COLLECT, s) } } + +func TestLevelCollectorStopDropsQueuedMetrics(t *testing.T) { + // Metrics from a complete collection that haven't been sent to the sinks + // when the LCO stops are dropped, not sent after the monitor stopped. + // Block TransformMetrics so that metrics are received and queued, then + // stop the LCO, then let TransformMetrics return. + const monitorId = "lco-stop-queued" + defer status.RemoveMonitor(monitorId) + TickerDuration(10*time.Millisecond, time.Second) + defer TickerDuration(time.Second, time.Second) + + entered := make(chan struct{}) + release := make(chan struct{}) + var enteredOnce, releaseOnce sync.Once + defer releaseOnce.Do(func() { close(release) }) + var sent atomic.Int32 + c := newStopTestLCO(monitorId, "every", mock.MetricsCollector{DomainFunc: func() string { return "test.lco-stop-queued" }}) + c.transformMetrics = func([]*blip.Metrics) error { + enteredOnce.Do(func() { close(entered) }) + <-release + return nil + } + c.sinks = []blip.Sink{mock.Sink{SendFunc: func(context.Context, *blip.Metrics) error { + sent.Add(1) + return nil + }}} + + runStopLCO(t, c, func() { + select { + case <-entered: + case <-time.After(5 * time.Second): + t.Fatal("timeout waiting for TransformMetrics") + } + time.Sleep(50 * time.Millisecond) // queue more metrics + }) + releaseOnce.Do(func() { close(release) }) + time.Sleep(100 * time.Millisecond) + if n := sent.Load(); n != 0 { + t.Errorf("sinks got %d sends after stop, expected 0", n) + } +} diff --git a/monitor/plan_changer.go b/monitor/plan_changer.go index a5ba11c..47d81e6 100644 --- a/monitor/plan_changer.go +++ b/monitor/plan_changer.go @@ -127,10 +127,11 @@ func (pch *planChanger) setErr(err error) { } // Run calls CheckState every second; or, if offline, it uses an exponential -// backoff up to 60 seconds until back online (MySQL connection ok). There is -// no logic in this function; it's just a timed loop to call CheckState. It's -// run as a goroutine from Monitor.Run only if config.monitors.plans.adjust -// is enabled (blip.ConfigPlanChange.Enabled returns true). +// backoff up to about 90 seconds (60 seconds plus jitter) until back online +// (MySQL connection ok). There is no logic in this function; it's just a timed +// loop to call CheckState. It's run as a goroutine from Monitor.Run only if +// config.monitors.plans.adjust is enabled (blip.ConfigPlanChange.Enabled +// returns true). func (pch *planChanger) Run(stopChan, doneChan chan struct{}) error { defer close(doneChan) defer status.Monitor(pch.monitorId, status.PLAN_CHANGER, "not running") diff --git a/monitor/plan_changer_stop_test.go b/monitor/plan_changer_stop_test.go index 0c5f8ad..d3a06e0 100644 --- a/monitor/plan_changer_stop_test.go +++ b/monitor/plan_changer_stop_test.go @@ -51,7 +51,7 @@ type offlineTestSQLDriver struct{} func (offlineTestSQLDriver) Open(string) (driver.Conn, error) { return nil, errors.New("offline") } func TestPlanChangerStopsDuringOfflineWait(t *testing.T) { - // While MySQL is offline, the PCH waits its retry backoff (up to 60s) + // While MySQL is offline, the PCH waits its retry backoff (up to about 90s: 60s plus jitter) // between state checks. Stopping it must not wait for that. const monitorId = "pch-stop" defer status.RemoveMonitor(monitorId) From 07b8ce6e3aae1c4b00d069cf43fb93fc139d99aa Mon Sep 17 00:00:00 2001 From: Ian Oberst Date: Tue, 27 Oct 2026 07:31:38 -0700 Subject: [PATCH 08/12] Clear the sinks status on drop and document more slow stops - When the level collector drops metrics on stop, it clears the level-sinks status instead of leaving "TransformMetrics" or "sending to"; it doesn't set one, which would add back a removed monitor. - Test the check right after recvMetrics receives metrics. - Docs: a restart after a subsystem fails drops metrics like a stop; a stop also waits for a plan change whose collector Prepare ignores cancel, and for the credentials function while starting. - Release note for CheckState(ctx) and for stop or restart. Co-Authored-By: Claude Opus 5.5 --- docs/content/about/release-notes.md | 3 +- docs/content/config/config-file.md | 7 ++-- monitor/level_collector.go | 3 ++ monitor/level_collector_stop_test.go | 51 ++++++++++++++++++++++++++++ monitor/monitor.go | 5 +-- 5 files changed, 63 insertions(+), 6 deletions(-) diff --git a/docs/content/about/release-notes.md b/docs/content/about/release-notes.md index dbb24dd..8f8a12d 100644 --- a/docs/content/about/release-notes.md +++ b/docs/content/about/release-notes.md @@ -47,7 +47,8 @@ Default sink HTTP clients now have a 10-second whole-request timeout plus bounde * Changed the loader to send the `monitor-stopped` and `monitor-stop-timeout` events from its stop goroutine, so an event receiver that panics on either event crashes Blip. Previously, when the API stopped a monitor, `net/http` recovered the panic. Event receivers must not panic. * **Behavior change:** `monitor.Loader.Stop` now keeps the monitor stopped across `Load` and `StartMonitors` until `Start`, unless its config changes or it's removed and loaded again. Go callers that relied on the next reload to restart a stopped monitor must call `Start`. `POST /monitors/stop` keeps a monitor stopped across reloads, as before. * **Behavior change:** `POST /monitors/start` now calls the `StartMonitor` plugin, so it returns 409 for a monitor that the plugin declines. Previously the API started the monitor directly and skipped the plugin, so it could start a monitor that the plugin filtered out. Integrations that start filtered monitors through the API must allow them in the plugin. -* A monitor stop no longer waits for the next collection tick, plan change check, or heartbeat. It cancels a collection in progress and drops its metrics and any metrics not yet sent to sinks, and it cancels the plan changer's read-only check and a heartbeat write in progress. +* A monitor stop, or a restart after one of its subsystems fails, no longer waits for the next collection tick, plan change check, or heartbeat. It cancels a collection in progress and drops its metrics and any metrics not yet sent to sinks, and it cancels the plan changer's read-only check and a heartbeat write in progress. +* Changed `CheckState` of the plan changer returned by `monitor.NewPlanChanger` to `CheckState(ctx)`. Canceling `ctx` cancels the read-only check. * **Behavior change:** Applied `timeout-connect` (default `10s`) to MySQL connections. Previously it was ignored, so connecting to an unreachable host waited for the operating system TCP connect timeout. Set it to `0` to retain the old behavior. An invalid `timeout-connect` value, which used to be ignored, now fails config validation, so Blip doesn't start, or a monitor in a reload doesn't load. For example, a bare number such as `timeout-connect: 5` has no unit and is invalid; use `5s`. * Added per-monitor fault counters, such as collector panics and timeouts, engine max runtime (EMR) timeouts, and dropped metrics, reported by the new [`GET /status/monitors/counters`]({{< ref "/api/status#get-statusmonitorscounters" >}}) API endpoint. Integrations can read them with the new `status.ReportCounters`, and register their own with `status.Counter`; the counter names are `status` constants like `status.COLLECTOR_PANICS`. * Added the optional `sink.CounterSink` interface for sinks that report monitor status counters; the monitor calls `RegisterCounters` each time it starts. A custom sink that wraps `sink.Retry` must forward `RegisterCounters` to it, else the retry buffer's `sink-retry-dropped` counter isn't reported. diff --git a/docs/content/config/config-file.md b/docs/content/config/config-file.md index 45903d1..d2816f4 100644 --- a/docs/content/config/config-file.md +++ b/docs/content/config/config-file.md @@ -196,14 +196,15 @@ If the monitor is reloaded with the same ID, the new monitor doesn't start until A stop doesn't wait for the next collection, plan change check, or heartbeat: -* The level collector stops right away. It cancels a collection in progress and drops its metrics, and it drops metrics that haven't been sent to the sinks yet (a send in progress isn't canceled). A canceled collection isn't counted as an engine max runtime timeout. +* The level collector stops right away. It cancels a collection in progress and drops its metrics, and it drops metrics that haven't been sent to the sinks yet (a send in progress isn't canceled). A canceled collection isn't counted as an engine max runtime timeout. The same happens when a monitor restarts because one of its subsystems failed. * The plan changer and the heartbeat writer stop right away while they wait. The plan changer waits between checks, and backs off while MySQL is offline (up to about 90 seconds: 60 seconds plus jitter). The heartbeat writer waits between heartbeats, before it retries a first heartbeat that failed, and while MySQL is read-only. A stop cancels the plan changer's read-only check and a heartbeat write in progress, so it doesn't wait for a hung or unreachable MySQL. A running monitor usually stops in milliseconds, even when MySQL is hung or unreachable. -A stop can take longer in two cases: +A stop can take longer in these cases: * A collector, or its cleanup function, doesn't return promptly when its collection is canceled. A stop waits for every collector to return, then runs their cleanup functions. -* The monitor is starting. A stop waits for it to load its plans. If they're in a table ([`plans.table`](#table-1)), that's the only time a starting monitor queries MySQL, and it can take up to 5 seconds (connecting included) when MySQL is unreachable or hung. A custom `DbProviderFactory` that blocks while making the monitor's connections also delays the stop. +* A plan change is in progress, and a collector's `Prepare` doesn't return promptly when it's canceled. A stop cancels the plan change and waits for it. +* The monitor is starting. A stop waits for it to load its plans. If they're in a table ([`plans.table`](#table-1)), that's the only time a starting monitor queries MySQL, and it can take up to 5 seconds (connecting included) when MySQL is unreachable or hung. Making the monitor's connections also calls its credentials function (for example, to get an IAM auth token), which can take up to 500 milliseconds each time, and a custom `DbProviderFactory` that blocks while making them delays the stop too. --- diff --git a/monitor/level_collector.go b/monitor/level_collector.go index 6275dc6..e86005c 100644 --- a/monitor/level_collector.go +++ b/monitor/level_collector.go @@ -170,6 +170,9 @@ func (c *lco) recvMetrics(stopSinksChan, doneChan chan struct{}) { select { case <-stopSinksChan: blip.Debug("%s: stopped, dropping metrics not sent", c.monitorId) + // Clear "TransformMetrics" or "sending to"; don't set a status, + // which would add back the status of a monitor that was removed + status.RemoveComponent(c.monitorId, status.LEVEL_SINKS) return true default: return false diff --git a/monitor/level_collector_stop_test.go b/monitor/level_collector_stop_test.go index c608bc3..3425ace 100644 --- a/monitor/level_collector_stop_test.go +++ b/monitor/level_collector_stop_test.go @@ -147,4 +147,55 @@ func TestLevelCollectorStopDropsQueuedMetrics(t *testing.T) { if n := sent.Load(); n != 0 { t.Errorf("sinks got %d sends after stop, expected 0", n) } + // The sinks status isn't left at "TransformMetrics" + if s, ok := status.ReportMonitors(monitorId)[monitorId][status.LEVEL_SINKS]; ok { + t.Errorf("%s status = %q, expected none after the metrics were dropped", status.LEVEL_SINKS, s) + } +} + +func TestLevelCollectorStopDropsMetricsAfterSend(t *testing.T) { + // After a sink Send returns, recvMetrics can receive metrics that were + // queued during the Send even though the LCO has stopped, because select + // picks randomly when both are ready. It must drop them before it calls + // TransformMetrics. Block the first Send so that metrics queue, stop the + // LCO, then let the Send return. One run catches a missing check half the + // time, so run several. + const monitorId = "lco-stop-after-send" + defer status.RemoveMonitor(monitorId) + TickerDuration(10*time.Millisecond, time.Second) + defer TickerDuration(time.Second, time.Second) + + for i := 0; i < 20; i++ { + entered := make(chan struct{}) + release := make(chan struct{}) + var enteredOnce, releaseOnce sync.Once + var transforms atomic.Int32 + c := newStopTestLCO(monitorId, "every", mock.MetricsCollector{DomainFunc: func() string { return "test.lco-stop-after-send" }}) + c.transformMetrics = func([]*blip.Metrics) error { + transforms.Add(1) + return nil + } + c.sinks = []blip.Sink{mock.Sink{SendFunc: func(context.Context, *blip.Metrics) error { + enteredOnce.Do(func() { + close(entered) + <-release + }) + return nil + }}} + + runStopLCO(t, c, func() { + select { + case <-entered: + case <-time.After(5 * time.Second): + t.Fatal("timeout waiting for sink Send") + } + time.Sleep(30 * time.Millisecond) // queue more metrics + }) + before := transforms.Load() + releaseOnce.Do(func() { close(release) }) + time.Sleep(30 * time.Millisecond) + if n := transforms.Load() - before; n != 0 { + t.Fatalf("run %d: TransformMetrics called %d times after stop, expected 0", i, n) + } + } } diff --git a/monitor/monitor.go b/monitor/monitor.go index 9011841..eb7ef50 100644 --- a/monitor/monitor.go +++ b/monitor/monitor.go @@ -246,8 +246,9 @@ func (m *Monitor) runLoop(stopChan, doneChan chan struct{}) { // Monitor is running. Wait for either Stop (which closes m.runLoopChan) // or one of the monitor subsystems to return/panic (which closes m.runChan). // On Stop, return immediately: user is stopping the monitor completely. - // On m.runChan close (via stop func), we restart almost immediately because - // Blip never stops trying to send metrics. + // On m.runChan close (via stop func), we restart almost immediately to + // resume collecting metrics. Like a stop, the restart drops the collection + // in progress and metrics not yet sent to sinks (see lco.Run). m.retry.Reset() status.Monitor(m.monitorId, status.MONITOR, "running since %s", blip.FormatTime(time.Now())) select { From 47c8ba36294be3380c4486c5be35a0c2e13c5962 Mon Sep 17 00:00:00 2001 From: Ian Oberst Date: Tue, 27 Oct 2026 09:46:38 -0700 Subject: [PATCH 09/12] Simplify the level collector, plan changer, and heartbeat stops - The LCO uses one stop context, from stopchan.Context, for its collections and its sink receiver instead of a separate stopSinksChan and collect context. Run cancels it first when it stops, so metrics not yet sent are still dropped before the engine stops. - recvMetrics checks ctx instead of a stopped closure, and its defer clears the level-sinks status when ctx is canceled. keepRecvMetrics restarts it after a sink panic until ctx is canceled. - Run checks for stop in one place, after waiting for the tick or stop. tickHook takes no argument; StopAfterTicks counts the ticks. - collect derives "stopped; metrics dropped" from ctx instead of a flag. - The plan changer drops its test-only readOnlyTimeout (the read-only query keeps its 10s timeout) and the stop check at the top of Run, which the sleep at the bottom covers. - The engine counts an EMR timeout only if emrCtx's deadline passed. - The heartbeat writer drops its stop checks after a write: a canceled write sets an error status for a moment, then the sleep that follows returns at once, and Write sets "stopped". - stopchan.Sleep uses time.After (Go 1.23+ collects unfired timers). - Tests: remove TestLoaderStopDoesNotWaitForTick, which TestLevelCollectorStopDoesNotWaitForTick covers. The PCH stop tests share stopPCH, the writer's hung-heartbeat case uses stopWriter with a fake driver that hangs from the Nth write, and TestAPIMonitorsStart uses a 200ms stop timeout now that a stop doesn't wait for the tick. Co-Authored-By: Claude Opus 5.5 --- heartbeat/writer.go | 6 -- heartbeat/writer_stop_test.go | 51 +++--------- internal/stopchan/stopchan.go | 4 +- monitor/engine.go | 5 +- monitor/export_test.go | 12 +-- monitor/fault_counters_test.go | 7 +- monitor/level_collector.go | 126 +++++++++++------------------- monitor/loader_stop_test.go | 29 ------- monitor/plan_changer.go | 10 +-- monitor/plan_changer_stop_test.go | 82 ++++++++----------- server/api_test.go | 3 +- 11 files changed, 107 insertions(+), 228 deletions(-) diff --git a/heartbeat/writer.go b/heartbeat/writer.go index ae976cc..9b06259 100644 --- a/heartbeat/writer.go +++ b/heartbeat/writer.go @@ -104,9 +104,6 @@ func (w *Writer) Write(stopChan, doneChan chan struct{}) error { ctx, cancel = context.WithTimeout(stopCtx, WriteTimeout) _, err = w.db.ExecContext(ctx, ping, params...) cancel() - if stopCtx.Err() != nil { - return nil // stopped - } if err == nil { // success status.Monitor(w.monitorId, status.HEARTBEAT_WRITER, "sleep") break @@ -144,9 +141,6 @@ func (w *Writer) Write(stopChan, doneChan chan struct{}) error { ctx, cancel = context.WithTimeout(stopCtx, WriteTimeout) _, err = w.db.ExecContext(ctx, ping, w.srcId) cancel() - if stopCtx.Err() != nil { - return nil // stopped - } if err != nil { blip.Debug("%s: %s", w.monitorId, err.Error()) if sqlutil.ReadOnly(err) { diff --git a/heartbeat/writer_stop_test.go b/heartbeat/writer_stop_test.go index 2cecfb1..7848cfa 100644 --- a/heartbeat/writer_stop_test.go +++ b/heartbeat/writer_stop_test.go @@ -21,13 +21,14 @@ import ( // writeTestDriver is a fake SQL driver for the heartbeat writer: every write // succeeds, or fails with a MySQL read-only error if writeTestReadOnly is set, -// or blocks until its context is canceled if writeTestHung is set. +// or blocks until its context is canceled from write number writeTestHangAt on +// (counted by writeTestWrites), if it's set. const writeTestDriver = "blip-heartbeat-write-test" var ( writeTestReadOnly atomic.Bool - writeTestHung atomic.Bool - writeTestHungExec = make(chan struct{}, 10) // a hung write has started + writeTestHangAt atomic.Int64 + writeTestWrites atomic.Int64 ) func init() { @@ -46,8 +47,8 @@ func (writeTestConn) Prepare(string) (driver.Stmt, error) { func (writeTestConn) Close() error { return nil } func (writeTestConn) Begin() (driver.Tx, error) { return nil, errors.New("no transactions") } func (writeTestConn) ExecContext(ctx context.Context, _ string, _ []driver.NamedValue) (driver.Result, error) { - if writeTestHung.Load() { - writeTestHungExec <- struct{}{} + n := writeTestWrites.Add(1) + if at := writeTestHangAt.Load(); at > 0 && n >= at { <-ctx.Done() return nil, ctx.Err() } @@ -130,48 +131,22 @@ func TestWriterStopsDuringWrite(t *testing.T) { defer func(d time.Duration) { heartbeat.WriteTimeout = d }(heartbeat.WriteTimeout) heartbeat.WriteTimeout = timeout - writeTestHung.Store(true) - defer writeTestHung.Store(false) + defer writeTestHangAt.Store(0) // First write (INSERT) hangs. The status is set before the write, so the // stop can come just before the write starts, which tests stopping it too. + writeTestWrites.Store(0) + writeTestHangAt.Store(1) w := heartbeat.NewWriter(monitorId, db, blip.ConfigHeartbeat{Freq: "1s", Table: "heartbeat"}) if d := stopWriter(t, w, monitorId, "first insert"); d > timeout/10 { t.Errorf("stop during first write took %s, expected well under %s", d, timeout) } - for len(writeTestHungExec) > 0 { - <-writeTestHungExec - } - // A heartbeat (UPDATE) hangs - writeTestHung.Store(false) + // The first write succeeds, and the first heartbeat (UPDATE) hangs + writeTestWrites.Store(0) + writeTestHangAt.Store(2) w = heartbeat.NewWriter(monitorId, db, blip.ConfigHeartbeat{Freq: "10ms", Table: "heartbeat"}) - stopChan := make(chan struct{}) - doneChan := make(chan struct{}) - go w.Write(stopChan, doneChan) - deadline := time.Now().Add(5 * time.Second) - for status.ReportMonitors(monitorId)[monitorId][status.HEARTBEAT_WRITER] != "sleep" { - if time.Now().After(deadline) { - close(stopChan) - t.Fatal("timeout waiting for first write") - } - time.Sleep(time.Millisecond) - } - writeTestHung.Store(true) - select { - case <-writeTestHungExec: - case <-time.After(5 * time.Second): - close(stopChan) - t.Fatal("timeout waiting for hung heartbeat") - } - t0 := time.Now() - close(stopChan) - select { - case <-doneChan: - case <-time.After(2 * timeout): - t.Fatal("heartbeat writer did not stop") - } - if d := time.Since(t0); d > timeout/10 { + if d := stopWriter(t, w, monitorId, "write"); d > timeout/10 { t.Errorf("stop during heartbeat write took %s, expected well under %s", d, timeout) } } diff --git a/internal/stopchan/stopchan.go b/internal/stopchan/stopchan.go index dd57897..c028bfa 100644 --- a/internal/stopchan/stopchan.go +++ b/internal/stopchan/stopchan.go @@ -13,12 +13,10 @@ import ( // Sleep waits for d or until stopChan is closed. It returns false if stopChan // was closed. To sleep until a context is done, pass ctx.Done() as stopChan. func Sleep(stopChan <-chan struct{}, d time.Duration) bool { - t := time.NewTimer(d) - defer t.Stop() select { case <-stopChan: return false - case <-t.C: + case <-time.After(d): // an unfired timer is collected (Go 1.23+) return true } } diff --git a/monitor/engine.go b/monitor/engine.go index e1fe071..2b0c309 100644 --- a/monitor/engine.go +++ b/monitor/engine.go @@ -445,10 +445,7 @@ SWEEP: // are missing from this interval. Not an error event because, when the // host is CPU-starved, this can happen every interval for every monitor. notReturned := len(running) - if emrTimeout && emrCtx.Err() == context.Canceled { - emrTimeout = false // canceled (stopping), not timed out - } - if emrTimeout { + if emrTimeout && emrCtx.Err() == context.DeadlineExceeded { // not canceled (stopping) e.counters.emrTimeouts.Add(1) e.counters.domainsNotStarted.Add(uint64(notStarted)) e.counters.domainsNotReturned.Add(uint64(notReturned)) diff --git a/monitor/export_test.go b/monitor/export_test.go index 480ed8a..0424536 100644 --- a/monitor/export_test.go +++ b/monitor/export_test.go @@ -2,18 +2,18 @@ package monitor -import "sync" +import "sync/atomic" // StopAfterTicks makes every LCO Run return at tick n+1, before it collects at // that tick, by closing stopChan. Use it for a test that runs one LCO, and call -// the returned func to remove it. If more than one LCO reaches tick n+1, +// the returned func to remove it. Ticks are counted across all LCOs, and // stopChan is closed only once. func StopAfterTicks(n uint, stopChan chan struct{}) func() { - var once sync.Once + var ticks atomic.Uint64 tickerMux.Lock() - tickHook = func(tick uint) { - if tick == n+1 { - once.Do(func() { close(stopChan) }) + tickHook = func() { + if ticks.Add(1) == uint64(n)+1 { + close(stopChan) } } tickerMux.Unlock() diff --git a/monitor/fault_counters_test.go b/monitor/fault_counters_test.go index e471575..0416c49 100644 --- a/monitor/fault_counters_test.go +++ b/monitor/fault_counters_test.go @@ -437,9 +437,10 @@ func TestLevelCollectorCountsTransformDrops(t *testing.T) { Config: blip.ConfigMonitor{MonitorId: monitorId}, TransformMetrics: func([]*blip.Metrics) error { return errors.New("transform failed") }, }, nil) - stopSinksChan := make(chan struct{}) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() doneChan := make(chan struct{}) - go c.recvMetrics(stopSinksChan, doneChan) + go c.recvMetrics(ctx, doneChan) c.metricsChan <- []*blip.Metrics{{MonitorId: monitorId}} c.metricsChan <- []*blip.Metrics{{MonitorId: monitorId}} @@ -447,7 +448,7 @@ func TestLevelCollectorCountsTransformDrops(t *testing.T) { for counters(t, monitorId)[status.LEVEL_TRANSFORM_DROPPED] < 2 && time.Now().Before(deadline) { time.Sleep(5 * time.Millisecond) } - close(stopSinksChan) + cancel() <-doneChan if v := counters(t, monitorId)[status.LEVEL_TRANSFORM_DROPPED]; v != 2 { t.Errorf("%s = %d, expected 2", status.LEVEL_TRANSFORM_DROPPED, v) diff --git a/monitor/level_collector.go b/monitor/level_collector.go index e86005c..7220570 100644 --- a/monitor/level_collector.go +++ b/monitor/level_collector.go @@ -142,21 +142,20 @@ var ( tickerDuration = 1 * time.Second // used for testing timeElapsed = 1 * time.Second // used for testing - // tickHook, if set, is called by Run after each tick with the tick number - // (1 for the first tick). Tests use it to stop Run after a number of ticks - // by closing stopChan, which Run sees before it collects at that tick. - tickHook func(tick uint) // used for testing + // tickHook, if set, is called by Run after each tick. Tests use it to stop + // Run by closing stopChan, which Run sees before it collects at that tick. + tickHook func() // used for testing ) // recvMetrics receives metrics on metricsChan and send them to all the sinks. // This is a goroutine run by keepRecvMetrics and restarted by keepRecvMetrics -// if a sink (or the transformMetrics) panics. It stops when stopSinksChan is -// closed in Run, and it closes doneChan when stopped. +// if a sink (or the transformMetrics) panics. It stops when ctx is canceled +// in Run, and it closes doneChan when stopped. // -// When stopSinksChan is closed, metrics not yet sent are dropped, so a stopped -// or removed monitor doesn't send metrics. recvMetrics checks after it receives +// When ctx is canceled, metrics not yet sent are dropped, so a stopped or +// removed monitor doesn't send metrics. recvMetrics checks after it receives // metrics and before each sink Send. A Send in progress isn't canceled. -func (c *lco) recvMetrics(stopSinksChan, doneChan chan struct{}) { +func (c *lco) recvMetrics(ctx context.Context, doneChan chan struct{}) { defer func() { close(doneChan) if r := recover(); r != nil { @@ -165,27 +164,20 @@ func (c *lco) recvMetrics(stopSinksChan, doneChan chan struct{}) { perr := fmt.Errorf("PANIC: sinks: %s: %v\n%s", c.monitorId, r, string(b[0:n])) c.event.Error(event.LCO_RECEIVER_PANIC, perr.Error()) } - }() - stopped := func() bool { - select { - case <-stopSinksChan: - blip.Debug("%s: stopped, dropping metrics not sent", c.monitorId) + if ctx.Err() != nil { // Clear "TransformMetrics" or "sending to"; don't set a status, // which would add back the status of a monitor that was removed status.RemoveComponent(c.monitorId, status.LEVEL_SINKS) - return true - default: - return false } - } + }() RECV: for { status.Monitor(c.monitorId, status.LEVEL_SINKS, "idle") select { - case <-stopSinksChan: + case <-ctx.Done(): return case metrics := <-c.metricsChan: - if stopped() { // select picks randomly when both are ready + if ctx.Err() != nil { // select picks randomly when both are ready return } if c.transformMetrics != nil { @@ -200,7 +192,7 @@ RECV: for _, m := range metrics { coId := fmt.Sprintf("%s/%s/%d", m.Plan, m.Level, m.Interval) for _, sink := range c.sinks { - if stopped() { + if ctx.Err() != nil { return } sinkName := sink.Name() @@ -220,34 +212,24 @@ RECV: // keepRecvMetrics keeps a recvMetrics goroutine running. If a sink or the // transformMetrics plugin panic, it must be restarted to keep metrics flowing. -func (c *lco) keepRecvMetrics(stopSinksChan chan struct{}) { - for { +func (c *lco) keepRecvMetrics(ctx context.Context) { + for ctx.Err() == nil { doneChan := make(chan struct{}) - go c.recvMetrics(stopSinksChan, doneChan) - select { - case <-stopSinksChan: - return - case <-doneChan: - // recvMetrics goroutine stopped but why? Did it return because - // stopSinksChan, or because of a panic? - select { - case <-stopSinksChan: - return // stopSinksChan closed - default: - // Probably a sink panic - } - } + go c.recvMetrics(ctx, doneChan) + <-doneChan // stopped, or restart after a sink panic } } func (c *lco) Run(stopChan, doneChan chan struct{}) error { defer close(doneChan) + // ctx is canceled on stop, so a stop doesn't wait for a collection in + // progress, and metrics not sent to sinks yet are dropped + ctx, cancel := stopchan.Context(stopChan) + defer cancel() + // Keep receiving metrics for as long as the LCO is running - stopSinksChan := make(chan struct{}) - stopSinks := sync.OnceFunc(func() { close(stopSinksChan) }) - go c.keepRecvMetrics(stopSinksChan) - defer stopSinks() + go c.keepRecvMetrics(ctx) // ----------------------------------------------------------------------- // LCO main loop: collect metrics every configured minimum level interval @@ -261,39 +243,33 @@ func (c *lco) Run(stopChan, doneChan chan struct{}) error { ticker := time.NewTicker(td) defer ticker.Stop() - // stop stops sending metrics to sinks, which drops metrics not sent yet, - // stops changePlan goroutine (if any) and prevents new ones in the - // pathological case that the LCH calls ChangePlan while the LCO is - // terminating, then stops the engine - stop := func() { - blip.Debug("stopChan closed s=%d interval=%d", s, interval) - stopSinks() - c.stopChangePlan() - c.engine.Stop() // stop all collectors and run their cleanup func - } - - // collectCtx is canceled when stopChan is closed, so a stop doesn't wait for - // a collection in progress - collectCtx, cancelCollect := stopchan.Context(stopChan) - defer cancelCollect() - var lastTick time.Time - tick := uint(0) for { - // Wait for the next tick or stop, whichever is first, so stopping - // doesn't wait up to one tick. Stop closes stopChan. + // Wait for the next tick or stop, so stopping doesn't wait for the tick var startTime time.Time select { case <-stopChan: - stop() - return nil case startTime = <-ticker.C: + if onTick != nil { + onTick() + } } - s = s + te - tick++ - if onTick != nil { - onTick(tick) + + // Was monitor stopped? Check after a tick, too, because select picks + // randomly when both the tick and stop are ready. + select { + case <-stopChan: // yes, return immediately + blip.Debug("stopChan closed s=%d interval=%d", s, interval) + cancel() // drop metrics not sent to sinks yet + // Stop changePlan goroutine (if any) and prevent new ones in the + // pathological case that the LCH calls ChangePlan while the LCO + // is terminating + c.stopChangePlan() + c.engine.Stop() // stop all collectors and run their cleanup func + return nil + default: // no } + s = s + te // Count ticks dropped while the previous iteration was blocked, usually // by a slow collection. Run doesn't catch up on dropped ticks, so each one @@ -303,15 +279,6 @@ func (c *lco) Run(stopChan, doneChan chan struct{}) error { } lastTick = startTime - // Was monitor stopped? Check again because select picks randomly - // when both the tick and stop are ready. - select { - case <-stopChan: // yes, return immediately - stop() - return nil - default: // no - } - // Paused because no plan is set or it's being changed? c.stateMux.Lock() // -- LOCK -- if c.paused { @@ -335,7 +302,7 @@ func (c *lco) Run(stopChan, doneChan chan struct{}) error { // Collect metrics at this level interval += 1 - c.collect(collectCtx, interval, c.levels[level].Name, startTime) + c.collect(ctx, interval, c.levels[level].Name, startTime) c.stateMux.Unlock() // -- UNLOCK -- } @@ -364,14 +331,13 @@ func (c *lco) stopChangePlan() { // recvMetrics hasn't sent yet are dropped too. func (c *lco) collect(ctx context.Context, interval uint, levelName string, startTime time.Time) { status.Monitor(c.monitorId, status.LEVEL_COLLECT, "%s/%s: collecting", c.plan.Name, levelName) - dropped := false // stopped during collection defer func() { if err := recover(); err != nil { // catch panic in engine and TransformMetrics b := make([]byte, 4096) n := runtime.Stack(b, false) c.event.Errorf(event.LCO_COLLECT_PANIC, "PANIC: %s: %s\n%s", c.monitorId, err, string(b[0:n])) } - if dropped { + if ctx.Err() != nil { status.Monitor(c.monitorId, status.LEVEL_COLLECT, "%s/%s/%d: stopped; metrics dropped", c.plan.Name, levelName, interval) return } @@ -388,9 +354,7 @@ func (c *lco) collect(ctx context.Context, interval uint, levelName string, star metrics, err := c.engine.Collect(emrCtx, interval, levelName, startTime) blip.Debug("%s: level %s: done in %s", c.monitorId, levelName, metrics[0].End.Sub(metrics[0].Begin)) if ctx.Err() != nil { - blip.Debug("%s: level %s: stopped, dropping metrics", c.monitorId, levelName) - dropped = true - return + return // stopped: drop the metrics } if err != nil { diff --git a/monitor/loader_stop_test.go b/monitor/loader_stop_test.go index 266e1f0..02074ca 100644 --- a/monitor/loader_stop_test.go +++ b/monitor/loader_stop_test.go @@ -1062,32 +1062,3 @@ func TestLoaderStartClearsHeld(t *testing.T) { setIds(st, &ids) require.NoError(t, loader.Load(context.Background())) } - -func TestLoaderStopDoesNotWaitForTick(t *testing.T) { - // A stop must not wait for the next LCO tick. Before the LCO selected on - // stop, a stop took up to one tick, depending on when in the tick it - // arrived. Use a long tick so the bound is relative to it and not to how - // fast a stop is under -race. TestLevelCollectorStop*, - // TestPlanChangerStopsDuringOfflineWait, and TestWriterStopsDuringWaits - // cover the other waits directly. - const tick = 4 * time.Second - ids := []string{"tick1"} - st, loader := setupStopTest(t, "10s", &ids, nil) // stop timeout > tick - defer status.RemoveMonitor("tick1") - monitor.TickerDuration(tick, time.Second) // setupStopTest restores the test tick - - require.NoError(t, loader.Load(context.Background())) - loader.StartMonitors() - waitPrepared(t, st, 1, "tick1") - - for i, offset := range []time.Duration{tick / 20, tick / 2} { // vary the offset into the tick - time.Sleep(offset) - t0 := time.Now() - require.NoError(t, loader.Stop("tick1", true)) - d := time.Since(t0) - t.Logf("stop %d: %s", i, d) - assert.Less(t, d, tick/4, "stop %d waited for the LCO tick", i) - require.NoError(t, loader.Start("tick1", true)) - waitPrepared(t, st, i+2, "tick1") - } -} diff --git a/monitor/plan_changer.go b/monitor/plan_changer.go index 47d81e6..0644b5b 100644 --- a/monitor/plan_changer.go +++ b/monitor/plan_changer.go @@ -67,8 +67,6 @@ type planChanger struct { event event.MonitorReceiver retry *backoff.ExponentialBackOff lerr error - - readOnlyTimeout time.Duration // how long to wait for readOnlyQuery } func NewPlanChanger(args PlanChangerArgs) *planChanger { @@ -112,8 +110,6 @@ func NewPlanChanger(args PlanChangerArgs) *planChanger { first: true, event: event.MonitorReceiver{MonitorId: args.MonitorId}, retry: retry, - - readOnlyTimeout: 10 * time.Second, } } @@ -144,10 +140,6 @@ func (pch *planChanger) Run(stopChan, doneChan chan struct{}) error { defer cancel() for { - if ctx.Err() != nil { - return nil // stopped - } - pch.CheckState(ctx) wait := 1 * time.Second @@ -264,7 +256,7 @@ func (pch *planChanger) state(ctx context.Context) string { status.Monitor(pch.monitorId, status.PLAN_CHANGER, "checking MySQL read-only") var ro, sro int - qctx, cancel := context.WithTimeout(ctx, pch.readOnlyTimeout) + qctx, cancel := context.WithTimeout(ctx, 10*time.Second) err := pch.db.QueryRowContext(qctx, readOnlyQuery).Scan(&ro, &sro) cancel() if ctx.Err() != nil { diff --git a/monitor/plan_changer_stop_test.go b/monitor/plan_changer_stop_test.go index d3a06e0..778fde6 100644 --- a/monitor/plan_changer_stop_test.go +++ b/monitor/plan_changer_stop_test.go @@ -50,82 +50,68 @@ type offlineTestSQLDriver struct{} func (offlineTestSQLDriver) Open(string) (driver.Conn, error) { return nil, errors.New("offline") } -func TestPlanChangerStopsDuringOfflineWait(t *testing.T) { - // While MySQL is offline, the PCH waits its retry backoff (up to about 90s: 60s plus jitter) - // between state checks. Stopping it must not wait for that. - const monitorId = "pch-stop" - defer status.RemoveMonitor(monitorId) - db, err := sql.Open(offlineTestDriver, "") +// stopPCH runs a PCH on a database opened with driverName, waits until ready +// returns true, then stops the PCH and fails the test if the stop takes 1s or +// longer. While MySQL is offline, the PCH waits 1 minute between checks. +func stopPCH(t *testing.T, monitorId, driverName string, ready func() bool) { + t.Helper() + db, err := sql.Open(driverName, "") if err != nil { t.Fatal(err) } defer db.Close() - - const wait = time.Minute pch := NewPlanChanger(PlanChangerArgs{MonitorId: monitorId, DB: db, HA: ha.Disabled}) - pch.retry.InitialInterval = wait - pch.retry.MaxInterval = wait + pch.retry.InitialInterval = time.Minute + pch.retry.MaxInterval = time.Minute pch.retry.Reset() stopChan := make(chan struct{}) doneChan := make(chan struct{}) go pch.Run(stopChan, doneChan) - - // Wait for the first state check (offline), then stop during the wait deadline := time.Now().Add(5 * time.Second) - for status.ReportMonitors(monitorId)[monitorId]["error:"+status.PLAN_CHANGER] == "" { + for !ready() { if time.Now().After(deadline) { close(stopChan) - t.Fatal("timeout waiting for PCH state check") + t.Fatal("timeout waiting for the PCH") } time.Sleep(5 * time.Millisecond) } + t0 := time.Now() close(stopChan) select { case <-doneChan: - case <-time.After(2 * wait): + case <-time.After(time.Minute): t.Fatal("PCH did not stop") } - if d := time.Since(t0); d > wait/10 { - t.Errorf("stop during offline wait took %s, expected well under %s", d, wait) + if d := time.Since(t0); d >= time.Second { + t.Errorf("stop took %s, expected less than 1s", d) } } +func TestPlanChangerStopsDuringOfflineWait(t *testing.T) { + // While MySQL is offline, the PCH waits its retry backoff (up to about 90s: + // 60s plus jitter) between state checks. Stopping it must not wait for that. + const monitorId = "pch-stop" + defer status.RemoveMonitor(monitorId) + stopPCH(t, monitorId, offlineTestDriver, func() bool { + return status.ReportMonitors(monitorId)[monitorId]["error:"+status.PLAN_CHANGER] != "" + }) +} + func TestPlanChangerStopsDuringStateQuery(t *testing.T) { - // The read-only query can wait up to the read-only timeout on a hung or - // partitioned MySQL. Stopping the PCH cancels it instead of waiting. + // The read-only query can wait up to 10s on a hung or partitioned MySQL. + // Stopping the PCH cancels it instead of waiting. const monitorId = "pch-stop-query" defer status.RemoveMonitor(monitorId) - db, err := sql.Open(hungTestDriver, "") - if err != nil { - t.Fatal(err) - } - defer db.Close() - - const timeout = time.Minute - pch := NewPlanChanger(PlanChangerArgs{MonitorId: monitorId, DB: db, HA: ha.Disabled}) - pch.readOnlyTimeout = timeout - stopChan := make(chan struct{}) - doneChan := make(chan struct{}) - go pch.Run(stopChan, doneChan) - select { - case <-hungTestQueries: - case <-time.After(5 * time.Second): - close(stopChan) - t.Fatal("timeout waiting for read-only query") - } - - t0 := time.Now() - close(stopChan) - select { - case <-doneChan: - case <-time.After(2 * timeout): - t.Fatal("PCH did not stop") - } - if d := time.Since(t0); d > timeout/10 { - t.Errorf("stop during read-only query took %s, expected well under %s", d, timeout) - } + stopPCH(t, monitorId, hungTestDriver, func() bool { + select { + case <-hungTestQueries: + return true + default: + return false + } + }) // A canceled query isn't an error: the PCH didn't see MySQL offline if e := status.ReportMonitors(monitorId)[monitorId]["error:"+status.PLAN_CHANGER]; e != "" { t.Errorf("PCH error status %q after stop, expected none", e) diff --git a/server/api_test.go b/server/api_test.go index fafab56..81c5f88 100644 --- a/server/api_test.go +++ b/server/api_test.go @@ -123,6 +123,7 @@ func TestAPIMonitorsStart(t *testing.T) { // running or the StartMonitor plugin declines to start it. A monitor // stopped by POST /monitors/stop stays stopped across a reload. cfg := blip.DefaultConfig() + cfg.MonitorLoader.StopTimeout = "200ms" dbf := blockingDbFactory{ DbFactory: dbconn.NewConnFactory(nil, nil), monitorId: "api-pending", @@ -231,7 +232,7 @@ func TestAPIMonitorsStart(t *testing.T) { } // Start api-pending and stop it while it's stuck in startup (making its - // DB), so the stop times out after the default stop timeout (3s). Starting + // DB), so the stop times out after the stop timeout (200ms). Starting // it while it's still stopping is pending, and starting it again is still // pending, not already running. if got := post("/monitors/start?id=api-pending"); got != http.StatusOK { From d906f486718f2f979a7e0b205d68e44439529aae Mon Sep 17 00:00:00 2001 From: Ian Oberst Date: Tue, 27 Oct 2026 09:46:41 -0700 Subject: [PATCH 10/12] Shorten the slow-stop docs and release note Co-Authored-By: Claude Opus 5.5 --- docs/content/about/release-notes.md | 2 +- docs/content/config/config-file.md | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/docs/content/about/release-notes.md b/docs/content/about/release-notes.md index 8f8a12d..086c8b1 100644 --- a/docs/content/about/release-notes.md +++ b/docs/content/about/release-notes.md @@ -47,7 +47,7 @@ Default sink HTTP clients now have a 10-second whole-request timeout plus bounde * Changed the loader to send the `monitor-stopped` and `monitor-stop-timeout` events from its stop goroutine, so an event receiver that panics on either event crashes Blip. Previously, when the API stopped a monitor, `net/http` recovered the panic. Event receivers must not panic. * **Behavior change:** `monitor.Loader.Stop` now keeps the monitor stopped across `Load` and `StartMonitors` until `Start`, unless its config changes or it's removed and loaded again. Go callers that relied on the next reload to restart a stopped monitor must call `Start`. `POST /monitors/stop` keeps a monitor stopped across reloads, as before. * **Behavior change:** `POST /monitors/start` now calls the `StartMonitor` plugin, so it returns 409 for a monitor that the plugin declines. Previously the API started the monitor directly and skipped the plugin, so it could start a monitor that the plugin filtered out. Integrations that start filtered monitors through the API must allow them in the plugin. -* A monitor stop, or a restart after one of its subsystems fails, no longer waits for the next collection tick, plan change check, or heartbeat. It cancels a collection in progress and drops its metrics and any metrics not yet sent to sinks, and it cancels the plan changer's read-only check and a heartbeat write in progress. +* A monitor stop, or a restart after a subsystem fails, no longer waits for the next collection tick, plan change check, or heartbeat. It cancels a collection, read-only check, or heartbeat write in progress, and drops metrics not yet sent to sinks. * Changed `CheckState` of the plan changer returned by `monitor.NewPlanChanger` to `CheckState(ctx)`. Canceling `ctx` cancels the read-only check. * **Behavior change:** Applied `timeout-connect` (default `10s`) to MySQL connections. Previously it was ignored, so connecting to an unreachable host waited for the operating system TCP connect timeout. Set it to `0` to retain the old behavior. An invalid `timeout-connect` value, which used to be ignored, now fails config validation, so Blip doesn't start, or a monitor in a reload doesn't load. For example, a bare number such as `timeout-connect: 5` has no unit and is invalid; use `5s`. * Added per-monitor fault counters, such as collector panics and timeouts, engine max runtime (EMR) timeouts, and dropped metrics, reported by the new [`GET /status/monitors/counters`]({{< ref "/api/status#get-statusmonitorscounters" >}}) API endpoint. Integrations can read them with the new `status.ReportCounters`, and register their own with `status.Counter`; the counter names are `status` constants like `status.COLLECTOR_PANICS`. diff --git a/docs/content/config/config-file.md b/docs/content/config/config-file.md index d2816f4..c96376b 100644 --- a/docs/content/config/config-file.md +++ b/docs/content/config/config-file.md @@ -204,7 +204,7 @@ A stop can take longer in these cases: * A collector, or its cleanup function, doesn't return promptly when its collection is canceled. A stop waits for every collector to return, then runs their cleanup functions. * A plan change is in progress, and a collector's `Prepare` doesn't return promptly when it's canceled. A stop cancels the plan change and waits for it. -* The monitor is starting. A stop waits for it to load its plans. If they're in a table ([`plans.table`](#table-1)), that's the only time a starting monitor queries MySQL, and it can take up to 5 seconds (connecting included) when MySQL is unreachable or hung. Making the monitor's connections also calls its credentials function (for example, to get an IAM auth token), which can take up to 500 milliseconds each time, and a custom `DbProviderFactory` that blocks while making them delays the stop too. +* The monitor is starting. A stop waits for it to make its connections, which calls its credentials function (for example, for an IAM auth token, up to 500 milliseconds each time) or a custom `DbProviderFactory`, and to load its plans, which queries MySQL only if they're in a table ([`plans.table`](#table-1)): up to 5 seconds, connecting included, when MySQL is unreachable or hung. --- From 3e4c3474d4cb026445c98da7877b3a811518a8ff Mon Sep 17 00:00:00 2001 From: Ian Oberst Date: Tue, 27 Oct 2026 10:42:41 -0700 Subject: [PATCH 11/12] Don't set stop statuses after a monitor stops A sink Send in progress at stop can return after the Loader removed the monitor's status. recvMetrics no longer sets the sink error or idle status then, which added back the removed monitor's status. It still sends the sink send error event. collect reports "stopped; metrics dropped" only if it didn't queue the metrics. If it queued them before the stop, it reports them collected, and recvMetrics drops them. The heartbeat writer returns right after a write canceled by a stop, instead of setting an error status first. Co-Authored-By: Claude Opus 5.5 --- heartbeat/writer.go | 6 +++ heartbeat/writer_stop_test.go | 54 ++++++++++++------- monitor/level_collector.go | 15 ++++-- monitor/level_collector_stop_test.go | 79 ++++++++++++++++++++++++++++ 4 files changed, 132 insertions(+), 22 deletions(-) diff --git a/heartbeat/writer.go b/heartbeat/writer.go index 9b06259..c9965b4 100644 --- a/heartbeat/writer.go +++ b/heartbeat/writer.go @@ -104,6 +104,9 @@ func (w *Writer) Write(stopChan, doneChan chan struct{}) error { ctx, cancel = context.WithTimeout(stopCtx, WriteTimeout) _, err = w.db.ExecContext(ctx, ping, params...) cancel() + if stopCtx.Err() != nil { + return nil // stopped; a canceled write isn't an error + } if err == nil { // success status.Monitor(w.monitorId, status.HEARTBEAT_WRITER, "sleep") break @@ -141,6 +144,9 @@ func (w *Writer) Write(stopChan, doneChan chan struct{}) error { ctx, cancel = context.WithTimeout(stopCtx, WriteTimeout) _, err = w.db.ExecContext(ctx, ping, w.srcId) cancel() + if stopCtx.Err() != nil { + return nil // stopped; a canceled write isn't an error + } if err != nil { blip.Debug("%s: %s", w.monitorId, err.Error()) if sqlutil.ReadOnly(err) { diff --git a/heartbeat/writer_stop_test.go b/heartbeat/writer_stop_test.go index 7848cfa..aea6778 100644 --- a/heartbeat/writer_stop_test.go +++ b/heartbeat/writer_stop_test.go @@ -75,14 +75,28 @@ func stopWriter(t *testing.T, w *heartbeat.Writer, monitorId, want string) time. time.Sleep(5 * time.Millisecond) } + // A stop isn't an error. The writer sets its final status, "stopped", when + // it returns, so check the status until then. This catches an error status + // only some of the time, so callers stop the writer several times. t0 := time.Now() close(stopChan) - select { - case <-doneChan: - case <-time.After(time.Minute): - t.Fatal("heartbeat writer did not stop") + timeout := time.After(time.Minute) + errStatus := "" + for { + if s := status.ReportMonitors(monitorId)[monitorId][status.HEARTBEAT_WRITER]; strings.Contains(s, "error") { + errStatus = s + } + select { + case <-doneChan: + if errStatus != "" { + t.Errorf("heartbeat writer status %q on stop, expected no error", errStatus) + } + return time.Since(t0) + case <-timeout: + t.Fatal("heartbeat writer did not stop") + default: + } } - return time.Since(t0) } func TestWriterStopsDuringWaits(t *testing.T) { @@ -133,20 +147,22 @@ func TestWriterStopsDuringWrite(t *testing.T) { defer writeTestHangAt.Store(0) - // First write (INSERT) hangs. The status is set before the write, so the - // stop can come just before the write starts, which tests stopping it too. - writeTestWrites.Store(0) - writeTestHangAt.Store(1) - w := heartbeat.NewWriter(monitorId, db, blip.ConfigHeartbeat{Freq: "1s", Table: "heartbeat"}) - if d := stopWriter(t, w, monitorId, "first insert"); d > timeout/10 { - t.Errorf("stop during first write took %s, expected well under %s", d, timeout) - } + for i := 0; i < 20; i++ { + // First write (INSERT) hangs. The status is set before the write, so the + // stop can come just before the write starts, which tests stopping it too. + writeTestWrites.Store(0) + writeTestHangAt.Store(1) + w := heartbeat.NewWriter(monitorId, db, blip.ConfigHeartbeat{Freq: "1s", Table: "heartbeat"}) + if d := stopWriter(t, w, monitorId, "first insert"); d > timeout/10 { + t.Errorf("stop during first write took %s, expected well under %s", d, timeout) + } - // The first write succeeds, and the first heartbeat (UPDATE) hangs - writeTestWrites.Store(0) - writeTestHangAt.Store(2) - w = heartbeat.NewWriter(monitorId, db, blip.ConfigHeartbeat{Freq: "10ms", Table: "heartbeat"}) - if d := stopWriter(t, w, monitorId, "write"); d > timeout/10 { - t.Errorf("stop during heartbeat write took %s, expected well under %s", d, timeout) + // The first write succeeds, and the first heartbeat (UPDATE) hangs + writeTestWrites.Store(0) + writeTestHangAt.Store(2) + w = heartbeat.NewWriter(monitorId, db, blip.ConfigHeartbeat{Freq: "10ms", Table: "heartbeat"}) + if d := stopWriter(t, w, monitorId, "write"); d > timeout/10 { + t.Errorf("stop during heartbeat write took %s, expected well under %s", d, timeout) + } } } diff --git a/monitor/level_collector.go b/monitor/level_collector.go index 7220570..debb9c9 100644 --- a/monitor/level_collector.go +++ b/monitor/level_collector.go @@ -154,7 +154,9 @@ var ( // // When ctx is canceled, metrics not yet sent are dropped, so a stopped or // removed monitor doesn't send metrics. recvMetrics checks after it receives -// metrics and before each sink Send. A Send in progress isn't canceled. +// metrics and before each sink Send. A Send in progress isn't canceled, but +// after it returns, recvMetrics doesn't set a status, which would add back the +// status of a monitor that was removed. func (c *lco) recvMetrics(ctx context.Context, doneChan chan struct{}) { defer func() { close(doneChan) @@ -171,7 +173,7 @@ func (c *lco) recvMetrics(ctx context.Context, doneChan chan struct{}) { } }() RECV: - for { + for ctx.Err() == nil { status.Monitor(c.monitorId, status.LEVEL_SINKS, "idle") select { case <-ctx.Done(): @@ -200,6 +202,11 @@ RECV: err := sink.Send(context.Background(), m) // @todo ctx with timeout if err != nil { c.event.Errorf(event.SINK_SEND_ERROR, "%s :%s", sinkName, err) // log by default + } + if ctx.Err() != nil { // stopped during Send + return + } + if err != nil { status.Monitor(c.monitorId, "error:"+sinkName, "%s", err.Error()) } else { status.RemoveComponent(c.monitorId, "error:"+sinkName) @@ -331,13 +338,14 @@ func (c *lco) stopChangePlan() { // recvMetrics hasn't sent yet are dropped too. func (c *lco) collect(ctx context.Context, interval uint, levelName string, startTime time.Time) { status.Monitor(c.monitorId, status.LEVEL_COLLECT, "%s/%s: collecting", c.plan.Name, levelName) + queued := false // for the sinks; recvMetrics drops them if ctx is canceled defer func() { if err := recover(); err != nil { // catch panic in engine and TransformMetrics b := make([]byte, 4096) n := runtime.Stack(b, false) c.event.Errorf(event.LCO_COLLECT_PANIC, "PANIC: %s: %s\n%s", c.monitorId, err, string(b[0:n])) } - if ctx.Err() != nil { + if !queued && ctx.Err() != nil { status.Monitor(c.monitorId, status.LEVEL_COLLECT, "%s/%s/%d: stopped; metrics dropped", c.plan.Name, levelName, interval) return } @@ -367,6 +375,7 @@ func (c *lco) collect(ctx context.Context, interval uint, levelName string, star status.Monitor(c.monitorId, status.LEVEL_COLLECT, "%s/%s: sending", c.plan.Name, levelName) select { case c.metricsChan <- metrics: + queued = true default: c.metricsDropped.Add(1) c.event.Errorf(event.LCO_METRICS_FAULT, "metrics channel blocked (check for sink errors), dropping metrics: %s", metrics) diff --git a/monitor/level_collector_stop_test.go b/monitor/level_collector_stop_test.go index 3425ace..f92bf25 100644 --- a/monitor/level_collector_stop_test.go +++ b/monitor/level_collector_stop_test.go @@ -4,6 +4,8 @@ package monitor import ( "context" + "errors" + "fmt" "strings" "sync" "sync/atomic" @@ -199,3 +201,80 @@ func TestLevelCollectorStopDropsMetricsAfterSend(t *testing.T) { } } } + +func TestLevelCollectorStopDuringSendDoesNotAddStatus(t *testing.T) { + // A sink Send in progress when the LCO stops can return after the Loader + // has removed the monitor's status. recvMetrics must not set a status + // then, which would add back the status of a removed monitor. Block the + // first Send, stop the LCO, remove the status, then let the Send return. + for _, sendErr := range []error{errors.New("send failed"), nil} { + t.Run(fmt.Sprintf("err=%v", sendErr), func(t *testing.T) { + const monitorId = "lco-stop-during-send" + defer status.RemoveMonitor(monitorId) + TickerDuration(10*time.Millisecond, time.Second) + defer TickerDuration(time.Second, time.Second) + + entered := make(chan struct{}) + release := make(chan struct{}) + returned := make(chan struct{}) + var once sync.Once + c := newStopTestLCO(monitorId, "every", mock.MetricsCollector{DomainFunc: func() string { return "test.lco-stop-during-send" }}) + c.sinks = []blip.Sink{mock.Sink{SendFunc: func(context.Context, *blip.Metrics) error { + err := errors.New("not first send") + once.Do(func() { + close(entered) + <-release + defer close(returned) + err = sendErr + }) + return err + }}} + + runStopLCO(t, c, func() { + select { + case <-entered: + case <-time.After(5 * time.Second): + t.Fatal("timeout waiting for sink Send") + } + }) + status.RemoveMonitor(monitorId) + close(release) + <-returned + time.Sleep(50 * time.Millisecond) // let recvMetrics return + if s, ok := status.ReportMonitors(monitorId)[monitorId]; ok { + t.Errorf("status of removed monitor added back: %v", s) + } + }) + } +} + +// cancelAfterFirstCheck is a context that reports canceled after the first +// Err call. In collect, that's the check right after Engine.Collect, so it's +// as if the LCO stopped after collect queued the metrics. +type cancelAfterFirstCheck struct { + context.Context + checks atomic.Int32 +} + +func (c *cancelAfterFirstCheck) Err() error { + if c.checks.Add(1) > 1 { + return context.Canceled + } + return nil +} + +func TestLevelCollectorStopAfterQueueReportsCollected(t *testing.T) { + // If the LCO stops after collect queued the metrics, collect reports + // them collected, not dropped: recvMetrics drops them, if it hasn't sent + // them yet, and clears the sinks status. + const monitorId = "lco-stop-after-queue" + defer status.RemoveMonitor(monitorId) + c := newStopTestLCO(monitorId, "every", mock.MetricsCollector{DomainFunc: func() string { return "test.lco-stop-after-queue" }}) + c.collect(&cancelAfterFirstCheck{Context: context.Background()}, 1, "every", time.Now()) + if n := len(c.metricsChan); n != 1 { + t.Fatalf("%d metrics queued, expected 1", n) + } + if s := status.ReportMonitors(monitorId)[monitorId][status.LEVEL_COLLECT]; !strings.Contains(s, ": collected in ") { + t.Errorf("%s status = %q, expected collected in", status.LEVEL_COLLECT, s) + } +} From 5769d7c2c946eaddb05056ab2bc496147dc1b09a Mon Sep 17 00:00:00 2001 From: Ian Oberst Date: Wed, 28 Oct 2026 09:06:58 -0700 Subject: [PATCH 12/12] Keep the plan changer's CheckState signature CheckState() again takes no arguments, so this isn't an API break. It calls the new checkState(ctx) with context.Background(), and Run calls checkState with its stop ctx. Co-Authored-By: Claude Opus 5.5 --- docs/content/about/release-notes.md | 1 - monitor/plan_changer.go | 18 ++++++++++++------ 2 files changed, 12 insertions(+), 7 deletions(-) diff --git a/docs/content/about/release-notes.md b/docs/content/about/release-notes.md index 086c8b1..7ccc1d9 100644 --- a/docs/content/about/release-notes.md +++ b/docs/content/about/release-notes.md @@ -48,7 +48,6 @@ Default sink HTTP clients now have a 10-second whole-request timeout plus bounde * **Behavior change:** `monitor.Loader.Stop` now keeps the monitor stopped across `Load` and `StartMonitors` until `Start`, unless its config changes or it's removed and loaded again. Go callers that relied on the next reload to restart a stopped monitor must call `Start`. `POST /monitors/stop` keeps a monitor stopped across reloads, as before. * **Behavior change:** `POST /monitors/start` now calls the `StartMonitor` plugin, so it returns 409 for a monitor that the plugin declines. Previously the API started the monitor directly and skipped the plugin, so it could start a monitor that the plugin filtered out. Integrations that start filtered monitors through the API must allow them in the plugin. * A monitor stop, or a restart after a subsystem fails, no longer waits for the next collection tick, plan change check, or heartbeat. It cancels a collection, read-only check, or heartbeat write in progress, and drops metrics not yet sent to sinks. -* Changed `CheckState` of the plan changer returned by `monitor.NewPlanChanger` to `CheckState(ctx)`. Canceling `ctx` cancels the read-only check. * **Behavior change:** Applied `timeout-connect` (default `10s`) to MySQL connections. Previously it was ignored, so connecting to an unreachable host waited for the operating system TCP connect timeout. Set it to `0` to retain the old behavior. An invalid `timeout-connect` value, which used to be ignored, now fails config validation, so Blip doesn't start, or a monitor in a reload doesn't load. For example, a bare number such as `timeout-connect: 5` has no unit and is invalid; use `5s`. * Added per-monitor fault counters, such as collector panics and timeouts, engine max runtime (EMR) timeouts, and dropped metrics, reported by the new [`GET /status/monitors/counters`]({{< ref "/api/status#get-statusmonitorscounters" >}}) API endpoint. Integrations can read them with the new `status.ReportCounters`, and register their own with `status.Counter`; the counter names are `status` constants like `status.COLLECTOR_PANICS`. * Added the optional `sink.CounterSink` interface for sinks that report monitor status counters; the monitor calls `RegisterCounters` each time it starts. A custom sink that wraps `sink.Retry` must forward `RegisterCounters` to it, else the retry buffer's `sink-retry-dropped` counter isn't reported. diff --git a/monitor/plan_changer.go b/monitor/plan_changer.go index 0644b5b..655af9f 100644 --- a/monitor/plan_changer.go +++ b/monitor/plan_changer.go @@ -122,10 +122,10 @@ func (pch *planChanger) setErr(err error) { } } -// Run calls CheckState every second; or, if offline, it uses an exponential +// Run calls checkState every second; or, if offline, it uses an exponential // backoff up to about 90 seconds (60 seconds plus jitter) until back online // (MySQL connection ok). There is no logic in this function; it's just a timed -// loop to call CheckState. It's run as a goroutine from Monitor.Run only if +// loop to call checkState. It's run as a goroutine from Monitor.Run only if // config.monitors.plans.adjust is enabled (blip.ConfigPlanChange.Enabled // returns true). func (pch *planChanger) Run(stopChan, doneChan chan struct{}) error { @@ -140,7 +140,7 @@ func (pch *planChanger) Run(stopChan, doneChan chan struct{}) error { defer cancel() for { - pch.CheckState(ctx) + pch.checkState(ctx) wait := 1 * time.Second if pch.curr.state == blip.STATE_OFFLINE { @@ -153,9 +153,15 @@ func (pch *planChanger) Run(stopChan, doneChan chan struct{}) error { } // CheckState checks the current state and changes state when necessary. -// This is the main logic, called periodically by Run. It does nothing if ctx -// is canceled (the PCH is stopping) while it checks the state. -func (pch *planChanger) CheckState(ctx context.Context) { +// It's the same check that Run does periodically, but it can't be canceled. +func (pch *planChanger) CheckState() { + pch.checkState(context.Background()) +} + +// checkState is the main logic of CheckState, called periodically by Run. It +// does nothing if ctx is canceled (the PCH is stopping) while it checks the +// state. +func (pch *planChanger) checkState(ctx context.Context) { now := Now() obsv := pch.state(ctx) if ctx.Err() != nil {