pkg/sqlutil/pg: create package; expand env config

go.mod

@@ -3,6 +3,7 @@ module github.com/smartcontractkit/chainlink-common
 go 1.21
 
 require (
+	github.com/XSAM/otelsql v0.29.0
 	github.com/confluentinc/confluent-kafka-go/v2 v2.3.0
 	github.com/ethereum/go-ethereum v1.13.8
 	github.com/fxamacker/cbor/v2 v2.5.0
@@ -12,25 +13,27 @@ require (
 	github.com/hashicorp/consul/sdk v0.16.0
 	github.com/hashicorp/go-hclog v1.5.0
 	github.com/hashicorp/go-plugin v1.6.0
+	github.com/jackc/pgx/v4 v4.18.3
 	github.com/jmoiron/sqlx v1.3.5
 	github.com/jonboulle/clockwork v0.4.0
 	github.com/jpillora/backoff v1.0.0
-	github.com/lib/pq v1.2.0
+	github.com/lib/pq v1.10.2
 	github.com/linkedin/goavro/v2 v2.12.0
 	github.com/mitchellh/mapstructure v1.5.0
 	github.com/mwitkow/grpc-proxy v0.0.0-20230212185441-f345521cb9c9
 	github.com/pelletier/go-toml/v2 v2.1.1
 	github.com/pkg/errors v0.9.1
 	github.com/prometheus/client_golang v1.17.0
 	github.com/riferrei/srclient v0.5.4
+	github.com/scylladb/go-reflectx v1.0.1
 	github.com/shopspring/decimal v1.3.1
 	github.com/smartcontractkit/libocr v0.0.0-20240326191951-2bbe9382d052
 	github.com/stretchr/testify v1.9.0
 	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.45.0
-	go.opentelemetry.io/otel v1.19.0
+	go.opentelemetry.io/otel v1.24.0
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.18.0
-	go.opentelemetry.io/otel/sdk v1.19.0
-	go.opentelemetry.io/otel/trace v1.19.0
+	go.opentelemetry.io/otel/sdk v1.24.0
+	go.opentelemetry.io/otel/trace v1.24.0
 	go.uber.org/goleak v1.2.1
 	go.uber.org/zap v1.26.0
 	golang.org/x/exp v0.0.0-20231110203233-9a3e6036ecaa
@@ -46,15 +49,21 @@ require (
 	github.com/cespare/xxhash/v2 v2.2.0 // indirect
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
 	github.com/fatih/color v1.14.1 // indirect
-	github.com/go-logr/logr v1.2.4 // indirect
+	github.com/go-logr/logr v1.4.1 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
 	github.com/golang/protobuf v1.5.3 // indirect
 	github.com/golang/snappy v0.0.5-0.20220116011046-fa5810519dcb // indirect
-	github.com/google/go-cmp v0.6.0 // indirect
 	github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.0.0-rc.3 // indirect
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.16.0 // indirect
 	github.com/hashicorp/yamux v0.1.1 // indirect
 	github.com/holiman/uint256 v1.2.4 // 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/mattn/go-colorable v0.1.13 // indirect
 	github.com/mattn/go-isatty v0.0.17 // indirect
 	github.com/matttproud/golang_protobuf_extensions v1.0.4 // indirect
@@ -69,13 +78,13 @@ require (
 	github.com/stretchr/objx v0.5.2 // indirect
 	github.com/x448/float16 v0.8.4 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.18.0 // indirect
-	go.opentelemetry.io/otel/metric v1.19.0 // indirect
+	go.opentelemetry.io/otel/metric v1.24.0 // indirect
 	go.opentelemetry.io/proto/otlp v1.0.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
-	golang.org/x/crypto v0.18.0 // indirect
-	golang.org/x/net v0.20.0 // indirect
+	golang.org/x/crypto v0.20.0 // indirect
+	golang.org/x/net v0.21.0 // indirect
 	golang.org/x/sync v0.6.0 // indirect
-	golang.org/x/sys v0.16.0 // indirect
+	golang.org/x/sys v0.17.0 // indirect
 	golang.org/x/text v0.14.0 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20230711160842-782d3b101e98 // indirect
 	google.golang.org/genproto/googleapis/rpc v0.0.0-20230711160842-782d3b101e98 // indirect

pkg/config/static/static.go

@@ -0,0 +1,49 @@
+package static
+
+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
+)
+
+func init() {
+	buildInfo, ok := debug.ReadBuildInfo()
+	if ok {
+		Program = buildInfo.Main.Path
+		if Version == "" {
+			Version = buildInfo.Main.Version
+		}
+		if Checksum == "" {
+			Checksum = buildInfo.Main.Sum
+		}
+	}
+}
+
+// Short returns a 7-character sha prefix and version, or Unset if blank.
+func Short() (shaPre string, ver string) {
+	return short(Checksum, Version)
+}
+
+func short(sha, ver string) (string, string) {
+	if sha == "" {
+		sha = Unset
+	} else if len(sha) > 7 {
+		sha = sha[:7]
+	}
+	if ver == "" {
+		ver = Unset
+	}
+	return sha, ver
+}

pkg/logger/logger.go

@@ -1,13 +1,16 @@
 package logger
 
 import (
+	"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/static"
 )
 
 // Logger is a minimal subset of smartcontractkit/chainlink/core/logger.Logger implemented by go.uber.org/zap.SugaredLogger
@@ -51,9 +54,17 @@ 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": staticShortVerSha(),
+		}
 	})
 }
 
