NIFI-2007: - Restoring bulletin functionality. - Ensuring appropriate merging of bulletins in clustered responses.
Project: http://git-wip-us.apache.org/repos/asf/nifi/repo Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/0b437e09 Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/0b437e09 Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/0b437e09 Branch: refs/heads/master Commit: 0b437e09a7a6a47ac0f717ffe695ed036d840f1c Parents: 548561e Author: Matt Gilman <[email protected]> Authored: Wed Jun 15 21:42:05 2016 -0400 Committer: Mark Payne <[email protected]> Committed: Fri Jun 17 13:20:20 2016 -0400 ---------------------------------------------------------------------- .../controller/status/ProcessGroupStatus.java | 13 -- .../status/RemoteProcessGroupStatus.java | 15 -- .../nifi/reporting/BulletinRepository.java | 10 ++ .../nifi/util/MockBulletinRepository.java | 10 +- .../nifi/web/api/dto/RemoteProcessGroupDTO.java | 13 +- .../nifi/web/api/entity/ComponentEntity.java | 19 +++ .../http/StandardHttpResponseMerger.java | 29 +++- .../endpoints/AbstractSingleDTOEndpoint.java | 50 +----- .../endpoints/AbstractSingleEntityEndpoint.java | 49 +----- .../endpoints/BulletinBoardEndpointMerger.java | 43 ++--- .../endpoints/ConnectionEndpointMerger.java | 58 +++++++ .../endpoints/ConnectionsEndpointMerger.java | 72 ++++++++ .../ControllerServiceEndpointMerger.java | 108 ++---------- ...ontrollerServiceReferenceEndpointMerger.java | 15 +- .../ControllerServicesEndpointMerger.java | 58 ++++--- .../coordination/http/endpoints/FlowMerger.java | 167 ++++--------------- .../endpoints/InputPortsEndpointMerger.java | 72 ++++++++ .../endpoints/OutputPortsEndpointMerger.java | 72 ++++++++ .../http/endpoints/PortEndpointMerger.java | 75 +++++++++ .../endpoints/ProcessGroupEndpointMerger.java | 85 ++-------- .../endpoints/ProcessGroupsEndpointMerger.java | 72 ++++++++ .../http/endpoints/ProcessorEndpointMerger.java | 54 +----- .../endpoints/ProcessorsEndpointMerger.java | 32 ++-- .../RemoteProcessGroupEndpointMerger.java | 92 ++-------- .../RemoteProcessGroupsEndpointMerger.java | 32 ++-- .../endpoints/ReportingTaskEndpointMerger.java | 39 +---- .../endpoints/ReportingTasksEndpointMerger.java | 54 +++--- .../nifi/cluster/manager/BulletinMerger.java | 85 ++++++++++ .../cluster/manager/ComponentEntityMerger.java | 51 ++++++ .../cluster/manager/ConnectionEntityMerger.java | 41 +++++ .../manager/ConnectionsEntityMerger.java | 38 +++++ .../manager/ControllerServiceEntityMerger.java | 146 ++++++++++++++++ .../manager/ControllerServicesEntityMerger.java | 38 +++++ .../nifi/cluster/manager/ErrorMerger.java | 66 ++++++++ .../nifi/cluster/manager/PortEntityMerger.java | 78 +++++++++ .../nifi/cluster/manager/PortsEntityMerger.java | 38 +++++ .../manager/ProcessGroupEntityMerger.java | 43 +++++ .../manager/ProcessGroupsEntityMerger.java | 38 +++++ .../cluster/manager/ProcessorEntityMerger.java | 78 +++++++++ .../cluster/manager/ProcessorsEntityMerger.java | 38 +++++ .../manager/RemoteProcessGroupEntityMerger.java | 119 +++++++++++++ .../RemoteProcessGroupsEntityMerger.java | 38 +++++ .../manager/ReportingTaskEntityMerger.java | 80 +++++++++ .../manager/ReportingTasksEntityMerger.java | 38 +++++ .../endpoints/TestProcessorEndpointMerger.java | 13 +- .../apache/nifi/controller/FlowController.java | 71 ++++---- .../nifi/events/VolatileBulletinRepository.java | 19 ++- .../nifi/web/StandardNiFiServiceFacade.java | 111 +++++++----- .../org/apache/nifi/web/api/dto/DtoFactory.java | 37 ++-- .../apache/nifi/web/api/dto/EntityFactory.java | 26 ++- .../nifi/web/controller/ControllerFacade.java | 56 +++---- .../src/main/resources/nifi-web-api-context.xml | 3 + .../src/main/webapp/css/flow-status.css | 4 + .../nf-ng-canvas-flow-status-controller.js | 10 +- .../main/webapp/js/nf/canvas/nf-canvas-utils.js | 15 +- .../js/nf/canvas/nf-controller-services.js | 4 +- .../webapp/js/nf/canvas/nf-process-group.js | 2 +- .../main/webapp/js/nf/canvas/nf-processor.js | 4 +- .../js/nf/canvas/nf-remote-process-group.js | 24 +-- .../src/main/webapp/js/nf/canvas/nf-settings.js | 4 +- .../webapp/js/nf/summary/nf-summary-table.js | 17 +- 61 files changed, 1990 insertions(+), 891 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/nifi/blob/0b437e09/nifi-api/src/main/java/org/apache/nifi/controller/status/ProcessGroupStatus.java ---------------------------------------------------------------------- diff --git a/nifi-api/src/main/java/org/apache/nifi/controller/status/ProcessGroupStatus.java b/nifi-api/src/main/java/org/apache/nifi/controller/status/ProcessGroupStatus.java index f60bda8..b890c85 100644 --- a/nifi-api/src/main/java/org/apache/nifi/controller/status/ProcessGroupStatus.java +++ b/nifi-api/src/main/java/org/apache/nifi/controller/status/ProcessGroupStatus.java @@ -19,7 +19,6 @@ package org.apache.nifi.controller.status; import java.util.ArrayList; import java.util.Collection; import java.util.HashMap; -import java.util.List; import java.util.Map; /** @@ -554,18 +553,6 @@ public class ProcessGroupStatus implements Cloneable { merged.setSentContentSize(merged.getSentContentSize() + statusToMerge.getSentContentSize()); merged.setSentCount(merged.getSentCount() + statusToMerge.getSentCount()); merged.setActiveThreadCount(merged.getActiveThreadCount() + statusToMerge.getActiveThreadCount()); - - List<String> mergedAuthenticationIssues = merged.getAuthorizationIssues(); - if (mergedAuthenticationIssues == null) { - mergedAuthenticationIssues = new ArrayList<>(); - } - - final List<String> nodeAuthorizationIssues = statusToMerge.getAuthorizationIssues(); - if (nodeAuthorizationIssues != null && !nodeAuthorizationIssues.isEmpty()) { - mergedAuthenticationIssues.addAll(nodeAuthorizationIssues); - } - - merged.setAuthorizationIssues(mergedAuthenticationIssues); } target.setRemoteProcessGroupStatus(mergedRemoteGroupMap.values()); http://git-wip-us.apache.org/repos/asf/nifi/blob/0b437e09/nifi-api/src/main/java/org/apache/nifi/controller/status/RemoteProcessGroupStatus.java ---------------------------------------------------------------------- diff --git a/nifi-api/src/main/java/org/apache/nifi/controller/status/RemoteProcessGroupStatus.java b/nifi-api/src/main/java/org/apache/nifi/controller/status/RemoteProcessGroupStatus.java index 110972e..2b0efff 100644 --- a/nifi-api/src/main/java/org/apache/nifi/controller/status/RemoteProcessGroupStatus.java +++ b/nifi-api/src/main/java/org/apache/nifi/controller/status/RemoteProcessGroupStatus.java @@ -16,9 +16,6 @@ */ package org.apache.nifi.controller.status; -import java.util.ArrayList; -import java.util.List; -import java.util.Objects; import java.util.concurrent.TimeUnit; /** @@ -40,7 +37,6 @@ public class RemoteProcessGroupStatus implements Cloneable { private Integer inactiveRemotePortCount; private long averageLineageDuration; - private List<String> authorizationIssues = new ArrayList<>(); public String getTargetUri() { return uri; @@ -138,14 +134,6 @@ public class RemoteProcessGroupStatus implements Cloneable { this.inactiveRemotePortCount = inactiveRemotePortCount; } - public List<String> getAuthorizationIssues() { - return new ArrayList<>(authorizationIssues); - } - - public void setAuthorizationIssues(List<String> authorizationIssues) { - this.authorizationIssues = new ArrayList<>(Objects.requireNonNull(authorizationIssues)); - } - public long getAverageLineageDuration() { return averageLineageDuration; } @@ -178,7 +166,6 @@ public class RemoteProcessGroupStatus implements Cloneable { clonedObj.activeRemotePortCount = activeRemotePortCount; clonedObj.inactiveRemotePortCount = inactiveRemotePortCount; clonedObj.averageLineageDuration = averageLineageDuration; - clonedObj.authorizationIssues = getAuthorizationIssues(); return clonedObj; } @@ -209,8 +196,6 @@ public class RemoteProcessGroupStatus implements Cloneable { builder.append(activeRemotePortCount); builder.append(", inactiveRemotePortCount="); builder.append(inactiveRemotePortCount); - builder.append(", authenticationIssues="); - builder.append(authorizationIssues); builder.append("]"); return builder.toString(); } http://git-wip-us.apache.org/repos/asf/nifi/blob/0b437e09/nifi-api/src/main/java/org/apache/nifi/reporting/BulletinRepository.java ---------------------------------------------------------------------- diff --git a/nifi-api/src/main/java/org/apache/nifi/reporting/BulletinRepository.java b/nifi-api/src/main/java/org/apache/nifi/reporting/BulletinRepository.java index 2679099..abc0595 100644 --- a/nifi-api/src/main/java/org/apache/nifi/reporting/BulletinRepository.java +++ b/nifi-api/src/main/java/org/apache/nifi/reporting/BulletinRepository.java @@ -25,6 +25,8 @@ import java.util.List; */ public interface BulletinRepository { + public static final int MAX_BULLETINS_PER_COMPONENT = 5; + /** * Adds a Bulletin to the repository. * @@ -51,6 +53,14 @@ public interface BulletinRepository { List<Bulletin> findBulletins(BulletinQuery bulletinQuery); /** + * Finds all bulletins for the specified source component. + * + * @param sourceId id of the source + * @return bulletins for the given source + */ + List<Bulletin> findBulletinsForSource(String sourceId); + + /** * Finds all bulletins for the specified group. * * @param groupId id of the group http://git-wip-us.apache.org/repos/asf/nifi/blob/0b437e09/nifi-mock/src/main/java/org/apache/nifi/util/MockBulletinRepository.java ---------------------------------------------------------------------- diff --git a/nifi-mock/src/main/java/org/apache/nifi/util/MockBulletinRepository.java b/nifi-mock/src/main/java/org/apache/nifi/util/MockBulletinRepository.java index 04fc496..a52853a 100644 --- a/nifi-mock/src/main/java/org/apache/nifi/util/MockBulletinRepository.java +++ b/nifi-mock/src/main/java/org/apache/nifi/util/MockBulletinRepository.java @@ -16,12 +16,12 @@ */ package org.apache.nifi.util; -import java.util.List; - import org.apache.nifi.reporting.Bulletin; import org.apache.nifi.reporting.BulletinQuery; import org.apache.nifi.reporting.BulletinRepository; +import java.util.List; + public class MockBulletinRepository implements BulletinRepository { @Override @@ -49,6 +49,12 @@ public class MockBulletinRepository implements BulletinRepository { } @Override + public List<Bulletin> findBulletinsForSource(String sourceId) { + // TODO: Implement + return null; + } + + @Override public List<Bulletin> findBulletinsForGroupBySource(String groupId) { // TODO: Implement return null; http://git-wip-us.apache.org/repos/asf/nifi/blob/0b437e09/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/RemoteProcessGroupDTO.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/RemoteProcessGroupDTO.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/RemoteProcessGroupDTO.java index 340e9b8..8a2f9e3 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/RemoteProcessGroupDTO.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/RemoteProcessGroupDTO.java @@ -17,11 +17,12 @@ package org.apache.nifi.web.api.dto; import com.wordnik.swagger.annotations.ApiModelProperty; -import java.util.Date; -import java.util.List; +import org.apache.nifi.web.api.dto.util.DateTimeAdapter; + import javax.xml.bind.annotation.XmlType; import javax.xml.bind.annotation.adapters.XmlJavaTypeAdapter; -import org.apache.nifi.web.api.dto.util.DateTimeAdapter; +import java.util.Collection; +import java.util.Date; /** * Details of a remote process group in this NiFi. @@ -42,7 +43,7 @@ public class RemoteProcessGroupDTO extends ComponentDTO { private String proxyUser; private String proxyPassword; - private List<String> authorizationIssues; + private Collection<String> authorizationIssues; private Boolean transmitting; private Integer inputPortCount; @@ -116,11 +117,11 @@ public class RemoteProcessGroupDTO extends ComponentDTO { @ApiModelProperty( value = "Any remote authorization issues for the remote process group." ) - public List<String> getAuthorizationIssues() { + public Collection<String> getAuthorizationIssues() { return authorizationIssues; } - public void setAuthorizationIssues(List<String> authorizationIssues) { + public void setAuthorizationIssues(Collection<String> authorizationIssues) { this.authorizationIssues = authorizationIssues; } http://git-wip-us.apache.org/repos/asf/nifi/blob/0b437e09/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/entity/ComponentEntity.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/entity/ComponentEntity.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/entity/ComponentEntity.java index 231ce53..fbeeb7f 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/entity/ComponentEntity.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/entity/ComponentEntity.java @@ -18,10 +18,12 @@ package org.apache.nifi.web.api.entity; import com.wordnik.swagger.annotations.ApiModelProperty; import org.apache.nifi.web.api.dto.AccessPolicyDTO; +import org.apache.nifi.web.api.dto.BulletinDTO; import org.apache.nifi.web.api.dto.PositionDTO; import org.apache.nifi.web.api.dto.RevisionDTO; import javax.xml.bind.annotation.XmlRootElement; +import java.util.List; /** * A base type for request/response entities. @@ -33,6 +35,7 @@ public class ComponentEntity extends Entity { private String id; private PositionDTO position; private AccessPolicyDTO accessPolicy; + private List<BulletinDTO> bulletins; /** * @return revision for this request/response @@ -96,6 +99,22 @@ public class ComponentEntity extends Entity { this.accessPolicy = accessPolicy; } + /** + * The bulletins for this component. + * + * @return The bulletins + */ + @ApiModelProperty( + value = "The bulletins for this component." + ) + public List<BulletinDTO> getBulletins() { + return bulletins; + } + + public void setBulletins(List<BulletinDTO> bulletins) { + this.bulletins = bulletins; + } + @Override public int hashCode() { return id.hashCode(); http://git-wip-us.apache.org/repos/asf/nifi/blob/0b437e09/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/StandardHttpResponseMerger.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/StandardHttpResponseMerger.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/StandardHttpResponseMerger.java index ce2e32a..63a2895 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/StandardHttpResponseMerger.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/StandardHttpResponseMerger.java @@ -17,18 +17,11 @@ package org.apache.nifi.cluster.coordination.http; -import java.io.IOException; -import java.net.URI; -import java.util.ArrayList; -import java.util.List; -import java.util.Set; -import java.util.stream.Collectors; - -import javax.ws.rs.core.StreamingOutput; - import org.apache.nifi.cluster.coordination.http.endpoints.BulletinBoardEndpointMerger; import org.apache.nifi.cluster.coordination.http.endpoints.ComponentStateEndpointMerger; +import org.apache.nifi.cluster.coordination.http.endpoints.ConnectionEndpointMerger; import org.apache.nifi.cluster.coordination.http.endpoints.ConnectionStatusEndpiontMerger; +import org.apache.nifi.cluster.coordination.http.endpoints.ConnectionsEndpointMerger; import org.apache.nifi.cluster.coordination.http.endpoints.ControllerServiceEndpointMerger; import org.apache.nifi.cluster.coordination.http.endpoints.ControllerServiceReferenceEndpointMerger; import org.apache.nifi.cluster.coordination.http.endpoints.ControllerServicesEndpointMerger; @@ -38,9 +31,13 @@ import org.apache.nifi.cluster.coordination.http.endpoints.DropRequestEndpiontMe import org.apache.nifi.cluster.coordination.http.endpoints.FlowMerger; import org.apache.nifi.cluster.coordination.http.endpoints.FlowSnippetEndpointMerger; import org.apache.nifi.cluster.coordination.http.endpoints.GroupStatusEndpointMerger; +import org.apache.nifi.cluster.coordination.http.endpoints.InputPortsEndpointMerger; import org.apache.nifi.cluster.coordination.http.endpoints.ListFlowFilesEndpointMerger; +import org.apache.nifi.cluster.coordination.http.endpoints.OutputPortsEndpointMerger; +import org.apache.nifi.cluster.coordination.http.endpoints.PortEndpointMerger; import org.apache.nifi.cluster.coordination.http.endpoints.PortStatusEndpointMerger; import org.apache.nifi.cluster.coordination.http.endpoints.ProcessGroupEndpointMerger; +import org.apache.nifi.cluster.coordination.http.endpoints.ProcessGroupsEndpointMerger; import org.apache.nifi.cluster.coordination.http.endpoints.ProcessorEndpointMerger; import org.apache.nifi.cluster.coordination.http.endpoints.ProcessorStatusEndpointMerger; import org.apache.nifi.cluster.coordination.http.endpoints.ProcessorsEndpointMerger; @@ -59,6 +56,14 @@ import org.apache.nifi.stream.io.NullOutputStream; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import javax.ws.rs.core.StreamingOutput; +import java.io.IOException; +import java.net.URI; +import java.util.ArrayList; +import java.util.List; +import java.util.Set; +import java.util.stream.Collectors; + public class StandardHttpResponseMerger implements HttpResponseMerger { private Logger logger = LoggerFactory.getLogger(StandardHttpResponseMerger.class); @@ -72,9 +77,15 @@ public class StandardHttpResponseMerger implements HttpResponseMerger { endpointMergers.add(new RemoteProcessGroupStatusEndpointMerger()); endpointMergers.add(new ProcessorEndpointMerger()); endpointMergers.add(new ProcessorsEndpointMerger()); + endpointMergers.add(new ConnectionEndpointMerger()); + endpointMergers.add(new ConnectionsEndpointMerger()); + endpointMergers.add(new PortEndpointMerger()); + endpointMergers.add(new InputPortsEndpointMerger()); + endpointMergers.add(new OutputPortsEndpointMerger()); endpointMergers.add(new RemoteProcessGroupEndpointMerger()); endpointMergers.add(new RemoteProcessGroupsEndpointMerger()); endpointMergers.add(new ProcessGroupEndpointMerger()); + endpointMergers.add(new ProcessGroupsEndpointMerger()); endpointMergers.add(new FlowSnippetEndpointMerger()); endpointMergers.add(new ProvenanceQueryEndpointMerger()); endpointMergers.add(new ProvenanceEventEndpointMerger()); http://git-wip-us.apache.org/repos/asf/nifi/blob/0b437e09/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/AbstractSingleDTOEndpoint.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/AbstractSingleDTOEndpoint.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/AbstractSingleDTOEndpoint.java index db3cfd1..2ef3129 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/AbstractSingleDTOEndpoint.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/AbstractSingleDTOEndpoint.java @@ -17,18 +17,16 @@ package org.apache.nifi.cluster.coordination.http.endpoints; -import java.net.URI; -import java.util.Collection; -import java.util.HashMap; -import java.util.HashSet; -import java.util.Map; -import java.util.Set; - import org.apache.nifi.cluster.coordination.http.EndpointResponseMerger; import org.apache.nifi.cluster.manager.NodeResponse; import org.apache.nifi.cluster.protocol.NodeIdentifier; import org.apache.nifi.web.api.entity.Entity; +import java.net.URI; +import java.util.HashMap; +import java.util.Map; +import java.util.Set; + public abstract class AbstractSingleDTOEndpoint<EntityType extends Entity, DtoType> implements EndpointResponseMerger { @Override @@ -51,44 +49,6 @@ public abstract class AbstractSingleDTOEndpoint<EntityType extends Entity, DtoTy return new NodeResponse(clientResponse, responseEntity); } - - /** - * Normalizes the validation errors by prepending the corresponding nodes when the error does not exist across all nodes. - * - * @param validationErrorMap map - * @param totalNodes total - * @return normalized errors - */ - protected Set<String> normalizedMergedValidationErrors(final Map<String, Set<NodeIdentifier>> validationErrorMap, int totalNodes) { - final Set<String> normalizedValidationErrors = new HashSet<>(); - for (final Map.Entry<String, Set<NodeIdentifier>> validationEntry : validationErrorMap.entrySet()) { - final String msg = validationEntry.getKey(); - final Set<NodeIdentifier> nodeIds = validationEntry.getValue(); - - if (nodeIds.size() == totalNodes) { - normalizedValidationErrors.add(msg); - } else { - nodeIds.forEach(id -> normalizedValidationErrors.add(id.getApiAddress() + ":" + id.getApiPort() + " -- " + msg)); - } - } - return normalizedValidationErrors; - } - - /** - * Merges the validation errors into the specified map, recording the corresponding node identifier. - * - * @param validationErrorMap map - * @param nodeId id - * @param nodeValidationErrors errors - */ - protected void mergeValidationErrors(final Map<String, Set<NodeIdentifier>> validationErrorMap, final NodeIdentifier nodeId, final Collection<String> nodeValidationErrors) { - if (nodeValidationErrors != null) { - nodeValidationErrors.stream().forEach( - err -> validationErrorMap.computeIfAbsent(err, k -> new HashSet<NodeIdentifier>()) - .add(nodeId)); - } - } - /** * @return the class that represents the type of Entity that is expected by this response mapper */ http://git-wip-us.apache.org/repos/asf/nifi/blob/0b437e09/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/AbstractSingleEntityEndpoint.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/AbstractSingleEntityEndpoint.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/AbstractSingleEntityEndpoint.java index 026ecec..a7ea4a1 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/AbstractSingleEntityEndpoint.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/AbstractSingleEntityEndpoint.java @@ -17,18 +17,16 @@ package org.apache.nifi.cluster.coordination.http.endpoints; -import java.net.URI; -import java.util.Collection; -import java.util.HashMap; -import java.util.HashSet; -import java.util.Map; -import java.util.Set; - import org.apache.nifi.cluster.coordination.http.EndpointResponseMerger; import org.apache.nifi.cluster.manager.NodeResponse; import org.apache.nifi.cluster.protocol.NodeIdentifier; import org.apache.nifi.web.api.entity.Entity; +import java.net.URI; +import java.util.HashMap; +import java.util.Map; +import java.util.Set; + public abstract class AbstractSingleEntityEndpoint<EntityType extends Entity> implements EndpointResponseMerger { @Override public final NodeResponse merge(final URI uri, final String method, final Set<NodeResponse> successfulResponses, final Set<NodeResponse> problematicResponses, final NodeResponse clientResponse) { @@ -49,43 +47,6 @@ public abstract class AbstractSingleEntityEndpoint<EntityType extends Entity> im } /** - * Merges the validation errors into the specified map, recording the corresponding node identifier. - * - * @param validationErrorMap map - * @param nodeId id - * @param nodeValidationErrors errors - */ - protected void mergeValidationErrors(final Map<String, Set<NodeIdentifier>> validationErrorMap, final NodeIdentifier nodeId, final Collection<String> nodeValidationErrors) { - if (nodeValidationErrors != null) { - nodeValidationErrors.stream().forEach( - err -> validationErrorMap.computeIfAbsent(err, k -> new HashSet<NodeIdentifier>()) - .add(nodeId)); - } - } - - /** - * Normalizes the validation errors by prepending the corresponding nodes when the error does not exist across all nodes. - * - * @param validationErrorMap map - * @param totalNodes total - * @return normalized errors - */ - protected Set<String> normalizedMergedValidationErrors(final Map<String, Set<NodeIdentifier>> validationErrorMap, int totalNodes) { - final Set<String> normalizedValidationErrors = new HashSet<>(); - for (final Map.Entry<String, Set<NodeIdentifier>> validationEntry : validationErrorMap.entrySet()) { - final String msg = validationEntry.getKey(); - final Set<NodeIdentifier> nodeIds = validationEntry.getValue(); - - if (nodeIds.size() == totalNodes) { - normalizedValidationErrors.add(msg); - } else { - nodeIds.forEach(id -> normalizedValidationErrors.add(id.getApiAddress() + ":" + id.getApiPort() + " -- " + msg)); - } - } - return normalizedValidationErrors; - } - - /** * @return the class that represents the type of Entity that is expected by this response mapper */ protected abstract Class<EntityType> getEntityClass(); http://git-wip-us.apache.org/repos/asf/nifi/blob/0b437e09/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/BulletinBoardEndpointMerger.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/BulletinBoardEndpointMerger.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/BulletinBoardEndpointMerger.java index 8da6353..8f0f88d 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/BulletinBoardEndpointMerger.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/BulletinBoardEndpointMerger.java @@ -17,21 +17,21 @@ package org.apache.nifi.cluster.coordination.http.endpoints; +import org.apache.nifi.cluster.manager.BulletinMerger; +import org.apache.nifi.cluster.manager.NodeResponse; +import org.apache.nifi.cluster.protocol.NodeIdentifier; +import org.apache.nifi.web.api.dto.BulletinBoardDTO; +import org.apache.nifi.web.api.dto.BulletinDTO; +import org.apache.nifi.web.api.entity.BulletinBoardEntity; + import java.net.URI; import java.util.ArrayList; -import java.util.Collections; -import java.util.Comparator; +import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Set; import java.util.regex.Pattern; -import org.apache.nifi.cluster.manager.NodeResponse; -import org.apache.nifi.cluster.protocol.NodeIdentifier; -import org.apache.nifi.web.api.dto.BulletinBoardDTO; -import org.apache.nifi.web.api.dto.BulletinDTO; -import org.apache.nifi.web.api.entity.BulletinBoardEntity; - public class BulletinBoardEndpointMerger extends AbstractSingleDTOEndpoint<BulletinBoardEntity, BulletinBoardDTO> { public static final Pattern BULLETIN_BOARD_URI_PATTERN = Pattern.compile("/nifi-api/flow/bulletin-board"); @@ -52,31 +52,16 @@ public class BulletinBoardEndpointMerger extends AbstractSingleDTOEndpoint<Bulle @Override protected void mergeResponses(BulletinBoardDTO clientDto, Map<NodeIdentifier, BulletinBoardDTO> dtoMap, Set<NodeResponse> successfulResponses, Set<NodeResponse> problematicResponses) { - final List<BulletinDTO> bulletinDtos = new ArrayList<>(); + final Map<NodeIdentifier, List<BulletinDTO>> bulletinDtos = new HashMap<>(); for (final Map.Entry<NodeIdentifier, BulletinBoardDTO> entry : dtoMap.entrySet()) { - final NodeIdentifier nodeId = entry.getKey(); + final NodeIdentifier nodeIdentifier = entry.getKey(); final BulletinBoardDTO boardDto = entry.getValue(); - final String nodeAddress = nodeId.getApiAddress() + ":" + nodeId.getApiPort(); - - for (final BulletinDTO bulletin : boardDto.getBulletins()) { - bulletin.setNodeAddress(nodeAddress); - bulletinDtos.add(bulletin); - } + boardDto.getBulletins().forEach(bulletin -> { + bulletinDtos.computeIfAbsent(nodeIdentifier, nodeId -> new ArrayList<>()).add(bulletin); + }); } - Collections.sort(bulletinDtos, new Comparator<BulletinDTO>() { - @Override - public int compare(final BulletinDTO o1, final BulletinDTO o2) { - final int timeComparison = o1.getTimestamp().compareTo(o2.getTimestamp()); - if (timeComparison != 0) { - return timeComparison; - } - - return o1.getNodeAddress().compareTo(o2.getNodeAddress()); - } - }); - - clientDto.setBulletins(bulletinDtos); + clientDto.setBulletins(BulletinMerger.mergeBulletins(bulletinDtos)); } } http://git-wip-us.apache.org/repos/asf/nifi/blob/0b437e09/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ConnectionEndpointMerger.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ConnectionEndpointMerger.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ConnectionEndpointMerger.java new file mode 100644 index 0000000..30c6f77 --- /dev/null +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ConnectionEndpointMerger.java @@ -0,0 +1,58 @@ +/* + * 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.cluster.coordination.http.endpoints; + +import org.apache.nifi.cluster.coordination.http.EndpointResponseMerger; +import org.apache.nifi.cluster.manager.ConnectionEntityMerger; +import org.apache.nifi.cluster.manager.NodeResponse; +import org.apache.nifi.cluster.protocol.NodeIdentifier; +import org.apache.nifi.web.api.entity.ConnectionEntity; + +import java.net.URI; +import java.util.Map; +import java.util.Set; +import java.util.regex.Pattern; + +public class ConnectionEndpointMerger extends AbstractSingleEntityEndpoint<ConnectionEntity> implements EndpointResponseMerger { + public static final Pattern PROCESSORS_URI_PATTERN = Pattern.compile("/nifi-api/process-groups/(?:(?:root)|(?:[a-f0-9\\-]{36}))/connections"); + public static final Pattern PROCESSOR_URI_PATTERN = Pattern.compile("/nifi-api/connections/[a-f0-9\\-]{36}"); + + @Override + public boolean canHandle(final URI uri, final String method) { + if (("GET".equalsIgnoreCase(method) || "PUT".equalsIgnoreCase(method)) && (PROCESSOR_URI_PATTERN.matcher(uri.getPath()).matches())) { + return true; + } else if ("POST".equalsIgnoreCase(method) && PROCESSORS_URI_PATTERN.matcher(uri.getPath()).matches()) { + return true; + } + + return false; + } + + @Override + protected Class<ConnectionEntity> getEntityClass() { + return ConnectionEntity.class; + } + + + @Override + protected void mergeResponses(final ConnectionEntity clientEntity, final Map<NodeIdentifier, ConnectionEntity> entityMap, final Set<NodeResponse> successfulResponses, + final Set<NodeResponse> problematicResponses) { + + ConnectionEntityMerger.mergeConnections(clientEntity, entityMap); + } +} http://git-wip-us.apache.org/repos/asf/nifi/blob/0b437e09/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ConnectionsEndpointMerger.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ConnectionsEndpointMerger.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ConnectionsEndpointMerger.java new file mode 100644 index 0000000..661350d --- /dev/null +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ConnectionsEndpointMerger.java @@ -0,0 +1,72 @@ +/* + * 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.cluster.coordination.http.endpoints; + +import org.apache.nifi.cluster.coordination.http.EndpointResponseMerger; +import org.apache.nifi.cluster.manager.ConnectionsEntityMerger; +import org.apache.nifi.cluster.manager.NodeResponse; +import org.apache.nifi.cluster.protocol.NodeIdentifier; +import org.apache.nifi.web.api.entity.ConnectionEntity; +import org.apache.nifi.web.api.entity.ConnectionsEntity; + +import java.net.URI; +import java.util.HashMap; +import java.util.Map; +import java.util.Set; +import java.util.regex.Pattern; + +public class ConnectionsEndpointMerger implements EndpointResponseMerger { + public static final Pattern CONNECTIONS_URI_PATTERN = Pattern.compile("/nifi-api/process-groups/(?:(?:root)|(?:[a-f0-9\\-]{36}))/connections"); + + @Override + public boolean canHandle(final URI uri, final String method) { + return "GET".equalsIgnoreCase(method) && CONNECTIONS_URI_PATTERN.matcher(uri.getPath()).matches(); + } + + @Override + public final NodeResponse merge(final URI uri, final String method, final Set<NodeResponse> successfulResponses, final Set<NodeResponse> problematicResponses, final NodeResponse clientResponse) { + if (!canHandle(uri, method)) { + throw new IllegalArgumentException("Cannot use Endpoint Mapper of type " + getClass().getSimpleName() + " to map responses for URI " + uri + ", HTTP Method " + method); + } + + final ConnectionsEntity responseEntity = clientResponse.getClientResponse().getEntity(ConnectionsEntity.class); + final Set<ConnectionEntity> connectionEntities = responseEntity.getConnections(); + + final Map<String, Map<NodeIdentifier, ConnectionEntity>> entityMap = new HashMap<>(); + for (final NodeResponse nodeResponse : successfulResponses) { + final ConnectionsEntity nodeResponseEntity = nodeResponse == clientResponse ? responseEntity : nodeResponse.getClientResponse().getEntity(ConnectionsEntity.class); + final Set<ConnectionEntity> nodeConnectionEntities = nodeResponseEntity.getConnections(); + + for (final ConnectionEntity nodeConnectionEntity : nodeConnectionEntities) { + final NodeIdentifier nodeId = nodeResponse.getNodeId(); + Map<NodeIdentifier, ConnectionEntity> innerMap = entityMap.get(nodeId); + if (innerMap == null) { + innerMap = new HashMap<>(); + entityMap.put(nodeConnectionEntity.getId(), innerMap); + } + + innerMap.put(nodeResponse.getNodeId(), nodeConnectionEntity); + } + } + + ConnectionsEntityMerger.mergeConnections(connectionEntities, entityMap); + + // create a new client response + return new NodeResponse(clientResponse, responseEntity); + } +} http://git-wip-us.apache.org/repos/asf/nifi/blob/0b437e09/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ControllerServiceEndpointMerger.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ControllerServiceEndpointMerger.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ControllerServiceEndpointMerger.java index 10fa281..879d3b5 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ControllerServiceEndpointMerger.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ControllerServiceEndpointMerger.java @@ -17,29 +17,27 @@ package org.apache.nifi.cluster.coordination.http.endpoints; +import org.apache.nifi.cluster.coordination.http.EndpointResponseMerger; +import org.apache.nifi.cluster.manager.ControllerServiceEntityMerger; import org.apache.nifi.cluster.manager.NodeResponse; import org.apache.nifi.cluster.protocol.NodeIdentifier; -import org.apache.nifi.controller.service.ControllerServiceState; -import org.apache.nifi.web.api.dto.ControllerServiceDTO; -import org.apache.nifi.web.api.dto.ControllerServiceReferencingComponentDTO; import org.apache.nifi.web.api.entity.ControllerServiceEntity; -import org.apache.nifi.web.api.entity.ControllerServiceReferencingComponentEntity; import java.net.URI; -import java.util.HashMap; import java.util.Map; import java.util.Set; import java.util.regex.Pattern; -public class ControllerServiceEndpointMerger extends AbstractSingleDTOEndpoint<ControllerServiceEntity, ControllerServiceDTO> { - public static final String CONTROLLER_SERVICES_URI = "/nifi-api/controller-services/node"; - public static final Pattern CONTROLLER_SERVICE_URI_PATTERN = Pattern.compile("/nifi-api/controller-services/node/[a-f0-9\\-]{36}"); +public class ControllerServiceEndpointMerger extends AbstractSingleEntityEndpoint<ControllerServiceEntity> implements EndpointResponseMerger { + public static final String CONTROLLER_CONTROLLER_SERVICES_URI = "/nifi-api/controller/controller-services"; + public static final Pattern PROCESS_GROUPS_CONTROLLER_SERVICES_URI = Pattern.compile("/nifi-api/process-groups/(?:(?:root)|(?:[a-f0-9\\-]{36}))/controller-services"); + public static final Pattern CONTROLLER_SERVICE_URI_PATTERN = Pattern.compile("/nifi-api/controller-services/[a-f0-9\\-]{36}"); @Override public boolean canHandle(URI uri, String method) { if (("GET".equalsIgnoreCase(method) || "PUT".equalsIgnoreCase(method)) && CONTROLLER_SERVICE_URI_PATTERN.matcher(uri.getPath()).matches()) { return true; - } else if ("POST".equalsIgnoreCase(method) && CONTROLLER_SERVICES_URI.equals(uri.getPath())) { + } else if ("POST".equalsIgnoreCase(method) && (CONTROLLER_CONTROLLER_SERVICES_URI.equals(uri.getPath()) || PROCESS_GROUPS_CONTROLLER_SERVICES_URI.matcher(uri.getPath()).matches())) { return true; } @@ -52,95 +50,9 @@ public class ControllerServiceEndpointMerger extends AbstractSingleDTOEndpoint<C } @Override - protected ControllerServiceDTO getDto(ControllerServiceEntity entity) { - return entity.getComponent(); - } - - @Override - protected void mergeResponses(ControllerServiceDTO clientDto, Map<NodeIdentifier, ControllerServiceDTO> dtoMap, Set<NodeResponse> successfulResponses, Set<NodeResponse> problematicResponses) { - final Map<String, Set<NodeIdentifier>> validationErrorMap = new HashMap<>(); - final Set<ControllerServiceReferencingComponentEntity> referencingComponents = clientDto.getReferencingComponents(); - final Map<NodeIdentifier, Set<ControllerServiceReferencingComponentEntity>> nodeReferencingComponentsMap = new HashMap<>(); - - String state = null; - for (final Map.Entry<NodeIdentifier, ControllerServiceDTO> nodeEntry : dtoMap.entrySet()) { - final NodeIdentifier nodeId = nodeEntry.getKey(); - final ControllerServiceDTO nodeControllerService = nodeEntry.getValue(); - - if (state == null) { - if (ControllerServiceState.DISABLING.name().equals(nodeControllerService.getState())) { - state = ControllerServiceState.DISABLING.name(); - } else if (ControllerServiceState.ENABLING.name().equals(nodeControllerService.getState())) { - state = ControllerServiceState.ENABLING.name(); - } - } - - nodeReferencingComponentsMap.put(nodeId, nodeControllerService.getReferencingComponents()); - - // merge the validation errors - mergeValidationErrors(validationErrorMap, nodeId, nodeControllerService.getValidationErrors()); - } - - // merge the referencing components - mergeControllerServiceReferences(referencingComponents, nodeReferencingComponentsMap); - - // store the 'transition' state is applicable - if (state != null) { - clientDto.setState(state); - } + protected void mergeResponses(ControllerServiceEntity clientEntity, Map<NodeIdentifier, ControllerServiceEntity> entityMap, + Set<NodeResponse> successfulResponses, Set<NodeResponse> problematicResponses) { - // set the merged the validation errors - clientDto.setValidationErrors(normalizedMergedValidationErrors(validationErrorMap, dtoMap.size())); + ControllerServiceEntityMerger.mergeControllerServices(clientEntity, entityMap); } - - public static void mergeControllerServiceReferences(Set<ControllerServiceReferencingComponentEntity> referencingComponents, - Map<NodeIdentifier, Set<ControllerServiceReferencingComponentEntity>> referencingComponentMap) { - - final Map<String, Integer> activeThreadCounts = new HashMap<>(); - final Map<String, String> states = new HashMap<>(); - for (final Map.Entry<NodeIdentifier, Set<ControllerServiceReferencingComponentEntity>> nodeEntry : referencingComponentMap.entrySet()) { - final Set<ControllerServiceReferencingComponentEntity> nodeReferencingComponents = nodeEntry.getValue(); - - // go through all the nodes referencing components - if (nodeReferencingComponents != null) { - for (final ControllerServiceReferencingComponentEntity nodeReferencingComponentEntity : nodeReferencingComponents) { - final ControllerServiceReferencingComponentDTO nodeReferencingComponent = nodeReferencingComponentEntity.getComponent(); - - // handle active thread counts - if (nodeReferencingComponent.getActiveThreadCount() != null && nodeReferencingComponent.getActiveThreadCount() > 0) { - final Integer current = activeThreadCounts.get(nodeReferencingComponent.getId()); - if (current == null) { - activeThreadCounts.put(nodeReferencingComponent.getId(), nodeReferencingComponent.getActiveThreadCount()); - } else { - activeThreadCounts.put(nodeReferencingComponent.getId(), nodeReferencingComponent.getActiveThreadCount() + current); - } - } - - // handle controller service state - final String state = states.get(nodeReferencingComponent.getId()); - if (state == null) { - if (ControllerServiceState.DISABLING.name().equals(nodeReferencingComponent.getState())) { - states.put(nodeReferencingComponent.getId(), ControllerServiceState.DISABLING.name()); - } else if (ControllerServiceState.ENABLING.name().equals(nodeReferencingComponent.getState())) { - states.put(nodeReferencingComponent.getId(), ControllerServiceState.ENABLING.name()); - } - } - } - } - } - - // go through each referencing components - for (final ControllerServiceReferencingComponentEntity referencingComponent : referencingComponents) { - final Integer activeThreadCount = activeThreadCounts.get(referencingComponent.getId()); - if (activeThreadCount != null) { - referencingComponent.getComponent().setActiveThreadCount(activeThreadCount); - } - - final String state = states.get(referencingComponent.getId()); - if (state != null) { - referencingComponent.getComponent().setState(state); - } - } - } - } http://git-wip-us.apache.org/repos/asf/nifi/blob/0b437e09/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ControllerServiceReferenceEndpointMerger.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ControllerServiceReferenceEndpointMerger.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ControllerServiceReferenceEndpointMerger.java index c299054..129d36f 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ControllerServiceReferenceEndpointMerger.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ControllerServiceReferenceEndpointMerger.java @@ -17,18 +17,19 @@ package org.apache.nifi.cluster.coordination.http.endpoints; -import java.net.URI; -import java.util.HashMap; -import java.util.Map; -import java.util.Set; -import java.util.regex.Pattern; - import org.apache.nifi.cluster.coordination.http.EndpointResponseMerger; +import org.apache.nifi.cluster.manager.ControllerServiceEntityMerger; import org.apache.nifi.cluster.manager.NodeResponse; import org.apache.nifi.cluster.protocol.NodeIdentifier; import org.apache.nifi.web.api.entity.ControllerServiceReferencingComponentEntity; import org.apache.nifi.web.api.entity.ControllerServiceReferencingComponentsEntity; +import java.net.URI; +import java.util.HashMap; +import java.util.Map; +import java.util.Set; +import java.util.regex.Pattern; + public class ControllerServiceReferenceEndpointMerger implements EndpointResponseMerger { public static final Pattern CONTROLLER_SERVICE_REFERENCES_URI_PATTERN = Pattern.compile("/nifi-api/controller-services/node/[a-f0-9\\-]{36}/references"); @@ -59,7 +60,7 @@ public class ControllerServiceReferenceEndpointMerger implements EndpointRespons resultsMap.put(nodeResponse.getNodeId(), nodeReferencingComponents); } - ControllerServiceEndpointMerger.mergeControllerServiceReferences(referencingComponents, resultsMap); + ControllerServiceEntityMerger.mergeControllerServiceReferences(referencingComponents, resultsMap); return new NodeResponse(clientResponse, responseEntity); } http://git-wip-us.apache.org/repos/asf/nifi/blob/0b437e09/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ControllerServicesEndpointMerger.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ControllerServicesEndpointMerger.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ControllerServicesEndpointMerger.java index ba09d9d..854a027 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ControllerServicesEndpointMerger.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ControllerServicesEndpointMerger.java @@ -17,47 +17,57 @@ package org.apache.nifi.cluster.coordination.http.endpoints; +import org.apache.nifi.cluster.coordination.http.EndpointResponseMerger; +import org.apache.nifi.cluster.manager.ControllerServicesEntityMerger; import org.apache.nifi.cluster.manager.NodeResponse; import org.apache.nifi.cluster.protocol.NodeIdentifier; import org.apache.nifi.web.api.entity.ControllerServiceEntity; import org.apache.nifi.web.api.entity.ControllerServicesEntity; import java.net.URI; +import java.util.HashMap; import java.util.Map; import java.util.Set; -import java.util.stream.Collectors; +import java.util.regex.Pattern; -public class ControllerServicesEndpointMerger extends AbstractMultiEntityEndpoint<ControllerServicesEntity, ControllerServiceEntity> { - public static final String CONTROLLER_SERVICES_URI = "/nifi-api/controller-services/node"; +public class ControllerServicesEndpointMerger implements EndpointResponseMerger { + public static final String CONTROLLER_SERVICES_URI = "/nifi-api/flow/controller/controller-services"; + public static final Pattern PROCESS_GROUPS_CONTROLLER_SERVICES_URI = Pattern.compile("/nifi-api/flow/process-groups/(?:(?:root)|(?:[a-f0-9\\-]{36}))/controller-services"); @Override public boolean canHandle(URI uri, String method) { - return "GET".equalsIgnoreCase(method) && CONTROLLER_SERVICES_URI.equals(uri.getPath()); + return "GET".equalsIgnoreCase(method) && (CONTROLLER_SERVICES_URI.equals(uri.getPath()) || PROCESS_GROUPS_CONTROLLER_SERVICES_URI.matcher(uri.getPath()).matches()); } @Override - protected Class<ControllerServicesEntity> getEntityClass() { - return ControllerServicesEntity.class; - } + public final NodeResponse merge(final URI uri, final String method, final Set<NodeResponse> successfulResponses, final Set<NodeResponse> problematicResponses, final NodeResponse clientResponse) { + if (!canHandle(uri, method)) { + throw new IllegalArgumentException("Cannot use Endpoint Mapper of type " + getClass().getSimpleName() + " to map responses for URI " + uri + ", HTTP Method " + method); + } - @Override - protected Set<ControllerServiceEntity> getDtos(ControllerServicesEntity entity) { - return entity.getControllerServices(); - } + final ControllerServicesEntity responseEntity = clientResponse.getClientResponse().getEntity(ControllerServicesEntity.class); + final Set<ControllerServiceEntity> controllerServiceEntities = responseEntity.getControllerServices(); - @Override - protected String getComponentId(ControllerServiceEntity entity) { - return entity.getComponent().getId(); - } + final Map<String, Map<NodeIdentifier, ControllerServiceEntity>> entityMap = new HashMap<>(); + for (final NodeResponse nodeResponse : successfulResponses) { + final ControllerServicesEntity nodeResponseEntity = nodeResponse == clientResponse ? responseEntity : nodeResponse.getClientResponse().getEntity(ControllerServicesEntity.class); + final Set<ControllerServiceEntity> nodeControllerServiceEntities = nodeResponseEntity.getControllerServices(); - @Override - protected void mergeResponses(ControllerServiceEntity entity, Map<NodeIdentifier, ControllerServiceEntity> entityMap, - Set<NodeResponse> successfulResponses, Set<NodeResponse> problematicResponses) { - - new ControllerServiceEndpointMerger().mergeResponses( - entity.getComponent(), - entityMap.entrySet().stream().collect(Collectors.toMap(Map.Entry::getKey, e -> e.getValue().getComponent())), - successfulResponses, - problematicResponses); + for (final ControllerServiceEntity nodeControllerServiceEntity : nodeControllerServiceEntities) { + final NodeIdentifier nodeId = nodeResponse.getNodeId(); + Map<NodeIdentifier, ControllerServiceEntity> innerMap = entityMap.get(nodeId); + if (innerMap == null) { + innerMap = new HashMap<>(); + entityMap.put(nodeControllerServiceEntity.getId(), innerMap); + } + + innerMap.put(nodeResponse.getNodeId(), nodeControllerServiceEntity); + } + } + + ControllerServicesEntityMerger.mergeControllerServices(controllerServiceEntities, entityMap); + + // create a new client response + return new NodeResponse(clientResponse, responseEntity); } } http://git-wip-us.apache.org/repos/asf/nifi/blob/0b437e09/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/FlowMerger.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/FlowMerger.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/FlowMerger.java index c033d19..0cbf213 100644 --- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/FlowMerger.java +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/FlowMerger.java @@ -17,16 +17,15 @@ package org.apache.nifi.cluster.coordination.http.endpoints; +import org.apache.nifi.cluster.manager.ConnectionsEntityMerger; import org.apache.nifi.cluster.manager.NodeResponse; -import org.apache.nifi.cluster.manager.StatusMerger; +import org.apache.nifi.cluster.manager.PortsEntityMerger; +import org.apache.nifi.cluster.manager.ProcessGroupsEntityMerger; +import org.apache.nifi.cluster.manager.ProcessorsEntityMerger; +import org.apache.nifi.cluster.manager.RemoteProcessGroupsEntityMerger; import org.apache.nifi.cluster.protocol.NodeIdentifier; import org.apache.nifi.web.api.dto.flow.FlowDTO; import org.apache.nifi.web.api.dto.flow.ProcessGroupFlowDTO; -import org.apache.nifi.web.api.dto.status.ConnectionStatusDTO; -import org.apache.nifi.web.api.dto.status.PortStatusDTO; -import org.apache.nifi.web.api.dto.status.ProcessGroupStatusDTO; -import org.apache.nifi.web.api.dto.status.ProcessorStatusDTO; -import org.apache.nifi.web.api.dto.status.RemoteProcessGroupStatusDTO; import org.apache.nifi.web.api.entity.ConnectionEntity; import org.apache.nifi.web.api.entity.FunnelEntity; import org.apache.nifi.web.api.entity.LabelEntity; @@ -68,23 +67,31 @@ public class FlowMerger extends AbstractSingleDTOEndpoint<ProcessGroupFlowEntity final Set<NodeResponse> successfulResponses, final Set<NodeResponse> problematicResponses) { final FlowDTO flowDto = clientDto.getFlow(); - - final Map<String, List<ConnectionEntity>> connections = new HashMap<>(); + final Set<ConnectionEntity> clientConnections = flowDto.getConnections(); + final Set<ProcessorEntity> clientProcessors = flowDto.getProcessors(); + final Set<PortEntity> clientInputPorts = flowDto.getInputPorts(); + final Set<PortEntity> clientOutputPorts = flowDto.getOutputPorts(); + final Set<RemoteProcessGroupEntity> clientRemoteProcessGroups = flowDto.getRemoteProcessGroups(); + final Set<ProcessGroupEntity> clientProcessGroups = flowDto.getProcessGroups(); + + final Map<String, Map<NodeIdentifier, ConnectionEntity>> connections = new HashMap<>(); final Map<String, List<FunnelEntity>> funnels = new HashMap<>(); - final Map<String, List<PortEntity>> inputPorts = new HashMap<>(); + final Map<String, Map<NodeIdentifier, PortEntity>> inputPorts = new HashMap<>(); final Map<String, List<LabelEntity>> labels = new HashMap<>(); - final Map<String, List<PortEntity>> outputPorts = new HashMap<>(); - final Map<String, List<ProcessorEntity>> processors = new HashMap<>(); - final Map<String, List<RemoteProcessGroupEntity>> rpgs = new HashMap<>(); - final Map<String, List<ProcessGroupEntity>> processGroups = new HashMap<>(); - - // Create mapping of ComponentID -> all components with that ID (one per node) - for (final ProcessGroupFlowDTO nodeGroupFlowDto : dtoMap.values()) { + final Map<String, Map<NodeIdentifier, PortEntity>> outputPorts = new HashMap<>(); + final Map<String, Map<NodeIdentifier, ProcessorEntity>> processors = new HashMap<>(); + final Map<String, Map<NodeIdentifier, RemoteProcessGroupEntity>> rpgs = new HashMap<>(); + final Map<String, Map<NodeIdentifier, ProcessGroupEntity>> processGroups = new HashMap<>(); + + // Create mapping of ComponentID -> [nodeId, entity on that node] + for (final Map.Entry<NodeIdentifier, ProcessGroupFlowDTO> nodeGroupFlowEntry : dtoMap.entrySet()) { + final NodeIdentifier nodeIdentifier = nodeGroupFlowEntry.getKey(); + final ProcessGroupFlowDTO nodeGroupFlowDto = nodeGroupFlowEntry.getValue(); final FlowDTO nodeFlowDto = nodeGroupFlowDto.getFlow(); // Merge connection statuses for (final ConnectionEntity entity : nodeFlowDto.getConnections()) { - connections.computeIfAbsent(entity.getId(), id -> new ArrayList<>()).add(entity); + connections.computeIfAbsent(entity.getId(), id -> new HashMap<>()).computeIfAbsent(nodeIdentifier, nodeId -> entity); } for (final FunnelEntity entity : nodeFlowDto.getFunnels()) { @@ -92,11 +99,11 @@ public class FlowMerger extends AbstractSingleDTOEndpoint<ProcessGroupFlowEntity } for (final PortEntity entity : nodeFlowDto.getInputPorts()) { - inputPorts.computeIfAbsent(entity.getId(), id -> new ArrayList<>()).add(entity); + inputPorts.computeIfAbsent(entity.getId(), id -> new HashMap<>()).computeIfAbsent(nodeIdentifier, nodeId -> entity); } for (final PortEntity entity : nodeFlowDto.getOutputPorts()) { - outputPorts.computeIfAbsent(entity.getId(), id -> new ArrayList<>()).add(entity); + outputPorts.computeIfAbsent(entity.getId(), id -> new HashMap<>()).computeIfAbsent(nodeIdentifier, nodeId -> entity); } for (final LabelEntity entity : nodeFlowDto.getLabels()) { @@ -104,15 +111,15 @@ public class FlowMerger extends AbstractSingleDTOEndpoint<ProcessGroupFlowEntity } for (final ProcessorEntity entity : nodeFlowDto.getProcessors()) { - processors.computeIfAbsent(entity.getId(), id -> new ArrayList<>()).add(entity); + processors.computeIfAbsent(entity.getId(), id -> new HashMap<>()).computeIfAbsent(nodeIdentifier, nodeId -> entity); } for (final RemoteProcessGroupEntity entity : nodeFlowDto.getRemoteProcessGroups()) { - rpgs.computeIfAbsent(entity.getId(), id -> new ArrayList<>()).add(entity); + rpgs.computeIfAbsent(entity.getId(), id -> new HashMap<>()).computeIfAbsent(nodeIdentifier, nodeId -> entity); } for (final ProcessGroupEntity entity : nodeFlowDto.getProcessGroups()) { - processGroups.computeIfAbsent(entity.getId(), id -> new ArrayList<>()).add(entity); + processGroups.computeIfAbsent(entity.getId(), id -> new HashMap<>()).computeIfAbsent(nodeIdentifier, nodeId -> entity); } } @@ -121,11 +128,7 @@ public class FlowMerger extends AbstractSingleDTOEndpoint<ProcessGroupFlowEntity // // Merge connections - final Set<ConnectionEntity> mergedConnections = new HashSet<>(); - for (final List<ConnectionEntity> connectionList : connections.values()) { - mergedConnections.add(mergeConnections(connectionList)); - } - flowDto.setConnections(mergedConnections); + ConnectionsEntityMerger.mergeConnections(clientConnections, connections); // Merge funnel statuses final Set<FunnelEntity> mergedFunnels = new HashSet<>(); @@ -135,18 +138,10 @@ public class FlowMerger extends AbstractSingleDTOEndpoint<ProcessGroupFlowEntity flowDto.setFunnels(mergedFunnels); // Merge input ports - final Set<PortEntity> mergedInputPorts = new HashSet<>(); - for (final List<PortEntity> portList : inputPorts.values()) { - mergedInputPorts.add(mergePorts(portList)); - } - flowDto.setInputPorts(mergedInputPorts); + PortsEntityMerger.mergePorts(clientInputPorts, inputPorts); // Merge output ports - final Set<PortEntity> mergedOutputPorts = new HashSet<>(); - for (final List<PortEntity> portList : outputPorts.values()) { - mergedOutputPorts.add(mergePorts(portList)); - } - flowDto.setOutputPorts(mergedOutputPorts); + PortsEntityMerger.mergePorts(clientOutputPorts, outputPorts); // Merge labels final Set<LabelEntity> mergedLabels = new HashSet<>(); @@ -155,57 +150,14 @@ public class FlowMerger extends AbstractSingleDTOEndpoint<ProcessGroupFlowEntity } flowDto.setLabels(mergedLabels); - // Merge processors - final Set<ProcessorEntity> mergedProcessors = new HashSet<>(); - for (final List<ProcessorEntity> processorList : processors.values()) { - mergedProcessors.add(mergeProcessors(processorList)); - } - flowDto.setProcessors(mergedProcessors); - + ProcessorsEntityMerger.mergeProcessors(clientProcessors, processors); // Merge Remote Process Groups - final Set<RemoteProcessGroupEntity> mergedRpgs = new HashSet<>(); - for (final List<RemoteProcessGroupEntity> rpgList : rpgs.values()) { - mergedRpgs.add(mergeRemoteProcessGroups(rpgList)); - } - flowDto.setRemoteProcessGroups(mergedRpgs); - + RemoteProcessGroupsEntityMerger.mergeRemoteProcessGroups(clientRemoteProcessGroups, rpgs); // Merge Process Groups - final Set<ProcessGroupEntity> mergedGroups = new HashSet<>(); - for (final List<ProcessGroupEntity> groupList : processGroups.values()) { - mergedGroups.add(mergeProcessGroups(groupList)); - } - flowDto.setProcessGroups(mergedGroups); - } - - private ConnectionEntity mergeConnections(final List<ConnectionEntity> connections) { - final ConnectionEntity merged = connections.get(0); - final ConnectionStatusDTO statusDto = merged.getStatus(); - statusDto.setNodeSnapshots(null); - - for (final ConnectionEntity entity : connections) { - if (entity != merged) { - StatusMerger.merge(merged.getStatus().getAggregateSnapshot(), entity.getStatus().getAggregateSnapshot()); - } - } - - return merged; - } - - private PortEntity mergePorts(final List<PortEntity> ports) { - final PortEntity merged = ports.get(0); - final PortStatusDTO statusDto = merged.getStatus(); - statusDto.setNodeSnapshots(null); - - for (final PortEntity entity : ports) { - if (entity != merged) { - StatusMerger.merge(merged.getStatus().getAggregateSnapshot(), entity.getStatus().getAggregateSnapshot()); - } - } - - return merged; + ProcessGroupsEntityMerger.mergeProcessGroups(clientProcessGroups, processGroups); } private FunnelEntity mergeFunnels(final List<FunnelEntity> funnels) { @@ -215,53 +167,4 @@ public class FlowMerger extends AbstractSingleDTOEndpoint<ProcessGroupFlowEntity private LabelEntity mergeLabels(final List<LabelEntity> labels) { return labels.get(0); } - - private ProcessorEntity mergeProcessors(final List<ProcessorEntity> processors) { - final ProcessorEntity merged = processors.get(0); - final ProcessorStatusDTO statusDto = merged.getStatus(); - statusDto.setNodeSnapshots(null); - - for (final ProcessorEntity entity : processors) { - if (entity != merged) { - StatusMerger.merge(merged.getStatus().getAggregateSnapshot(), entity.getStatus().getAggregateSnapshot()); - } - } - - return merged; - } - - - private RemoteProcessGroupEntity mergeRemoteProcessGroups(final List<RemoteProcessGroupEntity> rpgs) { - final RemoteProcessGroupEntity merged = rpgs.get(0); - final RemoteProcessGroupStatusDTO statusDto = merged.getStatus(); - statusDto.setNodeSnapshots(null); - - for (final RemoteProcessGroupEntity entity : rpgs) { - if (entity != merged) { - StatusMerger.merge(merged.getStatus().getAggregateSnapshot(), entity.getStatus().getAggregateSnapshot()); - } - } - - return merged; - } - - private ProcessGroupEntity mergeProcessGroups(final List<ProcessGroupEntity> groups) { - final ProcessGroupEntity merged = groups.get(0); - final ProcessGroupStatusDTO statusDto = merged.getStatus(); - statusDto.setNodeSnapshots(null); - - for (final ProcessGroupEntity entity : groups) { - if (entity != merged) { - StatusMerger.merge(merged.getStatus().getAggregateSnapshot(), entity.getStatus().getAggregateSnapshot()); - } - } - - // We merge only the statuses of the Process Groups. The child components are not - // necessary for a FlowProcessGroupDTO, so we just ensure that they are null - if (merged.getComponent() != null) { - merged.getComponent().setContents(null); - } - - return merged; - } } http://git-wip-us.apache.org/repos/asf/nifi/blob/0b437e09/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/InputPortsEndpointMerger.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/InputPortsEndpointMerger.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/InputPortsEndpointMerger.java new file mode 100644 index 0000000..b206508 --- /dev/null +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/InputPortsEndpointMerger.java @@ -0,0 +1,72 @@ +/* + * 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.cluster.coordination.http.endpoints; + +import org.apache.nifi.cluster.coordination.http.EndpointResponseMerger; +import org.apache.nifi.cluster.manager.NodeResponse; +import org.apache.nifi.cluster.manager.PortsEntityMerger; +import org.apache.nifi.cluster.protocol.NodeIdentifier; +import org.apache.nifi.web.api.entity.InputPortsEntity; +import org.apache.nifi.web.api.entity.PortEntity; + +import java.net.URI; +import java.util.HashMap; +import java.util.Map; +import java.util.Set; +import java.util.regex.Pattern; + +public class InputPortsEndpointMerger implements EndpointResponseMerger { + public static final Pattern INPUT_PORTS_URI_PATTERN = Pattern.compile("/nifi-api/process-groups/(?:(?:root)|(?:[a-f0-9\\-]{36}))/input-ports"); + + @Override + public boolean canHandle(final URI uri, final String method) { + return "GET".equalsIgnoreCase(method) && INPUT_PORTS_URI_PATTERN.matcher(uri.getPath()).matches(); + } + + @Override + public final NodeResponse merge(final URI uri, final String method, final Set<NodeResponse> successfulResponses, final Set<NodeResponse> problematicResponses, final NodeResponse clientResponse) { + if (!canHandle(uri, method)) { + throw new IllegalArgumentException("Cannot use Endpoint Mapper of type " + getClass().getSimpleName() + " to map responses for URI " + uri + ", HTTP Method " + method); + } + + final InputPortsEntity responseEntity = clientResponse.getClientResponse().getEntity(InputPortsEntity.class); + final Set<PortEntity> portEntities = responseEntity.getInputPorts(); + + final Map<String, Map<NodeIdentifier, PortEntity>> entityMap = new HashMap<>(); + for (final NodeResponse nodeResponse : successfulResponses) { + final InputPortsEntity nodeResponseEntity = nodeResponse == clientResponse ? responseEntity : nodeResponse.getClientResponse().getEntity(InputPortsEntity.class); + final Set<PortEntity> nodePortEntities = nodeResponseEntity.getInputPorts(); + + for (final PortEntity nodePortEntity : nodePortEntities) { + final NodeIdentifier nodeId = nodeResponse.getNodeId(); + Map<NodeIdentifier, PortEntity> innerMap = entityMap.get(nodeId); + if (innerMap == null) { + innerMap = new HashMap<>(); + entityMap.put(nodePortEntity.getId(), innerMap); + } + + innerMap.put(nodeResponse.getNodeId(), nodePortEntity); + } + } + + PortsEntityMerger.mergePorts(portEntities, entityMap); + + // create a new client response + return new NodeResponse(clientResponse, responseEntity); + } +} http://git-wip-us.apache.org/repos/asf/nifi/blob/0b437e09/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/OutputPortsEndpointMerger.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/OutputPortsEndpointMerger.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/OutputPortsEndpointMerger.java new file mode 100644 index 0000000..753c406 --- /dev/null +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/OutputPortsEndpointMerger.java @@ -0,0 +1,72 @@ +/* + * 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.cluster.coordination.http.endpoints; + +import org.apache.nifi.cluster.coordination.http.EndpointResponseMerger; +import org.apache.nifi.cluster.manager.NodeResponse; +import org.apache.nifi.cluster.manager.PortsEntityMerger; +import org.apache.nifi.cluster.protocol.NodeIdentifier; +import org.apache.nifi.web.api.entity.OutputPortsEntity; +import org.apache.nifi.web.api.entity.PortEntity; + +import java.net.URI; +import java.util.HashMap; +import java.util.Map; +import java.util.Set; +import java.util.regex.Pattern; + +public class OutputPortsEndpointMerger implements EndpointResponseMerger { + public static final Pattern OUTPUT_PORTS_URI_PATTERN = Pattern.compile("/nifi-api/process-groups/(?:(?:root)|(?:[a-f0-9\\-]{36}))/output-ports"); + + @Override + public boolean canHandle(final URI uri, final String method) { + return "GET".equalsIgnoreCase(method) && OUTPUT_PORTS_URI_PATTERN.matcher(uri.getPath()).matches(); + } + + @Override + public final NodeResponse merge(final URI uri, final String method, final Set<NodeResponse> successfulResponses, final Set<NodeResponse> problematicResponses, final NodeResponse clientResponse) { + if (!canHandle(uri, method)) { + throw new IllegalArgumentException("Cannot use Endpoint Mapper of type " + getClass().getSimpleName() + " to map responses for URI " + uri + ", HTTP Method " + method); + } + + final OutputPortsEntity responseEntity = clientResponse.getClientResponse().getEntity(OutputPortsEntity.class); + final Set<PortEntity> portEntities = responseEntity.getOutputPorts(); + + final Map<String, Map<NodeIdentifier, PortEntity>> entityMap = new HashMap<>(); + for (final NodeResponse nodeResponse : successfulResponses) { + final OutputPortsEntity nodeResponseEntity = nodeResponse == clientResponse ? responseEntity : nodeResponse.getClientResponse().getEntity(OutputPortsEntity.class); + final Set<PortEntity> nodePortEntities = nodeResponseEntity.getOutputPorts(); + + for (final PortEntity nodePortEntity : nodePortEntities) { + final NodeIdentifier nodeId = nodeResponse.getNodeId(); + Map<NodeIdentifier, PortEntity> innerMap = entityMap.get(nodeId); + if (innerMap == null) { + innerMap = new HashMap<>(); + entityMap.put(nodePortEntity.getId(), innerMap); + } + + innerMap.put(nodeResponse.getNodeId(), nodePortEntity); + } + } + + PortsEntityMerger.mergePorts(portEntities, entityMap); + + // create a new client response + return new NodeResponse(clientResponse, responseEntity); + } +} http://git-wip-us.apache.org/repos/asf/nifi/blob/0b437e09/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/PortEndpointMerger.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/PortEndpointMerger.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/PortEndpointMerger.java new file mode 100644 index 0000000..8ea6384 --- /dev/null +++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/PortEndpointMerger.java @@ -0,0 +1,75 @@ +/* + * 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.cluster.coordination.http.endpoints; + +import org.apache.nifi.cluster.coordination.http.EndpointResponseMerger; +import org.apache.nifi.cluster.manager.NodeResponse; +import org.apache.nifi.cluster.manager.PortEntityMerger; +import org.apache.nifi.cluster.protocol.NodeIdentifier; +import org.apache.nifi.web.api.entity.PortEntity; + +import java.net.URI; +import java.util.Map; +import java.util.Set; +import java.util.regex.Pattern; + +public class PortEndpointMerger extends AbstractSingleEntityEndpoint<PortEntity> implements EndpointResponseMerger { + public static final Pattern INPUT_PORTS_URI_PATTERN = Pattern.compile("/nifi-api/process-groups/(?:(?:root)|(?:[a-f0-9\\-]{36}))/input-ports"); + public static final Pattern INPUT_PORT_URI_PATTERN = Pattern.compile("/nifi-api/input-ports/[a-f0-9\\-]{36}"); + + public static final Pattern OUTPUT_PORTS_URI_PATTERN = Pattern.compile("/nifi-api/process-groups/(?:(?:root)|(?:[a-f0-9\\-]{36}))/output-ports"); + public static final Pattern OUTPUT_PORT_URI_PATTERN = Pattern.compile("/nifi-api/output-ports/[a-f0-9\\-]{36}"); + + @Override + public boolean canHandle(final URI uri, final String method) { + return canHandleInputPort(uri, method) || canHandleOutputPort(uri, method); + } + + private boolean canHandleInputPort(final URI uri, final String method) { + if (("GET".equalsIgnoreCase(method) || "PUT".equalsIgnoreCase(method)) && (INPUT_PORT_URI_PATTERN.matcher(uri.getPath()).matches())) { + return true; + } else if ("POST".equalsIgnoreCase(method) && INPUT_PORTS_URI_PATTERN.matcher(uri.getPath()).matches()) { + return true; + } + + return false; + } + + private boolean canHandleOutputPort(final URI uri, final String method) { + if (("GET".equalsIgnoreCase(method) || "PUT".equalsIgnoreCase(method)) && (OUTPUT_PORT_URI_PATTERN.matcher(uri.getPath()).matches())) { + return true; + } else if ("POST".equalsIgnoreCase(method) && OUTPUT_PORTS_URI_PATTERN.matcher(uri.getPath()).matches()) { + return true; + } + + return false; + } + + @Override + protected Class<PortEntity> getEntityClass() { + return PortEntity.class; + } + + + @Override + protected void mergeResponses(final PortEntity clientEntity, final Map<NodeIdentifier, PortEntity> entityMap, + final Set<NodeResponse> successfulResponses, final Set<NodeResponse> problematicResponses) { + + PortEntityMerger.mergePorts(clientEntity, entityMap); + } +}
