diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index b858874..192adab 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -8,6 +8,10 @@ on: - 'Cargo.toml' - 'Cargo.lock' - '.github/workflows/test.yml' + # config.rs asserts every K2I_* variable is documented in docs/kubernetes.md, + # so documentation changes can break the build and must be tested. + - 'docs/**' + - 'config/**' pull_request: branches: [main] paths: @@ -15,6 +19,10 @@ on: - 'Cargo.toml' - 'Cargo.lock' - '.github/workflows/test.yml' + # config.rs asserts every K2I_* variable is documented in docs/kubernetes.md, + # so documentation changes can break the build and must be tested. + - 'docs/**' + - 'config/**' permissions: contents: read @@ -133,13 +141,16 @@ jobs: ${{ runner.os }}-cargo-unit- ${{ runner.os }}-cargo-check- + # --include-ignored is required: every container-backed test is marked + # #[ignore = "requires Docker"] so it stays out of `cargo test`. Without + # this flag the job provisions Docker and then runs no Docker test at all. - name: Run integration tests - run: cargo test --test '*' --all-features -- --nocapture + run: cargo test --test '*' --all-features -- --nocapture --include-ignored env: DOCKER_HOST: unix:///var/run/docker.sock RUST_LOG: debug TESTCONTAINERS: "true" - timeout-minutes: 15 + timeout-minutes: 20 # Security audit security-audit: diff --git a/CHANGELOG.md b/CHANGELOG.md index 3f420a7..1f4d4d1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,16 +7,47 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- Implemented `create_gcs_store()` via `object_store::gcp::GoogleCloudStorageBuilder`, falling through to Application Default Credentials (Workload Identity on GKE) when `gcs_service_account_path` is unset. +- Implemented `create_azure_store()` via `object_store::azure::MicrosoftAzureBuilder`, falling through to `DefaultAzureCredential` (Managed Identity on AKS) when `azure_access_key` is unset. +- Documented that omitting `aws_access_key_id`/`aws_secret_access_key` activates the `AmazonS3Builder` default credential chain (env vars → IMDS → IRSA), unblocking EKS with IRSA and EC2 instance profiles without explicit config. +- Added `gcs_bucket_name`, `gcs_service_account_path`, `azure_container_name`, `azure_storage_account_name`, and `azure_access_key` fields to `IcebergConfig` for credential overrides and the Azure-required account name. + +### Security + +- `Secret` now redacts in `serde` serialization as well as `Debug`, emitting `REDACTED` in place of the value. `Config` derives `Serialize`, so previously any code that dumped or echoed the configuration would have emitted credentials in the clear. This follows the `secrecy` crate's convention of not implementing `Serialize` for secret-wrapped strings, so that emitting one must be a conscious act. The trade-off is deliberate and documented: a serialized `Config` no longer round-trips, since reading it back yields the literal `REDACTED` marker. + ### Changed +- `Config::validate` now rejects cloud warehouse settings that cannot be satisfied, rather than deferring the failure to the first flush: `azure_storage_account_name` is required for `az://` and `abfs://` paths, and `s3://`/`gs://` paths must name a bucket. A long-running ingest previously reported healthy and only failed minutes later, on its first write. +- Warehouse-path parsing (bucket, Azure container, in-bucket prefix) is now defined once in `config` and shared with the Iceberg writer, so what validation accepts is exactly what the writer can build a store from. +- Bumped the workspace version to 0.3.0 to absorb the semver-major addition of public fields on the externally-constructible `IcebergConfig` struct. The 0.x convention treats a minor bump (0.2 → 0.3) as the breaking-change boundary. - Upgraded the official Apache Iceberg Rust client from 0.7 to 0.10.0 and the Arrow/Parquet ecosystem from 54 to 58. - Removed the temporary standalone REST `update_schema` fallback now that `Transaction::update_schema()` is available in `iceberg-rust` 0.10.0. - Simplified `OfficialRestCommitter` by delegating all catalog operations to the official `RestCatalog` transaction APIs. ### Fixed -- Avoided manual OAuth2, route resolution, and multipart namespace encoding logic previously needed for the schema-update fallback. +- Aligned cloud object store uploads with the warehouse path recorded by the catalog/txlog. For cloud backends the store is rooted at the bucket, so `IcebergWriter` derives an in-bucket prefix from `warehouse_path` (e.g. `warehouse` for `s3://bucket/warehouse`) and applies it when addressing the store. Uploads previously landed at `s3://bucket/data/...` while the catalog recorded `s3://bucket/warehouse/data/...`, leaving every committed file unreadable. Preexisting on S3; the same handling now covers GCS and Azure. The prefix is applied **only** at upload time — paths handed to the catalog, transaction log, and read path stay warehouse-relative, since those consumers join them against `warehouse_path` themselves. +- Azure container parsing now handles the Hadoop ABFS form `abfs://container@account.dfs.core.windows.net/path` by extracting the container before the `@`, instead of treating the whole `container@account` segment as the container. +- `K2I_MONITORING_LOG_FORMAT` now takes effect. The tracing subscriber is configured before the full config is loaded and read the TOML value directly, so the environment override was silently ignored for all output. +- `K2I_RPC_ENABLED` no longer treats an unrecognized value as `false`. `K2I_RPC_ENABLED=yes` previously disabled the RPC server that the TOML had enabled; unparseable values now warn and preserve the configured value. +- Added `K2I_*` overrides for the remaining cloud object-store fields, including the Azure-required `azure_storage_account_name`, which could not previously be set by environment-only deployments. +- The unrecognized-variable warning no longer fires for `K2I_E2E_*` and the other harness variables that share the engine's environment during end-to-end runs. - Aligned Parquet writer properties with the parquet 58 API (`set_max_row_group_row_count`). +- Avoided manual OAuth2, route resolution, and multipart namespace encoding logic previously needed for the schema-update fallback. + +### Testing + +- Added container-backed S3 round-trip tests (MinIO) covering a prefixed warehouse (`s3://bucket/warehouse`), a multi-segment prefix, and a bucket-root warehouse. Each asserts that joining `warehouse_path` with the writer's reported path resolves to a real stored object, and that nothing was written to the doubled-prefix or bucket-root locations. These reproduce the warehouse-prefix defect above; they fail against the previous behaviour. +- CI's integration-tests job now passes `--include-ignored`. Every container-backed test is marked `#[ignore = "requires Docker"]`, so the job provisioned Docker and then ran no Docker test at all — including the pre-existing Kafka integration tests. +- The Tests workflow now also triggers on `docs/**` and `config/**`, since a test asserts that every `K2I_*` variable is documented in `docs/kubernetes.md`. + +### Documentation + +- Removed the `docs/configuration.md` claim that config values support `${VAR}` shell substitution. No such mechanism exists — following it would have authenticated with the literal string `${VAR}`. Replaced with the two real mechanisms: `{ file = "..." }` refs and `K2I_*` overrides. +- Updated `README.md`, `docs/architecture.md`, and `docs/configuration.md`, which still described GCS and Azure as declared-but-unwired. ### Requirements diff --git a/Cargo.lock b/Cargo.lock index ecaf850..81118eb 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3027,7 +3027,7 @@ dependencies = [ [[package]] name = "k2i-cli" -version = "0.2.2" +version = "0.3.0" dependencies = [ "anyhow", "axum", @@ -3052,7 +3052,7 @@ dependencies = [ [[package]] name = "k2i-core" -version = "0.2.2" +version = "0.3.0" dependencies = [ "ahash", "anyhow", @@ -3110,7 +3110,7 @@ dependencies = [ [[package]] name = "k2i-e2e-runner" -version = "0.2.2" +version = "0.3.0" dependencies = [ "anyhow", "arrow", @@ -3128,7 +3128,7 @@ dependencies = [ [[package]] name = "k2i-rpc" -version = "0.2.2" +version = "0.3.0" dependencies = [ "bincode", "chrono", @@ -3138,7 +3138,7 @@ dependencies = [ [[package]] name = "k2i-rpc-server" -version = "0.2.2" +version = "0.3.0" dependencies = [ "anyhow", "bytes", diff --git a/Cargo.toml b/Cargo.toml index b18d0a9..3fa4263 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -9,7 +9,7 @@ members = [ ] [workspace.package] -version = "0.2.2" +version = "0.3.0" edition = "2021" license = "Apache-2.0" authors = ["OSO DevOps"] @@ -119,6 +119,7 @@ testcontainers = "0.23" testcontainers-modules = { version = "0.11", features = [ "kafka", "localstack", + "minio", ] } tempfile = "3" tokio-test = "0.4" diff --git a/README.md b/README.md index 2f04410..b6688f4 100644 --- a/README.md +++ b/README.md @@ -257,7 +257,7 @@ K2I is ready for a first public release as a production-oriented Kafka-to-Iceber - Startup recovery computes state, but Kafka seeking/deduplication and startup orphan cleanup need further wiring. - Kafka offset commits are async; broker durability acknowledgement is not confirmed by the current helper. - Transaction-log entries are flushed, but not every entry is fsynced individually. -- GCS and Azure object-store configuration is declared, but writer creation is not complete for those backends. +- S3, GCS, and Azure object stores are wired end to end. S3 is covered by a container-backed round-trip test (MinIO); GCS and Azure are covered only at the configuration and store-construction level, so validate their credentials in your own environment before rollout. - Maintenance commands and task implementations exist; scheduler wiring should be reviewed for each deployment. See [Production Readiness](docs/production-readiness.md) for the detailed review checklist. diff --git a/config/example.toml b/config/example.toml index cb57234..9e44b3a 100644 --- a/config/example.toml +++ b/config/example.toml @@ -1,5 +1,31 @@ # K2I Example Configuration # Kafka to Iceberg streaming ingestion +# +# Configuration is loaded from this file and can be overridden by: +# - Secret file refs: `{ file = "path" }` inline tables (Kubernetes projected volumes) +# - `K2I_*` environment variables: override any field at runtime +# +# Override precedence (highest to lowest): +# 1. K2I_* environment variables +# 2. Inline TOML values (including `{ file = ... }` refs) +# +# Secret file refs — read secret contents from files at startup: +# [kafka.security] +# sasl_password = { file = "/etc/secrets/k2i/kafka-password" } +# +# [iceberg] +# aws_access_key_id = { file = "/etc/secrets/k2i/aws-key" } +# aws_secret_access_key = { file = "/etc/secrets/k2i/aws-secret" } +# +# Environment variable overrides (all K2I_* prefixed): +# export K2I_KAFKA_TOPIC=my-topic +# export K2I_KAFKA_SECURITY_SASL_PASSWORD=hunter2 +# export K2I_ICEBERG_WAREHOUSE_PATH=s3://prod-bucket/warehouse +# export K2I_ICEBERG_REST_CREDENTIAL=my-bearer-token +# +# Invalid numeric/enum env values are rejected with a warning; unrecognized +# K2I_* variables are also warned about. See docs/kubernetes.md for a full +# deployment guide. [kafka] bootstrap_servers = ["localhost:9092"] @@ -32,6 +58,14 @@ type = "raw" # sasl_mechanism = "SCRAM-SHA-256" # sasl_username = "user" # sasl_password = "password" +# +# Or load secrets from files (Kubernetes projected volumes): +# sasl_username = { file = "/etc/secrets/k2i/kafka-username" } +# sasl_password = { file = "/etc/secrets/k2i/kafka-password" } +# +# Or set via environment variables: +# K2I_KAFKA_SECURITY_SASL_USERNAME=user +# K2I_KAFKA_SECURITY_SASL_PASSWORD=hunter2 [schema_evolution] mode = "auto-additive" @@ -48,12 +82,67 @@ compression = "snappy" # REST catalog configuration rest_uri = "http://localhost:8181" +# +# REST catalog credential (bearer token or OAuth2), under [iceberg.rest]: +# [iceberg.rest] +# credential = "my-bearer-token" +# credential = { file = "/etc/secrets/k2i/rest-credential" } # K8s projected volume +# +# Or via environment variable: +# K2I_ICEBERG_REST_CREDENTIAL=my-bearer-token -# AWS configuration (for S3 storage) -# aws_region = "us-east-1" -# aws_access_key_id = "${AWS_ACCESS_KEY_ID}" -# aws_secret_access_key = "${AWS_SECRET_ACCESS_KEY}" -# s3_endpoint = "http://localhost:9000" # For MinIO +# AWS / S3 configuration +# Set aws_region for the S3 bucket region. +# For explicit key-based auth, set aws_access_key_id and aws_secret_access_key: +# aws_region = "us-east-1" +# aws_access_key_id = "AKIA..." +# aws_secret_access_key = "..." +# s3_endpoint = "http://localhost:9000" # For MinIO +# +# Or load from files (Kubernetes projected volumes): +# aws_access_key_id = { file = "/etc/secrets/k2i/aws-key" } +# aws_secret_access_key = { file = "/etc/secrets/k2i/aws-secret" } +# +# Or set via environment variables: +# K2I_ICEBERG_AWS_ACCESS_KEY_ID=AKIA... +# K2I_ICEBERG_AWS_SECRET_ACCESS_KEY=... +# +# When both access key fields are omitted, the SDK uses the default credential +# chain: environment variables (AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, etc.) +# -> IMDS (EC2 instance profiles) -> IRSA (EKS IAM Roles for Service Accounts). +# This allows EKS with IRSA, EC2 instance profiles, and local dev with env +# vars — no explicit config needed. + +# GCS configuration — set warehouse_path to gs://bucket/path +# warehouse_path = "gs://my-gcs-bucket/warehouse" +# gcs_bucket_name = "my-gcs-bucket" # optional: override bucket from path +# gcs_service_account_path = "/path/to/key.json" # optional: explicit SA key +# +# When gcs_service_account_path is omitted, the SDK uses Application Default +# Credentials (ADC). On GKE this means Workload Identity Federation; locally +# it picks up GOOGLE_APPLICATION_CREDENTIALS or gcloud auth. + +# Azure configuration — set warehouse_path to either: +# az://container/path (simple form) +# abfs://container@account.dfs.core.windows.net/path (Hadoop ABFS form) +# Both forms parse the container correctly; the account is taken from +# azure_storage_account_name (REQUIRED) since it cannot derive the endpoint host. +# +# Examples: +# warehouse_path = "az://my-container/warehouse" +# warehouse_path = "abfs://my-container@mystorageaccount.dfs.core.windows.net/warehouse" +# azure_storage_account_name = "mystorageaccount" # REQUIRED +# azure_container_name = "my-container" # optional: override container from path +# azure_access_key = "..." # optional; or { file = "/etc/secrets/k2i/azure-key" } +# +# Note: values are not shell-interpolated. To source a credential from the +# environment use K2I_ICEBERG_AZURE_ACCESS_KEY, and to source it from a file use +# the `{ file = "..." }` form — not "${VAR}". +# +# When azure_access_key is omitted, the SDK uses the DefaultAzureCredential +# chain: environment variables -> Managed Identity (on AKS) -> Azure CLI. +# This enables AKS with Workload Identity or system-assigned Managed Identity +# without explicit keys. # Partition specification # [[iceberg.partition_spec]] diff --git a/crates/k2i-cli/src/commands/dev.rs b/crates/k2i-cli/src/commands/dev.rs index 70037cc..5f11bf4 100644 --- a/crates/k2i-cli/src/commands/dev.rs +++ b/crates/k2i-cli/src/commands/dev.rs @@ -81,6 +81,11 @@ pub async fn run(options: DevOptions) -> Result<()> { aws_access_key_id: None, aws_secret_access_key: None, s3_endpoint: None, + gcs_bucket_name: None, + gcs_service_account_path: None, + azure_container_name: None, + azure_storage_account_name: None, + azure_access_key: None, catalog_manager: CatalogManagerConfig::default(), table_management: TableManagementConfig::default(), rest: Default::default(), diff --git a/crates/k2i-cli/src/main.rs b/crates/k2i-cli/src/main.rs index 423ec1b..22dbf3a 100644 --- a/crates/k2i-cli/src/main.rs +++ b/crates/k2i-cli/src/main.rs @@ -361,13 +361,19 @@ async fn main() { async fn run_cli() -> ExitCode { let cli = Cli::parse(); - // Try to load config for log format settings (optional - falls back to JSON) - let log_format = cli - .config - .as_ref() - .and_then(|path| std::fs::read_to_string(path).ok()) - .and_then(|content| toml::from_str::(&content).ok()) - .map(|config| config.monitoring.log_format) + // Try to load config for log format settings (optional - falls back to JSON). + // `K2I_MONITORING_LOG_FORMAT` wins over the TOML value here for the same + // reason it does in `Config::apply_env_overrides`; checking it directly means + // the override also applies when no config file is given, and before the + // subscriber exists to report a bad value. + let log_format = LogFormat::from_env() + .or_else(|| { + cli.config + .as_ref() + .and_then(|path| std::fs::read_to_string(path).ok()) + .and_then(|content| toml::from_str::(&content).ok()) + .map(|config| config.monitoring.log_format) + }) .unwrap_or(LogFormat::Json); // Initialize logging @@ -513,8 +519,5 @@ async fn execute_command(cli: Cli) -> Result<()> { fn load_config(path: &Option) -> Result { let path = path.clone().unwrap_or_else(|| PathBuf::from("config.toml")); - - let content = std::fs::read_to_string(&path)?; - let config: Config = toml::from_str(&content)?; - Ok(config) + Ok(Config::from_file(&path)?) } diff --git a/crates/k2i-core/src/backfill.rs b/crates/k2i-core/src/backfill.rs index 4d11692..90699a9 100644 --- a/crates/k2i-core/src/backfill.rs +++ b/crates/k2i-core/src/backfill.rs @@ -357,6 +357,11 @@ mod tests { aws_access_key_id: None, aws_secret_access_key: None, s3_endpoint: None, + gcs_bucket_name: None, + gcs_service_account_path: None, + azure_container_name: None, + azure_storage_account_name: None, + azure_access_key: None, catalog_manager: CatalogManagerConfig::default(), table_management: TableManagementConfig::default(), rest: Default::default(), diff --git a/crates/k2i-core/src/config.rs b/crates/k2i-core/src/config.rs index 6de0952..93bdc2e 100644 --- a/crates/k2i-core/src/config.rs +++ b/crates/k2i-core/src/config.rs @@ -5,6 +5,117 @@ use serde::{Deserialize, Serialize}; use std::path::PathBuf; +/// A configuration secret that redacts its value in `Debug` output. +/// +/// In TOML it accepts either a plain string or a `{ file = "path" }` table, +/// which supports Kubernetes projected-volume secrets: +/// +/// ```toml +/// sasl_password = "hunter2" +/// # or +/// sasl_password = { file = "/etc/secrets/k2i/kafka-password" } +/// ``` +/// +/// File contents are trimmed. Read the value explicitly with +/// [`Secret::expose`] or via `Deref` (`&*secret`). +/// +/// # Output is redacted +/// +/// Both `Debug` and [`Serialize`] redact: they emit [`Secret::REDACTED`], never +/// the value. `Config` derives `Serialize`, so any future code that dumps or +/// echoes the configuration — a `config dump` subcommand, an RPC response, a +/// diagnostic bundle — cannot leak a credential by accident. This mirrors the +/// `secrecy` crate, which deliberately does not implement `Serialize` for +/// secret-wrapped strings so that emitting one has to be a conscious act. +/// +/// The trade-off is that a serialized `Config` does not round-trip: reading one +/// back yields the literal `REDACTED` marker rather than the original +/// credential. That is intentional — a visibly broken credential is a far better +/// failure than a silently leaked one. Code that genuinely needs the plaintext +/// must call [`Secret::expose`] at the point of use. +#[derive(Clone, PartialEq, Eq)] +pub struct Secret(String); + +impl Secret { + /// The placeholder emitted by `Debug` and `Serialize` in place of the value. + pub const REDACTED: &'static str = "REDACTED"; + + /// Wrap a plaintext value as a secret. + pub fn new(value: impl Into) -> Self { + Self(value.into()) + } + + /// Expose the secret value. + pub fn expose(&self) -> &str { + &self.0 + } +} + +impl std::fmt::Debug for Secret { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "Secret({})", Self::REDACTED) + } +} + +impl Serialize for Secret { + /// Emits [`Secret::REDACTED`] rather than the value. See the type docs for + /// why this deliberately breaks round-tripping. + fn serialize(&self, serializer: S) -> Result + where + S: serde::Serializer, + { + serializer.serialize_str(Self::REDACTED) + } +} + +impl std::ops::Deref for Secret { + type Target = str; + + fn deref(&self) -> &str { + &self.0 + } +} + +impl From for Secret { + fn from(value: String) -> Self { + Self(value) + } +} + +impl From<&str> for Secret { + fn from(value: &str) -> Self { + Self(value.to_string()) + } +} + +impl<'de> Deserialize<'de> for Secret { + fn deserialize(deserializer: D) -> Result + where + D: serde::Deserializer<'de>, + { + #[derive(Deserialize)] + #[serde(untagged)] + enum Repr { + Plain(String), + FileRef { file: PathBuf }, + } + + match Repr::deserialize(deserializer)? { + Repr::Plain(value) => Ok(Secret(value)), + Repr::FileRef { file } => { + let contents = std::fs::read_to_string(&file).map_err(|e| { + serde::de::Error::custom(format!( + "failed to read secret file '{}': {}", + file.display(), + e + )) + })?; + Ok(Secret(contents.trim().to_string())) + } + } + } +} + /// Main configuration structure. #[derive(Debug, Clone, Deserialize, Serialize)] pub struct Config { @@ -205,11 +316,11 @@ pub struct KafkaSecurityConfig { /// SASL mechanism (PLAIN, SCRAM-SHA-256, SCRAM-SHA-512) pub sasl_mechanism: Option, - /// SASL username - pub sasl_username: Option, + /// SASL username (plain string or `{ file = "path" }`) + pub sasl_username: Option, - /// SASL password - pub sasl_password: Option, + /// SASL password (plain string or `{ file = "path" }`) + pub sasl_password: Option, /// SSL CA certificate location pub ssl_ca_location: Option, @@ -257,11 +368,11 @@ pub struct IcebergConfig { /// AWS region (for Glue catalog and S3) pub aws_region: Option, - /// AWS access key ID - pub aws_access_key_id: Option, + /// AWS access key ID (plain string or `{ file = "path" }`) + pub aws_access_key_id: Option, - /// AWS secret access key - pub aws_secret_access_key: Option, + /// AWS secret access key (plain string or `{ file = "path" }`) + pub aws_secret_access_key: Option, /// S3 endpoint (for MinIO or other S3-compatible storage) pub s3_endpoint: Option, @@ -290,6 +401,26 @@ pub struct IcebergConfig { #[serde(default)] pub sql_catalog: Option, + /// GCS bucket name override (default: parsed from warehouse_path) + #[serde(default)] + pub gcs_bucket_name: Option, + + /// GCS service account key path (optional; falls through to ADC when unset) + #[serde(default)] + pub gcs_service_account_path: Option, + + /// Azure container name override (default: parsed from warehouse_path) + #[serde(default)] + pub azure_container_name: Option, + + /// Azure storage account name (required for Azure backends) + #[serde(default)] + pub azure_storage_account_name: Option, + + /// Azure storage access key (optional; falls through to Managed Identity / env vars) + #[serde(default)] + pub azure_access_key: Option, + /// Object store configuration used by local-first deployments. #[serde(default)] pub object_store: ObjectStoreConfig, @@ -409,21 +540,21 @@ pub struct RestCatalogConfig { #[serde(default)] pub credential_type: CredentialType, - /// Credential value (token for bearer auth) + /// Credential value (token for bearer auth; plain string or `{ file = "path" }`) #[serde(default)] - pub credential: Option, + pub credential: Option, /// OAuth2 token endpoint (for oauth2 credential type) #[serde(default)] pub oauth2_token_endpoint: Option, - /// OAuth2 client ID + /// OAuth2 client ID (plain string or `{ file = "path" }`) #[serde(default)] - pub oauth2_client_id: Option, + pub oauth2_client_id: Option, - /// OAuth2 client secret + /// OAuth2 client secret (plain string or `{ file = "path" }`) #[serde(default)] - pub oauth2_client_secret: Option, + pub oauth2_client_secret: Option, /// OAuth2 scope (optional) #[serde(default)] @@ -854,6 +985,27 @@ pub enum LogFormat { Text, } +impl LogFormat { + /// Parse a `K2I_MONITORING_LOG_FORMAT` value, case-insensitively. + /// + /// Returns `None` for unrecognized values so callers can warn and fall back. + pub fn parse_env_value(value: &str) -> Option { + match value.to_lowercase().as_str() { + "json" => Some(Self::Json), + "text" => Some(Self::Text), + _ => None, + } + } + + /// Read the log format from `K2I_MONITORING_LOG_FORMAT`, if set and valid. + /// + /// Used to configure the tracing subscriber before the full config is + /// loaded, so the env override applies to startup logging too. + pub fn from_env() -> Option { + Self::parse_env_value(&std::env::var("K2I_MONITORING_LOG_FORMAT").ok()?) + } +} + // Default value functions fn default_batch_size() -> usize { 1000 @@ -974,11 +1126,135 @@ fn default_auto_create() -> bool { true } +/// Extract the bucket from a cloud warehouse path, e.g. `s3://bucket/prefix` +/// with `scheme = "s3://"` yields `Some("bucket")`. +/// +/// Returns `None` when the path does not use `scheme` or names no bucket. +/// Shared by [`Config::validate`] and the Iceberg writer so that what +/// validation accepts is exactly what the writer can build a store from. +pub(crate) fn bucket_from_warehouse(scheme: &str, warehouse_path: &str) -> Option { + let bucket = warehouse_path.strip_prefix(scheme)?.split('/').next()?; + if bucket.is_empty() { + None + } else { + Some(bucket.to_string()) + } +} + +/// Extract the in-bucket prefix that follows the bucket in a cloud warehouse +/// path: `gs://bucket/some/prefix` yields `Some("some/prefix")`. +/// +/// Returns `None` when the path has no subpath past the bucket, e.g. `s3://bucket`. +pub(crate) fn warehouse_prefix_after_bucket(scheme: &str, warehouse_path: &str) -> Option { + let after_bucket = warehouse_path.strip_prefix(scheme)?.split_once('/')?.1; + let trimmed = after_bucket.trim_matches('/'); + if trimmed.is_empty() { + None + } else { + Some(trimmed.to_string()) + } +} + +/// Strip either Azure warehouse scheme, returning the remainder. +fn strip_azure_scheme(warehouse_path: &str) -> Option<&str> { + warehouse_path + .strip_prefix("az://") + .or_else(|| warehouse_path.strip_prefix("abfs://")) +} + +/// Extract the Azure container from a warehouse path, supporting both the +/// simple form `az://container/path` and the Hadoop ABFS form +/// `abfs://container@account.dfs.core.windows.net/path`. +pub(crate) fn azure_container_from_warehouse(warehouse_path: &str) -> Option { + let first_segment = strip_azure_scheme(warehouse_path)?.split('/').next()?; + // ABFS form: take the segment before `@`. The simple form has no `@`. + let container = first_segment.split('@').next()?; + if container.is_empty() { + None + } else { + Some(container.to_string()) + } +} + +/// Extract the in-bucket prefix following the container in an Azure warehouse +/// path, for both the `az://` and `abfs://` forms. +pub(crate) fn azure_prefix_after_container(warehouse_path: &str) -> Option { + let after_container = strip_azure_scheme(warehouse_path)?.split_once('/')?.1; + let trimmed = after_container.trim_matches('/'); + if trimmed.is_empty() { + None + } else { + Some(trimmed.to_string()) + } +} + +/// `K2I_*` prefixes owned by tooling rather than by [`Config`]. +/// +/// These are consumed by the end-to-end test harness (`k2i-e2e-runner`) and its +/// helper subprocesses, which run with the same environment as the engine. They +/// are not config fields, so the typo warning below must not flag them. +const RESERVED_ENV_PREFIXES: &[&str] = &["K2I_E2E_", "K2I_PARQUET_", "K2I_ICEBERG_METADATA_PATH"]; + +/// Environment variables recognized by [`Config::apply_env_overrides`]. +/// +/// Keep in sync with the `env_val(...)` calls in that method. +const KNOWN_ENV_VARS: &[&str] = &[ + "K2I_KAFKA_BOOTSTRAP_SERVERS", + "K2I_KAFKA_TOPIC", + "K2I_KAFKA_CONSUMER_GROUP", + "K2I_KAFKA_BATCH_SIZE", + "K2I_KAFKA_BATCH_TIMEOUT_MS", + "K2I_KAFKA_SESSION_TIMEOUT_MS", + "K2I_KAFKA_HEARTBEAT_INTERVAL_MS", + "K2I_KAFKA_MAX_POLL_INTERVAL_MS", + "K2I_KAFKA_AUTO_OFFSET_RESET", + "K2I_KAFKA_SECURITY_PROTOCOL", + "K2I_KAFKA_SECURITY_SASL_MECHANISM", + "K2I_KAFKA_SECURITY_SASL_USERNAME", + "K2I_KAFKA_SECURITY_SASL_PASSWORD", + "K2I_ICEBERG_CATALOG_TYPE", + "K2I_ICEBERG_WAREHOUSE_PATH", + "K2I_ICEBERG_DATABASE_NAME", + "K2I_ICEBERG_TABLE_NAME", + "K2I_ICEBERG_AWS_REGION", + "K2I_ICEBERG_AWS_ACCESS_KEY_ID", + "K2I_ICEBERG_AWS_SECRET_ACCESS_KEY", + "K2I_ICEBERG_GCS_BUCKET_NAME", + "K2I_ICEBERG_GCS_SERVICE_ACCOUNT_PATH", + "K2I_ICEBERG_AZURE_STORAGE_ACCOUNT_NAME", + "K2I_ICEBERG_AZURE_CONTAINER_NAME", + "K2I_ICEBERG_AZURE_ACCESS_KEY", + "K2I_ICEBERG_S3_ENDPOINT", + "K2I_ICEBERG_REST_URI", + "K2I_ICEBERG_HIVE_METASTORE_URI", + "K2I_ICEBERG_REST_CREDENTIAL", + "K2I_ICEBERG_REST_OAUTH2_CLIENT_ID", + "K2I_ICEBERG_REST_OAUTH2_CLIENT_SECRET", + "K2I_SCHEMA_EVOLUTION_MODE", + "K2I_SCHEMA_EVOLUTION_ON_BREAKING_CHANGE", + "K2I_BUFFER_TTL_SECONDS", + "K2I_BUFFER_MAX_SIZE_MB", + "K2I_BUFFER_FLUSH_INTERVAL_SECONDS", + "K2I_TRANSACTION_LOG_LOG_DIR", + "K2I_MONITORING_HEALTH_PORT", + "K2I_MONITORING_METRICS_PORT", + "K2I_MONITORING_LOG_LEVEL", + "K2I_MONITORING_LOG_FORMAT", + "K2I_RPC_ENABLED", + "K2I_RPC_SOCKET_PATH", +]; + impl Config { /// Load configuration from a TOML file. + /// + /// Applies the following in order: + /// 1. Parse TOML values (secret `{ file = "..." }` refs resolve here) + /// 2. Apply `K2I_*` environment variable overrides + /// 3. Validate the merged configuration pub fn from_file(path: &std::path::Path) -> crate::Result { let content = std::fs::read_to_string(path)?; - let config: Config = toml::from_str(&content)?; + let mut config: Config = toml::from_str(&content)?; + config.apply_env_overrides(); config.validate()?; Ok(config) } @@ -1035,6 +1311,40 @@ impl Config { return Err(crate::Error::Config("Warehouse path is required".into())); } + // Fail at startup rather than on the first flush: the object store is + // not constructed until a writer is built, which for a long-running + // ingest means a misconfigured warehouse surfaces minutes in, after the + // process has already reported healthy. + let warehouse = &self.iceberg.warehouse_path; + if warehouse.starts_with("az://") || warehouse.starts_with("abfs://") { + if self.iceberg.azure_storage_account_name.is_none() { + return Err(crate::Error::Config( + "iceberg.azure_storage_account_name is required for az:// and abfs:// warehouse paths (it cannot be derived from the path)" + .into(), + )); + } + if self.iceberg.azure_container_name.is_none() + && azure_container_from_warehouse(warehouse).is_none() + { + return Err(crate::Error::Config(format!( + "could not determine the Azure container from warehouse path '{warehouse}'; set iceberg.azure_container_name" + ))); + } + } else if warehouse.starts_with("gs://") + && self.iceberg.gcs_bucket_name.is_none() + && bucket_from_warehouse("gs://", warehouse).is_none() + { + return Err(crate::Error::Config(format!( + "could not determine the GCS bucket from warehouse path '{warehouse}'; set iceberg.gcs_bucket_name" + ))); + } else if warehouse.starts_with("s3://") + && bucket_from_warehouse("s3://", warehouse).is_none() + { + return Err(crate::Error::Config(format!( + "could not determine the S3 bucket from warehouse path '{warehouse}'" + ))); + } + if self.iceberg.catalog_type == CatalogType::Sql { let sql_catalog = self.iceberg.sql_catalog.as_ref().ok_or_else(|| { crate::Error::Config("iceberg.sql_catalog is required when catalog_type=sql".into()) @@ -1079,6 +1389,273 @@ impl Config { Ok(()) } + + /// Apply `K2I_*` environment variable overrides on top of TOML values. + /// + /// Env vars use the convention `K2I_` + uppercase field path with `_` separators. + /// For example: `K2I_KAFKA_TOPIC`, `K2I_ICEBERG_WAREHOUSE_PATH`, `K2I_KAFKA_SECURITY_SASL_PASSWORD`. + /// + /// Invalid numeric/enum values are rejected with a warning (the TOML or + /// default value is preserved). Unrecognized `K2I_*` variables are also + /// logged so typos do not fail silently. + fn apply_env_overrides(&mut self) { + fn env_val(key: &str) -> Option { + std::env::var(key).ok() + } + + fn parse_num(key: &str, v: &str) -> Option { + match v.parse::() { + Ok(n) => Some(n), + Err(_) => { + tracing::warn!(var = key, value = %v, "Ignoring invalid numeric value"); + None + } + } + } + + fn warn_bad_enum(key: &str, v: &str, valid: &[&str]) { + tracing::warn!(var = key, value = %v, valid = ?valid, "Ignoring invalid enum value"); + } + + // --- Kafka --- + if let Some(v) = env_val("K2I_KAFKA_BOOTSTRAP_SERVERS") { + self.kafka.bootstrap_servers = v.split(',').map(String::from).collect(); + } + if let Some(v) = env_val("K2I_KAFKA_TOPIC") { + self.kafka.topic = v; + } + if let Some(v) = env_val("K2I_KAFKA_CONSUMER_GROUP") { + self.kafka.consumer_group = v; + } + if let Some(v) = env_val("K2I_KAFKA_BATCH_SIZE") { + if let Some(n) = parse_num("K2I_KAFKA_BATCH_SIZE", &v) { + self.kafka.batch_size = n; + } + } + if let Some(v) = env_val("K2I_KAFKA_BATCH_TIMEOUT_MS") { + if let Some(n) = parse_num("K2I_KAFKA_BATCH_TIMEOUT_MS", &v) { + self.kafka.batch_timeout_ms = n; + } + } + if let Some(v) = env_val("K2I_KAFKA_SESSION_TIMEOUT_MS") { + if let Some(n) = parse_num("K2I_KAFKA_SESSION_TIMEOUT_MS", &v) { + self.kafka.session_timeout_ms = n; + } + } + if let Some(v) = env_val("K2I_KAFKA_HEARTBEAT_INTERVAL_MS") { + if let Some(n) = parse_num("K2I_KAFKA_HEARTBEAT_INTERVAL_MS", &v) { + self.kafka.heartbeat_interval_ms = n; + } + } + if let Some(v) = env_val("K2I_KAFKA_MAX_POLL_INTERVAL_MS") { + if let Some(n) = parse_num("K2I_KAFKA_MAX_POLL_INTERVAL_MS", &v) { + self.kafka.max_poll_interval_ms = n; + } + } + if let Some(v) = env_val("K2I_KAFKA_AUTO_OFFSET_RESET") { + match v.to_lowercase().as_str() { + "earliest" => self.kafka.auto_offset_reset = OffsetReset::Earliest, + "latest" => self.kafka.auto_offset_reset = OffsetReset::Latest, + _ => warn_bad_enum("K2I_KAFKA_AUTO_OFFSET_RESET", &v, &["earliest", "latest"]), + } + } + + // Kafka security + if let Some(v) = env_val("K2I_KAFKA_SECURITY_PROTOCOL") { + self.kafka.security.protocol = Some(v); + } + if let Some(v) = env_val("K2I_KAFKA_SECURITY_SASL_MECHANISM") { + self.kafka.security.sasl_mechanism = Some(v); + } + if let Some(v) = env_val("K2I_KAFKA_SECURITY_SASL_USERNAME") { + self.kafka.security.sasl_username = Some(Secret::new(v)); + } + if let Some(v) = env_val("K2I_KAFKA_SECURITY_SASL_PASSWORD") { + self.kafka.security.sasl_password = Some(Secret::new(v)); + } + + // --- Iceberg --- + if let Some(v) = env_val("K2I_ICEBERG_CATALOG_TYPE") { + match v.to_lowercase().as_str() { + "rest" => self.iceberg.catalog_type = CatalogType::Rest, + "glue" => self.iceberg.catalog_type = CatalogType::Glue, + "hive" => self.iceberg.catalog_type = CatalogType::Hive, + "nessie" => self.iceberg.catalog_type = CatalogType::Nessie, + "sql" => self.iceberg.catalog_type = CatalogType::Sql, + _ => warn_bad_enum( + "K2I_ICEBERG_CATALOG_TYPE", + &v, + &["rest", "glue", "hive", "nessie", "sql"], + ), + } + } + if let Some(v) = env_val("K2I_ICEBERG_WAREHOUSE_PATH") { + self.iceberg.warehouse_path = v; + } + if let Some(v) = env_val("K2I_ICEBERG_DATABASE_NAME") { + self.iceberg.database_name = v; + } + if let Some(v) = env_val("K2I_ICEBERG_TABLE_NAME") { + self.iceberg.table_name = v; + } + if let Some(v) = env_val("K2I_ICEBERG_AWS_REGION") { + self.iceberg.aws_region = Some(v); + } + if let Some(v) = env_val("K2I_ICEBERG_AWS_ACCESS_KEY_ID") { + self.iceberg.aws_access_key_id = Some(Secret::new(v)); + } + if let Some(v) = env_val("K2I_ICEBERG_AWS_SECRET_ACCESS_KEY") { + self.iceberg.aws_secret_access_key = Some(Secret::new(v)); + } + if let Some(v) = env_val("K2I_ICEBERG_GCS_BUCKET_NAME") { + self.iceberg.gcs_bucket_name = Some(v); + } + if let Some(v) = env_val("K2I_ICEBERG_GCS_SERVICE_ACCOUNT_PATH") { + self.iceberg.gcs_service_account_path = Some(v); + } + if let Some(v) = env_val("K2I_ICEBERG_AZURE_STORAGE_ACCOUNT_NAME") { + self.iceberg.azure_storage_account_name = Some(v); + } + if let Some(v) = env_val("K2I_ICEBERG_AZURE_CONTAINER_NAME") { + self.iceberg.azure_container_name = Some(v); + } + if let Some(v) = env_val("K2I_ICEBERG_AZURE_ACCESS_KEY") { + self.iceberg.azure_access_key = Some(Secret::new(v)); + } + if let Some(v) = env_val("K2I_ICEBERG_S3_ENDPOINT") { + self.iceberg.s3_endpoint = Some(v); + } + if let Some(v) = env_val("K2I_ICEBERG_REST_URI") { + self.iceberg.rest_uri = Some(v); + } + if let Some(v) = env_val("K2I_ICEBERG_HIVE_METASTORE_URI") { + self.iceberg.hive_metastore_uri = Some(v); + } + + // REST catalog advanced + if let Some(v) = env_val("K2I_ICEBERG_REST_CREDENTIAL") { + self.iceberg.rest.credential = Some(Secret::new(v)); + } + if let Some(v) = env_val("K2I_ICEBERG_REST_OAUTH2_CLIENT_ID") { + self.iceberg.rest.oauth2_client_id = Some(Secret::new(v)); + } + if let Some(v) = env_val("K2I_ICEBERG_REST_OAUTH2_CLIENT_SECRET") { + self.iceberg.rest.oauth2_client_secret = Some(Secret::new(v)); + } + + // --- Schema evolution --- + if let Some(v) = env_val("K2I_SCHEMA_EVOLUTION_MODE") { + match v.to_lowercase().as_str() { + "manual" => self.schema_evolution.mode = SchemaEvolutionMode::Manual, + "auto-additive" => self.schema_evolution.mode = SchemaEvolutionMode::AutoAdditive, + "permissive" => self.schema_evolution.mode = SchemaEvolutionMode::Permissive, + _ => warn_bad_enum( + "K2I_SCHEMA_EVOLUTION_MODE", + &v, + &["manual", "auto-additive", "permissive"], + ), + } + } + if let Some(v) = env_val("K2I_SCHEMA_EVOLUTION_ON_BREAKING_CHANGE") { + match v.to_lowercase().as_str() { + "pause" => self.schema_evolution.on_breaking_change = OnBreakingChange::Pause, + "fail" => self.schema_evolution.on_breaking_change = OnBreakingChange::Fail, + "skip-message" => { + self.schema_evolution.on_breaking_change = OnBreakingChange::SkipMessage + } + _ => warn_bad_enum( + "K2I_SCHEMA_EVOLUTION_ON_BREAKING_CHANGE", + &v, + &["pause", "fail", "skip-message"], + ), + } + } + + // --- Buffer --- + if let Some(v) = env_val("K2I_BUFFER_TTL_SECONDS") { + if let Some(n) = parse_num("K2I_BUFFER_TTL_SECONDS", &v) { + self.buffer.ttl_seconds = n; + } + } + if let Some(v) = env_val("K2I_BUFFER_MAX_SIZE_MB") { + if let Some(n) = parse_num("K2I_BUFFER_MAX_SIZE_MB", &v) { + self.buffer.max_size_mb = n; + } + } + if let Some(v) = env_val("K2I_BUFFER_FLUSH_INTERVAL_SECONDS") { + if let Some(n) = parse_num("K2I_BUFFER_FLUSH_INTERVAL_SECONDS", &v) { + self.buffer.flush_interval_seconds = n; + } + } + + // --- Transaction log --- + if let Some(v) = env_val("K2I_TRANSACTION_LOG_LOG_DIR") { + self.transaction_log.log_dir = std::path::PathBuf::from(v); + } + + // --- Monitoring --- + if let Some(v) = env_val("K2I_MONITORING_HEALTH_PORT") { + if let Some(n) = parse_num("K2I_MONITORING_HEALTH_PORT", &v) { + self.monitoring.health_port = n; + } + } + if let Some(v) = env_val("K2I_MONITORING_METRICS_PORT") { + if let Some(n) = parse_num("K2I_MONITORING_METRICS_PORT", &v) { + self.monitoring.metrics_port = n; + } + } + if let Some(v) = env_val("K2I_MONITORING_LOG_LEVEL") { + match v.to_lowercase().as_str() { + "trace" => self.monitoring.log_level = LogLevel::Trace, + "debug" => self.monitoring.log_level = LogLevel::Debug, + "info" => self.monitoring.log_level = LogLevel::Info, + "warn" => self.monitoring.log_level = LogLevel::Warn, + "error" => self.monitoring.log_level = LogLevel::Error, + _ => warn_bad_enum( + "K2I_MONITORING_LOG_LEVEL", + &v, + &["trace", "debug", "info", "warn", "error"], + ), + } + } + if let Some(v) = env_val("K2I_MONITORING_LOG_FORMAT") { + match LogFormat::parse_env_value(&v) { + Some(format) => self.monitoring.log_format = format, + None => warn_bad_enum("K2I_MONITORING_LOG_FORMAT", &v, &["json", "text"]), + } + } + + // --- RPC --- + if let Some(v) = env_val("K2I_RPC_ENABLED") { + // Anything unrecognized keeps the TOML value rather than silently + // reading as `false` — `K2I_RPC_ENABLED=yes` must not disable RPC. + match v.to_lowercase().as_str() { + "true" | "1" | "yes" | "on" => self.rpc.enabled = true, + "false" | "0" | "no" | "off" => self.rpc.enabled = false, + _ => warn_bad_enum( + "K2I_RPC_ENABLED", + &v, + &["true", "false", "1", "0", "yes", "no", "on", "off"], + ), + } + } + if let Some(v) = env_val("K2I_RPC_SOCKET_PATH") { + self.rpc.socket_path = std::path::PathBuf::from(v); + } + + // Warn on unrecognized K2I_* variables (typo detection). + for (key, _) in std::env::vars_os() { + let key = key.to_string_lossy(); + if key.starts_with("K2I_") + && !KNOWN_ENV_VARS.contains(&key.as_ref()) + && !RESERVED_ENV_PREFIXES + .iter() + .any(|prefix| key.starts_with(prefix)) + { + tracing::warn!(var = %key, "Unrecognized K2I_* environment variable ignored"); + } + } + } } #[cfg(test)] @@ -1124,6 +1701,11 @@ mod tests { aws_access_key_id: None, aws_secret_access_key: None, s3_endpoint: None, + gcs_bucket_name: None, + gcs_service_account_path: None, + azure_container_name: None, + azure_storage_account_name: None, + azure_access_key: None, catalog_manager: CatalogManagerConfig::default(), table_management: TableManagementConfig::default(), rest: RestCatalogConfig::default(), @@ -1173,6 +1755,11 @@ mod tests { aws_access_key_id: None, aws_secret_access_key: None, s3_endpoint: None, + gcs_bucket_name: None, + gcs_service_account_path: None, + azure_container_name: None, + azure_storage_account_name: None, + azure_access_key: None, catalog_manager: CatalogManagerConfig::default(), table_management: TableManagementConfig::default(), rest: RestCatalogConfig::default(), @@ -1346,6 +1933,11 @@ mod tests { aws_access_key_id: None, aws_secret_access_key: None, s3_endpoint: None, + gcs_bucket_name: None, + gcs_service_account_path: None, + azure_container_name: None, + azure_storage_account_name: None, + azure_access_key: None, catalog_manager: CatalogManagerConfig::default(), table_management: TableManagementConfig::default(), rest: RestCatalogConfig::default(), @@ -1396,6 +1988,11 @@ mod tests { aws_access_key_id: None, aws_secret_access_key: None, s3_endpoint: None, + gcs_bucket_name: None, + gcs_service_account_path: None, + azure_container_name: None, + azure_storage_account_name: None, + azure_access_key: None, catalog_manager: CatalogManagerConfig::default(), table_management: TableManagementConfig::default(), rest: RestCatalogConfig::default(), @@ -1446,6 +2043,11 @@ mod tests { aws_access_key_id: None, aws_secret_access_key: None, s3_endpoint: None, + gcs_bucket_name: None, + gcs_service_account_path: None, + azure_container_name: None, + azure_storage_account_name: None, + azure_access_key: None, catalog_manager: CatalogManagerConfig::default(), table_management: TableManagementConfig::default(), rest: RestCatalogConfig::default(), @@ -1471,8 +2073,8 @@ mod tests { let config = KafkaSecurityConfig { protocol: Some("SASL_SSL".to_string()), sasl_mechanism: Some("SCRAM-SHA-256".to_string()), - sasl_username: Some("user".to_string()), - sasl_password: Some("pass".to_string()), + sasl_username: Some("user".into()), + sasl_password: Some("pass".into()), ssl_ca_location: Some(PathBuf::from("/path/to/ca.pem")), ssl_cert_location: None, ssl_key_location: None, @@ -1543,7 +2145,7 @@ mod tests { let config = RestCatalogConfig { credential_type: CredentialType::Bearer, - credential: Some("token123".to_string()), + credential: Some("token123".into()), oauth2_token_endpoint: None, oauth2_client_id: None, oauth2_client_secret: None, @@ -1553,7 +2155,7 @@ mod tests { }; assert_eq!(config.credential_type, CredentialType::Bearer); - assert_eq!(config.credential, Some("token123".to_string())); + assert_eq!(config.credential.as_deref(), Some("token123")); assert_eq!(config.request_timeout_seconds, Some(60)); assert_eq!(config.custom_headers.len(), 1); } @@ -1570,4 +2172,606 @@ mod tests { assert!(config.external_id.is_some()); assert!(config.catalog_id.is_some()); } + static ENV_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(()); + + #[test] + fn test_secret_file_ref_from_toml() { + let dir = tempfile::tempdir().unwrap(); + let pass_path = dir.path().join("kafka-password"); + std::fs::write(&pass_path, "hunter2\n").unwrap(); + + let toml = format!( + r#" +[kafka] +bootstrap_servers = ["localhost:9092"] +topic = "events" +consumer_group = "k2i" + +[kafka.security] +protocol = "SASL_SSL" +sasl_password = {{ file = "{}" }} + +[iceberg] +catalog_type = "sql" +warehouse_path = "/tmp/warehouse" +database_name = "db" +table_name = "tbl" +"#, + pass_path.display() + ); + + let config: Config = toml::from_str(&toml).unwrap(); + assert_eq!( + config.kafka.security.sasl_password.as_deref(), + Some("hunter2") + ); + } + + #[test] + fn test_secret_file_ref_missing_file_errors() { + let toml = r#" +[kafka] +bootstrap_servers = ["localhost:9092"] +topic = "events" +consumer_group = "k2i" + +[kafka.security] +sasl_password = { file = "/nonexistent/secret/path" } + +[iceberg] +catalog_type = "sql" +warehouse_path = "/tmp/warehouse" +database_name = "db" +table_name = "tbl" +"#; + + let err = toml::from_str::(toml).unwrap_err(); + assert!(err.to_string().contains("failed to read secret file")); + } + + #[test] + fn test_secret_plain_string_still_works() { + let toml = r#" +[kafka] +bootstrap_servers = ["localhost:9092"] +topic = "events" +consumer_group = "k2i" + +[kafka.security] +sasl_password = "hunter2" + +[iceberg] +catalog_type = "sql" +warehouse_path = "/tmp/warehouse" +database_name = "db" +table_name = "tbl" +"#; + + let config: Config = toml::from_str(toml).unwrap(); + assert_eq!( + config.kafka.security.sasl_password.as_deref(), + Some("hunter2") + ); + } + + #[test] + fn test_secret_debug_is_redacted() { + let config = KafkaSecurityConfig { + sasl_password: Some(Secret::new("hunter2")), + ..KafkaSecurityConfig::default() + }; + let debug = format!("{:?}", config); + assert!(!debug.contains("hunter2")); + assert!(debug.contains("REDACTED")); + } + + #[test] + fn test_env_override_string_field() { + let _guard = ENV_LOCK.lock().unwrap(); + std::env::set_var("K2I_KAFKA_TOPIC", "env-topic"); + + let mut config = test_config(); + config.kafka.topic = "toml-topic".into(); + + config.apply_env_overrides(); + + assert_eq!(config.kafka.topic, "env-topic"); + + std::env::remove_var("K2I_KAFKA_TOPIC"); + } + + #[test] + fn test_env_override_numeric_field() { + let _guard = ENV_LOCK.lock().unwrap(); + std::env::set_var("K2I_KAFKA_BATCH_SIZE", "500"); + + let mut config = test_config(); + config.apply_env_overrides(); + + assert_eq!(config.kafka.batch_size, 500); + + std::env::remove_var("K2I_KAFKA_BATCH_SIZE"); + } + + #[test] + fn test_env_override_iceberg_enum() { + let _guard = ENV_LOCK.lock().unwrap(); + std::env::set_var("K2I_ICEBERG_CATALOG_TYPE", "nessie"); + + let mut config = test_config(); + config.apply_env_overrides(); + + assert_eq!(config.iceberg.catalog_type, CatalogType::Nessie); + + std::env::remove_var("K2I_ICEBERG_CATALOG_TYPE"); + } + + #[test] + fn test_env_override_invalid_numeric_ignored() { + let _guard = ENV_LOCK.lock().unwrap(); + std::env::set_var("K2I_KAFKA_BATCH_SIZE", "not-a-number"); + + let mut config = test_config(); + config.kafka.batch_size = 1000; + config.apply_env_overrides(); + + // Invalid parse should leave the original value intact + assert_eq!(config.kafka.batch_size, 1000); + + std::env::remove_var("K2I_KAFKA_BATCH_SIZE"); + } + + #[test] + fn test_env_override_secret_field() { + let _guard = ENV_LOCK.lock().unwrap(); + std::env::set_var("K2I_KAFKA_SECURITY_SASL_PASSWORD", "env-secret"); + + let mut config = test_config(); + config.apply_env_overrides(); + + assert_eq!( + config.kafka.security.sasl_password.as_deref(), + Some("env-secret") + ); + + std::env::remove_var("K2I_KAFKA_SECURITY_SASL_PASSWORD"); + } + + /// `azure_storage_account_name` is mandatory for Azure backends, so a + /// Kubernetes deployment configuring purely through env must be able to set + /// it. Same for the other cloud-store fields. + #[test] + fn test_env_override_cloud_store_fields() { + let _guard = ENV_LOCK.lock().unwrap(); + std::env::set_var("K2I_ICEBERG_GCS_BUCKET_NAME", "env-bucket"); + std::env::set_var("K2I_ICEBERG_GCS_SERVICE_ACCOUNT_PATH", "/run/gcp/key.json"); + std::env::set_var("K2I_ICEBERG_AZURE_STORAGE_ACCOUNT_NAME", "envacct"); + std::env::set_var("K2I_ICEBERG_AZURE_CONTAINER_NAME", "env-container"); + std::env::set_var("K2I_ICEBERG_AZURE_ACCESS_KEY", "env-azure-key"); + + let mut config = test_config(); + config.apply_env_overrides(); + + assert_eq!( + config.iceberg.gcs_bucket_name.as_deref(), + Some("env-bucket") + ); + assert_eq!( + config.iceberg.gcs_service_account_path.as_deref(), + Some("/run/gcp/key.json") + ); + assert_eq!( + config.iceberg.azure_storage_account_name.as_deref(), + Some("envacct") + ); + assert_eq!( + config.iceberg.azure_container_name.as_deref(), + Some("env-container") + ); + assert_eq!( + config.iceberg.azure_access_key.as_deref(), + Some("env-azure-key") + ); + + for var in [ + "K2I_ICEBERG_GCS_BUCKET_NAME", + "K2I_ICEBERG_GCS_SERVICE_ACCOUNT_PATH", + "K2I_ICEBERG_AZURE_STORAGE_ACCOUNT_NAME", + "K2I_ICEBERG_AZURE_CONTAINER_NAME", + "K2I_ICEBERG_AZURE_ACCESS_KEY", + ] { + std::env::remove_var(var); + } + } + + /// An unparseable boolean must not read as `false` — that would silently + /// disable RPC for a deployment that set `K2I_RPC_ENABLED=yes`. + #[test] + fn test_env_override_rpc_enabled_booleans() { + let _guard = ENV_LOCK.lock().unwrap(); + + for (value, expected) in [ + ("true", true), + ("1", true), + ("yes", true), + ("ON", true), + ("false", false), + ("0", false), + ("no", false), + ] { + std::env::set_var("K2I_RPC_ENABLED", value); + let mut config = test_config(); + config.rpc.enabled = !expected; + config.apply_env_overrides(); + assert_eq!(config.rpc.enabled, expected, "K2I_RPC_ENABLED={value}"); + } + + // Garbage preserves the configured value in both directions. + std::env::set_var("K2I_RPC_ENABLED", "maybe"); + for configured in [true, false] { + let mut config = test_config(); + config.rpc.enabled = configured; + config.apply_env_overrides(); + assert_eq!(config.rpc.enabled, configured); + } + + std::env::remove_var("K2I_RPC_ENABLED"); + } + + /// `Config` derives `Serialize`. Anything that dumps or echoes it — a + /// diagnostic bundle, an RPC response, a future `config dump` — must not be + /// able to emit a credential in the clear. + #[test] + fn test_serializing_config_never_emits_a_plaintext_secret() { + let mut config = test_config(); + config.kafka.security.sasl_username = Some(Secret::new("svc-account")); + config.kafka.security.sasl_password = Some(Secret::new("hunter2")); + config.iceberg.aws_access_key_id = Some(Secret::new("AKIAPLAINTEXT")); + config.iceberg.aws_secret_access_key = Some(Secret::new("aws-secret-value")); + config.iceberg.azure_access_key = Some(Secret::new("azure-secret-value")); + config.iceberg.rest.credential = Some(Secret::new("bearer-token-value")); + config.iceberg.rest.oauth2_client_id = Some(Secret::new("oauth-client-id")); + config.iceberg.rest.oauth2_client_secret = Some(Secret::new("oauth-client-secret")); + + let json = serde_json::to_string(&config).expect("Config should serialize"); + let toml_out = toml::to_string(&config).expect("Config should serialize to TOML"); + let debug = format!("{config:?}"); + + for secret in [ + "svc-account", + "hunter2", + "AKIAPLAINTEXT", + "aws-secret-value", + "azure-secret-value", + "bearer-token-value", + "oauth-client-id", + "oauth-client-secret", + ] { + assert!(!json.contains(secret), "JSON leaked {secret}"); + assert!(!toml_out.contains(secret), "TOML leaked {secret}"); + assert!(!debug.contains(secret), "Debug leaked {secret}"); + } + + assert!(json.contains(Secret::REDACTED)); + assert!(debug.contains(Secret::REDACTED)); + } + + /// `expose()` remains the single deliberate way to read the value, so the + /// redacted output above does not come at the cost of actually using it. + #[test] + fn test_expose_still_returns_the_plaintext() { + let secret = Secret::new("hunter2"); + assert_eq!(secret.expose(), "hunter2"); + assert_eq!(&*secret, "hunter2"); + } + + #[test] + fn test_bucket_from_warehouse() { + assert_eq!( + bucket_from_warehouse("s3://", "s3://my-bucket/warehouse"), + Some("my-bucket".to_string()) + ); + // Bucket with no subpath + assert_eq!( + bucket_from_warehouse("s3://", "s3://my-bucket"), + Some("my-bucket".to_string()) + ); + // Wrong scheme + assert_eq!( + bucket_from_warehouse("s3://", "gs://bucket/warehouse"), + None + ); + // Missing bucket + assert_eq!(bucket_from_warehouse("s3://", "s3:///warehouse"), None); + assert_eq!(bucket_from_warehouse("s3://", "s3://"), None); + } + + #[test] + fn test_warehouse_prefix_after_bucket_strips_scheme_and_bucket() { + // Bucket-only path: no prefix + assert_eq!( + warehouse_prefix_after_bucket("s3://", "s3://my-bucket"), + None + ); + // Trailing slash normalizes to None + assert_eq!( + warehouse_prefix_after_bucket("s3://", "s3://my-bucket/"), + None + ); + // Single-segment prefix + assert_eq!( + warehouse_prefix_after_bucket("s3://", "s3://my-bucket/warehouse"), + Some("warehouse".to_string()) + ); + // Multi-segment prefix preserved + assert_eq!( + warehouse_prefix_after_bucket("gs://", "gs://bucket/warehouse/prod"), + Some("warehouse/prod".to_string()) + ); + // Wrong scheme returns None + assert_eq!( + warehouse_prefix_after_bucket("s3://", "gs://bucket/warehouse"), + None + ); + } + + #[test] + fn test_azure_container_from_warehouse_both_url_forms() { + // Simple form: az://container/path + assert_eq!( + azure_container_from_warehouse("az://my-container/warehouse"), + Some("my-container".to_string()) + ); + // Container-only (no subpath) + assert_eq!( + azure_container_from_warehouse("az://my-container"), + Some("my-container".to_string()) + ); + // Hadoop ABFS form must extract the container BEFORE the `@`, + // not the whole `container@account.dfs.core.windows.net` segment. + assert_eq!( + azure_container_from_warehouse("abfs://container@account.dfs.core.windows.net/path"), + Some("container".to_string()) + ); + // ABFS with multi-segment path + assert_eq!( + azure_container_from_warehouse("abfs://events@prodacct/warehouse/raw"), + Some("events".to_string()) + ); + // Not an azure scheme + assert_eq!(azure_container_from_warehouse("s3://bucket/path"), None); + // Empty container before `@` + assert_eq!(azure_container_from_warehouse("az://@account/path"), None); + } + + #[test] + fn test_azure_prefix_after_container() { + // No subpath past container + assert_eq!(azure_prefix_after_container("az://container"), None); + // Simple form subpath + assert_eq!( + azure_prefix_after_container("az://container/warehouse"), + Some("warehouse".to_string()) + ); + // ABFS form: prefix is the part AFTER the first `/`, unaffected by `@account` + assert_eq!( + azure_prefix_after_container( + "abfs://container@account.dfs.core.windows.net/warehouse/raw" + ), + Some("warehouse/raw".to_string()) + ); + // Wrong scheme + assert_eq!(azure_prefix_after_container("gs://bucket/warehouse"), None); + } + + /// `test_config()` intentionally uses a SQL catalog without its settings, so + /// it fails `validate()` for reasons unrelated to the warehouse path. These + /// tests need a baseline that otherwise passes. + fn validatable_config() -> Config { + let mut config = test_config(); + config.iceberg.catalog_type = CatalogType::Rest; + config.iceberg.sql_catalog = None; + config + } + + /// Azure needs a storage account name that cannot be derived from either + /// URL form. Catching it in `validate()` means a misconfigured deployment + /// fails at startup instead of on the first flush, minutes after the + /// process has already reported healthy. + #[test] + fn test_validate_requires_azure_storage_account_name() { + let mut config = validatable_config(); + config.iceberg.warehouse_path = "az://my-container/warehouse".into(); + + let err = config + .validate() + .expect_err("Azure warehouse without an account name must fail validation"); + assert!( + err.to_string().contains("azure_storage_account_name"), + "error should name the missing field, got: {err}" + ); + + config.iceberg.azure_storage_account_name = Some("myacct".into()); + config.validate().expect("should validate once set"); + } + + #[test] + fn test_validate_accepts_abfs_form_with_account_name() { + let mut config = validatable_config(); + config.iceberg.warehouse_path = "abfs://cont@myacct.dfs.core.windows.net/warehouse".into(); + config.iceberg.azure_storage_account_name = Some("myacct".into()); + config.validate().expect("ABFS form should validate"); + } + + #[test] + fn test_validate_rejects_cloud_warehouse_without_a_bucket() { + for path in ["s3://", "gs://", "s3:///warehouse"] { + let mut config = validatable_config(); + config.iceberg.warehouse_path = path.into(); + assert!( + config.validate().is_err(), + "{path} names no bucket and should fail validation" + ); + } + } + + #[test] + fn test_validate_accepts_ordinary_cloud_and_local_warehouses() { + for path in [ + "s3://bucket/warehouse", + "s3://bucket", + "gs://bucket/warehouse/prod", + "/tmp/warehouse", + "file:///tmp/warehouse", + ] { + let mut config = validatable_config(); + config.iceberg.warehouse_path = path.into(); + config + .validate() + .unwrap_or_else(|e| panic!("{path} should validate, got: {e}")); + } + } + + #[test] + fn test_log_format_parse_env_value() { + assert_eq!(LogFormat::parse_env_value("TEXT"), Some(LogFormat::Text)); + assert_eq!(LogFormat::parse_env_value("json"), Some(LogFormat::Json)); + assert_eq!(LogFormat::parse_env_value("yaml"), None); + } + + /// Values are never shell-interpolated. `docs/configuration.md` once + /// promised `${VAR}` substitution; following it would have authenticated + /// with the literal string. Pin the real behaviour so the documented + /// mechanisms (`{ file = ... }` and `K2I_*`) stay the only two. + #[test] + fn test_secret_values_are_not_shell_interpolated() { + let _guard = ENV_LOCK.lock().unwrap(); + std::env::set_var("SOME_KAFKA_PASSWORD", "real-secret"); + + let toml = r#" +[kafka] +bootstrap_servers = ["localhost:9092"] +topic = "events" +consumer_group = "k2i" + +[kafka.security] +sasl_password = "${SOME_KAFKA_PASSWORD}" + +[iceberg] +catalog_type = "rest" +warehouse_path = "/tmp/warehouse" +database_name = "db" +table_name = "tbl" +"#; + let config: Config = toml::from_str(toml).unwrap(); + assert_eq!( + config.kafka.security.sasl_password.as_deref(), + Some("${SOME_KAFKA_PASSWORD}"), + "`${{VAR}}` is stored literally; it is not an interpolation syntax" + ); + + std::env::remove_var("SOME_KAFKA_PASSWORD"); + } + + /// The env var table in `docs/kubernetes.md` is the deployment contract. + /// If a variable is added to `KNOWN_ENV_VARS` without documenting it, + /// operators cannot discover it — and an undocumented name silently does + /// nothing when mistyped. + #[test] + fn test_known_env_vars_are_documented() { + let docs = std::fs::read_to_string(concat!( + env!("CARGO_MANIFEST_DIR"), + "/../../docs/kubernetes.md" + )) + .expect("docs/kubernetes.md should be readable"); + + let undocumented: Vec<_> = KNOWN_ENV_VARS + .iter() + .filter(|var| !docs.contains(&format!("`{var}`"))) + .collect(); + + assert!( + undocumented.is_empty(), + "these K2I_* variables are not documented in docs/kubernetes.md: {undocumented:?}" + ); + } + + #[test] + fn test_known_env_vars_have_no_duplicates() { + let unique: std::collections::HashSet<_> = KNOWN_ENV_VARS.iter().collect(); + assert_eq!(unique.len(), KNOWN_ENV_VARS.len()); + } + + /// Harness variables share the engine's environment during e2e runs; they + /// must not be reported as typos. + #[test] + fn test_reserved_env_prefixes_cover_harness_variables() { + for var in [ + "K2I_E2E_TOPIC", + "K2I_E2E_WAREHOUSE", + "K2I_PARQUET_PATHS", + "K2I_ICEBERG_METADATA_PATH", + ] { + assert!( + RESERVED_ENV_PREFIXES.iter().any(|p| var.starts_with(p)), + "{var} should be treated as a reserved harness variable" + ); + } + // A genuine typo is still not reserved. + assert!(!RESERVED_ENV_PREFIXES + .iter() + .any(|p| "K2I_KAFKA_TOPC".starts_with(p))); + } + + /// Build a minimal valid Config for override tests. + fn test_config() -> Config { + Config { + kafka: KafkaConfig { + bootstrap_servers: vec!["localhost:9092".into()], + topic: "test".into(), + consumer_group: "test-group".into(), + batch_size: default_batch_size(), + batch_timeout_ms: default_batch_timeout_ms(), + session_timeout_ms: default_session_timeout_ms(), + heartbeat_interval_ms: default_heartbeat_interval_ms(), + max_poll_interval_ms: default_max_poll_interval_ms(), + auto_offset_reset: OffsetReset::Earliest, + security: KafkaSecurityConfig::default(), + format: KafkaFormatConfig::Raw, + }, + schema_evolution: SchemaEvolutionRuntimeConfig::default(), + iceberg: IcebergConfig { + catalog_type: CatalogType::Sql, + warehouse_path: "/tmp/warehouse".into(), + database_name: "db".into(), + table_name: "tbl".into(), + target_file_size_mb: default_target_file_size_mb(), + compression: ParquetCompression::Snappy, + partition_spec: vec![], + rest_uri: None, + hive_metastore_uri: None, + aws_region: None, + aws_access_key_id: None, + aws_secret_access_key: None, + s3_endpoint: None, + catalog_manager: CatalogManagerConfig::default(), + table_management: TableManagementConfig::default(), + rest: RestCatalogConfig::default(), + glue: GlueCatalogConfig::default(), + nessie: None, + sql_catalog: None, + gcs_bucket_name: None, + gcs_service_account_path: None, + azure_container_name: None, + azure_storage_account_name: None, + azure_access_key: None, + object_store: ObjectStoreConfig::default(), + }, + buffer: BufferConfig::default(), + transaction_log: TransactionLogConfig::default(), + maintenance: MaintenanceConfig::default(), + monitoring: MonitoringConfig::default(), + rpc: RpcConfig::default(), + } + } } diff --git a/crates/k2i-core/src/iceberg/catalog.rs b/crates/k2i-core/src/iceberg/catalog.rs index 3f6656c..d86c326 100644 --- a/crates/k2i-core/src/iceberg/catalog.rs +++ b/crates/k2i-core/src/iceberg/catalog.rs @@ -184,6 +184,11 @@ mod tests { aws_access_key_id: None, aws_secret_access_key: None, s3_endpoint: None, + gcs_bucket_name: None, + gcs_service_account_path: None, + azure_container_name: None, + azure_storage_account_name: None, + azure_access_key: None, catalog_manager: Default::default(), table_management: Default::default(), rest: Default::default(), diff --git a/crates/k2i-core/src/iceberg/glue.rs b/crates/k2i-core/src/iceberg/glue.rs index 47429f8..ac4f71d 100644 --- a/crates/k2i-core/src/iceberg/glue.rs +++ b/crates/k2i-core/src/iceberg/glue.rs @@ -131,8 +131,8 @@ impl GlueCatalogClient { { debug!("Using explicit AWS credentials"); let credentials = aws_credential_types::Credentials::new( - access_key, - secret_key, + access_key.expose(), + secret_key.expose(), None, // session token None, // expiry "k2i-explicit-credentials", @@ -801,6 +801,11 @@ mod tests { aws_access_key_id: Some("test_key".into()), aws_secret_access_key: Some("test_secret".into()), s3_endpoint: None, + gcs_bucket_name: None, + gcs_service_account_path: None, + azure_container_name: None, + azure_storage_account_name: None, + azure_access_key: None, catalog_manager: Default::default(), table_management: Default::default(), rest: Default::default(), diff --git a/crates/k2i-core/src/iceberg/hive.rs b/crates/k2i-core/src/iceberg/hive.rs index 9f12667..53f71e3 100644 --- a/crates/k2i-core/src/iceberg/hive.rs +++ b/crates/k2i-core/src/iceberg/hive.rs @@ -894,6 +894,11 @@ mod tests { aws_access_key_id: None, aws_secret_access_key: None, s3_endpoint: None, + gcs_bucket_name: None, + gcs_service_account_path: None, + azure_container_name: None, + azure_storage_account_name: None, + azure_access_key: None, catalog_manager: Default::default(), table_management: Default::default(), rest: Default::default(), diff --git a/crates/k2i-core/src/iceberg/nessie.rs b/crates/k2i-core/src/iceberg/nessie.rs index 9cf64d9..ec89890 100644 --- a/crates/k2i-core/src/iceberg/nessie.rs +++ b/crates/k2i-core/src/iceberg/nessie.rs @@ -115,7 +115,11 @@ impl NessieCatalogClient { timeout, max_retries: config.catalog_manager.max_retries, credential_type: config.rest.credential_type.clone(), - bearer_token: config.rest.credential.clone(), + bearer_token: config + .rest + .credential + .as_ref() + .map(|s| s.expose().to_string()), warehouse_path: config.warehouse_path.clone(), default_reference: default_ref.clone(), api_version, @@ -819,6 +823,11 @@ mod tests { aws_access_key_id: None, aws_secret_access_key: None, s3_endpoint: None, + gcs_bucket_name: None, + gcs_service_account_path: None, + azure_container_name: None, + azure_storage_account_name: None, + azure_access_key: None, catalog_manager: Default::default(), table_management: Default::default(), rest: Default::default(), diff --git a/crates/k2i-core/src/iceberg/official.rs b/crates/k2i-core/src/iceberg/official.rs index 0bf0613..4e7ddd3 100644 --- a/crates/k2i-core/src/iceberg/official.rs +++ b/crates/k2i-core/src/iceberg/official.rs @@ -776,15 +776,15 @@ fn apply_rest_auth_props( let token = config.rest.credential.as_ref().ok_or_else(|| { Error::Config("REST bearer auth requires iceberg.rest.credential".into()) })?; - props.insert("token".to_string(), token.clone()); + props.insert("token".to_string(), token.expose().to_string()); } CredentialType::OAuth2 => { let client_secret = config.rest.oauth2_client_secret.as_ref().ok_or_else(|| { Error::Config("REST OAuth2 requires iceberg.rest.oauth2_client_secret".into()) })?; let credential = match &config.rest.oauth2_client_id { - Some(client_id) => format!("{}:{}", client_id, client_secret), - None => client_secret.clone(), + Some(client_id) => format!("{}:{}", client_id.expose(), client_secret.expose()), + None => client_secret.expose().to_string(), }; props.insert("credential".to_string(), credential); if let Some(endpoint) = &config.rest.oauth2_token_endpoint { @@ -804,10 +804,16 @@ fn apply_file_io_props(config: &IcebergConfig, props: &mut HashMap for CatalogCommitOutcome { pub struct IcebergWriter { config: IcebergConfig, object_store: Arc, + /// In-bucket prefix derived from `warehouse_path` for cloud backends where + /// the object store root is the bucket, not the warehouse. `None` for local + /// filesystem (handled by `LocalFileSystem::new_with_prefix`). + /// + /// This is applied **only** when addressing the object store (see + /// [`IcebergWriter::storage_path`]). Every path handed to the catalog, the + /// transaction log, or the read path stays warehouse-relative, because those + /// consumers resolve it against `warehouse_path` themselves. + warehouse_prefix: Option, txlog: Option>, write_count: AtomicU64, /// Optional catalog operations for real catalog integration @@ -186,10 +198,11 @@ impl IcebergWriterBuilder { /// Build the IcebergWriter. pub async fn build(self) -> Result { - let object_store = IcebergWriter::create_object_store(&self.config)?; + let (object_store, warehouse_prefix) = IcebergWriter::create_object_store(&self.config)?; Ok(IcebergWriter { config: self.config, object_store, + warehouse_prefix, txlog: self.txlog, write_count: AtomicU64::new(0), catalog: self.catalog, @@ -235,7 +248,16 @@ impl IcebergWriter { } /// Create object store based on configuration. - fn create_object_store(config: &IcebergConfig) -> Result> { + /// + /// Returns the store and the in-bucket warehouse prefix that + /// [`IcebergWriter::storage_path`] prepends when addressing that store. For + /// cloud backends the store root is the bucket, so the prefix keeps uploads + /// aligned with the warehouse path recorded by the catalog/txlog. For local + /// filesystem the prefix is `None` because `LocalFileSystem::new_with_prefix` + /// handles it. + fn create_object_store( + config: &IcebergConfig, + ) -> Result<(Arc, Option)> { let warehouse_path = &config.warehouse_path; if warehouse_path.starts_with("s3://") { @@ -250,32 +272,35 @@ impl IcebergWriter { } } - fn create_s3_store(config: &IcebergConfig) -> Result> { + fn create_s3_store(config: &IcebergConfig) -> Result<(Arc, Option)> { use object_store::aws::AmazonS3Builder; - let bucket = config - .warehouse_path - .strip_prefix("s3://") - .and_then(|s| s.split('/').next()) - .ok_or_else(|| { - Error::Iceberg(IcebergError::CatalogConnection("Invalid S3 path".into())) - })?; + let bucket = bucket_from_warehouse("s3://", &config.warehouse_path).ok_or_else(|| { + Error::Iceberg(IcebergError::CatalogConnection("Invalid S3 path".into())) + })?; + + let warehouse_prefix = warehouse_prefix_after_bucket("s3://", &config.warehouse_path); let mut builder = AmazonS3Builder::new().with_bucket_name(bucket); - if let Some(ref region) = config.aws_region { + if let Some(region) = &config.aws_region { builder = builder.with_region(region); } - if let Some(ref access_key) = config.aws_access_key_id { - builder = builder.with_access_key_id(access_key); + // When access key / secret key are omitted, AmazonS3Builder falls through + // to the default credential chain: environment variables → IMDS → IRSA + // (IAM Roles for Service Accounts). This enables EKS with IRSA, + // EC2 instance profiles, and local dev with env vars like AWS_ACCESS_KEY_ID + // without any explicit config. + if let Some(access_key) = &config.aws_access_key_id { + builder = builder.with_access_key_id(access_key.expose()); } - if let Some(ref secret_key) = config.aws_secret_access_key { - builder = builder.with_secret_access_key(secret_key); + if let Some(secret_key) = &config.aws_secret_access_key { + builder = builder.with_secret_access_key(secret_key.expose()); } - if let Some(ref endpoint) = config.s3_endpoint { + if let Some(endpoint) = &config.s3_endpoint { builder = builder .with_endpoint(endpoint) .with_allow_http(endpoint.starts_with("http://")); @@ -285,24 +310,84 @@ impl IcebergWriter { .build() .map_err(|e| Error::Iceberg(IcebergError::FileUpload(e.to_string())))?; - Ok(Arc::new(store)) + Ok((Arc::new(store), warehouse_prefix)) } + fn create_gcs_store(config: &IcebergConfig) -> Result<(Arc, Option)> { + use object_store::gcp::GoogleCloudStorageBuilder; + + let bucket = config + .gcs_bucket_name + .clone() + .or_else(|| bucket_from_warehouse("gs://", &config.warehouse_path)) + .ok_or_else(|| { + Error::Iceberg(IcebergError::CatalogConnection( + "Invalid GCS path: could not determine bucket".into(), + )) + })?; + + // The in-bucket prefix is derived from warehouse_path regardless of + // whether gcs_bucket_name overrides the bucket, since the override only + // changes the bucket and leaves the path structure intact. + let warehouse_prefix = warehouse_prefix_after_bucket("gs://", &config.warehouse_path); - fn create_gcs_store(_config: &IcebergConfig) -> Result> { - // GCS support - for now just return an error, implement when needed - Err(Error::Iceberg(IcebergError::CatalogConnection( - "GCS storage not yet implemented".into(), - ))) + let mut builder = GoogleCloudStorageBuilder::new().with_bucket_name(&bucket); + + if let Some(sa_path) = &config.gcs_service_account_path { + builder = builder.with_service_account_path(sa_path); + } + // When no service account path is set, the builder falls through to + // Application Default Credentials (Workload Identity on GKE, env vars locally). + + let store = builder + .build() + .map_err(|e| Error::Iceberg(IcebergError::FileUpload(e.to_string())))?; + + Ok((Arc::new(store), warehouse_prefix)) } + fn create_azure_store( + config: &IcebergConfig, + ) -> Result<(Arc, Option)> { + use object_store::azure::MicrosoftAzureBuilder; + + let account_name = config.azure_storage_account_name.as_deref().ok_or_else(|| { + Error::Iceberg(IcebergError::CatalogConnection( + "Azure storage account name is required (set `azure_storage_account_name` in config)" + .into(), + )) + })?; + + let container = config + .azure_container_name + .clone() + .or_else(|| azure_container_from_warehouse(&config.warehouse_path)) + .ok_or_else(|| { + Error::Iceberg(IcebergError::CatalogConnection( + "Invalid Azure path: could not determine container".into(), + )) + })?; + + let warehouse_prefix = azure_prefix_after_container(&config.warehouse_path); - fn create_azure_store(_config: &IcebergConfig) -> Result> { - // Azure support - for now just return an error, implement when needed - Err(Error::Iceberg(IcebergError::CatalogConnection( - "Azure storage not yet implemented".into(), - ))) + let mut builder = MicrosoftAzureBuilder::new() + .with_account(account_name) + .with_container_name(&container); + + if let Some(access_key) = &config.azure_access_key { + builder = builder.with_access_key(access_key.expose()); + } + // When no access key is set, the builder falls through to + // the DefaultAzureCredential chain (Managed Identity on AKS, env vars locally). + + let store = builder + .build() + .map_err(|e| Error::Iceberg(IcebergError::FileUpload(e.to_string())))?; + + Ok((Arc::new(store), warehouse_prefix)) } - fn create_local_store(config: &IcebergConfig) -> Result> { + fn create_local_store( + config: &IcebergConfig, + ) -> Result<(Arc, Option)> { use object_store::local::LocalFileSystem; let path = std::path::Path::new(&config.warehouse_path); @@ -324,7 +409,9 @@ impl IcebergWriter { ))) })?; - Ok(Arc::new(store)) + // LocalFileSystem::new_with_prefix already roots uploads at the + // warehouse path, so no in-bucket prefix is needed. + Ok((Arc::new(store), None)) } /// Write a RecordBatch to Iceberg. @@ -533,8 +620,12 @@ impl IcebergWriter { } /// Upload a file to object storage. + /// + /// `path` is warehouse-relative; the in-bucket prefix is applied here so the + /// warehouse-relative form is what every other consumer sees. async fn upload_file(&self, path: &str, data: Bytes) -> Result<()> { - let object_path = ObjectPath::from(path); + let key = self.storage_path(path); + let object_path = ObjectPath::from(key.as_str()); let payload = PutPayload::from_bytes(data); self.object_store @@ -543,7 +634,7 @@ impl IcebergWriter { .map_err(|e| { Error::Iceberg(IcebergError::FileUpload(format!( "Failed to upload file to {}: {}", - path, e + key, e ))) })?; @@ -761,6 +852,11 @@ impl IcebergWriter { ); // Format: data/{db}/{table}/{time_partition}/kafka_partition={N}/part-{uuid}-{offset_range}.parquet + // + // Deliberately warehouse-relative: the catalog (`data_file_uri`), the + // read path (`absolute_data_path`), and the txlog all join this against + // `warehouse_path`. The in-bucket prefix is added separately by + // `storage_path` when addressing the object store. format!( "data/{}/{}/{}/kafka_partition={}/part-{}-{}.parquet", self.config.database_name, @@ -772,6 +868,22 @@ impl IcebergWriter { ) } + /// Map a warehouse-relative path to the key used to address the object store. + /// + /// For cloud backends the store is rooted at the bucket, so the in-bucket + /// warehouse prefix must be prepended: a warehouse of `s3://bucket/warehouse` + /// turns `data/db/tbl/f.parquet` into `warehouse/data/db/tbl/f.parquet`, which + /// lands at `s3://bucket/warehouse/data/db/tbl/f.parquet` — exactly where the + /// catalog's `warehouse_path` + relative-path join points. For the local + /// filesystem the store is already rooted at the warehouse, so the path is + /// returned unchanged. + fn storage_path(&self, relative_path: &str) -> String { + match &self.warehouse_prefix { + Some(prefix) => format!("{}/{}", prefix, relative_path.trim_start_matches('/')), + None => relative_path.to_string(), + } + } + /// Extract partition information from a RecordBatch. /// /// Extracts kafka partition, timestamp, and offset range from the batch columns. @@ -891,6 +1003,11 @@ mod tests { aws_access_key_id: None, aws_secret_access_key: None, s3_endpoint: None, + gcs_bucket_name: None, + gcs_service_account_path: None, + azure_container_name: None, + azure_storage_account_name: None, + azure_access_key: None, catalog_manager: Default::default(), table_management: Default::default(), rest: Default::default(), @@ -1149,4 +1266,149 @@ mod tests { assert!(stats.snapshot_id > 0); assert!(!writer.has_catalog_integration()); } + fn test_partition_info() -> PartitionInfo { + PartitionInfo { + topic: "test".to_string(), + kafka_partition: 0, + event_timestamp_ms: chrono::Utc::now().timestamp_millis(), + min_offset: 100, + max_offset: 200, + min_lsn: 1, + max_lsn: 2, + } + } + + #[tokio::test] + async fn test_generate_file_path_has_no_prefix_for_local() { + // Local filesystem store roots at the warehouse path itself, so the + // generated path must NOT carry an in-bucket prefix. + let temp_dir = TempDir::new().unwrap(); + let config = create_test_config(temp_dir.path().to_str().unwrap()); + let writer = IcebergWriter::new(config).await.unwrap(); + + let path = writer.generate_file_path(&test_partition_info()); + assert!( + path.starts_with("data/test_db/"), + "local store path should NOT have a warehouse prefix, got: {path}" + ); + assert_eq!(writer.warehouse_prefix, None); + assert_eq!(writer.storage_path(&path), path); + } + + /// The catalog (`data_file_uri`), the read path (`absolute_data_path`) and + /// the txlog all resolve a data file by joining `warehouse_path` with the + /// writer's path. If `generate_file_path` itself carried the in-bucket + /// prefix, that join would double it — the catalog would point at + /// `s3://bucket/warehouse/warehouse/data/...` while the upload landed at + /// `s3://bucket/warehouse/data/...`, so every committed file would be + /// unreadable. + #[tokio::test] + async fn test_cloud_file_path_is_warehouse_relative_but_upload_key_is_prefixed() { + let mut config = create_test_config("s3://my-bucket/warehouse"); + config.aws_region = Some("us-east-1".into()); + config.aws_access_key_id = Some("test-key".into()); + config.aws_secret_access_key = Some("test-secret".into()); + + let writer = IcebergWriter::new(config).await.unwrap(); + assert_eq!(writer.warehouse_prefix.as_deref(), Some("warehouse")); + + let relative = writer.generate_file_path(&test_partition_info()); + assert!( + relative.starts_with("data/test_db/test_table/"), + "catalog-visible path must stay warehouse-relative, got: {relative}" + ); + assert!( + !relative.starts_with("warehouse/"), + "catalog-visible path must not carry the in-bucket prefix, got: {relative}" + ); + + // The object store is rooted at the bucket, so the upload key does carry it. + let key = writer.storage_path(&relative); + assert_eq!(key, format!("warehouse/{relative}")); + + // Joining warehouse_path with the relative path, as `data_file_uri` and + // `absolute_data_path` do, must land on exactly that object. + let catalog_uri = format!("{}/{}", writer.config.warehouse_path, relative); + assert_eq!(catalog_uri, format!("s3://my-bucket/{key}")); + } + + /// A bucket-root warehouse has no prefix, so the upload key and the + /// catalog-visible path must be identical. + #[tokio::test] + async fn test_cloud_bucket_root_warehouse_has_no_prefix() { + let mut config = create_test_config("s3://my-bucket"); + config.aws_region = Some("us-east-1".into()); + config.aws_access_key_id = Some("test-key".into()); + config.aws_secret_access_key = Some("test-secret".into()); + + let writer = IcebergWriter::new(config).await.unwrap(); + assert_eq!(writer.warehouse_prefix, None); + + let relative = writer.generate_file_path(&test_partition_info()); + assert_eq!(writer.storage_path(&relative), relative); + } + + /// The credential fall-through paths must construct a store *without* + /// credentials present, so GKE Workload Identity / AKS Managed Identity + /// resolve at request time rather than failing at startup. + #[tokio::test] + async fn test_cloud_stores_build_without_explicit_credentials() { + let gcs = IcebergWriter::new(create_test_config("gs://my-bucket/warehouse/prod")) + .await + .expect("GCS store should build and defer to Application Default Credentials"); + assert_eq!(gcs.warehouse_prefix.as_deref(), Some("warehouse/prod")); + + let mut azure_config = + create_test_config("abfs://cont@myacct.dfs.core.windows.net/warehouse/raw"); + azure_config.azure_storage_account_name = Some("myacct".into()); + let azure = IcebergWriter::new(azure_config) + .await + .expect("Azure store should build and defer to DefaultAzureCredential"); + assert_eq!(azure.warehouse_prefix.as_deref(), Some("warehouse/raw")); + } + + /// The account name cannot be derived from either URL form, so its absence + /// must surface as a named configuration error rather than an opaque one. + #[tokio::test] + async fn test_azure_without_account_name_reports_the_missing_field() { + let err = match IcebergWriter::new(create_test_config("az://cont/warehouse")).await { + Err(err) => err, + Ok(_) => panic!("Azure without an account name must fail"), + }; + assert!( + err.to_string().contains("azure_storage_account_name"), + "error should name the missing field, got: {err}" + ); + } + + /// `gcs_bucket_name` overrides only the bucket; the in-bucket prefix still + /// comes from `warehouse_path`. + #[tokio::test] + async fn test_gcs_bucket_override_keeps_warehouse_prefix() { + let mut config = create_test_config("gs://path-bucket/warehouse"); + config.gcs_bucket_name = Some("override-bucket".into()); + let writer = IcebergWriter::new(config).await.unwrap(); + assert_eq!(writer.warehouse_prefix.as_deref(), Some("warehouse")); + } + + /// An end-to-end proof against a real (local) store: the path the writer + /// reports must be resolvable against `warehouse_path` on disk. + #[tokio::test] + async fn test_reported_file_path_resolves_against_warehouse_path() { + let temp_dir = TempDir::new().unwrap(); + let warehouse = temp_dir.path().to_str().unwrap().to_string(); + let writer = IcebergWriter::new(create_test_config(&warehouse)) + .await + .unwrap(); + + let stats = writer.write_batch(create_test_batch(), 100).await.unwrap(); + + let resolved = std::path::Path::new(&warehouse).join(&stats.file_path); + assert!( + resolved.exists(), + "reported file_path {} must resolve under warehouse {}", + stats.file_path, + warehouse + ); + } } diff --git a/crates/k2i-core/src/kafka/consumer.rs b/crates/k2i-core/src/kafka/consumer.rs index 24ba042..96c6dfc 100644 --- a/crates/k2i-core/src/kafka/consumer.rs +++ b/crates/k2i-core/src/kafka/consumer.rs @@ -167,11 +167,11 @@ impl KafkaConsumerBuilder { if let Some(ref mechanism) = self.config.security.sasl_mechanism { client_config.set("sasl.mechanism", mechanism); } - if let Some(ref username) = self.config.security.sasl_username { - client_config.set("sasl.username", username); + if let Some(username) = &self.config.security.sasl_username { + client_config.set("sasl.username", username.expose()); } - if let Some(ref password) = self.config.security.sasl_password { - client_config.set("sasl.password", password); + if let Some(password) = &self.config.security.sasl_password { + client_config.set("sasl.password", password.expose()); } if let Some(ref path) = self.config.security.ssl_ca_location { client_config.set("ssl.ca.location", path.to_string_lossy().as_ref()); diff --git a/crates/k2i-core/src/read/mod.rs b/crates/k2i-core/src/read/mod.rs index 32db860..30b1fb4 100644 --- a/crates/k2i-core/src/read/mod.rs +++ b/crates/k2i-core/src/read/mod.rs @@ -436,6 +436,11 @@ mod tests { aws_access_key_id: None, aws_secret_access_key: None, s3_endpoint: None, + gcs_bucket_name: None, + gcs_service_account_path: None, + azure_container_name: None, + azure_storage_account_name: None, + azure_access_key: None, catalog_manager: crate::config::CatalogManagerConfig::default(), table_management: crate::config::TableManagementConfig::default(), rest: crate::config::RestCatalogConfig::default(), diff --git a/crates/k2i-core/tests/integration_tests.rs b/crates/k2i-core/tests/integration_tests.rs index 182f849..aca13a6 100644 --- a/crates/k2i-core/tests/integration_tests.rs +++ b/crates/k2i-core/tests/integration_tests.rs @@ -662,3 +662,253 @@ mod end_to_end { assert!(!entries.is_empty()); } } + +/// End-to-end verification that data files land where the Iceberg catalog says +/// they do, against a real S3 implementation (MinIO). +/// +/// For cloud backends the object store is rooted at the *bucket*, while the +/// catalog, the transaction log, and the read path all locate a data file by +/// joining `warehouse_path` with the writer's reported path. A warehouse with +/// an in-bucket prefix (`s3://bucket/warehouse`) is therefore the case where +/// those two views can silently disagree — and a disagreement means every +/// committed file is unreadable while the pipeline reports success. +/// +/// These tests pin the contract: the reported path is warehouse-relative, and +/// `warehouse_path` + reported path resolves to a real object. +mod s3_object_store_integration { + use k2i_core::config::{ + CatalogManagerConfig, CatalogType, GlueCatalogConfig, IcebergConfig, ObjectStoreConfig, + ParquetCompression, RestCatalogConfig, TableManagementConfig, + }; + use k2i_core::iceberg::IcebergWriter; + use object_store::aws::AmazonS3Builder; + use object_store::path::Path as ObjectPath; + use object_store::ObjectStore; + use testcontainers::core::{ContainerPort, WaitFor}; + use testcontainers::runners::AsyncRunner; + use testcontainers::{GenericImage, ImageExt}; + + use arrow::array::{Int32Array, Int64Array, StringArray}; + use arrow::datatypes::{DataType, Field, Schema}; + use arrow::record_batch::RecordBatch; + use std::sync::Arc; + + const BUCKET: &str = "k2i-test-bucket"; + const ACCESS_KEY: &str = "minioadmin"; + const SECRET_KEY: &str = "minioadmin"; + + /// MinIO serves each top-level directory under its data dir as a bucket. + /// Seeding a file inside `/data//` therefore provisions the bucket + /// before startup, avoiding a dependency on an S3 admin client purely to + /// issue a CreateBucket call. The image entrypoint prefixes `minio` to the + /// command, so the command cannot be used to run a shell. + fn minio_image() -> testcontainers::ContainerRequest { + GenericImage::new("minio/minio", "RELEASE.2022-02-07T08-17-33Z") + .with_wait_for(WaitFor::message_on_stdout("API:")) + .with_exposed_port(ContainerPort::Tcp(9000)) + .with_env_var("MINIO_CONSOLE_ADDRESS", ":9001") + .with_env_var("MINIO_ROOT_USER", ACCESS_KEY) + .with_env_var("MINIO_ROOT_PASSWORD", SECRET_KEY) + .with_copy_to(format!("/data/{BUCKET}/.keep"), Vec::::new()) + .with_cmd(vec!["server".to_string(), "/data".to_string()]) + } + + fn s3_config(warehouse_path: &str, endpoint: &str) -> IcebergConfig { + IcebergConfig { + catalog_type: CatalogType::Rest, + warehouse_path: warehouse_path.to_string(), + database_name: "test_db".to_string(), + table_name: "test_table".to_string(), + target_file_size_mb: 128, + compression: ParquetCompression::Snappy, + partition_spec: vec![], + rest_uri: None, + hive_metastore_uri: None, + aws_region: Some("us-east-1".to_string()), + aws_access_key_id: Some(ACCESS_KEY.into()), + aws_secret_access_key: Some(SECRET_KEY.into()), + s3_endpoint: Some(endpoint.to_string()), + gcs_bucket_name: None, + gcs_service_account_path: None, + azure_container_name: None, + azure_storage_account_name: None, + azure_access_key: None, + catalog_manager: CatalogManagerConfig::default(), + table_management: TableManagementConfig::default(), + rest: RestCatalogConfig::default(), + glue: GlueCatalogConfig::default(), + nessie: None, + sql_catalog: None, + object_store: ObjectStoreConfig::default(), + } + } + + fn test_batch() -> RecordBatch { + let schema = Arc::new(Schema::new(vec![ + Field::new("id", DataType::Int32, false), + Field::new("name", DataType::Utf8, true), + Field::new("partition", DataType::Int32, false), + Field::new("offset", DataType::Int64, false), + Field::new("timestamp", DataType::Int64, false), + ])); + let now = chrono::Utc::now().timestamp_millis(); + RecordBatch::try_new( + schema, + vec![ + Arc::new(Int32Array::from(vec![1, 2, 3])), + Arc::new(StringArray::from(vec!["a", "b", "c"])), + Arc::new(Int32Array::from(vec![0, 0, 0])), + Arc::new(Int64Array::from(vec![100, 101, 102])), + Arc::new(Int64Array::from(vec![now, now, now])), + ], + ) + .expect("failed to build test batch") + } + + /// A store rooted at the bucket, matching how an external reader (Spark, + /// DuckDB, Trino) would resolve the URI the catalog recorded. + fn bucket_rooted_store(endpoint: &str) -> impl ObjectStore { + AmazonS3Builder::new() + .with_bucket_name(BUCKET) + .with_region("us-east-1") + .with_access_key_id(ACCESS_KEY) + .with_secret_access_key(SECRET_KEY) + .with_endpoint(endpoint) + .with_allow_http(true) + .build() + .expect("failed to build verification store") + } + + #[tokio::test] + #[ignore = "requires Docker"] + async fn test_s3_prefixed_warehouse_file_lands_where_catalog_points() { + let container = minio_image().start().await.expect("failed to start MinIO"); + let port = container + .get_host_port_ipv4(9000) + .await + .expect("failed to get MinIO port"); + let endpoint = format!("http://127.0.0.1:{port}"); + + // The interesting case: a warehouse with an in-bucket prefix. + let warehouse = format!("s3://{BUCKET}/warehouse"); + let writer = IcebergWriter::new(s3_config(&warehouse, &endpoint)) + .await + .expect("failed to build S3 writer"); + + let stats = writer + .write_batch(test_batch(), 102) + .await + .expect("write_batch against MinIO should succeed"); + + // 1. The reported path is warehouse-relative — it must NOT already + // contain the in-bucket prefix, or consumers that join it against + // `warehouse_path` would double it. + assert!( + stats.file_path.starts_with("data/test_db/test_table/"), + "reported path should be warehouse-relative, got: {}", + stats.file_path + ); + assert!( + !stats.file_path.starts_with("warehouse/"), + "reported path must not carry the in-bucket prefix, got: {}", + stats.file_path + ); + + let store = bucket_rooted_store(&endpoint); + + // 2. Joining `warehouse_path` with the reported path — exactly what the + // catalog records and what an external reader resolves — must land on + // a real object of the right size. + let expected_key = ObjectPath::from(format!("warehouse/{}", stats.file_path)); + let meta = store.head(&expected_key).await.unwrap_or_else(|e| { + panic!( + "catalog URI {}/{} does not resolve to a stored object: {e}", + warehouse, stats.file_path + ) + }); + assert_eq!( + meta.size as usize, stats.file_size_bytes, + "stored object size should match the reported write size" + ); + + // 3. Nothing was written to the double-prefixed location. + let doubled = ObjectPath::from(format!("warehouse/warehouse/{}", stats.file_path)); + assert!( + store.head(&doubled).await.is_err(), + "object must not be written under a doubled warehouse prefix" + ); + + // 4. Nothing was written at the bucket root either, which is where + // uploads landed before the prefix was applied at all. + let bucket_root = ObjectPath::from(stats.file_path.clone()); + assert!( + store.head(&bucket_root).await.is_err(), + "object must not be written at the bucket root, bypassing the warehouse prefix" + ); + } + + /// A bucket-root warehouse has no prefix to apply, so the reported path and + /// the storage key must coincide. + #[tokio::test] + #[ignore = "requires Docker"] + async fn test_s3_bucket_root_warehouse_writes_at_bucket_root() { + let container = minio_image().start().await.expect("failed to start MinIO"); + let port = container + .get_host_port_ipv4(9000) + .await + .expect("failed to get MinIO port"); + let endpoint = format!("http://127.0.0.1:{port}"); + + let warehouse = format!("s3://{BUCKET}"); + let writer = IcebergWriter::new(s3_config(&warehouse, &endpoint)) + .await + .expect("failed to build S3 writer"); + + let stats = writer + .write_batch(test_batch(), 102) + .await + .expect("write_batch against MinIO should succeed"); + + let store = bucket_rooted_store(&endpoint); + let key = ObjectPath::from(stats.file_path.clone()); + let meta = store.head(&key).await.unwrap_or_else(|e| { + panic!( + "object {} should exist at bucket root: {e}", + stats.file_path + ) + }); + assert_eq!(meta.size as usize, stats.file_size_bytes); + } + + /// A multi-segment prefix (`s3://bucket/warehouse/prod`) must be preserved + /// in full, not collapsed to its first segment. + #[tokio::test] + #[ignore = "requires Docker"] + async fn test_s3_multi_segment_warehouse_prefix_is_preserved() { + let container = minio_image().start().await.expect("failed to start MinIO"); + let port = container + .get_host_port_ipv4(9000) + .await + .expect("failed to get MinIO port"); + let endpoint = format!("http://127.0.0.1:{port}"); + + let warehouse = format!("s3://{BUCKET}/warehouse/prod"); + let writer = IcebergWriter::new(s3_config(&warehouse, &endpoint)) + .await + .expect("failed to build S3 writer"); + + let stats = writer + .write_batch(test_batch(), 102) + .await + .expect("write_batch against MinIO should succeed"); + + let store = bucket_rooted_store(&endpoint); + let expected_key = ObjectPath::from(format!("warehouse/prod/{}", stats.file_path)); + store.head(&expected_key).await.unwrap_or_else(|e| { + panic!( + "catalog URI {}/{} does not resolve to a stored object: {e}", + warehouse, stats.file_path + ) + }); + } +} diff --git a/crates/k2i-rpc-server/examples/fixture_server.rs b/crates/k2i-rpc-server/examples/fixture_server.rs index ef88692..c0007c2 100644 --- a/crates/k2i-rpc-server/examples/fixture_server.rs +++ b/crates/k2i-rpc-server/examples/fixture_server.rs @@ -102,6 +102,11 @@ fn fixture_config(socket_path: PathBuf) -> Config { aws_access_key_id: None, aws_secret_access_key: None, s3_endpoint: None, + gcs_bucket_name: None, + gcs_service_account_path: None, + azure_container_name: None, + azure_storage_account_name: None, + azure_access_key: None, catalog_manager: CatalogManagerConfig::default(), table_management: TableManagementConfig::default(), rest: RestCatalogConfig::default(), diff --git a/crates/k2i-rpc-server/src/lib.rs b/crates/k2i-rpc-server/src/lib.rs index d5c839e..f339129 100644 --- a/crates/k2i-rpc-server/src/lib.rs +++ b/crates/k2i-rpc-server/src/lib.rs @@ -274,6 +274,11 @@ mod tests { aws_access_key_id: None, aws_secret_access_key: None, s3_endpoint: None, + gcs_bucket_name: None, + gcs_service_account_path: None, + azure_container_name: None, + azure_storage_account_name: None, + azure_access_key: None, catalog_manager: CatalogManagerConfig::default(), table_management: TableManagementConfig::default(), rest: RestCatalogConfig::default(), diff --git a/docs/architecture.md b/docs/architecture.md index 9dca9ef..e71aa12 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -169,7 +169,7 @@ See [Iceberg REST Catalog](./iceberg-rest-catalog.md). - Startup recovery state is computed, but Kafka seeking/deduplication and startup orphan cleanup need further wiring. - Kafka commits are async in the current helper. - Transaction-log entries are flushed, but not every entry is fsynced individually. -- GCS and Azure object-store configuration is declared, but writer creation is not complete for those backends. +- S3, GCS, and Azure object stores are wired end to end. S3 is covered by a container-backed round-trip test (MinIO); GCS and Azure are covered only at the configuration and store-construction level, so validate their credentials in your own environment before rollout. - Maintenance commands and task implementations exist; scheduler wiring should be reviewed for each deployment. See [Production Readiness](./production-readiness.md). diff --git a/docs/configuration.md b/docs/configuration.md index 8605db4..c94c02e 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -255,18 +255,31 @@ Use `manual` when an operator or deployment pipeline owns table schema changes. | Option | Type | Required | Default | Description | |--------|------|----------|---------|-------------| | `aws_region` | String | No | - | AWS region for S3 | -| `aws_access_key_id` | String | No | - | AWS access key ID | -| `aws_secret_access_key` | String | No | - | AWS secret access key | +| `aws_access_key_id` | Secret | No | - | AWS access key ID; omit to use the default credential chain | +| `aws_secret_access_key` | Secret | No | - | AWS secret access key; omit to use the default credential chain | | `s3_endpoint` | String | No | - | Custom S3 endpoint (for MinIO, LocalStack) | +| `gcs_bucket_name` | String | No | parsed from `warehouse_path` | Overrides the GCS bucket | +| `gcs_service_account_path` | String | No | - | Service account key file; omit to use Application Default Credentials | +| `azure_storage_account_name` | String | Yes (Azure) | - | Azure storage account; cannot be derived from `warehouse_path` | +| `azure_container_name` | String | No | parsed from `warehouse_path` | Overrides the Azure container | +| `azure_access_key` | Secret | No | - | Azure storage key; omit to use `DefaultAzureCredential` | + +Fields marked `Secret` accept either a plain string or a `{ file = "path" }` +table. See [Injecting Secrets](#injecting-secrets). #### Warehouse Path Formats -| Storage | Format | Current writer status | -|---------|--------|-----------------------| -| AWS S3 / S3-compatible | `s3://bucket-name/path/` | Supported path; validate credentials and endpoint in your environment | -| Local filesystem | `file:///absolute/path/` | Supported for local development and tests | -| Google Cloud Storage | `gs://bucket-name/path/` | Declared in configuration, but writer creation still needs backend wiring | -| Azure Blob Storage | `az://container/path/` | Declared in configuration, but writer creation still needs backend wiring | +| Storage | Format | Credential fall-through when keys are omitted | +|---------|--------|-----------------------------------------------| +| AWS S3 / S3-compatible | `s3://bucket-name/path/` | Env vars → IMDS (EC2 instance profiles) → IRSA (EKS) | +| Local filesystem | `file:///absolute/path/` | n/a — used for local development and tests | +| Google Cloud Storage | `gs://bucket-name/path/` | Application Default Credentials (GKE Workload Identity, `GOOGLE_APPLICATION_CREDENTIALS`, gcloud) | +| Azure Blob Storage | `az://container/path/` or `abfs://container@account.dfs.core.windows.net/path/` | `DefaultAzureCredential` (env → Managed Identity on AKS → Azure CLI) | + +For cloud backends the object store is rooted at the bucket or container, and +the remainder of `warehouse_path` becomes an in-bucket prefix. A warehouse of +`s3://bucket/warehouse` writes data files to `s3://bucket/warehouse/data/...`, +matching the location the catalog records. ### [[iceberg.partition_spec]] @@ -448,20 +461,44 @@ The v1 protocol exposes `Health`, `ListTables`, `GetTableSchema`, `ScanTableBegi --- -## Environment Variable Substitution +## Injecting Secrets + +Values in the TOML file are **not** shell-interpolated: writing +`sasl_password = "${KAFKA_PASSWORD}"` authenticates with the literal string +`${KAFKA_PASSWORD}`. Use one of the two supported mechanisms instead. -Configuration values can reference environment variables: +**1. Read the value from a file** — for Kubernetes projected `Secret` volumes +and the Secrets Store CSI Driver. Credential fields accept a `{ file = "..." }` +table in place of a string; the file contents are read at startup and trimmed. ```toml [kafka.security] -sasl_username = "${KAFKA_USERNAME}" -sasl_password = "${KAFKA_PASSWORD}" +sasl_username = { file = "/etc/secrets/k2i/kafka-username" } +sasl_password = { file = "/etc/secrets/k2i/kafka-password" } [iceberg] -aws_access_key_id = "${AWS_ACCESS_KEY_ID}" -aws_secret_access_key = "${AWS_SECRET_ACCESS_KEY}" +aws_access_key_id = { file = "/etc/secrets/k2i/aws-access-key-id" } +aws_secret_access_key = { file = "/etc/secrets/k2i/aws-secret-access-key" } ``` +**2. Set a `K2I_*` environment variable** — these take precedence over the TOML +value, and suit `secretKeyRef` injection. + +```bash +export K2I_KAFKA_SECURITY_SASL_USERNAME=svc-account +export K2I_KAFKA_SECURITY_SASL_PASSWORD=... +export K2I_ICEBERG_AWS_ACCESS_KEY_ID=AKIA... +export K2I_ICEBERG_AWS_SECRET_ACCESS_KEY=... +``` + +Credential fields are redacted in both `Debug` output and serde serialization, +so neither a log line nor a configuration dump can leak them. Prefer file refs +over env vars where the threat model includes other processes reading +`/proc//environ`. + +See [Kubernetes deployment](./kubernetes.md) for the full variable table and the +manifest patterns for both approaches. + --- ## CLI Overrides diff --git a/docs/kubernetes.md b/docs/kubernetes.md new file mode 100644 index 0000000..d6632a4 --- /dev/null +++ b/docs/kubernetes.md @@ -0,0 +1,242 @@ +# Deploying K2I on Kubernetes + +K2I reads configuration from a TOML file and layers two secret-injection +mechanisms on top, so no secret ever needs to live in a ConfigMap: + +1. **Secret file refs** — `{ file = "path" }` inline tables in TOML, designed + for Kubernetes projected volumes and the Secrets Store CSI Driver. +2. **`K2I_*` environment variables** — override any field at runtime, designed + for `env` / `envFrom` injection from `Secret` resources. + +Precedence (highest wins): + +1. `K2I_*` environment variables +2. Inline TOML values (including `{ file = ... }` refs) + +Invalid numeric/enum env values are rejected with a warning (the TOML or +default value is preserved), and unrecognized `K2I_*` variables are logged so +typos do not fail silently. + +## Pattern A: projected secret files + +Mount a `Secret` as files and point secret fields at the mount paths. +Secret fields (`kafka.security.sasl_username`, `kafka.security.sasl_password`, +`iceberg.aws_access_key_id`, `iceberg.aws_secret_access_key`, +`iceberg.rest.credential`, `iceberg.rest.oauth2_client_id`, +`iceberg.rest.oauth2_client_secret`, `iceberg.azure_access_key`) accept either a plain string or a +`{ file = "path" }` table. File contents are trimmed; a missing file fails +startup with a clear error. + +```yaml +apiVersion: v1 +kind: Secret +metadata: + name: k2i-secrets +stringData: + kafka-password: hunter2 + aws-access-key-id: AKIA... + aws-secret-access-key: ... +--- +apiVersion: apps/v1 +kind: Deployment +metadata: + name: k2i +spec: + replicas: 1 # K2I is single-process by design + selector: + matchLabels: + app: k2i + template: + metadata: + labels: + app: k2i + spec: + containers: + - name: k2i + image: ghcr.io/osodevops/k2i:latest + args: ["ingest", "--config", "/etc/k2i/config.toml"] + volumeMounts: + - name: config + mountPath: /etc/k2i + - name: secrets + mountPath: /etc/secrets/k2i + readOnly: true + volumes: + - name: config + configMap: + name: k2i-config # non-sensitive TOML only + - name: secrets + secret: + secretName: k2i-secrets +``` + +```toml +# /etc/k2i/config.toml (ConfigMap — no secrets here) +[kafka] +bootstrap_servers = ["kafka:9092"] +topic = "events" +consumer_group = "k2i-ingestion" + +[kafka.security] +protocol = "SASL_SSL" +sasl_mechanism = "SCRAM-SHA-256" +sasl_password = { file = "/etc/secrets/k2i/kafka-password" } + +[iceberg] +catalog_type = "rest" +rest_uri = "http://iceberg-rest:8181" +warehouse_path = "s3://lakehouse/warehouse" +database_name = "raw" +table_name = "events" +aws_access_key_id = { file = "/etc/secrets/k2i/aws-access-key-id" } +aws_secret_access_key = { file = "/etc/secrets/k2i/aws-secret-access-key" } +``` + +## Pattern B: environment variables + +The fields listed in [Supported variables](#supported-variables) can be +overridden with a `K2I_` prefixed variable: `K2I_` + uppercase field path with +`_` separators. Secrets come from `secretKeyRef`; plain values from `env` or +`configMapKeyRef`. + +Anything outside that table has no env override — set it in the TOML file. A +`K2I_*` variable that is not recognized is logged at startup, so a typo such as +`K2I_KAFKA_TOPC` is reported rather than silently ignored. + +```yaml +spec: + containers: + - name: k2i + env: + - name: K2I_KAFKA_BOOTSTRAP_SERVERS + value: "kafka:9092" + - name: K2I_KAFKA_TOPIC + value: "events" + - name: K2I_KAFKA_SECURITY_SASL_PASSWORD + valueFrom: + secretKeyRef: + name: k2i-secrets + key: kafka-password + - name: K2I_ICEBERG_AWS_ACCESS_KEY_ID + valueFrom: + secretKeyRef: + name: k2i-secrets + key: aws-access-key-id +``` + +Note: `Vec` fields such as `K2I_KAFKA_BOOTSTRAP_SERVERS` are +comma-separated. + +### Supported variables + +| Variable | Field | +|---|---| +| `K2I_KAFKA_BOOTSTRAP_SERVERS` | `kafka.bootstrap_servers` (comma-separated) | +| `K2I_KAFKA_TOPIC` | `kafka.topic` | +| `K2I_KAFKA_CONSUMER_GROUP` | `kafka.consumer_group` | +| `K2I_KAFKA_BATCH_SIZE` | `kafka.batch_size` | +| `K2I_KAFKA_BATCH_TIMEOUT_MS` | `kafka.batch_timeout_ms` | +| `K2I_KAFKA_SESSION_TIMEOUT_MS` | `kafka.session_timeout_ms` | +| `K2I_KAFKA_HEARTBEAT_INTERVAL_MS` | `kafka.heartbeat_interval_ms` | +| `K2I_KAFKA_MAX_POLL_INTERVAL_MS` | `kafka.max_poll_interval_ms` | +| `K2I_KAFKA_AUTO_OFFSET_RESET` | `kafka.auto_offset_reset` (`earliest`, `latest`) | +| `K2I_KAFKA_SECURITY_PROTOCOL` | `kafka.security.protocol` | +| `K2I_KAFKA_SECURITY_SASL_MECHANISM` | `kafka.security.sasl_mechanism` | +| `K2I_KAFKA_SECURITY_SASL_USERNAME` | `kafka.security.sasl_username` | +| `K2I_KAFKA_SECURITY_SASL_PASSWORD` | `kafka.security.sasl_password` | +| `K2I_ICEBERG_CATALOG_TYPE` | `iceberg.catalog_type` (`rest`, `glue`, `hive`, `nessie`, `sql`) | +| `K2I_ICEBERG_WAREHOUSE_PATH` | `iceberg.warehouse_path` | +| `K2I_ICEBERG_DATABASE_NAME` | `iceberg.database_name` | +| `K2I_ICEBERG_TABLE_NAME` | `iceberg.table_name` | +| `K2I_ICEBERG_AWS_REGION` | `iceberg.aws_region` | +| `K2I_ICEBERG_AWS_ACCESS_KEY_ID` | `iceberg.aws_access_key_id` | +| `K2I_ICEBERG_AWS_SECRET_ACCESS_KEY` | `iceberg.aws_secret_access_key` | +| `K2I_ICEBERG_GCS_BUCKET_NAME` | `iceberg.gcs_bucket_name` | +| `K2I_ICEBERG_GCS_SERVICE_ACCOUNT_PATH` | `iceberg.gcs_service_account_path` | +| `K2I_ICEBERG_AZURE_STORAGE_ACCOUNT_NAME` | `iceberg.azure_storage_account_name` (required for Azure) | +| `K2I_ICEBERG_AZURE_CONTAINER_NAME` | `iceberg.azure_container_name` | +| `K2I_ICEBERG_AZURE_ACCESS_KEY` | `iceberg.azure_access_key` | +| `K2I_ICEBERG_S3_ENDPOINT` | `iceberg.s3_endpoint` | +| `K2I_ICEBERG_REST_URI` | `iceberg.rest_uri` | +| `K2I_ICEBERG_HIVE_METASTORE_URI` | `iceberg.hive_metastore_uri` | +| `K2I_ICEBERG_REST_CREDENTIAL` | `iceberg.rest.credential` | +| `K2I_ICEBERG_REST_OAUTH2_CLIENT_ID` | `iceberg.rest.oauth2_client_id` | +| `K2I_ICEBERG_REST_OAUTH2_CLIENT_SECRET` | `iceberg.rest.oauth2_client_secret` | +| `K2I_SCHEMA_EVOLUTION_MODE` | `schema_evolution.mode` (`manual`, `auto-additive`, `permissive`) | +| `K2I_SCHEMA_EVOLUTION_ON_BREAKING_CHANGE` | `schema_evolution.on_breaking_change` (`pause`, `fail`, `skip-message`) | +| `K2I_BUFFER_TTL_SECONDS` | `buffer.ttl_seconds` | +| `K2I_BUFFER_MAX_SIZE_MB` | `buffer.max_size_mb` | +| `K2I_BUFFER_FLUSH_INTERVAL_SECONDS` | `buffer.flush_interval_seconds` | +| `K2I_TRANSACTION_LOG_LOG_DIR` | `transaction_log.log_dir` | +| `K2I_MONITORING_HEALTH_PORT` | `monitoring.health_port` | +| `K2I_MONITORING_METRICS_PORT` | `monitoring.metrics_port` | +| `K2I_MONITORING_LOG_LEVEL` | `monitoring.log_level` (`trace`, `debug`, `info`, `warn`, `error`) | +| `K2I_MONITORING_LOG_FORMAT` | `monitoring.log_format` (`json`, `text`) | +| `K2I_RPC_ENABLED` | `rpc.enabled` (`true`/`1`) | +| `K2I_RPC_SOCKET_PATH` | `rpc.socket_path` | + +## Pattern C: Secrets Store CSI Driver + +The CSI driver mounts external secrets (AWS Secrets Manager, Azure Key Vault, +GCP Secret Manager, Vault) as files. Use the same `{ file = ... }` TOML refs +as Pattern A. + +```yaml +apiVersion: secrets-store.csi.x-k8s.io/v1 +kind: SecretProviderClass +metadata: + name: k2i-secrets +spec: + provider: aws + parameters: + objects: | + - objectName: "k2i/kafka-password" + objectType: "secretsmanager" + - objectName: "k2i/aws-secret-access-key" + objectType: "secretsmanager" +--- +# Pod spec: + volumes: + - name: secrets + csi: + driver: secrets-store.csi.k8s.io + readOnly: true + volumeAttributes: + secretProviderClass: k2i-secrets + containers: + - name: k2i + volumeMounts: + - name: secrets + mountPath: /mnt/secrets + readOnly: true +``` + +```toml +[kafka.security] +sasl_password = { file = "/mnt/secrets/kafka-password" } + +[iceberg] +aws_secret_access_key = { file = "/mnt/secrets/aws-secret-access-key" } +``` + +## Notes + +- **Secret redaction**: secret fields are wrapped in a `Secret` type that + redacts in **both** `Debug` and `serde` serialization, emitting `REDACTED` in + place of the value. Neither a `{:?}` dump nor a JSON/TOML serialization of the + configuration can leak a credential. The value is reachable only through the + explicit `expose()` accessor. The trade-off is deliberate: a serialized + configuration does not round-trip, because reading it back yields the literal + `REDACTED` marker rather than the original credential. +- **Fail-fast validation**: warehouse settings that cannot be derived from the + path — notably `azure_storage_account_name` — are checked at startup rather + than on the first flush, so a misconfigured deployment fails immediately + instead of after the process has reported healthy. +- **Env var visibility**: values injected via `env` are visible in + `/proc//environ` to other processes with sufficient privileges and in + `kubectl describe pod` output is limited to the reference (not the value), + but the pod spec still records the mapping. Prefer file refs for the most + sensitive credentials. +- **Rotation**: file contents are read once at startup. Rotating a secret + requires a pod restart to pick up new values. +- **Single replica**: K2I is single-process by design; run `replicas: 1`.