markap14 commented on a change in pull request #3536: NIFI-6380: Introduced the notion of Parameters and Parameter Contexts… URL: https://github.com/apache/nifi/pull/3536#discussion_r294500013
########## File path: nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ParameterContextResource.java ########## @@ -0,0 +1,1147 @@ +/* + * 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.web.api; + +import io.swagger.annotations.Api; +import io.swagger.annotations.ApiOperation; +import io.swagger.annotations.ApiParam; +import io.swagger.annotations.ApiResponse; +import io.swagger.annotations.ApiResponses; +import io.swagger.annotations.Authorization; +import org.apache.commons.lang3.StringUtils; +import org.apache.nifi.authorization.Authorizer; +import org.apache.nifi.authorization.RequestAction; +import org.apache.nifi.authorization.resource.Authorizable; +import org.apache.nifi.authorization.user.NiFiUser; +import org.apache.nifi.authorization.user.NiFiUserUtils; +import org.apache.nifi.cluster.manager.NodeResponse; +import org.apache.nifi.controller.ScheduledState; +import org.apache.nifi.controller.service.ControllerServiceState; +import org.apache.nifi.web.NiFiServiceFacade; +import org.apache.nifi.web.ResourceNotFoundException; +import org.apache.nifi.web.ResumeFlowException; +import org.apache.nifi.web.Revision; +import org.apache.nifi.web.api.concurrent.AsyncRequestManager; +import org.apache.nifi.web.api.concurrent.AsynchronousWebRequest; +import org.apache.nifi.web.api.concurrent.RequestManager; +import org.apache.nifi.web.api.concurrent.StandardAsynchronousWebRequest; +import org.apache.nifi.web.api.concurrent.StandardUpdateStep; +import org.apache.nifi.web.api.concurrent.UpdateStep; +import org.apache.nifi.web.api.dto.AffectedComponentDTO; +import org.apache.nifi.web.api.dto.DtoFactory; +import org.apache.nifi.web.api.dto.ParameterContextDTO; +import org.apache.nifi.web.api.dto.ParameterContextUpdateRequestDTO; +import org.apache.nifi.web.api.dto.ParameterContextUpdateStepDTO; +import org.apache.nifi.web.api.dto.ParameterContextValidationRequestDTO; +import org.apache.nifi.web.api.dto.RevisionDTO; +import org.apache.nifi.web.api.entity.AffectedComponentEntity; +import org.apache.nifi.web.api.entity.ComponentValidationResultEntity; +import org.apache.nifi.web.api.entity.ComponentValidationResultsEntity; +import org.apache.nifi.web.api.entity.Entity; +import org.apache.nifi.web.api.entity.ParameterContextEntity; +import org.apache.nifi.web.api.entity.ParameterContextUpdateRequestEntity; +import org.apache.nifi.web.api.entity.ParameterContextValidationRequestEntity; +import org.apache.nifi.web.api.entity.ParameterContextsEntity; +import org.apache.nifi.web.api.request.ClientIdParameter; +import org.apache.nifi.web.api.request.LongParameter; +import org.apache.nifi.web.util.AffectedComponentUtils; +import org.apache.nifi.web.util.CancellableTimedPause; +import org.apache.nifi.web.util.ComponentLifecycle; +import org.apache.nifi.web.util.InvalidComponentAction; +import org.apache.nifi.web.util.LifecycleManagementException; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.ws.rs.Consumes; +import javax.ws.rs.DELETE; +import javax.ws.rs.DefaultValue; +import javax.ws.rs.GET; +import javax.ws.rs.HttpMethod; +import javax.ws.rs.POST; +import javax.ws.rs.PUT; +import javax.ws.rs.Path; +import javax.ws.rs.PathParam; +import javax.ws.rs.Produces; +import javax.ws.rs.QueryParam; +import javax.ws.rs.core.MediaType; +import javax.ws.rs.core.Response; +import java.net.URI; +import java.net.URISyntaxException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.UUID; +import java.util.concurrent.TimeUnit; +import java.util.function.Consumer; +import java.util.stream.Collectors; + + +@Path("/parameter-contexts") +@Api(value = "/parameter-contexts", description = "Endpoint for managing version control for a flow") +public class ParameterContextResource extends ApplicationResource { + private static final Logger logger = LoggerFactory.getLogger(ParameterContextResource.class); + + private NiFiServiceFacade serviceFacade; + private Authorizer authorizer; + private DtoFactory dtoFactory; + private ComponentLifecycle clusterComponentLifecycle; + private ComponentLifecycle localComponentLifecycle; + + private RequestManager<ParameterContextEntity> updateRequestManager = new AsyncRequestManager<>(100, TimeUnit.MINUTES.toMillis(1L), "Parameter Context Update Thread"); + private RequestManager<ComponentValidationResultsEntity> validationRequestManager = new AsyncRequestManager<>(100, TimeUnit.MINUTES.toMillis(1L), + "Parameter Context Validation Thread"); + + + + @GET + @Consumes(MediaType.WILDCARD) + @Produces(MediaType.APPLICATION_JSON) + @ApiOperation( + value = "Gets all Parameter Contexts", + response = ParameterContextsEntity.class, + authorizations = { + @Authorization(value = "Read - /parameter-contexts/{id} for each Parameter Context") + } + ) + @ApiResponses( + value = { + @ApiResponse(code = 400, message = "NiFi was unable to complete the request because it was invalid. The request should not be retried without modification."), + @ApiResponse(code = 401, message = "Client could not be authenticated."), + @ApiResponse(code = 403, message = "Client is not authorized to make this request."), + @ApiResponse(code = 409, message = "The request was valid but NiFi was not in the appropriate state to process it. Retrying the same request later may be successful.") + } + ) + public Response getParameterContexts() { + if (isReplicateRequest()) { + return replicate(HttpMethod.GET); + } + + final Set<ParameterContextEntity> parameterContexts = serviceFacade.getParameterContexts(); + final ParameterContextsEntity entity = new ParameterContextsEntity(); + entity.setParameterContexts(parameterContexts); + + // generate the response + return generateOkResponse(entity).build(); + } + + + @GET + @Consumes(MediaType.WILDCARD) + @Produces(MediaType.APPLICATION_JSON) + @Path("{id}") + @ApiOperation( + value = "Returns the Parameter Context with the given ID", + response = ParameterContextEntity.class, + notes = "Returns the Parameter Context with the given ID.", + authorizations = { + @Authorization(value = "Read - /parameter-contexts/{id}") + }) + @ApiResponses(value = { + @ApiResponse(code = 400, message = "NiFi was unable to complete the request because it was invalid. The request should not be retried without modification."), + @ApiResponse(code = 401, message = "Client could not be authenticated."), + @ApiResponse(code = 403, message = "Client is not authorized to make this request."), + @ApiResponse(code = 404, message = "The specified resource could not be found."), + @ApiResponse(code = 409, message = "The request was valid but NiFi was not in the appropriate state to process it. Retrying the same request later may be successful.") + }) + public Response getParameterContext(@ApiParam("The ID of the Parameter Context") @PathParam("id") final String parameterContextId) { + if (isReplicateRequest()) { + return replicate(HttpMethod.GET); + } + + // authorize access + serviceFacade.authorizeAccess(lookup -> { + final Authorizable parameterContext = lookup.getParameterContext(parameterContextId); + parameterContext.authorize(authorizer, RequestAction.READ, NiFiUserUtils.getNiFiUser()); + }); + + // get the specified parameter context + final ParameterContextEntity entity = serviceFacade.getParameterContext(parameterContextId, NiFiUserUtils.getNiFiUser()); + entity.setUri(generateResourceUri("parameter-contexts", entity.getId())); + + // generate the response + return generateOkResponse(entity).build(); + } + + + @POST + @Consumes(MediaType.APPLICATION_JSON) + @Produces(MediaType.APPLICATION_JSON) + @ApiOperation( + value = "Create a Parameter Context", + response = ParameterContextEntity.class, + authorizations = { + @Authorization(value = "Write - /parameter-contexts") + }) + @ApiResponses(value = { + @ApiResponse(code = 400, message = "NiFi was unable to complete the request because it was invalid. The request should not be retried without modification."), + @ApiResponse(code = 401, message = "Client could not be authenticated."), + @ApiResponse(code = 403, message = "Client is not authorized to make this request."), + @ApiResponse(code = 404, message = "The specified resource could not be found."), + @ApiResponse(code = 409, message = "The request was valid but NiFi was not in the appropriate state to process it. Retrying the same request later may be successful.") + }) + public Response createParameterContext( + @ApiParam(value = "The Parameter Context.", required = true) final ParameterContextEntity requestEntity) { + + if (requestEntity == null || requestEntity.getComponent() == null) { + throw new IllegalArgumentException("Parameter Context must be specified"); + } + + if (requestEntity.getRevision() == null || (requestEntity.getRevision().getVersion() == null || requestEntity.getRevision().getVersion() != 0)) { + throw new IllegalArgumentException("A revision of 0 must be specified when creating a new Parameter Context."); + } + + final ParameterContextDTO context = requestEntity.getComponent(); + if (context.getName() == null) { + throw new IllegalArgumentException("Parameter Context's Name must be specified"); + } + + if (isReplicateRequest()) { + return replicate(HttpMethod.POST, requestEntity); + } else if (isDisconnectedFromCluster()) { + verifyDisconnectedNodeModification(requestEntity.isDisconnectedNodeAcknowledged()); + } + + return withWriteLock( + serviceFacade, + requestEntity, + lookup -> { + final Authorizable parameterContexts = lookup.getParameterContexts(); + parameterContexts.authorize(authorizer, RequestAction.WRITE, NiFiUserUtils.getNiFiUser()); + }, + () -> serviceFacade.verifyCreateParameterContext(requestEntity.getComponent()), + entity -> { + final String contextId = generateUuid(); + entity.getComponent().setId(contextId); + + final Revision revision = getRevision(entity.getRevision(), contextId); + final ParameterContextEntity contextEntity = serviceFacade.createParameterContext(revision, entity.getComponent()); + + // generate a 201 created response + final String uri = generateResourceUri("parameter-contexts", contextEntity.getId()); + contextEntity.setUri(uri); + return generateCreatedResponse(URI.create(uri), contextEntity).build(); + }); + } + + + @PUT + @Consumes(MediaType.APPLICATION_JSON) + @Produces(MediaType.APPLICATION_JSON) + @Path("{id}") + @ApiOperation( + value = "Modifies a Parameter Context", + response = ParameterContextEntity.class, + notes = "This endpoint will update a Parameter Context to match the provided entity. However, this request will fail if any component is running and is referencing a Parameter in the " + + "Parameter Context. Generally, this endpoint is not called directly. Instead, an update request should be submitted by making a POST to the " + + "/parameter-contexts/update-requests endpoint. That endpoint will, in turn, call this endpoint.", + authorizations = { + @Authorization(value = "Read - /parameter-contexts/{id}"), + @Authorization(value = "Write - /parameter-contexts/{id}") + } + ) + @ApiResponses(value = { + @ApiResponse(code = 400, message = "NiFi was unable to complete the request because it was invalid. The request should not be retried without modification."), + @ApiResponse(code = 401, message = "Client could not be authenticated."), + @ApiResponse(code = 403, message = "Client is not authorized to make this request."), + @ApiResponse(code = 404, message = "The specified resource could not be found."), + @ApiResponse(code = 409, message = "The request was valid but NiFi was not in the appropriate state to process it. Retrying the same request later may be successful.") + }) + public Response updateParameterContext( + @PathParam("id") String contextId, + @ApiParam(value = "The updated Parameter Context", required = true) ParameterContextEntity requestEntity) { + + // Validate request + if (requestEntity.getId() == null) { + throw new IllegalArgumentException("The ID of the Parameter Context must be specified"); + } + if (!requestEntity.getId().equals(contextId)) { + throw new IllegalArgumentException("The ID of the Parameter Context must match the ID specified in the URL's path"); + } + + final ParameterContextDTO updateDto = requestEntity.getComponent(); + if (updateDto == null) { + throw new IllegalArgumentException("The Parameter Context must be supplied"); + } + + final RevisionDTO revisionDto = requestEntity.getRevision(); + if (revisionDto == null) { + throw new IllegalArgumentException("The Revision of the Parameter Context must be specified."); + } + + // Perform the request + if (isReplicateRequest()) { + return replicate(HttpMethod.PUT, requestEntity); + } else if (isDisconnectedFromCluster()) { + verifyDisconnectedNodeModification(requestEntity.isDisconnectedNodeAcknowledged()); + } + + final Revision requestRevision = getRevision(requestEntity.getRevision(), updateDto.getId()); + return withWriteLock( + serviceFacade, + requestEntity, + requestRevision, + lookup -> { + final Authorizable parameterContext = lookup.getParameterContext(contextId); + parameterContext.authorize(authorizer, RequestAction.READ, NiFiUserUtils.getNiFiUser()); + parameterContext.authorize(authorizer, RequestAction.WRITE, NiFiUserUtils.getNiFiUser()); + }, + () -> serviceFacade.verifyUpdateParameterContext(updateDto, true), + (rev, entity) -> { + final ParameterContextEntity updatedEntity = serviceFacade.updateParameterContext(rev, entity.getComponent()); + + updatedEntity.setUri(generateResourceUri("parameter-contexts", entity.getId())); + return generateOkResponse(updatedEntity).build(); + } + ); + } + + + @POST + @Consumes(MediaType.APPLICATION_JSON) + @Produces(MediaType.APPLICATION_JSON) + @Path("update-requests") + @ApiOperation( + value = "Initiate the Update Request of a Parameter Context", + response = ParameterContextUpdateRequestEntity.class, + notes = "This will initiate the process of updating a Parameter Context. Changing the value of a Parameter may require that one or more components be stopped and " + + "restarted, so this acttion may take significantly more time than many other REST API actions. As a result, this endpoint will immediately return a ParameterContextUpdateRequestEntity, " + + "and the process of updating the necessary components will occur asynchronously in the background. The client may then periodically poll the status of the request by " + + "issuing a GET request to /parameter-contexts/update-requests/{requestId}. Once the request is completed, the client is expected to issue a DELETE request to " + + "/parameter-contexts/update-requests/{requestId}.", + authorizations = { + @Authorization(value = "Read - /parameter-contexts/{parameterContextId}"), + @Authorization(value = "Write - /parameter-contexts/{parameterContextId}"), + @Authorization(value = "Read - for every component that is affected by the update"), + @Authorization(value = "Write - for every component that is affected by the update") + }) + @ApiResponses(value = { + @ApiResponse(code = 400, message = "NiFi was unable to complete the request because it was invalid. The request should not be retried without modification."), + @ApiResponse(code = 401, message = "Client could not be authenticated."), + @ApiResponse(code = 403, message = "Client is not authorized to make this request."), + @ApiResponse(code = 404, message = "The specified resource could not be found."), + @ApiResponse(code = 409, message = "The request was valid but NiFi was not in the appropriate state to process it. Retrying the same request later may be successful.") + }) + public Response submitParameterContextUpdate( + @ApiParam(value = "The updated version of the parameter context.", required = true) final ParameterContextEntity requestEntity) { + + // Verify the request + final RevisionDTO revisionDto = requestEntity.getRevision(); + if (revisionDto == null) { + throw new IllegalArgumentException("Parameter Context Revision must be specified"); + } + + final ParameterContextDTO contextDto = requestEntity.getComponent(); + if (contextDto == null) { + throw new IllegalArgumentException("Parameter Context must be specified"); + } + + if (contextDto.getId() == null) { + throw new IllegalArgumentException("Parameter Context's ID must be specified"); + } + + // We will perform the updating of the Parameter Context in a background thread because it can be a long-running process. + // In order to do this, we will need some objects that are only available as Thread-Local variables to the current + // thread, so we will gather the values for these objects up front. + final boolean replicateRequest = isReplicateRequest(); + final ComponentLifecycle componentLifecycle = replicateRequest ? clusterComponentLifecycle : localComponentLifecycle; + final NiFiUser user = NiFiUserUtils.getNiFiUser(); + + // Workflow for this process: + // 1. Determine which components will be affected and are enabled/running + // 2. Verify READ and WRITE permissions for user, for every component that is affected + // 3. Verify READ and WRITE permissions for user, for Parameter Context + // 4. Stop all Processors that are affected. + // 5. Wait for all of the Processors to finish stopping. + // 6. Disable all Controller Services that are affected. + // 7. Wait for all Controller Services to finish disabling. + // 8. Update Parameter Context + // 9. Re-Enable all affected Controller Services + // 10. Re-Start all Processors + + final Set<AffectedComponentEntity> affectedComponents = serviceFacade.getComponentsAffectedByParameterContextUpdate(contextDto); + logger.debug("Received Update Request for Parameter Context: {}; the following {} components will be affected: {}", requestEntity, affectedComponents.size(), affectedComponents); + + final InitiateChangeParameterContextRequestWrapper requestWrapper = new InitiateChangeParameterContextRequestWrapper(requestEntity, componentLifecycle, getAbsolutePath(), + affectedComponents, replicateRequest, user); + + final Revision requestRevision = getRevision(requestEntity.getRevision(), contextDto.getId()); + return withWriteLock( + serviceFacade, + requestWrapper, + requestRevision, + lookup -> { + // Verify READ and WRITE permissions for user, for the Parameter Context itself + final Authorizable parameterContext = lookup.getParameterContext(requestEntity.getId()); + 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 + for (final AffectedComponentEntity entity : affectedComponents) { + final AffectedComponentDTO dto = entity.getComponent(); + if (AffectedComponentDTO.COMPONENT_TYPE_PROCESSOR.equals(dto.getReferenceType())) { + final Authorizable processor = lookup.getProcessor(dto.getId()).getAuthorizable(); + processor.authorize(authorizer, RequestAction.READ, user); + processor.authorize(authorizer, RequestAction.WRITE, user); + } + } + }, + () -> { + // Verify Request + serviceFacade.verifyUpdateParameterContext(contextDto, false); + }, + this::submitUpdateRequest + ); + } + + + @GET + @Consumes(MediaType.WILDCARD) + @Produces(MediaType.APPLICATION_JSON) + @Path("update-requests/{id}") + @ApiOperation( + value = "Returns the Update Request with the given ID", + response = ParameterContextUpdateRequestEntity.class, + notes = "Returns the Update Request with the given ID. Once an Update Request has been created by performing a POST to /nifi-api/parameter-contexts, " + + "that request can subsequently be retrieved via this endpoint, and the request that is fetched will contain the updated state, such as percent complete, the " + + "current state of the request, and any failures. ", + authorizations = { + @Authorization(value = "Only the user that submitted the request can get it") + }) + @ApiResponses(value = { + @ApiResponse(code = 400, message = "NiFi was unable to complete the request because it was invalid. The request should not be retried without modification."), + @ApiResponse(code = 401, message = "Client could not be authenticated."), + @ApiResponse(code = 403, message = "Client is not authorized to make this request."), + @ApiResponse(code = 404, message = "The specified resource could not be found."), + @ApiResponse(code = 409, message = "The request was valid but NiFi was not in the appropriate state to process it. Retrying the same request later may be successful.") + }) + public Response getParameterContextUpdate(@ApiParam("The ID of the Update Request") @PathParam("id") final String updateRequestId) { + return retrieveUpdateRequest("update-requests", updateRequestId); + } + + + @DELETE + @Consumes(MediaType.WILDCARD) + @Produces(MediaType.APPLICATION_JSON) + @Path("update-requests/{id}") + @ApiOperation( + value = "Deletes the Update Request with the given ID", + response = ParameterContextUpdateRequestEntity.class, + notes = "Deletes the Update Request with the given ID. After a request is created via a POST to /nifi-api/parameter-contexts, it is expected " + + "that the client will properly clean up the request by DELETE'ing it, once the Update process has completed. If the request is deleted before the request " + + "completes, then the Update request will finish the step that it is currently performing and then will cancel any subsequent steps.", + authorizations = { + @Authorization(value = "Only the user that submitted the request can remove it") + }) + @ApiResponses(value = { + @ApiResponse(code = 400, message = "NiFi was unable to complete the request because it was invalid. The request should not be retried without modification."), + @ApiResponse(code = 401, message = "Client could not be authenticated."), + @ApiResponse(code = 403, message = "Client is not authorized to make this request."), + @ApiResponse(code = 404, message = "The specified resource could not be found."), + @ApiResponse(code = 409, message = "The request was valid but NiFi was not in the appropriate state to process it. Retrying the same request later may be successful.") + }) + public Response deleteUpdateRequest( + @ApiParam( + value = "Acknowledges that this node is disconnected to allow for mutable requests to proceed.", + required = false + ) + @QueryParam(DISCONNECTED_NODE_ACKNOWLEDGED) @DefaultValue("false") final Boolean disconnectedNodeAcknowledged, + @ApiParam("The ID of the Update Request") @PathParam("id") final String updateRequestId) { + + return deleteUpdateRequest("update-requests", updateRequestId, disconnectedNodeAcknowledged.booleanValue()); + } + + + @DELETE + @Consumes(MediaType.WILDCARD) + @Produces(MediaType.APPLICATION_JSON) + @Path("{id}") + @ApiOperation( + value = "Deletes the Parameter Context with the given ID", + response = ParameterContextEntity.class, + notes = "Deletes the Parameter Context with the given ID.", + authorizations = { + @Authorization(value = "Read - /parameter-contexts/{uuid}"), + @Authorization(value = "Write - /parameter-contexts/{uuid}"), + @Authorization(value = "Read - /process-groups/{uuid}, for any Process Group that is currently bound to the Parameter Context"), + @Authorization(value = "Write - /process-groups/{uuid}, for any Process Group that is currently bound to the Parameter Context") + }) + @ApiResponses(value = { + @ApiResponse(code = 400, message = "NiFi was unable to complete the request because it was invalid. The request should not be retried without modification."), + @ApiResponse(code = 401, message = "Client could not be authenticated."), + @ApiResponse(code = 403, message = "Client is not authorized to make this request."), + @ApiResponse(code = 404, message = "The specified resource could not be found."), + @ApiResponse(code = 409, message = "The request was valid but NiFi was not in the appropriate state to process it. Retrying the same request later may be successful.") + }) + public Response deleteParameterContext( + @ApiParam( + value = "The version is used to verify the client is working with the latest version of the flow.", + required = false) Review comment: We mark the version and client id as not required, but the request must specify at least one of them. You can include a Client ID and no version, or a version and no Client ID, or both. This is the same pattern followed by all DELETE requests. ---------------------------------------------------------------- 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services
