feat: add metrics

controller/cmd/grpc_server/main.go

@@ -1,16 +1,15 @@
 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"
 	adminPb "github.com/moevm/grpc_server/pkg/proto/admin_service"
 	commPb "github.com/moevm/grpc_server/pkg/proto/communication"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/reflection"
+	"log"
+	"net"
 )
 
 func main() {

controller/pkg/proto/communication/communication.proto

@@ -62,4 +62,4 @@ message StatsReport {
     uint64 total_blocked = 3;
     uint64 total_allowed = 4;
     repeated ResourceStats resources = 5;
-}
+}

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"
+        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"

worker/include/metrics_collector.hpp

@@ -1,15 +1,29 @@
 #ifndef METRICS_COLLECTOR_HPP
 #define METRICS_COLLECTOR_HPP
 
+#include <atomic>
 #include <chrono>
+#include <prometheus/counter.h>
 #include <prometheus/gateway.h>
 #include <prometheus/gauge.h>
+#include <prometheus/histogram.h>
 #include <prometheus/registry.h>
+#include <sstream>
+#include <string>
+#include <thread>
+#include <unordered_map>
 
 class MetricsCollector {
-  prometheus::Gateway gateway;
-  std::shared_ptr<prometheus::Registry> 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<std::string, CPUInfo> 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<bool> is_running{true};
-  std::thread thread;
+  prometheus::Gateway gateway;
+  std::shared_ptr<prometheus::Registry> registry;
 
-  std::atomic<bool> is_task_running;
-  std::chrono::time_point<std::chrono::high_resolution_clock> task_start;
+  std::unordered_map<std::string, CPUInfo> 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<bool> is_running{true};
+  std::thread thread;
 };
 
-#endif
+#endif

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 <atomic>
 #include <cstdint>
 #include <grpcpp/grpcpp.h>
 #include <memory>
+#include <mutex>
 #include <rte_eal.h>
 #include <rte_ethdev.h>
 #include <rte_mbuf.h>
 #include <rte_mempool.h>
+#include <rte_timer.h>
 
 #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<MetricsCollector> metrics_collector_;
+
+  std::atomic<uint64_t> local_packets_received{0};
+  std::atomic<uint64_t> local_packets_passed{0};
+  std::atomic<uint64_t> 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

worker/src/dpdk_filter/proc_packets.c

@@ -2,6 +2,12 @@
 #include "domain_cache.h"
 #include "ip_cache.h"
 #include <stdatomic.h>
+#include <stdbool.h>
+#include <stdint.h>
+
+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,18 +60,22 @@ 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;
   }
 
   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]);
       }
     }
   }
-}
+}

worker/src/main.cpp

@@ -3,23 +3,6 @@
 
 #include <spdlog/spdlog.h>
 
-class FiltrWorker : public Worker {
-  MetricsCollector metrics_collector;
-
-protected:
-  void ProcessTask(const std::vector<char> &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;
-}
+}