Performance improvements

* Removes one second delay before scheduler starts to check if dependent
  resources can be created. This drastically speeds up unit tests
  and subsequent deployments when resources already exist and thus
  resource is created instantly
* Caches dependencies and definitions during deployment. When one flow
  is consumed from another, they were all fetched from k8s for each
  replica of the inner flow + once for the outer flow. Now it happens
  only once. It also makes deployment more consistent in case if
  definitions were modified in k8s in the middle of deployment
* Check dependency graph for cycles only once per deployment

Author: Stan Lagun

pkg/scheduler/dependency_graph.go

@@ -123,14 +123,21 @@ func (d *sortableDependencyList) Swap(i, j int) {
 	d.Items[i], d.Items[j] = d.Items[j], d.Items[i]
 }
 
-func (sched *scheduler) getDependencies() ([]client.Dependency, error) {
+func (sched *scheduler) getDependencies(silent bool) ([]client.Dependency, error) {
+	if sched.dependencyCache != nil {
+		return sched.dependencyCache, nil
+	}
+	if !silent {
+		log.Println("Getting dependencies")
+	}
 	depList, err := sched.client.Dependencies().List(api.ListOptions{LabelSelector: sched.selector})
 	if err != nil {
 		return nil, err
 	}
 	sortableDepList := sortableDependencyList(*depList)
 	sort.Stable(&sortableDepList)
 
+	sched.dependencyCache = sortableDepList.Items
 	return sortableDepList.Items, nil
 
 }
@@ -184,7 +191,13 @@ func getResourceName(resourceDefinition client.ResourceDefinition) (string, stri
 	return "", ""
 }
 
-func (sched *scheduler) getResourceDefinitions() (map[string]client.ResourceDefinition, error) {
+func (sched *scheduler) getResourceDefinitions(silent bool) (map[string]client.ResourceDefinition, error) {
+	if sched.resDefsCache != nil {
+		return sched.resDefsCache, nil
+	}
+	if !silent {
+		log.Println("Getting resource definitions")
+	}
 	resDefList, err := sched.client.ResourceDefinitions().List(api.ListOptions{LabelSelector: sched.selector})
 	if err != nil {
 		return nil, err
@@ -197,6 +210,7 @@ func (sched *scheduler) getResourceDefinitions() (map[string]client.ResourceDefi
 		}
 		result[kind+"/"+name] = resDef
 	}
+	sched.resDefsCache = result
 	return result, nil
 }
 
@@ -567,10 +581,7 @@ func (sched *scheduler) BuildDependencyGraph(options interfaces.DependencyGraphO
 		options.FlowName = interfaces.DefaultFlowName
 	}
 
-	if !options.Silent {
-		log.Println("Getting resource definitions")
-	}
-	resDefs, err := sched.getResourceDefinitions()
+	resDefs, err := sched.getResourceDefinitions(options.Silent)
 	if err != nil {
 		return nil, err
 	}
@@ -595,19 +606,19 @@ func (sched *scheduler) BuildDependencyGraph(options interfaces.DependencyGraphO
 		return nil, err
 	}
 
-	if !options.Silent {
-		log.Println("Getting dependencies")
-	}
-	depList, err := sched.getDependencies()
+	depList, err := sched.getDependencies(options.Silent)
 	if err != nil {
 		return nil, err
 	}
 
 	if !options.Silent {
 		log.Println("Making sure there is no cycles in the dependency graph")
 	}
-	if err = EnsureNoCycles(depList, resDefs); err != nil {
-		return nil, err
+	if !sched.graphHasNoCycles {
+		if err = EnsureNoCycles(depList, resDefs); err != nil {
+			return nil, err
+		}
+		sched.graphHasNoCycles = true
 	}
 
 	dependencies := groupDependencies(depList, resDefs)

pkg/scheduler/frontend.go

@@ -30,6 +30,10 @@ type scheduler struct {
 	client      client.Interface
 	selector    labels.Selector
 	concurrency int
+
+	resDefsCache     map[string]client.ResourceDefinition
+	dependencyCache  []client.Dependency
+	graphHasNoCycles bool
 }
 
 var _ interfaces.Scheduler = &scheduler{}

pkg/scheduler/scheduler.go

@@ -227,6 +227,9 @@ func createResources(toCreate chan *scheduledResource, finished chan<- *schedule
 					for _, reqKey := range r.requiredBy {
 						req := r.context.graph.graph[reqKey]
 						go func(req *scheduledResource, toCreate chan<- *scheduledResource) {
+							if req.requestCreation(toCreate) {
+								return
+							}
 							ticker := time.NewTicker(CheckInterval)
 							log.Printf("Requesting creation of dependency %v", req.Key())
 							for {