diff --git a/seaweed-vector/crates/server/Cargo.toml b/seaweed-vector/crates/server/Cargo.toml index 853aee0db8..22fcaf3c21 100644 --- a/seaweed-vector/crates/server/Cargo.toml +++ b/seaweed-vector/crates/server/Cargo.toml @@ -32,7 +32,7 @@ bytes = "1" clap = { version = "4", features = ["derive", "env"] } futures = "0.3" moka = { version = "0.12", features = ["sync"] } -object_store = "0.13" +object_store = { version = "0.13", features = ["aws"] } serde = { version = "1", features = ["derive"] } serde_json = "1" toml = "0.8" diff --git a/seaweed-vector/crates/server/src/main.rs b/seaweed-vector/crates/server/src/main.rs index d9eeb21267..6b5304877b 100644 --- a/seaweed-vector/crates/server/src/main.rs +++ b/seaweed-vector/crates/server/src/main.rs @@ -4,7 +4,6 @@ use anyhow::{Context, Result}; use clap::Parser; use tokio_stream::wrappers::TcpListenerStream; use tonic::transport::Server; -use weed_lance_worker::catalog::{NamespaceClient, SigningKey}; use weed_lance_worker::dataset::FallbackOptions; use weed_vector_server::pb::vector::vector_service_server::VectorServiceServer; use weed_vector_server::read_cache::ReadCache; @@ -15,18 +14,12 @@ use weed_vector_server::tls::{self, ServerTls}; /// The VectorService the S3 gateway fans vector calls out to. #[derive(Parser)] struct Args { - /// Lance Namespace base URL: the S3 gateway's Lance port (-s3.port.lance). - #[arg( - long, - env = "WEED_VECTOR_NAMESPACE", - default_value = "http://localhost:9101" - )] - namespace: String, /// gRPC listen address. #[arg(long, env = "WEED_VECTOR_LISTEN", default_value = "127.0.0.1:8090")] listen: String, - /// object_store option fallback for deployments that vend no credentials: - /// key=value, repeatable. + /// object_store options for deployments whose gateway vends no + /// credentials: key=value, repeatable. Credentials given here are used + /// only with the aws_endpoint given here. #[arg(long = "storage-option", alias = "storage.option", value_parser = parse_option)] storage_options: Vec<(String, String)>, /// Comma-separated master addresses to register this server with; the @@ -40,10 +33,6 @@ struct Args { /// tenth of the index's partitions, at least 8; more under a filter. #[arg(long, env = "WEED_VECTOR_NPROBES")] nprobes: Option, - /// The `jwt.filer_signing.key` of the cluster. The Lance namespace vends - /// credentials that write index data only to a caller holding it. - #[arg(long, env = "WEED_FILER_SIGNING_KEY")] - filer_signing_key: Option, /// The cluster's security.toml, for gRPC TLS: this server's identity is /// [grpc.vector], the masters are dialed with [grpc.client]. By default /// the file weed processes find in -config_dir, ., ~/.seaweedfs and @@ -54,6 +43,13 @@ struct Args { /// first. #[arg(long, alias = "config_dir", env = "WEED_VECTOR_CONFIG_DIR")] config_dir: Option, + /// Ignored: the gateway resolves every table it sends. Still accepted so + /// an existing command line starts. + #[arg(long, hide = true)] + namespace: Option, + /// Ignored, like --namespace. + #[arg(long, hide = true)] + filer_signing_key: Option, } fn parse_option(s: &str) -> Result<(String, String), String> { @@ -123,6 +119,11 @@ fn advertised_addr(listen: &str, masters: &[String]) -> String { async fn main() -> Result<()> { tracing_subscriber::fmt::init(); let args = Args::parse_from(normalize_args(std::env::args().collect())); + if args.namespace.is_some() || args.filer_signing_key.is_some() { + tracing::warn!( + "--namespace and --filer-signing-key are ignored: the gateway resolves every table it sends" + ); + } weed_lance_worker::tls::install_default_crypto_provider(); let security = Security::load(args.security_toml.as_deref(), args.config_dir.as_deref())?; if let Some(path) = security.path() { @@ -154,17 +155,12 @@ async fn main() -> Result<()> { )); } let fallback: FallbackOptions = args.storage_options.into_iter().collect(); - let signing_key = SigningKey::new(args.filer_signing_key); - let client = NamespaceClient::new(args.namespace.clone()) - .with_fallback(&fallback) - .with_signing_key(signing_key.as_ref()); - let mut service = VectorServer::new(client, fallback) - .with_nprobes(args.nprobes.map(std::num::NonZeroUsize::get)); + let mut service = + VectorServer::new(fallback).with_nprobes(args.nprobes.map(std::num::NonZeroUsize::get)); if args.read_cache_mb > 0 { service = service.with_read_cache(ReadCache::new(args.read_cache_mb << 20)); } tracing::info!( - namespace = %args.namespace, listen = %args.listen, "weed-vector; serving gRPC {}, dialing masters {}", match &server_tls { @@ -208,4 +204,21 @@ mod tests { let args = Args::try_parse_from(["weed-vector", "--nprobes", "12"]).unwrap(); assert_eq!(args.nprobes.map(std::num::NonZeroUsize::get), Some(12)); } + + #[test] + fn an_earlier_command_line_still_parses() { + let args = Args::try_parse_from(normalize_args( + [ + "weed-vector", + "--namespace", + "http://localhost:9101", + "-filer-signing-key", + "k", + ] + .map(String::from) + .to_vec(), + )) + .unwrap(); + assert!(args.namespace.is_some() && args.filer_signing_key.is_some()); + } } diff --git a/seaweed-vector/crates/server/src/read_cache.rs b/seaweed-vector/crates/server/src/read_cache.rs index 748bd4b66d..8260b1d690 100644 --- a/seaweed-vector/crates/server/src/read_cache.rs +++ b/seaweed-vector/crates/server/src/read_cache.rs @@ -37,26 +37,40 @@ struct Entry { /// Wraps each object store Lance opens; clones share one cache. #[derive(Clone)] -pub struct ReadCache(moka::sync::Cache); +pub struct ReadCache { + cache: moka::sync::Cache, + /// Keeps entries apart from those of another [`Self::scoped`] view, for + /// stores that share a prefix but not their bytes. + scope: Arc, +} impl ReadCache { pub fn new(capacity_bytes: u64) -> Self { - Self( - moka::sync::Cache::builder() + Self { + cache: moka::sync::Cache::builder() .max_capacity(capacity_bytes) .weigher(|key: &Key, entry: &Entry| { u32::try_from(key.path.as_ref().len() + entry.bytes.len()).unwrap_or(u32::MAX) }) .build(), - ) + scope: "".into(), + } + } + + /// The same cache, keyed apart under `scope`. + pub fn scoped(&self, scope: &str) -> Self { + Self { + cache: self.cache.clone(), + scope: scope.into(), + } } } impl fmt::Debug for ReadCache { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.debug_struct("ReadCache") - .field("entries", &self.0.entry_count()) - .field("bytes", &self.0.weighted_size()) + .field("entries", &self.cache.entry_count()) + .field("bytes", &self.cache.weighted_size()) .finish() } } @@ -65,8 +79,8 @@ impl WrappingObjectStore for ReadCache { fn wrap(&self, store_prefix: &str, original: Arc) -> Arc { Arc::new(CachedStore { inner: original, - store: store_prefix.into(), - cache: self.0.clone(), + store: format!("{}|{store_prefix}", self.scope).into(), + cache: self.cache.clone(), }) } } @@ -277,7 +291,6 @@ mod tests { use object_store::ObjectStoreExt; use object_store::memory::InMemory; use tonic::Request; - use weed_lance_worker::catalog::NamespaceClient; use weed_lance_worker::dataset::FallbackOptions; fn cached(inner: Arc) -> Arc { @@ -373,8 +386,8 @@ mod tests { .unwrap(); read(&store, &path, 0..(8 << 10)).await; } - cache.0.run_pending_tasks(); - assert!(cache.0.weighted_size() <= 64 << 10, "{cache:?}"); + cache.cache.run_pending_tasks(); + assert!(cache.cache.weighted_size() <= 64 << 10, "{cache:?}"); } #[tokio::test] @@ -392,40 +405,15 @@ mod tests { let backing = whole.as_ptr_range(); read(&store, &path, 10..20).await; let key = Key { - store: "memory".into(), + store: "|memory".into(), path, range: Some(10..20), }; - let cached = cache.0.get(&key).unwrap().bytes; + let cached = cache.cache.get(&key).unwrap().bytes; assert_eq!(cached.len(), 10); assert!(!backing.contains(&cached.as_ptr())); } - /// A catalog that answers every request with `location`. - async fn catalog_at(location: String) -> String { - use tokio::io::{AsyncReadExt, AsyncWriteExt}; - let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - let url = format!("http://{}", listener.local_addr().unwrap()); - let body = serde_json::json!({ "location": location }).to_string(); - tokio::spawn(async move { - loop { - let (mut conn, _) = listener.accept().await.unwrap(); - let body = body.clone(); - tokio::spawn(async move { - let mut request = [0u8; 8192]; - let _ = conn.read(&mut request).await; - let response = format!( - "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}", - body.len() - ); - let _ = conn.write_all(response.as_bytes()).await; - let _ = conn.shutdown().await; - }); - } - }); - url - } - fn vector(key: &str, data: Vec) -> pb::Vector { pb::Vector { key: key.to_string(), @@ -492,15 +480,15 @@ mod tests { /// dropped and recreated at the same location answers with its new rows. #[tokio::test] async fn cached_reads_follow_writes_and_a_recreated_index() { - let location = format!("shared-memory://cache-{}/vector/x", uuid_like(),); - let server = VectorServer::new( - NamespaceClient::new(catalog_at(location.clone()).await), - FallbackOptions::new(), - ) - .with_read_cache(ReadCache::new(16 << 20)); + let bucket = format!("cache-{}", uuid_like()); + let location = format!("shared-memory://{bucket}/vector/x"); + let server = VectorServer::new(FallbackOptions::new()) + .with_schemes(&["shared-memory"]) + .with_read_cache(ReadCache::new(16 << 20)); let table = pb::Table { - id: vec!["bkt".into(), "vector".into(), "x".into()], - version: 0, + id: vec![bucket, "vector".into(), "x".into()], + location: location.clone(), + ..Default::default() }; create(&server, &table, 4).await; diff --git a/seaweed-vector/crates/server/src/service.rs b/seaweed-vector/crates/server/src/service.rs index 6237b66b44..718aa1f77d 100644 --- a/seaweed-vector/crates/server/src/service.rs +++ b/seaweed-vector/crates/server/src/service.rs @@ -12,12 +12,16 @@ use lance::io::ObjectStoreParams; use lance::io::StorageOptionsAccessor; use lance::io::WrappingObjectStore; use lance::table::format::Fragment; +use object_store::aws::AmazonS3ConfigKey; use prost_types::{ListValue, Struct, Value, value::Kind}; use std::collections::{BTreeMap, HashMap}; +use std::ffi::OsString; +use std::str::FromStr; use std::sync::Arc; use tonic::{Request, Response, Status}; -use weed_lance_worker::catalog::NamespaceClient; +use weed_lance_worker::catalog::TableDescription; use weed_lance_worker::dataset::{self, FallbackOptions}; +use weed_lance_worker::jobs::vector_index::is_vector_table; use weed_vector_core::index; use weed_vector_core::query::{self, Query}; use weed_vector_core::table::VectorTable; @@ -35,17 +39,14 @@ fn invalid(err: impl Into) -> Status { fn open_status(err: anyhow::Error) -> Status { let missing = err.chain().any(|cause| { - cause - .downcast_ref::() - .is_some_and(|e| e.status() == Some(reqwest::StatusCode::NOT_FOUND)) - || cause.downcast_ref::().is_some_and(|e| { - matches!( - e, - lance::Error::DatasetNotFound { .. } - | lance::Error::NotFound { .. } - | lance::Error::VersionNotFound { .. } - ) - }) + cause.downcast_ref::().is_some_and(|e| { + matches!( + e, + lance::Error::DatasetNotFound { .. } + | lance::Error::NotFound { .. } + | lance::Error::VersionNotFound { .. } + ) + }) }); if missing { Status::not_found(format!("{:#}", err)) @@ -58,6 +59,69 @@ const KEY_COLUMN: &str = "key"; const METADATA_COLUMN: &str = "metadata"; const VECTOR_COLUMN: &str = "vector"; const ROW_ID_COLUMN: &str = "_rowid"; +const ENDPOINT_OPTION: &str = "aws_endpoint"; +/// The options a gateway's describe produces. Any other key a request sends, +/// such as a proxy or an endpoint alias, could route the engine's credentials +/// elsewhere, so it is not taken from the request. +const SENT_OPTIONS: [&str; 8] = [ + ENDPOINT_OPTION, + "allow_http", + "aws_region", + "aws_skip_signature", + "aws_access_key_id", + "aws_secret_access_key", + "aws_session_token", + "expires_at_millis", +]; +const CREDENTIAL_OPTIONS: [&str; 3] = [ + "aws_access_key_id", + "aws_secret_access_key", + "aws_session_token", +]; + +/// Whether a request's options may be used with the engine's fallback: known +/// keys only, and an endpoint that is the configured one, or any when no +/// credential of the engine's can reach it: none configured, and the request +/// either skips signing or signs with its own credentials alone. `env` names +/// the engine's environment variables. +fn options_trusted( + sent: &HashMap, + fallback: &FallbackOptions, + env: impl IntoIterator, +) -> bool { + if sent.keys().any(|key| !SENT_OPTIONS.contains(&key.as_str())) { + return false; + } + match (sent.get(ENDPOINT_OPTION), fallback.get(ENDPOINT_OPTION)) { + (None, _) => true, + (Some(sent), Some(pinned)) => sent.trim_end_matches('/') == pinned.trim_end_matches('/'), + (Some(_), None) => { + let configured = CREDENTIAL_OPTIONS + .iter() + .any(|key| fallback.contains_key(*key)); + let unsigned = sent.get("aws_skip_signature").is_some_and(|v| v == "true"); + !configured && (unsigned || signs_alone(sent, env)) + } + } +} + +/// Whether lance signs with the request's credentials alone. Lance fills an +/// option the request leaves out from an environment variable under any of +/// object_store's names for it, and asks the AWS default chain when the key +/// pair is incomplete. +fn signs_alone(sent: &HashMap, env: impl IntoIterator) -> bool { + let has = |key: &str| sent.contains_key(key); + let env_token = || { + env.into_iter().any(|name| { + name.to_str() + .and_then(|name| AmazonS3ConfigKey::from_str(&name.to_ascii_lowercase()).ok()) + == Some(AmazonS3ConfigKey::Token) + }) + }; + has("aws_access_key_id") + && has("aws_secret_access_key") + && (has("aws_session_token") || !env_token()) +} fn schema_for( dimension: usize, @@ -370,23 +434,50 @@ fn rows_as_vectors( out } +/// The description a gateway sent with the table. Only a vector index at its +/// colocated path is taken, since the engine is that path's writer already; +/// any other location would point the engine at storage the catalog does not +/// know. +fn resolved(table: &pb::Table, schemes: &[&str]) -> Option { + let colocated = dataset::table_location(&table.id)?; + let path = colocated.strip_prefix("s3://")?; + let location = table.location.trim_end_matches('/'); + let trusted = is_vector_table(&table.id) + && schemes + .iter() + .any(|scheme| location == format!("{scheme}://{path}")); + trusted.then(|| TableDescription { + location: location.to_string(), + storage_options: table.storage_options.clone(), + managed_versioning: false, + }) +} + pub struct VectorServer { - client: NamespaceClient, fallback: FallbackOptions, - read_cache: Option>, + read_cache: Option, nprobes: Option, + /// URI schemes a table's location may have. + schemes: &'static [&'static str], } impl VectorServer { - pub fn new(client: NamespaceClient, fallback: FallbackOptions) -> Self { + pub fn new(fallback: FallbackOptions) -> Self { Self { - client, fallback, read_cache: None, nprobes: None, + schemes: &["s3"], } } + /// Tables on an in-process store, for tests. + #[cfg(test)] + pub(crate) fn with_schemes(mut self, schemes: &'static [&'static str]) -> Self { + self.schemes = schemes; + self + } + /// Partitions every query probes, instead of the index's own default. pub fn with_nprobes(mut self, nprobes: Option) -> Self { self.nprobes = nprobes; @@ -395,21 +486,64 @@ impl VectorServer { /// Serve repeated reads of write-once table files from `cache`. pub fn with_read_cache(mut self, cache: ReadCache) -> Self { - self.read_cache = Some(Arc::new(cache)); + self.read_cache = Some(cache); self } - async fn open(&self, table: &pb::Table) -> Result { - let opened = dataset::open_with( - &self.client, - &table.id, - &self.fallback, - self.read_cache.clone(), - ) - .await?; - let mut dataset = opened.dataset; + /// The table as the gateway resolved it. A request that names no + /// location, another location than the index's own, or storage options + /// the engine's credentials must not be used with is refused. + fn describe(&self, table: &pb::Table) -> Result { + if table.location.is_empty() { + return Err(Status::invalid_argument("the table names no location")); + } + let description = resolved(table, self.schemes).ok_or_else(|| { + Status::permission_denied(format!( + "{} is not the location of index {}", + table.location, + table.id.join("/") + )) + })?; + let env = std::env::vars_os().map(|(name, _)| name); + if !options_trusted(&description.storage_options, &self.fallback, env) { + // The gateway answers AccessDenied without logging, so this is + // where an operator sees a pinned endpoint that does not match. + let endpoint = |options: &HashMap| { + options.get(ENDPOINT_OPTION).cloned().unwrap_or_default() + }; + tracing::warn!( + index = %table.id.join("/"), + sent = %endpoint(&description.storage_options), + pinned = %endpoint(&self.fallback), + "refused the request's storage options: an aws_endpoint other than the pinned one, a key other than the gateway's, or credentials the engine would have to complete" + ); + return Err(Status::permission_denied( + "the storage options would send the engine's credentials elsewhere", + )); + } + Ok(description) + } + + async fn open(&self, table: &pb::Table) -> Result { + let description = self.describe(table)?; + // The options name the store the bytes come from; the cache is scoped + // by it so the same bucket and path on another endpoint never share. + let cache = self.read_cache.as_ref().map(|cache| { + let options = dataset::table_options(&description, &self.fallback); + let endpoint = options + .get(ENDPOINT_OPTION) + .map(String::as_str) + .unwrap_or(""); + Arc::new(cache.scoped(endpoint)) as Arc + }); + let mut dataset = dataset::open_at(&description, &self.fallback, cache) + .await + .map_err(open_status)?; if table.version != 0 { - dataset = dataset.checkout_version(table.version).await?; + dataset = dataset + .checkout_version(table.version) + .await + .map_err(|err| open_status(err.into()))?; } Ok(dataset) } @@ -433,11 +567,7 @@ impl VectorService for VectorServer { return Err(Status::invalid_argument("dimension must be 1..=4096")); } // The gateway has already created the table through the catalog. - let description = self - .client - .describe_table(&table.id) - .await - .map_err(internal)?; + let description = self.describe(&table)?; let options = dataset::table_options(&description, &self.fallback); let schema = schema_for(dimension, metric, &req.non_filterable_metadata_keys); let empty = RecordBatch::new_empty(schema.clone()); @@ -470,7 +600,8 @@ impl VectorService for VectorServer { let table = req .table .ok_or_else(|| Status::invalid_argument("missing table"))?; - dataset::drop(&self.client, &table.id, &self.fallback) + let description = self.describe(&table)?; + dataset::drop_at(&table.id, &description, &self.fallback) .await .map_err(|err| { if err.is::() { @@ -490,7 +621,7 @@ impl VectorService for VectorServer { let table = req .table .ok_or_else(|| Status::invalid_argument("missing table"))?; - let dataset = self.open(&table).await.map_err(open_status)?; + let dataset = self.open(&table).await?; if req.vectors.is_empty() { return Ok(Response::new(pb::PutVectorsResponse { version: dataset.version().version, @@ -532,7 +663,7 @@ impl VectorService for VectorServer { let table = req .table .ok_or_else(|| Status::invalid_argument("missing table"))?; - let dataset = self.open(&table).await.map_err(open_status)?; + let dataset = self.open(&table).await?; // Lease the resolved version so maintenance cannot reclaim its files // while this query runs against it. let _lease = weed_vector_core::maintenance::try_pin(&dataset) @@ -556,7 +687,7 @@ impl VectorService for VectorServer { let table = req .table .ok_or_else(|| Status::invalid_argument("missing table"))?; - let mut dataset = self.open(&table).await.map_err(open_status)?; + let mut dataset = self.open(&table).await?; if !req.keys.is_empty() { dataset .delete(&key_list(&req.keys)) @@ -582,7 +713,7 @@ impl VectorService for VectorServer { "segment_index must be below segment_count, which is at most {MAX_SEGMENTS}" ))); } - let dataset = self.open(&table).await.map_err(open_status)?; + let dataset = self.open(&table).await?; let offset: u64 = if req.next_token.is_empty() { 0 } else { @@ -649,7 +780,7 @@ impl VectorService for VectorServer { let table = req .table .ok_or_else(|| Status::invalid_argument("missing table"))?; - let dataset = self.open(&table).await.map_err(open_status)?; + let dataset = self.open(&table).await?; // Lease the resolved version so maintenance cannot reclaim its files // while this query runs against it. let _lease = weed_vector_core::maintenance::try_pin(&dataset) @@ -941,45 +1072,210 @@ mod tests { assert_eq!(next, None); } - /// A catalog that answers one describe with `location`. - async fn catalog_describing(location: &str) -> String { - use tokio::io::{AsyncReadExt, AsyncWriteExt}; - let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - let url = format!("http://{}", listener.local_addr().unwrap()); - let body = serde_json::json!({ "location": location }).to_string(); - tokio::spawn(async move { - let (mut conn, _) = listener.accept().await.unwrap(); - let mut request = [0u8; 8192]; - let _ = conn.read(&mut request).await; - let response = format!( - "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}", - body.len() - ); - conn.write_all(response.as_bytes()).await.unwrap(); - let _ = conn.shutdown().await; - let _ = tokio::io::copy(&mut conn, &mut tokio::io::sink()).await; - }); - url - } - + /// A location that is not the index's own is refused before anything is + /// opened or removed. #[tokio::test] async fn drop_index_refuses_a_location_outside_the_table() { - for location in ["s3://bkt/vector", "s3://bkt/vector/prod"] { - let server = VectorServer::new( - NamespaceClient::new(catalog_describing(location).await), - FallbackOptions::new(), - ); + let server = VectorServer::new(FallbackOptions::new()); + for location in [ + "s3://bkt/vector", + "s3://bkt/vector/prod", + "s3://other/vector/x", + ] { let request = pb::DropIndexRequest { table: Some(pb::Table { id: vec!["bkt".into(), "vector".into(), "x".into()], - version: 0, + location: location.into(), + ..Default::default() }), }; let err = server .drop_index(Request::new(request)) .await .expect_err(location); - assert_eq!(err.code(), tonic::Code::FailedPrecondition, "{location}"); + assert_eq!(err.code(), tonic::Code::PermissionDenied, "{location}"); } } + + /// Only a vector index at its colocated path is taken from the request. + #[test] + fn resolved_takes_only_a_colocated_vector_index() { + let table = |id: &[&str], location: &str| pb::Table { + id: id.iter().map(|s| s.to_string()).collect(), + location: location.into(), + storage_options: HashMap::from([("aws_endpoint".to_string(), "http://gw".to_string())]), + ..Default::default() + }; + let s3 = &["s3"]; + let index = ["bkt", "vector", "x"]; + for location in ["s3://bkt/vector/x", "s3://bkt/vector/x/"] { + let description = resolved(&table(&index, location), s3).expect(location); + assert_eq!(description.location, "s3://bkt/vector/x"); + assert_eq!( + description + .storage_options + .get("aws_endpoint") + .map(String::as_str), + Some("http://gw") + ); + } + assert!(resolved(&table(&index, ""), s3).is_none()); + assert!(resolved(&table(&index, "s3://other/vector/x"), s3).is_none()); + assert!(resolved(&table(&index, "s3://bkt/vector/y"), s3).is_none()); + assert!(resolved(&table(&index, "file://bkt/vector/x"), s3).is_none()); + assert!(resolved(&table(&index, "memory://bkt/vector/x"), &["s3", "memory"]).is_some()); + assert!(resolved(&table(&["bkt", "ns", "x"], "s3://bkt/ns/x"), s3).is_none()); + assert!(resolved(&table(&["bkt", "x"], "s3://bkt/x"), s3).is_none()); + } + + /// A request names the table's location or is refused. + #[tokio::test] + async fn a_table_without_its_location_is_refused() { + let server = VectorServer::new(FallbackOptions::new()); + for (location, code) in [ + ("", tonic::Code::InvalidArgument), + ("s3://other/vector/x", tonic::Code::PermissionDenied), + ] { + let err = server + .get_vectors(Request::new(pb::GetVectorsRequest { + table: Some(pb::Table { + id: vec!["bkt".into(), "vector".into(), "x".into()], + location: location.into(), + ..Default::default() + }), + keys: vec!["k001".into()], + ..Default::default() + })) + .await + .expect_err(location); + assert_eq!(err.code(), code, "{location}"); + } + } + + /// A request's options are used with the engine's credentials only when + /// they cannot send those credentials elsewhere. + #[test] + fn options_are_trusted_only_where_credentials_cannot_leave() { + let options = |pairs: &[(&str, &str)]| -> HashMap { + pairs + .iter() + .map(|(k, v)| (k.to_string(), v.to_string())) + .collect() + }; + let creds = options(&[("aws_access_key_id", "k"), ("aws_secret_access_key", "s")]); + let pinned = options(&[ + ("aws_access_key_id", "k"), + ("aws_secret_access_key", "s"), + (ENDPOINT_OPTION, "http://gw"), + ]); + let gateway = options(&[(ENDPOINT_OPTION, "http://gw"), ("allow_http", "true")]); + let elsewhere = options(&[(ENDPOINT_OPTION, "http://elsewhere")]); + let unsigned = options(&[ + (ENDPOINT_OPTION, "http://gw"), + ("aws_skip_signature", "true"), + ]); + let vended = options(&[ + (ENDPOINT_OPTION, "http://gw"), + ("aws_access_key_id", "v"), + ("aws_secret_access_key", "v"), + ]); + let none = FallbackOptions::new(); + let trusted = |sent: &HashMap, fallback: &FallbackOptions| { + options_trusted(sent, fallback, Vec::::new()) + }; + + assert!(trusted(&gateway, &pinned)); + assert!(trusted( + &options(&[(ENDPOINT_OPTION, "http://gw/")]), + &pinned + )); + assert!(!trusted(&elsewhere, &pinned)); + assert!(!trusted(&gateway, &creds), "no configured endpoint"); + assert!(!trusted(&gateway, &none), "the default chain could sign"); + assert!(trusted(&unsigned, &none)); + assert!(trusted(&vended, &none)); + assert!( + !options_trusted(&vended, &none, [OsString::from("AWS_SESSION_TOKEN")]), + "the environment would add its session token" + ); + assert!( + !trusted( + &options(&[(ENDPOINT_OPTION, "http://gw"), ("aws_access_key_id", "v")]), + &none + ), + "a key without its secret is signed by the default chain" + ); + assert!( + !trusted(&vended, &creds), + "configured credentials stay home" + ); + assert!( + trusted(&options(&[("aws_region", "r")]), &creds), + "no endpoint sent" + ); + for key in ["proxy_url", "aws_endpoint_url_s3", "aws_endpoint_url"] { + let routed = options(&[(key, "http://elsewhere"), ("aws_skip_signature", "true")]); + assert!(!trusted(&routed, &none), "{key}"); + } + } + + /// A request signs alone only with a full key pair, and with its own + /// session token whenever the environment would supply one. + #[test] + fn a_request_signs_alone_without_the_environment() { + let sent = |pairs: &[(&str, &str)]| -> HashMap { + pairs + .iter() + .map(|(k, v)| (k.to_string(), v.to_string())) + .collect() + }; + let env = |names: &[&str]| -> Vec { names.iter().map(OsString::from).collect() }; + let pair = sent(&[("aws_access_key_id", "v"), ("aws_secret_access_key", "v")]); + let with_token = sent(&[ + ("aws_access_key_id", "v"), + ("aws_secret_access_key", "v"), + ("aws_session_token", "t"), + ]); + + assert!(signs_alone(&pair, env(&["AWS_REGION", "HOME"]))); + assert!(!signs_alone(&sent(&[("aws_access_key_id", "v")]), env(&[]))); + assert!(!signs_alone( + &sent(&[("aws_secret_access_key", "v")]), + env(&[]) + )); + for name in [ + "AWS_SESSION_TOKEN", + "aws_session_token", + "TOKEN", + "SESSION_TOKEN", + ] { + assert!(!signs_alone(&pair, env(&[name])), "{name}"); + assert!(signs_alone(&with_token, env(&[name])), "{name}"); + } + } + + /// Options the engine must not use are refused, for a drop as for an open. + #[tokio::test] + async fn untrusted_options_are_refused() { + let sent = |endpoint: &str| pb::Table { + id: vec!["bkt".into(), "vector".into(), "x".into()], + location: "s3://bkt/vector/x".into(), + storage_options: HashMap::from([(ENDPOINT_OPTION.to_string(), endpoint.to_string())]), + ..Default::default() + }; + let pinned = + FallbackOptions::from([(ENDPOINT_OPTION.to_string(), "http://gw".to_string())]); + let server = VectorServer::new(pinned); + let description = server.describe(&sent("http://gw")).unwrap(); + assert_eq!(description.location, "s3://bkt/vector/x"); + let err = server.describe(&sent("http://elsewhere")).unwrap_err(); + assert_eq!(err.code(), tonic::Code::PermissionDenied); + let err = server + .drop_index(Request::new(pb::DropIndexRequest { + table: Some(sent("http://elsewhere")), + })) + .await + .unwrap_err(); + assert_eq!(err.code(), tonic::Code::PermissionDenied); + } } diff --git a/seaweed-worker/crates/lance/src/dataset.rs b/seaweed-worker/crates/lance/src/dataset.rs index cdca46d0bd..e3cd6d0f0b 100644 --- a/seaweed-worker/crates/lance/src/dataset.rs +++ b/seaweed-worker/crates/lance/src/dataset.rs @@ -88,7 +88,8 @@ pub fn table_options( options } -async fn open_at( +/// Open a table the caller has already resolved. +pub async fn open_at( description: &TableDescription, fallback: &FallbackOptions, wrapper: Option>, @@ -118,13 +119,22 @@ pub async fn drop( Err(err) if describe_not_found(&err) => return Ok(()), Err(err) => return Err(err), }; + drop_at(id, &description, fallback).await +} + +/// [`drop`] for a table the caller has already resolved. +pub async fn drop_at( + id: &[String], + description: &TableDescription, + fallback: &FallbackOptions, +) -> Result<()> { check_drop_location(id, &description.location)?; let (store, path) = ObjectStore::from_uri_and_params( Arc::new(ObjectStoreRegistry::default()), &description.location, &ObjectStoreParams { storage_options_accessor: Some(Arc::new(StorageOptionsAccessor::with_static_options( - table_options(&description, fallback), + table_options(description, fallback), ))), ..Default::default() }, diff --git a/test/s3vectors/DESIGN.md b/test/s3vectors/DESIGN.md index 95aaa19a14..954eaaf2e6 100644 --- a/test/s3vectors/DESIGN.md +++ b/test/s3vectors/DESIGN.md @@ -45,10 +45,10 @@ engine RPC, so they work against a plain `weed server`: ### Tier 2 — vector engine required (skipped when absent) `TestVectorDataPlaneLifecycle` spawns a real `weed-vector` against the -environment's lance listener (`-s3.port.lance`), mirroring the cluster the -pytest suite builds in `conftest.py`: `--namespace` points at the lance -HTTP port, `--masters` at `master.http.grpc`, and the storage options carry -the same test credentials the IAM config writes. The binary resolves as +environment's S3 endpoint, mirroring the cluster the pytest suite builds in +`conftest.py`: `--masters` points at `master.http.grpc`, and the storage +options carry the same test credentials the IAM config writes, with +`aws_endpoint` at the S3 port. The binary resolves as `VECTOR_BINARY` env → `./weed-vector` (the Makefile's output) → PATH; the test skips when none is found, and `waitForEngineDiscovery` polls a data-plane op until the gateway routes to the registered engine instead of diff --git a/test/s3vectors/conftest.py b/test/s3vectors/conftest.py index 1279d60233..1fd95c49fc 100644 --- a/test/s3vectors/conftest.py +++ b/test/s3vectors/conftest.py @@ -304,8 +304,10 @@ def run_cluster(param, tmp_path_factory, identities): # would find it as a credential. if identities: config, unset = [f"-config={HERE / 's3.json'}"], () + # The engine uses its credentials only with the endpoint it is given. credentials = ["--storage-option", "aws_access_key_id=test", - "--storage-option", "aws_secret_access_key=test"] + "--storage-option", "aws_secret_access_key=test", + "--storage-option", f"aws_endpoint={endpoint}"] else: config, unset, credentials = [], ("AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY"), [] # VECTOR_ENCRYPT_VOLUMES=1 has the gateway encrypt every chunk it @@ -318,8 +320,7 @@ def run_cluster(param, tmp_path_factory, identities): wait_for(lambda: listening(s3), processes, "s3 not ready") if os.environ.get("VECTOR_TLS") or read_jwt_key or encrypt: probe_hardening(endpoint, filer, vol, directory, read_jwt_key) - start("vector", [engine, "--namespace", f"http://127.0.0.1:{lance}", - "--masters", master_addr, "--listen", f"127.0.0.1:{vector}", *credentials], unset) + start("vector", [engine, "--masters", master_addr, "--listen", f"127.0.0.1:{vector}", *credentials], unset) wait_for(lambda: listening(vector), processes, "vector engine not ready") if os.environ.get("VECTOR_TLS"): wait_for(lambda: listening(s3_grpc), processes, "s3 gRPC not ready") diff --git a/test/s3vectors/s3vectors_integration_test.go b/test/s3vectors/s3vectors_integration_test.go index a068e24bfd..93494577df 100644 --- a/test/s3vectors/s3vectors_integration_test.go +++ b/test/s3vectors/s3vectors_integration_test.go @@ -173,18 +173,18 @@ func vectorEngineBinary(t *testing.T) string { return bin } -// startVectorEngine spawns weed-vector against the environment's lance -// namespace endpoint and waits for its listener to come up — mirroring the -// cluster the pytest suite builds in conftest.py. +// startVectorEngine spawns weed-vector against the environment's S3 endpoint +// and waits for its listener to come up — mirroring the cluster the pytest +// suite builds in conftest.py. func (e *TestEnvironment) startVectorEngine(t *testing.T, bin string) { t.Helper() masters := fmt.Sprintf("%s:%d.%d", e.bindIP, e.masterPort, e.masterPort+testutil.GrpcPortOffset) cmd := exec.Command(bin, - "--namespace", fmt.Sprintf("http://%s:%d", e.bindIP, e.lancePort), "--masters", masters, "--listen", fmt.Sprintf("%s:%d", e.bindIP, e.vectorPort), "--storage-option", "aws_access_key_id="+e.accessKey, "--storage-option", "aws_secret_access_key="+e.secretKey, + "--storage-option", fmt.Sprintf("aws_endpoint=http://%s:%d", e.bindIP, e.s3Port), ) cmd.Dir = e.dataDir log, err := os.Create(filepath.Join(e.dataDir, "weed-vector.log")) diff --git a/weed/command/s3.go b/weed/command/s3.go index 29e9a23f2d..b6eb942b7d 100644 --- a/weed/command/s3.go +++ b/weed/command/s3.go @@ -104,6 +104,7 @@ type S3Options struct { // whatever process may hold the configured port. icebergBound atomic.Bool lanceBound atomic.Bool + lanceServer *lance.Server } func init() { @@ -425,6 +426,9 @@ func (s3opt *S3Options) startS3Server() bool { var httpServers []*http.Server + // Once, before either catalog serves a request that reads it. + s3opt.setCredentialRole(s3ApiServer) + // Start Iceberg REST Catalog server if enabled if *s3opt.portIceberg > 0 { if httpS := s3opt.startIcebergServer(s3ApiServer, listenOnLocalhost); httpS != nil { @@ -432,9 +436,11 @@ func (s3opt *S3Options) startS3Server() bool { } } - // Start Lance Namespace server if enabled + // The S3 Vectors API resolves its tables through the catalog in process, + // whether or not the catalog's own port is served. + s3opt.lanceServer = s3opt.newLanceServer(s3ApiServer) if s3opt.portLance != nil && *s3opt.portLance > 0 { - if httpS := s3opt.startLanceServer(s3ApiServer, listenOnLocalhost); httpS != nil { + if httpS := s3opt.startLanceServer(listenOnLocalhost); httpS != nil { httpServers = append(httpServers, httpS) } } @@ -615,6 +621,24 @@ func (s3opt *S3Options) newShutdown(stopGrpc func(), s3ApiServer *s3api.S3ApiSer return shutdown } +// vendsCredentials reports whether -iceberg.credentialRole is set, which +// both catalogs vend through, whichever of them runs. +func (s3opt *S3Options) vendsCredentials() bool { + return s3opt.icebergCredentialRole != nil && *s3opt.icebergCredentialRole != "" +} + +// setCredentialRole hands -iceberg.credentialRole to the S3 server. +func (s3opt *S3Options) setCredentialRole(s3ApiServer *s3api.S3ApiServer) { + if !s3opt.vendsCredentials() { + return + } + duration := int64(0) + if s3opt.icebergCredentialDuration != nil { + duration = int64(*s3opt.icebergCredentialDuration) + } + s3ApiServer.SetIcebergCredentialRole(*s3opt.icebergCredentialRole, duration) +} + // startIcebergServer starts the Iceberg REST Catalog server on a separate port. func (s3opt *S3Options) startIcebergServer(s3ApiServer *s3api.S3ApiServer, listenOnLocalhost bool) *http.Server { icebergRouter := mux.NewRouter().SkipClean(true) @@ -625,12 +649,7 @@ func (s3opt *S3Options) startIcebergServer(s3ApiServer *s3api.S3ApiServer, liste // Create Iceberg server using the S3ApiServer as filer client icebergServer := iceberg.NewServer(s3ApiServer, s3ApiServer) icebergServer.SetCredentialValidator(s3ApiServer) - if s3opt.icebergCredentialRole != nil && *s3opt.icebergCredentialRole != "" { - duration := int64(0) - if s3opt.icebergCredentialDuration != nil { - duration = int64(*s3opt.icebergCredentialDuration) - } - s3ApiServer.SetIcebergCredentialRole(*s3opt.icebergCredentialRole, duration) + if s3opt.vendsCredentials() { icebergServer.SetCredentialVendor(icebergCredentialVendor{s3ApiServer}) } icebergServer.SetS3Endpoint(s3opt.deriveS3AdvertisedEndpoint()) @@ -666,22 +685,26 @@ func (s3opt *S3Options) startIcebergServer(s3ApiServer *s3api.S3ApiServer, liste return httpS } -// startLanceServer starts the Lance Namespace server on a separate port. It -// shares the Iceberg catalog's credential role: one deployment vends table -// credentials one way, whichever catalog the client speaks to. -func (s3opt *S3Options) startLanceServer(s3ApiServer *s3api.S3ApiServer, listenOnLocalhost bool) *http.Server { - lanceRouter := mux.NewRouter().SkipClean(true) - lanceRouter.Use(util_http.EscapeSemicolonsInQuery) - +// newLanceServer builds the Lance Namespace catalog. It shares the Iceberg +// catalog's credential role: one deployment vends table credentials one way, +// whichever catalog the client speaks to. +func (s3opt *S3Options) newLanceServer(s3ApiServer *s3api.S3ApiServer) *lance.Server { lanceServer := lance.NewServer(s3ApiServer, s3ApiServer) lanceServer.SetCredentialValidator(s3ApiServer) lanceServer.SetSigningKey(s3ApiServer.FilerSigningKey) - if s3opt.icebergCredentialRole != nil && *s3opt.icebergCredentialRole != "" { + if s3opt.vendsCredentials() { lanceServer.SetCredentialVendor(lanceCredentialVendor{s3ApiServer}) } lanceServer.SetS3Endpoint(s3opt.deriveLanceStorageEndpoint()) lanceServer.SetS3Region(s3tables.DefaultRegion) - lanceServer.RegisterRoutes(lanceRouter) + return lanceServer +} + +// startLanceServer serves the Lance Namespace catalog on its own port. +func (s3opt *S3Options) startLanceServer(listenOnLocalhost bool) *http.Server { + lanceRouter := mux.NewRouter().SkipClean(true) + lanceRouter.Use(util_http.EscapeSemicolonsInQuery) + s3opt.lanceServer.RegisterRoutes(lanceRouter) listenAddress := fmt.Sprintf("%s:%d", *s3opt.bindIp, *s3opt.portLance) lanceListener, lanceLocalListener, err := util.NewIpAndLocalListeners( @@ -719,12 +742,25 @@ func (s3opt *S3Options) vectorsRouter(s3ApiServer *s3api.S3ApiServer, grpcDialOp return nil } backend := &vectorDiscovery{masters: masterAddresses, grpcDialOption: grpcDialOption, discover: discoverVectorWorkers} + // In-process describe — the router resolves locations and vended + // storage options through the catalog directly, never over HTTP. The + // lookup itself runs under the trusted identity: the caller's vector + // permission already authorized the query, and catalog grants must not + // gate an internal resolution. + describe := func(ctx context.Context, id []string) (string, map[string]string, error) { + if s3opt.lanceServer == nil { + return "", nil, fmt.Errorf("lance catalog is not running") + } + r := (&http.Request{Method: http.MethodPost, Header: make(http.Header)}).WithContext(ctx) + return s3opt.lanceServer.DescribeTableInternal(r, id) + } vectorsRouter := mux.NewRouter().SkipClean(true) vectorsServer := s3vectors.NewServer(s3ApiServer, s3ApiServer, backend) vectorsServer.SetRegion(s3tables.DefaultRegion) vectorsServer.SetAccountID(s3tables.DefaultAccountID) vectorsServer.SetEncryptionChecker(s3api.GetSSES3KeyManager().IsConfigured) + vectorsServer.SetDescribe(describe) vectorsServer.RegisterRoutes(vectorsRouter) return vectorsRouter } diff --git a/weed/command/s3_aux_listener_test.go b/weed/command/s3_aux_listener_test.go index 7b0e1abf0b..b96c49e57a 100644 --- a/weed/command/s3_aux_listener_test.go +++ b/weed/command/s3_aux_listener_test.go @@ -1,6 +1,7 @@ package command import ( + "context" "net" "testing" @@ -49,7 +50,8 @@ func TestAuxListenerBindFailureIsNonFatal(t *testing.T) { if s3opt.icebergBound.Load() { t.Error("icebergBound must stay false when the listener failed to bind") } - if s := s3opt.startLanceServer(s3a, false); s != nil { + s3opt.lanceServer = s3opt.newLanceServer(s3a) + if s := s3opt.startLanceServer(false); s != nil { s.Close() t.Error("startLanceServer on an occupied port should return nil") } @@ -57,3 +59,32 @@ func TestAuxListenerBindFailureIsNonFatal(t *testing.T) { t.Error("lanceBound must stay false when the listener failed to bind") } } + +// -iceberg.credentialRole reaches the S3 server once, before either catalog +// is built, so a gateway that serves no Iceberg port vends too. +func TestCredentialRoleIsSetForBothCatalogs(t *testing.T) { + bind := "127.0.0.1" + empty := "" + zero := 0 + role := "arn:aws:iam::000000000000:role/vend" + s3opt := &S3Options{ + bindIp: &bind, + ip: &bind, + port: &zero, + portHttps: &zero, + portIceberg: &zero, + portLance: &zero, + icebergCredentialRole: &role, + externalUrl: &empty, + tlsPrivateKey: &empty, + } + s3a := &s3api.S3ApiServer{} + s3opt.setCredentialRole(s3a) + if !s3opt.vendsCredentials() { + t.Fatal("a role is configured but vending is reported off") + } + // With no role, vending is off and answers nil without checking anything. + if _, err := s3a.VendTableCredentials(context.Background(), "", "bkt", "vector/x", true); err == nil { + t.Fatal("vending is off: the credential role was not set") + } +} diff --git a/weed/pb/vector.proto b/weed/pb/vector.proto index 8bdaf3e086..31ed317330 100644 --- a/weed/pb/vector.proto +++ b/weed/pb/vector.proto @@ -27,11 +27,16 @@ enum DistanceMetric { } // Table addresses a Lance table through the namespace: the id parts as the -// namespace catalog joins them. +// namespace catalog joins them. The gateway resolves the table before it +// calls, so the engine opens it without a catalog round trip of its own. message Table { repeated string id = 1; // Dataset version to pin reads to; 0 means latest. uint64 version = 2; + // The table's Lance location, as the catalog describes it. + string location = 3; + // object_store options from describe_table (vended creds / endpoint). + map storage_options = 4; } message CreateIndexRequest { diff --git a/weed/pb/vector_pb/vector.pb.go b/weed/pb/vector_pb/vector.pb.go index 330a3484d2..263770c4aa 100644 --- a/weed/pb/vector_pb/vector.pb.go +++ b/weed/pb/vector_pb/vector.pb.go @@ -75,14 +75,19 @@ func (DistanceMetric) EnumDescriptor() ([]byte, []int) { } // Table addresses a Lance table through the namespace: the id parts as the -// namespace catalog joins them. +// namespace catalog joins them. The gateway resolves the table before it +// calls, so the engine opens it without a catalog round trip of its own. type Table struct { state protoimpl.MessageState `protogen:"open.v1"` Id []string `protobuf:"bytes,1,rep,name=id,proto3" json:"id,omitempty"` // Dataset version to pin reads to; 0 means latest. - Version uint64 `protobuf:"varint,2,opt,name=version,proto3" json:"version,omitempty"` - unknownFields protoimpl.UnknownFields - sizeCache protoimpl.SizeCache + Version uint64 `protobuf:"varint,2,opt,name=version,proto3" json:"version,omitempty"` + // The table's Lance location, as the catalog describes it. + Location string `protobuf:"bytes,3,opt,name=location,proto3" json:"location,omitempty"` + // object_store options from describe_table (vended creds / endpoint). + StorageOptions map[string]string `protobuf:"bytes,4,rep,name=storage_options,json=storageOptions,proto3" json:"storage_options,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache } func (x *Table) Reset() { @@ -129,6 +134,20 @@ func (x *Table) GetVersion() uint64 { return 0 } +func (x *Table) GetLocation() string { + if x != nil { + return x.Location + } + return "" +} + +func (x *Table) GetStorageOptions() map[string]string { + if x != nil { + return x.StorageOptions + } + return nil +} + type CreateIndexRequest struct { state protoimpl.MessageState `protogen:"open.v1"` Table *Table `protobuf:"bytes,1,opt,name=table,proto3" json:"table,omitempty"` @@ -1061,10 +1080,15 @@ var File_vector_proto protoreflect.FileDescriptor const file_vector_proto_rawDesc = "" + "\n" + - "\fvector.proto\x12\tvector_pb\x1a\x1cgoogle/protobuf/struct.proto\"1\n" + + "\fvector.proto\x12\tvector_pb\x1a\x1cgoogle/protobuf/struct.proto\"\xdf\x01\n" + "\x05Table\x12\x0e\n" + "\x02id\x18\x01 \x03(\tR\x02id\x12\x18\n" + - "\aversion\x18\x02 \x01(\x04R\aversion\"\xdf\x01\n" + + "\aversion\x18\x02 \x01(\x04R\aversion\x12\x1a\n" + + "\blocation\x18\x03 \x01(\tR\blocation\x12M\n" + + "\x0fstorage_options\x18\x04 \x03(\v2$.vector_pb.Table.StorageOptionsEntryR\x0estorageOptions\x1aA\n" + + "\x13StorageOptionsEntry\x12\x10\n" + + "\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" + + "\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\"\xdf\x01\n" + "\x12CreateIndexRequest\x12&\n" + "\x05table\x18\x01 \x01(\v2\x10.vector_pb.TableR\x05table\x12\x1c\n" + "\tdimension\x18\x02 \x01(\rR\tdimension\x12B\n" + @@ -1162,7 +1186,7 @@ func file_vector_proto_rawDescGZIP() []byte { } var file_vector_proto_enumTypes = make([]protoimpl.EnumInfo, 1) -var file_vector_proto_msgTypes = make([]protoimpl.MessageInfo, 17) +var file_vector_proto_msgTypes = make([]protoimpl.MessageInfo, 18) var file_vector_proto_goTypes = []any{ (DistanceMetric)(0), // 0: vector_pb.DistanceMetric (*Table)(nil), // 1: vector_pb.Table @@ -1182,42 +1206,44 @@ var file_vector_proto_goTypes = []any{ (*QueryVectorsRequest)(nil), // 15: vector_pb.QueryVectorsRequest (*QueryResult)(nil), // 16: vector_pb.QueryResult (*QueryVectorsResponse)(nil), // 17: vector_pb.QueryVectorsResponse - (*structpb.Struct)(nil), // 18: google.protobuf.Struct + nil, // 18: vector_pb.Table.StorageOptionsEntry + (*structpb.Struct)(nil), // 19: google.protobuf.Struct } var file_vector_proto_depIdxs = []int32{ - 1, // 0: vector_pb.CreateIndexRequest.table:type_name -> vector_pb.Table - 0, // 1: vector_pb.CreateIndexRequest.distance_metric:type_name -> vector_pb.DistanceMetric - 1, // 2: vector_pb.DropIndexRequest.table:type_name -> vector_pb.Table - 18, // 3: vector_pb.Vector.metadata:type_name -> google.protobuf.Struct - 1, // 4: vector_pb.PutVectorsRequest.table:type_name -> vector_pb.Table - 6, // 5: vector_pb.PutVectorsRequest.vectors:type_name -> vector_pb.Vector - 1, // 6: vector_pb.GetVectorsRequest.table:type_name -> vector_pb.Table - 6, // 7: vector_pb.GetVectorsResponse.vectors:type_name -> vector_pb.Vector - 1, // 8: vector_pb.DeleteVectorsRequest.table:type_name -> vector_pb.Table - 1, // 9: vector_pb.ListVectorsRequest.table:type_name -> vector_pb.Table - 6, // 10: vector_pb.ListVectorsResponse.vectors:type_name -> vector_pb.Vector - 1, // 11: vector_pb.QueryVectorsRequest.table:type_name -> vector_pb.Table - 18, // 12: vector_pb.QueryResult.metadata:type_name -> google.protobuf.Struct - 16, // 13: vector_pb.QueryVectorsResponse.vectors:type_name -> vector_pb.QueryResult - 2, // 14: vector_pb.VectorService.CreateIndex:input_type -> vector_pb.CreateIndexRequest - 4, // 15: vector_pb.VectorService.DropIndex:input_type -> vector_pb.DropIndexRequest - 7, // 16: vector_pb.VectorService.PutVectors:input_type -> vector_pb.PutVectorsRequest - 9, // 17: vector_pb.VectorService.GetVectors:input_type -> vector_pb.GetVectorsRequest - 11, // 18: vector_pb.VectorService.DeleteVectors:input_type -> vector_pb.DeleteVectorsRequest - 13, // 19: vector_pb.VectorService.ListVectors:input_type -> vector_pb.ListVectorsRequest - 15, // 20: vector_pb.VectorService.QueryVectors:input_type -> vector_pb.QueryVectorsRequest - 3, // 21: vector_pb.VectorService.CreateIndex:output_type -> vector_pb.CreateIndexResponse - 5, // 22: vector_pb.VectorService.DropIndex:output_type -> vector_pb.DropIndexResponse - 8, // 23: vector_pb.VectorService.PutVectors:output_type -> vector_pb.PutVectorsResponse - 10, // 24: vector_pb.VectorService.GetVectors:output_type -> vector_pb.GetVectorsResponse - 12, // 25: vector_pb.VectorService.DeleteVectors:output_type -> vector_pb.DeleteVectorsResponse - 14, // 26: vector_pb.VectorService.ListVectors:output_type -> vector_pb.ListVectorsResponse - 17, // 27: vector_pb.VectorService.QueryVectors:output_type -> vector_pb.QueryVectorsResponse - 21, // [21:28] is the sub-list for method output_type - 14, // [14:21] is the sub-list for method input_type - 14, // [14:14] is the sub-list for extension type_name - 14, // [14:14] is the sub-list for extension extendee - 0, // [0:14] is the sub-list for field type_name + 18, // 0: vector_pb.Table.storage_options:type_name -> vector_pb.Table.StorageOptionsEntry + 1, // 1: vector_pb.CreateIndexRequest.table:type_name -> vector_pb.Table + 0, // 2: vector_pb.CreateIndexRequest.distance_metric:type_name -> vector_pb.DistanceMetric + 1, // 3: vector_pb.DropIndexRequest.table:type_name -> vector_pb.Table + 19, // 4: vector_pb.Vector.metadata:type_name -> google.protobuf.Struct + 1, // 5: vector_pb.PutVectorsRequest.table:type_name -> vector_pb.Table + 6, // 6: vector_pb.PutVectorsRequest.vectors:type_name -> vector_pb.Vector + 1, // 7: vector_pb.GetVectorsRequest.table:type_name -> vector_pb.Table + 6, // 8: vector_pb.GetVectorsResponse.vectors:type_name -> vector_pb.Vector + 1, // 9: vector_pb.DeleteVectorsRequest.table:type_name -> vector_pb.Table + 1, // 10: vector_pb.ListVectorsRequest.table:type_name -> vector_pb.Table + 6, // 11: vector_pb.ListVectorsResponse.vectors:type_name -> vector_pb.Vector + 1, // 12: vector_pb.QueryVectorsRequest.table:type_name -> vector_pb.Table + 19, // 13: vector_pb.QueryResult.metadata:type_name -> google.protobuf.Struct + 16, // 14: vector_pb.QueryVectorsResponse.vectors:type_name -> vector_pb.QueryResult + 2, // 15: vector_pb.VectorService.CreateIndex:input_type -> vector_pb.CreateIndexRequest + 4, // 16: vector_pb.VectorService.DropIndex:input_type -> vector_pb.DropIndexRequest + 7, // 17: vector_pb.VectorService.PutVectors:input_type -> vector_pb.PutVectorsRequest + 9, // 18: vector_pb.VectorService.GetVectors:input_type -> vector_pb.GetVectorsRequest + 11, // 19: vector_pb.VectorService.DeleteVectors:input_type -> vector_pb.DeleteVectorsRequest + 13, // 20: vector_pb.VectorService.ListVectors:input_type -> vector_pb.ListVectorsRequest + 15, // 21: vector_pb.VectorService.QueryVectors:input_type -> vector_pb.QueryVectorsRequest + 3, // 22: vector_pb.VectorService.CreateIndex:output_type -> vector_pb.CreateIndexResponse + 5, // 23: vector_pb.VectorService.DropIndex:output_type -> vector_pb.DropIndexResponse + 8, // 24: vector_pb.VectorService.PutVectors:output_type -> vector_pb.PutVectorsResponse + 10, // 25: vector_pb.VectorService.GetVectors:output_type -> vector_pb.GetVectorsResponse + 12, // 26: vector_pb.VectorService.DeleteVectors:output_type -> vector_pb.DeleteVectorsResponse + 14, // 27: vector_pb.VectorService.ListVectors:output_type -> vector_pb.ListVectorsResponse + 17, // 28: vector_pb.VectorService.QueryVectors:output_type -> vector_pb.QueryVectorsResponse + 22, // [22:29] is the sub-list for method output_type + 15, // [15:22] is the sub-list for method input_type + 15, // [15:15] is the sub-list for extension type_name + 15, // [15:15] is the sub-list for extension extendee + 0, // [0:15] is the sub-list for field type_name } func init() { file_vector_proto_init() } @@ -1231,7 +1257,7 @@ func file_vector_proto_init() { GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: unsafe.Slice(unsafe.StringData(file_vector_proto_rawDesc), len(file_vector_proto_rawDesc)), NumEnums: 1, - NumMessages: 17, + NumMessages: 18, NumExtensions: 0, NumServices: 1, }, diff --git a/weed/s3api/lance/handlers_table.go b/weed/s3api/lance/handlers_table.go index c2027d2d32..11d59ca81f 100644 --- a/weed/s3api/lance/handlers_table.go +++ b/weed/s3api/lance/handlers_table.go @@ -179,6 +179,55 @@ func (s *Server) handleDeclareTable(w http.ResponseWriter, r *http.Request) { // handleDescribeTable resolves a table to a location, and to credentials when // the caller asks for them. +// DescribeTable is the in-process form of handleDescribeTable for trusted +// callers that already carry the request identity — no HTTP layer. +func (s *Server) DescribeTable(r *http.Request, parts []string) (string, map[string]string, error) { + return s.describeTable(r, parts, false) +} + +// DescribeTableInternal resolves a table like DescribeTable but runs the +// catalog lookup through the trusted manager: callers reach it through the +// vector coordinator, which only invokes it after the caller's own vector +// permission authorized the operation. A caller's catalog grants must not +// gate an internal resolution. Storage options are still vended for the +// caller's principal, and for a table of the vector namespace they may write +// it: the engine is that namespace's writer, and the operation it carries +// out was authorized by the caller's vector permission, not by s3tables +// grants the caller need not hold. +func (s *Server) DescribeTableInternal(r *http.Request, parts []string) (string, map[string]string, error) { + return s.describeTable(r, parts, true) +} + +func (s *Server) describeTable(r *http.Request, parts []string, internal bool) (string, map[string]string, error) { + bucket, ns, name, err := identifier(parts).table() + if err != nil { + return "", nil, err + } + var table *lanceTable + if internal { + table, err = s.loadLanceTableInternal(r, bucket, ns, name) + } else { + table, err = s.loadLanceTable(r, bucket, ns, name) + } + if err != nil { + return "", nil, err + } + location := table.location + deregistered, _, err := s.datasetState(r, location) + if err != nil { + return "", nil, err + } + if deregistered { + return "", nil, &s3tables.S3TablesError{Type: s3tables.ErrCodeNoSuchTable, Message: "table is deregistered"} + } + canWrite := func() bool { return s.mayUpdateTable(r, bucket, ns, name) } + if internal && len(ns) == 1 && ns[0] == s3tables.VectorIndexNamespace { + canWrite = func() bool { return true } + } + options, err := s.storageOptions(r, bucket, location, true, canWrite) + return location, options, err +} + func (s *Server) handleDescribeTable(w http.ResponseWriter, r *http.Request) { id, _, ok := routeIdentifier(w, r) if !ok { @@ -534,13 +583,23 @@ type lanceTable struct { // loadLanceTable reads a table and refuses one that is not a Lance table, so a // Lance client never resolves an Iceberg table's location and writes over it. func (s *Server) loadLanceTable(r *http.Request, bucket string, ns []string, name string) (*lanceTable, error) { + return s.loadLanceTableVia(r, s.execute, bucket, ns, name) +} + +// loadLanceTableInternal is loadLanceTable through the trusted manager, for +// resolutions the caller's catalog grants must not gate. +func (s *Server) loadLanceTableInternal(r *http.Request, bucket string, ns []string, name string) (*lanceTable, error) { + return s.loadLanceTableVia(r, s.executeInternal, bucket, ns, name) +} + +func (s *Server) loadLanceTableVia(r *http.Request, exec func(*http.Request, string, interface{}, interface{}) error, bucket string, ns []string, name string) (*lanceTable, error) { req := &s3tables.GetTableRequest{ TableBucketARN: bucketARN(bucket), Namespace: ns, Name: name, } var resp s3tables.GetTableResponse - if err := s.execute(r, "GetTable", req, &resp); err != nil { + if err := exec(r, "GetTable", req, &resp); err != nil { return nil, err } if resp.Format != s3tables.FormatLance { diff --git a/weed/s3api/lance/handlers_test.go b/weed/s3api/lance/handlers_test.go index 037f66ff83..23ea7b610f 100644 --- a/weed/s3api/lance/handlers_test.go +++ b/weed/s3api/lance/handlers_test.go @@ -2,6 +2,7 @@ package lance import ( "encoding/json" + "errors" "net/http" "net/http/httptest" "strings" @@ -9,6 +10,7 @@ import ( "time" "github.com/gorilla/mux" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" "github.com/seaweedfs/seaweedfs/weed/s3api/s3err" "github.com/seaweedfs/seaweedfs/weed/s3api/s3tables" "github.com/seaweedfs/seaweedfs/weed/s3api/s3tables/s3tablestest" @@ -490,3 +492,64 @@ func TestDeclareVendsWriteOnlyForATableItCreates(t *testing.T) { t.Error("a declare that may not write the table brought it back") } } + +// An in-process describe for a trusted caller resolves the table without the +// caller's catalog grants, and still refuses a deregistered one. +func TestDescribeTableInternalIgnoresCatalogGrants(t *testing.T) { + h := newTestHarness(t) + h.createBucket(t, "lnc", s3tables.FormatLance) + h.mustDo(t, http.MethodPost, "/v1/namespace/lnc$ns/create", `{}`, http.StatusOK) + h.mustDo(t, http.MethodPost, "/v1/table/lnc$ns$t/declare", `{}`, http.StatusOK) + + server := NewServer(s3tables.NewManagerClient(h.filer.Client), createOnlyAuthenticator{}) + server.SetS3Endpoint("http://127.0.0.1:8333") + anonymous := httptest.NewRequest(http.MethodPost, "/", nil) + anonymous = anonymous.WithContext(s3_constants.SetIdentityNameInContext(anonymous.Context(), "stranger")) + if _, _, err := server.DescribeTable(anonymous, []string{"lnc", "ns", "t"}); err == nil { + t.Fatal("DescribeTable without grants resolved the table") + } + location, options, err := server.DescribeTableInternal(anonymous, []string{"lnc", "ns", "t"}) + if err != nil { + t.Fatalf("DescribeTableInternal: %v", err) + } + if location != "s3://lnc/ns/t" || options["aws_endpoint"] != "http://127.0.0.1:8333" { + t.Fatalf("resolved %q %v", location, options) + } + + h.mustDo(t, http.MethodPost, "/v1/table/lnc$ns$t/deregister", `{}`, http.StatusOK) + _, _, err = server.DescribeTableInternal(anonymous, []string{"lnc", "ns", "t"}) + var apiErr *s3tables.S3TablesError + if !errors.As(err, &apiErr) || apiErr.Type != s3tables.ErrCodeNoSuchTable { + t.Fatalf("a deregistered table resolved: %v, want %s", err, s3tables.ErrCodeNoSuchTable) + } +} + +// The options an internal describe vends may write a vector index, whose +// writer is the engine, but not a table whose writes s3tables grants gate. +func TestDescribeTableInternalVendsWriteForTheVectorNamespace(t *testing.T) { + h := newTestHarness(t) + h.seedVectorIndex(t, "vec", s3tables.FormatLance) + h.createBucket(t, "lnc", s3tables.FormatLance) + h.mustDo(t, http.MethodPost, "/v1/namespace/lnc$ns/create", `{}`, http.StatusOK) + h.mustDo(t, http.MethodPost, "/v1/table/lnc$ns$t/declare", `{}`, http.StatusOK) + + server := NewServer(s3tables.NewManagerClient(h.filer.Client), createOnlyAuthenticator{}) + server.SetCredentialVendor(recordingVendor{}) + stranger := httptest.NewRequest(http.MethodPost, "/", nil) + stranger = stranger.WithContext(s3_constants.SetIdentityNameInContext(stranger.Context(), "stranger")) + for _, tc := range []struct { + id []string + want string + }{ + {[]string{"vec", "vector", "idx"}, "read-write"}, + {[]string{"lnc", "ns", "t"}, "read-only"}, + } { + _, options, err := server.DescribeTableInternal(stranger, tc.id) + if err != nil { + t.Fatalf("%v: %v", tc.id, err) + } + if got := options["aws_access_key_id"]; got != tc.want { + t.Errorf("%v: vended %q, want %q", tc.id, got, tc.want) + } + } +} diff --git a/weed/s3api/lance/server.go b/weed/s3api/lance/server.go index 286205a70c..9dd33c99a1 100644 --- a/weed/s3api/lance/server.go +++ b/weed/s3api/lance/server.go @@ -61,6 +61,7 @@ type CredentialVendor interface { type Server struct { filerClient FilerClient tablesManager *s3tables.Manager + internal *s3tables.Manager authenticator S3Authenticator credentialValidator CredentialValidator credentialVendor CredentialVendor @@ -77,9 +78,12 @@ func NewServer(filerClient FilerClient, authenticator S3Authenticator) *Server { if authenticator != nil { manager.SetDefaultAllow(authenticator.DefaultAllow()) } + internal := s3tables.NewManager() + internal.SetTrusted(true) return &Server{ filerClient: filerClient, tablesManager: manager, + internal: internal, authenticator: authenticator, } } diff --git a/weed/s3api/lance/storage.go b/weed/s3api/lance/storage.go index e719680118..fb5bfaa50f 100644 --- a/weed/s3api/lance/storage.go +++ b/weed/s3api/lance/storage.go @@ -25,6 +25,15 @@ func (s *Server) execute(r *http.Request, operation string, req, resp interface{ }) } +// executeInternal runs one S3 Tables operation through the trusted manager — +// the caller's catalog grants must not gate an internal resolution that the +// caller's own vector permission already authorized. +func (s *Server) executeInternal(r *http.Request, operation string, req, resp interface{}) error { + return s.filerClient.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + return s.internal.Execute(r.Context(), s3tables.NewManagerClient(client), operation, req, resp, "") + }) +} + func bucketARN(bucket string) string { arn, _ := s3tables.BuildBucketARN(s3tables.DefaultRegion, s3_constants.AccountAdminId, bucket) return arn diff --git a/weed/s3api/s3vectors/describe.go b/weed/s3api/s3vectors/describe.go new file mode 100644 index 0000000000..b54d4657db --- /dev/null +++ b/weed/s3api/s3vectors/describe.go @@ -0,0 +1,108 @@ +package s3vectors + +import ( + "context" + "errors" + + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + + "github.com/seaweedfs/seaweedfs/weed/pb/vector_pb" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3tables" +) + +// DescribeFn resolves a table id to its Lance location and the storage +// options engines need to open it — the gateway supplies the catalog's +// in-process describe, so no unauthenticated HTTP call is needed. +type DescribeFn func(ctx context.Context, id []string) (string, map[string]string, error) + +// SetDescribe resolves every table this server sends to the engine, so the +// engine opens it without a catalog round trip of its own. +func (s *Server) SetDescribe(describe DescribeFn) { + s.vectors = &describedClient{VectorServiceClient: s.vectors, describe: describe} +} + +// describedClient fills in the table's location and storage options before a +// request reaches the engine. +type describedClient struct { + vector_pb.VectorServiceClient + describe DescribeFn +} + +func (c *describedClient) resolve(ctx context.Context, table *vector_pb.Table) error { + if table == nil || table.Location != "" { + return nil + } + location, options, err := c.describe(ctx, table.Id) + if err != nil { + return describeErr(err) + } + table.Location, table.StorageOptions = location, options + return nil +} + +// describeErr gives a catalog failure the gRPC code the engine would have +// answered with, so vectorErr maps it the same way. +func describeErr(err error) error { + var apiErr *s3tables.S3TablesError + if !errors.As(err, &apiErr) { + return status.Error(codes.Internal, err.Error()) + } + switch apiErr.Type { + case s3tables.ErrCodeNoSuchTable, s3tables.ErrCodeNoSuchNamespace, s3tables.ErrCodeNoSuchBucket: + return status.Error(codes.NotFound, apiErr.Message) + case s3tables.ErrCodeAccessDenied: + return status.Error(codes.PermissionDenied, apiErr.Message) + } + return status.Error(codes.Internal, apiErr.Message) +} + +func (c *describedClient) CreateIndex(ctx context.Context, in *vector_pb.CreateIndexRequest, opts ...grpc.CallOption) (*vector_pb.CreateIndexResponse, error) { + if err := c.resolve(ctx, in.Table); err != nil { + return nil, err + } + return c.VectorServiceClient.CreateIndex(ctx, in, opts...) +} + +func (c *describedClient) DropIndex(ctx context.Context, in *vector_pb.DropIndexRequest, opts ...grpc.CallOption) (*vector_pb.DropIndexResponse, error) { + if err := c.resolve(ctx, in.Table); err != nil { + return nil, err + } + return c.VectorServiceClient.DropIndex(ctx, in, opts...) +} + +func (c *describedClient) PutVectors(ctx context.Context, in *vector_pb.PutVectorsRequest, opts ...grpc.CallOption) (*vector_pb.PutVectorsResponse, error) { + if err := c.resolve(ctx, in.Table); err != nil { + return nil, err + } + return c.VectorServiceClient.PutVectors(ctx, in, opts...) +} + +func (c *describedClient) GetVectors(ctx context.Context, in *vector_pb.GetVectorsRequest, opts ...grpc.CallOption) (*vector_pb.GetVectorsResponse, error) { + if err := c.resolve(ctx, in.Table); err != nil { + return nil, err + } + return c.VectorServiceClient.GetVectors(ctx, in, opts...) +} + +func (c *describedClient) DeleteVectors(ctx context.Context, in *vector_pb.DeleteVectorsRequest, opts ...grpc.CallOption) (*vector_pb.DeleteVectorsResponse, error) { + if err := c.resolve(ctx, in.Table); err != nil { + return nil, err + } + return c.VectorServiceClient.DeleteVectors(ctx, in, opts...) +} + +func (c *describedClient) ListVectors(ctx context.Context, in *vector_pb.ListVectorsRequest, opts ...grpc.CallOption) (*vector_pb.ListVectorsResponse, error) { + if err := c.resolve(ctx, in.Table); err != nil { + return nil, err + } + return c.VectorServiceClient.ListVectors(ctx, in, opts...) +} + +func (c *describedClient) QueryVectors(ctx context.Context, in *vector_pb.QueryVectorsRequest, opts ...grpc.CallOption) (*vector_pb.QueryVectorsResponse, error) { + if err := c.resolve(ctx, in.Table); err != nil { + return nil, err + } + return c.VectorServiceClient.QueryVectors(ctx, in, opts...) +} diff --git a/weed/s3api/s3vectors/describe_test.go b/weed/s3api/s3vectors/describe_test.go new file mode 100644 index 0000000000..2a94e4bb80 --- /dev/null +++ b/weed/s3api/s3vectors/describe_test.go @@ -0,0 +1,96 @@ +package s3vectors + +import ( + "context" + "fmt" + "net/http" + "strings" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/s3api/s3tables" + "github.com/seaweedfs/seaweedfs/weed/s3api/s3tables/s3tablestest" +) + +func TestDescribeFillsTableLocation(t *testing.T) { + fs := s3tablestest.Start(t) + engine := &fakeVectors{} + s := NewServer(s3tables.NewManagerClient(fs.Client), allowAllAuth{}, engine) + var described [][]string + s.SetDescribe(func(_ context.Context, id []string) (string, map[string]string, error) { + described = append(described, id) + return "s3://" + strings.Join(id, "/"), map[string]string{"aws_endpoint": "http://gateway:8333"}, nil + }) + + if w := postJSON(t, s.handleCreateVectorBucket, &CreateVectorBucketRequest{VectorBucketName: "vec-bucket"}); w.Code != http.StatusOK { + t.Fatalf("create bucket: %d %s", w.Code, w.Body.String()) + } + if w := postJSON(t, s.handleCreateIndex, &CreateIndexRequest{ + VectorBucketName: "vec-bucket", + IndexName: "idx", + DataType: "float32", + Dimension: 3, + DistanceMetric: "euclidean", + }); w.Code != http.StatusOK { + t.Fatalf("create index: %d %s", w.Code, w.Body.String()) + } + if w := postJSON(t, s.handlePutVectors, &PutVectorsRequest{ + VectorBucketName: "vec-bucket", + IndexName: "idx", + Vectors: []*PutInputVector{{Key: "a", Data: &VectorData{Float32: []float32{1, 2, 3}}}}, + }); w.Code != http.StatusOK { + t.Fatalf("put vectors: %d %s", w.Code, w.Body.String()) + } + + if len(engine.created) != 1 || len(engine.puts) != 1 { + t.Fatalf("engine calls: %d creates, %d puts", len(engine.created), len(engine.puts)) + } + for _, table := range []interface{ GetLocation() string }{engine.created[0].Table, engine.puts[0].Table} { + if table.GetLocation() != "s3://vec-bucket/vector/idx" { + t.Fatalf("table location = %q", table.GetLocation()) + } + } + if got := engine.puts[0].Table.StorageOptions["aws_endpoint"]; got != "http://gateway:8333" { + t.Fatalf("storage options = %v", engine.puts[0].Table.StorageOptions) + } + if len(described) != 2 || strings.Join(described[1], "/") != "vec-bucket/vector/idx" { + t.Fatalf("described ids = %v", described) + } +} + +func TestDescribeErrorsReachTheCaller(t *testing.T) { + fs := s3tablestest.Start(t) + cases := []struct { + err error + code int + body string + }{ + {&s3tables.S3TablesError{Type: s3tables.ErrCodeNoSuchTable, Message: "gone"}, http.StatusNotFound, "NotFoundException"}, + {&s3tables.S3TablesError{Type: s3tables.ErrCodeAccessDenied, Message: "no"}, http.StatusForbidden, "AccessDeniedException"}, + {fmt.Errorf("lance catalog is not running"), http.StatusInternalServerError, "InternalFailure"}, + } + engine := &fakeVectors{} + s := NewServer(s3tables.NewManagerClient(fs.Client), allowAllAuth{}, engine) + var describeErr error + s.SetDescribe(func(context.Context, []string) (string, map[string]string, error) { + return "", nil, describeErr + }) + if w := postJSON(t, s.handleCreateVectorBucket, &CreateVectorBucketRequest{VectorBucketName: "vec-bucket"}); w.Code != http.StatusOK { + t.Fatalf("create bucket: %d %s", w.Code, w.Body.String()) + } + for _, tc := range cases { + describeErr = tc.err + w := postJSON(t, s.handleCreateIndex, &CreateIndexRequest{ + VectorBucketName: "vec-bucket", + IndexName: "idx", + DataType: "float32", + Dimension: 3, + DistanceMetric: "euclidean", + }) + if w.Code != tc.code || !strings.Contains(w.Body.String(), tc.body) { + t.Fatalf("%v: %d %s, want %d %s", tc.err, w.Code, w.Body.String(), tc.code, tc.body) + } + if len(engine.created) != 0 { + t.Fatalf("%v: the engine was called", tc.err) + } + } +}