backend: Add bun

backend/postgres/README.md

@@ -15,32 +15,12 @@ Configuration for the PostgreSQL connection pool of the microservice.
     * postgres user to access the database
 * `POSTGRES_DB` default: `postgres`
     * database to access
-* `POSTGRES_MAX_RETRIES` default: `5`
-    * Maximum number of retries before giving up
-* `POSTGRES_RETRY_STATEMENT_TIMEOUT` default: `false`
-    * Whether to retry queries cancelled because of statement_timeout
-* `POSTGRES_MIN_RETRY_BACKOFF` default: `250ms`
-    *  Minimum backoff between each retry
-* `POSTGRES_MAX_RETRY_BACKOFF` default: `4s`
-    * Maximum backoff between each retry
 * `POSTGRES_DIAL_TIMEOUT` default: `5s`
     * Dial timeout for establishing new connections
 * `POSTGRES_READ_TIMEOUT` default: `30s`
     *  Timeout for socket reads. If reached, commands will fail with a timeout instead of blocking
 * `POSTGRES_WRITE_TIMEOUT` default: `30s`
     * Timeout for socket writes. If reached, commands will fail with a timeout instead of blocking.
-* `POSTGRES_POOL_SIZE` default: `100`
-    * Maximum number of socket connections
-* `POSTGRES_MIN_IDLE_CONNECTIONS` default: `10`
-    * Minimum number of idle connections which is useful when establishing new connection is slow
-* `POSTGRES_MAX_CONN_AGE` default: `30m`
-    * Connection age at which client retires (closes) the connection
-* `POSTGRES_POOL_TIMEOUT` default: `31s`
-    * Time for which client waits for free connection if all connections are busy before returning an error
-* `POSTGRES_IDLE_TIMEOUT` default: `5m`
-    * Amount of time after which client closes idle connections
-* `POSTGRES_IDLE_CHECK_FREQUENCY` default: `1m`
-    * Frequency of idle checks made by idle connections reaper
 * `POSTGRES_HEALTH_CHECK_TABLE_NAME` default: `healthcheck`
     * Name of the Table that is created to try if database is writeable
 * `POSTGRES_HEALTH_CHECK_RESULT_TTL` default: `10s`
@@ -53,7 +33,6 @@ Prometheus metrics exposed.
 * `pace_postgres_query_total{database}` Collects stats about the number of postgres queries made
 * `pace_postgres_query_failed{database}` Collects stats about the number of postgres queries failed
 * `pace_postgres_query_duration_seconds{database}` Collects performance metrics for each postgres query
-* `pace_postgres_query_rows_total{database}` Collects stats about the number of rows returned by a postgres query
 * `pace_postgres_query_affected_total{database}` Collects stats about the number of rows affected by a postgres query
 * `pace_postgres_connection_pool_hits{database}` Collects number of times free connection was found in the pool
 * `pace_postgres_connection_pool_misses{database}` Collects number of times free connection was NOT found in the pool

backend/postgres/bun.go

