Bump opentelemetry collector 0.85

CHANGELOG.md

@@ -109,6 +109,15 @@ Main (unreleased)
 
 - Agent Management: Honor 503 ServiceUnavailable `Retry-After` header. (@jcreixell)
 
+- Bump opentelemetry-collector and opentelemetry-collector-contrib versions from v0.80 to v0.85 (@wildum):
+  - add `authoriy` attribute to `otelcol.exporter.loadbalancing` to override the default value in gRPC requests.
+  - add `exemplars` support to `otelcol.connector.spanmetrics`.
+  - add `exclude_dimensions` attribute to `otelcol.connector.spanmetrics` to exclude dimensions from the default set.
+  - add `authority` attribute to `otelcol.receiver.otlp` to override the default value in gRPC requests.
+  - add `disable_keep_alives` attribute to `otelcol.receiver.otlp` to disable the HTTP keep alive feature.
+  - add `traces_url_path`, `metrics_url_path` and `logs_url_path` attributes to `otelcol.receiver.otlp` to specify the URl path to respectively receive traces, metrics and logs on.
+  - add the value `json` to the `encoding` attribute of `otelcol.receiver.kafka`. The component is now able to decode `json` payload and to insert it into the body of a log record.
+
 - Added `scrape` block to customize the default behavior of `prometheus.operator.podmonitors`, `prometheus.operator.probes`, and `prometheus.operator.servicemonitors`. (@sberz)
 
 - The `instance` label of targets exposed by `prometheus.exporter.*` components

component/loki/source/kubernetes_events/event_controller.go

@@ -96,9 +96,11 @@ func (ctrl *eventController) runError(ctx context.Context) error {
 		return fmt.Errorf("adding core to scheme: %w", err)
 	}
 
