consul/agent/hcp/telemetry/otel_exporter.go

package telemetry

import (
	"context"
	"fmt"
	"net/url"

	goMetrics "github.com/armon/go-metrics"
	"go.opentelemetry.io/otel/sdk/metric"
	"go.opentelemetry.io/otel/sdk/metric/aggregation"
	"go.opentelemetry.io/otel/sdk/metric/metricdata"

	hcpclient "github.com/hashicorp/consul/agent/hcp/client"
)

// OTELExporter is a custom implementation of a OTEL Metrics SDK metrics.Exporter.
// The exporter is used by a OTEL Metrics SDK PeriodicReader to export aggregated metrics.
// This allows us to use a custom client - HCP authenticated MetricsClient.
type OTELExporter struct {
	client   hcpclient.MetricsClient
	endpoint *url.URL
}

// NewOTELExporter returns a configured OTELExporter
func NewOTELExporter(client hcpclient.MetricsClient, endpoint *url.URL) *OTELExporter {
	return &OTELExporter{
		client:   client,
		endpoint: endpoint,
	}
}

// Temporality returns the Cumulative temporality for metrics aggregation.
// Telemetry Gateway stores metrics in Prometheus format, so use Cummulative aggregation as default.
func (e *OTELExporter) Temporality(_ metric.InstrumentKind) metricdata.Temporality {
	return metricdata.CumulativeTemporality
}

// Aggregation returns the Aggregation to use for an instrument kind.
// The default implementation provided by the OTEL Metrics SDK library DefaultAggregationSelector panics.
// This custom version replicates that logic, but removes the panic.
func (e *OTELExporter) Aggregation(kind metric.InstrumentKind) aggregation.Aggregation {
	switch kind {
	case metric.InstrumentKindObservableGauge:
		return aggregation.LastValue{}
	case metric.InstrumentKindHistogram:
		return aggregation.ExplicitBucketHistogram{
			Boundaries: []float64{0, 5, 10, 25, 50, 75, 100, 250, 500, 750, 1000, 2500, 5000, 7500, 10000},
			NoMinMax:   false,
		}
	}
	// for metric.InstrumentKindCounter and others, default to sum.
	return aggregation.Sum{}
}

// Export serializes and transmits metric data to a receiver.
func (e *OTELExporter) Export(ctx context.Context, metrics *metricdata.ResourceMetrics) error {
	otlpMetrics := transformOTLP(metrics)
	if isEmpty(otlpMetrics) {
		return nil
	}
	err := e.client.ExportMetrics(ctx, otlpMetrics, e.endpoint.String())
	if err != nil {
		goMetrics.IncrCounter(internalMetricExportFailure, 1)
		return fmt.Errorf("failed to export metrics: %w", err)
	}

	goMetrics.IncrCounter(internalMetricExportSuccess, 1)
	return nil
}

// ForceFlush is a no-op, as the MetricsClient client holds no state.
func (e *OTELExporter) ForceFlush(ctx context.Context) error {
	goMetrics.IncrCounter(internalMetricExporterForceFlush, 1)
	return ctx.Err()
}

// Shutdown is a no-op, as the MetricsClient is a HTTP client that requires no graceful shutdown.
func (e *OTELExporter) Shutdown(ctx context.Context) error {
	goMetrics.IncrCounter(internalMetricExporterShutdown, 1)
	return ctx.Err()
}