Use same input if for policy if they have same tap and config

.github/workflows/build.yml

@@ -14,6 +14,7 @@ on:
 env:
   # Customize the CMake build type here (Release, Debug, RelWithDebInfo, etc.)
   BUILD_TYPE: Release
+  CTEST_OUTPUT_ON_FAILURE: 1
 
 jobs:
   build:

.github/workflows/static_build.yml

@@ -14,6 +14,7 @@ on:
 env:
   # Customize the CMake build type here (Release, Debug, RelWithDebInfo, etc.)
   BUILD_TYPE: Release
+  CTEST_OUTPUT_ON_FAILURE: 1
 
 jobs:
   build:

src/AbstractModule.cpp

@@ -16,9 +16,6 @@ void AbstractRunnableModule::common_info_json(json &j) const
 {
     j["module"]["name"] = _name;
     j["module"]["running"] = _running.load();
-    if (_policy) {
-        j["module"]["policy"]["name"] = _policy->name();
-    }
     config_json(j["module"]["config"]);
 }
 

src/AbstractModule.h

@@ -61,8 +61,6 @@ class AbstractRunnableModule : public AbstractModule
 protected:
     std::atomic_bool _running = false;
 
-    Policy *_policy = nullptr;
-
     void common_info_json(json &j) const;
 
 public:
@@ -71,16 +69,6 @@ class AbstractRunnableModule : public AbstractModule
     {
     }
 
-    void set_policy(Policy *policy)
-    {
-        _policy = policy;
-    }
-
-    const Policy *policy() const
-    {
-        return _policy;
-    }
-
     virtual ~AbstractRunnableModule(){};
 
     virtual void start() = 0;

src/Configurable.h

@@ -11,6 +11,7 @@
 #include <nlohmann/json.hpp>
 #include <regex>
 #include <shared_mutex>
+#include <sstream>
 #include <string>
 #include <unordered_map>
 #include <variant>
@@ -161,6 +162,41 @@ class Configurable
             }
         }
     }
+
+    const std::string config_hash()
+    {
+        std::shared_lock lock(_config_mutex);
+        std::vector<std::string> key_values;
+        for (const auto &[key, value] : _config) {
+            std::visit([&key_values, key = key](auto &&arg) {
+                std::string data;
+                data += key;
+                using T = std::decay_t<decltype(arg)>;
+                if constexpr (std::is_same_v<T, StringList>) {
+                    auto temp_list = arg;
+                    std::sort(temp_list.begin(), temp_list.end());
+                    for (const auto &s : temp_list) {
+                        data += s;
+                    }
+                } else if constexpr (std::is_same_v<T, std::string>) {
+                    data += arg;
+                } else {
+                    data += std::to_string(arg);
+                }
+                key_values.push_back(data);
+            },
+                value);
+        }
+        std::sort(key_values.begin(), key_values.end());
+        std::string hash;
+        for (auto &data : key_values) {
+            hash += data;
+        }
+        auto h1 = std::hash<std::string>{}(hash);
+        std::stringstream string_stream;
+        string_stream << std::hex << h1;
+        return string_stream.str();
+    }
 };
 
 class Config : public Configurable

src/InputStream.h

@@ -11,6 +11,8 @@ namespace visor {
 
 class InputStream : public AbstractRunnableModule
 {
+    mutable std::shared_mutex _input_mutex;
+    std::vector<const Policy *> _policies;
 
 public:
     InputStream(const std::string &name)
@@ -20,6 +22,24 @@ class InputStream : public AbstractRunnableModule
 
     virtual ~InputStream(){};
 
+    void add_policy(const Policy *policy)
+    {
+        std::unique_lock lock(_input_mutex);
+        _policies.push_back(policy);
+    }
+
+    void remove_policy(const Policy *policy)
+    {
+        std::unique_lock lock(_input_mutex);
+        _policies.erase(std::remove(_policies.begin(), _policies.end(), policy), _policies.end());
+    }
+
+    size_t policies_count() const
+    {
+        std::unique_lock lock(_input_mutex);
+        return _policies.size();
+    }
+
     virtual size_t consumer_count() const = 0;
 
     void common_info_json(json &j) const
@@ -31,4 +51,3 @@ class InputStream : public AbstractRunnableModule
 };
 
 }
