base

package
v1.6.1 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: May 27, 2026 License: Apache-2.0 Imports: 11 Imported by: 3

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var (
	ErrNotConnected  = errors.New("database not connected")
	ErrAlreadyClosed = errors.New("database already closed")
)

Functions

This section is empty.

Types

type AutoReconnector 

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

AutoReconnector performs automatic reconnection on connection loss

func NewAutoReconnector 

func NewAutoReconnector(target Reconnectable, interval time.Duration, maxAttempts int) *AutoReconnector

NewAutoReconnector creates a new auto-reconnector

func (*AutoReconnector) GetAttempts 

func (a *AutoReconnector) GetAttempts() int64

GetAttempts returns the current number of reconnection attempts

func (*AutoReconnector) Start 

func (a *AutoReconnector) Start(ctx context.Context)

Start starts the auto-reconnect routine

func (*AutoReconnector) Stop 

func (a *AutoReconnector) Stop()

Stop stops the auto-reconnector

type ConnectionPoolStats 

type ConnectionPoolStats struct {
	MaxOpenConnections int64         // Maximum number of open connections
	OpenConnections    int64         // Number of established connections
	InUse              int64         // Number of connections currently in use
	Idle               int64         // Number of idle connections
	MaxIdleConnections int64         // Maximum number of idle connections
	WaitCount          int64         // Total number of connections waited for
	WaitDuration       time.Duration // Total time blocked waiting for a new connection
	MaxIdleClosed      int64         // Total number of connections closed due to SetMaxIdleConns
	MaxLifetimeClosed  int64         // Total number of connections closed due to SetConnMaxLifetime
}

ConnectionPoolStats represents database connection pool statistics

type HealthCheckable 

type HealthCheckable interface {
	CheckHealth() error
	Name() string
}

HealthCheckable interface for health checkable components

type HealthChecker 

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

HealthChecker performs periodic health checks

func NewHealthChecker 

func NewHealthChecker(target HealthCheckable, interval time.Duration, customQuery string) *HealthChecker

NewHealthChecker creates a new health checker. Recovery (Reconnect) is attempted on first failure so the pool is replaced as soon as it is unhealthy.

func (*HealthChecker) IsHealthy 

func (h *HealthChecker) IsHealthy() bool

IsHealthy returns the current health status

func (*HealthChecker) Start 

func (h *HealthChecker) Start(ctx context.Context)

Start starts the health check routine

func (*HealthChecker) Stop 

func (h *HealthChecker) Stop()

Stop stops the health checker

type LeakDetector 

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

LeakDetector detects connection leaks by monitoring connection usage

func NewLeakDetector 

func NewLeakDetector(target Monitorable, threshold time.Duration) *LeakDetector

NewLeakDetector creates a new connection leak detector

func (*LeakDetector) Start 

func (l *LeakDetector) Start(ctx context.Context)

Start starts the leak detection routine

func (*LeakDetector) Stop 

func (l *LeakDetector) Stop()

Stop stops the leak detector

type MetricsConfig 

type MetricsConfig struct {
	// Enabled determines if metrics recording is enabled
	Enabled bool

	// Namespace for Prometheus metrics
	Namespace string

	// Subsystem for Prometheus metrics
	Subsystem string

	// Labels to add to all metrics
	Labels map[string]string

	// SlowQueryThreshold defines the threshold for slow query detection
	SlowQueryThreshold time.Duration
}

MetricsConfig defines configuration for metrics recording

func DefaultMetricsConfig 

func DefaultMetricsConfig() *MetricsConfig

DefaultMetricsConfig returns default metrics configuration

type MetricsRecorder 

type MetricsRecorder interface {
	// RecordConnectionPoolStats records connection pool statistics
	RecordConnectionPoolStats(stats *ConnectionPoolStats)

	// RecordHealthCheck records health check results
	RecordHealthCheck(success bool)

	// RecordQuery records SQL query duration and errors
	RecordQuery(duration time.Duration, err error, threshold time.Duration)

	// RecordTx records transaction duration and status
	RecordTx(duration time.Duration, committed bool)

	// IncConnectAttempt increments connection attempt counter
	IncConnectAttempt()

	// IncConnectRetry increments connection retry counter
	IncConnectRetry()

	// IncConnectSuccess increments connection success counter
	IncConnectSuccess()

	// IncConnectFailure increments connection failure counter
	IncConnectFailure()
}

MetricsRecorder defines the interface for recording database metrics

type Monitorable 

type Monitorable interface {
	GetStats() *ConnectionPoolStats
	Name() string
}

Monitorable interface for components that can be monitored

type NoOpMetricsRecorder 

type NoOpMetricsRecorder struct{}

NoOpMetricsRecorder provides a no-operation implementation of MetricsRecorder This is useful when metrics recording is disabled or not implemented

func (*NoOpMetricsRecorder) IncConnectAttempt 

func (n *NoOpMetricsRecorder) IncConnectAttempt()

IncConnectAttempt implements MetricsRecorder

func (*NoOpMetricsRecorder) IncConnectFailure 

func (n *NoOpMetricsRecorder) IncConnectFailure()

