Merge pull request #12954 from swordqiu/automated-cherry-pick-of-#12938-upstream-release-3.7

Automated cherry pick of #12938: fix: server migration halted by an unexpected STOP event

pkg/compute/models/guest_actions.go

@@ -126,7 +126,7 @@ func (self *SGuest) PerformMonitor(
 	query jsonutils.JSONObject,
 	data jsonutils.JSONObject,
 ) (jsonutils.JSONObject, error) {
-	if utils.IsInStringArray(self.Status, []string{api.VM_RUNNING, api.VM_BLOCK_STREAM}) {
+	if utils.IsInStringArray(self.Status, []string{api.VM_RUNNING, api.VM_BLOCK_STREAM, api.VM_MIGRATING}) {
 		cmd, err := data.GetString("command")
 		if err != nil {
 			return nil, httperrors.NewMissingParameterError("command")

pkg/hostman/guestman/guestman.go

@@ -925,7 +925,7 @@ func (m *SGuestManager) DeleteSnapshot(ctx context.Context, params interface{})
 
 func (m *SGuestManager) Resume(ctx context.Context, sid string, isLiveMigrate bool) (jsonutils.JSONObject, error) {
 	guest, _ := m.GetServer(sid)
-	resumeTask := NewGuestResumeTask(ctx, guest)
+	resumeTask := NewGuestResumeTask(ctx, guest, !isLiveMigrate)
 	if isLiveMigrate {
 		guest.StartPresendArp()
 	}

pkg/hostman/guestman/guesttasks.go

@@ -475,12 +475,17 @@ type SGuestLiveMigrateTask struct {
 	params *SLiveMigrate
 
 	c chan struct{}
+
+	timeoutAt        time.Time
+	doTimeoutMigrate bool
 }
 
 func NewGuestLiveMigrateTask(
 	ctx context.Context, guest *SKVMGuestInstance, params *SLiveMigrate,
 ) *SGuestLiveMigrateTask {
-	return &SGuestLiveMigrateTask{SKVMGuestInstance: guest, ctx: ctx, params: params}
+	task := &SGuestLiveMigrateTask{SKVMGuestInstance: guest, ctx: ctx, params: params}
+	task.migrateTask = task
+	return task
 }
 
 func (s *SGuestLiveMigrateTask) Start() {
@@ -489,18 +494,27 @@ func (s *SGuestLiveMigrateTask) Start() {
 
 func (s *SGuestLiveMigrateTask) onSetZeroBlocks(res string) {
 	if strings.Contains(strings.ToLower(res), "error") {
-		hostutils.TaskFailed(s.ctx, fmt.Sprintf("Migrate set capability error: %s", res))
+		hostutils.TaskFailed(s.ctx, fmt.Sprintf("Migrate set capability zero-blocks error: %s", res))
 		return
 	}
+	// https://wiki.qemu.org/Features/AutoconvergeLiveMigration
 	s.Monitor.MigrateSetCapability("auto-converge", "on", s.startMigrate)
 }
 
 func (s *SGuestLiveMigrateTask) startMigrate(res string) {
 	if strings.Contains(strings.ToLower(res), "error") {
-		hostutils.TaskFailed(s.ctx, fmt.Sprintf("Migrate set capability error: %s", res))
+		s.migrateTask = nil
+		hostutils.TaskFailed(s.ctx, fmt.Sprintf("Migrate set capability auto-converge error: %s", res))
 		return
 	}
 
+	memMb, _ := s.Desc.Int("mem")
+	migSeconds := int(memMb) / options.HostOptions.MigrateExpectRate
+	if migSeconds < options.HostOptions.MinMigrateTimeoutSeconds {
+		migSeconds = options.HostOptions.MinMigrateTimeoutSeconds
+	}
+	log.Infof("migrate timeout seconds: %d", migSeconds)
+	s.timeoutAt = time.Now().Add(time.Second * time.Duration(migSeconds))
 	var copyIncremental = false
 	if s.params.IsLocal {
 		copyIncremental = true
@@ -511,6 +525,7 @@ func (s *SGuestLiveMigrateTask) startMigrate(res string) {
 
 func (s *SGuestLiveMigrateTask) startMigrateStatusCheck(res string) {
 	if strings.Contains(strings.ToLower(res), "error") {
+		s.migrateTask = nil
 		hostutils.TaskFailed(s.ctx, fmt.Sprintf("Migrate error: %s", res))
 		return
 	}
@@ -521,7 +536,7 @@ func (s *SGuestLiveMigrateTask) startMigrateStatusCheck(res string) {
 		case <-s.c: // on c close
 			s.c = nil
 			break
-		case <-time.After(time.Second * 1):
+		case <-time.After(time.Second * 5):
 			s.Monitor.GetMigrateStatus(s.onGetMigrateStatus)
 		}
 	}
@@ -529,14 +544,40 @@ func (s *SGuestLiveMigrateTask) startMigrateStatusCheck(res string) {
 
 func (s *SGuestLiveMigrateTask) onGetMigrateStatus(status string) {
 	if status == "completed" {
-		close(s.c)
-		hostutils.TaskComplete(s.ctx, nil)
+		s.migrateComplete()
 	} else if status == "failed" {
+		s.migrateTask = nil
 		close(s.c)
 		hostutils.TaskFailed(s.ctx, fmt.Sprintf("Query migrate got status: %s", status))
+	} else if !s.doTimeoutMigrate {
+		if s.timeoutAt.After(time.Now()) {
+			log.Warningf("migrate timeout, force stop to finish migrate")
+			// timeout, start memory postcopy
+			// https://wiki.qemu.org/Features/PostCopyLiveMigration
+			s.Monitor.SimpleCommand("stop", s.onMigrateStartPostcopy)
+			s.doTimeoutMigrate = true
+		}
 	}
 }
 
+func (s *SGuestLiveMigrateTask) onMigrateStartPostcopy(res string) {
+	if strings.Contains(strings.ToLower(res), "error") {
+		s.migrateTask = nil
+		close(s.c)
+		hostutils.TaskFailed(s.ctx, fmt.Sprintf("onMigrateStartPostcopy error: %s", res))
+		return
+	} else {
+		log.Infof("onMigrateStartPostcopy success")
+	}
+}
+
+func (s *SGuestLiveMigrateTask) migrateComplete() {
+	s.migrateTask = nil
+	close(s.c)
+	s.Monitor.Disconnect()
+	hostutils.TaskComplete(s.ctx, nil)
+}
+
 /**
  *  GuestResumeTask
 **/
@@ -546,12 +587,15 @@ type SGuestResumeTask struct {
 
 	ctx       context.Context
 	startTime time.Time
+
+	isTimeout bool
 }
 
-func NewGuestResumeTask(ctx context.Context, s *SKVMGuestInstance) *SGuestResumeTask {
+func NewGuestResumeTask(ctx context.Context, s *SKVMGuestInstance, isTimeout bool) *SGuestResumeTask {
 	return &SGuestResumeTask{
 		SKVMGuestInstance: s,
 		ctx:               ctx,
+		isTimeout:         isTimeout,
 	}
 }
 
@@ -578,10 +622,16 @@ func (s *SGuestResumeTask) onConfirmRunning(status string) {
 	} else if strings.Contains(status, "paused") {
 		s.Monitor.GetBlocks(s.onGetBlockInfo)
 	} else {
-		if time.Now().Sub(s.startTime) >= time.Second*60 {
+		memMb, _ := s.Desc.Int("mem")
+		migSeconds := int(memMb) / options.HostOptions.MigrateExpectRate
+		if migSeconds < options.HostOptions.MinMigrateTimeoutSeconds {
+			migSeconds = options.HostOptions.MinMigrateTimeoutSeconds
+		}
+		log.Infof("start guest timeout seconds: %d", migSeconds)
+		if s.isTimeout && time.Now().Sub(s.startTime) >= time.Second*time.Duration(migSeconds) {
 			s.taskFailed("Timeout")
 		} else {
-			time.Sleep(time.Second * 1)
+			time.Sleep(time.Second * 3)
 			s.confirmRunning()
 		}
 	}

pkg/hostman/guestman/qemu-kvm.go

@@ -68,6 +68,7 @@ type SKVMGuestInstance struct {
 	Monitor     monitor.Monitor
 	manager     *SGuestManager
 	startupTask *SGuestResumeTask
+	migrateTask *SGuestLiveMigrateTask
 	stopping    bool
 	syncMeta    *jsonutils.JSONDict
 }
@@ -500,6 +501,11 @@ func (s *SKVMGuestInstance) onReceiveQMPEvent(event *monitor.Event) {
 		// 	modules.Servers.PerformAction(
 		// 		hostutils.GetComputeSession(context.Background()),
 		// 		s.GetId(), "event", params)
+	case event.Event == `"STOP"`:
+		if s.migrateTask != nil {
+			// migrating complete
+			s.migrateTask.migrateComplete()
+		}
 	}
 }
 
@@ -535,7 +541,7 @@ func (s *SKVMGuestInstance) onGetQemuVersion(ctx context.Context, version string
 	} else if s.IsMaster() {
 		s.startDiskBackupMirror(ctx)
 		if ctx != nil && len(appctx.AppContextTaskId(ctx)) > 0 {
-			s.DoResumeTask(ctx)
+			s.DoResumeTask(ctx, false)
 		} else {
 			if options.HostOptions.SetVncPassword {
 				s.SetVncPassword()
@@ -543,7 +549,7 @@ func (s *SKVMGuestInstance) onGetQemuVersion(ctx context.Context, version string
 			s.OnResumeSyncMetadataInfo()
 		}
 	} else {
-		s.DoResumeTask(ctx)
+		s.DoResumeTask(ctx, true)
 	}
 }
 
@@ -742,8 +748,8 @@ func (s *SKVMGuestInstance) saveVncPort(port int) error {
 	return fileutils2.FilePutContents(s.GetVncFilePath(), fmt.Sprintf("%d", port), false)
 }
 
-func (s *SKVMGuestInstance) DoResumeTask(ctx context.Context) {
-	s.startupTask = NewGuestResumeTask(ctx, s)
+func (s *SKVMGuestInstance) DoResumeTask(ctx context.Context, isTimeout bool) {
+	s.startupTask = NewGuestResumeTask(ctx, s, isTimeout)
 	s.startupTask.Start()
 }
 

pkg/hostman/monitor/hmp.go

@@ -341,6 +341,14 @@ func (m *HmpMonitor) GetMigrateStatus(callback StringCallback) {
 	m.Query("info migrate", cb)
 }
 
+func (m *HmpMonitor) MigrateStartPostcopy(callback StringCallback) {
+	cb := func(output string) {
+		log.Infof("MigrateStartPostcopy: %s", output)
+		callback(output)
+	}
+	m.Query("migrate_start_postcopy", cb)
+}
+
 func (m *HmpMonitor) GetBlockJobCounts(callback func(jobs int)) {
 	cb := func(output string) {
 		lines := strings.Split(strings.TrimSuffix(output, "\r\n"), "\r\n")

pkg/hostman/monitor/monitor.go

@@ -20,10 +20,9 @@ import (
 	"sync"
 	"time"
 
-	"github.com/pkg/errors"
-
 	"yunion.io/x/jsonutils"
 	"yunion.io/x/log"
+	"yunion.io/x/pkg/errors"
 )
 
 type StringCallback func(string)
@@ -65,6 +64,7 @@ type Monitor interface {
 	MigrateSetCapability(capability, state string, callback StringCallback)
 	Migrate(destStr string, copyIncremental, copyFull bool, callback StringCallback)
 	GetMigrateStatus(callback StringCallback)
+	MigrateStartPostcopy(callback StringCallback)
 
 	ReloadDiskBlkdev(device, path string, callback StringCallback)
 	SetVncPassword(proto, password string, callback StringCallback)

pkg/hostman/monitor/qmp.go

@@ -636,6 +636,27 @@ func (m *QmpMonitor) GetMigrateStatus(callback StringCallback) {
 	m.Query(cmd, cb)
 }
 
+func (m *QmpMonitor) MigrateStartPostcopy(callback StringCallback) {
+	var (
+		cmd = &Command{Execute: "migrate-start-postcopy"}
+		cb  = func(res *Response) {
+			if res.ErrorVal != nil {
+				callback(res.ErrorVal.Error())
+			} else {
+				ret, err := jsonutils.Parse(res.Return)
+				if err != nil {
+					log.Errorf("Parse qmp res error: %s", err)
+					callback("MigrateStartPostcopy error")
+				} else {
+					log.Infof("MigrateStartPostcopy: %s", ret.String())
+					callback("")
+				}
+			}
+		}
+	)
+	m.Query(cmd, cb)
+}
+
 func (m *QmpMonitor) GetBlockJobCounts(callback func(jobs int)) {
 	var cb = func(res *Response) {
 		if res.ErrorVal != nil {

pkg/hostman/options/options.go

@@ -96,7 +96,11 @@ type SHostOptions struct {
 	PingRegionInterval     int      `default:"60" help:"interval to ping region, deefault is 1 minute"`
 	ManageNtpConfiguration bool     `default:"true"`
 	LogSystemdUnits        []string `help:"Systemd units log collected by fluent-bit"`
-	BandwidthLimit         int      `default:"50" help:"Bandwidth upper bound when migrating disk image in MB/sec"`
+	// 更改默认带宽限速为400GBps, qiujian
+	BandwidthLimit int `default:"400000" help:"Bandwidth upper bound when migrating disk image in MB/sec, default 400GBps"`
+	// 热迁移带宽，预期不低于8MBps, 1G Memory takes 128 seconds
+	MigrateExpectRate        int `default:"8" help:"Expected memory migration rate in MB/sec, default 8MBps"`
+	MinMigrateTimeoutSeconds int `default:"30" help:"minimal timeout for a migration process, default 30 seconds"`
 
 	SnapshotDirSuffix  string `help:"Snapshot dir name equal diskId concat snapshot dir suffix" default:"_snap"`
 	SnapshotRecycleDay int    `default:"1" help:"Snapshot Recycle delete Duration day"`

pkg/util/sysutils/sysutils.go

@@ -83,6 +83,8 @@ func ParseCPUInfo(lines []string) (*types.SCPUInfo, error) {
 		freq      string
 		cache     string
 		microcode string
+		part      string
+		revision  string
 	)
 	lv := func(line string) string {
 		return strings.TrimSpace(line[strings.Index(line, ":")+1:])
@@ -100,10 +102,19 @@ func ParseCPUInfo(lines []string) (*types.SCPUInfo, error) {
 		if len(microcode) == 0 && strings.HasPrefix(line, "microcode") {
 			microcode = lv(line)
 		}
+		if len(part) == 0 && strings.HasPrefix(line, "CPU part") {
+			part = lv(line)
+		}
+		if len(revision) == 0 && strings.HasPrefix(line, "CPU revision") {
+			revision = lv(line)
+		}
 		if strings.HasPrefix(line, "processor") {
 			cnt += 1
 		}
 	}
+	if len(model) == 0 {
+		model = part
+	}
 	if len(model) == 0 {
 		log.Errorf("Failed to get cpu model")
 	}
@@ -113,6 +124,9 @@ func ParseCPUInfo(lines []string) (*types.SCPUInfo, error) {
 	if len(cache) == 0 {
 		log.Errorf("Failed to get cpu cache size")
 	}
+	if len(microcode) == 0 {
+		microcode = revision
+	}
 	model = strings.TrimSpace(model)
 	info := &types.SCPUInfo{
 		Count:     cnt,

pkg/util/sysutils/sysutils_test.go

@@ -159,6 +159,25 @@ func TestParseCPUInfo(t *testing.T) {
 			},
 			wantErr: false,
 		},
+		{
+			name: "arm",
+			args: args{lines: []string{
+				"processor\t: 62",
+				"BogoMIPS\t: 200.00",
+				"Features\t: fp asimd evtstrm aes pmull sha1 sha2 crc32 atomics fphp asimdhp cpuid asimdrdm jscvt fcma dcpop",
+				"CPU implementer\t: 0x48",
+				"CPU architecture: 8",
+				"CPU variant\t: 0x1",
+				"CPU part\t: 0xd01",
+				"CPU revision\t: 0",
+			}},
+			want: &types.SCPUInfo{
+				Model:     "0xd01",
+				Microcode: "0",
+				Count:     1,
+			},
+			wantErr: false,
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {