MINIFICPP-418: Add build time information, ability to deploy, and run time information to c2 response and build output MINIFICPP-395: Add transfer capability that runs a rollback command on failure MINIFICPP-417: Resolve issue with RapidJson changes.
MINIFICPP-417: Resolve linter issues MINIFICPP-418: Update package name MINIFICPP-418: Add flow URI introspection and update flow version MINIFICPP-468: Add configurable agent information and update readme This closes #314. Signed-off-by: Aldrin Piri <[email protected]> Project: http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/repo Commit: http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/commit/a330c57a Tree: http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/tree/a330c57a Diff: http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/diff/a330c57a Branch: refs/heads/master Commit: a330c57a59fc3dd1c12948f8aabf1854463bee8f Parents: eb9128c Author: Marc Parisi <[email protected]> Authored: Sat Jan 27 21:20:10 2018 -0500 Committer: Aldrin Piri <[email protected]> Committed: Fri May 4 09:19:55 2018 -0400 ---------------------------------------------------------------------- .gitignore | 1 + C2.md | 129 ++++ CMakeLists.txt | 26 +- README.md | 29 +- bin/minifi.sh | 32 + bootstrap.sh | 105 ++-- centos.sh | 2 + controller/Controller.h | 56 ++ controller/MiNiFiController.cpp | 11 +- darwin.sh | 4 +- deploy.sh | 27 + extensions/http-curl/HTTPCurlLoader.h | 1 - extensions/http-curl/client/HTTPClient.cpp | 7 +- extensions/http-curl/protocols/AgentPrinter.cpp | 141 +++++ extensions/http-curl/protocols/AgentPrinter.h | 80 +++ extensions/http-curl/protocols/RESTSender.cpp | 34 +- .../http-curl/tests/C2FailedUpdateTest.cpp | 186 ++++++ .../http-curl/tests/C2UpdateAgentTest.cpp | 184 ++++++ extensions/http-curl/tests/C2UpdateTest.cpp | 6 +- extensions/http-curl/tests/CMakeLists.txt | 2 + generateVersion.sh | 83 +++ libminifi/include/FlowController.h | 47 +- libminifi/include/agent/build_description.h | 101 ++++ libminifi/include/c2/C2Agent.h | 37 +- libminifi/include/c2/C2Payload.h | 67 ++- libminifi/include/c2/protocols/RESTProtocol.h | 12 +- libminifi/include/core/ClassLoader.h | 22 + libminifi/include/core/ConfigurableComponent.h | 17 +- libminifi/include/core/Core.h | 2 +- libminifi/include/core/ProcessGroup.h | 4 +- libminifi/include/core/Property.h | 2 +- .../core/repository/VolatileRepository.h | 14 +- libminifi/include/core/state/StateManager.h | 29 +- libminifi/include/core/state/UpdateController.h | 4 +- libminifi/include/core/state/Value.h | 214 +++++++ .../core/state/metrics/DeviceInformation.h | 313 ---------- .../include/core/state/metrics/MetricsBase.h | 161 ----- .../core/state/metrics/MetricsListener.h | 128 ---- .../include/core/state/metrics/ProcessMetrics.h | 102 ---- .../include/core/state/metrics/QueueMetrics.h | 106 ---- .../core/state/metrics/RepositoryMetrics.h | 101 ---- .../include/core/state/metrics/SystemMetrics.h | 108 ---- .../include/core/state/nodes/AgentInformation.h | 586 +++++++++++++++++++ .../include/core/state/nodes/BuildInformation.h | 135 +++++ .../core/state/nodes/DeviceInformation.h | 355 +++++++++++ .../include/core/state/nodes/FlowInformation.h | 273 +++++++++ .../include/core/state/nodes/MetricsBase.h | 264 +++++++++ .../include/core/state/nodes/ProcessMetrics.h | 103 ++++ .../include/core/state/nodes/QueueMetrics.h | 106 ++++ .../core/state/nodes/RepositoryMetrics.h | 101 ++++ .../include/core/state/nodes/StateMonitor.h | 87 +++ .../include/core/state/nodes/SystemMetrics.h | 117 ++++ .../core/state/nodes/TreeUpdateListener.h | 128 ++++ libminifi/include/processors/GetFile.h | 29 +- libminifi/include/processors/GetTCP.h | 30 +- libminifi/include/properties/Configure.h | 1 + libminifi/include/utils/ByteArrayCallback.h | 8 +- libminifi/include/utils/FileOutputCallback.h | 77 +++ libminifi/include/utils/HTTPClient.h | 1 + libminifi/include/utils/file/FileUtils.h | 9 + libminifi/src/Configure.cpp | 1 + libminifi/src/FlowController.cpp | 333 +++++++++-- libminifi/src/c2/C2Agent.cpp | 276 +++++++-- libminifi/src/c2/C2Payload.cpp | 23 +- libminifi/src/c2/ControllerSocketProtocol.cpp | 8 +- libminifi/src/c2/protocols/RESTProtocol.cpp | 234 +++++--- libminifi/src/core/ConfigurableComponent.cpp | 27 +- libminifi/src/core/ProcessGroup.cpp | 2 +- libminifi/src/core/Property.cpp | 3 +- libminifi/src/core/state/StateManager.cpp | 17 +- libminifi/src/core/yaml/YamlConfiguration.cpp | 7 +- libminifi/src/processors/GetFile.cpp | 2 +- libminifi/src/processors/GetTCP.cpp | 2 +- libminifi/src/utils/ByteArrayCallback.cpp | 2 - libminifi/src/utils/FileOutputCallback.cpp | 61 ++ libminifi/src/utils/HTTPClient.cpp | 46 ++ libminifi/test/resources/TestBad.yml | 74 +++ libminifi/test/unit/C2MetricsTests.cpp | 96 +-- libminifi/test/unit/ControllerTests.cpp | 4 +- 79 files changed, 4937 insertions(+), 1428 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a330c57a/.gitignore ---------------------------------------------------------------------- diff --git a/.gitignore b/.gitignore index c687dee..d89d4b6 100644 --- a/.gitignore +++ b/.gitignore @@ -47,6 +47,7 @@ target thirdparty/**/*.o thirdparty/**/*.a libminifi/test/**/*.a +libminifi/include/agent/agent_version.h docs/generated thirdparty/apache-rat/apache-rat* http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a330c57a/C2.md ---------------------------------------------------------------------- diff --git a/C2.md b/C2.md new file mode 100644 index 0000000..194467c --- /dev/null +++ b/C2.md @@ -0,0 +1,129 @@ +<!-- + 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. +--> + +# Apache NiFi - MiNiFi - C++ C2 Readme. + + +This readme defines the Command and control configuration options that work with Apache NiFi. All +options defined are located in minifi.properties. + +## Table of Contents + +- [Description](#description) +- [Configuration](#configuration) + - [Base Options](#base-options) + - [Metrics](#metrics) + +## Description + +Apache NiFi MiNiFI C++ can communicates with a C2 Server via a number of protocols. These protocols +deliver a C2 response the server, expecting requests in a hearbeat response. The protocol transforms +the C2 messages into a protocol specific representation. The internal representation is an AST therefore +you must define the root classes, which configure the classes that branch from the root. You can define +arbitrary nodes and sub-trees, but this isn't necessary and only advantageous for custom C2 servers. That +will be explained in greater detail in the metrics section. + +## Configuration + +### Base Options +For more more insight into the API used within the C2 agent, please visit: +https://cwiki.apache.org/confluence/display/MINIFI/C2+Design+Proposal + + + in minifi.properties + + # Disable/Enable C2 + nifi.c2.enable=true + + # specify classes for the AST response + nifi.c2.root.classes=DeviceInfoNode,AgentInformation,FlowInformation + + # specify C2 protocol -- default is RESTSender if this is not specified + c2.agent.protocol.class=RESTSender + + # control c2 heartbeat interval in millisecocnds + c2.agent.heartbeat.period=3000 + + # enable reporter classes + c2.agent.heartbeat.reporter.class=RESTReciver + + # specify the rest URIs if using RESTSender + c2.rest.url=http://localhost:10080/minifi-c2-api/c2-protocol/heartbeat + c2.rest.url.ack=http://localhost:10080/minifi-c2-api/c2-protocol/acknowledge + + # c2 agent identifier + nifi.c2.agent.identifier=<your identifier> + + # c2 agent class + nifi.c2.agent.class=<your agent class> + + # configure SSL Context service for REST Protocol + c2.rest.ssl.context.service + + +### Metrics + +Command and Control metrics can be used to send metrics through the heartbeat or via the DESCRIBE +operation. Since responses are formed in an AST, metrics can be formed as a sub tree. Metrics classes +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 + + + nifi.c2.root.class.definitions=metrics + nifi.c2.root.class.definitions.metrics.name=metrics + nifi.c2.root.class.definitions.metrics.metrics=typedmetrics,processorMetrics + nifi.c2.root.class.definitions.metrics.metrics.typedmetrics.name=RuntimeMetrics + nifi.c2.root.class.definitions.metrics.metrics.typedmetrics.classes=ProcessMetrics,SystemInformation + nifi.c2.root.class.definitions.metrics.metrics.processorMetrics.name=ProcessorMetrics + 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'. + +This is a comma separated list of all sub trees. In the example, above, only one sub tree exists: metrics. + +The options below metrics define the sub-trees within metrics: typedmetrics and processorMetrics. Each of these has a name. +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": { + "ProcessorMetrics": { + "GetFileMetrics": { + "AcceptedFiles": 22, + "InputBytes": 61755, + "OnTriggerInvocations": 1 + } + }, + "RuntimeMetrics": { + "ProcessMetrics": { + "CpuMetrics": { + "involcs": 1 + }, + "MemoryMetrics": { + "maxrss": 145804 + } + }, + "systeminfo": { + "systemInfo": { + "machinearch": "x86_64", + "physicalMem": 67361411072, + "vCores": 12 + }, + "identifier": "identifier" + } + } + } + http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a330c57a/CMakeLists.txt ---------------------------------------------------------------------- diff --git a/CMakeLists.txt b/CMakeLists.txt index 163e637..8594eaf 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -21,7 +21,7 @@ cmake_minimum_required(VERSION 3.0) project(nifi-minifi-cpp) set(PROJECT_NAME "nifi-minifi-cpp") set(PROJECT_VERSION_MAJOR 0) -set(PROJECT_VERSION_MINOR 4) +set(PROJECT_VERSION_MINOR 5) set(PROJECT_VERSION_PATCH 0) option(SKIP_TESTS "Skips building all tests." OFF) option(PORTABLE "Instructs the compiler to remove architecture specific optimizations" ON) @@ -292,6 +292,30 @@ if (NOT DISABLE_CURL) endif() +get_property(selected_extensions GLOBAL PROPERTY EXTENSION-OPTIONS) + +if (NOT BUILD_IDENTIFIER) + message("empty identifier ${BUILD_IDENTIFIER}") + if (NOT BUILD_IDENTIFIER) + message("empty identifier ${BUILD_IDENTIFIER}") + string(RANDOM LENGTH 24 BUILD_IDENTIFIER) + set(BUILD_IDENTIFIER "${BUILD_IDENTIFIER}" CACHE STRING "Build identifier") + endif() +endif() + +message("BUILD_IDENTIFIER is ${BUILD_IDENTIFIER}") + +execute_process(COMMAND + "${CMAKE_CURRENT_SOURCE_DIR}/generateVersion.sh" + "${PROJECT_VERSION_MAJOR}.${PROJECT_VERSION_MINOR}.${PROJECT_VERSION_PATCH}" + ${CMAKE_CURRENT_SOURCE_DIR} + ${CMAKE_CURRENT_SOURCE_DIR}/libminifi/include/agent/ + "${CMAKE_CXX_COMPILER}" + "${CMAKE_CXX_COMPILER_VERSION}" + "${CMAKE_CXX_FLAGS}" + "${selected_extensions}" + "${BUILD_IDENTIFIER}") + # Generate source assembly set(ASSEMBLY_BASE_NAME "${CMAKE_PROJECT_NAME}-${PROJECT_VERSION_MAJOR}.${PROJECT_VERSION_MINOR}.${PROJECT_VERSION_PATCH}") set(CPACK_SOURCE_GENERATOR "TGZ") http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a330c57a/README.md ---------------------------------------------------------------------- diff --git a/README.md b/README.md index f102faf..91c41c6 100644 --- a/README.md +++ b/README.md @@ -27,6 +27,7 @@ MiNiFi is a child project effort of Apache NiFi. This repository is for a nativ - [Cleaning](#cleaning) - [Configuring](#configuring) - [Running](#running) + - [Deploying](#deploying) - [Documentation](#documentation) - [License](#license) @@ -572,25 +573,7 @@ To enable HTTP Proxy for a remote process group. proxy password: ### Command and Control Configuration -For more more insight into the API used within the C2 agent, please visit: -https://cwiki.apache.org/confluence/display/MINIFI/C2+Design+Proposal - - in minifi.properties - - #Disable/Enable C2 - nifi.c2.enable=true - - #specify metrics classes - nifi.flow.metrics.classes=DeviceInformation,SystemInformation,ProcessMetrics - - #specify C2 protocol - c2.agent.protocol.class=RESTSender - - #control c2 heartbeat interval in millisecocnds - c2.agent.heartbeat.period=3000 - - # enable reporter classes - c2.agent.heartbeat.reporter.class=RESTReciver +Please see the [C2 readme](C2.md) for more informatoin ### Configuring Repository storage locations @@ -738,6 +721,14 @@ MiNiFi can also be installed as a system service using minifi.sh with an optiona $ ./bin/minifi.sh install [service name] +### Deploying +MiNiFi C++ comes with a deployment script. This will build and package minifi. Additionally, a file named build_output will be +created within the build directory that contains a manifest of build artifacts. + + $ deploy.sh <build identifier> + +The build identifier will be carried with the deployed binary for the configuration you specify. By default all extensions will be built. + ### Managing MiNFI C++ through the MiNiFi Controller The MiNiFi controller is an executable in the bin directory that can be used to control the MiNFi C++ agent while it runs. Currently the controller will let you stop subcomponents within a running instance, clear queues, get the status of queues, and update the flow for a warm re-deploy. http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a330c57a/bin/minifi.sh ---------------------------------------------------------------------- diff --git a/bin/minifi.sh b/bin/minifi.sh index dca94f3..4fc7123 100755 --- a/bin/minifi.sh +++ b/bin/minifi.sh @@ -225,6 +225,22 @@ case "\$1" in exit 3; fi ;; + update) + if [ -f \${bin_dir}/minifi.update ]; then + \${bin_dir}/minifi.sh stop + cp \${bin_dir}/minifi \${bin_dir}/minifi.bak + cp \${bin_dir}/minifi.update \${bin_dir}/minifi + # ensure that the command is now running + \${bin_dir}/minifi.sh start + saved_pid=\$(get_pid) + if [ "\${saved_pid}" -gt 0 ]; then + if [ \$(active_pid \${saved_pid}) -ne 0 ]; then + cp \${bin_dir}/minifi.bak \${bin_dir}/minifi + \${bin_dir}/minifi.sh start + fi + fi + fi + ;; restart) echo Restarting MiNiFi service \${bin_dir}/minifi.sh stop @@ -324,6 +340,22 @@ case "$1" in exit 3; fi ;; + update) + if [ -f ${bin_dir}/minifi.update ]; then + ${bin_dir}/minifi.sh stop + cp ${bin_dir}/minifi ${bin_dir}/minifi.bak + cp ${bin_dir}/minifi.update ${bin_dir}/minifi + # ensure that the command is now running + ${bin_dir}/minifi.sh start + saved_pid=$(get_pid) + if [ "${saved_pid}" -gt 0 ]; then + if [ $(active_pid ${saved_pid}) -ne 0 ]; then + cp ${bin_dir}/minifi.bak ${bin_dir}/minifi + ${bin_dir}/minifi.sh start + fi + fi + fi + ;; restart) echo Restarting MiNiFi service ${bin_dir}/minifi.sh stop http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a330c57a/bootstrap.sh ---------------------------------------------------------------------- diff --git a/bootstrap.sh b/bootstrap.sh index a23b0d5..7706a58 100755 --- a/bootstrap.sh +++ b/bootstrap.sh @@ -24,17 +24,19 @@ NO_COLOR='\033[0;0;39m' CORES=1 BUILD="false" PACKAGE="false" - +BUILD_IDENTIFIER="" TRUE="Enabled" FALSE="Disabled" FEATURES_SELECTED="false" AUTO_REMOVE_EXTENSIONS="true" export NO_PROMPT="false" +DEPLOY="false" OPTIONS=() CMAKE_OPTIONS_ENABLED=() CMAKE_OPTIONS_DISABLED=() CMAKE_MIN_VERSION=() +DEPLOY_LIMITS=() DEPENDENCIES=() @@ -61,6 +63,12 @@ add_disabled_option(){ if [ ! -z "$4" ]; then CMAKE_MIN_VERSION+=("$1:$4") fi + + if [ ! -z "$5" ]; then + if [ "$5" = "true" ]; then + DEPLOY_LIMITS+=("$1") + fi + fi } add_dependency(){ @@ -69,11 +77,37 @@ add_dependency(){ ### parse the command line arguments + +EnableAllFeatures(){ + for option in "${OPTIONS[@]}" ; do + feature_status=${!option} + if [ "$feature_status" = "${FALSE}" ]; then + ToggleFeature $option + fi + # eval "$option=${TRUE}" + done +} + while :; do case $1 in -n|--noprompt) NO_PROMPT="true" ;; + -e|--enableall) + NO_PROMPT="true" + FEATURES_SELECTED="true" + EnableAllFeatures + ;; + -d|--deploy) + NO_PROMPT="true" + DEPLOY="true" + FEATURES_SELECTED="true" + EnableAllFeatures + ;; + -t|--travis) + NO_PROMPT="true" + FEATURES_SELECTED="true" + ;; -p|--package) CORES=$(grep -c ^processor /proc/cpuinfo 2>/dev/null || sysctl -n hw.ncpu) BUILD="true" @@ -83,6 +117,9 @@ while :; do CORES=$(grep -c ^processor /proc/cpuinfo 2>/dev/null || sysctl -n hw.ncpu) BUILD="true" ;; + "--build_identifier="* ) + BUILD_IDENTIFIER="${1#*=}" + ;; *) break esac shift @@ -107,14 +144,7 @@ if [ "$NO_PROMPT" = "true" ]; then agree="N" echo "****************************************" echo "Welcome, this boostrap script will update your system to install MiNIFi C++" - echo "You have opted to skip prompts. Do you agree to this script installing" - echo "System packages without prompting you?" - read -p "Enter Y to continue, N to exit [ Y/N ] " agree - if [ "$agree" = "Y" ] || [ "$agree" = "y" ]; then - echo "Continuing..." - else - exit - fi + echo "You have opted to skip prompts. " fi @@ -239,18 +269,29 @@ add_disabled_option KAFKA_ENABLED ${FALSE} "ENABLE_LIBRDKAFKA" "3.4.0" add_disabled_option MQTT_ENABLED ${FALSE} "ENABLE_MQTT" -#add_disabled_option BUSTACHE_ENABLED ${FALSE} "ENABLE_BUSTACHE" -#add_dependency BUSTACHE_ENABLED "boost" +# Since the following extensions have limitations on + +add_disabled_option BUSTACHE_ENABLED ${FALSE} "ENABLE_BUSTACHE" "2.6" ${TRUE} +add_dependency BUSTACHE_ENABLED "boost" ## currently need to limit on certain platforms -#add_disabled_option TENSORFLOW_ENABLED ${FALSE} "ENABLE_TENSORFLOW" +add_disabled_option TENSORFLOW_ENABLED ${FALSE} "ENABLE_TENSORFLOW" "2.6" ${TRUE} +add_dependency TENSORFLOW_ENABLED "tensorflow" pause(){ read -p "Press [Enter] key to continue..." fackEnterKey } - +can_deploy(){ + for option in "${DEPLOY_LIMITS[@]}" ; do + OPT=${option%%:*} + if [ "${OPT}" = "$1" ]; then + echo "false" + fi + done + echo "true" +} ToggleFeature(){ VARIABLE_VALUE=${!1} @@ -278,21 +319,15 @@ ToggleFeature(){ fi done CAN_ENABLE=$(verify_enable $1) + CAN_DEPLOY=$(can_deploy $1) if [ "$CAN_ENABLE" = "true" ]; then - eval "$1=${TRUE}" + if [[ "$DEPLOY" = "true" && "$CAN_DEPLOY" = "true" ]] || [[ "$DEPLOY" = "false" ]]; then + eval "$1=${TRUE}" + fi fi fi } -EnableAllFeatures(){ - for option in "${OPTIONS[@]}" ; do - feature_status=${!option} - if [ "$feature_status" = "${FALSE}" ]; then - ToggleFeature $option - fi - # eval "$option=${TRUE}" - done -} print_feature_status(){ feature="$1" @@ -334,24 +369,6 @@ print_feature_status(){ } -### parse the command line arguments - -while :; do - case $1 in - -e|--enableall) - NO_PROMPT="true" - FEATURES_SELECTED="true" - EnableAllFeatures - ;; - -t|--travis) - NO_PROMPT="true" - FEATURES_SELECTED="true" - ;; - *) break - esac - shift -done - if [ ! -d "build" ]; then mkdir build/ else @@ -371,7 +388,7 @@ fi ## change to the directory -cd build +pushd build show_supported_features() { @@ -495,6 +512,8 @@ build_cmake_command(){ CMAKE_BUILD_COMMAND="${CMAKE_BUILD_COMMAND} -DPORTABLE=OFF " fi + CMAKE_BUILD_COMMAND="${CMAKE_BUILD_COMMAND} -DBUILD_IDENTIFIER=${BUILD_IDENTIFIER}" + add_os_flags curl -V | grep OpenSSL &> /dev/null @@ -530,4 +549,4 @@ if [ "$PACKAGE" = "true" ]; then make package fi - +popd http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a330c57a/centos.sh ---------------------------------------------------------------------- diff --git a/centos.sh b/centos.sh index 03d3b56..2794576 100644 --- a/centos.sh +++ b/centos.sh @@ -24,6 +24,8 @@ verify_enable() { echo "false" elif [ "$feature" = "USB_ENABLED" ]; then echo "false" + elif [ "$feature" = "TENSORFLOW_ENABLED" ]; then + echo "false" else echo "true" fi http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a330c57a/controller/Controller.h ---------------------------------------------------------------------- diff --git a/controller/Controller.h b/controller/Controller.h index 95e05b1..0a2b292 100644 --- a/controller/Controller.h +++ b/controller/Controller.h @@ -244,4 +244,60 @@ std::shared_ptr<core::controller::ControllerService> getControllerService(const return service; } + void printManifest(const std::shared_ptr<minifi::Configure> &configuration) { + + std::string prov_repo_class = "volatileprovenancerepository"; + std::string flow_repo_class = "volatileflowfilerepository"; + std::string nifi_configuration_class_name = "yamlconfiguration"; + std::string content_repo_class = "volatilecontentrepository"; + + std::shared_ptr<logging::LoggerProperties> log_properties = std::make_shared<logging::LoggerProperties>(); + log_properties->setHome("./"); + log_properties->set("appender.stdout","stdout"); + log_properties->set("logger.org::apache::nifi::minifi","OFF,stdout"); + logging::LoggerConfiguration::getConfiguration().initialize(log_properties); + + configuration->set(minifi::Configure::nifi_flow_configuration_file,"../conf/config.yml"); + configuration->get(minifi::Configure::nifi_provenance_repository_class_name, prov_repo_class); + // Create repos for flow record and provenance + std::shared_ptr<core::Repository> prov_repo = core::createRepository(prov_repo_class, true, "provenance"); + prov_repo->initialize(configuration); + + configuration->get(minifi::Configure::nifi_flow_repository_class_name, flow_repo_class); + + std::shared_ptr<core::Repository> flow_repo = core::createRepository(flow_repo_class, true, "flowfile"); + + flow_repo->initialize(configuration); + + configuration->get(minifi::Configure::nifi_content_repository_class_name, content_repo_class); + + std::shared_ptr<core::ContentRepository> content_repo = core::createContentRepository(content_repo_class, true, "content"); + + content_repo->initialize(configuration); + + std::string content_repo_path; + if (configuration->get(minifi::Configure::nifi_dbcontent_repository_directory_default, content_repo_path)) { + minifi::setDefaultDirectory(content_repo_path); + } + + configuration->set("c2.agent.heartbeat.period","25"); + configuration->set("nifi.c2.root.classes","AgentInformation"); + configuration->set("nifi.c2.enable","true"); + configuration->set("c2.agent.listen","true"); + configuration->set("c2.agent.heartbeat.reporter.classes","AgentPrinter"); + + configuration->get(minifi::Configure::nifi_configuration_class_name, nifi_configuration_class_name); + + std::shared_ptr<minifi::io::StreamFactory> stream_factory = std::make_shared<minifi::io::StreamFactory>(configuration); + + std::unique_ptr<core::FlowConfiguration> flow_configuration = core::createFlowConfiguration(prov_repo, flow_repo, content_repo, configuration, stream_factory, nifi_configuration_class_name); + + std::shared_ptr<minifi::FlowController> controller = std::unique_ptr<minifi::FlowController>( + new minifi::FlowController(prov_repo, flow_repo, configuration, std::move(flow_configuration), content_repo,"manifest",false)); + controller->load(); + controller->start(); + std::this_thread::sleep_for(std::chrono::milliseconds(10000)); + controller->stop(true); +} + #endif /* CONTROLLER_CONTROLLER_H_ */ http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a330c57a/controller/MiNiFiController.cpp ---------------------------------------------------------------------- diff --git a/controller/MiNiFiController.cpp b/controller/MiNiFiController.cpp index 98acd15..dc12306 100644 --- a/controller/MiNiFiController.cpp +++ b/controller/MiNiFiController.cpp @@ -75,7 +75,7 @@ int main(int argc, char **argv) { if (!validHome(minifiHome)) { logger->log_error("No valid MINIFI_HOME could be inferred. " "Please set MINIFI_HOME or run minifi from a valid location."); - return -1; + //return -1; } std::shared_ptr<minifi::Configure> configuration = std::make_shared<minifi::Configure>(); @@ -138,6 +138,7 @@ int main(int argc, char **argv) { ("getsize", "Reports the size of the associated connection queue", cxxopts::value<std::vector<std::string>>()) //NOLINT ("updateflow", "Updates the flow of the agent using the provided flow file", cxxopts::value<std::string>()) //NOLINT ("getfull", "Reports a list of full connections") //NOLINT + ("manifest", "Generates a manifest for the current binary") //NOLINT ("noheaders", "Removes headers from output streams"); bool show_headers = true; @@ -245,6 +246,14 @@ int main(int argc, char **argv) { if (updateFlow(std::move(socket), std::cout, flow_file) < 0) std::cout << "Could not connect to remote host " << host << ":" << port << std::endl; } + + if (result.count("manifest") > 0) { + printManifest(configuration); + } + }catch (const std::exception &exc) + { + // catch anything thrown within try block that derives from std::exception + std::cerr << exc.what() << std::endl; } catch (...) { std::cout << options.help( { "", "Group" }) << std::endl; exit(0); http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a330c57a/darwin.sh ---------------------------------------------------------------------- diff --git a/darwin.sh b/darwin.sh index e68c237..2283d3b 100644 --- a/darwin.sh +++ b/darwin.sh @@ -20,7 +20,7 @@ verify_enable() { feature="$1" feature_status=${!1} if [ "$feature" = "BUSTACHE_ENABLED" ]; then - BUSTACHE_MAX="9" + BUSTACHE_MAX="9" ## we should check the xcode version CLANG_VERSION=`clang --version | head -n 1 | awk '{print $4}'` CLANG_MAJOR=`echo $CLANG_VERSION | cut -d. -f1` @@ -73,6 +73,8 @@ build_deps(){ INSTALLED+=("flex") elif [ "$FOUND_VALUE" = "python" ]; then INSTALLED+=("python") + elif [ "$FOUND_VALUE" = "boost" ]; then + INSTALLED+=("boost") elif [ "$FOUND_VALUE" = "lua" ]; then INSTALLED+=("lua") elif [ "$FOUND_VALUE" = "gpsd" ]; then http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a330c57a/deploy.sh ---------------------------------------------------------------------- diff --git a/deploy.sh b/deploy.sh new file mode 100755 index 0000000..63d58db --- /dev/null +++ b/deploy.sh @@ -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. +#!/bin/bash + +build_identifier=$1 + +echo "${build_identifier}" > build_identifier + +./bootstrap.sh -d -p --build_identifier=${build_identifier} + +pushd build + ./controller/minificontroller --manifest >> build_output +popd http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a330c57a/extensions/http-curl/HTTPCurlLoader.h ---------------------------------------------------------------------- diff --git a/extensions/http-curl/HTTPCurlLoader.h b/extensions/http-curl/HTTPCurlLoader.h index ec90e99..23dcb44 100644 --- a/extensions/http-curl/HTTPCurlLoader.h +++ b/extensions/http-curl/HTTPCurlLoader.h @@ -49,7 +49,6 @@ class __attribute__((visibility("default"))) HttpCurlObjectFactory : public core */ virtual std::vector<std::string> getClassNames() override{ std::vector<std::string> class_names; - class_names.push_back("RESTProtocol"); class_names.push_back("HttpProtocol"); class_names.push_back("RESTSender"); class_names.push_back("InvokeHTTP"); http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a330c57a/extensions/http-curl/client/HTTPClient.cpp ---------------------------------------------------------------------- diff --git a/extensions/http-curl/client/HTTPClient.cpp b/extensions/http-curl/client/HTTPClient.cpp index eae8e1d..8f7ca42 100644 --- a/extensions/http-curl/client/HTTPClient.cpp +++ b/extensions/http-curl/client/HTTPClient.cpp @@ -297,8 +297,13 @@ const char *HTTPClient::getContentType() { } const std::vector<char> &HTTPClient::getResponseBody() { - if (response_body_.size() == 0) + if (response_body_.size() == 0){ + if (callback && callback->ptr){ + response_body_ = callback->ptr->to_string(); + }else{ response_body_ = read_callback_.to_string(); + } + } return response_body_; } http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a330c57a/extensions/http-curl/protocols/AgentPrinter.cpp ---------------------------------------------------------------------- diff --git a/extensions/http-curl/protocols/AgentPrinter.cpp b/extensions/http-curl/protocols/AgentPrinter.cpp new file mode 100644 index 0000000..efcf195 --- /dev/null +++ b/extensions/http-curl/protocols/AgentPrinter.cpp @@ -0,0 +1,141 @@ +/** + * + * 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 "AgentPrinter.h" +#include <algorithm> +#include <memory> +#include <utility> +#include <map> +#include <list> +#include <string> +#include <vector> + +namespace org { +namespace apache { +namespace nifi { +namespace minifi { +namespace c2 { + +AgentPrinter::AgentPrinter(std::string name, uuid_t uuid) + : HeartBeatReporter(name, uuid), + logger_(logging::LoggerFactory<AgentPrinter>::getLogger()) { +} + +void AgentPrinter::initialize(const std::shared_ptr<core::controller::ControllerServiceProvider> &controller, const std::shared_ptr<state::StateMonitor> &updateSink, + const std::shared_ptr<Configure> &configure) { + HeartBeatReporter::initialize(controller, updateSink, configure); +} +int16_t AgentPrinter::heartbeat(const C2Payload &payload) { + std::string outputConfig = serializeJsonRootPayload(payload); + return 0; +} + +std::string AgentPrinter::serializeJsonRootPayload(const C2Payload& payload) { + rapidjson::Document json_payload(payload.isContainer() ? rapidjson::kArrayType : rapidjson::kObjectType); + rapidjson::Document::AllocatorType &alloc = json_payload.GetAllocator(); + + rapidjson::Value opReqStrVal; + std::string operation_request_str = getOperation(payload); + opReqStrVal.SetString(operation_request_str.c_str(), operation_request_str.length(), alloc); + json_payload.AddMember("operation", opReqStrVal, alloc); + + std::string operationid = payload.getIdentifier(); + if (operationid.length() > 0) { + rapidjson::Value operationIdVal = getStringValue(operationid, alloc); + json_payload.AddMember("operationid", operationIdVal, alloc); + } + + mergePayloadContent(json_payload, payload, alloc); + + for (const auto &nested_payload : payload.getNestedPayloads()) { + rapidjson::Value np_key = getStringValue(nested_payload.getLabel(), alloc); + rapidjson::Value np_value = serializeJsonPayload(nested_payload, alloc); + json_payload.AddMember(np_key, np_value, alloc); + } + + rapidjson::StringBuffer buffer; + rapidjson::PrettyWriter<rapidjson::StringBuffer> writer(buffer); + json_payload.Accept(writer); + return buffer.GetString(); +} + +rapidjson::Value AgentPrinter::serializeJsonPayload(const C2Payload &payload, rapidjson::Document::AllocatorType &alloc) { + // get the name from the content + rapidjson::Value json_payload(payload.isContainer() ? rapidjson::kArrayType : rapidjson::kObjectType); + + std::map<std::string, std::list<rapidjson::Value*>> children; + bool print = payload.getLabel() == "agentManifest"; + for (const auto &nested_payload : payload.getNestedPayloads()) { + + rapidjson::Value* child_payload = new rapidjson::Value(serializeJsonPayload(nested_payload, alloc)); + children[nested_payload.getLabel()].push_back(child_payload); + + } + + // child_vector is Pair<string, vector<Value*>> + for (auto child_vector : children) { + rapidjson::Value children_json; + rapidjson::Value newMemberKey = getStringValue(child_vector.first, alloc); + if (child_vector.second.size() > 1) { + children_json.SetArray(); + for (auto child : child_vector.second) + children_json.PushBack(child->Move(), alloc); + if (json_payload.IsArray()) + json_payload.PushBack(children_json, alloc); + else + json_payload.AddMember(newMemberKey, children_json, alloc); + } else if (child_vector.second.size() == 1) { + rapidjson::Value* first = child_vector.second.front(); + + if (first->IsObject() && first->HasMember(newMemberKey)) { + if (json_payload.IsArray()) + json_payload.PushBack((*first)[newMemberKey].Move(), alloc); + else + json_payload.AddMember(newMemberKey, (*first)[newMemberKey].Move(), alloc); + } else { + if (json_payload.IsArray()) { + json_payload.PushBack(first->Move(), alloc); + } else { + json_payload.AddMember(newMemberKey, first->Move(), alloc); + } + } + } + + for (rapidjson::Value* child : child_vector.second) + delete child; + } + + mergePayloadContent(json_payload, payload, alloc); + + if (print) { + + rapidjson::StringBuffer buffer; + rapidjson::PrettyWriter<rapidjson::StringBuffer> writer(buffer); + json_payload.Accept(writer); + std::cout << buffer.GetString() << std::endl; + std::exit(1); + } + + return json_payload; +} + +} /* namespace c2 */ +} /* namespace minifi */ +} /* namespace nifi */ +} /* namespace apache */ +} /* namespace org */ http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a330c57a/extensions/http-curl/protocols/AgentPrinter.h ---------------------------------------------------------------------- diff --git a/extensions/http-curl/protocols/AgentPrinter.h b/extensions/http-curl/protocols/AgentPrinter.h new file mode 100644 index 0000000..e1f7480 --- /dev/null +++ b/extensions/http-curl/protocols/AgentPrinter.h @@ -0,0 +1,80 @@ +/** + * + * 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. + */ +#ifndef LIBMINIFI_INCLUDE_C2_AGENTPRINTER_H_ +#define LIBMINIFI_INCLUDE_C2_AGENTPRINTER_H_ + +#include <string> +#include <mutex> +#include "core/Resource.h" +#include "c2/protocols/RESTProtocol.h" +#include "CivetServer.h" +#include "c2/C2Protocol.h" +#include "controllers/SSLContextService.h" + +namespace org { +namespace apache { +namespace nifi { +namespace minifi { +namespace c2 { + +/** + * Purpose and Justification: Encapsulates printing agent information. + * + * Will be used to print agent information from the C2 response to stdout, selecting the agent's manifest + * + */ +class AgentPrinter : public RESTProtocol, public HeartBeatReporter { + public: + AgentPrinter(std::string name, uuid_t uuid = nullptr); + + /** + * Initialize agent printer. + */ + virtual void initialize(const std::shared_ptr<core::controller::ControllerServiceProvider> &controller, const std::shared_ptr<state::StateMonitor> &updateSink, + const std::shared_ptr<Configure> &configure) override; + + /** + * Accepts the heartbeat, only extracting AgentInformation. + */ + virtual int16_t heartbeat(const C2Payload &heartbeat) override; + + /** + * Overrides extracting the agent information from the root. + */ + virtual std::string serializeJsonRootPayload(const C2Payload& payload) override; + + /** + * Overrides extracting the agent information from the payload. + */ + virtual rapidjson::Value serializeJsonPayload(const C2Payload &payload, rapidjson::Document::AllocatorType &alloc) override; + + protected: + + private: + std::shared_ptr<logging::Logger> logger_; +}; + +REGISTER_RESOURCE(AgentPrinter); + +} /* namesapce c2 */ +} /* namespace minifi */ +} /* namespace nifi */ +} /* namespace apache */ +} /* namespace org */ + +#endif /* LIBMINIFI_INCLUDE_C2_AGENTPRINTER_H_ */ http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a330c57a/extensions/http-curl/protocols/RESTSender.cpp ---------------------------------------------------------------------- diff --git a/extensions/http-curl/protocols/RESTSender.cpp b/extensions/http-curl/protocols/RESTSender.cpp index a98ba03..c2fd56e 100644 --- a/extensions/http-curl/protocols/RESTSender.cpp +++ b/extensions/http-curl/protocols/RESTSender.cpp @@ -19,11 +19,16 @@ #include "RESTSender.h" #include <algorithm> +#include <iostream> #include <memory> #include <utility> #include <map> #include <string> #include <vector> +#include "utils/file/FileUtils.h" +#include "utils/StringUtils.h" +#include "utils/file/FileManager.h" +#include "utils/FileOutputCallback.h" namespace org { namespace apache { @@ -40,8 +45,17 @@ void RESTSender::initialize(const std::shared_ptr<core::controller::ControllerSe C2Protocol::initialize(controller, configure); // base URL when one is not specified. if (nullptr != configure) { + std::string update_str, ssl_context_service_str; configure->get("c2.rest.url", rest_uri_); configure->get("c2.rest.url.ack", ack_uri_); + if (configure->get("c2.rest.ssl.context.service", ssl_context_service_str)) { + auto service = controller->getControllerService(ssl_context_service_str); + if (nullptr != service) { + ssl_context_service_ = std::static_pointer_cast<minifi::controllers::SSLContextService>(service); + } + } + configure->get("c2.rest.heartbeat.minimize.updates", update_str); + utils::StringUtils::StringToBool(update_str, minimize_updates_); } logger_->log_debug("Submitting to %s", rest_uri_); } @@ -52,7 +66,6 @@ C2Payload RESTSender::consumePayload(const std::string &url, const C2Payload &pa if (direction == Direction::TRANSMIT) { outputConfig = serializeJsonRootPayload(payload); } - return sendPayload(url, direction, payload, outputConfig); } @@ -72,7 +85,6 @@ void RESTSender::update(const std::shared_ptr<Configure> &configure) { const C2Payload RESTSender::sendPayload(const std::string url, const Direction direction, const C2Payload &payload, const std::string outputConfig) { utils::HTTPClient client(url, ssl_context_service_); client.setConnectionTimeout(2); - std::unique_ptr<utils::ByteInputCallBack> input = nullptr; std::unique_ptr<utils::HTTPUploadCallback> callback = nullptr; if (direction == Direction::TRANSMIT) { @@ -89,15 +101,25 @@ const C2Payload RESTSender::sendPayload(const std::string url, const Direction d // since we are not uploading anything on a get client.set_request_method("GET"); } - client.appendHeader("Accept: application/json"); - client.setContentType("application/json"); + + std::unique_ptr<utils::FileOutputCallback> file_callback = nullptr; + utils::HTTPReadCallback read; + if (payload.getOperation() == TRANSFER) { + utils::file::FileManager file_man; + auto file = file_man.unique_file(true); + file_callback = std::unique_ptr<utils::FileOutputCallback>(new utils::FileOutputCallback(file)); + read.pos = 0; + read.ptr = file_callback.get(); + client.setReadCallback(&read); + } else { + client.appendHeader("Accept: application/json"); + client.setContentType("application/json"); + } bool isOkay = client.submit(); int64_t respCode = client.getResponseCode(); - if (isOkay && respCode) { if (payload.isRaw()) { C2Payload response_payload(payload.getOperation(), state::UpdateState::READ_COMPLETE, true, true); - response_payload.setRawData(client.getResponseBody()); return response_payload; } http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a330c57a/extensions/http-curl/tests/C2FailedUpdateTest.cpp ---------------------------------------------------------------------- diff --git a/extensions/http-curl/tests/C2FailedUpdateTest.cpp b/extensions/http-curl/tests/C2FailedUpdateTest.cpp new file mode 100644 index 0000000..2aed1f4 --- /dev/null +++ b/extensions/http-curl/tests/C2FailedUpdateTest.cpp @@ -0,0 +1,186 @@ +/** + * + * 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" + +void waitToVerifyProcessor() { + std::this_thread::sleep_for(std::chrono::seconds(10)); +} + +static std::vector<std::string> responses; + +class ConfigHandler : public CivetHandler { + public: + ConfigHandler() { + calls_ = 0; + } + bool handlePost(CivetServer *server, struct mg_connection *conn) { + calls_++; + if (responses.size() > 0) { + std::string top_str = responses.back(); + responses.pop_back(); + mg_printf(conn, "HTTP/1.1 200 OK\r\nContent-Type: " + "text/plain\r\nContent-Length: %lu\r\nConnection: close\r\n\r\n", + top_str.length()); + mg_printf(conn, "%s", top_str.c_str()); + } else { + mg_printf(conn, "HTTP/1.1 500 Internal Server Error\r\n"); + } + + return true; + } + + 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; + } + std::string test_file_location_; + std::string base_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().setDebug<minifi::c2::C2Agent>(); + + const char *options[] = { "document_root", ".", "listening_ports", "7071", 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]); + } + + CivetServer server(cpp_options); + ConfigHandler h_ex; + server.addHandler("/update", h_ex); + std::string key_dir, test_file_location; + if (argc > 1) { + h_ex.base_location_ = test_file_location = argv[1]; + h_ex.test_file_location_ = argv[2]; + key_dir = argv[3]; + } + std::string heartbeat_response = "{\"operation\" : \"heartbeat\",\"requested_operations\": [ {" + "\"operation\" : \"update\", " + "\"operationid\" : \"8675309\", " + "\"name\": \"configuration\"" + "}]}"; + + responses.push_back(heartbeat_response); + + 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(); + std::string response = "{\"operation\" : \"heartbeat\",\"requested_operations\": [ {" + "\"operation\" : \"update\", " + "\"operationid\" : \"8675309\", " + "\"name\": \"configuration\", \"content\": { \"location\": \"http://localhost:7071/update\"}}]}";; + responses.push_back(response); + } + + std::shared_ptr<minifi::Configure> configuration = std::make_shared<minifi::Configure>(); + + configuration->set("c2.rest.url", "http://localhost:7071/update";); + configuration->set("c2.agent.heartbeat.period", "1000"); + mkdir("content_repository", S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH); + + 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 = std::make_shared<minifi::io::StreamFactory>(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); + + 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(); + + controller->load(); + controller->start(); + waitToVerifyProcessor(); + + controller->waitUnload(60000); + auto then = std::chrono::system_clock::now(); + + auto milliseconds = std::chrono::duration_cast<std::chrono::milliseconds>(then - start).count(); + std::string logs = LogTestController::getInstance().log_output.str(); + assert(logs.find("Invalid configuration payload") != std::string::npos); + assert(logs.find("update failed.") != std::string::npos); + LogTestController::getInstance().reset(); + rmdir("./content_repository"); + assert(h_ex.calls_ <= (milliseconds / 1000) + 1); + + return 0; +} http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a330c57a/extensions/http-curl/tests/C2UpdateAgentTest.cpp ---------------------------------------------------------------------- diff --git a/extensions/http-curl/tests/C2UpdateAgentTest.cpp b/extensions/http-curl/tests/C2UpdateAgentTest.cpp new file mode 100644 index 0000000..26a3e86 --- /dev/null +++ b/extensions/http-curl/tests/C2UpdateAgentTest.cpp @@ -0,0 +1,184 @@ +/** + * + * 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" + +void waitToVerifyProcessor() { + std::this_thread::sleep_for(std::chrono::seconds(10)); +} + +static std::vector<std::string> responses; + +class ConfigHandler : public CivetHandler { + public: + ConfigHandler() { + calls_ = 0; + } + bool handlePost(CivetServer *server, struct mg_connection *conn) { + calls_++; + if (responses.size() > 0) { + std::string top_str = responses.back(); + responses.pop_back(); + mg_printf(conn, "HTTP/1.1 200 OK\r\nContent-Type: " + "text/plain\r\nContent-Length: %lu\r\nConnection: close\r\n\r\n", + top_str.length()); + mg_printf(conn, "%s", top_str.c_str()); + } else { + mg_printf(conn, "HTTP/1.1 500 Internal Server Error\r\n"); + } + + return true; + } + + 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; + } + 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>(); + + const char *options[] = { "document_root", ".", "listening_ports", "7072", 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]); + } + + CivetServer server(cpp_options); + ConfigHandler h_ex; + server.addHandler("/update", h_ex); + std::string key_dir, test_file_location; + if (argc > 1) { + h_ex.test_file_location_ = test_file_location = argv[1]; + key_dir = argv[2]; + } + std::string heartbeat_response = "{\"operation\" : \"heartbeat\",\"requested_operations\": [ {" + "\"operation\" : \"update\", " + "\"operationid\" : \"8675309\", " + "\"name\": \"agent\"" + "}]}"; + + responses.push_back(heartbeat_response); + + 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(); + std::string response = "{\"operation\" : \"heartbeat\",\"requested_operations\": [ {" + "\"operation\" : \"update\", " + "\"operationid\" : \"8675309\", " + "\"name\": \"agent\", \"content\": { \"location\": \"http://localhost:7072/update\"}}]}";; + responses.push_back(response); + } + + std::shared_ptr<minifi::Configure> configuration = std::make_shared<minifi::Configure>(); + + configuration->set("c2.rest.url", "http://localhost:7072/update";); + configuration->set("c2.agent.heartbeat.period", "1000"); + mkdir("content_repository", S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH); + + 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); + configuration->set("c2.agent.update.command", "echo \"verification command\""); + + std::shared_ptr<minifi::io::StreamFactory> stream_factory = std::make_shared<minifi::io::StreamFactory>(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); + + 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(); + + controller->load(); + controller->start(); + waitToVerifyProcessor(); + + controller->waitUnload(60000); + auto then = std::chrono::system_clock::now(); + + auto milliseconds = std::chrono::duration_cast<std::chrono::milliseconds>(then - start).count(); + std::string logs = LogTestController::getInstance().log_output.str(); + assert(logs.find("removing command") != std::string::npos); + LogTestController::getInstance().reset(); + rmdir("./content_repository"); + assert(h_ex.calls_ <= (milliseconds / 1000) + 1); + + return 0; +} http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a330c57a/extensions/http-curl/tests/C2UpdateTest.cpp ---------------------------------------------------------------------- diff --git a/extensions/http-curl/tests/C2UpdateTest.cpp b/extensions/http-curl/tests/C2UpdateTest.cpp index f21084b..0799ae5 100644 --- a/extensions/http-curl/tests/C2UpdateTest.cpp +++ b/extensions/http-curl/tests/C2UpdateTest.cpp @@ -102,7 +102,7 @@ int main(int argc, char **argv) { LogTestController::getInstance().setDebug<minifi::c2::RESTSender>(); LogTestController::getInstance().setDebug<minifi::c2::C2Agent>(); - const char *options[] = { "document_root", ".", "listening_ports", "9090", 0 }; + const char *options[] = { "document_root", ".", "listening_ports", "7070", 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]); @@ -134,13 +134,13 @@ int main(int argc, char **argv) { std::string response = "{\"operation\" : \"heartbeat\",\"requested_operations\": [ {" "\"operation\" : \"update\", " "\"operationid\" : \"8675309\", " - "\"name\": \"configuration\", \"content\": { \"location\": \"http://localhost:9090/update\"}}]}";; + "\"name\": \"configuration\", \"content\": { \"location\": \"http://localhost:7070/update\"}}]}";; responses.push_back(response); } std::shared_ptr<minifi::Configure> configuration = std::make_shared<minifi::Configure>(); - configuration->set("c2.rest.url", "http://localhost:9090/update";); + configuration->set("c2.rest.url", "http://localhost:7070/update";); configuration->set("c2.agent.heartbeat.period", "1000"); mkdir("content_repository", S_IRWXU | S_IRWXG | S_IROTH | S_IXOTH); http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a330c57a/extensions/http-curl/tests/CMakeLists.txt ---------------------------------------------------------------------- diff --git a/extensions/http-curl/tests/CMakeLists.txt b/extensions/http-curl/tests/CMakeLists.txt index 994177b..721c221 100644 --- a/extensions/http-curl/tests/CMakeLists.txt +++ b/extensions/http-curl/tests/CMakeLists.txt @@ -71,6 +71,8 @@ message("-- Finished building ${CURL_INT_TEST_COUNT} libcURL integration test fi 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 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}/") #add_test(NAME HttpGetIntegrationTestSecure COMMAND HttpGetIntegrationTest "${TEST_RESOURCES}/TestHTTPGetSecure.yml" "${TEST_RESOURCES}/") add_test(NAME HttpPostIntegrationTest COMMAND HttpPostIntegrationTest "${TEST_RESOURCES}/TestHTTPPost.yml" "${TEST_RESOURCES}/") http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a330c57a/generateVersion.sh ---------------------------------------------------------------------- diff --git a/generateVersion.sh b/generateVersion.sh new file mode 100755 index 0000000..a0796cb --- /dev/null +++ b/generateVersion.sh @@ -0,0 +1,83 @@ +#!/bin/bash +# 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. + +version=$1 +src_dir=$2 +out_dir=$3 +compiler=$4 +compiler_version=$5 +flags=$6 +extensions=$7 +buildident=$8 + +date=`date +%s` + +if [ -d ${src_dir}/.git ]; then + buildrev=`git log -1 --pretty=format:"%H"` + hostname=`hostname` +else + buildrev="Unknown" +fi + +IFS=';' read -r -a extensions_array <<< "$extensions" + +extension_list="${extension_list} } " + +cat >"$out_dir/agent_version.h" <<EOF +#ifndef AGENT_BUILD_H +#define AGENT_BUILD_H + +#include <vector> + +namespace org { +namespace apache { +namespace nifi { +namespace minifi { + +class AgentBuild { + public: + static constexpr const char* VERSION = "$version"; + static constexpr const char* BUILD_IDENTIFIER = "$buildident"; + static constexpr const char* BUILD_REV = "$buildrev"; + static constexpr const char* BUILD_DATE = "$date"; + static constexpr const char* COMPILER = "$compiler"; + static constexpr const char* COMPILER_VERSION = "$compiler_version"; + static constexpr const char* COMPILER_FLAGS = "$flags"; + static std::vector<std::string> getExtensions() { + static std::vector<std::string> extensions; + if (extensions.empty()){ +EOF + +for EXTENSION in "${extensions_array[@]}" +do +cat <<EOF >> "$out_dir/agent_version.h" + extensions.push_back("${EXTENSION}"); +EOF +done + +cat <<EOF >> "$out_dir/agent_version.h" + } + return extensions; + } +}; + +} /* namespace minifi */ +} /* namespace nifi */ +} /* namespace apache */ +} /* namespace org */ + +#endif /* AGENT_BUILD_H */ +EOF http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a330c57a/libminifi/include/FlowController.h ---------------------------------------------------------------------- diff --git a/libminifi/include/FlowController.h b/libminifi/include/FlowController.h index 2087f81..71d22cb 100644 --- a/libminifi/include/FlowController.h +++ b/libminifi/include/FlowController.h @@ -45,10 +45,10 @@ #include "EventDrivenSchedulingAgent.h" #include "FlowControlProtocol.h" #include "core/Property.h" +#include "core/state/nodes/MetricsBase.h" #include "utils/Id.h" -#include "core/state/metrics/MetricsBase.h" #include "core/state/StateManager.h" - +#include "core/state/nodes/FlowInformation.h" namespace org { namespace apache { namespace nifi { @@ -117,7 +117,7 @@ class FlowController : public core::controller::ControllerServiceProvider, publi // Whether the Flow Controller is start running virtual bool isRunning() { - return running_.load(); + return running_.load() || updating_.load(); } // Whether the Flow Controller has already been initialized (loaded flow XML) @@ -131,7 +131,7 @@ class FlowController : public core::controller::ControllerServiceProvider, publi } // Unload the current flow YAML, clean the root process group and all its children virtual int16_t stop(bool force, uint64_t timeToWait = 0); - virtual int16_t applyUpdate(const std::string &configuration); + virtual int16_t applyUpdate(const std::string &source, const std::string &configuration); virtual int16_t drainRepositories() { return -1; @@ -143,7 +143,7 @@ class FlowController : public core::controller::ControllerServiceProvider, publi virtual int16_t clearConnection(const std::string &connection); - virtual int16_t applyUpdate(const std::shared_ptr<state::Update> &updateController) { + virtual int16_t applyUpdate(const std::string &source, const std::shared_ptr<state::Update> &updateController) { return -1; } // Asynchronous function trigger unloading and wait for a period of time @@ -187,11 +187,11 @@ class FlowController : public core::controller::ControllerServiceProvider, publi } // get version - int getVersion() { + virtual std::string getVersion() { if (root_ != nullptr) - return root_->getVersion(); + return std::to_string( root_->getVersion() ); else - return 0; + return "0"; } /** @@ -299,6 +299,15 @@ class FlowController : public core::controller::ControllerServiceProvider, publi virtual void enableAllControllerServices(); /** + * Retrieves all root response nodes from this source. + * @param metric_vector -- metrics will be placed in this vector. + * @return result of the get operation. + * 0 Success + * 1 No error condition, but cannot obtain lock in timely manner. + * -1 failure + */ + virtual int16_t getResponseNodes(std::vector<std::shared_ptr<state::response::ResponseNode>> &metric_vector, uint16_t metricsClass); + /** * Retrieves all metrics from this source. * @param metric_vector -- metrics will be placed in this vector. * @return result of the get operation. @@ -306,17 +315,23 @@ class FlowController : public core::controller::ControllerServiceProvider, publi * 1 No error condition, but cannot obtain lock in timely manner. * -1 failure */ - virtual int16_t getMetrics(std::vector<std::shared_ptr<state::metrics::Metrics>> &metric_vector, uint16_t metricsClass); + virtual int16_t getMetricsNodes(std::vector<std::shared_ptr<state::response::ResponseNode>> &metric_vector, uint16_t metricsClass); virtual uint64_t getUptime(); + void initializeC2(); + protected: + void loadC2ResponseConfiguration(); + + void loadC2ResponseConfiguration(const std::string &prefix); + + std::shared_ptr<state::response::ResponseNode> loadC2ResponseConfiguration(const std::string &prefix, std::shared_ptr<state::response::ResponseNode>); + // function to load the flow file repo. void loadFlowRepo(); - void initializeC2(); - /** * Initializes flow controller paths. */ @@ -338,6 +353,7 @@ class FlowController : public core::controller::ControllerServiceProvider, publi // FlowFile Repo // Whether it is running std::atomic<bool> running_; + std::atomic<bool> updating_; // conifiguration filename std::string configuration_filename_; @@ -379,17 +395,20 @@ class FlowController : public core::controller::ControllerServiceProvider, publi std::chrono::steady_clock::time_point start_time_; 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::metrics::Metrics>> metrics_; + std::map<std::string, std::shared_ptr<state::response::ResponseNode>> device_information_; // metrics cache - std::map<std::string, std::shared_ptr<state::metrics::Metrics>> component_metrics_; + std::map<std::string, std::shared_ptr<state::response::ResponseNode>> component_metrics_; - std::map<uint8_t, std::vector<std::shared_ptr<state::metrics::Metrics>>>component_metrics_by_id_; + std::map<uint8_t, std::vector<std::shared_ptr<state::response::ResponseNode>>>component_metrics_by_id_; // metrics last run std::chrono::steady_clock::time_point last_metrics_capture_; private: + std::shared_ptr<state::response::FlowVersion> flow_version_; std::shared_ptr<logging::Logger> logger_; std::string serial_number_; static std::shared_ptr<utils::IdGenerator> id_generator_; http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a330c57a/libminifi/include/agent/build_description.h ---------------------------------------------------------------------- diff --git a/libminifi/include/agent/build_description.h b/libminifi/include/agent/build_description.h new file mode 100644 index 0000000..314282c --- /dev/null +++ b/libminifi/include/agent/build_description.h @@ -0,0 +1,101 @@ +/** + * + * 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. + */ +#ifndef BUILD_DESCRPTION_H +#define BUILD_DESCRPTION_H + +#include <vector> +#include "capi/expect.h" + +namespace org { +namespace apache { +namespace nifi { +namespace minifi { + + + +struct ClassDescription { + explicit ClassDescription(std::string name) + : class_name_(name), + support_dynamic_(false) { + } + explicit ClassDescription(std::string name, std::map<std::string, std::string> props, bool dyn) + : class_name_(name), + class_properties_(props), + support_dynamic_(dyn) { + + } + std::string class_name_; + std::map<std::string, std::string> class_properties_; + bool support_dynamic_; +}; + +struct Components { + std::vector<ClassDescription> processors_; + std::vector<ClassDescription> controller_services_; + std::vector<ClassDescription> other_components_; +}; + +class BuildDescription { + public: + + static struct Components getClassDescriptions() { + static struct Components classes; + if (UNLIKELY(IsNullOrEmpty(classes.processors_) && IsNullOrEmpty(classes.controller_services_))) { + for (auto clazz : core::ClassLoader::getDefaultClassLoader().getClasses()) { + + auto lastOfIdx = clazz.find_last_of("::"); + if (lastOfIdx != std::string::npos) { + lastOfIdx++; // if a value is found, increment to move beyond the . + int nameLength = clazz.length() - lastOfIdx; + std::string class_name = clazz.substr(lastOfIdx, nameLength); + + auto obj = core::ClassLoader::getDefaultClassLoader().instantiate(class_name, class_name); + + std::shared_ptr<core::ConfigurableComponent> component = std::dynamic_pointer_cast<core::ConfigurableComponent>(obj); + + ClassDescription description(clazz); + if (nullptr != component) { + + bool is_processor = std::dynamic_pointer_cast<core::Processor>(obj) != nullptr; + bool is_controller_service = LIKELY(is_processor == true) ? false : std::dynamic_pointer_cast<core::controller::ControllerService>(obj) != nullptr; + + component->initialize(); + description.class_properties_ = component->getProperties(); + description.support_dynamic_ = component->supportsDynamicProperties(); + if (is_processor) { + classes.processors_.emplace_back(description); + } else if (is_controller_service) { + classes.controller_services_.emplace_back(description); + } else { + classes.other_components_.emplace_back(description); + } + } + } + } + } + return classes; + } + +}; + +} /* namespace minifi */ +} /* namespace nifi */ +} /* namespace apache */ +} /* namespace org */ + +#endif /* BUILD_DESCRPTION_H */ http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a330c57a/libminifi/include/c2/C2Agent.h ---------------------------------------------------------------------- diff --git a/libminifi/include/c2/C2Agent.h b/libminifi/include/c2/C2Agent.h index 810eede..b902c04 100644 --- a/libminifi/include/c2/C2Agent.h +++ b/libminifi/include/c2/C2Agent.h @@ -24,8 +24,10 @@ #include <memory> #include <mutex> #include <thread> + +#include "../core/state/nodes/MetricsBase.h" #include "core/state/UpdateController.h" -#include "core/state/metrics/MetricsBase.h" +#include "core/state/Value.h" #include "C2Payload.h" #include "C2Protocol.h" #include "io/validation.h" @@ -47,13 +49,12 @@ namespace c2 { * 0 HeartBeat -- RESERVED * 1-255 Defined by the configuration file. */ -class C2Agent : public state::UpdateController, public state::metrics::MetricsSink, public std::enable_shared_from_this<C2Agent> { +class C2Agent : public state::UpdateController, public state::response::ResponseNodeSink, public std::enable_shared_from_this<C2Agent> { public: C2Agent(const std::shared_ptr<core::controller::ControllerServiceProvider> &controller, const std::shared_ptr<state::StateMonitor> &updateSink, const std::shared_ptr<Configure> &configure); virtual ~C2Agent() { - } /** @@ -71,7 +72,14 @@ class C2Agent : public state::UpdateController, public state::metrics::MetricsSi * @param metric metric to set * @param return 0 on success, -1 on failure. */ - virtual int16_t setMetrics(const std::shared_ptr<state::metrics::Metrics> &metric); + virtual int16_t setResponseNodes(const std::shared_ptr<state::response::ResponseNode> &metric); + + /** + * Sets the metric within this sink + * @param metric metric to set + * @param return 0 on success, -1 on failure. + */ + virtual int16_t setMetricsNodes(const std::shared_ptr<state::response::ResponseNode> &metric); int64_t getHeartBeatDelay(){ std::lock_guard<std::mutex> lock(heartbeat_mutex); @@ -80,6 +88,10 @@ class C2Agent : public state::UpdateController, public state::metrics::MetricsSi protected: + void restart_agent(); + + void update_agent(); + /** * Configure the C2 agent */ @@ -91,7 +103,7 @@ class C2Agent : public state::UpdateController, public state::metrics::MetricsSi * @param name name of this metric * @param metrics metrics to include. */ - void serializeMetrics(C2Payload &parent_payload, const std::string &name, const std::vector<state::metrics::MetricResponse> &metrics); + void serializeMetrics(C2Payload &parent_payload, const std::string &name, const std::vector<state::response::SerializedResponseNode> &metrics, bool is_container = false); /** * Extract the payload @@ -139,12 +151,17 @@ class C2Agent : public state::UpdateController, public state::metrics::MetricsSi void handle_describe(const C2ContentResponse &resp); std::timed_mutex metrics_mutex_; - std::map<std::string, std::shared_ptr<state::metrics::Metrics>> metrics_map_; + std::map<std::string, std::shared_ptr<state::response::ResponseNode>> metrics_map_; + + /** + * Device information stored in the metrics format + */ + std::map<std::string, std::shared_ptr<state::response::ResponseNode>> root_response_nodes_; /** * Device information stored in the metrics format */ - std::map<std::string, std::shared_ptr<state::metrics::Metrics>> device_information_; + std::map<std::string, std::shared_ptr<state::response::ResponseNode>> device_information_; // queue mutex std::timed_mutex queue_mutex; @@ -191,6 +208,12 @@ class C2Agent : public state::UpdateController, public state::metrics::MetricsSi std::atomic<C2Protocol*> protocol_; + bool allow_updates_; + + std::string update_command_; + + std::string update_location_; + std::shared_ptr<logging::Logger> logger_; } ; http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a330c57a/libminifi/include/c2/C2Payload.h ---------------------------------------------------------------------- diff --git a/libminifi/include/c2/C2Payload.h b/libminifi/include/c2/C2Payload.h index ca14584..4d4b0ad 100644 --- a/libminifi/include/c2/C2Payload.h +++ b/libminifi/include/c2/C2Payload.h @@ -22,6 +22,7 @@ #include <string> #include <map> #include "core/state/UpdateController.h" +#include "core/state/Value.h" namespace org { namespace apache { @@ -38,7 +39,8 @@ enum Operation { HEARTBEAT, UPDATE, VALIDATE, - CLEAR + CLEAR, + TRANSFER }; enum Direction { @@ -58,6 +60,24 @@ class C2ContentResponse { C2ContentResponse & operator=(const C2ContentResponse &other); + inline bool operator==(const C2ContentResponse &rhs) const { + if (op != rhs.op) + return false; + if (required != rhs.required) + return false; + if (ident != rhs.ident) + return false; + if (name != rhs.name) + return false; + if (operation_arguments != rhs.operation_arguments) + return false; + return true; + } + + inline bool operator!=(const C2ContentResponse &rhs) const { + return !(*this == rhs); + } + Operation op; // determines if the operation is required bool required; @@ -70,7 +90,7 @@ class C2ContentResponse { // name applied to commands std::string name; // commands that correspond with the operation. - std::map<std::string, std::string> operation_arguments; + std::map<std::string, state::response::ValueNode> operation_arguments; // std::vector<std::string> content; }; @@ -147,13 +167,21 @@ class C2Payload : public state::Update { /** * Returns raw data. */ - std::string getRawData() const; + std::vector<char> getRawData() const; /** * Add a nested payload. * @param payload payload to move into this object. */ void addPayload(const C2Payload &&payload); + + bool isContainer() const { + return is_container_; + } + + void setContainer(bool is_container) { + is_container_ = is_container; + } /** * Get nested payloads. */ @@ -162,6 +190,35 @@ class C2Payload : public state::Update { C2Payload &operator=(const C2Payload &&other); C2Payload &operator=(const C2Payload &other); + inline bool operator==(const C2Payload &rhs) const { + if (op_ != rhs.op_) { + return false; + } + if (ident_ != rhs.ident_) { + return false; + } + if (label_ != rhs.label_) { + return false; + } + if (payloads_ != rhs.payloads_) { + return false; + } + if (content_ != rhs.content_) { + return false; + } + if (raw_ != rhs.raw_) { + return false; + } + if (raw_data_ != rhs.raw_data_) { + return false; + } + return true; + } + + inline bool operator!=(const C2Payload &rhs) const { + return !(*this == rhs); + } + protected: // identifier for this payload. @@ -177,10 +234,12 @@ class C2Payload : public state::Update { bool raw_; - std::string raw_data_; + std::vector<char> raw_data_; bool isResponse; + bool is_container_; + }; } /* namesapce c2 */ http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a330c57a/libminifi/include/c2/protocols/RESTProtocol.h ---------------------------------------------------------------------- diff --git a/libminifi/include/c2/protocols/RESTProtocol.h b/libminifi/include/c2/protocols/RESTProtocol.h index ed6d3eb..53e3f4f 100644 --- a/libminifi/include/c2/protocols/RESTProtocol.h +++ b/libminifi/include/c2/protocols/RESTProtocol.h @@ -49,7 +49,8 @@ namespace c2 { */ class RESTProtocol { public: - RESTProtocol() { + RESTProtocol() + : minimize_updates_(false) { } @@ -59,10 +60,12 @@ class RESTProtocol { protected: + virtual rapidjson::Value getStringValue(const std::string& value, rapidjson::Document::AllocatorType& alloc); + virtual rapidjson::Value serializeJsonPayload(const C2Payload &payload, rapidjson::Document::AllocatorType &alloc); virtual std::string serializeJsonRootPayload(const C2Payload& payload); - + virtual void mergePayloadContent(rapidjson::Value &target, const C2Payload &payload, rapidjson::Document::AllocatorType &alloc); virtual const C2Payload parseJsonResponse(const C2Payload &payload, const std::vector<char> &response); @@ -71,6 +74,11 @@ class RESTProtocol { virtual Operation stringToOperation(const std::string str); + bool containsPayload(const C2Payload &o); + + std::mutex update_mutex_; + bool minimize_updates_; + std::map<std::string, C2Payload> nested_payloads_; }; } /* namesapce c2 */ http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a330c57a/libminifi/include/core/ClassLoader.h ---------------------------------------------------------------------- diff --git a/libminifi/include/core/ClassLoader.h b/libminifi/include/core/ClassLoader.h index 9297eef..6c17493 100644 --- a/libminifi/include/core/ClassLoader.h +++ b/libminifi/include/core/ClassLoader.h @@ -226,6 +226,28 @@ class ClassLoader { loaded_factories_.insert(std::make_pair(name, std::move(factory))); } + std::vector<std::string> getGroups() { + std::vector<std::string> groups; + std::lock_guard<std::mutex> lock(internal_mutex_); + for (auto & resource : loaded_factories_) { + groups.push_back(resource.first); + } + return groups; + } + + std::vector<std::string> getClasses() { + std::vector<std::string> groups; + std::lock_guard<std::mutex> lock(internal_mutex_); + for (auto & resource : loaded_factories_) { + if (nullptr != resource.second) { + auto classes = resource.second->getClassNames(); + groups.insert(groups.end(), classes.begin(), classes.end()); + }else{ + } + } + return groups; + } + /** * Instantiate object based on class_name * @param class_name class to create http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a330c57a/libminifi/include/core/ConfigurableComponent.h ---------------------------------------------------------------------- diff --git a/libminifi/include/core/ConfigurableComponent.h b/libminifi/include/core/ConfigurableComponent.h index 630ba59..e83e30e 100644 --- a/libminifi/include/core/ConfigurableComponent.h +++ b/libminifi/include/core/ConfigurableComponent.h @@ -129,12 +129,14 @@ class __attribute__((visibility("default"))) ConfigurableComponent { /** * Invoked anytime a static property is modified */ - virtual void onPropertyModified(const Property &old_property, const Property &new_property) {} + virtual void onPropertyModified(const Property &old_property, const Property &new_property) { + } /** * Invoked anytime a dynamic property is modified. */ - virtual void onDynamicPropertyModified(const Property &old_property, const Property &new_property) {} + virtual void onDynamicPropertyModified(const Property &old_property, const Property &new_property) { + } /** * Provides all dynamic property keys that have been set. @@ -143,8 +145,19 @@ class __attribute__((visibility("default"))) ConfigurableComponent { */ std::vector<std::string> getDynamicPropertyKeys(); + /** + * Returns a vector all properties + * + * @return map of property keys to their descriptions. + */ + std::map<std::string, std::string> getProperties(); + virtual ~ConfigurableComponent(); + virtual void initialize() { + + } + protected: /** http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a330c57a/libminifi/include/core/Core.h ---------------------------------------------------------------------- diff --git a/libminifi/include/core/Core.h b/libminifi/include/core/Core.h index 3ba051d..8109cd1 100644 --- a/libminifi/include/core/Core.h +++ b/libminifi/include/core/Core.h @@ -153,7 +153,7 @@ class CoreComponent { * Return the UUID string * @param constant reference to the UUID str */ - const std::string & getUUIDStr() { + const std::string & getUUIDStr() const { return uuidStr_; } http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a330c57a/libminifi/include/core/ProcessGroup.h ---------------------------------------------------------------------- diff --git a/libminifi/include/core/ProcessGroup.h b/libminifi/include/core/ProcessGroup.h index 737d6de..b5d437a 100644 --- a/libminifi/include/core/ProcessGroup.h +++ b/libminifi/include/core/ProcessGroup.h @@ -153,7 +153,7 @@ class ProcessGroup { } // getVersion int getVersion() { - return version_; + return config_version_; } // Start Processing void startProcessing(TimerDrivenSchedulingAgent *timeScheduler, EventDrivenSchedulingAgent *eventScheduler); @@ -216,7 +216,7 @@ class ProcessGroup { // Processor Group Name std::string name_; // version - int version_; + int config_version_; // Process Group Type ProcessGroupType type_; // Processors (ProcessNode) inside this process group which include Input/Output Port, Remote Process Group input/Output port http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/a330c57a/libminifi/include/core/Property.h ---------------------------------------------------------------------- diff --git a/libminifi/include/core/Property.h b/libminifi/include/core/Property.h index 788e452..ec228ec 100644 --- a/libminifi/include/core/Property.h +++ b/libminifi/include/core/Property.h @@ -82,7 +82,7 @@ class Property { // Get Name for the property std::string getName() const; // Get Description for the property - std::string getDescription(); + std::string getDescription() const; // Get value for the property std::string getValue() const; std::vector<std::string> &getValues();
