Version exchange and logging

.github/workflows/ci.yml

@@ -23,7 +23,7 @@ jobs:
     runs-on:
       - codebuild-defguard-core-runner-${{ github.run_id }}-${{ github.run_attempt }}
 
-    container: rust:1
+    container: public.ecr.aws/docker/library/rust:1
 
     services:
       postgres:
@@ -72,7 +72,9 @@ jobs:
           cargo clippy --all-targets --all-features -- -D warnings
 
       - name: Run cargo deny
-        uses: EmbarkStudios/cargo-deny-action@v2
+        run: |
+          cargo install cargo-deny
+          cargo deny check
 
       - name: Install nextest
         uses: taiki-e/install-action@nextest

Cargo.toml

@@ -14,6 +14,7 @@ resolver = "2"
 defguard_core = { path = "./crates/defguard_core", version = "1.5.0" }
 defguard_event_logger = { path = "./crates/defguard_event_logger", version = "0.0.0" }
 defguard_event_router = { path = "./crates/defguard_event_router", version = "0.0.0" }
+defguard_version = { path = "./crates/defguard_version", version = "0.0.0" }
 defguard_web_ui = { path = "./crates/defguard_web_ui", version = "0.0.0" }
 model_derive = { path = "./crates/model_derive", version = "0.0.0" }
 

crates/defguard/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "defguard"
-version = "1.5.0"
+version = "0.0.0"
 edition.workspace = true
 license-file.workspace = true
 homepage.workspace = true
@@ -12,6 +12,7 @@ rust-version.workspace = true
 defguard_core = { workspace = true }
 defguard_event_router = { workspace = true }
 defguard_event_logger = { workspace = true }
+defguard_version = { workspace = true }
 
 # external dependencies
 anyhow = { workspace = true }

crates/defguard/src/main.rs

@@ -1,9 +1,11 @@
+use bytes::Bytes;
+use secrecy::ExposeSecret;
 use std::{
     fs::read_to_string,
     sync::{Arc, Mutex},
 };
+use tokio::sync::{broadcast, mpsc::unbounded_channel};
 
