Fix monotone gauge behavior; Add monotone test

sdk/metric/aggregator/gauge/gauge.go

@@ -58,10 +58,16 @@ type (
 
 var _ export.MetricAggregator = &Aggregator{}
 
+// An unset gauge has zero timestamp and zero value.
+var unsetGauge = &gaugeData{}
+
 // New returns a new gauge aggregator.  This aggregator retains the
 // last value and timestamp that were recorded.
 func New() *Aggregator {
-	return &Aggregator{}
+	return &Aggregator{
+		live: unsafe.Pointer(unsetGauge),
+		save: unsafe.Pointer(unsetGauge),
+	}
 }
 
 // AsInt64 returns the recorded gauge value as an int64.
@@ -76,13 +82,7 @@ func (g *Aggregator) Timestamp() time.Time {
 
 // Collect saves the current value (atomically) and exports it.
 func (g *Aggregator) Collect(ctx context.Context, rec export.MetricRecord, exp export.MetricBatcher) {
-	g.save = atomic.SwapPointer(&g.live, nil)
-
-	if g.save == nil {
-		// There is no current value. This indicates a harmless race
-		// involving Collect() and DeleteHandle().
-		return
-	}
+	g.save = atomic.LoadPointer(&g.live)
 
 	exp.Export(ctx, rec, g)
 }
@@ -115,11 +115,9 @@ func (g *Aggregator) updateMonotonic(number core.Number, desc *export.Descriptor
 	for {
 		gd := (*gaugeData)(atomic.LoadPointer(&g.live))
 
-		if gd != nil {
-			if gd.value.CompareNumber(kind, number) >= 0 {
-				// TODO warn
-				return
-			}
+		if gd.value.CompareNumber(kind, number) >= 0 {
+			// TODO warn
+			return
 		}
 
 		if atomic.CompareAndSwapPointer(&g.live, unsafe.Pointer(gd), unsafe.Pointer(ngd)) {

sdk/metric/monotone_test.go

@@ -0,0 +1,115 @@
+// Copyright 2019, OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package metric_test
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/api/core"
+	"go.opentelemetry.io/api/key"
+	"go.opentelemetry.io/api/metric"
+	"go.opentelemetry.io/sdk/export"
+	sdk "go.opentelemetry.io/sdk/metric"
+	"go.opentelemetry.io/sdk/metric/aggregator/gauge"
+)
+
+type monotoneBatcher struct {
+	t *testing.T
+
+	collections  int
+	currentValue *core.Number
+	currentTime  *time.Time
+}
+
+func (m *monotoneBatcher) AggregatorFor(rec export.MetricRecord) export.MetricAggregator {
+	return gauge.New()
+}
+
+func (m *monotoneBatcher) Export(_ context.Context, record export.MetricRecord, agg export.MetricAggregator) {
+	require.Equal(m.t, "my.gauge.name", record.Descriptor().Name())
+	require.Equal(m.t, 1, len(record.Labels()))
+	require.Equal(m.t, "a", string(record.Labels()[0].Key))
+	require.Equal(m.t, "b", record.Labels()[0].Value.Emit())
+
+	gauge := agg.(*gauge.Aggregator)
+	val := gauge.AsNumber()
+	ts := gauge.Timestamp()
+
+	m.currentValue = &val
+	m.currentTime = &ts
+	m.collections++
+}
+
+func TestMonotoneGauge(t *testing.T) {
+	ctx := context.Background()
+	batcher := &monotoneBatcher{
+		t: t,
+	}
+	sdk := sdk.New(batcher)
+
+	gauge := sdk.NewInt64Gauge("my.gauge.name", metric.WithMonotonic(true))
+
+	handle := gauge.AcquireHandle(sdk.Labels(key.String("a", "b")))
+
+	require.Nil(t, batcher.currentTime)
+	require.Nil(t, batcher.currentValue)
+
+	before := time.Now()
+
+	handle.Set(ctx, 1)
+
+	// Until collection, expect nil.
+	require.Nil(t, batcher.currentTime)
+	require.Nil(t, batcher.currentValue)
+
+	sdk.Collect(ctx)
+
+	require.NotNil(t, batcher.currentValue)
+	require.Equal(t, core.NewInt64Number(1), *batcher.currentValue)
+	require.True(t, before.Before(*batcher.currentTime))
+
+	before = *batcher.currentTime
+
+	// Collect would ordinarily flush the record, except we're using a handle.
+	sdk.Collect(ctx)
+
+	require.Equal(t, 2, batcher.collections)
+
+	// Increase the value to 2.
+	handle.Set(ctx, 2)
+
+	sdk.Collect(ctx)
+
+	require.Equal(t, 3, batcher.collections)
+	require.Equal(t, core.NewInt64Number(2), *batcher.currentValue)
+	require.True(t, before.Before(*batcher.currentTime))
+
+	before = *batcher.currentTime
+
+	sdk.Collect(ctx)
+	require.Equal(t, 4, batcher.collections)
+
+	// Try to lower the value to 1, it will fail.
+	handle.Set(ctx, 1)
+	sdk.Collect(ctx)
+
+	// The value and timestamp are both unmodified
+	require.Equal(t, 5, batcher.collections)
+	require.Equal(t, core.NewInt64Number(2), *batcher.currentValue)
+	require.Equal(t, before, *batcher.currentTime)
+}