Merge branch 'master' into codex/fix-node-permission-codec

Author: ryankert01

Cargo.lock

@@ -64,6 +64,7 @@ members = [
     "core/shard",
     "core/shard_allocator",
     "core/simulator",
+    "core/system_stats",
     "core/tools",
     "examples/rust",
 ]
@@ -305,6 +306,7 @@ strum = { version = "0.28.0", features = ["derive"] }
 strum_macros = "0.28.0"
 syn = { version = "2", features = ["full", "extra-traits"] }
 sysinfo = "0.39.5"
+system_stats = { path = "core/system_stats" }
 tempfile = "3.27.0"
 terminal_size = { version = "0.4.4" }
 test-case = "3.3.1"

Cargo.toml

@@ -122,6 +122,8 @@ pub enum IggyError {
     InvalidPersonalAccessTokenExpiry = 56,
     #[error("Request transiently not committed; retry")]
     TransientNotCommitted = 57,
+    #[error("Request transiently not accepted; retry, on any replica")]
+    TransientNotAccepted = 58,
     #[error("Not connected")]
     NotConnected = 61,
     #[error("Client shutdown")]

core/common/src/error/iggy_error.rs

@@ -15,7 +15,6 @@
 // specific language governing permissions and limitations
 // under the License.
 
-use crate::traits::binary_auth::fail_if_not_authenticated;
 use crate::wire_conversions::cluster_metadata_from_wire;
 use crate::{BinaryClient, ClusterClient, ClusterMetadata, IggyError};
 use iggy_binary_protocol::codec::WireEncode;
@@ -26,7 +25,6 @@ use iggy_binary_protocol::responses::system::get_cluster_metadata::ClusterMetada
 #[async_trait::async_trait]
 impl<B: BinaryClient> ClusterClient for B {
     async fn get_cluster_metadata(&self) -> Result<ClusterMetadata, IggyError> {
-        fail_if_not_authenticated(self).await?;
         let response = self
             .send_raw_with_response(
                 GET_CLUSTER_METADATA_CODE,

core/common/src/traits/binary_impls/cluster.rs

@@ -23,6 +23,7 @@ use async_trait::async_trait;
 pub trait ClusterClient {
     /// Get the metadata of the cluster including node information, roles, and status.
     ///
-    /// Authentication is required.
+    /// Served pre-auth so an unauthenticated client can locate the cluster
+    /// leader before signing in; the server applies its own policy.
     async fn get_cluster_metadata(&self) -> Result<ClusterMetadata, IggyError>;
 }

core/common/src/traits/cluster_client.rs

@@ -70,6 +70,7 @@ serde_with = { workspace = true }
 serde_yaml_ng = { workspace = true }
 strum = { workspace = true }
 sysinfo = { workspace = true }
+system_stats = { workspace = true }
 thiserror = { workspace = true }
 tokio = { workspace = true }
 toml = { workspace = true }

core/connectors/runtime/Cargo.toml

@@ -39,6 +39,7 @@ use std::{
     env,
     sync::{Arc, atomic::AtomicU32},
 };
+use system_stats::capture_allowed_cpus;
 use tracing::{error, info};
 
 mod api;
@@ -114,6 +115,7 @@ fn print_ascii_art(text: &str) {
 
 #[tokio::main]
 async fn main() -> Result<(), RuntimeError> {
+    capture_allowed_cpus();
     Args::parse();
     print_ascii_art("Iggy Connectors");
 

core/connectors/runtime/src/main.rs

@@ -20,31 +20,17 @@ use crate::metrics::ConnectorType;
 use iggy_common::{IggyTimestamp, SemanticVersion};
 use iggy_connector_sdk::api::{ConnectorRuntimeStats, ConnectorStats};
 use std::str::FromStr;
-use std::sync::Arc;
+use std::sync::{Arc, Mutex, OnceLock, PoisonError};
 use sysinfo::System;
+use system_stats::SystemProbe;
 
 const VERSION: &str = env!("CARGO_PKG_VERSION");
 const SEMANTIC_VERSION: SemanticVersion = SemanticVersion::parse_const(VERSION);
 
-pub async fn get_runtime_stats(context: &Arc<RuntimeContext>) -> ConnectorRuntimeStats {
-    let pid = std::process::id();
-
-    let mut system = System::new_all();
-    system.refresh_cpu_all();
-    system.refresh_memory();
-    system.refresh_processes(
-        sysinfo::ProcessesToUpdate::Some(&[sysinfo::Pid::from_u32(pid)]),
-        true,
-    );
-
-    let total_cpu_usage = system.global_cpu_usage();
-    let total_memory = system.total_memory();
-    let available_memory = system.available_memory();
+static SYSINFO: OnceLock<Mutex<System>> = OnceLock::new();
 
-    let (cpu_usage, memory_usage) = system
-        .process(sysinfo::Pid::from_u32(pid))
-        .map(|p| (p.cpu_usage(), p.memory()))
-        .unwrap_or((0.0, 0));
+pub async fn get_runtime_stats(context: &Arc<RuntimeContext>) -> ConnectorRuntimeStats {
+    let system = probe_system();
 
     let sources = context.sources.get_all().await;
     let sinks = context.sinks.get_all().await;
@@ -113,12 +99,12 @@ pub async fn get_runtime_stats(context: &Arc<RuntimeContext>) -> ConnectorRuntim
     ConnectorRuntimeStats {
         connectors_runtime_version: VERSION.to_owned(),
         connectors_runtime_version_semver: SEMANTIC_VERSION.get_numeric_version().ok(),
-        process_id: pid,
-        cpu_usage,
-        total_cpu_usage,
-        memory_usage,
-        total_memory,
-        available_memory,
+        process_id: system.process_id,
+        cpu_usage: system.cpu_usage,
+        total_cpu_usage: system.total_cpu_usage,
+        memory_usage: system.memory_usage,
+        total_memory: system.total_memory,
+        available_memory: system.available_memory,
         run_time,
         start_time: start,
         sources_total,
@@ -128,3 +114,11 @@ pub async fn get_runtime_stats(context: &Arc<RuntimeContext>) -> ConnectorRuntim
         connectors,
     }
 }
+
+fn probe_system() -> SystemProbe {
+    let mut system = SYSINFO
+        .get_or_init(|| Mutex::new(System::new()))
+        .lock()
+        .unwrap_or_else(PoisonError::into_inner);
+    SystemProbe::capture(&mut system)
+}

core/connectors/runtime/src/stats.rs

@@ -73,21 +73,39 @@ impl ConnectorError {
 /// Runtime statistics response from `/stats` endpoint.
 #[derive(Debug, Serialize, Deserialize)]
 pub struct ConnectorRuntimeStats {
+    /// The version of the connectors runtime.
     pub connectors_runtime_version: String,
+    /// The semantic version of the connectors runtime in the numeric format,
+    /// e.g. 1.2.3 -> 1002003 (major followed by zero-padded three-digit minor and patch).
     #[serde(skip_serializing_if = "Option::is_none")]
     pub connectors_runtime_version_semver: Option<u32>,
+    /// The unique identifier of the runtime process.
     pub process_id: u32,
+    /// The CPU usage of the runtime process.
     pub cpu_usage: f32,
+    /// The total CPU usage of the system, scoped to the cores this process may run on
+    /// when confined by an affinity/cpuset mask.
     pub total_cpu_usage: f32,
+    /// The memory usage of the runtime process in bytes.
     pub memory_usage: u64,
+    /// The total memory of the system in bytes, or the effective cgroup memory limit when
+    /// the runtime runs inside a memory-capped cgroup (container, systemd slice).
     pub total_memory: u64,
+    /// The available memory of the system in bytes, scoped to the cgroup limit when one applies.
     pub available_memory: u64,
+    /// The elapsed time since the runtime started, in microseconds.
     pub run_time: u64,
+    /// The time the runtime started, in microseconds since the UNIX epoch.
     pub start_time: u64,
+    /// The number of configured source connectors, including disabled and failed ones.
     pub sources_total: u32,
+    /// The number of currently running source connectors.
     pub sources_running: u32,
+    /// The number of configured sink connectors, including disabled and failed ones.
     pub sinks_total: u32,
+    /// The number of currently running sink connectors.
     pub sinks_running: u32,
+    /// Per-connector statistics for every configured source and sink.
     pub connectors: Vec<ConnectorStats>,
 }