IncConnectFailure implements MetricsRecorder

func (*NoOpMetricsRecorder) IncConnectRetry 

func (n *NoOpMetricsRecorder) IncConnectRetry()

IncConnectRetry implements MetricsRecorder

func (*NoOpMetricsRecorder) IncConnectSuccess 

func (n *NoOpMetricsRecorder) IncConnectSuccess()

IncConnectSuccess implements MetricsRecorder

func (*NoOpMetricsRecorder) RecordConnectionPoolStats 

func (n *NoOpMetricsRecorder) RecordConnectionPoolStats(stats *ConnectionPoolStats)

RecordConnectionPoolStats implements MetricsRecorder

func (*NoOpMetricsRecorder) RecordHealthCheck 

func (n *NoOpMetricsRecorder) RecordHealthCheck(success bool)

RecordHealthCheck implements MetricsRecorder

func (*NoOpMetricsRecorder) RecordQuery 

func (n *NoOpMetricsRecorder) RecordQuery(duration time.Duration, err error, threshold time.Duration)

RecordQuery implements MetricsRecorder

func (*NoOpMetricsRecorder) RecordTx 

func (n *NoOpMetricsRecorder) RecordTx(duration time.Duration, committed bool)

RecordTx implements MetricsRecorder

type PoolMonitor 

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

PoolMonitor monitors connection pool health and triggers alerts

func NewPoolMonitor 

func NewPoolMonitor(target Monitorable, interval time.Duration, thresholds *PoolThresholds) *PoolMonitor

NewPoolMonitor creates a new connection pool monitor

func (*PoolMonitor) Start 

func (m *PoolMonitor) Start(ctx context.Context)

Start starts the monitoring routine

func (*PoolMonitor) Stop 

func (m *PoolMonitor) Stop()

Stop stops the monitor

type PoolThresholds 

type PoolThresholds struct {
	UsagePercentage float64       // Alert when pool usage exceeds this (0.0-1.0)
	WaitDuration    time.Duration // Alert when wait duration exceeds this
	WaitCount       int64         // Alert when wait count exceeds this
}

PoolThresholds defines alert thresholds for connection pool monitoring

type PrometheusMetrics 

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

PrometheusMetrics provides a unified Prometheus metrics implementation

func NewPrometheusMetrics 

func NewPrometheusMetrics(config *MetricsConfig) *PrometheusMetrics

NewPrometheusMetrics creates a new Prometheus metrics instance

func (*PrometheusMetrics) GetGatherer 

func (pm *PrometheusMetrics) GetGatherer() prometheus.Gatherer

GetGatherer returns the Prometheus gatherer

func (*PrometheusMetrics) IncConnectAttempt 

func (pm *PrometheusMetrics) IncConnectAttempt()

IncConnectAttempt implements MetricsRecorder

func (*PrometheusMetrics) IncConnectFailure 

func (pm *PrometheusMetrics) IncConnectFailure()

IncConnectFailure implements MetricsRecorder

func (*PrometheusMetrics) IncConnectRetry 

func (pm *PrometheusMetrics) IncConnectRetry()

IncConnectRetry implements MetricsRecorder

func (*PrometheusMetrics) IncConnectSuccess 

func (pm *PrometheusMetrics) IncConnectSuccess()

IncConnectSuccess implements MetricsRecorder

func (*PrometheusMetrics) RecordConnectionPoolStats 

func (pm *PrometheusMetrics) RecordConnectionPoolStats(stats *ConnectionPoolStats)

RecordConnectionPoolStats implements MetricsRecorder

func (*PrometheusMetrics) RecordHealthCheck 

func (pm *PrometheusMetrics) RecordHealthCheck(success bool)

RecordHealthCheck implements MetricsRecorder

func (*PrometheusMetrics) RecordQuery 

func (pm *PrometheusMetrics) RecordQuery(duration time.Duration, err error, threshold time.Duration)

RecordQuery implements MetricsRecorder

func (*PrometheusMetrics) RecordTx 

func (pm *PrometheusMetrics) RecordTx(duration time.Duration, committed bool)

RecordTx implements MetricsRecorder

type QueryMonitor 

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

QueryMonitor provides slow query monitoring and logging

func NewQueryMonitor 

func NewQueryMonitor(enabled bool, threshold time.Duration, recorder MetricsRecorder) *QueryMonitor

NewQueryMonitor creates a new query monitor

func (*QueryMonitor) MonitorExec 

func (m *QueryMonitor) MonitorExec(ctx context.Context, db *sql.DB, query string, args []interface{}) (sql.Result, error)

MonitorExec wraps a database exec with monitoring

func (*QueryMonitor) MonitorQuery 

func (m *QueryMonitor) MonitorQuery(ctx context.Context, db *sql.DB, query string, args []interface{}, fn func() error) error

MonitorQuery wraps a database query with monitoring

func (*QueryMonitor) MonitorQueryRow 

func (m *QueryMonitor) MonitorQueryRow(ctx context.Context, db *sql.DB, query string, args []interface{}, scan func(*sql.Row) error) error

MonitorQueryRow wraps a database query row with monitoring

type Reconnectable 

