Add optional proxy socket limit info to utils::SystemInfo()

rpc/core/src/model/message.rs

@@ -1899,6 +1899,7 @@ pub struct GetSystemInfoResponse {
     pub cpu_physical_cores: u16,
     pub total_memory: u64,
     pub fd_limit: u32,
+    pub proxy_socket_limit_per_cpu_core: Option<u32>,
 }
 
 impl std::fmt::Debug for GetSystemInfoResponse {
@@ -1910,35 +1911,39 @@ impl std::fmt::Debug for GetSystemInfoResponse {
             .field("cpu_physical_cores", &self.cpu_physical_cores)
             .field("total_memory", &self.total_memory)
             .field("fd_limit", &self.fd_limit)
+            .field("proxy_socket_limit_per_cpu_core", &self.proxy_socket_limit_per_cpu_core)
             .finish()
     }
 }
 
 impl Serializer for GetSystemInfoResponse {
     fn serialize<W: std::io::Write>(&self, writer: &mut W) -> std::io::Result<()> {
-        store!(u16, &1, writer)?;
+        store!(u16, &2, writer)?;
         store!(String, &self.version, writer)?;
         store!(Option<Vec<u8>>, &self.system_id, writer)?;
         store!(Option<Vec<u8>>, &self.git_hash, writer)?;
         store!(u16, &self.cpu_physical_cores, writer)?;
         store!(u64, &self.total_memory, writer)?;
         store!(u32, &self.fd_limit, writer)?;
+        store!(Option<u32>, &self.proxy_socket_limit_per_cpu_core, writer)?;
 
         Ok(())
     }
 }
 
 impl Deserializer for GetSystemInfoResponse {
     fn deserialize<R: std::io::Read>(reader: &mut R) -> std::io::Result<Self> {
-        let _version = load!(u16, reader)?;
+        let payload_version = load!(u16, reader)?;
         let version = load!(String, reader)?;
         let system_id = load!(Option<Vec<u8>>, reader)?;
         let git_hash = load!(Option<Vec<u8>>, reader)?;
         let cpu_physical_cores = load!(u16, reader)?;
         let total_memory = load!(u64, reader)?;
         let fd_limit = load!(u32, reader)?;
 
-        Ok(Self { version, system_id, git_hash, cpu_physical_cores, total_memory, fd_limit })
+        let proxy_socket_limit_per_cpu_core = if payload_version > 1 { load!(Option<u32>, reader)? } else { None };
+
+        Ok(Self { version, system_id, git_hash, cpu_physical_cores, total_memory, fd_limit, proxy_socket_limit_per_cpu_core })
     }
 }
 

rpc/core/src/model/tests.rs

@@ -970,6 +970,7 @@ mod mockery {
                 cpu_physical_cores: mock(),
                 total_memory: mock(),
                 fd_limit: mock(),
+                proxy_socket_limit_per_cpu_core: mock(),
             }
         }
     }

rpc/grpc/core/proto/rpc.proto

@@ -851,6 +851,7 @@ message GetSystemInfoResponseMessage{
   uint32 coreNum = 4;
   uint64 totalMemory = 5;
   uint32 fdLimit = 6;
+  uint32 proxySocketLimitPerCpuCore = 7;
   RPCError error = 1000;
 }
 

rpc/grpc/core/src/convert/message.rs

@@ -480,6 +480,7 @@ from!(item: RpcResult<&spectre_rpc_core::GetSystemInfoResponse>, protowire::GetS
         total_memory : item.total_memory,
         core_num : item.cpu_physical_cores as u32,
         fd_limit : item.fd_limit,
+        proxy_socket_limit_per_cpu_core : item.proxy_socket_limit_per_cpu_core.unwrap_or_default(),
         error: None,
     }
 });
@@ -962,6 +963,7 @@ try_from!(item: &protowire::GetSystemInfoResponseMessage, RpcResult<spectre_rpc_
         total_memory: item.total_memory,
         cpu_physical_cores: item.core_num as u16,
         fd_limit: item.fd_limit,
+        proxy_socket_limit_per_cpu_core : (item.proxy_socket_limit_per_cpu_core > 0).then_some(item.proxy_socket_limit_per_cpu_core),
     }
 });
 

