Limit concurrency of asset copy tasks

cmd/kops/get_assets.go

@@ -1,5 +1,5 @@
 /*
-Copyright 2020 The Kubernetes Authors.
+Copyright 2021 The Kubernetes Authors.
 
 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.
@@ -22,8 +22,7 @@ import (
 	"fmt"
 	"io"
 
-	"k8s.io/kops/upup/pkg/fi"
-	"k8s.io/kops/upup/pkg/fi/assettasks"
+	"k8s.io/kops/pkg/assets"
 	"k8s.io/kubectl/pkg/util/i18n"
 	"k8s.io/kubectl/pkg/util/templates"
 	"sigs.k8s.io/yaml"
@@ -58,9 +57,6 @@ type AssetResult struct {
 	Files []*File `json:"files,omitempty"`
 }
 
-type copyAssetsTarget struct {
-}
-
 func NewCmdGetAssets(f *util.Factory, out io.Writer, getOptions *GetOptions) *cobra.Command {
 	options := GetAssetsOptions{
 		GetOptions: getOptions,
@@ -120,7 +116,6 @@ func RunGetAssets(ctx context.Context, f *util.Factory, out io.Writer, options *
 		Images: make([]*Image, 0, len(updateClusterResults.ImageAssets)),
 		Files:  make([]*File, 0, len(updateClusterResults.FileAssets)),
 	}
-	tasks := map[string]fi.Task{}
 
 	seen := map[string]bool{}
 	for _, imageAsset := range updateClusterResults.ImageAssets {
@@ -132,24 +127,6 @@ func RunGetAssets(ctx context.Context, f *util.Factory, out io.Writer, options *
 			result.Images = append(result.Images, &image)
 			seen[image.Canonical] = true
 		}
-
-		if options.Copy && imageAsset.DownloadLocation != imageAsset.CanonicalLocation {
-			ctx := &fi.ModelBuilderContext{
-				Tasks: tasks,
-			}
-
-			copyImageTask := &assettasks.CopyImage{
-				Name:        fi.String(imageAsset.DownloadLocation),
-				SourceImage: fi.String(imageAsset.CanonicalLocation),
-				TargetImage: fi.String(imageAsset.DownloadLocation),
-				Lifecycle:   fi.LifecycleSync,
-			}
-
-			if err := ctx.EnsureTask(copyImageTask); err != nil {
-				return fmt.Errorf("error adding image-copy task: %v", err)
-			}
-			tasks = ctx.Tasks
-		}
 	}
 
 	seen = map[string]bool{}
@@ -163,41 +140,12 @@ func RunGetAssets(ctx context.Context, f *util.Factory, out io.Writer, options *
 			result.Files = append(result.Files, &file)
 			seen[file.Canonical] = true
 		}
-
-		// test if the asset needs to be copied
-		if options.Copy && fileAsset.DownloadURL.String() != fileAsset.CanonicalURL.String() {
-			ctx := &fi.ModelBuilderContext{
-				Tasks: tasks,
-			}
-
-			copyFileTask := &assettasks.CopyFile{
-				Name:       fi.String(fileAsset.CanonicalURL.String()),
-				TargetFile: fi.String(fileAsset.DownloadURL.String()),
-				SourceFile: fi.String(fileAsset.CanonicalURL.String()),
-				SHA:        fi.String(fileAsset.SHAValue),
-				Lifecycle:  fi.LifecycleSync,
-			}
-
-			if err := ctx.EnsureTask(copyFileTask); err != nil {
-				return fmt.Errorf("error adding file-copy task: %v", err)
-			}
-			tasks = ctx.Tasks
-		}
 	}
 
 	if options.Copy {
-		var options fi.RunTasksOptions
-		options.InitDefaults()
-
-		context, err := fi.NewContext(&copyAssetsTarget{}, updateClusterResults.Cluster, nil, nil, nil, nil, true, tasks)
-		if err != nil {
-			return fmt.Errorf("error building context: %v", err)
-		}
-		defer context.Close()
-
-		err = context.RunTasks(options)
+		err := assets.Copy(updateClusterResults.ImageAssets, updateClusterResults.FileAssets, updateClusterResults.Cluster)
 		if err != nil {
-			return fmt.Errorf("error running tasks: %v", err)
+			return err
 		}
 	}
 
@@ -260,11 +208,3 @@ func fileOutputTable(files []*File, out io.Writer) error {
 	columns := []string{"CANONICAL", "DOWNLOAD", "SHA"}
 	return t.Render(files, out, columns...)
 }
-
-func (c copyAssetsTarget) Finish(taskMap map[string]fi.Task) error {
-	return nil
-}
-
-func (c copyAssetsTarget) ProcessDeletions() bool {
-	return false
-}

pkg/assets/copy.go

@@ -0,0 +1,119 @@
+/*
+Copyright 2021 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package assets
+
+import (
+	"fmt"
+	"sort"
+
+	"k8s.io/klog/v2"
+	"k8s.io/kops/pkg/apis/kops"
+)
+
+type assetTask interface {
+	Run() error
+}
+
+func Copy(imageAssets []*ImageAsset, fileAssets []*FileAsset, cluster *kops.Cluster) error {
+	tasks := map[string]assetTask{}
+
+	for _, imageAsset := range imageAssets {
+		if imageAsset.DownloadLocation != imageAsset.CanonicalLocation {
+			copyImageTask := &CopyImage{
+				Name:        imageAsset.DownloadLocation,
+				SourceImage: imageAsset.CanonicalLocation,
+				TargetImage: imageAsset.DownloadLocation,
+			}
+
+			if existing, ok := tasks[copyImageTask.Name]; ok {
+				if existing.(*CopyImage).SourceImage != copyImageTask.SourceImage {
+					return fmt.Errorf("different sources for same image target %s: %s vs %s", copyImageTask.Name, copyImageTask.SourceImage, existing.(*CopyImage).SourceImage)
+				}
+			}
+
+			tasks[copyImageTask.Name] = copyImageTask
+		}
+	}
+
+	for _, fileAsset := range fileAssets {
+		if fileAsset.DownloadURL.String() != fileAsset.CanonicalURL.String() {
+			copyFileTask := &CopyFile{
+				Name:       fileAsset.CanonicalURL.String(),
+				TargetFile: fileAsset.DownloadURL.String(),
+				SourceFile: fileAsset.CanonicalURL.String(),
+				SHA:        fileAsset.SHAValue,
+				Cluster:    cluster,
+			}
+
+			if existing, ok := tasks[copyFileTask.Name]; ok {
+				e, ok := existing.(*CopyFile)
+				if !ok {
+					return fmt.Errorf("different types for copy target %s", copyFileTask.Name)
+				}
+				if e.TargetFile != copyFileTask.TargetFile {
+					return fmt.Errorf("different targets for same file %s: %s vs %s", copyFileTask.Name, copyFileTask.TargetFile, e.TargetFile)
+				}
+				if e.SHA != copyFileTask.SHA {
+					return fmt.Errorf("different sha for same file %s: %s vs %s", copyFileTask.Name, copyFileTask.SHA, e.SHA)
+				}
+			}
+
+			tasks[copyFileTask.Name] = copyFileTask
+		}
+	}
+
+	ch := make(chan error, 5)
+	for i := 0; i < cap(ch); i++ {
+		ch <- nil
+	}
+
+	gotError := false
+	names := make([]string, 0, len(tasks))
+	for name := range tasks {
+		names = append(names, name)
+	}
+	sort.Strings(names)
+	for _, name := range names {
+		task := tasks[name]
+		err := <-ch
+		if err != nil {
+			klog.Warning(err)
+			gotError = true
+		}
+		go func(n string, t assetTask) {
+			err := t.Run()
+			if err != nil {
+				err = fmt.Errorf("%s: %v", n, err)
+			}
+			ch <- err
+		}(name, task)
+	}
+
+	for i := 0; i < cap(ch); i++ {
+		err := <-ch
+		if err != nil {
+			klog.Warning(err)
+			gotError = true
+		}
+	}
+
+	close(ch)
+	if gotError {
+		return fmt.Errorf("not all assets copied successfully")
+	}
+	return nil
+}