fix(policies): start a policy's handlers before its new inputs

src/Policies.cpp

@@ -81,7 +81,6 @@ std::vector<Policy *> PolicyManager::load(const YAML::Node &policy_yaml, bool si
             }
         }
 
-        std::vector<std::string> added_resources_policies;
         std::vector<std::string> added_inputs;
         std::vector<std::string> added_handlers;
         try {
@@ -140,18 +139,8 @@ std::vector<Policy *> PolicyManager::load(const YAML::Node &policy_yaml, bool si
                 input_ptr->add_policy(policy_ptr);
             }
 
-            for (auto &p : added_inputs) {
-                auto [input_ptr, input_lock] = _registry->input_manager()->module_get_locked(p);
-                auto [resource_name, module_name] = create_resources_policy(policy_name, input_ptr, window_config);
-                added_resources_policies.push_back(resource_name);
-                added_handlers.push_back(module_name);
-            }
-
         } catch (std::runtime_error &e) {
             // failed to create policy
-            for (auto &p : added_resources_policies) {
-                module_remove(p);
-            }
             for (auto &m : added_handlers) {
                 _registry->handler_manager()->module_remove(m);
             }
@@ -161,10 +150,22 @@ std::vector<Policy *> PolicyManager::load(const YAML::Node &policy_yaml, bool si
             throw;
         }
 
+        std::vector<std::string> added_resources_policies;
         try {
+            // the policy starts its handlers before its inputs, so start it before the resources policies:
+            // an input that reads all its data as it starts, such as a pcap_file or dnstap_file, would
+            // otherwise be consumed before the policy's handlers are running
             policy_ptr->start();
+            for (auto &p : added_inputs) {
+                auto [input_ptr, input_lock] = _registry->input_manager()->module_get_locked(p);
+                auto [resource_name, module_name] = create_resources_policy(input_ptr, window_config);
+                added_resources_policies.push_back(resource_name);
+                added_handlers.push_back(module_name);
+            }
             module_add(std::move(policy));
         } catch (std::runtime_error &e) {
+            // undoes a completed start; Policy::start undoes a partial one itself
+            policy_ptr->stop();
             for (auto &p : added_resources_policies) {
                 module_remove(p);
             }
@@ -183,7 +184,7 @@ std::vector<Policy *> PolicyManager::load(const YAML::Node &policy_yaml, bool si
     return result;
 }
 
-std::pair<std::string, std::string> PolicyManager::create_resources_policy(const std::string &policy_name, InputStream *input, const Config &window_config)
+std::pair<std::string, std::string> PolicyManager::create_resources_policy(InputStream *input, const Config &window_config)
 {
     auto resources_handler_plugin = _registry->handler_plugins().find(std::make_pair("input_resources", CoreRegistry::DEFAULT_HANDLER_PLUGIN_VERSION));
     if (resources_handler_plugin == _registry->handler_plugins().end()) {
@@ -204,9 +205,9 @@ std::pair<std::string, std::string> PolicyManager::create_resources_policy(const
     try {
         resources_policy->start();
         module_add(std::move(resources_policy));
-    } catch (std::runtime_error &e) {
+    } catch (const std::runtime_error &) {
         resources_module->stop();
-        throw PolicyException(fmt::format("policy [{}] failed to start: {}", policy_name, e.what()));
+        throw;
     }
 
     _registry->handler_manager()->module_add(std::move(resources_module));
@@ -306,15 +307,25 @@ void Policy::start()
     // configurations
     _merge_like_handlers = (config_exists("merge_like_handlers") && config_get<bool>("merge_like_handlers"));
 
-    for (auto &mod : _modules) {
-        spdlog::get("visor")->debug("policy [{}]: starting handler instance: {}", _name, mod->name());
-        mod->start();
-    }
-    // start input stream _after_ modules, since input stream will create a new thread and we need to catch any startup errors
-    // from handlers in the same thread we are starting the policy from
-    for (auto &input : _input_streams) {
-        spdlog::get("visor")->debug("policy [{}]: starting input instance: {}", _name, input->name());
-        input->start();
+    try {
+        for (auto &mod : _modules) {
+            spdlog::get("visor")->debug("policy [{}]: starting handler instance: {}", _name, mod->name());
+            mod->start();
+        }
+        // start input stream _after_ modules, since input stream will create a new thread and we need to catch any startup errors
+        // from handlers in the same thread we are starting the policy from
+        for (auto &input : _input_streams) {
+            spdlog::get("visor")->debug("policy [{}]: starting input instance: {}", _name, input->name());
+            input->start();
+        }
+    } catch (...) {
+        // undo a partial start: in a sequence, a handler cannot be removed while the next one still runs connected to it
+        for (auto &mod : _modules) {
+            if (mod->running()) {
+                mod->stop();
+            }
+        }
+        throw;
     }
     _running = true;
 }

src/Policies.h

@@ -151,7 +151,7 @@ class PolicyManager : public AbstractManager<Policy>
 
     std::vector<Policy *> load_from_str(const std::string &str);
     std::vector<Policy *> load(const YAML::Node &tap_yaml, bool single = false);
-    std::pair<std::string, std::string> create_resources_policy(const std::string &policy_name, InputStream *input, const Config &window_config);
+    std::pair<std::string, std::string> create_resources_policy(InputStream *input, const Config &window_config);
     void remove_policy(const std::string &name);
     // Adds one scope per policy with a window to push, and returns whether any
     // policy had one, so the exporter sends nothing when none has.

src/tests/test_policies.cpp

@@ -797,6 +797,50 @@ version: "1.0"
        input_type: mock
 )";
 
+auto policies_config_pcap_file = R"(
+version: "1.0"
+
+visor:
+  taps:
+    recorded:
+      input_type: pcap
+      config:
+        pcap_file: tests/fixtures/dns_ipv4_udp.pcap
+  policies:
+    recorded_view:
+      kind: collection
+      input:
+        tap: recorded
+        input_type: pcap
+      handlers:
+        modules:
+          recorded_net:
+            type: net
+)";
+
+auto policies_config_pcap_file_missing_hseq = R"(
+version: "1.0"
+
+visor:
+  taps:
+    broken:
+      input_type: pcap
+      config:
+        pcap_file: tests/fixtures/does_not_exist.pcap
+  policies:
+    broken_view:
+      kind: collection
+      input:
+        tap: broken
+        input_type: pcap
+      handlers:
+        modules:
+          - broken_dns:
+            type: dns
+          - broken_net:
+            type: net
+)";
+
 TEST_CASE("Policies", "[policies]")
 {
 
@@ -888,6 +932,33 @@ TEST_CASE("Policies", "[policies]")
         CHECK(policy->modules()[1]->running());
     }
 
+    SECTION("Good Config pcap_file input feeds the policy's handlers")
+    {
+        CoreRegistry registry;
+        visor::load_builtin_plugins(registry);
+        registry.start(nullptr);
+        YAML::Node config_file = YAML::Load(policies_config_pcap_file);
+        REQUIRE_NOTHROW(registry.tap_manager()->load(config_file["visor"]["taps"]));
+        REQUIRE_NOTHROW(registry.policy_manager()->load(config_file["visor"]["policies"]));
+
+        // The file is read while the policy loads, so the policy's handlers
+        // must be running before its input starts.
+        std::string resources_name;
+        {
+            auto [policy, lock] = registry.policy_manager()->module_get_locked("recorded_view");
+            auto net_handler = dynamic_cast<handler::net::NetStreamHandler *>(policy->modules()[0]);
+            REQUIRE(net_handler);
+            auto event_data = net_handler->metrics()->bucket(0)->event_data_locked();
+            CHECK(event_data.num_events->value() == 140);
+            CHECK(net_handler->metrics()->start_tstamp().tv_sec == 1567706414);
+            resources_name = policy->input_stream().back()->name() + "-resources";
+        }
+
+        REQUIRE(registry.policy_manager()->module_exists(resources_name));
+        auto [resources, lock] = registry.policy_manager()->module_get_locked(resources_name);
+        CHECK(resources->running());
+    }
+
     // TODO multiple collection policies in the same yaml
 
     SECTION("Duplicate")
@@ -1000,6 +1071,26 @@ TEST_CASE("Policies", "[policies]")
 
         REQUIRE_NOTHROW(registry.tap_manager()->load_from_str(policies_config_bad4));
         REQUIRE_THROWS_WITH(registry.policy_manager()->load_from_str(policies_config_bad4), "policy [default_view] failed to start: mock error on start");
+        CHECK_FALSE(registry.policy_manager()->module_exists("default_view"));
+        REQUIRE_THROWS_WITH(registry.policy_manager()->load_from_str(policies_config_bad4), "policy [default_view] failed to start: mock error on start");
+    }
+
+    SECTION("Bad Config: exception on input start with sequence modules")
+    {
+        CoreRegistry registry;
+        visor::load_builtin_plugins(registry);
+        registry.start(nullptr);
+        YAML::Node config_file = YAML::Load(policies_config_pcap_file_missing_hseq);
+        REQUIRE_NOTHROW(registry.tap_manager()->load(config_file["visor"]["taps"]));
+
+        // the handlers started before the input failed, and each one consumes the one before it, so they must
+        // be stopped before they can be removed; a retry must fail the same way, not as a name clash
+        for (int attempt = 0; attempt < 2; ++attempt) {
+            REQUIRE_THROWS_WITH(registry.policy_manager()->load(config_file["visor"]["policies"]), "policy [broken_view] failed to start: Cannot open pcap/pcapng file");
+        }
+        CHECK_FALSE(registry.policy_manager()->module_exists("broken_view"));
+        CHECK_FALSE(registry.handler_manager()->module_exists("broken_view-broken-broken_dns"));
+        CHECK_FALSE(registry.handler_manager()->module_exists("broken_view-broken-broken_net"));
     }
 
     SECTION("Bad Config: mis-matched input_type on tap")