Dependency replication

This change adds ability to replicate dependency with index parameters
iterated over arbitrary number of lists.

For each dependency it is now possible to specify map of
indexVariableName -> listExpression

listExpression := range|item + [, listExpression]
range := number '..' number
item := STRING

for example, if for "i: 1..3" the dependency will be replicated into 3
clones, each one of them having argument i set to value in range [1, 3]

This also allows to consume N flow replicas by replicating the
dependency that leads to the consumed flow

Author: Stan Lagun

pkg/client/dependencies.go

@@ -41,6 +41,9 @@ type Dependency struct {
 
 	// Arguments passed to dependent resource
 	Args map[string]string `json:"args,omitempty"`
+
+	// map of variable name -> list expression. New dependencies are generated by replication and iteration over those lists
+	GenerateFor map[string]string `json:"generateFor,omitempty"`
 }
 
 // DependencyList is a k8s object representing list of dependencies

pkg/interfaces/interfaces.go

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

pkg/resources/flow.go

@@ -15,9 +15,7 @@
 package resources
 
 import (
-	"fmt"
 	"log"
-	"strings"
 
 	"github.com/Mirantis/k8s-AppController/pkg/client"
 	"github.com/Mirantis/k8s-AppController/pkg/interfaces"
@@ -52,19 +50,13 @@ 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)
 
-	dep := gc.Dependency()
-	var depName string
-	if dep != nil {
-		depName = strings.Replace(dep.Name, dep.GenerateName, "", 1)
-	}
-
 	return report.SimpleReporter{
 		BaseResource: &flow{
 			Base:         Base{def.Meta},
 			flow:         newFlow,
 			context:      gc,
 			originalName: def.Flow.Name,
-			instanceName: fmt.Sprintf("%s%s", depName, gc.GetArg("AC_NAME")),
+			instanceName: gc.GetArg("AC_ID"),
 		}}
 }
 

pkg/scheduler/dependency_graph.go

