You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by hg...@apache.org on 2026/01/27 15:21:47 UTC
(iggy) branch master updated: refactor(server): compile-time env var mappings eliminate runtime ambiguity (#2614)
This is an automated email from the ASF dual-hosted git repository.
hgruszecki pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 052431382 refactor(server): compile-time env var mappings eliminate runtime ambiguity (#2614)
052431382 is described below
commit 05243138255349a78bd1e086a0d7fb264682f980
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Tue Jan 27 16:21:41 2026 +0100
refactor(server): compile-time env var mappings eliminate runtime ambiguity (#2614)
Replace dynamic path inference with derive macro that generates
explicit mappings at compile time. Field attribute #[config_env(leaf)]
gives users control over nested vs value type handling without
hardcoding type names in the macro.
Added 2 new crates: `configs` and `configs_derive` which are meant
to not be published to crates.io.
Renamed `server.toml` to `config.toml` and moved it to `core/server`.
---
.github/config/components.yml | 54 +-
Cargo.lock | 178 ++--
Cargo.toml | 8 +
DEPENDENCIES.md | 60 +-
README.md | 8 +-
bdd/docker-compose.yml | 34 +-
core/ai/mcp/Cargo.toml | 3 +-
core/ai/mcp/src/configs.rs | 39 +-
core/ai/mcp/src/main.rs | 2 +-
core/common/Cargo.toml | 3 +-
core/common/src/configs/mod.rs | 893 ---------------------
core/common/src/lib.rs | 3 +-
core/configs/Cargo.toml | 30 +
core/configs/src/configs_impl/env_mapping.rs | 63 ++
core/configs/src/configs_impl/error.rs | 51 ++
core/configs/src/configs_impl/file_provider.rs | 142 ++++
core/configs/src/configs_impl/mod.rs | 41 +
core/configs/src/configs_impl/parsing.rs | 196 +++++
core/configs/src/configs_impl/traits.rs | 36 +
.../configs/src/configs_impl/typed_env_provider.rs | 597 ++++++++++++++
core/configs/src/lib.rs | 28 +
core/configs_derive/Cargo.toml | 32 +
core/configs_derive/src/config_env.rs | 583 ++++++++++++++
core/configs_derive/src/lib.rs | 120 +++
core/connectors/runtime/Cargo.toml | 2 +
core/connectors/runtime/src/api/config.rs | 8 +-
core/connectors/runtime/src/configs/connectors.rs | 17 +-
.../src/configs/connectors/local_provider.rs | 50 +-
core/connectors/runtime/src/configs/runtime.rs | 53 +-
core/connectors/runtime/src/main.rs | 2 +-
core/integration/Cargo.toml | 1 +
core/integration/src/test_connectors_runtime.rs | 5 -
core/integration/src/test_server.rs | 3 +-
core/integration/tests/config_provider/mod.rs | 23 +-
.../data_integrity/verify_after_server_restart.rs | 2 +-
core/server/Cargo.toml | 1 +
core/server/README.md | 2 +-
core/{configs/server.toml => server/config.toml} | 0
core/server/src/args.rs | 4 +-
core/server/src/configs/cluster.rs | 11 +-
core/server/src/configs/defaults.rs | 4 +-
core/server/src/configs/http.rs | 17 +-
core/server/src/configs/quic.rs | 15 +-
core/server/src/configs/server.rs | 83 +-
core/server/src/configs/sharding.rs | 4 +-
core/server/src/configs/system.rs | 43 +-
core/server/src/configs/tcp.rs | 11 +-
core/server/src/configs/validators.rs | 39 +-
core/server/src/configs/websocket.rs | 5 +-
core/server/src/server_error.rs | 2 +-
examples/java/README.md | 2 +-
examples/rust/README.md | 2 +-
foreign/node/src/e2e/tcp.cluster.e2e.ts | 2 +-
53 files changed, 2390 insertions(+), 1227 deletions(-)
diff --git a/.github/config/components.yml b/.github/config/components.yml
index 8a87480ee..6ea37faaa 100644
--- a/.github/config/components.yml
+++ b/.github/config/components.yml
@@ -37,10 +37,17 @@ components:
paths:
- "core/sdk/**"
+ rust-configs:
+ depends_on:
+ - "rust-workspace"
+ paths:
+ - "core/configs/**"
+ - "core/configs_derive/**"
+
rust-common:
depends_on:
- - "rust-workspace" # Common is affected by workspace changes
- - "ci-infrastructure" # CI changes trigger full regression
+ - "rust-workspace"
+ - "rust-configs"
paths:
- "core/common/**"
@@ -51,28 +58,52 @@ components:
paths:
- "core/binary_protocol/**"
+ rust-clock:
+ depends_on:
+ - "rust-workspace"
+ paths:
+ - "core/clock/**"
+
+ rust-journal:
+ depends_on:
+ - "rust-workspace"
+ paths:
+ - "core/journal/**"
+
+ rust-partitions:
+ depends_on:
+ - "rust-workspace"
+ paths:
+ - "core/partitions/**"
+
rust-server:
depends_on:
- - "rust-workspace" # Server is affected by workspace changes
- - "ci-infrastructure" # CI changes trigger full regression
+ - "rust-workspace"
+ - "rust-configs"
- "rust-common"
- "rust-binary-protocol"
+ - "rust-journal"
+ - "rust-partitions"
paths:
- "core/server/**"
rust-cluster:
depends_on:
- - "rust-workspace" # Cluster is affected by workspace changes
- - "ci-infrastructure" # CI changes trigger full regression
+ - "rust-workspace"
+ - "rust-clock"
paths:
- - "core/consensus/"
- - "core/metadata/"
- - "core/message_bus/"
+ - "core/consensus/**"
+ - "core/metadata/**"
+ - "core/message_bus/**"
# Main Rust workspace testing
rust:
depends_on:
- "rust-workspace"
+ - "rust-configs"
+ - "rust-clock"
+ - "rust-journal"
+ - "rust-partitions"
- "rust-sdk"
- "rust-common"
- "rust-binary-protocol"
@@ -84,7 +115,7 @@ components:
- "rust-connectors"
- "rust-mcp"
- "rust-integration"
- - "ci-infrastructure" # CI changes trigger full regression
+ - "ci-infrastructure"
paths:
- "Dockerfile*"
- "!foreign/**" # Exclude foreign SDKs
@@ -121,6 +152,7 @@ components:
rust-connectors:
depends_on:
- "rust-sdk"
+ - "rust-configs"
paths:
- "core/connectors/**"
@@ -128,6 +160,7 @@ components:
rust-mcp:
depends_on:
- "rust-sdk"
+ - "rust-configs"
paths:
- "core/ai/mcp/**"
@@ -138,6 +171,7 @@ components:
- "rust-common"
- "rust-binary-protocol"
- "rust-server"
+ - "rust-configs"
paths:
- "core/integration/**"
diff --git a/Cargo.lock b/Cargo.lock
index 4d342f57b..e87d41682 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -893,9 +893,9 @@ dependencies = [
[[package]]
name = "aws-lc-rs"
-version = "1.15.4"
+version = "1.15.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7b7b6141e96a8c160799cc2d5adecd5cbbe5054cb8c7c4af53da0f83bb7ad256"
+checksum = "6a88aab2464f1f25453baa7a07c84c5b7684e274054ba06817f382357f77a288"
dependencies = [
"aws-lc-sys",
"zeroize",
@@ -903,9 +903,9 @@ dependencies = [
[[package]]
name = "aws-lc-sys"
-version = "0.37.0"
+version = "0.35.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5c34dda4df7017c8db52132f0f8a2e0f8161649d15723ed63fc00c82d0f2081a"
+checksum = "b45afffdee1e7c9126814751f88dddc747f41d91da16c9551a0f1e8a11e788a1"
dependencies = [
"cc",
"cmake",
@@ -1317,7 +1317,7 @@ version = "3.8.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "89ec27229c38ed0eb3c0feee3d2c1d6a4379ae44f418a29a658890e062d8f365"
dependencies = [
- "darling 0.20.11",
+ "darling 0.23.0",
"ident_case",
"prettyplease",
"proc-macro2",
@@ -1498,9 +1498,9 @@ dependencies = [
[[package]]
name = "cc"
-version = "1.2.54"
+version = "1.2.52"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6354c81bbfd62d9cfa9cb3c773c2b7b2a3a482d569de977fd0e961f6e7c00583"
+checksum = "cd4932aefd12402b36c60956a4fe0035421f544799057659ff86f923657aada3"
dependencies = [
"find-msvc-tools",
"jobserver",
@@ -1665,7 +1665,7 @@ version = "3.1.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "faf9468729b8cbcea668e36183cb69d317348c2e08e994829fb56ebfdfbaac34"
dependencies = [
- "windows-sys 0.48.0",
+ "windows-sys 0.61.2",
]
[[package]]
@@ -1914,6 +1914,27 @@ dependencies = [
"crossbeam-utils",
]
+[[package]]
+name = "configs"
+version = "0.1.0"
+dependencies = [
+ "configs_derive",
+ "figment",
+ "serde",
+ "serde_json",
+ "tracing",
+]
+
+[[package]]
+name = "configs_derive"
+version = "0.1.0"
+dependencies = [
+ "darling 0.20.11",
+ "proc-macro2",
+ "quote",
+ "syn 2.0.114",
+]
+
[[package]]
name = "consensus"
version = "0.1.0"
@@ -2665,7 +2686,7 @@ dependencies = [
"libc",
"option-ext",
"redox_users",
- "windows-sys 0.59.0",
+ "windows-sys 0.61.2",
]
[[package]]
@@ -2969,7 +2990,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb"
dependencies = [
"libc",
- "windows-sys 0.59.0",
+ "windows-sys 0.61.2",
]
[[package]]
@@ -3160,20 +3181,21 @@ dependencies = [
[[package]]
name = "filetime"
-version = "0.2.27"
+version = "0.2.26"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f98844151eee8917efc50bd9e8318cb963ae8b297431495d3f758616ea5c57db"
+checksum = "bc0505cd1b6fa6580283f6bdf70a73fcf4aba1184038c90902b92b3dd0df63ed"
dependencies = [
"cfg-if",
"libc",
"libredox",
+ "windows-sys 0.60.2",
]
[[package]]
name = "find-msvc-tools"
-version = "0.1.8"
+version = "0.1.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8591b0bcc8a98a64310a2fae1bb3e9b8564dd10e381e6e28010fde8e8e8568db"
+checksum = "f449e6c6c08c865631d4890cfacf252b3d396c9bcc83adb6623cdb02a8336c41"
[[package]]
name = "flatbuffers"
@@ -4521,6 +4543,8 @@ dependencies = [
"axum-server",
"chrono",
"clap",
+ "configs",
+ "configs_derive",
"dashmap",
"derive_more",
"dirs",
@@ -4566,12 +4590,13 @@ version = "0.2.1-edge.5"
dependencies = [
"axum",
"axum-server",
+ "configs",
+ "configs_derive",
"dirs",
"dotenvy",
"figlet-rs",
"figment",
"iggy",
- "iggy_common",
"opentelemetry",
"opentelemetry-appender-tracing",
"opentelemetry-otlp",
@@ -4631,11 +4656,11 @@ dependencies = [
"compio-quic",
"compio-tls",
"compio-ws",
+ "configs_derive",
"crossbeam",
"derive_more",
"err_trail",
"fast-async-mutex",
- "figment",
"human-repr",
"humantime",
"nix 0.31.1",
@@ -4650,7 +4675,6 @@ dependencies = [
"strum",
"thiserror 2.0.18",
"tokio",
- "toml 0.9.11+spec-1.1.0",
"tracing",
"tungstenite",
"twox-hash",
@@ -4966,6 +4990,7 @@ dependencies = [
"bytes",
"chrono",
"compio",
+ "configs",
"ctor",
"derive_more",
"env_logger",
@@ -5088,7 +5113,7 @@ dependencies = [
"portable-atomic",
"portable-atomic-util",
"serde_core",
- "windows-sys 0.59.0",
+ "windows-sys 0.61.2",
]
[[package]]
@@ -5155,9 +5180,9 @@ version = "0.1.0"
[[package]]
name = "js-sys"
-version = "0.3.85"
+version = "0.3.83"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8c942ebf8e95485ca0d52d97da7c5a2c387d0e7f0ba4c35e93bfcaee045955b3"
+checksum = "464a3709c7f55f1f721e5389aa6ea4e3bc6aba669353300af094b29ffbdde1d8"
dependencies = [
"once_cell",
"wasm-bindgen",
@@ -5390,9 +5415,9 @@ dependencies = [
[[package]]
name = "liblzma-sys"
-version = "0.4.5"
+version = "0.4.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9f2db66f3268487b5033077f266da6777d057949b8f93c8ad82e441df25e6186"
+checksum = "01b9596486f6d60c3bbe644c0e1be1aa6ccc472ad630fe8927b456973d7cb736"
dependencies = [
"cc",
"libc",
@@ -5401,9 +5426,9 @@ dependencies = [
[[package]]
name = "libm"
-version = "0.2.16"
+version = "0.2.15"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b6d2cec3eae94f9f509c767b45932f1ada8350c4bdb85af2fcab4a3c14807981"
+checksum = "f9fbbcab51052fe104eb5e5d351cf728d30a5be1fe14d9be8a3b097481fb97de"
[[package]]
name = "libmimalloc-sys"
@@ -5861,12 +5886,9 @@ dependencies = [
[[package]]
name = "notify-types"
-version = "2.1.0"
+version = "2.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "42b8cfee0e339a0337359f3c88165702ac6e600dc01c0cc9579a92d62b08477a"
-dependencies = [
- "bitflags 2.10.0",
-]
+checksum = "5e0826a989adedc2a244799e823aece04662b66609d96af8dff7ac6df9a8925d"
[[package]]
name = "nougat"
@@ -5904,7 +5926,7 @@ version = "0.50.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5"
dependencies = [
- "windows-sys 0.59.0",
+ "windows-sys 0.61.2",
]
[[package]]
@@ -5959,9 +5981,9 @@ dependencies = [
[[package]]
name = "num-conv"
-version = "0.2.0"
+version = "0.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "cf97ec579c3c42f953ef76dbf8d55ac91fb219dde70e49aa4a6b7d74e9919050"
+checksum = "51d515d32fb182ee37cda2ccdcb92950d6a3c2893aa280e540671c2cd0f3b1d9"
[[package]]
name = "num-integer"
@@ -6198,9 +6220,9 @@ dependencies = [
[[package]]
name = "openssl-probe"
-version = "0.2.1"
+version = "0.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7c87def4c32ab89d880effc9e097653c8da5d6ef28e6b539d313baaacfbafcbe"
+checksum = "9f50d9b3dabb09ecd771ad0aa242ca6894994c130308ca3d7684634df8037391"
[[package]]
name = "openssl-src"
@@ -6360,7 +6382,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "7d8fae84b431384b68627d0f9b3b1245fcf9f46f6c0e3dc902e9dce64edd1967"
dependencies = [
"libc",
- "windows-sys 0.48.0",
+ "windows-sys 0.61.2",
]
[[package]]
@@ -6893,9 +6915,9 @@ dependencies = [
[[package]]
name = "proc-macro2"
-version = "1.0.106"
+version = "1.0.105"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8fd00f0bb2e90d81d1044c2b32617f68fcb9fa3bb7640c23e9c748e53fb30934"
+checksum = "535d180e0ecab6268a3e718bb9fd44db66bbbc256257165fc699dadf70d16fe7"
dependencies = [
"unicode-ident",
]
@@ -7122,14 +7144,14 @@ dependencies = [
"once_cell",
"socket2 0.5.10",
"tracing",
- "windows-sys 0.59.0",
+ "windows-sys 0.60.2",
]
[[package]]
name = "quote"
-version = "1.0.44"
+version = "1.0.43"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "21b2ebcf727b7760c461f091f9f0f539b77b8e87f2fd88131e7f1b433b3cece4"
+checksum = "dc74d9a594b72ae6656596548f56f667211f8a97b3d4c3d467150794690dc40a"
dependencies = [
"proc-macro2",
]
@@ -7736,7 +7758,7 @@ dependencies = [
"errno",
"libc",
"linux-raw-sys",
- "windows-sys 0.59.0",
+ "windows-sys 0.61.2",
]
[[package]]
@@ -7778,9 +7800,9 @@ dependencies = [
[[package]]
name = "rustls-pki-types"
-version = "1.14.0"
+version = "1.13.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "be040f8b0a225e40375822a563fa9524378b9d63112f53e19ffff34df5d33fdd"
+checksum = "21e6f2ab2928ca4291b86736a8bd920a277a399bba1589409d72154ff87c1282"
dependencies = [
"web-time",
"zeroize",
@@ -7804,7 +7826,7 @@ dependencies = [
"security-framework",
"security-framework-sys",
"webpki-root-certs",
- "windows-sys 0.59.0",
+ "windows-sys 0.61.2",
]
[[package]]
@@ -7815,9 +7837,9 @@ checksum = "f87165f0995f63a9fbeea62b64d10b4d9d8e78ec6d7d51fb2125fda7bb36788f"
[[package]]
name = "rustls-webpki"
-version = "0.103.9"
+version = "0.103.8"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d7df23109aa6c1567d1c575b9952556388da57401e4ace1d15f79eedad0d8f53"
+checksum = "2ffdfa2f5286e2247234e03f680868ac2815974dc39e00ea15adc445d0aafe52"
dependencies = [
"aws-lc-rs",
"ring",
@@ -7990,7 +8012,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5b55fb86dfd3a2f5f76ea78310a88f96c4ea21a3031f8d212443d56123fd0521"
dependencies = [
"libc",
- "windows-sys 0.59.0",
+ "windows-sys 0.61.2",
]
[[package]]
@@ -8234,6 +8256,7 @@ dependencies = [
"compio-quic",
"compio-tls",
"compio-ws",
+ "configs",
"ctrlc",
"cyper",
"cyper-axum",
@@ -8948,7 +8971,7 @@ dependencies = [
"getrandom 0.3.4",
"once_cell",
"rustix",
- "windows-sys 0.59.0",
+ "windows-sys 0.61.2",
]
[[package]]
@@ -9112,9 +9135,9 @@ dependencies = [
[[package]]
name = "time"
-version = "0.3.46"
+version = "0.3.45"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9da98b7d9b7dad93488a84b8248efc35352b0b2657397d4167e7ad67e5d535e5"
+checksum = "f9e442fc33d7fdb45aa9bfeb312c095964abdf596f7567261062b2a7107aaabd"
dependencies = [
"deranged",
"itoa",
@@ -9129,15 +9152,15 @@ dependencies = [
[[package]]
name = "time-core"
-version = "0.1.8"
+version = "0.1.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7694e1cfe791f8d31026952abf09c69ca6f6fa4e1a1229e18988f06a04a12dca"
+checksum = "8b36ee98fd31ec7426d599183e8fe26932a8dc1fb76ddb6214d05493377d34ca"
[[package]]
name = "time-macros"
-version = "0.2.26"
+version = "0.2.25"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "78cc610bac2dcee56805c99642447d4c5dbde4d01f752ffea0199aee1f601dc4"
+checksum = "71e552d1249bf61ac2a52db88179fd0673def1e1ad8243a00d9ec9ed71fee3dd"
dependencies = [
"num-conv",
"time-core",
@@ -9988,9 +10011,9 @@ checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b"
[[package]]
name = "wasip2"
-version = "1.0.2+wasi-0.2.9"
+version = "1.0.1+wasi-0.2.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9517f9239f02c069db75e65f174b3da828fe5f5b945c4dd26bd25d89c03ebcf5"
+checksum = "0562428422c63773dad2c345a1882263bbf4d65cf3f42e90921f787ef5ad58e7"
dependencies = [
"wit-bindgen",
]
@@ -10003,9 +10026,9 @@ checksum = "b8dad83b4f25e74f184f64c43b150b91efe7647395b42289f38e50566d82855b"
[[package]]
name = "wasm-bindgen"
-version = "0.2.108"
+version = "0.2.106"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "64024a30ec1e37399cf85a7ffefebdb72205ca1c972291c51512360d90bd8566"
+checksum = "0d759f433fa64a2d763d1340820e46e111a7a5ab75f993d1852d70b03dbb80fd"
dependencies = [
"cfg-if",
"once_cell",
@@ -10016,12 +10039,11 @@ dependencies = [
[[package]]
name = "wasm-bindgen-futures"
-version = "0.4.58"
+version = "0.4.56"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "70a6e77fd0ae8029c9ea0063f87c46fde723e7d887703d74ad2616d792e51e6f"
+checksum = "836d9622d604feee9e5de25ac10e3ea5f2d65b41eac0d9ce72eb5deae707ce7c"
dependencies = [
"cfg-if",
- "futures-util",
"js-sys",
"once_cell",
"wasm-bindgen",
@@ -10030,9 +10052,9 @@ dependencies = [
[[package]]
name = "wasm-bindgen-macro"
-version = "0.2.108"
+version = "0.2.106"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "008b239d9c740232e71bd39e8ef6429d27097518b6b30bdf9086833bd5b6d608"
+checksum = "48cb0d2638f8baedbc542ed444afc0644a29166f1595371af4fecf8ce1e7eeb3"
dependencies = [
"quote",
"wasm-bindgen-macro-support",
@@ -10040,9 +10062,9 @@ dependencies = [
[[package]]
name = "wasm-bindgen-macro-support"
-version = "0.2.108"
+version = "0.2.106"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5256bae2d58f54820e6490f9839c49780dff84c65aeab9e772f15d5f0e913a55"
+checksum = "cefb59d5cd5f92d9dcf80e4683949f15ca4b511f4ac0a6e14d4e1ac60c6ecd40"
dependencies = [
"bumpalo",
"proc-macro2",
@@ -10053,9 +10075,9 @@ dependencies = [
[[package]]
name = "wasm-bindgen-shared"
-version = "0.2.108"
+version = "0.2.106"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1f01b580c9ac74c8d8f0c0e4afb04eeef2acf145458e52c03845ee9cd23e3d12"
+checksum = "cbc538057e648b67f72a982e708d485b2efa771e1ac05fec311f9f63e5800db4"
dependencies = [
"unicode-ident",
]
@@ -10089,9 +10111,9 @@ dependencies = [
[[package]]
name = "web-sys"
-version = "0.3.85"
+version = "0.3.83"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "312e32e551d92129218ea9a2452120f4aabc03529ef03e4d0d82fb2780608598"
+checksum = "9b32828d774c412041098d182a8b38b16ea816958e07cf40eec2bc080ae137ac"
dependencies = [
"js-sys",
"wasm-bindgen",
@@ -10173,7 +10195,7 @@ version = "0.1.11"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22"
dependencies = [
- "windows-sys 0.48.0",
+ "windows-sys 0.61.2",
]
[[package]]
@@ -10698,9 +10720,9 @@ dependencies = [
[[package]]
name = "wit-bindgen"
-version = "0.51.0"
+version = "0.46.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d7249219f66ced02969388cf2bb044a09756a083d0fab1e566056b04d9fbcaa5"
+checksum = "f17a85883d4e6d00e8a97c586de764dabcc06133f7f1d55dce5cdc070ad7fe59"
[[package]]
name = "writeable"
@@ -10855,18 +10877,18 @@ dependencies = [
[[package]]
name = "zerocopy"
-version = "0.8.34"
+version = "0.8.33"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "71ddd76bcebeed25db614f82bf31a9f4222d3fbba300e6fb6c00afa26cbd4d9d"
+checksum = "668f5168d10b9ee831de31933dc111a459c97ec93225beb307aed970d1372dfd"
dependencies = [
"zerocopy-derive",
]
[[package]]
name = "zerocopy-derive"
-version = "0.8.34"
+version = "0.8.33"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d8187381b52e32220d50b255276aa16a084ec0a9017a0ca2152a1f55c539758d"
+checksum = "2c7962b26b0a8685668b671ee4b54d007a67d4eaf05fda79ac0ecf41e32270f1"
dependencies = [
"proc-macro2",
"quote",
@@ -10969,9 +10991,9 @@ checksum = "40990edd51aae2c2b6907af74ffb635029d5788228222c4bb811e9351c0caad3"
[[package]]
name = "zmij"
-version = "1.0.17"
+version = "1.0.14"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "02aae0f83f69aafc94776e879363e9771d7ecbffe2c7fbb6c14c5e00dfe88439"
+checksum = "bd8f3f50b848df28f887acb68e41201b5aea6bc8a8dacc00fb40635ff9a72fea"
[[package]]
name = "zopfli"
diff --git a/Cargo.toml b/Cargo.toml
index 129bc2203..e2aba3ca2 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -29,6 +29,8 @@ members = [
"core/cli",
"core/clock",
"core/common",
+ "core/configs",
+ "core/configs_derive",
"core/connectors/runtime",
"core/connectors/sdk",
"core/connectors/sinks/elasticsearch_sink",
@@ -110,10 +112,13 @@ compio-tls = { version = "0.8.0", features = [
"rustls",
], default-features = false }
compio-ws = "0.2.0"
+configs = { path = "core/configs", version = "0.1.0" }
+configs_derive = { path = "core/configs_derive", version = "0.1.0" }
console-subscriber = "0.5.0"
crossbeam = "0.8.4"
cyper = { version = "0.7.1", features = ["rustls"], default-features = false }
cyper-axum = { version = "0.7.1" }
+darling = "0.20"
dashmap = "6.1.0"
derive-new = "0.7.0"
derive_builder = "0.20.2"
@@ -173,7 +178,9 @@ parquet = "57.2.0"
passterm = "=2.0.1"
postcard = { version = "1.1.3", features = ["alloc"] }
predicates = "3.1.3"
+proc-macro2 = "1"
quinn = "0.11.9"
+quote = "1"
rand = "0.9.2"
rand_xoshiro = "0.7.0"
regex = "1.12.2"
@@ -200,6 +207,7 @@ simd-json = { version = "0.17.0", features = ["serde_impl"] }
slab = "0.4.11"
strum = { version = "0.27.2", features = ["derive"] }
strum_macros = "0.27.2"
+syn = { version = "2", features = ["full", "extra-traits"] }
sysinfo = "0.38.0"
tempfile = "3.24.0"
test-case = "3.3.1"
diff --git a/DEPENDENCIES.md b/DEPENDENCIES.md
index 1320800e7..c00b459ed 100644
--- a/DEPENDENCIES.md
+++ b/DEPENDENCIES.md
@@ -71,8 +71,8 @@ atomic-polyfill: 1.0.3, "Apache-2.0 OR MIT",
atomic-waker: 1.1.2, "Apache-2.0 OR MIT",
autocfg: 1.5.0, "Apache-2.0 OR MIT",
autotools: 0.2.7, "MIT",
-aws-lc-rs: 1.15.4, "(Apache-2.0 OR ISC) AND ISC",
-aws-lc-sys: 0.37.0, "(Apache-2.0 OR ISC) AND ISC AND OpenSSL",
+aws-lc-rs: 1.15.2, "(Apache-2.0 OR ISC) AND ISC",
+aws-lc-sys: 0.35.0, "(Apache-2.0 OR ISC) AND ISC AND OpenSSL",
axum: 0.8.8, "MIT",
axum-core: 0.5.6, "MIT",
axum-macros: 0.5.0, "MIT",
@@ -123,7 +123,7 @@ bzip2: 0.6.1, "Apache-2.0 OR MIT",
camino: 1.2.2, "Apache-2.0 OR MIT",
cargo-platform: 0.3.2, "Apache-2.0 OR MIT",
cargo_metadata: 0.23.1, "MIT",
-cc: 1.2.54, "Apache-2.0 OR MIT",
+cc: 1.2.52, "Apache-2.0 OR MIT",
cesu8: 1.1.0, "Apache-2.0 OR MIT",
cfg-if: 1.0.4, "Apache-2.0 OR MIT",
cfg_aliases: 0.2.1, "MIT",
@@ -158,6 +158,8 @@ compio-ws: 0.2.0, "MIT",
compression-codecs: 0.4.36, "Apache-2.0 OR MIT",
compression-core: 0.4.31, "Apache-2.0 OR MIT",
concurrent-queue: 2.5.0, "Apache-2.0 OR MIT",
+configs: 0.1.0, "Apache-2.0",
+configs_derive: 0.1.0, "Apache-2.0",
consensus: 0.1.0, "Apache-2.0",
console: 0.16.2, "MIT",
console_error_panic_hook: 0.1.7, "Apache-2.0 OR MIT",
@@ -277,8 +279,8 @@ fiat-crypto: 0.2.9, "Apache-2.0 OR BSD-1-Clause OR MIT",
figlet-rs: 0.1.5, "Apache-2.0",
figment: 0.10.19, "Apache-2.0 OR MIT",
file-operation: 0.8.17, "MIT",
-filetime: 0.2.27, "Apache-2.0 OR MIT",
-find-msvc-tools: 0.1.8, "Apache-2.0 OR MIT",
+filetime: 0.2.26, "Apache-2.0 OR MIT",
+find-msvc-tools: 0.1.7, "Apache-2.0 OR MIT",
flatbuffers: 25.12.19, "Apache-2.0",
flate2: 1.1.8, "Apache-2.0 OR MIT",
float-cmp: 0.10.0, "MIT",
@@ -431,7 +433,7 @@ jni: 0.21.1, "Apache-2.0 OR MIT",
jni-sys: 0.3.0, "Apache-2.0 OR MIT",
jobserver: 0.1.34, "Apache-2.0 OR MIT",
journal: 0.1.0, "Apache-2.0",
-js-sys: 0.3.85, "Apache-2.0 OR MIT",
+js-sys: 0.3.83, "Apache-2.0 OR MIT",
jsonwebtoken: 10.2.0, "MIT",
jwalk: 0.8.1, "MIT",
keccak: 0.1.5, "Apache-2.0 OR MIT",
@@ -454,8 +456,8 @@ libc: 0.2.180, "Apache-2.0 OR MIT",
libdbus-sys: 0.2.7, "Apache-2.0 OR MIT",
libgit2-sys: 0.18.3+1.9.2, "Apache-2.0 OR MIT",
liblzma: 0.4.5, "Apache-2.0 OR MIT",
-liblzma-sys: 0.4.5, "Apache-2.0 OR MIT",
-libm: 0.2.16, "MIT",
+liblzma-sys: 0.4.4, "Apache-2.0 OR MIT",
+libm: 0.2.15, "MIT",
libmimalloc-sys: 0.1.44, "MIT",
libredox: 0.1.12, "MIT",
libsqlite3-sys: 0.30.1, "MIT",
@@ -504,7 +506,7 @@ nonzero_ext: 0.3.0, "Apache-2.0",
nonzero_lit: 0.1.2, "Apache-2.0 OR CC0-1.0 OR MIT",
normalize-line-endings: 0.3.0, "Apache-2.0",
notify: 8.2.0, "CC0-1.0",
-notify-types: 2.1.0, "Apache-2.0 OR MIT",
+notify-types: 2.0.0, "Apache-2.0 OR MIT",
nougat: 0.2.4, "Apache-2.0 OR MIT OR Zlib",
nougat-proc_macros: 0.2.4, "Apache-2.0 OR MIT OR Zlib",
ntapi: 0.4.2, "Apache-2.0 OR MIT",
@@ -513,7 +515,7 @@ num: 0.4.3, "Apache-2.0 OR MIT",
num-bigint: 0.4.6, "Apache-2.0 OR MIT",
num-bigint-dig: 0.8.6, "Apache-2.0 OR MIT",
num-complex: 0.4.6, "Apache-2.0 OR MIT",
-num-conv: 0.2.0, "Apache-2.0 OR MIT",
+num-conv: 0.1.0, "Apache-2.0 OR MIT",
num-integer: 0.1.46, "Apache-2.0 OR MIT",
num-iter: 0.1.45, "Apache-2.0 OR MIT",
num-modular: 0.6.1, "Apache-2.0",
@@ -534,7 +536,7 @@ opaque-debug: 0.3.1, "Apache-2.0 OR MIT",
opendal: 0.55.0, "Apache-2.0",
openssl: 0.10.75, "Apache-2.0",
openssl-macros: 0.1.1, "Apache-2.0 OR MIT",
-openssl-probe: 0.2.1, "Apache-2.0 OR MIT",
+openssl-probe: 0.2.0, "Apache-2.0 OR MIT",
openssl-src: 300.5.4+3.5.4, "Apache-2.0 OR MIT",
openssl-sys: 0.9.111, "MIT",
opentelemetry: 0.31.0, "Apache-2.0",
@@ -601,7 +603,7 @@ proc-macro-crate: 1.3.1, "Apache-2.0 OR MIT",
proc-macro-crate: 3.4.0, "Apache-2.0 OR MIT",
proc-macro-error: 1.0.4, "Apache-2.0 OR MIT",
proc-macro-error-attr: 1.0.4, "Apache-2.0 OR MIT",
-proc-macro2: 1.0.106, "Apache-2.0 OR MIT",
+proc-macro2: 1.0.105, "Apache-2.0 OR MIT",
proc-macro2-diagnostics: 0.10.1, "Apache-2.0 OR MIT",
prometheus-client: 0.24.0, "Apache-2.0 OR MIT",
prometheus-client-derive-encode: 0.5.0, "Apache-2.0 OR MIT",
@@ -620,7 +622,7 @@ quick-xml: 0.38.4, "MIT",
quinn: 0.11.9, "Apache-2.0 OR MIT",
quinn-proto: 0.11.13, "Apache-2.0 OR MIT",
quinn-udp: 0.5.14, "Apache-2.0 OR MIT",
-quote: 1.0.44, "Apache-2.0 OR MIT",
+quote: 1.0.43, "Apache-2.0 OR MIT",
r-efi: 5.3.0, "Apache-2.0 OR LGPL-2.1-or-later OR MIT",
radium: 0.7.0, "MIT",
rand: 0.8.5, "Apache-2.0 OR MIT",
@@ -674,10 +676,10 @@ rustix: 1.1.3, "Apache-2.0 OR Apache-2.0 WITH LLVM-exception OR MIT",
rustls: 0.23.36, "Apache-2.0 OR ISC OR MIT",
rustls-native-certs: 0.8.3, "Apache-2.0 OR ISC OR MIT",
rustls-pemfile: 2.2.0, "Apache-2.0 OR ISC OR MIT",
-rustls-pki-types: 1.14.0, "Apache-2.0 OR MIT",
+rustls-pki-types: 1.13.2, "Apache-2.0 OR MIT",
rustls-platform-verifier: 0.6.2, "Apache-2.0 OR MIT",
rustls-platform-verifier-android: 0.1.1, "Apache-2.0 OR MIT",
-rustls-webpki: 0.103.9, "ISC",
+rustls-webpki: 0.103.8, "ISC",
rustversion: 1.0.22, "Apache-2.0 OR MIT",
ryu: 1.0.22, "Apache-2.0 OR BSL-1.0",
same-file: 1.0.6, "MIT OR Unlicense",
@@ -785,9 +787,9 @@ thiserror-impl: 1.0.69, "Apache-2.0 OR MIT",
thiserror-impl: 2.0.18, "Apache-2.0 OR MIT",
thread_local: 1.1.9, "Apache-2.0 OR MIT",
thrift: 0.17.0, "Apache-2.0",
-time: 0.3.46, "Apache-2.0 OR MIT",
-time-core: 0.1.8, "Apache-2.0 OR MIT",
-time-macros: 0.2.26, "Apache-2.0 OR MIT",
+time: 0.3.45, "Apache-2.0 OR MIT",
+time-core: 0.1.7, "Apache-2.0 OR MIT",
+time-macros: 0.2.25, "Apache-2.0 OR MIT",
tiny-keccak: 2.0.2, "CC0-1.0",
tinystr: 0.8.2, "Unicode-3.0",
tinyvec: 1.10.0, "Apache-2.0 OR MIT OR Zlib",
@@ -871,16 +873,16 @@ wait-timeout: 0.2.1, "Apache-2.0 OR MIT",
walkdir: 2.5.0, "MIT OR Unlicense",
want: 0.3.1, "MIT",
wasi: 0.11.1+wasi-snapshot-preview1, "Apache-2.0 OR Apache-2.0 WITH LLVM-exception OR MIT",
-wasip2: 1.0.2+wasi-0.2.9, "Apache-2.0 OR Apache-2.0 WITH LLVM-exception OR MIT",
+wasip2: 1.0.1+wasi-0.2.4, "Apache-2.0 OR Apache-2.0 WITH LLVM-exception OR MIT",
wasite: 0.1.0, "Apache-2.0 OR BSL-1.0 OR MIT",
-wasm-bindgen: 0.2.108, "Apache-2.0 OR MIT",
-wasm-bindgen-futures: 0.4.58, "Apache-2.0 OR MIT",
-wasm-bindgen-macro: 0.2.108, "Apache-2.0 OR MIT",
-wasm-bindgen-macro-support: 0.2.108, "Apache-2.0 OR MIT",
-wasm-bindgen-shared: 0.2.108, "Apache-2.0 OR MIT",
+wasm-bindgen: 0.2.106, "Apache-2.0 OR MIT",
+wasm-bindgen-futures: 0.4.56, "Apache-2.0 OR MIT",
+wasm-bindgen-macro: 0.2.106, "Apache-2.0 OR MIT",
+wasm-bindgen-macro-support: 0.2.106, "Apache-2.0 OR MIT",
+wasm-bindgen-shared: 0.2.106, "Apache-2.0 OR MIT",
wasm-streams: 0.4.2, "Apache-2.0 OR MIT",
wasmtimer: 0.4.3, "MIT",
-web-sys: 0.3.85, "Apache-2.0 OR MIT",
+web-sys: 0.3.83, "Apache-2.0 OR MIT",
web-time: 1.1.0, "Apache-2.0 OR MIT",
webpki-root-certs: 1.0.5, "CDLA-Permissive-2.0",
webpki-roots: 0.26.11, "CDLA-Permissive-2.0",
@@ -953,7 +955,7 @@ windows_x86_64_msvc: 0.52.6, "Apache-2.0 OR MIT",
windows_x86_64_msvc: 0.53.1, "Apache-2.0 OR MIT",
winnow: 0.5.40, "MIT",
winnow: 0.7.14, "MIT",
-wit-bindgen: 0.51.0, "Apache-2.0 OR Apache-2.0 WITH LLVM-exception OR MIT",
+wit-bindgen: 0.46.0, "Apache-2.0 OR Apache-2.0 WITH LLVM-exception OR MIT",
writeable: 0.6.2, "Unicode-3.0",
wyz: 0.5.1, "MIT",
x509-parser: 0.18.0, "Apache-2.0 OR MIT",
@@ -966,8 +968,8 @@ yew-router: 0.19.0, "Apache-2.0 OR MIT",
yew-router-macro: 0.19.0, "Apache-2.0 OR MIT",
yoke: 0.8.1, "Unicode-3.0",
yoke-derive: 0.8.1, "Unicode-3.0",
-zerocopy: 0.8.34, "Apache-2.0 OR BSD-2-Clause OR MIT",
-zerocopy-derive: 0.8.34, "Apache-2.0 OR BSD-2-Clause OR MIT",
+zerocopy: 0.8.33, "Apache-2.0 OR BSD-2-Clause OR MIT",
+zerocopy-derive: 0.8.33, "Apache-2.0 OR BSD-2-Clause OR MIT",
zerofrom: 0.1.6, "Unicode-3.0",
zerofrom-derive: 0.1.6, "Unicode-3.0",
zeroize: 1.8.2, "Apache-2.0 OR MIT",
@@ -977,7 +979,7 @@ zerovec: 0.11.5, "Unicode-3.0",
zerovec-derive: 0.11.2, "Unicode-3.0",
zip: 7.2.0, "MIT",
zlib-rs: 0.5.5, "Zlib",
-zmij: 1.0.17, "MIT",
+zmij: 1.0.14, "MIT",
zopfli: 0.8.3, "Apache-2.0",
zstd: 0.13.3, "MIT",
zstd-safe: 7.2.4, "Apache-2.0 OR MIT",
diff --git a/README.md b/README.md
index 3718f2931..a847968e7 100644
--- a/README.md
+++ b/README.md
@@ -211,11 +211,11 @@ ulimits:
## Configuration
-The default configuration can be found in `server.toml` file in `configs` directory.
+The default configuration can be found in `config.toml` file in `core/server` directory.
-The configuration file is loaded from the current working directory, but you can specify the path to the configuration file by setting `IGGY_CONFIG_PATH` environment variable, for example `export IGGY_CONFIG_PATH=configs/server.toml` (or other command depending on OS).
+The configuration file is loaded from the current working directory, but you can specify the path to the configuration file by setting `IGGY_CONFIG_PATH` environment variable, for example `export IGGY_CONFIG_PATH=core/server/config.toml` (or other command depending on OS).
-When config file is not found, the default values from embedded `server.toml` file are used.
+When config file is not found, the default values from embedded `config.toml` file are used.
For the detailed documentation of the configuration file, please refer to the [configuration](https://iggy.apache.org/docs/server/configuration) section.
@@ -257,7 +257,7 @@ Start the server:
`cargo run --bin iggy-server`
-All the data used by the server will be persisted under the `local_data` directory by default, unless specified differently in the configuration (see `system.path` in `server.toml`).
+All the data used by the server will be persisted under the `local_data` directory by default, unless specified differently in the configuration (see `system.path` in `config.toml`).
One can use default root credentials with optional `--with-default-root-credentials`.
This flag is equivalent to setting `IGGY_ROOT_USERNAME=iggy` and `IGGY_ROOT_PASSWORD=iggy`, plus
diff --git a/bdd/docker-compose.yml b/bdd/docker-compose.yml
index 495fbc080..1062bb780 100644
--- a/bdd/docker-compose.yml
+++ b/bdd/docker-compose.yml
@@ -28,7 +28,7 @@ services:
PREBUILT_IGGY_CLI: ${IGGY_CLI_PATH:-target/debug/iggy}
LIBC: glibc
PROFILE: debug
- command: ["--fresh", "--with-default-root-credentials"]
+ command: [ "--fresh", "--with-default-root-credentials" ]
cap_add:
- SYS_NICE
security_opt:
@@ -38,7 +38,7 @@ services:
soft: -1
hard: -1
healthcheck:
- test: ["CMD", "/usr/local/bin/iggy", "--tcp-server-address", "127.0.0.1:8090", "ping"]
+ test: [ "CMD", "/usr/local/bin/iggy", "--tcp-server-address", "127.0.0.1:8090", "ping" ]
interval: 20s
timeout: 3s
retries: 30
@@ -68,7 +68,7 @@ services:
LIBC: glibc
PROFILE: debug
# Leader runs without --follower flag
- command: ["--fresh", "--with-default-root-credentials"]
+ command: [ "--fresh", "--with-default-root-credentials" ]
cap_add:
- SYS_NICE
security_opt:
@@ -78,7 +78,7 @@ services:
soft: -1
hard: -1
healthcheck:
- test: ["CMD", "/usr/local/bin/iggy", "--tcp-server-address", "127.0.0.1:8091", "ping"]
+ test: [ "CMD", "/usr/local/bin/iggy", "--tcp-server-address", "127.0.0.1:8091", "ping" ]
interval: 20s
timeout: 3s
retries: 30
@@ -94,7 +94,6 @@ services:
# Cluster configuration
- IGGY_CLUSTER_ENABLED=true
- IGGY_CLUSTER_NAME=test-cluster
- - IGGY_CLUSTER_ID=1
# Current node identity
- IGGY_CLUSTER_NODE_CURRENT_NAME=leader-node
- IGGY_CLUSTER_NODE_CURRENT_IP=iggy-leader
@@ -123,7 +122,7 @@ services:
LIBC: glibc
PROFILE: debug
# Follower runs with --follower flag
- command: ["--fresh", "--with-default-root-credentials", "--follower"]
+ command: [ "--fresh", "--with-default-root-credentials", "--follower" ]
cap_add:
- SYS_NICE
security_opt:
@@ -133,7 +132,7 @@ services:
soft: -1
hard: -1
healthcheck:
- test: ["CMD", "/usr/local/bin/iggy", "--tcp-server-address", "127.0.0.1:8092", "ping"]
+ test: [ "CMD", "/usr/local/bin/iggy", "--tcp-server-address", "127.0.0.1:8092", "ping" ]
interval: 20s
timeout: 3s
retries: 30
@@ -149,7 +148,6 @@ services:
# Cluster configuration (same as leader)
- IGGY_CLUSTER_ENABLED=true
- IGGY_CLUSTER_NAME=test-cluster
- - IGGY_CLUSTER_ID=1
# Current node identity (different from leader)
- IGGY_CLUSTER_NODE_CURRENT_NAME=follower-node
- IGGY_CLUSTER_NODE_CURRENT_IP=iggy-follower
@@ -188,15 +186,7 @@ services:
volumes:
- ./scenarios/basic_messaging.feature:/app/features/basic_messaging.feature
- ./scenarios/leader_redirection.feature:/app/features/leader_redirection.feature
- command:
- [
- "cargo",
- "test",
- "-p",
- "bdd",
- "--features",
- "bdd",
- ]
+ command: [ "cargo", "test", "-p", "bdd", "--features", "bdd" ]
networks:
- iggy-bdd-network
@@ -213,7 +203,7 @@ services:
volumes:
- ./scenarios/basic_messaging.feature:/app/features/basic_messaging.feature
working_dir: /app
- command: ["pytest", "tests/", "-v"]
+ command: [ "pytest", "tests/", "-v" ]
networks:
- iggy-bdd-network
@@ -229,7 +219,7 @@ services:
- IGGY_TCP_ADDRESS=iggy-server:8090
volumes:
- ./scenarios/basic_messaging.feature:/app/features/basic_messaging.feature
- command: ["go", "test", "-v", "./..."]
+ command: [ "go", "test", "-v", "./..." ]
networks:
- iggy-bdd-network
@@ -244,7 +234,7 @@ services:
- IGGY_ROOT_USERNAME=iggy
- IGGY_ROOT_PASSWORD=iggy
- IGGY_TCP_ADDRESS=iggy-server:8090
- command: ["npm", "run", "test:bdd"]
+ command: [ "npm", "run", "test:bdd" ]
networks:
- iggy-bdd-network
@@ -258,7 +248,7 @@ services:
- IGGY_ROOT_USERNAME=iggy
- IGGY_ROOT_PASSWORD=iggy
- IGGY_TCP_ADDRESS=iggy-server:8090
- command: ["dotnet", "test"]
+ command: [ "dotnet", "test" ]
networks:
- iggy-bdd-network
@@ -275,7 +265,7 @@ services:
- IGGY_TCP_ADDRESS=iggy-server:8090
volumes:
- ./scenarios/basic_messaging.feature:/app/features/basic_messaging.feature
- command: ["gradle", "--no-daemon", "test"]
+ command: [ "gradle", "--no-daemon", "test" ]
networks:
- iggy-bdd-network
diff --git a/core/ai/mcp/Cargo.toml b/core/ai/mcp/Cargo.toml
index d9c8b9ad9..8cb193aa7 100644
--- a/core/ai/mcp/Cargo.toml
+++ b/core/ai/mcp/Cargo.toml
@@ -29,12 +29,13 @@ readme = "README.md"
[dependencies]
axum = { workspace = true }
axum-server = { workspace = true }
+configs = { workspace = true }
+configs_derive = { workspace = true }
dirs = { workspace = true }
dotenvy = { workspace = true }
figlet-rs = { workspace = true }
figment = { workspace = true }
iggy = { workspace = true }
-iggy_common = { workspace = true }
opentelemetry = { workspace = true }
opentelemetry-appender-tracing = { workspace = true }
opentelemetry-otlp = { workspace = true }
diff --git a/core/ai/mcp/src/configs.rs b/core/ai/mcp/src/configs.rs
index 7ec2846c3..f745ef60c 100644
--- a/core/ai/mcp/src/configs.rs
+++ b/core/ai/mcp/src/configs.rs
@@ -17,48 +17,53 @@
* under the License.
*/
+use ::configs::{FileConfigProvider, TypedEnvProvider};
use axum::http::Method;
+use configs_derive::ConfigEnv;
use figment::{
Metadata, Profile, Provider,
providers::{Format, Toml},
value::Dict,
};
use iggy::prelude::{DEFAULT_ROOT_PASSWORD, DEFAULT_ROOT_USERNAME};
-use iggy_common::{CustomEnvProvider, FileConfigProvider};
use serde::{Deserialize, Serialize};
use std::fmt::{Display as FmtDisplay, Formatter};
use std::str::FromStr;
use strum::Display;
use tower_http::cors::{AllowOrigin, CorsLayer};
-#[derive(Debug, Clone, Deserialize, Serialize)]
+#[derive(Debug, Clone, Deserialize, Serialize, ConfigEnv)]
+#[config_env(prefix = "IGGY_MCP_", name = "iggy-mcp-server-config")]
#[serde(default)]
pub struct McpServerConfig {
pub http: HttpConfig,
pub iggy: IggyConfig,
pub permissions: PermissionsConfig,
+ #[config_env(leaf)]
pub transport: McpTransport,
pub telemetry: TelemetryConfig,
}
-#[derive(Debug, Clone, Serialize, Deserialize)]
+#[derive(Debug, Clone, Serialize, Deserialize, ConfigEnv)]
pub struct IggyConfig {
pub address: String,
pub username: String,
+ #[config_env(secret)]
pub password: String,
+ #[config_env(secret)]
pub token: String,
pub consumer: String,
pub tls: IggyTlsConfig,
}
-#[derive(Debug, Default, Clone, Serialize, Deserialize)]
+#[derive(Debug, Default, Clone, Serialize, Deserialize, ConfigEnv)]
pub struct IggyTlsConfig {
pub enabled: bool,
pub ca_file: String,
pub domain: Option<String>,
}
-#[derive(Debug, Clone, Deserialize, Serialize)]
+#[derive(Debug, Clone, Deserialize, Serialize, ConfigEnv)]
pub struct PermissionsConfig {
pub create: bool,
pub read: bool,
@@ -66,7 +71,7 @@ pub struct PermissionsConfig {
pub delete: bool,
}
-#[derive(Debug, Clone, Deserialize, Serialize)]
+#[derive(Debug, Clone, Deserialize, Serialize, ConfigEnv)]
pub struct HttpConfig {
pub address: String,
pub path: String,
@@ -74,14 +79,14 @@ pub struct HttpConfig {
pub tls: HttpTlsConfig,
}
-#[derive(Debug, Default, Deserialize, Serialize, Clone)]
+#[derive(Debug, Default, Deserialize, Serialize, Clone, ConfigEnv)]
pub struct HttpTlsConfig {
pub enabled: bool,
pub cert_file: String,
pub key_file: String,
}
-#[derive(Debug, Default, Deserialize, Serialize, Clone)]
+#[derive(Debug, Default, Deserialize, Serialize, Clone, ConfigEnv)]
pub struct HttpCorsConfig {
pub enabled: bool,
pub allowed_methods: Vec<String>,
@@ -103,7 +108,7 @@ pub enum McpTransport {
Stdio,
}
-#[derive(Debug, Deserialize, Serialize, Clone)]
+#[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)]
pub struct TelemetryConfig {
pub enabled: bool,
pub service_name: String,
@@ -132,8 +137,9 @@ impl FmtDisplay for TelemetryConfig {
}
}
-#[derive(Debug, Deserialize, Serialize, Clone)]
+#[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)]
pub struct TelemetryLogsConfig {
+ #[config_env(leaf)]
pub transport: TelemetryTransport,
pub endpoint: String,
}
@@ -157,8 +163,9 @@ impl FmtDisplay for TelemetryLogsConfig {
}
}
-#[derive(Debug, Deserialize, Serialize, Clone)]
+#[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)]
pub struct TelemetryTracesConfig {
+ #[config_env(leaf)]
pub transport: TelemetryTransport,
pub endpoint: String,
}
@@ -396,25 +403,23 @@ impl McpServerConfig {
#[derive(Debug, Clone)]
pub struct McpServerEnvProvider {
- provider: CustomEnvProvider<McpServerConfig>,
+ provider: TypedEnvProvider<McpServerConfig>,
}
impl Default for McpServerEnvProvider {
fn default() -> Self {
Self {
- provider: CustomEnvProvider::new("IGGY_MCP_", &[]),
+ provider: TypedEnvProvider::from_config(McpServerConfig::ENV_PREFIX),
}
}
}
impl Provider for McpServerEnvProvider {
fn metadata(&self) -> Metadata {
- Metadata::named("iggy-mcp-server-config")
+ Metadata::named(McpServerConfig::ENV_PROVIDER_NAME)
}
fn data(&self) -> Result<figment::value::Map<Profile, Dict>, figment::Error> {
- self.provider.deserialize().map_err(|_| {
- figment::Error::from("Cannot deserialize environment variables for MCP config")
- })
+ self.provider.data()
}
}
diff --git a/core/ai/mcp/src/main.rs b/core/ai/mcp/src/main.rs
index dbbd1bf4c..73135cf1c 100644
--- a/core/ai/mcp/src/main.rs
+++ b/core/ai/mcp/src/main.rs
@@ -17,12 +17,12 @@
* under the License.
*/
+use ::configs::ConfigProvider;
use configs::{McpServerConfig, McpTransport};
use dotenvy::dotenv;
use error::McpRuntimeError;
use figlet_rs::FIGfont;
use iggy::prelude::{Client, Identifier};
-use iggy_common::ConfigProvider;
use rmcp::{ServiceExt, model::ErrorData, transport::stdio};
use service::IggyService;
use std::{env, sync::Arc};
diff --git a/core/common/Cargo.toml b/core/common/Cargo.toml
index 52ea4c5ea..fbcb02cd7 100644
--- a/core/common/Cargo.toml
+++ b/core/common/Cargo.toml
@@ -48,11 +48,11 @@ compio = { workspace = true }
compio-quic = { workspace = true }
compio-tls = { workspace = true }
compio-ws = { workspace = true }
+configs_derive = { workspace = true }
crossbeam = { workspace = true }
derive_more = { workspace = true }
err_trail = { workspace = true }
fast-async-mutex = { version = "0.6.7", optional = true }
-figment = { workspace = true }
human-repr = { workspace = true }
humantime = { workspace = true }
nix = { workspace = true }
@@ -66,7 +66,6 @@ serde_with = { workspace = true, features = ["base64"] }
strum = { workspace = true }
thiserror = { workspace = true }
tokio = { workspace = true }
-toml = { workspace = true }
tracing = { workspace = true }
tungstenite = { workspace = true }
twox-hash = { workspace = true }
diff --git a/core/common/src/configs/mod.rs b/core/common/src/configs/mod.rs
deleted file mode 100644
index fc440820f..000000000
--- a/core/common/src/configs/mod.rs
+++ /dev/null
@@ -1,893 +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.
- */
-
-//! Configuration module providing flexible configuration loading with file and environment support.
-//!
-//! This module provides a trait-based configuration system that supports:
-//! - Loading configuration from TOML files
-//! - Environment variable overrides with complex path resolution
-//! - Array and HashMap field overrides
-//! - JSON value field handling
-//! - Automatic type conversion and validation
-
-use figment::{
- Figment, Profile, Provider,
- providers::{Data, Format, Toml},
- value::{Dict, Map as FigmentMap, Tag, Value as FigmentValue},
-};
-use serde::{Serialize, de::DeserializeOwned};
-use std::{env, fmt::Display, future::Future, marker::PhantomData, path::Path};
-use toml::{Value as TomlValue, map::Map as TomlMap};
-use tracing::{error, info, warn};
-
-const SECRET_MASK: &str = "******";
-const ARRAY_SEPARATOR: char = '_';
-const PATH_SEPARATOR: &str = ".";
-const DISPLAY_CONFIG_ENV: &str = "IGGY_DISPLAY_CONFIG";
-
-type ProfileMap = FigmentMap<Profile, Dict>;
-
-/// Type alias for configuration that can be serialized, deserialized, has defaults, and can be displayed
-pub trait ConfigurationType: Serialize + DeserializeOwned + Default + Display {}
-impl<T: Serialize + DeserializeOwned + Default + Display> ConfigurationType for T {}
-
-#[derive(Debug, Clone, PartialEq, Eq)]
-pub enum ConfigurationError {
- CannotLoadConfiguration,
- DefaultSerializationFailed,
- DefaultParsingFailed,
- EnvironmentVariableParsingFailed,
- InvalidConfigurationValue,
-}
-
-impl Display for ConfigurationError {
- fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
- match self {
- Self::CannotLoadConfiguration => write!(f, "Cannot load configuration"),
- Self::DefaultSerializationFailed => {
- write!(f, "Failed to serialize default configuration")
- }
- Self::DefaultParsingFailed => write!(f, "Failed to parse default configuration"),
- Self::EnvironmentVariableParsingFailed => {
- write!(f, "Failed to parse environment variables")
- }
- Self::InvalidConfigurationValue => {
- write!(f, "Provided configuration value is invalid")
- }
- }
- }
-}
-
-impl std::error::Error for ConfigurationError {}
-
-pub trait ConfigProvider<T: ConfigurationType> {
- fn load_config(&self) -> impl Future<Output = Result<T, ConfigurationError>>;
-}
-
-/// File-based configuration provider that combines file, default, and environment configurations
-pub struct FileConfigProvider<P> {
- file_path: String,
- default_config: Option<Data<Toml>>,
- env_provider: P,
- display_config: bool,
-}
-
-impl<P: Provider> FileConfigProvider<P> {
- /// Create a new file configuration provider
- ///
- /// # Arguments
- /// * `file_path` - Path to the configuration file
- /// * `env_provider` - Environment variable provider
- /// * `display_config` - Whether to display the loaded configuration
- /// * `default_config` - Optional default configuration data
- pub fn new(
- file_path: String,
- env_provider: P,
- display_config: bool,
- default_config: Option<Data<Toml>>,
- ) -> Self {
- Self {
- file_path,
- env_provider,
- default_config,
- display_config,
- }
- }
-}
-
-#[derive(Debug, Clone)]
-pub struct CustomEnvProvider<T: ConfigurationType> {
- prefix: String,
- secret_keys: Vec<String>,
- _phantom: PhantomData<T>,
-}
-
-impl<T: ConfigurationType> CustomEnvProvider<T> {
- /// Create a new custom environment provider
- ///
- /// # Arguments
- /// * `prefix` - Environment variable prefix to filter by
- /// * `secret_keys` - Keys that should be masked in logs
- pub fn new(prefix: &str, secret_keys: &[&str]) -> Self {
- Self {
- prefix: prefix.to_string(),
- secret_keys: secret_keys.iter().map(|s| s.to_string()).collect(),
- _phantom: PhantomData,
- }
- }
-
- /// Deserialize environment variables into a configuration profile map
- pub fn deserialize(&self) -> Result<ProfileMap, ConfigurationError> {
- let default_config = toml::to_string(&T::default())
- .map_err(|_| ConfigurationError::DefaultSerializationFailed)?;
-
- let toml_value: TomlValue = toml::from_str(&default_config)
- .map_err(|_| ConfigurationError::DefaultParsingFailed)?;
-
- let mut source_dict = Dict::new();
- if let TomlValue::Table(table) = toml_value {
- Self::walk_toml_table_to_dict("", table, &mut source_dict);
- }
-
- let mut target_dict = Dict::new();
- for (key, mut value) in env::vars() {
- let env_key = key.to_uppercase();
- if !env_key.starts_with(&self.prefix) {
- continue;
- }
-
- let keys: Vec<String> = env_key[self.prefix.len()..]
- .split(ARRAY_SEPARATOR)
- .map(|k| k.to_lowercase())
- .collect();
-
- let expected_type = Self::find_expected_type(&source_dict, &keys);
- let env_var_value = Self::parse_environment_value_typed(&value, expected_type);
-
- if self.secret_keys.contains(&env_key) {
- value = SECRET_MASK.to_string();
- }
-
- info!("{env_key} value changed to: {value} from environment variable");
- Self::insert_environment_override(&source_dict, &mut target_dict, keys, env_var_value);
- }
-
- let mut data = ProfileMap::new();
- data.insert(Profile::default(), target_dict);
- Ok(data)
- }
-
- /// Walk through TOML table and convert to dictionary structure
- fn walk_toml_table_to_dict(prefix: &str, table: TomlMap<String, TomlValue>, dict: &mut Dict) {
- for (key, value) in table {
- let new_prefix = if prefix.is_empty() {
- key.clone()
- } else {
- format!("{prefix}{PATH_SEPARATOR}{key}")
- };
-
- match value {
- TomlValue::Table(inner_table) => {
- let mut nested_dict = Dict::new();
- Self::walk_toml_table_to_dict(&new_prefix, inner_table, &mut nested_dict);
- dict.insert(key, FigmentValue::from(nested_dict));
- }
- _ => {
- dict.insert(key, Self::toml_to_figment_value(&value));
- }
- }
- }
- }
-
- /// Insert environment variable override into target dictionary
- fn insert_environment_override(
- source: &Dict,
- target: &mut Dict,
- keys: Vec<String>,
- value: FigmentValue,
- ) {
- if keys.is_empty() {
- return;
- }
-
- // Detect array patterns (e.g., cluster.nodes.0.id)
- if let Some((idx_pos, array_index)) = Self::find_array_index(&keys) {
- Self::handle_array_override(source, target, &keys, idx_pos, array_index, value);
- } else {
- Self::handle_dictionary_override(source, target, &keys, value);
- }
- }
-
- /// Find array index in key path
- fn find_array_index(keys: &[String]) -> Option<(usize, usize)> {
- keys.iter()
- .enumerate()
- .find_map(|(i, key)| key.parse::<usize>().ok().map(|idx| (i, idx)))
- }
-
- /// Handle array field override with complex path resolution
- fn handle_array_override(
- source: &Dict,
- target: &mut Dict,
- keys: &[String],
- idx_pos: usize,
- array_index: usize,
- value: FigmentValue,
- ) {
- let path_to_array = &keys[..idx_pos];
- let remaining_path = &keys[idx_pos + 1..];
-
- // Navigate to array container
- let current_target = match Self::navigate_to_dictionary(
- target,
- &path_to_array[..path_to_array.len().saturating_sub(1)],
- ) {
- Some(dict) => dict,
- None => return,
- };
-
- let array_key = match path_to_array.last() {
- Some(key) => key.clone(),
- None => return,
- };
-
- // Copy existing array from source if needed
- if !current_target.contains_key(&array_key) {
- if let Some(existing_array) = Self::find_source_array(source, path_to_array, &array_key)
- {
- current_target.insert(array_key.clone(), existing_array);
- } else {
- current_target.insert(
- array_key.clone(),
- FigmentValue::Array(Tag::Default, Vec::new()),
- );
- }
- }
-
- // Update array element
- if let Some(FigmentValue::Array(_, arr)) = current_target.get_mut(&array_key) {
- // Ensure array is large enough
- while arr.len() <= array_index {
- arr.push(FigmentValue::Dict(Tag::Default, Dict::new()));
- }
-
- if remaining_path.is_empty() {
- arr[array_index] = value;
- } else if let FigmentValue::Dict(_, elem_dict) = &mut arr[array_index] {
- // TODO(hubcio): this is workaround done by Claude because this code is overly
- // complicated and I don't want to spend time on it.
- // For nested structures in arrays, check if we need to create intermediate dicts
- // Handle the "ports" case where it should be a nested structure
- if remaining_path.len() >= 2 && remaining_path[0] == "ports" {
- // Create the ports dict if it doesn't exist
- elem_dict
- .entry("ports".to_string())
- .or_insert_with(|| FigmentValue::Dict(Tag::Default, Dict::new()));
-
- if let Some(FigmentValue::Dict(_, ports_dict)) = elem_dict.get_mut("ports") {
- // Insert the specific port value (tcp, quic, http, websocket)
- ports_dict.insert(remaining_path[1].clone(), value);
- }
- } else {
- // Default behavior for other fields
- Self::insert_environment_override(
- &Dict::new(),
- elem_dict,
- remaining_path.to_vec(),
- value,
- );
- }
- }
- }
- }
-
- /// Handle dictionary field override with HashMap detection
- fn handle_dictionary_override(
- source: &Dict,
- target: &mut Dict,
- keys: &[String],
- value: FigmentValue,
- ) {
- if keys.is_empty() {
- return;
- }
-
- // Try to detect HashMap patterns
- if let Some(hashmap_result) =
- Self::try_handle_hashmap_override(source, target, keys, value.clone())
- && hashmap_result
- {
- return;
- }
-
- // Fallback to original logic for non-HashMap patterns
- let mut current_source = source;
- let mut current_target = target;
- let mut combined_keys = Vec::new();
-
- for (i, key) in keys.iter().enumerate() {
- combined_keys.push(key.clone());
- let key_to_check = combined_keys.join("_");
-
- match current_source.get(&key_to_check) {
- Some(FigmentValue::Dict(_, inner_source_dict)) => {
- current_target
- .entry(key_to_check.clone())
- .or_insert_with(|| FigmentValue::Dict(Tag::Default, Dict::new()));
-
- if let Some(FigmentValue::Dict(_, inner_target)) =
- current_target.get_mut(&key_to_check)
- {
- current_source = inner_source_dict;
- current_target = inner_target;
- combined_keys.clear();
- } else {
- return;
- }
- }
- Some(_) => {
- current_target.insert(key_to_check, value);
- return;
- }
- None if i == keys.len() - 1 => {
- current_target.insert(key_to_check, value);
- return;
- }
- _ => continue,
- }
- }
- }
-
- /// Try to handle HashMap override patterns
- fn try_handle_hashmap_override(
- source: &Dict,
- target: &mut Dict,
- keys: &[String],
- value: FigmentValue,
- ) -> Option<bool> {
- if keys.len() < 2 {
- return Some(false);
- }
-
- let potential_hashmap_key = &keys[0];
-
- // Check if this is actually a HashMap field in the source
- let source_hashmap = match source.get(potential_hashmap_key) {
- Some(FigmentValue::Dict(_, hashmap_dict)) => {
- // Determine if this is a HashMap or regular struct
- let has_dict_values = hashmap_dict
- .values()
- .any(|v| matches!(v, FigmentValue::Dict(_, _)));
- let has_non_dict_values = hashmap_dict
- .values()
- .any(|v| !matches!(v, FigmentValue::Dict(_, _)));
-
- // If it has non-Dict values mixed with Dict values, it's probably a regular struct
- if has_non_dict_values && has_dict_values {
- return Some(false);
- }
-
- // If it only has non-Dict values, it's definitely a regular struct
- if has_non_dict_values && !has_dict_values {
- return Some(false);
- }
-
- hashmap_dict
- }
- _ => return Some(false),
- };
-
- // Try to find the best HashMap entry key
- if let Some((entry_key, remaining_keys)) =
- Self::find_optimal_hashmap_split(source_hashmap, &keys[1..])
- {
- return Some(Self::apply_hashmap_override(
- target,
- potential_hashmap_key,
- &entry_key,
- &remaining_keys,
- source_hashmap,
- value,
- ));
- }
-
- Some(false)
- }
-
- /// Find optimal HashMap key split for environment variable path
- fn find_optimal_hashmap_split(
- source_hashmap: &Dict,
- keys: &[String],
- ) -> Option<(String, Vec<String>)> {
- if keys.is_empty() {
- return None;
- }
-
- // First, try existing HashMap entry keys if any exist
- if !source_hashmap.is_empty() {
- for (existing_key, existing_value) in source_hashmap {
- if let FigmentValue::Dict(_, entry_dict) = existing_value {
- // Try different ways to match this existing key
- for split_point in 1..=keys.len() {
- let candidate_key = keys[0..split_point].join("_");
-
- if candidate_key == *existing_key {
- let remaining_keys = keys[split_point..].to_vec();
-
- // Validate that the remaining keys form a valid path in the entry
- if remaining_keys.is_empty()
- || Self::is_valid_field_path(entry_dict, &remaining_keys)
- {
- return Some((existing_key.clone(), remaining_keys));
- }
- }
- }
- }
- }
- }
-
- // For empty HashMaps or when no existing keys match, use simple split
- let simple_entry_key = keys[0].clone();
- let simple_remaining = keys[1..].to_vec();
- Some((simple_entry_key, simple_remaining))
- }
-
- /// Check if a field path is valid within a dictionary structure
- fn is_valid_field_path(dict: &Dict, keys: &[String]) -> bool {
- if keys.is_empty() {
- return true;
- }
-
- // Try different combinations to account for fields with underscores
- for split_point in 1..=keys.len() {
- let field_name = keys[0..split_point].join("_");
- let remaining_keys = &keys[split_point..];
-
- if let Some(field_value) = dict.get(&field_name) {
- if remaining_keys.is_empty() {
- return true;
- } else if let FigmentValue::Dict(_, nested_dict) = field_value {
- return Self::is_valid_field_path(nested_dict, remaining_keys);
- }
- }
- }
-
- // Also try the original approach with combined keys
- let mut current_dict = dict;
- let mut combined_keys = Vec::new();
-
- for key in keys {
- combined_keys.push(key.clone());
- let combined_field = combined_keys.join("_");
-
- if let Some(field_value) = current_dict.get(&combined_field) {
- match field_value {
- FigmentValue::Dict(_, nested_dict) => {
- current_dict = nested_dict;
- combined_keys.clear();
- }
- _ => return true,
- }
- }
- }
-
- true
- }
-
- /// Apply HashMap override to target configuration
- fn apply_hashmap_override(
- target: &mut Dict,
- hashmap_key: &str,
- entry_key: &str,
- remaining_keys: &[String],
- source_hashmap: &Dict,
- value: FigmentValue,
- ) -> bool {
- // Ensure the HashMap exists in target
- target
- .entry(hashmap_key.to_string())
- .or_insert_with(|| FigmentValue::Dict(Tag::Default, Dict::new()));
-
- if let Some(FigmentValue::Dict(_, target_hashmap)) = target.get_mut(hashmap_key) {
- // Ensure the specific entry exists in the HashMap
- target_hashmap
- .entry(entry_key.to_string())
- .or_insert_with(|| {
- if let Some(existing_entry) = source_hashmap.get(entry_key) {
- existing_entry.clone()
- } else {
- FigmentValue::Dict(Tag::Default, Dict::new())
- }
- });
-
- if remaining_keys.is_empty() {
- target_hashmap.insert(entry_key.to_string(), value);
- } else if let Some(FigmentValue::Dict(_, entry_dict)) =
- target_hashmap.get_mut(entry_key)
- {
- // Check if we need special handling for JSON value fields
- if let Some((json_field, json_keys)) =
- Self::find_json_value_field_split(entry_dict, remaining_keys)
- {
- Self::handle_json_value_override(entry_dict, &json_field, json_keys, value);
- } else {
- Self::insert_environment_override(
- &Dict::new(),
- entry_dict,
- remaining_keys.to_vec(),
- value,
- );
- }
- }
- return true;
- }
-
- false
- }
-
- /// Find JSON value field split in remaining keys
- fn find_json_value_field_split(dict: &Dict, keys: &[String]) -> Option<(String, Vec<String>)> {
- if keys.is_empty() {
- return None;
- }
-
- // Try different split points to find a field that exists
- for split_point in 1..=keys.len() {
- let potential_json_field = keys[0..split_point].join("_");
- let remaining_keys = keys[split_point..].to_vec();
-
- if dict.contains_key(&potential_json_field) && !remaining_keys.is_empty() {
- return Some((potential_json_field, remaining_keys));
- }
- }
-
- // If no existing field matches and we have multiple keys, use first key
- if keys.len() > 1 {
- let potential_new_field = keys[0].clone();
- let remaining_keys = keys[1..].to_vec();
- return Some((potential_new_field, remaining_keys));
- }
-
- None
- }
-
- /// Handle JSON value field override
- fn handle_json_value_override(
- entry_dict: &mut Dict,
- json_field: &str,
- json_keys: Vec<String>,
- value: FigmentValue,
- ) {
- // Ensure the JSON value field exists
- entry_dict
- .entry(json_field.to_string())
- .or_insert_with(|| FigmentValue::Dict(Tag::Default, Dict::new()));
-
- // Get or create the JSON field as a Dict
- if let Some(json_value) = entry_dict.get_mut(json_field) {
- let json_dict = match json_value {
- FigmentValue::Dict(_, dict) => dict,
- _ => {
- *json_value = FigmentValue::Dict(Tag::Default, Dict::new());
- if let FigmentValue::Dict(_, dict) = json_value {
- dict
- } else {
- return;
- }
- }
- };
-
- Self::set_nested_json_field(json_dict, &json_keys, value);
- }
- }
-
- /// Set nested JSON field value
- fn set_nested_json_field(dict: &mut Dict, keys: &[String], value: FigmentValue) {
- if keys.is_empty() {
- return;
- }
-
- // Always try the full field name first (with underscores)
- let full_field_name = keys.join("_");
- dict.insert(full_field_name, value);
- }
-
- /// Navigate to a specific dictionary in the nested structure
- fn navigate_to_dictionary<'a>(target: &'a mut Dict, path: &[String]) -> Option<&'a mut Dict> {
- if path.is_empty() {
- return Some(target);
- }
-
- let mut current = target;
- let mut combined_keys = Vec::new();
-
- for key in path {
- combined_keys.push(key.clone());
- let key_to_check = combined_keys.join("_");
-
- current
- .entry(key_to_check.clone())
- .or_insert_with(|| FigmentValue::Dict(Tag::Default, Dict::new()));
-
- match current.get_mut(&key_to_check) {
- Some(FigmentValue::Dict(_, inner)) => {
- current = inner;
- combined_keys.clear();
- }
- _ => return None,
- }
- }
-
- Some(current)
- }
-
- /// Find source array value for copying to target
- fn find_source_array(source: &Dict, path: &[String], array_key: &str) -> Option<FigmentValue> {
- if path.is_empty() {
- return None;
- }
-
- let mut current = source;
- let mut combined_keys = Vec::new();
-
- for key in &path[..path.len() - 1] {
- combined_keys.push(key.clone());
- let key_to_check = combined_keys.join("_");
-
- match current.get(&key_to_check) {
- Some(FigmentValue::Dict(_, inner)) => {
- current = inner;
- combined_keys.clear();
- }
- _ => return None,
- }
- }
-
- current.get(array_key).cloned()
- }
-
- /// Convert TOML value to Figment value
- fn toml_to_figment_value(toml_value: &TomlValue) -> FigmentValue {
- match toml_value {
- TomlValue::String(s) => FigmentValue::from(s.clone()),
- TomlValue::Integer(i) => FigmentValue::from(*i),
- TomlValue::Float(f) => FigmentValue::from(*f),
- TomlValue::Boolean(b) => FigmentValue::from(*b),
- TomlValue::Array(arr) => {
- let vec: Vec<FigmentValue> = arr.iter().map(Self::toml_to_figment_value).collect();
- FigmentValue::from(vec)
- }
- TomlValue::Table(tbl) => {
- let mut dict = Dict::new();
- for (key, value) in tbl.iter() {
- dict.insert(key.clone(), Self::toml_to_figment_value(value));
- }
- FigmentValue::from(dict)
- }
- TomlValue::Datetime(_) => {
- // For now, convert datetime to string representation
- FigmentValue::from(toml_value.to_string())
- }
- }
- }
-
- fn parse_environment_value_typed(
- value: &str,
- expected_type: Option<&FigmentValue>,
- ) -> FigmentValue {
- if value.starts_with('[') && value.ends_with(']') {
- let inner_value = value.trim_start_matches('[').trim_end_matches(']');
- let element_type = match expected_type {
- Some(FigmentValue::Array(_, arr)) => arr.first(),
- _ => None,
- };
- let values: Vec<FigmentValue> = split_array_elements(inner_value)
- .into_iter()
- .map(|s| Self::parse_environment_value_typed(strip_quotes(s), element_type))
- .collect();
- return FigmentValue::from(values);
- }
-
- if let Some(FigmentValue::String(..)) = expected_type {
- return FigmentValue::from(value);
- }
-
- match value.to_lowercase().as_str() {
- "true" => return FigmentValue::from(true),
- "false" => return FigmentValue::from(false),
- _ => {}
- }
-
- if let Ok(uint_val) = value.parse::<u64>() {
- return FigmentValue::from(uint_val);
- }
- if let Ok(int_val) = value.parse::<i64>() {
- return FigmentValue::from(int_val);
- }
-
- if let Ok(float_val) = value.parse::<f64>() {
- return FigmentValue::from(float_val);
- }
-
- FigmentValue::from(value)
- }
-
- fn find_expected_type<'a>(source: &'a Dict, keys: &[String]) -> Option<&'a FigmentValue> {
- if keys.is_empty() {
- return None;
- }
-
- let mut current = source;
- let mut combined_keys = Vec::new();
-
- for (i, key) in keys.iter().enumerate() {
- combined_keys.push(key.clone());
- let key_to_check = combined_keys.join("_");
-
- match current.get(&key_to_check) {
- Some(FigmentValue::Dict(_, inner)) => {
- current = inner;
- combined_keys.clear();
- }
- Some(value) if i == keys.len() - 1 => {
- return Some(value);
- }
- _ => continue,
- }
- }
-
- let final_key = combined_keys.join("_");
- current.get(&final_key)
- }
-}
-
-fn split_array_elements(s: &str) -> Vec<&str> {
- let mut elements = Vec::new();
- let mut start = 0;
- let mut in_quotes = false;
- for (i, c) in s.char_indices() {
- match c {
- '"' => in_quotes = !in_quotes,
- ',' if !in_quotes => {
- elements.push(s[start..i].trim());
- start = i + 1;
- }
- _ => {}
- }
- }
- if start < s.len() {
- elements.push(s[start..].trim());
- }
- elements
-}
-
-fn strip_quotes(s: &str) -> &str {
- s.trim_matches('"')
-}
-
-pub fn parse_env_value_to_json(value: &str) -> serde_json::Value {
- if value.starts_with('[') && value.ends_with(']') {
- let inner = value.trim_start_matches('[').trim_end_matches(']');
- let values: Vec<serde_json::Value> = split_array_elements(inner)
- .into_iter()
- .map(|s| parse_env_value_to_json(strip_quotes(s)))
- .collect();
- return serde_json::Value::Array(values);
- }
-
- match value.to_lowercase().as_str() {
- "true" => return serde_json::Value::Bool(true),
- "false" => return serde_json::Value::Bool(false),
- _ => {}
- }
-
- if let Ok(n) = value.parse::<u64>() {
- return serde_json::Value::Number(n.into());
- }
- if let Ok(n) = value.parse::<i64>() {
- return serde_json::Value::Number(n.into());
- }
- if let Ok(n) = value.parse::<f64>()
- && let Some(num) = serde_json::Number::from_f64(n)
- {
- return serde_json::Value::Number(num);
- }
-
- serde_json::Value::String(value.to_owned())
-}
-
-fn file_exists<P: AsRef<Path>>(path: P) -> bool {
- let path = path.as_ref();
-
- if path.is_absolute() {
- return path.is_file();
- }
-
- let cwd = match std::env::current_dir() {
- Ok(dir) => dir,
- Err(_) => return false,
- };
-
- let mut current_dir = cwd.as_path();
- loop {
- let file_path = current_dir.join(path);
- if file_path.is_file() {
- return true;
- }
-
- current_dir = match current_dir.parent() {
- Some(parent) => parent,
- None => return false,
- };
- }
-}
-
-impl<T: ConfigurationType, P: Provider + Clone> ConfigProvider<T> for FileConfigProvider<P> {
- async fn load_config(&self) -> Result<T, ConfigurationError> {
- info!("Loading config from path: '{}'...", self.file_path);
-
- // Start with the default configuration if provided
- let mut config_builder = Figment::new();
- let has_default = self.default_config.is_some();
- if let Some(default) = &self.default_config {
- config_builder = config_builder.merge(default);
- } else {
- warn!("No default configuration provided.");
- }
-
- // If the config file exists, merge it into the configuration
- if file_exists(&self.file_path) {
- info!("Found configuration file at path: '{}'.", self.file_path);
- config_builder = config_builder.merge(Toml::file(&self.file_path));
- } else {
- warn!(
- "Configuration file not found at path: '{}'.",
- self.file_path
- );
- if has_default {
- info!(
- "Using default configuration embedded into server, as no config file was found."
- );
- }
- }
-
- // Merge environment variables into the configuration
- config_builder = config_builder.merge(self.env_provider.clone());
-
- // Finally, attempt to extract the final configuration
- let config_result: Result<T, figment::Error> = config_builder.extract();
-
- match config_result {
- Ok(config) => {
- info!("Config loaded successfully.");
- let display_config = env::var(DISPLAY_CONFIG_ENV)
- .map(|val| val == "1" || val.to_lowercase() == "true")
- .unwrap_or(self.display_config);
- if display_config {
- info!("Using Config: {config}");
- }
- Ok(config)
- }
- Err(e) => {
- error!("Failed to load config: {e}");
- Err(ConfigurationError::CannotLoadConfiguration)
- }
- }
- }
-}
diff --git a/core/common/src/lib.rs b/core/common/src/lib.rs
index 9171c4097..f5a955ac2 100644
--- a/core/common/src/lib.rs
+++ b/core/common/src/lib.rs
@@ -18,7 +18,6 @@
mod alloc;
mod certificates;
mod commands;
-mod configs;
mod error;
mod sender;
pub mod sharding;
@@ -44,7 +43,7 @@ pub use commands::system::get_cluster_metadata::*;
pub use commands::system::*;
pub use commands::topics::*;
pub use commands::users::*;
-pub use configs::*;
+pub use configs_derive::ConfigEnv;
pub use sender::{
QuicSender, Sender, SenderKind, TcpSender, TcpTlsSender, WebSocketSender, WebSocketTlsSender,
};
diff --git a/core/configs/Cargo.toml b/core/configs/Cargo.toml
new file mode 100644
index 000000000..d03f44c04
--- /dev/null
+++ b/core/configs/Cargo.toml
@@ -0,0 +1,30 @@
+# 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.
+
+[package]
+name = "configs"
+version = "0.1.0"
+edition = "2024"
+license = "Apache-2.0"
+publish = false
+
+[dependencies]
+configs_derive = { workspace = true }
+figment = { workspace = true }
+serde = { workspace = true }
+serde_json = { workspace = true }
+tracing = { workspace = true }
diff --git a/core/configs/src/configs_impl/env_mapping.rs b/core/configs/src/configs_impl/env_mapping.rs
new file mode 100644
index 000000000..6cb93f7d6
--- /dev/null
+++ b/core/configs/src/configs_impl/env_mapping.rs
@@ -0,0 +1,63 @@
+/*
+ * 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.
+ */
+
+//! Environment variable to config path mapping types.
+
+/// Represents a single environment variable to config path mapping.
+/// Used by the `ConfigEnv` derive macro to generate compile-time mappings.
+#[derive(Debug, Clone, Copy)]
+pub struct EnvVarMapping {
+ /// The environment variable name (e.g., "IGGY_HTTP_ENABLED")
+ pub env_name: &'static str,
+ /// The config path (e.g., "http.enabled")
+ pub config_path: &'static str,
+ /// Whether this field contains secret data
+ pub is_secret: bool,
+}
+
+/// Trait for configuration types that provide environment variable mappings.
+/// Implemented automatically by the `#[derive(ConfigEnv)]` macro.
+pub trait ConfigEnvMappings {
+ /// Returns all environment variable mappings for this config type.
+ fn env_mappings() -> &'static [EnvVarMapping];
+
+ /// Finds a mapping by environment variable name.
+ fn find_by_env_name(env_name: &str) -> Option<&'static EnvVarMapping> {
+ Self::env_mappings().iter().find(|m| m.env_name == env_name)
+ }
+
+ /// Finds a mapping by config path.
+ fn find_by_config_path(path: &str) -> Option<&'static EnvVarMapping> {
+ Self::env_mappings().iter().find(|m| m.config_path == path)
+ }
+
+ /// Returns all valid environment variable names.
+ fn all_env_var_names() -> Vec<&'static str> {
+ Self::env_mappings().iter().map(|m| m.env_name).collect()
+ }
+
+ /// Returns environment variable names marked as secrets.
+ fn secret_env_names() -> Vec<&'static str> {
+ Self::env_mappings()
+ .iter()
+ .filter(|m| m.is_secret)
+ .map(|m| m.env_name)
+ .collect()
+ }
+}
diff --git a/core/configs/src/configs_impl/error.rs b/core/configs/src/configs_impl/error.rs
new file mode 100644
index 000000000..1b36e89b1
--- /dev/null
+++ b/core/configs/src/configs_impl/error.rs
@@ -0,0 +1,51 @@
+/*
+ * 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.
+ */
+
+//! Configuration error types.
+
+use std::fmt::Display;
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum ConfigurationError {
+ CannotLoadConfiguration,
+ DefaultSerializationFailed,
+ DefaultParsingFailed,
+ EnvironmentVariableParsingFailed,
+ InvalidConfigurationValue,
+}
+
+impl Display for ConfigurationError {
+ fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ match self {
+ Self::CannotLoadConfiguration => write!(f, "Cannot load configuration"),
+ Self::DefaultSerializationFailed => {
+ write!(f, "Failed to serialize default configuration")
+ }
+ Self::DefaultParsingFailed => write!(f, "Failed to parse default configuration"),
+ Self::EnvironmentVariableParsingFailed => {
+ write!(f, "Failed to parse environment variables")
+ }
+ Self::InvalidConfigurationValue => {
+ write!(f, "Provided configuration value is invalid")
+ }
+ }
+ }
+}
+
+impl std::error::Error for ConfigurationError {}
diff --git a/core/configs/src/configs_impl/file_provider.rs b/core/configs/src/configs_impl/file_provider.rs
new file mode 100644
index 000000000..87463e38c
--- /dev/null
+++ b/core/configs/src/configs_impl/file_provider.rs
@@ -0,0 +1,142 @@
+/*
+ * 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.
+ */
+
+//! File-based configuration provider.
+
+use super::error::ConfigurationError;
+use super::traits::{ConfigProvider, ConfigurationType};
+use figment::{
+ Figment, Provider,
+ providers::{Data, Format, Toml},
+};
+use std::{env, path::Path};
+use tracing::{error, info, warn};
+
+const DISPLAY_CONFIG_ENV: &str = "IGGY_DISPLAY_CONFIG";
+
+/// File-based configuration provider that combines file, default, and environment configurations.
+pub struct FileConfigProvider<P> {
+ file_path: String,
+ default_config: Option<Data<Toml>>,
+ env_provider: P,
+ display_config: bool,
+}
+
+impl<P: Provider> FileConfigProvider<P> {
+ /// Create a new file configuration provider.
+ ///
+ /// # Arguments
+ /// * `file_path` - Path to the configuration file
+ /// * `env_provider` - Environment variable provider
+ /// * `display_config` - Whether to display the loaded configuration
+ /// * `default_config` - Optional default configuration data
+ pub fn new(
+ file_path: String,
+ env_provider: P,
+ display_config: bool,
+ default_config: Option<Data<Toml>>,
+ ) -> Self {
+ Self {
+ file_path,
+ env_provider,
+ default_config,
+ display_config,
+ }
+ }
+}
+
+impl<P: Provider + Clone> ConfigProvider for FileConfigProvider<P> {
+ async fn load_config<T: ConfigurationType>(&self) -> Result<T, ConfigurationError> {
+ info!("Loading config from path: '{}'...", self.file_path);
+
+ // Start with the default configuration if provided
+ let mut config_builder = Figment::new();
+ let has_default = self.default_config.is_some();
+ if let Some(default) = &self.default_config {
+ config_builder = config_builder.merge(default);
+ } else {
+ warn!("No default configuration provided.");
+ }
+
+ // If the config file exists, merge it into the configuration
+ if file_exists(&self.file_path) {
+ info!("Found configuration file at path: '{}'.", self.file_path);
+ config_builder = config_builder.merge(Toml::file(&self.file_path));
+ } else {
+ warn!(
+ "Configuration file not found at path: '{}'.",
+ self.file_path
+ );
+ if has_default {
+ info!(
+ "Using default configuration embedded into server, as no config file was found."
+ );
+ }
+ }
+
+ // Merge environment variables into the configuration
+ config_builder = config_builder.merge(self.env_provider.clone());
+
+ // Finally, attempt to extract the final configuration
+ let config_result: Result<T, figment::Error> = config_builder.extract();
+
+ match config_result {
+ Ok(config) => {
+ info!("Config loaded successfully.");
+ let display_config = env::var(DISPLAY_CONFIG_ENV)
+ .map(|val| val == "1" || val.to_lowercase() == "true")
+ .unwrap_or(self.display_config);
+ if display_config {
+ info!("Using Config: {config}");
+ }
+ Ok(config)
+ }
+ Err(e) => {
+ error!("Failed to load config: {e}");
+ Err(ConfigurationError::CannotLoadConfiguration)
+ }
+ }
+ }
+}
+
+fn file_exists<P: AsRef<Path>>(path: P) -> bool {
+ let path = path.as_ref();
+
+ if path.is_absolute() {
+ return path.is_file();
+ }
+
+ let cwd = match std::env::current_dir() {
+ Ok(dir) => dir,
+ Err(_) => return false,
+ };
+
+ let mut current_dir = cwd.as_path();
+ loop {
+ let file_path = current_dir.join(path);
+ if file_path.is_file() {
+ return true;
+ }
+
+ current_dir = match current_dir.parent() {
+ Some(parent) => parent,
+ None => return false,
+ };
+ }
+}
diff --git a/core/configs/src/configs_impl/mod.rs b/core/configs/src/configs_impl/mod.rs
new file mode 100644
index 000000000..7c2f23738
--- /dev/null
+++ b/core/configs/src/configs_impl/mod.rs
@@ -0,0 +1,41 @@
+/*
+ * 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.
+ */
+
+//! Configuration module providing flexible configuration loading with file and environment support.
+//!
+//! This module provides a trait-based configuration system that supports:
+//! - Loading configuration from TOML files
+//! - Environment variable overrides with complex path resolution
+//! - Array and HashMap field overrides
+//! - JSON value field handling
+//! - Automatic type conversion and validation
+
+mod env_mapping;
+mod error;
+mod file_provider;
+mod parsing;
+mod traits;
+mod typed_env_provider;
+
+pub use env_mapping::{ConfigEnvMappings, EnvVarMapping};
+pub use error::ConfigurationError;
+pub use file_provider::FileConfigProvider;
+pub use parsing::parse_env_value_to_json;
+pub use traits::{ConfigProvider, ConfigurationType};
+pub use typed_env_provider::TypedEnvProvider;
diff --git a/core/configs/src/configs_impl/parsing.rs b/core/configs/src/configs_impl/parsing.rs
new file mode 100644
index 000000000..d4675d7d3
--- /dev/null
+++ b/core/configs/src/configs_impl/parsing.rs
@@ -0,0 +1,196 @@
+/*
+ * 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.
+ */
+
+//! Environment variable value parsing utilities.
+
+use figment::value::Value as FigmentValue;
+
+/// Parse an environment variable value into a FigmentValue with type inference.
+pub(crate) fn parse_env_value(value: &str) -> FigmentValue {
+ // Handle array syntax [a, b, c]
+ if value.starts_with('[') && value.ends_with(']') {
+ let inner = value.trim_start_matches('[').trim_end_matches(']');
+ let elements: Vec<FigmentValue> = split_array_elements(inner)
+ .into_iter()
+ .map(|s| parse_env_value(strip_quotes(s)))
+ .collect();
+ return FigmentValue::from(elements);
+ }
+
+ // Boolean
+ match value.to_lowercase().as_str() {
+ "true" => return FigmentValue::from(true),
+ "false" => return FigmentValue::from(false),
+ _ => {}
+ }
+
+ // Try numeric types
+ if let Ok(n) = value.parse::<u64>() {
+ return FigmentValue::from(n);
+ }
+ if let Ok(n) = value.parse::<i64>() {
+ return FigmentValue::from(n);
+ }
+ if let Ok(n) = value.parse::<f64>() {
+ return FigmentValue::from(n);
+ }
+
+ // Default to string
+ FigmentValue::from(value)
+}
+
+/// Parse an environment variable value into a JSON value with type inference.
+pub fn parse_env_value_to_json(value: &str) -> serde_json::Value {
+ if value.starts_with('[') && value.ends_with(']') {
+ let inner = value.trim_start_matches('[').trim_end_matches(']');
+ let values: Vec<serde_json::Value> = split_array_elements(inner)
+ .into_iter()
+ .map(|s| parse_env_value_to_json(strip_quotes(s)))
+ .collect();
+ return serde_json::Value::Array(values);
+ }
+
+ match value.to_lowercase().as_str() {
+ "true" => return serde_json::Value::Bool(true),
+ "false" => return serde_json::Value::Bool(false),
+ _ => {}
+ }
+
+ if let Ok(n) = value.parse::<u64>() {
+ return serde_json::Value::Number(n.into());
+ }
+ if let Ok(n) = value.parse::<i64>() {
+ return serde_json::Value::Number(n.into());
+ }
+ if let Ok(n) = value.parse::<f64>()
+ && let Some(num) = serde_json::Number::from_f64(n)
+ {
+ return serde_json::Value::Number(num);
+ }
+
+ serde_json::Value::String(value.to_owned())
+}
+
+pub(crate) fn split_array_elements(s: &str) -> Vec<&str> {
+ let mut elements = Vec::new();
+ let mut start = 0;
+ let mut in_quotes = false;
+ for (i, c) in s.char_indices() {
+ match c {
+ '"' => in_quotes = !in_quotes,
+ ',' if !in_quotes => {
+ elements.push(s[start..i].trim());
+ start = i + 1;
+ }
+ _ => {}
+ }
+ }
+ if start < s.len() {
+ elements.push(s[start..].trim());
+ }
+ elements
+}
+
+pub(crate) fn strip_quotes(s: &str) -> &str {
+ s.trim_matches('"')
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn parse_env_value_to_json_handles_booleans() {
+ assert_eq!(
+ parse_env_value_to_json("true"),
+ serde_json::Value::Bool(true)
+ );
+ assert_eq!(
+ parse_env_value_to_json("TRUE"),
+ serde_json::Value::Bool(true)
+ );
+ assert_eq!(
+ parse_env_value_to_json("false"),
+ serde_json::Value::Bool(false)
+ );
+ assert_eq!(
+ parse_env_value_to_json("False"),
+ serde_json::Value::Bool(false)
+ );
+ }
+
+ #[test]
+ fn parse_env_value_to_json_handles_integers() {
+ assert_eq!(parse_env_value_to_json("42"), serde_json::json!(42));
+ assert_eq!(parse_env_value_to_json("0"), serde_json::json!(0));
+ assert_eq!(parse_env_value_to_json("-123"), serde_json::json!(-123));
+ }
+
+ #[test]
+ fn parse_env_value_to_json_handles_floats() {
+ assert_eq!(parse_env_value_to_json("1.5"), serde_json::json!(1.5));
+ assert_eq!(parse_env_value_to_json("-2.5"), serde_json::json!(-2.5));
+ }
+
+ #[test]
+ fn parse_env_value_to_json_handles_strings() {
+ assert_eq!(parse_env_value_to_json("hello"), serde_json::json!("hello"));
+ assert_eq!(
+ parse_env_value_to_json("127.0.0.1:8080"),
+ serde_json::json!("127.0.0.1:8080")
+ );
+ }
+
+ #[test]
+ fn parse_env_value_to_json_handles_arrays() {
+ assert_eq!(
+ parse_env_value_to_json("[1, 2, 3]"),
+ serde_json::json!([1, 2, 3])
+ );
+ assert_eq!(
+ parse_env_value_to_json("[\"a\", \"b\"]"),
+ serde_json::json!(["a", "b"])
+ );
+ assert_eq!(
+ parse_env_value_to_json("[true, false]"),
+ serde_json::json!([true, false])
+ );
+ }
+
+ #[test]
+ fn split_array_elements_handles_simple_arrays() {
+ assert_eq!(split_array_elements("a, b, c"), vec!["a", "b", "c"]);
+ assert_eq!(split_array_elements("1,2,3"), vec!["1", "2", "3"]);
+ }
+
+ #[test]
+ fn split_array_elements_handles_quoted_strings() {
+ assert_eq!(
+ split_array_elements("\"hello, world\", \"foo\""),
+ vec!["\"hello, world\"", "\"foo\""]
+ );
+ }
+
+ #[test]
+ fn strip_quotes_removes_surrounding_quotes() {
+ assert_eq!(strip_quotes("\"hello\""), "hello");
+ assert_eq!(strip_quotes("hello"), "hello");
+ assert_eq!(strip_quotes("\"\""), "");
+ }
+}
diff --git a/core/configs/src/configs_impl/traits.rs b/core/configs/src/configs_impl/traits.rs
new file mode 100644
index 000000000..213fa42ce
--- /dev/null
+++ b/core/configs/src/configs_impl/traits.rs
@@ -0,0 +1,36 @@
+/*
+ * 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.
+ */
+
+//! Configuration provider traits.
+
+use super::error::ConfigurationError;
+use serde::{Serialize, de::DeserializeOwned};
+use std::fmt::Display;
+use std::future::Future;
+
+/// Type alias for configuration that can be serialized, deserialized, has defaults, and can be displayed.
+pub trait ConfigurationType: Serialize + DeserializeOwned + Default + Display {}
+impl<T: Serialize + DeserializeOwned + Default + Display> ConfigurationType for T {}
+
+/// Trait for configuration providers that can load a configuration.
+pub trait ConfigProvider {
+ fn load_config<T: ConfigurationType>(
+ &self,
+ ) -> impl Future<Output = Result<T, ConfigurationError>>;
+}
diff --git a/core/configs/src/configs_impl/typed_env_provider.rs b/core/configs/src/configs_impl/typed_env_provider.rs
new file mode 100644
index 000000000..dd8926985
--- /dev/null
+++ b/core/configs/src/configs_impl/typed_env_provider.rs
@@ -0,0 +1,597 @@
+/*
+ * 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.
+ */
+
+use super::env_mapping::ConfigEnvMappings;
+use super::error::ConfigurationError;
+use super::parsing::parse_env_value;
+use figment::{
+ Profile, Provider,
+ value::{Dict, Map as FigmentMap, Tag, Value as FigmentValue},
+};
+use std::{collections::HashSet, env, marker::PhantomData};
+use tracing::{info, warn};
+
+const SECRET_MASK: &str = "******";
+
+/// Controls how env var names are resolved during deserialization.
+enum EnvNameResolution<'a> {
+ /// Use mapping.env_name directly (for configs with compile-time prefix)
+ Direct,
+ /// Prepend runtime prefix to mapping.env_name (for plugin configs)
+ PrependPrefix(&'a str),
+}
+
+/// Controls filtering and messaging for unknown env var warnings.
+enum WarningContext<'a> {
+ /// Main config: filter IGNORED_ENV_VARS and DELEGATED prefixes
+ MainConfig,
+ /// Connector config: filter PLUGIN_CONFIG_ prefix
+ ConnectorConfig(&'a str),
+}
+
+/// Environment variables starting with IGGY_ that are NOT config values.
+/// These are used for test control, CI, CLI behavior, config file paths, etc.
+const IGNORED_ENV_VARS: &[&str] = &[
+ "IGGY_CI_BUILD",
+ "IGGY_CONFIG_PATH",
+ "IGGY_CONNECTORS_CONFIG_PATH",
+ "IGGY_MCP_CONFIG_PATH",
+ "IGGY_ROOT_PASSWORD",
+ "IGGY_ROOT_USERNAME",
+ "IGGY_TEST_VERBOSE",
+];
+
+/// Prefixes for env vars handled by separate providers with runtime prefixes.
+/// The main config provider skips these; each sub-provider validates its own vars.
+const DELEGATED_ENV_VAR_PREFIXES: &[&str] = &["IGGY_CONNECTORS_SINK_", "IGGY_CONNECTORS_SOURCE_"];
+
+type ProfileMap = FigmentMap<Profile, Dict>;
+
+/// Type-safe environment variable provider that uses compile-time generated mappings.
+///
+/// Unlike runtime schema inference providers, this provider uses the `ConfigEnvMappings`
+/// trait to directly look up known env var names, eliminating ambiguity in path resolution.
+///
+/// # Example
+/// ```ignore
+/// let provider = TypedEnvProvider::<ServerConfig>::new("IGGY_", &["IGGY_SYSTEM_ENCRYPTION_KEY"]);
+/// ```
+#[derive(Debug, Clone)]
+pub struct TypedEnvProvider<T: ConfigEnvMappings> {
+ prefix: String,
+ secret_keys: Vec<String>,
+ _phantom: PhantomData<T>,
+}
+
+impl<T: ConfigEnvMappings> TypedEnvProvider<T> {
+ /// Create a new typed environment provider.
+ ///
+ /// # Arguments
+ /// * `prefix` - Environment variable prefix to validate against (e.g., "IGGY_")
+ /// * `secret_keys` - Environment variable names that should be masked in logs
+ pub fn new(prefix: &str, secret_keys: &[&str]) -> Self {
+ Self {
+ prefix: prefix.to_string(),
+ secret_keys: secret_keys.iter().map(|s| s.to_string()).collect(),
+ _phantom: PhantomData,
+ }
+ }
+
+ /// Create a provider with a runtime-determined prefix.
+ ///
+ /// Use this when the config struct has no prefix attribute (generates relative mappings)
+ /// but you need to apply a prefix at runtime.
+ ///
+ /// # Example
+ /// ```ignore
+ /// // SinkConfig has no prefix, generates: ENABLED, NAME, PATH
+ /// // We prepend "IGGY_CONNECTORS_SINK_POSTGRES_" at runtime
+ /// let provider = TypedEnvProvider::<SinkConfig>::with_runtime_prefix(
+ /// "IGGY_CONNECTORS_SINK_POSTGRES_",
+ /// &[]
+ /// );
+ /// ```
+ pub fn with_runtime_prefix(prefix: &str, secret_keys: &[&str]) -> Self {
+ Self::new(prefix, secret_keys)
+ }
+
+ /// Create a provider with default prefix "IGGY_" and no secret keys.
+ pub fn with_default_prefix() -> Self {
+ Self::new("IGGY_", &[])
+ }
+
+ /// Create a provider using auto-collected secrets from `#[config_env(secret)]` markers.
+ pub fn from_config(prefix: &str) -> Self {
+ let secret_keys: Vec<String> = T::secret_env_names()
+ .into_iter()
+ .map(String::from)
+ .collect();
+ Self {
+ prefix: prefix.to_string(),
+ secret_keys,
+ _phantom: PhantomData,
+ }
+ }
+
+ /// Deserialize with runtime prefix prepended to each mapping's env_name.
+ ///
+ /// Unlike `deserialize()`, this method prepends `self.prefix` to each mapping's
+ /// env_name, allowing for dynamic prefix construction at runtime.
+ pub fn deserialize_with_runtime_prefix(&self) -> Result<ProfileMap, ConfigurationError> {
+ self.warn_unknown_env_vars_inner(WarningContext::ConnectorConfig(&self.prefix));
+ self.deserialize_inner(EnvNameResolution::PrependPrefix(&self.prefix))
+ }
+
+ /// Deserialize environment variables into a configuration profile map.
+ ///
+ /// This method:
+ /// 1. Validates that all env vars with the prefix are known (warns on unknown)
+ /// 2. Iterates over compile-time generated mappings and applies set values
+ pub fn deserialize(&self) -> Result<ProfileMap, ConfigurationError> {
+ self.warn_unknown_env_vars_inner(WarningContext::MainConfig);
+ self.deserialize_inner(EnvNameResolution::Direct)
+ }
+
+ fn deserialize_inner(
+ &self,
+ resolution: EnvNameResolution,
+ ) -> Result<ProfileMap, ConfigurationError> {
+ let mut root_dict = Dict::new();
+
+ for mapping in T::env_mappings() {
+ let env_name = match &resolution {
+ EnvNameResolution::Direct => mapping.env_name.to_string(),
+ EnvNameResolution::PrependPrefix(prefix) => {
+ format!("{}{}", prefix, mapping.env_name)
+ }
+ };
+
+ let env_value = match env::var(&env_name) {
+ Ok(val) => val,
+ Err(_) => continue,
+ };
+
+ let is_secret = mapping.is_secret
+ || match &resolution {
+ EnvNameResolution::Direct => self.secret_keys.contains(&env_name),
+ EnvNameResolution::PrependPrefix(_) => {
+ self.secret_keys.iter().any(|s| env_name.ends_with(s))
+ }
+ };
+
+ let display_value = if is_secret {
+ SECRET_MASK.to_string()
+ } else {
+ env_value.clone()
+ };
+
+ info!(
+ "{} value changed to: {} from environment variable",
+ env_name, display_value
+ );
+
+ let parsed_value = parse_env_value(&env_value);
+ Self::insert_at_path(&mut root_dict, mapping.config_path, parsed_value);
+ }
+
+ let root_dict =
+ Self::convert_numeric_dicts_to_arrays(FigmentValue::Dict(Tag::Default, root_dict));
+ let root_dict = match root_dict {
+ FigmentValue::Dict(_, dict) => dict,
+ _ => Dict::new(),
+ };
+
+ let mut data = ProfileMap::new();
+ data.insert(Profile::default(), root_dict);
+ Ok(data)
+ }
+
+ /// Recursively convert dicts with all numeric string keys to arrays.
+ fn convert_numeric_dicts_to_arrays(value: FigmentValue) -> FigmentValue {
+ match value {
+ FigmentValue::Dict(tag, dict) => {
+ let processed: Dict = dict
+ .into_iter()
+ .map(|(k, v)| (k, Self::convert_numeric_dicts_to_arrays(v)))
+ .collect();
+
+ if Self::is_array_dict(&processed) {
+ Self::dict_to_array(processed)
+ } else {
+ FigmentValue::Dict(tag, processed)
+ }
+ }
+ FigmentValue::Array(tag, arr) => {
+ let processed: Vec<FigmentValue> = arr
+ .into_iter()
+ .map(Self::convert_numeric_dicts_to_arrays)
+ .collect();
+ FigmentValue::Array(tag, processed)
+ }
+ other => other,
+ }
+ }
+
+ fn is_array_dict(dict: &Dict) -> bool {
+ if dict.is_empty() {
+ return false;
+ }
+ dict.keys().all(|k| k.parse::<usize>().is_ok())
+ }
+
+ fn dict_to_array(dict: Dict) -> FigmentValue {
+ let mut indexed: Vec<(usize, FigmentValue)> = dict
+ .into_iter()
+ .filter_map(|(k, v)| k.parse::<usize>().ok().map(|i| (i, v)))
+ .collect();
+
+ indexed.sort_by_key(|(i, _)| *i);
+
+ let max_index = indexed.last().map(|(i, _)| *i).unwrap_or(0);
+ let mut array = vec![FigmentValue::Dict(Tag::Default, Dict::new()); max_index + 1];
+
+ for (i, v) in indexed {
+ array[i] = v;
+ }
+
+ FigmentValue::Array(Tag::Default, array)
+ }
+
+ fn warn_unknown_env_vars_inner(&self, context: WarningContext) {
+ let known_vars: HashSet<String> = match &context {
+ WarningContext::MainConfig => T::env_mappings()
+ .iter()
+ .map(|m| m.env_name.to_string())
+ .collect(),
+ WarningContext::ConnectorConfig(prefix) => T::env_mappings()
+ .iter()
+ .map(|m| format!("{}{}", prefix, m.env_name))
+ .collect(),
+ };
+ let known_refs: HashSet<&str> = known_vars.iter().map(|s| s.as_str()).collect();
+
+ for (key, _) in env::vars() {
+ if !key.starts_with(&self.prefix) {
+ continue;
+ }
+
+ if known_vars.contains(&key) {
+ continue;
+ }
+
+ let should_skip = match &context {
+ WarningContext::MainConfig => {
+ IGNORED_ENV_VARS.contains(&key.as_str())
+ || DELEGATED_ENV_VAR_PREFIXES
+ .iter()
+ .any(|p| key.starts_with(p))
+ }
+ WarningContext::ConnectorConfig(prefix) => {
+ let plugin_config_prefix = format!("{}PLUGIN_CONFIG_", prefix);
+ key.starts_with(&plugin_config_prefix)
+ }
+ };
+ if should_skip {
+ continue;
+ }
+
+ let suggestions = Self::find_similar_vars(&key, &known_refs);
+ let suggestion_hint = if suggestions.is_empty() {
+ String::new()
+ } else {
+ format!(" Similar: {}", suggestions.join(", "))
+ };
+
+ let debug_msg = match &context {
+ WarningContext::MainConfig => format!(
+ "Unknown IGGY_ env var: '{}'.{}. Add to IGNORED_ENV_VARS if intentional, \
+ or add #[derive(ConfigEnv)] to the config struct.",
+ key, suggestion_hint
+ ),
+ WarningContext::ConnectorConfig(_) => format!(
+ "Unknown connector env var: '{}'.{} Check field name or add #[config_env(skip)].",
+ key, suggestion_hint
+ ),
+ };
+ debug_assert!(false, "{}", debug_msg);
+
+ Self::warn_unknown_var(&key, &suggestions);
+ }
+ }
+
+ fn warn_unknown_var(unknown_var: &str, suggestions: &[String]) {
+ if suggestions.is_empty() {
+ warn!(
+ "Unknown environment variable '{}' will be ignored. \
+ Use --list-env-vars to see all valid environment variables.",
+ unknown_var
+ );
+ } else {
+ warn!(
+ "Unknown environment variable '{}' will be ignored. Similar variables: {}?",
+ unknown_var,
+ suggestions.join(", ")
+ );
+ }
+ }
+
+ /// Find similar variable names using Levenshtein edit distance.
+ /// Returns up to 3 suggestions sorted by similarity.
+ pub fn find_similar_vars(
+ unknown: &str,
+ known: &std::collections::HashSet<&str>,
+ ) -> Vec<String> {
+ let unknown_lower = unknown.to_lowercase();
+ let mut suggestions: Vec<(String, usize)> = known
+ .iter()
+ .filter_map(|&known_var| {
+ let known_lower = known_var.to_lowercase();
+ let distance = Self::levenshtein_distance(&unknown_lower, &known_lower);
+ let threshold = (unknown.len().max(known_var.len()) * 3) / 10;
+ if distance <= threshold.max(3) {
+ Some((known_var.to_string(), distance))
+ } else {
+ None
+ }
+ })
+ .collect();
+
+ suggestions.sort_by_key(|(_, dist)| *dist);
+ suggestions.truncate(3);
+ suggestions.into_iter().map(|(s, _)| s).collect()
+ }
+
+ fn levenshtein_distance(a: &str, b: &str) -> usize {
+ let a_chars: Vec<char> = a.chars().collect();
+ let b_chars: Vec<char> = b.chars().collect();
+ let a_len = a_chars.len();
+ let b_len = b_chars.len();
+
+ if a_len == 0 {
+ return b_len;
+ }
+ if b_len == 0 {
+ return a_len;
+ }
+
+ let mut prev_row: Vec<usize> = (0..=b_len).collect();
+ let mut curr_row = vec![0; b_len + 1];
+
+ for (i, a_char) in a_chars.iter().enumerate() {
+ curr_row[0] = i + 1;
+
+ for (j, b_char) in b_chars.iter().enumerate() {
+ let cost = if a_char == b_char { 0 } else { 1 };
+ curr_row[j + 1] = (prev_row[j + 1] + 1)
+ .min(curr_row[j] + 1)
+ .min(prev_row[j] + cost);
+ }
+
+ std::mem::swap(&mut prev_row, &mut curr_row);
+ }
+
+ prev_row[b_len]
+ }
+
+ fn insert_at_path(dict: &mut Dict, path: &str, value: FigmentValue) {
+ let segments: Vec<&str> = path.split('.').collect();
+ Self::insert_at_path_segments(dict, &segments, value);
+ }
+
+ fn insert_at_path_segments(dict: &mut Dict, segments: &[&str], value: FigmentValue) {
+ if segments.is_empty() {
+ return;
+ }
+
+ if segments.len() == 1 {
+ dict.insert(segments[0].to_string(), value);
+ return;
+ }
+
+ let key = segments[0].to_string();
+ dict.entry(key.clone())
+ .or_insert_with(|| FigmentValue::Dict(Tag::Default, Dict::new()));
+
+ if let Some(FigmentValue::Dict(_, inner_dict)) = dict.get_mut(&key) {
+ Self::insert_at_path_segments(inner_dict, &segments[1..], value);
+ }
+ }
+}
+
+impl<T: ConfigEnvMappings + Send + Sync> Provider for TypedEnvProvider<T> {
+ fn metadata(&self) -> figment::Metadata {
+ figment::Metadata::named("typed environment variables")
+ }
+
+ fn data(&self) -> Result<ProfileMap, figment::Error> {
+ self.deserialize()
+ .map_err(|e| figment::Error::from(format!("Failed to deserialize env vars: {e}")))
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::ConfigEnv;
+ use serde::{Deserialize, Serialize};
+ use std::collections::HashSet;
+
+ #[derive(Debug, Default, Clone, Serialize, Deserialize, ConfigEnv)]
+ #[config_env(prefix = "TEST_", name = "test-config")]
+ struct TestConfig {
+ enabled: bool,
+ name: String,
+ count: u32,
+ nested: NestedConfig,
+ }
+
+ #[derive(Debug, Default, Clone, Serialize, Deserialize, ConfigEnv)]
+ struct NestedConfig {
+ value: String,
+ flag: bool,
+ }
+
+ #[test]
+ fn levenshtein_distance_identical_strings() {
+ assert_eq!(
+ TypedEnvProvider::<TestConfig>::levenshtein_distance("hello", "hello"),
+ 0
+ );
+ }
+
+ #[test]
+ fn levenshtein_distance_single_char_difference() {
+ assert_eq!(
+ TypedEnvProvider::<TestConfig>::levenshtein_distance("hello", "hallo"),
+ 1
+ );
+ assert_eq!(
+ TypedEnvProvider::<TestConfig>::levenshtein_distance("cat", "car"),
+ 1
+ );
+ }
+
+ #[test]
+ fn levenshtein_distance_empty_strings() {
+ assert_eq!(
+ TypedEnvProvider::<TestConfig>::levenshtein_distance("", ""),
+ 0
+ );
+ assert_eq!(
+ TypedEnvProvider::<TestConfig>::levenshtein_distance("abc", ""),
+ 3
+ );
+ assert_eq!(
+ TypedEnvProvider::<TestConfig>::levenshtein_distance("", "abc"),
+ 3
+ );
+ }
+
+ #[test]
+ fn levenshtein_distance_insertions_deletions() {
+ assert_eq!(
+ TypedEnvProvider::<TestConfig>::levenshtein_distance("abc", "ab"),
+ 1
+ );
+ assert_eq!(
+ TypedEnvProvider::<TestConfig>::levenshtein_distance("ab", "abc"),
+ 1
+ );
+ }
+
+ #[test]
+ fn find_similar_vars_suggests_typos() {
+ let known: HashSet<&str> = ["TEST_ENABLED", "TEST_NAME", "TEST_COUNT"]
+ .into_iter()
+ .collect();
+
+ let suggestions = TypedEnvProvider::<TestConfig>::find_similar_vars("TEST_ENABELD", &known);
+ assert!(suggestions.contains(&"TEST_ENABLED".to_string()));
+
+ let suggestions = TypedEnvProvider::<TestConfig>::find_similar_vars("TEST_NANE", &known);
+ assert!(suggestions.contains(&"TEST_NAME".to_string()));
+ }
+
+ #[test]
+ fn find_similar_vars_returns_empty_for_unrelated() {
+ let known: HashSet<&str> = ["TEST_ENABLED", "TEST_NAME"].into_iter().collect();
+
+ let suggestions =
+ TypedEnvProvider::<TestConfig>::find_similar_vars("COMPLETELY_DIFFERENT_VAR", &known);
+ assert!(suggestions.is_empty());
+ }
+
+ #[test]
+ fn find_similar_vars_limits_to_three_suggestions() {
+ let known: HashSet<&str> = ["VAR_A", "VAR_B", "VAR_C", "VAR_D", "VAR_E"]
+ .into_iter()
+ .collect();
+
+ let suggestions = TypedEnvProvider::<TestConfig>::find_similar_vars("VAR_X", &known);
+ assert!(suggestions.len() <= 3);
+ }
+
+ #[test]
+ fn test_config_generates_env_mappings() {
+ let mappings = TestConfig::env_mappings();
+ assert!(!mappings.is_empty());
+
+ assert!(mappings.iter().any(|m| m.env_name == "TEST_ENABLED"));
+ assert!(mappings.iter().any(|m| m.env_name == "TEST_NAME"));
+ assert!(mappings.iter().any(|m| m.env_name == "TEST_COUNT"));
+ assert!(mappings.iter().any(|m| m.env_name == "TEST_NESTED_VALUE"));
+ assert!(mappings.iter().any(|m| m.env_name == "TEST_NESTED_FLAG"));
+ }
+
+ #[test]
+ fn test_config_mappings_have_correct_paths() {
+ let mappings = TestConfig::env_mappings();
+
+ let enabled = mappings
+ .iter()
+ .find(|m| m.env_name == "TEST_ENABLED")
+ .unwrap();
+ assert_eq!(enabled.config_path, "enabled");
+
+ let nested_value = mappings
+ .iter()
+ .find(|m| m.env_name == "TEST_NESTED_VALUE")
+ .unwrap();
+ assert_eq!(nested_value.config_path, "nested.value");
+ }
+
+ #[test]
+ fn typed_provider_deserializes_env_vars() {
+ unsafe {
+ env::set_var("TEST_ENABLED", "true");
+ env::set_var("TEST_NAME", "test-name");
+ env::set_var("TEST_COUNT", "42");
+ }
+
+ let provider = TypedEnvProvider::<TestConfig>::new("TEST_", &[]);
+ let data = provider.deserialize().expect("deserialize failed");
+
+ unsafe {
+ env::remove_var("TEST_ENABLED");
+ env::remove_var("TEST_NAME");
+ env::remove_var("TEST_COUNT");
+ }
+
+ let profile_data = data
+ .get(&figment::Profile::default())
+ .expect("no default profile");
+
+ let enabled = profile_data.get("enabled").expect("no enabled");
+ assert!(matches!(enabled, FigmentValue::Bool(_, true)));
+
+ let name = profile_data.get("name").expect("no name");
+ if let FigmentValue::String(_, s) = name {
+ assert_eq!(s, "test-name");
+ } else {
+ panic!("name should be string");
+ }
+
+ let count = profile_data.get("count").expect("no count");
+ if let FigmentValue::Num(_, figment::value::Num::U64(n)) = count {
+ assert_eq!(*n, 42);
+ } else {
+ panic!("count should be u64");
+ }
+ }
+}
diff --git a/core/configs/src/lib.rs b/core/configs/src/lib.rs
new file mode 100644
index 000000000..d1410ab8a
--- /dev/null
+++ b/core/configs/src/lib.rs
@@ -0,0 +1,28 @@
+/*
+ * 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.
+ */
+
+extern crate self as configs;
+
+mod configs_impl;
+
+pub use configs_derive::ConfigEnv;
+pub use configs_impl::{
+ ConfigEnvMappings, ConfigProvider, ConfigurationError, ConfigurationType, EnvVarMapping,
+ FileConfigProvider, TypedEnvProvider, parse_env_value_to_json,
+};
diff --git a/core/configs_derive/Cargo.toml b/core/configs_derive/Cargo.toml
new file mode 100644
index 000000000..6945a7e1d
--- /dev/null
+++ b/core/configs_derive/Cargo.toml
@@ -0,0 +1,32 @@
+# 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.
+
+[package]
+name = "configs_derive"
+version = "0.1.0"
+edition = "2024"
+license = "Apache-2.0"
+publish = false
+
+[lib]
+proc-macro = true
+
+[dependencies]
+darling = { workspace = true }
+proc-macro2 = { workspace = true }
+quote = { workspace = true }
+syn = { workspace = true }
diff --git a/core/configs_derive/src/config_env.rs b/core/configs_derive/src/config_env.rs
new file mode 100644
index 000000000..1eee37014
--- /dev/null
+++ b/core/configs_derive/src/config_env.rs
@@ -0,0 +1,583 @@
+/*
+ * 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.
+ */
+
+use darling::{FromDeriveInput, FromField, FromVariant};
+use proc_macro2::TokenStream as TokenStream2;
+use quote::{format_ident, quote};
+use syn::{DeriveInput, Generics, Ident, Type};
+
+/// Maximum number of array elements to generate env var mappings for.
+///
+/// For `Vec<T>` fields, mappings are generated for indices 0 through 9 (e.g.,
+/// `FIELD_0_NAME`, `FIELD_1_NAME`, ..., `FIELD_9_NAME`). Environment variables
+/// for indices beyond this limit are silently ignored.
+///
+/// This limit is sufficient for all known Iggy configuration arrays.
+const MAX_ARRAY_ELEMENTS: usize = 10;
+
+/// Container-level attributes for `#[config_env(...)]`
+#[derive(Debug, FromDeriveInput)]
+#[darling(attributes(config_env), supports(struct_named, enum_any))]
+pub struct ConfigEnvOpts {
+ ident: Ident,
+ generics: Generics,
+ data: darling::ast::Data<VariantOpts, FieldOpts>,
+
+ /// Environment variable prefix (e.g., "IGGY_"). Only needed on root config.
+ #[darling(default)]
+ prefix: Option<String>,
+
+ /// Metadata name for figment Provider (e.g., "iggy-server-config"). Only needed on root config.
+ #[darling(default)]
+ name: Option<String>,
+}
+
+/// Variant-level attributes for `#[config_env(...)]` on enums
+#[derive(Debug, FromVariant)]
+#[darling(attributes(config_env))]
+struct VariantOpts {
+ #[allow(dead_code)]
+ ident: Ident,
+ fields: darling::ast::Fields<FieldOpts>,
+
+ /// Skip this variant entirely
+ #[darling(default)]
+ skip: bool,
+}
+
+/// Field-level attributes for `#[config_env(...)]`
+#[derive(Debug, FromField)]
+#[darling(attributes(config_env))]
+struct FieldOpts {
+ ident: Option<Ident>,
+ ty: Type,
+
+ /// Explicit env var segment name (overrides field name conversion)
+ #[darling(default)]
+ name: Option<String>,
+
+ /// Skip this field entirely
+ #[darling(default)]
+ skip: bool,
+
+ /// Mark as secret (for builder documentation)
+ #[darling(default)]
+ secret: bool,
+
+ /// Treat this field as a leaf value (not a nested config struct)
+ #[darling(default)]
+ leaf: bool,
+}
+
+/// Represents a single environment variable mapping (leaf field)
+struct EnvMapping {
+ const_name: Ident,
+ /// Relative env name suffix (e.g., "ENABLED" not "IGGY_HTTP_ENABLED")
+ env_suffix: String,
+ /// Relative config path (e.g., "enabled" not "http.enabled")
+ config_path: String,
+ is_secret: bool,
+ builder_method_name: Ident,
+}
+
+/// Information about a nested field for mapping collection
+struct NestedFieldInfo {
+ /// The element type (inner type for Vec/Arc/Box/Option)
+ element_type: Type,
+ /// Whether this is a Vec field requiring array index expansion
+ is_vec: bool,
+ /// The field name in UPPER_CASE for env var construction
+ field_env_segment: String,
+ /// The field name for config path construction
+ field_name: String,
+}
+
+pub fn generate_impl(input: &DeriveInput) -> TokenStream2 {
+ match ConfigEnvOpts::from_derive_input(input) {
+ Ok(opts) => generate_from_opts(opts),
+ Err(e) => e.write_errors(),
+ }
+}
+
+fn generate_from_opts(opts: ConfigEnvOpts) -> TokenStream2 {
+ let type_name = &opts.ident;
+ let (impl_generics, ty_generics, where_clause) = opts.generics.split_for_impl();
+
+ match opts.data {
+ darling::ast::Data::Struct(fields) => generate_struct_impl(
+ type_name,
+ impl_generics,
+ ty_generics,
+ where_clause,
+ opts.prefix,
+ opts.name,
+ fields.fields,
+ ),
+ darling::ast::Data::Enum(variants) => generate_enum_impl(
+ type_name,
+ impl_generics,
+ ty_generics,
+ where_clause,
+ variants,
+ ),
+ }
+}
+
+/// Generate implementation for enum types.
+/// Combines mappings from all variant inner types that implement ConfigEnvMappings.
+fn generate_enum_impl(
+ enum_name: &Ident,
+ impl_generics: syn::ImplGenerics,
+ ty_generics: syn::TypeGenerics,
+ where_clause: Option<&syn::WhereClause>,
+ variants: Vec<VariantOpts>,
+) -> TokenStream2 {
+ // Collect inner types from newtype variants (single unnamed field)
+ let variant_types: Vec<&Type> = variants
+ .iter()
+ .filter(|v| !v.skip)
+ .filter_map(|v| {
+ // Only handle newtype variants: Variant(InnerType)
+ if v.fields.style == darling::ast::Style::Tuple && v.fields.fields.len() == 1 {
+ let field = &v.fields.fields[0];
+ if !field.skip {
+ return Some(&field.ty);
+ }
+ }
+ None
+ })
+ .collect();
+
+ if variant_types.is_empty() {
+ return quote! {
+ impl #impl_generics configs::ConfigEnvMappings for #enum_name #ty_generics #where_clause {
+ fn env_mappings() -> &'static [configs::EnvVarMapping] {
+ &[]
+ }
+ }
+ };
+ }
+
+ // Generate code to extend from each variant type
+ let extends: Vec<TokenStream2> = variant_types
+ .iter()
+ .map(|ty| {
+ quote! {
+ all_mappings.extend_from_slice(<#ty as configs::ConfigEnvMappings>::env_mappings());
+ }
+ })
+ .collect();
+
+ quote! {
+ impl #impl_generics configs::ConfigEnvMappings for #enum_name #ty_generics #where_clause {
+ fn env_mappings() -> &'static [configs::EnvVarMapping] {
+ static MAPPINGS: std::sync::OnceLock<Vec<configs::EnvVarMapping>> = std::sync::OnceLock::new();
+ MAPPINGS.get_or_init(|| {
+ let mut all_mappings: Vec<configs::EnvVarMapping> = Vec::new();
+ #(#extends)*
+ all_mappings
+ })
+ }
+ }
+ }
+}
+
+/// Generate implementation for struct types.
+fn generate_struct_impl(
+ struct_name: &Ident,
+ impl_generics: syn::ImplGenerics,
+ ty_generics: syn::TypeGenerics,
+ where_clause: Option<&syn::WhereClause>,
+ prefix: Option<String>,
+ name: Option<String>,
+ fields: Vec<FieldOpts>,
+) -> TokenStream2 {
+ let prefix_str = prefix.as_deref().unwrap_or("");
+ let has_prefix = !prefix_str.is_empty();
+
+ // Metadata name: use provided or derive from type name (e.g., "ServerConfig" -> "server-config")
+ let metadata_name = name.unwrap_or_else(|| {
+ let type_name = struct_name.to_string();
+ let mut result = String::new();
+ for (i, c) in type_name.chars().enumerate() {
+ if c.is_uppercase() && i > 0 {
+ result.push('-');
+ }
+ result.push(c.to_ascii_lowercase());
+ }
+ result
+ });
+
+ let (mappings, nested_fields) = collect_mappings(&fields);
+
+ let const_definitions = generate_const_definitions(&mappings, prefix_str);
+ let mapping_entries = generate_mapping_entries(&mappings);
+ let builder_methods = generate_builder_methods(&mappings, struct_name, prefix_str);
+ let builder_name = format_ident!("{}EnvBuilder", struct_name);
+ let mappings_count = mappings.len();
+
+ // Generate code to include nested type mappings
+ let nested_extends: Vec<TokenStream2> = nested_fields
+ .iter()
+ .map(|info| {
+ let ty = &info.element_type;
+ let field_env_segment = &info.field_env_segment;
+ let field_name = &info.field_name;
+
+ if info.is_vec {
+ // For Vec fields, expand with array indices
+ let max_elements = MAX_ARRAY_ELEMENTS;
+ quote! {
+ let nested_mappings = <#ty as configs::ConfigEnvMappings>::env_mappings();
+ for i in 0..#max_elements {
+ for mapping in nested_mappings {
+ // Transform env_name by prepending FIELD_INDEX_
+ let new_env_name: &'static str = Box::leak(
+ format!("{}_{}_{}", #field_env_segment, i, mapping.env_name).into_boxed_str()
+ );
+ // Transform config_path by prepending field.index.
+ let new_config_path: &'static str = Box::leak(
+ format!("{}.{}.{}", #field_name, i, mapping.config_path).into_boxed_str()
+ );
+ all_mappings.push(configs::EnvVarMapping {
+ env_name: new_env_name,
+ config_path: new_config_path,
+ is_secret: mapping.is_secret,
+ });
+ }
+ }
+ }
+ } else {
+ // For non-Vec nested types, transform by prepending field name
+ quote! {
+ let nested_mappings = <#ty as configs::ConfigEnvMappings>::env_mappings();
+ for mapping in nested_mappings {
+ // Transform env_name by prepending FIELD_
+ let new_env_name: &'static str = Box::leak(
+ format!("{}_{}", #field_env_segment, mapping.env_name).into_boxed_str()
+ );
+ // Transform config_path by prepending field.
+ let new_config_path: &'static str = Box::leak(
+ format!("{}.{}", #field_name, mapping.config_path).into_boxed_str()
+ );
+ all_mappings.push(configs::EnvVarMapping {
+ env_name: new_env_name,
+ config_path: new_config_path,
+ is_secret: mapping.is_secret,
+ });
+ }
+ }
+ }
+ })
+ .collect();
+
+ let prefix_application = if has_prefix {
+ quote! {
+ let all_mappings: Vec<configs::EnvVarMapping> = all_mappings
+ .into_iter()
+ .map(|m| configs::EnvVarMapping {
+ env_name: Box::leak(format!("{}{}", #prefix_str, m.env_name).into_boxed_str()),
+ config_path: m.config_path,
+ is_secret: m.is_secret,
+ })
+ .collect();
+ }
+ } else {
+ quote! {}
+ };
+
+ // If there are nested types, we need to use OnceLock to combine mappings
+ let env_mappings_impl = if nested_fields.is_empty() && !has_prefix {
+ // Simple case: no nested fields, no prefix transformation needed
+ quote! {
+ fn env_mappings() -> &'static [configs::EnvVarMapping] {
+ static MAPPINGS: [configs::EnvVarMapping; #mappings_count] = [
+ #(#mapping_entries),*
+ ];
+ &MAPPINGS
+ }
+ }
+ } else {
+ // Complex case: need OnceLock for dynamic construction
+ quote! {
+ fn env_mappings() -> &'static [configs::EnvVarMapping] {
+ static MAPPINGS: std::sync::OnceLock<Vec<configs::EnvVarMapping>> = std::sync::OnceLock::new();
+ MAPPINGS.get_or_init(|| {
+ let own_mappings: [configs::EnvVarMapping; #mappings_count] = [
+ #(#mapping_entries),*
+ ];
+ let mut all_mappings = Vec::from(own_mappings);
+ #(#nested_extends)*
+ #prefix_application
+ all_mappings
+ })
+ }
+ }
+ };
+
+ quote! {
+ impl #impl_generics #struct_name #ty_generics #where_clause {
+ /// Environment variable prefix for this config type.
+ pub const ENV_PREFIX: &'static str = #prefix_str;
+
+ /// Metadata name for figment Provider.
+ pub const ENV_PROVIDER_NAME: &'static str = #metadata_name;
+
+ #(#const_definitions)*
+ }
+
+ impl #impl_generics configs::ConfigEnvMappings for #struct_name #ty_generics #where_clause {
+ #env_mappings_impl
+ }
+
+ /// Type-safe builder for constructing environment variable maps for tests.
+ #[derive(Debug, Default, Clone)]
+ pub struct #builder_name {
+ envs: std::collections::HashMap<String, String>,
+ }
+
+ impl #builder_name {
+ /// Create a new builder instance.
+ pub fn new() -> Self {
+ Self::default()
+ }
+
+ #(#builder_methods)*
+
+ /// Set a custom environment variable.
+ pub fn custom(mut self, key: impl Into<String>, value: impl std::fmt::Display) -> Self {
+ self.envs.insert(key.into(), value.to_string());
+ self
+ }
+
+ /// Extend with another builder's values.
+ pub fn extend(mut self, other: Self) -> Self {
+ self.envs.extend(other.envs);
+ self
+ }
+
+ /// Build the final environment variable map.
+ pub fn build(self) -> std::collections::HashMap<String, String> {
+ self.envs
+ }
+
+ /// Build and merge with existing environment variables.
+ pub fn build_with(self, existing: std::collections::HashMap<String, String>) -> std::collections::HashMap<String, String> {
+ let mut result = existing;
+ result.extend(self.envs);
+ result
+ }
+ }
+ }
+}
+
+/// Check if a type is a Rust primitive that doesn't need nested expansion.
+fn is_primitive_type(ty: &Type) -> bool {
+ let Type::Path(type_path) = ty else {
+ return false;
+ };
+ let Some(segment) = type_path.path.segments.last() else {
+ return false;
+ };
+
+ let ident = segment.ident.to_string();
+
+ // Rust primitives only
+ if matches!(
+ ident.as_str(),
+ "bool"
+ | "u8"
+ | "u16"
+ | "u32"
+ | "u64"
+ | "u128"
+ | "i8"
+ | "i16"
+ | "i32"
+ | "i64"
+ | "i128"
+ | "f32"
+ | "f64"
+ | "usize"
+ | "isize"
+ | "String"
+ | "str"
+ ) {
+ return true;
+ }
+
+ // Option<T> - check inner type
+ if ident == "Option"
+ && let syn::PathArguments::AngleBracketed(args) = &segment.arguments
+ && let Some(syn::GenericArgument::Type(inner)) = args.args.first()
+ {
+ return is_primitive_type(inner);
+ }
+
+ false
+}
+
+/// Extract type info, unwrapping Vec/Arc/Box/Option.
+/// Returns (inner_type, is_vec).
+fn extract_type_info(ty: &Type) -> (Type, bool) {
+ let Type::Path(type_path) = ty else {
+ return (ty.clone(), false);
+ };
+ let Some(segment) = type_path.path.segments.last() else {
+ return (ty.clone(), false);
+ };
+
+ let ident = segment.ident.to_string();
+ match ident.as_str() {
+ "Vec" => {
+ if let syn::PathArguments::AngleBracketed(args) = &segment.arguments
+ && let Some(syn::GenericArgument::Type(inner)) = args.args.first()
+ {
+ let (inner_type, _) = extract_type_info(inner);
+ return (inner_type, true);
+ }
+ }
+ "Arc" | "Box" | "Rc" | "Option" => {
+ if let syn::PathArguments::AngleBracketed(args) = &segment.arguments
+ && let Some(syn::GenericArgument::Type(inner)) = args.args.first()
+ {
+ return extract_type_info(inner);
+ }
+ }
+ _ => {}
+ }
+ (ty.clone(), false)
+}
+
+fn collect_mappings(fields: &[FieldOpts]) -> (Vec<EnvMapping>, Vec<NestedFieldInfo>) {
+ let mut mappings = Vec::new();
+ let mut nested_fields = Vec::new();
+
+ for field in fields {
+ if field.skip {
+ continue;
+ }
+
+ let field_ident = match &field.ident {
+ Some(ident) => ident,
+ None => continue,
+ };
+
+ let field_name = field_ident.to_string();
+ let field_env_segment = field_name.to_uppercase();
+
+ // Check if this is a leaf type or needs nested handling
+ let (inner_type, is_vec) = extract_type_info(&field.ty);
+
+ // A field is a leaf if:
+ // 1. It's explicitly marked with #[config_env(leaf)]
+ // 2. It's a Rust primitive type (bool, u32, String, etc.)
+ let is_leaf = field.leaf || is_primitive_type(&inner_type);
+
+ if is_leaf {
+ // Leaf field: generate a direct mapping with relative paths
+ let env_suffix = field
+ .name
+ .clone()
+ .unwrap_or_else(|| field_env_segment.clone());
+
+ let const_name = format_ident!("ENV_VAR_{}", field_env_segment);
+ let builder_method_name = format_ident!("{}", field_name);
+
+ mappings.push(EnvMapping {
+ const_name,
+ env_suffix,
+ config_path: field_name,
+ is_secret: field.secret,
+ builder_method_name,
+ });
+ } else {
+ // Nested field: will include mappings from the nested type
+ nested_fields.push(NestedFieldInfo {
+ element_type: inner_type,
+ is_vec,
+ field_env_segment,
+ field_name,
+ });
+ }
+ }
+
+ (mappings, nested_fields)
+}
+
+fn generate_const_definitions(mappings: &[EnvMapping], prefix: &str) -> Vec<TokenStream2> {
+ mappings
+ .iter()
+ .map(|m| {
+ let const_name = &m.const_name;
+ let env_var_name = format!("{}{}", prefix, m.env_suffix);
+ quote! {
+ pub const #const_name: &'static str = #env_var_name;
+ }
+ })
+ .collect()
+}
+
+fn generate_mapping_entries(mappings: &[EnvMapping]) -> Vec<TokenStream2> {
+ mappings
+ .iter()
+ .map(|m| {
+ let env_suffix = &m.env_suffix;
+ let config_path = &m.config_path;
+ let is_secret = m.is_secret;
+
+ quote! {
+ configs::EnvVarMapping {
+ env_name: #env_suffix,
+ config_path: #config_path,
+ is_secret: #is_secret,
+ }
+ }
+ })
+ .collect()
+}
+
+fn generate_builder_methods(
+ mappings: &[EnvMapping],
+ struct_name: &Ident,
+ prefix: &str,
+) -> Vec<TokenStream2> {
+ mappings
+ .iter()
+ .map(|m| {
+ let method_name = &m.builder_method_name;
+ let const_name = &m.const_name;
+ let env_var_name = format!("{}{}", prefix, m.env_suffix);
+ let doc_secret = if m.is_secret {
+ " (secret - will be masked in logs)"
+ } else {
+ ""
+ };
+ let doc = format!("Set `{}` environment variable.{}", env_var_name, doc_secret);
+
+ quote! {
+ #[doc = #doc]
+ pub fn #method_name(mut self, value: impl std::fmt::Display) -> Self {
+ self.envs.insert(#struct_name::#const_name.into(), value.to_string());
+ self
+ }
+ }
+ })
+ .collect()
+}
diff --git a/core/configs_derive/src/lib.rs b/core/configs_derive/src/lib.rs
new file mode 100644
index 000000000..7435e42cb
--- /dev/null
+++ b/core/configs_derive/src/lib.rs
@@ -0,0 +1,120 @@
+/*
+ * 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.
+ */
+
+//! Proc macro for generating type-safe environment variable mappings for configuration structs.
+//!
+//! This crate provides `#[derive(ConfigEnv)]` which generates:
+//! - Compile-time constants for all valid env var names
+//! - Type-safe builder API for tests
+//! - `ConfigEnvMappings` trait implementation for runtime discovery
+//!
+//! # Path Deduction
+//!
+//! The macro automatically deduces config paths from field names:
+//! - Each struct generates **relative** mappings (just its own field names)
+//! - Parents transform nested mappings by prepending the field name
+//! - Only the root struct needs `prefix`
+//!
+//! # Type Inference
+//!
+//! - **Leaf types** (primitives, String, known value types): generate direct mappings
+//! - **Nested types** (structs with `ConfigEnv` derive): include their mappings recursively
+//! - **Vec<T>**: expands to indexed mappings (e.g., `FIELD_0_NAME`, `FIELD_1_NAME`, ...)
+//! - **Arc<T>**, **Box<T>**, **Option<T>**: transparently unwrapped
+
+mod config_env;
+
+use proc_macro::TokenStream;
+use syn::{DeriveInput, parse_macro_input};
+
+/// Derive macro for generating environment variable mappings.
+///
+/// # Container Attributes
+/// - `#[config_env(prefix = "IGGY_")]` - Sets the env var prefix (only on root config)
+/// - `#[config_env(name = "iggy-server-config")]` - Sets the figment Provider metadata name
+///
+/// Root configs should specify both `prefix` and `name`:
+/// ```ignore
+/// #[derive(ConfigEnv)]
+/// #[config_env(prefix = "IGGY_", name = "iggy-server-config")]
+/// pub struct ServerConfig { ... }
+/// ```
+///
+/// This generates constants `ServerConfig::ENV_PREFIX` and `ServerConfig::ENV_PROVIDER_NAME`.
+///
+/// # Field Attributes
+/// - `#[config_env(name = "CUSTOM")]` - Override the env var segment name
+/// - `#[config_env(skip)]` - Exclude this field from env var mapping
+/// - `#[config_env(secret)]` - Mark field as secret (masked in logs)
+/// - `#[config_env(leaf)]` - Treat as value type, not nested config (for custom types like enums)
+///
+/// # Type Handling
+/// - **Primitives** (bool, u8-u128, i8-i128, f32, f64, String): automatically treated as leaves
+/// - **Non-primitives**: assumed to be nested configs that implement `ConfigEnvMappings`
+/// - Use `#[config_env(leaf)]` for value types that shouldn't be recursively expanded
+/// (e.g., `IggyDuration`, `CompressionAlgorithm`, custom enums)
+///
+/// # Examples
+///
+/// ## Struct Configuration
+/// ```ignore
+/// // Root config with prefix and name
+/// #[derive(ConfigEnv)]
+/// #[config_env(prefix = "IGGY_", name = "iggy-server-config")]
+/// pub struct ServerConfig {
+/// pub http: HttpConfig, // Nested: IGGY_HTTP_*
+/// #[config_env(leaf)]
+/// pub compression: CompressionAlgorithm, // Leaf: IGGY_COMPRESSION
+/// }
+///
+/// // Nested config - no prefix needed, paths deduced from field names
+/// #[derive(ConfigEnv)]
+/// pub struct HttpConfig {
+/// pub enabled: bool, // ENABLED -> parent adds HTTP_ -> IGGY_HTTP_ENABLED
+/// pub port: u16, // PORT -> IGGY_HTTP_PORT
+/// #[config_env(leaf)]
+/// pub timeout: IggyDuration, // Leaf: TIMEOUT -> IGGY_HTTP_TIMEOUT
+/// }
+/// ```
+///
+/// ## Enum Configuration
+/// ```ignore
+/// // Enums with newtype variants collect mappings from inner types
+/// #[derive(ConfigEnv)]
+/// #[serde(tag = "type", rename_all = "lowercase")]
+/// pub enum ConnectorConfig {
+/// Local(LocalConfig), // Includes LocalConfig's mappings
+/// Http(HttpConfig), // Includes HttpConfig's mappings
+/// }
+///
+/// // Simple value enums should NOT derive ConfigEnv - mark the field as leaf instead
+/// #[derive(Serialize, Deserialize)]
+/// pub enum Transport { Tcp, Quic, Http }
+///
+/// #[derive(ConfigEnv)]
+/// pub struct ServerConfig {
+/// #[config_env(leaf)] // Transport is a simple enum, not a nested config
+/// pub transport: Transport,
+/// }
+/// ```
+#[proc_macro_derive(ConfigEnv, attributes(config_env))]
+pub fn derive_config_env(input: TokenStream) -> TokenStream {
+ let input = parse_macro_input!(input as DeriveInput);
+ config_env::generate_impl(&input).into()
+}
diff --git a/core/connectors/runtime/Cargo.toml b/core/connectors/runtime/Cargo.toml
index dea88451a..970c1a7e8 100644
--- a/core/connectors/runtime/Cargo.toml
+++ b/core/connectors/runtime/Cargo.toml
@@ -34,6 +34,8 @@ axum = { workspace = true }
axum-server = { workspace = true }
chrono = { workspace = true }
clap = { workspace = true }
+configs = { workspace = true }
+configs_derive = { workspace = true }
dashmap = { workspace = true }
derive_more = { workspace = true }
dirs = { workspace = true }
diff --git a/core/connectors/runtime/src/api/config.rs b/core/connectors/runtime/src/api/config.rs
index 1e11791b8..8efdb8ae5 100644
--- a/core/connectors/runtime/src/api/config.rs
+++ b/core/connectors/runtime/src/api/config.rs
@@ -20,6 +20,7 @@
use crate::configs::connectors::ConfigFormat;
use crate::error::RuntimeError;
use axum::http::{HeaderValue, Method};
+use configs_derive::ConfigEnv;
use serde::{Deserialize, Serialize};
use std::fmt::Formatter;
use tower_http::cors::{AllowOrigin, CorsLayer};
@@ -30,16 +31,17 @@ pub const YAML_HEADER: HeaderValue = HeaderValue::from_static("application/yaml"
pub const TOML_HEADER: HeaderValue = HeaderValue::from_static("application/toml");
pub const TEXT_HEADER: HeaderValue = HeaderValue::from_static("text/plain");
-#[derive(Debug, Clone, Deserialize, Serialize)]
+#[derive(Debug, Clone, Deserialize, Serialize, ConfigEnv)]
pub struct HttpConfig {
pub enabled: bool,
pub address: String,
+ #[config_env(secret)]
pub api_key: String,
pub cors: HttpCorsConfig,
pub tls: HttpTlsConfig,
}
-#[derive(Debug, Default, Deserialize, Serialize, Clone)]
+#[derive(Debug, Default, Deserialize, Serialize, Clone, ConfigEnv)]
pub struct HttpCorsConfig {
pub enabled: bool,
pub allowed_methods: Vec<String>,
@@ -50,7 +52,7 @@ pub struct HttpCorsConfig {
pub allow_private_network: bool,
}
-#[derive(Debug, Default, Deserialize, Serialize, Clone)]
+#[derive(Debug, Default, Deserialize, Serialize, Clone, ConfigEnv)]
pub struct HttpTlsConfig {
pub enabled: bool,
pub cert_file: String,
diff --git a/core/connectors/runtime/src/configs/connectors.rs b/core/connectors/runtime/src/configs/connectors.rs
index 14dd0bdf8..3c111cfaa 100644
--- a/core/connectors/runtime/src/configs/connectors.rs
+++ b/core/connectors/runtime/src/configs/connectors.rs
@@ -26,6 +26,7 @@ use crate::configs::runtime::ConnectorsConfig as RuntimeConnectorsConfig;
use crate::error::RuntimeError;
use async_trait::async_trait;
use chrono::{DateTime, Utc};
+use configs_derive::ConfigEnv;
use iggy_connector_sdk::Schema;
use iggy_connector_sdk::transforms::TransformType;
use serde::{Deserialize, Serialize};
@@ -101,16 +102,19 @@ impl CreateSinkConfig {
}
}
-#[derive(Debug, Default, Clone, Serialize, Deserialize)]
+#[derive(Debug, Default, Clone, Serialize, Deserialize, ConfigEnv)]
pub struct SinkConfig {
pub key: String,
pub enabled: bool,
pub version: u64,
pub name: String,
pub path: String,
+ #[config_env(skip)]
pub transforms: Option<TransformsConfig>,
pub streams: Vec<StreamConsumerConfig>,
+ #[config_env(leaf)]
pub plugin_config_format: Option<ConfigFormat>,
+ #[config_env(skip)]
pub plugin_config: Option<serde_json::Value>,
#[serde(default)]
pub verbose: bool,
@@ -146,16 +150,19 @@ impl CreateSourceConfig {
}
}
-#[derive(Debug, Default, Clone, Serialize, Deserialize)]
+#[derive(Debug, Default, Clone, Serialize, Deserialize, ConfigEnv)]
pub struct SourceConfig {
pub key: String,
pub enabled: bool,
pub version: u64,
pub name: String,
pub path: String,
+ #[config_env(skip)]
pub transforms: Option<TransformsConfig>,
pub streams: Vec<StreamProducerConfig>,
+ #[config_env(leaf)]
pub plugin_config_format: Option<ConfigFormat>,
+ #[config_env(skip)]
pub plugin_config: Option<serde_json::Value>,
#[serde(default)]
pub verbose: bool,
@@ -167,20 +174,22 @@ pub struct TransformsConfig {
pub transforms: HashMap<TransformType, serde_json::Value>,
}
-#[derive(Debug, Default, Clone, Serialize, Deserialize)]
+#[derive(Debug, Default, Clone, Serialize, Deserialize, ConfigEnv)]
pub struct StreamConsumerConfig {
pub stream: String,
pub topics: Vec<String>,
+ #[config_env(leaf)]
pub schema: Schema,
pub batch_length: Option<u32>,
pub poll_interval: Option<String>,
pub consumer_group: Option<String>,
}
-#[derive(Debug, Default, Clone, Serialize, Deserialize)]
+#[derive(Debug, Default, Clone, Serialize, Deserialize, ConfigEnv)]
pub struct StreamProducerConfig {
pub stream: String,
pub topic: String,
+ #[config_env(leaf)]
pub schema: Schema,
pub batch_length: Option<u32>,
pub linger_time: Option<String>,
diff --git a/core/connectors/runtime/src/configs/connectors/local_provider.rs b/core/connectors/runtime/src/configs/connectors/local_provider.rs
index fb363d9c5..6d1d3b45d 100644
--- a/core/connectors/runtime/src/configs/connectors/local_provider.rs
+++ b/core/connectors/runtime/src/configs/connectors/local_provider.rs
@@ -22,12 +22,12 @@ use crate::configs::connectors::{
ConnectorsConfigProvider, CreateSinkConfig, CreateSourceConfig, SinkConfig, SourceConfig,
};
use crate::error::RuntimeError;
+use ::configs::{ConfigProvider, FileConfigProvider, TypedEnvProvider};
use async_trait::async_trait;
use chrono::{DateTime, Utc};
use dashmap::DashMap;
use figment::value::Dict;
use figment::{Metadata, Profile, Provider};
-use iggy_common::{ConfigProvider, CustomEnvProvider, FileConfigProvider};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::path::Path;
@@ -338,7 +338,7 @@ impl<S: ProviderState> LocalConnectorsConfigProvider<S> {
let field_path = &env_key_upper[prefix.len()..];
let field_name = field_path.to_lowercase();
- let parsed_value = iggy_common::parse_env_value_to_json(&env_value);
+ let parsed_value = ::configs::parse_env_value_to_json(&env_value);
let config = plugin_config.get_or_insert_with(|| serde_json::json!({}));
if let serde_json::Value::Object(map) = config {
@@ -783,9 +783,15 @@ impl ConnectorsConfigProvider for LocalConnectorsConfigProvider<Initialized> {
}
#[derive(Debug, Clone)]
-pub struct ConnectorEnvProvider {
- connector_name: String,
- provider: CustomEnvProvider<ConnectorConfig>,
+pub enum ConnectorEnvProvider {
+ Sink {
+ connector_name: String,
+ provider: TypedEnvProvider<SinkConfig>,
+ },
+ Source {
+ connector_name: String,
+ provider: TypedEnvProvider<SourceConfig>,
+ },
}
impl ConnectorEnvProvider {
@@ -793,24 +799,38 @@ impl ConnectorEnvProvider {
let connector_type = base_config.connector_type().to_uppercase();
let key = base_config.key().to_uppercase();
let prefix = format!("IGGY_CONNECTORS_{}_{}_", connector_type, key);
- Self {
- connector_name: base_config.key().to_owned(),
- provider: CustomEnvProvider::new(&prefix, &[]),
+ let connector_name = base_config.key().to_owned();
+
+ match base_config {
+ BaseConnectorConfig::Sink { .. } => Self::Sink {
+ connector_name,
+ provider: TypedEnvProvider::with_runtime_prefix(&prefix, &[]),
+ },
+ BaseConnectorConfig::Source { .. } => Self::Source {
+ connector_name,
+ provider: TypedEnvProvider::with_runtime_prefix(&prefix, &[]),
+ },
}
}
}
impl Provider for ConnectorEnvProvider {
fn metadata(&self) -> Metadata {
- Metadata::named(format!("iggy-connectors-{}-config", self.connector_name))
+ let name = match self {
+ Self::Sink { connector_name, .. } => connector_name,
+ Self::Source { connector_name, .. } => connector_name,
+ };
+ Metadata::named(format!("iggy-connectors-{}-config", name))
}
fn data(&self) -> Result<figment::value::Map<Profile, Dict>, figment::Error> {
- self.provider.deserialize().map_err(|_| {
- figment::Error::from(format!(
- "Cannot deserialize environment variables for connector config {}",
- self.connector_name
- ))
- })
+ match self {
+ Self::Sink { provider, .. } => provider
+ .deserialize_with_runtime_prefix()
+ .map_err(|e| figment::Error::from(format!("Failed to deserialize env vars: {e}"))),
+ Self::Source { provider, .. } => provider
+ .deserialize_with_runtime_prefix()
+ .map_err(|e| figment::Error::from(format!("Failed to deserialize env vars: {e}"))),
+ }
}
}
diff --git a/core/connectors/runtime/src/configs/runtime.rs b/core/connectors/runtime/src/configs/runtime.rs
index 77d88ef07..edcf9fcca 100644
--- a/core/connectors/runtime/src/configs/runtime.rs
+++ b/core/connectors/runtime/src/configs/runtime.rs
@@ -18,24 +18,21 @@
*/
use crate::api::config::HttpConfig;
+use ::configs::{FileConfigProvider, TypedEnvProvider};
+use configs_derive::ConfigEnv;
use derive_more::Display;
use figment::providers::{Format, Toml};
use figment::value::Dict;
use figment::{Metadata, Profile, Provider};
+use iggy_common::IggyDuration;
use iggy_common::defaults::{DEFAULT_ROOT_PASSWORD, DEFAULT_ROOT_USERNAME};
-use iggy_common::{CustomEnvProvider, FileConfigProvider, IggyDuration};
use serde::{Deserialize, Serialize};
use serde_with::{DisplayFromStr, serde_as};
use std::collections::HashMap;
use std::fmt::{Display, Formatter};
use std::str::FromStr;
-const SECRET_KEYS: [&str; 2] = [
- "IGGY_CONNECTORS_IGGY_PASSWORD",
- "IGGY_CONNECTORS_IGGY_TOKEN",
-];
-
-#[derive(Debug, Deserialize, Serialize, Clone)]
+#[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)]
pub struct TelemetryConfig {
pub enabled: bool,
pub service_name: String,
@@ -64,8 +61,9 @@ impl Display for TelemetryConfig {
}
}
-#[derive(Debug, Deserialize, Serialize, Clone)]
+#[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)]
pub struct TelemetryLogsConfig {
+ #[config_env(leaf)]
pub transport: TelemetryTransport,
pub endpoint: String,
}
@@ -89,8 +87,9 @@ impl Display for TelemetryLogsConfig {
}
}
-#[derive(Debug, Deserialize, Serialize, Clone)]
+#[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)]
pub struct TelemetryTracesConfig {
+ #[config_env(leaf)]
pub transport: TelemetryTransport,
pub endpoint: String,
}
@@ -134,7 +133,8 @@ impl FromStr for TelemetryTransport {
}
}
-#[derive(Debug, Default, Clone, Deserialize, Serialize)]
+#[derive(Debug, Default, Clone, Deserialize, Serialize, ConfigEnv)]
+#[config_env(prefix = "IGGY_CONNECTORS_", name = "iggy-connectors-config")]
#[serde(default)]
pub struct ConnectorsRuntimeConfig {
pub http: HttpConfig,
@@ -144,16 +144,18 @@ pub struct ConnectorsRuntimeConfig {
pub telemetry: TelemetryConfig,
}
-#[derive(Debug, Clone, Serialize, Deserialize)]
+#[derive(Debug, Clone, Serialize, Deserialize, ConfigEnv)]
pub struct IggyConfig {
pub address: String,
pub username: String,
+ #[config_env(secret)]
pub password: String,
+ #[config_env(secret)]
pub token: String,
pub tls: IggyTlsConfig,
}
-#[derive(Debug, Default, Clone, Serialize, Deserialize)]
+#[derive(Debug, Default, Clone, Serialize, Deserialize, ConfigEnv)]
pub struct IggyTlsConfig {
pub enabled: bool,
pub ca_file: String,
@@ -161,12 +163,14 @@ pub struct IggyTlsConfig {
}
#[serde_as]
-#[derive(Debug, Clone, Serialize, Deserialize)]
+#[derive(Debug, Clone, Serialize, Deserialize, ConfigEnv)]
pub struct RetryConfig {
pub enabled: bool,
pub max_attempts: u32,
+ #[config_env(leaf)]
#[serde_as(as = "DisplayFromStr")]
pub initial_backoff: IggyDuration,
+ #[config_env(leaf)]
#[serde_as(as = "DisplayFromStr")]
pub max_backoff: IggyDuration,
pub backoff_multiplier: u32,
@@ -198,21 +202,24 @@ impl Display for RetryConfig {
}
}
-#[derive(Debug, Default, Clone, Deserialize, Serialize)]
+#[derive(Debug, Default, Clone, Deserialize, Serialize, ConfigEnv)]
#[serde(default)]
pub struct LocalConnectorsConfig {
pub config_dir: String,
}
#[serde_as]
-#[derive(Debug, Default, Clone, Deserialize, Serialize)]
+#[derive(Debug, Default, Clone, Deserialize, Serialize, ConfigEnv)]
pub struct HttpConnectorsConfig {
pub base_url: String,
+ #[config_env(leaf)]
#[serde_as(as = "DisplayFromStr")]
#[serde(default = "default_from_secs")]
pub timeout: IggyDuration,
+ #[config_env(skip)]
#[serde(default)]
pub request_headers: HashMap<String, String>,
+ #[config_env(skip)]
#[serde(default)]
pub url_templates: HashMap<String, String>,
#[serde(default)]
@@ -240,7 +247,7 @@ fn default_from_secs() -> IggyDuration {
IggyDuration::new_from_secs(10)
}
-#[derive(Debug, Default, Clone, Deserialize, Serialize)]
+#[derive(Debug, Default, Clone, Deserialize, Serialize, ConfigEnv)]
#[serde(default)]
pub struct ResponseConfig {
pub data_path: Option<String>,
@@ -248,7 +255,7 @@ pub struct ResponseConfig {
}
#[allow(clippy::large_enum_variant)]
-#[derive(Debug, Clone, Deserialize, Serialize)]
+#[derive(Debug, Clone, Deserialize, Serialize, ConfigEnv)]
#[serde(tag = "config_type", rename_all = "lowercase")]
pub enum ConnectorsConfig {
Local(LocalConnectorsConfig),
@@ -261,7 +268,7 @@ impl Default for ConnectorsConfig {
}
}
-#[derive(Debug, Clone, Serialize, Deserialize)]
+#[derive(Debug, Clone, Serialize, Deserialize, ConfigEnv)]
pub struct StateConfig {
pub path: String,
}
@@ -358,25 +365,23 @@ impl ConnectorsRuntimeConfig {
#[derive(Debug, Clone)]
pub struct ConnectorsEnvProvider {
- provider: CustomEnvProvider<ConnectorsRuntimeConfig>,
+ provider: TypedEnvProvider<ConnectorsRuntimeConfig>,
}
impl Default for ConnectorsEnvProvider {
fn default() -> Self {
Self {
- provider: CustomEnvProvider::new("IGGY_CONNECTORS_", &SECRET_KEYS),
+ provider: TypedEnvProvider::from_config(ConnectorsRuntimeConfig::ENV_PREFIX),
}
}
}
impl Provider for ConnectorsEnvProvider {
fn metadata(&self) -> Metadata {
- Metadata::named("iggy-connectors-config")
+ Metadata::named(ConnectorsRuntimeConfig::ENV_PROVIDER_NAME)
}
fn data(&self) -> Result<figment::value::Map<Profile, Dict>, figment::Error> {
- self.provider.deserialize().map_err(|_| {
- figment::Error::from("Cannot deserialize environment variables for connectors config")
- })
+ self.provider.data()
}
}
diff --git a/core/connectors/runtime/src/main.rs b/core/connectors/runtime/src/main.rs
index ab4a77208..657ed646a 100644
--- a/core/connectors/runtime/src/main.rs
+++ b/core/connectors/runtime/src/main.rs
@@ -18,6 +18,7 @@
*/
use crate::configs::connectors::{ConnectorsConfigProvider, create_connectors_config_provider};
+use ::configs::ConfigProvider;
use clap::Parser;
use configs::connectors::ConfigFormat;
use configs::runtime::ConnectorsRuntimeConfig;
@@ -26,7 +27,6 @@ use dotenvy::dotenv;
use error::RuntimeError;
use figlet_rs::FIGfont;
use iggy::prelude::{Client, IggyConsumer, IggyProducer};
-use iggy_common::ConfigProvider;
use iggy_connector_sdk::{
StreamDecoder, StreamEncoder,
sink::ConsumeCallback,
diff --git a/core/integration/Cargo.toml b/core/integration/Cargo.toml
index 87b95b6f1..3e9cb5376 100644
--- a/core/integration/Cargo.toml
+++ b/core/integration/Cargo.toml
@@ -33,6 +33,7 @@ async-trait = { workspace = true }
bytes = { workspace = true }
chrono = { workspace = true }
compio = { workspace = true }
+configs = { workspace = true }
ctor = "0.6.3"
derive_more = { workspace = true }
env_logger = { workspace = true }
diff --git a/core/integration/src/test_connectors_runtime.rs b/core/integration/src/test_connectors_runtime.rs
index 995feb69f..b7adb73ef 100644
--- a/core/integration/src/test_connectors_runtime.rs
+++ b/core/integration/src/test_connectors_runtime.rs
@@ -31,7 +31,6 @@ use uuid::Uuid;
pub const STATE_PATH_ENV_VAR: &str = "IGGY_CONNECTORS_STATE_PATH";
pub const TEST_VERBOSITY_ENV_VAR: &str = "IGGY_TEST_VERBOSE";
-pub const CONSUMER_NAME: &str = "connectors";
const LOCAL_STATE_PREFIX: &str = "local_state_";
#[derive(Debug)]
@@ -70,10 +69,6 @@ impl TestConnectorsRuntime {
"IGGY_CONNECTORS_IGGY_ADDRESS".to_string(),
iggy_tcp_server_address.to_string(),
);
- envs.insert(
- "IGGY_CONNECTORS_IGGY_CONSUMER".to_string(),
- CONSUMER_NAME.to_string(),
- );
Self::create(envs, server_executable_path)
}
diff --git a/core/integration/src/test_server.rs b/core/integration/src/test_server.rs
index 99846d4ad..a58ad85a3 100644
--- a/core/integration/src/test_server.rs
+++ b/core/integration/src/test_server.rs
@@ -19,11 +19,12 @@
use assert_cmd::prelude::CommandCargoExt;
use async_trait::async_trait;
+use configs::ConfigProvider;
use derive_more::Display;
use futures::executor::block_on;
use iggy::prelude::UserStatus::Active;
use iggy::prelude::*;
-use iggy_common::{ConfigProvider, TransportProtocol};
+use iggy_common::TransportProtocol;
use rand::Rng;
use server::configs::server::ServerConfig;
use std::collections::HashMap;
diff --git a/core/integration/tests/config_provider/mod.rs b/core/integration/tests/config_provider/mod.rs
index 82e039732..a705284f4 100644
--- a/core/integration/tests/config_provider/mod.rs
+++ b/core/integration/tests/config_provider/mod.rs
@@ -17,7 +17,7 @@
* under the License.
*/
-use iggy_common::ConfigProvider;
+use configs::ConfigProvider;
use integration::file::get_root_path;
use serial_test::serial;
use server::configs::server::ServerConfig;
@@ -44,13 +44,13 @@ async fn validate_config_env_override() {
);
}
- let config_path = get_root_path().join("../configs/server.toml");
+ let config_path = get_root_path().join("../server/config.toml");
let file_config_provider =
ServerConfig::config_provider(&config_path.as_path().display().to_string());
let config: ServerConfig = file_config_provider
.load_config()
.await
- .expect("Failed to load server.toml config");
+ .expect("Failed to load config.toml config");
assert_eq!(config.http.enabled, expected_http);
assert_eq!(config.tcp.enabled, expected_tcp);
@@ -86,13 +86,13 @@ async fn validate_socket_override() {
);
}
- let config_path = get_root_path().join("../configs/server.toml");
+ let config_path = get_root_path().join("../server/config.toml");
let file_config_provider =
ServerConfig::config_provider(&config_path.as_path().display().to_string());
let config: ServerConfig = file_config_provider
.load_config()
.await
- .expect("Failed to load server.toml config with socket override");
+ .expect("Failed to load config.toml config with socket override");
assert!(config.tcp.socket.override_defaults);
// Verify the buffer sizes match the expected byte counts
@@ -115,13 +115,13 @@ async fn validate_socket_override() {
#[serial]
#[tokio::test]
async fn validate_socket_no_override() {
- let config_path = get_root_path().join("../configs/server.toml");
+ let config_path = get_root_path().join("../server/config.toml");
let file_config_provider =
ServerConfig::config_provider(&config_path.as_path().display().to_string());
let config: ServerConfig = file_config_provider
.load_config()
.await
- .expect("Failed to load server.toml config without socket override");
+ .expect("Failed to load config.toml config without socket override");
assert!(!config.tcp.socket.override_defaults);
}
@@ -201,13 +201,13 @@ async fn validate_cluster_config_env_override() {
);
}
- let config_path = get_root_path().join("../configs/server.toml");
+ let config_path = get_root_path().join("../server/config.toml");
let file_config_provider =
ServerConfig::config_provider(&config_path.as_path().display().to_string());
let config: ServerConfig = file_config_provider
.load_config()
.await
- .expect("Failed to load server.toml config with cluster env overrides");
+ .expect("Failed to load config.toml config with cluster env overrides");
// Verify cluster configuration
assert_eq!(config.cluster.enabled, expected_cluster_enabled);
@@ -270,7 +270,6 @@ async fn validate_cluster_config_env_override() {
unsafe {
// Clean up environment variables
env::remove_var("IGGY_CLUSTER_ENABLED");
- env::remove_var("IGGY_CLUSTER_ID");
env::remove_var("IGGY_CLUSTER_NAME");
env::remove_var("IGGY_CLUSTER_NODE_CURRENT_NAME");
@@ -392,13 +391,13 @@ async fn validate_four_node_cluster_config_env_override() {
// IGGY_CLUSTER_NODE_OTHERS_2_PORTS_WEBSOCKET is NOT set - should use default (8092)
}
- let config_path = get_root_path().join("../configs/server.toml");
+ let config_path = get_root_path().join("../server/config.toml");
let file_config_provider =
ServerConfig::config_provider(&config_path.as_path().display().to_string());
let config: ServerConfig = file_config_provider
.load_config()
.await
- .expect("Failed to load server.toml config with 4-node cluster env overrides");
+ .expect("Failed to load config.toml config with 4-node cluster env overrides");
// Verify cluster configuration
assert_eq!(config.cluster.enabled, expected_cluster_enabled);
diff --git a/core/integration/tests/data_integrity/verify_after_server_restart.rs b/core/integration/tests/data_integrity/verify_after_server_restart.rs
index ac63d6e71..655fc8246 100644
--- a/core/integration/tests/data_integrity/verify_after_server_restart.rs
+++ b/core/integration/tests/data_integrity/verify_after_server_restart.rs
@@ -52,7 +52,7 @@ async fn should_fill_data_and_verify_after_restart(cache_setting: &'static str)
TestServer::get_random_path(),
),
(
- "IGGY_SEGMENT_CACHE_INDEXES".to_string(),
+ "IGGY_SYSTEM_SEGMENT_CACHE_INDEXES".to_string(),
cache_setting.to_string(),
),
]);
diff --git a/core/server/Cargo.toml b/core/server/Cargo.toml
index 3e4e7192f..d311bc870 100644
--- a/core/server/Cargo.toml
+++ b/core/server/Cargo.toml
@@ -53,6 +53,7 @@ compio-net = { workspace = true }
compio-quic = { workspace = true }
compio-tls = { workspace = true }
compio-ws = { workspace = true }
+configs = { workspace = true }
ctrlc = { version = "3.5", features = ["termination"] }
cyper = { workspace = true }
cyper-axum = { workspace = true }
diff --git a/core/server/README.md b/core/server/README.md
index 5608b4e0d..a57c62401 100644
--- a/core/server/README.md
+++ b/core/server/README.md
@@ -2,7 +2,7 @@
This is the core server component of Apache Iggy. You can run it directly with `cargo run --bin iggy-server --release` or use the Docker image `apache/iggy:latest` (the `edge` tag is for the latest development version).
-The configuration file is located at [core/configs/server.toml](https://github.com/apache/iggy/blob/master/core/configs/server.toml). You can customize the server settings by modifying this file or by using environment variables e.g. `IGGY_TCP_ADDRESS=0.0.0.0:8090`.
+The configuration file is located at [core/server/config.toml](https://github.com/apache/iggy/blob/master/core/server/config.toml). You can customize the server settings by modifying this file or by using environment variables e.g. `IGGY_TCP_ADDRESS=0.0.0.0:8090`.
diff --git a/core/configs/server.toml b/core/server/config.toml
similarity index 100%
rename from core/configs/server.toml
rename to core/server/config.toml
diff --git a/core/server/src/args.rs b/core/server/src/args.rs
index 8ec1dc09d..13451f411 100644
--- a/core/server/src/args.rs
+++ b/core/server/src/args.rs
@@ -38,12 +38,12 @@ DOCUMENTATION:
https://iggy.apache.org/docs
CONFIGURATION:
- The server uses a TOML configuration file. By default, it looks for 'configs/server.toml'
+ The server uses a TOML configuration file. By default, it looks for 'core/server/config.toml'
in the current working directory. You can override this with the IGGY_CONFIG_PATH environment
variable or use the --config-provider flag.
Examples:
- iggy-server # Uses default file provider (configs/server.toml)
+ iggy-server # Uses default file provider (core/server/config.toml)
iggy-server --config-provider file # Explicitly use file provider
IGGY_CONFIG_PATH=custom.toml iggy-server # Use custom config file path
diff --git a/core/server/src/configs/cluster.rs b/core/server/src/configs/cluster.rs
index 7e3893650..8468c3a81 100644
--- a/core/server/src/configs/cluster.rs
+++ b/core/server/src/configs/cluster.rs
@@ -16,35 +16,36 @@
* under the License.
*/
+use iggy_common::ConfigEnv;
use serde::{Deserialize, Serialize};
-#[derive(Debug, Deserialize, Serialize, Clone)]
+#[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)]
pub struct ClusterConfig {
pub enabled: bool,
pub name: String,
pub node: NodeConfig,
}
-#[derive(Debug, Deserialize, Serialize, Clone)]
+#[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)]
pub struct NodeConfig {
pub current: CurrentNodeConfig,
pub others: Vec<OtherNodeConfig>,
}
-#[derive(Debug, Deserialize, Serialize, Clone)]
+#[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)]
pub struct CurrentNodeConfig {
pub name: String,
pub ip: String,
}
-#[derive(Debug, Deserialize, Serialize, Clone)]
+#[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)]
pub struct OtherNodeConfig {
pub name: String,
pub ip: String,
pub ports: TransportPorts,
}
-#[derive(Debug, Deserialize, Serialize, Clone, Default)]
+#[derive(Debug, Deserialize, Serialize, Clone, Default, ConfigEnv)]
pub struct TransportPorts {
pub tcp: Option<u16>,
pub quic: Option<u16>,
diff --git a/core/server/src/configs/defaults.rs b/core/server/src/configs/defaults.rs
index f6d9d83e5..af283442e 100644
--- a/core/server/src/configs/defaults.rs
+++ b/core/server/src/configs/defaults.rs
@@ -42,8 +42,8 @@ use std::sync::Arc;
use std::time::Duration;
static_toml::static_toml! {
- // static_toml crate always starts from CARGO_MANIFEST_DIR (in this case iggy-server root directory)
- pub static SERVER_CONFIG = include_toml!("../configs/server.toml");
+ // static_toml crate always starts from CARGO_MANIFEST_DIR (core/server)
+ pub static SERVER_CONFIG = include_toml!("config.toml");
}
impl Default for ServerConfig {
diff --git a/core/server/src/configs/http.rs b/core/server/src/configs/http.rs
index 770632170..b21720cfb 100644
--- a/core/server/src/configs/http.rs
+++ b/core/server/src/configs/http.rs
@@ -16,6 +16,7 @@
* under the License.
*/
+use iggy_common::ConfigEnv;
use iggy_common::IggyByteSize;
use iggy_common::IggyDuration;
use iggy_common::IggyError;
@@ -25,10 +26,11 @@ use serde::{Deserialize, Serialize};
use serde_with::DisplayFromStr;
use serde_with::serde_as;
-#[derive(Debug, Deserialize, Serialize, Clone)]
+#[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)]
pub struct HttpConfig {
pub enabled: bool,
pub address: String,
+ #[config_env(leaf)]
pub max_request_size: IggyByteSize,
pub web_ui: bool,
pub cors: HttpCorsConfig,
@@ -37,7 +39,7 @@ pub struct HttpConfig {
pub tls: HttpTlsConfig,
}
-#[derive(Debug, Deserialize, Serialize, Clone)]
+#[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)]
pub struct HttpCorsConfig {
pub enabled: bool,
pub allowed_methods: Vec<String>,
@@ -49,25 +51,30 @@ pub struct HttpCorsConfig {
}
#[serde_as]
-#[derive(Debug, Deserialize, Serialize, Clone)]
+#[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)]
pub struct HttpJwtConfig {
pub algorithm: String,
pub issuer: String,
pub audience: String,
pub valid_issuers: Vec<String>,
pub valid_audiences: Vec<String>,
+ #[config_env(leaf)]
#[serde_as(as = "DisplayFromStr")]
pub access_token_expiry: IggyExpiry,
+ #[config_env(leaf)]
#[serde_as(as = "DisplayFromStr")]
pub clock_skew: IggyDuration,
+ #[config_env(leaf)]
#[serde_as(as = "DisplayFromStr")]
pub not_before: IggyDuration,
+ #[config_env(secret)]
pub encoding_secret: String,
+ #[config_env(secret)]
pub decoding_secret: String,
pub use_base64_secret: bool,
}
-#[derive(Debug, Deserialize, Serialize, Clone)]
+#[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)]
pub struct HttpMetricsConfig {
pub enabled: bool,
pub endpoint: String,
@@ -79,7 +86,7 @@ pub enum JwtSecret {
Base64(String),
}
-#[derive(Debug, Deserialize, Serialize, Clone)]
+#[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)]
pub struct HttpTlsConfig {
pub enabled: bool,
pub cert_file: String,
diff --git a/core/server/src/configs/quic.rs b/core/server/src/configs/quic.rs
index f2cedd6de..ee4c28c12 100644
--- a/core/server/src/configs/quic.rs
+++ b/core/server/src/configs/quic.rs
@@ -16,6 +16,7 @@
* under the License.
*/
+use iggy_common::ConfigEnv;
use iggy_common::IggyByteSize;
use iggy_common::IggyDuration;
use serde::{Deserialize, Serialize};
@@ -23,32 +24,40 @@ use serde_with::DisplayFromStr;
use serde_with::serde_as;
#[serde_as]
-#[derive(Debug, Deserialize, Serialize, Clone)]
+#[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)]
pub struct QuicConfig {
pub enabled: bool,
pub address: String,
pub max_concurrent_bidi_streams: u64,
+ #[config_env(leaf)]
pub datagram_send_buffer_size: IggyByteSize,
+ #[config_env(leaf)]
pub initial_mtu: IggyByteSize,
+ #[config_env(leaf)]
pub send_window: IggyByteSize,
+ #[config_env(leaf)]
pub receive_window: IggyByteSize,
+ #[config_env(leaf)]
#[serde_as(as = "DisplayFromStr")]
pub keep_alive_interval: IggyDuration,
+ #[config_env(leaf)]
#[serde_as(as = "DisplayFromStr")]
pub max_idle_timeout: IggyDuration,
pub certificate: QuicCertificateConfig,
pub socket: QuicSocketConfig,
}
-#[derive(Debug, Deserialize, Serialize, Clone)]
+#[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)]
pub struct QuicSocketConfig {
pub override_defaults: bool,
+ #[config_env(leaf)]
pub recv_buffer_size: IggyByteSize,
+ #[config_env(leaf)]
pub send_buffer_size: IggyByteSize,
pub keepalive: bool,
}
-#[derive(Debug, Deserialize, Serialize, Clone)]
+#[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)]
pub struct QuicCertificateConfig {
pub self_signed: bool,
pub cert_file: String,
diff --git a/core/server/src/configs/server.rs b/core/server/src/configs/server.rs
index 70008c4f7..f3fe593ce 100644
--- a/core/server/src/configs/server.rs
+++ b/core/server/src/configs/server.rs
@@ -16,7 +16,6 @@
* specific language governing permissions and limitations
* under the License.
*/
-use crate::IGGY_ROOT_PASSWORD_ENV;
use crate::configs::COMPONENT;
use crate::configs::cluster::ClusterConfig;
use crate::configs::http::HttpConfig;
@@ -25,13 +24,13 @@ use crate::configs::system::SystemConfig;
use crate::configs::tcp::TcpConfig;
use crate::configs::websocket::WebSocketConfig;
use crate::server_error::ConfigurationError;
+use configs::{ConfigEnvMappings, ConfigProvider, FileConfigProvider, TypedEnvProvider};
use derive_more::Display;
use err_trail::ErrContext;
use figment::providers::{Format, Toml};
use figment::value::Dict;
use figment::{Metadata, Profile, Provider};
-use iggy_common::{ConfigProvider, IggyByteSize, MemoryPoolConfigOther, Validatable};
-use iggy_common::{CustomEnvProvider, FileConfigProvider, IggyDuration};
+use iggy_common::{ConfigEnv, IggyByteSize, IggyDuration, MemoryPoolConfigOther, Validatable};
use serde::{Deserialize, Serialize};
use serde_with::DisplayFromStr;
use serde_with::serde_as;
@@ -39,17 +38,10 @@ use std::env;
use std::str::FromStr;
use std::sync::Arc;
-const DEFAULT_CONFIG_PATH: &str = "configs/server.toml";
-const SECRET_KEYS: [&str; 6] = [
- IGGY_ROOT_PASSWORD_ENV,
- "IGGY_DATA_MAINTENANCE_ARCHIVER_S3_KEY_SECRET",
- "IGGY_HTTP_JWT_ENCODING_SECRET",
- "IGGY_HTTP_JWT_DECODING_SECRET",
- "IGGY_TCP_TLS_PASSWORD",
- "IGGY_SYSTEM_ENCRYPTION_KEY",
-];
-
-#[derive(Debug, Deserialize, Serialize, Clone)]
+const DEFAULT_CONFIG_PATH: &str = "core/server/config.toml";
+
+#[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)]
+#[config_env(prefix = "IGGY_", name = "iggy-server-config")]
pub struct ServerConfig {
pub data_maintenance: DataMaintenanceConfig,
pub message_saver: MessageSaverConfig,
@@ -65,13 +57,11 @@ pub struct ServerConfig {
}
/// Configuration for the memory pool.
-#[derive(Debug, Deserialize, Serialize)]
+#[derive(Debug, Deserialize, Serialize, ConfigEnv)]
pub struct MemoryPoolConfig {
- /// Whether the pool is enabled.
pub enabled: bool,
- /// Maximum size of the pool.
+ #[config_env(leaf)]
pub size: IggyByteSize,
- /// Maximum number of buffers per bucket.
pub bucket_capacity: u32,
}
@@ -87,51 +77,55 @@ impl MemoryPoolConfig {
}
#[serde_as]
-#[derive(Debug, Default, Deserialize, Serialize, Clone)]
+#[derive(Debug, Default, Deserialize, Serialize, Clone, ConfigEnv)]
pub struct DataMaintenanceConfig {
pub messages: MessagesMaintenanceConfig,
}
#[serde_as]
-#[derive(Debug, Deserialize, Serialize, Clone)]
+#[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)]
pub struct MessagesMaintenanceConfig {
pub cleaner_enabled: bool,
+ #[config_env(leaf)]
#[serde_as(as = "DisplayFromStr")]
pub interval: IggyDuration,
}
#[serde_as]
-#[derive(Debug, Deserialize, Serialize, Clone)]
+#[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)]
pub struct MessageSaverConfig {
pub enabled: bool,
pub enforce_fsync: bool,
+ #[config_env(leaf)]
#[serde_as(as = "DisplayFromStr")]
pub interval: IggyDuration,
}
-#[derive(Debug, Deserialize, Serialize, Clone)]
+#[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)]
pub struct PersonalAccessTokenConfig {
pub max_tokens_per_user: u32,
pub cleaner: PersonalAccessTokenCleanerConfig,
}
#[serde_as]
-#[derive(Debug, Deserialize, Serialize, Clone)]
+#[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)]
pub struct PersonalAccessTokenCleanerConfig {
pub enabled: bool,
+ #[config_env(leaf)]
#[serde_as(as = "DisplayFromStr")]
pub interval: IggyDuration,
}
#[serde_as]
-#[derive(Debug, Deserialize, Serialize, Clone)]
+#[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)]
pub struct HeartbeatConfig {
pub enabled: bool,
+ #[config_env(leaf)]
#[serde_as(as = "DisplayFromStr")]
pub interval: IggyDuration,
}
-#[derive(Debug, Deserialize, Serialize, Clone)]
+#[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)]
pub struct TelemetryConfig {
pub enabled: bool,
pub service_name: String,
@@ -139,14 +133,16 @@ pub struct TelemetryConfig {
pub traces: TelemetryTracesConfig,
}
-#[derive(Debug, Deserialize, Serialize, Clone)]
+#[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)]
pub struct TelemetryLogsConfig {
+ #[config_env(leaf)]
pub transport: TelemetryTransport,
pub endpoint: String,
}
-#[derive(Debug, Deserialize, Serialize, Clone)]
+#[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)]
pub struct TelemetryTracesConfig {
+ #[config_env(leaf)]
pub transport: TelemetryTransport,
pub endpoint: String,
}
@@ -172,6 +168,9 @@ impl FromStr for TelemetryTransport {
}
impl ServerConfig {
+ /// Load server configuration from file and environment variables.
+ ///
+ /// Uses compile-time generated env var mappings for unambiguous resolution.
pub async fn load() -> Result<ServerConfig, ConfigurationError> {
let config_path =
env::var("IGGY_CONFIG_PATH").unwrap_or_else(|_| DEFAULT_CONFIG_PATH.to_string());
@@ -180,19 +179,20 @@ impl ServerConfig {
config_provider
.load_config()
.await
- .error(|e: &iggy_common::ConfigurationError| {
- format!("{COMPONENT} (error: {e}) - failed to load config provider config")
+ .error(|e: &configs::ConfigurationError| {
+ format!("{COMPONENT} (error: {e}) - failed to load config")
})?;
server_config
.validate()
- .error(|e: &iggy_common::ConfigurationError| {
+ .error(|e: &configs::ConfigurationError| {
format!("{COMPONENT} (error: {e}) - failed to validate server config")
})?;
Ok(server_config)
}
+ /// Create a config provider using compile-time generated env var mappings.
pub fn config_provider(config_path: &str) -> FileConfigProvider<ServerConfigEnvProvider> {
- let default_config = Toml::string(include_str!("../../../configs/server.toml"));
+ let default_config = Toml::string(include_str!("../../config.toml"));
FileConfigProvider::new(
config_path.to_string(),
ServerConfigEnvProvider::default(),
@@ -200,29 +200,40 @@ impl ServerConfig {
Some(default_config),
)
}
+
+ /// Returns all valid environment variable names for ServerConfig.
+ pub fn all_env_var_names() -> Vec<&'static str> {
+ <ServerConfig as ConfigEnvMappings>::all_env_var_names()
+ }
}
+/// Type-safe environment provider using compile-time generated mappings.
+///
+/// Uses the `ConfigEnvMappings` trait generated by `#[derive(ConfigEnv)]`
+/// to directly look up known environment variable names, eliminating path ambiguity.
#[derive(Debug, Clone)]
pub struct ServerConfigEnvProvider {
- provider: CustomEnvProvider<ServerConfig>,
+ provider: TypedEnvProvider<ServerConfig>,
}
impl Default for ServerConfigEnvProvider {
fn default() -> Self {
Self {
- provider: CustomEnvProvider::new("IGGY_", &SECRET_KEYS),
+ provider: TypedEnvProvider::from_config(ServerConfig::ENV_PREFIX),
}
}
}
impl Provider for ServerConfigEnvProvider {
fn metadata(&self) -> Metadata {
- Metadata::named("iggy-server config")
+ Metadata::named(ServerConfig::ENV_PROVIDER_NAME)
}
fn data(&self) -> Result<figment::value::Map<Profile, Dict>, figment::Error> {
- self.provider.deserialize().map_err(|_| {
- figment::Error::from("Cannot deserialize environment variables for server config")
+ self.provider.deserialize().map_err(|e| {
+ figment::Error::from(format!(
+ "Cannot deserialize environment variables for server config: {e}"
+ ))
})
}
}
diff --git a/core/server/src/configs/sharding.rs b/core/server/src/configs/sharding.rs
index 7891b8161..5b9d28718 100644
--- a/core/server/src/configs/sharding.rs
+++ b/core/server/src/configs/sharding.rs
@@ -31,10 +31,12 @@ use std::thread::available_parallelism;
use tracing::info;
use crate::server_error::ServerError;
+use iggy_common::ConfigEnv;
-#[derive(Debug, Deserialize, Serialize, Default)]
+#[derive(Debug, Deserialize, Serialize, Default, ConfigEnv)]
pub struct ShardingConfig {
#[serde(default)]
+ #[config_env(leaf)]
pub cpu_allocation: CpuAllocation,
}
diff --git a/core/server/src/configs/system.rs b/core/server/src/configs/system.rs
index cc45ba59d..35ba9ed55 100644
--- a/core/server/src/configs/system.rs
+++ b/core/server/src/configs/system.rs
@@ -19,6 +19,7 @@
use super::cache_indexes::CacheIndexesConfig;
use super::sharding::ShardingConfig;
use crate::configs::server::MemoryPoolConfig;
+use iggy_common::ConfigEnv;
use iggy_common::IggyByteSize;
use iggy_common::IggyError;
use iggy_common::IggyExpiry;
@@ -31,7 +32,7 @@ use serde_with::serde_as;
pub const INDEX_EXTENSION: &str = "index";
pub const LOG_EXTENSION: &str = "log";
-#[derive(Debug, Deserialize, Serialize)]
+#[derive(Debug, Deserialize, Serialize, ConfigEnv)]
pub struct SystemConfig {
pub path: String,
pub backup: BackupConfig,
@@ -50,104 +51,116 @@ pub struct SystemConfig {
pub sharding: ShardingConfig,
}
-#[derive(Debug, Deserialize, Serialize)]
+#[derive(Debug, Deserialize, Serialize, ConfigEnv)]
pub struct BackupConfig {
pub path: String,
pub compatibility: CompatibilityConfig,
}
-#[derive(Debug, Deserialize, Serialize)]
+#[derive(Debug, Deserialize, Serialize, ConfigEnv)]
pub struct CompatibilityConfig {
pub path: String,
}
-#[derive(Debug, Deserialize, Serialize)]
+#[derive(Debug, Deserialize, Serialize, ConfigEnv)]
pub struct DatabaseConfig {
pub path: String,
}
-#[derive(Debug, Deserialize, Serialize)]
+#[derive(Debug, Deserialize, Serialize, ConfigEnv)]
pub struct RuntimeConfig {
pub path: String,
}
-#[derive(Debug, Deserialize, Serialize)]
+#[derive(Debug, Deserialize, Serialize, ConfigEnv)]
pub struct CompressionConfig {
pub allow_override: bool,
+ #[config_env(leaf)]
pub default_algorithm: CompressionAlgorithm,
}
#[serde_as]
-#[derive(Debug, Deserialize, Serialize)]
+#[derive(Debug, Deserialize, Serialize, ConfigEnv)]
pub struct LoggingConfig {
pub path: String,
pub level: String,
pub file_enabled: bool,
+ #[config_env(leaf)]
pub max_size: IggyByteSize,
+ #[config_env(leaf)]
#[serde_as(as = "DisplayFromStr")]
pub retention: IggyDuration,
+ #[config_env(leaf)]
#[serde_as(as = "DisplayFromStr")]
pub sysinfo_print_interval: IggyDuration,
}
-#[derive(Debug, Deserialize, Serialize)]
+#[derive(Debug, Deserialize, Serialize, ConfigEnv)]
pub struct EncryptionConfig {
pub enabled: bool,
+ #[config_env(secret)]
pub key: String,
}
-#[derive(Debug, Deserialize, Serialize)]
+#[derive(Debug, Deserialize, Serialize, ConfigEnv)]
pub struct StreamConfig {
pub path: String,
}
#[serde_as]
-#[derive(Debug, Deserialize, Serialize)]
+#[derive(Debug, Deserialize, Serialize, ConfigEnv)]
pub struct TopicConfig {
pub path: String,
+ #[config_env(leaf)]
#[serde_as(as = "DisplayFromStr")]
pub max_size: MaxTopicSize,
pub delete_oldest_segments: bool,
}
-#[derive(Debug, Deserialize, Serialize)]
+#[derive(Debug, Deserialize, Serialize, ConfigEnv)]
pub struct PartitionConfig {
pub path: String,
pub messages_required_to_save: u32,
+ #[config_env(leaf)]
pub size_of_messages_required_to_save: IggyByteSize,
pub enforce_fsync: bool,
pub validate_checksum: bool,
}
#[serde_as]
-#[derive(Debug, Deserialize, Serialize)]
+#[derive(Debug, Deserialize, Serialize, ConfigEnv)]
pub struct MessageDeduplicationConfig {
pub enabled: bool,
pub max_entries: u64,
+ #[config_env(leaf)]
#[serde_as(as = "DisplayFromStr")]
pub expiry: IggyDuration,
}
-#[derive(Debug, Deserialize, Serialize)]
+#[derive(Debug, Deserialize, Serialize, ConfigEnv)]
pub struct RecoveryConfig {
pub recreate_missing_state: bool,
}
#[serde_as]
-#[derive(Debug, Deserialize, Serialize)]
+#[derive(Debug, Deserialize, Serialize, ConfigEnv)]
pub struct SegmentConfig {
+ #[config_env(leaf)]
pub size: IggyByteSize,
+ #[config_env(leaf)]
pub cache_indexes: CacheIndexesConfig,
+ #[config_env(leaf)]
#[serde_as(as = "DisplayFromStr")]
pub message_expiry: IggyExpiry,
pub archive_expired: bool,
}
#[serde_as]
-#[derive(Debug, Deserialize, Serialize)]
+#[derive(Debug, Deserialize, Serialize, ConfigEnv)]
pub struct StateConfig {
pub enforce_fsync: bool,
pub max_file_operation_retries: u32,
+ #[config_env(leaf)]
#[serde_as(as = "DisplayFromStr")]
pub retry_delay: IggyDuration,
}
diff --git a/core/server/src/configs/tcp.rs b/core/server/src/configs/tcp.rs
index 02ec8bbd5..fcc7d2de6 100644
--- a/core/server/src/configs/tcp.rs
+++ b/core/server/src/configs/tcp.rs
@@ -16,12 +16,12 @@
* under the License.
*/
-use iggy_common::{IggyByteSize, IggyDuration};
+use iggy_common::{ConfigEnv, IggyByteSize, IggyDuration};
use serde::{Deserialize, Serialize};
use serde_with::DisplayFromStr;
use serde_with::serde_as;
-#[derive(Debug, Deserialize, Serialize, Clone)]
+#[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)]
pub struct TcpConfig {
pub enabled: bool,
pub address: String,
@@ -31,7 +31,7 @@ pub struct TcpConfig {
pub socket_migration: bool,
}
-#[derive(Debug, Deserialize, Serialize, Clone)]
+#[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)]
pub struct TcpTlsConfig {
pub enabled: bool,
pub self_signed: bool,
@@ -40,13 +40,16 @@ pub struct TcpTlsConfig {
}
#[serde_as]
-#[derive(Debug, Deserialize, Serialize, Clone)]
+#[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)]
pub struct TcpSocketConfig {
pub override_defaults: bool,
+ #[config_env(leaf)]
pub recv_buffer_size: IggyByteSize,
+ #[config_env(leaf)]
pub send_buffer_size: IggyByteSize,
pub keepalive: bool,
pub nodelay: bool,
+ #[config_env(leaf)]
#[serde_as(as = "DisplayFromStr")]
pub linger: IggyDuration,
}
diff --git a/core/server/src/configs/validators.rs b/core/server/src/configs/validators.rs
index af12a4102..987b0d86a 100644
--- a/core/server/src/configs/validators.rs
+++ b/core/server/src/configs/validators.rs
@@ -28,11 +28,12 @@ use crate::configs::server::{MemoryPoolConfig, PersonalAccessTokenConfig, Server
use crate::configs::sharding::NumaTopology;
use crate::configs::system::SegmentConfig;
use crate::streaming::segments::*;
+use configs::ConfigurationError;
use err_trail::ErrContext;
+use iggy_common::CompressionAlgorithm;
use iggy_common::IggyExpiry;
use iggy_common::MaxTopicSize;
use iggy_common::Validatable;
-use iggy_common::{CompressionAlgorithm, ConfigurationError};
use std::thread::available_parallelism;
use tracing::{error, warn};
@@ -41,17 +42,17 @@ impl Validatable<ConfigurationError> for ServerConfig {
self.system
.memory_pool
.validate()
- .error(|e: &iggy_common::ConfigurationError| {
+ .error(|e: &ConfigurationError| {
format!("{COMPONENT} (error: {e}) - failed to validate memory pool config")
})?;
self.data_maintenance
.validate()
- .error(|e: &iggy_common::ConfigurationError| {
+ .error(|e: &ConfigurationError| {
format!("{COMPONENT} (error: {e}) - failed to validate data maintenance config")
})?;
self.personal_access_token
.validate()
- .error(|e: &iggy_common::ConfigurationError| {
+ .error(|e: &ConfigurationError| {
format!(
"{COMPONENT} (error: {e}) - failed to validate personal access token config"
)
@@ -59,31 +60,27 @@ impl Validatable<ConfigurationError> for ServerConfig {
self.system
.segment
.validate()
- .error(|e: &iggy_common::ConfigurationError| {
+ .error(|e: &ConfigurationError| {
format!("{COMPONENT} (error: {e}) - failed to validate segment config")
})?;
self.system
.compression
.validate()
- .error(|e: &iggy_common::ConfigurationError| {
+ .error(|e: &ConfigurationError| {
format!("{COMPONENT} (error: {e}) - failed to validate compression config")
})?;
- self.telemetry
- .validate()
- .error(|e: &iggy_common::ConfigurationError| {
- format!("{COMPONENT} (error: {e}) - failed to validate telemetry config")
- })?;
+ self.telemetry.validate().error(|e: &ConfigurationError| {
+ format!("{COMPONENT} (error: {e}) - failed to validate telemetry config")
+ })?;
self.system
.sharding
.validate()
- .error(|e: &iggy_common::ConfigurationError| {
+ .error(|e: &ConfigurationError| {
format!("{COMPONENT} (error: {e}) - failed to validate sharding config")
})?;
- self.cluster
- .validate()
- .error(|e: &iggy_common::ConfigurationError| {
- format!("{COMPONENT} (error: {e}) - failed to validate cluster config")
- })?;
+ self.cluster.validate().error(|e: &ConfigurationError| {
+ format!("{COMPONENT} (error: {e}) - failed to validate cluster config")
+ })?;
let topic_size = match self.system.topic.max_size {
MaxTopicSize::Custom(size) => Ok(size.as_bytes_u64()),
@@ -222,11 +219,9 @@ impl Validatable<ConfigurationError> for MessageSaverConfig {
impl Validatable<ConfigurationError> for DataMaintenanceConfig {
fn validate(&self) -> Result<(), ConfigurationError> {
- self.messages
- .validate()
- .error(|e: &iggy_common::ConfigurationError| {
- format!("{COMPONENT} (error: {e}) - failed to validate messages maintenance config")
- })?;
+ self.messages.validate().error(|e: &ConfigurationError| {
+ format!("{COMPONENT} (error: {e}) - failed to validate messages maintenance config")
+ })?;
Ok(())
}
}
diff --git a/core/server/src/configs/websocket.rs b/core/server/src/configs/websocket.rs
index 2c9c996ce..b911f4c15 100644
--- a/core/server/src/configs/websocket.rs
+++ b/core/server/src/configs/websocket.rs
@@ -16,12 +16,13 @@
* under the License.
*/
+use iggy_common::ConfigEnv;
use iggy_common::IggyByteSize;
use serde::{Deserialize, Serialize};
use std::fmt::{Display, Formatter};
use tungstenite::protocol::WebSocketConfig as TungsteniteConfig;
-#[derive(Debug, Deserialize, Serialize, Clone)]
+#[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)]
pub struct WebSocketConfig {
pub enabled: bool,
pub address: String,
@@ -41,7 +42,7 @@ pub struct WebSocketConfig {
pub tls: WebSocketTlsConfig,
}
-#[derive(Debug, Deserialize, Serialize, Clone)]
+#[derive(Debug, Deserialize, Serialize, Clone, ConfigEnv)]
pub struct WebSocketTlsConfig {
pub enabled: bool,
pub self_signed: bool,
diff --git a/core/server/src/server_error.rs b/core/server/src/server_error.rs
index f49e10468..2be3135ab 100644
--- a/core/server/src/server_error.rs
+++ b/core/server/src/server_error.rs
@@ -68,7 +68,7 @@ error_set!(
}
ConfigurationError := {
- ConfigurationError(iggy_common::ConfigurationError),
+ ConfigurationError(configs::ConfigurationError),
}
ArchiverError := {
diff --git a/examples/java/README.md b/examples/java/README.md
index 1589d5c15..41d2644b4 100644
--- a/examples/java/README.md
+++ b/examples/java/README.md
@@ -33,7 +33,7 @@ Iggy requires valid credentials to authenticate client requests. The examples as
> **Note** <br>
> This setup is intended only for development and testing, not production use.
-By default, all server data is stored in the `local_data` directory (this can be changed via `system.path` in `server.toml`).
+By default, all server data is stored in the `local_data` directory (this can be changed via `system.path` in `config.toml`).
Root credentials are applied **only on the very first startup**, when no data directory exists yet.
Once the server has created and populated the data directory, the existing stored credentials will always be used, and supplying the `--with-default-root-credentials` flag or setting the environment variables will no longer override them.
diff --git a/examples/rust/README.md b/examples/rust/README.md
index c406c30a4..b1de83d06 100644
--- a/examples/rust/README.md
+++ b/examples/rust/README.md
@@ -31,7 +31,7 @@ Iggy requires valid credentials to authenticate client requests. The examples as
> **Note** <br>
> This setup is intended only for development and testing, not production use.
-By default, all server data is stored in the `local_data` directory (this can be changed via `system.path` in `server.toml`).
+By default, all server data is stored in the `local_data` directory (this can be changed via `system.path` in `config.toml`).
Root credentials are applied **only on the very first startup**, when no data directory exists yet.
Once the server has created and populated the data directory, the existing stored credentials will always be used, and supplying the `--with-default-root-credentials` flag or setting the environment variables will no longer override them.
diff --git a/foreign/node/src/e2e/tcp.cluster.e2e.ts b/foreign/node/src/e2e/tcp.cluster.e2e.ts
index 61f3d47e5..e3317f80d 100644
--- a/foreign/node/src/e2e/tcp.cluster.e2e.ts
+++ b/foreign/node/src/e2e/tcp.cluster.e2e.ts
@@ -22,7 +22,7 @@ import assert from "node:assert/strict";
import { getTestClient } from "./test-client.utils.js";
// cluster mode still in dev atm
-// response is mocked from /core/configs/server.toml
+// response is mocked from /core/server/config.toml
const expectedMeta = {
name: "iggy-cluster",
nodes: [