[service/internal/graph] Add size throughput metrics (#13032)

Follows #12812

This PR adds the `size` metrics defined in the [Pipeline Component
Telemetry
RFC](https://github.com/open-telemetry/opentelemetry-collector/blob/main/docs/rfcs/component-universal-telemetry.md).

Author: djaglowski

.chloggen/pipeline-throughput.yaml

@@ -0,0 +1,32 @@
+# 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. otlpreceiver)
+component: service
+
+# A brief description of the change.  Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add size metrics defined in Pipeline Component Telemetry RFC
+
+# One or more tracking issues or pull requests related to the change
+issues: [13032]
+
+# (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: |
+  See [Pipeline Component Telemetry RFC](https://github.com/open-telemetry/opentelemetry-collector/blob/main/docs/rfcs/component-universal-telemetry.md) for more details:
+    - `otelcol.receiver.produced.size`
+    - `otelcol.processor.consumed.size`
+    - `otelcol.processor.produced.size`
+    - `otelcol.connector.consumed.size`
+    - `otelcol.connector.produced.size`
+    - `otelcol.exporter.consumed.size`
+
+# 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: []

service/internal/graph/connector.go

@@ -90,7 +90,7 @@ func (n *connectorNode) buildTraces(
 	for _, next := range nexts {
 		consumers[next.(*capabilitiesNode).pipelineID] = obsconsumer.NewTraces(
 			next.(consumer.Traces),
-			tb.ConnectorProducedItems,
+			tb.ConnectorProducedItems, tb.ConnectorProducedSize,
 			obsconsumer.WithStaticDataPointAttribute(
 				otelattr.String(
 					pipelineIDAttrKey,
@@ -107,32 +107,33 @@ func (n *connectorNode) buildTraces(
 		if err != nil {
 			return err
 		}
+
 		// Connectors which might pass along data must inherit capabilities of all nexts
 		n.consumer = obsconsumer.NewTraces(
 			capabilityconsumer.NewTraces(
 				n.Component.(consumer.Traces),
 				aggregateCap(n.Component.(consumer.Traces), nexts),
 			),
-			tb.ConnectorConsumedItems,
+			tb.ConnectorConsumedItems, tb.ConnectorConsumedSize,
 		)
 	case pipeline.SignalMetrics:
 		n.Component, err = builder.CreateMetricsToTraces(ctx, set, next)
 		if err != nil {
 			return err
 		}
-		n.consumer = obsconsumer.NewMetrics(n.Component.(consumer.Metrics), tb.ConnectorConsumedItems)
+		n.consumer = obsconsumer.NewMetrics(n.Component.(consumer.Metrics), tb.ConnectorConsumedItems, tb.ConnectorConsumedSize)
 	case pipeline.SignalLogs:
 		n.Component, err = builder.CreateLogsToTraces(ctx, set, next)
 		if err != nil {
 			return err
 		}
-		n.consumer = obsconsumer.NewLogs(n.Component.(consumer.Logs), tb.ConnectorConsumedItems)
+		n.consumer = obsconsumer.NewLogs(n.Component.(consumer.Logs), tb.ConnectorConsumedItems, tb.ConnectorConsumedSize)
 	case xpipeline.SignalProfiles:
 		n.Component, err = builder.CreateProfilesToTraces(ctx, set, next)
 		if err != nil {
 			return err
 		}
-		n.consumer = obsconsumer.NewProfiles(n.Component.(xconsumer.Profiles), tb.ConnectorConsumedItems)
+		n.consumer = obsconsumer.NewProfiles(n.Component.(xconsumer.Profiles), tb.ConnectorConsumedItems, tb.ConnectorConsumedSize)
 	}
 	return nil
 }
@@ -152,7 +153,7 @@ func (n *connectorNode) buildMetrics(
 	for _, next := range nexts {
 		consumers[next.(*capabilitiesNode).pipelineID] = obsconsumer.NewMetrics(
 			next.(consumer.Metrics),
-			tb.ConnectorProducedItems,
+			tb.ConnectorProducedItems, tb.ConnectorProducedSize,
 			obsconsumer.WithStaticDataPointAttribute(
 				otelattr.String(
 					pipelineIDAttrKey,
@@ -169,32 +170,33 @@ func (n *connectorNode) buildMetrics(
 		if err != nil {
 			return err
 		}
+
 		// Connectors which might pass along data must inherit capabilities of all nexts
 		n.consumer = obsconsumer.NewMetrics(
 			capabilityconsumer.NewMetrics(
 				n.Component.(consumer.Metrics),
 				aggregateCap(n.Component.(consumer.Metrics), nexts),
 			),
-			tb.ConnectorConsumedItems,
+			tb.ConnectorConsumedItems, tb.ConnectorConsumedSize,
 		)
 	case pipeline.SignalTraces:
 		n.Component, err = builder.CreateTracesToMetrics(ctx, set, next)
 		if err != nil {
 			return err
 		}
-		n.consumer = obsconsumer.NewTraces(n.Component.(consumer.Traces), tb.ConnectorConsumedItems)
+		n.consumer = obsconsumer.NewTraces(n.Component.(consumer.Traces), tb.ConnectorConsumedItems, tb.ConnectorConsumedSize)
 	case pipeline.SignalLogs:
 		n.Component, err = builder.CreateLogsToMetrics(ctx, set, next)
 		if err != nil {
 			return err
 		}
-		n.consumer = obsconsumer.NewLogs(n.Component.(consumer.Logs), tb.ConnectorConsumedItems)
+		n.consumer = obsconsumer.NewLogs(n.Component.(consumer.Logs), tb.ConnectorConsumedItems, tb.ConnectorConsumedSize)
 	case xpipeline.SignalProfiles:
 		n.Component, err = builder.CreateProfilesToMetrics(ctx, set, next)
 		if err != nil {
 			return err
 		}
-		n.consumer = obsconsumer.NewProfiles(n.Component.(xconsumer.Profiles), tb.ConnectorConsumedItems)
+		n.consumer = obsconsumer.NewProfiles(n.Component.(xconsumer.Profiles), tb.ConnectorConsumedItems, tb.ConnectorConsumedSize)
 	}
 	return nil
 }
@@ -214,7 +216,7 @@ func (n *connectorNode) buildLogs(
 	for _, next := range nexts {
 		consumers[next.(*capabilitiesNode).pipelineID] = obsconsumer.NewLogs(
 			next.(consumer.Logs),
-			tb.ConnectorProducedItems,
+			tb.ConnectorProducedItems, tb.ConnectorProducedSize,
 			obsconsumer.WithStaticDataPointAttribute(
 				otelattr.String(
 					pipelineIDAttrKey,
@@ -231,32 +233,33 @@ func (n *connectorNode) buildLogs(
 		if err != nil {
 			return err
 		}
+
 		// Connectors which might pass along data must inherit capabilities of all nexts
 		n.consumer = obsconsumer.NewLogs(
 			capabilityconsumer.NewLogs(
 				n.Component.(consumer.Logs),
 				aggregateCap(n.Component.(consumer.Logs), nexts),
 			),
-			tb.ConnectorConsumedItems,
+			tb.ConnectorConsumedItems, tb.ConnectorConsumedSize,
 		)
 	case pipeline.SignalTraces:
 		n.Component, err = builder.CreateTracesToLogs(ctx, set, next)
 		if err != nil {
 			return err
 		}
-		n.consumer = obsconsumer.NewTraces(n.Component.(consumer.Traces), tb.ConnectorConsumedItems)
+		n.consumer = obsconsumer.NewTraces(n.Component.(consumer.Traces), tb.ConnectorConsumedItems, tb.ConnectorConsumedSize)
 	case pipeline.SignalMetrics:
 		n.Component, err = builder.CreateMetricsToLogs(ctx, set, next)
 		if err != nil {
 			return err
 		}
-		n.consumer = obsconsumer.NewMetrics(n.Component.(consumer.Metrics), tb.ConnectorConsumedItems)
+		n.consumer = obsconsumer.NewMetrics(n.Component.(consumer.Metrics), tb.ConnectorConsumedItems, tb.ConnectorConsumedSize)
 	case xpipeline.SignalProfiles:
 		n.Component, err = builder.CreateProfilesToLogs(ctx, set, next)
 		if err != nil {
 			return err
 		}
-		n.consumer = obsconsumer.NewProfiles(n.Component.(xconsumer.Profiles), tb.ConnectorConsumedItems)
+		n.consumer = obsconsumer.NewProfiles(n.Component.(xconsumer.Profiles), tb.ConnectorConsumedItems, tb.ConnectorConsumedSize)
 	}
 	return nil
 }
@@ -276,7 +279,7 @@ func (n *connectorNode) buildProfiles(
 	for _, next := range nexts {
 		consumers[next.(*capabilitiesNode).pipelineID] = obsconsumer.NewProfiles(
 			next.(xconsumer.Profiles),
-			tb.ConnectorProducedItems,
+			tb.ConnectorProducedItems, tb.ConnectorProducedSize,
 			obsconsumer.WithStaticDataPointAttribute(
 				otelattr.String(
 					pipelineIDAttrKey,
@@ -293,32 +296,33 @@ func (n *connectorNode) buildProfiles(
 		if err != nil {
 			return err
 		}
+
 		// Connectors which might pass along data must inherit capabilities of all nexts
 		n.consumer = obsconsumer.NewProfiles(
 			capabilityconsumer.NewProfiles(
 				n.Component.(xconsumer.Profiles),
 				aggregateCap(n.Component.(xconsumer.Profiles), nexts),
 			),
-			tb.ConnectorConsumedItems,
+			tb.ConnectorConsumedItems, tb.ConnectorConsumedSize,
 		)
 	case pipeline.SignalTraces:
 		n.Component, err = builder.CreateTracesToProfiles(ctx, set, next)
 		if err != nil {
 			return err
 		}
-		n.consumer = obsconsumer.NewTraces(n.Component.(consumer.Traces), tb.ConnectorConsumedItems)
+		n.consumer = obsconsumer.NewTraces(n.Component.(consumer.Traces), tb.ConnectorConsumedItems, tb.ConnectorConsumedSize)
 	case pipeline.SignalMetrics:
 		n.Component, err = builder.CreateMetricsToProfiles(ctx, set, next)
 		if err != nil {
 			return err
 		}
-		n.consumer = obsconsumer.NewMetrics(n.Component.(consumer.Metrics), tb.ConnectorConsumedItems)
+		n.consumer = obsconsumer.NewMetrics(n.Component.(consumer.Metrics), tb.ConnectorConsumedItems, tb.ConnectorConsumedSize)
 	case pipeline.SignalLogs:
 		n.Component, err = builder.CreateLogsToProfiles(ctx, set, next)
 		if err != nil {
 			return err
 		}
-		n.consumer = obsconsumer.NewLogs(n.Component.(consumer.Logs), tb.ConnectorConsumedItems)
+		n.consumer = obsconsumer.NewLogs(n.Component.(consumer.Logs), tb.ConnectorConsumedItems, tb.ConnectorConsumedSize)
 	}
 	return nil
 }

service/internal/graph/exporter.go

@@ -67,25 +67,25 @@ func (n *exporterNode) buildComponent(
 		if err != nil {
 			return fmt.Errorf("failed to create %q exporter for data type %q: %w", set.ID, n.pipelineType, err)
 		}
-		n.consumer = obsconsumer.NewTraces(n.Component.(consumer.Traces), tb.ExporterConsumedItems)
+		n.consumer = obsconsumer.NewTraces(n.Component.(consumer.Traces), tb.ExporterConsumedItems, tb.ExporterConsumedSize)
 	case pipeline.SignalMetrics:
 		n.Component, err = builder.CreateMetrics(ctx, set)
 		if err != nil {
 			return fmt.Errorf("failed to create %q exporter for data type %q: %w", set.ID, n.pipelineType, err)
 		}
-		n.consumer = obsconsumer.NewMetrics(n.Component.(consumer.Metrics), tb.ExporterConsumedItems)
+		n.consumer = obsconsumer.NewMetrics(n.Component.(consumer.Metrics), tb.ExporterConsumedItems, tb.ExporterConsumedSize)
 	case pipeline.SignalLogs:
 		n.Component, err = builder.CreateLogs(ctx, set)
 		if err != nil {
 			return fmt.Errorf("failed to create %q exporter for data type %q: %w", set.ID, n.pipelineType, err)
 		}
-		n.consumer = obsconsumer.NewLogs(n.Component.(consumer.Logs), tb.ExporterConsumedItems)
+		n.consumer = obsconsumer.NewLogs(n.Component.(consumer.Logs), tb.ExporterConsumedItems, tb.ExporterConsumedSize)
 	case xpipeline.SignalProfiles:
 		n.Component, err = builder.CreateProfiles(ctx, set)
 		if err != nil {
 			return fmt.Errorf("failed to create %q exporter for data type %q: %w", set.ID, n.pipelineType, err)
 		}
-		n.consumer = obsconsumer.NewProfiles(n.Component.(xconsumer.Profiles), tb.ExporterConsumedItems)
+		n.consumer = obsconsumer.NewProfiles(n.Component.(xconsumer.Profiles), tb.ExporterConsumedItems, tb.ExporterConsumedSize)
 	default:
 		return fmt.Errorf("error creating exporter %q for data type %q is not supported", set.ID, n.pipelineType)
 	}