diff --git a/cmd/climc/shell/compute/servers.go b/cmd/climc/shell/compute/servers.go index 2c4ca63705..cc16becc6a 100644 --- a/cmd/climc/shell/compute/servers.go +++ b/cmd/climc/shell/compute/servers.go @@ -67,6 +67,8 @@ func init() { cmd.PrintObjectYAML().Perform("migrate-forecast", new(options.ServerMigrateForecastOptions)) cmd.Perform("migrate", new(options.ServerMigrateOptions)) cmd.Perform("live-migrate", new(options.ServerLiveMigrateOptions)) + cmd.BatchPerform("cancel-live-migrate", new(options.ServerIdsOptions)) + cmd.Perform("set-live-migrate-params", new(options.ServerSetLiveMigrateParamsOptions)) cmd.Perform("modify-src-check", new(options.ServerModifySrcCheckOptions)) cmd.Perform("set-secgroup", new(options.ServerSecGroupsOptions)) cmd.Perform("add-secgroup", new(options.ServerSecGroupsOptions)) diff --git a/pkg/apis/compute/api.go b/pkg/apis/compute/api.go index 8c2b9e09b7..d90d1bef83 100644 --- a/pkg/apis/compute/api.go +++ b/pkg/apis/compute/api.go @@ -581,6 +581,8 @@ type GuestBatchMigrateRequest struct { SkipCpuCheck bool `json:"skip_cpu_check"` SkipKernelCheck bool `json:"skip_kernel_check"` EnableTLS *bool `json:"enable_tls"` + MaxBandwidthMb *int64 `json:"max_bandwidth_mb"` + QuciklyFinish *bool `json:"quickly_finish"` } type GuestBatchMigrateParams struct { @@ -591,6 +593,8 @@ type GuestBatchMigrateParams struct { EnableTLS *bool RescueMode bool OldStatus string + MaxBandwidthMb *int64 + QuciklyFinish *bool } type HostLoginInfo struct { diff --git a/pkg/apis/compute/guest_const.go b/pkg/apis/compute/guest_const.go index ea094c3470..5958bd057a 100644 --- a/pkg/apis/compute/guest_const.go +++ b/pkg/apis/compute/guest_const.go @@ -71,6 +71,7 @@ const ( VM_START_MIGRATE = "start_migrate" VM_MIGRATING = "migrating" + VM_LIVE_MIGRATING = "live_migrating" VM_MIGRATE_FAILED = "migrate_failed" VM_CHANGE_FLAVOR = "change_flavor" diff --git a/pkg/apis/compute/guests.go b/pkg/apis/compute/guests.go index 00219f889c..f946958f3b 100644 --- a/pkg/apis/compute/guests.go +++ b/pkg/apis/compute/guests.go @@ -440,6 +440,12 @@ type GuestLiveMigrateInput struct { SkipKernelCheck *bool `json:"skip_kernel_check"` // 是否启用 tls EnableTLS *bool `json:"enable_tls"` + + // 迁移带宽限制 + MaxBandwidthMb *int64 `json:"max_bandwidth_mb"` + // 快速完成，内存同步一定周期后调整 downtime + QuicklyFinish *bool `json:"quickly_finish"` + KeepDestGuestOnFailed *bool `json:"keep_dest_guest_on_failed"` } type GuestSetSecgroupInput struct { @@ -916,3 +922,8 @@ type ServerSetPasswordInput struct { ResetPassword bool AutoStart bool } + +type ServerSetLiveMigrateParamsInput struct { + MaxBandwidthMB *int64 + DowntimeLimitMS *int64 +} diff --git a/pkg/compute/guestdrivers/kvm.go b/pkg/compute/guestdrivers/kvm.go index 7e86b126a4..7f5c0f138b 100644 --- a/pkg/compute/guestdrivers/kvm.go +++ b/pkg/compute/guestdrivers/kvm.go @@ -776,6 +776,9 @@ func (self *SKVMGuestDriver) CheckLiveMigrate(ctx context.Context, guest *models return httperrors.NewBadRequestError("Guest have backup, can't migrate") } if utils.IsInStringArray(guest.Status, []string{api.VM_RUNNING, api.VM_SUSPEND}) { + if *input.MaxBandwidthMb < 50 { + return httperrors.NewBadRequestError("max bandwidth must gratethan 100M") + } cdrom := guest.GetCdrom() if cdrom != nil && len(cdrom.ImageId) > 0 { return httperrors.NewBadRequestError("Cannot live migrate with cdrom") diff --git a/pkg/compute/models/guest_actions.go b/pkg/compute/models/guest_actions.go index fbceb2108e..f7985c24cf 100644 --- a/pkg/compute/models/guest_actions.go +++ b/pkg/compute/models/guest_actions.go @@ -32,7 +32,6 @@ import ( "yunion.io/x/pkg/errors" "yunion.io/x/pkg/tristate" "yunion.io/x/pkg/util/fileutils" - "yunion.io/x/pkg/util/osprofile" "yunion.io/x/pkg/util/regutils" "yunion.io/x/pkg/util/sets" "yunion.io/x/pkg/utils" @@ -557,10 +556,18 @@ func (self *SGuest) PerformLiveMigrate(ctx context.Context, userCred mcclient.To if input.EnableTLS == nil { input.EnableTLS = &options.Options.EnableTlsMigration } - return nil, self.StartGuestLiveMigrateTask(ctx, userCred, self.Status, input.PreferHost, input.SkipCpuCheck, input.SkipKernelCheck, input.EnableTLS, "") + return nil, self.StartGuestLiveMigrateTask(ctx, userCred, + self.Status, input.PreferHost, input.SkipCpuCheck, + input.SkipKernelCheck, input.EnableTLS, input.QuicklyFinish, input.MaxBandwidthMb, input.KeepDestGuestOnFailed, "", + ) } -func (self *SGuest) StartGuestLiveMigrateTask(ctx context.Context, userCred mcclient.TokenCredential, guestStatus, preferHostId string, skipCpuCheck *bool, skipKernelCheck *bool, enableTLS *bool, parentTaskId string) error { +func (self *SGuest) StartGuestLiveMigrateTask( + ctx context.Context, userCred mcclient.TokenCredential, + guestStatus, preferHostId string, + skipCpuCheck, skipKernelCheck, enableTLS, quicklyFinish *bool, + maxBandwidthMb *int64, keepDestGuestOnFailed *bool, parentTaskId string, +) error { self.SetStatus(userCred, api.VM_START_MIGRATE, "") data := jsonutils.NewDict() if len(preferHostId) > 0 { @@ -575,6 +582,16 @@ func (self *SGuest) StartGuestLiveMigrateTask(ctx context.Context, userCred mccl if enableTLS != nil { data.Set("enable_tls", jsonutils.NewBool(*enableTLS)) } + if quicklyFinish != nil { + data.Set("quickly_finish", jsonutils.NewBool(*quicklyFinish)) + } + if maxBandwidthMb != nil { + data.Set("max_bandwidth_mb", jsonutils.NewInt(*maxBandwidthMb)) + } + if keepDestGuestOnFailed != nil { + data.Set("keep_dest_guest_on_failed", jsonutils.NewBool(*keepDestGuestOnFailed)) + } + data.Set("guest_status", jsonutils.NewString(guestStatus)) dedicateMigrateTask := "GuestLiveMigrateTask" if self.GetHypervisor() != api.HYPERVISOR_KVM { @@ -589,6 +606,48 @@ func (self *SGuest) StartGuestLiveMigrateTask(ctx context.Context, userCred mccl return nil } +func (self *SGuest) PerformSetLiveMigrateParams( + ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject, input api.ServerSetLiveMigrateParamsInput, +) (jsonutils.JSONObject, error) { + if self.Status != api.VM_LIVE_MIGRATING { + return nil, httperrors.NewServerStatusError("cannot set migrate params in status %s", self.Status) + } + if input.MaxBandwidthMB == nil && input.DowntimeLimitMS == nil { + return nil, httperrors.NewInputParameterError("empty input") + } + + arguments := map[string]interface{}{} + if input.MaxBandwidthMB != nil { + arguments["max-bandwidth"] = *input.MaxBandwidthMB * 1024 * 1024 + } + if input.DowntimeLimitMS != nil { + arguments["downtime-limit"] = *input.DowntimeLimitMS + } + cmd := map[string]interface{}{ + "execute": "migrate-set-parameters", + "arguments": arguments, + } + log.Infof("set live migrate params input: %s", jsonutils.Marshal(cmd).String()) + monitorInput := &api.ServerMonitorInput{ + COMMAND: jsonutils.Marshal(cmd).String(), + QMP: true, + } + return self.SendMonitorCommand(ctx, userCred, monitorInput) +} + +func (self *SGuest) PerformCancelLiveMigrate( + ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject, data jsonutils.JSONObject, +) (jsonutils.JSONObject, error) { + if self.Status != api.VM_LIVE_MIGRATING { + return nil, httperrors.NewServerStatusError("cannot set migrate params in status %s", self.Status) + } + monitorInput := &api.ServerMonitorInput{ + COMMAND: "migrate_cancel", + QMP: false, + } + return self.SendMonitorCommand(ctx, userCred, monitorInput) +} + func (self *SGuest) PerformClone(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject, data jsonutils.JSONObject) (jsonutils.JSONObject, error) { if self.IsEncrypted() { return nil, httperrors.NewForbiddenError("cannot clone encrypted server") @@ -666,36 +725,6 @@ func (self *SGuest) GetDetailsCreateParams(ctx context.Context, userCred mcclien return input.JSON(input), nil } -func (self *SGuest) PerformSetPassword(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject, input api.ServerSetPasswordInput) (jsonutils.JSONObject, error) { - switch self.Status { - case api.VM_RUNNING: - inputQga := &api.ServerQgaSetPasswordInput{ - Username: input.Username, - Password: input.Password, - } - if inputQga.Username == "" { - if self.OsType == osprofile.OS_TYPE_WINDOWS { - inputQga.Username = api.VM_DEFAULT_WINDOWS_LOGIN_USER - } else { - inputQga.Username = api.VM_DEFAULT_LINUX_LOGIN_USER - } - } - if inputQga.Password == "" && input.ResetPassword { - inputQga.Password = seclib2.RandomPassword2(12) - } - return self.PerformQgaSetPassword(ctx, userCred, query, inputQga) - case api.VM_READY: - inputDeploy := api.ServerDeployInput{ - Password: input.Password, - ResetPassword: input.ResetPassword, - AutoStart: input.AutoStart, - } - return self.PerformDeploy(ctx, userCred, query, inputDeploy) - default: - return nil, httperrors.NewServerStatusError("Cannot deploy in status %s", self.Status) - } -} - func (self *SGuest) saveOldPassword(ctx context.Context, userCred mcclient.TokenCredential) { loginKey := self.GetMetadata(ctx, api.VM_METADATA_LOGIN_KEY, userCred) if len(loginKey) > 0 { @@ -4675,6 +4704,8 @@ func (manager *SGuestManager) PerformBatchMigrate(ctx context.Context, userCred SkipCpuCheck: params.SkipCpuCheck, SkipKernelCheck: params.SkipKernelCheck, EnableTLS: params.EnableTLS, + MaxBandwidthMb: params.MaxBandwidthMb, + QuciklyFinish: params.QuciklyFinish, } guests[i].SetStatus(userCred, api.VM_START_MIGRATE, "batch migrate") if _, ok := hostGuests[guests[i].HostId]; ok { diff --git a/pkg/compute/tasks/guest_live_migrate_task.go b/pkg/compute/tasks/guest_live_migrate_task.go index 3cc9eae947..41e2b57f96 100644 --- a/pkg/compute/tasks/guest_live_migrate_task.go +++ b/pkg/compute/tasks/guest_live_migrate_task.go @@ -79,7 +79,11 @@ func (self *GuestMigrateTask) GetSchedParams() (*schedapi.ScheduleInput, error) func (self *GuestMigrateTask) OnStartSchedule(obj IScheduleModel) { guest := obj.(*models.SGuest) - guest.SetStatus(self.UserCred, api.VM_MIGRATING, "") + guestStatus, _ := self.Params.GetString("guest_status") + if guestStatus != api.VM_RUNNING && guestStatus != api.VM_SUSPEND { + guest.SetStatus(self.UserCred, api.VM_MIGRATING, "") + } + db.OpsLog.LogEvent(guest, db.ACT_MIGRATING, "", self.UserCred) } @@ -102,6 +106,8 @@ func (self *GuestMigrateTask) SaveScheduleResult(ctx context.Context, obj ISched return } db.OpsLog.LogEvent(guest, db.ACT_MIGRATING, fmt.Sprintf("guest start migrate from host %s to %s", guest.HostId, targetHostId), self.UserCred) + logclient.AddActionLogWithContext(ctx, guest, logclient.ACT_MIGRATING, + fmt.Sprintf("guest start migrate from host %s to %s(%s)", guest.HostId, targetHostId, targetHost.GetName()), self.UserCred, true) body := jsonutils.NewDict() body.Set("target_host_id", jsonutils.NewString(targetHostId)) @@ -461,12 +467,18 @@ func (self *GuestLiveMigrateTask) OnStartDestComplete(ctx context.Context, guest body.Set("live_migrate_dest_port", liveMigrateDestPort) 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))) + if self.Params.Contains("max_bandwidth_mb") { + maxBandwidthMb, _ := self.Params.Get("max_bandwidth_mb") + body.Set("max_bandwidth_mb", maxBandwidthMb) + } headers := self.GetTaskRequestHeader() host, _ := guest.GetHost() url := fmt.Sprintf("%s/servers/%s/live-migrate", host.ManagerUri, guest.Id) self.SetStage("OnLiveMigrateComplete", nil) + guest.SetStatus(self.UserCred, api.VM_LIVE_MIGRATING, "") _, _, err = httputils.JSONRequest(httputils.GetDefaultClient(), ctx, "POST", url, headers, body, false) if err != nil { @@ -511,12 +523,19 @@ func (self *GuestMigrateTask) setGuest(ctx context.Context, guest *models.SGuest } func (self *GuestLiveMigrateTask) OnLiveMigrateCompleteFailed(ctx context.Context, guest *models.SGuest, data jsonutils.JSONObject) { - targetHostId, _ := self.Params.GetString("target_host_id") - guest.StartUndeployGuestTask(ctx, self.UserCred, "", targetHostId) + if !jsonutils.QueryBoolean(self.Params, "keep_dest_guest_on_failed", false) { + targetHostId, _ := self.Params.GetString("target_host_id") + guest.StartUndeployGuestTask(ctx, self.UserCred, "", targetHostId) + } self.TaskFailed(ctx, guest, data) } func (self *GuestLiveMigrateTask) OnLiveMigrateComplete(ctx context.Context, guest *models.SGuest, data jsonutils.JSONObject) { + if migInfo, err := data.Get("migration_info"); err != nil { + db.OpsLog.LogEvent(guest, db.ACT_MIGRATE, migInfo, self.UserCred) + logclient.AddActionLogWithContext(ctx, guest, logclient.ACT_MIGRATE, migInfo, self.UserCred, true) + } + headers := self.GetTaskRequestHeader() body := jsonutils.NewDict() body.Set("live_migrate", jsonutils.JSONTrue) diff --git a/pkg/compute/tasks/guest_sync_task.go b/pkg/compute/tasks/guest_sync_task.go index dd6688bb1b..a1bf45299f 100644 --- a/pkg/compute/tasks/guest_sync_task.go +++ b/pkg/compute/tasks/guest_sync_task.go @@ -91,6 +91,9 @@ func (self *GuestSyncConfTask) OnDiskSyncCompleteFailed(ctx context.Context, obj guest := obj.(*models.SGuest) db.OpsLog.LogEvent(guest, db.ACT_SYNC_CONF_FAIL, data, self.UserCred) logclient.AddActionLogWithStartable(self, guest, logclient.ACT_VM_SYNC_CONF, data, self.UserCred, false) + if !jsonutils.QueryBoolean(self.Params, "without_sync_status", false) { + guest.SetStatus(self.GetUserCred(), api.VM_SYNC_FAIL, data.String()) + } self.SetStageFailed(ctx, data) } diff --git a/pkg/compute/tasks/host_guests_migrate_task.go b/pkg/compute/tasks/host_guests_migrate_task.go index 59f242bdba..5c5cfd8b76 100644 --- a/pkg/compute/tasks/host_guests_migrate_task.go +++ b/pkg/compute/tasks/host_guests_migrate_task.go @@ -48,8 +48,10 @@ func (self *HostGuestsMigrateTask) OnInit(ctx context.Context, objs []db.IStanda for i := range objs { guest := objs[i].(*models.SGuest) if guests[i].LiveMigrate { - err := guest.StartGuestLiveMigrateTask( - ctx, self.UserCred, guests[i].OldStatus, preferHostId, &guests[i].SkipCpuCheck, &guests[i].SkipKernelCheck, guests[i].EnableTLS, self.Id) + err := guest.StartGuestLiveMigrateTask(ctx, self.UserCred, + guests[i].OldStatus, preferHostId, &guests[i].SkipCpuCheck, &guests[i].SkipKernelCheck, + guests[i].EnableTLS, guests[i].QuciklyFinish, guests[i].MaxBandwidthMb, nil, self.Id, + ) if err != nil { log.Errorln(err) } diff --git a/pkg/hostman/guestman/desc/desc.go b/pkg/hostman/guestman/desc/desc.go index 53abf85277..4d78623580 100644 --- a/pkg/hostman/guestman/desc/desc.go +++ b/pkg/hostman/guestman/desc/desc.go @@ -37,12 +37,31 @@ type SGuestCpu struct { // CpuCacheMode string } +type SMemObject struct { + *Object + SizeMB int64 +} + +type SMemDevice struct { + Type string + Id string +} + +type SMemSlot struct { + SizeMB int64 + + MemObj *Object + MemDev *SMemDevice +} + type SGuestMem struct { Slots uint MaxMem uint - // Backend string - // MemFilePath *string `json:",omitempty"` + SizeMB int64 + Mem *Object `json:",omitempty"` + + MemSlots []*SMemSlot `json:",omitempty"` } type SGuestHardwareDesc struct { diff --git a/pkg/hostman/guestman/guesthandlers/guesthandler.go b/pkg/hostman/guestman/guesthandlers/guesthandler.go index a2d40cdef6..75941146a7 100644 --- a/pkg/hostman/guestman/guesthandlers/guesthandler.go +++ b/pkg/hostman/guestman/guesthandlers/guesthandler.go @@ -439,13 +439,21 @@ func guestLiveMigrate(ctx context.Context, userCred mcclient.TokenCredential, si return nil, httperrors.NewMissingParameterError("is_local_storage") } enableTLS := jsonutils.QueryBoolean(body, "enable_tls", false) - hostutils.DelayTaskWithoutReqctx(ctx, guestman.GetGuestManager().LiveMigrate, &guestman.SLiveMigrate{ - Sid: sid, - DestPort: int(destPort), - DestIp: destIp, - IsLocal: isLocal, - EnableTLS: enableTLS, - }) + quicklyFinish := jsonutils.QueryBoolean(body, "quickly_finish", false) + params := &guestman.SLiveMigrate{ + Sid: sid, + DestPort: int(destPort), + DestIp: destIp, + IsLocal: isLocal, + EnableTLS: enableTLS, + QuicklyFinish: quicklyFinish, + } + if body.Contains("max_bandwidth_mb") { + maxBandwidthMb, _ := body.Int("max_bandwidth_mb") + params.MaxBandwidthMB = &maxBandwidthMb + } + + hostutils.DelayTaskWithoutReqctx(ctx, guestman.GetGuestManager().LiveMigrate, params) return nil, nil } diff --git a/pkg/hostman/guestman/guesthelper.go b/pkg/hostman/guestman/guesthelper.go index 3113cbf0ce..ca68a93b06 100644 --- a/pkg/hostman/guestman/guesthelper.go +++ b/pkg/hostman/guestman/guesthelper.go @@ -68,11 +68,13 @@ type SDestPrepareMigrate struct { } type SLiveMigrate struct { - Sid string - DestPort int - DestIp string - IsLocal bool - EnableTLS bool + Sid string + DestPort int + DestIp string + IsLocal bool + EnableTLS bool + MaxBandwidthMB *int64 + QuicklyFinish bool } type SDriverMirror struct { diff --git a/pkg/hostman/guestman/guestman.go b/pkg/hostman/guestman/guestman.go index e04b7b4f2a..8987942226 100644 --- a/pkg/hostman/guestman/guestman.go +++ b/pkg/hostman/guestman/guestman.go @@ -105,9 +105,14 @@ func NewGuestManager(host hostutils.IHost, serversPath string) *SGuestManager { manager.host.StartDHCPServer() manager.dirtyServersChan = make(chan struct{}) manager.dirtyServers = make([]*SKVMGuestInstance, 0) + procutils.NewCommand("mkdir", "-p", manager.QemuLogDir()).Run() return manager } +func (m *SGuestManager) QemuLogDir() string { + return path.Join(m.ServersPath, "logs") +} + func (m *SGuestManager) GetServer(sid string) (*SKVMGuestInstance, bool) { s, ok := m.Servers.Load(sid) if ok { diff --git a/pkg/hostman/guestman/guesttasks.go b/pkg/hostman/guestman/guesttasks.go index 7c13a302ae..b4d60118d7 100644 --- a/pkg/hostman/guestman/guesttasks.go +++ b/pkg/hostman/guestman/guesttasks.go @@ -43,6 +43,7 @@ import ( "yunion.io/x/onecloud/pkg/util/procutils" "yunion.io/x/onecloud/pkg/util/qemuimg" "yunion.io/x/onecloud/pkg/util/timeutils2" + "yunion.io/x/onecloud/pkg/util/version" ) type IGuestTasks interface { @@ -592,6 +593,8 @@ func (n *SGuestNetworkSyncTask) addNic(nic *desc.SGuestNetwork) { n.errors = append(n.errors, fmt.Errorf("netdev add failed %s", res)) n.syncNetworkConf() } else { + nic.UpscriptPath = upscript + nic.DownscriptPath = downscript n.onNetdevAdd(nic, pciRoot) } } @@ -605,7 +608,7 @@ func (n *SGuestNetworkSyncTask) onNetdevAdd(nic *desc.SGuestNetwork, pciRoot *de onFail := func(e error) { log.Errorln(e) n.errors = append(n.errors, e) - n.guest.Monitor.NetdevDel(id, func(res string) { + n.guest.Monitor.NetdevDel(nic.Ifname, func(res string) { log.Infof("netdev %s del %s", id, res) }) n.syncNetworkConf() @@ -816,12 +819,15 @@ type SGuestLiveMigrateTask struct { timeoutAt time.Time doTimeoutMigrate bool + + expectDowntime int64 } func NewGuestLiveMigrateTask( ctx context.Context, guest *SKVMGuestInstance, params *SLiveMigrate, ) *SGuestLiveMigrateTask { task := &SGuestLiveMigrateTask{SKVMGuestInstance: guest, ctx: ctx, params: params} + task.expectDowntime = 300 // qemu default downtime 300ms task.MigrateTask = task return task } @@ -835,8 +841,30 @@ func (s *SGuestLiveMigrateTask) onSetZeroBlocks(res string) { s.migrateFailed(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) + s.Monitor.MigrateSetCapability("auto-converge", "on", s.onSetAutoConverge) +} + +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)) + return + } + if version.LT(s.QemuVersion, "4.0.0") { + s.startMigrate() + 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() + } + log.Infof("migrate src guest enable multifd") + s.Monitor.MigrateSetCapability("multifd", "on", cb) } func (s *SGuestLiveMigrateTask) startRamMigrateTimeout() { @@ -853,11 +881,7 @@ func (s *SGuestLiveMigrateTask) startRamMigrateTimeout() { log.Infof("migrate timeout seconds: %d now: %v expectfinial: %v", migSeconds, time.Now(), s.timeoutAt) } -func (s *SGuestLiveMigrateTask) startMigrate(res string) { - if strings.Contains(strings.ToLower(res), "error") { - s.migrateFailed(fmt.Sprintf("Migrate set capability auto-converge error: %s", res)) - return - } +func (s *SGuestLiveMigrateTask) startMigrate() { if s.params.EnableTLS { // https://wiki.qemu.org/Features/MigrationTLS // first remove possible existing tls0 @@ -900,11 +924,20 @@ func (s *SGuestLiveMigrateTask) doMigrate() { copyIncremental = true } s.Monitor.Migrate(fmt.Sprintf("tcp:%s:%d", s.params.DestIp, s.params.DestPort), - copyIncremental, false, s.onSetMigrateDowntime) + copyIncremental, false, s.setMaxBandwidth) } -func (s *SGuestLiveMigrateTask) onSetMigrateDowntime(res string) { - s.Monitor.MigrateSetParameter("downtime-limit", int(options.HostOptions.DefaultLiveMigrateDowntime*1000), s.startMigrateStatusCheck) +func (s *SGuestLiveMigrateTask) setMaxBandwidth(res string) { + if strings.Contains(strings.ToLower(res), "error") { + s.migrateFailed(fmt.Sprintf("Migrate set capability auto-converge error: %s", res)) + return + } + + if s.params.MaxBandwidthMB != nil { + s.Monitor.MigrateSetParameter("max-bandwidth", *s.params.MaxBandwidthMB*1024*1024, s.startMigrateStatusCheck) + } else { + s.startMigrateStatusCheck("") + } } func (s *SGuestLiveMigrateTask) startMigrateStatusCheck(res string) { @@ -921,7 +954,7 @@ func (s *SGuestLiveMigrateTask) startMigrateStatusCheck(res string) { break case <-time.After(time.Second * 5): if s.Monitor != nil { - s.Monitor.GetMigrateStatus(s.onGetMigrateStatus) + s.Monitor.GetMigrateStats(s.onGetMigrateStats) } else { log.Errorf("server %s(%s) migrate stopped unexpectedly", s.GetId(), s.GetName()) s.migrateFailed(fmt.Sprintf("Migrate error: %s", res)) @@ -931,22 +964,76 @@ func (s *SGuestLiveMigrateTask) startMigrateStatusCheck(res string) { } } -func (s *SGuestLiveMigrateTask) onGetMigrateStatus(status string) { +func (s *SGuestLiveMigrateTask) onGetMigrateStats(stats *monitor.MigrationInfo, err error) { + if err != nil { + log.Errorf("%s get migrate stats failed %s", s.GetName(), err) + return + } + s.onGetMigrateStatus(stats) +} + +/* +{ 'enum': 'MigrationStatus', + 'data': [ 'none', 'setup', 'cancelling', 'cancelled', + 'active', 'postcopy-active', 'postcopy-paused', + 'postcopy-recover', 'completed', 'failed', 'colo', + 'pre-switchover', 'device', 'wait-unplug' ] } +*/ + +func (s *SGuestLiveMigrateTask) onGetMigrateStatus(stats *monitor.MigrationInfo) { + status := *stats.Status if status == "completed" { - s.migrateComplete() + 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 == "migrate_disk_copy" { - // do nothing, simply wait - } else if status == "migrate_ram_copy" { - if s.timeoutAt.IsZero() { - s.startRamMigrateTimeout() - } else if !s.doTimeoutMigrate && s.timeoutAt.Before(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 + } else if status == "active" { + var ( + ramTotal int64 + ramRemain int64 + mbps float64 + diskTotal int64 + diskRemain int64 + ) + 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 + } + progress := (1 - float64(diskRemain+ramRemain)/float64(diskTotal+ramTotal)) * 100.0 + hostutils.UpdateServerProgress(context.Background(), s.Id, progress, mbps) + if stats.Disk != nil && stats.Disk.Remaining > 0 { + // process disk copy + // do nothing, simply wait + } else if stats.RAM != nil && stats.RAM.Remaining > 0 { + if s.params.QuicklyFinish && + stats.RAM.DirtySyncCount > 50 && + stats.RAM.DirtyPagesRate > 0 && + s.expectDowntime < *stats.ExpectedDowntime { + cb := func(res string) { + if len(res) == 0 { + s.expectDowntime = *stats.ExpectedDowntime + log.Infof("migrate update downtime to %d", *stats.ExpectedDowntime) + } else { + log.Errorf("failed set migrate downtime %s", res) + } + } + s.Monitor.MigrateSetDowntime(float32(*stats.ExpectedDowntime), cb) + } else if !s.doTimeoutMigrate && + s.params.QuicklyFinish && + stats.RAM.DirtySyncCount > 100 && + stats.RAM.DirtyPagesRate > 0 { + 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 + } } } } @@ -960,7 +1047,7 @@ func (s *SGuestLiveMigrateTask) onMigrateStartPostcopy(res string) { } } -func (s *SGuestLiveMigrateTask) migrateComplete() { +func (s *SGuestLiveMigrateTask) migrateComplete(stats jsonutils.JSONObject) { s.MigrateTask = nil if s.c != nil { close(s.c) @@ -968,7 +1055,11 @@ func (s *SGuestLiveMigrateTask) migrateComplete() { } s.Monitor.Disconnect() s.Monitor = nil - hostutils.TaskComplete(s.ctx, nil) + res := jsonutils.NewDict() + if stats != nil { + res.Set("migration_info", stats) + } + hostutils.TaskComplete(s.ctx, res) } func (s *SGuestLiveMigrateTask) migrateFailed(msg string) { @@ -1002,6 +1093,7 @@ type SGuestResumeTask struct { isTimeout bool cleanTLS bool + resumed bool getTaskData func() (jsonutils.JSONObject, error) } @@ -1069,6 +1161,10 @@ func (s *SGuestResumeTask) onConfirmRunning(status string) { // handle error first, results may be 'paused (internal-error)' s.taskFailed(status) } else if strings.Contains(status, "paused") { + if s.resumed { + s.taskFailed("resume guest twice") + return + } if err := s.onGuestPrelaunch(); err != nil { s.ForceStop() s.taskFailed(err.Error()) @@ -1109,7 +1205,7 @@ func (s *SGuestResumeTask) onGetBlockInfo(blocks []monitor.QemuBlock) { // for _, drv := range results.GetArray() { // // encryption not work // } - time.Sleep(time.Second * 1) + //time.Sleep(time.Second * 1) s.resumeGuest() } @@ -1119,6 +1215,7 @@ func (s *SGuestResumeTask) resumeGuest() { } func (s *SGuestResumeTask) onResumeSucc(res string) { + s.resumed = true s.confirmRunning() } @@ -1729,7 +1826,9 @@ type SGuestHotplugCpuMemTask struct { originalCpuCount int addedCpuCount int + addedMemSize int memSlotNewIndex *int + memSlot *desc.SMemSlot } func NewGuestHotplugCpuMemTask( @@ -1792,6 +1891,11 @@ func (task *SGuestHotplugCpuMemTask) startAddMem() { func (task *SGuestHotplugCpuMemTask) onGetSlotIndex(index int) { var newIndex = index task.memSlotNewIndex = &newIndex + + var objType string + var id = fmt.Sprintf("mem%d", *task.memSlotNewIndex) + var options map[string]string + if task.manager.host.IsHugepagesEnabled() { memPath := fmt.Sprintf("/dev/hugepages/%s-%d", task.GetId(), index) @@ -1814,21 +1918,31 @@ func (task *SGuestHotplugCpuMemTask) onGetSlotIndex(index int) { return } - params := map[string]string{ - "id": fmt.Sprintf("mem%d", *task.memSlotNewIndex), + objType = "memory-backend-file" + options = map[string]string{ "size": fmt.Sprintf("%dM", task.addMemSize), "mem-path": memPath, "share": "on", "prealloc": "on", } - task.Monitor.ObjectAdd("memory-backend-file", params, task.onAddMemObject) } else { - params := map[string]string{ - "id": fmt.Sprintf("mem%d", *task.memSlotNewIndex), + objType = "memory-backend-ram" + options = map[string]string{ "size": fmt.Sprintf("%dM", task.addMemSize), } - task.Monitor.ObjectAdd("memory-backend-ram", params, task.onAddMemObject) } + options["id"] = id + cb := func(reason string) { + if reason == "" { + memObj := desc.NewObject(objType, id) + memObj.Options = options + task.memSlot = new(desc.SMemSlot) + task.memSlot.MemObj = memObj + task.memSlot.SizeMB = int64(task.addMemSize) + } + task.onAddMemObject(reason) + } + task.Monitor.ObjectAdd(objType, options, cb) } func (task *SGuestHotplugCpuMemTask) onAddMemFailed(reason string) { @@ -1847,7 +1961,17 @@ func (task *SGuestHotplugCpuMemTask) onAddMemObject(reason string) { "id": fmt.Sprintf("dimm%d", *task.memSlotNewIndex), "memdev": fmt.Sprintf("mem%d", *task.memSlotNewIndex), } - task.Monitor.DeviceAdd("pc-dimm", params, task.onAddMemDevice) + cb := func(reason string) { + if reason == "" { + task.memSlot.MemDev = &desc.SMemDevice{ + Type: "pc-dimm", + Id: fmt.Sprintf("dimm%d", *task.memSlotNewIndex), + } + } + task.onAddMemDevice(reason) + } + + task.Monitor.DeviceAdd("pc-dimm", params, cb) } func (task *SGuestHotplugCpuMemTask) onAddMemDevice(reason string) { @@ -1855,9 +1979,33 @@ func (task *SGuestHotplugCpuMemTask) onAddMemDevice(reason string) { task.onAddMemFailed(reason) return } + task.addedMemSize = task.addMemSize task.onSucc() } +func (task *SGuestHotplugCpuMemTask) updateGuestDesc() { + task.Desc.Cpu += int64(task.addedCpuCount) + task.Desc.Mem += int64(task.addedMemSize) + if task.addedMemSize > 0 { + if task.Desc.MemDesc.MemSlots == nil { + task.Desc.MemDesc.MemSlots = make([]*desc.SMemSlot, 0) + } + task.Desc.MemDesc.MemSlots = append(task.Desc.MemDesc.MemSlots, task.memSlot) + } + if task.addedCpuCount > 0 || task.addedMemSize > 0 { + task.SaveLiveDesc(task.Desc) + } + if task.addedMemSize > 0 { + vncPort := task.GetVncPort() + data := jsonutils.NewDict() + data.Set("vnc_port", jsonutils.NewInt(int64(vncPort))) + data.Set("sync_qemu_cmdline", jsonutils.JSONTrue) + if err := task.saveScripts(data); err != nil { + log.Errorf("failed save script: %s", err) + } + } +} + func (task *SGuestHotplugCpuMemTask) onFail(reason string) { body := jsonutils.NewDict() if task.addedCpuCount < task.addCpuCount { @@ -1866,10 +2014,12 @@ func (task *SGuestHotplugCpuMemTask) onFail(reason string) { } else if task.memSlotNewIndex != nil { body.Set("add_mem_failed", jsonutils.JSONTrue) } + task.updateGuestDesc() hostutils.TaskFailed2(task.ctx, reason, body) } func (task *SGuestHotplugCpuMemTask) onSucc() { + task.updateGuestDesc() hostutils.TaskComplete(task.ctx, nil) } diff --git a/pkg/hostman/guestman/pci.go b/pkg/hostman/guestman/pci.go index c09e956196..650390d9d0 100644 --- a/pkg/hostman/guestman/pci.go +++ b/pkg/hostman/guestman/pci.go @@ -51,7 +51,7 @@ func (s *SKVMGuestInstance) pciControllerFind(cont desc.PCI_CONTROLLER_TYPE) *de func (s *SKVMGuestInstance) initGuestDesc() error { s.initCpuDesc() - s.initMemDesc() + s.initMemDesc(s.Desc.Mem) s.initMachineDesc() pciRoot, pciBridge := s.initGuestPciControllers() @@ -87,6 +87,21 @@ func (s *SKVMGuestInstance) initGuestDesc() error { return s.ensurePciAddresses() } +func (s *SKVMGuestInstance) loadGuestPciAddresses() error { + err := s.initGuestPciAddresses() + if err != nil { + return errors.Wrap(err, "init guest pci addresses") + } + if err := s.initMachineDefaultDevices(); err != nil { + return errors.Wrap(err, "init machine default devices") + } + err = s.ensurePciAddresses() + if err != nil { + return errors.Wrap(err, "load desc ensure pci address") + } + return nil +} + func (s *SKVMGuestInstance) initMachineDefaultDevices() error { switch s.Desc.Machine { case "pc": @@ -642,8 +657,10 @@ func (s *SKVMGuestInstance) ensurePciAddresses() error { } // guests description no pci description before host-agent assign pci device address info -// in this case wo need query pci address info by `query-pci` command -func (s *SKVMGuestInstance) initGuestDescFromPCIInfoList(pciInfoList []monitor.PCIInfo) error { +// in this case wo need query pci address info by `query-pci` command. Also memory devices. +func (s *SKVMGuestInstance) initGuestDescFromExistingGuest( + pciInfoList []monitor.PCIInfo, memoryDevicesInfoList []monitor.MemoryDeviceInfo, +) error { if len(pciInfoList) > 1 { return errors.Errorf("unsupported pci info list with multi bus") } @@ -651,11 +668,14 @@ func (s *SKVMGuestInstance) initGuestDescFromPCIInfoList(pciInfoList []monitor.P unknownDevices := make([]monitor.PCIDeviceInfo, 0) s.initCpuDesc() - s.initMemDesc() + err := s.initMemDescFromMemoryInfo(memoryDevicesInfoList) + if err != nil { + return errors.Wrap(err, "init guest memory devices") + } s.initMachineDesc() pciRoot, _ := s.initGuestPciControllers() - err := s.initGuestPciAddresses() + err = s.initGuestPciAddresses() if err != nil { return errors.Wrap(err, "init guest pci addresses") } @@ -816,6 +836,10 @@ func (s *SKVMGuestInstance) initGuestDescFromPCIInfoList(pciInfoList []monitor.P s.Desc.AnonymousPCIDevs = make([]*desc.PCIDevice, len(unknownDevices)) } for i := 0; i < len(unknownDevices); i++ { + if unknownDevices[i].Bus == 0 && unknownDevices[i].Slot == 0 { + // host bridge + continue + } pciDev := desc.NewPCIDevice(pciRoot.CType, "", unknownDevices[i].QdevID) pciDev.PCIAddr = &desc.PCIAddr{ Bus: uint(unknownDevices[i].Bus), diff --git a/pkg/hostman/guestman/pci_test.go b/pkg/hostman/guestman/pci_test.go index 2413eea731..d38117881d 100644 --- a/pkg/hostman/guestman/pci_test.go +++ b/pkg/hostman/guestman/pci_test.go @@ -15,6 +15,7 @@ package guestman import ( + "fmt" "testing" "yunion.io/x/jsonutils" @@ -147,7 +148,14 @@ func TestSKVMGuestInstance_initGuestDesc(t *testing.T) { Accel: "kvm", } //s.initCpuDesc() - s.initMemDesc() + // s.initMemDesc() + s.Desc.MemDesc = new(desc.SGuestMem) + s.Desc.MemDesc.SizeMB = s.Desc.Mem + s.Desc.MemDesc.Mem = desc.NewObject("memory-backend-memfd", "mem") + s.Desc.MemDesc.Mem.Options = map[string]string{ + "size": fmt.Sprintf("%dM", s.Desc.Mem), + "share": "on", "prealloc": "on", + } // s.initMachineDesc() pciRoot, _ := s.initGuestPciControllers() diff --git a/pkg/hostman/guestman/qemu-kvm.go b/pkg/hostman/guestman/qemu-kvm.go index b98cb78f2a..860e74a213 100644 --- a/pkg/hostman/guestman/qemu-kvm.go +++ b/pkg/hostman/guestman/qemu-kvm.go @@ -24,6 +24,7 @@ import ( "regexp" "strconv" "strings" + "syscall" "time" "yunion.io/x/jsonutils" @@ -81,7 +82,7 @@ type SKVMInstanceRuntime struct { QemuVersion string VncPassword string - LiveMigrateDestPort *int + LiveMigrateDestPort *int64 LiveMigrateUseTls bool syncMeta *jsonutils.JSONDict @@ -247,6 +248,9 @@ func (s *SKVMGuestInstance) saveEncryptKeyFile(key string) error { } func (s *SKVMGuestInstance) getOriginId() string { + if s.Desc == nil { + return "" + } return s.Desc.Metadata["__origin_id"] } @@ -378,14 +382,7 @@ func (s *SKVMGuestInstance) LoadDesc() error { if s.IsRunning() { if len(s.Desc.PCIControllers) > 0 { - err = s.initGuestPciAddresses() - if err != nil { - return errors.Wrap(err, "init guest pci addresses") - } - err = s.ensurePciAddresses() - if err != nil { - return errors.Wrap(err, "load desc ensure pci address") - } + return s.loadGuestPciAddresses() } else { s.pciUninitialized = true } @@ -491,9 +488,12 @@ func (s *SKVMGuestInstance) asyncScriptStart(ctx context.Context, params interfa if err != nil { goto finally } else { - err = s.scriptStart() + err = s.scriptStart(ctx) if err == nil { isStarted = true + } else { + // call script stop if guest start failed + s.scriptStop() } } @@ -510,7 +510,6 @@ func (s *SKVMGuestInstance) asyncScriptStart(ctx context.Context, params interfa if isStarted { log.Infof("Async start server %s success!", s.GetName()) s.syncMeta = s.CleanImportMetadata() - s.StartMonitor(ctx, nil) return nil, nil } log.Errorf("Async start server %s failed: %s!!!", s.GetName(), err) @@ -551,6 +550,22 @@ func (s *SKVMGuestInstance) saveScripts(data *jsonutils.JSONDict) error { if err = fileutils2.FilePutContents(s.GetStartScriptPath(), startScript, false); err != nil { return err } + + if jsonutils.QueryBoolean(data, "sync_qemu_cmdline", false) { + cmdline, err := s.getQemuCmdlineFromContent(startScript) + if err != nil { + log.Errorf("failed parse cmdline from start script: %s", err) + } else { + log.Errorf("new cmd: %s", cmdline) + s.SyncQemuCmdline(cmdline) + } + } + + launcher := fmt.Sprintf(guestLauncher, s.GetStartScriptPath(), s.LogFilePath()) + if err := fileutils2.FilePutContents(s.pyLauncherPath(), launcher, false); err != nil { + return errors.Wrap(err, "generate guest launcher") + } + stopScript := s.generateStopScript(data) return fileutils2.FilePutContents(s.GetStopScriptPath(), stopScript, false) } @@ -677,73 +692,53 @@ func (s *SKVMGuestInstance) GetMonitorPath() string { return s.Desc.Metadata["__monitor_path"] } -func (s *SKVMGuestInstance) StartMonitorWithImportGuestSocketFile(ctx context.Context, socketFile string, cb func()) { - timeutils2.AddTimeout(100*time.Millisecond, func() { +func (s *SKVMGuestInstance) StartMonitorWithImportGuestSocketFile(ctx context.Context, socketFile string, cb func()) error { + var mon monitor.Monitor + mon = monitor.NewQmpMonitor( + s.GetName(), + s.Id, + s.onImportGuestMonitorDisConnect, // on monitor disconnect + func(err error) { s.onImportGuestMonitorTimeout(ctx, err) }, // on monitor timeout + func() { + s.Monitor = mon + s.onImportGuestMonitorConnected(ctx) + if cb != nil { + cb() + } + }, // on monitor connected + s.onReceiveQMPEvent, // on reveive qmp event + ) + return mon.ConnectWithSocket(socketFile) +} + +func (s *SKVMGuestInstance) StartMonitor(ctx context.Context, cb func()) error { + if s.GetQmpMonitorPort(-1) > 0 { var mon monitor.Monitor mon = monitor.NewQmpMonitor( - s.GetName(), - s.Id, - s.onImportGuestMonitorDisConnect, // on monitor disconnect - func(err error) { s.onImportGuestMonitorTimeout(ctx, err) }, // on monitor timeout + s.GetName(), s.Id, + s.onMonitorDisConnect, // on monitor disconnect + func(err error) { s.onMonitorTimeout(ctx, err) }, // on monitor timeout func() { s.Monitor = mon - s.onImportGuestMonitorConnected(ctx) + s.onMonitorConnected(ctx) if cb != nil { cb() } }, // on monitor connected - s.onReceiveQMPEvent, // on reveive qmp event + s.onReceiveQMPEvent, // on receive qmp event ) - mon.ConnectWithSocket(socketFile) - }) -} - -func (s *SKVMGuestInstance) StartMonitor(ctx context.Context, cb func()) { - if options.HostOptions.EnableQmpMonitor && s.GetQmpMonitorPort(-1) > 0 { - // try qmp first, if qmp connect failed, use hmp - timeutils2.AddTimeout(100*time.Millisecond, func() { - var mon monitor.Monitor - mon = monitor.NewQmpMonitor( - s.GetName(), - s.Id, - s.onMonitorDisConnect, // on monitor disconnect - func(err error) { s.onMonitorTimeout(ctx, err) }, // on monitor timeout - func() { - s.Monitor = mon - s.onMonitorConnected(ctx) - if cb != nil { - cb() - } - }, // on monitor connected - s.onReceiveQMPEvent, // on reveive qmp event - ) - err := mon.Connect("127.0.0.1", s.GetQmpMonitorPort(-1)) - if err != nil { - log.Errorf("Guest %s qmp monitor connect failed %s, try hmp", s.GetName(), err) - mon = monitor.NewHmpMonitor( - s.GetName(), - s.Id, - s.onMonitorDisConnect, // on monitor disconnect - func(err error) { s.onMonitorTimeout(ctx, err) }, // on monitor timeout - func() { - s.Monitor = mon - s.onMonitorConnected(ctx) - if cb != nil { - cb() - } - }, // on monitor connected - ) - err = mon.Connect("127.0.0.1", s.GetHmpMonitorPort(-1)) - if err != nil { - mon = nil - log.Errorf("Guest %s hmp monitor connect failed %s, something wrong", s.GetName(), err) - } - } - }) + err := mon.Connect("127.0.0.1", s.GetQmpMonitorPort(-1)) + if err != nil { + mon = nil + log.Errorf("Guest %s hmp monitor connect failed %s, something wrong", s.GetName(), err) + return errors.Errorf("connect qmp monitor: %s", err) + } + return nil } else if monitorPath := s.GetMonitorPath(); len(monitorPath) > 0 { - s.StartMonitorWithImportGuestSocketFile(ctx, monitorPath, cb) + return s.StartMonitorWithImportGuestSocketFile(ctx, monitorPath, cb) } else { - log.Errorf("Guest start monitor failed, can't get qmp monitor port or monitor path") + log.Errorf("Guest %s start monitor failed, can't get qmp monitor port or monitor path", s.Id) + return errors.Errorf("Guest %s start monitor failed, can't get qmp monitor port or monitor path", s.Id) } } @@ -760,7 +755,7 @@ func (s *SKVMGuestInstance) onReceiveQMPEvent(event *monitor.Event) { case event.Event == `"STOP"`: if s.MigrateTask != nil { // migrating complete - s.MigrateTask.migrateComplete() + s.MigrateTask.migrateComplete(nil) } hostutils.UpdateServerProgress(context.Background(), s.Id, 0.0, 0) } @@ -929,35 +924,77 @@ func (s *SKVMGuestInstance) setDestMigrateTLS(ctx context.Context, data *jsonuti }) } +func (s *SKVMGuestInstance) migrateEnableMultifd() error { + if version.LT(s.QemuVersion, "4.0.0") { + return nil + } + var err = make(chan error) + cb := func(res string) { + if len(res) > 0 { + err <- errors.Errorf("failed enable multifd %s", res) + } else { + err <- nil + } + } + log.Infof("migrate dest guest enable multifd") + s.Monitor.MigrateSetCapability("multifd", "on", cb) + 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) if s.pciUninitialized { - cb := func(pciInfoList []monitor.PCIInfo, err string) { - if err != "" && pciInfoList == nil { - log.Errorf("failed query pci %s", err) - } else { - // initialize pci address - if err := s.initGuestDescFromPCIInfoList(pciInfoList); err != nil { - log.Errorf("failed init pci devices %s", err) - } else { - s.pciUninitialized = false - s.SaveLiveDesc(s.Desc) - } - } - s.guestRun(ctx) - } - s.Monitor.QueryPci(cb) + s.getPciDevices(ctx) } else { s.guestRun(ctx) } } +func (s *SKVMGuestInstance) getPciDevices(ctx context.Context) { + cb := func(pciInfoList []monitor.PCIInfo, err string) { + if err != "" && pciInfoList == nil { + log.Errorf("failed query pci %s", err) + s.guestRun(ctx) + } else { + s.getMemoryDevices(ctx, pciInfoList) + } + } + s.Monitor.QueryPci(cb) +} + +func (s *SKVMGuestInstance) getMemoryDevices(ctx context.Context, pciInfoList []monitor.PCIInfo) { + cb := func(memoryDevicesInfoList []monitor.MemoryDeviceInfo, err string) { + if err != "" && memoryDevicesInfoList == nil { + log.Errorf("failed query memory device info: %s", err) + } else { + if err := s.initGuestDescFromExistingGuest(pciInfoList, memoryDevicesInfoList); err != nil { + log.Errorf("failed init guest devices: %s", err) + } else { + s.pciUninitialized = false + s.SaveLiveDesc(s.Desc) + vncPort := s.GetVncPort() + data := jsonutils.NewDict() + data.Set("vnc_port", jsonutils.NewInt(int64(vncPort))) + data.Set("sync_qemu_cmdline", jsonutils.JSONTrue) + s.saveScripts(data) + } + } + s.guestRun(ctx) + } + s.Monitor.GetMemoryDevicesInfo(cb) +} + func (s *SKVMGuestInstance) guestRun(ctx context.Context) { if s.LiveMigrateDestPort != nil && ctx != nil { // dest migrate guest body := jsonutils.NewDict() - body.Set("live_migrate_dest_port", jsonutils.NewInt(int64(*s.LiveMigrateDestPort))) + body.Set("live_migrate_dest_port", jsonutils.NewInt(*s.LiveMigrateDestPort)) + err := s.migrateEnableMultifd() + if err != nil { + hostutils.TaskFailed(ctx, err.Error()) + return + } if s.LiveMigrateUseTls { s.setDestMigrateTLS(ctx, body) } else { @@ -989,7 +1026,7 @@ func (s *SKVMGuestInstance) onMonitorDisConnect(err error) { log.Errorf("Guest %s on Monitor Disconnect reason: %v", s.Id, err) s.CleanStartupTask() s.scriptStop() - if !s.IsSlave() { + if !s.IsSlave() && s.LiveMigrateDestPort == nil { s.SyncStatus(fmt.Sprintf("monitor disconnect %v", err)) } if s.guestAgent != nil { @@ -1462,9 +1499,6 @@ func (s *SKVMGuestInstance) Delete(ctx context.Context, migrated bool) error { if err := s.delFlatFiles(ctx); err != nil { return errors.Wrap(err, "delFlatFiles") } - if fileutils2.Exists(s.getQemuLogPath()) { - procutils.NewRemoteCommandAsFarAsPossible("mv", s.getQemuLogPath(), fmt.Sprintf("/tmp/%s-qemu.log", s.GetId())).Run() - } output, err := procutils.NewCommand("rm", "-rf", s.HomeDir()).Output() if err != nil { return errors.Wrapf(err, "rm %s failed: %s", s.HomeDir(), output) @@ -1481,13 +1515,59 @@ func (s *SKVMGuestInstance) Stop() bool { } } -func (s *SKVMGuestInstance) scriptStart() error { - output, err := procutils.NewRemoteCommandAsFarAsPossible("bash", s.GetStartScriptPath()).Output() +func (s *SKVMGuestInstance) LogFilePath() string { + return path.Join(s.manager.QemuLogDir(), s.Id) +} + +func (s *SKVMGuestInstance) readQemuLogFileEnd(size int64) string { + fname := s.LogFilePath() + file, err := os.Open(fname) + if err != nil { + return fmt.Sprintf("failed open log file %s: %s", fname, err) + } + defer file.Close() + + buf := make([]byte, size) + stat, err := os.Stat(fname) + if err != nil { + return fmt.Sprintf("failed stat file %s: %s", fname, err) + } + start := stat.Size() - size + _, err = file.ReadAt(buf, start) + if err != nil { + return fmt.Sprintf("failed read logfile %s: %s", fname, err) + } + return string(buf) +} + +func (s *SKVMGuestInstance) pyLauncherPath() string { + return path.Join(s.HomeDir(), "startvm.py") +} + +func (s *SKVMGuestInstance) scriptStart(ctx context.Context) error { + output, err := procutils.NewRemoteCommandAsFarAsPossible("python2", s.pyLauncherPath()).Output() if err != nil { - s.scriptStop() return fmt.Errorf("Start VM Failed %s %s", output, err) } - return nil + pid, err := strconv.Atoi(string(output)) + if err != nil { + return errors.Wrapf(err, "failed parse pid %s", output) + } + proc, err := os.FindProcess(pid) + if err != nil { + return errors.Wrapf(err, "find process pid(%d)", pid) + } + for { + err = proc.Signal(syscall.Signal(0)) + if err != nil { // qemu process exited + log.Errorf("Guest %s check qemu(%d) process failed: %s", s.Id, pid, err) + return errors.Errorf(s.readQemuLogFileEnd(64)) + } + if err = s.StartMonitor(ctx, nil); err == nil { + return nil + } + time.Sleep(time.Millisecond * 10) + } } func (s *SKVMGuestInstance) scriptStop() bool { @@ -1772,6 +1852,8 @@ func (s *SKVMGuestInstance) SyncConfig( } // update guest live desc + s.Desc.Cpu = guestDesc.Cpu + s.Desc.Mem = guestDesc.Mem s.Desc.SGuestControlDesc = guestDesc.SGuestControlDesc s.Desc.SGuestProjectDesc = guestDesc.SGuestProjectDesc s.Desc.SGuestRegionDesc = guestDesc.SGuestRegionDesc @@ -1786,26 +1868,6 @@ func (s *SKVMGuestInstance) SyncConfig( var runTaskNames = []jsonutils.JSONObject{} var tasks = []IGuestTasks{} - var callBack = func(errs []error) { - s.SaveLiveDesc(s.Desc) - if len(tasks) > 0 { // devices updated, regenerate start script - vncPort := s.GetVncPort() - data := jsonutils.NewDict() - data.Set("vnc_port", jsonutils.NewInt(int64(vncPort))) - s.saveScripts(data) - } - - if len(errs) == 0 { - hostutils.TaskComplete(ctx, nil) - } else { - var reason string - for _, err := range errs { - reason += "; " + err.Error() - } - hostutils.TaskFailed(ctx, reason[2:]) - } - } - if len(delDisks)+len(addDisks) > 0 || cdrom != nil { task := NewGuestDiskSyncTask(s, delDisks, addDisks, cdrom) runTaskNames = append(runTaskNames, jsonutils.NewString("disksync")) @@ -1824,6 +1886,30 @@ func (s *SKVMGuestInstance) SyncConfig( tasks = append(tasks, task) } + lenTasks := len(tasks) + var callBack = func(errs []error) { + s.SaveLiveDesc(s.Desc) + if lenTasks > 0 { // devices updated, regenerate start script + vncPort := s.GetVncPort() + data := jsonutils.NewDict() + data.Set("vnc_port", jsonutils.NewInt(int64(vncPort))) + data.Set("sync_qemu_cmdline", jsonutils.JSONTrue) + if err := s.saveScripts(data); err != nil { + log.Errorf("failed save script: %s", err) + } + } + + if len(errs) == 0 { + hostutils.TaskComplete(ctx, nil) + } else { + var reason string + for _, err := range errs { + reason += "; " + err.Error() + } + hostutils.TaskFailed(ctx, reason[2:]) + } + } + NewGuestSyncConfigTaskExecutor(ctx, s, tasks, callBack).Start(1) res := jsonutils.NewDict() res.Set("task", jsonutils.NewArray(runTaskNames...)) @@ -2050,6 +2136,12 @@ func (s *SKVMGuestInstance) OnResumeSyncMetadataInfo() { s.SyncMetadata(meta) } +func (s *SKVMGuestInstance) SyncQemuCmdline(cmdline string) { + meta := jsonutils.NewDict() + meta.Set("__qemu_cmdline", jsonutils.NewString(cmdline)) + s.SyncMetadata(meta) +} + func (s *SKVMGuestInstance) doBlockIoThrottle() { disks := s.Desc.Disks if len(disks) > 0 { @@ -2062,6 +2154,7 @@ func (s *SKVMGuestInstance) doBlockIoThrottle() { } func (s *SKVMGuestInstance) onGuestPrelaunch() error { + s.LiveMigrateDestPort = nil if options.HostOptions.SetVncPassword { s.SetVncPassword() } diff --git a/pkg/hostman/guestman/qemu-kvmhelper.go b/pkg/hostman/guestman/qemu-kvmhelper.go index ac312290a0..0e8b38b7c7 100644 --- a/pkg/hostman/guestman/qemu-kvmhelper.go +++ b/pkg/hostman/guestman/qemu-kvmhelper.go @@ -36,6 +36,7 @@ import ( "yunion.io/x/onecloud/pkg/hostman/guestman/desc" "yunion.io/x/onecloud/pkg/hostman/guestman/qemu" qemucerts "yunion.io/x/onecloud/pkg/hostman/guestman/qemu/certs" + "yunion.io/x/onecloud/pkg/hostman/monitor" "yunion.io/x/onecloud/pkg/hostman/options" "yunion.io/x/onecloud/pkg/util/procutils" "yunion.io/x/onecloud/pkg/util/qemutils" @@ -60,6 +61,56 @@ const ( DISK_DRIVER_SATA = qemu.DISK_DRIVER_SATA ) +const guestLauncher = ` +#!/usr/bin/env python +import sys +import os +import time +import subprocess + +with open(os.devnull, 'w') as FNULL: + try: + cmd = subprocess.check_output(['bash', '%s'], stderr=FNULL).split() + except BaseException as e: + sys.stderr.write('%%s' %% e) + sys.exit(1) + +pid = os.fork() +if pid < 0: + sys.stderr.write('failed fork child process') + sys.exit(1) + +if pid > 0: + status_encoded = os.waitpid(pid, 0)[1] + sys.exit((status_encoded>>8) & 0xff) +else: + os.setsid() + os.chdir('/') + + pid = os.fork() + if pid < 0: + sys.stderr.write('failed fork child process') + sys.exit(1) + + if pid > 0: + sys.stdout.write('%%d' %% pid) + sys.exit(0) + else: + devnull = os.open('/dev/null', os.O_RDWR) + os.dup2(devnull, 0) + os.close(devnull) + + logfd = os.open('%s', os.O_RDWR|os.O_CREAT|os.O_APPEND) + os.dup2(logfd, 1) + os.dup2(logfd, 2) + os.write(logfd, '%%s Run command: %%s\n' %% (time.strftime('%%Y-%%m-%%d %%H:%%M:%%S', time.localtime()), cmd)) + os.close(logfd) + + if os.execv(cmd[0], cmd) < 0: + sys.stderr.write('exec error') + sys.exit(1) +` + func (s *SKVMGuestInstance) IsKvmSupport() bool { return s.manager.GetHost().IsKvmSupport() } @@ -357,9 +408,8 @@ function nic_mtu() { // Generate Start VM script cmd += `CMD="$QEMU_CMD $QEMU_CMD_KVM_ARG` - if options.HostOptions.LogLevel == "debug" { + if options.HostOptions.EnableQemuDebugLog { input.EnableLog = true - input.LogPath = s.getQemuLogPath() } // inject monitor @@ -368,12 +418,10 @@ function nic_mtu() { Port: uint(s.GetHmpMonitorPort(int(input.VNCPort))), Mode: MODE_READLINE, } - if options.HostOptions.EnableQmpMonitor { - input.QMPMonitor = &qemu.Monitor{ - Id: "qmqmon", - Port: uint(s.GetQmpMonitorPort(int(input.VNCPort))), - Mode: MODE_CONTROL, - } + input.QMPMonitor = &qemu.Monitor{ + Id: "qmqmon", + Port: uint(s.GetQmpMonitorPort(int(input.VNCPort))), + Mode: MODE_CONTROL, } input.EnableUUID = options.HostOptions.EnableVmUuid @@ -418,6 +466,7 @@ function nic_mtu() { if jsonutils.QueryBoolean(data, "need_migrate", false) { input.NeedMigrate = true input.LiveMigratePort = uint(liveMigratePort) + s.LiveMigrateDestPort = &liveMigratePort if jsonutils.QueryBoolean(data, "live_migrate_use_tls", false) { s.LiveMigrateUseTls = true input.LiveMigrateUseTLS = true @@ -439,7 +488,7 @@ if [ ! -z "$STATE_FILE" ] && [ -d "$STATE_FILE" ] && [ -f "$STATE_FILE/content" elif [ ! -z "$STATE_FILE" ] && [ -f "$STATE_FILE" ]; then CMD="$CMD --incoming \"exec: cat $STATE_FILE\"" fi -eval $CMD` +echo $CMD` return cmd, nil } @@ -458,10 +507,7 @@ func (s *SKVMGuestInstance) parseCmdline(input string) (*qemutils.Cmdline, []qem filterOpts = append(filterOpts, o) return true } - case "vnc": - filterOpts = append(filterOpts, o) - return true - case "spice": + case "vnc", "spice", "daemonize": filterOpts = append(filterOpts, o) return true case "chardev": @@ -691,11 +737,61 @@ func (s *SKVMGuestInstance) initCpuDesc() { s.Desc.CpuDesc = s.archMan.GenerateCpuDesc(uint(s.Desc.Cpu), osName, enableKVM, hideKVM) } -func (s *SKVMGuestInstance) initMemDesc() { - // hugepagesEnabled := s.manager.host.IsHugepagesEnabled() - // memPath := fmt.Sprintf("/dev/hugepages/%s", input.UUID) - // enableMemfd := s.isMemcleanEnabled() +func (s *SKVMGuestInstance) initMemDesc(memSizeMB int64) { s.Desc.MemDesc = s.archMan.GenerateMemDesc() + s.Desc.MemDesc.SizeMB = memSizeMB + if s.manager.host.IsHugepagesEnabled() { + s.Desc.MemDesc.Mem = desc.NewObject("memory-backend-file", "mem") + s.Desc.MemDesc.Mem.Options = map[string]string{ + "mem-path": fmt.Sprintf("/dev/hugepages/%s", s.Desc.Uuid), + "size": fmt.Sprintf("%dM", memSizeMB), + "share": "on", "prealloc": "on", + } + } else if s.isMemcleanEnabled() { + s.Desc.MemDesc.Mem = desc.NewObject("memory-backend-memfd", "mem") + s.Desc.MemDesc.Mem.Options = map[string]string{ + "size": fmt.Sprintf("%dM", memSizeMB), + "share": "on", "prealloc": "on", + } + } else { + s.Desc.MemDesc.Mem = desc.NewObject("memory-backend-ram", "mem") + s.Desc.MemDesc.Mem.Options = map[string]string{ + "size": fmt.Sprintf("%dM", memSizeMB), + } + } +} + +func (s *SKVMGuestInstance) initMemDescFromMemoryInfo(memoryDevicesInfoList []monitor.MemoryDeviceInfo) error { + var objType string + if s.manager.host.IsHugepagesEnabled() { + objType = "memory-backend-file" + } else if s.isMemcleanEnabled() { + objType = "memory-backend-memfd" + } else { + objType = "memory-backend-ram" + } + + memSize := s.Desc.Mem + memSlots := make([]*desc.SMemSlot, 0) + for i := 0; i < len(memoryDevicesInfoList); i++ { + if memoryDevicesInfoList[i].Type != "dimm" || memoryDevicesInfoList[i].Data.ID == nil { + return errors.Errorf("unsupported memory device type %s", memoryDevicesInfoList[i].Type) + } + memSize -= (memoryDevicesInfoList[i].Data.Size / 1024 / 1024) + memSlots = append(memSlots, &desc.SMemSlot{ + SizeMB: memoryDevicesInfoList[i].Data.Size / 1024 / 1024, + MemObj: desc.NewObject(objType, path.Base(memoryDevicesInfoList[i].Data.Memdev)), + MemDev: &desc.SMemDevice{ + Type: "pc-dimm", Id: *memoryDevicesInfoList[i].Data.ID, + }, + }) + } + if memSize <= 0 { + return errors.Errorf("wrong memsize %d", s.Desc.Mem) + } + s.initMemDesc(memSize) + s.Desc.MemDesc.MemSlots = memSlots + return nil } func (s *SKVMGuestInstance) initMachineDesc() { diff --git a/pkg/hostman/guestman/qemu/generate.go b/pkg/hostman/guestman/qemu/generate.go index db14dd1665..520e940b18 100644 --- a/pkg/hostman/guestman/qemu/generate.go +++ b/pkg/hostman/guestman/qemu/generate.go @@ -155,11 +155,20 @@ func generatePciControllerOptions(controllers []*desc.PCIController) []string { return opts } -func generateMemoryOption(sizeMB int64, memDesc *desc.SGuestMem) string { - return fmt.Sprintf( +func generateMemoryOption(memDesc *desc.SGuestMem) string { + cmds := []string{} + cmds = append(cmds, fmt.Sprintf( "-m %dM,slots=%d,maxmem=%dM", - sizeMB, memDesc.Slots, memDesc.MaxMem, - ) + memDesc.SizeMB, memDesc.Slots, memDesc.MaxMem, + )) + cmds = append(cmds, generateObjectOption(memDesc.Mem)) + for i := 0; i < len(memDesc.MemSlots); i++ { + memDev := memDesc.MemSlots[i].MemDev + memObj := memDesc.MemSlots[i].MemObj + cmds = append(cmds, generateObjectOption(memObj)) + cmds = append(cmds, fmt.Sprintf("-device %s,id=%s,memdev=%s", memDev.Type, memDev.Id, memObj.Id)) + } + return strings.Join(cmds, " ") } func generateMachineOption(machine string, machineDesc *desc.SGuestMachine) string { @@ -517,7 +526,6 @@ type GenerateStartOptionsInput struct { VNCPort uint VNCPassword bool EnableLog bool - LogPath string HMPMonitor *Monitor QMPMonitor *Monitor IsVdiSpice bool @@ -551,7 +559,7 @@ func GenerateStartOptions( opts = append(opts, drvOpt.FreezeCPU(), cpuOpt) if input.EnableLog { - opts = append(opts, drvOpt.Log(input.EnableLog, input.LogPath)) + opts = append(opts, drvOpt.Log(input.EnableLog)) } // TODO hmp - - @@ -562,7 +570,7 @@ func GenerateStartOptions( opts = append(opts, drvOpt.RTC(), - drvOpt.Daemonize(), + // drvOpt.Daemonize(), drvOpt.Nodefaults(), drvOpt.Nodefconfig(), // drvOpt.NoKVMPitReinjection(), @@ -572,22 +580,9 @@ func GenerateStartOptions( generateSMPOption(input.GuestDesc.CpuDesc), drvOpt.Name(input.GuestDesc.Name), drvOpt.UUID(input.EnableUUID, input.GuestDesc.Uuid), - generateMemoryOption(input.GuestDesc.Mem, input.GuestDesc.MemDesc), + generateMemoryOption(input.GuestDesc.MemDesc), ) - var memDev string - if input.HugepagesEnabled { - memDev = drvOpt.MemPath( - uint64(input.GuestDesc.Mem), - fmt.Sprintf("/dev/hugepages/%s", input.GuestDesc.Uuid), - ) - } else if input.EnableMemfd { - memDev = drvOpt.MemFd(uint64(input.GuestDesc.Mem)) - } else { - memDev = drvOpt.MemDev(uint64(input.GuestDesc.Mem)) - } - opts = append(opts, memDev) - // bootOrder enableMenu := false if input.GuestDesc.Cdrom != nil && input.GuestDesc.Cdrom.Path != "" { diff --git a/pkg/hostman/guestman/qemu/qemu.go b/pkg/hostman/guestman/qemu/qemu.go index 5e7fcde021..7bc69eb438 100644 --- a/pkg/hostman/guestman/qemu/qemu.go +++ b/pkg/hostman/guestman/qemu/qemu.go @@ -81,7 +81,7 @@ type CPUOption struct { type QemuOptions interface { IsArm() bool - Log(enable bool, qemuLogPath string) string + Log(enable bool) string RTC() string FreezeCPU() string Daemonize() string @@ -182,11 +182,11 @@ func (o baseOptions) IsArm() bool { return o.arch == Arch_aarch64 } -func (o baseOptions) Log(enable bool, qemuLogPath string) string { +func (o baseOptions) Log(enable bool) string { if !enable { return "" } - return fmt.Sprintf("-D %s -d all", qemuLogPath) + return "-d all" } func (o baseOptions) RTC() string { diff --git a/pkg/hostman/monitor/hmp.go b/pkg/hostman/monitor/hmp.go index 3523a8f8df..4fd8659073 100644 --- a/pkg/hostman/monitor/hmp.go +++ b/pkg/hostman/monitor/hmp.go @@ -24,6 +24,7 @@ import ( "time" "yunion.io/x/log" + "yunion.io/x/pkg/errors" "yunion.io/x/onecloud/pkg/util/regutils2" ) @@ -340,6 +341,10 @@ func (m *HmpMonitor) GetMigrateStatus(callback StringCallback) { m.Query("info migrate", cb) } +func (m *HmpMonitor) GetMigrateStats(callback MigrateStatsCallback) { + go callback(nil, errors.Errorf("unsupport get migrate stats")) +} + func (m *HmpMonitor) MigrateStartPostcopy(callback StringCallback) { cb := func(output string) { log.Infof("MigrateStartPostcopy %s: %s", m.server, output) @@ -475,6 +480,10 @@ func (m *HmpMonitor) GeMemtSlotIndex(callback func(index int)) { m.Query("info memory-devices", cb) } +func (m *HmpMonitor) GetMemoryDevicesInfo(cb QueryMemoryDevicesCallback) { + go cb(nil, "not supported") +} + func (m *HmpMonitor) ObjectAdd(objectType string, params map[string]string, callback StringCallback) { var paramsKvs = []string{} for k, v := range params { diff --git a/pkg/hostman/monitor/monitor.go b/pkg/hostman/monitor/monitor.go index 008d315bd0..fda7354be6 100644 --- a/pkg/hostman/monitor/monitor.go +++ b/pkg/hostman/monitor/monitor.go @@ -105,6 +105,49 @@ type QemuBlock struct { } } +type MigrationInfo struct { + Status *MigrationStatus `json:"status,omitempty"` + RAM *MigrationStats `json:"ram,omitempty"` + Disk *MigrationStats `json:"disk,omitempty"` + XbzrleCache *XbzrleCacheStats `json:"xbzrle-cache,omitempty"` + TotalTime *int64 `json:"total-time,omitempty"` + ExpectedDowntime *int64 `json:"expected-downtime,omitempty"` + Downtime *int64 `json:"downtime,omitempty"` + SetupTime *int64 `json:"setup-time,omitempty"` + CPUThrottlePercentage *int64 `json:"cpu-throttle-percentage,omitempty"` + ErrorDesc *string `json:"error-desc,omitempty"` +} + +type MigrationStats struct { + Transferred int64 `json:"transferred"` + Remaining int64 `json:"remaining"` + Total int64 `json:"total"` + Duplicate int64 `json:"duplicate"` + Skipped int64 `json:"skipped"` + Normal int64 `json:"normal"` + NormalBytes int64 `json:"normal-bytes"` + DirtyPagesRate int64 `json:"dirty-pages-rate"` + Mbps float64 `json:"mbps"` + DirtySyncCount int64 `json:"dirty-sync-count"` + PostcopyRequests int64 `json:"postcopy-requests"` + PageSize int64 `json:"page-size"` +} + +// XbzrleCacheStats implements the "XBZRLECacheStats" QMP API type. +type XbzrleCacheStats struct { + CacheSize int64 `json:"cache-size"` + Bytes int64 `json:"bytes"` + Pages int64 `json:"pages"` + CacheMiss int64 `json:"cache-miss"` + CacheMissRate float64 `json:"cache-miss-rate"` + Overflow int64 `json:"overflow"` +} + +// MigrationStatus implements the "MigrationStatus" QMP API type. +type MigrationStatus string + +type MigrateStatsCallback func(*MigrationInfo, error) + type blockSizeByte int64 func (self blockSizeByte) String() string { @@ -161,6 +204,7 @@ type Monitor interface { GetCpuCount(func(count int)) AddCpu(cpuIndex int, callback StringCallback) GeMemtSlotIndex(func(index int)) + GetMemoryDevicesInfo(QueryMemoryDevicesCallback) GetBlocks(callback func([]QemuBlock)) EjectCdrom(dev string, callback StringCallback) @@ -187,6 +231,7 @@ type Monitor interface { Migrate(destStr string, copyIncremental, copyFull bool, callback StringCallback) GetMigrateStatus(callback StringCallback) MigrateStartPostcopy(callback StringCallback) + GetMigrateStats(callback MigrateStatsCallback) ReloadDiskBlkdev(device, path string, callback StringCallback) SetVncPassword(proto, password string, callback StringCallback) diff --git a/pkg/hostman/monitor/pci.go b/pkg/hostman/monitor/pci.go index 87a22b5480..08e6a3e4f1 100644 --- a/pkg/hostman/monitor/pci.go +++ b/pkg/hostman/monitor/pci.go @@ -94,3 +94,21 @@ type PCIMemoryRegion struct { } type QueryPciCallback func(pciInfoList []PCIInfo, err string) + +type MemoryDeviceInfo struct { + Type string + Data PcdimmDeviceInfo +} + +type PcdimmDeviceInfo struct { + ID *string `json:"id,omitempty"` + Addr int64 `json:"addr"` + Size int64 `json:"size"` + Slot int64 `json:"slot"` + Node int64 `json:"node"` + Memdev string `json:"memdev"` + Hotplugged bool `json:"hotplugged"` + Hotpluggable bool `json:"hotpluggable"` +} + +type QueryMemoryDevicesCallback func(memoryDevicesInfoList []MemoryDeviceInfo, err string) diff --git a/pkg/hostman/monitor/qmp.go b/pkg/hostman/monitor/qmp.go index 2a5124b056..435d34e3e0 100644 --- a/pkg/hostman/monitor/qmp.go +++ b/pkg/hostman/monitor/qmp.go @@ -747,6 +747,26 @@ func (m *QmpMonitor) GetMigrateStatus(callback StringCallback) { m.Query(cmd, cb) } +func (m *QmpMonitor) GetMigrateStats(callback MigrateStatsCallback) { + var ( + cmd = &Command{Execute: "query-migrate"} + cb = func(res *Response) { + if res.ErrorVal != nil { + callback(nil, errors.Errorf(res.ErrorVal.Error())) + } else { + migStats := new(MigrationInfo) + err := json.Unmarshal(res.Return, migStats) + if err != nil { + callback(nil, err) + return + } + callback(migStats, nil) + } + } + ) + m.Query(cmd, cb) +} + func (m *QmpMonitor) MigrateStartPostcopy(callback StringCallback) { var ( cmd = &Command{Execute: "migrate-start-postcopy"} @@ -984,6 +1004,28 @@ func (m *QmpMonitor) GeMemtSlotIndex(callback func(index int)) { m.HumanMonitorCommand("info memory-devices", cb) } +func (m *QmpMonitor) GetMemoryDevicesInfo(callback QueryMemoryDevicesCallback) { + var ( + cb = func(res *Response) { + if res.ErrorVal != nil { + callback(nil, res.ErrorVal.Error()) + } else { + memDevices := make([]MemoryDeviceInfo, 0) + err := json.Unmarshal(res.Return, &memDevices) + if err != nil { + callback(nil, err.Error()) + } else { + callback(memDevices, "") + } + } + } + cmd = &Command{ + Execute: "query-memory-devices", + } + ) + m.Query(cmd, cb) +} + func (m *QmpMonitor) BlockIoThrottle(driveName string, bps, iops int64, callback StringCallback) { cmd := fmt.Sprintf("block_set_io_throttle %s %d 0 0 %d 0 0", driveName, bps, iops) m.HumanMonitorCommand(cmd, callback) diff --git a/pkg/hostman/options/options.go b/pkg/hostman/options/options.go index cca9acde33..2608da6166 100644 --- a/pkg/hostman/options/options.go +++ b/pkg/hostman/options/options.go @@ -83,7 +83,6 @@ type SHostOptions struct { BlockIoScheduler string `help:"Block IO scheduler, deadline or cfq" default:"deadline"` EnableKsm bool `help:"Enable Kernel Same Page Merging"` HugepagesOption string `help:"Hugepages option: disable|native|transparent" default:"transparent"` - EnableQmpMonitor bool `help:"Enable qmp monitor" default:"true"` PrivatePrefixes []string `help:"IPv4 private prefixes"` LocalImagePath []string `help:"Local image storage paths"` @@ -185,8 +184,9 @@ type SHostOptions struct { BinaryMemcleanPath string `help:"execute binary memclean path" default:"/opt/yunion/bin/memclean"` - MaxHotplugVCpuCount int `help:"maximal possible vCPU count that the platform kvm supports"` - PcieRootPortCount int `help:"pcie root port count" default:"2"` + MaxHotplugVCpuCount int `help:"maximal possible vCPU count that the platform kvm supports"` + PcieRootPortCount int `help:"pcie root port count" default:"2"` + EnableQemuDebugLog bool `help:"enable qemu debug logs" default:"false"` } var ( diff --git a/pkg/mcclient/options/compute/servers.go b/pkg/mcclient/options/compute/servers.go index d117844211..f2a27aa2ec 100644 --- a/pkg/mcclient/options/compute/servers.go +++ b/pkg/mcclient/options/compute/servers.go @@ -1009,6 +1009,10 @@ type ServerLiveMigrateOptions struct { SkipCpuCheck *bool `help:"Skip check CPU mode of the target host" json:"skip_cpu_check"` SkipKernelCheck *bool `help:"Skip target kernel version check" json:"skip_kernel_check"` EnableTLS *bool `help:"Enable tls migration" json:"enable_tls"` + QuicklyFinish *bool `help:"quickly finish, fix downtime after a few rounds of memory synchronization"` + MaxBandwidthMb *int64 `help:"live migrate downtime, unit MB"` + + KeepDestGuestOnFailed *bool `help:"do not delete dest guest on migrate failed, for debug"` } func (o *ServerLiveMigrateOptions) GetId() string { @@ -1019,6 +1023,20 @@ func (o *ServerLiveMigrateOptions) Params() (jsonutils.JSONObject, error) { return options.StructToParams(o) } +type ServerSetLiveMigrateParamsOptions struct { + ID string `help:"ID of server" json:"-"` + MaxBandwidthMB *int64 `help:"live migrate downtime, unit MB"` + DowntimeLimitMS *int64 `help:"live migrate downtime limit"` +} + +func (o *ServerSetLiveMigrateParamsOptions) GetId() string { + return o.ID +} + +func (o *ServerSetLiveMigrateParamsOptions) Params() (jsonutils.JSONObject, error) { + return options.StructToParams(o) +} + type ServerBatchMetadataOptions struct { Guests []string `help:"IDs or names of server" json:"-"` TAGS []string `help:"Tags info, eg: hypervisor=aliyun、os_type=Linux、os_version"` diff --git a/pkg/util/logclient/consts.go b/pkg/util/logclient/consts.go index c8220ff432..c8ac0c04a8 100644 --- a/pkg/util/logclient/consts.go +++ b/pkg/util/logclient/consts.go @@ -79,6 +79,7 @@ const ( ACT_SAVE_IMAGE = "save_image" ACT_SET_AUTO_RENEW = "set_auto_renew" ACT_MIGRATE = "migrate" + ACT_MIGRATING = "migrating" ACT_EIP_ASSOCIATE = "eip_associate" ACT_EIP_DISSOCIATE = "eip_dissociate" ACT_EIP_CONVERT = "eip_convert"