Add kubestash controller for changing kubeDB phase (#1096)

Signed-off-by: Anisur Rahman <anisur@appscode.com>

pkg/controller/initializer/stash/controller.go → pkg/controller/initializer/restore/controller.go

@@ -14,9 +14,10 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package stash
+package restore
 
 import (
+	"fmt"
 	"time"
 
 	amc "kubedb.dev/apimachinery/pkg/controller"
@@ -26,25 +27,31 @@ import (
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/rest"
 	"k8s.io/klog/v2"
+	kmapi "kmodules.xyz/client-go/api/v1"
 	dmcond "kmodules.xyz/client-go/dynamic/conditions"
 	"kmodules.xyz/client-go/tools/queue"
+	coreapi "kubestash.dev/apimachinery/apis/core/v1alpha1"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
 	"stash.appscode.dev/apimachinery/apis/stash/v1beta1"
 	scs "stash.appscode.dev/apimachinery/client/clientset/versioned"
 	stashinformer "stash.appscode.dev/apimachinery/client/informers/externalversions"
 )
 
 type Controller struct {
+	manager *manager.Manager
 	*amc.Controller
 	*amc.StashInitializer
 	restrictToNamespace string
 }
 
 func NewController(
+	mgr *manager.Manager,
 	ctrl *amc.Controller,
 	initializer *amc.StashInitializer,
 	restrictToNamespace string,
 ) *Controller {
 	return &Controller{
+		manager:             mgr,
 		Controller:          ctrl,
 		StashInitializer:    initializer,
 		restrictToNamespace: restrictToNamespace,
@@ -53,12 +60,22 @@ func NewController(
 
 type restoreInfo struct {
 	invoker    core.TypedLocalObjectReference
-	target     *v1beta1.RestoreTarget
-	phase      v1beta1.RestorePhase
+	stash      *stashInfo
+	kubestash  *kubestashInfo
 	do         dmcond.DynamicOptions
 	invokerUID types.UID
 }
 
+type stashInfo struct {
+	target *v1beta1.RestoreTarget
+	phase  v1beta1.RestorePhase
+}
+
+type kubestashInfo struct {
+	target *kmapi.TypedObjectReference
+	phase  coreapi.RestorePhase
+}
+
 func Configure(cfg *rest.Config, s *amc.StashInitializer, resyncPeriod time.Duration) error {
 	var err error
 	if s.StashClient, err = scs.NewForConfig(cfg); err != nil {
@@ -68,7 +85,7 @@ func Configure(cfg *rest.Config, s *amc.StashInitializer, resyncPeriod time.Dura
 	return nil
 }
 
-func (c *Controller) StartAfterStashInstalled(maxNumRequeues, numThreads int, selector metav1.LabelSelector, stopCh <-chan struct{}) {
+func (c *Controller) StartAfterStashInstalled(stopCh <-chan struct{}, maxNumRequeues, numThreads int, selector metav1.LabelSelector) {
 	// Wait until Stash operator installed
 	if err := c.waitUntilStashInstalled(stopCh); err != nil {
 		klog.Errorln("error during waiting for RestoreSession crd. Reason: ", err)
@@ -125,3 +142,17 @@ func (c *Controller) startController(stopCh <-chan struct{}) {
 	c.RSQueue.Run(stopCh)
 	c.RBQueue.Run(stopCh)
 }
+
+func (c *Controller) StartAfterKubeStashInstalled(stopCh <-chan struct{}, selector metav1.LabelSelector) {
+	// Here Wait until KubeStash operator installed
+	if err := c.waitUntilKubeStashInstalled(stopCh); err != nil {
+		klog.Errorln("error during waiting for RestoreSession crd. Reason: ", err)
+		return
+	}
+	if err := (&RestoreSessionReconciler{
+		ctrl: c,
+	}).SetupWithManager(*c.manager, selector); err != nil {
+		klog.Info(fmt.Errorf("unable to create RestoreSession controller. Reason: %w", err))
+		return
+	}
+}

pkg/controller/initializer/restore/kubestash-restoresession.go

@@ -0,0 +1,77 @@
+/*
+Copyright AppsCode Inc. and Contributors
+
+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 (
+	"context"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/klog/v2"
+	coreapi "kubestash.dev/apimachinery/apis/core/v1alpha1"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/builder"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/log"
+	"sigs.k8s.io/controller-runtime/pkg/predicate"
+)
+
+// RestoreSessionReconciler reconciles a RestoreSession object
+type RestoreSessionReconciler struct {
+	ctrl *Controller
+}
+
+func (r *RestoreSessionReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
+	logger := log.FromContext(ctx)
+	logger.Info("Reconciling: " + req.String())
+	c := r.ctrl.KBClient
+
+	rs := &coreapi.RestoreSession{}
+	if err := c.Get(ctx, req.NamespacedName, rs); err != nil {
+		return ctrl.Result{}, client.IgnoreNotFound(err)
+	}
+
+	ri, err := r.ctrl.extractRestoreInfo(rs)
+	if err != nil {
+		klog.Errorln("failed to extract kubeStash invoker info. Reason: ", err)
+		return ctrl.Result{}, err
+	}
+	if rs.DeletionTimestamp != nil {
+		return ctrl.Result{}, r.ctrl.handleTerminateEvent(ri)
+	}
+
+	return ctrl.Result{}, r.ctrl.handleRestoreInvokerEvent(ri)
+}
+
+// SetupWithManager sets up the controller with the Manager.
+func (r *RestoreSessionReconciler) SetupWithManager(mgr ctrl.Manager, selector metav1.LabelSelector) error {
+	return ctrl.NewControllerManagedBy(mgr).
+		For(&coreapi.RestoreSession{}, builder.WithPredicates(
+			predicate.NewPredicateFuncs(func(object client.Object) bool {
+				return hasRequiredLabels(object.GetLabels(), selector.MatchLabels)
+			}),
+		)).
+		Complete(r)
+}
+
+func hasRequiredLabels(actualLabels, requiredLabels map[string]string) bool {
+	for key, value := range requiredLabels {
+		if actualValue, found := actualLabels[key]; !found || actualValue != value {
+			return false
+		}
+	}
+	return true
+}

pkg/controller/initializer/stash/restorebatch.go → pkg/controller/initializer/restore/stash-restorebatch.go

@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package stash
+package restore
 
 import (
 	"time"

pkg/controller/initializer/stash/restoresession.go → pkg/controller/initializer/restore/stash-restoresession.go

@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package stash
+package restore
 
 import (
 	"time"