diff --git a/CHANGELOG.md b/CHANGELOG.md index d74cd6c4..766d5c48 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -30,6 +30,12 @@ All notable changes to this project will be documented in this file. - Fix broken link to the NiFi authorization usage guide in the `spec.clusterConfig.authorization` CRD doc (`usage-guide` -> `usage_guide`) ([#924]). +### Removed + +- BREAKING: Remove support for NiFi 1.x. + This removes the Prometheus reporting-task Job (and its `spec.clusterConfig.createReportingTaskJob` field), the pre-2.x non-rolling upgrade handling, the dedicated metrics port, and the sensitive-properties algorithms that were only supported on NiFi 1.x. + `status.deployed_version` is retained even though it no longer drives the (now removed) non-rolling upgrade state machine, as we don't want a breaking change to the status just for this ([#954]). + [#903]: https://github.com/stackabletech/nifi-operator/pull/903 [#916]: https://github.com/stackabletech/nifi-operator/pull/916 [#921]: https://github.com/stackabletech/nifi-operator/pull/921 @@ -41,6 +47,7 @@ All notable changes to this project will be documented in this file. [#940]: https://github.com/stackabletech/nifi-operator/pull/940 [#945]: https://github.com/stackabletech/nifi-operator/pull/945 [#953]: https://github.com/stackabletech/nifi-operator/pull/953 +[#954]: https://github.com/stackabletech/nifi-operator/pull/954 [#959]: https://github.com/stackabletech/nifi-operator/pull/959 ## [26.3.0] - 2026-03-16 diff --git a/deploy/helm/nifi-operator/templates/clusterrole-operator.yaml b/deploy/helm/nifi-operator/templates/clusterrole-operator.yaml index 38ecb56d..845b4ed5 100644 --- a/deploy/helm/nifi-operator/templates/clusterrole-operator.yaml +++ b/deploy/helm/nifi-operator/templates/clusterrole-operator.yaml @@ -81,17 +81,6 @@ rules: - list - patch - watch - # Optional reporting-task Job (NiFi 1.x only). Applied via SSA and tracked for orphan cleanup. - - apiGroups: - - batch - resources: - - jobs - verbs: - - create - - delete - - get - - list - - patch # PodDisruptionBudget created per role. Applied via SSA and tracked for orphan cleanup. - apiGroups: - policy diff --git a/docs/modules/nifi/pages/index.adoc b/docs/modules/nifi/pages/index.adoc index ead76339..f878e102 100644 --- a/docs/modules/nifi/pages/index.adoc +++ b/docs/modules/nifi/pages/index.adoc @@ -35,7 +35,8 @@ Every role group is accessible through it's own Service, and there is a Service == Dependencies -Apache NiFi 1.x depends on Apache ZooKeeper which you can run in Kubernetes with the xref:zookeeper:index.adoc[]. +NiFi builds its cluster quorum using Kubernetes by default and needs no additional dependencies. +It can optionally use Apache ZooKeeper instead, which you can run in Kubernetes with the xref:zookeeper:index.adoc[]. == [[demos]]Demos diff --git a/docs/modules/nifi/pages/usage_guide/clustering.adoc b/docs/modules/nifi/pages/usage_guide/clustering.adoc index 417c9f9b..e0485bfa 100644 --- a/docs/modules/nifi/pages/usage_guide/clustering.adoc +++ b/docs/modules/nifi/pages/usage_guide/clustering.adoc @@ -14,8 +14,6 @@ CAUTION: The cluster backend of an existing cluster should never be changed. Oth [#backend-kubernetes] == Kubernetes -NOTE: The Kubernetes provider is only supported by Apache NiFi 2.0 or newer. When using NiFi 1.x, use the xref:#backend-zookeeper[] backend instead. - The Kubernetes backend is used by default (unless the xref:#backend-zookeeper[] backend is configured), and stores all state in Kubernetes objects, in the same namespace as the `NifiCluster` object. It takes no configuration. diff --git a/docs/modules/nifi/pages/usage_guide/custom-components.adoc b/docs/modules/nifi/pages/usage_guide/custom-components.adoc index 1d7b3c3f..d091c95e 100644 --- a/docs/modules/nifi/pages/usage_guide/custom-components.adoc +++ b/docs/modules/nifi/pages/usage_guide/custom-components.adoc @@ -12,7 +12,7 @@ There are currently two types of custom components: 1. Custom NiFi Archives (NARs) + The {nifi-docs-developers-guide}[NiFi Developer’s Guide] provides further information on how to develop custom NARs. -2. Starting with NiFi 2.0, also custom Python extensions can be used. + +2. Additionally, custom Python extensions can be used. + In the {nifi-docs-python-developers-guide}[NiFi Python Developer’s Guide], the development of custom components using Python is described. The Stackable image contains the required tooling for both types. @@ -165,7 +165,7 @@ metadata: name: simple-nifi spec: image: - productVersion: 2.2.0 + productVersion: 2.9.0 clusterConfig: authentication: - authenticationClass: simple-nifi-admin-user @@ -202,7 +202,7 @@ The basic Dockerfile below shows how to achieve this: [source,Dockerfile] ---- -FROM oci.stackable.tech/sdp/nifi:2.2.0-stackable25.3.0 +FROM oci.stackable.tech/sdp/nifi:2.9.0-stackable26.7.0 COPY /path/to/your/nar.file /stackable/nifi/lib/ COPY /path/to/your/Python.file /stackable/nifi/python/extensions/ ---- @@ -213,8 +213,8 @@ You then need to make this image available to your Kubernetes cluster and specif ---- spec: image: - productVersion: 2.2.0 - custom: oci-registry.company.org/nifi:2.2.0-customprocessors + productVersion: 2.9.0 + custom: oci-registry.company.org/nifi:2.9.0-customprocessors ---- Also read the xref:guides:custom-images.adoc[Using customized product images] guide for additional information. @@ -285,7 +285,7 @@ metadata: name: simple-nifi spec: image: - productVersion: 2.2.0 + productVersion: 2.9.0 clusterConfig: authentication: - authenticationClass: simple-nifi-admin-user diff --git a/docs/modules/nifi/pages/usage_guide/flow-versioning.adoc b/docs/modules/nifi/pages/usage_guide/flow-versioning.adoc index eba3746c..92093135 100644 --- a/docs/modules/nifi/pages/usage_guide/flow-versioning.adoc +++ b/docs/modules/nifi/pages/usage_guide/flow-versioning.adoc @@ -1,8 +1,7 @@ = Flow Versioning -IMPORTANT: The way described in this document to version flows is only available in NiFi 2.0.0 and later versions. Previous versions of NiFi relied exclusively on the NiFi Registry service for flow versioning. - -Starting with Apache NiFi 2.0.0, you can connect directly to Git repositories using either GitHub or GitLab as your flow registry without requiring a separate NiFi Registry service. This document explains how to configure and use these registry clients to version your NiFi flows. +You can connect directly to Git repositories using either GitHub or GitLab as your flow registry without requiring a separate NiFi Registry service. +This document explains how to configure and use these registry clients to version your NiFi flows. == Setting up the Registry Client diff --git a/docs/modules/nifi/pages/usage_guide/monitoring.adoc b/docs/modules/nifi/pages/usage_guide/monitoring.adoc index 3a093f87..06998ac6 100644 --- a/docs/modules/nifi/pages/usage_guide/monitoring.adoc +++ b/docs/modules/nifi/pages/usage_guide/monitoring.adoc @@ -1,42 +1,10 @@ = Monitoring :description: The Stackable Operator for Apache NiFi automatically configures NiFi to export Prometheus metrics. -:k8s-job: https://kubernetes.io/docs/concepts/workloads/controllers/job/ -:k8s-network-policies: https://kubernetes.io/docs/concepts/services-networking/network-policies/ :prometheus-operator: https://prometheus-operator.dev/ -In November 2024, Apache NiFi released a new major version https://cwiki.apache.org/confluence/display/NIFI/Release+Notes#ReleaseNotes-Version2.0.0[`2.0.0`,window=_blank]. +== Configure metrics -The NiFi `2.0.0` release changed the way of exposing Prometheus metrics significantly. -The following steps explain on how to expose Metrics in NiFi versions `1.x.x` and `2.x.x`. - -== Configure metrics in NiFi `1.x.x` - -For NiFi versions `1.x.x`, the operator automatically configures NiFi to export Prometheus metrics. -This is done by creating a {k8s-job}[Job,window=_blank] that connects to NiFi and configures a https://nifi.apache.org/docs/nifi-docs/components/org.apache.nifi/nifi-prometheus-nar/1.26.0/org.apache.nifi.reporting.prometheus.PrometheusReportingTask/index.html[Prometheus Reporting Task,window=_blank]. - -IMPORTANT: Network access from the Job to NiFi is required. -If you are running a Kubernetes with restrictive {k8s-network-policies}[NetworkPolicies,window=_blank], make sure to allow access from the Job to NiFi. - -See xref:operators:monitoring.adoc[] for more details. - -== Disabling create-reporting-task Job - -It can be helpful to disable the Job, e.g. when you configOverride an authentication mechanism, which the Job currently cannot use to authenticate against NiFi. - -To achieve this use the following configuration: - -[source,yaml] ----- -spec: - clusterConfig: - createReportingTaskJob: - enabled: false ----- - -== Configure metrics in NiFi `2.x.x` - -The Prometheus Reporting Task was removed in NiFi `2.x.x` in https://issues.apache.org/jira/browse/NIFI-13507[NIFI-13507,window=_blank]. -Metrics are now always exposed and can be scraped using the NiFi `metrics` Service and the HTTP path `/nifi-api/flow/metrics/prometheus`. +Metrics are always exposed and can be scraped using the NiFi `metrics` Service and the HTTP path `/nifi-api/flow/metrics/prometheus`. For a deployed NiFi cluster called `simple-nifi`, containing a rolegroup called `default`, the metrics endpoint is reachable under: @@ -49,9 +17,9 @@ To scrape metrics from a particular Pod, the FQDN of the Pod and the `headless` IMPORTANT: If NiFi is configured to do any user authentication, requests to the metrics endpoint must be authenticated and authorized. -=== Authentication with NiFi `2.x.x` +=== Authentication with NiFi -To authenticate against the NiFi `2.x.x` API, you can configure mTLS between NiFi and the client calling NiFi. For more information about authentication between +To authenticate against the NiFi API, you can configure mTLS between NiFi and the client calling NiFi. For more information about authentication between Kubernetes Pods, check out the xref:home:secret-operator:index.adoc[Secret Operator documentation]. The following example illustrates the configuration of a Prometheus scraper for NiFi, using the aforementioned method of configuring mTLS diff --git a/docs/modules/nifi/pages/usage_guide/overrides.adoc b/docs/modules/nifi/pages/usage_guide/overrides.adoc index 8c862aea..ae24295a 100644 --- a/docs/modules/nifi/pages/usage_guide/overrides.adoc +++ b/docs/modules/nifi/pages/usage_guide/overrides.adoc @@ -94,25 +94,6 @@ Pod overrides allow you to configure any attributes that can be configured on a Read the xref:concepts:overrides.adoc#pod-overrides[Pod overrides concepts page] to learn more. -=== Pod overrides on create-reporting-task Job - -In addition to podOverrides on the created StatefulSet we also support podOverrides on the created Kubernetes Job, which enables the export of Prometheus metrics within NiFi. - -[source,yaml] ----- -spec: - clusterConfig: - createReportingTaskJob: - # enabled: false # You can also turn off the Job entirely - podOverrides: # podOverrides as usual - spec: - tolerations: - - key: "key1" - operator: "Equal" - value: "value1" - effect: "NoSchedule" ----- - == JVM argument overrides Stackable operators automatically determine the set of needed JVM arguments, such as memory settings or trust- and keystores. diff --git a/docs/modules/nifi/pages/usage_guide/updating.adoc b/docs/modules/nifi/pages/usage_guide/updating.adoc index f290add4..4df8ef62 100644 --- a/docs/modules/nifi/pages/usage_guide/updating.adoc +++ b/docs/modules/nifi/pages/usage_guide/updating.adoc @@ -17,16 +17,4 @@ spec: <1> Change the NiFi version here -[WARNING] -==== -NiFi clusters cannot be upgraded or downgraded in a rolling fashion due to a limitation in NiFi prior to version 2. - -When upgrading between NiFi 1 versions or from NiFi 1 to NiFi 2, any change to the NiFi version in the CRD triggers a full cluster restart with brief downtime. -However, the Stackable image version can be updated in a rolling manner, provided the NiFi version remains unchanged. - -Since NiFi version 2, rolling upgrades are supported. -==== - -== NiFi 2.0.0 - -Before you can upgrade to `2.0.0` you https://cwiki.apache.org/confluence/display/NIFI/Migration+Guidance[need to update] to at least version 1.27.x! +Rolling upgrades are supported, so changing the NiFi version in the CRD updates the cluster without downtime. diff --git a/extra/crds.yaml b/extra/crds.yaml index 08fd7441..73720bae 100644 --- a/extra/crds.yaml +++ b/extra/crds.yaml @@ -139,33 +139,6 @@ spec: - accessPolicyProvider type: object type: object - createReportingTaskJob: - default: - enabled: true - podOverrides: {} - description: |- - This section creates a `create-reporting-task` Kubernetes Job, which enables the export of - Prometheus metrics within NiFi. - properties: - enabled: - default: true - description: |- - Whether the Kubernetes Job should be created, defaults to true. It can be helpful to disable - the Job, e.g. when you configOverride an authentication mechanism, which the Job currently - can't use to authenticate against NiFi. - type: boolean - podOverrides: - default: {} - description: |- - Here you can define a - [PodTemplateSpec](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.27/#podtemplatespec-v1-core) - to override any property that can be set on the Pod of the create-reporting-task Kubernetes Job. - Read the - [Pod overrides documentation](https://docs.stackable.tech/home/nightly/concepts/overrides#pod-overrides) - for more information. - type: object - x-kubernetes-preserve-unknown-fields: true - type: object customComponentsGitSync: default: [] description: |- @@ -347,26 +320,11 @@ spec: `nifiPbkdf2AesGcm256` (the default value), `nifiArgon2AesGcm256`, - The following algorithms are deprecated and will be removed in future versions: - - `nifiArgon2AesGcm128`, - `nifiBcryptAesGcm128`, - `nifiBcryptAesGcm256`, - `nifiPbkdf2AesGcm128`, - `nifiScryptAesGcm128`, - `nifiScryptAesGcm256`. - Learn more about the specifics of the algorithm parameters in the [NiFi documentation](https://nifi.apache.org/docs/nifi-docs/html/administration-guide.html#property-encryption-algorithms). enum: - nifiPbkdf2AesGcm256 - nifiArgon2AesGcm256 - - nifiBcryptAesGcm128 - - nifiBcryptAesGcm256 - - nifiPbkdf2AesGcm128 - - nifiArgon2AesGcm128 - - nifiScryptAesGcm128 - - nifiScryptAesGcm256 - null nullable: true type: string @@ -422,8 +380,7 @@ spec: When using the [Stackable operator for Apache ZooKeeper](https://docs.stackable.tech/home/nightly/zookeeper/) to deploy a ZooKeeper cluster, this will simply be the name of your ZookeeperCluster resource. - The Kubernetes provider will be used if this field is unset. Kubernetes is only supported for NiFi 2.x and newer, - NiFi 1.x requires ZooKeeper. + The Kubernetes provider will be used if this field is unset. maxLength: 253 minLength: 1 pattern: ^[a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*$ diff --git a/rust/operator-binary/src/controller/build.rs b/rust/operator-binary/src/controller/build.rs index f636a911..8c713a32 100644 --- a/rust/operator-binary/src/controller/build.rs +++ b/rust/operator-binary/src/controller/build.rs @@ -23,8 +23,6 @@ pub const PROTOCOL_PORT_NAME: &str = "protocol"; pub const PROTOCOL_PORT: Port = Port(9088); pub const BALANCE_PORT_NAME: &str = "balance"; pub const BALANCE_PORT: Port = Port(6243); -pub const METRICS_PORT_NAME: &str = "metrics"; -pub const METRICS_PORT: Port = Port(8081); // Filesystem paths shared by multiple builders. Single-consumer paths live in their builder. pub const NIFI_CONFIG_DIRECTORY: &str = "/stackable/nifi/conf"; diff --git a/rust/operator-binary/src/controller/build/properties.rs b/rust/operator-binary/src/controller/build/properties.rs index b5d54c92..a03c6c5b 100644 --- a/rust/operator-binary/src/controller/build/properties.rs +++ b/rust/operator-binary/src/controller/build/properties.rs @@ -94,8 +94,8 @@ pub(crate) mod test_support { use crate::{ controller::{ - NifiRoleGroupConfig, ValidatedCluster, ValidatedClusterConfig, ValidatedReportingTask, - ValidatedRoleConfig, ValidatedSensitiveProperties, validate::build_role_group_configs, + NifiRoleGroupConfig, ValidatedCluster, ValidatedClusterConfig, ValidatedRoleConfig, + ValidatedSensitiveProperties, validate::build_role_group_configs, }, crd::{NifiRole, v1alpha1}, security::{ @@ -105,8 +105,7 @@ pub(crate) mod test_support { }; /// A minimal NiFi cluster YAML. Mirrors the fixture used by bootstrap_conf tests, - /// stripped down to the mandatory fields only (NiFi 2.x, Kubernetes clustering backend, - /// SingleUser auth). + /// stripped down to the mandatory fields only (Kubernetes clustering backend, SingleUser auth). pub const MINIMAL_NIFI_YAML: &str = r#" apiVersion: nifi.stackable.tech/v1alpha1 kind: NifiCluster @@ -196,15 +195,6 @@ pub(crate) mod test_support { }, server_tls_secret_class: nifi.server_tls_secret_class().clone(), extra_volumes: nifi.spec.cluster_config.extra_volumes.clone(), - reporting_task: ValidatedReportingTask { - enabled: nifi.spec.cluster_config.create_reporting_task_job.enabled, - pod_overrides: nifi - .spec - .cluster_config - .create_reporting_task_job - .pod_overrides - .clone(), - }, host_header_check: nifi.spec.cluster_config.host_header_check.clone(), }, ) diff --git a/rust/operator-binary/src/controller/build/properties/nifi_properties.rs b/rust/operator-binary/src/controller/build/properties/nifi_properties.rs index 2d854879..283d6f10 100644 --- a/rust/operator-binary/src/controller/build/properties/nifi_properties.rs +++ b/rust/operator-binary/src/controller/build/properties/nifi_properties.rs @@ -2,7 +2,7 @@ use std::collections::BTreeMap; -use snafu::{ResultExt, Snafu, ensure}; +use snafu::{ResultExt, Snafu}; use stackable_operator::{ memory::MemoryQuantity, role_utils::{ZeroReplicasCounting, fixed_replica_count}, @@ -51,11 +51,6 @@ pub enum Error { GenerateOidcConfig { source: crate::security::oidc::Error, }, - - #[snafu(display( - "NiFi 1.x requires ZooKeeper (hint: upgrade to NiFi 2.x or set .spec.clusterConfig.zookeeperConfigMapName)" - ))] - Nifi1RequiresZookeeper, } /// NiFi Python (`nipy`) extension directories, mounted only by the `nifi.properties` builder. @@ -72,36 +67,15 @@ pub fn build( proxy_hosts: &str, ) -> Result { let git_sync_resources = &rg.config.git_sync_resources; - let product_version = &cluster.image.product_version; let auth_config = &cluster.cluster_config.authentication; let resource_config = &rg.config.resources; - // TODO: Remove once we dropped support for all NiFi 1.x versions - let is_nifi_1 = product_version.starts_with("1."); - let mut properties = BTreeMap::new(); // Core Properties - // According to https://cwiki.apache.org/confluence/display/NIFI/Migration+Guidance#MigrationGuidance-Migratingto2.0.0-M1 - // The nifi.flow.configuration.file property in nifi.properties must be changed to reference - // "flow.json.gz" instead of "flow.xml.gz" - // TODO: Remove once we dropped support for all 1.x.x versions - // TODO(malte): In order to use CLI tools like: ./bin/nifi.sh set-sensitive-properties-algorithm NIFI_PBKDF2_AES_GCM_256 - // we have to set both "nifi.flow.configuration.file" and "nifi.flow.configuration.json.file" in NiFi 1.x.x. - if is_nifi_1 { - properties.insert( - "nifi.flow.configuration.file".to_string(), - NifiRepository::Database.mount_path() + "/flow.xml.gz", - ); - properties.insert( - "nifi.flow.configuration.json.file".to_string(), - NifiRepository::Database.mount_path() + "/flow.json.gz", - ); - } else { - properties.insert( - "nifi.flow.configuration.file".to_string(), - NifiRepository::Database.mount_path() + "/flow.json.gz", - ); - } + properties.insert( + "nifi.flow.configuration.file".to_string(), + NifiRepository::Database.mount_path() + "/flow.json.gz", + ); properties.insert( "nifi.flow.configuration.archive.enabled".to_string(), @@ -454,7 +428,6 @@ pub fn build( "".to_string(), ); - // The algorithm has already been validated in the validate step (check_for_nifi_version). properties.insert( "nifi.sensitive.props.algorithm".to_string(), cluster @@ -572,8 +545,6 @@ pub fn build( } v1alpha1::NifiClusteringBackend::Kubernetes {} => { - ensure!(!is_nifi_1, Nifi1RequiresZookeeperSnafu); - properties.insert( "nifi.cluster.leader.election.implementation".to_string(), "KubernetesLeaderElectionManager".to_string(), @@ -590,9 +561,6 @@ pub fn build( //#################### // Custom components # //#################### - // NiFi 1.x does not support Python components and the Python configuration below is just - // ignored. - // The command used to launch Python. // This property must be set to enable Python-based processors. properties.insert("nifi.python.command".to_string(), "python3".to_string()); @@ -807,53 +775,6 @@ mod tests { ); } - /// NiFi 1.x cannot use the Kubernetes clustering backend, so building must fail. - #[test] - fn kubernetes_backend_is_rejected_on_nifi_1() { - let mut cluster = minimal_validated_cluster(); - cluster.image.product_version = "1.27.0".to_string(); - let rg = default_rg(&cluster); - - let error = build(&cluster, rg, "*") - .expect_err("NiFi 1.x with the Kubernetes backend must be rejected"); - - assert!(matches!(error, Error::Nifi1RequiresZookeeper)); - } - - /// NiFi 1.x with the ZooKeeper backend is valid. - #[test] - fn zookeeper_backend_is_allowed_on_nifi_1() { - let mut cluster = minimal_validated_cluster(); - cluster.image.product_version = "1.27.0".to_string(); - cluster.cluster_config.clustering_backend = v1alpha1::NifiClusteringBackend::ZooKeeper { - zookeeper_config_map_name: "my-zk".parse().expect("valid ConfigMap name"), - }; - let rg = default_rg(&cluster); - - assert!(build(&cluster, rg, "*").is_ok()); - } - - /// NiFi 2.x configures only the JSON flow file; NiFi 1.x configures both the XML and JSON files. - #[test] - fn flow_configuration_files_differ_between_nifi_versions() { - let cluster_2 = minimal_validated_cluster(); // NiFi 2.x - let props_2 = build(&cluster_2, default_rg(&cluster_2), "*").expect("build should succeed"); - assert!(props_2.contains("nifi.flow.configuration.file=")); - assert!(props_2.contains("flow.json.gz")); - assert!(!props_2.contains("flow.xml.gz")); - assert!(!props_2.contains("nifi.flow.configuration.json.file=")); - - let mut cluster_1 = minimal_validated_cluster(); - cluster_1.image.product_version = "1.27.0".to_string(); - // NiFi 1.x must use the ZooKeeper backend. - cluster_1.cluster_config.clustering_backend = v1alpha1::NifiClusteringBackend::ZooKeeper { - zookeeper_config_map_name: "my-zk".parse().expect("valid ConfigMap name"), - }; - let props_1 = build(&cluster_1, default_rg(&cluster_1), "*").expect("build should succeed"); - assert!(props_1.contains("flow.xml.gz")); - assert!(props_1.contains("nifi.flow.configuration.json.file=")); - } - /// The flow archive max storage is the flowfile-repo capacity scaled by the 0.9 utilization /// factor (default capacity 1024Mi -> ~921 MB). #[test] diff --git a/rust/operator-binary/src/controller/build/proxy_hosts.rs b/rust/operator-binary/src/controller/build/proxy_hosts.rs index 2832264c..33749388 100644 --- a/rust/operator-binary/src/controller/build/proxy_hosts.rs +++ b/rust/operator-binary/src/controller/build/proxy_hosts.rs @@ -2,17 +2,11 @@ use std::collections::HashSet; -use stackable_operator::utils::cluster_info::KubernetesClusterInfo; - use crate::controller::{ ValidatedCluster, build::{ - HTTPS_PORT, properties::env_reference, - resource::{ - reporting_task, - statefulset::{LISTENER_DEFAULT_ADDRESS_ENV, LISTENER_DEFAULT_PORT_HTTPS_ENV}, - }, + resource::statefulset::{LISTENER_DEFAULT_ADDRESS_ENV, LISTENER_DEFAULT_PORT_HTTPS_ENV}, }, }; @@ -21,10 +15,7 @@ use crate::controller::{ /// The proxy hosts allow-list lets external users access NiFi via addresses we cannot predict, /// so all of them are added to the setting. For more information see /// -pub fn compute_proxy_hosts( - cluster: &ValidatedCluster, - cluster_info: &KubernetesClusterInfo, -) -> String { +pub fn compute_proxy_hosts(cluster: &ValidatedCluster) -> String { let host_header_check = &cluster.cluster_config.host_header_check; if host_header_check.allow_all { @@ -47,17 +38,6 @@ pub fn compute_proxy_hosts( )]); proxy_hosts.extend(host_header_check.additional_allowed_hosts.iter().cloned()); - // Reporting task only exists for NiFi 1.x - if cluster.image.product_version.starts_with("1.") { - let reporting_task_service_name = reporting_task::build_reporting_task_fqdn_service_name( - cluster.name.as_ref(), - &cluster.namespace, - cluster_info, - ); - - proxy_hosts.insert(format!("{reporting_task_service_name}:{HTTPS_PORT}")); - } - let mut proxy_hosts = Vec::from_iter(proxy_hosts); proxy_hosts.sort(); @@ -66,21 +46,9 @@ pub fn compute_proxy_hosts( #[cfg(test)] mod tests { - use std::str::FromStr; - - use stackable_operator::{ - commons::networking::DomainName, utils::cluster_info::KubernetesClusterInfo, - }; - use super::compute_proxy_hosts; use crate::controller::build::properties::test_support::minimal_validated_cluster; - fn cluster_info() -> KubernetesClusterInfo { - KubernetesClusterInfo { - cluster_domain: DomainName::from_str("cluster.local").expect("valid cluster domain"), - } - } - /// `allow_all` short-circuits to `*` and ignores any additional allowed hosts. #[test] fn allow_all_returns_wildcard_and_ignores_additional_hosts() { @@ -91,13 +59,13 @@ mod tests { .host_header_check .additional_allowed_hosts = vec!["ignored.example.com".to_string()]; - assert_eq!(compute_proxy_hosts(&cluster, &cluster_info()), "*"); + assert_eq!(compute_proxy_hosts(&cluster), "*"); } /// Without `allow_all`, the listener placeholder is always present and user-provided hosts are - /// merged in, sorted and comma-joined. On NiFi 2.x no reporting-task FQDN is added. + /// merged in, sorted and comma-joined. #[test] - fn explicit_hosts_include_listener_placeholder_on_nifi_2() { + fn explicit_hosts_include_listener_placeholder() { let mut cluster = minimal_validated_cluster(); cluster.cluster_config.host_header_check.allow_all = false; cluster @@ -105,7 +73,7 @@ mod tests { .host_header_check .additional_allowed_hosts = vec!["extra.example.com".to_string()]; - let hosts = compute_proxy_hosts(&cluster, &cluster_info()); + let hosts = compute_proxy_hosts(&cluster); assert!( hosts.contains("${env:LISTENER_DEFAULT_ADDRESS}:${env:LISTENER_DEFAULT_PORT_HTTPS}"), @@ -115,25 +83,5 @@ mod tests { hosts.contains("extra.example.com"), "expected the additional host in {hosts}" ); - // No reporting-task FQDN on NiFi 2.x. - assert!( - !hosts.contains("cluster.local"), - "did not expect a reporting-task FQDN on NiFi 2.x in {hosts}" - ); - } - - /// On NiFi 1.x the reporting-task FQDN (which embeds the cluster domain) is added. - #[test] - fn nifi_1_adds_reporting_task_fqdn() { - let mut cluster = minimal_validated_cluster(); - cluster.cluster_config.host_header_check.allow_all = false; - cluster.image.product_version = "1.27.0".to_string(); - - let hosts = compute_proxy_hosts(&cluster, &cluster_info()); - - assert!( - hosts.contains("cluster.local"), - "expected the reporting-task FQDN on NiFi 1.x in {hosts}" - ); } } diff --git a/rust/operator-binary/src/controller/build/resource/config_map.rs b/rust/operator-binary/src/controller/build/resource/config_map.rs index 1d93895b..d402b96a 100644 --- a/rust/operator-binary/src/controller/build/resource/config_map.rs +++ b/rust/operator-binary/src/controller/build/resource/config_map.rs @@ -3,8 +3,7 @@ use snafu::{ResultExt, Snafu}; use stackable_operator::{ builder::configmap::ConfigMapBuilder, k8s_openapi::api::core::v1::ConfigMap, - product_logging::framework::VECTOR_CONFIG_FILE, utils::cluster_info::KubernetesClusterInfo, - v2::types::operator::RoleGroupName, + product_logging::framework::VECTOR_CONFIG_FILE, v2::types::operator::RoleGroupName, }; use crate::controller::{ @@ -58,11 +57,10 @@ pub fn build_rolegroup_config_map( cluster: &ValidatedCluster, role_group_name: &RoleGroupName, rg: &NifiRoleGroupConfig, - cluster_info: &KubernetesClusterInfo, ) -> Result { tracing::debug!("building rolegroup ConfigMap"); - let proxy_hosts = proxy_hosts::compute_proxy_hosts(cluster, cluster_info); + let proxy_hosts = proxy_hosts::compute_proxy_hosts(cluster); let mut cm_builder = ConfigMapBuilder::new(); diff --git a/rust/operator-binary/src/controller/build/resource/mod.rs b/rust/operator-binary/src/controller/build/resource/mod.rs index 43b3b825..d040825b 100644 --- a/rust/operator-binary/src/controller/build/resource/mod.rs +++ b/rust/operator-binary/src/controller/build/resource/mod.rs @@ -5,6 +5,5 @@ pub mod config_map; pub mod listener; pub mod pdb; -pub mod reporting_task; pub mod service; pub mod statefulset; diff --git a/rust/operator-binary/src/controller/build/resource/reporting_task.rs b/rust/operator-binary/src/controller/build/resource/reporting_task.rs deleted file mode 100644 index 76ab68b4..00000000 --- a/rust/operator-binary/src/controller/build/resource/reporting_task.rs +++ /dev/null @@ -1,342 +0,0 @@ -//! The NiFi Reporting Task for Prometheus metrics is created via the NiFi Rest API. -//! -//! This module contains methods to create all required resources (Job, Service) and helper methods -//! to create the Prometheus Reporting Task. -//! -//! Before NiFi 1.25.0 there was only the actual Kubernetes Job required to create and run the Reporting Task Job. -//! -//! Due to changes in the JWT validation in 1.25.0, the issuer refers to the FQDN of the Pod that was created, e.g.: -//! { -//! "sub": "admin", -//! "iss": "test-nifi-node-default-0.test-nifi-node-default-headless.default.svc.cluster.local:8443", -//! } -//! which was different in e.g. 1.23.2 -//! { -//! "sub": "admin", -//! "iss": "SingleUserLoginIdentityProvider", -//! } -//! This caused problems when using the generated JWT against a different node (due to randomness of the service). -//! -//! "An error occurred while attempting to decode the Jwt: Signed JWT rejected: Another algorithm expected, or no matching key(s) found" -//! -//! Therefore, since the support of NiFi 1.25.0, an additional service for the Reporting Task Job containing a -//! random but deterministic NiFi node to ensure the communication with a single node. -//! -use std::{collections::BTreeMap, str::FromStr as _}; - -use snafu::{OptionExt, ResultExt, Snafu}; -use stackable_operator::{ - builder::{ - self, - meta::ObjectMetaBuilder, - pod::{ - PodBuilder, resources::ResourceRequirementsBuilder, - security::PodSecurityContextBuilder, volume::SecretFormat, - }, - }, - k8s_openapi::{ - DeepMerge, - api::{ - batch::v1::{Job, JobSpec}, - core::v1::{Service, ServicePort, ServiceSpec}, - }, - }, - shared::time::Duration, - utils::cluster_info::KubernetesClusterInfo, - v2::{ - builder::{meta::ownerreference_from_resource, pod::container::new_container_builder}, - types::{ - kubernetes::{ContainerName, NamespaceName, VolumeName}, - operator::RoleGroupName, - }, - }, -}; - -use crate::{ - controller::{ - ValidatedCluster, - build::{HTTPS_PORT, HTTPS_PORT_NAME, METRICS_PORT}, - }, - crd::NifiRole, - security::{authentication::STACKABLE_ADMIN_USERNAME, build_tls_volume}, -}; - -stackable_operator::constant!(REPORTING_TASK_CERT_VOLUME_NAME: VolumeName = "tls"); -const REPORTING_TASK_CERT_VOLUME_MOUNT: &str = "/stackable/cert"; -/// Path (inside the image) of the script that registers the NiFi 1.x reporting task. -const REPORTING_TASK_SCRIPT_PATH: &str = "/stackable/python/create_nifi_reporting_task.py"; -stackable_operator::constant!(REPORTING_TASK_CONTAINER_NAME: ContainerName = "reporting-task"); - -#[derive(Snafu, Debug)] -pub enum Error { - #[snafu(display("failed to add Authentication Volumes and VolumeMounts"))] - AddAuthVolumes { - source: crate::security::authentication::Error, - }, - - #[snafu(display("failed to build secret volume"))] - SecretVolumeBuildFailure { source: crate::security::Error }, - - #[snafu(display("failed to create reporting task service, no role groups defined"))] - FailedBuildReportingTaskService, - - #[snafu(display("failed to add needed volume"))] - AddVolume { source: builder::pod::Error }, - - #[snafu(display("failed to add needed volumeMount"))] - AddVolumeMount { - source: builder::pod::container::Error, - }, -} - -type Result = std::result::Result; - -/// Build required resources to create the reporting task in NiFi versions 1.x. -/// -/// This will return -/// * a Job that creates and runs the reporting task via the NiFi Rest API. -/// * a Service that contains of one single NiFi node. -/// -/// The Service is required in order to communicate only with one designated NiFi node. -/// This is necessary as the generated JWT was changed in 1.25.0 and corrected the issuer -/// from SingleUserLoginIdentityProvider to the FQDN of the pod. -/// The NiFi role service will randomly delegate to different NiFi nodes which will -/// then fail requests to other nodes. -/// -/// NiFi 2.x and above automatically server Prometheus metrics via the API, but as of 2024-11-08 -/// requires authentication. -pub fn build_maybe_reporting_task( - cluster: &ValidatedCluster, - cluster_info: &KubernetesClusterInfo, - sa_name: &str, -) -> Result> { - if cluster.image.product_version.starts_with("1.") { - Ok(Some(( - build_reporting_task_job(cluster, cluster_info, sa_name)?, - build_reporting_task_service(cluster)?, - ))) - } else { - Ok(None) - } -} - -/// The placeholder role-group name (`global`) used for the labels of the cluster-global reporting -/// task resources, which are not tied to a specific role group. -fn reporting_task_role_group() -> RoleGroupName { - RoleGroupName::from_str("global").expect("'global' is a valid role-group name") -} - -/// Return the name of the reporting task Service. -pub fn build_reporting_task_service_name(nifi_cluster_name: &str) -> String { - format!( - "{nifi_cluster_name}-{container}", - container = &*REPORTING_TASK_CONTAINER_NAME - ) -} - -/// Return the FQDN (with namespace, domain) of the reporting task. -pub fn build_reporting_task_fqdn_service_name( - cluster_name: &str, - namespace: &NamespaceName, - cluster_info: &KubernetesClusterInfo, -) -> String { - let reporting_task_service_name = build_reporting_task_service_name(cluster_name); - let cluster_domain = &cluster_info.cluster_domain; - format!("{reporting_task_service_name}.{namespace}.svc.{cluster_domain}") -} - -/// Return the name of the first pod belonging to the first role group that contains more than 0 replicas. -/// If no role group has replicas set (e.g. HPA, see ) -/// return the first rolegroup just in case. -/// This is required to only select a single node in the Reporting Task Service. -fn get_reporting_task_service_selector_pod(cluster: &ValidatedCluster) -> Result { - let node_name = NifiRole::Node.to_string(); - - // The role groups are already sorted by name (`BTreeMap`), avoiding random ordering and - // therefore unnecessary reconciles. - let role_groups = cluster - .role_group_configs - .get(&NifiRole::Node) - .context(FailedBuildReportingTaskServiceSnafu)?; - - let mut selector_role_group = None; - for (role_group_name, role_group) in role_groups { - // just pick the first rolegroup in case none has replicas set - if selector_role_group.is_none() { - selector_role_group = Some(role_group_name); - } - - if role_group.replicas.unwrap_or(1) > 0 { - selector_role_group = Some(role_group_name); - break; - } - } - - Ok(format!( - "{cluster_name}-{node_name}-{role_group_name}-0", - cluster_name = cluster.name, - role_group_name = selector_role_group.context(FailedBuildReportingTaskServiceSnafu)? - )) -} - -/// Build the internal Reporting Task Service in order to communicate with a single NiFi node. -fn build_reporting_task_service(cluster: &ValidatedCluster) -> Result { - let nifi_cluster_name = cluster.name.to_string(); - let mut selector: BTreeMap = cluster.role_selector().into(); - - let service_selector_pod = get_reporting_task_service_selector_pod(cluster)?; - selector.insert( - "statefulset.kubernetes.io/pod-name".to_string(), - service_selector_pod, - ); - - Ok(Service { - metadata: cluster - .object_meta( - build_reporting_task_service_name(&nifi_cluster_name), - &reporting_task_role_group(), - ) - .build(), - spec: Some(ServiceSpec { - ports: Some(vec![ServicePort { - name: Some(HTTPS_PORT_NAME.to_string()), - port: HTTPS_PORT.into(), - protocol: Some("TCP".to_string()), - ..ServicePort::default() - }]), - selector: Some(selector), - ..ServiceSpec::default() - }), - status: None, - }) -} - -/// Build the [`Job`](`stackable_operator::k8s_openapi::api::batch::v1::Job`) that creates a -/// NiFi `ReportingTask` in order to enable JVM and NiFi metrics. -/// -/// The Job is run via the [`tools`](https://github.com/stackabletech/docker-images/tree/main/tools) -/// docker image and more specifically the `create_nifi_reporting_task.py` Python script. -/// -/// This script uses the [`nipyapi`](https://nipyapi.readthedocs.io/en/latest/readme.html) -/// library to authenticate and run the required REST calls to the NiFi REST API. -/// -/// In order to authenticate we need the `username` and `password` from the -/// [`NifiAuthenticationConfig`](`crate::security::authentication::NifiAuthenticationConfig`) -/// as well as a public certificate provided by the Stackable -/// [`secret-operator`](https://github.com/stackabletech/secret-operator) -/// -fn build_reporting_task_job( - cluster: &ValidatedCluster, - cluster_info: &KubernetesClusterInfo, - sa_name: &str, -) -> Result { - let resolved_product_image = &cluster.image; - let nifi_auth_config = &cluster.cluster_config.authentication; - let reporting_task_fqdn_service_name = build_reporting_task_fqdn_service_name( - cluster.name.as_ref(), - &cluster.namespace, - cluster_info, - ); - let product_version = &resolved_product_image.product_version; - let nifi_connect_url = - format!("https://{reporting_task_fqdn_service_name}:{HTTPS_PORT}/nifi-api",); - - let (admin_username_file, admin_password_file) = - nifi_auth_config.get_user_and_password_file_paths(); - - let user_name_command = if admin_username_file.is_empty() { - // In case of the username being simple (e.g admin for SingleUser) just use it as is - format!("-u {STACKABLE_ADMIN_USERNAME}") - } else { - // If the username is a bind dn (e.g. cn=integrationtest,ou=my users,dc=example,dc=org) we have to extract the cn/dn/uid (in this case integrationtest) - format!( - "-u \"$(cat {admin_username_file} | grep -oP '((cn|dn|uid)=\\K[^,]+|.*)' | head -n 1)\"" - ) - }; - - let args = [ - REPORTING_TASK_SCRIPT_PATH.to_string(), - format!("-n {nifi_connect_url}"), - user_name_command, - format!("-p \"$(cat {admin_password_file})\""), - format!("-m {METRICS_PORT}"), - format!("-c {REPORTING_TASK_CERT_VOLUME_MOUNT}/ca.crt"), - ]; - let mut cb = new_container_builder(&REPORTING_TASK_CONTAINER_NAME); - cb.image_from_product_image(resolved_product_image) - .command(vec!["sh".to_string(), "-c".to_string()]) - .args(vec![args.join(" ")]) - // The VolumeMount for the secret operator key store certificates - .add_volume_mount( - REPORTING_TASK_CERT_VOLUME_NAME.to_string(), - REPORTING_TASK_CERT_VOLUME_MOUNT, - ) - .context(AddVolumeMountSnafu)? - .resources( - ResourceRequirementsBuilder::new() - .with_cpu_request("100m") - .with_cpu_limit("400m") - .with_memory_request("512Mi") - .with_memory_limit("512Mi") - .build(), - ); - - let job_name = format!( - "{}-create-reporting-task-{}", - cluster.name, - product_version.replace('.', "-").to_ascii_lowercase() - ); - - let mut pb = PodBuilder::new(); - nifi_auth_config - .add_volumes_and_mounts(&mut pb, vec![&mut cb]) - .context(AddAuthVolumesSnafu)?; - - let mut pod_template = pb - .metadata( - ObjectMetaBuilder::new() - .name_and_namespace(cluster) - .name(job_name.clone()) - .ownerreference(ownerreference_from_resource(cluster, None, Some(true))) - .build(), - ) - .image_pull_secrets_from_product_image(resolved_product_image) - .restart_policy("OnFailure") - .service_account_name(sa_name) - .security_context(PodSecurityContextBuilder::new().fs_group(1000).build()) - .add_container(cb.build()) - .add_volume( - build_tls_volume( - &cluster.cluster_config.server_tls_secret_class, - &REPORTING_TASK_CERT_VOLUME_NAME, - Vec::::new(), - SecretFormat::TlsPem, - // The certificate is only used for the REST API call, so a short lifetime is sufficient. - // There is no correct way to configure this job since it's an implementation detail. - // Also it will be dropped when support for 1.x is removed. - &Duration::from_days_unchecked(1), - // There is no listener volume we could get certs for - None, - ) - .context(SecretVolumeBuildFailureSnafu)?, - ) - .context(AddVolumeSnafu)? - .build_template(); - - pod_template.merge_from(cluster.cluster_config.reporting_task.pod_overrides.clone()); - - let job = Job { - metadata: cluster - .object_meta(job_name, &reporting_task_role_group()) - .build(), - spec: Some(JobSpec { - backoff_limit: Some(100), - ttl_seconds_after_finished: Some(120), - template: pod_template, - ..JobSpec::default() - }), - ..Job::default() - }; - - Ok(job) -} diff --git a/rust/operator-binary/src/controller/build/resource/service.rs b/rust/operator-binary/src/controller/build/resource/service.rs index 9e30816b..e2c6947e 100644 --- a/rust/operator-binary/src/controller/build/resource/service.rs +++ b/rust/operator-binary/src/controller/build/resource/service.rs @@ -9,7 +9,7 @@ use stackable_operator::{ use crate::controller::{ ValidatedCluster, - build::{HTTPS_PORT, HTTPS_PORT_NAME, METRICS_PORT, METRICS_PORT_NAME}, + build::{HTTPS_PORT, HTTPS_PORT_NAME}, }; /// The rolegroup headless [`Service`] is a service that allows direct access to the instances of a certain rolegroup @@ -32,7 +32,7 @@ pub fn build_rolegroup_headless_service( // Internal communication does not need to be exposed type_: Some("ClusterIP".to_string()), cluster_ip: Some("None".to_string()), - ports: Some(headless_service_ports()), + ports: Some(vec![headless_service_port()]), selector: Some(cluster.role_group_selector(role_group_name).into()), publish_not_ready_addresses: Some(true), ..ServiceSpec::default() @@ -46,7 +46,6 @@ pub fn build_rolegroup_metrics_service( cluster: &ValidatedCluster, role_group_name: &RoleGroupName, ) -> Service { - let product_version = &cluster.image.product_version; Service { metadata: cluster .object_meta( @@ -57,13 +56,13 @@ pub fn build_rolegroup_metrics_service( role_group_name, ) .with_labels(service::prometheus_labels(&Scraping::Enabled)) - .with_annotations(prometheus_annotations(product_version)) + .with_annotations(prometheus_annotations()) .build(), spec: Some(ServiceSpec { // Internal communication does not need to be exposed type_: Some("ClusterIP".to_string()), cluster_ip: Some("None".to_string()), - ports: Some(vec![metrics_service_port(product_version)]), + ports: Some(vec![metrics_service_port()]), selector: Some(cluster.role_group_selector(role_group_name).into()), publish_not_ready_addresses: Some(true), ..ServiceSpec::default() @@ -72,54 +71,31 @@ pub fn build_rolegroup_metrics_service( } } -fn headless_service_ports() -> Vec { - vec![ServicePort { +/// Returns the headless Service port, which exposes the HTTPS UI. +fn headless_service_port() -> ServicePort { + ServicePort { name: Some(HTTPS_PORT_NAME.into()), port: HTTPS_PORT.into(), protocol: Some("TCP".to_string()), ..ServicePort::default() - }] + } } -/// Returns the metrics port based on the NiFi version. -/// -/// NiFi 1.x exposes a dedicated metrics port via the JMX exporter; NiFi 2.x serves metrics on the -/// NiFi HTTP(S) port. -pub fn metrics_service_port(product_version: &str) -> ServicePort { - if product_version.starts_with("1.") { - ServicePort { - name: Some(METRICS_PORT_NAME.to_string()), - port: METRICS_PORT.into(), - protocol: Some("TCP".to_string()), - ..ServicePort::default() - } - } else { - ServicePort { - name: Some(HTTPS_PORT_NAME.into()), - port: HTTPS_PORT.into(), - protocol: Some("TCP".to_string()), - ..ServicePort::default() - } - } +/// Returns the metrics port, which is the same as the headless Service port +pub fn metrics_service_port() -> ServicePort { + headless_service_port() } /// Common annotations for Prometheus /// /// These annotations can be used in a ServiceMonitor. -fn prometheus_annotations(product_version: &str) -> Annotations { - // NiFi 1.x exposes metrics via the JMX exporter over HTTP on a dedicated port; NiFi 2.x serves - // them over HTTPS on the NiFi API port. - let (scheme, path, port) = if product_version.starts_with("1.") { - (Scheme::Http, "/metrics", METRICS_PORT) - } else { - ( - Scheme::Https, - "/nifi-api/flow/metrics/prometheus", - HTTPS_PORT, - ) - }; - - service::prometheus_annotations(&Scraping::Enabled, &scheme, path, &port) +fn prometheus_annotations() -> Annotations { + service::prometheus_annotations( + &Scraping::Enabled, + &Scheme::Https, + "/nifi-api/flow/metrics/prometheus", + &HTTPS_PORT, + ) } #[cfg(test)] @@ -127,27 +103,10 @@ mod tests { use std::str::FromStr as _; use pretty_assertions::assert_eq; - use rstest::rstest; - use stackable_operator::v2::types::common::Port; use super::*; use crate::controller::build::properties::test_support::minimal_validated_cluster; - #[rstest] - // NiFi 1.x exposes metrics on a dedicated JMX-exporter port ... - #[case("1.28.1", METRICS_PORT_NAME, METRICS_PORT)] - // ... while NiFi 2.x serves them on the HTTPS port. - #[case("2.9.0", HTTPS_PORT_NAME, HTTPS_PORT)] - fn metrics_service_port_depends_on_version( - #[case] product_version: &str, - #[case] expected_name: &str, - #[case] expected_port: Port, - ) { - let port = metrics_service_port(product_version); - assert_eq!(Some(expected_name.to_string()), port.name); - assert_eq!(i32::from(expected_port), port.port); - } - #[test] fn headless_service_is_cluster_ip_none_with_https_port() { let cluster = minimal_validated_cluster(); diff --git a/rust/operator-binary/src/controller/build/resource/statefulset.rs b/rust/operator-binary/src/controller/build/resource/statefulset.rs index 49312718..64bd3b37 100644 --- a/rust/operator-binary/src/controller/build/resource/statefulset.rs +++ b/rust/operator-binary/src/controller/build/resource/statefulset.rs @@ -49,17 +49,13 @@ use crate::{ controller::{ NifiRoleGroupConfig, ValidatedCluster, build::{ - BALANCE_PORT, BALANCE_PORT_NAME, HTTPS_PORT, HTTPS_PORT_NAME, METRICS_PORT, - METRICS_PORT_NAME, NIFI_CONFIG_DIRECTORY, NIFI_PYTHON_WORKING_DIRECTORY, PROTOCOL_PORT, - PROTOCOL_PORT_NAME, + BALANCE_PORT, BALANCE_PORT_NAME, HTTPS_PORT, HTTPS_PORT_NAME, NIFI_CONFIG_DIRECTORY, + NIFI_PYTHON_WORKING_DIRECTORY, PROTOCOL_PORT, PROTOCOL_PORT_NAME, graceful_shutdown::add_graceful_shutdown_config, properties::ConfigFileName, - resource::{ - listener::{ - LISTENER_VOLUME_DIR, LISTENER_VOLUME_NAME, build_group_listener_pvc, - group_listener_name, - }, - reporting_task::build_reporting_task_service_name, + resource::listener::{ + LISTENER_VOLUME_DIR, LISTENER_VOLUME_NAME, build_group_listener_pvc, + group_listener_name, }, }, }, @@ -174,7 +170,6 @@ pub(crate) async fn build_node_rolegroup_statefulset( cluster_info: &KubernetesClusterInfo, role_group_name: &RoleGroupName, rg: &NifiRoleGroupConfig, - rolling_update_supported: bool, effective_replicas: Option, service_account_name: &str, ) -> Result { @@ -463,11 +458,6 @@ pub(crate) async fn build_node_rolegroup_statefulset( }) .resources(merged_config.resources.clone().into()); - // NiFi 2.x.x offers nifi-api/flow/metrics/prometheus at the HTTPS_PORT, therefore METRICS_PORT is only required for NiFi 1.x.x. - if resolved_product_image.product_version.starts_with("1.") { - container_nifi.add_container_port(METRICS_PORT_NAME, METRICS_PORT.into()); - } - let mut pod_builder = PodBuilder::new(); let recommended_object_labels = cluster.recommended_labels(role_group_name); @@ -573,7 +563,6 @@ pub(crate) async fn build_node_rolegroup_statefulset( let requested_secret_lifetime = merged_config .requested_secret_lifetime .context(MissingSecretLifetimeSnafu)?; - let nifi_cluster_name = cluster.name.to_string(); pod_builder .metadata(metadata) .image_pull_secrets_from_product_image(resolved_product_image) @@ -616,13 +605,10 @@ pub(crate) async fn build_node_rolegroup_statefulset( build_tls_volume( &cluster.cluster_config.server_tls_secret_class, &KEYSTORE_VOLUME_NAME, - [ - cluster - .resource_names(role_group_name) - .metrics_service_name() - .to_string(), - build_reporting_task_service_name(&nifi_cluster_name), - ], + [cluster + .resource_names(role_group_name) + .metrics_service_name() + .to_string()], SecretFormat::TlsPkcs12, &requested_secret_lifetime, Some(LISTENER_VOLUME_NAME), @@ -683,11 +669,7 @@ pub(crate) async fn build_node_rolegroup_statefulset( service_name: Some(resource_names.headless_service_name().to_string()), template: pod_template, update_strategy: Some(StatefulSetUpdateStrategy { - type_: if rolling_update_supported { - Some("RollingUpdate".to_string()) - } else { - Some("OnDelete".to_string()) - }, + type_: Some("RollingUpdate".to_string()), ..StatefulSetUpdateStrategy::default() }), volume_claim_templates: Some(get_volume_claim_templates(cluster, role_group_name, rg)?), diff --git a/rust/operator-binary/src/controller/mod.rs b/rust/operator-binary/src/controller/mod.rs index b0e7dd3d..f9106741 100644 --- a/rust/operator-binary/src/controller/mod.rs +++ b/rust/operator-binary/src/controller/mod.rs @@ -11,16 +11,13 @@ use stackable_operator::{ resources::{NoRuntimeLimits, Resources}, }, crd::git_sync, - k8s_openapi::{ - api::core::v1::{PodTemplateSpec, Volume}, - apimachinery::pkg::apis::meta::v1::ObjectMeta, - }, + k8s_openapi::{api::core::v1::Volume, apimachinery::pkg::apis::meta::v1::ObjectMeta}, kube::Resource, kvp::Labels, shared::time::Duration, v2::{ HasName, HasUid, NameIsValidLabelValue, - kvp::label::{recommended_labels, role_group_selector, role_selector}, + kvp::label::{recommended_labels, role_group_selector}, product_logging::framework::{ValidatedContainerLogConfigChoice, VectorContainerLogConfig}, role_group_utils::ResourceNames, role_utils::{JavaCommonConfig, RoleGroupConfig}, @@ -48,7 +45,6 @@ use crate::{ pub(crate) mod build; pub(crate) mod dereference; -pub(crate) mod upgrade; pub(crate) mod validate; /// A validated, merged (default <- role <- role-group) NiFi rolegroup config. @@ -176,8 +172,6 @@ pub struct ValidatedClusterConfig { pub server_tls_secret_class: SecretClassName, /// User-provided extra volumes, mounted into every container under `/stackable/userdata/`. pub extra_volumes: Vec, - /// The resolved (NiFi 1.x-only) create-reporting-task Job configuration. - pub reporting_task: ValidatedReportingTask, } /// The resolved `spec.clusterConfig.sensitiveProperties`. @@ -190,14 +184,6 @@ pub struct ValidatedSensitiveProperties { pub auto_generate: bool, } -/// The resolved `spec.clusterConfig.createReportingTaskJob` (NiFi 1.x only). -pub struct ValidatedReportingTask { - /// Whether the reporting-task Job should be created. - pub enabled: bool, - /// Pod overrides for the reporting-task Job. - pub pod_overrides: PodTemplateSpec, -} - /// Per-role configuration extracted during validation. #[derive(Clone, Debug)] pub struct ValidatedRoleConfig { @@ -290,11 +276,6 @@ impl ValidatedCluster { role_group_selector(self, &product_name(), &Self::role_name(), role_group_name) } - /// Selector labels matching all pods of the (single) NiFi role. - pub fn role_selector(&self) -> Labels { - role_selector(self, &product_name(), &Self::role_name()) - } - /// Returns an [`ObjectMetaBuilder`](stackable_operator::builder::meta::ObjectMetaBuilder) /// pre-filled with the namespace, an owner reference back to this cluster, and the recommended /// labels for a resource named `name` in `role_group_name`. diff --git a/rust/operator-binary/src/controller/upgrade.rs b/rust/operator-binary/src/controller/upgrade.rs deleted file mode 100644 index 55e7fe6f..00000000 --- a/rust/operator-binary/src/controller/upgrade.rs +++ /dev/null @@ -1,108 +0,0 @@ -// TODO: This module can be removed once we don't support NiFi 1.x versions anymore -// It manages the version upgrade procedure for NiFi versions prior to NiFi 2, since rolling upgrade is not supported there yet - -use snafu::{ResultExt, Snafu}; -use stackable_operator::{ - client::Client, - k8s_openapi::{api::apps::v1::StatefulSet, apimachinery::pkg::apis::meta::v1::LabelSelector}, - v2::types::operator::ProductVersion, -}; - -use super::ValidatedCluster; - -#[derive(Snafu, Debug)] -pub enum Error { - #[snafu(display("failed to fetch deployed StatefulSets"))] - FetchStatefulsets { - source: stackable_operator::client::Error, - }, -} - -type Result = std::result::Result; - -// This struct is used for NiFi versions not supporting rolling upgrades since in that case -// we have to manage the restart process ourselves and need to track the state of it -#[derive(Debug, PartialEq, Eq)] -pub enum ClusterVersionUpdateState { - UpdateRequested, - UpdateInProgress, - ClusterStopped, - NoVersionChange, -} - -pub async fn cluster_version_update_state( - cluster: &ValidatedCluster, - client: &Client, - deployed_version: Option<&ProductVersion>, -) -> Result { - // The version we want to converge to, i.e. the resolved product image version. - let resolved_version = &cluster.image.product_version; - - // Handle full restarts for a version change - match deployed_version { - Some(deployed_version) => { - if deployed_version.as_ref() != resolved_version.as_str() { - // Check if statefulsets are already scaled to zero, if not - requeue - let selector = LabelSelector { - match_expressions: None, - match_labels: Some(cluster.role_selector().into()), - }; - - // Retrieve the deployed statefulsets to check on the current status of the restart - let deployed_statefulsets = client - .list_with_label_selector::(cluster.namespace.as_ref(), &selector) - .await - .context(FetchStatefulsetsSnafu)?; - - // Sum target replicas for all statefulsets - let target_replicas = deployed_statefulsets - .iter() - .filter_map(|statefulset| statefulset.spec.as_ref()) - .filter_map(|spec| spec.replicas) - .sum::(); - - // Sum current ready replicas for all statefulsets - let current_replicas = deployed_statefulsets - .iter() - .filter_map(|statefulset| statefulset.status.as_ref()) - .map(|status| status.replicas) - .sum::(); - - // If statefulsets have already been scaled to zero, but have remaining replicas - // we requeue to wait until a full stop has been performed. - if target_replicas == 0 && current_replicas > 0 { - tracing::info!( - "Cluster is performing a full restart at the moment and still shutting down, remaining replicas: [{}] - requeuing to wait for shutdown to finish", - current_replicas - ); - return Ok(ClusterVersionUpdateState::UpdateInProgress); - } - - // Otherwise we either still need to scale the statefulsets to 0 or all replicas have - // been stopped and we can restart the cluster. - // Both actions will be taken in the regular reconciliation, so we can simply continue - // here - if target_replicas > 0 { - tracing::info!( - "Version change detected, we'll need to scale down the cluster for a full restart." - ); - Ok(ClusterVersionUpdateState::UpdateRequested) - } else { - tracing::info!("Cluster has been stopped for a restart, will scale back up."); - Ok(ClusterVersionUpdateState::ClusterStopped) - } - } else { - // No version change detected, propagate this to the reconciliation - Ok(ClusterVersionUpdateState::NoVersionChange) - } - } - None => { - // No deployed version set in status, this is probably the first reconciliation ever - // for this cluster, so just let it progress normally - tracing::debug!( - "No deployed version found for this cluster, this is probably the first start, continue reconciliation" - ); - Ok(ClusterVersionUpdateState::NoVersionChange) - } - } -} diff --git a/rust/operator-binary/src/controller/validate.rs b/rust/operator-binary/src/controller/validate.rs index f0e1da38..a29ed629 100644 --- a/rust/operator-binary/src/controller/validate.rs +++ b/rust/operator-binary/src/controller/validate.rs @@ -30,11 +30,11 @@ use strum::{EnumDiscriminants, IntoStaticStr}; use super::{ NifiRoleGroupConfig, ValidatedCluster, ValidatedClusterConfig, ValidatedLogging, - ValidatedNifiConfig, ValidatedReportingTask, ValidatedRoleConfig, ValidatedSensitiveProperties, + ValidatedNifiConfig, ValidatedRoleConfig, ValidatedSensitiveProperties, }; use crate::{ controller::{build::git_sync::build_git_sync_resources, dereference::DereferencedObjects}, - crd::{Container, NifiConfig, NifiRole, sensitive_properties, v1alpha1}, + crd::{Container, NifiConfig, NifiRole, v1alpha1}, security::{ authentication::{self, NifiAuthenticationConfig}, authorization::ResolvedNifiAuthorizationConfig, @@ -80,9 +80,6 @@ pub enum Error { name: String, }, - #[snafu(display("invalid sensitive properties algorithm"))] - InvalidSensitivePropertiesAlgorithm { source: sensitive_properties::Error }, - #[snafu(display("failed to build git-sync resources"))] BuildGitSyncResources { source: crate::controller::build::git_sync::Error, @@ -135,9 +132,6 @@ pub fn validate( .algorithm .clone() .unwrap_or_default(); - sensitive_properties_algorithm - .check_for_nifi_version(&image.product_version) - .context(InvalidSensitivePropertiesAlgorithmSnafu)?; // The Vector aggregator discovery ConfigMap name is validated by the CRD's typed field. It is // only required when the Vector agent is enabled for a role group. @@ -193,15 +187,6 @@ pub fn validate( }, server_tls_secret_class: nifi.server_tls_secret_class().clone(), extra_volumes: nifi.spec.cluster_config.extra_volumes.clone(), - reporting_task: ValidatedReportingTask { - enabled: nifi.spec.cluster_config.create_reporting_task_job.enabled, - pod_overrides: nifi - .spec - .cluster_config - .create_reporting_task_job - .pod_overrides - .clone(), - }, host_header_check: nifi.spec.cluster_config.host_header_check.clone(), }, )) diff --git a/rust/operator-binary/src/crd/mod.rs b/rust/operator-binary/src/crd/mod.rs index 59679e3a..1d1acf67 100644 --- a/rust/operator-binary/src/crd/mod.rs +++ b/rust/operator-binary/src/crd/mod.rs @@ -23,17 +23,14 @@ use stackable_operator::{ config::{fragment::Fragment, merge::Merge}, crd::{authentication::core as auth_core, git_sync}, deep_merger::ObjectOverrides, - k8s_openapi::{ - api::core::v1::{PodTemplateSpec, Volume}, - apimachinery::pkg::api::resource::Quantity, - }, + k8s_openapi::{api::core::v1::Volume, apimachinery::pkg::api::resource::Quantity}, kube::CustomResource, product_logging::{self, spec::Logging}, role_utils::{GenericRoleConfig, Role}, schemars::{self, JsonSchema}, shared::time::Duration, status::condition::{ClusterCondition, HasStatusCondition}, - utils::crds::{raw_object_list_schema, raw_object_schema}, + utils::crds::raw_object_list_schema, v2::{ config_overrides::KeyValueConfigOverrides, role_utils::JavaCommonConfig, @@ -145,10 +142,6 @@ pub mod versioned { #[serde(default, skip_serializing_if = "Vec::is_empty")] #[schemars(schema_with = "raw_object_list_schema")] pub extra_volumes: Vec, - - // Docs are on the struct - #[serde(default)] - pub create_reporting_task_job: CreateReportingTaskJob, } // This is flattened in for backwards compatibility reasons, `zookeeper_config_map_name` already existed and used to be mandatory. @@ -163,8 +156,7 @@ pub mod versioned { /// When using the [Stackable operator for Apache ZooKeeper](DOCS_BASE_URL_PLACEHOLDER/zookeeper/) /// to deploy a ZooKeeper cluster, this will simply be the name of your ZookeeperCluster resource. /// - /// The Kubernetes provider will be used if this field is unset. Kubernetes is only supported for NiFi 2.x and newer, - /// NiFi 1.x requires ZooKeeper. + /// The Kubernetes provider will be used if this field is unset. zookeeper_config_map_name: ConfigMapName, }, Kubernetes {}, @@ -231,43 +223,6 @@ pub fn default_allow_all() -> bool { true } -/// This section creates a `create-reporting-task` Kubernetes Job, which enables the export of -/// Prometheus metrics within NiFi. -#[derive(Clone, Debug, Deserialize, JsonSchema, PartialEq, Serialize)] -#[serde(rename_all = "camelCase")] -pub struct CreateReportingTaskJob { - /// Whether the Kubernetes Job should be created, defaults to true. It can be helpful to disable - /// the Job, e.g. when you configOverride an authentication mechanism, which the Job currently - /// can't use to authenticate against NiFi. - #[serde(default = "CreateReportingTaskJob::default_enabled")] - pub enabled: bool, - - /// Here you can define a - /// [PodTemplateSpec](https://kubernetes.io/docs/reference/generated/kubernetes-api/v1.27/#podtemplatespec-v1-core) - /// to override any property that can be set on the Pod of the create-reporting-task Kubernetes Job. - /// Read the - /// [Pod overrides documentation](DOCS_BASE_URL_PLACEHOLDER/concepts/overrides#pod-overrides) - /// for more information. - #[serde(default)] - #[schemars(schema_with = "raw_object_schema")] - pub pod_overrides: PodTemplateSpec, -} - -impl Default for CreateReportingTaskJob { - fn default() -> Self { - Self { - enabled: Self::default_enabled(), - pod_overrides: Default::default(), - } - } -} - -impl CreateReportingTaskJob { - const fn default_enabled() -> bool { - true - } -} - #[derive( Clone, Debug, Deserialize, Eq, JsonSchema, Ord, PartialEq, PartialOrd, Serialize, strum::Display, )] @@ -585,7 +540,7 @@ mod tests { fn roundtrip_test_data() -> Vec { stackable_operator::utils::yaml_from_str_singleton_map(indoc::indoc! {r#" - image: - productVersion: 1.2.3 + productVersion: 2.9.0 pullPolicy: IfNotPresent clusterOperation: stopped: false diff --git a/rust/operator-binary/src/crd/sensitive_properties.rs b/rust/operator-binary/src/crd/sensitive_properties.rs index 8c5005b0..f72aed77 100644 --- a/rust/operator-binary/src/crd/sensitive_properties.rs +++ b/rust/operator-binary/src/crd/sensitive_properties.rs @@ -1,18 +1,9 @@ use serde::{Deserialize, Serialize}; -use snafu::Snafu; use stackable_operator::{ schemars::{self, JsonSchema}, v2::types::kubernetes::SecretName, }; -#[derive(Snafu, Debug)] -pub enum Error { - #[snafu(display( - "The sensitive properties algorithm '{algorithm}' is not supported in NiFi 2.X.X. Please see https://nifi.apache.org/docs/nifi-docs/html/administration-guide.html#updating-the-sensitive-properties-algorithm on how to upgrade the algorithm." - ))] - UnsupportedSensitivePropertiesAlgorithm { algorithm: String }, -} - /// These settings configure the encryption of sensitive properties in NiFi processors. /// NiFi supports encrypting sensitive properties in processors as they are written to disk. /// You can configure the encryption algorithm and the key to use. @@ -37,15 +28,6 @@ pub struct NifiSensitivePropertiesConfig { /// `nifiPbkdf2AesGcm256` (the default value), /// `nifiArgon2AesGcm256`, /// - /// The following algorithms are deprecated and will be removed in future versions: - /// - /// `nifiArgon2AesGcm128`, - /// `nifiBcryptAesGcm128`, - /// `nifiBcryptAesGcm256`, - /// `nifiPbkdf2AesGcm128`, - /// `nifiScryptAesGcm128`, - /// `nifiScryptAesGcm256`. - /// /// Learn more about the specifics of the algorithm parameters in the /// [NiFi documentation](https://nifi.apache.org/docs/nifi-docs/html/administration-guide.html#property-encryption-algorithms). pub algorithm: Option, @@ -56,104 +38,10 @@ pub struct NifiSensitivePropertiesConfig { )] #[serde(rename_all = "camelCase")] pub enum NifiSensitiveKeyAlgorithm { - // Supported in NiFi 2.x #[strum(serialize = "NIFI_PBKDF2_AES_GCM_256")] NifiPbkdf2AesGcm256, - // Supported in NiFi 2.x #[default] #[strum(serialize = "NIFI_ARGON2_AES_GCM_256")] NifiArgon2AesGcm256, - - // Deprecated in v1 -> can be removed when 1.x.x is no longer supported - #[strum(serialize = "NIFI_BCRYPT_AES_GCM_128")] - NifiBcryptAesGcm128, - - // Deprecated in v1 -> can be removed when 1.x.x is no longer supported - #[strum(serialize = "NIFI_BCRYPT_AES_GCM_256")] - NifiBcryptAesGcm256, - - // Deprecated in v1 -> can be removed when 1.x.x is no longer supported - #[strum(serialize = "NIFI_PBKDF2_AES_GCM_128")] - NifiPbkdf2AesGcm128, - - // Deprecated in v1 -> can be removed when 1.x.x is no longer supported - #[strum(serialize = "NIFI_ARGON2_AES_GCM_128")] - NifiArgon2AesGcm128, - - // Deprecated in v1 -> can be removed when 1.x.x is no longer supported - #[strum(serialize = "NIFI_SCRYPT_AES_GCM_128")] - NifiScryptAesGcm128, - - // Deprecated in v1 -> can be removed when 1.x.x is no longer supported - #[strum(serialize = "NIFI_SCRYPT_AES_GCM_256")] - NifiScryptAesGcm256, -} - -impl NifiSensitiveKeyAlgorithm { - /// Checks if the used encryption algorithm is supported or deprecated. - /// Will warn for deprecation and error out for missing support. - pub fn check_for_nifi_version(&self, product_version: &str) -> Result<(), Error> { - let algorithm = self.to_string(); - - match self { - // Allowed and supported in NiFi 1.x.x and 2.x.x - NifiSensitiveKeyAlgorithm::NifiPbkdf2AesGcm256 - | NifiSensitiveKeyAlgorithm::NifiArgon2AesGcm256 => {} - // All others are deprecated in 1.x.x and removed in 2.x.x - // see https://nifi.apache.org/docs/nifi-docs/html/administration-guide.html#property-encryption-algorithms - _ => { - if product_version.starts_with("1.") { - tracing::warn!( - "You are using a deprecated sensitive properties algorithm '{algorithm}'. Please update to '{pbkd}' or '{argon}'.", - pbkd = NifiSensitiveKeyAlgorithm::NifiPbkdf2AesGcm256, - argon = NifiSensitiveKeyAlgorithm::NifiArgon2AesGcm256 - ) - } else { - return Err(Error::UnsupportedSensitivePropertiesAlgorithm { algorithm }); - } - } - } - - Ok(()) - } -} - -#[cfg(test)] -mod tests { - use super::{Error, NifiSensitiveKeyAlgorithm}; - - #[test] - fn supported_algorithms_are_accepted_on_nifi_2() { - for algorithm in [ - NifiSensitiveKeyAlgorithm::NifiPbkdf2AesGcm256, - NifiSensitiveKeyAlgorithm::NifiArgon2AesGcm256, - ] { - assert!( - algorithm.check_for_nifi_version("2.9.0").is_ok(), - "{algorithm} must be supported on NiFi 2.x" - ); - } - } - - #[test] - fn deprecated_algorithm_is_allowed_with_a_warning_on_nifi_1() { - // Deprecated algorithms only warn (do not error) on NiFi 1.x. - assert!( - NifiSensitiveKeyAlgorithm::NifiArgon2AesGcm128 - .check_for_nifi_version("1.27.0") - .is_ok() - ); - } - - #[test] - fn deprecated_algorithm_is_rejected_on_nifi_2() { - let error = NifiSensitiveKeyAlgorithm::NifiScryptAesGcm256 - .check_for_nifi_version("2.9.0") - .expect_err("a deprecated algorithm must be rejected on NiFi 2.x"); - assert!(matches!( - error, - Error::UnsupportedSensitivePropertiesAlgorithm { .. } - )); - } } diff --git a/rust/operator-binary/src/nifi_controller.rs b/rust/operator-binary/src/nifi_controller.rs index 6f7d579d..b10e9013 100644 --- a/rust/operator-binary/src/nifi_controller.rs +++ b/rust/operator-binary/src/nifi_controller.rs @@ -35,13 +35,10 @@ use crate::{ build::resource::{ listener::{build_group_listener, group_listener_name}, pdb::build_pdb, - reporting_task::build_maybe_reporting_task, service::{build_rolegroup_headless_service, build_rolegroup_metrics_service}, statefulset::build_node_rolegroup_statefulset, }, - controller_name, dereference, operator_name, product_name, - upgrade::{self, ClusterVersionUpdateState}, - validate, + controller_name, dereference, operator_name, product_name, validate, }, crd::{APP_NAME, NifiRole, NifiStatus, v1alpha1}, security::{ @@ -116,16 +113,6 @@ pub enum Error { rolegroup: RoleGroupName, }, - #[snafu(display("failed to apply create ReportingTask service"))] - ApplyCreateReportingTaskService { - source: stackable_operator::cluster_resources::Error, - }, - - #[snafu(display("failed to apply create ReportingTask job"))] - ApplyCreateReportingTaskJob { - source: stackable_operator::cluster_resources::Error, - }, - #[snafu(display("failed to patch service account"))] ApplyServiceAccount { source: stackable_operator::cluster_resources::Error, @@ -155,14 +142,6 @@ pub enum Error { #[snafu(display("security failure"))] Security { source: crate::security::Error }, - #[snafu(display("reporting task failure"))] - ReportingTask { - source: crate::controller::build::resource::reporting_task::Error, - }, - - #[snafu(display("Failed to determine the state of the version upgrade procedure"))] - ClusterVersionUpdateState { source: upgrade::Error }, - #[snafu(display("failed to apply group listener"))] ApplyGroupListener { source: stackable_operator::cluster_resources::Error, @@ -212,30 +191,6 @@ pub async fn reconcile_nifi( .await .context(SecuritySnafu)?; - // If rolling upgrade is supported, kubernetes takes care of the cluster scaling automatically - // otherwise the operator handles it - // manage our own flow for upgrade from 1.x.x to 1.x.x/2.x.x - // TODO: this can be removed once 1.x.x is longer supported - let mut cluster_version_update_state = ClusterVersionUpdateState::NoVersionChange; - let deployed_version = nifi - .status - .as_ref() - .and_then(|status| status.deployed_version.as_ref()); - let rolling_upgrade_supported = resolved_product_image.product_version.starts_with("2.") - && deployed_version.is_some_and(|v| v.as_ref().starts_with("2.")); - - if !rolling_upgrade_supported { - cluster_version_update_state = - upgrade::cluster_version_update_state(&validated_cluster, client, deployed_version) - .await - .context(ClusterVersionUpdateStateSnafu)?; - - if cluster_version_update_state == ClusterVersionUpdateState::UpdateInProgress { - return Ok(Action::await_change()); - } - } - // end todo - let mut cluster_resources = cluster_resources_new( &product_name(), &operator_name(), @@ -295,25 +250,18 @@ pub async fn reconcile_nifi( &validated_cluster, role_group_name, rg, - &client.kubernetes_cluster_info, ) .context(BuildRoleGroupConfigMapSnafu { rolegroup: role_group_name.clone(), })?; - let effective_replicas = - if cluster_version_update_state == ClusterVersionUpdateState::UpdateRequested { - Some(0) - } else { - rg.replicas.map(i32::from) - }; + let effective_replicas = rg.replicas.map(i32::from); let rg_statefulset = build_node_rolegroup_statefulset( &validated_cluster, &client.kubernetes_cluster_info, role_group_name, rg, - rolling_upgrade_supported, effective_replicas, &rbac_sa.name_any(), ) @@ -383,27 +331,6 @@ pub async fn reconcile_nifi( .await .context(ApplyGroupListenerSnafu)?; - // Only add the reporting task in case it is enabled. - if validated_cluster.cluster_config.reporting_task.enabled { - if let Some((reporting_task_job, reporting_task_service)) = build_maybe_reporting_task( - &validated_cluster, - &client.kubernetes_cluster_info, - &rbac_sa.name_any(), - ) - .context(ReportingTaskSnafu)? - { - cluster_resources - .add(client, reporting_task_service) - .await - .context(ApplyCreateReportingTaskServiceSnafu)?; - - cluster_resources - .add(client, reporting_task_job) - .await - .context(ApplyCreateReportingTaskJobSnafu)?; - } - } - // Remove any orphaned resources that still exist in k8s, but have not been added to // the cluster resources during the reconciliation // TODO: this doesn't cater for a graceful cluster shrink, for that we'd need to predict @@ -419,24 +346,12 @@ pub async fn reconcile_nifi( let conditions = compute_conditions(nifi, &[&ss_cond_builder, &cluster_operation_cond_builder]); - // Update the deployed product version in the status after everything has been deployed, unless - // we are still in the process of updating - let status = if cluster_version_update_state != ClusterVersionUpdateState::UpdateRequested { - NifiStatus { - deployed_version: Some( - ProductVersion::from_str(&resolved_product_image.product_version) - .expect("the resolved product version is a valid product version label value"), - ), - conditions, - } - } else { - NifiStatus { - deployed_version: nifi - .status - .as_ref() - .and_then(|status| status.deployed_version.clone()), - conditions, - } + let status = NifiStatus { + deployed_version: Some( + ProductVersion::from_str(&resolved_product_image.product_version) + .expect("the resolved product version is a valid product version label value"), + ), + conditions, }; client diff --git a/tests/templates/kuttl/custom-components-git-sync/30_install-nifi.yaml.j2 b/tests/templates/kuttl/custom-components-git-sync/30_install-nifi.yaml.j2 index 50c25f40..14369e2d 100644 --- a/tests/templates/kuttl/custom-components-git-sync/30_install-nifi.yaml.j2 +++ b/tests/templates/kuttl/custom-components-git-sync/30_install-nifi.yaml.j2 @@ -135,31 +135,7 @@ data: "parameterContexts": [], "parameterProviders": [], "controllerServices": [], - "reportingTasks": [ - { - "identifier": "fd3cf892-0196-1000-0000-000074a555a2", - "instanceIdentifier": "fd3cf892-0196-1000-0000-000074a555a2", - "name": "StackablePrometheusReportingTask", - "type": "org.apache.nifi.reporting.prometheus.PrometheusReportingTask", - "bundle": { - "group": "org.apache.nifi", - "artifact": "nifi-prometheus-nar", - "version": "2.9.0" - }, - "properties": { - "prometheus-reporting-task-metrics-endpoint-port": "8081", - "prometheus-reporting-task-metrics-strategy": "All Components", - "prometheus-reporting-task-instance-id": "${hostname(true)}", - "prometheus-reporting-task-client-auth": "No Authentication", - "prometheus-reporting-task-metrics-send-jvm": "true" - }, - "propertyDescriptors": {}, - "scheduledState": "RUNNING", - "schedulingPeriod": "60 sec", - "schedulingStrategy": "TIMER_DRIVEN", - "componentType": "REPORTING_TASK" - } - ], + "reportingTasks": [], "templates": [], "rootGroup": { "identifier": "b4fe1571-02fd-3351-a2c0-e92f420b3ce9", diff --git a/tests/templates/kuttl/oidc-opa/25-opa-rego.yaml b/tests/templates/kuttl/oidc-opa/25-opa-rego.yaml index 16d60f5d..dc3d8e38 100644 --- a/tests/templates/kuttl/oidc-opa/25-opa-rego.yaml +++ b/tests/templates/kuttl/oidc-opa/25-opa-rego.yaml @@ -10,7 +10,6 @@ data: package my_nifi_package nifi_node_proxy := "CN=generated certificate for pod" - nifi_reporting_task_user := "admin" # Setting "resourceNotFound" to true results in the parent resource being evaluated for authorization, # e.g. the parent of a processor is the processor-group. @@ -64,16 +63,6 @@ data: group == "nifi-user" } - ### Reporting Task permissions (NiFi 1.x.x) - - # Allow the reporting task user - allow := { - "allowed": true, - "dumpCache": true - } if { - input.identity.name == nifi_reporting_task_user - } - ### NiFi Node permissions # NiFi Nodes need this permission to communicate with each other