Cloudpods/backend/pkg/compute/models/disks.go

// 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"
	"database/sql"
	"fmt"
	"path"
	"path/filepath"
	"strings"
	"time"

	"yunion.io/x/cloudmux/pkg/cloudprovider"
	"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/fileutils"
	"yunion.io/x/pkg/util/pinyinutils"
	"yunion.io/x/pkg/util/rbacscope"
	"yunion.io/x/pkg/utils"
	"yunion.io/x/sqlchemy"

	"yunion.io/x/onecloud/pkg/apis"
	billing_api "yunion.io/x/onecloud/pkg/apis/billing"
	api "yunion.io/x/onecloud/pkg/apis/compute"
	imageapi "yunion.io/x/onecloud/pkg/apis/image"
	schedapi "yunion.io/x/onecloud/pkg/apis/scheduler"
	"yunion.io/x/onecloud/pkg/cloudcommon/db"
	"yunion.io/x/onecloud/pkg/cloudcommon/db/lockman"
	"yunion.io/x/onecloud/pkg/cloudcommon/db/quotas"
	"yunion.io/x/onecloud/pkg/cloudcommon/db/taskman"
	"yunion.io/x/onecloud/pkg/cloudcommon/notifyclient"
	"yunion.io/x/onecloud/pkg/cloudcommon/validators"
	"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/mcclient/modules/image"
	"yunion.io/x/onecloud/pkg/util/logclient"
	"yunion.io/x/onecloud/pkg/util/stringutils2"
)

type SDiskManager struct {
	db.SVirtualResourceBaseManager
	db.SExternalizedResourceBaseManager
	SStorageResourceBaseManager
	SBillingResourceBaseManager
	db.SMultiArchResourceBaseManager
	db.SAutoDeleteResourceBaseManager
	db.SEncryptedResourceManager
}

var DiskManager *SDiskManager

func init() {
	DiskManager = &SDiskManager{
		SVirtualResourceBaseManager: db.NewVirtualResourceBaseManager(
			SDisk{},
			"disks_tbl",
			"disk",
			"disks",
		),
	}
	DiskManager.SetVirtualObject(DiskManager)
	DiskManager.TableSpec().AddIndex(false, "deleted", "disk_size", "status", "storage_id")
}

type SDisk struct {
	db.SVirtualResourceBase
	db.SExternalizedResourceBase

	SBillingResourceBase
	SStorageResourceBase `width:"128" charset:"ascii" nullable:"true" list:"admin" create:"optional"`
	db.SMultiArchResourceBase
	db.SAutoDeleteResourceBase

	db.SEncryptedResource

	// 磁盘存储类型
	// example: qcow2
	DiskFormat string `width:"32" charset:"ascii" nullable:"false" default:"qcow2" list:"user" json:"disk_format"`
	// 磁盘大小, 单位Mb
	// example: 10240
	DiskSize int `nullable:"false" list:"user" json:"disk_size"`
	// 磁盘路径
	AccessPath string `width:"256" charset:"utf8" nullable:"true" get:"user" json:"access_path"`
	PCIPath    string `width:"256" charset:"utf8" nullable:"true" get:"user" json:"pci_path"`

	// 存储Id
	// StorageId       string `width:"128" charset:"ascii" nullable:"true" list:"admin" create:"optional"`

	// 备份磁盘实例的存储ID
	BackupStorageId string `width:"128" charset:"ascii" nullable:"true" list:"admin" json:"backup_storage_id"`

	// 镜像Id
	TemplateId string `width:"256" charset:"ascii" nullable:"true" list:"user" json:"template_id"`
	// 快照Id
	SnapshotId string `width:"256" charset:"ascii" nullable:"true" list:"user" json:"snapshot_id"`
	// 备份Id
	BackupId string `width:"256" charset:"ascii" nullable:"true" list:"user" json:"backup_id"`

	// 设备名称
	Device string `width:"32" charset:"ascii" nullable:"true" get:"user" list:"user"`

	// 文件系统
	FsFormat string `width:"32" charset:"ascii" nullable:"true" list:"user" json:"fs_format"`
	// 文件系统特性
	FsFeatures *api.DiskFsFeatures `length:"medium" nullable:"true" list:"user" json:"fs_features"`

	// 磁盘类型
	// sys: 系统盘
	// data: 数据盘
	// swap: 交换盘
	// example: sys
	DiskType string `width:"32" charset:"ascii" nullable:"true" list:"user" update:"admin" json:"disk_type"`

	// 预分配策略
	// off: 关闭预分配，默认关闭
	// metadata: 精简制备
	// falloc: 厚制制备延迟置零
	// full: 厚制备快速置零
	Preallocation string `width:"12" default:"off" charset:"ascii" nullable:"true" list:"user" update:"admin" json:"preallocation"`
	// # is persistent
	Nonpersistent bool `default:"false" list:"user" json:"nonpersistent"`
	// auto reset disk after guest shutdown
	AutoReset bool `default:"false" list:"user" update:"user" json:"auto_reset"`

	// 是否标记为SSD磁盘
	IsSsd bool `nullable:"false" default:"false" list:"user" update:"user" create:"optional"`

	// 最大连接数
	Iops int `nullable:"true" list:"user" create:"optional"`

	// 磁盘吞吐量
	Throughput int `nullable:"true" list:"user" create:"optional"`
}

func (manager *SDiskManager) GetContextManagers() [][]db.IModelManager {
	return [][]db.IModelManager{
		{StorageManager},
	}
}

func (manager *SDiskManager) FetchDiskById(diskId string) *SDisk {
	disk, err := manager.FetchById(diskId)
	if err != nil {
		log.Errorf("FetchById fail %s", err)
		return nil
	}
	return disk.(*SDisk)
}

// 磁盘列表
func (manager *SDiskManager) ListItemFilter(
	ctx context.Context,
	q *sqlchemy.SQuery,
	userCred mcclient.TokenCredential,
	query api.DiskListInput,
) (*sqlchemy.SQuery, error) {
	var err error

	q, err = manager.SStorageResourceBaseManager.ListItemFilter(ctx, q, userCred, query.StorageFilterListInput)
	if err != nil {
		return nil, errors.Wrap(err, "SStorageResourceBaseManager.ListItemFilter")
	}

	q, err = manager.SExternalizedResourceBaseManager.ListItemFilter(ctx, q, userCred, query.ExternalizedResourceBaseListInput)
	if err != nil {
		return nil, errors.Wrap(err, "SExternalizedResourceBaseManager.ListItemFilter")
	}

	q, err = manager.SBillingResourceBaseManager.ListItemFilter(ctx, q, userCred, query.BillingResourceListInput)
	if err != nil {
		return nil, errors.Wrap(err, "SBillingResourceBaseManager.ListItemFilter")
	}

	q, err = manager.SVirtualResourceBaseManager.ListItemFilter(ctx, q, userCred, query.VirtualResourceListInput)
	if err != nil {
		return nil, errors.Wrap(err, "SVirtualResourceBaseManager.ListItemFilter")
	}

	q, err = manager.SMultiArchResourceBaseManager.ListItemFilter(ctx, q, userCred, query.MultiArchResourceBaseListInput)
	if err != nil {
		return nil, errors.Wrap(err, "SMultiArchResourceBaseManager.ListItemFilter")
	}

	q, err = manager.SAutoDeleteResourceBaseManager.ListItemFilter(ctx, q, userCred, query.AutoDeleteResourceBaseListInput)
	if err != nil {
		return nil, errors.Wrapf(err, "SAutoDeleteResourceBaseManager.ListItemFilter")
	}

	if query.Unused != nil {
		guestdisks := GuestdiskManager.Query().SubQuery()
		sq := guestdisks.Query(guestdisks.Field("disk_id"))
		if *query.Unused {
			q = q.Filter(sqlchemy.NotIn(q.Field("id"), sq))
		} else {
			q = q.Filter(sqlchemy.In(q.Field("id"), sq))
		}
	}

	if query.BindingSnapshotpolicy != nil {
		spjsq := SnapshotPolicyResourceManager.Query("resource_id").Equals("resource_type", api.SNAPSHOT_POLICY_TYPE_DISK).SubQuery()
		if *query.BindingSnapshotpolicy {
			q = q.In("id", spjsq)
		} else {
			q = q.NotIn("id", spjsq)
		}
	}
	if query.BindingServerSnapshotpolicy != nil {
		guestDisks := GuestdiskManager.Query("disk_id")
		sq := SnapshotPolicyResourceManager.Query("resource_id").Equals("resource_type", api.SNAPSHOT_POLICY_TYPE_SERVER).SubQuery()
		gdsq := guestDisks.Join(sq, sqlchemy.Equals(guestDisks.Field("guest_id"), sq.Field("resource_id"))).SubQuery()
		if *query.BindingServerSnapshotpolicy {
			q = q.In("id", gdsq)
		} else {
			q = q.NotIn("id", gdsq)
		}
	}

	guestId := query.ServerId
	if len(guestId) > 0 {
		guests := GuestManager.Query("id")
		sq := guests.Filter(
			sqlchemy.OR(
				sqlchemy.In(guests.Field("id"), guestId),
				sqlchemy.In(guests.Field("name"), guestId),
			),
		).SubQuery()
		guestDisks := GuestdiskManager.Query().SubQuery()
		q = q.Join(guestDisks, sqlchemy.AND(
			sqlchemy.Equals(guestDisks.Field("disk_id"), q.Field("id")),
			sqlchemy.In(guestDisks.Field("guest_id"), sq),
		)).Asc(guestDisks.Field("index"))
	}

	if diskType := query.DiskType; diskType != "" {
		q = q.Filter(sqlchemy.Equals(q.Field("disk_type"), diskType))
	}

	if len(query.SnapshotpolicyId) > 0 {
		_, err := validators.ValidateModel(ctx, userCred, SnapshotPolicyManager, &query.SnapshotpolicyId)
		if err != nil {
			return nil, err
		}
		sq := SnapshotPolicyResourceManager.Query("resource_id").Equals("resource_type", api.SNAPSHOT_POLICY_TYPE_DISK).Equals("snapshotpolicy_id", query.SnapshotpolicyId).SubQuery()
		q = q.In("id", sq)
	}

	if len(query.DiskFormat) > 0 {
		q = q.Equals("disk_format", query.DiskFormat)
	}

	if query.DiskSize > 0 {
		q = q.Equals("disk_size", query.DiskSize)
	}

	if len(query.FsFormat) > 0 {
		q = q.Equals("fs_format", query.FsFormat)
	}

	if len(query.ImageId) > 0 {
		img, err := CachedimageManager.getImageInfo(ctx, userCred, query.ImageId, false)
		if err != nil {
			return nil, errors.Wrap(err, "CachedimageManager.getImageInfo")
		}
		q = q.Equals("template_id", img.Id)
	}

	if len(query.SnapshotId) > 0 {
		_, err := validators.ValidateModel(ctx, userCred, SnapshotManager, &query.SnapshotId)
		if err != nil {
			return nil, err
		}
		q = q.Equals("snapshot_id", query.SnapshotId)
	}

	if len(query.GuestStatus) > 0 {
		guests := GuestManager.Query("id").Equals("status", query.GuestStatus).SubQuery()
		sq := GuestdiskManager.Query().In("guest_id", guests).SubQuery()
		q = q.Join(sq, sqlchemy.Equals(sq.Field("disk_id"), q.Field("id")))
	}

	return q, nil
}

func (manager *SDiskManager) OrderByExtraFields(
	ctx context.Context,
	q *sqlchemy.SQuery,
	userCred mcclient.TokenCredential,
	query api.DiskListInput,
) (*sqlchemy.SQuery, error) {
	var err error

	q, err = manager.SStorageResourceBaseManager.OrderByExtraFields(ctx, q, userCred, query.StorageFilterListInput)
	if err != nil {
		return nil, errors.Wrap(err, "SStorageResourceBaseManager.OrderByExtraFields")
	}

	q, err = manager.SBillingResourceBaseManager.OrderByExtraFields(ctx, q, userCred, query.BillingResourceListInput)
	if err != nil {
		return nil, errors.Wrap(err, "SBillingResourceBaseManager.OrderByExtraFields")
	}
	if db.NeedOrderQuery([]string{query.OrderByServer}) {
		guestDiskQuery := GuestdiskManager.Query("disk_id", "guest_id").SubQuery()
		q = q.LeftJoin(guestDiskQuery, sqlchemy.Equals(q.Field("id"), guestDiskQuery.Field("disk_id")))
		guestQuery := GuestManager.Query().SubQuery()
		q.AppendField(q.QueryFields()...)
		q.AppendField(guestQuery.Field("name", "guest_name"))
		q.Join(guestQuery, sqlchemy.Equals(guestQuery.Field("id"), guestDiskQuery.Field("guest_id")))
		db.OrderByFields(q, []string{query.OrderByServer}, []sqlchemy.IQueryField{guestQuery.Field("name")})
	}
	if db.NeedOrderQuery([]string{query.OrderByGuestCount}) {
		guestdisks := GuestdiskManager.Query().SubQuery()
		disks := DiskManager.Query().SubQuery()
		guestdiskQ := guestdisks.Query(
			guestdisks.Field("guest_id"),
			guestdisks.Field("disk_id"),
			sqlchemy.COUNT("guest_count", guestdisks.Field("guest_id")),
		)

		guestdiskQ = guestdiskQ.LeftJoin(disks, sqlchemy.Equals(guestdiskQ.Field("disk_id"), disks.Field("id")))
		guestdiskSQ := guestdiskQ.GroupBy(guestdiskQ.Field("disk_id")).SubQuery()
		q.AppendField(q.QueryFields()...)
		q.AppendField(guestdiskSQ.Field("guest_count"))
		q = q.LeftJoin(guestdiskSQ, sqlchemy.Equals(q.Field("id"), guestdiskSQ.Field("disk_id")))
		db.OrderByFields(q, []string{query.OrderByGuestCount}, []sqlchemy.IQueryField{guestdiskQ.Field("guest_count")})
	}
	q, err = manager.SVirtualResourceBaseManager.OrderByExtraFields(ctx, q, userCred, query.VirtualResourceListInput)
	if err != nil {
		return nil, errors.Wrap(err, "SVirtualResourceBaseManager.OrderByExtraFields")
	}

	return q, nil
}

func (manager *SDiskManager) QueryDistinctExtraField(q *sqlchemy.SQuery, field string) (*sqlchemy.SQuery, error) {
	var err error

	q, err = manager.SVirtualResourceBaseManager.QueryDistinctExtraField(q, field)
	if err == nil {
		return q, nil
	}
	if field == "guest_status" {
		guestDiskQuery := GuestdiskManager.Query("disk_id", "guest_id").SubQuery()
		q = q.LeftJoin(guestDiskQuery, sqlchemy.Equals(q.Field("id"), guestDiskQuery.Field("disk_id")))
		guestQuery := GuestManager.Query().SubQuery()
		q.AppendField(guestQuery.Field("status", field)).Distinct()
		q.Join(guestQuery, sqlchemy.Equals(guestQuery.Field("id"), guestDiskQuery.Field("guest_id")))
		return q, nil
	}
	if field == "server" {
		guestDiskQuery := GuestdiskManager.Query("disk_id", "guest_id").SubQuery()
		q = q.LeftJoin(guestDiskQuery, sqlchemy.Equals(q.Field("id"), guestDiskQuery.Field("disk_id")))
		guestQuery := GuestManager.Query().SubQuery()
		q.AppendField(guestQuery.Field("name", field)).Distinct()
		q.Join(guestQuery, sqlchemy.Equals(guestQuery.Field("id"), guestDiskQuery.Field("guest_id")))
		return q, nil
	}
	q, err = manager.SStorageResourceBaseManager.QueryDistinctExtraField(q, field)
	if err == nil {
		return q, nil
	}

	return q, httperrors.ErrNotFound
}

func (manager *SDiskManager) QueryDistinctExtraFields(q *sqlchemy.SQuery, resource string, fields []string) (*sqlchemy.SQuery, error) {
	switch resource {
	case GuestManager.Keyword():

		guestdisks := GuestdiskManager.Query("disk_id", "guest_id").SubQuery()
		q = q.LeftJoin(guestdisks, sqlchemy.Equals(q.Field("id"), guestdisks.Field("disk_id")))

		guestQuery := GuestManager.Query().SubQuery()
		for _, field := range fields {
			q = q.AppendField(guestQuery.Field(field))
		}
		q = q.Join(guestQuery, sqlchemy.Equals(q.Field("guest_id"), guestQuery.Field("id")))
		return q, nil
	}
	return q, httperrors.ErrNotFound
}

func (disk *SDisk) GetGuestDiskQuery() *sqlchemy.SQuery {
	guestdisks := GuestdiskManager.Query()
	guests := GuestManager.Query().SubQuery()
	guestdisks = guestdisks.Join(guests, sqlchemy.Equals(guestdisks.Field("guest_id"), guests.Field("id")))
	return guestdisks.Equals("disk_id", disk.Id)
}

func (self *SDisk) GetGuestDiskCount() (int, error) {
	return self.GetGuestDiskQuery().CountWithError()
}

func (disk *SDisk) GetGuestDisk() (*SGuestdisk, error) {
	guestdisk := &SGuestdisk{}
	err := disk.GetGuestDiskQuery().First(guestdisk)
	if err != nil {
		return nil, errors.Wrap(err, "First")
	}
	return guestdisk, nil
}

func (self *SDisk) isAttached() (bool, error) {
	cnt, err := self.GetGuestDiskCount()
	if err != nil {
		return false, err
	}
	return cnt > 0, nil
}

func (self *SDisk) GetGuestdisks() []SGuestdisk {
	guestdisks := make([]SGuestdisk, 0)
	q := GuestdiskManager.Query().Equals("disk_id", self.Id)
	err := db.FetchModelObjects(GuestdiskManager, q, &guestdisks)
	if err != nil {
		log.Errorf("%s", err)
		return nil
	}
	return guestdisks
}

