feat(adxexporter): Switch MetricsExporter to OTLP push mode (#1029)

* feat(adxexporter): Switch MetricsExporter to OTLP push mode

Replace Prometheus scraping with direct OTLP push using PromToOtlpExporter.

Changes:
- Add ToWriteRequest() to transform MetricData to prompb.WriteRequest with pooling
- Update MetricsExporterReconciler to use PromToOtlpExporter instead of OTel Meter
- Update CLI: --otlp-endpoint now required, remove deprecated --metrics-* flags
- Update tests to use mock OTLP server instead of metrics scraping
- Update documentation for OTLP push mode

Benefits:
- Timestamp fidelity: preserves actual KQL query timestamps
- Memory efficiency: uses pkg/prompb object pooling for high cardinality
- Simpler deployment: no Collector scraping configuration needed
- Reuses battle-tested PromToOtlpExporter from Collector

* Add --add-resource-attributes flag to AdxExporter

- Add --add-resource-attributes CLI flag for specifying OTLP resource attributes
- Merge ClusterLabels (base) with AddResourceAttributes (can override or add new)
- Add parseKeyValuePairs() helper function
- Add tests for resource attribute merging behavior

* fix(adxexporter): Add --add-resource-attributes flag to AdxExporter

Switches MetricsExporter from Prometheus scraping to OTLP push mode.
The metricNamePrefix field in the CRD controls the full metric name prefix.

Users who previously had metrics with 'adxexporter_' prefix should update
their CRDs to include it explicitly in metricNamePrefix, e.g.:
  metricNamePrefix: adxexporter_cluster_autoscaler

* feat(adxexporter): Add --default-metric-name-prefix CLI flag

Adds a new CLI flag that sets a default prefix for metric names when the
MetricsExporter CRD doesn't specify a metricNamePrefix. This provides:

1. Cluster-wide default naming convention without modifying every CRD
2. CRD-level override when specific prefixes are needed
3. Better user experience - operators configure once, users don't need to
   remember to add prefixes

Metric name precedence:
1. CRD's metricNamePrefix (if specified)
2. CLI's --default-metric-name-prefix (if CRD prefix is empty)
3. No prefix (if both are empty)

Example deployment:
  adxexporter --default-metric-name-prefix=adxexporter ...

This produces metrics like 'adxexporter_cluster_autoscaler_availability_*'
without requiring each MetricsExporter CRD to specify the prefix.

* feat(adxexporter): Change --metric-name-prefix to always prepend to CRD prefix

BREAKING CHANGE: Renamed --default-metric-name-prefix to --metric-name-prefix

The new behavior always prepends the CLI prefix to metric names, then appends
the CRD's metricNamePrefix. This allows operators to enforce a global prefix
(e.g., for allow-list compliance) while teams can still add sub-prefixes via CRD.

Old behavior (--default-metric-name-prefix):
- CLI: adxexporter, CRD: custom → Result: custom_* (CRD wins)

New behavior (--metric-name-prefix):
- CLI: adxexporter, CRD: custom → Result: adxexporter_custom_*
- CLI: adxexporter, CRD: empty → Result: adxexporter_*
- CLI: empty, CRD: custom → Result: custom_*

This ensures all MetricsExporter metrics can be prefixed uniformly without
requiring each CRD to manually include the global prefix.

Author: jessejlt

adxexporter/criteriaexpression_test.go

@@ -3,6 +3,9 @@ package adxexporter
 import (
 	"context"
 	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
 	"testing"
 	"time"
 
@@ -158,15 +161,25 @@ func TestMetricsExporter_CriteriaExpressionProcessing(t *testing.T) {
 			if tt.expectExecuted {
 				mockKusto.SetNextResult(t, [][]interface{}{{"test_metric", 1.0, time.Now()}})
 			}
+
+			// Create mock OTLP server
+			mockServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				io.Copy(io.Discard, r.Body)
+				r.Body.Close()
+				w.WriteHeader(http.StatusOK)
+				w.Write([]byte{})
+			}))
+			defer mockServer.Close()
+
 			reconciler := &MetricsExporterReconciler{
-				Client:                fakeClient,
-				Scheme:                s,
-				ClusterLabels:         labels,
-				KustoClusters:         map[string]string{"test-db": "https://test-cluster.kusto.windows.net"},
-				QueryExecutors:        map[string]*QueryExecutor{"test-db": NewQueryExecutor(mockKusto)},
-				EnableMetricsEndpoint: false,
+				Client:         fakeClient,
+				Scheme:         s,
+				ClusterLabels:  labels,
+				KustoClusters:  map[string]string{"test-db": "https://test-cluster.kusto.windows.net"},
+				QueryExecutors: map[string]*QueryExecutor{"test-db": NewQueryExecutor(mockKusto)},
+				OTLPEndpoint:   mockServer.URL,
 			}
-			require.NoError(t, reconciler.exposeMetricsServer())
+			require.NoError(t, reconciler.initOtlpExporter())
 			req := reconcile.Request{NamespacedName: types.NamespacedName{Name: me.Name, Namespace: me.Namespace}}
 			res, err := reconciler.Reconcile(context.Background(), req)
 			require.NoError(t, err)

