Add an OTel Collector with Patroni metrics

dsessler7 · cbandy · commit 33dc4ef68f62 · 2025-02-05T12:43:49.000-06:00
Issue: PGO-2043
diff --git a/internal/collector/config.go b/internal/collector/config.go
@@ -1,4 +1,4 @@
-// Copyright 2024 Crunchy Data Solutions, Inc.
+// Copyright 2024 - 2025 Crunchy Data Solutions, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 
diff --git a/internal/collector/config_test.go b/internal/collector/config_test.go
@@ -1,4 +1,4 @@
-// Copyright 2024 Crunchy Data Solutions, Inc.
+// Copyright 2024 - 2025 Crunchy Data Solutions, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 
diff --git a/internal/collector/instance.go b/internal/collector/instance.go
@@ -0,0 +1,85 @@
+// Copyright 2024 - 2025 Crunchy Data Solutions, Inc.
+//
+// SPDX-License-Identifier: Apache-2.0
+
+package collector
+
+import (
+	"context"
+
+	corev1 "k8s.io/api/core/v1"
+
+	"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/pkg/apis/postgres-operator.crunchydata.com/v1beta1"
+)
+
+// AddToConfigMap populates the shared ConfigMap with fields needed to run the Collector.
+func AddToConfigMap(
+	ctx context.Context,
+	inConfig *Config,
+	outInstanceConfigMap *corev1.ConfigMap,
+) error {
+	var err error
+	if outInstanceConfigMap.Data == nil {
+		outInstanceConfigMap.Data = make(map[string]string)
+	}
+
+	outInstanceConfigMap.Data["collector.yaml"], err = inConfig.ToYAML()
+
+	return err
+}
+
+// AddToPod adds the OpenTelemetry collector container to a given Pod
+func AddToPod(
+	ctx context.Context,
+	inCluster *v1beta1.PostgresCluster,
+	inInstanceConfigMap *corev1.ConfigMap,
+	outPod *corev1.PodSpec,
+	volumeMounts []corev1.VolumeMount,
+) {
+	if !feature.Enabled(ctx, feature.OpenTelemetryMetrics) {
+		return
+	}
+
+	configVolumeMount := corev1.VolumeMount{
+		Name:      "collector-config",
+		MountPath: "/etc/otel-collector",
+		ReadOnly:  true,
+	}
+	configVolume := corev1.Volume{Name: configVolumeMount.Name}
+	configVolume.Projected = &corev1.ProjectedVolumeSource{
+		Sources: []corev1.VolumeProjection{{
+			ConfigMap: &corev1.ConfigMapProjection{
+				LocalObjectReference: corev1.LocalObjectReference{
+					Name: inInstanceConfigMap.Name,
+				},
+				Items: []corev1.KeyToPath{{
+					Key:  "collector.yaml",
+					Path: "config.yaml",
+				}},
+			},
+		}},
+	}
+
+	container := corev1.Container{
+		Name: naming.ContainerCollector,
+
+		Image:           "ghcr.io/open-telemetry/opentelemetry-collector-releases/opentelemetry-collector-contrib:0.116.1",
+		ImagePullPolicy: inCluster.Spec.ImagePullPolicy,
+		Command:         []string{"/otelcol-contrib", "--config", "/etc/otel-collector/config.yaml"},
+
+		SecurityContext: initialize.RestrictedSecurityContext(),
+		VolumeMounts:    append(volumeMounts, configVolumeMount),
+	}
+
+	container.Ports = []corev1.ContainerPort{{
+		ContainerPort: int32(8889),
+		Name:          "otel-metrics",
+		Protocol:      corev1.ProtocolTCP,
+	}}
+
+	outPod.Containers = append(outPod.Containers, container)
+	outPod.Volumes = append(outPod.Volumes, configVolume)
+}
diff --git a/internal/collector/naming.go b/internal/collector/naming.go
@@ -1,4 +1,4 @@
-// Copyright 2024 Crunchy Data Solutions, Inc.
+// Copyright 2024 - 2025 Crunchy Data Solutions, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 
@@ -8,3 +8,5 @@ const CompactingProcessor = "groupbyattrs/compact"
 const DebugExporter = "debug"
 const OneSecondBatchProcessor = "batch/1s"
 const SubSecondBatchProcessor = "batch/200ms"
