Sequential flow replication

For sequential flow, each next replica is attached to the leafs
of previous one so that they will be deployed sequentially

Author: Stan Lagun

examples/etcd/resdefs/scale-flow.yaml

@@ -4,6 +4,8 @@ metadata:
   name: etcd-scale
 
 exported: true
+sequential: true
+
 construction:
   flow: etcd-scale
 destruction:

pkg/client/flows.go

@@ -46,6 +46,9 @@ type Flow struct {
 	// can only be triggered by other flows (including DEFAULT flow which is exported by-default)
 	Exported bool `json:"exported,omitempty"`
 
+	// Flow replicas must be deployed sequentially, one by one
+	Sequential bool `json:"sequential,omitempty"`
+
 	// Parameters that the flow can accept (i.e. valid inputs for the flow)
 	Parameters map[string]FlowParameter `json:"parameters,omitempty"`
 

pkg/scheduler/dependency_graph.go

@@ -661,27 +661,29 @@ func (sched *scheduler) BuildDependencyGraph(options interfaces.DependencyGraphO
 	return depGraph, nil
 }
 
+type interimGraphVertex struct {
+	dependency        client.Dependency
+	scheduledResource *ScheduledResource
+	parentContext     *graphContext
+}
+
 func (sched *scheduler) fillDependencyGraph(rootContext *graphContext,
 	resDefs map[string]client.ResourceDefinition,
 	dependencies map[string][]client.Dependency,
 	flow *client.Flow, replicas []client.Replica, useDestructionSelector bool) error {
 
-	type Block struct {
-		dependency        client.Dependency
-		scheduledResource *ScheduledResource
-		parentContext     *graphContext
-	}
-	blocks := map[string][]*Block{}
+	var vertices [][]interimGraphVertex
 	silent := rootContext.graph.Options().Silent
 
 	for _, replica := range replicas {
+		var replicaVertices []interimGraphVertex
 		replicaName := replica.ReplicaName()
 		replicaContext := sched.prepareContext(rootContext, nil, replicaName)
 		queue := list.New()
-		queue.PushFront(&Block{dependency: client.Dependency{Child: "flow/" + flow.Name}})
+		queue.PushFront(interimGraphVertex{dependency: client.Dependency{Child: "flow/" + flow.Name}})
 
 		for e := queue.Front(); e != nil; e = e.Next() {
-			parent := e.Value.(*Block)
+			parent := e.Value.(interimGraphVertex)
 
 			deps := filterDependencies(dependencies, parent.dependency.Child, flow, useDestructionSelector)
 
@@ -707,44 +709,163 @@ func (sched *scheduler) fillDependencyGraph(rootContext *graphContext,
 				}
 				sr.usedInReplicas = []string{replicaName}
 
-				block := &Block{
+				vertex := interimGraphVertex{
 					scheduledResource: sr,
 					dependency:        dep,
 					parentContext:     parentContext,
 				}
-
-				blocks[dep.Child] = append(blocks[dep.Child], block)
+				replicaVertices = append(replicaVertices, vertex)
 
 				if parent.scheduledResource != nil {
 					sr.Requires = append(sr.Requires, parent.scheduledResource.Key())
 					parent.scheduledResource.RequiredBy = append(parent.scheduledResource.RequiredBy, sr.Key())
 					sr.Meta[parent.dependency.Child] = dep.Meta
 				}
-				queue.PushBack(block)
+				queue.PushBack(vertex)
 			}
 		}
-		for _, block := range blocks {
-			for _, entry := range block {
-				key := entry.scheduledResource.Key()
-				existingSr := rootContext.graph.graph[key]
-				if existingSr == nil {
-					if !silent {
-						log.Printf("Adding resource %s to the dependency graph flow %s", key, flow.Name)
+		vertices = append(vertices, replicaVertices)
+	}
+
+	if flow.Sequential {
+		sched.concatenateReplicas(vertices, rootContext, rootContext.graph.Options())
+	} else {
+		sched.mergeReplicas(vertices, rootContext, rootContext.graph.Options())
+	}
+	return nil
+}
+
+func (sched *scheduler) mergeReplicas(vertices [][]interimGraphVertex, gc *graphContext,
+	options interfaces.DependencyGraphOptions) {
+
+	for _, replicaVertices := range vertices {
+		sched.mergeInterimGraphVertices(replicaVertices, gc.graph.graph, options)
+	}
+}
+
+func (sched *scheduler) concatenateReplicas(vertices [][]interimGraphVertex, gc *graphContext,
+	options interfaces.DependencyGraphOptions) {
+	graph := gc.graph.graph
+	var previousReplicaGraph map[string]*ScheduledResource
+	for i, replicaVertices := range vertices {
+		replicaGraph := map[string]*ScheduledResource{}
+		sched.mergeInterimGraphVertices(replicaVertices, replicaGraph, options)
+
+		if i > 0 {
+			toReplace := map[string]*ScheduledResource{}
+			for key, sr := range replicaGraph {
+				if graph[key] != nil {
+					toReplace[key] = sr
+				}
+			}
+			for key, sr := range toReplace {
+				sched.toVoid(sr, i, graph)
+				for _, rKey := range sr.RequiredBy {
+					requires := replicaGraph[rKey].Requires
+					for i, rKey2 := range requires {
+						if rKey2 == key {
+							requires[i] = sr.Key()
+							break
+						}
 					}
-					rootContext.graph.graph[key] = entry.scheduledResource
-				} else {
-					sched.updateContext(existingSr.context, entry.parentContext, entry.dependency)
-					existingSr.Requires = append(existingSr.Requires, entry.scheduledResource.Requires...)
-					existingSr.RequiredBy = append(existingSr.RequiredBy, entry.scheduledResource.RequiredBy...)
-					existingSr.usedInReplicas = append(existingSr.usedInReplicas, entry.scheduledResource.usedInReplicas...)
-					for metaKey, metaValue := range entry.scheduledResource.Meta {
-						existingSr.Meta[metaKey] = metaValue
+				}
+				for _, rKey := range sr.Requires {
+					requiredBy := replicaGraph[rKey].RequiredBy
+					for i, rKey2 := range requiredBy {
+						if rKey2 == key {
+							requiredBy[i] = sr.Key()
+							break
+						}
 					}
 				}
+				delete(replicaGraph, key)
+				replicaGraph[sr.Key()] = sr
+			}
+
+			syncPointName := fmt.Sprintf("splice-%d", i)
+			syncPoint, _ := sched.newScheduledResource(
+				"void", syncPointName, "", nil, gc, true)
+			syncPointName = "void/" + syncPointName
+			graph[syncPointName] = syncPoint
+
+			for _, leafName := range getLeafs(previousReplicaGraph) {
+				leaf := previousReplicaGraph[leafName]
+				syncPoint.Requires = append(syncPoint.Requires, leafName)
+				leaf.RequiredBy = append(leaf.RequiredBy, syncPointName)
+			}
+			for _, rootName := range getRoots(replicaGraph) {
+				root := replicaGraph[rootName]
+				syncPoint.RequiredBy = append(syncPoint.RequiredBy, rootName)
+				root.Requires = append(root.Requires, syncPointName)
+			}
+		}
+		previousReplicaGraph = replicaGraph
+
+		for key, value := range replicaGraph {
+			graph[key] = value
+		}
+
+	}
+}
+
+func getRoots(graph map[string]*ScheduledResource) []string {
+	var result []string
+	for key, sr := range graph {
+		if len(sr.Requires) == 0 {
+			result = append(result, key)
+		}
+	}
+	return result
+}
+
+func getLeafs(graph map[string]*ScheduledResource) []string {
+	var result []string
+	for key, sr := range graph {
+		if len(sr.RequiredBy) == 0 {
+			result = append(result, key)
+		}
+	}
+	return result
+}
+
+func (sched *scheduler) toVoid(sr *ScheduledResource, index int, graph map[string]*ScheduledResource) {
+	kind, name, _, _ := keyParts(sr.Key())
+	var newName string
+	for i := 1; ; i++ {
+		newName = fmt.Sprintf("%s-%s-%d", kind, name, index+1)
+		if i > 1 {
+			newName = newName + fmt.Sprintf("_%d", i)
+		}
+		if graph["void/"+newName] == nil {
+			break
+		}
+	}
+	resourceTemplate, _ := resources.KindToResourceTemplate["void"]
+	res, _, _ := sched.newResource("void", newName, nil, sr.context, resourceTemplate, true)
+	sr.Resource = res
+}
+
+func (sched *scheduler) mergeInterimGraphVertices(vertices []interimGraphVertex, graph map[string]*ScheduledResource,
+	options interfaces.DependencyGraphOptions) {
+
+	for _, entry := range vertices {
+		key := entry.scheduledResource.Key()
+		existingSr := graph[key]
+		if existingSr == nil {
+			if !options.Silent {
+				log.Printf("Adding resource %s to the dependency graph flow %s", key, options.FlowName)
+			}
+			graph[key] = entry.scheduledResource
+		} else {
+			sched.updateContext(existingSr.context, entry.parentContext, entry.dependency)
+			existingSr.Requires = append(existingSr.Requires, entry.scheduledResource.Requires...)
+			existingSr.RequiredBy = append(existingSr.RequiredBy, entry.scheduledResource.RequiredBy...)
+			existingSr.usedInReplicas = append(existingSr.usedInReplicas, entry.scheduledResource.usedInReplicas...)
+			for metaKey, metaValue := range entry.scheduledResource.Meta {
+				existingSr.Meta[metaKey] = metaValue
 			}
 		}
 	}
-	return nil
 }
 
 // getResourceDestructors builds a list of functions, each of them delete one of replica resources

pkg/scheduler/flows_test.go

@@ -1367,3 +1367,56 @@ func TestSyncOnVoidResource(t *testing.T) {
 	depGraph.Deploy(stopChan)
 	ensureReplicas(c, t, replicaCount, replicaCount)
 }
+
+// TestSequentialReplication tests that resources of sequentially replicated flows create in right order
+func TestSequentialReplication(t *testing.T) {
+	replicaCount := 3
+	flow := mocks.MakeFlow("test")
+	flow.Flow.Sequential = true
+
+	c, fake := mocks.NewClientWithFake(
+		flow,
+		mocks.MakeResourceDefinition("pod/ready-$AC_NAME"),
+		mocks.MakeResourceDefinition("secret/secret"),
+		mocks.MakeResourceDefinition("job/ready-$AC_NAME"),
+		mocks.MakeDependency("flow/test", "pod/ready-$AC_NAME", "flow=test"),
+		mocks.MakeDependency("pod/ready-$AC_NAME", "secret/secret", "flow=test"),
+		mocks.MakeDependency("secret/secret", "job/ready-$AC_NAME", "flow=test"),
+	)
+
+	stopChan := make(chan struct{})
+	var deployed []string
+	fake.PrependReactor("create", "*",
+		func(action k8stesting.Action) (handled bool, ret runtime.Object, err error) {
+			resource := action.GetResource().Resource
+			if resource != "replica" {
+				deployed = append(deployed, resource)
+			}
+
+			return false, nil, nil
+		})
+
+	depGraph, err := New(c, nil, 0).BuildDependencyGraph(
+		interfaces.DependencyGraphOptions{ReplicaCount: replicaCount, FlowName: "test"})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	graph := depGraph.(*dependencyGraph).graph
+	if len(graph) != 3*replicaCount+replicaCount-1 {
+		t.Error("wrong dependency graph length")
+	}
+
+	depGraph.Deploy(stopChan)
+	expected := []string{"pods", "secrets", "jobs", "pods", "jobs", "pods", "jobs"}
+	if len(deployed) != len(expected) {
+		t.Fatal("invalid resource sequence", deployed)
+	}
+	for i, r := range deployed {
+		if expected[i] != r {
+			t.Fatal("invalid resource sequence")
+		}
+	}
+
+	ensureReplicas(c, t, replicaCount, replicaCount)
+}