Repository: numaproj/numaflow (go)

Issue:
Auto-scaling appears not to be implemented for user-defined sources
**Describe the bug**
I found [this](https://github.com/numaproj/numaflow/blob/main/pkg/apis/numaflow/v1alpha1/vertex_types.go#L117) line of code which seems to indicate that Sources of particular names are auto-scalable, which seems like it should be replaced with something better:

```
func (v Vertex) Scalable() bool {
	if v.Spec.Scale.Disabled || v.IsReduceUDF() {
		return false
	}
	if v.IsASink() || v.IsMapUDF() {
		return true
	}
	if v.IsASource() {
		src := v.Spec.Source
		if src.Kafka != nil || src.RedisStreams != nil {
			return true
		}
	}
	return false
}
```

Moreover, it seems like it will not account for the user-defined sources that are scaleable.


---

<!-- Issue Author: Don't delete this message to encourage other users to support your issue! -->

**Message from the maintainers**:

Impacted by this bug? Give it a 👍. We often sort issues this way to know what to prioritize.


Pull request description:
fix: treat ALL user-defined source vertices as scalable
fixes #1121 

### Problem Statement

Before UDSource, Numaflow has the following logic to determine whether to enable auto-scaling for a source vertex.

```go
func (v Vertex) Scalable() bool {
  // ...
	if v.IsASource() {
		src := v.Spec.Source
		if src.Kafka != nil {
			return true
		}
	}
	return false
}
```

It basically says only Kafka source is scalable.

Now that we introduced UDSource, **how do we determine if it's scalable?** And if a UDSource is scalable, **how do we auto-scale it?**



### Solution

#### How do we determine if a UDSource is scalable?

I propose that **if a source vertex is a UDSource, it is scalable.**

#### How do we auto-scale it?

I propose to **continue using the existing auto-scaling logic.**



### Case Study - there are 3 scenarios.

#### Scenario 1

The UDSource implements a **real** `pending()` method. **real** means the method actually contains logic to consult the source and get a real pending count. The count can be positive, 0 and negative. **negative** means the pending information is temporarily unavailable on the server side.

#### Scenario 2

The UDSource doesn't support `pending`, hence it's always returning a negative number. [[The numaflow grpc_udsource](https://github.com/numaproj/numaflow/blob/main/pkg/sources/udsource/grpc_udsource.go#L77)](https://github.com/numaproj/numaflow/blob/main/pkg/sources/udsource/grpc_udsource.go#L77) will treat it as `isb.PendingNotAvailable`.

#### Scenario 3

There is an error occurring on the server side, hence `pending` returns an error. [[The numaflow metric server](https://github.com/numaproj/numaflow/blob/main/pkg/metrics/metrics.go#L179)](https://github.com/numaproj/numaflow/blob/main/pkg/metrics/metrics.go#L179) will treat the error as `isb.PendingNotAvailable` , log the error and continue the program without recording any pending information.

### The 3 scenarios essentially lead to two branches: 1. non-negative pending count and 2. `isb.PendingNotAvailable`

#### branch 1: Non-Negative pending count

* The Metrics Server will build the pending information.
* The vertex is defined as scalable because it's UDSource. The auto-scaler is watching the vertex.
* When scaling
  * The scaler successfully gets a pending count.
  * The scaler calculates `desiredReplicas` and then scales the vertex accordingly. **GOOD.**

#### branch 2: `isb.PendingNotAvailable`

* The Metrics Server will skip building the pending information.
* The auto-scaler watches the vertex.
* When scaling
  * The scaler can't get a pending count.
  * The scaler [[doesn't do anything](https://github.com/numaproj/numaflow/blob/main/pkg/reconciler/vertex/scaling/scaling.go#L254)](https://github.com/numaproj/numaflow/blob/main/pkg/reconciler/vertex/scaling/scaling.go#L254). **ALSO GOOD.**

### MISC

* For sources that are NOT scalable, like HTTP, how do we set the replica count?

Numaflow [[looks for](https://github.com/numaproj/numaflow/blob/main/pkg/apis/numaflow/v1alpha1/vertex_types.go#L402)](https://github.com/numaproj/numaflow/blob/main/pkg/apis/numaflow/v1alpha1/vertex_types.go#L402) the `replicas` specification from the pipeline spec. If not specified, 1.

The controller uses the spec as [[desired replicas](https://github.com/numaproj/numaflow/blob/main/pkg/reconciler/vertex/controller.go#L123)](https://github.com/numaproj/numaflow/blob/main/pkg/reconciler/vertex/controller.go#L123). So for HTTP, if not specified, the vertex by default has 1 replica and it will never scale up or down.

* If the source vertex ever scaled down to 0, with a UDSource not implementing a real `pending`. Both pending and rate will be nil, in this case, will the vertex ever scale up?

Yes. It will. Because when a vertex has been scaled down to 0 for a certain amount of time. Auto-scaler will [[scale it up to 1](https://github.com/numaproj/numaflow/blob/main/pkg/reconciler/vertex/scaling/scaling.go#L210)](https://github.com/numaproj/numaflow/blob/main/pkg/reconciler/vertex/scaling/scaling.go#L210). This happens before the `pending` and `rate` metrics kick in.

* How is the change tested?

Tested with a user-defined Kafka source. Verified that the source vertex can scale.

<!--

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.

-->


Patch:
diff --git a/Makefile b/Makefile
index 5b77844c..e5d8ed19 100644
--- a/Makefile
+++ b/Makefile
@@ -227,17 +227,29 @@ endif
 
 /usr/local/bin/mkdocs:
 	$(PYTHON) -m pip install mkdocs==1.3.0 mkdocs_material==8.3.9 mkdocs-embed-external-markdown==2.3.0
+/usr/local/bin/lychee:
+ifeq (, $(shell which lychee))
+ifeq ($(shell uname),Darwin)
+	brew install lychee
+else
+	curl -sSfL https://github.com/lycheeverse/lychee/releases/download/v0.13.0/lychee-v0.13.0-$(shell uname -m)-unknown-linux-gnu.tar.gz | sudo tar xz -C /usr/local/bin/
+endif
+endif
 
 # docs
 
 .PHONY: docs
-docs: /usr/local/bin/mkdocs
+docs: /usr/local/bin/mkdocs docs-linkcheck
 	mkdocs build
 
 .PHONY: docs-serve
 docs-serve: docs
 	mkdocs serve
 
+.PHONY: docs-linkcheck
+docs-linkcheck: /usr/local/bin/lychee
+	lychee --exclude-path=CHANGELOG.md --exclude-mail *.md
+
 # pre-push checks
 
 .git/hooks/%: hack/git/hooks/%
diff --git a/SECURITY.md b/SECURITY.md
index cd600267..97aa9d2c 100644
--- a/SECURITY.md
+++ b/SECURITY.md
@@ -18,4 +18,4 @@ Security vulnerabilities will be disclosed via [release notes](https://github.co
 
 ## Vulnerability Scanning
 
-See [static code analysis](docs/static-code-analysis.md).
+See [static code analysis](docs/development/static-code-analysis.md).
diff --git a/pkg/apis/numaflow/v1alpha1/vertex_types.go b/pkg/apis/numaflow/v1alpha1/vertex_types.go
index c8328e8a..e807bb22 100644
--- a/pkg/apis/numaflow/v1alpha1/vertex_types.go
+++ b/pkg/apis/numaflow/v1alpha1/vertex_types.go
@@ -114,7 +114,7 @@ func (v Vertex) Scalable() bool {
 	}
 	if v.IsASource() {
 		src := v.Spec.Source
-		if src.Kafka != nil {
+		if src.Kafka != nil || src.UDSource != nil {
 			return true
 		}
 	}


Test patch:
diff --git a/pkg/apis/numaflow/v1alpha1/vertex_types_test.go b/pkg/apis/numaflow/v1alpha1/vertex_types_test.go
index 24cfc541..5a879ef4 100644
--- a/pkg/apis/numaflow/v1alpha1/vertex_types_test.go
+++ b/pkg/apis/numaflow/v1alpha1/vertex_types_test.go
@@ -535,6 +535,10 @@ func TestScalable(t *testing.T) {
 		Kafka: &KafkaSource{},
 	}
 	assert.True(t, v.Scalable())
+	v.Spec.Source = &Source{
+		UDSource: &UDSource{},
+	}
+	assert.True(t, v.Scalable())
 }
 
 func Test_Scale_Parameters(t *testing.T) {