This is an automated email from the ASF dual-hosted git repository. numinnex pushed a commit to branch server_ng_decouple in repository https://gitbox.apache.org/repos/asf/iggy.git
commit fbe9499763d304f851633fdd839d5d91d1e1b358 Author: Grzegorz Koszyk <[email protected]> AuthorDate: Mon Jul 6 11:40:46 2026 +0200 refactor(server-ng): drop dependency on the legacy server crate --- Cargo.lock | 28 ++- core/configs/src/server_config/server.rs | 34 ++- core/configs/src/server_config/sharding.rs | 237 +-------------------- core/configs/src/server_config/system.rs | 15 ++ core/server-ng/Cargo.toml | 1 - core/server-ng/src/bootstrap.rs | 75 ++++++- core/server-ng/src/lib.rs | 6 + core/server-ng/src/main.rs | 6 +- core/server-ng/src/offset_recovery.rs | 191 +++++++++++++++++ core/server-ng/src/partition_helpers.rs | 41 ++-- core/server-ng/src/responses.rs | 4 +- core/server-ng/src/server_error.rs | 5 +- core/server/Cargo.toml | 18 -- core/server/src/io/mod.rs | 3 +- core/server/src/lib.rs | 4 +- core/server/src/log/mod.rs | 19 -- core/server/src/main.rs | 8 +- core/server/src/server_error.rs | 10 +- core/server_common/Cargo.toml | 22 ++ .../src/io => server_common/src}/fs_utils.rs | 0 core/server_common/src/lib.rs | 3 + core/{server => server_common}/src/log/logger.rs | 61 +++--- core/server_common/src/log/mod.rs | 50 +++++ core/{server => server_common}/src/log/runtime.rs | 0 core/server_common/src/log/settings.rs | 74 +++++++ .../src/shard_allocator.rs | 2 +- .../src/sharding/cpu_allocation.rs} | 161 -------------- core/server_common/src/sharding/mod.rs | 2 + 28 files changed, 531 insertions(+), 549 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 106aa2445..0057b0c8e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -11684,7 +11684,6 @@ dependencies = [ "futures", "hash32 1.0.0", "human-repr", - "hwlocality", "iggy_binary_protocol", "iggy_common", "jsonwebtoken", @@ -11692,16 +11691,10 @@ dependencies = [ "mimalloc", "mime_guess", "nix", - "opentelemetry", - "opentelemetry-appender-tracing", - "opentelemetry-otlp", - "opentelemetry-semantic-conventions", - "opentelemetry_sdk", "papaya", "prometheus-client", "ringbuffer", "rmp-serde", - "rolling-file", "rust-embed", "rustls", "rustls-pemfile", @@ -11720,9 +11713,6 @@ dependencies = [ "toml 1.1.2+spec-1.1.0", "tower-http 0.7.0", "tracing", - "tracing-appender", - "tracing-opentelemetry", - "tracing-subscriber", "ulid", "uuid", "vergen-git2", @@ -11789,7 +11779,6 @@ dependencies = [ "secrecy", "send_wrapper", "serde", - "server", "server_common", "shard", "slab", @@ -11820,21 +11809,38 @@ dependencies = [ "compio", "compio-buf", "crossbeam", + "derive_more", "err_trail", + "fs2", + "futures", "human-repr", + "hwlocality", "iggy_binary_protocol", "iggy_common", "lending-iterator", "moka", "nix", + "opentelemetry", + "opentelemetry-appender-tracing", + "opentelemetry-otlp", + "opentelemetry-semantic-conventions", + "opentelemetry_sdk", "rand 0.10.1", "rcgen", + "rolling-file", "rustls", + "send_wrapper", + "serde", + "serde_json", "serial_test", "smallvec", + "tempfile", "thiserror 2.0.18", "tokio", "tracing", + "tracing-appender", + "tracing-opentelemetry", + "tracing-subscriber", "twox-hash", ] diff --git a/core/configs/src/server_config/server.rs b/core/configs/src/server_config/server.rs index 4b7aba2fa..2ef47c23e 100644 --- a/core/configs/src/server_config/server.rs +++ b/core/configs/src/server_config/server.rs @@ -24,7 +24,6 @@ use super::tcp::TcpConfig; use super::websocket::WebSocketConfig; use crate::ConfigurationError; use configs::{ConfigEnv, ConfigEnvMappings, ConfigProvider, FileConfigProvider, TypedEnvProvider}; -use derive_more::Display; use err_trail::ErrContext; use figment::providers::{Format, Toml}; use figment::value::Dict; @@ -34,10 +33,12 @@ use serde::{Deserialize, Serialize}; use serde_with::DisplayFromStr; use serde_with::serde_as; use server_common::MemoryPoolConfigOther; +use server_common::log::{TelemetryEndpointSettings, TelemetrySettings}; use std::env; -use std::str::FromStr; use std::sync::Arc; +pub use server_common::log::TelemetryTransport; + const DEFAULT_CONFIG_PATH: &str = "core/server/config.toml"; #[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)] @@ -158,22 +159,19 @@ pub struct TelemetryTracesConfig { pub endpoint: String, } -#[derive(Debug, Serialize, Deserialize, PartialEq, Display, Copy, Clone)] -#[serde(rename_all = "lowercase")] -pub enum TelemetryTransport { - #[display("grpc")] - GRPC, - #[display("http")] - HTTP, -} - -impl FromStr for TelemetryTransport { - type Err = String; - fn from_str(s: &str) -> Result<Self, Self::Err> { - match s { - "grpc" => Ok(TelemetryTransport::GRPC), - "http" => Ok(TelemetryTransport::HTTP), - _ => Err(format!("Invalid telemetry transport: {s}")), +impl From<&TelemetryConfig> for TelemetrySettings { + fn from(config: &TelemetryConfig) -> Self { + Self { + enabled: config.enabled, + service_name: config.service_name.clone(), + logs: TelemetryEndpointSettings { + transport: config.logs.transport, + endpoint: config.logs.endpoint.clone(), + }, + traces: TelemetryEndpointSettings { + transport: config.traces.transport, + endpoint: config.traces.endpoint.clone(), + }, } } } diff --git a/core/configs/src/server_config/sharding.rs b/core/configs/src/server_config/sharding.rs index ec332462b..67f0489e6 100644 --- a/core/configs/src/server_config/sharding.rs +++ b/core/configs/src/server_config/sharding.rs @@ -16,13 +16,14 @@ // under the License. use iggy_common::IggyDuration; -use serde::{Deserialize, Deserializer, Serialize, Serializer}; +use serde::{Deserialize, Serialize}; use serde_with::{DisplayFromStr, serde_as}; -use std::str::FromStr; use std::time::Duration; use configs::ConfigEnv; +pub use server_common::sharding::{CpuAllocation, NumaConfig}; + /// Default capacity of the per-shard inter-shard inbox channel. Sized /// comfortably above the consensus working set, which is roughly /// `PIPELINE_PREPARE_QUEUE_MAX (= 32) * replica_count * directions` @@ -178,235 +179,3 @@ impl Default for ShardingConfig { } } } - -#[derive(Debug, Clone, PartialEq, Default)] -pub enum CpuAllocation { - #[default] - All, - Count(usize), - Range(usize, usize), - NumaAware(NumaConfig), -} - -/// NUMA specific configuration -#[derive(Debug, Clone, PartialEq, Default)] -pub struct NumaConfig { - /// Which NUMA nodes to use (empty = auto-detect all) - pub nodes: Vec<usize>, - /// Cores per node to use (0 = use all available) - pub cores_per_node: usize, - /// skip hyperthread sibling - pub avoid_hyperthread: bool, -} - -impl CpuAllocation { - fn parse_numa(s: &str) -> Result<CpuAllocation, String> { - let params = s - .strip_prefix("numa:") - .ok_or_else(|| "Numa config must start with 'numa:'".to_string())?; - - if params == "auto" { - return Ok(CpuAllocation::NumaAware(NumaConfig { - nodes: vec![], - cores_per_node: 0, - avoid_hyperthread: true, - })); - } - - let mut nodes = Vec::new(); - let mut cores_per_node = 0; - let mut avoid_hyperthread = true; - - for param in params.split(';') { - let kv: Vec<&str> = param.split('=').collect(); - if kv.len() != 2 { - return Err(format!( - "Invalid NUMA parameter: '{param}', only available: 'auto'" - )); - } - - match kv[0] { - "nodes" => { - nodes = kv[1] - .split(',') - .map(|n| { - n.parse::<usize>() - .map_err(|_| format!("Invalid node number: {n}")) - }) - .collect::<Result<Vec<_>, _>>()?; - } - "cores" => { - cores_per_node = kv[1] - .parse::<usize>() - .map_err(|_| format!("Invalid cores value: {}", kv[1]))?; - } - "no_ht" => { - avoid_hyperthread = kv[1] - .parse::<bool>() - .map_err(|_| format!("Invalid no ht value: {}", kv[1]))?; - } - _ => { - return Err(format!( - "Unknown NUMA parameter: {}, example: numa:nodes=0;cores=4;no_ht=true", - kv[0] - )); - } - } - } - - Ok(CpuAllocation::NumaAware(NumaConfig { - nodes, - cores_per_node, - avoid_hyperthread, - })) - } -} - -impl FromStr for CpuAllocation { - type Err = String; - - fn from_str(s: &str) -> Result<Self, Self::Err> { - match s { - "all" => Ok(CpuAllocation::All), - s if s.starts_with("numa:") => Self::parse_numa(s), - s if s.contains("..") => { - let parts: Vec<&str> = s.split("..").collect(); - if parts.len() != 2 { - return Err(format!("Invalid range format: {s}. Expected 'start..end'")); - } - let start = parts[0] - .parse::<usize>() - .map_err(|_| format!("Invalid start value: {}", parts[0]))?; - let end = parts[1] - .parse::<usize>() - .map_err(|_| format!("Invalid end value: {}", parts[1]))?; - Ok(CpuAllocation::Range(start, end)) - } - s => { - let count = s - .parse::<usize>() - .map_err(|_| format!("Invalid shard count: {s}"))?; - Ok(CpuAllocation::Count(count)) - } - } - } -} - -impl Serialize for CpuAllocation { - fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error> - where - S: Serializer, - { - match self { - CpuAllocation::All => serializer.serialize_str("all"), - CpuAllocation::Count(n) => serializer.serialize_u64(*n as u64), - CpuAllocation::Range(start, end) => { - serializer.serialize_str(&format!("{start}..{end}")) - } - CpuAllocation::NumaAware(numa) => { - if numa.nodes.is_empty() && numa.cores_per_node == 0 { - serializer.serialize_str("numa:auto") - } else { - let nodes_str = numa - .nodes - .iter() - .map(|n| n.to_string()) - .collect::<Vec<_>>() - .join(","); - - let full_str = format!( - "numa:nodes={};cores={};no_ht={}", - nodes_str, numa.cores_per_node, numa.avoid_hyperthread - ); - - serializer.serialize_str(&full_str) - } - } - } - } -} - -impl<'de> Deserialize<'de> for CpuAllocation { - fn deserialize<D>(deserializer: D) -> Result<Self, D::Error> - where - D: Deserializer<'de>, - { - #[derive(Deserialize)] - #[serde(untagged)] - enum CpuAllocationHelper { - String(String), - Number(usize), - } - - match CpuAllocationHelper::deserialize(deserializer)? { - CpuAllocationHelper::String(s) => { - CpuAllocation::from_str(&s).map_err(serde::de::Error::custom) - } - CpuAllocationHelper::Number(n) => Ok(CpuAllocation::Count(n)), - } - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn test_parse_all() { - assert_eq!(CpuAllocation::from_str("all").unwrap(), CpuAllocation::All); - } - - #[test] - fn test_parse_count() { - assert_eq!( - CpuAllocation::from_str("4").unwrap(), - CpuAllocation::Count(4) - ); - } - - #[test] - fn test_parse_range() { - assert_eq!( - CpuAllocation::from_str("2..8").unwrap(), - CpuAllocation::Range(2, 8) - ); - } - - #[test] - fn test_parse_numa_auto() { - let result = CpuAllocation::from_str("numa:auto").unwrap(); - match result { - CpuAllocation::NumaAware(numa) => { - assert!(numa.nodes.is_empty()); - assert_eq!(numa.cores_per_node, 0); - assert!(numa.avoid_hyperthread); - } - _ => panic!("Expected NumaAware"), - } - } - - #[test] - fn test_parse_numa_explicit() { - let result = CpuAllocation::from_str("numa:nodes=0,1;cores=4;no_ht=true").unwrap(); - match result { - CpuAllocation::NumaAware(numa) => { - assert_eq!(numa.nodes, vec![0, 1]); - assert_eq!(numa.cores_per_node, 4); - assert!(numa.avoid_hyperthread); - } - _ => panic!("Expected NumaAware"), - } - } - - #[test] - fn test_numa_explicit_serde_roundtrip() { - let original = CpuAllocation::NumaAware(NumaConfig { - nodes: vec![0, 1], - cores_per_node: 4, - avoid_hyperthread: true, - }); - let serialized = serde_json::to_string(&original).unwrap(); - let deserialized: CpuAllocation = serde_json::from_str(&serialized).unwrap(); - assert_eq!(original, deserialized); - } -} diff --git a/core/configs/src/server_config/system.rs b/core/configs/src/server_config/system.rs index 59c9b397f..b1d124c20 100644 --- a/core/configs/src/server_config/system.rs +++ b/core/configs/src/server_config/system.rs @@ -28,6 +28,7 @@ use serde::{Deserialize, Serialize}; use serde_with::DisplayFromStr; use serde_with::serde_as; use server_common::bootstrap::SystemPaths; +use server_common::log::LoggingSettings; pub const INDEX_EXTENSION: &str = "index"; pub const LOG_EXTENSION: &str = "log"; @@ -100,6 +101,20 @@ pub struct LoggingConfig { pub sysinfo_print_interval: IggyDuration, } +impl From<&LoggingConfig> for LoggingSettings { + fn from(config: &LoggingConfig) -> Self { + Self { + path: config.path.clone(), + level: config.level.clone(), + file_enabled: config.file_enabled, + max_file_size: config.max_file_size, + max_total_size: config.max_total_size, + rotation_check_interval: config.rotation_check_interval, + retention: config.retention, + } + } +} + #[derive(Debug, Deserialize, Serialize, ConfigEnv)] pub struct EncryptionConfig { pub enabled: bool, diff --git a/core/server-ng/Cargo.toml b/core/server-ng/Cargo.toml index bf42650ed..22e8400cb 100644 --- a/core/server-ng/Cargo.toml +++ b/core/server-ng/Cargo.toml @@ -146,7 +146,6 @@ rustls-pemfile = { workspace = true } secrecy = { workspace = true } send_wrapper = { workspace = true } serde = { workspace = true } -server = { workspace = true } server_common = { workspace = true } shard = { workspace = true } slab = { workspace = true } diff --git a/core/server-ng/src/bootstrap.rs b/core/server-ng/src/bootstrap.rs index 46bd45c29..02a567232 100644 --- a/core/server-ng/src/bootstrap.rs +++ b/core/server-ng/src/bootstrap.rs @@ -37,6 +37,10 @@ use consensus::{LocalPipeline, MetadataHandle, PartitionsHandle, Sequencer, VsrC // non-blocking variants for cancel-safe shutdown polling. use crossfire::{AsyncRxTrait, AsyncTxTrait}; use iggy_binary_protocol::Operation; +use iggy_common::defaults::{ + DEFAULT_ROOT_USERNAME, MAX_PASSWORD_LENGTH, MAX_USERNAME_LENGTH, MIN_PASSWORD_LENGTH, + MIN_USERNAME_LENGTH, +}; use iggy_common::{IggyByteSize, PartitionStats, variadic}; use journal::Journal; use journal::prepare_journal::PrepareJournal; @@ -70,13 +74,11 @@ use partitions::{ }; use rustls::pki_types::ServerName; use server_common::bootstrap::create_directories; +use server_common::crypto; use server_common::executor::create_shard_executor; +use server_common::log::{Logging, LoggingSettings, TelemetrySettings}; +use server_common::shard_allocator::{ShardAllocator, ShardInfo}; use server_common::sharding::{IggyNamespace, PartitionLocation, ShardId}; -// TODO: decouple bootstrap/storage helpers and logging from the `server` crate. -use server::log::logger::Logging; -use server::shard_allocator::{ShardAllocator, ShardInfo}; -use server::streaming::users::user::User as LegacyUser; -use server::{IGGY_ROOT_PASSWORD_ENV, IGGY_ROOT_USERNAME_ENV}; use shard::builder::IggyShardBuilder; use shard::metrics::{ShardMetrics, frame_drop_reason, frame_drop_variant}; use shard::shards_table::{PapayaShardsTable, ShardsTable, calculate_shard_assignment}; @@ -98,6 +100,9 @@ use tracing::{error, info, warn}; const SHARD_REPLICA_ID: u8 = 0; +pub const IGGY_ROOT_USERNAME_ENV: &str = "IGGY_ROOT_USERNAME"; +pub const IGGY_ROOT_PASSWORD_ENV: &str = "IGGY_ROOT_PASSWORD"; + type ServerNgMuxStateMachine = MuxStateMachine<variadic!(Users, Streams)>; /// Cross-thread bundle carrying one `ReadHandleFactory` per metadata @@ -435,8 +440,8 @@ pub async fn load_config(logging: &mut Logging) -> Result<ServerNgConfig, Server logging .late_init( config.system.get_system_path(), - &config.system.logging, - &config.telemetry, + &LoggingSettings::from(&config.system.logging), + &TelemetrySettings::from(&config.telemetry), ) .map_err(ServerNgError::Logging)?; @@ -2030,10 +2035,58 @@ fn ensure_default_root_user(mux_stm: &ServerNgMuxStateMachine) { return; } - let LegacyUser { - username, password, .. - } = server::bootstrap::create_root_user(); - mux_stm.users().ensure_root_user(&username, &password); + let (username, password_hash) = create_root_credentials(); + mux_stm.users().ensure_root_user(&username, &password_hash); +} + +/// Resolve the root user credentials from `IGGY_ROOT_USERNAME` / +/// `IGGY_ROOT_PASSWORD`, falling back to the default username with a +/// generated password (printed to stdout, mirroring the legacy server). +/// +/// Returns `(username, password_hash)`; the plaintext password never +/// leaves this function. +fn create_root_credentials() -> (String, String) { + let mut username = env::var(IGGY_ROOT_USERNAME_ENV); + let mut password = env::var(IGGY_ROOT_PASSWORD_ENV); + assert_eq!( + username.is_ok(), + password.is_ok(), + "When providing the custom root user credentials, both username and password must be set." + ); + if username.is_ok() && password.is_ok() { + info!("Using the custom root user credentials."); + } else { + info!("Using the default root user credentials..."); + username = Ok(DEFAULT_ROOT_USERNAME.to_string()); + let generated_password = crypto::generate_secret(20..40); + println!("Generated root user password: {generated_password}"); + password = Ok(generated_password); + } + + let username = username.expect("Root username is not set."); + let password = password.expect("Root password is not set."); + assert!( + !username.is_empty() && !password.is_empty(), + "Root user credentials cannot be empty." + ); + assert!( + username.len() >= MIN_USERNAME_LENGTH, + "Root username is too short." + ); + assert!( + username.len() <= MAX_USERNAME_LENGTH, + "Root username is too long." + ); + assert!( + password.len() >= MIN_PASSWORD_LENGTH, + "Root password is too short." + ); + assert!( + password.len() <= MAX_PASSWORD_LENGTH, + "Root password is too long." + ); + + (username, crypto::hash_password(&password)) } fn validate_cluster_root_bootstrap( diff --git a/core/server-ng/src/lib.rs b/core/server-ng/src/lib.rs index 22ead12fa..cf3cd0cbe 100644 --- a/core/server-ng/src/lib.rs +++ b/core/server-ng/src/lib.rs @@ -17,12 +17,18 @@ #![allow(clippy::future_not_send)] +use iggy_common::SemanticVersion; + +pub const VERSION: &str = env!("CARGO_PKG_VERSION"); +pub const SEMANTIC_VERSION: SemanticVersion = SemanticVersion::parse_const(VERSION); + pub mod auth; pub mod bootstrap; pub mod config_writer; pub mod consumer_group; pub mod dispatch; pub mod login_register; +pub(crate) mod offset_recovery; pub mod partition_helpers; pub mod partition_reconciler; pub mod pat; diff --git a/core/server-ng/src/main.rs b/core/server-ng/src/main.rs index 6459c8fdd..4aabb4df4 100644 --- a/core/server-ng/src/main.rs +++ b/core/server-ng/src/main.rs @@ -54,7 +54,7 @@ fn main() -> Result<(), ServerNgError> { ( configs::server_ng::ServerNgConfig, Option<u8>, - server::log::logger::Logging, + server_common::log::Logging, ), ServerNgError, > = bootstrap_runtime.block_on(async { @@ -65,9 +65,9 @@ fn main() -> Result<(), ServerNgError> { let _ = dotenvy::dotenv(); } - // TODO: decouple logging from the `server` crate. - let mut logging = server::log::logger::Logging::new(); + let mut logging = server_common::log::Logging::new(server_ng::VERSION); logging.early_init(); + server_common::print_build_info!(server_ng::VERSION); let config = load_config(&mut logging).await?; server_common::MemoryPool::init_pool(&config.system.memory_pool.into_other()); diff --git a/core/server-ng/src/offset_recovery.rs b/core/server-ng/src/offset_recovery.rs new file mode 100644 index 000000000..2e7029193 --- /dev/null +++ b/core/server-ng/src/offset_recovery.rs @@ -0,0 +1,191 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! server-ng-owned consumer offset recovery. +//! +//! Forked from `server::streaming::partitions::storage` (the legacy +//! `load_consumer_offsets` / `load_consumer_group_offsets`) so server-ng +//! owns the loaders for the offset files its own persistence path writes, +//! without depending on the legacy `server` crate. The on-disk format is +//! shared with the legacy server today: one file per consumer (numeric +//! file name = consumer id) holding a single little-endian `u64` offset. + +use iggy_common::{ConsumerGroupId, ConsumerKind, ConsumerOffset, IggyError}; +use std::io::Read; +use std::sync::atomic::AtomicU64; +use tracing::{error, trace, warn}; + +const COMPONENT: &str = "STREAMING_PARTITIONS"; + +pub fn load_consumer_offsets(path: &str) -> Result<Vec<ConsumerOffset>, IggyError> { + trace!("Loading consumer offsets from path: {path}..."); + let Ok(dir_entries) = std::fs::read_dir(path) else { + return Err(IggyError::CannotReadConsumerOffsets(path.to_owned())); + }; + + let mut consumer_offsets = Vec::new(); + for dir_entry in dir_entries { + let dir_entry = match dir_entry { + Ok(entry) => entry, + Err(e) => { + warn!( + "Failed to read directory entry in consumer offsets path: {path}, \ + error: {e}, skipping." + ); + continue; + } + }; + + let metadata = match dir_entry.metadata() { + Ok(m) => m, + Err(e) => { + warn!( + "Failed to read metadata for entry in consumer offsets path: {path}, \ + error: {e}, skipping." + ); + continue; + } + }; + + if metadata.is_dir() { + continue; + } + + let name = dir_entry.file_name().to_string_lossy().to_string(); + let Ok(consumer_id) = name.parse::<u32>() else { + warn!( + "Unexpected non-numeric consumer offset file: '{}', skipping.", + name + ); + continue; + }; + + let path = dir_entry.path(); + let Some(path) = path.to_str().map(str::to_owned) else { + error!("Invalid consumer ID path for file with name: '{}'.", name); + continue; + }; + + let Some(offset) = read_offset_file(&path, "consumer offset") else { + continue; + }; + + consumer_offsets.push(ConsumerOffset { + kind: ConsumerKind::Consumer, + consumer_id, + offset, + path, + }); + } + + consumer_offsets.sort_by_key(|consumer_offset| consumer_offset.consumer_id); + Ok(consumer_offsets) +} + +pub fn load_consumer_group_offsets( + path: &str, +) -> Result<Vec<(ConsumerGroupId, ConsumerOffset)>, IggyError> { + trace!("Loading consumer group offsets from path: {path}..."); + let Ok(dir_entries) = std::fs::read_dir(path) else { + return Err(IggyError::CannotReadConsumerOffsets(path.to_owned())); + }; + + let mut consumer_group_offsets = Vec::new(); + for dir_entry in dir_entries { + let dir_entry = match dir_entry { + Ok(entry) => entry, + Err(e) => { + warn!( + "Failed to read directory entry in consumer group offsets path: {path}, \ + error: {e}, skipping." + ); + continue; + } + }; + + let metadata = match dir_entry.metadata() { + Ok(m) => m, + Err(e) => { + warn!( + "Failed to read metadata for entry in consumer group offsets path: {path}, \ + error: {e}, skipping." + ); + continue; + } + }; + + if metadata.is_dir() { + continue; + } + + let name = dir_entry.file_name().to_string_lossy().to_string(); + let Ok(raw_consumer_group_id) = name.parse::<u32>() else { + warn!( + "Unexpected non-numeric consumer group offset file: '{}', skipping.", + name + ); + continue; + }; + let consumer_group_id = ConsumerGroupId(raw_consumer_group_id as usize); + + let path = dir_entry.path(); + let Some(path) = path.to_str().map(str::to_owned) else { + error!( + "Invalid consumer group offset path for file with name: '{}'.", + name + ); + continue; + }; + + let Some(offset) = read_offset_file(&path, "consumer group offset") else { + continue; + }; + + let consumer_offset = ConsumerOffset { + kind: ConsumerKind::ConsumerGroup, + consumer_id: raw_consumer_group_id, + offset, + path, + }; + + consumer_group_offsets.push((consumer_group_id, consumer_offset)); + } + + Ok(consumer_group_offsets) +} + +fn read_offset_file(path: &str, offset_kind: &'static str) -> Option<AtomicU64> { + let mut file = match std::fs::File::open(path) { + Ok(file) => file, + Err(e) => { + warn!( + "{COMPONENT} (error: {e}) - failed to open offset file, \ + path: {path}, skipping." + ); + return None; + } + }; + let mut offset = [0; 8]; + if let Err(e) = file.read_exact(&mut offset) { + warn!( + "{COMPONENT} (error: {e}) - failed to read {offset_kind} from file \ + (truncated or corrupt?), path: {path}, skipping." + ); + return None; + } + Some(AtomicU64::new(u64::from_le_bytes(offset))) +} diff --git a/core/server-ng/src/partition_helpers.rs b/core/server-ng/src/partition_helpers.rs index 6653819ad..a6aa71fb1 100644 --- a/core/server-ng/src/partition_helpers.rs +++ b/core/server-ng/src/partition_helpers.rs @@ -24,6 +24,7 @@ //! local partition yet. The two paths share namespace-bounds validation, //! consumer-offset configuration, and initial-segment provisioning. +use crate::offset_recovery::{load_consumer_group_offsets, load_consumer_offsets}; use crate::server_error::ServerNgError; use compio::fs::create_dir_all; use configs::server_ng::ServerNgConfig; @@ -33,9 +34,8 @@ use iggy_common::{ }; use message_bus::IggyMessageBus; use partitions::{IggyIndexWriter, IggyPartition, MessagesWriter, Segment}; -use server::io::fs_utils::remove_dir_all; -use server::streaming::partitions::storage::{load_consumer_group_offsets, load_consumer_offsets}; -use server::streaming::segments::storage::create_segment_storage; +use server_common::SegmentStorage; +use server_common::fs_utils::remove_dir_all; use server_common::sharding::IggyNamespace; use std::path::Path; use std::rc::Rc; @@ -333,26 +333,33 @@ pub async fn ensure_initial_segment( return Ok(()); } - // TODO: decouple segment storage creation from the `server` crate. - let storage = - create_segment_storage(&config.system, stream_id, topic_id, partition_id, 0, 0, 0) - .await - .map_err(|source| { - error!( - stream_id, - topic_id, - partition_id, - error = %source, - "failed to create initial segment storage" - ); - source - })?; let messages_path = config .system .get_messages_file_path(stream_id, topic_id, partition_id, 0); let index_path = config .system .get_index_path(stream_id, topic_id, partition_id, 0); + let enforce_fsync = config.system.partition.enforce_fsync; + let storage = SegmentStorage::new( + &messages_path, + &index_path, + 0, + 0, + enforce_fsync, + enforce_fsync, + false, + ) + .await + .map_err(|source| { + error!( + stream_id, + topic_id, + partition_id, + error = %source, + "failed to create initial segment storage" + ); + source + })?; partition.log.add_persisted_segment( Segment::new(0, config.system.segment.size), storage, diff --git a/core/server-ng/src/responses.rs b/core/server-ng/src/responses.rs index f3a0078d5..816d8340c 100644 --- a/core/server-ng/src/responses.rs +++ b/core/server-ng/src/responses.rs @@ -584,8 +584,8 @@ fn build_stats_response(shard: &Rc<ServerNgShard>) -> Result<StatsResponse, Iggy os_name: "unknown_os_name".to_owned(), os_version: "unknown_os_version".to_owned(), kernel_version: "unknown_kernel_version".to_owned(), - iggy_server_version: server::VERSION.to_owned(), - iggy_server_semver: server::SEMANTIC_VERSION.get_numeric_version().ok(), + iggy_server_version: crate::VERSION.to_owned(), + iggy_server_semver: crate::SEMANTIC_VERSION.get_numeric_version().ok(), cache_metrics: Vec::new(), threads_count: 0, free_disk_space: 0, diff --git a/core/server-ng/src/server_error.rs b/core/server-ng/src/server_error.rs index a3070dedd..e15ec1324 100644 --- a/core/server-ng/src/server_error.rs +++ b/core/server-ng/src/server_error.rs @@ -16,9 +16,8 @@ // under the License. use metadata::impls::recovery::RecoveryError; -// TODO: decouple logging errors from the `server` crate. -use server::server_error::LogError; -use server::shard_allocator::ShardingError; +use server_common::log::LogError; +use server_common::shard_allocator::ShardingError; use shard::ShardCtorError; use thiserror::Error; diff --git a/core/server/Cargo.toml b/core/server/Cargo.toml index c48abee12..5af835fec 100644 --- a/core/server/Cargo.toml +++ b/core/server/Cargo.toml @@ -22,9 +22,6 @@ edition = "2024" license = "Apache-2.0" publish = false -[package.metadata.cargo-udeps.ignore] -normal = ["tracing-appender"] - [package.metadata.cargo-machete] ignored = ["vergen-git2"] @@ -71,16 +68,10 @@ left-right = { workspace = true } mimalloc = { workspace = true, optional = true } mime_guess = { workspace = true, optional = true } nix = { workspace = true } -opentelemetry = { workspace = true } -opentelemetry-appender-tracing = { workspace = true } -opentelemetry-otlp = { workspace = true } -opentelemetry-semantic-conventions = { workspace = true } -opentelemetry_sdk = { workspace = true } papaya = { workspace = true } prometheus-client = { workspace = true } ringbuffer = { workspace = true } rmp-serde = { workspace = true } -rolling-file = { workspace = true } rust-embed = { workspace = true, optional = true } rustls = { workspace = true } rustls-pemfile = { workspace = true } @@ -99,17 +90,8 @@ thiserror = { workspace = true } toml = { workspace = true } tower-http = { workspace = true } tracing = { workspace = true } -tracing-appender = { workspace = true } -tracing-opentelemetry = { workspace = true } -tracing-subscriber = { workspace = true } ulid = { workspace = true } uuid = { workspace = true } -[target.'cfg(not(target_env = "musl"))'.dependencies] -hwlocality = { workspace = true } - -[target.'cfg(target_env = "musl")'.dependencies] -hwlocality = { workspace = true, features = ["vendored"] } - [build-dependencies] vergen-git2 = { workspace = true } diff --git a/core/server/src/io/mod.rs b/core/server/src/io/mod.rs index 49f06c069..4efe7ebd1 100644 --- a/core/server/src/io/mod.rs +++ b/core/server/src/io/mod.rs @@ -15,5 +15,6 @@ // specific language governing permissions and limitations // under the License. -pub mod fs_utils; +pub use server_common::fs_utils; + pub mod storage; diff --git a/core/server/src/lib.rs b/core/server/src/lib.rs index 6807fb6ba..fed758322 100644 --- a/core/server/src/lib.rs +++ b/core/server/src/lib.rs @@ -35,18 +35,18 @@ pub mod configs; pub mod diagnostics; pub mod http; pub mod io; -pub mod log; pub mod metadata; pub mod quic; pub mod sender; pub mod server_error; pub mod shard; -pub mod shard_allocator; pub mod state; pub mod streaming; pub mod tcp; pub mod websocket; +pub use server_common::{log, shard_allocator}; + pub const VERSION: &str = env!("CARGO_PKG_VERSION"); pub const SEMANTIC_VERSION: SemanticVersion = SemanticVersion::parse_const(VERSION); pub const IGGY_ROOT_USERNAME_ENV: &str = "IGGY_ROOT_USERNAME"; diff --git a/core/server/src/log/mod.rs b/core/server/src/log/mod.rs deleted file mode 100644 index 5c10dbcac..000000000 --- a/core/server/src/log/mod.rs +++ /dev/null @@ -1,19 +0,0 @@ -// Licensed to the Apache Software Foundation (ASF) under one -// or more contributor license agreements. See the NOTICE file -// distributed with this work for additional information -// regarding copyright ownership. The ASF licenses this file -// to you under the Apache License, Version 2.0 (the -// "License"); you may not use this file except in compliance -// with the License. You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, -// software distributed under the License is distributed on an -// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY -// KIND, either express or implied. See the License for the -// specific language governing permissions and limitations -// under the License. - -pub mod logger; -pub mod runtime; diff --git a/core/server/src/main.rs b/core/server/src/main.rs index d9baaac85..cc4aaed91 100644 --- a/core/server/src/main.rs +++ b/core/server/src/main.rs @@ -48,6 +48,7 @@ use server::streaming::diagnostics::metrics::Metrics; use server::streaming::storage::SystemStorage; use server::streaming::utils::ptr::EternalPtr; use server_common::MemoryPool; +use server_common::log::{LoggingSettings, TelemetrySettings}; use server_common::sharding::{IggyNamespace, PartitionLocation, ShardId}; use std::panic::AssertUnwindSafe; use std::rc::Rc; @@ -145,8 +146,9 @@ fn main() -> Result<(), ServerError> { // FIRST DISCRETE LOADING STEP. // Initialize early logging before config parsing so we can log during bootstrap. - let mut logging = Logging::new(); + let mut logging = Logging::new(server::VERSION); logging.early_init(); + server_common::print_build_info!(server::VERSION); // SECOND DISCRETE LOADING STEP. // Load config and create directories. @@ -176,8 +178,8 @@ fn main() -> Result<(), ServerError> { // From this point on, logs are persisted to file and telemetry is active. logging.late_init( config.system.get_system_path(), - &config.system.logging, - &config.telemetry, + &LoggingSettings::from(&config.system.logging), + &TelemetrySettings::from(&config.telemetry), )?; if is_follower { diff --git a/core/server/src/server_error.rs b/core/server/src/server_error.rs index 6a63fac8f..fb8b57b33 100644 --- a/core/server/src/server_error.rs +++ b/core/server/src/server_error.rs @@ -66,14 +66,8 @@ error_set!( } || IoError || CommonError LogError := { - #[display("Logging filter reload failure")] - FilterReloadFailure, - - #[display("Logging stdout reload failure")] - StdoutReloadFailure, - - #[display("Logging file reload failure")] - FileReloadFailure, + #[display("{0}")] + Logging(server_common::log::LogError), } CompatError := { diff --git a/core/server_common/Cargo.toml b/core/server_common/Cargo.toml index 2960cd11f..1a2fe81cd 100644 --- a/core/server_common/Cargo.toml +++ b/core/server_common/Cargo.toml @@ -31,23 +31,45 @@ bytes = { workspace = true } compio = { workspace = true } compio-buf = { workspace = true } crossbeam = { workspace = true } +derive_more = { workspace = true } err_trail = { workspace = true } +fs2 = { workspace = true } +futures = { workspace = true } human-repr = { workspace = true } iggy_binary_protocol = { workspace = true } iggy_common = { workspace = true } lending-iterator = { workspace = true } moka = { workspace = true } +opentelemetry = { workspace = true } +opentelemetry-appender-tracing = { workspace = true } +opentelemetry-otlp = { workspace = true } +opentelemetry-semantic-conventions = { workspace = true } +opentelemetry_sdk = { workspace = true } rand = { workspace = true } rcgen = { workspace = true } +rolling-file = { workspace = true } rustls = { workspace = true } +send_wrapper = { workspace = true, features = ["futures"] } +serde = { workspace = true } smallvec = { workspace = true } thiserror = { workspace = true } tracing = { workspace = true } +tracing-appender = { workspace = true } +tracing-opentelemetry = { workspace = true } +tracing-subscriber = { workspace = true } twox-hash = { workspace = true } [target.'cfg(unix)'.dependencies] nix = { workspace = true } +[target.'cfg(not(target_env = "musl"))'.dependencies] +hwlocality = { workspace = true } + +[target.'cfg(target_env = "musl")'.dependencies] +hwlocality = { workspace = true, features = ["vendored"] } + [dev-dependencies] +serde_json = { workspace = true } serial_test = { workspace = true } +tempfile = { workspace = true } tokio = { workspace = true } diff --git a/core/server/src/io/fs_utils.rs b/core/server_common/src/fs_utils.rs similarity index 100% rename from core/server/src/io/fs_utils.rs rename to core/server_common/src/fs_utils.rs diff --git a/core/server_common/src/lib.rs b/core/server_common/src/lib.rs index 5420adf15..9b1ede7e9 100644 --- a/core/server_common/src/lib.rs +++ b/core/server_common/src/lib.rs @@ -23,6 +23,7 @@ pub mod crypto; mod deduplication; pub mod diagnostics; pub mod executor; +pub mod fs_utils; mod in_flight; mod indexes_mut; // TODO(hubcio): iobuf was relocated verbatim from `core/binary_protocol/src/consensus/iobuf.rs` @@ -39,11 +40,13 @@ mod indexes_mut; clippy::use_self )] pub mod iobuf; +pub mod log; mod memory_pool; mod messages_batch_mut; mod messages_batch_set; mod segment_storage; pub mod send_messages2; +pub mod shard_allocator; pub mod sharding; pub use bootstrap::create_directories; diff --git a/core/server/src/log/logger.rs b/core/server_common/src/log/logger.rs similarity index 95% rename from core/server/src/log/logger.rs rename to core/server_common/src/log/logger.rs index 4ad885c0f..6ce0f919a 100644 --- a/core/server/src/log/logger.rs +++ b/core/server_common/src/log/logger.rs @@ -15,11 +15,8 @@ // specific language governing permissions and limitations // under the License. -use crate::VERSION; -use crate::configs::server::{TelemetryConfig, TelemetryTransport}; -use crate::configs::system::LoggingConfig; -use crate::log::runtime::CompioRuntime; -use crate::server_error::LogError; +use super::runtime::CompioRuntime; +use super::settings::{LoggingSettings, TelemetrySettings, TelemetryTransport}; use iggy_common::{IggyByteSize, IggyDuration}; use opentelemetry::KeyValue; use opentelemetry::global; @@ -53,6 +50,18 @@ use tracing_subscriber::{ const IGGY_LOG_FILE_PREFIX: &str = "iggy-server.log"; const ONE_HUNDRED_THOUSAND: u64 = 100_000; +#[derive(Debug, thiserror::Error)] +pub enum LogError { + #[error("Logging filter reload failure")] + FilterReloadFailure, + + #[error("Logging stdout reload failure")] + StdoutReloadFailure, + + #[error("Logging file reload failure")] + FileReloadFailure, +} + // Writer that does nothing struct NullWriter; impl Write for NullWriter { @@ -121,6 +130,8 @@ type EnvFilterReloadHandle = Handle<EnvFilter, Registry>; type FilteredRegistry = Layered<reload::Layer<EnvFilter, Registry>, Registry>; pub struct Logging { + server_version: &'static str, + stdout_guard: Option<WorkerGuard>, stdout_reload_handle: Option<ReloadHandle<FilteredRegistry>>, @@ -139,8 +150,9 @@ pub struct Logging { } impl Logging { - pub fn new() -> Self { + pub fn new(server_version: &'static str) -> Self { Self { + server_version, stdout_guard: None, stdout_reload_handle: None, file_guard: None, @@ -210,21 +222,20 @@ impl Logging { .with(env_filter_layer) .with(layers) .init(); - Self::print_build_info(); } pub fn late_init( &mut self, base_directory: String, - config: &LoggingConfig, - telemetry_config: &TelemetryConfig, + config: &LoggingSettings, + telemetry_config: &TelemetrySettings, ) -> Result<(), LogError> { // Write to stdout and file at the same time. // Use the non_blocking appender to avoid blocking the threads. // Use the rolling appender to avoid having a huge log file. // Make sure logs are dumped to the file during graceful shutdown. - trace!("Logging config: {config}"); + trace!("Logging config: {config:?}"); // Reload EnvFilter with config level if RUST_LOG is not set. // Config level supports EnvFilter syntax (e.g., "warn,server=debug,iggy=trace"). @@ -350,13 +361,13 @@ impl Logging { Ok(()) } - fn init_telemetry(&mut self, telemetry_config: &TelemetryConfig) -> Result<(), LogError> { + fn init_telemetry(&mut self, telemetry_config: &TelemetrySettings) -> Result<(), LogError> { let service_name = telemetry_config.service_name.to_owned(); let resource = Resource::builder() .with_service_name(service_name.clone()) .with_attribute(KeyValue::new( opentelemetry_semantic_conventions::resource::SERVICE_VERSION, - VERSION, + self.server_version, )) .build(); @@ -450,22 +461,6 @@ impl Logging { Format::default().with_thread_names(true) } - fn print_build_info() { - if option_env!("IGGY_CI_BUILD") == Some("true") { - let hash = option_env!("VERGEN_GIT_SHA").unwrap_or("unknown"); - let built_at = option_env!("VERGEN_BUILD_TIMESTAMP").unwrap_or("unknown"); - let rust_version = option_env!("VERGEN_RUSTC_SEMVER").unwrap_or("unknown"); - let target = option_env!("VERGEN_CARGO_TARGET_TRIPLE").unwrap_or("unknown"); - info!( - "Version: {VERSION}, hash: {hash}, built at: {built_at} using rust version: {rust_version} for target: {target}" - ); - } else { - info!( - "It seems that you are a developer. Environment variable IGGY_CI_BUILD is not set to 'true', skipping build info print." - ) - } - } - fn calculate_max_files( max_total_size_bytes: IggyByteSize, max_file_size_bytes: IggyByteSize, @@ -485,7 +480,7 @@ impl Logging { fn install_log_rotation_handler( &self, - config: &LoggingConfig, + config: &LoggingSettings, logs_path: Option<&PathBuf>, ) -> Option<std::thread::JoinHandle<()>> { let logs_path = logs_path?; @@ -742,12 +737,6 @@ impl Logging { } } -impl Default for Logging { - fn default() -> Self { - Self::new() - } -} - impl Drop for Logging { fn drop(&mut self) { self.rotation_should_stop @@ -865,7 +854,7 @@ mod tests { #[test] fn test_logging_creation() { - let logging = Logging::new(); + let logging = Logging::new("0.0.0-test"); assert!(logging.stdout_guard.is_none()); assert!(logging.file_guard.is_none()); assert!(logging.env_filter_reload_handle.is_none()); diff --git a/core/server_common/src/log/mod.rs b/core/server_common/src/log/mod.rs new file mode 100644 index 000000000..c932681ae --- /dev/null +++ b/core/server_common/src/log/mod.rs @@ -0,0 +1,50 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +pub mod logger; +pub mod runtime; +pub mod settings; + +pub use logger::{LogError, Logging}; +pub use settings::{ + LoggingSettings, TelemetryEndpointSettings, TelemetrySettings, TelemetryTransport, +}; + +/// Log the CI build fingerprint (or a developer-build notice). +/// +/// A macro rather than a function so the `option_env!` lookups expand in +/// the calling binary crate, where its `build.rs` emitted the `VERGEN_*` +/// values; expanded here they would always read as unset. +#[macro_export] +macro_rules! print_build_info { + ($version:expr) => { + if option_env!("IGGY_CI_BUILD") == Some("true") { + let hash = option_env!("VERGEN_GIT_SHA").unwrap_or("unknown"); + let built_at = option_env!("VERGEN_BUILD_TIMESTAMP").unwrap_or("unknown"); + let rust_version = option_env!("VERGEN_RUSTC_SEMVER").unwrap_or("unknown"); + let target = option_env!("VERGEN_CARGO_TARGET_TRIPLE").unwrap_or("unknown"); + ::tracing::info!( + "Version: {version}, hash: {hash}, built at: {built_at} using rust version: {rust_version} for target: {target}", + version = $version, + ); + } else { + ::tracing::info!( + "It seems that you are a developer. Environment variable IGGY_CI_BUILD is not set to 'true', skipping build info print." + ) + } + }; +} diff --git a/core/server/src/log/runtime.rs b/core/server_common/src/log/runtime.rs similarity index 100% rename from core/server/src/log/runtime.rs rename to core/server_common/src/log/runtime.rs diff --git a/core/server_common/src/log/settings.rs b/core/server_common/src/log/settings.rs new file mode 100644 index 000000000..a540dcd9e --- /dev/null +++ b/core/server_common/src/log/settings.rs @@ -0,0 +1,74 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Plain input structs for [`crate::log::logger::Logging`]. +//! +//! The `configs` crate depends on `server_common`, so the logger cannot +//! take the `ConfigEnv`-derived config structs directly; instead it takes +//! these mirrors of the fields it consumes, and `configs` provides `From` +//! conversions from `LoggingConfig` / `TelemetryConfig`. + +use derive_more::Display; +use serde::{Deserialize, Serialize}; +use std::str::FromStr; + +use iggy_common::{IggyByteSize, IggyDuration}; + +#[derive(Debug, Clone)] +pub struct LoggingSettings { + pub path: String, + pub level: String, + pub file_enabled: bool, + pub max_file_size: IggyByteSize, + pub max_total_size: IggyByteSize, + pub rotation_check_interval: IggyDuration, + pub retention: IggyDuration, +} + +#[derive(Debug, Clone)] +pub struct TelemetrySettings { + pub enabled: bool, + pub service_name: String, + pub logs: TelemetryEndpointSettings, + pub traces: TelemetryEndpointSettings, +} + +#[derive(Debug, Clone)] +pub struct TelemetryEndpointSettings { + pub transport: TelemetryTransport, + pub endpoint: String, +} + +#[derive(Debug, Serialize, Deserialize, PartialEq, Display, Copy, Clone)] +#[serde(rename_all = "lowercase")] +pub enum TelemetryTransport { + #[display("grpc")] + GRPC, + #[display("http")] + HTTP, +} + +impl FromStr for TelemetryTransport { + type Err = String; + fn from_str(s: &str) -> Result<Self, Self::Err> { + match s { + "grpc" => Ok(TelemetryTransport::GRPC), + "http" => Ok(TelemetryTransport::HTTP), + _ => Err(format!("Invalid telemetry transport: {s}")), + } + } +} diff --git a/core/server/src/shard_allocator.rs b/core/server_common/src/shard_allocator.rs similarity index 99% rename from core/server/src/shard_allocator.rs rename to core/server_common/src/shard_allocator.rs index 26a1f0e38..4d329f9bc 100644 --- a/core/server/src/shard_allocator.rs +++ b/core/server_common/src/shard_allocator.rs @@ -15,7 +15,7 @@ // specific language governing permissions and limitations // under the License. -use configs::sharding::{CpuAllocation, NumaConfig}; +use crate::sharding::{CpuAllocation, NumaConfig}; use hwlocality::Topology; use hwlocality::bitmap::SpecializedBitmapRef; use hwlocality::cpu::cpuset::CpuSet; diff --git a/core/configs/src/server_config/sharding.rs b/core/server_common/src/sharding/cpu_allocation.rs similarity index 50% copy from core/configs/src/server_config/sharding.rs copy to core/server_common/src/sharding/cpu_allocation.rs index ec332462b..fef2c3f46 100644 --- a/core/configs/src/server_config/sharding.rs +++ b/core/server_common/src/sharding/cpu_allocation.rs @@ -15,169 +15,8 @@ // specific language governing permissions and limitations // under the License. -use iggy_common::IggyDuration; use serde::{Deserialize, Deserializer, Serialize, Serializer}; -use serde_with::{DisplayFromStr, serde_as}; use std::str::FromStr; -use std::time::Duration; - -use configs::ConfigEnv; - -/// Default capacity of the per-shard inter-shard inbox channel. Sized -/// comfortably above the consensus working set, which is roughly -/// `PIPELINE_PREPARE_QUEUE_MAX (= 32) * replica_count * directions` -/// frames in flight per shard, without allowing a runaway producer to -/// eat unbounded memory. Tunable via `[system.sharding] inbox_capacity` -/// in TOML. -/// -/// The capacity must also absorb the worst-case cross-shard client -/// Reply burst. Unlike consensus frames, client Replies have no VSR -/// retransmit path: a Reply lost on full inbox is gone and the client -/// times out. A reasonable lower bound is -/// `max_inflight_client_requests / num_shards` (assuming requests are -/// distributed evenly across owning shards) plus the consensus -/// headroom above. -/// -/// Consensus frames and client-reply forwards share this one channel, -/// so the two headrooms are not independent: a consensus burst or -/// retransmit storm can fill the inbox with consensus frames exactly -/// when a client Reply needs the space. A single `inbox_capacity` knob -/// cannot isolate the two frame classes - size it for the sum of both -/// worst cases occurring together. Watch the drop-site `tracing` logs -/// (and, once a per-shard exporter lands, the `frame_drops_total` -/// `{variant="forward_client_send"}` counter) to detect when the bound -/// is too low in production. -pub const DEFAULT_INBOX_CAPACITY: usize = 1024; - -/// Maximum permitted per-shard inbox depth. The channel is allocated -/// up-front per shard, so a runaway value here OOMs the process at boot. -/// `1 << 20` (~1M frames) is several orders of magnitude above any -/// realistic backpressure target and still fits comfortably in process -/// address space. -pub const INBOX_CAPACITY_MAX: usize = 1 << 20; - -/// Default bus shutdown drain timeout. Sized larger than typical TCP RTT -/// times in-flight write-batch so writers receive their full last -/// `write_vectored_all` budget before the connection registry kicks in. -pub const DEFAULT_SHUTDOWN_DRAIN_TIMEOUT: Duration = Duration::from_secs(10); - -/// Default watchdog poll cadence for the cross-thread shutdown flag. -/// 50ms keeps Ctrl-C latency operator-visible without measurable wakeup -/// overhead. -pub const DEFAULT_SHUTDOWN_POLL_INTERVAL: Duration = Duration::from_millis(50); - -/// Hard upper bound on `shutdown_drain_timeout`. A drain that never -/// completes wedges process exit; capping at 10 minutes guarantees the -/// watchdog eventually force-tears the bus even with a pathological -/// config typo. -pub const SHUTDOWN_DRAIN_TIMEOUT_MAX: Duration = Duration::from_secs(600); - -/// Hard upper bound on `shutdown_poll_interval`. A pollerinterval longer -/// than the drain timeout makes the flag effectively unobservable; cap -/// at 5s so Ctrl-C latency stays bounded regardless of config. -pub const SHUTDOWN_POLL_INTERVAL_MAX: Duration = Duration::from_secs(5); - -/// Default safety-tick cadence for the partition reconciliation loop. -/// The reconciler also wakes on every `LifecycleFrame::MetadataCommitTick` -/// broadcast by shard 0; this fallback covers dropped wake-ups (the wake -/// channel is intentionally capacity-1) and the initial post-bootstrap -/// convergence window before shard 0's first tick. One second is -/// invisible to operators yet keeps idle clusters from burning CPU -/// re-reading the same target snapshot. -pub const DEFAULT_RECONCILE_PERIODIC_INTERVAL: Duration = Duration::from_secs(1); - -/// Hard upper bound on `reconcile_periodic_interval`. A tick longer -/// than ~30s makes post-failure recovery latency operator-visible; the -/// cap reins in pathological typos without disturbing reasonable -/// production values. -pub const RECONCILE_PERIODIC_INTERVAL_MAX: Duration = Duration::from_secs(30); - -const fn default_inbox_capacity() -> usize { - DEFAULT_INBOX_CAPACITY -} - -fn default_shutdown_drain_timeout() -> IggyDuration { - IggyDuration::new(DEFAULT_SHUTDOWN_DRAIN_TIMEOUT) -} - -fn default_shutdown_poll_interval() -> IggyDuration { - IggyDuration::new(DEFAULT_SHUTDOWN_POLL_INTERVAL) -} - -fn default_reconcile_periodic_interval() -> IggyDuration { - IggyDuration::new(DEFAULT_RECONCILE_PERIODIC_INTERVAL) -} - -#[serde_as] -#[derive(Debug, Deserialize, Serialize, ConfigEnv)] -pub struct ShardingConfig { - #[serde(default)] - #[config_env(leaf)] - pub cpu_allocation: CpuAllocation, - /// Per-shard inter-shard inbox channel capacity. Bounded by design. - /// Drops on full inbox of consensus frames are recovered by VSR - /// retransmit. Drops of cross-shard client Reply frames are terminal: - /// the client never receives the reply (no in-protocol retransmit). - /// Both frame classes share this one channel, so a consensus burst - /// can starve client-reply forwards: size against the worst-case sum - /// of consensus working set + peak client-reply fan-out per shard - /// occurring together; see `DEFAULT_INBOX_CAPACITY` for the - /// rationale. Used by `core/server-ng`; the legacy server uses its - /// own hard-coded inbox sizing. - /// - // TODO(hubcio): split into two priority lanes - one bounded queue for - // consensus frames (drops recovered by VSR retransmit) and one for - // client `Reply` frames (drops terminal, must be sized for worst-case - // fan-out). Current single-channel design is the minimum-viable - // wiring so `frame_drops_total{variant,reason}` surfaces under load - // and yields real numbers to size the split against. - #[serde(default = "default_inbox_capacity")] - pub inbox_capacity: usize, - /// Wall-clock budget for a single shard's bus drain on shutdown. - /// Drives `IggyMessageBus::shutdown(..)` from the per-shard watchdog - /// and the parallel-join survivor path. Sized larger than typical - /// TCP RTT times in-flight write-batch so writers receive their full - /// last `write_vectored_all` budget before the connection registry - /// force-tears the bus. Slow-fsync hosts may need to extend this past - /// the default; the cap is `SHUTDOWN_DRAIN_TIMEOUT_MAX` so a config - /// typo cannot wedge process exit. - #[serde(default = "default_shutdown_drain_timeout")] - #[serde_as(as = "DisplayFromStr")] - #[config_env(leaf)] - pub shutdown_drain_timeout: IggyDuration, - /// Poll cadence for the cross-thread shutdown flag and for the - /// `await_metadata_bundle` / `broadcast_metadata_bundle` poll loops. - /// Trades off Ctrl-C latency against idle wakeup cost; the default - /// keeps shutdown observably prompt without measurable scheduler - /// overhead. Capped at `SHUTDOWN_POLL_INTERVAL_MAX` so the flag - /// remains effectively observable regardless of config. - #[serde(default = "default_shutdown_poll_interval")] - #[serde_as(as = "DisplayFromStr")] - #[config_env(leaf)] - pub shutdown_poll_interval: IggyDuration, - /// Safety-tick cadence for the partition reconciliation loop; the - /// reconciler also wakes immediately on every - /// `LifecycleFrame::MetadataCommitTick` from shard 0. See - /// [`DEFAULT_RECONCILE_PERIODIC_INTERVAL`] for the rationale; values - /// above [`RECONCILE_PERIODIC_INTERVAL_MAX`] are rejected by the - /// validator. - #[serde(default = "default_reconcile_periodic_interval")] - #[serde_as(as = "DisplayFromStr")] - #[config_env(leaf)] - pub reconcile_periodic_interval: IggyDuration, -} - -impl Default for ShardingConfig { - fn default() -> Self { - Self { - cpu_allocation: CpuAllocation::default(), - inbox_capacity: DEFAULT_INBOX_CAPACITY, - shutdown_drain_timeout: default_shutdown_drain_timeout(), - shutdown_poll_interval: default_shutdown_poll_interval(), - reconcile_periodic_interval: default_reconcile_periodic_interval(), - } - } -} #[derive(Debug, Clone, PartialEq, Default)] pub enum CpuAllocation { diff --git a/core/server_common/src/sharding/mod.rs b/core/server_common/src/sharding/mod.rs index 761e41d04..5435a072c 100644 --- a/core/server_common/src/sharding/mod.rs +++ b/core/server_common/src/sharding/mod.rs @@ -15,11 +15,13 @@ // specific language governing permissions and limitations // under the License. +mod cpu_allocation; mod local_idx; mod namespace; mod partition_location; mod shard_id; +pub use cpu_allocation::{CpuAllocation, NumaConfig}; pub use local_idx::LocalIdx; pub use namespace::{ IggyNamespace, MAX_PARTITIONS, MAX_STREAMS, MAX_TOPICS, METADATA_CONSENSUS_NAMESPACE,
