sync: Completely parallelize the initial payload

pkg/cvo/internal/operatorstatus.go

@@ -99,7 +99,11 @@ func (b *clusterOperatorBuilder) Do(ctx context.Context) error {
 	if b.modifier != nil {
 		b.modifier(os)
 	}
-	ctxWithTimeout, cancel := context.WithTimeout(ctx, 1*time.Minute)
+	timeout := 1 * time.Minute
+	if b.mode == resourcebuilder.InitializingMode {
+		timeout = 6 * time.Minute
+	}
+	ctxWithTimeout, cancel := context.WithTimeout(ctx, timeout)
 	defer cancel()
 	return waitForOperatorStatusToBeDone(ctxWithTimeout, 1*time.Second, b.client, os, b.mode)
 }

pkg/cvo/sync_worker.go

@@ -429,17 +429,27 @@ func (w *SyncWorker) apply(ctx context.Context, payloadUpdate *payload.Update, w
 	}
 
 	var tasks []*payload.Task
+	backoff := w.backoff
+	if backoff.Steps > 1 && work.State == payload.InitializingPayload {
+		backoff = wait.Backoff{Steps: 4, Factor: 2, Duration: time.Second}
+	}
 	for i := range payloadUpdate.Manifests {
 		tasks = append(tasks, &payload.Task{
 			Index:    i + 1,
 			Total:    total,
 			Manifest: &payloadUpdate.Manifests[i],
-			Backoff:  w.backoff,
+			Backoff:  backoff,
 		})
 	}
 	graph := payload.NewTaskGraph(tasks)
 	graph.Split(payload.SplitOnJobs)