adxexporter/metricsexporter.go

@@ -3,25 +3,21 @@ package adxexporter
 import (
 	"context"
 	"fmt"
+	"strings"
 	"time"
 
 	adxmonv1 "github.com/Azure/adx-mon/api/v1"
+	"github.com/Azure/adx-mon/collector/export"
 	"github.com/Azure/adx-mon/pkg/kustoutil"
 	"github.com/Azure/adx-mon/pkg/logger"
+	"github.com/Azure/adx-mon/pkg/prompb"
 	"github.com/Azure/adx-mon/transform"
-	"go.opentelemetry.io/otel"
-	"go.opentelemetry.io/otel/attribute"
-	"go.opentelemetry.io/otel/exporters/prometheus"
-	"go.opentelemetry.io/otel/metric"
-	"go.opentelemetry.io/otel/metric/noop"
-	sdkmetric "go.opentelemetry.io/otel/sdk/metric"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/utils/clock"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
-	crmetrics "sigs.k8s.io/controller-runtime/pkg/metrics"
 )
 
 // MetricsExporterReconciler reconciles MetricsExporter objects
@@ -33,16 +29,18 @@ type MetricsExporterReconciler struct {
 	ClusterLabels         map[string]string
 	KustoClusters         map[string]string // database name -> endpoint URL
 	OTLPEndpoint          string
-	EnableMetricsEndpoint bool
-	MetricsPort           string // Used for controller-runtime metrics server configuration
-	MetricsPath           string // For documentation/consistency (controller-runtime uses /metrics)
+	AddResourceAttributes map[string]string // Additional OTLP resource attributes (merged with ClusterLabels)
+	MetricNamePrefix      string            // Global prefix prepended to all metric names (combined with CRD prefix)
+	EnableMetricsEndpoint bool              // Deprecated: kept for backward compatibility, not used with OTLP push
+	MetricsPort           string            // Deprecated: kept for backward compatibility
+	MetricsPath           string            // Deprecated: kept for backward compatibility
 
 	// Query execution components
 	QueryExecutors map[string]*QueryExecutor // keyed by database name
 	Clock          clock.Clock
 
-	// Metrics components
-	Meter metric.Meter
+	// OTLP push client for metrics delivery
+	OtlpExporter *export.PromToOtlpExporter
 }
 
 // Reconcile handles MetricsExporter reconciliation
@@ -101,34 +99,41 @@ func (r *MetricsExporterReconciler) Reconcile(ctx context.Context, req ctrl.Requ
 	return ctrl.Result{RequeueAfter: requeueAfter}, nil
 }
 
