diff --git a/src/Policies.cpp b/src/Policies.cpp index 21aa3ad99..021ed23a3 100644 --- a/src/Policies.cpp +++ b/src/Policies.cpp @@ -81,7 +81,6 @@ std::vector PolicyManager::load(const YAML::Node &policy_yaml, bool si } } - std::vector added_resources_policies; std::vector added_inputs; std::vector added_handlers; try { @@ -140,18 +139,8 @@ std::vector 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,33 @@ std::vector PolicyManager::load(const YAML::Node &policy_yaml, bool si throw; } + std::vector> resources_policies; + std::vector added_resources_policies; try { + // every handler on a new input must be running before the input starts: an input that reads all + // its data as it starts, such as a pcap_file or dnstap_file, is consumed by whichever policy starts + // it. So start the resources handlers, then this policy (its handlers, then its inputs), and only + // then the resources policies, whose handlers and inputs are already running. + for (auto &p : added_inputs) { + auto [input_ptr, input_lock] = _registry->input_manager()->module_get_locked(p); + auto [resources_policy, module_name] = create_resources_policy(input_ptr, window_config); + if (resources_policy) { + added_handlers.push_back(module_name); + resources_policies.push_back(std::move(resources_policy)); + } + } policy_ptr->start(); + for (auto &resources_policy : resources_policies) { + auto resources_policy_ptr = resources_policy.get(); + resources_policy_ptr->start(); + module_add(std::move(resources_policy)); + added_resources_policies.push_back(resources_policy_ptr->name()); + resources_policy_ptr->input_stream().back()->add_policy(resources_policy_ptr); + } 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,36 +195,32 @@ std::vector PolicyManager::load(const YAML::Node &policy_yaml, bool si return result; } -std::pair PolicyManager::create_resources_policy(const std::string &policy_name, InputStream *input, const Config &window_config) +std::pair, 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()) { spdlog::get("visor")->info("input_resources handler not available, not able to create input resources policy for input stream: {}", input->name()); - return std::make_pair(std::string(), std::string()); + return std::make_pair(nullptr, std::string()); } - // create new policy with resources handler for input stream + // create new policy with resources handler for input stream. the handler is started and registered here; + // the caller starts and registers the policy once the input is running auto resources_policy = std::make_unique(input->name() + "-resources"); - auto resources_policy_name = resources_policy->name(); - auto resources_policy_ptr = resources_policy.get(); resources_policy->add_input_stream(input); auto resources_module = resources_handler_plugin->second->instantiate(input->name() + "-resources", input->add_event_proxy(Configurable()), &window_config, nullptr); auto module_name = resources_module->name(); resources_policy->add_module(resources_module.get()); + resources_module->start(); try { - resources_policy->start(); - module_add(std::move(resources_policy)); - } catch (std::runtime_error &e) { + _registry->handler_manager()->module_add(std::move(resources_module)); + } 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)); - // success - input->add_policy(resources_policy_ptr); - return std::make_pair(resources_policy_name, module_name); + return std::make_pair(std::move(resources_policy), module_name); } std::string PolicyManager::_get_policy_name(YAML::const_iterator it) @@ -306,15 +314,25 @@ void Policy::start() // configurations _merge_like_handlers = (config_exists("merge_like_handlers") && config_get("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; } diff --git a/src/Policies.h b/src/Policies.h index ad932df06..507e9416f 100644 --- a/src/Policies.h +++ b/src/Policies.h @@ -151,7 +151,7 @@ class PolicyManager : public AbstractManager std::vector load_from_str(const std::string &str); std::vector load(const YAML::Node &tap_yaml, bool single = false); - std::pair create_resources_policy(const std::string &policy_name, InputStream *input, const Config &window_config); + std::pair, 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. diff --git a/src/tests/test_policies.cpp b/src/tests/test_policies.cpp index e481f4c34..8a66ed6bc 100644 --- a/src/tests/test_policies.cpp +++ b/src/tests/test_policies.cpp @@ -7,6 +7,7 @@ #include "Policies.h" #include "PrometheusSerializer.h" #include "handlers/dns/v1/DnsStreamHandler.h" +#include "handlers/input_resources/InputResourcesStreamHandler.h" #include "handlers/net/v1/NetStreamHandler.h" #include @@ -797,6 +798,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 +933,39 @@ 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 and + // the input's resources handler must both be running before it starts. + std::string resources_name; + { + auto [policy, lock] = registry.policy_manager()->module_get_locked("recorded_view"); + auto net_handler = dynamic_cast(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()); + auto resources_handler = dynamic_cast(resources->modules()[0]); + REQUIRE(resources_handler); + json j; + resources_handler->metrics()->bucket(0)->to_json(j); + CHECK(j["cpu_usage"]["p50"] != nullptr); + CHECK(j["memory_bytes"]["p50"] != nullptr); + } + // TODO multiple collection policies in the same yaml SECTION("Duplicate") @@ -1000,6 +1078,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")