diff --git a/internal/collector/pgbouncer.go b/internal/collector/pgbouncer.go index d59d936669..0ddf880d66 100644 --- a/internal/collector/pgbouncer.go +++ b/internal/collector/pgbouncer.go @@ -191,8 +191,8 @@ func EnablePgBouncerMetrics(ctx context.Context, inCluster *v1beta1.PostgresClus config.Receivers[SqlQuery] = map[string]any{ "driver": "postgres", "datasource": fmt.Sprintf( - `host=localhost dbname=pgbouncer port=5432 user=%s password=${env:PGPASSWORD}`, - sqlQueryUsername), + `host=localhost dbname=pgbouncer port=%d user=%s password=${env:PGPASSWORD}`, + *inCluster.Spec.Proxy.PGBouncer.Port, sqlQueryUsername), "queries": slices.Clone(pgBouncerMetricsQueries), } diff --git a/internal/collector/pgbouncer_test.go b/internal/collector/pgbouncer_test.go index f0b1e2035c..1e4cdc485f 100644 --- a/internal/collector/pgbouncer_test.go +++ b/internal/collector/pgbouncer_test.go @@ -11,7 +11,9 @@ import ( "gotest.tools/v3/assert" "github.com/crunchydata/postgres-operator/internal/feature" + "github.com/crunchydata/postgres-operator/internal/initialize" "github.com/crunchydata/postgres-operator/internal/naming" + "github.com/crunchydata/postgres-operator/internal/testing/cmp" "github.com/crunchydata/postgres-operator/internal/testing/require" "github.com/crunchydata/postgres-operator/pkg/apis/postgres-operator.crunchydata.com/v1beta1" ) @@ -337,6 +339,10 @@ func TestEnablePgBouncerMetrics(t *testing.T) { config := NewConfig(nil) cluster := new(v1beta1.PostgresCluster) + cluster.Spec.Proxy = &v1beta1.PostgresProxySpec{ + PGBouncer: &v1beta1.PGBouncerPodSpec{}, + } + cluster.Spec.Proxy.PGBouncer.Default() // Sets Port to 5432 require.UnmarshalInto(t, &cluster.Spec, `{ instrumentation: {} }`) @@ -513,6 +519,10 @@ service: config := NewConfig(testInstrumentationSpec()) cluster := new(v1beta1.PostgresCluster) + cluster.Spec.Proxy = &v1beta1.PostgresProxySpec{ + PGBouncer: &v1beta1.PGBouncerPodSpec{}, + } + cluster.Spec.Proxy.PGBouncer.Default() // Sets Port to 5432 cluster.Spec.Instrumentation = testInstrumentationSpec() EnablePgBouncerMetrics(ctx, cluster, config, "test_user") @@ -683,4 +693,30 @@ service: `) }) + + t.Run("CustomPort", func(t *testing.T) { + gate := feature.NewGate() + assert.NilError(t, gate.SetFromMap(map[string]bool{ + feature.OpenTelemetryMetrics: true, + })) + ctx := feature.NewContext(context.Background(), gate) + + config := NewConfig(nil) + cluster := new(v1beta1.PostgresCluster) + cluster.Spec.Proxy = &v1beta1.PostgresProxySpec{ + PGBouncer: &v1beta1.PGBouncerPodSpec{}, + } + cluster.Spec.Proxy.PGBouncer.Default() + cluster.Spec.Proxy.PGBouncer.Port = initialize.Int32(6432) // Override default + require.UnmarshalInto(t, &cluster.Spec, `{ + instrumentation: {} + }`) + + EnablePgBouncerMetrics(ctx, cluster, config, "test_user") + + // Verify datasource contains correct port + receiver := config.Receivers["sqlquery"].(map[string]any) + datasource := receiver["datasource"].(string) + assert.Assert(t, cmp.Contains(datasource, "port=6432")) + }) } diff --git a/internal/collector/postgres_metrics.go b/internal/collector/postgres_metrics.go index 7fc46cf943..17735fb203 100644 --- a/internal/collector/postgres_metrics.go +++ b/internal/collector/postgres_metrics.go @@ -59,6 +59,21 @@ type metric struct { ValueType string `json:"value_type,omitempty"` } +// updateServerAttribute updates the server static_attribute in all metrics +// to use the actual Postgres port from the cluster spec. +func updateServerAttribute(metrics []queryMetrics, port int32) { + serverValue := fmt.Sprintf("localhost:%d", port) + for i := range metrics { + for j := range metrics[i].Metrics { + if metrics[i].Metrics[j].StaticAttributes != nil { + if _, ok := metrics[i].Metrics[j].StaticAttributes["server"]; ok { + metrics[i].Metrics[j].StaticAttributes["server"] = serverValue + } + } + } + } +} + func EnablePostgresMetrics(ctx context.Context, inCluster *v1beta1.PostgresCluster, config *Config) { if OpenTelemetryMetricsEnabled(ctx, inCluster) { log := logging.FromContext(ctx) @@ -96,54 +111,47 @@ func EnablePostgresMetrics(ctx context.Context, inCluster *v1beta1.PostgresClust } } + // Always parse JSON to update static_attributes.server with actual port + var fiveSecondMetricsArr []queryMetrics + if err := json.Unmarshal(fiveSecondMetricsClone, &fiveSecondMetricsArr); err != nil { + log.Error(err, "error parsing five second postgres metrics") + } + updateServerAttribute(fiveSecondMetricsArr, *inCluster.Spec.Port) + + var fiveMinuteMetricsArr []queryMetrics + if err := json.Unmarshal(fiveMinuteMetricsClone, &fiveMinuteMetricsArr); err != nil { + log.Error(err, "error parsing five minute postgres metrics") + } + updateServerAttribute(fiveMinuteMetricsArr, *inCluster.Spec.Port) + + var fiveMinutePerDBMetricsArr []queryMetrics + if err := json.Unmarshal(fiveMinutePerDBMetricsClone, &fiveMinutePerDBMetricsArr); err != nil { + log.Error(err, "error parsing per-db postgres metrics") + } + updateServerAttribute(fiveMinutePerDBMetricsArr, *inCluster.Spec.Port) + // Remove any queries that user has specified in the spec if inCluster.Spec.Instrumentation != nil && inCluster.Spec.Instrumentation.Metrics != nil && inCluster.Spec.Instrumentation.Metrics.CustomQueries != nil && inCluster.Spec.Instrumentation.Metrics.CustomQueries.Remove != nil { - // Convert json to array of queryMetrics objects - var fiveSecondMetricsArr []queryMetrics - err := json.Unmarshal(fiveSecondMetricsClone, &fiveSecondMetricsArr) - if err != nil { - log.Error(err, "error compiling five second postgres metrics") - } - - // Remove any specified metrics from the five second metrics fiveSecondMetricsArr = removeMetricsFromQueries( inCluster.Spec.Instrumentation.Metrics.CustomQueries.Remove, fiveSecondMetricsArr) - - // Convert json to array of queryMetrics objects - var fiveMinuteMetricsArr []queryMetrics - err = json.Unmarshal(fiveMinuteMetricsClone, &fiveMinuteMetricsArr) - if err != nil { - log.Error(err, "error compiling five minute postgres metrics") - } - - // Remove any specified metrics from the five minute metrics fiveMinuteMetricsArr = removeMetricsFromQueries( inCluster.Spec.Instrumentation.Metrics.CustomQueries.Remove, fiveMinuteMetricsArr) - - // Convert json to array of queryMetrics objects - var fiveMinutePerDBMetricsArr []queryMetrics - err = json.Unmarshal(fiveMinutePerDBMetricsClone, &fiveMinutePerDBMetricsArr) - if err != nil { - log.Error(err, "error compiling per-db postgres metrics") - } - - // Remove any specified metrics from the five minute per-db metrics fiveMinutePerDBMetricsArr = removeMetricsFromQueries( inCluster.Spec.Instrumentation.Metrics.CustomQueries.Remove, fiveMinutePerDBMetricsArr) - - // Convert back to json data - // The error return value can be ignored as the errchkjson linter - // deems the []queryMetrics to be a safe argument: - // https://github.com/breml/errchkjson - fiveSecondMetricsClone, _ = json.Marshal(fiveSecondMetricsArr) - fiveMinuteMetricsClone, _ = json.Marshal(fiveMinuteMetricsArr) - fiveMinutePerDBMetricsClone, _ = json.Marshal(fiveMinutePerDBMetricsArr) } + // Always marshal back to JSON + // The error return value can be ignored as the errchkjson linter + // deems the []queryMetrics to be a safe argument: + // https://github.com/breml/errchkjson + fiveSecondMetricsClone, _ = json.Marshal(fiveSecondMetricsArr) + fiveMinuteMetricsClone, _ = json.Marshal(fiveMinuteMetricsArr) + fiveMinutePerDBMetricsClone, _ = json.Marshal(fiveMinutePerDBMetricsArr) + // Add Prometheus exporter config.Exporters[Prometheus] = map[string]any{ "endpoint": "0.0.0.0:" + strconv.Itoa(PrometheusPort), @@ -152,8 +160,8 @@ func EnablePostgresMetrics(ctx context.Context, inCluster *v1beta1.PostgresClust config.Receivers[FiveSecondSqlQuery] = map[string]any{ "driver": "postgres", "datasource": fmt.Sprintf( - `host=localhost dbname=postgres port=5432 user=%s password=${env:PGPASSWORD}`, - MonitoringUser), + `host=localhost dbname=postgres port=%d user=%s password=${env:PGPASSWORD}`, + *inCluster.Spec.Port, MonitoringUser), "collection_interval": "5s", // Give Postgres time to finish setup. "initial_delay": "15s", @@ -163,8 +171,8 @@ func EnablePostgresMetrics(ctx context.Context, inCluster *v1beta1.PostgresClust config.Receivers[FiveMinuteSqlQuery] = map[string]any{ "driver": "postgres", "datasource": fmt.Sprintf( - `host=localhost dbname=postgres port=5432 user=%s password=${env:PGPASSWORD}`, - MonitoringUser), + `host=localhost dbname=postgres port=%d user=%s password=${env:PGPASSWORD}`, + *inCluster.Spec.Port, MonitoringUser), "collection_interval": "300s", // Give Postgres time to finish setup. "initial_delay": "15s", @@ -209,8 +217,9 @@ func EnablePostgresMetrics(ctx context.Context, inCluster *v1beta1.PostgresClust config.Receivers[receiverName] = map[string]any{ "driver": "postgres", "datasource": fmt.Sprintf( - `host=localhost dbname=%s port=5432 user=%s password=${env:PGPASSWORD}`, + `host=localhost dbname=%s port=%d user=%s password=${env:PGPASSWORD}`, db, + *inCluster.Spec.Port, MonitoringUser), "collection_interval": querySet.CollectionInterval, // Give Postgres time to finish setup. @@ -234,8 +243,9 @@ func EnablePostgresMetrics(ctx context.Context, inCluster *v1beta1.PostgresClust config.Receivers[receiverName] = map[string]any{ "driver": "postgres", "datasource": fmt.Sprintf( - `host=localhost dbname=%s port=5432 user=%s password=${env:PGPASSWORD}`, + `host=localhost dbname=%s port=%d user=%s password=${env:PGPASSWORD}`, db, + *inCluster.Spec.Port, MonitoringUser), "collection_interval": "5m", // Give Postgres time to finish setup. diff --git a/internal/collector/postgres_test.go b/internal/collector/postgres_test.go index 3c6d44a63e..be3c1987df 100644 --- a/internal/collector/postgres_test.go +++ b/internal/collector/postgres_test.go @@ -6,11 +6,13 @@ package collector import ( "context" + "encoding/json" "testing" "gotest.tools/v3/assert" "github.com/crunchydata/postgres-operator/internal/feature" + "github.com/crunchydata/postgres-operator/internal/initialize" "github.com/crunchydata/postgres-operator/internal/postgres" "github.com/crunchydata/postgres-operator/internal/testing/cmp" "github.com/crunchydata/postgres-operator/internal/testing/require" @@ -903,6 +905,7 @@ func TestEnablePostgresMetrics(t *testing.T) { cluster := new(v1beta1.PostgresCluster) cluster.Spec.PostgresVersion = 99 + cluster.Default() // Sets Port to 5432 require.UnmarshalInto(t, &cluster.Spec, `{ instrumentation: {} }`) @@ -967,6 +970,7 @@ service: cluster := new(v1beta1.PostgresCluster) cluster.Spec.PostgresVersion = 99 + cluster.Default() // Sets Port to 5432 cluster.Spec.Instrumentation = testInstrumentationSpec() config := NewConfig(cluster.Spec.Instrumentation) @@ -1024,6 +1028,34 @@ service: - sqlquery/300s `) }) + + t.Run("CustomPort", func(t *testing.T) { + gate := feature.NewGate() + assert.NilError(t, gate.SetFromMap(map[string]bool{ + feature.OpenTelemetryMetrics: true, + })) + ctx := feature.NewContext(context.Background(), gate) + + cluster := new(v1beta1.PostgresCluster) + cluster.Spec.PostgresVersion = 99 + cluster.Default() + cluster.Spec.Port = initialize.Int32(5433) // Override default + require.UnmarshalInto(t, &cluster.Spec, `{ + instrumentation: {} + }`) + + config := NewConfig(nil) + EnablePostgresMetrics(ctx, cluster, config) + + // Verify datasource contains correct port + receiver := config.Receivers["sqlquery/5s"].(map[string]any) + datasource := receiver["datasource"].(string) + assert.Assert(t, cmp.Contains(datasource, "port=5433")) + + // Verify static_attributes.server contains correct port + queries := receiver["queries"].(json.RawMessage) + assert.Assert(t, cmp.Contains(string(queries), `"server":"localhost:5433"`)) + }) } func TestPostgresParameters(t *testing.T) {