If user has provided custom queries, add the queries file to the volume and create a new sqlquery receiver.

dsessler7 · dsessler7 · commit 4ebde146125b · 2025-03-06T13:01:29.000-08:00
Pass PodTemplateSpec to collector.AddToPod and add a collector label for discovery purposes.
diff --git a/internal/collector/instance.go b/internal/collector/instance.go
@@ -43,7 +43,7 @@ func AddToPod(
 	spec *v1beta1.InstrumentationSpec,
 	pullPolicy corev1.PullPolicy,
 	inInstanceConfigMap *corev1.ConfigMap,
-	outPod *corev1.PodSpec,
+	template *corev1.PodTemplateSpec,
 	volumeMounts []corev1.VolumeMount,
 	sqlQueryPassword string,
 	logDirectories []string,
@@ -76,13 +76,27 @@ func AddToPod(
 		}},
 	}
 
-	// If the user has specified files to be mounted in the spec, add them to the projected config volume
-	if spec != nil && spec.Config != nil && spec.Config.Files != nil {
-		configVolume.Projected.Sources = append(configVolume.Projected.Sources, spec.Config.Files...)
+	// If the user has specified files to be mounted in the spec, add them to
+	// the projected config volume
+	if spec.Config != nil && spec.Config.Files != nil {
+		configVolume.Projected.Sources = append(configVolume.Projected.Sources,
+			spec.Config.Files...)
+	}
+
+	// If the user has specified custom queries to add, put the queries file(s)
+	// in the projected config volume
+	if spec.Metrics != nil && spec.Metrics.CustomQueries != nil &&
+		spec.Metrics.CustomQueries.Add != nil {
+		for _, querySet := range spec.Metrics.CustomQueries.Add {
+			projection := querySet.Queries.AsProjection(querySet.Name +
+				"/" + querySet.Queries.Key)
+			configVolume.Projected.Sources = append(configVolume.Projected.Sources,
+				corev1.VolumeProjection{ConfigMap: &projection})
+		}
 	}
 
 	// Add configVolume to the pod's volumes
-	outPod.Volumes = append(outPod.Volumes, configVolume)
+	template.Spec.Volumes = append(template.Spec.Volumes, configVolume)
 
 	// Create collector container
 	container := corev1.Container{
@@ -136,18 +150,22 @@ func AddToPod(
 			}},
 		}
 		container.VolumeMounts = append(container.VolumeMounts, logrotateConfigVolumeMount)
-		outPod.Volumes = append(outPod.Volumes, logrotateConfigVolume)
+		template.Spec.Volumes = append(template.Spec.Volumes, logrotateConfigVolume)
 	}
 
 	if feature.Enabled(ctx, feature.OpenTelemetryMetrics) {
 		container.Ports = []corev1.ContainerPort{{
-			ContainerPort: int32(8889),
+			ContainerPort: int32(PrometheusPort),
 			Name:          "otel-metrics",
 			Protocol:      corev1.ProtocolTCP,
 		}}
 	}
 
-	outPod.Containers = append(outPod.Containers, container)
+	template.Spec.Containers = append(template.Spec.Containers, container)
+
+	// add the proper label to support Pod discovery by Prometheus
+	initialize.Labels(template)
+	template.Labels[naming.LabelCollectorDiscovery] = "true"
 }
 
 // startCommand generates the command script used by the collector container
diff --git a/internal/collector/naming.go b/internal/collector/naming.go
@@ -10,6 +10,7 @@ const LogsBatchProcessor = "batch/logs"
 const OneSecondBatchProcessor = "batch/1s"
 const SubSecondBatchProcessor = "batch/200ms"
 const Prometheus = "prometheus"
+const PrometheusPort = 9187
 const PGBouncerMetrics = "metrics/pgbouncer"
 const PostgresMetrics = "metrics/postgres"
 const PatroniMetrics = "metrics/patroni"
