diff --git a/docs/modules/nifi/assets/images/put-iceberg-processor.png b/docs/modules/nifi/assets/images/put-iceberg-processor.png deleted file mode 100644 index 92e1154c..00000000 Binary files a/docs/modules/nifi/assets/images/put-iceberg-processor.png and /dev/null differ diff --git a/docs/modules/nifi/pages/usage_guide/writing-to-iceberg-tables.adoc b/docs/modules/nifi/pages/usage_guide/writing-to-iceberg-tables.adoc index 9531a1be..154a693e 100644 --- a/docs/modules/nifi/pages/usage_guide/writing-to-iceberg-tables.adoc +++ b/docs/modules/nifi/pages/usage_guide/writing-to-iceberg-tables.adoc @@ -33,91 +33,3 @@ Please read on https://github.com/stackabletech/nifi-iceberg-bundle[its document You don't need any special configs on the `NiFiCluster` in case you are using S3 and no Kerberos. HDFS and Kerberos are also supported, please have a look at the https://github.com/stackabletech/nifi-operator/tree/main/tests/templates/kuttl/iceberg[Iceberg integration test] for that. - -== NiFi 1 - -Starting with `1.19.0`, NiFi supports writing to Iceberg tables. - -The following example shows an example NiFi setup using the Iceberg integration. - -[source,yaml] ----- -apiVersion: nifi.stackable.tech/v1alpha1 -kind: NifiCluster -metadata: - name: nifi -spec: - clusterConfig: - # ... - extraVolumes: - # Will be mounted at /stackable/userdata/nifi-hive-s3-config/ - - name: nifi-hive-s3-config - secret: - secretName: nifi-hive-s3-config ---- -apiVersion: v1 -kind: Secret -metadata: - name: nifi-hive-s3-config -stringData: - core-site.xml: | - - - fs.s3a.aws.credentials.provider - org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider - - - - fs.s3a.endpoint - http://minio:9000 - - - - fs.s3a.access.key - xxx - - - - fs.s3a.secret.key - xxx - - - - fs.s3a.path.style.access - true - - - - fs.s3a.connection.ssl.enabled - false - Enables or disables SSL connections to S3. - - ----- - -Please fill in the correct endpoint, access key and secret key for your S3 store, this is a classic Hadoop config file. - -Use e.g. Trino to create a table for Nifi to write into using something like - -[source,sql] ----- -CREATE SCHEMA IF NOT EXISTS lakehouse.demo WITH (location = 's3a://lakehouse/demo/'); - -CREATE TABLE IF NOT EXISTS lakehouse.demo.test ( - test varchar -); ----- - -In NiFi you need to create a `HiveCatalogService` first which allows you to access the Hive Metastore storing the Iceberg metadata. -Set `Hive Metastore URI` to something like `thrift://hive-iceberg.default.svc.cluster.local:9083`, -`Default Warehouse Location` to `s3a://lakehouse` -and `Hadoop Configuration Resources` to `/stackable/userdata/nifi-hive-s3-config/core-site.xml`. - -Afterwards you can create the `PutIceberg` processor and configure the `HiveCatalogService`. -Also set `Catalog Namespace` to your schema name and the `Table Name`. - -For the `File Format` it is recommended to use `PARQUET` or `ORC` rather than `AVRO` for performance reasons, but you can leave it empty or choose your desired format. - -You should end up with the following `PutIceberg` processor: - -image::put-iceberg-processor.png[] diff --git a/docs/modules/nifi/partials/supported-versions.adoc b/docs/modules/nifi/partials/supported-versions.adoc index 4b80fee5..c68a18df 100644 --- a/docs/modules/nifi/partials/supported-versions.adoc +++ b/docs/modules/nifi/partials/supported-versions.adoc @@ -3,9 +3,8 @@ // Stackable Platform documentation. * 2.9.0 -* 2.7.2 +* 2.7.2 (Deprecated) ** Please note that NiFi starting with 2.7.2 supports Iceberg, but only with S3 and Iceberg REST catalog (no Hive metastore or HDFS support). Please read on xref:nifi:usage_guide/writing-to-iceberg-tables.adoc[] for details. * 2.6.0 (LTS) -* 1.28.1 (Deprecated) For details on how to upgrade your NiFi version, refer to xref:nifi:usage_guide/updating.adoc[]. diff --git a/tests/templates/kuttl/cluster-sync-cs-bug/20-install-nifi.yaml.j2 b/tests/templates/kuttl/cluster-sync-cs-bug/20-install-nifi.yaml.j2 index f47dc621..61f43bcd 100644 --- a/tests/templates/kuttl/cluster-sync-cs-bug/20-install-nifi.yaml.j2 +++ b/tests/templates/kuttl/cluster-sync-cs-bug/20-install-nifi.yaml.j2 @@ -166,12 +166,12 @@ metadata: name: nifi spec: image: -{% if test_scenario['values']['nifi-v2'].find(",") > 0 %} - custom: "{{ test_scenario['values']['nifi-v2'].split(',')[1] }}" - productVersion: "{{ test_scenario['values']['nifi-v2'].split(',')[0] }}" +{% if test_scenario['values']['nifi'].find(",") > 0 %} + custom: "{{ test_scenario['values']['nifi'].split(',')[1] }}" + productVersion: "{{ test_scenario['values']['nifi'].split(',')[0] }}" {% else %} custom: null - productVersion: "{{ test_scenario['values']['nifi-v2'] }}" + productVersion: "{{ test_scenario['values']['nifi'] }}" {% endif %} pullPolicy: IfNotPresent clusterConfig: 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 0e716577..b1f43454 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 @@ -148,7 +148,7 @@ data: "bundle": { "group": "org.apache.nifi", "artifact": "nifi-prometheus-nar", - "version": "1.28.1" + "version": "2.9.0" }, "properties": { "prometheus-reporting-task-metrics-endpoint-port": "8081", @@ -190,7 +190,7 @@ data: "bundle": { "group": "org.apache.nifi", "artifact": "nifi-standard-nar", - "version": "1.28.1" + "version": "2.9.0" }, "properties": { "Regular Expression": "(?s)(^.*$)", @@ -235,7 +235,7 @@ data: "bundle": { "group": "org.apache.nifi", "artifact": "nifi-standard-nar", - "version": "1.28.1" + "version": "2.9.0" }, "properties": { "HTTP Context Map": "fd3b17c4-0196-1000-0000-000076ed786c", @@ -311,7 +311,7 @@ data: "bundle": { "group": "org.apache.nifi", "artifact": "nifi-standard-nar", - "version": "1.28.1" + "version": "2.9.0" }, "properties": { "multipart-request-max-size": "1 MB", @@ -475,7 +475,7 @@ data: "bundle": { "group": "org.apache.nifi", "artifact": "nifi-http-context-map-nar", - "version": "1.28.1" + "version": "2.9.0" }, "properties": { "Request Expiration": "1 min", @@ -488,7 +488,7 @@ data: "bundle": { "group": "org.apache.nifi", "artifact": "nifi-standard-services-api-nar", - "version": "1.28.1" + "version": "2.9.0" } } ], diff --git a/tests/templates/kuttl/smoke_v1/00-patch-ns.yaml.j2 b/tests/templates/kuttl/smoke/00-patch-ns.yaml.j2 similarity index 100% rename from tests/templates/kuttl/smoke_v1/00-patch-ns.yaml.j2 rename to tests/templates/kuttl/smoke/00-patch-ns.yaml.j2 diff --git a/tests/templates/kuttl/smoke_v1/00-range-limit.yaml b/tests/templates/kuttl/smoke/00-range-limit.yaml similarity index 100% rename from tests/templates/kuttl/smoke_v1/00-range-limit.yaml rename to tests/templates/kuttl/smoke/00-range-limit.yaml diff --git a/tests/templates/kuttl/smoke_v1/10-assert.yaml.j2 b/tests/templates/kuttl/smoke/10-assert.yaml.j2 similarity index 100% rename from tests/templates/kuttl/smoke_v1/10-assert.yaml.j2 rename to tests/templates/kuttl/smoke/10-assert.yaml.j2 diff --git a/tests/templates/kuttl/smoke_v1/10-install-vector-aggregator-discovery-configmap.yaml.j2 b/tests/templates/kuttl/smoke/10-install-vector-aggregator-discovery-configmap.yaml.j2 similarity index 100% rename from tests/templates/kuttl/smoke_v1/10-install-vector-aggregator-discovery-configmap.yaml.j2 rename to tests/templates/kuttl/smoke/10-install-vector-aggregator-discovery-configmap.yaml.j2 diff --git a/tests/templates/kuttl/smoke_v2/20-assert.yaml.j2 b/tests/templates/kuttl/smoke/20-assert.yaml.j2 similarity index 100% rename from tests/templates/kuttl/smoke_v2/20-assert.yaml.j2 rename to tests/templates/kuttl/smoke/20-assert.yaml.j2 diff --git a/tests/templates/kuttl/smoke_v2/20-install-zookeeper.yaml.j2 b/tests/templates/kuttl/smoke/20-install-zookeeper.yaml.j2 similarity index 100% rename from tests/templates/kuttl/smoke_v2/20-install-zookeeper.yaml.j2 rename to tests/templates/kuttl/smoke/20-install-zookeeper.yaml.j2 diff --git a/tests/templates/kuttl/smoke_v1/30-assert.yaml b/tests/templates/kuttl/smoke/30-assert.yaml similarity index 100% rename from tests/templates/kuttl/smoke_v1/30-assert.yaml rename to tests/templates/kuttl/smoke/30-assert.yaml diff --git a/tests/templates/kuttl/smoke_v2/30-install-nifi.yaml.j2 b/tests/templates/kuttl/smoke/30-install-nifi.yaml.j2 similarity index 90% rename from tests/templates/kuttl/smoke_v2/30-install-nifi.yaml.j2 rename to tests/templates/kuttl/smoke/30-install-nifi.yaml.j2 index ec39df14..b55c8c63 100644 --- a/tests/templates/kuttl/smoke_v2/30-install-nifi.yaml.j2 +++ b/tests/templates/kuttl/smoke/30-install-nifi.yaml.j2 @@ -42,12 +42,12 @@ metadata: name: nifi spec: image: -{% if test_scenario['values']['nifi-v2'].find(",") > 0 %} - custom: "{{ test_scenario['values']['nifi-v2'].split(',')[1] }}" - productVersion: "{{ test_scenario['values']['nifi-v2'].split(',')[0] }}" +{% if test_scenario['values']['nifi'].find(",") > 0 %} + custom: "{{ test_scenario['values']['nifi'].split(',')[1] }}" + productVersion: "{{ test_scenario['values']['nifi'].split(',')[0] }}" {% else %} custom: null - productVersion: "{{ test_scenario['values']['nifi-v2'] }}" + productVersion: "{{ test_scenario['values']['nifi'] }}" {% endif %} pullPolicy: IfNotPresent clusterConfig: diff --git a/tests/templates/kuttl/smoke_v1/31-assert.yaml.j2 b/tests/templates/kuttl/smoke/31-assert.yaml.j2 similarity index 100% rename from tests/templates/kuttl/smoke_v1/31-assert.yaml.j2 rename to tests/templates/kuttl/smoke/31-assert.yaml.j2 diff --git a/tests/templates/kuttl/smoke_v1/32-assert.yaml b/tests/templates/kuttl/smoke/32-assert.yaml similarity index 100% rename from tests/templates/kuttl/smoke_v1/32-assert.yaml rename to tests/templates/kuttl/smoke/32-assert.yaml diff --git a/tests/templates/kuttl/smoke_v1/33-assert.yaml b/tests/templates/kuttl/smoke/33-assert.yaml similarity index 100% rename from tests/templates/kuttl/smoke_v1/33-assert.yaml rename to tests/templates/kuttl/smoke/33-assert.yaml diff --git a/tests/templates/kuttl/smoke_v2/34-assert.yaml.j2 b/tests/templates/kuttl/smoke/34-assert.yaml.j2 similarity index 100% rename from tests/templates/kuttl/smoke_v2/34-assert.yaml.j2 rename to tests/templates/kuttl/smoke/34-assert.yaml.j2 diff --git a/tests/templates/kuttl/smoke_v2/35-assert.yaml.j2 b/tests/templates/kuttl/smoke/35-assert.yaml.j2 similarity index 100% rename from tests/templates/kuttl/smoke_v2/35-assert.yaml.j2 rename to tests/templates/kuttl/smoke/35-assert.yaml.j2 diff --git a/tests/templates/kuttl/smoke_v1/40-assert.yaml b/tests/templates/kuttl/smoke/40-assert.yaml similarity index 100% rename from tests/templates/kuttl/smoke_v1/40-assert.yaml rename to tests/templates/kuttl/smoke/40-assert.yaml diff --git a/tests/templates/kuttl/smoke_v1/40-scale-up-nifi.yaml.j2 b/tests/templates/kuttl/smoke/40-scale-up-nifi.yaml.j2 similarity index 100% rename from tests/templates/kuttl/smoke_v1/40-scale-up-nifi.yaml.j2 rename to tests/templates/kuttl/smoke/40-scale-up-nifi.yaml.j2 diff --git a/tests/templates/kuttl/smoke_v1/50-assert.yaml b/tests/templates/kuttl/smoke/50-assert.yaml similarity index 100% rename from tests/templates/kuttl/smoke_v1/50-assert.yaml rename to tests/templates/kuttl/smoke/50-assert.yaml diff --git a/tests/templates/kuttl/smoke_v2/50-install-test-nifi.yaml.j2 b/tests/templates/kuttl/smoke/50-install-test-nifi.yaml.j2 similarity index 100% rename from tests/templates/kuttl/smoke_v2/50-install-test-nifi.yaml.j2 rename to tests/templates/kuttl/smoke/50-install-test-nifi.yaml.j2 diff --git a/tests/templates/kuttl/smoke_v1/60-assert.yaml.j2 b/tests/templates/kuttl/smoke/60-assert.yaml.j2 similarity index 100% rename from tests/templates/kuttl/smoke_v1/60-assert.yaml.j2 rename to tests/templates/kuttl/smoke/60-assert.yaml.j2 diff --git a/tests/templates/kuttl/smoke_v1/60-prepare-test-nifi.yaml b/tests/templates/kuttl/smoke/60-prepare-test-nifi.yaml similarity index 100% rename from tests/templates/kuttl/smoke_v1/60-prepare-test-nifi.yaml rename to tests/templates/kuttl/smoke/60-prepare-test-nifi.yaml diff --git a/tests/templates/kuttl/smoke_v1/cacert.pem b/tests/templates/kuttl/smoke/cacert.pem similarity index 100% rename from tests/templates/kuttl/smoke_v1/cacert.pem rename to tests/templates/kuttl/smoke/cacert.pem diff --git a/tests/templates/kuttl/smoke_v1/test_nifi.py b/tests/templates/kuttl/smoke/test_nifi.py similarity index 100% rename from tests/templates/kuttl/smoke_v1/test_nifi.py rename to tests/templates/kuttl/smoke/test_nifi.py diff --git a/tests/templates/kuttl/smoke_v2/test_nifi_metrics.py b/tests/templates/kuttl/smoke/test_nifi_metrics.py similarity index 100% rename from tests/templates/kuttl/smoke_v2/test_nifi_metrics.py rename to tests/templates/kuttl/smoke/test_nifi_metrics.py diff --git a/tests/templates/kuttl/smoke_v1/20-assert.yaml b/tests/templates/kuttl/smoke_v1/20-assert.yaml deleted file mode 100644 index 49ba7437..00000000 --- a/tests/templates/kuttl/smoke_v1/20-assert.yaml +++ /dev/null @@ -1,12 +0,0 @@ ---- -apiVersion: kuttl.dev/v1beta1 -kind: TestAssert -timeout: 600 ---- -apiVersion: apps/v1 -kind: StatefulSet -metadata: - name: zookeeper-server-default -status: - readyReplicas: 1 - replicas: 1 diff --git a/tests/templates/kuttl/smoke_v1/20-install-zookeeper.yaml.j2 b/tests/templates/kuttl/smoke_v1/20-install-zookeeper.yaml.j2 deleted file mode 100644 index 2080faf6..00000000 --- a/tests/templates/kuttl/smoke_v1/20-install-zookeeper.yaml.j2 +++ /dev/null @@ -1,22 +0,0 @@ ---- -apiVersion: zookeeper.stackable.tech/v1alpha1 -kind: ZookeeperCluster -metadata: - name: zookeeper -spec: - image: - productVersion: "{{ test_scenario['values']['zookeeper'] }}" - pullPolicy: IfNotPresent - clusterConfig: -{% if lookup('env', 'VECTOR_AGGREGATOR') %} - vectorAggregatorConfigMapName: vector-aggregator-discovery -{% endif %} - servers: - config: - logging: - enableVectorAgent: {{ lookup('env', 'VECTOR_AGGREGATOR') | length > 0 }} - roleConfig: - listenerClass: {{ test_scenario['values']['listener-class'] }} - roleGroups: - default: - replicas: 1 diff --git a/tests/templates/kuttl/smoke_v1/30-install-nifi.yaml.j2 b/tests/templates/kuttl/smoke_v1/30-install-nifi.yaml.j2 deleted file mode 100644 index ee1798af..00000000 --- a/tests/templates/kuttl/smoke_v1/30-install-nifi.yaml.j2 +++ /dev/null @@ -1,88 +0,0 @@ ---- -apiVersion: authentication.stackable.tech/v1alpha1 -kind: AuthenticationClass -metadata: - name: nifi-users -spec: - provider: - static: - userCredentialsSecret: - name: nifi-users ---- -apiVersion: v1 -kind: Secret -metadata: - name: nifi-users -stringData: - admin: > - passwordWithSpecialCharacter\@<&>"' ---- -apiVersion: v1 -kind: Secret -metadata: - name: nifi-sensitive-property-key -stringData: - nifiSensitivePropsKey: mYsUp3rS3cr3tk3y ---- -apiVersion: zookeeper.stackable.tech/v1alpha1 -kind: ZookeeperZnode -metadata: - name: nifi-znode -spec: - clusterRef: - name: zookeeper ---- -# We need to create the NifiCluster last, so that the ConfigMaps/Secrets it mounts are already -# existing to prevent unnecessary Pod restarts. -apiVersion: nifi.stackable.tech/v1alpha1 -kind: NifiCluster -metadata: - name: nifi -spec: - image: -{% if test_scenario['values']['nifi-v1'].find(",") > 0 %} - custom: "{{ test_scenario['values']['nifi-v1'].split(',')[1] }}" - productVersion: "{{ test_scenario['values']['nifi-v1'].split(',')[0] }}" -{% else %} - custom: null - productVersion: "{{ test_scenario['values']['nifi-v1'] }}" -{% endif %} - pullPolicy: IfNotPresent - clusterConfig: - zookeeperConfigMapName: nifi-znode - authentication: - - authenticationClass: nifi-users - hostHeaderCheck: - allowAll: false - additionalAllowedHosts: - - example.com:1234 - sensitiveProperties: - keySecret: nifi-sensitive-property-key -{% if lookup('env', 'VECTOR_AGGREGATOR') %} - vectorAggregatorConfigMapName: vector-aggregator-discovery -{% endif %} - nodes: - envOverrides: - COMMON_VAR: role-value # overridden by role group below - ROLE_VAR: role-value # only defined here at role level - configOverrides: - "nifi.properties": - "nifi.diagnostics.on.shutdown.enabled": "true" - "nifi.diagnostics.on.shutdown.verbose": "false" - # Quicker startup, and we only have a single node - "nifi.cluster.flow.election.max.wait.time": "10 secs" - config: - logging: - enableVectorAgent: {{ lookup('env', 'VECTOR_AGGREGATOR') | length > 0 }} - roleConfig: - listenerClass: {{ test_scenario['values']['listener-class'] }} - roleGroups: - default: - replicas: 2 - envOverrides: - COMMON_VAR: group-value # overrides role value - GROUP_VAR: group-value # only defined here at group level - configOverrides: - "nifi.properties": - "nifi.diagnostics.on.shutdown.enabled": "false" - "nifi.diagnostics.on.shutdown.max.filecount": "20" diff --git a/tests/templates/kuttl/smoke_v1/50-install-test-nifi.yaml b/tests/templates/kuttl/smoke_v1/50-install-test-nifi.yaml deleted file mode 100644 index dc8a0b39..00000000 --- a/tests/templates/kuttl/smoke_v1/50-install-test-nifi.yaml +++ /dev/null @@ -1,28 +0,0 @@ ---- -apiVersion: apps/v1 -kind: StatefulSet -metadata: - name: test-nifi - labels: - app: test-nifi -spec: - replicas: 1 - selector: - matchLabels: - app: test-nifi - template: - metadata: - labels: - app: test-nifi - spec: - containers: - - name: test-nifi - image: oci.stackable.tech/sdp/testing-tools/nifi:0.3.0-stackable0.0.0-dev - command: ["sleep", "infinity"] - resources: - requests: - memory: "128Mi" - cpu: "100m" - limits: - memory: "128Mi" - cpu: "400m" diff --git a/tests/templates/kuttl/smoke_v1/test_nifi_metrics.py b/tests/templates/kuttl/smoke_v1/test_nifi_metrics.py deleted file mode 100755 index 86fd42a4..00000000 --- a/tests/templates/kuttl/smoke_v1/test_nifi_metrics.py +++ /dev/null @@ -1,62 +0,0 @@ -#!/usr/bin/env python -import argparse -import requests -import time -from requests.exceptions import ConnectionError - -if __name__ == "__main__": - # Construct an argument parser - all_args = argparse.ArgumentParser() - # Add arguments to the parser - all_args.add_argument( - "-m", - "--metric", - required=False, - default="nifi_amount_bytes_read", - help="The name of a certain metric to check", - ) - all_args.add_argument( - "-n", "--namespace", required=True, help="The namespace the test is running in" - ) - all_args.add_argument( - "-p", - "--port", - required=False, - default="8081", - help="The port where metrics are exposed", - ) - all_args.add_argument( - "-t", - "--timeout", - required=False, - default="120", - help="The timeout in seconds to wait for the metrics port to be opened", - ) - - args = vars(all_args.parse_args()) - metric_name = args["metric"] - namespace = args["namespace"] - port = args["port"] - timeout = int(args["timeout"]) - - url = f"http://nifi-node-default-metrics.{namespace}.svc.cluster.local:{port}/metrics/" - - # wait for 'timeout' seconds - t_end = time.time() + timeout - while time.time() < t_end: - try: - response = requests.get(url) - response.raise_for_status() - if metric_name in response.text: - print("Test metrics succeeded!") - exit(0) - else: - print( - f"Could not find metric [{metric_name}] in response:\n {response.text}" - ) - time.sleep(timeout) - except ConnectionError: - # NewConnectionError is expected until metrics are available - time.sleep(10) - - exit(-1) diff --git a/tests/templates/kuttl/smoke_v2/00-patch-ns.yaml.j2 b/tests/templates/kuttl/smoke_v2/00-patch-ns.yaml.j2 deleted file mode 100644 index 67185acf..00000000 --- a/tests/templates/kuttl/smoke_v2/00-patch-ns.yaml.j2 +++ /dev/null @@ -1,9 +0,0 @@ -{% if test_scenario['values']['openshift'] == 'true' %} -# see https://github.com/stackabletech/issues/issues/566 ---- -apiVersion: kuttl.dev/v1beta1 -kind: TestStep -commands: - - script: kubectl patch namespace $NAMESPACE -p '{"metadata":{"labels":{"pod-security.kubernetes.io/enforce":"privileged"}}}' - timeout: 120 -{% endif %} diff --git a/tests/templates/kuttl/smoke_v2/00-range-limit.yaml b/tests/templates/kuttl/smoke_v2/00-range-limit.yaml deleted file mode 100644 index 8fd02210..00000000 --- a/tests/templates/kuttl/smoke_v2/00-range-limit.yaml +++ /dev/null @@ -1,11 +0,0 @@ ---- -apiVersion: v1 -kind: LimitRange -metadata: - name: limit-request-ratio -spec: - limits: - - type: "Container" - maxLimitRequestRatio: - cpu: 5 - memory: 1 diff --git a/tests/templates/kuttl/smoke_v2/10-assert.yaml.j2 b/tests/templates/kuttl/smoke_v2/10-assert.yaml.j2 deleted file mode 100644 index 50b1d4c3..00000000 --- a/tests/templates/kuttl/smoke_v2/10-assert.yaml.j2 +++ /dev/null @@ -1,10 +0,0 @@ ---- -apiVersion: kuttl.dev/v1beta1 -kind: TestAssert -{% if lookup('env', 'VECTOR_AGGREGATOR') %} ---- -apiVersion: v1 -kind: ConfigMap -metadata: - name: vector-aggregator-discovery -{% endif %} diff --git a/tests/templates/kuttl/smoke_v2/10-install-vector-aggregator-discovery-configmap.yaml.j2 b/tests/templates/kuttl/smoke_v2/10-install-vector-aggregator-discovery-configmap.yaml.j2 deleted file mode 100644 index 2d6a0df5..00000000 --- a/tests/templates/kuttl/smoke_v2/10-install-vector-aggregator-discovery-configmap.yaml.j2 +++ /dev/null @@ -1,9 +0,0 @@ -{% if lookup('env', 'VECTOR_AGGREGATOR') %} ---- -apiVersion: v1 -kind: ConfigMap -metadata: - name: vector-aggregator-discovery -data: - ADDRESS: {{ lookup('env', 'VECTOR_AGGREGATOR') }} -{% endif %} diff --git a/tests/templates/kuttl/smoke_v2/30-assert.yaml b/tests/templates/kuttl/smoke_v2/30-assert.yaml deleted file mode 100644 index 9a31b9e7..00000000 --- a/tests/templates/kuttl/smoke_v2/30-assert.yaml +++ /dev/null @@ -1,27 +0,0 @@ ---- -apiVersion: kuttl.dev/v1beta1 -kind: TestAssert -timeout: 1200 ---- -apiVersion: apps/v1 -kind: StatefulSet -metadata: - name: nifi-node-default - labels: - restarter.stackable.tech/enabled: "true" -spec: - template: - spec: - terminationGracePeriodSeconds: 300 -status: - readyReplicas: 2 - replicas: 2 ---- -apiVersion: policy/v1 -kind: PodDisruptionBudget -metadata: - name: nifi-node -status: - expectedPods: 2 - currentHealthy: 2 - disruptionsAllowed: 1 diff --git a/tests/templates/kuttl/smoke_v2/31-assert.yaml.j2 b/tests/templates/kuttl/smoke_v2/31-assert.yaml.j2 deleted file mode 100644 index 7b2eacd4..00000000 --- a/tests/templates/kuttl/smoke_v2/31-assert.yaml.j2 +++ /dev/null @@ -1,6 +0,0 @@ ---- -apiVersion: kuttl.dev/v1beta1 -kind: TestAssert -timeout: 30 -commands: -- script: kubectl get cm -n $NAMESPACE nifi-node-default -o yaml | grep -- 'nifi.web.proxy.host=.*example.com:1234' | xargs test ! -z diff --git a/tests/templates/kuttl/smoke_v2/32-assert.yaml b/tests/templates/kuttl/smoke_v2/32-assert.yaml deleted file mode 100644 index 960b4789..00000000 --- a/tests/templates/kuttl/smoke_v2/32-assert.yaml +++ /dev/null @@ -1,19 +0,0 @@ ---- -apiVersion: kuttl.dev/v1beta1 -kind: TestAssert -timeout: 600 -commands: - # - # Test envOverrides - # - - script: | - kubectl -n $NAMESPACE get sts nifi-node-default -o yaml | yq -e '.spec.template.spec.containers[] | select (.name == "nifi") | .env[] | select (.name == "COMMON_VAR" and .value == "group-value")' - kubectl -n $NAMESPACE get sts nifi-node-default -o yaml | yq -e '.spec.template.spec.containers[] | select (.name == "nifi") | .env[] | select (.name == "GROUP_VAR" and .value == "group-value")' - kubectl -n $NAMESPACE get sts nifi-node-default -o yaml | yq -e '.spec.template.spec.containers[] | select (.name == "nifi") | .env[] | select (.name == "ROLE_VAR" and .value == "role-value")' - # - # Test configOverrides - # - - script: | - kubectl -n $NAMESPACE get cm nifi-node-default -o yaml | yq -e '.data."nifi.properties"' | grep "nifi.diagnostics.on.shutdown.enabled=false" - kubectl -n $NAMESPACE get cm nifi-node-default -o yaml | yq -e '.data."nifi.properties"' | grep "nifi.diagnostics.on.shutdown.verbose=false" - kubectl -n $NAMESPACE get cm nifi-node-default -o yaml | yq -e '.data."nifi.properties"' | grep "nifi.diagnostics.on.shutdown.max.filecount=20" diff --git a/tests/templates/kuttl/smoke_v2/33-assert.yaml b/tests/templates/kuttl/smoke_v2/33-assert.yaml deleted file mode 100644 index 271a8bec..00000000 --- a/tests/templates/kuttl/smoke_v2/33-assert.yaml +++ /dev/null @@ -1,7 +0,0 @@ ---- -# This test checks if the containerdebug-state.json file is present and valid -apiVersion: kuttl.dev/v1beta1 -kind: TestAssert -timeout: 600 -commands: - - script: kubectl exec -n $NAMESPACE --container nifi nifi-node-default-0 -- cat /stackable/log/containerdebug-state.json | jq --exit-status '"valid JSON"' diff --git a/tests/templates/kuttl/smoke_v2/40-assert.yaml b/tests/templates/kuttl/smoke_v2/40-assert.yaml deleted file mode 100644 index c3bcd76b..00000000 --- a/tests/templates/kuttl/smoke_v2/40-assert.yaml +++ /dev/null @@ -1,12 +0,0 @@ ---- -apiVersion: kuttl.dev/v1beta1 -kind: TestAssert -timeout: 1200 ---- -apiVersion: apps/v1 -kind: StatefulSet -metadata: - name: nifi-node-default -status: - readyReplicas: 3 - replicas: 3 diff --git a/tests/templates/kuttl/smoke_v2/40-scale-up-nifi.yaml.j2 b/tests/templates/kuttl/smoke_v2/40-scale-up-nifi.yaml.j2 deleted file mode 100644 index 37ded261..00000000 --- a/tests/templates/kuttl/smoke_v2/40-scale-up-nifi.yaml.j2 +++ /dev/null @@ -1,8 +0,0 @@ ---- -apiVersion: kuttl.dev/v1beta1 -kind: TestStep -commands: - - script: >- - kubectl --namespace $NAMESPACE - patch nificlusters.nifi.stackable.tech nifi - --type=merge --patch '{"spec":{"nodes": {"roleGroups": {"default": {"replicas": 3}}}}}' diff --git a/tests/templates/kuttl/smoke_v2/50-assert.yaml b/tests/templates/kuttl/smoke_v2/50-assert.yaml deleted file mode 100644 index f56ce239..00000000 --- a/tests/templates/kuttl/smoke_v2/50-assert.yaml +++ /dev/null @@ -1,12 +0,0 @@ ---- -apiVersion: kuttl.dev/v1beta1 -kind: TestAssert -timeout: 1200 ---- -apiVersion: apps/v1 -kind: StatefulSet -metadata: - name: test-nifi -status: - readyReplicas: 1 - replicas: 1 diff --git a/tests/templates/kuttl/smoke_v2/60-assert.yaml.j2 b/tests/templates/kuttl/smoke_v2/60-assert.yaml.j2 deleted file mode 100644 index 1d531af0..00000000 --- a/tests/templates/kuttl/smoke_v2/60-assert.yaml.j2 +++ /dev/null @@ -1,7 +0,0 @@ ---- -apiVersion: kuttl.dev/v1beta1 -kind: TestAssert -timeout: 300 -commands: - - script: kubectl exec -n $NAMESPACE test-nifi-0 -- python /tmp/test_nifi.py -u admin -p 'passwordWithSpecialCharacter\@<&>"'"'" -n $NAMESPACE -c 3 - - script: kubectl exec -n $NAMESPACE test-nifi-0 -- python /tmp/test_nifi_metrics.py -n $NAMESPACE diff --git a/tests/templates/kuttl/smoke_v2/60-prepare-test-nifi.yaml b/tests/templates/kuttl/smoke_v2/60-prepare-test-nifi.yaml deleted file mode 100644 index c3ac9b79..00000000 --- a/tests/templates/kuttl/smoke_v2/60-prepare-test-nifi.yaml +++ /dev/null @@ -1,7 +0,0 @@ ---- -apiVersion: kuttl.dev/v1beta1 -kind: TestStep -commands: - - script: kubectl cp -n $NAMESPACE ./test_nifi_metrics.py test-nifi-0:/tmp - - script: kubectl cp -n $NAMESPACE ./test_nifi.py test-nifi-0:/tmp - - script: kubectl cp -n $NAMESPACE ./cacert.pem test-nifi-0:/tmp diff --git a/tests/templates/kuttl/smoke_v2/cacert.pem b/tests/templates/kuttl/smoke_v2/cacert.pem deleted file mode 100644 index ebe73910..00000000 --- a/tests/templates/kuttl/smoke_v2/cacert.pem +++ /dev/null @@ -1,20 +0,0 @@ ------BEGIN CERTIFICATE----- -MIIDVTCCAj2gAwIBAgIJAJ8/0entaUgnMA0GCSqGSIb3DQEBCwUAMCYxJDAiBgNV -BAMMG3NlY3JldC1vcGVyYXRvciBzZWxmLXNpZ25lZDAeFw0yMjAxMTIxNDU3NDVa -Fw0yNDAxMTIxNTAyNDVaMCYxJDAiBgNVBAMMG3NlY3JldC1vcGVyYXRvciBzZWxm -LXNpZ25lZDCCASIwDQYJKoZIhvcNAQEBBQADggEPADCCAQoCggEBALLKNGhq5gE+ -mL9zFCLqtc22CLk8BSbjesjUEhBK3kxDvFDa2ou5atH0eUFjtOSszay2oBrCTVWK -wZBsdUkL0HkW/wq9A8EUkQ8EownXnsxpI61CLNGLPpBZc+CRHhyWDD6BqwGvEHEv -W546mh6k49//7zCiYfTK9/LCKBCFdDV6Sb7mNJ8HbNUj54uwC6iOgH25OCRDh4Bt -zXoSrV9GLAm6AM25ZFo+ONOUBMtv7pavaR0CFMnAixl2NKV2wyLBYAYnJgdJFzGD -8mP6HwuR7e2g7PkcyC01EnX4iOIuuKHT/Xl9ynut4nHI7g6popotgashrQ5Jf8MS -Kf98O12LzSMCAwEAAaOBhTCBgjAPBgNVHRMBAf8EBTADAQH/MB0GA1UdDgQWBBRf -U9OxCBwCqYiUjWqY05sz3a6cmjBABgNVHSMEOTA3oSqkKDAmMSQwIgYDVQQDDBtz -ZWNyZXQtb3BlcmF0b3Igc2VsZi1zaWduZWSCCQCfP9Hp7WlIJzAOBgNVHQ8BAf8E -BAMCAYYwDQYJKoZIhvcNAQELBQADggEBAA8Flk1XOb1pH33Mbie5ronP2xw/xf6t -Ox3PBEZ+5/jSPdIwoSaRp9JoP0L9Rg68jzcl5QMa4pOYWe+C1q8aZP0tjfq1eJfO -UD5ik2DQgEuoF1ELgW1xoM38vkd8wgE711swDHK2zAsOudSzO4XZ4rQ6kaXXtoej -2kFhxDYcC+na90LdkJM0kAqrjxlFaP7WgUK+HA2iN00CFSOI9FVdppLtootbcb+y -+WfXxM7gA9Exg4f2vKGVx7UxB/k4AbPvogBQZvK8VoAQocAhWrw7o2rqAesAw6JD -WwQjM69TlEfbHYXtTfMbi01Wi5TtVhFCjyXK6KDsqSgU+9McExIy70k= ------END CERTIFICATE----- diff --git a/tests/templates/kuttl/smoke_v2/test_nifi.py b/tests/templates/kuttl/smoke_v2/test_nifi.py deleted file mode 100755 index 5c72c1a6..00000000 --- a/tests/templates/kuttl/smoke_v2/test_nifi.py +++ /dev/null @@ -1,85 +0,0 @@ -#!/usr/bin/env python -import requests -import json -import argparse -import urllib3 -from time import sleep - - -def get_token(nifi_host, username, password): - nifi_headers = { - "content-type": "application/x-www-form-urlencoded; charset=UTF-8", - } - data = {"username": username, "password": password} - - # TODO: handle actual errors when connecting properly - nifi_url = nifi_host + "/nifi-api/access/token" - response = requests.post( - nifi_url, headers=nifi_headers, data=data, verify=False - ) # , cert='./tmp/cacert.pem') - - if response.ok: - nifi_token = response.content.decode("utf-8") - return "Bearer " + nifi_token - else: - print(f"Failed to get token: {response.status_code}: {response.content}") - exit(-1) - - -if __name__ == "__main__": - # Construct an argument parser - all_args = argparse.ArgumentParser() - - # Add arguments to the parser - all_args.add_argument("-u", "--user", required=True, help="Username to connect as") - all_args.add_argument( - "-p", "--password", required=True, help="Password for the user" - ) - all_args.add_argument( - "-n", "--namespace", required=True, help="Namespace the test is running in" - ) - all_args.add_argument( - "-c", "--count", required=True, help="The expected number of Nodes" - ) - args = vars(all_args.parse_args()) - - # disable warnings as we have specified non-verified https connections - urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) - - host = f"https://nifi-node-default-1.nifi-node-default-headless.{args['namespace']}.svc.cluster.local:8443" - token = get_token(host, args["user"], args["password"]) - headers = {"Authorization": token} - node_count = int(args["count"]) - - x = 0 - while x < 15: - url = host + "/nifi-api/controller/cluster" - cluster = requests.get( - url, headers=headers, verify=False - ) # , cert='/tmp/cacert.pem') - if cluster.status_code != 200: - print("Waiting for cluster...") - else: - cluster_data = json.loads(cluster.content.decode("utf-8")) - nodes = cluster_data["cluster"]["nodes"] - if len(nodes) != node_count: - print( - f"Cluster should have {node_count} nodes at this stage, but has: {len(nodes)}" - ) - else: - connected = True - for node in nodes: - if node["status"] != "CONNECTED": - print( - f"Node {node['nodeId']} is in state {node['status']} but should have been CONNECTED" - ) - connected = False - if connected: - print("Test succeeded!") - exit(0) - print("Retrying...") - x = x + 1 - sleep(10) - - print("Test failed") - exit(-1) diff --git a/tests/templates/kuttl/upgrade/03-install-test-nifi.yaml b/tests/templates/kuttl/upgrade/03-install-test-nifi.yaml index 243b09dd..7992be3d 100644 --- a/tests/templates/kuttl/upgrade/03-install-test-nifi.yaml +++ b/tests/templates/kuttl/upgrade/03-install-test-nifi.yaml @@ -20,3 +20,8 @@ spec: image: oci.stackable.tech/sdp/testing-tools/nifi:0.3.0-stackable0.0.0-dev stdin: true tty: true + env: + - name: NAMESPACE + valueFrom: + fieldRef: + fieldPath: metadata.namespace diff --git a/tests/templates/kuttl/upgrade/04-assert.yaml.j2 b/tests/templates/kuttl/upgrade/04-assert.yaml.j2 index 1f66464c..9297b0f7 100644 --- a/tests/templates/kuttl/upgrade/04-assert.yaml.j2 +++ b/tests/templates/kuttl/upgrade/04-assert.yaml.j2 @@ -4,13 +4,6 @@ kind: TestAssert timeout: 300 commands: - script: kubectl exec -n $NAMESPACE test-nifi-0 -- python /tmp/test_nifi.py -u admin -p supersecretpassword -n $NAMESPACE -c 3 -{% if test_scenario['values']['nifi_old'].split(',')[0].startswith('1.') %} - - script: kubectl exec -n $NAMESPACE test-nifi-0 -- python /tmp/test_nifi_metrics.py -n $NAMESPACE -{% endif %} -{% if test_scenario['values']['nifi_old'].split(',')[0] == '2.0.0' %} - - script: kubectl exec -n $NAMESPACE test-nifi-0 -- sh -c "python /tmp/flow.py -e https://test-nifi-node-default-0.test-nifi-node-default-headless.$NAMESPACE.svc.cluster.local:8443 run json /tmp/generate-and-log-flowfiles.json > /tmp/old_input" -{% else %} - - script: kubectl exec -n $NAMESPACE test-nifi-0 -- sh -c "python /tmp/flow.py -e https://test-nifi-node-default-0.test-nifi-node-default-headless.$NAMESPACE.svc.cluster.local:8443 run template /tmp/generate-and-log-flowfiles.xml > /tmp/old_input" -{% endif %} + - script: kubectl exec -n $NAMESPACE test-nifi-0 -- sh -c "python /tmp/flow.py > /tmp/old_input" # This tests if the output contains an Error or zero flow files are queued, which also indicates that something went wrong - script: kubectl exec -n $NAMESPACE test-nifi-0 -- sh -c "cat /tmp/old_input | grep -Eov 'Error|\b0\b'" diff --git a/tests/templates/kuttl/upgrade/04-prepare-test-nifi.yaml.j2 b/tests/templates/kuttl/upgrade/04-prepare-test-nifi.yaml.j2 index fc488c1e..7204710b 100644 --- a/tests/templates/kuttl/upgrade/04-prepare-test-nifi.yaml.j2 +++ b/tests/templates/kuttl/upgrade/04-prepare-test-nifi.yaml.j2 @@ -2,12 +2,7 @@ apiVersion: kuttl.dev/v1beta1 kind: TestStep commands: - - script: kubectl cp -n $NAMESPACE ./test_nifi_metrics.py test-nifi-0:/tmp - script: kubectl cp -n $NAMESPACE ./test_nifi.py test-nifi-0:/tmp - script: kubectl cp -n $NAMESPACE ./flow.py test-nifi-0:/tmp -{% if test_scenario['values']['nifi_old'].split(',')[0] == '2.0.0' %} - - script: kubectl cp -n $NAMESPACE ./generate-and-log-flowfiles.json test-nifi-0:/tmp -{% else %} - - script: kubectl cp -n $NAMESPACE ./generate-and-log-flowfiles.xml test-nifi-0:/tmp -{% endif %} + - script: kubectl cp -n $NAMESPACE ./generate-and-log-flowfiles.json test-nifi-0:/tmp - script: kubectl cp -n $NAMESPACE ./cacert.pem test-nifi-0:/tmp diff --git a/tests/templates/kuttl/upgrade/05-assert.yaml.j2 b/tests/templates/kuttl/upgrade/06-assert.yaml.j2 similarity index 100% rename from tests/templates/kuttl/upgrade/05-assert.yaml.j2 rename to tests/templates/kuttl/upgrade/06-assert.yaml.j2 diff --git a/tests/templates/kuttl/upgrade/06-sleep.yaml b/tests/templates/kuttl/upgrade/06-sleep.yaml new file mode 100644 index 00000000..5030b27a --- /dev/null +++ b/tests/templates/kuttl/upgrade/06-sleep.yaml @@ -0,0 +1,8 @@ +--- +apiVersion: kuttl.dev/v1beta1 +kind: TestStep +commands: + # Sleep for a little while to ensure the pod rotation of the NiFi cluster has started. Otherwise, + # the assertion in 06 will immediately succeed because it is run _before_ the pod rotation has + # started. + - script: sleep 15 diff --git a/tests/templates/kuttl/upgrade/07-assert.yaml.j2 b/tests/templates/kuttl/upgrade/07-assert.yaml.j2 index 6f10d137..dae1ecfc 100644 --- a/tests/templates/kuttl/upgrade/07-assert.yaml.j2 +++ b/tests/templates/kuttl/upgrade/07-assert.yaml.j2 @@ -6,9 +6,6 @@ metadata: timeout: 300 commands: - script: kubectl exec -n $NAMESPACE test-nifi-0 -- python /tmp/test_nifi.py -u admin -p supersecretpassword -n $NAMESPACE -c 3 -{% if test_scenario['values']['nifi_new'].split(',')[0].startswith('1.') %} - - script: kubectl exec -n $NAMESPACE test-nifi-0 -- python /tmp/test_nifi_metrics.py -n $NAMESPACE -{% endif %} - script: kubectl exec -n $NAMESPACE test-nifi-0 -- sh -c "python /tmp/flow.py -e https://test-nifi-node-default-0.test-nifi-node-default-headless.$NAMESPACE.svc.cluster.local:8443 query > /tmp/new_input" # This tests if the output contains an Error or zero flow files are queued, which also indicates that something went wrong - script: kubectl exec -n $NAMESPACE test-nifi-0 -- sh -c "cat /tmp/new_input | grep -Eov 'Error|\b0\b'" diff --git a/tests/templates/kuttl/upgrade/flow.py b/tests/templates/kuttl/upgrade/flow.py old mode 100755 new mode 100644 index 61834b43..1853a126 --- a/tests/templates/kuttl/upgrade/flow.py +++ b/tests/templates/kuttl/upgrade/flow.py @@ -1,6 +1,6 @@ #!/usr/bin/env python3 -# This script is used to create a flow from a template, run it for a couple -# of seconds, stop it and then query the number of queued flow files. +# This script is used to create a process group, run it for a couple of seconds, stop it and then +# query the number of queued flow files. import nipyapi from nipyapi.canvas import ( @@ -11,25 +11,19 @@ recurse_flow, ) from nipyapi.security import service_login -from nipyapi.templates import get_template, upload_template, deploy_template -from nipyapi.nifi.models import FlowEntity +from nipyapi.nifi.api_client import ApiClient -from typing import Union, List +from typing import Union from dataclasses import dataclass -import re import os -import sys import time -import argparse import urllib3 +ENDPOINT = f"https://test-nifi-node-default-0.test-nifi-node-default-headless.{os.environ['NAMESPACE']}.svc.cluster.local:8443" USERNAME = "admin" -PASSWORD = "supersecretpassword" -TEMPLATE_NAME = "generate-and-log-flowfiles" -TEMPLATE_FILE = f"{TEMPLATE_NAME}.xml" -JSON_NAME = "generate-and-log-flowfiles" -JSON_FILE = f"{JSON_NAME}.json" +PASSWORD = "supersecretpassword" # Can we use open("/nifi-users/admin").read() here? +FLOWFILE = "/tmp/generate-and-log-flowfiles.json" CONNECTION_NAME = "output-connection" @@ -38,170 +32,83 @@ class Error: message: str -def parse_args(args: List[str]) -> argparse.Namespace: - parser = argparse.ArgumentParser(description="Setup a flow and query the status.") - parser.add_argument( - "-e", - "--endpoint", - help="Nifi endpoint URL (must be https). As of 2022-08-29 we cant use https://nifi:8443 here because Nifi's check for the host header validity fails. The error message looks like:

