tengqm commented on code in PR #6013: URL: https://github.com/apache/gravitino/pull/6013#discussion_r1898152704
########## clients/filesystem-fuse/conf/gvfs_fuse.toml: ########## @@ -0,0 +1,39 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +# fuse settings +[fuse] +default_mask = 0o600 +fs_type = "memory" + +[fuse.properties] +key1 = "value1" +key2 = "value2" Review Comment: ? ########## clients/filesystem-fuse/src/config.rs: ########## @@ -0,0 +1,324 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +use crate::error::ErrorCode::{ConfigNotFound, InvalidConfig}; +use crate::utils::GvfsResult; +use config::{builder, Config}; +use log::{info, warn}; +use serde::Deserialize; +use std::collections::HashMap; +use std::fs; + +const FUSE_DEFAULT_FILE_MASK: ConfigEntity<u32> = ConfigEntity::new( + FuseConfig::MODULE_NAME, + "default_file_mask", + "The default file mask for the FUSE filesystem", + 0o600, +); + +const FUSE_DEFAULT_DIR_MASK: ConfigEntity<u32> = ConfigEntity::new( + FuseConfig::MODULE_NAME, + "default_dir_mask", + "The default directory mask for the FUSE filesystem", + 0o700, +); + +const FUSE_FS_TYPE: ConfigEntity<&'static str> = ConfigEntity::new( Review Comment: For naming conventions, I'd suggest we remove `FUSE_` from the constant names and add `DEFAULT_` to all default values for clarity. ########## clients/filesystem-fuse/src/gravitino_client.rs: ########## @@ -0,0 +1,277 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +use crate::config::GravitinoConfig; +use crate::error::{ErrorCode, GvfsError}; +use reqwest::Client; +use serde::Deserialize; +use std::collections::HashMap; +use std::fmt::Debug; +use urlencoding::encode; + +#[derive(Debug, Deserialize)] +pub(crate) struct Fileset { + pub(crate) name: String, + #[serde(rename = "type")] + pub(crate) fileset_type: String, + comment: String, + #[serde(rename = "storageLocation")] + pub(crate) storage_location: String, + properties: HashMap<String, String>, +} + +#[derive(Debug, Deserialize)] +struct FilesetResponse { + code: i32, + fileset: Fileset, +} + +#[derive(Debug, Deserialize)] +struct FileLocationResponse { + code: i32, Review Comment: here too ########## clients/filesystem-fuse/src/config.rs: ########## @@ -0,0 +1,324 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +use crate::error::ErrorCode::{ConfigNotFound, InvalidConfig}; +use crate::utils::GvfsResult; +use config::{builder, Config}; +use log::{info, warn}; +use serde::Deserialize; +use std::collections::HashMap; +use std::fs; + +const FUSE_DEFAULT_FILE_MASK: ConfigEntity<u32> = ConfigEntity::new( + FuseConfig::MODULE_NAME, + "default_file_mask", + "The default file mask for the FUSE filesystem", + 0o600, +); + +const FUSE_DEFAULT_DIR_MASK: ConfigEntity<u32> = ConfigEntity::new( + FuseConfig::MODULE_NAME, + "default_dir_mask", + "The default directory mask for the FUSE filesystem", + 0o700, +); + +const FUSE_FS_TYPE: ConfigEntity<&'static str> = ConfigEntity::new( + FuseConfig::MODULE_NAME, + "fs_type", + "The type of the FUSE filesystem", + "memory", +); + +const FUSE_CONFIG_PATH: ConfigEntity<&'static str> = ConfigEntity::new( + FuseConfig::MODULE_NAME, + "config_path", + "The path of the FUSE configuration file", + "/etc/gvfs/gvfs.toml", +); + +const FILESYSTEM_BLOCK_SIZE: ConfigEntity<u32> = ConfigEntity::new( + FilesystemConfig::MODULE_NAME, + "block_size", + "The block size of the gvfs fuse filesystem", + 4096, +); + +const GRAVITINO_URL: ConfigEntity<&'static str> = ConfigEntity::new( + GravitinoConfig::MODULE_NAME, + "gravitino_url", + "The URL of the Gravitino server", + "http://localhost:8090";, +); + +const GRAVITINO_METALAKE: ConfigEntity<&'static str> = ConfigEntity::new( + GravitinoConfig::MODULE_NAME, + "metalake", + "The metalake of the Gravitino server", + "", +); + +struct ConfigEntity<T: 'static> { + module: &'static str, + name: &'static str, + description: &'static str, + default: T, +} + +impl<T> ConfigEntity<T> { + const fn new( + module: &'static str, + name: &'static str, + description: &'static str, + default: T, + ) -> Self { + ConfigEntity { + module: module, + name: name, + description: description, + default: default, + } + } +} + +enum ConfigValue { + I32(ConfigEntity<i32>), + U32(ConfigEntity<u32>), + String(ConfigEntity<&'static str>), + Bool(ConfigEntity<bool>), + Float(ConfigEntity<f64>), +} + +struct DefaultConfig { + configs: HashMap<String, ConfigValue>, +} + +impl Default for DefaultConfig { + fn default() -> Self { + let mut configs = HashMap::new(); + + configs.insert( + Self::compose_key(FUSE_DEFAULT_FILE_MASK), + ConfigValue::U32(FUSE_DEFAULT_FILE_MASK), + ); + configs.insert( + Self::compose_key(FUSE_DEFAULT_DIR_MASK), + ConfigValue::U32(FUSE_DEFAULT_DIR_MASK), + ); + configs.insert( + Self::compose_key(FUSE_FS_TYPE), + ConfigValue::String(FUSE_FS_TYPE), + ); + configs.insert( + Self::compose_key(FUSE_CONFIG_PATH), + ConfigValue::String(FUSE_CONFIG_PATH), + ); + configs.insert( + Self::compose_key(GRAVITINO_URL), + ConfigValue::String(GRAVITINO_URL), + ); + configs.insert( + Self::compose_key(GRAVITINO_METALAKE), + ConfigValue::String(GRAVITINO_METALAKE), + ); + configs.insert( + Self::compose_key(FILESYSTEM_BLOCK_SIZE), + ConfigValue::U32(FILESYSTEM_BLOCK_SIZE), + ); + + DefaultConfig { configs } + } +} + +impl DefaultConfig { + fn compose_key<T>(entity: ConfigEntity<T>) -> String { + format!("{}.{}", entity.module, entity.name) + } +} + +#[derive(Debug, Deserialize)] +pub struct AppConfig { + #[serde(default)] + pub fuse: FuseConfig, + #[serde(default)] + pub filesystem: FilesystemConfig, + #[serde(default)] + pub gravitino: GravitinoConfig, + #[serde(default)] + pub extent_config: HashMap<String, String>, +} + +impl Default for AppConfig { + fn default() -> Self { + let builder = Self::crete_default_config_builder(); + let conf = builder + .build() + .expect("Failed to build default configuration"); + conf.try_deserialize::<AppConfig>() + .expect("Failed to deserialize default AppConfig") + } +} + +type ConfigBuilder = builder::ConfigBuilder<builder::DefaultState>; + +impl AppConfig { + fn crete_default_config_builder() -> ConfigBuilder { + let default = DefaultConfig::default(); + + default + .configs + .values() + .fold( + Config::builder(), + |builder, config_entity| match config_entity { + ConfigValue::I32(entity) => Self::add_config(builder, entity), + ConfigValue::U32(entity) => Self::add_config(builder, entity), + ConfigValue::String(entity) => Self::add_config(builder, entity), + ConfigValue::Bool(entity) => Self::add_config(builder, entity), + ConfigValue::Float(entity) => Self::add_config(builder, entity), + }, + ) + } + + fn add_config<T: Clone + Into<config::Value>>( + builder: ConfigBuilder, + entity: &ConfigEntity<T>, + ) -> ConfigBuilder { + let name = format!("{}.{}", entity.module, entity.name); + builder + .set_default(&name, entity.default.clone().into()) + .unwrap_or_else(|e| panic!("Failed to set default for {}: {}", entity.name, e)) + } + + pub fn from_file(config_file_path: Option<&str>) -> GvfsResult<AppConfig> { + let builder = Self::crete_default_config_builder(); + + let config_path = { + if config_file_path.is_some() { + let path = config_file_path.unwrap(); + //check config file exists + if fs::metadata(path).is_err() { + return Err( + ConfigNotFound.to_error("The configuration file not found".to_string()) + ); + } + info!("Use configuration file: {}", path); + path + } else { + //use default config + if fs::metadata(FUSE_CONFIG_PATH.default).is_err() { + warn!( + "The default configuration file not found, use the default configuration" + ); + return Ok(AppConfig::default()); + } else { + warn!( + "Use the default config file of {}", + FUSE_CONFIG_PATH.default + ); + } + FUSE_CONFIG_PATH.default + } + }; + let config = builder + .add_source(config::File::with_name(config_path).required(true)) + .build(); + if config.is_err() { + return Err(InvalidConfig.to_error("Failed to build configuration".to_string())); Review Comment: Can we make this error message more actionable? ########## clients/filesystem-fuse/src/open_dal_filesystem.rs: ########## @@ -0,0 +1,269 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +use crate::config::AppConfig; +use crate::error::ErrorCode::OpenDalError; +use crate::filesystem::{ + FileReader, FileStat, FileSystemCapacity, FileSystemContext, FileWriter, PathFileSystem, Result, +}; +use crate::gvfs_fuse::FileSystemSchema; +use crate::opened_file::{OpenFileFlags, OpenedFile}; +use crate::utils::GvfsResult; +use async_trait::async_trait; +use bytes::Bytes; +use fuse3::FileType::{Directory, RegularFile}; +use fuse3::{Errno, FileType, Timestamp}; +use log::{debug, error}; +use opendal::layers::LoggingLayer; +use opendal::services::S3; +use opendal::{Builder, EntryMode, ErrorKind, Metadata, Operator}; +use std::path::Path; +use std::time::SystemTime; + +pub(crate) struct OpenDalFileSystem { + op: Operator, +} + +impl OpenDalFileSystem {} + +impl OpenDalFileSystem { + fn new(op: Operator, _config: &AppConfig, _fs_context: &FileSystemContext) -> Self { + Self { op: op } + } + + pub(crate) fn create_file_system( + schema: &FileSystemSchema, + config: &AppConfig, + fs_context: &FileSystemContext, + ) -> GvfsResult<Box<dyn PathFileSystem>> { + match schema { + FileSystemSchema::S3 => { + let builder = S3::from_map(config.extent_config.clone()); + + let op = Operator::new(builder); + if let Err(e) = op { + error!("opendal create failed: {:?}", e); + return Err(OpenDalError.to_error(format!("opendal create failed: {:?}", e))); + } + let op = op.unwrap().layer(LoggingLayer::default()).finish(); + Ok(Box::new(OpenDalFileSystem::new(op, config, fs_context))) + } + } + } + + fn opendal_meta_to_file_stat(&self, meta: &Metadata, file_stat: &mut FileStat) { + let now = SystemTime::now(); + let mtime = meta.last_modified().map(|x| x.into()).unwrap_or(now); + + file_stat.size = meta.content_length(); + file_stat.kind = opendal_filemode_to_filetype(meta.mode()); + file_stat.ctime = Timestamp::from(mtime); + file_stat.atime = Timestamp::from(now); + file_stat.mtime = Timestamp::from(mtime); + } +} + +#[async_trait] +impl PathFileSystem for OpenDalFileSystem { + async fn init(&self) -> Result<()> { + Ok(()) + } + + async fn stat(&self, path: &Path) -> Result<FileStat> { + let file_name = path.to_string_lossy().to_string(); + let meta = self + .op + .stat_with(&file_name) + .await + .map_err(opendal_error_to_errno)?; + + let mut file_stat = FileStat::new_file_filestat_with_path(path, 0); + self.opendal_meta_to_file_stat(&meta, &mut file_stat); + Ok(file_stat) + } + + async fn read_dir(&self, path: &Path) -> Result<Vec<FileStat>> { + let file_name = path.to_string_lossy().to_string(); Review Comment: `path` might be something other than a directory? ########## clients/filesystem-fuse/src/main.rs: ########## @@ -16,18 +16,32 @@ * specific language governing permissions and limitations * under the License. */ +use fuse3::Errno; +use gvfs_fuse::config::AppConfig; use gvfs_fuse::{gvfs_mount, gvfs_unmount}; -use log::info; +use log::{error, info}; use tokio::signal; #[tokio::main] async fn main() -> fuse3::Result<()> { tracing_subscriber::fmt().init(); - tokio::spawn(async { gvfs_mount("gvfs").await }); + + let config = AppConfig::from_file(Some("conf/gvfs_fuse.toml)")); Review Comment: How can we ensure this relative path is always correct? ########## clients/filesystem-fuse/src/config.rs: ########## @@ -0,0 +1,324 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +use crate::error::ErrorCode::{ConfigNotFound, InvalidConfig}; +use crate::utils::GvfsResult; +use config::{builder, Config}; +use log::{info, warn}; +use serde::Deserialize; +use std::collections::HashMap; +use std::fs; + +const FUSE_DEFAULT_FILE_MASK: ConfigEntity<u32> = ConfigEntity::new( + FuseConfig::MODULE_NAME, + "default_file_mask", + "The default file mask for the FUSE filesystem", + 0o600, +); + +const FUSE_DEFAULT_DIR_MASK: ConfigEntity<u32> = ConfigEntity::new( + FuseConfig::MODULE_NAME, + "default_dir_mask", + "The default directory mask for the FUSE filesystem", + 0o700, +); + +const FUSE_FS_TYPE: ConfigEntity<&'static str> = ConfigEntity::new( + FuseConfig::MODULE_NAME, + "fs_type", + "The type of the FUSE filesystem", + "memory", +); + +const FUSE_CONFIG_PATH: ConfigEntity<&'static str> = ConfigEntity::new( + FuseConfig::MODULE_NAME, + "config_path", + "The path of the FUSE configuration file", + "/etc/gvfs/gvfs.toml", +); + +const FILESYSTEM_BLOCK_SIZE: ConfigEntity<u32> = ConfigEntity::new( + FilesystemConfig::MODULE_NAME, + "block_size", + "The block size of the gvfs fuse filesystem", + 4096, +); + +const GRAVITINO_URL: ConfigEntity<&'static str> = ConfigEntity::new( + GravitinoConfig::MODULE_NAME, + "gravitino_url", + "The URL of the Gravitino server", + "http://localhost:8090";, +); + +const GRAVITINO_METALAKE: ConfigEntity<&'static str> = ConfigEntity::new( + GravitinoConfig::MODULE_NAME, + "metalake", + "The metalake of the Gravitino server", + "", +); + +struct ConfigEntity<T: 'static> { + module: &'static str, + name: &'static str, + description: &'static str, + default: T, +} + +impl<T> ConfigEntity<T> { + const fn new( + module: &'static str, + name: &'static str, + description: &'static str, + default: T, + ) -> Self { + ConfigEntity { + module: module, + name: name, + description: description, + default: default, + } + } +} + +enum ConfigValue { + I32(ConfigEntity<i32>), + U32(ConfigEntity<u32>), + String(ConfigEntity<&'static str>), + Bool(ConfigEntity<bool>), + Float(ConfigEntity<f64>), +} + +struct DefaultConfig { + configs: HashMap<String, ConfigValue>, +} + +impl Default for DefaultConfig { + fn default() -> Self { + let mut configs = HashMap::new(); + + configs.insert( + Self::compose_key(FUSE_DEFAULT_FILE_MASK), + ConfigValue::U32(FUSE_DEFAULT_FILE_MASK), + ); + configs.insert( + Self::compose_key(FUSE_DEFAULT_DIR_MASK), + ConfigValue::U32(FUSE_DEFAULT_DIR_MASK), + ); + configs.insert( + Self::compose_key(FUSE_FS_TYPE), + ConfigValue::String(FUSE_FS_TYPE), + ); + configs.insert( + Self::compose_key(FUSE_CONFIG_PATH), + ConfigValue::String(FUSE_CONFIG_PATH), + ); + configs.insert( + Self::compose_key(GRAVITINO_URL), + ConfigValue::String(GRAVITINO_URL), + ); + configs.insert( + Self::compose_key(GRAVITINO_METALAKE), + ConfigValue::String(GRAVITINO_METALAKE), + ); + configs.insert( + Self::compose_key(FILESYSTEM_BLOCK_SIZE), + ConfigValue::U32(FILESYSTEM_BLOCK_SIZE), + ); + + DefaultConfig { configs } + } +} + +impl DefaultConfig { + fn compose_key<T>(entity: ConfigEntity<T>) -> String { + format!("{}.{}", entity.module, entity.name) + } +} + +#[derive(Debug, Deserialize)] +pub struct AppConfig { + #[serde(default)] + pub fuse: FuseConfig, + #[serde(default)] + pub filesystem: FilesystemConfig, + #[serde(default)] + pub gravitino: GravitinoConfig, + #[serde(default)] + pub extent_config: HashMap<String, String>, Review Comment: extent_config -> extend_config ########## clients/filesystem-fuse/conf/gvfs_fuse.toml: ########## @@ -0,0 +1,39 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +# fuse settings +[fuse] +default_mask = 0o600 +fs_type = "memory" + +[fuse.properties] +key1 = "value1" +key2 = "value2" + +# filesystem settings +[filesystem] +block_size = 8192 + +# Gravitino settings +[gravitino] +gravitino_url = "http://localhost:8090"; +metalake = "test" Review Comment: ? ########## clients/filesystem-fuse/src/config.rs: ########## @@ -0,0 +1,324 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +use crate::error::ErrorCode::{ConfigNotFound, InvalidConfig}; +use crate::utils::GvfsResult; +use config::{builder, Config}; +use log::{info, warn}; +use serde::Deserialize; +use std::collections::HashMap; +use std::fs; + +const FUSE_DEFAULT_FILE_MASK: ConfigEntity<u32> = ConfigEntity::new( + FuseConfig::MODULE_NAME, + "default_file_mask", + "The default file mask for the FUSE filesystem", + 0o600, +); + +const FUSE_DEFAULT_DIR_MASK: ConfigEntity<u32> = ConfigEntity::new( + FuseConfig::MODULE_NAME, + "default_dir_mask", + "The default directory mask for the FUSE filesystem", + 0o700, +); + +const FUSE_FS_TYPE: ConfigEntity<&'static str> = ConfigEntity::new( + FuseConfig::MODULE_NAME, + "fs_type", + "The type of the FUSE filesystem", + "memory", +); + +const FUSE_CONFIG_PATH: ConfigEntity<&'static str> = ConfigEntity::new( + FuseConfig::MODULE_NAME, + "config_path", + "The path of the FUSE configuration file", + "/etc/gvfs/gvfs.toml", +); + +const FILESYSTEM_BLOCK_SIZE: ConfigEntity<u32> = ConfigEntity::new( + FilesystemConfig::MODULE_NAME, + "block_size", + "The block size of the gvfs fuse filesystem", + 4096, +); + +const GRAVITINO_URL: ConfigEntity<&'static str> = ConfigEntity::new( + GravitinoConfig::MODULE_NAME, + "gravitino_url", + "The URL of the Gravitino server", + "http://localhost:8090";, +); + +const GRAVITINO_METALAKE: ConfigEntity<&'static str> = ConfigEntity::new( + GravitinoConfig::MODULE_NAME, + "metalake", + "The metalake of the Gravitino server", + "", +); + +struct ConfigEntity<T: 'static> { + module: &'static str, + name: &'static str, + description: &'static str, + default: T, +} + +impl<T> ConfigEntity<T> { + const fn new( + module: &'static str, + name: &'static str, + description: &'static str, + default: T, + ) -> Self { + ConfigEntity { + module: module, + name: name, + description: description, + default: default, + } + } +} + +enum ConfigValue { + I32(ConfigEntity<i32>), + U32(ConfigEntity<u32>), + String(ConfigEntity<&'static str>), + Bool(ConfigEntity<bool>), + Float(ConfigEntity<f64>), +} + +struct DefaultConfig { + configs: HashMap<String, ConfigValue>, +} + +impl Default for DefaultConfig { + fn default() -> Self { + let mut configs = HashMap::new(); + + configs.insert( + Self::compose_key(FUSE_DEFAULT_FILE_MASK), + ConfigValue::U32(FUSE_DEFAULT_FILE_MASK), + ); + configs.insert( + Self::compose_key(FUSE_DEFAULT_DIR_MASK), + ConfigValue::U32(FUSE_DEFAULT_DIR_MASK), + ); + configs.insert( + Self::compose_key(FUSE_FS_TYPE), + ConfigValue::String(FUSE_FS_TYPE), + ); + configs.insert( + Self::compose_key(FUSE_CONFIG_PATH), + ConfigValue::String(FUSE_CONFIG_PATH), + ); + configs.insert( + Self::compose_key(GRAVITINO_URL), + ConfigValue::String(GRAVITINO_URL), + ); + configs.insert( + Self::compose_key(GRAVITINO_METALAKE), + ConfigValue::String(GRAVITINO_METALAKE), + ); + configs.insert( + Self::compose_key(FILESYSTEM_BLOCK_SIZE), + ConfigValue::U32(FILESYSTEM_BLOCK_SIZE), + ); + + DefaultConfig { configs } + } +} + +impl DefaultConfig { + fn compose_key<T>(entity: ConfigEntity<T>) -> String { + format!("{}.{}", entity.module, entity.name) + } +} + +#[derive(Debug, Deserialize)] +pub struct AppConfig { + #[serde(default)] + pub fuse: FuseConfig, + #[serde(default)] + pub filesystem: FilesystemConfig, + #[serde(default)] + pub gravitino: GravitinoConfig, + #[serde(default)] + pub extent_config: HashMap<String, String>, +} + +impl Default for AppConfig { + fn default() -> Self { + let builder = Self::crete_default_config_builder(); + let conf = builder + .build() + .expect("Failed to build default configuration"); + conf.try_deserialize::<AppConfig>() + .expect("Failed to deserialize default AppConfig") + } +} + +type ConfigBuilder = builder::ConfigBuilder<builder::DefaultState>; + +impl AppConfig { + fn crete_default_config_builder() -> ConfigBuilder { + let default = DefaultConfig::default(); + + default + .configs + .values() + .fold( + Config::builder(), + |builder, config_entity| match config_entity { + ConfigValue::I32(entity) => Self::add_config(builder, entity), + ConfigValue::U32(entity) => Self::add_config(builder, entity), + ConfigValue::String(entity) => Self::add_config(builder, entity), + ConfigValue::Bool(entity) => Self::add_config(builder, entity), + ConfigValue::Float(entity) => Self::add_config(builder, entity), + }, + ) + } + + fn add_config<T: Clone + Into<config::Value>>( + builder: ConfigBuilder, + entity: &ConfigEntity<T>, + ) -> ConfigBuilder { + let name = format!("{}.{}", entity.module, entity.name); + builder + .set_default(&name, entity.default.clone().into()) + .unwrap_or_else(|e| panic!("Failed to set default for {}: {}", entity.name, e)) + } + + pub fn from_file(config_file_path: Option<&str>) -> GvfsResult<AppConfig> { + let builder = Self::crete_default_config_builder(); + + let config_path = { + if config_file_path.is_some() { + let path = config_file_path.unwrap(); + //check config file exists + if fs::metadata(path).is_err() { + return Err( + ConfigNotFound.to_error("The configuration file not found".to_string()) + ); + } + info!("Use configuration file: {}", path); + path + } else { + //use default config + if fs::metadata(FUSE_CONFIG_PATH.default).is_err() { + warn!( + "The default configuration file not found, use the default configuration" Review Comment: ```suggestion "The default configuration file is not found, using the default configuration" ``` ########## clients/filesystem-fuse/conf/gvfs_fuse.toml: ########## @@ -0,0 +1,39 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +# fuse settings +[fuse] +default_mask = 0o600 +fs_type = "memory" + +[fuse.properties] +key1 = "value1" +key2 = "value2" + +# filesystem settings +[filesystem] +block_size = 8192 + +# Gravitino settings +[gravitino] +gravitino_url = "http://localhost:8090"; +metalake = "test" + +# extent settings +[extent_config] +access_key = "XXX_access_key" +secret_key = "XXX_secret_key" Review Comment: ? ########## clients/filesystem-fuse/src/gravitino_client.rs: ########## @@ -0,0 +1,277 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +use crate::config::GravitinoConfig; +use crate::error::{ErrorCode, GvfsError}; +use reqwest::Client; +use serde::Deserialize; +use std::collections::HashMap; +use std::fmt::Debug; +use urlencoding::encode; + +#[derive(Debug, Deserialize)] +pub(crate) struct Fileset { + pub(crate) name: String, + #[serde(rename = "type")] + pub(crate) fileset_type: String, + comment: String, + #[serde(rename = "storageLocation")] + pub(crate) storage_location: String, + properties: HashMap<String, String>, +} + +#[derive(Debug, Deserialize)] +struct FilesetResponse { + code: i32, Review Comment: should be `u32`? ########## clients/filesystem-fuse/src/config.rs: ########## @@ -0,0 +1,324 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +use crate::error::ErrorCode::{ConfigNotFound, InvalidConfig}; +use crate::utils::GvfsResult; +use config::{builder, Config}; +use log::{info, warn}; +use serde::Deserialize; +use std::collections::HashMap; +use std::fs; + +const FUSE_DEFAULT_FILE_MASK: ConfigEntity<u32> = ConfigEntity::new( + FuseConfig::MODULE_NAME, + "default_file_mask", + "The default file mask for the FUSE filesystem", + 0o600, +); + +const FUSE_DEFAULT_DIR_MASK: ConfigEntity<u32> = ConfigEntity::new( + FuseConfig::MODULE_NAME, + "default_dir_mask", + "The default directory mask for the FUSE filesystem", + 0o700, +); + +const FUSE_FS_TYPE: ConfigEntity<&'static str> = ConfigEntity::new( + FuseConfig::MODULE_NAME, + "fs_type", + "The type of the FUSE filesystem", + "memory", +); + +const FUSE_CONFIG_PATH: ConfigEntity<&'static str> = ConfigEntity::new( + FuseConfig::MODULE_NAME, + "config_path", + "The path of the FUSE configuration file", + "/etc/gvfs/gvfs.toml", +); + +const FILESYSTEM_BLOCK_SIZE: ConfigEntity<u32> = ConfigEntity::new( + FilesystemConfig::MODULE_NAME, + "block_size", + "The block size of the gvfs fuse filesystem", + 4096, +); + +const GRAVITINO_URL: ConfigEntity<&'static str> = ConfigEntity::new( + GravitinoConfig::MODULE_NAME, + "gravitino_url", + "The URL of the Gravitino server", + "http://localhost:8090";, +); + +const GRAVITINO_METALAKE: ConfigEntity<&'static str> = ConfigEntity::new( + GravitinoConfig::MODULE_NAME, + "metalake", + "The metalake of the Gravitino server", + "", +); + +struct ConfigEntity<T: 'static> { + module: &'static str, + name: &'static str, + description: &'static str, + default: T, +} + +impl<T> ConfigEntity<T> { + const fn new( + module: &'static str, + name: &'static str, + description: &'static str, + default: T, + ) -> Self { + ConfigEntity { + module: module, + name: name, + description: description, + default: default, + } + } +} + +enum ConfigValue { + I32(ConfigEntity<i32>), + U32(ConfigEntity<u32>), + String(ConfigEntity<&'static str>), + Bool(ConfigEntity<bool>), + Float(ConfigEntity<f64>), +} + +struct DefaultConfig { + configs: HashMap<String, ConfigValue>, +} + +impl Default for DefaultConfig { + fn default() -> Self { + let mut configs = HashMap::new(); + + configs.insert( + Self::compose_key(FUSE_DEFAULT_FILE_MASK), + ConfigValue::U32(FUSE_DEFAULT_FILE_MASK), + ); + configs.insert( + Self::compose_key(FUSE_DEFAULT_DIR_MASK), + ConfigValue::U32(FUSE_DEFAULT_DIR_MASK), + ); + configs.insert( + Self::compose_key(FUSE_FS_TYPE), + ConfigValue::String(FUSE_FS_TYPE), + ); + configs.insert( + Self::compose_key(FUSE_CONFIG_PATH), + ConfigValue::String(FUSE_CONFIG_PATH), + ); + configs.insert( + Self::compose_key(GRAVITINO_URL), + ConfigValue::String(GRAVITINO_URL), + ); + configs.insert( + Self::compose_key(GRAVITINO_METALAKE), + ConfigValue::String(GRAVITINO_METALAKE), + ); + configs.insert( + Self::compose_key(FILESYSTEM_BLOCK_SIZE), + ConfigValue::U32(FILESYSTEM_BLOCK_SIZE), + ); + + DefaultConfig { configs } + } +} + +impl DefaultConfig { + fn compose_key<T>(entity: ConfigEntity<T>) -> String { + format!("{}.{}", entity.module, entity.name) Review Comment: This implies that we have some constraints on the values of `entity.module`, `entity.name`, right? For example, they cannot contain `.`. ########## clients/filesystem-fuse/src/config.rs: ########## @@ -0,0 +1,324 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +use crate::error::ErrorCode::{ConfigNotFound, InvalidConfig}; +use crate::utils::GvfsResult; +use config::{builder, Config}; +use log::{info, warn}; +use serde::Deserialize; +use std::collections::HashMap; +use std::fs; + +const FUSE_DEFAULT_FILE_MASK: ConfigEntity<u32> = ConfigEntity::new( + FuseConfig::MODULE_NAME, + "default_file_mask", + "The default file mask for the FUSE filesystem", + 0o600, +); + +const FUSE_DEFAULT_DIR_MASK: ConfigEntity<u32> = ConfigEntity::new( + FuseConfig::MODULE_NAME, + "default_dir_mask", + "The default directory mask for the FUSE filesystem", + 0o700, +); + +const FUSE_FS_TYPE: ConfigEntity<&'static str> = ConfigEntity::new( + FuseConfig::MODULE_NAME, + "fs_type", + "The type of the FUSE filesystem", + "memory", +); + +const FUSE_CONFIG_PATH: ConfigEntity<&'static str> = ConfigEntity::new( + FuseConfig::MODULE_NAME, + "config_path", + "The path of the FUSE configuration file", + "/etc/gvfs/gvfs.toml", +); + +const FILESYSTEM_BLOCK_SIZE: ConfigEntity<u32> = ConfigEntity::new( + FilesystemConfig::MODULE_NAME, + "block_size", + "The block size of the gvfs fuse filesystem", + 4096, +); + +const GRAVITINO_URL: ConfigEntity<&'static str> = ConfigEntity::new( + GravitinoConfig::MODULE_NAME, + "gravitino_url", + "The URL of the Gravitino server", + "http://localhost:8090";, +); + +const GRAVITINO_METALAKE: ConfigEntity<&'static str> = ConfigEntity::new( + GravitinoConfig::MODULE_NAME, + "metalake", + "The metalake of the Gravitino server", + "", +); + +struct ConfigEntity<T: 'static> { + module: &'static str, + name: &'static str, + description: &'static str, + default: T, +} + +impl<T> ConfigEntity<T> { + const fn new( + module: &'static str, + name: &'static str, + description: &'static str, + default: T, + ) -> Self { + ConfigEntity { + module: module, + name: name, + description: description, + default: default, + } + } +} + +enum ConfigValue { + I32(ConfigEntity<i32>), + U32(ConfigEntity<u32>), + String(ConfigEntity<&'static str>), + Bool(ConfigEntity<bool>), + Float(ConfigEntity<f64>), +} + +struct DefaultConfig { + configs: HashMap<String, ConfigValue>, +} + +impl Default for DefaultConfig { + fn default() -> Self { + let mut configs = HashMap::new(); + + configs.insert( + Self::compose_key(FUSE_DEFAULT_FILE_MASK), + ConfigValue::U32(FUSE_DEFAULT_FILE_MASK), + ); + configs.insert( + Self::compose_key(FUSE_DEFAULT_DIR_MASK), + ConfigValue::U32(FUSE_DEFAULT_DIR_MASK), + ); + configs.insert( + Self::compose_key(FUSE_FS_TYPE), + ConfigValue::String(FUSE_FS_TYPE), + ); + configs.insert( + Self::compose_key(FUSE_CONFIG_PATH), + ConfigValue::String(FUSE_CONFIG_PATH), + ); + configs.insert( + Self::compose_key(GRAVITINO_URL), + ConfigValue::String(GRAVITINO_URL), + ); + configs.insert( + Self::compose_key(GRAVITINO_METALAKE), + ConfigValue::String(GRAVITINO_METALAKE), + ); + configs.insert( + Self::compose_key(FILESYSTEM_BLOCK_SIZE), + ConfigValue::U32(FILESYSTEM_BLOCK_SIZE), + ); + + DefaultConfig { configs } + } +} + +impl DefaultConfig { + fn compose_key<T>(entity: ConfigEntity<T>) -> String { + format!("{}.{}", entity.module, entity.name) + } +} + +#[derive(Debug, Deserialize)] +pub struct AppConfig { + #[serde(default)] + pub fuse: FuseConfig, + #[serde(default)] + pub filesystem: FilesystemConfig, + #[serde(default)] + pub gravitino: GravitinoConfig, + #[serde(default)] + pub extent_config: HashMap<String, String>, +} + +impl Default for AppConfig { + fn default() -> Self { + let builder = Self::crete_default_config_builder(); + let conf = builder + .build() + .expect("Failed to build default configuration"); + conf.try_deserialize::<AppConfig>() + .expect("Failed to deserialize default AppConfig") + } +} + +type ConfigBuilder = builder::ConfigBuilder<builder::DefaultState>; + +impl AppConfig { + fn crete_default_config_builder() -> ConfigBuilder { + let default = DefaultConfig::default(); + + default + .configs + .values() + .fold( + Config::builder(), + |builder, config_entity| match config_entity { + ConfigValue::I32(entity) => Self::add_config(builder, entity), + ConfigValue::U32(entity) => Self::add_config(builder, entity), + ConfigValue::String(entity) => Self::add_config(builder, entity), + ConfigValue::Bool(entity) => Self::add_config(builder, entity), + ConfigValue::Float(entity) => Self::add_config(builder, entity), + }, + ) + } + + fn add_config<T: Clone + Into<config::Value>>( + builder: ConfigBuilder, + entity: &ConfigEntity<T>, + ) -> ConfigBuilder { + let name = format!("{}.{}", entity.module, entity.name); + builder + .set_default(&name, entity.default.clone().into()) + .unwrap_or_else(|e| panic!("Failed to set default for {}: {}", entity.name, e)) + } + + pub fn from_file(config_file_path: Option<&str>) -> GvfsResult<AppConfig> { + let builder = Self::crete_default_config_builder(); + + let config_path = { + if config_file_path.is_some() { + let path = config_file_path.unwrap(); + //check config file exists + if fs::metadata(path).is_err() { + return Err( + ConfigNotFound.to_error("The configuration file not found".to_string()) + ); + } + info!("Use configuration file: {}", path); + path + } else { + //use default config + if fs::metadata(FUSE_CONFIG_PATH.default).is_err() { + warn!( + "The default configuration file not found, use the default configuration" + ); + return Ok(AppConfig::default()); + } else { + warn!( + "Use the default config file of {}", Review Comment: ```suggestion "Using the default config file {}", ``` ########## clients/filesystem-fuse/src/filesystem.rs: ########## @@ -159,8 +161,21 @@ impl FileSystemContext { block_size: 4 * 1024, } } + + pub(crate) fn default() -> Self { + FileSystemContext { + uid: 0, + gid: 0, + default_file_perm: 0o644, + default_dir_perm: 0o755, Review Comment: Looks like we are inconsistent with the default permissions? In config file, the defaults were `0o600` and `0o700` respectively. ########## clients/filesystem-fuse/src/config.rs: ########## @@ -0,0 +1,324 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +use crate::error::ErrorCode::{ConfigNotFound, InvalidConfig}; +use crate::utils::GvfsResult; +use config::{builder, Config}; +use log::{info, warn}; +use serde::Deserialize; +use std::collections::HashMap; +use std::fs; + +const FUSE_DEFAULT_FILE_MASK: ConfigEntity<u32> = ConfigEntity::new( + FuseConfig::MODULE_NAME, + "default_file_mask", + "The default file mask for the FUSE filesystem", + 0o600, +); + +const FUSE_DEFAULT_DIR_MASK: ConfigEntity<u32> = ConfigEntity::new( + FuseConfig::MODULE_NAME, + "default_dir_mask", + "The default directory mask for the FUSE filesystem", + 0o700, +); + +const FUSE_FS_TYPE: ConfigEntity<&'static str> = ConfigEntity::new( + FuseConfig::MODULE_NAME, + "fs_type", + "The type of the FUSE filesystem", + "memory", +); + +const FUSE_CONFIG_PATH: ConfigEntity<&'static str> = ConfigEntity::new( + FuseConfig::MODULE_NAME, + "config_path", + "The path of the FUSE configuration file", + "/etc/gvfs/gvfs.toml", Review Comment: Writing this file needs a root account, is that okay? ########## clients/filesystem-fuse/src/gravitino_client.rs: ########## @@ -0,0 +1,277 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +use crate::config::GravitinoConfig; +use crate::error::{ErrorCode, GvfsError}; +use reqwest::Client; +use serde::Deserialize; +use std::collections::HashMap; +use std::fmt::Debug; +use urlencoding::encode; + +#[derive(Debug, Deserialize)] +pub(crate) struct Fileset { + pub(crate) name: String, + #[serde(rename = "type")] + pub(crate) fileset_type: String, + comment: String, + #[serde(rename = "storageLocation")] + pub(crate) storage_location: String, + properties: HashMap<String, String>, +} + +#[derive(Debug, Deserialize)] +struct FilesetResponse { + code: i32, + fileset: Fileset, +} + +#[derive(Debug, Deserialize)] +struct FileLocationResponse { + code: i32, + #[serde(rename = "fileLocation")] + location: String, +} + +pub(crate) struct GravitinoClient { + gravitino_uri: String, + metalake: String, + + http_client: Client, +} + +impl GravitinoClient { + pub fn new(config: &GravitinoConfig) -> Self { + Self { + gravitino_uri: config.gravitino_url.clone(), + metalake: config.metalake.clone(), + http_client: Client::new(), + } + } + + pub fn init(&self) {} + + pub fn do_post(&self, path: &str, data: &str) { + println!("POST request to {} with data: {}", path, data); Review Comment: This is for debug only ########## clients/filesystem-fuse/src/gravitino_client.rs: ########## @@ -0,0 +1,277 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +use crate::config::GravitinoConfig; +use crate::error::{ErrorCode, GvfsError}; +use reqwest::Client; +use serde::Deserialize; +use std::collections::HashMap; +use std::fmt::Debug; +use urlencoding::encode; + +#[derive(Debug, Deserialize)] +pub(crate) struct Fileset { + pub(crate) name: String, + #[serde(rename = "type")] + pub(crate) fileset_type: String, + comment: String, + #[serde(rename = "storageLocation")] + pub(crate) storage_location: String, + properties: HashMap<String, String>, +} + +#[derive(Debug, Deserialize)] +struct FilesetResponse { + code: i32, + fileset: Fileset, +} + +#[derive(Debug, Deserialize)] +struct FileLocationResponse { + code: i32, + #[serde(rename = "fileLocation")] + location: String, +} + +pub(crate) struct GravitinoClient { + gravitino_uri: String, + metalake: String, + + http_client: Client, Review Comment: This can be simply `client: Client`, right? ########## clients/filesystem-fuse/src/gvfs_fuse.rs: ########## @@ -0,0 +1,130 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +use crate::config::AppConfig; +use crate::default_raw_filesystem::DefaultRawFileSystem; +use crate::error::ErrorCode::UnSupportedFilesystem; +use crate::filesystem::FileSystemContext; +use crate::fuse_api_handle::FuseApiHandle; +use crate::fuse_server::FuseServer; +use crate::gvfs_creator::create_gvfs_filesystem; +use crate::gvfs_fileset_fs::GvfsFilesetFs; +use crate::memory_filesystem::MemoryFileSystem; +use crate::utils::GvfsResult; +use log::info; +use once_cell::sync::Lazy; +use std::sync::Arc; +use tokio::sync::Mutex; + +static SERVER: Lazy<Mutex<Option<Arc<FuseServer>>>> = Lazy::new(|| Mutex::new(None)); + +pub(crate) enum CreateFsResult { + Memory(MemoryFileSystem), + Gvfs(GvfsFilesetFs), + FuseMemoryFs(FuseApiHandle<DefaultRawFileSystem<MemoryFileSystem>>), + FuseGvfs(FuseApiHandle<DefaultRawFileSystem<GvfsFilesetFs>>), + None, +} + +pub enum FileSystemSchema { + S3, +} + +pub async fn mount(mount_to: &str, mount_from: &str, config: &AppConfig) -> GvfsResult<()> { + info!("Starting gvfs-fuse server..."); + let svr = Arc::new(FuseServer::new(mount_to)); + { + let mut server = SERVER.lock().await; + *server = Some(svr.clone()); + } + let fs = create_fuse_fs(mount_from, config).await?; + match fs { + CreateFsResult::FuseMemoryFs(vfs) => svr.start(vfs).await?, + CreateFsResult::FuseGvfs(vfs) => svr.start(vfs).await?, + _ => return Err(UnSupportedFilesystem.to_error("Unsupported filesystem type".to_string())), + } + Ok(()) +} + +pub async fn unmount() -> GvfsResult<()> { + info!("Stop gvfs-fuse server..."); Review Comment: ```suggestion info!("Stopping gvfs-fuse server..."); ``` ########## clients/filesystem-fuse/src/gvfs_creator.rs: ########## @@ -114,5 +78,72 @@ pub async fn create_gvfs_filesystem() { // // `XXXFileSystem is a filesystem that allows you to implement file access through your own extensions. - todo!("Implement the createGvfsFuseFileSystem function"); + let client = GravitinoClient::new(&config.gravitino); + + let (catalog, schema, fileset) = extract_fileset(mount_from)?; + let location = client + .get_fileset(&catalog, &schema, &fileset) + .await? + .storage_location; + let (schema, location) = extract_storage_filesystem(&location).unwrap(); + + let inner_fs = create_fs_by_schema(&schema, config, fs_context)?; + + let fs = GvfsFilesetFs::new(inner_fs, Path::new(&location), client, config, fs_context).await; + Ok(CreateFsResult::Gvfs(fs)) +} + +fn create_fs_by_schema( Review Comment: by -> of, with ? ########## clients/filesystem-fuse/src/gravitino_client.rs: ########## @@ -0,0 +1,277 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +use crate::config::GravitinoConfig; +use crate::error::{ErrorCode, GvfsError}; +use reqwest::Client; +use serde::Deserialize; +use std::collections::HashMap; +use std::fmt::Debug; +use urlencoding::encode; + +#[derive(Debug, Deserialize)] +pub(crate) struct Fileset { + pub(crate) name: String, + #[serde(rename = "type")] + pub(crate) fileset_type: String, + comment: String, + #[serde(rename = "storageLocation")] + pub(crate) storage_location: String, + properties: HashMap<String, String>, +} + +#[derive(Debug, Deserialize)] +struct FilesetResponse { + code: i32, + fileset: Fileset, +} + +#[derive(Debug, Deserialize)] +struct FileLocationResponse { + code: i32, + #[serde(rename = "fileLocation")] + location: String, +} + +pub(crate) struct GravitinoClient { + gravitino_uri: String, + metalake: String, + + http_client: Client, +} + +impl GravitinoClient { + pub fn new(config: &GravitinoConfig) -> Self { + Self { + gravitino_uri: config.gravitino_url.clone(), + metalake: config.metalake.clone(), + http_client: Client::new(), + } + } + + pub fn init(&self) {} + + pub fn do_post(&self, path: &str, data: &str) { + println!("POST request to {} with data: {}", path, data); + } + + pub fn request(&self, _path: &str, _data: &str) -> Result<(), GvfsError> { + todo!() + } + + pub fn list_schema(&self) -> Result<(), GvfsError> { + todo!() + } + + pub fn list_fileset(&self) -> Result<(), GvfsError> { + todo!() + } + + fn get_fileset_url(&self, catalog_name: &str, schema_name: &str, fileset_name: &str) -> String { + format!( + "{}/api/metalakes/{}/catalogs/{}/schemas/{}/filesets/{}", + self.gravitino_uri, self.metalake, catalog_name, schema_name, fileset_name + ) + } + + async fn do_get<T>(&self, url: &str) -> Result<T, GvfsError> + where + T: for<'de> Deserialize<'de>, + { + let http_resp = + self.http_client.get(url).send().await.map_err(|e| { + GvfsError::RestError(format!("Failed to send request to {}", url), e) + })?; + + let res = http_resp.json::<T>().await.map_err(|e| { + GvfsError::RestError(format!("Failed to parse response from {}", url), e) + })?; + + Ok(res) + } + + pub async fn get_fileset( + &self, + catalog_name: &str, + schema_name: &str, + fileset_name: &str, + ) -> Result<Fileset, GvfsError> { + let url = self.get_fileset_url(catalog_name, schema_name, fileset_name); + let res = self.do_get::<FilesetResponse>(&url).await?; + + if res.code != 0 { Review Comment: the code is not HTTP status code? ########## clients/filesystem-fuse/src/gvfs_fuse.rs: ########## @@ -0,0 +1,130 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +use crate::config::AppConfig; +use crate::default_raw_filesystem::DefaultRawFileSystem; +use crate::error::ErrorCode::UnSupportedFilesystem; +use crate::filesystem::FileSystemContext; +use crate::fuse_api_handle::FuseApiHandle; +use crate::fuse_server::FuseServer; +use crate::gvfs_creator::create_gvfs_filesystem; +use crate::gvfs_fileset_fs::GvfsFilesetFs; +use crate::memory_filesystem::MemoryFileSystem; +use crate::utils::GvfsResult; +use log::info; +use once_cell::sync::Lazy; +use std::sync::Arc; +use tokio::sync::Mutex; + +static SERVER: Lazy<Mutex<Option<Arc<FuseServer>>>> = Lazy::new(|| Mutex::new(None)); + +pub(crate) enum CreateFsResult { + Memory(MemoryFileSystem), + Gvfs(GvfsFilesetFs), + FuseMemoryFs(FuseApiHandle<DefaultRawFileSystem<MemoryFileSystem>>), + FuseGvfs(FuseApiHandle<DefaultRawFileSystem<GvfsFilesetFs>>), + None, +} + +pub enum FileSystemSchema { + S3, +} + +pub async fn mount(mount_to: &str, mount_from: &str, config: &AppConfig) -> GvfsResult<()> { + info!("Starting gvfs-fuse server..."); + let svr = Arc::new(FuseServer::new(mount_to)); + { + let mut server = SERVER.lock().await; + *server = Some(svr.clone()); + } + let fs = create_fuse_fs(mount_from, config).await?; + match fs { + CreateFsResult::FuseMemoryFs(vfs) => svr.start(vfs).await?, + CreateFsResult::FuseGvfs(vfs) => svr.start(vfs).await?, + _ => return Err(UnSupportedFilesystem.to_error("Unsupported filesystem type".to_string())), + } + Ok(()) +} + +pub async fn unmount() -> GvfsResult<()> { + info!("Stop gvfs-fuse server..."); + let svr = { + let mut server = SERVER.lock().await; + if server.is_none() { + info!("Server is already stopped."); + return Ok(()); + } + server.take().unwrap() + }; + svr.stop().await +} + +pub(crate) async fn create_fuse_fs( + mount_from: &str, + config: &AppConfig, +) -> GvfsResult<CreateFsResult> { + let uid = unsafe { libc::getuid() }; + let gid = unsafe { libc::getgid() }; + let fs_context = FileSystemContext { + uid: uid, + gid: gid, + default_file_perm: 0o644, + default_dir_perm: 0o755, Review Comment: We are not using the value from the config file? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
