Added Daemonset Status predicate

Added Daemonset Status predicate to optimize watch on DS.
Minor fixes on status update for the controllers by coalescing status
and CR update together.

Issue-ID: ONAPARC-461
Signed-off-by: Dileep Ranganathan <dileep.ranganathan@intel.com>
Change-Id: I2a56f0b93c2d7a56b9e8149c41f8c6f22be86ef1

1 files changed, 57 insertions, 14 deletions

diff --git a/vnfs/DAaaS/microservices/collectd-operator/pkg/controller/collectdplugin/collectdplugin_controller.go b/vnfs/DAaaS/microservices/collectd-operator/pkg/controller/collectdplugin/collectdplugin_controller.go
index 32775c5c..e0e62b87 100644
--- a/vnfs/DAaaS/microservices/collectd-operator/pkg/controller/collectdplugin/collectdplugin_controller.go
+++ b/vnfs/DAaaS/microservices/collectd-operator/pkg/controller/collectdplugin/collectdplugin_controller.go
@@ -4,17 +4,20 @@ import (
 	"context"
 	"fmt"
 	"reflect"
+	"strings"
 
 	"github.com/go-logr/logr"
 	"github.com/operator-framework/operator-sdk/pkg/predicate"
 
 	onapv1alpha1 "collectd-operator/pkg/apis/onap/v1alpha1"
 	collectdutils "collectd-operator/pkg/controller/utils"
+	dspredicate "collectd-operator/pkg/controller/utils"
 
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/client-go/util/retry"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller"
@@ -54,6 +57,35 @@ func add(mgr manager.Manager, r reconcile.Reconciler) error {
 		return err
 	}
 
+	log.V(1).Info("Add watcher for secondary resource Collectd Daemonset")
+	err = c.Watch(
+		&source.Kind{Type: &appsv1.DaemonSet{}},
+		&handler.EnqueueRequestsFromMapFunc{
+			ToRequests: handler.ToRequestsFunc(func(a handler.MapObject) []reconcile.Request {
+				labelSelector, err := collectdutils.GetWatchLabels()
+				labels := strings.Split(labelSelector, "=")
+				if err != nil {
+					log.Error(err, "Failed to get watch labels, continuing with default label")
+				}
+				rcp := r.(*ReconcileCollectdPlugin)
+				// Select the Daemonset with labelSelector (Default  is app=collectd)
+				if a.Meta.GetLabels()[labels[0]] == labels[1] {
+					var requests []reconcile.Request
+					cpList, err := collectdutils.GetCollectdPluginList(rcp.client, a.Meta.GetNamespace())
+					if err != nil || cpList == nil || cpList.Items == nil || len(cpList.Items) == 0 {
+						log.V(1).Info("No CollectdPlugin CR instance Exist")
+						return nil
+					}
+					for _, cp := range cpList.Items {
+						requests = append(requests, reconcile.Request{
+							NamespacedName: client.ObjectKey{Namespace: cp.Namespace, Name: cp.Name}})
+					}
+					return requests
+				}
+				return nil
+			}),
+		}, dspredicate.DaemonSetStatusChangedPredicate{})
+
 	return nil
 }
 
@@ -178,10 +210,13 @@ func (r *ReconcileCollectdPlugin) handleCollectdPlugin(reqLogger logr.Logger, cr
 		panic(fmt.Errorf("Update failed: %v", retryErr))
 	}
 
-	err := r.updateStatus(cr)
-	if err != nil {
-		reqLogger.Error(err, "Unable to update status")
+	retryErr = retry.RetryOnConflict(retry.DefaultRetry, func() error {
+		err := r.updateStatus(cr)
 		return err
+	})
+
+	if retryErr != nil {
+		panic(fmt.Errorf("Update failed: %v", retryErr))
 	}
 	// Reconcile success
 	reqLogger.Info("Reconcile success!!")
@@ -220,22 +255,29 @@ func (r *ReconcileCollectdPlugin) handleDelete(reqLogger logr.Logger, cr *onapv1
 }
 
 func (r *ReconcileCollectdPlugin) updateStatus(cr *onapv1alpha1.CollectdPlugin) error {
-	switch cr.Status.Status {
+	// Fetch the CollectdGlobal instance
+	instance := &onapv1alpha1.CollectdPlugin{}
+	key := types.NamespacedName{Namespace: cr.Namespace, Name: cr.Name}
+	err := r.client.Get(context.TODO(), key, instance)
+	if err != nil {
+		return err
+	}
+	switch instance.Status.Status {
 	case onapv1alpha1.Initial:
-		cr.Status.Status = onapv1alpha1.Created
+		instance.Status.Status = onapv1alpha1.Created
 	case onapv1alpha1.Created, onapv1alpha1.Enabled:
-		pods, err := collectdutils.GetPodList(r.client, cr.Namespace)
+		pods, err := collectdutils.GetPodList(r.client, instance.Namespace)
 		if err != nil {
 			return err
 		}
-		if !reflect.DeepEqual(pods, cr.Status.CollectdAgents) {
-			cr.Status.CollectdAgents = pods
-			cr.Status.Status = onapv1alpha1.Enabled
+		if !reflect.DeepEqual(pods, instance.Status.CollectdAgents) {
+			instance.Status.CollectdAgents = pods
+			instance.Status.Status = onapv1alpha1.Enabled
 		}
 	case onapv1alpha1.Deleting, onapv1alpha1.Deprecated:
 		return nil
 	}
-	err := r.client.Status().Update(context.TODO(), cr)
+	err = r.client.Status().Update(context.TODO(), instance)
 	return err
 }
 
@@ -255,11 +297,12 @@ func (r *ReconcileCollectdPlugin) addFinalizer(reqLogger logr.Logger, cr *onapv1
 	// Update status from Initial to Created
 	// Since addFinalizer will be executed only once,
 	// the status will be changed from Initial state to Created
-	updateErr := r.updateStatus(cr)
-	if updateErr != nil {
-		reqLogger.Error(updateErr, "Failed to update status from Initial state")
-	}
+	// updateErr := r.updateStatus(cr)
+	// if updateErr != nil {
+	// 	reqLogger.Error(updateErr, "Failed to update status from Initial state")
+	// }
 	// Update CR
+	cr.Status.Status = onapv1alpha1.Created
 	err := r.client.Update(context.TODO(), cr)
 	if err != nil {
 		reqLogger.Error(err, "Failed to update CollectdPlugin with finalizer")