diff --git a/controller/cmd/grpc_server/main.go b/controller/cmd/grpc_server/main.go index 821b4ff..04f12f5 100644 --- a/controller/cmd/grpc_server/main.go +++ b/controller/cmd/grpc_server/main.go @@ -1,9 +1,6 @@ package main import ( - "log" - "net" - "github.com/moevm/grpc_server/internal/config" "github.com/moevm/grpc_server/internal/grpcserver" "github.com/moevm/grpc_server/internal/manager" @@ -11,6 +8,8 @@ import ( commPb "github.com/moevm/grpc_server/pkg/proto/communication" "google.golang.org/grpc" "google.golang.org/grpc/reflection" + "log" + "net" ) func main() { diff --git a/controller/pkg/proto/communication/communication.proto b/controller/pkg/proto/communication/communication.proto index 9ad73f9..204d318 100644 --- a/controller/pkg/proto/communication/communication.proto +++ b/controller/pkg/proto/communication/communication.proto @@ -62,4 +62,4 @@ message StatsReport { uint64 total_blocked = 3; uint64 total_allowed = 4; repeated ResourceStats resources = 5; -} +} \ No newline at end of file diff --git a/worker/docker-compose.yml b/worker/docker-compose.yml index b61d03a..4a971ee 100644 --- a/worker/docker-compose.yml +++ b/worker/docker-compose.yml @@ -2,17 +2,22 @@ version: '3.8' services: worker: - build: + build: context: . dockerfile: Dockerfile.cc_x86_to_x86 + environment: + - WORKER_ID=1 + - CONTROLLER_GRPC_ADDR=controller:50051 + - METRICS_GATEWAY_ADDRESS=pushgateway + - METRICS_GATEWAY_PORT=9091 volumes: - - ./data:/data + - ./data:/data working_dir: /data logging: - driver: loki + driver: loki options: - loki-timeout: 10s - no-file: "true" - loki-external-labels: "container_name={{.Name}}" - labels: "container_name, host" - loki-url: "http://${LOKI_IP}:${LOG_PORT}/loki/api/v1/push" \ No newline at end of file + loki-timeout: 10s + no-file: "true" + loki-external-labels: "container_name={{.Name}}" + labels: "container_name, host" + loki-url: "http://${LOKI_IP}:${LOG_PORT}/loki/api/v1/push" \ No newline at end of file diff --git a/worker/include/metrics_collector.hpp b/worker/include/metrics_collector.hpp index 84f5c43..b449b65 100644 --- a/worker/include/metrics_collector.hpp +++ b/worker/include/metrics_collector.hpp @@ -1,15 +1,29 @@ #ifndef METRICS_COLLECTOR_HPP #define METRICS_COLLECTOR_HPP +#include #include +#include #include #include +#include #include +#include +#include +#include +#include class MetricsCollector { - prometheus::Gateway gateway; - std::shared_ptr registry; +public: + MetricsCollector(const char *gateway_address, const char *gateway_port, + const char *worker_name); + ~MetricsCollector(); + + void IncrementPacketsReceived(uint64_t count = 1); + void IncrementPacketsPassed(uint64_t count = 1); + void IncrementPacketsDropped(uint64_t count = 1); +private: struct CPUInfo { prometheus::Gauge *gauge; @@ -21,29 +35,29 @@ class MetricsCollector { }; Time time; - }; - std::unordered_map cpu_usage; + uint64_t last_total{0}; + uint64_t last_non_idle{0}; + }; - prometheus::Gauge *memory_used_gauge; - prometheus::Gauge *task_processing_time_gauge; + void GetCPUUsage(); + void MainLoop(); + void PushMetrics(); - std::atomic is_running{true}; - std::thread thread; + prometheus::Gateway gateway; + std::shared_ptr registry; - std::atomic is_task_running; - std::chrono::time_point task_start; + std::unordered_map cpu_usage; + prometheus::Gauge *memory_used_gauge; - void GetCPUUsage(); - void MainLoop(); + prometheus::Counter *packets_received_counter; + prometheus::Counter *packets_passed_counter; + prometheus::Counter *packets_dropped_counter; -public: - MetricsCollector(const char *gateway_address, const char *gateway_port, - const char *worker_name); - ~MetricsCollector(); + prometheus::Counter *push_errors_total; - void StartTask(); - void StopTask(); + std::atomic is_running{true}; + std::thread thread; }; -#endif +#endif \ No newline at end of file diff --git a/worker/include/worker.hpp b/worker/include/worker.hpp index 768b5b3..b0d5740 100644 --- a/worker/include/worker.hpp +++ b/worker/include/worker.hpp @@ -3,6 +3,7 @@ #include "communication.grpc.pb.h" #include "communication.pb.h" +#include "metrics_collector.hpp" extern "C" { #include "dpdk_filter/domain_cache.h" #include "dpdk_filter/filtr_packets.h" @@ -11,13 +12,16 @@ extern "C" { #include "dpdk_filter/proc_packets.h" #include "dpdk_filter/types.h" } +#include #include #include #include +#include #include #include #include #include +#include #define EXPECTED_POLICY_TIME 60 #define MIN_POLICY_TIME 30 @@ -58,8 +62,19 @@ class Worker { void LogStateChange(WorkerState new_state); void SetState(WorkerState new_state); + std::unique_ptr metrics_collector_; + + std::atomic local_packets_received{0}; + std::atomic local_packets_passed{0}; + std::atomic local_packets_dropped{0}; + + std::chrono::steady_clock::time_point last_metrics_push_time; + const int METRICS_FLUSH_INTERVAL_SEC = 5; + + void flushLocalCounters(); + public: - Worker(uint64_t id); + Worker(uint64_t id, const char *gateway_address, const char *gateway_port); ~Worker(); void initDPDK(int argc, char **argv); @@ -69,10 +84,13 @@ class Worker { struct requested_classification *out_req); void forward_to_out(struct net_port *incoming_port, struct net_port *outgoing_port, uint16_t queue_number); - void statsReport(); WorkerState GetState() const { return state; } static Worker *getInstance(); void MainLoop(); + + void RecordPacketReceived(); + void RecordPacketPassed(); + void RecordPacketDropped(); }; -#endif +#endif \ No newline at end of file diff --git a/worker/src/dpdk_filter/proc_packets.c b/worker/src/dpdk_filter/proc_packets.c index 19bb286..aee4047 100644 --- a/worker/src/dpdk_filter/proc_packets.c +++ b/worker/src/dpdk_filter/proc_packets.c @@ -2,6 +2,12 @@ #include "domain_cache.h" #include "ip_cache.h" #include +#include +#include + +extern void record_packet_received(); +extern void record_packet_passed(); +extern void record_packet_dropped(); extern bool worker_classify(const char *type, const char *target, struct requested_classification *out_req); @@ -17,11 +23,17 @@ void package_sending_decision(bool solution_is_send, struct rte_mbuf *pkt, if (ret < 1) { LOG_ERROR("Failed to send packet"); - // PLUG (to be added later) - need to add processing for this case + record_packet_dropped(); rte_pktmbuf_free(pkt); + return; } + + record_packet_passed(); + return; } + + record_packet_dropped(); rte_pktmbuf_free(pkt); } @@ -48,6 +60,7 @@ void pakage_processing(struct net_port *port_in, struct net_port *port_out, } if (atomic_load(&filtring_is_turned_off)) { for (int i = 0; i < nb_rx; i++) { + record_packet_received(); package_sending_decision(true, pkts[i], port_out, queue_number); } return; @@ -55,11 +68,14 @@ void pakage_processing(struct net_port *port_in, struct net_port *port_out, for (int i = 0; i < nb_rx; i++) { + record_packet_received(); + struct info_of_pakage info_pac; memset(&info_pac, 0, sizeof(info_pac)); parsing_pakage(pkts[i], &info_pac); LOG_INFO("[PKT] port = %hu", ntohs(info_pac.number_port)); + if (info_pac.domain[0] == '\0') { LOG_INFO("Packet without dns request"); struct node_cache_ip *cached_node_ip = NULL; @@ -91,7 +107,7 @@ void pakage_processing(struct net_port *port_in, struct net_port *port_out, } else if (ret == -ENOENT) { LOG_INFO("IP cache miss, applying filter"); - struct requested_classification req_clas; // query to ip controller + struct requested_classification req_clas; bool solution_is_send; @@ -142,6 +158,8 @@ void pakage_processing(struct net_port *port_in, struct net_port *port_out, } else { LOG_ERROR("Failed to search a key-value pair in the hash table: %s", strerror(-ret)); + record_packet_dropped(); + rte_pktmbuf_free(pkts[i]); } } else { LOG_INFO("[INFO] Packet with dns request"); @@ -165,7 +183,7 @@ void pakage_processing(struct net_port *port_in, struct net_port *port_out, LOG_INFO("Domain cache miss for '%s', applying filter", info_pac.domain); - struct requested_classification req_clas; // query to domain controller + struct requested_classification req_clas; bool solution_is_send; bool classification_success = @@ -179,6 +197,9 @@ void pakage_processing(struct net_port *port_in, struct net_port *port_out, LOG_WARNING("Classification failed for %s", info_pac.domain); } + package_sending_decision(solution_is_send, pkts[i], port_out, + queue_number); + struct node_cache_domain *new_node = rte_calloc("struct_node_cache", 1, sizeof(struct node_cache_domain), RTE_CACHE_LINE_SIZE); @@ -193,7 +214,9 @@ void pakage_processing(struct net_port *port_in, struct net_port *port_out, } else { LOG_ERROR("Failed to search a key-value pair in the hash table: %s", strerror(-ret)); + record_packet_dropped(); + rte_pktmbuf_free(pkts[i]); } } } -} +} \ No newline at end of file diff --git a/worker/src/main.cpp b/worker/src/main.cpp index 5d1ea2e..3cc244b 100644 --- a/worker/src/main.cpp +++ b/worker/src/main.cpp @@ -3,23 +3,6 @@ #include -class FiltrWorker : public Worker { - MetricsCollector metrics_collector; - -protected: - void ProcessTask(const std::vector &data) { - metrics_collector.StartTask(); - metrics_collector.StopTask(); - } - -public: - FiltrWorker(const char *gateway_address, const char *gateway_port, - uint64_t id) - : Worker(id), - metrics_collector(gateway_address, gateway_port, - ("worker-" + std::to_string(id)).c_str()) {} -}; - int main(int argc, char **argv) { const char *worker_id_str = getenv("WORKER_ID"); if (worker_id_str == nullptr) { @@ -42,7 +25,7 @@ int main(int argc, char **argv) { gateway_port); try { - Worker worker(worker_id); + Worker worker(worker_id, gateway_address, gateway_port); bool test_mode = false; if (getenv("TEST_REQUEST_POLICY") != nullptr) { test_mode = true; @@ -51,13 +34,6 @@ int main(int argc, char **argv) { worker.requestPolicyFromController(); } - if (getenv("TEST_STATS") != nullptr) { - test_mode = true; - spdlog::info("Test mode: send stats"); - std::this_thread::sleep_for(std::chrono::seconds(2)); - worker.statsReport(); - } - if (const char *target = getenv("TEST_CLASSIFY_TARGET")) { const char *type = getenv("TEST_CLASSIFY_TYPE"); if (!type) @@ -91,4 +67,4 @@ int main(int argc, char **argv) { } return 0; -} +} \ No newline at end of file diff --git a/worker/src/metrics_collector.cpp b/worker/src/metrics_collector.cpp index bf9559c..5162d3f 100644 --- a/worker/src/metrics_collector.cpp +++ b/worker/src/metrics_collector.cpp @@ -9,14 +9,14 @@ namespace { double GetMemoryUsed() { std::ifstream file("/proc/self/statm"); - if (!file.is_open()) { + if (!file.is_open()) return 0; - } - long mem_pages = 0; - file >> mem_pages; - file.close(); - return mem_pages * (double)getpagesize(); + long total_pages = 0; + long rss_pages = 0; + file >> total_pages >> rss_pages; + + return rss_pages * (double)getpagesize(); } } // namespace @@ -25,6 +25,7 @@ MetricsCollector::MetricsCollector(const char *gateway_address, const char *worker_name) : gateway(gateway_address, gateway_port, worker_name), registry(std::make_shared()) { + auto &cpu_usage_family = prometheus::BuildGauge() .Name("cpu_usage") .Help("CPU Usage in percents") @@ -34,12 +35,33 @@ MetricsCollector::MetricsCollector(const char *gateway_address, .Name("memory_used") .Help("Memory used by worker in bytes") .Register(*registry); + memory_used_gauge = &memory_used_family.Add({}); + + auto &packets_received_family = prometheus::BuildCounter() + .Name("packets_received_total") + .Help("Total number of packets received") + .Register(*registry); + packets_received_counter = &packets_received_family.Add({}); - auto &task_processing_time_family = - prometheus::BuildGauge() - .Name("task_processing_time") - .Help("Task processing time (in seconds)") + auto &packets_passed_family = + prometheus::BuildCounter() + .Name("packets_passed_total") + .Help("Total number of packets passed/forwarded") .Register(*registry); + packets_passed_counter = &packets_passed_family.Add({}); + + auto &packets_dropped_family = prometheus::BuildCounter() + .Name("packets_dropped_total") + .Help("Total number of packets dropped") + .Register(*registry); + + packets_dropped_counter = &packets_dropped_family.Add({}); + + auto &push_errors_family = prometheus::BuildCounter() + .Name("push_errors_total") + .Help("Total number of push gateway errors") + .Register(*registry); + push_errors_total = &push_errors_family.Add({}); std::ifstream file("/proc/stat"); int ign; @@ -60,12 +82,8 @@ MetricsCollector::MetricsCollector(const char *gateway_address, file.close(); - memory_used_gauge = &memory_used_family.Add({}); - task_processing_time_gauge = &task_processing_time_family.Add({}); - gateway.RegisterCollectable(registry); thread = std::thread(&MetricsCollector::MainLoop, this); - is_task_running = false; } void MetricsCollector::MainLoop() { @@ -75,73 +93,86 @@ void MetricsCollector::MainLoop() { memory_used_gauge->Set(::GetMemoryUsed()); GetCPUUsage(); - if (is_task_running) { - auto cur_time = std::chrono::high_resolution_clock::now(); - task_processing_time_gauge->Set( - std::chrono::duration(cur_time - task_start).count()); - } else { - task_processing_time_gauge->Set(0); - } + PushMetrics(); + } +} - int status = gateway.PushAdd(); - if (status != 200) { - spdlog::warn("Failed to push metrics. Status {}", status); +void MetricsCollector::PushMetrics() { + int status = gateway.PushAdd(); + if (status != 200) { + spdlog::warn("Failed to push metrics. Status {}", status); + if (push_errors_total) { + push_errors_total->Increment(); } } } MetricsCollector::~MetricsCollector() { is_running = false; - thread.join(); + if (thread.joinable()) { + thread.join(); + } } void MetricsCollector::GetCPUUsage() { std::ifstream file("/proc/stat"); - CPUInfo::Time cur_time; - double percent; + if (!file.is_open()) + return; - std::string cpu_name; - int ign; + std::string line; + while (std::getline(file, line)) { + if (line.find("cpu") != 0) + break; - while (true) { - file >> cpu_name >> cur_time.user >> cur_time.user_low >> cur_time.sys >> - cur_time.idle >> ign >> ign >> ign >> ign >> ign >> ign; + std::istringstream iss(line); + std::string cpu_name; + long user, nice, sys, idle, iowait, irq, softirq, steal, guest, guest_nice; - if (cpu_name.find("cpu") != 0) - break; + iss >> cpu_name >> user >> nice >> sys >> idle >> iowait >> irq >> + softirq >> steal >> guest >> guest_nice; - CPUInfo &cpu = cpu_usage[cpu_name]; - if (cur_time.user < cpu.time.user || - cur_time.user_low < cpu.time.user_low || cur_time.sys < cpu.time.sys || - cur_time.idle < cpu.time.idle) { - // overflow detection - percent = -1.0; - } else { - uint64_t total = (cur_time.user - cpu.time.user) + - (cur_time.user_low - cpu.time.user_low) + - (cur_time.sys - cpu.time.sys); - - percent = total; - total += (cur_time.idle - cpu.time.idle); - percent = (total == 0) ? -1.0 : (percent / total) * 100.0; - } + if (cpu_name.empty()) + continue; + + uint64_t non_idle = user + nice + sys + irq + softirq + steal; + uint64_t total = non_idle + idle + iowait; - cpu.time = cur_time; - cpu.gauge->Set(percent); + auto it = cpu_usage.find(cpu_name); + if (it != cpu_usage.end()) { + CPUInfo &cpu = it->second; + + if (cpu.last_total > 0) { + uint64_t total_diff = total - cpu.last_total; + uint64_t non_idle_diff = non_idle - cpu.last_non_idle; + + double percent = (total_diff == 0) + ? 0.0 + : (double)non_idle_diff / total_diff * 100.0; + cpu.gauge->Set(percent); + } + + cpu.last_total = total; + cpu.last_non_idle = non_idle; + } } file.close(); } -void MetricsCollector::StartTask() { - is_task_running = true; - task_start = std::chrono::high_resolution_clock::now(); - task_processing_time_gauge->Set(0); - gateway.PushAdd(); +void MetricsCollector::IncrementPacketsReceived(uint64_t count) { + if (packets_received_counter) { + packets_received_counter->Increment(count); + } } -void MetricsCollector::StopTask() { - is_task_running = false; - task_processing_time_gauge->Set(0); - gateway.PushAdd(); +void MetricsCollector::IncrementPacketsPassed(uint64_t count) { + if (packets_passed_counter) { + packets_passed_counter->Increment(count); + } } + +void MetricsCollector::IncrementPacketsDropped(uint64_t count) { + if (packets_dropped_counter) { + packets_dropped_counter->Increment(count); + } +} \ No newline at end of file diff --git a/worker/src/worker.cpp b/worker/src/worker.cpp index bd03edd..f11bb73 100644 --- a/worker/src/worker.cpp +++ b/worker/src/worker.cpp @@ -13,7 +13,7 @@ extern "C" bool worker_classify(const char *type, const char *target, struct requested_classification *out_req) { Worker *worker = Worker::getInstance(); if (!worker) { - fprintf(stderr, "worker_classify: worker is null\n"); + spdlog::error("worker_classify: worker is null"); return false; } return worker->classify(std::string(type), std::string(target), out_req); @@ -45,6 +45,55 @@ void Worker::SetState(WorkerState new_state) { } } +void Worker::RecordPacketReceived() { local_packets_received++; } + +extern "C" void record_packet_received() { + Worker *worker = Worker::getInstance(); + if (!worker) { + spdlog::error("record_packet_received: worker is null"); + return; + } + worker->RecordPacketReceived(); +} + +void Worker::RecordPacketPassed() { local_packets_passed++; } + +extern "C" void record_packet_passed() { + Worker *worker = Worker::getInstance(); + if (!worker) { + spdlog::error("record_packet_passed: worker is null"); + return; + } + worker->RecordPacketPassed(); +} + +void Worker::RecordPacketDropped() { local_packets_dropped++; } + +extern "C" void record_packet_dropped() { + Worker *worker = Worker::getInstance(); + if (!worker) { + spdlog::error("record_packet_dropped: worker is null"); + return; + } + worker->RecordPacketDropped(); +} + +void Worker::flushLocalCounters() { + if (!metrics_collector_) + return; + + uint64_t received = local_packets_received.exchange(0); + uint64_t passed = local_packets_passed.exchange(0); + uint64_t dropped = local_packets_dropped.exchange(0); + + if (received > 0) + metrics_collector_->IncrementPacketsReceived(received); + if (passed > 0) + metrics_collector_->IncrementPacketsPassed(passed); + if (dropped > 0) + metrics_collector_->IncrementPacketsDropped(dropped); +} + void Worker::initDPDK(int argc, char **argv) { unsigned mbuf_quantity_in_pool = 8192; unsigned cache_size_per_kernel = 250; @@ -100,13 +149,13 @@ void Worker::forward_to_out(struct net_port *incoming_port, rte_eth_tx_burst(outgoing_port->port_id, queue_number, &tap_pkts[i], 1); if (ret < 1) { spdlog::warn("Failed to send packet"); - // PLUG (to be added later) - need to add processing for this case rte_pktmbuf_free(tap_pkts[i]); } } } void Worker::requestPolicyFromController() { + try { spdlog::info("Worker {} requests policy", worker_id); GetPolicyRequest req; @@ -272,6 +321,7 @@ void Worker::requestPolicyFromController() { bool Worker::classify(const std::string &type, const std::string &target, struct requested_classification *out_req) { + try { spdlog::info("Worker {} classifying '{}' as {}", worker_id, target, type); @@ -305,6 +355,7 @@ bool Worker::classify(const std::string &type, const std::string &target, strncpy(out_req->get_categories[i], resp.categories(i).c_str(), CATEGORY_MAX_LEN - 1); } + return true; } catch (const std::exception &e) { spdlog::error(std::string("classifyDomain: ") + e.what()); @@ -312,36 +363,18 @@ bool Worker::classify(const std::string &type, const std::string &target, } } -void Worker::statsReport() { - try { - spdlog::info("Worker {} send stats", worker_id); - - StatsReport report; - report.set_worker_id(worker_id); - report.set_time(time(nullptr)); - - grpc::ClientContext context; - google::protobuf::Empty response; - - auto status = stub_->SendStats(&context, report, &response); - if (!status.ok()) { - spdlog::error("SendStats failed: " + status.error_message()); - return; - } - - spdlog::info("Stats sent successfully"); - - } catch (const std::exception &e) { - spdlog::error("statsReport failed: {}", e.what()); - } -} - -Worker::Worker(uint64_t id) : worker_id(id), state(WorkerState::FREE) { +Worker::Worker(uint64_t id, const char *gateway_address, + const char *gateway_port) + : worker_id(id), state(WorkerState::FREE) { instance = this; std::string controller_addr = "localhost:50051"; if (const char *env_addr = getenv("CONTROLLER_GRPC_ADDR")) { controller_addr = env_addr; } + + metrics_collector_ = std::make_unique( + gateway_address, gateway_port, ("worker-" + std::to_string(id)).c_str()); + auto channel = grpc::CreateChannel(controller_addr, grpc::InsecureChannelCredentials()); stub_ = DataService::NewStub(channel); @@ -382,6 +415,7 @@ void Worker::MainLoop() { last_policy_time = steady_clock::now(); last_stats_time = steady_clock::now(); + last_metrics_push_time = steady_clock::now(); struct rte_mbuf *pkts[32]; uint16_t nb_pkts = 32; @@ -404,15 +438,6 @@ void Worker::MainLoop() { auto now = steady_clock::now(); - int64_t seconds_since_stats = (now - last_stats_time) / 1s; - if (seconds_since_stats >= stats_interval) { - std::thread([this]() { statsReport(); }).detach(); - last_stats_time = now; - stats_interval = - MIN_STATS_TIME + (rand() % (MAX_STATS_TIME - MIN_STATS_TIME + 1)); - spdlog::info("Next stats report in {}s", stats_interval); - } - int64_t seconds_since_policy = (now - last_policy_time) / 1s; if (seconds_since_policy >= policy_interval) { std::thread([this]() { requestPolicyFromController(); }).detach(); @@ -421,9 +446,16 @@ void Worker::MainLoop() { MIN_POLICY_TIME + (rand() % (MAX_POLICY_TIME - MIN_POLICY_TIME + 1)); spdlog::info("Next policy request in {}s", policy_interval); } + int64_t seconds_since_metrics = (now - last_metrics_push_time) / 1s; + if (seconds_since_metrics >= METRICS_FLUSH_INTERVAL_SEC) { + flushLocalCounters(); + last_metrics_push_time = now; + } } + flushLocalCounters(); + if (stop_flag) { SetState(WorkerState::SHUTTING_DOWN); } -} +} \ No newline at end of file