+	defaultNamespaces := map[string]cache.Config{}
+	defaultNamespaces[ctrl.task.Namespace] = cache.Config{}
 	opts := cache.Options{
-		Scheme:     scheme,
-		Namespaces: []string{ctrl.task.Namespace},
+		Scheme:            scheme,
+		DefaultNamespaces: defaultNamespaces,
 	}
 	informers, err := cache.New(ctrl.task.Config, opts)
 	if err != nil {

component/otelcol/config_grpc.go

@@ -144,6 +144,7 @@ type GRPCClientArguments struct {
 	WaitForReady    bool              `river:"wait_for_ready,attr,optional"`
 	Headers         map[string]string `river:"headers,attr,optional"`
 	BalancerName    string            `river:"balancer_name,attr,optional"`
+	Authority       string            `river:"authority,attr,optional"`
 
 	// Auth is a binding to an otelcol.auth.* component extension which handles
 	// authentication.
@@ -180,6 +181,7 @@ func (args *GRPCClientArguments) Convert() *otelconfiggrpc.GRPCClientSettings {
 		WaitForReady:    args.WaitForReady,
 		Headers:         opaqueHeaders,
 		BalancerName:    args.BalancerName,
+		Authority:       args.Authority,
 
 		Auth: auth,
 	}

component/otelcol/config_http.go

@@ -89,6 +89,7 @@ type HTTPClientArguments struct {
 	MaxIdleConnsPerHost *int           `river:"max_idle_conns_per_host,attr,optional"`
 	MaxConnsPerHost     *int           `river:"max_conns_per_host,attr,optional"`
 	IdleConnTimeout     *time.Duration `river:"idle_conn_timeout,attr,optional"`
+	DisableKeepAlives   bool           `river:"disable_keep_alives,attr,optional"`
 
 	// Auth is a binding to an otelcol.auth.* component extension which handles
 	// authentication.

component/otelcol/connector/spanmetrics/spanmetrics.go

@@ -38,7 +38,8 @@ type Arguments struct {
 	// - status.code
 	// The dimensions will be fetched from the span's attributes. Examples of some conventionally used attributes:
 	// https://github.com/open-telemetry/opentelemetry-collector/blob/main/model/semconv/opentelemetry.go.
-	Dimensions []Dimension `river:"dimension,block,optional"`
+	Dimensions        []Dimension `river:"dimension,block,optional"`
+	ExcludeDimensions []string    `river:"exclude_dimensions,attr,optional"`
 
 	// DimensionsCacheSize defines the size of cache for storing Dimensions, which helps to avoid cache memory growing
 	// indefinitely over the lifetime of the collector.
@@ -54,6 +55,9 @@ type Arguments struct {
 	// Namespace is the namespace of the metrics emitted by the connector.
 	Namespace string `river:"namespace,attr,optional"`
 
+	// Exemplars defines the configuration for exemplars.
+	Exemplars ExemplarsConfig `river:"exemplars,block,optional"`
+
 	// Output configures where to send processed data. Required.
 	Output *otelcol.ConsumerArguments `river:"output,block"`
 }
@@ -131,13 +135,17 @@ func (args Arguments) Convert() (otelcomponent.Config, error) {
 		return nil, err
 	}
 
+	excludeDimensions := append([]string(nil), args.ExcludeDimensions...)
+
 	return &spanmetricsconnector.Config{
 		Dimensions:             dimensions,
+		ExcludeDimensions:      excludeDimensions,
 		DimensionsCacheSize:    args.DimensionsCacheSize,
 		AggregationTemporality: aggregationTemporality,
 		Histogram:              *histogram,
 		MetricsFlushInterval:   args.MetricsFlushInterval,
 		Namespace:              args.Namespace,
+		Exemplars:              *args.Exemplars.Convert(),
 	}, nil
 }
 

component/otelcol/connector/spanmetrics/spanmetrics_test.go

@@ -33,9 +33,11 @@ func TestArguments_UnmarshalRiver(t *testing.T) {
 			`,
 			expected: spanmetricsconnector.Config{
 				Dimensions:             []spanmetricsconnector.Dimension{},
+				ExcludeDimensions:      nil,
 				DimensionsCacheSize:    1000,
 				AggregationTemporality: "AGGREGATION_TEMPORALITY_CUMULATIVE",
 				Histogram: spanmetricsconnector.HistogramConfig{
+					Disable:     false,
 					Unit:        0,
 					Exponential: nil,
 					Explicit: &spanmetricsconnector.ExplicitHistogramConfig{
@@ -61,6 +63,9 @@ func TestArguments_UnmarshalRiver(t *testing.T) {
 				},
 				MetricsFlushInterval: 15 * time.Second,
 				Namespace:            "",
+				Exemplars: spanmetricsconnector.ExemplarsConfig{
+					Enabled: false,
+				},
 			},
 		},
 		{
@@ -75,8 +80,10 @@ func TestArguments_UnmarshalRiver(t *testing.T) {
 			expected: spanmetricsconnector.Config{
 				Dimensions:             []spanmetricsconnector.Dimension{},
 				DimensionsCacheSize:    1000,
+				ExcludeDimensions:      nil,
 				AggregationTemporality: "AGGREGATION_TEMPORALITY_CUMULATIVE",
 				Histogram: spanmetricsconnector.HistogramConfig{
+					Disable:     false,
 					Unit:        0,
 					Exponential: &spanmetricsconnector.ExponentialHistogramConfig{MaxSize: 160},
 					Explicit:    nil,
@@ -95,16 +102,21 @@ func TestArguments_UnmarshalRiver(t *testing.T) {
 				name = "http.method"
 				default = "GET"
 			}
+			exclude_dimensions = ["test_exclude_dim1", "test_exclude_dim2"]
 			dimensions_cache_size = 333
 			aggregation_temporality = "DELTA"
 			histogram {
+				disable = true
 				unit = "s"
 				explicit {
 					buckets = ["333ms", "777s", "999h"]
 				}
 			}
 			metrics_flush_interval = "33s"
 			namespace = "test.namespace"
+			exemplars {
+				enabled = true
+			}
 
 			output {}
 			`,
@@ -113,9 +125,11 @@ func TestArguments_UnmarshalRiver(t *testing.T) {
 					{Name: "http.status_code", Default: nil},
 					{Name: "http.method", Default: getStringPtr("GET")},
 				},
+				ExcludeDimensions:      []string{"test_exclude_dim1", "test_exclude_dim2"},
 				DimensionsCacheSize:    333,
 				AggregationTemporality: "AGGREGATION_TEMPORALITY_DELTA",
 				Histogram: spanmetricsconnector.HistogramConfig{
+					Disable:     true,
 					Unit:        1,
 					Exponential: nil,
 					Explicit: &spanmetricsconnector.ExplicitHistogramConfig{
@@ -128,6 +142,9 @@ func TestArguments_UnmarshalRiver(t *testing.T) {
 				},
 				MetricsFlushInterval: 33 * time.Second,
 				Namespace:            "test.namespace",
+				Exemplars: spanmetricsconnector.ExemplarsConfig{
+					Enabled: true,
+				},
 			},
 		},
 		{

component/otelcol/connector/spanmetrics/types.go

@@ -37,7 +37,7 @@ const (
 // The unit is a private type in an internal Otel package,
 // so we need to convert it to a map and then back to the internal type.
 // ConvertMetricUnit matches the Unit type in this internal package:
-// https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/v0.80.0/connector/spanmetricsconnector/internal/metrics/unit.go
+// https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/v0.85.0/connector/spanmetricsconnector/internal/metrics/unit.go
 func ConvertMetricUnit(unit string) (map[string]interface{}, error) {
 	switch unit {
 	case MetricsUnitMilliseconds:
@@ -56,6 +56,7 @@ func ConvertMetricUnit(unit string) (map[string]interface{}, error) {
 }
 
 type HistogramConfig struct {
+	Disable     bool                        `river:"disable,attr,optional"`
 	Unit        string                      `river:"unit,attr,optional"`
 	Exponential *ExponentialHistogramConfig `river:"exponential,block,optional"`
 	Explicit    *ExplicitHistogramConfig    `river:"explicit,block,optional"`
@@ -117,9 +118,20 @@ func (hc HistogramConfig) Convert() (*spanmetricsconnector.HistogramConfig, erro
 		result.Explicit = hc.Explicit.Convert()
 	}
 
+	result.Disable = hc.Disable
 	return &result, nil
 }
 
+type ExemplarsConfig struct {
+	Enabled bool `river:"enabled,attr,optional"`
+}
+
+func (ec ExemplarsConfig) Convert() *spanmetricsconnector.ExemplarsConfig {
+	return &spanmetricsconnector.ExemplarsConfig{
+		Enabled: ec.Enabled,
+	}
+}
+
 type ExponentialHistogramConfig struct {
 	MaxSize int32 `river:"max_size,attr,optional"`
 }

component/otelcol/exporter/loadbalancing/loadbalancing.go

@@ -208,6 +208,7 @@ type GRPCClientArguments struct {
 	WaitForReady    bool              `river:"wait_for_ready,attr,optional"`
 	Headers         map[string]string `river:"headers,attr,optional"`
 	BalancerName    string            `river:"balancer_name,attr,optional"`
+	Authority       string            `river:"authority,attr,optional"`
 
 	// Auth is a binding to an otelcol.auth.* component extension which handles
 	// authentication.
@@ -244,6 +245,7 @@ func (args *GRPCClientArguments) Convert() *otelconfiggrpc.GRPCClientSettings {
 		WaitForReady:    args.WaitForReady,
 		Headers:         opaqueHeaders,
 		BalancerName:    args.BalancerName,
+		Authority:       args.Authority,
 
 		Auth: auth,
 	}

component/otelcol/exporter/loadbalancing/loadbalancing_test.go

@@ -107,7 +107,9 @@ func TestConfigConversion(t *testing.T) {
 			protocol {
 				otlp {
 					timeout = "1s"
-					client {}
+					client {
+						authority = "authority"
+					}
 				}
 			}
 			resolver {
@@ -129,6 +131,7 @@ func TestConfigConversion(t *testing.T) {
 							WriteBufferSize: 512 * 1024,
 							Headers:         map[string]configopaque.String{},
 							BalancerName:    "pick_first",
+							Authority:       "authority",
 						},
 					},
 				},

component/otelcol/exporter/loki/internal/convert/convert.go

@@ -16,9 +16,7 @@ import (
 	"github.com/go-kit/log/level"
 	"github.com/grafana/agent/component/common/loki"
 	loki_translator "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/loki"
-	prometheus_translator "github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/prometheus"
 	"github.com/prometheus/client_golang/prometheus"
-	"github.com/prometheus/common/model"
 	"go.opentelemetry.io/collector/consumer"
 	"go.opentelemetry.io/collector/pdata/plog"
 )
@@ -69,27 +67,19 @@ func (conv *Converter) ConsumeLogs(ctx context.Context, ld plog.Logs) error {
 			for k := 0; k < logs.Len(); k++ {
 				conv.metrics.entriesTotal.Inc()
 
-				entry, err := loki_translator.LogToLokiEntry(logs.At(k), rls.At(i).Resource(), scope)
+				// TODO: loki added a parameter `defaultLabelsEnabled` to this function to add the possibility to disable default labels (exporter, job, instance, level)
+				// Is this interesting for us in any ways? (@wildum)
+				// https://github.com/open-telemetry/opentelemetry-collector-contrib/pull/23863/files#diff-ef7831fcba373f6e8aa7f799b5b89f4e113b2064cd7ef1688286ce193d2256a8
+				entry, err := loki_translator.LogToLokiEntry(logs.At(k), rls.At(i).Resource(), scope, nil)
 				if err != nil {
 					level.Error(conv.log).Log("msg", "failed to convert log to loki entry", "err", err)
 					conv.metrics.entriesFailed.Inc()
 					continue
 				}
 
-				// TODO: Remove this code once loki_translator.LogToLokiEntry() has been updated to normalise labels.
-				// We should firstly upgrade to a version of Collector which has this fix:
-				// https://github.com/open-telemetry/opentelemetry-collector-contrib/pull/26093
-				processed := model.LabelSet{}
-				for label := range entry.Labels {
-					// Loki doesn't support dots in label names
-					// labelName is normalized label name to follow Prometheus label names standard
-					labelName := prometheus_translator.NormalizeLabel(string(label))
-					processed[model.LabelName(labelName)] = entry.Labels[label]
-				}
-
 				conv.metrics.entriesProcessed.Inc()
 				entries = append(entries, loki.Entry{
-					Labels: processed,
+					Labels: entry.Labels,
 					Entry:  *entry.Entry,
 				})
 			}

component/otelcol/exporter/prometheus/internal/convert/convert.go

@@ -19,6 +19,7 @@ import (
 
 	"github.com/go-kit/log"
 	"github.com/go-kit/log/level"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/prometheus"
 	"github.com/prometheus/common/model"
 	"github.com/prometheus/prometheus/model/exemplar"
 	"github.com/prometheus/prometheus/model/labels"
@@ -31,8 +32,6 @@ import (
 	"go.opentelemetry.io/collector/pdata/pcommon"
 	"go.opentelemetry.io/collector/pdata/pmetric"
 	semconv "go.opentelemetry.io/collector/semconv/v1.6.1"
-
-	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/translator/prometheus"
 )
 
 var (
@@ -287,7 +286,8 @@ func (conv *Converter) consumeMetric(app storage.Appender, memResource *memorySe
 }
 
 func (conv *Converter) consumeGauge(app storage.Appender, memResource *memorySeries, memScope *memorySeries, m pmetric.Metric) {
-	metricName := prometheus.BuildPromCompliantName(m, "")
+	// TODO: should we make the param addMetricSuffixes configurable? For now we set the default value (true) to keep the same behavior as before
+	metricName := prometheus.BuildCompliantName(m, "", true)
 
 	metricMD := conv.createOrUpdateMetadata(metricName, metadata.Metadata{
 		Type: textparse.MetricTypeGauge,
@@ -389,7 +389,7 @@ func getNumberDataPointValue(dp pmetric.NumberDataPoint) float64 {
 }
 
 func (conv *Converter) consumeSum(app storage.Appender, memResource *memorySeries, memScope *memorySeries, m pmetric.Metric) {
-	metricName := prometheus.BuildPromCompliantName(m, "")
+	metricName := prometheus.BuildCompliantName(m, "", true)
 
 	// Excerpt from the spec:
 	//
@@ -447,7 +447,7 @@ func (conv *Converter) consumeSum(app storage.Appender, memResource *memorySerie
 }
 
 func (conv *Converter) consumeHistogram(app storage.Appender, memResource *memorySeries, memScope *memorySeries, m pmetric.Metric) {
-	metricName := prometheus.BuildPromCompliantName(m, "")
+	metricName := prometheus.BuildCompliantName(m, "", true)
 
 	if m.Histogram().AggregationTemporality() != pmetric.AggregationTemporalityCumulative {
 		// Drop non-cumulative histograms for now, which is permitted by the spec.
@@ -606,7 +606,7 @@ func (conv *Converter) convertExemplar(otelExemplar pmetric.Exemplar, ts time.Ti
 }
 
 func (conv *Converter) consumeSummary(app storage.Appender, memResource *memorySeries, memScope *memorySeries, m pmetric.Metric) {
-	metricName := prometheus.BuildPromCompliantName(m, "")
+	metricName := prometheus.BuildCompliantName(m, "", true)
 
 	metricMD := conv.createOrUpdateMetadata(metricName, metadata.Metadata{
 		Type: textparse.MetricTypeSummary,

component/otelcol/extension/jaeger_remote_sampling/internal/jaegerremotesampling/extension.go

@@ -18,7 +18,6 @@ import (
 	"context"
 	"fmt"
 
-	grpcStore "github.com/jaegertracing/jaeger/cmd/agent/app/configmanager/grpc"
 	"github.com/jaegertracing/jaeger/cmd/collector/app/sampling/strategystore"
 	"github.com/jaegertracing/jaeger/plugin/sampling/strategystore/static"
 	"go.opentelemetry.io/collector/component"
@@ -75,13 +74,16 @@ func (jrse *jrsExtension) Start(ctx context.Context, host component.Host) error
 	if jrse.cfg.Source.Remote != nil {
 		conn, err := jrse.cfg.Source.Remote.ToClientConn(ctx, host, jrse.telemetry)
 		if err != nil {
-			return fmt.Errorf("error while connecting to the remote sampling source: %w", err)
+			return fmt.Errorf("failed to create the remote strategy store: %w", err)
 		}
-
-		jrse.samplingStore = grpcStore.NewConfigManager(conn)
-		jrse.closers = append(jrse.closers, func() error {
-			return conn.Close()
-		})
+		jrse.closers = append(jrse.closers, conn.Close)
+		remoteStore, closer := internal.NewRemoteStrategyStore(
+			conn,
+			jrse.cfg.Source.Remote,
+			jrse.cfg.Source.ReloadInterval,
+		)
+		jrse.closers = append(jrse.closers, closer.Close)
+		jrse.samplingStore = remoteStore
 	}
 
 	if jrse.cfg.Source.Contents != "" {