type Reconnectable interface {
	Reconnect() error
	IsConnected() bool
	Name() string
}

Reconnectable interface for components that can be reconnected

type Recoverable 

type Recoverable interface {
	Reconnect() error
	IsConnected() bool
}

Recoverable interface for components that can recover from failures

type SQLPlugin 

type SQLPlugin struct {
	*plugins.BasePlugin
	// contains filtered or unexported fields
}

SQLPlugin provides common functionality for all SQL plugins

func NewBaseSQLPlugin 

func NewBaseSQLPlugin(
	id, name, desc, version, confPrefix string,
	weight int,
	config *interfaces.Config,
) *SQLPlugin

NewBaseSQLPlugin creates a new base SQL plugin

func (*SQLPlugin) CheckHealth 

func (p *SQLPlugin) CheckHealth() error

CheckHealth performs a health check. It ensures the database is reachable: if the current connection state is bad or the health query fails (e.g. one stale connection from the pool), it tries Reconnect() once and retries before reporting unhealthy, so a single bad connection does not cause the app to be marked down or stop serving.

func (*SQLPlugin) CheckHealthContext added in v1.6.1 

func (p *SQLPlugin) CheckHealthContext(ctx context.Context) error

func (*SQLPlugin) CleanupTasks 

func (p *SQLPlugin) CleanupTasks() error

CleanupTasks performs cleanup on shutdown

func (*SQLPlugin) GetAutoReconnector 

func (p *SQLPlugin) GetAutoReconnector() *AutoReconnector

GetAutoReconnector returns the auto-reconnector instance

func (*SQLPlugin) GetDB 

func (p *SQLPlugin) GetDB() (*sql.DB, error)

GetDB returns the database connection. Important: when auto-reconnect is enabled, do not cache the returned *sql.DB in long-lived structs (e.g. global Data/Repo). After Reconnect(), the previous *sql.DB is closed; cached references will get "sql: database is closed". Obtain the DB at the point of use (e.g. per request) or use a provider that calls GetDBWithContext when needed.

func (*SQLPlugin) GetDBWithContext 

func (p *SQLPlugin) GetDBWithContext(ctx context.Context) (*sql.DB, error)

GetDBWithContext returns the database connection with context support. When EnsureAliveBeforeHandout is true (default), it pings the pool before returning; on failure it triggers Reconnect() once so the pool is not handed out when already broken. Do not cache the returned *sql.DB when auto-reconnect is enabled; after Reconnect() it becomes closed.

func (*SQLPlugin) GetDialect 

func (p *SQLPlugin) GetDialect() string

GetDialect returns the database dialect This method is thread-safe

func (*SQLPlugin) GetMetricsRecorder 

func (p *SQLPlugin) GetMetricsRecorder() MetricsRecorder

GetMetricsRecorder returns the current metrics recorder

func (*SQLPlugin) GetQueryMonitor 

func (p *SQLPlugin) GetQueryMonitor() *QueryMonitor

GetQueryMonitor returns the query monitor for slow query detection

func (*SQLPlugin) GetStats 

func (p *SQLPlugin) GetStats() *ConnectionPoolStats

GetStats returns connection pool statistics This method is thread-safe and can be called concurrently

func (*SQLPlugin) GetValidatedConn added in v1.6.1 

func (p *SQLPlugin) GetValidatedConn(ctx context.Context) (*sql.Conn, error)

GetValidatedConn returns a single connection from the pool that has been verified alive (Ping). The returned connection is guaranteed to be usable at handoff time. Caller must call conn.Close() when done to return the connection to the pool.

func (*SQLPlugin) InitializeResources 

func (p *SQLPlugin) InitializeResources(rt plugins.Runtime) error

InitializeResources initializes plugin resources

func (*SQLPlugin) IsConnected 

func (p *SQLPlugin) IsConnected() bool

IsConnected checks if database is connected This method performs actual connection validation for accuracy

func (*SQLPlugin) IsConnectedContext added in v1.6.1 

func (p *SQLPlugin) IsConnectedContext(ctx context.Context) bool

func (*SQLPlugin) Reconnect 

func (p *SQLPlugin) Reconnect() error

Reconnect attempts to reconnect to the database This method is called by AutoReconnector when connection is lost

func (*SQLPlugin) ReconnectContext added in v1.6.1 

func (p *SQLPlugin) ReconnectContext(ctx context.Context) error

func (*SQLPlugin) SetMetricsRecorder 

func (p *SQLPlugin) SetMetricsRecorder(recorder MetricsRecorder)

SetMetricsRecorder sets the metrics recorder for this plugin

func (*SQLPlugin) SetProvider added in v1.6.1 

func (p *SQLPlugin) SetProvider(provider interfaces.DBProvider)

SetProvider sets the stable DB provider published into runtime resources.

func (*SQLPlugin) SharedProviderResourceName added in v1.6.1 

func (p *SQLPlugin) SharedProviderResourceName() string

SharedProviderResourceName returns the canonical shared resource name for a SQL provider.

func (*SQLPlugin) StartupTasks 

func (p *SQLPlugin) StartupTasks() error

StartupTasks performs startup initialization with retry support

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.