diff --git a/pkg/cloudcommon/workmanager/manager.go b/pkg/cloudcommon/workmanager/manager.go index f15a15a01a..897a3a0529 100644 --- a/pkg/cloudcommon/workmanager/manager.go +++ b/pkg/cloudcommon/workmanager/manager.go @@ -32,10 +32,10 @@ func (w *SWorkManager) done() { // If delay task is not panic and task func return err is nil // task complete will be called, otherwise called task failed -// Params is interface for receive any type, task func should do type assert +// Params is interface for receive any type, task func should do type assertion func (w *SWorkManager) DelayTask(ctx context.Context, task DelayTaskFunc, params interface{}) { if ctx == nil || ctx.Value(appctx.APP_CONTEXT_KEY_TASK_ID) == nil { - w.DelayTaskWithoutTask(ctx, task, params) + w.DelayTaskWithoutReqctx(ctx, task, params) return } else { w.add() @@ -67,17 +67,18 @@ func (w *SWorkManager) DelayTask(ctx context.Context, task DelayTaskFunc, params } } -func (w *SWorkManager) DelayTaskWithoutTask(ctx context.Context, task DelayTaskFunc, params interface{}) { +// response task by self, did not depend on work manager +func (w *SWorkManager) DelayTaskWithoutReqctx(ctx context.Context, task DelayTaskFunc, params interface{}) { w.add() go func() { defer w.done() defer func() { if r := recover(); r != nil { - log.Errorln("DelayTaskWithoutTask panic: ", r) + log.Errorln("DelayTaskWithoutReqctx panic: ", r) } }() if _, err := task(ctx, params); err != nil { - log.Errorln("DelayTaskWithoutTask error:", err) + log.Errorln("DelayTaskWithoutReqctx error:", err) } }() } diff --git a/pkg/hostman/guestman/guesthandler.go b/pkg/hostman/guestman/guesthandler.go index 7b4c2337d7..41ea21428a 100644 --- a/pkg/hostman/guestman/guesthandler.go +++ b/pkg/hostman/guestman/guesthandler.go @@ -148,7 +148,7 @@ func guestSuspend(ctx context.Context, sid string, body jsonutils.JSONObject) (i if !guestManger.IsGuestExist(sid) { return nil, httperrors.NewNotFoundError("Guest %s not found", sid) } - hostutils.DelayTaskWithoutTask(ctx, guestManger.GuestSuspend, sid) + hostutils.DelayTaskWithoutReqctx(ctx, guestManger.GuestSuspend, sid) return nil, nil } diff --git a/pkg/hostman/guestman/guestman.go b/pkg/hostman/guestman/guestman.go index 8137fef9b7..3c83b711a8 100644 --- a/pkg/hostman/guestman/guestman.go +++ b/pkg/hostman/guestman/guestman.go @@ -222,7 +222,7 @@ func (m *SGuestManager) PrepareDeploy(sid string) error { func (m *SGuestManager) Monitor(sid, cmd string, callback func(string)) error { if guest, ok := m.Servers[sid]; ok { if guest.IsRunning() { - guest.monitor.SimpleCommand(cmd, callback) + guest.Monitor.SimpleCommand(cmd, callback) return nil } else { return httperrors.NewBadRequestError("Server stopped??") @@ -331,7 +331,7 @@ func (m *SGuestManager) GuestStart(ctx context.Context, sid string, body jsonuti func (m *SGuestManager) GuestStop(ctx context.Context, sid string, timeout int64) error { if guest, ok := m.Servers[sid]; !ok { - guest.ExecStopTask(ctx, timeout) + hostutils.DelayTaskWithoutReqctx(ctx, guest.ExecStopTask, timeout) return nil } else { return httperrors.NewNotFoundError("Guest %s not found", sid) @@ -359,7 +359,7 @@ func (m *SGuestManager) GuestSuspend(ctx context.Context, params interface{}) (j return nil, fmt.Errorf("Unknown params") } guest := m.Servers[sid] - guest.ExecSuspendTask() + guest.ExecSuspendTask(ctx) return nil, nil } diff --git a/pkg/hostman/guestman/guesttasks.go b/pkg/hostman/guestman/guesttasks.go index f876a75958..e3b3b5a8d8 100644 --- a/pkg/hostman/guestman/guesttasks.go +++ b/pkg/hostman/guestman/guesttasks.go @@ -2,15 +2,18 @@ package guestman import ( "context" + "fmt" + "regexp" "time" "yunion.io/x/jsonutils" "yunion.io/x/onecloud/pkg/hostman/hostutils" + "yunion.io/x/onecloud/pkg/hostman/storageman" "yunion.io/x/onecloud/pkg/util/timeutils2" ) type IGuestTasks interface { - Start(func(error)) + Start(func(...error)) } type SGuestStopTask struct { @@ -32,7 +35,7 @@ func NewGuestStopTask(guest *SKVMGuestInstance, ctx context.Context, timeout int func (s *SGuestStopTask) Start() { if s.IsRunning() && s.IsMonitorAlive() { // Do Powerdown, - s.monitor.SimpleCommand("system_powerdown", s.onPowerdownGuest) + s.Monitor.SimpleCommand("system_powerdown", s.onPowerdownGuest) } else { s.CheckGuestRunningLater() } @@ -45,7 +48,7 @@ func (s *SGuestStopTask) onPowerdownGuest(results string) { } func (s *SGuestStopTask) checkGuestRunning() { - if !s.IsRunning() || time.Now().Sub(*s.startPowerdown) > (s.timeout*time.Duration) { + if !s.IsRunning() || time.Now().Sub(*s.startPowerdown) > (s.timeout*time.Second) { s.Stop() // force stop hostutils.TaskComplete(s.ctx, nil) } else { @@ -54,7 +57,8 @@ func (s *SGuestStopTask) checkGuestRunning() { } func (s *SGuestStopTask) CheckGuestRunningLater() { - timeutils2.AddTimeout(time.Second*1, s.checkGuestRunning()) + time.Sleep(time.Second * 1) + s.checkGuestRunning() } type SGuestSyncConfigTaskExecutor struct { @@ -91,9 +95,9 @@ func (t *SGuestSyncConfigTaskExecutor) doCallback() { } } -func (t *SGuestSyncConfigTaskExecutor) runNextTaskCallback(err error) { +func (t *SGuestSyncConfigTaskExecutor) runNextTaskCallback(err ...error) { if err != nil { - t.errors = append(t.errors, err) + t.errors = append(t.errors, err...) } t.runNextTask() } @@ -102,12 +106,12 @@ type SGuestDiskSyncTask struct { guest *SKVMGuestInstance delDisks []jsonutils.JSONObject addDisks []jsonutils.JSONObject - cdrom string + cdrom *string callback func(error) } -func (d *SGuestDiskSyncTask) NewGuestDiskSyncTask(guest *SKVMGuestInstance, delDisks, addDisks []jsonutils.JSONObject, cdrom string) *SGuestDiskSyncTask { +func NewGuestDiskSyncTask(guest *SKVMGuestInstance, delDisks, addDisks []jsonutils.JSONObject, cdrom *string) *SGuestDiskSyncTask { return &SGuestDiskSyncTask{guest, delDisks, addDisks, cdrom} } @@ -129,11 +133,158 @@ func (d *SGuestDiskSyncTask) syncDisksConf() { d.addDisk(disk) return } - if len(d.cdrom) > 0 { + if d.cdrom != nil { d.changeCdrom() return } - d.callback() + d.callback(nil) } -def change_cdrom(self): \ No newline at end of file +func (d *SGuestDiskSyncTask) changeCdrom() { + d.guest.Monitor.GetBlocks(d.onGetBlockInfo) +} + +func (d *SGuestDiskSyncTask) onGetBlockInfo(results *jsonutils.JSONArray) { + var cdName string + for _, r := range results.Value() { + device, _ := r.GetString("device") + if regexp.MustCompile(`^ide\d+-cd\d+$`, device) { + cdName = device + break + } + } + if len(cdName) > 0 { + d.changeCdromContent(cdName) + } +} + +func (d *SGuestDiskSyncTask) changeCdromContent(cdName string) { + if *d.cdrom == "" { + d.guest.Monitor.EjectCdrom(cdName, d.OnChangeCdromContentSucc) + } else { + d.guest.Monitor.ChangeCdrom(cdName, *d.cdrom, d.OnChangeCdromContentSucc) + } +} + +func (d *SGuestDiskSyncTask) OnChangeCdromContentSucc(results string) { + d.cdrom = nil + d.syncDisksConf() +} + +func (d *SGuestDiskSyncTask) removeDisk(disk jsonutils.JSONObject) { + index, _ := disk.Int("index") + devId = fmt.Sprintf("drive_%s", index) + d.guest.Monitor.DriveDel(devId, + func(results string) { d.onRemoveDriveSucc(devId, results) }) +} + +func (d *SGuestDiskSyncTask) onRemoveDriveSucc(devId, results string) { + d.guest.Monitor.DeviceDel(devId, d.onRemoveDiskSucc) +} + +func (d *SGuestDiskSyncTask) onRemoveDiskSucc(results string) { + d.syncDisksConf() +} + +func (d *SGuestDiskSyncTask) addDisk(disk jsonutils.JSONObject) { + diskPath, _ := disk.GetString("path") + iDisk := storageman.GetManager().GetDiskByPath(diskPath) + if iDisk == nil { + d.syncDisksConf() + return + } + + var ( + diskIndex, _ = disk.Int("index") + aio, _ = disk.GetString("aio_mode") + diskDirver, _ = disk.GetString("driver") + cacheMode, _ = disk.GetString("cache_mode") + ) + + var params = map[string]string{ + "file": iDisk.GetPath(), + "if": "none", + "id": fmt.Sprintf("drive_%s", diskIndex), + "cache": cacheMode, + "aio": aio, + } + + var bus string + switch diskDirver { + case DISK_DRIVER_SCSI: + bus = "scsi.0" + case DISK_DRIVER_VIRTIO: + bus = d.guest.GetPciBus() + case DISK_DRIVER_IDE: + bus = fmt.Sprintf("ide.%d", diskIndex/2) + case DISK_DRIVER_SATA: + bus = fmt.Sprintf("ide.%d", diskIndex) + } + d.guest.Monitor.DriveAdd(bus, params, func(result string) { d.onAddDiskSucc(disk, result) }) +} + +func (d *SGuestDiskSyncTask) onAddDiskSucc(disk jsonutils.JSONObject, results string) { + var ( + diskIndex, _ = disk.Int("index") + diskDirver, _ = disk.GetString("driver") + dev = d.guest.GetDiskDeviceModel(diskDirver) + ) + + var params = map[string]interface{}{ + "drive": fmt.Sprintf("drive_%s", diskIndex), + "id": fmt.Sprintf("drive_%s", diskIndex), + } + + if diskDirver == DISK_DRIVER_VIRTIO { + params["addr"] = fmt.Sprintf("0x%x", d.guest.GetDiskAddr(diskIndex)) + } else if DISK_DRIVER_IDE == diskDirver { + params["unit"] = diskIndex % 2 + } + d.guest.Monitor.DeviceAdd(dev, params, d.onAddDeviceSucc) +} + +func (d *SGuestDiskSyncTask) onAddDeviceSucc(results string) { + d.syncDisksConf() +} + +type SGuestNetworkSyncTask struct { + guest *SKVMGuestInstance + delNics []jsonutils.JSONObject + addNics []jsonutils.JSONObject + errors []error + + callback func(...error) +} + +func (n *SGuestNetworkSyncTask) Start(callback func(...error)) { + n.callback = callback + n.syncNetworkConf() +} + +func (n *SGuestNetworkSyncTask) syncNetworkConf() { + if len(n.delNics) > 0 { + nic = n.delNics[len(n.delNics)-1] + n.delNics = n.delNics[:len(n.delNics)-1] + n.removeNic(nic) + return + } else if len(n.addNics) > 0 { + nic = n.addNics[len(n.addNics)-1] + n.addNics = n.addNics[:len(n.addNics)-1] + n.addNic(nic) + return + } else { + n.callback() + } +} + +func (n *SGuestNetworkSyncTask) removeNic(nic jsonutils.JSONObject) { + // pass +} + +func (n *SGuestNetworkSyncTask) addNic(nic jsonutils.JSONObject) { + // pass +} + +func NewGuestNetworkSyncTask(guest *SKVMGuestInstance, delNics, addNics jsonutils.JSONObject) *SGuestNetworkSyncTask { + return &SGuestNetworkSyncTask{guest, delNics, addNics, make(error, 0)} +} diff --git a/pkg/hostman/guestman/qemu-kvm.go b/pkg/hostman/guestman/qemu-kvm.go index bd1e792d13..b7b806361b 100644 --- a/pkg/hostman/guestman/qemu-kvm.go +++ b/pkg/hostman/guestman/qemu-kvm.go @@ -41,7 +41,7 @@ type SKVMGuestInstance struct { QemuVersion string Desc *jsonutils.JSONDict - monitor monitor.Monitor + Monitor monitor.Monitor manager *SGuestManager } @@ -283,7 +283,7 @@ func (s *SKVMGuestInstance) IsSuspend() bool { } func (s *SKVMGuestInstance) IsMonitorAlive() bool { - return s.monitor != nil && s.monitor.IsConnected() + return s.Monitor != nil && s.Monitor.IsConnected() } func (s *SKVMGuestInstance) ListStateFilePaths() []string { @@ -306,15 +306,15 @@ func (s *SKVMGuestInstance) StartMonitor(ctx context.Context) { func (s *SKVMGuestInstance) delayStartMonitor(ctx context.Context) { if options.HostOptions.EnableQmpMonitor && s.GetQmpMonitorPort(-1) > 0 { - s.monitor = monitor.NewQmpMonitor(s.onMonitorDisConnect, s.onMonitorTimeout, + s.Monitor = monitor.NewQmpMonitor(s.onMonitorDisConnect, s.onMonitorTimeout, func() { s.onMonitorConnected(ctx) }) - s.monitor.Connect("127.0.0.1", s.GetQmpMonitorPort(-1)) + s.Monitor.Connect("127.0.0.1", s.GetQmpMonitorPort(-1)) } } func (s *SKVMGuestInstance) onMonitorConnected(ctx context.Context) { log.Infof("Monitor connected ...") - s.monitor.GetVersion(func(v string) { + s.Monitor.GetVersion(func(v string) { s.onGetQemuVersion(ctx, v) }) } @@ -379,7 +379,7 @@ func (s *SKVMGuestInstance) saveVncPort(port int64) error { func (s *SKVMGuestInstance) SyncStatus() { if s.IsRunning() { - s.monitor.GetBlockJobs(s.CheckBlockOrRunning) + s.Monitor.GetBlockJobs(s.CheckBlockOrRunning) return } var status = "ready" @@ -468,9 +468,9 @@ func (s *SKVMGuestInstance) ExitCleanup(clearCgroup bool) { s.ClearCgroup(pid) } } - if s.monitor != nil { - s.monitor.Disconnect() - s.monitor = nil + if s.Monitor != nil { + s.Monitor.Disconnect() + s.Monitor = nil } } @@ -522,7 +522,8 @@ func (s *SKVMGuestInstance) scriptStop() bool { return true } -func (s *SKVMGuestInstance) ExecStopTask(ctx context.Context, timeout int64) { +func (s *SKVMGuestInstance) ExecStopTask(ctx context.Context, params interface{}) { + timeout := params.(int64) NewGuestStopTask(s, ctx, timeout).Start() } @@ -593,9 +594,9 @@ func (s *SKVMGuestInstance) compareDescDisks(newDesc jsonutils.JSONObject) ([]js return delDisks, addDisks } -func (s *SKVMGuestInstance) compareDescCdrom(newDesc jsonutils.JSONObject) string { +func (s *SKVMGuestInstance) compareDescCdrom(newDesc jsonutils.JSONObject) *string { if !s.Desc.Contains("cdrom") && !newDesc.Contains("cdrom") { - return "" + return nil } else if !s.Desc.Contains("cdrom") && newDesc.Contains("cdrom") { cdrom, _ := newDesc.GetString("cdrom", "path") return cdrom @@ -605,7 +606,7 @@ func (s *SKVMGuestInstance) compareDescCdrom(newDesc jsonutils.JSONObject) strin cdrom, _ := s.Desc.GetString("cdrom", "path") ncdrom, _ := newDesc.GetString("cdrom", "path") if cdrom == ncdrom { - return "" + return nil } else { return ncdrom } @@ -694,7 +695,7 @@ func (s *SKVMGuestInstance) SyncConfig(ctx context.Context, desc jsonutils.JSONO } } - if len(delDisks)+len(addDisks) > 0 || len(cdrom) > 0 { + if len(delDisks)+len(addDisks) > 0 || cdrom != nil { task := NewGuestDiskSyncTask(s, delDisks, addDisks, cdrom) runTaskNames = append(runTaskNames, jsonutils.NewString("disksync")) tasks = append(tasks, tasks) @@ -706,6 +707,6 @@ func (s *SKVMGuestInstance) SyncConfig(ctx context.Context, desc jsonutils.JSONO tasks = append(tasks, tasks) } - NewGuestSyncConfigTaskExecutor(ctx, s, tasks).Start(1) + NewGuestSyncConfigTaskExecutor(ctx, s, tasks, callBack).Start(1) return jsonutils.NewDict(jsonutils.JSONPair{"task", runTaskNames}), nil } diff --git a/pkg/hostman/guestman/qemu-kvmhelper.go b/pkg/hostman/guestman/qemu-kvmhelper.go index f645438f59..13e52ec18f 100644 --- a/pkg/hostman/guestman/qemu-kvmhelper.go +++ b/pkg/hostman/guestman/qemu-kvmhelper.go @@ -80,7 +80,7 @@ func (s *SKVMGuestInstance) isQ35() bool { return s.getMachine() == "q35" } -func (s *SKVMGuestInstance) getPciBus() string { +func (s *SKVMGuestInstance) GetPciBus() string { if s.isQ35() { return "pcie.0" } else { @@ -111,7 +111,7 @@ func (s *SKVMGuestInstance) getDriveDesc(disk jsonutils.JSONObject, format strin return cmd } -func (s *SKVMGuestInstance) getDiskAddr(idx int) int { +func (s *SKVMGuestInstance) GetDiskAddr(idx int) int { var base = 5 if s.IsVdiSpice() { base += 10 @@ -120,7 +120,7 @@ func (s *SKVMGuestInstance) getDiskAddr(idx int) int { return base + idx } -func (s *SKVMGuestInstance) getDiskDeviceModel(driver string) string { +func (s *SKVMGuestInstance) GetDiskDeviceModel(driver string) string { if driver == DISK_DRIVER_VIRTIO { return "virtio-blk-pci" } else if utils.IsInStringArray(driver, []string{DISK_DRIVER_SCSI, DISK_DRIVER_PVSCSI}) { @@ -139,10 +139,10 @@ func (s *SKVMGuestInstance) getDriveDesc(disk jsonutils.JSONObject) string { diskDriver, _ := disk.GetString("driver") var cmd = "" - cmd += fmt.Sprintf(" -device %s", s.getDiskDeviceModel(diskDriver)) + cmd += fmt.Sprintf(" -device %s", s.GetDiskDeviceModel(diskDriver)) cmd += fmt.Sprintf(",drive=drive_%s", diskIndex) if diskDriver == DISK_DRIVER_VIRTIO { - cmd += fmt.Sprintf(",bus=%s,addr=0x%x", s.getPciBus(), s.getDiskAddr(int(diskIndex))) + cmd += fmt.Sprintf(",bus=%s,addr=0x%x", s.GetPciBus(), s.GetDiskAddr(int(diskIndex))) } else if utils.IsInStringArray(iskDriver, []string{DISK_DRIVER_SCSI, DISK_DRIVER_PVSCSI}) { cmd += ",bus=scsi.0" } else if diskDriver == DISK_DRIVER_IDE { @@ -200,7 +200,7 @@ func (s *SKVMGuestInstance) getNicAddr(index int) string { if len(disks) > 10 { diskCnt = 20 } - return s.getDiskAddr(diskCnt + index) + return s.GetDiskAddr(diskCnt + index) } func (s *SKVMGuestInstance) getVnicDesc(nic jsonutils.JSONObject) string { @@ -399,7 +399,7 @@ func (s *SKVMGuestInstance) generateStartScript(data *jsonutils.JSONDict) string cmd += " -device hda-duplex,id=sound0-codec0,bus=sound0.0,cad=0" cmd += fmt.Sprintf(" -spice port=%d,password=87654312,seamless-migration=on", 5900+vncPort) // # ,streaming-video=all,playback-compression=on,jpeg-wan-compression=always,zlib-glz-wan-compression=always,image-compression=glz" % (5900+vnc_port) - cmd += fmt.Sprintf(" -device virtio-serial-pci,id=virtio-serial0,max_ports=16,bus=%s", s.getPciBus()) + cmd += fmt.Sprintf(" -device virtio-serial-pci,id=virtio-serial0,max_ports=16,bus=%s", s.GetPciBus()) cmd += " -chardev spicevmc,name=vdagent,id=vdagent" cmd += " -device virtserialport,nr=1,bus=virtio-serial0.0,chardev=vdagent,name=com.redhat.spice.0" diff --git a/pkg/hostman/hostinfo/hostbridge.go b/pkg/hostman/hostinfo/hostbridge.go index d1ba0732ea..41682f9168 100644 --- a/pkg/hostman/hostinfo/hostbridge.go +++ b/pkg/hostman/hostinfo/hostbridge.go @@ -18,8 +18,17 @@ type IBridgeDriver interface { Setup() error Exists() bool Interfaces() []string + + GetMac() string } +/* + def get_mac(self): + if self._bridge.mac is None: + self._bridge.fetch_config() + return self._bridge.mac +*/ + type SBaseBridgeDriver struct { bridge *netutils2.SNetInterface ip string @@ -42,6 +51,13 @@ func NewBaseBridgeDriver(bridge, inter, ip string) (*SBaseBridgeDriver, error) { return bd, nil } +func (d *SBaseBridgeDriver) GetMac() string { + if len(d.bridge.Mac) == 0 { + d.bridge.FetchConfig() + } + return d.bridge.Mac +} + func (d *SBaseBridgeDriver) BringupInterface() error { var infs = []*netutils2.SNetInterface{d.bridge} if d.inter != nil { diff --git a/pkg/hostman/hostinfo/hostinfo.go b/pkg/hostman/hostinfo/hostinfo.go index 0b3d61b533..cb65803148 100644 --- a/pkg/hostman/hostinfo/hostinfo.go +++ b/pkg/hostman/hostinfo/hostinfo.go @@ -1,6 +1,7 @@ package hostinfo import ( + "context" "fmt" "io/ioutil" "os" @@ -8,12 +9,16 @@ import ( "path" "regexp" "strings" + "time" + "yunion.io/x/jsonutils" "yunion.io/x/log" "yunion.io/x/pkg/utils" "yunion.io/x/onecloud/pkg/cloudcommon/storagetypes" + "yunion.io/x/onecloud/pkg/hostman/hostutils" "yunion.io/x/onecloud/pkg/hostman/options" + "yunion.io/x/onecloud/pkg/mcclient/modules" "yunion.io/x/onecloud/pkg/util/fileutils2" "yunion.io/x/onecloud/pkg/util/netutils2" "yunion.io/x/onecloud/pkg/util/qemutils" @@ -43,8 +48,13 @@ type SHostInfo struct { // storageManager *storageman.SStorageManager MasterNic *netutils2.SNetInterface + Nics []*SNIC - Nics []*SNIC + Zone string + ZoneId string + Cloudregion string + CloudregionId string + ZoneManagerUri string } func (h *SHostInfo) IsKvmSupport() bool { @@ -582,8 +592,110 @@ func (h *SHostInfo) detectiveOvsVersion() { } } -func (h *SHostInfo) StartRegister() { - // TODO +func (h *SHostInfo) GetMasterIp() string { + if h.MasterNic != nil { + return h.MasterNic.Addr + } + for _, n := range h.Nics { + if len(n.Ip) > 0 { + return n.Ip + } + } + return "" +} + +func (h *SHostInfo) GetMasterMac() string { + if h.MasterNic != nil { + return h.MasterNic.Mac + } + for _, n := range h.Nics { + if len(n.Ip) > 0 { + return n.BridgeDev.GetMac() + } + } +} + +func (h *SHostInfo) StartRegister(delay int) { + timeutils2.AddTimeout(delay*time.Second, h.register) +} + +func (h *SHostInfo) register() { + if !h.isRegistered { + h.fetch_access_network_info() + } +} + +func (h *SHostInfo) onFail() { + log.Errorln("register failed, try 30 seconds later...") + h.StartRegister(30) +} + +func (h *SHostInfo) fetchAccessNetworkInfo() { + masterIp := h.GetMasterIp() + if len(masterIp) == 0 { + panic("master ip not found") + } + params := jsonutils.NewDict() + params.Set("ip", jsonutils.NewString(masterIp)) + params.Set("limit", jsonutils.NewInt(0)) + wire, err := hostutils.GetWireOfIp(context.Background(), params) + if err != nil { + log.Errorln(err) + h.onFail() + } else { + h.ZoneId, err = wire.GetString("zone_id") + if err != nil { + log.Errorln(err) + h.onFail() + } else { + h.getZoneInfo(h.ZoneId, false) + } + } +} + +func (h *SHostInfo) getZoneInfo(zoneId string, standalone bool) { + var params = jsonutils.NewDict() + params.Set("standalone", jsonutils.NewBool(standalone)) + res, err := modules.Zones.Get(hostutils.GetComputeSession(context.Background()), + zoneId, params) + if err != nil { + log.Errorln(err) + h.onFail(err) + } + + h.Zone, _ = res.GetString("name") + h.ZoneId, _ = res.GetString("id") + h.Cloudregion = res.GetString("cloudregion") + h.CloudregionId = res.GetString("cloudregion_id") + if res.Contains("manager_uri") { + h.ZoneManagerUri, _ = res.GetString("manager_uri") + } + if !standalone { + h.getHostInfo(h.ZoneId) + } +} + +func (h *SHostInfo) getHostInfo(zoneId string) { + masterMac := h.GetMasterMac() + if len(masterMac) == 0 { + panic("master mac not found") + } + params := jsonutils.NewDict() + params.Set("any_mac", jsonutils.NewString(masterMac)) + res, err := modules.Hosts.List(hostutils.GetComputeSession(context.Background()), params) + if err != nil { + log.Errorln(err) + h.onFail() + } + if len(res.Data) == 0 { + h.updateHostRecord() + } else { + host := res.Data[0] + name, _ := host.GetString("name") + id, _ := host.GetString("id") + h.setHostname(name) + + } } func NewHostInfo() (*SHostInfo, error) { diff --git a/pkg/hostman/hostutils/hostutils.go b/pkg/hostman/hostutils/hostutils.go index 458381b012..65c6fb1e9f 100644 --- a/pkg/hostman/hostutils/hostutils.go +++ b/pkg/hostman/hostutils/hostutils.go @@ -2,6 +2,7 @@ package hostutils import ( "context" + "fmt" "net/http" "yunion.io/x/jsonutils" @@ -37,6 +38,24 @@ func TaskComplete(ctx context.Context, params jsonutils.JSONObject) { } } +func GetWireOfIp(ctx context.Context, params jsonutils.JSONObject) (jsonutils.JSONObject, error) { + res, err := modules.Networks.List(GetComputeSession(ctx), params) + if err != nil { + return nil, err + } + + if len(res.Data) == 1 { + wireId, _ := res.Data[0].GetString("wire_id") + return GetWireInfo(ctx, wireId) + } else { + return nil, fmt.Errorf("Fail to get network info: no networks") + } +} + +func GetWireInfo(ctx context.Context, wireId string) (jsonutils.JSONObject, error) { + return modules.Wires.Get(GetComputeSession(ctx), wireId, nil) +} + func UpdateServerStatus(ctx context.Context, sid, status string) (jsonutils.JSONObject, error) { var body = jsonutils.NewDict() var stats = jsonutils.NewDict() @@ -75,8 +94,8 @@ func DelayTask(ctx context.Context, task workmanager.DelayTaskFunc, params inter wm.DelayTask(ctx, task, params) } -func DelayTaskWithoutTask(ctx context.Context, task workmanager.DelayTaskFunc, params interface{}) { - wm.DelayTaskWithoutTask(ctx, task, params) +func DelayTaskWithoutReqctx(ctx context.Context, task workmanager.DelayTaskFunc, params interface{}) { + wm.DelayTaskWithoutReqctx(ctx, task, params) } func init() { diff --git a/pkg/hostman/monitor/hmp.go b/pkg/hostman/monitor/hmp.go index 2b3a78dfa6..45566c4ef0 100644 --- a/pkg/hostman/monitor/hmp.go +++ b/pkg/hostman/monitor/hmp.go @@ -3,10 +3,13 @@ package monitor import ( "bufio" "bytes" + "fmt" "io" + "regexp" "strings" "time" + "yunion.io/x/jsonutils" "yunion.io/x/log" ) @@ -42,6 +45,10 @@ func (m *HmpMonitor) hmpSplitFunc(data []byte, atEOF bool) (advance int, token [ return 0, nil, nil } +func (m *HmpMonitor) actionResult(res string) string { + return res +} + func (m *HmpMonitor) read(r io.Reader) { if !m.checkReading() { return @@ -171,3 +178,84 @@ func (m *HmpMonitor) SimpleCommand(cmd string, callback StringCallback) { func (m *HmpMonitor) GetVersion(callback StringCallback) { m.Query("info version", callback) } + +func (m *HmpMonitor) GetBlocks(callback func(*jsonutils.JSONArray)) { + var cb = func(res string) { + var lines = strings.Split(res, "\r\n") + var mergedOutput = []string{} + + // merge output + for _, line := range lines { + parts := regexp.MustCompile(`\s+`).Split(line, -1) + if parts[0][len(parts[0])-1] == ':' { + mergedOutput = append(mergedOutput, "") + } else if regexp.MustCompile(`\(#block\d+\):`).MatchString(line) { + mergedOutput = append(mergedOutput, "") + } + mergedOutput[len(mergedOutput)-1] = mergedOutput[len(mergedOutput)-1] + " " + line + mergedOutput[len(mergedOutput)-1] = strings.TrimSpace(mergedOutput[len(mergedOutput)-1]) + } + + // parse to json + var outputJson = jsonutils.NewArray() + for _, line := range mergedOutput { + parts := regexp.MustCompile(`\s+`).Split(line, -1) + if parts[0][len(parts[0])-1] == ':' || + regexp.MustCompile(`\(#block\d+\):`).MatchString(parts[1]) { + + drv := jsonutils.NewDict() + drv.Set("device", jsonutils.NewString(parts[0])) + if parts[0][len(parts[0])-1] == ':' { + drv.Set("device", jsonutils.NewString(parts[0][:len(parts)-1])) + } + if regexp.MustCompile(`\(#block\d+\):`).MatchString(parts[1]) { + inserted := jsonutils.NewDict() + inserted.Set("file", jsonutils.NewString(parts[2])) + for i := 0; i < len(parts)-2; i++ { + if parts[i] == "Backing" && parts[i+1] == "file:" { + inserted.Set("backing_file", jsonutils.NewString(parts[i+2])) + break + } + } + } + outputJson.Add(drv) + } + } + + callback(outputJson) + } + + m.Query("info block", cb) +} + +func (m *HmpMonitor) EjectCdrom(dev string, callback StringCallback) { + m.Query(fmt.Sprintf("eject -f %s", dev), callback) +} + +func (m *HmpMonitor) ChangeCdrom(dev string, path string, callback StringCallback) { + m.Query(fmt.Sprintf("change %s %s", dev, path), callback) +} + +func (m *HmpMonitor) DriveDel(idstr string, callback StringCallback) { + m.Query(fmt.Sprintf("drive_del %s", idstr), callback) +} + +func (m *HmpMonitor) DeviceDel(idstr string, callback StringCallback) { + m.Query(fmt.Sprintf("device_del %s", idstr), callback) +} + +func (m *HmpMonitor) DriveAdd(bus string, params map[string]string, callback StringCallback) { + var paramsKvs = []string{} + for k, v := range params { + paramsKvs = append(paramsKvs, fmt.Sprintf("%s=%s", k, v)) + } + m.Query(fmt.Sprintf("drive_add %s %s", bus, strings.Join(paramsKvs, ",")), callback) +} + +func (m *HmpMonitor) DeviceAdd(dev string, params map[string]interface{}, callback StringCallback) { + var paramsKvs = []string{} + for k, v := range params { + paramsKvs = append(paramsKvs, fmt.Sprintf("%s=%v", k, v)) + } + m.Query(fmt.Sprintf("device_add %s,%s", dev, strings.Join(paramsKvs, ",")), callback) +} diff --git a/pkg/hostman/monitor/monitor.go b/pkg/hostman/monitor/monitor.go index 1358ae3a64..efb1c6e16d 100644 --- a/pkg/hostman/monitor/monitor.go +++ b/pkg/hostman/monitor/monitor.go @@ -6,6 +6,7 @@ import ( "sync" "time" + "yunion.io/x/jsonutils" "yunion.io/x/log" ) @@ -18,8 +19,18 @@ type Monitor interface { // The callback function will be called in another goroutine SimpleCommand(cmd string, callback StringCallback) - QueryStatus(callback StringCallback) - GetVersion(callback StringCallback) + QueryStatus(StringCallback) + GetVersion(StringCallback) + + GetBlocks(callback func(*jsonutils.JSONArray)) + EjectCdrom(dev string, callback StringCallback) + ChangeCdrom(dev string, path string, callback StringCallback) + + DriveDel(idstr string, callback StringCallback) + DeviceDel(idstr string, callback StringCallback) + + DriveAdd(bus string, params map[string]string, callback StringCallback) + DeviceAdd(dev string, params map[string]interface{}, callback StringCallback) } type MonitorErrorFunc func(error) diff --git a/pkg/hostman/monitor/qmp.go b/pkg/hostman/monitor/qmp.go index f8cd756ad3..0756ebb9bf 100644 --- a/pkg/hostman/monitor/qmp.go +++ b/pkg/hostman/monitor/qmp.go @@ -9,6 +9,7 @@ import ( "strings" "time" + "yunion.io/x/jsonutils" "yunion.io/x/log" ) @@ -100,6 +101,14 @@ func NewQmpMonitor(OnMonitorDisConnect, OnMonitorTimeout MonitorErrorFunc, return m } +func (m *QmpMonitor) actionResult(res *Response) string { + if res.ErrorVal != nil { + return res.ErrorVal.Error() + } else { + return "" + } +} + func (m *QmpMonitor) callBack(res *Response) { m.mutex.Lock() if len(m.callbackQueue) == 0 { @@ -308,3 +317,144 @@ func (m *QmpMonitor) parseVersion(callback StringCallback) qmpMonitorCallBack { } } } + +func (m *QmpMonitor) GetBlocks(callback func(*jsonutils.JSONArray)) { + var cb = func(res *Response) { + if res.ErrorVal != nil { + callback(nil) + } + jr, err := jsonutils.Parse(res.Return) + if err != nil { + log.Errorf("Get block error %s", err) + callback(nil) + } + callback(jr.(*jsonutils.JSONArray)) + } + + cmd := &Command{Execute: "query-block"} + m.Query(cmd, cb) +} + +func (m *QmpMonitor) ChangeCdrom(dev string, path string, callback StringCallback) { + var ( + args = map[string]interface{}{ + "arguments": map[string]interface{}{ + "device": dev, + "target": path, + }, + } + cmd = &Command{ + Execute: "change", + Args: args, + } + + cb = func(res *Response) { + callback(m.actionResult(res)) + } + ) + + m.Query(cmd, cb) +} + +func (m *QmpMonitor) EjectCdrom(dev string, callback StringCallback) { + var ( + args = map[string]interface{}{ + "arguments": map[string]interface{}{ + "device": dev, + "force": true, + }, + } + cmd = &Command{ + Execute: "eject", + Args: args, + } + + cb = func(res *Response) { + callback(m.actionResult(res)) + } + ) + + m.Query(cmd, cb) +} + +func (m *QmpMonitor) DriveDel(idstr string, callback StringCallback) { + var ( + args = map[string]interface{}{ + "arguments": map[string]interface{}{ + "device": idstr, + }, + } + cmd = &Command{ + Execute: "drive_del", + Args: args, + } + + cb = func(res *Response) { + callback(m.actionResult(res)) + } + ) + + m.Query(cmd, cb) +} + +func (m *QmpMonitor) DeviceDel(idstr string, callback StringCallback) { + var ( + args = map[string]interface{}{ + "arguments": map[string]interface{}{ + "device": idstr, + }, + } + cmd = &Command{ + Execute: "device_del", + Args: args, + } + + cb = func(res *Response) { + callback(m.actionResult(res)) + } + ) + + m.Query(cmd, cb) +} + +func (m *QmpMonitor) DriveAdd(bus string, params map[string]string, callback StringCallback) { + var ( + args = map[string]interface{}{ + "arguments": map[string]interface{}{ + "bus": bus, + "params": params, + }, + } + cmd = &Command{ + Execute: "drive_add", + Args: args, + } + + cb = func(res *Response) { + callback(m.actionResult(res)) + } + ) + + m.Query(cmd, cb) +} + +func (m *QmpMonitor) DeviceAdd(dev string, params map[string]interface{}, callback StringCallback) { + var ( + args = map[string]interface{}{ + "arguments": map[string]interface{}{ + "device": dev, + "params": params, + }, + } + cmd = &Command{ + Execute: "device_add", + Args: args, + } + + cb = func(res *Response) { + callback(m.actionResult(res)) + } + ) + + m.Query(cmd, cb) +}