pool

Documentation

Index

• Constants
• type AuroraPGPool
  • func NewAuroraPool(ctx context.Context, config *Config, logger *zap.Logger) (*AuroraPGPool, error)
  • func (p *AuroraPGPool) Acquire(ctx context.Context) (*pgxpool.Conn, error)
  • func (p *AuroraPGPool) AcquireAllIdle(ctx context.Context) []*pgxpool.Conn
  • func (p *AuroraPGPool) AcquireFunc(ctx context.Context, f func(*pgxpool.Conn) error) error
  • func (p *AuroraPGPool) Begin(ctx context.Context) (pgx.Tx, error)
  • func (p *AuroraPGPool) BeginTx(ctx context.Context, txOptions pgx.TxOptions) (pgx.Tx, error)
  • func (p *AuroraPGPool) Close()
  • func (p *AuroraPGPool) Config() *pgxpool.Config
  • func (p *AuroraPGPool) CopyFrom(ctx context.Context, tableName pgx.Identifier, columnNames []string, ...) (int64, error)
  • func (p *AuroraPGPool) Exec(ctx context.Context, sql string, arguments ...any) (pgconn.CommandTag, error)
  • func (p *AuroraPGPool) Ping(ctx context.Context) error
  • func (p *AuroraPGPool) Query(ctx context.Context, sql string, args ...interface{}) (pgx.Rows, error)
  • func (p *AuroraPGPool) QueryRow(ctx context.Context, sql string, args ...interface{}) pgx.Row
  • func (p *AuroraPGPool) Reset()
  • func (p *AuroraPGPool) SendBatch(ctx context.Context, b *pgx.Batch) pgx.BatchResults
  • func (p *AuroraPGPool) Stat() *pgxpool.Stat
• type Canary
• type Config
• type Metric
• type MetricsEmitterFunction
• type MetricsTag
• type PGXConnPool
• type ValidationFunction

Constants

const (
	// DefaultQueryHealthCheckPeriod is the default value for Config.QueryHealthCheckPeriod.
	DefaultQueryHealthCheckPeriod = time.Second * 60

	// DefaultMinAvailableConnectionFailSize is the default value for Config.MinAvailableConnectionFailSize.
	DefaultMinAvailableConnectionFailSize = 3

	// DefaultValidationCountDestroyTrigger is the default value for Config.ValidationCountDestroyTrigger.
	DefaultValidationCountDestroyTrigger = 2

	// DefaultQueryValidationTimeout is the default value for Config.QueryValidationTimeout.
	DefaultQueryValidationTimeout = time.Millisecond * 500

	// DefaultPGXHealthCheckPeriod is used to limit PGX's own internal health check
	// period, as when AuroraPGPool is used, there are two background check threads.
	DefaultPGXHealthCheckPeriod = time.Minute * 5
)

Types

type AuroraPGPool

type AuroraPGPool struct {
	// contains filtered or unexported fields
}

func NewAuroraPool

func NewAuroraPool(ctx context.Context, config *Config, logger *zap.Logger) (*AuroraPGPool, error)

NewAuroraPool instantiates a new *AuroraPGPool from the given config.

The passed in context is only used to instantiate a new PGX pool when Config.PGXConfig is provided.

If logger is left nil, no logs will be emitted.

func (*AuroraPGPool) Acquire

func (p *AuroraPGPool) Acquire(ctx context.Context) (*pgxpool.Conn, error)

func (*AuroraPGPool) AcquireAllIdle

func (p *AuroraPGPool) AcquireAllIdle(ctx context.Context) []*pgxpool.Conn

func (*AuroraPGPool) AcquireFunc

func (p *AuroraPGPool) AcquireFunc(ctx context.Context, f func(*pgxpool.Conn) error) error

func (*AuroraPGPool) Begin

func (p *AuroraPGPool) Begin(ctx context.Context) (pgx.Tx, error)

func (*AuroraPGPool) BeginTx

func (p *AuroraPGPool) BeginTx(ctx context.Context, txOptions pgx.TxOptions) (pgx.Tx, error)

func (*AuroraPGPool) Close

func (p *AuroraPGPool) Close()

func (*AuroraPGPool) Config

func (p *AuroraPGPool) Config() *pgxpool.Config

func (*AuroraPGPool) CopyFrom

func (p *AuroraPGPool) CopyFrom(ctx context.Context, tableName pgx.Identifier, columnNames []string,
	rowSrc pgx.CopyFromSource,
) (int64, error)

func (*AuroraPGPool) Exec

func (p *AuroraPGPool) Exec(ctx context.Context, sql string, arguments ...any) (pgconn.CommandTag, error)

func (*AuroraPGPool) Ping

