vnfs/DAaaS/microservices/collectd-operator/pkg/controller/collectdplugin/collectdplugin_controller.go

   1 package collectdplugin
   2
   3 import (
   4         "context"
   5         "crypto/sha256"
   6         "fmt"
   7         "github.com/go-logr/logr"
   8
   9         onapv1alpha1 "demo/vnfs/DAaaS/microservices/collectd-operator/pkg/apis/onap/v1alpha1"
  10
  11         corev1 "k8s.io/api/core/v1"
  12         extensionsv1beta1 "k8s.io/api/extensions/v1beta1"
  13         "k8s.io/apimachinery/pkg/api/errors"
  14         "k8s.io/apimachinery/pkg/runtime"
  15         "sigs.k8s.io/controller-runtime/pkg/client"
  16         "sigs.k8s.io/controller-runtime/pkg/controller"
  17         "sigs.k8s.io/controller-runtime/pkg/handler"
  18         "sigs.k8s.io/controller-runtime/pkg/manager"
  19         "sigs.k8s.io/controller-runtime/pkg/reconcile"
  20         logf "sigs.k8s.io/controller-runtime/pkg/runtime/log"
  21         "sigs.k8s.io/controller-runtime/pkg/source"
  22 )
  23
  24 var log = logf.Log.WithName("controller_collectdplugin")
  25
  26 // ResourceMap to hold objects to update/reload
  27 type ResourceMap struct {
  28         configMap       *corev1.ConfigMap
  29         daemonSet       *extensionsv1beta1.DaemonSet
  30         collectdPlugins *[]onapv1alpha1.CollectdPlugin
  31 }
  32
  33 /**
  34 * USER ACTION REQUIRED: This is a scaffold file intended for the user to modify with their own Controller
  35 * business logic.  Delete these comments after modifying this file.*
  36  */
  37
  38 // Add creates a new CollectdPlugin Controller and adds it to the Manager. The Manager will set fields on the Controller
  39 // and Start it when the Manager is Started.
  40 func Add(mgr manager.Manager) error {
  41         return add(mgr, newReconciler(mgr))
  42 }
  43
  44 // newReconciler returns a new reconcile.Reconciler
  45 func newReconciler(mgr manager.Manager) reconcile.Reconciler {
  46         return &ReconcileCollectdPlugin{client: mgr.GetClient(), scheme: mgr.GetScheme()}
  47 }
  48
  49 // add adds a new Controller to mgr with r as the reconcile.Reconciler
  50 func add(mgr manager.Manager, r reconcile.Reconciler) error {
  51         // Create a new controller
  52         log.V(1).Info("Creating a new controller for CollectdPlugin")
  53         c, err := controller.New("collectdplugin-controller", mgr, controller.Options{Reconciler: r})
  54         if err != nil {
  55                 return err
  56         }
  57
  58         // Watch for changes to primary resource CollectdPlugin
  59         log.V(1).Info("Add watcher for primary resource CollectdPlugin")
  60         err = c.Watch(&source.Kind{Type: &onapv1alpha1.CollectdPlugin{}}, &handler.EnqueueRequestForObject{})
  61         if err != nil {
  62                 return err
  63         }
  64
  65         return nil
  66 }
  67
  68 // blank assignment to verify that ReconcileCollectdPlugin implements reconcile.Reconciler
  69 var _ reconcile.Reconciler = &ReconcileCollectdPlugin{}
  70
  71 // ReconcileCollectdPlugin reconciles a CollectdPlugin object
  72 type ReconcileCollectdPlugin struct {
  73         // This client, initialized using mgr.Client() above, is a split client
  74         // that reads objects from the cache and writes to the apiserver
  75         client client.Client
  76         scheme *runtime.Scheme
  77 }
  78
  79 // Define the collectdPlugin finalizer for handling deletion
  80 const collectdPluginFinalizer = "finalizer.collectdplugin.onap.org"
  81
  82 // Reconcile reads that state of the cluster for a CollectdPlugin object and makes changes based on the state read
  83 // and what is in the CollectdPlugin.Spec
  84 // TODO(user): Modify this Reconcile function to implement your Controller logic.  This example creates
  85 // a Pod as an example
  86 // Note:
  87 // The Controller will requeue the Request to be processed again if the returned error is non-nil or
  88 // Result.Requeue is true, otherwise upon completion it will remove the work from the queue.
  89 func (r *ReconcileCollectdPlugin) Reconcile(request reconcile.Request) (reconcile.Result, error) {
  90         reqLogger := log.WithValues("Request.Namespace", request.Namespace, "Request.Name", request.Name)
  91         reqLogger.Info("Reconciling CollectdPlugin")
  92
  93         // Fetch the CollectdPlugin instance
  94         instance := &onapv1alpha1.CollectdPlugin{}
  95         err := r.client.Get(context.TODO(), request.NamespacedName, instance)
  96         if err != nil {
  97                 if errors.IsNotFound(err) {
  98                         // Request object not found, could have been deleted after reconcile request.
  99                         // Owned objects are automatically garbage collected. For additional cleanup logic use finalizers.
 100                         // Return and don't requeue
 101                         reqLogger.V(1).Info("CollectdPlugin object Not found")
 102                         return reconcile.Result{}, nil
 103                 }
 104                 // Error reading the object - requeue the request.
 105                 reqLogger.V(1).Info("Error reading the CollectdPlugin object, Requeuing")
 106                 return reconcile.Result{}, err
 107         }
 108
 109         // Handle Delete CR for additional cleanup
 110         isDelete, err := r.handleDelete(reqLogger, instance)
 111         if isDelete {
 112                 return reconcile.Result{}, err
 113         }
 114
 115         // Add finalizer for this CR
 116         if !contains(instance.GetFinalizers(), collectdPluginFinalizer) {
 117                 if err := r.addFinalizer(reqLogger, instance); err != nil {
 118                         return reconcile.Result{}, err
 119                 }
 120         }
 121         err = r.handleCollectdPlugin(reqLogger, instance)
 122         return reconcile.Result{}, err
 123 }
 124
 125 // handleCollectdPlugin regenerates the collectd conf on CR Create, Update, Delete events
 126 func (r *ReconcileCollectdPlugin) handleCollectdPlugin(reqLogger logr.Logger, cr *onapv1alpha1.CollectdPlugin) error {
 127
 128         rmap, err := r.findResourceMapForCR(cr)
 129         if err != nil {
 130                 reqLogger.Error(err, "Skip reconcile: Resources not found")
 131                 return err
 132         }
 133
 134         cm := rmap.configMap
 135         ds := rmap.daemonSet
 136         collectPlugins := rmap.collectdPlugins
 137         reqLogger.V(1).Info("Found ResourceMap")
 138         reqLogger.V(1).Info(":::: ConfigMap Info ::::", "ConfigMap.Namespace", cm.Namespace, "ConfigMap.Name", cm.Name)
 139         reqLogger.V(1).Info(":::: DaemonSet Info ::::", "DaemonSet.Namespace", ds.Namespace, "DaemonSet.Name", ds.Name)
 140
 141         collectdConf, err := rebuildCollectdConf(collectPlugins)
 142
 143         //Restart Collectd Pods
 144         //Restart only if hash of configmap has changed.
 145         ds.Spec.Template.SetAnnotations(map[string]string{
 146                 "daaas-random": ComputeSHA256([]byte(collectdConf)),
 147         })
 148         cm.SetAnnotations(map[string]string{
 149                 "daaas-random": ComputeSHA256([]byte(collectdConf)),
 150         })
 151
 152         cm.Data["node-collectd.conf"] = collectdConf
 153
 154         // Update the ConfigMap with new Spec and reload DaemonSets
 155         reqLogger.Info("Updating the ConfigMap", "ConfigMap.Namespace", cm.Namespace, "ConfigMap.Name", cm.Name)
 156         log.V(1).Info("ConfigMap Data", "Map: ", cm.Data)
 157         err = r.client.Update(context.TODO(), cm)
 158         if err != nil {
 159                 reqLogger.Error(err, "Update the ConfigMap failed", "ConfigMap.Namespace", cm.Namespace, "ConfigMap.Name", cm.Name)
 160                 return err
 161         }
 162
 163         err = r.client.Update(context.TODO(), ds)
 164         if err != nil {
 165                 reqLogger.Error(err, "Update the DaemonSet failed", "DaemonSet.Namespace", ds.Namespace, "DaemonSet.Name", ds.Name)
 166                 return err
 167         }
 168         r.updateStatus(cr)
 169         // Reconcile success
 170         reqLogger.Info("Reconcile success!!")
 171         return nil
 172 }
 173
 174 // ComputeSHA256  returns hash of data as string
 175 func ComputeSHA256(data []byte) string {
 176         hash := sha256.Sum256(data)
 177         return fmt.Sprintf("%x", hash)
 178 }
 179
 180 // findResourceMapForCR returns the configMap, collectd Daemonset and list of Collectd Plugins
 181 func (r *ReconcileCollectdPlugin) findResourceMapForCR(cr *onapv1alpha1.CollectdPlugin) (ResourceMap, error) {
 182         cmList := &corev1.ConfigMapList{}
 183         opts := &client.ListOptions{}
 184         rmap := ResourceMap{}
 185
 186         // Select ConfigMaps with label app=collectd
 187         opts.SetLabelSelector("app=collectd")
 188         opts.InNamespace(cr.Namespace)
 189         err := r.client.List(context.TODO(), opts, cmList)
 190         if err != nil {
 191                 return rmap, err
 192         }
 193
 194         if cmList.Items == nil || len(cmList.Items) == 0 {
 195                 return rmap, err
 196         }
 197
 198         // Select DaemonSets with label app=collectd
 199         dsList := &extensionsv1beta1.DaemonSetList{}
 200         err = r.client.List(context.TODO(), opts, dsList)
 201         if err != nil {
 202                 return rmap, err
 203         }
 204
 205         if dsList.Items == nil || len(dsList.Items) == 0 {
 206                 return rmap, err
 207         }
 208
 209         // Get all collectd plugins in the current namespace to rebuild conf.
 210         collectdPlugins := &onapv1alpha1.CollectdPluginList{}
 211         cpOpts := &client.ListOptions{}
 212         cpOpts.InNamespace(cr.Namespace)
 213         err = r.client.List(context.TODO(), cpOpts, collectdPlugins)
 214         if err != nil {
 215                 return rmap, err
 216         }
 217
 218         rmap.configMap = &cmList.Items[0]
 219         rmap.daemonSet = &dsList.Items[0]
 220         rmap.collectdPlugins = &collectdPlugins.Items //will be nil if no plugins exist
 221         return rmap, err
 222 }
 223
 224 // Get all collectd plugins and reconstruct, compute Hash and check for changes
 225 func rebuildCollectdConf(cpList *[]onapv1alpha1.CollectdPlugin) (string, error) {
 226         var collectdConf string
 227         if *cpList == nil || len(*cpList) == 0 {
 228                 return "", errors.NewNotFound(corev1.Resource("collectdplugin"), "CollectdPlugin")
 229         }
 230         loadPlugin := make(map[string]string)
 231         for _, cp := range *cpList {
 232                 if cp.Spec.PluginName == "global" {
 233                         collectdConf += cp.Spec.PluginConf + "\n"
 234                 } else {
 235                         loadPlugin[cp.Spec.PluginName] = cp.Spec.PluginConf
 236                 }
 237         }
 238
 239         log.V(1).Info("::::::: Plugins Map ::::::: ", "PluginMap ", loadPlugin)
 240
 241         for cpName, cpConf := range loadPlugin {
 242                 collectdConf += "LoadPlugin" + " " + cpName + "\n"
 243                 collectdConf += cpConf + "\n"
 244         }
 245
 246         collectdConf += "\n#Last line (collectd requires ‘\\n’ at the last line)"
 247
 248         return collectdConf, nil
 249 }
 250
 251 // Handle Delete CR event for additional cleanup
 252 func (r *ReconcileCollectdPlugin) handleDelete(reqLogger logr.Logger, cr *onapv1alpha1.CollectdPlugin) (bool, error) {
 253         // Check if the CollectdPlugin instance is marked to be deleted, which is
 254         // indicated by the deletion timestamp being set.
 255         isMarkedToBeDeleted := cr.GetDeletionTimestamp() != nil
 256         if isMarkedToBeDeleted {
 257                 if contains(cr.GetFinalizers(), collectdPluginFinalizer) {
 258                         // Run finalization logic for collectdPluginFinalizer. If the
 259                         // finalization logic fails, don't remove the finalizer so
 260                         // that we can retry during the next reconciliation.
 261                         if err := r.finalizeCollectdPlugin(reqLogger, cr); err != nil {
 262                                 return isMarkedToBeDeleted, err
 263                         }
 264
 265                         // Remove collectdPluginFinalizer. Once all finalizers have been
 266                         // removed, the object will be deleted.
 267                         cr.SetFinalizers(remove(cr.GetFinalizers(), collectdPluginFinalizer))
 268                         err := r.client.Update(context.TODO(), cr)
 269                         if err != nil {
 270                                 return isMarkedToBeDeleted, err
 271                         }
 272                 }
 273         }
 274         return isMarkedToBeDeleted, nil
 275 }
 276
 277 func (r *ReconcileCollectdPlugin) updateStatus(cr *onapv1alpha1.CollectdPlugin) error {
 278         podList := &corev1.PodList{}
 279         opts := &client.ListOptions{}
 280         opts.SetLabelSelector("app=collectd")
 281         var pods []string
 282         opts.InNamespace(cr.Namespace)
 283         err := r.client.List(context.TODO(), opts, podList)
 284         if err != nil {
 285                 return err
 286         }
 287
 288         if podList.Items == nil || len(podList.Items) == 0 {
 289                 return err
 290         }
 291
 292         for _, pod := range podList.Items {
 293                 pods = append(pods, pod.Name)
 294         }
 295         cr.Status.CollectdAgents = pods
 296         err = r.client.Status().Update(context.TODO(), cr)
 297         return err
 298 }
 299
 300 func (r *ReconcileCollectdPlugin) finalizeCollectdPlugin(reqLogger logr.Logger, cr *onapv1alpha1.CollectdPlugin) error {
 301         // Cleanup by regenerating new collectd conf and rolling update of DaemonSet
 302         if err := r.handleCollectdPlugin(reqLogger, cr); err != nil {
 303                 reqLogger.Error(err, "Finalize CollectdPlugin failed!!")
 304                 return err
 305         }
 306         reqLogger.Info("Successfully finalized CollectdPlugin!!")
 307         return nil
 308 }
 309
 310 func (r *ReconcileCollectdPlugin) addFinalizer(reqLogger logr.Logger, cr *onapv1alpha1.CollectdPlugin) error {
 311         reqLogger.Info("Adding Finalizer for the CollectdPlugin")
 312         cr.SetFinalizers(append(cr.GetFinalizers(), collectdPluginFinalizer))
 313
 314         // Update CR
 315         err := r.client.Update(context.TODO(), cr)
 316         if err != nil {
 317                 reqLogger.Error(err, "Failed to update CollectdPlugin with finalizer")
 318                 return err
 319         }
 320         return nil
 321 }
 322
 323 func contains(list []string, s string) bool {
 324         for _, v := range list {
 325                 if v == s {
 326                         return true
 327                 }
 328         }
 329         return false
 330 }
 331
 332 func remove(list []string, s string) []string {
 333         for i, v := range list {
 334                 if v == s {
 335                         list = append(list[:i], list[i+1:]...)
 336                 }
 337         }
 338         return list
 339 }