diff options
author | Yureka <tvl@yuka.dev> | 2024-07-18T17·09+0200 |
---|---|---|
committer | yuka <tvl@yuka.dev> | 2024-07-18T19·19+0000 |
commit | 168e4fda5909e535f33051051ef426e221ef20d4 (patch) | |
tree | e23b8ad4ced3f4232bdb0ad186f3b63f693c57e5 /tvix/castore/src/directoryservice | |
parent | 79317be214ce2f1e3347438319d3482bb773a649 (diff) |
refactor(tvix): use composition & registry for from_addr r/8368
Change-Id: I3c94ecb5958294b5973c6fcdf5ee9c0d37fa54ad Reviewed-on: https://cl.tvl.fyi/c/depot/+/11976 Reviewed-by: flokli <flokli@flokli.de> Tested-by: BuildkiteCI Autosubmit: yuka <yuka@yuka.dev>
Diffstat (limited to 'tvix/castore/src/directoryservice')
-rw-r--r-- | tvix/castore/src/directoryservice/bigtable.rs | 20 | ||||
-rw-r--r-- | tvix/castore/src/directoryservice/combinators.rs | 10 | ||||
-rw-r--r-- | tvix/castore/src/directoryservice/from_addr.rs | 113 | ||||
-rw-r--r-- | tvix/castore/src/directoryservice/grpc.rs | 13 | ||||
-rw-r--r-- | tvix/castore/src/directoryservice/memory.rs | 11 | ||||
-rw-r--r-- | tvix/castore/src/directoryservice/object_store.rs | 26 | ||||
-rw-r--r-- | tvix/castore/src/directoryservice/sled.rs | 25 |
7 files changed, 123 insertions, 95 deletions
diff --git a/tvix/castore/src/directoryservice/bigtable.rs b/tvix/castore/src/directoryservice/bigtable.rs index 69edc05d787d..596094930614 100644 --- a/tvix/castore/src/directoryservice/bigtable.rs +++ b/tvix/castore/src/directoryservice/bigtable.rs @@ -361,6 +361,26 @@ impl ServiceBuilder for BigtableParameters { } } +impl TryFrom<url::Url> for BigtableParameters { + type Error = Box<dyn std::error::Error + Send + Sync>; + fn try_from(mut url: url::Url) -> Result<Self, Self::Error> { + // parse the instance name from the hostname. + let instance_name = url + .host_str() + .ok_or_else(|| Error::StorageError("instance name missing".into()))? + .to_string(); + + // … but add it to the query string now, so we just need to parse that. + url.query_pairs_mut() + .append_pair("instance_name", &instance_name); + + let params: BigtableParameters = serde_qs::from_str(url.query().unwrap_or_default()) + .map_err(|e| Error::InvalidRequest(format!("failed to parse parameters: {}", e)))?; + + Ok(params) + } +} + fn default_app_profile_id() -> String { "default".to_owned() } diff --git a/tvix/castore/src/directoryservice/combinators.rs b/tvix/castore/src/directoryservice/combinators.rs index 2364a313d5f4..74d02f1ad2b9 100644 --- a/tvix/castore/src/directoryservice/combinators.rs +++ b/tvix/castore/src/directoryservice/combinators.rs @@ -151,6 +151,16 @@ pub struct CacheConfig { far: String, } +impl TryFrom<url::Url> for CacheConfig { + type Error = Box<dyn std::error::Error + Send + Sync>; + fn try_from(_url: url::Url) -> Result<Self, Self::Error> { + Err(Error::StorageError( + "Instantiating a CombinedDirectoryService from a url is not supported".into(), + ) + .into()) + } +} + #[async_trait] impl ServiceBuilder for CacheConfig { type Output = dyn DirectoryService; diff --git a/tvix/castore/src/directoryservice/from_addr.rs b/tvix/castore/src/directoryservice/from_addr.rs index 999170dcd13f..bc63f129fe9e 100644 --- a/tvix/castore/src/directoryservice/from_addr.rs +++ b/tvix/castore/src/directoryservice/from_addr.rs @@ -1,12 +1,13 @@ -use url::Url; +use std::sync::Arc; -use crate::{proto::directory_service_client::DirectoryServiceClient, Error}; +use url::Url; -use super::{ - DirectoryService, GRPCDirectoryService, MemoryDirectoryService, ObjectStoreDirectoryService, - SledDirectoryService, +use crate::composition::{ + with_registry, CompositionContext, DeserializeWithRegistry, ServiceBuilder, REG, }; +use super::DirectoryService; + /// Constructs a new instance of a [DirectoryService] from an URI. /// /// The following URIs are supported: @@ -21,101 +22,23 @@ use super::{ /// Connects to a local tvix-store gRPC service via Unix socket. /// - `grpc+http://host:port`, `grpc+https://host:port` /// Connects to a (remote) tvix-store gRPC service. -pub async fn from_addr(uri: &str) -> Result<Box<dyn DirectoryService>, crate::Error> { +pub async fn from_addr( + uri: &str, +) -> Result<Arc<dyn DirectoryService>, Box<dyn std::error::Error + Send + Sync>> { #[allow(unused_mut)] let mut url = Url::parse(uri) .map_err(|e| crate::Error::StorageError(format!("unable to parse url: {}", e)))?; - let directory_service: Box<dyn DirectoryService> = match url.scheme() { - "memory" => { - // memory doesn't support host or path in the URL. - if url.has_host() || !url.path().is_empty() { - return Err(Error::StorageError("invalid url".to_string())); - } - Box::<MemoryDirectoryService>::default() - } - "sled" => { - // sled doesn't support host, and a path can be provided (otherwise - // it'll live in memory only). - if url.has_host() { - return Err(Error::StorageError("no host allowed".to_string())); - } - - if url.path() == "/" { - return Err(Error::StorageError( - "cowardly refusing to open / with sled".to_string(), - )); - } - - // TODO: expose compression and other parameters as URL parameters? - - Box::new(if url.path().is_empty() { - SledDirectoryService::new_temporary() - .map_err(|e| Error::StorageError(e.to_string()))? - } else { - SledDirectoryService::new(url.path()) - .map_err(|e| Error::StorageError(e.to_string()))? - }) - } - scheme if scheme.starts_with("grpc+") => { - // schemes starting with grpc+ go to the GRPCPathInfoService. - // That's normally grpc+unix for unix sockets, and grpc+http(s) for the HTTP counterparts. - // - In the case of unix sockets, there must be a path, but may not be a host. - // - In the case of non-unix sockets, there must be a host, but no path. - // Constructing the channel is handled by tvix_castore::channel::from_url. - Box::new(GRPCDirectoryService::from_client( - DirectoryServiceClient::with_interceptor( - crate::tonic::channel_from_url(&url).await?, - tvix_tracing::propagate::tonic::send_trace, - ), - )) - } - scheme if scheme.starts_with("objectstore+") => { - // We need to convert the URL to string, strip the prefix there, and then - // parse it back as url, as Url::set_scheme() rejects some of the transitions we want to do. - let trimmed_url = { - let s = url.to_string(); - let mut url = Url::parse(s.strip_prefix("objectstore+").unwrap()).unwrap(); - // trim the query pairs, they might contain credentials or local settings we don't want to send as-is. - url.set_query(None); - url - }; - Box::new( - ObjectStoreDirectoryService::parse_url_opts(&trimmed_url, url.query_pairs()) - .map_err(|e| Error::StorageError(e.to_string()))?, - ) - } - #[cfg(feature = "cloud")] - "bigtable" => { - use super::bigtable::BigtableParameters; - use super::BigtableDirectoryService; - - // parse the instance name from the hostname. - let instance_name = url - .host_str() - .ok_or_else(|| Error::StorageError("instance name missing".into()))? - .to_string(); - - // … but add it to the query string now, so we just need to parse that. - url.query_pairs_mut() - .append_pair("instance_name", &instance_name); - - let params: BigtableParameters = serde_qs::from_str(url.query().unwrap_or_default()) - .map_err(|e| Error::InvalidRequest(format!("failed to parse parameters: {}", e)))?; + let directory_service_config = with_registry(®, || { + <DeserializeWithRegistry<Box<dyn ServiceBuilder<Output = dyn DirectoryService>>>>::try_from( + url, + ) + })? + .0; + let directory_service = directory_service_config + .build("anonymous", &CompositionContext::blank()) + .await?; - Box::new( - BigtableDirectoryService::connect(params) - .await - .map_err(|e| Error::StorageError(e.to_string()))?, - ) - } - _ => { - return Err(crate::Error::StorageError(format!( - "unknown scheme: {}", - url.scheme() - ))) - } - }; Ok(directory_service) } diff --git a/tvix/castore/src/directoryservice/grpc.rs b/tvix/castore/src/directoryservice/grpc.rs index e2ca3954c112..415796fa52cc 100644 --- a/tvix/castore/src/directoryservice/grpc.rs +++ b/tvix/castore/src/directoryservice/grpc.rs @@ -224,6 +224,19 @@ pub struct GRPCDirectoryServiceConfig { url: String, } +impl TryFrom<url::Url> for GRPCDirectoryServiceConfig { + type Error = Box<dyn std::error::Error + Send + Sync>; + fn try_from(url: url::Url) -> Result<Self, Self::Error> { + // This is normally grpc+unix for unix sockets, and grpc+http(s) for the HTTP counterparts. + // - In the case of unix sockets, there must be a path, but may not be a host. + // - In the case of non-unix sockets, there must be a host, but no path. + // Constructing the channel is handled by tvix_castore::channel::from_url. + Ok(GRPCDirectoryServiceConfig { + url: url.to_string(), + }) + } +} + #[async_trait] impl ServiceBuilder for GRPCDirectoryServiceConfig { type Output = dyn DirectoryService; diff --git a/tvix/castore/src/directoryservice/memory.rs b/tvix/castore/src/directoryservice/memory.rs index f12ef6e977d0..c1fc361f0d59 100644 --- a/tvix/castore/src/directoryservice/memory.rs +++ b/tvix/castore/src/directoryservice/memory.rs @@ -91,6 +91,17 @@ impl DirectoryService for MemoryDirectoryService { #[serde(deny_unknown_fields)] pub struct MemoryDirectoryServiceConfig {} +impl TryFrom<url::Url> for MemoryDirectoryServiceConfig { + type Error = Box<dyn std::error::Error + Send + Sync>; + fn try_from(url: url::Url) -> Result<Self, Self::Error> { + // memory doesn't support host or path in the URL. + if url.has_host() || !url.path().is_empty() { + return Err(Error::StorageError("invalid url".to_string()).into()); + } + Ok(MemoryDirectoryServiceConfig {}) + } +} + #[async_trait] impl ServiceBuilder for MemoryDirectoryServiceConfig { type Output = dyn DirectoryService; diff --git a/tvix/castore/src/directoryservice/object_store.rs b/tvix/castore/src/directoryservice/object_store.rs index 1977de18fbec..0f0423a49e5b 100644 --- a/tvix/castore/src/directoryservice/object_store.rs +++ b/tvix/castore/src/directoryservice/object_store.rs @@ -179,6 +179,32 @@ pub struct ObjectStoreDirectoryServiceConfig { object_store_options: HashMap<String, String>, } +impl TryFrom<url::Url> for ObjectStoreDirectoryServiceConfig { + type Error = Box<dyn std::error::Error + Send + Sync>; + fn try_from(url: url::Url) -> Result<Self, Self::Error> { + // We need to convert the URL to string, strip the prefix there, and then + // parse it back as url, as Url::set_scheme() rejects some of the transitions we want to do. + let trimmed_url = { + let s = url.to_string(); + let mut url = Url::parse( + s.strip_prefix("objectstore+") + .ok_or(Error::StorageError("Missing objectstore uri".into()))?, + )?; + // trim the query pairs, they might contain credentials or local settings we don't want to send as-is. + url.set_query(None); + url + }; + Ok(ObjectStoreDirectoryServiceConfig { + object_store_url: trimmed_url.into(), + object_store_options: url + .query_pairs() + .into_iter() + .map(|(k, v)| (k.to_string(), v.to_string())) + .collect(), + }) + } +} + #[async_trait] impl ServiceBuilder for ObjectStoreDirectoryServiceConfig { type Output = dyn DirectoryService; diff --git a/tvix/castore/src/directoryservice/sled.rs b/tvix/castore/src/directoryservice/sled.rs index 8e74227b3e62..61058b392bb3 100644 --- a/tvix/castore/src/directoryservice/sled.rs +++ b/tvix/castore/src/directoryservice/sled.rs @@ -145,6 +145,31 @@ pub struct SledDirectoryServiceConfig { path: Option<String>, } +impl TryFrom<url::Url> for SledDirectoryServiceConfig { + type Error = Box<dyn std::error::Error + Send + Sync>; + fn try_from(url: url::Url) -> Result<Self, Self::Error> { + // sled doesn't support host, and a path can be provided (otherwise + // it'll live in memory only). + if url.has_host() { + return Err(Error::StorageError("no host allowed".to_string()).into()); + } + + // TODO: expose compression and other parameters as URL parameters? + + Ok(if url.path().is_empty() { + SledDirectoryServiceConfig { + is_temporary: true, + path: None, + } + } else { + SledDirectoryServiceConfig { + is_temporary: false, + path: Some(url.path().to_string()), + } + }) + } +} + #[async_trait] impl ServiceBuilder for SledDirectoryServiceConfig { type Output = dyn DirectoryService; |