Merge pull request #2569 from stgraber/storage

Fix panic on storage migration

Author: hallyn

cmd/incusd/api_cluster.go

@@ -2979,7 +2979,7 @@ func clusterNodeStatePost(d *Daemon, r *http.Request) response.Response {
 			return nil
 		}
 
-		migrateFunc := func(ctx context.Context, s *state.State, inst instance.Instance, sourceMemberInfo *db.NodeInfo, targetMemberInfo *db.NodeInfo, live bool, startInstance bool, metadata map[string]any, op *operations.Operation) error {
+		migrateFunc := func(ctx context.Context, s *state.State, inst instance.Instance, sourceMemberInfo *db.NodeInfo, targetMemberInfo *db.NodeInfo, live bool, startInstance bool, op *operations.Operation) error {
 			// Migrate the instance.
 			req := api.InstancePost{
 				Migration: true,
@@ -3003,9 +3003,8 @@ func clusterNodeStatePost(d *Daemon, r *http.Request) response.Response {
 
 			dest = dest.UseProject(inst.Project().Name)
 
-			if metadata != nil && op != nil {
-				metadata["evacuation_progress"] = fmt.Sprintf("Starting %q in project %q", inst.Name(), inst.Project().Name)
-				_ = op.UpdateMetadata(metadata)
+			if op != nil {
+				_ = op.ExtendMetadata(map[string]any{"evacuation_progress": fmt.Sprintf("Starting %q in project %q", inst.Name(), inst.Project().Name)})
 			}
 
 			startOp, err := dest.UpdateInstanceState(inst.Name(), api.InstanceStatePut{Action: "start"}, "")

cmd/incusd/api_cluster_evacuation.go

@@ -40,7 +40,7 @@ import (
 
 type (
 	evacuateStopFunc    func(inst instance.Instance, action string) error
-	evacuateMigrateFunc func(ctx context.Context, s *state.State, inst instance.Instance, sourceMemberInfo *db.NodeInfo, targetMemberInfo *db.NodeInfo, live bool, startInstance bool, metadata map[string]any, op *operations.Operation) error
+	evacuateMigrateFunc func(ctx context.Context, s *state.State, inst instance.Instance, sourceMemberInfo *db.NodeInfo, targetMemberInfo *db.NodeInfo, live bool, startInstance bool, op *operations.Operation) error
 )
 
 type evacuateOpts struct {
@@ -185,8 +185,6 @@ func evacuateInstances(ctx context.Context, opts evacuateOpts) error {
 }
 
 func evacuateInstancesFunc(ctx context.Context, inst instance.Instance, opts evacuateOpts) error {
-	metadata := make(map[string]any)
-
 	instProject := inst.Project()
 	l := logger.AddContext(logger.Ctx{"project": instProject.Name, "instance": inst.Name()})
 
@@ -214,8 +212,7 @@ func evacuateInstancesFunc(ctx context.Context, inst instance.Instance, opts eva
 	isRunning := inst.IsRunning()
 	if action != "live-migrate" {
 		if opts.stopInstance != nil && isRunning {
-			metadata["evacuation_progress"] = fmt.Sprintf("Stopping %q in project %q", inst.Name(), instProject.Name)
-			_ = opts.op.UpdateMetadata(metadata)
+			_ = opts.op.ExtendMetadata(map[string]any{"evacuation_progress": fmt.Sprintf("Stopping %q in project %q", inst.Name(), instProject.Name)})
 
 			err := opts.stopInstance(inst, action)
 			if err != nil {
@@ -245,16 +242,15 @@ func evacuateInstancesFunc(ctx context.Context, inst instance.Instance, opts eva
 	}
 
 	// Start migrating the instance.
-	metadata["evacuation_progress"] = fmt.Sprintf("Migrating %q in project %q to %q", inst.Name(), instProject.Name, targetMemberInfo.Name)
-	_ = opts.op.UpdateMetadata(metadata)
+	_ = opts.op.ExtendMetadata(map[string]any{"evacuation_progress": fmt.Sprintf("Migrating %q in project %q to %q", inst.Name(), instProject.Name, targetMemberInfo.Name)})
 
 	// Set origin server (but skip if already set as that suggests more than one server being evacuated).
 	if inst.LocalConfig()["volatile.evacuate.origin"] == "" {
 		_ = inst.VolatileSet(map[string]string{"volatile.evacuate.origin": opts.srcMemberName})
 	}
 
 	start := isRunning || instanceShouldAutoStart(inst)
-	err = opts.migrateInstance(ctx, opts.s, inst, sourceMemberInfo, targetMemberInfo, action == "live-migrate", start, metadata, opts.op)
+	err = opts.migrateInstance(ctx, opts.s, inst, sourceMemberInfo, targetMemberInfo, action == "live-migrate", start, opts.op)
 	if err != nil {
 		return err
 	}
@@ -342,9 +338,7 @@ func restoreClusterMember(d *Daemon, r *http.Request) response.Response {
 			}
 
 			// Start the instance.
-			metadata := make(map[string]any)
-			metadata["evacuation_progress"] = fmt.Sprintf("Starting %q in project %q", inst.Name(), inst.Project().Name)
-			_ = op.UpdateMetadata(metadata)
+			_ = op.ExtendMetadata(map[string]any{"evacuation_progress": fmt.Sprintf("Starting %q in project %q", inst.Name(), inst.Project().Name)})
 
 			// If configured for stateful stop, try restoring its state.
 			action := inst.CanMigrate()
@@ -396,15 +390,13 @@ func restoreClusterMemberFunc(inst instance.Instance, op *operations.Operation,
 	var err error
 	var source incus.InstanceServer
 	var sourceNode db.NodeInfo
-	metadata := make(map[string]any)
 
 	l := logger.AddContext(logger.Ctx{"project": inst.Project().Name, "instance": inst.Name()})
 
 	// Check the action.
 	live := inst.CanMigrate() == "live-migrate"
 
-	metadata["evacuation_progress"] = fmt.Sprintf("Migrating %q in project %q from %q", inst.Name(), inst.Project().Name, inst.Location())
-	_ = op.UpdateMetadata(metadata)
+	_ = op.ExtendMetadata(map[string]any{"evacuation_progress": fmt.Sprintf("Migrating %q in project %q from %q", inst.Name(), inst.Project().Name, inst.Location())})
 
 	err = s.DB.Cluster.Transaction(context.Background(), func(ctx context.Context, tx *db.ClusterTx) error {
 		sourceNode, err = tx.GetNodeByName(ctx, inst.Location())
@@ -432,8 +424,7 @@ func restoreClusterMemberFunc(inst instance.Instance, op *operations.Operation,
 
 	isRunning := apiInst.StatusCode == api.Running
 	if isRunning && !live {
-		metadata["evacuation_progress"] = fmt.Sprintf("Stopping %q in project %q", inst.Name(), inst.Project().Name)
-		_ = op.UpdateMetadata(metadata)
+		_ = op.ExtendMetadata(map[string]any{"evacuation_progress": fmt.Sprintf("Stopping %q in project %q", inst.Name(), inst.Project().Name)})
 
 		timeout := inst.ExpandedConfig()["boot.host_shutdown_timeout"]
 		val, err := strconv.Atoi(timeout)
@@ -514,8 +505,7 @@ func restoreClusterMemberFunc(inst instance.Instance, op *operations.Operation,
 		return nil
 	}
 
-	metadata["evacuation_progress"] = fmt.Sprintf("Starting %q in project %q", inst.Name(), inst.Project().Name)
-	_ = op.UpdateMetadata(metadata)
+	_ = op.ExtendMetadata(map[string]any{"evacuation_progress": fmt.Sprintf("Starting %q in project %q", inst.Name(), inst.Project().Name)})
 
 	err = inst.Start(false)
 	if err != nil {
@@ -728,7 +718,7 @@ func healClusterMember(d *Daemon, op *operations.Operation, name string) error {
 	logger.Info("Starting cluster healing", logger.Ctx{"server": name})
 	defer logger.Info("Completed cluster healing", logger.Ctx{"server": name})
 
-	migrateFunc := func(ctx context.Context, s *state.State, inst instance.Instance, sourceMemberInfo *db.NodeInfo, targetMemberInfo *db.NodeInfo, live bool, startInstance bool, metadata map[string]any, op *operations.Operation) error {
+	migrateFunc := func(ctx context.Context, s *state.State, inst instance.Instance, sourceMemberInfo *db.NodeInfo, targetMemberInfo *db.NodeInfo, live bool, startInstance bool, op *operations.Operation) error {
 		// This returns an error if the instance's storage pool is local.
 		// Since we only care about remote backed instances, this can be ignored and return nil instead.
 		poolName, err := inst.StoragePool()

cmd/incusd/backup.go

@@ -152,14 +152,7 @@ func backupCreate(s *state.State, args db.InstanceBackup, sourceInst instance.In
 	backupProgressWriter := &ioprogress.ProgressWriter{
 		Tracker: &ioprogress.ProgressTracker{
 			Handler: func(value, speed int64) {
-				meta := op.Metadata()
-				if meta == nil {
-					meta = make(map[string]any)
-				}
-
-				progressText := fmt.Sprintf("%s (%s/s)", units.GetByteSizeString(value, 2), units.GetByteSizeString(speed, 2))
-				meta["create_backup_progress"] = progressText
-				_ = op.UpdateMetadata(meta)
+				_ = op.ExtendMetadata(map[string]any{"create_backup_progress": fmt.Sprintf("%s (%s/s)", units.GetByteSizeString(value, 2), units.GetByteSizeString(speed, 2))})
 			},
 		},
 	}

cmd/incusd/daemon_images.go

@@ -341,8 +341,7 @@ func ImageDownload(ctx context.Context, r *http.Request, s *state.State, op *ope
 		}
 
 		if meta["download_progress"] != progress.Text {
-			meta["download_progress"] = progress.Text
-			_ = op.UpdateMetadata(meta)
+			_ = op.ExtendMetadata(map[string]any{"download_progress": progress.Text})
 		}
 	}
 

cmd/incusd/images.go

@@ -278,7 +278,6 @@ func imgPostInstanceInfo(ctx context.Context, s *state.State, r *http.Request, r
 	}
 
 	// Track progress creating image.
-	metadata := make(map[string]any)
 	metaProgressWriter := &ioprogress.ProgressWriter{
 		Tracker: &ioprogress.ProgressTracker{
 			Handler: func(value, speed int64) {
@@ -292,6 +291,7 @@ func imgPostInstanceInfo(ctx context.Context, s *state.State, r *http.Request, r
 					processed = value
 				}
 
+				metadata := make(map[string]any)
 				operations.SetProgressMetadata(metadata, "create_image_from_container_pack", "Image pack", percent, processed, speed)
 				_ = op.UpdateMetadata(metadata)
 			},
@@ -312,6 +312,7 @@ func imgPostInstanceInfo(ctx context.Context, s *state.State, r *http.Request, r
 					processed = value
 				}
 
+				metadata := make(map[string]any)
 				operations.SetProgressMetadata(metadata, "create_image_from_container_pack", "Image pack", percent, processed, speed)
 				_ = op.UpdateMetadata(metadata)
 			},
@@ -406,6 +407,7 @@ func imgPostInstanceInfo(ctx context.Context, s *state.State, r *http.Request, r
 	// Tracker instance for the export phase.
 	tracker := &ioprogress.ProgressTracker{
 		Handler: func(value, speed int64) {
+			metadata := make(map[string]any)
 			operations.SetProgressMetadata(metadata, "create_image_from_container_pack", "Exporting", value, 0, 0)
 			_ = op.UpdateMetadata(metadata)
 		},

internal/server/instance/drivers/driver_common.go

@@ -857,8 +857,7 @@ func (d *common) updateProgress(progress string) {
 	}
 
 	if meta["container_progress"] != progress {
-		meta["container_progress"] = progress
-		_ = d.op.UpdateMetadata(meta)
+		_ = d.op.ExtendMetadata(map[string]any{"container_progress": progress})
 	}
 }
 

internal/server/instance/drivers/qmp/log.go

@@ -47,6 +47,10 @@ func (ql *qmpLog) open() error {
 
 // Write writes len(b) bytes from b to the channel.
 func (ql *qmpLog) Write(p []byte) (n int, err error) {
+	if ql == nil || ql.log == nil {
+		return 0, nil
+	}
+
 	ql.mu.Lock()
 	defer ql.mu.Unlock()
 
@@ -60,14 +64,14 @@ func (ql *qmpLog) Write(p []byte) (n int, err error) {
 
 // Close closes the log and wait the channel clean.
 func (ql *qmpLog) Close() error {
-	if ql.log != nil {
-		ql.mu.Lock()
-		defer ql.mu.Unlock()
-
-		err := ql.log.Close()
-		ql.log = nil
-		return err
+	if ql == nil || ql.log == nil {
+		return nil
 	}
 
-	return nil
+	ql.mu.Lock()
+	defer ql.mu.Unlock()
+
+	err := ql.log.Close()
+	ql.log = nil
+	return err
 }

internal/server/migration/migration_volumes.go

@@ -248,10 +248,7 @@ func MatchTypes(offer *migration.MigrationHeader, fallbackType migration.Migrati
 }
 
 func progressWrapperRender(op *operations.Operation, key string, description string, progressInt int64, speedInt int64) {
-	meta := op.Metadata()
-	if meta == nil {
-		meta = make(map[string]any)
-	}
+	meta := map[string]any{}
 
 	progress := fmt.Sprintf("%s (%s/s)", units.GetByteSizeString(progressInt, 2), units.GetByteSizeString(speedInt, 2))
 	if description != "" {
@@ -260,7 +257,7 @@ func progressWrapperRender(op *operations.Operation, key string, description str
 
 	if meta[key] != progress {
 		meta[key] = progress
-		_ = op.UpdateMetadata(meta)
+		_ = op.ExtendMetadata(meta)
 	}
 }