From ee69f890bd75f54c8d50843e07dcbad38880691e Mon Sep 17 00:00:00 2001 From: wanyaoqi Date: Mon, 18 Feb 2019 18:38:42 +0800 Subject: [PATCH 1/4] bug fix --- Gopkg.lock | 4 +- ...p_task.go => guest_delete_on_host_task.go} | 6 ++- pkg/hostman/guestman/qemu-kvm.go | 9 ++-- pkg/hostman/hostinfo/hostbridge/hostbridge.go | 4 +- pkg/hostman/hostinfo/hostdhcp/dhcprelay.go | 6 ++- pkg/hostman/hostinfo/hostdhcp/dhcpserver.go | 5 ++- pkg/hostman/hostinfo/hostinfohelper.go | 4 +- pkg/hostman/options/options.go | 4 +- pkg/hostman/system_service/system_service.go | 7 +--- pkg/util/fileutils2/fileutils.go | 42 ++++++++++++++----- pkg/util/netutils2/netutils.go | 5 +++ vendor/yunion.io/x/structarg/structarg.go | 6 ++- 12 files changed, 67 insertions(+), 35 deletions(-) rename pkg/compute/tasks/{guest_delete_backup_task.go => guest_delete_on_host_task.go} (95%) diff --git a/Gopkg.lock b/Gopkg.lock index 7e20fb308d..633bfb02d5 100644 --- a/Gopkg.lock +++ b/Gopkg.lock @@ -1722,11 +1722,11 @@ [[projects]] branch = "master" - digest = "1:cec4892868ae1c9fec0d2cace79d4c43e620345cc3c070d004dec09caf197208" + digest = "1:6a0bb09937441c871c34278e62ea9c80320a6b9894d63261f5c4246c07aeb510" name = "yunion.io/x/structarg" packages = ["."] pruneopts = "UT" - revision = "f39549642b91f03f0b46403f9856d4750e3e59eb" + revision = "f370514c4558e64e798bd83abaff263dabef9f3a" [solve-meta] analyzer-name = "dep" diff --git a/pkg/compute/tasks/guest_delete_backup_task.go b/pkg/compute/tasks/guest_delete_on_host_task.go similarity index 95% rename from pkg/compute/tasks/guest_delete_backup_task.go rename to pkg/compute/tasks/guest_delete_on_host_task.go index dde3ed2d09..d66655c2c0 100644 --- a/pkg/compute/tasks/guest_delete_backup_task.go +++ b/pkg/compute/tasks/guest_delete_on_host_task.go @@ -4,6 +4,8 @@ import ( "context" "yunion.io/x/jsonutils" + "yunion.io/x/log" + "yunion.io/x/onecloud/pkg/cloudcommon/db" "yunion.io/x/onecloud/pkg/cloudcommon/db/taskman" "yunion.io/x/onecloud/pkg/compute/models" @@ -33,8 +35,8 @@ func (self *GuestDeleteOnHostTask) OnInit(ctx context.Context, obj db.IStandalon self.SetStage("OnStopGuest", nil) self.Params.Set("is_force", jsonutils.JSONTrue) if err := guest.GetDriver().RequestStopOnHost(ctx, guest, host, self); err != nil { - self.SetStageFailed(ctx, err.Error()) - return + log.Errorf("RequestStopGuestForDelete fail %s", err) + self.OnStopGuest(ctx, guest, nil) } } diff --git a/pkg/hostman/guestman/qemu-kvm.go b/pkg/hostman/guestman/qemu-kvm.go index ab8467b98a..dd309afb4f 100644 --- a/pkg/hostman/guestman/qemu-kvm.go +++ b/pkg/hostman/guestman/qemu-kvm.go @@ -544,10 +544,11 @@ func (s *SKVMGuestInstance) StartGuest(ctx context.Context, params jsonutils.JSO func (s *SKVMGuestInstance) DeployFs(deployInfo *guestfs.SDeployInfo) (jsonutils.JSONObject, error) { disks, _ := s.Desc.GetArray("disks") if len(disks) > 0 { - storageId, _ := disks[0].GetString("storage_id") - diskId, _ := disks[0].GetString("disk_id") - - disk := storageman.GetManager().GetStorageDisk(storageId, diskId) + diskPath, _ := disks[0].GetString("path") + disk := storageman.GetManager().GetDiskByPath(diskPath) + if disk == nil { + return nil, fmt.Errorf("Cannot find disk index 0") + } return disk.DeployGuestFs(disk.GetPath(), s.Desc, deployInfo) } else { return nil, fmt.Errorf("Guest dosen't have disk ??") diff --git a/pkg/hostman/hostinfo/hostbridge/hostbridge.go b/pkg/hostman/hostinfo/hostbridge/hostbridge.go index a2aded7584..7cbc697566 100644 --- a/pkg/hostman/hostinfo/hostbridge/hostbridge.go +++ b/pkg/hostman/hostinfo/hostbridge/hostbridge.go @@ -45,7 +45,7 @@ func NewBaseBridgeDriver(bridge, inter, ip string) (*SBaseBridgeDriver, error) { bd.bridge = netutils2.NewNetInterface(bridge) if len(inter) > 0 { bd.inter = netutils2.NewNetInterface(inter) - if bd.inter == nil { + if !bd.inter.Exist() { return nil, fmt.Errorf("%s not exists", inter) } bd.ip = ip @@ -132,7 +132,7 @@ func (d *SBaseBridgeDriver) ConfirmToConfig(exists bool, infs []string) (bool, e func (d *SBaseBridgeDriver) SetupAddresses(mask net.IPMask) error { var addr string - if len(d.ip) > 0 { + if len(d.ip) == 0 { addr, mask = netutils2.GetSecretInterfaceAddress() } else { addr = d.ip diff --git a/pkg/hostman/hostinfo/hostdhcp/dhcprelay.go b/pkg/hostman/hostinfo/hostdhcp/dhcprelay.go index d23358930c..82d7c8d81e 100644 --- a/pkg/hostman/hostinfo/hostdhcp/dhcprelay.go +++ b/pkg/hostman/hostinfo/hostdhcp/dhcprelay.go @@ -46,6 +46,8 @@ func NewDHCPRelay(guestDHCPConn *dhcp.Conn, addrs []string) (*SDHCPRelay, error) log.Errorln(err) return nil, err } + + log.Infof("Set Relay To Address: %s, %d", addr, port) relay.destaddr = net.ParseIP(addr) relay.destport = port relay.cache = sync.Map{} @@ -66,7 +68,7 @@ func (r *SDHCPRelay) Start() { func (r *SDHCPRelay) Setup(addr string) error { var err error r.srcaddr = addr - log.Infof("DHCP Relay Bind addr %s port %d", r.srcaddr, DEFAULT_DHCP_RELAY_PORT) + log.Infof("DHCP Relay Server Bind addr %s port %d", r.srcaddr, DEFAULT_DHCP_RELAY_PORT) r.server, r.conn, err = dhcp.NewDHCPServer2(r.srcaddr, DEFAULT_DHCP_RELAY_PORT) if err != nil { log.Errorln(err) @@ -94,7 +96,7 @@ func (r *SDHCPRelay) ServeDHCP(pkt dhcp.Packet, addr *net.UDPAddr, intf *net.Int } func (r *SDHCPRelay) Relay(pkt dhcp.Packet, addr *net.UDPAddr, intf *net.Interface) (dhcp.Packet, error) { - log.Infof("Receive DHCP Relay Rquest FROM %s", pkt.CHAddr()) + log.Infof("Receive DHCP Relay Rquest FROM %s %s", pkt.SIAddr(), pkt.CHAddr()) // clean cache first var now = time.Now().Add(time.Second * -30) diff --git a/pkg/hostman/hostinfo/hostdhcp/dhcpserver.go b/pkg/hostman/hostinfo/hostdhcp/dhcpserver.go index 34fb6d8e3f..f80fe655ae 100644 --- a/pkg/hostman/hostinfo/hostdhcp/dhcpserver.go +++ b/pkg/hostman/hostinfo/hostdhcp/dhcpserver.go @@ -32,7 +32,10 @@ func NewGuestDHCPServer(iface string, relay []string) (*SGuestDHCPServer, error) guestdhcp = new(SGuestDHCPServer) ) - guestdhcp.server, guestdhcp.conn, err = dhcp.NewDHCPServer2(DEFAULT_DHCP_BIND_ADDR, options.HostOptions.DhcpServerPort) + log.Infof("DHCP Server Bind: %s %d", + DEFAULT_DHCP_BIND_ADDR, options.HostOptions.DhcpServerPort) + guestdhcp.server, guestdhcp.conn, err = dhcp.NewDHCPServer2( + DEFAULT_DHCP_BIND_ADDR, options.HostOptions.DhcpServerPort) if err != nil { return nil, err } diff --git a/pkg/hostman/hostinfo/hostinfohelper.go b/pkg/hostman/hostinfo/hostinfohelper.go index 04b2686518..481a584303 100644 --- a/pkg/hostman/hostinfo/hostinfohelper.go +++ b/pkg/hostman/hostinfo/hostinfohelper.go @@ -184,7 +184,7 @@ func (n *SNIC) EnableDHCPRelay() bool { log.Errorln(err) return false } - if len(options.HostOptions.GoDhcpRelay) > 0 && !netutils.IsExitAddress(v4Ip) { + if len(options.HostOptions.DhcpRelay) > 0 && !netutils.IsExitAddress(v4Ip) { return true } else { return false @@ -269,7 +269,7 @@ func NewNIC(desc string) (*SNIC, error) { var dhcpRelay []string if nic.EnableDHCPRelay() { - dhcpRelay = options.HostOptions.GoDhcpRelay + dhcpRelay = options.HostOptions.DhcpRelay } nic.dhcpServer, err = hostdhcp.NewGuestDHCPServer(nic.Bridge, dhcpRelay) if err != nil { diff --git a/pkg/hostman/options/options.go b/pkg/hostman/options/options.go index 39fef19f08..725bb7f44b 100644 --- a/pkg/hostman/options/options.go +++ b/pkg/hostman/options/options.go @@ -51,9 +51,7 @@ type SHostOptions struct { DefaultQemuVersion string `help:"Default qemu version" default:"2.9.1"` - // XXX: Need configure - // dhcp_relay = ('10.168.222.236', 67) => dhcp_relay = ['10.168.222.236', '67'] - GoDhcpRelay []string `help:"DHCP relay upstream"` + DhcpRelay []string `help:"DHCP relay upstream"` DhcpLeaseTime int `default:"100663296" help:"DHCP lease time in seconds"` DhcpRenewalTime int `default:"67108864" help:"DHCP renewal time in seconds"` diff --git a/pkg/hostman/system_service/system_service.go b/pkg/hostman/system_service/system_service.go index 31cfa2ec09..60f1381a3d 100644 --- a/pkg/hostman/system_service/system_service.go +++ b/pkg/hostman/system_service/system_service.go @@ -72,12 +72,7 @@ func (s *SBaseSystemService) IsInstalled() bool { } func (s *SBaseSystemService) GetStatus() map[string]string { - res, err := procutils.NewCommand("systemctl", "status", s.name).Run() - if err != nil { - log.Errorln(err) - return nil - } - + res, _ := procutils.NewCommand("systemctl", "status", s.name).Run() var ret = make(map[string]string, 0) lines := strings.Split(string(res), "\n") for _, line := range lines { diff --git a/pkg/util/fileutils2/fileutils.go b/pkg/util/fileutils2/fileutils.go index 04bdd938af..c35eb11d4e 100644 --- a/pkg/util/fileutils2/fileutils.go +++ b/pkg/util/fileutils2/fileutils.go @@ -488,7 +488,7 @@ func ResizeDiskFs(diskPath string, sizeMb int) error { return fmt.Errorf("%s", output) } if len(part[6]) > 0 { - err := ResizePartitionFs(part[7], part[6]) + err, _ := ResizePartitionFs(part[7], part[6], false) if err != nil { return err } @@ -498,12 +498,26 @@ func ResizeDiskFs(diskPath string, sizeMb int) error { } func FsckExtFs(fpath string) bool { - cmd := []string{"e2fsck", "-f", "-p", fpath} - if _, err := procutils.NewCommand(cmd[0], cmd[1:]...).Run(); err != nil { + log.Debugf("Exec command: %v", []string{"e2fsck", "-f", "-p", fpath}) + cmd := exec.Command("e2fsck", "-f", "-p", fpath) + if err := cmd.Start(); err != nil { log.Errorln(err) return false + } else { + err = cmd.Wait() + if err != nil { + if exiterr, ok := err.(*exec.ExitError); ok { + ws := exiterr.Sys().(syscall.WaitStatus) + if ws.ExitStatus() < 4 { + return true + } + } + log.Errorln(err) + return false + } else { + return true + } } - return true } func FsckXfsFs(fpath string) bool { @@ -515,9 +529,9 @@ func FsckXfsFs(fpath string) bool { return true } -func ResizePartitionFs(fpath, fs string) error { +func ResizePartitionFs(fpath, fs string, raiseError bool) (error, bool) { if len(fs) == 0 { - return nil + return nil, false } var ( cmds = [][]string{} @@ -531,7 +545,11 @@ func ResizePartitionFs(fpath, fs string) error { } } else if strings.HasPrefix(fs, "ext") { if !FsckExtFs(fpath) { - return fmt.Errorf("Failed to fsck ext fs %s", fpath) + if raiseError { + return fmt.Errorf("Failed to fsck ext fs %s", fpath), false + } else { + return nil, false + } } cmds = [][]string{{"resize2fs", fpath}} } else if fs == "xfs" { @@ -540,7 +558,7 @@ func ResizePartitionFs(fpath, fs string) error { _, err = procutils.NewCommand("umount", "-f", tmpPoint).Run() if err != nil { log.Errorln(err) - return err + return err, false } } FsckXfsFs(fpath) @@ -559,11 +577,15 @@ func ResizePartitionFs(fpath, fs string) error { _, err := procutils.NewCommand(cmd[0], cmd[1:]...).Run() if err != nil { log.Errorln(err) - return err + if raiseError { + return err, false + } else { + return nil, false + } } } } - return nil + return nil, true } func GetDevUuid(dev string) map[string]string { diff --git a/pkg/util/netutils2/netutils.go b/pkg/util/netutils2/netutils.go index 94d6da49fa..dfdccf3dfa 100644 --- a/pkg/util/netutils2/netutils.go +++ b/pkg/util/netutils2/netutils.go @@ -179,6 +179,11 @@ func (n *SNetInterface) String() string { return n.name } +func (n *SNetInterface) Exist() bool { + _, err := net.InterfaceByName(n.name) + return err == nil +} + func (n *SNetInterface) FetchInter() *net.Interface { inter, err := net.InterfaceByName(n.name) if err != nil { diff --git a/vendor/yunion.io/x/structarg/structarg.go b/vendor/yunion.io/x/structarg/structarg.go index 08a39fd397..5020249dc2 100644 --- a/vendor/yunion.io/x/structarg/structarg.go +++ b/vendor/yunion.io/x/structarg/structarg.go @@ -888,7 +888,11 @@ func (this *ArgumentParser) parseKeyValue(key, value string) error { arg := this.findOptionalArgument(key) if arg != nil { if arg.IsMulti() { - value = strings.Trim(value, "[]") + if value[0] == '(' { + value = strings.Trim(value, "()") + } else { + value = strings.Trim(value, "[]") + } values := utils.FindWords([]byte(value), 0) for _, v := range values { e := arg.SetValue(v) From 50a28caab4bed0189de18e8ef3b5fc7bb88e9d9b Mon Sep 17 00:00:00 2001 From: wanyaoqi Date: Tue, 19 Feb 2019 18:26:51 +0800 Subject: [PATCH 2/4] 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) { From 0ad75780ee6e5501446cd0d86f94019f2a1315b1 Mon Sep 17 00:00:00 2001 From: wanyaoqi Date: Wed, 20 Feb 2019 11:15:14 +0800 Subject: [PATCH 3/4] add backup delete status --- pkg/compute/models/guest_actions.go | 4 ++++ pkg/compute/models/guests.go | 2 ++ .../tasks/guest_delete_on_host_task.go | 24 ++++++++++++++----- 3 files changed, 24 insertions(+), 6 deletions(-) diff --git a/pkg/compute/models/guest_actions.go b/pkg/compute/models/guest_actions.go index c6d681bc4a..744753d22e 100644 --- a/pkg/compute/models/guest_actions.go +++ b/pkg/compute/models/guest_actions.go @@ -2370,6 +2370,10 @@ func (self *SGuest) PerformDeleteBackup(ctx context.Context, userCred mcclient.T taskData := jsonutils.NewDict() taskData.Set("pruge", jsonutils.NewBool(jsonutils.QueryBoolean(data, "purge", false))) taskData.Set("host_id", jsonutils.NewString(self.BackupHostId)) + taskData.Set("old_status", jsonutils.NewString(self.Status)) + taskData.Set("failed_status", jsonutils.NewString(VM_BACKUP_DELETE_FAILED)) + + self.SetStatus(userCred, VM_DELETING_BACKUP, "delete backup server") if task, err := taskman.TaskManager.NewTask( ctx, "GuestDeleteOnHostTask", self, userCred, taskData, "", "", nil); err != nil { log.Errorf(err.Error()) diff --git a/pkg/compute/models/guests.go b/pkg/compute/models/guests.go index 8cef315626..47c37082ca 100644 --- a/pkg/compute/models/guests.go +++ b/pkg/compute/models/guests.go @@ -64,6 +64,8 @@ const ( VM_BACKUP_CREATE_FAILED = "backup_create_fail" VM_DEPLOYING_BACKUP = "deploying_backup" VM_DEPLOYING_BACKUP_FAILED = "deploging_backup_fail" + VM_DELETING_BACKUP = "deleting_backup" + VM_BACKUP_DELETE_FAILED = "backup_delete_fail" VM_ATTACH_DISK_FAILED = "attach_disk_fail" VM_DETACH_DISK_FAILED = "detach_disk_fail" diff --git a/pkg/compute/tasks/guest_delete_on_host_task.go b/pkg/compute/tasks/guest_delete_on_host_task.go index d66655c2c0..1c6185c7ca 100644 --- a/pkg/compute/tasks/guest_delete_on_host_task.go +++ b/pkg/compute/tasks/guest_delete_on_host_task.go @@ -20,17 +20,17 @@ type GuestDeleteOnHostTask struct { } func (self *GuestDeleteOnHostTask) OnInit(ctx context.Context, obj db.IStandaloneModel, data jsonutils.JSONObject) { + guest := obj.(*models.SGuest) hostId, err := self.Params.GetString("host_id") if err != nil { - self.SetStageFailed(ctx, "Missing param host id") + self.OnFail(ctx, guest, "Missing param host id") return } host := models.HostManager.FetchHostById(hostId) if host == nil { - self.SetStageFailed(ctx, "Host is nil") + self.OnFail(ctx, guest, "Host is nil") return } - guest := obj.(*models.SGuest) self.SetStage("OnStopGuest", nil) self.Params.Set("is_force", jsonutils.JSONTrue) @@ -52,7 +52,7 @@ func (self *GuestDeleteOnHostTask) OnStopGuest(ctx context.Context, guest *model storage := models.StorageManager.FetchStorageById(disk.BackupStorageId) if storage != nil && !isPurge { if err := host.GetHostDriver().RequestDeallocateBackupDiskOnHost(ctx, host, storage, disk, self); err != nil { - self.SetStageFailed(ctx, err.Error()) + self.OnFail(ctx, guest, err.Error()) return } } @@ -61,7 +61,7 @@ func (self *GuestDeleteOnHostTask) OnStopGuest(ctx context.Context, guest *model return nil }) if err != nil { - self.SetStageFailed(ctx, err.Error()) + self.OnFail(ctx, guest, err.Error()) return } } @@ -75,15 +75,27 @@ func (self *GuestDeleteOnHostTask) OnStopGuest(ctx context.Context, guest *model func (self *GuestDeleteOnHostTask) OnUnDeployGuest(ctx context.Context, guest *models.SGuest, data jsonutils.JSONObject) { hostId, _ := self.Params.GetString("host_id") + oldStatus, _ := self.Params.GetString("old_status") if guest.BackupHostId == hostId { _, err := models.GuestManager.TableSpec().Update(guest, func() error { guest.BackupHostId = "" + if len(oldStatus) > 0 { + guest.Status = oldStatus + } return nil }) if err != nil { - self.SetStageFailed(ctx, err.Error()) + self.OnFail(ctx, guest, err.Error()) return } } self.SetStageComplete(ctx, nil) } + +func (self *GuestDeleteOnHostTask) OnFail(ctx context.Context, guest *models.SGuest, reason string) { + failedStatus, _ := self.Params.GetString("failed_status") + if len(failedStatus) > 0 { + guest.SetStatus(self.UserCred, failedStatus, reason) + } + self.SetStageFailed(ctx, reason) +} From 91f109977f65a9b6419975948ff2dd90fb61d649 Mon Sep 17 00:00:00 2001 From: wanyaoqi Date: Wed, 20 Feb 2019 15:09:25 +0800 Subject: [PATCH 4/4] fix backup server --- pkg/compute/guestdrivers/kvm.go | 6 +++-- pkg/compute/models/guest_actions.go | 5 ++--- .../tasks/guest_delete_on_host_task.go | 9 ++++---- pkg/compute/tasks/guest_sync_task.go | 2 +- pkg/hostman/guesthandlers/guesthandler.go | 6 ++++- pkg/hostman/guestman/guesthelper.go | 1 + pkg/hostman/guestman/guestman.go | 3 +++ pkg/hostman/guestman/guesttasks.go | 14 +----------- pkg/hostman/guestman/qemu-kvm.go | 22 +++++++++++++++++-- 9 files changed, 42 insertions(+), 26 deletions(-) diff --git a/pkg/compute/guestdrivers/kvm.go b/pkg/compute/guestdrivers/kvm.go index 22bd714317..024f563727 100644 --- a/pkg/compute/guestdrivers/kvm.go +++ b/pkg/compute/guestdrivers/kvm.go @@ -340,9 +340,11 @@ func (self *SKVMGuestDriver) RequestRebuildRootDisk(ctx context.Context, guest * } func (self *SKVMGuestDriver) RequestSyncToBackup(ctx context.Context, guest *models.SGuest, task taskman.ITask) error { - body := jsonutils.NewDict() - body.Set("backup_nbd_server_uri", jsonutils.NewString(guest.GetMetadata("backup_nbd_server_uri", task.GetUserCred()))) host := guest.GetHost() + desc := guest.GetDriver().GetJsonDescAtHost(ctx, task.GetUserCred(), guest, host) + body := jsonutils.NewDict() + body.Add(desc, "desc") + body.Set("backup_nbd_server_uri", jsonutils.NewString(guest.GetMetadata("backup_nbd_server_uri", task.GetUserCred()))) url := fmt.Sprintf("%s/servers/%s/drive-mirror", host.ManagerUri, guest.Id) header := self.getTaskRequestHeader(task) _, _, err := httputils.JSONRequest(httputils.GetDefaultClient(), ctx, "POST", url, header, body, false) diff --git a/pkg/compute/models/guest_actions.go b/pkg/compute/models/guest_actions.go index 744753d22e..b7935c9460 100644 --- a/pkg/compute/models/guest_actions.go +++ b/pkg/compute/models/guest_actions.go @@ -432,10 +432,10 @@ func (self *SGuest) PerformAttachdisk(ctx context.Context, userCred mcclient.Tok return nil, nil } -func (self *SGuest) StartSyncTask(ctx context.Context, userCred mcclient.TokenCredential, fw_only bool, parentTaskId string) error { +func (self *SGuest) StartSyncTask(ctx context.Context, userCred mcclient.TokenCredential, fwOnly bool, parentTaskId string) error { data := jsonutils.NewDict() - if fw_only { + if fwOnly { data.Add(jsonutils.JSONTrue, "fw_only") } else if err := self.SetStatus(userCred, VM_SYNC_CONFIG, ""); err != nil { log.Errorf(err.Error()) @@ -2370,7 +2370,6 @@ func (self *SGuest) PerformDeleteBackup(ctx context.Context, userCred mcclient.T taskData := jsonutils.NewDict() taskData.Set("pruge", jsonutils.NewBool(jsonutils.QueryBoolean(data, "purge", false))) taskData.Set("host_id", jsonutils.NewString(self.BackupHostId)) - taskData.Set("old_status", jsonutils.NewString(self.Status)) taskData.Set("failed_status", jsonutils.NewString(VM_BACKUP_DELETE_FAILED)) self.SetStatus(userCred, VM_DELETING_BACKUP, "delete backup server") diff --git a/pkg/compute/tasks/guest_delete_on_host_task.go b/pkg/compute/tasks/guest_delete_on_host_task.go index 1c6185c7ca..44e8d80b05 100644 --- a/pkg/compute/tasks/guest_delete_on_host_task.go +++ b/pkg/compute/tasks/guest_delete_on_host_task.go @@ -75,13 +75,9 @@ func (self *GuestDeleteOnHostTask) OnStopGuest(ctx context.Context, guest *model func (self *GuestDeleteOnHostTask) OnUnDeployGuest(ctx context.Context, guest *models.SGuest, data jsonutils.JSONObject) { hostId, _ := self.Params.GetString("host_id") - oldStatus, _ := self.Params.GetString("old_status") if guest.BackupHostId == hostId { _, err := models.GuestManager.TableSpec().Update(guest, func() error { guest.BackupHostId = "" - if len(oldStatus) > 0 { - guest.Status = oldStatus - } return nil }) if err != nil { @@ -89,6 +85,11 @@ func (self *GuestDeleteOnHostTask) OnUnDeployGuest(ctx context.Context, guest *m return } } + self.SetStage("OnSync", nil) + guest.StartSyncTask(ctx, self.UserCred, false, self.GetTaskId()) +} + +func (self *GuestDeleteOnHostTask) OnSync(ctx context.Context, guest *models.SGuest, data jsonutils.JSONObject) { self.SetStageComplete(ctx, nil) } diff --git a/pkg/compute/tasks/guest_sync_task.go b/pkg/compute/tasks/guest_sync_task.go index 4d92cb72de..c8c65ac534 100644 --- a/pkg/compute/tasks/guest_sync_task.go +++ b/pkg/compute/tasks/guest_sync_task.go @@ -36,7 +36,7 @@ func (self *GuestSyncConfTask) OnInit(ctx context.Context, obj db.IStandaloneMod func (self *GuestSyncConfTask) OnSyncComplete(ctx context.Context, obj db.IStandaloneModel, data jsonutils.JSONObject) { guest := obj.(*models.SGuest) - if fw_only, _ := self.GetParams().Bool("fw_only"); fw_only { + if fwOnly, _ := self.GetParams().Bool("fw_only"); fwOnly { db.OpsLog.LogEvent(guest, db.ACT_SYNC_CONF, nil, self.UserCred) self.SetStageComplete(ctx, guest.GetShortDesc(ctx)) } else if data.Contains("task") { diff --git a/pkg/hostman/guesthandlers/guesthandler.go b/pkg/hostman/guesthandlers/guesthandler.go index dd145ceea1..50760596c0 100644 --- a/pkg/hostman/guesthandlers/guesthandler.go +++ b/pkg/hostman/guesthandlers/guesthandler.go @@ -311,8 +311,12 @@ func guestDriveMirror(ctx context.Context, sid string, body jsonutils.JSONObject if err != nil { return nil, httperrors.NewMissingParameterError("backup_nbd_server_uri") } + desc, err := body.Get("desc") + if err != nil { + return nil, httperrors.NewMissingParameterError("desc") + } hostutils.DelayTaskWithoutReqctx(ctx, guestman.GetGuestManager().StartDriveMirror, - &guestman.SDriverMirror{sid, backupNbdServerUri}) + &guestman.SDriverMirror{sid, backupNbdServerUri, desc}) return nil, nil } diff --git a/pkg/hostman/guestman/guesthelper.go b/pkg/hostman/guestman/guesthelper.go index 18802d5de7..6323f424ba 100644 --- a/pkg/hostman/guestman/guesthelper.go +++ b/pkg/hostman/guestman/guesthelper.go @@ -45,6 +45,7 @@ type SLiveMigrate struct { type SDriverMirror struct { Sid string NbdServerUri string + Desc jsonutils.JSONObject } type SReloadDisk struct { diff --git a/pkg/hostman/guestman/guestman.go b/pkg/hostman/guestman/guestman.go index cf9bdf6c0f..cd15102474 100644 --- a/pkg/hostman/guestman/guestman.go +++ b/pkg/hostman/guestman/guestman.go @@ -642,6 +642,9 @@ func (m *SGuestManager) StartDriveMirror(ctx context.Context, params interface{} return nil, hostutils.ParamsError } guest := guestManger.Servers[mirrorParams.Sid] + if err := guest.SaveDesc(mirrorParams.Desc); err != nil { + return nil, err + } task := NewDriveMirrorTask(ctx, guest, mirrorParams.NbdServerUri, "top", nil) task.Start() return nil, nil diff --git a/pkg/hostman/guestman/guesttasks.go b/pkg/hostman/guestman/guesttasks.go index ac019f3c9b..9530929967 100644 --- a/pkg/hostman/guestman/guesttasks.go +++ b/pkg/hostman/guestman/guesttasks.go @@ -452,7 +452,7 @@ func (s *SGuestResumeTask) onStartRunning() { if options.HostOptions.SetVncPassword { s.SetVncPassword() } - s.syncMetadata() + s.SyncMetadataInfo() s.SyncStatus() timeutils2.AddTimeout(time.Second*5, s.SetCgroup) disksIdx := s.GetNeedMergeBackingFileDiskIndexs() @@ -481,18 +481,6 @@ func (s *SGuestResumeTask) onStreamComplete(disksIdx []int) { } } -func (s *SGuestResumeTask) syncMetadata() { - meta := jsonutils.NewDict() - meta.Set("__qemu_version", jsonutils.NewString(s.GetQemuVersionStr())) - meta.Set("__vnc_port", jsonutils.NewInt(int64(s.GetVncPort()))) - - if len(s.VncPassword) > 0 { - meta.Set("__vnc_password", jsonutils.NewString(s.VncPassword)) - } - - s.SyncMetadata(meta) -} - func (s *SGuestResumeTask) removeStatefile() { go s.CleanStatefiles() } diff --git a/pkg/hostman/guestman/qemu-kvm.go b/pkg/hostman/guestman/qemu-kvm.go index 3ca3323813..d1791909e6 100644 --- a/pkg/hostman/guestman/qemu-kvm.go +++ b/pkg/hostman/guestman/qemu-kvm.go @@ -360,8 +360,6 @@ func (s *SKVMGuestInstance) onReceiveQMPEvent(event *monitor.Event) { } } } - } else if event.Event == `"BLOCK_JOB_ERROR"` && s.IsMaster() { - modules.Servers.PerformAction(hostutils.GetComputeSession(context.Background()), s.GetId(), "mirror-job-failed", nil) } } @@ -388,6 +386,11 @@ func (s *SKVMGuestInstance) onGetQemuVersion(ctx context.Context, version string s.startDiskBackupMirror(ctx) if ctx != nil { s.DoResumeTask(ctx) + } else { + if options.HostOptions.SetVncPassword { + s.SetVncPassword() + } + s.SyncMetadataInfo() } } else { s.DoResumeTask(ctx) @@ -475,6 +478,9 @@ func (s *SKVMGuestInstance) IsMirrorJobSucc() bool { return false case v := <-res: if v != nil { + if len(v.Value()) == 0 { + return true + } mirrorSuccCount := 0 for _, val := range v.Value() { jobType, _ := val.GetString("type") @@ -1070,6 +1076,18 @@ func (s *SKVMGuestInstance) SetVncPassword() { func() { s.Monitor.SetVncPassword(s.GetVdiProtocol(), password, callback) }) } +func (s *SKVMGuestInstance) SyncMetadataInfo() { + meta := jsonutils.NewDict() + meta.Set("__qemu_version", jsonutils.NewString(s.GetQemuVersionStr())) + meta.Set("__vnc_port", jsonutils.NewInt(int64(s.GetVncPort()))) + + if len(s.VncPassword) > 0 { + meta.Set("__vnc_password", jsonutils.NewString(s.VncPassword)) + } + + s.SyncMetadata(meta) +} + func (s *SKVMGuestInstance) ListStateFilePaths() []string { var ret = []string{} if fileutils2.Exists(s.HomeDir()) {