You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by gk...@apache.org on 2026/07/06 09:40:57 UTC
(iggy) branch server_ng_decouple created (now fbe949976)
This is an automated email from the ASF dual-hosted git repository.
numinnex pushed a change to branch server_ng_decouple
in repository https://gitbox.apache.org/repos/asf/iggy.git
at fbe949976 refactor(server-ng): drop dependency on the legacy server crate
This branch includes the following new commits:
new fbe949976 refactor(server-ng): drop dependency on the legacy server crate
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iggy) 01/01: refactor(server-ng): drop dependency on the legacy server crate
Posted by gk...@apache.org.
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 <g....@gmail.com>
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,