From a053e372ef22862ffc5033a52aace70529d088b6 Mon Sep 17 00:00:00 2001 From: wanyaoqi Date: Fri, 21 Dec 2018 17:53:51 +0800 Subject: [PATCH] 20181221 temp commit --- pkg/cloudcommon/httpclients/compute_client.go | 23 ++++ pkg/cloudcommon/workmanager/manager.go | 31 ++++- pkg/hostman/guestman/guesthandler.go | 73 ++++++------ pkg/hostman/guestman/guesthelper.go | 9 ++ pkg/hostman/guestman/guestman.go | 50 +++++--- pkg/hostman/guestman/qemu-kvm.go | 108 ++++++++++++++---- pkg/hostman/guestman/utils.go | 1 - pkg/hostman/hostinfo/host_cpu.go | 86 +++++++++++++- pkg/hostman/hostinfo/hostinfo.go | 25 ++-- pkg/hostman/hostinfo/hostmemory.go | 65 +++++++++++ pkg/hostman/monitor/monitor.go | 8 ++ pkg/hostman/options/options.go | 2 + pkg/hostman/storageman/core.go | 103 ----------------- pkg/hostman/storageman/diskbase.go | 49 ++++++++ pkg/hostman/storageman/localdisk.go | 25 ++++ pkg/hostman/storageman/localstorage.go | 26 ----- pkg/hostman/storageman/storagebase.go | 41 +++++++ pkg/hostman/storageman/storagemanager.go | 67 +++++++++++ 18 files changed, 572 insertions(+), 220 deletions(-) create mode 100644 pkg/hostman/guestman/guesthelper.go delete mode 100644 pkg/hostman/guestman/utils.go create mode 100644 pkg/hostman/hostinfo/hostmemory.go delete mode 100644 pkg/hostman/storageman/core.go create mode 100644 pkg/hostman/storageman/diskbase.go create mode 100644 pkg/hostman/storageman/localdisk.go create mode 100644 pkg/hostman/storageman/storagebase.go create mode 100644 pkg/hostman/storageman/storagemanager.go diff --git a/pkg/cloudcommon/httpclients/compute_client.go b/pkg/cloudcommon/httpclients/compute_client.go index 365e4b152e..153464ecf1 100644 --- a/pkg/cloudcommon/httpclients/compute_client.go +++ b/pkg/cloudcommon/httpclients/compute_client.go @@ -6,6 +6,9 @@ import ( "yunion.io/x/jsonutils" "yunion.io/x/log" + + "yunion.io/x/onecloud/pkg/appctx" + "yunion.io/x/onecloud/pkg/cloudcommon/httpclients" "yunion.io/x/onecloud/pkg/hostman/options" ) @@ -65,3 +68,23 @@ func (c *SComputeClient) UpdateServerStatus(sid, status string) { body.Set("server", stus) c.Request(context.Background(), "POST", url, nil, body, false) } + +func TaskFailed(ctx context.Context, reason string) error { + if taskId := ctx.Value(appctx.APP_CONTEXT_KEY_TASK_ID); taskId != nil { + httpclients.GetDefaultComputeClient().TaskFail(ctx, taskId.(string), reason) + return nil + } else { + log.Errorln("Reqeuest task failed missing task id, with reason(%s)", reason) + return fmt.Errorf("Reqeuest task failed missing task id") + } +} + +func TaskComplete(ctx context.Context, data jsonutils.JSONObject) error { + if taskId := ctx.Value(appctx.APP_CONTEXT_KEY_TASK_ID); taskId != nil { + httpclients.GetDefaultComputeClient().TaskComplete(ctx, taskId.(string), data, 0) + return nil + } else { + log.Errorln("Reqeuest task complete missing task id") + return fmt.Errorf("Reqeuest task complete missing task id") + } +} diff --git a/pkg/cloudcommon/workmanager/manager.go b/pkg/cloudcommon/workmanager/manager.go index f0ffd344ed..e8a7ff5497 100644 --- a/pkg/cloudcommon/workmanager/manager.go +++ b/pkg/cloudcommon/workmanager/manager.go @@ -7,9 +7,10 @@ import ( "yunion.io/x/jsonutils" "yunion.io/x/log" + "yunion.io/x/onecloud/pkg/cloudcommon/httpclients" ) -type DelayTaskFunc func(ctx context.Context, params jsonutils.JSONObject) +type DelayTaskFunc func(ctx context.Context, params interface{}) (jsonutils.JSONObject, error) type SWorkManager struct { curCount int32 @@ -23,11 +24,29 @@ func (w *SWorkManager) done() { atomic.AddInt32(&w.curCount, -1) } -func (w *SWorkManager) DelayTask(task func()) { +func (w *SWorkManager) DelayTask(ctx context.Context, task DelayTaskFunc, params interface{}) { w.add() go func() { defer w.done() - task() + defer func() { + if r := recover(); r != nil { + log.Errorln("Delay task recover: ", r) + switch val := r.(type) { + case string: + httpclients.TaskFailed(ctx, val) + case error: + httpclients.TaskFailed(ctx, val.Error()) + default: + httpclients.TaskFailed(ctx, "Unknown panic") + } + } + }() + res, err := task(ctx, params) + if err != nil { + httpclients.TaskFailed(ctx, err.Error()) + } else { + httpclients.TaskComplete(ctx, res) + } }() } @@ -35,10 +54,14 @@ func (w *SWorkManager) Stop() { log.Infof("WorkManager To stop, wait for workers ...") for w.curCount > 0 { log.Warningf("Busy workers count %d, waiting stopped", w.curCount) - time.Sleep(1) + time.Sleep(1 * time.Second) } } func NewWorkManger() *SWorkManager { return &SWorkManager{} } + +// type ITaskPramas interface { +// To +// } diff --git a/pkg/hostman/guestman/guesthandler.go b/pkg/hostman/guestman/guesthandler.go index 5bc5483cab..c5b7765920 100644 --- a/pkg/hostman/guestman/guesthandler.go +++ b/pkg/hostman/guestman/guesthandler.go @@ -22,17 +22,6 @@ func AddGuestTaskHandler(prefix string, app *appsrv.Application) { app.AddHandler("DELETE", "/servers/", auth.Authenticate(deleteGuest)) } -func getStatus(ctx context.Context, w http.ResponseWriter, r *http.Request) { - params, _, _ := appsrv.FetchEnv(ctx, w, r) - var status = guestManger.Status(params[""]) - appsrv.SendStruct(w, strDict{"status": status}) -} - -func cpusetBalance(ctx context.Context, w http.ResponseWriter, r *http.Request) { - wm.DelayTask(func() { guestManger.CpusetBalance(ctx) }) - responseOk(ctx, w) -} - func guestActions(ctx context.Context, w http.ResponseWriter, r *http.Request) { params, _, body := appsrv.FetchEnv(ctx, w, r) var sid = params[""] @@ -51,6 +40,17 @@ func guestActions(ctx context.Context, w http.ResponseWriter, r *http.Request) { } } +func getStatus(ctx context.Context, w http.ResponseWriter, r *http.Request) { + params, _, _ := appsrv.FetchEnv(ctx, w, r) + var status = guestManger.Status(params[""]) + appsrv.SendStruct(w, strDict{"status": status}) +} + +func cpusetBalance(ctx context.Context, w http.ResponseWriter, r *http.Request) { + wm.DelayTask(ctx, guestManger.CpusetBalance, nil) + responseOk(ctx, w) +} + func deleteGuest(ctx context.Context, w http.ResponseWriter, r *http.Request) { params, _, body := appsrv.FetchEnv(ctx, w, r) var sid = params[""] @@ -59,22 +59,47 @@ func deleteGuest(ctx context.Context, w http.ResponseWriter, r *http.Request) { if err != nil { response(ctx, w, err) } else { - wm.DelayTask(func() { guest.CleanGuest(ctx, migrated) }) + // TODO: CleanGuest + wm.DelayTask(ctx, guest.CleanGuest, migrated) response(ctx, w, map[string]bool{"delay_clean": true}) } } +func responseOk(ctx context.Context, w http.ResponseWriter) { + response(ctx, w, strDict{"result": "ok"}) +} + +func response(ctx context.Context, w http.ResponseWriter, res interface{}) { + if taskId := ctx.Value(appctx.APP_CONTEXT_KEY_TASK_ID); taskId != nil { + w.Header().Set("X-Request-Id", taskId.(string)) + } + switch res.(type) { + case string: + appsrv.Send(w, res.(string)) + case jsonutils.JSONObject: + appsrv.SendJSON(w, res.(jsonutils.JSONObject)) + case error: + httperrors.GeneralServerError(w, res.(error)) + default: + appsrv.SendStruct(w, res) + } +} + func doCreate(ctx context.Context, sid string, body jsonutils.JSONObject) (interface{}, error) { err := guestManger.PrepareCreate(sid) if err != nil { return nil, httperrors.NewBadRequestError(err.Error()) } - wm.DelayTask(func() { guestManger.DoDeploy(ctx, sid, body, true) }) + wm.DelayTask(ctx, guestManger.DoDeploy, &SGuestDeploy{sid, body, true}) return nil, nil } func doDeploy(ctx context.Context, sid string, body jsonutils.JSONObject) (interface{}, error) { - // TODO + err := guestManger.PrepareDeploy(sid) + if err != nil { + return nil, httperrors.NewBadRequestError(err.Error()) + } + wm.DelayTask(ctx, guestManger.DoDeploy, &SGuestDeploy{sid, body, false}) return nil, nil } @@ -107,26 +132,6 @@ func doMonitor(ctx context.Context, sid string, body jsonutils.JSONObject) (inte } } -func responseOk(ctx context.Context, w http.ResponseWriter) { - response(ctx, w, strDict{"result": "ok"}) -} - -func response(ctx context.Context, w http.ResponseWriter, res interface{}) { - if taskId := ctx.Value(appctx.APP_CONTEXT_KEY_TASK_ID); taskId != nil { - w.Header().Set("X-Request-Id", taskId.(string)) - } - switch res.(type) { - case string: - appsrv.Send(w, res.(string)) - case jsonutils.JSONObject: - appsrv.SendJSON(w, res.(jsonutils.JSONObject)) - case error: - httperrors.GeneralServerError(w, res.(error)) - default: - appsrv.SendStruct(w, res) - } -} - var actionFuncs = map[string]actionFunc{ "create": doCreate, "deploy": doDeploy, diff --git a/pkg/hostman/guestman/guesthelper.go b/pkg/hostman/guestman/guesthelper.go new file mode 100644 index 0000000000..284076a9c7 --- /dev/null +++ b/pkg/hostman/guestman/guesthelper.go @@ -0,0 +1,9 @@ +package guestman + +import "yunion.io/x/jsonutils" + +type SGuestDeploy struct { + sid string + body jsonutils.JSONObject + isInit bool +} diff --git a/pkg/hostman/guestman/guestman.go b/pkg/hostman/guestman/guestman.go index 32fb5b9a15..b9492ce665 100644 --- a/pkg/hostman/guestman/guestman.go +++ b/pkg/hostman/guestman/guestman.go @@ -177,13 +177,26 @@ func (m *SGuestManager) PrepareCreate(sid string) error { m.ServersLock.Lock() defer m.ServersLock.Unlock() if _, ok := m.Servers[sid]; ok { - return httperrors.NewBadRequestError("Guest %s exists", sid) + return fmt.Errorf("Guest %s exists", sid) } guest := NewKVMGuestInstance(sid, m) m.Servers[sid] = guest return guest.PrepareDir() } +func (m *SGuestManager) PrepareDeploy(sid string) error { + m.ServersLock.Lock() + defer m.ServersLock.Unlock() + if guest, ok := m.Servers[sid]; !ok { + return fmt.Errorf("Guest %s not exists", sid) + } else { + if guest.IsRunning() || guest.IsSuspend() { + return fmt.Errorf("Cannot deploy on running/suspend guest") + } + } + return nil +} + func (m *SGuestManager) Monitor(sid, cmd string, callback func(string)) error { if guest, ok := m.Servers[sid]; ok { if guest.IsRunning() { @@ -197,41 +210,44 @@ func (m *SGuestManager) Monitor(sid, cmd string, callback func(string)) error { } } -func (m *SGuestManager) DoDeploy(ctx context.Context, sid string, body jsonutils.JSONObject, isInit bool) { - guest, ok := m.Servers[sid] +// Delay process +func (m *SGuestManager) DoDeploy(ctx context.Context, params interface{}) (jsonutils.JSONObject, error) { + deployParams, ok := params.(*SGuestDeploy) + if !ok { + return nil, fmt.Errorf("Unknown params") + } + guest, ok := m.Servers[deployParams.sid] if ok { - desc, _ := body.Get("desc") + desc, _ := deployParams.body.Get("desc") if desc != nil { guest.SaveDesc(desc) } - if jsonutils.QueryBoolean(body, "k8s_pod", false) { - TaskComplete(ctx, nil) - return + if jsonutils.QueryBoolean(deployParams.body, "k8s_pod", false) { + return nil, nil } - // TODO - publicKey := sshkeys.GetKeys(body) - deploys, _ := body.GetArray("deploys") - password, _ := body.GetString("password") - resetPassword := jsonutils.QueryBoolean(body, "reset_password", false) + publicKey := sshkeys.GetKeys(deployParams.body) + deploys, _ := deployParams.body.GetArray("deploys") + password, _ := deployParams.body.GetString("password") + resetPassword := jsonutils.QueryBoolean(deployParams.body, "reset_password", false) if resetPassword && len(password) == 0 { password = seclib.RandomPassword(12) } guestInfo, err := guest.DeployFs(&guestfs.SDeployInfo{ - publicKey, deploys, password, isInit}) + publicKey, deploys, password, deployParams.isInit}) if err != nil { log.Errorf("Deploy guest fs error: %s", err) - TaskFailed(ctx, err.Error()) + return nil, err } else { - TaskComplete(ctx, guestInfo) + return guestInfo, nil } } else { - TaskFailed(ctx, fmt.Sprinft("Guest %s not found", sid)) + return nil, fmt.Errorf("Guest %s not found", sid) } } // delay cpuset balance -func (m *SGuestManager) CpusetBalance(ctx context.Context) { +func (m *SGuestManager) CpusetBalance(ctx context.Context, params interface{}) (jsonutils.JSONObject, error) { // TODO } diff --git a/pkg/hostman/guestman/qemu-kvm.go b/pkg/hostman/guestman/qemu-kvm.go index 4213ca7dbc..7a1da1bbdd 100644 --- a/pkg/hostman/guestman/qemu-kvm.go +++ b/pkg/hostman/guestman/qemu-kvm.go @@ -16,7 +16,6 @@ import ( "yunion.io/x/log" "yunion.io/x/pkg/util/regutils" - "yunion.io/x/onecloud/pkg/appctx" "yunion.io/x/onecloud/pkg/cloudcommon" "yunion.io/x/onecloud/pkg/cloudcommon/httpclients" "yunion.io/x/onecloud/pkg/hostman/guestfs" @@ -71,6 +70,10 @@ func (s *SKVMGuestInstance) GetPidFilePath() string { return path.Join(s.HomeDir(), "pid") } +func (s *SKVMGuestInstance) GetVncFilePath() string { + return path.Join(s.HomeDir(), "vnc") +} + func (s *SKVMGuestInstance) GetPid() int { pidFile := s.GetPidFilePath() fi, err := os.Stat(pidFile) @@ -196,7 +199,7 @@ func (s *SKVMGuestInstance) onAsyncScriptStart(ctx context.Context, isStarted bo } else { log.Infof("Async start server %s failed: %s!!!", s.GetName(), err) if ctx != nil { - TaskFailed(ctx, fmt.Sprintf("Async start server failed: %s", err)) + httpclients.TaskFailed(ctx, fmt.Sprintf("Async start server failed: %s", err)) } s.SyncStatus() } @@ -286,7 +289,7 @@ func (s *SKVMGuestInstance) onGetQemuVersion(ctx context.Context, version string migratePort, _ := s.Desc.Get("live_migrate_dest_port") body := jsonutils.NewDict( jsonutils.JSONPair{"live_migrate_dest_port", migratePort}) - TaskComplete(ctx, body) + httpclients.TaskComplete(ctx, body) } else if jsonutils.QueryBoolean(s.Desc, "is_slave", false) { // TODO } else if jsonutils.QueryBoolean(s.Desc, "is_master", false) && ctx == nil { @@ -343,26 +346,6 @@ func (s *SKVMGuestInstance) SyncStatus() { httpclients.GetDefaultComputeClient().UpdateServerStatus(s.GetId(), status) } -func TaskFailed(ctx context.Context, reason string) error { - if taskId := ctx.Value(appctx.APP_CONTEXT_KEY_TASK_ID); taskId != nil { - httpclients.GetDefaultComputeClient().TaskFail(ctx, taskId.(string), reason) - return nil - } else { - log.Errorln("Reqeuest task failed missing task id, with reason(%s)", reason) - return fmt.Errorf("Reqeuest task failed missing task id") - } -} - -func TaskComplete(ctx context.Context, data jsonutils.JSONObject) error { - if taskId := ctx.Value(appctx.APP_CONTEXT_KEY_TASK_ID); taskId != nil { - httpclients.GetDefaultComputeClient().TaskComplete(ctx, taskId.(string), data, 0) - return nil - } else { - log.Errorln("Reqeuest task complete missing task id") - return fmt.Errorf("Reqeuest task complete missing task id") - } -} - func (s *SKVMGuestInstance) SaveDesc(desc jsonutils.JSONObject) error { // TODO // bw_info = self._get_bw_info() @@ -394,3 +377,82 @@ func (s *SKVMGuestInstance) DeployFs(deployInfo *guestfs.SDeployInfo) (jsonutils return nil, fmt.Errorf("Guest dosen't have disk ??") } } + +func (s *SKVMGuestInstance) CleanGuest(ctx context.Context, params interface{}) (jsonutils.JSONObject, error) { + migrated, ok := params.(bool) + if !ok { + return nil, fmt.Errorf("Unknown params") + } + if err := s.StartDelete(ctx, migrated); err != nil { + return nil, err + } + return nil, nil +} + +func (s *SKVMGuestInstance) StartDelete(ctx context.Context, migrated bool) error { + for s.IsRunning() { + s.ForceStop() + time.Sleep(time.Second * 1) + } + return s.Delete(ctx, migrated) +} + +func (s *SKVMGuestInstance) ForceStop() bool { + s.ExitCleanup(true) + if s.IsRunning() { + err := exec.Command("kill", "-9", fmt.Sprintf("%d", s.GetPid())).Run() + if err != nil { + log.Errorln(err) + return false + } + for _, f := range s.GetCleanFiles() { + err := exec.Command("rm", "-f", f).Run() + if err != nil { + log.Errorln(err) + return false + } + } + return true + } + return false +} + +func (s *SKVMGuestInstance) ExitCleanup(clearCgroup bool) { + if clearCgroup { + pid := s.GetPid() + if pid > 0 { + // TODO: ClearCgroup + s.ClearCgroup(pid) + } + } + if s.monitor != nil { + s.monitor.Disconnect() + s.monitor = nil + } +} + +func (s *SKVMGuestInstance) GetCleanFiles() []string { + return []string{s.GetPidFilePath(), s.GetVncFilePath()} +} + +func (s *SKVMGuestInstance) delTmpDisks(ctx context.Context, migrated bool) { + disks, _ := s.Desc.GetArray("disks") + for _, disk := range disks { + if disk.Contains("path") { + diskPath, _ := disk.GetString("path") + // TODO GetDisksByPath, storagetypes, deleteallsnapshot, delete + d := hostinfo.GetStorageManager().GetDiskByPath(diskPath) + if d != nil && d.GetType == storagetypes.STORAGE_LOCAL && migrated { + d.DeleteAllSnapshot() + d.Delete(ctx) + } + } + } +} + +func (s *SKVMGuestInstance) Delete(ctx context.Context, migrated bool) error { + // self._del_bw_limit() + // self._del_netmon_nic() ?? 需要开发？ + s.delTmpDisks(ctx, migrated) + return exec.Command("rm", "-rf", s.HomeDir()).Run() +} diff --git a/pkg/hostman/guestman/utils.go b/pkg/hostman/guestman/utils.go deleted file mode 100644 index 234f946cba..0000000000 --- a/pkg/hostman/guestman/utils.go +++ /dev/null @@ -1 +0,0 @@ -package guestman diff --git a/pkg/hostman/hostinfo/host_cpu.go b/pkg/hostman/hostinfo/host_cpu.go index 2d4f61e61d..486d40f4f1 100644 --- a/pkg/hostman/hostinfo/host_cpu.go +++ b/pkg/hostman/hostinfo/host_cpu.go @@ -1,20 +1,98 @@ package hostinfo import ( + "bufio" + "os" + "os/exec" + "strconv" + "strings" + "github.com/shirou/gopsutil/cpu" + "yunion.io/x/log" + "yunion.io/x/onecloud/pkg/cloudcommon" + "yunion.io/x/onecloud/pkg/cloudcommon/types" + "yunion.io/x/onecloud/pkg/util/sysutils" ) type SCPUInfo struct { CpuCount int cpuFreq float32 // MHZ cpuFeatures []string + + cpuInfoProc *types.CPUInfo + cpuInfoDmi *types.DMICPUInfo } -func DetectCpuInfo() *SCPUInfo { +func DetectCpuInfo() (*SCPUInfo, err) { + cpuinfo := new(SCPUInfo) cpuCount, _ := cpu.Counts(true) - log.Errorln(cpu.Percent(0, false)) // /// - return &SCPUInfo{ - CpuCount: cpuCount, + cpuinfo.CpuCount = cpuCount + spec, err := cpuinfo.fetchCpuSpecs() + if err != nil { + return nil, err } + strCpuFreq := spec["cpu_freq"] + freq, err := strconv.ParseInt(strCpuFreq, 10, 0) + if err != nil { + log.Errorln(err) + return nil, err + } + cpu.Percent(interval, percpu) + ret, err := cloudcommon.FileGetContents("/proc/cpuinfo") + if err != nil { + log.Errorln(err) + return nil, err + } + cpuinfo.cpuInfoProc, err = sysutils.ParseCPUInfo(strings.Split(ret, "\n")) + if err != nil { + log.Errorln(err) + return nil, err + } + ret, err = exec.Command("dmidecode", "-t", "4").Output() + if err != nil { + log.Errorln(err) + return nil, err + } + cpuinfo.cpuInfoDmi, err = sysutils.ParseDMICPUInfo(strings.Split(string(ret), "\n")) + if err != nil { + log.Errorln(err) + return nil, err + } + return cpuinfo +} + +func (c *SCPUInfo) fetchCpuSpecs() (map[string]string, error) { + f, err := os.Open("/proc/cpuinfo") + if err != nil { + return nil, err + } + defer f.Close() + + var spec = make(map[string]string, 0) + scanner := bufio.NewScanner(f) + for scanner.Scan() { + line := scanner.Text() + colon := strings.Index(line, ":") + if colon > 0 { + key := strings.TrimSpace(line[:colon]) + val := strings.TrimSpace(line[colon+1:]) + if key == "cpu MHz" { + spec["cpu_freq"] = val + } else if key == "flags" { + spec["flags"] = val + } + } + } + if err := scanner.Err(); err != nil { + log.Errorln(err) + return nil, err + } + return spec, nil +} + +// percentInterval(ms) +func (c *SCPUInfo) GetJsonDesc(percentInterval int) { + // perc, err := cpu.Percent(time.Millisecond*percentInterval, false) + // os. ?????可能不需要要写 } diff --git a/pkg/hostman/hostinfo/hostinfo.go b/pkg/hostman/hostinfo/hostinfo.go index e1173a343c..381611f546 100644 --- a/pkg/hostman/hostinfo/hostinfo.go +++ b/pkg/hostman/hostinfo/hostinfo.go @@ -18,6 +18,7 @@ type SHostInfo struct { isRegistered bool Cpu *SCPUInfo + Mem *SMemory storageManager *storageman.SStorageManager } @@ -190,6 +191,10 @@ func (h *SHostInfo) EnableTransparentHugepages() { } } +func (h *SHostInfo) GetMemory() int { + return h.Mem.Total // - options.reserved_memory +} + func (h *SHostInfo) EnableNativeHugepages() error { content, err := ioutil.ReadFile("/proc/sys/vm/nr_hugepages") if err != nil { @@ -203,9 +208,7 @@ func (h *SHostInfo) EnableNativeHugepages() error { for k, v := range kv { h.setSysConfig(k, v) } - // TODO - // h.Memory 还未实现 - preAllocPagesNum := h.GetMemory/h.Memory.GetHugepagesizeMb() + 1 + preAllocPagesNum := h.GetMemory()/h.Memory.GetHugepagesizeMb() + 1 cmd := cloudcommon.CommandWithTimeout(1, "sh", "-c", fmt.Sprintf("echo %d > /proc/sys/vm/nr_hugepages", preAllocPagesNum)) _, err := cmd.Output() if err != nil { @@ -275,11 +278,17 @@ func (h *SHostInfo) StartRegister() { // TODO } -func NewHostInfo() *SHostInfo { - cpu := DetectCpuInfo() - // memory := DetectMemoryInfo() - return &SHostInfo{ - Cpu: cpu, +func NewHostInfo() (*SHostInfo, error) { + var res = new(SHostInfo) + cpu, err := DetectCpuInfo() + if err != nil { + return nil, err + } else { + res.Cpu = cpu + } + mem, err := DetectMemoryInfo() + if err != nil { + } } diff --git a/pkg/hostman/hostinfo/hostmemory.go b/pkg/hostman/hostinfo/hostmemory.go new file mode 100644 index 0000000000..7f27215e0b --- /dev/null +++ b/pkg/hostman/hostinfo/hostmemory.go @@ -0,0 +1,65 @@ +package hostinfo + +import ( + "bufio" + "os" + "os/exec" + "regexp" + "strconv" + "strings" + + "yunion.io/x/log" + "yunion.io/x/onecloud/pkg/cloudcommon/types" + "yunion.io/x/onecloud/pkg/util/sysutils" +) + +type SMemory struct { + Total int + Free int + Used int + MemInfo *types.DMIMemInfo +} + +func DetectMemoryInfo() (*SMemory, error) { + var mem = new(SMemory) + info, err := mem.VirtualMemory() + if err != nil { + return nil, err + } + mem.Total = int(info.Total / 1024 / 1024) + mem.Free = int(info.Available / 1024 / 1024) + mem.Used = mem.Total - mem.Free + ret, err := exec.Command("dmidecode", "-t", "17").Output() + if err != nil { + return nil, err + } + mem.MemInfo = sysutils.ParseDMIMemInfo(strings.Split(string(ret), "\n")) + return mem, nil +} + +func (m *SMemory) GetHugepagesizeMb() int { + file, err := os.Open("/proc/meminfo") + if err != nil { + log.Errorln(err) + return 0 + } + defer file.Close() + scanner := bufio.NewScanner(file) + for scanner.Scan() { + line := scanner.Text() + if strings.HasPrefix(line, "Hugepagesize:") { + re := regexp.MustCompile(`\s+`) + segs := re.Split(line, -1) + v, err := strconv.Atoi(segs[1]) + if err != nil { + log.Errorln(err) + return 0 + } + return int(v) / 1024 + } + } + if err := scanner.Err(); err != nil { + log.Errorln(err) + } + return 0 +} diff --git a/pkg/hostman/monitor/monitor.go b/pkg/hostman/monitor/monitor.go index 79a14399ec..4fbf068711 100644 --- a/pkg/hostman/monitor/monitor.go +++ b/pkg/hostman/monitor/monitor.go @@ -13,6 +13,7 @@ type StringCallback func(string) type Monitor interface { Connect(host string, port int) error + Dicconnect() // The callback function will be called in another goroutine SimpleCommand(cmd string, callback StringCallback) @@ -59,6 +60,13 @@ func (m *SBaseMonitor) Connect(host string, port int) error { return nil } +func (m *SBaseMonitor) Disconnect() { + if m.connected { + m.connected = false + m.rwc.Close() + } +} + func (m *SBaseMonitor) checkReading() bool { m.mutex.Lock() defer m.mutex.Unlock() diff --git a/pkg/hostman/options/options.go b/pkg/hostman/options/options.go index f79d699efd..f83ea98b3c 100644 --- a/pkg/hostman/options/options.go +++ b/pkg/hostman/options/options.go @@ -49,6 +49,8 @@ type SHostOptions struct { EnableQmpMonitor bool `help:"Enable qmp monitor" default:"true"` PrivatePrefixes []string `help:"IPv4 private prefixes"` + LocalImagePath []string `help:"Local image storage paths"` + SharedStorages []string `help:"Path of shared storages"` } var HostOptions SHostOptions diff --git a/pkg/hostman/storageman/core.go b/pkg/hostman/storageman/core.go deleted file mode 100644 index 76f5a96538..0000000000 --- a/pkg/hostman/storageman/core.go +++ /dev/null @@ -1,103 +0,0 @@ -package storageman - -import ( - "fmt" - "path" - "sync" - - "yunion.io/x/jsonutils" - "yunion.io/x/log" - "yunion.io/x/onecloud/pkg/hostman/guestfs" -) - -type IStorage interface { - StorageType() string - GetPath() string - - // Find owner disks first, if not found, call create disk - GetDiskById(diskId string) IDisk - CreateDisk(diskId string) IDisk -} - -type IDisk interface { - GetId() string - Probe() bool - DeployGuestFs(guestDesc *jsonutils.JSONDict, - deployInfo *guestfs.SDeployInfo) (jsonutils.JSONObject, error) -} - -type SBaseStorage struct { - Manager *SStorageManager - StorageId string - Path string - StorageName string - StorageConf *jsonutils.JSONDict - StoragecacheId string - - Disks []IDisk - DiskLock *sync.Mutex -} - -func (s *SBaseStorage) GetPath() string { - return s.Path -} - -func NewBaseStorage(manager *SStorageManager, path string) *SBaseStorage { - var ret = new(SBaseStorage) - ret.Disks = make([]IDisk, 0) - ret.DiskLock = new(sync.Mutex) - ret.Manager = manager - ret.Path = path - return ret -} - -type SBaseDisk struct { - Id string - Storage IStorage -} - -func (d *SBaseDisk) getPath() string { - return path.Join(d.Storage.GetPath(), d.Id) -} - -func (d *SBaseDisk) DeployGuestFs( - guestDesc *jsonutils.JSONDict, - deployInfo *guestfs.SDeployInfo) (jsonutils.JSONObject, error) { - // TODO - var kvmDisk = NewKVMGuestDisk(d.getPath()) - if kvmDisk.Connect() { - defer kvmDisk.Disconnect() - log.Infof("Kvm Disk Connect Success !!") - if root := kvmDisk.Mount(); root != nil { - defer kvmDisk.Umount(root) - return root.DeployGuestFs(root, guestDesc, deployInfo) - } - } - return nil, fmt.Errorf("Kvm disk connect or mount error") -} - -func NewBaseDisk(storage IStorage, id string) *SBaseDisk { - var ret = new(SBaseDisk) - ret.Storage = storage - ret.Id = id - return ret -} - -type SStorageManager struct { - storages map[string]IStorage -} - -func NewStorageManager() *SStorageManager { - var ret = new(SStorageManager) - // TODO - ret.storages = make(map[string]IStorage, 0) - - return ret -} - -func (m *SStorageManager) GetStorageDisk(storageId, diskId string) IDisk { - if storage, ok := m.storages[storageId]; ok { - return storage.GetDiskById(diskId) - } - return nil -} diff --git a/pkg/hostman/storageman/diskbase.go b/pkg/hostman/storageman/diskbase.go new file mode 100644 index 0000000000..5f63b7ad3d --- /dev/null +++ b/pkg/hostman/storageman/diskbase.go @@ -0,0 +1,49 @@ +package storageman + +import ( + "fmt" + "path" + + "yunion.io/x/jsonutils" + "yunion.io/x/log" + "yunion.io/x/onecloud/pkg/hostman/guestfs" +) + +type IDisk interface { + GetId() string + Probe() bool + DeployGuestFs(guestDesc *jsonutils.JSONDict, + deployInfo *guestfs.SDeployInfo) (jsonutils.JSONObject, error) +} + +type SBaseDisk struct { + Id string + Storage IStorage +} + +func NewBaseDisk(storage IStorage, id string) *SBaseDisk { + var ret = new(SBaseDisk) + ret.Storage = storage + ret.Id = id + return ret +} + +func (d *SBaseDisk) getPath() string { + return path.Join(d.Storage.GetPath(), d.Id) +} + +func (d *SBaseDisk) DeployGuestFs( + guestDesc *jsonutils.JSONDict, + deployInfo *guestfs.SDeployInfo) (jsonutils.JSONObject, error) { + var kvmDisk = NewKVMGuestDisk(d.getPath()) + if kvmDisk.Connect() { + defer kvmDisk.Disconnect() + log.Infof("Kvm Disk Connect Success !!") + + if root := kvmDisk.Mount(); root != nil { + defer kvmDisk.Umount(root) + return root.DeployGuestFs(root, guestDesc, deployInfo) + } + } + return nil, fmt.Errorf("Kvm disk connect or mount error") +} diff --git a/pkg/hostman/storageman/localdisk.go b/pkg/hostman/storageman/localdisk.go new file mode 100644 index 0000000000..1e387a312a --- /dev/null +++ b/pkg/hostman/storageman/localdisk.go @@ -0,0 +1,25 @@ +package storageman + +import "os" + +type SLocalDisk struct { + *SBaseDisk +} + +func NewLocalDisk(storage IStorage, id string) *SLocalDisk { + var ret = new(SLocalDisk) + ret.SBaseDisk = NewBaseDisk(storage, id) + return ret +} + +func (d *SLocalDisk) GetId() string { + return d.Id +} + +func (d *SLocalDisk) Probe() bool { + if _, err := os.Stat(d.getPath()); !os.IsNotExist(err) { + return true + } + // TODO alter ?? + return false +} diff --git a/pkg/hostman/storageman/localstorage.go b/pkg/hostman/storageman/localstorage.go index 0e41d91a28..2a9e52b90f 100644 --- a/pkg/hostman/storageman/localstorage.go +++ b/pkg/hostman/storageman/localstorage.go @@ -1,9 +1,5 @@ package storageman -import ( - "os" -) - type SLocalStorage struct { *SBaseStorage } @@ -42,25 +38,3 @@ func (s *SLocalStorage) CreateDisk(diskId string) IDisk { func (s *SLocalStorage) StartSnapshotRecycle() { //TODO } - -type SLocalDisk struct { - *SBaseDisk -} - -func NewLocalDisk(storage IStorage, id string) *SLocalDisk { - var ret = new(SLocalDisk) - ret.SBaseDisk = NewBaseDisk(storage, id) - return ret -} - -func (d *SLocalDisk) GetId() string { - return d.Id -} - -func (d *SLocalDisk) Probe() bool { - if _, err := os.Stat(d.getPath()); !os.IsNotExist(err) { - return true - } - // TODO alter ?? - return false -} diff --git a/pkg/hostman/storageman/storagebase.go b/pkg/hostman/storageman/storagebase.go new file mode 100644 index 0000000000..76b94ee5da --- /dev/null +++ b/pkg/hostman/storageman/storagebase.go @@ -0,0 +1,41 @@ +package storageman + +import ( + "sync" + + "yunion.io/x/jsonutils" +) + +type IStorage interface { + StorageType() string + GetPath() string + + // Find owner disks first, if not found, call create disk + GetDiskById(diskId string) IDisk + CreateDisk(diskId string) IDisk +} + +type SBaseStorage struct { + Manager *SStorageManager + StorageId string + Path string + StorageName string + StorageConf *jsonutils.JSONDict + StoragecacheId string + + Disks []IDisk + DiskLock *sync.Mutex +} + +func (s *SBaseStorage) GetPath() string { + return s.Path +} + +func NewBaseStorage(manager *SStorageManager, path string) *SBaseStorage { + var ret = new(SBaseStorage) + ret.Disks = make([]IDisk, 0) + ret.DiskLock = new(sync.Mutex) + ret.Manager = manager + ret.Path = path + return ret +} diff --git a/pkg/hostman/storageman/storagemanager.go b/pkg/hostman/storageman/storagemanager.go new file mode 100644 index 0000000000..95b2ff7576 --- /dev/null +++ b/pkg/hostman/storageman/storagemanager.go @@ -0,0 +1,67 @@ +package storageman + +import ( + "yunion.io/x/log" + "yunion.io/x/onecloud/pkg/hostman/options" +) + +/***************************************************************************/ +/****************************StorageManager*********************************/ +/***************************************************************************/ + +const MINIMAL_FREE_SPACE = 128 + +type SStorageManager struct { + storages []IStorage + LocalStorageImagecache IStorageCache +} + +func NewStorageManager() *SStorageManager { + var ret = new(SStorageManager) + ret.storages = make([]IStorage, 0) + var allFull = true + for _, d := range options.HostOptions.LocalImagePath { + s := NewLocalStorage(ret, d) + // TODO Accessible GetFreeSizeMb + if s.Accessible() { + ret.storages = append(ret.storages, s) + if allFull && s.GetFreeSizeMb > MINIMAL_FREE_SPACE { + allFull = false + } + } + } + for _, d := range options.HostOptions.SharedStorages { + s := NewSharedStorage(ret, d) + ret.storages = append(ret.storages, s) + allFull = False + } + if allFull { + log.Fatalf("Not enough storage space!") + } + ret.initLocalStorageImagecache() + ret.initAgentStorageImagecache() + ret.initAgentStorage() + return ret +} + +func (s *SStorageManager) GetStorageDisk(storageId, diskId string) IDisk { + if storage, ok := s.storages[storageId]; ok { + return storage.GetDiskById(diskId) + } + return nil +} + +func (s *SStorageManager) initLocalStorageImagecache() { + var cacheDir = "image_cache" + cachePath := options.HostOptions.ImageCachePath + limit := options.HostOptions.ImageCacheLimit + if len(cachePath) == 0 { + cachePath = s.getLeasedUsedLocalStorage(cacheDir, limit) + } + if len(cachePath) == 0 { + // TODO NewLocalImageCacheManager + s.LocalStorageImagecache = NewLocalImageCacheManager(s, cachePath, limit, true) + } else { + log.Fatalf("Cannot allocate image cache storage") + } +}