feat(glance,region,host): auto cache glance images to ceph storages (#22076)

pkg/apis/compute/storage.go

@@ -80,6 +80,9 @@ type StorageCreateInput struct {
 	// enable ceph messenger v2
 	EnableMessengerV2 *bool `json:"enable_messenger_v2"`
 
+	// rbd storage auto cache glance images
+	AutoCacheImages *bool `json:"auto_cache_images"`
+
 	// swagger:ignore
 	MonHost string
 
@@ -246,6 +249,8 @@ type StorageUpdateInput struct {
 
 	// enable ceph messenger v2
 	EnableMessengerV2 *bool `json:"enable_messenger_v2"`
+	// rbd storage auto cache glance images
+	AutoCacheImages *bool `json:"auto_cache_images"`
 
 	RbdTimeoutInput
 
@@ -259,6 +264,18 @@ type StorageUpdateInput struct {
 	MasterHost   string
 }
 
+type RbdStorageConf struct {
+	RadosMonOpTimeout  int `json:"rados_mon_op_timeout"`
+	RadosOsdOpTimeout  int `json:"rados_osd_op_timeout"`
+	ClientMountTimeout int `json:"client_mount_timeout"`
+
+	MonHost           string `json:"mon_host"`
+	Pool              string `json:"pool"`
+	Key               string `json:"key"`
+	EnableMessengerV2 bool   `json:"enable_messenger_v2"`
+	AutoCacheImages   bool   `json:"auto_cache_images"`
+}
+
 type StorageSetCmtBoundInput struct {
 	Cmtbound *float32
 }

pkg/apis/compute/storage_const.go

@@ -282,4 +282,7 @@ type StorageListInput struct {
 
 	// filter storages of baremetal host
 	IsBaremetal *bool `json:"is_baremetal"`
+
+	// filter by storage type
+	StorageType string `json:"storage_type"`
 }

pkg/compute/models/storages.go

@@ -26,6 +26,7 @@ import (
 	"yunion.io/x/jsonutils"
 	"yunion.io/x/log"
 	"yunion.io/x/pkg/errors"
+	"yunion.io/x/pkg/gotypes"
 	"yunion.io/x/pkg/tristate"
 	"yunion.io/x/pkg/util/compare"
 	"yunion.io/x/pkg/util/rbacscope"
@@ -118,11 +119,14 @@ func (self *SStorage) ValidateUpdateData(ctx context.Context, userCred mcclient.
 	if err != nil {
 		return input, err
 	}
-	input.StorageConf = jsonutils.NewDict()
+
+	if gotypes.IsNil(input.StorageConf) {
+		input.StorageConf = jsonutils.NewDict()
+	}
 	if self.StorageConf != nil {
 		confs, _ := self.StorageConf.GetMap()
 		for k, v := range confs {
-			if !input.StorageConf.Contains(k) {
+			if input.StorageConf.Contains(k) {
 				continue
 			}
 			input.StorageConf.Set(k, v)
@@ -1742,6 +1746,10 @@ func (manager *SStorageManager) ListItemFilter(
 		q = q.Filter(sqlchemy.In(q.Field("storage_type"), api.STORAGE_LOCAL_TYPES))
 	}
 
+	if len(query.StorageType) > 0 {
+		q = q.Equals("storage_type", query.StorageType)
+	}
+
 	if len(query.SchedtagId) > 0 {
 		schedTag, err := SchedtagManager.FetchByIdOrName(ctx, nil, query.SchedtagId)
 		if err != nil {

pkg/compute/storagedrivers/rbd.go

@@ -91,6 +91,10 @@ func (self *SRbdStorageDriver) ValidateCreateData(ctx context.Context, userCred
 	if input.EnableMessengerV2 != nil {
 		enableMessengerV2 = *input.EnableMessengerV2
 	}
+	autoCacheImages := false
+	if input.AutoCacheImages != nil {
+		autoCacheImages = *input.AutoCacheImages
+	}
 	input.StorageConf.Update(
 		jsonutils.Marshal(map[string]interface{}{
 			"mon_host":             input.MonHost,
@@ -100,6 +104,7 @@ func (self *SRbdStorageDriver) ValidateCreateData(ctx context.Context, userCred
 			"rados_osd_op_timeout": input.RadosOsdOpTimeout,
 			"client_mount_timeout": input.ClientMountTimeout,
 			"enable_messenger_v2":  enableMessengerV2,
+			"auto_cache_images":    autoCacheImages,
 		}))
 	return nil
 }
@@ -120,6 +125,11 @@ func (self *SRbdStorageDriver) ValidateUpdateData(ctx context.Context, userCred
 		input.UpdateStorageConf = true
 	}
 
+	if input.AutoCacheImages != nil {
+		input.StorageConf.Set("auto_cache_images", jsonutils.NewBool(*input.AutoCacheImages))
+		input.UpdateStorageConf = true
+	}
+
 	if len(input.RbdKey) > 0 {
 		input.StorageConf.Set("key", jsonutils.NewString(strings.Trim(input.RbdKey, " ")))
 		input.UpdateStorageConf = true

pkg/hostman/storageman/imagecache_rbd.go

@@ -73,6 +73,12 @@ func (r *SRbdImageCache) Load() error {
 }
 
 func (r *SRbdImageCache) Acquire(ctx context.Context, input api.CacheImageInput, callback func(progress, progressMbps float64, totalSizeMb int64)) error {
+	if err := r.Load(); err == nil {
+		log.Infof("rbd image %s has been cached at pool %s", r.imageId, r.Manager.GetPath())
+		r.imageName = r.imageId
+		return nil
+	}
+
 	input.ImageId = r.imageId
 	localImageCache, err := storageManager.LocalStorageImagecacheManager.AcquireImage(ctx, input, func(progress, progressMbps float64, totalSizeMb int64) {
 		if len(input.ServerId) > 0 {

pkg/hostman/storageman/storage_rbd.go

@@ -19,7 +19,6 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
-	"strings"
 
 	"yunion.io/x/jsonutils"
 	"yunion.io/x/log"
@@ -142,23 +141,7 @@ func (s *SRbdStorage) GetImgsaveBackupPath() string {
 
 // Tip Configuration values containing :, @, or = can be escaped with a leading \ character.
 func (s *SRbdStorage) getStorageConfString() string {
-	conf := []string{}
-	conf = append(conf, "mon_host="+strings.ReplaceAll(s.MonHost, ",", `\;`))
-	key := s.Key
-	if len(key) > 0 {
-		for _, k := range []string{":", "@", "="} {
-			key = strings.ReplaceAll(key, k, fmt.Sprintf(`\%s`, k))
-		}
-		conf = append(conf, "key="+key)
-	}
-	for k, timeout := range map[string]int64{
-		"rados_mon_op_timeout": s.RadosMonOpTimeout,
-		"rados_osd_op_timeout": s.RadosOsdOpTimeout,
-		"client_mount_timeout": s.ClientMountTimeout,
-	} {
-		conf = append(conf, fmt.Sprintf("%s=%d", k, timeout))
-	}
-	return ":" + strings.Join(conf, ":")
+	return cephutils.CephConfString(s.MonHost, s.Key, s.RadosMonOpTimeout, s.RadosOsdOpTimeout, s.ClientMountTimeout)
 }
 
 func (s *SRbdStorage) listImages() ([]string, error) {

pkg/image/models/image_ceph_cache_storages.go

@@ -0,0 +1,129 @@
+// Copyright 2019 Yunion
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package models
+
+import (
+	"context"
+
+	"yunion.io/x/jsonutils"
+	"yunion.io/x/log"
+	"yunion.io/x/pkg/errors"
+
+	computeapi "yunion.io/x/onecloud/pkg/apis/compute"
+	"yunion.io/x/onecloud/pkg/image/options"
+	"yunion.io/x/onecloud/pkg/mcclient/auth"
+	"yunion.io/x/onecloud/pkg/mcclient/modules/compute"
+	"yunion.io/x/onecloud/pkg/util/cephutils"
+)
+
+func GetRegionCephStorages() (map[string]*computeapi.RbdStorageConf, error) {
+	q := struct {
+		Scope       string `json:"scope"`
+		StorageType string `json:"storage_type"`
+	}{
+		"system", computeapi.STORAGE_RBD,
+	}
+
+	res, err := compute.Storages.List(auth.GetAdminSession(context.Background(), options.Options.Region), jsonutils.Marshal(q))
+	if err != nil {
+		return nil, errors.Wrap(err, "compute.Storages.List")
+	}
+	if res.Total <= 0 {
+		return nil, nil
+	}
+	storages := []computeapi.StorageDetails{}
+	err = jsonutils.Update(&storages, res.Data)
+	if err != nil {
+		return nil, errors.Wrap(err, "json parse storage details")
+	}
+	cephStorages := map[string]*computeapi.RbdStorageConf{}
+	for i := range storages {
+		if storages[i].StorageType != computeapi.STORAGE_RBD {
+			continue
+		}
+		if jsonutils.QueryBoolean(storages[i].StorageConf, "auto_cache_images", false) {
+			conf := new(computeapi.RbdStorageConf)
+			if err := storages[i].StorageConf.Unmarshal(conf); err != nil {
+				log.Errorf("failed unmarshal storage %s: %s", storages[i].StorageConf, err)
+				continue
+			}
+			cephStorages[storages[i].Id] = conf
+		}
+	}
+	return cephStorages, nil
+}
+
+type SCephStorageConf struct {
+	StorageIdConf     map[string]*computeapi.RbdStorageConf
+	CephFsidStorageId map[string][]string
+}
+
+var (
+	cephStorages          *SCephStorageConf
+	requestedCephStorages bool
+)
+
+func GetCephStorages() *SCephStorageConf {
+	cephutils.SetCephConfTempDir(options.Options.DefaultImageServiceHomeDir)
+	if requestedCephStorages {
+		return cephStorages
+	}
+	cephStorages = getCephStorages()
+	requestedCephStorages = true
+	return cephStorages
+}
+
+func getCephStorages() *SCephStorageConf {
+	storagesConf, err := GetRegionCephStorages()
+	if err != nil {
+		log.Errorf("failed GetCephStorages %s", err)
+		return nil
+	}
+	if len(storagesConf) == 0 {
+		log.Infof("No enable auto_cache_image ceph storage found...")
+		return nil
+	}
+	fsidStorageMap := map[string][]string{}
+	for id := range storagesConf {
+		fsid := getStorageFsid(storagesConf[id])
+		if fsid == "" {
+			continue
+		}
+		storages, ok := fsidStorageMap[fsid]
+		if !ok {
+			storages = make([]string, 0)
+		}
+		fsidStorageMap[fsid] = append(storages, id)
+	}
+	return &SCephStorageConf{
+		StorageIdConf:     storagesConf,
+		CephFsidStorageId: fsidStorageMap,
+	}
+}
+
+func getStorageFsid(conf *computeapi.RbdStorageConf) string {
+	cli, err := cephutils.NewClient(conf.MonHost, conf.Key, conf.Pool, conf.EnableMessengerV2)
+	if err != nil {
+		log.Errorf("failed new client of ceph storage %s:%s", conf.MonHost, conf.Pool)
+		return ""
+	}
+	defer cli.Close()
+	fsid, err := cli.Fsid()
+	if err != nil {
+		log.Errorf("failed get fsid of ceph storage %s:%s: %s", conf.MonHost, conf.Pool, err)
+		return ""
+	}
+	return fsid
+}

pkg/image/models/images.go

@@ -41,6 +41,7 @@ import (
 	"yunion.io/x/sqlchemy"
 
 	"yunion.io/x/onecloud/pkg/apis"
+	computeapi "yunion.io/x/onecloud/pkg/apis/compute"
 	api "yunion.io/x/onecloud/pkg/apis/image"
 	noapi "yunion.io/x/onecloud/pkg/apis/notify"
 	"yunion.io/x/onecloud/pkg/appsrv"
@@ -59,10 +60,12 @@ import (
 	identity_modules "yunion.io/x/onecloud/pkg/mcclient/modules/identity"
 	modules "yunion.io/x/onecloud/pkg/mcclient/modules/image"
 	"yunion.io/x/onecloud/pkg/mcclient/modules/notify"
+	"yunion.io/x/onecloud/pkg/util/cephutils"
 	"yunion.io/x/onecloud/pkg/util/fileutils2"
 	"yunion.io/x/onecloud/pkg/util/logclient"
 	"yunion.io/x/onecloud/pkg/util/procutils"
 	"yunion.io/x/onecloud/pkg/util/qemuimg"
+	"yunion.io/x/onecloud/pkg/util/qemutils"
 	"yunion.io/x/onecloud/pkg/util/rbacutils"
 	"yunion.io/x/onecloud/pkg/util/stringutils2"
 )
@@ -2085,7 +2088,7 @@ func (img *SImage) Pipeline(ctx context.Context, userCred mcclient.TokenCredenti
 		}
 	}
 	{
-		// do conert
+		// do convert
 		converted, err := img.doConvert(ctx, userCred)
 		if err != nil {
 			return errors.Wrap(err, "doConvert")
@@ -2104,6 +2107,12 @@ func (img *SImage) Pipeline(ctx context.Context, userCred mcclient.TokenCredenti
 			updated = true
 		}
 	}
+	{
+		// do cache to ceph storages
+		if img.GetImageType() != api.ImageTypeTarGzip {
+			img.cacheToCephStorages(ctx)
+		}
+	}
 	if img.Status != api.IMAGE_STATUS_ACTIVE {
 		img.SetStatus(ctx, userCred, api.IMAGE_STATUS_ACTIVE, "image pipeline complete")
 	}
@@ -2120,6 +2129,192 @@ func (img *SImage) Pipeline(ctx context.Context, userCred mcclient.TokenCredenti
 	return nil
 }
 
+func (img *SImage) cacheToCephStorages(ctx context.Context) {
+	// skip if image converting
+	localPath := img.GetPath(img.DiskFormat)
+	if procutils.NewRemoteCommandAsFarAsPossible("sh", "-c",
+		fmt.Sprintf("ps -ef | grep [q]emu-img | grep convert | grep %s", localPath)) == nil {
+		log.Warningf("image %s has converting progress", img.Id)
+		return
+	}
+
+	cephStorages := GetCephStorages()
+	if cephStorages == nil || len(cephStorages.StorageIdConf) == 0 {
+		return
+	}
+	for fsid, storageIds := range cephStorages.CephFsidStorageId {
+		storageCachedImages := map[string]*cephutils.SImage{}
+		var cachedRbdimgStorageId string
+		for i := range storageIds {
+			storageConf := cephStorages.StorageIdConf[storageIds[i]]
+			rbdimg, err := img.getCephImage(storageConf, "")
+			if err != nil {
+				log.Errorf("failed get img %s by storage conf %#v: %s", img.Id, storageConf, err)
+				continue
+			}
+			if rbdimg != nil && cachedRbdimgStorageId == "" {
+				cachedRbdimgStorageId = storageIds[i]
+			}
+			if rbdimg != nil {
+				log.Infof("image %s has been cached at ceph pool: %s", img.Id, storageConf.Pool)
+			}
+			storageCachedImages[storageIds[i]] = rbdimg
+		}
+		if len(storageCachedImages) == 0 {
+			// ceph storage unreachable
+			log.Errorf("all of cpeh storage with fsid %s failed get ceph image", fsid)
+			continue
+		}
+		if cachedRbdimgStorageId == "" {
+			// do cache img to ceph storage
+			for storageId := range storageCachedImages {
+				storageConf := cephStorages.StorageIdConf[storageId]
+				imgTmpName := "image_cache_" + img.Id + ".tmp"
+				if !fileutils2.Exists(localPath) {
+					log.Errorf("image localpath %s not exist", localPath)
+					continue
+				}
+				// remove tmp image first
+				if err := img.removeCephImage(storageConf, imgTmpName); err != nil {
+					log.Errorf("remove existing tmp img %s failed: %s", imgTmpName, err)
+					continue
+				}
+				storageConfString := cephutils.CephConfString(
+					storageConf.MonHost,
+					storageConf.Key,
+					int64(storageConf.RadosMonOpTimeout),
+					int64(storageConf.RadosOsdOpTimeout),
+					int64(storageConf.ClientMountTimeout),
+				)
+				rbdPath := fmt.Sprintf("rbd:%s/%s%s", storageConf.Pool, imgTmpName, storageConfString)
+				log.Infof("convert local image %s to rbd pool %s", img.Id, storageConf.Pool)
+				out, err := procutils.NewRemoteCommandAsFarAsPossible(qemutils.GetQemuImg(),
+					"convert", "-W", "-m", "16", "-O", "raw", localPath, rbdPath).Output()
+				if err != nil {
+					log.Errorf("convert local image %s to rbd pool %s failed: %s %s", img.Id, storageConf.Pool, out, err)
+					continue
+				}
+				log.Infof("Success cached img %s to pool %s by convert", imgTmpName, storageConf.Pool)
+				rbdimg, err := img.getCephImage(storageConf, imgTmpName)
+				if err != nil {
+					log.Errorf("failed get ceph image %s after convert to ceph: %s", imgTmpName, err)
+					continue
+				} else if rbdimg == nil {
+					log.Errorf("failed get ceph image %s after convert to ceph, rbdimage not found", imgTmpName)
+					continue
+				} else {
+					imgName := "image_cache_" + img.Id
+					if err = img.renameCephImage(storageConf, imgTmpName, imgName); err != nil {
+						log.Errorf("failed rename from tmp image %s to %s: %s", imgTmpName, imgName, err)
+						continue
+					}
+					cachedRbdimgStorageId = storageId
+					delete(storageCachedImages, storageId)
+					break
+				}
+			}
+		}
+		if cachedRbdimgStorageId == "" {
+			log.Errorf("failed cache img %s to ceph storages fsid: %s", img.Id, fsid)
+			continue
+		}
+
+		for storageId := range storageCachedImages {
+			if storageId != cachedRbdimgStorageId && storageCachedImages[storageId] == nil {
+				srcConf := cephStorages.StorageIdConf[cachedRbdimgStorageId]
+				destConf := cephStorages.StorageIdConf[storageId]
+				err := img.cloneToCephStorage(ctx, srcConf.MonHost, srcConf.Key, srcConf.Pool, srcConf.EnableMessengerV2, destConf.Pool)
+				if err != nil {
+					log.Errorf("failed cache img %s to pool %s: %s", img.Id, destConf.Pool, err)
+					continue
+				}
+				log.Infof("Success cached img %s to pool %s by clone", img.Id, destConf.Pool)
+			}
+		}
+	}
+}
+
+func (img *SImage) removeCephImage(storageConf *computeapi.RbdStorageConf, imgName string) error {
+	cli, err := cephutils.NewClient(storageConf.MonHost, storageConf.Key, storageConf.Pool, storageConf.EnableMessengerV2)
+	if err != nil {
+		return errors.Wrap(err, "cephutils.NewClient")
+	}
+	defer cli.Close()
+	rbdimg, err := cli.GetImage(imgName)
+	if err != nil {
+		if errors.Cause(err) == errors.ErrNotFound {
+			return nil
+		}
+		return errors.Wrapf(err, "GetImage")
+	}
+	return rbdimg.Remove()
+}
+
+func (img *SImage) renameCephImage(storageConf *computeapi.RbdStorageConf, srcImgName, destImgName string) error {
+	cli, err := cephutils.NewClient(storageConf.MonHost, storageConf.Key, storageConf.Pool, storageConf.EnableMessengerV2)
+	if err != nil {
+		return errors.Wrap(err, "cephutils.NewClient")
+	}
+	defer cli.Close()
+	rbdimg, err := cli.GetImage(srcImgName)
+	if err != nil {
+		return errors.Wrapf(err, "GetImage")
+	}
+	return rbdimg.Rename(destImgName)
+}
+
+func (img *SImage) getCephImage(storageConf *computeapi.RbdStorageConf, imgName string) (*cephutils.SImage, error) {
+	if imgName == "" {
+		imgName = "image_cache_" + img.Id
+	}
+	cli, err := cephutils.NewClient(storageConf.MonHost, storageConf.Key, storageConf.Pool, storageConf.EnableMessengerV2)
+	if err != nil {
+		return nil, errors.Wrap(err, "cephutils.NewClient")
+	}
+	defer cli.Close()
+	rbdimg, err := cli.GetImage(imgName)
+	if err != nil {
+		if errors.Cause(err) == errors.ErrNotFound {
+			return nil, nil
+		}
+		return nil, errors.Wrapf(err, "GetImage")
+	}
+	storageConfString := cephutils.CephConfString(
+		storageConf.MonHost,
+		storageConf.Key,
+		int64(storageConf.RadosMonOpTimeout),
+		int64(storageConf.RadosOsdOpTimeout),
+		int64(storageConf.ClientMountTimeout),
+	)
+	rbdPath := fmt.Sprintf("rbd:%s/%s%s", storageConf.Pool, imgName, storageConfString)
+	origin, err := qemuimg.NewQemuImage(rbdPath)
+	if err != nil {
+		return nil, errors.Wrapf(err, "NewQemuImage %s", rbdPath)
+	}
+	if !origin.IsValid() {
+		return nil, errors.Errorf("rbd img %s is invalid", rbdPath)
+	}
+	return rbdimg, nil
+}
+
+func (img *SImage) cloneToCephStorage(ctx context.Context, monHost, key, pool string, enableMessengerV2 bool, destPool string) error {
+	imgName := "image_cache_" + img.Id
+	cli, err := cephutils.NewClient(monHost, key, pool, enableMessengerV2)
+	if err != nil {
+		return errors.Wrap(err, "cephutils.NewClient")
+	}
+	defer cli.Close()
+	rbdimg, err := cli.GetImage(imgName)
+	if err != nil {
+		return errors.Wrap(err, "cli.GetImage(imgName)")
+	}
+	_, err = rbdimg.Clone(ctx, destPool, imgName)
+	if err != nil {
+		return errors.Wrapf(err, "rbdimg.Clone to destPool %s", destPool)
+	}
+	return nil
+}
+
 func (img *SImage) getGuestImageCount() (int, error) {
 	gis, err := GuestImageJointManager.GetByImageId(img.Id)
 	if err != nil {

pkg/image/options/options.go

@@ -40,6 +40,8 @@ type SImageOptions struct {
 
 	TorrentClientPath string `help:"path to torrent executable" default:"/opt/yunion/bin/torrent"`
 
+	DefaultImageServiceHomeDir string `help:"Default image service home dir" default:"/opt/cloud/workspace/data/glance"`
+
 	// DeployServerSocketPath string `help:"Deploy server listen socket path" default:"/var/run/onecloud/deploy.sock"`
 
 	StorageDriver string `help:"image backend storage" default:"local" choices:"s3|local"`

pkg/mcclient/options/compute/storage.go

@@ -58,6 +58,7 @@ type StorageUpdateOptions struct {
 	Reserved              string `help:"Reserved storage space"`
 	Capacity              int    `help:"Capacity for storage"`
 	MasterHost            string `help:"slvm storage master host"`
+	AutoCacheImages       *bool  `help:"ceph storage auto cache glance images"`
 }
 
 func (opts *StorageUpdateOptions) Params() (jsonutils.JSONObject, error) {
@@ -77,6 +78,7 @@ type StorageCreateOptions struct {
 	RbdClientMountTimeout int64  `help:"ceph client_mount_timeout"`
 	RbdKey                string `help:"Ceph key config"`
 	RbdPool               string `help:"Ceph Pool Name"`
+	AutoCacheImages       *bool  `help:"ceph storage auto cache glance images"`
 	NfsHost               string `help:"NFS host"`
 	NfsSharedDir          string `help:"NFS shared dir"`
 	ClvmVgName            string `help:"clvm vg name"`

pkg/util/cephutils/ceph.go

@@ -195,7 +195,7 @@ func (cli *CephClient) GetCapacity() (*SCapacity, error) {
 }
 
 func writeFile(pattern string, content string) (string, error) {
-	file, err := ioutil.TempFile("", pattern)
+	file, err := ioutil.TempFile(cephConfTmpDir, pattern)
 	if err != nil {
 		return "", errors.Wrapf(err, "TempFile")
 	}
@@ -230,6 +230,12 @@ func (cli *CephClient) SetTimeout(timeout int) {
 
 const DEFAULT_TIMTOUT_SECOND = 15
 
+var cephConfTmpDir = ""
+
+func SetCephConfTempDir(dir string) {
+	cephConfTmpDir = dir
+}
+
 func NewClient(monHost, key, pool string, enableMessengerV2 bool) (*CephClient, error) {
 	client := &CephClient{
 		monHost: monHost,
@@ -283,12 +289,50 @@ keyring = %s
 	return client, nil
 }
 
+func CephConfString(monHost, key string, radosMonOpTimeout, radosOsdOpTimeout, clientMountTimeout int64) string {
+	conf := []string{}
+	conf = append(conf, "mon_host="+strings.ReplaceAll(monHost, ",", `\;`))
+	if len(key) > 0 {
+		for _, k := range []string{":", "@", "="} {
+			key = strings.ReplaceAll(key, k, fmt.Sprintf(`\%s`, k))
+		}
+		conf = append(conf, "key="+key)
+	}
+	for k, timeout := range map[string]int64{
+		"rados_mon_op_timeout": radosMonOpTimeout,
+		"rados_osd_op_timeout": radosOsdOpTimeout,
+		"client_mount_timeout": clientMountTimeout,
+	} {
+		conf = append(conf, fmt.Sprintf("%s=%d", k, timeout))
+	}
+	return ":" + strings.Join(conf, ":")
+}
+
 func (cli *CephClient) Child(pool string) *CephClient {
 	newCli := *cli
 	newCli.pool = pool
 	return &newCli
 }
 
+type SFsid struct {
+	Fsid string
+}
+
+func (cli *CephClient) Fsid() (string, error) {
+	opts := cli.options()
+	opts = append(opts, "fsid")
+	resp, err := cli.output("ceph", opts, true)
+	if err != nil {
+		return "", err
+	}
+	fsid := SFsid{}
+	err = resp.Unmarshal(&fsid)
+	if err != nil {
+		return "", err
+	}
+	return fsid.Fsid, nil
+}
+
 type SImage struct {
 	name   string
 	client *CephClient