The request contained an invalid host header [nifi:8443] in the request [/nifi-api]. Check for request manipulation or third-party intercept.

", - required=True, - ) - parser.add_argument( - "-u", - "--user", - help="Nifi login user.", - ) - parser.add_argument( - "-p", - "--password", - help="Nifi login password.", - ) - subparsers = parser.add_subparsers(dest="subcommand", required=True) - run_parser = subparsers.add_parser( - "run", - help="Install and run a flow.", - ) - format_parser = run_parser.add_subparsers(dest="formatcommand") - template_parser = format_parser.add_parser( - "template", help="Use the template format for the flow." - ) - template_parser.add_argument( - "template", - help="Nifi template file.", - ) - json_parser = format_parser.add_parser( - "json", help="Use the json format for the flow." - ) - json_parser.add_argument( - "json", - help="Nifi json file.", - ) - subparsers.add_parser( - "query", - help="Query the flow.", - ) - return parser.parse_args(args) - - -def login(endpoint: str, user: str, passwd: str) -> bool: - nipyapi.config.nifi_config.host = f"{endpoint}/nifi-api" - nipyapi.config.nifi_config.verify_ssl = False - return service_login(username=user, password=passwd) - - -def flow_from_template( - pg_id: str, template_file: str, template_name: str -) -> FlowEntity: - upload_template(pg_id, template_file) - template_id = get_template(template_name).id - return deploy_template(pg_id, template_id, 200, 0) +urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) def schedule(pg_id: str, toggle: bool) -> bool: for controller in list_all_controllers(pg_id): schedule_controller(controller, scheduled=toggle) + return schedule_process_group(pg_id, scheduled=toggle) def flow_files_queued(connection_name: str) -> Union[str, Error]: """Returns the current input record number for the given component""" flow = recurse_flow() - c = [ - c - for c in flow.process_group_flow.flow.connections - if c.status.name == connection_name - ] - if c: - return c[0].status.aggregate_snapshot.flow_files_queued - else: - # for nifi 2.x the flow is loaded via api into a new process group - # we need to look for the connection in that process group instead - if flow.process_group_flow.flow.process_groups: - inner_flow = recurse_flow(flow.process_group_flow.flow.process_groups[0].id) - inner_c = [ - inner_c - for inner_c in inner_flow.process_group_flow.flow.connections - if inner_c.status.name == connection_name - ] - if inner_c: - return inner_c[0].status.aggregate_snapshot.flow_files_queued - else: - return Error(f"No connection named '{connection_name}' found.") + + if flow.process_group_flow.flow.process_groups: + inner_flow = recurse_flow(flow.process_group_flow.flow.process_groups[0].id) + conn = [ + conn + for conn in inner_flow.process_group_flow.flow.connections + if conn.status.name == connection_name + ] + if conn: + return conn[0].status.aggregate_snapshot.flow_files_queued else: return Error(f"No connection named '{connection_name}' found.") + else: + return Error(f"No connection named '{connection_name}' found.") def main(): - # turn this on to debug Nifi api calls. Also need to import logging - # logging.getLogger().setLevel(logging.INFO) - - args = parse_args(sys.argv[1:]) - - urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) - - username = args.user or USERNAME - password = args.password or PASSWORD - - if args.subcommand == "run": - login(args.endpoint, username, password) - pg_id = get_root_pg_id() - - if args.formatcommand == "template": - template_file = args.template or TEMPLATE_FILE - template_name = re.sub(r"\..+$", "", os.path.basename(template_file)) - - flow_from_template(pg_id, template_file, template_name) - - elif args.formatcommand == "json": - json_file = args.json or JSON_FILE - json_name = re.sub(r"\..+$", "", os.path.basename(json_file)) - - if not nipyapi.config.nifi_config.api_client: - nipyapi.config.nifi_config.api_client = nipyapi.nifi.ApiClient() - - header_params = {} - header_params["Accept"] = ( - nipyapi.config.nifi_config.api_client.select_header_accept( - ["application/json"] - ) - ) - header_params["Content-Type"] = ( - nipyapi.config.nifi_config.api_client.select_header_content_type( - ["multipart/form-data"] - ) - ) - - nipyapi.config.nifi_config.api_client.call_api( - "/process-groups/{pg_id}/process-groups/upload", - "POST", - path_params={"pg_id": pg_id}, - header_params=header_params, - _return_http_data_only=True, - post_params=[ - ("id", pg_id), - ("groupName", json_name), - ("positionX", 100), - ("positionY", 10), - ("clientId", nipyapi.nifi.FlowApi().generate_client_id()), - ], - files={"file": json_file}, - auth_settings=["tokenAuth"], - ) - - # get the process group id of the newly created process group to pass it to the schedule() function - # for nifi 2.x we need to schedule the inner process group and instead of the root process group - flow = recurse_flow() - inner_flow_id = flow.process_group_flow.flow.process_groups[0].id - pg_id = inner_flow_id - - schedule(pg_id, True) # start - time.sleep(5) # give the flow 5 seconds to run - schedule(pg_id, False) # stop - print(flow_files_queued(CONNECTION_NAME)) - - elif args.subcommand == "query": - login(args.endpoint, username, password) - print(flow_files_queued(CONNECTION_NAME)) + nipyapi.config.nifi_config.host = f"{ENDPOINT}/nifi-api" + nipyapi.config.nifi_config.verify_ssl = False + + print(f"Logging in as {USERNAME}") + service_login(username=USERNAME, password=PASSWORD) + print("Logged in") + + pg_id = get_root_pg_id() + print(f"Got root process group id: {pg_id}") + + if not nipyapi.config.nifi_config.api_client: + nipyapi.config.nifi_config.api_client = ApiClient() + + header_params = {} + header_params["Accept"] = ( + nipyapi.config.nifi_config.api_client.select_header_accept(["application/json"]) + ) + header_params["Content-Type"] = ( + nipyapi.config.nifi_config.api_client.select_header_content_type( + ["multipart/form-data"] + ) + ) + + print("Uploading process group") + nipyapi.config.nifi_config.api_client.call_api( + "/process-groups/{pg_id}/process-groups/upload", + "POST", + path_params={"pg_id": pg_id}, + header_params=header_params, + _return_http_data_only=True, + post_params=[ + ("id", pg_id), + ("groupName", "Upgrade Test"), + ("positionX", 100), + ("positionY", 10), + ("clientId", nipyapi.nifi.FlowApi().generate_client_id()), + ], + files={"file": FLOWFILE}, + auth_settings=["tokenAuth"], + ) + print("Process group uploaded") + + schedule(pg_id, True) # Start the flow + time.sleep(5) # Let the flow run for 5 seconds + schedule(pg_id, False) # Stop the flow + + print(flow_files_queued(CONNECTION_NAME)) if __name__ == "__main__": diff --git a/tests/templates/kuttl/upgrade/generate-and-log-flowfiles.xml b/tests/templates/kuttl/upgrade/generate-and-log-flowfiles.xml deleted file mode 100644 index 547e69f4..00000000 --- a/tests/templates/kuttl/upgrade/generate-and-log-flowfiles.xml +++ /dev/null @@ -1,327 +0,0 @@ - -