Ability to parametrize resource key in dependency

Currently, flow resource name is auto-generated from replica name of the
outer (consuming) graph and dependency name, whose child is the flow
resource. As a result, if we want a flow call to depend on two parent
resources, we will get two flow calls after each parent becomes ready,
rather than one call after both turn ready.

If we make flow name generation depend on replica name only, the problem
above will be solved, however it will still not be possible to have one
flow call for entire graph rather than for each replica. Also in some
cases we might want it to be called for each dependency, if there are
different arguments passed along each of them.

If we remove name generation at all, the only way, how we'll be able to
achieve behavior other, than one call per graph will be to make the
flow name by dynamic, i.e. include things like $AC_NAME or $arg1.
Though technically it solves the problem, it has its own major cons:
* Names of the flows will become ugly and flow name is a user facing
  thing because it will propagate into CLI commandlines
* Flows were supposed to be reusable. Flow author might not know
  how it will be used and not include needed variable in flow name
* Even if he did, one flow may still be used in different ways.
  For example, what if the flow name is "flow-$AC_NAME" and I want
  to call it just once for all replicas of my flow. There is no way
  to pass $AC_NAME explicitly to the flow to make it be a fixed value.

This commit proposes another solution to the problem. There is no need
to use variables in flow names. Instead, it can be parametrized on the
caller side with new dependency syntax. Now I can add optional suffix
(3rd component of the name) to parent/child names in dependencies.
For example: `child: flow/my-flow/$AC_NAME`

Resource definition is looked up by first 2 components (flow/my-flow),
but resources from different replicas are merged only if the whole
3-component key is the same.

So we get:
* Ability to call flows by simple names
* It is now possible to call flow several times within another flow.
  Without this change it is impossible, because node names are unique
  in the graph and if we include several flow calls, we will get cycle
  in the graph. However, with this change node identity is identified
  by 3-part string, while resource definition by just two, so we can
  have several usage of the same resource that differ by 3-rd component
  only
* Similar to flows, it becomes possible to include the same resource
  more than once in a graph. For example it will become possible to
  create resource, do something and then update the same resource. Or
  delete it.

Author: Stan Lagun

e2e/flows_test.go

