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         "time"
   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 // Reconcile reads that state of the cluster for a CollectdPlugin object and makes changes based on the state read
  80 // and what is in the CollectdPlugin.Spec
  81 // TODO(user): Modify this Reconcile function to implement your Controller logic.  This example creates
  82 // a Pod as an example
  83 // Note:
  84 // The Controller will requeue the Request to be processed again if the returned error is non-nil or
  85 // Result.Requeue is true, otherwise upon completion it will remove the work from the queue.
  86 func (r *ReconcileCollectdPlugin) Reconcile(request reconcile.Request) (reconcile.Result, error) {
  87         reqLogger := log.WithValues("Request.Namespace", request.Namespace, "Request.Name", request.Name)
  88         reqLogger.Info("Reconciling CollectdPlugin")
  89
  90         // Fetch the CollectdPlugin instance
  91         instance := &onapv1alpha1.CollectdPlugin{}
  92         err := r.client.Get(context.TODO(), request.NamespacedName, instance)
  93         if err != nil {
  94                 if errors.IsNotFound(err) {
  95                         // Request object not found, could have been deleted after reconcile request.
  96                         // Owned objects are automatically garbage collected. For additional cleanup logic use finalizers.
  97                         // Return and don't requeue
  98                         reqLogger.V(1).Info("CollectdPlugin object Not found")
  99                         return reconcile.Result{}, nil
 100                 }
 101                 // Error reading the object - requeue the request.
 102                 reqLogger.V(1).Info("Error reading the CollectdPlugin object, Requeuing")
 103                 return reconcile.Result{}, err
 104         }
 105
 106         rmap, err := findResourceMapForCR(r, instance)
 107         if err != nil {
 108                 reqLogger.Info("Skip reconcile: ConfigMap not found")
 109                 return reconcile.Result{}, err
 110         }
 111
 112         cm := rmap.configMap
 113         ds := rmap.daemonSet
 114         collectPlugins := rmap.collectdPlugins
 115         reqLogger.V(1).Info("Found ResourceMap")
 116         reqLogger.V(1).Info("ConfigMap.Namespace", cm.Namespace, "ConfigMap.Name", cm.Name)
 117         reqLogger.V(1).Info("DaemonSet.Namespace", ds.Namespace, "DaemonSet.Name", ds.Name)
 118
 119         collectdConf, err := rebuildCollectdConf(collectPlugins)
 120
 121         //Restart Collectd Pods
 122
 123         ts := time.Now().Format(time.RFC850)
 124         reqLogger.V(1).Info("Timestamp : ", ts)
 125         ds.Spec.Template.SetAnnotations(map[string]string{
 126                 "daaas-random": ComputeSHA256([]byte(ts)),
 127         })
 128         cm.SetAnnotations(map[string]string{
 129                 "daaas-random": ComputeSHA256([]byte(ts)),
 130         })
 131
 132         cm.Data["node-collectd.conf"] = collectdConf
 133
 134         // Update the ConfigMap with new Spec and reload DaemonSets
 135         reqLogger.Info("Updating the ConfigMap", "ConfigMap.Namespace", cm.Namespace, "ConfigMap.Name", cm.Name)
 136         log.Info("ConfigMap Data", "Map: ", cm.Data)
 137         err = r.client.Update(context.TODO(), cm)
 138         if err != nil {
 139                 return reconcile.Result{}, err
 140         }
 141
 142         err = r.client.Update(context.TODO(), ds)
 143         if err != nil {
 144                 return reconcile.Result{}, err
 145         }
 146         // Reconcile success
 147         reqLogger.Info("Updated the ConfigMap", "ConfigMap.Namespace", cm.Namespace, "ConfigMap.Name", cm.Name)
 148         return reconcile.Result{}, nil
 149 }
 150
 151 // ComputeSHA256  returns hash of data as string
 152 func ComputeSHA256(data []byte) string {
 153         hash := sha256.Sum256(data)
 154         return fmt.Sprintf("%x", hash)
 155 }
 156
 157 // findConfigMapForCR returns the configMap used by collectd Daemonset
 158 func findResourceMapForCR(r *ReconcileCollectdPlugin, cr *onapv1alpha1.CollectdPlugin) (ResourceMap, error) {
 159         cmList := &corev1.ConfigMapList{}
 160         opts := &client.ListOptions{}
 161         rmap := ResourceMap{}
 162
 163         // Select ConfigMaps with label app=collectd
 164         opts.SetLabelSelector("app=collectd")
 165         opts.InNamespace(cr.Namespace)
 166         err := r.client.List(context.TODO(), opts, cmList)
 167         if err != nil {
 168                 return rmap, err
 169         }
 170
 171         if cmList.Items == nil || len(cmList.Items) == 0 {
 172                 return rmap, err
 173         }
 174
 175         // Select DaemonSets with label app=collectd
 176         dsList := &extensionsv1beta1.DaemonSetList{}
 177         err = r.client.List(context.TODO(), opts, dsList)
 178         if err != nil {
 179                 return rmap, err
 180         }
 181
 182         if dsList.Items == nil || len(dsList.Items) == 0 {
 183                 return rmap, err
 184         }
 185
 186         // Get all collectd plugins in the current namespace to rebuild conf.
 187         collectdPlugins := &onapv1alpha1.CollectdPluginList{}
 188         cpOpts := &client.ListOptions{}
 189         cpOpts.InNamespace(cr.Namespace)
 190         err = r.client.List(context.TODO(), cpOpts, collectdPlugins)
 191         if err != nil {
 192                 return rmap, err
 193         }
 194
 195         rmap.configMap = &cmList.Items[0]
 196         rmap.daemonSet = &dsList.Items[0]
 197         rmap.collectdPlugins = &collectdPlugins.Items //will be nil if no plugins exist
 198         return rmap, err
 199 }
 200
 201 // Get all collectd plugins and reconstruct, compute Hash and check for changes
 202 func rebuildCollectdConf(cpList *[]onapv1alpha1.CollectdPlugin) (string, error) {
 203         var collectdConf string
 204         if *cpList == nil || len(*cpList) == 0 {
 205                 return "", errors.NewNotFound(corev1.Resource("collectdplugin"), "CollectdPlugin")
 206         }
 207         loadPlugin := make(map[string]string)
 208         for _, cp := range *cpList {
 209                 if cp.Spec.PluginName == "global" {
 210                         collectdConf += cp.Spec.PluginConf + "\n"
 211                 } else {
 212                         loadPlugin[cp.Spec.PluginName] = cp.Spec.PluginConf
 213                 }
 214         }
 215
 216         log.V(1).Info("::::::: Plugins Map ::::::: ", "PluginMap ", loadPlugin)
 217
 218         for cpName, cpConf := range loadPlugin {
 219                 collectdConf += "LoadPlugin" + " " + cpName + "\n"
 220                 collectdConf += cpConf + "\n"
 221         }
 222
 223         collectdConf += "\n#Last line (collectd requires ‘\\n’ at the last line)"
 224
 225         return collectdConf, nil
 226 }