Repository: numaproj/numaflow (go)

Issue:
nil pointer dereference when running 1-simple-pipeline with minimal crd
**Describe the bug**
When installing with a namespaced mode it is suggested that user should use minimal crd (ref: https://github.com/numaproj/numaflow/blob/29f15d57a125c54b6de8c4cd29d16e3d8473b655/docs/installation.md#namespace-scope).  When doing this the `simple-pipeline` example (ref: https://github.com/numaproj/numaflow/blob/27e6a8755bd1092b31d93fac891d1decd006a093/examples/1-simple-pipeline.yaml) errors with a panic:

```
panic: runtime error: invalid memory address or nil pointer dereference
[signal SIGSEGV: segmentation violation code=0x1 addr=0x0 pc=0x1a66f29]

goroutine 1 [running]:
github.com/numaproj/numaflow/pkg/sources.(*SourceProcessor).getSourcer(0xc0009a7d78, {0xc0001abd80, 0x1, 0x1}, {0x23b7b80, 0xc0005ee690}, 0x23b2100?, {0x23b7bd0, 0xc0007f82e0}, 0xc000493b70)
	/home/dseapy/github/numaflow/pkg/sources/source.go:145 +0x4c9
github.com/numaproj/numaflow/pkg/sources.(*SourceProcessor).Start(0xc0009a7d78, {0x23cc038, 0xc0001e83f0})
	/home/dseapy/github/numaflow/pkg/sources/source.go:95 +0x465
github.com/numaproj/numaflow/cmd/commands.NewProcessorCommand.func1(0xc00059f680?, {0x2076944?, 0x2?, 0x2?})
	/home/dseapy/github/numaflow/cmd/commands/processor.go:67 +0x525
github.com/spf13/cobra.(*Command).execute(0xc00059f680, {0xc000130fe0, 0x2, 0x2})
	/home/dseapy/go/pkg/mod/github.com/spf13/cobra@v1.2.1/command.go:856 +0x67c
github.com/spf13/cobra.(*Command).ExecuteC(0x33a1440)
	/home/dseapy/go/pkg/mod/github.com/spf13/cobra@v1.2.1/command.go:974 +0x3bd
github.com/spf13/cobra.(*Command).Execute(...)
	/home/dseapy/go/pkg/mod/github.com/spf13/cobra@v1.2.1/command.go:902
github.com/numaproj/numaflow/cmd/commands.Execute()
	/home/dseapy/github/numaflow/cmd/commands/root.go:16 +0x25
main.main()
	/home/dseapy/github/numaflow/cmd/main.go:8 +0x17
```

This is caused due to `msgSize` missing in the example, which doesn't get defaulted when using minimal crd.

**To Reproduce**
Steps to reproduce the behavior:

1. Install minimal crd in cluster
2. kubectl apply the `1-simple-pipeline.yaml` example
3. kubectl logs the `simple-pipeline-in-0-...` pod

**Expected behavior**
* Error message provides indication as to what is missing.
* Warning in installation guide that full crd leverages defaults which will not be populated by the minimal crd.  Therefore some examples may be incomplete if using the minimal crd.


**Environment (please complete the following information):**

- Kubernetes: v1.25.3
- Numaflow: head of main

---

<!-- 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:
fix: nil pointer deref when running example with minimal CRD. Fixes #260
Signed-off-by: David Seapy <dseapy@gmail.com>

Fixes #260.

Avoids nil pointer dereference.  Adds note about using minimal CRD with examples that may rely on defaults present in full CRD.

<!--

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/docs/installation.md b/docs/installation.md
index 87b4809d..01547d31 100644
--- a/docs/installation.md
+++ b/docs/installation.md
@@ -28,7 +28,7 @@ namespace: numaflow-system
 
 A namespace installation only watches and executes pipelines in the namespace it is installed (typically `numaflow-system`).
 
-Add an argument `--namespaced` to the `numaflow-controller` deployment to achieve namespace scope installation.
+Add an argument `--namespaced` to the `numaflow-controller` and `numaflow-server` deployments to achieve namespace scope installation.
 
 ```
       - args:
@@ -61,7 +61,7 @@ namespace: numaflow-system
 
 A managed namespace installation watches and executes pipelines in a specific namespace.
 
-To do managed namespace installation, besides `--namespaced`, add `--managed-namespace` and the specific namespace to the `numaflow-controller` deployment arguments.
+To do managed namespace installation, besides `--namespaced`, add `--managed-namespace` and the specific namespace to the `numaflow-controller` and `numaflow-server` deployment arguments.
 
 ```
       - args:
diff --git a/pkg/apis/numaflow/v1alpha1/vertex_types.go b/pkg/apis/numaflow/v1alpha1/vertex_types.go
index 96562e69..4d40bd6a 100644
--- a/pkg/apis/numaflow/v1alpha1/vertex_types.go
+++ b/pkg/apis/numaflow/v1alpha1/vertex_types.go
@@ -137,7 +137,7 @@ func (v Vertex) getServiceObj(name string, headless bool, port int, servicePortN
 	return svc
 }
 
-func (v Vertex) commonEvns() []corev1.EnvVar {
+func (v Vertex) commonEnvs() []corev1.EnvVar {
 	return []corev1.EnvVar{
 		{Name: EnvNamespace, ValueFrom: &corev1.EnvVarSource{FieldRef: &corev1.ObjectFieldSelector{FieldPath: "metadata.namespace"}}},
 		{Name: EnvPod, ValueFrom: &corev1.EnvVarSource{FieldRef: &corev1.ObjectFieldSelector{FieldPath: "metadata.name"}}},
@@ -163,7 +163,7 @@ func (v Vertex) GetPodSpec(req GetVertexPodSpecReq) (*corev1.PodSpec, error) {
 	envVars := []corev1.EnvVar{
 		{Name: EnvVertexObject, Value: encodedVertexSpec},
 	}
-	envVars = append(envVars, v.commonEvns()...)
+	envVars = append(envVars, v.commonEnvs()...)
 	envVars = append(envVars, req.Env...)
 	resources := standardResources
 	if v.Spec.ContainerTemplate != nil {
@@ -221,7 +221,7 @@ func (v Vertex) GetPodSpec(req GetVertexPodSpecReq) (*corev1.PodSpec, error) {
 	}
 
 	if len(containers) > 1 { // udf and udsink
-		containers[1].Env = append(containers[1].Env, v.commonEvns()...)
+		containers[1].Env = append(containers[1].Env, v.commonEnvs()...)
 	}
 
 	spec := &corev1.PodSpec{
diff --git a/pkg/sources/generator/tickgen.go b/pkg/sources/generator/tickgen.go
index 8a42be69..1a068ee1 100644
--- a/pkg/sources/generator/tickgen.go
+++ b/pkg/sources/generator/tickgen.go
@@ -111,17 +111,26 @@ func WithReadTimeOut(timeout time.Duration) Option {
 //	is created for use by this vertex.
 //
 // name - name of this vertex
-// rpu  - no of records to generate per time unit. by default the channel buffer size is set to 5*rpu
-// msgSize - size of each generated message
-// timeunit - unit of time per tick. could be any golang time.Duration.
 // writers - destinations to write to
 func NewMemGen(vertexInstance *dfv1.VertexInstance,
-	rpu int,
-	msgSize int32,
-	timeunit time.Duration,
 	writers []isb.BufferWriter,
 	fetchWM fetch.Fetcher, publishWM map[string]publish.Publisher, publishWMStores store.WatermarkStorer, // watermarks
 	opts ...Option) (*memgen, error) {
+
+	// minimal CRDs don't have defaults
+	rpu := 5
+	if vertexInstance.Vertex.Spec.Source.Generator.RPU != nil {
+		rpu = int(*(vertexInstance.Vertex.Spec.Source.Generator.RPU))
+	}
+	msgSize := int32(8)
+	if vertexInstance.Vertex.Spec.Source.Generator.MsgSize != nil {
+		msgSize = *vertexInstance.Vertex.Spec.Source.Generator.MsgSize
+	}
+	timeunit := time.Second
+	if vertexInstance.Vertex.Spec.Source.Generator.Duration != nil {
+		timeunit = vertexInstance.Vertex.Spec.Source.Generator.Duration.Duration
+	}
+
 	gensrc := &memgen{
 		rpu:            rpu,
 		msgSize:        msgSize,
diff --git a/pkg/sources/source.go b/pkg/sources/source.go
index 0e66a3cc..d1e4d915 100644
--- a/pkg/sources/source.go
+++ b/pkg/sources/source.go
@@ -142,7 +142,7 @@ func (sp *SourceProcessor) getSourcer(writers []isb.BufferWriter, fetchWM fetch.
 		if l := sp.VertexInstance.Vertex.Spec.Limits; l != nil && l.ReadTimeout != nil {
 			readOptions = append(readOptions, generator.WithReadTimeOut(l.ReadTimeout.Duration))
 		}
-		return generator.NewMemGen(sp.VertexInstance, int(*x.RPU), *x.MsgSize, x.Duration.Duration, writers, fetchWM, publishWM, publishWMStores, readOptions...)
+		return generator.NewMemGen(sp.VertexInstance, writers, fetchWM, publishWM, publishWMStores, readOptions...)
 	} else if x := src.Kafka; x != nil {
 		readOptions := []kafka.Option{
 			kafka.WithGroupName(x.ConsumerGroupName),


Test patch:
diff --git a/pkg/sources/generator/tickgen_test.go b/pkg/sources/generator/tickgen_test.go
index 89b88cca..358c968e 100644
--- a/pkg/sources/generator/tickgen_test.go
+++ b/pkg/sources/generator/tickgen_test.go
@@ -20,9 +20,20 @@ import (
 func TestRead(t *testing.T) {
 	dest := simplebuffer.NewInMemoryBuffer("writer", 20)
 	ctx := context.Background()
-	vertex := &dfv1.Vertex{ObjectMeta: v1.ObjectMeta{
-		Name: "memgen",
-	}}
+	vertex := &dfv1.Vertex{
+		ObjectMeta: v1.ObjectMeta{
+			Name: "memgen",
+		},
+		Spec: dfv1.VertexSpec{
+			PipelineName: "testPipeline",
+			AbstractVertex: dfv1.AbstractVertex{
+				Name: "testVertex",
+				Source: &dfv1.Source{
+					Generator: &dfv1.GeneratorSource{},
+				},
+			},
+		},
+	}
 	m := &dfv1.VertexInstance{
 		Vertex:   vertex,
 		Hostname: "TestRead",
@@ -34,7 +45,7 @@ func TestRead(t *testing.T) {
 		"writer": dest,
 	}
 	fetchWatermark, publishWatermark := generic.BuildNoOpWatermarkProgressorsFromBufferMap(toSteps)
-	mgen, err := NewMemGen(m, 5, 8, time.Millisecond, []isb.BufferWriter{dest}, fetchWatermark, publishWatermark, publishWMStore)
+	mgen, err := NewMemGen(m, []isb.BufferWriter{dest}, fetchWatermark, publishWatermark, publishWMStore)
 	assert.NoError(t, err)
 	_ = mgen.Start()
 
@@ -54,9 +65,20 @@ func TestStop(t *testing.T) {
 	ctx := context.Background()
 
 	dest := simplebuffer.NewInMemoryBuffer("writer", 100)
-	vertex := &dfv1.Vertex{ObjectMeta: v1.ObjectMeta{
-		Name: "memgen",
-	}}
+	vertex := &dfv1.Vertex{
+		ObjectMeta: v1.ObjectMeta{
+			Name: "memgen",
+		},
+		Spec: dfv1.VertexSpec{
+			PipelineName: "testPipeline",
+			AbstractVertex: dfv1.AbstractVertex{
+				Name: "testVertex",
+				Source: &dfv1.Source{
+					Generator: &dfv1.GeneratorSource{},
+				},
+			},
+		},
+	}
 	m := &dfv1.VertexInstance{
 		Vertex:   vertex,
 		Hostname: "TestRead",
@@ -67,7 +89,7 @@ func TestStop(t *testing.T) {
 		"writer": dest,
 	}
 	fetchWatermark, publishWatermark := generic.BuildNoOpWatermarkProgressorsFromBufferMap(toSteps)
-	mgen, err := NewMemGen(m, 5, 8, time.Millisecond, []isb.BufferWriter{dest}, fetchWatermark, publishWatermark, publishWMStore)
+	mgen, err := NewMemGen(m, []isb.BufferWriter{dest}, fetchWatermark, publishWatermark, publishWMStore)
 	assert.NoError(t, err)
 	stop := mgen.Start()
 
diff --git a/pkg/sources/generator/watermark_test.go b/pkg/sources/generator/watermark_test.go
index 37a8371b..bdb5ea6a 100644
--- a/pkg/sources/generator/watermark_test.go
+++ b/pkg/sources/generator/watermark_test.go
@@ -26,16 +26,27 @@ func TestWatermark(t *testing.T) {
 	defer func() { _ = os.Unsetenv(dfv1.EnvWatermarkDisabled) }()
 
 	dest := simplebuffer.NewInMemoryBuffer("writer", 1000)
-	vertex := &dfv1.Vertex{ObjectMeta: v1.ObjectMeta{
-		Name: "memgen",
-	}}
+	vertex := &dfv1.Vertex{
+		ObjectMeta: v1.ObjectMeta{
+			Name: "memgen",
+		},
+		Spec: dfv1.VertexSpec{
+			PipelineName: "testPipeline",
+			AbstractVertex: dfv1.AbstractVertex{
+				Name: "testVertex",
+				Source: &dfv1.Source{
+					Generator: &dfv1.GeneratorSource{},
+				},
+			},
+		},
+	}
 	m := &dfv1.VertexInstance{
 		Vertex:   vertex,
 		Hostname: "TestRead",
 		Replica:  0,
 	}
 	publishWMStore := store.BuildWatermarkStore(noop.NewKVNoOpStore(), noop.NewKVNoOpStore())
-	mgen, err := NewMemGen(m, 1, 8, time.Millisecond, []isb.BufferWriter{dest}, nil, nil, publishWMStore)
+	mgen, err := NewMemGen(m, []isb.BufferWriter{dest}, nil, nil, publishWMStore)
 	assert.NoError(t, err)
 	stop := mgen.Start()