From d4863fc185a30949324fe5616d712b19df8f8a09 Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Mon, 12 Oct 2026 08:40:47 -0700 Subject: [PATCH 1/2] fix(io): propagate User-Agent options to Jindo OSS --- crates/paimon/src/io/storage_jindo.rs | 168 +++++++++++++++++++++++++- 1 file changed, 164 insertions(+), 4 deletions(-) diff --git a/crates/paimon/src/io/storage_jindo.rs b/crates/paimon/src/io/storage_jindo.rs index bfda2a3dc..96437f286 100644 --- a/crates/paimon/src/io/storage_jindo.rs +++ b/crates/paimon/src/io/storage_jindo.rs @@ -30,6 +30,10 @@ use opendal::{Error as OpendalError, OperationContext, Operator, Result as Opend use tokio::sync::{OnceCell, Semaphore}; use crate::error::{Error, JINDO_FORK_ERROR}; +use crate::io::user_agent::{ + DLF_ACCESS_TRACKING_EXTENDED_INFO, OSS_USER_AGENT_EXTENDED, OSS_USER_AGENT_FEATURES, + OSS_USER_AGENT_MODULE, USER_AGENT_EXTENDED, USER_AGENT_FEATURES, USER_AGENT_MODULE, +}; use crate::Result; const OSS_IMPL: &str = "fs.oss.impl"; @@ -159,10 +163,7 @@ pub(crate) fn jindo_config_parse(mut props: HashMap) -> Result DEFAULT_JINDO_MAX_CONCURRENT_READS, }; props.remove(OSS_IMPL); - props.insert( - "fs.oss.user.agent.features".to_string(), - "paimon-rust".to_string(), - ); + configure_jindo_user_agent(&mut props); props .entry("logger.appender".to_string()) .or_insert_with(|| "file".to_string()); @@ -184,6 +185,50 @@ pub(crate) fn jindo_config_parse(mut props: HashMap) -> Result) { + let effective = |oss_key: &str, common_key: &str| { + props + .get(oss_key) + .filter(|value| !value.trim().is_empty()) + .or_else(|| { + props + .get(common_key) + .filter(|value| !value.trim().is_empty()) + }) + .map(|value| value.trim().to_string()) + }; + let module = effective(OSS_USER_AGENT_MODULE, USER_AGENT_MODULE); + let features = effective(OSS_USER_AGENT_FEATURES, USER_AGENT_FEATURES); + let extended = effective(OSS_USER_AGENT_EXTENDED, USER_AGENT_EXTENDED); + let access_tracking = props + .get(DLF_ACCESS_TRACKING_EXTENDED_INFO) + .map(|value| value.trim().to_string()) + .filter(|value| !value.is_empty()); + + if let Some(module) = module { + props.insert(OSS_USER_AGENT_MODULE.to_string(), module); + } + let mut features: Vec<&str> = features + .as_deref() + .map(str::split_whitespace) + .into_iter() + .flatten() + .collect(); + if !features.contains(&"paimon-rust") { + features.insert(0, "paimon-rust"); + } + props.insert(OSS_USER_AGENT_FEATURES.to_string(), features.join(" ")); + + let extended = [extended, access_tracking] + .into_iter() + .flatten() + .collect::>() + .join(" "); + if !extended.is_empty() { + props.insert(OSS_USER_AGENT_EXTENDED.to_string(), extended); + } +} + pub(crate) fn jindo_config_build(config: &JindoStorageConfig, bucket: &str) -> Result { let builder = JindoBuilder { config: Some(config.clone()), @@ -1257,6 +1302,91 @@ mod tests { ); assert!(!config.properties.contains_key(OSS_IMPL)); assert!(!config.properties.contains_key(JINDO_MAX_CONCURRENT_READS)); + assert_eq!( + config + .properties + .get(OSS_USER_AGENT_FEATURES) + .map(String::as_str), + Some("paimon-rust") + ); + } + + #[test] + fn test_jindo_user_agent_from_catalog_options() { + let config = jindo_config_parse(HashMap::from([ + ( + USER_AGENT_MODULE.to_string(), + "training-app/1.0".to_string(), + ), + ( + USER_AGENT_FEATURES.to_string(), + "sdk-v2 training".to_string(), + ), + (USER_AGENT_EXTENDED.to_string(), "env=prod".to_string()), + ( + DLF_ACCESS_TRACKING_EXTENDED_INFO.to_string(), + "username=test@example.com".to_string(), + ), + ])) + .unwrap(); + assert_eq!( + config + .properties + .get(OSS_USER_AGENT_MODULE) + .map(String::as_str), + Some("training-app/1.0") + ); + assert_eq!( + config + .properties + .get(OSS_USER_AGENT_FEATURES) + .map(String::as_str), + Some("paimon-rust sdk-v2 training") + ); + assert_eq!( + config + .properties + .get(OSS_USER_AGENT_EXTENDED) + .map(String::as_str), + Some("env=prod username=test@example.com") + ); + } + + #[test] + fn test_jindo_oss_user_agent_options_take_precedence() { + let config = jindo_config_parse(HashMap::from([ + (USER_AGENT_MODULE.to_string(), "common/1.0".to_string()), + (OSS_USER_AGENT_MODULE.to_string(), "oss/1.0".to_string()), + (USER_AGENT_FEATURES.to_string(), "common".to_string()), + ( + OSS_USER_AGENT_FEATURES.to_string(), + "oss paimon-rust".to_string(), + ), + (USER_AGENT_EXTENDED.to_string(), "common-info".to_string()), + (OSS_USER_AGENT_EXTENDED.to_string(), "oss-info".to_string()), + ])) + .unwrap(); + assert_eq!( + config + .properties + .get(OSS_USER_AGENT_MODULE) + .map(String::as_str), + Some("oss/1.0") + ); + assert_eq!( + config + .properties + .get(OSS_USER_AGENT_FEATURES) + .map(String::as_str), + Some("oss paimon-rust") + ); + assert_eq!( + config + .properties + .get(OSS_USER_AGENT_EXTENDED) + .map(String::as_str), + Some("oss-info") + ); } #[test] @@ -1510,6 +1640,7 @@ mod tests { list_page_requests: Arc, range_requests: Arc, unavailable_requests: Arc, + user_agents: Arc>>, } #[cfg(target_os = "linux")] @@ -1520,6 +1651,13 @@ mod tests { headers: HeaderMap, ) -> Response { const DATA: &[u8] = b"0123456789"; + state.user_agents.lock().unwrap().push( + headers + .get("user-agent") + .and_then(|value| value.to_str().ok()) + .unwrap_or_default() + .to_string(), + ); if method == Method::HEAD { if uri.path().ends_with("/collision") { return Response::builder() @@ -1725,6 +1863,15 @@ mod tests { "true".to_string(), ), ("fs.oss.retry.count".to_string(), "0".to_string()), + ( + USER_AGENT_MODULE.to_string(), + "training-app/1.0".to_string(), + ), + ( + USER_AGENT_FEATURES.to_string(), + "sdk-v2 training".to_string(), + ), + (USER_AGENT_EXTENDED.to_string(), "env=test".to_string()), ]); let config = jindo_config_parse(properties.clone()).unwrap(); let operator = jindo_config_build(&config, "jindo-test-bucket").unwrap(); @@ -1787,6 +1934,19 @@ mod tests { assert_eq!(state.list_page_requests.load(Ordering::SeqCst), 3); assert_eq!(state.range_requests.load(Ordering::SeqCst), 1); assert_eq!(state.unavailable_requests.load(Ordering::SeqCst), 1); + let user_agents = state.user_agents.lock().unwrap(); + assert!(!user_agents.is_empty()); + assert!( + user_agents.iter().all(|value| { + value.contains("training-app/1.0") + && value.contains("paimon-rust") + && value.contains("sdk-v2") + && value.contains("training") + && value.contains("env=test") + }), + "unexpected Jindo User-Agent: {user_agents:?}" + ); + drop(user_agents); let fork_operator = jindo_config_build(&config, "jindo-test-bucket").unwrap(); assert_operator_rejects_reuse_after_fork(fork_operator).await; From 32855d3a45939ca764cac0f9cf5bd28a062bb2a0 Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Tue, 13 Oct 2026 00:45:27 -0700 Subject: [PATCH 2/2] test(io): release Jindo UA assertion lock before await --- crates/paimon/src/io/storage_jindo.rs | 27 ++++++++++++++------------- 1 file changed, 14 insertions(+), 13 deletions(-) diff --git a/crates/paimon/src/io/storage_jindo.rs b/crates/paimon/src/io/storage_jindo.rs index 96437f286..2e080aad7 100644 --- a/crates/paimon/src/io/storage_jindo.rs +++ b/crates/paimon/src/io/storage_jindo.rs @@ -1934,19 +1934,20 @@ mod tests { assert_eq!(state.list_page_requests.load(Ordering::SeqCst), 3); assert_eq!(state.range_requests.load(Ordering::SeqCst), 1); assert_eq!(state.unavailable_requests.load(Ordering::SeqCst), 1); - let user_agents = state.user_agents.lock().unwrap(); - assert!(!user_agents.is_empty()); - assert!( - user_agents.iter().all(|value| { - value.contains("training-app/1.0") - && value.contains("paimon-rust") - && value.contains("sdk-v2") - && value.contains("training") - && value.contains("env=test") - }), - "unexpected Jindo User-Agent: {user_agents:?}" - ); - drop(user_agents); + { + let user_agents = state.user_agents.lock().unwrap(); + assert!(!user_agents.is_empty()); + assert!( + user_agents.iter().all(|value| { + value.contains("training-app/1.0") + && value.contains("paimon-rust") + && value.contains("sdk-v2") + && value.contains("training") + && value.contains("env=test") + }), + "unexpected Jindo User-Agent: {user_agents:?}" + ); + } let fork_operator = jindo_config_build(&config, "jindo-test-bucket").unwrap(); assert_operator_rejects_reuse_after_fork(fork_operator).await;