func (self *SDisk) GetGuests() []SGuest {
	result := make([]SGuest, 0)
	query := GuestManager.Query()
	guestdisks := GuestdiskManager.Query().SubQuery()
	q := query.Join(guestdisks, sqlchemy.AND(
		sqlchemy.Equals(guestdisks.Field("guest_id"), query.Field("id")))).
		Filter(sqlchemy.Equals(guestdisks.Field("disk_id"), self.Id))
	err := db.FetchModelObjects(GuestManager, q, &result)
	if err != nil {
		log.Errorln(err)
		return nil
	}
	return result
}

func (self *SDisk) GetGuest() *SGuest {
	guests := self.GetGuests()
	if len(guests) > 0 {
		return &guests[0]
	}
	return nil
}

func (self *SDisk) GetGuestsCount() (int, error) {
	guests := GuestManager.Query().SubQuery()
	guestdisks := GuestdiskManager.Query().SubQuery()
	return guests.Query().Join(guestdisks, sqlchemy.AND(
		sqlchemy.Equals(guestdisks.Field("guest_id"), guests.Field("id")))).
		Filter(sqlchemy.Equals(guestdisks.Field("disk_id"), self.Id)).CountWithError()
}

func (self *SDisk) GetRuningGuestCount() (int, error) {
	guests := GuestManager.Query().SubQuery()
	guestdisks := GuestdiskManager.Query().SubQuery()
	return guests.Query().Join(guestdisks, sqlchemy.AND(
		sqlchemy.Equals(guestdisks.Field("guest_id"), guests.Field("id")))).
		Filter(sqlchemy.Equals(guestdisks.Field("disk_id"), self.Id)).
		Filter(sqlchemy.Equals(guests.Field("status"), api.VM_RUNNING)).CountWithError()
}

func (self *SDisk) CustomizeCreate(ctx context.Context, userCred mcclient.TokenCredential, ownerId mcclient.IIdentityProvider, query jsonutils.JSONObject, data jsonutils.JSONObject) error {
	input := new(api.DiskCreateInput)
	if err := data.Unmarshal(input); err != nil {
		return errors.Wrap(err, "Unmarshal json")
	}
	if err := self.fetchDiskInfo(input.DiskConfig); err != nil {
		return errors.Wrap(err, "fetch disk info")
	}
	err := self.SEncryptedResource.CustomizeCreate(ctx, userCred, ownerId, data, "disk-"+pinyinutils.Text2Pinyin(self.Name))
	if err != nil {
		return errors.Wrap(err, "SEncryptedResource.CustomizeCreate")
	}
	return self.SVirtualResourceBase.CustomizeCreate(ctx, userCred, ownerId, query, data)
}

func (self *SDisk) ValidateUpdateData(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject, input *api.DiskUpdateInput) (*api.DiskUpdateInput, error) {
	var err error

	if input.DiskType != "" {
		if !utils.IsInStringArray(input.DiskType, []string{api.DISK_TYPE_DATA, api.DISK_TYPE_VOLUME, api.DISK_TYPE_SYS}) {
			return input, httperrors.NewInputParameterError("not support update disk_type %s", input.DiskType)
		}
	}

	if input.AutoReset != nil && *input.AutoReset != self.AutoReset {
		if guest := self.GetGuest(); guest != nil && guest.Status != api.VM_READY {
			return input, httperrors.NewBadRequestError("Can't set disk auto_reset on guest status %s", guest.Status)
		}
	}

	storage, _ := self.GetStorage()
	if storage == nil {
		return input, httperrors.NewNotFoundError("failed to find storage for disk %s", self.Name)
	}

	host, err := storage.GetMasterHost()
	if err != nil {
		return nil, errors.Wrapf(err, "GetMasterHost")
	}

	driver, err := host.GetHostDriver()
	if err != nil {
		return nil, errors.Wrapf(err, "GetHostDriver")
	}

	input, err = driver.ValidateUpdateDisk(ctx, userCred, input)
	if err != nil {
		return input, errors.Wrap(err, "GetHostDriver().ValidateUpdateDisk")
	}

	input.VirtualResourceBaseUpdateInput, err = self.SVirtualResourceBase.ValidateUpdateData(ctx, userCred, query, input.VirtualResourceBaseUpdateInput)
	if err != nil {
		return input, errors.Wrap(err, "SVirtualResourceBase.ValidateUpdateData")
	}

	return input, nil
}

func (man *SDiskManager) BatchCreateValidateCreateData(ctx context.Context, userCred mcclient.TokenCredential, ownerId mcclient.IIdentityProvider, query jsonutils.JSONObject, input api.DiskCreateInput) (*jsonutils.JSONDict, error) {
	input, err := man.ValidateCreateData(ctx, userCred, ownerId, query, input)
	if err != nil {
		return nil, err
	}
	return input.JSON(input), nil
}

func diskCreateInput2ComputeQuotaKeys(input api.DiskCreateInput, ownerId mcclient.IIdentityProvider) (SComputeResourceKeys, error) {
	var keys SComputeResourceKeys
	if len(input.PreferHost) > 0 {
		hostObj, err := HostManager.FetchById(input.PreferHost)
		if err != nil {
			return keys, err
		}
		host := hostObj.(*SHost)
		input.PreferZone = host.ZoneId
		keys.ZoneId = host.ZoneId
	}
	if len(input.PreferWire) > 0 {
		wireObj, err := WireManager.FetchById(input.PreferWire)
		if err != nil {
			return keys, err
		}
		wire := wireObj.(*SWire)
		if len(wire.ZoneId) > 0 {
			input.PreferZone = wire.ZoneId
			keys.ZoneId = wire.ZoneId
		}
	}
	if len(input.PreferZone) > 0 {
		zoneObj, err := ZoneManager.FetchById(input.PreferZone)
		if err != nil {
			return keys, err
		}
		zone := zoneObj.(*SZone)
		input.PreferRegion = zone.CloudregionId
		keys.ZoneId = zone.Id
		keys.RegionId = zone.CloudregionId
	}
	if len(input.PreferRegion) > 0 {
		regionObj, err := CloudregionManager.FetchById(input.PreferRegion)
		if err != nil {
			return keys, err
		}
		region := regionObj.(*SCloudregion)
		keys.RegionId = region.GetId()
		keys.Brand = region.Provider
	}
	return keys, nil
}

func (manager *SDiskManager) ValidateCreateData(ctx context.Context, userCred mcclient.TokenCredential, ownerId mcclient.IIdentityProvider, query jsonutils.JSONObject, input api.DiskCreateInput) (api.DiskCreateInput, error) {
	diskConfig := input.DiskConfig
	diskConfig, err := parseDiskInfo(ctx, userCred, diskConfig)
	if err != nil {
		return input, err
	}
	if input.ExistingPath != "" && input.Storage == "" {
		return input, httperrors.NewInputParameterError("disk create from existing disk must give storage")
	}

	input.ProjectId = ownerId.GetProjectId()
	input.ProjectDomainId = ownerId.GetProjectDomainId()

	var quotaKey quotas.IQuotaKeys

	storageID := input.Storage

	if storageID != "" {
		storageObj, err := StorageManager.FetchByIdOrName(ctx, nil, storageID)
		if err != nil {
			return input, httperrors.NewResourceNotFoundError("Storage %s not found", storageID)
		}
		storage := storageObj.(*SStorage)

		provider := storage.GetCloudprovider()
		if provider != nil && !provider.IsAvailable() {
			return input, httperrors.NewResourceNotReadyError("cloudprovider %s not available", provider.Name)
		}

		host, err := storage.GetMasterHost()
		if err != nil {
			return input, errors.Wrapf(err, "GetMasterHost")
		}
		hostDriver, err := host.GetHostDriver()
		if err != nil {
			return input, errors.Wrapf(err, "GetHostDriver")
		}
		input.Hypervisor = hostDriver.GetHypervisor()
		if len(diskConfig.Backend) == 0 {
			diskConfig.Backend = storage.StorageType
		}
		err = manager.validateDiskOnStorage(diskConfig, storage)
		if err != nil {
			return input, err
		}
		input.Storage = storage.Id

		zone, _ := storage.getZone()
		quotaKey = fetchComputeQuotaKeys(
			rbacscope.ScopeProject,
			ownerId,
			zone,
			provider,
			input.Hypervisor,
		)

	} else {
		if len(diskConfig.Backend) == 0 {
			diskConfig.Backend = api.STORAGE_LOCAL
		}
		if len(input.PreferManager) > 0 {
			_manager, err := CloudproviderManager.FetchByIdOrName(ctx, userCred, input.PreferManager)
			if err != nil {
				if errors.Cause(err) == sql.ErrNoRows {
					return input, httperrors.NewResourceNotFoundError2("cloudprovider", input.PreferManager)
				}
				return input, httperrors.NewGeneralError(err)
			}
			manager := _manager.(*SCloudprovider)
			if !manager.IsAvailable() {
				return input, httperrors.NewResourceNotReadyError("cloudprovider %s not available", manager.Name)
			}
			input.PreferManager = manager.Id
		}
		serverInput, err := ValidateScheduleCreateData(ctx, userCred, input.ToServerCreateInput(), input.Hypervisor)
		if err != nil {
			return input, err
		}
		// preserve encrypt info
		encInput := input.EncryptedResourceCreateInput
		input = *serverInput.ToDiskCreateInput()
		input.EncryptedResourceCreateInput = encInput
		quotaKey, _ = diskCreateInput2ComputeQuotaKeys(input, ownerId)
	}

	input.VirtualResourceCreateInput, err = manager.SVirtualResourceBaseManager.ValidateCreateData(ctx, userCred, ownerId, query, input.VirtualResourceCreateInput)
	if err != nil {
		return input, errors.Wrap(err, "SVirtualResourceBaseManager.ValidateCreateData")
	}
	input.EncryptedResourceCreateInput, err = manager.SEncryptedResourceManager.ValidateCreateData(ctx, userCred, ownerId, query, input.EncryptedResourceCreateInput)
	if err != nil {
		return input, errors.Wrap(err, "SEncryptedResourceManager.ValidateCreateData")
	}
	if err := manager.ValidateFsFeatures(input.Fs, input.FsFeatures); err != nil {
		return input, err
	}

	pendingUsage := SQuota{Storage: diskConfig.SizeMb}
	pendingUsage.SetKeys(quotaKey)
	if err := quotas.CheckSetPendingQuota(ctx, userCred, &pendingUsage); err != nil {
		return input, httperrors.NewOutOfQuotaError("%s", err)
	}
	return input, nil
}

func (manager *SDiskManager) ValidateFsFeatures(fsType string, feature *api.DiskFsFeatures) error {
	if feature == nil {
		return nil
	}
	if feature.Ext4 != nil {
		if fsType != "ext4" {
			return httperrors.NewInputParameterError("only ext4 fs can set fs_features.ext4, current is %q", fsType)
		}
		if feature.Ext4.ReservedBlocksPercentage < 0 || feature.Ext4.ReservedBlocksPercentage >= 100 {
			return httperrors.NewInputParameterError("ext4.reserved_blocks_percentage must in range [1, 99]")
		}
	}
	if feature.F2fs != nil {
		if fsType != "f2fs" {
			return httperrors.NewInputParameterError("only f2fs fs can set fs_features.f2fs, current is %q", fsType)
		}
		if feature.F2fs.OverprovisionRatioPercentage < 0 || feature.F2fs.OverprovisionRatioPercentage >= 100 {
			return httperrors.NewInputParameterError("f2fs.reserved_blocks_percentage must in range [1, 99]")
		}
	}
	return nil
}

func (manager *SDiskManager) validateDiskOnStorage(diskConfig *api.DiskConfig, storage *SStorage) error {
	if storage.Enabled.IsFalse() {
		return httperrors.NewInputParameterError("Cannot create disk with disabled storage[%s]", storage.Name)
	}
	if !utils.IsInStringArray(storage.Status, []string{api.STORAGE_ENABLED, api.STORAGE_ONLINE}) {
		return httperrors.NewInputParameterError("Cannot create disk with offline storage[%s]", storage.Name)
	}
	if storage.StorageType != diskConfig.Backend {
		return httperrors.NewInputParameterError("Storage type[%s] not match backend %s", storage.StorageType, diskConfig.Backend)
	}
	if diskConfig.ExistingPath != "" {
		if !utils.IsInStringArray(storage.StorageType, api.FIEL_STORAGE) {
			return httperrors.NewInputParameterError(
				"Disk create from existing path, unsupport storage type %s", storage.StorageType)
		}
	}

	var guestdriver IGuestDriver = nil
	if host, _ := storage.GetMasterHost(); host != nil {
		//公有云磁盘大小检查。
		hostDriver, err := host.GetHostDriver()
		if err != nil {
			return errors.Wrapf(err, "GetHostDriver")
		}
		if err := hostDriver.ValidateDiskSize(storage, diskConfig.SizeMb>>10); err != nil {
			return httperrors.NewInputParameterError("%v", err)
		}

		guestdriver, _ = GetDriver(hostDriver.GetHypervisor(), hostDriver.GetProvider())
	}
	hoststorages := HoststorageManager.Query().SubQuery()
	hoststorage := make([]SHoststorage, 0)
	if err := hoststorages.Query().Equals("storage_id", storage.Id).All(&hoststorage); err != nil {
		return err
	}
	if len(hoststorage) == 0 {
		return httperrors.NewInputParameterError("Storage[%s] must attach to a host", storage.Name)
	}
	if guestdriver == nil || guestdriver.DoScheduleStorageFilter() {
		if int64(diskConfig.SizeMb) > storage.GetFreeCapacity() && !storage.IsEmulated {
			return httperrors.NewInputParameterError("Not enough free space")
		}
	}
	return nil
}

func (disk *SDisk) SetStorage(storageId string, diskConfig *api.DiskConfig) error {
	backend := diskConfig.Backend
	if backend == "" {
		return fmt.Errorf("Backend is empty")
	}
	storage := StorageManager.FetchStorageById(storageId)
	if storage == nil {
		return fmt.Errorf("Not found backend %s storage %s", backend, storageId)
	}
	err := DiskManager.validateDiskOnStorage(diskConfig, storage)
	if err != nil {
		return err
	}
	_, err = db.Update(disk, func() error {
		disk.StorageId = storage.Id
		return nil
	})
	return err
}

func (disk *SDisk) SetStorageByHost(hostId string, diskConfig *api.DiskConfig, storageIds []string) error {
	host := HostManager.FetchHostById(hostId)
	backend := diskConfig.Backend
	if backend == "" {
		return fmt.Errorf("Backend is empty")
	}
	var storage *SStorage
	if len(storageIds) != 0 {
		storage = StorageManager.FetchStorageById(storageIds[0])
	} else if utils.IsInStringArray(backend, api.STORAGE_LIMITED_TYPES) {
		storage = host.GetLeastUsedStorage(backend)
	} else {
		// unlimited pulic cloud storages
		storages := host.GetAttachedEnabledHostStorages(nil)
		for _, s := range storages {
			if s.StorageType == backend {
				tmpS := s
				storage = &tmpS
			}
		}
	}
	if storage == nil {
		return fmt.Errorf("Not found host %s backend %s storage", host.Name, backend)
	}
	err := DiskManager.validateDiskOnStorage(diskConfig, storage)
	if err != nil {
		return err
	}
	_, err = db.Update(disk, func() error {
		disk.StorageId = storage.Id
		disk.IsSsd = (storage.MediumType == api.DISK_TYPE_SSD)
		return nil
	})
	return err
}

func getDiskResourceRequirements(ctx context.Context, userCred mcclient.TokenCredential, ownerId mcclient.IIdentityProvider, input api.DiskCreateInput, count int) SQuota {
	req := SQuota{
		Storage: input.SizeMb * count,
	}
	var quotaKey SComputeResourceKeys
	if len(input.Storage) > 0 {
		storageObj, _ := StorageManager.FetchById(input.Storage)
		storage := storageObj.(*SStorage)
		zone, _ := storage.getZone()
		quotaKey = fetchComputeQuotaKeys(
			rbacscope.ScopeProject,
			ownerId,
			zone,
			storage.GetCloudprovider(),
			input.Hypervisor,
		)
	} else {
		quotaKey, _ = diskCreateInput2ComputeQuotaKeys(input, ownerId)
	}
	req.SetKeys(quotaKey)
	return req
}

func (disk *SDisk) OnMetadataUpdated(ctx context.Context, userCred mcclient.TokenCredential) {
	if len(disk.ExternalId) == 0 || options.Options.KeepTagLocalization {
		return
	}
	err := disk.StartRemoteUpdateTask(ctx, userCred, true, "")
	if err != nil {
		log.Errorf("StartRemoteUpdateTask fail: %s", err)
	}
}

func (disk *SDisk) StartRemoteUpdateTask(ctx context.Context, userCred mcclient.TokenCredential, replaceTags bool, parentTaskId string) error {
	data := jsonutils.NewDict()
	if replaceTags {
		data.Add(jsonutils.JSONTrue, "replace_tags")
	}
	task, err := taskman.TaskManager.NewTask(ctx, "DiskRemoteUpdateTask", disk, userCred, data, parentTaskId, "", nil)
	if err != nil {
		return errors.Wrap(err, "Start DiskRemoteUpdateTask")
	}
	disk.SetStatus(ctx, userCred, apis.STATUS_UPDATE_TAGS, "StartRemoteUpdateTask")
	return task.ScheduleRun(nil)
}

func (disk *SDisk) PostCreate(ctx context.Context, userCred mcclient.TokenCredential, ownerId mcclient.IIdentityProvider, query jsonutils.JSONObject, data jsonutils.JSONObject) {
	disk.SVirtualResourceBase.PostCreate(ctx, userCred, ownerId, query, data)
	input := api.DiskCreateInput{}
	err := data.Unmarshal(&input)
	if err != nil {
		log.Errorf("!!!data.Unmarshal api.DiskCreateInput fail %s", err)
	}
	if input.ExistingPath != "" {
		disk.SetMetadata(ctx, api.DISK_META_EXISTING_PATH, input.ExistingPath, userCred)
	}
}

