Merge pull request #527 from masnax/fixes

Fix concurrent access & cleanup timeout

Author: stgraber

cmd/migration-managerd/internal/api/workers.go

@@ -396,15 +396,13 @@ func (d *Daemon) beginImports(ctx context.Context, cleanupInstances bool) error
 	// This is used so that we ensure each pool in each target is checked only once for volumes in a particular project, for a particular VM OS.
 	visitedLocations := map[string]map[string]map[string]bool{}
 	ignoredBatches := []string{}
+	var volLock sync.Mutex
 	err = util.RunConcurrentMap(migrationState, func(batchName string, state queue.MigrationState) error {
 		log := log.With(slog.String("batch", state.Batch.Name))
 		for instUUID, q := range state.QueueEntries {
 			for _, pool := range q.Placement.StoragePools {
+				volLock.Lock()
 				// for every instance in this batch, check volumes at the corresponding target, unless we did already.
-				if visitedLocations == nil {
-					visitedLocations = map[string]map[string]map[string]bool{}
-				}
-
 				if visitedLocations[q.Placement.TargetName] == nil {
 					visitedLocations[q.Placement.TargetName] = map[string]map[string]bool{}
 				}
@@ -413,18 +411,27 @@ func (d *Daemon) beginImports(ctx context.Context, cleanupInstances bool) error
 					visitedLocations[q.Placement.TargetName][q.Placement.TargetProject] = map[string]bool{}
 				}
 
-				if !visitedLocations[q.Placement.TargetName][q.Placement.TargetProject][pool] {
+				shouldCreateVol := !visitedLocations[q.Placement.TargetName][q.Placement.TargetProject][pool]
+				if shouldCreateVol {
+					visitedLocations[q.Placement.TargetName][q.Placement.TargetProject][pool] = true
+				}
+
+				volLock.Unlock()
+
+				if shouldCreateVol {
 					err := d.ensureISOImagesExistInStoragePool(ctx, state.Instances[instUUID], state.Targets[instUUID], state.Batch, pool, q.Placement.TargetProject)
 					if err != nil {
+						volLock.Lock()
+						visitedLocations[q.Placement.TargetName][q.Placement.TargetProject][pool] = false
+						volLock.Unlock()
+
 						log.Error("Failed to validate batch", logger.Err(err))
 						_, err := d.batch.UpdateStatusByName(ctx, state.Batch.Name, api.BATCHSTATUS_ERROR, err.Error())
 						if err != nil {
 							return fmt.Errorf("Failed to set batch status to %q: %w", api.BATCHSTATUS_ERROR, err)
 						}
 
 						ignoredBatches = append(ignoredBatches, state.Batch.Name)
-					} else {
-						visitedLocations[q.Placement.TargetName][q.Placement.TargetProject][pool] = true
 					}
 				}
 			}
@@ -672,7 +679,32 @@ func (d *Daemon) createTargetVM(ctx context.Context, b migration.Batch, inst mig
 	if cleanupInstances {
 		reverter.Add(func() {
 			slog.Error("Cleaning up new instance after failure", slog.String("revert", "instance cleanup"), slog.Any("error", err))
-			cleanup()
+
+			// Instantiate a new client.
+			it, err := target.NewTarget(t.ToAPI())
+			if err != nil {
+				slog.Error("Failed to construct target", slog.String("name", t.Name), slog.Any("error", err))
+				return
+			}
+
+			timeoutCtx, cancel := context.WithTimeout(ctx, it.Timeout())
+			defer cancel()
+
+			// Connect to the target.
+			err = it.Connect(timeoutCtx)
+			if err != nil {
+				slog.Error("Failed to connect to target", slog.String("name", it.GetName()), slog.Any("error", err))
+				return
+			}
+
+			// Set the project.
+			err = it.SetProject(q.Placement.TargetProject)
+			if err != nil {
+				slog.Error("Failed to set project", slog.String("project", q.Placement.TargetProject), slog.String("name", it.GetName()), slog.Any("error", err))
+				return
+			}
+
+			cleanup(it)
 		})
 	}
 

internal/target/incus.go

@@ -593,7 +593,7 @@ func (t *InternalIncusTarget) CreateVMDefinition(instanceDef migration.Instance,
 	return ret, nil
 }
 
-func (t *InternalIncusTarget) CreateNewVM(ctx context.Context, instDef migration.Instance, apiDef incusAPI.InstancesPost, placement api.Placement, bootISOImage string) (func(context.Context) error, func(), error) {
+func (t *InternalIncusTarget) CreateNewVM(ctx context.Context, instDef migration.Instance, apiDef incusAPI.InstancesPost, placement api.Placement, bootISOImage string) (func(context.Context) error, func(t Target), error) {
 	reverter := revert.New()
 	defer reverter.Fail()
 
@@ -618,7 +618,7 @@ func (t *InternalIncusTarget) CreateNewVM(ctx context.Context, instDef migration
 		return nil, nil, err
 	}
 
-	cleanup := func() {
+	cleanup := func(t Target) {
 		err := t.CleanupVM(context.Background(), apiDef.Name, true)
 		if err != nil {
 			slog.Error("Failed to clean up instance after error", slog.String("name", apiDef.Name), slog.Any("error", err))
@@ -678,7 +678,6 @@ func (t *InternalIncusTarget) SetupVM(ctx context.Context, instDef migration.Ins
 			instInfo.Devices[diskKey]["source"] = diskName
 		}
 
-		apiDef.Start = true
 		op, err := tgtClient.UpdateInstance(instInfo.Name, instInfo.InstancePut, etag)
 		if err != nil {
 			return err

internal/target/interface.go

@@ -82,7 +82,7 @@ type Target interface {
 	CreateVMDefinition(instanceDef migration.Instance, usedNetworks migration.Networks, q migration.QueueEntry, fingerprint string, endpoint string, targetNetwork api.MigrationNetworkPlacement) (incusAPI.InstancesPost, error)
 
 	// Creates a new VM from the pre-populated API definition.
-	CreateNewVM(ctx context.Context, instDef migration.Instance, apiDef incusAPI.InstancesPost, placement api.Placement, bootISOImage string) (func(context.Context) error, func(), error)
+	CreateNewVM(ctx context.Context, instDef migration.Instance, apiDef incusAPI.InstancesPost, placement api.Placement, bootISOImage string) (func(context.Context) error, func(t Target), error)
 
 	// Creates a new VM from the pre-populated API definition.
 	SetupVM(ctx context.Context, instDef migration.Instance, apiDef incusAPI.InstancesPost, placement api.Placement) error