Start webserver before database init, and improve database init/error handling (#479)

Signed-off-by: Anders Swanson <anders.swanson@oracle.com>

Author: anders-swanson

alertlog/alertlog.go

@@ -129,8 +129,11 @@ func readLastMatchingLogRecord(logDestination, database string, perDatabaseFiles
 
 // UpdateLog appends newly queried alert log records for a database to the configured log destination.
 func UpdateLog(logDestination string, perDatabaseFiles bool, logger *slog.Logger, d *collector.Database) {
+	if !d.StartupReady() {
+		return
+	}
 	// Do not try to query the alert log if the database configuration is invalid.
-	if !d.IsValid() {
+	if d.IsValid() != nil {
 		return
 	}
 	now := time.Now()

alertlog/alertlog_test.go

@@ -9,6 +9,11 @@ import (
 	"path/filepath"
 	"testing"
 	"time"
+
+	"io"
+	"log/slog"
+
+	"github.com/oracle/oracle-db-appdev-monitoring/collector"
 )
 
 func TestNullStringValue(t *testing.T) {
@@ -156,3 +161,15 @@ func TestRetryTrackerRecordSuccessResetsState(t *testing.T) {
 		t.Fatalf("expected retry to be allowed after success reset, got retry_after=%v", retryAfter)
 	}
 }
+
+func TestUpdateLogSkipsWhenStartupNotReady(t *testing.T) {
+	logger := slog.New(slog.NewTextHandler(io.Discard, nil))
+	logPath := filepath.Join(t.TempDir(), "alert.log")
+	db := &collector.Database{Name: "db1"}
+
+	UpdateLog(logPath, false, logger, db)
+
+	if _, err := os.Stat(logPath); !os.IsNotExist(err) {
+		t.Fatalf("expected log file to not be created while startup is in progress, got err=%v", err)
+	}
+}

collector/collector.go

@@ -48,7 +48,6 @@ func maskDsn(dsn string) string {
 // NewExporter creates a new Exporter instance
 func NewExporter(logger *slog.Logger, m *MetricsConfiguration) *Exporter {
 	var databases []*Database
-	wg := &sync.WaitGroup{}
 
 	var allConstLabels []string
 	// All the metrics of the same name need to have the same set of labels
@@ -64,16 +63,10 @@ func NewExporter(logger *slog.Logger, m *MetricsConfiguration) *Exporter {
 	}
 
 	for dbname, dbconfig := range m.Databases {
-		logger.Info("Initializing database", "database", dbname)
+		logger.Info("Registering database", "database", dbname)
 		database := NewDatabase(logger, m.DatabaseLabel(), dbname, dbconfig)
 		databases = append(databases, database)
-		wg.Add(1)
-		go func() {
-			defer wg.Done()
-			database.WarmupConnectionPool(logger)
-		}()
 	}
-	wg.Wait()
 	e := &Exporter{
 		mu: &sync.Mutex{},
 		duration: prometheus.NewGauge(prometheus.GaugeOpts{
@@ -116,6 +109,15 @@ func NewExporter(logger *slog.Logger, m *MetricsConfiguration) *Exporter {
 	return e
 }
 
+func (e *Exporter) InitializeDatabases() {
+	for _, database := range e.databases {
+		e.logger.Info("Starting database connection warmup", "database", database.Name)
+		if err := database.WarmupConnectionPool(e.logger, e.MetricsConfiguration.ConnectionBackoff()); err != nil {
+			e.logger.Error("Database startup warmup failed", "error", err, "database", database.Name)
+		}
+	}
+}
+
 func (e *Exporter) constLabels() map[string]string {
 	// All the metrics of the same name need to have the same labels
 	// If a label is set for a particular database, it must be included also
@@ -252,11 +254,16 @@ func (e *Exporter) scrapeDatabase(ch chan<- prometheus.Metric, errChan chan<- er
 	}()
 
 	// If the database configuration is invalid, do not attempt to ping or reestablish the database connection.
-	if !d.IsValid() {
-		e.logger.Warn("Invalid database configuration, will not attempt reconnection", "database", d.Name)
+	if retryAfter := d.IsValid(); retryAfter != nil {
+		e.logger.Warn("Invalid database configuration", "database", d.Name, "retry_after", retryAfter)
 		errChan <- fmt.Errorf("database %s is invalid, will not be scraped", d.Name)
 		return
 	}
+	if !d.StartupReady() {
+		e.logger.Info("Database connection in progress", "database", d.Name)
+		errChan <- nil
+		return
+	}
 	// If ping fails, we will try again on the next iteration of metrics scraping
 	if err := d.ping(e.logger, e.MetricsConfiguration.ConnectionBackoff()); err != nil {
 		e.logger.Error("Error pinging database", "error", err, "database", d.Name)

collector/connect_godror.go

@@ -76,7 +76,6 @@ func connect(logger *slog.Logger, dbname string, dbconfig DatabaseConfig) *sql.D
 	// note that this just configures the connection, it does not actually connect until later
 	// when we call db.Ping()
 	db := sql.OpenDB(godror.NewConnector(P))
-	initdb(logger, dbname, dbconfig, db)
 	return db
 }
 
@@ -91,3 +90,20 @@ func isInvalidCredentialsError(err error) bool {
 	}
 	return oraErr.Code() == ora01017code || oraErr.Code() == ora28000code
 }
+
+func isTemporaryConnectionError(err error) bool {
+	err = errors.Unwrap(err)
+	if err == nil {
+		return false
+	}
+	oraErr, ok := err.(*godror.OraErr)
+	if !ok {
+		return false
+	}
+	switch oraErr.Code() {
+	case ora01033code, ora03113code, ora03114code, ora12537code:
+		return true
+	default:
+		return false
+	}
+}

collector/connect_goora.go

@@ -51,7 +51,6 @@ func connect(logger *slog.Logger, dbname string, dbconfig DatabaseConfig) *sql.D
 
 	// Configure connection pool (sql.DB handles pooling)
 	setConnectionPool(logger, dbname, dbconfig, db)
-	initdb(logger, dbname, dbconfig, db)
 	return db
 }
 
@@ -81,3 +80,20 @@ func isInvalidCredentialsError(err error) bool {
 	}
 	return oraErr.ErrCode == ora01017code || oraErr.ErrCode == ora28000code
 }
+
+func isTemporaryConnectionError(err error) bool {
+	if err == nil {
+		return false
+	}
+	var oraErr *network.OracleError
+	ok := errors.As(err, &oraErr)
+	if !ok {
+		return false
+	}
+	switch oraErr.ErrCode {
+	case ora01033code, ora03113code, ora03114code, ora12537code:
+		return true
+	default:
+		return false
+	}
+}

collector/database.go

@@ -6,16 +6,22 @@ package collector
 import (
 	"context"
 	"database/sql"
+	"errors"
 	"fmt"
 	"github.com/prometheus/client_golang/prometheus"
 	"log/slog"
 	"strings"
+	"sync"
 	"time"
 )
 
 const (
 	ora01017code = 1017
+	ora01033code = 1033
 	ora28000code = 28000
+	ora03113code = 3113
+	ora03114code = 3114
+	ora12537code = 12537
 )
 
 func (d *Database) UpMetric(exporterLabels map[string]string) prometheus.Metric {
@@ -49,9 +55,14 @@ func NewDatabase(logger *slog.Logger, dblabel, dbname string, dbconfig DatabaseC
 		Session:       db,
 		Config:        dbconfig,
 		DatabaseLabel: dblabel,
+		reconnectMU:   sync.Mutex{},
 	}
 }
 
+func (d *Database) StartupReady() bool {
+	return d.startupReady.Load()
+}
+
 // initCache resets the metrics cached. Used on startup and when metrics are reloaded.
 func (d *Database) initCache(metrics map[string]*Metric) {
 	d.MetricsCache = NewMetricsCache(metrics)
@@ -60,7 +71,18 @@ func (d *Database) initCache(metrics map[string]*Metric) {
 // WarmupConnectionPool serially acquires connections to "warm up" the connection pool.
 // This is a workaround for a perceived bug in ODPI_C where rapid acquisition of connections
 // results in a SIGABRT.
-func (d *Database) WarmupConnectionPool(logger *slog.Logger) {
+func (d *Database) WarmupConnectionPool(logger *slog.Logger, backoff time.Duration) error {
+	defer d.startupReady.Store(true)
+	return d.warmupSession(logger, backoff, d.Session)
+}
+
+func (d *Database) warmupSession(logger *slog.Logger, backoff time.Duration, session *sql.DB) error {
+	if session == nil {
+		d.Up = 0
+		d.invalidate(backoff)
+		return errors.New("database session is not initialized")
+	}
+
 	var connections []*sql.Conn
 	poolSize := d.Config.GetMaxOpenConns()
 	if poolSize < 1 {
@@ -74,67 +96,109 @@ func (d *Database) WarmupConnectionPool(logger *slog.Logger) {
 		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 		defer cancel()
 
-		conn, err := d.Session.Conn(ctx)
+		conn, err := session.Conn(ctx)
 		if err != nil {
 			return err
 		}
 		connections = append(connections, conn)
 		return nil
 	}
 
-	func() {
-		for i := 0; i < poolSize; i++ {
-			// short circuit warmup for inaccessible databases
-			if err := warmup(i + 1); err != nil {
-				d.Up = 0
-				logger.Error("Failed warmup database connection pool", "conn", i, "error", err, "database", d.Name)
-				return
-			}
+	initdb(logger, d.Name, d.Config, session)
+
+	for i := 0; i < poolSize; i++ {
+		// short circuit warmup for inaccessible databases
+		if err := warmup(i + 1); err != nil {
+			d.Up = 0
+			d.invalidate(backoff)
+			logger.Debug("Failed warmup database connection pool", "conn", i, "error", err, "database", d.Name)
+			return err
 		}
-	}()
+	}
 
 	logger.Debug("Warmed connection pool", "total", len(connections), "database", d.Name)
 	for i, conn := range connections {
 		if err := conn.Close(); err != nil {
 			logger.Debug("Failed to return database connection to pool on warmup", "conn", i+1, "error", err, "database", d.Name)
 		}
 	}
+	d.Up = 1
+	d.clearInvalid()
+	return nil
+}
+
+func (d *Database) reconnect(logger *slog.Logger, backoff time.Duration) error {
+	d.reconnectMU.Lock()
+	defer d.reconnectMU.Unlock()
+
+	logger.Info("Reconnecting database session", "database", d.Name)
+
+	session := connect(logger, d.Name, d.Config)
+	if err := d.warmupSession(logger, backoff, session); err != nil {
+		if session != nil {
+			_ = session.Close()
+		}
+		return err
+	}
+
+	oldSession := d.Session
+	d.Session = session
+	if oldSession != nil && oldSession != session {
+		_ = oldSession.Close()
+	}
+	return nil
 }
 
 // ping the database. If the database is disconnected, try to reconnect.
 // If the database type is unknown, try to reload it.
 func (d *Database) ping(logger *slog.Logger, backoff time.Duration) error {
+	if d.Session == nil {
+		return d.reconnect(logger, backoff)
+	}
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 	defer cancel()
 	err := d.Session.PingContext(ctx)
 	if err != nil {
 		d.Up = 0
-		if isInvalidCredentialsError(err) {
+		if isInvalidCredentialsError(err) || isTemporaryConnectionError(err) {
 			d.invalidate(backoff)
 			return err
 		}
-		// If database is closed, try to reconnect
-		if strings.Contains(err.Error(), "sql: database is closed") {
-			d.Session = connect(logger, d.Name, d.Config)
+		// If database is closed, rebuild the handle and rerun init/warmup.
+		if isClosedDatabaseError(err) {
+			return d.reconnect(logger, backoff)
 		}
 		return err
 	}
 	d.Up = 1
+	d.clearInvalid()
 	return nil
 }
 
-func (d *Database) IsValid() bool {
+func (d *Database) IsValid() *time.Duration {
 	if d.invalidUntil == nil {
-		return true
+		return nil
+	}
+	retryAfter := time.Until(*d.invalidUntil)
+	if retryAfter <= 0 {
+		return nil
 	}
-	return time.Now().After(*d.invalidUntil)
+	return &retryAfter
 }
 
 func (d *Database) invalidate(backoff time.Duration) {
 	until := time.Now().Add(backoff)
 	d.invalidUntil = &until
 }
 
+func (d *Database) clearInvalid() {
+	d.invalidUntil = nil
+}
+
+func isClosedDatabaseError(err error) bool {
+	return errors.Is(err, sql.ErrConnDone) || strings.Contains(err.Error(), "sql: database is closed")
+}
+
 func initdb(logger *slog.Logger, dbname string, dbconfig DatabaseConfig, db *sql.DB) {
 	logger.Debug(fmt.Sprintf("set max idle connections to %d", dbconfig.MaxIdleConns), "database", dbname)
 	db.SetMaxIdleConns(dbconfig.GetMaxIdleConns())