diff --git a/snapshot/README.md b/snapshot/README.md
index 6d289d8e4e4..7cb4c1375ce 100644
--- a/snapshot/README.md
+++ b/snapshot/README.md
@@ -28,15 +28,15 @@ A Volume plugin must provide `RegisterPlugin()` to return plugin struct, `GetPlu
 import (
 	"k8s.io/client-go/pkg/api/v1"
 
-	crdv1 "github.com/rootfs/snapshot/pkg/apis/crd/v1"
-	"github.com/rootfs/snapshot/pkg/cloudprovider"
+	crdv1 "github.com/kubernetes-incubator/external-storage/snapshot/pkg/apis/crd/v1"
+	"github.com/kubernetes-incubator/external-storage/snapshot/pkg/cloudprovider"
 )
 
 type VolumePlugin interface {
 	// Init inits volume plugin
 	Init(cloudprovider.Interface)
 	// SnapshotCreate creates a VolumeSnapshot from a PersistentVolumeSpec
-	SnapshotCreate(*v1.PersistentVolume) (*crdv1.VolumeSnapshotDataSource, error)
+	SnapshotCreate(*v1.PersistentVolume, *map[string]string) (*crdv1.VolumeSnapshotDataSource, *[]crdv1.VolumeSnapshotCondition, error)
 	// SnapshotDelete deletes a VolumeSnapshot
 	// PersistentVolume is provided for volume types, if any, that need PV Spec to delete snapshot
 	SnapshotDelete(*crdv1.VolumeSnapshotDataSource, *v1.PersistentVolume) error
@@ -44,7 +44,9 @@ type VolumePlugin interface {
 	SnapshotRestore(*crdv1.VolumeSnapshotData, *v1.PersistentVolumeClaim, string, map[string]string) (*v1.PersistentVolumeSource, map[string]string, error)
 	// Describe volume snapshot status.
 	// return true if the snapshot is ready
-	DescribeSnapshot(snapshotData *crdv1.VolumeSnapshotData) (isCompleted bool, err error)
+	DescribeSnapshot(snapshotData *crdv1.VolumeSnapshotData) (snapConditions *[]crdv1.VolumeSnapshotCondition, isCompleted bool, err error)
+	// FindSnapshot finds a VolumeSnapshot by matching metadata
+	FindSnapshot(tags *map[string]string) (*crdv1.VolumeSnapshotDataSource, *[]crdv1.VolumeSnapshotCondition, error)
 	// VolumeDelete deletes a PV
 	// TODO in the future pass kubernetes client for certain volumes (e.g. rbd) so they can access storage class to retrieve secret
 	VolumeDelete(pv *v1.PersistentVolume) error
diff --git a/snapshot/pkg/apis/crd/v1/types.go b/snapshot/pkg/apis/crd/v1/types.go
index 7b739bc95cf..2baad44ad9f 100644
--- a/snapshot/pkg/apis/crd/v1/types.go
+++ b/snapshot/pkg/apis/crd/v1/types.go
@@ -36,7 +36,7 @@ type VolumeSnapshotStatus struct {
 	// +optional
 	CreationTimestamp metav1.Time `json:"creationTimestamp" protobuf:"bytes,1,opt,name=creationTimestamp"`
 
-	// Representes the lates available observations about the volume snapshot
+	// Representes the latest available observations about the volume snapshot
 	Conditions []VolumeSnapshotCondition `json:"conditions" protobuf:"bytes,2,rep,name=conditions"`
 }
 
@@ -44,8 +44,16 @@ type VolumeSnapshotConditionType string
 
 // These are valid conditions of a volume snapshot.
 const (
-	// VolumeSnapshotReady is added when the snapshot has been successfully created and is ready to be used.
+	// VolumeSnapshotConditionPending means the snapshot is cut and the application
+	// can resume accessing data if core_v1.ConditionStatus is True. It corresponds
+	// to "Uploading" in GCE PD or "Pending" in AWS and core_v1.ConditionStatus is True.
+	// It also corresponds to "Creating" in OpenStack Cinder and core_v1.ConditionStatus
+	// is Unknown.
+	VolumeSnapshotConditionPending VolumeSnapshotConditionType = "Pending"
+	// VolumeSnapshotConditionReady is added when the snapshot has been successfully created and is ready to be used.
 	VolumeSnapshotConditionReady VolumeSnapshotConditionType = "Ready"
+	// VolumeSnapshotConditionError means an error occurred during snapshot creation.
+	VolumeSnapshotConditionError VolumeSnapshotConditionType = "Error"
 )
 
 // VolumeSnapshot Condition describes the state of a volume snapshot  at a certain point.
diff --git a/snapshot/pkg/cloudprovider/providers/aws/aws.go b/snapshot/pkg/cloudprovider/providers/aws/aws.go
index 08155b66302..b05e5fa442c 100644
--- a/snapshot/pkg/cloudprovider/providers/aws/aws.go
+++ b/snapshot/pkg/cloudprovider/providers/aws/aws.go
@@ -313,6 +313,7 @@ type VolumeOptions struct {
 // VolumeOptions specifies volume snapshot options.
 type SnapshotOptions struct {
 	VolumeId string
+	Tags     *map[string]string
 }
 
 // Volumes is an interface for managing cloud-provisioned volumes
@@ -348,14 +349,17 @@ type Volumes interface {
 	DisksAreAttached(map[types.NodeName][]KubernetesVolumeID) (map[types.NodeName]map[KubernetesVolumeID]bool, error)
 
 	// Create an EBS volume snapshot
-	CreateSnapshot(snapshotOptions *SnapshotOptions) (snapshotId string, err error)
+	CreateSnapshot(snapshotOptions *SnapshotOptions) (snapshotId string, status string, err error)
 
 	// Delete an EBS volume snapshot
 	DeleteSnapshot(snapshotId string) (bool, error)
 
 	// Describe an EBS volume snapshot status for create or delete.
 	// return status (completed or pending or error), and error
-	DescribeSnapshot(snapshotId string) (isCompleted bool, err error)
+	DescribeSnapshot(snapshotId string) (status string, isCompleted bool, err error)
+
+	// Find snapshot by tags
+	FindSnapshot(tags map[string]string) ([]string, []string, error)
 }
 
 // InstanceGroups is an interface for managing cloud-managed instance groups / autoscaling instance groups
@@ -1904,7 +1908,7 @@ func (c *Cloud) DisksAreAttached(nodeDisks map[types.NodeName][]KubernetesVolume
 }
 
 // CreateSnapshot creates an EBS volume snapshot
-func (c *Cloud) CreateSnapshot(snapshotOptions *SnapshotOptions) (snapshotId string, err error) {
+func (c *Cloud) CreateSnapshot(snapshotOptions *SnapshotOptions) (snapshotId string, status string, err error) {
 	request := &ec2.CreateSnapshotInput{}
 	request.VolumeId = aws.String(snapshotOptions.VolumeId)
 	request.DryRun = aws.Bool(false)
@@ -1912,12 +1916,23 @@ func (c *Cloud) CreateSnapshot(snapshotOptions *SnapshotOptions) (snapshotId str
 	request.Description = aws.String(descriptions)
 	res, err := c.ec2.CreateSnapshot(request)
 	if err != nil {
-		return "", err
+		return "", "", err
 	}
 	if res == nil {
-		return "", fmt.Errorf("nil CreateSnapshotResponse")
+		return "", "", fmt.Errorf("nil CreateSnapshotResponse")
+	}
+	if snapshotOptions.Tags != nil {
+		awsID := awsVolumeID(aws.StringValue(res.SnapshotId))
+		// apply tags
+		if err := c.tagging.createTags(c.ec2, string(awsID), ResourceLifecycleOwned, *snapshotOptions.Tags); err != nil {
+			_, delerr := c.DeleteSnapshot(*res.SnapshotId)
+			if delerr != nil {
+				return "", "", fmt.Errorf("error tagging snapshot %s, could not delete the snapshot: %v", *res.SnapshotId, delerr)
+			}
+			return "", "", fmt.Errorf("error tagging snapshot %s: %v", *res.SnapshotId, err)
+		}
 	}
-	return *res.SnapshotId, nil
+	return *res.SnapshotId, *res.State, nil
 
 }
 
@@ -1934,7 +1949,7 @@ func (c *Cloud) DeleteSnapshot(snapshotId string) (bool, error) {
 }
 
 // DescribeSnapshot returns the status of the snapshot
-func (c *Cloud) DescribeSnapshot(snapshotId string) (isCompleted bool, err error) {
+func (c *Cloud) DescribeSnapshot(snapshotId string) (status string, isCompleted bool, err error) {
 	request := &ec2.DescribeSnapshotsInput{
 		SnapshotIds: []*string{
 			aws.String(snapshotId),
@@ -1942,21 +1957,50 @@ func (c *Cloud) DescribeSnapshot(snapshotId string) (isCompleted bool, err error
 	}
 	result, err := c.ec2.DescribeSnapshots(request)
 	if err != nil {
-		return false, err
+		return "", false, err
 	}
 	if len(result) != 1 {
-		return false, fmt.Errorf("wrong result from DescribeSnapshots: %#v", result)
+		return "", false, fmt.Errorf("wrong result from DescribeSnapshots: %#v", result)
 	}
 	if result[0].State == nil {
-		return false, fmt.Errorf("missing state from DescribeSnapshots: %#v", result)
+		return "", false, fmt.Errorf("missing state from DescribeSnapshots: %#v", result)
 	}
 	if *result[0].State == ec2.SnapshotStateCompleted {
-		return true, nil
+		return *result[0].State, true, nil
 	}
 	if *result[0].State == ec2.SnapshotStateError {
-		return false, fmt.Errorf("snapshot state is error: %s", *result[0].StateMessage)
+		return *result[0].State, false, fmt.Errorf("snapshot state is error: %s", *result[0].StateMessage)
+	}
+	if *result[0].State == ec2.SnapshotStatePending {
+		return *result[0].State, false, nil
+	}
+	return *result[0].State, false, fmt.Errorf("unknown state")
+}
+
+// FindSnapshot returns the found snapshot
+func (c *Cloud) FindSnapshot(tags map[string]string) ([]string, []string, error) {
+	request := &ec2.DescribeSnapshotsInput{}
+	for k, v := range tags {
+		filter := &ec2.Filter{}
+		filter.SetName(k)
+		filter.SetValues([]*string{&v})
+
+		request.Filters = append(request.Filters, filter)
+	}
+
+	result, err := c.ec2.DescribeSnapshots(request)
+	if err != nil {
+		return nil, nil, err
+	}
+	var snapshotIDs, statuses []string
+	for _, snapshot := range result {
+		id := *snapshot.SnapshotId
+		status := *snapshot.State
+		glog.Infof("found %s, status %s", id, status)
+		snapshotIDs = append(snapshotIDs, id)
+		statuses = append(statuses, status)
 	}
-	return false, fmt.Errorf(*result[0].StateMessage)
+	return snapshotIDs, statuses, nil
 }
 
 // Gets the current load balancer state
diff --git a/snapshot/pkg/cloudprovider/providers/gce/gce.go b/snapshot/pkg/cloudprovider/providers/gce/gce.go
index 0cd6de64011..09da81e12f0 100644
--- a/snapshot/pkg/cloudprovider/providers/gce/gce.go
+++ b/snapshot/pkg/cloudprovider/providers/gce/gce.go
@@ -152,7 +152,10 @@ type Disks interface {
 
 	// Describe a GCE PD volume snapshot status for create or delete.
 	// return status (completed or pending or error), and error
-	DescribeSnapshot(snapshotToGet string) (isCompleted bool, err error)
+	DescribeSnapshot(snapshotToGet string) (status string, isCompleted bool, err error)
+
+	// Find snapshot by tags
+	FindSnapshot(tags map[string]string) ([]string, []string, error)
 
 	// GetAutoLabelsForPD returns labels to apply to PersistentVolume
 	// representing this PD, namely failure domain and zone.
@@ -2843,19 +2846,25 @@ func (gce *GCECloud) CreateDiskFromSnapshot(snapshot string,
 	return err
 }
 
-func (gce *GCECloud) DescribeSnapshot(snapshotToGet string) (isCompleted bool, err error) {
+func (gce *GCECloud) DescribeSnapshot(snapshotToGet string) (status string, isCompleted bool, err error) {
 	snapshot, err := gce.getSnapshotByName(snapshotToGet)
 	if err != nil {
-		return false, err
+		return "", false, err
 	}
 	//no snapshot is found
 	if snapshot == nil {
-		return false, fmt.Errorf("snapshot %s is found", snapshotToGet)
+		return "", false, fmt.Errorf("snapshot %s is found", snapshotToGet)
 	}
 	if snapshot.Status == "READY" {
-		return true, nil
+		return snapshot.Status, true, nil
 	}
-	return false, nil
+	return snapshot.Status, false, nil
+}
+
+// FindSnapshot returns the found snapshots
+func (gce *GCECloud) FindSnapshot(tags map[string]string) ([]string, []string, error) {
+	var snapshotIDs, statuses []string
+	return snapshotIDs, statuses, nil
 }
 
 func (gce *GCECloud) DeleteSnapshot(snapshotToDelete string) error {
@@ -2892,6 +2901,7 @@ func (gce *GCECloud) getSnapshotByName(snapshotName string) (*gceSnapshot, error
 	return nil, nil
 }
 
+// TODO: CreateSnapshot should return snapshot status
 func (gce *GCECloud) CreateSnapshot(diskName string, zone string, snapshotName string, tags map[string]string) error {
 	isManaged := false
 	for _, managedZone := range gce.managedZones {
diff --git a/snapshot/pkg/cloudprovider/providers/openstack/openstack_snapshots.go b/snapshot/pkg/cloudprovider/providers/openstack/openstack_snapshots.go
index 3aed5dcf1bf..d1a90b5cf14 100644
--- a/snapshot/pkg/cloudprovider/providers/openstack/openstack_snapshots.go
+++ b/snapshot/pkg/cloudprovider/providers/openstack/openstack_snapshots.go
@@ -23,6 +23,7 @@ import (
 	snapshotsV2 "github.com/gophercloud/gophercloud/openstack/blockstorage/v2/snapshots"
 
 	"github.com/golang/glog"
+	ctrlsnap "github.com/kubernetes-incubator/external-storage/snapshot/pkg/controller/snapshotter"
 )
 
 // SnapshotsV2 is the Cinder V2 Snapshot service from gophercloud
@@ -37,6 +38,8 @@ type Snapshot struct {
 	Name           string
 	Status         string
 	SourceVolumeID string
+	Description    string
+	Metadata       map[string]string
 }
 
 // SnapshotCreateOpts are the valid create options for Cinder Snapshots
@@ -48,13 +51,21 @@ type SnapshotCreateOpts struct {
 	Metadata    map[string]string
 }
 
+// SnapshotListOpts are the valid list options for Cinder Snapshots
+type SnapshotListOpts struct {
+	Name     string
+	Status   string
+	VolumeID string
+}
+
 type snapshotService interface {
-	createSnapshot(opts SnapshotCreateOpts) (string, error)
+	createSnapshot(opts SnapshotCreateOpts) (string, string, error)
 	deleteSnapshot(snapshotName string) error
 	getSnapshot(snapshotID string) (Snapshot, error)
+	listSnapshots(opts SnapshotListOpts) ([]Snapshot, error)
 }
 
-func (snapshots *SnapshotsV2) createSnapshot(opts SnapshotCreateOpts) (string, error) {
+func (snapshots *SnapshotsV2) createSnapshot(opts SnapshotCreateOpts) (string, string, error) {
 
 	createOpts := snapshotsV2.CreateOpts{
 		VolumeID:    opts.VolumeID,
@@ -66,9 +77,9 @@ func (snapshots *SnapshotsV2) createSnapshot(opts SnapshotCreateOpts) (string, e
 
 	snap, err := snapshotsV2.Create(snapshots.blockstorage, createOpts).Extract()
 	if err != nil {
-		return "", err
+		return "", "", err
 	}
-	return snap.ID, nil
+	return snap.ID, snap.Status, nil
 }
 
 func (snapshots *SnapshotsV2) getSnapshot(snapshotID string) (Snapshot, error) {
@@ -96,12 +107,45 @@ func (snapshots *SnapshotsV2) deleteSnapshot(snapshotID string) error {
 	return err
 }
 
+func (snapshots *SnapshotsV2) listSnapshots(opts SnapshotListOpts) ([]Snapshot, error) {
+	var snaplist []Snapshot
+
+	listOpts := snapshotsV2.ListOpts{
+		Name:     opts.Name,
+		Status:   opts.Status,
+		VolumeID: opts.VolumeID,
+	}
+
+	snapPages, err := snapshotsV2.List(snapshots.blockstorage, listOpts).AllPages()
+	if err != nil {
+		return snaplist, err
+	}
+	allSnaps, err := snapshotsV2.ExtractSnapshots(snapPages)
+	if err != nil {
+		return snaplist, err
+	}
+
+	for _, snapshot := range allSnaps {
+		glog.Infof("Snapshot details: %#v", snapshot)
+		var snap Snapshot
+		snap.ID = snapshot.ID
+		snap.Name = snapshot.Name
+		snap.Status = snapshot.Status
+		snap.SourceVolumeID = snapshot.VolumeID
+		snap.Description = snapshot.Description
+		snap.Metadata = snapshot.Metadata
+		snaplist = append(snaplist, snap)
+	}
+
+	return snaplist, nil
+}
+
 // CreateSnapshot from the specified volume
-func (os *OpenStack) CreateSnapshot(sourceVolumeID, name, description string, tags map[string]string) (string, error) {
+func (os *OpenStack) CreateSnapshot(sourceVolumeID, name, description string, tags map[string]string) (string, string, error) {
 	snapshots, err := os.snapshotService()
 	if err != nil || snapshots == nil {
 		glog.Errorf("Unable to initialize cinder client for region: %s", os.region)
-		return "", err
+		return "", "", fmt.Errorf("Failed to create snapshot for volume %s: %v", sourceVolumeID, err)
 	}
 
 	opts := SnapshotCreateOpts{
@@ -113,15 +157,15 @@ func (os *OpenStack) CreateSnapshot(sourceVolumeID, name, description string, ta
 		opts.Metadata = tags
 	}
 
-	snapshotID, err := snapshots.createSnapshot(opts)
+	snapshotID, status, err := snapshots.createSnapshot(opts)
 
 	if err != nil {
 		glog.Errorf("Failed to snapshot volume %s : %v", sourceVolumeID, err)
-		return "", err
+		return "", "", err
 	}
 
 	glog.Infof("Created snapshot %v from volume: %v", snapshotID, sourceVolumeID)
-	return snapshotID, nil
+	return snapshotID, status, nil
 }
 
 // DeleteSnapshot deletes the specified snapshot
@@ -141,11 +185,11 @@ func (os *OpenStack) DeleteSnapshot(snapshotID string) error {
 
 // FIXME(j-griffith): Name doesn't fit at all here, this is actually more like is `IsAvailable`
 // DescribeSnapshot returns the status of the snapshot
-func (os *OpenStack) DescribeSnapshot(snapshotID string) (isCompleted bool, err error) {
+func (os *OpenStack) DescribeSnapshot(snapshotID string) (status string, isCompleted bool, err error) {
 	ss, err := os.snapshotService()
 	if err != nil || ss == nil {
 		glog.Errorf("Unable to initialize cinder client for region: %s", os.region)
-		return false, err
+		return "", false, fmt.Errorf("Failed to describe snapshot %s: %v", snapshotID, err)
 	}
 
 	snap, err := ss.getSnapshot(snapshotID)
@@ -154,10 +198,56 @@ func (os *OpenStack) DescribeSnapshot(snapshotID string) (isCompleted bool, err
 	}
 
 	if err != nil {
-		return false, err
+		return "", false, err
 	}
 	if snap.Status != "available" {
-		return false, fmt.Errorf("current snapshot status is: %s", snap.Status)
+		return snap.Status, false, nil
+	}
+	return snap.Status, true, nil
+}
+
+// Find snapshot by metadata
+func (os *OpenStack) FindSnapshot(tags map[string]string) ([]string, []string, error) {
+	var snapshotIDs, statuses []string
+	ss, err := os.snapshotService()
+	if err != nil || ss == nil {
+		glog.Errorf("Unable to initialize cinder client for region: %s", os.region)
+		return snapshotIDs, statuses, fmt.Errorf("Failed to find snapshot by tags %v: %v", tags, err)
 	}
-	return true, nil
+
+	opts := SnapshotListOpts{}
+	snapshots, err := ss.listSnapshots(opts)
+
+	if err != nil {
+		glog.Errorf("Failed to list snapshots. Error: %v", err)
+		return snapshotIDs, statuses, err
+	}
+	glog.Infof("Listed [%v] snapshots.", len(snapshots))
+
+	glog.Infof("Looking for matching tags [%#v] in snapshots.", tags)
+	// Loop around to find the snapshot with the matching input metadata
+	for _, snapshot := range snapshots {
+		glog.Infof("Looking for matching tags in snapshot [%#v].", snapshot)
+		namespaceVal, ok := snapshot.Metadata[ctrlsnap.CloudSnapshotCreatedForVolumeSnapshotNamespaceTag]
+		if ok {
+			if tags[ctrlsnap.CloudSnapshotCreatedForVolumeSnapshotNamespaceTag] == namespaceVal {
+				nameVal, ok := snapshot.Metadata[ctrlsnap.CloudSnapshotCreatedForVolumeSnapshotNameTag]
+				if ok {
+					if tags[ctrlsnap.CloudSnapshotCreatedForVolumeSnapshotNameTag] == nameVal {
+						timeVal, ok := snapshot.Metadata[ctrlsnap.CloudSnapshotCreatedForVolumeSnapshotTimestampTag]
+						if ok {
+							if tags[ctrlsnap.CloudSnapshotCreatedForVolumeSnapshotTimestampTag] == timeVal {
+								snapshotIDs = append(snapshotIDs, snapshot.ID)
+								statuses = append(statuses, snapshot.Status)
+								glog.Infof("Add snapshot [%#v].", snapshot)
+							}
+						}
+
+					}
+				}
+			}
+		}
+	}
+
+	return snapshotIDs, statuses, nil
 }
diff --git a/snapshot/pkg/controller/reconciler/reconciler.go b/snapshot/pkg/controller/reconciler/reconciler.go
index fea1472068d..0b726a0fb2a 100644
--- a/snapshot/pkg/controller/reconciler/reconciler.go
+++ b/snapshot/pkg/controller/reconciler/reconciler.go
@@ -23,8 +23,11 @@ import (
 	"time"
 
 	"github.com/golang/glog"
+	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/wait"
 
+	crdv1 "github.com/kubernetes-incubator/external-storage/snapshot/pkg/apis/crd/v1"
 	"github.com/kubernetes-incubator/external-storage/snapshot/pkg/controller/cache"
 	"github.com/kubernetes-incubator/external-storage/snapshot/pkg/controller/snapshotter"
 )
@@ -132,7 +135,15 @@ func (rc *reconciler) reconcile() {
 				// Seems the write to API server failed before. Find the SnapshotData and fix
 				// this VolumeSnapshot reference and status
 				glog.Infof("Volume snapshot %s is missing the snapshot data name - updating.", name)
-				err := rc.snapshotter.UpdateVolumeSnapshot(name)
+				snapConditions := []crdv1.VolumeSnapshotCondition{
+					{
+						Type:               crdv1.VolumeSnapshotConditionReady,
+						Status:             v1.ConditionTrue,
+						Message:            "Snapshot created succsessfully",
+						LastTransitionTime: metav1.Now(),
+					},
+				}
+				_, err := rc.snapshotter.UpdateVolumeSnapshot(name, &snapConditions)
 				if err != nil {
 					glog.Errorf("Error updating VolumeSnapshot %s: %v", name, err)
 				}
diff --git a/snapshot/pkg/controller/snapshotter/snapshotter.go b/snapshot/pkg/controller/snapshotter/snapshotter.go
index 34edd9e8c36..7d4aca8959d 100644
--- a/snapshot/pkg/controller/snapshotter/snapshotter.go
+++ b/snapshot/pkg/controller/snapshotter/snapshotter.go
@@ -48,7 +48,7 @@ type VolumeSnapshotter interface {
 	CreateVolumeSnapshot(snapshot *crdv1.VolumeSnapshot)
 	DeleteVolumeSnapshot(snapshot *crdv1.VolumeSnapshot)
 	PromoteVolumeSnapshotToPV(snapshot *crdv1.VolumeSnapshot)
-	UpdateVolumeSnapshot(snapshotName string) error
+	UpdateVolumeSnapshot(snapshotName string, status *[]crdv1.VolumeSnapshotCondition) (*crdv1.VolumeSnapshot, error)
 	UpdateVolumeSnapshotData(snapshotDataName string, status *[]crdv1.VolumeSnapshotDataCondition) error
 }
 
@@ -69,6 +69,20 @@ const (
 	createVolumeSnapshotDataRetryCount = 5
 	// Interval between retries when we create a VolumeSnapshotData object.
 	createVolumeSnapshotDataInterval = 10 * time.Second
+	// CloudSnapshotCreatedForVolumeSnapshotNamespaceTag is a name of a tag attached to a real snapshot in cloud
+	// (e.g. AWS EBS or GCE PD) with namespace of a volumesnapshot used to create this snapshot.
+	CloudSnapshotCreatedForVolumeSnapshotNamespaceTag = "kubernetes.io/created-for/snapshot/namespace"
+	// CloudSnapshotCreatedForVolumeSnapshotNameTag is a name of a tag attached to a real snapshot in cloud
+	// (e.g. AWS EBS or GCE PD) with name of a volumesnapshot used to create this snapshot.
+	CloudSnapshotCreatedForVolumeSnapshotNameTag = "kubernetes.io/created-for/snapshot/name"
+	// CloudSnapshotCreatedForVolumeSnapshotTimestampTag is a name of a tag attached to a real snapshot in cloud
+	// (e.g. AWS EBS or GCE PD) with timestamp when the create snapshot request is issued.
+	CloudSnapshotCreatedForVolumeSnapshotTimestampTag = "kubernetes.io/created-for/snapshot/timestamp"
+	// Statuses of snapshot creation process
+	statusReady   string = "ready"
+	statusError   string = "error"
+	statusPending string = "pending"
+	statusNew     string = "new"
 )
 
 func NewVolumeSnapshotter(
@@ -87,6 +101,7 @@ func NewVolumeSnapshotter(
 	}
 }
 
+// TODO(xyang): Cache PV volume information into meta data to avoid query api server
 // Helper function to get PV from VolumeSnapshot
 func (vs *volumeSnapshotter) getPVFromVolumeSnapshot(snapshotName string, snapshot *crdv1.VolumeSnapshot) (*v1.PersistentVolume, error) {
 	pvcName := snapshot.Spec.PersistentVolumeClaimName
@@ -128,13 +143,12 @@ func (vs *volumeSnapshotter) getSnapshotDataFromSnapshotName(snapshotName string
 		return nil
 	}
 	if len(snapshotDataList.Items) == 0 {
-		glog.Errorf("Error: no VolumeSnapshotData objects found on the API server")
+		glog.Infof("No VolumeSnapshotData objects found on the API server")
 		return nil
 	}
 	for _, snapData := range snapshotDataList.Items {
 		if snapData.Spec.VolumeSnapshotRef != nil {
 			name := snapData.Spec.VolumeSnapshotRef.Namespace + "/" + snapData.Spec.VolumeSnapshotRef.Name
-			// if snapData.Spec.VolumeSnapshotRef.Name == snapshotName
 			if name == snapshotName || snapData.Spec.VolumeSnapshotRef.Name == snapshotName {
 				snapshotDataObj = snapData
 				found = true
@@ -150,6 +164,95 @@ func (vs *volumeSnapshotter) getSnapshotDataFromSnapshotName(snapshotName string
 	return &snapshotDataObj
 }
 
+// Query status of the snapshot from plugin and update the status of VolumeSnapshot and VolumeSnapshotData
+// if needed. Finish waiting when the snapshot becomes available/ready or error.
+func (vs *volumeSnapshotter) waitForSnapshot(snapshotName string, snapshot *crdv1.VolumeSnapshot, snapshotDataObj *crdv1.VolumeSnapshotData) error {
+	// Get a fresh VolumeSnapshot from API
+	var snapshotObj crdv1.VolumeSnapshot
+	err := vs.restClient.Get().
+		Name(snapshot.Metadata.Name).
+		Resource(crdv1.VolumeSnapshotResourcePlural).
+		Namespace(snapshot.Metadata.Namespace).
+		Do().Into(&snapshotObj)
+	if err != nil {
+		return fmt.Errorf("Error getting snapshot object %s from API server: %v", snapshotName, err)
+	}
+	snapshotDataName := snapshotObj.Spec.SnapshotDataName
+	glog.Infof("In waitForSnapshot: snapshot %s snapshot data %s", snapshotName, snapshotDataName)
+	if snapshotDataObj == nil {
+		return fmt.Errorf("Failed to update VolumeSnapshot for snapshot %s: no VolumeSnapshotData.", snapshotName)
+	}
+
+	plugin, err := vs.getPlugin(snapshotName, snapshot)
+	if plugin == nil {
+		return fmt.Errorf("Failed to get volume plugin to wait for snapshot creation %s. %v", snapshotName, err)
+	}
+
+	var newSnapshot *crdv1.VolumeSnapshot = nil
+	var conditions *[]crdv1.VolumeSnapshotCondition = nil
+	var status, newstatus string = "", ""
+	// Wait until the snapshot is successfully created by the plugin or an error occurs that
+	// fails the snapshot creation.
+	for {
+		conditions, _, err = (*plugin).DescribeSnapshot(snapshotDataObj)
+		if err != nil {
+			glog.Warningf("failed to get snapshot %v, err: %v", snapshotName, err)
+			continue
+		}
+		newstatus = vs.getSimplifiedSnapshotStatus(conditions)
+		if newstatus == statusPending {
+			if newstatus != status {
+				status = newstatus
+				glog.V(5).Infof("Snapshot %s creation is not complete yet. Status: [%#v] Retrying...", snapshotName, conditions)
+				// UpdateVolmeSnapshot status
+				newSnapshot, err = vs.UpdateVolumeSnapshot(snapshotName, conditions)
+				if err != nil {
+					glog.Errorf("Error updating volume snapshot %s: %v", snapshotName, err)
+					return fmt.Errorf("Failed to update VolumeSnapshot for snapshot %s: %v", snapshotName, err)
+				}
+			}
+			time.Sleep(createVolumeSnapshotDataInterval)
+			continue
+		} else if newstatus == statusError {
+			return fmt.Errorf("Status for snapshot %s is error.", snapshotName)
+		}
+		glog.Infof("waitForSnapshot: Snapshot %s creation is complete: %#v", snapshotName, conditions)
+
+		newSnapshot, err = vs.UpdateVolumeSnapshot(snapshotName, conditions)
+		if err != nil {
+			glog.Errorf("Error updating volume snapshot %s: %v", snapshotName, err)
+			return fmt.Errorf("Failed to update VolumeSnapshot for snapshot %s: %v", snapshotName, err)
+		}
+
+		ind := len(*conditions) - 1
+		snapDataConditions := []crdv1.VolumeSnapshotDataCondition{
+			{
+				Type:               (crdv1.VolumeSnapshotDataConditionType)((*conditions)[ind].Type),
+				Status:             (*conditions)[ind].Status,
+				Message:            (*conditions)[ind].Message,
+				LastTransitionTime: metav1.Now(),
+			},
+		}
+
+		// Update VolumeSnapshotData status
+		err = vs.UpdateVolumeSnapshotData(snapshotDataName, &snapDataConditions)
+		if err != nil {
+			return fmt.Errorf("Error update snapshotData object %s: %v", snapshotName, err)
+		}
+
+		if newstatus == statusReady {
+			glog.Infof("waitForSnapshot: Snapshot %s created successfully. Adding it to Actual State of World.", snapshotName)
+			vs.actualStateOfWorld.AddSnapshot(newSnapshot)
+			// Break out of the for loop
+			return nil
+		} else if newstatus == statusError {
+			return fmt.Errorf("Failed to create snapshot %s.", snapshotName)
+		}
+	}
+
+	return fmt.Errorf("Snapshot %s is NOT completed successfully.", snapshotName)
+}
+
 func (vs *volumeSnapshotter) updateSnapshotDataStatus(snapshotName string, snapshot *crdv1.VolumeSnapshot) error {
 	var snapshotDataObj crdv1.VolumeSnapshotData
 	snapshotDataName := snapshot.Spec.SnapshotDataName
@@ -178,7 +281,7 @@ func (vs *volumeSnapshotter) updateSnapshotDataStatus(snapshotName string, snaps
 		if !ok {
 			return fmt.Errorf("%s is not supported volume for %#v", volumeType, spec)
 		}
-		completed, err := plugin.DescribeSnapshot(&snapshotDataObj)
+		_, completed, err := plugin.DescribeSnapshot(&snapshotDataObj)
 		if !completed {
 			return fmt.Errorf("snapshot is not completed yet: %v", err)
 		}
@@ -192,7 +295,7 @@ func (vs *volumeSnapshotter) updateSnapshotDataStatus(snapshotName string, snaps
 			},
 		}
 
-		// Bind VolumeSnapshot to VolumeSnapshotData
+		// Update VolumeSnapshotData status
 		err = vs.UpdateVolumeSnapshotData(snapshotDataName, &status)
 		if err != nil {
 			return fmt.Errorf("Error update snapshotData object %s: %v", snapshotName, err)
@@ -205,24 +308,25 @@ func (vs *volumeSnapshotter) updateSnapshotDataStatus(snapshotName string, snaps
 
 // This is the function responsible for determining the correct volume plugin to use,
 // asking it to make a snapshot and assigning it some name that it returns to the caller.
-func (vs *volumeSnapshotter) takeSnapshot(pv *v1.PersistentVolume) (*crdv1.VolumeSnapshotDataSource, error) {
+func (vs *volumeSnapshotter) takeSnapshot(pv *v1.PersistentVolume, tags *map[string]string) (*crdv1.VolumeSnapshotDataSource, *[]crdv1.VolumeSnapshotCondition, error) {
 	spec := &pv.Spec
 	volumeType := crdv1.GetSupportedVolumeFromPVSpec(spec)
 	if len(volumeType) == 0 {
-		return nil, fmt.Errorf("unsupported volume type found in PV %#v", spec)
+		return nil, nil, fmt.Errorf("unsupported volume type found in PV %#v", spec)
 	}
 	plugin, ok := (*vs.volumePlugins)[volumeType]
 	if !ok {
-		return nil, fmt.Errorf("%s is not supported volume for %#v", volumeType, spec)
+		return nil, nil, fmt.Errorf("%s is not supported volume for %#v", volumeType, spec)
 	}
-	snap, err := plugin.SnapshotCreate(pv)
+
+	snap, snapConditions, err := plugin.SnapshotCreate(pv, tags)
 	if err != nil {
 		glog.Warningf("failed to snapshot %#v, err: %v", spec, err)
 	} else {
-		glog.Infof("snapshot created: %v", snap)
-		return snap, nil
+		glog.Infof("snapshot created: %v. Conditions: %#v", snap, snapConditions)
+		return snap, snapConditions, nil
 	}
-	return nil, nil
+	return nil, nil, nil
 }
 
 // This is the function responsible for determining the correct volume plugin to use,
@@ -245,98 +349,303 @@ func (vs *volumeSnapshotter) deleteSnapshot(spec *v1.PersistentVolumeSpec, sourc
 	return nil
 }
 
-// Below are the closures meant to build the functions for the GoRoutineMap operations.
+func (vs *volumeSnapshotter) getSimplifiedSnapshotStatus(conditions *[]crdv1.VolumeSnapshotCondition) string {
+	if conditions == nil {
+		glog.Errorf("Invalid input conditions for snapshot.")
+		return statusError
+	}
+	index := len(*conditions) - 1
+	if len(*conditions) > 0 &&
+		((*conditions)[index].Type == crdv1.VolumeSnapshotConditionReady &&
+			(*conditions)[index].Status == v1.ConditionTrue) {
+		return statusReady
+	} else if len(*conditions) > 0 &&
+		(*conditions)[index].Type == crdv1.VolumeSnapshotConditionError {
+		return statusError
+	} else if len(*conditions) > 0 &&
+		(*conditions)[index].Type == crdv1.VolumeSnapshotConditionPending &&
+		((*conditions)[index].Status == v1.ConditionTrue ||
+			(*conditions)[index].Status == v1.ConditionUnknown) {
+		return statusPending
+	} else {
+		return statusNew
+	}
+}
 
-func (vs *volumeSnapshotter) getSnapshotCreateFunc(snapshotName string, snapshot *crdv1.VolumeSnapshot) func() error {
-	// Create a snapshot:
-	// 1. If Snapshot references SnapshotData object, try to find it
-	//   1a. If doesn't exist, log error and finish, if it exists already, check its SnapshotRef
-	//   1b. If it's empty, check its Spec UID (or fins out what PV/PVC does and copies the mechanism)
-	//   1c. If it matches the user (TODO: how to find out?), bind the two objects and finish
-	//   1d. If it doesn't match, log error and finish.
-	// 2. Create the SnapshotData object
-	// 3. Ask the backend to create the snapshot (device)
-	// 4. If OK, update the SnapshotData and Snapshot objects
-	// 5. Add the Snapshot to the ActualStateOfWorld
-	// 6. Finish (we have created snapshot for an user)
-	return func() error {
-		glog.Infof("Enter getSnapshotCreateFunc: snapshotName %s snapshot [%#v]", snapshotName, snapshot)
-		if snapshot.Spec.SnapshotDataName != "" {
-			// update snapshot data status and if complete, adds to asw
-			return vs.updateSnapshotDataStatus(snapshotName, snapshot)
+func (vs *volumeSnapshotter) findVolumeSnapshotMetadata(snapshot *crdv1.VolumeSnapshot) *map[string]string {
+	var tags *map[string]string = nil
+	if snapshot.Metadata.Name != "" && snapshot.Metadata.Namespace != "" {
+		if snapshot.Metadata.Labels != nil {
+			timestamp, ok := snapshot.Metadata.Labels["Timestamp"]
+			if ok {
+				tags := make(map[string]string)
+				tags[CloudSnapshotCreatedForVolumeSnapshotNamespaceTag] = snapshot.Metadata.Namespace
+				tags[CloudSnapshotCreatedForVolumeSnapshotNameTag] = snapshot.Metadata.Name
+				tags[CloudSnapshotCreatedForVolumeSnapshotTimestampTag] = timestamp
+				glog.Infof("findVolumeSnapshotMetadata: returning tags [%#v]", tags)
+			}
 		}
-		pv, err := vs.getPVFromVolumeSnapshot(snapshotName, snapshot)
+	}
+	return tags
+}
+
+func (vs *volumeSnapshotter) getPlugin(snapshotName string, snapshot *crdv1.VolumeSnapshot) (*volume.VolumePlugin, error) {
+	pv, err := vs.getPVFromVolumeSnapshot(snapshotName, snapshot)
+	if err != nil {
+		return nil, err
+	}
+	spec := &pv.Spec
+	volumeType := crdv1.GetSupportedVolumeFromPVSpec(spec)
+	if len(volumeType) == 0 {
+		return nil, fmt.Errorf("Unsupported volume type found in PV %#v", spec)
+	}
+	plugin, ok := (*vs.volumePlugins)[volumeType]
+	if !ok {
+		return nil, fmt.Errorf("%s is not supported volume for %#v", volumeType, spec)
+	}
+	return &plugin, nil
+}
+
+func (vs *volumeSnapshotter) getSnapshotStatus(snapshot *crdv1.VolumeSnapshot) (string, *crdv1.VolumeSnapshotData, error) {
+	var bCreateSnapData bool = false
+	var snapshotName string = snapshot.Metadata.Name
+	var snapshotDataObj *crdv1.VolumeSnapshotData = nil
+	var snapshotDataSource *crdv1.VolumeSnapshotDataSource = nil
+	var conditions *[]crdv1.VolumeSnapshotCondition = nil
+	var err error = nil
+	status := vs.getSimplifiedSnapshotStatus(&snapshot.Status.Conditions)
+	if status == statusReady {
+		return status, nil, nil
+	} else if status == statusPending {
+		// If we are here, takeSnapshot has already happened.
+		// Check whether the VolumeSnapshotData object is already created
+		snapshotDataObj = vs.getSnapshotDataFromSnapshotName(snapshotName)
+		if snapshotDataObj == nil {
+			bCreateSnapData = true
+			// Find snapshot by existing tags, and create VolumeSnapshotData
+		} else {
+			// Bind VolumeSnapshotData to VolumeSnapshot if it has not happened yet
+			if snapshot.Spec.SnapshotDataName != snapshotDataObj.Metadata.Name {
+				glog.Infof("getSnapshotStatus: bind VolumeSnapshotData to VolumeSnapshot %s.", snapshotName)
+				err = vs.bindVolumeSnapshotDataToVolumeSnapshot(snapshotName, snapshotDataObj.Metadata.Name)
+				if err != nil {
+					glog.Errorf("getSnapshotStatus: Error updating volume snapshot %s: %v", snapshotName, err)
+					return statusError, nil, err
+				}
+			}
+			return status, snapshotDataObj, nil
+		}
+	} else if status == statusError {
+		return status, nil, fmt.Errorf("Failed to find snapshot %s", snapshotName)
+	} else {
+		bCreateSnapData = true
+		// Find snapshot by existing tags, and create VolumeSnapshotData
+	}
+
+	if bCreateSnapData {
+		snapshotDataSource, conditions, err = vs.findSnapshot(snapshotName, snapshot)
 		if err != nil {
-			return err
+			return statusNew, nil, nil
 		}
-		pvName := pv.Name
-		snapshotDataSource, err := vs.takeSnapshot(pv)
-		if err != nil || snapshotDataSource == nil {
-			return fmt.Errorf("Failed to take snapshot of the volume %s: %q", pvName, err)
+		glog.Infof("getSnapshotStatus: create VolumeSnapshotData object for VolumeSnapshot %s.", snapshotName)
+		snapshotDataObj, err := vs.createVolumeSnapshotData(snapshotName, snapshot, snapshotDataSource)
+		if err != nil {
+			return statusError, nil, err
 		}
-		// Snapshot has been created, made an object for it
-		readyCondition := crdv1.VolumeSnapshotDataCondition{
-			Type:    crdv1.VolumeSnapshotDataConditionPending,
-			Status:  v1.ConditionTrue,
-			Message: "Snapshot data is being created",
+		if status != statusReady {
+			_, err = vs.UpdateVolumeSnapshot(snapshotName, conditions)
+			if err != nil {
+				glog.Errorf("getSnapshotStatus: Error updating volume snapshot %s: %v", snapshotName, err)
+				return statusError, nil, err
+			}
 		}
-		snapName := "k8s-volume-snapshot-" + string(uuid.NewUUID())
-
-		snapshotData := &crdv1.VolumeSnapshotData{
-			Metadata: metav1.ObjectMeta{
-				Name: snapName,
-			},
-			Spec: crdv1.VolumeSnapshotDataSpec{
-				VolumeSnapshotRef: &v1.ObjectReference{
-					Kind: "VolumeSnapshot",
-					Name: snapshotName,
-				},
-				PersistentVolumeRef: &v1.ObjectReference{
-					Kind: "PersistentVolume",
-					Name: pvName,
-				},
-				VolumeSnapshotDataSource: *snapshotDataSource,
-			},
-			Status: crdv1.VolumeSnapshotDataStatus{
-				Conditions: []crdv1.VolumeSnapshotDataCondition{
-					readyCondition,
-				},
-			},
+		glog.Infof("getSnapshotStatus: bind VolumeSnapshotData to VolumeSnapshot %s.", snapshotName)
+		err = vs.bindVolumeSnapshotDataToVolumeSnapshot(snapshotName, snapshotDataObj.Metadata.Name)
+		if err != nil {
+			glog.Errorf("getSnapshotStatus: Error binding VolumeSnapshotData to VolumeSnapshot %s: %v", snapshotName, err)
+			return statusError, nil, err
 		}
-		var result crdv1.VolumeSnapshotData
-		for i := 0; i < createVolumeSnapshotDataRetryCount; i++ {
-			err = vs.restClient.Post().
-				Resource(crdv1.VolumeSnapshotDataResourcePlural).
-				Namespace(v1.NamespaceDefault).
-				Body(snapshotData).
-				Do().Into(&result)
+		return statusPending, snapshotDataObj, nil
+	}
+	return statusError, nil, nil
+}
+
+// Below are the closures meant to build the functions for the GoRoutineMap operations.
+// syncSnapshot is the main controller method to decide what to do to create a snapshot.
+func (vs *volumeSnapshotter) syncSnapshot(snapshotName string, snapshot *crdv1.VolumeSnapshot) func() error {
+	return func() error {
+		status, snapshotDataObj, err := vs.getSnapshotStatus(snapshot)
+		switch status {
+		case statusReady:
+			return nil
+		case statusError:
+			glog.Infof("syncSnapshot: Error creating snapshot %s.", snapshotName)
+			return fmt.Errorf("Error creating snapshot %s.", snapshotName)
+		case statusPending:
+			glog.Infof("syncSnapshot: Snapshot %s is Pending.", snapshotName)
+			// Query the volume plugin for the status of the snapshot with snapshot id
+			// from VolumeSnapshotData object.
+			// Add snapshot to Actual State of World when snapshot is Ready.
+			err = vs.waitForSnapshot(snapshotName, snapshot, snapshotDataObj)
 			if err != nil {
-				// Re-Try it as errors writing to the API server are common
-				glog.Infof("Error creating the VolumeSnapshotData %s: %v. Retrying...", snapshotName, err)
-				time.Sleep(createVolumeSnapshotDataInterval)
-			} else {
-				break
+				return fmt.Errorf("Failed to create snapshot %s.", snapshotName)
 			}
+			glog.Infof("syncSnapshot: Snapshot %s created successfully.", snapshotName)
+			return nil
+		case statusNew:
+			glog.Infof("syncSnapshot: Creating snapshot %s ...", snapshotName)
+			ret := vs.createSnapshot(snapshotName, snapshot)
+			return ret
 		}
+		return fmt.Errorf("Error occurred when creating snapshot %s.", snapshotName)
+	}
+}
 
-		if err != nil {
-			glog.Errorf("Error creating the VolumeSnapshotData %s: %v", snapshotName, err)
-			// Don't proceed to create snapshot using the plugin due to error creating
-			// VolumeSnapshotData
-			return fmt.Errorf("Failed to create the VolumeSnapshotData %s for snapshot %s", snapName, snapshotName)
+func (vs *volumeSnapshotter) findSnapshot(snapshotName string, snapshot *crdv1.VolumeSnapshot) (*crdv1.VolumeSnapshotDataSource, *[]crdv1.VolumeSnapshotCondition, error) {
+	glog.Infof("findSnapshot: snapshot %s", snapshotName)
+	var snapshotDataSource *crdv1.VolumeSnapshotDataSource = nil
+	var conditions *[]crdv1.VolumeSnapshotCondition = nil
+	tags := vs.findVolumeSnapshotMetadata(snapshot)
+	if tags != nil {
+		plugin, err := vs.getPlugin(snapshotName, snapshot)
+		if plugin == nil {
+			glog.Errorf("Failed to get volume plugin. %v", err)
+			return nil, nil, fmt.Errorf("Failed to get volume plugin to create snapshot %s", snapshotName)
 		}
+		// Check whether the real snapshot is already created by the plugin
+		glog.Infof("findSnapshot: find snapshot %s by tags %v.", snapshotName, tags)
+		snapshotDataSource, conditions, err = (*plugin).FindSnapshot(tags)
+		if err == nil {
+			glog.Infof("findSnapshot: found snapshot %s.", snapshotName)
+			return snapshotDataSource, conditions, nil
+		}
+		return nil, nil, err
+	}
+	return nil, nil, fmt.Errorf("No metadata found in snapshot %s", snapshotName)
+}
+
+func (vs *volumeSnapshotter) createSnapshot(snapshotName string, snapshot *crdv1.VolumeSnapshot) error {
+	var snapshotDataSource *crdv1.VolumeSnapshotDataSource = nil
+	var snapStatus *[]crdv1.VolumeSnapshotCondition = nil
+	var err error = nil
+	var tags *map[string]string = nil
+	glog.Infof("createSnapshot: Create metadata for snapshot %s.", snapshotName)
+	tags, err = vs.updateVolumeSnapshotMetadata(snapshot)
+	if err != nil {
+		return fmt.Errorf("Failed to update metadata for volume snapshot %s: %q", snapshotName, err)
+	}
+
+	glog.Infof("createSnapshot: Creating snapshot %s through the plugin ...", snapshotName)
+	pv, err := vs.getPVFromVolumeSnapshot(snapshotName, snapshot)
+	if err != nil {
+		return err
+	}
+	snapshotDataSource, snapStatus, err = vs.takeSnapshot(pv, tags)
+	if err != nil || snapshotDataSource == nil {
+		return fmt.Errorf("Failed to take snapshot of the volume %s: %q", pv.Name, err)
+	}
+
+	glog.Infof("createSnapshot: Update status for VolumeSnapshot object %s.", snapshotName)
+	_, err = vs.UpdateVolumeSnapshot(snapshotName, snapStatus)
+	if err != nil {
+		glog.Errorf("createSnapshot: Error updating volume snapshot %s: %v", snapshotName, err)
+		return fmt.Errorf("Failed to update VolumeSnapshot for snapshot %s", snapshotName)
+	}
+
+	glog.Infof("createSnapshot: create VolumeSnapshotData object for VolumeSnapshot %s.", snapshotName)
+	snapshotDataObj, err := vs.createVolumeSnapshotData(snapshotName, snapshot, snapshotDataSource)
+	if err != nil {
+		return err
+	}
 
-		// Update the VolumeSnapshot object too
-		err = vs.UpdateVolumeSnapshot(snapshotName)
+	glog.Infof("createSnapshot: bind VolumeSnapshotData to VolumeSnapshot %s.", snapshotName)
+	err = vs.bindVolumeSnapshotDataToVolumeSnapshot(snapshotName, snapshotDataObj.Metadata.Name)
+	if err != nil {
+		glog.Errorf("createSnapshot: Error binding VolumeSnapshotData to VolumeSnapshot %s: %v", snapshotName, err)
+		return fmt.Errorf("Failed to bind VolumeSnapshotData to VolumeSnapshot %s", snapshotName)
+	}
+
+	// Waiting for snapshot to be ready
+	err = vs.waitForSnapshot(snapshotName, snapshot, snapshotDataObj)
+	if err != nil {
+		return fmt.Errorf("Failed to create snapshot %s.", snapshotName)
+	}
+	glog.Infof("createSnapshot: Snapshot %s created successfully.", snapshotName)
+	return nil
+}
+
+func (vs *volumeSnapshotter) createVolumeSnapshotData(snapshotName string, snapshot *crdv1.VolumeSnapshot, snapshotDataSource *crdv1.VolumeSnapshotDataSource) (*crdv1.VolumeSnapshotData, error) {
+	var snapshotObj crdv1.VolumeSnapshot
+	// Need to get a fresh copy of the VolumeSnapshot with the updated status
+	err := vs.restClient.Get().
+		Name(snapshot.Metadata.Name).
+		Resource(crdv1.VolumeSnapshotResourcePlural).
+		Namespace(snapshot.Metadata.Namespace).
+		Do().Into(&snapshotObj)
+
+	conditions := snapshotObj.Status.Conditions
+	if len(conditions) == 0 {
+		glog.Infof("createVolumeSnapshotData: Failed to create VolumeSnapshotDate for snapshot %s. No status info from the VolumeSnapshot object.", snapshotName)
+		return nil, fmt.Errorf("Failed to create VolumeSnapshotData for snapshot %s", snapshotName)
+	}
+	ind := len(conditions) - 1
+	glog.Infof("createVolumeSnapshotData: Snapshot %s. Conditions: %#v", snapshotName, conditions)
+	readyCondition := crdv1.VolumeSnapshotDataCondition{
+		Type:    (crdv1.VolumeSnapshotDataConditionType)(conditions[ind].Type),
+		Status:  conditions[ind].Status,
+		Message: conditions[ind].Message,
+	}
+	snapName := "k8s-volume-snapshot-" + string(uuid.NewUUID())
+
+	pv, err := vs.getPVFromVolumeSnapshot(snapshotName, snapshot)
+	if err != nil {
+		return nil, err
+	}
+	pvName := pv.Name
+
+	snapshotData := &crdv1.VolumeSnapshotData{
+		Metadata: metav1.ObjectMeta{
+			Name: snapName,
+		},
+		Spec: crdv1.VolumeSnapshotDataSpec{
+			VolumeSnapshotRef: &v1.ObjectReference{
+				Kind: "VolumeSnapshot",
+				Name: snapshotName,
+			},
+			PersistentVolumeRef: &v1.ObjectReference{
+				Kind: "PersistentVolume",
+				Name: pvName,
+			},
+			VolumeSnapshotDataSource: *snapshotDataSource,
+		},
+		Status: crdv1.VolumeSnapshotDataStatus{
+			Conditions: []crdv1.VolumeSnapshotDataCondition{
+				readyCondition,
+			},
+		},
+	}
+	var result crdv1.VolumeSnapshotData
+	for i := 0; i < createVolumeSnapshotDataRetryCount; i++ {
+		err = vs.restClient.Post().
+			Resource(crdv1.VolumeSnapshotDataResourcePlural).
+			Namespace(v1.NamespaceDefault).
+			Body(snapshotData).
+			Do().Into(&result)
 		if err != nil {
-			glog.Errorf("Error updating volume snapshot %s: %v", snapshotName, err)
-			// NOTE(xyang): Return error if failed to update VolumeSnapshot after
-			// create snapshot request is sent to the plugin and VolumeSnapshotData is updated
-			return fmt.Errorf("Failed to update VolumeSnapshot for snapshot %s", snapshotName)
+			// TODO(xyang): Use exponential backoff instead of a fixed internal.
+			// https://github.com/jpillora/backoff
+			// Re-Try it as errors writing to the API server are common
+			time.Sleep(createVolumeSnapshotDataInterval)
+		} else {
+			break
 		}
+	}
 
-		return nil
+	if err != nil {
+		glog.Errorf("createVolumeSnapshotData: Error creating the VolumeSnapshotData %s: %v", snapshotName, err)
+		return nil, fmt.Errorf("Failed to create the VolumeSnapshotData %s for snapshot %s", snapName, snapshotName)
 	}
+	return &result, nil
 }
 
 func (vs *volumeSnapshotter) getSnapshotDeleteFunc(snapshotName string, snapshot *crdv1.VolumeSnapshot) func() error {
@@ -395,9 +704,9 @@ func (vs *volumeSnapshotter) getSnapshotPromoteFunc(snapshotName string, snapsho
 func (vs *volumeSnapshotter) CreateVolumeSnapshot(snapshot *crdv1.VolumeSnapshot) {
 	snapshotName := cache.MakeSnapshotName(snapshot.Metadata.Namespace, snapshot.Metadata.Name)
 	operationName := snapshotOpCreatePrefix + snapshotName + snapshot.Spec.PersistentVolumeClaimName
-	glog.Infof("Snapshotter is about to create volume snapshot operation named %s, spec %#v", operationName, snapshot.Spec)
+	//glog.Infof("Snapshotter is about to create volume snapshot operation named %s, spec %#v", operationName, snapshot.Spec)
 
-	err := vs.runningOperation.Run(operationName, vs.getSnapshotCreateFunc(snapshotName, snapshot))
+	err := vs.runningOperation.Run(operationName, vs.syncSnapshot(snapshotName, snapshot))
 
 	if err != nil {
 		switch {
@@ -449,16 +758,61 @@ func (vs *volumeSnapshotter) PromoteVolumeSnapshotToPV(snapshot *crdv1.VolumeSna
 	}
 }
 
-func (vs *volumeSnapshotter) UpdateVolumeSnapshot(snapshotName string) error {
+func (vs *volumeSnapshotter) updateVolumeSnapshotMetadata(snapshot *crdv1.VolumeSnapshot) (*map[string]string, error) {
+	glog.Infof("In updateVolumeSnapshotMetadata")
+	var snapshotObj crdv1.VolumeSnapshot
+	// Need to get a fresh copy of the VolumeSnapshot from the API server
+	err := vs.restClient.Get().
+		Name(snapshot.Metadata.Name).
+		Resource(crdv1.VolumeSnapshotResourcePlural).
+		Namespace(snapshot.Metadata.Namespace).
+		Do().Into(&snapshotObj)
+
+	// Copy the snapshot object before updating it
+	objCopy, err := vs.scheme.DeepCopy(&snapshotObj)
+	if err != nil {
+		return nil, fmt.Errorf("Error copying snapshot object %s object: %v", snapshot.Metadata.Name, err)
+	}
+	snapshotCopy, ok := objCopy.(*crdv1.VolumeSnapshot)
+	if !ok {
+		return nil, fmt.Errorf("Error: expecting type VolumeSnapshot but received type %T", objCopy)
+	}
+
+	tags := make(map[string]string)
+	tags["Timestamp"] = fmt.Sprintf("%d", time.Now().UnixNano())
+	snapshotCopy.Metadata.Labels = tags
+	glog.Infof("updateVolumeSnapshotMetadata: Metadata Name: %s Metadata Namespace: %s Setting tags in Metadata Labels: %#v.", snapshotCopy.Metadata.Name, snapshotCopy.Metadata.Namespace, snapshotCopy.Metadata.Labels)
+
+	var result crdv1.VolumeSnapshot
+	err = vs.restClient.Put().
+		Name(snapshot.Metadata.Name).
+		Resource(crdv1.VolumeSnapshotResourcePlural).
+		Namespace(snapshotCopy.Metadata.Namespace).
+		Body(snapshotCopy).
+		Do().Into(&result)
+	if err != nil {
+		return nil, fmt.Errorf("Error updating snapshot object %s on the API server: %v", snapshot.Metadata.Name, err)
+	}
+
+	cloudTags := make(map[string]string)
+	cloudTags[CloudSnapshotCreatedForVolumeSnapshotNamespaceTag] = result.Metadata.Namespace
+	cloudTags[CloudSnapshotCreatedForVolumeSnapshotNameTag] = result.Metadata.Name
+	cloudTags[CloudSnapshotCreatedForVolumeSnapshotTimestampTag] = result.Metadata.Labels["Timestamp"]
+
+	glog.Infof("updateVolumeSnapshotMetadata: returning cloudTags [%#v]", cloudTags)
+	return &cloudTags, nil
+}
+
+func (vs *volumeSnapshotter) UpdateVolumeSnapshot(snapshotName string, status *[]crdv1.VolumeSnapshotCondition) (*crdv1.VolumeSnapshot, error) {
 	var snapshotObj crdv1.VolumeSnapshot
 
 	glog.Infof("In UpdateVolumeSnapshot")
-	// Get a fresh copy of the VolumeSnapshotData from the API server
 	// Get a fresh copy of the VolumeSnapshot from the API server
 	snapNameSpace, snapName, err := cache.GetNameAndNameSpaceFromSnapshotName(snapshotName)
 	if err != nil {
-		return fmt.Errorf("Error getting namespace and name from VolumeSnapshot name %s: %v", snapshotName, err)
+		return nil, fmt.Errorf("Error getting namespace and name from VolumeSnapshot name %s: %v", snapshotName, err)
 	}
+	glog.Infof("UpdateVolumeSnapshot: Namespace %s Name %s", snapNameSpace, snapName)
 	err = vs.restClient.Get().
 		Name(snapName).
 		Resource(crdv1.VolumeSnapshotResourcePlural).
@@ -467,29 +821,76 @@ func (vs *volumeSnapshotter) UpdateVolumeSnapshot(snapshotName string) error {
 
 	objCopy, err := vs.scheme.DeepCopy(&snapshotObj)
 	if err != nil {
-		return fmt.Errorf("Error copying snapshot object %s object from API server: %v", snapshotName, err)
+		return nil, fmt.Errorf("Error copying snapshot object %s object from API server: %v", snapshotName, err)
 	}
 	snapshotCopy, ok := objCopy.(*crdv1.VolumeSnapshot)
 	if !ok {
-		return fmt.Errorf("Error: expecting type VolumeSnapshot but received type %T", objCopy)
+		return nil, fmt.Errorf("Error: expecting type VolumeSnapshot but received type %T", objCopy)
 	}
 
 	snapshotDataObj := vs.getSnapshotDataFromSnapshotName(snapshotName)
 	if snapshotDataObj == nil {
-		return fmt.Errorf("Error getting VolumeSnapshotData for VolumeSnapshot %s", snapshotName)
+		glog.Infof("UpdateVolumeSnapshot: VolumeSnapshotData not created for snapshot %s yet.", snapName)
+	} else {
+		glog.Infof("UpdateVolumeSnapshot: Setting VolumeSnapshotData name %s in VolumeSnapshot object %s", snapshotDataObj.Metadata.Name, snapName)
+		snapshotCopy.Spec.SnapshotDataName = snapshotDataObj.Metadata.Name
+	}
+
+	if status != nil && len(*status) > 0 {
+		glog.Infof("UpdateVolumeSnapshot: Setting status in VolumeSnapshot object.")
+		// TODO(xyang): Only the last status is recorded for now. Will revisit later whether previous statuses
+		// should be kept and the last status should be added to existing ones.
+		ind := len(*status) - 1
+		ind2 := len(snapshotCopy.Status.Conditions)
+		if ind2 < 1 || snapshotCopy.Status.Conditions[ind2-1].Type != (*status)[ind].Type {
+			snapshotCopy.Status.Conditions = append(snapshotCopy.Status.Conditions, (*status)[ind])
+		} else if snapshotCopy.Status.Conditions[ind2-1].Type == (*status)[ind].Type {
+			snapshotCopy.Status.Conditions[ind2-1] = (*status)[ind]
+		}
+	}
+	glog.Infof("Updating VolumeSnapshot object [%#v]", snapshotCopy)
+	// TODO: Make diff of the two objects and then use restClient.Patch to update it
+	var result crdv1.VolumeSnapshot
+	err = vs.restClient.Put().
+		Name(snapName).
+		Resource(crdv1.VolumeSnapshotResourcePlural).
+		Namespace(snapNameSpace).
+		Body(snapshotCopy).
+		Do().Into(&result)
+	if err != nil {
+		return nil, fmt.Errorf("Error updating snapshot object %s on the API server: %v", snapshotName, err)
 	}
 
-	glog.Infof("UpdateVolumeSnapshot: Setting VolumeSnapshotData name in VolumeSnapshotSpec of VolumeSnapshot object")
-	snapshotCopy.Spec.SnapshotDataName = snapshotDataObj.Metadata.Name
-	snapshotCopy.Status.Conditions = []crdv1.VolumeSnapshotCondition{
-		{
-			Type:               crdv1.VolumeSnapshotConditionReady,
-			Status:             v1.ConditionTrue,
-			Message:            "Snapshot created succsessfully",
-			LastTransitionTime: metav1.Now(),
-		},
+	return &result, nil
+}
+
+func (vs *volumeSnapshotter) bindVolumeSnapshotDataToVolumeSnapshot(snapshotName string, snapshotDataName string) error {
+	var snapshotObj crdv1.VolumeSnapshot
+
+	glog.Infof("In bindVolumeSnapshotDataToVolumeSnapshot")
+	// Get a fresh copy of the VolumeSnapshot from the API server
+	snapNameSpace, snapName, err := cache.GetNameAndNameSpaceFromSnapshotName(snapshotName)
+	if err != nil {
+		return fmt.Errorf("Error getting namespace and name from VolumeSnapshot name %s: %v", snapshotName, err)
 	}
-	glog.Infof("Updating VolumeSnapshot object")
+	glog.Infof("bindVolumeSnapshotDataToVolumeSnapshot: Namespace %s Name %s", snapNameSpace, snapName)
+	err = vs.restClient.Get().
+		Name(snapName).
+		Resource(crdv1.VolumeSnapshotResourcePlural).
+		Namespace(snapNameSpace).
+		Do().Into(&snapshotObj)
+
+	objCopy, err := vs.scheme.DeepCopy(&snapshotObj)
+	if err != nil {
+		return fmt.Errorf("Error copying snapshot object %s object from API server: %v", snapshotName, err)
+	}
+	snapshotCopy, ok := objCopy.(*crdv1.VolumeSnapshot)
+	if !ok {
+		return fmt.Errorf("Error: expecting type VolumeSnapshot but received type %T", objCopy)
+	}
+
+	snapshotCopy.Spec.SnapshotDataName = snapshotDataName
+	glog.Infof("bindVolumeSnapshotDataToVolumeSnapshot: Updating VolumeSnapshot object [%#v]", snapshotCopy)
 	// TODO: Make diff of the two objects and then use restClient.Patch to update it
 	var result crdv1.VolumeSnapshot
 	err = vs.restClient.Put().
@@ -501,13 +902,14 @@ func (vs *volumeSnapshotter) UpdateVolumeSnapshot(snapshotName string) error {
 	if err != nil {
 		return fmt.Errorf("Error updating snapshot object %s on the API server: %v", snapshotName, err)
 	}
+
 	return nil
 }
 
 func (vs *volumeSnapshotter) UpdateVolumeSnapshotData(snapshotDataName string, status *[]crdv1.VolumeSnapshotDataCondition) error {
 	var snapshotDataObj crdv1.VolumeSnapshotData
 
-	glog.Infof("In UpdateVolumeSnapshotData")
+	glog.Infof("In UpdateVolumeSnapshotData %s", snapshotDataName)
 	err := vs.restClient.Get().
 		Name(snapshotDataName).
 		Resource(crdv1.VolumeSnapshotDataResourcePlural).
@@ -523,8 +925,16 @@ func (vs *volumeSnapshotter) UpdateVolumeSnapshotData(snapshotDataName string, s
 		return fmt.Errorf("Error: expecting type VolumeSnapshotData but received type %T", objCopy)
 	}
 
-	snapshotDataCopy.Status.Conditions = *status
-	glog.Infof("Updating VolumeSnapshotData object")
+	// TODO(xyang): Only the last status is recorded for now. Will revisit later whether previous statuses
+	// should be kept and the last status should be added to existing ones.
+	ind := len(*status) - 1
+	ind2 := len(snapshotDataCopy.Status.Conditions)
+	if ind2 < 1 || snapshotDataCopy.Status.Conditions[ind2-1].Type != (*status)[ind].Type {
+		snapshotDataCopy.Status.Conditions = append(snapshotDataCopy.Status.Conditions, (*status)[ind])
+	} else if snapshotDataCopy.Status.Conditions[ind2-1].Type == (*status)[ind].Type {
+		snapshotDataCopy.Status.Conditions[ind2-1] = (*status)[ind]
+	}
+	glog.Infof("Updating VolumeSnapshotData object. Conditions: [%v]", snapshotDataCopy.Status.Conditions)
 	// TODO: Make diff of the two objects and then use restClient.Patch to update it
 	var result crdv1.VolumeSnapshotData
 	err = vs.restClient.Put().
diff --git a/snapshot/pkg/volume/aws_ebs/processor.go b/snapshot/pkg/volume/aws_ebs/processor.go
index 52c5d90c36e..023ae6afc86 100644
--- a/snapshot/pkg/volume/aws_ebs/processor.go
+++ b/snapshot/pkg/volume/aws_ebs/processor.go
@@ -22,6 +22,7 @@ import (
 	"strings"
 
 	"k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	kvol "k8s.io/kubernetes/pkg/volume"
 
 	"github.com/golang/glog"
@@ -50,10 +51,10 @@ func (a *awsEBSPlugin) Init(cloud cloudprovider.Interface) {
 	a.cloud = cloud.(*aws.Cloud)
 }
 
-func (a *awsEBSPlugin) SnapshotCreate(pv *v1.PersistentVolume) (*crdv1.VolumeSnapshotDataSource, error) {
+func (a *awsEBSPlugin) SnapshotCreate(pv *v1.PersistentVolume, tags *map[string]string) (*crdv1.VolumeSnapshotDataSource, *[]crdv1.VolumeSnapshotCondition, error) {
 	spec := &pv.Spec
 	if spec == nil || spec.AWSElasticBlockStore == nil {
-		return nil, fmt.Errorf("invalid PV spec %v", spec)
+		return nil, nil, fmt.Errorf("invalid PV spec %v", spec)
 	}
 	volumeId := spec.AWSElasticBlockStore.VolumeID
 	if ind := strings.LastIndex(volumeId, "/"); ind >= 0 {
@@ -61,16 +62,18 @@ func (a *awsEBSPlugin) SnapshotCreate(pv *v1.PersistentVolume) (*crdv1.VolumeSna
 	}
 	snapshotOpt := &aws.SnapshotOptions{
 		VolumeId: volumeId,
+		Tags:     tags,
 	}
-	snapshotId, err := a.cloud.CreateSnapshot(snapshotOpt)
+	// TODO: Convert AWS EBS snapshot status to crdv1.VolumeSnapshotCondition
+	snapshotId, status, err := a.cloud.CreateSnapshot(snapshotOpt)
 	if err != nil {
-		return nil, err
+		return nil, nil, err
 	}
 	return &crdv1.VolumeSnapshotDataSource{
 		AWSElasticBlockStore: &crdv1.AWSElasticBlockStoreVolumeSnapshotSource{
 			SnapshotID: snapshotId,
 		},
-	}, nil
+	}, convertAWSStatus(status), nil
 }
 
 func (a *awsEBSPlugin) SnapshotDelete(src *crdv1.VolumeSnapshotDataSource, _ *v1.PersistentVolume) error {
@@ -79,6 +82,7 @@ func (a *awsEBSPlugin) SnapshotDelete(src *crdv1.VolumeSnapshotDataSource, _ *v1
 	}
 	snapshotId := src.AWSElasticBlockStore.SnapshotID
 	_, err := a.cloud.DeleteSnapshot(snapshotId)
+	glog.Infof("delete snapshot %s, err: %v", snapshotId, err)
 	if err != nil {
 		return err
 	}
@@ -86,12 +90,25 @@ func (a *awsEBSPlugin) SnapshotDelete(src *crdv1.VolumeSnapshotDataSource, _ *v1
 	return nil
 }
 
-func (a *awsEBSPlugin) DescribeSnapshot(snapshotData *crdv1.VolumeSnapshotData) (isCompleted bool, err error) {
+func (a *awsEBSPlugin) DescribeSnapshot(snapshotData *crdv1.VolumeSnapshotData) (snapConditions *[]crdv1.VolumeSnapshotCondition, isCompleted bool, err error) {
 	if snapshotData == nil || snapshotData.Spec.AWSElasticBlockStore == nil {
-		return false, fmt.Errorf("invalid VolumeSnapshotDataSource: %v", snapshotData)
+		return nil, false, fmt.Errorf("invalid VolumeSnapshotDataSource: %v", snapshotData)
 	}
 	snapshotId := snapshotData.Spec.AWSElasticBlockStore.SnapshotID
-	return a.cloud.DescribeSnapshot(snapshotId)
+	status, isCompleted, err := a.cloud.DescribeSnapshot(snapshotId)
+	return convertAWSStatus(status), isCompleted, err
+}
+
+// FindSnapshot finds a VolumeSnapshot by matching metadata
+func (a *awsEBSPlugin) FindSnapshot(tags *map[string]string) (*crdv1.VolumeSnapshotDataSource, *[]crdv1.VolumeSnapshotCondition, error) {
+	glog.Infof("FindSnapshot by tags: %#v", *tags)
+
+	// TODO: Implement FindSnapshot
+	return &crdv1.VolumeSnapshotDataSource{
+		AWSElasticBlockStore: &crdv1.AWSElasticBlockStoreVolumeSnapshotSource{
+			SnapshotID: "",
+		},
+	}, nil, nil
 }
 
 func (a *awsEBSPlugin) SnapshotRestore(snapshotData *crdv1.VolumeSnapshotData, pvc *v1.PersistentVolumeClaim, pvName string, parameters map[string]string) (*v1.PersistentVolumeSource, map[string]string, error) {
@@ -187,3 +204,37 @@ func (a *awsEBSPlugin) VolumeDelete(pv *v1.PersistentVolume) error {
 
 	return nil
 }
+
+func convertAWSStatus(status string) *[]crdv1.VolumeSnapshotCondition {
+	var snapConditions []crdv1.VolumeSnapshotCondition
+	if strings.ToLower(status) == "completed" {
+		snapConditions = []crdv1.VolumeSnapshotCondition{
+			{
+				Type:               crdv1.VolumeSnapshotConditionReady,
+				Status:             v1.ConditionTrue,
+				Message:            "Snapshot created succsessfully and it is ready",
+				LastTransitionTime: metav1.Now(),
+			},
+		}
+	} else if strings.ToLower(status) == "pending" {
+		snapConditions = []crdv1.VolumeSnapshotCondition{
+			{
+				Type:               crdv1.VolumeSnapshotConditionPending,
+				Status:             v1.ConditionUnknown,
+				Message:            "Snapshot is being created",
+				LastTransitionTime: metav1.Now(),
+			},
+		}
+	} else {
+		snapConditions = []crdv1.VolumeSnapshotCondition{
+			{
+				Type:               crdv1.VolumeSnapshotConditionError,
+				Status:             v1.ConditionTrue,
+				Message:            "Snapshot creation failed",
+				LastTransitionTime: metav1.Now(),
+			},
+		}
+	}
+
+	return &snapConditions
+}
diff --git a/snapshot/pkg/volume/cinder/processor.go b/snapshot/pkg/volume/cinder/processor.go
index 67a8b0394dd..d7f5e6b1669 100644
--- a/snapshot/pkg/volume/cinder/processor.go
+++ b/snapshot/pkg/volume/cinder/processor.go
@@ -22,6 +22,7 @@ import (
 	"time"
 
 	"k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 
 	"github.com/golang/glog"
 
@@ -68,26 +69,25 @@ func (c *cinderPlugin) VolumeDelete(pv *v1.PersistentVolume) error {
 }
 
 // SnapshotCreate creates a VolumeSnapshot from a PersistentVolumeSpec
-func (c *cinderPlugin) SnapshotCreate(pv *v1.PersistentVolume) (*crdv1.VolumeSnapshotDataSource, error) {
+func (c *cinderPlugin) SnapshotCreate(pv *v1.PersistentVolume, tags *map[string]string) (*crdv1.VolumeSnapshotDataSource, *[]crdv1.VolumeSnapshotCondition, error) {
 	spec := &pv.Spec
 	if spec == nil || spec.Cinder == nil {
-		return nil, fmt.Errorf("invalid PV spec %v", spec)
+		return nil, nil, fmt.Errorf("invalid PV spec %v", spec)
 	}
 	volumeID := spec.Cinder.VolumeID
 	snapshotName := string(pv.Name) + fmt.Sprintf("%d", time.Now().UnixNano())
 	snapshotDescription := "kubernetes snapshot"
-	tags := make(map[string]string)
-	glog.Infof("issuing Cinder.CreateSnapshot - SourceVol: %s, Name: %s", volumeID, snapshotName)
-	snapID, err := c.cloud.CreateSnapshot(volumeID, snapshotName, snapshotDescription, tags)
+	glog.Infof("issuing Cinder.CreateSnapshot - SourceVol: %s, Name: %s, tags: %#v", volumeID, snapshotName, *tags)
+	snapID, status, err := c.cloud.CreateSnapshot(volumeID, snapshotName, snapshotDescription, *tags)
 	if err != nil {
-		return nil, err
+		return nil, nil, err
 	}
 
 	return &crdv1.VolumeSnapshotDataSource{
 		CinderSnapshot: &crdv1.CinderVolumeSnapshotSource{
 			SnapshotID: snapID,
 		},
-	}, nil
+	}, c.convertSnapshotStatus(status), nil
 }
 
 // SnapshotDelete deletes a VolumeSnapshot
@@ -151,14 +151,71 @@ func (c *cinderPlugin) SnapshotRestore(snapshotData *crdv1.VolumeSnapshotData, p
 }
 
 // DescribeSnapshot retrieves info for the specified Snapshot
-func (c *cinderPlugin) DescribeSnapshot(snapshotData *crdv1.VolumeSnapshotData) (bool, error) {
+func (c *cinderPlugin) DescribeSnapshot(snapshotData *crdv1.VolumeSnapshotData) (*[]crdv1.VolumeSnapshotCondition, bool, error) {
 	if snapshotData == nil || snapshotData.Spec.CinderSnapshot == nil {
-		return false, fmt.Errorf("invalid VolumeSnapshotDataSource: %v", snapshotData)
+		return nil, false, fmt.Errorf("invalid VolumeSnapshotDataSource: %v", snapshotData)
 	}
 	snapshotID := snapshotData.Spec.CinderSnapshot.SnapshotID
-	isComplete, err := c.cloud.DescribeSnapshot(snapshotID)
+	status, isComplete, err := c.cloud.DescribeSnapshot(snapshotID)
+	glog.Infof("DescribeSnapshot: Snapshot %s, Status %s, isComplete: %v", snapshotID, status, isComplete)
 	if err != nil {
-		return false, err
+		return c.convertSnapshotStatus(status), false, err
+	}
+	return c.convertSnapshotStatus(status), isComplete, nil
+}
+
+// convertSnapshotStatus converts Cinder snapshot status to crdv1.VolumeSnapshotCondition
+func (c *cinderPlugin) convertSnapshotStatus(status string) *[]crdv1.VolumeSnapshotCondition {
+	var snapConditions []crdv1.VolumeSnapshotCondition
+	if status == "available" {
+		snapConditions = []crdv1.VolumeSnapshotCondition{
+			{
+				Type:               crdv1.VolumeSnapshotConditionReady,
+				Status:             v1.ConditionTrue,
+				Message:            "Snapshot created succsessfully and it is ready",
+				LastTransitionTime: metav1.Now(),
+			},
+		}
+	} else if status == "creating" {
+		snapConditions = []crdv1.VolumeSnapshotCondition{
+			{
+				Type:               crdv1.VolumeSnapshotConditionPending,
+				Status:             v1.ConditionUnknown,
+				Message:            "Snapshot is being created",
+				LastTransitionTime: metav1.Now(),
+			},
+		}
+	} else {
+		snapConditions = []crdv1.VolumeSnapshotCondition{
+			{
+				Type:               crdv1.VolumeSnapshotConditionError,
+				Status:             v1.ConditionTrue,
+				Message:            "Snapshot creation failed",
+				LastTransitionTime: metav1.Now(),
+			},
+		}
 	}
-	return isComplete, nil
+
+	return &snapConditions
+}
+
+// FindSnapshot finds a VolumeSnapshot by matching metadata
+func (c *cinderPlugin) FindSnapshot(tags *map[string]string) (*crdv1.VolumeSnapshotDataSource, *[]crdv1.VolumeSnapshotCondition, error) {
+	glog.Infof("Cinder.FindSnapshot by tags: %#v", *tags)
+	snapIDs, statuses, err := c.cloud.FindSnapshot(*tags)
+	if err != nil {
+		glog.Infof("Cinder.FindSnapshot by tags: %#v. Error: %v", *tags, err)
+		//return nil, err
+	}
+
+	if len(snapIDs) > 0 && len(statuses) > 0 {
+		glog.Infof("Found snapshot %s by tags: %#v", snapIDs[0], *tags)
+		return &crdv1.VolumeSnapshotDataSource{
+			CinderSnapshot: &crdv1.CinderVolumeSnapshotSource{
+				SnapshotID: snapIDs[0],
+			},
+		}, c.convertSnapshotStatus(statuses[0]), nil
+	}
+
+	return nil, nil, nil
 }
diff --git a/snapshot/pkg/volume/gce_pd/processor.go b/snapshot/pkg/volume/gce_pd/processor.go
index dcd27b81668..8490e88f27e 100644
--- a/snapshot/pkg/volume/gce_pd/processor.go
+++ b/snapshot/pkg/volume/gce_pd/processor.go
@@ -53,30 +53,30 @@ func (plugin *gcePersistentDiskPlugin) Init(cloud cloudprovider.Interface) {
 	plugin.cloud = cloud.(*gce.GCECloud)
 }
 
-func (plugin *gcePersistentDiskPlugin) SnapshotCreate(pv *v1.PersistentVolume) (*crdv1.VolumeSnapshotDataSource, error) {
+func (plugin *gcePersistentDiskPlugin) SnapshotCreate(pv *v1.PersistentVolume, tags *map[string]string) (*crdv1.VolumeSnapshotDataSource, *[]crdv1.VolumeSnapshotCondition, error) {
 	spec := &pv.Spec
 	if spec == nil || spec.GCEPersistentDisk == nil {
-		return nil, fmt.Errorf("invalid PV spec %v", spec)
+		return nil, nil, fmt.Errorf("invalid PV spec %v", spec)
 	}
 	diskName := spec.GCEPersistentDisk.PDName
 	zone := pv.Labels[apis.LabelZoneFailureDomain]
 	snapshotName := createSnapshotName(string(pv.Name))
 	glog.Infof("Jing snapshotName %s", snapshotName)
 	// Gather provisioning options
-	tags := make(map[string]string)
+	//tags := make(map[string]string)
 	//tags["kubernetes.io/created-for/snapshot/namespace"] = claim.Namespace
 	//tags[CloudVolumeCreatedForClaimNameTag] = claim.Name
 	//tags[CloudVolumeCreatedForVolumeNameTag] = pvName
 
-	err := plugin.cloud.CreateSnapshot(diskName, zone, snapshotName, tags)
+	err := plugin.cloud.CreateSnapshot(diskName, zone, snapshotName, *tags)
 	if err != nil {
-		return nil, err
+		return nil, nil, err
 	}
 	return &crdv1.VolumeSnapshotDataSource{
 		GCEPersistentDiskSnapshot: &crdv1.GCEPersistentDiskSnapshotSource{
 			SnapshotName: snapshotName,
 		},
-	}, nil
+	}, nil, nil
 }
 
 func (plugin *gcePersistentDiskPlugin) SnapshotRestore(snapshotData *crdv1.VolumeSnapshotData, pvc *v1.PersistentVolumeClaim, pvName string, parameters map[string]string) (*v1.PersistentVolumeSource, map[string]string, error) {
@@ -166,12 +166,26 @@ func (plugin *gcePersistentDiskPlugin) SnapshotDelete(src *crdv1.VolumeSnapshotD
 	return nil
 }
 
-func (plugin *gcePersistentDiskPlugin) DescribeSnapshot(snapshotData *crdv1.VolumeSnapshotData) (isCompleted bool, err error) {
+func (plugin *gcePersistentDiskPlugin) DescribeSnapshot(snapshotData *crdv1.VolumeSnapshotData) (snapConditions *[]crdv1.VolumeSnapshotCondition, isCompleted bool, err error) {
 	if snapshotData == nil || snapshotData.Spec.GCEPersistentDiskSnapshot == nil {
-		return false, fmt.Errorf("invalid VolumeSnapshotDataSource: %v", snapshotData)
+		return nil, false, fmt.Errorf("invalid VolumeSnapshotDataSource: %v", snapshotData)
 	}
 	snapshotId := snapshotData.Spec.GCEPersistentDiskSnapshot.SnapshotName
-	return plugin.cloud.DescribeSnapshot(snapshotId)
+	_, isCompleted, err = plugin.cloud.DescribeSnapshot(snapshotId)
+	// TODO: Convert GCE status to []crdv1.VolumeSnapshotCondition
+	return nil, isCompleted, err
+}
+
+// FindSnapshot finds a VolumeSnapshot by matching metadata
+func (a *gcePersistentDiskPlugin) FindSnapshot(tags *map[string]string) (*crdv1.VolumeSnapshotDataSource, *[]crdv1.VolumeSnapshotCondition, error) {
+	glog.Infof("FindSnapshot by tags: %#v", *tags)
+
+	// TODO: Implement FindSnapshot
+	return &crdv1.VolumeSnapshotDataSource{
+		GCEPersistentDiskSnapshot: &crdv1.GCEPersistentDiskSnapshotSource{
+			SnapshotName: "",
+		},
+	}, nil, nil
 }
 
 func (plugin *gcePersistentDiskPlugin) VolumeDelete(pv *v1.PersistentVolume) error {
diff --git a/snapshot/pkg/volume/hostpath/processor.go b/snapshot/pkg/volume/hostpath/processor.go
index 0f16b5af6c9..57e8f211efb 100644
--- a/snapshot/pkg/volume/hostpath/processor.go
+++ b/snapshot/pkg/volume/hostpath/processor.go
@@ -21,7 +21,9 @@ import (
 	"os"
 	"os/exec"
 
+	"github.com/golang/glog"
 	"k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/uuid"
 
 	crdv1 "github.com/kubernetes-incubator/external-storage/snapshot/pkg/apis/crd/v1"
@@ -50,20 +52,41 @@ func GetPluginName() string {
 func (h *hostPathPlugin) Init(_ cloudprovider.Interface) {
 }
 
-func (h *hostPathPlugin) SnapshotCreate(pv *v1.PersistentVolume) (*crdv1.VolumeSnapshotDataSource, error) {
+func (h *hostPathPlugin) SnapshotCreate(pv *v1.PersistentVolume, tags *map[string]string) (*crdv1.VolumeSnapshotDataSource, *[]crdv1.VolumeSnapshotCondition, error) {
 	spec := &pv.Spec
 	if spec == nil || spec.HostPath == nil {
-		return nil, fmt.Errorf("invalid PV spec %v", spec)
+		return nil, nil, fmt.Errorf("invalid PV spec %v", spec)
 	}
 	path := spec.HostPath.Path
 	file := depot + string(uuid.NewUUID()) + ".tgz"
 	cmd := exec.Command("tar", "czf", file, path)
+	runErr := cmd.Run()
+	cond := []crdv1.VolumeSnapshotCondition{}
+	if runErr == nil {
+		cond = []crdv1.VolumeSnapshotCondition{
+			{
+				Status:             v1.ConditionTrue,
+				Message:            "Snapshot created successfully",
+				LastTransitionTime: metav1.Now(),
+				Type:               crdv1.VolumeSnapshotConditionReady,
+			},
+		}
+	} else {
+		cond = []crdv1.VolumeSnapshotCondition{
+			{
+				Status:             v1.ConditionTrue,
+				Message:            fmt.Sprintf("Failed to create the snapshot: %v", runErr),
+				LastTransitionTime: metav1.Now(),
+				Type:               crdv1.VolumeSnapshotConditionError,
+			},
+		}
+	}
 	res := &crdv1.VolumeSnapshotDataSource{
 		HostPath: &crdv1.HostPathVolumeSnapshotSource{
 			Path: file,
 		},
 	}
-	return res, cmd.Run()
+	return res, &cond, runErr
 }
 
 func (h *hostPathPlugin) SnapshotDelete(src *crdv1.VolumeSnapshotDataSource, _ *v1.PersistentVolume) error {
@@ -74,15 +97,27 @@ func (h *hostPathPlugin) SnapshotDelete(src *crdv1.VolumeSnapshotDataSource, _ *
 	return os.Remove(path)
 }
 
-func (a *hostPathPlugin) DescribeSnapshot(snapshotData *crdv1.VolumeSnapshotData) (isCompleted bool, err error) {
+func (a *hostPathPlugin) DescribeSnapshot(snapshotData *crdv1.VolumeSnapshotData) (snapConditions *[]crdv1.VolumeSnapshotCondition, isCompleted bool, err error) {
 	if snapshotData == nil || snapshotData.Spec.HostPath == nil {
-		return false, fmt.Errorf("failed to retrieve Snapshot spec")
+		return nil, false, fmt.Errorf("failed to retrieve Snapshot spec")
 	}
 	path := snapshotData.Spec.HostPath.Path
 	if _, err := os.Stat(path); err != nil {
-		return false, err
+		return nil, false, err
 	}
-	return true, nil
+	return nil, true, nil
+}
+
+// FindSnapshot finds a VolumeSnapshot by matching metadata
+func (a *hostPathPlugin) FindSnapshot(tags *map[string]string) (*crdv1.VolumeSnapshotDataSource, *[]crdv1.VolumeSnapshotCondition, error) {
+	glog.Infof("FindSnapshot by tags: %#v", *tags)
+
+	// TODO: Implement FindSnapshot
+	return &crdv1.VolumeSnapshotDataSource{
+		HostPath: &crdv1.HostPathVolumeSnapshotSource{
+			Path: "",
+		},
+	}, nil, nil
 }
 
 func (h *hostPathPlugin) SnapshotRestore(snapshotData *crdv1.VolumeSnapshotData, _ *v1.PersistentVolumeClaim, _ string, _ map[string]string) (*v1.PersistentVolumeSource, map[string]string, error) {
diff --git a/snapshot/pkg/volume/interface.go b/snapshot/pkg/volume/interface.go
index c1cb02e0523..742af3f620e 100644
--- a/snapshot/pkg/volume/interface.go
+++ b/snapshot/pkg/volume/interface.go
@@ -27,7 +27,7 @@ type VolumePlugin interface {
 	// Init inits volume plugin
 	Init(cloudprovider.Interface)
 	// SnapshotCreate creates a VolumeSnapshot from a PersistentVolumeSpec
-	SnapshotCreate(*v1.PersistentVolume) (*crdv1.VolumeSnapshotDataSource, error)
+	SnapshotCreate(*v1.PersistentVolume, *map[string]string) (*crdv1.VolumeSnapshotDataSource, *[]crdv1.VolumeSnapshotCondition, error)
 	// SnapshotDelete deletes a VolumeSnapshot
 	// PersistentVolume is provided for volume types, if any, that need PV Spec to delete snapshot
 	SnapshotDelete(*crdv1.VolumeSnapshotDataSource, *v1.PersistentVolume) error
@@ -35,7 +35,9 @@ type VolumePlugin interface {
 	SnapshotRestore(*crdv1.VolumeSnapshotData, *v1.PersistentVolumeClaim, string, map[string]string) (*v1.PersistentVolumeSource, map[string]string, error)
 	// Describe an EBS volume snapshot status for create or delete.
 	// return status (completed or pending or error), and error
-	DescribeSnapshot(snapshotData *crdv1.VolumeSnapshotData) (isCompleted bool, err error)
+	DescribeSnapshot(snapshotData *crdv1.VolumeSnapshotData) (snapConditions *[]crdv1.VolumeSnapshotCondition, isCompleted bool, err error)
+	// FindSnapshot finds a VolumeSnapshot by matching metadata
+	FindSnapshot(tags *map[string]string) (*crdv1.VolumeSnapshotDataSource, *[]crdv1.VolumeSnapshotCondition, error)
 	// VolumeDelete deletes a PV
 	// TODO in the future pass kubernetes client for certain volumes (e.g. rbd) so they can access storage class to retrieve secret
 	VolumeDelete(pv *v1.PersistentVolume) error