-

src/Policies.cpp

@@ -118,30 +118,39 @@ std::vector<Policy *> PolicyManager::load(const YAML::Node &policy_yaml)
                 throw PolicyException(fmt::format("invalid input config for tap '{}': {}", tap_name, e.what()));
             }
         }
+        // TODO separate config and filter. for now, they merge
+        tap_filter.config_merge(tap_config);
 
-        // Create Policy
-        auto policy = std::make_unique<Policy>(policy_name, tap);
-        // if and only if policy succeeds, we will return this in result set
-        Policy *policy_ptr = policy.get();
+        std::string input_stream_module_name = tap_filter.config_hash();
+        input_stream_module_name.insert(0, tap->name() + "-");
 
-        // Instantiate stream from tap
         std::unique_ptr<InputStream> input_stream;
-        std::string input_stream_module_name;
-        try {
-            spdlog::get("visor")->info("policy [{}]: instantiating Tap: {}", policy_name, tap_name);
-            // TODO separate config and filter. for now, they merge
-            tap_filter.config_merge(tap_config);
-            input_stream = tap->instantiate(policy.get(), &tap_filter);
-            // ensure tap input type matches policy input tap
-            if (input_node["input_type"].as<std::string>() != tap->input_plugin()->plugin()) {
-                throw PolicyException(fmt::format("input_type for policy specified tap '{}' doesn't match tap's defined input type: {}/{}", tap_name, input_node["input_type"].as<std::string>(), tap->input_plugin()->plugin()));
+        InputStream *input_ptr;
+        std::unique_lock<std::shared_mutex> input_lock;
+        if (_registry->input_manager()->module_exists(input_stream_module_name)) {
+            spdlog::get("visor")->info("policy [{}]: input stream already exists. reusing: {}", policy_name, input_stream_module_name);
+            auto result_input = _registry->input_manager()->module_get_locked(input_stream_module_name);
+            input_ptr = result_input.module;
+            input_lock = std::move(result_input.lock);
+        } else {
+            // Instantiate stream from tap
+            try {
+                spdlog::get("visor")->info("policy [{}]: instantiating Tap: {}", policy_name, tap_name);
+                input_stream = tap->instantiate(&tap_filter, input_stream_module_name);
+                // ensure tap input type matches policy input tap
+                if (input_node["input_type"].as<std::string>() != tap->input_plugin()->plugin()) {
+                    throw PolicyException(fmt::format("input_type for policy specified tap '{}' doesn't match tap's defined input type: {}/{}", tap_name, input_node["input_type"].as<std::string>(), tap->input_plugin()->plugin()));
+                }
+                input_ptr = input_stream.get();
+            } catch (std::runtime_error &e) {
+                throw PolicyException(fmt::format("unable to instantiate tap '{}': {}", tap_name, e.what()));
             }
-            input_stream_module_name = input_stream->name();
-        } catch (std::runtime_error &e) {
-            throw PolicyException(fmt::format("unable to instantiate tap '{}': {}", tap_name, e.what()));
         }
-        policy->set_input_stream(input_stream.get());
-
+        // Create Policy
+        auto policy = std::make_unique<Policy>(policy_name, tap);
+        // if and only if policy succeeds, we will return this in result set
+        Policy *policy_ptr = policy.get();
+        policy->set_input_stream(input_ptr);
         // Handler Section
         if (!it->second["handlers"] || !it->second["handlers"].IsMap()) {
             throw PolicyException("missing or invalid handler configuration at key 'handlers'");
@@ -224,7 +233,7 @@ std::vector<Policy *> PolicyManager::load(const YAML::Node &policy_yaml)
 
             std::unique_ptr<StreamHandler> handler_module;
             if (!handler_sequence || handler_modules.empty()) {
-                handler_module = handler_plugin->second->instantiate(policy_name + "-" + handler_module_name, input_stream.get(), &handler_config);
+                handler_module = handler_plugin->second->instantiate(policy_name + "-" + handler_module_name, input_ptr, &handler_config);
             } else {
                 // for sequence, use only previous handler
                 handler_module = handler_plugin->second->instantiate(policy_name + "-" + handler_module_name, nullptr, &handler_config, handler_modules.back().get());
@@ -249,7 +258,9 @@ std::vector<Policy *> PolicyManager::load(const YAML::Node &policy_yaml)
             throw PolicyException(fmt::format("policy [{}] creation failed (policy): {}", policy_name, e.what()));
         }
         try {
-            _registry->input_manager()->module_add(std::move(input_stream));
+            if (input_stream) {
+                _registry->input_manager()->module_add(std::move(input_stream));
+            }
         } catch (ModuleException &e) {
             // note that if this call excepts, we are in an unknown state and the exception will propagate
             module_remove(policy_name);
@@ -276,6 +287,7 @@ std::vector<Policy *> PolicyManager::load(const YAML::Node &policy_yaml)
         }
 
         // success
+        input_ptr->add_policy(policy_ptr);
         result.push_back(policy_ptr);
     }
     return result;
@@ -299,7 +311,10 @@ void PolicyManager::remove_policy(const std::string &name)
     for (const auto &name : module_names) {
         _registry->handler_manager()->module_remove(name);
     }
-    _registry->input_manager()->module_remove(input_name);
+
+    if (!policy->input_stream()->policies_count()) {
+        _registry->input_manager()->module_remove(input_name);
+    }
 
     _map.erase(name);
 }
@@ -335,9 +350,14 @@ void Policy::stop()
     }
     spdlog::get("visor")->info("policy [{}]: stopping", _name);
     if (_input_stream->running()) {
-        spdlog::get("visor")->info("policy [{}]: stopping input instance: {}", _name, _input_stream->name());
-        _input_stream->stop();
+        if (_input_stream->policies_count() <= 1) {
+            spdlog::get("visor")->info("policy [{}]: stopping input instance: {}", _name, _input_stream->name());
+            _input_stream->stop();
+        } else {
+            spdlog::get("visor")->info("policy [{}]: input instance {} not stopped because it is in use by another policy.", _name, _input_stream->name());
+        }
     }
