serve/examples/live_object_store.rs
//! Live check: **`skade-katalog serve` fronting a warehouse on a real
//! S3/MinIO**, end to end, over the wire an Iceberg-REST client uses.
//!
//! `serve/tests/data_plane_warehouse.rs` proves the same round trip against a
//! local-fs object store — deterministic, offline, and the right thing for
//! `cargo test`. It cannot prove endpoint URLs, TLS, path-style addressing,
//! SigV4 or bucket policy, because none of those exist in it. This example is
//! the other half: point it at a bucket and it writes a table with the data
//! plane, serves that warehouse, and loads it back over REST.
//!
//! It **skips cleanly** (exit 0, one line on stderr) when the endpoint or
//! bucket is unset, so it is safe to run anywhere.
//!
//! ```sh
//! export MINIO_ACCESS_KEY=… MINIO_SECRET_KEY=… # never a flag
//! export SKADE_LIVE_S3_ENDPOINT=https://minio.example:9000
//! export SKADE_LIVE_S3_BUCKET=red
//! cargo run --release --features object-store --example live_object_store
//! ```
//!
//! Knobs: `SKADE_LIVE_S3_PREFIX` (default `skade-live-check/<pid>`),
//! `SKADE_LIVE_S3_REGION` (default `us-east-1`),
//! `SKADE_LIVE_S3_VIRTUAL_HOSTED=1` (default is path-style, what MinIO needs),
//! `SKADE_LIVE_S3_KEEP=1` (leave the objects behind for inspection).
fn main() -> anyhow::Result<()> {
#[cfg(not(feature = "object-store"))]
{
eprintln!(
"live_object_store: needs the `object-store` feature \
(cargo run --features object-store --example live_object_store)"
);
Ok(())
}
#[cfg(feature = "object-store")]
{
tokio::runtime::Builder::new_multi_thread()
.enable_all()
.build()?
.block_on(run())
}
}
#[cfg(feature = "object-store")]
mod live {
use std::net::SocketAddr;
use std::sync::Arc;
use anyhow::{Context, Result, bail};
use skade::object_store::{ObjectStore, ObjectStoreConfig, RustS3Store, S3Config};
use skade_katalog_serve::{ServeConfig, WarehouseStorage};
pub fn env(key: &str) -> Option<String> {
std::env::var(key).ok().filter(|v| !v.is_empty())
}
/// Credentials from the environment only — the same names `serve` itself
/// accepts. A flag would land in `ps` output and in shell history.
pub fn creds() -> Result<(String, String)> {
let pick = |names: &[&str]| names.iter().find_map(|n| env(n));
let id = pick(&[
"SKADE_S3_ACCESS_KEY_ID",
"AWS_ACCESS_KEY_ID",
"MINIO_ACCESS_KEY",
])
.context("no S3 access key in the environment")?;
let secret = pick(&[
"SKADE_S3_SECRET_ACCESS_KEY",
"AWS_SECRET_ACCESS_KEY",
"MINIO_SECRET_KEY",
])
.context("no S3 secret key in the environment")?;
Ok((id, secret))
}
pub fn schema() -> Arc<skade::arrow_schema::Schema> {
use skade::arrow_schema::{DataType, Field, Schema};
Arc::new(Schema::new(vec![
Field::new("id", DataType::Int64, false),
Field::new("namn", DataType::Utf8, false),
]))
}
pub fn batch(n: i64) -> skade::arrow_array::RecordBatch {
use skade::arrow_array::{Int64Array, RecordBatch, StringArray};
RecordBatch::try_new(
schema(),
vec![
Arc::new(Int64Array::from((0..n).collect::<Vec<_>>())),
Arc::new(StringArray::from(
(0..n).map(|i| format!("rad-{i}")).collect::<Vec<_>>(),
)),
],
)
.expect("batch")
}
/// One `Connection: close` GET, returning status and body. No HTTP client
/// dep in this crate, and the REST front's answers are small.
pub async fn get(addr: &SocketAddr, path: &str) -> Result<(u16, String)> {
use tokio::io::{AsyncReadExt, AsyncWriteExt};
let mut sock = tokio::net::TcpStream::connect(addr).await?;
sock.write_all(
format!("GET {path} HTTP/1.1\r\nHost: {addr}\r\nConnection: close\r\n\r\n").as_bytes(),
)
.await?;
let mut raw = Vec::new();
sock.read_to_end(&mut raw).await?;
let text = String::from_utf8_lossy(&raw).to_string();
let status = text
.split_whitespace()
.nth(1)
.and_then(|s| s.parse().ok())
.unwrap_or(0u16);
let body = text
.split("\r\n\r\n")
.nth(1)
.unwrap_or("")
.trim()
.to_string();
Ok((status, body))
}
/// Delete everything this run wrote, so a shared bucket does not silt up.
pub async fn cleanup(store: &RustS3Store, prefix: &str) -> Result<usize> {
let keys = store.list(prefix).await?;
for k in &keys {
store.delete(k).await?;
}
Ok(keys.len())
}
pub fn config(endpoint: String, bucket: String) -> Result<S3Config> {
let (access_key_id, secret_access_key) = creds()?;
Ok(S3Config {
endpoint,
region: env("SKADE_LIVE_S3_REGION").unwrap_or_else(|| "us-east-1".into()),
bucket,
access_key_id,
secret_access_key,
// Path-style is the default because it is what MinIO needs; the
// knob exists for a real AWS endpoint.
path_style: env("SKADE_LIVE_S3_VIRTUAL_HOSTED").is_none(),
tls: None,
})
}
pub fn serve_config(
db_path: std::path::PathBuf,
local_root: std::path::PathBuf,
warehouse_uri: &str,
s3: S3Config,
) -> ServeConfig {
let mut cfg = ServeConfig::new(local_root);
cfg.db_path = db_path;
cfg.warehouse_uri = Some(warehouse_uri.to_string());
cfg.storage = WarehouseStorage::ObjectStore(ObjectStoreConfig::RustS3(s3));
cfg
}
pub fn require(cond: bool, what: &str) -> Result<()> {
if cond {
println!(" ok {what}");
Ok(())
} else {
bail!("FAILED: {what}")
}
}
}
#[cfg(feature = "object-store")]
async fn run() -> anyhow::Result<()> {
use anyhow::Context;
use live::*;
use skade::object_store::{ObjectStoreConfig, RustS3Store};
let (Some(endpoint), Some(bucket)) =
(env("SKADE_LIVE_S3_ENDPOINT"), env("SKADE_LIVE_S3_BUCKET"))
else {
eprintln!(
"live_object_store: SKIP — set SKADE_LIVE_S3_ENDPOINT and \
SKADE_LIVE_S3_BUCKET (plus MINIO_ACCESS_KEY / MINIO_SECRET_KEY) to run"
);
return Ok(());
};
let prefix = env("SKADE_LIVE_S3_PREFIX")
.unwrap_or_else(|| format!("skade-live-check/{}", std::process::id()));
let warehouse_uri = format!("s3://{bucket}/{prefix}");
let s3 = config(endpoint.clone(), bucket.clone())?;
println!("live_object_store: {warehouse_uri} via {endpoint}");
let tmp = tempfile::TempDir::new()?;
let db_path = tmp.path().join("catalog.redb");
// ── 1. write with the DATA PLANE, blobs into the bucket ──
{
let wh = skade::Warehouse::open_with_object_store_config(
&db_path,
&warehouse_uri,
ObjectStoreConfig::RustS3(s3.clone()),
)
.await
.context("open the object-store warehouse (endpoint/TLS/credentials)")?;
let mut t = wh.table_or_create("personer", &schema()).await?;
t.append(&[batch(256)]).await?;
println!(" ok wrote 256 rows into {warehouse_uri}/main/personer");
}
// Dropped → the redb write lock is free for the serve front.
// ── 2. serve that warehouse ──
let mut cfg = serve_config(db_path, tmp.path().join("root"), &warehouse_uri, s3.clone());
cfg.rest = Some("127.0.0.1:0".parse()?);
let (rest, _) = skade_katalog_serve::spawn(&cfg).await?;
let rest = rest.expect("rest bound");
println!(" ok REST front on http://{rest}");
// ── 3. drive it the way Doris/Trino/Spark would ──
let (code, body) = get(&rest, "/v1/namespaces").await?;
println!(" GET /v1/namespaces -> {code} {body}");
require(
code == 200 && body.contains("main"),
"listNamespaces returns `main`",
)?;
let (code, body) = get(&rest, "/v1/namespaces/main/tables").await?;
println!(" GET /v1/namespaces/main/tables -> {code} {body}");
require(
code == 200 && body.contains("personer"),
"listTables returns `personer`",
)?;
let (code, body) = get(&rest, "/v1/namespaces/main/tables/personer").await?;
println!(" GET /v1/namespaces/main/tables/personer -> {code}");
require(
code == 200,
"loadTable succeeds over an object-store warehouse",
)?;
let meta: serde_json::Value = serde_json::from_str(&body)?;
let loc = meta["metadata-location"].as_str().unwrap_or_default();
println!(" metadata-location: {loc}");
require(
loc.starts_with(&format!("s3://{bucket}/")),
"the served metadata-location is in the bucket, not a local path",
)?;
require(
meta["metadata"]["current-snapshot-id"].is_i64(),
"the metadata document itself came back through the object store",
)?;
let manifest_list = meta["metadata"]["snapshots"][0]["manifest-list"]
.as_str()
.unwrap_or_default();
println!(" manifest-list: {manifest_list}");
require(
manifest_list.starts_with(&format!("s3://{bucket}/")),
"manifests are addressed in the bucket too",
)?;
// ── 4. tidy up ──
let store = RustS3Store::new(s3)?;
if env("SKADE_LIVE_S3_KEEP").is_some() {
println!(" kept {warehouse_uri} (SKADE_LIVE_S3_KEEP)");
} else {
let n = cleanup(&store, &format!("{prefix}/")).await?;
println!(" ok removed {n} objects under {prefix}/");
}
println!("live_object_store: PASS");
Ok(())
}