-use bytes::Bytes;
 use defguard_core::{
     SERVER_CONFIG, VERSION,
     auth::failed_login::FailedLoginMap,
@@ -28,9 +30,6 @@ use defguard_core::{
 };
 use defguard_event_logger::{message::EventLoggerMessage, run_event_logger};
 use defguard_event_router::{RouterReceiverSet, run_event_router};
-use secrecy::ExposeSecret;
-use tokio::sync::{broadcast, mpsc::unbounded_channel};
-use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
 
 #[macro_use]
 extern crate tracing;
@@ -42,14 +41,12 @@ async fn main() -> Result<(), anyhow::Error> {
     }
     let config = DefGuardConfig::new();
     SERVER_CONFIG.set(config.clone())?;
-    // initialize tracing
-    tracing_subscriber::registry()
-        .with(
-            tracing_subscriber::EnvFilter::try_from_default_env()
-                .unwrap_or_else(|_| format!("{},h2=info", config.log_level).into()),
-        )
-        .with(tracing_subscriber::fmt::layer())
-        .init();
+
+    // initialize tracing with version formatter
+    defguard_version::tracing::init(
+        defguard_version::Version::parse(VERSION)?,
+        &config.log_level,
+    )?;
 
     info!("Starting ... version v{}", VERSION);
     debug!("Using config: {config:?}");

crates/defguard_core/Cargo.toml

@@ -10,6 +10,7 @@ rust-version.workspace = true
 [dependencies]
 # internal crates
 defguard_web_ui = { workspace = true }
+defguard_version = { workspace = true }
 model_derive = { workspace = true }
 
 # external dependencies
@@ -85,6 +86,7 @@ strum = { workspace = true }
 strum_macros = { workspace = true }
 bytes = { workspace = true }
 ed25519-dalek = { version = "2.2.0", features = ["rand_core"] }
+tower = "0.5.2"
 
 # https://github.com/juhaku/utoipa/issues/1345
 [dependencies.zip]

crates/defguard_core/src/grpc/gateway/mod.rs

@@ -8,6 +8,7 @@ use std::{
 
 use chrono::{DateTime, TimeDelta, Utc};
 use client_state::ClientMap;
+use defguard_version::{DefguardComponent, version_info_from_metadata};
 use sqlx::{Error as SqlxError, PgExecutor, PgPool, query};
 use thiserror::Error;
 use tokio::{
@@ -127,6 +128,14 @@ impl WireguardNetwork<Id> {
     }
 }
 
+/// Utility struct encapsulating commonly extracted metadata fields during gRPC communication.
+struct GatewayMetadata {
+    network_id: Id,
+    hostname: String,
+    version: String,
+    info: String,
+}
+
 impl GatewayServer {
     /// Create new gateway server instance
     #[must_use]
@@ -158,10 +167,10 @@ impl GatewayServer {
     }
 
     // parse network id from gateway request metadata from intercepted information from JWT token
-    fn get_network_id_from_metadata(metadata: &MetadataMap) -> Option<i64> {
+    fn get_network_id_from_metadata(metadata: &MetadataMap) -> Option<Id> {
         if let Some(ascii_value) = metadata.get("gateway_network_id") {
             if let Ok(slice) = ascii_value.clone().to_str() {
-                if let Ok(id) = slice.parse::<i64>() {
+                if let Ok(id) = slice.parse::<Id>() {
                     return Some(id);
                 }
             }
@@ -276,6 +285,17 @@ impl GatewayServer {
 
         Ok(user)
     }
+
+    /// Utility function extracting metadata fields during gRPC communication.
+    fn extract_metadata(metadata: &MetadataMap) -> Result<GatewayMetadata, Status> {
+        let (version, info) = version_info_from_metadata(metadata);
+        Ok(GatewayMetadata {
+            network_id: Self::get_network_id(metadata)?,
+            hostname: Self::get_gateway_hostname(metadata)?,
+            version,
+            info,
+        })
+    }
 }
 
 fn gen_config(
@@ -721,11 +741,17 @@ impl gateway_service_server::GatewayService for GatewayServer {
         &self,
         request: Request<tonic::Streaming<StatsUpdate>>,
     ) -> Result<Response<()>, Status> {
-        let network_id = Self::get_network_id(request.metadata())?;
-        let gateway_hostname = Self::get_gateway_hostname(request.metadata())?;
+        let GatewayMetadata {
+            network_id,
+            hostname,
+            version,
+            info,
+        } = Self::extract_metadata(request.metadata())?;
         let mut stream = request.into_inner();
         let mut disconnect_timer = interval(Duration::from_secs(PEER_DISCONNECT_INTERVAL));
 
+        let span = tracing::info_span!("gateway_stats", component = %DefguardComponent::Gateway, version, info);
+        let _guard = span.enter();
         loop {
             // wait for a message or update client map at least once a mninute if no messages are received
             let stats_update = tokio::select! {
@@ -821,7 +847,7 @@ impl gateway_service_server::GatewayService for GatewayServer {
                                 // mark new VPN client as connected
                                 client_map.connect_vpn_client(
                                     network_id,
-                                    &gateway_hostname,
+                                    &hostname,
                                     &public_key,
                                     &device,
                                     &user,
@@ -884,8 +910,14 @@ impl gateway_service_server::GatewayService for GatewayServer {
         request: Request<ConfigurationRequest>,
     ) -> Result<Response<Configuration>, Status> {
         debug!("Sending configuration to gateway client.");
-        let network_id = Self::get_network_id(request.metadata())?;
-        let hostname = Self::get_gateway_hostname(request.metadata())?;
+        let GatewayMetadata {
+            network_id,
+            hostname,
+            version,
+            info,
+        } = Self::extract_metadata(request.metadata())?;
+        let span = tracing::info_span!("gateway_config", component = %DefguardComponent::Gateway, version, info);
+        let _guard = span.enter();
 
         let mut conn = self.pool.acquire().await.map_err(|e| {
             error!("Failed to acquire DB connection: {e}");
@@ -956,22 +988,28 @@ impl gateway_service_server::GatewayService for GatewayServer {
     }
 
     async fn updates(&self, request: Request<()>) -> Result<Response<Self::UpdatesStream>, Status> {
-        let gateway_network_id = Self::get_network_id(request.metadata())?;
-        let hostname = Self::get_gateway_hostname(request.metadata())?;
+        let GatewayMetadata {
+            network_id,
+            hostname,
+            version,
+            info,
+        } = Self::extract_metadata(request.metadata())?;
+        let span = tracing::info_span!("gateway_updates", component = %DefguardComponent::Gateway, version, info);
+        let _guard = span.enter();
 
-        let Some(network) = WireguardNetwork::find_by_id(&self.pool, gateway_network_id)
+        let Some(network) = WireguardNetwork::find_by_id(&self.pool, network_id)
             .await
             .map_err(|_| {
-                error!("Failed to fetch network {gateway_network_id} from the database");
+                error!("Failed to fetch network {network_id} from the database");
                 Status::new(
                     Code::Internal,
-                    format!("Failed to retrieve network {gateway_network_id} from the database"),
+                    format!("Failed to retrieve network {network_id} from the database"),
                 )
             })?
         else {
             return Err(Status::new(
                 Code::Internal,
-                format!("Network with id {gateway_network_id} not found"),
+                format!("Network with id {network_id} not found"),
             ));
         };
 
@@ -981,9 +1019,9 @@ impl gateway_service_server::GatewayService for GatewayServer {
         let events_rx = self.wireguard_tx.subscribe();
         let mut state = self.gateway_state.lock().unwrap();
         state
-            .connect_gateway(gateway_network_id, &hostname, &self.pool)
+            .connect_gateway(network_id, &hostname, &self.pool)
             .map_err(|err| {
-                error!("Failed to connect gateway on network {gateway_network_id}: {err}");
+                error!("Failed to connect gateway on network {network_id}: {err}");
                 Status::new(
                     Code::Internal,
                     "Failed to connect gateway on network {gateway_network_id}",
@@ -993,20 +1031,15 @@ impl gateway_service_server::GatewayService for GatewayServer {
         // clone here before moving into a closure
         let gateway_hostname = hostname.clone();
         let handle = tokio::spawn(async move {
-            let mut update_handler = GatewayUpdatesHandler::new(
-                gateway_network_id,
-                network,
-                gateway_hostname,
-                events_rx,
-                tx,
-            );
+            let mut update_handler =
+                GatewayUpdatesHandler::new(network_id, network, gateway_hostname, events_rx, tx);
             update_handler.run().await;
         });
 
         Ok(Response::new(GatewayUpdatesStream::new(
             handle,
             rx,
-            gateway_network_id,
+            network_id,
             hostname,
             Arc::clone(&self.gateway_state),
             self.pool.clone(),