+    _input_stream->remove_policy(this);
     for (auto &mod : _modules) {
         if (mod->running()) {
             spdlog::get("visor")->info("policy [{}]: stopping handler instance: {}", _name, mod->name());

src/Taps.cpp

@@ -58,13 +58,13 @@ void TapManager::load(const YAML::Node &tap_yaml, bool strict)
     }
 }
 
-std::unique_ptr<InputStream> Tap::instantiate(Policy *policy, const Configurable *filter_config)
+std::unique_ptr<InputStream> Tap::instantiate(const Configurable *filter_config, std::string input_name)
 {
     Config c;
     c.config_merge(dynamic_cast<const Configurable &>(*this));
     c.config_merge(*filter_config);
-    auto module = _input_plugin->instantiate(_name + "-" + policy->name(), &c);
-    module->set_policy(policy);
+    auto module = _input_plugin->instantiate(input_name, &c);
+
     return module;
 }
 

src/Taps.h

@@ -28,7 +28,7 @@ class Tap : public AbstractModule
         assert(input_plugin);
     }
 
-    std::unique_ptr<InputStream> instantiate(Policy *policy, const Configurable *filter_config);
+    std::unique_ptr<InputStream> instantiate(const Configurable *filter_config, std::string input_name);
 
     const InputModulePlugin *input_plugin() const
     {

src/tests/test_policies.cpp

@@ -90,6 +90,31 @@ version: "1.0"
             type: net
 )";
 
+auto policies_config_same_input = R"(
+version: "1.0"
+
+visor:
+  policies:
+    # policy name and description
+    same_input:
+      kind: collection
+      input:
+        # this must reference a tap name, or application of the policy will fail
+        tap: anycast
+        input_type: mock
+        config:
+          sample: value
+        filter:
+          bpf: "tcp or udp"
+      handlers:
+        window_config:
+          num_periods: 5
+          deep_sample_rate: 100
+        modules:
+          net:
+            type: net
+)";
+
 auto policies_config_bad1 = R"(
 visor:
   policies:
@@ -335,7 +360,7 @@ TEST_CASE("Policies", "[policies]")
         REQUIRE(registry.policy_manager()->module_exists("default_view"));
         auto [policy, lock] = registry.policy_manager()->module_get_locked("default_view");
         CHECK(policy->name() == "default_view");
