MINIFICPP-2314 - Send asset state hash in heartbeat, implement c2 asset sync

C2.md

@@ -66,9 +66,10 @@ be requested via C2 DESCRIBE manifest command.
     #   DeviceInfoNode: basic info about the system (OS, number of cores etc)
     #   AgentInformation: info about the MiNiFi agent, may include the manifest
     #   FlowInformation: information about the current flow, including queue sizes
+    #   AssetInformation: the state of the asset directory managed by the agent
     #   ConfigurationChecksums: hashes of the configuration files; can be used to detect unexpected modifications
     # the default is
-    nifi.c2.root.classes=DeviceInfoNode,AgentInformation,FlowInformation
+    nifi.c2.root.classes=DeviceInfoNode,AgentInformation,FlowInformation,AssetInformation
 
     # control c2 heartbeat interval
     nifi.c2.agent.heartbeat.period=30 sec
@@ -80,8 +81,10 @@ be requested via C2 DESCRIBE manifest command.
     nifi.c2.rest.listener.cacert=<SSL Cert path>
 
     # specify the rest URIs if using RESTSender
-    nifi.c2.rest.url=http://<your-c2-server>/<c2-api-path>/c2-protocol/heartbeat
-    nifi.c2.rest.url.ack=http://<your-c2-server>/<c2-api-path>/c2-protocol/acknowledge
+    nifi.c2.rest.path.base=https://<your-c2-server>/<c2-api-path>
+    # specify either absolute url or relative to the nifi.c2.rest.path.base url for hearbeat and acknowledge
+    nifi.c2.rest.url=/c2-protocol/heartbeat
+    nifi.c2.rest.url.ack=/c2-protocol/acknowledge
     nifi.c2.flow.base.url=http://<your-c2-server>/<c2-api-path>/c2-protocol/
 
     # c2 agent identifier -- must be defined to run agent

CONFIGURE.md

@@ -567,8 +567,13 @@ Additionally, a unique hexadecimal uid.minifi.device.segment should be assigned
 
 ### Asset directory
 
-It is possible to make agents download an asset (triggered through the c2 protocol). The target directory can be specified
-using the `nifi.asset.directory` agent property, which defaults to `${MINIFI_HOME}/asset`.
+There is an asset directory specified using the `nifi.asset.directory` agent property, which defaults to `${MINIFI_HOME}/asset`.
+The files referenced in the `.state` file in this directory are managed by the agent. They are deleted, updated, downloaded
+using the asset sync c2 command. For the asset sync command to work, the c2 server must be made aware of the current state of the
+managed assets by adding the `AssetInformation` entry to the `nifi.c2.root.classes` property.
+
+Files and directories not referenced in the `.state` file are not directly controlled by the agent, although
+it is possible to download an asset (triggered through the c2 protocol) into the asset directory instead.
 
 ### Controller Services
  If you need to reference a controller service in your config.yml file, use the following template. In the example, below, ControllerServiceClass is the name of the class defining the controller Service. ControllerService1

conf/minifi.properties

@@ -84,10 +84,11 @@ nifi.content.repository.class.name=DatabaseContentRepository
 ## base URL of the c2 server,
 ## very likely the same base url of rest urls
 #nifi.c2.flow.base.url=
+#nifi.c2.rest.path.base=
 #nifi.c2.rest.url=
 #nifi.c2.rest.url.ack=
 #nifi.c2.rest.ssl.context.service=
-nifi.c2.root.classes=DeviceInfoNode,AgentInformation,FlowInformation
+nifi.c2.root.classes=DeviceInfoNode,AgentInformation,FlowInformation,AssetInformation
 ## Minimize heartbeat payload size by excluding agent manifest from the heartbeat
 nifi.c2.full.heartbeat=false
 ## heartbeat twice a minute

docker/test/integration/cluster/containers/MinifiContainer.py

