[mdatagen] add support for async instruments (#10159)

This PR adds the ability to configure asynchronous (observable)
instruments via mdatagen. This requires providing a mechanism to set
options to pass in the callbacks that will be called at the time of the
observation.

---------

Signed-off-by: Alex Boten <223565+codeboten@users.noreply.github.com>

cmd/mdatagen/internal/samplereceiver/metadata.yaml

@@ -165,3 +165,11 @@ telemetry:
       unit: s
       histogram:
         value_type: double
+    process_runtime_total_alloc_bytes:
+      enabled: true
+      description: Cumulative bytes allocated for heap objects (see 'go doc runtime.MemStats.TotalAlloc')
+      unit: By
+      sum:
+        async: true
+        value_type: int
+        monotonic: true

cmd/mdatagen/loader_test.go

@@ -250,6 +250,18 @@ func TestLoadMetadata(t *testing.T) {
 								MetricValueType: MetricValueType{pmetric.NumberDataPointValueTypeDouble},
 							},
 						},
+						"process_runtime_total_alloc_bytes": {
+							Enabled:     true,
+							Description: "Cumulative bytes allocated for heap objects (see 'go doc runtime.MemStats.TotalAlloc')",
+							Unit:        strPtr("By"),
+							Sum: &sum{
+								Mono: Mono{true},
+								MetricValueType: MetricValueType{
+									ValueType: pmetric.NumberDataPointValueTypeInt,
+								},
+								Async: true,
+							},
+						},
 					},
 				},
 				ScopeName:       "go.opentelemetry.io/collector/internal/receiver/samplereceiver",

cmd/mdatagen/metricdata.go

@@ -27,6 +27,7 @@ type MetricData interface {
 	HasAggregated() bool
 	HasMetricInputType() bool
 	Instrument() string
+	IsAsync() bool
 }
 
 // AggregationTemporality defines a metric aggregation type.
