Repository: numaproj/numaflow
Pull request #639 (2023-03-27 20:39:02)

fix: desired replicas should not be greater than pending
1. Excluding `ackpending` from Pending() will cause unexpected scalding down to 0 when it takes a long time to process a message, and it happens there's no real pending messages in the buffer. Reverted that logic.
2. Let the desired replicas <= pending.

<!--

Before you push your changes:

* Run `make pre-push -B` to fix codegen and lint problems.

Then, you MUST:

* Sign-off your commit (otherwise the DCO check will fail).
* Use [a conventional commit message](https://www.conventionalcommits.org/en/v1.0.0/) (otherwise the commit message check will fail).

If you did not do this, reset all your commit and replace them with a single commit:

```
git reset HEAD~1 ;# change 1 to how many commits you made
git commit --signoff -m 'feat: my feat. Fixes #1234'
```

When creating your PR: 

* Make sure that "Fixes #" or "Closes #" is in both the PR title (for release notes) and description (to automatically link and close the issue).
* Say how you tested your changes. If you changed the UI, attach screenshots.
* Set your PR as a draft initially.
* Your PR needs to pass the required checks before it can be approved. 
* Once required tests have passed, mark your PR "Ready for review".

If changes were requested, once you've made them, you MUST dismiss the review to get it reviewed again.

-->


vigith: this will reopen https://github.com/numaproj/numaflow/issues/630, right?
whynowy: > this will reopen #630, right?

There's not a perfect solution for all the use cases, we will not do anything for that issue - having something to fix one corner case makes trouble for some other cases. Simply applying a `max` will mitigate that issue.

whynowy: > > this will reopen #630, right?
> 
> There's not a perfect solution for all the use cases, we will not do anything for that issue - having something to fix one corner case makes trouble for some other cases. Simply applying a `max` will mitigate that issue.

Actually there's already a 200 max in place, in this case, the parameters should be fined tuned.
whynowy: > > > this will reopen #630, right?
> > 
> > 
> > There's not a perfect solution for all the use cases, we will not do anything for that issue - having something to fix one corner case makes trouble for some other cases. Simply applying a `max` will mitigate that issue.
> 
> Actually there's already a 200 max in place, in this case, the parameters should be fined tuned.

Or maybe some simple logic we can add, like the replica number should not be more than the calculated `pending`.

Patch:
diff --git a/Makefile b/Makefile
index 81ba344e..f19bda21 100644
--- a/Makefile
+++ b/Makefile
@@ -81,7 +81,7 @@ dist/$(BINARY_NAME)-%:
 
 .PHONY: test
 test:
-	go test $(shell go list ./... | grep -v /vendor/ | grep -v /numaflow/test/) -race -short -v
+	go test $(shell go list ./... | grep -v /vendor/ | grep -v /numaflow/test/) -race -short -v -timeout 60s
 
 .PHONY: test-coverage
 test-coverage:
@@ -100,7 +100,7 @@ test-coverage-with-isb:
 
 .PHONY: test-code
 test-code:
-	go test -tags=isb_redis -race -v $(shell go list ./... | grep -v /vendor/ | grep -v /numaflow/test/)
+	go test -tags=isb_redis -race -v $(shell go list ./... | grep -v /vendor/ | grep -v /numaflow/test/) -timeout 120s
 
 test-e2e:
 test-kafka-e2e:
diff --git a/pkg/isb/stores/jetstream/reader.go b/pkg/isb/stores/jetstream/reader.go
index 72df5614..84421556 100644
--- a/pkg/isb/stores/jetstream/reader.go
+++ b/pkg/isb/stores/jetstream/reader.go
@@ -191,7 +191,7 @@ func (jr *jetStreamReader) Pending(_ context.Context) (int64, error) {
 	if err != nil {
 		return isb.PendingNotAvailable, fmt.Errorf("failed to get consumer info, %w", err)
 	}
-	return int64(c.NumPending), nil
+	return int64(c.NumPending) + int64(c.NumAckPending), nil
 }
 
 // Rate returns the ack rate (tps) in the past seconds
diff --git a/pkg/isb/stores/simplebuffer/buffer.go b/pkg/isb/stores/simplebuffer/buffer.go
index 9a202f4f..1d448c8b 100644
--- a/pkg/isb/stores/simplebuffer/buffer.go
+++ b/pkg/isb/stores/simplebuffer/buffer.go
@@ -180,6 +180,9 @@ func (b *InMemoryBuffer) Read(ctx context.Context, count int64) ([]*isb.ReadMess
 			if errors.Is(err, context.Canceled) {
 				return readMessages, nil
 			}
+			if errors.Is(err, context.DeadlineExceeded) {
+				return readMessages, nil
+			}
 			return readMessages, isb.BufferReadErr{Name: b.name, Empty: true, Message: err.Error()}
 		}
 		// access buffer via lock
diff --git a/pkg/reconciler/vertex/scaling/scaling.go b/pkg/reconciler/vertex/scaling/scaling.go
index c2cc810c..e26d12ac 100644
--- a/pkg/reconciler/vertex/scaling/scaling.go
+++ b/pkg/reconciler/vertex/scaling/scaling.go
@@ -246,14 +246,16 @@ func (s *Scaler) scaleOneVertex(ctx context.Context, key string, worker int) err
 	}
 	current := int32(vertex.GetReplicas())
 	desired := s.desiredReplicas(ctx, vertex, rate, pending, totalBufferLength, targetAvailableBufferLength)
