Create a new recorder rather than reuse one for same labels (#610)

Co-authored-by: Rahul Patel <[email protected]>

sdk/metric/correct_test.go

@@ -285,15 +285,15 @@ func TestObserverCollection(t *testing.T) {
 	meter := metric.WrapMeterImpl(sdk, "test")
 
 	_ = Must(meter).RegisterFloat64Observer("float.observer", func(result metric.Float64ObserverResult) {
-		// TODO: The spec says the last-value wins in observer
-		// instruments, but it is not implemented yet, i.e., with the
-		// following line we get 1-1==0 instead of -1:
-		// result.Observe(1, meter.Labels(key.String("A", "B")))
-
+		result.Observe(1, key.String("A", "B"))
+		// last value wins
 		result.Observe(-1, key.String("A", "B"))
 		result.Observe(-1, key.String("C", "D"))
 	})
 	_ = Must(meter).RegisterInt64Observer("int.observer", func(result metric.Int64ObserverResult) {
+		result.Observe(-1, key.String("A", "B"))
+		result.Observe(1)
+		// last value wins
 		result.Observe(1, key.String("A", "B"))
 		result.Observe(1)
 	})

sdk/metric/sdk.go

@@ -216,7 +216,13 @@ func (a *asyncInstrument) getRecorder(kvs []core.KeyValue) export.Aggregator {
 
 	lrec, ok := a.recorders[labels.ordered]
 	if ok {
-		lrec.modifiedEpoch = a.meter.currentEpoch
+		if lrec.modifiedEpoch == a.meter.currentEpoch {
+			// last value wins for Observers, so if we see the same labels
+			// in the current epoch, we replace the old recorder
+			lrec.recorder = a.meter.batcher.AggregatorFor(&a.descriptor)
+		} else {
+			lrec.modifiedEpoch = a.meter.currentEpoch
+		}
 		a.recorders[labels.ordered] = lrec
 		return lrec.recorder
 	}