Github user markap14 commented on a diff in the pull request:
https://github.com/apache/nifi/pull/461#discussion_r64259242
--- Diff:
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/SnippetResource.java
---
@@ -0,0 +1,397 @@
+/*
+ * 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 com.wordnik.swagger.annotations.Api;
+import com.wordnik.swagger.annotations.ApiOperation;
+import com.wordnik.swagger.annotations.ApiParam;
+import com.wordnik.swagger.annotations.ApiResponse;
+import com.wordnik.swagger.annotations.ApiResponses;
+import com.wordnik.swagger.annotations.Authorization;
+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.impl.WebClusterManager;
+import org.apache.nifi.controller.Snippet;
+import org.apache.nifi.util.NiFiProperties;
+import org.apache.nifi.web.NiFiServiceFacade;
+import org.apache.nifi.web.Revision;
+import org.apache.nifi.web.api.dto.SnippetDTO;
+import org.apache.nifi.web.api.entity.SnippetEntity;
+
+import javax.servlet.http.HttpServletRequest;
+import javax.ws.rs.Consumes;
+import javax.ws.rs.DELETE;
+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.core.Context;
+import javax.ws.rs.core.MediaType;
+import javax.ws.rs.core.Response;
+import java.net.URI;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * RESTful endpoint for querying dataflow snippets.
+ */
+@Path("/snippets")
+@Api(
+ value = "/snippets",
+ description = "Endpoint for accessing dataflow snippets."
+)
+public class SnippetResource extends ApplicationResource {
+
+ private NiFiServiceFacade serviceFacade;
+ private WebClusterManager clusterManager;
+ private NiFiProperties properties;
+ private Authorizer authorizer;
+
+ /**
+ * Populate the uri's for the specified snippet.
+ *
+ * @param entity processors
+ * @return dtos
+ */
+ private SnippetEntity
populateRemainingSnippetEntityContent(SnippetEntity entity) {
+ if (entity.getSnippet() != null) {
+ populateRemainingSnippetContent(entity.getSnippet());
+ }
+ return entity;
+ }
+
+ /**
+ * Populates the uri for the specified snippet.
+ */
+ private SnippetDTO populateRemainingSnippetContent(SnippetDTO snippet)
{
+ String snippetGroupId = snippet.getParentGroupId();
+
+ // populate the snippet href
+ snippet.setUri(generateResourceUri("process-groups",
snippetGroupId, "snippets", snippet.getId()));
+
+ return snippet;
+ }
+
+ // --------
+ // snippets
+ // --------
+
+ /**
+ * Creates a snippet based off the specified configuration.
+ *
+ * @param httpServletRequest request
+ * @param snippetEntity A snippetEntity
+ * @return A snippetEntity
+ */
+ @POST
+ @Consumes(MediaType.APPLICATION_JSON)
+ @Produces(MediaType.APPLICATION_JSON)
+ // TODO - @PreAuthorize("hasRole('ROLE_DFM')")
+ @ApiOperation(
+ value = "Creates a snippet",
+ response = SnippetEntity.class,
+ authorizations = {
+ @Authorization(value = "Read Only", type = "ROLE_MONITOR"),
+ @Authorization(value = "Data Flow Manager", type = "ROLE_DFM"),
+ @Authorization(value = "Administrator", type = "ROLE_ADMIN")
+ }
+ )
+ @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 createSnippet(
+ @Context HttpServletRequest httpServletRequest,
+ @ApiParam(
+ value = "The snippet configuration details.",
+ required = true
+ )
+ final SnippetEntity snippetEntity) {
+
+ if (snippetEntity == null || snippetEntity.getSnippet() == null) {
+ throw new IllegalArgumentException("Snippet details must be
specified.");
+ }
+
+ if (snippetEntity.getSnippet().getId() != null) {
+ throw new IllegalArgumentException("Snippet ID cannot be
specified.");
+ }
+
+ if (properties.isClusterManager()) {
+ return clusterManager.applyRequest(HttpMethod.POST,
getAbsolutePath(), snippetEntity, getHeaders()).getResponse();
+ }
+
+ // handle expects request (usually from the cluster manager)
+ final boolean validationPhase =
isValidationPhase(httpServletRequest);
+ if (validationPhase || !isTwoPhaseRequest(httpServletRequest)) {
+ // authorize access
+ serviceFacade.authorizeAccess(lookup -> {
+ final SnippetDTO snippet = snippetEntity.getSnippet();
+
+ // ensure read permission to every component in the snippet
+ final Set<Authorizable> authorizables = new HashSet<>();
+
authorizables.addAll(snippet.getProcessGroups().keySet().stream().map(id ->
lookup.getProcessGroup(id)).collect(Collectors.toSet()));
+
authorizables.addAll(snippet.getRemoteProcessGroups().keySet().stream().map(id
-> lookup.getRemoteProcessGroup(id)).collect(Collectors.toSet()));
+
authorizables.addAll(snippet.getProcessors().keySet().stream().map(id ->
lookup.getProcessor(id)).collect(Collectors.toSet()));
+
authorizables.addAll(snippet.getInputPorts().keySet().stream().map(id ->
lookup.getInputPort(id)).collect(Collectors.toSet()));
+
authorizables.addAll(snippet.getOutputPorts().keySet().stream().map(id ->
lookup.getOutputPort(id)).collect(Collectors.toSet()));
+
authorizables.addAll(snippet.getConnections().keySet().stream().map(id ->
lookup.getConnection(id)).collect(Collectors.toSet()));
+
authorizables.addAll(snippet.getConnections().keySet().stream().map(id ->
lookup.getConnection(id)).collect(Collectors.toSet()));
+
authorizables.addAll(snippet.getFunnels().keySet().stream().map(id ->
lookup.getFunnel(id)).collect(Collectors.toSet()));
+ authorizables.stream().forEach(authorizable ->
authorizable.authorize(authorizer, RequestAction.READ));
+ });
+ }
+ if (validationPhase) {
+ return generateContinueResponse().build();
+ }
+
+ // set the processor id as appropriate
+ snippetEntity.getSnippet().setId(generateUuid());
+
+ // create the snippet
+ final SnippetEntity entity =
serviceFacade.createSnippet(snippetEntity.getSnippet());
+ populateRemainingSnippetEntityContent(entity);
+
+ // build the response
+ return
clusterContext(generateCreatedResponse(URI.create(entity.getSnippet().getUri()),
entity)).build();
+ }
+
+ /**
+ * Updates the specified snippet. The contents of the snippet
(component
+ * ids) cannot be updated once the snippet is created.
+ *
+ * @param httpServletRequest request
+ * @param snippetId The id of the snippet.
+ * @param snippetEntity A snippetEntity
+ * @return A snippetEntity
+ */
+ @PUT
+ @Consumes(MediaType.APPLICATION_JSON)
+ @Produces(MediaType.APPLICATION_JSON)
+ @Path("{id}")
+ // TODO - @PreAuthorize("hasRole('ROLE_DFM')")
+ @ApiOperation(
+ value = "Updates a snippet",
+ response = SnippetEntity.class,
+ authorizations = {
+ @Authorization(value = "Data Flow Manager", type = "ROLE_DFM")
+ }
+ )
+ @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 updateSnippet(
+ @Context HttpServletRequest httpServletRequest,
+ @ApiParam(
+ value = "The snippet id.",
+ required = true
+ )
+ @PathParam("id") String snippetId,
+ @ApiParam(
+ value = "The snippet configuration details.",
+ required = true
+ ) final SnippetEntity snippetEntity) {
+
+ if (snippetEntity == null || snippetEntity.getSnippet() == null) {
+ throw new IllegalArgumentException("Snippet details must be
specified.");
+ }
+
+ // ensure the ids are the same
+ final SnippetDTO requestSnippetDTO = snippetEntity.getSnippet();
+ if (!snippetId.equals(requestSnippetDTO.getId())) {
+ throw new IllegalArgumentException(String.format("The snippet
id (%s) in the request body does not equal the "
+ + "snippet id of the requested resource (%s).",
requestSnippetDTO.getId(), snippetId));
+ }
+
+ // replicate if cluster manager
+ if (properties.isClusterManager()) {
+ return clusterManager.applyRequest(HttpMethod.PUT,
getAbsolutePath(), snippetEntity, getHeaders()).getResponse();
+ }
+
+ final NiFiUser user = NiFiUserUtils.getNiFiUser();
+
+ // get the revision from this snippet
+ final Set<Revision> revisions =
serviceFacade.getRevisionsFromSnippet(snippetId);
+
+ // handle expects request (usually from the cluster manager)
+ final boolean validationPhase =
isValidationPhase(httpServletRequest);
+ if (validationPhase || !isTwoPhaseRequest(httpServletRequest)) {
+ // authorize access
+ serviceFacade.authorizeAccess(lookup -> {
+ // ensure write access to the target process group
+ if (requestSnippetDTO.getParentGroupId() != null) {
+
lookup.getProcessGroup(requestSnippetDTO.getParentGroupId()).authorize(authorizer,
RequestAction.WRITE);
+ }
+
+ // ensure read permission to every component in the snippet
+ final Snippet snippet = lookup.getSnippet(snippetId);
+ final Set<Authorizable> authorizables = new HashSet<>();
+
authorizables.addAll(snippet.getProcessGroups().keySet().stream().map(id ->
lookup.getProcessGroup(id)).collect(Collectors.toSet()));
+
authorizables.addAll(snippet.getRemoteProcessGroups().keySet().stream().map(id
-> lookup.getRemoteProcessGroup(id)).collect(Collectors.toSet()));
+
authorizables.addAll(snippet.getProcessors().keySet().stream().map(id ->
lookup.getProcessor(id)).collect(Collectors.toSet()));
+
authorizables.addAll(snippet.getInputPorts().keySet().stream().map(id ->
lookup.getInputPort(id)).collect(Collectors.toSet()));
+
authorizables.addAll(snippet.getOutputPorts().keySet().stream().map(id ->
lookup.getOutputPort(id)).collect(Collectors.toSet()));
+
authorizables.addAll(snippet.getConnections().keySet().stream().map(id ->
lookup.getConnection(id)).collect(Collectors.toSet()));
+
authorizables.addAll(snippet.getConnections().keySet().stream().map(id ->
lookup.getConnection(id)).collect(Collectors.toSet()));
+
authorizables.addAll(snippet.getFunnels().keySet().stream().map(id ->
lookup.getFunnel(id)).collect(Collectors.toSet()));
+ authorizables.stream().forEach(authorizable ->
authorizable.authorize(authorizer, RequestAction.WRITE));
+ });
+
+ serviceFacade.claimRevisions(revisions, user);
+ }
+
+ try {
+ if (validationPhase) {
+ serviceFacade.verifyUpdateSnippet(requestSnippetDTO,
revisions.stream().map(rev ->
rev.getComponentId()).collect(Collectors.toSet()));
+ return generateContinueResponse().build();
+ }
+ } catch (final Exception e) {
+ serviceFacade.cancelRevisions(revisions);
+ throw e;
+ }
+
+ try {
+ // update the snippet
+ final SnippetEntity entity =
serviceFacade.updateSnippet(revisions, snippetEntity.getSnippet());
+ populateRemainingSnippetEntityContent(entity);
+ return clusterContext(generateOkResponse(entity)).build();
+ } finally {
+ serviceFacade.cancelRevisions(revisions);
+ }
+ }
+
+ /**
+ * Removes the specified snippet.
+ *
+ * @param httpServletRequest request
+ * @param snippetId The id of the snippet to remove.
+ * @return A entity containing the client id and an updated revision.
+ */
+ @DELETE
+ @Consumes(MediaType.WILDCARD)
+ @Produces(MediaType.APPLICATION_JSON)
+ @Path("{id}")
+ // TODO - @PreAuthorize("hasRole('ROLE_DFM')")
+ @ApiOperation(
+ value = "Deletes the components in a snippet and drops the
snippet",
+ response = SnippetEntity.class,
+ authorizations = {
+ @Authorization(value = "Data Flow Manager", type = "ROLE_DFM")
+ }
+ )
+ @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 deleteSnippet(
+ @Context final HttpServletRequest httpServletRequest,
+ @ApiParam(
+ value = "The snippet id.",
+ required = true
+ )
+ @PathParam("id") final String snippetId) {
+
+ // replicate if cluster manager
+ if (properties.isClusterManager()) {
+ return clusterManager.applyRequest(HttpMethod.DELETE,
getAbsolutePath(), getRequestParameters(true), getHeaders()).getResponse();
+ }
+
+ final NiFiUser user = NiFiUserUtils.getNiFiUser();
+
+ // get the revision from this snippet
+ final Set<Revision> revisions =
serviceFacade.getRevisionsFromSnippet(snippetId);
+
+ // handle expects request (usually from the cluster manager)
+ final boolean validationPhase =
isValidationPhase(httpServletRequest);
+ if (validationPhase || !isTwoPhaseRequest(httpServletRequest)) {
+ // authorize access
+ serviceFacade.authorizeAccess(lookup -> {
+ // ensure read permission to every component in the snippet
+ final Snippet snippet = lookup.getSnippet(snippetId);
+ final Set<Authorizable> authorizables = new HashSet<>();
+
authorizables.addAll(snippet.getProcessGroups().keySet().stream().map(id ->
lookup.getProcessGroup(id)).collect(Collectors.toSet()));
--- End diff --
And again, same as above... we should probably actually pull this out into
its own method, as it's a decent chunk of code being repeated a few times.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
