Use the PrecomputedSum for async counters

sdk/metric/pipeline.go

@@ -266,7 +266,7 @@ func (i *inserter[N]) cachedAggregator(inst view.Instrument, u unit.Unit) (inter
 	// still be applied and a warning should be logged.
 	i.logConflict(id)
 	return i.cache.LookupAggregator(id, func() (internal.Aggregator[N], error) {
-		agg, err := i.aggregator(inst.Aggregation, id.Temporality, id.Monotonic)
+		agg, err := i.aggregator(inst.Aggregation, inst.Kind, id.Temporality, id.Monotonic)
 		if err != nil {
 			return nil, err
 		}
@@ -322,16 +322,24 @@ func (i *inserter[N]) instrumentID(vi view.Instrument, u unit.Unit) instrumentID
 	return id
 }
 
-// aggregator returns a new Aggregator matching agg, temporality, and
+// aggregator returns a new Aggregator matching agg, kind, temporality, and
 // monotonic. If the agg is unknown or temporality is invalid, an error is
 // returned.
-func (i *inserter[N]) aggregator(agg aggregation.Aggregation, temporality metricdata.Temporality, monotonic bool) (internal.Aggregator[N], error) {
+func (i *inserter[N]) aggregator(agg aggregation.Aggregation, kind view.InstrumentKind, temporality metricdata.Temporality, monotonic bool) (internal.Aggregator[N], error) {
 	switch a := agg.(type) {
 	case aggregation.Drop:
 		return nil, nil
 	case aggregation.LastValue:
 		return internal.NewLastValue[N](), nil
 	case aggregation.Sum:
+		switch kind {
+		case view.AsyncCounter, view.AsyncUpDownCounter:
+			// Asynchronous counters and up-down-counters are defined to record
+			// the absolute value of the count:
+			// https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/metrics/api.md#asynchronous-counter-creation
+			return internal.NewPrecomputedSum[N](monotonic, temporality), nil
+		}
+
 		switch temporality {
 		case metricdata.CumulativeTemporality:
 			return internal.NewCumulativeSum[N](monotonic), nil