Implement ThreadMonitor for linux systems

src/Policies.cpp

@@ -174,6 +174,18 @@ std::vector<Policy *> PolicyManager::load(const YAML::Node &policy_yaml)
             window_config.config_set<uint64_t>("deep_sample_rate", _default_deep_sample_rate);
         }
 
+        std::unique_ptr<Policy> input_resources_policy;
+        std::unique_ptr<StreamHandler> resources_module;
+        if (input_stream) {
+            // create new policy with resources handler for input stream
+            input_resources_policy = std::make_unique<Policy>(input_stream_module_name + "-resources", tap);
+            input_resources_policy->set_input_stream(input_ptr);
+            auto resources_handler_plugin = _registry->handler_plugins().find("input_resources");
+            resources_module = resources_handler_plugin->second->instantiate(input_stream_module_name + "-resources", input_ptr, &window_config);
+            input_resources_policy->add_module(resources_module.get());
+            input_stream->add_policy(input_resources_policy.get());
+        }
+
         std::vector<std::unique_ptr<StreamHandler>> handler_modules;
         for (YAML::const_iterator h_it = handler_node["modules"].begin(); h_it != handler_node["modules"].end(); ++h_it) {
 
@@ -268,6 +280,9 @@ std::vector<Policy *> PolicyManager::load(const YAML::Node &policy_yaml)
 
         // make sure policy starts before committing
         try {
+            if (input_resources_policy) {
+                input_resources_policy->start();
+            }
             policy->start();
         } catch (std::runtime_error &e) {
             throw PolicyException(fmt::format("policy [{}] failed to start: {}", policy_name, e.what()));
@@ -278,12 +293,16 @@ std::vector<Policy *> PolicyManager::load(const YAML::Node &policy_yaml)
         // roll back during exception ensures no modules have been added to any of the managers
         try {
             module_add(std::move(policy));
+            if (input_resources_policy) {
+                module_add(std::move(input_resources_policy));
+            }
         } catch (ModuleException &e) {
             throw PolicyException(fmt::format("policy [{}] creation failed (policy): {}", policy_name, e.what()));
         }
         try {
-            if (input_stream) {
+            if (input_stream && resources_module) {
                 _registry->input_manager()->module_add(std::move(input_stream));
+                _registry->handler_manager()->module_add(std::move(resources_module));
             }
         } catch (ModuleException &e) {
             // note that if this call excepts, we are in an unknown state and the exception will propagate
@@ -336,8 +355,13 @@ void PolicyManager::remove_policy(const std::string &name)
         _registry->handler_manager()->module_remove(name);
     }
 
-    if (!policy->input_stream()->policies_count()) {
+    if (policy->input_stream()->policies_count() <= 1) {
+        auto resources_name = input_name + "-resources";
+        auto resources_policy = _map[resources_name].get();
+        resources_policy->stop();
+        _registry->handler_manager()->module_remove(resources_name);
         _registry->input_manager()->module_remove(input_name);
+        _map.erase(resources_name);
     }
 
     _map.erase(name);

src/handlers/CMakeLists.txt

@@ -4,5 +4,6 @@ add_subdirectory(dns)
 add_subdirectory(dhcp)
 add_subdirectory(pcap)
 add_subdirectory(mock)
+add_subdirectory(input_resources)
 
 set(VISOR_STATIC_PLUGINS ${VISOR_STATIC_PLUGINS} PARENT_SCOPE)

src/handlers/README.md

@@ -10,3 +10,4 @@ See the individual READMEs for more information:
 * [Mock](mock/)
 * [Network](net/)
 * [PCAP](pcap/)
+* [Resources](resources/)

src/handlers/input_resources/CMakeLists.txt

@@ -0,0 +1,28 @@
+message(STATUS "Handler Module: Input Resources")
+
+set_directory_properties(PROPERTIES CORRADE_USE_PEDANTIC_FLAGS ON)
+
+corrade_add_static_plugin(VisorHandlerInputResources
+        ${CMAKE_CURRENT_BINARY_DIR}
+        InputResourcesHandler.conf
+        InputResourcesHandlerModulePlugin.cpp
+        InputResourcesStreamHandler.cpp
+        )
+add_library(Visor::Handler::InputResources ALIAS VisorHandlerInputResources)
+
+target_include_directories(VisorHandlerInputResources
+        INTERFACE
+        $<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}>
+        )
+
+target_link_libraries(VisorHandlerInputResources
+        PUBLIC
+        Visor::Input::Pcap
+        Visor::Input::Dnstap
+        Visor::Input::Mock
+        Visor::Input::Sflow
+        )
+
+set(VISOR_STATIC_PLUGINS ${VISOR_STATIC_PLUGINS} Visor::Handler::InputResources PARENT_SCOPE)
+
+add_subdirectory(tests)

src/handlers/input_resources/InputResourcesHandler.conf

@@ -0,0 +1,5 @@
+# Aliases
+provides=input_resources
+[data]
+desc=Input Resources analyzer
+type=handler

src/handlers/input_resources/InputResourcesHandlerModulePlugin.cpp

@@ -0,0 +1,31 @@
+/* This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at https://mozilla.org/MPL/2.0/. */
+
+#include "InputResourcesHandlerModulePlugin.h"
+#include "CoreRegistry.h"
+#include "InputResourcesStreamHandler.h"
+#include "HandlerManager.h"
+#include "InputStreamManager.h"
+#include <Corrade/PluginManager/AbstractManager.h>
+#include <nlohmann/json.hpp>
+
+CORRADE_PLUGIN_REGISTER(VisorHandlerInputResources, visor::handler::resources::InputResourcesHandlerModulePlugin,
+    "visor.module.handler/1.0")
+
+namespace visor::handler::resources {
+
+using json = nlohmann::json;
+
+void InputResourcesHandlerModulePlugin::setup_routes(HttpServer *svr)
+{
+}
+std::unique_ptr<StreamHandler> InputResourcesHandlerModulePlugin::instantiate(const std::string &name, InputStream *input_stream, const Configurable *config, StreamHandler *stream_handler)
+{
+    // TODO using config as both window config and module config
+    auto handler_module = std::make_unique<InputResourcesStreamHandler>(name, input_stream, config, stream_handler);
+    handler_module->config_merge(*config);
+    return handler_module;
+}
+
+}

src/handlers/input_resources/InputResourcesHandlerModulePlugin.h

@@ -0,0 +1,24 @@
+/* This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at https://mozilla.org/MPL/2.0/. */
+
+#pragma once
+
+#include "HandlerModulePlugin.h"
+
+namespace visor::handler::resources {
+
+class InputResourcesHandlerModulePlugin : public HandlerModulePlugin
+{
+
+protected:
+    void setup_routes(HttpServer *svr) override;
+
+public:
+    explicit InputResourcesHandlerModulePlugin(Corrade::PluginManager::AbstractManager &manager, const std::string &plugin)
+        : visor::HandlerModulePlugin{manager, plugin}
+    {
+    }
+    std::unique_ptr<StreamHandler> instantiate(const std::string &name, InputStream *input_stream, const Configurable *config, StreamHandler *stream_handler = nullptr) override;
+};
+}

src/handlers/input_resources/InputResourcesStreamHandler.cpp

@@ -0,0 +1,158 @@
+/* This Source Code Form is subject to the terms of the Mozilla Public
+ * License, v. 2.0. If a copy of the MPL was not distributed with this
+ * file, You can obtain one at https://mozilla.org/MPL/2.0/. */
+
+#include "InputResourcesStreamHandler.h"
+
+namespace visor::handler::resources {
+
+InputResourcesStreamHandler::InputResourcesStreamHandler(const std::string &name, InputStream *stream, const Configurable *window_config, StreamHandler *handler)
+    : visor::StreamMetricsHandler<InputResourcesMetricsManager>(name, window_config)
+    , _timer(0)
+    , _timestamp(timespec())
+{
+    if (handler) {
+        throw StreamHandlerException(fmt::format("ResourcesStreamHandler: unsupported upstream chained stream handler {}", handler->name()));
+    }
+
+    assert(stream);
+    // figure out which input stream we have
+    if (stream) {
+        _pcap_stream = dynamic_cast<PcapInputStream *>(stream);
+        _dnstap_stream = dynamic_cast<DnstapInputStream *>(stream);
+        _mock_stream = dynamic_cast<MockInputStream *>(stream);
+        _sflow_stream = dynamic_cast<SflowInputStream *>(stream);
+        if (!_pcap_stream && !_mock_stream && !_dnstap_stream && !_sflow_stream) {
+            throw StreamHandlerException(fmt::format("NetStreamHandler: unsupported input stream {}", stream->name()));
+        }
+    }
+}
+
+void InputResourcesStreamHandler::start()
+{
+    if (_running) {
+        return;
+    }
+
+    if (config_exists("recorded_stream")) {
+        _metrics->set_recorded_stream();
+    }
+
+    if (_pcap_stream) {
+        _pkt_connection = _pcap_stream->packet_signal.connect(&InputResourcesStreamHandler::process_packet_cb, this);
+    } else if (_dnstap_stream) {
+        _dnstap_connection = _dnstap_stream->dnstap_signal.connect(&InputResourcesStreamHandler::process_dnstap_cb, this);
+    } else if (_sflow_stream) {
+        _sflow_connection = _sflow_stream->sflow_signal.connect(&InputResourcesStreamHandler::process_sflow_cb, this);
+    }
+
+    _running = true;
+}
+
+void InputResourcesStreamHandler::stop()
+{
+    if (!_running) {
+        return;
+    }
+
+    if (_pcap_stream) {
+        _pkt_connection.disconnect();
+    } else if (_dnstap_stream) {
+        _dnstap_connection.disconnect();
+    } else if (_sflow_stream) {
+        _sflow_connection.disconnect();
+    }
+
+    _running = false;
+}
+
+void InputResourcesStreamHandler::process_sflow_cb([[maybe_unused]] const SFSample &)
+{
+    if (difftime(time(NULL), _timer) >= MEASURE_INTERVAL) {
+        _timer = time(NULL);
+        _metrics->process_resources();
+    }
+}
+
+void InputResourcesStreamHandler::process_dnstap_cb([[maybe_unused]] const dnstap::Dnstap &)
+{
+    if (difftime(time(NULL), _timer) >= MEASURE_INTERVAL) {
+        _timer = time(NULL);
+        _metrics->process_resources();
+    }
+}
+
+void InputResourcesStreamHandler::process_packet_cb([[maybe_unused]] pcpp::Packet &payload, [[maybe_unused]] PacketDirection dir, [[maybe_unused]] pcpp::ProtocolType l3, [[maybe_unused]] pcpp::ProtocolType l4, [[maybe_unused]] timespec stamp)
+{
+    if (stamp.tv_sec >= MEASURE_INTERVAL + _timestamp.tv_sec) {
+        _timestamp = stamp;
+        _metrics->process_resources(stamp);
+    }
+}
+
+void InputResourcesMetricsBucket::specialized_merge(const AbstractMetricsBucket &o)
+{
+    // static because caller guarantees only our own bucket type
+    const auto &other = static_cast<const InputResourcesMetricsBucket &>(o);
+
+    std::shared_lock r_lock(other._mutex);
+    std::unique_lock w_lock(_mutex);
+
+    _cpu_percentage.merge(other._cpu_percentage);
+    _memory_usage_kb.merge(other._memory_usage_kb);
+}
+
+void InputResourcesMetricsBucket::to_prometheus(std::stringstream &out, Metric::LabelMap add_labels) const
+{
+    {
+        auto [num_events, num_samples, event_rate, event_lock] = event_data_locked(); // thread safe
+
+        event_rate->to_prometheus(out, add_labels);
+        num_events->to_prometheus(out, add_labels);
+        num_samples->to_prometheus(out, add_labels);
+    }
+
+    std::shared_lock r_lock(_mutex);
+
+    _cpu_percentage.to_prometheus(out, add_labels);
+    _memory_usage_kb.to_prometheus(out, add_labels);
+}
+
+void InputResourcesMetricsBucket::to_json(json &j) const
+{
+    bool live_rates = !read_only() && !recorded_stream();
+
+    {
+        auto [num_events, num_samples, event_rate, event_lock] = event_data_locked(); // thread safe
+
+        event_rate->to_json(j, live_rates);
+        num_events->to_json(j);
+        num_samples->to_json(j);
+    }
+
+    std::shared_lock r_lock(_mutex);
+
+    _cpu_percentage.to_json(j);
+    _memory_usage_kb.to_json(j);
+}
+
+void InputResourcesMetricsBucket::process_resources()
+{
+    std::unique_lock lock(_mutex);
+
+    _cpu_percentage.update(_monitor.cpu_percentage());
+    _memory_usage_kb.update(_monitor.memory_usage());
+}
+
+void InputResourcesMetricsManager::process_resources(timespec stamp)
+{
+    if(stamp.tv_sec == 0) {
+        // use now()
+        std::timespec_get(&stamp, TIME_UTC);
+    }
+    // base event
+    new_event(stamp);
+    // process in the "live" bucket. this will parse the resources if we are deep sampling
+    live_bucket()->process_resources();
+}
+}