exceptionfactory commented on code in PR #9138:
URL: https://github.com/apache/nifi/pull/9138#discussion_r1704502027
##########
nifi-commons/nifi-utils/src/main/java/org/apache/nifi/util/file/FileUtils.java:
##########
@@ -582,4 +582,39 @@ public static long getContainerUsableSpace(final Path
path) {
return path.toFile().getUsableSpace();
}
+ // The invalid character list is derived from this Stackoverflow page.
+ //
https://stackoverflow.com/questions/1155107/is-there-a-cross-platform-java-method-to-remove-filename-special-chars
+ private final static int[] INVALID_CHARS = {34, 60, 62, 124, 0, 1, 2, 3,
4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15,
+ 16, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27, 28, 29, 30, 31,
58, 42, 63, 92, 47, 32};
+
+ static {
+ Arrays.sort(INVALID_CHARS);
+ }
+
+ /**
+ * Replaces invalid characters for a file system name within a given
filename string to underscore '_'.
+ * Be careful not to pass a file path as this method replaces path
delimiter characters (i.e forward/back slashes).
+ * @param filename The filename to clean
+ * @return sanitized filename
+ */
+ public static String sanitizeFilename(String filename) {
Review Comment:
Minor naming recommendation:
```suggestion
public static String getSanitizedFilename(String filename) {
```
##########
nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/framework/configuration/FlowControllerConfiguration.java:
##########
@@ -434,4 +439,24 @@ public NarManager narManager(@Autowired final
NarPersistenceProvider narPersiste
properties
);
}
+
+ /**
+ * Asset Manager from Flow Controller
+ *
+ * @return Asset Manager
+ */
+ @Bean
+ public AssetManager assetManager(@Autowired final FlowController
flowController) {
Review Comment:
Instead of wiring the `FlowController` as an argument, it should be possible
to call `flowController()`.
##########
nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/dto/DtoFactory.java:
##########
@@ -1509,6 +1512,26 @@ private ParameterProviderConfigurationEntity
createParameterProviderConfiguratio
return config;
}
+ public AssetEntity createAssetEntity(final Asset asset) {
+ final AssetEntity entity = new AssetEntity();
+ entity.setAsset(createAssetDto(asset));
+ return entity;
+ }
+
+ public AssetDTO createAssetDto(final Asset asset) {
+ final File assetFile = asset.getFile();
+ final AssetDTO dto = new AssetDTO();
+ dto.setId(asset.getIdentifier());
+ dto.setName(asset.getName());
+ dto.setDigest(asset.getDigest().orElse(null));
+ dto.setMissingContent(!assetFile.exists());
Review Comment:
What do you think about naming this property `contentNotFound` instead of
`missingContent`?
##########
nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ParameterContextResource.java:
##########
@@ -323,6 +356,204 @@ public Response updateParameterContext(
);
}
+ @POST
+ @Consumes(MediaType.APPLICATION_OCTET_STREAM)
+ @Produces(MediaType.APPLICATION_JSON)
+ @Path("{contextId}/assets")
+ @Operation(
+ summary = "Creates a new Asset in the given Parameter Context",
+ responses = @ApiResponse(content = @Content(schema =
@Schema(implementation = AssetEntity.class))),
+ description = "This endpoint will create a new Asset in the given
Parameter Context. The Asset will be created with the given name and the
contents of the file that is uploaded. " +
+ "The Asset will be created in the given Parameter Context,
and will be available for use by any component that references the Parameter
Context.",
+ security = {
+ @SecurityRequirement(name = "Read -
/parameter-contexts/{parameterContextId}"),
+ @SecurityRequirement(name = "Write -
/parameter-contexts/{parameterContextId}"),
+ @SecurityRequirement(name = "Read - for every component that
is affected by the update"),
+ @SecurityRequirement(name = "Write - for every component that
is affected by the update"),
+ @SecurityRequirement(name = "Read - for every currently
inherited parameter context")
+ }
+ )
+ @ApiResponses(
+ value = {
+ @ApiResponse(responseCode = "400", description = "NiFi was unable
to complete the request because it was invalid. The request should not be
retried without modification."),
+ @ApiResponse(responseCode = "401", description = "Client could not
be authenticated."),
+ @ApiResponse(responseCode = "403", description = "Client is not
authorized to make this request."),
+ @ApiResponse(responseCode = "404", description = "The specified
resource could not be found."),
+ @ApiResponse(responseCode = "409", description = "The request was
valid but NiFi was not in the appropriate state to process it.")
+ }
+ )
+ public Response createAsset(
+ @PathParam("contextId") final String contextId,
+ @HeaderParam(FILENAME_HEADER) final String assetName,
+ @Parameter(description = "The contents of the asset.", required =
true) final InputStream assetContents) throws IOException {
+
+ // Validate input
+ if (StringUtils.isBlank(assetName)) {
+ throw new IllegalArgumentException(FILENAME_HEADER + " header is
required");
+ }
+ if (assetContents == null) {
+ throw new IllegalArgumentException("Asset contents must be
specified.");
+ }
+
+ // If clustered and not all nodes are connected, do not allow creating
an asset.
+ // Generally, we allow the flow to be modified when nodes are
disconnected, but we do not allow creating an asset because
+ // the cluster has no mechanism for synchronizing those assets after
the upload.
+ final ClusterCoordinator clusterCoordinator = getClusterCoordinator();
+ if (clusterCoordinator != null) {
+ final Set<NodeIdentifier> disconnectedNodes =
clusterCoordinator.getNodeIdentifiers(NodeConnectionState.CONNECTING,
NodeConnectionState.DISCONNECTED, NodeConnectionState.DISCONNECTING);
+ if (!disconnectedNodes.isEmpty()) {
+ throw new IllegalStateException("Cannot create an Asset
because the following %s nodes are not currently connected:
%s".formatted(disconnectedNodes.size(), disconnectedNodes));
+ }
+ }
+
+ // Get the context or throw ResourceNotFoundException
+ final NiFiUser user = NiFiUserUtils.getNiFiUser();
+ final ParameterContextEntity contextEntity =
serviceFacade.getParameterContext(contextId, false, user);
+ final Set<AffectedComponentEntity> affectedComponents =
serviceFacade.getComponentsAffectedByParameterContextUpdate(Collections.singletonList(contextEntity.getComponent()));
+
+ // Authorize the request
+ serviceFacade.authorizeAccess(lookup -> {
+ // Verify READ and WRITE permissions for user, for the Parameter
Context itself
+ final ParameterContext parameterContext =
lookup.getParameterContext(contextId);
+ parameterContext.authorize(authorizer, RequestAction.READ, user);
+ parameterContext.authorize(authorizer, RequestAction.WRITE, user);
+
+ // Verify READ and WRITE permissions for user, for every component
that is affected
+ affectedComponents.forEach(component ->
parameterUpdateManager.authorizeAffectedComponent(component, lookup, user,
true, true));
+ });
+
+ // If we need to replicate the request, we do so using the Upload
Request Replicator, rather than the typical replicate() method.
+ // This is because Upload Request Replication works differently in
that it needs to be able to replicate the InputStream multiple times,
+ // so it must create a file on disk to do so and then use the file's
content to replicate the request. It also bypasses the two-phase
+ // commit process that is used for other requests because doing so
would result in uploading the file twice to each node or providing a
+ // different request for each of the two phases.
+
+ final long startTime = System.currentTimeMillis();
+ final InputStream maxLengthInputStream = new
MaxLengthInputStream(assetContents, (long) DataUnit.GB.toB(1));
+
+ final AssetEntity assetEntity;
+ if (isReplicateRequest()) {
+ final UploadRequest<AssetEntity> uploadRequest = new
UploadRequest.Builder<AssetEntity>()
+ .user(NiFiUserUtils.getNiFiUser())
+ .filename(assetName)
+ .identifier(UUID.randomUUID().toString())
+ .contents(maxLengthInputStream)
+ .header(FILENAME_HEADER, assetName)
+ .header(CONTENT_TYPE_HEADER, UPLOAD_CONTENT_TYPE)
+ .exampleRequestUri(getAbsolutePath())
+ .responseClass(AssetEntity.class)
+ .successfulResponseStatus(HttpResponseStatus.OK.getCode())
+ .build();
+ assetEntity = uploadRequestReplicator.upload(uploadRequest);
+ } else {
+ final String existingContextId = contextEntity.getId();
+ final String sanitizedAssetName =
FileUtils.sanitizeFilename(assetName);
Review Comment:
Instead of simply sanitizing the filename, should an input filename be
rejected if it contains unexpected characters?
##########
nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ParameterContextResource.java:
##########
@@ -323,6 +356,204 @@ public Response updateParameterContext(
);
}
+ @POST
+ @Consumes(MediaType.APPLICATION_OCTET_STREAM)
+ @Produces(MediaType.APPLICATION_JSON)
+ @Path("{contextId}/assets")
+ @Operation(
+ summary = "Creates a new Asset in the given Parameter Context",
+ responses = @ApiResponse(content = @Content(schema =
@Schema(implementation = AssetEntity.class))),
+ description = "This endpoint will create a new Asset in the given
Parameter Context. The Asset will be created with the given name and the
contents of the file that is uploaded. " +
+ "The Asset will be created in the given Parameter Context,
and will be available for use by any component that references the Parameter
Context.",
+ security = {
+ @SecurityRequirement(name = "Read -
/parameter-contexts/{parameterContextId}"),
+ @SecurityRequirement(name = "Write -
/parameter-contexts/{parameterContextId}"),
+ @SecurityRequirement(name = "Read - for every component that
is affected by the update"),
+ @SecurityRequirement(name = "Write - for every component that
is affected by the update"),
+ @SecurityRequirement(name = "Read - for every currently
inherited parameter context")
+ }
+ )
+ @ApiResponses(
+ value = {
+ @ApiResponse(responseCode = "400", description = "NiFi was unable
to complete the request because it was invalid. The request should not be
retried without modification."),
+ @ApiResponse(responseCode = "401", description = "Client could not
be authenticated."),
+ @ApiResponse(responseCode = "403", description = "Client is not
authorized to make this request."),
+ @ApiResponse(responseCode = "404", description = "The specified
resource could not be found."),
+ @ApiResponse(responseCode = "409", description = "The request was
valid but NiFi was not in the appropriate state to process it.")
+ }
+ )
+ public Response createAsset(
+ @PathParam("contextId") final String contextId,
+ @HeaderParam(FILENAME_HEADER) final String assetName,
+ @Parameter(description = "The contents of the asset.", required =
true) final InputStream assetContents) throws IOException {
+
+ // Validate input
+ if (StringUtils.isBlank(assetName)) {
+ throw new IllegalArgumentException(FILENAME_HEADER + " header is
required");
+ }
+ if (assetContents == null) {
+ throw new IllegalArgumentException("Asset contents must be
specified.");
+ }
+
+ // If clustered and not all nodes are connected, do not allow creating
an asset.
+ // Generally, we allow the flow to be modified when nodes are
disconnected, but we do not allow creating an asset because
+ // the cluster has no mechanism for synchronizing those assets after
the upload.
+ final ClusterCoordinator clusterCoordinator = getClusterCoordinator();
+ if (clusterCoordinator != null) {
+ final Set<NodeIdentifier> disconnectedNodes =
clusterCoordinator.getNodeIdentifiers(NodeConnectionState.CONNECTING,
NodeConnectionState.DISCONNECTED, NodeConnectionState.DISCONNECTING);
+ if (!disconnectedNodes.isEmpty()) {
+ throw new IllegalStateException("Cannot create an Asset
because the following %s nodes are not currently connected:
%s".formatted(disconnectedNodes.size(), disconnectedNodes));
+ }
+ }
+
+ // Get the context or throw ResourceNotFoundException
+ final NiFiUser user = NiFiUserUtils.getNiFiUser();
+ final ParameterContextEntity contextEntity =
serviceFacade.getParameterContext(contextId, false, user);
+ final Set<AffectedComponentEntity> affectedComponents =
serviceFacade.getComponentsAffectedByParameterContextUpdate(Collections.singletonList(contextEntity.getComponent()));
+
+ // Authorize the request
+ serviceFacade.authorizeAccess(lookup -> {
+ // Verify READ and WRITE permissions for user, for the Parameter
Context itself
+ final ParameterContext parameterContext =
lookup.getParameterContext(contextId);
+ parameterContext.authorize(authorizer, RequestAction.READ, user);
+ parameterContext.authorize(authorizer, RequestAction.WRITE, user);
+
+ // Verify READ and WRITE permissions for user, for every component
that is affected
+ affectedComponents.forEach(component ->
parameterUpdateManager.authorizeAffectedComponent(component, lookup, user,
true, true));
+ });
+
+ // If we need to replicate the request, we do so using the Upload
Request Replicator, rather than the typical replicate() method.
+ // This is because Upload Request Replication works differently in
that it needs to be able to replicate the InputStream multiple times,
+ // so it must create a file on disk to do so and then use the file's
content to replicate the request. It also bypasses the two-phase
+ // commit process that is used for other requests because doing so
would result in uploading the file twice to each node or providing a
+ // different request for each of the two phases.
+
+ final long startTime = System.currentTimeMillis();
+ final InputStream maxLengthInputStream = new
MaxLengthInputStream(assetContents, (long) DataUnit.GB.toB(1));
Review Comment:
For tracking purposes, it would be helpful to promote the maximum length
value to a static value. Although this could be made configurable, it seems
unlikely to change, and 1 GB seems like it should be large enough to handle
anything of reasonable size.
##########
nifi-toolkit/nifi-toolkit-cli/src/main/java/org/apache/nifi/toolkit/cli/impl/client/nifi/ParamContextClient.java:
##########
@@ -40,4 +43,10 @@ public interface ParamContextClient {
ParameterContextUpdateRequestEntity deleteParamContextUpdateRequest(String
contextId, String updateRequestId) throws NiFiClientException, IOException;
+ AssetEntity createAsset(String contextId, String assetName, File file)
throws NiFiClientException, IOException;
+
+ AssetsEntity getAssets(String contextId) throws NiFiClientException,
IOException;
+
+ File getAssetContent(String contextId, String assetId, File
outputDirectory) throws NiFiClientException, IOException;
Review Comment:
What do you think about returning a `Path` instead of `File` for this method?
--
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]
