避免挂载ceph磁盘失败

pkg/compute/models/cachedimages.go

@@ -175,7 +175,7 @@ func (manager *SCachedimageManager) cacheGlanceImageInfo(ctx context.Context, us
 
 	name = db.GenerateName(manager, "", name)
 
-	err := manager.Query().Equals("id", imgId).First(&imageCache)
+	err := manager.RawQuery().Equals("id", imgId).First(&imageCache)
 	if err != nil {
 		if err == sql.ErrNoRows { // insert
 			imageCache.Id = imgId
@@ -200,6 +200,12 @@ func (manager *SCachedimageManager) cacheGlanceImageInfo(ctx context.Context, us
 			imageCache.Size = size
 			imageCache.Info = info
 			imageCache.LastSync = timeutils.UtcNow()
+			if imageCache.Deleted == true {
+				imageCache.Deleted = false
+				imageCache.DeletedAt = time.Time{}
+				imageCache.RefCount = 0
+				imageCache.UpdateVersion = 0
+			}
 			return nil
 		})
 		if err != nil {

pkg/compute/models/storages.go

@@ -175,13 +175,7 @@ func (self *SStorage) AllowDeleteItem(ctx context.Context, userCred mcclient.Tok
 func (manager *SStorageManager) ValidateCreateData(ctx context.Context, userCred mcclient.TokenCredential, ownerProjId string, query jsonutils.JSONObject, data *jsonutils.JSONDict) (*jsonutils.JSONDict, error) {
 	storageType, _ := data.GetString("storage_type")
 	mediumType, _ := data.GetString("medium_type")
-	/*
-		capacity, _ := data.Int("capacity")
-		if capacity < 0 {
-			return nil, httperrors.NewInputParameterError("Invalid capacity")
-		}
-		data.Set("capacity", jsonutils.NewInt(capacity))
-	*/
+
 	if !utils.IsInStringArray(storageType, STORAGE_TYPES) {
 		return nil, httperrors.NewInputParameterError("Invalid storage type %s", storageType)
 	}
@@ -207,73 +201,10 @@ func (manager *SStorageManager) ValidateCreateData(ctx context.Context, userCred
 	if err != nil {
 		return nil, err
 	}
-	/*
-		if storageType == STORAGE_RBD {
-			conf, err := manager.ValidateRbdConfData(data)
-			if err != nil {
-				return nil, httperrors.NewBadRequestError("Vaildata rbd conf error: %s", err.Error())
-			}
-			data.Set("storage_conf", conf)
-			// data.Set("capacity", rbdConf)
-		} else if storageType == STORAGE_NFS {
-			conf, err := manager.ValidataNfsConfdata(data)
-			if err != nil {
-				return nil, httperrors.NewBadRequestError("Vaildata nfs conf error: %s", err.Error())
-			}
-			data.Set("storage_conf", conf)
-		}
-	*/
+
 	return manager.SStandaloneResourceBaseManager.ValidateCreateData(ctx, userCred, ownerProjId, query, data)
 }
 
-/*
-func (manager *SStorageManager) ValidataNfsConfdata(data *jsonutils.JSONDict) (*jsonutils.JSONDict, error) {
-	conf := jsonutils.NewDict()
-	if nfsHost, err := data.GetString("nfs_host"); err != nil {
-		return nil, httperrors.NewInputParameterError("Get nfs conf host error: %s", err.Error())
-	} else {
-		conf.Set("nfs_host", jsonutils.NewString(nfsHost))
-	}
-	if nfsSharedDir, err := data.GetString("nfs_shared_dir"); err != nil {
-		return nil, httperrors.NewInputParameterError("Get nfs conf shared dir error: %s", err.Error())
-	} else {
-		conf.Set("nfs_shared_dir", jsonutils.NewString(nfsSharedDir))
-	}
-	return conf, nil
-}
-*/
-
-/*
-func (manager *SStorageManager) ValidateRbdConfData(data *jsonutils.JSONDict) (*jsonutils.JSONDict, error) {
-	conf := jsonutils.NewDict()
-	for k, v := range data.Value() {
-		if strings.HasPrefix(k, fmt.Sprintf("%s_", STORAGE_RBD)) {
-			k = k[len(STORAGE_RBD)+1:]
-			if len(k) > 0 {
-				conf.Set(k, v)
-			}
-		}
-	}
-	requireFields := []string{"mon_host", "key", "pool"}
-	for _, field := range requireFields {
-		if !conf.Contains(field) {
-			return nil, httperrors.NewMissingParameterError(field)
-		}
-	}
-	storages := make([]SStorage, 0)
-	err := manager.Query().Equals("storage_type", STORAGE_RBD).All(&storages)
-	if err != nil {
-		return nil, err
-	}
-	for i := 0; i < len(storages); i++ {
-		if conf.Equals(storages[i].StorageConf) {
-			return nil, httperrors.NewDuplicateResourceError("This RBD Storage[%s/%s] has already exist", storages[i].Name, conf.String())
-		}
-	}
-	// TODO??? ensure rbd pool can use and get capacity
-	return conf, nil
-}*/
-
 func (self *SStorage) ValidateDeleteCondition(ctx context.Context) error {
 	if self.GetHostCount() > 0 || self.GetDiskCount() > 0 || self.GetSnapshotCount() > 0 {
 		return httperrors.NewNotEmptyError("Not an empty storage provider")
@@ -288,76 +219,6 @@ func (self *SStorage) PostCreate(ctx context.Context, userCred mcclient.TokenCre
 	if storageDriver != nil {
 		storageDriver.PostCreate(ctx, userCred, self, data)
 	}
-
-	/*
-		storageConf, _ := data.Get("storage_conf")
-		if storageConf != nil {
-			_, err := self.GetModelManager().TableSpec().Update(self, func() error {
-				self.StorageConf = storageConf
-				return nil
-			})
-			if err != nil {
-				log.Errorln(err)
-				return
-			}
-		}
-		if self.StorageType == STORAGE_RBD {
-			var storages = make([]SStorage, 0)
-			err := StorageManager.Query().Equals("storage_type", STORAGE_RBD).All(&storages)
-			if err != nil {
-				log.Errorln(err)
-				return
-			}
-			nMonHost, _ := storageConf.GetString("mon_host")
-			nKey, _ := storageConf.GetString("key")
-			for i := 0; i < len(storages); i++ {
-				monHost, _ := storages[i].StorageConf.GetString("mon_host")
-				key, _ := storages[i].StorageConf.GetString("key")
-				if monHost == nMonHost && nKey == key {
-					_, err := self.GetModelManager().TableSpec().Update(self, func() error {
-						self.StoragecacheId = storages[i].StoragecacheId
-						return nil
-					})
-					if err != nil {
-						log.Errorln(err)
-						return
-					}
-					break
-				}
-			}
-			if len(self.StoragecacheId) == 0 {
-				sc := &SStoragecache{}
-				sc.SetModelManager(StoragecacheManager)
-				sc.Name = fmt.Sprintf("imagecache-%s", self.Id)
-				pool, _ := storageConf.GetString("pool")
-				sc.Path = fmt.Sprintf("rbd:%s", pool)
-				err := StorageManager.TableSpec().Insert(sc)
-				if err != nil {
-					log.Errorln(err)
-				}
-			}
-		} else if self.StorageType == STORAGE_NFS {
-			sc := &SStoragecache{}
-			sc.Path = options.Options.NfsDefaultImageCacheDir
-			sc.ExternalId = self.Id
-			sc.Name = "nfs-" + self.Name + time.Now().Format("2006-01-02 15:04:05")
-			if err := StoragecacheManager.TableSpec().Insert(sc); err != nil {
-				log.Errorln(err)
-				return
-			}
-			if err := StoragecacheManager.Query().Equals("external_id", self.Id).First(sc); err != nil {
-				log.Errorln(err)
-				return
-			}
-			_, err := self.GetModelManager().TableSpec().Update(self, func() error {
-				self.StoragecacheId = sc.Id
-				self.Status = STORAGE_ONLINE
-				return nil
-			})
-			if err != nil {
-				log.Errorln(err)
-			}
-		}*/
 }
 
 func (self *SStorage) SetStatus(userCred mcclient.TokenCredential, status string, reason string) error {

pkg/compute/storagedrivers/rbd.go

@@ -39,6 +39,12 @@ func (self *SRbdStorageDriver) ValidateCreateData(ctx context.Context, userCred
 		conf.Add(jsonutils.NewString(key), "key")
 	}
 
+	if timeout, _ := data.Int("rbd_timeout"); timeout > 0 {
+		conf.Add(jsonutils.NewInt(timeout), "rados_osd_op_timeout")
+		conf.Add(jsonutils.NewInt(timeout), "rados_mon_op_timeout")
+		conf.Add(jsonutils.NewInt(timeout), "client_mount_timeout")
+	}
+
 	storages := []models.SStorage{}
 	q := models.StorageManager.Query().Equals("storage_type", models.STORAGE_RBD)
 	if err := db.FetchModelObjects(models.StorageManager, q, &storages); err != nil {

pkg/hostman/storageman/disk_rbd.go

@@ -10,6 +10,7 @@ import (
 	"yunion.io/x/onecloud/pkg/appctx"
 	"yunion.io/x/onecloud/pkg/cloudcommon/storagetypes"
 	"yunion.io/x/onecloud/pkg/hostman/hostutils"
+	"yunion.io/x/pkg/utils"
 )
 
 type SRBDDisk struct {
@@ -121,7 +122,8 @@ func (d *SRBDDisk) PrepareSaveToGlance(ctx context.Context, params interface{})
 	}
 	storage := d.Storage.(*SRbdStorage)
 	pool, _ := storage.GetStorageConf().GetString("pool")
-	if err := storage.copyImage(pool, d.Id, imageCache.GetPath(), imageName); err != nil {
+	if err := storage.cloneImage(pool, d.Id, imageCache.GetPath(), imageName); err != nil {
+		log.Errorf("clone image %s from pool %s to %s/%s error: %v", d.Id, pool, imageCache.GetPath(), imageName, err)
 		return nil, err
 	}
 	return jsonutils.Marshal(map[string]string{"backup": imageName}), nil
@@ -174,7 +176,7 @@ func (d *SRBDDisk) createFromTemplate(ctx context.Context, imageId, format strin
 	defer imageCacheManager.ReleaseImage(imageId)
 	storage := d.Storage.(*SRbdStorage)
 	destPool, _ := storage.StorageConf.GetString("pool")
-	if err := storage.copyImage(imageCacheManager.GetPath(), imageCache.GetName(), destPool, d.Id); err != nil {
+	if err := storage.cloneImage(imageCacheManager.GetPath(), imageCache.GetName(), destPool, d.Id); err != nil {
 		return nil, err
 	}
 	return d.GetDiskDesc(), nil
@@ -190,9 +192,30 @@ func (d *SRBDDisk) CreateRaw(ctx context.Context, sizeMb int, diskFromat string,
 	if err := storage.createImage(pool, diskId, uint64(sizeMb)); err != nil {
 		return nil, err
 	}
+
+	if utils.IsInStringArray(fsFormat, []string{"swap", "ext2", "ext3", "ext4", "xfs"}) {
+		d.FormatFs(fsFormat, diskId)
+	}
+
 	return d.GetDiskDesc(), nil
 }
 
+func (d *SRBDDisk) 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, uuid)
+			if err != nil {
+				log.Errorln(err)
+			}
+		} else {
+			log.Errorln(err)
+		}
+	}
+}
+
 func (d *SRBDDisk) PostCreateFromImageFuse() {
 	log.Errorf("Not support PostCreateFromImageFuse")
 }

pkg/hostman/storageman/diskutils.go

@@ -3,6 +3,7 @@ package storageman
 import (
 	"fmt"
 	"io/ioutil"
+	"os"
 	"path"
 	"path/filepath"
 	"strings"
@@ -42,6 +43,21 @@ func (d *SKVMGuestDisk) Connect() bool {
 
 	var cmd []string
 	if strings.HasPrefix(d.imagePath, "rbd:") || d.getImageFormat() == "raw" {
+		//qemu-nbd 连接ceph时 /etc/ceph/ceph.conf 必须存在
+		if strings.HasPrefix(d.imagePath, "rbd:") {
+			if !fileutils2.Exists("/etc/ceph") {
+				if err := os.Mkdir("/etc/ceph", 0755); err != nil {
+					log.Errorf("failed to mkdir /etc/ceph error: %v", err)
+					return false
+				}
+			}
+			if !fileutils2.IsFile("/etc/ceph/ceph.conf") {
+				if _, err := os.Create("/etc/ceph/ceph.conf"); err != nil {
+					log.Errorf("failed to create /etc/ceph/ceph.conf error: %v", err)
+					return false
+				}
+			}
+		}
 		cmd = []string{qemutils.GetQemuNbd(), "-c", d.nbdDev, "-f", "raw", d.imagePath}
 	} else {
 		cmd = []string{qemutils.GetQemuNbd(), "-c", d.nbdDev, d.imagePath}

pkg/hostman/storageman/storage_rbd.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"os"
 	"strings"
 
 	"github.com/ceph/go-ceph/rados"
@@ -12,14 +13,19 @@ import (
 	"yunion.io/x/log"
 	"yunion.io/x/onecloud/pkg/cloudcommon/storagetypes"
 	"yunion.io/x/onecloud/pkg/compute/models"
+	"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/mcclient/modules"
+	"yunion.io/x/onecloud/pkg/util/procutils"
+	"yunion.io/x/onecloud/pkg/util/qemutils"
 	"yunion.io/x/pkg/utils"
 )
 
 const (
-	RBD_FEATURE = 2
-	RBD_ORDER   = 22 //为rbd对应到rados中每个对象的大小，默认为4MB
+	RBD_FEATURE     = 3
+	RBD_ORDER       = 22  //为rbd对应到rados中每个对象的大小，默认为4MB
+	DEFAULT_TIMEOUT = 240 //4 minutes
 )
 
 var (
@@ -64,7 +70,7 @@ func (s *SRbdStorage) GetImgsaveBackupPath() string {
 //Tip Configuration values containing :, @, or = can be escaped with a leading \ character.
 func (s *SRbdStorage) getStorageConfString() string {
 	conf := ""
-	for _, key := range []string{"mon_host", "key", "rados_osd_op_timeout", "rados_mon_op_timeout", "client_mount_timeout", "rbd_default_format"} {
+	for _, key := range []string{"mon_host", "key"} {
 		if value, _ := s.StorageConf.GetString(key); len(value) > 0 {
 			if key == "mon_host" {
 				value = strings.Replace(value, ",", `\;`, -1)
@@ -77,6 +83,13 @@ func (s *SRbdStorage) getStorageConfString() string {
 			conf += fmt.Sprintf(":%s=%s", key, value)
 		}
 	}
+	for _, key := range []string{"rados_osd_op_timeout", "rados_mon_op_timeout", "client_mount_timeout"} {
+		var timeout int64
+		if timeout, _ = s.StorageConf.Int(key); timeout == 0 {
+			timeout = DEFAULT_TIMEOUT
+		}
+		conf += fmt.Sprintf(":%s=%d", key, timeout)
+	}
 	return conf
 }
 
@@ -122,6 +135,7 @@ func (s *SRbdStorage) deleteImage(pool string, name string) error {
 	return err
 }
 
+// 比较费时
 func (s *SRbdStorage) copyImage(srcPool string, srcImage string, destPool string, destImage string) error {
 	_, err := s.withImage(srcPool, srcImage, func(src *rbd.Image) (interface{}, error) {
 		imageSize, err := src.GetSize()
@@ -140,6 +154,41 @@ func (s *SRbdStorage) copyImage(srcPool string, srcImage string, destPool string
 	return err
 }
 
+// 速度快
+func (s *SRbdStorage) cloneImage(srcPool string, srcImage string, destPool string, destImage string) error {
+	_, err := s.withImage(srcPool, srcImage, func(src *rbd.Image) (interface{}, error) {
+		snapshot, err := src.CreateSnapshot(destImage)
+		if err != nil {
+			log.Errorf("create snapshot error: %v", err)
+			return nil, err
+		}
+		names, err := src.GetSnapshotNames()
+
+		defer snapshot.Remove()
+		isProtect, err := snapshot.IsProtected()
+		if err != nil {
+			return nil, err
+		}
+		if !isProtect {
+			if err := snapshot.Protect(); err != nil {
+				log.Errorf("snapshot protect error: %v", err)
+				return nil, err
+			}
+		}
+		defer snapshot.Unprotect()
+
+		return s.withIOContext(destPool, func(ioctx *rados.IOContext) (interface{}, error) {
+			dest, err := src.Clone(destImage, ioctx, destImage, RBD_FEATURE, RBD_ORDER)
+			if err != nil {
+				return nil, err
+			}
+			defer dest.Close()
+			return nil, dest.Flatten()
+		})
+	})
+	return err
+}
+
 func (s *SRbdStorage) withImage(pool string, name string, doFunc func(*rbd.Image) (interface{}, error)) (interface{}, error) {
 	return s.withIOContext(pool, func(ioctx *rados.IOContext) (interface{}, error) {
 		names, err := rbd.GetImageNames(ioctx)
@@ -210,6 +259,13 @@ func (s *SRbdStorage) createImage(pool string, name string, sizeMb uint64) error
 	return err
 }
 
+func (s *SRbdStorage) renameImage(pool string, src string, dest string) error {
+	_, err := s.withImage(pool, src, func(image *rbd.Image) (interface{}, error) {
+		return nil, image.Rename(dest)
+	})
+	return err
+}
+
 func (s *SRbdStorage) createSnapshot(pool string, diskId string, snapshotId string) error {
 	_, err := s.withImage(pool, diskId, func(image *rbd.Image) (interface{}, error) {
 		return image.CreateSnapshot(snapshotId)
@@ -299,9 +355,132 @@ func (s *SRbdStorage) Accessible() bool {
 }
 
 func (s *SRbdStorage) SaveToGlance(ctx context.Context, params interface{}) (jsonutils.JSONObject, error) {
+	data, ok := params.(*jsonutils.JSONDict)
+	if !ok {
+		return nil, hostutils.ParamsError
+	}
+
+	rbdImageCache := storageManager.GetStoragecacheById(s.GetStoragecacheId())
+	if rbdImageCache == nil {
+		return nil, fmt.Errorf("failed to find storage image cache for storage %s", s.GetStorageName())
+	}
+
+	imagePath, _ := data.GetString("image_path")
+	compress := jsonutils.QueryBoolean(data, "compress", true)
+	format, _ := data.GetString("format")
+	imageId, _ := data.GetString("image_id")
+	imageName := "image_cache_" + imageId
+	if err := s.renameImage(rbdImageCache.GetPath(), imagePath, imageName); err != nil {
+		return nil, err
+	}
+
+	imagePath = fmt.Sprintf("rbd:%s/%s%s", rbdImageCache.GetPath(), imageName, s.getStorageConfString())
+
+	if err := s.saveToGlance(ctx, imageId, imagePath, compress, format); err != nil {
+		log.Errorf("Save to glance failed: %s", err)
+		s.onSaveToGlanceFailed(ctx, imageId)
+	}
+
+	rbdImageCache.LoadImageCache(imageId)
+	_, err := hostutils.RemoteStoragecacheCacheImage(ctx, rbdImageCache.GetId(), imageId, "ready", imagePath)
+	if err != nil {
+		log.Errorf("ail to remote cache image: %v", err)
+	}
 	return nil, nil
 }
 
+func (s *SRbdStorage) onSaveToGlanceFailed(ctx context.Context, imageId string) {
+	params := jsonutils.NewDict()
+	params.Set("status", jsonutils.NewString("killed"))
+	_, err := modules.Images.Update(hostutils.GetImageSession(ctx, s.GetZone()),
+		imageId, params)
+	if err != nil {
+		log.Errorln(err)
+	}
+}
+
+func (s *SRbdStorage) saveToGlance(ctx context.Context, imageId, imagePath string, compress bool, format string) error {
+	var (
+		kvmDisk = NewKVMGuestDisk(imagePath)
+		osInfo  string
+		relInfo *fsdriver.SReleaseInfo
+	)
+
+	if err := func() error {
+		if kvmDisk.Connect() {
+			defer kvmDisk.Disconnect()
+
+			if root := kvmDisk.MountKvmRootfs(); root != nil {
+				defer kvmDisk.UmountKvmRootfs(root)
+
+				osInfo = root.GetOs()
+				relInfo = root.GetReleaseInfo(root.GetPartition())
+				if compress {
+					if err := root.PrepareFsForTemplate(root.GetPartition()); err != nil {
+						log.Errorln(err)
+						return err
+					}
+				}
+			}
+
+			if compress {
+				kvmDisk.Zerofree()
+			}
+		}
+		return nil
+	}(); err != nil {
+		return err
+	}
+
+	tmpImageFile := fmt.Sprintf("/tmp/%s.img", imageId)
+	if len(format) == 0 {
+		format = options.HostOptions.DefaultImageSaveFormat
+	}
+
+	_, err := procutils.NewCommand(qemutils.GetQemuImg(), "convert", "-f", "raw", "-O", format, imagePath, tmpImageFile).Run()
+	if err != nil {
+		return err
+	}
+
+	f, err := os.Open(tmpImageFile)
+	if err != nil {
+		return err
+	}
+
+	defer os.Remove(tmpImageFile)
+
+	finfo, err := f.Stat()
+	if err != nil {
+		return err
+	}
+	size := finfo.Size()
+
+	var params = jsonutils.NewDict()
+	if len(osInfo) > 0 {
+		params.Set("os_type", jsonutils.NewString(osInfo))
+	}
+	if relInfo != nil {
+		params.Set("os_distribution", jsonutils.NewString(relInfo.Distro))
+		if len(relInfo.Version) > 0 {
+			params.Set("os_version", jsonutils.NewString(relInfo.Version))
+		}
+		if len(relInfo.Arch) > 0 {
+			params.Set("os_arch", jsonutils.NewString(relInfo.Arch))
+		}
+		if len(relInfo.Version) > 0 {
+			params.Set("os_language", jsonutils.NewString(relInfo.Language))
+		}
+	}
+	params.Set("image_id", jsonutils.NewString(imageId))
+
+	_, err = modules.Images.Upload(hostutils.GetImageSession(ctx, s.GetZone()),
+		params, f, size)
+	f.Close()
+	// TODO
+	// notify_template_ready
+	return err
+}
+
 func (s *SRbdStorage) CreateSnapshotFormUrl(ctx context.Context, snapshotUrl, diskId, snapshotPath string) error {
 	return nil
 }