+func staticShortVerSha() string {
+	sha, ver := static.Short() //TODO rename?
+	return fmt.Sprintf("%s@%s", ver, sha)
+}
+
 // NewWith returns a new Logger from a modified [zap.Config].
 func NewWith(cfgFn func(*zap.Config)) (Logger, error) {
 	cfg := zap.NewProductionConfig()
@@ -67,7 +78,7 @@ func NewWith(cfgFn func(*zap.Config)) (Logger, error) {
 
 // Test returns a new test Logger for tb.
 func Test(tb testing.TB) Logger {
-	return &logger{zaptest.NewLogger(tb).Sugar()}
+	return &logger{zaptest.NewLogger(tb).With(zap.String("version", staticShortVerSha())).Sugar()}
 }
 
 // TestSugared returns a new test SugaredLogger.

pkg/loop/config.go

@@ -6,10 +6,18 @@ import (
 	"os"
 	"strconv"
 	"strings"
+	"time"
 )
 
 const (
-	envDatabaseURL            = "CL_DATABASE_URL"
+	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"
@@ -21,7 +29,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
 
@@ -45,6 +59,12 @@ func (e *EnvConfig) AsCmdEnv() (env []string) {
 	// DatabaseURL is optional
 	if e.DatabaseURL != nil {
 		injectEnv[envDatabaseURL] = e.DatabaseURL.String()
+		injectEnv[envDatabaseIdleInTxSessionTimeout] = e.DatabaseIdleInTxSessionTimeout.String()
+		injectEnv[envDatabaseLockTimeout] = e.DatabaseLockTimeout.String()
+		injectEnv[envDatabaseQueryTimeout] = e.DatabaseQueryTimeout.String()
+		injectEnv[envDatabaseLogSQL] = strconv.FormatBool(e.DatabaseLogSQL)
+		injectEnv[envDatabaseMaxOpenConns] = strconv.Itoa(e.DatabaseMaxOpenConns)
+		injectEnv[envDatabaseMaxIdleConns] = strconv.Itoa(e.DatabaseMaxIdleConns)
 	}
 
 	for k, v := range e.TracingAttributes {
@@ -59,13 +79,39 @@ 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()
 	if err != nil {
-		return fmt.Errorf("failed to parse %s: %q", envDatabaseURL, err)
+		return err
+	}
+	if e.DatabaseURL != nil {
+		e.DatabaseIdleInTxSessionTimeout, err = getDatabaseIdleInTxSessionTimeout()
+		if err != nil {
+			return err
+		}
+		e.DatabaseLockTimeout, err = getDatabaseLockTimeout()
+		if err != nil {
+			return err
+		}
+		e.DatabaseQueryTimeout, err = getDatabaseQueryTimeout()
+		if err != nil {
+			return err
+		}
+		e.DatabaseLogSQL, err = getDatabaseLogSQL()
+		if err != nil {
+			return err
+		}
+		e.DatabaseMaxOpenConns, err = getDatabaseMaxOpenConns()
+		if err != nil {
+			return err
+		}
+		e.DatabaseMaxIdleConns, err = getDatabaseMaxIdleConns()
+		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)
@@ -150,3 +196,27 @@ func getDatabaseURL() (*url.URL, error) {
 	}
 	return u, nil
 }
+
+func getDatabaseIdleInTxSessionTimeout() (time.Duration, error) {
+	return time.ParseDuration(os.Getenv(envDatabaseIdleInTxSessionTimeout))
+}
+
+func getDatabaseLockTimeout() (time.Duration, error) {
+	return time.ParseDuration(os.Getenv(envDatabaseLockTimeout))
+}
+
+func getDatabaseQueryTimeout() (time.Duration, error) {
+	return time.ParseDuration(os.Getenv(envDatabaseQueryTimeout))
+}
+
+func getDatabaseLogSQL() (bool, error) {
+	return strconv.ParseBool(os.Getenv(envDatabaseLogSQL))
+}
+
+func getDatabaseMaxOpenConns() (int, error) {
+	return strconv.Atoi(os.Getenv(envDatabaseMaxOpenConns))
+}
+
+func getDatabaseMaxIdleConns() (int, error) {
+	return strconv.Atoi(os.Getenv(envDatabaseMaxIdleConns))
+}

pkg/loop/config_test.go

@@ -5,17 +5,26 @@ import (
 	"strconv"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
 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
@@ -25,16 +34,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",
@@ -85,6 +109,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)
 					}