Work around high cardinality metrics from otelcol receivers

component/otelcol/config_debug_metrics.go

@@ -0,0 +1,16 @@
+package otelcol
+
+// DebugMetricsArguments configures internal metrics of the components
+type DebugMetricsArguments struct {
+	DisableHighCardinalityMetrics bool `river:"disable_high_cardinality_metrics,attr,optional"`
+}
+
+// DefaultDebugMetricsArguments holds default settings for DebugMetricsArguments.
+var DefaultDebugMetricsArguments = DebugMetricsArguments{
+	DisableHighCardinalityMetrics: false,
+}
+
+// SetToDefault implements river.Defaulter.
+func (args *DebugMetricsArguments) SetToDefault() {
+	*args = DefaultDebugMetricsArguments
+}

component/otelcol/exporter/exporter.go

@@ -12,6 +12,7 @@ import (
 	"github.com/grafana/agent/component/otelcol/internal/lazycollector"
 	"github.com/grafana/agent/component/otelcol/internal/lazyconsumer"
 	"github.com/grafana/agent/component/otelcol/internal/scheduler"
+	"github.com/grafana/agent/component/otelcol/internal/views"
 	"github.com/grafana/agent/pkg/build"
 	"github.com/grafana/agent/pkg/util/zapadapter"
 	"github.com/prometheus/client_golang/prometheus"
@@ -40,6 +41,9 @@ type Arguments interface {
 	// Exporters returns the set of exporters that are exposed to the configured
 	// component.
 	Exporters() map[otelcomponent.DataType]map[otelcomponent.ID]otelcomponent.Component
+
+	// DebugMetricsConfig returns the configuration for debug metrics
+	DebugMetricsConfig() otelcol.DebugMetricsArguments
 }
 
 // Exporter is a Flow component shim which manages an OpenTelemetry Collector
@@ -127,12 +131,17 @@ func (e *Exporter) Update(args component.Arguments) error {
 		return err
 	}
 
+	metricOpts := []metric.Option{metric.WithReader(promExporter)}
+	if eargs.DebugMetricsConfig().DisableHighCardinalityMetrics {
+		metricOpts = append(metricOpts, metric.WithView(views.DropHighCardinalityServerAttributes()...))
+	}
+
 	settings := otelexporter.CreateSettings{
 		TelemetrySettings: otelcomponent.TelemetrySettings{
 			Logger: zapadapter.New(e.opts.Logger),
 
 			TracerProvider: e.opts.Tracer,
-			MeterProvider:  metric.NewMeterProvider(metric.WithReader(promExporter)),
+			MeterProvider:  metric.NewMeterProvider(metricOpts...),
 		},
 
 		BuildInfo: otelcomponent.BuildInfo{

component/otelcol/exporter/exporter_test.go

@@ -121,8 +121,7 @@ func (te *testEnvironment) Start() {
 	}()
 }
 
-type fakeExporterArgs struct {
-}
+type fakeExporterArgs struct{}
 
 var _ exporter.Arguments = fakeExporterArgs{}
 
@@ -138,6 +137,10 @@ func (fa fakeExporterArgs) Exporters() map[otelcomponent.DataType]map[otelcompon
 	return nil
 }
 
+func (fe fakeExporterArgs) DebugMetricsConfig() otelcol.DebugMetricsArguments {
+	return otelcol.DefaultDebugMetricsArguments
+}
+
 type fakeExporter struct {
 	StartFunc         func(ctx context.Context, host otelcomponent.Host) error
 	ShutdownFunc      func(ctx context.Context) error
@@ -178,7 +181,7 @@ func (fe *fakeExporter) ConsumeTraces(ctx context.Context, td ptrace.Traces) err
 func createTestTraces() ptrace.Traces {
 	// Matches format from the protobuf definition:
 	// https://github.com/open-telemetry/opentelemetry-proto/blob/main/opentelemetry/proto/trace/v1/trace.proto
-	var bb = `{
+	bb := `{
 		"resource_spans": [{
 			"scope_spans": [{
 				"spans": [{

component/otelcol/exporter/jaeger/jaeger.go

@@ -33,12 +33,13 @@ type Arguments struct {
 	Queue otelcol.QueueArguments `river:"sending_queue,block,optional"`
 	Retry otelcol.RetryArguments `river:"retry_on_failure,block,optional"`
 
+	// DebugMetrics configures component internal metrics. Optional.
+	DebugMetrics otelcol.DebugMetricsArguments `river:"debug_metrics,block,optional"`
+
 	Client GRPCClientArguments `river:"client,block"`
 }
 
-var (
-	_ exporter.Arguments = Arguments{}
-)
+var _ exporter.Arguments = Arguments{}
 
 // DefaultArguments holds default values for Arguments.
 var DefaultArguments = Arguments{
@@ -75,6 +76,11 @@ func (args Arguments) Exporters() map[otelcomponent.DataType]map[otelcomponent.I
 	return nil
 }
 
+// DebugMetricsConfig implements receiver.Arguments.
+func (args Arguments) DebugMetricsConfig() otelcol.DebugMetricsArguments {
+	return args.DebugMetrics
+}
+
 // GRPCClientArguments is used to configure otelcol.exporter.jaeger with
 // component-specific defaults.
 type GRPCClientArguments otelcol.GRPCClientArguments

component/otelcol/exporter/loadbalancing/loadbalancing.go

@@ -38,6 +38,9 @@ type Arguments struct {
 	Protocol   Protocol         `river:"protocol,block"`
 	Resolver   ResolverSettings `river:"resolver,block"`
 	RoutingKey string           `river:"routing_key,attr,optional"`
+
+	// DebugMetrics configures component internal metrics. Optional.
+	DebugMetrics otelcol.DebugMetricsArguments `river:"debug_metrics,block,optional"`
 }
 
 var (
@@ -137,9 +140,7 @@ type DNSResolver struct {
 	Timeout  time.Duration `river:"timeout,attr,optional"`
 }
 
-var (
-	_ river.Defaulter = &DNSResolver{}
-)
+var _ river.Defaulter = &DNSResolver{}
 
 // DefaultDNSResolver holds default values for DNSResolver.
 var DefaultDNSResolver = DNSResolver{
@@ -172,6 +173,11 @@ func (args Arguments) Exporters() map[otelcomponent.DataType]map[otelcomponent.I
 	return nil
 }
 
+// DebugMetricsConfig implements receiver.Arguments.
+func (args Arguments) DebugMetricsConfig() otelcol.DebugMetricsArguments {
+	return args.DebugMetrics
+}
+
 // GRPCClientArguments is the same as otelcol.GRPCClientArguments, but without an "endpoint" attribute
 type GRPCClientArguments struct {
 	Compression otelcol.CompressionType `river:"compression,attr,optional"`
@@ -190,9 +196,7 @@ type GRPCClientArguments struct {
 	Auth *auth.Handler `river:"auth,attr,optional"`
 }
 
-var (
-	_ river.Defaulter = &GRPCClientArguments{}
-)
+var _ river.Defaulter = &GRPCClientArguments{}
 
 // Convert converts args into the upstream type.
 func (args *GRPCClientArguments) Convert() *otelconfiggrpc.GRPCClientSettings {

component/otelcol/exporter/logging/logging.go

@@ -29,11 +29,12 @@ type Arguments struct {
 	Verbosity          configtelemetry.Level `river:"verbosity,attr,optional"`
 	SamplingInitial    int                   `river:"sampling_initial,attr,optional"`
 	SamplingThereafter int                   `river:"sampling_thereafter,attr,optional"`
+
+	// DebugMetrics configures component internal metrics. Optional.
+	DebugMetrics otelcol.DebugMetricsArguments `river:"debug_metrics,block,optional"`
 }
 
-var (
-	_ exporter.Arguments = Arguments{}
-)
+var _ exporter.Arguments = Arguments{}
 
 // DefaultArguments holds default values for Arguments.
 var DefaultArguments = Arguments{
@@ -65,3 +66,8 @@ func (args Arguments) Extensions() map[otelcomponent.ID]otelextension.Extension
 func (args Arguments) Exporters() map[otelcomponent.DataType]map[otelcomponent.ID]otelcomponent.Component {
 	return nil
 }
+
+// DebugMetricsConfig implements receiver.Arguments.
+func (args Arguments) DebugMetricsConfig() otelcol.DebugMetricsArguments {
+	return args.DebugMetrics
+}

component/otelcol/exporter/otlp/otlp.go

@@ -33,12 +33,13 @@ type Arguments struct {
 	Queue otelcol.QueueArguments `river:"sending_queue,block,optional"`
 	Retry otelcol.RetryArguments `river:"retry_on_failure,block,optional"`
 
+	// DebugMetrics configures component internal metrics. Optional.
+	DebugMetrics otelcol.DebugMetricsArguments `river:"debug_metrics,block,optional"`
+
 	Client GRPCClientArguments `river:"client,block"`
 }
 
-var (
-	_ exporter.Arguments = Arguments{}
-)
+var _ exporter.Arguments = Arguments{}
 
 // DefaultArguments holds default values for Arguments.
 var DefaultArguments = Arguments{
@@ -75,6 +76,11 @@ func (args Arguments) Exporters() map[otelcomponent.DataType]map[otelcomponent.I
 	return nil
 }
 
+// DebugMetricsConfig implements receiver.Arguments.
+func (args Arguments) DebugMetricsConfig() otelcol.DebugMetricsArguments {
+	return args.DebugMetrics
+}
+
 // GRPCClientArguments is used to configure otelcol.exporter.otlp with
 // component-specific defaults.
 type GRPCClientArguments otelcol.GRPCClientArguments

component/otelcol/exporter/otlp/otlp_test.go

@@ -45,9 +45,14 @@ func Test(t *testing.T) {
 				insecure_skip_verify = true
 			}
 		}
+
+		debug_metrics {
+			disable_high_cardinality_metrics = true
+		}
 	`, tracesServer)
 	var args otlp.Arguments
 	require.NoError(t, river.Unmarshal([]byte(cfg), &args))
+	require.Equal(t, args.DebugMetricsConfig().DisableHighCardinalityMetrics, true)
 
 	go func() {
 		err := ctrl.Run(ctx, args)
@@ -121,7 +126,7 @@ func (ms *mockTracesReceiver) Export(_ context.Context, req ptraceotlp.ExportReq
 func createTestTraces() ptrace.Traces {
 	// Matches format from the protobuf definition:
 	// https://github.com/open-telemetry/opentelemetry-proto/blob/main/opentelemetry/proto/trace/v1/trace.proto
-	var bb = `{
+	bb := `{
 		"resource_spans": [{
 			"scope_spans": [{
 				"spans": [{

component/otelcol/exporter/otlphttp/otlphttp.go

@@ -32,6 +32,9 @@ type Arguments struct {
 	Queue  otelcol.QueueArguments `river:"sending_queue,block,optional"`
 	Retry  otelcol.RetryArguments `river:"retry_on_failure,block,optional"`
 
+	// DebugMetrics configures component internal metrics. Optional.
+	DebugMetrics otelcol.DebugMetricsArguments `river:"debug_metrics,block,optional"`
+
 	// The URLs to send metrics/logs/traces to. If omitted the exporter will
 	// use Client.Endpoint by appending "/v1/metrics", "/v1/logs" or
 	// "/v1/traces", respectively. If set, these settings override
@@ -41,9 +44,7 @@ type Arguments struct {
 	LogsEndpoint    string `river:"logs_endpoint,attr,optional"`
 }
 
-var (
-	_ exporter.Arguments = Arguments{}
-)
+var _ exporter.Arguments = Arguments{}
 
 // DefaultArguments holds default values for Arguments.
 var DefaultArguments = Arguments{
@@ -79,6 +80,11 @@ func (args Arguments) Exporters() map[otelcomponent.DataType]map[otelcomponent.I
 	return nil
 }
 
+// DebugMetricsConfig implements receiver.Arguments.
+func (args Arguments) DebugMetricsConfig() otelcol.DebugMetricsArguments {
+	return args.DebugMetrics
+}
+
 // Validate implements river.Validator.
 func (args *Arguments) Validate() error {
 	if args.Client.Endpoint == "" && args.TracesEndpoint == "" && args.MetricsEndpoint == "" && args.LogsEndpoint == "" {

component/otelcol/internal/views/views.go

@@ -0,0 +1,57 @@
+package views
+
+import (
+	semconv "go.opentelemetry.io/collector/semconv/v1.13.0"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/sdk/instrumentation"
+	"go.opentelemetry.io/otel/sdk/metric"
+)
+
+var (
+	grpcScope = "go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
+	// grpcUnacceptableKeyValues is a list of high cardinality grpc attributes that should be filtered out.
+	grpcUnacceptableKeyValues = []attribute.KeyValue{
+		attribute.String(semconv.AttributeNetSockPeerAddr, ""),
+		attribute.String(semconv.AttributeNetSockPeerPort, ""),
+		attribute.String(semconv.AttributeNetSockPeerName, ""),
+	}
+
+	httpScope = "go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
+	// httpUnacceptableKeyValues is a list of high cardinality http attributes that should be filtered out.
+	httpUnacceptableKeyValues = []attribute.KeyValue{
+		attribute.String(semconv.AttributeNetHostName, ""),
+		attribute.String(semconv.AttributeNetHostPort, ""),
+		attribute.String(semconv.AttributeNetSockPeerPort, ""),
+		attribute.String(semconv.AttributeNetSockPeerAddr, ""),
+		attribute.String(semconv.AttributeHTTPClientIP, ""),
+	}
+)
+
+func cardinalityFilter(kvs ...attribute.KeyValue) attribute.Filter {
+	filter := attribute.NewSet(kvs...)
+	return func(kv attribute.KeyValue) bool {
+		return !filter.HasValue(kv.Key)
+	}
+}
+
+// DropHighCardinalityServerAttributes drops certain high cardinality attributes from grpc/http server metrics
+//
+// This is a fix to an upstream issue:
+// https://github.com/open-telemetry/opentelemetry-go-contrib/issues/3765
+// The long-term solution for the Collector is to set view settings in the Collector config:
+// https://github.com/open-telemetry/opentelemetry-collector/issues/7517#issuecomment-1511168350
+// In the future, when Collector supports such config, we may want to support similar view settings in the Agent.
+func DropHighCardinalityServerAttributes() []metric.View {
+	var views []metric.View
+
+	views = append(views, metric.NewView(
+		metric.Instrument{Scope: instrumentation.Scope{Name: grpcScope}},
+		metric.Stream{AttributeFilter: cardinalityFilter(grpcUnacceptableKeyValues...)}))
+
+	views = append(views, metric.NewView(
+		metric.Instrument{Scope: instrumentation.Scope{Name: httpScope}},
+		metric.Stream{AttributeFilter: cardinalityFilter(httpUnacceptableKeyValues...)},
+	))
+
+	return views
+}

component/otelcol/receiver/jaeger/jaeger.go

@@ -31,13 +31,14 @@ func init() {
 type Arguments struct {
 	Protocols ProtocolsArguments `river:"protocols,block"`
 
+	// DebugMetrics configures component internal metrics. Optional.
+	DebugMetrics otelcol.DebugMetricsArguments `river:"debug_metrics,block,optional"`
+
 	// Output configures where to send received data. Required.
 	Output *otelcol.ConsumerArguments `river:"output,block"`
 }
 
-var (
-	_ receiver.Arguments = Arguments{}
-)
+var _ receiver.Arguments = Arguments{}
 
 // Validate implements river.Validator.
 func (args *Arguments) Validate() error {
@@ -210,3 +211,8 @@ func (args *ThriftBinary) Convert() *jaegerreceiver.ProtocolUDP {
 
 	return args.ProtocolUDP.Convert()
 }
+
+// DebugMetricsConfig implements receiver.Arguments.
+func (args Arguments) DebugMetricsConfig() otelcol.DebugMetricsArguments {
+	return args.DebugMetrics
+}

component/otelcol/receiver/kafka/kafka.go

@@ -41,13 +41,14 @@ type Arguments struct {
 	AutoCommit     AutoCommitArguments     `river:"autocommit,block,optional"`
 	MessageMarking MessageMarkingArguments `river:"message_marking,block,optional"`
 
+	// DebugMetrics configures component internal metrics. Optional.
+	DebugMetrics otelcol.DebugMetricsArguments `river:"debug_metrics,block,optional"`
+
 	// Output configures where to send received data. Required.
 	Output *otelcol.ConsumerArguments `river:"output,block"`
 }
 
-var (
-	_ receiver.Arguments = Arguments{}
-)
+var _ receiver.Arguments = Arguments{}
 
 // DefaultArguments holds default values for Arguments.
 var DefaultArguments = Arguments{
@@ -279,3 +280,8 @@ func (args MessageMarkingArguments) Convert() kafkareceiver.MessageMarking {
 		OnError: args.IncludeUnsuccessful,
 	}
 }
+
+// DebugMetricsConfig implements receiver.Arguments.
+func (args Arguments) DebugMetricsConfig() otelcol.DebugMetricsArguments {
+	return args.DebugMetrics
+}