[service/internal/graph] Measure telemetry as it is passed between pipeline components (#12812)

Depends on
#12856

Resolves #12676

This is a reboot of #11311, incorporating metrics defined in the
[component telemetry
RFC](https://github.com/open-telemetry/opentelemetry-collector/blob/main/docs/rfcs/component-universal-telemetry.md)
and attributes added in #12617.

The basic pattern is:
- When building any pipeline component which produces data, wrap the
"next consumer" with instrumentation to measure the number of items
being passed. This wrapped consumer is then passed into the constructor
of the component.
- When building any pipeline component which consumes data, wrap the
component itself. This wrapped consumer is saved onto the graph node so
that it can be retrieved during graph assembly.

---------

Co-authored-by: Pablo Baeyens <[email protected]>

Author: djaglowski

.chloggen/pipeline-component-metrics.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 item count metrics defined in Pipeline Component Telemetry RFC
+
+# One or more tracking issues or pull requests related to the change
+issues: [12812]
+
+# (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.items`
+    - `otelcol.processor.consumed.items`
+    - `otelcol.processor.produced.items`
+    - `otelcol.connector.consumed.items`
+    - `otelcol.connector.produced.items`
+    - `otelcol.exporter.consumed.items`
+
+# 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: []

cmd/mdatagen/internal/metric.go

@@ -59,6 +59,9 @@ type Metric struct {
 
 	// Attributes is the list of attributes that the metric emits.
 	Attributes []AttributeName `mapstructure:"attributes"`
+
+	// Override the default prefix for the metric name.
+	Prefix string `mapstructure:"prefix"`
 }
 
 type Stability struct {

cmd/mdatagen/internal/templates/documentation.md.tmpl

@@ -38,7 +38,7 @@
 {{- $metricName := . }}
 {{- $metric := $metricName | telemetryInfo -}}
 
-### otelcol_{{ $metricName }}
+### {{ if $metric.Prefix -}}{{ $metric.Prefix }}{{- else -}}otelcol_{{- end -}}{{ $metricName }}
 
 {{ $metric.Description }}{{ $metric.Stability }}
 

cmd/mdatagen/internal/templates/telemetry.go.tmpl

@@ -136,7 +136,11 @@ func NewTelemetryBuilder(settings component.TelemetrySettings, options ...Teleme
 
     {{- range $name, $metric := .Telemetry.Metrics }}
     builder.{{ $name.Render }}, err = builder.meter.{{ $metric.Data.Instrument }}(
+        {{ if $metric.Prefix -}}
+        "{{ $metric.Prefix }}{{ $name }}",
+        {{ else -}}
         "otelcol_{{ $name }}",
+        {{ end -}}
         metric.WithDescription("{{ $metric.Description }}{{ $metric.Stability }}"),
         metric.WithUnit("{{ $metric.Unit }}"),
         {{ if eq $metric.Data.Type "Histogram" -}}

cmd/mdatagen/internal/templates/telemetrytest.go.tmpl

@@ -32,7 +32,11 @@ return set
 
 func AssertEqual{{ $name.Render }}(t *testing.T, tt *componenttest.Telemetry, dps []metricdata.{{- if eq $metric.Data.Type "Histogram" }} {{$metric.Data.Type}} {{- end }}DataPoint[{{ $metric.Data.BasicType }}], opts ...metricdatatest.Option) {
 	want := metricdata.Metrics{
+		{{ if $metric.Prefix -}}
+		Name:        "{{ $metric.Prefix }}{{ $name }}",
+		{{ else -}}
 		Name:        "otelcol_{{ $name }}",
+		{{ end -}}
 		Description: "{{ $metric.Description }}{{ $metric.Stability }}",
 		Unit:        "{{ $metric.Unit }}",
 		Data:        metricdata.{{ $metric.Data.Type }}[{{ $metric.Data.BasicType }}]{
@@ -45,7 +49,11 @@ func AssertEqual{{ $name.Render }}(t *testing.T, tt *componenttest.Telemetry, dp
 			DataPoints: dps,
 		},
 	}
+	{{ if $metric.Prefix -}}
+	got, err := tt.GetMetric("{{ $metric.Prefix }}{{ $name }}")
+	{{ else -}}
 	got, err := tt.GetMetric("otelcol_{{ $name }}")
+	{{ end -}}
 	require.NoError(t, err)
 	metricdatatest.AssertEqual(t, want, got, opts...)
 }

service/documentation.md

@@ -6,6 +6,30 @@
 
 The following telemetry is emitted by this component.
 
+### otelcol.connector.consumed.items
+
+Number of items passed to the connector.
+
+| Unit | Metric Type | Value Type | Monotonic |
+| ---- | ----------- | ---------- | --------- |
+| {item} | Sum | Int | true |
+
+### otelcol.connector.produced.items
+
+Number of items emitted from the connector.
+
+| Unit | Metric Type | Value Type | Monotonic |
+| ---- | ----------- | ---------- | --------- |
+| {item} | Sum | Int | true |
+
+### otelcol.exporter.consumed.items
+
+Number of items passed to the exporter.
+
+| Unit | Metric Type | Value Type | Monotonic |
+| ---- | ----------- | ---------- | --------- |
+| {item} | Sum | Int | true |
+
 ### otelcol_process_cpu_seconds
 
 Total CPU user and system time in seconds [alpha]
@@ -53,3 +77,27 @@ Uptime of the process [alpha]
 | Unit | Metric Type | Value Type | Monotonic |
 | ---- | ----------- | ---------- | --------- |
 | s | Sum | Double | true |
+
+### otelcol.processor.consumed.items
+
+Number of items passed to the processor.
+
+| Unit | Metric Type | Value Type | Monotonic |
+| ---- | ----------- | ---------- | --------- |
+| {item} | Sum | Int | true |
+
+### otelcol.processor.produced.items
+
+Number of items emitted from the processor.
+
+| Unit | Metric Type | Value Type | Monotonic |
+| ---- | ----------- | ---------- | --------- |
+| {item} | Sum | Int | true |
+
+### otelcol.receiver.produced.items
+
+Number of items emitted from the receiver.
+
+| Unit | Metric Type | Value Type | Monotonic |
+| ---- | ----------- | ---------- | --------- |
+| {item} | Sum | Int | true |