pkg/sqlutil/pg: create package; expand env config; add example relay

Author: jmank88

go.mod

@@ -3,6 +3,7 @@ module github.com/smartcontractkit/chainlink-common
 go 1.23.3
 
 require (
+	github.com/XSAM/otelsql v0.29.0
 	github.com/andybalholm/brotli v1.1.0
 	github.com/atombender/go-jsonschema v0.16.1-0.20240916205339-a74cd4e2851c
 	github.com/bytecodealliance/wasmtime-go/v23 v23.0.0
@@ -20,6 +21,7 @@ require (
 	github.com/hashicorp/go-plugin v1.6.2
 	github.com/iancoleman/strcase v0.3.0
 	github.com/invopop/jsonschema v0.12.0
+	github.com/jackc/pgx/v4 v4.18.3
 	github.com/jmoiron/sqlx v1.4.0
 	github.com/jonboulle/clockwork v0.4.0
 	github.com/jpillora/backoff v1.0.0
@@ -29,6 +31,7 @@ require (
 	github.com/prometheus/client_golang v1.17.0
 	github.com/riferrei/srclient v0.5.4
 	github.com/santhosh-tekuri/jsonschema/v5 v5.2.0
+	github.com/scylladb/go-reflectx v1.0.1
 	github.com/shopspring/decimal v1.4.0
 	github.com/smartcontractkit/grpc-proxy v0.0.0-20240830132753-a7e17fec5ab7
 	github.com/smartcontractkit/libocr v0.0.0-20241007185508-adbe57025f12
@@ -79,6 +82,13 @@ require (
 	github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.1.0 // indirect
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.22.0 // indirect
 	github.com/hashicorp/yamux v0.1.1 // indirect
+	github.com/jackc/chunkreader/v2 v2.0.1 // indirect
+	github.com/jackc/pgconn v1.14.3 // indirect
+	github.com/jackc/pgio v1.0.0 // indirect
+	github.com/jackc/pgpassfile v1.0.0 // indirect
+	github.com/jackc/pgproto3/v2 v2.3.3 // indirect
+	github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a // indirect
+	github.com/jackc/pgtype v1.14.0 // indirect
 	github.com/leodido/go-urn v1.2.0 // indirect
 	github.com/mailru/easyjson v0.7.7 // indirect
 	github.com/mattn/go-colorable v0.1.13 // indirect

go.sum

@@ -0,0 +1,34 @@
+package build
+
+import (
+	"os"
+	"runtime/debug"
+)
+
+// Unset is a sentinel value.
+const Unset = "unset"
+
+// Version and Checksum are set at compile time via build arguments.
+var (
+	// Program is updated to the full main program path if [debug.BuildInfo] is available.
+	Program = os.Args[0]
+	// Version is the semantic version of the build or Unset.
+	Version = Unset
+	// Checksum is the commit hash of the build or Unset.
+	Checksum       = Unset
+	ChecksumPrefix = Unset
+)
+
+func init() {
+	buildInfo, ok := debug.ReadBuildInfo()
+	if ok {
+		Program = buildInfo.Main.Path
+		if Version == Unset && buildInfo.Main.Version != "" {
+			Version = buildInfo.Main.Version
+		}
+		if Checksum == Unset && buildInfo.Main.Sum != "" {
+			Checksum = buildInfo.Main.Sum
+		}
+	}
+	ChecksumPrefix = Checksum[:min(7, len(Checksum))]
+}

pkg/config/build/build.go

@@ -2,13 +2,16 @@ package logger
 
 import (
 	"io"
+	"fmt"
 	"reflect"
 	"testing"
 
 	"go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
 	"go.uber.org/zap/zaptest"
 	"go.uber.org/zap/zaptest/observer"
+
+	"github.com/smartcontractkit/chainlink-common/pkg/config/build"
 )
 
 // Logger is a minimal subset of smartcontractkit/chainlink/core/logger.Logger implemented by go.uber.org/zap.SugaredLogger
@@ -52,9 +55,16 @@ func New() (Logger, error) { return defaultConfig.New() }
 func (c *Config) New() (Logger, error) {
 	return NewWith(func(cfg *zap.Config) {
 		cfg.Level.SetLevel(c.Level)
+		cfg.InitialFields = map[string]interface{}{
+			"version": buildVersion(),
+		}
 	})
 }
 
+func buildVersion() string {
+	return fmt.Sprintf("%s@%s", build.Version, build.ChecksumPrefix)
+}
+
 // NewWith returns a new Logger from a modified [zap.Config].
 func NewWith(cfgFn func(*zap.Config)) (Logger, error) {
 	cfg := zap.NewProductionConfig()
@@ -83,7 +93,7 @@ func Test(tb testing.TB) Logger {
 			zapcore.DebugLevel,
 		),
 	)
-	return &logger{lggr.Sugar()}
+	return &logger{lggr.With(zap.String("version", buildVersion())).Sugar()}
 }
 
 // TestSugared returns a new test SugaredLogger.

pkg/logger/logger.go

@@ -12,8 +12,15 @@ import (
 )
 
 const (
-	envDatabaseURL = "CL_DATABASE_URL"
-	envPromPort    = "CL_PROMETHEUS_PORT"
+	envDatabaseURL                    = "CL_DATABASE_URL"
+	envDatabaseIdleInTxSessionTimeout = "CL_DATABASE_IDLE_IN_TX_SESSION_TIMEOUT"
+	envDatabaseLockTimeout            = "CL_DATABASE_LOCK_TIMEOUT"
+	envDatabaseQueryTimeout           = "CL_DATABASE_QUERY_TIMEOUT"
+	envDatabaseLogSQL                 = "CL_DATABASE_LOG_SQL"
+	envDatabaseMaxOpenConns           = "CL_DATABASE_MAX_OPEN_CONNS"
+	envDatabaseMaxIdleConns           = "CL_DATABASE_MAX_IDLE_CONNS"
+
+	envPromPort = "CL_PROMETHEUS_PORT"
 
 	envTracingEnabled         = "CL_TRACING_ENABLED"
 	envTracingCollectorTarget = "CL_TRACING_COLLECTOR_TARGET"
@@ -36,7 +43,13 @@ const (
 // EnvConfig is the configuration between the application and the LOOP executable. The values
 // are fully resolved and static and passed via the environment.
 type EnvConfig struct {
-	DatabaseURL *url.URL
+	DatabaseURL                    *url.URL
+	DatabaseIdleInTxSessionTimeout time.Duration
+	DatabaseLockTimeout            time.Duration
+	DatabaseQueryTimeout           time.Duration
+	DatabaseLogSQL                 bool
+	DatabaseMaxOpenConns           int
+	DatabaseMaxIdleConns           int
 
 	PrometheusPort int
 
@@ -66,7 +79,14 @@ func (e *EnvConfig) AsCmdEnv() (env []string) {
 
 	if e.DatabaseURL != nil { // optional
 		add(envDatabaseURL, e.DatabaseURL.String())
+		add(envDatabaseIdleInTxSessionTimeout, e.DatabaseIdleInTxSessionTimeout.String())
+		add(envDatabaseLockTimeout, e.DatabaseLockTimeout.String())
+		add(envDatabaseQueryTimeout, e.DatabaseQueryTimeout.String())
+		add(envDatabaseLogSQL, strconv.FormatBool(e.DatabaseLogSQL))
+		add(envDatabaseMaxOpenConns, strconv.Itoa(e.DatabaseMaxOpenConns))
+		add(envDatabaseMaxIdleConns, strconv.Itoa(e.DatabaseMaxIdleConns))
 	}
+
 	add(envPromPort, strconv.Itoa(e.PrometheusPort))
 
 	add(envTracingEnabled, strconv.FormatBool(e.TracingEnabled))
@@ -99,13 +119,44 @@ func (e *EnvConfig) AsCmdEnv() (env []string) {
 
 // parse deserializes environment variables
 func (e *EnvConfig) parse() error {
-	promPortStr := os.Getenv(envPromPort)
 	var err error
-	e.DatabaseURL, err = getDatabaseURL()
+	e.DatabaseURL, err = getEnv(envDatabaseURL, func(s string) (*url.URL, error) {
+		if s == "" { // DatabaseURL is optional
+			return nil, nil
+		}
+		return url.Parse(s)
+	})
 	if err != nil {
-		return fmt.Errorf("failed to parse %s: %w", envDatabaseURL, err)
+		return err
+	}
+	if e.DatabaseURL != nil {
+		e.DatabaseIdleInTxSessionTimeout, err = getEnv(envDatabaseIdleInTxSessionTimeout, time.ParseDuration)
+		if err != nil {
+			return err
+		}
+		e.DatabaseLockTimeout, err = getEnv(envDatabaseLockTimeout, time.ParseDuration)
+		if err != nil {
+			return err
+		}
+		e.DatabaseQueryTimeout, err = getEnv(envDatabaseQueryTimeout, time.ParseDuration)
+		if err != nil {
+			return err
+		}
+		e.DatabaseLogSQL, err = getEnv(envDatabaseLogSQL, strconv.ParseBool)
+		if err != nil {
+			return err
+		}
+		e.DatabaseMaxOpenConns, err = getEnv(envDatabaseMaxOpenConns, strconv.Atoi)
+		if err != nil {
+			return err
+		}
+		e.DatabaseMaxIdleConns, err = getEnv(envDatabaseMaxIdleConns, strconv.Atoi)
+		if err != nil {
+			return err
+		}
 	}
 
+	promPortStr := os.Getenv(envPromPort)
 	e.PrometheusPort, err = strconv.Atoi(promPortStr)
 	if err != nil {
 		return fmt.Errorf("failed to parse %s = %q: %w", envPromPort, promPortStr, err)
@@ -203,16 +254,11 @@ func getFloat64OrZero(envKey string) float64 {
 	return f
 }
 
-// getDatabaseURL parses the CL_DATABASE_URL environment variable.
-func getDatabaseURL() (*url.URL, error) {
-	databaseURL := os.Getenv(envDatabaseURL)
-	if databaseURL == "" {
-		// DatabaseURL is optional
-		return nil, nil
-	}
-	u, err := url.Parse(databaseURL)
+func getEnv[T any](key string, parse func(string) (T, error)) (t T, err error) {
+	v := os.Getenv(key)
+	t, err = parse(v)
 	if err != nil {
-		return nil, fmt.Errorf("invalid %s: %w", envDatabaseURL, err)
+		err = fmt.Errorf("failed to parse %s=%s: %w", key, v, err)
 	}
-	return u, nil
+	return
 }

pkg/loop/config.go

@@ -18,10 +18,18 @@ import (
 
 func TestEnvConfig_parse(t *testing.T) {
 	cases := []struct {
-		name                           string
-		envVars                        map[string]string
-		expectError                    bool
-		expectedDatabaseURL            string
+		name        string
+		envVars     map[string]string
+		expectError bool
+
+		expectedDatabaseURL                    string
+		expectedDatabaseIdleInTxSessionTimeout time.Duration
+		expectedDatabaseLockTimeout            time.Duration
+		expectedDatabaseQueryTimeout           time.Duration
+		expectedDatabaseLogSQL                 bool
+		expectedDatabaseMaxOpenConns           int
+		expectedDatabaseMaxIdleConns           int
+
 		expectedPrometheusPort         int
 		expectedTracingEnabled         bool
 		expectedTracingCollectorTarget string
@@ -31,16 +39,31 @@ func TestEnvConfig_parse(t *testing.T) {
 		{
 			name: "All variables set correctly",
 			envVars: map[string]string{
-				envDatabaseURL:              "postgres://user:password@localhost:5432/db",
+				envDatabaseURL:                    "postgres://user:password@localhost:5432/db",
+				envDatabaseIdleInTxSessionTimeout: "42s",
+				envDatabaseLockTimeout:            "8m",
+				envDatabaseQueryTimeout:           "7s",
+				envDatabaseLogSQL:                 "true",
+				envDatabaseMaxOpenConns:           "9999",
+				envDatabaseMaxIdleConns:           "8080",
+
 				envPromPort:                 "8080",
 				envTracingEnabled:           "true",
 				envTracingCollectorTarget:   "some:target",
 				envTracingSamplingRatio:     "1.0",
 				envTracingTLSCertPath:       "internal/test/fixtures/client.pem",
 				envTracingAttribute + "XYZ": "value",
 			},
-			expectError:                    false,
-			expectedDatabaseURL:            "postgres://user:password@localhost:5432/db",
+			expectError: false,
+
+			expectedDatabaseURL:                    "postgres://user:password@localhost:5432/db",
+			expectedDatabaseIdleInTxSessionTimeout: 42 * time.Second,
+			expectedDatabaseLockTimeout:            8 * time.Minute,
+			expectedDatabaseQueryTimeout:           7 * time.Second,
+			expectedDatabaseLogSQL:                 true,
+			expectedDatabaseMaxOpenConns:           9999,
+			expectedDatabaseMaxIdleConns:           8080,
+
 			expectedPrometheusPort:         8080,
 			expectedTracingEnabled:         true,
 			expectedTracingCollectorTarget: "some:target",
@@ -91,6 +114,25 @@ func TestEnvConfig_parse(t *testing.T) {
 					if config.DatabaseURL.String() != tc.expectedDatabaseURL {
 						t.Errorf("Expected Database URL %s, got %s", tc.expectedDatabaseURL, config.DatabaseURL)
 					}
+					if config.DatabaseIdleInTxSessionTimeout != tc.expectedDatabaseIdleInTxSessionTimeout {
+						t.Errorf("Expected Database idle in tx session timeout %s, got %s", tc.expectedDatabaseIdleInTxSessionTimeout, config.DatabaseIdleInTxSessionTimeout)
+					}
+					if config.DatabaseLockTimeout != tc.expectedDatabaseLockTimeout {
+						t.Errorf("Expected Database lock timeout %s, got %s", tc.expectedDatabaseLockTimeout, config.DatabaseLockTimeout)
+					}
+					if config.DatabaseQueryTimeout != tc.expectedDatabaseQueryTimeout {
+						t.Errorf("Expected Database query timeout %s, got %s", tc.expectedDatabaseQueryTimeout, config.DatabaseQueryTimeout)
+					}
+					if config.DatabaseLogSQL != tc.expectedDatabaseLogSQL {
+						t.Errorf("Expected Database log sql %t, got %t", tc.expectedDatabaseLogSQL, config.DatabaseLogSQL)
+					}
+					if config.DatabaseMaxOpenConns != tc.expectedDatabaseMaxOpenConns {
+						t.Errorf("Expected Database max open conns %d, got %d", tc.expectedDatabaseMaxOpenConns, config.DatabaseMaxOpenConns)
+					}
+					if config.DatabaseMaxIdleConns != tc.expectedDatabaseMaxIdleConns {
+						t.Errorf("Expected Database max idle conns %d, got %d", tc.expectedDatabaseMaxIdleConns, config.DatabaseMaxIdleConns)
+					}
+
 					if config.PrometheusPort != tc.expectedPrometheusPort {
 						t.Errorf("Expected Prometheus port %d, got %d", tc.expectedPrometheusPort, config.PrometheusPort)
 					}