support sync disk snapshot status

pkg/apis/compute/disk_const.go

@@ -55,4 +55,7 @@ const (
 	DISK_BACKING_IMAGE = "image"
 
 	DISK_SIZE_AUTOEXTEND = -1
+
+	DISK_NOT_EXIST = "not_exist"
+	DISK_EXIST     = "exist"
 )

pkg/apis/compute/snapshot_const.go

@@ -52,4 +52,7 @@ const (
 	SNAPSHOT_POLICY_CACHE_STATUS_READY         = "ready"
 	SNAPSHOT_POLICY_CACHE_STATUS_DELETING      = "deleting"
 	SNAPSHOT_POLICY_CACHE_STATUS_DELETE_FAILED = "delete_failed"
+
+	SNAPSHOT_EXIST     = "exist"
+	SNAPSHOT_NOT_EXIST = "not_exist"
 )

pkg/compute/regiondrivers/kvm.go

@@ -33,6 +33,7 @@ import (
 	"yunion.io/x/onecloud/pkg/httperrors"
 	"yunion.io/x/onecloud/pkg/mcclient"
 	"yunion.io/x/onecloud/pkg/util/choices"
+	"yunion.io/x/onecloud/pkg/util/httputils"
 	"yunion.io/x/onecloud/pkg/util/rand"
 )
 
@@ -925,6 +926,52 @@ func (self *SKVMRegionDriver) OnSnapshotDelete(ctx context.Context, snapshot *mo
 	return nil
 }
 
+func (self *SKVMRegionDriver) RequestSyncDiskStatus(ctx context.Context, userCred mcclient.TokenCredential, disk *models.SDisk, task taskman.ITask) error {
+	taskman.LocalTaskRun(task, func() (jsonutils.JSONObject, error) {
+		storage := disk.GetStorage()
+		host := storage.GetMasterHost()
+		header := task.GetTaskRequestHeader()
+		url := fmt.Sprintf("%s/disks/%s/%s/status", host.ManagerUri, storage.Id, disk.Id)
+		_, res, err := httputils.JSONRequest(httputils.GetDefaultClient(), ctx, "GET", url, header, nil, false)
+		if err != nil {
+			return nil, err
+		}
+		var diskStatus string
+		originStatus, _ := task.GetParams().GetString("origin_status")
+		status, _ := res.GetString("status")
+		if status == api.DISK_EXIST {
+			diskStatus = originStatus
+		} else {
+			diskStatus = api.DISK_UNKNOWN
+		}
+		return nil, disk.SetStatus(userCred, diskStatus, "sync status")
+	})
+	return nil
+}
+
+func (self *SKVMRegionDriver) RequestSyncSnapshotStatus(ctx context.Context, userCred mcclient.TokenCredential, snapshot *models.SSnapshot, task taskman.ITask) error {
+	taskman.LocalTaskRun(task, func() (jsonutils.JSONObject, error) {
+		storage := snapshot.GetStorage()
+		host := storage.GetMasterHost()
+		header := task.GetTaskRequestHeader()
+		url := fmt.Sprintf("%s/snapshots/%s/%s/%s/status", host.ManagerUri, storage.Id, snapshot.DiskId, snapshot.Id)
+		_, res, err := httputils.JSONRequest(httputils.GetDefaultClient(), ctx, "GET", url, header, nil, false)
+		if err != nil {
+			return nil, err
+		}
+		var snapshotStatus string
+		originStatus, _ := task.GetParams().GetString("origin_status")
+		status, _ := res.GetString("status")
+		if status == api.SNAPSHOT_EXIST {
+			snapshotStatus = originStatus
+		} else {
+			snapshotStatus = api.SNAPSHOT_UNKNOWN
+		}
+		return nil, snapshot.SetStatus(userCred, snapshotStatus, "sync status")
+	})
+	return nil
+}
+
 func (self *SKVMRegionDriver) DealNatGatewaySpec(spec string) string {
 	return spec
 }

pkg/hostman/storageman/diskhandlers/diskhandler.go

