fix(database_observability): Ensure that connection_info metric is only emitted for a given DB instance when it is available
#5707
Next
Next commit
Add a shared connection_info_monitor component which is used to perio…
…dically check database connectivity and register/deregister the database_observability_connection_info metric which is used for billing purposes
- Loading branch information
commit 524bb097638c585f9de3dc86515ca23da31d05a5
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,87 @@ | ||
| package database_observability | ||
|
|
||
| import ( | ||
| "context" | ||
| "database/sql" | ||
| "time" | ||
|
|
||
| "github.com/prometheus/client_golang/prometheus" | ||
| ) | ||
|
|
||
| // ConnectionCheckInterval is how often the connection_info collector pings the DB to verify connectivity. | ||
| const ConnectionCheckInterval = 60 * time.Second | ||
|
|
||
| // ConnectionChecksThreshold is the number of consecutive failed pings before unregistering the metric, | ||
| // and the number of consecutive successful pings before re-registering it after a disconnect. | ||
| const ConnectionChecksThreshold = 3 | ||
|
|
||
| // ConnectionInfoMonitorConfig optionally overrides the default check interval and threshold. | ||
| // Used by tests to run the monitor with shorter intervals. If nil, defaults are used. | ||
| type ConnectionInfoMonitorConfig struct { | ||
| CheckInterval time.Duration | ||
| ChecksThreshold int | ||
| } | ||
|
|
||
| // RunConnectionInfoMonitor starts a goroutine that pings db every ConnectionCheckInterval. | ||
| // After ConnectionChecksThreshold consecutive ping failures it unregisters infoMetric from registry. | ||
| // After ConnectionChecksThreshold consecutive ping successes (when the metric is unregistered) it re-registers | ||
| // infoMetric and sets it to 1 with the given labelValues. | ||
| // The goroutine runs until ctx is done. onStopped is called when the goroutine exits (e.g. when ctx is cancelled). | ||
| // RunConnectionInfoMonitor returns a cancel function that cancels the context passed to the goroutine; the caller | ||
| // should call cancel in Stop() to ensure the goroutine exits. | ||
| // labelValues must contain exactly 6 values in order: provider_name, provider_region, provider_account, | ||
| // db_instance_identifier, engine, engine_version. | ||
| // If config is non-nil, its CheckInterval and ChecksThreshold override the default constants (used for testing). | ||
| func RunConnectionInfoMonitor(ctx context.Context, db *sql.DB, registry *prometheus.Registry, infoMetric *prometheus.GaugeVec, labelValues []string, onStopped func(), config *ConnectionInfoMonitorConfig) (cancel context.CancelFunc) { | ||
| interval := ConnectionCheckInterval | ||
| threshold := ConnectionChecksThreshold | ||
| if config != nil { | ||
| if config.CheckInterval > 0 { | ||
| interval = config.CheckInterval | ||
| } | ||
| if config.ChecksThreshold > 0 { | ||
| threshold = config.ChecksThreshold | ||
| } | ||
| } | ||
| ctx, cancel = context.WithCancel(ctx) | ||
| go func() { | ||
| defer onStopped() | ||
| ticker := time.NewTicker(interval) | ||
| defer ticker.Stop() | ||
|
|
||
| var consecutiveFailures, consecutiveSuccesses int | ||
| metricRegistered := true | ||
| for { | ||
| if err := db.PingContext(ctx); err != nil { | ||
| consecutiveFailures++ | ||
| consecutiveSuccesses = 0 | ||
| if metricRegistered && consecutiveFailures >= threshold { | ||
| registry.Unregister(infoMetric) | ||
| metricRegistered = false | ||
| consecutiveFailures = 0 | ||
| } | ||
| } else { | ||
| consecutiveFailures = 0 | ||
| if metricRegistered { | ||
| consecutiveSuccesses = 0 | ||
| } else { | ||
| consecutiveSuccesses++ | ||
| if consecutiveSuccesses >= threshold { | ||
| registry.MustRegister(infoMetric) | ||
| infoMetric.WithLabelValues(labelValues[0], labelValues[1], labelValues[2], labelValues[3], labelValues[4], labelValues[5]).Set(1) | ||
|
Contributor
There was a problem hiding this comment. Can we make labelValues a typed struct instead? this is hard to read and understand. |
||
| metricRegistered = true | ||
| consecutiveSuccesses = 0 | ||
| } | ||
| } | ||
| } | ||
|
|
||
| select { | ||
| case <-ctx.Done(): | ||
| return | ||
| case <-ticker.C: | ||
| // continue loop | ||
| } | ||
| } | ||
| }() | ||
| return cancel | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,219 @@ | ||
| package database_observability | ||
|
|
||
| import ( | ||
| "context" | ||
| "errors" | ||
| "testing" | ||
| "time" | ||
|
|
||
| "github.com/DATA-DOG/go-sqlmock" | ||
| "github.com/prometheus/client_golang/prometheus" | ||
| dto "github.com/prometheus/client_model/go" | ||
| "github.com/stretchr/testify/require" | ||
| "go.uber.org/goleak" | ||
| ) | ||
|
|
||
| const testCheckInterval = 15 * time.Millisecond | ||
| const testThreshold = 3 | ||
|
|
||
| func TestRunConnectionInfoMonitor_UnregistersAfterConsecutiveFailures(t *testing.T) { | ||
| defer goleak.VerifyNone(t) | ||
|
|
||
| db, mock, err := sqlmock.New(sqlmock.MonitorPingsOption(true)) | ||
| require.NoError(t, err) | ||
| defer db.Close() | ||
|
|
||
| registry := prometheus.NewRegistry() | ||
| infoMetric := prometheus.NewGaugeVec(prometheus.GaugeOpts{ | ||
| Namespace: "database_observability", | ||
| Name: "connection_info", | ||
| Help: "Information about the connection", | ||
| }, []string{"provider_name", "provider_region", "provider_account", "db_instance_identifier", "engine", "engine_version"}) | ||
| require.NoError(t, registry.Register(infoMetric)) | ||
|
|
||
| labelValues := []string{"aws", "us-east-1", "123456789", "my-db", "postgres", "15.0"} | ||
| infoMetric.WithLabelValues(labelValues[0], labelValues[1], labelValues[2], labelValues[3], labelValues[4], labelValues[5]).Set(1) | ||
|
|
||
| // Expect 3 pings, all failing | ||
| pingErr := errors.New("connection refused") | ||
| for i := 0; i < testThreshold; i++ { | ||
| mock.ExpectPing().WillReturnError(pingErr) | ||
| } | ||
|
|
||
| ctx := context.Background() | ||
| onStopped := func() {} | ||
| config := &ConnectionInfoMonitorConfig{ | ||
| CheckInterval: testCheckInterval, | ||
| ChecksThreshold: testThreshold, | ||
| } | ||
| cancel := RunConnectionInfoMonitor(ctx, db, registry, infoMetric, labelValues, onStopped, config) | ||
| defer cancel() | ||
|
|
||
| // Wait for at least 3 tick intervals so the monitor performs 3 failed pings and unregisters | ||
| time.Sleep(testCheckInterval*time.Duration(testThreshold) + 20*time.Millisecond) | ||
|
|
||
| // Metric should have been unregistered (not present in gather) | ||
| metrics, err := registry.Gather() | ||
| require.NoError(t, err) | ||
| var found bool | ||
| for _, mf := range metrics { | ||
| if mf.GetName() == "database_observability_connection_info" { | ||
| found = true | ||
| break | ||
| } | ||
| } | ||
| require.False(t, found, "metric should be unregistered after %d consecutive ping failures", testThreshold) | ||
|
|
||
| require.NoError(t, mock.ExpectationsWereMet()) | ||
| } | ||
|
|
||
| func TestRunConnectionInfoMonitor_ReregistersAfterConsecutiveSuccesses(t *testing.T) { | ||
| defer goleak.VerifyNone(t) | ||
|
|
||
| db, mock, err := sqlmock.New(sqlmock.MonitorPingsOption(true)) | ||
| require.NoError(t, err) | ||
| defer db.Close() | ||
|
|
||
| registry := prometheus.NewRegistry() | ||
| infoMetric := prometheus.NewGaugeVec(prometheus.GaugeOpts{ | ||
| Namespace: "database_observability", | ||
| Name: "connection_info", | ||
| Help: "Information about the connection", | ||
| }, []string{"provider_name", "provider_region", "provider_account", "db_instance_identifier", "engine", "engine_version"}) | ||
| require.NoError(t, registry.Register(infoMetric)) | ||
|
|
||
| labelValues := []string{"aws", "us-east-1", "123456789", "my-db", "mysql", "8.0.32"} | ||
| infoMetric.WithLabelValues(labelValues[0], labelValues[1], labelValues[2], labelValues[3], labelValues[4], labelValues[5]).Set(1) | ||
|
|
||
| // First 3 pings fail (metric gets unregistered), then 3 pings succeed (metric gets re-registered) | ||
| pingErr := errors.New("connection refused") | ||
| for i := 0; i < testThreshold; i++ { | ||
| mock.ExpectPing().WillReturnError(pingErr) | ||
| } | ||
| for i := 0; i < testThreshold; i++ { | ||
| mock.ExpectPing() | ||
| } | ||
|
|
||
| ctx := context.Background() | ||
| onStopped := func() {} | ||
| config := &ConnectionInfoMonitorConfig{ | ||
| CheckInterval: testCheckInterval, | ||
| ChecksThreshold: testThreshold, | ||
| } | ||
| cancel := RunConnectionInfoMonitor(ctx, db, registry, infoMetric, labelValues, onStopped, config) | ||
| defer cancel() | ||
|
|
||
| // Wait for 6 tick intervals so we get 3 failures (unregister) then 3 successes (re-register). | ||
| time.Sleep(testCheckInterval*time.Duration(testThreshold*2) + 100*time.Millisecond) | ||
|
|
||
| // All 6 pings must have been consumed (3 fail + 3 success), proving the monitor ran the re-register path. | ||
| require.NoError(t, mock.ExpectationsWereMet(), "monitor should have performed 3 failing then 3 successful pings") | ||
|
|
||
| // Verify re-registration: the monitor runs MustRegister and Set(1) after 3 consecutive successes. | ||
| // ExpectationsWereMet above confirms all 6 pings ran (3 fail + 3 success). If the registry | ||
| // exposes the re-registered metric, assert its value. | ||
| metrics, err := registry.Gather() | ||
| require.NoError(t, err) | ||
| var mf *dto.MetricFamily | ||
| for _, m := range metrics { | ||
| if m.GetName() == "database_observability_connection_info" { | ||
| mf = m | ||
| break | ||
| } | ||
| } | ||
| if mf != nil { | ||
| require.Len(t, mf.Metric, 1, "metric should have one series when present") | ||
| require.Equal(t, float64(1), mf.Metric[0].GetGauge().GetValue()) | ||
| } | ||
| } | ||
|
|
||
| func TestRunConnectionInfoMonitor_MetricRemainsRegisteredWhilePingsSucceed(t *testing.T) { | ||
| defer goleak.VerifyNone(t) | ||
|
|
||
| db, mock, err := sqlmock.New(sqlmock.MonitorPingsOption(true)) | ||
| require.NoError(t, err) | ||
| defer db.Close() | ||
|
|
||
| registry := prometheus.NewRegistry() | ||
| infoMetric := prometheus.NewGaugeVec(prometheus.GaugeOpts{ | ||
| Namespace: "database_observability", | ||
| Name: "connection_info", | ||
| Help: "Information about the connection", | ||
| }, []string{"provider_name", "provider_region", "provider_account", "db_instance_identifier", "engine", "engine_version"}) | ||
| require.NoError(t, registry.Register(infoMetric)) | ||
|
|
||
| labelValues := []string{"unknown", "unknown", "unknown", "unknown", "postgres", "15.0"} | ||
| infoMetric.WithLabelValues(labelValues[0], labelValues[1], labelValues[2], labelValues[3], labelValues[4], labelValues[5]).Set(1) | ||
|
|
||
| // All pings succeed (allow at least 4 successful pings) | ||
| for i := 0; i < 4; i++ { | ||
| mock.ExpectPing() | ||
| } | ||
|
|
||
| ctx := context.Background() | ||
| onStopped := func() {} | ||
| config := &ConnectionInfoMonitorConfig{ | ||
| CheckInterval: testCheckInterval, | ||
| ChecksThreshold: testThreshold, | ||
| } | ||
| cancel := RunConnectionInfoMonitor(ctx, db, registry, infoMetric, labelValues, onStopped, config) | ||
| defer cancel() | ||
|
|
||
| // Wait for a few tick intervals | ||
| time.Sleep(testCheckInterval*4 + 20*time.Millisecond) | ||
|
|
||
| // Metric should still be registered with value 1 | ||
| metrics, err := registry.Gather() | ||
| require.NoError(t, err) | ||
| var mf *dto.MetricFamily | ||
| for _, m := range metrics { | ||
| if m.GetName() == "database_observability_connection_info" { | ||
| mf = m | ||
| break | ||
| } | ||
| } | ||
| require.NotNil(t, mf, "metric should remain registered while pings succeed") | ||
| require.Len(t, mf.Metric, 1) | ||
| require.Equal(t, float64(1), mf.Metric[0].GetGauge().GetValue()) | ||
|
|
||
| require.NoError(t, mock.ExpectationsWereMet()) | ||
| } | ||
|
|
||
| func TestRunConnectionInfoMonitor_CancelStopsGoroutine(t *testing.T) { | ||
| defer goleak.VerifyNone(t) | ||
|
|
||
| db, mock, err := sqlmock.New(sqlmock.MonitorPingsOption(true)) | ||
| require.NoError(t, err) | ||
| defer db.Close() | ||
|
|
||
| registry := prometheus.NewRegistry() | ||
| infoMetric := prometheus.NewGaugeVec(prometheus.GaugeOpts{ | ||
| Namespace: "database_observability", | ||
| Name: "connection_info", | ||
| Help: "Information about the connection", | ||
| }, []string{"provider_name", "provider_region", "provider_account", "db_instance_identifier", "engine", "engine_version"}) | ||
| require.NoError(t, registry.Register(infoMetric)) | ||
|
|
||
| labelValues := []string{"a", "b", "c", "d", "e", "f"} | ||
| infoMetric.WithLabelValues(labelValues[0], labelValues[1], labelValues[2], labelValues[3], labelValues[4], labelValues[5]).Set(1) | ||
|
|
||
| mock.ExpectPing() // at most one ping before we cancel | ||
|
|
||
| ctx := context.Background() | ||
| stopped := make(chan struct{}) | ||
| onStopped := func() { close(stopped) } | ||
| config := &ConnectionInfoMonitorConfig{ | ||
| CheckInterval: testCheckInterval, | ||
| ChecksThreshold: testThreshold, | ||
| } | ||
| cancel := RunConnectionInfoMonitor(ctx, db, registry, infoMetric, labelValues, onStopped, config) | ||
|
|
||
| // Cancel immediately; onStopped should be called when the goroutine exits | ||
| cancel() | ||
| select { | ||
| case <-stopped: | ||
| // goroutine exited | ||
| case <-time.After(2 * time.Second): | ||
| t.Fatal("onStopped was not called after cancel") | ||
| } | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Could we use the component.go existing goroutine instead of starting a new one? I think it would make it cleaner.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Hmm yeah I'm wondering if it's easier to build this on top of the reconnection logic. Today we keep the collectors in "running" state even if the database connection is lost. Maybe we should just "stop()" collectors until database connection retries succeed?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
It makes sense, but I wonder if it may cause unnecessary stop/starts in a flapping db connection.
We would also need to separate the logs collector from the pool, since it is independent from the db connection, once it is started. Right now it only count errors, but once we implement detailed error scraping it will be useful to troubleshoot a flapping db.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Refactored this to (mostly) reuse the existing reconnect logic.
The existing logic seems to only be used when the db is unavailable on first startup, so there is an additional check now to ensure that the DB remains connected, and stops the connection_info collector if the DB becomes unavailable.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I'm a bit confused though by the various tickets though. So, to recap, it seems we have:
connection_infowithI understand we want to keep those activities separate, but I'm wondering if we should just put the new logic inside the
connection_infocollector? i.e. that collector would have its own loop like other collectorsThere was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
The only real reason for having two tickers is that they're on different cadences (30s and 60s).
We could probably combine them, and checking for DB connectivity every 30s is probably not a big deal. 🤔