From 767cc95d96f632b26166e2972f8922cf412eaaf2 Mon Sep 17 00:00:00 2001 From: vovanbravin Date: Thu, 14 May 2026 21:14:16 +0300 Subject: [PATCH 01/12] feat: add metrics --- controller/cmd/grpc_server/main.go | 37 ++++- controller/internal/grpcserver/data_server.go | 14 +- controller/internal/manager/manager.go | 106 +++++++++++++++ .../proto/communication/communication.proto | 5 +- worker/include/metrics_collector.hpp | 57 +++++--- worker/include/worker.hpp | 19 ++- worker/src/dpdk_filter/proc_packets.c | 34 ++++- worker/src/metrics_collector.cpp | 126 ++++++++++++++++-- worker/src/worker.cpp | 93 ++++++++++++- 9 files changed, 455 insertions(+), 36 deletions(-) diff --git a/controller/cmd/grpc_server/main.go b/controller/cmd/grpc_server/main.go index 821b4ff..ba23dd5 100644 --- a/controller/cmd/grpc_server/main.go +++ b/controller/cmd/grpc_server/main.go @@ -1,8 +1,10 @@ package main import ( + "fmt" "log" "net" + "net/http" "github.com/moevm/grpc_server/internal/config" "github.com/moevm/grpc_server/internal/grpcserver" @@ -29,6 +31,39 @@ func main() { log.Fatalf("Failed to create data server: %v", err) } + go func() { + http.HandleFunc("/metrics", func(w http.ResponseWriter, r *http.Request) { + stats := mgr.GetAggregatedStats() + + w.Header().Set("Content-Type", "text/plain; version=0.0.4") + + fmt.Fprintf(w, "# HELP packets_passed_total Total number of packets passed\n") + fmt.Fprintf(w, "# TYPE packets_passed_total counter\n") + fmt.Fprintf(w, "packets_passed_total %d\n", stats.PacketsPassed) + + fmt.Fprintf(w, "# HELP packets_dropped_total Total number of packets dropped\n") + fmt.Fprintf(w, "# TYPE packets_dropped_total counter\n") + fmt.Fprintf(w, "packets_dropped_total %d\n", stats.PacketsDropped) + + fmt.Fprintf(w, "# HELP packets_received_total Total number of packets received\n") + fmt.Fprintf(w, "# TYPE packets_received_total counter\n") + fmt.Fprintf(w, "packets_received_total %d\n", stats.PacketsReceived) + + fmt.Fprintf(w, "# HELP worker_last_update Last update timestamp\n") + fmt.Fprintf(w, "# TYPE worker_last_update gauge\n") + fmt.Fprintf(w, "worker_last_update %d\n", stats.LastUpdate.Unix()) + + fmt.Fprintf(w, "# HELP worker_up Worker is reporting metrics\n") + fmt.Fprintf(w, "# TYPE worker_up gauge\n") + fmt.Fprintf(w, "worker_up 1\n") + }) + + log.Println("Metrics HTTP server starting on :8081") + if err := http.ListenAndServe("0.0.0.0:8081", nil); err != nil { + log.Printf("Metrics HTTP server error: %v", err) + } + }() + lis, err := net.Listen("tcp", net.JoinHostPort(cfg.Host, cfg.Port)) if err != nil { log.Fatalf("failed to listen: %v", err) @@ -48,4 +83,4 @@ func main() { if err := service.Serve(lis); err != nil { log.Fatalf("failed to serve: %v", err) } -} +} \ No newline at end of file diff --git a/controller/internal/grpcserver/data_server.go b/controller/internal/grpcserver/data_server.go index 3db8292..0de6331 100644 --- a/controller/internal/grpcserver/data_server.go +++ b/controller/internal/grpcserver/data_server.go @@ -108,7 +108,15 @@ func (s *DataServer) Classify(ctx context.Context, req *pb.ClassifyRequest) (*pb } func (s *DataServer) SendStats(ctx context.Context, report *pb.StatsReport) (*emptypb.Empty, error) { - log.Printf("gRPC Stats from worker %d: blocked=%d allowed=%d", - report.WorkerId, report.TotalBlocked, report.TotalAllowed) + log.Printf("gRPC Stats from worker %d: received=%d passed=%d dropped=%d", + report.WorkerId, + report.PacketsReceived, + report.PacketsPassed, + report.PacketsDropped) + + if err := s.manager.HandleStatsReport(report); err != nil { + log.Printf("Error handling stats from worker %d: %v", report.WorkerId, err) + } + return &emptypb.Empty{}, nil -} +} \ No newline at end of file diff --git a/controller/internal/manager/manager.go b/controller/internal/manager/manager.go index 5c99619..31bd426 100644 --- a/controller/internal/manager/manager.go +++ b/controller/internal/manager/manager.go @@ -3,6 +3,8 @@ package manager import ( "fmt" "log" + "sync" + "time" communication "github.com/moevm/grpc_server/pkg/proto/communication" "google.golang.org/protobuf/proto" @@ -10,11 +12,30 @@ import ( type Manager struct { policyManager *PolicyManager + statsStore map[uint64]*WorkerStats // Хранилище статистики воркеров + statsMutex sync.RWMutex +} + +type WorkerStats struct { + LastUpdate time.Time + PacketsReceived uint64 + PacketsPassed uint64 + PacketsDropped uint64 + TotalBlocked uint64 + TotalAllowed uint64 + Resources map[string]*ResourceStats // Статистика по доменам/IP +} + +type ResourceStats struct { + Domain string + Blocked uint64 + Allowed uint64 } func NewManager() (*Manager, error) { return &Manager{ policyManager: NewPolicyManager(), + statsStore: make(map[uint64]*WorkerStats), }, nil } @@ -51,3 +72,88 @@ func (m *Manager) UpdateConfig(configData []byte) error { } return m.policyManager.UpdateConfig(configData) } + +func (m *Manager) HandleStatsReport(report *communication.StatsReport) error { + m.statsMutex.Lock() + defer m.statsMutex.Unlock() + + log.Printf("Processing stats from worker %d: received=%d passed=%d dropped=%d", + report.WorkerId, + report.PacketsReceived, + report.PacketsPassed, + report.PacketsDropped) + + stats, exists := m.statsStore[report.WorkerId] + if !exists { + stats = &WorkerStats{ + Resources: make(map[string]*ResourceStats), + } + m.statsStore[report.WorkerId] = stats + } + + stats.LastUpdate = time.Unix(int64(report.Time), 0) + stats.PacketsReceived = report.PacketsReceived + stats.PacketsPassed = report.PacketsPassed + stats.PacketsDropped = report.PacketsDropped + stats.TotalBlocked = report.TotalBlocked + stats.TotalAllowed = report.TotalAllowed + + for _, resource := range report.Resources { + if _, ok := stats.Resources[resource.Domain]; !ok { + stats.Resources[resource.Domain] = &ResourceStats{ + Domain: resource.Domain, + } + } + stats.Resources[resource.Domain].Blocked = resource.Blocked + stats.Resources[resource.Domain].Allowed = resource.Allowed + } + + log.Printf("Stats saved for worker %d", report.WorkerId) + return nil +} + +func (m *Manager) GetWorkerStats(workerID uint64) (*WorkerStats, bool) { + m.statsMutex.RLock() + defer m.statsMutex.RUnlock() + stats, ok := m.statsStore[workerID] + return stats, ok +} + +func (m *Manager) GetAllStats() map[uint64]*WorkerStats { + m.statsMutex.RLock() + defer m.statsMutex.RUnlock() + + result := make(map[uint64]*WorkerStats) + for k, v := range m.statsStore { + result[k] = v + } + return result +} + +func (m *Manager) GetAggregatedStats() *WorkerStats { + m.statsMutex.RLock() + defer m.statsMutex.RUnlock() + + aggregated := &WorkerStats{ + Resources: make(map[string]*ResourceStats), + } + + for _, stats := range m.statsStore { + aggregated.PacketsReceived += stats.PacketsReceived + aggregated.PacketsPassed += stats.PacketsPassed + aggregated.PacketsDropped += stats.PacketsDropped + aggregated.TotalBlocked += stats.TotalBlocked + aggregated.TotalAllowed += stats.TotalAllowed + + for domain, resource := range stats.Resources { + if _, ok := aggregated.Resources[domain]; !ok { + aggregated.Resources[domain] = &ResourceStats{Domain: domain} + } + aggregated.Resources[domain].Blocked += resource.Blocked + aggregated.Resources[domain].Allowed += resource.Allowed + } + } + + aggregated.LastUpdate = time.Now() + return aggregated +} \ No newline at end of file diff --git a/controller/pkg/proto/communication/communication.proto b/controller/pkg/proto/communication/communication.proto index 1fe6dc4..e203cfb 100644 --- a/controller/pkg/proto/communication/communication.proto +++ b/controller/pkg/proto/communication/communication.proto @@ -59,4 +59,7 @@ message StatsReport { uint64 total_blocked = 3; uint64 total_allowed = 4; repeated ResourceStats resources = 5; -} + uint64 packets_received = 6; + uint64 packets_passed = 7; + uint64 packets_dropped = 8; +} \ No newline at end of file diff --git a/worker/include/metrics_collector.hpp b/worker/include/metrics_collector.hpp index 84f5c43..5acaa70 100644 --- a/worker/include/metrics_collector.hpp +++ b/worker/include/metrics_collector.hpp @@ -1,15 +1,33 @@ #ifndef METRICS_COLLECTOR_HPP #define METRICS_COLLECTOR_HPP +#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(int count = 1); + void IncrementPacketsPassed(int count = 1); + void IncrementPacketsDropped(const std::string &reason, int count = 1); + void IncrementBlockedDomain(const std::string &domain_or_ip); + + void StartTask(); + void StopTask(); + void ObserveCollectionDuration(double seconds); +private: struct CPUInfo { prometheus::Gauge *gauge; @@ -23,27 +41,34 @@ class MetricsCollector { Time time; }; - std::unordered_map cpu_usage; + void GetCPUUsage(); + void MainLoop(); + void PushMetrics(); + + prometheus::Gateway gateway; + std::shared_ptr registry; + std::unordered_map cpu_usage; prometheus::Gauge *memory_used_gauge; prometheus::Gauge *task_processing_time_gauge; - std::atomic is_running{true}; - std::thread thread; + prometheus::Counter *packets_received_counter; + prometheus::Counter *packets_passed_counter; + prometheus::Counter *packets_dropped_counter; + prometheus::Family *packets_dropped_by_reason_family; + prometheus::Family *blocked_domains_family; - std::atomic is_task_running; - std::chrono::time_point task_start; + prometheus::Counter *tasks_completed_counter; + prometheus::Histogram *task_duration_histogram; - void GetCPUUsage(); - void MainLoop(); + prometheus::Histogram *metrics_collection_duration; + prometheus::Counter *push_errors_total; -public: - MetricsCollector(const char *gateway_address, const char *gateway_port, - const char *worker_name); - ~MetricsCollector(); + std::atomic is_running{true}; + std::thread thread; - void StartTask(); - void StopTask(); + std::atomic is_task_running{false}; + std::chrono::time_point task_start; }; -#endif +#endif \ No newline at end of file diff --git a/worker/include/worker.hpp b/worker/include/worker.hpp index a3e2101..3781303 100644 --- a/worker/include/worker.hpp +++ b/worker/include/worker.hpp @@ -1,6 +1,7 @@ #ifndef WORKER_HPP #define WORKER_HPP +#include "../include/metrics_collector.hpp" #include "communication.grpc.pb.h" #include "communication.pb.h" extern "C" { @@ -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 @@ -56,6 +60,12 @@ class Worker { void LogStateChange(WorkerState new_state); void SetState(WorkerState new_state); + std::unique_ptr metrics_collector_; + + std::atomic packets_received_count{0}; + std::atomic packets_passed_count{0}; + std::atomic packets_dropped_count{0}; + public: Worker(uint64_t id); ~Worker(); @@ -71,6 +81,13 @@ class Worker { WorkerState GetState() const { return state; } static Worker *getInstance(); void MainLoop(); + + void RecordPacketReceived(); + void RecordPacketPassed(); + void RecordPacketDropped(const std::string &reason); + void RecordDomainBlocked(const std::string &domain_or_ip); + void RecordTaskStart(); + void RecordTaskEnd(); }; -#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 37c864d..a83bbd6 100644 --- a/worker/src/dpdk_filter/proc_packets.c +++ b/worker/src/dpdk_filter/proc_packets.c @@ -2,6 +2,11 @@ #include "domain_cache.h" #include "ip_cache.h" +extern void record_packet_received(); +extern void record_packet_passed(); +extern void record_packet_droped(char *reason); +extern void record_domain_blocked(char *domain); + extern bool worker_classify(const char *type, const char *target, struct requested_classification *out_req); @@ -11,16 +16,19 @@ void package_sending_decision(bool solution_is_send, struct rte_mbuf *pkt, struct net_port *port_out, uint16_t queue_number) { if (solution_is_send) { + record_packet_passed(); + struct rte_mbuf *tx_pkt[1] = {pkt}; uint16_t ret = rte_eth_tx_burst(port_out->port_id, queue_number, tx_pkt, 1); if (ret < 1) { LOG_ERROR("Failed to send packet"); - // PLUG (to be added later) - need to add processing for this case rte_pktmbuf_free(pkt); } return; } + + record_packet_droped("blocked"); rte_pktmbuf_free(pkt); } @@ -42,12 +50,15 @@ void pakage_processing(struct net_port *port_in, struct net_port *port_out, rte_eth_rx_burst(port_in->port_id, queue_number, pkts, nb_pkts); 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; @@ -74,7 +85,7 @@ void pakage_processing(struct net_port *port_in, struct net_port *port_out, port_out, queue_number); } else if (ret == -ENOENT) { - struct requested_classification req_clas; // query to ip controller + struct requested_classification req_clas; bool solution_is_send; @@ -98,6 +109,10 @@ void pakage_processing(struct net_port *port_in, struct net_port *port_out, LOG_WARNING("Classification failed for IP %s", ip_str); } + if (!solution_is_send) { + record_domain_blocked(ip_str); + } + package_sending_decision(solution_is_send, pkts[i], port_out, queue_number); @@ -140,9 +155,13 @@ void pakage_processing(struct net_port *port_in, struct net_port *port_out, if (ret >= 0 && cached_node_domain) { package_sending_decision(cached_node_domain->solution_is_send, pkts[i], port_out, queue_number); + + if (!cached_node_domain->solution_is_send) { + record_domain_blocked(info_pac.domain); + } } else if (ret == -ENOENT) { - struct requested_classification req_clas; // query to domain controller + struct requested_classification req_clas; bool solution_is_send; bool classification_success = @@ -156,6 +175,13 @@ void pakage_processing(struct net_port *port_in, struct net_port *port_out, LOG_WARNING("Classification failed for %s", info_pac.domain); } + if (!solution_is_send) { + record_domain_blocked(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); @@ -173,4 +199,4 @@ void pakage_processing(struct net_port *port_in, struct net_port *port_out, } } } -} +} \ No newline at end of file diff --git a/worker/src/metrics_collector.cpp b/worker/src/metrics_collector.cpp index bf9559c..c0f12c7 100644 --- a/worker/src/metrics_collector.cpp +++ b/worker/src/metrics_collector.cpp @@ -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,67 @@ 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 &task_processing_time_family = prometheus::BuildGauge() .Name("task_processing_time") .Help("Task processing time (in seconds)") .Register(*registry); + task_processing_time_gauge = &task_processing_time_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 &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({}); + + packets_dropped_by_reason_family = + &prometheus::BuildCounter() + .Name("packets_dropped_by_reason_total") + .Help("Packets dropped by reason") + .Register(*registry); + + blocked_domains_family = &prometheus::BuildCounter() + .Name("blocked_domains_total") + .Help("Number of blocked requests by domain/IP") + .Register(*registry); + + auto &tasks_completed_family = prometheus::BuildCounter() + .Name("tasks_completed_total") + .Help("Total number of completed tasks") + .Register(*registry); + tasks_completed_counter = &tasks_completed_family.Add({}); + + auto &task_duration_family = prometheus::BuildHistogram() + .Name("task_duration_seconds") + .Help("Task execution duration in seconds") + .Register(*registry); + + prometheus::Histogram::BucketBoundaries task_buckets = { + 0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, + 0.5, 1.0, 2.5, 5.0, 10.0, 30.0, 60.0}; + task_duration_histogram = + &task_duration_family.Add({}, std::move(task_buckets)); + + 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,9 +116,6 @@ 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; @@ -83,16 +136,25 @@ void MetricsCollector::MainLoop() { task_processing_time_gauge->Set(0); } - int status = gateway.PushAdd(); - if (status != 200) { - spdlog::warn("Failed to push metrics. Status {}", status); + PushMetrics(); + } +} + +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() { @@ -133,15 +195,61 @@ void MetricsCollector::GetCPUUsage() { file.close(); } +void MetricsCollector::IncrementPacketsReceived(int count) { + if (packets_received_counter) { + packets_received_counter->Increment(count); + } +} + +void MetricsCollector::IncrementPacketsPassed(int count) { + if (packets_passed_counter) { + packets_passed_counter->Increment(count); + } +} + +void MetricsCollector::IncrementPacketsDropped(const std::string &reason, + int count) { + if (packets_dropped_counter) { + packets_dropped_counter->Increment(count); + } + if (packets_dropped_by_reason_family) { + packets_dropped_by_reason_family->Add({{"reason", reason}}) + .Increment(count); + } +} + +void MetricsCollector::IncrementBlockedDomain(const std::string &domain_or_ip) { + if (blocked_domains_family) { + blocked_domains_family->Add({{"domain", domain_or_ip}}).Increment(); + } +} + void MetricsCollector::StartTask() { is_task_running = true; task_start = std::chrono::high_resolution_clock::now(); task_processing_time_gauge->Set(0); - gateway.PushAdd(); + PushMetrics(); } void MetricsCollector::StopTask() { + auto duration = std::chrono::duration( + std::chrono::high_resolution_clock::now() - task_start) + .count(); + + if (task_duration_histogram) { + task_duration_histogram->Observe(duration); + } + + if (tasks_completed_counter) { + tasks_completed_counter->Increment(); + } + is_task_running = false; task_processing_time_gauge->Set(0); - gateway.PushAdd(); + PushMetrics(); } + +void MetricsCollector::ObserveCollectionDuration(double seconds) { + // Метод оставлен для совместимости + (void)seconds; +} \ No newline at end of file diff --git a/worker/src/worker.cpp b/worker/src/worker.cpp index 18da562..0e8e262 100644 --- a/worker/src/worker.cpp +++ b/worker/src/worker.cpp @@ -45,6 +45,82 @@ void Worker::SetState(WorkerState new_state) { } } +void Worker::RecordPacketReceived() { + packets_received_count++; + if (metrics_collector_) { + metrics_collector_->IncrementPacketsReceived(1); + } +} + +extern "C" void record_packet_received() { + Worker *worker = Worker::getInstance(); + if (!worker) { + fprintf(stderr, "worker_classify: worker is null\n"); + return; + } + worker->RecordPacketReceived(); +} + +void Worker::RecordPacketPassed() { + packets_passed_count++; + if (metrics_collector_) { + metrics_collector_->IncrementPacketsPassed(1); + } +} + +extern "C" void record_packet_passed() { + Worker *worker = Worker::getInstance(); + if (!worker) { + fprintf(stderr, "worker_classify: worker is null\n"); + return; + } + worker->RecordPacketPassed(); +} + +void Worker::RecordPacketDropped(const std::string &reason) { + packets_dropped_count++; + if (metrics_collector_) { + metrics_collector_->IncrementPacketsDropped(reason, 1); + } +} + +extern "C" void record_packet_droped(char * reson) { + Worker *worker = Worker::getInstance(); + if (!worker) { + fprintf(stderr, "worker_classify: worker is null\n"); + return; + } + worker->RecordPacketDropped(reson); +} + +void Worker::RecordDomainBlocked(const std::string &domain_or_ip) { + if (metrics_collector_) { + metrics_collector_->IncrementBlockedDomain(domain_or_ip); + } +} + +extern "C" void record_domain_blocked(char *endpoint) { + Worker *worker = Worker::getInstance(); + if (!worker) { + fprintf(stderr, "worker_classify: worker is null\n"); + return; + } + worker->RecordDomainBlocked(endpoint); +} + + +void Worker::RecordTaskStart() { + if (metrics_collector_) { + metrics_collector_->StartTask(); + } +} + +void Worker::RecordTaskEnd() { + if (metrics_collector_) { + metrics_collector_->StopTask(); + } +} + void Worker::initDPDK(int argc, char **argv) { unsigned mbuf_quantity_in_pool = 8192; unsigned cache_size_per_kernel = 250; @@ -100,13 +176,14 @@ 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() { + RecordTaskStart(); + try { spdlog::info("Worker {} requests policy", worker_id); GetPolicyRequest req; @@ -120,6 +197,7 @@ void Worker::requestPolicyFromController() { if (!status.ok()) { spdlog::error("GetPolicy failed: " + status.error_message()); + RecordTaskEnd(); return; } @@ -263,10 +341,14 @@ void Worker::requestPolicyFromController() { } catch (const std::exception &e) { spdlog::error("requestPolicyFromController exception: {}", e.what()); } + + RecordTaskEnd(); } bool Worker::classify(const std::string &type, const std::string &target, struct requested_classification *out_req) { + RecordTaskStart(); + try { spdlog::info("Worker {} classifying '{}' as {}", worker_id, target, type); @@ -281,6 +363,7 @@ bool Worker::classify(const std::string &type, const std::string &target, auto status = stub_->Classify(&context, req, &resp); if (!status.ok()) { spdlog::error("Classify failed: " + status.error_message()); + RecordTaskEnd(); return false; } @@ -300,14 +383,19 @@ 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); } + + RecordTaskEnd(); return true; } catch (const std::exception &e) { spdlog::error(std::string("classifyDomain: ") + e.what()); + RecordTaskEnd(); return false; } } void Worker::statsReport() { + RecordTaskStart(); + try { spdlog::info("Worker {} send stats", worker_id); @@ -321,6 +409,7 @@ void Worker::statsReport() { auto status = stub_->SendStats(&context, report, &response); if (!status.ok()) { spdlog::error("SendStats failed: " + status.error_message()); + RecordTaskEnd(); return; } @@ -329,6 +418,8 @@ void Worker::statsReport() { } catch (const std::exception &e) { spdlog::error("statsReport failed: {}", e.what()); } + + RecordTaskEnd(); } Worker::Worker(uint64_t id) : worker_id(id), state(WorkerState::FREE) { From f155a86b277e8671933e365aa082376262c14f1f Mon Sep 17 00:00:00 2001 From: vovanbravin Date: Thu, 14 May 2026 21:25:57 +0300 Subject: [PATCH 02/12] fix linter --- controller/cmd/grpc_server/main.go | 2 +- controller/internal/grpcserver/data_server.go | 2 +- controller/internal/manager/manager.go | 2 +- worker/include/metrics_collector.hpp | 2 +- worker/src/dpdk_filter/proc_packets.c | 10 +++++----- worker/src/metrics_collector.cpp | 2 +- worker/src/worker.cpp | 15 +++++++-------- 7 files changed, 17 insertions(+), 18 deletions(-) diff --git a/controller/cmd/grpc_server/main.go b/controller/cmd/grpc_server/main.go index ba23dd5..8d53f57 100644 --- a/controller/cmd/grpc_server/main.go +++ b/controller/cmd/grpc_server/main.go @@ -83,4 +83,4 @@ func main() { if err := service.Serve(lis); err != nil { log.Fatalf("failed to serve: %v", err) } -} \ No newline at end of file +} diff --git a/controller/internal/grpcserver/data_server.go b/controller/internal/grpcserver/data_server.go index 0de6331..e8522b9 100644 --- a/controller/internal/grpcserver/data_server.go +++ b/controller/internal/grpcserver/data_server.go @@ -119,4 +119,4 @@ func (s *DataServer) SendStats(ctx context.Context, report *pb.StatsReport) (*em } return &emptypb.Empty{}, nil -} \ No newline at end of file +} diff --git a/controller/internal/manager/manager.go b/controller/internal/manager/manager.go index 31bd426..d6041c4 100644 --- a/controller/internal/manager/manager.go +++ b/controller/internal/manager/manager.go @@ -156,4 +156,4 @@ func (m *Manager) GetAggregatedStats() *WorkerStats { aggregated.LastUpdate = time.Now() return aggregated -} \ No newline at end of file +} diff --git a/worker/include/metrics_collector.hpp b/worker/include/metrics_collector.hpp index 5acaa70..0b971f2 100644 --- a/worker/include/metrics_collector.hpp +++ b/worker/include/metrics_collector.hpp @@ -8,9 +8,9 @@ #include #include #include +#include #include #include -#include class MetricsCollector { public: diff --git a/worker/src/dpdk_filter/proc_packets.c b/worker/src/dpdk_filter/proc_packets.c index a83bbd6..8310474 100644 --- a/worker/src/dpdk_filter/proc_packets.c +++ b/worker/src/dpdk_filter/proc_packets.c @@ -17,7 +17,7 @@ void package_sending_decision(bool solution_is_send, struct rte_mbuf *pkt, uint16_t queue_number) { if (solution_is_send) { record_packet_passed(); - + struct rte_mbuf *tx_pkt[1] = {pkt}; uint16_t ret = rte_eth_tx_burst(port_out->port_id, queue_number, tx_pkt, 1); @@ -27,7 +27,7 @@ void package_sending_decision(bool solution_is_send, struct rte_mbuf *pkt, } return; } - + record_packet_droped("blocked"); rte_pktmbuf_free(pkt); } @@ -50,7 +50,7 @@ void pakage_processing(struct net_port *port_in, struct net_port *port_out, rte_eth_rx_burst(port_in->port_id, queue_number, pkts, nb_pkts); for (int i = 0; i < nb_rx; i++) { - + record_packet_received(); struct info_of_pakage info_pac; @@ -58,7 +58,7 @@ void pakage_processing(struct net_port *port_in, struct net_port *port_out, 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; @@ -155,7 +155,7 @@ void pakage_processing(struct net_port *port_in, struct net_port *port_out, if (ret >= 0 && cached_node_domain) { package_sending_decision(cached_node_domain->solution_is_send, pkts[i], port_out, queue_number); - + if (!cached_node_domain->solution_is_send) { record_domain_blocked(info_pac.domain); } diff --git a/worker/src/metrics_collector.cpp b/worker/src/metrics_collector.cpp index c0f12c7..cb91acc 100644 --- a/worker/src/metrics_collector.cpp +++ b/worker/src/metrics_collector.cpp @@ -25,7 +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") diff --git a/worker/src/worker.cpp b/worker/src/worker.cpp index 0e8e262..e5538f9 100644 --- a/worker/src/worker.cpp +++ b/worker/src/worker.cpp @@ -84,7 +84,7 @@ void Worker::RecordPacketDropped(const std::string &reason) { } } -extern "C" void record_packet_droped(char * reson) { +extern "C" void record_packet_droped(char *reson) { Worker *worker = Worker::getInstance(); if (!worker) { fprintf(stderr, "worker_classify: worker is null\n"); @@ -108,7 +108,6 @@ extern "C" void record_domain_blocked(char *endpoint) { worker->RecordDomainBlocked(endpoint); } - void Worker::RecordTaskStart() { if (metrics_collector_) { metrics_collector_->StartTask(); @@ -183,7 +182,7 @@ void Worker::forward_to_out(struct net_port *incoming_port, void Worker::requestPolicyFromController() { RecordTaskStart(); - + try { spdlog::info("Worker {} requests policy", worker_id); GetPolicyRequest req; @@ -341,14 +340,14 @@ void Worker::requestPolicyFromController() { } catch (const std::exception &e) { spdlog::error("requestPolicyFromController exception: {}", e.what()); } - + RecordTaskEnd(); } bool Worker::classify(const std::string &type, const std::string &target, struct requested_classification *out_req) { RecordTaskStart(); - + try { spdlog::info("Worker {} classifying '{}' as {}", worker_id, target, type); @@ -383,7 +382,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); } - + RecordTaskEnd(); return true; } catch (const std::exception &e) { @@ -395,7 +394,7 @@ bool Worker::classify(const std::string &type, const std::string &target, void Worker::statsReport() { RecordTaskStart(); - + try { spdlog::info("Worker {} send stats", worker_id); @@ -418,7 +417,7 @@ void Worker::statsReport() { } catch (const std::exception &e) { spdlog::error("statsReport failed: {}", e.what()); } - + RecordTaskEnd(); } From 8672009159d56cbb5521a28fdbeb061b383eceee Mon Sep 17 00:00:00 2001 From: vovanbravin Date: Tue, 19 May 2026 22:11:20 +0300 Subject: [PATCH 03/12] fix push --- controller/cmd/grpc_server/main.go | 33 ------ controller/internal/grpcserver/data_server.go | 14 +-- controller/internal/manager/manager.go | 108 +----------------- .../proto/communication/communication.proto | 3 - worker/include/worker.hpp | 1 - worker/src/worker.cpp | 37 ------ 6 files changed, 4 insertions(+), 192 deletions(-) diff --git a/controller/cmd/grpc_server/main.go b/controller/cmd/grpc_server/main.go index 8d53f57..7119d77 100644 --- a/controller/cmd/grpc_server/main.go +++ b/controller/cmd/grpc_server/main.go @@ -31,39 +31,6 @@ func main() { log.Fatalf("Failed to create data server: %v", err) } - go func() { - http.HandleFunc("/metrics", func(w http.ResponseWriter, r *http.Request) { - stats := mgr.GetAggregatedStats() - - w.Header().Set("Content-Type", "text/plain; version=0.0.4") - - fmt.Fprintf(w, "# HELP packets_passed_total Total number of packets passed\n") - fmt.Fprintf(w, "# TYPE packets_passed_total counter\n") - fmt.Fprintf(w, "packets_passed_total %d\n", stats.PacketsPassed) - - fmt.Fprintf(w, "# HELP packets_dropped_total Total number of packets dropped\n") - fmt.Fprintf(w, "# TYPE packets_dropped_total counter\n") - fmt.Fprintf(w, "packets_dropped_total %d\n", stats.PacketsDropped) - - fmt.Fprintf(w, "# HELP packets_received_total Total number of packets received\n") - fmt.Fprintf(w, "# TYPE packets_received_total counter\n") - fmt.Fprintf(w, "packets_received_total %d\n", stats.PacketsReceived) - - fmt.Fprintf(w, "# HELP worker_last_update Last update timestamp\n") - fmt.Fprintf(w, "# TYPE worker_last_update gauge\n") - fmt.Fprintf(w, "worker_last_update %d\n", stats.LastUpdate.Unix()) - - fmt.Fprintf(w, "# HELP worker_up Worker is reporting metrics\n") - fmt.Fprintf(w, "# TYPE worker_up gauge\n") - fmt.Fprintf(w, "worker_up 1\n") - }) - - log.Println("Metrics HTTP server starting on :8081") - if err := http.ListenAndServe("0.0.0.0:8081", nil); err != nil { - log.Printf("Metrics HTTP server error: %v", err) - } - }() - lis, err := net.Listen("tcp", net.JoinHostPort(cfg.Host, cfg.Port)) if err != nil { log.Fatalf("failed to listen: %v", err) diff --git a/controller/internal/grpcserver/data_server.go b/controller/internal/grpcserver/data_server.go index e8522b9..5b33cbd 100644 --- a/controller/internal/grpcserver/data_server.go +++ b/controller/internal/grpcserver/data_server.go @@ -108,15 +108,7 @@ func (s *DataServer) Classify(ctx context.Context, req *pb.ClassifyRequest) (*pb } func (s *DataServer) SendStats(ctx context.Context, report *pb.StatsReport) (*emptypb.Empty, error) { - log.Printf("gRPC Stats from worker %d: received=%d passed=%d dropped=%d", - report.WorkerId, - report.PacketsReceived, - report.PacketsPassed, - report.PacketsDropped) - - if err := s.manager.HandleStatsReport(report); err != nil { - log.Printf("Error handling stats from worker %d: %v", report.WorkerId, err) - } - + log.Printf("gRPC Stats from worker %d: blocked=%d allowed=%d", + report.WorkerId, report.TotalBlocked, report.TotalAllowed) return &emptypb.Empty{}, nil -} +} \ No newline at end of file diff --git a/controller/internal/manager/manager.go b/controller/internal/manager/manager.go index d6041c4..af8b475 100644 --- a/controller/internal/manager/manager.go +++ b/controller/internal/manager/manager.go @@ -3,8 +3,6 @@ package manager import ( "fmt" "log" - "sync" - "time" communication "github.com/moevm/grpc_server/pkg/proto/communication" "google.golang.org/protobuf/proto" @@ -12,30 +10,11 @@ import ( type Manager struct { policyManager *PolicyManager - statsStore map[uint64]*WorkerStats // Хранилище статистики воркеров - statsMutex sync.RWMutex -} - -type WorkerStats struct { - LastUpdate time.Time - PacketsReceived uint64 - PacketsPassed uint64 - PacketsDropped uint64 - TotalBlocked uint64 - TotalAllowed uint64 - Resources map[string]*ResourceStats // Статистика по доменам/IP -} - -type ResourceStats struct { - Domain string - Blocked uint64 - Allowed uint64 } func NewManager() (*Manager, error) { return &Manager{ policyManager: NewPolicyManager(), - statsStore: make(map[uint64]*WorkerStats), }, nil } @@ -71,89 +50,4 @@ func (m *Manager) UpdateConfig(configData []byte) error { return fmt.Errorf("policyManager is nil") } return m.policyManager.UpdateConfig(configData) -} - -func (m *Manager) HandleStatsReport(report *communication.StatsReport) error { - m.statsMutex.Lock() - defer m.statsMutex.Unlock() - - log.Printf("Processing stats from worker %d: received=%d passed=%d dropped=%d", - report.WorkerId, - report.PacketsReceived, - report.PacketsPassed, - report.PacketsDropped) - - stats, exists := m.statsStore[report.WorkerId] - if !exists { - stats = &WorkerStats{ - Resources: make(map[string]*ResourceStats), - } - m.statsStore[report.WorkerId] = stats - } - - stats.LastUpdate = time.Unix(int64(report.Time), 0) - stats.PacketsReceived = report.PacketsReceived - stats.PacketsPassed = report.PacketsPassed - stats.PacketsDropped = report.PacketsDropped - stats.TotalBlocked = report.TotalBlocked - stats.TotalAllowed = report.TotalAllowed - - for _, resource := range report.Resources { - if _, ok := stats.Resources[resource.Domain]; !ok { - stats.Resources[resource.Domain] = &ResourceStats{ - Domain: resource.Domain, - } - } - stats.Resources[resource.Domain].Blocked = resource.Blocked - stats.Resources[resource.Domain].Allowed = resource.Allowed - } - - log.Printf("Stats saved for worker %d", report.WorkerId) - return nil -} - -func (m *Manager) GetWorkerStats(workerID uint64) (*WorkerStats, bool) { - m.statsMutex.RLock() - defer m.statsMutex.RUnlock() - stats, ok := m.statsStore[workerID] - return stats, ok -} - -func (m *Manager) GetAllStats() map[uint64]*WorkerStats { - m.statsMutex.RLock() - defer m.statsMutex.RUnlock() - - result := make(map[uint64]*WorkerStats) - for k, v := range m.statsStore { - result[k] = v - } - return result -} - -func (m *Manager) GetAggregatedStats() *WorkerStats { - m.statsMutex.RLock() - defer m.statsMutex.RUnlock() - - aggregated := &WorkerStats{ - Resources: make(map[string]*ResourceStats), - } - - for _, stats := range m.statsStore { - aggregated.PacketsReceived += stats.PacketsReceived - aggregated.PacketsPassed += stats.PacketsPassed - aggregated.PacketsDropped += stats.PacketsDropped - aggregated.TotalBlocked += stats.TotalBlocked - aggregated.TotalAllowed += stats.TotalAllowed - - for domain, resource := range stats.Resources { - if _, ok := aggregated.Resources[domain]; !ok { - aggregated.Resources[domain] = &ResourceStats{Domain: domain} - } - aggregated.Resources[domain].Blocked += resource.Blocked - aggregated.Resources[domain].Allowed += resource.Allowed - } - } - - aggregated.LastUpdate = time.Now() - return aggregated -} +} \ No newline at end of file diff --git a/controller/pkg/proto/communication/communication.proto b/controller/pkg/proto/communication/communication.proto index e203cfb..0e9a095 100644 --- a/controller/pkg/proto/communication/communication.proto +++ b/controller/pkg/proto/communication/communication.proto @@ -59,7 +59,4 @@ message StatsReport { uint64 total_blocked = 3; uint64 total_allowed = 4; repeated ResourceStats resources = 5; - uint64 packets_received = 6; - uint64 packets_passed = 7; - uint64 packets_dropped = 8; } \ No newline at end of file diff --git a/worker/include/worker.hpp b/worker/include/worker.hpp index 3781303..be560c7 100644 --- a/worker/include/worker.hpp +++ b/worker/include/worker.hpp @@ -77,7 +77,6 @@ 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(); diff --git a/worker/src/worker.cpp b/worker/src/worker.cpp index e5538f9..3fd2ffb 100644 --- a/worker/src/worker.cpp +++ b/worker/src/worker.cpp @@ -392,35 +392,6 @@ bool Worker::classify(const std::string &type, const std::string &target, } } -void Worker::statsReport() { - RecordTaskStart(); - - 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()); - RecordTaskEnd(); - return; - } - - spdlog::info("Stats sent successfully"); - - } catch (const std::exception &e) { - spdlog::error("statsReport failed: {}", e.what()); - } - - RecordTaskEnd(); -} - Worker::Worker(uint64_t id) : worker_id(id), state(WorkerState::FREE) { instance = this; std::string controller_addr = "localhost:50051"; @@ -488,14 +459,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)); - } - int64_t seconds_since_policy = (now - last_policy_time) / 1s; if (seconds_since_policy >= policy_interval) { std::thread([this]() { requestPolicyFromController(); }).detach(); From fd6914e16d500f3e389b6ce53edeb8f3fa6ae37f Mon Sep 17 00:00:00 2001 From: vovanbravin Date: Thu, 21 May 2026 20:09:09 +0300 Subject: [PATCH 04/12] fix build --- controller/cmd/grpc_server/main.go | 3 --- worker/include/worker.hpp | 1 + worker/src/worker.cpp | 37 ++++++++++++++++++++++++++++++ 3 files changed, 38 insertions(+), 3 deletions(-) diff --git a/controller/cmd/grpc_server/main.go b/controller/cmd/grpc_server/main.go index 7119d77..79039e0 100644 --- a/controller/cmd/grpc_server/main.go +++ b/controller/cmd/grpc_server/main.go @@ -1,11 +1,8 @@ package main import ( - "fmt" "log" "net" - "net/http" - "github.com/moevm/grpc_server/internal/config" "github.com/moevm/grpc_server/internal/grpcserver" "github.com/moevm/grpc_server/internal/manager" diff --git a/worker/include/worker.hpp b/worker/include/worker.hpp index be560c7..d3c73fc 100644 --- a/worker/include/worker.hpp +++ b/worker/include/worker.hpp @@ -81,6 +81,7 @@ class Worker { static Worker *getInstance(); void MainLoop(); + void statsReport(); void RecordPacketReceived(); void RecordPacketPassed(); void RecordPacketDropped(const std::string &reason); diff --git a/worker/src/worker.cpp b/worker/src/worker.cpp index 3fd2ffb..e5538f9 100644 --- a/worker/src/worker.cpp +++ b/worker/src/worker.cpp @@ -392,6 +392,35 @@ bool Worker::classify(const std::string &type, const std::string &target, } } +void Worker::statsReport() { + RecordTaskStart(); + + 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()); + RecordTaskEnd(); + return; + } + + spdlog::info("Stats sent successfully"); + + } catch (const std::exception &e) { + spdlog::error("statsReport failed: {}", e.what()); + } + + RecordTaskEnd(); +} + Worker::Worker(uint64_t id) : worker_id(id), state(WorkerState::FREE) { instance = this; std::string controller_addr = "localhost:50051"; @@ -459,6 +488,14 @@ 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)); + } + int64_t seconds_since_policy = (now - last_policy_time) / 1s; if (seconds_since_policy >= policy_interval) { std::thread([this]() { requestPolicyFromController(); }).detach(); From 01077260b7273b857143e2b92160efc6d97fa94f Mon Sep 17 00:00:00 2001 From: vovanbravin Date: Wed, 27 May 2026 19:26:27 +0300 Subject: [PATCH 05/12] add local count --- worker/BUILD | 4 -- worker/include/metrics_collector.hpp | 6 +- worker/include/worker.hpp | 16 +++-- worker/src/dpdk_filter/proc_packets.c | 16 +---- worker/src/main.cpp | 21 +----- worker/src/metrics_collector.cpp | 98 +++++++++++---------------- worker/src/worker.cpp | 67 +++++++++--------- 7 files changed, 90 insertions(+), 138 deletions(-) diff --git a/worker/BUILD b/worker/BUILD index 8df6f52..6f43ca8 100644 --- a/worker/BUILD +++ b/worker/BUILD @@ -74,10 +74,6 @@ cc_binary( "@curl//:curl", ], copts = [ - "-mssse3", - "-msse4.2", - "-mpclmul", - "-maes", "-I$(GENDIR)/..", ] + select({ "@platforms//cpu:x86_64": [ diff --git a/worker/include/metrics_collector.hpp b/worker/include/metrics_collector.hpp index 0b971f2..b1d645f 100644 --- a/worker/include/metrics_collector.hpp +++ b/worker/include/metrics_collector.hpp @@ -11,6 +11,7 @@ #include #include #include +#include class MetricsCollector { public: @@ -39,6 +40,9 @@ class MetricsCollector { }; Time time; + + uint64_t last_total{0}; + uint64_t last_non_idle{0}; }; void GetCPUUsage(); @@ -55,8 +59,6 @@ class MetricsCollector { prometheus::Counter *packets_received_counter; prometheus::Counter *packets_passed_counter; prometheus::Counter *packets_dropped_counter; - prometheus::Family *packets_dropped_by_reason_family; - prometheus::Family *blocked_domains_family; prometheus::Counter *tasks_completed_counter; prometheus::Histogram *task_duration_histogram; diff --git a/worker/include/worker.hpp b/worker/include/worker.hpp index d3c73fc..96ee7b6 100644 --- a/worker/include/worker.hpp +++ b/worker/include/worker.hpp @@ -62,12 +62,17 @@ class Worker { std::unique_ptr metrics_collector_; - std::atomic packets_received_count{0}; - std::atomic packets_passed_count{0}; - std::atomic packets_dropped_count{0}; + 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_PUSH_INTERVAL_SEC = 5; + + void pushMetricsToPrometheus(); public: - Worker(uint64_t id); + Worker(uint64_t id, const char *gateway_address, const char *gateway_port); ~Worker(); void initDPDK(int argc, char **argv); @@ -84,8 +89,7 @@ class Worker { void statsReport(); void RecordPacketReceived(); void RecordPacketPassed(); - void RecordPacketDropped(const std::string &reason); - void RecordDomainBlocked(const std::string &domain_or_ip); + void RecordPacketDropped(); void RecordTaskStart(); void RecordTaskEnd(); }; diff --git a/worker/src/dpdk_filter/proc_packets.c b/worker/src/dpdk_filter/proc_packets.c index 8310474..981ec95 100644 --- a/worker/src/dpdk_filter/proc_packets.c +++ b/worker/src/dpdk_filter/proc_packets.c @@ -4,8 +4,7 @@ extern void record_packet_received(); extern void record_packet_passed(); -extern void record_packet_droped(char *reason); -extern void record_domain_blocked(char *domain); +extern void record_packet_droped(); extern bool worker_classify(const char *type, const char *target, struct requested_classification *out_req); @@ -28,7 +27,7 @@ void package_sending_decision(bool solution_is_send, struct rte_mbuf *pkt, return; } - record_packet_droped("blocked"); + record_packet_droped(); rte_pktmbuf_free(pkt); } @@ -109,10 +108,6 @@ void pakage_processing(struct net_port *port_in, struct net_port *port_out, LOG_WARNING("Classification failed for IP %s", ip_str); } - if (!solution_is_send) { - record_domain_blocked(ip_str); - } - package_sending_decision(solution_is_send, pkts[i], port_out, queue_number); @@ -155,10 +150,6 @@ void pakage_processing(struct net_port *port_in, struct net_port *port_out, if (ret >= 0 && cached_node_domain) { package_sending_decision(cached_node_domain->solution_is_send, pkts[i], port_out, queue_number); - - if (!cached_node_domain->solution_is_send) { - record_domain_blocked(info_pac.domain); - } } else if (ret == -ENOENT) { struct requested_classification req_clas; @@ -175,9 +166,6 @@ void pakage_processing(struct net_port *port_in, struct net_port *port_out, LOG_WARNING("Classification failed for %s", info_pac.domain); } - if (!solution_is_send) { - record_domain_blocked(info_pac.domain); - } package_sending_decision(solution_is_send, pkts[i], port_out, queue_number); diff --git a/worker/src/main.cpp b/worker/src/main.cpp index 7795627..f5edf92 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) { @@ -41,7 +24,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; @@ -90,4 +73,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 cb91acc..66b9649 100644 --- a/worker/src/metrics_collector.cpp +++ b/worker/src/metrics_collector.cpp @@ -9,14 +9,13 @@ namespace { double GetMemoryUsed() { std::ifstream file("/proc/self/statm"); - if (!file.is_open()) { - return 0; - } + 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 @@ -61,18 +60,8 @@ MetricsCollector::MetricsCollector(const char *gateway_address, .Name("packets_dropped_total") .Help("Total number of packets dropped") .Register(*registry); - packets_dropped_counter = &packets_dropped_family.Add({}); - packets_dropped_by_reason_family = - &prometheus::BuildCounter() - .Name("packets_dropped_by_reason_total") - .Help("Packets dropped by reason") - .Register(*registry); - - blocked_domains_family = &prometheus::BuildCounter() - .Name("blocked_domains_total") - .Help("Number of blocked requests by domain/IP") - .Register(*registry); + packets_dropped_counter = &packets_dropped_family.Add({}); auto &tasks_completed_family = prometheus::BuildCounter() .Name("tasks_completed_total") @@ -159,39 +148,40 @@ MetricsCollector::~MetricsCollector() { void MetricsCollector::GetCPUUsage() { std::ifstream file("/proc/stat"); - CPUInfo::Time cur_time; - double percent; - - std::string cpu_name; - int ign; - - while (true) { - file >> cpu_name >> cur_time.user >> cur_time.user_low >> cur_time.sys >> - cur_time.idle >> ign >> ign >> ign >> ign >> ign >> ign; - - if (cpu_name.find("cpu") != 0) - break; - - 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 (!file.is_open()) return; + + std::string line; + while (std::getline(file, line)) { + if (line.find("cpu") != 0) break; + + std::istringstream iss(line); + std::string cpu_name; + long user, nice, sys, idle, iowait, irq, softirq, steal, guest, guest_nice; + + iss >> cpu_name >> user >> nice >> sys >> idle >> iowait >> irq >> softirq >> steal >> guest >> guest_nice; + + if (cpu_name.empty()) continue; + + uint64_t non_idle = user + nice + sys + irq + softirq + steal; + uint64_t total = non_idle + idle + iowait; + + 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; } - - cpu.time = cur_time; - cpu.gauge->Set(percent); } - + file.close(); } @@ -212,16 +202,6 @@ void MetricsCollector::IncrementPacketsDropped(const std::string &reason, if (packets_dropped_counter) { packets_dropped_counter->Increment(count); } - if (packets_dropped_by_reason_family) { - packets_dropped_by_reason_family->Add({{"reason", reason}}) - .Increment(count); - } -} - -void MetricsCollector::IncrementBlockedDomain(const std::string &domain_or_ip) { - if (blocked_domains_family) { - blocked_domains_family->Add({{"domain", domain_or_ip}}).Increment(); - } } void MetricsCollector::StartTask() { diff --git a/worker/src/worker.cpp b/worker/src/worker.cpp index e5538f9..8f1d530 100644 --- a/worker/src/worker.cpp +++ b/worker/src/worker.cpp @@ -46,10 +46,7 @@ void Worker::SetState(WorkerState new_state) { } void Worker::RecordPacketReceived() { - packets_received_count++; - if (metrics_collector_) { - metrics_collector_->IncrementPacketsReceived(1); - } + local_packets_received++; } extern "C" void record_packet_received() { @@ -62,10 +59,7 @@ extern "C" void record_packet_received() { } void Worker::RecordPacketPassed() { - packets_passed_count++; - if (metrics_collector_) { - metrics_collector_->IncrementPacketsPassed(1); - } + local_packets_passed++; } extern "C" void record_packet_passed() { @@ -77,35 +71,17 @@ extern "C" void record_packet_passed() { worker->RecordPacketPassed(); } -void Worker::RecordPacketDropped(const std::string &reason) { - packets_dropped_count++; - if (metrics_collector_) { - metrics_collector_->IncrementPacketsDropped(reason, 1); - } -} - -extern "C" void record_packet_droped(char *reson) { - Worker *worker = Worker::getInstance(); - if (!worker) { - fprintf(stderr, "worker_classify: worker is null\n"); - return; - } - worker->RecordPacketDropped(reson); -} - -void Worker::RecordDomainBlocked(const std::string &domain_or_ip) { - if (metrics_collector_) { - metrics_collector_->IncrementBlockedDomain(domain_or_ip); - } +void Worker::RecordPacketDropped() { + local_packets_dropped++; } -extern "C" void record_domain_blocked(char *endpoint) { +extern "C" void record_packet_droped() { Worker *worker = Worker::getInstance(); if (!worker) { fprintf(stderr, "worker_classify: worker is null\n"); return; } - worker->RecordDomainBlocked(endpoint); + worker->RecordPacketDropped(); } void Worker::RecordTaskStart() { @@ -120,6 +96,19 @@ void Worker::RecordTaskEnd() { } } +void Worker::pushMetricsToPrometheus() { + 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("total", dropped); + +} + void Worker::initDPDK(int argc, char **argv) { unsigned mbuf_quantity_in_pool = 8192; unsigned cache_size_per_kernel = 250; @@ -421,12 +410,18 @@ void Worker::statsReport() { RecordTaskEnd(); } -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); @@ -466,6 +461,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; @@ -503,11 +499,14 @@ void Worker::MainLoop() { policy_interval = MIN_POLICY_TIME + (rand() % (MAX_POLICY_TIME - MIN_POLICY_TIME + 1)); } + int64_t seconds_since_metrics = (now - last_metrics_push_time) / 1s; + if (seconds_since_metrics >= METRICS_PUSH_INTERVAL_SEC) { + pushMetricsToPrometheus(); + last_metrics_push_time = now; + } } - if (stop_flag) { - SetState(WorkerState::SHUTTING_DOWN); - } + pushMetricsToPrometheus(); if (stop_flag) { SetState(WorkerState::SHUTTING_DOWN); From 052c32a4ba56664bc2e6198864fb762c86dc0459 Mon Sep 17 00:00:00 2001 From: vovanbravin Date: Fri, 29 May 2026 13:24:59 +0300 Subject: [PATCH 06/12] fix: count packets when filter:off --- worker/src/dpdk_filter/proc_packets.c | 1 + worker/src/worker.cpp | 2 +- 2 files changed, 2 insertions(+), 1 deletion(-) diff --git a/worker/src/dpdk_filter/proc_packets.c b/worker/src/dpdk_filter/proc_packets.c index 123674e..e499997 100644 --- a/worker/src/dpdk_filter/proc_packets.c +++ b/worker/src/dpdk_filter/proc_packets.c @@ -55,6 +55,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; diff --git a/worker/src/worker.cpp b/worker/src/worker.cpp index 5ec7c75..fc1c306 100644 --- a/worker/src/worker.cpp +++ b/worker/src/worker.cpp @@ -1,4 +1,4 @@ -#include "worker.hpp" +#include "../include/worker.hpp" #include "communication.grpc.pb.h" #include "proc_packets.h" #include From 218a5c7fc6094dd1757b9e9356ced30574c2b1e0 Mon Sep 17 00:00:00 2001 From: vovanbravin Date: Fri, 29 May 2026 13:34:41 +0300 Subject: [PATCH 07/12] fix linter --- controller/cmd/grpc_server/main.go | 4 +- controller/internal/grpcserver/data_server.go | 2 +- worker/include/metrics_collector.hpp | 4 +- worker/include/worker.hpp | 2 +- worker/src/dpdk_filter/proc_packets.c | 3 +- worker/src/metrics_collector.cpp | 45 +++++++++++-------- worker/src/worker.cpp | 35 +++++++-------- 7 files changed, 49 insertions(+), 46 deletions(-) diff --git a/controller/cmd/grpc_server/main.go b/controller/cmd/grpc_server/main.go index 79039e0..04f12f5 100644 --- a/controller/cmd/grpc_server/main.go +++ b/controller/cmd/grpc_server/main.go @@ -1,8 +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" @@ -10,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/internal/grpcserver/data_server.go b/controller/internal/grpcserver/data_server.go index bd8a11d..15ddc7e 100644 --- a/controller/internal/grpcserver/data_server.go +++ b/controller/internal/grpcserver/data_server.go @@ -113,4 +113,4 @@ func (s *DataServer) SendStats(ctx context.Context, report *pb.StatsReport) (*em log.Printf("gRPC Stats from worker %d: blocked=%d allowed=%d", report.WorkerId, report.TotalBlocked, report.TotalAllowed) return &emptypb.Empty{}, nil -} \ No newline at end of file +} diff --git a/worker/include/metrics_collector.hpp b/worker/include/metrics_collector.hpp index b1d645f..e102767 100644 --- a/worker/include/metrics_collector.hpp +++ b/worker/include/metrics_collector.hpp @@ -8,10 +8,10 @@ #include #include #include +#include #include #include #include -#include class MetricsCollector { public: @@ -40,7 +40,7 @@ class MetricsCollector { }; Time time; - + uint64_t last_total{0}; uint64_t last_non_idle{0}; }; diff --git a/worker/include/worker.hpp b/worker/include/worker.hpp index 43d389b..ea78410 100644 --- a/worker/include/worker.hpp +++ b/worker/include/worker.hpp @@ -69,7 +69,7 @@ class Worker { std::atomic local_packets_dropped{0}; std::chrono::steady_clock::time_point last_metrics_push_time; - const int METRICS_PUSH_INTERVAL_SEC = 5; + const int METRICS_PUSH_INTERVAL_SEC = 5; void pushMetricsToPrometheus(); diff --git a/worker/src/dpdk_filter/proc_packets.c b/worker/src/dpdk_filter/proc_packets.c index e499997..2398b7f 100644 --- a/worker/src/dpdk_filter/proc_packets.c +++ b/worker/src/dpdk_filter/proc_packets.c @@ -55,7 +55,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(); + record_packet_received(); package_sending_decision(true, pkts[i], port_out, queue_number); } return; @@ -190,7 +190,6 @@ 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); diff --git a/worker/src/metrics_collector.cpp b/worker/src/metrics_collector.cpp index 66b9649..242cee2 100644 --- a/worker/src/metrics_collector.cpp +++ b/worker/src/metrics_collector.cpp @@ -9,12 +9,13 @@ namespace { double GetMemoryUsed() { std::ifstream file("/proc/self/statm"); - if (!file.is_open()) return 0; + if (!file.is_open()) + return 0; + + long total_pages = 0; + long rss_pages = 0; + file >> total_pages >> rss_pages; - long total_pages = 0; - long rss_pages = 0; - file >> total_pages >> rss_pages; - return rss_pages * (double)getpagesize(); } } // namespace @@ -148,40 +149,46 @@ MetricsCollector::~MetricsCollector() { void MetricsCollector::GetCPUUsage() { std::ifstream file("/proc/stat"); - if (!file.is_open()) return; + if (!file.is_open()) + return; std::string line; while (std::getline(file, line)) { - if (line.find("cpu") != 0) break; - + if (line.find("cpu") != 0) + break; + std::istringstream iss(line); std::string cpu_name; long user, nice, sys, idle, iowait, irq, softirq, steal, guest, guest_nice; - - iss >> cpu_name >> user >> nice >> sys >> idle >> iowait >> irq >> softirq >> steal >> guest >> guest_nice; - - if (cpu_name.empty()) continue; - + + iss >> cpu_name >> user >> nice >> sys >> idle >> iowait >> irq >> + softirq >> steal >> guest >> guest_nice; + + if (cpu_name.empty()) + continue; + uint64_t non_idle = user + nice + sys + irq + softirq + steal; uint64_t total = non_idle + idle + iowait; - + 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; + + 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(); } diff --git a/worker/src/worker.cpp b/worker/src/worker.cpp index fc1c306..c73ce5f 100644 --- a/worker/src/worker.cpp +++ b/worker/src/worker.cpp @@ -45,9 +45,7 @@ void Worker::SetState(WorkerState new_state) { } } -void Worker::RecordPacketReceived() { - local_packets_received++; -} +void Worker::RecordPacketReceived() { local_packets_received++; } extern "C" void record_packet_received() { Worker *worker = Worker::getInstance(); @@ -58,9 +56,7 @@ extern "C" void record_packet_received() { worker->RecordPacketReceived(); } -void Worker::RecordPacketPassed() { - local_packets_passed++; -} +void Worker::RecordPacketPassed() { local_packets_passed++; } extern "C" void record_packet_passed() { Worker *worker = Worker::getInstance(); @@ -71,9 +67,7 @@ extern "C" void record_packet_passed() { worker->RecordPacketPassed(); } -void Worker::RecordPacketDropped() { - local_packets_dropped++; -} +void Worker::RecordPacketDropped() { local_packets_dropped++; } extern "C" void record_packet_droped() { Worker *worker = Worker::getInstance(); @@ -97,16 +91,19 @@ void Worker::RecordTaskEnd() { } void Worker::pushMetricsToPrometheus() { - if (!metrics_collector_) return; - + 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("total", dropped); + if (received > 0) + metrics_collector_->IncrementPacketsReceived(received); + if (passed > 0) + metrics_collector_->IncrementPacketsPassed(passed); + if (dropped > 0) + metrics_collector_->IncrementPacketsDropped("total", dropped); } void Worker::initDPDK(int argc, char **argv) { @@ -415,7 +412,9 @@ void Worker::statsReport() { RecordTaskEnd(); } -Worker::Worker(uint64_t id, const char *gateway_address, const char *gateway_port) : 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")) { @@ -423,9 +422,7 @@ Worker::Worker(uint64_t id, const char *gateway_address, const char *gateway_por } metrics_collector_ = std::make_unique( - gateway_address, - gateway_port, - ("worker-" + std::to_string(id)).c_str()); + gateway_address, gateway_port, ("worker-" + std::to_string(id)).c_str()); auto channel = grpc::CreateChannel(controller_addr, grpc::InsecureChannelCredentials()); From 74e665856fbbbcc430742aa509f4eb892b7ef599 Mon Sep 17 00:00:00 2001 From: vovanbravin Date: Thu, 11 Jun 2026 17:06:20 +0300 Subject: [PATCH 08/12] fix: path and fprintf --- worker/docker-compose.yml | 5 +++++ worker/include/worker.hpp | 2 +- worker/src/worker.cpp | 12 ++++++------ 3 files changed, 12 insertions(+), 7 deletions(-) diff --git a/worker/docker-compose.yml b/worker/docker-compose.yml index b61d03a..c60d17c 100644 --- a/worker/docker-compose.yml +++ b/worker/docker-compose.yml @@ -5,6 +5,11 @@ services: build: context: . dockerfile: Dockerfile.cc_x86_to_x86 + environment: + - WORKER_ID=1 + - CONTROLLER_GRPC_ADDR=controller:50051 + - PUSH_GATEWAY_ADDR=pushgateway + - PUSH_GATEWAY_PORT=9091 volumes: - ./data:/data working_dir: /data diff --git a/worker/include/worker.hpp b/worker/include/worker.hpp index ea78410..5c95d5c 100644 --- a/worker/include/worker.hpp +++ b/worker/include/worker.hpp @@ -1,7 +1,7 @@ #ifndef WORKER_HPP #define WORKER_HPP -#include "../include/metrics_collector.hpp" +#include "metrics_collector.hpp" #include "communication.grpc.pb.h" #include "communication.pb.h" extern "C" { diff --git a/worker/src/worker.cpp b/worker/src/worker.cpp index c73ce5f..b513ec2 100644 --- a/worker/src/worker.cpp +++ b/worker/src/worker.cpp @@ -1,4 +1,4 @@ -#include "../include/worker.hpp" +#include "worker.hpp" #include "communication.grpc.pb.h" #include "proc_packets.h" #include @@ -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); @@ -50,7 +50,7 @@ void Worker::RecordPacketReceived() { local_packets_received++; } extern "C" void record_packet_received() { Worker *worker = Worker::getInstance(); if (!worker) { - fprintf(stderr, "worker_classify: worker is null\n"); + spdlog::error("record_packet_received: worker is null"); return; } worker->RecordPacketReceived(); @@ -61,7 +61,7 @@ void Worker::RecordPacketPassed() { local_packets_passed++; } extern "C" void record_packet_passed() { Worker *worker = Worker::getInstance(); if (!worker) { - fprintf(stderr, "worker_classify: worker is null\n"); + spdlog::error("record_packet_passed: worker is null"); return; } worker->RecordPacketPassed(); @@ -72,7 +72,7 @@ void Worker::RecordPacketDropped() { local_packets_dropped++; } extern "C" void record_packet_droped() { Worker *worker = Worker::getInstance(); if (!worker) { - fprintf(stderr, "worker_classify: worker is null\n"); + spdlog::error("record_packet_dropped: worker is null"); return; } worker->RecordPacketDropped(); @@ -516,4 +516,4 @@ void Worker::MainLoop() { if (stop_flag) { SetState(WorkerState::SHUTTING_DOWN); } -} +} \ No newline at end of file From 397a0cd807a668f75b60ae18cb7d6a3701386ca9 Mon Sep 17 00:00:00 2001 From: vovanbravin Date: Thu, 18 Jun 2026 21:43:10 +0300 Subject: [PATCH 09/12] fix --- worker/docker-compose.yml | 20 ++++----- worker/include/metrics_collector.hpp | 8 +--- worker/include/worker.hpp | 7 +-- worker/src/dpdk_filter/proc_packets.c | 11 ++++- worker/src/main.cpp | 7 --- worker/src/metrics_collector.cpp | 33 +------------- worker/src/worker.cpp | 64 ++------------------------- 7 files changed, 27 insertions(+), 123 deletions(-) diff --git a/worker/docker-compose.yml b/worker/docker-compose.yml index c60d17c..4a971ee 100644 --- a/worker/docker-compose.yml +++ b/worker/docker-compose.yml @@ -2,22 +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 - - PUSH_GATEWAY_ADDR=pushgateway - - PUSH_GATEWAY_PORT=9091 + - 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 e102767..0b06627 100644 --- a/worker/include/metrics_collector.hpp +++ b/worker/include/metrics_collector.hpp @@ -21,12 +21,8 @@ class MetricsCollector { void IncrementPacketsReceived(int count = 1); void IncrementPacketsPassed(int count = 1); - void IncrementPacketsDropped(const std::string &reason, int count = 1); - void IncrementBlockedDomain(const std::string &domain_or_ip); + void IncrementPacketsDropped(int count = 1); - void StartTask(); - void StopTask(); - void ObserveCollectionDuration(double seconds); private: struct CPUInfo { @@ -61,9 +57,7 @@ class MetricsCollector { prometheus::Counter *packets_dropped_counter; prometheus::Counter *tasks_completed_counter; - prometheus::Histogram *task_duration_histogram; - prometheus::Histogram *metrics_collection_duration; prometheus::Counter *push_errors_total; std::atomic is_running{true}; diff --git a/worker/include/worker.hpp b/worker/include/worker.hpp index 5c95d5c..a4e001b 100644 --- a/worker/include/worker.hpp +++ b/worker/include/worker.hpp @@ -1,9 +1,9 @@ #ifndef WORKER_HPP #define WORKER_HPP -#include "metrics_collector.hpp" #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" @@ -71,7 +71,7 @@ class Worker { std::chrono::steady_clock::time_point last_metrics_push_time; const int METRICS_PUSH_INTERVAL_SEC = 5; - void pushMetricsToPrometheus(); + void flushLocalCounters(); public: Worker(uint64_t id, const char *gateway_address, const char *gateway_port); @@ -88,12 +88,9 @@ class Worker { static Worker *getInstance(); void MainLoop(); - void statsReport(); void RecordPacketReceived(); void RecordPacketPassed(); void RecordPacketDropped(); - void RecordTaskStart(); - void RecordTaskEnd(); }; #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 2398b7f..e6d9a7a 100644 --- a/worker/src/dpdk_filter/proc_packets.c +++ b/worker/src/dpdk_filter/proc_packets.c @@ -2,6 +2,8 @@ #include "domain_cache.h" #include "ip_cache.h" #include +#include +#include extern void record_packet_received(); extern void record_packet_passed(); @@ -16,15 +18,18 @@ void package_sending_decision(bool solution_is_send, struct rte_mbuf *pkt, struct net_port *port_out, uint16_t queue_number) { if (solution_is_send) { - record_packet_passed(); - struct rte_mbuf *tx_pkt[1] = {pkt}; uint16_t ret = rte_eth_tx_burst(port_out->port_id, queue_number, tx_pkt, 1); if (ret < 1) { LOG_ERROR("Failed to send packet"); + record_packet_droped(); rte_pktmbuf_free(pkt); + return; } + + record_packet_passed(); + return; } @@ -207,6 +212,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_droped(); + rte_pktmbuf_free(pkts[i]); } } } diff --git a/worker/src/main.cpp b/worker/src/main.cpp index 8267fc0..3cc244b 100644 --- a/worker/src/main.cpp +++ b/worker/src/main.cpp @@ -34,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) diff --git a/worker/src/metrics_collector.cpp b/worker/src/metrics_collector.cpp index 242cee2..5e8a06a 100644 --- a/worker/src/metrics_collector.cpp +++ b/worker/src/metrics_collector.cpp @@ -204,39 +204,8 @@ void MetricsCollector::IncrementPacketsPassed(int count) { } } -void MetricsCollector::IncrementPacketsDropped(const std::string &reason, - int count) { +void MetricsCollector::IncrementPacketsDropped(int count) { if (packets_dropped_counter) { packets_dropped_counter->Increment(count); } -} - -void MetricsCollector::StartTask() { - is_task_running = true; - task_start = std::chrono::high_resolution_clock::now(); - task_processing_time_gauge->Set(0); - PushMetrics(); -} - -void MetricsCollector::StopTask() { - auto duration = std::chrono::duration( - std::chrono::high_resolution_clock::now() - task_start) - .count(); - - if (task_duration_histogram) { - task_duration_histogram->Observe(duration); - } - - if (tasks_completed_counter) { - tasks_completed_counter->Increment(); - } - - is_task_running = false; - task_processing_time_gauge->Set(0); - PushMetrics(); -} - -void MetricsCollector::ObserveCollectionDuration(double seconds) { - // Метод оставлен для совместимости - (void)seconds; } \ No newline at end of file diff --git a/worker/src/worker.cpp b/worker/src/worker.cpp index b513ec2..9dd3a01 100644 --- a/worker/src/worker.cpp +++ b/worker/src/worker.cpp @@ -78,19 +78,7 @@ extern "C" void record_packet_droped() { worker->RecordPacketDropped(); } -void Worker::RecordTaskStart() { - if (metrics_collector_) { - metrics_collector_->StartTask(); - } -} - -void Worker::RecordTaskEnd() { - if (metrics_collector_) { - metrics_collector_->StopTask(); - } -} - -void Worker::pushMetricsToPrometheus() { +void Worker::flushLocalCounters() { if (!metrics_collector_) return; @@ -103,7 +91,7 @@ void Worker::pushMetricsToPrometheus() { if (passed > 0) metrics_collector_->IncrementPacketsPassed(passed); if (dropped > 0) - metrics_collector_->IncrementPacketsDropped("total", dropped); + metrics_collector_->IncrementPacketsDropped(dropped); } void Worker::initDPDK(int argc, char **argv) { @@ -167,7 +155,6 @@ void Worker::forward_to_out(struct net_port *incoming_port, } void Worker::requestPolicyFromController() { - RecordTaskStart(); try { spdlog::info("Worker {} requests policy", worker_id); @@ -182,7 +169,6 @@ void Worker::requestPolicyFromController() { if (!status.ok()) { spdlog::error("GetPolicy failed: " + status.error_message()); - RecordTaskEnd(); return; } @@ -332,12 +318,10 @@ void Worker::requestPolicyFromController() { spdlog::error("requestPolicyFromController exception: {}", e.what()); } - RecordTaskEnd(); } bool Worker::classify(const std::string &type, const std::string &target, struct requested_classification *out_req) { - RecordTaskStart(); try { spdlog::info("Worker {} classifying '{}' as {}", worker_id, target, type); @@ -353,7 +337,6 @@ bool Worker::classify(const std::string &type, const std::string &target, auto status = stub_->Classify(&context, req, &resp); if (!status.ok()) { spdlog::error("Classify failed: " + status.error_message()); - RecordTaskEnd(); return false; } @@ -374,44 +357,13 @@ bool Worker::classify(const std::string &type, const std::string &target, CATEGORY_MAX_LEN - 1); } - RecordTaskEnd(); return true; } catch (const std::exception &e) { spdlog::error(std::string("classifyDomain: ") + e.what()); - RecordTaskEnd(); return false; } } -void Worker::statsReport() { - RecordTaskStart(); - - 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()); - RecordTaskEnd(); - return; - } - - spdlog::info("Stats sent successfully"); - - } catch (const std::exception &e) { - spdlog::error("statsReport failed: {}", e.what()); - } - - RecordTaskEnd(); -} - Worker::Worker(uint64_t id, const char *gateway_address, const char *gateway_port) : worker_id(id), state(WorkerState::FREE) { @@ -487,14 +439,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) { @@ -506,12 +450,12 @@ void Worker::MainLoop() { } int64_t seconds_since_metrics = (now - last_metrics_push_time) / 1s; if (seconds_since_metrics >= METRICS_PUSH_INTERVAL_SEC) { - pushMetricsToPrometheus(); + flushLocalCounters(); last_metrics_push_time = now; } } - pushMetricsToPrometheus(); + flushLocalCounters(); if (stop_flag) { SetState(WorkerState::SHUTTING_DOWN); From aaab892b805eda5640c69c2773677858dadcb137 Mon Sep 17 00:00:00 2001 From: vovanbravin Date: Fri, 19 Jun 2026 11:37:21 +0300 Subject: [PATCH 10/12] fix build --- worker/src/metrics_collector.cpp | 11 ----------- 1 file changed, 11 deletions(-) diff --git a/worker/src/metrics_collector.cpp b/worker/src/metrics_collector.cpp index 5e8a06a..aa9df6a 100644 --- a/worker/src/metrics_collector.cpp +++ b/worker/src/metrics_collector.cpp @@ -70,17 +70,6 @@ MetricsCollector::MetricsCollector(const char *gateway_address, .Register(*registry); tasks_completed_counter = &tasks_completed_family.Add({}); - auto &task_duration_family = prometheus::BuildHistogram() - .Name("task_duration_seconds") - .Help("Task execution duration in seconds") - .Register(*registry); - - prometheus::Histogram::BucketBoundaries task_buckets = { - 0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, - 0.5, 1.0, 2.5, 5.0, 10.0, 30.0, 60.0}; - task_duration_histogram = - &task_duration_family.Add({}, std::move(task_buckets)); - auto &push_errors_family = prometheus::BuildCounter() .Name("push_errors_total") .Help("Total number of push gateway errors") From 896ad1eea947d14bb5c6ae3debe123d619193bf4 Mon Sep 17 00:00:00 2001 From: vovanbravin Date: Thu, 2 Jul 2026 16:09:37 +0300 Subject: [PATCH 11/12] fix --- worker/include/metrics_collector.hpp | 13 +++---------- worker/include/worker.hpp | 2 +- worker/src/dpdk_filter/proc_packets.c | 10 ++++++---- worker/src/metrics_collector.cpp | 28 +++------------------------ worker/src/worker.cpp | 4 +--- 5 files changed, 14 insertions(+), 43 deletions(-) diff --git a/worker/include/metrics_collector.hpp b/worker/include/metrics_collector.hpp index 0b06627..b449b65 100644 --- a/worker/include/metrics_collector.hpp +++ b/worker/include/metrics_collector.hpp @@ -19,10 +19,9 @@ class MetricsCollector { const char *worker_name); ~MetricsCollector(); - void IncrementPacketsReceived(int count = 1); - void IncrementPacketsPassed(int count = 1); - void IncrementPacketsDropped(int count = 1); - + void IncrementPacketsReceived(uint64_t count = 1); + void IncrementPacketsPassed(uint64_t count = 1); + void IncrementPacketsDropped(uint64_t count = 1); private: struct CPUInfo { @@ -50,21 +49,15 @@ class MetricsCollector { std::unordered_map cpu_usage; prometheus::Gauge *memory_used_gauge; - prometheus::Gauge *task_processing_time_gauge; prometheus::Counter *packets_received_counter; prometheus::Counter *packets_passed_counter; prometheus::Counter *packets_dropped_counter; - prometheus::Counter *tasks_completed_counter; - prometheus::Counter *push_errors_total; std::atomic is_running{true}; std::thread thread; - - std::atomic is_task_running{false}; - std::chrono::time_point task_start; }; #endif \ No newline at end of file diff --git a/worker/include/worker.hpp b/worker/include/worker.hpp index a4e001b..b0d5740 100644 --- a/worker/include/worker.hpp +++ b/worker/include/worker.hpp @@ -69,7 +69,7 @@ class Worker { std::atomic local_packets_dropped{0}; std::chrono::steady_clock::time_point last_metrics_push_time; - const int METRICS_PUSH_INTERVAL_SEC = 5; + const int METRICS_FLUSH_INTERVAL_SEC = 5; void flushLocalCounters(); diff --git a/worker/src/dpdk_filter/proc_packets.c b/worker/src/dpdk_filter/proc_packets.c index e6d9a7a..aee4047 100644 --- a/worker/src/dpdk_filter/proc_packets.c +++ b/worker/src/dpdk_filter/proc_packets.c @@ -7,7 +7,7 @@ extern void record_packet_received(); extern void record_packet_passed(); -extern void record_packet_droped(); +extern void record_packet_dropped(); extern bool worker_classify(const char *type, const char *target, struct requested_classification *out_req); @@ -23,7 +23,7 @@ void package_sending_decision(bool solution_is_send, struct rte_mbuf *pkt, if (ret < 1) { LOG_ERROR("Failed to send packet"); - record_packet_droped(); + record_packet_dropped(); rte_pktmbuf_free(pkt); return; } @@ -33,7 +33,7 @@ void package_sending_decision(bool solution_is_send, struct rte_mbuf *pkt, return; } - record_packet_droped(); + record_packet_dropped(); rte_pktmbuf_free(pkt); } @@ -158,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"); @@ -212,7 +214,7 @@ 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_droped(); + record_packet_dropped(); rte_pktmbuf_free(pkts[i]); } } diff --git a/worker/src/metrics_collector.cpp b/worker/src/metrics_collector.cpp index aa9df6a..5162d3f 100644 --- a/worker/src/metrics_collector.cpp +++ b/worker/src/metrics_collector.cpp @@ -37,13 +37,6 @@ MetricsCollector::MetricsCollector(const char *gateway_address, .Register(*registry); memory_used_gauge = &memory_used_family.Add({}); - auto &task_processing_time_family = - prometheus::BuildGauge() - .Name("task_processing_time") - .Help("Task processing time (in seconds)") - .Register(*registry); - task_processing_time_gauge = &task_processing_time_family.Add({}); - auto &packets_received_family = prometheus::BuildCounter() .Name("packets_received_total") .Help("Total number of packets received") @@ -64,12 +57,6 @@ MetricsCollector::MetricsCollector(const char *gateway_address, packets_dropped_counter = &packets_dropped_family.Add({}); - auto &tasks_completed_family = prometheus::BuildCounter() - .Name("tasks_completed_total") - .Help("Total number of completed tasks") - .Register(*registry); - tasks_completed_counter = &tasks_completed_family.Add({}); - auto &push_errors_family = prometheus::BuildCounter() .Name("push_errors_total") .Help("Total number of push gateway errors") @@ -97,7 +84,6 @@ MetricsCollector::MetricsCollector(const char *gateway_address, gateway.RegisterCollectable(registry); thread = std::thread(&MetricsCollector::MainLoop, this); - is_task_running = false; } void MetricsCollector::MainLoop() { @@ -107,14 +93,6 @@ 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(); } } @@ -181,19 +159,19 @@ void MetricsCollector::GetCPUUsage() { file.close(); } -void MetricsCollector::IncrementPacketsReceived(int count) { +void MetricsCollector::IncrementPacketsReceived(uint64_t count) { if (packets_received_counter) { packets_received_counter->Increment(count); } } -void MetricsCollector::IncrementPacketsPassed(int count) { +void MetricsCollector::IncrementPacketsPassed(uint64_t count) { if (packets_passed_counter) { packets_passed_counter->Increment(count); } } -void MetricsCollector::IncrementPacketsDropped(int count) { +void MetricsCollector::IncrementPacketsDropped(uint64_t count) { if (packets_dropped_counter) { packets_dropped_counter->Increment(count); } diff --git a/worker/src/worker.cpp b/worker/src/worker.cpp index 9dd3a01..2e369aa 100644 --- a/worker/src/worker.cpp +++ b/worker/src/worker.cpp @@ -317,7 +317,6 @@ void Worker::requestPolicyFromController() { } catch (const std::exception &e) { spdlog::error("requestPolicyFromController exception: {}", e.what()); } - } bool Worker::classify(const std::string &type, const std::string &target, @@ -439,7 +438,6 @@ void Worker::MainLoop() { auto now = steady_clock::now(); - int64_t seconds_since_policy = (now - last_policy_time) / 1s; if (seconds_since_policy >= policy_interval) { std::thread([this]() { requestPolicyFromController(); }).detach(); @@ -449,7 +447,7 @@ void Worker::MainLoop() { 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_PUSH_INTERVAL_SEC) { + if (seconds_since_metrics >= METRICS_FLUSH_INTERVAL_SEC) { flushLocalCounters(); last_metrics_push_time = now; } From 98ebcd92375e86b458bf5e8bd29195160aa2e3a0 Mon Sep 17 00:00:00 2001 From: vovanbravin Date: Thu, 2 Jul 2026 16:13:01 +0300 Subject: [PATCH 12/12] fix record_packets_dropped --- worker/src/worker.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/worker/src/worker.cpp b/worker/src/worker.cpp index 2e369aa..f11bb73 100644 --- a/worker/src/worker.cpp +++ b/worker/src/worker.cpp @@ -69,7 +69,7 @@ extern "C" void record_packet_passed() { void Worker::RecordPacketDropped() { local_packets_dropped++; } -extern "C" void record_packet_droped() { +extern "C" void record_packet_dropped() { Worker *worker = Worker::getInstance(); if (!worker) { spdlog::error("record_packet_dropped: worker is null");