"context"
"fmt"
"reflect"
- "strings"
"github.com/go-logr/logr"
+ "github.com/operator-framework/operator-sdk/pkg/predicate"
- onapv1alpha1 "demo/vnfs/DAaaS/microservices/collectd-operator/pkg/apis/onap/v1alpha1"
- collectdutils "demo/vnfs/DAaaS/microservices/collectd-operator/pkg/controller/utils"
+ 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"
var log = logf.Log.WithName("controller_collectdplugin")
-
// 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 {
// 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{})
- 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
- }),
- })
+ err = c.Watch(&source.Kind{Type: &onapv1alpha1.CollectdPlugin{}}, &handler.EnqueueRequestForObject{}, predicate.GenerationChangedPredicate{})
if err != nil {
return err
}
return nil
}
-
// blank assignment to verify that ReconcileCollectdPlugin implements reconcile.Reconciler
var _ reconcile.Reconciler = &ReconcileCollectdPlugin{}
scheme *runtime.Scheme
}
-
// Reconcile reads that state of the cluster for a CollectdPlugin object and makes changes based on the state read
// and what is in the CollectdPlugin.Spec
// TODO(user): Modify this Reconcile function to implement your Controller logic. This example creates
if err := r.addFinalizer(reqLogger, instance); err != nil {
return reconcile.Result{}, err
}
- return reconcile.Result{}, nil
+ //return reconcile.Result{}, nil
}
// Handle the reconciliation for CollectdPlugin.
// At this stage the Status of the CollectdPlugin should NOT be ""
// handleCollectdPlugin regenerates the collectd conf on CR Create, Update, Delete events
func (r *ReconcileCollectdPlugin) handleCollectdPlugin(reqLogger logr.Logger, cr *onapv1alpha1.CollectdPlugin, isDelete bool) error {
+ collectdutils.ReconcileLock.Lock()
+ defer collectdutils.ReconcileLock.Unlock()
+ retryErr := retry.RetryOnConflict(retry.DefaultRetry, func() error {
+ cm, err := collectdutils.GetConfigMap(r.client, reqLogger, cr.Namespace)
+ if err != nil {
+ reqLogger.Error(err, "Skip reconcile: ConfigMap not found")
+ return err
+ }
+ reqLogger.V(1).Info(":::: ConfigMap Info ::::", "ConfigMap.Namespace", cm.Namespace, "ConfigMap.Name", cm.Name)
- 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
- }
-
- cm.SetAnnotations(map[string]string{
- "daaas-random": collectdutils.ComputeSHA256([]byte(collectdConf)),
- })
-
- cm.Data["collectd.conf"] = collectdConf
+ 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
+ }
- // Update the ConfigMap with new Spec and reload DaemonSets
- reqLogger.Info("Updating the ConfigMap", "ConfigMap.Namespace", cm.Namespace, "ConfigMap.Name", cm.Name)
- log.V(1).Info("ConfigMap Data", "Map: ", cm.Data)
- err = r.client.Update(context.TODO(), cm)
- if err != nil {
- reqLogger.Error(err, "Update the ConfigMap failed", "ConfigMap.Namespace", cm.Namespace, "ConfigMap.Name", cm.Name)
- return err
- }
+ cm.SetAnnotations(map[string]string{
+ "daaas-random": collectdutils.ComputeSHA256([]byte(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)
+ updateErr := r.client.Update(context.TODO(), cm)
+ if updateErr != nil {
+ reqLogger.Error(updateErr, "Update ConfigMap failed")
+ return updateErr
+ }
- 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{}
+ dsList := &appsv1.DaemonSetList{}
opts := &client.ListOptions{}
labelSelector, err := collectdutils.GetWatchLabels()
if err != nil {
ds.Spec.Template.SetAnnotations(map[string]string{
"daaas-random": collectdutils.ComputeSHA256([]byte(collectdConf)),
})
- updateErr := r.client.Update(context.TODO(), ds)
+ updateErr = r.client.Update(context.TODO(), ds)
return updateErr
})
if retryErr != nil {
panic(fmt.Errorf("Update failed: %v", retryErr))
}
- err = r.updateStatus(cr)
+ err := r.updateStatus(cr)
if err != nil {
reqLogger.Error(err, "Unable to update status")
return err
Code Review / demo.git / blobdiff