[Feature/NifiConnection] Implementation of NifiConnection controller

CHANGELOG.md

@@ -2,6 +2,8 @@
 
 ### Added
 
+- [PR #168](https://github.com/konpyutaika/nifikop/pull/168) - **[Operator/NifiConnection]** Implementation on NifiConnection controller.
+
 ### Changed
 
 - [PR #160](https://github.com/konpyutaika/nifikop/pull/160) - **[Documentation]** Upgrade documentation dependencies.

PROJECT

@@ -58,4 +58,13 @@ resources:
   kind: NifiNodeGroupAutoscaler
   path: github.com/konpyutaika/nifikop/api/v1alpha1
   version: v1alpha1
+- api:
+    crdVersion: v1
+    namespaced: true
+  controller: true
+  domain: konpyutaika.com
+  group: nifi
+  kind: NifiConnection
+  path: github.com/konpyutaika/nifikop/api/v1alpha1
+  version: v1alpha1
 version: "3"

api/v1alpha1/common_types.go

@@ -9,11 +9,15 @@ import (
 // DataflowState defines the state of a NifiDataflow
 type DataflowState string
 
+// ConnectionState defines the state of a NifiConnection
+type ConnectionState string
+
 // DataflowUpdateRequestType defines the type of versioned flow update request
 type DataflowUpdateRequestType string
 
-// DataflowUpdateStrategy defines the type of strategy to update a flow
-type DataflowUpdateStrategy string
+// ComponentUpdateStrategy defines the type of strategy to update a component
+// +kubebuilder:validation:Enum={"drop","drain"}
+type ComponentUpdateStrategy string
 
 // RackAwarenessState stores info about rack awareness status
 type RackAwarenessState string
@@ -46,12 +50,15 @@ type ConfigurationState string
 type InitClusterNode bool
 
 // PKIBackend represents an interface implementing the PKIManager
+// +kubebuilder:validation:Enum={"cert-manager","vault"}
 type PKIBackend string
 
 // ClientConfigType represents an interface implementing the ClientConfigManager
+// +kubebuilder:validation:Enum={"tls","basic"}
 type ClientConfigType string
 
 // ClusterType represents an interface implementing the  ClientConfigManager
+// +kubebuilder:validation:Enum={"external","internal"}
 type ClusterType string
 
 // AccessPolicyType represents the type of access policy
@@ -278,16 +285,23 @@ const (
 	// DataflowStateInSync describes the status of a NifiDataflow as in sync
 	DataflowStateInSync DataflowState = "InSync"
 
+	// ConnectionStateOutOfSync describes the status of a NifiConnection as out of sync
+	ConnectionStateOutOfSync ConnectionState = "OutOfSync"
+	// ConnectionStateInSync describes the status of a NifiConnection as in sync
+	ConnectionStateInSync ConnectionState = "InSync"
+	// ConnectionStateCreated describes the status of a NifiConnection as created
+	ConnectionStateCreated ConnectionState = "Created"
+
 	// RevertRequestType defines a revert changes request.
 	RevertRequestType DataflowUpdateRequestType = "Revert"
 	// UpdateRequestType defines an update version request.
 	UpdateRequestType DataflowUpdateRequestType = "Update"
 
 	// DrainStrategy leads to shutting down only input components (Input processors, remote input process group)
 	// and dropping all flowfiles from the flow.
-	DrainStrategy DataflowUpdateStrategy = "drain"
+	DrainStrategy ComponentUpdateStrategy = "drain"
 	// DropStrategy leads to shutting down all components and dropping all flowfiles from the flow.
-	DropStrategy DataflowUpdateStrategy = "drop"
+	DropStrategy ComponentUpdateStrategy = "drop"
 
 	// UserStateCreated describes the status of a NifiUser as created
 	UserStateCreated UserState = "created"
@@ -455,10 +469,86 @@ func SecretRefsEquals(secretRefs []SecretReference) bool {
 	return true
 }
 
+func ComponentRefsEquals(componentRefs []ComponentReference) bool {
+	c1 := componentRefs[0]
+	name := c1.Name
+	ns := c1.Namespace
+
+	for _, component := range componentRefs {
+		if name != component.Name || ns != component.Namespace || ns != string(component.Type) || ns != component.SubName {
+			return false
+		}
+	}
+
+	return true
+}
+
+// +kubebuilder:validation:Enum={"never","always","once"}
 type DataflowSyncMode string
 
 const (
 	SyncNever  DataflowSyncMode = "never"
 	SyncOnce   DataflowSyncMode = "once"
 	SyncAlways DataflowSyncMode = "always"
 )
+
+// Change the list to {"dataflow","input-port","output-port","processor","process-group"} when all the type are available
+// +kubebuilder:validation:Enum={"dataflow"}
+type ComponentType string
+
+const (
+	ComponentDataflow     ComponentType = "dataflow"
+	ComponentInputPort    ComponentType = "input-port"
+	ComponentOutputPort   ComponentType = "output-port"
+	ComponentProcessor    ComponentType = "processor"
+	ComponentFunnel       ComponentType = "funnel"
+	ComponentProcessGroup ComponentType = "process-group"
+)
+
+type ComponentInformation struct {
+	Id            string           `json:"id"`
+	GroupId       string           `json:"groupId"`
+	Type          string           `json:"type"`
+	ParentGroupId string           `json:"parentGroupId"`
+	ClusterRef    ClusterReference `json:"clusterRef"`
+}
+
+// +kubebuilder:validation:Enum={"DO_NOT_LOAD_BALANCE","PARTITION_BY_ATTRIBUTE","ROUND_ROBIN","SINGLE"}
+type ConnectionLoadBalanceStrategy string
+
+const (
+	// Do not load balance FlowFiles between nodes in the cluster.
+	StrategyDoNotLoadBalance ConnectionLoadBalanceStrategy = "DO_NOT_LOAD_BALANCE"
+	// Determine which node to send a given FlowFile to based on the value of a user-specified FlowFile Attribute. All FlowFiles that have the same value for said Attribute will be sent to the same node in the cluster.
+	StrategyPartitionByAttribute ConnectionLoadBalanceStrategy = "PARTITION_BY_ATTRIBUTE"
+	// FlowFiles will be distributed to nodes in the cluster in a Round-Robin fashion. However, if a node in the cluster is not able to receive data as fast as other nodes, that node may be skipped in one or more iterations in order to maximize throughput of data distribution across the cluster.
+	StrategyRoundRobin ConnectionLoadBalanceStrategy = "ROUND_ROBIN"
+	// All FlowFiles will be sent to the same node. Which node they are sent to is not defined.
+	StrategySingle ConnectionLoadBalanceStrategy = "SINGLE"
+)
+
+// +kubebuilder:validation:Enum={"DO_NOT_COMPRESS","COMPRESS_ATTRIBUTES_ONLY","COMPRESS_ATTRIBUTES_AND_CONTENT"}
+type ConnectionLoadBalanceCompression string
+
+const (
+	// FlowFiles will not be compressed.
+	CompressionDoNotCompress ConnectionLoadBalanceCompression = "DO_NOT_COMPRESS"
+	// FlowFiles' attributes will be compressed, but the FlowFiles' contents will not be
+	CompressionCompressAttributesOnly ConnectionLoadBalanceCompression = "COMPRESS_ATTRIBUTES_ONLY"
+	// FlowFiles' attributes and content will be compressed
+	CompressionCompressAttributesAndContent ConnectionLoadBalanceCompression = "COMPRESS_ATTRIBUTES_AND_CONTENT"
+)
+
+// +kubebuilder:validation:Enum={"FirstInFirstOutPrioritizer","NewestFlowFileFirstPrioritizer","OldestFlowFileFirstPrioritizer","PriorityAttributePrioritizer"}
+type ConnectionPrioritizer string
+
+const (
+	// Given two FlowFiles, the one that reached the connection first will be processed first.
+	PrioritizerFirstInFirstOutPrioritizer ConnectionPrioritizer = "FirstInFirstOutPrioritizer"
+	// Given two FlowFiles, the one that is newest in the dataflow will be processed first.
+	PrioritizerNewestFlowFileFirstPrioritizer ConnectionPrioritizer = "NewestFlowFileFirstPrioritizer"
+	// Given two FlowFiles, the one that is oldest in the dataflow will be processed first. 'This is the default scheme that is used if no prioritizers are selected'.
+	PrioritizerOldestFlowFileFirstPrioritizer ConnectionPrioritizer = "OldestFlowFileFirstPrioritizer"
+	// Given two FlowFiles, an attribute called “priority” will be extracted. The one that has the lowest priority value will be processed first.
+	PrioritizerPriorityAttributePrioritizer ConnectionPrioritizer = "PriorityAttributePrioritizer"
+)

api/v1alpha1/nificluster_types.go

@@ -27,10 +27,8 @@ const (
 // NifiClusterSpec defines the desired state of NifiCluster
 type NifiClusterSpec struct {
 	// clientType defines if the operator will use basic or tls authentication to query the NiFi cluster.
-	// +kubebuilder:validation:Enum={"tls","basic"}
 	ClientType ClientConfigType `json:"clientType,omitempty"`
 	// type defines if the cluster is internal (i.e manager by the operator) or external.
-	// +kubebuilder:validation:Enum={"external","internal"}
 	Type ClusterType `json:"type,omitempty"`
 	// nodeURITemplate used to dynamically compute node uri (used if external type)
 	NodeURITemplate string `json:"nodeURITemplate,omitempty"`
@@ -374,7 +372,6 @@ type SSLSecrets struct {
 	// https://cert-manager.io/docs/concepts/issuer/
 	IssuerRef *cmmeta.ObjectReference `json:"issuerRef,omitempty"`
 	// TODO : add vault
-	// +kubebuilder:validation:Enum={"cert-manager","vault"}
 	PKIBackend PKIBackend `json:"pkiBackend,omitempty"`
 	//,"vault"
 }

api/v1alpha1/nificonnection_types.go

@@ -0,0 +1,167 @@
+package v1alpha1
+
+import (
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+)
+
+// EDIT THIS FILE!  THIS IS SCAFFOLDING FOR YOU TO OWN!
+// NOTE: json tags are required.  Any new fields you add must have json tags for the fields to be serialized.
+
+// NifiConnectionSpec defines the desired state of NifiConnection
+type NifiConnectionSpec struct {
+	// the Source component of the connection.
+	Source ComponentReference `json:"source"`
+	// the Destination component of the connection.
+	Destination ComponentReference `json:"destination"`
+	// the Configuration of the connection.
+	Configuration ConnectionConfiguration `json:"configuration,omitempty"`
+	// describes the way the operator will deal with data when a connection will be updated : drop or drain.
+	UpdateStrategy ComponentUpdateStrategy `json:"updateStrategy"`
+}
+
+type ComponentReference struct {
+	// the name of the component.
+	Name string `json:"name"`
+	// the namespace of the component.
+	Namespace string `json:"namespace,omitempty"`
+	// the type of the component (e.g. nifidataflow).
+	Type ComponentType `json:"type"`
+	// the name of the sub component (e.g. queue or port name).
+	SubName string `json:"subName,omitempty"`
+}
+
+type ConnectionConfiguration struct {
+	// the maximum amount of time an object may be in the flow before it will be automatically aged out of the flow.
+	FlowFileExpiration string `json:"flowFileExpiration,omitempty"`
+	// the maximum data size of objects that can be queued before back pressure is applied.
+	BackPressureDataSizeThreshold string `json:"backPressureDataSizeThreshold,omitempty"`
+	// the maximum number of objects that can be queued before back pressure is applied.
+	BackPressureObjectThreshold *int64 `json:"backPressureObjectThreshold,omitempty"`
+	// how to load balance the data in this Connection across the nodes in the cluster.
+	LoadBalanceStrategy ConnectionLoadBalanceStrategy `json:"loadBalanceStrategy,omitempty"`
+	// the FlowFile Attribute to use for determining which node a FlowFile will go to.
+	LoadBalancePartitionAttribute string `json:"loadBalancePartitionAttribute,omitempty"`
+	// whether or not data should be compressed when being transferred between nodes in the cluster.
+	LoadBalanceCompression ConnectionLoadBalanceCompression `json:"loadBalanceCompression,omitempty"`
+	// the comparators used to prioritize the queue.
+	Prioritizers []ConnectionPrioritizer `json:"prioritizers,omitempty"`
+	// the index of the bend point where to place the connection label.
+	LabelIndex *int32 `json:"labelIndex,omitempty"`
+	// the bend points on the connection.
+	Bends []ConnectionBend `json:"bends,omitempty"`
+}
+
+type ConnectionBend struct {
+	// The x coordinate.
+	X *int64 `json:"posX,omitempty"`
+	// The y coordinate.
+	Y *int64 `json:"posY,omitempty"`
+}
+
+// NifiConnectionStatus defines the observed state of NifiConnection
+type NifiConnectionStatus struct {
+	// connection ID.
+	ConnectionId string `json:"connectionID"`
+	// connection current state.
+	State ConnectionState `json:"state"`
+}
+
+//+kubebuilder:object:root=true
+//+kubebuilder:subresource:status
+
+// NifiConnection is the Schema for the nificonnections API
+type NifiConnection struct {
+	metav1.TypeMeta   `json:",inline"`
+	metav1.ObjectMeta `json:"metadata,omitempty"`
+
+	Spec   NifiConnectionSpec   `json:"spec,omitempty"`
+	Status NifiConnectionStatus `json:"status,omitempty"`
+}
+
+//+kubebuilder:object:root=true
+
+// NifiConnectionList contains a list of NifiConnection
+type NifiConnectionList struct {
+	metav1.TypeMeta `json:",inline"`
+	metav1.ListMeta `json:"metadata,omitempty"`
+	Items           []NifiConnection `json:"items"`
+}
+
+func init() {
+	SchemeBuilder.Register(&NifiConnection{}, &NifiConnectionList{})
+}
+
+func (nCon *NifiConnectionSpec) IsValid() bool {
+	return nCon.Source.IsValid() && nCon.Destination.IsValid() && nCon.Configuration.IsValid()
+}
+
+func (ref *ComponentReference) IsValid() bool {
+	return ref.Type == ComponentDataflow && ref.SubName != ""
+}
+
+func (conf *ConnectionConfiguration) IsValid() bool {
+	if conf.GetLoadBalanceStrategy() == StrategyPartitionByAttribute && len(conf.GetLoadBalancePartitionAttribute()) == 0 {
+		return false
+	}
+	return true
+}
+
+func (conf *ConnectionConfiguration) GetFlowFileExpiration() string {
+	return conf.FlowFileExpiration
+}
+
+func (conf *ConnectionConfiguration) GetBackPressureDataSizeThreshold() string {
+	if len(conf.BackPressureDataSizeThreshold) > 0 {
+		return conf.BackPressureDataSizeThreshold
+	}
+	return "1 GB"
+}
+
+func (conf *ConnectionConfiguration) GetBackPressureObjectThreshold() int64 {
+	if conf.BackPressureObjectThreshold != nil {
+		return *conf.BackPressureObjectThreshold
+	}
+	return 10000
+}
+
+func (conf *ConnectionConfiguration) GetLoadBalanceStrategy() ConnectionLoadBalanceStrategy {
+	if len(conf.LoadBalanceStrategy) > 0 {
+		return conf.LoadBalanceStrategy
+	}
+	return StrategyDoNotLoadBalance
+}
+
+func (conf *ConnectionConfiguration) GetLoadBalancePartitionAttribute() string {
+	return conf.LoadBalancePartitionAttribute
+}
+
+func (conf *ConnectionConfiguration) GetLoadBalanceCompression() ConnectionLoadBalanceCompression {
+	if len(conf.LoadBalanceCompression) > 0 {
+		return conf.LoadBalanceCompression
+	}
+	return CompressionDoNotCompress
+}
+
+func (conf *ConnectionConfiguration) GetPrioritizers() []ConnectionPrioritizer {
+	return conf.Prioritizers
+}
+
+func (conf *ConnectionConfiguration) GetStringPrioritizers() []string {
+	var prefix string = "org.apache.nifi.prioritizer."
+	prioritizers := []string{}
+	for _, prioritizer := range conf.Prioritizers {
+		prioritizers = append(prioritizers, prefix+string(prioritizer))
+	}
+	return prioritizers
+}
+
+func (conf *ConnectionConfiguration) GetLabelIndex() int32 {
+	if conf.LabelIndex != nil {
+		return *conf.LabelIndex
+	}
+	return 0
+}
+
+func (conf *ConnectionConfiguration) GetBends() []ConnectionBend {
+	return conf.Bends
+}

api/v1alpha1/nifidataflow_types.go

@@ -22,7 +22,6 @@ type NifiDataflowSpec struct {
 	// contains the reference to the ParameterContext with the one the dataflow is linked.
 	ParameterContextRef *ParameterContextReference `json:"parameterContextRef,omitempty"`
 	// if the flow will be synchronized once, continuously or never
-	// +kubebuilder:validation:Enum={"never","always","once"}
 	SyncMode *DataflowSyncMode `json:"syncMode,omitempty"`
 	// whether the flow is considered as ran if some controller services are still invalid or not.
 	SkipInvalidControllerService bool `json:"skipInvalidControllerService,omitempty"`
@@ -33,8 +32,7 @@ type NifiDataflowSpec struct {
 	// contains the reference to the NifiRegistry with the one the dataflow is linked.
 	RegistryClientRef *RegistryClientReference `json:"registryClientRef,omitempty"`
 	// describes the way the operator will deal with data when a dataflow will be updated : drop or drain
-	// +kubebuilder:validation:Enum={"drop","drain"}
-	UpdateStrategy DataflowUpdateStrategy `json:"updateStrategy"`
+	UpdateStrategy ComponentUpdateStrategy `json:"updateStrategy"`
 }
 
 type FlowPosition struct {