@@ -16,6 +16,7 @@ package integration
 
 import (
 	"strings"
+	"time"
 
 	testutils "github.com/Mirantis/k8s-AppController/e2e/utils"
 	"github.com/Mirantis/k8s-AppController/pkg/interfaces"
@@ -33,56 +34,56 @@ var _ = Describe("Flows Suite", func() {
 		framework.CreateRunAndVerify("flows", interfaces.DependencyGraphOptions{MinReplicaCount: 1})
 		Eventually(func() int {
 			return framework.countJobs("a-job-", false)
-		}).Should(Equal(1), "1 a-job* should have been created")
+		}, 300*time.Second, 5*time.Second).Should(Equal(1), "1 a-job* should have been created")
 		Eventually(func() int {
 			return framework.countJobs("b-job-", false)
-		}).Should(Equal(1), "1 a-job* should have been created")
+		}, 300*time.Second, 5*time.Second).Should(Equal(1), "1 a-job* should have been created")
 		Eventually(func() int {
 			return framework.countJobs("test-job", true)
-		}).Should(Equal(1), "1 test-job should have been created")
+		}, 300*time.Second, 5*time.Second).Should(Equal(1), "1 test-job should have been created")
 		Eventually(func() int {
 			return framework.countPods("a-pod-", false)
-		}).Should(Equal(1), "1 a-pod* should have been created")
+		}, 300*time.Second, 5*time.Second).Should(Equal(1), "1 a-pod* should have been created")
 		Eventually(func() int {
 			return framework.countPods("b-pod-", false)
-		}).Should(Equal(1), "1 a-pod* should have been created")
+		}, 300*time.Second, 5*time.Second).Should(Equal(1), "1 a-pod* should have been created")
 		Eventually(func() int {
 			return framework.countPods("test-pod", true)
-		}).Should(Equal(1), "1 test-pod should have been created")
+		}, 300*time.Second, 5*time.Second).Should(Equal(1), "1 test-pod should have been created")
 		Eventually(func() int {
 			return framework.countReplicas("test-flow", true)
-		}).Should(Equal(2), "2 test-flow replicas should have been created")
+		}, 300*time.Second, 5*time.Second).Should(Equal(2), "2 test-flow replicas should have been created")
 		Eventually(func() int {
 			return framework.countReplicas("DEFAULT", true)
-		}).Should(Equal(1), "1 DEFAULT flow replica should have been created")
+		}, 300*time.Second, 5*time.Second).Should(Equal(1), "1 DEFAULT flow replica should have been created")
 	})
 
 	It("Example 'flows' with replication should finish and create two replicas", func() {
 		framework.CreateRunAndVerify("flows", interfaces.DependencyGraphOptions{ReplicaCount: 2})
 		Eventually(func() int {
 			return framework.countJobs("a-job-", false)
-		}).Should(Equal(2), "1 a-job* should have been created")
+		}, 300*time.Second, 5*time.Second).Should(Equal(2), "1 a-job* should have been created")
 		Eventually(func() int {
 			return framework.countJobs("b-job-", false)
-		}).Should(Equal(2), "1 a-job* should have been created")
+		}, 300*time.Second, 5*time.Second).Should(Equal(2), "1 a-job* should have been created")
 		Eventually(func() int {
 			return framework.countJobs("test-job", true)
-		}).Should(Equal(1), "1 test-job should have been created")
+		}, 300*time.Second, 5*time.Second).Should(Equal(1), "1 test-job should have been created")
 		Eventually(func() int {
 			return framework.countPods("a-pod-", false)
-		}).Should(Equal(2), "1 a-pod* should have been created")
+		}, 300*time.Second, 5*time.Second).Should(Equal(2), "1 a-pod* should have been created")
 		Eventually(func() int {
 			return framework.countPods("b-pod-", false)
-		}).Should(Equal(2), "1 a-pod* should have been created")
+		}, 300*time.Second, 5*time.Second).Should(Equal(2), "1 a-pod* should have been created")
 		Eventually(func() int {
 			return framework.countPods("test-pod", true)
-		}).Should(Equal(1), "1 test-pod should have been created")
+		}, 300*time.Second, 5*time.Second).Should(Equal(1), "1 test-pod should have been created")
 		Eventually(func() int {
 			return framework.countReplicas("test-flow", true)
-		}).Should(Equal(4), "2 test-flow replicas should have been created")
+		}, 300*time.Second, 5*time.Second).Should(Equal(4), "2 test-flow replicas should have been created")
 		Eventually(func() int {
 			return framework.countReplicas("DEFAULT", true)
-		}).Should(Equal(2), "1 DEFAULT flow replica should have been created")
+		}, 300*time.Second, 5*time.Second).Should(Equal(2), "1 DEFAULT flow replica should have been created")
 	})
 
 	It("Example 'flows' should cleanup after itself", func() {
@@ -96,13 +97,13 @@ var _ = Describe("Flows Suite", func() {
 
 		Eventually(func() int {
 			return framework.countReplicas("", false)
-		}).Should(Equal(0), "0 replicas should remain")
+		}, 300*time.Second, 5*time.Second).Should(Equal(0), "0 replicas should remain")
 		Eventually(func() int {
 			return framework.countJobs("", false)
-		}).Should(Equal(0), "0 jobs should remain")
+		}, 300*time.Second, 5*time.Second).Should(Equal(0), "0 jobs should remain")
 		Eventually(func() int {
 			return framework.countPods("", false)
-		}).Should(Equal(1), "only AC pod should remain")
+		}, 300*time.Second, 5*time.Second).Should(Equal(1), "only AC pod should remain")
 	})
 })
 

examples/flows/create.sh

@@ -15,6 +15,6 @@ cat job2.yaml | $KUBECTL_NAME exec -i k8s-appcontroller kubeac wrap | $KUBECTL_N
 cat pod.yaml | $KUBECTL_NAME exec -i k8s-appcontroller kubeac wrap | $KUBECTL_NAME create -f -
 cat pod2.yaml | $KUBECTL_NAME exec -i k8s-appcontroller kubeac wrap | $KUBECTL_NAME create -f -
 
-$KUBECTL_NAME exec k8s-appcontroller run
+$KUBECTL_NAME exec k8s-appcontroller kubeac run
 
 $KUBECTL_NAME logs -f k8s-appcontroller

examples/flows/deps.yaml

@@ -3,7 +3,7 @@ kind: Dependency
 metadata:
   generateName: dependency-
 parent: flow/DEFAULT
-child: flow/test-flow
+child: flow/test-flow/$prefix$AC_NAME
 args:
   prefix: a
 ---
@@ -19,17 +19,18 @@ kind: Dependency
 metadata:
   generateName: dependency-
 parent: pod/test-pod
