Repository: numaproj/numaflow
Pull request #1966 (2024-08-19 20:31:23)

chore: send negative value rateNotAvailable
fixes https://github.com/numaproj/numaflow/issues/1960

When the rater cannot calculate the processing rate due to any missing data, we should send a negative value to indicate that the rate is not available instead of 0
This helps with a corner case where the scaler can consider this as the TPS to be 0 and cause unnecessary scale downs. 

kohlisid: @vigith 
Yes, same behavior in Pipeline

The flow is the same for that as well as documented here
https://github.com/numaproj/numaflow/issues/1960#issuecomment-2297415677
codecov[bot]: ## [Codecov](https://app.codecov.io/gh/numaproj/numaflow/pull/1966?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=numaproj) Report
All modified and coverable lines are covered by tests :white_check_mark:
> Project coverage is 60.60%. Comparing base [(`4267113`)](https://app.codecov.io/gh/numaproj/numaflow/commit/42671138250d67f6eacddf33d4b5d5e069e5674f?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=numaproj) to head [(`375be63`)](https://app.codecov.io/gh/numaproj/numaflow/commit/375be63d345cc6b9d47e9fee0e15270df64d23a4?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=numaproj).

<details><summary>Additional details and impacted files</summary>


```diff
@@            Coverage Diff             @@
##             main    #1966      +/-   ##
==========================================
+ Coverage   60.59%   60.60%   +0.01%     
==========================================
  Files         317      317              
  Lines       28158    28158              
==========================================
+ Hits        17061    17064       +3     
+ Misses      10153    10152       -1     
+ Partials      944      942       -2     
```

</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/numaproj/numaflow/pull/1966?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=numaproj).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=numaproj).

kohlisid: I see another case in Pipeline though which we should potentially address, There can be scenario that we have a lookup timeline but there is no entry for a specific partition in that. But in the calculation we would send 0 as we do a simple lookup

Just trying to think whether sending 0 as rate here can have any side effects or not.

https://github.com/numaproj/numaflow/blob/42671138250d67f6eacddf33d4b5d5e069e5674f/pkg/daemon/server/service/rater/helper_test.go#L330

kohlisid: @KeranYang I have added it to the Unit tests, they seem to be passing correctly. 
I do wrap the MinInt with a float64 as well for consistency 