func (manager *SDiskManager) OnCreateComplete(ctx context.Context, items []db.IModel, userCred mcclient.TokenCredential, ownerId mcclient.IIdentityProvider, query jsonutils.JSONObject, data []jsonutils.JSONObject) {
	input := api.DiskCreateInput{}
	err := data[0].Unmarshal(&input)
	if err != nil {
		log.Errorf("!!!data.Unmarshal api.DiskCreateInput fail %s", err)
	}

	pendingUsage := getDiskResourceRequirements(ctx, userCred, ownerId, input, len(items))
	parentTaskId, _ := data[0].GetString("parent_task_id")
	RunBatchCreateTask(ctx, items, userCred, data, pendingUsage, SRegionQuota{}, "DiskBatchCreateTask", parentTaskId)
}

func (self *SDisk) StartDiskCreateTask(ctx context.Context, userCred mcclient.TokenCredential, rebuild bool, snapshot string, parentTaskId string) error {
	kwargs := jsonutils.NewDict()
	if rebuild {
		kwargs.Add(jsonutils.JSONTrue, "rebuild")
	}
	if len(snapshot) > 0 {
		kwargs.Add(jsonutils.NewString(snapshot), "snapshot")
	}

	taskName := "DiskCreateTask"
	if self.BackupStorageId != "" {
		taskName = "HADiskCreateTask"
	}
	if task, err := taskman.TaskManager.NewTask(ctx, taskName, self, userCred, kwargs, parentTaskId, "", nil); err != nil {
		return err
	} else {
		task.ScheduleRun(nil)
	}
	return nil
}

func (self *SDisk) GetSnapshotFuseUrl() (string, error) {
	snapObj, err := SnapshotManager.FetchById(self.SnapshotId)
	if err != nil {
		return "", errors.Wrapf(err, "SnapshotManager.FetchById(%s)", self.SnapshotId)
	}
	snapshot := snapObj.(*SSnapshot)
	return snapshot.GetFuseUrl()
}

func (self *SDisk) GetSnapshotCount() (int, error) {
	q := SnapshotManager.Query()
	return q.Filter(sqlchemy.AND(sqlchemy.Equals(q.Field("disk_id"), self.Id),
		sqlchemy.Equals(q.Field("fake_deleted"), false))).CountWithError()
}

func (self *SDisk) GetManualSnapshotCount() (int, error) {
	return SnapshotManager.Query().
		Equals("disk_id", self.Id).Equals("fake_deleted", false).
		Equals("created_by", api.SNAPSHOT_MANUAL).CountWithError()
}

func (self *SDisk) getDiskAllocateFromBackupInput(ctx context.Context, backupId string) (*api.DiskAllocateFromBackupInput, error) {
	ibackup, err := DiskBackupManager.FetchById(backupId)
	if err != nil {
		return nil, errors.Wrapf(err, "unable to get backup %s", backupId)
	}
	backup := ibackup.(*SDiskBackup)
	bs, err := backup.GetBackupStorage()
	if err != nil {
		return nil, errors.Wrapf(err, "unable to get backupstorage of backup %s", backupId)
	}
	accessInfo, err := bs.GetAccessInfo()
	if err != nil {
		return nil, errors.Wrap(err, "backupStorage.GetAccessInfo")
	}
	return &api.DiskAllocateFromBackupInput{
		BackupId:                backupId,
		BackupStorageId:         bs.GetId(),
		BackupStorageAccessInfo: jsonutils.Marshal(accessInfo).(*jsonutils.JSONDict),
		DiskConfig:              &backup.DiskConfig.DiskConfig,
		BackupAsTar:             backup.DiskConfig.BackupAsTar,
	}, nil
}

func (self *SDisk) StartAllocate(ctx context.Context, host *SHost, storage *SStorage, taskId string, userCred mcclient.TokenCredential, rebuild bool, snapshot string, task taskman.ITask) error {
	log.Infof("Allocating disk on host %s ...", host.GetName())

	templateId := self.GetTemplateId()
	fsFormat := self.GetFsFormat()

	input := api.DiskAllocateInput{
		Format:     self.DiskFormat,
		DiskSizeMb: self.DiskSize,
		SnapshotId: snapshot,
	}
	if self.BackupId != "" {
		allocateInput, err := self.getDiskAllocateFromBackupInput(ctx, self.BackupId)
		if err != nil {
			return errors.Wrap(err, "unable to getDiskAllocateFromBackupInput")
		}
		input.Backup = allocateInput
	}
	if len(snapshot) > 0 {
		if utils.IsInStringArray(storage.StorageType, api.FIEL_STORAGE) {
			SnapshotManager.AddRefCount(self.SnapshotId, 1)
			self.SetMetadata(ctx, "merge_snapshot", jsonutils.JSONTrue, userCred)
		}
	} else if len(templateId) > 0 {
		input.ImageId = templateId
		s := auth.GetAdminSession(ctx, options.Options.Region)
		img, err := image.Images.Get(s, templateId, nil)
		if err != nil {
			return errors.Wrapf(err, "get image details from glance")
		}
		input.ImageFormat, _ = img.GetString("disk_format")
	}
	if len(fsFormat) > 0 {
		input.FsFormat = fsFormat
		if self.FsFeatures != nil {
			input.FsFeatures = self.FsFeatures
		}
	}
	if self.IsEncrypted() {
		var err error
		input.Encryption = true
		input.EncryptInfo, err = self.GetEncryptInfo(ctx, userCred)
		if err != nil {
			return errors.Wrap(err, "GetEncryptInfo")
		}
	}
	if ePath := self.GetMetadata(ctx, api.DISK_META_EXISTING_PATH, userCred); ePath != "" {
		input.ExistingPath = ePath
	}

	driver, err := host.GetHostDriver()
	if err != nil {
		return errors.Wrapf(err, "GetHostDriver")
	}

	if rebuild {
		return driver.RequestRebuildDiskOnStorage(ctx, host, storage, self, task, input)
	}
	return driver.RequestAllocateDiskOnStorage(ctx, userCred, host, storage, self, task, input)
}

// make snapshot after reset out of chain
func (self *SDisk) CleanUpDiskSnapshots(ctx context.Context, userCred mcclient.TokenCredential, snapshot *SSnapshot) error {
	dest := make([]SSnapshot, 0)
	query := SnapshotManager.Query()
	query.Filter(sqlchemy.Equals(query.Field("disk_id"), self.Id)).
		GT("created_at", snapshot.CreatedAt).Asc("created_at").All(&dest)
	if len(dest) == 0 {
		return nil
	}
	convertSnapshots := jsonutils.NewArray()
	deleteSnapshots := jsonutils.NewArray()
	for i := 0; i < len(dest); i++ {
		if !dest[i].FakeDeleted && !dest[i].OutOfChain {
			convertSnapshots.Add(jsonutils.NewString(dest[i].Id))
		} else if dest[i].FakeDeleted {
			deleteSnapshots.Add(jsonutils.NewString(dest[i].Id))
		}
	}
	params := jsonutils.NewDict()
	params.Set("convert_snapshots", convertSnapshots)
	params.Set("delete_snapshots", deleteSnapshots)
	task, err := taskman.TaskManager.NewTask(ctx, "DiskCleanUpSnapshotsTask", self, userCred, params, "", "", nil)
	if err != nil {
		return err
	} else {
		task.ScheduleRun(nil)
	}
	return nil
}

func (self *SDisk) PerformDiskReset(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject, input *api.DiskResetInput) (jsonutils.JSONObject, error) {
	err := self.ValidateEncryption(ctx, userCred)
	if err != nil {
		return nil, errors.Wrap(err, "ValidateEncryption")
	}
	if !utils.IsInStringArray(self.Status, []string{api.DISK_READY}) {
		return nil, httperrors.NewInputParameterError("Cannot reset disk in status %s", self.Status)
	}
	storage, err := self.GetStorage()
	if err != nil {
		return nil, httperrors.NewGeneralError(errors.Wrapf(err, "GetStorage"))
	}

	host, err := storage.GetMasterHost()
	if err != nil {
		return nil, httperrors.NewGeneralError(errors.Wrapf(err, "GetMasterHost"))
	}

	snapshotObj, err := validators.ValidateModel(ctx, userCred, SnapshotManager, &input.SnapshotId)
	if err != nil {
		return nil, err
	}
	snapshot := snapshotObj.(*SSnapshot)
	if snapshot.Status != api.SNAPSHOT_READY {
		return nil, httperrors.NewBadRequestError("Cannot reset disk with snapshot in status %s", snapshot.Status)
	}

	if snapshot.DiskId != self.Id {
		return nil, httperrors.NewBadRequestError("Cannot reset disk %s(%s),Snapshot is belong to disk %s", self.Name, self.Id, snapshot.DiskId)
	}

	driver, err := host.GetHostDriver()
	if err != nil {
		return nil, errors.Wrapf(err, "GetHostDriver")
	}

	guests := self.GetGuests()
	input, err = driver.ValidateResetDisk(ctx, userCred, self, snapshot, guests, input)
	if err != nil {
		return nil, err
	}

	var guest *SGuest = nil
	if len(guests) > 0 {
		guest = &guests[0]
	}
	return nil, self.StartResetDisk(ctx, userCred, snapshot.Id, input.AutoStart, guest, "")
}

func (self *SDisk) validateMigrate(ctx context.Context, userCred mcclient.TokenCredential, input *api.DiskMigrateInput) error {
	hypervisor := self.getHypervisor()
	if !utils.IsInStringArray(hypervisor, []string{api.HYPERVISOR_KVM, api.HYPERVISOR_POD}) {
		return httperrors.NewNotAcceptableError("Not allow for hypervisor %s", hypervisor)
	}
	if guest := self.GetGuest(); guest != nil {
		return httperrors.NewBadRequestError("Disk attached guest, cannot migrate")
	}
	if input.TargetStorageId != "" {
		srcStorage, err := self.GetStorage()
		if err != nil {
			return errors.Wrap(err, "get src storage")
		}
		iDstStorage, err := StorageManager.FetchByIdOrName(ctx, userCred, input.TargetStorageId)
		if err != nil {
			return errors.Wrap(err, "get target storage")
		}
		if srcStorage.StorageType != iDstStorage.(*SStorage).StorageType {
			return httperrors.NewBadRequestError("Cannot migrate disk from storage type %s to %s", srcStorage.StorageType, iDstStorage.(*SStorage).StorageType)
		}
		input.TargetStorageId = iDstStorage.GetId()
	}
	return nil
}

func (self *SDisk) PerformMigrate(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject, input *api.DiskMigrateInput) (jsonutils.JSONObject, error) {
	err := self.validateMigrate(ctx, userCred, input)
	if err != nil {
		return nil, err
	}
	self.SetStatus(ctx, userCred, api.DISK_START_MIGRATE, "")
	params := jsonutils.NewDict()
	params.Set("target_storage_id", jsonutils.NewString(input.TargetStorageId))
	task, err := taskman.TaskManager.NewTask(ctx, "DiskMigrateTask", self, userCred, params, "", "", nil)
	if err != nil {
		return nil, errors.Wrapf(err, "NewTask")
	}
	return nil, task.ScheduleRun(nil)
}

func (self *SDisk) PerformChangeStorageType(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject, input *api.DiskChagneStorageTypeInput) (jsonutils.JSONObject, error) {
	if len(input.StorageType) == 0 {
		return nil, httperrors.NewMissingParameterError("storage_type")
	}
	storage, err := self.GetStorage()
	if err != nil {
		return nil, errors.Wrapf(err, "GetStorage")
	}
	if storage.StorageType == input.StorageType {
		return nil, httperrors.NewInputParameterError("Storage type is already %s", input.StorageType)
	}
	storages := []SStorage{}
	q := StorageManager.Query().Equals("zone_id", storage.ZoneId).Equals("storage_type", input.StorageType).IsTrue("enabled").Equals("status", api.STORAGE_ONLINE).Equals("manager_id", storage.ManagerId)
	err = q.All(&storages)
	if err != nil {
		return nil, errors.Wrapf(err, "CountWithError")
	}
	if len(storages) == 0 {
		return nil, httperrors.NewInputParameterError("no available storage type %s to change", input.StorageType)
	}
	if len(storages) > 1 {
		return nil, httperrors.NewInputParameterError("duplicate storage type %s found", input.StorageType)
	}

	self.SetStatus(ctx, userCred, api.DISK_MIGRATING, "")
	params := jsonutils.NewDict()
	params.Set("storage_id", jsonutils.NewString(storages[0].Id))
	task, err := taskman.TaskManager.NewTask(ctx, "DiskChangeStorageTypeTask", self, userCred, params, "", "", nil)
	if err != nil {
		return nil, errors.Wrapf(err, "NewTask")
	}
	return nil, task.ScheduleRun(nil)
}

func (self *SDisk) GetSchedMigrateParams(targetStorageId string) (*schedapi.ScheduleInput, error) {
	diskConfig := self.ToDiskConfig()
	diskConfig.Medium = ""
	diskConfig.Storage = targetStorageId
	input := new(api.DiskCreateInput)
	input.DiskConfig = diskConfig

	srvInput := input.ToServerCreateInput()
	ret := new(schedapi.ScheduleInput)
	err := srvInput.JSON(srvInput).Unmarshal(ret)
	if err != nil {
		return nil, err
	}

	if targetStorageId == "" {
		storage, err := self.GetStorage()
		if err != nil {
			return nil, err
		}
		host, err := storage.GetMasterHost()
		if err != nil {
			return nil, err
		}
		ret.HostId = host.Id
		ret.LiveMigrate = false
	}
	return ret, err
}

func (self *SDisk) StartResetDisk(
	ctx context.Context, userCred mcclient.TokenCredential,
	snapshotId string, autoStart bool, guest *SGuest, parentTaskId string,
) error {
	self.SetStatus(ctx, userCred, api.DISK_RESET, "")
	if guest != nil {
		guest.SetStatus(ctx, userCred, api.VM_DISK_RESET, "disk reset")
	}
	params := jsonutils.NewDict()
	params.Set("snapshot_id", jsonutils.NewString(snapshotId))
	params.Set("auto_start", jsonutils.NewBool(autoStart))
	task, err := taskman.TaskManager.NewTask(ctx, "DiskResetTask", self, userCred, params, parentTaskId, "", nil)
	if err != nil {
		return errors.Wrapf(err, "NewTask")
	}
	return task.ScheduleRun(nil)
}

func (disk *SDisk) PerformResize(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject, input api.DiskResizeInput) (jsonutils.JSONObject, error) {
	guest := disk.GetGuest()
	sizeMb, err := input.SizeMb()
	if err != nil {
		return nil, err
	}
	err = disk.doResize(ctx, userCred, sizeMb, guest)
	if err != nil {
		return nil, err
	}
	return nil, nil
}

func (disk *SDisk) getHypervisor() string {
	storage, _ := disk.GetStorage()
	if storage != nil {
		host, _ := storage.GetMasterHost()
		if host != nil {
			driver, _ := host.GetHostDriver()
			if driver != nil {
				return driver.GetHypervisor()
			}
		}
	}
	hypervisor := disk.GetMetadata(context.Background(), "hypervisor", nil)
	return hypervisor
}

func (disk *SDisk) GetQuotaKeys() (quotas.IQuotaKeys, error) {
	storage, _ := disk.GetStorage()
	if storage == nil {
		return nil, errors.Wrap(httperrors.ErrInvalidStatus, "no valid storage")
	}
	provider := storage.GetCloudprovider()
	if provider == nil && len(storage.ManagerId) > 0 {
		return nil, errors.Wrap(httperrors.ErrInvalidStatus, "no valid manager")
	}
	zone, _ := storage.getZone()
	if zone == nil {
		return nil, errors.Wrap(httperrors.ErrInvalidStatus, "no valid zone")
	}
	return fetchComputeQuotaKeys(
		rbacscope.ScopeProject,
		disk.GetOwnerId(),
		zone,
		provider,
		disk.getHypervisor(),
	), nil
}

func (disk *SDisk) doResize(ctx context.Context, userCred mcclient.TokenCredential, sizeMb int, guest *SGuest) error {
	if disk.Status != api.DISK_READY {
		return httperrors.NewResourceNotReadyError("Resize disk when disk is READY")
	}
	if sizeMb < disk.DiskSize {
		return httperrors.NewUnsupportOperationError("Disk cannot be thrink")
	}
	if sizeMb == disk.DiskSize {
		return nil
	}
	addDisk := sizeMb - disk.DiskSize
	storage, _ := disk.GetStorage()
	if storage == nil {
		return httperrors.NewInternalServerError("disk has no valid storage")
	}

	var guestdriver IGuestDriver
	if host, _ := storage.GetMasterHost(); host != nil {
		hostDriver, err := host.GetHostDriver()
		if err != nil {
			return errors.Wrapf(err, "GetHostDriver")
		}
		if err := hostDriver.ValidateDiskSize(storage, sizeMb>>10); err != nil {
			return httperrors.NewInputParameterError("%v", err)
		}
		guestdriver, _ = GetDriver(hostDriver.GetHypervisor(), hostDriver.GetProvider())
	}
	if guestdriver == nil || guestdriver.DoScheduleStorageFilter() {
		if int64(addDisk) > storage.GetFreeCapacity() && !storage.IsEmulated {
			return httperrors.NewOutOfResourceError("Not enough free space")
		}
	}
	if guest != nil {
		if err := guest.ValidateResizeDisk(disk, storage); err != nil {
			return httperrors.NewInputParameterError("%v", err)
		}
	}
	pendingUsage := SQuota{Storage: int(addDisk)}
	keys, err := disk.GetQuotaKeys()
	if err != nil {
		return httperrors.NewInternalServerError("disk.GetQuotaKeys fail %s", err)
	}
	pendingUsage.SetKeys(keys)
	err = quotas.CheckSetPendingQuota(ctx, userCred, &pendingUsage)
	if err != nil {
		return httperrors.NewGeneralError(err)
	}

	if guest != nil {
		return guest.StartGuestDiskResizeTask(ctx, userCred, disk.Id, int64(sizeMb), "", &pendingUsage)
	} else {
		return disk.StartDiskResizeTask(ctx, userCred, int64(sizeMb), "", &pendingUsage)
	}
}