@@ -20,8 +20,10 @@ import (
 	"net/http"
 
 	"yunion.io/x/jsonutils"
+	"yunion.io/x/log"
 	"yunion.io/x/pkg/util/regutils"
 
+	"yunion.io/x/onecloud/pkg/apis/compute"
 	"yunion.io/x/onecloud/pkg/appsrv"
 	"yunion.io/x/onecloud/pkg/cloudcommon/workmanager"
 	"yunion.io/x/onecloud/pkg/hostman/guestman"
@@ -32,7 +34,9 @@ import (
 )
 
 var (
-	keyWords    = []string{"disks"}
+	keyWords         = []string{"disks"}
+	snapshotKeywords = []string{"snapshots"}
+
 	actionFuncs = map[string]actionFunc{
 		"create":            diskCreate,
 		"delete":            diskDelete,
@@ -66,7 +70,17 @@ func AddDiskHandler(prefix string, app *appsrv.Application) {
 		app.AddHandler("POST",
 			fmt.Sprintf("%s/%s/<storageId>/<action>/<diskId>", prefix, keyWord),
 			auth.Authenticate(perfomrDiskActions))
+		app.AddHandler("GET",
+			fmt.Sprintf("%s/%s/<storageId>/<diskId>/status", prefix, keyWord),
+			auth.Authenticate(getDiskStatus))
+	}
+	for _, keyWord := range snapshotKeywords {
+		app.AddHandler("GET",
+			fmt.Sprintf("%s/%s/<storageId>/<diskId>/<snapshotId>/status", prefix, keyWord),
+			auth.Authenticate(getSnapshotStatus),
+		)
 	}
+
 }
 
 func performImageCache(
@@ -113,6 +127,59 @@ func deleteImageCache(ctx context.Context, w http.ResponseWriter, r *http.Reques
 
 }
 
+func getDiskStatus(ctx context.Context, w http.ResponseWriter, r *http.Request) {
+	params, _, _ := appsrv.FetchEnv(ctx, w, r)
+	var (
+		storageId = params["<storageId>"]
+		diskId    = params["<diskId>"]
+	)
+
+	storage := storageman.GetManager().GetStorage(storageId)
+	if storage == nil {
+		hostutils.Response(ctx, w, httperrors.NewNotFoundError("Storage %s not found", storageId))
+		return
+	}
+	ret := jsonutils.NewDict()
+	disk := storage.GetDiskById(diskId)
+	if disk == nil {
+		ret.Set("status", jsonutils.NewString(compute.DISK_NOT_EXIST))
+	} else {
+		// Note: the statuses of disk on host are either exist or not exist
+		ret.Set("status", jsonutils.NewString(compute.DISK_EXIST))
+	}
+	hostutils.Response(ctx, w, ret)
+}
+
+func getSnapshotStatus(ctx context.Context, w http.ResponseWriter, r *http.Request) {
+	params, _, _ := appsrv.FetchEnv(ctx, w, r)
+	var (
+		storageId  = params["<storageId>"]
+		diskId     = params["<diskId>"]
+		snapshotId = params["<snapshotId>"]
+	)
+
+	storage := storageman.GetManager().GetStorage(storageId)
+	if storage == nil {
+		hostutils.Response(ctx, w, httperrors.NewNotFoundError("Storage %s not found", storageId))
+		return
+	}
+
+	var (
+		ret    = jsonutils.NewDict()
+		status string
+	)
+	if exist, err := storage.IsSnapshotExist(diskId, snapshotId); exist {
+		status = compute.SNAPSHOT_EXIST
+	} else if !exist && err == nil {
+		status = compute.SNAPSHOT_NOT_EXIST
+	} else {
+		log.Errorf("fetch snapshot exist failed %s", err)
+		status = compute.SNAPSHOT_UNKNOWN
+	}
+	ret.Set("status", jsonutils.NewString(status))
+	hostutils.Response(ctx, w, ret)
+}
+
 func saveToGlance(ctx context.Context, w http.ResponseWriter, r *http.Request) {
 	params, _, body := appsrv.FetchEnv(ctx, w, r)
 	var (

pkg/hostman/storageman/storage_base.go

@@ -89,6 +89,7 @@ type IStorage interface {
 	GetSnapshotDir() string
 	GetSnapshotPathByIds(diskId, snapshotId string) string
 	DeleteSnapshots(ctx context.Context, params interface{}) (jsonutils.JSONObject, error)
+	IsSnapshotExist(diskId, snapshotId string) (bool, error)
 
 	GetFreeSizeMb() int
 	GetCapacity() int

pkg/hostman/storageman/storage_local.go

@@ -77,6 +77,10 @@ func (s *SLocalStorage) GetSnapshotPathByIds(diskId, snapshotId string) string {
 	return path.Join(s.GetSnapshotDir(), diskId+options.HostOptions.SnapshotDirSuffix, snapshotId)
 }
 
+func (s *SLocalStorage) IsSnapshotExist(diskId, snapshotId string) (bool, error) {
+	return fileutils2.Exists(s.GetSnapshotPathByIds(diskId, snapshotId)), nil
+}
+
 func (s *SLocalStorage) GetComposedName() string {
 	return fmt.Sprintf("host_%s_%s_storage_%d", s.Manager.host.GetMasterIp(), s.StorageType(), s.Index)
 }

pkg/hostman/storageman/storage_rbd.go

@@ -89,6 +89,30 @@ func (s *SRbdStorage) GetSnapshotPathByIds(diskId, snapshotId string) string {
 	return ""
 }
 
+func (s *SRbdStorage) IsSnapshotExist(diskId, snapshotId string) (bool, error) {
+	var exist bool
+	pool, _ := s.StorageConf.GetString("pool")
+	_, err := s.withImage(pool, diskId,
+		func(src *rbd.Image) (interface{}, error) {
+			sps, err := src.GetSnapshotNames()
+			if err != nil {
+				return nil, errors.Wrap(err, "get snapshot names")
+			}
+			for i := 0; i < len(sps); i++ {
+				if sps[i].Name == snapshotId {
+					exist = true
+					break
+				}
+			}
+			return nil, nil
+		},
+	)
+	if err != nil {
+		return false, err
+	}
+	return exist, nil
+}
+
 func (s *SRbdStorage) GetSnapshotDir() string {
 	return ""
 }