Bug Fix - resource discovery deadlocks and improve error handling

Author: vimalmeena238

internal/resourcediscovery/export_compartment.go

@@ -19,6 +19,7 @@ import (
 	"sync"
 	"time"
 
+	"github.com/hashicorp/go-multierror"
 	"github.com/hashicorp/go-version"
 
 	"github.com/oracle/terraform-provider-oci/internal/globalvar"
@@ -489,6 +490,7 @@ func runExportCommand(ctx *resourceDiscoveryContext) error {
 			utils.Debugf("[DEBUG] discover: Running step %d", i)
 			defer elapsed(fmt.Sprintf("time taken in discovering resources for step %d", i), step.getBaseStep(), Discovery)()
 			defer func() {
+				<-sem
 				if r := recover(); r != nil {
 					utils.Logf("[ERROR] panic in discover goroutine")
 					utils.Logf("[ERROR] panic in discover goroutine")
@@ -520,7 +522,6 @@ func runExportCommand(ctx *resourceDiscoveryContext) error {
 
 			utils.Debugf("[DEBUG] discover: Completed step %d", i)
 			utils.Debugf("[DEBUG] discovered %d resources for step %d", len(step.getDiscoveredResources()), i)
-			<-sem
 		}(i, step)
 
 	}
@@ -560,7 +561,13 @@ func runExportCommand(ctx *resourceDiscoveryContext) error {
 	// Reset discovered resources if already set by writeTmpConfigurationForImport
 	ctx.discoveredResources = make([]*OCIResource, 0)
 
-	errorChannel := make(chan error)
+	/*
+		sem allows number of steps equals to arg.Parallelism to execute in parallel.
+		arg.Parallelism is very less compare to total number of steps
+		So, errorChannel should be atleast equals to number of steps to allow all steps getting chance to execute.
+		in case of multiple steps fails, steps will be blocked to write errors and pending steps will wait to acquire sem which leads to deadlock situation
+	*/
+	errorChannel := make(chan error, len(steps))
 	var configWg sync.WaitGroup
 	configWg.Add(len(steps))
 
@@ -574,6 +581,7 @@ func runExportCommand(ctx *resourceDiscoveryContext) error {
 		go func(i int, step resourceDiscoveryStep) {
 			utils.Debugf("[DEBUG] writeConfiguration: Running step %d", i)
 			defer func() {
+				<-sem
 				if r := recover(); r != nil {
 					utils.Logf("[ERROR] panic in writeConfiguration goroutine")
 					debug.PrintStack()
@@ -586,7 +594,6 @@ func runExportCommand(ctx *resourceDiscoveryContext) error {
 			}
 
 			utils.Debugf("[DEBUG] writeConfiguration: Completed step %d", i)
-			<-sem
 		}(i, step)
 	}
 
@@ -603,10 +610,13 @@ func runExportCommand(ctx *resourceDiscoveryContext) error {
 		utils.Debugf("[DEBUG] ~~~~~~ writeConfiguration steps completed ~~~~~~")
 		utils.Debugf("[DEBUG] writing config took %v\n", time.Since(configStart))
 		break
-	case err := <-errorChannel:
+	case errs := <-errorChannel:
 		close(errorChannel)
-		utils.Logf("[ERROR] error writing final configuration for resources found: %s", err.Error())
-		return err
+		for i := 0; i < len(errorChannel); i++ {
+			errs = multierror.Append(errs, <-errorChannel)
+		}
+		utils.Logf("[ERROR] error writing final configuration for resources found: %s", errs.Error())
+		return errs
 	}
 
 	region, err := exportConfigProvider.Region()
@@ -651,7 +661,13 @@ func generateStateParallel(ctx *resourceDiscoveryContext, steps []resourceDiscov
 	defer cleanupTempStateFiles(ctx)
 	defer elapsed("generating state in parallel", nil, 0)()
 
-	errorChannel := make(chan error)
+	/*
+		sem allows number of steps equals to arg.Parallelism to execute in parallel.
+		arg.Parallelism is very less compare to total number of steps
+		So, errorChannel should be atleast equals to number of steps to allow all steps getting chance to execute.
+		in case of multiple steps fails, steps will be blocked to write errors and pending steps will wait to acquire sem which leads to deadlock situation
+	*/
+	errorChannel := make(chan error, len(steps))
 	var stateWg sync.WaitGroup
 	wgDone := make(chan bool)
 
@@ -669,6 +685,7 @@ func generateStateParallel(ctx *resourceDiscoveryContext, steps []resourceDiscov
 			utils.Debugf("[DEBUG] writing temp config and state: Running step %d", i)
 			defer elapsed(fmt.Sprintf("time taken by step %s to generate state", fmt.Sprint(i)), step.getBaseStep(), GeneratingState)()
 			defer func() {
+				<-sem
 				if r := recover(); r != nil {
 					utils.Logf("[ERROR] panic in writing temp config and state goroutine")
 					debug.PrintStack()
@@ -680,17 +697,20 @@ func generateStateParallel(ctx *resourceDiscoveryContext, steps []resourceDiscov
 			/* Generate temporary HCL configs from all discovered resources to run import
 			   Final configuration will be generated after import so that we can exclude the resources for which import failed
 			   and also remove the references to failed resources from the referenceMap */
-			if err := step.writeTmpConfigurationForImport(); err != nil {
-				errorChannel <- fmt.Errorf("[ERROR] error writing temp config for resources found: %s", err.Error())
+			var err error = nil
+			if err = step.writeTmpConfigurationForImport(); err != nil {
+				errorChannel <- fmt.Errorf("[ERROR] error while writing temp config for resources found in step %d: %s", i, err.Error())
 			}
 
 			// Write temp state file for each service, this step will import resources into a separate state file for each service in parallel
-			if err := step.writeTmpState(); err != nil {
-				errorChannel <- fmt.Errorf("[ERROR] error writing temp state for resources found: %s", err.Error())
+			// If writing temp configuration thrown error, won't attempt to write temp state
+			if err == nil {
+				if err = step.writeTmpState(); err != nil {
+					errorChannel <- fmt.Errorf("[ERROR] error while writing temp state for resources found in step %d: %s", i, err.Error())
+				}
 			}
 
 			utils.Debugf("writing temp config and state: Completed step %d", i)
-			<-sem
 		}(i, step)
 	}
 
@@ -706,9 +726,13 @@ func generateStateParallel(ctx *resourceDiscoveryContext, steps []resourceDiscov
 	case <-wgDone:
 		utils.Debugf("[DEBUG] ~~~~~~ writing temp config and state steps completed ~~~~~~")
 		break
-	case err := <-errorChannel:
+	case errs := <-errorChannel:
 		close(errorChannel)
-		return err
+		for i := 0; i < len(errorChannel); i++ {
+			errs = multierror.Append(errs, <-errorChannel)
+		}
+		utils.Logf("[ERROR] error writing temp config and state: %s", errs.Error())
+		return errs
 	}
 
 	// Generate final state by merging state json generated for all services

internal/resourcediscovery/export_compartment_test.go

@@ -2222,6 +2222,94 @@ func TestUnitGenerateStateParallel(t *testing.T) {
 	assert.NoError(t, err, "")
 }
 
+func TestUnitGenerateStateParallelWhenTfInitFails(t *testing.T) {
+
+	ctx := getTestCtx()
+	nSteps := 80 // number of steps
+	steps := make([]resourceDiscoveryStep, nSteps)
+
+	compartmentId := resourceDiscoveryTestCompartmentOcid
+	if err := os.Setenv("export_tenancy_id", resourceDiscoveryTestTenancyOcid); err != nil {
+		t.Logf("unable to set export_tenancy_id. err: %v", err)
+		t.Fail()
+	}
+	outputDir, err := os.Getwd()
+	outputDir = fmt.Sprintf("%s%sdiscoveryTest-%d", outputDir, string(os.PathSeparator), time.Now().Nanosecond())
+	if err = os.Mkdir(outputDir, os.ModePerm); err != nil {
+		t.Logf("unable to mkdir %s. err: %v", outputDir, err)
+		t.Fail()
+	}
+
+	for i := 0; i < nSteps; i++ {
+		discoveredResources := []*OCIResource{}
+		discoveredResources = append(discoveredResources, &OCIResource{
+			compartmentId: compartmentId,
+			TerraformResource: TerraformResource{
+				id:             "ocid1.a.b.c",
+				terraformClass: "oci_resource_type1",
+				terraformName:  "type1_res1",
+			},
+			parent: &OCIResource{
+				TerraformResource: TerraformResource{terraformName: "tf"},
+			},
+		})
+
+		steps[i] = &resourceDiscoveryWithTargetIds{
+			resourceDiscoveryBaseStep: resourceDiscoveryBaseStep{
+				ctx:                 ctx,
+				name:                "resources" + fmt.Sprint(i),
+				discoveredResources: discoveredResources,
+				omittedResources:    []*OCIResource{},
+			},
+		}
+	}
+
+	type args struct {
+		steps []resourceDiscoveryStep
+		ctx   *resourceDiscoveryContext
+	}
+	t_args := args{
+		steps: steps,
+		ctx:   ctx,
+	}
+	tests := []struct {
+		name      string
+		args      args
+		mock      func()
+		wantError bool
+	}{
+		{
+			name: "If Import failed ,should Return error",
+			args: t_args,
+			mock: func() {
+				t_args.ctx.terraformProviderBinaryPath = "tf"
+				t_args.ctx.OutputDir = &outputDir
+				ctxTerraformImportVar = func(ctx *resourceDiscoveryContext, ctxBackground context.Context, address, id string, importArgs ...tfexec.ImportOption) error {
+					return nil
+				}
+				terraformInitMockVar = func(r *resourceDiscoveryBaseStep, backgroundCtx context.Context, initArgs []tfexec.InitOption) error {
+					return errors.New("Init failed")
+				}
+				sem = make(chan struct{}, 4) // Parallelism=4
+				resourcesMap = mockResourcesMap()
+			},
+			wantError: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			tt.mock()
+			err := generateStateParallel(ctx, steps)
+			if tt.wantError {
+				assert.Error(t, err)
+			} else {
+				assert.NoError(t, err, "")
+			}
+		})
+	}
+}
+
 func TestUnitGenerateState(test *testing.T) {
 	defer func() {
 		outputDir := getOutputDir()

internal/resourcediscovery/export_resource_helpers.go

@@ -335,6 +335,10 @@ func (r *resourceDiscoveryBaseStep) writeTmpState() error {
 	if !isInitDone {
 		utils.Debugf("[DEBUG] acquiring lock to run terraform init")
 		initLock.Lock()
+		defer func() {
+			utils.Debugf("[DEBUG] releasing lock")
+			initLock.Unlock()
+		}()
 		// Check for existence of .terraform folder to make sure init is not run already by another thread
 		if _, err := os.Stat(fmt.Sprintf("%s%s.terraform", *r.ctx.OutputDir, string(os.PathSeparator))); os.IsNotExist(err) {
 			// Run init command if not already run
@@ -349,12 +353,11 @@ func (r *resourceDiscoveryBaseStep) writeTmpState() error {
 			}
 
 			if err := terraformInitMockVar(r, backgroundCtx, initArgs); err != nil {
+				utils.Debugf("[ERROR] error occured while terraform init: %s", err.Error())
 				return err
 			}
 			isInitDone = true
 		}
-		initLock.Unlock()
-		utils.Debugf("[DEBUG] releasing lock")
 	}
 	tmpStateOutputDir := filepath.Join(*r.ctx.OutputDir, "tmp", r.name)
 	tmpStateOutputFilePrefix := filepath.Join(tmpStateOutputDir, globalvar.DefaultTmpStateFile)