Add an OpenTelemetry Collector for logs and metrics

Makefile

@@ -254,16 +254,16 @@ generate-kuttl: ## Generate kuttl tests
 ##@ Generate
 
 .PHONY: check-generate
-check-generate: ## Check crd, deepcopy functions, and rbac generation
-check-generate: generate-crd
-check-generate: generate-deepcopy
-check-generate: generate-rbac
+check-generate: ## Check everything generated is also committed
+check-generate: generate
 	git diff --exit-code -- config/crd
 	git diff --exit-code -- config/rbac
+	git diff --exit-code -- internal/collector
 	git diff --exit-code -- pkg/apis
 
 .PHONY: generate
-generate: ## Generate crd, deepcopy functions, and rbac
+generate: ## Generate everything
+generate: generate-collector
 generate: generate-crd
 generate: generate-deepcopy
 generate: generate-rbac
@@ -276,6 +276,10 @@ generate-crd: tools/controller-gen
 		paths='./pkg/apis/...' \
 		output:dir='config/crd/bases' # {directory}/{group}_{plural}.yaml
 
+.PHONY: generate-collector
+generate-collector: ## Generate OTel Collector files
+	$(GO) generate ./internal/collector
+
 .PHONY: generate-deepcopy
 generate-deepcopy: ## Generate DeepCopy functions
 generate-deepcopy: tools/controller-gen

internal/collector/config.go

@@ -0,0 +1,93 @@
+// Copyright 2024 - 2025 Crunchy Data Solutions, Inc.
+//
+// SPDX-License-Identifier: Apache-2.0
+
+package collector
+
+import (
+	"k8s.io/apimachinery/pkg/util/sets"
+	"sigs.k8s.io/yaml"
+)
+
+// ComponentID represents a component identifier within an OpenTelemetry
+// Collector YAML configuration. Each value is a "type" followed by an optional
+// slash-then-name: `type[/name]`
+type ComponentID string
+
+// Config represents an OpenTelemetry Collector YAML configuration.
+// See: https://opentelemetry.io/docs/collector/configuration
+type Config struct {
+	Exporters  map[ComponentID]any
+	Extensions map[ComponentID]any
+	Processors map[ComponentID]any
+	Receivers  map[ComponentID]any
+
+	Pipelines map[PipelineID]Pipeline
+}
+
+// Pipeline represents the YAML configuration of a flow of telemetry data
+// through an OpenTelemetry Collector.
+// See: https://opentelemetry.io/docs/collector/configuration#pipelines
+type Pipeline struct {
+	Extensions []ComponentID
+	Exporters  []ComponentID
+	Processors []ComponentID
+	Receivers  []ComponentID
+}
+
+// PipelineID represents a pipeline identifier within an OpenTelemetry Collector
+// YAML configuration. Each value is a signal followed by an optional
+// slash-then-name: `signal[/name]`
+type PipelineID string
+
+func (c *Config) ToYAML() (string, error) {
+	const yamlGeneratedWarning = "" +
+		"# Generated by postgres-operator. DO NOT EDIT.\n" +
+		"# Your changes will not be saved.\n"
+
+	extensions := sets.New[ComponentID]()
+	pipelines := make(map[PipelineID]any, len(c.Pipelines))
+
+	for id, p := range c.Pipelines {
+		extensions.Insert(p.Extensions...)
+		pipelines[id] = map[string]any{
+			"exporters":  p.Exporters,
+			"processors": p.Processors,
+			"receivers":  p.Receivers,
+		}
+	}
+
+	b, err := yaml.Marshal(map[string]any{
+		"exporters":  c.Exporters,
+		"extensions": c.Extensions,
+		"processors": c.Processors,
+		"receivers":  c.Receivers,
+		"service": map[string]any{
+			"extensions": sets.List(extensions), // Sorted
+			"pipelines":  pipelines,
+		},
+	})
+	return string(append([]byte(yamlGeneratedWarning), b...)), err
+}
+
+// NewConfig creates a base config for an OTel collector container
+func NewConfig() *Config {
+	return &Config{
+		Exporters: map[ComponentID]any{
+			// TODO: Do we want a DebugExporter outside of development?
+			// https://pkg.go.dev/go.opentelemetry.io/collector/exporter/debugexporter#section-readme
+			DebugExporter: map[string]any{"verbosity": "detailed"},
+		},
+		Extensions: map[ComponentID]any{},
+		Processors: map[ComponentID]any{
+			// https://pkg.go.dev/go.opentelemetry.io/collector/processor/batchprocessor#section-readme
+			OneSecondBatchProcessor: map[string]any{"timeout": "1s"},
+			SubSecondBatchProcessor: map[string]any{"timeout": "200ms"},
+
+			// https://pkg.go.dev/github.com/open-telemetry/opentelemetry-collector-contrib/processor/groupbyattrsprocessor#readme-compaction
+			CompactingProcessor: map[string]any{},
+		},
+		Receivers: map[ComponentID]any{},
+		Pipelines: map[PipelineID]Pipeline{},
+	}
+}

internal/collector/config_test.go

