Repository: numaproj/numaflow (go)

Issue:
Better handling for periods in pipeline name
# Summary

This passes validation in the controller:
```yaml
apiVersion: numaflow.numaproj.io/v1alpha1
kind: Pipeline
metadata:
  name: inf-udf-v0.2
...
```
But then errors trying to create the service for daemon:
```json
{
  "level": "error",
  "ts": 1668268595.0106926,
  "logger": "numaflow.controller-manager",
  "caller": "pipeline/controller.go:82",
  "msg": "Reconcile error",
  "namespace": "default",
  "pipeline": "inf-udf-v0.2",
  "error": "failed to create daemon service, Service \"inf-udf-v0.2-daemon-svc\" is invalid: metadata.name: Invalid value: \"inf-udf-v0.2-daemon-svc\": a DNS-1035 label must consist of lower case alphanumeric characters or '-', start with an alphabetic character, and end with an alphanumeric character (e.g. 'my-name',  or 'abc-123', regex used for validation is '[a-z]([-a-z0-9]*[a-z0-9])?')",
  "stacktrace": "..."
}
```

Would be good to not pass Pipeline validation if periods aren't allowed, or allow them and adjust service name to compensate.

---
<!-- Issue Author: Don't delete this message to encourage other users to support your issue! -->
**Message from the maintainers**:

If you wish to see this enhancement implemented please add a 👍 reaction to this issue! We often sort issues this way to know what to prioritize.


Pull request description:
chore: add pipeline and vertex name validation. Closes #342
Signed-off-by: Derek Wang <whynowy@gmail.com>

Closes #342

<!--

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/pkg/reconciler/pipeline/validate.go b/pkg/reconciler/pipeline/validate.go
index 7cf5bd83..fffc8729 100644
--- a/pkg/reconciler/pipeline/validate.go
+++ b/pkg/reconciler/pipeline/validate.go
@@ -19,6 +19,8 @@ package pipeline
 import (
 	"fmt"
 
+	k8svalidation "k8s.io/apimachinery/pkg/util/validation"
+
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 )
 
@@ -26,6 +28,11 @@ func ValidatePipeline(pl *dfv1.Pipeline) error {
 	if pl == nil {
 		return fmt.Errorf("nil pipeline")
 	}
+
+	if errs := k8svalidation.IsDNS1035Label(pl.Name); len(errs) > 0 {
+		return fmt.Errorf("invalid pipeline name %q, %v", pl.Name, errs)
+	}
+
 	if len(pl.Spec.Vertices) == 0 {
 		return fmt.Errorf("empty vertices")
 	}
@@ -169,6 +176,9 @@ func ValidatePipeline(pl *dfv1.Pipeline) error {
 }
 
 func validateVertex(v dfv1.AbstractVertex) error {
+	if errs := k8svalidation.IsDNS1035Label(v.Name); len(errs) > 0 {
+		return fmt.Errorf("invalid vertex name %q, %v", v.Name, errs)
+	}
 	min, max := int32(0), int32(dfv1.DefaultMaxReplicas)
 	if v.Scale.Min != nil {
 		min = *v.Scale.Min


Test patch:
diff --git a/pkg/reconciler/pipeline/validate_test.go b/pkg/reconciler/pipeline/validate_test.go
index f1ce40bc..f78b7dc9 100644
--- a/pkg/reconciler/pipeline/validate_test.go
+++ b/pkg/reconciler/pipeline/validate_test.go
@@ -128,6 +128,14 @@ func TestValidatePipeline(t *testing.T) {
 	})
 
 	t.Run("test nil pipeline", func(t *testing.T) {
+		testObj := testPipeline.DeepCopy()
+		testObj.Name = "invalid.name"
+		err := ValidatePipeline(testObj)
+		assert.Error(t, err)
+		assert.Contains(t, err.Error(), "invalid pipeline name")
+	})
+
+	t.Run("test invalid pipeline name", func(t *testing.T) {
 		err := ValidatePipeline(nil)
 		assert.Error(t, err)
 	})
@@ -313,11 +321,21 @@ func TestValidateReducePipeline(t *testing.T) {
 }
 
 func TestValidateVertex(t *testing.T) {
+	t.Run("test invalid vertex name", func(t *testing.T) {
+		v := dfv1.AbstractVertex{
+			Name: "invalid.name",
+		}
+		err := validateVertex(v)
+		assert.Error(t, err)
+		assert.Contains(t, err.Error(), "invalid vertex name")
+	})
+
 	goodContainers := []corev1.Container{{Name: "my-test-image", Image: "my-image:latest"}}
 	badContainers := []corev1.Container{{Name: dfv1.CtrInit, Image: "my-image:latest"}}
 
 	t.Run("bad min", func(t *testing.T) {
 		v := dfv1.AbstractVertex{
+			Name: "my-vertex",
 			Scale: dfv1.Scale{
 				Min: pointer.Int32(-1),
 				Max: pointer.Int32(1),
@@ -330,6 +348,7 @@ func TestValidateVertex(t *testing.T) {
 
 	t.Run("min > max", func(t *testing.T) {
 		v := dfv1.AbstractVertex{
+			Name: "my-vertex",
 			Scale: dfv1.Scale{
 				Min: pointer.Int32(2),
 				Max: pointer.Int32(1),
@@ -341,26 +360,26 @@ func TestValidateVertex(t *testing.T) {
 	})
 
 	t.Run("good init container", func(t *testing.T) {
-		v := dfv1.AbstractVertex{InitContainers: goodContainers}
+		v := dfv1.AbstractVertex{Name: "my-vertex", InitContainers: goodContainers}
 		err := validateVertex(v)
 		assert.NoError(t, err)
 	})
 
 	t.Run("bad init container name", func(t *testing.T) {
-		v := dfv1.AbstractVertex{InitContainers: badContainers}
+		v := dfv1.AbstractVertex{Name: "my-vertex", InitContainers: badContainers}
 		err := validateVertex(v)
 		assert.Error(t, err)
 		assert.Contains(t, err.Error(), "is reserved for containers created by numaflow")
 	})
 
 	t.Run("good sidecar container", func(t *testing.T) {
-		v := dfv1.AbstractVertex{Sidecars: goodContainers}
+		v := dfv1.AbstractVertex{Name: "my-vertex", Sidecars: goodContainers}
 		err := validateVertex(v)
 		assert.NoError(t, err)
 	})
 
 	t.Run("bad sidecar container name", func(t *testing.T) {
-		v := dfv1.AbstractVertex{Sidecars: badContainers}
+		v := dfv1.AbstractVertex{Name: "my-vertex", Sidecars: badContainers}
 		err := validateVertex(v)
 		assert.Error(t, err)
 		assert.Contains(t, err.Error(), "is reserved for containers created by numaflow")
@@ -368,6 +387,7 @@ func TestValidateVertex(t *testing.T) {
 
 	t.Run("sidecar on source vertex", func(t *testing.T) {
 		v := dfv1.AbstractVertex{
+			Name: "my-vertex",
 			Source: &dfv1.Source{
 				Generator: &dfv1.GeneratorSource{},
 			},