-	log.Debugf("Calculated desired replica number of vertex %q is: %v", vertex.Name, desired)
+	log.Debugf("Calculated desired replica number of vertex %q is: %t", vertex.Name, desired)
 	max := vertex.Spec.Scale.GetMaxReplicas()
 	min := vertex.Spec.Scale.GetMinReplicas()
 	if desired > max {
 		desired = max
+		log.Debugf("Calculated desired replica number %t of vertex %q is greater than max, using max %t", vertex.Name, desired, max)
 	}
 	if desired < min {
 		desired = min
+		log.Debugf("Calculated desired replica number %t of vertex %q is smaller than min, using min %t", vertex.Name, desired, min)
 	}
 	if current > max || current < min { // Someone might have manually scaled up/down the vertex
 		return s.patchVertexReplicas(ctx, vertex, desired)
@@ -294,31 +296,35 @@ func (s *Scaler) desiredReplicas(ctx context.Context, vertex *dfv1.Vertex, rate
 	if rate == 0 && pending == 0 { // This could scale down to 0
 		return 0
 	}
-	if rate == 0 { // Something is wrong, we don't do anything.
+	if pending == 0 || rate == 0 {
+		// Pending is 0 and rate is not 0, or rate is 0 and pending is not 0, we don't do anything.
+		// Technically this would not happen because the pending includes ackpending, which means rate and pending are either both 0, or both > 0.
+		// But we still keep this check here for safety.
 		return int32(vertex.Status.Replicas)
 	}
+	var desired int32
 	if vertex.IsASource() {
 		// For sources, we calculate the time of finishing processing the pending messages,
 		// and then we know how many replicas are needed to get them done in target seconds.
-		desired := int32(math.Round(((float64(pending) / rate) / float64(vertex.Spec.Scale.GetTargetProcessingSeconds())) * float64(vertex.Status.Replicas)))
-		if desired == 0 {
-			desired = 1
-		}
-		return desired
+		desired = int32(math.Round(((float64(pending) / rate) / float64(vertex.Spec.Scale.GetTargetProcessingSeconds())) * float64(vertex.Status.Replicas)))
 	} else {
 		// For UDF and sinks, we calculate the available buffer length, and consider it is the contribution of current replicas,
 		// then we figure out how many replicas are needed to keep the available buffer length at target level.
 		if pending >= totalBufferLength {
 			// Simply return current replica number + max allowed if the pending messages are more than available buffer length
-			return int32(vertex.Status.Replicas) + int32(vertex.Spec.Scale.GetReplicasPerScale())
-		}
-		singleReplicaContribution := float64(totalBufferLength-pending) / float64(vertex.Status.Replicas)
-		desired := int32(math.Round(float64(targetAvailableBufferLength) / singleReplicaContribution))
-		if desired == 0 {
-			desired = 1
+			desired = int32(vertex.Status.Replicas) + int32(vertex.Spec.Scale.GetReplicasPerScale())
+		} else {
+			singleReplicaContribution := float64(totalBufferLength-pending) / float64(vertex.Status.Replicas)
+			desired = int32(math.Round(float64(targetAvailableBufferLength) / singleReplicaContribution))
 		}
-		return desired
 	}
+	if desired == 0 {
+		desired = 1
+	}
+	if desired > int32(pending) { // For some corner cases, we don't want to scale up to more than pending.
+		desired = int32(pending)
+	}
+	return desired
 }
 
 // Start function starts the autoscaling worker group.


Test patch:
diff --git a/pkg/reconciler/vertex/scaling/scaling_test.go b/pkg/reconciler/vertex/scaling/scaling_test.go
index 004d4c33..5a39012d 100644
--- a/pkg/reconciler/vertex/scaling/scaling_test.go
+++ b/pkg/reconciler/vertex/scaling/scaling_test.go
@@ -65,7 +65,7 @@ func Test_desiredReplicas(t *testing.T) {
 	assert.Equal(t, int32(7), s.desiredReplicas(context.TODO(), src, 2500, 8749, 30000, 20000))
 	assert.Equal(t, int32(2), s.desiredReplicas(context.TODO(), src, 0, 9950, 30000, 20000))
 	assert.Equal(t, int32(1), s.desiredReplicas(context.TODO(), src, 2500, 2, 30000, 20000))
-	assert.Equal(t, int32(1), s.desiredReplicas(context.TODO(), src, 2500, 0, 30000, 20000))
+	assert.Equal(t, int32(2), s.desiredReplicas(context.TODO(), src, 2500, 0, 30000, 20000))
 
 	udf := &dfv1.Vertex{
 		Spec: dfv1.VertexSpec{