You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by ma...@apache.org on 2016/03/03 19:29:53 UTC
[1/5] nifi git commit: NIFI-1563: Federate requests and merge
responses from nodes instead of storing bulletins and stats at NCM
Repository: nifi
Updated Branches:
refs/heads/NIFI-1563 [created] ad32cb824
http://git-wip-us.apache.org/repos/asf/nifi/blob/ad32cb82/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ProcessGroupResource.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ProcessGroupResource.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ProcessGroupResource.java
index a67e74b..b5fadcc 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ProcessGroupResource.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ProcessGroupResource.java
@@ -16,19 +16,13 @@
*/
package org.apache.nifi.web.api;
-import com.sun.jersey.api.core.ResourceContext;
-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 java.net.URI;
import java.net.URISyntaxException;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
+
import javax.servlet.http.HttpServletRequest;
import javax.ws.rs.Consumes;
import javax.ws.rs.DELETE;
@@ -46,13 +40,13 @@ import javax.ws.rs.WebApplicationException;
import javax.ws.rs.core.Context;
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.Response;
+
+import org.apache.commons.lang3.StringUtils;
import org.apache.nifi.cluster.manager.impl.WebClusterManager;
import org.apache.nifi.util.NiFiProperties;
import org.apache.nifi.web.ConfigurationSnapshot;
-import org.apache.nifi.web.IllegalClusterResourceRequestException;
import org.apache.nifi.web.NiFiServiceFacade;
import org.apache.nifi.web.Revision;
-import static org.apache.nifi.web.api.ApplicationResource.CLIENT_ID;
import org.apache.nifi.web.api.dto.FlowSnippetDTO;
import org.apache.nifi.web.api.dto.PositionDTO;
import org.apache.nifi.web.api.dto.ProcessGroupDTO;
@@ -67,9 +61,16 @@ import org.apache.nifi.web.api.entity.StatusHistoryEntity;
import org.apache.nifi.web.api.request.ClientIdParameter;
import org.apache.nifi.web.api.request.DoubleParameter;
import org.apache.nifi.web.api.request.LongParameter;
-import org.apache.commons.lang3.StringUtils;
import org.springframework.security.access.prepost.PreAuthorize;
+import com.sun.jersey.api.core.ResourceContext;
+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;
+
/**
* RESTful endpoint for managing a Group.
*/
@@ -979,7 +980,7 @@ public class ProcessGroupResource extends ApplicationResource {
headersToOverride.put("content-type", MediaType.APPLICATION_JSON);
// replicate put request
- return (Response) clusterManager.applyRequest(HttpMethod.PUT, putUri, updateClientId(processGroupEntity), getHeaders(headersToOverride)).getResponse();
+ return clusterManager.applyRequest(HttpMethod.PUT, putUri, updateClientId(processGroupEntity), getHeaders(headersToOverride)).getResponse();
}
// handle expects request (usually from the cluster manager)
@@ -1284,6 +1285,10 @@ public class ProcessGroupResource extends ApplicationResource {
@QueryParam(CLIENT_ID) @DefaultValue(StringUtils.EMPTY) ClientIdParameter clientId,
@QueryParam("recursive") @DefaultValue(RECURSIVE) Boolean recursive) {
+ if (properties.isClusterManager()) {
+ return clusterManager.applyRequest(HttpMethod.GET, getAbsolutePath(), getRequestParameters(true), getHeaders()).getResponse();
+ }
+
// get the status
final ProcessGroupStatusDTO statusReport = serviceFacade.getProcessGroupStatus(groupId);
@@ -1345,7 +1350,7 @@ public class ProcessGroupResource extends ApplicationResource {
// replicate if cluster manager
if (properties.isClusterManager()) {
- throw new IllegalClusterResourceRequestException("This request is only supported in standalone mode.");
+ return clusterManager.applyRequest(HttpMethod.GET, getAbsolutePath(), getRequestParameters(true), getHeaders()).getResponse();
}
// get the specified processor status history
http://git-wip-us.apache.org/repos/asf/nifi/blob/ad32cb82/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ProcessorResource.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ProcessorResource.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ProcessorResource.java
index 9fb90b5..0e4c979 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ProcessorResource.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ProcessorResource.java
@@ -16,12 +16,6 @@
*/
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 java.net.URI;
import java.net.URISyntaxException;
import java.util.Arrays;
@@ -52,50 +46,49 @@ import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.MultivaluedMap;
import javax.ws.rs.core.Response;
+import org.apache.commons.lang3.StringUtils;
import org.apache.nifi.cluster.manager.exception.IllegalClusterStateException;
import org.apache.nifi.cluster.manager.impl.WebClusterManager;
import org.apache.nifi.cluster.node.Node;
import org.apache.nifi.scheduling.SchedulingStrategy;
+import org.apache.nifi.ui.extension.UiExtension;
+import org.apache.nifi.ui.extension.UiExtensionMapping;
import org.apache.nifi.util.NiFiProperties;
import org.apache.nifi.web.ConfigurationSnapshot;
-import org.apache.nifi.web.IllegalClusterResourceRequestException;
import org.apache.nifi.web.NiFiServiceFacade;
import org.apache.nifi.web.Revision;
-import static org.apache.nifi.web.api.ApplicationResource.CLIENT_ID;
-
+import org.apache.nifi.web.UiExtensionType;
import org.apache.nifi.web.api.dto.ComponentStateDTO;
import org.apache.nifi.web.api.dto.PositionDTO;
import org.apache.nifi.web.api.dto.ProcessorConfigDTO;
import org.apache.nifi.web.api.dto.ProcessorDTO;
+import org.apache.nifi.web.api.dto.PropertyDescriptorDTO;
import org.apache.nifi.web.api.dto.RevisionDTO;
import org.apache.nifi.web.api.dto.status.StatusHistoryDTO;
import org.apache.nifi.web.api.entity.ComponentStateEntity;
import org.apache.nifi.web.api.entity.ProcessorEntity;
import org.apache.nifi.web.api.entity.ProcessorsEntity;
+import org.apache.nifi.web.api.entity.PropertyDescriptorEntity;
import org.apache.nifi.web.api.entity.StatusHistoryEntity;
import org.apache.nifi.web.api.request.ClientIdParameter;
import org.apache.nifi.web.api.request.DoubleParameter;
import org.apache.nifi.web.api.request.IntegerParameter;
import org.apache.nifi.web.api.request.LongParameter;
-
-import org.apache.commons.lang3.StringUtils;
-import org.apache.nifi.ui.extension.UiExtension;
-import org.apache.nifi.ui.extension.UiExtensionMapping;
-import org.apache.nifi.web.UiExtensionType;
-import org.apache.nifi.web.api.dto.PropertyDescriptorDTO;
-import org.apache.nifi.web.api.entity.PropertyDescriptorEntity;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import org.springframework.security.access.prepost.PreAuthorize;
+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;
+
/**
* RESTful endpoint for managing a Processor.
*/
@Api(hidden = true)
public class ProcessorResource extends ApplicationResource {
- private static final Logger logger = LoggerFactory.getLogger(ProcessorResource.class);
-
private static final List<Long> POSSIBLE_RUN_DURATIONS = Arrays.asList(0L, 25L, 50L, 100L, 250L, 500L, 1000L, 2000L);
private NiFiServiceFacade serviceFacade;
@@ -328,7 +321,7 @@ public class ProcessorResource extends ApplicationResource {
headersToOverride.put("content-type", MediaType.APPLICATION_JSON);
// replicate put request
- return (Response) clusterManager.applyRequest(HttpMethod.PUT, putUri, updateClientId(processorEntity), getHeaders(headersToOverride)).getResponse();
+ return clusterManager.applyRequest(HttpMethod.PUT, putUri, updateClientId(processorEntity), getHeaders(headersToOverride)).getResponse();
}
@@ -467,7 +460,7 @@ public class ProcessorResource extends ApplicationResource {
// replicate if cluster manager
if (properties.isClusterManager()) {
- throw new IllegalClusterResourceRequestException("This request is only supported in standalone mode.");
+ return clusterManager.applyRequest(HttpMethod.GET, getAbsolutePath(), getRequestParameters(true), getHeaders()).getResponse();
}
// get the specified processor status history
http://git-wip-us.apache.org/repos/asf/nifi/blob/ad32cb82/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/RemoteProcessGroupResource.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/RemoteProcessGroupResource.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/RemoteProcessGroupResource.java
index e466666..f1e10f7 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/RemoteProcessGroupResource.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/RemoteProcessGroupResource.java
@@ -16,18 +16,13 @@
*/
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 java.net.URI;
import java.net.URISyntaxException;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
+
import javax.servlet.http.HttpServletRequest;
import javax.ws.rs.Consumes;
import javax.ws.rs.DELETE;
@@ -45,19 +40,19 @@ import javax.ws.rs.WebApplicationException;
import javax.ws.rs.core.Context;
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.Response;
+
+import org.apache.commons.lang3.StringUtils;
import org.apache.nifi.cluster.manager.impl.WebClusterManager;
import org.apache.nifi.util.NiFiProperties;
import org.apache.nifi.web.ConfigurationSnapshot;
-import org.apache.nifi.web.IllegalClusterResourceRequestException;
import org.apache.nifi.web.NiFiServiceFacade;
import org.apache.nifi.web.Revision;
-import static org.apache.nifi.web.api.ApplicationResource.CLIENT_ID;
-import static org.apache.nifi.web.api.ApplicationResource.VERSION;
import org.apache.nifi.web.api.dto.PositionDTO;
import org.apache.nifi.web.api.dto.RemoteProcessGroupDTO;
import org.apache.nifi.web.api.dto.RemoteProcessGroupPortDTO;
import org.apache.nifi.web.api.dto.RevisionDTO;
import org.apache.nifi.web.api.dto.status.StatusHistoryDTO;
+import org.apache.nifi.web.api.entity.ConnectionsEntity;
import org.apache.nifi.web.api.entity.RemoteProcessGroupEntity;
import org.apache.nifi.web.api.entity.RemoteProcessGroupPortEntity;
import org.apache.nifi.web.api.entity.RemoteProcessGroupsEntity;
@@ -66,20 +61,21 @@ import org.apache.nifi.web.api.request.ClientIdParameter;
import org.apache.nifi.web.api.request.DoubleParameter;
import org.apache.nifi.web.api.request.IntegerParameter;
import org.apache.nifi.web.api.request.LongParameter;
-import org.apache.commons.lang3.StringUtils;
-import org.apache.nifi.web.api.entity.ConnectionsEntity;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import org.springframework.security.access.prepost.PreAuthorize;
+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;
+
/**
* RESTful endpoint for managing a Remote group.
*/
@Api(hidden = true)
public class RemoteProcessGroupResource extends ApplicationResource {
- private static final Logger logger = LoggerFactory.getLogger(RemoteProcessGroupResource.class);
-
private static final String VERBOSE_DEFAULT_VALUE = "false";
private NiFiServiceFacade serviceFacade;
@@ -300,7 +296,7 @@ public class RemoteProcessGroupResource extends ApplicationResource {
// replicate if cluster manager
if (properties.isClusterManager()) {
- throw new IllegalClusterResourceRequestException("This request is only supported in standalone mode.");
+ return clusterManager.applyRequest(HttpMethod.GET, getAbsolutePath(), getRequestParameters(true), getHeaders()).getResponse();
}
// get the specified processor status history
@@ -446,7 +442,7 @@ public class RemoteProcessGroupResource extends ApplicationResource {
headersToOverride.put("content-type", MediaType.APPLICATION_JSON);
// replicate put request
- return (Response) clusterManager.applyRequest(HttpMethod.PUT, putUri, updateClientId(remoteProcessGroupEntity), getHeaders(headersToOverride)).getResponse();
+ return clusterManager.applyRequest(HttpMethod.PUT, putUri, updateClientId(remoteProcessGroupEntity), getHeaders(headersToOverride)).getResponse();
}
// handle expects request (usually from the cluster manager)
http://git-wip-us.apache.org/repos/asf/nifi/blob/ad32cb82/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ReportingTaskResource.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ReportingTaskResource.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ReportingTaskResource.java
index d7b77b2..802f46f 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ReportingTaskResource.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ReportingTaskResource.java
@@ -16,12 +16,6 @@
*/
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 java.net.URI;
import java.net.URISyntaxException;
import java.util.HashMap;
@@ -30,6 +24,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
+
import javax.servlet.ServletContext;
import javax.servlet.http.HttpServletRequest;
import javax.ws.rs.Consumes;
@@ -49,39 +44,42 @@ import javax.ws.rs.core.Context;
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.MultivaluedMap;
import javax.ws.rs.core.Response;
+
+import org.apache.commons.lang3.StringUtils;
import org.apache.nifi.cluster.manager.impl.WebClusterManager;
+import org.apache.nifi.ui.extension.UiExtension;
+import org.apache.nifi.ui.extension.UiExtensionMapping;
import org.apache.nifi.util.NiFiProperties;
import org.apache.nifi.web.ConfigurationSnapshot;
import org.apache.nifi.web.NiFiServiceFacade;
import org.apache.nifi.web.Revision;
-import org.apache.nifi.web.api.dto.ComponentStateDTO;
-import org.apache.nifi.web.api.dto.RevisionDTO;
-import org.apache.nifi.web.api.entity.ComponentStateEntity;
-import org.apache.nifi.web.api.request.ClientIdParameter;
-import org.apache.nifi.web.api.request.LongParameter;
-import org.apache.commons.lang3.StringUtils;
-import org.apache.nifi.ui.extension.UiExtension;
-import org.apache.nifi.ui.extension.UiExtensionMapping;
import org.apache.nifi.web.UiExtensionType;
-import static org.apache.nifi.web.api.ApplicationResource.CLIENT_ID;
+import org.apache.nifi.web.api.dto.ComponentStateDTO;
import org.apache.nifi.web.api.dto.PropertyDescriptorDTO;
import org.apache.nifi.web.api.dto.ReportingTaskDTO;
+import org.apache.nifi.web.api.dto.RevisionDTO;
+import org.apache.nifi.web.api.entity.ComponentStateEntity;
import org.apache.nifi.web.api.entity.PropertyDescriptorEntity;
import org.apache.nifi.web.api.entity.ReportingTaskEntity;
import org.apache.nifi.web.api.entity.ReportingTasksEntity;
+import org.apache.nifi.web.api.request.ClientIdParameter;
+import org.apache.nifi.web.api.request.LongParameter;
import org.apache.nifi.web.util.Availability;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import org.springframework.security.access.prepost.PreAuthorize;
+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;
+
/**
* RESTful endpoint for managing a Reporting Task.
*/
@Api(hidden = true)
public class ReportingTaskResource extends ApplicationResource {
- private static final Logger logger = LoggerFactory.getLogger(ReportingTaskResource.class);
-
private NiFiServiceFacade serviceFacade;
private WebClusterManager clusterManager;
private NiFiProperties properties;
@@ -344,7 +342,7 @@ public class ReportingTaskResource extends ApplicationResource {
headersToOverride.put("content-type", MediaType.APPLICATION_JSON);
// replicate put request
- return (Response) clusterManager.applyRequest(HttpMethod.PUT, putUri, updateClientId(reportingTaskEntity), getHeaders(headersToOverride)).getResponse();
+ return clusterManager.applyRequest(HttpMethod.PUT, putUri, updateClientId(reportingTaskEntity), getHeaders(headersToOverride)).getResponse();
}
// handle expects request (usually from the cluster manager)
http://git-wip-us.apache.org/repos/asf/nifi/blob/ad32cb82/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/dto/DtoFactory.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/dto/DtoFactory.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/dto/DtoFactory.java
index 652fd04..1b4e920 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/dto/DtoFactory.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/dto/DtoFactory.java
@@ -127,6 +127,7 @@ 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.dto.status.StatusDTO;
+import org.apache.nifi.web.api.dto.status.StatusMerger;
import javax.ws.rs.WebApplicationException;
import java.text.Collator;
@@ -322,7 +323,7 @@ public final class DtoFactory {
final StateMapDTO dto = new StateMapDTO();
dto.setScope(scope.toString());
- final TreeMap<String, String> sortedState = new TreeMap(SortedStateUtils.getKeyComparator());
+ final TreeMap<String, String> sortedState = new TreeMap<>(SortedStateUtils.getKeyComparator());
final Map<String, String> state = stateMap.toMap();
sortedState.putAll(state);
@@ -708,13 +709,6 @@ public final class DtoFactory {
return copy;
}
- private String formatCount(final Integer intStatus) {
- return intStatus == null ? "-" : FormatUtils.formatCount(intStatus);
- }
-
- private String formatDataSize(final Long longStatus) {
- return longStatus == null ? "-" : FormatUtils.formatDataSize(longStatus);
- }
public RemoteProcessGroupStatusDTO createRemoteProcessGroupStatusDto(final RemoteProcessGroupStatus remoteProcessGroupStatus) {
final RemoteProcessGroupStatusDTO dto = new RemoteProcessGroupStatusDTO();
@@ -724,10 +718,13 @@ public final class DtoFactory {
dto.setName(remoteProcessGroupStatus.getName());
dto.setTransmissionStatus(remoteProcessGroupStatus.getTransmissionStatus().toString());
dto.setActiveThreadCount(remoteProcessGroupStatus.getActiveThreadCount());
- dto.setSent(formatCount(remoteProcessGroupStatus.getSentCount()) + " / " + formatDataSize(remoteProcessGroupStatus.getSentContentSize()));
- dto.setReceived(formatCount(remoteProcessGroupStatus.getReceivedCount()) + " / " + formatDataSize(remoteProcessGroupStatus.getReceivedContentSize()));
+ dto.setFlowFilesSent(remoteProcessGroupStatus.getSentCount());
+ dto.setBytesSent(remoteProcessGroupStatus.getSentContentSize());
+ dto.setFlowFilesReceived(remoteProcessGroupStatus.getReceivedCount());
+ dto.setBytesReceived(remoteProcessGroupStatus.getReceivedContentSize());
dto.setAuthorizationIssues(remoteProcessGroupStatus.getAuthorizationIssues());
+ StatusMerger.updatePrettyPrintedFields(dto);
return dto;
}
@@ -737,20 +734,20 @@ public final class DtoFactory {
processGroupStatusDto.setId(processGroupStatus.getId());
processGroupStatusDto.setName(processGroupStatus.getName());
processGroupStatusDto.setStatsLastRefreshed(new Date(processGroupStatus.getCreationTimestamp()));
- processGroupStatusDto.setRead(formatDataSize(processGroupStatus.getBytesRead()));
- processGroupStatusDto.setWritten(formatDataSize(processGroupStatus.getBytesWritten()));
- processGroupStatusDto.setInput(formatCount(processGroupStatus.getInputCount()) + " / " + formatDataSize(processGroupStatus.getInputContentSize()));
- processGroupStatusDto.setOutput(formatCount(processGroupStatus.getOutputCount()) + " / " + formatDataSize(processGroupStatus.getOutputContentSize()));
- processGroupStatusDto.setTransferred(formatCount(processGroupStatus.getFlowFilesTransferred()) + " / " + formatDataSize(processGroupStatus.getBytesTransferred()));
- processGroupStatusDto.setSent(formatCount(processGroupStatus.getFlowFilesSent()) + " / " + formatDataSize(processGroupStatus.getBytesSent()));
- processGroupStatusDto.setReceived(formatCount(processGroupStatus.getFlowFilesReceived()) + " / " + formatDataSize(processGroupStatus.getBytesReceived()));
+ processGroupStatusDto.setBytesRead(processGroupStatus.getBytesRead());
+ processGroupStatusDto.setBytesWritten(processGroupStatus.getBytesWritten());
+ processGroupStatusDto.setFlowFilesIn(processGroupStatus.getInputCount());
+ processGroupStatusDto.setBytesIn(processGroupStatus.getInputContentSize());
+ processGroupStatusDto.setFlowFilesOut(processGroupStatus.getOutputCount());
+ processGroupStatusDto.setBytesOut(processGroupStatus.getOutputContentSize());
+ processGroupStatusDto.setFlowFilesTransferred(processGroupStatus.getFlowFilesTransferred());
+ processGroupStatusDto.setBytesTransferred(processGroupStatus.getBytesTransferred());
+ processGroupStatusDto.setFlowFilesSent(processGroupStatus.getFlowFilesSent());
+ processGroupStatusDto.setBytesSent(processGroupStatus.getBytesSent());
+ processGroupStatusDto.setFlowFilesReceived(processGroupStatus.getFlowFilesReceived());
+ processGroupStatusDto.setBytesReceived(processGroupStatus.getBytesReceived());
processGroupStatusDto.setActiveThreadCount(processGroupStatus.getActiveThreadCount());
-
- final String queuedCount = FormatUtils.formatCount(processGroupStatus.getQueuedCount());
- final String queuedSize = FormatUtils.formatDataSize(processGroupStatus.getQueuedContentSize());
- processGroupStatusDto.setQueuedCount(queuedCount);
- processGroupStatusDto.setQueuedSize(queuedSize);
- processGroupStatusDto.setQueued(queuedCount + " / " + queuedSize);
+ StatusMerger.updatePrettyPrintedFields(processGroupStatusDto);
final Map<String, StatusDTO> componentStatusDtoMap = new HashMap<>();
@@ -896,7 +893,6 @@ public final class DtoFactory {
}
public ConnectionStatusDTO createConnectionStatusDto(final ConnectionStatus connectionStatus) {
-
final ConnectionStatusDTO connectionStatusDto = new ConnectionStatusDTO();
connectionStatusDto.setGroupId(connectionStatus.getGroupId());
connectionStatusDto.setId(connectionStatus.getId());
@@ -906,46 +902,36 @@ public final class DtoFactory {
connectionStatusDto.setDestinationId(connectionStatus.getDestinationId());
connectionStatusDto.setDestinationName(connectionStatus.getDestinationName());
- final String queuedCount = FormatUtils.formatCount(connectionStatus.getQueuedCount());
- final String queuedSize = FormatUtils.formatDataSize(connectionStatus.getQueuedBytes());
- connectionStatusDto.setQueuedCount(queuedCount);
- connectionStatusDto.setQueuedSize(queuedSize);
- connectionStatusDto.setQueued(queuedCount + " / " + queuedSize);
+ connectionStatusDto.setFlowFilesQueued(connectionStatus.getQueuedCount());
+ connectionStatusDto.setBytesQueued(connectionStatus.getQueuedBytes());
- final int inputCount = connectionStatus.getInputCount();
- final long inputBytes = connectionStatus.getInputBytes();
- connectionStatusDto.setInput(FormatUtils.formatCount(inputCount) + " / " + FormatUtils.formatDataSize(inputBytes));
+ connectionStatusDto.setFlowFilesIn(connectionStatus.getInputCount());
+ connectionStatusDto.setBytesIn(connectionStatus.getInputBytes());
- final int outputCount = connectionStatus.getOutputCount();
- final long outputBytes = connectionStatus.getOutputBytes();
- connectionStatusDto.setOutput(FormatUtils.formatCount(outputCount) + " / " + FormatUtils.formatDataSize(outputBytes));
+ connectionStatusDto.setFlowFilesOut(connectionStatus.getOutputCount());
+ connectionStatusDto.setBytesOut(connectionStatus.getOutputBytes());
+ StatusMerger.updatePrettyPrintedFields(connectionStatusDto);
return connectionStatusDto;
}
public ProcessorStatusDTO createProcessorStatusDto(final ProcessorStatus procStatus) {
-
final ProcessorStatusDTO dto = new ProcessorStatusDTO();
dto.setId(procStatus.getId());
dto.setGroupId(procStatus.getGroupId());
dto.setName(procStatus.getName());
- final int processedCount = procStatus.getOutputCount();
- final long numProcessedBytes = procStatus.getOutputBytes();
- dto.setOutput(FormatUtils.formatCount(processedCount) + " / " + FormatUtils.formatDataSize(numProcessedBytes));
-
- final int inputCount = procStatus.getInputCount();
- final long inputBytes = procStatus.getInputBytes();
- dto.setInput(FormatUtils.formatCount(inputCount) + " / " + FormatUtils.formatDataSize(inputBytes));
+ dto.setFlowFilesOut(procStatus.getOutputCount());
+ dto.setBytesOut(procStatus.getOutputBytes());
- final long readBytes = procStatus.getBytesRead();
- dto.setRead(FormatUtils.formatDataSize(readBytes));
+ dto.setFlowFilesIn(procStatus.getInputCount());
+ dto.setBytesIn(procStatus.getInputBytes());
- final long writtenBytes = procStatus.getBytesWritten();
- dto.setWritten(FormatUtils.formatDataSize(writtenBytes));
+ dto.setBytesRead(procStatus.getBytesRead());
+ dto.setBytesWritten(procStatus.getBytesWritten());
- dto.setTasksDuration(FormatUtils.formatHoursMinutesSeconds(procStatus.getProcessingNanos(), TimeUnit.NANOSECONDS));
- dto.setTasks(FormatUtils.formatCount(procStatus.getInvocations()));
+ dto.setTaskCount(procStatus.getInvocations());
+ dto.setTaskDuration(procStatus.getProcessingNanos());
// determine the run status
dto.setRunStatus(procStatus.getRunStatus().toString());
@@ -953,6 +939,7 @@ public final class DtoFactory {
dto.setActiveThreadCount(procStatus.getActiveThreadCount());
dto.setType(procStatus.getType());
+ StatusMerger.updatePrettyPrintedFields(dto);
return dto;
}
@@ -971,13 +958,12 @@ public final class DtoFactory {
dto.setRunStatus(portStatus.getRunStatus().toString());
dto.setTransmitting(portStatus.isTransmitting());
- final int processedCount = portStatus.getOutputCount();
- final long numProcessedBytes = portStatus.getOutputBytes();
- dto.setOutput(FormatUtils.formatCount(processedCount) + " / " + FormatUtils.formatDataSize(numProcessedBytes));
+ dto.setFlowFilesOut(portStatus.getOutputCount());
+ dto.setBytesOut(portStatus.getOutputBytes());
- final int inputCount = portStatus.getInputCount();
- final long inputBytes = portStatus.getInputBytes();
- dto.setInput(FormatUtils.formatCount(inputCount) + " / " + FormatUtils.formatDataSize(inputBytes));
+ dto.setFlowFilesIn(portStatus.getInputCount());
+ dto.setBytesIn(portStatus.getInputBytes());
+ StatusMerger.updatePrettyPrintedFields(dto);
return dto;
}
@@ -1765,6 +1751,7 @@ public final class DtoFactory {
* @param node node
* @return dto
*/
+ @SuppressWarnings("deprecation")
public ProvenanceNodeDTO createProvenanceEventNodeDTO(final ProvenanceEventLineageNode node) {
final ProvenanceNodeDTO dto = new ProvenanceNodeDTO();
dto.setId(node.getIdentifier());
@@ -1785,6 +1772,7 @@ public final class DtoFactory {
* @param node node
* @return dto
*/
+ @SuppressWarnings("deprecation")
public ProvenanceNodeDTO createFlowFileNodeDTO(final LineageNode node) {
final ProvenanceNodeDTO dto = new ProvenanceNodeDTO();
dto.setId(node.getIdentifier());
http://git-wip-us.apache.org/repos/asf/nifi/blob/ad32cb82/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/resources/nifi-web-api-context.xml
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/resources/nifi-web-api-context.xml b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/resources/nifi-web-api-context.xml
index c9326b2..1a57e22 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/resources/nifi-web-api-context.xml
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/resources/nifi-web-api-context.xml
@@ -245,6 +245,7 @@
<bean id="clusterResource" class="org.apache.nifi.web.api.ClusterResource" scope="singleton">
<property name="serviceFacade" ref="serviceFacade"/>
<property name="properties" ref="nifiProperties"/>
+ <property name="clusterManager" ref="clusterManager"/>
</bean>
<bean id="nodeResource" class="org.apache.nifi.web.api.NodeResource" scope="singleton">
<property name="serviceFacade" ref="serviceFacade"/>
[4/5] nifi git commit: NIFI-1563: Federate requests and merge
responses from nodes instead of storing bulletins and stats at NCM
Posted by ma...@apache.org.
http://git-wip-us.apache.org/repos/asf/nifi/blob/ad32cb82/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/status/StatusMerger.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/status/StatusMerger.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/status/StatusMerger.java
new file mode 100644
index 0000000..2eb95f7
--- /dev/null
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/status/StatusMerger.java
@@ -0,0 +1,375 @@
+/*
+ * 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.dto.status;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.nifi.controller.status.RunStatus;
+import org.apache.nifi.controller.status.TransmissionStatus;
+import org.apache.nifi.util.FormatUtils;
+import org.apache.nifi.web.api.dto.BulletinDTO;
+
+public class StatusMerger {
+ public static void merge(final ControllerStatusDTO target, final ControllerStatusDTO toMerge) {
+ if (target == null || toMerge == null) {
+ return;
+ }
+
+ target.setActiveRemotePortCount(target.getActiveRemotePortCount() + toMerge.getActiveRemotePortCount());
+ target.setActiveThreadCount(target.getActiveThreadCount() + toMerge.getActiveThreadCount());
+ target.setBytesQueued(target.getBytesQueued() + toMerge.getBytesQueued());
+ target.setDisabledCount(target.getDisabledCount() + toMerge.getDisabledCount());
+ target.setFlowFilesQueued(target.getFlowFilesQueued() + toMerge.getFlowFilesQueued());
+ target.setInactiveRemotePortCount(target.getInactiveRemotePortCount() + toMerge.getInactiveRemotePortCount());
+ target.setInvalidCount(target.getInvalidCount() + toMerge.getInvalidCount());
+ target.setRunningCount(target.getRunningCount() + toMerge.getRunningCount());
+ target.setStoppedCount(target.getStoppedCount() + toMerge.getStoppedCount());
+
+ target.setBulletins(mergeBulletins(target.getBulletins(), toMerge.getBulletins()));
+ target.setControllerServiceBulletins(mergeBulletins(target.getControllerServiceBulletins(), toMerge.getControllerServiceBulletins()));
+ target.setReportingTaskBulletins(mergeBulletins(target.getReportingTaskBulletins(), toMerge.getReportingTaskBulletins()));
+
+ updatePrettyPrintedFields(target);
+ }
+
+ public static void updatePrettyPrintedFields(final ControllerStatusDTO target) {
+ target.setQueued(prettyPrint(target.getFlowFilesQueued(), target.getBytesQueued()));
+ target.setConnectedNodes(formatCount(target.getConnectedNodeCount()) + " / " + formatCount(target.getTotalNodeCount()));
+ }
+
+ public static void merge(final ProcessGroupStatusDTO target, final ProcessGroupStatusDTO toMerge) {
+ if (target == null || toMerge == null) {
+ return;
+ }
+
+ target.setBytesIn(target.getBytesIn() + toMerge.getBytesIn());
+ target.setFlowFilesIn(target.getFlowFilesIn() + toMerge.getFlowFilesIn());
+
+ target.setBytesQueued(target.getBytesQueued() + toMerge.getBytesQueued());
+ target.setFlowFilesQueued(target.getFlowFilesQueued() + toMerge.getFlowFilesQueued());
+
+ target.setBytesRead(target.getBytesRead() + toMerge.getBytesRead());
+ target.setBytesWritten(target.getBytesWritten() + toMerge.getBytesWritten());
+
+ target.setBytesOut(target.getBytesOut() + toMerge.getBytesOut());
+ target.setFlowFilesOut(target.getFlowFilesOut() + toMerge.getFlowFilesOut());
+
+ target.setBytesTransferred(target.getBytesTransferred() + toMerge.getBytesTransferred());
+ target.setFlowFilesTransferred(target.getFlowFilesTransferred() + toMerge.getFlowFilesTransferred());
+
+ target.setBytesReceived(target.getBytesReceived() + toMerge.getBytesReceived());
+ target.setFlowFilesReceived(target.getFlowFilesReceived() + toMerge.getFlowFilesReceived());
+
+ target.setBytesSent(target.getBytesSent() + toMerge.getBytesSent());
+ target.setFlowFilesSent(target.getFlowFilesSent() + toMerge.getFlowFilesSent());
+
+ target.setActiveThreadCount(target.getActiveThreadCount() + toMerge.getActiveThreadCount());
+ target.setBulletins(mergeBulletins(target.getBulletins(), toMerge.getBulletins()));
+ updatePrettyPrintedFields(target);
+
+ // connection status
+ // sort by id
+ final Map<String, ConnectionStatusDTO> mergedConnectionMap = new HashMap<>();
+ for (final ConnectionStatusDTO status : replaceNull(target.getConnectionStatus())) {
+ mergedConnectionMap.put(status.getId(), status);
+ }
+
+ for (final ConnectionStatusDTO statusToMerge : replaceNull(toMerge.getConnectionStatus())) {
+ ConnectionStatusDTO merged = mergedConnectionMap.get(statusToMerge.getId());
+ if (merged == null) {
+ mergedConnectionMap.put(statusToMerge.getId(), statusToMerge.clone());
+ continue;
+ }
+
+ merge(merged, statusToMerge);
+ }
+ target.setConnectionStatus(mergedConnectionMap.values());
+
+ // processor status
+ final Map<String, ProcessorStatusDTO> mergedProcessorMap = new HashMap<>();
+ for (final ProcessorStatusDTO status : replaceNull(target.getProcessorStatus())) {
+ mergedProcessorMap.put(status.getId(), status);
+ }
+
+ for (final ProcessorStatusDTO statusToMerge : replaceNull(toMerge.getProcessorStatus())) {
+ ProcessorStatusDTO merged = mergedProcessorMap.get(statusToMerge.getId());
+ if (merged == null) {
+ mergedProcessorMap.put(statusToMerge.getId(), statusToMerge.clone());
+ continue;
+ }
+
+ merge(merged, statusToMerge);
+ }
+ target.setProcessorStatus(mergedProcessorMap.values());
+
+ // input ports
+ final Map<String, PortStatusDTO> mergedInputPortMap = new HashMap<>();
+ for (final PortStatusDTO status : replaceNull(target.getInputPortStatus())) {
+ mergedInputPortMap.put(status.getId(), status);
+ }
+
+ for (final PortStatusDTO statusToMerge : replaceNull(toMerge.getInputPortStatus())) {
+ PortStatusDTO merged = mergedInputPortMap.get(statusToMerge.getId());
+ if (merged == null) {
+ mergedInputPortMap.put(statusToMerge.getId(), statusToMerge.clone());
+ continue;
+ }
+
+ merge(merged, statusToMerge);
+ }
+ target.setInputPortStatus(mergedInputPortMap.values());
+
+ // output ports
+ final Map<String, PortStatusDTO> mergedOutputPortMap = new HashMap<>();
+ for (final PortStatusDTO status : replaceNull(target.getOutputPortStatus())) {
+ mergedOutputPortMap.put(status.getId(), status);
+ }
+
+ for (final PortStatusDTO statusToMerge : replaceNull(toMerge.getOutputPortStatus())) {
+ PortStatusDTO merged = mergedOutputPortMap.get(statusToMerge.getId());
+ if (merged == null) {
+ mergedOutputPortMap.put(statusToMerge.getId(), statusToMerge.clone());
+ continue;
+ }
+
+ merge(merged, statusToMerge);
+ }
+ target.setOutputPortStatus(mergedOutputPortMap.values());
+
+ // child groups
+ final Map<String, ProcessGroupStatusDTO> mergedGroupMap = new HashMap<>();
+ for (final ProcessGroupStatusDTO status : replaceNull(target.getProcessGroupStatus())) {
+ mergedGroupMap.put(status.getId(), status);
+ }
+
+ for (final ProcessGroupStatusDTO statusToMerge : replaceNull(toMerge.getProcessGroupStatus())) {
+ ProcessGroupStatusDTO merged = mergedGroupMap.get(statusToMerge.getId());
+ if (merged == null) {
+ mergedGroupMap.put(statusToMerge.getId(), statusToMerge.clone());
+ continue;
+ }
+
+ merge(merged, statusToMerge);
+ }
+ target.setOutputPortStatus(mergedOutputPortMap.values());
+
+ // remote groups
+ final Map<String, RemoteProcessGroupStatusDTO> mergedRemoteGroupMap = new HashMap<>();
+ for (final RemoteProcessGroupStatusDTO status : replaceNull(target.getRemoteProcessGroupStatus())) {
+ mergedRemoteGroupMap.put(status.getId(), status);
+ }
+
+ for (final RemoteProcessGroupStatusDTO statusToMerge : replaceNull(toMerge.getRemoteProcessGroupStatus())) {
+ RemoteProcessGroupStatusDTO merged = mergedRemoteGroupMap.get(statusToMerge.getId());
+ if (merged == null) {
+ mergedRemoteGroupMap.put(statusToMerge.getId(), statusToMerge.clone());
+ continue;
+ }
+
+ merge(merged, statusToMerge);
+ }
+ target.setRemoteProcessGroupStatus(mergedRemoteGroupMap.values());
+ }
+
+ private static <T> Collection<T> replaceNull(final Collection<T> collection) {
+ return (collection == null) ? Collections.<T> emptyList() : collection;
+ }
+
+ public static List<BulletinDTO> mergeBulletins(final List<BulletinDTO> targetBulletins, final List<BulletinDTO> toMerge) {
+ final List<BulletinDTO> bulletins = new ArrayList<>();
+ if (targetBulletins != null) {
+ bulletins.addAll(targetBulletins);
+ }
+
+ if (toMerge != null) {
+ bulletins.addAll(toMerge);
+ }
+
+ return bulletins;
+ }
+
+ /**
+ * Updates the fields that are "pretty printed" based on the raw values currently set. For example,
+ * {@link ProcessGroupStatusDTO#setInput(String)} will be called with the pretty-printed form of the
+ * FlowFile counts and sizes retrieved via {@link ProcessGroupStatusDTO#getFlowFilesIn()} and
+ * {@link ProcessGroupStatusDTO#getBytesIn()}.
+ *
+ * This logic is performed here, rather than in the DTO itself because the DTO needs to be kept purely
+ * getters & setters - otherwise the automatic marshalling and unmarshalling to/from JSON becomes very
+ * complicated.
+ *
+ * @param target the DTO to update
+ */
+ public static void updatePrettyPrintedFields(final ProcessGroupStatusDTO target) {
+ target.setQueued(prettyPrint(target.getFlowFilesQueued(), target.getBytesQueued()));
+ target.setQueuedCount(formatCount(target.getFlowFilesQueued()));
+ target.setQueuedSize(formatDataSize(target.getBytesQueued()));
+ target.setInput(prettyPrint(target.getFlowFilesIn(), target.getBytesIn()));
+ target.setRead(formatDataSize(target.getBytesRead()));
+ target.setWritten(formatDataSize(target.getBytesWritten()));
+ target.setOutput(prettyPrint(target.getFlowFilesOut(), target.getBytesOut()));
+ target.setTransferred(prettyPrint(target.getFlowFilesTransferred(), target.getBytesTransferred()));
+ target.setReceived(prettyPrint(target.getFlowFilesReceived(), target.getBytesReceived()));
+ target.setSent(prettyPrint(target.getFlowFilesSent(), target.getBytesSent()));
+ }
+
+
+ public static void merge(final ProcessorStatusDTO target, final ProcessorStatusDTO toMerge) {
+ if (target == null || toMerge == null) {
+ return;
+ }
+
+ // if the status to merge is invalid allow it to take precedence. whether the
+ // processor run status is disabled/stopped/running is part of the flow configuration
+ // and should not differ amongst nodes. however, whether a processor is invalid
+ // can be driven by environmental conditions. this check allows any of those to
+ // take precedence over the configured run status.
+ if (RunStatus.Invalid.name().equals(toMerge.getRunStatus())) {
+ target.setRunStatus(RunStatus.Invalid.name());
+ }
+
+ target.setBytesRead(target.getBytesRead() + toMerge.getBytesRead());
+ target.setBytesWritten(target.getBytesWritten() + toMerge.getBytesWritten());
+ target.setFlowFilesIn(target.getFlowFilesIn() + toMerge.getFlowFilesIn());
+ target.setBytesIn(target.getBytesIn() + toMerge.getBytesIn());
+ target.setFlowFilesOut(target.getFlowFilesOut() + toMerge.getFlowFilesOut());
+ target.setBytesOut(target.getBytesOut() + toMerge.getBytesOut());
+ target.setTaskCount(target.getTaskCount() + toMerge.getTaskCount());
+ target.setTaskDuration(target.getTaskDuration() + toMerge.getTaskDuration());
+ target.setActiveThreadCount(target.getActiveThreadCount() + toMerge.getActiveThreadCount());
+ target.setBulletins(mergeBulletins(target.getBulletins(), toMerge.getBulletins()));
+ updatePrettyPrintedFields(target);
+ }
+
+ public static void updatePrettyPrintedFields(final ProcessorStatusDTO target) {
+ target.setInput(prettyPrint(target.getFlowFilesIn(), target.getBytesIn()));
+ target.setRead(formatDataSize(target.getBytesRead()));
+ target.setWritten(formatDataSize(target.getBytesWritten()));
+ target.setOutput(prettyPrint(target.getFlowFilesOut(), target.getBytesOut()));
+
+ final Integer taskCount = target.getTaskCount();
+ final String tasks = (taskCount == null) ? "-" : formatCount(taskCount);
+ target.setTasks(tasks);
+
+ target.setTasksDuration(FormatUtils.formatHoursMinutesSeconds(target.getTaskDuration(), TimeUnit.NANOSECONDS));
+ }
+
+
+ public static void merge(final ConnectionStatusDTO target, final ConnectionStatusDTO toMerge) {
+ if (target == null || toMerge == null) {
+ return;
+ }
+
+ target.setFlowFilesIn(target.getFlowFilesIn() + toMerge.getFlowFilesIn());
+ target.setBytesIn(target.getBytesIn() + toMerge.getBytesIn());
+ target.setFlowFilesOut(target.getFlowFilesOut() + toMerge.getFlowFilesOut());
+ target.setBytesOut(target.getBytesOut() + toMerge.getBytesOut());
+ target.setFlowFilesQueued(target.getFlowFilesQueued() + toMerge.getFlowFilesQueued());
+ target.setBytesQueued(target.getBytesQueued() + toMerge.getBytesQueued());
+ updatePrettyPrintedFields(target);
+ }
+
+ public static void updatePrettyPrintedFields(final ConnectionStatusDTO target) {
+ target.setQueued(prettyPrint(target.getFlowFilesQueued(), target.getBytesQueued()));
+ target.setQueuedCount(formatCount(target.getFlowFilesQueued()));
+ target.setQueuedSize(formatDataSize(target.getBytesQueued()));
+ target.setInput(prettyPrint(target.getFlowFilesIn(), target.getBytesIn()));
+ target.setOutput(prettyPrint(target.getFlowFilesOut(), target.getBytesOut()));
+ }
+
+
+
+ public static void merge(final RemoteProcessGroupStatusDTO target, final RemoteProcessGroupStatusDTO toMerge) {
+ final String transmittingValue = TransmissionStatus.Transmitting.name();
+ if (transmittingValue.equals(target.getTransmissionStatus()) || transmittingValue.equals(toMerge.getTransmissionStatus())) {
+ target.setTransmissionStatus(transmittingValue);
+ }
+
+ target.setActiveThreadCount(target.getActiveThreadCount() + toMerge.getActiveThreadCount());
+
+ final List<String> authIssues = new ArrayList<>();
+ if (target.getAuthorizationIssues() != null) {
+ authIssues.addAll(target.getAuthorizationIssues());
+ }
+ if (toMerge.getAuthorizationIssues() != null) {
+ authIssues.addAll(toMerge.getAuthorizationIssues());
+ }
+ target.setAuthorizationIssues(authIssues);
+
+ target.setFlowFilesSent(target.getFlowFilesSent() + toMerge.getFlowFilesSent());
+ target.setBytesSent(target.getBytesSent() + toMerge.getBytesSent());
+ target.setFlowFilesReceived(target.getFlowFilesReceived() + toMerge.getFlowFilesReceived());
+ target.setBytesReceived(target.getBytesReceived() + toMerge.getBytesReceived());
+ target.setBulletins(mergeBulletins(target.getBulletins(), toMerge.getBulletins()));
+ updatePrettyPrintedFields(target);
+ }
+
+ public static void updatePrettyPrintedFields(final RemoteProcessGroupStatusDTO target) {
+ target.setReceived(prettyPrint(target.getFlowFilesReceived(), target.getBytesReceived()));
+ target.setSent(prettyPrint(target.getFlowFilesSent(), target.getBytesSent()));
+ }
+
+
+
+ public static void merge(final PortStatusDTO target, final PortStatusDTO toMerge) {
+ if (target == null || toMerge == null) {
+ return;
+ }
+
+ target.setActiveThreadCount(target.getActiveThreadCount() + toMerge.getActiveThreadCount());
+ target.setFlowFilesIn(target.getFlowFilesIn() + toMerge.getFlowFilesIn());
+ target.setBytesIn(target.getBytesIn() + toMerge.getBytesIn());
+ target.setFlowFilesOut(target.getFlowFilesOut() + toMerge.getFlowFilesOut());
+ target.setBytesOut(target.getBytesOut() + toMerge.getBytesOut());
+ target.setTransmitting(Boolean.TRUE.equals(target.isTransmitting()) || Boolean.TRUE.equals(toMerge.isTransmitting()));
+
+ // should be unnecessary here since ports run status not should be affected by
+ // environmental conditions but doing so in case that changes
+ if (RunStatus.Invalid.name().equals(toMerge.getRunStatus())) {
+ target.setRunStatus(RunStatus.Invalid.name());
+ }
+
+ target.setBulletins(mergeBulletins(target.getBulletins(), toMerge.getBulletins()));
+ updatePrettyPrintedFields(target);
+ }
+
+ public static void updatePrettyPrintedFields(final PortStatusDTO target) {
+ target.setInput(prettyPrint(target.getFlowFilesIn(), target.getBytesIn()));
+ target.setOutput(prettyPrint(target.getFlowFilesOut(), target.getBytesOut()));
+ }
+
+
+ public static String formatCount(final Integer intStatus) {
+ return intStatus == null ? "-" : FormatUtils.formatCount(intStatus);
+ }
+
+ public static String formatDataSize(final Long longStatus) {
+ return longStatus == null ? "-" : FormatUtils.formatDataSize(longStatus);
+ }
+
+ public static String prettyPrint(final Integer count, final Long bytes) {
+ return formatCount(count) + " / " + formatDataSize(bytes);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/ad32cb82/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/entity/ClusterStatusHistoryEntity.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/entity/ClusterStatusHistoryEntity.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/entity/ClusterStatusHistoryEntity.java
deleted file mode 100644
index b13984a..0000000
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/entity/ClusterStatusHistoryEntity.java
+++ /dev/null
@@ -1,43 +0,0 @@
-/*
- * 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.entity;
-
-import javax.xml.bind.annotation.XmlRootElement;
-import org.apache.nifi.web.api.dto.status.ClusterStatusHistoryDTO;
-
-/**
- * A serialized representation of this class can be placed in the entity body of a request or response to or from the API. This particular entity holds a reference to a ClusterStatusHistoryDTO.
- */
-@XmlRootElement(name = "clusterStatusHistoryEntity")
-public class ClusterStatusHistoryEntity extends Entity {
-
- private ClusterStatusHistoryDTO clusterStatusHistory;
-
- /**
- * The ClusterStatusHistoryDTO that is being serialized.
- *
- * @return The ClusterStatusHistoryDTO object
- */
- public ClusterStatusHistoryDTO getClusterStatusHistory() {
- return clusterStatusHistory;
- }
-
- public void setClusterStatusHistory(ClusterStatusHistoryDTO clusterStatusHistory) {
- this.clusterStatusHistory = clusterStatusHistory;
- }
-
-}
http://git-wip-us.apache.org/repos/asf/nifi/blob/ad32cb82/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/NodeProtocolSender.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/NodeProtocolSender.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/NodeProtocolSender.java
index f3e5df4..be0c339 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/NodeProtocolSender.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/NodeProtocolSender.java
@@ -20,7 +20,6 @@ import org.apache.nifi.cluster.protocol.message.ConnectionRequestMessage;
import org.apache.nifi.cluster.protocol.message.ConnectionResponseMessage;
import org.apache.nifi.cluster.protocol.message.ControllerStartupFailureMessage;
import org.apache.nifi.cluster.protocol.message.HeartbeatMessage;
-import org.apache.nifi.cluster.protocol.message.NodeBulletinsMessage;
import org.apache.nifi.cluster.protocol.message.ReconnectionFailureMessage;
/**
@@ -52,15 +51,6 @@ public interface NodeProtocolSender {
void heartbeat(HeartbeatMessage msg) throws ProtocolException, UnknownServiceAddressException;
/**
- * Sends a bulletins message to the cluster manager.
- *
- * @param msg a message
- * @throws ProtocolException pe
- * @throws UnknownServiceAddressException ex
- */
- void sendBulletins(NodeBulletinsMessage msg) throws ProtocolException, UnknownServiceAddressException;
-
- /**
* Sends a failure notification if the controller was unable start.
*
* @param msg a message
http://git-wip-us.apache.org/repos/asf/nifi/blob/ad32cb82/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/impl/NodeProtocolSenderImpl.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/impl/NodeProtocolSenderImpl.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/impl/NodeProtocolSenderImpl.java
index 993dea5..9ae6182 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/impl/NodeProtocolSenderImpl.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/impl/NodeProtocolSenderImpl.java
@@ -32,7 +32,6 @@ import org.apache.nifi.cluster.protocol.message.ConnectionRequestMessage;
import org.apache.nifi.cluster.protocol.message.ConnectionResponseMessage;
import org.apache.nifi.cluster.protocol.message.ControllerStartupFailureMessage;
import org.apache.nifi.cluster.protocol.message.HeartbeatMessage;
-import org.apache.nifi.cluster.protocol.message.NodeBulletinsMessage;
import org.apache.nifi.cluster.protocol.message.ProtocolMessage;
import org.apache.nifi.cluster.protocol.message.ProtocolMessage.MessageType;
import org.apache.nifi.cluster.protocol.message.ReconnectionFailureMessage;
@@ -117,11 +116,6 @@ public class NodeProtocolSenderImpl implements NodeProtocolSender {
}
@Override
- public void sendBulletins(NodeBulletinsMessage msg) throws ProtocolException, UnknownServiceAddressException {
- sendProtocolMessage(msg);
- }
-
- @Override
public void notifyControllerStartupFailure(final ControllerStartupFailureMessage msg) throws ProtocolException, UnknownServiceAddressException {
sendProtocolMessage(msg);
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/ad32cb82/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/impl/NodeProtocolSenderListener.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/impl/NodeProtocolSenderListener.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/impl/NodeProtocolSenderListener.java
index 2992e38..0a9a064 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/impl/NodeProtocolSenderListener.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/impl/NodeProtocolSenderListener.java
@@ -24,7 +24,6 @@ import org.apache.nifi.cluster.protocol.ProtocolException;
import org.apache.nifi.cluster.protocol.ProtocolHandler;
import org.apache.nifi.cluster.protocol.ProtocolListener;
import org.apache.nifi.cluster.protocol.UnknownServiceAddressException;
-import org.apache.nifi.cluster.protocol.message.NodeBulletinsMessage;
import org.apache.nifi.cluster.protocol.message.ConnectionRequestMessage;
import org.apache.nifi.cluster.protocol.message.ConnectionResponseMessage;
import org.apache.nifi.cluster.protocol.message.ControllerStartupFailureMessage;
@@ -104,11 +103,6 @@ public class NodeProtocolSenderListener implements NodeProtocolSender, ProtocolL
}
@Override
- public void sendBulletins(NodeBulletinsMessage msg) throws ProtocolException, UnknownServiceAddressException {
- sender.sendBulletins(msg);
- }
-
- @Override
public void setBulletinRepository(final BulletinRepository bulletinRepository) {
listener.setBulletinRepository(bulletinRepository);
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/ad32cb82/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/jaxb/message/ObjectFactory.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/jaxb/message/ObjectFactory.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/jaxb/message/ObjectFactory.java
index f0a9fa7..516b67e 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/jaxb/message/ObjectFactory.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/jaxb/message/ObjectFactory.java
@@ -26,7 +26,6 @@ import org.apache.nifi.cluster.protocol.message.FlowRequestMessage;
import org.apache.nifi.cluster.protocol.message.FlowResponseMessage;
import org.apache.nifi.cluster.protocol.message.HeartbeatMessage;
import org.apache.nifi.cluster.protocol.message.MulticastProtocolMessage;
-import org.apache.nifi.cluster.protocol.message.NodeBulletinsMessage;
import org.apache.nifi.cluster.protocol.message.PingMessage;
import org.apache.nifi.cluster.protocol.message.PrimaryRoleAssignmentMessage;
import org.apache.nifi.cluster.protocol.message.ReconnectionFailureMessage;
@@ -97,8 +96,4 @@ public class ObjectFactory {
public PrimaryRoleAssignmentMessage createPrimaryRoleAssignmentMessage() {
return new PrimaryRoleAssignmentMessage();
}
-
- public NodeBulletinsMessage createBulletinsMessage() {
- return new NodeBulletinsMessage();
- }
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/ad32cb82/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/message/NodeBulletinsMessage.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/message/NodeBulletinsMessage.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/message/NodeBulletinsMessage.java
deleted file mode 100644
index 6df3ba4..0000000
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/message/NodeBulletinsMessage.java
+++ /dev/null
@@ -1,42 +0,0 @@
-/*
- * 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.protocol.message;
-
-import org.apache.nifi.cluster.protocol.NodeBulletins;
-import javax.xml.bind.annotation.XmlRootElement;
-
-/**
- */
-@XmlRootElement(name = "nodeBulletinsMessage")
-public class NodeBulletinsMessage extends ProtocolMessage {
-
- private NodeBulletins bulletins;
-
- @Override
- public MessageType getType() {
- return MessageType.BULLETINS;
- }
-
- public NodeBulletins getBulletins() {
- return bulletins;
- }
-
- public void setBulletins(NodeBulletins bulletins) {
- this.bulletins = bulletins;
- }
-
-}
http://git-wip-us.apache.org/repos/asf/nifi/blob/ad32cb82/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/message/ProtocolMessage.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/message/ProtocolMessage.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/message/ProtocolMessage.java
index c6f7ce0..f01efd8 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/message/ProtocolMessage.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster-protocol/src/main/java/org/apache/nifi/cluster/protocol/message/ProtocolMessage.java
@@ -21,8 +21,6 @@ public abstract class ProtocolMessage {
private volatile String requestorDN;
public static enum MessageType {
-
- BULLETINS,
CONNECTION_REQUEST,
CONNECTION_RESPONSE,
CONTROLLER_STARTUP_FAILURE,
http://git-wip-us.apache.org/repos/asf/nifi/blob/ad32cb82/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/ClusterManager.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/ClusterManager.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/ClusterManager.java
index 336d675..99141c4 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/ClusterManager.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/ClusterManager.java
@@ -33,7 +33,6 @@ import org.apache.nifi.cluster.protocol.ConnectionRequest;
import org.apache.nifi.cluster.protocol.ConnectionResponse;
import org.apache.nifi.cluster.protocol.Heartbeat;
import org.apache.nifi.cluster.protocol.NodeIdentifier;
-import org.apache.nifi.controller.status.ProcessGroupStatus;
import org.apache.nifi.diagnostics.SystemDiagnostics;
import org.apache.nifi.remote.cluster.NodeInformant;
import org.apache.nifi.reporting.BulletinRepository;
@@ -170,12 +169,6 @@ public interface ClusterManager extends NodeInformant {
BulletinRepository getBulletinRepository();
/**
- * @param groupId groupId
- * @return a {@link ProcessGroupStatus} that represents the status of all nodes with the given {@link Status}es for the given ProcessGroup id, or null if no nodes exist with the given statuses
- */
- ProcessGroupStatus getProcessGroupStatus(String groupId);
-
- /**
* @return a merged representation of the System Diagnostics for all nodes in the cluster
*/
SystemDiagnostics getSystemDiagnostics();
http://git-wip-us.apache.org/repos/asf/nifi/blob/ad32cb82/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/impl/ClusteredEventAccess.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/impl/ClusteredEventAccess.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/impl/ClusteredEventAccess.java
index d3d5559..3966a31 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/impl/ClusteredEventAccess.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/impl/ClusteredEventAccess.java
@@ -46,7 +46,7 @@ public class ClusteredEventAccess implements EventAccess {
@Override
public ProcessGroupStatus getControllerStatus() {
- return clusterManager.getProcessGroupStatus(WebClusterManager.ROOT_GROUP_ID_ALIAS);
+ return new ProcessGroupStatus();
}
@Override
[5/5] nifi git commit: NIFI-1563: Federate requests and merge
responses from nodes instead of storing bulletins and stats at NCM
Posted by ma...@apache.org.
NIFI-1563: Federate requests and merge responses from nodes instead of storing bulletins and stats at NCM
Project: http://git-wip-us.apache.org/repos/asf/nifi/repo
Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/ad32cb82
Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/ad32cb82
Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/ad32cb82
Branch: refs/heads/NIFI-1563
Commit: ad32cb8243cc20ce7b04cc1fce0263503322ce5b
Parents: 6776060
Author: Mark Payne <ma...@hotmail.com>
Authored: Thu Mar 3 13:29:34 2016 -0500
Committer: Mark Payne <ma...@hotmail.com>
Committed: Thu Mar 3 13:29:34 2016 -0500
----------------------------------------------------------------------
.../nifi-framework/nifi-client-dto/pom.xml | 8 +
.../api/dto/status/ClusterStatusHistoryDTO.java | 79 --
.../web/api/dto/status/ConnectionStatusDTO.java | 168 +++--
.../web/api/dto/status/ControllerStatusDTO.java | 134 ++--
.../api/dto/status/NodeStatusHistoryDTO.java | 60 --
.../nifi/web/api/dto/status/PortStatusDTO.java | 100 ++-
.../api/dto/status/ProcessGroupStatusDTO.java | 372 +++++++---
.../web/api/dto/status/ProcessorStatusDTO.java | 159 +++-
.../dto/status/RemoteProcessGroupStatusDTO.java | 107 ++-
.../nifi/web/api/dto/status/StatusDTO.java | 14 +-
.../web/api/dto/status/StatusHistoryDTO.java | 66 +-
.../api/dto/status/StatusHistoryDetailDTO.java | 59 --
.../nifi/web/api/dto/status/StatusMerger.java | 375 ++++++++++
.../api/entity/ClusterStatusHistoryEntity.java | 43 --
.../cluster/protocol/NodeProtocolSender.java | 10 -
.../protocol/impl/NodeProtocolSenderImpl.java | 6 -
.../impl/NodeProtocolSenderListener.java | 6 -
.../protocol/jaxb/message/ObjectFactory.java | 5 -
.../protocol/message/NodeBulletinsMessage.java | 42 --
.../protocol/message/ProtocolMessage.java | 2 -
.../nifi/cluster/manager/ClusterManager.java | 7 -
.../manager/impl/ClusteredEventAccess.java | 2 +-
.../cluster/manager/impl/WebClusterManager.java | 737 +++++++++----------
.../apache/nifi/controller/FlowController.java | 192 +----
.../history/ConnectionStatusDescriptor.java | 110 +++
.../history/ProcessGroupStatusDescriptor.java | 143 ++++
.../history/ProcessorStatusDescriptor.java | 220 ++++++
.../RemoteProcessGroupStatusDescriptor.java | 127 ++++
.../status/history/StatusHistoryUtil.java | 4 +-
.../VolatileComponentStatusRepository.java | 418 +----------
.../events/NodeBulletinProcessingStrategy.java | 38 +-
.../nifi/events/VolatileBulletinRepository.java | 20 -
.../org/apache/nifi/web/NiFiServiceFacade.java | 24 -
.../nifi/web/StandardNiFiServiceFacade.java | 129 +---
.../apache/nifi/web/api/ClusterResource.java | 221 +-----
.../apache/nifi/web/api/ConnectionResource.java | 89 ++-
.../apache/nifi/web/api/ControllerResource.java | 4 +
.../nifi/web/api/ProcessGroupResource.java | 29 +-
.../apache/nifi/web/api/ProcessorResource.java | 37 +-
.../web/api/RemoteProcessGroupResource.java | 30 +-
.../nifi/web/api/ReportingTaskResource.java | 38 +-
.../org/apache/nifi/web/api/dto/DtoFactory.java | 98 ++-
.../src/main/resources/nifi-web-api-context.xml | 1 +
43 files changed, 2364 insertions(+), 2169 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/nifi/blob/ad32cb82/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/pom.xml
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/pom.xml b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/pom.xml
index c1cb557..3318f3b 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/pom.xml
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/pom.xml
@@ -26,5 +26,13 @@
<groupId>com.wordnik</groupId>
<artifactId>swagger-annotations</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-utils</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-api</artifactId>
+ </dependency>
</dependencies>
</project>
http://git-wip-us.apache.org/repos/asf/nifi/blob/ad32cb82/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/status/ClusterStatusHistoryDTO.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/status/ClusterStatusHistoryDTO.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/status/ClusterStatusHistoryDTO.java
deleted file mode 100644
index 997e24b..0000000
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/status/ClusterStatusHistoryDTO.java
+++ /dev/null
@@ -1,79 +0,0 @@
-/*
- * 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.dto.status;
-
-import com.wordnik.swagger.annotations.ApiModelProperty;
-import java.util.Collection;
-import java.util.Date;
-import javax.xml.bind.annotation.XmlType;
-import javax.xml.bind.annotation.adapters.XmlJavaTypeAdapter;
-import org.apache.nifi.web.api.dto.util.TimeAdapter;
-
-/**
- * DTO for serializing the a status history across the cluster.
- */
-@XmlType(name = "clusterStatusHistory")
-public class ClusterStatusHistoryDTO {
-
- private Collection<NodeStatusHistoryDTO> nodeStatusHistory;
- private StatusHistoryDTO clusterStatusHistory;
- private Date generated;
-
- /**
- * @return when this status history was generated
- */
- @XmlJavaTypeAdapter(TimeAdapter.class)
- @ApiModelProperty(
- value = "When the status history was generated."
- )
- public Date getGenerated() {
- return generated;
- }
-
- public void setGenerated(Date generated) {
- this.generated = generated;
- }
-
- /**
- * @return status history from each node in the cluster
- */
- @ApiModelProperty(
- value = "The status history from each node."
- )
- public Collection<NodeStatusHistoryDTO> getNodeStatusHistory() {
- return nodeStatusHistory;
- }
-
- public void setNodeStatusHistory(Collection<NodeStatusHistoryDTO> nodeStatusHistory) {
- this.nodeStatusHistory = nodeStatusHistory;
- }
-
- /**
- * @return status history for this component across the entire cluster
- */
- @ApiModelProperty(
- value = "The status history for the entire cluster."
- )
- public StatusHistoryDTO getClusterStatusHistory() {
- return clusterStatusHistory;
- }
-
- public void setClusterStatusHistory(StatusHistoryDTO clusterStatusHistory) {
- this.clusterStatusHistory = clusterStatusHistory;
- }
-
-}
http://git-wip-us.apache.org/repos/asf/nifi/blob/ad32cb82/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/status/ConnectionStatusDTO.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/status/ConnectionStatusDTO.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/status/ConnectionStatusDTO.java
index dc17c21..4083e1f 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/status/ConnectionStatusDTO.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/status/ConnectionStatusDTO.java
@@ -23,29 +23,34 @@ import javax.xml.bind.annotation.XmlType;
* DTO for serializing the status of a connection.
*/
@XmlType(name = "connectionStatus")
-public class ConnectionStatusDTO {
+public class ConnectionStatusDTO implements Cloneable {
private String id;
private String groupId;
private String name;
- private String input;
- private String queuedCount;
- private String queuedSize;
- private String queued;
- private String output;
private String sourceId;
private String sourceName;
private String destinationId;
private String destinationName;
+ private Integer flowFilesIn = 0;
+ private Long bytesIn = 0L;
+ private String input;
+ private Integer flowFilesOut = 0;
+ private Long bytesOut = 0L;
+ private String output;
+ private Integer flowFilesQueued = 0;
+ private Long bytesQueued = 0L;
+ private String queued;
+ private String queuedSize;
+ private String queuedCount;
+
/* getters / setters */
/**
* @return The connection id
*/
- @ApiModelProperty(
- value = "The id of the connection."
- )
+ @ApiModelProperty("The id of the connection.")
public String getId() {
return id;
}
@@ -57,9 +62,7 @@ public class ConnectionStatusDTO {
/**
* @return the ID of the Process Group to which this connection belongs.
*/
- @ApiModelProperty(
- value = "The id of the process group the connection belongs to."
- )
+ @ApiModelProperty("The id of the process group the connection belongs to.")
public String getGroupId() {
return groupId;
}
@@ -71,9 +74,7 @@ public class ConnectionStatusDTO {
/**
* @return name of this connection
*/
- @ApiModelProperty(
- value = "The name of the connection."
- )
+ @ApiModelProperty("The name of the connection.")
public String getName() {
return name;
}
@@ -85,9 +86,7 @@ public class ConnectionStatusDTO {
/**
* @return total count of flow files that are queued
*/
- @ApiModelProperty(
- value = "The number of flowfiles that are queued."
- )
+ @ApiModelProperty("The number of flowfiles that are queued, pretty printed.")
public String getQueuedCount() {
return queuedCount;
}
@@ -96,16 +95,28 @@ public class ConnectionStatusDTO {
this.queuedCount = queuedCount;
}
+
/**
* @return total size of flow files that are queued
*/
- @ApiModelProperty(
- value = "The total size of flowfiles that are queued formatted."
- )
+ @ApiModelProperty("The total size of flowfiles that are queued formatted.")
public String getQueuedSize() {
return queuedSize;
}
+
+ public void setInput(String input) {
+ this.input = input;
+ }
+
+ public void setOutput(String output) {
+ this.output = output;
+ }
+
+ public void setQueued(String queued) {
+ this.queued = queued;
+ }
+
public void setQueuedSize(String queuedSize) {
this.queuedSize = queuedSize;
}
@@ -113,23 +124,16 @@ public class ConnectionStatusDTO {
/**
* @return The total count and size of queued flow files
*/
- @ApiModelProperty(
- value = "The total count and size of queued flowfiles formatted."
- )
+ @ApiModelProperty("The total count and size of queued flowfiles formatted.")
public String getQueued() {
return queued;
}
- public void setQueued(String queued) {
- this.queued = queued;
- }
/**
* @return id of the source of this connection
*/
- @ApiModelProperty(
- value = "The id of the source of the connection."
- )
+ @ApiModelProperty("The id of the source of the connection.")
public String getSourceId() {
return sourceId;
}
@@ -141,9 +145,7 @@ public class ConnectionStatusDTO {
/**
* @return name of the source of this connection
*/
- @ApiModelProperty(
- value = "The name of the source of the connection."
- )
+ @ApiModelProperty("The name of the source of the connection.")
public String getSourceName() {
return sourceName;
}
@@ -155,9 +157,7 @@ public class ConnectionStatusDTO {
/**
* @return id of the destination of this connection
*/
- @ApiModelProperty(
- value = "The id of the destination of the connection."
- )
+ @ApiModelProperty("The id of the destination of the connection.")
public String getDestinationId() {
return destinationId;
}
@@ -169,9 +169,7 @@ public class ConnectionStatusDTO {
/**
* @return name of the destination of this connection
*/
- @ApiModelProperty(
- value = "The name of the destination of the connection."
- )
+ @ApiModelProperty("The name of the destination of the connection.")
public String getDestinationName() {
return destinationName;
}
@@ -183,29 +181,99 @@ public class ConnectionStatusDTO {
/**
* @return input for this connection
*/
- @ApiModelProperty(
- value = "The input count/size for the connection in the last 5 minutes."
- )
+ @ApiModelProperty("The input count/size for the connection in the last 5 minutes, pretty printed.")
public String getInput() {
return input;
}
- public void setInput(String input) {
- this.input = input;
- }
/**
* @return output for this connection
*/
- @ApiModelProperty(
- value = "The output count/sie for the connection in the last 5 minutes."
- )
+ @ApiModelProperty("The output count/sie for the connection in the last 5 minutes, pretty printed.")
public String getOutput() {
return output;
}
- public void setOutput(String output) {
- this.output = output;
+
+ @ApiModelProperty("The number of FlowFiles that have come into the connection in the last 5 minutes.")
+ public Integer getFlowFilesIn() {
+ return flowFilesIn;
+ }
+
+ public void setFlowFilesIn(Integer flowFilesIn) {
+ this.flowFilesIn = flowFilesIn;
+ }
+
+ @ApiModelProperty("The size of the FlowFiles that have come into the connection in the last 5 minutes.")
+ public Long getBytesIn() {
+ return bytesIn;
+ }
+
+ public void setBytesIn(Long bytesIn) {
+ this.bytesIn = bytesIn;
+ }
+
+ @ApiModelProperty("The number of FlowFiles that have left the connection in the last 5 minutes.")
+ public Integer getFlowFilesOut() {
+ return flowFilesOut;
+ }
+
+ public void setFlowFilesOut(Integer flowFilesOut) {
+ this.flowFilesOut = flowFilesOut;
+ }
+
+ @ApiModelProperty("The number of bytes that have left the connection in the last 5 minutes.")
+ public Long getBytesOut() {
+ return bytesOut;
+ }
+
+ public void setBytesOut(Long bytesOut) {
+ this.bytesOut = bytesOut;
+ }
+
+ @ApiModelProperty("The number of FlowFiles that are currently queued in the connection.")
+ public Integer getFlowFilesQueued() {
+ return flowFilesQueued;
+ }
+
+ public void setFlowFilesQueued(Integer flowFilesQueued) {
+ this.flowFilesQueued = flowFilesQueued;
}
+ @ApiModelProperty("The size of the FlowFiles that are currently queued in the connection.")
+ public Long getBytesQueued() {
+ return bytesQueued;
+ }
+
+ public void setBytesQueued(Long bytesQueued) {
+ this.bytesQueued = bytesQueued;
+ }
+
+
+ @Override
+ public ConnectionStatusDTO clone() {
+ final ConnectionStatusDTO other = new ConnectionStatusDTO();
+ other.setDestinationId(getDestinationId());
+ other.setDestinationName(getDestinationName());
+ other.setGroupId(getGroupId());
+ other.setId(getId());
+ other.setName(getName());
+ other.setSourceId(getSourceId());
+ other.setSourceName(getSourceName());
+
+ other.setFlowFilesIn(getFlowFilesIn());
+ other.setBytesIn(getBytesIn());
+ other.setInput(getInput());
+ other.setFlowFilesOut(getFlowFilesOut());
+ other.setBytesOut(getBytesOut());
+ other.setOutput(getOutput());
+ other.setFlowFilesQueued(getFlowFilesQueued());
+ other.setBytesQueued(getBytesQueued());
+ other.setQueued(getQueued());
+ other.setQueuedCount(getQueuedCount());
+ other.setQueuedSize(getQueuedSize());
+
+ return other;
+ }
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/ad32cb82/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/status/ControllerStatusDTO.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/status/ControllerStatusDTO.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/status/ControllerStatusDTO.java
index 5d5eddf..03e2124 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/status/ControllerStatusDTO.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/status/ControllerStatusDTO.java
@@ -17,6 +17,8 @@
package org.apache.nifi.web.api.dto.status;
import com.wordnik.swagger.annotations.ApiModelProperty;
+
+import java.util.ArrayList;
import java.util.List;
import javax.xml.bind.annotation.XmlType;
import org.apache.nifi.web.api.dto.BulletinDTO;
@@ -25,19 +27,25 @@ import org.apache.nifi.web.api.dto.BulletinDTO;
* The status of this NiFi controller.
*/
@XmlType(name = "controllerStatus")
-public class ControllerStatusDTO {
+public class ControllerStatusDTO implements Cloneable {
- private Integer activeThreadCount;
+ private Integer activeThreadCount = 0;
private String queued;
+ private Integer flowFilesQueued = 0;
+ private Long bytesQueued = 0L;
+
private String connectedNodes;
+ private Integer connectedNodeCount = 0;
+ private Integer totalNodeCount = 0;
+
private Boolean hasPendingAccounts;
- private Integer runningCount;
- private Integer stoppedCount;
- private Integer invalidCount;
- private Integer disabledCount;
- private Integer activeRemotePortCount;
- private Integer inactiveRemotePortCount;
+ private Integer runningCount = 0;
+ private Integer stoppedCount = 0;
+ private Integer invalidCount = 0;
+ private Integer disabledCount = 0;
+ private Integer activeRemotePortCount = 0;
+ private Integer inactiveRemotePortCount = 0;
private List<BulletinDTO> bulletins;
private List<BulletinDTO> controllerServiceBulletins;
@@ -48,9 +56,7 @@ public class ControllerStatusDTO {
*
* @return The active thread count
*/
- @ApiModelProperty(
- value = "The number of active threads in the NiFi."
- )
+ @ApiModelProperty("The number of active threads in the NiFi.")
public Integer getActiveThreadCount() {
return activeThreadCount;
}
@@ -62,9 +68,7 @@ public class ControllerStatusDTO {
/**
* @return queue for the controller
*/
- @ApiModelProperty(
- value = "The number of flowfilew queued in the NiFi."
- )
+ @ApiModelProperty("The number of flowfilew queued in the NiFi.")
public String getQueued() {
return queued;
}
@@ -77,9 +81,7 @@ public class ControllerStatusDTO {
* @return Used in clustering, will report the number of nodes connected vs
* the number of nodes in the cluster
*/
- @ApiModelProperty(
- value = "When clustered, reports the number of nodes connected vs the number of nodes in the cluster."
- )
+ @ApiModelProperty("When clustered, reports the number of nodes connected vs the number of nodes in the cluster.")
public String getConnectedNodes() {
return connectedNodes;
}
@@ -91,9 +93,7 @@ public class ControllerStatusDTO {
/**
* @return System bulletins to be reported to the user
*/
- @ApiModelProperty(
- value = "System level bulletins to be reported to the user."
- )
+ @ApiModelProperty("System level bulletins to be reported to the user.")
public List<BulletinDTO> getBulletins() {
return bulletins;
}
@@ -105,9 +105,7 @@ public class ControllerStatusDTO {
/**
* @return Controller service bulletins to be reported to the user
*/
- @ApiModelProperty(
- value = "Controller service bulletins to be reported to the user."
- )
+ @ApiModelProperty("Controller service bulletins to be reported to the user.")
public List<BulletinDTO> getControllerServiceBulletins() {
return controllerServiceBulletins;
}
@@ -119,9 +117,7 @@ public class ControllerStatusDTO {
/**
* @return Reporting task bulletins to be reported to the user
*/
- @ApiModelProperty(
- value = "Reporting task bulletins to be reported to the user."
- )
+ @ApiModelProperty("Reporting task bulletins to be reported to the user.")
public List<BulletinDTO> getReportingTaskBulletins() {
return reportingTaskBulletins;
}
@@ -133,9 +129,7 @@ public class ControllerStatusDTO {
/**
* @return whether or not there are pending user requests
*/
- @ApiModelProperty(
- value = "Whether there are any pending user account requests."
- )
+ @ApiModelProperty("Whether there are any pending user account requests.")
public Boolean getHasPendingAccounts() {
return hasPendingAccounts;
}
@@ -147,9 +141,7 @@ public class ControllerStatusDTO {
/**
* @return number of running components in this controller
*/
- @ApiModelProperty(
- value = "The number of running components in the NiFi."
- )
+ @ApiModelProperty("The number of running components in the NiFi.")
public Integer getRunningCount() {
return runningCount;
}
@@ -161,9 +153,7 @@ public class ControllerStatusDTO {
/**
* @return number of stopped components in this controller
*/
- @ApiModelProperty(
- value = "The number of stopped components in the NiFi."
- )
+ @ApiModelProperty("The number of stopped components in the NiFi.")
public Integer getStoppedCount() {
return stoppedCount;
}
@@ -175,9 +165,7 @@ public class ControllerStatusDTO {
/**
* @return number of invalid components in this controller
*/
- @ApiModelProperty(
- value = "The number of invalid components in the NiFi."
- )
+ @ApiModelProperty("The number of invalid components in the NiFi.")
public Integer getInvalidCount() {
return invalidCount;
}
@@ -189,9 +177,7 @@ public class ControllerStatusDTO {
/**
* @return number of disabled components in this controller
*/
- @ApiModelProperty(
- value = "The number of disabled components in the NiFi."
- )
+ @ApiModelProperty("The number of disabled components in the NiFi.")
public Integer getDisabledCount() {
return disabledCount;
}
@@ -203,9 +189,7 @@ public class ControllerStatusDTO {
/**
* @return number of active remote ports in this controller
*/
- @ApiModelProperty(
- value = "The number of active remote ports in the NiFi."
- )
+ @ApiModelProperty("The number of active remote ports in the NiFi.")
public Integer getActiveRemotePortCount() {
return activeRemotePortCount;
}
@@ -217,9 +201,7 @@ public class ControllerStatusDTO {
/**
* @return number of inactive remote ports in this controller
*/
- @ApiModelProperty(
- value = "The number of inactive remote ports in the NiFi."
- )
+ @ApiModelProperty("The number of inactive remote ports in the NiFi.")
public Integer getInactiveRemotePortCount() {
return inactiveRemotePortCount;
}
@@ -228,4 +210,62 @@ public class ControllerStatusDTO {
this.inactiveRemotePortCount = inactiveRemotePortCount;
}
+ @ApiModelProperty("The number of FlowFiles queued across the entire flow")
+ public Integer getFlowFilesQueued() {
+ return flowFilesQueued;
+ }
+
+ public void setFlowFilesQueued(Integer flowFilesQueued) {
+ this.flowFilesQueued = flowFilesQueued;
+ }
+
+ @ApiModelProperty("The size of the FlowFiles queued across the entire flow")
+ public Long getBytesQueued() {
+ return bytesQueued;
+ }
+
+ public void setBytesQueued(Long bytesQueued) {
+ this.bytesQueued = bytesQueued;
+ }
+
+ @ApiModelProperty("The number of nodes that are currently connected to the cluster")
+ public Integer getConnectedNodeCount() {
+ return connectedNodeCount;
+ }
+
+ public void setConnectedNodeCount(Integer connectedNodeCount) {
+ this.connectedNodeCount = connectedNodeCount;
+ }
+
+ @ApiModelProperty("The number of nodes in the cluster, regardless of whether or not they are connected")
+ public Integer getTotalNodeCount() {
+ return totalNodeCount;
+ }
+
+ public void setTotalNodeCount(Integer totalNodeCount) {
+ this.totalNodeCount = totalNodeCount;
+ }
+
+ @Override
+ public ControllerStatusDTO clone() {
+ final ControllerStatusDTO other = new ControllerStatusDTO();
+ other.setActiveThreadCount(getActiveThreadCount());
+ other.setQueued(getQueued());
+ other.setFlowFilesQueued(getFlowFilesQueued());
+ other.setBytesQueued(getBytesQueued());
+ other.setConnectedNodes(getConnectedNodes());
+ other.setConnectedNodeCount(getConnectedNodeCount());
+ other.setTotalNodeCount(getTotalNodeCount());
+ other.setHasPendingAccounts(getHasPendingAccounts());
+ other.setRunningCount(getRunningCount());
+ other.setStoppedCount(getStoppedCount());
+ other.setInvalidCount(getInvalidCount());
+ other.setDisabledCount(getDisabledCount());
+ other.setActiveRemotePortCount(getActiveRemotePortCount());
+ other.setInactiveRemotePortCount(getInactiveRemotePortCount());
+ other.setBulletins(getBulletins() == null ? null : new ArrayList<>(getBulletins()));
+ other.setControllerServiceBulletins(getControllerServiceBulletins() == null ? null : new ArrayList<>(getControllerServiceBulletins()));
+ other.setReportingTaskBulletins(getReportingTaskBulletins() == null ? null : new ArrayList<>(getReportingTaskBulletins()));
+ return other;
+ }
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/ad32cb82/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/status/NodeStatusHistoryDTO.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/status/NodeStatusHistoryDTO.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/status/NodeStatusHistoryDTO.java
deleted file mode 100644
index 5cf9f41..0000000
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/status/NodeStatusHistoryDTO.java
+++ /dev/null
@@ -1,60 +0,0 @@
-/*
- * 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.dto.status;
-
-import com.wordnik.swagger.annotations.ApiModelProperty;
-import javax.xml.bind.annotation.XmlType;
-import org.apache.nifi.web.api.dto.NodeDTO;
-
-/**
- * DTO for serializing the status history for a particular node.
- */
-@XmlType(name = "nodeStatusHistory")
-public class NodeStatusHistoryDTO {
-
- private NodeDTO node;
- private StatusHistoryDTO statusHistory;
-
- /**
- * @return the node
- */
- @ApiModelProperty(
- value = "The node."
- )
- public NodeDTO getNode() {
- return node;
- }
-
- public void setNode(NodeDTO node) {
- this.node = node;
- }
-
- /**
- * @return processor status history
- */
- @ApiModelProperty(
- value = "The processor status for each node."
- )
- public StatusHistoryDTO getStatusHistory() {
- return statusHistory;
- }
-
- public void setStatusHistory(StatusHistoryDTO statusHistory) {
- this.statusHistory = statusHistory;
- }
-
-}
http://git-wip-us.apache.org/repos/asf/nifi/blob/ad32cb82/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/status/PortStatusDTO.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/status/PortStatusDTO.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/status/PortStatusDTO.java
index c1d95d0..6a2a972 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/status/PortStatusDTO.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/status/PortStatusDTO.java
@@ -16,30 +16,35 @@
*/
package org.apache.nifi.web.api.dto.status;
-import com.wordnik.swagger.annotations.ApiModelProperty;
import javax.xml.bind.annotation.XmlType;
+import com.wordnik.swagger.annotations.ApiModelProperty;
+
/**
* The status for a port in this NiFi.
*/
@XmlType(name = "portStatus")
-public class PortStatusDTO extends StatusDTO {
+public class PortStatusDTO extends StatusDTO implements Cloneable {
private String id;
private String groupId;
private String name;
- private Integer activeThreadCount;
+
+ private Integer activeThreadCount = 0;
+ private Integer flowFilesIn = 0;
+ private Long bytesIn = 0L;
private String input;
+ private Integer flowFilesOut = 0;
+ private Long bytesOut = 0L;
private String output;
+
private Boolean transmitting;
private String runStatus;
/**
* @return whether this port has incoming or outgoing connections to a remote NiFi
*/
- @ApiModelProperty(
- value = "Whether the port has incoming or outgoing connections to a remote NiFi."
- )
+ @ApiModelProperty("Whether the port has incoming or outgoing connections to a remote NiFi.")
public Boolean isTransmitting() {
return transmitting;
}
@@ -51,9 +56,7 @@ public class PortStatusDTO extends StatusDTO {
/**
* @return the active thread count for this port
*/
- @ApiModelProperty(
- value = "The active thread count for the port."
- )
+ @ApiModelProperty("The active thread count for the port.")
public Integer getActiveThreadCount() {
return activeThreadCount;
}
@@ -65,9 +68,7 @@ public class PortStatusDTO extends StatusDTO {
/**
* @return id of this port
*/
- @ApiModelProperty(
- value = "The id of the port."
- )
+ @ApiModelProperty("The id of the port.")
public String getId() {
return id;
}
@@ -79,9 +80,7 @@ public class PortStatusDTO extends StatusDTO {
/**
* @return id of the group this port resides in
*/
- @ApiModelProperty(
- value = "The id of the parent process group of the port."
- )
+ @ApiModelProperty("The id of the parent process group of the port.")
public String getGroupId() {
return groupId;
}
@@ -93,9 +92,7 @@ public class PortStatusDTO extends StatusDTO {
/**
* @return name of this port
*/
- @ApiModelProperty(
- value = "The name of the port."
- )
+ @ApiModelProperty("The name of the port.")
public String getName() {
return name;
}
@@ -107,9 +104,7 @@ public class PortStatusDTO extends StatusDTO {
/**
* @return run status of this port
*/
- @ApiModelProperty(
- value = "The run status of the port."
- )
+ @ApiModelProperty("The run status of the port.")
public String getRunStatus() {
return runStatus;
}
@@ -121,9 +116,7 @@ public class PortStatusDTO extends StatusDTO {
/**
* @return The total count and size of flow files that have been accepted in the last five minutes
*/
- @ApiModelProperty(
- value = "The count/size of flowfiles that have been accepted in the last 5 minutes."
- )
+ @ApiModelProperty("The count/size of flowfiles that have been accepted in the last 5 minutes.")
public String getInput() {
return input;
}
@@ -135,9 +128,7 @@ public class PortStatusDTO extends StatusDTO {
/**
* @return The total count and size of flow files that have been processed in the last five minutes
*/
- @ApiModelProperty(
- value = "The count/size of flowfiles that have been processed in the last 5 minutes."
- )
+ @ApiModelProperty("The count/size of flowfiles that have been processed in the last 5 minutes.")
public String getOutput() {
return output;
}
@@ -146,4 +137,59 @@ public class PortStatusDTO extends StatusDTO {
this.output = output;
}
+ @ApiModelProperty("The number of FlowFiles that have been accepted in the last 5 minutes.")
+ public Integer getFlowFilesIn() {
+ return flowFilesIn;
+ }
+
+ public void setFlowFilesIn(Integer flowFilesIn) {
+ this.flowFilesIn = flowFilesIn;
+ }
+
+ @ApiModelProperty("The size of hte FlowFiles that have been accepted in the last 5 minutes.")
+ public Long getBytesIn() {
+ return bytesIn;
+ }
+
+ public void setBytesIn(Long bytesIn) {
+ this.bytesIn = bytesIn;
+ }
+
+ @ApiModelProperty("The number of FlowFiles that have been processed in the last 5 minutes.")
+ public Integer getFlowFilesOut() {
+ return flowFilesOut;
+ }
+
+ public void setFlowFilesOut(Integer flowFilesOut) {
+ this.flowFilesOut = flowFilesOut;
+ }
+
+ @ApiModelProperty("The number of bytes that have been processed in the last 5 minutes.")
+ public Long getBytesOut() {
+ return bytesOut;
+ }
+
+ public void setBytesOut(Long bytesOut) {
+ this.bytesOut = bytesOut;
+ }
+
+ @Override
+ public PortStatusDTO clone() {
+ final PortStatusDTO other = new PortStatusDTO();
+ other.setId(getId());
+ other.setGroupId(getGroupId());
+ other.setName(getName());
+ other.setActiveThreadCount(getActiveThreadCount());
+ other.setFlowFilesIn(getFlowFilesIn());
+ other.setBytesIn(getBytesIn());
+ other.setFlowFilesOut(getFlowFilesOut());
+ other.setBytesOut(getBytesOut());
+ other.setTransmitting(isTransmitting());
+ other.setRunStatus(getRunStatus());
+ other.setInput(getInput());
+ other.setOutput(getOutput());
+ other.setBulletins(cloneBulletins());
+
+ return other;
+ }
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/ad32cb82/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/status/ProcessGroupStatusDTO.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/status/ProcessGroupStatusDTO.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/status/ProcessGroupStatusDTO.java
index 3ebae8f..ff7eecc 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/status/ProcessGroupStatusDTO.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/status/ProcessGroupStatusDTO.java
@@ -16,18 +16,23 @@
*/
package org.apache.nifi.web.api.dto.status;
-import com.wordnik.swagger.annotations.ApiModelProperty;
+import java.util.ArrayList;
import java.util.Collection;
import java.util.Date;
+import java.util.List;
+
import javax.xml.bind.annotation.XmlType;
import javax.xml.bind.annotation.adapters.XmlJavaTypeAdapter;
+
import org.apache.nifi.web.api.dto.util.TimeAdapter;
+import com.wordnik.swagger.annotations.ApiModelProperty;
+
/**
* The status for a process group in this NiFi.
*/
@XmlType(name = "processGroupStatus")
-public class ProcessGroupStatusDTO extends StatusDTO {
+public class ProcessGroupStatusDTO extends StatusDTO implements Cloneable {
private String id;
private String name;
@@ -38,17 +43,38 @@ public class ProcessGroupStatusDTO extends StatusDTO {
private Collection<PortStatusDTO> inputPortStatus;
private Collection<PortStatusDTO> outputPortStatus;
+ private Integer flowFilesIn = 0;
+ private Long bytesIn = 0L;
private String input;
+
+ private Integer flowFilesQueued = 0;
+ private Long bytesQueued = 0L;
+ private String queued;
private String queuedCount;
private String queuedSize;
- private String queued;
+
+ private Long bytesRead = 0L;
private String read;
+ private Long bytesWritten = 0L;
private String written;
+
+ private Integer flowFilesOut = 0;
+ private Long bytesOut = 0L;
private String output;
+
+ private Integer flowFilesTransferred = 0;
+ private Long bytesTransferred = 0L;
private String transferred;
+
+ private Long bytesReceived = 0L;
+ private Integer flowFilesReceived = 0;
private String received;
+
+ private Long bytesSent = 0L;
+ private Integer flowFilesSent = 0;
private String sent;
- private Integer activeThreadCount;
+
+ private Integer activeThreadCount = 0;
private Date statsLastRefreshed;
/**
@@ -56,9 +82,7 @@ public class ProcessGroupStatusDTO extends StatusDTO {
*
* @return The id for the process group
*/
- @ApiModelProperty(
- value = "The id of the process group."
- )
+ @ApiModelProperty("The id of the process group.")
public String getId() {
return id;
}
@@ -70,9 +94,7 @@ public class ProcessGroupStatusDTO extends StatusDTO {
/**
* @return name of this process group
*/
- @ApiModelProperty(
- value = "The name of this process group."
- )
+ @ApiModelProperty("The name of this process group.")
public String getName() {
return name;
}
@@ -84,9 +106,7 @@ public class ProcessGroupStatusDTO extends StatusDTO {
/**
* @return active thread count for this process group
*/
- @ApiModelProperty(
- value = "The active thread count for this process group."
- )
+ @ApiModelProperty("The active thread count for this process group.")
public Integer getActiveThreadCount() {
return activeThreadCount;
}
@@ -100,9 +120,7 @@ public class ProcessGroupStatusDTO extends StatusDTO {
*
* @return The status of all connections
*/
- @ApiModelProperty(
- value = "The status of all conenctions in the process group."
- )
+ @ApiModelProperty("The status of all conenctions in the process group.")
public Collection<ConnectionStatusDTO> getConnectionStatus() {
return connectionStatus;
}
@@ -116,9 +134,7 @@ public class ProcessGroupStatusDTO extends StatusDTO {
*
* @return The status of all process groups
*/
- @ApiModelProperty(
- value = "The status of all process groups in the process group."
- )
+ @ApiModelProperty("The status of all process groups in the process group.")
public Collection<ProcessGroupStatusDTO> getProcessGroupStatus() {
return processGroupStatus;
}
@@ -132,9 +148,7 @@ public class ProcessGroupStatusDTO extends StatusDTO {
*
* @return The status of all remote process groups
*/
- @ApiModelProperty(
- value = "The status of all remote process groups in the process group.."
- )
+ @ApiModelProperty("The status of all remote process groups in the process group.")
public Collection<RemoteProcessGroupStatusDTO> getRemoteProcessGroupStatus() {
return remoteProcessGroupStatus;
}
@@ -148,9 +162,7 @@ public class ProcessGroupStatusDTO extends StatusDTO {
*
* @return The status of all processors
*/
- @ApiModelProperty(
- value = "The status of all processors in the process group."
- )
+ @ApiModelProperty("The status of all processors in the process group.")
public Collection<ProcessorStatusDTO> getProcessorStatus() {
return processorStatus;
}
@@ -164,9 +176,7 @@ public class ProcessGroupStatusDTO extends StatusDTO {
*
* @return The status of all input ports
*/
- @ApiModelProperty(
- value = "The status of all input ports in the process group."
- )
+ @ApiModelProperty("The status of all input ports in the process group.")
public Collection<PortStatusDTO> getInputPortStatus() {
return inputPortStatus;
}
@@ -180,9 +190,7 @@ public class ProcessGroupStatusDTO extends StatusDTO {
*
* @return The status of all output ports
*/
- @ApiModelProperty(
- value = "The status of all output ports in the process group."
- )
+ @ApiModelProperty("The status of all output ports in the process group.")
public Collection<PortStatusDTO> getOutputPortStatus() {
return outputPortStatus;
}
@@ -196,159 +204,233 @@ public class ProcessGroupStatusDTO extends StatusDTO {
*
* @return The output stats
*/
- @ApiModelProperty(
- value = "The output count/size for the process group in the last 5 minutes."
- )
+ @ApiModelProperty("The output count/size for the process group in the last 5 minutes.")
public String getOutput() {
return output;
}
- public void setOutput(String output) {
- this.output = output;
- }
-
/**
* The transferred stats for this process group. This represents the count/size of flowfiles transferred to/from queues.
*
* @return The transferred status for this process group
*/
- @ApiModelProperty(
- value = "The count/size transferred to/frome queues in the process group in the last 5 minutes."
- )
+ @ApiModelProperty("The count/size transferred to/frome queues in the process group in the last 5 minutes.")
public String getTransferred() {
return transferred;
}
- public void setTransferred(String transferred) {
- this.transferred = transferred;
- }
-
/**
* The received stats for this process group. This represents the count/size of flowfiles received.
*
* @return The received stats for this process group
*/
- @ApiModelProperty(
- value = "The count/size sent to the process group in the last 5 minutes."
- )
+ @ApiModelProperty("The count/size sent to the process group in the last 5 minutes.")
public String getReceived() {
return received;
}
- public void setReceived(String received) {
- this.received = received;
- }
/**
* The sent stats for this process group. This represents the count/size of flowfiles sent.
*
* @return The sent stats for this process group
*/
- @ApiModelProperty(
- value = "The count/size sent from this process group in the last 5 minutes."
- )
+ @ApiModelProperty("The count/size sent from this process group in the last 5 minutes.")
public String getSent() {
return sent;
}
- public void setSent(String sent) {
- this.sent = sent;
- }
/**
* The queued count for this process group.
*
* @return The queued count for this process group
*/
- @ApiModelProperty(
- value = "The count that is queued for the process group."
- )
+ @ApiModelProperty("The count that is queued for the process group.")
public String getQueuedCount() {
return queuedCount;
}
- public void setQueuedCount(String queuedCount) {
- this.queuedCount = queuedCount;
- }
/**
* The queued size for this process group.
*
* @return The queued size for this process group
*/
- @ApiModelProperty(
- value = "The size that is queued for the process group."
- )
+ @ApiModelProperty("The size that is queued for the process group.")
public String getQueuedSize() {
return queuedSize;
}
- public void setQueuedSize(String queuedSize) {
- this.queuedSize = queuedSize;
- }
/**
* The queued stats for this process group.
*
* @return The queued stats
*/
- @ApiModelProperty(
- value = "The count/size that is queued in the the process group."
- )
+ @ApiModelProperty("The count/size that is queued in the the process group.")
public String getQueued() {
return queued;
}
- public void setQueued(String queued) {
- this.queued = queued;
- }
/**
* The read stats for this process group.
*
* @return The read stats
*/
- @ApiModelProperty(
- value = "The number of bytes read in the last 5 minutes."
- )
+ @ApiModelProperty("The number of bytes read in the last 5 minutes.")
public String getRead() {
return read;
}
- public void setRead(String read) {
- this.read = read;
- }
/**
* The written stats for this process group.
*
* @return The written stats
*/
- @ApiModelProperty(
- value = "The number of bytes written in the last 5 minutes."
- )
+ @ApiModelProperty("The number of bytes written in the last 5 minutes.")
public String getWritten() {
return written;
}
- public void setWritten(String written) {
- this.written = written;
- }
/**
* The input stats for this process group.
*
* @return The input stats
*/
- @ApiModelProperty(
- value = "The input count/size for the process group in the last 5 minutes."
- )
+ @ApiModelProperty("The input count/size for the process group in the last 5 minutes (pretty printed).")
public String getInput() {
return input;
}
- public void setInput(String input) {
- this.input = input;
+
+ @ApiModelProperty("The number of FlowFiles that have come into this ProcessGroup in the last 5 minutes")
+ public Integer getFlowFilesIn() {
+ return flowFilesIn;
+ }
+
+ public void setFlowFilesIn(Integer flowFilesIn) {
+ this.flowFilesIn = flowFilesIn;
+ }
+
+ @ApiModelProperty("The number of bytes that have come into this ProcessGroup in the last 5 minutes")
+ public Long getBytesIn() {
+ return bytesIn;
+ }
+
+ public void setBytesIn(Long bytesIn) {
+ this.bytesIn = bytesIn;
+ }
+
+ @ApiModelProperty("The number of FlowFiles that are queued up in this ProcessGroup right now")
+ public Integer getFlowFilesQueued() {
+ return flowFilesQueued;
+ }
+
+ public void setFlowFilesQueued(Integer flowFilesQueued) {
+ this.flowFilesQueued = flowFilesQueued;
+ }
+
+ @ApiModelProperty("The number of bytes that are queued up in this ProcessGroup right now")
+ public Long getBytesQueued() {
+ return bytesQueued;
+ }
+
+ public void setBytesQueued(Long bytesQueued) {
+ this.bytesQueued = bytesQueued;
+ }
+
+ @ApiModelProperty("The number of bytes read by components in this ProcessGroup in the last 5 minutes")
+ public Long getBytesRead() {
+ return bytesRead;
+ }
+
+ public void setBytesRead(Long bytesRead) {
+ this.bytesRead = bytesRead;
+ }
+
+ @ApiModelProperty("The number of bytes written by components in this ProcessGroup in the last 5 minutes")
+ public Long getBytesWritten() {
+ return bytesWritten;
+ }
+
+ public void setBytesWritten(Long bytesWritten) {
+ this.bytesWritten = bytesWritten;
+ }
+
+ @ApiModelProperty("The number of FlowFiles transferred out of this ProcessGroup in the last 5 minutes")
+ public Integer getFlowFilesOut() {
+ return flowFilesOut;
+ }
+
+ public void setFlowFilesOut(Integer flowFilesOut) {
+ this.flowFilesOut = flowFilesOut;
+ }
+
+ @ApiModelProperty("The number of bytes transferred out of this ProcessGroup in the last 5 minutes")
+ public Long getBytesOut() {
+ return bytesOut;
+ }
+
+ public void setBytesOut(Long bytesOut) {
+ this.bytesOut = bytesOut;
+ }
+
+ @ApiModelProperty("The number of FlowFiles transferred in this ProcessGroup in the last 5 minutes")
+ public Integer getFlowFilesTransferred() {
+ return flowFilesTransferred;
+ }
+
+ public void setFlowFilesTransferred(Integer flowFilesTransferred) {
+ this.flowFilesTransferred = flowFilesTransferred;
+ }
+
+ @ApiModelProperty("The number of bytes transferred in this ProcessGroup in the last 5 minutes")
+ public Long getBytesTransferred() {
+ return bytesTransferred;
+ }
+
+ public void setBytesTransferred(Long bytesTransferred) {
+ this.bytesTransferred = bytesTransferred;
+ }
+
+ @ApiModelProperty("The number of bytes received from external sources by components within this ProcessGroup in the last 5 minutes")
+ public Long getBytesReceived() {
+ return bytesReceived;
+ }
+
+ public void setBytesReceived(Long bytesReceived) {
+ this.bytesReceived = bytesReceived;
+ }
+
+ @ApiModelProperty("The number of bytes sent to an external sink by components within this ProcessGroup in the last 5 minutes")
+ public Long getBytesSent() {
+ return bytesSent;
+ }
+
+ public void setBytesSent(Long bytesSent) {
+ this.bytesSent = bytesSent;
+ }
+
+ @ApiModelProperty("The number of FlowFiles sent to an external sink by components within this ProcessGroup in the last 5 minutes")
+ public Integer getFlowFilesSent() {
+ return flowFilesSent;
+ }
+
+ public void setFlowFilesSent(Integer flowFilesSent) {
+ this.flowFilesSent = flowFilesSent;
+ }
+
+ @ApiModelProperty("The number of FlowFiles received from external sources by components within this ProcessGroup in the last 5 minutes")
+ public Integer getFlowFilesReceived() {
+ return flowFilesReceived;
+ }
+
+ public void setFlowFilesReceived(Integer flowFilesReceived) {
+ this.flowFilesReceived = flowFilesReceived;
}
/**
@@ -368,4 +450,110 @@ public class ProcessGroupStatusDTO extends StatusDTO {
this.statsLastRefreshed = statsLastRefreshed;
}
+
+ public void setInput(String input) {
+ this.input = input;
+ }
+
+ public void setQueued(String queued) {
+ this.queued = queued;
+ }
+
+ public void setQueuedCount(String queuedCount) {
+ this.queuedCount = queuedCount;
+ }
+
+ public void setQueuedSize(String queuedSize) {
+ this.queuedSize = queuedSize;
+ }
+
+ public void setRead(String read) {
+ this.read = read;
+ }
+
+ public void setWritten(String written) {
+ this.written = written;
+ }
+
+ public void setOutput(String output) {
+ this.output = output;
+ }
+
+ public void setTransferred(String transferred) {
+ this.transferred = transferred;
+ }
+
+ public void setReceived(String received) {
+ this.received = received;
+ }
+
+ public void setSent(String sent) {
+ this.sent = sent;
+ }
+
+ @Override
+ public ProcessGroupStatusDTO clone() {
+ final ProcessGroupStatusDTO other = new ProcessGroupStatusDTO();
+ other.setId(getId());
+ other.setName(getName());
+
+ other.setBytesIn(getBytesIn());
+ other.setFlowFilesIn(getFlowFilesIn());
+ other.setInput(getInput());
+
+ other.setBytesQueued(getBytesQueued());
+ other.setFlowFilesQueued(getFlowFilesQueued());
+ other.setQueued(getQueued());
+ other.setQueuedCount(getQueuedCount());
+ other.setQueuedSize(getQueuedSize());
+
+ other.setBytesRead(getBytesRead());
+ other.setRead(getRead());
+ other.setBytesWritten(getBytesWritten());
+ other.setWritten(getWritten());
+
+ other.setBytesOut(getBytesOut());
+ other.setFlowFilesOut(getFlowFilesOut());
+ other.setOutput(getOutput());
+
+ other.setBytesTransferred(getBytesTransferred());
+ other.setFlowFilesTransferred(getFlowFilesTransferred());
+ other.setTransferred(getTransferred());
+
+ other.setBytesReceived(getBytesReceived());
+ other.setFlowFilesReceived(getFlowFilesReceived());
+ other.setReceived(getReceived());
+ other.setBytesSent(getBytesSent());
+ other.setFlowFilesSent(getFlowFilesSent());
+ other.setSent(getSent());
+
+ other.setActiveThreadCount(getActiveThreadCount());
+ other.setStatsLastRefreshed(getStatsLastRefreshed());
+
+ other.setConnectionStatus(copy(getConnectionStatus()));
+ other.setProcessorStatus(copy(getProcessorStatus()));
+ other.setRemoteProcessGroupStatus(copy(getRemoteProcessGroupStatus()));
+ other.setInputPortStatus(copy(getInputPortStatus()));
+ other.setOutputPortStatus(copy(getOutputPortStatus()));
+
+ if (processGroupStatus != null) {
+ final List<ProcessGroupStatusDTO> childGroups = new ArrayList<>();
+ for (final ProcessGroupStatusDTO procGroupStatus : processGroupStatus) {
+ childGroups.add(procGroupStatus.clone());
+ }
+ other.setProcessGroupStatus(childGroups);
+ }
+
+ other.setBulletins(cloneBulletins());
+
+ return other;
+ }
+
+ private <T> Collection<T> copy(final Collection<T> original) {
+ if (original == null) {
+ return null;
+ }
+
+ return new ArrayList<T>(original);
+ }
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/ad32cb82/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/status/ProcessorStatusDTO.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/status/ProcessorStatusDTO.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/status/ProcessorStatusDTO.java
index 1899418..f1b26db 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/status/ProcessorStatusDTO.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/status/ProcessorStatusDTO.java
@@ -16,14 +16,15 @@
*/
package org.apache.nifi.web.api.dto.status;
-import com.wordnik.swagger.annotations.ApiModelProperty;
import javax.xml.bind.annotation.XmlType;
+import com.wordnik.swagger.annotations.ApiModelProperty;
+
/**
* DTO for serializing the status of a processor.
*/
@XmlType(name = "processorStatus")
-public class ProcessorStatusDTO extends StatusDTO {
+public class ProcessorStatusDTO extends StatusDTO implements Cloneable {
private String id;
private String groupId;
@@ -31,23 +32,30 @@ public class ProcessorStatusDTO extends StatusDTO {
private String type;
private String runStatus;
+ private Long bytesRead = 0L;
+ private Long bytesWritten = 0L;
private String read;
private String written;
+ private Integer flowFilesIn = 0;
+ private Long bytesIn = 0L;
private String input;
+
+ private Integer flowFilesOut = 0;
+ private Long bytesOut = 0L;
private String output;
+ private Integer taskCount = 0;
+ private Long taskDurationNanos = 0L;
private String tasks;
private String tasksDuration;
- private Integer activeThreadCount;
+ private Integer activeThreadCount = 0;
/* getters / setters */
/**
* @return The processor id
*/
- @ApiModelProperty(
- value = "The id of the processor."
- )
+ @ApiModelProperty("The id of the processor.")
public String getId() {
return id;
}
@@ -59,9 +67,7 @@ public class ProcessorStatusDTO extends StatusDTO {
/**
* @return The processor name
*/
- @ApiModelProperty(
- value = "The name of the prcessor."
- )
+ @ApiModelProperty("The name of the prcessor.")
public String getName() {
return name;
}
@@ -73,9 +79,7 @@ public class ProcessorStatusDTO extends StatusDTO {
/**
* @return The processor type
*/
- @ApiModelProperty(
- value = "The type of the processor."
- )
+ @ApiModelProperty("The type of the processor.")
public String getType() {
return type;
}
@@ -102,9 +106,7 @@ public class ProcessorStatusDTO extends StatusDTO {
/**
* @return The total count and size of flow files that have been accepted in the last five minutes
*/
- @ApiModelProperty(
- value = "The count/size of flowfiles that have been accepted in the last 5 minutes."
- )
+ @ApiModelProperty("The count/size of flowfiles that have been accepted in the last 5 minutes.")
public String getInput() {
return input;
}
@@ -116,9 +118,7 @@ public class ProcessorStatusDTO extends StatusDTO {
/**
* @return number of bytes read
*/
- @ApiModelProperty(
- value = "The number of bytes read in the last 5 minutes."
- )
+ @ApiModelProperty("The number of bytes read in the last 5 minutes.")
public String getRead() {
return read;
}
@@ -130,9 +130,7 @@ public class ProcessorStatusDTO extends StatusDTO {
/**
* @return number of bytes written
*/
- @ApiModelProperty(
- value = "The number of bytes written in the last 5 minutes."
- )
+ @ApiModelProperty("The number of bytes written in the last 5 minutes.")
public String getWritten() {
return written;
}
@@ -144,9 +142,7 @@ public class ProcessorStatusDTO extends StatusDTO {
/**
* @return the ID of the Process Group to which this processor belongs.
*/
- @ApiModelProperty(
- value = "The id of the parent process group to which the processor belongs."
- )
+ @ApiModelProperty("The id of the parent process group to which the processor belongs.")
public String getGroupId() {
return groupId;
}
@@ -158,9 +154,7 @@ public class ProcessorStatusDTO extends StatusDTO {
/**
* @return The total count and size of flow files that have been processed in the last five minutes
*/
- @ApiModelProperty(
- value = "The count/size of flowfiles that have been processed in the last 5 minutes."
- )
+ @ApiModelProperty("The count/size of flowfiles that have been processed in the last 5 minutes.")
public String getOutput() {
return output;
}
@@ -172,9 +166,7 @@ public class ProcessorStatusDTO extends StatusDTO {
/**
* @return number of threads currently running for this Processor
*/
- @ApiModelProperty(
- value = "The number of threads currently executing in the processor."
- )
+ @ApiModelProperty("The number of threads currently executing in the processor.")
public Integer getActiveThreadCount() {
return activeThreadCount;
}
@@ -186,9 +178,7 @@ public class ProcessorStatusDTO extends StatusDTO {
/**
* @return number of task this connectable has had over the last 5 minutes
*/
- @ApiModelProperty(
- value = "The total number of task this connectable has completed over the last 5 minutes."
- )
+ @ApiModelProperty("The total number of task this connectable has completed over the last 5 minutes.")
public String getTasks() {
return tasks;
}
@@ -200,9 +190,7 @@ public class ProcessorStatusDTO extends StatusDTO {
/**
* @return total duration of all tasks for this connectable over the last 5 minutes
*/
- @ApiModelProperty(
- value = "The total duration of all tasks for this connectable over the last 5 minutes."
- )
+ @ApiModelProperty("The total duration of all tasks for this connectable over the last 5 minutes.")
public String getTasksDuration() {
return tasksDuration;
}
@@ -211,4 +199,103 @@ public class ProcessorStatusDTO extends StatusDTO {
this.tasksDuration = tasksDuration;
}
+ @ApiModelProperty("The number of bytes read by this Processor in the last 5 mintues")
+ public Long getBytesRead() {
+ return bytesRead;
+ }
+
+ public void setBytesRead(Long bytesRead) {
+ this.bytesRead = bytesRead;
+ }
+
+ @ApiModelProperty("The number of bytes written by this Processor in the last 5 minutes")
+ public Long getBytesWritten() {
+ return bytesWritten;
+ }
+
+ public void setBytesWritten(Long bytesWritten) {
+ this.bytesWritten = bytesWritten;
+ }
+
+ @ApiModelProperty("The number of FlowFiles that have been accepted in the last 5 minutes")
+ public Integer getFlowFilesIn() {
+ return flowFilesIn;
+ }
+
+ public void setFlowFilesIn(Integer flowFilesIn) {
+ this.flowFilesIn = flowFilesIn;
+ }
+
+ @ApiModelProperty("The size of the FlowFiles that have been accepted in the last 5 minutes")
+ public Long getBytesIn() {
+ return bytesIn;
+ }
+
+ public void setBytesIn(Long bytesIn) {
+ this.bytesIn = bytesIn;
+ }
+
+ @ApiModelProperty("The number of FlowFiles transferred to a Connection in the last 5 minutes")
+ public Integer getFlowFilesOut() {
+ return flowFilesOut;
+ }
+
+ public void setFlowFilesOut(Integer flowFilesOut) {
+ this.flowFilesOut = flowFilesOut;
+ }
+
+ @ApiModelProperty("The size of the FlowFiles transferred to a Connection in the last 5 minutes")
+ public Long getBytesOut() {
+ return bytesOut;
+ }
+
+ public void setBytesOut(Long bytesOut) {
+ this.bytesOut = bytesOut;
+ }
+
+ @ApiModelProperty("The number of times this Processor has run in the last 5 minutes")
+ public Integer getTaskCount() {
+ return taskCount;
+ }
+
+ public void setTaskCount(Integer taskCount) {
+ this.taskCount = taskCount;
+ }
+
+ @ApiModelProperty("The number of nanoseconds that this Processor has spent running in the last 5 minutes")
+ public Long getTaskDuration() {
+ return taskDurationNanos;
+ }
+
+ public void setTaskDuration(Long taskNanos) {
+ this.taskDurationNanos = taskNanos;
+ }
+
+ @Override
+ public ProcessorStatusDTO clone() {
+ final ProcessorStatusDTO other = new ProcessorStatusDTO();
+ other.setId(getId());
+ other.setGroupId(getGroupId());
+ other.setName(getName());
+ other.setType(getType());
+
+ other.setRunStatus(getRunStatus());
+ other.setBytesRead(getBytesRead());
+ other.setBytesWritten(getBytesWritten());
+ other.setFlowFilesIn(getFlowFilesIn());
+ other.setBytesIn(getBytesIn());
+ other.setFlowFilesOut(getFlowFilesOut());
+ other.setBytesOut(getBytesOut());
+ other.setTaskCount(getTaskCount());
+ other.setTaskDuration(getTaskDuration());
+ other.setActiveThreadCount(getActiveThreadCount());
+ other.setInput(getInput());
+ other.setOutput(getOutput());
+ other.setRead(getRead());
+ other.setWritten(getWritten());
+ other.setTasks(getTasks());
+ other.setBulletins(cloneBulletins());
+
+ return other;
+ }
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/ad32cb82/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/status/RemoteProcessGroupStatusDTO.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/status/RemoteProcessGroupStatusDTO.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/status/RemoteProcessGroupStatusDTO.java
index f556deb..e1dad2a 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/status/RemoteProcessGroupStatusDTO.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/status/RemoteProcessGroupStatusDTO.java
@@ -16,15 +16,18 @@
*/
package org.apache.nifi.web.api.dto.status;
-import com.wordnik.swagger.annotations.ApiModelProperty;
+import java.util.ArrayList;
import java.util.List;
+
import javax.xml.bind.annotation.XmlType;
+import com.wordnik.swagger.annotations.ApiModelProperty;
+
/**
* The status of a remote process group in this NiFi.
*/
@XmlType(name = "remoteProcessGroupStatus")
-public class RemoteProcessGroupStatusDTO extends StatusDTO {
+public class RemoteProcessGroupStatusDTO extends StatusDTO implements Cloneable {
private String id;
private String groupId;
@@ -35,15 +38,18 @@ public class RemoteProcessGroupStatusDTO extends StatusDTO {
private List<String> authorizationIssues;
+ private Integer flowFilesSent = 0;
+ private Long bytesSent = 0L;
private String sent;
+
+ private Integer flowFilesReceived = 0;
+ private Long bytesReceived = 0L;
private String received;
/**
* @return The id for the remote process group
*/
- @ApiModelProperty(
- value = "The id of the remote process group."
- )
+ @ApiModelProperty("The id of the remote process group.")
public String getId() {
return id;
}
@@ -55,9 +61,7 @@ public class RemoteProcessGroupStatusDTO extends StatusDTO {
/**
* @return id of the group this remote process group is in
*/
- @ApiModelProperty(
- value = "The id of the parent process group the remote process group resides in."
- )
+ @ApiModelProperty("The id of the parent process group the remote process group resides in.")
public String getGroupId() {
return groupId;
}
@@ -69,9 +73,7 @@ public class RemoteProcessGroupStatusDTO extends StatusDTO {
/**
* @return URI of the target system
*/
- @ApiModelProperty(
- value = "The URI of the target system."
- )
+ @ApiModelProperty("The URI of the target system.")
public String getTargetUri() {
return targetUri;
}
@@ -83,9 +85,7 @@ public class RemoteProcessGroupStatusDTO extends StatusDTO {
/**
* @return name of this remote process group
*/
- @ApiModelProperty(
- value = "The name of the remote process group."
- )
+ @ApiModelProperty("The name of the remote process group.")
public String getName() {
return name;
}
@@ -97,9 +97,7 @@ public class RemoteProcessGroupStatusDTO extends StatusDTO {
/**
* @return transmission status of this remote process group
*/
- @ApiModelProperty(
- value = "The transmission status of the remote process group."
- )
+ @ApiModelProperty("The transmission status of the remote process group.")
public String getTransmissionStatus() {
return transmissionStatus;
}
@@ -111,9 +109,7 @@ public class RemoteProcessGroupStatusDTO extends StatusDTO {
/**
* @return number of active threads
*/
- @ApiModelProperty(
- value = "The number of active threads for the remote process group."
- )
+ @ApiModelProperty("The number of active threads for the remote process group.")
public Integer getActiveThreadCount() {
return activeThreadCount;
}
@@ -125,9 +121,7 @@ public class RemoteProcessGroupStatusDTO extends StatusDTO {
/**
* @return any remote authorization issues for this remote process group
*/
- @ApiModelProperty(
- value = "Any remote authorization issues for the remote process group."
- )
+ @ApiModelProperty("Any remote authorization issues for the remote process group.")
public List<String> getAuthorizationIssues() {
return authorizationIssues;
}
@@ -139,9 +133,7 @@ public class RemoteProcessGroupStatusDTO extends StatusDTO {
/**
* @return Formatted description of the amount of data sent to this remote process group
*/
- @ApiModelProperty(
- value = "The count/size of the flowfiles sent to the remote process group in the last 5 minutes."
- )
+ @ApiModelProperty("The count/size of the flowfiles sent to the remote process group in the last 5 minutes.")
public String getSent() {
return sent;
}
@@ -150,12 +142,11 @@ public class RemoteProcessGroupStatusDTO extends StatusDTO {
this.sent = sent;
}
+
/**
* @return Formatted description of the amount of data received from this remote process group
*/
- @ApiModelProperty(
- value = "The count/size of the flowfiles received from the remote process group in the last 5 minutes."
- )
+ @ApiModelProperty("The count/size of the flowfiles received from the remote process group in the last 5 minutes.")
public String getReceived() {
return received;
}
@@ -164,4 +155,62 @@ public class RemoteProcessGroupStatusDTO extends StatusDTO {
this.received = received;
}
+ @ApiModelProperty("The number of FlowFiles sent to the remote process group in the last 5 minutes.")
+ public Integer getFlowFilesSent() {
+ return flowFilesSent;
+ }
+
+ public void setFlowFilesSent(Integer flowFilesSent) {
+ this.flowFilesSent = flowFilesSent;
+ }
+
+ @ApiModelProperty("The size of the FlowFiles sent to the remote process group in the last 5 minutes.")
+ public Long getBytesSent() {
+ return bytesSent;
+ }
+
+ public void setBytesSent(Long bytesSent) {
+ this.bytesSent = bytesSent;
+ }
+
+ @ApiModelProperty("The number of FlowFiles received from the remote process group in the last 5 minutes.")
+ public Integer getFlowFilesReceived() {
+ return flowFilesReceived;
+ }
+
+ public void setFlowFilesReceived(Integer flowFilesReceived) {
+ this.flowFilesReceived = flowFilesReceived;
+ }
+
+ @ApiModelProperty("The size of the FlowFiles received from the remote process group in the last 5 minutes.")
+ public Long getBytesReceived() {
+ return bytesReceived;
+ }
+
+ public void setBytesReceived(Long bytesReceived) {
+ this.bytesReceived = bytesReceived;
+ }
+
+
+ @Override
+ public RemoteProcessGroupStatusDTO clone() {
+ final RemoteProcessGroupStatusDTO other = new RemoteProcessGroupStatusDTO();
+ other.setId(getId());
+ other.setGroupId(getGroupId());
+ other.setName(getName());
+ other.setTargetUri(getTargetUri());
+ other.setTransmissionStatus(getTransmissionStatus());
+ other.setActiveThreadCount(getActiveThreadCount());
+ other.setAuthorizationIssues(getAuthorizationIssues() == null ? null : new ArrayList<String>(getAuthorizationIssues()));
+ other.setFlowFilesSent(getFlowFilesSent());
+ other.setBytesSent(getBytesSent());
+ other.setFlowFilesReceived(getFlowFilesReceived());
+ other.setBytesReceived(getBytesReceived());
+ other.setReceived(getReceived());
+ other.setSent(getSent());
+ other.setBulletins(cloneBulletins());
+
+ return other;
+ }
+
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/ad32cb82/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/status/StatusDTO.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/status/StatusDTO.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/status/StatusDTO.java
index 39b9c06..ad6d64f 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/status/StatusDTO.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/status/StatusDTO.java
@@ -16,11 +16,15 @@
*/
package org.apache.nifi.web.api.dto.status;
-import com.wordnik.swagger.annotations.ApiModelProperty;
+import java.util.ArrayList;
import java.util.List;
+
import javax.xml.bind.annotation.XmlType;
+
import org.apache.nifi.web.api.dto.BulletinDTO;
+import com.wordnik.swagger.annotations.ApiModelProperty;
+
/**
* The status of a component in this NiFi.
*/
@@ -42,4 +46,12 @@ public abstract class StatusDTO {
public void setBulletins(List<BulletinDTO> bulletins) {
this.bulletins = bulletins;
}
+
+ public List<BulletinDTO> cloneBulletins() {
+ if (bulletins == null) {
+ return null;
+ }
+
+ return new ArrayList<>(bulletins);
+ }
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/ad32cb82/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/status/StatusHistoryDTO.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/status/StatusHistoryDTO.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/status/StatusHistoryDTO.java
index c0ef33b..4a9e1e3 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/status/StatusHistoryDTO.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/status/StatusHistoryDTO.java
@@ -16,34 +16,37 @@
*/
package org.apache.nifi.web.api.dto.status;
-import com.wordnik.swagger.annotations.ApiModelProperty;
import java.util.Date;
import java.util.LinkedHashMap;
import java.util.List;
+import java.util.Map;
+
import javax.xml.bind.annotation.XmlType;
import javax.xml.bind.annotation.adapters.XmlJavaTypeAdapter;
+
import org.apache.nifi.web.api.dto.util.TimeAdapter;
+import com.wordnik.swagger.annotations.ApiModelProperty;
+
/**
- * History status for a component in this NiFi.
+ * DTO for serializing the status history of a single component across the cluster.
*/
@XmlType(name = "statusHistory")
public class StatusHistoryDTO {
private Date generated;
- private LinkedHashMap<String, String> details;
-
+ private LinkedHashMap<String, String> componentDetails;
private List<StatusDescriptorDTO> fieldDescriptors;
- private List<StatusSnapshotDTO> statusSnapshots;
+ private List<StatusSnapshotDTO> aggregateStatusSnapshots;
+ private Map<String, List<StatusSnapshotDTO>> nodeStatusSnapshots;
+
/**
* @return when this status history was generated
*/
@XmlJavaTypeAdapter(TimeAdapter.class)
- @ApiModelProperty(
- value = "The timestamp when the status history was generated."
- )
+ @ApiModelProperty("When the status history was generated.")
public Date getGenerated() {
return generated;
}
@@ -53,25 +56,18 @@ public class StatusHistoryDTO {
}
/**
- * @return The component details for this status history
+ * @return key/value pairs that describe the component that the status history belongs to
*/
- @ApiModelProperty(
- value = "The component details for the status history."
- )
- public LinkedHashMap<String, String> getDetails() {
- return details;
+ @ApiModelProperty("A Map of key/value pairs that describe the component that the status history belongs to")
+ public LinkedHashMap<String, String> getComponentDetails() {
+ return componentDetails;
}
- public void setDetails(LinkedHashMap<String, String> details) {
- this.details = details;
+ public void setComponentDetails(LinkedHashMap<String, String> componentDetails) {
+ this.componentDetails = componentDetails;
}
- /**
- * @return Descriptors for each supported status field
- */
- @ApiModelProperty(
- value = "The descriptor for each support status field."
- )
+ @ApiModelProperty("The Descriptors that provide information on each of the metrics provided in the status history")
public List<StatusDescriptorDTO> getFieldDescriptors() {
return fieldDescriptors;
}
@@ -80,18 +76,24 @@ public class StatusHistoryDTO {
this.fieldDescriptors = fieldDescriptors;
}
- /**
- * @return The status snapshots
- */
- @ApiModelProperty(
- value = "The status snapshots."
- )
- public List<StatusSnapshotDTO> getStatusSnapshots() {
- return statusSnapshots;
+ @ApiModelProperty("A list of StatusSnapshotDTO objects that provide the actual metric values for the component. If the NiFi instance "
+ + "is clustered, this will represent the aggregate status across all nodes. If the NiFi instance is not clustered, this will represent "
+ + "the status of the entire NiFi instance.")
+ public List<StatusSnapshotDTO> getAggregateStatusSnapshots() {
+ return aggregateStatusSnapshots;
}
- public void setStatusSnapshots(List<StatusSnapshotDTO> statusSnapshots) {
- this.statusSnapshots = statusSnapshots;
+ public void setAggregateStatusSnapshots(List<StatusSnapshotDTO> aggregateStatusSnapshots) {
+ this.aggregateStatusSnapshots = aggregateStatusSnapshots;
}
+ @ApiModelProperty("A Map of Node Address to a list of StatusSnapshotDTO objects that provide the actual metric values for the component, for that node. "
+ + "If the NiFi instance is not clustered, this value will be null.")
+ public Map<String, List<StatusSnapshotDTO>> getNodeStatusSnapshots() {
+ return nodeStatusSnapshots;
+ }
+
+ public void setNodeStatusSnapshots(Map<String, List<StatusSnapshotDTO>> nodeStatusSnapshots) {
+ this.nodeStatusSnapshots = nodeStatusSnapshots;
+ }
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/ad32cb82/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/status/StatusHistoryDetailDTO.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/status/StatusHistoryDetailDTO.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/status/StatusHistoryDetailDTO.java
deleted file mode 100644
index e78641e..0000000
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-client-dto/src/main/java/org/apache/nifi/web/api/dto/status/StatusHistoryDetailDTO.java
+++ /dev/null
@@ -1,59 +0,0 @@
-/*
- * 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.dto.status;
-
-import com.wordnik.swagger.annotations.ApiModelProperty;
-import javax.xml.bind.annotation.XmlType;
-
-/**
- * Detail of a status history metric.
- */
-@XmlType(name = "statusHistoryDetail")
-public class StatusHistoryDetailDTO {
-
- private String label;
- private String value;
-
- /**
- * @return label for this status detail
- */
- @ApiModelProperty(
- value = "The label for the status detail."
- )
- public String getLabel() {
- return label;
- }
-
- public void setLabel(String label) {
- this.label = label;
- }
-
- /**
- * @return value for this status detail
- */
- @ApiModelProperty(
- value = "The value for the status detail."
- )
- public String getValue() {
- return value;
- }
-
- public void setValue(String value) {
- this.value = value;
- }
-
-}
[2/5] nifi git commit: NIFI-1563: Federate requests and merge
responses from nodes instead of storing bulletins and stats at NCM
Posted by ma...@apache.org.
http://git-wip-us.apache.org/repos/asf/nifi/blob/ad32cb82/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/ProcessorStatusDescriptor.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/ProcessorStatusDescriptor.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/ProcessorStatusDescriptor.java
new file mode 100644
index 0000000..89e8aa0
--- /dev/null
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/ProcessorStatusDescriptor.java
@@ -0,0 +1,220 @@
+/*
+ * 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.controller.status.history;
+
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.nifi.controller.status.ProcessorStatus;
+import org.apache.nifi.controller.status.history.MetricDescriptor.Formatter;
+
+public enum ProcessorStatusDescriptor {
+ BYTES_READ(new StandardMetricDescriptor<ProcessorStatus>(
+ "bytesRead",
+ "Bytes Read (5 mins)",
+ "The total number of bytes read from the Content Repository by this Processor in the past 5 minutes",
+ Formatter.DATA_SIZE,
+ new ValueMapper<ProcessorStatus>() {
+ @Override
+ public Long getValue(final ProcessorStatus status) {
+ return status.getBytesRead();
+ }
+ })),
+
+ BYTES_WRITTEN(new StandardMetricDescriptor<ProcessorStatus>(
+ "bytesWritten",
+ "Bytes Written (5 mins)",
+ "The total number of bytes written to the Content Repository by this Processor in the past 5 minutes",
+ Formatter.DATA_SIZE,
+ new ValueMapper<ProcessorStatus>() {
+ @Override
+ public Long getValue(final ProcessorStatus status) {
+ return status.getBytesWritten();
+ }
+ })),
+
+ BYTES_TRANSFERRED(new StandardMetricDescriptor<ProcessorStatus>(
+ "bytesTransferred",
+ "Bytes Transferred (5 mins)",
+ "The total number of bytes read from or written to the Content Repository by this Processor in the past 5 minutes",
+ Formatter.DATA_SIZE,
+ new ValueMapper<ProcessorStatus>() {
+ @Override
+ public Long getValue(final ProcessorStatus status) {
+ return status.getBytesRead() + status.getBytesWritten();
+ }
+ })),
+
+ INPUT_BYTES(new StandardMetricDescriptor<ProcessorStatus>(
+ "inputBytes",
+ "Bytes In (5 mins)",
+ "The cumulative size of all FlowFiles that this Processor has pulled from its queues in the past 5 minutes",
+ Formatter.DATA_SIZE,
+ new ValueMapper<ProcessorStatus>() {
+ @Override
+ public Long getValue(final ProcessorStatus status) {
+ return status.getInputBytes();
+ }
+ })),
+
+ INPUT_COUNT(new StandardMetricDescriptor<ProcessorStatus>(
+ "inputCount",
+ "FlowFiles In (5 mins)",
+ "The number of FlowFiles that this Processor has pulled from its queues in the past 5 minutes",
+ Formatter.COUNT, new ValueMapper<ProcessorStatus>() {
+ @Override
+ public Long getValue(final ProcessorStatus status) {
+ return Long.valueOf(status.getInputCount());
+ }
+ })),
+ OUTPUT_BYTES(new StandardMetricDescriptor<ProcessorStatus>(
+ "outputBytes",
+ "Bytes Out (5 mins)",
+ "The cumulative size of all FlowFiles that this Processor has transferred to downstream queues in the past 5 minutes",
+ Formatter.DATA_SIZE,
+ new ValueMapper<ProcessorStatus>() {
+ @Override
+ public Long getValue(final ProcessorStatus status) {
+ return status.getOutputBytes();
+ }
+ })),
+
+ OUTPUT_COUNT(new StandardMetricDescriptor<ProcessorStatus>(
+ "outputCount",
+ "FlowFiles Out (5 mins)",
+ "The number of FlowFiles that this Processor has transferred to downstream queues in the past 5 minutes",
+ Formatter.COUNT,
+ new ValueMapper<ProcessorStatus>() {
+ @Override
+ public Long getValue(final ProcessorStatus status) {
+ return Long.valueOf(status.getOutputCount());
+ }
+ })),
+
+ TASK_COUNT(new StandardMetricDescriptor<ProcessorStatus>(
+ "taskCount",
+ "Tasks (5 mins)",
+ "The number of tasks that this Processor has completed in the past 5 minutes",
+ Formatter.COUNT,
+ new ValueMapper<ProcessorStatus>() {
+ @Override
+ public Long getValue(final ProcessorStatus status) {
+ return Long.valueOf(status.getInvocations());
+ }
+ })),
+
+ TASK_MILLIS(new StandardMetricDescriptor<ProcessorStatus>(
+ "taskMillis",
+ "Total Task Duration (5 mins)",
+ "The total number of thread-milliseconds that the Processor has used to complete its tasks in the past 5 minutes",
+ Formatter.DURATION,
+ new ValueMapper<ProcessorStatus>() {
+ @Override
+ public Long getValue(final ProcessorStatus status) {
+ return TimeUnit.MILLISECONDS.convert(status.getProcessingNanos(), TimeUnit.NANOSECONDS);
+ }
+ })),
+
+ FLOWFILES_REMOVED(new StandardMetricDescriptor<ProcessorStatus>(
+ "flowFilesRemoved",
+ "FlowFiles Removed (5 mins)",
+ "The total number of FlowFiles removed by this Processor in the last 5 minutes",
+ Formatter.COUNT,
+ new ValueMapper<ProcessorStatus>() {
+ @Override
+ public Long getValue(final ProcessorStatus status) {
+ return Long.valueOf(status.getFlowFilesRemoved());
+ }
+ })),
+
+ AVERAGE_LINEAGE_DURATION(new StandardMetricDescriptor<ProcessorStatus>(
+ "averageLineageDuration",
+ "Average Lineage Duration (5 mins)",
+ "The average amount of time that a FlowFile took to process (from receipt until this Processor finished processing it) in the past 5 minutes.",
+ Formatter.DURATION,
+ new ValueMapper<ProcessorStatus>() {
+ @Override
+ public Long getValue(final ProcessorStatus status) {
+ return status.getAverageLineageDuration(TimeUnit.MILLISECONDS);
+ }
+ }, new ValueReducer<StatusSnapshot, Long>() {
+ @Override
+ public Long reduce(final List<StatusSnapshot> values) {
+ long millis = 0L;
+ int count = 0;
+
+ for (final StatusSnapshot snapshot : values) {
+ final long removed = snapshot.getStatusMetrics().get(FLOWFILES_REMOVED.getDescriptor()).longValue();
+ count += removed;
+
+ count += snapshot.getStatusMetrics().get(OUTPUT_COUNT.getDescriptor()).longValue();
+
+ final long avgMillis = snapshot.getStatusMetrics().get(AVERAGE_LINEAGE_DURATION.getDescriptor()).longValue();
+ final long totalMillis = avgMillis * removed;
+ millis += totalMillis;
+ }
+
+ return count == 0 ? 0 : millis / count;
+ }
+ }
+ )),
+
+ AVERAGE_TASK_MILLIS(new StandardMetricDescriptor<ProcessorStatus>(
+ "averageTaskMillis",
+ "Average Task Duration",
+ "The average duration it took this Processor to complete a task, as averaged over the past 5 minutes",
+ Formatter.DURATION,
+ new ValueMapper<ProcessorStatus>() {
+ @Override
+ public Long getValue(final ProcessorStatus status) {
+ return status.getInvocations() == 0 ? 0 : TimeUnit.MILLISECONDS.convert(status.getProcessingNanos(), TimeUnit.NANOSECONDS) / status.getInvocations();
+ }
+ },
+ new ValueReducer<StatusSnapshot, Long>() {
+ @Override
+ public Long reduce(final List<StatusSnapshot> values) {
+ long procMillis = 0L;
+ int invocations = 0;
+
+ for (final StatusSnapshot snapshot : values) {
+ procMillis += snapshot.getStatusMetrics().get(TASK_MILLIS.getDescriptor()).longValue();
+ invocations += snapshot.getStatusMetrics().get(TASK_COUNT.getDescriptor()).intValue();
+ }
+
+ if (invocations == 0) {
+ return 0L;
+ }
+
+ return procMillis / invocations;
+ }
+ }));
+
+ private MetricDescriptor<ProcessorStatus> descriptor;
+
+ private ProcessorStatusDescriptor(final MetricDescriptor<ProcessorStatus> descriptor) {
+ this.descriptor = descriptor;
+ }
+
+ public String getField() {
+ return descriptor.getField();
+ }
+
+ public MetricDescriptor<ProcessorStatus> getDescriptor() {
+ return descriptor;
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/ad32cb82/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/RemoteProcessGroupStatusDescriptor.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/RemoteProcessGroupStatusDescriptor.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/RemoteProcessGroupStatusDescriptor.java
new file mode 100644
index 0000000..0499d65
--- /dev/null
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/RemoteProcessGroupStatusDescriptor.java
@@ -0,0 +1,127 @@
+/*
+ * 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.controller.status.history;
+
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.nifi.controller.status.RemoteProcessGroupStatus;
+import org.apache.nifi.controller.status.history.MetricDescriptor.Formatter;
+
+public enum RemoteProcessGroupStatusDescriptor {
+ SENT_BYTES(new StandardMetricDescriptor<RemoteProcessGroupStatus>("sentBytes", "Bytes Sent (5 mins)",
+ "The cumulative size of all FlowFiles that have been successfully sent to the remote system in the past 5 minutes", Formatter.DATA_SIZE, new ValueMapper<RemoteProcessGroupStatus>() {
+ @Override
+ public Long getValue(final RemoteProcessGroupStatus status) {
+ return status.getSentContentSize();
+ }
+ })),
+
+ SENT_COUNT(new StandardMetricDescriptor<RemoteProcessGroupStatus>("sentCount", "FlowFiles Sent (5 mins)",
+ "The number of FlowFiles that have been successfully sent to the remote system in the past 5 minutes", Formatter.COUNT, new ValueMapper<RemoteProcessGroupStatus>() {
+ @Override
+ public Long getValue(final RemoteProcessGroupStatus status) {
+ return Long.valueOf(status.getSentCount().longValue());
+ }
+ })),
+
+ RECEIVED_BYTES(new StandardMetricDescriptor<RemoteProcessGroupStatus>("receivedBytes", "Bytes Received (5 mins)",
+ "The cumulative size of all FlowFiles that have been received from the remote system in the past 5 minutes", Formatter.DATA_SIZE, new ValueMapper<RemoteProcessGroupStatus>() {
+ @Override
+ public Long getValue(final RemoteProcessGroupStatus status) {
+ return status.getReceivedContentSize();
+ }
+ })),
+
+ RECEIVED_COUNT(new StandardMetricDescriptor<RemoteProcessGroupStatus>("receivedCount", "FlowFiles Received (5 mins)",
+ "The number of FlowFiles that have been received from the remote system in the past 5 minutes", Formatter.COUNT, new ValueMapper<RemoteProcessGroupStatus>() {
+ @Override
+ public Long getValue(final RemoteProcessGroupStatus status) {
+ return Long.valueOf(status.getReceivedCount().longValue());
+ }
+ })),
+
+ RECEIVED_BYTES_PER_SECOND(new StandardMetricDescriptor<RemoteProcessGroupStatus>("receivedBytesPerSecond", "Received Bytes Per Second",
+ "The data rate at which data was received from the remote system in the past 5 minutes in terms of Bytes Per Second",
+ Formatter.DATA_SIZE, new ValueMapper<RemoteProcessGroupStatus>() {
+ @Override
+ public Long getValue(final RemoteProcessGroupStatus status) {
+ return Long.valueOf(status.getReceivedContentSize().longValue() / 300L);
+ }
+ })),
+
+ SENT_BYTES_PER_SECOND(new StandardMetricDescriptor<RemoteProcessGroupStatus>("sentBytesPerSecond", "Sent Bytes Per Second",
+ "The data rate at which data was received from the remote system in the past 5 minutes in terms of Bytes Per Second", Formatter.DATA_SIZE, new ValueMapper<RemoteProcessGroupStatus>() {
+ @Override
+ public Long getValue(final RemoteProcessGroupStatus status) {
+ return Long.valueOf(status.getSentContentSize().longValue() / 300L);
+ }
+ })),
+
+ TOTAL_BYTES_PER_SECOND(new StandardMetricDescriptor<RemoteProcessGroupStatus>("totalBytesPerSecond", "Total Bytes Per Second",
+ "The sum of the send and receive data rate from the remote system in the past 5 minutes in terms of Bytes Per Second",
+ Formatter.DATA_SIZE, new ValueMapper<RemoteProcessGroupStatus>() {
+ @Override
+ public Long getValue(final RemoteProcessGroupStatus status) {
+ return Long.valueOf((status.getReceivedContentSize().longValue() + status.getSentContentSize().longValue()) / 300L);
+ }
+ })),
+
+ AVERAGE_LINEAGE_DURATION(new StandardMetricDescriptor<RemoteProcessGroupStatus>(
+ "averageLineageDuration",
+ "Average Lineage Duration (5 mins)",
+ "The average amount of time that a FlowFile took to process from receipt to drop in the past 5 minutes. For Processors that do not terminate FlowFiles, this value will be 0.",
+ Formatter.DURATION,
+ new ValueMapper<RemoteProcessGroupStatus>() {
+ @Override
+ public Long getValue(final RemoteProcessGroupStatus status) {
+ return status.getAverageLineageDuration(TimeUnit.MILLISECONDS);
+ }
+ }, new ValueReducer<StatusSnapshot, Long>() {
+ @Override
+ public Long reduce(final List<StatusSnapshot> values) {
+ long millis = 0L;
+ int count = 0;
+
+ for (final StatusSnapshot snapshot : values) {
+ final long sent = snapshot.getStatusMetrics().get(SENT_COUNT.getDescriptor()).longValue();
+ count += sent;
+
+ final long avgMillis = snapshot.getStatusMetrics().get(AVERAGE_LINEAGE_DURATION.getDescriptor()).longValue();
+ final long totalMillis = avgMillis * sent;
+ millis += totalMillis;
+ }
+
+ return count == 0 ? 0 : millis / count;
+ }
+ }));
+
+ private final MetricDescriptor<RemoteProcessGroupStatus> descriptor;
+
+ private RemoteProcessGroupStatusDescriptor(final MetricDescriptor<RemoteProcessGroupStatus> descriptor) {
+ this.descriptor = descriptor;
+ }
+
+ public String getField() {
+ return descriptor.getField();
+ }
+
+ public MetricDescriptor<RemoteProcessGroupStatus> getDescriptor() {
+ return descriptor;
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/ad32cb82/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/StatusHistoryUtil.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/StatusHistoryUtil.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/StatusHistoryUtil.java
index 014b0a6..2b0cd9e 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/StatusHistoryUtil.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/StatusHistoryUtil.java
@@ -44,9 +44,9 @@ public class StatusHistoryUtil {
final StatusHistoryDTO dto = new StatusHistoryDTO();
dto.setGenerated(new Date());
- dto.setDetails(componentDetails);
+ dto.setComponentDetails(componentDetails);
dto.setFieldDescriptors(StatusHistoryUtil.createFieldDescriptorDtos(metricDescriptors));
- dto.setStatusSnapshots(snapshotDtos);
+ dto.setAggregateStatusSnapshots(snapshotDtos);
return dto;
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/ad32cb82/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/VolatileComponentStatusRepository.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/VolatileComponentStatusRepository.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/VolatileComponentStatusRepository.java
index 02f7d6b..36288d5 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/VolatileComponentStatusRepository.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/VolatileComponentStatusRepository.java
@@ -20,19 +20,16 @@ import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.List;
-import java.util.concurrent.TimeUnit;
import org.apache.nifi.controller.status.ConnectionStatus;
import org.apache.nifi.controller.status.ProcessGroupStatus;
import org.apache.nifi.controller.status.ProcessorStatus;
import org.apache.nifi.controller.status.RemoteProcessGroupStatus;
-import org.apache.nifi.controller.status.history.MetricDescriptor.Formatter;
import org.apache.nifi.util.ComponentStatusReport;
-import org.apache.nifi.util.NiFiProperties;
import org.apache.nifi.util.ComponentStatusReport.ComponentType;
+import org.apache.nifi.util.NiFiProperties;
import org.apache.nifi.util.RingBuffer;
import org.apache.nifi.util.RingBuffer.ForEachEvaluator;
-
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -233,22 +230,9 @@ public class VolatileComponentStatusRepository implements ComponentStatusReposit
return history;
}
- private static long calculateTaskMillis(final ProcessGroupStatus status) {
- long nanos = 0L;
- for (final ProcessorStatus procStatus : status.getProcessorStatus()) {
- nanos += procStatus.getProcessingNanos();
- }
-
- for (final ProcessGroupStatus childStatus : status.getProcessGroupStatus()) {
- nanos += calculateTaskMillis(childStatus);
- }
-
- return TimeUnit.MILLISECONDS.convert(nanos, TimeUnit.NANOSECONDS);
- }
private static class Capture {
-
private final Date captureDate;
private final ComponentStatusReport statusReport;
@@ -266,407 +250,7 @@ public class VolatileComponentStatusRepository implements ComponentStatusReposit
}
}
- public static enum RemoteProcessGroupStatusDescriptor {
-
- SENT_BYTES(new StandardMetricDescriptor<RemoteProcessGroupStatus>("sentBytes", "Bytes Sent (5 mins)",
- "The cumulative size of all FlowFiles that have been successfully sent to the remote system in the past 5 minutes", Formatter.DATA_SIZE, new ValueMapper<RemoteProcessGroupStatus>() {
- @Override
- public Long getValue(final RemoteProcessGroupStatus status) {
- return status.getSentContentSize();
- }
- })),
- SENT_COUNT(new StandardMetricDescriptor<RemoteProcessGroupStatus>("sentCount", "FlowFiles Sent (5 mins)",
- "The number of FlowFiles that have been successfully sent to the remote system in the past 5 minutes", Formatter.COUNT, new ValueMapper<RemoteProcessGroupStatus>() {
- @Override
- public Long getValue(final RemoteProcessGroupStatus status) {
- return Long.valueOf(status.getSentCount().longValue());
- }
- })),
- RECEIVED_BYTES(new StandardMetricDescriptor<RemoteProcessGroupStatus>("receivedBytes", "Bytes Received (5 mins)",
- "The cumulative size of all FlowFiles that have been received from the remote system in the past 5 minutes", Formatter.DATA_SIZE, new ValueMapper<RemoteProcessGroupStatus>() {
- @Override
- public Long getValue(final RemoteProcessGroupStatus status) {
- return status.getReceivedContentSize();
- }
- })),
- RECEIVED_COUNT(new StandardMetricDescriptor<RemoteProcessGroupStatus>("receivedCount", "FlowFiles Received (5 mins)",
- "The number of FlowFiles that have been received from the remote system in the past 5 minutes", Formatter.COUNT, new ValueMapper<RemoteProcessGroupStatus>() {
- @Override
- public Long getValue(final RemoteProcessGroupStatus status) {
- return Long.valueOf(status.getReceivedCount().longValue());
- }
- })),
- RECEIVED_BYTES_PER_SECOND(new StandardMetricDescriptor<RemoteProcessGroupStatus>("receivedBytesPerSecond", "Received Bytes Per Second",
- "The data rate at which data was received from the remote system in the past 5 minutes in terms of Bytes Per Second",
- Formatter.DATA_SIZE, new ValueMapper<RemoteProcessGroupStatus>() {
- @Override
- public Long getValue(final RemoteProcessGroupStatus status) {
- return Long.valueOf(status.getReceivedContentSize().longValue() / 300L);
- }
- })),
- SENT_BYTES_PER_SECOND(new StandardMetricDescriptor<RemoteProcessGroupStatus>("sentBytesPerSecond", "Sent Bytes Per Second",
- "The data rate at which data was received from the remote system in the past 5 minutes in terms of Bytes Per Second", Formatter.DATA_SIZE, new ValueMapper<RemoteProcessGroupStatus>() {
- @Override
- public Long getValue(final RemoteProcessGroupStatus status) {
- return Long.valueOf(status.getSentContentSize().longValue() / 300L);
- }
- })),
- TOTAL_BYTES_PER_SECOND(new StandardMetricDescriptor<RemoteProcessGroupStatus>("totalBytesPerSecond", "Total Bytes Per Second",
- "The sum of the send and receive data rate from the remote system in the past 5 minutes in terms of Bytes Per Second",
- Formatter.DATA_SIZE, new ValueMapper<RemoteProcessGroupStatus>() {
- @Override
- public Long getValue(final RemoteProcessGroupStatus status) {
- return Long.valueOf((status.getReceivedContentSize().longValue() + status.getSentContentSize().longValue()) / 300L);
- }
- })),
- AVERAGE_LINEAGE_DURATION(new StandardMetricDescriptor<RemoteProcessGroupStatus>(
- "averageLineageDuration",
- "Average Lineage Duration (5 mins)",
- "The average amount of time that a FlowFile took to process from receipt to drop in the past 5 minutes. For Processors that do not terminate FlowFiles, this value will be 0.",
- Formatter.DURATION,
- new ValueMapper<RemoteProcessGroupStatus>() {
- @Override
- public Long getValue(final RemoteProcessGroupStatus status) {
- return status.getAverageLineageDuration(TimeUnit.MILLISECONDS);
- }
- }, new ValueReducer<StatusSnapshot, Long>() {
- @Override
- public Long reduce(final List<StatusSnapshot> values) {
- long millis = 0L;
- int count = 0;
-
- for (final StatusSnapshot snapshot : values) {
- final long sent = snapshot.getStatusMetrics().get(SENT_COUNT.getDescriptor()).longValue();
- count += sent;
-
- final long avgMillis = snapshot.getStatusMetrics().get(AVERAGE_LINEAGE_DURATION.getDescriptor()).longValue();
- final long totalMillis = avgMillis * sent;
- millis += totalMillis;
- }
-
- return count == 0 ? 0 : millis / count;
- }
- }
- ));
-
- private final MetricDescriptor<RemoteProcessGroupStatus> descriptor;
-
- private RemoteProcessGroupStatusDescriptor(final MetricDescriptor<RemoteProcessGroupStatus> descriptor) {
- this.descriptor = descriptor;
- }
-
- public String getField() {
- return descriptor.getField();
- }
-
- public MetricDescriptor<RemoteProcessGroupStatus> getDescriptor() {
- return descriptor;
- }
- }
-
- public static enum ProcessGroupStatusDescriptor {
-
- BYTES_READ(new StandardMetricDescriptor<ProcessGroupStatus>("bytesRead", "Bytes Read (5 mins)",
- "The total number of bytes read from Content Repository by Processors in this Process Group in the past 5 minutes", Formatter.DATA_SIZE, new ValueMapper<ProcessGroupStatus>() {
- @Override
- public Long getValue(final ProcessGroupStatus status) {
- return status.getBytesRead();
- }
- })),
- BYTES_WRITTEN(new StandardMetricDescriptor<ProcessGroupStatus>("bytesWritten", "Bytes Written (5 mins)",
- "The total number of bytes written to Content Repository by Processors in this Process Group in the past 5 minutes", Formatter.DATA_SIZE, new ValueMapper<ProcessGroupStatus>() {
- @Override
- public Long getValue(final ProcessGroupStatus status) {
- return status.getBytesWritten();
- }
- })),
- BYTES_TRANSFERRED(new StandardMetricDescriptor<ProcessGroupStatus>("bytesTransferred", "Bytes Transferred (5 mins)",
- "The total number of bytes read from or written to Content Repository by Processors in this Process Group in the past 5 minutes",
- Formatter.DATA_SIZE, new ValueMapper<ProcessGroupStatus>() {
- @Override
- public Long getValue(final ProcessGroupStatus status) {
- return status.getBytesRead() + status.getBytesWritten();
- }
- })),
- INPUT_BYTES(new StandardMetricDescriptor<ProcessGroupStatus>("inputBytes", "Bytes In (5 mins)",
- "The cumulative size of all FlowFiles that have entered this Process Group via its Input Ports in the past 5 minutes",
- Formatter.DATA_SIZE, new ValueMapper<ProcessGroupStatus>() {
- @Override
- public Long getValue(final ProcessGroupStatus status) {
- return status.getInputContentSize();
- }
- })),
- INPUT_COUNT(new StandardMetricDescriptor<ProcessGroupStatus>("inputCount", "FlowFiles In (5 mins)",
- "The number of FlowFiles that have entered this Process Group via its Input Ports in the past 5 minutes",
- Formatter.COUNT, new ValueMapper<ProcessGroupStatus>() {
- @Override
- public Long getValue(final ProcessGroupStatus status) {
- return status.getInputCount().longValue();
- }
- })),
- OUTPUT_BYTES(new StandardMetricDescriptor<ProcessGroupStatus>("outputBytes", "Bytes Out (5 mins)",
- "The cumulative size of all FlowFiles that have exited this Process Group via its Output Ports in the past 5 minutes",
- Formatter.DATA_SIZE, new ValueMapper<ProcessGroupStatus>() {
- @Override
- public Long getValue(final ProcessGroupStatus status) {
- return status.getOutputContentSize();
- }
- })),
- OUTPUT_COUNT(new StandardMetricDescriptor<ProcessGroupStatus>("outputCount", "FlowFiles Out (5 mins)",
- "The number of FlowFiles that have exited this Process Group via its Output Ports in the past 5 minutes",
- Formatter.COUNT, new ValueMapper<ProcessGroupStatus>() {
- @Override
- public Long getValue(final ProcessGroupStatus status) {
- return status.getOutputCount().longValue();
- }
- })),
- QUEUED_BYTES(new StandardMetricDescriptor<ProcessGroupStatus>("queuedBytes", "Queued Bytes",
- "The cumulative size of all FlowFiles queued in all Connections of this Process Group",
- Formatter.DATA_SIZE, new ValueMapper<ProcessGroupStatus>() {
- @Override
- public Long getValue(final ProcessGroupStatus status) {
- return status.getQueuedContentSize();
- }
- })),
- QUEUED_COUNT(new StandardMetricDescriptor<ProcessGroupStatus>("queuedCount", "Queued Count",
- "The number of FlowFiles queued in all Connections of this Process Group", Formatter.COUNT, new ValueMapper<ProcessGroupStatus>() {
- @Override
- public Long getValue(final ProcessGroupStatus status) {
- return status.getQueuedCount().longValue();
- }
- })),
- TASK_MILLIS(new StandardMetricDescriptor<ProcessGroupStatus>("taskMillis", "Total Task Duration (5 mins)",
- "The total number of thread-milliseconds that the Processors within this ProcessGroup have used to complete their tasks in the past 5 minutes",
- Formatter.DURATION, new ValueMapper<ProcessGroupStatus>() {
- @Override
- public Long getValue(final ProcessGroupStatus status) {
- return calculateTaskMillis(status);
- }
- }));
-
- private MetricDescriptor<ProcessGroupStatus> descriptor;
-
- private ProcessGroupStatusDescriptor(final MetricDescriptor<ProcessGroupStatus> descriptor) {
- this.descriptor = descriptor;
- }
-
- public String getField() {
- return descriptor.getField();
- }
-
- public MetricDescriptor<ProcessGroupStatus> getDescriptor() {
- return descriptor;
- }
- }
-
- public static enum ConnectionStatusDescriptor {
-
- INPUT_BYTES(new StandardMetricDescriptor<ConnectionStatus>("inputBytes", "Bytes In (5 mins)",
- "The cumulative size of all FlowFiles that were transferred to this Connection in the past 5 minutes", Formatter.DATA_SIZE, new ValueMapper<ConnectionStatus>() {
- @Override
- public Long getValue(final ConnectionStatus status) {
- return status.getInputBytes();
- }
- })),
- INPUT_COUNT(new StandardMetricDescriptor<ConnectionStatus>("inputCount", "FlowFiles In (5 mins)",
- "The number of FlowFiles that were transferred to this Connection in the past 5 minutes", Formatter.COUNT, new ValueMapper<ConnectionStatus>() {
- @Override
- public Long getValue(final ConnectionStatus status) {
- return Long.valueOf(status.getInputCount());
- }
- })),
- OUTPUT_BYTES(new StandardMetricDescriptor<ConnectionStatus>("outputBytes", "Bytes Out (5 mins)",
- "The cumulative size of all FlowFiles that were pulled from this Connection in the past 5 minutes", Formatter.DATA_SIZE, new ValueMapper<ConnectionStatus>() {
- @Override
- public Long getValue(final ConnectionStatus status) {
- return status.getOutputBytes();
- }
- })),
- OUTPUT_COUNT(new StandardMetricDescriptor<ConnectionStatus>("outputCount", "FlowFiles Out (5 mins)",
- "The number of FlowFiles that were pulled from this Connection in the past 5 minutes", Formatter.COUNT, new ValueMapper<ConnectionStatus>() {
- @Override
- public Long getValue(final ConnectionStatus status) {
- return Long.valueOf(status.getOutputCount());
- }
- })),
- QUEUED_BYTES(new StandardMetricDescriptor<ConnectionStatus>("queuedBytes", "Queued Bytes",
- "The number of Bytes queued in this Connection", Formatter.DATA_SIZE, new ValueMapper<ConnectionStatus>() {
- @Override
- public Long getValue(final ConnectionStatus status) {
- return status.getQueuedBytes();
- }
- })),
- QUEUED_COUNT(new StandardMetricDescriptor<ConnectionStatus>("queuedCount", "Queued Count",
- "The number of FlowFiles queued in this Connection", Formatter.COUNT, new ValueMapper<ConnectionStatus>() {
- @Override
- public Long getValue(final ConnectionStatus status) {
- return Long.valueOf(status.getQueuedCount());
- }
- }));
-
- private MetricDescriptor<ConnectionStatus> descriptor;
-
- private ConnectionStatusDescriptor(final MetricDescriptor<ConnectionStatus> descriptor) {
- this.descriptor = descriptor;
- }
-
- public String getField() {
- return descriptor.getField();
- }
-
- public MetricDescriptor<ConnectionStatus> getDescriptor() {
- return descriptor;
- }
- }
- public static enum ProcessorStatusDescriptor {
-
- BYTES_READ(new StandardMetricDescriptor<ProcessorStatus>("bytesRead", "Bytes Read (5 mins)",
- "The total number of bytes read from the Content Repository by this Processor in the past 5 minutes", Formatter.DATA_SIZE, new ValueMapper<ProcessorStatus>() {
- @Override
- public Long getValue(final ProcessorStatus status) {
- return status.getBytesRead();
- }
- })),
- BYTES_WRITTEN(new StandardMetricDescriptor<ProcessorStatus>("bytesWritten", "Bytes Written (5 mins)",
- "The total number of bytes written to the Content Repository by this Processor in the past 5 minutes", Formatter.DATA_SIZE, new ValueMapper<ProcessorStatus>() {
- @Override
- public Long getValue(final ProcessorStatus status) {
- return status.getBytesWritten();
- }
- })),
- BYTES_TRANSFERRED(new StandardMetricDescriptor<ProcessorStatus>("bytesTransferred", "Bytes Transferred (5 mins)",
- "The total number of bytes read from or written to the Content Repository by this Processor in the past 5 minutes", Formatter.DATA_SIZE, new ValueMapper<ProcessorStatus>() {
- @Override
- public Long getValue(final ProcessorStatus status) {
- return status.getBytesRead() + status.getBytesWritten();
- }
- })),
- INPUT_BYTES(new StandardMetricDescriptor<ProcessorStatus>("inputBytes", "Bytes In (5 mins)",
- "The cumulative size of all FlowFiles that this Processor has pulled from its queues in the past 5 minutes", Formatter.DATA_SIZE, new ValueMapper<ProcessorStatus>() {
- @Override
- public Long getValue(final ProcessorStatus status) {
- return status.getInputBytes();
- }
- })),
- INPUT_COUNT(new StandardMetricDescriptor<ProcessorStatus>("inputCount", "FlowFiles In (5 mins)",
- "The number of FlowFiles that this Processor has pulled from its queues in the past 5 minutes", Formatter.COUNT, new ValueMapper<ProcessorStatus>() {
- @Override
- public Long getValue(final ProcessorStatus status) {
- return Long.valueOf(status.getInputCount());
- }
- })),
- OUTPUT_BYTES(new StandardMetricDescriptor<ProcessorStatus>("outputBytes", "Bytes Out (5 mins)",
- "The cumulative size of all FlowFiles that this Processor has transferred to downstream queues in the past 5 minutes", Formatter.DATA_SIZE, new ValueMapper<ProcessorStatus>() {
- @Override
- public Long getValue(final ProcessorStatus status) {
- return status.getOutputBytes();
- }
- })),
- OUTPUT_COUNT(new StandardMetricDescriptor<ProcessorStatus>("outputCount", "FlowFiles Out (5 mins)",
- "The number of FlowFiles that this Processor has transferred to downstream queues in the past 5 minutes", Formatter.COUNT, new ValueMapper<ProcessorStatus>() {
- @Override
- public Long getValue(final ProcessorStatus status) {
- return Long.valueOf(status.getOutputCount());
- }
- })),
- TASK_COUNT(new StandardMetricDescriptor<ProcessorStatus>("taskCount", "Tasks (5 mins)", "The number of tasks that this Processor has completed in the past 5 minutes",
- Formatter.COUNT, new ValueMapper<ProcessorStatus>() {
- @Override
- public Long getValue(final ProcessorStatus status) {
- return Long.valueOf(status.getInvocations());
- }
- })),
- TASK_MILLIS(new StandardMetricDescriptor<ProcessorStatus>("taskMillis", "Total Task Duration (5 mins)",
- "The total number of thread-milliseconds that the Processor has used to complete its tasks in the past 5 minutes", Formatter.DURATION, new ValueMapper<ProcessorStatus>() {
- @Override
- public Long getValue(final ProcessorStatus status) {
- return TimeUnit.MILLISECONDS.convert(status.getProcessingNanos(), TimeUnit.NANOSECONDS);
- }
- })),
- FLOWFILES_REMOVED(new StandardMetricDescriptor<ProcessorStatus>("flowFilesRemoved", "FlowFiles Removed (5 mins)",
- "The total number of FlowFiles removed by this Processor in the last 5 minutes", Formatter.COUNT, new ValueMapper<ProcessorStatus>() {
- @Override
- public Long getValue(final ProcessorStatus status) {
- return Long.valueOf(status.getFlowFilesRemoved());
- }
- })),
- AVERAGE_LINEAGE_DURATION(new StandardMetricDescriptor<ProcessorStatus>(
- "averageLineageDuration",
- "Average Lineage Duration (5 mins)",
- "The average amount of time that a FlowFile took to process (from receipt until this Processor finished processing it) in the past 5 minutes.",
- Formatter.DURATION,
- new ValueMapper<ProcessorStatus>() {
- @Override
- public Long getValue(final ProcessorStatus status) {
- return status.getAverageLineageDuration(TimeUnit.MILLISECONDS);
- }
- }, new ValueReducer<StatusSnapshot, Long>() {
- @Override
- public Long reduce(final List<StatusSnapshot> values) {
- long millis = 0L;
- int count = 0;
-
- for (final StatusSnapshot snapshot : values) {
- final long removed = snapshot.getStatusMetrics().get(FLOWFILES_REMOVED.getDescriptor()).longValue();
- count += removed;
-
- count += snapshot.getStatusMetrics().get(OUTPUT_COUNT.getDescriptor()).longValue();
-
- final long avgMillis = snapshot.getStatusMetrics().get(AVERAGE_LINEAGE_DURATION.getDescriptor()).longValue();
- final long totalMillis = avgMillis * removed;
- millis += totalMillis;
- }
-
- return count == 0 ? 0 : millis / count;
- }
- }
- )),
- AVERAGE_TASK_MILLIS(new StandardMetricDescriptor<ProcessorStatus>(
- "averageTaskMillis",
- "Average Task Duration",
- "The average duration it took this Processor to complete a task, as averaged over the past 5 minutes",
- Formatter.DURATION,
- new ValueMapper<ProcessorStatus>() {
- @Override
- public Long getValue(final ProcessorStatus status) {
- return status.getInvocations() == 0 ? 0 : TimeUnit.MILLISECONDS.convert(status.getProcessingNanos(), TimeUnit.NANOSECONDS) / status.getInvocations();
- }
- },
- new ValueReducer<StatusSnapshot, Long>() {
- @Override
- public Long reduce(final List<StatusSnapshot> values) {
- long procMillis = 0L;
- int invocations = 0;
-
- for (final StatusSnapshot snapshot : values) {
- procMillis += snapshot.getStatusMetrics().get(TASK_MILLIS.getDescriptor()).longValue();
- invocations += snapshot.getStatusMetrics().get(TASK_COUNT.getDescriptor()).intValue();
- }
-
- if (invocations == 0) {
- return 0L;
- }
-
- return procMillis / invocations;
- }
- }
- ));
-
- private MetricDescriptor<ProcessorStatus> descriptor;
-
- private ProcessorStatusDescriptor(final MetricDescriptor<ProcessorStatus> descriptor) {
- this.descriptor = descriptor;
- }
-
- public String getField() {
- return descriptor.getField();
- }
-
- public MetricDescriptor<ProcessorStatus> getDescriptor() {
- return descriptor;
- }
- }
@Override
public List<MetricDescriptor<ConnectionStatus>> getConnectionMetricDescriptors() {
http://git-wip-us.apache.org/repos/asf/nifi/blob/ad32cb82/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/events/NodeBulletinProcessingStrategy.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/events/NodeBulletinProcessingStrategy.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/events/NodeBulletinProcessingStrategy.java
index d3cfd9e..ad9208e 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/events/NodeBulletinProcessingStrategy.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/events/NodeBulletinProcessingStrategy.java
@@ -17,50 +17,24 @@
package org.apache.nifi.events;
import java.util.HashSet;
-import java.util.LinkedHashSet;
import java.util.Set;
-import java.util.concurrent.locks.Lock;
-import java.util.concurrent.locks.ReentrantLock;
+import org.apache.commons.collections4.queue.CircularFifoQueue;
import org.apache.nifi.reporting.Bulletin;
/**
*
*/
public class NodeBulletinProcessingStrategy implements BulletinProcessingStrategy {
-
- private final Lock lock;
- private final Set<Bulletin> bulletins;
-
- public NodeBulletinProcessingStrategy() {
- lock = new ReentrantLock();
- bulletins = new LinkedHashSet<>();
- }
+ private final CircularFifoQueue<Bulletin> ringBuffer = new CircularFifoQueue<>(5);
@Override
- public void update(final Bulletin bulletin) {
- lock.lock();
- try {
- bulletins.add(bulletin);
- } finally {
- lock.unlock();
- }
+ public synchronized void update(final Bulletin bulletin) {
+ ringBuffer.add(bulletin);
}
- public Set<Bulletin> getBulletins() {
- final Set<Bulletin> response = new HashSet<>();
-
- lock.lock();
- try {
- // get all the bulletins currently stored
- response.addAll(bulletins);
-
- // remove the bulletins
- bulletins.clear();
- } finally {
- lock.unlock();
- }
-
+ public synchronized Set<Bulletin> getBulletins() {
+ final Set<Bulletin> response = new HashSet<>(ringBuffer);
return response;
}
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/ad32cb82/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/events/VolatileBulletinRepository.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/events/VolatileBulletinRepository.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/events/VolatileBulletinRepository.java
index 8aeb34d..7202546 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/events/VolatileBulletinRepository.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/events/VolatileBulletinRepository.java
@@ -19,7 +19,6 @@ package org.apache.nifi.events;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
-import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
@@ -207,25 +206,6 @@ public class VolatileBulletinRepository implements BulletinRepository {
return controllerBulletins;
}
- /**
- * Overrides the default bulletin processing strategy. When a custom
- * bulletin strategy is employed, bulletins will not be persisted in this
- * repository and will sent to the specified strategy instead.
- *
- * @param strategy bulletin strategy
- */
- public void overrideDefaultBulletinProcessing(final BulletinProcessingStrategy strategy) {
- Objects.requireNonNull(strategy);
- this.processingStrategy = strategy;
- }
-
- /**
- * Restores the default bulletin processing strategy.
- */
- public void restoreDefaultBulletinProcessing() {
- this.processingStrategy = new DefaultBulletinProcessingStrategy();
- }
-
private List<RingBuffer<Bulletin>> getBulletinBuffers(final Bulletin bulletin) {
final String storageKey = getBulletinStoreKey(bulletin);
http://git-wip-us.apache.org/repos/asf/nifi/blob/ad32cb82/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/NiFiServiceFacade.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/NiFiServiceFacade.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/NiFiServiceFacade.java
index bc5245c..a290fec 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/NiFiServiceFacade.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/NiFiServiceFacade.java
@@ -67,7 +67,6 @@ import org.apache.nifi.web.api.dto.status.ClusterProcessGroupStatusDTO;
import org.apache.nifi.web.api.dto.status.ClusterProcessorStatusDTO;
import org.apache.nifi.web.api.dto.status.ClusterRemoteProcessGroupStatusDTO;
import org.apache.nifi.web.api.dto.status.ClusterStatusDTO;
-import org.apache.nifi.web.api.dto.status.ClusterStatusHistoryDTO;
import org.apache.nifi.web.api.dto.status.ControllerStatusDTO;
import org.apache.nifi.web.api.dto.status.NodeStatusDTO;
import org.apache.nifi.web.api.dto.status.ProcessGroupStatusDTO;
@@ -1538,11 +1537,6 @@ public interface NiFiServiceFacade {
*/
ClusterProcessorStatusDTO getClusterProcessorStatus(String processorId);
- /**
- * @param processorId id
- * @return the processor status history for each node connected to the cluster
- */
- ClusterStatusHistoryDTO getClusterProcessorStatusHistory(String processorId);
/**
* Returns a connection's status for each node connected to the cluster.
@@ -1552,17 +1546,6 @@ public interface NiFiServiceFacade {
*/
ClusterConnectionStatusDTO getClusterConnectionStatus(String connectionId);
- /**
- * @param connectionId id
- * @return the connection status history for each node connected to the cluster
- */
- ClusterStatusHistoryDTO getClusterConnectionStatusHistory(String connectionId);
-
- /**
- * @param processGroupId id
- * @return the process group status history for each node connected to the cluster
- */
- ClusterStatusHistoryDTO getClusterProcessGroupStatusHistory(String processGroupId);
/**
* Returns a process group's status for each node connected to the cluster.
@@ -1572,13 +1555,6 @@ public interface NiFiServiceFacade {
*/
ClusterProcessGroupStatusDTO getClusterProcessGroupStatus(String processorId);
- /**
- * Returns the remote process group status history for each node connected to the cluster.
- *
- * @param remoteProcessGroupId a remote process group identifier
- * @return The cluster status history
- */
- ClusterStatusHistoryDTO getClusterRemoteProcessGroupStatusHistory(String remoteProcessGroupId);
/**
* Returns a remote process group's status for each node connected to the cluster.
http://git-wip-us.apache.org/repos/asf/nifi/blob/ad32cb82/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiServiceFacade.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiServiceFacade.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiServiceFacade.java
index 945c671..098fd64 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiServiceFacade.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiServiceFacade.java
@@ -16,6 +16,25 @@
*/
package org.apache.nifi.web;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.ListIterator;
+import java.util.Map;
+import java.util.Set;
+import java.util.TimeZone;
+import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+
+import javax.ws.rs.WebApplicationException;
+
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.nifi.action.Action;
@@ -72,7 +91,6 @@ import org.apache.nifi.remote.RootGroupPort;
import org.apache.nifi.reporting.Bulletin;
import org.apache.nifi.reporting.BulletinQuery;
import org.apache.nifi.reporting.BulletinRepository;
-import org.apache.nifi.reporting.ComponentType;
import org.apache.nifi.user.AccountStatus;
import org.apache.nifi.user.NiFiUser;
import org.apache.nifi.user.NiFiUserGroup;
@@ -131,7 +149,6 @@ import org.apache.nifi.web.api.dto.status.ClusterProcessGroupStatusDTO;
import org.apache.nifi.web.api.dto.status.ClusterProcessorStatusDTO;
import org.apache.nifi.web.api.dto.status.ClusterRemoteProcessGroupStatusDTO;
import org.apache.nifi.web.api.dto.status.ClusterStatusDTO;
-import org.apache.nifi.web.api.dto.status.ClusterStatusHistoryDTO;
import org.apache.nifi.web.api.dto.status.ControllerStatusDTO;
import org.apache.nifi.web.api.dto.status.NodeConnectionStatusDTO;
import org.apache.nifi.web.api.dto.status.NodePortStatusDTO;
@@ -160,24 +177,6 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.security.access.AccessDeniedException;
-import javax.ws.rs.WebApplicationException;
-import java.nio.charset.StandardCharsets;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.Date;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.LinkedHashMap;
-import java.util.LinkedHashSet;
-import java.util.List;
-import java.util.ListIterator;
-import java.util.Map;
-import java.util.Set;
-import java.util.TimeZone;
-import java.util.UUID;
-import java.util.concurrent.TimeUnit;
-
/**
* Implementation of NiFiServiceFacade that performs revision checking.
*/
@@ -2105,78 +2104,13 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
@Override
public ProcessGroupStatusDTO getProcessGroupStatus(String groupId) {
- ProcessGroupStatusDTO statusReport;
- if (properties.isClusterManager()) {
- final ProcessGroupStatus mergedProcessGroupStatus = clusterManager.getProcessGroupStatus(groupId);
- if (mergedProcessGroupStatus == null) {
- throw new ResourceNotFoundException(String.format("Unable to find status for process group %s.", groupId));
- }
- statusReport = dtoFactory.createProcessGroupStatusDto(clusterManager.getBulletinRepository(), mergedProcessGroupStatus);
- } else {
- statusReport = controllerFacade.getProcessGroupStatus(groupId);
- }
- return statusReport;
+ return controllerFacade.getProcessGroupStatus(groupId);
}
@Override
public ControllerStatusDTO getControllerStatus() {
- final ControllerStatusDTO controllerStatus;
-
- if (properties.isClusterManager()) {
- final Set<Node> connectedNodes = clusterManager.getNodes(Node.Status.CONNECTED);
-
- if (connectedNodes.isEmpty()) {
- throw new NoConnectedNodesException();
- }
-
- int activeThreadCount = 0;
- long totalFlowFileObjectCount = 0;
- long totalFlowFileByteCount = 0;
- for (final Node node : connectedNodes) {
- final HeartbeatPayload nodeHeartbeatPayload = node.getHeartbeatPayload();
- if (nodeHeartbeatPayload == null) {
- continue;
- }
-
- activeThreadCount += nodeHeartbeatPayload.getActiveThreadCount();
- totalFlowFileObjectCount += nodeHeartbeatPayload.getTotalFlowFileCount();
- totalFlowFileByteCount += nodeHeartbeatPayload.getTotalFlowFileBytes();
- }
-
- controllerStatus = new ControllerStatusDTO();
- controllerStatus.setActiveThreadCount(activeThreadCount);
- controllerStatus.setQueued(FormatUtils.formatCount(totalFlowFileObjectCount) + " / " + FormatUtils.formatDataSize(totalFlowFileByteCount));
-
- final int numNodes = clusterManager.getNodeIds().size();
- controllerStatus.setConnectedNodes(connectedNodes.size() + " / " + numNodes);
-
- // get the bulletins for the controller
- final BulletinRepository bulletinRepository = clusterManager.getBulletinRepository();
- controllerStatus.setBulletins(dtoFactory.createBulletinDtos(bulletinRepository.findBulletinsForController()));
-
- // get the controller service bulletins
- final BulletinQuery controllerServiceQuery = new BulletinQuery.Builder().sourceType(ComponentType.CONTROLLER_SERVICE).build();
- controllerStatus.setControllerServiceBulletins(dtoFactory.createBulletinDtos(bulletinRepository.findBulletins(controllerServiceQuery)));
-
- // get the reporting task bulletins
- final BulletinQuery reportingTaskQuery = new BulletinQuery.Builder().sourceType(ComponentType.REPORTING_TASK).build();
- controllerStatus.setReportingTaskBulletins(dtoFactory.createBulletinDtos(bulletinRepository.findBulletins(reportingTaskQuery)));
-
- // get the component counts by extracting them from the roots' group status
- final ProcessGroupStatus status = clusterManager.getProcessGroupStatus("root");
- if (status != null) {
- final ProcessGroupCounts counts = extractProcessGroupCounts(status);
- controllerStatus.setRunningCount(counts.getRunningCount());
- controllerStatus.setStoppedCount(counts.getStoppedCount());
- controllerStatus.setInvalidCount(counts.getInvalidCount());
- controllerStatus.setDisabledCount(counts.getDisabledCount());
- controllerStatus.setActiveRemotePortCount(counts.getActiveRemotePortCount());
- controllerStatus.setInactiveRemotePortCount(counts.getInactiveRemotePortCount());
- }
- } else {
- // get the controller status
- controllerStatus = controllerFacade.getControllerStatus();
- }
+ // get the controller status
+ final ControllerStatusDTO controllerStatus = controllerFacade.getControllerStatus();
// determine if there are any pending user accounts - only include if appropriate
if (NiFiUserUtils.getAuthorities().contains(Authority.ROLE_ADMIN.toString())) {
@@ -3371,25 +3305,6 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
return clusterRemoteProcessGroupStatusDto;
}
- @Override
- public ClusterStatusHistoryDTO getClusterProcessorStatusHistory(String processorId) {
- return clusterManager.getProcessorStatusHistory(processorId);
- }
-
- @Override
- public ClusterStatusHistoryDTO getClusterConnectionStatusHistory(String connectionId) {
- return clusterManager.getConnectionStatusHistory(connectionId);
- }
-
- @Override
- public ClusterStatusHistoryDTO getClusterProcessGroupStatusHistory(String processGroupId) {
- return clusterManager.getProcessGroupStatusHistory(processGroupId);
- }
-
- @Override
- public ClusterStatusHistoryDTO getClusterRemoteProcessGroupStatusHistory(String remoteProcessGroupId) {
- return clusterManager.getRemoteProcessGroupStatusHistory(remoteProcessGroupId);
- }
@Override
public NodeStatusDTO getNodeStatus(String nodeId) {
http://git-wip-us.apache.org/repos/asf/nifi/blob/ad32cb82/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ClusterResource.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ClusterResource.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ClusterResource.java
index 857df56..3086ab4 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ClusterResource.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ClusterResource.java
@@ -16,6 +16,8 @@
*/
package org.apache.nifi.web.api;
+import java.net.URI;
+import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.List;
@@ -25,6 +27,7 @@ import javax.ws.rs.DefaultValue;
import javax.ws.rs.FormParam;
import javax.ws.rs.GET;
import javax.ws.rs.HEAD;
+import javax.ws.rs.HttpMethod;
import javax.ws.rs.PUT;
import javax.ws.rs.Path;
import javax.ws.rs.PathParam;
@@ -34,6 +37,7 @@ import javax.ws.rs.core.Context;
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.Response;
+import org.apache.commons.lang3.StringUtils;
import org.apache.nifi.cluster.manager.impl.WebClusterManager;
import org.apache.nifi.cluster.node.Node;
import org.apache.nifi.util.NiFiProperties;
@@ -49,23 +53,22 @@ import org.apache.nifi.web.api.dto.RevisionDTO;
import org.apache.nifi.web.api.dto.search.NodeSearchResultDTO;
import org.apache.nifi.web.api.dto.status.ClusterConnectionStatusDTO;
import org.apache.nifi.web.api.dto.status.ClusterPortStatusDTO;
+import org.apache.nifi.web.api.dto.status.ClusterProcessGroupStatusDTO;
import org.apache.nifi.web.api.dto.status.ClusterProcessorStatusDTO;
import org.apache.nifi.web.api.dto.status.ClusterRemoteProcessGroupStatusDTO;
import org.apache.nifi.web.api.dto.status.ClusterStatusDTO;
-import org.apache.nifi.web.api.dto.status.ClusterStatusHistoryDTO;
import org.apache.nifi.web.api.entity.ClusterConnectionStatusEntity;
import org.apache.nifi.web.api.entity.ClusterEntity;
import org.apache.nifi.web.api.entity.ClusterPortStatusEntity;
+import org.apache.nifi.web.api.entity.ClusterProcessGroupStatusEntity;
import org.apache.nifi.web.api.entity.ClusterProcessorStatusEntity;
import org.apache.nifi.web.api.entity.ClusterRemoteProcessGroupStatusEntity;
import org.apache.nifi.web.api.entity.ClusterSearchResultsEntity;
import org.apache.nifi.web.api.entity.ClusterStatusEntity;
-import org.apache.nifi.web.api.entity.ClusterStatusHistoryEntity;
import org.apache.nifi.web.api.entity.ProcessorEntity;
+import org.apache.nifi.web.api.entity.StatusHistoryEntity;
import org.apache.nifi.web.api.request.ClientIdParameter;
import org.apache.nifi.web.api.request.LongParameter;
-
-import org.apache.commons.lang3.StringUtils;
import org.springframework.security.access.prepost.PreAuthorize;
import com.sun.jersey.api.core.ResourceContext;
@@ -75,8 +78,6 @@ 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.web.api.dto.status.ClusterProcessGroupStatusDTO;
-import org.apache.nifi.web.api.entity.ClusterProcessGroupStatusEntity;
/**
* RESTful endpoint for managing a cluster.
@@ -92,6 +93,7 @@ public class ClusterResource extends ApplicationResource {
private ResourceContext resourceContext;
private NiFiServiceFacade serviceFacade;
private NiFiProperties properties;
+ private WebClusterManager clusterManager;
/**
* Locates the ClusterConnection sub-resource.
@@ -595,7 +597,7 @@ public class ClusterResource extends ApplicationResource {
@PreAuthorize("hasAnyRole('ROLE_MONITOR', 'ROLE_DFM', 'ROLE_ADMIN')")
@ApiOperation(
value = "Gets processor status history across the cluster",
- response = ClusterStatusHistoryEntity.class,
+ response = StatusHistoryEntity.class,
authorizations = {
@Authorization(value = "Read Only", type = "ROLE_MONITOR"),
@Authorization(value = "DFM", type = "ROLE_DFM"),
@@ -624,19 +626,16 @@ public class ClusterResource extends ApplicationResource {
@PathParam("id") String id) {
if (properties.isClusterManager()) {
- final ClusterStatusHistoryDTO dto = serviceFacade.getClusterProcessorStatusHistory(id);
-
- // create the revision
- RevisionDTO revision = new RevisionDTO();
- revision.setClientId(clientId.getClientId());
-
- // create entity
- final ClusterStatusHistoryEntity entity = new ClusterStatusHistoryEntity();
- entity.setClusterStatusHistory(dto);
- entity.setRevision(revision);
+ final URI uri;
+ try {
+ final URI originalUri = getAbsolutePath();
+ final String newPath = "/nifi-api/processors/" + id + "/status/history";
+ uri = new URI(originalUri.getScheme(), originalUri.getAuthority(), newPath, originalUri.getQuery(), originalUri.getFragment());
+ } catch (final URISyntaxException use) {
+ throw new RuntimeException(use);
+ }
- // generate the response
- return generateOkResponse(entity).build();
+ return clusterManager.applyRequest(HttpMethod.GET, uri, getRequestParameters(true), getHeaders()).getResponse();
}
throw new IllegalClusterResourceRequestException("Only a cluster manager can process the request.");
@@ -705,67 +704,6 @@ public class ClusterResource extends ApplicationResource {
}
/**
- * Gets the connections status history for every node.
- *
- * @param clientId Optional client id. If the client id is not specified, a new one will be generated. This value (whether specified or generated) is included in the response.
- * @param id The id of the processor
- * @return A clusterProcessorStatusHistoryEntity
- */
- @GET
- @Consumes(MediaType.WILDCARD)
- @Produces({MediaType.APPLICATION_XML, MediaType.APPLICATION_JSON})
- @Path("/connections/{id}/status/history")
- @PreAuthorize("hasAnyRole('ROLE_MONITOR', 'ROLE_DFM', 'ROLE_ADMIN')")
- @ApiOperation(
- value = "Gets connection status history across the cluster",
- response = ClusterStatusHistoryEntity.class,
- authorizations = {
- @Authorization(value = "Read Only", type = "ROLE_MONITOR"),
- @Authorization(value = "DFM", type = "ROLE_DFM"),
- @Authorization(value = "Admin", 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 getConnectionStatusHistory(
- @ApiParam(
- value = "If the client id is not specified, new one will be generated. This value (whether specified or generated) is included in the response.",
- required = false
- )
- @QueryParam(CLIENT_ID) @DefaultValue(StringUtils.EMPTY) ClientIdParameter clientId,
- @ApiParam(
- value = "The connection id.",
- required = true
- )
- @PathParam("id") String id) {
-
- if (properties.isClusterManager()) {
- final ClusterStatusHistoryDTO dto = serviceFacade.getClusterConnectionStatusHistory(id);
-
- // create the revision
- RevisionDTO revision = new RevisionDTO();
- revision.setClientId(clientId.getClientId());
-
- // create entity
- final ClusterStatusHistoryEntity entity = new ClusterStatusHistoryEntity();
- entity.setClusterStatusHistory(dto);
- entity.setRevision(revision);
-
- // generate the response
- return generateOkResponse(entity).build();
- }
-
- throw new IllegalClusterResourceRequestException("Only a cluster manager can process the request.");
- }
-
- /**
* Gets the process group status for every node.
*
* @param clientId Optional client id. If the client id is not specified, a new one will be generated. This value (whether specified or generated) is included in the response.
@@ -827,66 +765,6 @@ public class ClusterResource extends ApplicationResource {
throw new IllegalClusterResourceRequestException("Only a cluster manager can process the request.");
}
- /**
- * Gets the process group status history for every node.
- *
- * @param clientId Optional client id. If the client id is not specified, a new one will be generated. This value (whether specified or generated) is included in the response.
- * @param id The id of the process group
- * @return A clusterProcessGroupStatusHistoryEntity
- */
- @GET
- @Consumes(MediaType.WILDCARD)
- @Produces({MediaType.APPLICATION_XML, MediaType.APPLICATION_JSON})
- @Path("/process-groups/{id}/status/history")
- @PreAuthorize("hasAnyRole('ROLE_MONITOR', 'ROLE_DFM', 'ROLE_ADMIN')")
- @ApiOperation(
- value = "Gets process group status history across the cluster",
- response = ClusterStatusHistoryEntity.class,
- authorizations = {
- @Authorization(value = "Read Only", type = "ROLE_MONITOR"),
- @Authorization(value = "DFM", type = "ROLE_DFM"),
- @Authorization(value = "Admin", 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 getProcessGroupStatusHistory(
- @ApiParam(
- value = "If the client id is not specified, new one will be generated. This value (whether specified or generated) is included in the response.",
- required = false
- )
- @QueryParam(CLIENT_ID) @DefaultValue(StringUtils.EMPTY) ClientIdParameter clientId,
- @ApiParam(
- value = "The process group id.",
- required = true
- )
- @PathParam("id") String id) {
-
- if (properties.isClusterManager()) {
- final ClusterStatusHistoryDTO dto = serviceFacade.getClusterProcessGroupStatusHistory(id);
-
- // create the revision
- RevisionDTO revision = new RevisionDTO();
- revision.setClientId(clientId.getClientId());
-
- // create entity
- final ClusterStatusHistoryEntity entity = new ClusterStatusHistoryEntity();
- entity.setClusterStatusHistory(dto);
- entity.setRevision(revision);
-
- // generate the response
- return generateOkResponse(entity).build();
- }
-
- throw new IllegalClusterResourceRequestException("Only a cluster manager can process the request.");
- }
/**
* Gets the remote process group status for every node.
@@ -1074,66 +952,6 @@ public class ClusterResource extends ApplicationResource {
throw new IllegalClusterResourceRequestException("Only a cluster manager can process the request.");
}
- /**
- * Gets the remote process group status history for every node.
- *
- * @param clientId Optional client id. If the client id is not specified, a new one will be generated. This value (whether specified or generated) is included in the response.
- * @param id The id of the processor
- * @return A clusterRemoteProcessGroupStatusHistoryEntity
- */
- @GET
- @Consumes(MediaType.WILDCARD)
- @Produces({MediaType.APPLICATION_XML, MediaType.APPLICATION_JSON})
- @Path("/remote-process-groups/{id}/status/history")
- @PreAuthorize("hasAnyRole('ROLE_MONITOR', 'ROLE_DFM', 'ROLE_ADMIN')")
- @ApiOperation(
- value = "Gets the remote process group status history across the cluster",
- response = ClusterStatusHistoryEntity.class,
- authorizations = {
- @Authorization(value = "Read Only", type = "ROLE_MONITOR"),
- @Authorization(value = "DFM", type = "ROLE_DFM"),
- @Authorization(value = "Admin", 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 getRemoteProcessGroupStatusHistory(
- @ApiParam(
- value = "If the client id is not specified, new one will be generated. This value (whether specified or generated) is included in the response.",
- required = false
- )
- @QueryParam(CLIENT_ID) @DefaultValue(StringUtils.EMPTY) ClientIdParameter clientId,
- @ApiParam(
- value = "The remote process group id.",
- required = true
- )
- @PathParam("id") String id) {
-
- if (properties.isClusterManager()) {
- final ClusterStatusHistoryDTO dto = serviceFacade.getClusterRemoteProcessGroupStatusHistory(id);
-
- // create the revision
- RevisionDTO revision = new RevisionDTO();
- revision.setClientId(clientId.getClientId());
-
- // create entity
- final ClusterStatusHistoryEntity entity = new ClusterStatusHistoryEntity();
- entity.setClusterStatusHistory(dto);
- entity.setRevision(revision);
-
- // generate the response
- return generateOkResponse(entity).build();
- }
-
- throw new IllegalClusterResourceRequestException("Only a cluster manager can process the request.");
- }
// setters
public void setServiceFacade(NiFiServiceFacade serviceFacade) {
@@ -1144,4 +962,7 @@ public class ClusterResource extends ApplicationResource {
this.properties = properties;
}
+ public void setClusterManager(WebClusterManager clusterManager) {
+ this.clusterManager = clusterManager;
+ }
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/ad32cb82/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ConnectionResource.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ConnectionResource.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ConnectionResource.java
index 738da04..ddc60cd 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ConnectionResource.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ConnectionResource.java
@@ -16,12 +16,41 @@
*/
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 java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+
+import javax.servlet.http.HttpServletRequest;
+import javax.ws.rs.Consumes;
+import javax.ws.rs.DELETE;
+import javax.ws.rs.DefaultValue;
+import javax.ws.rs.FormParam;
+import javax.ws.rs.GET;
+import javax.ws.rs.HttpMethod;
+import javax.ws.rs.POST;
+import javax.ws.rs.PUT;
+import javax.ws.rs.Path;
+import javax.ws.rs.PathParam;
+import javax.ws.rs.Produces;
+import javax.ws.rs.QueryParam;
+import javax.ws.rs.WebApplicationException;
+import javax.ws.rs.core.Context;
+import javax.ws.rs.core.MediaType;
+import javax.ws.rs.core.MultivaluedMap;
+import javax.ws.rs.core.Response;
+import javax.ws.rs.core.Response.Status;
+import javax.ws.rs.core.StreamingOutput;
+
import org.apache.commons.lang3.StringUtils;
import org.apache.nifi.cluster.context.ClusterContext;
import org.apache.nifi.cluster.context.ClusterContextThreadLocal;
@@ -33,7 +62,6 @@ import org.apache.nifi.stream.io.StreamUtils;
import org.apache.nifi.util.NiFiProperties;
import org.apache.nifi.web.ConfigurationSnapshot;
import org.apache.nifi.web.DownloadableContent;
-import org.apache.nifi.web.IllegalClusterResourceRequestException;
import org.apache.nifi.web.NiFiServiceFacade;
import org.apache.nifi.web.Revision;
import org.apache.nifi.web.api.dto.ConnectableDTO;
@@ -55,43 +83,14 @@ import org.apache.nifi.web.api.request.ClientIdParameter;
import org.apache.nifi.web.api.request.ConnectableTypeParameter;
import org.apache.nifi.web.api.request.IntegerParameter;
import org.apache.nifi.web.api.request.LongParameter;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import org.springframework.security.access.prepost.PreAuthorize;
-import javax.servlet.http.HttpServletRequest;
-import javax.ws.rs.Consumes;
-import javax.ws.rs.DELETE;
-import javax.ws.rs.DefaultValue;
-import javax.ws.rs.FormParam;
-import javax.ws.rs.GET;
-import javax.ws.rs.HttpMethod;
-import javax.ws.rs.POST;
-import javax.ws.rs.PUT;
-import javax.ws.rs.Path;
-import javax.ws.rs.PathParam;
-import javax.ws.rs.Produces;
-import javax.ws.rs.QueryParam;
-import javax.ws.rs.WebApplicationException;
-import javax.ws.rs.core.Context;
-import javax.ws.rs.core.MediaType;
-import javax.ws.rs.core.MultivaluedMap;
-import javax.ws.rs.core.Response;
-import javax.ws.rs.core.Response.Status;
-import javax.ws.rs.core.StreamingOutput;
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
-import java.net.URI;
-import java.net.URISyntaxException;
-import java.nio.charset.StandardCharsets;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.UUID;
+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;
/**
* RESTful endpoint for managing a Connection.
@@ -99,8 +98,6 @@ import java.util.UUID;
@Api(hidden = true)
public class ConnectionResource extends ApplicationResource {
- private static final Logger logger = LoggerFactory.getLogger(ConnectionResource.class);
-
private NiFiServiceFacade serviceFacade;
private WebClusterManager clusterManager;
private NiFiProperties properties;
@@ -327,7 +324,7 @@ public class ConnectionResource extends ApplicationResource {
// replicate if cluster manager
if (properties.isClusterManager()) {
- throw new IllegalClusterResourceRequestException("This request is only supported in standalone mode.");
+ return clusterManager.applyRequest(HttpMethod.GET, getAbsolutePath(), getRequestParameters(true), getHeaders()).getResponse();
}
// get the specified processor status history
@@ -565,7 +562,7 @@ public class ConnectionResource extends ApplicationResource {
headersToOverride.put("content-type", MediaType.APPLICATION_JSON);
// replicate put request
- return (Response) clusterManager.applyRequest(HttpMethod.PUT, putUri, updateClientId(connectionEntity), getHeaders(headersToOverride)).getResponse();
+ return clusterManager.applyRequest(HttpMethod.PUT, putUri, updateClientId(connectionEntity), getHeaders(headersToOverride)).getResponse();
}
// get the connection
http://git-wip-us.apache.org/repos/asf/nifi/blob/ad32cb82/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ControllerResource.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ControllerResource.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ControllerResource.java
index ef62a62..36b25f2 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ControllerResource.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ControllerResource.java
@@ -524,6 +524,10 @@ public class ControllerResource extends ApplicationResource {
)
@QueryParam(CLIENT_ID) @DefaultValue(StringUtils.EMPTY) ClientIdParameter clientId) {
+ if (properties.isClusterManager()) {
+ return clusterManager.applyRequest(HttpMethod.GET, getAbsolutePath(), getRequestParameters(true), getHeaders()).getResponse();
+ }
+
final ControllerStatusDTO controllerStatus = serviceFacade.getControllerStatus();
// create the revision
[3/5] nifi git commit: NIFI-1563: Federate requests and merge
responses from nodes instead of storing bulletins and stats at NCM
Posted by ma...@apache.org.
http://git-wip-us.apache.org/repos/asf/nifi/blob/ad32cb82/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/impl/WebClusterManager.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/impl/WebClusterManager.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/impl/WebClusterManager.java
index c0f4c63..3d21b55 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/impl/WebClusterManager.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/manager/impl/WebClusterManager.java
@@ -30,7 +30,6 @@ import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
-import java.util.LinkedHashSet;
import java.util.List;
import java.util.ListIterator;
import java.util.Map;
@@ -73,7 +72,6 @@ import org.apache.nifi.admin.service.AuditService;
import org.apache.nifi.annotation.lifecycle.OnAdded;
import org.apache.nifi.annotation.lifecycle.OnConfigurationRestored;
import org.apache.nifi.annotation.lifecycle.OnRemoved;
-import org.apache.nifi.cluster.BulletinsPayload;
import org.apache.nifi.cluster.HeartbeatPayload;
import org.apache.nifi.cluster.context.ClusterContext;
import org.apache.nifi.cluster.context.ClusterContextImpl;
@@ -109,7 +107,6 @@ import org.apache.nifi.cluster.node.Node.Status;
import org.apache.nifi.cluster.protocol.ConnectionRequest;
import org.apache.nifi.cluster.protocol.ConnectionResponse;
import org.apache.nifi.cluster.protocol.Heartbeat;
-import org.apache.nifi.cluster.protocol.NodeBulletins;
import org.apache.nifi.cluster.protocol.NodeIdentifier;
import org.apache.nifi.cluster.protocol.ProtocolException;
import org.apache.nifi.cluster.protocol.ProtocolHandler;
@@ -121,7 +118,6 @@ import org.apache.nifi.cluster.protocol.message.ConnectionResponseMessage;
import org.apache.nifi.cluster.protocol.message.ControllerStartupFailureMessage;
import org.apache.nifi.cluster.protocol.message.DisconnectMessage;
import org.apache.nifi.cluster.protocol.message.HeartbeatMessage;
-import org.apache.nifi.cluster.protocol.message.NodeBulletinsMessage;
import org.apache.nifi.cluster.protocol.message.PrimaryRoleAssignmentMessage;
import org.apache.nifi.cluster.protocol.message.ProtocolMessage;
import org.apache.nifi.cluster.protocol.message.ProtocolMessage.MessageType;
@@ -153,11 +149,13 @@ import org.apache.nifi.controller.service.ControllerServiceState;
import org.apache.nifi.controller.service.StandardControllerServiceProvider;
import org.apache.nifi.controller.state.SortedStateUtils;
import org.apache.nifi.controller.state.manager.StandardStateManagerProvider;
-import org.apache.nifi.controller.status.ProcessGroupStatus;
-import org.apache.nifi.controller.status.RemoteProcessGroupStatus;
import org.apache.nifi.controller.status.history.ComponentStatusRepository;
+import org.apache.nifi.controller.status.history.ConnectionStatusDescriptor;
import org.apache.nifi.controller.status.history.MetricDescriptor;
-import org.apache.nifi.controller.status.history.StatusHistory;
+import org.apache.nifi.controller.status.history.ProcessGroupStatusDescriptor;
+import org.apache.nifi.controller.status.history.ProcessorStatusDescriptor;
+import org.apache.nifi.controller.status.history.RemoteProcessGroupStatusDescriptor;
+import org.apache.nifi.controller.status.history.StandardStatusSnapshot;
import org.apache.nifi.controller.status.history.StatusHistoryUtil;
import org.apache.nifi.controller.status.history.StatusSnapshot;
import org.apache.nifi.diagnostics.GarbageCollection;
@@ -188,7 +186,9 @@ import org.apache.nifi.remote.cluster.ClusterNodeInformation;
import org.apache.nifi.remote.cluster.NodeInformation;
import org.apache.nifi.remote.protocol.socket.ClusterManagerServerProtocol;
import org.apache.nifi.reporting.Bulletin;
+import org.apache.nifi.reporting.BulletinQuery;
import org.apache.nifi.reporting.BulletinRepository;
+import org.apache.nifi.reporting.ComponentType;
import org.apache.nifi.reporting.InitializationException;
import org.apache.nifi.reporting.ReportingInitializationContext;
import org.apache.nifi.reporting.ReportingTask;
@@ -202,6 +202,7 @@ import org.apache.nifi.util.ReflectionUtils;
import org.apache.nifi.web.OptimisticLockingManager;
import org.apache.nifi.web.Revision;
import org.apache.nifi.web.UpdateRevision;
+import org.apache.nifi.web.api.dto.BulletinDTO;
import org.apache.nifi.web.api.dto.ComponentStateDTO;
import org.apache.nifi.web.api.dto.ControllerServiceDTO;
import org.apache.nifi.web.api.dto.ControllerServiceReferencingComponentDTO;
@@ -209,7 +210,6 @@ import org.apache.nifi.web.api.dto.DropRequestDTO;
import org.apache.nifi.web.api.dto.FlowFileSummaryDTO;
import org.apache.nifi.web.api.dto.FlowSnippetDTO;
import org.apache.nifi.web.api.dto.ListingRequestDTO;
-import org.apache.nifi.web.api.dto.NodeDTO;
import org.apache.nifi.web.api.dto.ProcessGroupDTO;
import org.apache.nifi.web.api.dto.ProcessorDTO;
import org.apache.nifi.web.api.dto.QueueSizeDTO;
@@ -223,18 +223,23 @@ import org.apache.nifi.web.api.dto.provenance.ProvenanceDTO;
import org.apache.nifi.web.api.dto.provenance.ProvenanceEventDTO;
import org.apache.nifi.web.api.dto.provenance.ProvenanceRequestDTO;
import org.apache.nifi.web.api.dto.provenance.ProvenanceResultsDTO;
-import org.apache.nifi.web.api.dto.status.ClusterStatusHistoryDTO;
-import org.apache.nifi.web.api.dto.status.NodeStatusHistoryDTO;
+import org.apache.nifi.web.api.dto.status.ControllerStatusDTO;
+import org.apache.nifi.web.api.dto.status.ProcessGroupStatusDTO;
+import org.apache.nifi.web.api.dto.status.RemoteProcessGroupStatusDTO;
+import org.apache.nifi.web.api.dto.status.StatusDTO;
import org.apache.nifi.web.api.dto.status.StatusHistoryDTO;
+import org.apache.nifi.web.api.dto.status.StatusMerger;
import org.apache.nifi.web.api.dto.status.StatusSnapshotDTO;
import org.apache.nifi.web.api.entity.ComponentStateEntity;
import org.apache.nifi.web.api.entity.ControllerServiceEntity;
import org.apache.nifi.web.api.entity.ControllerServiceReferencingComponentsEntity;
import org.apache.nifi.web.api.entity.ControllerServicesEntity;
+import org.apache.nifi.web.api.entity.ControllerStatusEntity;
import org.apache.nifi.web.api.entity.DropRequestEntity;
import org.apache.nifi.web.api.entity.FlowSnippetEntity;
import org.apache.nifi.web.api.entity.ListingRequestEntity;
import org.apache.nifi.web.api.entity.ProcessGroupEntity;
+import org.apache.nifi.web.api.entity.ProcessGroupStatusEntity;
import org.apache.nifi.web.api.entity.ProcessorEntity;
import org.apache.nifi.web.api.entity.ProcessorsEntity;
import org.apache.nifi.web.api.entity.ProvenanceEntity;
@@ -243,6 +248,7 @@ import org.apache.nifi.web.api.entity.RemoteProcessGroupEntity;
import org.apache.nifi.web.api.entity.RemoteProcessGroupsEntity;
import org.apache.nifi.web.api.entity.ReportingTaskEntity;
import org.apache.nifi.web.api.entity.ReportingTasksEntity;
+import org.apache.nifi.web.api.entity.StatusHistoryEntity;
import org.apache.nifi.web.util.WebUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -321,6 +327,8 @@ public class WebClusterManager implements HttpClusterManager, ProtocolHandler, C
public static final Pattern REMOTE_PROCESS_GROUP_URI_PATTERN = Pattern.compile("/nifi-api/controller/process-groups/(?:(?:root)|(?:[a-f0-9\\-]{36}))/remote-process-groups/[a-f0-9\\-]{36}");
public static final Pattern PROCESS_GROUP_URI_PATTERN = Pattern.compile("/nifi-api/controller/process-groups/(?:(?:root)|(?:[a-f0-9\\-]{36}))");
+ public static final Pattern GROUP_STATUS_URI_PATTERN = Pattern.compile("/nifi-api/controller/process-groups/(?:(?:root)|(?:[a-f0-9\\-]{36}))/status");
+ public static final Pattern CONTROLLER_STATUS_URI_PATTERN = Pattern.compile("/nifi-api/controller/status");
public static final Pattern TEMPLATE_INSTANCE_URI_PATTERN = Pattern.compile("/nifi-api/controller/process-groups/(?:(?:root)|(?:[a-f0-9\\-]{36}))/template-instance");
public static final Pattern FLOW_SNIPPET_INSTANCE_URI_PATTERN = Pattern.compile("/nifi-api/controller/process-groups/(?:(?:root)|(?:[a-f0-9\\-]{36}))/snippet-instance");
@@ -337,6 +345,15 @@ public class WebClusterManager implements HttpClusterManager, ProtocolHandler, C
public static final Pattern REPORTING_TASK_URI_PATTERN = Pattern.compile("/nifi-api/controller/reporting-tasks/node/[a-f0-9\\-]{36}");
public static final Pattern REPORTING_TASK_STATE_URI_PATTERN = Pattern.compile("/nifi-api/controller/reporting-tasks/node/[a-f0-9\\-]{36}/state");
+ public static final Pattern PROCESSOR_STATUS_HISTORY_URI_PATTERN =
+ Pattern.compile("/nifi-api/controller/process-groups/(?:(?:root)|(?:[a-f0-9\\-]{36}))/processors/[a-f0-9\\-]{36}/status/history");
+ public static final Pattern PROCESS_GROUP_STATUS_HISTORY_URI_PATTERN = Pattern.compile("/nifi-api/controller/process-groups/(?:(?:root)|(?:[a-f0-9\\-]{36}))/status/history");
+ public static final Pattern REMOTE_PROCESS_GROUP_STATUS_HISTORY_URI_PATTERN = Pattern
+ .compile("/nifi-api/controller/process-groups/(?:(?:root)|(?:[a-f0-9\\-]{36}))/remote-process-groups/[a-f0-9\\-]{36}/status/history");
+ public static final Pattern CONNECTION_STATUS_HISTORY_URI_PATTERN = Pattern
+ .compile("/nifi-api/controller/process-groups/(?:(?:root)|(?:[a-f0-9\\-]{36}))/connections/[a-f0-9\\-]{36}/status/history");
+
+
@Deprecated
public static final Pattern QUEUE_CONTENTS_URI = Pattern.compile("/nifi-api/controller/process-groups/(?:(?:root)|(?:[a-f0-9\\-]{36}))/connections/[a-f0-9\\-]{36}/contents");
public static final Pattern DROP_REQUESTS_URI = Pattern.compile("/nifi-api/controller/process-groups/(?:(?:root)|(?:[a-f0-9\\-]{36}))/connections/[a-f0-9\\-]{36}/drop-requests");
@@ -378,7 +395,6 @@ public class WebClusterManager implements HttpClusterManager, ProtocolHandler, C
private final BulletinRepository bulletinRepository;
private final String instanceId;
private final FlowEngine reportingTaskEngine;
- private final Map<NodeIdentifier, ComponentStatusRepository> componentMetricsRepositoryMap = new HashMap<>();
private final StandardProcessScheduler processScheduler;
private final StateManagerProvider stateManagerProvider;
private final long componentStatusSnapshotMillis;
@@ -463,7 +479,6 @@ public class WebClusterManager implements HttpClusterManager, ProtocolHandler, C
processScheduler.setSchedulingAgent(SchedulingStrategy.CRON_DRIVEN, new QuartzSchedulingAgent(null, reportingTaskEngine, null, encryptor));
processScheduler.setMaxThreadCount(SchedulingStrategy.TIMER_DRIVEN, 10);
processScheduler.setMaxThreadCount(SchedulingStrategy.CRON_DRIVEN, 10);
- processScheduler.scheduleFrameworkTask(new CaptureComponentMetrics(), "Capture Component Metrics", componentStatusSnapshotMillis, componentStatusSnapshotMillis, TimeUnit.MILLISECONDS);
controllerServiceProvider = new StandardControllerServiceProvider(processScheduler, bulletinRepository, stateManagerProvider);
}
@@ -620,7 +635,6 @@ public class WebClusterManager implements HttpClusterManager, ProtocolHandler, C
return MessageType.CONNECTION_REQUEST == msg.getType()
|| MessageType.HEARTBEAT == msg.getType()
|| MessageType.CONTROLLER_STARTUP_FAILURE == msg.getType()
- || MessageType.BULLETINS == msg.getType()
|| MessageType.RECONNECTION_FAILURE == msg.getType();
}
@@ -654,10 +668,6 @@ public class WebClusterManager implements HttpClusterManager, ProtocolHandler, C
}
}, "Handle Reconnection Failure Message from " + ((ReconnectionFailureMessage) protocolMessage).getNodeId()).start();
return null;
- case BULLETINS:
- final NodeBulletinsMessage bulletinsMessage = (NodeBulletinsMessage) protocolMessage;
- handleBulletins(bulletinsMessage.getBulletins());
- return null;
default:
throw new ProtocolException("No handler defined for message type: " + protocolMessage.getType());
}
@@ -1686,22 +1696,6 @@ public class WebClusterManager implements HttpClusterManager, ProtocolHandler, C
processScheduler.enableReportingTask(reportingTask);
}
- /**
- * Handle a bulletins message.
- *
- * @param bulletins bulletins
- */
- public void handleBulletins(final NodeBulletins bulletins) {
- final NodeIdentifier nodeIdentifier = bulletins.getNodeIdentifier();
- final String nodeAddress = nodeIdentifier.getApiAddress() + ":" + nodeIdentifier.getApiPort();
-
- // unmarshal the message
- final BulletinsPayload payload = BulletinsPayload.unmarshal(bulletins.getPayload());
- for (final Bulletin bulletin : payload.getBulletins()) {
- bulletin.setNodeAddress(nodeAddress);
- bulletinRepository.addBulletin(bulletin);
- }
- }
/**
* Handles a node's heartbeat. If this heartbeat is a node's first heartbeat since its connection request, then the manager will mark the node as connected. If the node was previously disconnected
@@ -2442,6 +2436,14 @@ public class WebClusterManager implements HttpClusterManager, ProtocolHandler, C
return ("GET".equalsIgnoreCase(method) || "PUT".equalsIgnoreCase(method)) && PROCESS_GROUP_URI_PATTERN.matcher(uri.getPath()).matches();
}
+ private static boolean isGroupStatusEndpoint(final URI uri, final String method) {
+ return "GET".equalsIgnoreCase(method) && GROUP_STATUS_URI_PATTERN.matcher(uri.getPath()).matches();
+ }
+
+ private static boolean isControllerStatusEndpoint(final URI uri, final String method) {
+ return "GET".equalsIgnoreCase(method) && CONTROLLER_STATUS_URI_PATTERN.matcher(uri.getPath()).matches();
+ }
+
private static boolean isTemplateEndpoint(final URI uri, final String method) {
return "POST".equalsIgnoreCase(method) && TEMPLATE_INSTANCE_URI_PATTERN.matcher(uri.getPath()).matches();
}
@@ -2454,6 +2456,23 @@ public class WebClusterManager implements HttpClusterManager, ProtocolHandler, C
return "GET".equalsIgnoreCase(method) && REMOTE_PROCESS_GROUPS_URI_PATTERN.matcher(uri.getPath()).matches();
}
+ private static boolean isProcessorStatusHistoryEndpoint(final URI uri, final String method) {
+ return "GET".equalsIgnoreCase(method) && PROCESSOR_STATUS_HISTORY_URI_PATTERN.matcher(uri.getPath()).matches();
+ }
+
+ private static boolean isProcessGroupStatusHistoryEndpoint(final URI uri, final String method) {
+ return "GET".equalsIgnoreCase(method) && PROCESS_GROUP_STATUS_HISTORY_URI_PATTERN.matcher(uri.getPath()).matches();
+ }
+
+ private static boolean isRemoteProcessGroupStatusHistoryEndpoint(final URI uri, final String method) {
+ return "GET".equalsIgnoreCase(method) && REMOTE_PROCESS_GROUP_STATUS_HISTORY_URI_PATTERN.matcher(uri.getPath()).matches();
+ }
+
+ private static boolean isConnectionStatusHistoryEndpoint(final URI uri, final String method) {
+ return "GET".equalsIgnoreCase(method) && CONNECTION_STATUS_HISTORY_URI_PATTERN.matcher(uri.getPath()).matches();
+ }
+
+
private static boolean isRemoteProcessGroupEndpoint(final URI uri, final String method) {
if (("GET".equalsIgnoreCase(method) || "PUT".equalsIgnoreCase(method)) && REMOTE_PROCESS_GROUP_URI_PATTERN.matcher(uri.getPath()).matches()) {
return true;
@@ -2556,7 +2575,10 @@ public class WebClusterManager implements HttpClusterManager, ProtocolHandler, C
|| isControllerServicesEndpoint(uri, method) || isControllerServiceEndpoint(uri, method)
|| isControllerServiceReferenceEndpoint(uri, method) || isControllerServiceStateEndpoint(uri, method)
|| isReportingTasksEndpoint(uri, method) || isReportingTaskEndpoint(uri, method) || isReportingTaskStateEndpoint(uri, method)
- || isDropRequestEndpoint(uri, method) || isListFlowFilesEndpoint(uri, method);
+ || isDropRequestEndpoint(uri, method) || isListFlowFilesEndpoint(uri, method)
+ || isGroupStatusEndpoint(uri, method) || isControllerStatusEndpoint(uri, method)
+ || isProcessorStatusHistoryEndpoint(uri, method) || isProcessGroupStatusHistoryEndpoint(uri, method)
+ || isRemoteProcessGroupStatusHistoryEndpoint(uri, method) || isConnectionStatusHistoryEndpoint(uri, method);
}
private void mergeProcessorValidationErrors(final ProcessorDTO processor, Map<NodeIdentifier, ProcessorDTO> processorMap) {
@@ -2608,6 +2630,159 @@ public class WebClusterManager implements HttpClusterManager, ProtocolHandler, C
componentState.getLocalState().setState(localStateEntries);
}
+ @SuppressWarnings("unchecked")
+ private void updateBulletins(final String nodeAddress, final Collection<? extends StatusDTO>... dtos) {
+ for (final Collection<? extends StatusDTO> collection : dtos) {
+ if (collection != null) {
+ for (final StatusDTO dto : collection) {
+ final List<BulletinDTO> bulletins = dto.getBulletins();
+ if (bulletins != null) {
+ for (final BulletinDTO bulletin : bulletins) {
+ bulletin.setNodeAddress(nodeAddress);
+ }
+ }
+ }
+ }
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private void updateBulletins(final ProcessGroupStatusDTO dto, final String nodeAddress) {
+ for (final BulletinDTO bulletin : dto.getBulletins()) {
+ bulletin.setNodeAddress(nodeAddress);
+ }
+
+ updateBulletins(nodeAddress, dto.getProcessorStatus(), dto.getInputPortStatus(), dto.getOutputPortStatus(), dto.getRemoteProcessGroupStatus());
+
+ if (dto.getProcessGroupStatus() != null) {
+ for (final ProcessGroupStatusDTO childGroup : dto.getProcessGroupStatus()) {
+ updateBulletins(childGroup, nodeAddress);
+ }
+ }
+ }
+
+ private void mergeGroupStatus(final ProcessGroupStatusDTO statusDto, final Map<NodeIdentifier, ProcessGroupStatusDTO> resultMap) {
+ ProcessGroupStatusDTO mergedProcessGroupStatus = statusDto;
+ for (final Map.Entry<NodeIdentifier, ProcessGroupStatusDTO> entry : resultMap.entrySet()) {
+ final NodeIdentifier nodeId = entry.getKey();
+ final ProcessGroupStatusDTO nodeProcessGroupStatus = entry.getValue();
+
+ final String nodeAddress = nodeId.getApiAddress() + ":" + nodeId.getApiPort();
+ updateBulletins(mergedProcessGroupStatus, nodeAddress);
+
+ if (nodeProcessGroupStatus == mergedProcessGroupStatus) {
+ continue;
+ }
+
+ final ProcessGroupStatusDTO nodeClone = nodeProcessGroupStatus.clone();
+ for (final RemoteProcessGroupStatusDTO remoteProcessGroupStatus : nodeClone.getRemoteProcessGroupStatus()) {
+ final List<String> nodeAuthorizationIssues = remoteProcessGroupStatus.getAuthorizationIssues();
+ if (!nodeAuthorizationIssues.isEmpty()) {
+ for (final ListIterator<String> iter = nodeAuthorizationIssues.listIterator(); iter.hasNext();) {
+ final String Issue = iter.next();
+ iter.set("[" + nodeId.getApiAddress() + ":" + nodeId.getApiPort() + "] -- " + Issue);
+ }
+ remoteProcessGroupStatus.setAuthorizationIssues(nodeAuthorizationIssues);
+ }
+ }
+
+ StatusMerger.merge(mergedProcessGroupStatus, nodeClone);
+ }
+
+ StatusMerger.updatePrettyPrintedFields(mergedProcessGroupStatus);
+ }
+
+
+ private void mergeControllerStatus(final ControllerStatusDTO statusDto, final Map<NodeIdentifier, ControllerStatusDTO> resultMap) {
+ ControllerStatusDTO mergedStatus = statusDto;
+ for (final Map.Entry<NodeIdentifier, ControllerStatusDTO> entry : resultMap.entrySet()) {
+ final NodeIdentifier nodeId = entry.getKey();
+ final ControllerStatusDTO nodeStatus = entry.getValue();
+
+ final String nodeAddress = nodeId.getApiAddress() + ":" + nodeId.getApiPort();
+ for (final BulletinDTO bulletin : nodeStatus.getBulletins()) {
+ bulletin.setNodeAddress(nodeAddress);
+ }
+ for (final BulletinDTO bulletin : nodeStatus.getControllerServiceBulletins()) {
+ bulletin.setNodeAddress(nodeAddress);
+ }
+ for (final BulletinDTO bulletin : nodeStatus.getReportingTaskBulletins()) {
+ bulletin.setNodeAddress(nodeAddress);
+ }
+
+ if (nodeStatus == mergedStatus) {
+ continue;
+ }
+
+ StatusMerger.merge(mergedStatus, nodeStatus);
+ }
+
+ final int totalNodeCount = getNodeIds().size();
+ final int connectedNodeCount = getNodeIds(Status.CONNECTED).size();
+
+ final List<Bulletin> ncmControllerBulletins = getBulletinRepository().findBulletinsForController();
+ mergedStatus.setBulletins(mergeNCMBulletins(mergedStatus.getBulletins(), ncmControllerBulletins));
+
+ // get the controller service bulletins
+ final BulletinQuery controllerServiceQuery = new BulletinQuery.Builder().sourceType(ComponentType.CONTROLLER_SERVICE).build();
+ final List<Bulletin> ncmServiceBulletins = getBulletinRepository().findBulletins(controllerServiceQuery);
+ mergedStatus.setControllerServiceBulletins(mergeNCMBulletins(mergedStatus.getControllerServiceBulletins(), ncmServiceBulletins));
+
+ // get the reporting task bulletins
+ final BulletinQuery reportingTaskQuery = new BulletinQuery.Builder().sourceType(ComponentType.REPORTING_TASK).build();
+ final List<Bulletin> ncmReportingTaskBulletins = getBulletinRepository().findBulletins(reportingTaskQuery);
+ mergedStatus.setReportingTaskBulletins(mergeNCMBulletins(mergedStatus.getReportingTaskBulletins(), ncmReportingTaskBulletins));
+
+ mergedStatus.setConnectedNodeCount(connectedNodeCount);
+ mergedStatus.setTotalNodeCount(totalNodeCount);
+ StatusMerger.updatePrettyPrintedFields(mergedStatus);
+ }
+
+ private List<BulletinDTO> mergeNCMBulletins(final List<BulletinDTO> nodeBulletins, final List<Bulletin> ncmBulletins) {
+ if (ncmBulletins == null || ncmBulletins.isEmpty()) {
+ return nodeBulletins;
+ }
+
+ final List<BulletinDTO> mergedBulletins = new ArrayList<>(nodeBulletins.size() + ncmBulletins.size());
+ mergedBulletins.addAll(nodeBulletins);
+ mergedBulletins.addAll(createBulletinDtos(ncmBulletins));
+ return mergedBulletins;
+ }
+
+ /**
+ * Creates BulletinDTOs for the specified Bulletins.
+ *
+ * @param bulletins bulletin
+ * @return dto
+ */
+ public List<BulletinDTO> createBulletinDtos(final List<Bulletin> bulletins) {
+ final List<BulletinDTO> bulletinDtos = new ArrayList<>(bulletins.size());
+ for (final Bulletin bulletin : bulletins) {
+ bulletinDtos.add(createBulletinDto(bulletin));
+ }
+ return bulletinDtos;
+ }
+
+ /**
+ * Creates a BulletinDTO for the specified Bulletin.
+ *
+ * @param bulletin bulletin
+ * @return dto
+ */
+ public BulletinDTO createBulletinDto(final Bulletin bulletin) {
+ final BulletinDTO dto = new BulletinDTO();
+ dto.setId(bulletin.getId());
+ dto.setNodeAddress(bulletin.getNodeAddress());
+ dto.setTimestamp(bulletin.getTimestamp());
+ dto.setGroupId(bulletin.getGroupId());
+ dto.setSourceId(bulletin.getSourceId());
+ dto.setSourceName(bulletin.getSourceName());
+ dto.setCategory(bulletin.getCategory());
+ dto.setLevel(bulletin.getLevel());
+ dto.setMessage(bulletin.getMessage());
+ return dto;
+ }
+
private void mergeProvenanceQueryResults(final ProvenanceDTO provenanceDto, final Map<NodeIdentifier, ProvenanceDTO> resultMap, final Set<NodeResponse> problematicResponses) {
final ProvenanceResultsDTO results = provenanceDto.getResults();
final ProvenanceRequestDTO request = provenanceDto.getRequest();
@@ -3545,6 +3720,70 @@ public class WebClusterManager implements HttpClusterManager, ProtocolHandler, C
mergeComponentState(componentState, resultsMap);
clientResponse = new NodeResponse(clientResponse, responseEntity);
+ } else if (hasSuccessfulClientResponse && isGroupStatusEndpoint(uri, method)) {
+ final ProcessGroupStatusEntity responseEntity = clientResponse.getClientResponse().getEntity(ProcessGroupStatusEntity.class);
+ final ProcessGroupStatusDTO statusRequest = responseEntity.getProcessGroupStatus();
+
+ final Map<NodeIdentifier, ProcessGroupStatusDTO> resultsMap = new HashMap<>();
+ for (final NodeResponse nodeResponse : updatedNodesMap.values()) {
+ if (problematicNodeResponses.contains(nodeResponse)) {
+ continue;
+ }
+
+ final ProcessGroupStatusEntity nodeResponseEntity = nodeResponse == clientResponse ? responseEntity : nodeResponse.getClientResponse().getEntity(ProcessGroupStatusEntity.class);
+ final ProcessGroupStatusDTO nodeStatus = nodeResponseEntity.getProcessGroupStatus();
+
+ resultsMap.put(nodeResponse.getNodeId(), nodeStatus);
+ }
+ mergeGroupStatus(statusRequest, resultsMap);
+
+ clientResponse = new NodeResponse(clientResponse, responseEntity);
+ } else if (hasSuccessfulClientResponse && isControllerStatusEndpoint(uri, method)) {
+ final ControllerStatusEntity responseEntity = clientResponse.getClientResponse().getEntity(ControllerStatusEntity.class);
+ final ControllerStatusDTO statusRequest = responseEntity.getControllerStatus();
+
+ final Map<NodeIdentifier, ControllerStatusDTO> resultsMap = new HashMap<>();
+ for (final NodeResponse nodeResponse : updatedNodesMap.values()) {
+ if (problematicNodeResponses.contains(nodeResponse)) {
+ continue;
+ }
+
+ final ControllerStatusEntity nodeResponseEntity = nodeResponse == clientResponse ? responseEntity : nodeResponse.getClientResponse().getEntity(ControllerStatusEntity.class);
+ final ControllerStatusDTO nodeStatus = nodeResponseEntity.getControllerStatus();
+
+ resultsMap.put(nodeResponse.getNodeId(), nodeStatus);
+ }
+ mergeControllerStatus(statusRequest, resultsMap);
+
+ clientResponse = new NodeResponse(clientResponse, responseEntity);
+ } else if (hasSuccessfulClientResponse && isProcessorStatusHistoryEndpoint(uri, method)) {
+ final Map<String, MetricDescriptor<?>> metricDescriptors = new HashMap<>();
+ for (final ProcessorStatusDescriptor descriptor : ProcessorStatusDescriptor.values()) {
+ metricDescriptors.put(descriptor.getField(), descriptor.getDescriptor());
+ }
+
+ clientResponse = mergeStatusHistoryResponses(clientResponse, updatedNodesMap, problematicNodeResponses, metricDescriptors);
+ } else if (hasSuccessfulClientResponse && isConnectionStatusHistoryEndpoint(uri, method)) {
+ final Map<String, MetricDescriptor<?>> metricDescriptors = new HashMap<>();
+ for (final ConnectionStatusDescriptor descriptor : ConnectionStatusDescriptor.values()) {
+ metricDescriptors.put(descriptor.getField(), descriptor.getDescriptor());
+ }
+
+ clientResponse = mergeStatusHistoryResponses(clientResponse, updatedNodesMap, problematicNodeResponses, metricDescriptors);
+ } else if (hasSuccessfulClientResponse && isProcessGroupStatusHistoryEndpoint(uri, method)) {
+ final Map<String, MetricDescriptor<?>> metricDescriptors = new HashMap<>();
+ for (final ProcessGroupStatusDescriptor descriptor : ProcessGroupStatusDescriptor.values()) {
+ metricDescriptors.put(descriptor.getField(), descriptor.getDescriptor());
+ }
+
+ clientResponse = mergeStatusHistoryResponses(clientResponse, updatedNodesMap, problematicNodeResponses, metricDescriptors);
+ } else if (hasSuccessfulClientResponse && isRemoteProcessGroupStatusHistoryEndpoint(uri, method)) {
+ final Map<String, MetricDescriptor<?>> metricDescriptors = new HashMap<>();
+ for (final RemoteProcessGroupStatusDescriptor descriptor : RemoteProcessGroupStatusDescriptor.values()) {
+ metricDescriptors.put(descriptor.getField(), descriptor.getDescriptor());
+ }
+
+ clientResponse = mergeStatusHistoryResponses(clientResponse, updatedNodesMap, problematicNodeResponses, metricDescriptors);
} else {
if (!nodeResponsesToDrain.isEmpty()) {
drainResponses(nodeResponsesToDrain);
@@ -3603,6 +3842,44 @@ public class WebClusterManager implements HttpClusterManager, ProtocolHandler, C
return clientResponse;
}
+
+ private NodeResponse mergeStatusHistoryResponses(NodeResponse clientResponse, Map<Node, NodeResponse> updatedNodesMap, Set<NodeResponse> problematicNodeResponses,
+ Map<String, MetricDescriptor<?>> metricDescriptors) {
+ final StatusHistoryEntity responseEntity = clientResponse.getClientResponse().getEntity(StatusHistoryEntity.class);
+
+ StatusHistoryDTO lastStatusHistory = null;
+ final Map<String, List<StatusSnapshotDTO>> nodeStatusHistories = new HashMap<>(updatedNodesMap.size());
+ for (final NodeResponse nodeResponse : updatedNodesMap.values()) {
+ if (problematicNodeResponses.contains(nodeResponse)) {
+ continue;
+ }
+
+ final StatusHistoryEntity nodeResponseEntity = nodeResponse == clientResponse ? responseEntity : nodeResponse.getClientResponse().getEntity(StatusHistoryEntity.class);
+ final StatusHistoryDTO nodeStatus = nodeResponseEntity.getStatusHistory();
+ lastStatusHistory = nodeStatus;
+
+ final NodeIdentifier nodeId = nodeResponse.getNodeId();
+ final String nodeName = nodeId.getApiAddress() + ":" + nodeId.getApiPort();
+ nodeStatusHistories.put(nodeName, nodeStatus.getAggregateStatusSnapshots());
+ }
+
+ final StatusHistoryDTO clusterStatusHistory = new StatusHistoryDTO();
+ clusterStatusHistory.setAggregateStatusSnapshots(mergeStatusHistories(nodeStatusHistories, metricDescriptors));
+ clusterStatusHistory.setGenerated(new Date());
+ clusterStatusHistory.setNodeStatusSnapshots(nodeStatusHistories);
+ if (lastStatusHistory != null) {
+ clusterStatusHistory.setComponentDetails(lastStatusHistory.getComponentDetails());
+ clusterStatusHistory.setFieldDescriptors(lastStatusHistory.getFieldDescriptors());
+ }
+
+ final StatusHistoryEntity clusterEntity = new StatusHistoryEntity();
+ clusterEntity.setStatusHistory(clusterStatusHistory);
+ clusterEntity.setRevision(responseEntity.getRevision());
+
+ return new NodeResponse(clientResponse, clusterEntity);
+ }
+
+
/**
* Determines if all problematic responses were due to 404 NOT_FOUND. Assumes that problematicNodeResponses is not empty and is not comprised of responses from all nodes in the cluster (at least
* one node contained the counter in question).
@@ -4026,78 +4303,6 @@ public class WebClusterManager implements HttpClusterManager, ProtocolHandler, C
return bulletinRepository;
}
- @Override
- public ProcessGroupStatus getProcessGroupStatus(final String groupId) {
- final Set<Node> connectedNodes = getNodes(Node.Status.CONNECTED);
-
- // ensure there are some nodes in the cluster
- if (connectedNodes.isEmpty()) {
- throw new NoConnectedNodesException();
- }
-
- ProcessGroupStatus mergedProcessGroupStatus = null;
- for (final Node node : connectedNodes) {
- final NodeIdentifier nodeId = node.getNodeId();
- final HeartbeatPayload nodeHeartbeatPayload = node.getHeartbeatPayload();
- if (nodeHeartbeatPayload == null) {
- continue;
- }
- final ProcessGroupStatus nodeRootProcessGroupStatus = nodeHeartbeatPayload.getProcessGroupStatus();
- final ProcessGroupStatus nodeProcessGroupStatus = groupId.equals(ROOT_GROUP_ID_ALIAS) ? nodeRootProcessGroupStatus : getProcessGroupStatus(nodeRootProcessGroupStatus, groupId);
- if (nodeProcessGroupStatus == null) {
- continue;
- }
-
- if (mergedProcessGroupStatus == null) {
- mergedProcessGroupStatus = nodeProcessGroupStatus.clone();
-
- // update any issues with the node label
- if (mergedProcessGroupStatus.getRemoteProcessGroupStatus() != null) {
- for (final RemoteProcessGroupStatus remoteProcessGroupStatus : mergedProcessGroupStatus.getRemoteProcessGroupStatus()) {
- final List<String> nodeAuthorizationIssues = remoteProcessGroupStatus.getAuthorizationIssues();
- if (!nodeAuthorizationIssues.isEmpty()) {
- for (final ListIterator<String> iter = nodeAuthorizationIssues.listIterator(); iter.hasNext();) {
- final String Issue = iter.next();
- iter.set("[" + nodeId.getApiAddress() + ":" + nodeId.getApiPort() + "] -- " + Issue);
- }
- remoteProcessGroupStatus.setAuthorizationIssues(nodeAuthorizationIssues);
- }
- }
- }
- } else {
- final ProcessGroupStatus nodeClone = nodeProcessGroupStatus.clone();
- for (final RemoteProcessGroupStatus remoteProcessGroupStatus : nodeClone.getRemoteProcessGroupStatus()) {
- final List<String> nodeAuthorizationIssues = remoteProcessGroupStatus.getAuthorizationIssues();
- if (!nodeAuthorizationIssues.isEmpty()) {
- for (final ListIterator<String> iter = nodeAuthorizationIssues.listIterator(); iter.hasNext();) {
- final String Issue = iter.next();
- iter.set("[" + nodeId.getApiAddress() + ":" + nodeId.getApiPort() + "] -- " + Issue);
- }
- remoteProcessGroupStatus.setAuthorizationIssues(nodeAuthorizationIssues);
- }
- }
-
- ProcessGroupStatus.merge(mergedProcessGroupStatus, nodeClone);
- }
- }
-
- return mergedProcessGroupStatus;
- }
-
- private ProcessGroupStatus getProcessGroupStatus(final ProcessGroupStatus parent, final String groupId) {
- if (parent.getId().equals(groupId)) {
- return parent;
- }
-
- for (final ProcessGroupStatus child : parent.getProcessGroupStatus()) {
- final ProcessGroupStatus matching = getProcessGroupStatus(child, groupId);
- if (matching != null) {
- return matching;
- }
- }
-
- return null;
- }
@Override
public SystemDiagnostics getSystemDiagnostics() {
@@ -4214,19 +4419,6 @@ public class WebClusterManager implements HttpClusterManager, ProtocolHandler, C
return new Date(time - time % numMillis);
}
- private NodeDTO createNodeDTO(final Node node) {
- final NodeDTO nodeDto = new NodeDTO();
- final NodeIdentifier nodeId = node.getNodeId();
- nodeDto.setNodeId(nodeId.getId());
- nodeDto.setAddress(nodeId.getApiAddress());
- nodeDto.setApiPort(nodeId.getApiPort());
- nodeDto.setStatus(node.getStatus().name());
- nodeDto.setPrimary(node.equals(getPrimaryNode()));
- final Date connectionRequested = new Date(node.getConnectionRequestedTimestamp());
- nodeDto.setConnectionRequested(connectionRequested);
-
- return nodeDto;
- }
private List<StatusSnapshotDTO> aggregate(Map<Date, List<StatusSnapshot>> snapshotsToAggregate) {
// Aggregate the snapshots
@@ -4245,185 +4437,34 @@ public class WebClusterManager implements HttpClusterManager, ProtocolHandler, C
return aggregatedSnapshotDtos;
}
- public ClusterStatusHistoryDTO getProcessorStatusHistory(final String processorId) {
- return getProcessorStatusHistory(processorId, null, null, Integer.MAX_VALUE);
- }
-
- public ClusterStatusHistoryDTO getProcessorStatusHistory(final String processorId, final Date startDate, final Date endDate, final int preferredDataPoints) {
- final List<NodeStatusHistoryDTO> nodeHistories = new ArrayList<>();
-
- StatusHistoryDTO lastStatusHistory = null;
- final Set<MetricDescriptor<?>> processorDescriptors = new LinkedHashSet<>();
- final Map<Date, List<StatusSnapshot>> snapshotsToAggregate = new TreeMap<>();
- for (final Node node : getRawNodes()) {
- final ComponentStatusRepository statusRepository = componentMetricsRepositoryMap.get(node.getNodeId());
- if (statusRepository == null) {
- continue;
- }
-
- final StatusHistory statusHistory = statusRepository.getProcessorStatusHistory(processorId, startDate, endDate, preferredDataPoints);
- if (statusHistory == null) {
- continue;
- }
+ private StatusSnapshot createSnapshot(final StatusSnapshotDTO snapshotDto, final Map<String, MetricDescriptor<?>> metricDescriptors) {
+ final StandardStatusSnapshot snapshot = new StandardStatusSnapshot();
+ snapshot.setTimestamp(snapshotDto.getTimestamp());
- processorDescriptors.addAll(statusRepository.getProcessorMetricDescriptors());
+ final Map<String, Long> metrics = snapshotDto.getStatusMetrics();
+ for (final Map.Entry<String, Long> entry : metrics.entrySet()) {
+ final String metricId = entry.getKey();
+ final Long value = entry.getValue();
- // record the status history (last) to get the component details for use later
- final StatusHistoryDTO statusHistoryDto = createStatusHistoryDto(statusHistory);
- lastStatusHistory = statusHistoryDto;
-
- final NodeStatusHistoryDTO nodeHistory = new NodeStatusHistoryDTO();
- nodeHistory.setStatusHistory(statusHistoryDto);
- nodeHistory.setNode(createNodeDTO(node));
- nodeHistories.add(nodeHistory);
-
- // collect all of the snapshots to aggregate
- for (final StatusSnapshot snapshot : statusHistory.getStatusSnapshots()) {
- final Date normalizedDate = normalizeStatusSnapshotDate(snapshot.getTimestamp(), componentStatusSnapshotMillis);
- List<StatusSnapshot> snapshots = snapshotsToAggregate.get(normalizedDate);
- if (snapshots == null) {
- snapshots = new ArrayList<>();
- snapshotsToAggregate.put(normalizedDate, snapshots);
- }
- snapshots.add(snapshot);
- }
- }
-
- // Aggregate the snapshots
- final List<StatusSnapshotDTO> aggregatedSnapshotDtos = aggregate(snapshotsToAggregate);
-
- // get the details for this component from the last status history
- final LinkedHashMap<String, String> clusterStatusHistoryDetails = new LinkedHashMap<>();
- clusterStatusHistoryDetails.putAll(lastStatusHistory.getDetails());
-
- final StatusHistoryDTO clusterStatusHistory = new StatusHistoryDTO();
- clusterStatusHistory.setGenerated(new Date());
- clusterStatusHistory.setFieldDescriptors(StatusHistoryUtil.createFieldDescriptorDtos(processorDescriptors));
- clusterStatusHistory.setDetails(clusterStatusHistoryDetails);
- clusterStatusHistory.setStatusSnapshots(aggregatedSnapshotDtos);
-
- final ClusterStatusHistoryDTO history = new ClusterStatusHistoryDTO();
- history.setGenerated(new Date());
- history.setNodeStatusHistory(nodeHistories);
- history.setClusterStatusHistory(clusterStatusHistory);
- return history;
- }
-
- public StatusHistoryDTO createStatusHistoryDto(final StatusHistory statusHistory) {
- final StatusHistoryDTO dto = new StatusHistoryDTO();
-
- dto.setDetails(new LinkedHashMap<>(statusHistory.getComponentDetails()));
- dto.setFieldDescriptors(StatusHistoryUtil.createFieldDescriptorDtos(statusHistory));
- dto.setGenerated(statusHistory.getDateGenerated());
-
- final List<StatusSnapshotDTO> statusSnapshots = new ArrayList<>();
- for (final StatusSnapshot statusSnapshot : statusHistory.getStatusSnapshots()) {
- statusSnapshots.add(StatusHistoryUtil.createStatusSnapshotDto(statusSnapshot));
- }
- dto.setStatusSnapshots(statusSnapshots);
-
- return dto;
- }
-
- public ClusterStatusHistoryDTO getConnectionStatusHistory(final String connectionId) {
- return getConnectionStatusHistory(connectionId, null, null, Integer.MAX_VALUE);
- }
-
- public ClusterStatusHistoryDTO getConnectionStatusHistory(final String connectionId, final Date startDate, final Date endDate, final int preferredDataPoints) {
- final List<NodeStatusHistoryDTO> nodeHistories = new ArrayList<>();
-
- StatusHistoryDTO lastStatusHistory = null;
- final Set<MetricDescriptor<?>> connectionDescriptors = new LinkedHashSet<>();
- final Map<Date, List<StatusSnapshot>> snapshotsToAggregate = new TreeMap<>();
-
- for (final Node node : getRawNodes()) {
- final ComponentStatusRepository statusRepository = componentMetricsRepositoryMap.get(node.getNodeId());
- if (statusRepository == null) {
- continue;
- }
-
- final StatusHistory statusHistory = statusRepository.getConnectionStatusHistory(connectionId, startDate, endDate, preferredDataPoints);
- if (statusHistory == null) {
- continue;
- }
-
- final StatusHistoryDTO statusHistoryDto = createStatusHistoryDto(statusHistory);
- // record the status history (last) to get the componet details for use later
- lastStatusHistory = statusHistoryDto;
- connectionDescriptors.addAll(statusRepository.getConnectionMetricDescriptors());
-
- final NodeStatusHistoryDTO nodeHistory = new NodeStatusHistoryDTO();
- nodeHistory.setStatusHistory(statusHistoryDto);
- nodeHistory.setNode(createNodeDTO(node));
- nodeHistories.add(nodeHistory);
-
- // collect all of the snapshots to aggregate
- for (final StatusSnapshot snapshot : statusHistory.getStatusSnapshots()) {
- final Date normalizedDate = normalizeStatusSnapshotDate(snapshot.getTimestamp(), componentStatusSnapshotMillis);
- List<StatusSnapshot> snapshots = snapshotsToAggregate.get(normalizedDate);
- if (snapshots == null) {
- snapshots = new ArrayList<>();
- snapshotsToAggregate.put(normalizedDate, snapshots);
- }
- snapshots.add(snapshot);
+ final MetricDescriptor<?> descriptor = metricDescriptors.get(metricId);
+ if (descriptor != null) {
+ snapshot.addStatusMetric(descriptor, value);
}
}
- // Aggregate the snapshots
- final List<StatusSnapshotDTO> aggregatedSnapshotDtos = aggregate(snapshotsToAggregate);
-
- // get the details for this component from the last status history
- final LinkedHashMap<String, String> clusterStatusHistoryDetails = new LinkedHashMap<>();
- clusterStatusHistoryDetails.putAll(lastStatusHistory.getDetails());
-
- final StatusHistoryDTO clusterStatusHistory = new StatusHistoryDTO();
- clusterStatusHistory.setGenerated(new Date());
- clusterStatusHistory.setFieldDescriptors(StatusHistoryUtil.createFieldDescriptorDtos(connectionDescriptors));
- clusterStatusHistory.setDetails(clusterStatusHistoryDetails);
- clusterStatusHistory.setStatusSnapshots(aggregatedSnapshotDtos);
-
- final ClusterStatusHistoryDTO history = new ClusterStatusHistoryDTO();
- history.setGenerated(new Date());
- history.setNodeStatusHistory(nodeHistories);
- history.setClusterStatusHistory(clusterStatusHistory);
- return history;
- }
-
- public ClusterStatusHistoryDTO getProcessGroupStatusHistory(final String processGroupId) {
- return getProcessGroupStatusHistory(processGroupId, null, null, Integer.MAX_VALUE);
+ return snapshot;
}
- public ClusterStatusHistoryDTO getProcessGroupStatusHistory(final String processGroupId, final Date startDate, final Date endDate, final int preferredDataPoints) {
- final List<NodeStatusHistoryDTO> nodeHistories = new ArrayList<>();
-
- StatusHistoryDTO lastStatusHistory = null;
- final Set<MetricDescriptor<?>> processGroupDescriptors = new LinkedHashSet<>();
+ private List<StatusSnapshotDTO> mergeStatusHistories(final Map<String, List<StatusSnapshotDTO>> nodeStatusHistories, final Map<String, MetricDescriptor<?>> metricDescriptors) {
+ // Map of "normalized Date" (i.e., a time range, essentially) to all Snapshots for that time. The list
+ // will contain one snapshot for each node.
final Map<Date, List<StatusSnapshot>> snapshotsToAggregate = new TreeMap<>();
- for (final Node node : getRawNodes()) {
- final ComponentStatusRepository statusRepository = componentMetricsRepositoryMap.get(node.getNodeId());
- if (statusRepository == null) {
- continue;
- }
-
- final StatusHistory statusHistory = statusRepository.getProcessGroupStatusHistory(processGroupId, startDate, endDate, preferredDataPoints);
- if (statusHistory == null) {
- continue;
- }
-
- final StatusHistoryDTO statusHistoryDto = createStatusHistoryDto(statusHistory);
- // record the status history (last) to get the componet details for use later
- lastStatusHistory = statusHistoryDto;
- processGroupDescriptors.addAll(statusRepository.getProcessGroupMetricDescriptors());
-
- final NodeStatusHistoryDTO nodeHistory = new NodeStatusHistoryDTO();
- nodeHistory.setStatusHistory(statusHistoryDto);
- nodeHistory.setNode(createNodeDTO(node));
- nodeHistories.add(nodeHistory);
-
- // collect all of the snapshots to aggregate
- for (final StatusSnapshot snapshot : statusHistory.getStatusSnapshots()) {
+ // group status snapshot's for each node by date
+ for (final List<StatusSnapshotDTO> snapshotDtos : nodeStatusHistories.values()) {
+ for (final StatusSnapshotDTO snapshotDto : snapshotDtos) {
+ final StatusSnapshot snapshot = createSnapshot(snapshotDto, metricDescriptors);
final Date normalizedDate = normalizeStatusSnapshotDate(snapshot.getTimestamp(), componentStatusSnapshotMillis);
List<StatusSnapshot> snapshots = snapshotsToAggregate.get(normalizedDate);
if (snapshots == null) {
@@ -4434,89 +4475,12 @@ public class WebClusterManager implements HttpClusterManager, ProtocolHandler, C
}
}
- // Aggregate the snapshots
- final List<StatusSnapshotDTO> aggregatedSnapshotDtos = aggregate(snapshotsToAggregate);
-
- // get the details for this component from the last status history
- final LinkedHashMap<String, String> clusterStatusHistoryDetails = new LinkedHashMap<>();
- clusterStatusHistoryDetails.putAll(lastStatusHistory.getDetails());
-
- final StatusHistoryDTO clusterStatusHistory = new StatusHistoryDTO();
- clusterStatusHistory.setGenerated(new Date());
- clusterStatusHistory.setDetails(clusterStatusHistoryDetails);
- clusterStatusHistory.setFieldDescriptors(StatusHistoryUtil.createFieldDescriptorDtos(processGroupDescriptors));
- clusterStatusHistory.setStatusSnapshots(aggregatedSnapshotDtos);
-
- final ClusterStatusHistoryDTO history = new ClusterStatusHistoryDTO();
- history.setGenerated(new Date());
- history.setNodeStatusHistory(nodeHistories);
- history.setClusterStatusHistory(clusterStatusHistory);
- return history;
+ // aggregate the snapshots by (normalized) timestamp
+ final List<StatusSnapshotDTO> aggregatedSnapshots = aggregate(snapshotsToAggregate);
+ return aggregatedSnapshots;
}
- public ClusterStatusHistoryDTO getRemoteProcessGroupStatusHistory(final String remoteGroupId) {
- return getRemoteProcessGroupStatusHistory(remoteGroupId, null, null, Integer.MAX_VALUE);
- }
-
- public ClusterStatusHistoryDTO getRemoteProcessGroupStatusHistory(final String remoteGroupId, final Date startDate, final Date endDate, final int preferredDataPoints) {
- final List<NodeStatusHistoryDTO> nodeHistories = new ArrayList<>();
-
- StatusHistoryDTO lastStatusHistory = null;
- final Set<MetricDescriptor<?>> remoteProcessGroupDescriptors = new LinkedHashSet<>();
- final Map<Date, List<StatusSnapshot>> snapshotsToAggregate = new TreeMap<>();
-
- for (final Node node : getRawNodes()) {
- final ComponentStatusRepository statusRepository = componentMetricsRepositoryMap.get(node.getNodeId());
- if (statusRepository == null) {
- continue;
- }
-
- final StatusHistory statusHistory = statusRepository.getRemoteProcessGroupStatusHistory(remoteGroupId, startDate, endDate, preferredDataPoints);
- if (statusHistory == null) {
- continue;
- }
-
- final StatusHistoryDTO statusHistoryDto = createStatusHistoryDto(statusHistory);
- // record the status history (last) to get the componet details for use later
- lastStatusHistory = statusHistoryDto;
- remoteProcessGroupDescriptors.addAll(statusRepository.getRemoteProcessGroupMetricDescriptors());
- final NodeStatusHistoryDTO nodeHistory = new NodeStatusHistoryDTO();
- nodeHistory.setStatusHistory(statusHistoryDto);
- nodeHistory.setNode(createNodeDTO(node));
- nodeHistories.add(nodeHistory);
-
- // collect all of the snapshots to aggregate
- for (final StatusSnapshot snapshot : statusHistory.getStatusSnapshots()) {
- final Date normalizedDate = normalizeStatusSnapshotDate(snapshot.getTimestamp(), componentStatusSnapshotMillis);
- List<StatusSnapshot> snapshots = snapshotsToAggregate.get(normalizedDate);
- if (snapshots == null) {
- snapshots = new ArrayList<>();
- snapshotsToAggregate.put(normalizedDate, snapshots);
- }
- snapshots.add(snapshot);
- }
- }
-
- // Aggregate the snapshots
- final List<StatusSnapshotDTO> aggregatedSnapshotDtos = aggregate(snapshotsToAggregate);
-
- // get the details for this component from the last status history
- final LinkedHashMap<String, String> clusterStatusHistoryDetails = new LinkedHashMap<>();
- clusterStatusHistoryDetails.putAll(lastStatusHistory.getDetails());
-
- final StatusHistoryDTO clusterStatusHistory = new StatusHistoryDTO();
- clusterStatusHistory.setGenerated(new Date());
- clusterStatusHistory.setDetails(clusterStatusHistoryDetails);
- clusterStatusHistory.setFieldDescriptors(StatusHistoryUtil.createFieldDescriptorDtos(remoteProcessGroupDescriptors));
- clusterStatusHistory.setStatusSnapshots(aggregatedSnapshotDtos);
-
- final ClusterStatusHistoryDTO history = new ClusterStatusHistoryDTO();
- history.setGenerated(new Date());
- history.setNodeStatusHistory(nodeHistories);
- history.setClusterStatusHistory(clusterStatusHistory);
- return history;
- }
private static class ClusterManagerLock {
@@ -4583,41 +4547,4 @@ public class WebClusterManager implements HttpClusterManager, ProtocolHandler, C
public Set<String> getControllerServiceIdentifiers(final Class<? extends ControllerService> serviceType) {
return controllerServiceProvider.getControllerServiceIdentifiers(serviceType);
}
-
- /**
- * Captures snapshots of components' metrics
- */
- private class CaptureComponentMetrics implements Runnable {
- @Override
- public void run() {
- readLock.lock();
- try {
- for (final Node node : nodes) {
- if (Status.CONNECTED.equals(node.getStatus())) {
- ComponentStatusRepository statusRepository = componentMetricsRepositoryMap.get(node.getNodeId());
- if (statusRepository == null) {
- statusRepository = createComponentStatusRepository();
- componentMetricsRepositoryMap.put(node.getNodeId(), statusRepository);
- }
-
- // ensure this node has a payload
- if (node.getHeartbeat() != null && node.getHeartbeatPayload() != null) {
- // if nothing has been captured or the current heartbeat is newer, capture it - comparing the heatbeat created timestamp
- // is safe since its marked as XmlTransient so we're assured that its based off the same clock that created the last capture date
- if (statusRepository.getLastCaptureDate() == null || node.getHeartbeat().getCreatedTimestamp() > statusRepository.getLastCaptureDate().getTime()) {
- statusRepository.capture(node.getHeartbeatPayload().getProcessGroupStatus());
- }
- }
- }
- }
- } catch (final Throwable t) {
- logger.warn("Unable to capture component metrics from Node heartbeats: " + t);
- if (logger.isDebugEnabled()) {
- logger.warn("", t);
- }
- } finally {
- readLock.unlock("capture component metrics from node heartbeats");
- }
- }
- }
}
http://git-wip-us.apache.org/repos/asf/nifi/blob/ad32cb82/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
index 6d09bf6..7ea0408 100644
--- a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
@@ -16,7 +16,40 @@
*/
package org.apache.nifi.controller;
-import com.sun.jersey.api.client.ClientHandlerException;
+import static java.util.Objects.requireNonNull;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.text.DateFormat;
+import java.text.SimpleDateFormat;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Date;
+import java.util.HashSet;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.LockSupport;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+import javax.net.ssl.SSLContext;
+
import org.apache.commons.lang3.StringUtils;
import org.apache.nifi.action.Action;
import org.apache.nifi.admin.service.AuditService;
@@ -27,16 +60,13 @@ import org.apache.nifi.annotation.lifecycle.OnRemoved;
import org.apache.nifi.annotation.lifecycle.OnShutdown;
import org.apache.nifi.annotation.notification.OnPrimaryNodeStateChange;
import org.apache.nifi.annotation.notification.PrimaryNodeState;
-import org.apache.nifi.cluster.BulletinsPayload;
import org.apache.nifi.cluster.HeartbeatPayload;
import org.apache.nifi.cluster.protocol.DataFlow;
import org.apache.nifi.cluster.protocol.Heartbeat;
-import org.apache.nifi.cluster.protocol.NodeBulletins;
import org.apache.nifi.cluster.protocol.NodeIdentifier;
import org.apache.nifi.cluster.protocol.NodeProtocolSender;
import org.apache.nifi.cluster.protocol.UnknownServiceAddressException;
import org.apache.nifi.cluster.protocol.message.HeartbeatMessage;
-import org.apache.nifi.cluster.protocol.message.NodeBulletinsMessage;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.components.state.StateManagerProvider;
import org.apache.nifi.connectable.Connectable;
@@ -109,7 +139,6 @@ import org.apache.nifi.encrypt.StringEncryptor;
import org.apache.nifi.engine.FlowEngine;
import org.apache.nifi.events.BulletinFactory;
import org.apache.nifi.events.EventReporter;
-import org.apache.nifi.events.NodeBulletinProcessingStrategy;
import org.apache.nifi.events.VolatileBulletinRepository;
import org.apache.nifi.flowfile.FlowFilePrioritizer;
import org.apache.nifi.flowfile.attributes.CoreAttributes;
@@ -182,39 +211,7 @@ import org.apache.zookeeper.server.quorum.QuorumPeerConfig.ConfigException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import javax.net.ssl.SSLContext;
-
-import java.io.ByteArrayInputStream;
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
-import java.text.DateFormat;
-import java.text.SimpleDateFormat;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.Date;
-import java.util.HashSet;
-import java.util.LinkedHashSet;
-import java.util.List;
-import java.util.Locale;
-import java.util.Map;
-import java.util.Set;
-import java.util.UUID;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ScheduledFuture;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.concurrent.atomic.AtomicReference;
-import java.util.concurrent.locks.Lock;
-import java.util.concurrent.locks.LockSupport;
-import java.util.concurrent.locks.ReentrantReadWriteLock;
-
-import static java.util.Objects.requireNonNull;
+import com.sun.jersey.api.client.ClientHandlerException;
public class FlowController implements EventAccess, ControllerServiceProvider, ReportingTaskProvider, Heartbeater, QueueProvider {
@@ -306,7 +303,6 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R
/**
* timer to periodically send heartbeats to the cluster
*/
- private ScheduledFuture<?> bulletinFuture;
private ScheduledFuture<?> heartbeatGeneratorFuture;
private ScheduledFuture<?> heartbeatSenderFuture;
@@ -316,8 +312,6 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R
*/
private final AtomicReference<HeartbeatMessageGeneratorTask> heartbeatMessageGeneratorTaskRef = new AtomicReference<>(null);
- private final AtomicReference<NodeBulletinProcessingStrategy> nodeBulletinSubscriber;
-
// guarded by rwLock
/**
* the node identifier;
@@ -418,7 +412,6 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R
counterRepositoryRef = new AtomicReference<CounterRepository>(new StandardCounterRepository());
bulletinRepository = new VolatileBulletinRepository();
- nodeBulletinSubscriber = new AtomicReference<>();
try {
this.provenanceEventRepository = createProvenanceRepository(properties);
@@ -2953,8 +2946,6 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R
stopHeartbeating();
- bulletinFuture = clusterTaskExecutor.scheduleWithFixedDelay(new BulletinsTask(protocolSender), 250, 2000, TimeUnit.MILLISECONDS);
-
final HeartbeatMessageGeneratorTask heartbeatMessageGeneratorTask = new HeartbeatMessageGeneratorTask();
heartbeatMessageGeneratorTaskRef.set(heartbeatMessageGeneratorTask);
heartbeatGeneratorFuture = clusterTaskExecutor.scheduleWithFixedDelay(heartbeatMessageGeneratorTask, 0, heartbeatDelaySeconds, TimeUnit.SECONDS);
@@ -3005,10 +2996,6 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R
if (heartbeatSenderFuture != null) {
heartbeatSenderFuture.cancel(false);
}
-
- if (bulletinFuture != null) {
- bulletinFuture.cancel(false);
- }
} finally {
writeLock.unlock();
}
@@ -3133,8 +3120,6 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R
// update the bulletin repository
if (isChanging) {
if (clustered) {
- nodeBulletinSubscriber.set(new NodeBulletinProcessingStrategy());
- bulletinRepository.overrideDefaultBulletinProcessing(nodeBulletinSubscriber.get());
stateManagerProvider.enableClusterProvider();
if (zooKeeperStateServer != null) {
@@ -3173,7 +3158,6 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R
LockSupport.parkNanos(TimeUnit.SECONDS.toNanos(1L));
}
} else {
- bulletinRepository.restoreDefaultBulletinProcessing();
if (zooKeeperStateServer != null) {
zooKeeperStateServer.shutdown();
}
@@ -3472,6 +3456,7 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R
return replayFlowFile(record, requestor);
}
+ @SuppressWarnings("deprecation")
public ProvenanceEventRecord replayFlowFile(final ProvenanceEventRecord event, final String requestor) throws IOException {
if (event == null) {
throw new NullPointerException();
@@ -3627,110 +3612,6 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R
}
}
- private class BulletinsTask implements Runnable {
-
- private final NodeProtocolSender protocolSender;
- private final DateFormat dateFormatter = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss,SSS", Locale.US);
-
- public BulletinsTask(final NodeProtocolSender protocolSender) {
- if (protocolSender == null) {
- throw new IllegalArgumentException("NodeProtocolSender may not be null.");
- }
- this.protocolSender = protocolSender;
- }
-
- @Override
- public void run() {
- try {
- final NodeBulletinsMessage message = createBulletinsMessage();
- if (message == null) {
- return;
- }
-
- protocolSender.sendBulletins(message);
- if (LOG.isDebugEnabled()) {
- LOG.debug(
- String.format(
- "Sending bulletins to cluster manager at %s",
- dateFormatter.format(new Date())));
- }
-
- } catch (final UnknownServiceAddressException usae) {
- if (LOG.isDebugEnabled()) {
- LOG.debug(usae.getMessage());
- }
- } catch (final Exception ex) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("Failed to send bulletins to cluster manager due to: " + ex, ex);
- }
- }
- }
-
- private boolean isIllegalXmlChar(final char c) {
- return c < 0x20 && c != 0x09 && c != 0x0A && c != 0x0D;
- }
-
- private boolean containsIllegalXmlChars(final Bulletin bulletin) {
- final String message = bulletin.getMessage();
- for (int i = 0; i < message.length(); i++) {
- final char c = message.charAt(i);
- if (isIllegalXmlChar(c)) {
- return true;
- }
- }
-
- return false;
- }
-
- private String stripIllegalXmlChars(final String value) {
- final StringBuilder sb = new StringBuilder(value.length());
- for (int i = 0; i < value.length(); i++) {
- final char c = value.charAt(i);
- sb.append(isIllegalXmlChar(c) ? '?' : c);
- }
-
- return sb.toString();
- }
-
- private NodeBulletinsMessage createBulletinsMessage() {
- final Set<Bulletin> nodeBulletins = nodeBulletinSubscriber.get().getBulletins();
- final Set<Bulletin> escapedNodeBulletins = new HashSet<>(nodeBulletins.size());
-
- // ensure there are some bulletins to report
- if (nodeBulletins.isEmpty()) {
- return null;
- }
-
- for (final Bulletin bulletin : nodeBulletins) {
- final Bulletin escapedBulletin;
- if (containsIllegalXmlChars(bulletin)) {
- final String escapedBulletinMessage = stripIllegalXmlChars(bulletin.getMessage());
-
- if (bulletin.getGroupId() == null) {
- escapedBulletin = BulletinFactory.createBulletin(bulletin.getCategory(), bulletin.getLevel(), escapedBulletinMessage);
- } else {
- escapedBulletin = BulletinFactory.createBulletin(bulletin.getGroupId(), bulletin.getSourceId(), bulletin.getSourceType(),
- bulletin.getSourceName(), bulletin.getCategory(), bulletin.getLevel(), escapedBulletinMessage);
- }
- } else {
- escapedBulletin = bulletin;
- }
-
- escapedNodeBulletins.add(escapedBulletin);
- }
-
- // create the bulletin payload
- final BulletinsPayload payload = new BulletinsPayload();
- payload.setBulletins(escapedNodeBulletins);
-
- // create bulletin message
- final NodeBulletins bulletins = new NodeBulletins(getNodeId(), payload.marshal());
- final NodeBulletinsMessage message = new NodeBulletinsMessage();
- message.setBulletins(bulletins);
-
- return message;
- }
- }
private class HeartbeatSendTask implements Runnable {
@@ -3819,7 +3700,6 @@ public class FlowController implements EventAccess, ControllerServiceProvider, R
hbPayload.setCounters(getCounters());
hbPayload.setSystemDiagnostics(getSystemDiagnostics());
- hbPayload.setProcessGroupStatus(procGroupStatus);
// create heartbeat message
final Heartbeat heartbeat = new Heartbeat(getNodeId(), bean.isPrimary(), bean.isConnected(), hbPayload.marshal());
http://git-wip-us.apache.org/repos/asf/nifi/blob/ad32cb82/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/ConnectionStatusDescriptor.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/ConnectionStatusDescriptor.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/ConnectionStatusDescriptor.java
new file mode 100644
index 0000000..8b9f383
--- /dev/null
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/ConnectionStatusDescriptor.java
@@ -0,0 +1,110 @@
+/*
+ * 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.controller.status.history;
+
+import org.apache.nifi.controller.status.ConnectionStatus;
+import org.apache.nifi.controller.status.history.MetricDescriptor.Formatter;
+
+public enum ConnectionStatusDescriptor {
+ INPUT_BYTES(new StandardMetricDescriptor<ConnectionStatus>(
+ "inputBytes",
+ "Bytes In (5 mins)",
+ "The cumulative size of all FlowFiles that were transferred to this Connection in the past 5 minutes",
+ Formatter.DATA_SIZE,
+ new ValueMapper<ConnectionStatus>() {
+ @Override
+ public Long getValue(final ConnectionStatus status) {
+ return status.getInputBytes();
+ }
+ })),
+
+ INPUT_COUNT(new StandardMetricDescriptor<ConnectionStatus>(
+ "inputCount",
+ "FlowFiles In (5 mins)",
+ "The number of FlowFiles that were transferred to this Connection in the past 5 minutes",
+ Formatter.COUNT,
+ new ValueMapper<ConnectionStatus>() {
+ @Override
+ public Long getValue(final ConnectionStatus status) {
+ return Long.valueOf(status.getInputCount());
+ }
+ })),
+
+ OUTPUT_BYTES(new StandardMetricDescriptor<ConnectionStatus>(
+ "outputBytes",
+ "Bytes Out (5 mins)",
+ "The cumulative size of all FlowFiles that were pulled from this Connection in the past 5 minutes",
+ Formatter.DATA_SIZE,
+ new ValueMapper<ConnectionStatus>() {
+ @Override
+ public Long getValue(final ConnectionStatus status) {
+ return status.getOutputBytes();
+ }
+ })),
+
+ OUTPUT_COUNT(new StandardMetricDescriptor<ConnectionStatus>(
+ "outputCount",
+ "FlowFiles Out (5 mins)",
+ "The number of FlowFiles that were pulled from this Connection in the past 5 minutes",
+ Formatter.COUNT,
+ new ValueMapper<ConnectionStatus>() {
+ @Override
+ public Long getValue(final ConnectionStatus status) {
+ return Long.valueOf(status.getOutputCount());
+ }
+ })),
+
+ QUEUED_BYTES(new StandardMetricDescriptor<ConnectionStatus>(
+ "queuedBytes",
+ "Queued Bytes",
+ "The number of Bytes queued in this Connection",
+ Formatter.DATA_SIZE,
+ new ValueMapper<ConnectionStatus>() {
+ @Override
+ public Long getValue(final ConnectionStatus status) {
+ return status.getQueuedBytes();
+ }
+ })),
+
+ QUEUED_COUNT(new StandardMetricDescriptor<ConnectionStatus>(
+ "queuedCount",
+ "Queued Count",
+ "The number of FlowFiles queued in this Connection",
+ Formatter.COUNT,
+ new ValueMapper<ConnectionStatus>() {
+ @Override
+ public Long getValue(final ConnectionStatus status) {
+ return Long.valueOf(status.getQueuedCount());
+ }
+ }));
+
+
+ private MetricDescriptor<ConnectionStatus> descriptor;
+
+ private ConnectionStatusDescriptor(final MetricDescriptor<ConnectionStatus> descriptor) {
+ this.descriptor = descriptor;
+ }
+
+ public String getField() {
+ return descriptor.getField();
+ }
+
+ public MetricDescriptor<ConnectionStatus> getDescriptor() {
+ return descriptor;
+ }
+}
http://git-wip-us.apache.org/repos/asf/nifi/blob/ad32cb82/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/ProcessGroupStatusDescriptor.java
----------------------------------------------------------------------
diff --git a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/ProcessGroupStatusDescriptor.java b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/ProcessGroupStatusDescriptor.java
new file mode 100644
index 0000000..d5325d0
--- /dev/null
+++ b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/status/history/ProcessGroupStatusDescriptor.java
@@ -0,0 +1,143 @@
+/*
+ * 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.controller.status.history;
+
+import java.util.concurrent.TimeUnit;
+
+import org.apache.nifi.controller.status.ProcessGroupStatus;
+import org.apache.nifi.controller.status.ProcessorStatus;
+import org.apache.nifi.controller.status.history.MetricDescriptor.Formatter;
+
+public enum ProcessGroupStatusDescriptor {
+
+ BYTES_READ(new StandardMetricDescriptor<ProcessGroupStatus>("bytesRead", "Bytes Read (5 mins)",
+ "The total number of bytes read from Content Repository by Processors in this Process Group in the past 5 minutes", Formatter.DATA_SIZE, new ValueMapper<ProcessGroupStatus>() {
+ @Override
+ public Long getValue(final ProcessGroupStatus status) {
+ return status.getBytesRead();
+ }
+ })),
+
+ BYTES_WRITTEN(new StandardMetricDescriptor<ProcessGroupStatus>("bytesWritten", "Bytes Written (5 mins)",
+ "The total number of bytes written to Content Repository by Processors in this Process Group in the past 5 minutes", Formatter.DATA_SIZE, new ValueMapper<ProcessGroupStatus>() {
+ @Override
+ public Long getValue(final ProcessGroupStatus status) {
+ return status.getBytesWritten();
+ }
+ })),
+
+ BYTES_TRANSFERRED(new StandardMetricDescriptor<ProcessGroupStatus>("bytesTransferred", "Bytes Transferred (5 mins)",
+ "The total number of bytes read from or written to Content Repository by Processors in this Process Group in the past 5 minutes",
+ Formatter.DATA_SIZE, new ValueMapper<ProcessGroupStatus>() {
+ @Override
+ public Long getValue(final ProcessGroupStatus status) {
+ return status.getBytesRead() + status.getBytesWritten();
+ }
+ })),
+
+ INPUT_BYTES(new StandardMetricDescriptor<ProcessGroupStatus>("inputBytes", "Bytes In (5 mins)",
+ "The cumulative size of all FlowFiles that have entered this Process Group via its Input Ports in the past 5 minutes",
+ Formatter.DATA_SIZE, new ValueMapper<ProcessGroupStatus>() {
+ @Override
+ public Long getValue(final ProcessGroupStatus status) {
+ return status.getInputContentSize();
+ }
+ })),
+
+ INPUT_COUNT(new StandardMetricDescriptor<ProcessGroupStatus>("inputCount", "FlowFiles In (5 mins)",
+ "The number of FlowFiles that have entered this Process Group via its Input Ports in the past 5 minutes",
+ Formatter.COUNT, new ValueMapper<ProcessGroupStatus>() {
+ @Override
+ public Long getValue(final ProcessGroupStatus status) {
+ return status.getInputCount().longValue();
+ }
+ })),
+
+ OUTPUT_BYTES(new StandardMetricDescriptor<ProcessGroupStatus>("outputBytes", "Bytes Out (5 mins)",
+ "The cumulative size of all FlowFiles that have exited this Process Group via its Output Ports in the past 5 minutes",
+ Formatter.DATA_SIZE, new ValueMapper<ProcessGroupStatus>() {
+ @Override
+ public Long getValue(final ProcessGroupStatus status) {
+ return status.getOutputContentSize();
+ }
+ })),
+
+ OUTPUT_COUNT(new StandardMetricDescriptor<ProcessGroupStatus>("outputCount", "FlowFiles Out (5 mins)",
+ "The number of FlowFiles that have exited this Process Group via its Output Ports in the past 5 minutes",
+ Formatter.COUNT, new ValueMapper<ProcessGroupStatus>() {
+ @Override
+ public Long getValue(final ProcessGroupStatus status) {
+ return status.getOutputCount().longValue();
+ }
+ })),
+
+ QUEUED_BYTES(new StandardMetricDescriptor<ProcessGroupStatus>("queuedBytes", "Queued Bytes",
+ "The cumulative size of all FlowFiles queued in all Connections of this Process Group",
+ Formatter.DATA_SIZE, new ValueMapper<ProcessGroupStatus>() {
+ @Override
+ public Long getValue(final ProcessGroupStatus status) {
+ return status.getQueuedContentSize();
+ }
+ })),
+
+ QUEUED_COUNT(new StandardMetricDescriptor<ProcessGroupStatus>("queuedCount", "Queued Count",
+ "The number of FlowFiles queued in all Connections of this Process Group", Formatter.COUNT, new ValueMapper<ProcessGroupStatus>() {
+ @Override
+ public Long getValue(final ProcessGroupStatus status) {
+ return status.getQueuedCount().longValue();
+ }
+ })),
+
+ TASK_MILLIS(new StandardMetricDescriptor<ProcessGroupStatus>("taskMillis", "Total Task Duration (5 mins)",
+ "The total number of thread-milliseconds that the Processors within this ProcessGroup have used to complete their tasks in the past 5 minutes",
+ Formatter.DURATION, new ValueMapper<ProcessGroupStatus>() {
+ @Override
+ public Long getValue(final ProcessGroupStatus status) {
+ return calculateTaskMillis(status);
+ }
+ }));
+
+ private MetricDescriptor<ProcessGroupStatus> descriptor;
+
+ private ProcessGroupStatusDescriptor(final MetricDescriptor<ProcessGroupStatus> descriptor) {
+ this.descriptor = descriptor;
+ }
+
+ public String getField() {
+ return descriptor.getField();
+ }
+
+ public MetricDescriptor<ProcessGroupStatus> getDescriptor() {
+ return descriptor;
+ }
+
+
+ private static long calculateTaskMillis(final ProcessGroupStatus status) {
+ long nanos = 0L;
+
+ for (final ProcessorStatus procStatus : status.getProcessorStatus()) {
+ nanos += procStatus.getProcessingNanos();
+ }
+
+ for (final ProcessGroupStatus childStatus : status.getProcessGroupStatus()) {
+ nanos += calculateTaskMillis(childStatus);
+ }
+
+ return TimeUnit.MILLISECONDS.convert(nanos, TimeUnit.NANOSECONDS);
+ }
+}