32669 wireframe sum connector

.chloggen/32669-wireframe-sum.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: new_component
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: sumconnector
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: "creates a wireframe and initial pr to develop from"
+
+# Mandatory: One or more tracking issues related to the change. You can use the PR number here if no issue exists.
+issues: [32669]
+
+# (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: []

.github/CODEOWNERS

@@ -33,6 +33,7 @@ connector/roundrobinconnector/                                      @open-teleme
 connector/routingconnector/                                         @open-telemetry/collector-contrib-approvers @jpkrohling @mwear
 connector/servicegraphconnector/                                    @open-telemetry/collector-contrib-approvers @jpkrohling @mapno
 connector/spanmetricsconnector/                                     @open-telemetry/collector-contrib-approvers @portertech @Frapschen
+connector/sumconnector/                                             @open-telemetry/collector-contrib-approvers @greatestusername @shalper2
 
 examples/demo/                                                      @open-telemetry/collector-contrib-approvers @open-telemetry/collector-approvers
 

.github/ISSUE_TEMPLATE/bug_report.yaml

@@ -34,6 +34,7 @@ body:
       - connector/routing
       - connector/servicegraph
       - connector/spanmetrics
+      - connector/sum
       - examples/demo
       - exporter/alertmanager
       - exporter/alibabacloudlogservice

.github/ISSUE_TEMPLATE/feature_request.yaml

@@ -28,6 +28,7 @@ body:
       - connector/routing
       - connector/servicegraph
       - connector/spanmetrics
+      - connector/sum
       - examples/demo
       - exporter/alertmanager
       - exporter/alibabacloudlogservice

.github/ISSUE_TEMPLATE/other.yaml

@@ -28,6 +28,7 @@ body:
       - connector/routing
       - connector/servicegraph
       - connector/spanmetrics
+      - connector/sum
       - examples/demo
       - exporter/alertmanager
       - exporter/alibabacloudlogservice

.github/ISSUE_TEMPLATE/unmaintained.yaml

@@ -33,6 +33,7 @@ body:
       - connector/routing
       - connector/servicegraph
       - connector/spanmetrics
+      - connector/sum
       - examples/demo
       - exporter/alertmanager
       - exporter/alibabacloudlogservice

connector/sumconnector/Makefile

@@ -0,0 +1 @@
+include ../../Makefile.Common

connector/sumconnector/README.md