+const Prometheus = "prometheus"
+const Metrics = "metrics"
diff --git a/internal/collector/postgres.go b/internal/collector/postgres.go
@@ -0,0 +1,53 @@
+// Copyright 2024 - 2025 Crunchy Data Solutions, Inc.
+//
+// SPDX-License-Identifier: Apache-2.0
+
+package collector
+
+import (
+	"context"
+
+	"github.com/crunchydata/postgres-operator/internal/feature"
+)
+
+func NewConfigForPostgresPod(ctx context.Context) *Config {
+	config := NewConfig()
+
+	if feature.Enabled(ctx, feature.OpenTelemetryMetrics) {
+		// Add Prometheus exporter
+		config.Exporters[Prometheus] = map[string]any{
+			"endpoint": "0.0.0.0:8889",
+		}
+
+		// Add Prometheus Receiver
+		config.Receivers[Prometheus] = map[string]any{
+			"config": map[string]any{
+				"scrape_configs": []map[string]any{
+					{
+						"job_name": "patroni",
+						"scheme":   "https",
+						"tls_config": map[string]any{
+							"insecure_skip_verify": true,
+						},
+						"scrape_interval": "10s",
+						"static_configs": []map[string]any{
+							{
+								"targets": []string{
+									"0.0.0.0:8008",
+								},
+							},
+						},
+					},
+				},
+			},
+		}
+
+		// Add Metrics Pipeline
+		config.Pipelines[Metrics] = Pipeline{
+			Receivers: []ComponentID{Prometheus},
+			Exporters: []ComponentID{Prometheus},
+		}
+	}
+
+	return config
+}
diff --git a/internal/controller/postgrescluster/controller.go b/internal/controller/postgrescluster/controller.go
@@ -27,6 +27,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 
+	"github.com/crunchydata/postgres-operator/internal/collector"
 	"github.com/crunchydata/postgres-operator/internal/config"
 	"github.com/crunchydata/postgres-operator/internal/controller/runtime"
 	"github.com/crunchydata/postgres-operator/internal/initialize"
@@ -241,6 +242,8 @@ func (r *Reconciler) Reconcile(
 	pgbackrest.PostgreSQL(cluster, &pgParameters, backupsSpecFound)
 	pgmonitor.PostgreSQLParameters(cluster, &pgParameters)
 
+	otelConfig := collector.NewConfigForPostgresPod(ctx)
+
 	// Set huge_pages = try if a hugepages resource limit > 0, otherwise set "off"
 	postgres.SetHugePages(cluster, &pgParameters)
 
@@ -349,7 +352,7 @@ func (r *Reconciler) Reconcile(
 			ctx, cluster, clusterConfigMap, clusterReplicationSecret, rootCA,
 			clusterPodService, instanceServiceAccount, instances, patroniLeaderService,
 			primaryCertificate, clusterVolumes, exporterQueriesConfig, exporterWebConfig,
-			backupsSpecFound,
+			backupsSpecFound, otelConfig,
 		)
 	}
 
diff --git a/internal/controller/postgrescluster/instance.go b/internal/controller/postgrescluster/instance.go
@@ -24,6 +24,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 
+	"github.com/crunchydata/postgres-operator/internal/collector"
 	"github.com/crunchydata/postgres-operator/internal/config"
 	"github.com/crunchydata/postgres-operator/internal/controller/runtime"
 	"github.com/crunchydata/postgres-operator/internal/feature"
