From fa370a7c6d251da6ecb1e19d1619a4f8b542a0c8 Mon Sep 17 00:00:00 2001 From: wanyaoqi Date: Tue, 25 Jun 2019 14:27:16 +0800 Subject: [PATCH 1/2] - live migrate filter cpu model - fix list export keys --- pkg/apis/scheduler/api.go | 9 ++++++--- pkg/cloudcommon/db/db_dispatcher.go | 2 ++ pkg/cloudcommon/db/modelbase.go | 1 + pkg/cloudcommon/types/types.go | 9 +++++---- pkg/compute/models/guests.go | 9 +++++++-- pkg/compute/models/hosts.go | 15 ++++++++------- pkg/compute/tasks/guest_live_migrate_task.go | 7 +++++++ pkg/hostman/hostinfo/hostinfo.go | 3 +++ pkg/hostman/hostinfo/hostinfohelper.go | 2 ++ pkg/scheduler/algorithm/predicates/error.go | 18 ++++++++++-------- .../predicates/guest/migrate_predicate.go | 16 +++++++++++++++- pkg/scheduler/cache/candidate/base.go | 4 ++++ pkg/scheduler/core/types.go | 1 + pkg/util/sysutils/sysutils.go | 15 ++++++++++----- 14 files changed, 81 insertions(+), 30 deletions(-) diff --git a/pkg/apis/scheduler/api.go b/pkg/apis/scheduler/api.go index 8fe9af2d18..d900406f48 100644 --- a/pkg/apis/scheduler/api.go +++ b/pkg/apis/scheduler/api.go @@ -52,9 +52,6 @@ type ServerConfig struct { Name string `json:"name"` GuestStatus string `json:"guest_status"` - // HostId used by migrate - HostId string `json:"host_id"` - // DEPRECATED Metadata map[string]string `json:"__meta__"` ForGuests []*ForGuest `json:"for_guests"` @@ -70,6 +67,12 @@ type ScheduleInput struct { ScheduleBaseConfig ServerConfig + + // HostId used by migrate + HostId string `json:"host_id"` + LiveMigrate bool `json:"live_migrate"` + CpuDesc string `json:"cpu_desc"` + CpuMicrocode string `json:"cpu_microcode"` } func (input ScheduleInput) ToConditionInput() *jsonutils.JSONDict { diff --git a/pkg/cloudcommon/db/db_dispatcher.go b/pkg/cloudcommon/db/db_dispatcher.go index 880e4a7047..7885ce875d 100644 --- a/pkg/cloudcommon/db/db_dispatcher.go +++ b/pkg/cloudcommon/db/db_dispatcher.go @@ -334,6 +334,7 @@ func Query2List(manager IModelManager, ctx context.Context, userCred mcclient.To if extraKeys != nil { extraData.Update(extraKeys) } + log.Errorf("RAWMAP %#v", RowMap) err = q.RowMap2Struct(RowMap, item) if err != nil { return nil, err @@ -352,6 +353,7 @@ func Query2List(manager IModelManager, ctx context.Context, userCred mcclient.To } jsonDict := jsonutils.Marshal(item).(*jsonutils.JSONDict) + log.Errorln(jsonDict) jsonDict = jsonDict.CopyIncludes([]string(listF)...) jsonDict.Update(extraData) if showDetails && !query.Contains("export_keys") { diff --git a/pkg/cloudcommon/db/modelbase.go b/pkg/cloudcommon/db/modelbase.go index 36b302c14c..6182934897 100644 --- a/pkg/cloudcommon/db/modelbase.go +++ b/pkg/cloudcommon/db/modelbase.go @@ -190,6 +190,7 @@ func (manager *SModelBaseManager) InitializeData() error { } func (manager *SModelBaseManager) ListItemExportKeys(ctx context.Context, q *sqlchemy.SQuery, userCred mcclient.TokenCredential, query jsonutils.JSONObject) (*sqlchemy.SQuery, error) { + q = q.AppendField(q.QueryFields()...) return q, nil } diff --git a/pkg/cloudcommon/types/types.go b/pkg/cloudcommon/types/types.go index 96bfc2179f..c8f79888d4 100644 --- a/pkg/cloudcommon/types/types.go +++ b/pkg/cloudcommon/types/types.go @@ -39,10 +39,11 @@ func (info *SDMISystemInfo) ToIPMISystemInfo() *SIPMISystemInfo { } type SCPUInfo struct { - Count int `json:"count"` - Model string `json:"desc"` - Freq int `json:"freq"` - Cache int `json:"cache"` + Count int `json:"count"` + Model string `json:"desc"` + Freq int `json:"freq"` + Cache int `json:"cache"` + Microcode string `json:"microcode"` } type SDMICPUInfo struct { diff --git a/pkg/compute/models/guests.go b/pkg/compute/models/guests.go index 7494d2cbb5..e071bcbd22 100644 --- a/pkg/compute/models/guests.go +++ b/pkg/compute/models/guests.go @@ -1482,9 +1482,14 @@ func (self *SGuest) GetExtraDetails(ctx context.Context, userCred mcclient.Token } func (manager *SGuestManager) ListItemExportKeys(ctx context.Context, q *sqlchemy.SQuery, userCred mcclient.TokenCredential, query jsonutils.JSONObject) (*sqlchemy.SQuery, error) { + var err error + q, err = manager.SModelBaseManager.ListItemExportKeys(ctx, q, userCred, query) + if err != nil { + return nil, err + } + exportKeys, _ := query.GetString("export_keys") keys := strings.Split(exportKeys, ",") - // guest_id as filter key if utils.IsInStringArray("ips", keys) { guestIpsQuery := GuestnetworkManager.Query("guest_id").GroupBy("guest_id") @@ -4150,7 +4155,7 @@ func (self *SGuest) ToSchedDesc() *schedapi.ScheduleInput { self.FillDiskSchedDesc(config.ServerConfigs) self.FillNetSchedDesc(config.ServerConfigs) if len(self.HostId) > 0 && regutils.MatchUUID(self.HostId) { - config.HostId = self.HostId + desc.HostId = self.HostId } config.Project = self.ProjectId /*tags := self.GetApptags() diff --git a/pkg/compute/models/hosts.go b/pkg/compute/models/hosts.go index a466bcd1f3..2f0b51b89a 100644 --- a/pkg/compute/models/hosts.go +++ b/pkg/compute/models/hosts.go @@ -157,13 +157,14 @@ type SHost struct { SysInfo jsonutils.JSONObject `nullable:"true" search:"admin" list:"admin" update:"admin" create:"admin_optional"` // Column(JSONEncodedDict, nullable=True) SN string `width:"128" charset:"ascii" nullable:"true" list:"admin" update:"admin" create:"admin_optional"` // Column(VARCHAR(128, charset='ascii'), nullable=True) - CpuCount int `nullable:"true" list:"admin" update:"admin" create:"admin_optional"` // Column(TINYINT, nullable=True) # cpu count - NodeCount int8 `nullable:"true" list:"admin" update:"admin" create:"admin_optional"` // Column(TINYINT, nullable=True) - CpuDesc string `width:"64" charset:"ascii" nullable:"true" get:"admin" update:"admin" create:"admin_optional"` // Column(VARCHAR(64, charset='ascii'), nullable=True) - CpuMhz int `nullable:"true" get:"admin" update:"admin" create:"admin_optional"` // Column(Integer, nullable=True) # cpu MHz - CpuCache int `nullable:"true" get:"admin" update:"admin" create:"admin_optional"` // Column(Integer, nullable=True) # cpu Cache in KB - CpuReserved int `nullable:"true" default:"0" list:"admin" update:"admin" create:"admin_optional"` // Column(TINYINT, nullable=True, default=0) - CpuCmtbound float32 `nullable:"true" default:"8" list:"admin" update:"admin" create:"admin_optional"` // = Column(Float, nullable=True) + CpuCount int `nullable:"true" list:"admin" update:"admin" create:"admin_optional"` // Column(TINYINT, nullable=True) # cpu count + NodeCount int8 `nullable:"true" list:"admin" update:"admin" create:"admin_optional"` // Column(TINYINT, nullable=True) + CpuDesc string `width:"64" charset:"ascii" nullable:"true" get:"admin" update:"admin" create:"admin_optional"` // Column(VARCHAR(64, charset='ascii'), nullable=True) + CpuMhz int `nullable:"true" get:"admin" update:"admin" create:"admin_optional"` // Column(Integer, nullable=True) # cpu MHz + CpuCache int `nullable:"true" get:"admin" update:"admin" create:"admin_optional"` // Column(Integer, nullable=True) # cpu Cache in KB + CpuReserved int `nullable:"true" default:"0" list:"admin" update:"admin" create:"admin_optional"` // Column(TINYINT, nullable=True, default=0) + CpuCmtbound float32 `nullable:"true" default:"8" list:"admin" update:"admin" create:"admin_optional"` // = Column(Float, nullable=True) + CpuMicrocode string `width:"64" charset:"ascii" nullable:"true" get:"admin" update:"admin" create:"admin_optional"` MemSize int `nullable:"true" list:"admin" update:"admin" create:"admin_optional"` // Column(Integer, nullable=True) # memory size in MB MemReserved int `nullable:"true" default:"0" list:"admin" update:"admin" create:"admin_optional"` // Column(Integer, nullable=True, default=0) # memory reserved in MB diff --git a/pkg/compute/tasks/guest_live_migrate_task.go b/pkg/compute/tasks/guest_live_migrate_task.go index 08bd2e0dcc..38b0b9f267 100644 --- a/pkg/compute/tasks/guest_live_migrate_task.go +++ b/pkg/compute/tasks/guest_live_migrate_task.go @@ -56,6 +56,13 @@ func (self *GuestMigrateTask) GetSchedParams() (*schedapi.ScheduleInput, error) preferHostId, _ := self.Params.GetString("prefer_host_id") schedDesc.ServerConfig.PreferHost = preferHostId } + guestStatus, _ := self.Params.GetString("guest_status") + if !jsonutils.QueryBoolean(self.Params, "is_rescue_mode", false) && (guestStatus == api.VM_RUNNING || guestStatus == api.VM_SUSPEND) { + schedDesc.LiveMigrate = true + host := guest.GetHost() + schedDesc.CpuDesc = host.CpuDesc + schedDesc.CpuMicrocode = host.CpuMicrocode + } return schedDesc, nil } diff --git a/pkg/hostman/hostinfo/hostinfo.go b/pkg/hostman/hostinfo/hostinfo.go index 94737b2f52..c6dc53257f 100644 --- a/pkg/hostman/hostinfo/hostinfo.go +++ b/pkg/hostman/hostinfo/hostinfo.go @@ -742,6 +742,7 @@ func (h *SHostInfo) updateHostRecord(hostId string) { content.Set("cpu_count", jsonutils.NewInt(int64(h.Cpu.cpuInfoProc.Count))) content.Set("node_count", jsonutils.NewInt(int64(h.Cpu.cpuInfoDmi.Nodes))) content.Set("cpu_desc", jsonutils.NewString(h.Cpu.cpuInfoProc.Model)) + content.Set("cpu_microcode", jsonutils.NewString(h.Cpu.cpuInfoProc.Microcode)) content.Set("cpu_mhz", jsonutils.NewInt(int64(h.Cpu.cpuInfoProc.Freq))) content.Set("cpu_cache", jsonutils.NewInt(int64(h.Cpu.cpuInfoProc.Cache))) content.Set("mem_size", jsonutils.NewInt(int64(h.Mem.MemInfo.Total))) @@ -1353,6 +1354,8 @@ func NewHostInfo() (*SHostInfo, error) { res.Cpu = cpu } + log.Infof("CPU Model %s Microcode %s", cpu.cpuInfoProc.Model, cpu.cpuInfoProc.Microcode) + mem, err := DetectMemoryInfo() if err != nil { return nil, err diff --git a/pkg/hostman/hostinfo/hostinfohelper.go b/pkg/hostman/hostinfo/hostinfohelper.go index 26429d49ce..3655957881 100644 --- a/pkg/hostman/hostinfo/hostinfohelper.go +++ b/pkg/hostman/hostinfo/hostinfohelper.go @@ -311,6 +311,8 @@ type SSysInfo struct { QemuVersion string `json:"qemu_version"` OvsVersion string `json:"ovs_version"` KvmModule string `json:"kvm_module"` + CpuModelName string `json:"cpu_model_name"` + CpuMicrocode string `json:"cpu_microcode"` StorageType string `json:"storage_type"` } diff --git a/pkg/scheduler/algorithm/predicates/error.go b/pkg/scheduler/algorithm/predicates/error.go index be769a4bbb..35e882b008 100644 --- a/pkg/scheduler/algorithm/predicates/error.go +++ b/pkg/scheduler/algorithm/predicates/error.go @@ -31,14 +31,16 @@ const ( ErrNoEnoughAvailableGPUs = `no enough available GPUs` ErrNotSupportNest = `nested function not supported` - ErrRequireMvs = `require mvs` - ErrRequireNoMvs = `require not mvs` - ErrHostIsSpecifiedForMigration = `host_id specified for migration` - ErrMoreThanOneSizeUnspecificSplit = `more than 1 size unspecific split` - ErrNoMoreSpaceForUnspecificSplit = `no more space for an unspecific split` - ErrSubtotalOfSplitExceedsDiskSize = `subtotal of split exceeds disk size` - ErrBaremetalHasAlreadyBeenOccupied = `baremetal has already been occupied` - ErrPrepaidHostOccupied = `prepaid host occupied` + ErrRequireMvs = `require mvs` + ErrRequireNoMvs = `require not mvs` + ErrHostIsSpecifiedForMigration = `host_id specified for migration` + ErrHostCpuModelIsNotMatchForLiveMigrate = `host cpu mode not match for live migrate` + ErrHostCpuMicrocodeNotMatchForLiveMigrate = `host cpu microcode not match for live migrate` + ErrMoreThanOneSizeUnspecificSplit = `more than 1 size unspecific split` + ErrNoMoreSpaceForUnspecificSplit = `no more space for an unspecific split` + ErrSubtotalOfSplitExceedsDiskSize = `subtotal of split exceeds disk size` + ErrBaremetalHasAlreadyBeenOccupied = `baremetal has already been occupied` + ErrPrepaidHostOccupied = `prepaid host occupied` ErrUnknown = `unknown error` ) diff --git a/pkg/scheduler/algorithm/predicates/guest/migrate_predicate.go b/pkg/scheduler/algorithm/predicates/guest/migrate_predicate.go index 57fcf38a6a..522a89532b 100644 --- a/pkg/scheduler/algorithm/predicates/guest/migrate_predicate.go +++ b/pkg/scheduler/algorithm/predicates/guest/migrate_predicate.go @@ -38,9 +38,23 @@ func (p *MigratePredicate) PreExecute(u *core.Unit, cs []core.Candidater) (bool, func (p *MigratePredicate) Execute(u *core.Unit, c core.Candidater) (bool, []core.PredicateFailureReason, error) { h := predicates.NewPredicateHelper(p, u, c) + schedData := u.SchedData() - if u.SchedData().HostId == c.IndexKey() { + if schedData.HostId == c.IndexKey() { h.Exclude(predicates.ErrHostIsSpecifiedForMigration) + return h.GetResult() + } + + if schedData.LiveMigrate { + host := c.Getter().Host() + if schedData.CpuDesc != host.CpuDesc { + h.Exclude(predicates.ErrHostCpuModelIsNotMatchForLiveMigrate) + return h.GetResult() + } + if len(schedData.CpuMicrocode) > 0 && schedData.CpuMicrocode != host.CpuMicrocode { + h.Exclude(predicates.ErrHostCpuMicrocodeNotMatchForLiveMigrate) + return h.GetResult() + } } return h.GetResult() diff --git a/pkg/scheduler/cache/candidate/base.go b/pkg/scheduler/cache/candidate/base.go index c39bae63b3..ff888b764f 100644 --- a/pkg/scheduler/cache/candidate/base.go +++ b/pkg/scheduler/cache/candidate/base.go @@ -63,6 +63,10 @@ func (b baseHostGetter) Zone() *computemodels.SZone { return b.h.Zone } +func (b baseHostGetter) Host() *computemodels.SHost { + return b.h.SHost +} + func (b baseHostGetter) Cloudprovider() *computemodels.SCloudprovider { return b.h.Cloudprovider } diff --git a/pkg/scheduler/core/types.go b/pkg/scheduler/core/types.go index bfcc876c54..5aa8f93df1 100644 --- a/pkg/scheduler/core/types.go +++ b/pkg/scheduler/core/types.go @@ -56,6 +56,7 @@ type CandidatePropertyGetter interface { Id() string Name() string Zone() *computemodels.SZone + Host() *computemodels.SHost Cloudprovider() *computemodels.SCloudprovider IsPublic() bool DomainId() string diff --git a/pkg/util/sysutils/sysutils.go b/pkg/util/sysutils/sysutils.go index 6934ea1ccb..1476df680b 100644 --- a/pkg/util/sysutils/sysutils.go +++ b/pkg/util/sysutils/sysutils.go @@ -78,9 +78,10 @@ func ParseDMISysinfo(lines []string) (*types.SDMISystemInfo, error) { func ParseCPUInfo(lines []string) (*types.SCPUInfo, error) { cnt := 0 var ( - model string - freq string - cache string + model string + freq string + cache string + microcode string ) lv := func(line string) string { return strings.TrimSpace(line[strings.Index(line, ":")+1:]) @@ -95,6 +96,9 @@ func ParseCPUInfo(lines []string) (*types.SCPUInfo, error) { if len(cache) == 0 && strings.HasPrefix(line, "cache size") { cache = strings.TrimSpace(line[strings.Index(line, ":")+1 : strings.Index(line, " KB")]) } + if len(microcode) == 0 && strings.HasPrefix(line, "microcode") { + microcode = lv(line) + } if strings.HasPrefix(line, "processor") { cnt += 1 } @@ -110,8 +114,9 @@ func ParseCPUInfo(lines []string) (*types.SCPUInfo, error) { } model = strings.TrimSpace(model) info := &types.SCPUInfo{ - Count: cnt, - Model: model, + Count: cnt, + Model: model, + Microcode: microcode, } info.Cache, _ = strconv.Atoi(cache) freqF, _ := strconv.ParseFloat(freq, 32) From 73c14b9e226b95ad8ccfd7bbfab4662b4583bad4 Mon Sep 17 00:00:00 2001 From: wanyaoqi Date: Tue, 25 Jun 2019 15:27:06 +0800 Subject: [PATCH 2/2] optimize sync status with backup --- pkg/cloudcommon/db/db_dispatcher.go | 2 - pkg/compute/models/guest_actions.go | 2 +- pkg/compute/tasks/guest_backup_tasks.go | 5 ++ pkg/compute/tasks/guest_syncstatus_task.go | 2 +- pkg/hostman/guestman/guestman.go | 23 +++-- pkg/hostman/guestman/qemu-kvm.go | 98 +++++++++++++++------- pkg/image/tasks/image_convert_task.go | 2 +- 7 files changed, 90 insertions(+), 44 deletions(-) diff --git a/pkg/cloudcommon/db/db_dispatcher.go b/pkg/cloudcommon/db/db_dispatcher.go index 7885ce875d..880e4a7047 100644 --- a/pkg/cloudcommon/db/db_dispatcher.go +++ b/pkg/cloudcommon/db/db_dispatcher.go @@ -334,7 +334,6 @@ func Query2List(manager IModelManager, ctx context.Context, userCred mcclient.To if extraKeys != nil { extraData.Update(extraKeys) } - log.Errorf("RAWMAP %#v", RowMap) err = q.RowMap2Struct(RowMap, item) if err != nil { return nil, err @@ -353,7 +352,6 @@ func Query2List(manager IModelManager, ctx context.Context, userCred mcclient.To } jsonDict := jsonutils.Marshal(item).(*jsonutils.JSONDict) - log.Errorln(jsonDict) jsonDict = jsonDict.CopyIncludes([]string(listF)...) jsonDict.Update(extraData) if showDetails && !query.Contains("export_keys") { diff --git a/pkg/compute/models/guest_actions.go b/pkg/compute/models/guest_actions.go index 0f8d49e6bd..17b1e3e078 100644 --- a/pkg/compute/models/guest_actions.go +++ b/pkg/compute/models/guest_actions.go @@ -2728,7 +2728,7 @@ func (self *SGuest) PerformBlockStreamFailed(ctx context.Context, userCred mccli if len(self.BackupHostId) > 0 { self.SetMetadata(ctx, "__mirror_job_status", "failed", userCred) } - if self.Status == api.VM_BLOCK_STREAM { + if self.Status == api.VM_BLOCK_STREAM || self.Status == api.VM_RUNNING { reason, _ := data.GetString("reason") return nil, self.SetStatus(userCred, api.VM_BLOCK_STREAM_FAIL, reason) } diff --git a/pkg/compute/tasks/guest_backup_tasks.go b/pkg/compute/tasks/guest_backup_tasks.go index 7c0fb63d79..8ab398a11c 100644 --- a/pkg/compute/tasks/guest_backup_tasks.go +++ b/pkg/compute/tasks/guest_backup_tasks.go @@ -192,6 +192,11 @@ func (self *GuestStartAndSyncToBackupTask) OnRequestSyncToBackup(ctx context.Con self.SetStageComplete(ctx, nil) } +func (self *GuestStartAndSyncToBackupTask) OnRequestSyncToBackupFailed(ctx context.Context, guest *models.SGuest, data jsonutils.JSONObject) { + guest.SetStatus(self.UserCred, api.VM_BLOCK_STREAM_FAIL, "OnSyncToBackup") + self.SetStageFailed(ctx, data.String()) +} + type GuestCreateBackupTask struct { SSchedTask } diff --git a/pkg/compute/tasks/guest_syncstatus_task.go b/pkg/compute/tasks/guest_syncstatus_task.go index 6c8d644716..b7e5dd9fae 100644 --- a/pkg/compute/tasks/guest_syncstatus_task.go +++ b/pkg/compute/tasks/guest_syncstatus_task.go @@ -62,7 +62,7 @@ func (self *GuestSyncstatusTask) OnGetStatusSucc(ctx context.Context, guest *mod statusStr = api.VM_SUSPEND case cloudprovider.CloudVMStatusStopped: statusStr = api.VM_READY - case api.VM_BLOCK_STREAM: /// XXX ??? + case api.VM_BLOCK_STREAM, api.VM_BLOCK_STREAM_FAIL: break default: statusStr = api.VM_UNKNOWN diff --git a/pkg/hostman/guestman/guestman.go b/pkg/hostman/guestman/guestman.go index 546b148d40..7be2b23caa 100644 --- a/pkg/hostman/guestman/guestman.go +++ b/pkg/hostman/guestman/guestman.go @@ -29,6 +29,7 @@ import ( "yunion.io/x/log" "yunion.io/x/pkg/util/regutils" + "yunion.io/x/onecloud/pkg/apis/compute" "yunion.io/x/onecloud/pkg/appsrv" "yunion.io/x/onecloud/pkg/cloudcommon/sshkeys" "yunion.io/x/onecloud/pkg/hostman/guestfs" @@ -47,11 +48,12 @@ import ( const ( VNC_PORT_BASE = 5900 - GUEST_RUNNING = "running" - GUEST_BLOCK_STREAM = "block_stream" - GUEST_SUSPEND = "suspend" - GUSET_STOPPED = "stopped" - GUEST_NOT_FOUND = "notfound" + GUEST_RUNNING = compute.VM_RUNNING + GUEST_BLOCK_STREAM = compute.VM_BLOCK_STREAM + GUEST_BLOCK_STREAM_FAIL = compute.VM_BLOCK_STREAM_FAIL + GUEST_SUSPEND = compute.VM_SUSPEND + GUSET_STOPPED = "stopped" + GUEST_NOT_FOUND = "notfound" ) type SGuestManager struct { @@ -360,8 +362,15 @@ func (m *SGuestManager) Status(sid string) string { func (m *SGuestManager) GetStatus(sid string) string { if guest, ok := m.Servers[sid]; ok { - if guest.Monitor != nil && guest.IsMaster() && !guest.IsMirrorJobSucc() { - return GUEST_BLOCK_STREAM + if guest.IsRunning() && guest.Monitor != nil && guest.IsMaster() { + mirrorStatus := guest.MirrorJobStatus() + if mirrorStatus.InProcess() { + return GUEST_BLOCK_STREAM + } else if mirrorStatus.IsFailed() { + timeutils2.AddTimeout(1*time.Second, + func() { guest.SyncMirrorJobFailed("Block job missing") }) + return GUEST_BLOCK_STREAM_FAIL + } } if guest.IsRunning() { return GUEST_RUNNING diff --git a/pkg/hostman/guestman/qemu-kvm.go b/pkg/hostman/guestman/qemu-kvm.go index dcbc2a901c..1b90b52c54 100644 --- a/pkg/hostman/guestman/qemu-kvm.go +++ b/pkg/hostman/guestman/qemu-kvm.go @@ -30,7 +30,7 @@ import ( "yunion.io/x/pkg/util/seclib" "yunion.io/x/pkg/utils" - api "yunion.io/x/onecloud/pkg/apis/compute" + "yunion.io/x/onecloud/pkg/apis/compute" "yunion.io/x/onecloud/pkg/appctx" "yunion.io/x/onecloud/pkg/hostman/guestfs" "yunion.io/x/onecloud/pkg/hostman/hostinfo/hostbridge" @@ -329,7 +329,9 @@ func (s *SKVMGuestInstance) ImportServer(pendingDelete bool) { action = "suspend" } log.Infof("%s is %s, pending_delete=%t", s.GetName(), action, pendingDelete) - s.SyncStatus() + if !s.IsSlave() { + s.SyncStatus() + } } } @@ -445,24 +447,19 @@ func (s *SKVMGuestInstance) onReceiveQMPEvent(event *monitor.Event) { if itype, ok := event.Data["type"]; ok { stype, _ := itype.(string) if stype == "mirror" { - if s.IsMirrorJobSucc() { - _, err := hostutils.UpdateServerStatus(context.Background(), s.GetId(), "running") + mirrorStatus := s.MirrorJobStatus() + if mirrorStatus.IsSucc() { + _, err := hostutils.UpdateServerStatus(context.Background(), s.GetId(), compute.VM_RUNNING) if err != nil { log.Errorf("onReceiveQMPEvent update server status error: %s", err) } + } else if mirrorStatus.IsFailed() { + s.SyncMirrorJobFailed("Block job missing") } } } case event.Event == `"BLOCK_JOB_ERROR"`: - params := jsonutils.NewDict() - params.Set("reason", jsonutils.NewString("BLOCK_JOB_ERROR")) - _, err := modules.Servers.PerformAction( - hostutils.GetComputeSession(context.Background()), - s.GetId(), "block-stream-failed", params, - ) - if err != nil { - log.Errorf("Server %s perform block-stream-failed got error %s", s.GetId(), err) - } + s.SyncMirrorJobFailed("BLOCK_JOB_ERROR") case event.Event == `"GUEST_PANICKED"`: // qemu runc state event source qemu/src/qapi/run-state.json params := jsonutils.NewDict() @@ -486,6 +483,18 @@ func (s *SKVMGuestInstance) onReceiveQMPEvent(event *monitor.Event) { } } +func (s *SKVMGuestInstance) SyncMirrorJobFailed(reason string) { + params := jsonutils.NewDict() + params.Set("reason", jsonutils.NewString(reason)) + _, err := modules.Servers.PerformAction( + hostutils.GetComputeSession(context.Background()), + s.GetId(), "block-stream-failed", params, + ) + if err != nil { + log.Errorf("Server %s perform block-stream-failed got error %s", s.GetId(), err) + } +} + func (s *SKVMGuestInstance) onMonitorConnected(ctx context.Context) { log.Infof("Monitor connected ...") s.Monitor.GetVersion(func(v string) { @@ -507,7 +516,7 @@ func (s *SKVMGuestInstance) onGetQemuVersion(ctx context.Context, version string } } else if jsonutils.QueryBoolean(s.Desc, "is_master", false) { s.startDiskBackupMirror(ctx) - if ctx != nil { + if ctx != nil && len(appctx.AppContextTaskId(ctx)) > 0 { s.DoResumeTask(ctx) } else { if options.HostOptions.SetVncPassword { @@ -533,9 +542,13 @@ func (s *SKVMGuestInstance) onMonitorDisConnect(err error) { func (s *SKVMGuestInstance) startDiskBackupMirror(ctx context.Context) { if ctx == nil || len(appctx.AppContextTaskId(ctx)) == 0 { - status := "running" - if !s.IsMirrorJobSucc() { - status = "block_stream" + status := compute.VM_RUNNING + mirrorStatus := s.MirrorJobStatus() + if mirrorStatus.InProcess() { + status = compute.VM_BLOCK_STREAM + } else if mirrorStatus.IsFailed() { + status = compute.VM_BLOCK_STREAM_FAIL + s.SyncMirrorJobFailed("mirror job missing") } hostutils.UpdateServerStatus(context.Background(), s.GetId(), status) } else { @@ -586,24 +599,39 @@ func (s *SKVMGuestInstance) IsMaster() bool { return jsonutils.QueryBoolean(s.Desc, "is_master", false) } +func (s *SKVMGuestInstance) IsSlave() bool { + return jsonutils.QueryBoolean(s.Desc, "is_slave", false) +} + func (s *SKVMGuestInstance) DiskCount() int { disks, _ := s.Desc.GetArray("disks") return len(disks) } -func (s *SKVMGuestInstance) IsMirrorJobSucc() bool { +type MirrorJob int + +func (ms MirrorJob) IsSucc() bool { + return ms == 1 +} + +func (ms MirrorJob) IsFailed() bool { + return ms == -1 +} + +func (ms MirrorJob) InProcess() bool { + return ms == 0 +} + +func (s *SKVMGuestInstance) MirrorJobStatus() MirrorJob { res := make(chan *jsonutils.JSONArray) s.Monitor.GetBlockJobs(func(jobs *jsonutils.JSONArray) { res <- jobs }) select { case <-time.After(time.Second * 3): - return false + return 0 case v := <-res: - if v != nil { - if len(v.Value()) == 0 { - return true - } + if v != nil && v.Length() >= s.DiskCount() { mirrorSuccCount := 0 for _, val := range v.Value() { jobType, _ := val.GetString("type") @@ -612,9 +640,13 @@ func (s *SKVMGuestInstance) IsMirrorJobSucc() bool { mirrorSuccCount += 1 } } - return mirrorSuccCount == s.DiskCount() + if mirrorSuccCount == s.DiskCount() { + return 1 + } else { + return 0 + } } else { - return false + return -1 } } } @@ -696,12 +728,14 @@ func (s *SKVMGuestInstance) SyncStatus() { } func (s *SKVMGuestInstance) CheckBlockOrRunning(jobs int) { - var status = "running" + var status = compute.VM_RUNNING if jobs > 0 { - if s.IsMaster() && s.IsMirrorJobSucc() { - status = "running" - } else { - status = "block_stream" + mirrorStatus := s.MirrorJobStatus() + if mirrorStatus.InProcess() { + status = compute.VM_BLOCK_STREAM + } else if mirrorStatus.IsFailed() { + status = compute.VM_BLOCK_STREAM_FAIL + s.SyncMirrorJobFailed("Block job missing") } } _, err := hostutils.UpdateServerStatus(context.Background(), s.Id, status) @@ -811,7 +845,7 @@ func (s *SKVMGuestInstance) delTmpDisks(ctx context.Context, migrated bool) erro if disk.Contains("path") { diskPath, _ := disk.GetString("path") d := storageman.GetManager().GetDiskByPath(diskPath) - if d != nil && d.GetType() == api.STORAGE_LOCAL && migrated { + if d != nil && d.GetType() == compute.STORAGE_LOCAL && migrated { if err := d.DeleteAllSnapshot(); err != nil { log.Errorln(err) return err @@ -1365,7 +1399,7 @@ func (s *SKVMGuestInstance) PrepareMigrate(liveMigrage bool) (*jsonutils.JSONDic if disk.Contains("path") { diskPath, _ := disk.GetString("path") d := storageman.GetManager().GetDiskByPath(diskPath) - if d.GetType() == api.STORAGE_LOCAL { + if d.GetType() == compute.STORAGE_LOCAL { back, err := d.PrepareMigrate(liveMigrage) if err != nil { return nil, err diff --git a/pkg/image/tasks/image_convert_task.go b/pkg/image/tasks/image_convert_task.go index 1a1692003e..c3cb57b50f 100644 --- a/pkg/image/tasks/image_convert_task.go +++ b/pkg/image/tasks/image_convert_task.go @@ -58,7 +58,7 @@ func (self *ImageConvertTask) OnInit(ctx context.Context, obj db.IStandaloneMode kwargs := jsonutils.NewDict() kwargs.Set("name", jsonutils.NewString(image.GetName())) osType, err := models.ImagePropertyManager.GetProperty(image.Id, api.IMAGE_OS_TYPE) - if err != nil { + if err == nil { kwargs.Set("os_type", jsonutils.NewString(osType.Value)) } notifyclient.SystemNotify(notify.NotifyPriorityNormal, notifyclient.IMAGE_ACTIVED, kwargs)