@@ -19,6 +19,7 @@ import (
 	"fmt"
 	"log"
 	"sort"
+	"strconv"
 	"strings"
 	"time"
 
@@ -40,12 +41,12 @@ type dependencyGraph struct {
 }
 
 type graphContext struct {
-	args       map[string]string
-	graph      *dependencyGraph
-	scheduler  *scheduler
-	flow       *client.Flow
-	dependency *client.Dependency
-	replica    string
+	args      map[string]string
+	graph     *dependencyGraph
+	scheduler *scheduler
+	flow      *client.Flow
+	id        string
+	replica   string
 }
 
 var _ interfaces.GraphContext = &graphContext{}
@@ -62,6 +63,8 @@ func (gc graphContext) GetArg(name string) string {
 		return gc.replica
 	case "AC_FLOW_NAME":
 		return gc.flow.Name
+	case "AC_ID":
+		return gc.id
 	default:
 		val, ok := gc.args[name]
 		if ok {
@@ -84,11 +87,6 @@ func (gc graphContext) Graph() interfaces.DependencyGraph {
 	return gc.graph
 }
 
-// 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, suffix string, context *graphContext, existing bool) *ScheduledResource {
 	return &ScheduledResource{
@@ -159,7 +157,9 @@ func groupDependencies(dependencies []client.Dependency,
 		defaultFlow = []client.Dependency{}
 		addResource := func(name string) {
 			if !strings.HasPrefix(name, "flow/") && !isDependant[name] {
-				defaultFlow = append(defaultFlow, client.Dependency{Parent: defaultFlowName, Child: name})
+				dep := client.Dependency{Parent: defaultFlowName, Child: name}
+				dep.Name = name
+				defaultFlow = append(defaultFlow, dep)
 				isDependant[name] = true
 			}
 		}
@@ -328,11 +328,11 @@ func getArgFunc(gc interfaces.GraphContext) func(string) string {
 
 func (sched *scheduler) prepareContext(parentContext *graphContext, dependency *client.Dependency, replica string) *graphContext {
 	context := &graphContext{
-		scheduler:  sched,
-		graph:      parentContext.graph,
-		flow:       parentContext.flow,
-		replica:    replica,
-		dependency: dependency,
+		scheduler: sched,
+		graph:     parentContext.graph,
+		flow:      parentContext.flow,
+		replica:   replica,
+		id:        getVertexID(dependency, replica),
 	}
 
 	context.args = make(map[string]string)
@@ -344,6 +344,15 @@ func (sched *scheduler) prepareContext(parentContext *graphContext, dependency *
 	return context
 }
 
+func getVertexID(dependency *client.Dependency, replica string) string {
+	var depName string
+	if dependency != nil {
+		depName = strings.Replace(dependency.Name, dependency.GenerateName, "", 1)
+	}
+	depName += replica
+	return depName
+}
+
 func (sched *scheduler) updateContext(context, parentContext *graphContext, dependency client.Dependency) {
 	for key, value := range dependency.Args {
 		context.args[key] = copier.EvaluateString(value, parentContext.GetArg)
@@ -661,6 +670,110 @@ func (sched *scheduler) BuildDependencyGraph(options interfaces.DependencyGraphO
 	return depGraph, nil
 }
 
+func listDependencies(dependencies map[string][]client.Dependency, parent string, flow *client.Flow,
+	useDestructionSelector bool) []client.Dependency {
+
+	deps := filterDependencies(dependencies, parent, flow, useDestructionSelector)
+	var result []client.Dependency
+	for _, dep := range deps {
+		if len(dep.GenerateFor) == 0 {
+			result = append(result, dep)
+			continue
+		}
+
+		var keys []string
+		for k := range dep.GenerateFor {
+			keys = append(keys, k)
+		}
+		sort.Strings(keys)
+		lists := make([][]string, len(dep.GenerateFor))
+		for i, key := range keys {
+			lists[i] = expandListExpression(dep.GenerateFor[key])
+		}
+		for n, combination := range permute(lists) {
+			newArgs := make(map[string]string, len(dep.Args)+len(keys))
+			for k, v := range dep.Args {
+				newArgs[k] = v
+			}
+			for i, key := range keys {
+				newArgs[key] = combination[i]
+			}
+			depCopy := dep
+			depCopy.Args = newArgs
+			depCopy.Name += strconv.Itoa(n + 1)
+			result = append(result, depCopy)
+		}
+	}
+	return result
+}
+
+func permute(variants [][]string) [][]string {
+	switch len(variants) {
+	case 0:
+		return variants
+	case 1:
+		var result [][]string
+		for _, v := range variants[0] {
+			result = append(result, []string{v})
+		}
+		return result
+	default:
+		var result [][]string
+		for _, tail := range variants[len(variants)-1] {
+			for _, p := range permute(variants[:len(variants)-1]) {
+				result = append(result, append(p, tail))
+			}
+		}
+		return result
+	}
+}
+
+func expandListExpression(expr string) []string {
+	var result []string
+
+	for _, part := range strings.Split(expr, ",") {
+		part = strings.TrimSpace(part)
+
+		isRange := true
+		var from, to int
+
+		rangeParts := strings.SplitN(part, "..", 2)
+		if len(rangeParts) != 2 {
+			isRange = false
+		}
+
+		var err error
+		if isRange {
+			from, err = strconv.Atoi(rangeParts[0])
+			if err != nil {
+				isRange = false
+			}
+		}
+		if isRange {
+			to, err = strconv.Atoi(rangeParts[1])
+			if err != nil {
+				isRange = false
+			}
+		}
+
+		if isRange {
+			step := 1
+			if to < from {
+				step = -1
+			}
+			for i := from; ; i += step {
+				result = append(result, strconv.Itoa(i))
+				if i == to {
+					break
+				}
+			}
+		} else {
+			result = append(result, part)
+		}
+	}
+	return result
+}
+
 func (sched *scheduler) fillDependencyGraph(rootContext *graphContext,
 	resDefs map[string]client.ResourceDefinition,
 	dependencies map[string][]client.Dependency,
@@ -683,7 +796,7 @@ func (sched *scheduler) fillDependencyGraph(rootContext *graphContext,
 		for e := queue.Front(); e != nil; e = e.Next() {
 			parent := e.Value.(*Block)
 
-			deps := filterDependencies(dependencies, parent.dependency.Child, flow, useDestructionSelector)
+			deps := listDependencies(dependencies, parent.dependency.Child, flow, useDestructionSelector)
 
 			for _, dep := range deps {
 				if parent.scheduledResource != nil && strings.HasPrefix(parent.scheduledResource.Key(), "flow/") {

pkg/scheduler/dependency_graph_test.go

@@ -15,6 +15,7 @@
 package scheduler
 
 import (
+	"strings"
 	"testing"
 
 	"github.com/Mirantis/k8s-AppController/pkg/client"
@@ -319,3 +320,72 @@ func TestDependencyToFlowMatching(t *testing.T) {
 		}
 	}
 }
+
+// TestPermute tests permute function
+func TestPermute(t *testing.T) {
+	alphabets := [][]string{
+		{"1", "2", "3"},
+		{"+", "-"},
+		{"a", "b"},
+		{"="},
+	}
+
+	expected := map[string]bool{
+		"1+a=": true,
+		"1+b=": true,
+		"1-a=": true,
+		"1-b=": true,
+		"2+a=": true,
+		"2+b=": true,
+		"2-a=": true,
+		"2-b=": true,
+		"3+a=": true,
+		"3+b=": true,
+		"3-a=": true,
+		"3-b=": true,
+	}
+	permutations := permute(alphabets)
+	for _, combination := range permutations {
+		combinationStr := strings.Join(combination, "")
+		if !expected[combinationStr] {
+			t.Errorf("unexpected combination %s", combinationStr)
+		} else {
+			delete(expected, combinationStr)
+		}
+	}
+	if len(expected) != 0 {
+		t.Error("not all combinations were generated")
+	}
+
+	alphabets = append(alphabets, make([]string, 0))
+	if len(permute(alphabets)) != 0 {
+		t.Error("empty alphabet didin't result in empty permutation list")
+	}
+}
+
+// TestExpendListExpression tests list expression translation to list of strings
+func TestExpendListExpression(t *testing.T) {
+	table := map[string][]string{
+		"1":              {"1"},
+		"1..5":           {"1", "2", "3", "4", "5"},
+		"2..-1":          {"2", "1", "0", "-1"},
+		"a, b":           {"a", "b"},
+		"a, b, 2..4":     {"a", "b", "2", "3", "4"},
+		"-1..1, 2..4, x": {"-1", "0", "1", "2", "3", "4", "x"},
+		"a..b":           {"a..b"},
+		"a..b, 1..3":     {"a..b", "1", "2", "3"},
+		"a..b, c..d":     {"a..b", "c..d"},
+	}
+	for expr, expected := range table {
+		result := expandListExpression(expr)
+		if len(result) != len(expected) {
+			t.Errorf("unexpected result length for expression %s: %d != %d", expr, len(result), len(expected))
+		} else {
+			for i := range expected {
+				if expected[i] != result[i] {
+					t.Errorf("invalid entry %d for expression %s: %s != %s", i, expr, expected[i], result[i])
+				}
+			}
+		}
+	}
+}