You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@nifi.apache.org by "ferencerdei (via GitHub)" <gi...@apache.org> on 2024/06/03 08:41:53 UTC
Re: [PR] NIFI-13242 MiNiFi Sync Resource C2 command [nifi]
ferencerdei commented on code in PR #8898:
URL: https://github.com/apache/nifi/pull/8898#discussion_r1623931164
##########
c2/c2-client-bundle/c2-client-service/src/main/java/org/apache/nifi/c2/client/service/operation/C2OperationHandler.java:
##########
@@ -63,4 +68,43 @@ default boolean requiresRestart() {
* @return the result of the operation handling
*/
C2OperationAck handle(C2Operation operation);
+
+ /**
+ * Commonly used logic for creating an C2OperationState object
+ *
+ * @param operationState the state of the operation
+ * @param details additional status info to detail the state
+ * @return the created state
+ */
+ default C2OperationState operationState(C2OperationState.OperationState operationState, String details) {
+ C2OperationState state = new C2OperationState();
+ state.setState(operationState);
+ state.setDetails(details);
+ return state;
+ }
+
+ /**
+ * Commonly used logic for creating an C2OperationAck object
+ *
+ * @param operationId the identifier of the operation
+ * @param operationState the state of the operation
+ * @return the created operation ack object
+ */
+ default C2OperationAck operationAck(String operationId, C2OperationState operationState) {
+ C2OperationAck operationAck = new C2OperationAck();
+ operationAck.setOperationState(operationState);
+ operationAck.setOperationId(operationId);
+ return operationAck;
+ }
+
+ /**
+ * Commonly used logic for retrieving a string value from the operation arguments' map
+ *
+ * @param operation the operation with arguments
+ * @param argument the name of the argument to retrieve
+ * @return the retrieved argument value
+ */
+ default String getOperationArg(C2Operation operation, String argument) {
Review Comment:
We could return Optional<String> here. What do you think?
##########
c2/c2-client-bundle/c2-client-service/src/main/java/org/apache/nifi/c2/client/service/operation/DescribeManifestOperationHandler.java:
##########
@@ -58,25 +59,23 @@ public OperandType getOperandType() {
@Override
public C2OperationAck handle(C2Operation operation) {
- String opIdentifier = Optional.ofNullable(operation.getIdentifier())
- .orElse(EMPTY);
- C2OperationAck operationAck = new C2OperationAck();
- C2OperationState state = new C2OperationState();
- operationAck.setOperationState(state);
- operationAck.setOperationId(opIdentifier);
+ String operationId = ofNullable(operation.getIdentifier()).orElse(EMPTY);
Review Comment:
Shouldn't we add the ResourceInfo to the C2OperationAck? The Describe Manifest is basically a full Heartbeat so the fields should match with the Heartbeat in my opinion.
##########
minifi/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/main/java/org/apache/nifi/minifi/c2/command/syncresource/DefaultSyncResourceStrategy.java:
##########
@@ -0,0 +1,212 @@
+/*
+ * 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.minifi.c2.command.syncresource;
+
+import static java.nio.file.Files.copy;
+import static java.nio.file.Files.createTempFile;
+import static java.nio.file.Files.deleteIfExists;
+import static java.nio.file.StandardCopyOption.COPY_ATTRIBUTES;
+import static java.nio.file.StandardCopyOption.REPLACE_EXISTING;
+import static java.util.Map.entry;
+import static java.util.Optional.empty;
+import static java.util.Optional.ofNullable;
+import static java.util.UUID.randomUUID;
+import static java.util.function.Predicate.not;
+import static java.util.stream.Collectors.toSet;
+import static org.apache.commons.io.file.PathUtils.createParentDirectories;
+import static org.apache.nifi.c2.protocol.api.C2OperationState.OperationState.FULLY_APPLIED;
+import static org.apache.nifi.c2.protocol.api.C2OperationState.OperationState.NOT_APPLIED;
+import static org.apache.nifi.c2.protocol.api.C2OperationState.OperationState.NO_OPERATION;
+import static org.apache.nifi.c2.protocol.api.C2OperationState.OperationState.PARTIALLY_APPLIED;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.Map.Entry;
+import java.util.Optional;
+import java.util.Set;
+import java.util.function.BiFunction;
+import java.util.function.Function;
+import org.apache.nifi.c2.client.service.operation.SyncResourceStrategy;
+import org.apache.nifi.c2.protocol.api.C2OperationState.OperationState;
+import org.apache.nifi.c2.protocol.api.ResourceItem;
+import org.apache.nifi.c2.protocol.api.ResourcesGlobalHash;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class DefaultSyncResourceStrategy implements SyncResourceStrategy {
+
+ private static final Logger LOG = LoggerFactory.getLogger(DefaultSyncResourceStrategy.class);
+
+ private static final Set<Entry<OperationState, OperationState>> SUCCESS_RESULT_PAIRS = Set.of(
+ entry(NO_OPERATION, NO_OPERATION),
+ entry(NO_OPERATION, FULLY_APPLIED),
+ entry(FULLY_APPLIED, NO_OPERATION),
+ entry(FULLY_APPLIED, FULLY_APPLIED));
+
+ private static final Set<Entry<OperationState, OperationState>> FAILED_RESULT_PAIRS = Set.of(
+ entry(NO_OPERATION, NOT_APPLIED),
+ entry(NOT_APPLIED, NO_OPERATION),
+ entry(NOT_APPLIED, NOT_APPLIED));
+
+ private final ResourceRepository resourceRepository;
+ private final Path assetDirectory;
+ private final Path extensionDirectory;
+
+ public DefaultSyncResourceStrategy(ResourceRepository resourceRepository, Path assetDirectory, Path extensionDirectory) {
+ this.resourceRepository = resourceRepository;
+ this.assetDirectory = assetDirectory;
+ this.extensionDirectory = extensionDirectory;
+ }
+
+ @Override
+ public OperationState synchronizeResourceRepository(ResourcesGlobalHash c2GlobalHash, List<ResourceItem> c2ServerItems,
+ BiFunction<String, Function<InputStream, Optional<Path>>, Optional<Path>> resourceDownloadFunction,
+ Function<String, Optional<String>> urlEnrichFunction) {
+ Set<ResourceItem> c2Items = c2ServerItems.stream().collect(toSet());
+ Set<ResourceItem> agentItems = resourceRepository.findAllResourceItems().stream().collect(toSet());
+
+ OperationState deleteResult = deleteItems(c2Items, agentItems);
+ OperationState saveResult = saveNewItems(c2Items, agentItems, resourceDownloadFunction, urlEnrichFunction);
+
+ Entry<OperationState, OperationState> resultPair = entry(deleteResult, saveResult);
+
+ return SUCCESS_RESULT_PAIRS.contains(resultPair)
+ ? saveGlobalHash(c2GlobalHash, deleteResult, saveResult)
+ : FAILED_RESULT_PAIRS.contains(resultPair) ? NOT_APPLIED : PARTIALLY_APPLIED;
+ }
+
+ private OperationState deleteItems(Set<ResourceItem> c2Items, Set<ResourceItem> agentItems) {
+ List<ResourceItem> toDeleteItems = agentItems.stream().filter(not(c2Items::contains)).toList();
+ if (toDeleteItems.isEmpty()) {
+ return NO_OPERATION;
+ }
+
+ List<ResourceItem> deletedItems = toDeleteItems.stream()
+ .map(this::deleteItem)
+ .flatMap(Optional::stream)
+ .toList();
+
+ return deletedItems.isEmpty()
+ ? NOT_APPLIED
+ : deletedItems.size() == toDeleteItems.size() ? FULLY_APPLIED : PARTIALLY_APPLIED;
+ }
+
+ private Optional<ResourceItem> deleteItem(ResourceItem item) {
+ try {
+ Path resourcePath = resourcePath(item);
+ ResourceItem deletedItem = resourceRepository.deleteResourceItem(item);
+ deleteSilently(resourcePath, "Unable to delete resource file");
Review Comment:
Why don't we encapsulate all the delete logic in the repository's deleteResourceItem method - same for the add.
Maybe I misunderstood the intent of the FileResourceRepository. Is it only the FileMetadata Repository?
##########
c2/c2-client-bundle/c2-client-http/src/main/java/org/apache/nifi/c2/client/http/C2HttpClient.java:
##########
@@ -95,6 +102,26 @@ public Optional<byte[]> retrieveUpdateAssetContent(String callbackUrl) {
return retrieveContent(callbackUrl, Map.of());
}
+ @Override
+ public Optional<Path> retrieveResourceItem(String callbackUrl, Function<InputStream, Optional<Path>> resourceConsumer) {
+ Request request = new Request.Builder()
+ .get()
+ .url(callbackUrl)
+ .build();
+
+ try (Response response = httpClientReference.get().newCall(request).execute()) {
+ ResponseBody responseBody = response.body();
+ if (!response.isSuccessful() || responseBody == null) {
+ throw new C2ServerException(
+ format("Resource content retrieval failed with HTTP return code %d and body:\n%s", response.code(), responseBody == null ? EMPTY : responseBody.toString()));
Review Comment:
I would rather not pass the response body in the exception message. Is it always a string? I'm not sure how it behaves in case of network issues.
##########
c2/c2-client-bundle/c2-client-service/src/main/java/org/apache/nifi/c2/client/service/operation/SyncResourceOperationHandler.java:
##########
@@ -0,0 +1,132 @@
+/*
+ * 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.c2.client.service.operation;
+
+import static java.util.Optional.ofNullable;
+import static org.apache.commons.lang3.StringUtils.EMPTY;
+import static org.apache.nifi.c2.protocol.api.C2OperationState.OperationState.NOT_APPLIED;
+import static org.apache.nifi.c2.protocol.api.OperandType.RESOURCE;
+import static org.apache.nifi.c2.protocol.api.OperationType.SYNC;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import java.util.List;
+import java.util.Map;
+import org.apache.nifi.c2.client.api.C2Client;
+import org.apache.nifi.c2.protocol.api.C2Operation;
+import org.apache.nifi.c2.protocol.api.C2OperationAck;
+import org.apache.nifi.c2.protocol.api.C2OperationState;
+import org.apache.nifi.c2.protocol.api.C2OperationState.OperationState;
+import org.apache.nifi.c2.protocol.api.OperandType;
+import org.apache.nifi.c2.protocol.api.OperationType;
+import org.apache.nifi.c2.protocol.api.ResourceItem;
+import org.apache.nifi.c2.protocol.api.ResourcesGlobalHash;
+import org.apache.nifi.c2.serializer.C2Serializer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class SyncResourceOperationHandler implements C2OperationHandler {
+
+ static final String GLOBAL_HASH_FIELD = "globalHash";
+ static final String RESOURCE_LIST_FIELD = "resourceList";
+
+ private static final Logger LOG = LoggerFactory.getLogger(SyncResourceOperationHandler.class);
+
+ private final C2Client c2Client;
+ private final OperandPropertiesProvider operandPropertiesProvider;
+ private final SyncResourceStrategy syncResourceStrategy;
+ private final C2Serializer c2Serializer;
+
+ public SyncResourceOperationHandler(C2Client c2Client, OperandPropertiesProvider operandPropertiesProvider, SyncResourceStrategy syncResourceStrategy,
+ C2Serializer c2Serializer) {
+ this.c2Client = c2Client;
+ this.operandPropertiesProvider = operandPropertiesProvider;
+ this.syncResourceStrategy = syncResourceStrategy;
+ this.c2Serializer = c2Serializer;
+ }
+
+ public static SyncResourceOperationHandler create(C2Client c2Client, OperandPropertiesProvider operandPropertiesProvider, SyncResourceStrategy syncResourceStrategy,
+ C2Serializer c2Serializer) {
+ if (c2Client == null) {
Review Comment:
What do you think about creating a utility class with checkArgument in a minifi common module?
##########
c2/c2-client-bundle/c2-client-service/src/main/java/org/apache/nifi/c2/client/service/operation/SyncResourceOperationHandler.java:
##########
@@ -0,0 +1,132 @@
+/*
+ * 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.c2.client.service.operation;
+
+import static java.util.Optional.ofNullable;
+import static org.apache.commons.lang3.StringUtils.EMPTY;
+import static org.apache.nifi.c2.protocol.api.C2OperationState.OperationState.NOT_APPLIED;
+import static org.apache.nifi.c2.protocol.api.OperandType.RESOURCE;
+import static org.apache.nifi.c2.protocol.api.OperationType.SYNC;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import java.util.List;
+import java.util.Map;
+import org.apache.nifi.c2.client.api.C2Client;
+import org.apache.nifi.c2.protocol.api.C2Operation;
+import org.apache.nifi.c2.protocol.api.C2OperationAck;
+import org.apache.nifi.c2.protocol.api.C2OperationState;
+import org.apache.nifi.c2.protocol.api.C2OperationState.OperationState;
+import org.apache.nifi.c2.protocol.api.OperandType;
+import org.apache.nifi.c2.protocol.api.OperationType;
+import org.apache.nifi.c2.protocol.api.ResourceItem;
+import org.apache.nifi.c2.protocol.api.ResourcesGlobalHash;
+import org.apache.nifi.c2.serializer.C2Serializer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class SyncResourceOperationHandler implements C2OperationHandler {
+
+ static final String GLOBAL_HASH_FIELD = "globalHash";
+ static final String RESOURCE_LIST_FIELD = "resourceList";
+
+ private static final Logger LOG = LoggerFactory.getLogger(SyncResourceOperationHandler.class);
+
+ private final C2Client c2Client;
+ private final OperandPropertiesProvider operandPropertiesProvider;
+ private final SyncResourceStrategy syncResourceStrategy;
+ private final C2Serializer c2Serializer;
+
+ public SyncResourceOperationHandler(C2Client c2Client, OperandPropertiesProvider operandPropertiesProvider, SyncResourceStrategy syncResourceStrategy,
+ C2Serializer c2Serializer) {
+ this.c2Client = c2Client;
+ this.operandPropertiesProvider = operandPropertiesProvider;
+ this.syncResourceStrategy = syncResourceStrategy;
+ this.c2Serializer = c2Serializer;
+ }
+
+ public static SyncResourceOperationHandler create(C2Client c2Client, OperandPropertiesProvider operandPropertiesProvider, SyncResourceStrategy syncResourceStrategy,
+ C2Serializer c2Serializer) {
+ if (c2Client == null) {
+ throw new IllegalArgumentException("C2Client should not be null");
+ }
+ if (operandPropertiesProvider == null) {
+ throw new IllegalArgumentException("OperandPropertiesProvider should not be not null");
+ }
+ if (syncResourceStrategy == null) {
+ throw new IllegalArgumentException("Sync resource strategy should not be null");
+ }
+ if (c2Serializer == null) {
+ throw new IllegalArgumentException("C2 serializer should not be null");
+ }
+ return new SyncResourceOperationHandler(c2Client, operandPropertiesProvider, syncResourceStrategy, c2Serializer);
+ }
+
+ @Override
+ public OperationType getOperationType() {
+ return SYNC;
+ }
+
+ @Override
+ public OperandType getOperandType() {
+ return RESOURCE;
+ }
+
+ @Override
+ public Map<String, Object> getProperties() {
+ return operandPropertiesProvider.getProperties();
+ }
+
+ @Override
+ public C2OperationAck handle(C2Operation operation) {
+ String operationId = ofNullable(operation.getIdentifier()).orElse(EMPTY);
+
+ ResourcesGlobalHash resourcesGlobalHash = getOperationArg(operation, GLOBAL_HASH_FIELD, new TypeReference<>() {
+ });
+ if (resourcesGlobalHash == null) {
+ LOG.error("Resources global hash could not be constructed from C2 request");
+ return operationAck(operationId, operationState(NOT_APPLIED, "Resources global hash element was not found"));
+ }
+ List<ResourceItem> resourceItems = getOperationArg(operation, RESOURCE_LIST_FIELD, new TypeReference<>() {
+ });
+ if (resourceItems == null) {
+ LOG.error("Resource item list could not be constructed from C2 request");
+ return operationAck(operationId, operationState(NOT_APPLIED, "Resource item list element was not found"));
+ }
+
+ OperationState operationState = syncResourceStrategy.synchronizeResourceRepository(resourcesGlobalHash, resourceItems, c2Client::retrieveResourceItem,
+ relativeUrl -> c2Client.getCallbackUrl(null, relativeUrl));
+ C2OperationState resultState = operationState(
+ operationState,
+ switch (operationState) {
+ case NOT_APPLIED -> "No resource items were retrieved, please check the log for errors";
+ case PARTIALLY_APPLIED -> "Resource repository is partially synced, retrieving some items failed. Pleas check log for errors";
+ case FULLY_APPLIED -> "Agent Resource repository is in sync with the C2 server";
+ default -> "Unexpected status, please check the log for errors";
+ }
+ );
+
+ return operationAck(operationId, resultState);
+ }
+
+ private <T> T getOperationArg(C2Operation operation, String key, TypeReference<T> type) {
Review Comment:
Any way to move this to the C2OperationHandler or to an abstract class? Also changing the return type to Optional would make the caller side cleaner.
##########
minifi/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/main/java/org/apache/nifi/minifi/c2/C2NifiClientService.java:
##########
@@ -222,14 +231,18 @@ private C2OperationHandlerProvider c2OperationHandlerProvider(NiFiProperties niF
new StandardFlowEnrichService(niFiProperties), flowPropertyEncryptor,
StandardFlowSerDeService.defaultInstance(), niFiProperties.getProperty(FLOW_CONFIGURATION_FILE));
+ SyncResourceStrategy syncResourceStrategy =
+ new DefaultSyncResourceStrategy(resourceRepository, Path.of(c2AssetDirectory), niFiProperties.getNarAutoLoadDirectory().toPath());
Review Comment:
Is it a good idea to use the same asset dir for the asset sync and the asset push? Files could be overridden accidentally via the single asset push.
##########
minifi/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/main/java/org/apache/nifi/minifi/c2/command/syncresource/DefaultSyncResourceStrategy.java:
##########
@@ -0,0 +1,212 @@
+/*
+ * 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.minifi.c2.command.syncresource;
+
+import static java.nio.file.Files.copy;
+import static java.nio.file.Files.createTempFile;
+import static java.nio.file.Files.deleteIfExists;
+import static java.nio.file.StandardCopyOption.COPY_ATTRIBUTES;
+import static java.nio.file.StandardCopyOption.REPLACE_EXISTING;
+import static java.util.Map.entry;
+import static java.util.Optional.empty;
+import static java.util.Optional.ofNullable;
+import static java.util.UUID.randomUUID;
+import static java.util.function.Predicate.not;
+import static java.util.stream.Collectors.toSet;
+import static org.apache.commons.io.file.PathUtils.createParentDirectories;
+import static org.apache.nifi.c2.protocol.api.C2OperationState.OperationState.FULLY_APPLIED;
+import static org.apache.nifi.c2.protocol.api.C2OperationState.OperationState.NOT_APPLIED;
+import static org.apache.nifi.c2.protocol.api.C2OperationState.OperationState.NO_OPERATION;
+import static org.apache.nifi.c2.protocol.api.C2OperationState.OperationState.PARTIALLY_APPLIED;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.Map.Entry;
+import java.util.Optional;
+import java.util.Set;
+import java.util.function.BiFunction;
+import java.util.function.Function;
+import org.apache.nifi.c2.client.service.operation.SyncResourceStrategy;
+import org.apache.nifi.c2.protocol.api.C2OperationState.OperationState;
+import org.apache.nifi.c2.protocol.api.ResourceItem;
+import org.apache.nifi.c2.protocol.api.ResourcesGlobalHash;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class DefaultSyncResourceStrategy implements SyncResourceStrategy {
+
+ private static final Logger LOG = LoggerFactory.getLogger(DefaultSyncResourceStrategy.class);
+
+ private static final Set<Entry<OperationState, OperationState>> SUCCESS_RESULT_PAIRS = Set.of(
+ entry(NO_OPERATION, NO_OPERATION),
+ entry(NO_OPERATION, FULLY_APPLIED),
+ entry(FULLY_APPLIED, NO_OPERATION),
+ entry(FULLY_APPLIED, FULLY_APPLIED));
+
+ private static final Set<Entry<OperationState, OperationState>> FAILED_RESULT_PAIRS = Set.of(
+ entry(NO_OPERATION, NOT_APPLIED),
+ entry(NOT_APPLIED, NO_OPERATION),
+ entry(NOT_APPLIED, NOT_APPLIED));
+
+ private final ResourceRepository resourceRepository;
+ private final Path assetDirectory;
+ private final Path extensionDirectory;
+
+ public DefaultSyncResourceStrategy(ResourceRepository resourceRepository, Path assetDirectory, Path extensionDirectory) {
+ this.resourceRepository = resourceRepository;
+ this.assetDirectory = assetDirectory;
+ this.extensionDirectory = extensionDirectory;
+ }
+
+ @Override
+ public OperationState synchronizeResourceRepository(ResourcesGlobalHash c2GlobalHash, List<ResourceItem> c2ServerItems,
+ BiFunction<String, Function<InputStream, Optional<Path>>, Optional<Path>> resourceDownloadFunction,
+ Function<String, Optional<String>> urlEnrichFunction) {
+ Set<ResourceItem> c2Items = c2ServerItems.stream().collect(toSet());
Review Comment:
There 2 collects methods could be replaced with new HashSet<>()
##########
minifi/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/main/java/org/apache/nifi/minifi/c2/command/syncresource/FileResourceRepository.java:
##########
@@ -0,0 +1,123 @@
+/*
+ * 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.minifi.c2.command.syncresource;
+
+import static java.nio.file.Files.exists;
+import static java.nio.file.Files.readString;
+import static java.nio.file.Files.writeString;
+import static java.nio.file.StandardOpenOption.CREATE;
+import static java.nio.file.StandardOpenOption.SYNC;
+import static java.nio.file.StandardOpenOption.TRUNCATE_EXISTING;
+import static java.nio.file.StandardOpenOption.WRITE;
+import static java.util.stream.Collectors.toList;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.nifi.c2.protocol.api.ResourceItem;
+import org.apache.nifi.c2.protocol.api.ResourcesGlobalHash;
+import org.apache.nifi.c2.serializer.C2Serializer;
+
+public class FileResourceRepository implements ResourceRepository {
+
+ static final String RESOURCE_REPOSITORY_FILE_NAME = "resources.json";
+
+ private static final ResourceRepositoryDescriptor EMTPY_HOLDER = new ResourceRepositoryDescriptor(new ResourcesGlobalHash(), List.of());
+
+ private final Path resourceRepositoryFile;
+ private final C2Serializer c2Serializer;
+
+ private ResourceRepositoryDescriptor resourceRepositoryDescriptor;
+
+ public FileResourceRepository(Path configDirectory, C2Serializer c2Serializer) {
Review Comment:
I guess, with this solution there is no way to recover if the resourceRepositoryDescriptor get corrupted in the file system, or out of sync with the actual files?
Let say, we want to deploy an agent with extensions that are already mounted (to save bandwidth). If I'm not wrong with the current approach, we can not avoid to re-download everything. Can we consider somehow avoiding this?
##########
minifi/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/main/java/org/apache/nifi/minifi/c2/command/syncresource/DefaultSyncResourceStrategy.java:
##########
@@ -0,0 +1,212 @@
+/*
+ * 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.minifi.c2.command.syncresource;
+
+import static java.nio.file.Files.copy;
+import static java.nio.file.Files.createTempFile;
+import static java.nio.file.Files.deleteIfExists;
+import static java.nio.file.StandardCopyOption.COPY_ATTRIBUTES;
+import static java.nio.file.StandardCopyOption.REPLACE_EXISTING;
+import static java.util.Map.entry;
+import static java.util.Optional.empty;
+import static java.util.Optional.ofNullable;
+import static java.util.UUID.randomUUID;
+import static java.util.function.Predicate.not;
+import static java.util.stream.Collectors.toSet;
+import static org.apache.commons.io.file.PathUtils.createParentDirectories;
+import static org.apache.nifi.c2.protocol.api.C2OperationState.OperationState.FULLY_APPLIED;
+import static org.apache.nifi.c2.protocol.api.C2OperationState.OperationState.NOT_APPLIED;
+import static org.apache.nifi.c2.protocol.api.C2OperationState.OperationState.NO_OPERATION;
+import static org.apache.nifi.c2.protocol.api.C2OperationState.OperationState.PARTIALLY_APPLIED;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.Map.Entry;
+import java.util.Optional;
+import java.util.Set;
+import java.util.function.BiFunction;
+import java.util.function.Function;
+import org.apache.nifi.c2.client.service.operation.SyncResourceStrategy;
+import org.apache.nifi.c2.protocol.api.C2OperationState.OperationState;
+import org.apache.nifi.c2.protocol.api.ResourceItem;
+import org.apache.nifi.c2.protocol.api.ResourcesGlobalHash;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class DefaultSyncResourceStrategy implements SyncResourceStrategy {
+
+ private static final Logger LOG = LoggerFactory.getLogger(DefaultSyncResourceStrategy.class);
+
+ private static final Set<Entry<OperationState, OperationState>> SUCCESS_RESULT_PAIRS = Set.of(
+ entry(NO_OPERATION, NO_OPERATION),
+ entry(NO_OPERATION, FULLY_APPLIED),
+ entry(FULLY_APPLIED, NO_OPERATION),
+ entry(FULLY_APPLIED, FULLY_APPLIED));
+
+ private static final Set<Entry<OperationState, OperationState>> FAILED_RESULT_PAIRS = Set.of(
+ entry(NO_OPERATION, NOT_APPLIED),
+ entry(NOT_APPLIED, NO_OPERATION),
+ entry(NOT_APPLIED, NOT_APPLIED));
+
+ private final ResourceRepository resourceRepository;
+ private final Path assetDirectory;
+ private final Path extensionDirectory;
+
+ public DefaultSyncResourceStrategy(ResourceRepository resourceRepository, Path assetDirectory, Path extensionDirectory) {
+ this.resourceRepository = resourceRepository;
+ this.assetDirectory = assetDirectory;
+ this.extensionDirectory = extensionDirectory;
+ }
+
+ @Override
+ public OperationState synchronizeResourceRepository(ResourcesGlobalHash c2GlobalHash, List<ResourceItem> c2ServerItems,
+ BiFunction<String, Function<InputStream, Optional<Path>>, Optional<Path>> resourceDownloadFunction,
+ Function<String, Optional<String>> urlEnrichFunction) {
+ Set<ResourceItem> c2Items = c2ServerItems.stream().collect(toSet());
+ Set<ResourceItem> agentItems = resourceRepository.findAllResourceItems().stream().collect(toSet());
+
+ OperationState deleteResult = deleteItems(c2Items, agentItems);
+ OperationState saveResult = saveNewItems(c2Items, agentItems, resourceDownloadFunction, urlEnrichFunction);
+
+ Entry<OperationState, OperationState> resultPair = entry(deleteResult, saveResult);
+
+ return SUCCESS_RESULT_PAIRS.contains(resultPair)
+ ? saveGlobalHash(c2GlobalHash, deleteResult, saveResult)
+ : FAILED_RESULT_PAIRS.contains(resultPair) ? NOT_APPLIED : PARTIALLY_APPLIED;
+ }
+
+ private OperationState deleteItems(Set<ResourceItem> c2Items, Set<ResourceItem> agentItems) {
+ List<ResourceItem> toDeleteItems = agentItems.stream().filter(not(c2Items::contains)).toList();
+ if (toDeleteItems.isEmpty()) {
+ return NO_OPERATION;
+ }
+
+ List<ResourceItem> deletedItems = toDeleteItems.stream()
+ .map(this::deleteItem)
+ .flatMap(Optional::stream)
+ .toList();
+
+ return deletedItems.isEmpty()
+ ? NOT_APPLIED
+ : deletedItems.size() == toDeleteItems.size() ? FULLY_APPLIED : PARTIALLY_APPLIED;
+ }
+
+ private Optional<ResourceItem> deleteItem(ResourceItem item) {
+ try {
+ Path resourcePath = resourcePath(item);
+ ResourceItem deletedItem = resourceRepository.deleteResourceItem(item);
+ deleteSilently(resourcePath, "Unable to delete resource file");
+ return Optional.of(deletedItem);
+ } catch (Exception e) {
+ LOG.error("Unable to delete resources from repository", e);
+ return empty();
+ }
+ }
+
+ private OperationState saveNewItems(Set<ResourceItem> c2Items, Set<ResourceItem> agentItems,
+ BiFunction<String, Function<InputStream, Optional<Path>>, Optional<Path>> resourceDownloadFunction,
+ Function<String, Optional<String>> urlEnrichFunction) {
+ List<ResourceItem> newItems = c2Items.stream().filter(not(agentItems::contains)).toList();
+ if (newItems.isEmpty()) {
+ return NO_OPERATION;
+ }
+
+ List<ResourceItem> addedItems = newItems.stream()
+ .map(newItem -> urlEnrichFunction.apply(newItem.getUrl())
+ .flatMap(enrichedUrl -> resourceDownloadFunction.apply(enrichedUrl, this::persistToTemporaryLocation))
+ .flatMap(tempResourcePath -> moveToFinalLocation(newItem, tempResourcePath))
+ .map(finalPath -> addItemToRepositoryOrCleanupWhenFailure(newItem, finalPath))
+ )
+ .flatMap(Optional::stream)
+ .toList();
+
+ return addedItems.isEmpty()
+ ? NOT_APPLIED
+ : newItems.size() == addedItems.size() ? FULLY_APPLIED : PARTIALLY_APPLIED;
+ }
+
+ private Optional<Path> persistToTemporaryLocation(InputStream inputStream) {
+ String tempResourceId = randomUUID().toString();
+ try {
+ Path tempFile = createTempFile(tempResourceId, null);
+ copy(inputStream, tempFile, REPLACE_EXISTING);
+ return ofNullable(tempFile);
+ } catch (IOException e) {
+ LOG.error("Unable to download resource. Will retry in next heartbeat iteration", e);
+ return empty();
+ }
+ }
+
+ private Optional<Path> moveToFinalLocation(ResourceItem resourceItem, Path tempPath) {
+ try {
+ Path finalPath = resourcePath(resourceItem);
+
+ createParentDirectories(finalPath);
+ copy(tempPath, finalPath, REPLACE_EXISTING, COPY_ATTRIBUTES);
+
+ return ofNullable(finalPath);
+ } catch (IOException e) {
+ LOG.error("Unable to move asset to final location. Syncing this asset will be retried in next heartbeat iteration", e);
+ return empty();
+ } finally {
+ deleteSilently(tempPath, "Unable to cleanup temporary file");
+ }
+ }
+
+ private ResourceItem addItemToRepositoryOrCleanupWhenFailure(ResourceItem newItem, Path finalPath) {
+ try {
+ return resourceRepository.addResourceItem(newItem);
+ } catch (Exception e) {
+ LOG.error("Unable to add resource to repository", e);
+ deleteSilently(finalPath, "Unable to cleanup resource file");
+ return null;
+ }
+ }
+
+ private void deleteSilently(Path tempPath, String errorMessage) {
+ try {
+ deleteIfExists(tempPath);
+ } catch (IOException e) {
+ LOG.error(errorMessage, e);
+ }
+ }
+
+ private Path resourcePath(ResourceItem resourceItem) {
+ return switch (resourceItem.getResourceType()) {
+ case ASSET -> ofNullable(resourceItem.getResourcePath())
Review Comment:
we should throw error if resourceItem.getResourcePath() contains ".."
--
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: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org