@@ -0,0 +1,123 @@
+// Copyright © 2024 by PACE Telematics GmbH. All rights reserved.
+
+package postgres
+
+import (
+	"context"
+	"database/sql"
+	"net"
+	"os"
+	"path/filepath"
+	"strconv"
+	"time"
+
+	"github.com/caarlos0/env/v10"
+	"github.com/pace/bricks/backend/postgres/hooks"
+	"github.com/pace/bricks/maintenance/health/servicehealthcheck"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/uptrace/bun"
+	"github.com/uptrace/bun/dialect/pgdialect"
+	"github.com/uptrace/bun/driver/pgdriver"
+
+	"github.com/pace/bricks/maintenance/log"
+)
+
+type Config struct {
+	Port     int    `env:"POSTGRES_PORT" envDefault:"5432"`
+	Host     string `env:"POSTGRES_HOST" envDefault:"postgres"`
+	Password string `env:"POSTGRES_PASSWORD" envDefault:"mysecretpassword"`
+	User     string `env:"POSTGRES_USER" envDefault:"postgres"`
+	Database string `env:"POSTGRES_DB" envDefault:"postgres"`
+
+	// ApplicationName is the application name. Used in logs on Pg side.
+	// Only availaible from pg-9.0.
+	ApplicationName string `env:"POSTGRES_APPLICATION_NAME" envDefault:"-"`
+	// Dial timeout for establishing new connections.
+	DialTimeout time.Duration `env:"POSTGRES_DIAL_TIMEOUT" envDefault:"5s"`
+	// Name of the Table that is created to try if database is writeable
+	HealthCheckTableName string `env:"POSTGRES_HEALTH_CHECK_TABLE_NAME" envDefault:"healthcheck"`
+	// Amount of time to cache the last health check result
+	HealthCheckResultTTL time.Duration `env:"POSTGRES_HEALTH_CHECK_RESULT_TTL" envDefault:"10s"`
+	// Indicator whether write (insert,update,delete) queries should be logged
+	LogWrite bool `env:"POSTGRES_LOG_WRITES" envDefault:"true"`
+	// Indicator whether read (select) queries should be logged
+	LogRead bool `env:"POSTGRES_LOG_READS" envDefault:"false"`
+	// Timeout for socket reads. If reached, commands will fail
+	// with a timeout instead of blocking.
+	ReadTimeout time.Duration `env:"POSTGRES_READ_TIMEOUT" envDefault:"30s"`
+	// Timeout for socket writes. If reached, commands will fail
+	// with a timeout instead of blocking.
+	WriteTimeout time.Duration `env:"POSTGRES_WRITE_TIMEOUT" envDefault:"30s"`
+}
+
+var cfg Config
+
+func init() {
+	prometheus.MustRegister(hooks.MetricQueryTotal)
+	prometheus.MustRegister(hooks.MetricQueryFailed)
+	prometheus.MustRegister(hooks.MetricQueryDurationSeconds)
+	prometheus.MustRegister(hooks.MetricQueryAffectedTotal)
+
+	err := env.Parse(&cfg)
+	if err != nil {
+		log.Fatalf("Failed to parse postgres environment: %v", err)
+	}
+
+	// if the application name is unset infer it from the:
+	// jaeger service name , service name or executable name
+	if cfg.ApplicationName == "-" {
+		names := []string{
+			os.Getenv("JAEGER_SERVICE_NAME"),
+			os.Getenv("SERVICE_NAME"),
+			filepath.Base(os.Args[0]),
+		}
+		for _, name := range names {
+			if name != "" {
+				cfg.ApplicationName = name
+				break
+			}
+		}
+	}
+
+	servicehealthcheck.RegisterHealthCheck("postgresdefault", &HealthCheck{
+		db: NewDB(context.Background()),
+	})
+}
+
+func NewDB(ctx context.Context, options ...ConfigOption) *bun.DB {
+	for _, opt := range options {
+		opt(&cfg)
+	}
+
+	connector := pgdriver.NewConnector(
+		pgdriver.WithAddr(net.JoinHostPort(cfg.Host, strconv.Itoa(cfg.Port))),
+		pgdriver.WithApplicationName(cfg.ApplicationName),
+		pgdriver.WithDatabase(cfg.Database),
+		pgdriver.WithDialTimeout(cfg.DialTimeout),
+		pgdriver.WithPassword(cfg.Password),
+		pgdriver.WithReadTimeout(cfg.ReadTimeout),
+		pgdriver.WithUser(cfg.User),
+		pgdriver.WithWriteTimeout(cfg.WriteTimeout),
+	)
+
+	sqldb := sql.OpenDB(connector)
+	db := bun.NewDB(sqldb, pgdialect.New())
+
+	log.Ctx(ctx).Info().Str("addr", connector.Config().Addr).
+		Str("user", connector.Config().User).
+		Str("database", connector.Config().Database).
+		Str("as", connector.Config().AppName).
+		Msg("PostgreSQL connection pool created")
+
+	// Add hooks
+	db.AddQueryHook(&hooks.TracingHook{})
+	db.AddQueryHook(hooks.NewMetricsHook(cfg.Host, cfg.Database))
+
+	if cfg.LogWrite || cfg.LogRead {
+		db.AddQueryHook(hooks.NewLoggingHook(cfg.LogRead, cfg.LogWrite))
+	} else {
+		log.Ctx(ctx).Warn().Msg("Connection pool has logging queries disabled completely")
+	}
+
+	return db
+}

