This is an automated email from the ASF dual-hosted git repository. aboda pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit 9552757b07ce95e22601f85ae661aa62555272d7 Author: Murtuza <[email protected]> AuthorDate: Mon Feb 10 16:20:09 2020 -0500 MINIFICPP-1157 Implement light weight heartbeat. Remove agent manifest from regular heartbeat messages. Send agent manifest in response to DESCRIBE manifest request Signed-off-by: Arpad Boda <[email protected]> --- C2.md | 5 + conf/minifi.properties | 2 + extensions/coap/tests/CoapIntegrationBase.h | 8 +- .../http-curl/tests/C2DescribeManifestTest.cpp | 91 +++++++++++++ extensions/http-curl/tests/C2JstackTest.cpp | 133 +++++-------------- .../http-curl/tests/C2VerifyHeartbeatAndStop.cpp | 144 +++++++-------------- extensions/http-curl/tests/CMakeLists.txt | 1 + extensions/http-curl/tests/HTTPHandlers.h | 102 +++++++++++++++ extensions/http-curl/tests/HTTPIntegrationBase.h | 78 ++++++++++- libminifi/include/FlowController.h | 20 ++- libminifi/include/c2/C2Agent.h | 7 + .../include/core/state/nodes/AgentInformation.h | 137 +++++++++++++++++--- libminifi/include/core/state/nodes/MetricsBase.h | 20 ++- libminifi/include/properties/Properties.h | 2 +- libminifi/src/FlowController.cpp | 38 ++++++ libminifi/src/c2/C2Agent.cpp | 143 ++++++++++---------- libminifi/test/integration/IntegrationBase.h | 11 +- 17 files changed, 631 insertions(+), 311 deletions(-) diff --git a/C2.md b/C2.md index b907926..dc99778 100644 --- a/C2.md +++ b/C2.md @@ -49,6 +49,11 @@ Release 0.6.0: Please note that all c2 properties now exist as `nifi.c2.*`. If y files contain the former naming convention of `c2.*`, we will continue to support that as an alternate key, but you are encouraged to switch your configuration options as soon as possible. +Note: In release 0.8.0 there is a configuration option to minizime the heartbeat payload size by excluding agent manifest. +For that, replace "AgentInformation" with "AgentInformationWithoutManifest in nifi.c2.root.classes property value. +With this change, heartbeat with agent manifest included is sent only for the first time then falls back to sending +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 diff --git a/conf/minifi.properties b/conf/minifi.properties index 4f7166a..249ca7a 100644 --- a/conf/minifi.properties +++ b/conf/minifi.properties @@ -52,6 +52,8 @@ nifi.database.content.repository.directory.default=${MINIFI_HOME}/content_reposi #nifi.c2.rest.url= #nifi.c2.rest.url.ack= nifi.c2.root.classes=DeviceInfoNode,AgentInformation,FlowInformation +## Minimize heartbeat payload size by excluding agent manifest from the heartbeat +#nifi.c2.root.classes=DeviceInfoNode,AgentInformationWithoutManifest,FlowInformation ## heartbeat 4 times a second #nifi.c2.agent.heartbeat.period=250 ## define parameters about your agent diff --git a/extensions/coap/tests/CoapIntegrationBase.h b/extensions/coap/tests/CoapIntegrationBase.h index 83935bf..bd7740e 100644 --- a/extensions/coap/tests/CoapIntegrationBase.h +++ b/extensions/coap/tests/CoapIntegrationBase.h @@ -41,7 +41,7 @@ class CoapIntegrationBase : public IntegrationBase { void setUrl(std::string url, CivetHandler *handler); - virtual ~CoapIntegrationBase(); + virtual ~CoapIntegrationBase() = default; void shutdownBeforeFlowController() override { stop_webserver(server); @@ -68,6 +68,8 @@ class CoapIntegrationBase : public IntegrationBase { queryRootProcessGroup(pg); + configureC2RootClasses(); + ptr.release(); std::shared_ptr<TestRepository> repo = std::static_pointer_cast<TestRepository>(test_repo); @@ -90,10 +92,6 @@ class CoapIntegrationBase : public IntegrationBase { CivetServer *server; }; -CoapIntegrationBase::~CoapIntegrationBase() { - -} - void CoapIntegrationBase::setUrl(std::string url, CivetHandler *handler) { parse_http_components(url, port, scheme, path); diff --git a/extensions/http-curl/tests/C2DescribeManifestTest.cpp b/extensions/http-curl/tests/C2DescribeManifestTest.cpp new file mode 100644 index 0000000..5a84c5c --- /dev/null +++ b/extensions/http-curl/tests/C2DescribeManifestTest.cpp @@ -0,0 +1,91 @@ +/** + * + * 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 <sys/stat.h> +#undef NDEBUG +#include <cassert> +#include <utility> +#include <chrono> +#include <fstream> +#include <memory> +#include <string> +#include <thread> +#include <type_traits> +#include <vector> +#include <iostream> +#include <sstream> +#include "HTTPClient.h" +#include "InvokeHTTP.h" +#include "TestBase.h" +#include "utils/StringUtils.h" +#include "core/Core.h" +#include "core/logging/Logger.h" +#include "core/ProcessGroup.h" +#include "core/yaml/YamlConfiguration.h" +#include "FlowController.h" +#include "properties/Configure.h" +#include "unit/ProvenanceTestHelper.h" +#include "io/StreamFactory.h" +#include "c2/C2Agent.h" +#include "CivetServer.h" +#include <cstring> +#include "protocols/RESTSender.h" +#include "HTTPIntegrationBase.h" +#include "HTTPHandlers.h" + +class DescribeManifestHandler: public HeartbeatHandler { +public: + + explicit DescribeManifestHandler(bool isSecure) + : HeartbeatHandler(isSecure) { + } + + virtual void handleHeartbeat(const rapidjson::Document& root, struct mg_connection * conn) { + sendHeartbeatResponse("DESCRIBE", "manifest", "889345", conn); + } + + virtual void handleAcknowledge(const rapidjson::Document& root) { + verifyJsonHasAgentManifest(root); + } +}; +int main(int argc, char **argv) { + std::string key_dir, test_file_location, url; + url = "http://localhost:0/api/heartbeat";; + if (argc > 1) { + test_file_location = argv[1]; + if (argc > 2) { + url = "https://localhost:0/api/heartbeat";; + key_dir = argv[2]; + } + } + + bool isSecure = false; + if (url.find("https") != std::string::npos) { + isSecure = true; + } + + VerifyC2Describe harness(isSecure); + + harness.setKeyDir(key_dir); + + DescribeManifestHandler responder(isSecure); + + harness.setUrl(url, &responder); + + harness.run(test_file_location); +} diff --git a/extensions/http-curl/tests/C2JstackTest.cpp b/extensions/http-curl/tests/C2JstackTest.cpp index dd2ad27..241af1c 100644 --- a/extensions/http-curl/tests/C2JstackTest.cpp +++ b/extensions/http-curl/tests/C2JstackTest.cpp @@ -45,123 +45,60 @@ #include "CivetServer.h" #include <cstring> #include "protocols/RESTSender.h" +#include "HTTPIntegrationBase.h" +#include "HTTPHandlers.h" -void waitToVerifyProcessor() { - std::this_thread::sleep_for(std::chrono::seconds(10)); -} - - -class ConfigHandler : public CivetHandler { +class VerifyC2DescribeJstack : public VerifyC2Describe { public: - ConfigHandler() { - calls_ = 0; - } - bool handlePost(CivetServer *server, struct mg_connection *conn) { - calls_++; - std::string heartbeat_response = "{\"operation\" : \"heartbeat\",\"requested_operations\": [ {" - "\"operation\" : \"describe\", " - "\"operationid\" : \"8675309\", " - "\"name\": \"jstack\"" - "}]}"; - mg_printf(conn, "HTTP/1.1 200 OK\r\nContent-Type: " - "text/plain\r\nContent-Length: %lu\r\nConnection: close\r\n\r\n", - heartbeat_response.length()); - mg_printf(conn, "%s", heartbeat_response.c_str()); - - - return true; + explicit VerifyC2DescribeJstack(bool isSecure) + : VerifyC2Describe(isSecure) { } - bool handleGet(CivetServer *server, struct mg_connection *conn) { - std::ifstream myfile(test_file_location_.c_str()); - - if (myfile.is_open()) { - std::stringstream buffer; - buffer << myfile.rdbuf(); - std::string str = buffer.str(); - myfile.close(); - mg_printf(conn, "HTTP/1.1 200 OK\r\nContent-Type: " - "text/plain\r\nContent-Length: %lu\r\nConnection: close\r\n\r\n", - str.length()); - mg_printf(conn, "%s", str.c_str()); - } else { - mg_printf(conn, "HTTP/1.1 500 Internal Server Error\r\n"); - } - - return true; + virtual void runAssertions() { + assert(LogTestController::getInstance().contains("SchedulingAgent") == true); } - std::string test_file_location_; - std::atomic<size_t> calls_; }; -int main(int argc, char **argv) { - mg_init_library(0); - LogTestController::getInstance().setInfo<minifi::FlowController>(); - LogTestController::getInstance().setDebug<minifi::utils::HTTPClient>(); - LogTestController::getInstance().setDebug<minifi::c2::RESTSender>(); - LogTestController::getInstance().setTrace<minifi::c2::C2Agent>(); +class DescribeJstackHandler : public HeartbeatHandler { + public: + explicit DescribeJstackHandler(bool isSecure) + : HeartbeatHandler(isSecure) { + } - const char *options[] = { "document_root", ".", "listening_ports", "0", 0 }; - std::vector<std::string> cpp_options; - for (int i = 0; i < (sizeof(options) / sizeof(options[0]) - 1); i++) { - cpp_options.push_back(options[i]); + virtual void handleHeartbeat(const rapidjson::Document& root, struct mg_connection * conn) { + sendHeartbeatResponse("DESCRIBE", "jstack", "889398", conn); } - CivetServer server(cpp_options); + virtual void handleAcknowledge(const rapidjson::Document& root) { + assert(root.HasMember("SchedulingAgent #0") == true); + } - std::string port_str = std::to_string(server.getListeningPorts()[0]); +}; - ConfigHandler h_ex; - server.addHandler("/update", h_ex); - std::string key_dir, test_file_location; +int main(int argc, char **argv) { + std::string key_dir, test_file_location, url; + url = "http://localhost:0/api/heartbeat";; if (argc > 1) { - h_ex.test_file_location_ = test_file_location = argv[1]; - key_dir = argv[2]; + test_file_location = argv[1]; + if (argc > 2) { + url = "https://localhost:0/api/heartbeat";; + key_dir = argv[2]; + } } + bool isSecure = false; + if (url.find("https") != std::string::npos) { + isSecure = true; + } - std::shared_ptr<minifi::Configure> configuration = std::make_shared<minifi::Configure>(); - - std::string c2_rest_url = "http://localhost:"; + port_str + "/update"; - - configuration->set("c2.rest.url", c2_rest_url); - configuration->set("c2.agent.heartbeat.period", "1000"); - - std::shared_ptr<core::Repository> test_repo = std::make_shared<TestRepository>(); - std::shared_ptr<core::Repository> test_flow_repo = std::make_shared<TestFlowRepository>(); - - configuration->set(minifi::Configure::nifi_flow_configuration_file, test_file_location); - - std::shared_ptr<minifi::io::StreamFactory> stream_factory = minifi::io::StreamFactory::getInstance(configuration); - std::shared_ptr<core::ContentRepository> content_repo = std::make_shared<core::repository::VolatileContentRepository>(); - std::unique_ptr<core::FlowConfiguration> yaml_ptr = std::unique_ptr<core::YamlConfiguration>( - new core::YamlConfiguration(test_repo, test_repo, content_repo, stream_factory, configuration, test_file_location)); - std::shared_ptr<TestRepository> repo = std::static_pointer_cast<TestRepository>(test_repo); - - std::shared_ptr<minifi::FlowController> controller = std::make_shared<minifi::FlowController>(test_repo, test_flow_repo, configuration, std::move(yaml_ptr), content_repo, DEFAULT_ROOT_GROUP_NAME, - true); - - core::YamlConfiguration yaml_config(test_repo, test_repo, content_repo, stream_factory, configuration, test_file_location); + VerifyC2DescribeJstack harness(isSecure); - std::unique_ptr<core::ProcessGroup> ptr = yaml_config.getRoot(test_file_location); - std::shared_ptr<core::ProcessGroup> pg = std::shared_ptr<core::ProcessGroup>(ptr.get()); - ptr.release(); - auto start = std::chrono::system_clock::now(); + harness.setKeyDir(key_dir); - controller->load(); - controller->start(); - waitToVerifyProcessor(); + DescribeJstackHandler responder(isSecure); - controller->waitUnload(60000); - auto then = std::chrono::system_clock::now(); + harness.setUrl(url, &responder); - auto milliseconds = std::chrono::duration_cast<std::chrono::milliseconds>(then - start).count(); - std::string logs = LogTestController::getInstance().log_output.str(); - #ifndef WIN32 - assert(logs.find("SchedulingAgent") != std::string::npos); - #endif - LogTestController::getInstance().reset(); - assert(h_ex.calls_ <= (milliseconds / 1000) + 1); + harness.run(test_file_location); - return 0; } diff --git a/extensions/http-curl/tests/C2VerifyHeartbeatAndStop.cpp b/extensions/http-curl/tests/C2VerifyHeartbeatAndStop.cpp index 2eaa10b..6e0c1c2 100644 --- a/extensions/http-curl/tests/C2VerifyHeartbeatAndStop.cpp +++ b/extensions/http-curl/tests/C2VerifyHeartbeatAndStop.cpp @@ -50,98 +50,47 @@ #include "protocols/RESTReceiver.h" #include "protocols/RESTSender.h" #include "HTTPIntegrationBase.h" +#include "HTTPHandlers.h" #include "agent/build_description.h" #include "processors/LogAttribute.h" -class Responder : public CivetHandler { +class LightWeightC2Handler : public HeartbeatHandler { public: - explicit Responder(bool isSecure) - : isSecure(isSecure) { + explicit LightWeightC2Handler(bool isSecure) + : HeartbeatHandler(isSecure), + calls_(0) { } - std::string readPost(struct mg_connection *conn) { - std::string response; - int blockSize = 1024 * sizeof(char), readBytes; + virtual ~LightWeightC2Handler() = default; - char buffer[1024]; - while ((readBytes = mg_read(conn, buffer, blockSize)) > 0) { - response.append(buffer, 0, (readBytes / sizeof(char))); + virtual void handleHeartbeat(const rapidjson::Document& root, struct mg_connection * conn) { + (void)conn; + if (calls_ == 0) { + verifyJsonHasAgentManifest(root); + } else { + assert(root.HasMember("agentInfo") == true); + assert(root["agentInfo"].HasMember("agentManifest") == false); } - return response; + calls_++; } - bool handlePost(CivetServer *server, struct mg_connection *conn) { - auto post_data = readPost(conn); - - std::cerr << post_data << std::endl; - - if (!IsNullOrEmpty(post_data)) { - rapidjson::Document root; - rapidjson::ParseResult ok = root.Parse(post_data.data(), post_data.size()); - bool found = false; - std::string operation = root["operation"].GetString(); - if (operation == "heartbeat") { - assert(root.HasMember("agentInfo") == true); - assert(root["agentInfo"]["agentManifest"].HasMember("bundles") == true); - - for (auto &bundle : root["agentInfo"]["agentManifest"]["bundles"].GetArray()) { - assert(bundle.HasMember("artifact")); - std::string str = bundle["artifact"].GetString(); - if (str == "minifi-system") { - - std::vector<std::string> classes; - for (auto &proc : bundle["componentManifest"]["processors"].GetArray()) { - classes.push_back(proc["type"].GetString()); - } - - auto group = minifi::BuildDescription::getClassDescriptions(str); - for (auto proc : group.processors_) { - assert(std::find(classes.begin(), classes.end(), proc.class_name_) != std::end(classes)); - found = true; - } - - } - } - assert(found == true); - } - } - std::string resp = "{\"operation\" : \"heartbeat\", \"requested_operations\" : [{ \"operationid\" : 41, \"operation\" : \"stop\", \"name\" : \"invoke\" }, " - "{ \"operationid\" : 42, \"operation\" : \"stop\", \"name\" : \"FlowController\" } ]}"; - mg_printf(conn, "HTTP/1.1 200 OK\r\nContent-Type: " - "text/plain\r\nContent-Length: %lu\r\nConnection: close\r\n\r\n", - resp.length()); - mg_printf(conn, "%s", resp.c_str()); - return true; - } - - protected: - bool isSecure; + private: + std::atomic<size_t> calls_; }; -class VerifyC2Heartbeat : public CoapIntegrationBase { +class VerifyC2Heartbeat : public VerifyC2Base { public: explicit VerifyC2Heartbeat(bool isSecure) - : isSecure(isSecure) { - char format[] = "/tmp/ssth.XXXXXX"; - dir = testController.createTempDirectory(format); + : VerifyC2Base(isSecure) { } - void testSetup() { - LogTestController::getInstance().setDebug<utils::HTTPClient>(); + virtual ~VerifyC2Heartbeat() = default; + + virtual void testSetup() { LogTestController::getInstance().setTrace<minifi::c2::C2Agent>(); - LogTestController::getInstance().setDebug<LogTestController>(); LogTestController::getInstance().setDebug<minifi::c2::RESTSender>(); LogTestController::getInstance().setDebug<minifi::c2::RESTProtocol>(); LogTestController::getInstance().setDebug<minifi::c2::RESTReceiver>(); - std::fstream file; - ss << dir << "/" << "tstFile.ext"; - file.open(ss.str(), std::ios::out); - file << "tempFile"; - file.close(); - } - - void cleanup() { - LogTestController::getInstance().reset(); - unlink(ss.str().c_str()); + VerifyC2Base::testSetup(); } void runAssertions() { @@ -152,32 +101,22 @@ class VerifyC2Heartbeat : public CoapIntegrationBase { assert(LogTestController::getInstance().contains("C2Agent] [debug] Stopping component FlowController") == true); } - void queryRootProcessGroup(std::shared_ptr<core::ProcessGroup> pg) { - std::shared_ptr<core::Processor> proc = pg->findProcessor("invoke"); - assert(proc != nullptr); - - std::shared_ptr<minifi::processors::InvokeHTTP> inv = std::dynamic_pointer_cast<minifi::processors::InvokeHTTP>(proc); + void configureC2RootClasses() { + configuration->set("nifi.c2.root.classes", "DeviceInfoNode,AgentInformation,FlowInformation"); + } +}; - assert(inv != nullptr); - std::string url = ""; - inv->getProperty(minifi::processors::InvokeHTTP::URL.getName(), url); +class VerifyLightWeightC2Heartbeat : public VerifyC2Heartbeat { +public: + explicit VerifyLightWeightC2Heartbeat(bool isSecure) + : VerifyC2Heartbeat(isSecure) { + } - std::string c2_url = std::string("http") + (isSecure ? "s" : "") + "://localhost:" + getWebPort() + "/api/heartbeat"; + virtual ~VerifyLightWeightC2Heartbeat() = default; - configuration->set("nifi.c2.agent.protocol.class", "RESTSender"); - configuration->set("nifi.c2.enable", "true"); - configuration->set("nifi.c2.agent.class", "test"); - configuration->set("nifi.c2.rest.url", c2_url); - configuration->set("nifi.c2.agent.heartbeat.period", "1000"); - configuration->set("nifi.c2.rest.url.ack", c2_url); - configuration->set("nifi.c2.root.classes", "DeviceInfoNode,AgentInformation,FlowInformation"); + void configureC2RootClasses() { + configuration->set("nifi.c2.root.classes", "DeviceInfoNode,AgentInformationWithoutManifest,FlowInformation"); } - - protected: - bool isSecure; - std::string dir; - std::stringstream ss; - TestController testController; }; int main(int argc, char **argv) { @@ -195,12 +134,23 @@ int main(int argc, char **argv) { if (url.find("https") != std::string::npos) { isSecure = true; } + { + VerifyC2Heartbeat harness(isSecure); + + harness.setKeyDir(key_dir); + + HeartbeatHandler responder(isSecure); + + harness.setUrl(url, &responder); + + harness.run(test_file_location); + } - VerifyC2Heartbeat harness(isSecure); + VerifyLightWeightC2Heartbeat harness(isSecure); harness.setKeyDir(key_dir); - Responder responder(isSecure); + LightWeightC2Handler responder(isSecure); harness.setUrl(url, &responder); diff --git a/extensions/http-curl/tests/CMakeLists.txt b/extensions/http-curl/tests/CMakeLists.txt index ab9176d..ff848b5 100644 --- a/extensions/http-curl/tests/CMakeLists.txt +++ b/extensions/http-curl/tests/CMakeLists.txt @@ -72,6 +72,7 @@ add_test(NAME HTTPStreamingCallbackTests COMMAND "HTTPStreamingCallbackTests" WO add_test(NAME HttpGetIntegrationTest COMMAND HttpGetIntegrationTest "${TEST_RESOURCES}/TestHTTPGet.yml" "${TEST_RESOURCES}/") add_test(NAME C2UpdateTest COMMAND C2UpdateTest "${TEST_RESOURCES}/TestHTTPGet.yml" "${TEST_RESOURCES}/") add_test(NAME C2JstackTest COMMAND C2JstackTest "${TEST_RESOURCES}/TestHTTPGet.yml" "${TEST_RESOURCES}/") +add_test(NAME C2DescribeManifestTest COMMAND C2DescribeManifestTest "${TEST_RESOURCES}/TestHTTPGet.yml" "${TEST_RESOURCES}/") add_test(NAME C2UpdateAgentTest COMMAND C2UpdateAgentTest "${TEST_RESOURCES}/TestHTTPGet.yml" "${TEST_RESOURCES}/") add_test(NAME C2FailedUpdateTest COMMAND C2FailedUpdateTest "${TEST_RESOURCES}/TestHTTPGet.yml" "${TEST_RESOURCES}/TestBad.yml" "${TEST_RESOURCES}/") add_test(NAME C2NullConfiguration COMMAND C2NullConfiguration "${TEST_RESOURCES}/TestNull.yml" "${TEST_RESOURCES}/") diff --git a/extensions/http-curl/tests/HTTPHandlers.h b/extensions/http-curl/tests/HTTPHandlers.h index 282b470..1de39b3 100644 --- a/extensions/http-curl/tests/HTTPHandlers.h +++ b/extensions/http-curl/tests/HTTPHandlers.h @@ -20,6 +20,8 @@ #include "concurrentqueue.h" #include "CivetStream.h" #include "io/CRCStream.h" +#include "rapidjson/document.h" + #ifndef LIBMINIFI_TEST_CURL_TESTS_SITETOSITEHTTP_HTTPHANDLERS_H_ #define LIBMINIFI_TEST_CURL_TESTS_SITETOSITEHTTP_HTTPHANDLERS_H_ static std::atomic<int> transaction_id; @@ -343,4 +345,104 @@ class DeleteTransactionResponder : public CivetHandler { std::string response_code; }; +class HeartbeatHandler : public CivetHandler { + public: + explicit HeartbeatHandler(bool isSecure) + : isSecure(isSecure) { + } + + std::string readPost(struct mg_connection *conn) { + std::string response; + int blockSize = 1024 * sizeof(char), readBytes; + + char buffer[1024]; + while ((readBytes = mg_read(conn, buffer, blockSize)) > 0) { + response.append(buffer, 0, (readBytes / sizeof(char))); + } + return response; + } + + void sendStopOperation(struct mg_connection *conn) { + std::string resp = "{\"operation\" : \"heartbeat\", \"requested_operations\" : [{ \"operationid\" : 41, \"operation\" : \"stop\", \"name\" : \"invoke\" }, " + "{ \"operationid\" : 42, \"operation\" : \"stop\", \"name\" : \"FlowController\" } ]}"; + mg_printf(conn, "HTTP/1.1 200 OK\r\nContent-Type: " + "text/plain\r\nContent-Length: %lu\r\nConnection: close\r\n\r\n", + resp.length()); + mg_printf(conn, "%s", resp.c_str()); + } + + void sendHeartbeatResponse(const std::string& operation, const std::string& operand, const std::string& operationId, struct mg_connection * conn) { + std::string heartbeat_response = "{\"operation\" : \"heartbeat\",\"requested_operations\": [ {" + "\"operation\" : \"" + operation + "\"," + "\"operationid\" : \"" + operationId + "\"," + "\"operand\": \"" + operand + "\"}]}"; + + mg_printf(conn, "HTTP/1.1 200 OK\r\nContent-Type: " + "text/plain\r\nContent-Length: %lu\r\nConnection: close\r\n\r\n", + heartbeat_response.length()); + mg_printf(conn, "%s", heartbeat_response.c_str()); + } + + void verifyJsonHasAgentManifest(const rapidjson::Document& root) { + bool found = false; + assert(root.HasMember("agentInfo") == true); + assert(root["agentInfo"].HasMember("agentManifest") == true); + assert(root["agentInfo"]["agentManifest"].HasMember("bundles") == true); + + for (auto &bundle : root["agentInfo"]["agentManifest"]["bundles"].GetArray()) { + assert(bundle.HasMember("artifact")); + std::string str = bundle["artifact"].GetString(); + if (str == "minifi-system") { + + std::vector<std::string> classes; + for (auto &proc : bundle["componentManifest"]["processors"].GetArray()) { + classes.push_back(proc["type"].GetString()); + } + + auto group = minifi::BuildDescription::getClassDescriptions(str); + for (auto proc : group.processors_) { + assert(std::find(classes.begin(), classes.end(), proc.class_name_) != std::end(classes)); + found = true; + } + + } + } + assert(found == true); + } + + virtual void handleHeartbeat(const rapidjson::Document& root, struct mg_connection * conn) { + (void)conn; + verifyJsonHasAgentManifest(root); + } + + virtual void handleAcknowledge(const rapidjson::Document& root) { + } + + void verify(struct mg_connection *conn) { + auto post_data = readPost(conn); + std::cerr << post_data << std::endl; + if (!IsNullOrEmpty(post_data)) { + rapidjson::Document root; + rapidjson::ParseResult ok = root.Parse(post_data.data(), post_data.size()); + std::string operation = root["operation"].GetString(); + if (operation == "heartbeat") { + handleHeartbeat(root, conn); + } else if (operation == "acknowledge") { + handleAcknowledge(root); + } else { + throw std::runtime_error("operation not supported " + operation); + } + } + } + + bool handlePost(CivetServer *server, struct mg_connection *conn) { + verify(conn); + sendStopOperation(conn); + return true; + } + + protected: + bool isSecure; +}; + #endif /* LIBMINIFI_TEST_CURL_TESTS_SITETOSITEHTTP_HTTPHANDLERS_H_ */ diff --git a/extensions/http-curl/tests/HTTPIntegrationBase.h b/extensions/http-curl/tests/HTTPIntegrationBase.h index 4ab623b..4226630 100644 --- a/extensions/http-curl/tests/HTTPIntegrationBase.h +++ b/extensions/http-curl/tests/HTTPIntegrationBase.h @@ -21,6 +21,8 @@ #include "../tests/TestServer.h" #include "CivetServer.h" #include "integration/IntegrationBase.h" +#include "c2/C2Agent.h" +#include "protocols/RESTSender.h" int log_message(const struct mg_connection *conn, const char *message) { puts(message); @@ -41,8 +43,6 @@ class CoapIntegrationBase : public IntegrationBase { void setUrl(std::string url, CivetHandler *handler); - virtual ~CoapIntegrationBase(); - void shutdownBeforeFlowController() { stop_webserver(server); } @@ -59,10 +59,6 @@ class CoapIntegrationBase : public IntegrationBase { CivetServer *server; }; -CoapIntegrationBase::~CoapIntegrationBase() { - -} - void CoapIntegrationBase::setUrl(std::string url, CivetHandler *handler) { parse_http_components(url, port, scheme, path); @@ -91,4 +87,74 @@ void CoapIntegrationBase::setUrl(std::string url, CivetHandler *handler) { } } +class VerifyC2Base : public CoapIntegrationBase { + public: + explicit VerifyC2Base(bool isSecure) + : isSecure(isSecure) { + char format[] = "/tmp/ssth.XXXXXX"; + dir = testController.createTempDirectory(format); + } + + virtual void testSetup() { + LogTestController::getInstance().setDebug<utils::HTTPClient>(); + LogTestController::getInstance().setDebug<LogTestController>(); + std::fstream file; + ss << dir << "/" << "tstFile.ext"; + file.open(ss.str(), std::ios::out); + file << "tempFile"; + file.close(); + } + + void runAssertions() { + } + + virtual void queryRootProcessGroup(std::shared_ptr<core::ProcessGroup> pg) { + std::shared_ptr<core::Processor> proc = pg->findProcessor("invoke"); + assert(proc != nullptr); + + std::shared_ptr<minifi::processors::InvokeHTTP> inv = std::dynamic_pointer_cast<minifi::processors::InvokeHTTP>(proc); + + assert(inv != nullptr); + std::string url = ""; + inv->getProperty(minifi::processors::InvokeHTTP::URL.getName(), url); + + std::string c2_url = std::string("http") + (isSecure ? "s" : "") + "://localhost:" + getWebPort() + "/api/heartbeat"; + + configuration->set("nifi.c2.agent.protocol.class", "RESTSender"); + configuration->set("nifi.c2.enable", "true"); + configuration->set("nifi.c2.agent.class", "test"); + configuration->set("nifi.c2.rest.url", c2_url); + configuration->set("nifi.c2.agent.heartbeat.period", "1000"); + configuration->set("nifi.c2.rest.url.ack", c2_url); + } + + void cleanup() { + LogTestController::getInstance().reset(); + unlink(ss.str().c_str()); + } + + protected: + bool isSecure; + std::string dir; + std::stringstream ss; + TestController testController; +}; + +class VerifyC2Describe : public VerifyC2Base { + public: + explicit VerifyC2Describe(bool isSecure) + : VerifyC2Base(isSecure) { + } + + void testSetup() { + LogTestController::getInstance().setTrace<minifi::c2::C2Agent>(); + LogTestController::getInstance().setDebug<minifi::c2::RESTSender>(); + LogTestController::getInstance().setInfo<minifi::FlowController>(); + VerifyC2Base::testSetup(); + } + + void configureC2RootClasses() { + configuration->set("nifi.c2.root.classes", "DeviceInfoNode,AgentInformationWithoutManifest,FlowInformation"); + } +}; #endif /* LIBMINIFI_TEST_INTEGRATION_HTTPINTEGRATIONBASE_H_ */ diff --git a/libminifi/include/FlowController.h b/libminifi/include/FlowController.h index 5602452..bef607d 100644 --- a/libminifi/include/FlowController.h +++ b/libminifi/include/FlowController.h @@ -303,6 +303,19 @@ class FlowController : public core::controller::ControllerServiceProvider, publi */ virtual int16_t getMetricsNodes(std::vector<std::shared_ptr<state::response::ResponseNode>> &metric_vector, uint16_t metricsClass); + /** + * Retrieves agent information with manifest only from this source. + * @param manifest_vector -- manifest nodes vector. + * @return 0 on Success, -1 on failure + */ + virtual int16_t getManifestNodes(std::vector<std::shared_ptr<state::response::ResponseNode>>& manifest_vector) const; + + /** + * Returns a response node containing all agent information with manifest and agent status + * @return a shared pointer to agent information + */ + virtual std::shared_ptr<state::response::ResponseNode> getAgentInformation() const; + virtual uint64_t getUptime(); virtual std::vector<BackTrace> getTraces(); @@ -384,9 +397,10 @@ class FlowController : public core::controller::ControllerServiceProvider, publi std::chrono::steady_clock::time_point start_time_; - std::mutex metrics_mutex_; + mutable std::mutex metrics_mutex_; // root_nodes cache std::map<std::string, std::shared_ptr<state::response::ResponseNode>> root_response_nodes_; + // metrics cache std::map<std::string, std::shared_ptr<state::response::ResponseNode>> device_information_; @@ -394,6 +408,10 @@ class FlowController : public core::controller::ControllerServiceProvider, publi std::map<std::string, std::shared_ptr<state::response::ResponseNode>> component_metrics_; std::map<uint8_t, std::vector<std::shared_ptr<state::response::ResponseNode>>> component_metrics_by_id_; + + // manifest cache + std::map<std::string, std::shared_ptr<state::response::ResponseNode>> agent_information_; + // metrics last run std::chrono::steady_clock::time_point last_metrics_capture_; diff --git a/libminifi/include/c2/C2Agent.h b/libminifi/include/c2/C2Agent.h index f84c131..c39589f 100644 --- a/libminifi/include/c2/C2Agent.h +++ b/libminifi/include/c2/C2Agent.h @@ -171,6 +171,11 @@ class C2Agent : public state::UpdateController, public state::response::Response */ bool update_property(const std::string &property_name, const std::string &property_value, bool persist = false); + /** + * Creates configuration options C2 payload for response + */ + C2Payload prepareConfigurationOptions(const C2ContentResponse &resp) const; + std::timed_mutex metrics_mutex_; std::map<std::string, std::shared_ptr<state::response::ResponseNode>> metrics_map_; @@ -244,6 +249,8 @@ class C2Agent : public state::UpdateController, public state::response::Response std::string bin_location_; std::shared_ptr<logging::Logger> logger_; + + bool manifest_sent_; }; } /* namesapce c2 */ diff --git a/libminifi/include/core/state/nodes/AgentInformation.h b/libminifi/include/core/state/nodes/AgentInformation.h index abf0c6f..5809a4c 100644 --- a/libminifi/include/core/state/nodes/AgentInformation.h +++ b/libminifi/include/core/state/nodes/AgentInformation.h @@ -68,7 +68,7 @@ namespace response { class ComponentManifest : public DeviceInformation { public: - ComponentManifest(std::string name, utils::Identifier & uuid) + ComponentManifest(const std::string& name, utils::Identifier & uuid) : DeviceInformation(name, uuid) { } @@ -92,7 +92,7 @@ class ComponentManifest : public DeviceInformation { } protected: - void serializeClassDescription(const std::vector<ClassDescription> &descriptions, const std::string name, SerializedResponseNode &response) { + void serializeClassDescription(const std::vector<ClassDescription> &descriptions, const std::string name, SerializedResponseNode &response) const { if (!descriptions.empty()) { SerializedResponseNode type; type.name = name; @@ -307,7 +307,7 @@ class ComponentManifest : public DeviceInformation { class ExternalManifest : public ComponentManifest { public: - ExternalManifest(std::string name, utils::Identifier & uuid) + ExternalManifest(const std::string& name, utils::Identifier & uuid) : ComponentManifest(name, uuid) { } @@ -329,7 +329,7 @@ class ExternalManifest : public ComponentManifest { class Bundles : public DeviceInformation { public: - Bundles(std::string name, utils::Identifier & uuid) + Bundles(const std::string& name, utils::Identifier & uuid) : DeviceInformation(name, uuid) { setArray(true); } @@ -452,8 +452,17 @@ class AgentStatus : public StateMonitorNode { queuesize.name = "size"; queuesize.value = repo.second->getRepoSize(); - repoNode.children.push_back(queuesize); + SerializedResponseNode isRunning; + isRunning.name = "running"; + isRunning.value = repo.second->isRunning(); + + SerializedResponseNode isFull; + isFull.name = "full"; + isFull.value = repo.second->isFull(); + repoNode.children.push_back(queuesize); + repoNode.children.push_back(isRunning); + repoNode.children.push_back(isFull); repositories.children.push_back(repoNode); } @@ -540,14 +549,12 @@ class AgentMonitor { class AgentManifest : public DeviceInformation { public: - AgentManifest(std::string name, utils::Identifier & uuid) + AgentManifest(const std::string& name, utils::Identifier & uuid) : DeviceInformation(name, uuid) { - //setArray(true); } AgentManifest(const std::string &name) : DeviceInformation(name) { - // setArray(true); } std::string getName() const { @@ -622,25 +629,21 @@ class AgentManifest : public DeviceInformation { } }; -/** - * Purpose and Justification: Prints classes along with their properties for the current agent. - */ -class AgentInformation : public DeviceInformation, public AgentMonitor, public AgentIdentifier { - public: - AgentInformation(std::string name, utils::Identifier & uuid) +class AgentNode : public DeviceInformation, public AgentMonitor, public AgentIdentifier { +public: + + AgentNode(const std::string& name, utils::Identifier & uuid) : DeviceInformation(name, uuid) { setArray(false); } - AgentInformation(const std::string &name) + explicit AgentNode(const std::string& name) : DeviceInformation(name) { setArray(false); } - std::string getName() const { - return "agentInfo"; - } +protected: std::vector<SerializedResponseNode> serialize() { std::vector<SerializedResponseNode> serialized; @@ -654,6 +657,14 @@ class AgentInformation : public DeviceInformation, public AgentMonitor, public A agentClass.name = "agentClass"; agentClass.value = agent_class_; + serialized.push_back(ident); + serialized.push_back(agentClass); + + return serialized; + } + + std::vector<SerializedResponseNode> getAgentManifest() const { + std::vector<SerializedResponseNode> serialized; AgentManifest manifest("manifest"); SerializedResponseNode agentManifest; @@ -661,6 +672,34 @@ class AgentInformation : public DeviceInformation, public AgentMonitor, public A for (auto &ser : manifest.serialize()) { agentManifest.children.push_back(std::move(ser)); } + serialized.push_back(agentManifest); + return serialized; + } +}; + +/** + * This class is used for regular heartbeat without manifest + * A light weight heartbeat + */ +class AgentInformationWithoutManifest : public AgentNode { +public: + + AgentInformationWithoutManifest(const std::string& name, utils::Identifier & uuid) + : AgentNode(name, uuid) { + setArray(false); + } + + explicit AgentInformationWithoutManifest(const std::string &name) + : AgentNode(name) { + setArray(false); + } + + std::string getName() const { + return "agentInfo"; + } + + std::vector<SerializedResponseNode> serialize() { + std::vector<SerializedResponseNode> serialized(AgentNode::serialize()); AgentStatus status("status"); status.setRepositories(repositories_); @@ -672,16 +711,72 @@ class AgentInformation : public DeviceInformation, public AgentMonitor, public A agentStatus.children.push_back(std::move(ser)); } - serialized.push_back(ident); - serialized.push_back(agentClass); - serialized.push_back(agentManifest); serialized.push_back(agentStatus); return serialized; } +}; + + +/** + * This class is used for sending all agent information including manifest and status + * A heavy weight heartbeat. Here to maintain backward compatibility + */ +class AgentInformation : public AgentInformationWithoutManifest { + public: + + AgentInformation(const std::string& name, utils::Identifier & uuid) + : AgentInformationWithoutManifest(name, uuid) { + setArray(false); + } + + explicit AgentInformation(const std::string &name) + : AgentInformationWithoutManifest(name) { + setArray(false); + } + + std::string getName() const { + return "agentInfo"; + } + + std::vector<SerializedResponseNode> serialize() { + std::vector<SerializedResponseNode> serialized(AgentInformationWithoutManifest::serialize()); + auto manifest = getAgentManifest(); + serialized.insert(serialized.end(), std::make_move_iterator(manifest.begin()), std::make_move_iterator(manifest.end())); + return serialized; + } + +}; + +/** + * This class is used for response to DESCRIBE manifest request + * It contains static information only + */ +class AgentInformationWithManifest : public AgentNode { +public: + AgentInformationWithManifest(const std::string& name, utils::Identifier & uuid) + : AgentNode(name, uuid) { + setArray(false); + } + explicit AgentInformationWithManifest(const std::string &name) + : AgentNode(name) { + setArray(false); + } + + std::string getName() const { + return "agentInfo"; + } + + std::vector<SerializedResponseNode> serialize() { + std::vector<SerializedResponseNode> serialized(AgentNode::serialize()); + auto manifest = getAgentManifest(); + serialized.insert(serialized.end(), std::make_move_iterator(manifest.begin()), std::make_move_iterator(manifest.end())); + return serialized; + } }; REGISTER_RESOURCE(AgentInformation, "Node part of an AST that defines all agent information, to include the manifest, and bundle information as part of a healthy hearbeat."); +REGISTER_RESOURCE(AgentInformationWithoutManifest, "Node part of an AST that defines all agent information, without the manifest and bundle information as part of a healthy hearbeat."); } /* namespace metrics */ } /* namespace state */ diff --git a/libminifi/include/core/state/nodes/MetricsBase.h b/libminifi/include/core/state/nodes/MetricsBase.h index 40a0c0c..d91df6e 100644 --- a/libminifi/include/core/state/nodes/MetricsBase.h +++ b/libminifi/include/core/state/nodes/MetricsBase.h @@ -43,12 +43,12 @@ class ResponseNode : public core::Connectable { is_array_(false) { } - ResponseNode(std::string name) + ResponseNode(const std::string& name) : core::Connectable(name), is_array_(false) { } - ResponseNode(std::string name, utils::Identifier & uuid) + ResponseNode(const std::string& name, utils::Identifier & uuid) : core::Connectable(name, uuid), is_array_(false) { } @@ -90,10 +90,10 @@ class ResponseNode : public core::Connectable { */ class DeviceInformation : public ResponseNode { public: - DeviceInformation(std::string name, utils::Identifier & uuid) + DeviceInformation(const std::string& name, utils::Identifier & uuid) : ResponseNode(name, uuid) { } - DeviceInformation(std::string name) + DeviceInformation(const std::string& name) : ResponseNode(name) { } }; @@ -228,6 +228,18 @@ class NodeReporter { */ virtual int16_t getMetricsNodes(std::vector<std::shared_ptr<ResponseNode>> &metric_vector, uint16_t metricsClass) = 0; + /** + * Retrieves agent information with manifest only from this source. + * @param manifest_vector -- manifest nodes vector. + * @return 0 on Success, -1 on failure + */ + virtual int16_t getManifestNodes(std::vector<std::shared_ptr<state::response::ResponseNode>>& manifest_vector) const = 0; + + /** + * Returns a response node containing all agent information with manifest and agent status + * @return a shared pointer to agent information + */ + virtual std::shared_ptr<state::response::ResponseNode> getAgentInformation() const = 0; }; /** diff --git a/libminifi/include/properties/Properties.h b/libminifi/include/properties/Properties.h index 2a855ba..648d86f 100644 --- a/libminifi/include/properties/Properties.h +++ b/libminifi/include/properties/Properties.h @@ -96,7 +96,7 @@ class Properties { minifi_home_ = minifiHome; } - std::vector<std::string> getConfiguredKeys() { + std::vector<std::string> getConfiguredKeys() const { std::vector<std::string> keys; for (auto &property : properties_) { keys.push_back(property.first); diff --git a/libminifi/src/FlowController.cpp b/libminifi/src/FlowController.cpp index 31160c8..1a251b5 100644 --- a/libminifi/src/FlowController.cpp +++ b/libminifi/src/FlowController.cpp @@ -466,6 +466,7 @@ void FlowController::initializeC2() { device_information_.clear(); component_metrics_.clear(); component_metrics_by_id_.clear(); + agent_information_.clear(); std::string class_csv; if (root_ != nullptr) { @@ -484,6 +485,15 @@ void FlowController::initializeC2() { repoMetrics->addRepository(flow_file_repo_); device_information_[repoMetrics->getName()] = repoMetrics; + + std::shared_ptr<state::response::AgentInformationWithManifest> manifest = std::make_shared<state::response::AgentInformationWithManifest>("agentInformation"); + auto identifier = std::dynamic_pointer_cast<state::response::AgentIdentifier>(manifest); + + if (identifier != nullptr) { + identifier->setIdentifier(identifier_str); + identifier->setAgentClass(class_str); + agent_information_[manifest->getName()] = manifest; + } } if (configuration_->get("nifi.c2.root.classes", class_csv)) { @@ -932,6 +942,34 @@ int16_t FlowController::getMetricsNodes(std::vector<std::shared_ptr<state::respo return 0; } +int16_t FlowController::getManifestNodes(std::vector<std::shared_ptr<state::response::ResponseNode>>& manifest_vector) const { + std::lock_guard<std::mutex> lock(metrics_mutex_); + for (const auto& metric : agent_information_) { + manifest_vector.push_back(metric.second); + } + return 0; +} + +std::shared_ptr<state::response::ResponseNode> FlowController::getAgentInformation() const { + auto agentInfo = std::make_shared<state::response::AgentInformation>("agentInfo"); + auto identifier = std::dynamic_pointer_cast<state::response::AgentIdentifier>(agentInfo); + + if (identifier != nullptr) { + std::string class_str; + configuration_->get("nifi.c2.agent.class", "c2.agent.class", class_str); + + std::string identifier_str; + if (!configuration_->get("nifi.c2.agent.identifier", "c2.agent.identifier", identifier_str) || identifier_str.empty()) { + identifier_str = uuidStr_; + } + + identifier->setIdentifier(identifier_str); + identifier->setAgentClass(class_str); + return agentInfo; + } + return nullptr; +} + std::vector<std::shared_ptr<state::StateController>> FlowController::getAllComponents() { std::vector<std::shared_ptr<state::StateController>> vec; vec.push_back(shared_from_this()); diff --git a/libminifi/src/c2/C2Agent.cpp b/libminifi/src/c2/C2Agent.cpp index 7c4b6c2..77d3f7a 100644 --- a/libminifi/src/c2/C2Agent.cpp +++ b/libminifi/src/c2/C2Agent.cpp @@ -50,6 +50,8 @@ C2Agent::C2Agent(const std::shared_ptr<core::controller::ControllerServiceProvid logger_(logging::LoggerFactory<C2Agent>::getLogger()) { allow_updates_ = true; + manifest_sent_ = false; + running_c2_configuration = std::make_shared<Configure>(); last_run_ = std::chrono::steady_clock::now(); @@ -299,32 +301,34 @@ void C2Agent::performHeartBeat() { payload.addPayload(std::move(metrics)); } - if (device_information_.size() > 0) { - C2Payload deviceInfo(Operation::HEARTBEAT); - deviceInfo.setLabel("AgentInformation"); - - for (auto metric : device_information_) { - C2Payload child_metric_payload(Operation::HEARTBEAT); - child_metric_payload.setLabel(metric.first); - if (metric.second->isArray()) { - child_metric_payload.setContainer(true); - } - serializeMetrics(child_metric_payload, metric.first, metric.second->serialize(), metric.second->isArray()); - deviceInfo.addPayload(std::move(child_metric_payload)); + for (auto metric : root_response_nodes_) { + C2Payload child_metric_payload(Operation::HEARTBEAT); + bool isArray{false}; + std::string metricName; + std::vector<state::response::SerializedResponseNode> metrics; + std::shared_ptr<state::response::NodeReporter> reporter; + std::shared_ptr<state::response::ResponseNode> agentInfo; + + // Send agent manifest in first heartbeat + if (!manifest_sent_ + && (reporter = std::dynamic_pointer_cast<state::response::NodeReporter>(update_sink_)) + && (agentInfo = reporter->getAgentInformation()) + && metric.first == agentInfo->getName()) { + metricName = agentInfo->getName(); + isArray = agentInfo->isArray(); + metrics = agentInfo->serialize(); + manifest_sent_ = true; + } else { + metricName = metric.first; + isArray = metric.second->isArray(); + metrics = metric.second->serialize(); } - payload.addPayload(std::move(deviceInfo)); - } - - if (!root_response_nodes_.empty()) { - for (auto metric : root_response_nodes_) { - C2Payload child_metric_payload(Operation::HEARTBEAT); - child_metric_payload.setLabel(metric.first); - if (metric.second->isArray()) { - child_metric_payload.setContainer(true); - } - serializeMetrics(child_metric_payload, metric.first, metric.second->serialize(), metric.second->isArray()); - payload.addPayload(std::move(child_metric_payload)); + child_metric_payload.setLabel(metricName); + if (isArray) { + child_metric_payload.setContainer(true); } + serializeMetrics(child_metric_payload, metricName, metrics, isArray); + payload.addPayload(std::move(child_metric_payload)); } C2Payload && response = protocol_.load()->consumePayload(payload); @@ -485,6 +489,28 @@ void C2Agent::handle_c2_server_response(const C2ContentResponse &resp) { } } +C2Payload C2Agent::prepareConfigurationOptions(const C2ContentResponse &resp) const { + auto unsanitized_keys = configuration_->getConfiguredKeys(); + std::vector<std::string> keys; + std::copy_if(unsanitized_keys.begin(), unsanitized_keys.end(), std::back_inserter(keys), + [](std::string key) {return key.find("pass") == std::string::npos;}); + + C2Payload response(Operation::ACKNOWLEDGE, resp.ident, false, true); + C2Payload options(Operation::ACKNOWLEDGE); + options.setLabel("configuration_options"); + std::string value; + for (auto key : keys) { + C2ContentResponse option(Operation::ACKNOWLEDGE); + option.name = key; + if (configuration_->get(key, value)) { + option.operation_arguments[key] = value; + options.addContent(std::move(option)); + } + } + response.addPayload(std::move(options)); + return response; +} + /** * Descriptions are special types of requests that require information * to be put into the acknowledgement @@ -506,67 +532,36 @@ void C2Agent::handle_describe(const C2ContentResponse &resp) { } std::vector<std::shared_ptr<state::response::ResponseNode>> metrics_vec; - - reporter->getResponseNodes(metrics_vec, metric_class_id); C2Payload response(Operation::ACKNOWLEDGE, resp.ident, false, true); - response.setLabel("metrics"); + C2Payload metrics(Operation::ACKNOWLEDGE); + metrics.setLabel("metrics"); + reporter->getResponseNodes(metrics_vec, 0); for (auto metric : metrics_vec) { - serializeMetrics(response, metric->getName(), metric->serialize()); + serializeMetrics(metrics, metric->getName(), metric->serialize()); } + response.addPayload(std::move(metrics)); enqueue_c2_response(std::move(response)); } } else if (resp.name == "configuration") { - auto unsanitized_keys = configuration_->getConfiguredKeys(); - std::vector<std::string> keys; - std::copy_if(unsanitized_keys.begin(), unsanitized_keys.end(), std::back_inserter(keys), [](std::string key) {return key.find("pass") == std::string::npos;}); - C2Payload response(Operation::ACKNOWLEDGE, resp.ident, false, true); - response.setLabel("configuration_options"); - C2Payload options(Operation::ACKNOWLEDGE, resp.ident, false, true); - options.setLabel("configuration_options"); - std::string value; - for (auto key : keys) { - C2ContentResponse option(Operation::ACKNOWLEDGE); - option.name = key; - if (configuration_->get(key, value)) { - option.operation_arguments[key] = value; - options.addContent(std::move(option)); - } - } - response.addPayload(std::move(options)); - enqueue_c2_response(std::move(response)); + auto configOptions = prepareConfigurationOptions(resp); + enqueue_c2_response(std::move(configOptions)); return; } else if (resp.name == "manifest") { - auto keys = configuration_->getConfiguredKeys(); - C2Payload response(Operation::ACKNOWLEDGE, resp.ident, false, true); - response.setLabel("configuration_options"); - C2Payload options(Operation::ACKNOWLEDGE, resp.ident, false, true); - options.setLabel("configuration_options"); - std::string value; - for (auto key : keys) { - C2ContentResponse option(Operation::ACKNOWLEDGE); - option.name = key; - if (configuration_->get(key, value)) { - option.operation_arguments[key] = value; - options.addContent(std::move(option)); - } - } - response.addPayload(std::move(options)); + C2Payload response(prepareConfigurationOptions(resp)); + + auto reporter = std::dynamic_pointer_cast<state::response::NodeReporter>(update_sink_); + if (reporter != nullptr) { + std::vector<std::shared_ptr<state::response::ResponseNode>> metrics_vec; - if (device_information_.size() > 0) { - C2Payload deviceInfo(Operation::HEARTBEAT); - deviceInfo.setLabel("AgentInformation"); + C2Payload agentInfo(Operation::ACKNOWLEDGE, resp.ident, false, true); + agentInfo.setLabel("agentInfo"); - for (auto metric : device_information_) { - C2Payload child_metric_payload(Operation::HEARTBEAT); - child_metric_payload.setLabel(metric.first); - if (metric.second->isArray()) { - child_metric_payload.setContainer(true); - } - serializeMetrics(child_metric_payload, metric.first, metric.second->serialize(), metric.second->isArray()); - deviceInfo.addPayload(std::move(child_metric_payload)); + reporter->getManifestNodes(metrics_vec); + for (const auto& metric : metrics_vec) { + serializeMetrics(agentInfo, metric->getName(), metric->serialize()); } - response.addPayload(std::move(deviceInfo)); + response.addPayload(std::move(agentInfo)); } enqueue_c2_response(std::move(response)); @@ -581,7 +576,6 @@ void C2Agent::handle_describe(const C2ContentResponse &resp) { } auto keys = configuration_->getConfiguredKeys(); C2Payload response(Operation::ACKNOWLEDGE, resp.ident, false, true); - response.setLabel("configuration_options"); for (const auto &trace : traces) { C2Payload options(Operation::ACKNOWLEDGE, resp.ident, false, true); options.setLabel(trace.getName()); @@ -596,6 +590,7 @@ void C2Agent::handle_describe(const C2ContentResponse &resp) { } enqueue_c2_response(std::move(response)); } + return; } C2Payload response(Operation::ACKNOWLEDGE, resp.ident, false, true); enqueue_c2_response(std::move(response)); diff --git a/libminifi/test/integration/IntegrationBase.h b/libminifi/test/integration/IntegrationBase.h index 0cc4cc6..30f771a 100644 --- a/libminifi/test/integration/IntegrationBase.h +++ b/libminifi/test/integration/IntegrationBase.h @@ -35,7 +35,7 @@ class IntegrationBase { public: IntegrationBase(uint64_t waitTime = DEFAULT_WAITTIME_MSECS); - virtual ~IntegrationBase(); + virtual ~IntegrationBase() = default; virtual void run(std::string test_file_location); @@ -75,6 +75,10 @@ class IntegrationBase { } + virtual void configureC2RootClasses() { + + } + virtual void updateProperties(std::shared_ptr<minifi::FlowController> fc) { } @@ -92,9 +96,6 @@ IntegrationBase::IntegrationBase(uint64_t waitTime) wait_time_(waitTime) { } -IntegrationBase::~IntegrationBase() { -} - void IntegrationBase::configureSecurity() { if (!key_dir.empty()) { configuration->set(minifi::Configure::nifi_security_client_certificate, key_dir + "cn.crt.pem"); @@ -126,6 +127,8 @@ void IntegrationBase::run(std::string test_file_location) { queryRootProcessGroup(pg); + configureC2RootClasses(); + ptr.release(); std::shared_ptr<TestRepository> repo = std::static_pointer_cast<TestRepository>(test_repo);
