feat: cloud object stores, credential chains, and config secret injection

.github/workflows/test.yml

@@ -8,13 +8,21 @@ 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:
       - 'crates/**'
       - '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:

CHANGELOG.md

@@ -7,16 +7,47 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+
+- Implemented `create_gcs_store()` via `object_store::gcp::GoogleCloudStorageBuilder`, falling through to Application Default Credentials (Workload Identity on GKE) when `gcs_service_account_path` is unset.
+- Implemented `create_azure_store()` via `object_store::azure::MicrosoftAzureBuilder`, falling through to `DefaultAzureCredential` (Managed Identity on AKS) when `azure_access_key` is unset.
+- Documented that omitting `aws_access_key_id`/`aws_secret_access_key` activates the `AmazonS3Builder` default credential chain (env vars → IMDS → IRSA), unblocking EKS with IRSA and EC2 instance profiles without explicit config.
+- Added `gcs_bucket_name`, `gcs_service_account_path`, `azure_container_name`, `azure_storage_account_name`, and `azure_access_key` fields to `IcebergConfig` for credential overrides and the Azure-required account name.
+
+### Security
+
+- `Secret` now redacts in `serde` serialization as well as `Debug`, emitting `REDACTED` in place of the value. `Config` derives `Serialize`, so previously any code that dumped or echoed the configuration would have emitted credentials in the clear. This follows the `secrecy` crate's convention of not implementing `Serialize` for secret-wrapped strings, so that emitting one must be a conscious act. The trade-off is deliberate and documented: a serialized `Config` no longer round-trips, since reading it back yields the literal `REDACTED` marker.
+
 ### Changed
 
+- `Config::validate` now rejects cloud warehouse settings that cannot be satisfied, rather than deferring the failure to the first flush: `azure_storage_account_name` is required for `az://` and `abfs://` paths, and `s3://`/`gs://` paths must name a bucket. A long-running ingest previously reported healthy and only failed minutes later, on its first write.
+- Warehouse-path parsing (bucket, Azure container, in-bucket prefix) is now defined once in `config` and shared with the Iceberg writer, so what validation accepts is exactly what the writer can build a store from.
+- Bumped the workspace version to 0.3.0 to absorb the semver-major addition of public fields on the externally-constructible `IcebergConfig` struct. The 0.x convention treats a minor bump (0.2 → 0.3) as the breaking-change boundary.
 - Upgraded the official Apache Iceberg Rust client from 0.7 to 0.10.0 and the Arrow/Parquet ecosystem from 54 to 58.
 - Removed the temporary standalone REST `update_schema` fallback now that `Transaction::update_schema()` is available in `iceberg-rust` 0.10.0.
 - Simplified `OfficialRestCommitter` by delegating all catalog operations to the official `RestCatalog` transaction APIs.
 
 ### Fixed
 
-- Avoided manual OAuth2, route resolution, and multipart namespace encoding logic previously needed for the schema-update fallback.
+- Aligned cloud object store uploads with the warehouse path recorded by the catalog/txlog. For cloud backends the store is rooted at the bucket, so `IcebergWriter` derives an in-bucket prefix from `warehouse_path` (e.g. `warehouse` for `s3://bucket/warehouse`) and applies it when addressing the store. Uploads previously landed at `s3://bucket/data/...` while the catalog recorded `s3://bucket/warehouse/data/...`, leaving every committed file unreadable. Preexisting on S3; the same handling now covers GCS and Azure. The prefix is applied **only** at upload time — paths handed to the catalog, transaction log, and read path stay warehouse-relative, since those consumers join them against `warehouse_path` themselves.
+- Azure container parsing now handles the Hadoop ABFS form `abfs://container@account.dfs.core.windows.net/path` by extracting the container before the `@`, instead of treating the whole `container@account` segment as the container.
+- `K2I_MONITORING_LOG_FORMAT` now takes effect. The tracing subscriber is configured before the full config is loaded and read the TOML value directly, so the environment override was silently ignored for all output.
+- `K2I_RPC_ENABLED` no longer treats an unrecognized value as `false`. `K2I_RPC_ENABLED=yes` previously disabled the RPC server that the TOML had enabled; unparseable values now warn and preserve the configured value.
+- Added `K2I_*` overrides for the remaining cloud object-store fields, including the Azure-required `azure_storage_account_name`, which could not previously be set by environment-only deployments.
+- The unrecognized-variable warning no longer fires for `K2I_E2E_*` and the other harness variables that share the engine's environment during end-to-end runs.
 - Aligned Parquet writer properties with the parquet 58 API (`set_max_row_group_row_count`).
