lordgamez commented on code in PR #1340: URL: https://github.com/apache/nifi-minifi-cpp/pull/1340#discussion_r893443384
########## Windows.md: ########## @@ -67,6 +67,7 @@ After the build directory it will take optional parameters modifying the CMake c | /L | Enables Linter | | /O | Enables OpenCV | | /PDH | Enables Performance Monitor | +| /P | Enables Prometheus | Review Comment: Good catch, fixed in 4f53b2c73daee9f416fab8ad038404064002d0e8 ########## extensions/http-curl/tests/C2MetricsTest.cpp: ########## @@ -0,0 +1,209 @@ +/** + * + * 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 <string> +#include <iostream> +#include <filesystem> + +#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 VerifyC2Metrics : public VerifyC2Base { + public: + explicit VerifyC2Metrics(const std::atomic_bool& metrics_updated_successfully) : metrics_updated_successfully_(metrics_updated_successfully) { + } + + 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_updated_successfully_.load(); }, 1s)); + } + + private: + const std::atomic_bool& metrics_updated_successfully_; +}; + +class MetricsHandler: public HeartbeatHandler { + public: + explicit MetricsHandler(std::atomic_bool& metrics_updated_successfully, std::shared_ptr<minifi::Configure> configuration, const std::string& replacement_config_path) + : HeartbeatHandler(std::move(configuration)), + metrics_updated_successfully_(metrics_updated_successfully), + replacement_config_(getReplacementConfigAsJsonValue(replacement_config_path)) { + } + + void handleHeartbeat(const rapidjson::Document& root, struct mg_connection* conn) override { + switch (test_state_) { + case TestState::VERIFY_INITIAL_METRICS: { + verifyMetrics(root); + sendEmptyHeartbeatResponse(conn); + break; + } + case TestState::SEND_NEW_CONFIG: { + sendHeartbeatResponse("UPDATE", "configuration", "889348", conn, {{"configuration_data", replacement_config_}}); + test_state_ = TestState::VERIFY_UPDATED_METRICS; + break; + } + case TestState::VERIFY_UPDATED_METRICS: { + verifyUpdatedMetrics(root); + sendEmptyHeartbeatResponse(conn); + break; + } + } + } + + private: + enum class TestState { + VERIFY_INITIAL_METRICS, + SEND_NEW_CONFIG, + VERIFY_UPDATED_METRICS + }; + + 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("RuntimeMetrics") && + root["metrics"].HasMember("LoadMetrics") && + root["metrics"].HasMember("ProcessorMetrics"); + if (initial_metrics_verified) { + test_state_ = TestState::SEND_NEW_CONFIG; + } + } + + void verifyUpdatedMetrics(const rapidjson::Document& root) { + auto updated_metrics_verified = + root.HasMember("metrics") && + root["metrics"].HasMember("RuntimeMetrics") && + root["metrics"].HasMember("LoadMetrics") && + !root["metrics"].HasMember("ProcessorMetrics") && + verifyUpdatedRuntimeMetrics(root["metrics"]["RuntimeMetrics"]) && + verifyUpdatedLoadMetrics(root["metrics"]["LoadMetrics"]); + + if (updated_metrics_verified) { + metrics_updated_successfully_ = true; + } + } + + static bool verifyRuntimeMetrics(const rapidjson::Value& runtime_metrics) { + return runtime_metrics.HasMember("deviceInfo") && + runtime_metrics.HasMember("flowInfo") && + runtime_metrics["flowInfo"].HasMember("versionedFlowSnapshotURI") && + runtime_metrics["flowInfo"].HasMember("queues") && + runtime_metrics["flowInfo"].HasMember("components") && + runtime_metrics["flowInfo"]["queues"].HasMember("2438e3c8-015a-1000-79ca-83af40ec1997") && + runtime_metrics["flowInfo"]["components"].HasMember("FlowController") && + runtime_metrics["flowInfo"]["components"].HasMember("GetTCP") && + runtime_metrics["flowInfo"]["components"].HasMember("LogAttribute"); + } Review Comment: I checked it, but I think in this case it's not really viable. There are a lot of environment or flow state specific metrics that cannot be checked for and exact expected json and I think it's better to check on these specific fields separately. ########## extensions/prometheus/PrometheusExposerWrapper.cpp: ########## @@ -0,0 +1,34 @@ +/** + * 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. + */ +#include "PrometheusExposerWrapper.h" + +namespace org::apache::nifi::minifi::extensions::prometheus { + +PrometheusExposerWrapper::PrometheusExposerWrapper(uint32_t port) + : exposer_(std::to_string(port)) { + logger_->log_info("Started Prometheus metrics publisher on port %u", port); Review Comment: Good point, updated in 4f53b2c73daee9f416fab8ad038404064002d0e8 ########## libminifi/include/core/state/nodes/QueueMetrics.h: ########## @@ -94,15 +82,20 @@ class QueueMetrics : public ResponseNode { return serialized; } - protected: - std::map<std::string, std::unique_ptr<minifi::Connection>> connections; + std::vector<PublishedMetric> calculateMetrics() override { + std::vector<PublishedMetric> metrics; + for (const auto& [_, connection] : connections_) { + metrics.push_back({"queue_data_size", static_cast<double>(connection->getQueueDataSize()), + {{"connection_uuid", connection->getUUIDStr()}, {"connection_name", connection->getName()}, {"metric_class", getName()}}}); + metrics.push_back({"queue_data_size_max", static_cast<double>(connection->getMaxQueueDataSize()), + {{"connection_uuid", connection->getUUIDStr()}, {"connection_name", connection->getName()}, {"metric_class", getName()}}}); + metrics.push_back({"queue_size", static_cast<double>(connection->getQueueSize()), + {{"connection_uuid", connection->getUUIDStr()}, {"connection_name", connection->getName()}, {"metric_class", getName()}}}); + metrics.push_back({"queue_size_max", static_cast<double>(connection->getMaxQueueSize()), + {{"connection_uuid", connection->getUUIDStr()}, {"connection_name", connection->getName()}, {"metric_class", getName()}}}); + } + return metrics; Review Comment: I moved the connection metric calculation to the ConnectionStore base class which returns a PublishedMetrics collection. Currently it has the name, value and labels structure as a data format for providing metrics. We shall see if that holds up to be generic enough when another metric publisher is implemented or it needs to be changed. ########## libminifi/src/core/state/nodes/QueueMetrics.cpp: ########## @@ -0,0 +1,27 @@ +/** + * + * 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. + */ + +#include "core/state/nodes/QueueMetrics.h" +#include "core/Resource.h" + +namespace org::apache::nifi::minifi::state::response { + +REGISTER_RESOURCE(QueueMetrics, "Node part of an AST that defines queue metric information"); Review Comment: I think it's a legacy term used in all the metric node descriptions, not sure why, as in C2 the metric nodes are represented in kind of a tree format, but not an AST. I changed it to be a more generic "Metric node" in the descriptions in 4f53b2c73daee9f416fab8ad038404064002d0e8. ########## docker/test/integration/minifi/core/DockerTestCluster.py: ########## @@ -283,3 +282,79 @@ def write_content_to_container(self, content, dst): tar.addfile(info, io.BytesIO(content.encode('utf-8'))) with open(os.path.join(td, 'content.tar'), 'rb') as data: return container.put_archive(os.path.dirname(dst_path), data.read()) + + def check_metric_class_on_prometheus(self, metric_class, timeout_seconds): + start_time = time.perf_counter() + while (time.perf_counter() - start_time) < timeout_seconds: + if self.verify_metric_class(metric_class): + return True + time.sleep(1) + return False Review Comment: Good point, changed it in 4f53b2c73daee9f416fab8ad038404064002d0e8 ########## libminifi/include/FlowController.h: ########## @@ -252,8 +255,7 @@ class FlowController : public core::controller::ForwardingControllerServiceProvi // Thread pool for schedulers utils::ThreadPool<utils::TaskRescheduleInfo> thread_pool_; std::map<utils::Identifier, std::unique_ptr<state::ProcessorController>> processor_to_controller_; + std::unique_ptr<state::MetricsPublisher> metrics_publisher_; Review Comment: I agree, it should not depend on it, and I started out with that concept at the beginning, but unfortunately I had a few major problems with the current design in the FlowContrroller, C2Client and the metric ResponseNodes. 1. There are some metric nodes that depend on parts of the FlowController like the AgentNode which depends on the controller service provider to be initialized when loading that node. This requires that the ResponseNodeLoader should only be initialized after the FlowController initialized its `controller_service_provider_impl_` member and only after that we can initialize the metrics node publisher to make sure we can load any of the configured metric nodes. 2. In case of a flow update we need to make sure to remove all the metric nodes before applying the flow config, to avoid problems when a metric is collected on another thread and tries to query non-existing connections or processgroups. After the update these metric nodes need to be repopulated and as the FlowController controls this configuration update and the new flow configuration, we unfortunately have to depend on it to provide the new flow and take care of the thread synchronization while updating. These dependencies are not ideal at all and should be changed in the future, but looks to be a cumbersome change to figure out how these dependencies should be fixed and managed later in the future. I would propose a separate jira ticket for a future PR that resolves these issue, if that okay? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
