feat(zfspv): adding backup and restore support (#162)

This commit adds support for Backup and Restore controller, which will be watching for
the events. The velero plugin will create a Backup CR to create a backup
with the remote location information, the controller will send the data
to that remote location.

In the same way, the velero plugin will create a Restore CR to restore the
volume from the the remote location and the restore controller will restore
the data.

Steps to use velero plugin for ZFS-LocalPV are :

1. install velero

2. add openebs plugin

velero plugin add openebs/velero-plugin:latest

3. Create the volumesnapshot location :

for full backup :-

```yaml
apiVersion: velero.io/v1
kind: VolumeSnapshotLocation
metadata:
  name: default
  namespace: velero
spec:
  provider: openebs.io/zfspv-blockstore
  config:
    bucket: velero
    prefix: zfs
    namespace: openebs
    provider: aws
    region: minio
    s3ForcePathStyle: "true"
    s3Url: http://minio.velero.svc:9000
```

for incremental backup :-

```yaml
apiVersion: velero.io/v1
kind: VolumeSnapshotLocation
metadata:
  name: default
  namespace: velero
spec:
  provider: openebs.io/zfspv-blockstore
  config:
    bucket: velero
    prefix: zfs
    backup: incremental
    namespace: openebs
    provider: aws
    region: minio
    s3ForcePathStyle: "true"
    s3Url: http://minio.velero.svc:9000
```

4. Create backup

velero backup create my-backup --snapshot-volumes --include-namespaces=velero-ns --volume-snapshot-locations=aws-cloud-default --storage-location=default

5. Create Schedule

velero create schedule newschedule  --schedule="*/1 * * * *" --snapshot-volumes --include-namespaces=velero-ns --volume-snapshot-locations=aws-local-default --storage-location=default

6. Restore from backup

velero restore create --from-backup my-backup --restore-volumes=true --namespace-mappings velero-ns:ns1



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

pkg/mgmt/backup/backup.go

@@ -0,0 +1,250 @@
+/*
+Copyright 2020 The OpenEBS Authors
+
+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 backup
+
+import (
+	"fmt"
+	"k8s.io/klog"
+	"time"
+
+	apis "github.com/openebs/zfs-localpv/pkg/apis/openebs.io/zfs/v1"
+	zfs "github.com/openebs/zfs-localpv/pkg/zfs"
+	k8serror "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/client-go/tools/cache"
+)
+
+// isDeletionCandidate checks if a zfs backup is a deletion candidate.
+func (c *BkpController) isDeletionCandidate(bkp *apis.ZFSBackup) bool {
+	return bkp.ObjectMeta.DeletionTimestamp != nil
+}
+
+// syncHandler compares the actual state with the desired, and attempts to
+// converge the two.
+func (c *BkpController) syncHandler(key string) error {
+	// Convert the namespace/name string into a distinct namespace and name
+	namespace, name, err := cache.SplitMetaNamespaceKey(key)
+	if err != nil {
+		runtime.HandleError(fmt.Errorf("invalid resource key: %s", key))
+		return nil
+	}
+
+	// Get the bkp resource with this namespace/name
+	bkp, err := c.bkpLister.ZFSBackups(namespace).Get(name)
+	if k8serror.IsNotFound(err) {
+		runtime.HandleError(fmt.Errorf("zfs backup '%s' has been deleted", key))
+		return nil
+	}
+	if err != nil {
+		return err
+	}
+	bkpCopy := bkp.DeepCopy()
+	err = c.syncBkp(bkpCopy)
+	return err
+}
+
+// enqueueBkp takes a ZFSBackup resource and converts it into a namespace/name
+// string which is then put onto the work queue. This method should *not* be
+// passed resources of any type other than ZFSBackup.
+func (c *BkpController) enqueueBkp(obj interface{}) {
+	var key string
+	var err error
+	if key, err = cache.MetaNamespaceKeyFunc(obj); err != nil {
+		runtime.HandleError(err)
+		return
+	}
+	c.workqueue.Add(key)
+}
+
+// synBkp is the function which tries to converge to a desired state for the
+// ZFSBackup
+func (c *BkpController) syncBkp(bkp *apis.ZFSBackup) error {
+	var err error = nil
+	// ZFSBackup should be deleted. Check if deletion timestamp is set
+	if c.isDeletionCandidate(bkp) {
+		// reconcile for the Destroy error
+		err = zfs.DestoryBackup(bkp)
+		if err == nil {
+			err = zfs.RemoveBkpFinalizer(bkp)
+		}
+	} else {
+		// if status is init then it means we are creating the zfs backup.
+		if bkp.Status == apis.BKPZFSStatusInit {
+			err = zfs.CreateBackup(bkp)
+			if err == nil {
+				klog.Infof("backup %s done %s@%s prevsnap [%s]", bkp.Name, bkp.Spec.VolumeName, bkp.Spec.SnapName, bkp.Spec.PrevSnapName)
+				err = zfs.UpdateBkpInfo(bkp, apis.BKPZFSStatusDone)
+			} else {
+				klog.Errorf("backup %s failed %s@%s err %v", bkp.Name, bkp.Spec.VolumeName, bkp.Spec.SnapName, err)
+				err = zfs.UpdateBkpInfo(bkp, apis.BKPZFSStatusFailed)
+			}
+		}
+	}
+	return err
+}
+
+// addBkp is the add event handler for ZFSBackup
+func (c *BkpController) addBkp(obj interface{}) {
+	bkp, ok := obj.(*apis.ZFSBackup)
+	if !ok {
+		runtime.HandleError(fmt.Errorf("Couldn't get backup object %#v", obj))
+		return
+	}
+
+	if zfs.NodeID != bkp.Spec.OwnerNodeID {
+		return
+	}
+	klog.Infof("Got add event for Bkp %s snap %s@%s", bkp.Name, bkp.Spec.VolumeName, bkp.Spec.SnapName)
+	c.enqueueBkp(bkp)
+}
+
+// updateBkp is the update event handler for ZFSBackup
+func (c *BkpController) updateBkp(oldObj, newObj interface{}) {
+
+	newBkp, ok := newObj.(*apis.ZFSBackup)
+	if !ok {
+		runtime.HandleError(fmt.Errorf("Couldn't get bkp object %#v", newBkp))
+		return
+	}
+
+	if zfs.NodeID != newBkp.Spec.OwnerNodeID {
+		return
+	}
+
+	if c.isDeletionCandidate(newBkp) {
+		klog.Infof("Got update event for Bkp %s snap %s@%s", newBkp.Name, newBkp.Spec.VolumeName, newBkp.Spec.SnapName)
+		c.enqueueBkp(newBkp)
+	}
+}
+
+// deleteBkp is the delete event handler for ZFSBackup
+func (c *BkpController) deleteBkp(obj interface{}) {
+	bkp, ok := obj.(*apis.ZFSBackup)
+	if !ok {
+		tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
+		if !ok {
+			runtime.HandleError(fmt.Errorf("Couldn't get object from tombstone %#v", obj))
+			return
+		}
+		bkp, ok = tombstone.Obj.(*apis.ZFSBackup)
+		if !ok {
+			runtime.HandleError(fmt.Errorf("Tombstone contained object that is not a zfsbackup %#v", obj))
+			return
+		}
+	}
+
+	if zfs.NodeID != bkp.Spec.OwnerNodeID {
+		return
+	}
+
+	klog.Infof("Got delete event for Bkp %s snap %s@%s", bkp.Name, bkp.Spec.VolumeName, bkp.Spec.SnapName)
+	c.enqueueBkp(bkp)
+}
+
+// Run will set up the event handlers for types we are interested in, as well
+// as syncing informer caches and starting workers. It will block until stopCh
+// is closed, at which point it will shutdown the workqueue and wait for
+// workers to finish processing their current work items.
+func (c *BkpController) Run(threadiness int, stopCh <-chan struct{}) error {
+	defer runtime.HandleCrash()
+	defer c.workqueue.ShutDown()
+
+	// Start the informer factories to begin populating the informer caches
+	klog.Info("Starting Bkp controller")
+
+	// Wait for the k8s caches to be synced before starting workers
+	klog.Info("Waiting for informer caches to sync")
+	if ok := cache.WaitForCacheSync(stopCh, c.bkpSynced); !ok {
+		return fmt.Errorf("failed to wait for caches to sync")
+	}
+	klog.Info("Starting Bkp workers")
+	// Launch worker to process Bkp resources
+	// Threadiness will decide the number of workers you want to launch to process work items from queue
+	for i := 0; i < threadiness; i++ {
+		go wait.Until(c.runWorker, time.Second, stopCh)
+	}
+
+	klog.Info("Started Bkp workers")
+	<-stopCh
+	klog.Info("Shutting down Bkp workers")
+
+	return nil
+}
+
+// runWorker is a long-running function that will continually call the
+// processNextWorkItem function in order to read and process a message on the
+// workqueue.
+func (c *BkpController) runWorker() {
+	for c.processNextWorkItem() {
+	}
+}
+
+// processNextWorkItem will read a single work item off the workqueue and
+// attempt to process it, by calling the syncHandler.
+func (c *BkpController) processNextWorkItem() bool {
+	obj, shutdown := c.workqueue.Get()
+
+	if shutdown {
+		return false
+	}
+
+	// We wrap this block in a func so we can defer c.workqueue.Done.
+	err := func(obj interface{}) error {
+		// We call Done here so the workqueue knows we have finished
+		// processing this item. We also must remember to call Forget if we
+		// do not want this work item being re-queued. For example, we do
+		// not call Forget if a transient error occurs, instead the item is
+		// put back on the workqueue and attempted again after a back-off
+		// period.
+		defer c.workqueue.Done(obj)
+		var key string
+		var ok bool
+		// We expect strings to come off the workqueue. These are of the
+		// form namespace/name. We do this as the delayed nature of the
+		// workqueue means the items in the informer cache may actually be
+		// more up to date that when the item was initially put onto the
+		// workqueue.
+		if key, ok = obj.(string); !ok {
+			// As the item in the workqueue is actually invalid, we call
+			// Forget here else we'd go into a loop of attempting to
+			// process a work item that is invalid.
+			c.workqueue.Forget(obj)
+			runtime.HandleError(fmt.Errorf("expected string in workqueue but got %#v", obj))
+			return nil
+		}
+		// Run the syncHandler, passing it the namespace/name string of the
+		// Bkp resource to be synced.
+		if err := c.syncHandler(key); err != nil {
+			// Put the item back on the workqueue to handle any transient errors.
+			c.workqueue.AddRateLimited(key)
+			return fmt.Errorf("error syncing '%s': %s, requeuing", key, err.Error())
+		}
+		// Finally, if no error occurs we Forget this item so it does not
+		// get queued again until another change happens.
+		c.workqueue.Forget(obj)
+		klog.Infof("Successfully synced '%s'", key)
+		return nil
+	}(obj)
+
+	if err != nil {
+		runtime.HandleError(err)
+		return true
+	}
+
+	return true
+}

pkg/mgmt/backup/builder.go

@@ -0,0 +1,136 @@
+/*
+Copyright 2020 The OpenEBS Authors
+
+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 backup
+
+import (
+	"k8s.io/klog"
+
+	clientset "github.com/openebs/zfs-localpv/pkg/generated/clientset/internalclientset"
+	openebsScheme "github.com/openebs/zfs-localpv/pkg/generated/clientset/internalclientset/scheme"
+	informers "github.com/openebs/zfs-localpv/pkg/generated/informer/externalversions"
+	listers "github.com/openebs/zfs-localpv/pkg/generated/lister/zfs/v1"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/kubernetes/scheme"
+	typedcorev1 "k8s.io/client-go/kubernetes/typed/core/v1"
+	"k8s.io/client-go/tools/cache"
+	"k8s.io/client-go/tools/record"
+	"k8s.io/client-go/util/workqueue"
+)
+
+const controllerAgentName = "zfsbackup-controller"
+
+// BkpController is the controller implementation for Bkp resources
+type BkpController struct {
+	// kubeclientset is a standard kubernetes clientset
+	kubeclientset kubernetes.Interface
+
+	// clientset is a openebs custom resource package generated for custom API group.
+	clientset clientset.Interface
+
+	bkpLister listers.ZFSBackupLister
+
+	// backupSynced is used for caches sync to get populated
+	bkpSynced cache.InformerSynced
+
+	// workqueue is a rate limited work queue. This is used to queue work to be
+	// processed instead of performing it as soon as a change happens. This
+	// means we can ensure we only process a fixed amount of resources at a
+	// time, and makes it easy to ensure we are never processing the same item
+	// simultaneously in two different workers.
+	workqueue workqueue.RateLimitingInterface
+
+	// recorder is an event recorder for recording Event resources to the
+	// Kubernetes API.
+	recorder record.EventRecorder
+}
+
+// BkpControllerBuilder is the builder object for controller.
+type BkpControllerBuilder struct {
+	BkpController *BkpController
+}
+
+// NewBkpControllerBuilder returns an empty instance of controller builder.
+func NewBkpControllerBuilder() *BkpControllerBuilder {
+	return &BkpControllerBuilder{
+		BkpController: &BkpController{},
+	}
+}
+
+// withKubeClient fills kube client to controller object.
+func (cb *BkpControllerBuilder) withKubeClient(ks kubernetes.Interface) *BkpControllerBuilder {
+	cb.BkpController.kubeclientset = ks
+	return cb
+}
+
+// withOpenEBSClient fills openebs client to controller object.
+func (cb *BkpControllerBuilder) withOpenEBSClient(cs clientset.Interface) *BkpControllerBuilder {
+	cb.BkpController.clientset = cs
+	return cb
+}
+
+// withBkpLister fills bkp lister to controller object.
+func (cb *BkpControllerBuilder) withBkpLister(sl informers.SharedInformerFactory) *BkpControllerBuilder {
+	bkpInformer := sl.Zfs().V1().ZFSBackups()
+	cb.BkpController.bkpLister = bkpInformer.Lister()
+	return cb
+}
+
+// withBkpSynced adds object sync information in cache to controller object.
+func (cb *BkpControllerBuilder) withBkpSynced(sl informers.SharedInformerFactory) *BkpControllerBuilder {
+	bkpInformer := sl.Zfs().V1().ZFSBackups()
+	cb.BkpController.bkpSynced = bkpInformer.Informer().HasSynced
+	return cb
+}
+
+// withWorkqueue adds workqueue to controller object.
+func (cb *BkpControllerBuilder) withWorkqueueRateLimiting() *BkpControllerBuilder {
+	cb.BkpController.workqueue = workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "Bkp")
+	return cb
+}
+
+// withRecorder adds recorder to controller object.
+func (cb *BkpControllerBuilder) withRecorder(ks kubernetes.Interface) *BkpControllerBuilder {
+	klog.Infof("Creating event broadcaster")
+	eventBroadcaster := record.NewBroadcaster()
+	eventBroadcaster.StartLogging(klog.Infof)
+	eventBroadcaster.StartRecordingToSink(&typedcorev1.EventSinkImpl{Interface: ks.CoreV1().Events("")})
+	recorder := eventBroadcaster.NewRecorder(scheme.Scheme, corev1.EventSource{Component: controllerAgentName})
+	cb.BkpController.recorder = recorder
+	return cb
+}
+
+// withEventHandler adds event handlers controller object.
+func (cb *BkpControllerBuilder) withEventHandler(cvcInformerFactory informers.SharedInformerFactory) *BkpControllerBuilder {
+	cvcInformer := cvcInformerFactory.Zfs().V1().ZFSBackups()
+	// Set up an event handler for when Bkp resources change
+	cvcInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
+		AddFunc:    cb.BkpController.addBkp,
+		UpdateFunc: cb.BkpController.updateBkp,
+		DeleteFunc: cb.BkpController.deleteBkp,
+	})
+	return cb
+}
+
+// Build returns a controller instance.
+func (cb *BkpControllerBuilder) Build() (*BkpController, error) {
+	err := openebsScheme.AddToScheme(scheme.Scheme)
+	if err != nil {
+		return nil, err
+	}
+	return cb.BkpController, nil
+}

pkg/mgmt/backup/doc.go

@@ -0,0 +1,47 @@
+/*
+Copyright 2020 The OpenEBS Authors
+
+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.
+*/
+
+/*
+The Backup flow is as follows:
+
+- plugin takes the backup of ZFSVolume CR so that it can be restored.
+
+- It will save the namespace information where the pvc is created also while taking the backup. Plugin will use this info if restoring without a namespace mapping to find if volume has already been restored.
+
+- plugin then creates the ZFSBackup CR with status as Init and with the destination volume and remote location where the data needs to be send.
+
+- Backup controller (on node) keeps a watch for new CRs associated with the node id. This node ID will be same as the Node ID present in the ZFSVolume resource.
+
+- if Backup status == init and not marked for deletion, the Backup controller will take a snapshot which needs to be send for the Backup purpose.
+
+- Backup controller will execute the `zfs send | remote-write` command which will send the data to the Backup server which is a server running by the plugin. The plugin will read the data and send that to remote location S3 or minio.
+
+- If Backup is deleted then corresponsing snapshot also gets deleted.
+
+
+Limitation :-
+
+- there should be enough space in the pool to accomodate the snapshot.
+
+- if there is a network error and backup failed and :
+    * Backup status update also failed, then backup will be retried from the beginning (TODO optimize it)
+    * Backup status update is successful, the Backup operation will fail.
+
+- A snapshot will exist as long as Backup is present and it will be cleaned up when the Backup is deleted.
+
+*/
+
+package backup

pkg/mgmt/backup/start.go

@@ -0,0 +1,107 @@
+/*
+Copyright 2020 The OpenEBS Authors
+
+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 backup
+
+import (
+	"sync"
+
+	"github.com/pkg/errors"
+	"k8s.io/klog"
+
+	"time"
+
+	clientset "github.com/openebs/zfs-localpv/pkg/generated/clientset/internalclientset"
+	informers "github.com/openebs/zfs-localpv/pkg/generated/informer/externalversions"
+	kubeinformers "k8s.io/client-go/informers"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/clientcmd"
+)
+
+var (
+	masterURL  string
+	kubeconfig string
+)
+
+// Start starts the zfsbackup controller.
+func Start(controllerMtx *sync.RWMutex, stopCh <-chan struct{}) error {
+
+	// Get in cluster config
+	cfg, err := getClusterConfig(kubeconfig)
+	if err != nil {
+		return errors.Wrap(err, "error building kubeconfig")
+	}
+
+	// Building Kubernetes Clientset
+	kubeClient, err := kubernetes.NewForConfig(cfg)
+	if err != nil {
+		return errors.Wrap(err, "error building kubernetes clientset")
+	}
+
+	// Building OpenEBS Clientset
+	openebsClient, err := clientset.NewForConfig(cfg)
+	if err != nil {
+		return errors.Wrap(err, "error building openebs clientset")
+	}
+
+	kubeInformerFactory := kubeinformers.NewSharedInformerFactory(kubeClient, time.Second*30)
+	bkpInformerFactory := informers.NewSharedInformerFactory(openebsClient, time.Second*30)
+	// Build() fn of all controllers calls AddToScheme to adds all types of this
+	// clientset into the given scheme.
+	// If multiple controllers happen to call this AddToScheme same time,
+	// it causes panic with error saying concurrent map access.
+	// This lock is used to serialize the AddToScheme call of all controllers.
+	controllerMtx.Lock()
+
+	controller, err := NewBkpControllerBuilder().
+		withKubeClient(kubeClient).
+		withOpenEBSClient(openebsClient).
+		withBkpSynced(bkpInformerFactory).
+		withBkpLister(bkpInformerFactory).
+		withRecorder(kubeClient).
+		withEventHandler(bkpInformerFactory).
+		withWorkqueueRateLimiting().Build()
+
+	// blocking call, can't use defer to release the lock
+	controllerMtx.Unlock()
+
+	if err != nil {
+		return errors.Wrapf(err, "error building controller instance")
+	}
+
+	go kubeInformerFactory.Start(stopCh)
+	go bkpInformerFactory.Start(stopCh)
+
+	// Threadiness defines the number of workers to be launched in Run function
+	return controller.Run(2, stopCh)
+}
+
+// GetClusterConfig return the config for k8s.
+func getClusterConfig(kubeconfig string) (*rest.Config, error) {
+	cfg, err := rest.InClusterConfig()
+	if err != nil {
+		klog.Errorf("Failed to get k8s Incluster config. %+v", err)
+		if kubeconfig == "" {
+			return nil, errors.Wrap(err, "kubeconfig is empty")
+		}
+		cfg, err = clientcmd.BuildConfigFromFlags(masterURL, kubeconfig)
+		if err != nil {
+			return nil, errors.Wrap(err, "error building kubeconfig")
+		}
+	}
+	return cfg, err
+}

pkg/mgmt/restore/builder.go

@@ -0,0 +1,136 @@
+/*
+Copyright 2020 The OpenEBS Authors
+
+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 restore
+
+import (
+	"k8s.io/klog"
+
+	clientset "github.com/openebs/zfs-localpv/pkg/generated/clientset/internalclientset"
+	openebsScheme "github.com/openebs/zfs-localpv/pkg/generated/clientset/internalclientset/scheme"
+	informers "github.com/openebs/zfs-localpv/pkg/generated/informer/externalversions"
+	listers "github.com/openebs/zfs-localpv/pkg/generated/lister/zfs/v1"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/kubernetes/scheme"
+	typedcorev1 "k8s.io/client-go/kubernetes/typed/core/v1"
+	"k8s.io/client-go/tools/cache"
+	"k8s.io/client-go/tools/record"
+	"k8s.io/client-go/util/workqueue"
+)
+
+const controllerAgentName = "zfsrestore-controller"
+
+// RstrController is the controller implementation for Restore resources
+type RstrController struct {
+	// kubeclientset is a standard kubernetes clientset
+	kubeclientset kubernetes.Interface
+
+	// clientset is a openebs custom resource package generated for custom API group.
+	clientset clientset.Interface
+
+	rstrLister listers.ZFSRestoreLister
+
+	// backupSynced is used for caches sync to get populated
+	rstrSynced cache.InformerSynced
+
+	// workqueue is a rate limited work queue. This is used to queue work to be
+	// processed instead of performing it as soon as a change happens. This
+	// means we can ensure we only process a fixed amount of resources at a
+	// time, and makes it easy to ensure we are never processing the same item
+	// simultaneously in two different workers.
+	workqueue workqueue.RateLimitingInterface
+
+	// recorder is an event recorder for recording Event resources to the
+	// Kubernetes API.
+	recorder record.EventRecorder
+}
+
+// RstrControllerBuilder is the builder object for controller.
+type RstrControllerBuilder struct {
+	RstrController *RstrController
+}
+
+// NewRstrControllerBuilder returns an empty instance of controller builder.
+func NewRstrControllerBuilder() *RstrControllerBuilder {
+	return &RstrControllerBuilder{
+		RstrController: &RstrController{},
+	}
+}
+
+// withKubeClient fills kube client to controller object.
+func (cb *RstrControllerBuilder) withKubeClient(ks kubernetes.Interface) *RstrControllerBuilder {
+	cb.RstrController.kubeclientset = ks
+	return cb
+}
+
+// withOpenEBSClient fills openebs client to controller object.
+func (cb *RstrControllerBuilder) withOpenEBSClient(cs clientset.Interface) *RstrControllerBuilder {
+	cb.RstrController.clientset = cs
+	return cb
+}
+
+// withRestoreLister fills rstr lister to controller object.
+func (cb *RstrControllerBuilder) withRestoreLister(sl informers.SharedInformerFactory) *RstrControllerBuilder {
+	rstrInformer := sl.Zfs().V1().ZFSRestores()
+	cb.RstrController.rstrLister = rstrInformer.Lister()
+	return cb
+}
+
+// withRestoreSynced adds object sync information in cache to controller object.
+func (cb *RstrControllerBuilder) withRestoreSynced(sl informers.SharedInformerFactory) *RstrControllerBuilder {
+	rstrInformer := sl.Zfs().V1().ZFSRestores()
+	cb.RstrController.rstrSynced = rstrInformer.Informer().HasSynced
+	return cb
+}
+
+// withWorkqueue adds workqueue to controller object.
+func (cb *RstrControllerBuilder) withWorkqueueRateLimiting() *RstrControllerBuilder {
+	cb.RstrController.workqueue = workqueue.NewNamedRateLimitingQueue(workqueue.DefaultControllerRateLimiter(), "Restore")
+	return cb
+}
+
+// withRecorder adds recorder to controller object.
+func (cb *RstrControllerBuilder) withRecorder(ks kubernetes.Interface) *RstrControllerBuilder {
+	klog.Infof("Creating event broadcaster")
+	eventBroadcaster := record.NewBroadcaster()
+	eventBroadcaster.StartLogging(klog.Infof)
+	eventBroadcaster.StartRecordingToSink(&typedcorev1.EventSinkImpl{Interface: ks.CoreV1().Events("")})
+	recorder := eventBroadcaster.NewRecorder(scheme.Scheme, corev1.EventSource{Component: controllerAgentName})
+	cb.RstrController.recorder = recorder
+	return cb
+}
+
+// withEventHandler adds event handlers controller object.
+func (cb *RstrControllerBuilder) withEventHandler(cvcInformerFactory informers.SharedInformerFactory) *RstrControllerBuilder {
+	cvcInformer := cvcInformerFactory.Zfs().V1().ZFSRestores()
+	// Set up an event handler for when Restore resources change
+	cvcInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
+		AddFunc:    cb.RstrController.addRestore,
+		UpdateFunc: cb.RstrController.updateRestore,
+		DeleteFunc: cb.RstrController.deleteRestore,
+	})
+	return cb
+}
+
+// Build returns a controller instance.
+func (cb *RstrControllerBuilder) Build() (*RstrController, error) {
+	err := openebsScheme.AddToScheme(scheme.Scheme)
+	if err != nil {
+		return nil, err
+	}
+	return cb.RstrController, nil
+}

pkg/mgmt/restore/doc.go

@@ -0,0 +1,49 @@
+/*
+Copyright 2020 The OpenEBS Authors
+
+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.
+*/
+
+/*
+The restore flow is as follows:
+
+- plugin creates a restore storage volume(zvol or dataset)
+
+At the backup time, the plugin backs up the ZFSVolume CR and while doing the restore we have all the information related to that volume. The plugin first creates the restore destination to store the data.
+
+- plugin then creates the ZFSRestore CR with the destination volume and remote location as its server information from where the data will be read for restore purpose.
+
+- restore controller (on node) keeps a watch for new CRs associated with the node id. This node ID will be same as the Node ID present in the ZFSVolume resource.
+
+- if Restore status == init and not marked for deletion, Restore controller will execute the `remote-read | zfs recv` command.
+
+
+Limitation with the Initial Version :-
+
+- The destination cluster should have same node ID and Zpool present.
+
+- If volume was thick provisioned, then destination Zpool should have enough space for that volume.
+
+- destination volume should be present before starting the Restore Operation.
+
+- If the restore fails due to network issues and
+        * the status update succeed, the Restore will not be re-attempted.
+	* the status update fails, the Restore will be re-attempted from the beginning (TODO optimize it).
+
+- If the restore doesn't have the specified backup, the plugin itself fails that restore request as there is no Backup to Restore from.
+
+- If the same volume is restored twice, the data will be written again. The plugin itself fails this kind of request.
+
+*/
+
+package restore

pkg/mgmt/restore/restore.go

@@ -0,0 +1,245 @@
+/*
+Copyright 2020 The OpenEBS Authors
+
+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 restore
+
+import (
+	"fmt"
+	"time"
+
+	"k8s.io/klog"
+
+	apis "github.com/openebs/zfs-localpv/pkg/apis/openebs.io/zfs/v1"
+	zfs "github.com/openebs/zfs-localpv/pkg/zfs"
+	k8serror "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/util/runtime"
+	"k8s.io/apimachinery/pkg/util/wait"
+	"k8s.io/client-go/tools/cache"
+)
+
+// isDeletionCandidate checks if a zfs backup is a deletion candidate.
+func (c *RstrController) isDeletionCandidate(rstr *apis.ZFSRestore) bool {
+	return rstr.ObjectMeta.DeletionTimestamp != nil
+}
+
+// syncHandler compares the actual state with the desired, and attempts to
+// converge the two.
+func (c *RstrController) syncHandler(key string) error {
+	// Convert the namespace/name string into a distinct namespace and name
+	namespace, name, err := cache.SplitMetaNamespaceKey(key)
+	if err != nil {
+		runtime.HandleError(fmt.Errorf("invalid resource key: %s", key))
+		return nil
+	}
+
+	// Get the rstr resource with this namespace/name
+	rstr, err := c.rstrLister.ZFSRestores(namespace).Get(name)
+	if k8serror.IsNotFound(err) {
+		runtime.HandleError(fmt.Errorf("zfs restore '%s' has been deleted", key))
+		return nil
+	}
+	if err != nil {
+		return err
+	}
+	rstrCopy := rstr.DeepCopy()
+	err = c.syncRestore(rstrCopy)
+	return err
+}
+
+// enqueueRestore takes a ZFSRestore resource and converts it into a namespace/name
+// string which is then put onto the work queue. This method should *not* be
+// passed resources of any type other than ZFSRestore.
+func (c *RstrController) enqueueRestore(obj interface{}) {
+	var key string
+	var err error
+	if key, err = cache.MetaNamespaceKeyFunc(obj); err != nil {
+		runtime.HandleError(err)
+		return
+	}
+	c.workqueue.Add(key)
+}
+
+// synRestore is the function which tries to converge to a desired state for the
+// ZFSRestore
+func (c *RstrController) syncRestore(rstr *apis.ZFSRestore) error {
+	var err error = nil
+	// ZFSRestore should not be deleted. Check if deletion timestamp is set
+	if !c.isDeletionCandidate(rstr) {
+		// if status is Init, then only do the restore
+		if rstr.Status == apis.RSTZFSStatusInit {
+			err = zfs.CreateRestore(rstr)
+			if err == nil {
+				klog.Infof("restore %s done %s", rstr.Name, rstr.Spec.VolumeName)
+				err = zfs.UpdateRestoreInfo(rstr, apis.RSTZFSStatusDone)
+			} else {
+				klog.Errorf("restore %s failed %s err %v", rstr.Name, rstr.Spec.VolumeName, err)
+				err = zfs.UpdateRestoreInfo(rstr, apis.RSTZFSStatusFailed)
+			}
+		}
+	}
+	return err
+}
+
+// addRestore is the add event handler for ZFSRestore
+func (c *RstrController) addRestore(obj interface{}) {
+	rstr, ok := obj.(*apis.ZFSRestore)
+	if !ok {
+		runtime.HandleError(fmt.Errorf("Couldn't get rstr object %#v", obj))
+		return
+	}
+
+	if zfs.NodeID != rstr.Spec.OwnerNodeID {
+		return
+	}
+	klog.Infof("Got add event for Restore %s vol %s", rstr.Name, rstr.Spec.VolumeName)
+	c.enqueueRestore(rstr)
+}
+
+// updateRestore is the update event handler for ZFSRestore
+func (c *RstrController) updateRestore(oldObj, newObj interface{}) {
+
+	newRstr, ok := newObj.(*apis.ZFSRestore)
+	if !ok {
+		runtime.HandleError(fmt.Errorf("Couldn't get rstr object %#v", newRstr))
+		return
+	}
+
+	if zfs.NodeID != newRstr.Spec.OwnerNodeID {
+		return
+	}
+
+	if c.isDeletionCandidate(newRstr) {
+		klog.Infof("Got update event for Restore %s vol %s", newRstr.Name, newRstr.Spec.VolumeName)
+		c.enqueueRestore(newRstr)
+	}
+}
+
+// deleteRestore is the delete event handler for ZFSRestore
+func (c *RstrController) deleteRestore(obj interface{}) {
+	rstr, ok := obj.(*apis.ZFSRestore)
+	if !ok {
+		tombstone, ok := obj.(cache.DeletedFinalStateUnknown)
+		if !ok {
+			runtime.HandleError(fmt.Errorf("Couldn't get object from tombstone %#v", obj))
+			return
+		}
+		rstr, ok = tombstone.Obj.(*apis.ZFSRestore)
+		if !ok {
+			runtime.HandleError(fmt.Errorf("Tombstone contained object that is not a zfsbackup %#v", obj))
+			return
+		}
+	}
+
+	if zfs.NodeID != rstr.Spec.OwnerNodeID {
+		return
+	}
+
+	klog.Infof("Got delete event for Restore %s", rstr.Spec.VolumeName)
+	c.enqueueRestore(rstr)
+}
+
+// Run will set up the event handlers for types we are interested in, as well
+// as syncing informer caches and starting workers. It will block until stopCh
+// is closed, at which point it will shutdown the workqueue and wait for
+// workers to finish processing their current work items.
+func (c *RstrController) Run(threadiness int, stopCh <-chan struct{}) error {
+	defer runtime.HandleCrash()
+	defer c.workqueue.ShutDown()
+
+	// Start the informer factories to begin populating the informer caches
+	klog.Info("Starting Restore controller")
+
+	// Wait for the k8s caches to be synced before starting workers
+	klog.Info("Waiting for informer caches to sync")
+	if ok := cache.WaitForCacheSync(stopCh, c.rstrSynced); !ok {
+		return fmt.Errorf("failed to wait for caches to sync")
+	}
+	klog.Info("Starting Restore workers")
+	// Launch worker to process Restore resources
+	// Threadiness will decide the number of workers you want to launch to process work items from queue
+	for i := 0; i < threadiness; i++ {
+		go wait.Until(c.runWorker, time.Second, stopCh)
+	}
+
+	klog.Info("Started Restore workers")
+	<-stopCh
+	klog.Info("Shutting down Restore workers")
+
+	return nil
+}
+
+// runWorker is a long-running function that will continually call the
+// processNextWorkItem function in order to read and process a message on the
+// workqueue.
+func (c *RstrController) runWorker() {
+	for c.processNextWorkItem() {
+	}
+}
+
+// processNextWorkItem will read a single work item off the workqueue and
+// attempt to process it, by calling the syncHandler.
+func (c *RstrController) processNextWorkItem() bool {
+	obj, shutdown := c.workqueue.Get()
+
+	if shutdown {
+		return false
+	}
+
+	// We wrap this block in a func so we can defer c.workqueue.Done.
+	err := func(obj interface{}) error {
+		// We call Done here so the workqueue knows we have finished
+		// processing this item. We also must remember to call Forget if we
+		// do not want this work item being re-queued. For example, we do
+		// not call Forget if a transient error occurs, instead the item is
+		// put back on the workqueue and attempted again after a back-off
+		// period.
+		defer c.workqueue.Done(obj)
+		var key string
+		var ok bool
+		// We expect strings to come off the workqueue. These are of the
+		// form namespace/name. We do this as the delayed nature of the
+		// workqueue means the items in the informer cache may actually be
+		// more up to date that when the item was initially put onto the
+		// workqueue.
+		if key, ok = obj.(string); !ok {
+			// As the item in the workqueue is actually invalid, we call
+			// Forget here else we'd go into a loop of attempting to
+			// process a work item that is invalid.
+			c.workqueue.Forget(obj)
+			runtime.HandleError(fmt.Errorf("expected string in workqueue but got %#v", obj))
+			return nil
+		}
+		// Run the syncHandler, passing it the namespace/name string of the
+		// Restore resource to be synced.
+		if err := c.syncHandler(key); err != nil {
+			// Put the item back on the workqueue to handle any transient errors.
+			c.workqueue.AddRateLimited(key)
+			return fmt.Errorf("error syncing '%s': %s, requeuing", key, err.Error())
+		}
+		// Finally, if no error occurs we Forget this item so it does not
+		// get queued again until another change happens.
+		c.workqueue.Forget(obj)
+		klog.Infof("Successfully synced '%s'", key)
+		return nil
+	}(obj)
+
+	if err != nil {
+		runtime.HandleError(err)
+		return true
+	}
+
+	return true
+}

pkg/mgmt/restore/start.go

@@ -0,0 +1,107 @@
+/*
+Copyright 2020 The OpenEBS Authors
+
+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 restore
+
+import (
+	"sync"
+
+	"github.com/pkg/errors"
+	"k8s.io/klog"
+
+	"time"
+
+	clientset "github.com/openebs/zfs-localpv/pkg/generated/clientset/internalclientset"
+	informers "github.com/openebs/zfs-localpv/pkg/generated/informer/externalversions"
+	kubeinformers "k8s.io/client-go/informers"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/clientcmd"
+)
+
+var (
+	masterURL  string
+	kubeconfig string
+)
+
+// Start starts the zfsrestore controller.
+func Start(controllerMtx *sync.RWMutex, stopCh <-chan struct{}) error {
+
+	// Get in cluster config
+	cfg, err := getClusterConfig(kubeconfig)
+	if err != nil {
+		return errors.Wrap(err, "error building kubeconfig")
+	}
+
+	// Building Kubernetes Clientset
+	kubeClient, err := kubernetes.NewForConfig(cfg)
+	if err != nil {
+		return errors.Wrap(err, "error building kubernetes clientset")
+	}
+
+	// Building OpenEBS Clientset
+	openebsClient, err := clientset.NewForConfig(cfg)
+	if err != nil {
+		return errors.Wrap(err, "error building openebs clientset")
+	}
+
+	kubeInformerFactory := kubeinformers.NewSharedInformerFactory(kubeClient, time.Second*30)
+	bkpInformerFactory := informers.NewSharedInformerFactory(openebsClient, time.Second*30)
+	// Build() fn of all controllers calls AddToScheme to adds all types of this
+	// clientset into the given scheme.
+	// If multiple controllers happen to call this AddToScheme same time,
+	// it causes panic with error saying concurrent map access.
+	// This lock is used to serialize the AddToScheme call of all controllers.
+	controllerMtx.Lock()
+
+	controller, err := NewRstrControllerBuilder().
+		withKubeClient(kubeClient).
+		withOpenEBSClient(openebsClient).
+		withRestoreSynced(bkpInformerFactory).
+		withRestoreLister(bkpInformerFactory).
+		withRecorder(kubeClient).
+		withEventHandler(bkpInformerFactory).
+		withWorkqueueRateLimiting().Build()
+
+	// blocking call, can't use defer to release the lock
+	controllerMtx.Unlock()
+
+	if err != nil {
+		return errors.Wrapf(err, "error building controller instance")
+	}
+
+	go kubeInformerFactory.Start(stopCh)
+	go bkpInformerFactory.Start(stopCh)
+
+	// Threadiness defines the number of workers to be launched in Run function
+	return controller.Run(2, stopCh)
+}
+
+// GetClusterConfig return the config for k8s.
+func getClusterConfig(kubeconfig string) (*rest.Config, error) {
+	cfg, err := rest.InClusterConfig()
+	if err != nil {
+		klog.Errorf("Failed to get k8s Incluster config. %+v", err)
+		if kubeconfig == "" {
+			return nil, errors.Wrap(err, "kubeconfig is empty")
+		}
+		cfg, err = clientcmd.BuildConfigFromFlags(masterURL, kubeconfig)
+		if err != nil {
+			return nil, errors.Wrap(err, "error building kubeconfig")
+		}
+	}
+	return cfg, err
+}