@@ -0,0 +1,33 @@
+// Copyright 2024 - 2025 Crunchy Data Solutions, Inc.
+//
+// SPDX-License-Identifier: Apache-2.0
+
+package collector
+
+import (
+	"testing"
+
+	"gotest.tools/v3/assert"
+)
+
+func TestConfigToYAML(t *testing.T) {
+	result, err := NewConfig().ToYAML()
+	assert.NilError(t, err)
+	assert.DeepEqual(t, result, `# Generated by postgres-operator. DO NOT EDIT.
+# Your changes will not be saved.
+exporters:
+  debug:
+    verbosity: detailed
+extensions: {}
+processors:
+  batch/1s:
+    timeout: 1s
+  batch/200ms:
+    timeout: 200ms
+  groupbyattrs/compact: {}
+receivers: {}
+service:
+  extensions: []
+  pipelines: {}
+`)
+}

internal/collector/generate.go

@@ -0,0 +1,60 @@
+// Copyright 2024 - 2025 Crunchy Data Solutions, Inc.
+//
+// SPDX-License-Identifier: Apache-2.0
+
+// [pg_query.Parse] requires CGO to compile and call https://github.com/pganalyze/libpg_query
+//go:build cgo && generate
+
+//go:generate go run generate.go
+
+package main
+
+import (
+	"bytes"
+	"log/slog"
+	"os"
+	"path/filepath"
+	"strings"
+
+	pg_query "github.com/pganalyze/pg_query_go/v5"
+	"sigs.k8s.io/yaml"
+)
+
+func main() {
+	cwd := need(os.Getwd())
+	fileNames := map[string][]string{}
+
+	slog.Info("Reading", "directory", cwd)
+	for _, entry := range need(os.ReadDir(cwd)) {
+		if entry.Type() == 0 {
+			ext := filepath.Ext(entry.Name())
+			fileNames[ext] = append(fileNames[ext], entry.Name())
+		}
+	}
+
+	for _, sqlName := range fileNames[".sql"] {
+		slog.Info("Reading", "file", sqlName)
+		sqlData := need(pg_query.Parse(string(need(os.ReadFile(sqlName)))))
+		sqlPath := filepath.Join("generated", sqlName)
+
+		slog.Info("Writing", "file", sqlPath)
+		must(os.WriteFile(sqlPath, []byte(need(pg_query.Deparse(sqlData))+"\n"), 0o644))
+	}
+
+	for _, yamlName := range fileNames[".yaml"] {
+		slog.Info("Reading", "file", yamlName)
+		jsonData := need(yaml.YAMLToJSONStrict(need(os.ReadFile(yamlName))))
+		jsonPath := filepath.Join("generated", strings.TrimSuffix(yamlName, ".yaml")+".json")
+
+		slog.Info("Writing", "file", jsonPath)
+		must(os.WriteFile(jsonPath, append(bytes.TrimSpace(jsonData), '\n'), 0o644))
+	}
+}
+
+func must(err error) { need(0, err) }
+func need[V any](v V, err error) V {
+	if err != nil {
+		panic(err)
+	}
+	return v
+}

internal/collector/generated/.gitattributes

@@ -0,0 +1,2 @@
+# https://docs.github.com/en/repositories/working-with-files/managing-files/customizing-how-changed-files-appear-on-github
+/*.json linguist-generated=true

internal/collector/instance.go

@@ -0,0 +1,104 @@
+// 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"
+)
+
+// 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,
+	pullPolicy corev1.PullPolicy,
+	inInstanceConfigMap *corev1.ConfigMap,
+	outPod *corev1.PodSpec,
+	volumeMounts []corev1.VolumeMount,
+	sqlQueryPassword string,
+) {
+	if !(feature.Enabled(ctx, feature.OpenTelemetryLogs) || 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.117.0",
+		ImagePullPolicy: pullPolicy,
+		Command:         []string{"/otelcol-contrib", "--config", "/etc/otel-collector/config.yaml"},
+		Env: []corev1.EnvVar{
+			{
+				Name: "K8S_POD_NAMESPACE",
+				ValueFrom: &corev1.EnvVarSource{FieldRef: &corev1.ObjectFieldSelector{
+					FieldPath: "metadata.namespace",
+				}},
+			},
+			{
+				Name: "K8S_POD_NAME",
+				ValueFrom: &corev1.EnvVarSource{FieldRef: &corev1.ObjectFieldSelector{
+					FieldPath: "metadata.name",
+				}},
+			},
+			{
+				Name:  "PGPASSWORD",
+				Value: sqlQueryPassword,
+			},
+		},
+
+		SecurityContext: initialize.RestrictedSecurityContext(),
+		VolumeMounts:    append(volumeMounts, configVolumeMount),
+	}
+
+	if feature.Enabled(ctx, feature.OpenTelemetryMetrics) {
+		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)
+}

internal/collector/naming.go

@@ -0,0 +1,13 @@
+// Copyright 2024 - 2025 Crunchy Data Solutions, Inc.
+//
+// SPDX-License-Identifier: Apache-2.0
+
+package collector
+
+const CompactingProcessor = "groupbyattrs/compact"
+const DebugExporter = "debug"
+const OneSecondBatchProcessor = "batch/1s"
+const SubSecondBatchProcessor = "batch/200ms"
+const Prometheus = "prometheus"
+const Metrics = "metrics"
+const SqlQuery = "sqlquery"