@@ -123,6 +124,7 @@ func (mvt MetricValueType) BasicType() string {
 type gauge struct {
 	MetricValueType `mapstructure:"value_type"`
 	MetricInputType `mapstructure:",squash"`
+	Async           bool `mapstructure:"async,omitempty"`
 }
 
 // Unmarshal is a custom unmarshaler for gauge. Needed mostly to avoid MetricValueType.Unmarshal inheritance.
@@ -146,14 +148,26 @@ func (d gauge) HasAggregated() bool {
 }
 
 func (d gauge) Instrument() string {
-	return ""
+	instrumentName := cases.Title(language.English).String(d.MetricValueType.BasicType())
+
+	if d.Async {
+		instrumentName += "Observable"
+	}
+
+	instrumentName += "Gauge"
+	return instrumentName
+}
+
+func (d gauge) IsAsync() bool {
+	return d.Async
 }
 
 type sum struct {
 	AggregationTemporality `mapstructure:"aggregation_temporality"`
 	Mono                   `mapstructure:",squash"`
 	MetricValueType        `mapstructure:"value_type"`
 	MetricInputType        `mapstructure:",squash"`
+	Async                  bool `mapstructure:"async,omitempty"`
 }
 
 // Unmarshal is a custom unmarshaler for sum. Needed mostly to avoid MetricValueType.Unmarshal inheritance.
@@ -190,18 +204,26 @@ func (d sum) HasAggregated() bool {
 func (d sum) Instrument() string {
 	instrumentName := cases.Title(language.English).String(d.MetricValueType.BasicType())
 
+	if d.Async {
+		instrumentName += "Observable"
+	}
 	if !d.Monotonic {
 		instrumentName += "UpDown"
 	}
 	instrumentName += "Counter"
 	return instrumentName
 }
 
+func (d sum) IsAsync() bool {
+	return d.Async
+}
+
 type histogram struct {
 	AggregationTemporality `mapstructure:"aggregation_temporality"`
 	Mono                   `mapstructure:",squash"`
 	MetricValueType        `mapstructure:"value_type"`
 	MetricInputType        `mapstructure:",squash"`
+	Async                  bool `mapstructure:"async,omitempty"`
 }
 
 func (d histogram) Type() string {
@@ -228,3 +250,7 @@ func (d *histogram) Unmarshal(parser *confmap.Conf) error {
 	}
 	return parser.Unmarshal(d, confmap.WithIgnoreUnused())
 }
+
+func (d histogram) IsAsync() bool {
+	return d.Async
+}

cmd/mdatagen/metricdata_test.go

@@ -7,21 +7,34 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/assert"
+
+	"go.opentelemetry.io/collector/pdata/pmetric"
 )
 
 func TestMetricData(t *testing.T) {
 	for _, arg := range []struct {
-		metricData    MetricData
-		typ           string
-		hasAggregated bool
-		hasMonotonic  bool
+		metricData        MetricData
+		wantType          string
+		wantHasAggregated bool
+		wantHasMonotonic  bool
+		wantInstrument    string
+		wantAsync         bool
 	}{
-		{&gauge{}, "Gauge", false, false},
-		{&sum{}, "Sum", true, true},
-		{&histogram{}, "Histogram", false, false},
+		{&gauge{}, "Gauge", false, false, "Gauge", false},
+		{&gauge{Async: true}, "Gauge", false, false, "ObservableGauge", true},
+		{&gauge{MetricValueType: MetricValueType{pmetric.NumberDataPointValueTypeInt}, Async: true}, "Gauge", false, false, "Int64ObservableGauge", true},
+		{&gauge{MetricValueType: MetricValueType{pmetric.NumberDataPointValueTypeDouble}, Async: true}, "Gauge", false, false, "Float64ObservableGauge", true},
+		{&sum{}, "Sum", true, true, "UpDownCounter", false},
+		{&sum{Mono: Mono{true}}, "Sum", true, true, "Counter", false},
+		{&sum{Async: true}, "Sum", true, true, "ObservableUpDownCounter", true},
+		{&sum{MetricValueType: MetricValueType{pmetric.NumberDataPointValueTypeInt}, Async: true}, "Sum", true, true, "Int64ObservableUpDownCounter", true},
+		{&sum{MetricValueType: MetricValueType{pmetric.NumberDataPointValueTypeDouble}, Async: true}, "Sum", true, true, "Float64ObservableUpDownCounter", true},
+		{&histogram{}, "Histogram", false, false, "Histogram", false},
 	} {
-		assert.Equal(t, arg.typ, arg.metricData.Type())
-		assert.Equal(t, arg.hasAggregated, arg.metricData.HasAggregated())
-		assert.Equal(t, arg.hasMonotonic, arg.metricData.HasMonotonic())
+		assert.Equal(t, arg.wantType, arg.metricData.Type())
+		assert.Equal(t, arg.wantHasAggregated, arg.metricData.HasAggregated())
+		assert.Equal(t, arg.wantHasMonotonic, arg.metricData.HasMonotonic())
+		assert.Equal(t, arg.wantInstrument, arg.metricData.Instrument())
+		assert.Equal(t, arg.wantAsync, arg.metricData.IsAsync())
 	}
 }

cmd/mdatagen/templates/telemetry.go.tmpl

@@ -4,6 +4,7 @@ package {{ .Package }}
 
 import (
     {{- if .Telemetry.Metrics }}
+    "context"
     "errors"
     {{- end }}
 
@@ -26,23 +27,46 @@ func Tracer(settings component.TelemetrySettings) trace.Tracer {
 type TelemetryBuilder struct {
 	{{- range $name, $metric := .Telemetry.Metrics }}
 	{{ $name.Render }} metric.{{ $metric.Data.Instrument }}
+    {{- if $metric.Data.Async }}
+    observe{{ $name.Render }} func() {{ $metric.Data.BasicType }}
+    {{- end }}
 	{{- end }}
 }
 
 // telemetryBuilderOption applies changes to default builder.
 type telemetryBuilderOption func(*TelemetryBuilder)
 
+{{- range $name, $metric := .Telemetry.Metrics }}
+    {{ if $metric.Data.Async -}}
+// With{{ $name.Render }}Callback sets callback for observable {{ $name.Render }} metric.
+func With{{ $name.Render }}Callback(cb func() {{ $metric.Data.BasicType }}) telemetryBuilderOption {
+    return func(builder *TelemetryBuilder) {
+        builder.observe{{ $name.Render }} = cb
+    }
+}
+    {{- end }}
+{{- end }}
+
 // NewTelemetryBuilder provides a struct with methods to update all internal telemetry
 // for a component
 func NewTelemetryBuilder(settings component.TelemetrySettings, options ...telemetryBuilderOption) (*TelemetryBuilder, error) {
     builder := TelemetryBuilder{}
+	for _, op := range options {
+		op(&builder)
+	}
     var err, errs error
     meter := Meter(settings)
     {{- range $name, $metric := .Telemetry.Metrics }}
     builder.{{ $name.Render }}, err = meter.{{ $metric.Data.Instrument }}(
         "{{ $name }}",
         metric.WithDescription("{{ $metric.Description }}"),
         metric.WithUnit("{{ $metric.Unit }}"),
+        {{ if $metric.Data.Async -}}
+        metric.With{{ casesTitle $metric.Data.BasicType }}Callback(func(_ context.Context, o metric.{{ casesTitle $metric.Data.BasicType }}Observer) error {
+            o.Observe(builder.observe{{ $name.Render }}())
+            return nil
+        }),
+        {{- end }}
     )
     errs = errors.Join(errs, err)
     {{- end }}

cmd/mdatagen/templates/telemetry_test.go.tmpl

@@ -61,3 +61,18 @@ func TestProviders(t *testing.T) {
 		require.Fail(t, "returned Meter not mockTracer")
 	}
 }
+{{- if .Telemetry.Metrics }}
+
+func TestNewTelemetryBuilder(t *testing.T) {
+	set := component.TelemetrySettings{
+		MeterProvider: mockMeterProvider{},
+		TracerProvider: mockTracerProvider{},
+	}
+	applied := false
+	_, err := NewTelemetryBuilder(set, func(b *TelemetryBuilder) {
+		applied = true
+	})
+	require.NoError(t, err)
+	require.True(t, applied)
+}
+{{- end }}