func (self *SDisk) GetIStorage(ctx context.Context) (cloudprovider.ICloudStorage, error) {
	storage, err := self.GetStorage()
	if err != nil {
		return nil, errors.Wrapf(err, "GetStorage")
	}
	istorage, err := storage.GetIStorage(ctx)
	if err != nil {
		return nil, err
	}
	return istorage, nil
}

func (self *SDisk) GetIDisk(ctx context.Context) (cloudprovider.ICloudDisk, error) {
	if len(self.ExternalId) == 0 {
		return nil, errors.Wrapf(cloudprovider.ErrNotFound, "empty external id")
	}
	iStorage, err := self.GetIStorage(ctx)
	if err != nil {
		return nil, errors.Wrapf(err, "GetIStorage")
	}
	return iStorage.GetIDiskById(self.GetExternalId())
}

func (self *SDisk) GetZone() (*SZone, error) {
	storage, err := self.GetStorage()
	if err != nil {
		return nil, err
	}
	return storage.getZone()
}

func (m *SDiskManager) CheckGlanceImage(ctx context.Context, userCred mcclient.TokenCredential, name string, generateName string) error {
	if len(generateName) == 0 {
		s := auth.GetAdminSession(ctx, options.Options.Region)
		imageList, err := image.Images.List(s, jsonutils.Marshal(map[string]string{"name": name, "admin": "true"}))
		if err != nil {
			return err
		}
		if imageList.Total > 0 {
			return httperrors.NewConflictError("Duplicate image name %s", name)
		}
	}
	return nil
}

type CreateGlanceImageInput struct {
	Name          string
	GenerateName  string
	VirtualSize   int
	DiskFormat    string
	OsArch        string
	Properties    map[string]string
	ProjectId     string
	EncryptKeyId  string
	ClassMetadata map[string]string
}

func (m *SDiskManager) CreateGlanceImage(ctx context.Context, userCred mcclient.TokenCredential, input *CreateGlanceImageInput) (string, error) {
	if err := DiskManager.CheckGlanceImage(ctx, userCred, input.Name, input.GenerateName); err != nil {
		return "", err
	}
	/*
		no need to check quota anymore
		session := auth.GetSession(userCred, options.Options.Region, "v2")
		quota := image_models.SQuota{Image: 1}
		if _, err := image.ImageQuotas.DoQuotaCheck(session, jsonutils.Marshal(&quota)); err != nil {
			return "", err
		}*/
	us := auth.GetSession(ctx, userCred, options.Options.Region)
	result, err := image.Images.Create(us, jsonutils.Marshal(input))
	if err != nil {
		return "", err
	}
	imageId, err := result.GetString("id")
	if err != nil {
		return "", err
	}
	if len(input.ClassMetadata) > 0 {
		_, err = image.Images.PerformAction(us, imageId, "set-class-metadata", jsonutils.Marshal(input.ClassMetadata))
		if err != nil {
			return "", errors.Wrapf(err, "unable to SetClassMetadata for image %s", imageId)
		}
	}
	return imageId, nil
}

func (self *SDisk) PrepareSaveImage(ctx context.Context, userCred mcclient.TokenCredential, input api.ServerSaveImageInput) (string, error) {
	zone, _ := self.GetZone()
	if zone == nil {
		return "", httperrors.NewResourceNotFoundError("No zone for this disk")
	}
	imageInput := &CreateGlanceImageInput{
		Name:         input.Name,
		GenerateName: input.GenerateName,
		VirtualSize:  self.DiskSize,
		DiskFormat:   self.DiskFormat,
		OsArch:       input.OsArch,
		Properties: map[string]string{
			"notes":   input.Notes,
			"os_type": input.OsType,
			"os_arch": input.OsArch,
		},

		// inherit the ownership of disk
		ProjectId: self.ProjectId,
	}
	if self.IsEncrypted() {
		encKey, err := self.GetEncryptInfo(ctx, userCred)
		if err != nil {
			return "", errors.Wrap(err, "GetEncryptInfo")
		}
		imageInput.EncryptKeyId = encKey.Id
	}
	// check class metadata
	cm, err := self.GetAllClassMetadata()
	if err != nil {
		return "", errors.Wrap(err, "unable to GetAllClassMetadata")
	}
	imageInput.ClassMetadata = cm

	return DiskManager.CreateGlanceImage(ctx, userCred, imageInput)
}

func (self *SDisk) PerformSave(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject, input api.DiskSaveInput) (jsonutils.JSONObject, error) {
	if self.Status != api.DISK_READY {
		return nil, httperrors.NewResourceNotReadyError("Save disk when disk is READY")

	}
	cnt, err := self.GetRuningGuestCount()
	if err != nil {
		return nil, httperrors.NewInternalServerError("GetRuningGuestCount fail %s", err)
	}
	if cnt > 0 {
		return nil, httperrors.NewResourceNotReadyError("Save disk when not being USED")
	}

	if len(input.Name) == 0 {
		return nil, httperrors.NewInputParameterError("Image name is required")
	}
	opts := api.ServerSaveImageInput{
		Name: input.Name,
	}

	input.ImageId, err = self.PrepareSaveImage(ctx, userCred, opts)
	if err != nil {
		return nil, errors.Wrapf(err, "PrepareSaveImage")
	}
	return nil, self.StartDiskSaveTask(ctx, userCred, input, "")
}

func (self *SDisk) StartDiskSaveTask(ctx context.Context, userCred mcclient.TokenCredential, input api.DiskSaveInput, parentTaskId string) error {
	data := jsonutils.Marshal(input).(*jsonutils.JSONDict)
	task, err := taskman.TaskManager.NewTask(ctx, "DiskSaveTask", self, userCred, data, parentTaskId, "", nil)
	if err != nil {
		return errors.Wrapf(err, "NewTask")
	}
	self.SetStatus(ctx, userCred, api.DISK_START_SAVE, "")
	task.ScheduleRun(nil)
	return nil
}

func (self *SDisk) ValidateDeleteCondition(ctx context.Context, info api.DiskDetails) error {
	if len(info.Guests) > 0 {
		return httperrors.NewNotEmptyError("Virtual disk %s(%s) used by virtual servers", self.Name, self.Id)
	}
	if self.IsNotDeletablePrePaid() {
		return httperrors.NewForbiddenError("not allow to delete prepaid disk in valid status")
	}
	return self.SVirtualResourceBase.ValidateDeleteCondition(ctx, nil)
}

func (self *SDisk) validateDeleteCondition(ctx context.Context, isPurge bool) error {
	if !isPurge {
		storage, _ := self.GetStorage()
		if storage == nil {
			// storage is empty, a dirty data, allow to delete
			return nil
		}
		host, _ := storage.GetMasterHost()
		if host == nil {
			return httperrors.NewBadRequestError("storage of disk %s no valid host", self.Id)
		}
	}
	cnt, err := self.GetGuestDiskCount()
	if err != nil {
		return httperrors.NewInternalServerError("GetGuestDiskCount for disk %s fail %s", self.Id, err)
	}
	if cnt > 0 {
		return httperrors.NewNotEmptyError("Virtual disk %s(%s) used by virtual servers", self.Name, self.Id)
	}
	if !isPurge && self.IsNotDeletablePrePaid() {
		return httperrors.NewForbiddenError("not allow to delete prepaid disk in valid status")
	}
	return self.SVirtualResourceBase.ValidateDeleteCondition(ctx, nil)
}

func (self *SDisk) AllowDeleteItem(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject, data jsonutils.JSONObject) bool {
	provider := self.GetCloudprovider()
	if provider != nil {
		if !provider.IsAvailable() {
			return false
		}

		account, _ := provider.GetCloudaccount()
		if account != nil && !account.IsAvailable() {
			return false
		}
	}

	overridePendingDelete := false
	purge := false
	if query != nil {
		overridePendingDelete = jsonutils.QueryBoolean(query, "override_pending_delete", false)
		purge = jsonutils.QueryBoolean(query, "purge", false)
	}
	if (overridePendingDelete || purge) && !db.IsAdminAllowDelete(ctx, userCred, self) {
		return false
	}
	return self.IsOwner(userCred) || db.IsAdminAllowDelete(ctx, userCred, self)
}

func (self *SDisk) GetTemplateId() string {
	if len(self.TemplateId) == 0 {
		return ""
	}
	imageObj, err := CachedimageManager.FetchById(self.TemplateId)
	if err != nil || imageObj == nil {
		log.Errorf("failed to found disk %s(%s) templateId %s: %s", self.Name, self.Id, self.TemplateId, err)
		return ""
	}
	return self.TemplateId
}

func (self *SDisk) IsLocal() bool {
	storage, _ := self.GetStorage()
	if storage != nil {
		return storage.IsLocal()
	}
	return false
}

func (self *SDisk) GetCloudproviderId() string {
	storage, _ := self.GetStorage()
	if storage != nil {
		return storage.GetCloudproviderId()
	}
	return ""
}

func (self *SDisk) GetStorage() (*SStorage, error) {
	store, err := StorageManager.FetchById(self.StorageId)
	if err != nil {
		return nil, errors.Wrapf(err, "GetStorage(%s)", self.StorageId)
	}
	return store.(*SStorage), nil
}

func (self *SDisk) GetRegionDriver() (IRegionDriver, error) {
	storage, _ := self.GetStorage()
	if storage == nil {
		return nil, fmt.Errorf("failed to found storage for disk %s(%s)", self.Name, self.Id)
	}
	return storage.GetRegionDriver()
}

func (self *SDisk) GetBackupStorage() *SStorage {
	if len(self.BackupStorageId) == 0 {
		return nil
	}
	store, _ := StorageManager.FetchById(self.BackupStorageId)
	if store != nil {
		return store.(*SStorage)
	}
	return nil
}

func (self *SDisk) GetCloudprovider() *SCloudprovider {
	if storage, _ := self.GetStorage(); storage != nil {
		return storage.GetCloudprovider()
	}
	return nil
}

func (self *SDisk) GetPathAtHost(host *SHost) string {
	hostStorage := host.GetHoststorageOfId(self.StorageId)
	if hostStorage != nil {
		return path.Join(hostStorage.MountPoint, self.Id)
	} else if len(self.BackupStorageId) > 0 {
		hostStorage = host.GetHoststorageOfId(self.BackupStorageId)
		if hostStorage != nil {
			return path.Join(hostStorage.MountPoint, self.Id)
		}
	}
	return ""
}

func (disk *SDisk) getCandidateHostIds() ([]string, error) {
	hss, err := HoststorageManager.GetHostStoragesByStorageId(disk.StorageId)
	if err != nil {
		return nil, errors.Wrap(err, "GetHostStoragesByStorageId")
	}
	candidates := make([]string, 0)
	for i := range hss {
		candidates = append(candidates, hss[i].HostId)
	}
	return candidates, nil
}

func (self *SDisk) GetMasterHost(storage *SStorage) (*SHost, error) {
	if storage.StorageType == api.STORAGE_SLVM {
		if guest := self.GetGuest(); guest != nil {
			return guest.GetHost()
		}
	}

	if storage.MasterHost != "" {
		return storage.GetMasterHost()
	}

	hosts := HostManager.Query().SubQuery()
	hoststorages := HoststorageManager.Query().SubQuery()

	q := hosts.Query().Join(hoststorages, sqlchemy.Equals(hoststorages.Field("host_id"), hosts.Field("id")))
	q = q.Filter(sqlchemy.Equals(hoststorages.Field("storage_id"), self.StorageId))
	q = q.IsTrue("enabled")
	q = q.Equals("host_status", api.HOST_ONLINE).Asc("id")
	guest := self.GetGuest()
	if guest != nil && len(guest.OsArch) > 0 {
		switch guest.OsArch {
		case apis.OS_ARCH_X86:
			q = q.In("cpu_architecture", apis.ARCH_X86)
		case apis.OS_ARCH_ARM:
			q = q.In("cpu_architecture", apis.ARCH_ARM)
		case apis.OS_ARCH_RISCV:
			q = q.In("cpu_architecture", apis.ARCH_RISCV)
		}
	}
	host := SHost{}
	host.SetModelManager(HostManager, &host)
	err := q.First(&host)
	if err != nil {
		return nil, errors.Wrapf(err, "q.First")
	}
	return &host, nil
}

func (self *SDisk) GetFetchUrl() (string, error) {
	storage, err := self.GetStorage()
	if err != nil {
		return "", errors.Wrapf(err, "self.GetStorage")
	}
	host, err := storage.GetMasterHost()
	if err != nil {
		return "", errors.Wrapf(err, "storage.GetMasterHost")
	}
	return fmt.Sprintf("%s/disks/%s", host.GetFetchUrl(true), self.Id), nil
}

func (self *SDisk) GetFsFormat() string {
	return self.FsFormat
}

func (self *SDisk) GetCacheImageFormat(ctx context.Context) (string, error) {
	if self.TemplateId != "" {
		s := auth.GetAdminSession(ctx, options.Options.Region)
		img, err := image.Images.Get(s, self.TemplateId, nil)
		if err == nil {
			diskFmt, err := img.GetString("disk_format")
			if err != nil {
				return "", errors.Wrapf(err, "not found disk_format of image %s", self.TemplateId)
			}
			if diskFmt == imageapi.IMAGE_DISK_FORMAT_TGZ {
				return imageapi.IMAGE_DISK_FORMAT_TGZ, nil
			}
		}
	}
	if self.DiskFormat == "raw" {
		return "qcow2", nil
	}
	return self.DiskFormat, nil
}

func (manager *SDiskManager) getDisksByStorage(storage *SStorage) ([]SDisk, error) {
	disks := make([]SDisk, 0)
	q := manager.Query().Equals("storage_id", storage.Id)
	err := db.FetchModelObjects(manager, q, &disks)
	if err != nil {
		log.Errorf("%s", err)
		return nil, err
	}
	return disks, nil
}

func (manager *SDiskManager) findOrCreateDisk(ctx context.Context, userCred mcclient.TokenCredential, provider cloudprovider.ICloudProvider, vdisk cloudprovider.ICloudDisk, index int, syncOwnerId mcclient.IIdentityProvider, managerId string) (*SDisk, error) {
	diskId := vdisk.GetGlobalId()
	diskObj, err := db.FetchByExternalIdAndManagerId(manager, diskId, func(q *sqlchemy.SQuery) *sqlchemy.SQuery {
		sq := StorageManager.Query().SubQuery()
		return q.Join(sq, sqlchemy.Equals(sq.Field("id"), q.Field("storage_id"))).Filter(sqlchemy.Equals(sq.Field("manager_id"), managerId))
	})
	if err != nil {
		if errors.Cause(err) != sql.ErrNoRows {
			return nil, errors.Wrapf(err, "db.FetchByExternalIdAndManagerId %s", diskId)
		}
		vstorage, err := vdisk.GetIStorage()
		if err != nil {
			return nil, errors.Wrapf(err, "unable to GetIStorage of vdisk %q", vdisk.GetName())
		}

		storageObj, err := db.FetchByExternalIdAndManagerId(StorageManager, vstorage.GetGlobalId(), func(q *sqlchemy.SQuery) *sqlchemy.SQuery {
			return q.Equals("manager_id", managerId)
		})
		if err != nil {
			return nil, errors.Wrapf(err, "cannot find storage of vdisk %s", vdisk.GetName())
		}
		storage := storageObj.(*SStorage)
		return manager.newFromCloudDisk(ctx, userCred, provider, vdisk, storage, -1, syncOwnerId)
	}
	return diskObj.(*SDisk), nil
}

