From 771083b3ed2db24f2c3b51f5c38d0511a6faf518 Mon Sep 17 00:00:00 2001 From: Alex Demidoff Date: Thu, 27 Aug 2026 10:43:47 +0300 Subject: [PATCH 1/4] PMM-14807 Enable the global connection pool by default The exporter created a fresh mongo.Client for every scrape and disconnected it immediately afterwards, so each scrape cost the monitored mongod a TCP connect, a TLS handshake where configured, and a SCRAM authentication -- work paid for by the database, not by the exporter. --mongodb.global-conn-pool already avoided that, but it defaulted to off and was not reachable through pmm-admin, so nobody got the benefit. Default it to on; --no-mongodb.global-conn-pool restores per-scrape connections. Measured on a single-node replica set with SCRAM auth, 200 scrapes with diagnosticdata, dbstats, collstats, indexstats, top, currentop, replset status and config enabled: pool off pool on connections created 605 4 SCRAM-SHA-256 handshakes 408 6 mongod CPU per scrape 11.2 ms 4.8 ms metric names exposed 5277 5277 Also stop pinning a disconnected client in the cache. Ping failures were returned to the caller with the client left in place, which is right for a transient error -- the driver reconnects the pool on its own, and tearing it down would add churn while MongoDB is already struggling -- but a client that returns ErrClientDisconnected never recovers, and every later scrape would fail with it. Drop that one so the next scrape builds a new client. --- REFERENCE.md | 2 +- exporter/exporter.go | 50 +++++++++++++++++----------- exporter/exporter_test.go | 69 +++++++++++++++++++++++++++++++++++++++ main.go | 26 +++++++-------- main_test.go | 28 ++++++++++++++++ 5 files changed, 142 insertions(+), 33 deletions(-) diff --git a/REFERENCE.md b/REFERENCE.md index 38c2eea61..90c3ab356 100644 --- a/REFERENCE.md +++ b/REFERENCE.md @@ -9,7 +9,7 @@ | --mongodb.collstats-colls | List of comma separared databases.collections to get $collStats | --mongodb.collstats-colls=db1,db2.col2 | | --mongodb.indexstats-colls | List of comma separared databases.collections to get $indexStats | --mongodb.indexstats-colls=db1.col1,db2.col2 | | --[no-]mongodb.direct-connect | Whether or not a direct connect should be made. Direct connections are not valid if multiple hosts are specified or an SRV URI is used | | -| --[no-]mongodb.global-conn-pool | Use global connection pool instead of creating new pool for each http request | | +| --[no-]mongodb.global-conn-pool | Use global connection pool instead of creating new pool for each http request. Enabled by default | --no-mongodb.global-conn-pool | | --mongodb.uri | MongoDB connection URI ($MONGODB_URI) | --mongodb.uri=mongodb://user:pass@127.0.0.1:27017/admin?ssl=true | | --split-cluster | Whether to treat cluster members from the connection URI as separate targets | | --web.listen-address | Address to listen on for web interface and telemetry | --web.listen-address=":9216" | diff --git a/exporter/exporter.go b/exporter/exporter.go index ce7dabd01..77372ec31 100644 --- a/exporter/exporter.go +++ b/exporter/exporter.go @@ -18,6 +18,7 @@ package exporter import ( "context" + "errors" "fmt" "log/slog" "net/http" @@ -268,36 +269,47 @@ func (e *Exporter) makeRegistry(ctx context.Context, client *mongo.Client, topol } func (e *Exporter) getClient(ctx context.Context) (*mongo.Client, error) { - if e.opts.GlobalConnPool { - // Get global client. Maybe it must be initialized first. - // Initialization is retried with every scrape until it succeeds once. - e.clientMu.Lock() - defer e.clientMu.Unlock() - - // If client is already initialized, and Ping is successful -- return it. - if e.client != nil { - err := e.client.Ping(ctx, nil) - if err != nil { - return nil, fmt.Errorf("cannot connect to MongoDB: %w", err) - } + if !e.opts.GlobalConnPool { + // Create a new client for every scrape. The caller disconnects it. + client, err := connect(ctx, e.opts) + if err != nil { + return nil, err + } + + return client, nil + } + + // Get global client. Maybe it must be initialized first. + // Initialization is retried with every scrape until it succeeds once. + e.clientMu.Lock() + defer e.clientMu.Unlock() + // If client is already initialized, and Ping is successful -- return it. + if e.client != nil { + err := e.client.Ping(ctx, nil) + if err == nil { return e.client, nil } - client, err := connect(context.Background(), e.opts) - if err != nil { - return nil, err + // A disconnected client never recovers, so drop it and let the next scrape build + // a new one. Every other error is transient -- an unreachable server, a scrape + // that ran out of time -- and the driver reconnects the pool on its own. Tearing + // it down would only add connection churn while MongoDB is already struggling. + if errors.Is(err, mongo.ErrClientDisconnected) { + e.logger.Warn("Dropping disconnected MongoDB client, reconnecting on next scrape") + e.client = nil } - e.client = client - return client, nil + return nil, fmt.Errorf("cannot connect to MongoDB: %w", err) } - // !e.opts.GlobalConnPool: create new client for every scrape. - client, err := connect(ctx, e.opts) + // The pooled client outlives the scrape that happens to create it, so it must not + // inherit that scrape's context: cancelling the request would close the shared pool. + client, err := connect(context.Background(), e.opts) //nolint:contextcheck if err != nil { return nil, err } + e.client = client return client, nil } diff --git a/exporter/exporter_test.go b/exporter/exporter_test.go index 4f0202107..e76392c78 100644 --- a/exporter/exporter_test.go +++ b/exporter/exporter_test.go @@ -256,6 +256,75 @@ func TestConnect(t *testing.T) { }) } +// newPooledExporter builds an exporter that reuses one client. It skips New, whose +// background initial connect would race with tests that drive getClient themselves. +func newPooledExporter(t *testing.T) *Exporter { + t.Helper() + + log := promslog.New(&promslog.Config{}) //nolint:exhaustruct_v5 + + opts := &Opts{ //nolint:exhaustruct_v5 + Logger: log, + URI: fmt.Sprintf("mongodb://127.0.0.1:%s/admin", tu.MongoDBS1PrimaryPort), + GlobalConnPool: true, + DirectConnect: true, + } + + return &Exporter{ //nolint:exhaustruct_v5 + logger: log, + opts: opts, + lock: &sync.Mutex{}, + totalCollectionsCount: -1, + } +} + +func TestGlobalConnPoolReplacesDisconnectedClient(t *testing.T) { + t.Parallel() + + ctx := t.Context() + e := newPooledExporter(t) + + first, err := e.getClient(ctx) + require.NoError(t, err) + require.NotNil(t, first) + + // Make the cached client permanently unusable. Every later Ping on it fails with + // ErrClientDisconnected, no matter how healthy the server is. + require.NoError(t, first.Disconnect(ctx)) + + // The scrape that discovers it still fails, but it must not leave the dead client + // cached, or every later scrape would fail with it too. + _, err = e.getClient(ctx) + require.Error(t, err) + require.Nil(t, e.client, "disconnected client stayed cached") + + second, err := e.getClient(ctx) + require.NoError(t, err) + assert.NotSame(t, first, second) + assert.NoError(t, second.Ping(ctx, nil)) + + require.NoError(t, second.Disconnect(ctx)) +} + +func TestGlobalConnPoolKeepsClientOnTransientError(t *testing.T) { + t.Parallel() + + e := newPooledExporter(t) + + first, err := e.getClient(t.Context()) + require.NoError(t, err) + t.Cleanup(func() { _ = first.Disconnect(context.Background()) }) + + // A scrape that runs out of time must not cost us the pool: the client is healthy, + // only this request is over. + expired, cancel := context.WithCancel(t.Context()) + cancel() + + _, err = e.getClient(expired) + require.Error(t, err) + assert.Same(t, first, e.client, "healthy client was dropped after a transient error") +} + // How this test works? // When connected to a MongoS instance, the makeRegistry method should skip // adding replSetGetStatusCollector. To test that, we try to unregister a diff --git a/main.go b/main.go index b327cab7c..99d415ab7 100644 --- a/main.go +++ b/main.go @@ -40,19 +40,19 @@ var ( // GlobalFlags has command line flags to configure the exporter. type GlobalFlags struct { - User string `env:"MONGODB_USER" help:"monitor user, need clusterMonitor role in admin db and read role in local db" name:"mongodb.user" placeholder:"monitorUser"` - Password string `env:"MONGODB_PASSWORD" help:"monitor user password" name:"mongodb.password" placeholder:"monitorPassword"` - CollStatsNamespaces string `help:"List of comma separared databases.collections to get $collStats" name:"mongodb.collstats-colls" placeholder:"db1,db2.col2"` - IndexStatsCollections string `help:"List of comma separared databases.collections to get $indexStats" name:"mongodb.indexstats-colls" placeholder:"db1.col1,db2.col2"` - URI []string `env:"MONGODB_URI" help:"MongoDB connection URI" name:"mongodb.uri" placeholder:"mongodb://user:pass@127.0.0.1:27017/admin?ssl=true"` - GlobalConnPool bool `help:"Use global connection pool instead of creating new pool for each http request." name:"mongodb.global-conn-pool" negatable:""` - DirectConnect bool `default:"true" help:"Whether or not a direct connect should be made. Direct connections are not valid if multiple hosts are specified or an SRV URI is used." name:"mongodb.direct-connect" negatable:""` - WebListenAddress string `default:":9216" help:"Address to listen on for web interface and telemetry" name:"web.listen-address"` - WebTelemetryPath string `default:"/metrics" help:"Metrics expose path" name:"web.telemetry-path"` - TLSConfigPath string `help:"Path to the file having Prometheus TLS config for basic auth" name:"web.config"` - TimeoutOffset int `default:"1" help:"Offset to subtract from the request timeout in seconds" name:"web.timeout-offset"` - LogLevel string `default:"error" enum:"debug,info,warn,error,fatal" help:"Only log messages with the given severity or above. Valid levels: [debug, info, warn, error, fatal]" name:"log.level"` - ConnectTimeoutMS int `default:"5000" help:"Connection timeout in milliseconds" name:"mongodb.connect-timeout-ms"` + User string `env:"MONGODB_USER" help:"monitor user, need clusterMonitor role in admin db and read role in local db" name:"mongodb.user" placeholder:"monitorUser"` + Password string `env:"MONGODB_PASSWORD" help:"monitor user password" name:"mongodb.password" placeholder:"monitorPassword"` + CollStatsNamespaces string `help:"List of comma separared databases.collections to get $collStats" name:"mongodb.collstats-colls" placeholder:"db1,db2.col2"` + IndexStatsCollections string `help:"List of comma separared databases.collections to get $indexStats" name:"mongodb.indexstats-colls" placeholder:"db1.col1,db2.col2"` + URI []string `env:"MONGODB_URI" help:"MongoDB connection URI" name:"mongodb.uri" placeholder:"mongodb://user:pass@127.0.0.1:27017/admin?ssl=true"` + GlobalConnPool bool `default:"true" help:"Use global connection pool instead of creating new pool for each http request." name:"mongodb.global-conn-pool" negatable:""` + DirectConnect bool `default:"true" help:"Whether or not a direct connect should be made. Direct connections are not valid if multiple hosts are specified or an SRV URI is used." name:"mongodb.direct-connect" negatable:""` + WebListenAddress string `default:":9216" help:"Address to listen on for web interface and telemetry" name:"web.listen-address"` + WebTelemetryPath string `default:"/metrics" help:"Metrics expose path" name:"web.telemetry-path"` + TLSConfigPath string `help:"Path to the file having Prometheus TLS config for basic auth" name:"web.config"` + TimeoutOffset int `default:"1" help:"Offset to subtract from the request timeout in seconds" name:"web.timeout-offset"` + LogLevel string `default:"error" enum:"debug,info,warn,error,fatal" help:"Only log messages with the given severity or above. Valid levels: [debug, info, warn, error, fatal]" name:"log.level"` + ConnectTimeoutMS int `default:"5000" help:"Connection timeout in milliseconds" name:"mongodb.connect-timeout-ms"` EnableExporterMetrics bool `default:"True" help:"Enable collecting metrics about the exporter itself (process_*, go_*)" name:"collector.exporter-metrics" negatable:""` EnableDiagnosticData bool `help:"Enable collecting metrics from getDiagnosticData" name:"collector.diagnosticdata"` diff --git a/main_test.go b/main_test.go index 3d54f866e..aec284edd 100644 --- a/main_test.go +++ b/main_test.go @@ -20,9 +20,11 @@ import ( "strings" "testing" + "github.com/alecthomas/kong" "github.com/foxcpp/go-mockdns" "github.com/prometheus/common/promslog" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" "github.com/percona/mongodb_exporter/internal/tu" ) @@ -100,6 +102,32 @@ func TestSplitCluster(t *testing.T) { } } +func TestGlobalConnPoolFlagDefault(t *testing.T) { + t.Parallel() + + tests := map[string]struct { + args []string + want bool + }{ + "default": {args: []string{}, want: true}, + "negated": {args: []string{"--no-mongodb.global-conn-pool"}, want: false}, + "explicit": {args: []string{"--mongodb.global-conn-pool"}, want: true}, + } + + for name, test := range tests { + t.Run(name, func(t *testing.T) { + t.Parallel() + var opts GlobalFlags + parser, err := kong.New(&opts, kong.Vars{"version": ""}) + require.NoError(t, err) + + _, err = parser.Parse(append(test.args, "--mongodb.uri=mongodb://127.0.0.1:27017")) + require.NoError(t, err) + assert.Equal(t, test.want, opts.GlobalConnPool) + }) + } +} + func TestBuildExporter(t *testing.T) { t.Parallel() opts := GlobalFlags{ From e81efa0e88a7a046cfa6337ef5c57780adcf6c77 Mon Sep 17 00:00:00 2001 From: Alex Demidoff Date: Thu, 27 Aug 2026 11:27:25 +0300 Subject: [PATCH 2/4] Bound the pooled client's initial connect by the scrape context connect pings, so on an unreachable server context.Background() let that ping run for the whole server-selection timeout (5s by default) while holding clientMu. A scrape with a shorter budget then overran it and Prometheus got nothing at all, rather than mongodb_up 0 -- the same failure mode the disconnected-client handling above deliberately avoids. The comment claiming the scrape context would close the shared pool was wrong. Topology.Connect takes no context and mongo.Connect does not retain the one it is given, so the pooled client outlives the scrape that created it either way. Verified: a client still serves pings and queries after the context that created it is cancelled. Both invariants are now covered by tests. --- exporter/exporter.go | 9 +++++--- exporter/exporter_test.go | 45 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 51 insertions(+), 3 deletions(-) diff --git a/exporter/exporter.go b/exporter/exporter.go index 77372ec31..f0acaf5b5 100644 --- a/exporter/exporter.go +++ b/exporter/exporter.go @@ -303,9 +303,12 @@ func (e *Exporter) getClient(ctx context.Context) (*mongo.Client, error) { return nil, fmt.Errorf("cannot connect to MongoDB: %w", err) } - // The pooled client outlives the scrape that happens to create it, so it must not - // inherit that scrape's context: cancelling the request would close the shared pool. - client, err := connect(context.Background(), e.opts) //nolint:contextcheck + // The scrape context bounds this attempt: connect pings, and on an unreachable server + // that ping would otherwise run for the whole server-selection timeout while holding + // clientMu, overrunning the scrape budget so Prometheus gets nothing at all instead of + // mongodb_up 0. The pooled client still outlives the scrape that created it -- the + // driver connects the topology without a context and does not retain this one. + client, err := connect(ctx, e.opts) if err != nil { return nil, err } diff --git a/exporter/exporter_test.go b/exporter/exporter_test.go index e76392c78..757cff833 100644 --- a/exporter/exporter_test.go +++ b/exporter/exporter_test.go @@ -27,6 +27,7 @@ import ( "strings" "sync" "testing" + "time" "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus/testutil" @@ -306,6 +307,50 @@ func TestGlobalConnPoolReplacesDisconnectedClient(t *testing.T) { require.NoError(t, second.Disconnect(ctx)) } +// The pooled client is built during whichever scrape happens to find the cache empty, but +// it has to outlive that scrape. This holds because the driver connects the topology +// without a context and does not retain the one passed to mongo.Connect. +func TestGlobalConnPoolClientOutlivesCreatingScrape(t *testing.T) { + t.Parallel() + + e := newPooledExporter(t) + + scrape, cancel := context.WithCancel(t.Context()) + first, err := e.getClient(scrape) + require.NoError(t, err) + t.Cleanup(func() { _ = first.Disconnect(context.Background()) }) + + // The scrape that created the client ends. + cancel() + + second, err := e.getClient(t.Context()) + require.NoError(t, err) + assert.Same(t, first, second, "pooled client was lost when its creating scrape ended") + assert.NoError(t, second.Ping(t.Context(), nil)) +} + +// An unreachable server must not hold clientMu for the whole server-selection timeout: +// the scrape needs to come back in time to report mongodb_up 0. +func TestGlobalConnPoolInitialConnectHonoursScrapeDeadline(t *testing.T) { + t.Parallel() + + e := newPooledExporter(t) + e.opts.URI = "mongodb://127.0.0.1:1/admin" + e.opts.ConnectTimeoutMS = 5000 + + budget := 500 * time.Millisecond + ctx, cancel := context.WithTimeout(t.Context(), budget) + defer cancel() + + start := time.Now() + _, err := e.getClient(ctx) + elapsed := time.Since(start) + + require.Error(t, err) + assert.Less(t, elapsed, time.Duration(e.opts.ConnectTimeoutMS)*time.Millisecond, + "initial connect ignored the scrape deadline and ran to the server-selection timeout") +} + func TestGlobalConnPoolKeepsClientOnTransientError(t *testing.T) { t.Parallel() From f8ef8b43084f20d1bf147832ebfaa26537be60e7 Mon Sep 17 00:00:00 2001 From: Alex Demidoff Date: Thu, 27 Aug 2026 19:05:07 +0300 Subject: [PATCH 3/4] Keep a scrape's budget while another connect holds the client lock Bounding the initial connect by the scrape context did not fully land: waiting for the mutex guarding the cached client was not bounded by anything. New() connects in the background holding that lock, and with the pool now on by default it takes this path, so a scrape arriving during startup blocked until server selection gave up -- measured at 5.0s against an unreachable server with a 1s budget. Concurrent scrapes serialised the same way while the cache was empty. Guard the client with a capacity-1 channel instead, so taking it selects on ctx.Done() and a scrape gives up with its own budget: 5.0s -> 1.1s. Bound the background connect too. It ran on context.Background(), so a server that never answered held the lock for the process lifetime -- forever when Opts.ConnectTimeoutMS is 0, since connect then passes 0 to SetServerSelectionTimeout and the driver creates no timer at all. Also disconnect the client that background connect creates when the global pool is off. Nothing else ever owned it, so it leaked a topology, its monitoring goroutines and a connection per exporter, times N under --split-cluster. Measured on a live mongod: connections.current +2 after startup before, +0 now. This makes the "caller disconnects it" comment on that branch true. --- exporter/exporter.go | 48 +++++++++++++++++++++++++++++++++------ exporter/exporter_test.go | 26 +++++++++++++++++++++ 2 files changed, 67 insertions(+), 7 deletions(-) diff --git a/exporter/exporter.go b/exporter/exporter.go index f0acaf5b5..4caecb4a5 100644 --- a/exporter/exporter.go +++ b/exporter/exporter.go @@ -37,8 +37,10 @@ import ( // Exporter holds Exporter methods and attributes. type Exporter struct { - client *mongo.Client - clientMu sync.Mutex + client *mongo.Client + // clientLock guards client. It is a channel rather than a sync.Mutex so that waiting + // for it can honour a scrape's context: see lockClient. + clientLock chan struct{} logger *slog.Logger opts *Opts lock *sync.Mutex @@ -95,6 +97,12 @@ var ( const ( defaultCacheSize = 1000 + + // initialConnectTimeout bounds the connect New starts in the background. Without it a + // server that never answers would hold clientLock for the process lifetime -- forever + // if Opts.ConnectTimeoutMS is 0, since connect then asks the driver for no + // server-selection timeout at all -- and every scrape would fail on the lock. + initialConnectTimeout = 30 * time.Second ) // New connects to the database and returns a new Exporter instance. @@ -108,19 +116,36 @@ func New(opts *Opts) *Exporter { opts.Logger = promslog.New(promslogConfig) } - ctx := context.Background() - exp := &Exporter{ logger: opts.Logger, opts: opts, + clientLock: make(chan struct{}, 1), lock: &sync.Mutex{}, totalCollectionsCount: -1, // Not calculated yet. waiting the db connection. } // Try initial connect. Connection will be retried with every scrape. go func() { - _, err := exp.getClient(ctx) + ctx, cancel := context.WithTimeout(context.Background(), initialConnectTimeout) + defer cancel() + + client, err := exp.getClient(ctx) if err != nil { exp.logger.Error("Cannot connect to MongoDB", "error", err) + + return + } + + // With the global pool this client is the one every later scrape reuses. Without it + // the client belongs to nobody, since each scrape builds its own, so leaving it + // connected would leak a topology, its monitoring goroutines and a connection for + // the lifetime of the process. + if exp.opts.GlobalConnPool { + return + } + + err = client.Disconnect(ctx) + if err != nil { + exp.logger.Error("Cannot disconnect client", "error", err) } }() @@ -281,8 +306,17 @@ func (e *Exporter) getClient(ctx context.Context) (*mongo.Client, error) { // Get global client. Maybe it must be initialized first. // Initialization is retried with every scrape until it succeeds once. - e.clientMu.Lock() - defer e.clientMu.Unlock() + // + // Taking the lock honours ctx, which a sync.Mutex could not: whoever holds it may be + // inside a connect -- the one New starts in the background, or another scrape's -- and + // waiting that out would overrun this scrape's budget, so Prometheus would get nothing + // at all instead of mongodb_up 0. + select { + case e.clientLock <- struct{}{}: + case <-ctx.Done(): + return nil, fmt.Errorf("cannot connect to MongoDB: %w", ctx.Err()) + } + defer func() { <-e.clientLock }() // If client is already initialized, and Ping is successful -- return it. if e.client != nil { diff --git a/exporter/exporter_test.go b/exporter/exporter_test.go index 757cff833..54ad4e618 100644 --- a/exporter/exporter_test.go +++ b/exporter/exporter_test.go @@ -274,6 +274,7 @@ func newPooledExporter(t *testing.T) *Exporter { return &Exporter{ //nolint:exhaustruct_v5 logger: log, opts: opts, + clientLock: make(chan struct{}, 1), lock: &sync.Mutex{}, totalCollectionsCount: -1, } @@ -351,6 +352,31 @@ func TestGlobalConnPoolInitialConnectHonoursScrapeDeadline(t *testing.T) { "initial connect ignored the scrape deadline and ran to the server-selection timeout") } +// A scrape must not wait out somebody else's connect -- the background connect New starts, +// or another scrape's -- because waiting on the lock is not covered by the connect's own +// context bound. +func TestGlobalConnPoolScrapeGivesUpOnHeldLock(t *testing.T) { + t.Parallel() + + e := newPooledExporter(t) + + // Stand in for a connect in progress: the lock is held and e.client is still nil. + e.clientLock <- struct{}{} + t.Cleanup(func() { <-e.clientLock }) + + budget := 300 * time.Millisecond + ctx, cancel := context.WithTimeout(t.Context(), budget) + defer cancel() + + start := time.Now() + _, err := e.getClient(ctx) + elapsed := time.Since(start) + + require.Error(t, err) + assert.Less(t, elapsed, initialConnectTimeout, + "scrape blocked on the client lock instead of giving up with its budget") +} + func TestGlobalConnPoolKeepsClientOnTransientError(t *testing.T) { t.Parallel() From 998ec66df32b6a23281efabc7cb9e478312c026d Mon Sep 17 00:00:00 2001 From: Alex Demidoff Date: Thu, 27 Aug 2026 21:33:40 +0300 Subject: [PATCH 4/4] Keep the pooled client's health check off the exclusive lock Two problems with guarding the cached client by a lock held across network I/O, both introduced by making the pool the default. The Ping ran under an exclusive lock, so concurrent scrapes of one target -- PMM scrapes each exporter at three resolutions with different collect[] sets -- had to queue behind each other, letting one slow scrape push the next past its budget and report mongodb_up 0 for a healthy client. The lock now guards the client pointer only and is never held across a command. Bounding the connect by the scrape context fixed one failure and created another: a scrape budget shorter than one connect cancelled every attempt, so the pool stayed empty and every scrape kept paying for a connect that could never finish. The connect now gets its own budget, taken from ConnectTimeoutMS rather than from the caller, concurrent attempts collapse onto it, and each caller gives up on its own deadline while it carries on in the background. That also removes the hardcoded startup bound, which truncated the connect at 30s for an operator who had asked for more via --mongodb.connect-timeout-ms. The ConnectTimeoutMS=0 case is still handled: connect would pass 0 to SetServerSelectionTimeout, which the driver reads as no timeout at all, so the budget falls back to defaultConnectTimeout. Both tests fail against the previous shape. Also state the flipped default in the --help text, since kong does not print defaults for bool flags. --- exporter/exporter.go | 127 ++++++++++++++++++++++---------------- exporter/exporter_test.go | 112 +++++++++++++++++++++++++-------- go.mod | 2 +- main.go | 2 +- 4 files changed, 162 insertions(+), 81 deletions(-) diff --git a/exporter/exporter.go b/exporter/exporter.go index 4caecb4a5..179a0f33a 100644 --- a/exporter/exporter.go +++ b/exporter/exporter.go @@ -31,6 +31,7 @@ import ( "github.com/prometheus/client_golang/prometheus/promhttp" "github.com/prometheus/common/promslog" "go.mongodb.org/mongo-driver/mongo" + "golang.org/x/sync/singleflight" "github.com/percona/mongodb_exporter/exporter/dsn_fix" ) @@ -38,9 +39,11 @@ import ( // Exporter holds Exporter methods and attributes. type Exporter struct { client *mongo.Client - // clientLock guards client. It is a channel rather than a sync.Mutex so that waiting - // for it can honour a scrape's context: see lockClient. - clientLock chan struct{} + // clientMu guards the client pointer only. It is never held across a command, so + // concurrent scrapes of this target do not wait for each other. + clientMu sync.RWMutex + // clientGroup collapses concurrent attempts to build client into one connect. + clientGroup singleflight.Group logger *slog.Logger opts *Opts lock *sync.Mutex @@ -98,11 +101,11 @@ var ( const ( defaultCacheSize = 1000 - // initialConnectTimeout bounds the connect New starts in the background. Without it a - // server that never answers would hold clientLock for the process lifetime -- forever - // if Opts.ConnectTimeoutMS is 0, since connect then asks the driver for no - // server-selection timeout at all -- and every scrape would fail on the lock. - initialConnectTimeout = 30 * time.Second + // defaultConnectTimeout bounds building the pooled client when Opts.ConnectTimeoutMS is + // unset. connect would otherwise hand the driver a server-selection timeout of 0, which + // it reads as no timeout at all, and an unanswering server would block a connect for the + // lifetime of the process. + defaultConnectTimeout = 30 * time.Second ) // New connects to the database and returns a new Exporter instance. @@ -119,14 +122,13 @@ func New(opts *Opts) *Exporter { exp := &Exporter{ logger: opts.Logger, opts: opts, - clientLock: make(chan struct{}, 1), lock: &sync.Mutex{}, totalCollectionsCount: -1, // Not calculated yet. waiting the db connection. } // Try initial connect. Connection will be retried with every scrape. + // getClient bounds the connect itself, so no deadline is imposed here. go func() { - ctx, cancel := context.WithTimeout(context.Background(), initialConnectTimeout) - defer cancel() + ctx := context.Background() client, err := exp.getClient(ctx) if err != nil { @@ -135,10 +137,9 @@ func New(opts *Opts) *Exporter { return } - // With the global pool this client is the one every later scrape reuses. Without it - // the client belongs to nobody, since each scrape builds its own, so leaving it - // connected would leak a topology, its monitoring goroutines and a connection for - // the lifetime of the process. + // With the global pool this client is the one every later scrape reuses. Otherwise + // nothing owns it, since each scrape builds its own, so leaving it connected would + // leak a topology, its monitoring goroutines and a connection. if exp.opts.GlobalConnPool { return } @@ -293,62 +294,82 @@ func (e *Exporter) makeRegistry(ctx context.Context, client *mongo.Client, topol return registry } +// getClient returns a client to scrape with. Everything below keeps to one rule: a scrape +// must come back within its own budget, so that a struggling MongoDB is reported as +// mongodb_up 0 rather than as a scrape Prometheus never gets an answer to. func (e *Exporter) getClient(ctx context.Context) (*mongo.Client, error) { if !e.opts.GlobalConnPool { // Create a new client for every scrape. The caller disconnects it. - client, err := connect(ctx, e.opts) - if err != nil { - return nil, err - } - - return client, nil + return connect(ctx, e.opts) } - // Get global client. Maybe it must be initialized first. - // Initialization is retried with every scrape until it succeeds once. - // - // Taking the lock honours ctx, which a sync.Mutex could not: whoever holds it may be - // inside a connect -- the one New starts in the background, or another scrape's -- and - // waiting that out would overrun this scrape's budget, so Prometheus would get nothing - // at all instead of mongodb_up 0. - select { - case e.clientLock <- struct{}{}: - case <-ctx.Done(): - return nil, fmt.Errorf("cannot connect to MongoDB: %w", ctx.Err()) - } - defer func() { <-e.clientLock }() + e.clientMu.RLock() + client := e.client + e.clientMu.RUnlock() - // If client is already initialized, and Ping is successful -- return it. - if e.client != nil { - err := e.client.Ping(ctx, nil) + // Health-check outside the lock. Holding it across the Ping would make concurrent + // scrapes of this target queue behind each other, letting one slow scrape push the + // next past its budget. + if client != nil { + err := client.Ping(ctx, nil) if err == nil { - return e.client, nil + return client, nil } - // A disconnected client never recovers, so drop it and let the next scrape build - // a new one. Every other error is transient -- an unreachable server, a scrape - // that ran out of time -- and the driver reconnects the pool on its own. Tearing - // it down would only add connection churn while MongoDB is already struggling. + // A disconnected client never recovers, so forget it and let the next scrape build a + // new one. Every other error is transient -- an unreachable server, a scrape that ran + // out of time -- and the driver reconnects the pool on its own; tearing it down would + // only add churn while MongoDB is already struggling. if errors.Is(err, mongo.ErrClientDisconnected) { e.logger.Warn("Dropping disconnected MongoDB client, reconnecting on next scrape") - e.client = nil + e.clientMu.Lock() + if e.client == client { + e.client = nil + } + e.clientMu.Unlock() } return nil, fmt.Errorf("cannot connect to MongoDB: %w", err) } - // The scrape context bounds this attempt: connect pings, and on an unreachable server - // that ping would otherwise run for the whole server-selection timeout while holding - // clientMu, overrunning the scrape budget so Prometheus gets nothing at all instead of - // mongodb_up 0. The pooled client still outlives the scrape that created it -- the - // driver connects the topology without a context and does not retain this one. - client, err := connect(ctx, e.opts) - if err != nil { - return nil, err + // Build the client. Initialization is retried with every scrape until it succeeds once. + // + // The connect gets its own budget rather than the scrape's: a scrape shorter than one + // connect would cancel every attempt and leave the pool permanently empty, so every + // scrape would keep paying for a connect that can never finish. Concurrent scrapes + // collapse onto that one connect, and each gives up on its own deadline while it + // carries on in the background. + connectTimeout := time.Duration(e.opts.ConnectTimeoutMS) * time.Millisecond + if connectTimeout <= 0 { + connectTimeout = defaultConnectTimeout } - e.client = client - return client, nil + built := e.clientGroup.DoChan("", func() (any, error) { //nolint:contextcheck + connectCtx, cancel := context.WithTimeout(context.Background(), connectTimeout) + defer cancel() + + newClient, err := connect(connectCtx, e.opts) + if err != nil { + return nil, err + } + + e.clientMu.Lock() + e.client = newClient + e.clientMu.Unlock() + + return newClient, nil + }) + + select { + case res := <-built: + if res.Err != nil { + return nil, res.Err + } + + return res.Val.(*mongo.Client), nil //nolint:forcetypeassert + case <-ctx.Done(): + return nil, fmt.Errorf("cannot connect to MongoDB: %w", ctx.Err()) + } } // Handler returns an http.Handler that serves metrics. Can be used instead of diff --git a/exporter/exporter_test.go b/exporter/exporter_test.go index 54ad4e618..33542713a 100644 --- a/exporter/exporter_test.go +++ b/exporter/exporter_test.go @@ -262,24 +262,55 @@ func TestConnect(t *testing.T) { func newPooledExporter(t *testing.T) *Exporter { t.Helper() - log := promslog.New(&promslog.Config{}) //nolint:exhaustruct_v5 + log := promslog.New(&promslog.Config{}) - opts := &Opts{ //nolint:exhaustruct_v5 + opts := &Opts{ Logger: log, URI: fmt.Sprintf("mongodb://127.0.0.1:%s/admin", tu.MongoDBS1PrimaryPort), GlobalConnPool: true, DirectConnect: true, } - return &Exporter{ //nolint:exhaustruct_v5 + return &Exporter{ logger: log, opts: opts, - clientLock: make(chan struct{}, 1), lock: &sync.Mutex{}, totalCollectionsCount: -1, } } +// blackHoleMongo returns the address of a listener that accepts connections and never +// answers, so a driver handshake against it blocks until its own timeout rather than +// failing fast the way a closed port would. The returned channel is closed once the first +// connection has been accepted, which is when a connect is provably in flight. +func blackHoleMongo(t *testing.T) (string, <-chan struct{}) { + t.Helper() + + var listenCfg net.ListenConfig + listener, err := listenCfg.Listen(t.Context(), "tcp", "127.0.0.1:0") + require.NoError(t, err) + t.Cleanup(func() { _ = listener.Close() }) + + dialed := make(chan struct{}) + go func() { + first := true + for { + conn, err := listener.Accept() + if err != nil { + return + } + if first { + close(dialed) + first = false + } + // Hold the connection open and stay silent. The process end closes it. + _ = conn + } + }() + + return listener.Addr().String(), dialed +} + func TestGlobalConnPoolReplacesDisconnectedClient(t *testing.T) { t.Parallel() @@ -330,39 +361,62 @@ func TestGlobalConnPoolClientOutlivesCreatingScrape(t *testing.T) { assert.NoError(t, second.Ping(t.Context(), nil)) } -// An unreachable server must not hold clientMu for the whole server-selection timeout: -// the scrape needs to come back in time to report mongodb_up 0. -func TestGlobalConnPoolInitialConnectHonoursScrapeDeadline(t *testing.T) { +// Giving up on the scrape budget must not cancel the connect: a budget shorter than one +// connect would otherwise leave the pool permanently empty, so every scrape would keep +// paying for an attempt that can never finish. +func TestGlobalConnPoolCacheWarmsAfterScrapeGivesUp(t *testing.T) { t.Parallel() e := newPooledExporter(t) - e.opts.URI = "mongodb://127.0.0.1:1/admin" - e.opts.ConnectTimeoutMS = 5000 + t.Cleanup(func() { + e.clientMu.RLock() + defer e.clientMu.RUnlock() - budget := 500 * time.Millisecond - ctx, cancel := context.WithTimeout(t.Context(), budget) - defer cancel() + if e.client != nil { + _ = e.client.Disconnect(context.Background()) + } + }) - start := time.Now() - _, err := e.getClient(ctx) - elapsed := time.Since(start) + // No budget at all, so the scrape cannot wait for the connect it starts. + ctx, cancel := context.WithCancel(t.Context()) + cancel() + _, err := e.getClient(ctx) require.Error(t, err) - assert.Less(t, elapsed, time.Duration(e.opts.ConnectTimeoutMS)*time.Millisecond, - "initial connect ignored the scrape deadline and ran to the server-selection timeout") + + require.Eventually(t, func() bool { + e.clientMu.RLock() + defer e.clientMu.RUnlock() + + return e.client != nil + }, 15*time.Second, 50*time.Millisecond, + "the connect was cancelled along with the scrape, leaving the pool empty") } -// A scrape must not wait out somebody else's connect -- the background connect New starts, -// or another scrape's -- because waiting on the lock is not covered by the connect's own -// context bound. -func TestGlobalConnPoolScrapeGivesUpOnHeldLock(t *testing.T) { +// A scrape must not wait out a connect somebody else started -- the one New runs in the +// background, or another scrape's -- even though that connect is deliberately given a +// budget of its own, longer than any single scrape's. +func TestGlobalConnPoolScrapeGivesUpOnConnectInFlight(t *testing.T) { t.Parallel() e := newPooledExporter(t) + addr, dialed := blackHoleMongo(t) + e.opts.URI = "mongodb://" + addr + "/admin" + e.opts.ConnectTimeoutMS = 3000 + + inFlight := make(chan struct{}) + go func() { + defer close(inFlight) + _, _ = e.getClient(context.Background()) + }() - // Stand in for a connect in progress: the lock is held and e.client is still nil. - e.clientLock <- struct{}{} - t.Cleanup(func() { <-e.clientLock }) + // Only once the listener has accepted is a connect provably under way. Without this the + // scrape below could win the race and simply do its own connect. + select { + case <-dialed: + case <-time.After(10 * time.Second): + t.Fatal("the background connect never dialled") + } budget := 300 * time.Millisecond ctx, cancel := context.WithTimeout(t.Context(), budget) @@ -373,8 +427,14 @@ func TestGlobalConnPoolScrapeGivesUpOnHeldLock(t *testing.T) { elapsed := time.Since(start) require.Error(t, err) - assert.Less(t, elapsed, initialConnectTimeout, - "scrape blocked on the client lock instead of giving up with its budget") + assert.Less(t, elapsed, 4*budget, + "scrape waited for the in-flight connect instead of giving up on its budget") + + select { + case <-inFlight: + case <-time.After(time.Duration(e.opts.ConnectTimeoutMS) * time.Millisecond * 2): + t.Fatal("the background connect never returned") + } } func TestGlobalConnPoolKeepsClientOnTransientError(t *testing.T) { diff --git a/go.mod b/go.mod index 0400d5203..1843d5d4b 100644 --- a/go.mod +++ b/go.mod @@ -20,6 +20,7 @@ require github.com/foxcpp/go-mockdns v1.2.0 require ( github.com/hashicorp/go-version v1.9.0 github.com/percona/percona-backup-mongodb v1.8.1-0.20251124214042-d06cab743541 + golang.org/x/sync v0.22.0 ) require ( @@ -133,7 +134,6 @@ require ( golang.org/x/mod v0.37.0 // indirect golang.org/x/net v0.57.0 // indirect golang.org/x/oauth2 v0.36.0 // indirect - golang.org/x/sync v0.22.0 // indirect golang.org/x/sys v0.47.0 // indirect golang.org/x/term v0.45.0 // indirect golang.org/x/text v0.40.0 // indirect diff --git a/main.go b/main.go index 99d415ab7..bf865dbbb 100644 --- a/main.go +++ b/main.go @@ -45,7 +45,7 @@ type GlobalFlags struct { CollStatsNamespaces string `help:"List of comma separared databases.collections to get $collStats" name:"mongodb.collstats-colls" placeholder:"db1,db2.col2"` IndexStatsCollections string `help:"List of comma separared databases.collections to get $indexStats" name:"mongodb.indexstats-colls" placeholder:"db1.col1,db2.col2"` URI []string `env:"MONGODB_URI" help:"MongoDB connection URI" name:"mongodb.uri" placeholder:"mongodb://user:pass@127.0.0.1:27017/admin?ssl=true"` - GlobalConnPool bool `default:"true" help:"Use global connection pool instead of creating new pool for each http request." name:"mongodb.global-conn-pool" negatable:""` + GlobalConnPool bool `default:"true" help:"Use global connection pool instead of creating new pool for each http request. Enabled by default." name:"mongodb.global-conn-pool" negatable:""` DirectConnect bool `default:"true" help:"Whether or not a direct connect should be made. Direct connections are not valid if multiple hosts are specified or an SRV URI is used." name:"mongodb.direct-connect" negatable:""` WebListenAddress string `default:":9216" help:"Address to listen on for web interface and telemetry" name:"web.listen-address"` WebTelemetryPath string `default:"/metrics" help:"Metrics expose path" name:"web.telemetry-path"`