Repository: nifi-minifi Updated Branches: refs/heads/master 9fc758b88 -> e8c4aae4d
MINIFI-457 Allowing the usage of both component name and identifier to interface with flow status queries. This closes #128. Signed-off-by: Aldrin Piri <[email protected]> Project: http://git-wip-us.apache.org/repos/asf/nifi-minifi/repo Commit: http://git-wip-us.apache.org/repos/asf/nifi-minifi/commit/e8c4aae4 Tree: http://git-wip-us.apache.org/repos/asf/nifi-minifi/tree/e8c4aae4 Diff: http://git-wip-us.apache.org/repos/asf/nifi-minifi/diff/e8c4aae4 Branch: refs/heads/master Commit: e8c4aae4d20de5a640ec46ac9be50fc17bb0666e Parents: 9fc758b Author: Aldrin Piri <[email protected]> Authored: Tue May 22 15:46:45 2018 -0400 Committer: Aldrin Piri <[email protected]> Committed: Thu May 24 13:15:51 2018 -0400 ---------------------------------------------------------------------- .../status/common/AbstractStatusBean.java | 42 +++++++ .../status/connection/ConnectionStatusBean.java | 20 +--- .../status/processor/ProcessorStatusBean.java | 18 +-- .../status/util/StatusReportPopulator.java | 6 +- .../src/main/markdown/System_Admin_Guide.md | 8 +- .../minifi/status/StatusConfigReporter.java | 117 ++++++++++++------- .../nifi/minifi/status/StatusRequestParser.java | 2 + .../minifi/status/StatusConfigReporterTest.java | 19 ++- 8 files changed, 155 insertions(+), 77 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/e8c4aae4/minifi-commons/minifi-utils/src/main/java/org/apache/nifi/minifi/commons/status/common/AbstractStatusBean.java ---------------------------------------------------------------------- diff --git a/minifi-commons/minifi-utils/src/main/java/org/apache/nifi/minifi/commons/status/common/AbstractStatusBean.java b/minifi-commons/minifi-utils/src/main/java/org/apache/nifi/minifi/commons/status/common/AbstractStatusBean.java new file mode 100644 index 0000000..aaf2ad0 --- /dev/null +++ b/minifi-commons/minifi-utils/src/main/java/org/apache/nifi/minifi/commons/status/common/AbstractStatusBean.java @@ -0,0 +1,42 @@ +/* + * 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.minifi.commons.status.common; + +import java.io.Serializable; + +public abstract class AbstractStatusBean implements Serializable { + + private String id; + private String name; + + public String getId() { + return id; + } + + public void setId(final String id) { + this.id = id; + } + + public String getName() { + return name; + } + + public void setName(String name) { + this.name = name; + } +} http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/e8c4aae4/minifi-commons/minifi-utils/src/main/java/org/apache/nifi/minifi/commons/status/connection/ConnectionStatusBean.java ---------------------------------------------------------------------- diff --git a/minifi-commons/minifi-utils/src/main/java/org/apache/nifi/minifi/commons/status/connection/ConnectionStatusBean.java b/minifi-commons/minifi-utils/src/main/java/org/apache/nifi/minifi/commons/status/connection/ConnectionStatusBean.java index 378b95e..2efe605 100644 --- a/minifi-commons/minifi-utils/src/main/java/org/apache/nifi/minifi/commons/status/connection/ConnectionStatusBean.java +++ b/minifi-commons/minifi-utils/src/main/java/org/apache/nifi/minifi/commons/status/connection/ConnectionStatusBean.java @@ -17,21 +17,12 @@ package org.apache.nifi.minifi.commons.status.connection; -public class ConnectionStatusBean implements java.io.Serializable { - private String name; - private ConnectionHealth connectionHealth; - private ConnectionStats connectionStats; - - public ConnectionStatusBean() { - } +import org.apache.nifi.minifi.commons.status.common.AbstractStatusBean; - public String getName() { - return name; - } +public class ConnectionStatusBean extends AbstractStatusBean { - public void setName(String name) { - this.name = name; - } + private ConnectionHealth connectionHealth; + private ConnectionStats connectionStats; public ConnectionHealth getConnectionHealth() { return connectionHealth; @@ -73,7 +64,8 @@ public class ConnectionStatusBean implements java.io.Serializable { @Override public String toString() { return "{" + - "name='" + name + '\'' + + "id='" + getId() + '\'' + + "name='" + getName() + '\'' + ", connectionHealth=" + connectionHealth + ", connectionStats=" + connectionStats + '}'; http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/e8c4aae4/minifi-commons/minifi-utils/src/main/java/org/apache/nifi/minifi/commons/status/processor/ProcessorStatusBean.java ---------------------------------------------------------------------- diff --git a/minifi-commons/minifi-utils/src/main/java/org/apache/nifi/minifi/commons/status/processor/ProcessorStatusBean.java b/minifi-commons/minifi-utils/src/main/java/org/apache/nifi/minifi/commons/status/processor/ProcessorStatusBean.java index 99b597a..5823511 100644 --- a/minifi-commons/minifi-utils/src/main/java/org/apache/nifi/minifi/commons/status/processor/ProcessorStatusBean.java +++ b/minifi-commons/minifi-utils/src/main/java/org/apache/nifi/minifi/commons/status/processor/ProcessorStatusBean.java @@ -17,28 +17,17 @@ package org.apache.nifi.minifi.commons.status.processor; +import org.apache.nifi.minifi.commons.status.common.AbstractStatusBean; import org.apache.nifi.minifi.commons.status.common.BulletinStatus; import java.util.List; -public class ProcessorStatusBean implements java.io.Serializable { +public class ProcessorStatusBean extends AbstractStatusBean { - private String name; private ProcessorHealth processorHealth; private ProcessorStats processorStats; private List<BulletinStatus> bulletinList; - public ProcessorStatusBean() { - } - - public String getName() { - return name; - } - - public void setName(String name) { - this.name = name; - } - public ProcessorHealth getProcessorHealth() { return processorHealth; } @@ -89,7 +78,8 @@ public class ProcessorStatusBean implements java.io.Serializable { @Override public String toString() { return "{" + - "name='" + name + '\'' + + "id='" + getId()+ '\'' + + "name='" + getName() + '\'' + ", processorHealth=" + processorHealth + ", processorStats=" + processorStats + ", bulletinList=" + bulletinList + http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/e8c4aae4/minifi-commons/minifi-utils/src/main/java/org/apache/nifi/minifi/commons/status/util/StatusReportPopulator.java ---------------------------------------------------------------------- diff --git a/minifi-commons/minifi-utils/src/main/java/org/apache/nifi/minifi/commons/status/util/StatusReportPopulator.java b/minifi-commons/minifi-utils/src/main/java/org/apache/nifi/minifi/commons/status/util/StatusReportPopulator.java index 5e81ea6..263c7f4 100644 --- a/minifi-commons/minifi-utils/src/main/java/org/apache/nifi/minifi/commons/status/util/StatusReportPopulator.java +++ b/minifi-commons/minifi-utils/src/main/java/org/apache/nifi/minifi/commons/status/util/StatusReportPopulator.java @@ -55,7 +55,8 @@ public class StatusReportPopulator { public static void addProcessorStatus(FlowStatusReport flowStatusReport, boolean addHealth, boolean validationErrors, boolean addStats, boolean addBulletins, boolean populateBulletins) { ProcessorStatusBean expectedProcessorStatus = new ProcessorStatusBean(); - expectedProcessorStatus.setName("UpdateAttributeProcessorId"); + expectedProcessorStatus.setId("UpdateAttributeProcessorId"); + expectedProcessorStatus.setName("UpdateAttributeProcessorName"); if (addHealth) { ProcessorHealth processorHealth = new ProcessorHealth(); @@ -114,7 +115,8 @@ public class StatusReportPopulator { ConnectionStatusBean expectedConnectionStatus = new ConnectionStatusBean(); - expectedConnectionStatus.setName("connectionId"); + expectedConnectionStatus.setId("connectionId"); + expectedConnectionStatus.setName("connectionName"); if (addHealth) { ConnectionHealth connectionHealth = new ConnectionHealth(); connectionHealth.setQueuedCount(10); http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/e8c4aae4/minifi-docs/src/main/markdown/System_Admin_Guide.md ---------------------------------------------------------------------- diff --git a/minifi-docs/src/main/markdown/System_Admin_Guide.md b/minifi-docs/src/main/markdown/System_Admin_Guide.md index bda996b..719fef2 100644 --- a/minifi-docs/src/main/markdown/System_Admin_Guide.md +++ b/minifi-docs/src/main/markdown/System_Admin_Guide.md @@ -131,7 +131,7 @@ In NiFi there is a lot of information, such as stats and bulletins, that is only ## FlowStatus Script Query -From the minifi.sh script, you can manually query to get the current status of your dataflow. The following is an example of a minifi.sh query you might run to view health, stats, and bulletins for the TailFile processor. This query returns information to your command-line. +From the minifi.sh script, you can manually query to get the current status of your dataflow. The following is an example of a minifi.sh query you might run to view health, stats, and bulletins for the TailFile processor. This query returns information to your command-line. ``` minifi.sh flowStatus processor:TailFile:health,stats,bulletins @@ -202,7 +202,7 @@ This section outlines each option to query the MiNiFi instance for the FlowStatu ### Processors -To query the processors use the "processor" flag followed by the processor ID to get (or "all") followed by one of the processor options. The processor options are below. +To query the processors use the "processor" flag followed by the processor ID or name, to get (or "all") followed by one of the processor options. The processor options are below. Option | Description ------ | ----------- @@ -216,7 +216,7 @@ minifi.sh flowStatus processor:TailFile:health,stats,bulletins ``` ### Connections -To query the connections use the "connection" flag followed by the connection ID to get (or "all") followed by one of the connection options. The connection options are below. +To query the connections use the "connection" flag followed by the connection ID or name, to get (or "all") followed by one of the connection options. The connection options are below. Option | Description ------ | ----------- @@ -230,7 +230,7 @@ minifi.sh flowStatus connection:TailToS2S:health,stats ### Remote Process Groups -To query the remote process groups (RPG) use the "remoteProcessGroup" flag followed by the RPG ID to get (or "all") followed by one of the remote process group options. The remote process group options are below. +To query the remote process groups (RPG) use the "remoteProcessGroup" flag followed by the RPG ID or name to get (or "all") followed by one of the remote process group options. The remote process group options are below. Option | Description ------ | ----------- http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/e8c4aae4/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/main/java/org/apache/nifi/minifi/status/StatusConfigReporter.java ---------------------------------------------------------------------- diff --git a/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/main/java/org/apache/nifi/minifi/status/StatusConfigReporter.java b/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/main/java/org/apache/nifi/minifi/status/StatusConfigReporter.java index 75350b2..36251c0 100644 --- a/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/main/java/org/apache/nifi/minifi/status/StatusConfigReporter.java +++ b/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/main/java/org/apache/nifi/minifi/status/StatusConfigReporter.java @@ -35,8 +35,10 @@ import org.apache.nifi.minifi.commons.status.rpg.RemoteProcessGroupStatusBean; import org.apache.nifi.minifi.commons.status.system.SystemDiagnosticsStatus; import org.slf4j.Logger; +import java.util.Arrays; import java.util.Collection; import java.util.HashMap; +import java.util.HashSet; import java.util.LinkedList; import java.util.List; import java.util.Map; @@ -78,15 +80,15 @@ public final class StatusConfigReporter { Map<String, RemoteProcessGroupStatus> remoteProcessGroupStatusMap = null; for (String item : itemsToReport) { - String[] sections = item.split(":"); try { - switch (sections[0].toLowerCase().trim()) { + final RequestItem requestItem = new RequestItem(item); + switch (requestItem.queryType.toLowerCase().trim()) { case "systemdiagnostics": - SystemDiagnosticsStatus systemDiagnosticsStatus = parseSystemDiagnosticsRequest(flowController.getSystemDiagnostics(), sections[1]); + SystemDiagnosticsStatus systemDiagnosticsStatus = parseSystemDiagnosticsRequest(flowController.getSystemDiagnostics(), requestItem.options); flowStatusReport.setSystemDiagnosticsStatus(systemDiagnosticsStatus); break; case "instance": - InstanceStatus instanceStatus = parseInstanceRequest(sections[1], flowController, rootGroupStatus); + InstanceStatus instanceStatus = parseInstanceRequest(requestItem.options, flowController, rootGroupStatus); flowStatusReport.setInstanceStatus(instanceStatus); break; case "remoteprocessgroup": @@ -94,35 +96,35 @@ public final class StatusConfigReporter { List<RemoteProcessGroupStatusBean> remoteProcessGroupStatusList = new LinkedList<>(); flowStatusReport.setRemoteProcessGroupStatusList(remoteProcessGroupStatusList); } - handleRemoteProcessGroupRequest(sections, rootGroupStatus, flowController, flowStatusReport.getRemoteProcessGroupStatusList(), remoteProcessGroupStatusMap, logger); + handleRemoteProcessGroupRequest(requestItem, rootGroupStatus, flowController, flowStatusReport.getRemoteProcessGroupStatusList(), remoteProcessGroupStatusMap, logger); break; case "processor": if (flowStatusReport.getProcessorStatusList() == null) { List<ProcessorStatusBean> processorStatusList = new LinkedList<>(); flowStatusReport.setProcessorStatusList(processorStatusList); } - handleProcessorRequest(sections, rootGroupStatus, flowController, flowStatusReport.getProcessorStatusList(), processorStatusMap, logger); + handleProcessorRequest(requestItem, rootGroupStatus, flowController, flowStatusReport.getProcessorStatusList(), processorStatusMap, logger); break; case "connection": if (flowStatusReport.getConnectionStatusList() == null) { List<ConnectionStatusBean> connectionStatusList = new LinkedList<>(); flowStatusReport.setConnectionStatusList(connectionStatusList); } - handleConnectionRequest(sections, rootGroupStatus, flowStatusReport.getConnectionStatusList(), connectionStatusMap, logger); + handleConnectionRequest(requestItem, rootGroupStatus, flowStatusReport.getConnectionStatusList(), connectionStatusMap, logger); break; case "provenancereporting": if (flowStatusReport.getRemoteProcessGroupStatusList() == null) { List<ReportingTaskStatus> reportingTaskStatusList = new LinkedList<>(); flowStatusReport.setReportingTaskStatusList(reportingTaskStatusList); } - handleReportingTaskRequest(sections, flowController, flowStatusReport.getReportingTaskStatusList(), logger); + handleReportingTaskRequest(requestItem, flowController, flowStatusReport.getReportingTaskStatusList(), logger); break; case "controllerservices": if (flowStatusReport.getControllerServiceStatusList() == null) { List<ControllerServiceStatus> controllerServiceStatusList = new LinkedList<>(); flowStatusReport.setControllerServiceStatusList(controllerServiceStatusList); } - handleControllerServices(sections, flowController, flowStatusReport.getControllerServiceStatusList(), logger); + handleControllerServices(requestItem, flowController, flowStatusReport.getControllerServiceStatusList(), logger); break; } } catch (Exception e) { @@ -133,98 +135,98 @@ public final class StatusConfigReporter { return flowStatusReport; } - private static void handleControllerServices(String[] sections, FlowController flowController, List<ControllerServiceStatus> controllerServiceStatusList, Logger logger) { + private static void handleControllerServices(RequestItem requestItem, FlowController flowController, List<ControllerServiceStatus> controllerServiceStatusList, Logger logger) { Collection<ControllerServiceNode> controllerServiceNodeSet = flowController.getAllControllerServices(); if (!controllerServiceNodeSet.isEmpty()) { for (ControllerServiceNode controllerServiceNode : controllerServiceNodeSet) { - controllerServiceStatusList.add(parseControllerServiceStatusRequest(controllerServiceNode, sections[1], flowController, logger)); + controllerServiceStatusList.add(parseControllerServiceStatusRequest(controllerServiceNode, requestItem.options, flowController, logger)); } } } - private static void handleProcessorRequest(String[] sections, ProcessGroupStatus rootGroupStatus, FlowController flowController, List<ProcessorStatusBean> processorStatusBeanList, + private static void handleProcessorRequest(final RequestItem requestItem, ProcessGroupStatus rootGroupStatus, FlowController flowController, List<ProcessorStatusBean> processorStatusBeanList, Map<String, ProcessorStatus> processorStatusMap, Logger logger) throws StatusRequestException { if (processorStatusMap == null) { processorStatusMap = transformStatusCollection(rootGroupStatus.getProcessorStatus()); } String rootGroupId = flowController.getRootGroupId(); - if (sections[1].equalsIgnoreCase("all")) { + if (requestItem.identifier.equalsIgnoreCase("all")) { if (!processorStatusMap.isEmpty()) { - for (ProcessorStatus processorStatus : processorStatusMap.values()) { + for (ProcessorStatus processorStatus : new HashSet<>(processorStatusMap.values())) { Collection<ValidationResult> validationResults = flowController.getGroup(rootGroupId).getProcessor(processorStatus.getId()).getValidationErrors(); - processorStatusBeanList.add(parseProcessorStatusRequest(processorStatus, sections[2], flowController, validationResults)); + processorStatusBeanList.add(parseProcessorStatusRequest(processorStatus, requestItem.options, flowController, validationResults)); } } } else { - if (processorStatusMap.containsKey(sections[1])) { - ProcessorStatus processorStatus = processorStatusMap.get(sections[1]); + if (processorStatusMap.containsKey(requestItem.identifier)) { + ProcessorStatus processorStatus = processorStatusMap.get(requestItem.identifier); Collection<ValidationResult> validationResults = flowController.getGroup(rootGroupId).getProcessor(processorStatus.getId()).getValidationErrors(); - processorStatusBeanList.add(parseProcessorStatusRequest(processorStatus, sections[2], flowController, validationResults)); + processorStatusBeanList.add(parseProcessorStatusRequest(processorStatus, requestItem.options, flowController, validationResults)); } else { - logger.warn("Status for processor with key " + sections[1] + " was requested but one does not exist"); - throw new StatusRequestException("No processor with key " + sections[1] + " to report status on"); + logger.warn("Status for processor with key " + requestItem.identifier + " was requested but one does not exist"); + throw new StatusRequestException("No processor with key " + requestItem.identifier + " to report status on"); } } } - private static void handleConnectionRequest(String[] sections, ProcessGroupStatus rootGroupStatus, List<ConnectionStatusBean> connectionStatusList, + private static void handleConnectionRequest(RequestItem requestItem, ProcessGroupStatus rootGroupStatus, List<ConnectionStatusBean> connectionStatusList, Map<String, ConnectionStatus> connectionStatusMap, Logger logger) throws StatusRequestException { if (connectionStatusMap == null) { connectionStatusMap = transformStatusCollection(rootGroupStatus.getConnectionStatus()); } - if (sections[1].equalsIgnoreCase("all")) { + if (requestItem.identifier.equalsIgnoreCase("all")) { if (!connectionStatusMap.isEmpty()) { - for (ConnectionStatus connectionStatus : connectionStatusMap.values()) { - connectionStatusList.add(parseConnectionStatusRequest(connectionStatus, sections[2], logger)); + for (ConnectionStatus connectionStatus : new HashSet<>(connectionStatusMap.values())) { + connectionStatusList.add(parseConnectionStatusRequest(connectionStatus, requestItem.options, logger)); } } } else { - if (connectionStatusMap.containsKey(sections[1])) { - connectionStatusList.add(parseConnectionStatusRequest(connectionStatusMap.get(sections[1]), sections[2], logger)); + if (connectionStatusMap.containsKey(requestItem.identifier)) { + connectionStatusList.add(parseConnectionStatusRequest(connectionStatusMap.get(requestItem.identifier), requestItem.options, logger)); } else { - logger.warn("Status for connection with key " + sections[1] + " was requested but one does not exist"); - throw new StatusRequestException("No connection with key " + sections[1] + " to report status on"); + logger.warn("Status for connection with key " + requestItem.identifier + " was requested but one does not exist"); + throw new StatusRequestException("No connection with key " + requestItem.identifier + " to report status on"); } } } - private static void handleRemoteProcessGroupRequest(String[] sections, ProcessGroupStatus rootGroupStatus, FlowController flowController, + private static void handleRemoteProcessGroupRequest(RequestItem requestItem, ProcessGroupStatus rootGroupStatus, FlowController flowController, List<RemoteProcessGroupStatusBean> remoteProcessGroupStatusList, Map<String, RemoteProcessGroupStatus> remoteProcessGroupStatusMap, Logger logger) throws StatusRequestException { if (remoteProcessGroupStatusMap == null) { remoteProcessGroupStatusMap = transformStatusCollection(rootGroupStatus.getRemoteProcessGroupStatus()); } - if (sections[1].equalsIgnoreCase("all")) { + if (requestItem.identifier.equalsIgnoreCase("all")) { if (!remoteProcessGroupStatusMap.isEmpty()) { - for (RemoteProcessGroupStatus remoteProcessGroupStatus : remoteProcessGroupStatusMap.values()) { - remoteProcessGroupStatusList.add(parseRemoteProcessGroupStatusRequest(remoteProcessGroupStatus, sections[2], flowController)); + for (RemoteProcessGroupStatus remoteProcessGroupStatus : new HashSet<>(remoteProcessGroupStatusMap.values())) { + remoteProcessGroupStatusList.add(parseRemoteProcessGroupStatusRequest(remoteProcessGroupStatus, requestItem.options, flowController)); } } } else { - if (remoteProcessGroupStatusMap.containsKey(sections[1])) { - RemoteProcessGroupStatus remoteProcessGroupStatus = remoteProcessGroupStatusMap.get(sections[1]); - remoteProcessGroupStatusList.add(parseRemoteProcessGroupStatusRequest(remoteProcessGroupStatus, sections[2], flowController)); + if (remoteProcessGroupStatusMap.containsKey(requestItem.identifier)) { + RemoteProcessGroupStatus remoteProcessGroupStatus = remoteProcessGroupStatusMap.get(requestItem.identifier); + remoteProcessGroupStatusList.add(parseRemoteProcessGroupStatusRequest(remoteProcessGroupStatus, requestItem.options, flowController)); } else { - logger.warn("Status for Remote Process Group with key " + sections[1] + " was requested but one does not exist"); - throw new StatusRequestException("No Remote Process Group with key " + sections[1] + " to report status on"); + logger.warn("Status for Remote Process Group with key " + requestItem.identifier + " was requested but one does not exist"); + throw new StatusRequestException("No Remote Process Group with key " + requestItem.identifier + " to report status on"); } } } - private static void handleReportingTaskRequest(String[] sections, FlowController flowController, List<ReportingTaskStatus> reportingTaskStatusList, Logger logger) { + private static void handleReportingTaskRequest(RequestItem requestItem, FlowController flowController, List<ReportingTaskStatus> reportingTaskStatusList, Logger logger) { Set<ReportingTaskNode> reportingTaskNodes = flowController.getAllReportingTasks(); if (!reportingTaskNodes.isEmpty()) { for (ReportingTaskNode reportingTaskNode : reportingTaskNodes) { - reportingTaskStatusList.add(parseReportingTaskStatusRequest(reportingTaskNode.getIdentifier(), reportingTaskNode, sections[1], flowController, logger)); + reportingTaskStatusList.add(parseReportingTaskStatusRequest(reportingTaskNode.getIdentifier(), reportingTaskNode, requestItem.options, flowController, logger)); } } } @@ -234,16 +236,49 @@ public final class StatusConfigReporter { for (E status : statusCollection) { if (status instanceof ProcessorStatus) { statusMap.put(((ProcessorStatus) status).getId(), status); + if (((ProcessorStatus) status).getName() != null) { + statusMap.put(((ProcessorStatus) status).getName(), status); + } } else if (status instanceof ConnectionStatus) { statusMap.put(((ConnectionStatus) status).getId(), status); + if (((ConnectionStatus) status).getName() != null) { + statusMap.put(((ConnectionStatus) status).getName(), status); + } } else if (status instanceof RemoteProcessGroupStatus) { statusMap.put(((RemoteProcessGroupStatus) status).getId(), status); - } else { - // TODO + if (((RemoteProcessGroupStatus) status).getName() != null) { + statusMap.put(((RemoteProcessGroupStatus) status).getName(), status); + } } } return statusMap; } + private static class RequestItem { + + private final int EXPECTED_REQUEST_COMPONENTS = 2; + + private String queryType; + private String identifier; + private String options; + + public RequestItem(final String requestString) { + final List<String> reqComponents = Arrays.asList(requestString.split(":")); + final int numComponents = reqComponents.size(); + if (numComponents < EXPECTED_REQUEST_COMPONENTS) { + throw new IllegalArgumentException(String.format("Cannot perform a FlowStatusQuery request for '%s'. Expected at least %d components but got %d.", + requestString, EXPECTED_REQUEST_COMPONENTS, numComponents)); + } + this.queryType = reqComponents.get(0).toLowerCase(); + if (numComponents == 2) { + this.options = reqComponents.get(1); + } else { + this.identifier = reqComponents.get(1); + this.options = reqComponents.get(2); + } + // normalize options + this.options = this.options.toLowerCase(); + } + } } http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/e8c4aae4/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/main/java/org/apache/nifi/minifi/status/StatusRequestParser.java ---------------------------------------------------------------------- diff --git a/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/main/java/org/apache/nifi/minifi/status/StatusRequestParser.java b/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/main/java/org/apache/nifi/minifi/status/StatusRequestParser.java index 9f09d06..50a8c58 100644 --- a/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/main/java/org/apache/nifi/minifi/status/StatusRequestParser.java +++ b/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/main/java/org/apache/nifi/minifi/status/StatusRequestParser.java @@ -73,6 +73,7 @@ public final class StatusRequestParser { static ProcessorStatusBean parseProcessorStatusRequest(ProcessorStatus inputProcessorStatus, String statusTypes, FlowController flowController, Collection<ValidationResult> validationResults) { ProcessorStatusBean processorStatusBean = new ProcessorStatusBean(); + processorStatusBean.setId(inputProcessorStatus.getId()); processorStatusBean.setName(inputProcessorStatus.getName()); String[] statusSplits = statusTypes.split(","); @@ -175,6 +176,7 @@ public final class StatusRequestParser { static ConnectionStatusBean parseConnectionStatusRequest(ConnectionStatus inputConnectionStatus, String statusTypes, Logger logger) { ConnectionStatusBean connectionStatusBean = new ConnectionStatusBean(); + connectionStatusBean.setId(inputConnectionStatus.getId()); connectionStatusBean.setName(inputConnectionStatus.getName()); String[] statusSplits = statusTypes.split(","); http://git-wip-us.apache.org/repos/asf/nifi-minifi/blob/e8c4aae4/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/test/java/org/apache/nifi/minifi/status/StatusConfigReporterTest.java ---------------------------------------------------------------------- diff --git a/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/test/java/org/apache/nifi/minifi/status/StatusConfigReporterTest.java b/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/test/java/org/apache/nifi/minifi/status/StatusConfigReporterTest.java index 171400a..004742c 100644 --- a/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/test/java/org/apache/nifi/minifi/status/StatusConfigReporterTest.java +++ b/minifi-nar-bundles/minifi-framework-bundle/minifi-framework/minifi-framework-core/src/test/java/org/apache/nifi/minifi/status/StatusConfigReporterTest.java @@ -105,6 +105,21 @@ public class StatusConfigReporterTest { } @Test + public void individualProcessorStatusHealth() throws Exception { + populateProcessor(false, false); + + String statusRequest = "processor:UpdateAttributeProcessorId:health"; + FlowStatusReport actual = StatusConfigReporter.getStatus(mockFlowController, statusRequest, LoggerFactory.getLogger(StatusConfigReporterTest.class)); + + FlowStatusReport expected = new FlowStatusReport(); + expected.setErrorsGeneratingReport(Collections.EMPTY_LIST); + + addProcessorStatus(expected, true, false, false, false, false); + + assertEquals(expected, actual); + } + + @Test public void processorStatusWithValidationErrors() throws Exception { populateProcessor(true, false); @@ -636,7 +651,7 @@ public class StatusConfigReporterTest { ConnectionStatus connectionStatus = new ConnectionStatus(); connectionStatus.setQueuedBytes(100); connectionStatus.setId("connectionId"); - connectionStatus.setName("connectionId"); + connectionStatus.setName("connectionName"); connectionStatus.setQueuedCount(10); connectionStatus.setInputCount(1); connectionStatus.setInputBytes(2); @@ -657,7 +672,7 @@ public class StatusConfigReporterTest { ProcessorStatus processorStatus = new ProcessorStatus(); processorStatus.setType("org.apache.nifi.processors.attributes.UpdateAttribute"); processorStatus.setId("UpdateAttributeProcessorId"); - processorStatus.setName("UpdateAttributeProcessorId"); + processorStatus.setName("UpdateAttributeProcessorName"); processorStatus.setRunStatus(RunStatus.Stopped); processorStatus.setActiveThreadCount(1); processorStatus.setFlowFilesReceived(2);