func (manager *SDiskManager) SyncDisks(ctx context.Context, userCred mcclient.TokenCredential, provider cloudprovider.ICloudProvider, storage *SStorage, disks []cloudprovider.ICloudDisk, syncOwnerId mcclient.IIdentityProvider, xor bool) ([]SDisk, []cloudprovider.ICloudDisk, compare.SyncResult) {
	lockman.LockRawObject(ctx, manager.Keyword(), storage.Id)
	defer lockman.ReleaseRawObject(ctx, manager.Keyword(), storage.Id)

	localDisks := make([]SDisk, 0)
	remoteDisks := make([]cloudprovider.ICloudDisk, 0)
	syncResult := compare.SyncResult{}

	dbDisks, err := manager.getDisksByStorage(storage)
	if err != nil {
		syncResult.Error(err)
		return nil, nil, syncResult
	}

	removed := make([]SDisk, 0)
	commondb := make([]SDisk, 0)
	commonext := make([]cloudprovider.ICloudDisk, 0)
	added := make([]cloudprovider.ICloudDisk, 0)

	err = compare.CompareSets(dbDisks, disks, &removed, &commondb, &commonext, &added)
	if err != nil {
		syncResult.Error(err)
		return nil, nil, syncResult
	}

	for i := 0; i < len(removed); i += 1 {
		err = removed[i].syncRemoveCloudDisk(ctx, userCred)
		if err != nil {
			// vm not sync, so skip disk used by vm error
			if errors.Cause(err) == httperrors.ErrNotEmpty {
				continue
			}
			syncResult.DeleteError(err)
		} else {
			syncResult.Delete()
		}
	}

	for i := 0; i < len(commondb); i += 1 {
		guest := commondb[i].GetGuest()
		// 仅独立磁盘判断是否需要通过标签跳过同步，避免虚拟机有标签，磁盘没标签导致磁盘不断同步删除后再同步
		if gotypes.IsNil(guest) {
			skip, key := IsNeedSkipSync(commonext[i])
			if skip {
				log.Infof("delete disk %s(%s) with tag key or value: %s", commonext[i].GetName(), commonext[i].GetGlobalId(), key)
				err := commondb[i].RealDelete(ctx, userCred)
				if err != nil {
					syncResult.DeleteError(err)
					continue
				}
				syncResult.Delete()
				continue
			}
		}
		if !xor {
			err = commondb[i].syncWithCloudDisk(ctx, userCred, provider, commonext[i], -1, syncOwnerId, storage.ManagerId)
			if err != nil {
				syncResult.UpdateError(err)
				continue
			}
		}
		localDisks = append(localDisks, commondb[i])
		remoteDisks = append(remoteDisks, commonext[i])
		syncResult.Update()
	}

	for i := 0; i < len(added); i += 1 {
		skip, key := IsNeedSkipSync(added[i])
		if skip {
			log.Infof("skip disk %s(%s) sync with tag key or value: %s", added[i].GetName(), added[i].GetGlobalId(), key)
			continue
		}
		extId := added[i].GetGlobalId()
		_disk, err := db.FetchByExternalIdAndManagerId(manager, extId, func(q *sqlchemy.SQuery) *sqlchemy.SQuery {
			sq := StorageManager.Query().SubQuery()
			return q.Join(sq, sqlchemy.Equals(sq.Field("id"), q.Field("storage_id"))).Filter(sqlchemy.Equals(sq.Field("manager_id"), storage.ManagerId))
		})
		if err != nil && err != sql.ErrNoRows {
			//主要是显示duplicate err及 general err,方便排错
			msg := fmt.Errorf("failed to found disk by external Id %s error: %v", extId, err)
			syncResult.Error(msg)
			continue
		}
		if _disk != nil {
			disk := _disk.(*SDisk)
			err = disk.syncDiskStorage(ctx, userCred, added[i], storage.ManagerId)
			if err != nil {
				syncResult.UpdateError(err)
			} else {
				syncResult.Update()
			}
			continue
		}
		new, err := manager.newFromCloudDisk(ctx, userCred, provider, added[i], storage, -1, syncOwnerId)
		if err != nil {
			syncResult.AddError(err)
		} else {
			localDisks = append(localDisks, *new)
			remoteDisks = append(remoteDisks, added[i])
			syncResult.Add()
		}
	}

	return localDisks, remoteDisks, syncResult
}

func (self *SDisk) syncDiskStorage(ctx context.Context, userCred mcclient.TokenCredential, idisk cloudprovider.ICloudDisk, managerId string) error {
	istorage, err := idisk.GetIStorage()
	if err != nil {
		return errors.Wrapf(err, "idisk.GetIStorage %s", idisk.GetGlobalId())
	}
	storageExtId := istorage.GetGlobalId()
	storage, err := db.FetchByExternalIdAndManagerId(StorageManager, storageExtId, func(q *sqlchemy.SQuery) *sqlchemy.SQuery {
		return q.Equals("manager_id", managerId)
	})
	if err != nil {
		return errors.Wrapf(err, "storage db.FetchByExternalIdAndManagerId(%s)", storageExtId)
	}
	diff, err := db.UpdateWithLock(ctx, self, func() error {
		self.StorageId = storage.GetId()
		self.Status = idisk.GetStatus()
		return nil
	})
	if err != nil {
		return errors.Wrapf(err, "db.UpdateWithLock")
	}
	db.OpsLog.LogSyncUpdate(self, diff, userCred)
	return nil
}

func (self *SDisk) GetIRegion(ctx context.Context) (cloudprovider.ICloudRegion, error) {
	storage, _ := self.GetStorage()
	if storage == nil {
		return nil, fmt.Errorf("failed to get storage for disk %s(%s)", self.Name, self.Id)
	}

	provider, err := storage.GetDriver(ctx)
	if err != nil {
		return nil, fmt.Errorf("No cloudprovider for storage %s(%s) error: %v", storage.Name, storage.Id, err)
	}

	if provider.GetFactory().IsOnPremise() {
		return provider.GetOnPremiseIRegion()
	}
	region, err := storage.GetRegion()
	if err != nil {
		return nil, err
	}
	return provider.GetIRegionById(region.ExternalId)
}

func (self *SDisk) syncRemoveCloudDisk(ctx context.Context, userCred mcclient.TokenCredential) error {
	lockman.LockObject(ctx, self)
	defer lockman.ReleaseObject(ctx, self)

	iregion, err := self.GetIRegion(ctx)
	if err != nil {
		return err
	}
	iDisk, err := iregion.GetIDiskById(self.ExternalId)
	if err == nil {
		if storageId := iDisk.GetIStorageId(); len(storageId) > 0 {
			storage, err := db.FetchByExternalIdAndManagerId(StorageManager, storageId, func(q *sqlchemy.SQuery) *sqlchemy.SQuery {
				if s, _ := self.GetStorage(); s != nil {
					return q.Equals("manager_id", s.ManagerId)
				}
				return q
			})
			if err == nil {
				_, err = db.Update(self, func() error {
					self.StorageId = storage.GetId()
					return nil
				})
				return err
			}
		}
	} else if errors.Cause(err) != cloudprovider.ErrNotFound {
		return err
	}

	err = self.validateDeleteCondition(ctx, true)
	if err != nil {
		self.SetStatus(ctx, userCred, api.DISK_UNKNOWN, "missing original disk after sync")
		return err
	}
	err = self.RealDelete(ctx, userCred)
	if err != nil {
		return err
	}
	notifyclient.EventNotify(ctx, userCred, notifyclient.SEventNotifyParam{
		Obj:    self,
		Action: notifyclient.ActionSyncDelete,
	})
	return nil
}

func (self *SDisk) syncWithCloudDisk(ctx context.Context, userCred mcclient.TokenCredential, provider cloudprovider.ICloudProvider, extDisk cloudprovider.ICloudDisk, index int, syncOwnerId mcclient.IIdentityProvider, managerId string) error {
	recycle := false
	guests := self.GetGuests()
	if provider.GetFactory().IsSupportPrepaidResources() && len(guests) == 1 && guests[0].IsPrepaidRecycle() {
		recycle = true
	}

	diff, err := db.UpdateWithLock(ctx, self, func() error {
		if options.Options.EnableSyncName {
			newName, _ := db.GenerateAlterName(self, extDisk.GetName())
			if len(newName) > 0 {
				self.Name = newName
			}
		}

		self.Status = extDisk.GetStatus()
		self.DiskFormat = extDisk.GetDiskFormat()
		self.DiskSize = extDisk.GetDiskSizeMB()
		self.AccessPath = extDisk.GetAccessPath()
		self.Preallocation = extDisk.GetPreallocation()
		if iops := extDisk.GetIops(); iops > 0 {
			self.Iops = iops
		}
		if extDisk.GetIsAutoDelete() {
			self.AutoDelete = true
		}
		if device := extDisk.GetDeviceName(); len(device) > 0 {
			self.Device = device
		}
		// self.TemplateId = extDisk.GetTemplateId() no sync template ID
		if templateId := extDisk.GetTemplateId(); len(templateId) > 0 {
			cachedImage, err := db.FetchByExternalId(CachedimageManager, templateId)
			if err == nil && cachedImage != nil {
				self.TemplateId = cachedImage.GetId()
			}
		}
		self.DiskType = extDisk.GetDiskType()
		if index == 0 {
			self.DiskType = api.DISK_TYPE_SYS
		}
		// self.FsFormat = extDisk.GetFsFormat()
		self.Nonpersistent = extDisk.GetIsNonPersistent()

		self.IsEmulated = extDisk.IsEmulated()

		if provider.GetFactory().IsSupportPrepaidResources() && !recycle {
			if billintType := extDisk.GetBillingType(); len(billintType) > 0 {
				self.BillingType = billing_api.TBillingType(extDisk.GetBillingType())
				self.ExpiredAt = time.Time{}
				self.AutoRenew = false
				if self.BillingType == billing_api.BILLING_TYPE_PREPAID {
					self.ExpiredAt = extDisk.GetExpiredAt()
					self.AutoRenew = extDisk.IsAutoRenew()
				}
			}
		}

		if createdAt := extDisk.GetCreatedAt(); !createdAt.IsZero() {
			self.CreatedAt = createdAt
		}

		return nil
	})
	if err != nil {
		return errors.Wrapf(err, "db.UpdateWithLock")
	}
	storage, err := self.GetStorage()
	if err != nil {
		return errors.Wrapf(err, "GetStorage")
	}

	db.OpsLog.LogSyncUpdate(self, diff, userCred)

	if len(diff) > 0 {
		notifyclient.EventNotify(ctx, userCred, notifyclient.SEventNotifyParam{
			Obj:    self,
			Action: notifyclient.ActionSyncUpdate,
		})
	}

	if account := storage.GetCloudaccount(); account != nil {
		syncVirtualResourceMetadata(ctx, userCred, self, extDisk, account.ReadOnly)
	}

	if len(guests) == 0 {
		if provider := storage.GetCloudprovider(); provider != nil {
			SyncCloudProject(ctx, userCred, self, syncOwnerId, extDisk, provider)
		}
	} else {
		self.SyncCloudProjectId(userCred, guests[0].GetOwnerId())
	}

	return nil
}

func (manager *SDiskManager) newFromCloudDisk(ctx context.Context, userCred mcclient.TokenCredential, provider cloudprovider.ICloudProvider, extDisk cloudprovider.ICloudDisk, storage *SStorage, index int, syncOwnerId mcclient.IIdentityProvider) (*SDisk, error) {
	disk := SDisk{}
	disk.SetModelManager(manager, &disk)

	disk.Status = extDisk.GetStatus()
	disk.ExternalId = extDisk.GetGlobalId()
	disk.StorageId = storage.Id

	disk.Iops = extDisk.GetIops()
	disk.DiskFormat = extDisk.GetDiskFormat()
	disk.DiskSize = extDisk.GetDiskSizeMB()
	disk.AutoDelete = extDisk.GetIsAutoDelete()
	disk.Preallocation = extDisk.GetPreallocation()
	disk.DiskType = extDisk.GetDiskType()
	if index == 0 {
		disk.DiskType = api.DISK_TYPE_SYS
	}
	disk.Nonpersistent = extDisk.GetIsNonPersistent()
	disk.Device = extDisk.GetDeviceName()

	disk.IsEmulated = extDisk.IsEmulated()

	if templateId := extDisk.GetTemplateId(); len(templateId) > 0 {
		cachedImage, err := db.FetchByExternalId(CachedimageManager, templateId)
		if err == nil && cachedImage != nil {
			disk.TemplateId = cachedImage.GetId()
		}
	}

	if provider.GetFactory().IsSupportPrepaidResources() {
		disk.BillingType = billing_api.TBillingType(extDisk.GetBillingType())
		if expired := extDisk.GetExpiredAt(); !expired.IsZero() {
			disk.ExpiredAt = expired
		}
		disk.AutoRenew = extDisk.IsAutoRenew()
	}

	if createAt := extDisk.GetCreatedAt(); !createAt.IsZero() {
		disk.CreatedAt = createAt
	}

	var err = func() error {
		lockman.LockRawObject(ctx, manager.Keyword(), "name")
		defer lockman.ReleaseRawObject(ctx, manager.Keyword(), "name")

		newName, err := db.GenerateName(ctx, manager, syncOwnerId, extDisk.GetName())
		if err != nil {
			return err
		}
		disk.Name = newName

		return manager.TableSpec().Insert(ctx, &disk)
	}()
	if err != nil {
		return nil, errors.Wrapf(err, "newFromCloudDisk")
	}

	syncVirtualResourceMetadata(ctx, userCred, &disk, extDisk, false)

	if provider := storage.GetCloudprovider(); provider != nil {
		SyncCloudProject(ctx, userCred, &disk, syncOwnerId, extDisk, provider)
	}

	db.OpsLog.LogEvent(&disk, db.ACT_CREATE, disk.GetShortDesc(ctx), userCred)

	notifyclient.EventNotify(ctx, userCred, notifyclient.SEventNotifyParam{
		Obj:    &disk,
		Action: notifyclient.ActionSyncCreate,
	})

	return &disk, nil
}

func totalDiskSize(
	scope rbacscope.TRbacScope,
	ownerId mcclient.IIdentityProvider,
	active tristate.TriState,
	ready tristate.TriState,
	includeSystem bool,
	pendingDelete bool,
	rangeObjs []db.IStandaloneModel,
	providers []string,
	brands []string,
	cloudEnv string,
	hypervisors []string,
) int {
	disks := DiskManager.Query().SubQuery()
	q := disks.Query(sqlchemy.SUM("total", disks.Field("disk_size")))
	storages := StorageManager.Query().SubQuery()
	q = q.Join(storages, sqlchemy.Equals(storages.Field("id"), disks.Field("storage_id")))
	q = CloudProviderFilter(q, storages.Field("manager_id"), providers, brands, cloudEnv)
	q = RangeObjectsFilter(q, rangeObjs, nil, storages.Field("zone_id"), storages.Field("manager_id"), nil, storages.Field("id"))
	if len(hypervisors) > 0 {
		hoststorages := HoststorageManager.Query().SubQuery()
		hosts := HostManager.Query().SubQuery()
		q = q.Join(hoststorages, sqlchemy.Equals(storages.Field("id"), hoststorages.Field("storage_id")))
		q = q.Join(hosts, sqlchemy.Equals(hoststorages.Field("host_id"), hosts.Field("id")))
		q = q.Filter(sqlchemy.In(hosts.Field("host_type"), Hypervisors2HostTypes(hypervisors)))
	}
	if !active.IsNone() {
		if active.IsTrue() {
			q = q.Filter(sqlchemy.In(storages.Field("status"), []string{api.STORAGE_ENABLED, api.STORAGE_ONLINE}))
		} else {
			q = q.Filter(sqlchemy.NotIn(storages.Field("status"), []string{api.STORAGE_ENABLED, api.STORAGE_ONLINE}))
		}
	}

	switch scope {
	case rbacscope.ScopeSystem:
		// do nothing
	case rbacscope.ScopeDomain:
		q = q.Filter(sqlchemy.Equals(disks.Field("domain_id"), ownerId.GetProjectDomainId()))
	case rbacscope.ScopeProject:
		q = q.Filter(sqlchemy.Equals(disks.Field("tenant_id"), ownerId.GetProjectId()))
	}

	if !ready.IsNone() {
		if ready.IsTrue() {
			q = q.Filter(sqlchemy.Equals(disks.Field("status"), api.DISK_READY))
		} else {
			q = q.Filter(sqlchemy.NotEquals(disks.Field("status"), api.DISK_READY))
		}
	}
	if !includeSystem {
		q = q.Filter(sqlchemy.OR(sqlchemy.IsNull(disks.Field("is_system")),
			sqlchemy.IsFalse(disks.Field("is_system"))))
	}
	if pendingDelete {
		q = q.Filter(sqlchemy.IsTrue(disks.Field("pending_deleted")))
	} else {
		q = q.Filter(sqlchemy.OR(sqlchemy.IsNull(disks.Field("pending_deleted")), sqlchemy.IsFalse(disks.Field("pending_deleted"))))
	}

	row := q.Row()
	size := sql.NullInt64{}
	err := row.Scan(&size)
	if err != nil {
		log.Errorf("totalDiskSize error %s: %s", err, q.String())
		return 0
	}
	if size.Valid {
		return int(size.Int64)
	} else {
		return 0
	}
}

func parseDiskInfo(ctx context.Context, userCred mcclient.TokenCredential, info *api.DiskConfig) (*api.DiskConfig, error) {
	if info.Storage != "" {
		if err := fillDiskConfigByStorage(ctx, userCred, info, info.Storage); err != nil {
			return nil, errors.Wrap(err, "fillDiskConfigByStorage")
		}
	}
	if info.DiskId != "" {
		if err := fillDiskConfigByDisk(ctx, userCred, info, info.DiskId); err != nil {
			return nil, errors.Wrap(err, "fillDiskConfigByDisk")
		}
	}
	if info.SnapshotId != "" {
		if err := fillDiskConfigBySnapshot(ctx, userCred, info, info.SnapshotId); err != nil {
			return nil, errors.Wrap(err, "fillDiskConfigBySnapshot")
		}
	}
	if info.BackupId != "" {
		if err := fillDiskConfigByBackup(ctx, userCred, info, info.BackupId); err != nil {
			return nil, errors.Wrap(err, "fillDiskConfigByBackup")
		}
	}
	if info.ImageId != "" {
		if err := fillDiskConfigByImage(ctx, userCred, info, info.ImageId); err != nil {
			if len(info.SnapshotId) == 0 && len(info.BackupId) == 0 {
				// return error only if no valid snapshotId and backId
				// otherwise, the disk was crated by snapshot or backup, not depends on vald image info
				return nil, errors.Wrap(err, "fillDiskConfigByImage")
			}
		}
	}
	if info.ExistingPath != "" {
		info.ExistingPath = strings.TrimSpace(info.ExistingPath)
		_, err := filepath.Rel("/", info.ExistingPath)
		if err != nil {
			return nil, errors.Wrap(err, "invaild existing path")
		}
	}

	// XXX: do not set default disk size here, set it by each hypervisor driver
	// if len(diskConfig.ImageId) > 0 && diskConfig.SizeMb == 0 {
	// 	diskConfig.SizeMb = options.Options.DefaultDiskSize // MB
	// else
	if len(info.ImageId) == 0 && info.SizeMb == 0 && info.ExistingPath == "" && info.NVMEDevice == nil {
		return nil, httperrors.NewInputParameterError("Diskinfo index %d: both imageID and size are absent", info.Index)
	}
	return info, nil
}

