opensource/cloudpods
3
0
Fork 0
mirror of https://github.com/yunionio/cloudpods.git synced 2026-09-21 14:19:49 +08:00
Code Issues Packages Projects Releases Wiki Activity

20181221 temp commit

Browse Source
This commit is contained in:
wanyaoqi
2019-01-22 16:56:47 +08:00
committed by Zexi Li
parent 5b16b5b423
commit a053e372ef
18 changed files with 572 additions and 220 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/cloudcommon/httpclients/compute_client.go
+23
View File
@@ -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")
}
}
pkg/cloudcommon/workmanager/manager.go
+27 -4
View File
@@ -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
// }
pkg/hostman/guestman/guesthandler.go
+39 -34
View File
@@ -22,17 +22,6 @@ func AddGuestTaskHandler(prefix string, app *appsrv.Application) {
app.AddHandler("DELETE", "/servers/<sid>", 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["<sid>"])
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["<sid>"]
@@ -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["<sid>"])
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["<sid>"]
@@ -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,
pkg/hostman/guestman/guesthelper.go
+9
View File
@@ -0,0 +1,9 @@
package guestman
import "yunion.io/x/jsonutils"
type SGuestDeploy struct {
sid string
body jsonutils.JSONObject
isInit bool
}
pkg/hostman/guestman/guestman.go
+33 -17
View File
@@ -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
}
pkg/hostman/guestman/qemu-kvm.go
+85 -23
View File
@@ -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()
}
pkg/hostman/guestman/utils.go
-1
View File
@@ -1 +0,0 @@
package guestman
pkg/hostman/hostinfo/host_cpu.go
+82 -4
View File
@@ -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. ?????可能不需要要写
}
pkg/hostman/hostinfo/hostinfo.go
+17 -8
View File
@@ -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 {
}
}
pkg/hostman/hostinfo/hostmemory.go
+65
View File
@@ -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
}
pkg/hostman/monitor/monitor.go
+8
View File
@@ -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()
pkg/hostman/options/options.go
+2
View File
@@ -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
pkg/hostman/storageman/core.go
-103
View File
@@ -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
}
pkg/hostman/storageman/diskbase.go
+49
View File
@@ -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")
}
pkg/hostman/storageman/localdisk.go
+25
View File
@@ -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
}
pkg/hostman/storageman/localstorage.go
-26
View File
@@ -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
}
pkg/hostman/storageman/storagebase.go
+41
View File
@@ -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
}
pkg/hostman/storageman/storagemanager.go
+67
View File
@@ -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")
}
}