From b1d8e6080218885f095287b37aef25f520d75f34 Mon Sep 17 00:00:00 2001 From: junxiang Mu <1948535941@qq.com> Date: Mon, 21 Apr 2025 01:42:57 +0000 Subject: [PATCH 1/5] fix sql Signed-off-by: junxiang Mu <1948535941@qq.com> --- Cargo.lock | 1 + rustfs/src/server/service_state.rs | 8 +- rustfs/src/storage/ecfs.rs | 7 +- s3select/api/src/query/session.rs | 16 ++- s3select/query/Cargo.toml | 1 + s3select/query/src/dispatcher/manager.rs | 119 +++++++++++++++++++---- s3select/query/src/instance.rs | 18 +++- 7 files changed, 139 insertions(+), 31 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index f69232ecc..47a013b25 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6560,6 +6560,7 @@ dependencies = [ "datafusion", "derive_builder", "futures", + "lazy_static", "parking_lot 0.12.3", "s3s", "snafu", diff --git a/rustfs/src/server/service_state.rs b/rustfs/src/server/service_state.rs index d27296e35..ad137f66d 100644 --- a/rustfs/src/server/service_state.rs +++ b/rustfs/src/server/service_state.rs @@ -96,7 +96,9 @@ impl ServiceStateManager { ServiceState::Starting => { info!("Service is starting..."); #[cfg(target_os = "linux")] - if let Err(e) = libsystemd::daemon::notify(false, &[libsystemd::daemon::NotifyState::Status("Starting...".to_string())]) { + if let Err(e) = + libsystemd::daemon::notify(false, &[libsystemd::daemon::NotifyState::Status("Starting...".to_string())]) + { tracing::error!("Failed to notify systemd of starting state: {}", e); } } @@ -111,7 +113,9 @@ impl ServiceStateManager { ServiceState::Stopped => { info!("Service has stopped"); #[cfg(target_os = "linux")] - if let Err(e) = libsystemd::daemon::notify(false, &[libsystemd::daemon::NotifyState::Status("Stopped".to_string())]) { + if let Err(e) = + libsystemd::daemon::notify(false, &[libsystemd::daemon::NotifyState::Status("Stopped".to_string())]) + { tracing::error!("Failed to notify systemd of stopped state: {}", e); } } diff --git a/rustfs/src/storage/ecfs.rs b/rustfs/src/storage/ecfs.rs index f1f53b9fb..252cd316f 100644 --- a/rustfs/src/storage/ecfs.rs +++ b/rustfs/src/storage/ecfs.rs @@ -1895,10 +1895,13 @@ impl S3 for FS { let db = make_rustfsms(input.clone(), false).await.map_err(|e| { error!("make db failed, {}", e.to_string()); - s3_error!(InternalError) + s3_error!(InternalError, "{}", e.to_string()) })?; let query = Query::new(Context { input: input.clone() }, input.request.expression); - let result = db.execute(&query).await.map_err(|_| s3_error!(InternalError))?; + let result = db + .execute(&query) + .await + .map_err(|e| s3_error!(InternalError, "{}", e.to_string()))?; let results = result.result().chunk_result().await.unwrap().to_vec(); diff --git a/s3select/api/src/query/session.rs b/s3select/api/src/query/session.rs index c9d91f51d..286ee9f8e 100644 --- a/s3select/api/src/query/session.rs +++ b/s3select/api/src/query/session.rs @@ -1,8 +1,8 @@ use std::sync::Arc; -use bytes::Bytes; use datafusion::{ execution::{context::SessionState, runtime_env::RuntimeEnvBuilder, SessionStateBuilder}, + parquet::data_type::AsBytes, prelude::SessionContext, }; use object_store::{memory::InMemory, path::Path, ObjectStore}; @@ -65,7 +65,19 @@ impl SessionCtxFactory { 8,Henry,32,IT,6200 9,Ivy,24,Marketing,4800 10,Jack,38,Finance,7500"; - let data_bytes = Bytes::from(data.to_vec()); + let data_bytes = data.as_bytes(); + // let data = r#""year"╦"gender"╦"ethnicity"╦"firstname"╦"count"╦"rank" + // "2011"╦"FEMALE"╦"ASIAN AND PACIFIC ISLANDER"╦"SOPHIA"╦"119"╦"1" + // "2011"╦"FEMALE"╦"ASIAN AND PACIFIC ISLANDER"╦"CHLOE"╦"106"╦"2" + // "2011"╦"FEMALE"╦"ASIAN AND PACIFIC ISLANDER"╦"EMILY"╦"93"╦"3" + // "2011"╦"FEMALE"╦"ASIAN AND PACIFIC ISLANDER"╦"OLIVIA"╦"89"╦"4" + // "2011"╦"FEMALE"╦"ASIAN AND PACIFIC ISLANDER"╦"EMMA"╦"75"╦"5" + // "2011"╦"FEMALE"╦"ASIAN AND PACIFIC ISLANDER"╦"ISABELLA"╦"67"╦"6" + // "2011"╦"FEMALE"╦"ASIAN AND PACIFIC ISLANDER"╦"TIFFANY"╦"54"╦"7" + // "2011"╦"FEMALE"╦"ASIAN AND PACIFIC ISLANDER"╦"ASHLEY"╦"52"╦"8" + // "2011"╦"FEMALE"╦"ASIAN AND PACIFIC ISLANDER"╦"FIONA"╦"48"╦"9" + // "2011"╦"FEMALE"╦"ASIAN AND PACIFIC ISLANDER"╦"ANGELA"╦"47"╦"10""#; + // let data_bytes = Bytes::from(data); let path = Path::from(context.input.key.clone()); store.put(&path, data_bytes.into()).await.map_err(|e| { error!("put data into memory failed: {}", e.to_string()); diff --git a/s3select/query/Cargo.toml b/s3select/query/Cargo.toml index 8d9b159c2..559eab27a 100644 --- a/s3select/query/Cargo.toml +++ b/s3select/query/Cargo.toml @@ -10,6 +10,7 @@ async-trait.workspace = true datafusion = { workspace = true } derive_builder = { workspace = true } futures = { workspace = true } +lazy_static = { workspace = true } parking_lot = { version = "0.12.3" } s3s.workspace = true snafu = { workspace = true, features = ["backtrace"] } diff --git a/s3select/query/src/dispatcher/manager.rs b/s3select/query/src/dispatcher/manager.rs index e85e7a543..05f763435 100644 --- a/s3select/query/src/dispatcher/manager.rs +++ b/s3select/query/src/dispatcher/manager.rs @@ -1,4 +1,5 @@ use std::{ + ops::Deref, pin::Pin, sync::Arc, task::{Context, Poll}, @@ -19,8 +20,10 @@ use api::{ }; use async_trait::async_trait; use datafusion::{ - arrow::{datatypes::SchemaRef, record_batch::RecordBatch}, - config::CsvOptions, + arrow::{ + datatypes::{Schema, SchemaRef}, + record_batch::RecordBatch, + }, datasource::{ file_format::{csv::CsvFormat, json::JsonFormat, parquet::ParquetFormat}, listing::{ListingOptions, ListingTable, ListingTableConfig, ListingTableUrl}, @@ -29,7 +32,8 @@ use datafusion::{ execution::{RecordBatchStream, SendableRecordBatchStream}, }; use futures::{Stream, StreamExt}; -use s3s::dto::SelectObjectContentInput; +use lazy_static::lazy_static; +use s3s::dto::{FileHeaderInfo, SelectObjectContentInput}; use crate::{ execution::factory::QueryExecutionFactoryRef, @@ -37,6 +41,12 @@ use crate::{ sql::logical::planner::DefaultLogicalPlanner, }; +lazy_static! { + static ref IGNORE: FileHeaderInfo = FileHeaderInfo::from_static(FileHeaderInfo::IGNORE); + static ref NONE: FileHeaderInfo = FileHeaderInfo::from_static(FileHeaderInfo::NONE); + static ref USE: FileHeaderInfo = FileHeaderInfo::from_static(FileHeaderInfo::USE); +} + #[derive(Clone)] pub struct SimpleQueryDispatcher { input: SelectObjectContentInput, @@ -138,25 +148,94 @@ impl SimpleQueryDispatcher { async fn build_scheme_provider(&self, session: &SessionCtx) -> QueryResult { let path = format!("s3://{}/{}", self.input.bucket, self.input.key); let table_path = ListingTableUrl::parse(path)?; - let listing_options = if self.input.request.input_serialization.csv.is_some() { - let file_format = CsvFormat::default().with_options(CsvOptions::default().with_has_header(true)); - ListingOptions::new(Arc::new(file_format)).with_file_extension(".csv") - } else if self.input.request.input_serialization.parquet.is_some() { - let file_format = ParquetFormat::new(); - ListingOptions::new(Arc::new(file_format)).with_file_extension(".parquet") - } else if self.input.request.input_serialization.json.is_some() { - let file_format = JsonFormat::default(); - ListingOptions::new(Arc::new(file_format)).with_file_extension(".json") - } else { - return Err(QueryError::NotImplemented { - err: "not support this file type".to_string(), - }); - }; + let (listing_options, need_rename_volume_name, need_ignore_volume_name) = + if let Some(csv) = self.input.request.input_serialization.csv.as_ref() { + let mut need_rename_volume_name = false; + let mut need_ignore_volume_name = false; + let mut file_format = CsvFormat::default() + .with_comment( + csv.comments + .clone() + .map(|c| c.as_bytes().first().copied().unwrap_or_default()), + ) + .with_escape( + csv.quote_escape_character + .clone() + .map(|e| e.as_bytes().first().copied().unwrap_or_default()), + ); + if let Some(delimiter) = csv.field_delimiter.as_ref() { + file_format = file_format.with_delimiter(delimiter.as_bytes().first().copied().unwrap_or_default()); + } + if csv.file_header_info.is_some() {} + match csv.file_header_info.as_ref() { + Some(info) => { + if *info == *NONE { + file_format = file_format.with_has_header(false); + need_rename_volume_name = true; + } else if *info == *IGNORE { + file_format = file_format.with_has_header(true); + need_rename_volume_name = true; + need_ignore_volume_name = true; + } else if *info == *USE { + file_format = file_format.with_has_header(true); + } else { + return Err(QueryError::NotImplemented { + err: "unsupported FileHeaderInfo".to_string(), + }); + } + } + _ => { + return Err(QueryError::NotImplemented { + err: "unsupported FileHeaderInfo".to_string(), + }); + } + } + if let Some(quote) = csv.quote_character.as_ref() { + file_format = file_format.with_quote(quote.as_bytes().first().copied().unwrap_or_default()); + } + ( + ListingOptions::new(Arc::new(file_format)).with_file_extension(".csv"), + need_rename_volume_name, + need_ignore_volume_name, + ) + } else if self.input.request.input_serialization.parquet.is_some() { + let file_format = ParquetFormat::new(); + (ListingOptions::new(Arc::new(file_format)).with_file_extension(".parquet"), false, false) + } else if self.input.request.input_serialization.json.is_some() { + let file_format = JsonFormat::default(); + (ListingOptions::new(Arc::new(file_format)).with_file_extension(".json"), false, false) + } else { + return Err(QueryError::NotImplemented { + err: "not support this file type".to_string(), + }); + }; let resolve_schema = listing_options.infer_schema(session.inner(), &table_path).await?; - let config = ListingTableConfig::new(table_path) - .with_listing_options(listing_options) - .with_schema(resolve_schema); + let config = if need_rename_volume_name { + let mut new_fields = Vec::new(); + for (i, field) in resolve_schema.fields().iter().enumerate() { + let f_name = field.name(); + let mut_field = field.deref().clone(); + if f_name.starts_with("column_") { + let re_name = f_name.replace("column_", "_"); + new_fields.push(mut_field.with_name(re_name)); + } else if need_ignore_volume_name { + let re_name = format!("_{}", i + 1); + new_fields.push(mut_field.with_name(re_name)); + } else { + new_fields.push(mut_field); + } + } + let new_schema = Arc::new(Schema::new(new_fields).with_metadata(resolve_schema.metadata().clone())); + ListingTableConfig::new(table_path) + .with_listing_options(listing_options) + .with_schema(new_schema) + } else { + ListingTableConfig::new(table_path) + .with_listing_options(listing_options) + .with_schema(resolve_schema) + }; + // rename default let provider = Arc::new(ListingTable::try_new(config)?); let current_session_table_provider = self.build_table_handle_provider()?; let metadata_provider = diff --git a/s3select/query/src/instance.rs b/s3select/query/src/instance.rs index 3ad8941a8..44952a4ac 100644 --- a/s3select/query/src/instance.rs +++ b/s3select/query/src/instance.rs @@ -101,8 +101,8 @@ mod tests { }; use datafusion::{arrow::util::pretty, assert_batches_eq}; use s3s::dto::{ - CSVInput, CSVOutput, ExpressionType, InputSerialization, OutputSerialization, SelectObjectContentInput, - SelectObjectContentRequest, + CSVInput, CSVOutput, ExpressionType, FieldDelimiter, FileHeaderInfo, InputSerialization, OutputSerialization, + RecordDelimiter, SelectObjectContentInput, SelectObjectContentRequest, }; use crate::instance::make_rustfsms; @@ -122,7 +122,10 @@ mod tests { expression: sql.to_string(), expression_type: ExpressionType::from_static("SQL"), input_serialization: InputSerialization { - csv: Some(CSVInput::default()), + csv: Some(CSVInput { + file_header_info: Some(FileHeaderInfo::from_static(FileHeaderInfo::USE)), + ..Default::default() + }), ..Default::default() }, output_serialization: OutputSerialization { @@ -164,7 +167,7 @@ mod tests { #[tokio::test] #[ignore] async fn test_func_sql() { - let sql = "select count(s.id) from S3Object as s"; + let sql = "SELECT s._1 FROM S3Object s"; let input = SelectObjectContentInput { bucket: "dandan".to_string(), expected_bucket_owner: None, @@ -176,7 +179,12 @@ mod tests { expression: sql.to_string(), expression_type: ExpressionType::from_static("SQL"), input_serialization: InputSerialization { - csv: Some(CSVInput::default()), + csv: Some(CSVInput { + file_header_info: Some(FileHeaderInfo::from_static(FileHeaderInfo::IGNORE)), + field_delimiter: Some(FieldDelimiter::from("╦")), + record_delimiter: Some(RecordDelimiter::from("\n")), + ..Default::default() + }), ..Default::default() }, output_serialization: OutputSerialization { From 29fbfc3d6ecf2239285926b7e675924fc98dc463 Mon Sep 17 00:00:00 2001 From: junxiang Mu <1948535941@qq.com> Date: Mon, 21 Apr 2025 03:09:54 +0000 Subject: [PATCH 2/5] move protobuf generate into bin crate Signed-off-by: junxiang Mu <1948535941@qq.com> --- README.md | 4 +++ common/protos/Cargo.toml | 6 ++-- common/protos/{build.rs => src/main.rs} | 48 ++++++++++++++----------- 3 files changed, 36 insertions(+), 22 deletions(-) rename common/protos/{build.rs => src/main.rs} (88%) diff --git a/README.md b/README.md index 3e1de2cb0..511f6736e 100644 --- a/README.md +++ b/README.md @@ -12,6 +12,10 @@ https://github.com/google/flatbuffers/releases/download/v25.2.10/Linux.flatc.bin https://github.com/protocolbuffers/protobuf/releases/download/v30.2/protoc-30.2-linux-x86_64.zip +generate protobuf code: + +```cargo run --bin gproto``` + Or use Docker: ```yml diff --git a/common/protos/Cargo.toml b/common/protos/Cargo.toml index 581fb2d02..2041375c4 100644 --- a/common/protos/Cargo.toml +++ b/common/protos/Cargo.toml @@ -6,6 +6,10 @@ edition.workspace = true [lints] workspace = true +[[bin]] +name = "gproto" +path = "src/main.rs" + [dependencies] #async-backtrace = { workspace = true, optional = true } common.workspace = true @@ -15,7 +19,5 @@ protobuf = { workspace = true } tokio = { workspace = true } tonic = { workspace = true, features = ["transport"] } tower = { workspace = true } - -[build-dependencies] prost-build = { workspace = true } tonic-build = { workspace = true } diff --git a/common/protos/build.rs b/common/protos/src/main.rs similarity index 88% rename from common/protos/build.rs rename to common/protos/src/main.rs index fc55ec7a6..44fe20a10 100644 --- a/common/protos/build.rs +++ b/common/protos/src/main.rs @@ -1,13 +1,7 @@ -use std::{ - cmp, env, fs, - io::Write, - path::{Path, PathBuf}, - process::Command, -}; +use std::{cmp, env, fs, io::Write, path::Path, process::Command}; type AnyError = Box; -const ENV_OUT_DIR: &str = "OUT_DIR"; const VERSION_PROTOBUF: Version = Version(30, 2, 0); // 30.2.0 const VERSION_FLATBUFFERS: Version = Version(24, 3, 25); // 24.3.25 /// Build protos if the major version of `flatc` or `protoc` is greater @@ -37,16 +31,16 @@ fn main() -> Result<(), AnyError> { } // path of proto file - let project_root_dir = env::current_dir()?; - let proto_dir = project_root_dir.join("src"); + let project_root_dir = env::current_dir()?.join("common/protos/src"); + let proto_dir = project_root_dir.clone(); let proto_files = &["node.proto"]; - let proto_out_dir = project_root_dir.join("src").join("generated").join("proto_gen"); - let flatbuffer_out_dir = project_root_dir.join("src").join("generated").join("flatbuffers_generated"); - let descriptor_set_path = PathBuf::from(env::var(ENV_OUT_DIR).unwrap()).join("proto-descriptor.bin"); + let proto_out_dir = project_root_dir.join("generated").join("proto_gen"); + let flatbuffer_out_dir = project_root_dir.join("generated").join("flatbuffers_generated"); + // let descriptor_set_path = PathBuf::from(env::var(ENV_OUT_DIR).unwrap()).join("proto-descriptor.bin"); tonic_build::configure() .out_dir(proto_out_dir) - .file_descriptor_set_path(descriptor_set_path) + // .file_descriptor_set_path(descriptor_set_path) .protoc_arg("--experimental_allow_proto3_optional") .compile_well_known_types(true) .emit_rerun_if_changed(false) @@ -54,17 +48,13 @@ fn main() -> Result<(), AnyError> { .map_err(|e| format!("Failed to generate protobuf file: {e}."))?; // protos/gen/mod.rs - let generated_mod_rs_path = project_root_dir - .join("src") - .join("generated") - .join("proto_gen") - .join("mod.rs"); + let generated_mod_rs_path = project_root_dir.join("generated").join("proto_gen").join("mod.rs"); let mut generated_mod_rs = fs::File::create(generated_mod_rs_path)?; writeln!(&mut generated_mod_rs, "pub mod node_service;")?; generated_mod_rs.flush()?; - let generated_mod_rs_path = project_root_dir.join("src").join("generated").join("mod.rs"); + let generated_mod_rs_path = project_root_dir.join("generated").join("mod.rs"); let mut generated_mod_rs = fs::File::create(generated_mod_rs_path)?; writeln!(&mut generated_mod_rs, "#![allow(unused_imports)]")?; @@ -80,7 +70,6 @@ fn main() -> Result<(), AnyError> { Err(_) => "flatc".to_string(), }; - // build src/protos/*.fbs files to src/protos/gen/ compile_flatbuffers_models( &mut generated_mod_rs, &flatc_path, @@ -88,6 +77,8 @@ fn main() -> Result<(), AnyError> { flatbuffer_out_dir.clone(), vec!["models"], )?; + + fmt(); Ok(()) } @@ -260,3 +251,20 @@ fn protobuf_compiler_version() -> Result { } }) } + +fn fmt() { + let output = Command::new("cargo").arg("fmt").arg("-p").arg("protos").status(); + + match output { + Ok(status) => { + if status.success() { + println!("cargo fmt executed successfully."); + } else { + eprintln!("cargo fmt failed with status: {:?}", status); + } + } + Err(e) => { + eprintln!("Failed to execute cargo fmt: {}", e); + } + } +} From 536b2e2ed914d557b562a966b1e1260fe064dead Mon Sep 17 00:00:00 2001 From: DamonXue Date: Mon, 21 Apr 2025 22:05:31 +0800 Subject: [PATCH 3/5] feat: update launch configuration and docker-compose for enhanced observability and logging --- .vscode/launch.json | 38 ++++++++++++++++++++++++++------------ docker-compose.yaml | 13 ------------- 2 files changed, 26 insertions(+), 25 deletions(-) diff --git a/.vscode/launch.json b/.vscode/launch.json index 8c61c1833..28578442c 100644 --- a/.vscode/launch.json +++ b/.vscode/launch.json @@ -20,19 +20,33 @@ } }, "env": { - "RUST_LOG": "rustfs=debug,ecstore=info,s3s=debug" + "RUST_LOG": "rustfs=debug,ecstore=info,s3s=debug", + "RUSTFS_VOLUMES": "./target/volume/test{0...3}", + "RUSTFS_ADDRESS": "[::]:9000", + "RUSTFS_CONSOLE_ENABLE": "true", + "RUSTFS_CONSOLE_ADDRESS": "[::]:9002", + "RUSTFS_SERVER_DOMAINS": "localhost:9000", + "RUSTFS_TLS_PATH": "./deploy/certs", + "RUSTFS_OBS_CONFIG": "./deploy/config/obs.example.toml", + "RUSTFS__OBSERVABILITY__ENDPOINT": "http://localhost:4317", + "RUSTFS__OBSERVABILITY__USE_STDOUT": "true", + "RUSTFS__OBSERVABILITY__SAMPLE_RATIO": "2.0", + "RUSTFS__OBSERVABILITY__METER_INTERVAL": "30", + "RUSTFS__OBSERVABILITY__SERVICE_NAME": "rustfs", + "RUSTFS__OBSERVABILITY__SERVICE_VERSION": "0.1.0", + "RUSTFS__OBSERVABILITY__ENVIRONMENT": "develop", + "RUSTFS__OBSERVABILITY__LOGGER_LEVEL": "info", + "RUSTFS__SINKS__FILE__ENABLED": "true", + "RUSTFS__SINKS__FILE__PATH": "./deploy/logs/rustfs.log", + "RUSTFS__SINKS__WEBHOOK__ENABLED": "false", + "RUSTFS__SINKS__WEBHOOK__ENDPOINT": "", + "RUSTFS__SINKS__WEBHOOK__AUTH_TOKEN": "", + "RUSTFS__SINKS__KAFKA__ENABLED": "false", + "RUSTFS__SINKS__KAFKA__BOOTSTRAP_SERVERS": "", + "RUSTFS__SINKS__KAFKA__TOPIC": "", + "RUSTFS__LOGGER__QUEUE_CAPACITY": "10" + }, - "args": [ - "--access-key", - "AKEXAMPLERUSTFS", - "--secret-key", - "SKEXAMPLERUSTFS", - "--address", - "0.0.0.0:9010", - "--domain-name", - "127.0.0.1:9010", - "./target/volume/test{0...4}" - ], "cwd": "${workspaceFolder}" }, { diff --git a/docker-compose.yaml b/docker-compose.yaml index bee2f4a0a..b26b84a63 100644 --- a/docker-compose.yaml +++ b/docker-compose.yaml @@ -67,16 +67,3 @@ services: - ./target/x86_64-unknown-linux-musl/release/rustfs:/app/rustfs # - ./data/node3:/data command: "/app/rustfs" - - - - - - 2025-03-14T05:23:15.661154Z INFO ecstore::disk::os: reliable_rename rm dst failed. src_file_path: "/data/rustfs1/.rustfs.sys/tmp/c7fabb9c-48c8-4827-b5e2-13271c3867c3x1741929793/part.38", dst_file_path: "/data/rustfs1/.rustfs.sys/multipart/494d877741f5e87d5160dc4e1bd4fbdacda64559ea0b7d16cdbeed61f252b98f/a83dc20f-e73a-46d0-a02b-11b330ba6e7ex1741929773056730169/641d3efd-cca0-418e-983b-ca2d47652900/part.38", base_dir: "/data/rustfs1/.rustfs.sys/multipart", err: Os { code: 2, kind: NotFound, message: "No such file or directory" } - at ecstore/src/disk/os.rs:144 - - 2025-03-14T05:23:15.953116Z INFO ecstore::disk::os: reliable_rename rm dst failed. src_file_path: "/data/rustfs3/.rustfs.sys/tmp/e712821f-bc3f-4ffe-8a0c-0daa379d00d4x1741929793/part.39", dst_file_path: "/data/rustfs3/.rustfs.sys/multipart/494d877741f5e87d5160dc4e1bd4fbdacda64559ea0b7d16cdbeed61f252b98f/a83dc20f-e73a-46d0-a02b-11b330ba6e7ex1741929773056730169/641d3efd-cca0-418e-983b-ca2d47652900/part.39", base_dir: "/data/rustfs3/.rustfs.sys/multipart", err: Os { code: 2, kind: NotFound, message: "No such file or directory" } - at ecstore/src/disk/os.rs:144 - - 2025-03-14T05:23:15.953218Z INFO ecstore::disk::os: reliable_rename rm dst failed. src_file_path: "/data/rustfs2/.rustfs.sys/tmp/e712821f-bc3f-4ffe-8a0c-0daa379d00d4x1741929793/part.39", dst_file_path: "/data/rustfs2/.rustfs.sys/multipart/494d877741f5e87d5160dc4e1bd4fbdacda64559ea0b7d16cdbeed61f252b98f/a83dc20f-e73a-46d0-a02b-11b330ba6e7ex1741929773056730169/641d3efd-cca0-418e-983b-ca2d47652900/part.39", base_dir: "/data/rustfs2/.rustfs.sys/multipart", err: Os { code: 2, kind: NotFound, message: "No such file or directory" } - at ecstore/src/disk/os.rs:144 \ No newline at end of file From 59c008b0eeb390eb1f88e48fb6b6c9cd9e88276c Mon Sep 17 00:00:00 2001 From: weisd Date: Mon, 21 Apr 2025 17:29:00 +0800 Subject: [PATCH 4/5] fix pool select idx --- Cargo.toml | 60 ++++++++++++++++++++++---------- ecstore/src/file_meta.rs | 6 ++-- ecstore/src/heal/data_scanner.rs | 1 - ecstore/src/store.rs | 41 +++++++++++----------- 4 files changed, 65 insertions(+), 43 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 381305c48..e2ef3b562 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,17 +1,17 @@ [workspace] members = [ - "madmin", # Management dashboard and admin API interface - "rustfs", # Core file system implementation - "ecstore", # Erasure coding storage implementation - "e2e_test", # End-to-end test suite - "common/common", # Shared utilities and data structures - "common/lock", # Distributed locking implementation - "common/protos", # Protocol buffer definitions + "madmin", # Management dashboard and admin API interface + "rustfs", # Core file system implementation + "ecstore", # Erasure coding storage implementation + "e2e_test", # End-to-end test suite + "common/common", # Shared utilities and data structures + "common/lock", # Distributed locking implementation + "common/protos", # Protocol buffer definitions "common/workers", # Worker thread pools and task scheduling - "iam", # Identity and Access Management - "crypto", # Cryptography and security features + "iam", # Identity and Access Management + "crypto", # Cryptography and security features "cli/rustfs-gui", # Graphical user interface client - "crates/obs", # Observability utilities + "crates/obs", # Observability utilities "s3select/api", "s3select/query", "appauth", @@ -67,7 +67,11 @@ http = "1.3.1" http-body = "1.0.1" humantime = "2.2.0" jsonwebtoken = "9.3.1" -keyring = { version = "3.6.2", features = ["apple-native", "windows-native", "sync-secret-service"] } +keyring = { version = "3.6.2", features = [ + "apple-native", + "windows-native", + "sync-secret-service", +] } lock = { path = "./common/lock" } lazy_static = "1.5.0" libsystemd = { version = "0.7" } @@ -77,12 +81,17 @@ md-5 = "0.10.6" mime = "0.3.17" netif = "0.1.6" opentelemetry = { version = "0.29.1" } -opentelemetry-appender-tracing = { version = "0.29.1", features = ["experimental_use_tracing_span_context", "experimental_metadata_attributes"] } +opentelemetry-appender-tracing = { version = "0.29.1", features = [ + "experimental_use_tracing_span_context", + "experimental_metadata_attributes", +] } opentelemetry_sdk = { version = "0.29" } opentelemetry-stdout = { version = "0.29.0" } opentelemetry-otlp = { version = "0.29" } opentelemetry-prometheus = { version = "0.29.1" } -opentelemetry-semantic-conventions = { version = "0.29.0", features = ["semconv_experimental"] } +opentelemetry-semantic-conventions = { version = "0.29.0", features = [ + "semconv_experimental", +] } pin-project-lite = "0.2" prometheus = "0.14.0" # pin-utils = "0.1.0" @@ -93,8 +102,19 @@ protobuf = "3.7" protos = { path = "./common/protos" } rand = "0.8.5" rdkafka = { version = "0.37", features = ["tokio"] } -reqwest = { version = "0.12.15", default-features = false, features = ["rustls-tls", "charset", "http2", "macos-system-configuration", "stream", "json", "blocking"] } -rfd = { version = "0.15.3", default-features = false, features = ["xdg-portal", "tokio"] } +reqwest = { version = "0.12.15", default-features = false, features = [ + "rustls-tls", + "charset", + "http2", + "macos-system-configuration", + "stream", + "json", + "blocking", +] } +rfd = { version = "0.15.3", default-features = false, features = [ + "xdg-portal", + "tokio", +] } rmp = "0.8.14" rmp-serde = "1.3.0" rustfs-obs = { path = "crates/obs", version = "0.0.1" } @@ -154,6 +174,10 @@ inherits = "dev" inherits = "dev" [profile.release] -opt-level = 3 # Optimization Level (0-3) -lto = true # Optimize when linking -codegen-units = 1 # Reduce code generation units to improve optimization +opt-level = 3 +lto = "thin" + +[profile.production] +inherits = "release" +lto = "fat" +codegen-units = 1 diff --git a/ecstore/src/file_meta.rs b/ecstore/src/file_meta.rs index c9558578e..e4da3882a 100644 --- a/ecstore/src/file_meta.rs +++ b/ecstore/src/file_meta.rs @@ -99,7 +99,7 @@ impl FileMeta { Ok((bin_len, &buf[5..])) } - #[tracing::instrument] + pub fn unmarshal_msg(&mut self, buf: &[u8]) -> Result { let i = buf.len() as u64; @@ -711,7 +711,6 @@ impl FileMetaVersion { Ok(data_dir) } - #[tracing::instrument] pub fn unmarshal_msg(&mut self, buf: &[u8]) -> Result { let mut cur = Cursor::new(buf); @@ -998,7 +997,7 @@ impl FileMetaVersionHeader { Ok(wr) } - #[tracing::instrument] + pub fn unmarshal_msg(&mut self, buf: &[u8]) -> Result { let mut cur = Cursor::new(buf); let alen = rmp::decode::read_array_len(&mut cur)?; @@ -1144,7 +1143,6 @@ pub struct MetaObject { } impl MetaObject { - #[tracing::instrument] pub fn unmarshal_msg(&mut self, buf: &[u8]) -> Result { let mut cur = Cursor::new(buf); diff --git a/ecstore/src/heal/data_scanner.rs b/ecstore/src/heal/data_scanner.rs index 315b21fef..78c5b9f64 100644 --- a/ecstore/src/heal/data_scanner.rs +++ b/ecstore/src/heal/data_scanner.rs @@ -344,7 +344,6 @@ impl CurrentScannerCycle { Ok(result) } - #[tracing::instrument] pub fn unmarshal_msg(&mut self, buf: &[u8]) -> Result { let mut cur = Cursor::new(buf); diff --git a/ecstore/src/store.rs b/ecstore/src/store.rs index 4bd2067f1..927672afb 100644 --- a/ecstore/src/store.rs +++ b/ecstore/src/store.rs @@ -695,7 +695,7 @@ impl ECStore { let at = a.object_info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH); let bt = b.object_info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH); - at.cmp(&bt) + bt.cmp(&at) }); let mut def_pool = PoolObjInfo::default(); @@ -911,29 +911,30 @@ impl ECStore { // TODO: test order idx_res.sort_by(|a, b| { - if let Some(obj1) = &a.res { - if let Some(obj2) = &b.res { - let cmp = obj1.mod_time.cmp(&obj2.mod_time); - match cmp { - // eq use lowest - Ordering::Equal => { - if a.idx < b.idx { - Ordering::Greater - } else { - Ordering::Less - } - } - _ => cmp, - } - } else { - Ordering::Greater - } + let a_mod = if let Some(o1) = &a.res { + o1.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH) } else { - Ordering::Less + OffsetDateTime::UNIX_EPOCH + }; + + let b_mod = if let Some(o2) = &b.res { + o2.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH) + } else { + OffsetDateTime::UNIX_EPOCH + }; + + if a_mod == b_mod { + if a.idx < b.idx { + return Ordering::Greater; + } else { + return Ordering::Less; + } } + + b_mod.cmp(&a_mod) }); - for res in idx_res { + for res in idx_res.into_iter() { if let Some(obj) = res.res { return Ok((obj, res.idx)); } From 18b50c17522f09a3f5787733682d98061d7166c3 Mon Sep 17 00:00:00 2001 From: weisd Date: Mon, 21 Apr 2025 17:33:50 +0800 Subject: [PATCH 5/5] otel debug filter tower --- crates/obs/src/telemetry.rs | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/crates/obs/src/telemetry.rs b/crates/obs/src/telemetry.rs index 82c93de73..8efa869dd 100644 --- a/crates/obs/src/telemetry.rs +++ b/crates/obs/src/telemetry.rs @@ -254,7 +254,11 @@ pub fn init_telemetry(config: &OtelConfig) -> OtelGuard { let filter_otel = match logger_level { "trace" | "debug" => { info!("OpenTelemetry tracing initialized with level: {}", logger_level); - EnvFilter::new(logger_level) + let mut filter = EnvFilter::new(logger_level); + for directive in ["hyper", "tonic", "h2", "reqwest", "tower"] { + filter = filter.add_directive(format!("{}=off", directive).parse().unwrap()); + } + filter } _ => { let mut filter = EnvFilter::new(logger_level);