func fillDiskConfigBySnapshot(ctx context.Context, userCred mcclient.TokenCredential, diskConfig *api.DiskConfig, snapshotId string) error {
	iSnapshot, err := SnapshotManager.FetchByIdOrName(ctx, userCred, snapshotId)
	if err != nil {
		if err == sql.ErrNoRows {
			return httperrors.NewNotFoundError("Snapshot %s not found", snapshotId)
		}
		return err
	}
	var snapshot = iSnapshot.(*SSnapshot)
	if storage := StorageManager.FetchStorageById(snapshot.StorageId); storage == nil {
		return httperrors.NewBadRequestError("Snapshot %s storage %s not found, is public cloud?",
			snapshotId, snapshot.StorageId)
	} else {
		if disk := DiskManager.FetchDiskById(snapshot.DiskId); disk != nil {
			diskConfig.Fs = disk.FsFormat
			if len(diskConfig.Format) == 0 {
				diskConfig.Format = disk.DiskFormat
			}
		}
		diskConfig.SnapshotId = snapshot.Id
		diskConfig.DiskType = snapshot.DiskType
		diskConfig.SizeMb = snapshot.VirtualSize
		diskConfig.Backend = storage.StorageType
		diskConfig.Fs = ""
		diskConfig.Mountpoint = ""
		diskConfig.OsArch = snapshot.OsArch
	}
	return nil
}

func fillDiskConfigByBackup(ctx context.Context, userCred mcclient.TokenCredential, diskConfig *api.DiskConfig, backupId string) error {
	iBakcup, err := DiskBackupManager.FetchByIdOrName(ctx, userCred, backupId)
	if err != nil {
		if err == sql.ErrNoRows {
			return httperrors.NewNotFoundError("Backup %s not found", backupId)
		}
		return err
	}
	backup := iBakcup.(*SDiskBackup)
	if diskConfig.DiskType == "" {
		diskConfig.DiskType = backup.DiskType
	}
	diskConfig.BackupId = backup.GetId()
	return nil
}

func fillDiskConfigByImage(ctx context.Context, userCred mcclient.TokenCredential,
	diskConfig *api.DiskConfig, imageId string) error {
	if userCred == nil {
		diskConfig.ImageId = imageId
	} else {
		image, err := CachedimageManager.getImageInfo(ctx, userCred, imageId, false)
		if err != nil {
			log.Errorf("getImageInfo %s fail %s", imageId, err)
			return err
		}
		if image.Status != cloudprovider.IMAGE_STATUS_ACTIVE {
			return httperrors.NewInvalidStatusError("Image status is not active")
		}
		diskConfig.ImageId = image.Id
		diskConfig.ImageEncryptKeyId = image.EncryptKeyId
		diskConfig.ImageProperties = image.Properties
		diskConfig.ImageProperties[imageapi.IMAGE_DISK_FORMAT] = image.DiskFormat
		// if len(diskConfig.Format) == 0 {
		// 	diskConfig.Format = image.DiskFormat
		// }
		// diskConfig.ImageDiskFormat = image.DiskFormat
		CachedimageManager.ImageAddRefCount(image.Id)
		if diskConfig.SizeMb != api.DISK_SIZE_AUTOEXTEND && diskConfig.SizeMb < image.MinDiskMB {
			diskConfig.SizeMb = image.MinDiskMB // MB
		}
		if strings.Contains(image.Properties["os_arch"], "aarch") {
			diskConfig.OsArch = apis.OS_ARCH_AARCH64
		} else {
			diskConfig.OsArch = image.Properties["os_arch"]
		}
	}
	return nil
}

func fillDiskConfigByDisk(ctx context.Context, userCred mcclient.TokenCredential,
	diskConfig *api.DiskConfig, diskId string) error {
	diskObj, err := DiskManager.FetchByIdOrName(ctx, userCred, diskId)
	if err != nil {
		if errors.Cause(err) == sql.ErrNoRows {
			return httperrors.NewResourceNotFoundError2("disk", diskId)
		} else {
			return errors.Wrapf(err, "DiskManager.FetchByIdOrName %s", diskId)
		}
	}
	disk := diskObj.(*SDisk)
	if disk.Status != api.DISK_READY {
		return errors.Wrapf(httperrors.ErrInvalidStatus, "disk status %s not ready", disk.Status)
	}
	guests := disk.GetGuests()
	if len(guests) > 0 {
		return errors.Wrapf(httperrors.ErrInvalidStatus, "disk %s has been used", diskId)
	}

	diskConfig.DiskId = disk.Id
	diskConfig.SizeMb = disk.DiskSize
	if disk.SnapshotId != "" {
		diskConfig.SnapshotId = disk.SnapshotId
	}
	if disk.TemplateId != "" {
		diskConfig.ImageId = disk.TemplateId
	}
	if disk.OsArch != "" {
		diskConfig.OsArch = disk.OsArch
	}
	storage, err := disk.GetStorage()
	if err != nil {
		return errors.Wrap(err, "disk.GetStorage")
	}
	if !storage.Enabled.IsTrue() {
		return errors.Wrap(httperrors.ErrInvalidStatus, "storage not enabled")
	}
	if storage.Status != api.STORAGE_ONLINE {
		return errors.Wrap(httperrors.ErrInvalidStatus, "storage not online")
	}
	diskConfig.Storage = disk.StorageId
	return nil
}

func fillDiskConfigByStorage(ctx context.Context, userCred mcclient.TokenCredential,
	diskConfig *api.DiskConfig, storageId string) error {
	storageObj, err := StorageManager.FetchByIdOrName(ctx, userCred, storageId)
	if err != nil {
		if errors.Cause(err) == sql.ErrNoRows {
			return httperrors.NewResourceNotFoundError2("storage", storageId)
		} else {
			return errors.Wrapf(err, "StorageManager.FetchByIdOrName %s", storageId)
		}
	}
	storage := storageObj.(*SStorage)
	if !storage.Enabled.IsTrue() {
		return errors.Wrap(httperrors.ErrInvalidStatus, "storage not enabled")
	}
	if storage.Status != api.STORAGE_ONLINE {
		return errors.Wrap(httperrors.ErrInvalidStatus, "storage not online")
	}
	if storage.StorageType == api.STORAGE_NVME_PT {
		return httperrors.NewBadRequestError("storage type %s require assign isolated device", api.STORAGE_NVME_PT)
	}
	diskConfig.Storage = storage.Id
	diskConfig.Backend = storage.StorageType
	return nil
}

func parseIsoInfo(ctx context.Context, userCred mcclient.TokenCredential, imageId string) (*cloudprovider.SImage, error) {
	image, err := CachedimageManager.getImageInfo(ctx, userCred, imageId, false)
	if err != nil {
		log.Errorf("getImageInfo fail %s", err)
		return nil, err
	}
	if image.Status != cloudprovider.IMAGE_STATUS_ACTIVE {
		return nil, httperrors.NewInvalidStatusError("Image status is not active")
	}
	return image, nil
}

func (self *SDisk) fetchDiskInfo(diskConfig *api.DiskConfig) error {
	if len(diskConfig.SnapshotId) > 0 {
		self.SnapshotId = diskConfig.SnapshotId
		self.DiskType = diskConfig.DiskType
	}
	if len(diskConfig.BackupId) > 0 {
		self.BackupId = diskConfig.BackupId
		self.DiskType = diskConfig.DiskType
	}
	if len(diskConfig.ImageId) > 0 {
		self.TemplateId = diskConfig.ImageId
		// support for create vm from guest image
		if len(diskConfig.DiskType) == 0 {
			self.DiskType = api.DISK_TYPE_SYS
		} else {
			self.DiskType = diskConfig.DiskType
		}
	}
	if len(diskConfig.Fs) > 0 {
		self.FsFormat = diskConfig.Fs
	}
	self.FsFeatures = diskConfig.FsFeatures
	if err := DiskManager.ValidateFsFeatures(self.FsFormat, diskConfig.FsFeatures); err != nil {
		return err
	}
	if self.FsFormat == "swap" {
		self.DiskType = api.DISK_TYPE_SWAP
		self.Nonpersistent = true
	} else {
		if len(self.DiskType) == 0 {
			diskType := api.DISK_TYPE_DATA
			if diskConfig.DiskType == api.DISK_TYPE_VOLUME {
				diskType = api.DISK_TYPE_VOLUME
			}
			self.DiskType = diskType
		}
		self.Nonpersistent = false
	}
	if len(diskConfig.DiskId) > 0 && utils.IsMatchUUID(diskConfig.DiskId) {
		self.Id = diskConfig.DiskId
	}
	self.DiskFormat = diskConfig.Format
	self.DiskSize = diskConfig.SizeMb
	self.OsArch = diskConfig.OsArch
	return nil
}

type DiskInfo struct {
	ImageId    string
	Fs         string
	MountPoint string
	Format     string
	Size       int64
	Storage    string
	Backend    string
	MediumType string
	Driver     string
	Cache      string
	DiskType   string
}

// DEPRECATE: will be remove in future, use ToDiskConfig
func (self *SDisk) ToDiskInfo() DiskInfo {
	ret := DiskInfo{
		ImageId:    self.GetTemplateId(),
		Fs:         self.GetFsFormat(),
		MountPoint: self.GetMountPoint(),
		Format:     self.DiskFormat,
		Size:       int64(self.DiskSize),
		DiskType:   self.DiskType,
	}
	storage, _ := self.GetStorage()
	if storage == nil {
		return ret
	}
	ret.Storage = storage.Id
	ret.Backend = storage.StorageType
	ret.MediumType = storage.MediumType
	return ret
}

func (self *SDisk) ToDiskConfig() *api.DiskConfig {
	ret := &api.DiskConfig{
		Index:      -1,
		ImageId:    self.GetTemplateId(),
		Fs:         self.GetFsFormat(),
		Mountpoint: self.GetMountPoint(),
		Format:     self.DiskFormat,
		SizeMb:     self.DiskSize,
		DiskType:   self.DiskType,
	}
	storage, _ := self.GetStorage()
	if storage == nil {
		return ret
	}
	ret.Storage = storage.Id
	ret.Backend = storage.StorageType
	ret.Medium = storage.MediumType
	return ret
}

func (self *SDisk) Delete(ctx context.Context, userCred mcclient.TokenCredential) error {
	// override
	log.Infof("disk delete do nothing")
	return nil
}

func (self *SDisk) RealDelete(ctx context.Context, userCred mcclient.TokenCredential) error {
	// diskbackups := DiskBackupManager.Query("id").Equals("disk_id", self.Id)
	guestdisks := GuestdiskManager.Query("row_id").Equals("disk_id", self.Id)
	err := SnapshotPolicyResourceManager.RemoveByResource(self.Id, api.SNAPSHOT_POLICY_TYPE_DISK)
	if err != nil {
		return errors.Wrapf(err, "RemoveByResource")
	}
	pairs := []purgePair{
		// {manager: DiskBackupManager, key: "id", q: diskbackups},
		{manager: GuestdiskManager, key: "row_id", q: guestdisks},
	}
	for i := range pairs {
		err := pairs[i].purgeAll(ctx)
		if err != nil {
			return err
		}
	}
	SnapshotPolicyResourceManager.RemoveByResource(self.Id, api.SNAPSHOT_POLICY_TYPE_DISK)
	return self.SVirtualResourceBase.Delete(ctx, userCred)
}

func (self *SDisk) DoPendingDelete(ctx context.Context, userCred mcclient.TokenCredential) error {
	SnapshotPolicyResourceManager.RemoveByResource(self.Id, api.SNAPSHOT_POLICY_TYPE_DISK)
	return self.SVirtualResourceBase.DoPendingDelete(ctx, userCred)
}

func (self *SDisk) RecordLastAttachedHost(ctx context.Context, userCred mcclient.TokenCredential, hostId string) error {
	storage, err := self.GetStorage()
	if err != nil {
		return err
	}
	if storage.StorageType != api.STORAGE_SLVM {
		return nil
	}
	return self.SetMetadata(ctx, api.DISK_META_LAST_ATTACHED_HOST, hostId, userCred)
}

func (self *SDisk) GetLastAttachedHost(ctx context.Context, userCred mcclient.TokenCredential) string {
	return self.GetMetadata(ctx, api.DISK_META_LAST_ATTACHED_HOST, userCred)
}

func (self *SDisk) RecordDiskSnapshotsLastHost(ctx context.Context, userCred mcclient.TokenCredential, hostId string) error {
	storage, err := self.GetStorage()
	if err != nil {
		return err
	}
	if storage.StorageType != api.STORAGE_SLVM {
		return nil
	}
	// record disk snapshots master host
	snaps := SnapshotManager.GetDiskSnapshots(self.Id)
	for i := range snaps {
		err = snaps[i].SetMetadata(ctx, api.DISK_META_LAST_ATTACHED_HOST, hostId, userCred)
		if err != nil {
			log.Errorf("snapshot %s failed set last attached host: %s", snaps[i].Id, err)
		}
	}
	return nil
}

// 同步磁盘状态
func (self *SDisk) PerformSyncstatus(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject, input api.DiskSyncstatusInput) (jsonutils.JSONObject, error) {
	var openTask = true
	count, err := taskman.TaskManager.QueryTasksOfObject(self, time.Now().Add(-3*time.Minute), &openTask).CountWithError()
	if err != nil {
		return nil, err
	}
	if count > 0 {
		return nil, httperrors.NewBadRequestError("Disk has %d task active, can't sync status", count)
	}

	return nil, self.StartSyncstatus(ctx, userCred, "")
}

func (disk *SDisk) StartSyncstatus(ctx context.Context, userCred mcclient.TokenCredential, parentTaskId string) error {
	return StartResourceSyncStatusTask(ctx, userCred, disk, "DiskSyncstatusTask", parentTaskId)
}

func (self *SDisk) PerformPurge(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject, data jsonutils.JSONObject) (jsonutils.JSONObject, error) {
	err := self.validateDeleteCondition(ctx, true)
	if err != nil {
		return nil, err
	}

	provider := self.GetCloudprovider()
	if provider != nil && utils.IsInStringArray(provider.Provider, []string{api.CLOUD_PROVIDER_HUAWEI, api.CLOUD_PROVIDER_HCSO, api.CLOUD_PROVIDER_HCS}) {
		cnt, err := self.GetSnapshotCount()
		if err != nil {
			return nil, httperrors.NewInternalServerError("GetSnapshotCount fail %s", err)
		}
		if cnt > 0 {
			return nil, httperrors.NewForbiddenError("not allow to purge. Virtual disk must not have snapshots")
		}
	}

	return nil, self.StartDiskDeleteTask(ctx, userCred, "", true, false, false)
}

func (self *SDisk) CustomizeDelete(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject, data jsonutils.JSONObject) error {
	if !jsonutils.QueryBoolean(query, "delete_snapshots", false) {
		if provider := self.GetCloudprovider(); provider != nil && utils.IsInStringArray(provider.Provider, []string{api.CLOUD_PROVIDER_HUAWEI, api.CLOUD_PROVIDER_HCSO, api.CLOUD_PROVIDER_HCS}) {
			cnt, err := self.GetSnapshotCount()
			if err != nil {
				return httperrors.NewInternalServerError("GetSnapshotCount fail %s", err)
			}
			if cnt > 0 {
				return httperrors.NewForbiddenError("not allow to delete. Virtual disk must not have snapshots")
			}
		} else if storage, _ := self.GetStorage(); storage != nil && storage.StorageType == api.STORAGE_RBD {
			scnt, err := self.GetSnapshotCount()
			if err != nil {
				return err
			}
			if scnt > 0 {
				return httperrors.NewBadRequestError("not allow to delete %s disk with snapshots", storage.StorageType)
			}
		}
	}

	return self.StartDiskDeleteTask(ctx, userCred, "", false,
		jsonutils.QueryBoolean(query, "override_pending_delete", false),
		jsonutils.QueryBoolean(query, "delete_snapshots", false))
}