-        CHECK(policy->input_stream()->name() == "anycast-default_view");
+        CHECK(policy->input_stream()->name().find("anycast-") != std::string::npos);
         CHECK(policy->input_stream()->config_get<std::string>("bpf") == "tcp or udp"); // TODO this will move to filter member variable
         CHECK(policy->input_stream()->config_get<std::string>("sample") == "value");
         CHECK(policy->modules()[0]->name() == "default_view-default_net");
@@ -354,7 +379,6 @@ TEST_CASE("Policies", "[policies]")
         CoreRegistry registry;
         registry.start(nullptr);
         YAML::Node config_file = YAML::Load(policies_config_hseq);
-
         CHECK(config_file["visor"]["policies"]);
         CHECK(config_file["visor"]["policies"].IsMap());
         REQUIRE_NOTHROW(registry.tap_manager()->load(config_file["visor"]["taps"], true));
@@ -363,7 +387,7 @@ TEST_CASE("Policies", "[policies]")
         REQUIRE(registry.policy_manager()->module_exists("default_view"));
         auto [policy, lock] = registry.policy_manager()->module_get_locked("default_view");
         CHECK(policy->name() == "default_view");
-        CHECK(policy->input_stream()->name() == "anycast-default_view");
+        CHECK(policy->input_stream()->name().find("anycast-") != std::string::npos);
         CHECK(policy->modules()[0]->name() == "default_view-default_dns");
         CHECK(policy->modules()[1]->name() == "default_view-default_net");
         CHECK(policy->input_stream()->running());
@@ -519,9 +543,17 @@ TEST_CASE("Policies", "[policies]")
         registry.handler_manager()->module_add(std::move(mod));
         REQUIRE_THROWS_WITH(registry.policy_manager()->load(config_file["visor"]["policies"]), "policy [default_view-default_net] creation failed (handler: default_view): module name 'default_view-default_net' already exists");
 
+        auto input_node = config_file["visor"]["policies"]["default_view"]["input"];
+        Config input_filter;
+        input_filter.config_set_yaml(input_node["filter"]);
+        Config input_config;
+        input_config.config_set_yaml(input_node["config"]);
+        input_filter.config_merge(input_config);
+        auto hash = input_filter.config_hash();
+
         // ensure the modules were rolled back
         REQUIRE(!registry.policy_manager()->module_exists("default_view"));
-        REQUIRE(!registry.input_manager()->module_exists("anycast-default_view"));
+        REQUIRE(!registry.input_manager()->module_exists("anycast-" + hash));
     }
     SECTION("Good Config, test stop()")
     {
@@ -585,4 +617,35 @@ TEST_CASE("Policies", "[policies]")
         new_lock.unlock();
         REQUIRE_NOTHROW(registry.policy_manager()->remove_policy("default_view"));
     }
+
+    SECTION("Good Config, policies with same tap and input")
+    {
+        CoreRegistry registry;
+        registry.start(nullptr);
+        YAML::Node config_file = YAML::Load(policies_config);
+
+        CHECK(config_file["visor"]["policies"]);
+        CHECK(config_file["visor"]["policies"].IsMap());
+
+        REQUIRE_NOTHROW(registry.tap_manager()->load(config_file["visor"]["taps"], true));
+        REQUIRE_NOTHROW(registry.policy_manager()->load(config_file["visor"]["policies"]));
+
+        REQUIRE(registry.policy_manager()->module_exists("default_view"));
+        auto [policy, lock] = registry.policy_manager()->module_get_locked("default_view");
+        CHECK(policy->name() == "default_view");
+        CHECK(policy->input_stream()->name().find("anycast-") != std::string::npos);
+        lock.unlock();
+
+        YAML::Node config_file2 = YAML::Load(policies_config_same_input);
+        CHECK(config_file2["visor"]["policies"]);
+        CHECK(config_file2["visor"]["policies"].IsMap());
+
+        REQUIRE_NOTHROW(registry.policy_manager()->load(config_file2["visor"]["policies"]));
+
+        REQUIRE(registry.policy_manager()->module_exists("same_input"));
+        auto [policy2, lock2] = registry.policy_manager()->module_get_locked("same_input");
+        CHECK(policy2->name() == "same_input");
+        CHECK(policy2->input_stream()->name() == policy->input_stream()->name());
+        lock2.unlock();
+    }
 }