diff options
Diffstat (limited to 'vnfs/DAaaS/microservices/collectd-operator/pkg/controller')
4 files changed, 152 insertions, 41 deletions
diff --git a/vnfs/DAaaS/microservices/collectd-operator/pkg/controller/collectdglobal/collectdglobal_controller.go b/vnfs/DAaaS/microservices/collectd-operator/pkg/controller/collectdglobal/collectdglobal_controller.go index 0d3e2bbf..8c6023b8 100644 --- a/vnfs/DAaaS/microservices/collectd-operator/pkg/controller/collectdglobal/collectdglobal_controller.go +++ b/vnfs/DAaaS/microservices/collectd-operator/pkg/controller/collectdglobal/collectdglobal_controller.go @@ -12,6 +12,7 @@ import ( onapv1alpha1 "collectd-operator/pkg/apis/onap/v1alpha1" collectdutils "collectd-operator/pkg/controller/utils" + dspredicate "collectd-operator/pkg/controller/utils" dsutils "collectd-operator/pkg/controller/utils" appsv1 "k8s.io/api/apps/v1" @@ -69,22 +70,13 @@ func add(mgr manager.Manager, r reconcile.Reconciler) error { log.Error(err, "Failed to get watch labels, continuing with default label") } rcp := r.(*ReconcileCollectdGlobal) - // Select the Daemonset with labelSelector (Defautl is app=collectd) + // Select the Daemonset with labelSelector (Default is app=collectd) if a.Meta.GetLabels()[labels[0]] == labels[1] { var requests []reconcile.Request cg, err := collectdutils.GetCollectdGlobal(rcp.client, a.Meta.GetNamespace()) if err != nil || cg == nil { log.V(1).Info("No CollectdGlobal CR instance Exist") - 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 } requests = append(requests, reconcile.Request{ NamespacedName: client.ObjectKey{Namespace: cg.Namespace, Name: cg.Name}}) @@ -92,7 +84,7 @@ func add(mgr manager.Manager, r reconcile.Reconciler) error { } return nil }), - }, predicate.GenerationChangedPredicate{}) + }, dspredicate.DaemonSetStatusChangedPredicate{}) if err != nil { return err } @@ -224,10 +216,13 @@ func (r *ReconcileCollectdGlobal) handleCollectdGlobal(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!!") @@ -266,22 +261,29 @@ func (r *ReconcileCollectdGlobal) handleDelete(reqLogger logr.Logger, cr *onapv1 } func (r *ReconcileCollectdGlobal) updateStatus(cr *onapv1alpha1.CollectdGlobal) error { - switch cr.Status.Status { + // Fetch the CollectdGlobal instance + instance := &onapv1alpha1.CollectdGlobal{} + 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 } @@ -301,11 +303,12 @@ func (r *ReconcileCollectdGlobal) 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 CollectdGlobal with finalizer") 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") diff --git a/vnfs/DAaaS/microservices/collectd-operator/pkg/controller/utils/collectdutils.go b/vnfs/DAaaS/microservices/collectd-operator/pkg/controller/utils/collectdutils.go index 17cad0e5..b379d916 100644 --- a/vnfs/DAaaS/microservices/collectd-operator/pkg/controller/utils/collectdutils.go +++ b/vnfs/DAaaS/microservices/collectd-operator/pkg/controller/utils/collectdutils.go @@ -211,7 +211,7 @@ func RebuildCollectdConf(rc client.Client, ns string, isDelete bool, delPlugin s collectdConf += cpConf + "\n" } - collectdConf += "#Last line (collectd requires ‘\\n’ at the last line)\n" + collectdConf += "#Last line (collectd requires '\\n' at the last line)\n" return collectdConf, nil } diff --git a/vnfs/DAaaS/microservices/collectd-operator/pkg/controller/utils/predicate.go b/vnfs/DAaaS/microservices/collectd-operator/pkg/controller/utils/predicate.go new file mode 100644 index 00000000..a9ec1dc4 --- /dev/null +++ b/vnfs/DAaaS/microservices/collectd-operator/pkg/controller/utils/predicate.go @@ -0,0 +1,65 @@ +// Copyright 2018 The Operator-SDK Authors +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package utils + +import ( + appsv1 "k8s.io/api/apps/v1" + + "sigs.k8s.io/controller-runtime/pkg/event" + "sigs.k8s.io/controller-runtime/pkg/predicate" + logf "sigs.k8s.io/controller-runtime/pkg/runtime/log" +) + +var plog = logf.Log.WithName("predicate").WithName("eventFilters") + +// DaemonSetStatusChangedPredicate implements a default update predicate function on status change for Daemonsets +// (adapted from sigs.k8s.io/controller-runtime/pkg/predicate/predicate.GenerationChangedPredicate) +type DaemonSetStatusChangedPredicate struct { + predicate.Funcs +} + +// Update implements default UpdateEvent filter for validating generation change +func (DaemonSetStatusChangedPredicate) Update(e event.UpdateEvent) bool { + newDS := e.ObjectNew.DeepCopyObject().(*appsv1.DaemonSet) + oldDS := e.ObjectOld.DeepCopyObject().(*appsv1.DaemonSet) + plog.V(2).Info("newDS", "nUNS:=", newDS.Status.UpdatedNumberScheduled, "oUNS:=", oldDS.Status.UpdatedNumberScheduled, "nDNS:=", newDS.Status.DesiredNumberScheduled, "nNR:=", newDS.Status.NumberReady, "nNA:=", newDS.Status.NumberAvailable) + if newDS.Status.UpdatedNumberScheduled >= oldDS.Status.UpdatedNumberScheduled { + if (newDS.Status.UpdatedNumberScheduled == newDS.Status.NumberReady) && + (newDS.Status.UpdatedNumberScheduled == newDS.Status.NumberAvailable) { + return true + } + } + if e.MetaOld == nil { + plog.Error(nil, "Update event has no old metadata", "event", e) + return false + } + if e.ObjectOld == nil { + plog.Error(nil, "Update event has no old runtime object to update", "event", e) + return false + } + if e.ObjectNew == nil { + plog.Error(nil, "Update event has no new runtime object for update", "event", e) + return false + } + if e.MetaNew == nil { + plog.Error(nil, "Update event has no new metadata", "event", e) + return false + } + if e.MetaNew.GetGeneration() == e.MetaOld.GetGeneration() { + return false + } + + return true +} |