func (manager *SDiskManager) FetchCustomizeColumns(
	ctx context.Context,
	userCred mcclient.TokenCredential,
	query jsonutils.JSONObject,
	objs []interface{},
	fields stringutils2.SSortedStrings,
	isList bool,
) []api.DiskDetails {
	rows := make([]api.DiskDetails, len(objs))
	virtRows := manager.SVirtualResourceBaseManager.FetchCustomizeColumns(ctx, userCred, query, objs, fields, isList)
	storeRows := manager.SStorageResourceBaseManager.FetchCustomizeColumns(ctx, userCred, query, objs, fields, isList)
	encRows := manager.SEncryptedResourceManager.FetchCustomizeColumns(ctx, userCred, query, objs, fields, isList)
	diskIds := make([]string, len(objs))
	for i := range rows {
		rows[i] = api.DiskDetails{
			VirtualResourceDetails: virtRows[i],
			StorageResourceInfo:    storeRows[i],

			EncryptedResourceDetails: encRows[i],
		}

		disk := objs[i].(*SDisk)
		if disk.PendingDeleted {
			pendingDeletedAt := disk.PendingDeletedAt.Add(time.Second * time.Duration(options.Options.PendingDeleteExpireSeconds))
			rows[i].AutoDeleteAt = pendingDeletedAt
		}
		diskIds[i] = disk.Id
	}

	guestSQ := GuestManager.Query().SubQuery()
	gds := GuestdiskManager.Query().SubQuery()
	q := guestSQ.Query(
		guestSQ.Field("id"),
		guestSQ.Field("name"),
		guestSQ.Field("status"),
		guestSQ.Field("billing_type"),
		gds.Field("disk_id"),
		gds.Field("index"),
		gds.Field("driver"),
		gds.Field("cache_mode"),
		gds.Field("iops"),
		gds.Field("bps"),
	).
		Join(gds, sqlchemy.Equals(gds.Field("guest_id"), guestSQ.Field("id"))).
		Filter(sqlchemy.In(gds.Field("disk_id"), diskIds))

	guestInfo := []struct {
		Id     string
		Name   string
		Status string
		DiskId string

		Index       int
		Driver      string
		CacheMode   string
		Iops        int
		Bps         int
		BillingType string
	}{}
	err := q.All(&guestInfo)
	if err != nil {
		log.Errorf("query disk guest info error: %v", err)
		return rows
	}

	guests, guestIds := map[string][]api.SimpleGuest{}, []string{}
	for _, guest := range guestInfo {
		_, ok := guests[guest.DiskId]
		if !ok {
			guests[guest.DiskId] = []api.SimpleGuest{}
			guestIds = append(guestIds, guest.Id)
		}
		guests[guest.DiskId] = append(guests[guest.DiskId], api.SimpleGuest{
			Id:     guest.Id,
			Name:   guest.Name,
			Status: guest.Status,

			Index:       guest.Index,
			Driver:      guest.Driver,
			CacheMode:   guest.CacheMode,
			Iops:        guest.Iops,
			Bps:         guest.Bps,
			BillingType: guest.BillingType,
		})
	}

	policySQ := SnapshotPolicyManager.Query().SubQuery()
	dps := SnapshotPolicyResourceManager.Query().SubQuery()

	q = policySQ.Query(
		policySQ.Field("id"),
		policySQ.Field("name"),
		policySQ.Field("time_points"),
		policySQ.Field("repeat_weekdays"),
		dps.Field("resource_id"),
		dps.Field("resource_type"),
	).Join(dps, sqlchemy.Equals(dps.Field("snapshotpolicy_id"), policySQ.Field("id"))).
		Filter(sqlchemy.OR(sqlchemy.In(dps.Field("resource_id"), diskIds), sqlchemy.In(dps.Field("resource_id"), guestIds)))

	policyInfo := []struct {
		Id             string
		Name           string
		Status         string
		TimePoints     []int
		RepeatWeekdays []int
		ResourceId     string
		ResourceType   string
	}{}
	err = q.All(&policyInfo)
	if err != nil {
		log.Errorf("query disk snapshot policy info error: %v", err)
		return rows
	}

	policies := map[string][]api.SimpleSnapshotPolicy{}
	for _, policy := range policyInfo {
		_, ok := policies[policy.ResourceId]
		if !ok {
			policies[policy.ResourceId] = []api.SimpleSnapshotPolicy{}
		}
		policies[policy.ResourceId] = append(policies[policy.ResourceId], api.SimpleSnapshotPolicy{
			Id:             policy.Id,
			Name:           policy.Name,
			RepeatWeekdays: policy.RepeatWeekdays,
			TimePoints:     policy.TimePoints,
			ResourceType:   policy.ResourceType,
		})
	}

	for i := range rows {
		rows[i].Guests, _ = guests[diskIds[i]]
		names, status, billingTypes := []string{}, []string{}, []string{}
		var iops, bps int
		for j := range rows[i].Guests {
			guest := rows[i].Guests[j]
			names = append(names, guest.Name)
			status = append(status, guest.Status)
			iops = guest.Iops
			bps = guest.Bps
			billingTypes = append(billingTypes, guest.BillingType)
			rows[i].Guests[j].Snapshotpolicy, _ = policies[guest.Id]
			rows[i].GuestSnapshotpolicyCount += len(rows[i].Guests[j].Snapshotpolicy)
		}
		rows[i].GuestCount = len(rows[i].Guests)
		rows[i].Guest = strings.Join(names, ",")
		rows[i].GuestStatus = strings.Join(status, ",")
		rows[i].GuestBillingType = strings.Join(billingTypes, ",")

		rows[i].Snapshotpolicies, _ = policies[diskIds[i]]

		disk := objs[i].(*SDisk)
		if len(disk.StorageId) == 0 && disk.Status == api.VM_SCHEDULE_FAILED {
			rows[i].Brand = "Unknown"
			rows[i].Provider = "Unknown"
		}
		// 仅kvm使用
		if len(rows[i].ManagerId) == 0 {
			disk.Iops = iops
			disk.Throughput = bps
		}
	}
	return rows
}

func (self *SDisk) StartDiskResizeTask(ctx context.Context, userCred mcclient.TokenCredential, sizeMb int64, parentTaskId string, pendingUsage quotas.IQuota) error {
	self.SetStatus(ctx, userCred, api.DISK_START_RESIZE, "StartDiskResizeTask")
	params := jsonutils.NewDict()
	params.Add(jsonutils.NewInt(sizeMb), "size")
	task, err := taskman.TaskManager.NewTask(ctx, "DiskResizeTask", self, userCred, params, parentTaskId, "", pendingUsage)
	if err != nil {
		return err
	}
	task.ScheduleRun(nil)
	return nil
}

func (self *SDisk) StartDiskDeleteTask(
	ctx context.Context, userCred mcclient.TokenCredential, parentTaskId string,
	isPurge, overridePendingDelete, deleteSnapshots bool,
) error {
	params := jsonutils.NewDict()
	if isPurge {
		params.Add(jsonutils.JSONTrue, "purge")
	}
	if overridePendingDelete {
		params.Add(jsonutils.JSONTrue, "override_pending_delete")
	}
	if deleteSnapshots {
		params.Add(jsonutils.JSONTrue, "delete_snapshots")
	}
	task, err := taskman.TaskManager.NewTask(ctx, "DiskDeleteTask", self, userCred, params, parentTaskId, "", nil)
	if err != nil {
		log.Errorf("%s", err)
		return err
	}
	task.ScheduleRun(nil)
	return nil
}

func (self *SDisk) GetAttachedGuests() []SGuest {
	guests := GuestManager.Query().SubQuery()
	guestdisks := GuestdiskManager.Query().SubQuery()

	q := guests.Query()
	q = q.Join(guestdisks, sqlchemy.AND(sqlchemy.Equals(guestdisks.Field("guest_id"), guests.Field("id")),
		sqlchemy.IsFalse(guestdisks.Field("deleted"))))
	q = q.Filter(sqlchemy.Equals(guestdisks.Field("disk_id"), self.Id))

	ret := make([]SGuest, 0)
	if err := db.FetchModelObjects(GuestManager, q, &ret); err != nil {
		log.Errorf("Fetch Geusts Objects %v", err)
		return nil
	}
	return ret
}

func (self *SDisk) SetDiskReady(ctx context.Context, userCred mcclient.TokenCredential, reason string) {
	self.SetStatus(ctx, userCred, api.DISK_READY, reason)
	guests := self.GetAttachedGuests()
	if guests != nil {
		for _, guest := range guests {
			guest.StartSyncstatus(ctx, userCred, "")
		}
	}
}

func (self *SDisk) SwitchToBackup(userCred mcclient.TokenCredential) error {
	diff, err := db.Update(self, func() error {
		self.StorageId, self.BackupStorageId = self.BackupStorageId, self.StorageId
		return nil
	})
	if err != nil {
		log.Errorf("SwitchToBackup fail %s", err)
		return err
	}
	db.OpsLog.LogEvent(self, db.ACT_UPDATE, diff, userCred)
	return nil
}

func (self *SDisk) ClearHostSchedCache() error {
	storage, _ := self.GetStorage()
	if storage == nil {
		return fmt.Errorf("no valid storage")
	}
	hosts := storage.GetAllAttachingHosts()
	if hosts == nil {
		return fmt.Errorf("get attaching host error")
	}
	for _, h := range hosts {
		err := h.ClearSchedDescCache()
		if err != nil {
			return err
		}
	}
	return nil
}

func (self *SDisk) GetShortDesc(ctx context.Context) *jsonutils.JSONDict {
	desc := self.SVirtualResourceBase.GetShortDesc(ctx)
	desc.Add(jsonutils.NewInt(int64(self.DiskSize)), "size")
	desc.Add(jsonutils.NewString(self.DiskType), "disk_type")
	storage, _ := self.GetStorage()
	if storage != nil {
		desc.Add(jsonutils.NewString(storage.StorageType), "storage_type")
		desc.Add(jsonutils.NewString(storage.MediumType), "medium_type")
	}

	if hypervisor := self.GetMetadata(ctx, "hypervisor", nil); len(hypervisor) > 0 {
		desc.Add(jsonutils.NewString(hypervisor), "hypervisor")
	}

	if len(self.ExternalId) > 0 {
		desc.Add(jsonutils.NewString(self.ExternalId), "externalId")
	}

	if self.IsSsd {
		desc.Add(jsonutils.JSONTrue, "is_ssd")
	}

	fs := self.GetFsFormat()
	if len(fs) > 0 {
		desc.Add(jsonutils.NewString(fs), "fs_format")
	}
	tid := self.GetTemplateId()
	if len(tid) > 0 {
		desc.Add(jsonutils.NewString(tid), "template_id")
	}

	var billingInfo SCloudBillingInfo

	if storage != nil {
		billingInfo.SCloudProviderInfo = storage.getCloudProviderInfo()
	}

	if priceKey := self.GetMetadata(ctx, "ext:price_key", nil); len(priceKey) > 0 {
		billingInfo.PriceKey = priceKey
	}

	billingInfo.SBillingBaseInfo = self.getBillingBaseInfo()

	desc.Update(jsonutils.Marshal(billingInfo))

	return desc
}

func (self *SDisk) getDev() string {
	return self.GetMetadata(context.Background(), "dev", nil)
}

func (self *SDisk) GetMountPoint() string {
	return self.GetMetadata(context.Background(), "mountpoint", nil)
}

func (self *SDisk) isReady() bool {
	return self.Status == api.DISK_READY
}

func (self *SDisk) isInit() bool {
	return self.Status == api.DISK_INIT
}

func (self *SDisk) PerformCancelDelete(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject, data jsonutils.JSONObject) (jsonutils.JSONObject, error) {
	if self.PendingDeleted && !self.Deleted {
		err := self.DoCancelPendingDelete(ctx, userCred)
		if err != nil {
			return nil, err
		}
		self.RecoverUsages(ctx, userCred)
	}
	return nil, nil
}

func (manager *SDiskManager) getExpiredPendingDeleteDisks() []SDisk {
	deadline := time.Now().Add(time.Duration(options.Options.PendingDeleteExpireSeconds*-1) * time.Second)

	q := manager.Query()
	q = q.IsTrue("pending_deleted").LT("pending_deleted_at", deadline).Limit(options.Options.PendingDeleteMaxCleanBatchSize)

	disks := make([]SDisk, 0)
	err := db.FetchModelObjects(DiskManager, q, &disks)
	if err != nil {
		log.Errorf("fetch disks error %s", err)
		return nil
	}

	return disks
}

func (manager *SDiskManager) CleanPendingDeleteDisks(ctx context.Context, userCred mcclient.TokenCredential, isStart bool) {
	disks := manager.getExpiredPendingDeleteDisks()
	if disks == nil {
		return
	}
	for i := 0; i < len(disks); i += 1 {
		disks[i].StartDiskDeleteTask(ctx, userCred, "", false, true, false)
	}
}

func (manager *SDiskManager) GetNeedAutoSnapshotDisks() ([]SSnapshotPolicyResource, error) {
	tz, _ := time.LoadLocation(options.Options.TimeZone)
	t := time.Now().In(tz)
	week := t.Weekday()
	if week == 0 { // sunday is zero
		week += 7
	}
	timePoint := t.Hour()

	policy := SnapshotPolicyManager.Query().Equals("type", api.SNAPSHOT_POLICY_TYPE_DISK).Equals("cloudregion_id", api.DEFAULT_REGION_ID)
	policy = policy.Filter(sqlchemy.Contains(policy.Field("repeat_weekdays"), fmt.Sprintf("%d", week)))
	sq := policy.Filter(
		sqlchemy.OR(
			sqlchemy.Contains(policy.Field("time_points"), fmt.Sprintf(",%d,", timePoint)),
			sqlchemy.Startswith(policy.Field("time_points"), fmt.Sprintf("[%d,", timePoint)),
			sqlchemy.Endswith(policy.Field("time_points"), fmt.Sprintf(",%d]", timePoint)),
			sqlchemy.Equals(policy.Field("time_points"), fmt.Sprintf("[%d]", timePoint)),
		),
	).SubQuery()
	disks := DiskManager.Query().SubQuery()
	q := SnapshotPolicyResourceManager.Query().Equals("resource_type", api.SNAPSHOT_POLICY_TYPE_DISK)
	q = q.Join(sq, sqlchemy.Equals(q.Field("snapshotpolicy_id"), sq.Field("id")))
	q = q.Join(disks, sqlchemy.Equals(q.Field("resource_id"), disks.Field("id")))
	ret := []SSnapshotPolicyResource{}
	err := db.FetchModelObjects(SnapshotPolicyResourceManager, q, &ret)
	if err != nil {
		return nil, err
	}
	return ret, nil
}

func (disk *SDisk) validateDiskAutoCreateSnapshot() error {
	guests := disk.GetGuests()
	if len(guests) == 0 {
		return fmt.Errorf("Disks %s not attach guest, can't create snapshot", disk.GetName())
	}
	storage, err := disk.GetStorage()
	if err != nil {
		return errors.Wrapf(err, "GetStorage")
	}
	if len(guests) == 1 && utils.IsInStringArray(storage.StorageType, api.FIEL_STORAGE) {
		if !utils.IsInStringArray(guests[0].Status, []string{api.VM_RUNNING, api.VM_READY}) {
			return fmt.Errorf("Guest(%s) in status(%s) cannot do disk snapshot", guests[0].Id, guests[0].Status)
		}
	}
	if storageFree := storage.GetFreeCapacity(); storageFree < int64(disk.DiskSize) {
		return fmt.Errorf("Storage(%s) space not enough", storage.GetName())
	}
	return nil
}

func (manager *SDiskManager) AutoDiskSnapshot(ctx context.Context, userCred mcclient.TokenCredential, isStart bool) {
	disks, err := manager.GetNeedAutoSnapshotDisks()
	if err != nil {
		log.Errorf("Get auto snapshot disks id failed: %s", err)
		return
	}
	log.Infof("auto snapshot %d disks", len(disks))

	guestDps := map[string][]SSnapshotPolicyResource{}
	for i := 0; i < len(disks); i++ {
		disk, err := disks[i].GetDisk()
		if err != nil {
			log.Errorf("get disk error: %v", err)
			continue
		}
		if guest := disk.GetGuest(); guest != nil {
			if dps, ok := guestDps[guest.Id]; ok {
				guestDps[guest.Id] = append(dps, disks[i])
			} else {
				guestDps[guest.Id] = []SSnapshotPolicyResource{disks[i]}
			}
			continue
		}

		err = manager.DoAutoSnapshot(ctx, userCred, &disks[i], disk, "")
		if err != nil {
			log.Errorf("auto snapshot %s error: %v", disk.Name, err)
			db.OpsLog.LogEvent(disk, db.ACT_DISK_AUTO_SNAPSHOT_FAIL, err.Error(), userCred)
			notifyclient.NotifySystemErrorWithCtx(ctx, disk.Id, disk.Name, db.ACT_DISK_AUTO_SNAPSHOT_FAIL, errors.Wrapf(err, "Disk auto create snapshot").Error())
		}
	}

	for gid, diskSnapshotPolicies := range guestDps {
		guest := GuestManager.FetchGuestById(gid)
		err = manager.OrderCreateDisksSnapshotsBySnapshotPolicy(ctx, userCred, guest, diskSnapshotPolicies)
		if err != nil {
			log.Errorf("failed start OrderCreateDisksSnapshotsBySnapshotPolicy")
		}
	}
}

func (manager *SDiskManager) OrderCreateDisksSnapshotsBySnapshotPolicy(
	ctx context.Context, userCred mcclient.TokenCredential, guest *SGuest, snapshotPolicyDisks []SSnapshotPolicyResource,
) error {
	params := jsonutils.NewDict()
	params.Set("snapshot_policy_disks", jsonutils.Marshal(snapshotPolicyDisks))

	task, err := taskman.TaskManager.NewTask(ctx, "GuestDisksSnapshotPolicyExecuteTask", guest, userCred, params, "", "", nil)
	if err != nil {
		return errors.Wrapf(err, "NewTask")
	}
	return task.ScheduleRun(nil)
}

func (manager *SDiskManager) DoAutoSnapshot(
	ctx context.Context, userCred mcclient.TokenCredential,
	diskSnapshotPolicy *SSnapshotPolicyResource, disk *SDisk, parentTaskId string,
) error {
	policy, err := diskSnapshotPolicy.GetSnapshotPolicy()
	if err != nil {
		return errors.Wrapf(err, "GetSnapshotPolicy")
	}

	if len(disk.ExternalId) == 0 {
		err = disk.validateDiskAutoCreateSnapshot()
		if err != nil {
			return errors.Wrapf(err, "validateDiskAutoCreateSnapshot")
		}
	}

	snapshot, err := disk.CreateSnapshotAuto(ctx, userCred, policy, parentTaskId)
	if err != nil {
		return errors.Wrapf(err, "CreateSnapshotAuto")
	}

	db.OpsLog.LogEvent(disk, db.ACT_DISK_AUTO_SNAPSHOT, snapshot.Name, userCred)
	policy.ExecuteNotify(ctx, userCred, disk.GetName())
	return nil
}

