- add: metrics exporting

docker-compose.yml

@@ -103,7 +103,8 @@
       dockerfile: ./packet-service/Dockerfile
     restart: on-failure
     ports:
-      - "4000:4000"
+      - "29000:29000"
+      - "4001:4001"
     env_file:
       - ./packet-service/.env
       - .env

packet-service/Cargo.toml

@@ -26,6 +26,7 @@ warp = "0.3.7"
 dashmap = "6.1.0"
 uuid = { version = "1.11.0", features = ["v4"] }
 chrono = "0.4.39"
+prometheus_exporter = "0.8.5"
 
 [build-dependencies]
 tonic-build = "0.12.3"

packet-service/Dockerfile

@@ -21,5 +21,6 @@ RUN apk add --no-cache libssl3 libgcc
 COPY --from=builder /usr/src/packet-service/target/release/packet-service /usr/local/bin/packet-service
 
 EXPOSE 29000
+EXPOSE 4001
 
 CMD ["packet-service"]

packet-service/src/main.rs

@@ -6,6 +6,8 @@ use crate::metrics::{ACTIVE_CONNECTIONS, PACKETS_RECEIVED};
 use crate::packet::Packet;
 use crate::router::PacketRouter;
 use dotenv::dotenv;
+use prometheus::{self, Encoder, TextEncoder};
+use prometheus_exporter;
 use std::collections::HashMap;
 use std::env;
 use std::error::Error;
@@ -66,6 +68,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     // Set the gRPC server address
     let addr = env::var("LISTEN_ADDR").unwrap_or_else(|_| "0.0.0.0".to_string());
     let port = env::var("PACKET_SERVICE_PORT").unwrap_or_else(|_| "4000".to_string());
+    let metrics_port = env::var("PACKET_METRICS_PORT").unwrap_or_else(|_| "4001".to_string());
     let health_port = env::var("HEALTH_CHECK_PORT").unwrap_or_else(|_| "8082".to_string());
 
     let consul_url = env::var("CONSUL_URL").unwrap_or_else(|_| "http://127.0.0.1:8500".to_string());
@@ -93,7 +96,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         meta,
         &health_check_url,
     )
-    .await?;
+        .await?;
 
     // Start health-check endpoint
     consul_registration::start_health_check(addr.as_str()).await?;
@@ -153,6 +156,9 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         }
     });
 
+    let binding = format!("{}:{}", &addr, metrics_port);
+    prometheus_exporter::start(binding.parse().unwrap()).unwrap();
+
     utils::signal_handler::wait_for_signal().await;
 
     consul_registration::deregister_service(&consul_url, service_id.as_str())

packet-service/src/metrics.rs

@@ -11,6 +11,12 @@ lazy_static! {
         "Total number of packets received"
     ).unwrap();
 
+    // Counter to track the number of packets sent
+    pub static ref PACKETS_SENT: Counter = register_counter!(
+        "packets_sent_total",
+        "Total number of packets sent"
+    ).unwrap();
+
     // Gauge to track the number of active connections
     pub static ref ACTIVE_CONNECTIONS: Gauge = register_gauge!(
         "active_connections",

packet-service/src/packet.rs

@@ -1,3 +1,4 @@
+use crate::metrics::PACKETS_SENT;
 use crate::packet_type::PacketType;
 use bincode::{Decode, Encode};
 use std::error::Error;
@@ -112,6 +113,7 @@ impl Packet {
 pub async fn send_packet(stream: &mut TcpStream, packet: &Packet) -> Result<(), std::io::Error> {
     let data = packet.to_raw();
     debug!("Sending '{:#X}' bytes of data. {:?}", data.len(), data);
+    PACKETS_SENT.inc();
     stream.write_all(&data).await?;
     Ok(())
 }

packet-service/src/router.rs

@@ -3,7 +3,7 @@ use crate::bufferpool::BufferPool;
 use crate::character_client::CharacterClient;
 use crate::connection_service::ConnectionService;
 use crate::handlers::*;
-use crate::metrics::{ACTIVE_CONNECTIONS, PACKETS_RECEIVED};
+use crate::metrics::{ACTIVE_CONNECTIONS, PACKETS_RECEIVED, PACKET_PROCESSING_TIME};
 use crate::packet::Packet;
 use crate::packet_type::PacketType;
 use std::error::Error;
@@ -46,6 +46,7 @@ impl PacketRouter {
 
             PACKETS_RECEIVED.inc();
 
+            let timer = PACKET_PROCESSING_TIME.start_timer();
             // Process the packet
             match Packet::from_raw(&buffer[..packet_size]) {
                 Ok(packet) => {
@@ -56,6 +57,7 @@ impl PacketRouter {
                 }
                 Err(e) => warn!("Failed to parse packet: {}", e),
             }
+            timer.stop_and_record();
 
             pool.release(buffer).await;
         }