feat(zfspv): pvc should be bound only if volume has been created.

The controller does not check whether the volume has been created or not
and return successful. Which in turn binds the pvc to the pv.

The PVC should not bound until corresponding zfs volume has been created.
Now controller will check the ZFSVolume CR state to be "Ready" before returning
successful. The CSI will retry the CreateVolume request when it will get
a error reply and when the ZFS node agent creates the ZFS volume and sets the
ZFSVolume CR state to be "Ready", the controller will return success for the
CreateVolume Request and then PVC will be bound.

Signed-off-by: Pawan <pawan@mayadata.io>

pkg/apis/openebs.io/zfs/v1alpha1/zfsvolume.go

@@ -30,6 +30,7 @@ import (
 // +kubebuilder:printcolumn:name="ZPool",type=string,JSONPath=`.spec.poolName`,description="ZFS Pool where the volume is created"
 // +kubebuilder:printcolumn:name="Node",type=string,JSONPath=`.spec.ownerNodeID`,description="Node where the volume is created"
 // +kubebuilder:printcolumn:name="Size",type=string,JSONPath=`.spec.capacity`,description="Size of the volume"
+// +kubebuilder:printcolumn:name="Status",type=string,JSONPath=`.status.state`,description="Status of the volume"
 // +kubebuilder:printcolumn:name="volblocksize",type=string,JSONPath=`.spec.volblocksize`,description="volblocksize of volume"
 // +kubebuilder:printcolumn:name="recordsize",type=string,JSONPath=`.spec.recordsize`,description="recordsize of created zfs dataset"
 // +kubebuilder:printcolumn:name="Filesystem",type=string,JSONPath=`.spec.fsType`,description="filesystem created on the volume"
@@ -39,7 +40,8 @@ type ZFSVolume struct {
 	metav1.TypeMeta   `json:",inline"`
 	metav1.ObjectMeta `json:"metadata,omitempty"`
 
-	Spec VolumeInfo `json:"spec"`
+	Spec   VolumeInfo `json:"spec"`
+	Status VolStatus  `json:"status,omitempty"`
 }
 
 // MountInfo contains the volume related info
@@ -198,3 +200,12 @@ type VolumeInfo struct {
 	// Default Value: ext4.
 	FsType string `json:"fsType,omitempty"`
 }
+
+type VolStatus struct {
+	// State specifies the current state of the volume provisioning request.
+	// The state "Pending" means that the volume creation request has not
+	// processed yet. The state "Ready" means that the volume has been created
+	// and it is ready for the use.
+	// +kubebuilder:validation:Enum=Pending;Ready
+	State string `json:"state,omitempty"`
+}

pkg/apis/openebs.io/zfs/v1alpha1/zz_generated.deepcopy.go

@@ -66,6 +66,22 @@ func (in *SnapStatus) DeepCopy() *SnapStatus {
 	return out
 }
 
+// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
+func (in *VolStatus) DeepCopyInto(out *VolStatus) {
+	*out = *in
+	return
+}
+
+// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new VolStatus.
+func (in *VolStatus) DeepCopy() *VolStatus {
+	if in == nil {
+		return nil
+	}
+	out := new(VolStatus)
+	in.DeepCopyInto(out)
+	return out
+}
+
 // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
 func (in *VolumeInfo) DeepCopyInto(out *VolumeInfo) {
 	*out = *in
@@ -149,6 +165,7 @@ func (in *ZFSVolume) DeepCopyInto(out *ZFSVolume) {
 	out.TypeMeta = in.TypeMeta
 	in.ObjectMeta.DeepCopyInto(&out.ObjectMeta)
 	out.Spec = in.Spec
+	out.Status = in.Status
 	return
 }
 

pkg/builder/volbuilder/build.go

@@ -160,6 +160,12 @@ func (b *Builder) WithVolumeType(vtype string) *Builder {
 	return b
 }
 
+// WithVolumeStatus sets ZFSVolume status
+func (b *Builder) WithVolumeStatus(status string) *Builder {
+	b.volume.Object.Status.State = status
+	return b
+}
+
 // WithFsType sets filesystem for the ZFSVolume
 func (b *Builder) WithFsType(fstype string) *Builder {
 	b.volume.Object.Spec.FsType = fstype

pkg/driver/controller.go

@@ -110,6 +110,7 @@ func CreateZFSVolume(req *csi.CreateVolumeRequest) (string, error) {
 		WithThinProv(tp).
 		WithOwnerNode(selected).
 		WithVolumeType(vtype).
+		WithVolumeStatus(zfs.ZFSStatusPending).
 		WithFsType(fstype).
 		WithCompression(compression).Build()
 
@@ -161,7 +162,9 @@ func CreateZFSClone(req *csi.CreateVolumeRequest, snapshot string) (string, erro
 	selected := snap.Spec.OwnerNodeID
 
 	volObj, err := volbuilder.NewBuilder().
-		WithName(volName).Build()
+		WithName(volName).
+		WithVolumeStatus(zfs.ZFSStatusPending).
+		Build()
 
 	volObj.Spec = snap.Spec
 	volObj.Spec.SnapName = snapshot
@@ -187,12 +190,22 @@ func (cs *controller) CreateVolume(
 	volName := req.GetName()
 	pool := req.GetParameters()["poolname"]
 	size := req.GetCapacityRange().RequiredBytes
+	contentSource := req.GetVolumeContentSource()
 
 	if err = cs.validateVolumeCreateReq(req); err != nil {
 		return nil, err
 	}
 
-	contentSource := req.GetVolumeContentSource()
+	selected, state, err := zfs.GetZFSVolumeState(req.Name)
+
+	if err == nil {
+		// ZFSVolume CR has been created, check if it is in Ready state
+		if state == zfs.ZFSStatusReady {
+			goto CreateVolumeResponse
+		}
+		return nil, status.Errorf(codes.Internal, "volume %s creation is Pending", volName)
+	}
+
 	if contentSource != nil && contentSource.GetSnapshot() != nil {
 		snapshotID := contentSource.GetSnapshot().GetSnapshotId()
 
@@ -205,6 +218,18 @@ func (cs *controller) CreateVolume(
 		return nil, status.Error(codes.Internal, err.Error())
 	}
 
+	_, state, err = zfs.GetZFSVolumeState(req.Name)
+
+	if err != nil {
+		return nil, status.Errorf(codes.Internal, "createvolume: failed to fetch the volume %v", err.Error())
+	}
+
+	if state == zfs.ZFSStatusPending {
+		return nil, status.Errorf(codes.Internal, "volume %s is being created", volName)
+	}
+
+CreateVolumeResponse:
+
 	sendEventOrIgnore(volName, strconv.FormatInt(int64(size), 10), "zfs-localpv", analytics.VolumeProvision)
 
 	topology := map[string]string{zfs.ZFSTopologyKey: selected}

pkg/generated/clientset/internalclientset/typed/zfs/v1alpha1/fake/fake_zfsvolume.go

@@ -100,6 +100,18 @@ func (c *FakeZFSVolumes) Update(zFSVolume *v1alpha1.ZFSVolume) (result *v1alpha1
 	return obj.(*v1alpha1.ZFSVolume), err
 }
 
+// UpdateStatus was generated because the type contains a Status member.
+// Add a +genclient:noStatus comment above the type to avoid generating UpdateStatus().
+func (c *FakeZFSVolumes) UpdateStatus(zFSVolume *v1alpha1.ZFSVolume) (*v1alpha1.ZFSVolume, error) {
+	obj, err := c.Fake.
+		Invokes(testing.NewUpdateSubresourceAction(zfsvolumesResource, "status", c.ns, zFSVolume), &v1alpha1.ZFSVolume{})
+
+	if obj == nil {
+		return nil, err
+	}
+	return obj.(*v1alpha1.ZFSVolume), err
+}
+
 // Delete takes name of the zFSVolume and deletes it. Returns an error if one occurs.
 func (c *FakeZFSVolumes) Delete(name string, options *v1.DeleteOptions) error {
 	_, err := c.Fake.

pkg/generated/clientset/internalclientset/typed/zfs/v1alpha1/zfsvolume.go

@@ -39,6 +39,7 @@ type ZFSVolumesGetter interface {
 type ZFSVolumeInterface interface {
 	Create(*v1alpha1.ZFSVolume) (*v1alpha1.ZFSVolume, error)
 	Update(*v1alpha1.ZFSVolume) (*v1alpha1.ZFSVolume, error)
+	UpdateStatus(*v1alpha1.ZFSVolume) (*v1alpha1.ZFSVolume, error)
 	Delete(name string, options *v1.DeleteOptions) error
 	DeleteCollection(options *v1.DeleteOptions, listOptions v1.ListOptions) error
 	Get(name string, options v1.GetOptions) (*v1alpha1.ZFSVolume, error)
@@ -132,6 +133,22 @@ func (c *zFSVolumes) Update(zFSVolume *v1alpha1.ZFSVolume) (result *v1alpha1.ZFS
 	return
 }
 
+// UpdateStatus was generated because the type contains a Status member.
+// Add a +genclient:noStatus comment above the type to avoid generating UpdateStatus().
+
+func (c *zFSVolumes) UpdateStatus(zFSVolume *v1alpha1.ZFSVolume) (result *v1alpha1.ZFSVolume, err error) {
+	result = &v1alpha1.ZFSVolume{}
+	err = c.client.Put().
+		Namespace(c.ns).
+		Resource("zfsvolumes").
+		Name(zFSVolume.Name).
+		SubResource("status").
+		Body(zFSVolume).
+		Do().
+		Into(result)
+	return
+}
+
 // Delete takes name of the zFSVolume and deletes it. Returns an error if one occurs.
 func (c *zFSVolumes) Delete(name string, options *v1.DeleteOptions) error {
 	return c.client.Delete().

pkg/zfs/volume.go

@@ -157,6 +157,21 @@ func GetZFSVolume(volumeID string) (*apis.ZFSVolume, error) {
 	return vol, err
 }
 
+// GetZFSVolumeState returns ZFSVolume OwnerNode and State for
+// the given volume. CreateVolume request may call it again and
+// again until volume is "Ready".
+func GetZFSVolumeState(volID string) (string, string, error) {
+	getOptions := metav1.GetOptions{}
+	vol, err := volbuilder.NewKubeclient().
+		WithNamespace(OpenEBSNamespace).Get(volID, getOptions)
+
+	if err != nil {
+		return "", "", err
+	}
+
+	return vol.Spec.OwnerNodeID, vol.Status.State, nil
+}
+
 // UpdateZvolInfo updates ZFSVolume CR with node id and finalizer
 func UpdateZvolInfo(vol *apis.ZFSVolume) error {
 	finalizers := []string{ZFSFinalizer}
@@ -168,6 +183,7 @@ func UpdateZvolInfo(vol *apis.ZFSVolume) error {
 
 	newVol, err := volbuilder.BuildFrom(vol).
 		WithFinalizer(finalizers).
+		WithVolumeStatus(ZFSStatusReady).
 		WithLabels(labels).Build()
 
 	if err != nil {