func (self *SDisk) CreateSnapshotAuto(
	ctx context.Context, userCred mcclient.TokenCredential, policy *SSnapshotPolicy, parentTaskId string,
) (*SSnapshot, error) {
	storage, err := self.GetStorage()
	if err != nil {
		return nil, errors.Wrapf(err, "GetStorage")
	}
	snapshot := &SSnapshot{}
	snapshot.SetModelManager(SnapshotManager, snapshot)
	snapshot.ProjectId = self.ProjectId
	snapshot.DomainId = self.DomainId
	snapshot.DiskId = self.Id
	if len(self.ExternalId) == 0 {
		snapshot.StorageId = self.StorageId
	}

	// inherit encrypt_key_id
	snapshot.EncryptKeyId = self.EncryptKeyId

	driver, err := storage.GetRegionDriver()
	if err != nil {
		return nil, errors.Wrapf(err, "GetRegionDriver")
	}
	snapshot.OutOfChain = driver.SnapshotIsOutOfChain(self)
	snapshot.VirtualSize = self.DiskSize
	snapshot.DiskType = self.DiskType
	snapshot.Location = ""
	snapshot.CreatedBy = api.SNAPSHOT_AUTO
	snapshot.ManagerId = storage.ManagerId
	if cloudregion, _ := storage.GetRegion(); cloudregion != nil {
		snapshot.CloudregionId = cloudregion.GetId()
	}
	snapshot.Name = fmt.Sprintf("%s-auto-snapshot-%d", self.Name, time.Now().Unix())
	snapshot.Status = api.SNAPSHOT_CREATING
	if policy.RetentionDays > 0 {
		snapshot.ExpiredAt = time.Now().AddDate(0, 0, policy.RetentionDays)
	}
	snapshot.IsSystem = self.IsSystem
	err = SnapshotManager.TableSpec().Insert(ctx, snapshot)
	if err != nil {
		return nil, errors.Wrapf(err, "Insert")
	}

	db.OpsLog.LogEvent(snapshot, db.ACT_CREATE, "disk create snapshot auto", userCred)
	err = snapshot.StartSnapshotCreateTask(ctx, userCred, nil, parentTaskId)
	if err != nil {
		return nil, errors.Wrap(err, "disk auto snapshot start snapshot task")
	}
	return snapshot, nil
}

func (self *SDisk) StartCreateBackupTask(ctx context.Context, userCred mcclient.TokenCredential, parentTaskId string) error {
	if task, err := taskman.TaskManager.NewTask(ctx, "DiskCreateBackupTask", self, userCred, nil, parentTaskId, "", nil); err != nil {
		log.Errorln(err)
		return err
	} else {
		task.ScheduleRun(nil)
	}
	return nil
}

func (self *SDisk) DeleteSnapshots(ctx context.Context, userCred mcclient.TokenCredential, parentTaskId string) error {
	if task, err := taskman.TaskManager.NewTask(ctx, "DiskDeleteSnapshotsTask", self, userCred, nil, parentTaskId, "", nil); err != nil {
		log.Errorln(err)
		return err
	} else {
		task.ScheduleRun(nil)
	}
	return nil
}

func (self *SDisk) IsDetachable() bool {
	storage, _ := self.GetStorage()
	if storage == nil {
		return true
	}
	if storage.IsLocal() {
		return false
	}
	if self.BillingType == billing_api.BILLING_TYPE_PREPAID {
		return false
	}
	if utils.IsInStringArray(self.DiskType, []string{api.DISK_TYPE_SYS, api.DISK_TYPE_SWAP}) {
		return false
	}
	if self.AutoDelete {
		return false
	}
	return true
}

func (self *SDisk) GetDynamicConditionInput() *jsonutils.JSONDict {
	conf := self.ToDiskConfig()
	return conf.JSON(conf)
}

func (self *SDisk) IsNeedWaitSnapshotsDeleted() (bool, error) {
	storage, _ := self.GetStorage()
	if storage == nil {
		return false, fmt.Errorf("no valid storage")
	}
	if storage.StorageType == api.STORAGE_RBD {
		scnt, err := self.GetSnapshotCount()
		if err != nil {
			return false, err
		}
		if scnt > 0 {
			return true, nil
		}
	}
	return false, nil
}

func (self *SDisk) UpdataSnapshotsBackingDisk(backingDiskId string) error {
	snapshots := make([]SSnapshot, 0)
	err := SnapshotManager.Query().Equals("disk_id", self.Id).IsNullOrEmpty("backing_disk_id").All(&snapshots)
	if err != nil {
		return err
	}
	for i := 0; i < len(snapshots); i++ {
		snapshots[i].SetModelManager(SnapshotManager, &snapshots[i])
		_, err := db.Update(&snapshots[i], func() error {
			snapshots[i].BackingDiskId = backingDiskId
			return nil
		})
		if err != nil {
			log.Errorln(err)
		}
	}
	return nil
}

func (self *SDisk) GetSnapshotsNotInInstanceSnapshot() ([]SSnapshot, error) {
	snapshots := make([]SSnapshot, 0)
	sq := InstanceSnapshotJointManager.Query("snapshot_id").SubQuery()
	q := SnapshotManager.Query().IsFalse("fake_deleted").Equals("disk_id", self.Id)
	q = q.LeftJoin(sq, sqlchemy.Equals(q.Field("id"), sq.Field("snapshot_id"))).
		Filter(sqlchemy.IsNull(sq.Field("snapshot_id")))
	err := db.FetchModelObjects(SnapshotManager, q, &snapshots)
	if err != nil {
		log.Errorf("Fetch db snapshots failed %s", err)
		return nil, err
	}
	return snapshots, nil
}

func (self *SDisk) PerformChangeOwner(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject,
	input apis.PerformChangeProjectOwnerInput) (jsonutils.JSONObject, error) {

	_, err := self.SVirtualResourceBase.PerformChangeOwner(ctx, userCred, query, input)
	if err != nil {
		return nil, err
	}
	snapshotQuery := SnapshotManager.Query().Equals("disk_id", self.Id)
	snapshots := make([]SSnapshot, 0, 1)
	err = db.FetchModelObjects(SnapshotManager, snapshotQuery, &snapshots)
	if err != nil {
		return nil, errors.Wrapf(err, "fail to fetch snapshots of disk %s", self.Id)
	}
	for i := range snapshots {
		snapshot := snapshots[i]
		err := func() error {
			lockman.LockObject(ctx, &snapshot)
			defer lockman.ReleaseObject(ctx, &snapshot)
			_, err := snapshot.PerformChangeOwner(ctx, userCred, query, input)
			if err != nil {
				return err
			}
			return nil
		}()
		if err != nil {
			return nil, errors.Wrapf(err, "fail to change owner of this disk(%s)'s snapshot %s", self.Id, snapshot.Id)
		}
	}
	return nil, nil
}

func (disk *SDisk) GetUsages() []db.IUsage {
	if disk.PendingDeleted || disk.Deleted {
		return nil
	}
	usage := SQuota{Storage: disk.DiskSize}
	keys, err := disk.GetQuotaKeys()
	if err != nil {
		log.Errorf("disk.GetQuotaKeys fail %s", err)
		return nil
	}
	usage.SetKeys(keys)
	return []db.IUsage{
		&usage,
	}
}

// 绑定磁盘快照策略
// 磁盘只能绑定一个快照策略，已绑定时报错
// 若磁盘所属主机已绑定主机快照策略，则磁盘不能再绑定快照策略
func (disk *SDisk) PerformBindSnapshotpolicy(
	ctx context.Context,
	userCred mcclient.TokenCredential,
	query jsonutils.JSONObject,
	input *api.DiskSnapshotpolicyInput,
) (jsonutils.JSONObject, error) {
	// 磁盘只能绑定一个快照策略，已绑定时报错
	cnt, err := SnapshotPolicyResourceManager.GetBindingCount(disk.Id, api.SNAPSHOT_POLICY_TYPE_DISK)
	if err != nil {
		return nil, errors.Wrap(err, "GetBindingCount")
	}
	if cnt > 0 {
		return nil, httperrors.NewConflictError("disk already bound to a snapshot policy")
	}
	// 若磁盘所属主机已绑定主机快照策略，则磁盘不能再绑定快照策略
	if guest := disk.GetGuest(); guest != nil {
		guestCnt, err := SnapshotPolicyResourceManager.GetBindingCount(guest.Id, api.SNAPSHOT_POLICY_TYPE_SERVER)
		if err != nil {
			return nil, errors.Wrap(err, "GetBindingCount for guest")
		}
		if guestCnt > 0 {
			return nil, httperrors.NewConflictError("guest already has server snapshot policy, disk cannot bind snapshot policy")
		}
	}
	spObj, err := validators.ValidateModel(ctx, userCred, SnapshotPolicyManager, &input.SnapshotpolicyId)
	if err != nil {
		return nil, err
	}
	sp := spObj.(*SSnapshotPolicy)
	if len(sp.ManagerId) > 0 {
		storage, err := disk.GetStorage()
		if err != nil {
			return nil, errors.Wrapf(err, "GetStorage")
		}
		if storage.ManagerId != sp.ManagerId {
			return nil, httperrors.NewConflictError("The snapshot policy %s and disk account are different", sp.Name)
		}
		zone, err := storage.GetZone()
		if err != nil {
			return nil, errors.Wrapf(err, "GetZone")
		}
		if sp.CloudregionId != zone.CloudregionId {
			return nil, httperrors.NewConflictError("The snapshot policy %s and the disk are in different region", sp.Name)
		}
	}
	return nil, sp.StartBindDisksTask(ctx, userCred, []string{disk.Id})
}

// 设置磁盘快照策略
// 可覆盖当前磁盘绑定的快照策略，若磁盘所属主机已绑定主机快照策略，则自动解除主机快照策略
func (disk *SDisk) PerformSetSnapshotpolicy(
	ctx context.Context,
	userCred mcclient.TokenCredential,
	query jsonutils.JSONObject,
	input *api.DiskSnapshotpolicyInput,
) (jsonutils.JSONObject, error) {
	spObj, err := validators.ValidateModel(ctx, userCred, SnapshotPolicyManager, &input.SnapshotpolicyId)
	if err != nil {
		return nil, err
	}
	sp := spObj.(*SSnapshotPolicy)
	if sp.Type != api.SNAPSHOT_POLICY_TYPE_DISK {
		return nil, httperrors.NewBadRequestError("The snapshot policy %s is not a disk snapshot policy", sp.Name)
	}
	if len(sp.ManagerId) > 0 {
		storage, err := disk.GetStorage()
		if err != nil {
			return nil, errors.Wrapf(err, "GetStorage")
		}
		if storage.ManagerId != sp.ManagerId {
			return nil, httperrors.NewConflictError("The snapshot policy %s and disk account are different", sp.Name)
		}
		zone, err := storage.GetZone()
		if err != nil {
			return nil, errors.Wrapf(err, "GetZone")
		}
		if sp.CloudregionId != zone.CloudregionId {
			return nil, httperrors.NewConflictError("The snapshot policy %s and the disk are in different region", sp.Name)
		}
	}
	// 先解除当前绑定再绑定新策略
	if err := SnapshotPolicyResourceManager.RemoveByResource(disk.Id, api.SNAPSHOT_POLICY_TYPE_DISK); err != nil {
		return nil, errors.Wrap(err, "RemoveByResource")
	}
	// 若磁盘所属主机已绑定主机快照策略，则磁盘不能再绑定快照策略
	if guest := disk.GetGuest(); guest != nil {
		if err := SnapshotPolicyResourceManager.RemoveByResource(guest.Id, api.SNAPSHOT_POLICY_TYPE_SERVER); err != nil {
			return nil, errors.Wrap(err, "RemoveByResource")
		}
	}
	return nil, sp.StartBindDisksTask(ctx, userCred, []string{disk.Id})
}

// 解绑自动快照策略
func (disk *SDisk) PerformUnbindSnapshotpolicy(
	ctx context.Context,
	userCred mcclient.TokenCredential,
	query jsonutils.JSONObject,
	input *api.DiskSnapshotpolicyInput,
) (jsonutils.JSONObject, error) {
	spObj, err := validators.ValidateModel(ctx, userCred, SnapshotPolicyManager, &input.SnapshotpolicyId)
	if err != nil {
		return nil, err
	}
	sp := spObj.(*SSnapshotPolicy)
	return nil, sp.StartUnbindDisksTask(ctx, userCred, []string{disk.Id})
}

func (manager *SDiskManager) ListItemExportKeys(ctx context.Context,
	q *sqlchemy.SQuery,
	userCred mcclient.TokenCredential,
	keys stringutils2.SSortedStrings,
) (*sqlchemy.SQuery, error) {
	var err error

	q, err = manager.SVirtualResourceBaseManager.ListItemExportKeys(ctx, q, userCred, keys)
	if err != nil {
		return nil, errors.Wrap(err, "SVirtualResourceBaseManager.ListItemExportKeys")
	}
	if keys.ContainsAny(manager.SStorageResourceBaseManager.GetExportKeys()...) {
		q, err = manager.SStorageResourceBaseManager.ListItemExportKeys(ctx, q, userCred, keys)
		if err != nil {
			return nil, errors.Wrap(err, "SStorageResourceBaseManager.ListItemExportKeys")
		}
	}
	return q, nil
}

func (disk *SDisk) PerformRebuild(
	ctx context.Context,
	userCred mcclient.TokenCredential,
	query jsonutils.JSONObject,
	input api.DiskRebuildInput,
) (jsonutils.JSONObject, error) {
	guests := disk.GetGuests()
	for _, guest := range guests {
		if guest.GetStatus() != api.VM_READY {
			return nil, httperrors.NewInvalidStatusError("Guest %s status is %s", guest.GetId(), guest.GetStatus())
		}
		guest.SetStatus(ctx, userCred, api.VM_DISK_RESET, "disk rebuild")
	}
	err := disk.resetDiskinfo(ctx, userCred, input)
	if err != nil {
		return nil, errors.Wrap(err, "disk.resetDiskinfo")
	}
	disk.SetStatus(ctx, userCred, api.DISK_REBUILD, "disk rebuild")
	return nil, disk.StartDiskCreateTask(ctx, userCred, true, disk.SnapshotId, "")
}

func (disk *SDisk) resetDiskinfo(
	ctx context.Context,
	userCred mcclient.TokenCredential,
	input api.DiskRebuildInput,
) error {
	if disk.DiskFormat != "raw" {
		return errors.Wrapf(errors.ErrInvalidStatus, "disk_format must be raw, not %s", disk.DiskFormat)
	}
	if len(disk.FsFormat) == 0 {
		return errors.Wrap(errors.ErrInvalidStatus, "fs_format must be set")
	}
	if input.Fs != nil {
		if disk.FsFeatures != nil {
			err := DiskManager.ValidateFsFeatures(*input.Fs, input.FsFeatures)
			if err != nil {
				return err
			}
		}
	}

	if input.TemplateId != nil {
		if len(*input.TemplateId) > 0 {
			imageObj, err := CachedimageManager.FetchByIdOrName(ctx, userCred, *input.TemplateId)
			if err != nil {
				if errors.Cause(err) == sql.ErrNoRows {
					return httperrors.NewResourceNotFoundError2(CachedimageManager.Keyword(), *input.TemplateId)
				} else {
					return errors.Wrap(err, "CachedimageManager.FetchById")
				}
			}
			image := imageObj.(*SCachedimage)
			input.TemplateId = &image.Id
		}
	}
	if input.BackupId != nil {
		if len(*input.BackupId) > 0 {
			bkObj, err := DiskBackupManager.FetchByIdOrName(ctx, userCred, *input.BackupId)
			if err != nil {
				if errors.Cause(err) == sql.ErrNoRows {
					return httperrors.NewResourceNotFoundError2(DiskBackupManager.Keyword(), *input.BackupId)
				} else {
					return errors.Wrap(err, "DiskBackupManager.FetchByIdOrName")
				}
			}
			backup := bkObj.(*SDiskBackup)
			input.BackupId = &backup.Id
		}
	}
	diskSize := 0
	if input.Size != nil {
		size, err := fileutils.GetSizeMb(*input.Size, 'M', 1024)
		if err != nil {
			return errors.Wrapf(err, "GetSizeMb %s", *input.Size)
		}
		diskSize = size
	}
	notes, err := db.Update(disk, func() error {
		if input.TemplateId != nil {
			disk.TemplateId = *input.TemplateId
		}
		if input.BackupId != nil {
			disk.BackupId = *input.BackupId
		}
		if diskSize > 0 {
			disk.DiskSize = diskSize
		}
		if input.Fs != nil {
			disk.FsFormat = *input.Fs
			disk.FsFeatures = input.FsFeatures
		}
		return nil
	})
	if err != nil {
		logclient.AddActionLogWithContext(ctx, disk, logclient.ACT_UPDATE, err, userCred, false)
		return errors.Wrap(err, "Update")
	}
	logclient.AddActionLogWithContext(ctx, disk, logclient.ACT_UPDATE, err, userCred, true)
	db.OpsLog.LogEvent(disk, db.ACT_UPDATE, notes, userCred)
	return nil
}

func (disk *SDisk) PerformChangeBillingType(ctx context.Context, userCred mcclient.TokenCredential, _ jsonutils.JSONObject, input *api.DiskChangeBillingTypeInput) (jsonutils.JSONObject, error) {
	if !utils.IsInStringArray(disk.Status, []string{api.DISK_READY}) {
		return nil, httperrors.NewServerStatusError("Cannot change disk billing type in status %s", disk.Status)
	}
	if len(input.BillingType) == 0 {
		return nil, httperrors.NewMissingParameterError("billing_type")
	}
	if !utils.IsInStringArray(string(input.BillingType), []string{string(billing_api.BILLING_TYPE_POSTPAID), string(billing_api.BILLING_TYPE_PREPAID)}) {
		return nil, httperrors.NewInputParameterError("invalid billing_type %s", input.BillingType)
	}
	if disk.BillingType == input.BillingType {
		return nil, nil
	}
	return nil, disk.StartChangeBillingTypeTask(ctx, userCred, "")
}

func (disk *SDisk) StartChangeBillingTypeTask(ctx context.Context, userCred mcclient.TokenCredential, parentTaskId string) error {
	disk.SetStatus(ctx, userCred, apis.STATUS_CHANGE_BILLING_TYPE, "")
	kwargs := jsonutils.NewDict()
	task, err := taskman.TaskManager.NewTask(ctx, "DiskChangeBillingTypeTask", disk, userCred, kwargs, parentTaskId, "", nil)
	if err != nil {
		return err
	}
	return task.ScheduleRun(nil)
}