perSecond returns empty datapoint for negative rate

transformation/binary.go

@@ -29,10 +29,19 @@ const (
 	nanosPerSecond = time.Second / time.Nanosecond
 )
 
+// perSecond computes the derivative between consecutive datapoints, taking into
+// account the time interval between the values.
+// * It skips NaN values.
+// * It assumes the timestamps are monotonically increasing, and values are non-decreasing.
+//   If either of the two conditions is not met, an empty datapoint is returned.
 func perSecond(prev, curr Datapoint) Datapoint {
 	if prev.TimeNanos >= curr.TimeNanos || math.IsNaN(prev.Value) || math.IsNaN(curr.Value) {
 		return emptyDatapoint
 	}
-	rate := (curr.Value - prev.Value) * float64(nanosPerSecond) / float64(curr.TimeNanos-prev.TimeNanos)
+	diff := curr.Value - prev.Value
+	if diff < 0 {
+		return emptyDatapoint
+	}
+	rate := diff * float64(nanosPerSecond) / float64(curr.TimeNanos-prev.TimeNanos)
 	return Datapoint{TimeNanos: curr.TimeNanos, Value: rate}
 }

transformation/binary_test.go

@@ -42,8 +42,8 @@ func TestPerSecond(t *testing.T) {
 		},
 		{
 			prev:     Datapoint{TimeNanos: time.Unix(1230, 0).UnixNano(), Value: 25},
-			curr:     Datapoint{TimeNanos: time.Unix(1240, 0).UnixNano(), Value: 20},
-			expected: Datapoint{TimeNanos: time.Unix(1240, 0).UnixNano(), Value: -0.5},
+			curr:     Datapoint{TimeNanos: time.Unix(1240, 0).UnixNano(), Value: 30},
+			expected: Datapoint{TimeNanos: time.Unix(1240, 0).UnixNano(), Value: 0.5},
 		},
 		{
 			prev:        Datapoint{TimeNanos: time.Unix(1230, 0).UnixNano(), Value: 25},
@@ -63,6 +63,12 @@ func TestPerSecond(t *testing.T) {
 			expectedNaN: true,
 			expected:    emptyDatapoint,
 		},
+		{
+			prev:        Datapoint{TimeNanos: time.Unix(1230, 0).UnixNano(), Value: 30},
+			curr:        Datapoint{TimeNanos: time.Unix(1240, 0).UnixNano(), Value: 20},
+			expectedNaN: true,
+			expected:    emptyDatapoint,
+		},
 	}
 
 	for _, input := range inputs {