Merge pull request #429 from harishsurf/parallelize-export

Bug 1880127: Parallelize opm index export

Author: openshift-merge-robot

cmd/opm/index/export.go

@@ -12,8 +12,8 @@ import (
 var exportLong = templates.LongDesc(`
 	Export an operator from an index image into the appregistry format. 
 
-	This command will take an index image (specified by the --index option), parse it for the given operator (set by 
-	the --operator option) and export the operator metadata into an appregistry compliant format (a package.yaml file). 
+	This command will take an index image (specified by the --index option), parse it for the given operator(s) (set by 
+	the --package option) and export the operator metadata into an appregistry compliant format (a package.yaml file). 
 	This command requires access to docker or podman to complete successfully.
 
 	Note: the appregistry format is being deprecated in favor of the new index image and image bundle format. 
@@ -40,10 +40,7 @@ func newIndexExportCmd() *cobra.Command {
 	if err := indexCmd.MarkFlagRequired("index"); err != nil {
 		logrus.Panic("Failed to set required `index` flag for `index export`")
 	}
-	indexCmd.Flags().StringP("package", "o", "", "the package to export")
-	if err := indexCmd.MarkFlagRequired("package"); err != nil {
-		logrus.Panic("Failed to set required `package` flag for `index export`")
-	}
+	indexCmd.Flags().StringSliceP("package", "p", nil, "comma separated list of packages to export")
 	indexCmd.Flags().StringP("download-folder", "f", "downloaded", "directory where downloaded operator bundle(s) will be stored")
 	indexCmd.Flags().StringP("container-tool", "c", "none", "tool to interact with container images (save, build, etc.). One of: [none, docker, podman]")
 	if err := indexCmd.Flags().MarkHidden("debug"); err != nil {
@@ -60,7 +57,7 @@ func runIndexExportCmdFunc(cmd *cobra.Command, args []string) error {
 		return err
 	}
 
-	packageName, err := cmd.Flags().GetString("package")
+	packages, err := cmd.Flags().GetStringSlice("package")
 	if err != nil {
 		return err
 	}
@@ -80,15 +77,15 @@ func runIndexExportCmdFunc(cmd *cobra.Command, args []string) error {
 		return err
 	}
 
-	logger := logrus.WithFields(logrus.Fields{"index": index, "package": packageName})
+	logger := logrus.WithFields(logrus.Fields{"index": index, "package": packages})
 
 	logger.Info("export from the index")
 
 	indexExporter := indexer.NewIndexExporter(containertools.NewContainerTool(containerTool, containertools.NoneTool), logger)
 
 	request := indexer.ExportFromIndexRequest{
 		Index:         index,
-		Package:       packageName,
+		Packages:      packages,
 		DownloadPath:  downloadPath,
 		ContainerTool: containertools.NewContainerTool(containerTool, containertools.NoneTool),
 		SkipTLS:       skipTLS,

pkg/lib/bundle/exporter.go

@@ -44,7 +44,7 @@ func (i *BundleExporter) Export() error {
 	var rerr error
 	switch i.containerTool {
 	case containertools.NoneTool:
-		reg, rerr = containerdregistry.NewRegistry(containerdregistry.WithLog(log))
+		reg, rerr = containerdregistry.NewRegistry(containerdregistry.WithLog(log), containerdregistry.WithCacheDir(filepath.Join(tmpDir, "cacheDir")))
 	case containertools.PodmanTool:
 		fallthrough
 	case containertools.DockerTool:

pkg/lib/indexer/indexer.go

@@ -6,9 +6,12 @@ import (
 	"fmt"
 	"io"
 	"io/ioutil"
+	"math/rand"
 	"os"
 	"path"
 	"path/filepath"
+	"strconv"
+	"sync"
 
 	"github.com/sirupsen/logrus"
 	"gopkg.in/yaml.v2"
@@ -26,12 +29,13 @@ import (
 )
 
 const (
-	defaultDockerfileName = "index.Dockerfile"
-	defaultImageTag       = "operator-registry-index:latest"
-	defaultDatabaseFolder = "database"
-	defaultDatabaseFile   = "index.db"
-	tmpDirPrefix          = "index_tmp_"
-	tmpBuildDirPrefix     = "index_build_tmp"
+	defaultDockerfileName     = "index.Dockerfile"
+	defaultImageTag           = "operator-registry-index:latest"
+	defaultDatabaseFolder     = "database"
+	defaultDatabaseFile       = "index.db"
+	tmpDirPrefix              = "index_tmp_"
+	tmpBuildDirPrefix         = "index_build_tmp"
+	concurrencyLimitForExport = 10
 )
 
 // ImageIndexer is a struct implementation of the Indexer interface
@@ -467,7 +471,7 @@ func write(dockerfileText, outDockerfile string, logger *logrus.Entry) error {
 // ExportFromIndexRequest defines the parameters to send to the ExportFromIndex API
 type ExportFromIndexRequest struct {
 	Index         string
-	Package       string
+	Packages      []string
 	DownloadPath  string
 	ContainerTool containertools.ContainerTool
 	CaFile        string
@@ -497,14 +501,20 @@ func (i ImageIndexer) ExportFromIndex(request ExportFromIndexRequest) error {
 	defer db.Close()
 
 	dbQuerier := sqlite.NewSQLLiteQuerierFromDb(db)
-	if err != nil {
-		return err
+
+	// fetch all packages from the index image if packages is empty
+	if len(request.Packages) == 0 {
+		request.Packages, err = dbQuerier.ListPackages(context.TODO())
+		if err != nil {
+			return err
+		}
 	}
 
-	bundles, err := getBundlesToExport(dbQuerier, request.Package)
+	bundles, err := getBundlesToExport(dbQuerier, request.Packages)
 	if err != nil {
 		return err
 	}
+
 	i.Logger.Infof("Preparing to pull bundles %+q", bundles)
 
 	// Creating downloadPath dir
@@ -513,40 +523,68 @@ func (i ImageIndexer) ExportFromIndex(request ExportFromIndexRequest) error {
 	}
 
 	var errs []error
-	for _, bundleImage := range bundles {
-		// try to name the folder
-		folderName, err := dbQuerier.GetBundleVersion(context.TODO(), bundleImage)
-		if err != nil {
-			return err
-		}
-		if folderName == "" {
-			// operator-registry does not care about the folder name
-			folderName = bundleImage
-		}
-		exporter := bundle.NewExporterForBundle(bundleImage, filepath.Join(request.DownloadPath, folderName), request.ContainerTool)
-		if err := exporter.Export(); err != nil {
-			err = fmt.Errorf("error exporting bundle from image: %s", err)
-			errs = append(errs, err)
-		}
-	}
-	if err != nil {
-		errs = append(errs, err)
+	var wg sync.WaitGroup
+	wg.Add(len(bundles))
+	var mu = &sync.Mutex{}
+
+	sem := make(chan struct{}, concurrencyLimitForExport)
+
+	for bundleImage, bundleDir := range bundles {
+		go func(bundleImage string, bundleDir bundleDirPrefix) {
+			defer wg.Done()
+
+			sem <- struct{}{}
+			defer func() {
+				<-sem
+			}()
+
+			// generate a random folder name if bundle version is empty
+			if bundleDir.bundleVersion == "" {
+				bundleDir.bundleVersion = strconv.Itoa(rand.Intn(10000))
+			}
+			exporter := bundle.NewExporterForBundle(bundleImage, filepath.Join(request.DownloadPath, bundleDir.pkgName, bundleDir.bundleVersion), request.ContainerTool)
+			if err := exporter.Export(); err != nil {
+				err = fmt.Errorf("exporting bundle image:%s failed with %s", bundleImage, err)
+				mu.Lock()
+				errs = append(errs, err)
+				mu.Unlock()
+			}
+		}(bundleImage, bundleDir)
+	}
+	// Wait for all the go routines to finish export
+	wg.Wait()
+
+	if errs != nil {
 		return utilerrors.NewAggregate(errs)
 	}
 
-	err = generatePackageYaml(dbQuerier, request.Package, request.DownloadPath)
-	if err != nil {
-		errs = append(errs, err)
+	for _, packageName := range request.Packages {
+		err := generatePackageYaml(dbQuerier, packageName, filepath.Join(request.DownloadPath, packageName))
+		if err != nil {
+			errs = append(errs, err)
+		}
 	}
 	return utilerrors.NewAggregate(errs)
 }
 
-func getBundlesToExport(dbQuerier pregistry.Query, packageName string) ([]string, error) {
-	bundles, err := dbQuerier.GetBundlePathsForPackage(context.TODO(), packageName)
-	if err != nil {
-		return nil, err
+type bundleDirPrefix struct {
+	pkgName, bundleVersion string
+}
+
+func getBundlesToExport(dbQuerier pregistry.Query, packages []string) (map[string]bundleDirPrefix, error) {
+	bundleMap := make(map[string]bundleDirPrefix)
+
+	for _, packageName := range packages {
+		bundlesForPackage, err := dbQuerier.GetBundlesForPackage(context.TODO(), packageName)
+		if err != nil {
+			return nil, err
+		}
+		for k, _ := range bundlesForPackage {
+			bundleMap[k.BundlePath] = bundleDirPrefix{pkgName: packageName, bundleVersion: k.Version}
+		}
 	}
-	return bundles, nil
+
+	return bundleMap, nil
 }
 
 func generatePackageYaml(dbQuerier pregistry.Query, packageName, downloadPath string) error {

pkg/lib/indexer/indexer_test.go

@@ -30,10 +30,16 @@ func TestGetBundlesToExport(t *testing.T) {
 		t.Fatalf("creating querier: %s", err)
 	}
 
-	bundleImages, err := getBundlesToExport(dbQuerier, "etcd")
+	bundleMap, err := getBundlesToExport(dbQuerier, []string {"etcd"})
 	if err != nil {
 		t.Fatalf("exporting bundles from db: %s", err)
 	}
+
+	var bundleImages []string
+	for bundlePath, _ := range bundleMap {
+		bundleImages = append(bundleImages, bundlePath)
+	}
+
 	sort.Strings(bundleImages)
 
 	if !reflect.DeepEqual(expected, bundleImages) {

test/e2e/opm_test.go

@@ -137,13 +137,30 @@ func pushWith(containerTool, image string) error {
 	return dockerpush.Run()
 }
 
-func exportWith(containerTool string) error {
-	logger := logrus.WithFields(logrus.Fields{"package": packageName})
+func exportPackageWith(containerTool string) error {
+	packages := []string{packageName}
+	logger := logrus.WithFields(logrus.Fields{"package": packages})
 	indexExporter := indexer.NewIndexExporter(containertools.NewContainerTool(containerTool, containertools.NoneTool), logger)
 
 	request := indexer.ExportFromIndexRequest{
 		Index:         indexImage2,
-		Package:       packageName,
+		Packages:      packages,
+		DownloadPath:  "downloaded",
+		ContainerTool: containertools.NewContainerTool(containerTool, containertools.NoneTool),
+	}
+
+	return indexExporter.ExportFromIndex(request)
+}
+
+
+func exportIndexImageWith(containerTool string) error {
+
+	logger := logrus.NewEntry(logrus.New())
+	indexExporter := indexer.NewIndexExporter(containertools.NewContainerTool(containerTool, containertools.NoneTool), logger)
+
+	request := indexer.ExportFromIndexRequest{
+		Index:         indexImage2,
+		Packages:      []string{},
 		DownloadPath:  "downloaded",
 		ContainerTool: containertools.NewContainerTool(containerTool, containertools.NoneTool),
 	}
@@ -272,8 +289,8 @@ var _ = Describe("opm", func() {
 			err = pushWith(containerTool, indexImage3)
 			Expect(err).NotTo(HaveOccurred())
 
-			By("exporting an index to disk")
-			err = exportWith(containerTool)
+			By("exporting a package from an index to disk")
+			err = exportPackageWith(containerTool)
 			Expect(err).NotTo(HaveOccurred())
 
 			By("loading manifests from a directory")
@@ -284,13 +301,26 @@ var _ = Describe("opm", func() {
 			err = os.RemoveAll("downloaded")
 			Expect(err).NotTo(HaveOccurred())
 
-			By("exporting an index to disk with containerd")
-			err = exportWith(containertools.NoneTool.String())
+			By("exporting a package from an index to disk with containerd")
+			err = exportPackageWith(containertools.NoneTool.String())
 			Expect(err).NotTo(HaveOccurred())
 
 			By("loading manifests from a containerd-extracted directory")
 			err = initialize()
 			Expect(err).NotTo(HaveOccurred())
+
+			// clean containerd-extracted directory
+			err = os.RemoveAll("downloaded")
+			Expect(err).NotTo(HaveOccurred())
+
+			By("exporting an entire index to disk")
+			err = exportIndexImageWith(containerTool)
+			Expect(err).NotTo(HaveOccurred())
+
+			By("loading manifests from a directory")
+			err = initialize()
+			Expect(err).NotTo(HaveOccurred())
+
 		})
 
 		It("build bundles and index via inference", func() {