bbende commented on code in PR #7092: URL: https://github.com/apache/nifi/pull/7092#discussion_r1157769951
########## nifi-toolkit/nifi-toolkit-cli/src/main/java/org/apache/nifi/toolkit/cli/impl/command/registry/flow/ExportAllFlows.java: ########## @@ -0,0 +1,182 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.nifi.toolkit.cli.impl.command.registry.flow; + +import org.apache.commons.cli.ParseException; +import org.apache.nifi.registry.bucket.Bucket; +import org.apache.nifi.registry.client.NiFiRegistryClient; +import org.apache.nifi.registry.client.NiFiRegistryException; +import org.apache.nifi.registry.flow.VersionedFlow; +import org.apache.nifi.registry.flow.VersionedFlowSnapshot; +import org.apache.nifi.registry.flow.VersionedFlowSnapshotMetadata; +import org.apache.nifi.toolkit.cli.api.CommandException; +import org.apache.nifi.toolkit.cli.api.Context; +import org.apache.nifi.toolkit.cli.impl.command.CommandOption; +import org.apache.nifi.toolkit.cli.impl.command.registry.AbstractNiFiRegistryCommand; +import org.apache.nifi.toolkit.cli.impl.command.registry.bucket.ListBuckets; +import org.apache.nifi.toolkit.cli.impl.result.registry.VersionedFlowSnapshotsResult; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.Properties; +import java.util.function.Function; +import java.util.stream.Collectors; + +public class ExportAllFlows extends AbstractNiFiRegistryCommand<VersionedFlowSnapshotsResult> { + private static final String ALL_BUCKETS_COLLECTED = "All buckets collected..."; + private static final String ALL_FLOWS_COLLECTED = "All flows collected..."; + private static final String ALL_FLOW_VERSIONS_COLLECTED = "All flow versions collected..."; + private static final String EXPORTING_FLOW_VERSIONS = "Exporting flow versions..."; + private final ListBuckets listBuckets; + private final ListFlows listFlows; + private final ListFlowVersions listFlowVersions; + private final ExportFlowVersion exportFlowVersion; + + public ExportAllFlows() { + super("export-all-flows", VersionedFlowSnapshotsResult.class); + this.listBuckets = new ListBuckets(); + this.listFlows = new ListFlows(); + this.listFlowVersions = new ListFlowVersions(); + this.exportFlowVersion = new ExportFlowVersion(); + } + + @Override + public void doInitialize(final Context context) { + addOption(CommandOption.OUTPUT_DIR.createOption()); + + listBuckets.initialize(context); + listFlows.initialize(context); + listFlowVersions.initialize(context); + exportFlowVersion.initialize(context); + } + + @Override + public String getDescription() { + return "List all the buckets, for each bucket, list all the flows, for each flow, list all versions and export each version." + + "Versions will be saved in the provided target directory."; + } + + @Override + public VersionedFlowSnapshotsResult doExecute(NiFiRegistryClient client, Properties properties) throws IOException, NiFiRegistryException, ParseException, CommandException { + final String outputDirectory = getRequiredArg(properties, CommandOption.OUTPUT_DIR); + final boolean isInteractive = getContext().isInteractive(); + + // Gather all buckets and create a map for quick access by bucket id + final Map<String, Bucket> bucketMap = getBucketMap(client, isInteractive); + + // Gather all flows and create a map for quick access by flow id + final Map<String, VersionedFlow> flowMap = getFlowMap(client, bucketMap, isInteractive); + + // Gather all versions for all the flows + final List<VersionedFlowSnapshotMetadata> versionedFlowSnapshotMetadataList = getVersionedFlowSnapshotMetadataList(client, flowMap, isInteractive); + + // Prepare flow version exports + final List<VersionedFlowSnapshot> versionedFlowSnapshotList = getVersionedFlowSnapshotResults(client, outputDirectory, bucketMap, flowMap, versionedFlowSnapshotMetadataList, isInteractive); Review Comment: I think the way the result classes worked kind of led it to be implemented this way, but I agree with David's concern that we should see if we can write out the snapshots one-by-one. Maybe we can pass some kind iterator/helper object to the result class that allow it to keep calling next() and writing out the result? ########## nifi-toolkit/nifi-toolkit-cli/src/main/java/org/apache/nifi/toolkit/cli/impl/command/registry/flow/ImportAllFlows.java: ########## @@ -0,0 +1,298 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.nifi.toolkit.cli.impl.command.registry.flow; + +import com.fasterxml.jackson.databind.ObjectMapper; +import org.apache.commons.cli.MissingOptionException; +import org.apache.commons.cli.ParseException; +import org.apache.commons.lang3.tuple.ImmutablePair; +import org.apache.commons.lang3.tuple.Pair; +import org.apache.curator.shaded.com.google.common.collect.ComparisonChain; +import org.apache.nifi.registry.bucket.Bucket; +import org.apache.nifi.registry.client.BucketClient; +import org.apache.nifi.registry.client.FlowClient; +import org.apache.nifi.registry.client.NiFiRegistryClient; +import org.apache.nifi.registry.client.NiFiRegistryException; +import org.apache.nifi.registry.flow.VersionedFlow; +import org.apache.nifi.registry.flow.VersionedFlowSnapshot; +import org.apache.nifi.registry.flow.VersionedFlowSnapshotMetadata; +import org.apache.nifi.toolkit.cli.api.CommandException; +import org.apache.nifi.toolkit.cli.api.Context; +import org.apache.nifi.toolkit.cli.impl.command.CommandOption; +import org.apache.nifi.toolkit.cli.impl.command.registry.AbstractNiFiRegistryCommand; +import org.apache.nifi.toolkit.cli.impl.command.registry.bucket.ListBuckets; +import org.apache.nifi.toolkit.cli.impl.result.StringResult; +import org.apache.nifi.toolkit.cli.impl.util.JacksonUtils; + +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Properties; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +public class ImportAllFlows extends AbstractNiFiRegistryCommand<StringResult> { + private static final String FILE_NAME_PREFIX = "toolkit_registry_export_all_"; + private static final String SKIPPING_BUCKET_CREATION = " already exists, skipping bucket creation..."; + private static final String SKIPPING_IMPORT = " already exists, skipping import..."; + private static final String SKIPPING_FLOW_CREATION = " already exists, skipping flow creation..."; + private static final String IMPORT_COMPLETED = "Import completed..."; + private static final String ALL_BUCKETS_COLLECTED = "All buckets collected..."; + private static final String ALL_FLOWS_COLLECTED = "All flows collected..."; + private static final String ALL_FLOW_VERSIONS_COLLECTED = "All flow versions collected..."; + private final ListBuckets listBuckets; + private final ListFlows listFlows; + private final ListFlowVersions listFlowVersions; + private final ImportFlowVersion importFlowVersion; + + public ImportAllFlows() { + super("import-all-flows", StringResult.class); + this.listBuckets = new ListBuckets(); + this.listFlows = new ListFlows(); + this.listFlowVersions = new ListFlowVersions(); + this.importFlowVersion = new ImportFlowVersion(); + } + + @Override + protected void doInitialize(Context context) { + addOption(CommandOption.INPUT_SOURCE.createOption()); + addOption(CommandOption.SKIP.createOption()); + + listBuckets.initialize(context); + listFlows.initialize(context); + listFlowVersions.initialize(context); + importFlowVersion.initialize(context); + } + + @Override + public String getDescription() { + return "From a provided directory as input, the directory content must be generated by the export-all-flows command, " + + "based on the file contents, the corresponding buckets, flows and flow versions will be created." + + "If not configured otherwise, already existing objects will be skipped."; + } + + @Override + public StringResult doExecute(final NiFiRegistryClient client, final Properties properties) throws IOException, NiFiRegistryException, ParseException, CommandException { + final boolean skip = Boolean.parseBoolean(getRequiredArg(properties, CommandOption.SKIP)); + final boolean isInteractive = getContext().isInteractive(); + + //Gather all buckets and create a map for easier search by bucket name + final Map<String, String> bucketMap = getBucketMap(client, isInteractive); + + // Gather all flows and create a map for easier search by flow name. + // As flow name is only unique within the same bucket we need to use the bucket id in the key as well + final Map<Pair<String, String>, String> flowMap = getFlowMap(client, bucketMap, isInteractive); + final Map<Pair<String, String>, String> flowCreated = new HashMap<>(); + + // Gather all flow versions and create a map for easier search by flow id + final Map<String, List<Integer>> versionMap = getVersionMap(client, flowMap, isInteractive); + + // Create file path list + final List<String> files = getFilePathList(properties); + + // Deserialize file content + final List<ImportedSnapshot> importedSnapshots = deserializeSnapshots(files); + + // As we need to keep the version order the snapshot list needs to be sorted + importedSnapshots.sort((o1, o2) -> ComparisonChain.start() + .compare(o1.getSnapshot().getBucket().getName(), o2.getSnapshot().getBucket().getName()) + .compare(o1.getSnapshot().getFlow().getName(), o2.getSnapshot().getFlow().getName()) + .compare(o1.getSnapshot().getSnapshotMetadata().getVersion(), o2.getSnapshot().getSnapshotMetadata().getVersion()) + .result()); + + for (final ImportedSnapshot snapshot : importedSnapshots) { + + final String inputSource = snapshot.getInputSource(); + final String bucketName = snapshot.getSnapshot().getBucket().getName(); + final String bucketDescription = snapshot.getSnapshot().getBucket().getDescription(); + final String flowName = snapshot.getSnapshot().getFlow().getName(); + final String flowDescription = snapshot.getSnapshot().getFlow().getDescription(); + final int flowVersion = snapshot.getSnapshot().getSnapshotMetadata().getVersion(); + // The original bucket and flow ids must be kept otherwise NiFi won't be able to synchronize with the NiFi Registry + final String bucketId = snapshot.getSnapshot().getBucket().getIdentifier(); + final String flowId = snapshot.getSnapshot().getFlow().getIdentifier(); + + // Create bucket if missing + if (bucketMap.containsKey(bucketName)) { + printMessage(isInteractive, bucketName + SKIPPING_BUCKET_CREATION); + } else { + createBucket(client, bucketMap, bucketName, bucketDescription, bucketId); + } + + // Create flow if missing + if (flowMap.containsKey(new ImmutablePair<>(bucketId, flowName))) { + if (skip) { + printMessage(isInteractive, flowName + SKIPPING_IMPORT); + continue; + } else { + printMessage(isInteractive, flowName + SKIPPING_FLOW_CREATION); + } + } else if (!flowCreated.containsKey(new ImmutablePair<>(bucketId, flowName))) { + createFlow(client, flowCreated, flowId, flowName, flowDescription, bucketId); + } + + // Create missing flow versions + if (!versionMap.getOrDefault(flowId, Collections.emptyList()).contains(flowVersion)) { + createFlowVersion(client, inputSource, flowId); Review Comment: What is the expectation for handling nested versioning during a migration? For example, `Outer PG` under version control, and inside it there is `Child PG` also under version control. In registry, the snapshot of `Outer PG` will have a process group for `Child PG` with coordinates pointing to the registry it came from: ``` "versionedFlowCoordinates" : { "storageLocation" : "http://localhost:18080/nifi-registry-api/buckets/876e0978-ffe5-43c5-ab2c-69dd0a8d3c52/flows/412512c1-c95f-4a2a-9df3-8a988e326df7/versions/1";, "registryUrl" : "http://localhost:18080";, "bucketId" : "876e0978-ffe5-43c5-ab2c-69dd0a8d3c52", "flowId" : "412512c1-c95f-4a2a-9df3-8a988e326df7", "version" : 1 } ``` If these commands were being used to replicate/move between two different registries at different locations, then when moving `Outer PG` to the new registry, it would need to update `storageLocation` and `registryUrl` to the location of the new registry. Otherwise when a NiFi instance pulls `Outer PG` from the new registry, it will then come to 'Child PG' and try to pull it from the old registry, or fail because no registry client exists for the URL of the old registry. ########## nifi-toolkit/nifi-toolkit-cli/src/main/java/org/apache/nifi/toolkit/cli/impl/command/registry/flow/ImportAllFlows.java: ########## @@ -0,0 +1,298 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.nifi.toolkit.cli.impl.command.registry.flow; + +import com.fasterxml.jackson.databind.ObjectMapper; +import org.apache.commons.cli.MissingOptionException; +import org.apache.commons.cli.ParseException; +import org.apache.commons.lang3.tuple.ImmutablePair; +import org.apache.commons.lang3.tuple.Pair; +import org.apache.curator.shaded.com.google.common.collect.ComparisonChain; +import org.apache.nifi.registry.bucket.Bucket; +import org.apache.nifi.registry.client.BucketClient; +import org.apache.nifi.registry.client.FlowClient; +import org.apache.nifi.registry.client.NiFiRegistryClient; +import org.apache.nifi.registry.client.NiFiRegistryException; +import org.apache.nifi.registry.flow.VersionedFlow; +import org.apache.nifi.registry.flow.VersionedFlowSnapshot; +import org.apache.nifi.registry.flow.VersionedFlowSnapshotMetadata; +import org.apache.nifi.toolkit.cli.api.CommandException; +import org.apache.nifi.toolkit.cli.api.Context; +import org.apache.nifi.toolkit.cli.impl.command.CommandOption; +import org.apache.nifi.toolkit.cli.impl.command.registry.AbstractNiFiRegistryCommand; +import org.apache.nifi.toolkit.cli.impl.command.registry.bucket.ListBuckets; +import org.apache.nifi.toolkit.cli.impl.result.StringResult; +import org.apache.nifi.toolkit.cli.impl.util.JacksonUtils; + +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Properties; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +public class ImportAllFlows extends AbstractNiFiRegistryCommand<StringResult> { + private static final String FILE_NAME_PREFIX = "toolkit_registry_export_all_"; + private static final String SKIPPING_BUCKET_CREATION = " already exists, skipping bucket creation..."; + private static final String SKIPPING_IMPORT = " already exists, skipping import..."; + private static final String SKIPPING_FLOW_CREATION = " already exists, skipping flow creation..."; + private static final String IMPORT_COMPLETED = "Import completed..."; + private static final String ALL_BUCKETS_COLLECTED = "All buckets collected..."; + private static final String ALL_FLOWS_COLLECTED = "All flows collected..."; + private static final String ALL_FLOW_VERSIONS_COLLECTED = "All flow versions collected..."; + private final ListBuckets listBuckets; + private final ListFlows listFlows; + private final ListFlowVersions listFlowVersions; + private final ImportFlowVersion importFlowVersion; + + public ImportAllFlows() { + super("import-all-flows", StringResult.class); + this.listBuckets = new ListBuckets(); + this.listFlows = new ListFlows(); + this.listFlowVersions = new ListFlowVersions(); + this.importFlowVersion = new ImportFlowVersion(); + } + + @Override + protected void doInitialize(Context context) { + addOption(CommandOption.INPUT_SOURCE.createOption()); + addOption(CommandOption.SKIP.createOption()); + + listBuckets.initialize(context); + listFlows.initialize(context); + listFlowVersions.initialize(context); + importFlowVersion.initialize(context); + } + + @Override + public String getDescription() { + return "From a provided directory as input, the directory content must be generated by the export-all-flows command, " + + "based on the file contents, the corresponding buckets, flows and flow versions will be created." + + "If not configured otherwise, already existing objects will be skipped."; + } + + @Override + public StringResult doExecute(final NiFiRegistryClient client, final Properties properties) throws IOException, NiFiRegistryException, ParseException, CommandException { + final boolean skip = Boolean.parseBoolean(getRequiredArg(properties, CommandOption.SKIP)); + final boolean isInteractive = getContext().isInteractive(); + + //Gather all buckets and create a map for easier search by bucket name + final Map<String, String> bucketMap = getBucketMap(client, isInteractive); + + // Gather all flows and create a map for easier search by flow name. + // As flow name is only unique within the same bucket we need to use the bucket id in the key as well + final Map<Pair<String, String>, String> flowMap = getFlowMap(client, bucketMap, isInteractive); + final Map<Pair<String, String>, String> flowCreated = new HashMap<>(); + + // Gather all flow versions and create a map for easier search by flow id + final Map<String, List<Integer>> versionMap = getVersionMap(client, flowMap, isInteractive); + + // Create file path list + final List<String> files = getFilePathList(properties); + + // Deserialize file content + final List<ImportedSnapshot> importedSnapshots = deserializeSnapshots(files); Review Comment: I'm wondering if we can change the filenames that are produced by export to contain the 3 fields below that we are sorting on, then we can sort the `List<String> files` in the same order? So exported filenames would be: ``` <Bucket-Name>-<Flow-Name>-<Version>.json ``` This would also have the added benefit of being human readable for anyone looking at the exported files. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
