Parallelize initializing provider in the importing directories (#341)

* parallelize init

* Add error log for importing error

* go mod tidy

go.mod

@@ -24,7 +24,7 @@ require (
 	github.com/magodo/textinput v0.0.0-20210913072708-7d24f2b4b0c0
 	github.com/magodo/tfadd v0.10.1-0.20230106064825-378b3ebb9a4e
 	github.com/magodo/tfmerge v0.0.0-20221214062955-f52e46d03402
-	github.com/magodo/workerpool v0.0.0-20211124060943-1c48f3e5a514
+	github.com/magodo/workerpool v0.0.0-20230119025400-40192d2716ea
 	github.com/mitchellh/go-wordwrap v1.0.0
 	github.com/muesli/reflow v0.3.0
 	github.com/stretchr/testify v1.8.0

go.sum

@@ -208,8 +208,8 @@ github.com/magodo/tfpluginschema v0.0.0-20220905090502-2d6a05ebaefd h1:L0kTduNwp
 github.com/magodo/tfpluginschema v0.0.0-20220905090502-2d6a05ebaefd/go.mod h1:u625f3VQoOZTAxoDjeDLmrBdKqriRK4OHCpSWt7FQc8=
 github.com/magodo/tfstate v0.0.0-20220409052014-9b9568dda918 h1:yZ5ZEMSKZNCM7KpivKhDNNQEZYSDxg0Wyi5p0hQ8dVo=
 github.com/magodo/tfstate v0.0.0-20220409052014-9b9568dda918/go.mod h1:BW96zQS6A92qWVONOjviK73K0HlKdt5ufNBioGnOaEs=
-github.com/magodo/workerpool v0.0.0-20211124060943-1c48f3e5a514 h1:9JtvsO+tAKh70rXqUb39Ldn4p6zDqDOgrCpo15MM1cw=
-github.com/magodo/workerpool v0.0.0-20211124060943-1c48f3e5a514/go.mod h1:oI7XLq0SfJZISAwYhT7DHmi1Fqbr1Q8ZE8gegIe7iAI=
+github.com/magodo/workerpool v0.0.0-20230119025400-40192d2716ea h1:QK7gPkX5ubzDLTImfsSvAlVv/n/x1smVkeYpECA0mPo=
+github.com/magodo/workerpool v0.0.0-20230119025400-40192d2716ea/go.mod h1:oI7XLq0SfJZISAwYhT7DHmi1Fqbr1Q8ZE8gegIe7iAI=
 github.com/matryer/is v1.2.0/go.mod h1:2fLPjFQM9rhQ15aVEtbuwhJinnOqrmgXPNdZsdwlWXA=
 github.com/mattn/go-colorable v0.1.9/go.mod h1:u6P/XSegPjTcexA+o6vUJrdnUu04hMope9wVRipJSqc=
 github.com/mattn/go-colorable v0.1.12 h1:jF+Du6AlPIjs2BiUiQlKOX0rt3SujHxPnksPKZbaA40=

internal/meta/base_meta.go

@@ -263,36 +263,6 @@ func (meta *baseMeta) CleanTFState(ctx context.Context, addr string) {
 	meta.tf.StateRm(ctx, addr)
 }
 
-func (meta *baseMeta) importItem(ctx context.Context, item *ImportItem, importIdx int) {
-	if item.Skip() {
-		log.Printf("[INFO] Skipping %s", item.TFResourceId)
-		return
-	}
-
-	moduleDir := meta.importModuleDirs[importIdx]
-	tf := meta.importTFs[importIdx]
-
-	// Construct the empty cfg file for importing
-	cfgFile := filepath.Join(moduleDir, meta.filenameTmpCfg())
-	tpl := fmt.Sprintf(`resource "%s" "%s" {}`, item.TFAddr.Type, item.TFAddr.Name)
-	// #nosec G306
-	if err := os.WriteFile(cfgFile, []byte(tpl), 0644); err != nil {
-		item.ImportError = fmt.Errorf("generating resource template file: %w", err)
-		return
-	}
-	defer os.Remove(cfgFile)
-
-	// Import resources
-	addr := item.TFAddr.String()
-	if meta.moduleAddr != "" {
-		addr = meta.moduleAddr + "." + addr
-	}
-	log.Printf("[INFO] Importing %s as %s", item.TFResourceId, addr)
-	err := tf.Import(ctx, addr, item.TFResourceId)
-	item.ImportError = err
-	item.Imported = err == nil
-}
-
 func (meta *baseMeta) ParallelImport(ctx context.Context, items []*ImportItem) {
 	itemsCh := make(chan *ImportItem, len(items))
 	for _, item := range items {
@@ -606,30 +576,78 @@ func (meta *baseMeta) initProvider(ctx context.Context) error {
 	for _, opt := range meta.backendConfig {
 		opts = append(opts, tfexec.BackendConfig(opt))
 	}
+
+	log.Printf(`[DEBUG] Run "terraform init" for the output directory %s`, meta.outdir)
 	if err := meta.tf.Init(ctx, opts...); err != nil {
-		return fmt.Errorf("error running terraform init: %s", err)
+		return fmt.Errorf("error running terraform init for the output directory: %s", err)
 	}
 
 	// Initialize provider for the import directories.
+	wp := workerpool.NewWorkPool(meta.parallelism)
+	wp.Run(nil)
 	for i := range meta.importBaseDirs {
-		providerFile := filepath.Join(meta.importBaseDirs[i], "provider.tf")
-		// #nosec G306
-		if err := os.WriteFile(providerFile, []byte(meta.providerConfig()), 0644); err != nil {
-			return fmt.Errorf("error creating provider config: %w", err)
-		}
-		terraformFile := filepath.Join(meta.importBaseDirs[i], "terraform.tf")
-		// #nosec G306
-		if err := os.WriteFile(terraformFile, []byte(meta.terraformConfig("local")), 0644); err != nil {
-			return fmt.Errorf("error creating terraform config: %w", err)
-		}
-		if err := meta.importTFs[i].Init(ctx); err != nil {
-			return fmt.Errorf("error running terraform init: %s", err)
-		}
+		i := i
+		wp.AddTask(func() (interface{}, error) {
+			providerFile := filepath.Join(meta.importBaseDirs[i], "provider.tf")
+			// #nosec G306
+			if err := os.WriteFile(providerFile, []byte(meta.providerConfig()), 0644); err != nil {
+				return nil, fmt.Errorf("error creating provider config: %w", err)
+			}
+			terraformFile := filepath.Join(meta.importBaseDirs[i], "terraform.tf")
+			// #nosec G306
+			if err := os.WriteFile(terraformFile, []byte(meta.terraformConfig("local")), 0644); err != nil {
+				return nil, fmt.Errorf("error creating terraform config: %w", err)
+			}
+			log.Printf(`[DEBUG] Run "terraform init" for the import directory %s`, meta.importBaseDirs[i])
+			if err := meta.importTFs[i].Init(ctx); err != nil {
+				return nil, fmt.Errorf("error running terraform init: %s", err)
+			}
+			return nil, nil
+		})
+	}
+	if err := wp.Done(); err != nil {
+		return fmt.Errorf("initializing provider for the import directories: %v", err)
 	}
 
 	return nil
 }
 
+func (meta *baseMeta) importItem(ctx context.Context, item *ImportItem, importIdx int) {
+	if item.Skip() {
+		log.Printf("[INFO] Skipping %s", item.TFResourceId)
+		return
+	}
+
+	moduleDir := meta.importModuleDirs[importIdx]
+	tf := meta.importTFs[importIdx]
+
+	// Construct the empty cfg file for importing
+	cfgFile := filepath.Join(moduleDir, meta.filenameTmpCfg())
+	tpl := fmt.Sprintf(`resource "%s" "%s" {}`, item.TFAddr.Type, item.TFAddr.Name)
+	// #nosec G306
+	if err := os.WriteFile(cfgFile, []byte(tpl), 0644); err != nil {
+		err := fmt.Errorf("generating resource template file for %s: %w", item.TFAddr, err)
+		log.Printf("[ERROR] %v", err)
+		item.ImportError = err
+		return
+	}
+	defer os.Remove(cfgFile)
+
+	// Import resources
+	addr := item.TFAddr.String()
+	if meta.moduleAddr != "" {
+		addr = meta.moduleAddr + "." + addr
+	}
+	log.Printf("[INFO] Importing %s as %s", item.TFResourceId, addr)
+	err := tf.Import(ctx, addr, item.TFResourceId)
+	if err != nil {
+		err = fmt.Errorf("importing %s: %w", item.TFAddr, err)
+		log.Printf("[ERROR] %v", err)
+	}
+	item.ImportError = err
+	item.Imported = err == nil
+}
+
 func (meta baseMeta) stateToConfig(ctx context.Context, list ImportList) (ConfigInfos, error) {
 	out := make([]ConfigInfo, len(list.Imported()))
 

main.go

@@ -4,7 +4,6 @@ import (
 	"bytes"
 	"context"
 	"fmt"
-	internalconfig "github.com/Azure/aztfy/internal/config"
 	"io"
 	golog "log"
 	"os"
@@ -13,6 +12,8 @@ import (
 	"strconv"
 	"strings"
 
+	internalconfig "github.com/Azure/aztfy/internal/config"
+
 	"github.com/Azure/aztfy/pkg/config"
 	"github.com/Azure/aztfy/pkg/log"
 
@@ -281,7 +282,7 @@ The output directory is not empty. Please choose one of actions below:
 		&cli.StringFlag{
 			Name:        "log-level",
 			EnvVars:     []string{"AZTFY_LOG_LEVEL"},
-			Usage:       `Log level, can be one of "ERROR", "WARN", "INFO", "DEBUG"`,
+			Usage:       `Log level, can be one of "ERROR", "WARN", "INFO", "DEBUG" and "TRACE"`,
 			Destination: &flagLogLevel,
 			Value:       "INFO",
 		},
@@ -530,6 +531,8 @@ func logLevel(level string) (hclog.Level, error) {
 		return hclog.Info, nil
 	case "DEBUG":
 		return hclog.Debug, nil
+	case "TRACE":
+		return hclog.Trace, nil
 	default:
 		return hclog.NoLevel, fmt.Errorf("unknown log level: %s", level)
 	}
@@ -562,7 +565,7 @@ func initLog(path string, level hclog.Level) error {
 		// Enable log for azure sdk
 		os.Setenv("AZURE_SDK_GO_LOGGING", "all") // #nosec G104
 		azlog.SetListener(func(cls azlog.Event, msg string) {
-			logger.Printf("[DEBUG] %s: %s\n", cls, msg)
+			logger.Printf("[TRACE] %s: %s\n", cls, msg)
 		})
 	}
 	return nil