-child: flow/test-flow
+child: flow/test-flow/$prefix$AC_NAME
 args:
   prefix: b
 ---
 apiVersion: appcontroller.k8s/v1alpha1
 kind: Dependency
 metadata:
   generateName: dependency-
-parent: flow/test-flow
+parent: flow/test-flow/$prefix$AC_NAME
 child: job/test-job
 ---
+
 apiVersion: appcontroller.k8s/v1alpha1
 kind: Dependency
 metadata:

pkg/interfaces/interfaces.go

@@ -82,7 +82,7 @@ type GraphContext interface {
 	Scheduler() Scheduler
 	GetArg(string) string
 	Graph() DependencyGraph
-	Dependencies() []client.Dependency
+	Dependency() *client.Dependency
 }
 
 // DependencyGraphOptions contains all the input required to build a dependency graph

pkg/resources/flow.go

@@ -26,11 +26,11 @@ import (
 
 type flow struct {
 	Base
-	flow          *client.Flow
-	context       interfaces.GraphContext
-	generatedName string
-	originalName  string
-	currentGraph  interfaces.DependencyGraph
+	flow         *client.Flow
+	context      interfaces.GraphContext
+	originalName string
+	instanceName string
+	currentGraph interfaces.DependencyGraph
 }
 
 type flowTemplateFactory struct{}
@@ -52,19 +52,19 @@ func (flowTemplateFactory) Kind() string {
 func (flowTemplateFactory) New(def client.ResourceDefinition, c client.Interface, gc interfaces.GraphContext) interfaces.Resource {
 	newFlow := parametrizeResource(def.Flow, gc, []string{"*"}).(*client.Flow)
 
-	deps := gc.Dependencies()
+	dep := gc.Dependency()
 	var depName string
-	if len(deps) > 0 {
-		depName = strings.Replace(deps[0].Name, deps[0].GenerateName, "", 1)
+	if dep != nil {
+		depName = strings.Replace(dep.Name, dep.GenerateName, "", 1)
 	}
 
 	return report.SimpleReporter{
 		BaseResource: &flow{
-			Base:          Base{def.Meta},
-			flow:          newFlow,
-			context:       gc,
-			generatedName: fmt.Sprintf("%s-%s%s", newFlow.Name, depName, gc.GetArg("AC_NAME")),
-			originalName:  def.Flow.Name,
+			Base:         Base{def.Meta},
+			flow:         newFlow,
+			context:      gc,
+			originalName: def.Flow.Name,
+			instanceName: fmt.Sprintf("%s%s", depName, gc.GetArg("AC_NAME")),
 		}}
 }
 
@@ -77,7 +77,7 @@ func (flowTemplateFactory) NewExisting(name string, c client.Interface, gc inter
 
 // Key return Flow identifier
 func (f flow) Key() string {
-	return "flow/" + f.generatedName
+	return "flow/" + f.flow.Name
 }
 
 func (f *flow) buildDependencyGraph(replicaCount int, silent bool) (interfaces.DependencyGraph, error) {
@@ -95,7 +95,7 @@ func (f *flow) buildDependencyGraph(replicaCount int, silent bool) (interfaces.D
 	options := interfaces.DependencyGraphOptions{
 		FlowName:              f.originalName,
 		Args:                  args,
-		FlowInstanceName:      f.generatedName,
+		FlowInstanceName:      f.instanceName,
 		ReplicaCount:          replicaCount,
 		Silent:                silent,
 		FixedNumberOfReplicas: fixedNumberOfReplicas,

pkg/scheduler/dependency_graph.go

@@ -40,12 +40,12 @@ type dependencyGraph struct {
 }
 
 type graphContext struct {
-	args         map[string]string
-	graph        *dependencyGraph
-	scheduler    *scheduler
-	flow         *client.Flow
-	replica      string
-	dependencies []client.Dependency
+	args       map[string]string
+	graph      *dependencyGraph
+	scheduler  *scheduler
+	flow       *client.Flow
+	dependency *client.Dependency
+	replica    string
 }
 
 var _ interfaces.GraphContext = &graphContext{}
@@ -84,13 +84,13 @@ func (gc graphContext) Graph() interfaces.DependencyGraph {
 	return gc.graph
 }
 
-// Dependencies method returns list of incoming dependencies for the node on which operation is performed
-func (gc graphContext) Dependencies() []client.Dependency {
-	return gc.dependencies
+// Dependency returns Dependency for which child is the resource being created with this context
+func (gc graphContext) Dependency() *client.Dependency {
+	return gc.dependency
 }
 
 // newScheduledResourceFor returns new scheduled resource for given resource in init state
-func newScheduledResourceFor(r interfaces.Resource, context *graphContext, existing bool) *ScheduledResource {
+func newScheduledResourceFor(r interfaces.Resource, suffix string, context *graphContext, existing bool) *ScheduledResource {
 	return &ScheduledResource{
 		Started:  false,
 		Ignored:  false,
@@ -99,6 +99,7 @@ func newScheduledResourceFor(r interfaces.Resource, context *graphContext, exist
 		Meta:     map[string]map[string]string{},
 		context:  context,
 		Existing: existing,
+		suffix:   copier.EvaluateString(suffix, getArgFunc(context)),
 	}
 }
 
@@ -252,7 +253,7 @@ func isMapContainedIn(contained, containing map[string]string) bool {
 }
 
 // newScheduledResource is a constructor for ScheduledResource
-func (sched scheduler) newScheduledResource(kind, name string, resDefs map[string]client.ResourceDefinition,
+func (sched scheduler) newScheduledResource(kind, name, suffix string, resDefs map[string]client.ResourceDefinition,
 	gc *graphContext, silent bool) (*ScheduledResource, error) {
 	var r interfaces.Resource
 
@@ -266,7 +267,7 @@ func (sched scheduler) newScheduledResource(kind, name string, resDefs map[strin
 		return nil, err
 	}
 
-	return newScheduledResourceFor(r, gc, existing), nil
+	return newScheduledResourceFor(r, suffix, gc, existing), nil
 }
 
 // newResource returns creates a resource controller for a given resources name and factory.
@@ -294,14 +295,17 @@ func (sched scheduler) newResource(kind, name string, resDefs map[string]client.
 	return r, true, nil
 }
 
-func keyParts(key string) (kind, name string, err error) {
-	parts := strings.Split(key, "/")
+func keyParts(key string) (kind, name, suffix string, err error) {
+	parts := strings.SplitN(key, "/", 31)
 
 	if len(parts) < 2 {
-		return "", "", fmt.Errorf("not a proper resource key: %s. Expected KIND/NAME", key)
+		return "", "", "", fmt.Errorf("not a proper resource key: %s. Expected KIND/NAME", key)
+	}
+	if len(parts) == 3 {
+		suffix = parts[2]
 	}
 
-	return parts[0], parts[1], nil
+	return parts[0], parts[1], suffix, nil
 }
 
 func newDependencyGraph(sched *scheduler, options interfaces.DependencyGraphOptions) *dependencyGraph {
@@ -322,17 +326,17 @@ func getArgFunc(gc interfaces.GraphContext) func(string) string {
 	}
 }
 
-func (sched *scheduler) prepareContext(parentContext *graphContext, dependencies []client.Dependency, replica string) *graphContext {
+func (sched *scheduler) prepareContext(parentContext *graphContext, dependency *client.Dependency, replica string) *graphContext {
 	context := &graphContext{
-		scheduler:    sched,
-		graph:        parentContext.graph,
-		flow:         parentContext.flow,
-		replica:      replica,
-		dependencies: dependencies,
+		scheduler:  sched,
+		graph:      parentContext.graph,
+		flow:       parentContext.flow,
+		replica:    replica,
+		dependency: dependency,
 	}
 
 	context.args = make(map[string]string)
-	for _, dependency := range dependencies {
+	if dependency != nil {
 		for key, value := range dependency.Args {
 			context.args[key] = copier.EvaluateString(value, getArgFunc(parentContext))
 		}
@@ -344,7 +348,6 @@ func (sched *scheduler) updateContext(context, parentContext *graphContext, depe
 	for key, value := range dependency.Args {
 		context.args[key] = copier.EvaluateString(value, parentContext.GetArg)
 	}
-	context.dependencies = append(context.dependencies, dependency)
 }
 
 func newDefaultFlowObject() *client.Flow {
@@ -695,10 +698,10 @@ func (sched *scheduler) fillDependencyGraph(rootContext *graphContext,
 					parentContext = parent.scheduledResource.context
 				}
 
-				kind, name, err := keyParts(dep.Child)
+				kind, name, suffix, err := keyParts(dep.Child)
 
-				context := sched.prepareContext(parentContext, []client.Dependency{dep}, replicaName)
-				sr, err := sched.newScheduledResource(kind, name, resDefs, context, silent)
+				context := sched.prepareContext(parentContext, &dep, replicaName)
+				sr, err := sched.newScheduledResource(kind, name, suffix, resDefs, context, silent)
 				if err != nil {
 					return err
 				}