[exporter/loadbalancing] allow metrics routing

.chloggen/loadbalance-metrics.yaml

@@ -0,0 +1,27 @@
+# Use this changelog template to create an entry for release notes.
+
+# 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. filelogreceiver)
+component: exporter/loadbalancing
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Allow metrics routing
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [25858]
+
+# (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:
+
+# If your change doesn't affect end users or the exported elements of any package,
+# you should instead start your pull request title with [chore] or use the "Skip Changelog" label.
+# Optional: The change log or logs in which this entry should be included.
+# e.g. '[user]' or '[user, api]'
+# Include 'user' if the change is relevant to end users.
+# Include 'api' if there is a change to a library API.
+# Default: '[user]'
+change_logs: [user]

exporter/loadbalancingexporter/README.md

@@ -4,19 +4,32 @@
 | Status        |           |
 | ------------- |-----------|
 | Stability     | [beta]: traces, logs   |
+|               | [development]: metrics   |
 | Distributions | [contrib], [aws], [grafana], [observiq], [sumo] |
 | Issues        | [![Open issues](https://img.shields.io/github/issues-search/open-telemetry/opentelemetry-collector-contrib?query=is%3Aissue%20is%3Aopen%20label%3Aexporter%2Floadbalancing%20&label=open&color=orange&logo=opentelemetry)](https://github.com/open-telemetry/opentelemetry-collector-contrib/issues?q=is%3Aopen+is%3Aissue+label%3Aexporter%2Floadbalancing) [![Closed issues](https://img.shields.io/github/issues-search/open-telemetry/opentelemetry-collector-contrib?query=is%3Aissue%20is%3Aclosed%20label%3Aexporter%2Floadbalancing%20&label=closed&color=blue&logo=opentelemetry)](https://github.com/open-telemetry/opentelemetry-collector-contrib/issues?q=is%3Aclosed+is%3Aissue+label%3Aexporter%2Floadbalancing) |
 | [Code Owners](https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/main/CONTRIBUTING.md#becoming-a-code-owner)    | [@jpkrohling](https://www.github.com/jpkrohling) |
 
 [beta]: https://github.com/open-telemetry/opentelemetry-collector#beta
+[development]: https://github.com/open-telemetry/opentelemetry-collector#development
 [contrib]: https://github.com/open-telemetry/opentelemetry-collector-releases/tree/main/distributions/otelcol-contrib
 [aws]: https://github.com/aws-observability/aws-otel-collector
 [grafana]: https://github.com/grafana/agent
 [observiq]: https://github.com/observIQ/observiq-otel-collector
 [sumo]: https://github.com/SumoLogic/sumologic-otel-collector
 <!-- end autogenerated section -->
 
-This is an exporter that will consistently export spans and logs depending on the `routing_key` configured. If no `routing_key` is configured, the default routing mechanism is `traceID`. This means that spans belonging to the same `traceID` (or `service.name`, when `service` is used as the `routing_key`) will be sent to the same backend.
+This is an exporter that will consistently export spans, metrics and logs depending on the `routing_key` configured.
+
+The options for `routing_key` are: `service`, `traceID`, `metric` (metric name), `resource`.
+
+| routing_key        | can be used for |
+| ------------- |-----------|
+| service | logs, spans, metrics |
+| traceID | logs, spans |
+| resource | metrics |
+| metric | metrics |
+
+If no `routing_key` is configured, the default routing mechanism is `traceID`  for traces, while `service` is the default for metrics. This means that spans belonging to the same `traceID` (or `service.name`, when `service` is used as the `routing_key`) will be sent to the same backend.
 
 It requires a source of backend information to be provided: static, with a fixed list of backends, or DNS, with a hostname that will resolve to all IP addresses to use. The DNS resolver will periodically check for updates.
 

exporter/loadbalancingexporter/config.go

@@ -14,6 +14,8 @@ type routingKey int
 const (
 	traceIDRouting routingKey = iota
 	svcRouting
+	metricNameRouting
+	resourceRouting
 )
 
 // Config defines configuration for the exporter.

exporter/loadbalancingexporter/factory.go

@@ -25,6 +25,7 @@ func NewFactory() exporter.Factory {
 		createDefaultConfig,
 		exporter.WithTraces(createTracesExporter, metadata.TracesStability),
 		exporter.WithLogs(createLogsExporter, metadata.LogsStability),
+		exporter.WithMetrics(createMetricsExporter, metadata.MetricsStability),
 	)
 }
 
@@ -46,3 +47,7 @@ func createTracesExporter(_ context.Context, params exporter.CreateSettings, cfg
 func createLogsExporter(_ context.Context, params exporter.CreateSettings, cfg component.Config) (exporter.Logs, error) {
 	return newLogsExporter(params, cfg)
 }
+
+func createMetricsExporter(_ context.Context, params exporter.CreateSettings, cfg component.Config) (exporter.Metrics, error) {
+	return newMetricsExporter(params, cfg)
+}

exporter/loadbalancingexporter/metadata.yaml

@@ -4,6 +4,7 @@ status:
   class: exporter
   stability:
     beta: [traces, logs]
+    development: [metrics]
   distributions:
   - contrib
   - grafana

exporter/loadbalancingexporter/metrics_exporter.go

@@ -0,0 +1,215 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package loadbalancingexporter // import "github.com/open-telemetry/opentelemetry-collector-contrib/exporter/loadbalancingexporter"
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"sort"
+	"strings"
+	"sync"
+	"time"
+
+	"go.opencensus.io/stats"
+	"go.opencensus.io/tag"
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/consumer"
+	"go.opentelemetry.io/collector/exporter"
+	"go.opentelemetry.io/collector/exporter/otlpexporter"
+	"go.opentelemetry.io/collector/pdata/pcommon"
+	"go.opentelemetry.io/collector/pdata/pmetric"
+	conventions "go.opentelemetry.io/collector/semconv/v1.6.1"
+	"go.uber.org/multierr"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/batchpersignal"
+)
+
+var _ exporter.Metrics = (*metricExporterImp)(nil)
+
+type metricExporterImp struct {
+	loadBalancer loadBalancer
+	routingKey   routingKey
+
+	stopped    bool
+	shutdownWg sync.WaitGroup
+}
+
+func newMetricsExporter(params exporter.CreateSettings, cfg component.Config) (*metricExporterImp, error) {
+	exporterFactory := otlpexporter.NewFactory()
+
+	lb, err := newLoadBalancer(params, cfg, func(ctx context.Context, endpoint string) (component.Component, error) {
+		oCfg := buildExporterConfig(cfg.(*Config), endpoint)
+		return exporterFactory.CreateMetricsExporter(ctx, params, &oCfg)
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	metricExporter := metricExporterImp{loadBalancer: lb, routingKey: svcRouting}
+
+	switch cfg.(*Config).RoutingKey {
+	case "service", "":
+		// default case for empty routing key
+		metricExporter.routingKey = svcRouting
+	case "resource":
+		metricExporter.routingKey = resourceRouting
+	case "metric":
+		metricExporter.routingKey = metricNameRouting
+	default:
+		return nil, fmt.Errorf("unsupported routing_key: %q", cfg.(*Config).RoutingKey)
+	}
+	return &metricExporter, nil
+
+}
+
+func (e *metricExporterImp) Capabilities() consumer.Capabilities {
+	return consumer.Capabilities{MutatesData: false}
+}
+
+func (e *metricExporterImp) Start(ctx context.Context, host component.Host) error {
+	return e.loadBalancer.Start(ctx, host)
+}
+
+func (e *metricExporterImp) Shutdown(context.Context) error {
+	e.stopped = true
+	e.shutdownWg.Wait()
+	return nil
+}
+
+func (e *metricExporterImp) ConsumeMetrics(ctx context.Context, md pmetric.Metrics) error {
+	var errs error
+	batches := batchpersignal.SplitMetrics(md)
+	for _, batch := range batches {
+		errs = multierr.Append(errs, e.consumeMetric(ctx, batch))
+	}
+
+	return errs
+}
+
+func (e *metricExporterImp) consumeMetric(ctx context.Context, md pmetric.Metrics) error {
+	var exp component.Component
+	routingIds, err := routingIdentifiersFromMetrics(md, e.routingKey)
+	if err != nil {
+		return err
+	}
+	for rid := range routingIds {
+		endpoint := e.loadBalancer.Endpoint([]byte(rid))
+		exp, err = e.loadBalancer.Exporter(endpoint)
+		if err != nil {
+			return err
+		}
+
+		te, ok := exp.(exporter.Metrics)
+		if !ok {
+			return fmt.Errorf("unable to export metrics, unexpected exporter type: expected exporter.Metrics but got %T", exp)
+		}
+
+		start := time.Now()
+		err = te.ConsumeMetrics(ctx, md)
+		duration := time.Since(start)
+
+		if err == nil {
+			_ = stats.RecordWithTags(
+				ctx,
+				[]tag.Mutator{tag.Upsert(endpointTagKey, endpoint), successTrueMutator},
+				mBackendLatency.M(duration.Milliseconds()))
+		} else {
+			_ = stats.RecordWithTags(
+				ctx,
+				[]tag.Mutator{tag.Upsert(endpointTagKey, endpoint), successFalseMutator},
+				mBackendLatency.M(duration.Milliseconds()))
+		}
+	}
+
+	return err
+}
+
+func routingIdentifiersFromMetrics(mds pmetric.Metrics, key routingKey) (map[string]bool, error) {
+	ids := make(map[string]bool)
+
+	// no need to test "empty labels"
+	// no need to test "empty resources"
+
+	rs := mds.ResourceMetrics()
+	if rs.Len() == 0 {
+		return nil, errors.New("empty resource metrics")
+	}
+
+	ils := rs.At(0).ScopeMetrics()
+	if ils.Len() == 0 {
+		return nil, errors.New("empty scope metrics")
+	}
+
+	metrics := ils.At(0).Metrics()
+	if metrics.Len() == 0 {
+		return nil, errors.New("empty metrics")
+	}
+
+	for i := 0; i < rs.Len(); i++ {
+		resource := rs.At(i).Resource()
+		switch key {
+		default:
+		case svcRouting, traceIDRouting:
+			svc, ok := resource.Attributes().Get(conventions.AttributeServiceName)
+			if !ok {
+				return nil, errors.New("unable to get service name")
+			}
+			ids[svc.Str()] = true
+		case metricNameRouting:
+			sm := rs.At(i).ScopeMetrics()
+			for j := 0; j < sm.Len(); j++ {
+				metrics := sm.At(j).Metrics()
+				for k := 0; k < metrics.Len(); k++ {
+					md := metrics.At(k)
+					rKey := metricRoutingKey(md)
+					ids[rKey] = true
+				}
+			}
+		case resourceRouting:
+			sm := rs.At(i).ScopeMetrics()
+			for j := 0; j < sm.Len(); j++ {
+				metrics := sm.At(j).Metrics()
+				for k := 0; k < metrics.Len(); k++ {
+					md := metrics.At(k)
+					rKey := resourceRoutingKey(md, resource.Attributes())
+					ids[rKey] = true
+				}
+			}
+		}
+	}
+
+	return ids, nil
+
+}
+
+// maintain
+func sortedMapAttrs(attrs pcommon.Map) []string {
+	keys := make([]string, 0)
+	for k := range attrs.AsRaw() {
+		keys = append(keys, k)
+	}
+	sort.Strings(keys)
+
+	attrsHash := make([]string, 0)
+	for _, k := range keys {
+		attrsHash = append(attrsHash, k)
+		if v, ok := attrs.Get(k); ok {
+			attrsHash = append(attrsHash, v.AsString())
+		}
+	}
+	return attrsHash
+}
+
+func resourceRoutingKey(md pmetric.Metric, attrs pcommon.Map) string {
+	attrsHash := sortedMapAttrs(attrs)
+	attrsHash = append(attrsHash, md.Name())
+	routingRef := strings.Join(attrsHash, "")
+
+	return routingRef
+}
+
+func metricRoutingKey(md pmetric.Metric) string {
+	return md.Name()
+}