fix create bakcup server and some bugs

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 {

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

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() {

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)
 			}
 		}
 	}

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) {

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<type>\w+), device (?P<device>\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)
 }

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)

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) {