fix(common.postgresql): Skip connection liveness ping for PgBouncer

plugins/common/postgresql/config.go

@@ -1,6 +1,7 @@
 package postgresql
 
 import (
+	"context"
 	"fmt"
 	"net"
 	"net/url"
@@ -24,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) {
@@ -49,14 +58,27 @@ 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
-	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
 	}
 
+	// 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
+		}))
+	}
+
 	// Provide the connection string without sensitive information for use as
 	// tag or other output properties
 	sanitizedAddr, err := c.sanitizedAddress()
@@ -70,7 +92,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
 }
 

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)

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,67 @@ 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")
+	}
+
+	pgBouncerServicePort := "6432"
+
+	// 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},
+		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")

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 {

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")

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()