Merge branch 'master' into fix/simplenificluster_config

CHANGELOG.md

@@ -3,6 +3,8 @@
 ### Added
 
 - [PR #76](https://github.com/konpyutaika/nifikop/pull/76) - **[Operator/NiFiCluster]** Add ability to override default authorizers.xml template.
+- [PR #95](https://github.com/konpyutaika/nifikop/pull/95) - **[Operator/NiFiParameterContext]** Allow the operator to take over existing parameter context.
+- [PR #96](https://github.com/konpyutaika/nifikop/pull/96) - **[Operator/NifiCluster]** Add ability to specify pod priority class
 
 ### Changed
 
@@ -18,7 +20,7 @@
 
 - [PR #76](https://github.com/konpyutaika/nifikop/pull/88) - **[Operator/NiFiCluster]** Re-ordering config out of sync steps.
 - [PR #93](https://github.com/konpyutaika/nifikop/pull/93) - **[Documentation]** Remove serviceAnnotations mentions and fix docs.
-
+- [PR #101](https://github.com/konpyutaika/nifikop/pull/101) - **[Operator]** Handle finalizer removal case where `NifiCluster` is aggressively torn down and pods are no longer available to communicate with.
 
 ## v0.10.0
 

api/v1alpha1/nificluster_types.go

@@ -298,6 +298,9 @@ type NodeConfig struct {
 	Tolerations []corev1.Toleration `json:"tolerations,omitempty"`
 	// podMetadata allows to add additionnal metadata to the node pods
 	PodMetadata Metadata `json:"podMetadata,omitempty"`
+	// priorityClassName can be used to set the priority class applied to the node
+	// +optional
+	PriorityClassName *string `json:"priorityClassName,omitempty"`
 }
 
 type Metadata struct {
@@ -655,6 +658,14 @@ func (nConfig *NodeConfig) GetResources() *corev1.ResourceRequirements {
 	}
 }
 
+// GetPriorityClass returns the name of the priority class to use for the given node
+func (nConfig *NodeConfig) GetPriorityClass() string {
+	if nConfig.PriorityClassName != nil {
+		return *nConfig.PriorityClassName
+	}
+	return ""
+}
+
 //
 func (nConfig *NodeConfig) GetRunAsUser() *int64 {
 	var defaultUserID int64 = 1000

api/v1alpha1/nifiparametercontext_types.go

@@ -17,6 +17,8 @@ type NifiParameterContextSpec struct {
 	ClusterRef ClusterReference `json:"clusterRef,omitempty"`
 	// a list of secret containing sensitive parameters (the key will name of the parameter).
 	SecretRefs []SecretReference `json:"secretRefs,omitempty"`
+	// whether or not the operator should take over an existing parameter context if its name is the same.
+	DisableTakeOver *bool `json:"disableTakeOver,omitempty"`
 }
 
 type Parameter struct {
@@ -83,3 +85,10 @@ type NifiParameterContextList struct {
 func init() {
 	SchemeBuilder.Register(&NifiParameterContext{}, &NifiParameterContextList{})
 }
+
+func (d *NifiParameterContextSpec) IsTakeOverEnabled() bool {
+	if d.DisableTakeOver == nil {
+		return true
+	}
+	return !*d.DisableTakeOver
+}

config/crd/bases/nifi.konpyutaika.com_nificlusters.yaml

@@ -3427,6 +3427,10 @@ spec:
                             associated https://kubernetes.io/docs/concepts/overview/working-with-objects/labels/#syntax-and-character-set
                           type: object
                       type: object
+                    priorityClassName:
+                      description: priorityClassName can be used to set the priority
+                        class applied to the node
+                      type: string
                     provenanceStorage:
                       description: provenanceStorage allow to specify the maximum
                         amount of data provenance information to store at a time https://nifi.apache.org/docs/nifi-docs/html/administration-guide.html#write-ahead-provenance-repository-properties
@@ -5670,6 +5674,10 @@ spec:
                                 associated https://kubernetes.io/docs/concepts/overview/working-with-objects/labels/#syntax-and-character-set
                               type: object
                           type: object
+                        priorityClassName:
+                          description: priorityClassName can be used to set the priority
+                            class applied to the node
+                          type: string
                         provenanceStorage:
                           description: provenanceStorage allow to specify the maximum
                             amount of data provenance information to store at a time

config/crd/bases/nifi.konpyutaika.com_nifiparametercontexts.yaml

@@ -50,6 +50,10 @@ spec:
               description:
                 description: the Description of the Parameter Context.
                 type: string
+              disableTakeOver:
+                description: whether or not the operator should take over an existing
+                  parameter context if its name is the same.
+                type: boolean
               parameters:
                 description: a list of non-sensitive Parameters.
                 items:

controllers/nifidataflow_controller.go

@@ -235,6 +235,13 @@ func (r *NifiDataflowReconciler) Reconcile(ctx context.Context, req ctrl.Request
 		r.Recorder.Event(instance, corev1.EventTypeWarning, "ReferenceClusterError",
 			fmt.Sprintf("Failed to create HTTP client for the referenced cluster : %s in %s",
 				instance.Spec.ClusterRef.Name, currentClusterRef.Namespace))
+		// the cluster is gone, so just remove the finalizer
+		if k8sutil.IsMarkedForDeletion(instance.ObjectMeta) {
+			if err = r.removeFinalizer(ctx, instance); err != nil {
+				return RequeueWithError(r.Log, fmt.Sprintf("failed to remove finalizer from NifiDataflow %s", instance.Name), err)
+			}
+			return Reconciled()
+		}
 		// the cluster does not exist - should have been caught pre-flight
 		return RequeueWithError(r.Log, "failed to create HTTP client the for referenced cluster", err)
 	}
@@ -491,8 +498,7 @@ func (r *NifiDataflowReconciler) updateAndFetchLatest(ctx context.Context,
 
 func (r *NifiDataflowReconciler) checkFinalizers(ctx context.Context, flow *v1alpha1.NifiDataflow,
 	config *clientconfig.NifiConfig) (reconcile.Result, error) {
-
-	r.Log.Info("NiFi dataflow is marked for deletion")
+	r.Log.Info(fmt.Sprintf("NiFi dataflow %s is marked for deletion", flow.Name))
 	var err error
 	if util.StringSliceContains(flow.GetFinalizers(), dataflowFinalizer) {
 		if err = r.finalizeNifiDataflow(flow, config); err != nil {
@@ -512,6 +518,7 @@ func (r *NifiDataflowReconciler) checkFinalizers(ctx context.Context, flow *v1al
 }
 
 func (r *NifiDataflowReconciler) removeFinalizer(ctx context.Context, flow *v1alpha1.NifiDataflow) error {
+	r.Log.V(5).Info(fmt.Sprintf("Removing finalizer for NifiDataflow %s", flow.Name))
 	flow.SetFinalizers(util.StringSliceRemove(flow.GetFinalizers(), dataflowFinalizer))
 	_, err := r.updateAndFetchLatest(ctx, flow)
 	return err

controllers/nifiparametercontext_controller.go

@@ -18,9 +18,12 @@ package controllers
 
 import (
 	"context"
-	"emperror.dev/errors"
 	"encoding/json"
 	"fmt"
+	"reflect"
+	"time"
+
+	"emperror.dev/errors"
 	"github.com/banzaicloud/k8s-objectmatcher/patch"
 	"github.com/konpyutaika/nifikop/pkg/clientwrappers/parametercontext"
 	errorfactory "github.com/konpyutaika/nifikop/pkg/errorfactory"
@@ -31,9 +34,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/client-go/tools/record"
-	"reflect"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
-	"time"
 
 	"github.com/go-logr/logr"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -170,6 +171,13 @@ func (r *NifiParameterContextReconciler) Reconcile(ctx context.Context, req ctrl
 		r.Recorder.Event(instance, corev1.EventTypeWarning, "ReferenceClusterError",
 			fmt.Sprintf("Failed to create HTTP client for the referenced cluster : %s in %s",
 				instance.Spec.ClusterRef.Name, clusterRef.Namespace))
+		// the cluster is gone, so just remove the finalizer
+		if k8sutil.IsMarkedForDeletion(instance.ObjectMeta) {
+			if err = r.removeFinalizer(ctx, instance); err != nil {
+				return RequeueWithError(r.Log, fmt.Sprintf("failed to remove finalizer from NifiParameterContext %s", instance.Name), err)
+			}
+			return Reconciled()
+		}
 		// the cluster does not exist - should have been caught pre-flight
 		return RequeueWithError(r.Log, "failed to create HTTP client the for referenced cluster", err)
 	}
@@ -223,9 +231,23 @@ func (r *NifiParameterContextReconciler) Reconcile(ctx context.Context, req ctrl
 		r.Recorder.Event(instance, corev1.EventTypeNormal, "Creating",
 			fmt.Sprintf("Creating parameter context %s", instance.Name))
 
-		status, err := parametercontext.CreateParameterContext(instance, parameterSecrets, clientConfig)
+		var status *v1alpha1.NifiParameterContextStatus
+
+		status, err = parametercontext.FindParameterContextByName(instance, clientConfig)
 		if err != nil {
-			return RequeueWithError(r.Log, "failure creating parameter context", err)
+			return RequeueWithError(r.Log, "failure finding parameter context", err)
+		}
+
+                 if status != nil  && !instance.Spec.IsTakeOverEnabled()  {
+                         // TakeOver disabled
+			return RequeueWithError(r.Log, fmt.Sprintf("parameter context name %s already used and takeOver disabled", instance.GetName()), err)
+                 }
+		if status == nil {
+			// Create NiFi parameter context
+			status, err = parametercontext.CreateParameterContext(instance, parameterSecrets, clientConfig)
+			if err != nil {
+				return RequeueWithError(r.Log, "failure creating parameter context", err)
+			}
 		}
 
 		instance.Status = *status
@@ -320,8 +342,7 @@ func (r *NifiParameterContextReconciler) checkFinalizers(
 	parameterContext *v1alpha1.NifiParameterContext,
 	parameterSecrets []*corev1.Secret,
 	config *clientconfig.NifiConfig) (reconcile.Result, error) {
-
-	r.Log.Info("NiFi parameter context is marked for deletion")
+	r.Log.Info(fmt.Sprintf("NiFi parameter context %s is marked for deletion", parameterContext.Name))
 	var err error
 	if util.StringSliceContains(parameterContext.GetFinalizers(), parameterContextFinalizer) {
 		if err = r.finalizeNifiParameterContext(parameterContext, parameterSecrets, config); err != nil {
@@ -334,9 +355,10 @@ func (r *NifiParameterContextReconciler) checkFinalizers(
 	return Reconciled()
 }
 
-func (r *NifiParameterContextReconciler) removeFinalizer(ctx context.Context, flow *v1alpha1.NifiParameterContext) error {
-	flow.SetFinalizers(util.StringSliceRemove(flow.GetFinalizers(), parameterContextFinalizer))
-	_, err := r.updateAndFetchLatest(ctx, flow)
+func (r *NifiParameterContextReconciler) removeFinalizer(ctx context.Context, paramCtxt *v1alpha1.NifiParameterContext) error {
+	r.Log.V(5).Info(fmt.Sprintf("Removing finalizer for NifiParameterContext %s", paramCtxt.Name))
+	paramCtxt.SetFinalizers(util.StringSliceRemove(paramCtxt.GetFinalizers(), parameterContextFinalizer))
+	_, err := r.updateAndFetchLatest(ctx, paramCtxt)
 	return err
 }
 

controllers/nifiregistryclient_controller.go

@@ -146,6 +146,13 @@ func (r *NifiRegistryClientReconciler) Reconcile(ctx context.Context, req ctrl.R
 		r.Recorder.Event(instance, corev1.EventTypeWarning, "ReferenceClusterError",
 			fmt.Sprintf("Failed to create HTTP client for the referenced cluster : %s in %s",
 				instance.Spec.ClusterRef.Name, clusterRef.Namespace))
+		// the cluster is gone, so just remove the finalizer
+		if k8sutil.IsMarkedForDeletion(instance.ObjectMeta) {
+			if err = r.removeFinalizer(ctx, instance); err != nil {
+				return RequeueWithError(r.Log, fmt.Sprintf("failed to remove finalizer from NifiRegistryClient %s", instance.Name), err)
+			}
+			return Reconciled()
+		}
 		// the cluster does not exist - should have been caught pre-flight
 		return RequeueWithError(r.Log, "failed to create HTTP client the for referenced cluster", err)
 	}
@@ -291,8 +298,7 @@ func (r *NifiRegistryClientReconciler) updateAndFetchLatest(ctx context.Context,
 
 func (r *NifiRegistryClientReconciler) checkFinalizers(ctx context.Context, reqLogger logr.Logger,
 	registryClient *v1alpha1.NifiRegistryClient, config *clientconfig.NifiConfig) (reconcile.Result, error) {
-
-	reqLogger.Info("NiFi registry client is marked for deletion")
+	reqLogger.Info(fmt.Sprintf("NiFi registry client %s is marked for deletion", registryClient.Name))
 	var err error
 	if util.StringSliceContains(registryClient.GetFinalizers(), registryClientFinalizer) {
 		if err = r.finalizeNifiRegistryClient(reqLogger, registryClient, config); err != nil {
@@ -306,6 +312,7 @@ func (r *NifiRegistryClientReconciler) checkFinalizers(ctx context.Context, reqL
 }
 
 func (r *NifiRegistryClientReconciler) removeFinalizer(ctx context.Context, registryClient *v1alpha1.NifiRegistryClient) error {
+	r.Log.V(5).Info(fmt.Sprintf("Removing finalizer for NifiRegistryClient %s", registryClient.Name))
 	registryClient.SetFinalizers(util.StringSliceRemove(registryClient.GetFinalizers(), registryClientFinalizer))
 	_, err := r.updateAndFetchLatest(ctx, registryClient)
 	return err

controllers/nifiuser_controller.go

@@ -225,6 +225,13 @@ func (r *NifiUserReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
 		r.Recorder.Event(instance, corev1.EventTypeWarning, "ReferenceClusterError",
 			fmt.Sprintf("Failed to create HTTP client for the referenced cluster : %s in %s",
 				instance.Spec.ClusterRef.Name, clusterRef.Namespace))
+		// the cluster is gone, so just remove the finalizer
+		if k8sutil.IsMarkedForDeletion(instance.ObjectMeta) {
+			if err = r.removeFinalizer(ctx, instance); err != nil {
+				return RequeueWithError(r.Log, fmt.Sprintf("failed to remove finalizer from NifiUser %s", instance.Name), err)
+			}
+			return Reconciled()
+		}
 		// the cluster does not exist - should have been caught pre-flight
 		return RequeueWithError(r.Log, "failed to create HTTP client the for referenced cluster", err)
 	}
@@ -384,7 +391,7 @@ func (r *NifiUserReconciler) updateAndFetchLatest(ctx context.Context, user *v1a
 }
 
 func (r *NifiUserReconciler) checkFinalizers(ctx context.Context, user *v1alpha1.NifiUser, config *clientconfig.NifiConfig) (reconcile.Result, error) {
-	r.Log.Info("NiFi user is marked for deletion")
+	r.Log.Info(fmt.Sprintf("NiFi user %s is marked for deletion", user.Name))
 	var err error
 	if util.StringSliceContains(user.GetFinalizers(), userFinalizer) {
 		if err = r.finalizeNifiUser(user, config); err != nil {
@@ -399,6 +406,7 @@ func (r *NifiUserReconciler) checkFinalizers(ctx context.Context, user *v1alpha1
 }
 
 func (r *NifiUserReconciler) removeFinalizer(ctx context.Context, user *v1alpha1.NifiUser) error {
+	r.Log.V(5).Info(fmt.Sprintf("Removing finalizer for NifiUser %s", user.Name))
 	user.SetFinalizers(util.StringSliceRemove(user.GetFinalizers(), userFinalizer))
 	_, err := r.updateAndFetchLatest(ctx, user)
 	return err
@@ -408,12 +416,12 @@ func (r *NifiUserReconciler) finalizeNifiUser(user *v1alpha1.NifiUser, config *c
 	if err := usercli.RemoveUser(user, config); err != nil {
 		return err
 	}
-	r.Log.Info("Delete user")
+	r.Log.Info(fmt.Sprintf("Deleted user %s", user.Name))
 	return nil
 }
 
 func (r *NifiUserReconciler) addFinalizer(user *v1alpha1.NifiUser) {
-	r.Log.Info("Adding Finalizer for the NifiUser")
+	r.Log.Info(fmt.Sprintf("Adding Finalizer for the NifiUser %s", user.Name))
 	user.SetFinalizers(append(user.GetFinalizers(), userFinalizer))
 	return
 }

controllers/nifiusergroup_controller.go

@@ -188,6 +188,13 @@ func (r *NifiUserGroupReconciler) Reconcile(ctx context.Context, req ctrl.Reques
 		r.Recorder.Event(instance, corev1.EventTypeWarning, "ReferenceClusterError",
 			fmt.Sprintf("Failed to create HTTP client for the referenced cluster : %s in %s",
 				instance.Spec.ClusterRef.Name, clusterRef.Namespace))
+		// the cluster is gone, so just remove the finalizer
+		if k8sutil.IsMarkedForDeletion(instance.ObjectMeta) {
+			if err = r.removeFinalizer(ctx, instance); err != nil {
+				return RequeueWithError(r.Log, fmt.Sprintf("failed to remove finalizer from NifiUserGroup %s", instance.Name), err)
+			}
+			return Reconciled()
+		}
 		// the cluster does not exist - should have been caught pre-flight
 		return RequeueWithError(r.Log, "failed to create HTTP client the for referenced cluster", err)
 	}
@@ -328,8 +335,7 @@ func (r *NifiUserGroupReconciler) updateAndFetchLatest(ctx context.Context,
 
 func (r *NifiUserGroupReconciler) checkFinalizers(ctx context.Context, userGroup *v1alpha1.NifiUserGroup,
 	users []*v1alpha1.NifiUser, config *clientconfig.NifiConfig) (reconcile.Result, error) {
-
-	r.Log.Info("NiFi user group is marked for deletion")
+	r.Log.Info(fmt.Sprintf("NiFi user group %s is marked for deletion", userGroup.Name))
 	var err error
 	if util.StringSliceContains(userGroup.GetFinalizers(), userGroupFinalizer) {
 		if err = r.finalizeNifiNifiUserGroup(userGroup, users, config); err != nil {
@@ -343,6 +349,7 @@ func (r *NifiUserGroupReconciler) checkFinalizers(ctx context.Context, userGroup
 }
 
 func (r *NifiUserGroupReconciler) removeFinalizer(ctx context.Context, userGroup *v1alpha1.NifiUserGroup) error {
+	r.Log.V(5).Info(fmt.Sprintf("Removing finalizer for NifiUserGroup %s", userGroup.Name))
 	userGroup.SetFinalizers(util.StringSliceRemove(userGroup.GetFinalizers(), userGroupFinalizer))
 	_, err := r.updateAndFetchLatest(ctx, userGroup)
 	return err

helm/nifikop/crds/nifi.konpyutaika.com_nificlusters.yaml

@@ -3427,6 +3427,10 @@ spec:
                             associated https://kubernetes.io/docs/concepts/overview/working-with-objects/labels/#syntax-and-character-set
                           type: object
                       type: object
+                    priorityClassName:
+                      description: priorityClassName can be used to set the priority
+                        class applied to the node
+                      type: string
                     provenanceStorage:
                       description: provenanceStorage allow to specify the maximum
                         amount of data provenance information to store at a time https://nifi.apache.org/docs/nifi-docs/html/administration-guide.html#write-ahead-provenance-repository-properties
@@ -5670,6 +5674,10 @@ spec:
                                 associated https://kubernetes.io/docs/concepts/overview/working-with-objects/labels/#syntax-and-character-set
                               type: object
                           type: object
+                        priorityClassName:
+                          description: priorityClassName can be used to set the priority
+                            class applied to the node
+                          type: string
                         provenanceStorage:
                           description: provenanceStorage allow to specify the maximum
                             amount of data provenance information to store at a time

helm/nifikop/crds/nifi.konpyutaika.com_nifiparametercontexts.yaml

@@ -50,6 +50,10 @@ spec:
               description:
                 description: the Description of the Parameter Context.
                 type: string
+              disableTakeOver:
+                description: whether or not the operator should take over an existing
+                  parameter context if its name is the same.
+                type: boolean
               parameters:
                 description: a list of non-sensitive Parameters.
                 items: