feat(target allocator): bridge Prometheus-only metrics into OTLP, add resource, shutdown, validation

- Export self-telemetry over OTLP with the same metric set as /metrics by
  bridging the default Prometheus registry (SD internals, Go/process collectors)
  into the OTLP reader via the Prometheus bridge; SDK metrics use a dedicated
  registry to avoid double counting, and /metrics serves the union.
- Attach a resource with a default service.name (overridable via OTEL_* env).
- Shut down the meter provider on exit to flush the final export.
- Expand ${env:VAR} in OTLP header/endpoint values so secrets (API tokens) can be
  sourced from a Secret via pod env instead of being written into the ConfigMap.
- Accept both a full URL and host:port endpoint for gRPC (consistent with HTTP).
- Validate telemetry config in the TA binary for standalone use.
- Unit + end-to-end export tests, config-file README docs, changelog entry.

Author: odubajDT

.chloggen/ta-otlp-self-telemetry.yaml

@@ -0,0 +1,21 @@
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. collector, target allocator, auto-instrumentation, opamp, github action)
+component: target allocator
+
+# A brief description of the change. Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: "Support exporting the TargetAllocator's self-telemetry metrics via OTLP, in addition to the Prometheus /metrics endpoint."
+
+# One or more tracking issues related to the change
+issues: [5047]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext: |
+  Configure it under `spec.targetAllocator.telemetry.metrics.otlp` (OpenTelemetryCollector CR) or
+  `spec.telemetry.metrics.otlp` (TargetAllocator CR), with endpoint, protocol (grpc/http), temporality,
+  headers, TLS and export interval/timeout. Metrics registered directly on the Prometheus registry
+  (Prometheus service discovery, Go runtime and process collectors) are bridged into the OTLP export
+  so the Prometheus endpoint and OTLP expose the same metric set.

cmd/otel-allocator/README.md

