From 40604ef17a08957ec100a113ca15c3b6a6fb664a Mon Sep 17 00:00:00 2001 From: skartikey <1942366+skartikey@users.noreply.github.com> Date: Mon, 13 Jul 2026 11:57:09 +0100 Subject: [PATCH 1/2] fix(common.postgresql): Skip connection liveness ping for PgBouncer --- plugins/common/postgresql/config.go | 13 +++- plugins/common/postgresql/service.go | 12 +--- plugins/inputs/pgbouncer/pgbouncer_test.go | 74 ++++++++++++++++++++++ 3 files changed, 89 insertions(+), 10 deletions(-) diff --git a/plugins/common/postgresql/config.go b/plugins/common/postgresql/config.go index b4cb06507c1ac..2022757357659 100644 --- a/plugins/common/postgresql/config.go +++ b/plugins/common/postgresql/config.go @@ -1,6 +1,7 @@ package postgresql import ( + "context" "fmt" "net" "net/url" @@ -51,10 +52,20 @@ func (c *Config) CreateService() (*Service, error) { // Specific support to make it work with PgBouncer too // See https://github.com/influxdata/telegraf/issues/3253#issuecomment-357505343 + var opts []stdlib.OptionOpenDB if c.IsPgBouncer { // Remove DriveConfig and revert it by the ParseConfig method // See https://github.com/influxdata/telegraf/issues/9134 connConfig.DefaultQueryExecMode = pgx.QueryExecModeSimpleProtocol + + // PgBouncer's admin console is not a SQL parser and rejects the + // "-- ping" liveness query the driver sends before reusing an idle + // connection, flooding the PgBouncer log and forcing a reconnect on + // every gather cycle. Skip the ping for PgBouncer connections. + // See https://github.com/influxdata/telegraf/issues/19252 + opts = append(opts, stdlib.OptionShouldPing(func(context.Context, stdlib.ShouldPingParams) bool { + return false + })) } // Provide the connection string without sensitive information for use as @@ -70,7 +81,7 @@ func (c *Config) CreateService() (*Service, error) { maxIdle: c.MaxIdle, maxOpen: c.MaxOpen, maxLifetime: time.Duration(c.MaxLifetime), - dsn: stdlib.RegisterConnConfig(connConfig), + connector: stdlib.GetConnector(*connConfig, opts...), }, nil } diff --git a/plugins/common/postgresql/service.go b/plugins/common/postgresql/service.go index 6ce793f96826d..abb93fa971cf3 100644 --- a/plugins/common/postgresql/service.go +++ b/plugins/common/postgresql/service.go @@ -2,10 +2,8 @@ package postgresql import ( "database/sql" + "database/sql/driver" "time" - - // Blank import required to register driver - _ "github.com/jackc/pgx/v5/stdlib" ) // Service common functionality shared between the postgresql and postgresql_extensible @@ -15,18 +13,14 @@ type Service struct { SanitizedAddress string ConnectionDatabase string - dsn string + connector driver.Connector maxIdle int maxOpen int maxLifetime time.Duration } func (p *Service) Start() error { - db, err := sql.Open("pgx", p.dsn) - if err != nil { - return err - } - p.DB = db + p.DB = sql.OpenDB(p.connector) p.DB.SetMaxOpenConns(p.maxOpen) p.DB.SetMaxIdleConns(p.maxIdle) diff --git a/plugins/inputs/pgbouncer/pgbouncer_test.go b/plugins/inputs/pgbouncer/pgbouncer_test.go index 4af835f1b8d77..be4c4f581ccbb 100644 --- a/plugins/inputs/pgbouncer/pgbouncer_test.go +++ b/plugins/inputs/pgbouncer/pgbouncer_test.go @@ -2,7 +2,9 @@ package pgbouncer import ( "fmt" + "strings" "testing" + "time" "github.com/stretchr/testify/require" "github.com/testcontainers/testcontainers-go/wait" @@ -101,6 +103,78 @@ func TestPgBouncerGeneratesMetricsIntegration(t *testing.T) { require.Equal(t, len(intMetricsPgBouncer)+len(intMetricsPgBouncerPools), metricsCounted) } +func TestPgBouncerIdleConnectionReuseIntegration(t *testing.T) { + if testing.Short() { + t.Skip("Skipping integration test in short mode") + } + + postgresServicePort := "5432" + pgBouncerServicePort := "6432" + + backend := testutil.Container{ + Image: "postgres:alpine", + ExposedPorts: []string{postgresServicePort}, + Env: map[string]string{ + "POSTGRES_HOST_AUTH_METHOD": "trust", + }, + WaitingFor: wait.ForLog("database system is ready to accept connections").WithOccurrence(2), + } + require.NoError(t, backend.Start(), "failed to start container") + defer backend.Terminate() + + container := testutil.Container{ + Image: "z9pascal/pgbouncer-container:1.24.1-latest", + ExposedPorts: []string{pgBouncerServicePort}, + Env: map[string]string{ + "PG_ENV_POSTGRESQL_USER": "pgbouncer", + "PG_ENV_POSTGRESQL_PASS": "pgbouncer", + }, + WaitingFor: wait.ForAll( + wait.ForListeningPort(pgBouncerServicePort), + wait.ForLog("LOG process up"), + ), + } + require.NoError(t, container.Start(), "failed to start container") + defer func() { + container.Terminate() + + // The container log can only be inspected after Terminate stopped the + // log streaming. PgBouncer's admin console logs an error for every + // statement it cannot parse, e.g. the "-- ping" liveness query the + // driver sends by default when reusing an idle connection. + require.NotContains(t, strings.Join(container.Logs.Msgs, ""), "invalid command") + }() + + addr := fmt.Sprintf( + "host=%s user=pgbouncer password=pgbouncer dbname=pgbouncer port=%s sslmode=disable", + container.Address, + container.Ports[pgBouncerServicePort], + ) + + // Mirror the connection-pool defaults of the plugin factory as keeping the + // connection idle and reusing it is the precondition for the driver's + // liveness check + p := &PgBouncer{ + Config: postgresql.Config{ + Address: config.NewSecret([]byte(addr)), + MaxIdle: 1, + MaxOpen: 1, + IsPgBouncer: true, + }, + } + require.NoError(t, p.Init()) + + var acc testutil.Accumulator + require.NoError(t, p.Start(&acc)) + defer p.Stop() + require.NoError(t, p.Gather(&acc)) + + // Let the connection sit idle beyond the one-second threshold after which + // the driver checks connection liveness before reusing it + time.Sleep(1500 * time.Millisecond) + require.NoError(t, p.Gather(&acc)) +} + func TestPgBouncerGeneratesMetricsIntegrationShowCommands(t *testing.T) { if testing.Short() { t.Skip("Skipping integration test in short mode") From 0e33ce4e73d265ef7d3c64cd93b5e5e2bbe01778 Mon Sep 17 00:00:00 2001 From: skartikey <1942366+skartikey@users.noreply.github.com> Date: Wed, 15 Jul 2026 12:32:06 +0100 Subject: [PATCH 2/2] fix(common.postgresql): Restrict liveness ping skip to the PgBouncer admin console The IsPgBouncer flag is also set by the postgresql and postgresql_extensible plugins whenever prepared_statements is disabled, which is the documented setting for reaching a server through a PgBouncer with pool_mode set to transaction. Those connections talk to a real PostgreSQL that parses the "-- ping" liveness query fine, so skipping the ping there only drops the liveness check without fixing anything. Let IsPgBouncer mark the admin console only and move the prepared-statement handling to a separate SimpleProtocol flag. Cover the PostgreSQL side with an integration test asserting that disabling prepared statements keeps the ping. --- plugins/common/postgresql/config.go | 33 ++++++---- plugins/inputs/pgbouncer/pgbouncer_test.go | 13 +--- plugins/inputs/postgresql/postgresql.go | 2 +- plugins/inputs/postgresql/postgresql_test.go | 63 +++++++++++++++++++ .../postgresql_extensible.go | 2 +- 5 files changed, 88 insertions(+), 25 deletions(-) diff --git a/plugins/common/postgresql/config.go b/plugins/common/postgresql/config.go index 2022757357659..ec43ba7b2a836 100644 --- a/plugins/common/postgresql/config.go +++ b/plugins/common/postgresql/config.go @@ -25,7 +25,15 @@ type Config struct { MaxIdle int `toml:"max_idle"` MaxOpen int `toml:"max_open"` MaxLifetime config.Duration `toml:"max_lifetime"` - IsPgBouncer bool `toml:"-"` + + // Internal flags + + // IsPgBouncer marks a connection to the PgBouncer admin console, which + // only understands PgBouncer's own SHOW commands + IsPgBouncer bool `toml:"-"` + // SimpleProtocol disables prepared statements, required for the admin + // console and for servers reached through a PgBouncer in transaction mode + SimpleProtocol bool `toml:"-"` } func (c *Config) CreateService() (*Service, error) { @@ -50,19 +58,22 @@ func (c *Config) CreateService() (*Service, error) { // Remove the socket name from the path connConfig.Host = socketRegexp.ReplaceAllLiteralString(connConfig.Host, "") - // Specific support to make it work with PgBouncer too + // Neither the PgBouncer admin console nor a server reached through a + // PgBouncer with pool_mode set to transaction supports prepared statements // See https://github.com/influxdata/telegraf/issues/3253#issuecomment-357505343 - var opts []stdlib.OptionOpenDB - if c.IsPgBouncer { - // Remove DriveConfig and revert it by the ParseConfig method - // See https://github.com/influxdata/telegraf/issues/9134 + // and https://github.com/influxdata/telegraf/issues/9134 + if c.IsPgBouncer || c.SimpleProtocol { connConfig.DefaultQueryExecMode = pgx.QueryExecModeSimpleProtocol + } - // PgBouncer's admin console is not a SQL parser and rejects the - // "-- ping" liveness query the driver sends before reusing an idle - // connection, flooding the PgBouncer log and forcing a reconnect on - // every gather cycle. Skip the ping for PgBouncer connections. - // See https://github.com/influxdata/telegraf/issues/19252 + // The PgBouncer admin console is not a SQL parser and rejects the "-- ping" + // liveness query the driver sends before reusing an idle connection, + // flooding the PgBouncer log and forcing a reconnect on every gather cycle. + // Servers reached through a PgBouncer are not affected as the ping is + // forwarded to PostgreSQL, so those keep the liveness check. + // See https://github.com/influxdata/telegraf/issues/19252 + var opts []stdlib.OptionOpenDB + if c.IsPgBouncer { opts = append(opts, stdlib.OptionShouldPing(func(context.Context, stdlib.ShouldPingParams) bool { return false })) diff --git a/plugins/inputs/pgbouncer/pgbouncer_test.go b/plugins/inputs/pgbouncer/pgbouncer_test.go index be4c4f581ccbb..f8f511b5cde99 100644 --- a/plugins/inputs/pgbouncer/pgbouncer_test.go +++ b/plugins/inputs/pgbouncer/pgbouncer_test.go @@ -108,20 +108,9 @@ func TestPgBouncerIdleConnectionReuseIntegration(t *testing.T) { t.Skip("Skipping integration test in short mode") } - postgresServicePort := "5432" pgBouncerServicePort := "6432" - backend := testutil.Container{ - Image: "postgres:alpine", - ExposedPorts: []string{postgresServicePort}, - Env: map[string]string{ - "POSTGRES_HOST_AUTH_METHOD": "trust", - }, - WaitingFor: wait.ForLog("database system is ready to accept connections").WithOccurrence(2), - } - require.NoError(t, backend.Start(), "failed to start container") - defer backend.Terminate() - + // The admin console serves the SHOW commands without a PostgreSQL backend container := testutil.Container{ Image: "z9pascal/pgbouncer-container:1.24.1-latest", ExposedPorts: []string{pgBouncerServicePort}, diff --git a/plugins/inputs/postgresql/postgresql.go b/plugins/inputs/postgresql/postgresql.go index 46b2354874cb2..9dda138d480a2 100644 --- a/plugins/inputs/postgresql/postgresql.go +++ b/plugins/inputs/postgresql/postgresql.go @@ -32,7 +32,7 @@ func (*Postgresql) SampleConfig() string { } func (p *Postgresql) Init() error { - p.IsPgBouncer = !p.PreparedStatements + p.SimpleProtocol = !p.PreparedStatements service, err := p.Config.CreateService() if err != nil { diff --git a/plugins/inputs/postgresql/postgresql_test.go b/plugins/inputs/postgresql/postgresql_test.go index 3d44d08db3eda..62dfb1223f534 100644 --- a/plugins/inputs/postgresql/postgresql_test.go +++ b/plugins/inputs/postgresql/postgresql_test.go @@ -2,7 +2,9 @@ package postgresql import ( "fmt" + "strings" "testing" + "time" "github.com/stretchr/testify/require" "github.com/testcontainers/testcontainers-go/wait" @@ -35,6 +37,67 @@ func launchTestContainer(t *testing.T) *testutil.Container { return &container } +func TestPostgresqlIdleConnectionPingIntegration(t *testing.T) { + if testing.Short() { + t.Skip("Skipping integration test in short mode") + } + + // Log every statement to make the driver's liveness ping observable + container := testutil.Container{ + Image: "postgres:alpine", + ExposedPorts: []string{servicePort}, + Env: map[string]string{ + "POSTGRES_HOST_AUTH_METHOD": "trust", + }, + Cmd: []string{"postgres", "-c", "log_statement=all"}, + WaitingFor: wait.ForAll( + wait.ForLog("database system is ready to accept connections").WithOccurrence(2), + wait.ForListeningPort(servicePort), + ), + } + require.NoError(t, container.Start(), "failed to start container") + defer func() { + container.Terminate() + + // The container log can only be inspected after Terminate stopped the + // log streaming. PostgreSQL parses the "-- ping" liveness query just + // fine, so disabling prepared statements must not cost the connection + // its liveness check. + require.Contains(t, strings.Join(container.Logs.Msgs, ""), "statement: -- ping") + }() + + addr := fmt.Sprintf( + "host=%s port=%s user=postgres sslmode=disable", + container.Address, + container.Ports[servicePort], + ) + + // Disabling prepared statements is what users configure to reach a server + // through a PgBouncer with pool_mode set to transaction. Mirror the + // connection-pool defaults of the plugin factory as keeping the connection + // idle and reusing it is the precondition for the driver's liveness check + p := &Postgresql{ + Config: postgresql.Config{ + Address: config.NewSecret([]byte(addr)), + MaxIdle: 1, + MaxOpen: 1, + }, + Databases: []string{"postgres"}, + PreparedStatements: false, + } + require.NoError(t, p.Init()) + + var acc testutil.Accumulator + require.NoError(t, p.Start(&acc)) + defer p.Stop() + require.NoError(t, p.Gather(&acc)) + + // Let the connection sit idle beyond the one-second threshold after which + // the driver checks connection liveness before reusing it + time.Sleep(1500 * time.Millisecond) + require.NoError(t, p.Gather(&acc)) +} + func TestPostgresqlGeneratesMetricsIntegration(t *testing.T) { if testing.Short() { t.Skip("Skipping integration test in short mode") diff --git a/plugins/inputs/postgresql_extensible/postgresql_extensible.go b/plugins/inputs/postgresql_extensible/postgresql_extensible.go index e8236d3b3bbe1..801e9e775e84c 100644 --- a/plugins/inputs/postgresql_extensible/postgresql_extensible.go +++ b/plugins/inputs/postgresql_extensible/postgresql_extensible.go @@ -90,7 +90,7 @@ func (p *Postgresql) Init() error { } p.Query[i] = q } - p.Config.IsPgBouncer = !p.PreparedStatements + p.Config.SimpleProtocol = !p.PreparedStatements // Create a service to access the PostgreSQL server service, err := p.Config.CreateService()