-	graph.Parallelize(payload.ByNumberAndComponent)
+	if work.State == payload.InitializingPayload {
+		// get the payload out via brute force
+		maxWorkers = len(tasks)
+		graph.Parallelize(payload.FlattenByNumberAndComponent)
+	} else {
+		graph.Parallelize(payload.ByNumberAndComponent)
+	}
 
 	// update each object
 	err := payload.RunGraph(ctx, graph, maxWorkers, func(ctx context.Context, tasks []*payload.Task) error {

pkg/payload/payload_test.go

@@ -48,8 +48,8 @@ func Test_loadUpdatePayload(t *testing.T) {
 				ManifestHash: "6GC9TkkG9PA=",
 				Manifests: []lib.Manifest{
 					{
-						OriginalFilename: "file.json",
-						Raw:              mustRead(filepath.Join("..", "cvo", "testdata", "payloadtest", "release-manifests", "file.json")),
+						OriginalFilename: "0000_10_a_file.json",
+						Raw:              mustRead(filepath.Join("..", "cvo", "testdata", "payloadtest", "release-manifests", "0000_10_a_file.json")),
 						GVK: schema.GroupVersionKind{
 							Kind:    "Test",
 							Version: "v1",
@@ -65,7 +65,7 @@ func Test_loadUpdatePayload(t *testing.T) {
 						},
 					},
 					{
-						OriginalFilename: "file.yaml",
+						OriginalFilename: "0000_10_a_file.yaml",
 						Raw:              []byte(`{"apiVersion":"v1","kind":"Test","metadata":{"name":"file-yaml"}}`),
 						GVK: schema.GroupVersionKind{
 							Kind:    "Test",
@@ -82,7 +82,7 @@ func Test_loadUpdatePayload(t *testing.T) {
 						},
 					},
 					{
-						OriginalFilename: "file.yml",
+						OriginalFilename: "0000_10_a_file.yml",
 						Raw:              []byte(`{"apiVersion":"v1","kind":"Test","metadata":{"name":"file-yml"}}`),
 						GVK: schema.GroupVersionKind{
 							Kind:    "Test",

pkg/payload/task_graph.go

@@ -78,6 +78,54 @@ func ByNumberAndComponent(tasks []*Task) [][]*TaskNode {
 	return buckets
 }
 
+// FlattenByNumberAndComponent creates parallelization for tasks whose original filenames are of the form
+// 0000_NN_NAME_* - files that share 0000_NN_NAME_ are run in serial, but chunks of files that have
+// different 0000_NN_NAME can be run in parallel. This splitter does *not* preserve ordering within run
+// levels and is intended only for use cases where order is not important.
+func FlattenByNumberAndComponent(tasks []*Task) [][]*TaskNode {
+	if len(tasks) <= 1 {
+		return nil
+	}
+	count := len(tasks)
+	matches := make([][]string, 0, count)
+	for i := 0; i < len(tasks); i++ {
+		matches = append(matches, reMatchPattern.FindStringSubmatch(tasks[i].Manifest.OriginalFilename))
+	}
+
+	var lastNode *TaskNode
+	var groups []*TaskNode
+	for i := 0; i < count; {
+		matchBase := matches[i]
+		j := i + 1
+		for ; j < count; j++ {
+			matchNext := matches[j]
+			if matchBase == nil || matchNext == nil || matchBase[groupNumber] != matchNext[groupNumber] {
+				break
+			}
+			if matchBase[groupComponent] != matchNext[groupComponent] {
+				groups = append(groups, &TaskNode{Tasks: tasks[i:j]})
+				i = j
+			}
+			matchBase = matchNext
+		}
+		if len(groups) > 0 {
+			groups = append(groups, &TaskNode{Tasks: tasks[i:j]})
+			i = j
+			lastNode = nil
+			continue
+		}
+		if lastNode == nil {
+			lastNode = &TaskNode{Tasks: append([]*Task(nil), tasks[i:j]...)}
+			i = j
+			groups = append(groups, lastNode)
+			continue
+		}
+		lastNode.Tasks = append(lastNode.Tasks, tasks[i:j]...)
+		i = j
+	}
+	return [][]*TaskNode{groups}
+}
+
 type TaskNode struct {
 	In    []int
 	Tasks []*Task

pkg/payload/task_graph_test.go

@@ -233,6 +233,143 @@ func TestByNumberAndComponent(t *testing.T) {
 	}
 }
 
+func TestFlattenByNumberAndComponent(t *testing.T) {
+	tasks := func(names ...string) []*Task {
+		var arr []*Task
+		for _, name := range names {
+			arr = append(arr, &Task{Manifest: &lib.Manifest{OriginalFilename: name}})
+		}
+		return arr
+	}
+	tests := []struct {
+		name  string
+		tasks []*Task
+		want  [][]*TaskNode
+	}{
+		{
+			name:  "empty tasks",
+			tasks: tasks(),
+			want:  nil,
+		},
+		{
+			name:  "no grouping possible",
+			tasks: tasks("a"),
+			want:  nil,
+		},
+		{
+			name:  "no recognizable groups",
+			tasks: tasks("a", "b", "c"),
+			want: [][]*TaskNode{
+				{
+					&TaskNode{Tasks: tasks("a")},
+					&TaskNode{Tasks: tasks("b")},
+					&TaskNode{Tasks: tasks("c")},
+				},
+			},
+		},
+		{
+			name:  "single grouped item",
+			tasks: tasks("0000_01_x-y-z_file1"),
+			want:  nil,
+		},
+		{
+			name:  "multiple grouped items in single node",
+			tasks: tasks("0000_01_x-y-z_file1", "0000_01_x-y-z_file2"),
+			want: [][]*TaskNode{
+				{
+					&TaskNode{Tasks: tasks("0000_01_x-y-z_file1", "0000_01_x-y-z_file2")},
+				},
+			},
+		},
+		{
+			tasks: tasks("a", "0000_01_x-y-z_file1", "c"),
+			want: [][]*TaskNode{
+				{
+					&TaskNode{Tasks: tasks("a")},
+					&TaskNode{Tasks: tasks("0000_01_x-y-z_file1")},
+					&TaskNode{Tasks: tasks("c")},
+				},
+			},
+		},
+		{
+			tasks: tasks("0000_01_x-y-z_file1", "0000_01_x-y-z_file2"),
+			want: [][]*TaskNode{
+				{
+					&TaskNode{Tasks: tasks("0000_01_x-y-z_file1", "0000_01_x-y-z_file2")},
+				},
+			},
+		},
+		{
+			tasks: tasks("0000_01_a-b-c_file1", "0000_01_x-y-z_file2"),
+			want: [][]*TaskNode{
+				{
+					&TaskNode{Tasks: tasks("0000_01_a-b-c_file1")},
+					&TaskNode{Tasks: tasks("0000_01_x-y-z_file2")},
+				},
+			},
+		},
+		{
+			tasks: tasks(
+				"0000_01_a-b-c_file1",
+				"0000_01_x-y-z_file1",
+				"0000_01_x-y-z_file2",
+				"a",
+				"0000_01_x-y-z_file2",
+				"0000_01_x-y-z_file3",
+			),
+			want: [][]*TaskNode{
+				{
+					&TaskNode{Tasks: tasks(
+						"0000_01_a-b-c_file1",
+					)},
+					&TaskNode{Tasks: tasks(
+						"0000_01_x-y-z_file1",
+						"0000_01_x-y-z_file2",
+					)},
+					&TaskNode{Tasks: tasks(
+						"a",
+					)},
+					&TaskNode{Tasks: tasks(
+						"0000_01_x-y-z_file2",
+						"0000_01_x-y-z_file3",
+					)},
+				},
+			},
+		},
+		{
+			tasks: tasks(
+				"0000_01_a-b-c_file1",
+				"0000_01_x-y-z_file1",
+				"0000_01_x-y-z_file2",
+				"0000_02_x-y-z_file2",
+				"0000_02_x-y-z_file3",
+			),
+			want: [][]*TaskNode{
+				{
+					&TaskNode{Tasks: tasks(
+						"0000_01_a-b-c_file1",
+					)},
+					&TaskNode{Tasks: tasks(
+						"0000_01_x-y-z_file1",
+						"0000_01_x-y-z_file2",
+					)},
+					&TaskNode{Tasks: tasks(
+						"0000_02_x-y-z_file2",
+						"0000_02_x-y-z_file3",
+					)},
+				},
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := FlattenByNumberAndComponent(tt.tasks); !reflect.DeepEqual(got, tt.want) {
+				t.Fatalf("%s", diff.ObjectReflectDiff(tt.want, got))
+			}
+		})
+	}
+}
+
 func Test_TaskGraph_real(t *testing.T) {
 	path := os.Getenv("TEST_GRAPH_PATH")
 	if len(path) == 0 {