From 3a6e7f55ab06d2005eeaf2c6364c59065b7e7348 Mon Sep 17 00:00:00 2001 From: Zexi Li Date: Tue, 27 Apr 2021 11:31:52 +0800 Subject: [PATCH] fix(region): download cached image from source host before migrating --- pkg/compute/hostdrivers/kvm.go | 59 ++++++++++--------- pkg/compute/models/cachedimages.go | 7 +++ pkg/compute/models/storagecaches.go | 8 +++ pkg/compute/options/options.go | 1 - pkg/compute/tasks/guest_live_migrate_task.go | 4 +- pkg/hostman/downloader/downloadhandler.go | 15 +++++ .../downloader/imagecache_downloader.go | 17 ++++++ pkg/hostman/storageman/disk_local.go | 2 +- pkg/hostman/storageman/disk_rbd.go | 2 +- pkg/hostman/storageman/imagecache_base.go | 2 +- pkg/hostman/storageman/imagecache_local.go | 8 +-- pkg/hostman/storageman/imagecache_rbd.go | 4 +- .../storageman/imagecachemanager_base.go | 2 +- .../storageman/imagecachemanager_local.go | 7 ++- .../storageman/imagecachemanager_rbd.go | 7 ++- .../storageman/remotefile/remotefile.go | 24 +++++--- 16 files changed, 112 insertions(+), 57 deletions(-) diff --git a/pkg/compute/hostdrivers/kvm.go b/pkg/compute/hostdrivers/kvm.go index 29f3523257..9089f1985a 100644 --- a/pkg/compute/hostdrivers/kvm.go +++ b/pkg/compute/hostdrivers/kvm.go @@ -28,7 +28,6 @@ import ( "yunion.io/x/pkg/errors" "yunion.io/x/pkg/utils" - "yunion.io/x/onecloud/pkg/apis" api "yunion.io/x/onecloud/pkg/apis/compute" "yunion.io/x/onecloud/pkg/cloudcommon/cmdline" "yunion.io/x/onecloud/pkg/cloudcommon/db" @@ -39,7 +38,6 @@ import ( "yunion.io/x/onecloud/pkg/compute/options" "yunion.io/x/onecloud/pkg/httperrors" "yunion.io/x/onecloud/pkg/mcclient" - "yunion.io/x/onecloud/pkg/mcclient/auth" "yunion.io/x/onecloud/pkg/util/httputils" "yunion.io/x/onecloud/pkg/util/k8s/tokens" ) @@ -164,6 +162,15 @@ func (self *SKVMHostDriver) CheckAndSetCacheImage(ctx context.Context, host *mod format, _ := params.GetString("format") isForce := jsonutils.QueryBoolean(params, "is_force", false) + srcHostId, _ := params.GetString("source_host_id") + var srcHost *models.SHost + if srcHostId != "" { + srcHost = models.HostManager.FetchHostById(srcHostId) + if srcHost == nil { + return errors.Errorf("Source host %s not found", srcHostId) + } + } + type contentStruct struct { ImageId string Format string @@ -176,37 +183,31 @@ func (self *SKVMHostDriver) CheckAndSetCacheImage(ctx context.Context, host *mod content.ImageId = imageId content.Format = format - if options.Options.ImageCacheFromHost { - obj, err := models.CachedimageManager.FetchById(imageId) - if err != nil { - return errors.Wrapf(err, "Fetch cached image by image_id %s", imageId) - } - cacheImage := obj.(*models.SCachedimage) - srcHostCacheImage, err := cacheImage.ChooseSourceStoragecacheInRange(api.HOST_TYPE_HYPERVISOR, []string{host.Id}, []interface{}{host.GetZone()}) + obj, err := models.CachedimageManager.FetchById(imageId) + if err != nil { + return errors.Wrapf(err, "Fetch cached image by image_id %s", imageId) + } + cacheImage := obj.(*models.SCachedimage) + rangeObjs := []interface{}{host.GetZone()} + if srcHost != nil { + rangeObjs = append(rangeObjs, srcHost) + } + srcHostCacheImage, err := cacheImage.ChooseSourceStoragecacheInRange(api.HOST_TYPE_HYPERVISOR, []string{host.Id}, rangeObjs) + if err != nil { + return errors.Wrapf(err, "Choose source storagecache") + } + if srcHostCacheImage != nil { + err = srcHostCacheImage.AddDownloadRefcount() if err != nil { return err } - if srcHostCacheImage != nil { - err = srcHostCacheImage.AddDownloadRefcount() - if err != nil { - return err - } - srcHost, err := srcHostCacheImage.GetHost() - if err != nil { - return err - } - content.SrcUrl = fmt.Sprintf("%s/download/images/%s", srcHost.ManagerUri, imageId) - } - } else { - // from glance service - glanceURL, err := auth.GetServiceURL(apis.SERVICE_TYPE_IMAGE, "", host.GetZone().GetName(), "") + + srcHost, err := srcHostCacheImage.GetHost() if err != nil { - return errors.Wrapf(err, "Get %s service url", apis.SERVICE_TYPE_IMAGE) - } - content.SrcUrl = fmt.Sprintf("%s/images/%s", glanceURL, imageId) - if content.Format != "" { - content.SrcUrl = fmt.Sprintf("%s?format=%s", content.SrcUrl, content.Format) + return errors.Wrapf(err, "Get storage cached image %s host", srcHostCacheImage.GetId()) } + content.SrcUrl = fmt.Sprintf("%s/download/images/%s", srcHost.ManagerUri, imageId) + } url := fmt.Sprintf("%s/disks/image_cache", host.ManagerUri) @@ -222,7 +223,7 @@ func (self *SKVMHostDriver) CheckAndSetCacheImage(ctx context.Context, host *mod _, _, err = httputils.JSONRequest(httputils.GetDefaultClient(), ctx, "POST", url, header, body, false) if err != nil { - return err + return errors.Wrapf(err, "POST %s", url) } return nil } diff --git a/pkg/compute/models/cachedimages.go b/pkg/compute/models/cachedimages.go index 52573b06bc..81f66fa1b8 100644 --- a/pkg/compute/models/cachedimages.go +++ b/pkg/compute/models/cachedimages.go @@ -170,6 +170,11 @@ func (self *SCachedimage) GetHypervisor() string { return osType } +func (self *SCachedimage) GetChecksum() string { + checksum, _ := self.Info.GetString("checksum") + return checksum +} + func (self *SCachedimage) getStoragecacheQuery() *sqlchemy.SQuery { q := StoragecachedimageManager.Query().Equals("cachedimage_id", self.Id) return q @@ -450,6 +455,8 @@ func (self *SCachedimage) ChooseSourceStoragecacheInRange(hostType string, exclu for _, rangeObj := range rangeObjs { switch v := rangeObj.(type) { + case *SHost: + q = q.Filter(sqlchemy.Equals(host.Field("id"), v.Id)) case *SZone: q = q.Filter(sqlchemy.Equals(host.Field("zone_id"), v.Id)) case *SCloudprovider: diff --git a/pkg/compute/models/storagecaches.go b/pkg/compute/models/storagecaches.go index 26c44deb84..96e1b31fa6 100644 --- a/pkg/compute/models/storagecaches.go +++ b/pkg/compute/models/storagecaches.go @@ -390,6 +390,10 @@ func (self *SStoragecache) getCachedImageSize() int64 { } func (self *SStoragecache) StartImageCacheTask(ctx context.Context, userCred mcclient.TokenCredential, imageId string, format string, isForce bool, parentTaskId string) error { + return self.StartImageCacheTaskFromHost(ctx, userCred, imageId, format, isForce, "", parentTaskId) +} + +func (self *SStoragecache) StartImageCacheTaskFromHost(ctx context.Context, userCred mcclient.TokenCredential, imageId string, format string, isForce bool, srcHostId string, parentTaskId string) error { StoragecachedimageManager.Register(ctx, userCred, self.Id, imageId, "") data := jsonutils.NewDict() data.Add(jsonutils.NewString(imageId), "image_id") @@ -413,6 +417,10 @@ func (self *SStoragecache) StartImageCacheTask(ctx context.Context, userCred mcc if isForce { data.Add(jsonutils.JSONTrue, "is_force") } + + if srcHostId != "" { + data.Add(jsonutils.NewString(srcHostId), "source_host_id") + } task, err := taskman.TaskManager.NewTask(ctx, "StorageCacheImageTask", self, userCred, data, parentTaskId, "", nil) if err != nil { log.Errorf("create StorageCacheImageTask fail %s", err) diff --git a/pkg/compute/options/options.go b/pkg/compute/options/options.go index 565de73ddb..68333837ef 100644 --- a/pkg/compute/options/options.go +++ b/pkg/compute/options/options.go @@ -50,7 +50,6 @@ type ComputeOptions struct { LoadbalancerPendingDeleteCheckInterval int `default:"3600" help:"Interval between checks of pending deleted loadbalancer objects, defaults to 1h"` ImageCacheStoragePolicy string `default:"least_used" choices:"best_fit|least_used" help:"Policy to choose storage for image cache, best_fit or least_used"` - ImageCacheFromHost bool `default:"false" help:"Download cached image from host"` MetricsRetentionDays int32 `default:"30" help:"Retention days for monitoring metrics in influxdb"` DefaultBandwidth int `default:"1000" help:"Default bandwidth"` diff --git a/pkg/compute/tasks/guest_live_migrate_task.go b/pkg/compute/tasks/guest_live_migrate_task.go index f89363fb65..2d78896d92 100644 --- a/pkg/compute/tasks/guest_live_migrate_task.go +++ b/pkg/compute/tasks/guest_live_migrate_task.go @@ -138,8 +138,8 @@ func (self *GuestMigrateTask) SaveScheduleResult(ctx context.Context, obj ISched if len(disk.TemplateId) > 0 && isLocalStorage { targetStorageCache := targetHost.GetLocalStoragecache() if targetStorageCache != nil { - err := targetStorageCache.StartImageCacheTask( - ctx, self.UserCred, disk.TemplateId, disk.DiskFormat, false, self.GetTaskId()) + err := targetStorageCache.StartImageCacheTaskFromHost( + ctx, self.UserCred, disk.TemplateId, disk.DiskFormat, false, guest.HostId, self.GetTaskId()) if err != nil { self.TaskFailed(ctx, guest, jsonutils.NewString(err.Error())) } diff --git a/pkg/hostman/downloader/downloadhandler.go b/pkg/hostman/downloader/downloadhandler.go index 0f34a10798..001b1d4383 100644 --- a/pkg/hostman/downloader/downloadhandler.go +++ b/pkg/hostman/downloader/downloadhandler.go @@ -60,6 +60,8 @@ func AddDownloadHandler(prefix string, app *appsrv.Application) { app.AddHandler("HEAD", fmt.Sprintf("%s/%s/snapshots///", prefix, kerword), auth.Authenticate(snapshotHead)) + app.AddHandler("HEAD", + fmt.Sprintf("%s/%s/images/", prefix, kerword), auth.Authenticate(imageCacheHead)) } } @@ -198,3 +200,16 @@ func snapshotHead(ctx context.Context, w http.ResponseWriter, r *http.Request) { } } } + +func imageCacheHead(ctx context.Context, w http.ResponseWriter, r *http.Request) { + params, _, _ := appsrv.FetchEnv(ctx, w, r) + imageId := params[""] + rateLimit := options.HostOptions.BandwidthLimit + compress := isCompress(r) + + hand := NewImageCacheDownloadProvider(w, compress, rateLimit, imageId) + + if err := hand.HandlerHead(); err != nil { + hostutils.Response(ctx, w, err) + } +} diff --git a/pkg/hostman/downloader/imagecache_downloader.go b/pkg/hostman/downloader/imagecache_downloader.go index 3a534a41e0..576c1d0719 100644 --- a/pkg/hostman/downloader/imagecache_downloader.go +++ b/pkg/hostman/downloader/imagecache_downloader.go @@ -18,7 +18,10 @@ import ( "net/http" "path" + "yunion.io/x/pkg/errors" + "yunion.io/x/onecloud/pkg/hostman/storageman" + "yunion.io/x/onecloud/pkg/util/fileutils2" ) type SImageCacheDownloadProvider struct { @@ -44,6 +47,20 @@ func (s *SImageCacheDownloadProvider) downloadFilePath() string { storageman.GetManager().LocalStorageImagecacheManager.GetPath(), s.imageId) } +func (s *SImageCacheDownloadProvider) HandlerHead() error { + headers := s.getHeaders() + checksum, err := fileutils2.MD5(s.downloadFilePath()) + if err != nil { + return errors.Wrapf(err, "MD5SUM %s", s.downloadFilePath()) + } + headers.Set("X-Image-Meta-Checksum", checksum) + for k := range headers { + s.w.Header().Add(k, headers.Get(k)) + } + s.w.WriteHeader(200) + return nil +} + func (s *SImageCacheDownloadProvider) Start() error { return s.SDownloadProvider.Start(nil, nil, s.downloadFilePath(), s.getHeaders()) } diff --git a/pkg/hostman/storageman/disk_local.go b/pkg/hostman/storageman/disk_local.go index 307b7b87bb..2da9cdc52e 100644 --- a/pkg/hostman/storageman/disk_local.go +++ b/pkg/hostman/storageman/disk_local.go @@ -216,7 +216,7 @@ func (d *SLocalDisk) CreateFromTemplate(ctx context.Context, imageId, format str func (d *SLocalDisk) createFromTemplate( ctx context.Context, imageId, format string, imageCacheManager IImageCacheManger, ) (jsonutils.JSONObject, error) { - imageCache := imageCacheManager.AcquireImage(ctx, imageId, d.GetZoneName(), "", "") + imageCache := imageCacheManager.AcquireImage(ctx, imageId, d.GetZoneName(), "", "", "") if imageCache != nil { defer imageCacheManager.ReleaseImage(ctx, imageId) cacheImagePath := imageCache.GetPath() diff --git a/pkg/hostman/storageman/disk_rbd.go b/pkg/hostman/storageman/disk_rbd.go index 3edda08a49..23e57376fd 100644 --- a/pkg/hostman/storageman/disk_rbd.go +++ b/pkg/hostman/storageman/disk_rbd.go @@ -182,7 +182,7 @@ func (d *SRBDDisk) createFromTemplate(ctx context.Context, imageId, format strin if imageCacheManager == nil { return nil, fmt.Errorf("failed to find image cache manger for storage %s", d.Storage.GetStorageName()) } - imageCache := imageCacheManager.AcquireImage(ctx, imageId, d.GetZoneName(), "", "") + imageCache := imageCacheManager.AcquireImage(ctx, imageId, d.GetZoneName(), "", "", "") if imageCache == nil { return nil, fmt.Errorf("failed to qcquire image for storage %s", d.Storage.GetStorageName()) } diff --git a/pkg/hostman/storageman/imagecache_base.go b/pkg/hostman/storageman/imagecache_base.go index fa6690800c..a7aa562bc7 100644 --- a/pkg/hostman/storageman/imagecache_base.go +++ b/pkg/hostman/storageman/imagecache_base.go @@ -24,7 +24,7 @@ type IImageCache interface { GetPath() string GetName() string Load() bool - Acquire(ctx context.Context, zone, srcUrl, format string) bool + Acquire(ctx context.Context, zone, srcUrl, format, checksum string) bool Release() Remove(ctx context.Context) error GetImageId() string diff --git a/pkg/hostman/storageman/imagecache_local.go b/pkg/hostman/storageman/imagecache_local.go index 378592f9a2..3ac3686f71 100644 --- a/pkg/hostman/storageman/imagecache_local.go +++ b/pkg/hostman/storageman/imagecache_local.go @@ -155,15 +155,15 @@ func (l *SLocalImageCache) Release() { l.consumerCount -= 1 } -func (l *SLocalImageCache) Acquire(ctx context.Context, zone, srcUrl, format string) bool { - ret, exit := l.prepare(ctx, zone, srcUrl, format) +func (l *SLocalImageCache) Acquire(ctx context.Context, zone, srcUrl, format, preChksum string) bool { + ret, exit := l.prepare(ctx, zone, srcUrl, format, preChksum) if exit { return ret } return l.fetch(ctx, zone, srcUrl, format) } -func (l *SLocalImageCache) prepare(ctx context.Context, zone, srcUrl, format string) (bool, bool) { +func (l *SLocalImageCache) prepare(ctx context.Context, zone, srcUrl, format, preChksum string) (bool, bool) { l.cond.L.Lock() defer l.cond.L.Unlock() @@ -187,7 +187,7 @@ func (l *SLocalImageCache) prepare(ctx context.Context, zone, srcUrl, format str url += fmt.Sprintf("?format=%s&scope=system", format) l.remoteFile = remotefile.NewRemoteFile(ctx, url, - l.GetPath(), false, "", -1, nil, l.GetTmpPath(), srcUrl) + l.GetPath(), false, preChksum, -1, nil, l.GetTmpPath(), srcUrl) return false, false } diff --git a/pkg/hostman/storageman/imagecache_rbd.go b/pkg/hostman/storageman/imagecache_rbd.go index b9662215eb..bd47b4ab1a 100644 --- a/pkg/hostman/storageman/imagecache_rbd.go +++ b/pkg/hostman/storageman/imagecache_rbd.go @@ -66,8 +66,8 @@ func (r *SRbdImageCache) Load() bool { return origin.IsValid() } -func (r *SRbdImageCache) Acquire(ctx context.Context, zone, srcUrl, format string) bool { - localImageCache := storageManager.LocalStorageImagecacheManager.AcquireImage(ctx, r.imageId, zone, srcUrl, format) +func (r *SRbdImageCache) Acquire(ctx context.Context, zone, srcUrl, format, checksum string) bool { + localImageCache := storageManager.LocalStorageImagecacheManager.AcquireImage(ctx, r.imageId, zone, srcUrl, format, checksum) if localImageCache == nil { log.Errorf("failed to acquireimage %s ", r.imageId) return false diff --git a/pkg/hostman/storageman/imagecachemanager_base.go b/pkg/hostman/storageman/imagecachemanager_base.go index a8fc532e11..9732a607ac 100644 --- a/pkg/hostman/storageman/imagecachemanager_base.go +++ b/pkg/hostman/storageman/imagecachemanager_base.go @@ -51,7 +51,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, imageId, zone, srcUrl, format string) IImageCache + AcquireImage(ctx context.Context, imageId, zone, srcUrl, format, checksum string) IImageCache ReleaseImage(ctx context.Context, imageId string) LoadImageCache(imageId string) } diff --git a/pkg/hostman/storageman/imagecachemanager_local.go b/pkg/hostman/storageman/imagecachemanager_local.go index 8c4268e56a..ce2feff5ad 100644 --- a/pkg/hostman/storageman/imagecachemanager_local.go +++ b/pkg/hostman/storageman/imagecachemanager_local.go @@ -73,7 +73,7 @@ func (c *SLocalImageCacheManager) LoadImageCache(imageId string) { } } -func (c *SLocalImageCacheManager) AcquireImage(ctx context.Context, imageId, zone, srcUrl, format string) IImageCache { +func (c *SLocalImageCacheManager) AcquireImage(ctx context.Context, imageId, zone, srcUrl, format, checksum string) IImageCache { c.lock.LockRawObject(ctx, "image-cache", imageId) defer c.lock.ReleaseRawObject(ctx, "image-cache", imageId) @@ -82,7 +82,7 @@ func (c *SLocalImageCacheManager) AcquireImage(ctx context.Context, imageId, zon img = NewLocalImageCache(imageId, c) c.cachedImages[imageId] = img } - if img.Acquire(ctx, zone, srcUrl, format) { + if img.Acquire(ctx, zone, srcUrl, format, checksum) { return img } else { return nil @@ -131,9 +131,10 @@ func (c *SLocalImageCacheManager) PrefetchImageCache(ctx context.Context, data i } format, _ := body.GetString("format") srcUrl, _ := body.GetString("src_url") + checksum, _ := body.GetString("checksum") if imgCache := c.AcquireImage(ctx, imageId, c.GetStorageManager().GetZoneName(), - srcUrl, format); imgCache != nil { + srcUrl, format, checksum); imgCache != nil { defer imgCache.Release() res := jsonutils.NewDict() diff --git a/pkg/hostman/storageman/imagecachemanager_rbd.go b/pkg/hostman/storageman/imagecachemanager_rbd.go index 751c2ba095..54611f8e74 100644 --- a/pkg/hostman/storageman/imagecachemanager_rbd.go +++ b/pkg/hostman/storageman/imagecachemanager_rbd.go @@ -114,8 +114,9 @@ func (c *SRbdImageCacheManager) PrefetchImageCache(ctx context.Context, data int format, _ := body.GetString("format") srcUrl, _ := body.GetString("src_url") zone, _ := body.GetString("zone") + checksum, _ := body.GetString("checksum") - cache := c.AcquireImage(ctx, imageId, zone, srcUrl, format) + cache := c.AcquireImage(ctx, imageId, zone, srcUrl, format, checksum) if cache == nil { return nil, fmt.Errorf("failed to cache image %s.%s", imageId, format) } @@ -152,7 +153,7 @@ func (c *SRbdImageCacheManager) removeImage(ctx context.Context, imageId string) return nil } -func (c *SRbdImageCacheManager) AcquireImage(ctx context.Context, imageId, zone, srcUrl, format string) IImageCache { +func (c *SRbdImageCacheManager) AcquireImage(ctx context.Context, imageId, zone, srcUrl, format, checksum string) IImageCache { lockman.LockRawObject(ctx, "image-cache", imageId) defer lockman.ReleaseRawObject(ctx, "image-cache", imageId) @@ -161,7 +162,7 @@ func (c *SRbdImageCacheManager) AcquireImage(ctx context.Context, imageId, zone, img = NewRbdImageCache(imageId, c) c.cachedImages[imageId] = img } - if img.Acquire(ctx, zone, srcUrl, format) { + if img.Acquire(ctx, zone, srcUrl, format, checksum) { return img } return nil diff --git a/pkg/hostman/storageman/remotefile/remotefile.go b/pkg/hostman/storageman/remotefile/remotefile.go index 96eeccb5a3..b0f12f8c9e 100644 --- a/pkg/hostman/storageman/remotefile/remotefile.go +++ b/pkg/hostman/storageman/remotefile/remotefile.go @@ -114,12 +114,18 @@ func (r *SRemoteFile) GetInfo() *SImageDesc { } func (r *SRemoteFile) VerifyIntegrity() bool { - if r.download(false, "") { - localChksum, err := fileutils2.MD5(r.localPath) - if err != nil { - log.Errorln(err) - return false + localChksum, err := fileutils2.MD5(r.localPath) + if err != nil { + log.Errorf("MD5SUM local file %s error: %v", r.localPath, err) + return false + } + if r.preChksum != "" { + if localChksum == r.preChksum { + log.Infof("Identical preChksum, skip download") + return true } + } + if r.download(false, "") { if localChksum == r.chksum { log.Infof("Identical chksum, skip download") return true @@ -204,10 +210,10 @@ func (r *SRemoteFile) downloadInternal(getData bool, preChksum string) bool { } } var method, url = "HEAD", r.url + if len(r.downloadUrl) > 0 { + url = r.downloadUrl + } if getData { - if len(r.downloadUrl) > 0 { - url = r.downloadUrl - } method = "GET" } @@ -274,7 +280,7 @@ func (r *SRemoteFile) downloadInternal(getData bool, preChksum string) bool { } return true } else { - log.Errorf("Remote file fetch error %d", resp.StatusCode) + log.Errorf("Remote file fetch %s %s error %d", method, url, resp.StatusCode) return false } }