Add size throughput metrics

Author: djaglowski

service/internal/graph/connector.go

@@ -126,7 +126,13 @@ func (n *connectorNode) buildTraces(
 		if err != nil {
 			return err
 		}
-		n.consumer = obsconsumer.NewLogs(n.Component.(consumer.Logs), tb.ConnectorConsumedItems)
+		consumedOpts := []obsconsumer.Option{
+			obsconsumer.WithLogsSizeCounter(&tb.ConnectorConsumedItems),
+		}
+		if isEnabled(tb.ConnectorConsumedSize) {
+			consumedOpts = append(consumedOpts, obsconsumer.WithLogsSizeCounter(&tb.ConnectorConsumedSize))
+		}
+		n.consumer = obsconsumer.NewLogs(n.Component.(consumer.Logs), consumedOpts...)
 	case xpipeline.SignalProfiles:
 		n.Component, err = builder.CreateProfilesToTraces(ctx, set, next)
 		if err != nil {
@@ -188,7 +194,13 @@ func (n *connectorNode) buildMetrics(
 		if err != nil {
 			return err
 		}
-		n.consumer = obsconsumer.NewLogs(n.Component.(consumer.Logs), tb.ConnectorConsumedItems)
+		consumedOpts := []obsconsumer.Option{
+			obsconsumer.WithLogsSizeCounter(&tb.ConnectorConsumedItems),
+		}
+		if isEnabled(tb.ConnectorConsumedSize) {
+			consumedOpts = append(consumedOpts, obsconsumer.WithLogsSizeCounter(&tb.ConnectorConsumedSize))
+		}
+		n.consumer = obsconsumer.NewLogs(n.Component.(consumer.Logs), consumedOpts...)
 	case xpipeline.SignalProfiles:
 		n.Component, err = builder.CreateProfilesToMetrics(ctx, set, next)
 		if err != nil {
@@ -212,16 +224,20 @@ func (n *connectorNode) buildLogs(
 
 	consumers := make(map[pipeline.ID]consumer.Logs, len(nexts))
 	for _, next := range nexts {
-		consumers[next.(*capabilitiesNode).pipelineID] = obsconsumer.NewLogs(
-			next.(consumer.Logs),
-			tb.ConnectorProducedItems,
+		producedOpts := []obsconsumer.Option{
 			obsconsumer.WithStaticDataPointAttribute(
 				otelattr.String(
 					pipelineIDAttrKey,
 					next.(*capabilitiesNode).pipelineID.String(),
 				),
 			),
-		)
+			obsconsumer.WithLogsSizeCounter(&tb.ConnectorProducedSize),
+		}
+		if isEnabled(tb.ConnectorProducedSize) {
+			producedOpts = append(producedOpts, obsconsumer.WithLogsSizeCounter(&tb.ConnectorProducedSize))
+		}
+		consumers[next.(*capabilitiesNode).pipelineID] = obsconsumer.NewLogs(
+			next.(consumer.Logs), producedOpts...)
 	}
 	next := connector.NewLogsRouter(consumers)
 
@@ -231,13 +247,19 @@ func (n *connectorNode) buildLogs(
 		if err != nil {
 			return err
 		}
+		consumedOpts := []obsconsumer.Option{
+			obsconsumer.WithLogsSizeCounter(&tb.ConnectorConsumedItems),
+		}
+		if isEnabled(tb.ConnectorConsumedSize) {
+			consumedOpts = append(consumedOpts, obsconsumer.WithLogsSizeCounter(&tb.ConnectorConsumedSize))
+		}
 		// 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,
+			consumedOpts...,
 		)
 	case pipeline.SignalTraces:
 		n.Component, err = builder.CreateTracesToLogs(ctx, set, next)
@@ -318,7 +340,13 @@ func (n *connectorNode) buildProfiles(
 		if err != nil {
 			return err
 		}
-		n.consumer = obsconsumer.NewLogs(n.Component.(consumer.Logs), tb.ConnectorConsumedItems)
+		consumedOpts := []obsconsumer.Option{
+			obsconsumer.WithLogsSizeCounter(&tb.ConnectorConsumedItems),
+		}
+		if isEnabled(tb.ConnectorConsumedSize) {
+			consumedOpts = append(consumedOpts, obsconsumer.WithLogsSizeCounter(&tb.ConnectorConsumedSize))
+		}
+		n.consumer = obsconsumer.NewLogs(n.Component.(consumer.Logs), consumedOpts...)
 	}
 	return nil
 }

service/internal/graph/enabled.go

@@ -0,0 +1,19 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package graph // import "go.opentelemetry.io/collector/service/internal/graph"
+
+import (
+	"context"
+
+	"go.opentelemetry.io/otel/metric"
+)
+
+type enabledInstrument interface {
+	Enabled(context.Context) bool
+}
+
+func isEnabled(inst metric.Int64Counter) bool {
+	_, ok := inst.(enabledInstrument)
+	return ok
+}

service/internal/graph/exporter.go

@@ -79,7 +79,13 @@ 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.NewLogs(n.Component.(consumer.Logs), tb.ExporterConsumedItems)
+		consumedOpts := []obsconsumer.Option{
+			obsconsumer.WithLogsSizeCounter(&tb.ExporterConsumedItems),
+		}
+		if isEnabled(tb.ExporterConsumedSize) {
+			consumedOpts = append(consumedOpts, obsconsumer.WithLogsSizeCounter(&tb.ExporterConsumedSize))
+		}
+		n.consumer = obsconsumer.NewLogs(n.Component.(consumer.Logs), consumedOpts...)
 	case xpipeline.SignalProfiles:
 		n.Component, err = builder.CreateProfiles(ctx, set)
 		if err != nil {

service/internal/graph/processor.go

@@ -77,12 +77,24 @@ func (n *processorNode) buildComponent(ctx context.Context,
 		}
 		n.consumer = obsconsumer.NewMetrics(n.Component.(consumer.Metrics), tb.ProcessorConsumedItems)
 	case pipeline.SignalLogs:
+		producedOpts := []obsconsumer.Option{
+			obsconsumer.WithLogsSizeCounter(&tb.ProcessorProducedItems),
+		}
+		if isEnabled(tb.ProcessorProducedSize) {
+			producedOpts = append(producedOpts, obsconsumer.WithLogsSizeCounter(&tb.ProcessorProducedSize))
+		}
 		n.Component, err = builder.CreateLogs(ctx, set,
-			obsconsumer.NewLogs(next.(consumer.Logs), tb.ProcessorProducedItems))
+			obsconsumer.NewLogs(next.(consumer.Logs), producedOpts...))
 		if err != nil {
 			return fmt.Errorf("failed to create %q processor, in pipeline %q: %w", set.ID, n.pipelineID.String(), err)
 		}
-		n.consumer = obsconsumer.NewLogs(n.Component.(consumer.Logs), tb.ProcessorConsumedItems)
+		consumedOpts := []obsconsumer.Option{
+			obsconsumer.WithLogsSizeCounter(&tb.ProcessorConsumedItems),
+		}
+		if isEnabled(tb.ProcessorConsumedSize) {
+			consumedOpts = append(consumedOpts, obsconsumer.WithLogsSizeCounter(&tb.ProcessorConsumedSize))
+		}
+		n.consumer = obsconsumer.NewLogs(n.Component.(consumer.Logs), consumedOpts...)
 	case xpipeline.SignalProfiles:
 		n.Component, err = builder.CreateProfiles(ctx, set,
 			obsconsumer.NewProfiles(next.(xconsumer.Profiles), tb.ProcessorProducedItems))

service/internal/graph/receiver.go

@@ -75,8 +75,14 @@ func (n *receiverNode) buildComponent(ctx context.Context,
 		for _, next := range nexts {
 			consumers = append(consumers, next.(consumer.Logs))
 		}
+		consumedOpts := []obsconsumer.Option{
+			obsconsumer.WithLogsSizeCounter(&tb.ReceiverProducedItems),
+		}
+		if isEnabled(tb.ReceiverProducedSize) {
+			consumedOpts = append(consumedOpts, obsconsumer.WithLogsSizeCounter(&tb.ReceiverProducedSize))
+		}
 		n.Component, err = builder.CreateLogs(ctx, set,
-			obsconsumer.NewLogs(fanoutconsumer.NewLogs(consumers), tb.ReceiverProducedItems))
+			obsconsumer.NewLogs(fanoutconsumer.NewLogs(consumers), consumedOpts...))
 	case xpipeline.SignalProfiles:
 		var consumers []xconsumer.Profiles
 		for _, next := range nexts {