fix(region,host): virtio dirver local disk live migrate

Guest using the virtio-blk driver for local disks, iothread is used to
handle IO operations, Qemu will segfault when copying disks during live
migration. To address this issue, use nbd server and drive_mirror
approach to copy the disks.

pkg/compute/tasks/guest_live_migrate_task.go

@@ -473,6 +473,11 @@ func (self *GuestLiveMigrateTask) OnStartDestComplete(ctx context.Context, guest
 		self.TaskFailed(ctx, guest, jsonutils.NewString(fmt.Sprintf("Get migrate port error: %s", err)))
 		return
 	}
+	nbdServerPort, err := data.Get("nbd_server_port")
+	if err != nil {
+		self.TaskFailed(ctx, guest, jsonutils.NewString(fmt.Sprintf("Get nbd server port error: %s", err)))
+		return
+	}
 
 	targetHostId, _ := self.Params.GetString("target_host_id")
 	targetHost := models.HostManager.FetchHostById(targetHostId)
@@ -481,6 +486,7 @@ func (self *GuestLiveMigrateTask) OnStartDestComplete(ctx context.Context, guest
 	isLocalStorage, _ := self.Params.Get("is_local_storage")
 	body.Set("is_local_storage", isLocalStorage)
 	body.Set("live_migrate_dest_port", liveMigrateDestPort)
+	body.Set("nbd_server_port", nbdServerPort)
 	body.Set("dest_ip", jsonutils.NewString(targetHost.AccessIp))
 	body.Set("enable_tls", jsonutils.NewBool(jsonutils.QueryBoolean(self.GetParams(), "enable_tls", false)))
 	body.Set("quickly_finish", jsonutils.NewBool(jsonutils.QueryBoolean(self.GetParams(), "quickly_finish", false)))

pkg/hostman/guestman/guesthandlers/guesthandler.go

@@ -443,6 +443,10 @@ func guestLiveMigrate(ctx context.Context, userCred mcclient.TokenCredential, si
 	if err != nil {
 		return nil, httperrors.NewMissingParameterError("live_migrate_dest_port")
 	}
+	nbdServerPort, err := body.Int("nbd_server_port")
+	if err != nil {
+		return nil, httperrors.NewMissingParameterError("live_migrate_dest_port")
+	}
 	destIp, err := body.GetString("dest_ip")
 	if err != nil {
 		return nil, httperrors.NewMissingParameterError("dest_ip")
@@ -456,6 +460,7 @@ func guestLiveMigrate(ctx context.Context, userCred mcclient.TokenCredential, si
 	params := &guestman.SLiveMigrate{
 		Sid:           sid,
 		DestPort:      int(destPort),
+		NbdServerPort: int(nbdServerPort),
 		DestIp:        destIp,
 		IsLocal:       isLocal,
 		EnableTLS:     enableTLS,

pkg/hostman/guestman/guesthelper.go

@@ -77,6 +77,7 @@ type SDestPrepareMigrate struct {
 type SLiveMigrate struct {
 	Sid            string
 	DestPort       int
+	NbdServerPort  int
 	DestIp         string
 	IsLocal        bool
 	EnableTLS      bool

pkg/hostman/guestman/guestman.go

@@ -1210,17 +1210,23 @@ func (m *SGuestManager) Resume(ctx context.Context, sid string, isLiveMigrate bo
 	if guest.IsStopping() || guest.IsStopped() {
 		return nil, httperrors.NewInvalidStatusError("resume stopped server???")
 	}
-	var cb = func() {
+
+	var onLiveMigrateCleanup = func(res string) {
+		guest.DoResumeTask(ctx, !isLiveMigrate, cleanTLS)
+	}
+	var onMonitorConnected = func() {
 		if isLiveMigrate {
 			guest.StartPresendArp()
+			guest.Monitor.StopNbdServer(onLiveMigrateCleanup)
+		} else {
+			onLiveMigrateCleanup("")
 		}
-		guest.DoResumeTask(ctx, !isLiveMigrate, cleanTLS)
 	}
 	if guest.Monitor == nil {
 		guest.StartMonitor(ctx, nil)
 		return nil, nil
 	} else {
-		cb()
+		onMonitorConnected()
 	}
 	return nil, nil
 }

pkg/hostman/guestman/guesttasks.go

@@ -29,6 +29,7 @@ import (
 	"yunion.io/x/pkg/appctx"
 	"yunion.io/x/pkg/errors"
 	"yunion.io/x/pkg/util/version"
+	"yunion.io/x/pkg/utils"
 
 	api "yunion.io/x/onecloud/pkg/apis/compute"
 	hostapi "yunion.io/x/onecloud/pkg/apis/host"
@@ -1002,8 +1003,12 @@ type SGuestLiveMigrateTask struct {
 	timeoutAt        time.Time
 	doTimeoutMigrate bool
 
-	expectDowntime int64
-	dirtySyncCount int64
+	expectDowntime        int64
+	dirtySyncCount        int64
+	diskDriverMirrorIndex int
+
+	onBlockJobsCancelled func()
+	totalTransferMb      int64
 }
 
 func NewGuestLiveMigrateTask(
@@ -1012,6 +1017,13 @@ func NewGuestLiveMigrateTask(
 	task := &SGuestLiveMigrateTask{SKVMGuestInstance: guest, ctx: ctx, params: params}
 	task.expectDowntime = 300 // qemu default downtime 300ms
 	task.MigrateTask = task
+
+	task.totalTransferMb = task.Desc.Mem
+	for i := 0; i < len(task.Desc.Disks); i++ {
+		if utils.IsInStringArray(task.Desc.Disks[i].StorageType, api.STORAGE_LOCAL_TYPES) {
+			task.totalTransferMb += int64(task.Desc.Disks[i].Size)
+		}
+	}
 	return task
 }
 
@@ -1031,7 +1043,25 @@ func (s *SGuestLiveMigrateTask) onSetZeroBlocks(res string) {
 
 func (s *SGuestLiveMigrateTask) onSetAutoConverge(res string) {
 	if strings.Contains(strings.ToLower(res), "error") {
-		s.migrateFailed(fmt.Sprintf("Migrate set capability auto-converge error: %s", res))
+		s.migrateFailed(fmt.Sprintf("Migrate set capability zero-blocks error: %s", res))
+		return
+	}
+
+	// https://wiki.qemu.org/Features/AutoconvergeLiveMigration
+	s.Monitor.MigrateSetCapability("events", "on", s.onMigrateEnableEvents)
+}
+
+func (s SGuestLiveMigrateTask) onMigrateEnableEvents(res string) {
+	if strings.Contains(strings.ToLower(res), "error") {
+		s.migrateFailed(fmt.Sprintf("Migrate set capability events error: %s", res))
+		return
+	}
+	s.Monitor.MigrateSetCapability("pause-before-switchover", "on", s.onMigrateSetPauseBeforeSwitchover)
+}
+
+func (s *SGuestLiveMigrateTask) onMigrateSetPauseBeforeSwitchover(res string) {
+	if strings.Contains(strings.ToLower(res), "error") {
+		s.migrateFailed(fmt.Sprintf("Migrate set capability pause-before-switchover error: %s", res))
 		return
 	}
 
@@ -1040,20 +1070,21 @@ func (s *SGuestLiveMigrateTask) onSetAutoConverge(res string) {
 		return
 	}
 
-	cb := func(res string) {
-		if strings.Contains(strings.ToLower(res), "error") {
-			s.migrateFailed(fmt.Sprintf("Migrate set capability auto-converge error: %s", res))
-			return
-		}
-		s.startMigrate()
-	}
 	if s.params.EnableTLS {
-		s.Monitor.MigrateSetCapability("multifd", "off", cb)
+		s.Monitor.MigrateSetCapability("multifd", "off", s.onSetMulitfd)
 		return
 	}
 
 	log.Infof("migrate src guest enable multifd")
-	s.Monitor.MigrateSetCapability("multifd", "on", cb)
+	s.Monitor.MigrateSetCapability("multifd", "on", s.onSetMulitfd)
+}
+
+func (s *SGuestLiveMigrateTask) onSetMulitfd(res string) {
+	if strings.Contains(strings.ToLower(res), "error") {
+		s.migrateFailed(fmt.Sprintf("Migrate set capability multifd error: %s", res))
+		return
+	}
+	s.startMigrate()
 }
 
 func (s *SGuestLiveMigrateTask) startRamMigrateTimeout() {
@@ -1106,14 +1137,64 @@ func (s *SGuestLiveMigrateTask) startMigrate() {
 	}
 }
 
-func (s *SGuestLiveMigrateTask) doMigrate() {
-	var copyIncremental = false
-	if s.params.IsLocal {
-		// copy disk data
-		copyIncremental = true
+func (s *SGuestLiveMigrateTask) waitMirrorJobsReady() {
+	cb := func(jobs []monitor.BlockJob) {
+		var allReady = true
+		var remaining = s.Desc.Mem * 1024 * 1024
+		var mbps float64
+		for i := 0; i < len(jobs); i++ {
+			if jobs[i].Status != "ready" {
+				allReady = false
+				remaining += (jobs[i].Len - jobs[i].Offset)
+				mbps += float64(jobs[i].Speed) / 1024 / 1024
+			}
+		}
+
+		if !allReady {
+			progress := (1 - float64(remaining)/float64(s.totalTransferMb*1024*1024)) * 100.0
+			hostutils.UpdateServerProgress(context.Background(), s.Id, progress, mbps)
+			time.Sleep(time.Second * 3)
+			s.waitMirrorJobsReady()
+			return
+		}
+		s.Monitor.Migrate(fmt.Sprintf("tcp:%s:%d", s.params.DestIp, s.params.DestPort),
+			false, false, s.setMaxBandwidth)
 	}
-	s.Monitor.Migrate(fmt.Sprintf("tcp:%s:%d", s.params.DestIp, s.params.DestPort),
-		copyIncremental, false, s.setMaxBandwidth)
+	s.Monitor.GetBlockJobs(cb)
+}
+
+func (s *SGuestLiveMigrateTask) mirrorDisks(res string) {
+	if len(res) > 0 {
+		log.Errorf("disk %d driver mirror failed %s", s.diskDriverMirrorIndex, res)
+		s.onDriveMirrorDisksFailed(res)
+		return
+	}
+	if s.diskDriverMirrorIndex == len(s.Desc.Disks) {
+		s.waitMirrorJobsReady()
+		return
+	}
+
+	i := s.diskDriverMirrorIndex
+	s.diskDriverMirrorIndex += 1
+	if utils.IsInStringArray(s.Desc.Disks[i].StorageType, api.STORAGE_LOCAL_TYPES) {
+		var drive = fmt.Sprintf("drive_%d", s.Desc.Disks[i].Index)
+		var target = fmt.Sprintf("nbd:%s:%d:exportname=drive_%d", s.params.DestIp, s.params.NbdServerPort, s.Desc.Disks[i].Index)
+		var speed int64 = 0
+		if s.params.MaxBandwidthMB != nil {
+			speed = *s.params.MaxBandwidthMB * 1024 * 1024
+		}
+		s.Monitor.DriveMirror(s.mirrorDisks, drive, target, "top", s.Desc.Disks[i].Format, true, false, speed)
+	} else {
+		s.mirrorDisks("")
+	}
+}
+
+func (s *SGuestLiveMigrateTask) onDriveMirrorDisksFailed(res string) {
+	s.migrateFailed(fmt.Sprintf("Migrate error: %s", res))
+}
+
+func (s *SGuestLiveMigrateTask) doMigrate() {
+	s.mirrorDisks("")
 }
 
 func (s *SGuestLiveMigrateTask) setMaxBandwidth(res string) {
@@ -1176,28 +1257,21 @@ func (s *SGuestLiveMigrateTask) onGetMigrateStatus(stats *monitor.MigrationInfo)
 	if status == "completed" {
 		jsonStats := jsonutils.Marshal(stats)
 		log.Infof("migration info %s", jsonStats)
-		s.migrateComplete(jsonStats)
 	} else if status == "failed" || status == "cancelled" {
 		s.migrateFailed(fmt.Sprintf("Query migrate got status: %s", status))
 	} else if status == "active" {
 		var (
-			ramTotal   int64
-			ramRemain  int64
-			mbps       float64
-			diskTotal  int64
-			diskRemain int64
+			ramRemain int64
+			mbps      float64
 		)
 		if stats.RAM != nil {
-			ramTotal = stats.RAM.Total
 			mbps = stats.RAM.Mbps
 			ramRemain = stats.RAM.Remaining
 		}
-		if stats.Disk != nil {
-			diskTotal = stats.Disk.Total
-			diskRemain = stats.Disk.Remaining
+		if ramRemain > 0 {
+			progress := (1 - float64(ramRemain)/float64(s.totalTransferMb*1024*1024)) * 100.0
+			hostutils.UpdateServerProgress(context.Background(), s.Id, progress, mbps)
 		}
-		progress := (1 - float64(diskRemain+ramRemain)/float64(diskTotal+ramTotal)) * 100.0
-		hostutils.UpdateServerProgress(context.Background(), s.Id, progress, mbps)
 
 		if s.params.QuicklyFinish && stats.RAM != nil && stats.RAM.Remaining > 0 {
 			if stats.CPUThrottlePercentage == nil {
@@ -1243,23 +1317,23 @@ func (s *SGuestLiveMigrateTask) onMigrateStartPostcopy(res string) {
 	}
 }
 
-func (s *SGuestLiveMigrateTask) onMigrateReceivedStopEvent() {
-	s.Monitor.GetMigrateStats(func(stats *monitor.MigrationInfo, err error) {
-		if err != nil {
-			log.Errorf("%s get migrate stats failed %s", s.GetName(), err)
-			return
-		}
+func (s *SGuestLiveMigrateTask) migrateContinueFromPreSwitchover() {
+	s.Monitor.MigrateContinue("pre-switchover", s.onMigrateContinue)
+}
 
-		switch *stats.Status {
-		case "completed":
-			s.migrateComplete(jsonutils.Marshal(stats))
-		case "failed", "cancelled":
-			s.migrateFailed(fmt.Sprintf("Query migrate got status: %s", *stats.Status))
-		case "active":
-			time.Sleep(10 * time.Millisecond)
-			s.onMigrateReceivedStopEvent()
-		}
-	})
+func (s *SGuestLiveMigrateTask) onMigrateContinue(res string) {
+	if len(res) > 0 {
+		s.migrateFailed(res)
+	}
+}
+
+func (s *SGuestLiveMigrateTask) onMigrateReceivedPreSwitchoverEvent() {
+	s.onBlockJobsCancelled = s.migrateContinueFromPreSwitchover
+	s.cancelBlockJobs("")
+}
+
+func (s *SGuestLiveMigrateTask) onMigrateReceivedBlockJobError(res string) {
+	s.migrateFailed(res)
 }
 
 func (s *SGuestLiveMigrateTask) migrateComplete(stats jsonutils.JSONObject) {
@@ -1278,7 +1352,30 @@ func (s *SGuestLiveMigrateTask) migrateComplete(stats jsonutils.JSONObject) {
 	hostutils.UpdateServerProgress(context.Background(), s.Id, 0.0, 0)
 }
 
+func (s *SGuestLiveMigrateTask) cancelBlockJobs(res string) {
+	log.Infof("%s cancel block jobs %s", s.GetName(), res)
+	if s.diskDriverMirrorIndex == 0 {
+		s.onBlockJobsCancelled()
+		return
+	}
+
+	s.diskDriverMirrorIndex -= 1
+	i := s.diskDriverMirrorIndex
+	if utils.IsInStringArray(s.Desc.Disks[i].StorageType, api.STORAGE_LOCAL_TYPES) {
+		s.Monitor.CancelBlockJob(fmt.Sprintf("drive_%d", s.Desc.Disks[i].Index), false, s.cancelBlockJobs)
+	} else {
+		s.cancelBlockJobs("")
+	}
+}
+
 func (s *SGuestLiveMigrateTask) migrateFailed(msg string) {
+	s.onBlockJobsCancelled = func() {
+		s.onMigrateFailBlockJobsCancelled(msg)
+	}
+	s.cancelBlockJobs("")
+}
+
+func (s *SGuestLiveMigrateTask) onMigrateFailBlockJobsCancelled(msg string) {
 	cleanup := func() {
 		s.MigrateTask = nil
 		if s.c != nil {
@@ -2056,8 +2153,7 @@ func (s *SDriveMirrorTask) startMirror(res string) {
 	}
 	if s.index < len(s.Desc.Disks) {
 		target := fmt.Sprintf("%s:exportname=drive_%d_backend", s.nbdUri, s.index)
-		s.Monitor.DriveMirror(s.startMirror, fmt.Sprintf("drive_%d", s.index),
-			target, s.syncMode, "", true, blockReplication)
+		s.Monitor.DriveMirror(s.startMirror, fmt.Sprintf("drive_%d", s.index), target, s.syncMode, "", true, blockReplication, 0)
 		s.index += 1
 	} else {
 		if s.onSucc != nil {
@@ -2675,15 +2771,7 @@ func (t *SGuestStorageCloneDiskTask) Start(guestRunning bool) {
 		t.diskIndex = diskIndex
 	}
 
-	t.Monitor.DriveMirror(
-		t.onDriveMirror,
-		fmt.Sprintf("drive_%d", diskIndex),
-		targetDisk.GetPath(),
-		"full",
-		targetDiskFormat,
-		true,
-		false,
-	)
+	t.Monitor.DriveMirror(t.onDriveMirror, fmt.Sprintf("drive_%d", diskIndex), targetDisk.GetPath(), "full", targetDiskFormat, true, false, 0)
 }
 
 func (t *SGuestStorageCloneDiskTask) onDriveMirror(res string) {
@@ -2719,15 +2807,7 @@ func (t *SGuestStorageCloneDiskTask) OnGetBlockJobs(jobs []monitor.BlockJob) {
 				)
 				return
 			}
-			t.Monitor.DriveMirror(
-				t.onDriveMirror,
-				fmt.Sprintf("drive_%d", t.diskIndex),
-				targetDisk.GetPath(),
-				"full",
-				targetDiskFormat,
-				true,
-				false,
-			)
+			t.Monitor.DriveMirror(t.onDriveMirror, fmt.Sprintf("drive_%d", t.diskIndex), targetDisk.GetPath(), "full", targetDiskFormat, true, false, 0)
 		} else {
 			hostutils.TaskFailed(t.ctx, fmt.Sprintf("Disk %s Block job not found", t.params.SourceDisk.GetId()))
 		}

pkg/hostman/guestman/qemu-kvm.go

@@ -807,19 +807,39 @@ func (s *SKVMGuestInstance) onReceiveQMPEvent(event *monitor.Event) {
 		s.eventGuestStop()
 	case `"QUORUM_REPORT_BAD"`:
 		s.eventQuorumReportBad(event)
+	case `"MIGRATION"`:
+		s.eventMigration(event)
+	}
+}
+
+func (s *SKVMGuestInstance) eventMigration(event *monitor.Event) {
+	if s.MigrateTask == nil {
+		return
+	}
+
+	status, ok := event.Data["status"]
+	if !ok {
+		return
+	}
+	if status == "pre-switchover" {
+		// migrating complete
+		s.MigrateTask.onMigrateReceivedPreSwitchoverEvent()
+		hostutils.UpdateServerProgress(context.Background(), s.Id, 0.0, 0)
+	} else if status == "completed" {
+		s.MigrateTask.migrateComplete(nil)
 	}
 }
 
 func (s *SKVMGuestInstance) eventBlockJobError(event *monitor.Event) {
-	s.SyncMirrorJobFailed(event.String())
+	if s.MigrateTask != nil {
+		s.MigrateTask.onMigrateReceivedBlockJobError(event.String())
+	} else {
+		s.SyncMirrorJobFailed(event.String())
+	}
 }
 
 func (s *SKVMGuestInstance) eventGuestStop() {
-	if s.MigrateTask != nil {
-		// migrating complete
-		s.MigrateTask.onMigrateReceivedStopEvent()
-	}
-	hostutils.UpdateServerProgress(context.Background(), s.Id, 0.0, 0)
+	// do nothing
 }
 
 func (s *SKVMGuestInstance) eventQuorumReportBad(event *monitor.Event) {
@@ -870,6 +890,10 @@ func (s *SKVMGuestInstance) eventGuestPaniced(event *monitor.Event) {
 }
 
 func (s *SKVMGuestInstance) eventBlockJobReady(event *monitor.Event) {
+	if !s.IsSlave() {
+		return
+	}
+
 	itype, ok := event.Data["type"]
 	if !ok {
 		log.Errorf("block job missing event type")
@@ -906,25 +930,23 @@ func (s *SKVMGuestInstance) eventBlockJobReady(event *monitor.Event) {
 		return
 	}
 
-	if s.IsSlave() { // is backup server
-		disk, err := storageman.GetManager().GetDiskByPath(diskPath)
-		if err != nil {
-			log.Errorf("eventBlockJobReady failed get disk %s", diskPath)
-			return
-		}
-		disk.PostCreateFromImageFuse()
-		blockJobCount := s.BlockJobsCount()
-		if blockJobCount == 0 {
-			for {
-				_, err := modules.Servers.PerformAction(
-					hostutils.GetComputeSession(context.Background()), s.GetId(), "slave-block-stream-ready", nil,
-				)
-				if err != nil {
-					log.Errorf("onReceiveQMPEvent sync slave block stream ready error: %s", err)
-					time.Sleep(3 * time.Second)
-				} else {
-					break
-				}
+	disk, err := storageman.GetManager().GetDiskByPath(diskPath)
+	if err != nil {
+		log.Errorf("eventBlockJobReady failed get disk %s", diskPath)
+		return
+	}
+	disk.PostCreateFromImageFuse()
+	blockJobCount := s.BlockJobsCount()
+	if blockJobCount == 0 {
+		for {
+			_, err := modules.Servers.PerformAction(
+				hostutils.GetComputeSession(context.Background()), s.GetId(), "slave-block-stream-ready", nil,
+			)
+			if err != nil {
+				log.Errorf("onReceiveQMPEvent sync slave block stream ready error: %s", err)
+				time.Sleep(3 * time.Second)
+			} else {
+				break
 			}
 		}
 	}
@@ -1010,6 +1032,20 @@ func (s *SKVMGuestInstance) migrateEnableMultifd() error {
 	return <-err
 }
 
+func (s *SKVMGuestInstance) migrateStartNbdServer(nbdServerPort int) error {
+	var err = make(chan error)
+	onNbdServerStarted := func(res string) {
+		if len(res) > 0 {
+			err <- errors.Errorf("failed enable multifd %s", res)
+		} else {
+			err <- nil
+		}
+	}
+	log.Infof("migrate dest guest start nbd server on %d", nbdServerPort)
+	s.Monitor.StartNbdServer(nbdServerPort, true, true, onNbdServerStarted)
+	return <-err
+}
+
 func (s *SKVMGuestInstance) onGetQemuVersion(ctx context.Context, version string) {
 	s.QemuVersion = version
 	log.Infof("Guest(%s) qemu version %s", s.Id, s.QemuVersion)
@@ -1151,6 +1187,15 @@ func (s *SKVMGuestInstance) guestRun(ctx context.Context) {
 			hostutils.TaskFailed(ctx, err.Error())
 			return
 		}
+		nbdServerPort := s.manager.GetNBDServerFreePort()
+		defer s.manager.unsetPort(nbdServerPort)
+		err = s.migrateStartNbdServer(nbdServerPort)
+		if err != nil {
+			hostutils.TaskFailed(ctx, err.Error())
+			return
+		}
+		body.Set("nbd_server_port", jsonutils.NewInt(int64(nbdServerPort)))
+
 		if s.LiveMigrateUseTls {
 			s.setDestMigrateTLS(ctx, body)
 		} else {

pkg/hostman/monitor/hmp.go

@@ -320,6 +320,11 @@ func (m *HmpMonitor) MigrateIncoming(address string, callback StringCallback) {
 	m.Query(cmd, callback)
 }
 
+func (m *HmpMonitor) MigrateContinue(state string, callback StringCallback) {
+	cmd := fmt.Sprintf("migrate_continue %s", state)
+	m.Query(cmd, callback)
+}
+
 func (m *HmpMonitor) Migrate(
 	destStr string, copyIncremental, copyFull bool, callback StringCallback,
 ) {
@@ -402,7 +407,7 @@ func (m *HmpMonitor) ReloadDiskBlkdev(device, path string, callback StringCallba
 	m.Query(fmt.Sprintf("reload_disk_snapshot_blkdev -n %s %s", device, path), callback)
 }
 
-func (m *HmpMonitor) DriveMirror(callback StringCallback, drive, target, syncMode, format string, unmap, blockReplication bool) {
+func (m *HmpMonitor) DriveMirror(callback StringCallback, drive, target, syncMode, format string, unmap, blockReplication bool, speed int64) {
 	cmd := "drive_mirror -n"
 	if blockReplication {
 		cmd += " -c"
@@ -467,6 +472,10 @@ func (m *HmpMonitor) StartNbdServer(port int, exportAllDevice, writable bool, ca
 	m.Query(cmd, callback)
 }
 
+func (m *HmpMonitor) StopNbdServer(callback StringCallback) {
+	m.Query("nbd_server_stop", callback)
+}
+
 func (m *HmpMonitor) ResizeDisk(driveName string, sizeMB int64, callback StringCallback) {
 	cmd := fmt.Sprintf("block_resize %s %d", driveName, sizeMB)
 	m.Query(cmd, callback)

pkg/hostman/monitor/monitor.go

@@ -223,7 +223,7 @@ type Monitor interface {
 
 	XBlockdevChange(parent, node, child string, callback StringCallback)
 	BlockStream(drive string, callback StringCallback)
-	DriveMirror(callback StringCallback, drive, target, syncMode, format string, unmap, blockReplication bool)
+	DriveMirror(callback StringCallback, drive, target, syncMode, format string, unmap, blockReplication bool, speed int64)
 	DriveBackup(callback StringCallback, drive, target, syncMode, format string)
 	BlockJobComplete(drive string, cb StringCallback)
 	BlockReopenImage(drive, newImagePath, format string, cb StringCallback)
@@ -234,6 +234,7 @@ type Monitor interface {
 	MigrateSetParameter(key string, val interface{}, callback StringCallback)
 	MigrateIncoming(address string, callback StringCallback)
 	Migrate(destStr string, copyIncremental, copyFull bool, callback StringCallback)
+	MigrateContinue(state string, callback StringCallback)
 	GetMigrateStatus(callback StringCallback)
 	MigrateStartPostcopy(callback StringCallback)
 	GetMigrateStats(callback MigrateStatsCallback)
@@ -241,6 +242,7 @@ type Monitor interface {
 	ReloadDiskBlkdev(device, path string, callback StringCallback)
 	SetVncPassword(proto, password string, callback StringCallback)
 	StartNbdServer(port int, exportAllDevice, writable bool, callback StringCallback)
+	StopNbdServer(callback StringCallback)
 
 	ResizeDisk(driveName string, sizeMB int64, callback StringCallback)
 	BlockIoThrottle(driveName string, bps, iops int64, callback StringCallback)

pkg/hostman/monitor/qmp.go

@@ -698,6 +698,11 @@ func (m *QmpMonitor) MigrateIncoming(address string, callback StringCallback) {
 	m.HumanMonitorCommand(cmd, callback)
 }
 
+func (m *QmpMonitor) MigrateContinue(state string, callback StringCallback) {
+	cmd := fmt.Sprintf("migrate_continue %s", state)
+	m.HumanMonitorCommand(cmd, callback)
+}
+
 func (m *QmpMonitor) Migrate(
 	destStr string, copyIncremental, copyFull bool, callback StringCallback,
 ) {
@@ -858,7 +863,7 @@ func (m *QmpMonitor) ReloadDiskBlkdev(device, path string, callback StringCallba
 	m.Query(cmd, cb)
 }
 
-func (m *QmpMonitor) DriveMirror(callback StringCallback, drive, target, syncMode, format string, unmap, blockReplication bool) {
+func (m *QmpMonitor) DriveMirror(callback StringCallback, drive, target, syncMode, format string, unmap, blockReplication bool, speed int64) {
 	var (
 		cb = func(res *Response) {
 			callback(m.actionResult(res))
@@ -871,6 +876,9 @@ func (m *QmpMonitor) DriveMirror(callback StringCallback, drive, target, syncMod
 			"unmap":  unmap,
 		}
 	)
+	if speed > 0 {
+		args["speed"] = speed
+	}
 	if blockReplication {
 		args["block-replication"] = true
 	}
@@ -951,6 +959,10 @@ func (m *QmpMonitor) StartNbdServer(port int, exportAllDevice, writable bool, ca
 	m.HumanMonitorCommand(cmd, callback)
 }
 
+func (m *QmpMonitor) StopNbdServer(callback StringCallback) {
+	m.HumanMonitorCommand("nbd_server_stop", callback)
+}
+
 func (m *QmpMonitor) ResizeDisk(driveName string, sizeMB int64, callback StringCallback) {
 	cmd := fmt.Sprintf("block_resize %s %d", driveName, sizeMB)
 	m.HumanMonitorCommand(cmd, callback)