X-Git-Url: https://gerrit.onap.org/r/gitweb?a=blobdiff_plain;f=vnfs%2FDAaaS%2Fmicroservices%2Fcollectd-operator%2Fpkg%2Fcontroller%2Fcollectdplugin%2Fcollectdplugin_controller.go;h=98d770a1fbfe161e39548806b4edac979a0e9c5f;hb=ebc40b61c64d5f06b5fe2b4d886ed4e5748db6d3;hp=a7990a0d6defc37ab70a30111aedea2860280d0d;hpb=83714f905f0483a84b8af8132a14ea1464bd0355;p=demo.git

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 a7990a0d..98d770a1 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
@@ -2,16 +2,22 @@ package collectdplugin
 
 import (
 	"context"
-	"crypto/sha256"
 	"fmt"
-	"time"
+	"reflect"
+	"strings"
 
-	onapv1alpha1 "demo/vnfs/DAaaS/microservices/collectd-operator/pkg/apis/onap/v1alpha1"
+	"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"
+
+	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	extensionsv1beta1 "k8s.io/api/extensions/v1beta1"
 	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/util/retry"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
@@ -23,18 +29,6 @@ import (
 
 var log = logf.Log.WithName("controller_collectdplugin")
 
-// ResourceMap to hold objects to update/reload
-type ResourceMap struct {
-	configMap       *corev1.ConfigMap
-	daemonSet       *extensionsv1beta1.DaemonSet
-	collectdPlugins *[]onapv1alpha1.CollectdPlugin
-}
-
-/**
-* USER ACTION REQUIRED: This is a scaffold file intended for the user to modify with their own Controller
-* business logic.  Delete these comments after modifying this file.*
- */
-
 // Add creates a new CollectdPlugin Controller and adds it to the Manager. The Manager will set fields on the Controller
 // and Start it when the Manager is Started.
 func Add(mgr manager.Manager) error {
@@ -57,7 +51,38 @@ func add(mgr manager.Manager, r reconcile.Reconciler) error {
 
 	// Watch for changes to primary resource CollectdPlugin
 	log.V(1).Info("Add watcher for primary resource CollectdPlugin")
-	err = c.Watch(&source.Kind{Type: &onapv1alpha1.CollectdPlugin{}}, &handler.EnqueueRequestForObject{})
+	err = c.Watch(&source.Kind{Type: &onapv1alpha1.CollectdPlugin{}}, &handler.EnqueueRequestForObject{}, predicate.GenerationChangedPredicate{})
+	if err != nil {
+		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 (Defautl  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 {
+						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
+			}),
+		})
 	if err != nil {
 		return err
 	}
@@ -103,124 +128,179 @@ func (r *ReconcileCollectdPlugin) Reconcile(request reconcile.Request) (reconcil
 		return reconcile.Result{}, err
 	}
 
-	rmap, err := findResourceMapForCR(r, instance)
-	if err != nil {
-		reqLogger.Info("Skip reconcile: ConfigMap not found")
+	// Handle Delete CR for additional cleanup
+	isDelete, err := r.handleDelete(reqLogger, instance)
+	if isDelete {
 		return reconcile.Result{}, err
 	}
 
-	cm := rmap.configMap
-	ds := rmap.daemonSet
-	collectPlugins := rmap.collectdPlugins
-	reqLogger.V(1).Info("Found ResourceMap")
-	reqLogger.V(1).Info("ConfigMap.Namespace", cm.Namespace, "ConfigMap.Name", cm.Name)
-	reqLogger.V(1).Info("DaemonSet.Namespace", ds.Namespace, "DaemonSet.Name", ds.Name)
+	// Add finalizer for this CR
+	if !collectdutils.Contains(instance.GetFinalizers(), collectdutils.CollectdFinalizer) {
+		if err := r.addFinalizer(reqLogger, instance); err != nil {
+			return reconcile.Result{}, err
+		}
+		return reconcile.Result{}, nil
+	}
+	// Handle the reconciliation for CollectdPlugin.
+	// At this stage the Status of the CollectdPlugin should NOT be ""
+	err = r.handleCollectdPlugin(reqLogger, instance, false)
+	return reconcile.Result{}, err
+}
 
-	collectdConf, err := rebuildCollectdConf(collectPlugins)
+// handleCollectdPlugin regenerates the collectd conf on CR Create, Update, Delete events
+func (r *ReconcileCollectdPlugin) handleCollectdPlugin(reqLogger logr.Logger, cr *onapv1alpha1.CollectdPlugin, isDelete bool) error {
 
-	//Restart Collectd Pods
+	rmap, err := collectdutils.FindResourceMapForCR(r.client, reqLogger, cr.Namespace)
+	if err != nil {
+		reqLogger.Error(err, "Skip reconcile: Resources not found")
+		return err
+	}
+
+	cm := rmap.ConfigMap
+	reqLogger.V(1).Info("Found ResourceMap")
+	reqLogger.V(1).Info(":::: ConfigMap Info ::::", "ConfigMap.Namespace", cm.Namespace, "ConfigMap.Name", cm.Name)
+
+	collectdConf, err := collectdutils.RebuildCollectdConf(r.client, cr.Namespace, isDelete, cr.Spec.PluginName)
+	if err != nil {
+		reqLogger.Error(err, "Skip reconcile: Rebuild conf failed")
+		return err
+	}
 
-	ts := time.Now().Format(time.RFC850)
-	reqLogger.V(1).Info("Timestamp : ", ts)
-	ds.Spec.Template.SetAnnotations(map[string]string{
-		"daaas-random": ComputeSHA256([]byte(ts)),
-	})
 	cm.SetAnnotations(map[string]string{
-		"daaas-random": ComputeSHA256([]byte(ts)),
+		"daaas-random": collectdutils.ComputeSHA256([]byte(collectdConf)),
 	})
 
-	cm.Data["node-collectd.conf"] = collectdConf
+	cm.Data["collectd.conf"] = collectdConf
 
 	// Update the ConfigMap with new Spec and reload DaemonSets
 	reqLogger.Info("Updating the ConfigMap", "ConfigMap.Namespace", cm.Namespace, "ConfigMap.Name", cm.Name)
-	log.Info("ConfigMap Data", "Map: ", cm.Data)
+	log.V(1).Info("ConfigMap Data", "Map: ", cm.Data)
 	err = r.client.Update(context.TODO(), cm)
 	if err != nil {
-		return reconcile.Result{}, err
-	}
-
-	err = r.client.Update(context.TODO(), ds)
-	if err != nil {
-		return reconcile.Result{}, err
+		reqLogger.Error(err, "Update the ConfigMap failed", "ConfigMap.Namespace", cm.Namespace, "ConfigMap.Name", cm.Name)
+		return err
 	}
-	// Reconcile success
-	reqLogger.Info("Updated the ConfigMap", "ConfigMap.Namespace", cm.Namespace, "ConfigMap.Name", cm.Name)
-	return reconcile.Result{}, nil
-}
 
-// ComputeSHA256  returns hash of data as string
-func ComputeSHA256(data []byte) string {
-	hash := sha256.Sum256(data)
-	return fmt.Sprintf("%x", hash)
-}
-
-// findConfigMapForCR returns the configMap used by collectd Daemonset
-func findResourceMapForCR(r *ReconcileCollectdPlugin, cr *onapv1alpha1.CollectdPlugin) (ResourceMap, error) {
-	cmList := &corev1.ConfigMapList{}
-	opts := &client.ListOptions{}
-	rmap := ResourceMap{}
-
-	// Select ConfigMaps with label app=collectd
-	opts.SetLabelSelector("app=collectd")
-	opts.InNamespace(cr.Namespace)
-	err := r.client.List(context.TODO(), opts, cmList)
-	if err != nil {
-		return rmap, err
-	}
+	retryErr := retry.RetryOnConflict(retry.DefaultRetry, func() error {
+		// Retrieve the latest version of Daemonset before attempting update
+		// RetryOnConflict uses exponential backoff to avoid exhausting the apiserver
+		// Select DaemonSets with label
+		dsList := &extensionsv1beta1.DaemonSetList{}
+		opts := &client.ListOptions{}
+		labelSelector, err := collectdutils.GetWatchLabels()
+		if err != nil {
+			reqLogger.Error(err, "Failed to get watch labels, continuing with default label")
+		}
+		opts.SetLabelSelector(labelSelector)
+		opts.InNamespace(cr.Namespace)
+		err = r.client.List(context.TODO(), opts, dsList)
+		if err != nil {
+			panic(fmt.Errorf("Failed to get latest version of DaemonSet: %v", err))
+		}
 
-	if cmList.Items == nil || len(cmList.Items) == 0 {
-		return rmap, err
+		if dsList.Items == nil || len(dsList.Items) == 0 {
+			return errors.NewNotFound(corev1.Resource("daemonset"), "DaemonSet")
+		}
+		ds := &dsList.Items[0]
+		//Restart Collectd Pods
+		reqLogger.Info("Reloading the Daemonset", "DaemonSet.Namespace", ds.Namespace, "DaemonSet.Name", ds.Name)
+		//Restart only if hash of conf has changed.
+		ds.Spec.Template.SetAnnotations(map[string]string{
+			"daaas-random": collectdutils.ComputeSHA256([]byte(collectdConf)),
+		})
+		updateErr := r.client.Update(context.TODO(), ds)
+		return updateErr
+	})
+	if retryErr != nil {
+		panic(fmt.Errorf("Update failed: %v", retryErr))
 	}
 
-	// Select DaemonSets with label app=collectd
-	dsList := &extensionsv1beta1.DaemonSetList{}
-	err = r.client.List(context.TODO(), opts, dsList)
+	err = r.updateStatus(cr)
 	if err != nil {
-		return rmap, err
-	}
-
-	if dsList.Items == nil || len(dsList.Items) == 0 {
-		return rmap, err
+		reqLogger.Error(err, "Unable to update status")
+		return err
 	}
+	// Reconcile success
+	reqLogger.Info("Reconcile success!!")
+	return nil
+}
 
-	// Get all collectd plugins in the current namespace to rebuild conf.
-	collectdPlugins := &onapv1alpha1.CollectdPluginList{}
-	cpOpts := &client.ListOptions{}
-	cpOpts.InNamespace(cr.Namespace)
-	err = r.client.List(context.TODO(), cpOpts, collectdPlugins)
-	if err != nil {
-		return rmap, err
+// Handle Delete CR event for additional cleanup
+func (r *ReconcileCollectdPlugin) handleDelete(reqLogger logr.Logger, cr *onapv1alpha1.CollectdPlugin) (bool, error) {
+	// Check if the CollectdPlugin instance is marked to be deleted, which is
+	// indicated by the deletion timestamp being set.
+	isMarkedToBeDeleted := cr.GetDeletionTimestamp() != nil
+	if isMarkedToBeDeleted {
+		// Update status to Deleting state
+		cr.Status.Status = onapv1alpha1.Deleting
+		cr.Status.CollectdAgents = nil
+		_ = r.client.Status().Update(context.TODO(), cr)
+
+		if collectdutils.Contains(cr.GetFinalizers(), collectdutils.CollectdFinalizer) {
+			// Run finalization logic for collectdPluginFinalizer. If the
+			// finalization logic fails, don't remove the finalizer so
+			// that we can retry during the next reconciliation.
+			if err := r.finalizeCollectdPlugin(reqLogger, cr); err != nil {
+				return isMarkedToBeDeleted, err
+			}
+
+			// Remove collectdPluginFinalizer. Once all finalizers have been
+			// removed, the object will be deleted.
+			cr.SetFinalizers(collectdutils.Remove(cr.GetFinalizers(), collectdutils.CollectdFinalizer))
+			err := r.client.Update(context.TODO(), cr)
+			if err != nil {
+				return isMarkedToBeDeleted, err
+			}
+		}
 	}
-
-	rmap.configMap = &cmList.Items[0]
-	rmap.daemonSet = &dsList.Items[0]
-	rmap.collectdPlugins = &collectdPlugins.Items //will be nil if no plugins exist
-	return rmap, err
+	return isMarkedToBeDeleted, nil
 }
 
-// Get all collectd plugins and reconstruct, compute Hash and check for changes
-func rebuildCollectdConf(cpList *[]onapv1alpha1.CollectdPlugin) (string, error) {
-	var collectdConf string
-	if *cpList == nil || len(*cpList) == 0 {
-		return "", errors.NewNotFound(corev1.Resource("collectdplugin"), "CollectdPlugin")
-	}
-	loadPlugin := make(map[string]string)
-	for _, cp := range *cpList {
-		if cp.Spec.PluginName == "global" {
-			collectdConf += cp.Spec.PluginConf + "\n"
-		} else {
-			loadPlugin[cp.Spec.PluginName] = cp.Spec.PluginConf
+func (r *ReconcileCollectdPlugin) updateStatus(cr *onapv1alpha1.CollectdPlugin) error {
+	switch cr.Status.Status {
+	case onapv1alpha1.Initial:
+		cr.Status.Status = onapv1alpha1.Created
+	case onapv1alpha1.Created, onapv1alpha1.Enabled:
+		pods, err := collectdutils.GetPodList(r.client, cr.Namespace)
+		if err != nil {
+			return err
+		}
+		if !reflect.DeepEqual(pods, cr.Status.CollectdAgents) {
+			cr.Status.CollectdAgents = pods
+			cr.Status.Status = onapv1alpha1.Enabled
 		}
+	case onapv1alpha1.Deleting, onapv1alpha1.Deprecated:
+		return nil
 	}
+	err := r.client.Status().Update(context.TODO(), cr)
+	return err
+}
 
-	log.V(1).Info("::::::: Plugins Map ::::::: ", "PluginMap ", loadPlugin)
-
-	for cpName, cpConf := range loadPlugin {
-		collectdConf += "LoadPlugin" + " " + cpName + "\n"
-		collectdConf += cpConf + "\n"
+func (r *ReconcileCollectdPlugin) finalizeCollectdPlugin(reqLogger logr.Logger, cr *onapv1alpha1.CollectdPlugin) error {
+	// Cleanup by regenerating new collectd conf and rolling update of DaemonSet
+	if err := r.handleCollectdPlugin(reqLogger, cr, true); err != nil {
+		reqLogger.Error(err, "Finalize CollectdPlugin failed!!")
+		return err
 	}
+	reqLogger.Info("Successfully finalized CollectdPlugin!!")
+	return nil
+}
 
-	collectdConf += "\n#Last line (collectd requires ‘\\n’ at the last line)"
-
-	return collectdConf, nil
+func (r *ReconcileCollectdPlugin) addFinalizer(reqLogger logr.Logger, cr *onapv1alpha1.CollectdPlugin) error {
+	reqLogger.Info("Adding Finalizer for the CollectdPlugin")
+	cr.SetFinalizers(append(cr.GetFinalizers(), collectdutils.CollectdFinalizer))
+	// 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")
+	}
+	// Update CR
+	err := r.client.Update(context.TODO(), cr)
+	if err != nil {
+		reqLogger.Error(err, "Failed to update CollectdPlugin with finalizer")
+		return err
+	}
+	return nil
 }