feat: add forest validation for pipelines. Fixes #1002 #1063
Conversation
Signed-off-by: Dillen Padhiar <[email protected]>
Signed-off-by: Dillen Padhiar <[email protected]>
| assert.Error(t, err) | ||
| assert.Contains(t, err.Error(), "invalid pipeline") | ||
| }) | ||
|
|
There was a problem hiding this comment.
looks like some good test cases for invalid pipelines. I wonder if we should add one for an "interesting" valid pipeline - something with multiple sources and multiple sinks?
|
|
||
| // true if forest is found | ||
| return len(visited) != len(pl.Spec.Vertices) | ||
|
|
There was a problem hiding this comment.
might be useful to have a comment for this function to go over how it works
pkg/reconciler/pipeline/validate.go
Outdated
|
|
||
| visited[vtxName] = struct{}{} | ||
|
|
||
| // construct list all to and from vertices |
There was a problem hiding this comment.
"construct list of all"...
pkg/reconciler/pipeline/validate.go
Outdated
| @@ -107,6 +107,10 @@ func ValidatePipeline(pl *dfv1.Pipeline) error { | |||
| return fmt.Errorf("invalid user-defined source vertex %q, only one of 'http', 'kafka', 'nats', 'redisStreams', 'generator' and 'udSource' can be specified", k) | |||
| } | |||
| } | |||
|
|
|||
| if findForests(s, pl) { | |||
There was a problem hiding this comment.
Do we need to findForests for all the sources? Can we define the findForests as below?
func findForests(pl *dfv1.Pipeline) bool
Inside the function, we randomly pick a vertex and DFS. If the pipeline is a forest, then ANY of the vertices cannot visit ALL vertices. Hence if the eventual lengths don't match, the pipeline is a forest.
There was a problem hiding this comment.
Correct. I didn't notice we were doing it for all sources. We should only do it for one vertex.
There was a problem hiding this comment.
In this case, how would we randomly pick the vertex? I'm in favor of just calling it on pl.Spec.Vertices[0] each time as well.
pkg/reconciler/pipeline/validate.go
Outdated
| func findForests(vtx dfv1.AbstractVertex, pl *dfv1.Pipeline) bool { | ||
|
|
||
| visited := map[string]struct{}{} | ||
| findForestHelper(vtx.Name, visited, pl) |
There was a problem hiding this comment.
nit - naming: we have findForests and findForestHelper, do we need the extra s? Also consider naming it isAForest
There was a problem hiding this comment.
This is a better name, will make this change. findForests isn't as descriptive/accurate for what we're doing.
pkg/reconciler/pipeline/validate.go
Outdated
|
|
||
| } | ||
|
|
||
| func findForestHelper(vtxName string, visited map[string]struct{}, pl *dfv1.Pipeline) { |
There was a problem hiding this comment.
| func findForestHelper(vtxName string, visited map[string]struct{}, pl *dfv1.Pipeline) { | |
| func buildVisitedMap(vtxName string, visited map[string]struct{}, pl *dfv1.Pipeline) { |
NIT: how about we make the function name a good representative of what it is doing? It took me a couple of seconds to figure out that visited is mutated even though it is an arg.
There was a problem hiding this comment.
Adding a few comments would be helpful as well, will do.
Signed-off-by: Dillen Padhiar <[email protected]>
Signed-off-by: Dillen Padhiar <[email protected]>
…roj#1063) Signed-off-by: Dillen Padhiar <[email protected]> Signed-off-by: Sidhant Kohli <[email protected]>
…roj#1063) Signed-off-by: Dillen Padhiar <[email protected]> Signed-off-by: Sidhant Kohli <[email protected]>
Explain what this PR does.
Fixes #1002
Adds validation to prevent users from submitting pipeline specs that are forests.