@@ -99,7 +99,7 @@ def _create_properties(self):
                 f.write(f"nifi.c2.rest.url=http://minifi-c2-server-{self.feature_context.id}:10090/c2/config/heartbeat\n")
                 f.write(f"nifi.c2.rest.url.ack=http://minifi-c2-server-{self.feature_context.id}:10090/c2/config/acknowledge\n")
                 f.write(f"nifi.c2.flow.base.url=http://minifi-c2-server-{self.feature_context.id}:10090/c2/config/\n")
-                f.write("nifi.c2.root.classes=DeviceInfoNode,AgentInformation,FlowInformation\n")
+                f.write("nifi.c2.root.classes=DeviceInfoNode,AgentInformation,FlowInformation,AssetInformation\n")
                 f.write("nifi.c2.full.heartbeat=false\n")
                 f.write("nifi.c2.agent.class=minifi-test-class\n")
                 f.write("nifi.c2.agent.identifier=minifi-test-id\n")
@@ -109,7 +109,7 @@ def _create_properties(self):
                 f.write(f"nifi.c2.rest.url.ack=https://minifi-c2-server-{self.feature_context.id}:10090/c2/config/acknowledge\n")
                 f.write("nifi.c2.rest.ssl.context.service=SSLContextService\n")
                 f.write(f"nifi.c2.flow.base.url=https://minifi-c2-server-{self.feature_context.id}:10090/c2/config/\n")
-                f.write("nifi.c2.root.classes=DeviceInfoNode,AgentInformation,FlowInformation\n")
+                f.write("nifi.c2.root.classes=DeviceInfoNode,AgentInformation,FlowInformation,AssetInformation\n")
                 f.write("nifi.c2.full.heartbeat=false\n")
                 f.write("nifi.c2.agent.class=minifi-test-class\n")
                 f.write("nifi.c2.agent.identifier=minifi-test-id\n")

encrypt-config/tests/resources/minifi.properties

@@ -55,7 +55,7 @@ nifi.c2.enable=true
 nifi.c2.flow.base.url=http://localhost:10080/c2-server/api
 nifi.c2.rest.url=http://localhost:10080/c2-server/api/c2-protocol/heartbeat
 nifi.c2.rest.url.ack=http://localhost:10080/c2-server/api/c2-protocol/acknowledge
-nifi.c2.root.classes=DeviceInfoNode,AgentInformation,FlowInformation
+nifi.c2.root.classes=DeviceInfoNode,AgentInformation,FlowInformation,AssetInformation
 ## Minimize heartbeat payload size by excluding agent manifest from the heartbeat
 #nifi.c2.full.heartbeat=false
 ## heartbeat 4 times a second

encrypt-config/tests/resources/with-additional-sensitive-props.minifi.properties

@@ -57,7 +57,7 @@ nifi.c2.enable=true
 nifi.c2.flow.base.url=http://localhost:10080/c2-server/api
 nifi.c2.rest.url=http://localhost:10080/c2-server/api/c2-protocol/heartbeat
 nifi.c2.rest.url.ack=http://localhost:10080/c2-server/api/c2-protocol/acknowledge
-nifi.c2.root.classes=DeviceInfoNode,AgentInformation,FlowInformation
+nifi.c2.root.classes=DeviceInfoNode,AgentInformation,FlowInformation,AssetInformation
 ## Minimize heartbeat payload size by excluding agent manifest from the heartbeat
 #nifi.c2.full.heartbeat=false
 ## heartbeat 4 times a second

libminifi/include/FlowController.h

@@ -55,6 +55,7 @@
 #include "TimerDrivenSchedulingAgent.h"
 #include "utils/Id.h"
 #include "utils/file/FileSystem.h"
+#include "utils/file/AssetManager.h"
 #include "core/state/nodes/ResponseNodeLoader.h"
 #include "core/state/MetricsPublisher.h"
 #include "core/state/MetricsPublisherStore.h"