@@ -29,9 +29,40 @@ The Target Allocator uses a configuration file (by default under `/conf/targetal
 | `https`                            | Whether to expose the target allocator endpoint over https                    |                                               |                      |
 | `allow_insecure_auth_secrets`      | Serve auth secret values over plain HTTP without mTLS                         | `false`                                       | `ALLOW_INSECURE_AUTH_SECRETS` |
 | `collector_not_ready_grace_period` | Wait time before assigning jobs to a new collector.                           | 30s                                           |                      |
+| `telemetry`                        | Self-telemetry configuration, including optional OTLP metric export (see below)|                                              |                      |
 
 Additional configuration options are present under [./internal/config/config.go](./internal/config/config.go).
 
+### Self-telemetry
+
+The Target Allocator always exposes its own metrics on the Prometheus `/metrics` endpoint. It can
+additionally push them over OTLP by configuring `telemetry.metrics.otlp`. When set, an OTLP exporter
+is added alongside the Prometheus endpoint; the metrics registered directly on the Prometheus registry
+(Prometheus service discovery internals, Go runtime and process collectors) are bridged into the OTLP
+export so both surfaces expose the same metric set. The `service.name` resource attribute defaults to
+`target-allocator` and can be overridden with the standard `OTEL_SERVICE_NAME` /
+`OTEL_RESOURCE_ATTRIBUTES` environment variables.
+
+```yaml
+telemetry:
+  metrics:
+    otlp:
+      endpoint: https://example.com:4318   # base URL (HTTP, /v1/metrics appended) or host:port (gRPC)
+      protocol: http                        # "grpc" (default) or "http"
+      temporality: delta                    # "cumulative" (default), "delta", or "lowmemory"
+      export_interval: 30s                  # default 60s
+      timeout: 15s                          # default 10s
+      insecure: false                       # disable TLS (local dev)
+      headers:
+        Authorization: "Api-Token ${env:DT_API_TOKEN}"
+```
+
+Header values support `${env:VAR}` references, which are expanded from the Target Allocator's
+process environment at runtime. Use this to keep secrets (such as an API token) out of the
+ConfigMap: reference an environment variable here and source that variable from a Secret on the
+Target Allocator pod. The `endpoint` field supports the same expansion. Both the `grpc` and `http`
+protocols accept either a full URL (with scheme) or a bare `host:port` endpoint.
+
 ## Even Distribution of Prometheus Targets
 
 The Target Allocator’s first job is to discover targets to scrape and OTel Collectors to allocate targets to. Then it can distribute the targets it discovers among the Collectors. The Collectors in turn query the Target Allocator for Metrics endpoints to scrape, and then the Collectors’ [Prometheus Receivers](https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/main/receiver/prometheusreceiver/README.md) scrape the Metrics targets.

cmd/otel-allocator/internal/config/config.go

@@ -475,6 +475,30 @@ func ValidateConfig(config *Config) error {
 	if len(config.PrometheusCR.AllowNamespaces) != 0 && len(config.PrometheusCR.DenyNamespaces) != 0 {
 		return errors.New("only one of allowNamespaces or denyNamespaces can be set")
 	}
+	return validateTelemetry(config.Telemetry)
+}
+
+// validateTelemetry validates the self-telemetry configuration. The operator sets these
+// values from a validated CRD, but the Target Allocator can also be run standalone with a
+// config file, so we validate here as well for a clear error instead of a runtime failure.
+func validateTelemetry(t TelemetryConfig) error {
+	otlp := t.Metrics.OTLP
+	if otlp == nil {
+		return nil
+	}
+	if otlp.Endpoint == "" {
+		return errors.New("telemetry.metrics.otlp.endpoint must be set")
+	}
+	switch otlp.Protocol {
+	case "", "grpc", "http":
+	default:
+		return fmt.Errorf("telemetry.metrics.otlp.protocol must be 'grpc' or 'http', got %q", otlp.Protocol)
+	}
+	switch otlp.Temporality {
+	case "", "cumulative", "delta", "lowmemory":
+	default:
+		return fmt.Errorf("telemetry.metrics.otlp.temporality must be 'cumulative', 'delta', or 'lowmemory', got %q", otlp.Temporality)
+	}
 	return nil
 }
 

cmd/otel-allocator/internal/config/telemetry_test.go

@@ -0,0 +1,39 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package config
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestValidateTelemetry(t *testing.T) {
+	otlp := func(o OTLPExporterConfig) TelemetryConfig {
+		return TelemetryConfig{Metrics: MetricsConfig{OTLP: &o}}
+	}
+
+	tests := []struct {
+		name    string
+		cfg     TelemetryConfig
+		wantErr bool
+	}{
+		{name: "empty is valid", cfg: TelemetryConfig{}},
+		{name: "valid grpc delta", cfg: otlp(OTLPExporterConfig{Protocol: "grpc", Endpoint: "example.com:4317", Temporality: "delta"})},
+		{name: "valid http defaults", cfg: otlp(OTLPExporterConfig{Endpoint: "http://gw:4318"})},
+		{name: "missing endpoint", cfg: otlp(OTLPExporterConfig{Protocol: "grpc"}), wantErr: true},
+		{name: "bad protocol", cfg: otlp(OTLPExporterConfig{Protocol: "thrift", Endpoint: "gw:4317"}), wantErr: true},
+		{name: "bad temporality", cfg: otlp(OTLPExporterConfig{Endpoint: "gw:4317", Temporality: "gauge"}), wantErr: true},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			err := validateTelemetry(tt.cfg)
+			if tt.wantErr {
+				assert.Error(t, err)
+			} else {
+				assert.NoError(t, err)
+			}
+		})
+	}
+}

cmd/otel-allocator/internal/server/server.go

@@ -21,6 +21,7 @@ import (
 	"github.com/gin-gonic/gin"
 	"github.com/go-logr/logr"
 	"github.com/goccy/go-json"
+	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 	promcommconfig "github.com/prometheus/common/config"
 	promconfig "github.com/prometheus/prometheus/config"
@@ -62,10 +63,22 @@ type Server struct {
 	ScrapeConfigMarshalledSecretResponse []byte
 	httpDuration                         metric.Float64Histogram
 	allowInsecureAuthSecrets             bool
+	// metricsGatherer, when set, is used to serve the /metrics endpoint. Defaults to
+	// the global Prometheus default gatherer.
+	metricsGatherer prometheus.Gatherer
 }
 
 type Option func(*Server)
 
+// WithMetricsGatherer sets the Prometheus gatherer used to serve /metrics. This allows
+// the caller to combine multiple registries (e.g. the default registry plus a dedicated
+// OTel SDK registry) into a single endpoint.
+func WithMetricsGatherer(gatherer prometheus.Gatherer) Option {
+	return func(s *Server) {
+		s.metricsGatherer = gatherer
+	}
+}
+
 // Option to create an additional https server with mTLS configuration.
 // Used for getting the scrape config with real secret values.
 func WithTLSConfig(tlsConfig *tls.Config, httpsListenAddr string) Option {
@@ -83,6 +96,15 @@ func WithInsecureAuthSecrets() Option {
 	}
 }
 
+// metricsHandler returns the HTTP handler serving /metrics. It uses the configured
+// gatherer if one was provided, otherwise the global Prometheus default.
+func (s *Server) metricsHandler() http.Handler {
+	if s.metricsGatherer != nil {
+		return promhttp.HandlerFor(s.metricsGatherer, promhttp.HandlerOpts{})
+	}
+	return promhttp.Handler()
+}
+
 func (s *Server) setRouter(router *gin.Engine) {
 	router.Use(gin.Recovery())
 	router.UseRawPath = true
@@ -100,7 +122,11 @@ func (s *Server) setRouter(router *gin.Engine) {
 	router.GET("/scrape_configs", s.ScrapeConfigsHandler)
 	router.GET("/jobs", s.JobsHandler)
 	router.GET("/jobs/:job_id/targets", s.TargetsHandler)
-	router.GET("/metrics", gin.WrapH(promhttp.Handler()))
+	// The handler is resolved per request so that the gatherer configured via
+	// WithMetricsGatherer (applied after the router is built) is honored.
+	router.GET("/metrics", func(c *gin.Context) {
+		s.metricsHandler().ServeHTTP(c.Writer, c.Request)
+	})
 	router.GET("/livez", s.LivenessProbeHandler)
 	router.GET("/readyz", s.ReadinessProbeHandler)
 	registerPprof(router.Group("/debug/pprof/"))

cmd/otel-allocator/main.go

@@ -9,20 +9,24 @@ import (
 	"fmt"
 	"os"
 	"os/signal"
+	"regexp"
 	"strings"
 	"syscall"
 
 	"github.com/oklog/run"
 	monitoringclient "github.com/prometheus-operator/prometheus-operator/pkg/client/versioned"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/prometheus/discovery"
+	prometheusbridge "go.opentelemetry.io/contrib/bridges/prometheus"
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/attribute"
-	otelprom "go.opentelemetry.io/otel/exporters/prometheus"
 	"go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc"
 	"go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp"
+	otelprom "go.opentelemetry.io/otel/exporters/prometheus"
 	"go.opentelemetry.io/otel/metric"
 	sdkmetric "go.opentelemetry.io/otel/sdk/metric"
+	"go.opentelemetry.io/otel/sdk/resource"
+	semconv "go.opentelemetry.io/otel/semconv/v1.27.0"
 	"k8s.io/client-go/kubernetes"
 	_ "k8s.io/client-go/plugin/pkg/client/auth/gcp"
 	ctrl "sigs.k8s.io/controller-runtime"
@@ -84,7 +88,14 @@ func main() {
 		os.Exit(1)
 	}
 
-	metricExporter, promErr := otelprom.New()
+	// The OTel SDK metrics are exported to Prometheus through a dedicated registry
+	// rather than the default one. This keeps them separate from the metrics that are
+	// registered directly on the default Prometheus registry (Prometheus service
+	// discovery internals, Go runtime and process collectors). That separation lets the
+	// OTLP reader below pull those Prometheus-only metrics via the Prometheus bridge
+	// without double-counting the SDK metrics, which it already collects natively.
+	sdkRegistry := prometheus.NewRegistry()
+	metricExporter, promErr := otelprom.New(otelprom.WithRegisterer(sdkRegistry))
 	if promErr != nil {
 		panic(promErr)
 	}
@@ -96,11 +107,23 @@ func main() {
 			setupLog.Error(otlpErr, "Failed to create OTLP metric reader")
 			os.Exit(1)
 		}
-		meterProviderOpts = append(meterProviderOpts, sdkmetric.WithReader(otlpReader))
+		telemetryResource, resErr := telemetryResource(ctx)
+		if resErr != nil {
+			setupLog.Error(resErr, "Failed to build telemetry resource")
+			os.Exit(1)
+		}
+		meterProviderOpts = append(meterProviderOpts, sdkmetric.WithReader(otlpReader), sdkmetric.WithResource(telemetryResource))
 	}
 
 	meterProvider := sdkmetric.NewMeterProvider(meterProviderOpts...)
 	otel.SetMeterProvider(meterProvider)
+	defer func() {
+		// Flush and close exporters (notably the OTLP reader) on shutdown so the final
+		// batch of metrics is delivered.
+		if shutdownErr := meterProvider.Shutdown(context.Background()); shutdownErr != nil {
+			setupLog.Error(shutdownErr, "Error shutting down meter provider")
+		}
+	}()
 
 	allocatorPrehook = prehook.New(cfg.FilterStrategy, log)
 	allocator, allocErr := allocation.New(cfg.AllocationStrategy, log, allocation.WithFilter(allocatorPrehook), allocation.WithFallbackStrategy(cfg.AllocationFallbackStrategy))
@@ -123,6 +146,10 @@ func main() {
 	if cfg.AllowInsecureAuthSecrets {
 		httpOptions = append(httpOptions, server.WithInsecureAuthSecrets())
 	}
+	// Serve /metrics from both the default registry (Prometheus service discovery, Go
+	// runtime and process collectors) and the dedicated SDK registry (OTel SDK metrics),
+	// preserving the full metric set that used to be exposed via the default registry alone.
+	httpOptions = append(httpOptions, server.WithMetricsGatherer(prometheus.Gatherers{prometheus.DefaultGatherer, sdkRegistry}))
 	srv, serverErr := server.NewServer(log, allocator, cfg.ListenAddr, httpOptions...)
 	if serverErr != nil {
 		panic(serverErr)
@@ -316,18 +343,48 @@ func main() {
 	setupLog.Info("Target allocator exited.")
 }
 
+// envRefPattern matches ${env:VAR} references, mirroring the OTel Collector's
+// environment-variable substitution syntax.
+var envRefPattern = regexp.MustCompile(`\$\{env:([a-zA-Z_][a-zA-Z0-9_]*)\}`)
+
+// expandEnvRefs replaces ${env:VAR} occurrences in s with the value of the
+// environment variable VAR (empty if unset). This lets sensitive header values
+// (e.g. an API token) be sourced from the pod environment / a Secret at runtime
+// rather than being written in plaintext into the Target Allocator ConfigMap.
+func expandEnvRefs(s string) string {
+	return envRefPattern.ReplaceAllStringFunc(s, func(match string) string {
+		name := envRefPattern.FindStringSubmatch(match)[1]
+		return os.Getenv(name)
+	})
+}
+
+// expandHeaders returns a copy of headers with ${env:VAR} references expanded in
+// every value.
+func expandHeaders(headers map[string]string) map[string]string {
+	if len(headers) == 0 {
+		return nil
+	}
+	expanded := make(map[string]string, len(headers))
+	for k, v := range headers {
+		expanded[k] = expandEnvRefs(v)
+	}
+	return expanded
+}
+
 // newOTLPMetricReader creates a PeriodicExportingMetricReader backed by either
 // an OTLP/gRPC or OTLP/HTTP exporter depending on cfg.Protocol.
 func newOTLPMetricReader(ctx context.Context, cfg *config.OTLPExporterConfig) (sdkmetric.Reader, error) {
 	temporality := temporalitySelector(cfg.Temporality)
 	readerOpts := periodicReaderOptions(cfg)
+	endpoint := expandEnvRefs(cfg.Endpoint)
+	headers := expandHeaders(cfg.Headers)
 	switch cfg.Protocol {
 	case "http":
 		// WithEndpointURL does not append /v1/metrics automatically. We do it here
 		// so the endpoint convention matches the OTel collector's otlphttp exporter
 		// (where users specify the base URL, not the full signal path).
 		// Skip if the caller already included the signal path.
-		endpoint := strings.TrimRight(cfg.Endpoint, "/")
+		endpoint = strings.TrimRight(endpoint, "/")
 		if !strings.HasSuffix(endpoint, "/v1/metrics") {
 			endpoint += "/v1/metrics"
 		}
@@ -338,8 +395,8 @@ func newOTLPMetricReader(ctx context.Context, cfg *config.OTLPExporterConfig) (s
 		if cfg.Insecure {
 			opts = append(opts, otlpmetrichttp.WithInsecure())
 		}
-		if len(cfg.Headers) > 0 {
-			opts = append(opts, otlpmetrichttp.WithHeaders(cfg.Headers))
+		if len(headers) > 0 {
+			opts = append(opts, otlpmetrichttp.WithHeaders(headers))
 		}
 		exp, err := otlpmetrichttp.New(ctx, opts...)
 		if err != nil {
@@ -348,14 +405,21 @@ func newOTLPMetricReader(ctx context.Context, cfg *config.OTLPExporterConfig) (s
 		return sdkmetric.NewPeriodicReader(exp, readerOpts...), nil
 	default: // "grpc"
 		opts := []otlpmetricgrpc.Option{
-			otlpmetricgrpc.WithEndpoint(cfg.Endpoint),
 			otlpmetricgrpc.WithTemporalitySelector(temporality),
 		}
+		// Accept both a full URL (with scheme, consistent with the HTTP protocol and
+		// the collector's otlp exporter) and a bare host:port. WithEndpointURL requires
+		// a scheme; WithEndpoint expects host:port.
+		if strings.Contains(endpoint, "://") {
+			opts = append(opts, otlpmetricgrpc.WithEndpointURL(endpoint))
+		} else {
+			opts = append(opts, otlpmetricgrpc.WithEndpoint(endpoint))
+		}
 		if cfg.Insecure {
 			opts = append(opts, otlpmetricgrpc.WithInsecure())
 		}
-		if len(cfg.Headers) > 0 {
-			opts = append(opts, otlpmetricgrpc.WithHeaders(cfg.Headers))
+		if len(headers) > 0 {
+			opts = append(opts, otlpmetricgrpc.WithHeaders(headers))
 		}
 		exp, err := otlpmetricgrpc.New(ctx, opts...)
 		if err != nil {
@@ -367,7 +431,15 @@ func newOTLPMetricReader(ctx context.Context, cfg *config.OTLPExporterConfig) (s
 
 // periodicReaderOptions builds PeriodicReaderOptions from the export interval and timeout config.
 func periodicReaderOptions(cfg *config.OTLPExporterConfig) []sdkmetric.PeriodicReaderOption {
-	var opts []sdkmetric.PeriodicReaderOption
+	opts := []sdkmetric.PeriodicReaderOption{
+		// Bridge the metrics registered directly on the default Prometheus registry
+		// (Prometheus service discovery internals, Go runtime and process collectors)
+		// into the OTLP export. Without this, those metrics would be visible on the
+		// Prometheus /metrics endpoint but missing from OTLP. The SDK's own metrics live
+		// on a separate registry and are collected natively by this reader, so bridging
+		// the default registry does not double-count them.
+		sdkmetric.WithProducer(prometheusbridge.NewMetricProducer()),
+	}
 	if cfg.ExportInterval > 0 {
 		opts = append(opts, sdkmetric.WithInterval(cfg.ExportInterval))
 	}
@@ -377,6 +449,19 @@ func periodicReaderOptions(cfg *config.OTLPExporterConfig) []sdkmetric.PeriodicR
 	return opts
 }
 
+// telemetryResource builds the OpenTelemetry resource attached to exported metrics.
+// It sets a default service.name and honors the standard OTEL_SERVICE_NAME /
+// OTEL_RESOURCE_ATTRIBUTES environment variables, so additional attributes (e.g. k8s.*)
+// can be injected without code changes.
+func telemetryResource(ctx context.Context) (*resource.Resource, error) {
+	return resource.New(ctx,
+		resource.WithTelemetrySDK(),
+		resource.WithAttributes(semconv.ServiceName("target-allocator")),
+		// Applied last so environment configuration wins on conflict.
+		resource.WithFromEnv(),
+	)
+}
+
 // temporalitySelector maps a config string to an SDK TemporalitySelector.
 // "delta" → all instruments delta; "lowmemory" → delta for counters/histograms,
 // cumulative for gauges; anything else → cumulative (SDK default).