You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by sz...@apache.org on 2022/08/19 10:20:15 UTC
[nifi-minifi-cpp] 01/03: MINIFICPP-1853 Support multiple metrics of the same processor
This is an automated email from the ASF dual-hosted git repository.
szaszm pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit 5bdb8912cde35223c15bddc457512ac1ebd831a1
Author: Gabor Gyimesi <ga...@gmail.com>
AuthorDate: Thu Aug 18 18:27:18 2022 +0200
MINIFICPP-1853 Support multiple metrics of the same processor
Closes #1360
Signed-off-by: Marton Szasz <sz...@apache.org>
---
C2.md | 321 ++++++++++-----------
.../test/integration/features/prometheus.feature | 13 +
.../http-curl/tests/C2DescribeMetricsTest.cpp | 161 +++++++++++
extensions/http-curl/tests/C2MetricsTest.cpp | 7 +-
.../http-curl/tests/C2SameProcessorMetrics.cpp | 115 ++++++++
extensions/http-curl/tests/CMakeLists.txt | 2 +
.../prometheus/PrometheusMetricsPublisher.cpp | 6 +-
.../standard-processors/processors/GetFile.h | 10 +-
extensions/standard-processors/processors/GetTCP.h | 6 +-
libminifi/include/c2/C2Client.h | 8 +-
libminifi/include/core/state/nodes/MetricsBase.h | 30 +-
.../include/core/state/nodes/ResponseNodeLoader.h | 9 +-
libminifi/src/c2/C2Client.cpp | 81 +++---
.../src/core/state/nodes/ResponseNodeLoader.cpp | 44 +--
.../test/resources/TestSameProcessorMetrics.yml | 123 ++++++++
15 files changed, 699 insertions(+), 237 deletions(-)
diff --git a/C2.md b/C2.md
index 2f2af37d5..5394531c5 100644
--- a/C2.md
+++ b/C2.md
@@ -21,14 +21,16 @@ options defined are located in minifi.properties.
## Table of Contents
-- [Description](#description)
-- [Configuration](#configuration)
- - [Base Options](#base-options)
- - [Metrics](#metrics)
- - [Protocols](#protocols)
- - [Triggers](#triggers)
- - [UpdatePolicies](#updatepolicies)
- - [Documentation](#documentation)
+- [Apache NiFi - MiNiFi - C++ C2 Readme.](#apache-nifi---minifi---c-c2-readme)
+ - [Table of Contents](#table-of-contents)
+ - [Description](#description)
+ - [Configuration](#configuration)
+ - [Base Options](#base-options)
+ - [Metrics](#metrics)
+ - [Protocols](#protocols)
+ - [UpdatePolicies](#updatepolicies)
+ - [Triggers](#triggers)
+ - [C2 File triggers](#c2-file-triggers)
## Description
@@ -55,46 +57,46 @@ With this change, heartbeat with agent manifest included is sent only for the fi
light weight heartbeat. If for some reason the C2 server does not receive the first full heartbeat, the manifest can
be requested via C2 DESCRIBE manifest command.
- #in minifi.properties
- # Disable/Enable C2
- nifi.c2.enable=true
+ #in minifi.properties
- # specify which sections the heartbeat message should contain
- # DeviceInfoNode: basic info about the system (OS, number of cores etc)
- # AgentInformation: info about the MiNiFi agent, may include the manifest
- # FlowInformation: information about the current flow, including queue sizes
- # ConfigurationChecksums: hashes of the configuration files; can be used to detect unexpected modifications
- # the default is
- nifi.c2.root.classes=DeviceInfoNode,AgentInformation,FlowInformation
+ # Disable/Enable C2
+ nifi.c2.enable=true
- # specify C2 protocol -- default is RESTSender if not specified
- nifi.c2.agent.protocol.class=RESTSender
- # nifi.c2.agent.protocol.class=CoapProtocol
+ # specify which sections the heartbeat message should contain
+ # DeviceInfoNode: basic info about the system (OS, number of cores etc)
+ # AgentInformation: info about the MiNiFi agent, may include the manifest
+ # FlowInformation: information about the current flow, including queue sizes
+ # ConfigurationChecksums: hashes of the configuration files; can be used to detect unexpected modifications
+ # the default is
+ nifi.c2.root.classes=DeviceInfoNode,AgentInformation,FlowInformation
- # control c2 heartbeat interval in milliseconds
- nifi.c2.agent.heartbeat.period=250
+ # specify C2 protocol -- default is RESTSender if not specified
+ nifi.c2.agent.protocol.class=RESTSender
+ # nifi.c2.agent.protocol.class=CoapProtocol
- # enable reporter classes
- nifi.c2.agent.heartbeat.reporter.class=RESTReciver
+ # control c2 heartbeat interval in milliseconds
+ nifi.c2.agent.heartbeat.period=250
- # specify the rest URIs if using RESTSender
- nifi.c2.rest.url=http://<your-c2-server>/<c2-api-path>/c2-protocol/heartbeat
- nifi.c2.rest.url.ack=http://<your-c2-server>/<c2-api-path>/c2-protocol/acknowledge
- nifi.c2.flow.base.url=http://<your-c2-server>/<c2-api-path>/c2-protocol/
+ # enable reporter classes
+ nifi.c2.agent.heartbeat.reporter.class=RESTReciver
- # c2 agent identifier -- must be defined to run agent
- nifi.c2.agent.identifier=<your identifier>
+ # specify the rest URIs if using RESTSender
+ nifi.c2.rest.url=http://<your-c2-server>/<c2-api-path>/c2-protocol/heartbeat
+ nifi.c2.rest.url.ack=http://<your-c2-server>/<c2-api-path>/c2-protocol/acknowledge
+ nifi.c2.flow.base.url=http://<your-c2-server>/<c2-api-path>/c2-protocol/
- # c2 agent class -- must be defined to run agent
- nifi.c2.agent.class=<your agent class>
+ # c2 agent identifier -- must be defined to run agent
+ nifi.c2.agent.identifier=<your identifier>
- # configure SSL Context service for REST Protocol
- #nifi.c2.rest.ssl.context.service
+ # c2 agent class -- must be defined to run agent
+ nifi.c2.agent.class=<your agent class>
- # specify encoding strategy for c2 requests (gzip, none)
- #nifi.c2.rest.request.encoding=none
+ # configure SSL Context service for REST Protocol
+ #nifi.c2.rest.ssl.context.service
+ # specify encoding strategy for c2 requests (gzip, none)
+ #nifi.c2.rest.request.encoding=none
### Metrics
@@ -103,18 +105,17 @@ operation. Since responses are formed in an AST, metrics can be formed as a sub
are defined apriori and may reference a metrics class specific to a processor. The following example describes
a configuration of an agent
- # in minifi.properties
+ # in minifi.properties
-
- nifi.c2.root.class.definitions=metrics
- nifi.c2.root.class.definitions.metrics.name=metrics
- nifi.c2.root.class.definitions.metrics.metrics=runtimemetrics,loadmetrics,processorMetrics
- nifi.c2.root.class.definitions.metrics.metrics.runtimemetrics.name=RuntimeMetrics
- nifi.c2.root.class.definitions.metrics.metrics.runtimemetrics.classes=DeviceInfoNode,FlowInformation
- nifi.c2.root.class.definitions.metrics.metrics.loadmetrics.name=LoadMetrics
- nifi.c2.root.class.definitions.metrics.metrics.loadmetrics.classes=QueueMetrics,RepositoryMetrics
- nifi.c2.root.class.definitions.metrics.metrics.processorMetrics.name=ProcessorMetric
- nifi.c2.root.class.definitions.metrics.metrics.processorMetrics.classes=GetFileMetrics
+ nifi.c2.root.class.definitions=metrics
+ nifi.c2.root.class.definitions.metrics.name=metrics
+ nifi.c2.root.class.definitions.metrics.metrics=runtimemetrics,loadmetrics,processorMetrics
+ nifi.c2.root.class.definitions.metrics.metrics.runtimemetrics.name=RuntimeMetrics
+ nifi.c2.root.class.definitions.metrics.metrics.runtimemetrics.classes=DeviceInfoNode,FlowInformation
+ nifi.c2.root.class.definitions.metrics.metrics.loadmetrics.name=LoadMetrics
+ nifi.c2.root.class.definitions.metrics.metrics.loadmetrics.classes=QueueMetrics,RepositoryMetrics
+ nifi.c2.root.class.definitions.metrics.metrics.processorMetrics.name=ProcessorMetric
+ nifi.c2.root.class.definitions.metrics.metrics.processorMetrics.classes=GetFileMetrics
This example shows a metrics sub tree defined by the option 'nifi.c2.root.class.definitions'.
@@ -124,85 +125,87 @@ The options below metrics define the sub-trees within metrics: typedmetrics and
The classes sub option will define the metrics classes that are placed within this sub-tree. For the RESTProtocol, the above
configuration produces the following JSON:
- "metrics": {
- "RuntimeMetrics": {
- "deviceInfo": {
- "systemInfo": {
- "cpuUtilization": -1,
- "machinearch": "x86_64",
- "memoryUsage": 13103550464,
- "operatingSystem": "Linux",
- "physicalMem": 67024097280,
- "vCores": 12
- },
- "networkInfo": {
- "hostname": "ggyimesi-5540-ubuntu",
- "ipAddress": "192.168.50.181"
- },
- "identifier": "13396751919892753964"
- },
- "flowInfo": {
- "versionedFlowSnapshotURI": {
- "bucketId": "default",
- "flowId": "8db40550-db5d-11ec-95d7-0433c2c9832b"
- },
- "queues": {
- "2438e3c8-015a-1000-79ca-83af40ec1997": {
- "dataSize": 0,
- "dataSizeMax": 1048576,
- "name": "GetFile/success/LogAttribute",
- "size": 0,
- "sizeMax": 0,
- "uuid": "2438e3c8-015a-1000-79ca-83af40ec1997"
- }
- },
- "components": {
- "FlowController": {
- "running": true,
- "uuid": "2438e3c8-015a-1000-79ca-83af40ec1990"
- },
- "GetFile": {
- "running": false,
- "uuid": "2438e3c8-015a-1000-79ca-83af40ec1991"
- },
- "LogAttribute": {
- "running": true,
- "uuid": "2438e3c8-015a-1000-79ca-83af40ec1992"
- }
- },
- "flowId": "8db40550-db5d-11ec-95d7-0433c2c9832b"
- }
- },
- "LoadMetrics": {
- "QueueMetrics": {
- "GetFile/success/LogAttribute": {
- "datasize": "0",
- "datasizemax": "1048576",
- "queued": "0",
- "queuedmax": "0"
- }
- },
- "RepositoryMetrics": {
- "ff": {
- "full": false,
- "running": false,
- "size": "0"
- },
- "repo_name": {
- "full": false,
- "running": true,
- "size": "0"
- }
- }
- },
- "ProcessorMetrics": {
- "GetFileMetrics": {
- "AcceptedFiles": 0,
- "InputBytes": 0,
- "OnTriggerInvocations": 0
- }
+ "metrics": {
+ "RuntimeMetrics": {
+ "deviceInfo": {
+ "systemInfo": {
+ "cpuUtilization": -1,
+ "machinearch": "x86_64",
+ "memoryUsage": 13103550464,
+ "operatingSystem": "Linux",
+ "physicalMem": 67024097280,
+ "vCores": 12
+ },
+ "networkInfo": {
+ "hostname": "ggyimesi-5540-ubuntu",
+ "ipAddress": "192.168.50.181"
+ },
+ "identifier": "13396751919892753964"
+ },
+ "flowInfo": {
+ "versionedFlowSnapshotURI": {
+ "bucketId": "default",
+ "flowId": "8db40550-db5d-11ec-95d7-0433c2c9832b"
+ },
+ "queues": {
+ "2438e3c8-015a-1000-79ca-83af40ec1997": {
+ "dataSize": 0,
+ "dataSizeMax": 1048576,
+ "name": "GetFile/success/LogAttribute",
+ "size": 0,
+ "sizeMax": 0,
+ "uuid": "2438e3c8-015a-1000-79ca-83af40ec1997"
+ }
+ },
+ "components": {
+ "FlowController": {
+ "running": true,
+ "uuid": "2438e3c8-015a-1000-79ca-83af40ec1990"
+ },
+ "GetFile": {
+ "running": false,
+ "uuid": "2438e3c8-015a-1000-79ca-83af40ec1991"
+ },
+ "LogAttribute": {
+ "running": true,
+ "uuid": "2438e3c8-015a-1000-79ca-83af40ec1992"
+ }
+ },
+ "flowId": "8db40550-db5d-11ec-95d7-0433c2c9832b"
+ }
+ },
+ "LoadMetrics": {
+ "QueueMetrics": {
+ "GetFile/success/LogAttribute": {
+ "datasize": "0",
+ "datasizemax": "1048576",
+ "queued": "0",
+ "queuedmax": "0"
+ }
+ },
+ "RepositoryMetrics": {
+ "ff": {
+ "full": false,
+ "running": false,
+ "size": "0"
+ },
+ "repo_name": {
+ "full": false,
+ "running": true,
+ "size": "0"
+ }
+ }
+ },
+ "ProcessorMetrics": {
+ "GetFileMetrics": {
+ "2438e3c8-015a-1000-79ca-83af40ec1991": {
+ "AcceptedFiles": 0,
+ "InputBytes": 0,
+ "OnTriggerInvocations": 0
+ }
+ }
+ }
}
- }
### Protocols
@@ -213,50 +216,46 @@ Once configured, COAP uses a controller service within the flow OR minifi proper
Note that with CoAP, the payload will be significantly smaller, paring down metrics that are sent in each heartbeat. This will be useful for
constrained environments.
- nifi.c2.agent.coap.host=hostname
- nifi.c2.agent.coap.port=<port number>
-
+ nifi.c2.agent.coap.host=hostname
+ nifi.c2.agent.coap.port=<port number>
If you wish to use the Controller service you made add a controller service named CoapConnectorService with the properties in the example config
below. Note that Max Queue Size is the only non-required property:
- Controller Services:
- - id: 94491a38-015a-1000-0000-000000000001
- name: coapservice
- class: CoapConnectorService
- Properties:
- Remote Server: server
- Remote Port: port
- Max Queue Size: 1000
-
+ Controller Services:
+ - id: 94491a38-015a-1000-0000-000000000001
+ name: coapservice
+ class: CoapConnectorService
+ Properties:
+ Remote Server: server
+ Remote Port: port
+ Max Queue Size: 1000
### UpdatePolicies
Updates to MiNiFi C++ properties can be controlled through an UpdatePolicyControllerService named
C2UpdatePolicy. The service supports several configuration options. They are defined in the following example:
-
- Controller Services:
- - id: 94491a38-015a-1000-0000-000000000001
- name: C2UpdatePolicy
- class: UpdatePolicyControllerService
- Properties:
- # true enables all properties to be updated.
- Allow All Properties: true
- # allowed properties are those which can be updated
- Allowed Properties:
- - value: Property_1
- - value: Property_2
- Disallowed Properties:
- - value: Property_3
- - value: Property_4
-
+ Controller Services:
+ - id: 94491a38-015a-1000-0000-000000000001
+ name: C2UpdatePolicy
+ class: UpdatePolicyControllerService
+ Properties:
+ # true enables all properties to be updated.
+ Allow All Properties: true
+ # allowed properties are those which can be updated
+ Allowed Properties:
+ - value: Property_1
+ - value: Property_2
+ Disallowed Properties:
+ - value: Property_3
+ - value: Property_4
### Triggers
- C2 Triggers can be activated to perform some C2 activity via a local event. Currently only FileUpdateTrigger exists, which monitors
- for C2 File triggers to update the flow configuration. Classes can be defined as a comma separated list of classes to load via the option
- nifi.c2.agent.trigger.classes
+C2 Triggers can be activated to perform some C2 activity via a local event. Currently only FileUpdateTrigger exists, which monitors
+for C2 File triggers to update the flow configuration. Classes can be defined as a comma separated list of classes to load via the option
+nifi.c2.agent.trigger.classes
#### C2 File triggers
@@ -267,8 +266,8 @@ changes while the agent is running, it will be copied into place of the file def
will then be restarted with the new flow configuration. If a failure occurs in reading that file or it is an invalid YAML file, the
update process will be halted.
- in minifi.properties to activate the file update trigger specify
+in minifi.properties to activate the file update trigger specify
- # specifying a trigger
- nifi.c2.agent.trigger.classes=FileUpdateTrigger
- nifi.c2.file.watch=<full path of file to monitor>
+ # specifying a trigger
+ nifi.c2.agent.trigger.classes=FileUpdateTrigger
+ nifi.c2.file.watch=<full path of file to monitor>
diff --git a/docker/test/integration/features/prometheus.feature b/docker/test/integration/features/prometheus.feature
index 7f1f1aa29..e22aa5faf 100644
--- a/docker/test/integration/features/prometheus.feature
+++ b/docker/test/integration/features/prometheus.feature
@@ -15,3 +15,16 @@ Feature: MiNiFi can publish metrics to Prometheus server
And "GetFileMetrics" processor metric is published to the Prometheus server in less than 60 seconds for "GetFile1" processor
And "FlowInformation" is published to the Prometheus server in less than 60 seconds
And "DeviceInfoNode" is published to the Prometheus server in less than 60 seconds
+
+ Scenario: Multiple GetFile metrics are reported by Prometheus
+ Given a GetFile processor with the name "GetFile1" and the "Input Directory" property set to "/tmp/input"
+ And a GetFile processor with the name "GetFile2" and the "Input Directory" property set to "/tmp/input"
+ And "GetFile2" processor is a start node
+ And a file with the content "test" is present in "/tmp/input"
+ And a PutFile processor with the "Directory" property set to "/tmp/output"
+ And the "success" relationship of the GetFile1 processor is connected to the PutFile
+ And the "success" relationship of the GetFile2 processor is connected to the PutFile
+ And a Prometheus server is set up
+ When all instances start up
+ Then "GetFileMetrics" processor metric is published to the Prometheus server in less than 60 seconds for "GetFile1" processor
+ And "GetFileMetrics" processor metric is published to the Prometheus server in less than 60 seconds for "GetFile2" processor
diff --git a/extensions/http-curl/tests/C2DescribeMetricsTest.cpp b/extensions/http-curl/tests/C2DescribeMetricsTest.cpp
new file mode 100644
index 000000000..acf5d80cb
--- /dev/null
+++ b/extensions/http-curl/tests/C2DescribeMetricsTest.cpp
@@ -0,0 +1,161 @@
+/**
+ *
+ * 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.
+ */
+
+#undef NDEBUG
+#include <memory>
+
+#include "TestBase.h"
+#include "HTTPIntegrationBase.h"
+#include "HTTPHandlers.h"
+#include "processors/TailFile.h"
+#include "state/ProcessorController.h"
+#include "utils/file/FileUtils.h"
+#include "utils/TestUtils.h"
+#include "processors/GetTCP.h"
+#include "utils/StringUtils.h"
+#include "utils/file/PathUtils.h"
+
+using namespace std::literals::chrono_literals;
+
+namespace org::apache::nifi::minifi::test {
+
+class VerifyEmptyC2Metric : public VerifyC2Base {
+ public:
+ explicit VerifyEmptyC2Metric(const std::atomic_bool& metrics_found) : metrics_found_(metrics_found) {
+ }
+
+ void testSetup() override {
+ LogTestController::getInstance().setTrace<c2::C2Agent>();
+ LogTestController::getInstance().setTrace<c2::C2Client>();
+ LogTestController::getInstance().setDebug<c2::RESTSender>();
+ LogTestController::getInstance().setDebug<FlowController>();
+ LogTestController::getInstance().setOff<processors::GetTCP>();
+ VerifyC2Base::testSetup();
+ }
+
+ void runAssertions() override {
+ using org::apache::nifi::minifi::utils::verifyEventHappenedInPollTime;
+ assert(verifyEventHappenedInPollTime(40s, [&] { return metrics_found_.load(); }, 1s));
+ }
+
+ private:
+ const std::atomic_bool& metrics_found_;
+};
+
+class MetricsHandler: public HeartbeatHandler {
+ public:
+ explicit MetricsHandler(std::atomic_bool& metrics_found, std::shared_ptr<Configure> configuration)
+ : HeartbeatHandler(std::move(configuration)),
+ metrics_found_(metrics_found) {
+ }
+
+ void handleHeartbeat(const rapidjson::Document&, struct mg_connection* conn) override {
+ switch (state_) {
+ case TestState::DESCRIBE_SPECIFIC_METRIC: {
+ sendHeartbeatResponse("DESCRIBE", "metrics", "889347", conn, {{"metricsClass", "GetFileMetrics"}});
+ break;
+ }
+ case TestState::DESCRIBE_ALL_METRICS: {
+ sendHeartbeatResponse("DESCRIBE", "metrics", "889347", conn);
+ break;
+ }
+ default:
+ throw std::runtime_error("Unhandled test state");
+ }
+ }
+
+ void handleAcknowledge(const rapidjson::Document& root) override {
+ switch (state_) {
+ case TestState::DESCRIBE_SPECIFIC_METRIC: {
+ verifySpecificMetrics(root);
+ break;
+ }
+ case TestState::DESCRIBE_ALL_METRICS: {
+ verifyAllMetrics(root);
+ break;
+ }
+ default:
+ throw std::runtime_error("Unhandled test state");
+ }
+ }
+
+ private:
+ enum class TestState {
+ DESCRIBE_SPECIFIC_METRIC,
+ DESCRIBE_ALL_METRICS
+ };
+
+ static constexpr const char* GETFILE1_UUID = "471deef6-2a6e-4a7d-912a-81cc17e3a206";
+ static constexpr const char* GETFILE2_UUID = "471deef6-2a6e-4a7d-912a-81cc17e3a207";
+ static constexpr const char* GETTCP1_UUID = "2438e3c8-015a-1000-79ca-83af40ec1995";
+ static constexpr const char* GETTCP2_UUID = "2438e3c8-015a-1000-79ca-83af40ec1996";
+
+ static void sendEmptyHeartbeatResponse(struct mg_connection* conn) {
+ mg_printf(conn, "HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nContent-Length: 0\r\nConnection: close\r\n\r\n");
+ }
+
+ void verifySpecificMetrics(const rapidjson::Document& root) {
+ auto getfile_metrics_verified =
+ !root.HasMember("metrics") &&
+ root.HasMember("GetFileMetrics") &&
+ root["GetFileMetrics"].HasMember(GETFILE1_UUID) &&
+ root["GetFileMetrics"].HasMember(GETFILE2_UUID);
+ if (getfile_metrics_verified) {
+ state_ = TestState::DESCRIBE_ALL_METRICS;
+ }
+ }
+
+ void verifyAllMetrics(const rapidjson::Document& root) {
+ auto all_metrics_verified =
+ root.HasMember("metrics") &&
+ root["metrics"].HasMember("ProcessorMetrics") &&
+ root["metrics"]["ProcessorMetrics"].HasMember("GetFileMetrics") &&
+ root["metrics"]["ProcessorMetrics"]["GetFileMetrics"].HasMember(GETFILE1_UUID) &&
+ root["metrics"]["ProcessorMetrics"]["GetFileMetrics"].HasMember(GETFILE2_UUID) &&
+ root["metrics"]["ProcessorMetrics"]["GetTCPMetrics"].HasMember(GETTCP1_UUID) &&
+ root["metrics"]["ProcessorMetrics"]["GetTCPMetrics"].HasMember(GETTCP2_UUID) &&
+ root["metrics"].HasMember("SystemMetrics") &&
+ root["metrics"]["SystemMetrics"].HasMember("QueueMetrics");
+ if (all_metrics_verified) {
+ metrics_found_ = true;
+ }
+ }
+
+ TestState state_ = TestState::DESCRIBE_SPECIFIC_METRIC;
+ std::atomic_bool& metrics_found_;
+};
+
+} // namespace org::apache::nifi::minifi::test
+
+int main(int argc, char **argv) {
+ std::atomic_bool metrics_found{false};
+ const cmd_args args = parse_cmdline_args(argc, argv, "api/heartbeat");
+ org::apache::nifi::minifi::test::VerifyEmptyC2Metric harness(metrics_found);
+ harness.getConfiguration()->set("nifi.c2.root.class.definitions", "metrics");
+ harness.getConfiguration()->set("nifi.c2.root.class.definitions.metrics.name", "metrics");
+ harness.getConfiguration()->set("nifi.c2.root.class.definitions.metrics.metrics", "processormetrics,systemmetrics");
+ harness.getConfiguration()->set("nifi.c2.root.class.definitions.metrics.metrics.processormetrics.name", "ProcessorMetrics");
+ harness.getConfiguration()->set("nifi.c2.root.class.definitions.metrics.metrics.processormetrics.classes", "GetFileMetrics,GetTCPMetrics");
+ harness.getConfiguration()->set("nifi.c2.root.class.definitions.metrics.metrics.systemmetrics.name", "SystemMetrics");
+ harness.getConfiguration()->set("nifi.c2.root.class.definitions.metrics.metrics.systemmetrics.classes", "QueueMetrics");
+ harness.setKeyDir(args.key_dir);
+ org::apache::nifi::minifi::test::MetricsHandler handler(metrics_found, harness.getConfiguration());
+ harness.setUrl(args.url, &handler);
+ harness.run(args.test_file);
+ return 0;
+}
diff --git a/extensions/http-curl/tests/C2MetricsTest.cpp b/extensions/http-curl/tests/C2MetricsTest.cpp
index 7661fdd84..0e06d7478 100644
--- a/extensions/http-curl/tests/C2MetricsTest.cpp
+++ b/extensions/http-curl/tests/C2MetricsTest.cpp
@@ -94,6 +94,8 @@ class MetricsHandler: public HeartbeatHandler {
VERIFY_UPDATED_METRICS
};
+ static constexpr const char* GETTCP1_UUID = "2438e3c8-015a-1000-79ca-83af40ec1991";
+
static void sendEmptyHeartbeatResponse(struct mg_connection* conn) {
mg_printf(conn, "HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nContent-Length: 0\r\nConnection: close\r\n\r\n");
}
@@ -169,8 +171,9 @@ class MetricsHandler: public HeartbeatHandler {
static bool verifyProcessorMetrics(const rapidjson::Value& processor_metrics) {
return processor_metrics.HasMember("GetTCPMetrics") &&
- processor_metrics["GetTCPMetrics"].HasMember("OnTriggerInvocations") &&
- processor_metrics["GetTCPMetrics"]["OnTriggerInvocations"].GetUint() > 0;
+ processor_metrics["GetTCPMetrics"].HasMember(GETTCP1_UUID) &&
+ processor_metrics["GetTCPMetrics"][GETTCP1_UUID].HasMember("OnTriggerInvocations") &&
+ processor_metrics["GetTCPMetrics"][GETTCP1_UUID]["OnTriggerInvocations"].GetUint() > 0;
}
[[nodiscard]] static std::string getReplacementConfigAsJsonValue(const std::string& replacement_config_path) {
diff --git a/extensions/http-curl/tests/C2SameProcessorMetrics.cpp b/extensions/http-curl/tests/C2SameProcessorMetrics.cpp
new file mode 100644
index 000000000..c7d1b29aa
--- /dev/null
+++ b/extensions/http-curl/tests/C2SameProcessorMetrics.cpp
@@ -0,0 +1,115 @@
+/**
+ *
+ * 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.
+ */
+
+#undef NDEBUG
+#include <memory>
+
+#include "TestBase.h"
+#include "HTTPIntegrationBase.h"
+#include "HTTPHandlers.h"
+#include "processors/TailFile.h"
+#include "state/ProcessorController.h"
+#include "utils/file/FileUtils.h"
+#include "utils/TestUtils.h"
+#include "processors/GetTCP.h"
+#include "utils/StringUtils.h"
+#include "utils/file/PathUtils.h"
+
+using namespace std::literals::chrono_literals;
+
+namespace org::apache::nifi::minifi::test {
+
+class VerifyEmptyC2Metric : public VerifyC2Base {
+ public:
+ explicit VerifyEmptyC2Metric(const std::atomic_bool& metrics_found) : metrics_found_(metrics_found) {
+ }
+
+ void testSetup() override {
+ LogTestController::getInstance().setTrace<minifi::c2::C2Agent>();
+ LogTestController::getInstance().setTrace<minifi::c2::C2Client>();
+ LogTestController::getInstance().setDebug<minifi::c2::RESTSender>();
+ LogTestController::getInstance().setDebug<minifi::FlowController>();
+ LogTestController::getInstance().setOff<minifi::processors::GetTCP>();
+ VerifyC2Base::testSetup();
+ }
+
+ void runAssertions() override {
+ using org::apache::nifi::minifi::utils::verifyEventHappenedInPollTime;
+ assert(verifyEventHappenedInPollTime(40s, [&] { return metrics_found_.load(); }, 1s));
+ }
+
+ private:
+ const std::atomic_bool& metrics_found_;
+};
+
+class MetricsHandler: public HeartbeatHandler {
+ public:
+ explicit MetricsHandler(std::atomic_bool& metrics_found, std::shared_ptr<minifi::Configure> configuration)
+ : HeartbeatHandler(std::move(configuration)),
+ metrics_found_(metrics_found) {
+ }
+
+ void handleHeartbeat(const rapidjson::Document& root, struct mg_connection* conn) override {
+ verifyMetrics(root);
+ sendEmptyHeartbeatResponse(conn);
+ }
+
+ private:
+ static constexpr const char* GETFILE1_UUID = "471deef6-2a6e-4a7d-912a-81cc17e3a206";
+ static constexpr const char* GETFILE2_UUID = "471deef6-2a6e-4a7d-912a-81cc17e3a207";
+ static constexpr const char* GETTCP1_UUID = "2438e3c8-015a-1000-79ca-83af40ec1995";
+ static constexpr const char* GETTCP2_UUID = "2438e3c8-015a-1000-79ca-83af40ec1996";
+
+ static void sendEmptyHeartbeatResponse(struct mg_connection* conn) {
+ mg_printf(conn, "HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nContent-Length: 0\r\nConnection: close\r\n\r\n");
+ }
+
+ void verifyMetrics(const rapidjson::Document& root) {
+ auto initial_metrics_verified =
+ root.HasMember("metrics") &&
+ root["metrics"].HasMember("ProcessorMetrics") &&
+ root["metrics"]["ProcessorMetrics"].HasMember("GetFileMetrics") &&
+ root["metrics"]["ProcessorMetrics"]["GetFileMetrics"].HasMember(GETFILE1_UUID) &&
+ root["metrics"]["ProcessorMetrics"]["GetFileMetrics"].HasMember(GETFILE2_UUID) &&
+ root["metrics"]["ProcessorMetrics"]["GetTCPMetrics"].HasMember(GETTCP1_UUID) &&
+ root["metrics"]["ProcessorMetrics"]["GetTCPMetrics"].HasMember(GETTCP2_UUID);
+ if (initial_metrics_verified) {
+ metrics_found_ = true;
+ }
+ }
+
+ std::atomic_bool& metrics_found_;
+};
+
+} // namespace org::apache::nifi::minifi::test
+
+int main(int argc, char **argv) {
+ std::atomic_bool metrics_found{false};
+ const cmd_args args = parse_cmdline_args(argc, argv, "api/heartbeat");
+ org::apache::nifi::minifi::test::VerifyEmptyC2Metric harness(metrics_found);
+ harness.getConfiguration()->set("nifi.c2.root.class.definitions", "metrics");
+ harness.getConfiguration()->set("nifi.c2.root.class.definitions.metrics.name", "metrics");
+ harness.getConfiguration()->set("nifi.c2.root.class.definitions.metrics.metrics", "processormetrics");
+ harness.getConfiguration()->set("nifi.c2.root.class.definitions.metrics.metrics.processormetrics.name", "ProcessorMetrics");
+ harness.getConfiguration()->set("nifi.c2.root.class.definitions.metrics.metrics.processormetrics.classes", "GetFileMetrics,GetTCPMetrics");
+ harness.setKeyDir(args.key_dir);
+ org::apache::nifi::minifi::test::MetricsHandler handler(metrics_found, harness.getConfiguration());
+ harness.setUrl(args.url, &handler);
+ harness.run(args.test_file);
+ return 0;
+}
diff --git a/extensions/http-curl/tests/CMakeLists.txt b/extensions/http-curl/tests/CMakeLists.txt
index d24d0ac9e..3de352c0a 100644
--- a/extensions/http-curl/tests/CMakeLists.txt
+++ b/extensions/http-curl/tests/CMakeLists.txt
@@ -106,3 +106,5 @@ add_test(NAME C2UpdateAssetTest COMMAND C2UpdateAssetTest)
add_test(NAME C2CompressTest COMMAND C2CompressTest)
add_test(NAME C2MetricsTest COMMAND C2MetricsTest "${TEST_RESOURCES}/TestC2Metrics.yml")
add_test(NAME C2EmptyMetricTest COMMAND C2EmptyMetricTest "${TEST_RESOURCES}/TestEmpty.yml")
+add_test(NAME C2SameProcessorMetrics COMMAND C2SameProcessorMetrics "${TEST_RESOURCES}/TestSameProcessorMetrics.yml")
+add_test(NAME C2DescribeMetricsTest COMMAND C2DescribeMetricsTest "${TEST_RESOURCES}/TestSameProcessorMetrics.yml")
diff --git a/extensions/prometheus/PrometheusMetricsPublisher.cpp b/extensions/prometheus/PrometheusMetricsPublisher.cpp
index 007723367..85f122120 100644
--- a/extensions/prometheus/PrometheusMetricsPublisher.cpp
+++ b/extensions/prometheus/PrometheusMetricsPublisher.cpp
@@ -71,12 +71,12 @@ std::vector<std::shared_ptr<state::response::ResponseNode>> PrometheusMetricsPub
if (auto metric_classes_str = configuration_->get(minifi::Configuration::nifi_metrics_publisher_metrics)) {
auto metric_classes = utils::StringUtils::split(*metric_classes_str, ",");
for (const std::string& clazz : metric_classes) {
- auto response_node = response_node_loader_->loadResponseNode(clazz, root);
- if (!response_node) {
+ auto response_nodes = response_node_loader_->loadResponseNodes(clazz, root);
+ if (response_nodes.empty()) {
logger_->log_warn("Metric class '%s' could not be loaded.", clazz);
continue;
}
- nodes.push_back(response_node);
+ nodes.insert(nodes.end(), response_nodes.begin(), response_nodes.end());
}
}
return nodes;
diff --git a/extensions/standard-processors/processors/GetFile.h b/extensions/standard-processors/processors/GetFile.h
index 004427982..086ec41bd 100644
--- a/extensions/standard-processors/processors/GetFile.h
+++ b/extensions/standard-processors/processors/GetFile.h
@@ -62,23 +62,27 @@ class GetFileMetrics : public state::response::ResponseNode {
std::vector<state::response::SerializedResponseNode> serialize() override {
std::vector<state::response::SerializedResponseNode> resp;
+ state::response::SerializedResponseNode root_node;
+ root_node.name = source_component_.getUUIDStr();
+
state::response::SerializedResponseNode iter;
iter.name = "OnTriggerInvocations";
iter.value = (uint32_t)iterations_.load();
- resp.push_back(iter);
+ root_node.children.push_back(iter);
state::response::SerializedResponseNode accepted_files;
accepted_files.name = "AcceptedFiles";
accepted_files.value = (uint32_t)accepted_files_.load();
- resp.push_back(accepted_files);
+ root_node.children.push_back(accepted_files);
state::response::SerializedResponseNode input_bytes;
input_bytes.name = "InputBytes";
input_bytes.value = (uint32_t)input_bytes_.load();
- resp.push_back(input_bytes);
+ root_node.children.push_back(input_bytes);
+ resp.push_back(root_node);
return resp;
}
diff --git a/extensions/standard-processors/processors/GetTCP.h b/extensions/standard-processors/processors/GetTCP.h
index a2fa3b819..7e326eb19 100644
--- a/extensions/standard-processors/processors/GetTCP.h
+++ b/extensions/standard-processors/processors/GetTCP.h
@@ -111,11 +111,15 @@ class GetTCPMetrics : public state::response::ResponseNode {
std::vector<state::response::SerializedResponseNode> serialize() override {
std::vector<state::response::SerializedResponseNode> resp;
+ state::response::SerializedResponseNode root_node;
+ root_node.name = source_component_.getUUIDStr();
+
state::response::SerializedResponseNode iter;
iter.name = "OnTriggerInvocations";
iter.value = (uint32_t)iterations_.load();
- resp.push_back(iter);
+ root_node.children.push_back(iter);
+ resp.push_back(root_node);
return resp;
}
diff --git a/libminifi/include/c2/C2Client.h b/libminifi/include/c2/C2Client.h
index 26da4605b..816474d76 100644
--- a/libminifi/include/c2/C2Client.h
+++ b/libminifi/include/c2/C2Client.h
@@ -18,7 +18,7 @@
#pragma once
-#include <map>
+#include <unordered_map>
#include <memory>
#include <mutex>
#include <optional>
@@ -77,7 +77,11 @@ class C2Client : public core::Flow, public state::response::NodeReporter {
bool initialized_ = false;
std::shared_ptr<core::logging::Logger> logger_;
mutable std::mutex metrics_mutex_;
- std::map<std::string, std::shared_ptr<state::response::ResponseNode>> root_response_nodes_;
+
+ // Name and response node value of the root response nodes defined in nifi.c2.root.classes and nifi.c2.root.class.definitions
+ // In case a root class is defined to be a processor metric there can be multiple response nodes if the same processor is defined
+ // multiple times in the flow
+ std::unordered_map<std::string, std::vector<std::shared_ptr<state::response::ResponseNode>>> root_response_nodes_;
protected:
std::atomic<bool> flow_update_{false};
diff --git a/libminifi/include/core/state/nodes/MetricsBase.h b/libminifi/include/core/state/nodes/MetricsBase.h
index 6fe7dcb54..da545fb14 100644
--- a/libminifi/include/core/state/nodes/MetricsBase.h
+++ b/libminifi/include/core/state/nodes/MetricsBase.h
@@ -22,6 +22,7 @@
#include <memory>
#include <string>
#include <optional>
+#include <unordered_map>
#include "../Value.h"
#include "../PublishedMetricProvider.h"
@@ -51,6 +52,19 @@ class ResponseNode : public core::Connectable, public PublishedMetricProvider {
}
virtual ~ResponseNode() = default;
+ static std::vector<state::response::SerializedResponseNode> serializeAndMergeResponseNodes(const std::vector<std::shared_ptr<ResponseNode>>& nodes) {
+ if (nodes.empty()) {
+ return {};
+ }
+
+ std::vector<state::response::SerializedResponseNode> result;
+ for (const auto& node: nodes) {
+ auto serialized = node->serialize();
+ result.insert(result.end(), serialized.begin(), serialized.end());
+ }
+ return result;
+ }
+
virtual std::vector<SerializedResponseNode> serialize() = 0;
virtual void yield() {
@@ -101,7 +115,7 @@ class ObjectNode : public ResponseNode {
}
void add_node(const std::shared_ptr<ResponseNode> &node) {
- nodes_.push_back(node);
+ nodes_[node->getName()].push_back(node);
}
std::string getName() const override {
@@ -110,19 +124,19 @@ class ObjectNode : public ResponseNode {
std::vector<SerializedResponseNode> serialize() override {
std::vector<SerializedResponseNode> serialized;
-// SerializedResponseNode outer_node;
- // outer_node.name = getName();
- for (auto &node : nodes_) {
+ for (const auto& [name, nodes] : nodes_) {
+ if (nodes.empty()) {
+ continue;
+ }
SerializedResponseNode inner_node;
- inner_node.name = node->getName();
- for (auto &embed : node->serialize()) {
+ inner_node.name = nodes[0]->getName();
+ for (auto &embed : ResponseNode::serializeAndMergeResponseNodes(nodes)) {
if (!embed.empty() || embed.keep_empty) {
inner_node.children.push_back(std::move(embed));
}
}
serialized.push_back(std::move(inner_node));
}
- // serialized.push_back(std::move(outer_node));
return serialized;
}
@@ -131,7 +145,7 @@ class ObjectNode : public ResponseNode {
}
protected:
- std::vector<std::shared_ptr<ResponseNode>> nodes_;
+ std::unordered_map<std::string, std::vector<std::shared_ptr<ResponseNode>>> nodes_;
};
/**
diff --git a/libminifi/include/core/state/nodes/ResponseNodeLoader.h b/libminifi/include/core/state/nodes/ResponseNodeLoader.h
index ffd7a57f9..58d2c8d85 100644
--- a/libminifi/include/core/state/nodes/ResponseNodeLoader.h
+++ b/libminifi/include/core/state/nodes/ResponseNodeLoader.h
@@ -23,6 +23,7 @@
#include <memory>
#include <mutex>
#include <functional>
+#include <vector>
#include "MetricsBase.h"
#include "core/ProcessGroup.h"
@@ -38,14 +39,14 @@ class ResponseNodeLoader {
public:
ResponseNodeLoader(std::shared_ptr<Configure> configuration, std::shared_ptr<core::Repository> provenance_repo,
std::shared_ptr<core::Repository> flow_file_repo, core::FlowConfiguration* flow_configuration);
- std::shared_ptr<ResponseNode> loadResponseNode(const std::string& clazz, core::ProcessGroup* root);
- std::shared_ptr<state::response::ResponseNode> getComponentMetricsNode(const std::string& metrics_class) const;
+ std::vector<std::shared_ptr<ResponseNode>> loadResponseNodes(const std::string& clazz, core::ProcessGroup* root);
+ std::vector<std::shared_ptr<ResponseNode>> getComponentMetricsNodes(const std::string& metrics_class) const;
void setControllerServiceProvider(core::controller::ControllerServiceProvider* controller);
void setStateMonitor(state::StateMonitor* update_sink);
void initializeComponentMetrics(core::ProcessGroup* root);
private:
- std::shared_ptr<ResponseNode> getResponseNode(const std::string& clazz) const;
+ std::vector<std::shared_ptr<ResponseNode>> getResponseNodes(const std::string& clazz) const;
void initializeRepositoryMetrics(const std::shared_ptr<ResponseNode>& response_node);
static void initializeQueueMetrics(const std::shared_ptr<ResponseNode>& response_node, core::ProcessGroup* root);
void initializeAgentIdentifier(const std::shared_ptr<ResponseNode>& response_node);
@@ -55,7 +56,7 @@ class ResponseNodeLoader {
void initializeFlowMonitor(const std::shared_ptr<ResponseNode>& response_node, core::ProcessGroup* root);
mutable std::mutex component_metrics_mutex_;
- std::unordered_map<std::string, std::shared_ptr<ResponseNode>> component_metrics_;
+ std::unordered_map<std::string, std::vector<std::shared_ptr<ResponseNode>>> component_metrics_;
std::shared_ptr<Configure> configuration_;
std::shared_ptr<core::Repository> provenance_repo_;
std::shared_ptr<core::Repository> flow_file_repo_;
diff --git a/libminifi/src/c2/C2Client.cpp b/libminifi/src/c2/C2Client.cpp
index 14829aa75..c7adee079 100644
--- a/libminifi/src/c2/C2Client.cpp
+++ b/libminifi/src/c2/C2Client.cpp
@@ -32,6 +32,7 @@
#include "core/state/nodes/FlowInformation.h"
#include "utils/file/FileSystem.h"
#include "utils/file/FileUtils.h"
+#include "utils/gsl.h"
namespace org::apache::nifi::minifi::c2 {
@@ -102,11 +103,13 @@ std::optional<std::string> C2Client::fetchFlow(const std::string& uri) const {
void C2Client::loadNodeClasses(const std::string& class_definitions, const std::shared_ptr<state::response::ResponseNode>& new_node) {
auto classes = utils::StringUtils::split(class_definitions, ",");
for (const std::string& clazz : classes) {
- auto response_node = response_node_loader_.loadResponseNode(clazz, root_.get());
- if (!response_node) {
+ auto response_nodes = response_node_loader_.loadResponseNodes(clazz, root_.get());
+ if (response_nodes.empty()) {
continue;
}
- std::static_pointer_cast<state::response::ObjectNode>(new_node)->add_node(response_node);
+ for (const auto& response_node : response_nodes) {
+ std::static_pointer_cast<state::response::ObjectNode>(new_node)->add_node(response_node);
+ }
}
}
@@ -137,7 +140,7 @@ void C2Client::loadC2ResponseConfiguration(const std::string &prefix) {
}
// We don't need to lock here, we already do it in the initializeResponseNodes member function
- root_response_nodes_[name] = new_node;
+ root_response_nodes_[name].push_back(new_node);
} catch (...) {
logger_->log_error("Could not create metrics class %s", metricsClass);
}
@@ -191,23 +194,24 @@ std::shared_ptr<state::response::ResponseNode> C2Client::loadC2ResponseConfigura
std::optional<state::response::NodeReporter::ReportedNode> C2Client::getMetricsNode(const std::string& metrics_class) const {
std::lock_guard<std::mutex> guard{metrics_mutex_};
+ const auto createReportedNode = [](const std::vector<std::shared_ptr<state::response::ResponseNode>>& nodes) {
+ gsl_Expects(!nodes.empty());
+ state::response::NodeReporter::ReportedNode reported_node;
+ reported_node.is_array = nodes[0]->isArray();
+ reported_node.name = nodes[0]->getName();
+ reported_node.serialized_nodes = state::response::ResponseNode::serializeAndMergeResponseNodes(nodes);
+ return reported_node;
+ };
+
if (!metrics_class.empty()) {
- auto metrics_node = response_node_loader_.getComponentMetricsNode(metrics_class);
- if (metrics_node) {
- state::response::NodeReporter::ReportedNode reported_node;
- reported_node.is_array = metrics_node->isArray();
- reported_node.name = metrics_node->getName();
- reported_node.serialized_nodes = metrics_node->serialize();
- return reported_node;
+ auto metrics_nodes = response_node_loader_.getComponentMetricsNodes(metrics_class);
+ if (!metrics_nodes.empty()) {
+ return createReportedNode(metrics_nodes);
}
} else {
- const auto iter = root_response_nodes_.find("metrics");
- if (iter != root_response_nodes_.end()) {
- state::response::NodeReporter::ReportedNode reported_node;
- reported_node.is_array = iter->second->isArray();
- reported_node.name = iter->second->getName();
- reported_node.serialized_nodes = iter->second->serialize();
- return reported_node;
+ const auto metrics_it = root_response_nodes_.find("metrics");
+ if (metrics_it != root_response_nodes_.end()) {
+ return createReportedNode(metrics_it->second);
}
}
return std::nullopt;
@@ -218,38 +222,45 @@ std::vector<state::response::NodeReporter::ReportedNode> C2Client::getHeartbeatN
configuration_->get(minifi::Configuration::nifi_c2_full_heartbeat, fullHb);
const bool include = include_manifest || fullHb == "true";
- std::vector<state::response::NodeReporter::ReportedNode> nodes;
+ std::vector<state::response::NodeReporter::ReportedNode> reported_nodes;
std::lock_guard<std::mutex> guard{metrics_mutex_};
- nodes.reserve(root_response_nodes_.size());
- for (const auto &entry : root_response_nodes_) {
- auto identifier = std::dynamic_pointer_cast<state::response::AgentIdentifier>(entry.second);
- if (identifier) {
- identifier->includeAgentManifest(include);
- }
- if (entry.second) {
- state::response::NodeReporter::ReportedNode reported_node;
- reported_node.name = entry.second->getName();
- reported_node.is_array = entry.second->isArray();
- reported_node.serialized_nodes = entry.second->serialize();
- nodes.push_back(reported_node);
+ reported_nodes.reserve(root_response_nodes_.size());
+ for (const auto& [name, node_values] : root_response_nodes_) {
+ for (const auto& node : node_values) {
+ auto identifier = std::dynamic_pointer_cast<state::response::AgentIdentifier>(node);
+ if (identifier) {
+ identifier->includeAgentManifest(include);
+ }
+ if (node) {
+ state::response::NodeReporter::ReportedNode reported_node;
+ reported_node.name = node->getName();
+ reported_node.is_array = node->isArray();
+ reported_node.serialized_nodes = node->serialize();
+ reported_nodes.push_back(reported_node);
+ }
}
}
- return nodes;
+ return reported_nodes;
}
void C2Client::initializeResponseNodes(core::ProcessGroup* root) {
+ if (!root_response_nodes_.empty()) {
+ return;
+ }
std::string class_csv;
std::lock_guard<std::mutex> guard{metrics_mutex_};
if (configuration_->get(minifi::Configuration::nifi_c2_root_classes, class_csv)) {
std::vector<std::string> classes = utils::StringUtils::split(class_csv, ",");
for (const std::string& clazz : classes) {
- auto response_node = response_node_loader_.loadResponseNode(clazz, root);
- if (!response_node) {
+ auto response_nodes = response_node_loader_.loadResponseNodes(clazz, root);
+ if (response_nodes.empty()) {
continue;
}
- root_response_nodes_[response_node->getName()] = std::move(response_node);
+ for (auto&& response_node: response_nodes) {
+ root_response_nodes_[response_node->getName()].push_back(std::move(response_node));
+ }
}
}
diff --git a/libminifi/src/core/state/nodes/ResponseNodeLoader.cpp b/libminifi/src/core/state/nodes/ResponseNodeLoader.cpp
index f98d29379..feb44a00e 100644
--- a/libminifi/src/core/state/nodes/ResponseNodeLoader.cpp
+++ b/libminifi/src/core/state/nodes/ResponseNodeLoader.cpp
@@ -26,6 +26,7 @@
#include "core/state/nodes/AgentInformation.h"
#include "core/state/nodes/ConfigurationChecksums.h"
#include "c2/C2Agent.h"
+#include "utils/gsl.h"
namespace org::apache::nifi::minifi::state::response {
@@ -59,17 +60,22 @@ void ResponseNodeLoader::initializeComponentMetrics(core::ProcessGroup* root) {
node_source->getResponseNodes(metric_vector);
std::lock_guard<std::mutex> guard(component_metrics_mutex_);
for (const auto& metric : metric_vector) {
- component_metrics_[metric->getName()] = metric;
+ component_metrics_[metric->getName()].push_back(metric);
}
}
}
-std::shared_ptr<ResponseNode> ResponseNodeLoader::getResponseNode(const std::string& clazz) const {
+std::vector<std::shared_ptr<ResponseNode>> ResponseNodeLoader::getResponseNodes(const std::string& clazz) const {
std::shared_ptr<core::CoreComponent> ptr = core::ClassLoader::getDefaultClassLoader().instantiate(clazz, clazz);
if (ptr == nullptr) {
- return getComponentMetricsNode(clazz);
+ return getComponentMetricsNodes(clazz);
}
- return std::dynamic_pointer_cast<ResponseNode>(ptr);
+ auto response_node = std::dynamic_pointer_cast<ResponseNode>(ptr);
+ if (!response_node) {
+ logger_->log_error("Instantiated class '%s' is not a ResponseNode!", clazz);
+ return {};
+ }
+ return {response_node};
}
void ResponseNodeLoader::initializeRepositoryMetrics(const std::shared_ptr<ResponseNode>& response_node) {
@@ -153,24 +159,26 @@ void ResponseNodeLoader::initializeFlowMonitor(const std::shared_ptr<ResponseNod
}
}
-std::shared_ptr<ResponseNode> ResponseNodeLoader::loadResponseNode(const std::string& clazz, core::ProcessGroup* root) {
- auto response_node = getResponseNode(clazz);
- if (!response_node) {
+std::vector<std::shared_ptr<ResponseNode>> ResponseNodeLoader::loadResponseNodes(const std::string& clazz, core::ProcessGroup* root) {
+ auto response_nodes = getResponseNodes(clazz);
+ if (response_nodes.empty()) {
logger_->log_error("No metric defined for %s", clazz);
- return nullptr;
+ return {};
}
- initializeRepositoryMetrics(response_node);
- initializeQueueMetrics(response_node, root);
- initializeAgentIdentifier(response_node);
- initializeAgentMonitor(response_node);
- initializeAgentNode(response_node);
- initializeConfigurationChecksums(response_node);
- initializeFlowMonitor(response_node, root);
- return response_node;
+ for (const auto& response_node : response_nodes) {
+ initializeRepositoryMetrics(response_node);
+ initializeQueueMetrics(response_node, root);
+ initializeAgentIdentifier(response_node);
+ initializeAgentMonitor(response_node);
+ initializeAgentNode(response_node);
+ initializeConfigurationChecksums(response_node);
+ initializeFlowMonitor(response_node, root);
+ }
+ return response_nodes;
}
-std::shared_ptr<state::response::ResponseNode> ResponseNodeLoader::getComponentMetricsNode(const std::string& metrics_class) const {
+std::vector<std::shared_ptr<ResponseNode>> ResponseNodeLoader::getComponentMetricsNodes(const std::string& metrics_class) const {
if (!metrics_class.empty()) {
std::lock_guard<std::mutex> lock(component_metrics_mutex_);
const auto citer = component_metrics_.find(metrics_class);
@@ -178,7 +186,7 @@ std::shared_ptr<state::response::ResponseNode> ResponseNodeLoader::getComponentM
return citer->second;
}
}
- return nullptr;
+ return {};
}
void ResponseNodeLoader::setControllerServiceProvider(core::controller::ControllerServiceProvider* controller) {
diff --git a/libminifi/test/resources/TestSameProcessorMetrics.yml b/libminifi/test/resources/TestSameProcessorMetrics.yml
new file mode 100644
index 000000000..2c842b8b8
--- /dev/null
+++ b/libminifi/test/resources/TestSameProcessorMetrics.yml
@@ -0,0 +1,123 @@
+#
+# 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.
+#
+Flow Controller:
+ name: MiNiFi Flow
+ id: 2438e3c8-015a-1000-79ca-83af40ec1990
+Processors:
+ - name: GetFile1
+ id: 471deef6-2a6e-4a7d-912a-81cc17e3a206
+ class: org.apache.nifi.processors.standard.GetFile
+ max concurrent tasks: 1
+ scheduling strategy: TIMER_DRIVEN
+ scheduling period: 1 sec
+ penalization period: 30 sec
+ yield period: 1 sec
+ run duration nanos: 0
+ auto-terminated relationships list:
+ Properties:
+ Input Directory: /tmp/getfile1
+ Keep Source File: true
+ - name: GetFile2
+ id: 471deef6-2a6e-4a7d-912a-81cc17e3a207
+ class: org.apache.nifi.processors.standard.GetFile
+ max concurrent tasks: 1
+ scheduling strategy: TIMER_DRIVEN
+ scheduling period: 1 sec
+ penalization period: 30 sec
+ yield period: 1 sec
+ run duration nanos: 0
+ auto-terminated relationships list:
+ Properties:
+ Input Directory: /tmp/getfile2
+ Keep Source File: true
+ - name: GetTCP1
+ id: 2438e3c8-015a-1000-79ca-83af40ec1995
+ class: org.apache.nifi.processors.standard.GetTCP
+ max concurrent tasks: 1
+ scheduling strategy: TIMER_DRIVEN
+ scheduling period: 10 msec
+ penalization period: 30 msec
+ yield period: 10 msec
+ run duration nanos: 0
+ auto-terminated relationships list:
+ - success
+ - partial
+ Properties:
+ SSL Context Service: SSLContextService
+ endpoint-list: localhost:8776
+ end-of-message-byte: d
+ reconnect-interval: 100ms
+ connection-attempt-timeout: 2000
+ - name: GetTCP2
+ id: 2438e3c8-015a-1000-79ca-83af40ec1996
+ class: org.apache.nifi.processors.standard.GetTCP
+ max concurrent tasks: 1
+ scheduling strategy: TIMER_DRIVEN
+ scheduling period: 10 msec
+ penalization period: 30 msec
+ yield period: 10 msec
+ run duration nanos: 0
+ auto-terminated relationships list:
+ - success
+ - partial
+ Properties:
+ SSL Context Service: SSLContextService
+ endpoint-list: localhost:8776
+ end-of-message-byte: d
+ reconnect-interval: 100ms
+ connection-attempt-timeout: 2000
+ - name: LogAttribute
+ id: 2438e3c8-015a-1000-79ca-83af40ec1992
+ class: org.apache.nifi.processors.standard.LogAttribute
+ max concurrent tasks: 1
+ scheduling strategy: TIMER_DRIVEN
+ scheduling period: 1 sec
+ penalization period: 30 sec
+ yield period: 1 sec
+ run duration nanos: 0
+ auto-terminated relationships list:
+ - success
+ Properties:
+ Log Level: info
+ Log Payload: true
+
+Connections:
+ - name: GetFile1/success/LogAttribute
+ id: 2438e3c8-015a-1000-79ca-83af40ec1997
+ source name: GetFile1
+ source id: 471deef6-2a6e-4a7d-912a-81cc17e3a206
+ source relationship name: success
+ destination name: LogAttribute
+ destination id: 2438e3c8-015a-1000-79ca-83af40ec1992
+ max work queue size: 0
+ max work queue data size: 1 MB
+ flowfile expiration: 60 sec
+ - name: GetFile2/success/LogAttribute
+ id: 2438e3c8-015a-1000-79ca-83af40ec1998
+ source name: GetFile2
+ source id: 471deef6-2a6e-4a7d-912a-81cc17e3a207
+ source relationship name: success
+ destination name: LogAttribute
+ destination id: 2438e3c8-015a-1000-79ca-83af40ec1992
+ max work queue size: 0
+ max work queue data size: 1 MB
+ flowfile expiration: 60 sec
+
+Remote Processing Groups:
+