+- Avoided manual OAuth2, route resolution, and multipart namespace encoding logic previously needed for the schema-update fallback.
+
+### Testing
+
+- Added container-backed S3 round-trip tests (MinIO) covering a prefixed warehouse (`s3://bucket/warehouse`), a multi-segment prefix, and a bucket-root warehouse. Each asserts that joining `warehouse_path` with the writer's reported path resolves to a real stored object, and that nothing was written to the doubled-prefix or bucket-root locations. These reproduce the warehouse-prefix defect above; they fail against the previous behaviour.
+- CI's integration-tests job now passes `--include-ignored`. Every container-backed test is marked `#[ignore = "requires Docker"]`, so the job provisioned Docker and then ran no Docker test at all — including the pre-existing Kafka integration tests.
+- The Tests workflow now also triggers on `docs/**` and `config/**`, since a test asserts that every `K2I_*` variable is documented in `docs/kubernetes.md`.
+
+### Documentation
+
+- Removed the `docs/configuration.md` claim that config values support `${VAR}` shell substitution. No such mechanism exists — following it would have authenticated with the literal string `${VAR}`. Replaced with the two real mechanisms: `{ file = "..." }` refs and `K2I_*` overrides.
+- Updated `README.md`, `docs/architecture.md`, and `docs/configuration.md`, which still described GCS and Azure as declared-but-unwired.
 
 ### Requirements
 

Cargo.toml

@@ -9,7 +9,7 @@ members = [
 ]
 
 [workspace.package]
-version = "0.2.2"
+version = "0.3.0"
 edition = "2021"
 license = "Apache-2.0"
 authors = ["OSO DevOps"]
@@ -119,6 +119,7 @@ testcontainers = "0.23"
 testcontainers-modules = { version = "0.11", features = [
   "kafka",
   "localstack",
+  "minio",
 ] }
 tempfile = "3"
 tokio-test = "0.4"

README.md

@@ -257,7 +257,7 @@ K2I is ready for a first public release as a production-oriented Kafka-to-Iceber
 - Startup recovery computes state, but Kafka seeking/deduplication and startup orphan cleanup need further wiring.
 - Kafka offset commits are async; broker durability acknowledgement is not confirmed by the current helper.
 - Transaction-log entries are flushed, but not every entry is fsynced individually.
-- GCS and Azure object-store configuration is declared, but writer creation is not complete for those backends.
+- S3, GCS, and Azure object stores are wired end to end. S3 is covered by a container-backed round-trip test (MinIO); GCS and Azure are covered only at the configuration and store-construction level, so validate their credentials in your own environment before rollout.
 - Maintenance commands and task implementations exist; scheduler wiring should be reviewed for each deployment.
 
 See [Production Readiness](docs/production-readiness.md) for the detailed review checklist.

config/example.toml

@@ -1,5 +1,31 @@
 # K2I Example Configuration
 # Kafka to Iceberg streaming ingestion
+#
+# Configuration is loaded from this file and can be overridden by:
+#   - Secret file refs: `{ file = "path" }` inline tables (Kubernetes projected volumes)
+#   - `K2I_*` environment variables: override any field at runtime
+#
+# Override precedence (highest to lowest):
+#   1. K2I_* environment variables
+#   2. Inline TOML values (including `{ file = ... }` refs)
+#
+# Secret file refs — read secret contents from files at startup:
+#   [kafka.security]
+#   sasl_password = { file = "/etc/secrets/k2i/kafka-password" }
+#
+#   [iceberg]
+#   aws_access_key_id = { file = "/etc/secrets/k2i/aws-key" }
+#   aws_secret_access_key = { file = "/etc/secrets/k2i/aws-secret" }
+#
+# Environment variable overrides (all K2I_* prefixed):
+#   export K2I_KAFKA_TOPIC=my-topic
+#   export K2I_KAFKA_SECURITY_SASL_PASSWORD=hunter2
+#   export K2I_ICEBERG_WAREHOUSE_PATH=s3://prod-bucket/warehouse
+#   export K2I_ICEBERG_REST_CREDENTIAL=my-bearer-token
+#
+# Invalid numeric/enum env values are rejected with a warning; unrecognized
+# K2I_* variables are also warned about. See docs/kubernetes.md for a full
+# deployment guide.
 
 [kafka]
 bootstrap_servers = ["localhost:9092"]
@@ -32,6 +58,14 @@ type = "raw"
 # sasl_mechanism = "SCRAM-SHA-256"
 # sasl_username = "user"
 # sasl_password = "password"
