From 919957ab39c188ec9d30c67abfe5ef74c0d5a814 Mon Sep 17 00:00:00 2001 From: Manudev <38869988+manudiv16@users.noreply.github.com> Date: Sat, 25 Jul 2026 19:46:19 +0200 Subject: [PATCH 1/3] feat: cloud object stores, credential chains, and secret injection GCS/Azure object stores + credential chain docs (PR #8): - GCS backend with ADC / Workload Identity Federation - Azure backend (az:// and abfs://) with DefaultAzureCredential chain - S3 default credential chain docs (env -> IMDS -> IRSA) Config secret injection (PR #9): - Secret newtype for credential fields: plain string or { file = "path" } TOML forms, Debug redacted, expose() accessors - K2I_* env var overrides with warnings on invalid numeric/enum values and unrecognized variables (typo detection) - Mutex-serialized env-var tests - docs/kubernetes.md: projected volumes, env injection, Secrets Store CSI Driver, full variable table - azure_access_key uses Secret for consistency Closes #1 --- CHANGELOG.md | 12 +- Cargo.lock | 10 +- Cargo.toml | 2 +- config/example.toml | 99 ++- crates/k2i-cli/src/commands/dev.rs | 5 + crates/k2i-cli/src/main.rs | 5 +- crates/k2i-core/src/backfill.rs | 5 + crates/k2i-core/src/config.rs | 674 +++++++++++++++++- crates/k2i-core/src/iceberg/catalog.rs | 5 + crates/k2i-core/src/iceberg/glue.rs | 9 +- crates/k2i-core/src/iceberg/hive.rs | 5 + crates/k2i-core/src/iceberg/nessie.rs | 11 +- crates/k2i-core/src/iceberg/official.rs | 16 +- crates/k2i-core/src/iceberg/sql.rs | 5 + crates/k2i-core/src/iceberg/writer.rs | 315 +++++++- crates/k2i-core/src/kafka/consumer.rs | 8 +- crates/k2i-core/src/read/mod.rs | 5 + .../k2i-rpc-server/examples/fixture_server.rs | 5 + crates/k2i-rpc-server/src/lib.rs | 5 + docs/kubernetes.md | 225 ++++++ 20 files changed, 1350 insertions(+), 76 deletions(-) create mode 100644 docs/kubernetes.md diff --git a/CHANGELOG.md b/CHANGELOG.md index 3f420a7..3afd4ff 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,16 +7,26 @@ 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. + ### Changed +- 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: `IcebergWriter` now derives an in-bucket prefix from `warehouse_path` (e.g. `warehouse` for `s3://bucket/warehouse`) and prepends it to every data file path, fixing a silent mismatch where uploads landed at `s3://bucket/data/...` while the catalog expected `s3://bucket/warehouse/data/...`. Preexisting on S3; now applied uniformly to GCS and Azure. +- 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. - 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. ### 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..39e1605 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"] diff --git a/config/example.toml b/config/example.toml index cb57234..b0bc779 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,18 +82,65 @@ 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. -# Partition specification -# [[iceberg.partition_spec]] -# source_field = "event_timestamp" -# transform = "day" +# 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 = "${AZURE_STORAGE_ACCESS_KEY}" # optional +# +# 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 [buffer] ttl_seconds = 60 max_size_mb = 500 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..f36ff18 100644 --- a/crates/k2i-cli/src/main.rs +++ b/crates/k2i-cli/src/main.rs @@ -513,8 +513,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..dd84d50 100644 --- a/crates/k2i-core/src/config.rs +++ b/crates/k2i-core/src/config.rs @@ -5,6 +5,89 @@ 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`). +#[derive(Clone, PartialEq, Eq, Serialize)] +#[serde(transparent)] +pub struct Secret(String); + +impl Secret { + /// 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 { + f.write_str("Secret(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 +288,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 +340,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 +373,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 +512,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)] @@ -974,11 +1077,62 @@ fn default_auto_create() -> bool { true } +/// 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_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) } @@ -1079,6 +1233,247 @@ 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_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 v.to_lowercase().as_str() { + "json" => self.monitoring.log_format = LogFormat::Json, + "text" => self.monitoring.log_format = LogFormat::Text, + _ => warn_bad_enum("K2I_MONITORING_LOG_FORMAT", &v, &["json", "text"]), + } + } + + // --- RPC --- + if let Some(v) = env_val("K2I_RPC_ENABLED") { + self.rpc.enabled = v.eq_ignore_ascii_case("true") || v == "1"; + } + 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()) { + tracing::warn!(var = %key, "Unrecognized K2I_* environment variable ignored"); + } + } + } } #[cfg(test)] @@ -1124,6 +1519,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 +1573,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 +1751,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 +1806,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 +1861,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 +1891,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 +1963,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 +1973,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 +1990,220 @@ 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"); + } + + /// 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`). + warehouse_prefix: Option, txlog: Option>, write_count: AtomicU64, /// Optional catalog operations for real catalog integration @@ -186,10 +190,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 +240,15 @@ impl IcebergWriter { } /// Create object store based on configuration. - fn create_object_store(config: &IcebergConfig) -> Result> { + /// + /// Returns the store and the in-bucket warehouse prefix to prepend to every + /// data/metadata path the writer uploads. 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,7 +263,38 @@ impl IcebergWriter { } } - fn create_s3_store(config: &IcebergConfig) -> Result> { + /// Extract the in-bucket prefix from a cloud warehouse path. + /// + /// Given `gs://bucket/some/prefix` returns `Some("some/prefix")`. Returns + /// `None` when the path has no subpath past the bucket (e.g. `s3://bucket`). + fn warehouse_prefix_after_bucket(scheme: &str, warehouse_path: &str) -> Option { + let after_scheme = warehouse_path.strip_prefix(scheme)?; + let after_bucket = after_scheme.split_once('/')?.1; + let trimmed = after_bucket.trim_matches('/'); + if trimmed.is_empty() { + None + } else { + Some(trimmed.to_string()) + } + } + + /// Extract the in-bucket prefix from an Azure warehouse path that may use + /// the Hadoop ABFS form `abfs://container@account.../prefix` or the simpler + /// `az://container/prefix`. + fn warehouse_prefix_after_azure_container(warehouse_path: &str) -> Option { + let after_scheme = warehouse_path + .strip_prefix("az://") + .or_else(|| warehouse_path.strip_prefix("abfs://"))?; + let after_container = after_scheme.split_once('/')?.1; + let trimmed = after_container.trim_matches('/'); + if trimmed.is_empty() { + None + } else { + Some(trimmed.to_string()) + } + } + + fn create_s3_store(config: &IcebergConfig) -> Result<(Arc, Option)> { use object_store::aws::AmazonS3Builder; let bucket = config @@ -261,21 +305,28 @@ impl IcebergWriter { Error::Iceberg(IcebergError::CatalogConnection("Invalid S3 path".into())) })?; + let warehouse_prefix = Self::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 +336,108 @@ 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(|| { + config + .warehouse_path + .strip_prefix("gs://") + .and_then(|s| s.split('/').next()) + .map(|s| s.to_string()) + }) + .ok_or_else(|| { + Error::Iceberg(IcebergError::CatalogConnection( + "Invalid GCS path: could not determine bucket".into(), + )) + })?; - 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(), - ))) + // 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 = Self::warehouse_prefix_after_bucket("gs://", &config.warehouse_path); + + 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(|| Self::parse_azure_container(&config.warehouse_path)) + .ok_or_else(|| { + Error::Iceberg(IcebergError::CatalogConnection( + "Invalid Azure path: could not determine container".into(), + )) + })?; - 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 warehouse_prefix = Self::warehouse_prefix_after_azure_container(&config.warehouse_path); + + 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> { + /// Parse the Azure container name 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`. + fn parse_azure_container(warehouse_path: &str) -> Option { + let after_scheme = warehouse_path + .strip_prefix("az://") + .or_else(|| warehouse_path.strip_prefix("abfs://"))?; + let first_segment = after_scheme.split('/').next()?; + // ABFS form: `container@account.dfs.core.windows.net` — take the + // segment before `@` as the container. Simple form has no `@`. + let container = first_segment.split('@').next()?; + if container.is_empty() { + None + } else { + Some(container.to_string()) + } + } + + 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 +459,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. @@ -760,8 +897,10 @@ impl IcebergWriter { partition_info.min_offset, partition_info.max_offset ); - // Format: data/{db}/{table}/{time_partition}/kafka_partition={N}/part-{uuid}-{offset_range}.parquet - format!( + // Format: [warehouse_prefix]data/{db}/{table}/{time_partition}/kafka_partition={N}/part-{uuid}-{offset_range}.parquet + // The warehouse_prefix (in-bucket subpath for cloud backends) keeps + // uploads aligned with the warehouse path recorded by the catalog/txlog. + let relative = format!( "data/{}/{}/{}/kafka_partition={}/part-{}-{}.parquet", self.config.database_name, self.config.table_name, @@ -769,7 +908,12 @@ impl IcebergWriter { partition_info.kafka_partition, uuid, offset_range - ) + ); + + match &self.warehouse_prefix { + Some(prefix) => format!("{prefix}/{relative}"), + None => relative, + } } /// Extract partition information from a RecordBatch. @@ -891,6 +1035,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 +1298,120 @@ mod tests { assert!(stats.snapshot_id > 0); assert!(!writer.has_catalog_integration()); } + #[test] + fn test_warehouse_prefix_after_bucket_strips_scheme_and_bucket() { + // Bucket-only path: no prefix + assert_eq!( + IcebergWriter::warehouse_prefix_after_bucket("s3://my-bucket", "s3://my-bucket"), + None + ); + // Trailing slash normalizes to None + assert_eq!( + IcebergWriter::warehouse_prefix_after_bucket("s3://my-bucket/", "s3://my-bucket/"), + None + ); + // Single-segment prefix + assert_eq!( + IcebergWriter::warehouse_prefix_after_bucket("s3://", "s3://my-bucket/warehouse"), + Some("warehouse".to_string()) + ); + // Multi-segment prefix preserved + assert_eq!( + IcebergWriter::warehouse_prefix_after_bucket("gs://", "gs://bucket/warehouse/prod"), + Some("warehouse/prod".to_string()) + ); + // Wrong scheme returns None + assert_eq!( + IcebergWriter::warehouse_prefix_after_bucket("s3://", "gs://bucket/warehouse"), + None + ); + } + + #[test] + fn test_parse_azure_container_both_url_forms() { + // Simple form: az://container/path + assert_eq!( + IcebergWriter::parse_azure_container("az://my-container/warehouse"), + Some("my-container".to_string()) + ); + // Container-only (no subpath) + assert_eq!( + IcebergWriter::parse_azure_container("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!( + IcebergWriter::parse_azure_container( + "abfs://container@account.dfs.core.windows.net/path" + ), + Some("container".to_string()) + ); + // ABFS with multi-segment path + assert_eq!( + IcebergWriter::parse_azure_container("abfs://events@prodacct/warehouse/raw"), + Some("events".to_string()) + ); + // Not an azure scheme + assert_eq!( + IcebergWriter::parse_azure_container("s3://bucket/path"), + None + ); + // Empty container after `@` + assert_eq!( + IcebergWriter::parse_azure_container("az://@account/path"), + None + ); + } + + #[test] + fn test_warehouse_prefix_after_azure_container_only_subpath() { + // No subpath past container + assert_eq!( + IcebergWriter::warehouse_prefix_after_azure_container("az://container"), + None + ); + // Simple form subpath + assert_eq!( + IcebergWriter::warehouse_prefix_after_azure_container("az://container/warehouse"), + Some("warehouse".to_string()) + ); + // ABFS form: prefix is the part AFTER the first `/`, not affected by `@account` + assert_eq!( + IcebergWriter::warehouse_prefix_after_azure_container( + "abfs://container@account.dfs.core.windows.net/warehouse/raw" + ), + Some("warehouse/raw".to_string()) + ); + // Wrong scheme + assert_eq!( + IcebergWriter::warehouse_prefix_after_azure_container("gs://bucket/warehouse"), + None + ); + } + + #[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 partition_info = 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, + }; + let path = writer.generate_file_path(&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); + } } 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-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/kubernetes.md b/docs/kubernetes.md new file mode 100644 index 0000000..f0d1df3 --- /dev/null +++ b/docs/kubernetes.md @@ -0,0 +1,225 @@ +# 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 + +Every field can be overridden with a `K2I_` prefixed variable: +`K2I_` + uppercase field path with `_` separators. Secrets come from +`secretKeyRef`; plain values from `env` or `configMapKeyRef`. + +```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_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 whose + `Debug` output is `Secret(REDACTED)`, so `{:?}` dumps of the configuration + do not leak values. The value is only exposed through explicit accessors. +- **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`. From de8cdf72c34dfad18bb24a78e7681cc6e49e1f91 Mon Sep 17 00:00:00 2001 From: Sion Smith Date: Tue, 28 Jul 2026 14:12:01 +0100 Subject: [PATCH 2/3] fix: correct cloud warehouse prefix handling and env override gaps MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The in-bucket warehouse prefix was applied inside generate_file_path, but that path is also what goes to the catalog, the transaction log, and the read path — all of which resolve it by joining against warehouse_path. For a warehouse of s3://bucket/warehouse the catalog recorded s3://bucket/warehouse/warehouse/data/... while the upload landed at s3://bucket/warehouse/data/..., so every committed file was unreadable. Apply the prefix in a dedicated storage_path() used only by upload_file, keeping every externally-visible path warehouse-relative. Also fixes three defects in the env-override layer: - K2I_MONITORING_LOG_FORMAT never took effect: the tracing subscriber is configured before the config is loaded and parsed the TOML directly. - K2I_RPC_ENABLED read any unrecognized value as false, so =yes silently disabled an RPC server the TOML had enabled. - The cloud object-store fields had no overrides at all, including the Azure-required azure_storage_account_name. And suppresses the spurious unrecognized-variable warnings for K2I_E2E_* and the other harness variables that share the engine's environment. Docs: docs/configuration.md promised ${VAR} shell substitution, which does not exist — following it would have authenticated with the literal string. Replaced with the two real mechanisms. README.md, docs/architecture.md and docs/configuration.md still described GCS/Azure as unwired. Verified locally: 319 tests (up from 289), clippy -D warnings, fmt, cargo-audit, cargo-semver-checks, the Docker Kafka integration tests, and all five docker/e2e suites. Co-Authored-By: Claude Opus 5 (1M context) --- CHANGELOG.md | 11 +- README.md | 2 +- config/example.toml | 10 +- crates/k2i-cli/src/main.rs | 20 ++- crates/k2i-core/src/config.rs | 249 +++++++++++++++++++++++++- crates/k2i-core/src/iceberg/writer.rs | 193 +++++++++++++++++--- docs/architecture.md | 2 +- docs/configuration.md | 63 +++++-- docs/kubernetes.md | 15 +- 9 files changed, 507 insertions(+), 58 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3afd4ff..b4a274c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -23,11 +23,20 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed -- Aligned cloud object store uploads with the warehouse path recorded by the catalog/txlog: `IcebergWriter` now derives an in-bucket prefix from `warehouse_path` (e.g. `warehouse` for `s3://bucket/warehouse`) and prepends it to every data file path, fixing a silent mismatch where uploads landed at `s3://bucket/data/...` while the catalog expected `s3://bucket/warehouse/data/...`. Preexisting on S3; now applied uniformly to GCS and Azure. +- 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. +### 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 - Raised the documented minimum supported Rust version to 1.94, matching `iceberg` 0.10.0 and the updated AWS SDK dependency graph. diff --git a/README.md b/README.md index 2f04410..62a12c7 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, but only S3 and the local filesystem are covered by automated tests; validate GCS and Azure 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 b0bc779..9e44b3a 100644 --- a/config/example.toml +++ b/config/example.toml @@ -133,7 +133,11 @@ rest_uri = "http://localhost:8181" # 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 = "${AZURE_STORAGE_ACCESS_KEY}" # optional +# 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. @@ -141,6 +145,10 @@ rest_uri = "http://localhost:8181" # without explicit keys. # Partition specification +# [[iceberg.partition_spec]] +# source_field = "event_timestamp" +# transform = "day" + [buffer] ttl_seconds = 60 max_size_mb = 500 diff --git a/crates/k2i-cli/src/main.rs b/crates/k2i-cli/src/main.rs index f36ff18..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 diff --git a/crates/k2i-core/src/config.rs b/crates/k2i-core/src/config.rs index dd84d50..696f68e 100644 --- a/crates/k2i-core/src/config.rs +++ b/crates/k2i-core/src/config.rs @@ -18,6 +18,13 @@ use std::path::PathBuf; /// /// File contents are trimmed. Read the value explicitly with /// [`Secret::expose`] or via `Deref` (`&*secret`). +/// +/// Note that `Debug` is the only redacted output. [`Serialize`] is +/// `transparent` and emits the plaintext, so that a `Config` round-trips +/// faithfully. Nothing currently serializes `Config`; if that changes — a +/// `config dump` subcommand, an RPC response echoing settings — the secret +/// would go out in the clear. Redact at that call site, or give `Secret` a +/// redacting `Serialize` and accept that the output no longer round-trips. #[derive(Clone, PartialEq, Eq, Serialize)] #[serde(transparent)] pub struct Secret(String); @@ -957,6 +964,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 @@ -1077,6 +1105,13 @@ fn default_auto_create() -> bool { true } +/// `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. @@ -1101,6 +1136,10 @@ const KNOWN_ENV_VARS: &[&str] = &[ "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", @@ -1351,6 +1390,18 @@ impl Config { 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)); } @@ -1451,16 +1502,25 @@ impl Config { } } if let Some(v) = env_val("K2I_MONITORING_LOG_FORMAT") { - match v.to_lowercase().as_str() { - "json" => self.monitoring.log_format = LogFormat::Json, - "text" => self.monitoring.log_format = LogFormat::Text, - _ => warn_bad_enum("K2I_MONITORING_LOG_FORMAT", &v, &["json", "text"]), + 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") { - self.rpc.enabled = v.eq_ignore_ascii_case("true") || v == "1"; + // 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); @@ -1469,7 +1529,12 @@ impl Config { // 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()) { + 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"); } } @@ -2155,6 +2220,178 @@ table_name = "tbl" 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"); + } + + #[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 { diff --git a/crates/k2i-core/src/iceberg/writer.rs b/crates/k2i-core/src/iceberg/writer.rs index 5832cbd..3a985e2 100644 --- a/crates/k2i-core/src/iceberg/writer.rs +++ b/crates/k2i-core/src/iceberg/writer.rs @@ -118,6 +118,11 @@ pub struct IcebergWriter { /// 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, @@ -241,11 +246,12 @@ impl IcebergWriter { /// Create object store based on configuration. /// - /// Returns the store and the in-bucket warehouse prefix to prepend to every - /// data/metadata path the writer uploads. 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. + /// 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)> { @@ -670,8 +676,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 @@ -680,7 +690,7 @@ impl IcebergWriter { .map_err(|e| { Error::Iceberg(IcebergError::FileUpload(format!( "Failed to upload file to {}: {}", - path, e + key, e ))) })?; @@ -897,10 +907,13 @@ impl IcebergWriter { partition_info.min_offset, partition_info.max_offset ); - // Format: [warehouse_prefix]data/{db}/{table}/{time_partition}/kafka_partition={N}/part-{uuid}-{offset_range}.parquet - // The warehouse_prefix (in-bucket subpath for cloud backends) keeps - // uploads aligned with the warehouse path recorded by the catalog/txlog. - let relative = format!( + // 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, self.config.table_name, @@ -908,11 +921,22 @@ impl IcebergWriter { partition_info.kafka_partition, uuid, offset_range - ); + ) + } + /// 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}"), - None => relative, + Some(prefix) => format!("{}/{}", prefix, relative_path.trim_start_matches('/')), + None => relative_path.to_string(), } } @@ -1390,6 +1414,18 @@ mod tests { ); } + 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 @@ -1398,20 +1434,129 @@ mod tests { let config = create_test_config(temp_dir.path().to_str().unwrap()); let writer = IcebergWriter::new(config).await.unwrap(); - let partition_info = 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, - }; - let path = writer.generate_file_path(&partition_info); + 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/docs/architecture.md b/docs/architecture.md index 9dca9ef..f374780 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, but only S3 and the local filesystem are covered by automated tests; validate GCS and Azure 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..35e98bd 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,42 @@ 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 `Debug` output. 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 index f0d1df3..640c75d 100644 --- a/docs/kubernetes.md +++ b/docs/kubernetes.md @@ -94,9 +94,14 @@ aws_secret_access_key = { file = "/etc/secrets/k2i/aws-secret-access-key" } ## Pattern B: environment variables -Every field can be overridden with a `K2I_` prefixed variable: -`K2I_` + uppercase field path with `_` separators. Secrets come from -`secretKeyRef`; plain values from `env` or `configMapKeyRef`. +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: @@ -146,6 +151,10 @@ comma-separated. | `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` | From 71773f62cb8c32300012cb18cf50bc5b08a90c49 Mon Sep 17 00:00:00 2001 From: Sion Smith Date: Tue, 28 Jul 2026 15:03:33 +0100 Subject: [PATCH 3/3] fix: redact secrets on serialize, validate cloud warehouses, cover S3 for real Three gaps remained after the warehouse-prefix fix. Secret redacted Debug but not Serialize, and Config derives Serialize, so any code that dumped or echoed the configuration would emit credentials in the clear. Redact in both, following the secrecy crate's convention that emitting a secret must be a conscious act. A serialized Config no longer round-trips; that trade-off is deliberate and documented, since a visibly broken credential beats a silently leaked one. Cloud warehouse settings that cannot be derived from the path are now checked in Config::validate rather than at writer construction. Azure needs a storage account name that neither URL form carries; previously a long-running ingest reported healthy and failed minutes later on its first flush. Warehouse-path parsing now lives in one place and is shared with the writer, so validation and store construction cannot drift. The prefix fix was only unit-tested. Added container-backed S3 round-trip tests (MinIO) for a prefixed warehouse, a multi-segment prefix, and a bucket-root warehouse, each asserting that warehouse_path joined with the reported path resolves to a real object and that nothing landed at the doubled-prefix or bucket-root locations. Verified they fail against the previous behaviour with the exact production symptom: a 404 on s3://bucket/warehouse/prod/warehouse/prod/data/... Those tests would not have run in CI: every container-backed test is #[ignore = "requires Docker"], and the integration job never passed --include-ignored, so it provisioned Docker and ran no Docker test at all, including the pre-existing Kafka ones. Fixed, and the workflow now also triggers on docs/ and config/ since a test asserts every K2I_* variable is documented. Verified locally: 326 tests, the CI integration command (13 tests incl. 5 container-backed), clippy -D warnings, fmt, cargo-audit, cargo-semver-checks, and all five docker/e2e suites. Co-Authored-By: Claude Opus 5 (1M context) --- .github/workflows/test.yml | 15 +- CHANGELOG.md | 12 + Cargo.toml | 1 + README.md | 2 +- crates/k2i-core/src/config.rs | 349 ++++++++++++++++++++- crates/k2i-core/src/iceberg/writer.rs | 172 +--------- crates/k2i-core/tests/integration_tests.rs | 250 +++++++++++++++ docs/architecture.md | 2 +- docs/configuration.md | 6 +- docs/kubernetes.md | 14 +- 10 files changed, 645 insertions(+), 178 deletions(-) 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 b4a274c..1f4d4d1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,8 +14,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - 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. @@ -32,6 +38,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - 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. diff --git a/Cargo.toml b/Cargo.toml index 39e1605..3fa4263 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -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 62a12c7..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. -- S3, GCS, and Azure object stores are wired end to end, but only S3 and the local filesystem are covered by automated tests; validate GCS and Azure credentials in your own environment before rollout. +- 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/crates/k2i-core/src/config.rs b/crates/k2i-core/src/config.rs index 696f68e..93bdc2e 100644 --- a/crates/k2i-core/src/config.rs +++ b/crates/k2i-core/src/config.rs @@ -19,17 +19,27 @@ use std::path::PathBuf; /// File contents are trimmed. Read the value explicitly with /// [`Secret::expose`] or via `Deref` (`&*secret`). /// -/// Note that `Debug` is the only redacted output. [`Serialize`] is -/// `transparent` and emits the plaintext, so that a `Config` round-trips -/// faithfully. Nothing currently serializes `Config`; if that changes — a -/// `config dump` subcommand, an RPC response echoing settings — the secret -/// would go out in the clear. Redact at that call site, or give `Secret` a -/// redacting `Serialize` and accept that the output no longer round-trips. -#[derive(Clone, PartialEq, Eq, Serialize)] -#[serde(transparent)] +/// # 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()) @@ -43,7 +53,18 @@ impl Secret { impl std::fmt::Debug for Secret { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - f.write_str("Secret(REDACTED)") + 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) } } @@ -1105,6 +1126,68 @@ 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 @@ -1228,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()) @@ -2301,6 +2418,220 @@ table_name = "tbl" 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)); diff --git a/crates/k2i-core/src/iceberg/writer.rs b/crates/k2i-core/src/iceberg/writer.rs index 3a985e2..808494d 100644 --- a/crates/k2i-core/src/iceberg/writer.rs +++ b/crates/k2i-core/src/iceberg/writer.rs @@ -17,7 +17,10 @@ //! - Atomic commits with CAS semantics via TransactionCoordinator //! - Metadata caching via MetadataCache -use crate::config::{IcebergConfig, ParquetCompression}; +use crate::config::{ + azure_container_from_warehouse, azure_prefix_after_container, bucket_from_warehouse, + warehouse_prefix_after_bucket, IcebergConfig, ParquetCompression, +}; use crate::iceberg::factory::{ CatalogOperations, DataFileInfo, SnapshotCommit, SnapshotCommitResult, }; @@ -269,49 +272,14 @@ impl IcebergWriter { } } - /// Extract the in-bucket prefix from a cloud warehouse path. - /// - /// Given `gs://bucket/some/prefix` returns `Some("some/prefix")`. Returns - /// `None` when the path has no subpath past the bucket (e.g. `s3://bucket`). - fn warehouse_prefix_after_bucket(scheme: &str, warehouse_path: &str) -> Option { - let after_scheme = warehouse_path.strip_prefix(scheme)?; - let after_bucket = after_scheme.split_once('/')?.1; - let trimmed = after_bucket.trim_matches('/'); - if trimmed.is_empty() { - None - } else { - Some(trimmed.to_string()) - } - } - - /// Extract the in-bucket prefix from an Azure warehouse path that may use - /// the Hadoop ABFS form `abfs://container@account.../prefix` or the simpler - /// `az://container/prefix`. - fn warehouse_prefix_after_azure_container(warehouse_path: &str) -> Option { - let after_scheme = warehouse_path - .strip_prefix("az://") - .or_else(|| warehouse_path.strip_prefix("abfs://"))?; - let after_container = after_scheme.split_once('/')?.1; - let trimmed = after_container.trim_matches('/'); - if trimmed.is_empty() { - None - } else { - Some(trimmed.to_string()) - } - } - 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 = Self::warehouse_prefix_after_bucket("s3://", &config.warehouse_path); + let warehouse_prefix = warehouse_prefix_after_bucket("s3://", &config.warehouse_path); let mut builder = AmazonS3Builder::new().with_bucket_name(bucket); @@ -350,13 +318,7 @@ impl IcebergWriter { let bucket = config .gcs_bucket_name .clone() - .or_else(|| { - config - .warehouse_path - .strip_prefix("gs://") - .and_then(|s| s.split('/').next()) - .map(|s| s.to_string()) - }) + .or_else(|| bucket_from_warehouse("gs://", &config.warehouse_path)) .ok_or_else(|| { Error::Iceberg(IcebergError::CatalogConnection( "Invalid GCS path: could not determine bucket".into(), @@ -366,7 +328,7 @@ impl IcebergWriter { // 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 = Self::warehouse_prefix_after_bucket("gs://", &config.warehouse_path); + let warehouse_prefix = warehouse_prefix_after_bucket("gs://", &config.warehouse_path); let mut builder = GoogleCloudStorageBuilder::new().with_bucket_name(&bucket); @@ -397,14 +359,14 @@ impl IcebergWriter { let container = config .azure_container_name .clone() - .or_else(|| Self::parse_azure_container(&config.warehouse_path)) + .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 = Self::warehouse_prefix_after_azure_container(&config.warehouse_path); + let warehouse_prefix = azure_prefix_after_container(&config.warehouse_path); let mut builder = MicrosoftAzureBuilder::new() .with_account(account_name) @@ -423,24 +385,6 @@ impl IcebergWriter { Ok((Arc::new(store), warehouse_prefix)) } - /// Parse the Azure container name 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`. - fn parse_azure_container(warehouse_path: &str) -> Option { - let after_scheme = warehouse_path - .strip_prefix("az://") - .or_else(|| warehouse_path.strip_prefix("abfs://"))?; - let first_segment = after_scheme.split('/').next()?; - // ABFS form: `container@account.dfs.core.windows.net` — take the - // segment before `@` as the container. Simple form has no `@`. - let container = first_segment.split('@').next()?; - if container.is_empty() { - None - } else { - Some(container.to_string()) - } - } - fn create_local_store( config: &IcebergConfig, ) -> Result<(Arc, Option)> { @@ -1322,98 +1266,6 @@ mod tests { assert!(stats.snapshot_id > 0); assert!(!writer.has_catalog_integration()); } - #[test] - fn test_warehouse_prefix_after_bucket_strips_scheme_and_bucket() { - // Bucket-only path: no prefix - assert_eq!( - IcebergWriter::warehouse_prefix_after_bucket("s3://my-bucket", "s3://my-bucket"), - None - ); - // Trailing slash normalizes to None - assert_eq!( - IcebergWriter::warehouse_prefix_after_bucket("s3://my-bucket/", "s3://my-bucket/"), - None - ); - // Single-segment prefix - assert_eq!( - IcebergWriter::warehouse_prefix_after_bucket("s3://", "s3://my-bucket/warehouse"), - Some("warehouse".to_string()) - ); - // Multi-segment prefix preserved - assert_eq!( - IcebergWriter::warehouse_prefix_after_bucket("gs://", "gs://bucket/warehouse/prod"), - Some("warehouse/prod".to_string()) - ); - // Wrong scheme returns None - assert_eq!( - IcebergWriter::warehouse_prefix_after_bucket("s3://", "gs://bucket/warehouse"), - None - ); - } - - #[test] - fn test_parse_azure_container_both_url_forms() { - // Simple form: az://container/path - assert_eq!( - IcebergWriter::parse_azure_container("az://my-container/warehouse"), - Some("my-container".to_string()) - ); - // Container-only (no subpath) - assert_eq!( - IcebergWriter::parse_azure_container("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!( - IcebergWriter::parse_azure_container( - "abfs://container@account.dfs.core.windows.net/path" - ), - Some("container".to_string()) - ); - // ABFS with multi-segment path - assert_eq!( - IcebergWriter::parse_azure_container("abfs://events@prodacct/warehouse/raw"), - Some("events".to_string()) - ); - // Not an azure scheme - assert_eq!( - IcebergWriter::parse_azure_container("s3://bucket/path"), - None - ); - // Empty container after `@` - assert_eq!( - IcebergWriter::parse_azure_container("az://@account/path"), - None - ); - } - - #[test] - fn test_warehouse_prefix_after_azure_container_only_subpath() { - // No subpath past container - assert_eq!( - IcebergWriter::warehouse_prefix_after_azure_container("az://container"), - None - ); - // Simple form subpath - assert_eq!( - IcebergWriter::warehouse_prefix_after_azure_container("az://container/warehouse"), - Some("warehouse".to_string()) - ); - // ABFS form: prefix is the part AFTER the first `/`, not affected by `@account` - assert_eq!( - IcebergWriter::warehouse_prefix_after_azure_container( - "abfs://container@account.dfs.core.windows.net/warehouse/raw" - ), - Some("warehouse/raw".to_string()) - ); - // Wrong scheme - assert_eq!( - IcebergWriter::warehouse_prefix_after_azure_container("gs://bucket/warehouse"), - None - ); - } - fn test_partition_info() -> PartitionInfo { PartitionInfo { topic: "test".to_string(), 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/docs/architecture.md b/docs/architecture.md index f374780..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. -- S3, GCS, and Azure object stores are wired end to end, but only S3 and the local filesystem are covered by automated tests; validate GCS and Azure credentials in your own environment before rollout. +- 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 35e98bd..c94c02e 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -491,8 +491,10 @@ export K2I_ICEBERG_AWS_ACCESS_KEY_ID=AKIA... export K2I_ICEBERG_AWS_SECRET_ACCESS_KEY=... ``` -Credential fields are redacted in `Debug` output. Prefer file refs over env vars -where the threat model includes other processes reading `/proc//environ`. +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. diff --git a/docs/kubernetes.md b/docs/kubernetes.md index 640c75d..d6632a4 100644 --- a/docs/kubernetes.md +++ b/docs/kubernetes.md @@ -221,9 +221,17 @@ aws_secret_access_key = { file = "/mnt/secrets/aws-secret-access-key" } ## Notes -- **Secret redaction**: secret fields are wrapped in a `Secret` type whose - `Debug` output is `Secret(REDACTED)`, so `{:?}` dumps of the configuration - do not leak values. The value is only exposed through explicit accessors. +- **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),