bbende commented on code in PR #7092: URL: https://github.com/apache/nifi/pull/7092#discussion_r1170134766
########## nifi-docs/src/main/asciidoc/toolkit-guide.adoc: ########## @@ -768,6 +770,325 @@ This command migrates nifi.properties back from AWS_KMS to AES_GCM protection sc -v ---- +[export_import_all_flows] +=== Export All Flows +You can use the `export-all-flows` to perform the following tasks: + +* List all the buckets +* For each bucket, list all flows +* For each flow, list all versions +* Export each version into a provided directory + +Running the command requires an `--outputDirectory` parameter. + +=== Import All Flows +You can use the `import-all-flows` to perform the following tasks: + +* List all files, representing a flow version, from a directory created by export-all-flows +* Create all the corresponding buckets +* Create all the corresponding flows +* Import all the corresponding flow versions + +Running the command requires 2 parameters: + +* `--input` parameter represents a directory to read files from +* `--skipExisting` parameter, configuring how to handle existing flow and flow version creation. +If true the flow and flow version creation will be skipped regardless of there are missing flow versions. +If false the missing flow versions will be created. The default value is true, skip creation. + +=== Usage +The input source for an import-all-flows command must be created by an export-all-flows command. +To avoid migration conflicts, no modification should be performed in the NiFi Registry during this activity. +Buckets and flows with the same name are considered equal. + +* Export all flow versions: + + ./bin/cli.sh registry export-all-flows -u http://localhost:18080 --outputDirectory "/my_dir/flow_exports" + +* Import all flow versions: + + ./bin/cli.sh registry import-all-flows -u http://localhost:18080 --input "/my_dir/flow_exports" --skipExisting false + +=== Expected behaviour +=== Use case 1: reconfiguring an existing NiFi Registry + +NiFi is connecting to NiFi Registry, the NiFi Registry does not change, only its configuration. +All the data will be created. + +1. Export versions: + + ./bin/cli.sh registry export-all-flows -u http://localhost:18080 --outputDirectory "/my_dir/flow_exports" + +2. Stop registry + +3. Switch provider + +4. Start registry + +5. Import versions + + ./bin/cli.sh registry import-all-flows -u http://localhost:18080 --input "/my_dir/flow_exports" --skipExisting true + + +=== Use case 2: data replication + +NiFi_1 is connecting to NiFi Registry_1 and NiFi_2 is connecting to NiFi Registry_2. + +For disaster recovery purposes the data from NiFi Registry_1 needs to be periodically replicated to NiFi Registry_2 via a scheduled job. + +The initial version of Nifi Registry_2 needs to be created by this tool. + +The missing buckets, flows and versions will be created. If bucket and flow exist the missing versions will be created. + +1. Export versions: + + ./bin/cli.sh registry export-all-flows -u http://nifi-registry-1:18080 --outputDirectory "/my_dir/flow_exports" + +2. Import versions: + + ./bin/cli.sh registry import-all-flows -u http://nifi-registry-2:18080 --input "/my_dir/flow_exports" --skipExisting false + + Review Comment: I think this is adding back a bunch of sections that were removed on main. Need to update so that this is only adding your new section, and everything afterwards starting with `File Manager` down to `tls_toolkit` is removed. ########## nifi-toolkit/nifi-toolkit-cli/src/main/java/org/apache/nifi/toolkit/cli/impl/command/registry/flow/ImportAllFlows.java: ########## @@ -0,0 +1,331 @@ +/* + * 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.flow.VersionedFlowCoordinates; +import org.apache.nifi.flow.VersionedProcessGroup; +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.FlowSnapshotClient; +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 static final String FILE_NAME_SEPARATOR = "_"; + private static final String STORAGE_LOCATION_URL = "%s/nifi-registry-api/buckets/%s/flows/%s/versions/%s"; + private static final ObjectMapper MAPPER = JacksonUtils.getObjectMapper(); + private final ListBuckets listBuckets; + private final ListFlows listFlows; + private final ListFlowVersions listFlowVersions; + + public ImportAllFlows() { + super("import-all-flows", StringResult.class); + this.listBuckets = new ListBuckets(); + this.listFlows = new ListFlows(); + this.listFlowVersions = new ListFlowVersions(); + } + + @Override + protected void doInitialize(Context context) { + addOption(CommandOption.INPUT_SOURCE.createOption()); + addOption(CommandOption.SKIP_EXISTING.createOption()); + + listBuckets.initialize(context); + listFlows.initialize(context); + listFlowVersions.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_EXISTING)); + 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<VersionFileMetaData> files = getFilePathList(properties); + + // As we need to keep the version order the list needs to be sorted + files.sort((o1, o2) -> ComparisonChain.start() + .compare(o1.getBucketName(), o2.getBucketName()) + .compare(o1.getFlowName(), o2.getFlowName()) + .compare(o1.getVersion(), o2.getVersion()) + .result()); + + for (VersionFileMetaData file : files) { + final String inputSource = file.getInputSource(); + final String fileContent = getInputSourceContent(inputSource); + final VersionedFlowSnapshot snapshot = MAPPER.readValue(fileContent, VersionedFlowSnapshot.class); + + final String bucketName = snapshot.getBucket().getName(); + final String bucketDescription = snapshot.getBucket().getDescription(); + final String flowName = snapshot.getFlow().getName(); + final String flowDescription = snapshot.getFlow().getDescription(); + final int flowVersion = snapshot.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 flowId = snapshot.getFlow().getIdentifier(); + final String bucketId = snapshot.getBucket().getIdentifier(); Review Comment: Could we print a message at the beginning and end of this loop like? ``` Importing <flow Name> - <flowVersion> to <bucketName> ... Successfully imported <flow Name> - <flowVersion> to <bucketName> ``` I think this would help keep track of the current messages to know where it is during the import. I guess the successful message would only print if not skipping. ########## nifi-toolkit/nifi-toolkit-cli/src/main/java/org/apache/nifi/toolkit/cli/impl/result/registry/VersionedFlowSnapshotsResult.java: ########## @@ -0,0 +1,77 @@ +/* + * 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.result.registry; + +import org.apache.commons.lang3.Validate; +import org.apache.nifi.registry.flow.VersionedFlowSnapshot; +import org.apache.nifi.toolkit.cli.api.WritableResult; +import org.apache.nifi.toolkit.cli.impl.util.JacksonUtils; + +import java.io.IOException; +import java.io.OutputStream; +import java.io.PrintStream; +import java.nio.file.Files; +import java.nio.file.Paths; +import java.util.Iterator; + +/** + * Result for a list of VersionedFlowSnapshots. + * + * If this result was created with a non-null exportDirectoryName, then the write method will ignore + * the passed in PrintStream, and will write the serialized snapshot to the given directory. + * The file name will be generated from the flow name and its version. + * + * If this result was created with a null exportDirectoryName, then the write method will write the + * serialized snapshots to the given PrintStream. + */ +public class VersionedFlowSnapshotsResult implements WritableResult<Iterator<VersionedFlowSnapshot>> { + private static final String FILE_NAME_PREFIX = "toolkit_registry_export_all"; + private static final String EXPORT_FILE_NAME = "%s/%s_%s_%s_%d"; + private static final String SEPARATOR = "_"; + private static final String REPLACEMENT = "-"; + private final Iterator<VersionedFlowSnapshot> versionedFlowSnapshots; + private final String exportDirectoryName; + + public VersionedFlowSnapshotsResult(final Iterator<VersionedFlowSnapshot> versionedFlowSnapshots, final String exportDirectoryName) { + this.versionedFlowSnapshots = versionedFlowSnapshots; + this.exportDirectoryName = exportDirectoryName; + Validate.notNull(this.versionedFlowSnapshots); + } + + @Override + public Iterator<VersionedFlowSnapshot> getResult() { + return versionedFlowSnapshots; + } + + @Override + public void write(final PrintStream output) throws IOException { + while (versionedFlowSnapshots.hasNext()) { + final VersionedFlowSnapshot versionedFlowSnapshot = versionedFlowSnapshots.next(); + if (exportDirectoryName != null) { + final String bucketName = versionedFlowSnapshot.getBucket().getName().replaceAll(SEPARATOR, REPLACEMENT); + final String flowName = versionedFlowSnapshot.getFlow().getName().replaceAll(SEPARATOR, REPLACEMENT); + final int version = versionedFlowSnapshot.getSnapshotMetadata().getVersion(); + final String exportFileName = String.format(EXPORT_FILE_NAME, exportDirectoryName, FILE_NAME_PREFIX, bucketName, flowName, version); + try (final OutputStream resultOut = Files.newOutputStream(Paths.get(exportFileName))) { + JacksonUtils.write(versionedFlowSnapshot, resultOut); + } Review Comment: If the output directory doesn't exist, currently the export fails, but it isn't very clear why... ``` #> registry export-all-flows --outputDirectory /tmp/registry-export -verbose All buckets collected... All flows collected... All flow versions collected... ERROR: /tmp/registry-export/toolkit_registry_export_all_B1_Resuable_1 ``` With adding `-verbose` we can see: ``` java.nio.file.NoSuchFileException: /tmp/registry-export/toolkit_registry_export_all_B1_Resuable_1 at java.base/sun.nio.fs.UnixException.translateToIOException(UnixException.java:92) at java.base/sun.nio.fs.UnixException.rethrowAsIOException(UnixException.java:111) at java.base/sun.nio.fs.UnixException.rethrowAsIOException(UnixException.java:116) at java.base/sun.nio.fs.UnixFileSystemProvider.newByteChannel(UnixFileSystemProvider.java:219) at java.base/java.nio.file.spi.FileSystemProvider.newOutputStream(FileSystemProvider.java:478) at java.base/java.nio.file.Files.newOutputStream(Files.java:220) at org.apache.nifi.toolkit.cli.impl.result.registry.VersionedFlowSnapshotsResult.write(VersionedFlowSnapshotsResult.java:69) at org.apache.nifi.toolkit.cli.impl.command.CommandProcessor.processCommand(CommandProcessor.java:256) at org.apache.nifi.toolkit.cli.impl.command.CommandProcessor.processGroupCommand(CommandProcessor.java:233) at org.apache.nifi.toolkit.cli.impl.command.CommandProcessor.process(CommandProcessor.java:188) at org.apache.nifi.toolkit.cli.CLIMain.runInteractiveCLI(CLIMain.java:124) at org.apache.nifi.toolkit.cli.CLIMain.main(CLIMain.java:68) ``` Match we should add something here like: ``` catch(Exception e) { throw new RuntimeException("Unable to write flow snapshot to: " + exportFileName, e); } ``` ? ########## nifi-registry/nifi-registry-core/nifi-registry-web-api/src/main/java/org/apache/nifi/registry/web/api/BucketFlowResource.java: ########## @@ -279,15 +279,21 @@ public Response createFlowVersion( @ApiParam(value = "The flow identifier") final String flowId, @ApiParam(value = "The new versioned flow snapshot.", required = true) - final VersionedFlowSnapshot snapshot) { + final VersionedFlowSnapshot snapshot, + @ApiParam( + value = "Whether source properties like author should be kept") + @QueryParam("preserveSourceProperties") + final boolean preserveSourceProperties) { verifyPathParamsMatchBody(bucketId, flowId, snapshot); // bucketId and flowId fields are optional in the body parameter, but required before calling the service layer setSnaphotMetadataIfMissing(bucketId, flowId, snapshot); - final String userIdentity = NiFiUserUtils.getNiFiUserIdentity(); - snapshot.getSnapshotMetadata().setAuthor(userIdentity); + if (!preserveSourceProperties) { + final String userIdentity = NiFiUserUtils.getNiFiUserIdentity(); + snapshot.getSnapshotMetadata().setAuthor(userIdentity); + } Review Comment: If `preserveSourceProperties` is `true` , we should enforce that `snapshot.snapshotMetadata.author` is not blank. It is possible some lower layer already enforces that author can't be blank/null, and if so then that is probably ok, but if not then we should enforce it here by throwing `BadRequestException`. ########## nifi-toolkit/nifi-toolkit-cli/src/main/java/org/apache/nifi/toolkit/cli/impl/command/registry/flow/ImportAllFlows.java: ########## @@ -0,0 +1,331 @@ +/* + * 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.flow.VersionedFlowCoordinates; +import org.apache.nifi.flow.VersionedProcessGroup; +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.FlowSnapshotClient; +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 static final String FILE_NAME_SEPARATOR = "_"; + private static final String STORAGE_LOCATION_URL = "%s/nifi-registry-api/buckets/%s/flows/%s/versions/%s"; + private static final ObjectMapper MAPPER = JacksonUtils.getObjectMapper(); + private final ListBuckets listBuckets; + private final ListFlows listFlows; + private final ListFlowVersions listFlowVersions; + + public ImportAllFlows() { + super("import-all-flows", StringResult.class); + this.listBuckets = new ListBuckets(); + this.listFlows = new ListFlows(); + this.listFlowVersions = new ListFlowVersions(); + } + + @Override + protected void doInitialize(Context context) { + addOption(CommandOption.INPUT_SOURCE.createOption()); + addOption(CommandOption.SKIP_EXISTING.createOption()); + + listBuckets.initialize(context); + listFlows.initialize(context); + listFlowVersions.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_EXISTING)); Review Comment: It seems like `SKIP_EXISTING` really only relates to skipping the import of versions for a flow when the flow already exists. So say my registry already has `Flow 1` with `v1` and `v2` and now I am trying to do an import that has the same flow, but has one additional version of `v3`. If I specify `--skipExisting true` then `v3` does not get imported, and if specify `--skipExisting false` then it does get imported. Is there a situation in one of the uses cases where you would want to use `--skipExisting true` and not get `v3` imported? If we do keep `--skipExisting`, I am thinking it might be clearer if the way it was used was to make it a non-required argument, and if it is not specified then the behavior is as if it was `--skipExisting false`. If they do specify it, then it can be an argument with no value and specifying `--skipExisting` means `--skipExisting true`. Default usage would just be: ``` registry import-all-flows -i /path/to/export ``` Thoughts? -- 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]
