pkg/ansible: Changing the reconciliation loop to use re-queue. (#599)

* Adding annotation to override reconciliation time.
* Adding default time as part of controller creation.
* Adding option to specify duration as controller option.

Author: Shawn Hurley

Gopkg.lock

@@ -134,6 +134,7 @@ func upLocalAnsible() {
 	}
 
 	printVersion()
+	logrus.Infof("watching namespace: %s", namespace)
 	done := make(chan error)
 
 	// start the proxy
@@ -144,7 +145,7 @@ func upLocalAnsible() {
 	})
 
 	// start the operator
-	go ansibleOperator.Run(done, mgr, namespace)
+	go ansibleOperator.Run(done, mgr)
 
 	// wait for either to finish
 	err = <-done

commands/operator-sdk/cmd/up/local.go

@@ -35,29 +35,30 @@ import (
 
 // Options - options for your controller
 type Options struct {
-	EventHandlers []events.EventHandler
-	LoggingLevel  events.LogLevel
-	Runner        runner.Runner
-	Namespace     string
-	GVK           schema.GroupVersionKind
-	// StopChannel is used to deal with the bug:
-	// https://github.com/kubernetes-sigs/controller-runtime/issues/103
-	StopChannel <-chan struct{}
+	EventHandlers   []events.EventHandler
+	LoggingLevel    events.LogLevel
+	Runner          runner.Runner
+	GVK             schema.GroupVersionKind
+	ReconcilePeriod time.Duration
 }
 
 // Add - Creates a new ansible operator controller and adds it to the manager
 func Add(mgr manager.Manager, options Options) {
-	logrus.Infof("Watching %s/%v, %s, %s", options.GVK.Group, options.GVK.Version, options.GVK.Kind, options.Namespace)
+	logrus.Infof("Watching %s/%v, %s", options.GVK.Group, options.GVK.Version, options.GVK.Kind)
 	if options.EventHandlers == nil {
 		options.EventHandlers = []events.EventHandler{}
 	}
 	eventHandlers := append(options.EventHandlers, events.NewLoggingEventHandler(options.LoggingLevel))
+	if options.ReconcilePeriod == time.Duration(0) {
+		options.ReconcilePeriod = time.Minute
+	}
 
 	aor := &AnsibleOperatorReconciler{
-		Client:        mgr.GetClient(),
-		GVK:           options.GVK,
-		Runner:        options.Runner,
-		EventHandlers: eventHandlers,
+		Client:          mgr.GetClient(),
+		GVK:             options.GVK,
+		Runner:          options.Runner,
+		EventHandlers:   eventHandlers,
+		ReconcilePeriod: options.ReconcilePeriod,
 	}
 
 	// Register the GVK with the schema
@@ -79,13 +80,4 @@ func Add(mgr manager.Manager, options Options) {
 	if err := c.Watch(&source.Kind{Type: u}, &crthandler.EnqueueRequestForObject{}); err != nil {
 		log.Fatal(err)
 	}
-
-	r := NewReconcileLoop(time.Minute*1, options.GVK, mgr.GetClient(), options.Namespace)
-	r.Stop = options.StopChannel
-	cs := &source.Channel{Source: r.Source}
-	cs.InjectStopChannel(options.StopChannel)
-	if err := c.Watch(cs, &crthandler.EnqueueRequestForObject{}); err != nil {
-		log.Fatal(err)
-	}
-	r.Start()
 }

pkg/ansible/controller/controller.go

@@ -19,6 +19,7 @@ import (
 	"encoding/json"
 	"errors"
 	"os"
+	"time"
 
 	"github.com/operator-framework/operator-sdk/pkg/ansible/events"
 	"github.com/operator-framework/operator-sdk/pkg/ansible/proxy/kubeconfig"
@@ -34,12 +35,20 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 )
 
+const (
+	// ReconcilePeriodAnnotation - annotation used by a user to specify the reconcilation interval for the CR.
+	// To use create a CR with an annotation "ansible.operator-sdk/reconcile-period: 30s" or some other valid
+	// Duration. This will override the operators/or controllers reconcile period for that particular CR.
+	ReconcilePeriodAnnotation = "ansible.operator-sdk/reconcile-period"
+)
+
 // AnsibleOperatorReconciler - object to reconcile runner requests
 type AnsibleOperatorReconciler struct {
-	GVK           schema.GroupVersionKind
-	Runner        runner.Runner
-	Client        client.Client
-	EventHandlers []events.EventHandler
+	GVK             schema.GroupVersionKind
+	Runner          runner.Runner
+	Client          client.Client
+	EventHandlers   []events.EventHandler
+	ReconcilePeriod time.Duration
 }
 
 // Reconcile - handle the event.
@@ -53,6 +62,14 @@ func (r *AnsibleOperatorReconciler) Reconcile(request reconcile.Request) (reconc
 	if err != nil {
 		return reconcile.Result{}, err
 	}
+	reconcileResult := reconcile.Result{RequeueAfter: r.ReconcilePeriod}
+	if ds, ok := u.GetAnnotations()[ReconcilePeriodAnnotation]; ok {
+		duration, err := time.ParseDuration(ds)
+		if err != nil {
+			return reconcileResult, err
+		}
+		reconcileResult.RequeueAfter = duration
+	}
 
 	deleted := u.GetDeletionTimestamp() != nil
 	finalizer, finalizerExists := r.Runner.GetFinalizer()
@@ -63,11 +80,11 @@ func (r *AnsibleOperatorReconciler) Reconcile(request reconcile.Request) (reconc
 		finalizers := append(pendingFinalizers, finalizer)
 		u.SetFinalizers(finalizers)
 		err := r.Client.Update(context.TODO(), u)
-		return reconcile.Result{}, err
+		return reconcileResult, err
 	}
 	if !contains(pendingFinalizers, finalizer) && deleted {
 		logrus.Info("Resource is terminated, skipping reconcilation")
-		return reconcile.Result{}, nil
+		return reconcileResult, nil
 	}
 
 	spec := u.Object["spec"]
@@ -77,9 +94,10 @@ func (r *AnsibleOperatorReconciler) Reconcile(request reconcile.Request) (reconc
 		u.Object["spec"] = map[string]interface{}{}
 		err = r.Client.Update(context.TODO(), u)
 		if err != nil {
-			return reconcile.Result{}, err
+			return reconcileResult, err
 		}
-		return reconcile.Result{Requeue: true}, nil
+		reconcileResult.Requeue = true
+		return reconcileResult, nil
 	}
 	status := u.Object["status"]
 	_, ok = status.(map[string]interface{})
@@ -88,9 +106,10 @@ func (r *AnsibleOperatorReconciler) Reconcile(request reconcile.Request) (reconc
 		u.Object["status"] = map[string]interface{}{}
 		err = r.Client.Update(context.TODO(), u)
 		if err != nil {
-			return reconcile.Result{}, err
+			return reconcileResult, err
 		}
-		return reconcile.Result{Requeue: true}, nil
+		reconcileResult.Requeue = true
+		return reconcileResult, nil
 	}
 
 	// If status is an empty map we can assume CR was just created
@@ -101,9 +120,10 @@ func (r *AnsibleOperatorReconciler) Reconcile(request reconcile.Request) (reconc
 		}
 		err = r.Client.Update(context.TODO(), u)
 		if err != nil {
-			return reconcile.Result{}, err
+			return reconcileResult, err
 		}
-		return reconcile.Result{Requeue: true}, nil
+		reconcileResult.Requeue = true
+		return reconcileResult, nil
 	}
 
 	ownerRef := metav1.OwnerReference{
@@ -115,12 +135,12 @@ func (r *AnsibleOperatorReconciler) Reconcile(request reconcile.Request) (reconc
 
 	kc, err := kubeconfig.Create(ownerRef, "http://localhost:8888", u.GetNamespace())
 	if err != nil {
-		return reconcile.Result{}, err
+		return reconcileResult, err
 	}
 	defer os.Remove(kc.Name())
 	eventChan, err := r.Runner.Run(u, kc.Name())
 	if err != nil {
-		return reconcile.Result{}, err
+		return reconcileResult, err
 	}
 
 	// iterate events from ansible, looking for the final one
@@ -144,7 +164,7 @@ func (r *AnsibleOperatorReconciler) Reconcile(request reconcile.Request) (reconc
 	if statusEvent.Event == "" {
 		err := errors.New("did not receive playbook_on_stats event")
 		logrus.Error(err.Error())
-		return reconcile.Result{}, err
+		return reconcileResult, err
 	}
 
 	// We only want to update the CustomResource once, so we'll track changes and do it at the end
@@ -186,9 +206,10 @@ func (r *AnsibleOperatorReconciler) Reconcile(request reconcile.Request) (reconc
 		err = r.Client.Update(context.TODO(), u)
 	}
 	if !runSuccessful {
-		return reconcile.Result{Requeue: true}, err
+		reconcileResult.Requeue = true
+		return reconcileResult, err
 	}
-	return reconcile.Result{}, err
+	return reconcileResult, err
 }
 
 func contains(l []string, s string) bool {

pkg/ansible/controller/reconcile.go

@@ -30,10 +30,7 @@ import (
 // Run - A blocking function which starts a controller-runtime manager
 // It starts an Operator by reading in the values in `./watches.yaml`, adds a controller
 // to the manager, and finally running the manager.
-func Run(done chan error, mgr manager.Manager, namespace string) {
-	if namespace == "" {
-		namespace = "default"
-	}
+func Run(done chan error, mgr manager.Manager) {
 	watches, err := runner.NewFromWatches("./watches.yaml")
 	if err != nil {
 		logrus.Error("Failed to get watches")
@@ -45,10 +42,8 @@ func Run(done chan error, mgr manager.Manager, namespace string) {
 
 	for gvk, runner := range watches {
 		controller.Add(mgr, controller.Options{
-			GVK:         gvk,
-			Namespace:   namespace,
-			Runner:      runner,
-			StopChannel: c,
+			GVK:    gvk,
+			Runner: runner,
 		})
 	}
 	log.Fatal(mgr.Start(c))

pkg/ansible/controller/source.go

@@ -71,7 +71,6 @@ func main() {
 }
 
 func runSDK(done chan error, mgr manager.Manager) {
-	namespace := "default"
 	watches, err := runner.NewFromWatches("/opt/ansible/watches.yaml")
 	if err != nil {
 		logrus.Error("Failed to get watches")
@@ -83,10 +82,8 @@ func runSDK(done chan error, mgr manager.Manager) {
 
 	for gvk, runner := range watches {
 		controller.Add(mgr, controller.Options{
-			GVK:         gvk,
-			Namespace:   namespace,
-			Runner:      runner,
-			StopChannel: c,
+			GVK:    gvk,
+			Runner: runner,
 		})
 	}
 	log.Fatal(mgr.Start(c))