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

fix(host): add progress for migrate (#13101)

Browse Source
This commit is contained in:
屈轩
2021-12-31 06:09:16 +08:00
committed by GitHub
parent 71c646128a
commit 9f92441349
14 changed files with 40 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/hostman/guestman/guestman.go
+3 -1
View File
@@ -807,7 +807,7 @@ func (m *SGuestManager) DestPrepareMigrate(ctx context.Context, params interface
err := iStorage.DestinationPrepareMigrate(
ctx, migParams.LiveMigrate, migParams.DisksUri, migParams.SnapshotsUri,
migParams.DisksBackingFile, migParams.SrcSnapshots, migParams.RebaseDisks, disks[i],
migParams.DisksBackingFile, migParams.SrcSnapshots, migParams.RebaseDisks, disks[i], migParams.Sid, i+1, len(disks),
)
if err != nil {
return nil, fmt.Errorf("dest prepare migrate failed %s", err)
@@ -825,6 +825,8 @@ func (m *SGuestManager) DestPrepareMigrate(ctx context.Context, params interface
startParams.Set("qemu_version", jsonutils.NewString(migParams.QemuVersion))
startParams.Set("need_migrate", jsonutils.JSONTrue)
hostutils.DelayTaskWithoutReqctx(ctx, guest.asyncScriptStart, startParams)
} else {
hostutils.UpdateServerProgress(context.Background(), migParams.Sid, 100.0, 0)
}
return nil, nil
pkg/hostman/storageman/disk_base.go
+2 -2
View File
@@ -51,7 +51,7 @@ type IDisk interface {
CleanupSnapshots(ctx context.Context, params interface{}) (jsonutils.JSONObject, error)
PrepareMigrate(liveMigrate bool) (string, error)
CreateFromUrl(ctx context.Context, url string, size int64) error
CreateFromUrl(ctx context.Context, url string, size int64, callback func(progress, progressMbps float64, totalSizeMb int64)) error
CreateFromTemplate(context.Context, string, string, int64) (jsonutils.JSONObject, error)
CreateFromSnapshotLocation(ctx context.Context, location string, size int64) error
CreateFromRbdSnapshot(ctx context.Context, snapshotId, srcDiskId, srcPool string) error
@@ -97,7 +97,7 @@ func (d *SBaseDisk) OnRebuildRoot(ctx context.Context, params api.DiskAllocateIn
return fmt.Errorf("Not implemented")
}
func (d *SBaseDisk) CreateFromUrl(ctx context.Context, url string, size int64) error {
func (d *SBaseDisk) CreateFromUrl(ctx context.Context, url string, size int64, callback func(progress, progressMbps float64, totalSizeMb int64)) error {
return fmt.Errorf("Not implemented")
}
pkg/hostman/storageman/disk_local.go
+2 -2
View File
@@ -248,9 +248,9 @@ func (d *SLocalDisk) createFromTemplate(
return d.GetDiskDesc(), nil
}
func (d *SLocalDisk) CreateFromUrl(ctx context.Context, url string, size int64) error {
func (d *SLocalDisk) CreateFromUrl(ctx context.Context, url string, size int64, callback func(progress, progressMbps float64, totalSizeMb int64)) error {
remoteFile := remotefile.NewRemoteFile(ctx, url, d.getPath(), false, "", -1, nil, "", "")
err := remoteFile.Fetch(nil)
err := remoteFile.Fetch(callback)
if err != nil {
return errors.Wrapf(err, "fetch image from %s", url)
}
pkg/hostman/storageman/disk_rbd.go
-3
View File
@@ -12,9 +12,6 @@
// See the License for the specific language governing permissions and
// limitations under the License.
//go:build linux && cgo
// +build linux,cgo
package storageman
import (
pkg/hostman/storageman/imagecache_base.go
+1 -1
View File
@@ -25,7 +25,7 @@ type IImageCache interface {
GetPath() string
GetName() string
Load() error
Acquire(ctx context.Context, input api.CacheImageInput, callback func(progress float32)) error
Acquire(ctx context.Context, input api.CacheImageInput, callback func(progress, progressMbps float64, totalSizeMb int64)) error
Release()
Remove(ctx context.Context) error
GetImageId() string
pkg/hostman/storageman/imagecache_local.go
+2 -2
View File
@@ -149,7 +149,7 @@ func (l *SLocalImageCache) Release() {
l.consumerCount -= 1
}
func (l *SLocalImageCache) Acquire(ctx context.Context, input api.CacheImageInput, callback func(progress float32)) error {
func (l *SLocalImageCache) Acquire(ctx context.Context, input api.CacheImageInput, callback func(progress, progressMbps float64, totalSizeMb int64)) error {
isOk, err := l.prepare(ctx, input)
if err != nil {
return errors.Wrapf(err, "prepare")
@@ -187,7 +187,7 @@ func (l *SLocalImageCache) prepare(ctx context.Context, input api.CacheImageInpu
return false, nil
}
func (l *SLocalImageCache) fetch(ctx context.Context, input api.CacheImageInput, callback func(progress float32)) error {
func (l *SLocalImageCache) fetch(ctx context.Context, input api.CacheImageInput, callback func(progress, progressMbps float64, totalSizeMb int64)) error {
var _fetch = func() error {
if len(l.Manager.GetId()) > 0 {
_, err := hostutils.RemoteStoragecacheCacheImage(ctx,
pkg/hostman/storageman/imagecache_rbd.go
+3 -5
View File
@@ -12,8 +12,6 @@
// See the License for the specific language governing permissions and
// limitations under the License.
// +build linux,cgo
package storageman
import (
@@ -73,11 +71,11 @@ func (r *SRbdImageCache) Load() error {
return fmt.Errorf("invalid rbd image %s at host %s", origin.String(), options.HostOptions.Hostname)
}
func (r *SRbdImageCache) Acquire(ctx context.Context, input api.CacheImageInput, callback func(progress float32)) error {
func (r *SRbdImageCache) Acquire(ctx context.Context, input api.CacheImageInput, callback func(progress, progressMbps float64, totalSizeMb int64)) error {
input.ImageId = r.imageId
localImageCache, err := storageManager.LocalStorageImagecacheManager.AcquireImage(ctx, input, func(percent float32) {
localImageCache, err := storageManager.LocalStorageImagecacheManager.AcquireImage(ctx, input, func(progress, progressMbps float64, totalSizeMb int64) {
if len(input.ServerId) > 0 {
modules.Servers.Update(hostutils.GetComputeSession(context.Background()), input.ServerId, jsonutils.Marshal(map[string]float32{"progress": percent / 1.2}))
hostutils.UpdateServerProgress(context.Background(), input.ServerId, progress/1.2, progressMbps)
}
})
if err != nil {
pkg/hostman/storageman/imagecachemanager_base.go
+1 -1
View File
@@ -53,7 +53,7 @@ type IImageCacheManger interface {
PrefetchImageCache(ctx context.Context, data interface{}) (jsonutils.JSONObject, error)
DeleteImageCache(ctx context.Context, data interface{}) (jsonutils.JSONObject, error)
AcquireImage(ctx context.Context, input api.CacheImageInput, callback func(progress float32)) (IImageCache, error)
AcquireImage(ctx context.Context, input api.CacheImageInput, callback func(progress, progressMbps float64, totalSizeMb int64)) (IImageCache, error)
ReleaseImage(ctx context.Context, imageId string)
LoadImageCache(imageId string)
}
pkg/hostman/storageman/imagecachemanager_local.go
+3 -4
View File
@@ -28,7 +28,6 @@ import (
"yunion.io/x/onecloud/pkg/cloudcommon/db/lockman"
"yunion.io/x/onecloud/pkg/hostman/hostutils"
"yunion.io/x/onecloud/pkg/httperrors"
modules "yunion.io/x/onecloud/pkg/mcclient/modules/compute"
"yunion.io/x/onecloud/pkg/util/fileutils2"
"yunion.io/x/onecloud/pkg/util/procutils"
)
@@ -77,7 +76,7 @@ func (c *SLocalImageCacheManager) LoadImageCache(imageId string) {
}
}
func (c *SLocalImageCacheManager) AcquireImage(ctx context.Context, input api.CacheImageInput, callback func(progress float32)) (IImageCache, error) {
func (c *SLocalImageCacheManager) AcquireImage(ctx context.Context, input api.CacheImageInput, callback func(progress, progressMbps float64, totalSizeMb int64)) (IImageCache, error) {
c.lock.LockRawObject(ctx, "image-cache", input.ImageId)
defer c.lock.ReleaseRawObject(ctx, "image-cache", input.ImageId)
@@ -87,9 +86,9 @@ func (c *SLocalImageCacheManager) AcquireImage(ctx context.Context, input api.Ca
c.cachedImages[input.ImageId] = img
}
if callback == nil && len(input.ServerId) > 0 {
callback = func(progress float32) {
callback = func(progress, progressMbps float64, totalSizeMb int64) {
if len(input.ServerId) > 0 {
modules.Servers.Update(hostutils.GetComputeSession(context.Background()), input.ServerId, jsonutils.Marshal(map[string]float32{"progress": progress}))
hostutils.UpdateServerProgress(context.Background(), input.ServerId, progress, progressMbps)
}
}
}
pkg/hostman/storageman/imagecachemanager_rbd.go
+1 -3
View File
@@ -12,8 +12,6 @@
// See the License for the specific language governing permissions and
// limitations under the License.
// +build linux,cgo
package storageman
import (
@@ -158,7 +156,7 @@ func (c *SRbdImageCacheManager) removeImage(ctx context.Context, imageId string)
return nil
}
func (c *SRbdImageCacheManager) AcquireImage(ctx context.Context, input api.CacheImageInput, callback func(float32)) (IImageCache, error) {
func (c *SRbdImageCacheManager) AcquireImage(ctx context.Context, input api.CacheImageInput, callback func(float64, float64, int64)) (IImageCache, error) {
lockman.LockRawObject(ctx, "image-cache", input.ImageId)
defer lockman.ReleaseRawObject(ctx, "image-cache", input.ImageId)
pkg/hostman/storageman/remotefile/remotefile.go
+9 -7
View File
@@ -83,7 +83,7 @@ func NewRemoteFile(
}
}
func (r *SRemoteFile) Fetch(callback func(progress float32)) error {
func (r *SRemoteFile) Fetch(callback func(progress, progressMbps float64, totalSizeMb int64)) error {
if len(r.preChksum) > 0 {
log.Infof("Fetch remote file with precheck sum: %s", r.preChksum)
return r.fetch(r.preChksum, callback)
@@ -115,7 +115,7 @@ func (r *SRemoteFile) GetInfo() (*SImageDesc, error) {
}, nil
}
func (r *SRemoteFile) VerifyIntegrity(callback func(progress float32)) error {
func (r *SRemoteFile) VerifyIntegrity(callback func(progress, progressMbps float64, totalSizeMb int64)) error {
localChksum, err := fileutils2.MD5(r.localPath)
if err != nil {
return errors.Wrapf(err, "fileutils2.MD5(%s)", r.localPath)
@@ -133,7 +133,7 @@ func (r *SRemoteFile) VerifyIntegrity(callback func(progress float32)) error {
return r.fetch("", callback)
}
func (r *SRemoteFile) fetch(preChksum string, callback func(progress float32)) error {
func (r *SRemoteFile) fetch(preChksum string, callback func(progress, progressMbps float64, totalSizeMb int64)) error {
var err error
for i := 0; i < 3; i++ {
err = r.download(true, preChksum, callback)
@@ -158,7 +158,7 @@ func (r *SRemoteFile) fetch(preChksum string, callback func(progress float32)) e
}
// retry download
func (r *SRemoteFile) download(getData bool, preChksum string, callback func(progress float32)) error {
func (r *SRemoteFile) download(getData bool, preChksum string, callback func(progress, progressMbps float64, totalSizeMb int64)) error {
if getData {
// fetch image headers and set resource properties
err := r.downloadInternal(false, preChksum, callback)
@@ -182,7 +182,7 @@ func (r *SRemoteFile) download(getData bool, preChksum string, callback func(pro
return r.downloadInternal(getData, preChksum, callback)
}
func (r *SRemoteFile) downloadInternal(getData bool, preChksum string, callback func(progress float32)) error {
func (r *SRemoteFile) downloadInternal(getData bool, preChksum string, callback func(progress, progressMbps float64, totalSizeMb int64)) error {
var header = http.Header{}
header.Set("X-Auth-Token", auth.GetTokenString())
if len(preChksum) > 0 {
@@ -234,6 +234,7 @@ func (r *SRemoteFile) downloadInternal(getData bool, preChksum string, callback
var finishChan = make(chan struct{})
go func() {
defer recover()
preSizeMb := int64(0)
for {
select {
case <-time.After(1 * time.Second):
@@ -249,11 +250,12 @@ func (r *SRemoteFile) downloadInternal(getData bool, preChksum string, callback
}
log.Infof("written file %s size %dM%s", r.tmpPath, info.Size()/1024/1024, percentInfo)
if callback != nil && percent > 0 {
callback(float32(percent))
callback(percent, float64(info.Size()-preSizeMb)/1024/1024, totalSize/1024/1024)
}
preSizeMb = info.Size()
case <-finishChan:
if callback != nil {
callback(100)
callback(100, 0, totalSize/1024/1024)
}
return
}
pkg/hostman/storageman/storage_base.go
+2 -2
View File
@@ -130,7 +130,7 @@ type IStorage interface {
GetImgsaveBackupPath() string
DestinationPrepareMigrate(ctx context.Context, liveMigrate bool, disksUri string, snapshotsUri string,
disksBackingFile, srcSnapshots jsonutils.JSONObject, rebaseDisks bool, diskDesc jsonutils.JSONObject) error
disksBackingFile, srcSnapshots jsonutils.JSONObject, rebaseDisks bool, diskDesc jsonutils.JSONObject, serverId string, idx, totalDiskCount int) error
Accessible() error
Detach() error
@@ -365,7 +365,7 @@ func (s *SBaseStorage) CreateDiskFromSnpashot(ctx context.Context, disk IDisk, i
func (s *SBaseStorage) DestinationPrepareMigrate(
ctx context.Context, liveMigrate bool, disksUri string, snapshotsUri string,
disksBackingFile, srcSnapshots jsonutils.JSONObject, rebaseDisks bool, diskinfo jsonutils.JSONObject,
disksBackingFile, srcSnapshots jsonutils.JSONObject, rebaseDisks bool, diskinfo jsonutils.JSONObject, serverId string, idx, totalDiskCount int,
) error {
return nil
}
pkg/hostman/storageman/storage_local.go
+11 -5
View File
@@ -399,7 +399,7 @@ func (s *SLocalStorage) DeleteSnapshots(ctx context.Context, params interface{})
func (s *SLocalStorage) DestinationPrepareMigrate(
ctx context.Context, liveMigrate bool, disksUri string, snapshotsUri string,
disksBackingFile, srcSnapshots jsonutils.JSONObject, rebaseDisks bool, diskinfo jsonutils.JSONObject) error {
disksBackingFile, srcSnapshots jsonutils.JSONObject, rebaseDisks bool, diskinfo jsonutils.JSONObject, serverId string, idx, totalDiskCount int) error {
var (
diskId, _ = diskinfo.GetString("disk_id")
snapshots, _ = srcSnapshots.GetArray(diskId)
@@ -459,10 +459,16 @@ func (s *SLocalStorage) DestinationPrepareMigrate(
} else {
// download disk form remote url
diskUrl := fmt.Sprintf("%s/%s/%s", disksUri, diskStorageId, diskId)
if err := disk.CreateFromUrl(ctx, diskUrl, 0); err != nil {
log.Errorf("CreateFromUrl %q: %v", diskUrl, err)
err = errors.Wrap(err, "CreateFromUrl")
return err
err := disk.CreateFromUrl(ctx, diskUrl, 0, func(progress, progressMbps float64, totalSizeMb int64) {
log.Debugf("[%.2f / %d] disk %s create %.2f with speed %.2fMbps", progress*float64(totalSizeMb)/100, totalSizeMb, disk.GetId(), progress, progressMbps)
newProgress := float64(idx-1)/float64(totalDiskCount)*100.0 + 1/float64(totalDiskCount)*progress
if len(serverId) > 0 {
log.Debugf("server %s migrate %.2f with speed %.2fMbps", serverId, newProgress, progressMbps)
hostutils.UpdateServerProgress(context.Background(), serverId, newProgress, progressMbps)
}
})
if err != nil {
return errors.Wrap(err, "CreateFromUrl")
}
}
if rebaseDisks && len(templateId) > 0 && len(baseImagePath) == 0 {
pkg/hostman/storageman/storage_rbd.go
-3
View File
@@ -12,9 +12,6 @@
// See the License for the specific language governing permissions and
// limitations under the License.
//go:build linux && cgo
// +build linux,cgo
package storageman
import (