@@ -0,0 +1,59 @@
+# Sum Connector
+<!-- status autogenerated section -->
+| Status        |           |
+| ------------- |-----------|
+| Distributions | [contrib] |
+| Issues        | [![Open issues](https://img.shields.io/github/issues-search/open-telemetry/opentelemetry-collector-contrib?query=is%3Aissue%20is%3Aopen%20label%3Aconnector%2Fsum%20&label=open&color=orange&logo=opentelemetry)](https://github.com/open-telemetry/opentelemetry-collector-contrib/issues?q=is%3Aopen+is%3Aissue+label%3Aconnector%2Fsum) [![Closed issues](https://img.shields.io/github/issues-search/open-telemetry/opentelemetry-collector-contrib?query=is%3Aissue%20is%3Aclosed%20label%3Aconnector%2Fsum%20&label=closed&color=blue&logo=opentelemetry)](https://github.com/open-telemetry/opentelemetry-collector-contrib/issues?q=is%3Aclosed+is%3Aissue+label%3Aconnector%2Fsum) |
+| [Code Owners](https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/main/CONTRIBUTING.md#becoming-a-code-owner)    | [@greatestusername](https://www.github.com/greatestusername), [@shalper2](https://www.github.com/shalper2) |
+
+[development]: https://github.com/open-telemetry/opentelemetry-collector#development
+[contrib]: https://github.com/open-telemetry/opentelemetry-collector-releases/tree/main/distributions/otelcol-contrib
+
+## Supported Pipeline Types
+
+| [Exporter Pipeline Type] | [Receiver Pipeline Type] | [Stability Level] |
+| ------------------------ | ------------------------ | ----------------- |
+| traces | metrics | [development] |
+| metrics | metrics | [development] |
+| logs | metrics | [development] |
+
+[Exporter Pipeline Type]: https://github.com/open-telemetry/opentelemetry-collector/blob/main/connector/README.md#exporter-pipeline-type
+[Receiver Pipeline Type]: https://github.com/open-telemetry/opentelemetry-collector/blob/main/connector/README.md#receiver-pipeline-type
+[Stability Level]: https://github.com/open-telemetry/opentelemetry-collector#stability-levels
+<!-- end autogenerated section -->
+
+The `sum` connector can be used to sum attribute values from spans, span events, metrics, data points, and log records.
+## Configuration
+
+If you are not already familiar with connectors, you may find it helpful to first visit the [Connectors README].
+
+### Example Configuration
+
+```yaml
+receivers:
+  foo:
+connectors:
+  sum:
+    logs:
+      checkout.total:
+        source_attribute: 
+          - attributes["total.payment"]
+        conditions:
+          - attributes["total.payment"] != "NULL"
+        attributes:
+          - key: payment.processor
+            default_value: unspecified_processor
+exporters:
+  bar:
+
+service:
+  pipelines:
+    metrics/sum:
+       receivers: [sum]
+       exporters: [bar]
+    logs:
+       receivers: [foo]
+       exporters: [sum]
+```
+
+[Connectors README]: https://github.com/open-telemetry/opentelemetry-collector/blob/main/connector/README.md

connector/sumconnector/config.go

@@ -0,0 +1,25 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package sumconnector // import "github.com/open-telemetry/opentelemetry-collector-contrib/connector/sumconnector"
+
+// Config for the connector
+type Config struct {
+	Spans      map[string]MetricInfo `mapstructure:"spans"`
+	SpanEvents map[string]MetricInfo `mapstructure:"spanevents"`
+	Metrics    map[string]MetricInfo `mapstructure:"metrics"`
+	DataPoints map[string]MetricInfo `mapstructure:"datapoints"`
+	Logs       map[string]MetricInfo `mapstructure:"logs"`
+}
+
+// MetricInfo for a data type
+type MetricInfo struct {
+	Description string            `mapstructure:"description"`
+	Conditions  []string          `mapstructure:"conditions"`
+	Attributes  []AttributeConfig `mapstructure:"attributes"`
+}
+
+type AttributeConfig struct {
+	Key          string `mapstructure:"key"`
+	DefaultValue any    `mapstructure:"default_value"`
+}

connector/sumconnector/connector.go

@@ -0,0 +1,61 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package sumconnector // import "github.com/open-telemetry/opentelemetry-collector-contrib/connector/sumconnector"
+
+import (
+	"context"
+
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/consumer"
+	"go.opentelemetry.io/collector/pdata/plog"
+	"go.opentelemetry.io/collector/pdata/pmetric"
+	"go.opentelemetry.io/collector/pdata/ptrace"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottldatapoint"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottllog"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottlmetric"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottlspan"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottlspanevent"
+)
+
+// sum can sum attribute values from spans, span event, metrics, data points, or log records
+// and emit the sums onto a metrics pipeline.
+type sum struct {
+	metricsConsumer consumer.Metrics
+	component.StartFunc
+	component.ShutdownFunc
+
+	spansMetricDefs      map[string]metricDef[ottlspan.TransformContext]
+	spanEventsMetricDefs map[string]metricDef[ottlspanevent.TransformContext]
+	metricsMetricDefs    map[string]metricDef[ottlmetric.TransformContext]
+	dataPointsMetricDefs map[string]metricDef[ottldatapoint.TransformContext]
+	logsMetricDefs       map[string]metricDef[ottllog.TransformContext]
+}
+
+func (c *sum) Capabilities() consumer.Capabilities {
+	return consumer.Capabilities{MutatesData: false}
+}
+
+func (c *sum) ConsumeTraces(ctx context.Context, td ptrace.Traces) error {
+
+	sumMetrics := pmetric.NewMetrics()
+	sumMetrics.ResourceMetrics().EnsureCapacity(td.ResourceSpans().Len())
+
+	return c.metricsConsumer.ConsumeMetrics(ctx, sumMetrics)
+}
+
+func (c *sum) ConsumeMetrics(ctx context.Context, md pmetric.Metrics) error {
+
+	sumMetrics := pmetric.NewMetrics()
+	sumMetrics.ResourceMetrics().EnsureCapacity(md.ResourceMetrics().Len())
+
+	return c.metricsConsumer.ConsumeMetrics(ctx, sumMetrics)
+}
+
+func (c *sum) ConsumeLogs(ctx context.Context, ld plog.Logs) error {
+	sumMetrics := pmetric.NewMetrics()
+	sumMetrics.ResourceMetrics().EnsureCapacity(ld.ResourceLogs().Len())
+
+	return c.metricsConsumer.ConsumeMetrics(ctx, sumMetrics)
+}

connector/sumconnector/factory.go

@@ -0,0 +1,162 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+//go:generate mdatagen metadata.yaml
+
+package sumconnector // import "github.com/open-telemetry/opentelemetry-collector-contrib/connector/sumconnector"
+
+import (
+	"context"
+
+	"go.opentelemetry.io/collector/component"
+	"go.opentelemetry.io/collector/connector"
+	"go.opentelemetry.io/collector/consumer"
+
+	"github.com/open-telemetry/opentelemetry-collector-contrib/connector/sumconnector/internal/metadata"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/filter/expr"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/internal/filter/filterottl"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottldatapoint"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottllog"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottlmetric"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottlspan"
+	"github.com/open-telemetry/opentelemetry-collector-contrib/pkg/ottl/contexts/ottlspanevent"
+)
+
+// NewFactory returns a ConnectorFactory.
+func NewFactory() connector.Factory {
+	return connector.NewFactory(
+		metadata.Type,
+		createDefaultConfig,
+		connector.WithTracesToMetrics(createTracesToMetrics, metadata.TracesToMetricsStability),
+		connector.WithMetricsToMetrics(createMetricsToMetrics, metadata.MetricsToMetricsStability),
+		connector.WithLogsToMetrics(createLogsToMetrics, metadata.LogsToMetricsStability),
+	)
+}
+
+// createDefaultConfig creates the default configuration.
+func createDefaultConfig() component.Config {
+	return &Config{}
+}
+
+// createTracesToMetrics creates a traces to metrics connector based on provided config.
+func createTracesToMetrics(
+	_ context.Context,
+	set connector.Settings,
+	cfg component.Config,
+	nextConsumer consumer.Metrics,
+) (connector.Traces, error) {
+	c := cfg.(*Config)
+
+	spanMetricDefs := make(map[string]metricDef[ottlspan.TransformContext], len(c.Spans))
+	for name, info := range c.Spans {
+		md := metricDef[ottlspan.TransformContext]{
+			desc:  info.Description,
+			attrs: info.Attributes,
+		}
+		if len(info.Conditions) > 0 {
+			// Error checked in Config.Validate()
+			condition, _ := filterottl.NewBoolExprForSpan(info.Conditions, filterottl.StandardSpanFuncs(), ottl.PropagateError, set.TelemetrySettings)
+			md.condition = condition
+		}
+		spanMetricDefs[name] = md
+	}
+
+	spanEventMetricDefs := make(map[string]metricDef[ottlspanevent.TransformContext], len(c.SpanEvents))
+	for name, info := range c.SpanEvents {
+		md := metricDef[ottlspanevent.TransformContext]{
+			desc:  info.Description,
+			attrs: info.Attributes,
+		}
+		if len(info.Conditions) > 0 {
+			// Error checked in Config.Validate()
+			condition, _ := filterottl.NewBoolExprForSpanEvent(info.Conditions, filterottl.StandardSpanEventFuncs(), ottl.PropagateError, set.TelemetrySettings)
+			md.condition = condition
+		}
+		spanEventMetricDefs[name] = md
+	}
+
+	return &sum{
+		metricsConsumer:      nextConsumer,
+		spansMetricDefs:      spanMetricDefs,
+		spanEventsMetricDefs: spanEventMetricDefs,
+	}, nil
+}
+
+// createMetricsToMetrics creates a metricds to metrics connector based on provided config.
+func createMetricsToMetrics(
+	_ context.Context,
+	set connector.Settings,
+	cfg component.Config,
+	nextConsumer consumer.Metrics,
+) (connector.Metrics, error) {
+	c := cfg.(*Config)
+
+	metricMetricDefs := make(map[string]metricDef[ottlmetric.TransformContext], len(c.Metrics))
+	for name, info := range c.Metrics {
+		md := metricDef[ottlmetric.TransformContext]{
+			desc: info.Description,
+		}
+		if len(info.Conditions) > 0 {
+			// Error checked in Config.Validate()
+			condition, _ := filterottl.NewBoolExprForMetric(info.Conditions, filterottl.StandardMetricFuncs(), ottl.PropagateError, set.TelemetrySettings)
+			md.condition = condition
+		}
+		metricMetricDefs[name] = md
+	}
+
+	dataPointMetricDefs := make(map[string]metricDef[ottldatapoint.TransformContext], len(c.DataPoints))
+	for name, info := range c.DataPoints {
+		md := metricDef[ottldatapoint.TransformContext]{
+			desc:  info.Description,
+			attrs: info.Attributes,
+		}
+		if len(info.Conditions) > 0 {
+			// Error checked in Config.Validate()
+			condition, _ := filterottl.NewBoolExprForDataPoint(info.Conditions, filterottl.StandardDataPointFuncs(), ottl.PropagateError, set.TelemetrySettings)
+			md.condition = condition
+		}
+		dataPointMetricDefs[name] = md
+	}
+
+	return &sum{
+		metricsConsumer:      nextConsumer,
+		metricsMetricDefs:    metricMetricDefs,
+		dataPointsMetricDefs: dataPointMetricDefs,
+	}, nil
+}
+
+// createLogsToMetrics creates a logs to metrics connector based on provided config.
+func createLogsToMetrics(
+	_ context.Context,
+	set connector.Settings,
+	cfg component.Config,
+	nextConsumer consumer.Metrics,
+) (connector.Logs, error) {
+	c := cfg.(*Config)
+
+	metricDefs := make(map[string]metricDef[ottllog.TransformContext], len(c.Logs))
+	for name, info := range c.Logs {
+		md := metricDef[ottllog.TransformContext]{
+			desc:  info.Description,
+			attrs: info.Attributes,
+		}
+		if len(info.Conditions) > 0 {
+			// Error checked in Config.Validate()
+			condition, _ := filterottl.NewBoolExprForLog(info.Conditions, filterottl.StandardLogFuncs(), ottl.PropagateError, set.TelemetrySettings)
+			md.condition = condition
+		}
+		metricDefs[name] = md
+	}
+
+	return &sum{
+		metricsConsumer: nextConsumer,
+		logsMetricDefs:  metricDefs,
+	}, nil
+}
+
+type metricDef[K any] struct {
+	condition expr.BoolExpr[K]
+	desc      string
+	attrs     []AttributeConfig
+}