opensource/cloudpods
3
0
Fork 0
mirror of https://github.com/yunionio/cloudpods.git synced 2026-09-01 15:07:17 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(region): download cached image from source host before migrating

Browse Source
This commit is contained in:
Zexi Li
2021-04-27 11:31:52 +08:00
parent 5788704501
commit 3a6e7f55ab
16 changed files with 112 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/compute/hostdrivers/kvm.go
+30 -29
View File
@@ -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
}
pkg/compute/models/cachedimages.go
+7
View File
@@ -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:
pkg/compute/models/storagecaches.go
+8
View File
@@ -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)
pkg/compute/options/options.go
-1
View File
@@ -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"`
pkg/compute/tasks/guest_live_migrate_task.go
+2 -2
View File
@@ -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()))
}
pkg/hostman/downloader/downloadhandler.go
+15
View File
@@ -60,6 +60,8 @@ func AddDownloadHandler(prefix string, app *appsrv.Application) {
app.AddHandler("HEAD",
fmt.Sprintf("%s/%s/snapshots/<storageId>/<diskId>/<snapshotId>",
prefix, kerword), auth.Authenticate(snapshotHead))
app.AddHandler("HEAD",
fmt.Sprintf("%s/%s/images/<id>", 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["<id>"]
rateLimit := options.HostOptions.BandwidthLimit
compress := isCompress(r)
hand := NewImageCacheDownloadProvider(w, compress, rateLimit, imageId)
if err := hand.HandlerHead(); err != nil {
hostutils.Response(ctx, w, err)
}
}
pkg/hostman/downloader/imagecache_downloader.go
+17
View File
@@ -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())
}
pkg/hostman/storageman/disk_local.go
+1 -1
View File
@@ -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()
pkg/hostman/storageman/disk_rbd.go
+1 -1
View File
@@ -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())
}
pkg/hostman/storageman/imagecache_base.go
+1 -1
View File
@@ -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
pkg/hostman/storageman/imagecache_local.go
+4 -4
View File
@@ -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
}
pkg/hostman/storageman/imagecache_rbd.go
+2 -2
View File
@@ -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
pkg/hostman/storageman/imagecachemanager_base.go
+1 -1
View File
@@ -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)
}
pkg/hostman/storageman/imagecachemanager_local.go
+4 -3
View File
@@ -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()
pkg/hostman/storageman/imagecachemanager_rbd.go
+4 -3
View File
@@ -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
pkg/hostman/storageman/remotefile/remotefile.go
+15 -9
View File
@@ -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
}
}