From 50a28caab4bed0189de18e8ef3b5fc7bb88e9d9b Mon Sep 17 00:00:00 2001 From: wanyaoqi Date: Tue, 19 Feb 2019 18:26:51 +0800 Subject: [PATCH] fix create bakcup server and some bugs --- pkg/compute/models/guest_actions.go | 2 +- pkg/hostman/guesthandlers/guesthandler.go | 3 - pkg/hostman/guestman/guestman.go | 5 +- pkg/hostman/guestman/guesttasks.go | 4 +- pkg/hostman/guestman/qemu-kvm.go | 74 +++++++++++++++++++---- pkg/hostman/monitor/hmp.go | 29 ++++++++- pkg/hostman/monitor/monitor.go | 3 +- pkg/hostman/monitor/qmp.go | 26 +++++++- 8 files changed, 121 insertions(+), 25 deletions(-) diff --git a/pkg/compute/models/guest_actions.go b/pkg/compute/models/guest_actions.go index e99279e3c4..c6d681bc4a 100644 --- a/pkg/compute/models/guest_actions.go +++ b/pkg/compute/models/guest_actions.go @@ -2250,7 +2250,7 @@ func (self *SGuest) AllowPerformSwitchToBackup(ctx context.Context, userCred mcc } func (self *SGuest) PerformSwitchToBackup(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject, data jsonutils.JSONObject) (jsonutils.JSONObject, error) { - if self.Status == VM_BLOCK_STREAM { + if !utils.IsInStringArray(self.Status, []string{VM_READY, VM_RUNNING}) { return nil, httperrors.NewBadRequestError("Cannot swith to backup when guest in status %s", self.Status) } if len(self.BackupHostId) == 0 { diff --git a/pkg/hostman/guesthandlers/guesthandler.go b/pkg/hostman/guesthandlers/guesthandler.go index ffddfc4b33..dd145ceea1 100644 --- a/pkg/hostman/guesthandlers/guesthandler.go +++ b/pkg/hostman/guesthandlers/guesthandler.go @@ -161,9 +161,6 @@ func guestMonitor(ctx context.Context, sid string, body jsonutils.JSONObject) (i return nil, err } else { var res = <-c - if len(res) > 1 { - res = res[1 : len(res)-1] - } lines := strings.Split(res, "\\r\\n") return strDict{"results": strings.Join(lines, "\n")}, nil diff --git a/pkg/hostman/guestman/guestman.go b/pkg/hostman/guestman/guestman.go index e0eaa1eff0..cf9bdf6c0f 100644 --- a/pkg/hostman/guestman/guestman.go +++ b/pkg/hostman/guestman/guestman.go @@ -74,10 +74,9 @@ func (m *SGuestManager) VerifyExistingGuests(pendingDelete bool) { params.Set("limit", jsonutils.NewInt(0)) params.Set("admin", jsonutils.JSONTrue) params.Set("system", jsonutils.JSONTrue) + params.Set("host", jsonutils.NewString(m.host.GetHostId())) params.Set("pending_delete", jsonutils.NewBool(pendingDelete)) params.Set("get_backup_guests_on_host", jsonutils.JSONTrue) - params.Set("filter.0", jsonutils.NewString( - fmt.Sprintf("host_id.equals(%s)", m.host.GetHostId()))) if len(m.CandidateServers) > 0 { keys := make([]string, len(m.CandidateServers)) var index = 0 @@ -314,7 +313,7 @@ func (m *SGuestManager) CpusetBalance(ctx context.Context, params interface{}) ( func (m *SGuestManager) Status(sid string) string { if guest, ok := m.Servers[sid]; ok { - if guest.IsMaster() && !guest.IsMirrorJobSucc() { + if guest.Monitor != nil && guest.IsMaster() && !guest.IsMirrorJobSucc() { return "block_stream" } if guest.IsRunning() { diff --git a/pkg/hostman/guestman/guesttasks.go b/pkg/hostman/guestman/guesttasks.go index 2ca49c9734..ac019f3c9b 100644 --- a/pkg/hostman/guestman/guesttasks.go +++ b/pkg/hostman/guestman/guesttasks.go @@ -522,7 +522,7 @@ func NewGuestStreamDisksTask(ctx context.Context, guest *SKVMGuestInstance, call } func (s *SGuestStreamDisksTask) Start() { - s.Monitor.GetBlockJobs(s.onInitCheckStreamJobs) + s.Monitor.GetBlockJobCounts(s.onInitCheckStreamJobs) } func (s *SGuestStreamDisksTask) onInitCheckStreamJobs(jobs int) { @@ -592,7 +592,7 @@ func (s *SGuestStreamDisksTask) startWaitBlockStream(res string) { s.c = nil return case <-time.After(time.Second * 1): - s.Monitor.GetBlockJobs(s.checkStreamJobs) + s.Monitor.GetBlockJobCounts(s.checkStreamJobs) } } } diff --git a/pkg/hostman/guestman/qemu-kvm.go b/pkg/hostman/guestman/qemu-kvm.go index dd309afb4f..3ca3323813 100644 --- a/pkg/hostman/guestman/qemu-kvm.go +++ b/pkg/hostman/guestman/qemu-kvm.go @@ -345,7 +345,7 @@ func (s *SKVMGuestInstance) delayStartMonitor(ctx context.Context) { } func (s *SKVMGuestInstance) onReceiveQMPEvent(event *monitor.Event) { - if event.Event == "BLOCK_JOB_READY" && s.IsMaster() { + if event.Event == `"BLOCK_JOB_READY"` && s.IsMaster() { if itype, ok := event.Data["type"]; ok { stype, _ := itype.(string) if stype == "mirror" { @@ -355,12 +355,12 @@ func (s *SKVMGuestInstance) onReceiveQMPEvent(event *monitor.Event) { s.mirrorJobSuccCount = new(int) *s.mirrorJobSuccCount = 1 } - if *s.mirrorJobSuccCount == s.DiskCount() { + if *s.mirrorJobSuccCount >= s.DiskCount() { hostutils.UpdateServerStatus(context.Background(), s.GetId(), "running") } } } - } else if event.Event == "BLOCK_JOB_ERROR" && s.IsMaster() { + } else if event.Event == `"BLOCK_JOB_ERROR"` && s.IsMaster() { modules.Servers.PerformAction(hostutils.GetComputeSession(context.Background()), s.GetId(), "mirror-job-failed", nil) } } @@ -381,11 +381,14 @@ func (s *SKVMGuestInstance) onGetQemuVersion(ctx context.Context, version string body.Set("live_migrate_dest_port", migratePort) hostutils.TaskComplete(ctx, body) } else if jsonutils.QueryBoolean(s.Desc, "is_slave", false) { - if len(appctx.AppContextTaskId(ctx)) > 0 { + if ctx != nil && len(appctx.AppContextTaskId(ctx)) > 0 { s.startQemuBuiltInNbdServer(ctx) } - } else if jsonutils.QueryBoolean(s.Desc, "is_master", false) && ctx == nil { - return + } else if jsonutils.QueryBoolean(s.Desc, "is_master", false) { + s.startDiskBackupMirror(ctx) + if ctx != nil { + s.DoResumeTask(ctx) + } } else { s.DoResumeTask(ctx) } @@ -402,14 +405,42 @@ func (s *SKVMGuestInstance) onMonitorDisConnect(err error) { s.Monitor = nil } +func (s *SKVMGuestInstance) startDiskBackupMirror(ctx context.Context) { + if ctx == nil || len(appctx.AppContextTaskId(ctx)) == 0 { + status := "running" + if !s.IsMirrorJobSucc() { + status = "block_stream" + } + hostutils.UpdateServerStatus(context.Background(), s.GetId(), status) + } else { + metadata, _ := s.Desc.Get("metadata") + if metadata == nil || !metadata.Contains("backup_nbd_server_uri") { + hostutils.TaskFailed(ctx, "Missing dest nbd location") + } + nbdUri, _ := metadata.GetString("backup_nbd_server_uri") + + onSucc := func() { + cb := func(res string) { log.Infof("On backup mirror server(%s) resume start", s.Id) } + s.Monitor.SimpleCommand("cont", cb) + } + NewDriveMirrorTask(ctx, s, nbdUri, "top", onSucc).Start() + } +} + func (s *SKVMGuestInstance) startQemuBuiltInNbdServer(ctx context.Context) { + if ctx == nil || len(appctx.AppContextTaskId(ctx)) == 0 { + return + } + nbdServerPort := s.manager.GetFreePortByBase(BUILT_IN_NBD_SERVER_PORT_BASE) var onNbdServerStarted = func(res string) { if len(res) > 0 { - log.Errorln("Start Qemu Builtin nbd server error %s", res) + log.Errorf("Start Qemu Builtin nbd server error %s", res) hostutils.TaskFailed(ctx, res) } else { - hostutils.TaskComplete(ctx, nil) + res := jsonutils.NewDict() + res.Set("nbd_server_port", jsonutils.NewInt(int64(nbdServerPort))) + hostutils.TaskComplete(ctx, res) } } s.Monitor.StartNbdServer(nbdServerPort, true, true, onNbdServerStarted) @@ -435,7 +466,28 @@ func (s *SKVMGuestInstance) DiskCount() int { } func (s *SKVMGuestInstance) IsMirrorJobSucc() bool { - return s.mirrorJobSuccCount != nil && *s.mirrorJobSuccCount == s.DiskCount() + res := make(chan *jsonutils.JSONArray) + s.Monitor.GetBlockJobs(func(jobs *jsonutils.JSONArray) { + res <- jobs + }) + select { + case <-time.After(time.Second * 3): + return false + case v := <-res: + if v != nil { + mirrorSuccCount := 0 + for _, val := range v.Value() { + jobType, _ := val.GetString("type") + jobStatus, _ := val.GetString("status") + if jobType == "mirror" && jobStatus == "ready" { + mirrorSuccCount += 1 + } + } + return mirrorSuccCount == s.DiskCount() + } else { + return false + } + } } func (s *SKVMGuestInstance) CleanStartupTask() { @@ -503,7 +555,7 @@ func (s *SKVMGuestInstance) DoResumeTask(ctx context.Context) { func (s *SKVMGuestInstance) SyncStatus() { if s.IsRunning() { - s.Monitor.GetBlockJobs(s.CheckBlockOrRunning) + s.Monitor.GetBlockJobCounts(s.CheckBlockOrRunning) return } var status = "ready" @@ -538,7 +590,7 @@ func (s *SKVMGuestInstance) SaveDesc(desc jsonutils.JSONObject) error { } func (s *SKVMGuestInstance) StartGuest(ctx context.Context, params jsonutils.JSONObject) { - hostutils.DelayTask(ctx, s.asyncScriptStart, params) + hostutils.DelayTaskWithoutReqctx(ctx, s.asyncScriptStart, params) } func (s *SKVMGuestInstance) DeployFs(deployInfo *guestfs.SDeployInfo) (jsonutils.JSONObject, error) { diff --git a/pkg/hostman/monitor/hmp.go b/pkg/hostman/monitor/hmp.go index af21ce9d21..a48269ac46 100644 --- a/pkg/hostman/monitor/hmp.go +++ b/pkg/hostman/monitor/hmp.go @@ -11,6 +11,7 @@ import ( "yunion.io/x/jsonutils" "yunion.io/x/log" + "yunion.io/x/onecloud/pkg/util/regutils2" ) type HmpMonitor struct { @@ -299,7 +300,7 @@ func (m *HmpMonitor) GetMigrateStatus(callback StringCallback) { m.Query("info migrate", cb) } -func (m *HmpMonitor) GetBlockJobs(callback func(jobs int)) { +func (m *HmpMonitor) GetBlockJobCounts(callback func(jobs int)) { cb := func(output string) { lines := strings.Split(output, "\n") if lines[0] == "No active jobs" { @@ -312,6 +313,32 @@ func (m *HmpMonitor) GetBlockJobs(callback func(jobs int)) { m.Query("info block-jobs", cb) } +func (m *HmpMonitor) GetBlockJobs(callback func(*jsonutils.JSONArray)) { + cb := func(output string) { + lines := strings.Split(output, "\n") + if lines[0] == "No active jobs" { + callback(nil) + } else { + res := jsonutils.NewArray() + re := regexp.MustCompile(`Type (?P\w+), device (?P\w+)`) + for i := 0; i < len(lines); i++ { + m := regutils2.GetParams(re, lines[i]) + if len(m) > 0 { + jobType, _ := m["type"] + device, _ := m["device"] + jobInfo := jsonutils.NewDict() + jobInfo.Set("type", jsonutils.NewString(jobType)) + jobInfo.Set("device", jsonutils.NewString(device)) + res.Add(jobInfo) + } + } + callback(res) + } + } + + m.Query("info block-jobs", cb) +} + func (m *HmpMonitor) ReloadDiskBlkdev(device, path string, callback StringCallback) { m.Query(fmt.Sprintf("reload_disk_snapshot_blkdev -n %s %s", device, path), callback) } diff --git a/pkg/hostman/monitor/monitor.go b/pkg/hostman/monitor/monitor.go index 4573de4d4e..2e0151c150 100644 --- a/pkg/hostman/monitor/monitor.go +++ b/pkg/hostman/monitor/monitor.go @@ -23,7 +23,8 @@ type Monitor interface { QueryStatus(StringCallback) GetVersion(StringCallback) - GetBlockJobs(func(jobs int)) + GetBlockJobCounts(func(jobs int)) + GetBlockJobs(func(*jsonutils.JSONArray)) GetBlocks(callback func(*jsonutils.JSONArray)) EjectCdrom(dev string, callback StringCallback) diff --git a/pkg/hostman/monitor/qmp.go b/pkg/hostman/monitor/qmp.go index 83705d2663..01cc6c6561 100644 --- a/pkg/hostman/monitor/qmp.go +++ b/pkg/hostman/monitor/qmp.go @@ -149,7 +149,7 @@ func (m *QmpMonitor) read(r io.Reader) { m.callBack(res) } else if val, ok := objmap["return"]; ok { var res = &Response{} - res.Return = *val + res.Return = []byte(*val) if id, ok := objmap["id"]; ok { res.Id = string(*id) } @@ -305,10 +305,11 @@ func (m *QmpMonitor) HumanMonitorCommand(cmd string, callback StringCallback) { } cb = func(res *Response) { + log.Debugf("Monitor ret: %s", res.Return) if res.ErrorVal != nil { callback(res.ErrorVal.Error()) } else { - callback(string(res.Return)) + callback(strings.Trim(string(res.Return), `""`)) } } ) @@ -595,7 +596,7 @@ func (m *QmpMonitor) GetMigrateStatus(callback StringCallback) { m.Query(cmd, cb) } -func (m *QmpMonitor) GetBlockJobs(callback func(jobs int)) { +func (m *QmpMonitor) GetBlockJobCounts(callback func(jobs int)) { var cb = func(res *Response) { if res.ErrorVal != nil { log.Errorln(res.ErrorVal.Error()) @@ -614,6 +615,25 @@ func (m *QmpMonitor) GetBlockJobs(callback func(jobs int)) { m.Query(&Command{Execute: "query-block-jobs"}, cb) } +func (m *QmpMonitor) GetBlockJobs(callback func(*jsonutils.JSONArray)) { + var cb = func(res *Response) { + if res.ErrorVal != nil { + log.Errorln(res.ErrorVal.Error()) + callback(nil) + } else { + ret, err := jsonutils.Parse(res.Return) + if err != nil { + log.Errorf("Parse qmp res error: %s", err) + callback(nil) + } else { + jobs, _ := ret.(*jsonutils.JSONArray) + callback(jobs) + } + } + } + m.Query(&Command{Execute: "query-block-jobs"}, cb) +} + func (m *QmpMonitor) ReloadDiskBlkdev(device, path string, callback StringCallback) { var ( cb = func(res *Response) {