diff --git a/internal/collector/patroni.go b/internal/collector/patroni.go
@@ -7,6 +7,7 @@ package collector
 import (
 	"context"
 	"slices"
+	"strconv"
 
 	"github.com/crunchydata/postgres-operator/internal/feature"
 	"github.com/crunchydata/postgres-operator/internal/naming"
@@ -136,7 +137,7 @@ func EnablePatroniMetrics(ctx context.Context,
 	if feature.Enabled(ctx, feature.OpenTelemetryMetrics) {
 		// Add Prometheus exporter
 		outConfig.Exporters[Prometheus] = map[string]any{
-			"endpoint": "0.0.0.0:9187",
+			"endpoint": "0.0.0.0:" + strconv.Itoa(PrometheusPort),
 		}
 
 		// Add Prometheus Receiver
diff --git a/internal/collector/pgbouncer.go b/internal/collector/pgbouncer.go
@@ -10,6 +10,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"slices"
+	"strconv"
 
 	"github.com/crunchydata/postgres-operator/internal/feature"
 	"github.com/crunchydata/postgres-operator/internal/naming"
@@ -174,7 +175,7 @@ func EnablePgBouncerMetrics(ctx context.Context, config *Config, sqlQueryUsernam
 	if feature.Enabled(ctx, feature.OpenTelemetryMetrics) {
 		// Add Prometheus exporter
 		config.Exporters[Prometheus] = map[string]any{
-			"endpoint": "0.0.0.0:9187",
+			"endpoint": "0.0.0.0:" + strconv.Itoa(PrometheusPort),
 		}
 
 		// Add SqlQuery Receiver
diff --git a/internal/collector/postgres_metrics.go b/internal/collector/postgres_metrics.go
@@ -10,6 +10,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"slices"
+	"strconv"
 
 	"github.com/crunchydata/postgres-operator/internal/feature"
 	"github.com/crunchydata/postgres-operator/internal/pgmonitor"
@@ -56,7 +57,7 @@ func EnablePostgresMetrics(ctx context.Context, inCluster *v1beta1.PostgresClust
 
 		// Add Prometheus exporter
 		config.Exporters[Prometheus] = map[string]any{
-			"endpoint": "0.0.0.0:9187",
+			"endpoint": "0.0.0.0:" + strconv.Itoa(PrometheusPort),
 		}
 
 		config.Receivers[FiveSecondSqlQuery] = map[string]any{
@@ -76,6 +77,7 @@ func EnablePostgresMetrics(ctx context.Context, inCluster *v1beta1.PostgresClust
 			"initial_delay": "10s",
 			"queries":       slices.Clone(fiveMinuteMetrics),
 		}
+
 		// Add Metrics Pipeline
 		config.Pipelines[PostgresMetrics] = Pipeline{
 			Receivers: []ComponentID{FiveSecondSqlQuery, FiveMinuteSqlQuery},
@@ -85,6 +87,30 @@ func EnablePostgresMetrics(ctx context.Context, inCluster *v1beta1.PostgresClust
 			},
 			Exporters: []ComponentID{Prometheus},
 		}
+
+		// Add custom queries if they are defined in the spec
+		if inCluster.Spec.Instrumentation != nil &&
+			inCluster.Spec.Instrumentation.Metrics != nil &&
+			inCluster.Spec.Instrumentation.Metrics.CustomQueries != nil &&
+			inCluster.Spec.Instrumentation.Metrics.CustomQueries.Add != nil {
+			for _, querySet := range inCluster.Spec.Instrumentation.Metrics.CustomQueries.Add {
+				// Create a receiver for the query set
+				receiverName := "sqlquery/" + querySet.Name
+				config.Receivers[receiverName] = map[string]any{
+					"driver":              "postgres",
+					"datasource":          fmt.Sprintf(`host=localhost dbname=postgres port=5432 user=%s password=${env:PGPASSWORD}`, pgmonitor.MonitoringUser),
+					"collection_interval": querySet.CollectionInterval,
+					// Give Postgres time to finish setup.
+					"initial_delay": "10s",
+					"queries":       "${file:/etc/otel-collector/" + querySet.Name + "/" + querySet.Queries.Key + "}",
+				}
+
+				// Add the receiver to the pipeline
+				pipeline := config.Pipelines[PostgresMetrics]
+				pipeline.Receivers = append(pipeline.Receivers, receiverName)
+				config.Pipelines[PostgresMetrics] = pipeline
+			}
+		}
 	}
 }
 
diff --git a/internal/controller/postgrescluster/instance.go b/internal/controller/postgrescluster/instance.go
@@ -1220,7 +1220,7 @@ func (r *Reconciler) reconcileInstance(
 
 		// For now, we are not using logrotate to rotate postgres or patroni logs
 		// but we are using it for pgbackrest logs in the postgres pod
-		collector.AddToPod(ctx, cluster.Spec.Instrumentation, cluster.Spec.ImagePullPolicy, instanceConfigMap, &instance.Spec.Template.Spec,
+		collector.AddToPod(ctx, cluster.Spec.Instrumentation, cluster.Spec.ImagePullPolicy, instanceConfigMap, &instance.Spec.Template,
 			[]corev1.VolumeMount{postgres.DataVolumeMount()}, pgPassword,
 			[]string{naming.PGBackRestPGDataLogPath}, true)
 	}
diff --git a/internal/controller/postgrescluster/pgbackrest.go b/internal/controller/postgrescluster/pgbackrest.go
@@ -697,7 +697,7 @@ func (r *Reconciler) generateRepoHostIntent(ctx context.Context, postgresCluster
 		if postgresCluster.Spec.Instrumentation != nil && feature.Enabled(ctx, feature.OpenTelemetryLogs) {
 			collector.AddToPod(ctx, postgresCluster.Spec.Instrumentation, postgresCluster.Spec.ImagePullPolicy,
 				&corev1.ConfigMap{ObjectMeta: naming.PGBackRestConfig(postgresCluster)},
-				&repo.Spec.Template.Spec, []corev1.VolumeMount{}, "",
+				&repo.Spec.Template, []corev1.VolumeMount{}, "",
 				[]string{pgBackRestLogPath}, true)
 
 			containersToAdd = append(containersToAdd, naming.ContainerCollector)
diff --git a/internal/controller/postgrescluster/pgbouncer.go b/internal/controller/postgrescluster/pgbouncer.go
@@ -472,7 +472,7 @@ func (r *Reconciler) generatePGBouncerDeployment(
 	err := errors.WithStack(r.setControllerReference(cluster, deploy))
 
 	if err == nil {
-		pgbouncer.Pod(ctx, cluster, configmap, primaryCertificate, secret, &deploy.Spec.Template.Spec)
+		pgbouncer.Pod(ctx, cluster, configmap, primaryCertificate, secret, &deploy.Spec.Template)
 	}
 
 	// Add tmp directory and volume for log files
diff --git a/internal/controller/standalone_pgadmin/statefulset.go b/internal/controller/standalone_pgadmin/statefulset.go
@@ -132,7 +132,7 @@ func statefulset(
 		}
 
 		collector.AddToPod(ctx, pgadmin.Spec.Instrumentation, pgadmin.Spec.ImagePullPolicy,
-			configmap, &sts.Spec.Template.Spec, volumeMounts, "", []string{}, false)
+			configmap, &sts.Spec.Template, volumeMounts, "", []string{}, false)
 	}
 
 	return sts
diff --git a/internal/naming/labels.go b/internal/naming/labels.go
@@ -40,6 +40,10 @@ const (
 	// LabelMovePGWalDir is used to identify the Job that moves an existing pg_wal directory.
 	LabelMovePGWalDir = labelPrefix + "move-pgwal-dir"
 
+	// LabelCollectorDiscovery is added to Pods running the OpenTelemetry "collector"
+	// container to support discovery by Prometheus
+	LabelCollectorDiscovery = labelPrefix + "crunchy-otel-collector"
+
 	// LabelPGBackRest is used to indicate that a resource is for pgBackRest
 	LabelPGBackRest = labelPrefix + "pgbackrest"
 
diff --git a/internal/pgbouncer/reconcile.go b/internal/pgbouncer/reconcile.go
@@ -110,7 +110,7 @@ func Pod(
 	inConfigMap *corev1.ConfigMap,
 	inPostgreSQLCertificate *corev1.SecretProjection,
 	inSecret *corev1.Secret,
-	outPod *corev1.PodSpec,
+	template *corev1.PodTemplateSpec,
 ) {
 	if inCluster.Spec.Proxy == nil || inCluster.Spec.Proxy.PGBouncer == nil {
 		// PgBouncer is disabled; there is nothing to do.
@@ -179,20 +179,20 @@ func Pod(
 		reloader.Resources = *inCluster.Spec.Proxy.PGBouncer.Sidecars.PGBouncerConfig.Resources
 	}
 
-	outPod.Containers = []corev1.Container{container, reloader}
+	template.Spec.Containers = []corev1.Container{container, reloader}
 
 	// If the PGBouncerSidecars feature gate is enabled and custom pgBouncer
 	// sidecars are defined, add the defined container to the Pod.
 	if feature.Enabled(ctx, feature.PGBouncerSidecars) &&
 		inCluster.Spec.Proxy.PGBouncer.Containers != nil {
-		outPod.Containers = append(outPod.Containers, inCluster.Spec.Proxy.PGBouncer.Containers...)
+		template.Spec.Containers = append(template.Spec.Containers, inCluster.Spec.Proxy.PGBouncer.Containers...)
 	}
 
-	outPod.Volumes = []corev1.Volume{configVolume}
+	template.Spec.Volumes = []corev1.Volume{configVolume}
 
 	if feature.Enabled(ctx, feature.OpenTelemetryLogs) || feature.Enabled(ctx, feature.OpenTelemetryMetrics) {
 		collector.AddToPod(ctx, inCluster.Spec.Instrumentation, inCluster.Spec.ImagePullPolicy, inConfigMap,
-			outPod, []corev1.VolumeMount{configVolumeMount}, string(inSecret.Data["pgbouncer-password"]), []string{naming.PGBouncerLogPath},
+			template, []corev1.VolumeMount{configVolumeMount}, string(inSecret.Data["pgbouncer-password"]), []string{naming.PGBouncerLogPath},
 			true)
 	}
 }
diff --git a/internal/pgbouncer/reconcile_test.go b/internal/pgbouncer/reconcile_test.go
@@ -101,16 +101,16 @@ func TestPod(t *testing.T) {
 	configMap := new(corev1.ConfigMap)
 	primaryCertificate := new(corev1.SecretProjection)
 	secret := new(corev1.Secret)
-	pod := new(corev1.PodSpec)
+	template := new(corev1.PodTemplateSpec)
 
-	call := func() { Pod(ctx, cluster, configMap, primaryCertificate, secret, pod) }
+	call := func() { Pod(ctx, cluster, configMap, primaryCertificate, secret, template) }
 
 	t.Run("Disabled", func(t *testing.T) {
-		before := pod.DeepCopy()
+		before := template.DeepCopy()
 		call()
 
 		// No change when PgBouncer is not requested in the spec.
-		assert.DeepEqual(t, before, pod)
+		assert.DeepEqual(t, before, template)
 	})
 
 	t.Run("Defaults", func(t *testing.T) {
@@ -120,7 +120,7 @@ func TestPod(t *testing.T) {
 
 		call()
 
-		assert.Assert(t, cmp.MarshalMatches(pod, `
+		assert.Assert(t, cmp.MarshalMatches(template.Spec, `
 containers:
 - command:
   - pgbouncer
@@ -209,9 +209,9 @@ volumes:
 		`))
 
 		// No change when called again.
-		before := pod.DeepCopy()
+		before := template.DeepCopy()
 		call()
-		assert.DeepEqual(t, before, pod)
+		assert.DeepEqual(t, before, template)
 	})
 
 	t.Run("Customizations", func(t *testing.T) {
@@ -230,7 +230,7 @@ volumes:
 
 		call()
 
-		assert.Assert(t, cmp.MarshalMatches(pod, `
+		assert.Assert(t, cmp.MarshalMatches(template.Spec, `
 containers:
 - command:
   - pgbouncer
@@ -340,7 +340,7 @@ volumes:
 
 		call()
 
-		assert.Assert(t, cmp.MarshalMatches(pod, `
+		assert.Assert(t, cmp.MarshalMatches(template.Spec, `
 containers:
 - command:
   - pgbouncer
@@ -444,7 +444,7 @@ volumes:
 		t.Run("SidecarNotEnabled", func(t *testing.T) {
 
 			call()
-			assert.Equal(t, len(pod.Containers), 2, "expected 2 containers in Pod, got %d", len(pod.Containers))
+			assert.Equal(t, len(template.Spec.Containers), 2, "expected 2 containers in Pod, got %d", len(template.Spec.Containers))
 		})
 
 		t.Run("SidecarEnabled", func(t *testing.T) {
@@ -453,11 +453,11 @@ volumes:
 			}))
 			call()
 
-			assert.Equal(t, len(pod.Containers), 3, "expected 3 containers in Pod, got %d", len(pod.Containers))
+			assert.Equal(t, len(template.Spec.Containers), 3, "expected 3 containers in Pod, got %d", len(template.Spec.Containers))
 
 			var found bool
-			for i := range pod.Containers {
-				if pod.Containers[i].Name == "customsidecar1" {
+			for i := range template.Spec.Containers {
+				if template.Spec.Containers[i].Name == "customsidecar1" {
 					found = true
 					break
 				}

Original file line number	Diff line number	Diff line change
`@@ -1220,7 +1220,7 @@ func (r *Reconciler) reconcileInstance(`
`1220`	`1220`
`1221`	`1221`	`// For now, we are not using logrotate to rotate postgres or patroni logs`
`1222`	`1222`	`// but we are using it for pgbackrest logs in the postgres pod`
`1223`		`- collector.AddToPod(ctx, cluster.Spec.Instrumentation, cluster.Spec.ImagePullPolicy, instanceConfigMap, &instance.Spec.Template.Spec,`
	`1223`	`+ collector.AddToPod(ctx, cluster.Spec.Instrumentation, cluster.Spec.ImagePullPolicy, instanceConfigMap, &instance.Spec.Template,`
`1224`	`1224`	`[]corev1.VolumeMount{postgres.DataVolumeMount()}, pgPassword,`
`1225`	`1225`	`[]string{naming.PGBackRestPGDataLogPath}, true)`
`1226`	`1226`	`}`
Original file line number	Diff line number	Diff line change
`@@ -472,7 +472,7 @@ func (r *Reconciler) generatePGBouncerDeployment(`
`472`	`472`	`err := errors.WithStack(r.setControllerReference(cluster, deploy))`
`473`	`473`
`474`	`474`	`if err == nil {`
`475`		`- pgbouncer.Pod(ctx, cluster, configmap, primaryCertificate, secret, &deploy.Spec.Template.Spec)`
	`475`	`+ pgbouncer.Pod(ctx, cluster, configmap, primaryCertificate, secret, &deploy.Spec.Template)`
`476`	`476`	`}`
`477`	`477`
`478`	`478`	`// Add tmp directory and volume for log files`
Original file line number	Diff line number	Diff line change
`@@ -132,7 +132,7 @@ func statefulset(`
`132`	`132`	`}`
`133`	`133`
`134`	`134`	`collector.AddToPod(ctx, pgadmin.Spec.Instrumentation, pgadmin.Spec.ImagePullPolicy,`
`135`		`- configmap, &sts.Spec.Template.Spec, volumeMounts, "", []string{}, false)`
	`135`	`+ configmap, &sts.Spec.Template, volumeMounts, "", []string{}, false)`
`136`	`136`	`}`
`137`	`137`
`138`	`138`	`return sts`