func (p *AuroraPGPool) Ping(ctx context.Context) error

func (*AuroraPGPool) Query

func (p *AuroraPGPool) Query(ctx context.Context, sql string, args ...interface{}) (pgx.Rows, error)

func (*AuroraPGPool) QueryRow

func (p *AuroraPGPool) QueryRow(ctx context.Context, sql string, args ...interface{}) pgx.Row

func (*AuroraPGPool) Reset

func (p *AuroraPGPool) Reset()

func (*AuroraPGPool) SendBatch

func (p *AuroraPGPool) SendBatch(ctx context.Context, b *pgx.Batch) pgx.BatchResults

func (*AuroraPGPool) Stat

func (p *AuroraPGPool) Stat() *pgxpool.Stat

type Canary

type Canary struct {
	ID          int64     `json:"ID"`
	LastUpdated time.Time `json:"lastUpdated"`
	DiffMS      float64   `json:"diffMS"`
}

type Config

type Config struct {
	// QueryValidator represents the required health check validation function.
	QueryValidator ValidationFunction

	// QueryValidationTimeout represents how long the query validation function is allowed to run.
	// Defaulted to DefaultQueryValidationTimeout when not specified.
	QueryValidationTimeout time.Duration

	// QueryHealthCheckPeriod represents how often the provided Config.QueryValidator function will run.
	// Defaulted to DefaultQueryHealthCheckPeriod when not specified.
	QueryHealthCheckPeriod time.Duration

	// MinAvailableConnectionFailSize is used in conjunction with Config.ValidationCountDestroyTrigger, and gates
	// when all connections on the pool are allowed to be reset. Specifically, the number of active connections
	// at the time of validation must be larger than this value, in order for all connections to be reset.
	//
	// Defaulted to DefaultMinAvailableConnectionFailSize when not specified.
	//
	// TODO(tjasko): This behavior seems strange and documentation is not provided on why this
	//  was done. Leaving this for a rainy day to figure out if this needs to be kept.
	MinAvailableConnectionFailSize int

	// ValidationCountDestroyTrigger represents how many consecutive validation attempts need to fail until all
	// connections on the pool are reset. When this count is reached, the pool will be reset on the next attempt.
	//
	// Defaulted to DefaultValidationCountDestroyTrigger when not specified.
	ValidationCountDestroyTrigger int

	// MetricsEmitter is an optional function used to collect metrics.
	MetricsEmitter MetricsEmitterFunction

	// PGXConfig is used to instantiate a new PGX pool instance on
	// behalf of the caller. Must not be used with Config.PGXPool.
	PGXConfig *pgxpool.Config

	// PGXPool is used to pass in a pre-instantiated PGX pool.
	// Must not be used with Config.PGXConfig.
	//
	// The caller is expected to set PGX's health check period to
	// an appropriate value, e.g.: DefaultPGXHealthCheckPeriod.
	PGXPool *pgxpool.Pool
}

Config is used to instantiate a new AuroraPGPool.

type Metric

type Metric struct {
	Key   string
	Value float64
}

type MetricsEmitterFunction

type MetricsEmitterFunction func(metrics interface{}, tags []MetricsTag)

MetricsEmitterFunction the pool can emit the pgxpool.Stat or raw metrics

type MetricsTag

type MetricsTag struct {
	Key   string
	Value string
}

type PGXConnPool

type PGXConnPool interface {
	Close()
	Acquire(ctx context.Context) (*pgxpool.Conn, error)
	AcquireFunc(ctx context.Context, f func(*pgxpool.Conn) error) error
	AcquireAllIdle(ctx context.Context) []*pgxpool.Conn
	Config() *pgxpool.Config
	Stat() *pgxpool.Stat
	Exec(ctx context.Context, sql string, arguments ...interface{}) (pgconn.CommandTag, error)
	Query(ctx context.Context, sql string, args ...interface{}) (pgx.Rows, error)
	QueryRow(ctx context.Context, sql string, args ...interface{}) pgx.Row
	SendBatch(ctx context.Context, b *pgx.Batch) pgx.BatchResults
	Begin(ctx context.Context) (pgx.Tx, error)
	BeginTx(ctx context.Context, txOptions pgx.TxOptions) (pgx.Tx, error)
	CopyFrom(ctx context.Context, tableName pgx.Identifier, columnNames []string, rowSrc pgx.CopyFromSource) (int64, error)
	Ping(ctx context.Context) error
	Reset()
}

type ValidationFunction

type ValidationFunction func(ctx context.Context, conn *pgxpool.Conn, logger *zap.Logger) bool

var DefaultReaderValidator ValidationFunction = readerValidator

var DefaultWriteValidator ValidationFunction = writeValidator