@@ -591,6 +592,7 @@ func (r *Reconciler) reconcileInstanceSets(
 	clusterVolumes []*corev1.PersistentVolumeClaim,
 	exporterQueriesConfig, exporterWebConfig *corev1.ConfigMap,
 	backupsSpecFound bool,
+	otelConfig *collector.Config,
 ) error {
 
 	// Go through the observed instances and check if a primary has been determined.
@@ -628,7 +630,7 @@ func (r *Reconciler) reconcileInstanceSets(
 			patroniLeaderService, primaryCertificate,
 			findAvailableInstanceNames(*set, instances, clusterVolumes),
 			numInstancePods, clusterVolumes, exporterQueriesConfig, exporterWebConfig,
-			backupsSpecFound,
+			backupsSpecFound, otelConfig,
 		)
 
 		if err == nil {
@@ -1063,6 +1065,7 @@ func (r *Reconciler) scaleUpInstances(
 	clusterVolumes []*corev1.PersistentVolumeClaim,
 	exporterQueriesConfig, exporterWebConfig *corev1.ConfigMap,
 	backupsSpecFound bool,
+	otelConfig *collector.Config,
 ) ([]*appsv1.StatefulSet, error) {
 	log := logging.FromContext(ctx)
 
@@ -1109,7 +1112,7 @@ func (r *Reconciler) scaleUpInstances(
 			rootCA, clusterPodService, instanceServiceAccount,
 			patroniLeaderService, primaryCertificate, instances[i],
 			numInstancePods, clusterVolumes, exporterQueriesConfig, exporterWebConfig,
-			backupsSpecFound,
+			backupsSpecFound, otelConfig,
 		)
 	}
 	if err == nil {
@@ -1140,6 +1143,7 @@ func (r *Reconciler) reconcileInstance(
 	clusterVolumes []*corev1.PersistentVolumeClaim,
 	exporterQueriesConfig, exporterWebConfig *corev1.ConfigMap,
 	backupsSpecFound bool,
+	otelConfig *collector.Config,
 ) error {
 	log := logging.FromContext(ctx).WithValues("instance", instance.Name)
 	ctx = logging.NewContext(ctx, log)
@@ -1164,7 +1168,7 @@ func (r *Reconciler) reconcileInstance(
 	)
 
 	if err == nil {
-		instanceConfigMap, err = r.reconcileInstanceConfigMap(ctx, cluster, spec, instance)
+		instanceConfigMap, err = r.reconcileInstanceConfigMap(ctx, cluster, spec, instance, otelConfig)
 	}
 	if err == nil {
 		instanceCertificates, err = r.reconcileInstanceCertificates(
@@ -1196,6 +1200,10 @@ func (r *Reconciler) reconcileInstance(
 			spec, instanceCertificates, instanceConfigMap, &instance.Spec.Template)
 	}
 
+	if err == nil && feature.Enabled(ctx, feature.OpenTelemetryMetrics) {
+		collector.AddToPod(ctx, cluster, instanceConfigMap, &instance.Spec.Template.Spec, nil)
+	}
+
 	// Add pgMonitor resources to the instance Pod spec
 	if err == nil {
 		err = addPGMonitorToInstancePodSpec(ctx, cluster, &instance.Spec.Template, exporterQueriesConfig, exporterWebConfig)
@@ -1377,7 +1385,7 @@ func addPGBackRestToInstancePodSpec(
 // files (etc) that apply to instance of cluster.
 func (r *Reconciler) reconcileInstanceConfigMap(
 	ctx context.Context, cluster *v1beta1.PostgresCluster, spec *v1beta1.PostgresInstanceSetSpec,
-	instance *appsv1.StatefulSet,
+	instance *appsv1.StatefulSet, otelConfig *collector.Config,
 ) (*corev1.ConfigMap, error) {
 	instanceConfigMap := &corev1.ConfigMap{ObjectMeta: naming.InstanceConfigMap(instance)}
 	instanceConfigMap.SetGroupVersionKind(corev1.SchemeGroupVersion.WithKind("ConfigMap"))
@@ -1397,6 +1405,9 @@ func (r *Reconciler) reconcileInstanceConfigMap(
 			naming.LabelInstance:    instance.Name,
 		})
 
+	if err == nil && feature.Enabled(ctx, feature.OpenTelemetryMetrics) {
+		err = collector.AddToConfigMap(ctx, otelConfig, instanceConfigMap)
+	}
 	if err == nil {
 		err = patroni.InstanceConfigMap(ctx, cluster, spec, instanceConfigMap)
 	}
diff --git a/internal/feature/features.go b/internal/feature/features.go
@@ -81,6 +81,9 @@ const (
 	// Support custom sidecars for PostgreSQL instance Pods
 	InstanceSidecars = "InstanceSidecars"
 
+	// Export metrics using OpenTelemetry
+	OpenTelemetryMetrics = "OpenTelemetryMetrics"
+
 	// Support custom sidecars for pgBouncer Pods
 	PGBouncerSidecars = "PGBouncerSidecars"
 
@@ -104,6 +107,7 @@ func NewGate() MutableGate {
 		AutoGrowVolumes:         {Default: false, PreRelease: featuregate.Alpha},
 		BridgeIdentifiers:       {Default: false, PreRelease: featuregate.Deprecated},
 		InstanceSidecars:        {Default: false, PreRelease: featuregate.Alpha},
+		OpenTelemetryMetrics:    {Default: false, PreRelease: featuregate.Alpha},
 		PGBouncerSidecars:       {Default: false, PreRelease: featuregate.Alpha},
 		PGUpgradeCPUConcurrency: {Default: false, PreRelease: featuregate.Alpha},
 		TablespaceVolumes:       {Default: false, PreRelease: featuregate.Alpha},
diff --git a/internal/feature/features_test.go b/internal/feature/features_test.go
@@ -21,6 +21,7 @@ func TestDefaults(t *testing.T) {
 	assert.Assert(t, false == gate.Enabled(AutoGrowVolumes))
 	assert.Assert(t, false == gate.Enabled(BridgeIdentifiers))
 	assert.Assert(t, false == gate.Enabled(InstanceSidecars))
+	assert.Assert(t, false == gate.Enabled(OpenTelemetryMetrics))
 	assert.Assert(t, false == gate.Enabled(PGBouncerSidecars))
 	assert.Assert(t, false == gate.Enabled(PGUpgradeCPUConcurrency))
 	assert.Assert(t, false == gate.Enabled(TablespaceVolumes))

Original file line number	Diff line number	Diff line change
`@@ -1,4 +1,4 @@`
`1`		`-// Copyright 2024 Crunchy Data Solutions, Inc.`
	`1`	`+// Copyright 2024 - 2025 Crunchy Data Solutions, Inc.`
`2`	`2`	`//`
`3`	`3`	`// SPDX-License-Identifier: Apache-2.0`
`4`	`4`