+#
+# Or load secrets from files (Kubernetes projected volumes):
+# sasl_username = { file = "/etc/secrets/k2i/kafka-username" }
+# sasl_password = { file = "/etc/secrets/k2i/kafka-password" }
+#
+# Or set via environment variables:
+# K2I_KAFKA_SECURITY_SASL_USERNAME=user
+# K2I_KAFKA_SECURITY_SASL_PASSWORD=hunter2
 
 [schema_evolution]
 mode = "auto-additive"
@@ -48,12 +82,67 @@ compression = "snappy"
 
 # REST catalog configuration
 rest_uri = "http://localhost:8181"
+#
+# REST catalog credential (bearer token or OAuth2), under [iceberg.rest]:
+# [iceberg.rest]
+# credential = "my-bearer-token"
+# credential = { file = "/etc/secrets/k2i/rest-credential" }  # K8s projected volume
+#
+# Or via environment variable:
+# K2I_ICEBERG_REST_CREDENTIAL=my-bearer-token
 
-# AWS configuration (for S3 storage)
-# aws_region = "us-east-1"
-# aws_access_key_id = "${AWS_ACCESS_KEY_ID}"
-# aws_secret_access_key = "${AWS_SECRET_ACCESS_KEY}"
-# s3_endpoint = "http://localhost:9000"  # For MinIO
+# AWS / S3 configuration
+# Set aws_region for the S3 bucket region.
+# For explicit key-based auth, set aws_access_key_id and aws_secret_access_key:
+#   aws_region = "us-east-1"
+#   aws_access_key_id = "AKIA..."
+#   aws_secret_access_key = "..."
+#   s3_endpoint = "http://localhost:9000"  # For MinIO
+#
+# Or load from files (Kubernetes projected volumes):
+# aws_access_key_id = { file = "/etc/secrets/k2i/aws-key" }
+# aws_secret_access_key = { file = "/etc/secrets/k2i/aws-secret" }
+#
+# Or set via environment variables:
+# K2I_ICEBERG_AWS_ACCESS_KEY_ID=AKIA...
+# K2I_ICEBERG_AWS_SECRET_ACCESS_KEY=...
+#
+# When both access key fields are omitted, the SDK uses the default credential
+# chain: environment variables (AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY, etc.)
+# -> IMDS (EC2 instance profiles) -> IRSA (EKS IAM Roles for Service Accounts).
+# This allows EKS with IRSA, EC2 instance profiles, and local dev with env
+# vars — no explicit config needed.
+
+# GCS configuration — set warehouse_path to gs://bucket/path
+#   warehouse_path = "gs://my-gcs-bucket/warehouse"
+#   gcs_bucket_name = "my-gcs-bucket"  # optional: override bucket from path
+#   gcs_service_account_path = "/path/to/key.json"  # optional: explicit SA key
+#
+# When gcs_service_account_path is omitted, the SDK uses Application Default
+# Credentials (ADC). On GKE this means Workload Identity Federation; locally
+# it picks up GOOGLE_APPLICATION_CREDENTIALS or gcloud auth.
+
+# Azure configuration — set warehouse_path to either:
+#   az://container/path                          (simple form)
+#   abfs://container@account.dfs.core.windows.net/path  (Hadoop ABFS form)
+# Both forms parse the container correctly; the account is taken from
+# azure_storage_account_name (REQUIRED) since it cannot derive the endpoint host.
+#
+# Examples:
+#   warehouse_path = "az://my-container/warehouse"
+#   warehouse_path = "abfs://my-container@mystorageaccount.dfs.core.windows.net/warehouse"
+#   azure_storage_account_name = "mystorageaccount"  # REQUIRED
+#   azure_container_name = "my-container"  # optional: override container from path
+#   azure_access_key = "..."  # optional; or { file = "/etc/secrets/k2i/azure-key" }
+#
+# Note: values are not shell-interpolated. To source a credential from the
+# environment use K2I_ICEBERG_AZURE_ACCESS_KEY, and to source it from a file use
+# the `{ file = "..." }` form — not "${VAR}".
+#
+# When azure_access_key is omitted, the SDK uses the DefaultAzureCredential
+# chain: environment variables -> Managed Identity (on AKS) -> Azure CLI.
+# This enables AKS with Workload Identity or system-assigned Managed Identity
+# without explicit keys.
 
 # Partition specification
 # [[iceberg.partition_spec]]

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(),

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::<Config>(&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::<Config>(&content).ok())
+                .map(|config| config.monitoring.log_format)
+        })
         .unwrap_or(LogFormat::Json);
 
     // Initialize logging
@@ -513,8 +519,5 @@ async fn execute_command(cli: Cli) -> Result<()> {
 
 fn load_config(path: &Option<PathBuf>) -> Result<Config> {
     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)?)
 }

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(),