Patch:
diff --git a/pkg/daemon/server/service/rater/helper.go b/pkg/daemon/server/service/rater/helper.go
index 929ac99d..73683c06 100644
--- a/pkg/daemon/server/service/rater/helper.go
+++ b/pkg/daemon/server/service/rater/helper.go
@@ -17,6 +17,7 @@ limitations under the License.
 package rater
 
 import (
+	"math"
 	"time"
 
 	sharedqueue "github.com/numaproj/numaflow/pkg/shared/queue"
@@ -24,6 +25,9 @@ import (
 
 const (
 	indexNotFound = -1
+	// rateNotAvailable is returned when the processing rate cannot be derived from the currently
+	// available pod data, a negative min is returned to indicate this.
+	rateNotAvailable = float64(math.MinInt)
 )
 
 // UpdateCount updates the count of processed messages for a pod at a given time
@@ -48,14 +52,14 @@ func UpdateCount(q *sharedqueue.OverflowQueue[*TimestampedCounts], time int64, p
 func CalculateRate(q *sharedqueue.OverflowQueue[*TimestampedCounts], lookbackSeconds int64, partitionName string) float64 {
 	counts := q.Items()
 	if len(counts) <= 1 {
-		return 0
+		return rateNotAvailable
 	}
 	startIndex := findStartIndex(lookbackSeconds, counts)
 	// we consider the last but one element as the end index because the last element might be incomplete
 	// we can be sure that the last but one element in the queue is complete.
 	endIndex := len(counts) - 2
 	if startIndex == indexNotFound {
-		return 0
+		return rateNotAvailable
 	}
 
 	// time diff in seconds.
@@ -63,7 +67,7 @@ func CalculateRate(q *sharedqueue.OverflowQueue[*TimestampedCounts], lookbackSec
 	if timeDiff == 0 {
 		// if the time difference is 0, we return 0 to avoid division by 0
 		// this should not happen in practice because we are using a 10s interval
-		return 0
+		return rateNotAvailable
 	}
 
 	delta := float64(0)
diff --git a/pkg/mvtxdaemon/server/service/rater/helper.go b/pkg/mvtxdaemon/server/service/rater/helper.go
index 0973b3c5..ce9aac1e 100644
--- a/pkg/mvtxdaemon/server/service/rater/helper.go
+++ b/pkg/mvtxdaemon/server/service/rater/helper.go
@@ -17,6 +17,7 @@ limitations under the License.
 package rater
 
 import (
+	"math"
 	"time"
 
 	sharedqueue "github.com/numaproj/numaflow/pkg/shared/queue"
@@ -25,6 +26,9 @@ import (
 const (
 	// indexNotFound is returned when the start index cannot be found in the queue.
 	indexNotFound = -1
+	// rateNotAvailable is returned when the processing rate cannot be derived from the currently
+	// available pod data, a negative min is returned to indicate this.
+	rateNotAvailable = float64(math.MinInt)
 )
 
 // UpdateCount updates the count for a given timestamp in the queue.
@@ -49,14 +53,14 @@ func UpdateCount(q *sharedqueue.OverflowQueue[*TimestampedCounts], time int64, p
 func CalculateRate(q *sharedqueue.OverflowQueue[*TimestampedCounts], lookbackSeconds int64) float64 {
 	counts := q.Items()
 	if len(counts) <= 1 {
-		return 0
+		return rateNotAvailable
 	}
 	startIndex := findStartIndex(lookbackSeconds, counts)
 	// we consider the last but one element as the end index because the last element might be incomplete
 	// we can be sure that the last but one element in the queue is complete.
 	endIndex := len(counts) - 2
 	if startIndex == indexNotFound {
-		return 0
+		return rateNotAvailable
 	}
 
 	// time diff in seconds.
@@ -64,7 +68,7 @@ func CalculateRate(q *sharedqueue.OverflowQueue[*TimestampedCounts], lookbackSec
 	if timeDiff == 0 {
 		// if the time difference is 0, we return 0 to avoid division by 0
 		// this should not happen in practice because we are using a 10s interval
-		return 0
+		return rateNotAvailable
 	}
 
 	delta := float64(0)


Test patch:
diff --git a/pkg/daemon/server/service/rater/helper_test.go b/pkg/daemon/server/service/rater/helper_test.go
index e420bcc0..8873f998 100644
--- a/pkg/daemon/server/service/rater/helper_test.go
+++ b/pkg/daemon/server/service/rater/helper_test.go
@@ -119,17 +119,17 @@ func TestUpdateCount(t *testing.T) {
 }
 
 func TestCalculateRate(t *testing.T) {
-	t.Run("givenCollectedTimeLessThanTwo_whenCalculateRate_thenReturnZero", func(t *testing.T) {
+	t.Run("givenCollectedTimeLessThanTwo_whenCalculateRate_thenReturnRateNotAvailable", func(t *testing.T) {
 		q := sharedqueue.New[*TimestampedCounts](1800)
 		// no data
-		assert.Equal(t, 0.0, CalculateRate(q, 10, "partition1"))
+		assert.Equal(t, rateNotAvailable, CalculateRate(q, 10, "partition1"))
 
 		// only one data
 		now := time.Now()
 		tc1 := NewTimestampedCounts(now.Truncate(CountWindow).Unix() - 20)
 		tc1.Update(&PodReadCount{"pod1", map[string]float64{"partition1": 5.0}})
 		q.Append(tc1)
-		assert.Equal(t, 0.0, CalculateRate(q, 10, "partition1"))
+		assert.Equal(t, rateNotAvailable, CalculateRate(q, 10, "partition1"))
 	})
 
 	t.Run("singlePod_givenCountIncreases_whenCalculateRate_thenReturnRate", func(t *testing.T) {
@@ -147,9 +147,9 @@ func TestCalculateRate(t *testing.T) {
 		q.Append(tc3)
 
 		// no enough data collected within lookback seconds, expect rate 0
-		assert.Equal(t, 0.0, CalculateRate(q, 5, "partition1"))
+		assert.Equal(t, rateNotAvailable, CalculateRate(q, 5, "partition1"))
 		// no enough data collected within lookback seconds, expect rate 0
-		assert.Equal(t, 0.0, CalculateRate(q, 15, "partition1"))
+		assert.Equal(t, rateNotAvailable, CalculateRate(q, 15, "partition1"))
 		// tc1 and tc2 are used to calculate the rate
 		assert.Equal(t, 0.5, CalculateRate(q, 25, "partition1"))
 		// tc1 and tc2 are used to calculate the rate
@@ -174,9 +174,9 @@ func TestCalculateRate(t *testing.T) {
 		q.Append(tc4)
 
 		// no enough data collected within lookback seconds, expect rate 0
-		assert.Equal(t, 0.0, CalculateRate(q, 5, "partition1"))
+		assert.Equal(t, rateNotAvailable, CalculateRate(q, 5, "partition1"))
 		// no enough data collected within lookback seconds, expect rate 0
-		assert.Equal(t, 0.0, CalculateRate(q, 15, "partition1"))
+		assert.Equal(t, rateNotAvailable, CalculateRate(q, 15, "partition1"))
 		// tc2 and tc3 are used to calculate the rate
 		assert.Equal(t, 5.0, CalculateRate(q, 25, "partition1"))
 		// tc1, 2 and 3 are used to calculate the rate
@@ -203,11 +203,11 @@ func TestCalculateRate(t *testing.T) {
 		q.Append(tc3)
 
 		// no enough data collected within lookback seconds, expect rate 0
-		assert.Equal(t, 0.0, CalculateRate(q, 5, "partition1"))
+		assert.Equal(t, rateNotAvailable, CalculateRate(q, 5, "partition1"))
 		// no enough data collected within lookback seconds, expect rate 0
-		assert.Equal(t, 0.0, CalculateRate(q, 15, "partition1"))
+		assert.Equal(t, rateNotAvailable, CalculateRate(q, 15, "partition1"))
 		// no enough data collected within lookback seconds, expect rate 0
-		assert.Equal(t, 0.0, CalculateRate(q, 25, "partition1"))
+		assert.Equal(t, rateNotAvailable, CalculateRate(q, 25, "partition1"))
 		// tc1 and tc2 are used to calculate the rate
 		assert.Equal(t, 15.0, CalculateRate(q, 35, "partition1"))
 	})
@@ -230,11 +230,11 @@ func TestCalculateRate(t *testing.T) {
 		q.Append(tc3)
 
 		// no enough data collected within lookback seconds, expect rate 0
-		assert.Equal(t, 0.0, CalculateRate(q, 5, "partition1"))
+		assert.Equal(t, rateNotAvailable, CalculateRate(q, 5, "partition1"))
 		// no enough data collected within lookback seconds, expect rate 0
-		assert.Equal(t, 0.0, CalculateRate(q, 15, "partition1"))
+		assert.Equal(t, rateNotAvailable, CalculateRate(q, 15, "partition1"))
 		// no enough data collected within lookback seconds, expect rate 0
-		assert.Equal(t, 0.0, CalculateRate(q, 25, "partition1"))
+		assert.Equal(t, rateNotAvailable, CalculateRate(q, 25, "partition1"))
 		// tc1 and tc2 are used to calculate the rate
 		assert.Equal(t, 30.0, CalculateRate(q, 35, "partition1"))
 	})
@@ -257,11 +257,11 @@ func TestCalculateRate(t *testing.T) {
 		q.Append(tc3)
 
 		// no enough data collected within lookback seconds, expect rate 0
-		assert.Equal(t, 0.0, CalculateRate(q, 5, "partition1"))
+		assert.Equal(t, rateNotAvailable, CalculateRate(q, 5, "partition1"))
 		// no enough data collected within lookback seconds, expect rate 0
-		assert.Equal(t, 0.0, CalculateRate(q, 15, "partition1"))
+		assert.Equal(t, rateNotAvailable, CalculateRate(q, 15, "partition1"))
 		// no enough data collected within lookback seconds, expect rate 0
-		assert.Equal(t, 0.0, CalculateRate(q, 25, "partition1"))
+		assert.Equal(t, rateNotAvailable, CalculateRate(q, 25, "partition1"))
 		// tc1 and tc2 are used to calculate the rate
 		assert.Equal(t, 25.0, CalculateRate(q, 35, "partition1"))
 	})
@@ -292,9 +292,9 @@ func TestCalculateRate(t *testing.T) {
 
 		// partition1 rate
 		// no enough data collected within lookback seconds, expect rate 0
-		assert.Equal(t, 0.0, CalculateRate(q, 5, "partition1"))
+		assert.Equal(t, rateNotAvailable, CalculateRate(q, 5, "partition1"))
 		// no enough data collected within lookback seconds, expect rate 0
-		assert.Equal(t, 0.0, CalculateRate(q, 15, "partition1"))
+		assert.Equal(t, rateNotAvailable, CalculateRate(q, 15, "partition1"))
 		// tc2 and tc3 are used to calculate the rate
 		assert.Equal(t, 5.0, CalculateRate(q, 25, "partition1"))
 		// tc1, 2 and 3 are used to calculate the rate
@@ -303,29 +303,29 @@ func TestCalculateRate(t *testing.T) {
 		assert.Equal(t, 7.5, CalculateRate(q, 100, "partition1"))
 
 		// partition2 rate
-		assert.Equal(t, 0.0, CalculateRate(q, 5, "partition2"))
-		assert.Equal(t, 0.0, CalculateRate(q, 15, "partition2"))
+		assert.Equal(t, rateNotAvailable, CalculateRate(q, 5, "partition2"))
+		assert.Equal(t, rateNotAvailable, CalculateRate(q, 15, "partition2"))
 		assert.Equal(t, 10.0, CalculateRate(q, 25, "partition2"))
 		assert.Equal(t, 10.5, CalculateRate(q, 35, "partition2"))
 		assert.Equal(t, 10.5, CalculateRate(q, 100, "partition2"))
 
 		// partition3 rate
-		assert.Equal(t, 0.0, CalculateRate(q, 5, "partition3"))
-		assert.Equal(t, 0.0, CalculateRate(q, 15, "partition3"))
+		assert.Equal(t, rateNotAvailable, CalculateRate(q, 5, "partition3"))
+		assert.Equal(t, rateNotAvailable, CalculateRate(q, 15, "partition3"))
 		assert.Equal(t, 20.0, CalculateRate(q, 25, "partition3"))
 		assert.Equal(t, 10.0, CalculateRate(q, 35, "partition3"))
 		assert.Equal(t, 10.0, CalculateRate(q, 100, "partition3"))
 
 		// partition4 rate
-		assert.Equal(t, 0.0, CalculateRate(q, 5, "partition4"))
-		assert.Equal(t, 0.0, CalculateRate(q, 15, "partition4"))
+		assert.Equal(t, rateNotAvailable, CalculateRate(q, 5, "partition4"))
+		assert.Equal(t, rateNotAvailable, CalculateRate(q, 15, "partition4"))
 		assert.Equal(t, 10.0, CalculateRate(q, 25, "partition4"))
 		assert.Equal(t, 5.0, CalculateRate(q, 35, "partition4"))
 		assert.Equal(t, 5.0, CalculateRate(q, 100, "partition4"))
 
 		// partition100 rate
-		assert.Equal(t, 0.0, CalculateRate(q, 5, "partition100"))
-		assert.Equal(t, 0.0, CalculateRate(q, 15, "partition100"))
+		assert.Equal(t, rateNotAvailable, CalculateRate(q, 5, "partition100"))
+		assert.Equal(t, rateNotAvailable, CalculateRate(q, 15, "partition100"))
 		assert.Equal(t, 0.0, CalculateRate(q, 25, "partition100"))
 		assert.Equal(t, 0.0, CalculateRate(q, 35, "partition100"))
 		assert.Equal(t, 0.0, CalculateRate(q, 100, "partition100"))
@@ -359,9 +359,9 @@ func TestCalculateRate(t *testing.T) {
 
 		// partition1 rate
 		// no enough data collected within lookback seconds, expect rate 0
-		assert.Equal(t, 0.0, CalculateRate(q, 5, "partition1"))
+		assert.Equal(t, rateNotAvailable, CalculateRate(q, 5, "partition1"))
 		// no enough data collected within lookback seconds, expect rate 0
-		assert.Equal(t, 0.0, CalculateRate(q, 15, "partition1"))
+		assert.Equal(t, rateNotAvailable, CalculateRate(q, 15, "partition1"))
 		// tc2 and tc3 are used to calculate the rate
 		assert.Equal(t, 111.0, CalculateRate(q, 25, "partition1"))
 		// tc1, 2 and 3 are used to calculate the rate
@@ -370,8 +370,8 @@ func TestCalculateRate(t *testing.T) {
 		assert.Equal(t, 111.0, CalculateRate(q, 100, "partition1"))
 
 		// partition2 rate
-		assert.Equal(t, 0.0, CalculateRate(q, 5, "partition2"))
-		assert.Equal(t, 0.0, CalculateRate(q, 15, "partition2"))
+		assert.Equal(t, rateNotAvailable, CalculateRate(q, 5, "partition2"))
+		assert.Equal(t, rateNotAvailable, CalculateRate(q, 15, "partition2"))
 		assert.Equal(t, 111.0, CalculateRate(q, 25, "partition2"))
 		assert.Equal(t, 111.0, CalculateRate(q, 35, "partition2"))
 		assert.Equal(t, 111.0, CalculateRate(q, 100, "partition2"))
diff --git a/pkg/mvtxdaemon/server/service/rater/helper_test.go b/pkg/mvtxdaemon/server/service/rater/helper_test.go
index 6ac87824..f3e61e33 100644
--- a/pkg/mvtxdaemon/server/service/rater/helper_test.go
+++ b/pkg/mvtxdaemon/server/service/rater/helper_test.go
@@ -106,17 +106,17 @@ func TestUpdateCount(t *testing.T) {
 }
 
 func TestCalculateRate(t *testing.T) {
-	t.Run("givenCollectedTimeLessThanTwo_whenCalculateRate_thenReturnZero", func(t *testing.T) {
+	t.Run("givenCollectedTimeLessThanTwo_whenCalculateRate_thenReturnRateNotAvailable", func(t *testing.T) {
 		q := sharedqueue.New[*TimestampedCounts](1800)
 		// no data
-		assert.Equal(t, 0.0, CalculateRate(q, 10))
+		assert.Equal(t, rateNotAvailable, CalculateRate(q, 10))
 
 		// only one data
 		now := time.Now()
 		tc1 := NewTimestampedCounts(now.Truncate(CountWindow).Unix() - 20)
 		tc1.Update(&PodReadCount{"pod1", 5.0})
 		q.Append(tc1)
-		assert.Equal(t, 0.0, CalculateRate(q, 10))
+		assert.Equal(t, rateNotAvailable, CalculateRate(q, 10))
 	})
 
 	t.Run("singlePod_givenCountIncreases_whenCalculateRate_thenReturnRate", func(t *testing.T) {
@@ -134,9 +134,9 @@ func TestCalculateRate(t *testing.T) {
 		q.Append(tc3)
 
 		// no enough data collected within lookback seconds, expect rate 0
-		assert.Equal(t, 0.0, CalculateRate(q, 5))
+		assert.Equal(t, rateNotAvailable, CalculateRate(q, 5))
 		// no enough data collected within lookback seconds, expect rate 0
-		assert.Equal(t, 0.0, CalculateRate(q, 15))
+		assert.Equal(t, rateNotAvailable, CalculateRate(q, 15))
 		// tc1 and tc2 are used to calculate the rate
 		assert.Equal(t, 0.5, CalculateRate(q, 25))
 		// tc1 and tc2 are used to calculate the rate
@@ -161,9 +161,9 @@ func TestCalculateRate(t *testing.T) {
 		q.Append(tc4)
 
 		// no enough data collected within lookback seconds, expect rate 0
-		assert.Equal(t, 0.0, CalculateRate(q, 5))
+		assert.Equal(t, rateNotAvailable, CalculateRate(q, 5))
 		// no enough data collected within lookback seconds, expect rate 0
-		assert.Equal(t, 0.0, CalculateRate(q, 15))
+		assert.Equal(t, rateNotAvailable, CalculateRate(q, 15))
 		// tc2 and tc3 are used to calculate the rate
 		assert.Equal(t, 5.0, CalculateRate(q, 25))
 		// tc1, 2 and 3 are used to calculate the rate
@@ -190,11 +190,11 @@ func TestCalculateRate(t *testing.T) {
 		q.Append(tc3)
 
 		// no enough data collected within lookback seconds, expect rate 0
-		assert.Equal(t, 0.0, CalculateRate(q, 5))
+		assert.Equal(t, rateNotAvailable, CalculateRate(q, 5))
 		// no enough data collected within lookback seconds, expect rate 0
-		assert.Equal(t, 0.0, CalculateRate(q, 15))
+		assert.Equal(t, rateNotAvailable, CalculateRate(q, 15))
 		// no enough data collected within lookback seconds, expect rate 0
-		assert.Equal(t, 0.0, CalculateRate(q, 25))
+		assert.Equal(t, rateNotAvailable, CalculateRate(q, 25))
 		// tc1 and tc2 are used to calculate the rate
 		assert.Equal(t, 15.0, CalculateRate(q, 35))
 	})
@@ -217,11 +217,11 @@ func TestCalculateRate(t *testing.T) {
 		q.Append(tc3)
 
 		// no enough data collected within lookback seconds, expect rate 0
-		assert.Equal(t, 0.0, CalculateRate(q, 5))
+		assert.Equal(t, rateNotAvailable, CalculateRate(q, 5))
 		// no enough data collected within lookback seconds, expect rate 0
-		assert.Equal(t, 0.0, CalculateRate(q, 15))
+		assert.Equal(t, rateNotAvailable, CalculateRate(q, 15))
 		// no enough data collected within lookback seconds, expect rate 0
-		assert.Equal(t, 0.0, CalculateRate(q, 25))
+		assert.Equal(t, rateNotAvailable, CalculateRate(q, 25))
 		// tc1 and tc2 are used to calculate the rate
 		assert.Equal(t, 30.0, CalculateRate(q, 35))
 	})
@@ -244,11 +244,11 @@ func TestCalculateRate(t *testing.T) {
 		q.Append(tc3)
 
 		// no enough data collected within lookback seconds, expect rate 0
-		assert.Equal(t, 0.0, CalculateRate(q, 5))
+		assert.Equal(t, rateNotAvailable, CalculateRate(q, 5))
 		// no enough data collected within lookback seconds, expect rate 0
-		assert.Equal(t, 0.0, CalculateRate(q, 15))
+		assert.Equal(t, rateNotAvailable, CalculateRate(q, 15))
 		// no enough data collected within lookback seconds, expect rate 0
-		assert.Equal(t, 0.0, CalculateRate(q, 25))
+		assert.Equal(t, rateNotAvailable, CalculateRate(q, 25))
 		// tc1 and tc2 are used to calculate the rate
 		assert.Equal(t, 25.0, CalculateRate(q, 35))
 	})
@@ -279,8 +279,8 @@ func TestCalculateRate(t *testing.T) {
 		q.Append(tc4)
 
 		// vertex rate
-		assert.Equal(t, 0.0, CalculateRate(q, 5))
-		assert.Equal(t, 0.0, CalculateRate(q, 15))
+		assert.Equal(t, rateNotAvailable, CalculateRate(q, 5))
+		assert.Equal(t, rateNotAvailable, CalculateRate(q, 15))
 		assert.Equal(t, 25.0, CalculateRate(q, 25))
 		assert.Equal(t, 23.0, CalculateRate(q, 35))
 		assert.Equal(t, 23.0, CalculateRate(q, 100))