rpc/service/src/service.rs

@@ -1067,6 +1067,7 @@ NOTE: This error usually indicates an RPC conversion error between the node and
             cpu_physical_cores: self.system_info.cpu_physical_cores,
             total_memory: self.system_info.total_memory,
             fd_limit: self.system_info.fd_limit,
+            proxy_socket_limit_per_cpu_core: self.system_info.proxy_socket_limit_per_cpu_core,
         };
 
         Ok(response)

utils/src/sysinfo.rs

@@ -2,21 +2,32 @@ use crate::fd_budget;
 use crate::git;
 use crate::hex::ToHex;
 use sha2::{Digest, Sha256};
-use std::fs::File;
+use std::fs::{read_to_string, File};
 use std::io::Read;
+use std::path::PathBuf;
+// use std::fs::read_to_string;
 use std::sync::OnceLock;
 
 static SYSTEM_INFO: OnceLock<SystemInfo> = OnceLock::new();
 
 #[derive(Clone)]
 pub struct SystemInfo {
+    /// unique system (machine) identifier
     pub system_id: Option<Vec<u8>>,
+    /// full git commit hash
     pub git_hash: Option<Vec<u8>>,
+    /// short git commit hash
     pub git_short_hash: Option<Vec<u8>>,
+    /// crate (workspace) version
     pub version: String,
+    /// number of physical CPU cores
     pub cpu_physical_cores: u16,
+    /// total system memory in bytes
     pub total_memory: u64,
+    /// file descriptor limit of the current process
     pub fd_limit: u32,
+    /// maximum number of sockets per CPU core
+    pub proxy_socket_limit_per_cpu_core: Option<u32>,
 }
 
 // provide hex encoding for system_id, git_hash, and git_short_hash
@@ -30,6 +41,7 @@ impl std::fmt::Debug for SystemInfo {
             .field("cpu_physical_cores", &self.cpu_physical_cores)
             .field("total_memory", &self.total_memory)
             .field("fd_limit", &self.fd_limit)
+            .field("proxy_socket_limit_per_cpu_core", &self.proxy_socket_limit_per_cpu_core)
             .finish()
     }
 }
@@ -46,8 +58,18 @@ impl Default for SystemInfo {
             let git_hash = git::hash();
             let git_short_hash = git::short_hash();
             let version = git::version();
+            let proxy_socket_limit_per_cpu_core = Self::try_proxy_socket_limit_per_cpu_core();
 
-            SystemInfo { system_id, git_hash, git_short_hash, version, cpu_physical_cores, total_memory, fd_limit }
+            SystemInfo {
+                system_id,
+                git_hash,
+                git_short_hash,
+                version,
+                cpu_physical_cores,
+                total_memory,
+                fd_limit,
+                proxy_socket_limit_per_cpu_core,
+            }
         });
         (*system_info).clone()
     }
@@ -72,10 +94,34 @@ impl SystemInfo {
         sha256.update(some_id.as_bytes());
         Some(sha256.finalize().to_vec())
     }
+
+    fn try_proxy_socket_limit_per_cpu_core() -> Option<u32> {
+        let nginx_config_path = PathBuf::from("/etc/nginx/nginx.conf");
+        if nginx_config_path.exists() {
+            read_to_string(nginx_config_path)
+                .ok()
+                .and_then(|content| content.lines().find(|line| line.trim().starts_with("worker_connections")).map(String::from))
+                .and_then(|line| line.split_whitespace().nth(1).map(|v| v.replace(";", "")))
+                .and_then(|value| value.parse::<u32>().ok())
+        } else {
+            None
+        }
+    }
 }
 
 impl AsRef<SystemInfo> for SystemInfo {
     fn as_ref(&self) -> &SystemInfo {
         self
     }
 }
+
+// #[cfg(test)]
+// mod tests {
+//     use super::*;
+
+//     #[test]
+//     fn test_system_info() {
+//         let system_info = SystemInfo::default();
+//         println!("{:#?}", system_info);
+//     }
+// }