diff --git a/collector.go b/collector.go index 4a53dffb..1b6ee970 100644 --- a/collector.go +++ b/collector.go @@ -21,11 +21,18 @@ type Collector interface { // Prepare prepares a plan for future calls to Collect. The return function // is called once when the collector is destroyed; it allows the collector - // to clean up. It's also called if Prepare returns an error with a non-nil - // function, and if a later collector in the plan fails to prepare, because - // then the engine discards the collectors it already prepared. If Prepare - // returns an error, Blip will retry preparing the plan. Therefore, Prepare - // should not retry on error (for example, if the database is not online yet). + // to clean up. It's also called if a later collector in the plan fails to + // prepare, because then the engine discards the collectors it already + // prepared. If Prepare returns an error, Blip will retry preparing the + // plan. Therefore, Prepare should not retry on error (for example, if the + // database is not online yet). + // + // If Prepare returns an error, Blip discards the collector without calling + // Collect, so Prepare must release any resources that it started, like + // goroutines, before it returns, and it should return a nil cleanup func. + // As a backstop, Blip calls a non-nil cleanup func returned with an error, + // so the cleanup func must be safe to call after Prepare released the + // resources. Prepare(ctx context.Context, plan Plan) (func(), error) // Collect collects metrics for the previously prepared plan. Collect is only diff --git a/docs/content/about/release-notes.md b/docs/content/about/release-notes.md index 9b9b89d1..10fca47c 100644 --- a/docs/content/about/release-notes.md +++ b/docs/content/about/release-notes.md @@ -61,6 +61,10 @@ Default sink HTTP clients now have a 10-second whole-request timeout plus bounde * Added validation of the retry options: `buffer-size` must be from 1 to 10000, `send-timeout` must be greater than zero and at most 1m, and `send-retry-wait` must be greater than zero. Other values prevent the monitor from loading. Previously a built-in sink rejected any retry option, so no config that loaded before is affected. * Changed the retry sink to stop waiting `send-retry-wait` between sends when `send-timeout` expires. Previously a send cycle could run past `send-timeout` by up to one `send-retry-wait`. * Added `sink.RetryArgs.Validate` and the `sink.MAX_RETRY_BUFFER_SIZE` and `sink.MAX_RETRY_SEND_TIMEOUT` constants. `sink.NewRetry` doesn't call `Validate`, so custom sinks that build `RetryArgs` from user input should call it first. +* Fixed `repl.lag` at multiple levels with the Blip heartbeat writer: `report-no-heartbeat` now applies at every level, not only at one random level. Levels that don't set `writer`, or set it to `auto`, use the writer set at another level. Previously each of these levels auto-detected its own writer, so it could use `pfs` while another level used `blip`. The writer options `repl-check`, `table`, `source-id`, `source-role`, and `network-latency` are read from one level: the first level, by name, that sets `writer` to `pfs` or `blip`, else the first level that collects `repl.lag`. +* **Behavior change:** The `repl.lag` heartbeat reader now uses `repl-check`; previously it ignored it at every level. With the Blip heartbeat writer and `repl-check` set, an instance that isn't a replica now drops `repl.lag.current`, or reports -1 with `report-not-a-replica=yes`, instead of reporting heartbeat lag. +* **Behavior change:** Plans that set the `repl.lag` `writer` to `pfs` at one level and `blip` at another now fail to prepare, and the monitor keeps retrying the plan. Previously each level used its own writer. Set the same `writer` at every level, or set it at one level only. +* Changed `heartbeat.BlipReader.Stop` to wait for the reader to return, which takes up to the heartbeat read timeout (2s) if a heartbeat query is running. Previously the reader could report status after its monitor was stopped and removed, so the removed monitor reappeared in [`GET /status/monitors`]({{< ref "/api/status#get-statusmonitors" >}}). ### v2.1.0 (10 Aug 2026) diff --git a/docs/content/develop/collectors.md b/docs/content/develop/collectors.md index 24838860..ab3dc595 100644 --- a/docs/content/develop/collectors.md +++ b/docs/content/develop/collectors.md @@ -161,6 +161,7 @@ Then, when `Collect` is called, the collector quickly returns the latest value(s Using this method, the collector should return a cleanup function (callback) from `Prepare`. Blip calls collector cleanup functions if/when a monitor is stopped or restarted. +If `Prepare` returns an error, Blip discards the collector, so `Prepare` must stop any background threads that it started before it returns. ### ErrMore diff --git a/docs/content/metrics/domains/repl.lag/_index.md b/docs/content/metrics/domains/repl.lag/_index.md index 3e5a10ca..5c393e5e 100644 --- a/docs/content/metrics/domains/repl.lag/_index.md +++ b/docs/content/metrics/domains/repl.lag/_index.md @@ -71,6 +71,10 @@ Only available with MySQL 8.x Performance Schema. ## Options +When `repl.lag` is collected at more than one level, `writer`, `repl-check`, `table`, `source-id`, `source-role`, and `network-latency` are read from one level only: the first level, by name, that sets `writer` to `pfs` or `blip`, else the first level that collects `repl.lag`. +The options `report-no-heartbeat`, `report-not-a-replica`, and `default-channel-name` apply per level. +If one level sets `writer: pfs` and another sets `writer: blip`, the plan fails to prepare. + ### Common #### `repl-check` diff --git a/heartbeat/heartbeat_test.go b/heartbeat/heartbeat_test.go index e167ddd5..9c289b5e 100644 --- a/heartbeat/heartbeat_test.go +++ b/heartbeat/heartbeat_test.go @@ -189,11 +189,15 @@ func TestReader(t *testing.T) { // the real lag (and wait time, which we ignore) that the real waiter // returns to the reader. hbChan := make(chan int64, 1) + done := make(chan struct{}) // closed when test returns, so a blocked send doesn't block Stop realWaiter := heartbeat.SlowFastWaiter{NetworkLatency: 10 * time.Millisecond} mockWaiter := mock.LagWaiter{ WaitFunc: func(now, then time.Time, f int, srcId string) (int64, time.Duration) { lag, wait := realWaiter.Wait(now, then, f, srcId) - hbChan <- lag + select { + case hbChan <- lag: + case <-done: + } return lag, wait }, } @@ -204,6 +208,8 @@ func TestReader(t *testing.T) { Waiter: mockWaiter, }) hr.Start() + defer hr.Stop() // before ReadErrorWait is reset: Stop waits for the reader to return + defer close(done) // before Stop: unblocks the reader if it's sending to hbChan timeout := time.After(5 * time.Second) // this whole test should take <1s diff --git a/heartbeat/reader.go b/heartbeat/reader.go index 725b7b8e..28f69fd5 100644 --- a/heartbeat/reader.go +++ b/heartbeat/reader.go @@ -54,7 +54,8 @@ type BlipReader struct { lag int64 last time.Time stopChan chan struct{} - doneChan chan struct{} + doneChan chan struct{} // closed when run returns + started bool // run started, so Stop waits for doneChan isRepl bool event event.MonitorReceiver query string @@ -113,7 +114,20 @@ func NewBlipReader(args BlipReaderArgs) *BlipReader { return r } +// Start starts the reader in a goroutine. It doesn't start the reader if Start +// was called before, or if Stop was called first. func (r *BlipReader) Start() error { + r.Lock() + defer r.Unlock() + select { + case <-r.stopChan: + return nil // stopped before started + default: + } + if r.started { + return nil + } + r.started = true go r.run() return nil } @@ -211,15 +225,24 @@ func (r *BlipReader) wait(d time.Duration) bool { } } +// Stop stops the reader and waits for it to return, so it doesn't report +// status after Stop returns. If a heartbeat query is running, Stop waits for it +// to return, which takes up to ReadTimeout. Stop can be called more than once, +// concurrently, and before Start. func (r *BlipReader) Stop() { r.Lock() select { case <-r.stopChan: - case <-r.doneChan: default: close(r.stopChan) } + started := r.started r.Unlock() + + // Wait without the lock: run locks to update lag + if started { + <-r.doneChan + } } func (r *BlipReader) Lag(_ context.Context) (Lag, error) { diff --git a/heartbeat/reader_query_test.go b/heartbeat/reader_query_test.go new file mode 100644 index 00000000..4914b538 --- /dev/null +++ b/heartbeat/reader_query_test.go @@ -0,0 +1,20 @@ +// Copyright 2026 Block, Inc. + +package heartbeat + +import ( + "strings" + "testing" +) + +func TestBlipReaderQuerySelectsReplCheck(t *testing.T) { + r := NewBlipReader(BlipReaderArgs{MonitorId: "m1", Table: "blip.heartbeat", ReplCheck: "read_only"}) + if !strings.Contains(r.query, "@@read_only") { + t.Errorf("query %q does not select @@read_only", r.query) + } + + r = NewBlipReader(BlipReaderArgs{MonitorId: "m1", Table: "blip.heartbeat"}) + if strings.Contains(r.query, "@@") { + t.Errorf("query %q selects a variable, expected none without repl check", r.query) + } +} diff --git a/heartbeat/reader_stop_test.go b/heartbeat/reader_stop_test.go new file mode 100644 index 00000000..da6057df --- /dev/null +++ b/heartbeat/reader_stop_test.go @@ -0,0 +1,108 @@ +// Copyright 2026 Block, Inc. + +package heartbeat + +import ( + "context" + "database/sql" + "database/sql/driver" + "errors" + "sync" + "testing" + "time" + + "github.com/cashapp/blip/v2/status" +) + +// blockingDB is a database whose queries block until release is closed (or +// the query context is done), then fail. +type blockingDB struct { + running chan struct{} // receives when a query starts + release chan struct{} +} + +func newBlockingDB(t *testing.T) (*sql.DB, *blockingDB) { + b := &blockingDB{running: make(chan struct{}, 1), release: make(chan struct{})} + db := sql.OpenDB(b) + t.Cleanup(func() { db.Close() }) + return db, b +} + +func (b *blockingDB) Connect(context.Context) (driver.Conn, error) { return blockingConn{b}, nil } +func (b *blockingDB) Driver() driver.Driver { return nil } + +type blockingConn struct{ b *blockingDB } + +func (c blockingConn) QueryContext(ctx context.Context, _ string, _ []driver.NamedValue) (driver.Rows, error) { + select { + case c.b.running <- struct{}{}: + default: + } + select { + case <-c.b.release: + case <-ctx.Done(): + } + return nil, errors.New("query failed") +} + +func (blockingConn) Prepare(string) (driver.Stmt, error) { return nil, errors.New("not supported") } +func (blockingConn) Close() error { return nil } +func (blockingConn) Begin() (driver.Tx, error) { return nil, errors.New("not supported") } + +func TestBlipReaderStopWaitsForRunningQuery(t *testing.T) { + // The monitor loader removes a monitor's status after stopping it. If Stop + // returned while a heartbeat query was running, the reader would report + // status after that, which makes the removed monitor reappear. + monitorId := "stop-waits" + db, d := newBlockingDB(t) + r := NewBlipReader(BlipReaderArgs{MonitorId: monitorId, DB: db, Table: "blip.heartbeat"}) + r.Start() + select { + case <-d.running: + case <-time.After(5 * time.Second): + t.Fatal("heartbeat query didn't start") + } + + // Stop while the query is running, which returns after Stop is called + time.AfterFunc(100*time.Millisecond, func() { close(d.release) }) + var wg sync.WaitGroup + for i := 0; i < 2; i++ { // concurrent Stop calls + wg.Add(1) + go func() { + defer wg.Done() + r.Stop() + }() + } + wg.Wait() + status.RemoveMonitor(monitorId) + + <-r.doneChan // reader returned, so it reported any status it was going to + if got := status.ReportMonitors(monitorId)[monitorId]; len(got) != 0 { + t.Errorf("monitor status after Stop and RemoveMonitor: %v, expected none", got) + } + r.Stop() // again, after it stopped +} + +func TestBlipReaderStopBeforeStart(t *testing.T) { + // Repl.lag calls Start in a goroutine, so Stop can be called first + db, _ := newBlockingDB(t) + r := NewBlipReader(BlipReaderArgs{MonitorId: "stop-before-start", DB: db, Table: "blip.heartbeat"}) + stopped := make(chan struct{}) + go func() { + r.Stop() + r.Start() // doesn't start a stopped reader + r.Stop() + close(stopped) + }() + select { + case <-stopped: + case <-time.After(5 * time.Second): + t.Fatal("Stop blocked for a reader that wasn't started") + } + r.Lock() + started := r.started + r.Unlock() + if started { + t.Error("Start started a stopped reader") + } +} diff --git a/metrics/repl.lag/lag.go b/metrics/repl.lag/lag.go index ed35a50e..5972dbd9 100644 --- a/metrics/repl.lag/lag.go +++ b/metrics/repl.lag/lag.go @@ -6,6 +6,7 @@ import ( "context" "database/sql" "fmt" + "sort" "strconv" "time" @@ -33,7 +34,8 @@ const ( type Lag struct { db *sql.DB lagReader heartbeat.Reader - stopReader func() // stops lagReader; returned by Prepare as cleanup + stopReader func() // stops lagReader; returned by Prepare as cleanup + newReader func(heartbeat.BlipReaderArgs) heartbeat.Reader // tests replace it to inspect the args lagWriterIn map[string]string dropNoHeartbeat map[string]bool dropNotAReplica map[string]bool @@ -45,9 +47,14 @@ type Lag struct { var _ blip.Collector = &Lag{} +func newBlipReader(args heartbeat.BlipReaderArgs) heartbeat.Reader { + return heartbeat.NewBlipReader(args) +} + func NewLag(db *sql.DB) *Lag { return &Lag{ db: db, + newReader: newBlipReader, lagWriterIn: map[string]string{}, dropNoHeartbeat: map[string]bool{}, dropNotAReplica: map[string]bool{}, @@ -149,11 +156,19 @@ func (c *Lag) Help() blip.CollectorHelp { // domain collects only one metric (repl.lag.current), there's no need to // collect different metrics at different frequencies. // +// One level sets up the writer: the first level, by name, that sets the +// writer option to pfs or blip; else the first level that collects lag, which +// auto-detects the writer. Levels that don't set the writer, or set it to +// auto, use that writer. It's an error if one level sets pfs and another sets +// blip. The writer and its options (repl-check, table, source-id, +// source-role, network-latency) are read only from that level. +// The report-no-heartbeat, report-not-a-replica, and default-channel-name +// options are per level. +// // If lag is collected from a Blip heartbeat, the returned cleanup func stops -// the heartbeat reader; else it's nil. +// the heartbeat reader; else it's nil. On error, Prepare stops the reader +// and returns a nil cleanup func, as required by blip.Collector. func (c *Lag) Prepare(ctx context.Context, plan blip.Plan) (cleanup func(), err error) { - configured := "" // set after first level to its writer value - // The engine discards this collector if Prepare fails, so stop the // reader if it was started before the error. defer func() { @@ -162,67 +177,82 @@ func (c *Lag) Prepare(ctx context.Context, plan blip.Plan) (cleanup func(), err } }() -LEVEL: + // Levels that collect lag, sorted so the result doesn't depend on map order + levels := []string{} for levelName, level := range plan.Levels { - dom, ok := level.Collect[DOMAIN] - if !ok { - continue LEVEL // not collected in this level - } - - writer := dom.Options[OPT_WRITER] - - // Already configured? If yes and same writer, that's ok and expected - // (lag collected at multiple levels). But if writer is different, that's - // and error. - if configured != "" { - if configured != writer { - return nil, fmt.Errorf("different writer configuration: %s != %s", configured, writer) - } - c.lagWriterIn[levelName] = writer // collect at this level - continue LEVEL + if _, ok := level.Collect[DOMAIN]; ok { + levels = append(levels, levelName) } + } + if len(levels) == 0 { + return nil, nil + } + sort.Strings(levels) - blip.Debug("repl.lag: config from level %s", levelName) - switch writer { - case LAG_WRITER_PFS: - // Try collecting, discard metrics - if _, err = c.collectPFS(ctx, levelName); err != nil { - return nil, err - } - case LAG_WRITER_BLIP: - cleanup, err = c.prepareBlip(levelName, plan.MonitorId, plan.Name, dom.Options) - if err != nil { - return nil, err - } - case "auto", "": // default - // Try PFS first - if _, err = c.collectPFS(ctx, levelName); err == nil { - blip.Debug("repl.lag auto-detected PFS") - writer = LAG_WRITER_PFS - } else { - // then Blip HeartBeat - if cleanup, err = c.prepareBlip(levelName, plan.MonitorId, plan.Name, dom.Options); err == nil { - blip.Debug("repl.lag auto-detected Blip heartbeat") - writer = LAG_WRITER_BLIP - } else { - return nil, fmt.Errorf("failed to auto-detect source, set %s manually", OPT_WRITER) - } - } + // Find the level that sets up the writer + configLevel := levels[0] + writer := "auto" + for _, levelName := range levels { + opt := plan.Levels[levelName].Collect[DOMAIN].Options[OPT_WRITER] + switch opt { + case "", "auto": + continue + case LAG_WRITER_PFS, LAG_WRITER_BLIP: default: - return nil, fmt.Errorf("invalid lag writer: %q; valid values: auto, pfs, blip", writer) + return nil, fmt.Errorf("invalid lag writer: %q; valid values: auto, pfs, blip", opt) } + if writer == "auto" { + configLevel = levelName + writer = opt + } else if opt != writer { + return nil, fmt.Errorf("different writer configuration: %s at level %s != %s at level %s", writer, configLevel, opt, levelName) + } + } + + if writer, err = c.configure(ctx, configLevel, plan, writer, plan.Levels[configLevel].Collect[DOMAIN].Options); err != nil { + return nil, err + } + for _, levelName := range levels { + options := plan.Levels[levelName].Collect[DOMAIN].Options c.lagWriterIn[levelName] = writer // collect at this level + c.dropNoHeartbeat[levelName] = !blip.Bool(options[OPT_REPORT_NO_HEARTBEAT]) + c.dropNotAReplica[levelName] = !blip.Bool(options[OPT_REPORT_NOT_A_REPLICA]) + c.defaultChannelNameOverrides[levelName] = options[OPT_DEFAULT_CHANNEL_NAME] + } - c.dropNotAReplica[levelName] = !blip.Bool(dom.Options[OPT_REPORT_NOT_A_REPLICA]) - c.defaultChannelNameOverrides[levelName] = dom.Options[OPT_DEFAULT_CHANNEL_NAME] - if err := c.verifyReplCheck(ctx, dom.Options[OPT_REPL_CHECK]); err != nil { - return nil, err + return c.stopReader, nil +} + +// configure sets up the writer from the options of one level. It returns the +// writer, which is never auto. +func (c *Lag) configure(ctx context.Context, levelName string, plan blip.Plan, writer string, options map[string]string) (string, error) { + blip.Debug("repl.lag: config from level %s", levelName) + switch writer { + case LAG_WRITER_PFS: + // Try collecting, discard metrics + if _, err := c.collectPFS(ctx, levelName); err != nil { + return "", err + } + case LAG_WRITER_BLIP: + c.prepareBlip(levelName, plan.MonitorId, plan.Name, options) + default: // auto + // Try PFS first, then Blip heartbeat + if _, err := c.collectPFS(ctx, levelName); err == nil { + blip.Debug("repl.lag auto-detected PFS") + writer = LAG_WRITER_PFS + } else { + c.prepareBlip(levelName, plan.MonitorId, plan.Name, options) + blip.Debug("repl.lag auto-detected Blip heartbeat") + writer = LAG_WRITER_BLIP } - c.replCheck = dom.Options[OPT_REPL_CHECK] } - return cleanup, nil + if err := c.verifyReplCheck(ctx, options[OPT_REPL_CHECK]); err != nil { + return "", err + } + c.replCheck = options[OPT_REPL_CHECK] + return writer, nil } func (c *Lag) verifyReplCheck(ctx context.Context, variable string) error { @@ -253,15 +283,12 @@ func (c *Lag) Collect(ctx context.Context, levelName string) ([]blip.MetricValue // Internal methods // ////////////////////////////////////////////////////////////////////////// -func (c *Lag) prepareBlip(levelName string, monitorID string, planName string, options map[string]string) (func(), error) { - // Prepare calls this func for each level that collects lag. Return the - // same cleanup func for every level, else the reader is never stopped. +// prepareBlip starts the heartbeat reader. Only 1 reader per collector. +func (c *Lag) prepareBlip(levelName string, monitorID string, planName string, options map[string]string) { if c.lagReader != nil { - return c.stopReader, nil + return } - c.dropNoHeartbeat[levelName] = !blip.Bool(options[OPT_REPORT_NO_HEARTBEAT]) - table := options[OPT_HEARTBEAT_TABLE] if table == "" { table = blip.DEFAULT_HEARTBEAT_TABLE @@ -275,14 +302,13 @@ func (c *Lag) prepareBlip(levelName string, monitorID string, planName string, o netLatency = time.Duration(n) * time.Millisecond } } - // Only 1 reader per plan - c.lagReader = heartbeat.NewBlipReader(heartbeat.BlipReaderArgs{ + c.lagReader = c.newReader(heartbeat.BlipReaderArgs{ MonitorId: monitorID, DB: c.db, Table: table, SourceId: options[OPT_HEARTBEAT_SOURCE_ID], SourceRole: options[OPT_HEARTBEAT_SOURCE_ROLE], - ReplCheck: c.replCheck, + ReplCheck: options[OPT_REPL_CHECK], // not c.replCheck: it's set after the reader starts Waiter: heartbeat.SlowFastWaiter{ MonitorId: monitorID, NetworkLatency: netLatency, @@ -290,12 +316,10 @@ func (c *Lag) prepareBlip(levelName string, monitorID string, planName string, o }) go c.lagReader.Start() blip.Debug("%s: started reader: %s/%s (network latency: %s)", monitorID, planName, levelName, netLatency) - c.lagWriterIn[levelName] = LAG_WRITER_BLIP c.stopReader = func() { blip.Debug("%s: stopping reader", monitorID) c.lagReader.Stop() } - return c.stopReader, nil } func (c *Lag) collectBlip(ctx context.Context, levelName string) ([]blip.MetricValue, error) { diff --git a/metrics/repl.lag/lag_test.go b/metrics/repl.lag/lag_test.go index 16f88241..5baddb99 100644 --- a/metrics/repl.lag/lag_test.go +++ b/metrics/repl.lag/lag_test.go @@ -6,12 +6,14 @@ import ( "context" "database/sql" "errors" + "strings" "testing" "time" "github.com/stretchr/testify/assert" "github.com/cashapp/blip/v2" + "github.com/cashapp/blip/v2/heartbeat" "github.com/cashapp/blip/v2/test" "github.com/cashapp/blip/v2/test/mock" ) @@ -223,3 +225,194 @@ func TestCollectPFSReturnsDeadlineExceeded(t *testing.T) { t.Errorf("Collect returned error %v, expected an error wrapping context.DeadlineExceeded", err) } } + +// fakeReader is a heartbeat reader that reports lag without a database. +type fakeReader struct { + lag heartbeat.Lag +} + +func (r fakeReader) Start() error { return nil } +func (r fakeReader) Stop() {} +func (r fakeReader) Lag(context.Context) (heartbeat.Lag, error) { return r.lag, nil } + +func TestPrepareBlipAtMultipleLevelsDropsNoHeartbeat(t *testing.T) { + // report-no-heartbeat defaults to no (drop), and that applies to every + // level, not only to the level that started the reader. The reader can't + // reach the database, so it reports no heartbeat. + for _, report := range []string{"", "no", "yes"} { + t.Run("report-no-heartbeat="+report, func(t *testing.T) { + c := NewLag(unreachableDB(t)) + plan := blipPlan("fast", "medium", "slow") + for _, level := range plan.Levels { + if report != "" { + level.Collect[DOMAIN].Options[OPT_REPORT_NO_HEARTBEAT] = report + } + } + cleanup, err := c.Prepare(context.Background(), plan) + stopReader(t, cleanup) + if err != nil { + t.Fatal(err) + } + for level := range plan.Levels { + metrics, err := c.Collect(context.Background(), level) + if err != nil { + t.Fatalf("%s: %s", level, err) + } + if report == "yes" { + if len(metrics) != 1 || metrics[0].Value != -1 { + t.Errorf("%s: got %+v, expected repl.lag.current = -1", level, metrics) + } + } else if len(metrics) != 0 { + t.Errorf("%s: got %+v, expected no metrics (dropped)", level, metrics) + } + } + }) + } +} + +func TestPrepareBlipPassesReplCheckToReader(t *testing.T) { + // The reader selects @@ to report whether the instance is a + // replica, so it must be made with the repl-check option. The repl check + // is verified after the reader starts, and that fails because the database + // is unreachable, but by then the reader has been made. + c := NewLag(unreachableDB(t)) + var args []heartbeat.BlipReaderArgs + c.newReader = func(a heartbeat.BlipReaderArgs) heartbeat.Reader { + args = append(args, a) + return fakeReader{} + } + plan := blipPlan("fast", "slow") + for _, level := range plan.Levels { + level.Collect[DOMAIN].Options[OPT_REPL_CHECK] = "read_only" + } + if _, err := c.Prepare(context.Background(), plan); err == nil { + t.Fatal("Prepare returned nil error, expected repl check error") + } + if len(args) != 1 { + t.Fatalf("Prepare made %d heartbeat readers, expected 1", len(args)) + } + assert.Equal(t, "read_only", args[0].ReplCheck) +} + +func TestPrepareWritersAtDifferentLevels(t *testing.T) { + // Levels that don't set the writer, or set auto, use the writer set at + // another level. Plans have them because levels inherit options only from + // some lower levels. But pfs at one level and blip at another is an error. + // Each pair is tested in both orders, and many times because Prepare + // ranges over a map. + tests := []struct { + fast, slow string + writer string // expected at every level, or "" if Prepare fails + err string // expected error, if writer is "" + }{ + {fast: "", slow: LAG_WRITER_BLIP, writer: LAG_WRITER_BLIP}, + {fast: LAG_WRITER_BLIP, slow: "", writer: LAG_WRITER_BLIP}, + {fast: "auto", slow: LAG_WRITER_BLIP, writer: LAG_WRITER_BLIP}, + {fast: LAG_WRITER_BLIP, slow: "auto", writer: LAG_WRITER_BLIP}, + {fast: LAG_WRITER_BLIP, slow: LAG_WRITER_BLIP, writer: LAG_WRITER_BLIP}, + {fast: "", slow: "auto", writer: LAG_WRITER_BLIP}, // auto-detected: PFS fails because database is unreachable + {fast: LAG_WRITER_PFS, slow: LAG_WRITER_BLIP, err: "different writer configuration"}, + {fast: LAG_WRITER_BLIP, slow: LAG_WRITER_PFS, err: "different writer configuration"}, + // pfs is set up, not auto-detected or blip, so it fails because the + // database is unreachable. See TestPrepareAutoAndPFSOnMySQL80. + {fast: "auto", slow: LAG_WRITER_PFS, err: "could not check replication lag"}, + {fast: LAG_WRITER_PFS, slow: "", err: "could not check replication lag"}, + } + for _, tt := range tests { + t.Run(tt.fast+"/"+tt.slow, func(t *testing.T) { + for i := 0; i < 20; i++ { + c := NewLag(unreachableDB(t)) + readers := 0 + c.newReader = func(heartbeat.BlipReaderArgs) heartbeat.Reader { + readers++ + return fakeReader{} + } + plan := blipPlan("fast", "slow") + plan.Levels["fast"].Collect[DOMAIN].Options[OPT_WRITER] = tt.fast + plan.Levels["slow"].Collect[DOMAIN].Options[OPT_WRITER] = tt.slow + _, err := c.Prepare(context.Background(), plan) + if tt.writer == "" { + if err == nil || !strings.Contains(err.Error(), tt.err) { + t.Fatalf("run %d: got error %v, expected %q", i, err, tt.err) + } + continue + } + if err != nil { + t.Fatalf("run %d: %s", i, err) + } + if readers != 1 { + t.Fatalf("run %d: Prepare made %d heartbeat readers, expected 1", i, readers) + } + want := map[string]string{"fast": tt.writer, "slow": tt.writer} + if !assert.Equal(t, want, c.lagWriterIn, "run %d", i) { + return + } + } + }) + } +} + +func TestPrepareSetsOptionsAtEveryLevel(t *testing.T) { + // The writer is set up once, but every level has its own report and + // channel name options. + c := NewLag(unreachableDB(t)) + c.newReader = func(heartbeat.BlipReaderArgs) heartbeat.Reader { + return fakeReader{lag: heartbeat.Lag{Replica: false, Milliseconds: -1}} + } + plan := blipPlan("fast", "medium", "slow") + plan.Levels["fast"].Collect[DOMAIN].Options[OPT_REPORT_NO_HEARTBEAT] = "yes" + plan.Levels["medium"].Collect[DOMAIN].Options[OPT_REPORT_NOT_A_REPLICA] = "yes" + plan.Levels["slow"].Collect[DOMAIN].Options[OPT_DEFAULT_CHANNEL_NAME] = "main" + if _, err := c.Prepare(context.Background(), plan); err != nil { + t.Fatal(err) + } + assert.Equal(t, map[string]string{"fast": LAG_WRITER_BLIP, "medium": LAG_WRITER_BLIP, "slow": LAG_WRITER_BLIP}, c.lagWriterIn) + assert.Equal(t, map[string]bool{"fast": false, "medium": true, "slow": true}, c.dropNoHeartbeat) + assert.Equal(t, map[string]bool{"fast": true, "medium": false, "slow": true}, c.dropNotAReplica) + assert.Equal(t, map[string]string{"fast": "", "medium": "", "slow": "main"}, c.defaultChannelNameOverrides) + + // Not a replica: reported only at the level with report-not-a-replica=yes + for level := range plan.Levels { + metrics, err := c.Collect(context.Background(), level) + if err != nil { + t.Fatalf("%s: %s", level, err) + } + if level == "medium" { + assert.Len(t, metrics, 1, level) + } else { + assert.Empty(t, metrics, level) + } + } +} + +func TestPrepareAutoAndPFSOnMySQL80(t *testing.T) { + _, db, err := test.Connection("mysql80") + if err != nil { + if test.Build { + t.Skip("mysql80 not running") + } else { + t.Fatal(err) + } + } + defer db.Close() + + // Same as auto and blip in TestPrepareWritersAtDifferentLevels, but pfs + // needs MySQL + for _, writers := range [][2]string{{"auto", LAG_WRITER_PFS}, {LAG_WRITER_PFS, ""}} { + for i := 0; i < 20; i++ { + c := NewLag(db) + plan := blipPlan("fast", "slow") + plan.Levels["fast"].Collect[DOMAIN].Options[OPT_WRITER] = writers[0] + plan.Levels["slow"].Collect[DOMAIN].Options[OPT_WRITER] = writers[1] + cleanup, err := c.Prepare(context.Background(), plan) + stopReader(t, cleanup) + if err != nil { + t.Fatalf("%v run %d: %s", writers, i, err) + } + if c.lagReader != nil { + t.Fatalf("%v run %d: Prepare made a heartbeat reader for pfs", writers, i) + } + assert.Equal(t, map[string]string{"fast": LAG_WRITER_PFS, "slow": LAG_WRITER_PFS}, c.lagWriterIn) + } + } +} diff --git a/monitor/engine.go b/monitor/engine.go index 2b0c3097..c955aa3e 100644 --- a/monitor/engine.go +++ b/monitor/engine.go @@ -224,6 +224,9 @@ func (e *Engine) Prepare(ctx context.Context, plan blip.Plan, before, after func status.Monitor(e.monitorId, status.ENGINE_PREPARE, "%s: prepare collector %s", plan.Name, domain) cleanup, err := c.Prepare(ctx, plan) if err != nil { + // The collector should have released its resources and returned + // nil, but call a returned cleanup func as a backstop. See + // blip.Collector.Prepare. if cleanup != nil { cleanup() }