Allow number of simultaneous workers to be configured

docs/hugo/content/tools/asoctl.md

@@ -252,7 +252,8 @@ Flags:
   -n, --namespace string       Write the imported resources to the specified namespace
   -o, --output string          Write ARM resource CRDs to a single file
   -f, --output-folder string   Write ARM resource CRDs to individual files in a folder
-
+  -w, --workers int            Specify the number of parallel workers to use when importing resources (default 4)
+
 Global Flags:
       --quiet     Silence most logging
       --verbose   Enable verbose logging

v2/cmd/asoctl/cmd/import_azure_resource.go

@@ -66,13 +66,15 @@ https://docs.microsoft.com/azure/active-directory/develop/authentication-nationa
 		},
 	}
 
-	options.outputPath = cmd.Flags().StringP(
+	cmd.Flags().StringVarP(
+		&options.outputPath,
 		"output",
 		"o",
 		"",
 		"Write ARM resource CRDs to a single file")
 
-	options.outputFolder = cmd.Flags().StringP(
+	cmd.Flags().StringVarP(
+		&options.outputFolder,
 		"output-folder",
 		"f",
 		"",
@@ -86,19 +88,28 @@ https://docs.microsoft.com/azure/active-directory/develop/authentication-nationa
 		"n",
 		"",
 		"Write the imported resources to the specified namespace")
+
 	cmd.Flags().StringSliceVarP(
 		&options.labels,
 		"label",
 		"l",
 		nil,
 		"Add the specified labels to the imported resources. Multiple comma-separated labels can be specified (--label example.com/mylabel=foo,example.com/mylabel2=bar) or the --label (-l) argument can be used multiple times (-l example.com/mylabel=foo -l example.com/mylabel2=bar)")
+
 	cmd.Flags().StringSliceVarP(
 		&options.annotations,
 		"annotation",
 		"a",
 		nil,
 		"Add the specified annotations to the imported resources. Multiple comma-separated annotations can be specified (--annotation example.com/myannotation=foo,example.com/myannotation2=bar) or the --annotation (-a) argument can be used multiple times (-a example.com/myannotation=foo -a example.com/myannotation2=bar)")
 
+	cmd.Flags().IntVarP(
+		&options.workers,
+		"workers",
+		"w",
+		4,
+		"Specify the number of parallel workers to use when importing resources")
+
 	return cmd
 }
 
@@ -131,7 +142,11 @@ func importAzureResource(
 	done := make(chan struct{}) // signal that we're done
 	pb := importreporter.NewBar("Import Azure Resources", progressBar, done)
 
-	importer := importresources.New(api.CreateScheme(), client, log, pb)
+	importerOptions := importresources.ResourceImporterOptions{
+		Workers: options.workers,
+	}
+
+	importer := importresources.New(api.CreateScheme(), client, log, pb, importerOptions)
 	for _, armID := range armIDs {
 		err = importer.AddARMID(armID)
 		if err != nil {
@@ -205,11 +220,12 @@ func importAzureResource(
 }
 
 type importAzureResourceOptions struct {
-	outputPath   *string
-	outputFolder *string
+	outputPath   string
+	outputFolder string
 	namespace    string
 	annotations  []string
 	labels       []string
+	workers      int
 
 	readCloud               sync.Once
 	azureAuthorityHost      string
@@ -218,16 +234,16 @@ type importAzureResourceOptions struct {
 }
 
 func (option *importAzureResourceOptions) writeToFile() (string, bool) {
-	if option.outputPath != nil && *option.outputPath != "" {
-		return *option.outputPath, true
+	if option.outputPath != "" {
+		return option.outputPath, true
 	}
 
 	return "", false
 }
 
 func (option *importAzureResourceOptions) writeToFolder() (string, bool) {
-	if option.outputFolder != nil && *option.outputFolder != "" {
-		return *option.outputFolder, true
+	if option.outputFolder != "" {
+		return option.outputFolder, true
 	}
 
 	return "", false

v2/cmd/asoctl/pkg/importresources/resource_importer.go

@@ -27,6 +27,13 @@ type ResourceImporter struct {
 	imported  map[string]ImportedResource     // A set of importers that have been successfully imported
 	log       logr.Logger                     // Logger to use for logging
 	reporter  importreporter.Interface        // Reporter to use for reporter updates
+	options   ResourceImporterOptions         // Options for the importer
+}
+
+// ResourceImporterOptions are optional configuration items for the importer
+type ResourceImporterOptions struct {
+	// Workers is the number of concurrent imports to run at the same time. If not specified, a default of 4 is used.
+	Workers int
 }
 
 type ImportResourceResult struct {
@@ -41,13 +48,15 @@ func New(
 	client *genericarmclient.GenericClient,
 	log logr.Logger,
 	reporter importreporter.Interface,
+	options ResourceImporterOptions,
 ) *ResourceImporter {
 	return &ResourceImporter{
 		scheme:   scheme,
 		client:   client,
 		imported: make(map[string]ImportedResource),
 		log:      log,
 		reporter: reporter,
+		options:  options,
 	}
 }
 
@@ -73,7 +82,7 @@ func (ri *ResourceImporter) Import(
 	ctx context.Context,
 	done chan struct{},
 ) (*Result, error) {
-	workers := 4
+	workersRequired := ri.desiredWorkers()
 	candidates := make(chan ImportableResource)  // candidates that need to be deduped
 	pending := make(chan ImportableResource)     // importers that are pending import
 	completed := make(chan ImportResourceResult) // importers that have been executed successfully
@@ -83,7 +92,7 @@ func (ri *ResourceImporter) Import(
 	go ri.queueUniqueImporters(candidates, pending, ri.reporter)
 
 	// Create workers to run the import
-	for i := 0; i < workers; i++ {
+	for i := 0; i < workersRequired; i++ {
 		go ri.importWorker(ctx, pending, completed, ri.reporter)
 	}
 
@@ -282,3 +291,12 @@ func (ri *ResourceImporter) importResource(
 
 	return result
 }
+
+// desiredWorkers returns the number of workers to use for importing resources.
+func (ri *ResourceImporter) desiredWorkers() int {
+	if ri.options.Workers > 0 {
+		return ri.options.Workers
+	}
+
+	return 4
+}