parallelise the importer for speed - note logging output becomes non-sequential, needs a sequence number prefix maybe?

Author: jackofallops

tools/importer-rest-api-specs/internal/components/apidefinitions/parse_service.go

@@ -23,10 +23,10 @@ func ParseService(input discoveryModels.AvailableDataSet) (*sdkModels.Service, e
 	apiVersions := make(map[string]sdkModels.APIVersion)
 
 	for apiVersionName, dataSet := range input.DataSetsForAPIVersions {
-		logging.Infof("Parsing Data for API Version %q..", apiVersionName)
+		logging.Infof("Parsing Data for %q API Version %q..", input.ServiceName, apiVersionName)
 		parsed, err := parseAPIVersion(input.ServiceName, dataSet, input.ResourceProvider)
 		if err != nil {
-			return nil, fmt.Errorf("parsing API Version %q: %+v", apiVersionName, err)
+			return nil, fmt.Errorf("parsing %q API Version %q: %+v", input.ServiceName, apiVersionName, err)
 		}
 
 		apiVersions[apiVersionName] = *parsed

tools/importer-rest-api-specs/internal/pipeline/import.go

@@ -4,8 +4,11 @@
 package pipeline
 
 import (
+	"context"
 	"fmt"
+	"runtime"
 
+	"golang.org/x/sync/errgroup"
 	"github.com/hashicorp/pandora/tools/data-api-repository/repository"
 	"github.com/hashicorp/pandora/tools/importer-rest-api-specs/internal/components/terraform"
 	"github.com/hashicorp/pandora/tools/importer-rest-api-specs/internal/logging"
@@ -47,7 +50,13 @@ func RunImporter(opts Options) error {
 	logging.Debug("Completed - Clearing any existing API Definitions.")
 
 	logging.Infof("Processing %d Services..", len(p.servicesFromConfigurationFiles))
+
+	g, ctx := errgroup.WithContext(context.Background())
+	semaphore := make(chan struct{}, runtime.NumCPU())
+
 	for _, service := range p.servicesFromConfigurationFiles {
+		service := service
+
 		if len(opts.ServiceNamesToLimitTo) > 0 {
 			processThisService := false
 			for _, serviceNameToLimitTo := range opts.ServiceNamesToLimitTo {
@@ -62,37 +71,52 @@ func RunImporter(opts Options) error {
 			}
 		}
 
-		logging.Infof("Discovering the Data for Service %q..", service.Name)
-		data, err := p.parseDataForService(service)
-		if err != nil {
-			return fmt.Errorf("parsing Data for the Service %q: %+v", service.Name, err)
-		}
-		logging.Debugf("Completed - Discovering the Data for Service %q.", service.Name)
+		g.Go(func() error {
+			select {
+			case semaphore <- struct{}{}:
+			case <-ctx.Done():
+				return ctx.Err()
+			}
+			defer func() { <-semaphore }()
 
-		terraformDetails, ok := p.servicesToTerraformDetails[service.Name]
-		if ok {
-			logging.Infof("Building the Terraform Data for the Service %q..", service.Name)
-			data, err = terraform.BuildForService(*data, terraformDetails.resourceLabelToResourceDefinitions, p.opts.ProviderPrefix, terraformDetails.terraformPackageName)
+			logging.Infof("Discovering the Data for Service %q..", service.Name)
+			data, err := p.parseDataForService(service)
 			if err != nil {
-				return fmt.Errorf("building the Terraform Data for Service %q: %+v", service.Name, err)
+				return fmt.Errorf("parsing Data for the Service %q: %+v", service.Name, err)
 			}
-			logging.Debugf("Completed - Building the Terraform Data for the Service %q.", service.Name)
-		} else {
-			logging.Debugf("Skipping - no Terraform Definitions for the Service %q..", service.Name)
-		}
+			logging.Debugf("Completed - Discovering the Data for Service %q.", service.Name)
 
-		logging.Infof("Writing Data for Service %q..", service.Name)
-		saveServiceOpts := repository.SaveServiceOptions{
-			ResourceProvider: service.ResourceProvider,
-			Service:          *data,
-			ServiceName:      service.Name,
-			SourceCommitSHA:  restAPISpecsCommitSHA,
-			SourceDataOrigin: p.opts.SourceDataOrigin,
-		}
-		if err := p.repository.SaveService(saveServiceOpts); err != nil {
-			return fmt.Errorf("saving the Service %q: %+v", service.Name, err)
-		}
-		logging.Debugf("Completed - writing Data for Service %q.", service.Name)
+			terraformDetails, ok := p.servicesToTerraformDetails[service.Name]
+			if ok {
+				logging.Infof("Building the Terraform Data for the Service %q..", service.Name)
+				data, err = terraform.BuildForService(*data, terraformDetails.resourceLabelToResourceDefinitions, p.opts.ProviderPrefix, terraformDetails.terraformPackageName)
+				if err != nil {
+					return fmt.Errorf("building the Terraform Data for Service %q: %+v", service.Name, err)
+				}
+				logging.Debugf("Completed - Building the Terraform Data for the Service %q.", service.Name)
+			} else {
+				logging.Debugf("Skipping - no Terraform Definitions for the Service %q..", service.Name)
+			}
+
+			logging.Infof("Writing Data for Service %q..", service.Name)
+			saveServiceOpts := repository.SaveServiceOptions{
+				ResourceProvider: service.ResourceProvider,
+				Service:          *data,
+				ServiceName:      service.Name,
+				SourceCommitSHA:  restAPISpecsCommitSHA,
+				SourceDataOrigin: p.opts.SourceDataOrigin,
+			}
+			if err := p.repository.SaveService(saveServiceOpts); err != nil {
+				return fmt.Errorf("saving the Service %q: %+v", service.Name, err)
+			}
+			logging.Debugf("Completed - writing Data for Service %q.", service.Name)
+
+			return nil
+		})
+	}
+
+	if err := g.Wait(); err != nil {
+		return err
 	}
 
 	logging.Info("Completed - Importing Data.")