backend/postgres/health_postgres.go → backend/postgres/health.go

@@ -1,30 +1,33 @@
-// Copyright © 2019 by PACE Telematics GmbH. All rights reserved.
+// Copyright © 2024 by PACE Telematics GmbH. All rights reserved.
 
 package postgres
 
 import (
 	"context"
 	"time"
 
-	"github.com/go-pg/pg/orm"
 	"github.com/pace/bricks/maintenance/health/servicehealthcheck"
+	"github.com/uptrace/bun"
 )
 
 // HealthCheck checks the state of a postgres connection. It must not be changed
 // after it was registered as a health check.
 type HealthCheck struct {
 	state servicehealthcheck.ConnectionState
-	Pool  postgresQueryExecutor
+	db    *bun.DB
 }
 
-type postgresQueryExecutor interface {
-	Exec(ctx context.Context, query interface{}, params ...interface{}) (res orm.Result, err error)
+type healthcheck struct {
+	bun.BaseModel
+
+	OK bool `bun:"column:ok"`
 }
 
 // Init initializes the test table
 func (h *HealthCheck) Init(ctx context.Context) error {
-	_, errWrite := h.Pool.Exec(ctx, `CREATE TABLE IF NOT EXISTS `+cfg.HealthCheckTableName+`(ok boolean);`)
-	return errWrite
+	_, err := h.db.NewCreateTable().Table(cfg.HealthCheckTableName).Model((*healthcheck)(nil)).Exec(ctx)
+
+	return err
 }
 
 // HealthCheck performs the read test on the database. If enabled, it performs a
@@ -36,19 +39,23 @@ func (h *HealthCheck) HealthCheck(ctx context.Context) servicehealthcheck.Health
 	}
 
 	// Readcheck
-	if _, err := h.Pool.Exec(ctx, `SELECT 1;`); err != nil {
+	if _, err := h.db.NewSelect().Column("1").Exec(ctx); err != nil {
 		h.state.SetErrorState(err)
 		return h.state.GetState()
 	}
+
+	hc := &healthcheck{OK: true}
+
 	// writecheck - add Data to configured Table
-	_, err := h.Pool.Exec(ctx, "INSERT INTO "+cfg.HealthCheckTableName+"(ok) VALUES (true);")
+	_, err := h.db.NewInsert().Model(hc).Exec(ctx)
 	if err != nil {
 		h.state.SetErrorState(err)
 		return h.state.GetState()
 	}
+
 	// and while we're at it, check delete as well (so as not to clutter the database
 	// because UPSERT is impractical here
-	_, err = h.Pool.Exec(ctx, "DELETE FROM "+cfg.HealthCheckTableName+";")
+	_, err = h.db.NewDelete().Exec(ctx)
 	if err != nil {
 		h.state.SetErrorState(err)
 		return h.state.GetState()
@@ -60,6 +67,7 @@ func (h *HealthCheck) HealthCheck(ctx context.Context) servicehealthcheck.Health
 
 // CleanUp drops the test table.
 func (h *HealthCheck) CleanUp(ctx context.Context) error {
-	_, err := h.Pool.Exec(ctx, "DROP TABLE IF EXISTS "+cfg.HealthCheckTableName)
+	_, err := h.db.NewDropTable().Table(cfg.HealthCheckTableName).IfExists().Exec(ctx)
+
 	return err
 }