Treat logs as remote file backends

Resolves #43.

Author: kezhuw

src/bin/seamdbd.rs

@@ -20,7 +20,6 @@ use clap::Parser;
 use pgwire::tokio::process_socket;
 use seamdb::cluster::{ClusterEnv, EtcdClusterMetaDaemon, EtcdNodeRegistry, NodeId};
 use seamdb::endpoint::{Endpoint, ServiceUri};
-use seamdb::fs::MemoryFileSystemFactory;
 use seamdb::log::{KafkaLogFactory, LogManager, MemoryLogFactory};
 use seamdb::protos::TableDescriptor;
 use seamdb::sql::postgresql::PostgresqlHandlerFactory;
@@ -85,7 +84,7 @@ async fn main() {
     let (nodes, lease) =
         EtcdNodeRegistry::join(cluster_uri.clone(), node_id.clone(), Some(endpoint.to_owned())).await.unwrap();
     let log_manager = new_log_manager(log_uri).await.unwrap();
-    let cluster_env = ClusterEnv::new(log_manager.into(), MemoryFileSystemFactory.into(), nodes).with_replicas(1);
+    let cluster_env = ClusterEnv::new(log_manager.into(), nodes).with_replicas(1);
     let mut cluster_meta_handle =
         EtcdClusterMetaDaemon::start(args.cluster_name, cluster_uri.clone(), cluster_env.clone()).await.unwrap();
     let descriptor_watcher = cluster_meta_handle.watch_descriptor(None).await.unwrap();

src/cluster/env.rs

@@ -35,7 +35,7 @@ pub struct ClusterEnv {
 }
 
 impl ClusterEnv {
-    pub fn new(log: Arc<LogManager>, fs: Arc<FileSystemManager>, nodes: Arc<dyn NodeRegistry>) -> Self {
+    pub fn with_filesystem(log: Arc<LogManager>, fs: Arc<FileSystemManager>, nodes: Arc<dyn NodeRegistry>) -> Self {
         Self {
             log,
             fs,
@@ -48,6 +48,11 @@ impl ClusterEnv {
         }
     }
 
+    pub fn new(log: Arc<LogManager>, nodes: Arc<dyn NodeRegistry>) -> Self {
+        let fs = Arc::new(FileSystemManager::from(log.clone()));
+        Self::with_filesystem(log, fs, nodes)
+    }
+
     pub fn with_replicas(self, replicas: usize) -> Self {
         Self { replicas: replicas.max(1), ..self }
     }

src/cluster/meta.rs

@@ -840,7 +840,6 @@ mod tests {
     use crate::cluster::etcd::EtcdHelper;
     use crate::cluster::{EtcdNodeRegistry, *};
     use crate::endpoint::{Endpoint, ServiceUri};
-    use crate::fs::MemoryFileSystemFactory;
     use crate::log::tests::*;
     use crate::log::LogRegistry;
     use crate::protos::{
@@ -1029,7 +1028,7 @@ mod tests {
 
         let (nodes, _cluster_lease) =
             EtcdNodeRegistry::join(cluster_uri.clone(), NodeId::new_random(), None).await.unwrap();
-        let cluster_env = ClusterEnv::new(log_manager.clone(), MemoryFileSystemFactory.into(), nodes).with_replicas(2);
+        let cluster_env = ClusterEnv::new(log_manager.clone(), nodes).with_replicas(2);
 
         let mut cluster_meta_handle =
             EtcdClusterMetaDaemon::start("seamdb1", cluster_uri.clone(), cluster_env).await.unwrap();
@@ -1095,7 +1094,7 @@ mod tests {
 
         let (nodes2, _cluster_lease2) =
             EtcdNodeRegistry::join(cluster_uri.clone(), NodeId::new_random(), None).await.unwrap();
-        let cluster_env2 = ClusterEnv::new(log_manager, MemoryFileSystemFactory.into(), nodes2).with_replicas(2);
+        let cluster_env2 = ClusterEnv::new(log_manager, nodes2).with_replicas(2);
         let mut cluster_meta_handle2 =
             EtcdClusterMetaDaemon::start("seamdb1", cluster_uri.clone(), cluster_env2).await.unwrap();
 

src/fs/mod.rs

@@ -29,6 +29,8 @@ use thiserror::Error;
 
 pub use self::memory::MemoryFileSystemFactory;
 use crate::endpoint::{OwnedResourceUri, OwnedServiceUri, ResourceUri};
+use crate::log::fs::{LogFileSystem, LogFileSystemFactory};
+use crate::log::LogManager;
 
 pub struct FileMeta {
     pub last_modified: SystemTime,
@@ -132,6 +134,20 @@ impl From<MemoryFileSystemFactory> for Arc<FileSystemManager> {
     }
 }
 
+impl From<Arc<LogManager>> for FileSystemManager {
+    fn from(logs: Arc<LogManager>) -> Self {
+        let filesystems = logs
+            .clients()
+            .cloned()
+            .map(|client| (client.location().into_owned(), Arc::new(LogFileSystem::new(client)) as Arc<dyn FileSystem>))
+            .collect();
+        let default = Arc::new(LogFileSystem::new(logs.default_client().clone()));
+        let mut factories = FileSystemFactories::default();
+        factories.register(Arc::new(LogFileSystemFactory::new(logs))).unwrap();
+        Self { factories, default, filesystems }
+    }
+}
+
 impl FileSystemManager {
     pub async fn new(factories: FileSystemFactories, filesystem: OwnedServiceUri) -> Result<Self> {
         let default = factories.open(filesystem).await?;
@@ -140,6 +156,15 @@ impl FileSystemManager {
         Ok(manager)
     }
 
+    pub fn add_log_filesystems(&mut self, logs: Arc<LogManager>) -> Result<()> {
+        self.factories.register(Arc::new(LogFileSystemFactory::new(logs.clone())))?;
+        for client in logs.clients() {
+            let fs = Arc::new(LogFileSystem::new(client.clone()));
+            self.mount_fs(fs);
+        }
+        Ok(())
+    }
+
     pub async fn open(&self, uri: &ResourceUri<'_>) -> Result<Box<dyn FileReader>> {
         let (fs, path) = self.locate_file(uri)?;
         fs.open(path).await