@@ -72,7 +73,8 @@ class FlowController : public core::controller::ForwardingControllerServiceProvi
   FlowController(std::shared_ptr<core::Repository> provenance_repo, std::shared_ptr<core::Repository> flow_file_repo,
                  std::shared_ptr<Configure> configure, std::shared_ptr<core::FlowConfiguration> flow_configuration,
                  std::shared_ptr<core::ContentRepository> content_repo, std::unique_ptr<state::MetricsPublisherStore> metrics_publisher_store = nullptr,
-                 std::shared_ptr<utils::file::FileSystem> filesystem = std::make_shared<utils::file::FileSystem>(), std::function<void()> request_restart = []{});
+                 std::shared_ptr<utils::file::FileSystem> filesystem = std::make_shared<utils::file::FileSystem>(), std::function<void()> request_restart = []{},
+                 std::shared_ptr<utils::file::AssetManager> asset_manager = {});
 
   ~FlowController() override;
 

libminifi/include/c2/C2Agent.h

@@ -43,6 +43,7 @@
 #include "utils/ThreadPool.h"
 #include "utils/file/FileSystem.h"
 #include "C2Utils.h"
+#include "utils/file/AssetManager.h"
 
 namespace org::apache::nifi::minifi::c2 {
 
@@ -62,7 +63,8 @@ class C2Agent : public state::UpdateController {
   C2Agent(std::shared_ptr<Configure> configuration,
           std::weak_ptr<state::response::NodeReporter> node_reporter,
           std::shared_ptr<utils::file::FileSystem> filesystem,
-          std::function<void()> request_restart);
+          std::function<void()> request_restart,
+          std::shared_ptr<utils::file::AssetManager> asset_manager);
 
   void initialize(core::controller::ControllerServiceProvider *controller, state::Pausable *pause_handler, state::StateMonitor* update_sink);
   void start() override;
@@ -131,6 +133,8 @@ class C2Agent : public state::UpdateController {
    */
   void handle_describe(const C2ContentResponse &resp);
 
+  void handle_sync(const C2ContentResponse &resp);
+
 
   enum class UpdateResult {
     NO_UPDATE,
@@ -235,6 +239,8 @@ class C2Agent : public state::UpdateController {
 
   // time point the last time we performed a heartbeat.
   std::chrono::steady_clock::time_point last_run_;
+
+  std::shared_ptr<utils::file::AssetManager> asset_manager_;
 };
 
 }  // namespace org::apache::nifi::minifi::c2

libminifi/include/c2/C2Payload.h

@@ -29,6 +29,7 @@
 #include "utils/Enum.h"
 #include "utils/gsl.h"
 #include "utils/span.h"
+#include "rapidjson/document.h"
 
 namespace org::apache::nifi::minifi::c2 {
 
@@ -43,7 +44,8 @@ enum class Operation : uint8_t {
   clear,
   transfer,
   pause,
-  resume
+  resume,
+  sync
 };
 
 enum class DescribeOperand : uint8_t {
@@ -70,6 +72,10 @@ enum class ClearOperand : uint8_t{
   corecomponentstate
 };
 
+enum class SyncOperand : uint8_t{
+  resource
+};
+
 #define PAYLOAD_NO_STATUS 0
 #define PAYLOAD_SUCCESS 1
 #define PAYLOAD_FAILURE 2
@@ -79,21 +85,65 @@ enum Direction {
   RECEIVE
 };
 
-struct AnnotatedValue : state::response::ValueNode {
-  using state::response::ValueNode::ValueNode;
-  using state::response::ValueNode::operator=;
+class C2Value {
+ public:
+  friend std::ostream& operator<<(std::ostream& out, const C2Value& val);
+
+  C2Value() = default;
+  C2Value(const C2Value& other) {
+    (*this) = other;
+  }
+  C2Value(C2Value&&) = default;
+  template<typename T>
+  requires(std::constructible_from<state::response::ValueNode, T>)
+  C2Value(T&& value) { value_ = state::response::ValueNode{std::forward<T>(value)}; }  // NOLINT(runtime/explicit)
+  C2Value(const rapidjson::Value& json_value) {  // NOLINT(runtime/explicit)
+    value_.emplace<rapidjson::Document>();
+    get<rapidjson::Document>(value_).CopyFrom(json_value, get<rapidjson::Document>(value_).GetAllocator());
+  }
+  C2Value(rapidjson::Document&& json_doc) {  // NOLINT(runtime/explicit)
+    value_ = std::move(json_doc);
+  }
 
-  [[nodiscard]] std::optional<std::reference_wrapper<const AnnotatedValue>> getAnnotation(const std::string& name) const {
-    auto it = annotations.find(name);
-    if (it == annotations.end()) {
-      return {};
+  C2Value& operator=(const C2Value& other) {
+    if (auto* other_val_node = get_if<state::response::ValueNode>(&other.value_)) {
+      value_ = *other_val_node;
+    } else {
+      value_.emplace<rapidjson::Document>();
+      get<rapidjson::Document>(value_).CopyFrom(get<rapidjson::Document>(other.value_), get<rapidjson::Document>(value_).GetAllocator());
     }
-    return std::cref(it->second);
+    return *this;
+  }
+
+  C2Value& operator=(C2Value&&) = default;
+
+
+  bool empty() const {
+    if (auto* val_node = get_if<state::response::ValueNode>(&value_)) {
+      return val_node->empty();
+    }
+    return false;
+  }
+
+  std::string to_string() const {
+    if (auto* val_node = get_if<state::response::ValueNode>(&value_)) {
+      return val_node->to_string();
+    }
+    return std::string{get<rapidjson::Document>(value_).GetString(), get<rapidjson::Document>(value_).GetStringLength()};
+  }
+
+  const rapidjson::Document* json() const {
+    return get_if<rapidjson::Document>(&value_);
+  }
+
+  const state::response::ValueNode* valueNode() const {
+    return get_if<state::response::ValueNode>(&value_);
   }
 
-  friend std::ostream& operator<<(std::ostream& out, const AnnotatedValue& val);
+  bool operator==(const C2Value&) const = default;
 
-  std::map<std::string, AnnotatedValue> annotations;
+ private:
+  std::variant<state::response::ValueNode, rapidjson::Document> value_;
 };
 
 struct C2ContentResponse {
@@ -115,7 +165,7 @@ struct C2ContentResponse {
 
   friend std::ostream& operator<<(std::ostream& out, const C2ContentResponse& response);
 
-  std::optional<std::string> getArgument(const std::string& arg_name) const {
+  std::optional<std::string> getStringArgument(const std::string& arg_name) const {
     if (auto it = operation_arguments.find(arg_name); it != operation_arguments.end()) {
       return it->second.to_string();
     }
@@ -134,7 +184,7 @@ struct C2ContentResponse {
   // name applied to commands
   std::string name;
   // commands that correspond with the operation.
-  std::map<std::string, AnnotatedValue> operation_arguments;
+  std::map<std::string, C2Value> operation_arguments;
 };
 
 /**

libminifi/include/c2/PayloadParser.h

@@ -138,27 +138,20 @@ class PayloadParser {
   }
 
   template<typename T>
-  inline T getAs(const std::string &field) {
+  inline T getAs(const std::string &field, const std::optional<T>& fallback = std::nullopt) {
     for (const auto &cmd : ref_.getContent()) {
-      auto exists = cmd.operation_arguments.find(field);
-      if (exists != cmd.operation_arguments.end()) {
-        return convert_if<T>(exists->second.getValue())();
+      if (auto it = cmd.operation_arguments.find(field); it != cmd.operation_arguments.end()) {
+        if (auto* val_node = it->second.valueNode()) {
+          return convert_if<T>(val_node->getValue())();
+        }
       }
     }
-    std::stringstream ss;
-    ss << "Invalid Field. Could not find " << field << " in " << ref_.getLabel();
-    throw PayloadParseException(ss.str());
-  }
-
-  template<typename T>
-  inline T getAs(const std::string &field, const T &fallback) {
-    for (const auto &cmd : ref_.getContent()) {
-      auto exists = cmd.operation_arguments.find(field);
-      if (exists != cmd.operation_arguments.end()) {
-        return convert_if<T>(exists->second.getValue())();
-      }
+    if (!fallback) {
+      std::stringstream ss;
+      ss << "Invalid Field. Could not find " << field << " in " << ref_.getLabel();
+      throw PayloadParseException(ss.str());
     }
-    return fallback;
+    return fallback.value();
   }
 
   size_t getSize() const {

libminifi/include/c2/PayloadSerializer.h

@@ -42,7 +42,7 @@ class PayloadSerializer {
   /**
    * Static function that serializes the value nodes
    */
-  static void serializeValueNode(state::response::ValueNode &value, std::shared_ptr<io::OutputStream> stream) {
+  static void serializeValueNode(const state::response::ValueNode &value, std::shared_ptr<io::OutputStream> stream) {
     auto base_type = value.getValue();
     if (!base_type) {
       uint8_t type = 0;
@@ -95,7 +95,7 @@ class PayloadSerializer {
         stream->write(size);
         for (auto content : payload_content.operation_arguments) {
           stream->write(content.first);
-          serializeValueNode(content.second, stream);
+          serializeValueNode(*gsl::not_null(content.second.valueNode()), stream);
         }
       }
       if (nested_payload.getNestedPayloads().size() > 0) {
@@ -170,7 +170,7 @@ class PayloadSerializer {
       stream->write(size);
       for (auto content : payload_content.operation_arguments) {
         stream->write(content.first);
-        serializeValueNode(content.second, stream);
+        serializeValueNode(*gsl::not_null(content.second.valueNode()), stream);
       }
     }
     serialize(op, payload, stream);

libminifi/include/c2/protocols/RESTProtocol.h

@@ -43,7 +43,7 @@ class RESTProtocol : public HeartbeatJsonSerializer {
  protected:
   void initialize(core::controller::ControllerServiceProvider* controller, const std::shared_ptr<Configure> &configure);
   void serializeNestedPayload(rapidjson::Value& target, const C2Payload& payload, rapidjson::Document::AllocatorType& alloc) override;
-  static C2Payload parseJsonResponse(const C2Payload &payload, std::span<const std::byte> response);
+  C2Payload parseJsonResponse(const C2Payload &payload, std::span<const std::byte> response);
 
  private:
   bool containsPayload(const C2Payload &o);

libminifi/include/core/state/MetricsPublisherStore.h

@@ -27,13 +27,14 @@
 #include "core/state/nodes/ResponseNodeLoader.h"
 #include "utils/gsl.h"
 #include "core/ProcessGroup.h"
+#include "utils/file/AssetManager.h"
 
 namespace org::apache::nifi::minifi::state {
 
 class MetricsPublisherStore {
  public:
   MetricsPublisherStore(std::shared_ptr<Configure> configuration, const std::vector<std::shared_ptr<core::RepositoryMetricsSource>>& repository_metric_sources,
-    std::shared_ptr<core::FlowConfiguration> flow_configuration);
+    std::shared_ptr<core::FlowConfiguration> flow_configuration, std::shared_ptr<utils::file::AssetManager> asset_manager = nullptr);
   void initialize(core::controller::ControllerServiceProvider* controller, state::StateMonitor* update_sink);
   void loadMetricNodes(core::ProcessGroup* root);
   void clearMetricNodes();

libminifi/include/core/state/nodes/AssetInformation.h

@@ -0,0 +1,42 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#pragma once
+
+#include "core/state/nodes/MetricsBase.h"
+#include "utils/file/AssetManager.h"
+#include "core/logging/Logger.h"
+
+namespace org::apache::nifi::minifi::state::response {
+
+class AssetInformation : public ResponseNode {
+ public:
+  AssetInformation();
+  explicit AssetInformation(std::string_view name, const utils::Identifier& uuid = {}) : ResponseNode(name, uuid) {}
+
+  MINIFIAPI static constexpr const char* Description = "Metric node that defines hash for all asset identifiers";
+
+  void setAssetManager(std::shared_ptr<utils::file::AssetManager> asset_manager);
+
+  std::string getName() const override { return "resourceInfo"; }
+  std::vector<SerializedResponseNode> serialize() override;
+
+ private:
+  std::shared_ptr<utils::file::AssetManager> asset_manager_;
+  std::shared_ptr<core::logging::Logger> logger_;
+};
+
+}  // namespace org::apache::nifi::minifi::state::response