From 71856e430e551217a4e9fd009137b20b134c79ed Mon Sep 17 00:00:00 2001 From: wanyaoqi Date: Mon, 14 Jan 2019 16:44:01 +0800 Subject: [PATCH] guest migrate snapshot --- pkg/cloudcommon/workmanager/manager.go | 1 + pkg/hostman/guestman/guesthandler.go | 249 +++++++- pkg/hostman/guestman/guesthelper.go | 55 +- pkg/hostman/guestman/guestman.go | 178 +++++- pkg/hostman/guestman/guesttasks.go | 603 +++++++++++++++++- pkg/hostman/guestman/qemu-kvm.go | 327 +++++++++- pkg/hostman/guestman/qemu-kvmhelper.go | 10 +- .../hostinfo/{ => hostbridge}/hostbridge.go | 13 +- pkg/hostman/hostinfo/hostinfo.go | 8 +- pkg/hostman/hostinfo/hostinfohelper.go | 7 +- pkg/hostman/monitor/hmp.go | 52 ++ pkg/hostman/monitor/monitor.go | 7 + pkg/hostman/monitor/qmp.go | 105 +++ pkg/hostman/options/options.go | 6 + pkg/hostman/storageman/core.go | 62 +- pkg/hostman/storageman/diskbase.go | 9 +- pkg/hostman/storageman/diskhandler.go | 1 - pkg/hostman/storageman/disklocal.go | 40 +- pkg/hostman/storageman/diskutils.go | 10 +- pkg/hostman/storageman/imagecachebase.go | 22 +- pkg/hostman/storageman/imagecachemanager.go | 20 +- pkg/hostman/storageman/{ => nbd}/nbdman.go | 2 +- pkg/hostman/storageman/storagebase.go | 3 + pkg/hostman/storageman/storagelocal.go | 17 + pkg/util/cgrouputils/cgrouputils.go | 367 ++++++++++- pkg/util/cgrouputils/cpusetutils.go | 74 +++ pkg/util/cgrouputils/cpuutils.go | 0 pkg/util/fileutils2/fileutils.go | 51 ++ pkg/util/fileutils2/fileutils_test.go | 24 +- pkg/util/ovsutils/ovsutils.go | 90 +++ pkg/util/qemuimg/qemuimg.go | 23 + pkg/util/regutils2/regutils.go | 12 +- 32 files changed, 2308 insertions(+), 140 deletions(-) rename pkg/hostman/hostinfo/{ => hostbridge}/hostbridge.go (96%) rename pkg/hostman/storageman/{ => nbd}/nbdman.go (98%) create mode 100644 pkg/util/cgrouputils/cpusetutils.go create mode 100644 pkg/util/cgrouputils/cpuutils.go create mode 100644 pkg/util/ovsutils/ovsutils.go diff --git a/pkg/cloudcommon/workmanager/manager.go b/pkg/cloudcommon/workmanager/manager.go index 3ae727c912..3fdcf667d9 100644 --- a/pkg/cloudcommon/workmanager/manager.go +++ b/pkg/cloudcommon/workmanager/manager.go @@ -79,6 +79,7 @@ func (w *SWorkManager) DelayTaskWithoutReqctx(ctx context.Context, task DelayTas }() if _, err := task(ctx, params); err != nil { log.Errorln("DelayTaskWithoutReqctx error: ", err) + w.onFailed(ctx, err.Error()) } }() } diff --git a/pkg/hostman/guestman/guesthandler.go b/pkg/hostman/guestman/guesthandler.go index 0a796b6d0a..71454ef374 100644 --- a/pkg/hostman/guestman/guesthandler.go +++ b/pkg/hostman/guestman/guesthandler.go @@ -9,6 +9,7 @@ import ( "yunion.io/x/jsonutils" "yunion.io/x/onecloud/pkg/appsrv" "yunion.io/x/onecloud/pkg/hostman/hostutils" + "yunion.io/x/onecloud/pkg/hostman/storageman" "yunion.io/x/onecloud/pkg/httperrors" "yunion.io/x/onecloud/pkg/mcclient/auth" ) @@ -152,31 +153,241 @@ func guestSuspend(ctx context.Context, sid string, body jsonutils.JSONObject) (i return nil, nil } -// snapshot 相关的有待重构，放入diskhandler中去，先不写 +func guestSrcPrepareMigrate(ctx context.Context, sid string, body jsonutils.JSONObject) (interface{}, error) { + if !guestManger.IsGuestExist(sid) { + return nil, httperrors.NewNotFoundError("Guest %s not found", sid) + } + liveMigrate := jsonutils.QueryBoolean(body, "live_migrate", false) + hostutils.DelayTask(ctx, guestManger.SrcPrepareMigrate, + &SSrcPrepareMigrate{sid, liveMigrate}) + return nil, nil +} + +func guestDestPrepareMigrate(ctx context.Context, sid string, body jsonutils.JSONObject) (interface{}, error) { + if !guestManger.CanMigrate(sid) { + return nil, httperrors.NewBadRequestError("Guest exist") + } + desc, err := body.Get("desc") + if err != nil { + return nil, httperrors.NewMissingParameterError("desc") + } + qemuVersion, err := body.GetString("qemu_version") + if err != nil { + return nil, httperrors.NewMissingParameterError("qemu_version") + } + liveMigrate := jsonutils.QueryBoolean(body, "live_migrate", false) + isLocal, err := body.Bool("is_local_storage") + if err != nil { + return nil, httperrors.NewMissingParameterError("is_local_storage") + } + var params = &SDestPrepareMigrate{} + params.Sid = sid + params.Desc = desc + params.QemuVersion = qemuVersion + params.LiveMigrate = liveMigrate + if isLocal { + serverUrl, err := body.GetString("server_url") + if err != nil { + return nil, httperrors.NewMissingParameterError("server_url") + } else { + params.ServerUrl = serverUrl + } + snapshotsUri, err := body.GetString("snapshots_uri") + if err != nil { + return nil, httperrors.NewMissingParameterError("snapshots_uri") + } else { + params.SnapshotsUri = snapshotsUri + } + disksUri, err := body.GetString("disks_uri") + if err != nil { + return nil, httperrors.NewMissingParameterError("disks_uri") + } else { + params.DisksUri = disksUri + } + srcSnapshots, err := body.Get("src_snapshots") + if err != nil { + return nil, httperrors.NewMissingParameterError("src_snapshots") + } else { + params.SrcSnapshots = srcSnapshots + } + disksBack, err := body.Get("disks_back") + if err != nil { + return nil, httperrors.NewMissingParameterError("disks_back") + } else { + params.DisksBackingFile = disksBack + } + disks, err := desc.GetArray("disks") + if err != nil { + return nil, httperrors.NewInputParameterError("Get desc disks error") + } else { + targetStorageId, _ := disks[0].GetString("target_storage_id") + if len(targetStorageId) == 0 { + return nil, httperrors.NewInputParameterError("Disk desc missing target storage id") + } + params.TargetStorageId = targetStorageId + } + } + hostutils.DelayTask(ctx, guestManger.DestPrepareMigrate, params) + return nil, nil +} + +func guestLiveMigrate(ctx context.Context, sid string, body jsonutils.JSONObject) (interface{}, error) { + if !guestManger.IsGuestExist(sid) { + return nil, httperrors.NewNotFoundError("Guest %s not found", sid) + } + destPort, err := body.Int("live_migrate_dest_port") + if err != nil { + return nil, httperrors.NewMissingParameterError("live_migrate_dest_port") + } + destIp, err := body.GetString("dest_ip") + if err != nil { + return nil, httperrors.NewMissingParameterError("dest_ip") + } + isLocal, err := body.Bool("is_local_storage") + if err != nil { + return nil, httperrors.NewMissingParameterError("is_local_storage") + } + hostutils.DelayTaskWithoutReqctx(ctx, guestManger.LiveMigrate, &SLiveMigrate{ + Sid: sid, DestPort: int(destPort), DestIp: destIp, IsLocal: isLocal, + }) + return nil, nil +} + +func guestResume(ctx context.Context, sid string, body jsonutils.JSONObject) (interface{}, error) { + if !guestManger.IsGuestExist(sid) { + return nil, httperrors.NewNotFoundError("Guest %s not found", sid) + } + isLiveMigrate := jsonutils.QueryBoolean(body, "live_migrate", false) + guestManger.Resume(ctx, sid, isLiveMigrate) + return nil, nil +} + +func guestStartNbdServer(ctx context.Context, sid string, body jsonutils.JSONObject) (interface{}, error) { + if !guestManger.IsGuestExist(sid) { + return nil, httperrors.NewNotFoundError("Guest %s not found", sid) + } + hostutils.DelayTask(ctx, guestManger.StartNbdServer, sid) + return nil, nil +} + +func guestDriveMirror(ctx context.Context, sid string, body jsonutils.JSONObject) (interface{}, error) { + if !guestManger.IsGuestExist(sid) { + return nil, httperrors.NewNotFoundError("Guest %s not found", sid) + } + backupNbdServerUri, err := body.GetString("backup_ndb_server_uri") + if err != nil { + return nil, httperrors.NewMissingParameterError("backup_ndb_server_uri") + } + hostutils.DelayTask(ctx, guestManger.StartDriveMirror, + &SDriverMirror{sid, backupNbdServerUri}) + return nil, nil +} + +func guestReloadDiskSnapshot(ctx context.Context, sid string, body jsonutils.JSONObject) (interface{}, error) { + if !guestManger.IsGuestExist(sid) { + return nil, httperrors.NewNotFoundError("Guest %s not found", sid) + } + diskId, err := body.GetString("disk_id") + if err != nil { + return nil, httperrors.NewMissingParameterError("disk_id") + } + + var disk storageman.IDisk + guest := guestManger.Servers[sid] + disks, _ := guest.Desc.GetArray("disks") + for _, disk := range disks { + id, _ := disk.GetString("disk_id") + if diskId == id { + diskPath, _ := disk.GetString("path") + disk := storageman.GetManager().GetDiskByPath(diskPath) + break + } + } + if disk == nil { + return nil, httperrors.NewNotFoundError("Disk not found") + } + + hostutils.DelayTaskWithoutReqctx(ctx, guestManger.ReloadDiskSnapshot, &SReloadDisk{sid, disk}) + return nil, nil +} + func guestSnapshot(ctx context.Context, sid string, body jsonutils.JSONObject) (interface{}, error) { if !guestManger.IsGuestExist(sid) { return nil, httperrors.NewNotFoundError("Guest %s not found", sid) } - hostutils.DelayTask(ctx, guestManger.DoSnapshot, params) + snapshotId, err := body.GetString("snapshot_id") + if err != nil { + return nil, httperrors.NewMissingParameterError("snapshot_id") + } + diskId, err := body.GetString("disk_id") + if err != nil { + return nil, httperrors.NewMissingParameterError("disk_id") + } + + var disk storageman.IDisk + guest := guestManger.Servers[sid] + disks, _ := guest.Desc.GetArray("disks") + for _, disk := range disks { + id, _ := disk.GetString("disk_id") + if diskId == id { + diskPath, _ := disk.GetString("path") + disk := storageman.GetManager().GetDiskByPath(diskPath) + break + } + } + if disk == nil { + return nil, httperrors.NewNotFoundError("Disk not found") + } + + hostutils.DelayTask(ctx, guestManger.DoSnapshot, &SDiskSnapshot{sid, snapshotId, disk}) + return nil, nil } -func guestSrcPrepareMigrate(ctx context.Context, sid string, body jsonutils.JSONObject) (interface{}, error) { +func guestDeleteSnapshot(ctx context.Context, sid string, body jsonutils.JSONObject) (interface{}, error) { + deleteSnapshot, err := body.GetString("delete_snapshot") + if err != nil { + return nil, httperrors.NewMissingParameterError("delete_snapshot") + } + diskId, err := body.GetString("disk_id") + if err != nil { + return nil, httperrors.NewMissingParameterError("disk_id") + } -} + var disk storageman.IDisk + guest := guestManger.Servers[sid] + disks, _ := guest.Desc.GetArray("disks") + for _, disk := range disks { + id, _ := disk.GetString("disk_id") + if diskId == id { + diskPath, _ := disk.GetString("path") + disk := storageman.GetManager().GetDiskByPath(diskPath) + break + } + } + if disk == nil { + return nil, httperrors.NewNotFoundError("Disk not found") + } -func guestDestPrepareMigrate(ctx context.Context, sid string, body jsonutils.JSONObject) (interface{}, error) { -} + params := &SDeleteDiskSnapshot{ + Sid: sid, + DeleteSnapshot: deleteSnapshot, + Disk: disk, + } -func guestLiveMigrate(ctx context.Context, sid string, body jsonutils.JSONObject) (interface{}, error) { -} - -func guestResume(ctx context.Context, sid string, body jsonutils.JSONObject) (interface{}, error) { -} - -func guestStartNbdServer(ctx context.Context, sid string, body jsonutils.JSONObject) (interface{}, error) { -} - -func guestDriveMirror(ctx context.Context, sid string, body jsonutils.JSONObject) (interface{}, error) { + if !jsonutils.QueryBoolean(body, "auto_delete", false) { + convertSnapshot, err := body.GetString("convert_snapshot") + if err != nil { + return nil, httperrors.NewMissingParameterError("convert_snapshot") + } + params.ConvertSnapshot = convertSnapshot + pendingDelete, err := body.Bool("pending_delete") + if err != nil { + return nil, httperrors.NewMissingParameterError("pending_delete") + } + params.PendingDelete = pendingDelete + } + hostutils.DelayTask(ctx, guestManger.DeleteSnapshot, params) + return nil, nil } var actionFuncs = map[string]actionFunc{ @@ -188,9 +399,9 @@ var actionFuncs = map[string]actionFunc{ "sync": guestSync, "suspend": guestSuspend, - "snapshot": guestSnapshot, - // "delete-snapshot": guestDeleteSnapshot, - // "reload-disk-snapshot": guestReloadDiskSnapshot, + "snapshot": guestSnapshot, + "delete-snapshot": guestDeleteSnapshot, + "reload-disk-snapshot": guestReloadDiskSnapshot, // "remove-statefile": guestRemoveStatefile, // "io-throttle": guestIoThrottle, diff --git a/pkg/hostman/guestman/guesthelper.go b/pkg/hostman/guestman/guesthelper.go index 4b777df9d2..3d76678f7b 100644 --- a/pkg/hostman/guestman/guesthelper.go +++ b/pkg/hostman/guestman/guesthelper.go @@ -1,6 +1,9 @@ package guestman -import "yunion.io/x/jsonutils" +import ( + "yunion.io/x/jsonutils" + "yunion.io/x/onecloud/pkg/hostman/storageman" +) type SBaseParms struct { Sid string @@ -11,3 +14,53 @@ type SGuestDeploy struct { SGuestBaseParms IsInit bool } + +type SSrcPrepareMigrate struct { + Sid string + LiveMigrate bool +} + +type SDestPrepareMigrate struct { + Sid string + ServerUrl string + QemuVersion string + SnapshotsUri string + DisksUri string + TargetStorageId string + LiveMigrate bool + + Desc jsonutils.JSONObject + DisksBackingFile jsonutils.JSONObject + SrcSnapshots jsonutils.JSONObject +} + +type SLiveMigrate struct { + Sid string + DestPort int + DestIp string + IsLocal bool +} + +type SDriverMirror struct { + Sid string + NbdServerUri string +} + +type SReloadDisk struct { + Sid string + Disk storageman.IDisk +} + +type SDiskSnapshot struct { + Sid string + SnapshotId string + Disk storageman.IDisk +} + +type SDeleteDiskSnapshot struct { + Sid string + DeleteSnapshot string + Disk storageman.IDisk + ConvertSnapshot string + PendingDelete bool +} diff --git a/pkg/hostman/guestman/guestman.go b/pkg/hostman/guestman/guestman.go index 31a5890666..62e5decfba 100644 --- a/pkg/hostman/guestman/guestman.go +++ b/pkg/hostman/guestman/guestman.go @@ -6,6 +6,7 @@ import ( "io/ioutil" "net/url" "os" + "os/exec" "path" "strings" "sync" @@ -19,8 +20,11 @@ import ( "yunion.io/x/onecloud/pkg/cloudcommon/sshkeys" "yunion.io/x/onecloud/pkg/hostman/guestfs" "yunion.io/x/onecloud/pkg/hostman/hostutils" + "yunion.io/x/onecloud/pkg/hostman/storageman" "yunion.io/x/onecloud/pkg/httperrors" "yunion.io/x/onecloud/pkg/mcclient/modules" + "yunion.io/x/onecloud/pkg/util/cgrouputils" + "yunion.io/x/onecloud/pkg/util/fileutils2" "yunion.io/x/onecloud/pkg/util/netutils2" "yunion.io/x/onecloud/pkg/util/timeutils2" ) @@ -348,7 +352,6 @@ func (m *SGuestManager) GuestSync(ctx context.Context, params interface{}) (json if syncParams.Body.Contains("desc") { desc, _ := syncParams.Body.Get("desc") fwOnly := jsonutils.QueryBoolean(syncParams.Body, "fw_only", false) - // TODO :SyncConfig return guest.SyncConfig(ctx, desc, fwOnly) } return nil, nil @@ -364,11 +367,131 @@ func (m *SGuestManager) GuestSuspend(ctx context.Context, params interface{}) (j return nil, nil } -func (m *SGuestManager) DoSnapshot(ctx context.Context, params interface{}) (jsonutils.JSONObject, error) { - // TODO +func (m *SGuestManager) SrcPrepareMigrate(ctx context.Context, params interface{}) (jsonutils.JSONObject, error) { + migParams, ok := params.(*SSrcPrepareMigrate) + if !ok { + return nil, hostutils.ParamsError + } + guest := m.Servers[migParams.Sid] + disksPrepare := guest.PrepareMigrate(migParams.LiveMigrate) + if len(disksPrepare) > 0 { + return map[string]interface{}{"disks_back": disksPrepare}, nil + } return nil, nil } +func (m *SGuestManager) DestPrepareMigrate(ctx context.Context, params interface{}) (jsonutils.JSONObject, error) { + migParams, ok := params.(*SDestPrepareMigrate) + if !ok { + return nil, hostutils.ParamsError + } + + guest := m.Servers[migParams.Sid] + if err := guest.CreateFromDesc(desc); err != nil { + return nil, err + } + + if len(migParams.TargetStorageId) > 0 { + iStorage := storageman.GetManager().GetStorage(migParams.TargetStorageId) + if iStorage == nil { + return nil, fmt.Errorf("Target storage %s not found", migParams.TargetStorageId) + } + + // 可能可以不用? + // guest.CreateFromUrl(ctx, migParams.ServerUrl, migParams.Desc) + + disks, _ := migParams.Desc.GetArray("disks") + for _, diskinfo := range disks { + var ( + diskId, _ = diskinfo.GetString("disk_id") + snapshots, _ = migParams.SrcSnapshots.GetArray(diskId) + disk = iStorage.CreateDisk(diskId) + ) + + if disk == nil { + return nil, fmt.Errorf( + "Storage %s create disk %s failed", iStorage.GetId(), diskId) + } + + // prepare disk snapshot dir + if len(snapshots) > 0 && !fileutils2.Exists(disk.GetSnapshotDir()) { + err := exec.Command("mkdir", "-p", disk.GetSnapshotDir()).Run() + if err != nil { + return nil, err + } + } + + // create snapshots form remote url + diskStorageId, _ := diskinfo.GetString("storage_id") + for _, snapshotId := range snapshots { + snapshotUrl := path.Join(migParams.SnapshotsUri, diskStorageId, diskId, snapshotId) + snapshotPath := path.Join(disk.GetSnapshotDir(), snapshotId) + log.Infof("Disk %s snapshot %s url: %s", diskId, snapshotId, snapshotUrl) + iStorage.CreateSnapshotFormUrl(ctx, snapshotUrl, diskId, snapshotPath) + } + + if migParams.LiveMigrate { + // create local disk + backingFile, _ := migParams.DisksBackingFile.GetString(diskId) + size, _ := diskinfo.Int("size") + _, err := disk.CreateRaw(ctx, size, "qcow2", "", false, "", backingFile) + if err != nil { + log.Errorln(err) + return nil, err + } + } else { + // download disk form remote url + diskUrl := path.Join(migParams.DisksUri, diskStorageId, diskId) + if err := disk.CreateFromUrl(); err != nil { + log.Errorln(err) + return nil, err + } + } + } + + // 可能可以不要 + if err := guest.SaveDesc(migParams.Desc); err != nil { + log.Errorln(err) + return nil, err + } + } + + if migParams.LiveMigrate { + startParams := jsonutils.NewDict() + startParams.Set("qemu_version", jsonutils.NewString(migParams.QemuVersion)) + startParams.Set("need_migrate", jsonutils.JSONTrue) + guest.StartGuest(ctx, params) + } + + return nil, nil +} + +func (m *SGuestManager) LiveMigrate(ctx context.Context, params interface{}) (jsonutils.JSONObject, error) { + migParams, ok := params.(*SLiveMigrate) + if !ok { + return nil, hostutils.ParamsError + } + + guest := m.Servers[migParams.Sid] + task := NewGuestLiveMigrateTask(ctx, guest, migParams) + task.Start() + return nil, nil +} + +func (m *SGuestManager) CanMigrate(sid string) bool { + m.ServersLock.Lock() + defer m.ServersLock.Unlock() + + if _, ok := m.Servers[sid]; ok { + log.Infof("Guest %s exists", sid) + return false + } + + guest := NewKVMGuestInstance(sid, m) + m.Servers[sid] = guest + return true +} + func (m *SGuestManager) GetFreePortByBase(basePort int64) int64 { var port = 1 for { @@ -401,10 +524,57 @@ func (m *SGuestManager) GetFreeVncPort() int64 { return port } +func (m *SGuestManager) ReloadDiskSnapshot( + ctx context.Context, params interface{}, +) (jsonutils.JSONObject, error) { + reloadParams, ok := params.(*SReloadDisk) + if !ok { + return nil, hostutils.ParamsError + } + guest := guestManger.Servers[reloadParams.Sid] + return guest.ExecReloadDiskTask(ctx, reloadParams.Disk) +} + +func (m *SGuestManager) DoSnapshot(ctx context.Context, params interface{}) (jsonutils.JSONObject, error) { + snapshotParams, ok := params.(*SDiskSnapshot) + if !ok { + return nil, hostutils.ParamsError + } + guest := guestManger.Servers[snapshotParams.Sid] + return guest.ExecDiskSnapshotTask(ctx, snapshotParams.Disk, snapshotParams.SnapshotId) +} + +func (m *SGuestManager) DeleteSnapshot(ctx context.Context, params interface{}) (jsonutils.JSONObject, error) { + delParams, ok := params.(*SDeleteDiskSnapshot) + if !ok { + return nil, hostutils.ParamsError + } + + if len(delParams.ConvertSnapshot) > 0 { + guest := guestManger.Servers[delParams.Sid] + return guest.ExecDeleteSnapshotTask(ctx, delParams.Disk, delParams.DeleteSnapshot, + delParams.ConvertSnapshot, delParams.PendingDelete) + } else { + return nil, delParams.Disk.DeleteSnapshot(delParams.DeleteSnapshot) + } +} + +func (m *SGuestManager) ExitGuestCleanup() { + for _, guest := range m.Servers { + guest.ExitCleanup(false) + } + + // TODO + m.StopCpusetBalancer() + cgrouputils.CgroupCleanAll() + // TODO + // hostmetrics? +} + var guestManger *SGuestManager func Stop() { - // guestManger.ExitGuestCleanup() + guestManger.ExitGuestCleanup() } func Init(serversPath string) { diff --git a/pkg/hostman/guestman/guesttasks.go b/pkg/hostman/guestman/guesttasks.go index e3b3b5a8d8..32a69e54d9 100644 --- a/pkg/hostman/guestman/guesttasks.go +++ b/pkg/hostman/guestman/guesttasks.go @@ -3,12 +3,22 @@ package guestman import ( "context" "fmt" + "os" + "os/exec" + "path" "regexp" + "strings" "time" "yunion.io/x/jsonutils" + "yunion.io/x/log" + + "yunion.io/x/onecloud/pkg/appctx" "yunion.io/x/onecloud/pkg/hostman/hostutils" + "yunion.io/x/onecloud/pkg/hostman/options" "yunion.io/x/onecloud/pkg/hostman/storageman" + "yunion.io/x/onecloud/pkg/util/fileutils2" + "yunion.io/x/onecloud/pkg/util/qemuimg" "yunion.io/x/onecloud/pkg/util/timeutils2" ) @@ -16,6 +26,10 @@ type IGuestTasks interface { Start(func(...error)) } +/** + * GuestStopTask +**/ + type SGuestStopTask struct { *SKVMGuestInstance ctx context.Context @@ -61,6 +75,10 @@ func (s *SGuestStopTask) CheckGuestRunningLater() { s.checkGuestRunning() } +/** + * GuestSyncConfigTaskExecutor +**/ + type SGuestSyncConfigTaskExecutor struct { ctx context.Context guest *SKVMGuestInstance @@ -102,6 +120,10 @@ func (t *SGuestSyncConfigTaskExecutor) runNextTaskCallback(err ...error) { t.runNextTask() } +/** + * GuestDiskSyncTask +**/ + type SGuestDiskSyncTask struct { guest *SKVMGuestInstance delDisks []jsonutils.JSONObject @@ -247,6 +269,10 @@ func (d *SGuestDiskSyncTask) onAddDeviceSucc(results string) { d.syncDisksConf() } +/** + * GuestNetworkSyncTask +**/ + type SGuestNetworkSyncTask struct { guest *SKVMGuestInstance delNics []jsonutils.JSONObject @@ -278,13 +304,586 @@ func (n *SGuestNetworkSyncTask) syncNetworkConf() { } func (n *SGuestNetworkSyncTask) removeNic(nic jsonutils.JSONObject) { - // pass + // pass not implement } func (n *SGuestNetworkSyncTask) addNic(nic jsonutils.JSONObject) { - // pass + // pass not implement } func NewGuestNetworkSyncTask(guest *SKVMGuestInstance, delNics, addNics jsonutils.JSONObject) *SGuestNetworkSyncTask { return &SGuestNetworkSyncTask{guest, delNics, addNics, make(error, 0)} } + +/** + * GuestLiveMigrateTask +**/ + +type SGuestLiveMigrateTask struct { + *SKVMGuestInstance + + ctx context.Context + params *SLiveMigrate + + c chan struct{} +} + +func NewGuestLiveMigrateTask( + ctx context.Context, guest *SKVMGuestInstance, params *SLiveMigrate, +) *SGuestLiveMigrateTask { + return &SGuestLiveMigrateTask{SKVMGuestInstance: guest, ctx: ctx, params: params} +} + +func (s *SGuestLiveMigrateTask) Start() { + s.Monitor.MigrateSetCapability("zero-blocks", "on", s.startMigrate) +} + +func (s *SGuestLiveMigrateTask) startMigrate(string) { + var copyIncremental = false + if s.params.IsLocal { + copyIncremental = true + } + s.Monitor.Migrate(fmt.Sprintf("tcp:%s:%d", s.params.DestIp, s.params.DestPort), + copyIncremental, false, s.startMigrateStatusCheck) +} + +func (s *SGuestLiveMigrateTask) startMigrateStatusCheck(string) { + s.c = make(chan struct{}) + for { + select { + case <-c: // on c close + break + case <-time.After(time.Second * 1): + s.Monitor.getMigrateStatus(s.onGetMigrateStatus) + } + } +} + +func (s *SGuestLiveMigrateTask) onGetMigrateStatus(status string) { + if status == "completed" { + close(s.c) + hostutils.TaskComplete(ctx, nil) + } else if status == "failed" { + close(c) + hostutils.TaskFailed(ctx, fmt.Sprintf("Query migrate got status: %s", status)) + } +} + +/** + * GuestResumeTask +**/ + +type SGuestResumeTask struct { + *SKVMGuestInstance + + ctx context.Context + startTime time.Time +} + +func (s *SGuestResumeTask) Start() { + s.startTime = time.Now() + s.confirmRunning() +} + +func (s *SGuestResumeTask) confirmRunning() { + s.Monitor.QueryStatus(s.onConfirmRunning) +} + +func (s *SGuestResumeTask) onConfirmRunning(status string) { + if status == "running" || status == "paused (suspended)" || status == "paused (perlaunch)" { + s.onStartRunning() + } else if strings.Contains(status, "paused") { + s.Monitor.GetBlocks(s.onGetBlockInfo) + } else if strings.Contains(status, "error") { + s.taskFailed(status) + } else { + if time.Now().Sub(s.startTime) >= time.Second*60 { + s.taskFailed("Timeout") + } else { + time.Sleep(time.Second * 1) + s.confirmRunning() + } + } +} + +func (s *SGuestResumeTask) taskFailed(reason string) { + log.Infof("Start guest %s failed: %s", s.GetId(), reason) + s.ForceStop() + if len(appctx.AppContextTaskId(ctx)) > 0 { + hostutils.TaskFailed(ctx, reason) + } else { + s.SyncStatus() + } +} + +func (s *SGuestResumeTask) onGetBlockInfo(results *jsonutils.JSONArray) { + // for _, drv := range results.GetArray() { + // // encryption not work + // } + time.Sleep(time.Second * 1) + s.resumeGuest() +} + +func (s *SGuestResumeTask) resumeGuest() { + s.startTime = time.Now() + s.Monitor.Cont(s.onResumeSucc) +} + +func (s *SGuestResumeTask) onResumeSucc(res string) { + s.confirmRunning() +} + +func (s *SGuestResumeTask) onStartRunning() { + s.removeStatefile() + if len(appctx.AppContextTaskId(ctx)) > 0 { + hostutils.TaskComplete(ctx, nil) + } + if options.HostOptions.SetVncPassword { + s.SetVncPassword() + } + s.syncMetadata() + s.SyncStatus() + timeutils2.AddTimeout(time.Second*5, s.SetCgroup) + disksIdx := s.GetNeedMergeBackingFileDiskIndexs() + if len(disksIdx) > 0 { + timeutils2.AddTimeout(time.Second*5, func() { s.startStreamDisks(disksIdx) }) + } else if options.HostOptions.AutoMergeBackingTemplate { + timeutils2.AddTimeout(time.Second*options.HostOptions.AutoMergeDelaySeconds, + func() { s.startStreamDisks(nil) }) + } +} + +func (s *SGuestResumeTask) startStreamDisks(disksIdx []int) { + s.startTime = nil + s.CleanStartupTask() + if s.IsMonitorAlive() { + s.StreamDisks(s.ctx, func() { s.onStreamComplete(disksIdx) }, disksIdx) + } +} + +func (s *SGuestResumeTask) onStreamComplete(disksIdx []int) { + if len(disksIdx) == 0 { + s.SyncStatus() + } else { + s.streamDisksComplete(s.ctx) + } +} + +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() +} + +/** + * GuestStreamDisksTask +**/ + +type SGuestStreamDisksTask struct { + *SKVMGuestInstance + + ctx context.Context + callback func() + disksIdx []int + + c chan struct{} + streamDevs []string +} + +func NewGuestStreamDisksTask(ctx context.Context, guest *SKVMGuestInstance, callback func(), disksIdx []int) { + return &SGuestStreamDisksTask{ + SKVMGuestInstance: guest, + ctx: ctx, + callback: callback, + disksIdx: disksIdx, + } +} + +func (s *SGuestStreamDisksTask) Start() { + s.Monitor.GetBlockJobs(s.onInitCheckStreamJobs) +} + +func (s *SGuestStreamDisksTask) onInitCheckStreamJobs(jobs int) { + if jobs > 0 { + log.Warningf("GuestStreamDisksTask: duplicate block streaming???") + s.startWaitBlockStream() + } else if jobs == 0 { + s.startBlockStreaming() + } +} + +func (s *SGuestStreamDisksTask) startBlockStreaming() { + s.checkBlockDrives() +} + +func (s *SGuestStreamDisksTask) checkBlockDrives() { + s.Monitor.GetBlocks(s.onBlockDrivesSucc) +} + +func (s *SGuestStreamDisksTask) onBlockDrivesSucc(res *jsonutils.JSONArray) { + streamDevs := []string{} + drvs, _ := res.GetArray() + for _, drv := range drvs { + device, err := drv.GetString("device") + if err != nil { + log.Errorln(err) + continue + } + inserted, err := drv.Get("inserted") + if err != nil && inserted.Contains("file") && inserted.Contains("backing_file") { + var stream = false + idx := device[len(device)-1] - '0' + for i := 0; i < len(s.disksIdx); i++ { + if int(idx) == s.disksIdx[i] { + stream = true + } + } + if !stream { + continue + } + streamDevs = append(streamDevs, device) + } + } + s.streamDevs = streamDevs + if len(streamDevs) == 0 { + s.taskComplete() + } else { + s.SyncStatus() + s.startDoBlockStream() + } +} + +func (s *SGuestStreamDisksTask) startDoBlockStream() { + if len(s.streamDevs) > 0 { + dev = s.streamDevs[0] + s.streamDevs = s.streamDevs[1:] + s.Monitor.BlockStream(dev, s.startWaitBlockStream) + } +} + +func (s *SGuestStreamDisksTask) startWaitBlockStream() { + if s.c == nil { + s.c = make(chan struct{}) + for { + select { + case <-s.c: + s.c = nil + return + case <-time.After(time.Second * 1): + s.Monitor.GetBlockJobs(s.checkStreamJobs) + } + } + } +} + +func (s *SGuestStreamDisksTask) checkStreamJobs(jobs int) { + if jobs == 0 { + if len(s.streamDevs) == 0 { + close(s.c) + s.taskComplete() + } else { + close(s.c) + } + } +} + +func (s *SGuestStreamDisksTask) taskComplete() { + s.SyncStatus() + + // XXX: region disk post-migrate not implement + + // disks, _ := s.Desc.GetArray("disks") + // var needSync = fale + // for i, disk := range disks { + // if disk.Contains("url") && disk.Contains("path") { + // diskId, _ := disk.GetString("disk_id") + // targetStroageId, _ := disk.GetString("target_storage_id") + // params := jsonutils.NewDict() + // params.Set("storage_id", jsonutils.NewString(targetStroageId)) + // modules.Disks.PerformAction(hostutils.GetComputeSession(context.Background()), + // diskId, "post-migrate", params) + // needSync = true + // } + // } + // if needSync { + // s.SaveDesc(s.Desc) + // } + + if s.callback != nil { + s.callback() + } +} + +/** + * GuestReloadDiskTask +**/ + +type SGuestReloadDiskTask struct { + *SKVMGuestInstance + + ctx context.Context + disk storageman.IDisk +} + +func NewGuestReloadDiskTask( + ctx context.Context, s *SKVMGuestInstance, disk *storageman.IDisk, +) *SGuestReloadDiskTask { + return &SGuestReloadDiskTask{ + SKVMGuestInstance: s, + ctx: ctx, + disk: disk, + } +} + +func (s *SGuestReloadDiskTask) WaitSnapshotReplaced(callback func()) { + // TODO + callback() +} + +func (s *SGuestReloadDiskTask) Start() { + s.fetchDisksInfo(s.startReloadDisk) +} + +func (s *SGuestReloadDiskTask) fetchDisksInfo(callback func(string)) { + s.Monitor.GetBlocks(func(res *jsonutils.JSONArray) { s.onGetBlocksSucc(res, callback) }) +} + +func (s *SGuestReloadDiskTask) onGetBlocksSucc(res *jsonutils.JSONArray, callback func(string)) { + var device string + devs, _ := res.GetArray() + for _, d := range devs { + device := s.getDiskOfDrive(d) + if len(device) > 0 { + callback(device) + break + } + } + + if len(device) == 0 { + s.taskFailed("Device not found") + } +} + +func (s *SGuestReloadDiskTask) getDiskOfDrive(d jsonutils.JSONObject) string { + inserted, err := d.Get("inserted") + if err != nil { + return "" + } + file, err := inserted.GetString("file") + if err != nil { + return "" + } + if file == s.disk.GetPath() { + drive, _ := d.GetString("device") + return drive + } + return "" +} + +func (s *SGuestReloadDiskTask) startReloadDisk(device string) { + s.doReloadDisk(device, s.onReloadSucc) +} + +func (s *SGuestReloadDiskTask) doReloadDisk(device string, callback func(string)) { + s.Monitor.SimpleCommand("stop", func(string) { + s.Monitor.ReloadDiskBlkdev(device, s.disk.GetPath(), callback) + }) +} + +func (s *SGuestReloadDiskTask) onReloadSucc(err string) { + if len(err) > 0 { + log.Errorf("monitor new snapshot blkdev error: %s", err) + } + s.Monitor.SimpleCommand(cont, s.onResumeSucc) +} + +func (s *SGuestReloadDiskTask) onResumeSucc(results string) { + log.Infof("guest reload disk task resume succ %s", results) + if _, err := hostutils.TaskComplete(ctx, nil); err != nil { + log.Errorln(err) + } +} + +func (s *SGuestReloadDiskTask) taskFailed(reason string) { + log.Errorf("SGuestReloadDiskTask error: %s", reason) + if _, err := hostutils.TaskFailed(s.ctx, reason); err != nil { + log.Errorln(err) + } +} + +/** + * GuestDiskSnapshotTask +**/ + +type SGuestDiskSnapshotTask struct { + *SGuestReloadDiskTask + + snapshotId string +} + +func NewGuestDiskSnapshotTask( + ctx context.Context, s *SKVMGuestInstance, disk *storageman.IDisk, snapshotId string, +) *SGuestDiskSnapshotTask { + return &SGuestDiskSnapshotTask{ + SGuestReloadDiskTask: NewGuestReloadDiskTask(ctx, s, disk), + snapshotId: snapshotId, + } +} + +func (s *SGuestDiskSnapshotTask) Start() { + s.fetchDisksInfo(s.startSnapshot) +} + +func (s *SGuestDiskSnapshotTask) startSnapshot(device string) { + s.doReloadDisk(device, s.onReloadBlkdevSucc) +} + +func (s *SGuestDiskSnapshotTask) onReloadBlkdevSucc(res string) { + var cb = s.onResumeSucc + if len(res) > 0 { + log.Errorf("Monitor reload blkdev error: %s", res) + cb = s.onSnapshotBlkdevFail + } + s.Monitor.SimpleCommand("cont", cb) +} + +func (s *SGuestDiskSnapshotTask) onSnapshotBlkdevFail(string) { + snapshotDir := s.disk.GetSnapshotDir() + snapshotPath := path.Join(snapshotDir, s.snapshotId) + err := exec.Command("rm", "-rf", snapshotPath).Run() + if err != nil { + log.Errorln(err) + } + _, err = hostutils.TaskFailed(s.ctx, "Reload blkdev error") + if err != nil { + log.Errorln(err) + } +} + +func (s *SGuestDiskSnapshotTask) onResumeSucc(res string) { + log.Infof("guest disk snapshot task resume succ %s", res) + snapshotDir := s.disk.GetSnapshotDir() + snapshotLocation := path.Join(snapshotDir, s.snapshotId) + _, err := hostutils.TaskComplete(s.ctx, jsonutils.NewDict( + jsonutils.JSONPair{"localtion", jsonutils.NewString(snapshotLocation)})) + if err != nil { + log.Errorln(err) + } +} + +/** + * GuestSnapshotDeleteTask +**/ + +type SGuestSnapshotDeleteTask struct { + *SGuestReloadDiskTask + deleteSnapshot string + convertSnapshot string + pendingDelete bool + + tmpPath string +} + +func NewGuestSnapshotDeleteTask( + ctx context.Context, s *SKVMGuestInstance, disk *storageman.IDisk, + deleteSnapshot, convertSnapshot string, pendingDelete string, +) *SGuestSnapshotDeleteTask { + return &SGuestSnapshotDeleteTask{ + SGuestReloadDiskTask: NewGuestReloadDiskTask(ctx, s, disk), + deleteSnapshot: deleteSnapshot, + convertSnapshot: convertSnapshot, + pendingDelete: pendingDelete, + } +} + +func (s *SGuestSnapshotDeleteTask) Start() { + if err := s.doDiskConvert(); err != nil { + s.taskFailed(err.Error()) + } + s.fetchDisksInfo(s.doReloadDisk) +} + +func (s *SGuestSnapshotDeleteTask) doDiskConvert() error { + snapshotDir := s.disk.GetSnapshotDir() + snapshotPath := path.Join(snapshotDir, s.convertSnapshot) + img, err := qemuimg.NewQemuImage(snapshotPath) + if err != nil { + log.Errorln(err) + return err + } + convertedDisk := snapshotPath + ".tmp" + if err = img.Convert2Qcow2To(convertedDisk, true); err != nil { + log.Errorln(err) + if fileutils2.Exists(convertedDisk) { + os.Remove(convertedDisk) + } + return err + } + + s.tmpPath = snapshotPath + ".swap" + if err := exec.Command("mv", "-f", snapshotPath, s.tmpPath).Run(); err != nil { + log.Errorln(err) + if fileutils2.Exists(s.tmpPath) { + exec.Command("mv", "-f", s.tmpPath, snapshotPath).Run() + } + return err + } + if err := exec.Command("mv", "-f", convertedDisk, snapshotPath).Run(); err != nil { + log.Errorln(err) + if fileutils2.Exists(s.tmpPath) { + exec.Command("mv", "-f", s.tmpPath, snapshotPath).Run() + } + return err + } + return nil +} + +func (s *SGuestSnapshotDeleteTask) doReloadDisk(device string) { + s.SGuestReloadDiskTask.doReloadDisk(device, s.onReloadBlkdevSucc) +} + +func (s *SGuestSnapshotDeleteTask) onReloadBlkdevSucc(err string) { + var callback = s.onResumeSucc + if len(err) > 0 { + log.Errorln("Reload blkdev failed: %s", err) + callback = s.onSnapshotBlkdevFail + } + s.Monitor.SimpleCommand("cont", callback) +} + +func (s *SGuestSnapshotDeleteTask) onSnapshotBlkdevFail(res string) { + snapshotPath := path.Join(s.disk.GetSnapshotDir(), s.convertSnapshot) + if err := exec.Command("rm", "-f", s.tmpPath, snapshotPath).Run(); err != nil { + log.Errorln(err) + } + s.taskFailed("Reload blkdev failed") +} + +func (s *SGuestSnapshotDeleteTask) onResumeSucc(res string) { + log.Infof("guest do new snapshot task resume succ %s", res) + if len(s.tmpPath) > 0 { + err := exec.Command("rm", "-f", s.tmpPath).Run() + if err != nil { + log.Errorln(err) + } + } + if !s.pendingDelete { + snapshotDir := s.disk.GetSnapshotDir() + exec.Command("rm", "-f", path.Join(snapshotDir, s.deleteSnapshot)) + } + _, err := hostutils.TaskComplete(s.ctx, + jsonutils.NewDict(jsonutils.JSONPair{"localtion", jsonutils.JSONTrue})) + if err != nil { + log.Errorln(err) + } +} diff --git a/pkg/hostman/guestman/qemu-kvm.go b/pkg/hostman/guestman/qemu-kvm.go index dc7f5c11b0..d9da4c14d0 100644 --- a/pkg/hostman/guestman/qemu-kvm.go +++ b/pkg/hostman/guestman/qemu-kvm.go @@ -15,17 +15,21 @@ import ( "yunion.io/x/jsonutils" "yunion.io/x/log" "yunion.io/x/pkg/util/regutils" + "yunion.io/x/pkg/util/seclib" "yunion.io/x/pkg/utils" "yunion.io/x/onecloud/pkg/cloudcommon/storagetypes" "yunion.io/x/onecloud/pkg/hostman/guestfs" + "yunion.io/x/onecloud/pkg/hostman/hostinfo/hostbridge" "yunion.io/x/onecloud/pkg/hostman/hostutils" "yunion.io/x/onecloud/pkg/hostman/monitor" "yunion.io/x/onecloud/pkg/hostman/options" "yunion.io/x/onecloud/pkg/hostman/storageman" "yunion.io/x/onecloud/pkg/mcclient/modules" + "yunion.io/x/onecloud/pkg/util/cgrouputils" "yunion.io/x/onecloud/pkg/util/fileutils2" "yunion.io/x/onecloud/pkg/util/timeutils2" + "yunion.io/x/onecloud/pkg/util/version" ) const ( @@ -38,11 +42,15 @@ const ( type SKVMGuestInstance struct { Id string + cgroupId string QemuVersion string + VncPassword string Desc *jsonutils.JSONDict Monitor monitor.Monitor manager *SGuestManager + + startupTask *SGuestResumeTask } func NewKVMGuestInstance(id string, manager *SGuestManager) *SKVMGuestInstance { @@ -68,8 +76,7 @@ func (s *SKVMGuestInstance) HomeDir() string { } func (s *SKVMGuestInstance) PrepareDir() error { - _, err := exec.Command("mkdir", "-p", s.HomeDir()).Output() - return err + return exec.Command("mkdir", "-p", s.HomeDir()).Run() } func (s *SKVMGuestInstance) GetPidFilePath() string { @@ -181,7 +188,7 @@ func (s *SKVMGuestInstance) asyncScriptStart(ctx context.Context, params interfa return nil, hostutils.ParamsError } - // TODO hostinof.instace().clean_deleted_ports + hostbridge.CleanDeletedPorts() time.Sleep(100 * time.Millisecond) var isStarted, tried = false, 0 @@ -305,8 +312,11 @@ 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, - func() { s.onMonitorConnected(ctx) }) + s.Monitor = monitor.NewQmpMonitor( + s.onMonitorDisConnect, + func(err error) { s.onMonitorTimeout(ctx, err) }, + func() { s.onMonitorConnected(ctx) }, + ) s.Monitor.Connect("127.0.0.1", s.GetQmpMonitorPort(-1)) } } @@ -337,11 +347,32 @@ func (s *SKVMGuestInstance) onGetQemuVersion(ctx context.Context, version string } func (s *SKVMGuestInstance) onMonitorDisConnect(err error) { + s.CleanStartupTask() + s.scriptStop() + if jsonutils.QueryBoolean(s.Desc, "is_slave", false) { + // TODO + // sync slave status + } else { + s.SyncStatus() + } // TODO + // s.clearCgroup() + s.Monitor = nil } -func (s *SKVMGuestInstance) onMonitorTimeout(err error) { - // TODO +func (s *SKVMGuestInstance) CleanStartupTask() { + log.Infof("Clean startup task ...") + if s.startupTask != nil { + s.startupTask.Stop() + s.startupTask = nil + } +} + +func (s *SKVMGuestInstance) onMonitorTimeout(ctx context.Context, err error) { + log.Errorf("Monitor connect timeout, VM %s frozen!! force restart!!!!", s.GetId()) + s.ForceStop() + timeutils2.AddTimeout(time.Second*3, + func() { s.asyncScriptStart(ctx, jsonutils.NewDict()) }) } func (s *SKVMGuestInstance) GetQmpMonitorPort(vncPort int) int { @@ -389,17 +420,22 @@ func (s *SKVMGuestInstance) SyncStatus() { hostutils.UpdateServerStatus(context.Background(), s.GetId(), status) } +func (s *SKVMGuestInstance) CheckBlockOrRunning(jobs int) { + var status = "running" + if jobs > 0 { + status = "block_stream" + } + _, err := hostutils.UpdateServerStatus(context.Background(), s.GetId(), status) + if err != nil { + log.Errorln(err) + } +} + func (s *SKVMGuestInstance) SaveDesc(desc jsonutils.JSONObject) error { - // TODO - // bw_info = self._get_bw_info() - // netmon_info = self._get_netmon_info() s.Desc = desc.(*jsonutils.JSONDict) if err := fileutils2.FilePutContents(s.GetDescFilePath(), desc.String()); err != nil { log.Errorln(err) } - // TODO - // self._update_bw_limit(bw_info) - // self._update_netmon_nic(netmon_info) } func (s *SKVMGuestInstance) StartGuest(ctx context.Context, params jsonutils.JSONObject) { @@ -709,3 +745,268 @@ func (s *SKVMGuestInstance) SyncConfig(ctx context.Context, desc jsonutils.JSONO NewGuestSyncConfigTaskExecutor(ctx, s, tasks, callBack).Start(1) return jsonutils.NewDict(jsonutils.JSONPair{"task", runTaskNames}), nil } + +func (s *SKVMGuestInstance) getApptags() []string { + var tags []string + meta, _ := s.Desc.Get("metadata") + if meta != nil && meta.Contains("app_tags") { + tagsStr, _ := meta.GetString("app_tags") + if len(tagsStr) > 0 { + return strings.Split(tagsStr, ",") + } + } + return tags +} + +func (s *SKVMGuestInstance) getStorageDeviceId() string { + disks, _ := s.Desc.GetArray("disks") + if len(disks) > 0 { + diskPath, _ := disks[0].GetString("path") + if len(diskPath) > 0 { + return fileutils2.GetDevId(diskPath) + } + } + return "" +} + +func (s *SKVMGuestInstance) SetCgroup() { + s.cgroupId = s.GetPid() + s.setCgroupIo() + s.setCgroupCpu() +} + +func (s *SKVMGuestInstance) setCgroupIo() { + appTags := s.getApptags() + params := map[string]int{} + if utils.IsInStringArray("io_hardlimit", appTags) { + devId := s.getStorageDeviceId() + if len(devId) == 0 { + log.Errorln("failed to get device ID (MAJOR:MINOR)") + return + } + params["blkio.throttle.read_bps_device"] = options.HostOptions.DefaultReadBpsPerCpu + params["blkio.throttle.read_iops_device"] = options.HostOptions.DefaultReadIopsPerCpu + params["blkio.throttle.write_bps_device"] = options.HostOptions.DefaultWriteBpsPerCpu + params["blkio.throttle.write_iops_device"] = options.HostOptions.DefaultWriteIopsPerCpu + cpu, _ := s.Desc.Int("cpu") + cgrouputils.CgroupIoHardlimitSet(s.cgroupId, int(cpu), params, devId) + } +} + +func (s *SKVMGuestInstance) setCgroupCpu() { + cpu, _ := s.Desc.Int("cpu") + cgrouputils.CgroupSet(s.cgroupId, int(cpu)) + + // TODO XXX + /* + var ( + cpuWeight = 1024 + cpuPeriod = 0 + cpuQuota = 0 + appTags = s.getApptags() + meta, _ = s.Desc.Get("metadata") + ) + + if meta != nil { + if meta.Contains("__cpu_weight") { + cpuWeight, _ = meta.Int("__cpu_weight") + } + if meta.Contains("__cpu_period") { + cpuPeriod, _ = meta.Int("__cpu_period") + } else { + cpuPeriod = -1 + } + if meta.Contains("__cpu_quota") { + cpuQuota, _ = meta.Int("__cpu_quota") + } else { + cpuQuota = -1 + } + } + */ +} + +func (s *SKVMGuestInstance) CreateFromDesc(desc jsonutils.JSONObject) error { + if !s.PrepareDir() { + uuid, _ := desc.GetString("uuid") + return fmt.Errorf("Failed to create server dir %s", uuid) + } + return s.SaveDesc(desc) +} + +func (s *SKVMGuestInstance) GetNeedMergeBackingFileDiskIndexs() []int { + res := make([]int, 0) + disks, _ := s.Desc.GetArray("disks") + for _, disk := range disks { + if jsonutils.QueryBoolean(disk, "merge_snapshot", false) { + diskIdx, _ := disk.Int("index") + res = append(res, int(diskIdx)) + } + } + return res +} + +func (s *SKVMGuestInstance) streamDisksComplete(ctx context.Context) { + disks, _ := s.Desc.GetArray("disks") + for i, disk := range disks { + diskpath, _ := disk.GetString("path") + d := storageman.GetManager().GetDiskByPath(diskPath) + if d != nil { + d.PostCreateFromImageFuse() + } + if jsonutils.QueryBoolean(disk, "merge_snapshot", false) { + d := disks[i].(*jsonutils.JSONDict) + d.Set("merge_snapshot", jsonutils.JSONFalse) + } + } + s.SaveDesc(s.Desc) + _, err := modules.Servers.PerformAction(ctx, s.GetId(), "stream-disks-complete", nil) + if err != nil { + log.Infof("stream disks complete sync error %s", err) + } +} + +func (s *SKVMGuestInstance) GetQemuVersionStr() string { + return s.QemuVersion +} + +func (s *SKVMGuestInstance) SyncMetadata(meta *jsonutils.JSONDict) { + _, err := modules.Servers.SetMetadata(hostutils.GetComputeSession(context.Background()), + s.GetId(), meta) + if err != nil { + log.Errorln(err) + } +} + +func (s *SKVMGuestInstance) SetVncPassword() { + password := seclib.RandomPassword(8) + s.VncPassword = password + timeutils2.AddTimeout(time.Second*3, + func() { s.Monitor.SetVncPassword(s.GetVdiProtocol(), password, nil) }) +} + +func (s *SKVMGuestInstance) ListStateFilePaths() []string { + var ret = []string{} + if fileutils2.Exists(s.HomeDir()) { + files, err := ioutil.ReadDir(s.HomeDir()) + if err != nil { + log.Errorln(err) + return nil + } + for _, f := range files { + if strings.HasPrefix(f.Name(), STATE_FILE_PREFIX) { + ret = append(ret, path.Join(s.HomeDir(), f.Name())) + } + } + } + return ret +} + +// 好像不用了 +func (s *SKVMGuestInstance) CleanStatefiles() { + for _, stateFile := range s.ListStateFilePaths() { + if err := exec.Command("mountpoint", stateFile).Run(); err == nil { + if err = exec.Command("umount", stateFile).Run(); err != nil { + log.Errorln(err) + } + } + if err := exec.Command("rm", "-rf", stateFile).Run(); err != nil { + log.Errorln(err) + } + } + if err := exec.Command("rm", "-rf", s.GetFuseTmpPath()); err != nil { + log.Errorln(err) + } +} + +func (s *SKVMGuestInstance) GetFuseTmpPath() string { + return path.Join(s.HomeDir(), "tmp") +} + +func (s *SKVMGuestInstance) StreamDisks(ctx context.Context, callback func(), disksIdx []int) { + log.Infof("Start guest block stream task ...") + task := NewGuestStreamDisksTask(ctx, s, callback, disksIdx) + task.Start() +} + +func (s *SKVMGuestInstance) isLiveSnapshotEnabled() bool { + if version.GE(s.QemuVersion, "2.12.1") { + return true + } else { + return false + } +} + +func (s *SKVMGuestInstance) ExecReloadDiskTask(ctx context.Context, disk storageman.IDisk) (jsonutils.JSONObject, error) { + if s.IsRunning() { + if s.isLiveSnapshotEnabled() { + task := NewGuestReloadDiskTask(ctx, s, disk) + task.WaitSnapshotReplaced(task.Start) + return nil, nil + } else { + return nil, fmt.Errorf("Guest dosen't support reload disk") + } + } else { + return jsonutils.NewDict(jsonutils.JSONPair{"reopen", jsonutils.JSONTrue}), nil + } +} + +func (s *SKVMGuestInstance) ExecDiskSnapshotTask( + ctx context.Context, disk storageman.IDisk, snapshotId string, +) (jsonutils.JSONObject, error) { + if s.IsRunning() { + if !s.isLiveSnapshotEnabled() { + return nil, fmt.Errorf("Guest dosen't support live snapshot") + } + err := disk.CreateSnapshot(snapshotId) + if err != nil { + return nil, err + } + task := NewGuestDiskSnapshotTask(ctx, s, disk, snapshotId) + task.Start() + return nil, nil + } else { + return s.StaticSaveSnapshot(ctx, disk, snapshotId) + } +} + +func (s *SKVMGuestInstance) StaticSaveSnapshot( + ctx context.Context, disk storageman.IDisk, snapshotId string, +) (jsonutils.JSONObject, error) { + err := disk.CreateSnapshot(snapshotId) + if err != nil { + return nil, err + } + location := path.Join(disk.GetSnapshotDir(), snapshotId) + return jsonutils.NewDict(jsonutils.JSONPair{"localtion", + jsonutils.NewString(snapshotLocation)}), nil +} + +func (s *SKVMGuestInstance) ExecDeleteSnapshotTask( + ctx context.Context, disk storageman.IDisk, + deleteSnapshot string, convertSnapshot string, pendingDelete bool, +) (jsonutils.JSONObject, error) { + if s.IsRunning() { + if s.isLiveSnapshotEnabled() { + task := NewGuestSnapshotDeleteTask(ctx, s, disk, + deleteSnapshot, convertSnapshot, pendingDelete) + task.Start() + return nil, nil + } else { + return nil, fmt.Errorf("Guest dosen't support live snapshot delete") + } + } else { + return s.deleteStaticSnapshotFile(ctx, disk, deleteSnapshot, + convertSnapshot, pendingDelete) + } +} + +func (s *SKVMGuestInstance) deleteStaticSnapshotFile( + ctx context.Context, disk storageman.IDisk, + deleteSnapshot string, convertSnapshot string, pendingDelete bool, +) (jsonutils.JSONObject, error) { + if err := disk.DeleteSnapshot(deleteSnapshot, convertSnapshot, pendingDelete); err != nil { + log.Errorln(err) + return nil, err + } + return jsonutils.NewDict(jsonutils.JSONPair{"localtion", jsonutils.JSONTrue}), nil +} diff --git a/pkg/hostman/guestman/qemu-kvmhelper.go b/pkg/hostman/guestman/qemu-kvmhelper.go index 74f6bf5e91..29b5c2af37 100644 --- a/pkg/hostman/guestman/qemu-kvmhelper.go +++ b/pkg/hostman/guestman/qemu-kvmhelper.go @@ -80,6 +80,14 @@ func (s *SKVMGuestInstance) isQ35() bool { return s.getMachine() == "q35" } +func (s *SKVMGuestInstance) GetVdiProtocol() string { + vdi, err := s.Desc.GetString("vdi") + if err != nil { + vdi = "vnc" + } + return vdi +} + func (s *SKVMGuestInstance) GetPciBus() string { if s.isQ35() { return "pcie.0" @@ -512,7 +520,7 @@ func (s *SKVMGuestInstance) generateStartScript(data *jsonutils.JSONDict) string return cmd } -func generateStartScript(data *jsonutils.JSONDict) string { +func (s *SKVMGuestInstance) generateStopScript(data *jsonutils.JSONDict) string { var ( uuid, _ = s.Desc.GetString("uuid") nics, _ = s.Desc.GetArray("nics") diff --git a/pkg/hostman/hostinfo/hostbridge.go b/pkg/hostman/hostinfo/hostbridge/hostbridge.go similarity index 96% rename from pkg/hostman/hostinfo/hostbridge.go rename to pkg/hostman/hostinfo/hostbridge/hostbridge.go index db61efc4d9..1c1e271757 100644 --- a/pkg/hostman/hostinfo/hostbridge.go +++ b/pkg/hostman/hostinfo/hostbridge/hostbridge.go @@ -1,4 +1,4 @@ -package hostinfo +package hostbridge import ( "fmt" @@ -11,6 +11,7 @@ import ( "yunion.io/x/onecloud/pkg/hostman/options" "yunion.io/x/onecloud/pkg/util/netutils2" + "yunion.io/x/onecloud/pkg/util/ovsutils" ) type IBridgeDriver interface { @@ -254,6 +255,10 @@ func (o *SOVSBridgeDriver) SetupBridgeDev() error { return nil } +func CleanOvsBridge() { + ovsutils.CleanAllHiddenPorts() +} + func NewOVSBridgeDriver(bridge, inter, ip string) (*SOVSBridgeDriver, error) { base, err := NewBaseBridgeDriver(bridge, inter, ip) if err != nil { @@ -269,3 +274,9 @@ func NewDriver(bridgeDriver, bridge, inter, ip string) (IBridgeDriver, error) { return nil, fmt.Errorf("Not Implentment") } } + +func CleanDeletedPorts() { + if options.HostOptions.BridgeDriver == "openvswitch" { + CleanOvsBridge() + } +} diff --git a/pkg/hostman/hostinfo/hostinfo.go b/pkg/hostman/hostinfo/hostinfo.go index e15f08eaa8..63162421fd 100644 --- a/pkg/hostman/hostinfo/hostinfo.go +++ b/pkg/hostman/hostinfo/hostinfo.go @@ -22,6 +22,7 @@ import ( "yunion.io/x/onecloud/pkg/hostman/options" "yunion.io/x/onecloud/pkg/hostman/storageman" "yunion.io/x/onecloud/pkg/mcclient/modules" + "yunion.io/x/onecloud/pkg/util/cgrouputils" "yunion.io/x/onecloud/pkg/util/fileutils2" "yunion.io/x/onecloud/pkg/util/netutils2" "yunion.io/x/onecloud/pkg/util/qemutils" @@ -178,10 +179,9 @@ func (h *SHostInfo) prepareEnv() error { e := err.(*exec.ExitError) log.Errorln(e.Stderr) } - // TODO: cgrouputils 还未实现 - // if !cgrouputils.Init() { - // return fmt.Errorf("Cannot initialize control group subsystem") - // } + if !cgrouputils.Init() { + return fmt.Errorf("Cannot initialize control group subsystem") + } _, err = exec.Command("rmmod", "nbd").Output() if err != nil { diff --git a/pkg/hostman/hostinfo/hostinfohelper.go b/pkg/hostman/hostinfo/hostinfohelper.go index e6f2d44867..e3b9e4e395 100644 --- a/pkg/hostman/hostinfo/hostinfohelper.go +++ b/pkg/hostman/hostinfo/hostinfohelper.go @@ -18,6 +18,7 @@ import ( "yunion.io/x/pkg/util/regutils" "yunion.io/x/onecloud/pkg/cloudcommon/types" + "yunion.io/x/onecloud/pkg/hostman/hostinfo/hostbridge" "yunion.io/x/onecloud/pkg/hostman/hostinfo/hostdhcp" "yunion.io/x/onecloud/pkg/hostman/hostutils" "yunion.io/x/onecloud/pkg/hostman/options" @@ -170,7 +171,7 @@ type SNIC struct { WireId string Bandwidth int - BridgeDev IBridgeDriver + BridgeDev hostbridge.IBridgeDriver dhcpServer *hostdhcp.SGuestDHCPServer } @@ -228,9 +229,9 @@ func NewNIC(desc string) (*SNIC, error) { } } - // default openvswitch var err error - nic.BridgeDev, err = NewDriver(options.HostOptions.BridgeDriver, + + nic.BridgeDev, err = hostbridge.NewDriver(options.HostOptions.BridgeDriver, nic.Bridge, nic.Inter, nic.Ip) if err != nil { log.Errorln(err) diff --git a/pkg/hostman/monitor/hmp.go b/pkg/hostman/monitor/hmp.go index 45566c4ef0..be04ae2f28 100644 --- a/pkg/hostman/monitor/hmp.go +++ b/pkg/hostman/monitor/hmp.go @@ -217,6 +217,7 @@ func (m *HmpMonitor) GetBlocks(callback func(*jsonutils.JSONArray)) { break } } + drv.Set("inserted", inserted) } outputJson.Add(drv) } @@ -259,3 +260,54 @@ func (m *HmpMonitor) DeviceAdd(dev string, params map[string]interface{}, callba } m.Query(fmt.Sprintf("device_add %s,%s", dev, strings.Join(paramsKvs, ",")), callback) } + +func (m *HmpMonitor) MigrateSetCapability(capability, state string, callback StringCallback) { + m.Query(fmt.Sprintf("migrate_set_capability %s %s", capability, state), callback) +} + +func (m *HmpMonitor) Migrate( + destStr string, copyIncremental, copyFull bool, callback StringCallback, +) { + cmd := "migrate -d" + if copyIncremental { + cmd += " -i" + } else if copyIncremental { + cmd += " -b" + } + cmd += " " + destStr + m.Query(cmd, callback) +} + +func (m *HmpMonitor) GetMigrateStatus(callback StringCallback) { + cb := func(output string) { + log.Infof("Query migrate status: %s", output) + + var status string + for _, line := range strings.Split(output, "\n") { + if strings.HasPrefix(line, "Migration status") { + status = line[strings.LastIndex(line, " ")+1:] + break + } + } + callback(status) + } + + m.Query("info migrate", cb) +} + +func (m *HmpMonitor) GetBlockJobs(callback func(jobs int)) { + cb := func(output string) { + lines := strings.Split(output, "\n") + if lines[0] == "No active jobs" { + callback(0) + } else { + callback(len(lines)) + } + } + + 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 efb1c6e16d..7dc9851105 100644 --- a/pkg/hostman/monitor/monitor.go +++ b/pkg/hostman/monitor/monitor.go @@ -21,6 +21,7 @@ type Monitor interface { SimpleCommand(cmd string, callback StringCallback) QueryStatus(StringCallback) GetVersion(StringCallback) + GetBlockJobs(func(jobs int)) GetBlocks(callback func(*jsonutils.JSONArray)) EjectCdrom(dev string, callback StringCallback) @@ -31,6 +32,12 @@ type Monitor interface { DriveAdd(bus string, params map[string]string, callback StringCallback) DeviceAdd(dev string, params map[string]interface{}, callback StringCallback) + + MigrateSetCapability(capability, state string, callback StringCallback) + Migrate(destStr string, copyIncremental, copyFull bool, callback StringCallback) + GetMigrateStatus(callback StringCallback) + + ReloadDiskBlkdev(device, path string, callback StringCallback) } type MonitorErrorFunc func(error) diff --git a/pkg/hostman/monitor/qmp.go b/pkg/hostman/monitor/qmp.go index 0756ebb9bf..c78805cdfd 100644 --- a/pkg/hostman/monitor/qmp.go +++ b/pkg/hostman/monitor/qmp.go @@ -458,3 +458,108 @@ func (m *QmpMonitor) DeviceAdd(dev string, params map[string]interface{}, callba m.Query(cmd, cb) } + +func (m *QmpMonitor) MigrateSetCapability(capability, state string, callback StringCallback) { + var ( + cb = func(res *Response) { + callback(m.actionResult(res)) + } + st = false + ) + if state == "on" { + st = true + } + + cmd := &Command{ + Execute: "query-migrate-capabilities", + Args: map[string]interface{}{ + "capabilities": []interface{}{ + map[string]interface{}{ + "capability": capability, + "state": st, + }, + }, + }, + } + + m.Query(cmd, cb) +} + +func (m *QmpMonitor) Migrate( + destStr string, copyIncremental, copyFull bool, callback StringCallback, +) { + var ( + cb = func(res *Response) { + callback(m.actionResult(res)) + } + cmd = &Command{ + Execute: "migrate", + Args: map[string]interface{}{ + "uri": destStr, + "blk": copyFull, + "inc": copyIncremental, + }, + } + ) + + m.Query(cmd, cb) +} + +func (m *QmpMonitor) GetMigrateStatus(callback StringCallback) { + var ( + cmd = &Command{Execute: "query-migrate"} + cb = func(res *Response) { + if res.ErrorVal != nil { + callback(res.ErrorVal.Error()) + } else { + ret, err := jsonutils.Parse(res.Return) + if err != nil { + log.Errorf("Parse qmp res error: %s", err) + callback("") + } else { + log.Infof("Query migrate status: %s", ret.String()) + status, _ := ret.GetString("status") + callback(status) + } + } + } + ) + m.Query(cmd, cb) +} + +func (m *QmpMonitor) GetBlockJobs(callback func(jobs int)) { + var cb = func(res *Response) { + if res.ErrorVal != nil { + log.Errorln(res.ErrorVal.Error()) + callback(-1) + } else { + ret, err := jsonutils.Parse(res.Return) + if err != nil { + log.Errorf("Parse qmp res error: %s", err) + callback(-1) + } else { + jobs, _ := ret.GetArray() + callback(len(jobs)) + } + } + } + m.Query(&Command{Execute: "query-block-jobs"}, cb) +} + +func (m *QmpMonitor) ReloadDiskBlkdev(device, path string, callback StringCallback) { + var ( + cb = func(res *Response) { + callback(m.actionResult(res)) + } + cmd = &Command{ + Execute: "reload-disk-snapshot-blkdev-sync", + Args: map[string]string{ + "device": device, + "snapshot-file": path, + "mode": "existing", + "format": "qcow2", + }, + } + ) + m.Query(cmd, cb) +} diff --git a/pkg/hostman/options/options.go b/pkg/hostman/options/options.go index fabae363de..8f6bb9da95 100644 --- a/pkg/hostman/options/options.go +++ b/pkg/hostman/options/options.go @@ -65,6 +65,12 @@ type SHostOptions struct { FetcherfsPath string `default:"/opt/yunion/fetchclient/bin/fetcherfs" help:"Fuse fetcherfs path"` DefaultImageSaveFormat string `default:"qcow2" help:"Default image save format, default is qcow2, canbe vmdk"` + + DefaultReadBpsPerCpu int `default:"163840000" help:"Default read bps per cpu for hard IO limit"` + DefaultReadIopsPerCpu int `default:"1250" help:"Default read iops per cpu for hard IO limit"` + DefaultWriteBpsPerCpu int `default:"54525952" help:"Default write bps per cpu for hard IO limit"` + DefaultWriteIopsPerCpu int `default:"416" help:"Default write iops per cpu for hard IO limit"` + SetVncPassword bool `default:"true" help:"Auto set vnc password after monitor connected"` } var HostOptions SHostOptions diff --git a/pkg/hostman/storageman/core.go b/pkg/hostman/storageman/core.go index 0f591da6a0..186628b3b4 100644 --- a/pkg/hostman/storageman/core.go +++ b/pkg/hostman/storageman/core.go @@ -10,6 +10,8 @@ import ( "yunion.io/x/onecloud/pkg/hostman/options" ) +const MINIMAL_FREE_SPACE = 128 + type IHost interface { GetZone() string GetHostId() string @@ -17,25 +19,6 @@ type IHost interface { GetMediumType() string } -const MINIMAL_FREE_SPACE = 128 - -var storageManager *SStorageManager - -func Init(host IHost) error { - var ( - err error - zone = host.GetZone() - hostId = host.GetHostId() - ) - - storageManager, err = NewStorageManager(host) - return err -} - -func GetManager() *SStorageManager { - return storageManager -} - type SStorageManager struct { host IHost @@ -186,16 +169,16 @@ func (s *SStorageManager) initLocalStorageImagecache() error { // } // } -func (s *SStorageManager) AddNfsStorage(storagecacheId, cachePath string) { - if len(cachePath) == 0 { - return - } - if s.NfsStorageImagecacheManagers == nil { - s.NfsStorageImagecacheManagers = make(map[string]IImageCacheManger, 0) - } - s.NfsStorageImagecacheManagers[storagecacheId] = NewLocalImageCacheManager(s, cachePath, - options.HostOptions.ImageCacheLimit, true, storagecacheId) -} +// func (s *SStorageManager) AddNfsStorage(storagecacheId, cachePath string) { +// if len(cachePath) == 0 { +// return +// } +// if s.NfsStorageImagecacheManagers == nil { +// s.NfsStorageImagecacheManagers = make(map[string]IImageCacheManger, 0) +// } +// s.NfsStorageImagecacheManagers[storagecacheId] = NewLocalImageCacheManager(s, cachePath, +// options.HostOptions.ImageCacheLimit, true, storagecacheId) +// } func (s *SStorageManager) GetStorage(storageId string) IStorage { for _, storage := range s.Storages { @@ -282,3 +265,24 @@ func (s *SStorageManager) InitSharedStorageImageCache(storageType, storagecacheI } } + +var storageManager *SStorageManager + +func GetManager() *SStorageManager { + return storageManager +} + +func Manager() *SStorageManager { + return storageManager +} + +func Init(host IHost) error { + var ( + err error + zone = host.GetZone() + hostId = host.GetHostId() + ) + + storageManager, err = NewStorageManager(host) + return err +} diff --git a/pkg/hostman/storageman/diskbase.go b/pkg/hostman/storageman/diskbase.go index 3524955f0d..9425593fb1 100644 --- a/pkg/hostman/storageman/diskbase.go +++ b/pkg/hostman/storageman/diskbase.go @@ -24,6 +24,8 @@ type IDisk interface { Resize(ctx context.Context, params interface{}) (jsonutils.JSONObject, error) GetPath() string + GetSnapshotDir() string + CreateFromUrl(context.Context, string) error // CreateFromSnapshot CreateFromTemplate(context.Context, string, string, int64) (jsonutils.JSONObject, error) @@ -31,8 +33,13 @@ type IDisk interface { CreateRaw(ctx context.Context, sizeMb int, diskFromat string, fsFormat string, encryption bool, diskId string, back string) (jsonutils.JSONObject, error) + CreateSnapshot(snapshotId string) error + DeleteSnapshot(snapshotId string) error + // @params: diskPath, guestDesc, deployInfo DeployGuestFs(string, *jsonutils.JSONDict, *guestfs.SDeployInfo) (jsonutils.JSONObject, error) + + PostCreateFromImageFuse() } type SBaseDisk struct { @@ -88,7 +95,7 @@ func (d *SBaseDisk) DeployGuestFs(diskPath string, guestDesc *jsonutils.JSONDict if root := kvmDisk.Mount(); root != nil { defer kvmDisk.Umount(root) - return root.DeployGuestFs(root, guestDesc, deployInfo) + return guestfs.DeployGuestFs(root, guestDesc, deployInfo) } } return nil, fmt.Errorf("Kvm disk connect or mount error") diff --git a/pkg/hostman/storageman/diskhandler.go b/pkg/hostman/storageman/diskhandler.go index 1a320fff0a..093de8ce70 100644 --- a/pkg/hostman/storageman/diskhandler.go +++ b/pkg/hostman/storageman/diskhandler.go @@ -38,7 +38,6 @@ func AddDiskHandler(prefix string, app *appsrv.Application) { auth.Authenticate(deleteImageCache)) } - // excuse me ？ app.AddHandler("POST", fmt.Sprintf("%s/%s//upload", prefix, keyWord), auth.Authenticate(saveToGlance)) diff --git a/pkg/hostman/storageman/disklocal.go b/pkg/hostman/storageman/disklocal.go index 1a818a809b..77b89ac87d 100644 --- a/pkg/hostman/storageman/disklocal.go +++ b/pkg/hostman/storageman/disklocal.go @@ -11,6 +11,7 @@ import ( "yunion.io/x/log" "yunion.io/x/onecloud/pkg/hostman/hostutils" "yunion.io/x/onecloud/pkg/hostman/options" + "yunion.io/x/onecloud/pkg/hostman/storageman/remotefile" "yunion.io/x/onecloud/pkg/mcclient/auth" "yunion.io/x/onecloud/pkg/util/fileutils2" "yunion.io/x/onecloud/pkg/util/fuseutils" @@ -47,6 +48,10 @@ func (d *SLocalDisk) GetPath() string { } } +func (d *SLocalDisk) GetSnapshotDir() string { + return path.Join(d.Storage.GetSnapshotDir(), d.Id+"_snap") +} + func (d *SLocalDisk) Probe() error { if _, err := os.Stat(d.getPath()); !os.IsNotExist(err) { d.isAlter = false @@ -168,7 +173,7 @@ func (d *SLocalDisk) CreateFromImageFuse(ctx context.Context, url string) error func (d *SLocalDisk) CreateFromTemplate(ctx context.Context, imageId, format string, size int64) (jsonutils.JSONObject, error) { var imageCacheManager = storageManager.LocalStorageImagecacheManager - imageCache := imageCacheManager.AcquireImage(ctx, imageId, d.GetZone(), "") + imageCache := imageCacheManager.AcquireImage(ctx, imageId, d.GetZone(), "", "") if imageCache != nil { defer imageCacheManager.ReleaseImage(imageId) cacheImagePath := imageCache.GetPath() @@ -197,8 +202,21 @@ func (d *SLocalDisk) CreateFromTemplate(ctx context.Context, imageId, format str } } +func (d *SLocalDisk) CreateFromUrl(ctx context.Context, url string) error { + remoteFile := remotefile.NewRemoteFile(ctx, url, d.getPath(), false, "", -1, nil, "", "") + if remoteFile.Fetch() { + if options.HostOptions.EnableFallocateDisk { + //TODO + // d.fallocate() + } + return nil + } else { + return fmt.Errorf("Fail to fetch image from %s", url) + } +} + func (d *SLocalDisk) CreateRaw(ctx context.Context, sizeMB int, diskFormat, fsFormat string, - encryption bool, diskId string, back string) (jsonutils.JSONObject, error) { + encryption bool, uuid string, back string) (jsonutils.JSONObject, error) { if fileutils2.Exists(d.GetPath()) { os.Remove(d.GetPath()) } @@ -229,19 +247,19 @@ func (d *SLocalDisk) CreateRaw(ctx context.Context, sizeMB int, diskFormat, fsFo } if utils.IsInStringArray(fsFormat, []string{"swap", "ext2", "ext3", "ext4", "xfs"}) { - d.FormatFs(fsFormat, diskId) + d.FormatFs(fsFormat, uuid) } return d.GetDiskDesc(), nil } -func (d *SLocalDisk) FormatFs(fsFormat, diskId string) { - log.Infof("Make disk %s fs %s", diskId, fsFormat) +func (d *SLocalDisk) FormatFs(fsFormat, uuid string) { + log.Infof("Make disk %s fs %s", uuid, fsFormat) gd := NewKVMGuestDisk(d.GetPath()) if gd.Connect() { defer gd.Disconnect() if err := gd.MakePartition(fsFormat); err == nil { - err = gd.FormatPartition(fsFormat, diskId) + err = gd.FormatPartition(fsFormat, uuid) if err != nil { log.Errorln(err) } @@ -274,3 +292,13 @@ func (d *SLocalDisk) GetDiskSetupScripts(diskIndex int) string { cmd += "fi\n" return cmd } + +func (d *SLocalDisk) PostCreateFromImageFuse() { + mntPath := path.Join(d.Storage.GetFuseMountPath(), d.Id) + if err := exec.Command("umount", mntPath).Run(); err != nil { + log.Errorln(err) + } + if err := exec.Command("rm", "-rf", mntPath).Run(); err != nil { + log.Errorln(err) + } +} diff --git a/pkg/hostman/storageman/diskutils.go b/pkg/hostman/storageman/diskutils.go index 153acb6d40..4a688e61b6 100644 --- a/pkg/hostman/storageman/diskutils.go +++ b/pkg/hostman/storageman/diskutils.go @@ -12,6 +12,8 @@ import ( "yunion.io/x/log" "yunion.io/x/onecloud/pkg/hostman/guestfs" + "yunion.io/x/onecloud/pkg/hostman/guestfs/fsdriver" + "yunion.io/x/onecloud/pkg/hostman/storageman/nbd" "yunion.io/x/onecloud/pkg/util/fileutils2" "yunion.io/x/onecloud/pkg/util/qemutils" ) @@ -32,7 +34,7 @@ func NewKVMGuestDisk(imagePath string) *SKVMGuestDisk { } func (d *SKVMGuestDisk) Connect() bool { - d.nbdDev = nbdManager.AcquireNbddev() + d.nbdDev = nbd.GetNBDManager().AcquireNbddev() if len(d.nbdDev) == 0 { log.Errorln("Cannot get nbd device") return false @@ -111,7 +113,7 @@ func (d *SKVMGuestDisk) Disconnect() bool { log.Errorln(err.Error()) return false } - nbdManager.ReleaseNbddev(d.nbdDev) + nbd.GetNBDManager().ReleaseNbddev(d.nbdDev) d.nbdDev = "" d.partitions = d.partitions[len(d.partitions):] return true @@ -120,7 +122,7 @@ func (d *SKVMGuestDisk) Disconnect() bool { } } -func (d *SKVMGuestDisk) Mount() guestfs.IRootFsDriver { +func (d *SKVMGuestDisk) Mount() fsdriver.IRootFsDriver { for i := 0; i < len(d.partitions); i++ { if d.partitions[i].Mount() { if fs := guestfs.DetectRootFs(d.partitions[i]); fs != nil { @@ -134,7 +136,7 @@ func (d *SKVMGuestDisk) Mount() guestfs.IRootFsDriver { return nil } -func (d *SKVMGuestDisk) Umount(fd guestfs.IRootFsDriver) { +func (d *SKVMGuestDisk) Umount(fd fsdriver.IRootFsDriver) { if part := fd.GetPartition(); part != nil { part.Umount() } diff --git a/pkg/hostman/storageman/imagecachebase.go b/pkg/hostman/storageman/imagecachebase.go index 9e4a856151..3dbcc38f0b 100644 --- a/pkg/hostman/storageman/imagecachebase.go +++ b/pkg/hostman/storageman/imagecachebase.go @@ -31,7 +31,7 @@ type IImageCache interface { Load() bool Acquire(ctx context.Context, zone, srcUrl, format string) bool Release() - Remove() error + Remove(ctx context.Context) error GetImageId() string GetDesc() *remotefile.SImageDesc @@ -62,6 +62,10 @@ func (l *SLocalImageCache) GetDesc() *remotefile.SImageDesc { return l.Desc } +func (l *SLocalImageCache) GetImageId() string { + return l.imageId +} + func (l *SLocalImageCache) Load() bool { var ( imgPath = l.GetPath() @@ -218,15 +222,21 @@ func (l *SLocalImageCache) fetch(ctx context.Context, zone, srcUrl, format strin } } -func (l *SLocalImageCache) Remove(ctx context.Context) { +func (l *SLocalImageCache) Remove(ctx context.Context) error { if fileutils2.Exists(l.GetPath()) { - syscall.Unlink(l.GetPath()) + if err := syscall.Unlink(l.GetPath()); err != nil { + return err + } } if fileutils2.Exists(l.GetInfPath()) { - syscall.Unlink(l.GetInfPath()) + if err := syscall.Unlink(l.GetInfPath()); err != nil { + return err + } } if fileutils2.Exists(l.GetTmpPath()) { - syscall.Unlink(l.GetTmpPath()) + if err := syscall.Unlink(l.GetTmpPath()); err != nil { + return err + } } go func() { @@ -236,6 +246,8 @@ func (l *SLocalImageCache) Remove(ctx context.Context) { log.Errorf("Fail to delete host cached image: %s", err) } }() + + return nil } func (l *SLocalImageCache) GetPath() string { diff --git a/pkg/hostman/storageman/imagecachemanager.go b/pkg/hostman/storageman/imagecachemanager.go index 623f9e05c2..0e7760c592 100644 --- a/pkg/hostman/storageman/imagecachemanager.go +++ b/pkg/hostman/storageman/imagecachemanager.go @@ -85,7 +85,7 @@ func (c *SLocalImageCacheManager) loadCache() { func (c *SLocalImageCacheManager) loadImageCache(imageId string) { imageCache := NewLocalImageCache(imageId, c) - if err := imageCache.Load(); err != nil { + if imageCache.Load() { c.cachedImages[imageId] = imageCache } } @@ -101,7 +101,15 @@ func (c *SLocalImageCacheManager) AcquireImage(ctx context.Context, imageId, zon if img.Acquire(ctx, zone, srcUrl, format) { return img } else { - return false + return nil + } +} + +func (c *SLocalImageCacheManager) ReleaseImage(imageId string) { + c.mutex.Lock() + defer c.mutex.Unlock() + if img, ok := c.cachedImages[imageId]; ok { + img.Release() } } @@ -112,18 +120,18 @@ func (c *SLocalImageCacheManager) DeleteImageCache(ctx context.Context, data int } imageId, _ := body.GetString("image_id") - c.removeImage(ctx, imageId) - return nil, nil + return nil, c.removeImage(ctx, imageId) } -func (c *SLocalImageCacheManager) removeImage(ctx context.Context, imageId string) { +func (c *SLocalImageCacheManager) removeImage(ctx context.Context, imageId string) error { c.mutex.Lock() defer c.mutex.Unlock() if img, ok := c.cachedImages[imageId]; ok { delete(c.cachedImages, imageId) - img.Remove(ctx) + return img.Remove(ctx) } + return nil } func (c *SLocalImageCacheManager) PrefetchImageCache(ctx context.Context, data interface{}) (jsonutils.JSONObject, error) { diff --git a/pkg/hostman/storageman/nbdman.go b/pkg/hostman/storageman/nbd/nbdman.go similarity index 98% rename from pkg/hostman/storageman/nbdman.go rename to pkg/hostman/storageman/nbd/nbdman.go index 0d51f85d78..b1a4d57e2b 100644 --- a/pkg/hostman/storageman/nbdman.go +++ b/pkg/hostman/storageman/nbd/nbdman.go @@ -1,4 +1,4 @@ -package storageman +package nbd import ( "fmt" diff --git a/pkg/hostman/storageman/storagebase.go b/pkg/hostman/storageman/storagebase.go index a7965ab015..8e59f218c0 100644 --- a/pkg/hostman/storageman/storagebase.go +++ b/pkg/hostman/storageman/storagebase.go @@ -27,6 +27,7 @@ type IStorage interface { SetPath(string) GetPath() string + GetSnapshotDir() string GetFreeSizeMb() int GetCapacity() int @@ -41,6 +42,8 @@ type IStorage interface { CreateDiskByDiskinfo(context.Context, interface{}) (jsonutils.JSONObject, error) SaveToGlance(context.Context, interface{}) (jsonutils.JSONObject, error) + CreateSnapshotFormUrl(ctx context.Context, snapshotUrl, diskId, snapshotPath string) error + DeleteDiskfile(diskPath string) error GetFuseTmpPath() string GetFuseMountPath() string diff --git a/pkg/hostman/storageman/storagelocal.go b/pkg/hostman/storageman/storagelocal.go index 56df546e96..34ee4205f8 100644 --- a/pkg/hostman/storageman/storagelocal.go +++ b/pkg/hostman/storageman/storagelocal.go @@ -14,6 +14,7 @@ import ( "yunion.io/x/onecloud/pkg/hostman/guestfs/fsdriver" "yunion.io/x/onecloud/pkg/hostman/hostutils" "yunion.io/x/onecloud/pkg/hostman/options" + "yunion.io/x/onecloud/pkg/hostman/storageman/remotefile" "yunion.io/x/onecloud/pkg/mcclient/modules" "yunion.io/x/onecloud/pkg/util/fileutils2" "yunion.io/x/onecloud/pkg/util/qemuimg" @@ -49,6 +50,10 @@ func (s *SLocalStorage) StorageType() string { return storagetypes.STORAGE_LOCAL } +func (s *SLocalStorage) GetSnapshotDir() string { + return path.Join(s.Path, _SNAPSHOT_PATH_) +} + func (s *SLocalStorage) SyncStorageInfo() { content := jsonutils.NewDict() content.Set("name", jsonutils.NewString(s.StorageName)) @@ -289,3 +294,15 @@ func (s *SLocalStorage) onSaveToGlanceFailed(ctx context.Context, imageId string log.Errorln(err) } } + +func (s *SLocalStorage) CreateSnapshotFormUrl( + ctx context.Context, snapshotUrl, diskId, snapshotPath string, +) error { + remoteFile := remotefile.NewRemoteFile(ctx, snapshotUrl, snapshotPath, + false, "", -1, nil, "", "") + if remoteFile.Fetch() { + return nil + } else { + return fmt.Errorf("Fail to fetch snapshot from %s", snapshotUrl) + } +} diff --git a/pkg/util/cgrouputils/cgrouputils.go b/pkg/util/cgrouputils/cgrouputils.go index db112d2088..a437e7fe7c 100644 --- a/pkg/util/cgrouputils/cgrouputils.go +++ b/pkg/util/cgrouputils/cgrouputils.go @@ -11,6 +11,7 @@ import ( "strings" "yunion.io/x/log" + "yunion.io/x/onecloud/pkg/util/fileutils2" ) @@ -24,18 +25,35 @@ const ( ) var ( - module string cgroupsPath = getGroupPath() ) +type ICGroupTask interface { + SetPid(string) + SetWeight(coreNum int) + SetHand(hand ICGroupTask) + + GetStaticConfig() map[string]string + GetConfig() map[string]string + Module() string + RemoveTask() bool + SetTask() bool + + init() bool +} + type CGroupTask struct { pid string weight float64 + + hand ICGroupTask } -type ICGroupTask interface { - GetStaticConfig() map[string]string - GetConfig() map[string]string +func NewCGroupTask(pid string, coreNum int) *CGroupTask { + return &CGroupTask{ + pid: pid, + weight: float64(coreNum) / normalizeBase, + } } func getGroupPath() string { @@ -65,36 +83,36 @@ func ModuleIsMounted(module string) bool { return exec.Command("mountpoint", fullPath).Run() == nil } -func RootTaskPath() string { +func RootTaskPath(module string) string { return path.Join(cgroupsPath, module) } -func GetTaskParamPath(name, pid string) string { - spath := RootTaskPath() +func GetTaskParamPath(module, name, pid string) string { + spath := RootTaskPath(module) if len(pid) > 0 { spath = path.Join(spath, pid) } return path.Join(spath, name) } -func GetRootParam(name, pid string) (string, bool) { - param, err := fileutils2.FileGetContents(GetTaskParamPath(name, pid)) +func GetRootParam(module, name, pid string) string { + param, err := fileutils2.FileGetContents(GetTaskParamPath(module, name, pid)) if err != nil { log.Errorln(err) - return param, false + return "" } - return strings.TrimSpace(param), true + return strings.TrimSpace(param) } -func SetRootParam(name, value, pid string) bool { - if param, succ := GetRootParam(name, pid); !succ { - return false - } else if param != value { - fi, err := os.Open(GetTaskParamPath(name, pid)) +func SetRootParam(module, name, value, pid string) bool { + if param := GetRootParam(module, name, pid); param != value { + fi, err := os.Open(GetTaskParamPath(module, name, pid)) if err == nil { fi.Write([]byte(value)) err = fi.Sync() } + defer fi.Close() + if err != nil { if len(pid) == 0 { pid = "root" @@ -106,6 +124,54 @@ func SetRootParam(name, value, pid string) bool { return true } +func CleanupNonexistPids(module string) { + var root = RootTaskPath(module) + files, err := ioutil.ReadDir(root) + if err != nil { + log.Errorf("GetTaskIds failed: %s", err) + return + } + ids := []string{} + for _, file := range files { + ids = append(ids, file.Name()) + } + re := regexp.MustCompile(`^\d+$`) + for _, pid := range ids { + if re.MatchString(pid) && fileutils2.IsDir(path.Join(root, pid)) { + if !fileutils2.Exists(path.Join("/proc", pid)) { + log.Infof("Cgroup clenup %s", pid) + if err := os.Remove(path.Join(root, pid)); err != nil { + log.Errorf("CleanupNonexistPids pid=%s error: %s", pid, err) + } + } + } + } +} + +func (c *CGroupTask) SetWeight(coreNum int) { + c.weight = float64(coreNum) / normalizeBase +} + +func (c *CGroupTask) SetHand(hand ICGroupTask) { + c.hand = hand +} + +func (c *CGroupTask) SetPid(pid string) { + c.pid = pid +} + +func (c *CGroupTask) Module() string { + return "" +} + +func (c *CGroupTask) GetWeight() float64 { + if c.weight < maxWeight { + return c.weight + } else { + return maxWeight + } +} + func (c *CGroupTask) GetTaskIds() []string { files, err := ioutil.ReadDir(fmt.Sprintf("/proc/%s/task", c.pid)) if err != nil { @@ -120,7 +186,7 @@ func (c *CGroupTask) GetTaskIds() []string { } func (c *CGroupTask) TaskPath() string { - return path.Join(RootTaskPath(), c.pid) + return path.Join(RootTaskPath(c.hand.Module()), c.pid) } func (c *CGroupTask) taskIsExist() bool { @@ -135,6 +201,33 @@ func (c *CGroupTask) createTask() bool { return true } +func (c *CGroupTask) GetParam(name string) string { + return GetRootParam(c.hand.Module(), name, c.pid) +} + +func (c *CGroupTask) MoveTasksToRoot() { + procs := c.GetParam(CGROUP_TASKS) + if len(procs) > 0 { + for _, proc := range strings.Split(procs, "\n") { + proc = strings.TrimSpace(proc) + if len(proc) > 0 { + c.PushPid(proc, true) + } + } + } +} + +func (c *CGroupTask) RemoveTask() bool { + if c.taskIsExist() { + c.MoveTasksToRoot() + if err := os.RemoveAll(c.TaskPath()); err != nil { + log.Errorf("Remove task %s", err) + return false + } + } + return true +} + func (c *CGroupTask) GetStaticConfig() map[string]string { return nil } @@ -144,13 +237,13 @@ func (c *CGroupTask) GetConfig() map[string]string { } func (c *CGroupTask) SetParam(name, value string) bool { - return SetRootParam(name, value, c.pid) + return SetRootParam(c.hand.Module(), name, value, c.pid) } func (c *CGroupTask) SetParams(conf map[string]string) bool { for k, v := range conf { if !c.SetParam(k, v) { - log.Errorf("Fail to set %s/%s=%s for %s", module, k, v, c.pid) + log.Errorf("Fail to set %s/%s=%s for %s", c.hand.Module(), k, v, c.pid) return false } } @@ -158,18 +251,14 @@ func (c *CGroupTask) SetParams(conf map[string]string) bool { } func (c *CGroupTask) SetTask() bool { - return c.setTask(c) -} - -func (c *CGroupTask) setTask(t ICGroupTask) bool { if !c.taskIsExist() { if !c.createTask() { return false } } - conf := t.GetStaticConfig() + conf := c.hand.GetStaticConfig() if c.SetParams(conf) { - conf = t.GetConfig() + conf = c.hand.GetConfig() if c.SetParams(conf) { pids := c.GetTaskIds() if len(pids) > 0 { @@ -185,7 +274,25 @@ func (c *CGroupTask) setTask(t ICGroupTask) bool { func (c *CGroupTask) PushPid(pid string, isRoot bool) { subdir := fmt.Sprintf("/proc/%s/task/%s", c.pid, pid) - // if os.Stat(subdir) + if fi, err := os.Stat(subdir); err != nil { + log.Errorf("Fail to put pid in task %s", err) + return + } else if fi.Mode().IsDir() { + stat, err := fileutils2.FileGetContents(path.Join(subdir, "stat")) + if err != nil { + log.Errorf("Fail to put pid in task %s", err) + return + } + re := regexp.MustCompile(`\s+`) + data := re.Split(stat, -1) + if data[2] != "Z" { + if isRoot { + SetRootParam(c.hand.Module(), CGROUP_TASKS, pid, "") + } else { + c.SetParam(CGROUP_TASKS, pid) + } + } + } } func (c *CGroupTask) init() bool { @@ -240,6 +347,212 @@ func (c *CGroupTask) init() bool { return true } -func Init() { +/** + * CGroupCPUTask + */ +type CGroupCPUTask struct { + *CGroupTask +} + +const ( + CgroupsSharesWeight = 1024 + CPU_SHARES = "cpu.shares" +) + +func (c *CGroupCPUTask) Module() string { + return "cpu" +} + +func (c *CGroupCPUTask) GetConfig() map[string]string { + wt := int(CgroupsSharesWeight * c.GetWeight()) + return map[string]string{CPU_SHARES: fmt.Sprintf("%d", wt)} +} + +func (c *CGroupCPUTask) init() bool { + return SetRootParam(c.Module(), CPU_SHARES, + fmt.Sprintf("%d", CgroupsSharesWeight), "") +} + +func NewCGroupCPUTask(pid string, coreNum int) *CGroupCPUTask { + cgroup := &CGroupCPUTask{NewCGroupTask(pid, coreNum)} + cgroup.hand = cgroup + return cgroup +} + +/** + * CGroupIOTask + */ + +type CGroupIOTask struct { + *CGroupTask +} + +const ( + IoWeightBase = 100 + IoWeightMax = 1000 + IoWeightMin = 100 + BLOCK_IO_WEIGHT = "blkio.weight" +) + +func (c *CGroupIOTask) Module() string { + return "blkio" +} + +func (c *CGroupIOTask) GetConfig() map[string]string { + wt := int(c.GetWeight() * IoWeightBase) + if wt > IoWeightMax { + wt = IoWeightMax + } else if wt < IoWeightMin { + wt = IoWeightMin + } + return map[string]string{BLOCK_IO_WEIGHT: fmt.Sprintf("%d", wt)} +} + +func (c *CGroupIOTask) init() bool { + return SetRootParam(c.Module(), BLOCK_IO_WEIGHT, fmt.Sprintf("%d", IoWeightMax), "") +} + +func NewCGroupIOTask(pid string, coreNum int) *CGroupIOTask { + return &CGroupIOTask{NewCGroupTask(pid, coreNum)} +} + +/** + * CGroupIOHardlimitTask + */ + +type CGroupIOHardlimitTask struct { + *CGroupIOTask + + cpuNum int + params map[string]int + devId string +} + +func (c *CGroupIOHardlimitTask) GetConfig() map[string]string { + config := make(map[string]string, 0) + for k, v := range c.params { + if v != 0 { + config[k] = fmt.Sprintf("%s %d", c.devId, v*c.cpuNum) + } + } + return config +} + +func NewCGroupIOHardlimitTask(pid string, mem int, params map[string]int, devId string) *CGroupIOHardlimitTask { + return &CGroupIOHardlimitTask{ + CGroupIOTask: NewCGroupIOTask(pid, 0), + cpuNum: mem, + params: params, + devId: devId, + } +} + +/** + * CGroupMemoryTask + */ + +type CGroupMemoryTask struct { + *CGroupTask +} + +const ( + root_swappiness = 60 + vm_swappiness = 0 + MEMORY_SWAPPINESS = "memory.swappiness" +) + +func (c *CGroupMemoryTask) Module() string { + return "memory" +} + +func (c *CGroupMemoryTask) GetConfig() map[string]string { + return map[string]string{MEMORY_SWAPPINESS: fmt.Sprintf("%d", vm_swappiness)} +} + +func NewCGroupMemoryTask(pid string, coreNum int) *CGroupMemoryTask { + return &CGroupMemoryTask{ + CGroupTask: NewCGroupTask(pid, coreNum), + } +} + +/** + * CGroupCPUSetTask + */ + +type CGroupCPUSetTask struct { + *CGroupTask + + cpuset string +} + +const ( + CPUSET_CPUS = "cpuset.cpus" + CPUSET_MEMS = "cpuset.mems" +) + +func (c *CGroupCPUSetTask) Module() string { + return "cpuset" +} + +func (c *CGroupCPUSetTask) GetStaticConfig() map[string]string { + return map[string]string{CPUSET_MEMS: GetRootParam(c.Module(), CPUSET_MEMS, "")} +} + +func (c *CGroupCPUSetTask) GetConfig() map[string]string { + return map[string]string{CPUSET_CPUS: c.cpuset} +} + +func NewCGroupCPUSetTask(pid string, coreNum int, cpuset string) *CGroupCPUSetTask { + return &CGroupCPUSetTask{ + CGroupTask: NewCGroupTask(pid, coreNum), + cpuset: cpuset, + } +} + +func Init() bool { + for _, hand := range []ICGroupTask{&CGroupTask{}, &CGroupCPUTask{}, &CGroupIOTask{}} { + if !hand.init() { + return false + } + } + return true +} + +func CgroupSet(pid string, coreNum int) bool { + for _, hand := range []ICGroupTask{&CGroupCPUTask{}, &CGroupIOTask{}, &CGroupMemoryTask{}} { + hand.SetHand(hand) + hand.SetPid(pid) + hand.SetWeight(coreNum) + if !hand.SetTask() { + return false + } + } + return true +} + +func CgroupIoHardlimitSet( + pid string, coreNum int, + params map[string]int, devId string, +) bool { + cg := NewCGroupIOHardlimitTask(pid, coreNum, params, devId) + return cg.SetTask() +} + +func CgroupDestroy(pid string) bool { + for _, hand := range []ICGroupTask{&CGroupCPUTask{}, &CGroupIOTask{}, &CGroupMemoryTask{}, &CGroupCPUSetTask{}, &CGroupIOHardlimitTask{}} { + hand.SetHand(hand) + hand.SetPid(pid) + if !hand.RemoveTask() { + return false + } + } + return true +} + +func CgroupCleanAll() { + for _, hand := range []ICGroupTask{&CGroupCPUTask{}, &CGroupIOTask{}, &CGroupMemoryTask{}, &CGroupCPUSetTask{}, &CGroupIOHardlimitTask{}} { + hand.SetHand(hand) + CleanupNonexistPids(hand.Module()) + } } diff --git a/pkg/util/cgrouputils/cpusetutils.go b/pkg/util/cgrouputils/cpusetutils.go new file mode 100644 index 0000000000..0c5ae203e6 --- /dev/null +++ b/pkg/util/cgrouputils/cpusetutils.go @@ -0,0 +1,74 @@ +package cgrouputils + +import ( + "encoding/json" + "sync" + + "yunion.io/x/log" + "yunion.io/x/onecloud/pkg/util/fileutils2" +) + +const ( + utilHistoryFile = "/tmp/util.history" + MAX_HISTORY_UTIL_COUNT = 5 +) + +var ( + rebalanceProcessesLock = sync.Mutex{} + rebalanceProcessesRunning = false + + utilHistory map[string][]float64 +) + +func RebalanceProcesses(pids string) { + rebalanceProcessesLock.Lock() + defer rebalanceProcessesLock.Unlock() + + if rebalanceProcessesRunning { + return + } + + rebalanceProcesses(pids) +} + +func rebalanceProcesses(pids string) { + FetchHistoryUtil() + // ... + SaveHistoryUtil() +} + +func FetchHistoryUtil() map[string][]float64 { + if utilHistory == nil { + utilHistory = make(map[string][]float64) + if fileutils2.Exists(utilHistoryFile) { + contents, err := fileutils2.FileGetContents(utilHistoryFile) + if err != nil { + log.Errorf("FetchHistoryUtil error: %s", err) + return utilHistory + } + var objmap map[string]*json.RawMessage + if err := json.Unmarshal([]byte(contents), objmap); err != nil { + log.Errorf("FetchHistoryUtil error: %s", err) + return utilHistory + } + for k, v := range objmap { + var s = []float64{} + if err := json.Unmarshal(*v, &s); err != nil { + log.Errorf("FetchHistoryUtil error: %s", err) + break + } + utilHistory[k] = s + } + } + } + return utilHistory +} + +func SaveHistoryUtil() { + content, err := json.Marshal(utilHistory) + if err != nil { + log.Errorf("SaveHistoryUtil error: %s", err) + } else { + fileutils2.FilePutContents(utilHistoryFile, string(content), false) + } +} diff --git a/pkg/util/cgrouputils/cpuutils.go b/pkg/util/cgrouputils/cpuutils.go new file mode 100644 index 0000000000..e69de29bb2 diff --git a/pkg/util/fileutils2/fileutils.go b/pkg/util/fileutils2/fileutils.go index aea5a9a5e8..39efddb1e9 100644 --- a/pkg/util/fileutils2/fileutils.go +++ b/pkg/util/fileutils2/fileutils.go @@ -8,6 +8,7 @@ import ( "os" "os/exec" "path" + "path/filepath" "regexp" "strconv" "strings" @@ -589,3 +590,53 @@ func GetDevUuid(dev string) map[string]string { } return nil } + +func GetDevOfPath(spath string) string { + spath, err := filepath.Abs(spath) + if err != nil { + log.Errorln(err) + return "" + } + lines, err := exec.Command("mount").Output() + if err != nil { + log.Errorln(err) + return "" + } + var ( + maxMatchLen int + matchDev string + ) + + for _, line := range strings.Split(string(lines), "\n") { + segs := strings.Split(line, " ") + if len(segs) < 3 { + continue + } + if strings.HasPrefix(segs[0], "/dev/") { + if strings.HasPrefix(spath, segs[2]) { + matchLen := len(segs[2]) + if maxMatchLen < matchLen { + maxMatchLen = matchLen + matchDev = segs[0] + } + } + } + } + return matchDev +} + +func GetDevId(spath string) string { + dev := GetDevOfPath(spath) + if len(dev) == 0 { + return "" + } + devInfo, err := exec.Command("ls", "-l", dev).Output() + if err != nil { + log.Errorln(err) + return "" + } + devInfos := strings.Split(string(devInfo), "\n") + data := strings.Split(string(devInfos[0]), " ") + data[4] = data[4][:len(data[4])-1] + return strings.Join(data, ":") +} diff --git a/pkg/util/fileutils2/fileutils_test.go b/pkg/util/fileutils2/fileutils_test.go index 63065fee93..329185791d 100644 --- a/pkg/util/fileutils2/fileutils_test.go +++ b/pkg/util/fileutils2/fileutils_test.go @@ -1,6 +1,8 @@ package fileutils2 -import "testing" +import ( + "testing" +) func TestIsBlockDeviceUsed(t *testing.T) { type args struct { @@ -30,3 +32,23 @@ func TestIsBlockDeviceUsed(t *testing.T) { }) } } + +func TestGetDevId(t *testing.T) { + type args struct { + spath string + } + tests := []struct { + name string + args args + want string + }{ + // TODO: Add test cases. + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := GetDevId(tt.args.spath); got != tt.want { + t.Errorf("GetDevId() = %v, want %v", got, tt.want) + } + }) + } +} diff --git a/pkg/util/ovsutils/ovsutils.go b/pkg/util/ovsutils/ovsutils.go new file mode 100644 index 0000000000..26da622e21 --- /dev/null +++ b/pkg/util/ovsutils/ovsutils.go @@ -0,0 +1,90 @@ +package ovsutils + +import ( + "os/exec" + "regexp" + "strings" + + "yunion.io/x/log" + "yunion.io/x/onecloud/pkg/util/regutils2" + "yunion.io/x/pkg/utils" +) + +func GetDbPorts(brname string) []string { + output, err := exec.Command("ovs-vsctl", "list-ifaces", brname).Output() + if err != nil { + log.Errorln(err) + return nil + } + ifaces := make([]string, 0) + re := regexp.MustCompile(`^[a-zA-Z0-9._@-]+$`) + for _, line := range strings.Split(string(output), "\n") { + ifname := strings.TrimSpace(line) + if len(ifname) > 0 && re.MatchString(ifname) { + ifaces = append(ifaces, ifname) + } + } + return ifaces +} + +func GetDpPorts(brname string) []string { + output, err := exec.Command("ovs-dpctl", "show").Output() + if err != nil { + log.Errorln(err) + return nil + } + + ifaces := make([]string, 0) + re := regexp.MustCompile(`port \d+: (?P[a-zA-Z0-9._@-]+)`) + for _, line := range strings.Split(string(output), "\n") { + m := regutils2.GetParams(re, line) + if len(m) > 0 { + ifnmae := m["name"] + if ifnmae != brname { + ifaces = append(ifaces, ifnmae) + } + } + } + return ifaces +} + +func GetBridges() []string { + output, err := exec.Command("ovs_vsctl", "list-br").Output() + if err != nil { + log.Errorln(err) + return nil + } + + brs := make([]string, 0) + for _, line := range strings.Split(string(output), "\n") { + brname := strings.TrimSpace(line) + if len(brname) > 0 { + brs = append(brs, brname) + } + } + return brs +} + +func RemovePortFromBridge(brname, port string) { + log.Infof("remove_port_from_bridge %s %s", brname, port) + if err := exec.Command("ovs-vsctl", "del-port", brname, port).Run(); err != nil { + log.Errorln(err) + } +} + +func CleanHiddenPorts(brname string) { + dbPorts := GetDbPorts(brname) + dpPorts := GetDpPorts(brname) + for _, p := range dbPorts { + if !utils.IsInStringArray(p, dpPorts) { + RemovePortFromBridge(brname, p) + } + } +} + +func CleanAllHiddenPorts() { + brs := GetBridges() + for _, br := range brs { + CleanHiddenPorts(br) + } +} diff --git a/pkg/util/qemuimg/qemuimg.go b/pkg/util/qemuimg/qemuimg.go index 89419f0f23..57de61f0f6 100644 --- a/pkg/util/qemuimg/qemuimg.go +++ b/pkg/util/qemuimg/qemuimg.go @@ -230,6 +230,17 @@ func (img *SQemuImage) convert(format TImageFormat, options []string, compact bo return img.parse() } +func (img *SQemuImage) convertTo( + format TImageFormat, options []string, compact bool, password string, output string, +) error { + err := img.doConvert(output, format, options, compact, password) + if err != nil { + return err + } + img.Password = password + return img.parse() +} + func (img *SQemuImage) Copy(name string) (*SQemuImage, error) { if !img.IsValid() { return nil, fmt.Errorf("self is not valid") @@ -244,6 +255,18 @@ func (img *SQemuImage) Copy(name string) (*SQemuImage, error) { return NewQemuImage(name) } +func (img *SQemuImage) Convert2Qcow2To(output string, compact bool) error { + options := make([]string, 0) + // if len(backPath) > 0 { + // options = append(options, fmt.Sprintf("backing_file=%s", backPath)) + //} else + if !compact { + sparseOpts := qcow2SparseOptions() + options = append(options, sparseOpts...) + } + return img.convertTo(QCOW2, options, compact, "", output) +} + func (img *SQemuImage) Convert2Qcow2(compact bool) error { options := make([]string, 0) // if len(backPath) > 0 { diff --git a/pkg/util/regutils2/regutils.go b/pkg/util/regutils2/regutils.go index 4b474e055f..de7ceacee5 100644 --- a/pkg/util/regutils2/regutils.go +++ b/pkg/util/regutils2/regutils.go @@ -20,15 +20,5 @@ func GetParams(compRegEx *regexp.Regexp, val string) map[string]string { func SubGroupMatch(pattern string, line string) map[string]string { regEx := regexp.MustCompile(pattern) - params := make(map[string]string) - matches := regEx.FindStringSubmatch(line) - if len(matches) == 0 { - return params - } - for i, name := range regEx.SubexpNames() { - if i > 0 && i <= len(matches) { - params[name] = matches[i] - } - } - return params + return GetParams(regEx, line) }