-func (r *MetricsExporterReconciler) exposeMetricsServer() error {
-	if !r.EnableMetricsEndpoint {
-		r.Meter = noop.NewMeterProvider().Meter("noop")
-		return nil
+func (r *MetricsExporterReconciler) initOtlpExporter() error {
+	if r.OTLPEndpoint == "" {
+		return fmt.Errorf("OTLP endpoint is required: specify --otlp-endpoint")
 	}
 
-	// Register with controller-runtime's shared metrics registry, replacing the default registry
-	exporter, err := prometheus.New(
-		prometheus.WithRegisterer(crmetrics.Registry),
-		// Adds a namespace prefix to all metrics
-		prometheus.WithNamespace("adxexporter"),
-		// Disables the long otel specific scope string since we're only exposing through metrics
-		prometheus.WithoutScopeInfo(),
-	)
-	if err != nil {
-		return fmt.Errorf("failed to create metrics exporter: %w", err)
+	// Create resource attributes by merging cluster labels with explicit attributes.
+	// ClusterLabels provide a base set of resource attributes for identification.
+	// AddResourceAttributes allows explicit overrides or additional attributes.
+	resourceAttrs := make(map[string]string, len(r.ClusterLabels)+len(r.AddResourceAttributes))
+	for k, v := range r.ClusterLabels {
+		resourceAttrs[k] = v
+	}
+	// Explicit attributes take precedence over cluster labels
+	for k, v := range r.AddResourceAttributes {
+		resourceAttrs[k] = v
+	}
+
+	// Create a pass-through transformer (no filtering - KQL already selected the data)
+	transformer := &transform.RequestTransformer{
+		DefaultDropMetrics: false,
 	}
-	provider := sdkmetric.NewMeterProvider(sdkmetric.WithReader(exporter))
-	otel.SetMeterProvider(provider)
 
-	r.Meter = otel.GetMeterProvider().Meter("adxexporter")
+	opts := export.PromToOtlpExporterOpts{
+		Transformer:           transformer,
+		Destination:           r.OTLPEndpoint,
+		AddResourceAttributes: resourceAttrs,
+	}
 
+	r.OtlpExporter = export.NewPromToOtlpExporter(opts)
 	return nil
 }
 
 // SetupWithManager sets up the service with the Manager
 func (r *MetricsExporterReconciler) SetupWithManager(mgr ctrl.Manager) error {
-	if err := r.exposeMetricsServer(); err != nil {
+	if err := r.initOtlpExporter(); err != nil {
 		return err
 	}
 
@@ -265,7 +270,7 @@ func (r *MetricsExporterReconciler) executeMetricsExporter(ctx context.Context,
 	return true, nil
 }
 
-// transformAndRegisterMetrics converts KQL query results to metrics and registers them
+// transformAndRegisterMetrics converts KQL query results to metrics and pushes them via OTLP
 func (r *MetricsExporterReconciler) transformAndRegisterMetrics(ctx context.Context, me *adxmonv1.MetricsExporter, rows []map[string]any) error {
 	if len(rows) == 0 {
 		if logger.IsDebug() {
@@ -274,17 +279,32 @@ func (r *MetricsExporterReconciler) transformAndRegisterMetrics(ctx context.Cont
 		return nil
 	}
 
+	// Build the effective metric name prefix by combining CLI and CRD prefixes:
+	// - CLI prefix (--metric-name-prefix) is always prepended first if set
+	// - CRD prefix (transform.metricNamePrefix) is appended after CLI prefix if set
+	// This ensures operators can enforce a global prefix (e.g., for allow-lists)
+	// while teams can still add their own sub-prefixes via CRD.
+	var prefixParts []string
+	if r.MetricNamePrefix != "" {
+		prefixParts = append(prefixParts, r.MetricNamePrefix)
+	}
+	if me.Spec.Transform.MetricNamePrefix != "" {
+		prefixParts = append(prefixParts, me.Spec.Transform.MetricNamePrefix)
+	}
+	effectivePrefix := strings.Join(prefixParts, "_")
+
 	// Create transformer with the MetricsExporter's transform configuration
+	// Note: meter parameter is nil since we're using OTLP push instead of OTel SDK registration
 	transformer := transform.NewKustoToMetricsTransformer(
 		transform.TransformConfig{
 			MetricNameColumn:  me.Spec.Transform.MetricNameColumn,
-			MetricNamePrefix:  me.Spec.Transform.MetricNamePrefix,
+			MetricNamePrefix:  effectivePrefix,
 			ValueColumns:      me.Spec.Transform.ValueColumns,
 			TimestampColumn:   me.Spec.Transform.TimestampColumn,
 			LabelColumns:      me.Spec.Transform.LabelColumns,
 			DefaultMetricName: me.Spec.Transform.DefaultMetricName,
 		},
-		r.Meter,
+		nil, // meter not needed for OTLP push path
 	)
 
 	// Validate the transform configuration against the query results
@@ -298,44 +318,32 @@ func (r *MetricsExporterReconciler) transformAndRegisterMetrics(ctx context.Cont
 		return fmt.Errorf("failed to transform rows to metrics: %w", err)
 	}
 
-	if err := r.registerMetrics(ctx, metrics); err != nil {
-		return fmt.Errorf("failed to register metrics: %w", err)
+	if err := r.pushMetrics(ctx, metrics); err != nil {
+		return fmt.Errorf("failed to push metrics: %w", err)
 	}
 
-	logger.Infof("Successfully transformed and registered %d metrics for MetricsExporter %s/%s",
+	logger.Infof("Successfully transformed and pushed %d metrics for MetricsExporter %s/%s",
 		len(metrics), me.Namespace, me.Name)
 
 	return nil
 }
 
-func (r *MetricsExporterReconciler) registerMetrics(ctx context.Context, metrics []transform.MetricData) error {
-	// Group metrics by name for efficient registration
-	metricsByName := make(map[string][]transform.MetricData)
-	for _, metric := range metrics {
-		metricsByName[metric.Name] = append(metricsByName[metric.Name], metric)
+// pushMetrics converts MetricData to WriteRequest and sends via OTLP
+func (r *MetricsExporterReconciler) pushMetrics(ctx context.Context, metrics []transform.MetricData) error {
+	if len(metrics) == 0 {
+		return nil
 	}
 
-	// Register each unique metric name as a gauge
-	for metricName, metricData := range metricsByName {
-		gauge, err := r.Meter.Float64Gauge(metricName)
-		if err != nil {
-			return fmt.Errorf("failed to create gauge for metric '%s': %w", metricName, err)
-		}
-
-		// Record all data points for this metric
-		for _, data := range metricData {
-			// Convert labels to OpenTelemetry attributes
-			attrs := make([]attribute.KeyValue, 0, len(data.Labels))
-			for key, value := range data.Labels {
-				attrs = append(attrs, attribute.String(key, value))
-			}
+	// Convert to WriteRequest using pooled objects for efficiency
+	wr := transform.ToWriteRequest(metrics)
+	defer func() {
+		// Return pooled objects
+		wr.Reset()
+		prompb.WriteRequestPool.Put(wr)
+	}()
 
-			// Record the metric value
-			gauge.Record(ctx, data.Value, metric.WithAttributes(attrs...))
-		}
-	}
-
-	return nil
+	// Push via OTLP exporter
+	return r.OtlpExporter.Write(ctx, wr)
 }
 
 // initializeQueryExecutors creates QueryExecutors for all configured databases