1212// See the License for the specific language governing permissions and
1313// limitations under the License.
1414
15+ use std:: collections:: BTreeMap ;
16+ use std:: sync:: Arc ;
17+ use std:: time:: Duration ;
18+
19+ use async_channel:: bounded;
1520use databend_common_base:: base:: tokio;
1621use databend_common_exception:: Result ;
1722use databend_common_tracing:: convert_to_batch;
23+ use databend_common_tracing:: Config ;
1824use databend_common_tracing:: RemoteLog ;
25+ use databend_common_tracing:: RemoteLogBuffer ;
1926use databend_common_tracing:: RemoteLogElement ;
27+ use databend_common_tracing:: RemoteLogGuard ;
28+ use log:: Level ;
29+ use log:: Record ;
2030use opendal:: services;
2131use opendal:: Operator ;
32+ use serde_json:: Value ;
33+
34+ fn setup ( ) -> Result < ( RemoteLog , Box < RemoteLogGuard > ) > {
35+ let mut labels = BTreeMap :: new ( ) ;
36+ labels. insert ( "cluster_id" . into ( ) , "cluster_id1" . into ( ) ) ;
37+ labels. insert ( "node_id" . into ( ) , "node_id1" . into ( ) ) ;
38+ let ( remote_log, guard) = RemoteLog :: new ( & labels, & Config :: default ( ) ) ?;
39+
40+ Ok ( ( remote_log, guard) )
41+ }
42+
43+ fn get_remote_log_elements ( ) -> RemoteLogElement {
44+ RemoteLogElement {
45+ timestamp : chrono:: Local :: now ( ) . timestamp_micros ( ) ,
46+ path : "databend_query::interpreters::common::query_log: query_log.rs:71" . to_string ( ) ,
47+ target : "databend::log::query" . to_string ( ) ,
48+ cluster_id : "test_cluster" . to_string ( ) ,
49+ node_id : "izs9przqhAN4n5hbJanJm2" . to_string ( ) ,
50+ query_id : Some ( "89ad07ad-83fe-4424-8005-4c5b318a7212" . to_string ( ) ) ,
51+ warehouse_id : None ,
52+ log_level : "INFO" . to_string ( ) ,
53+ fields : r#"{"message":"test"}"# . to_string ( ) ,
54+ }
55+ }
56+
57+ #[ test]
58+ fn test_basic_parse ( ) -> Result < ( ) > {
59+ let ( remote_log, _guard) = setup ( ) ?;
60+ let record = Record :: builder ( )
61+ . args ( format_args ! ( "begin to list files" ) )
62+ . level ( Level :: Info )
63+ . target ( "databend_query::sessions::query_ctx" )
64+ . module_path ( Some ( "databend_query::sessions::query_ctx" ) )
65+ . file ( Some ( "query_ctx.rs" ) )
66+ . line ( Some ( 656 ) )
67+ . build ( ) ;
68+
69+ let remote_log_element = remote_log. prepare_log_element ( & record) ;
70+
71+ assert_eq ! ( remote_log_element. cluster_id, "cluster_id1" ) ;
72+ assert_eq ! ( remote_log_element. node_id, "node_id1" ) ;
73+ assert_eq ! (
74+ remote_log_element. path,
75+ "databend_query::sessions::query_ctx: query_ctx.rs:656"
76+ ) ;
77+
78+ let fields: Value = serde_json:: from_str ( & remote_log_element. fields ) ?;
79+
80+ assert_eq ! ( fields[ "message" ] , "begin to list files" ) ;
81+
82+ Ok ( ( ) )
83+ }
2284
2385#[ test]
24- fn test_convert_to_batch ( ) {
25- let _ = convert_to_batch ( get_remote_log ( ) ) . unwrap ( ) ;
86+ fn test_convert_to_batch ( ) -> Result < ( ) > {
87+ let elements = vec ! [ get_remote_log_elements( ) ] ;
88+ let _ = convert_to_batch ( elements) . unwrap ( ) ;
89+ Ok ( ( ) )
90+ }
91+
92+ #[ tokio:: test( flavor = "multi_thread" , worker_threads = 1 ) ]
93+ async fn test_buffer_flush_with_buffer_limit ( ) -> Result < ( ) > {
94+ let ( tx, rx) = bounded ( 10 ) ;
95+ let interval = Duration :: from_secs ( 100 ) . as_micros ( ) as u64 ;
96+ let buffer = Arc :: new ( RemoteLogBuffer :: new ( tx. clone ( ) , interval) ) ;
97+ for _i in 0 ..5005 {
98+ buffer. log ( get_remote_log_elements ( ) ) ?
99+ }
100+ let res = rx. recv ( ) . await . unwrap ( ) ;
101+ assert_eq ! ( res. len( ) , 5000 ) ;
102+ Ok ( ( ) )
103+ }
104+
105+ #[ tokio:: test( flavor = "multi_thread" , worker_threads = 1 ) ]
106+ async fn test_buffer_flush_with_buffer_interval ( ) -> Result < ( ) > {
107+ let ( tx, rx) = bounded ( 10 ) ;
108+ let interval = Duration :: from_secs ( 1 ) . as_micros ( ) as u64 ;
109+ let buffer = Arc :: new ( RemoteLogBuffer :: new ( tx. clone ( ) , interval) ) ;
110+ for _i in 0 ..5 {
111+ buffer. log ( get_remote_log_elements ( ) ) ?
112+ }
113+ tokio:: time:: sleep ( Duration :: from_secs ( 1 ) ) . await ;
114+ buffer. log ( get_remote_log_elements ( ) ) ?;
115+ let res = rx. recv ( ) . await . unwrap ( ) ;
116+ assert_eq ! ( res. len( ) , 6 ) ;
117+ Ok ( ( ) )
118+ }
119+
120+ #[ tokio:: test( flavor = "multi_thread" , worker_threads = 1 ) ]
121+ async fn test_buffer_flush_with_force_collect ( ) -> Result < ( ) > {
122+ // This simulates guard is dropped and collect is called
123+ let ( tx, rx) = bounded ( 10 ) ;
124+ let interval = Duration :: from_secs ( 100 ) . as_micros ( ) as u64 ;
125+ let buffer = Arc :: new ( RemoteLogBuffer :: new ( tx. clone ( ) , interval) ) ;
126+ for _i in 0 ..500 {
127+ buffer. log ( get_remote_log_elements ( ) ) ?
128+ }
129+ buffer. collect ( ) ?;
130+ let res = rx. recv ( ) . await . unwrap ( ) ;
131+ assert_eq ! ( res. len( ) , 500 ) ;
132+ Ok ( ( ) )
26133}
27134
28135#[ tokio:: test( flavor = "multi_thread" , worker_threads = 1 ) ]
@@ -31,35 +138,12 @@ async fn test_do_flush() -> Result<()> {
31138 let op = Operator :: new ( builder) ?. finish ( ) ;
32139
33140 let path = "test_logs.parquet" ;
34-
35- let result = RemoteLog :: do_flush ( op. clone ( ) , get_remote_log ( ) , path) . await ;
141+ let elements = vec ! [ get_remote_log_elements ( ) ] ;
142+ let result = RemoteLog :: do_flush ( op. clone ( ) , elements , path) . await ;
36143 assert ! ( result. is_ok( ) ) ;
37144
38145 let exists = op. exists ( path) . await ?;
39146 assert ! ( exists) ;
40147
41148 Ok ( ( ) )
42149}
43-
44- fn get_remote_log ( ) -> Vec < RemoteLogElement > {
45- vec ! [
46- RemoteLogElement {
47- timestamp: 1741680446186595 ,
48- path: "test_path" . to_string( ) ,
49- messages: r#"{"message":"MetaGrpcClient(0.0.0.0:9191)::worker spawned"}"# . to_string( ) ,
50- cluster_id: "cluster_id" . to_string( ) ,
51- node_id: "node_id" . to_string( ) ,
52- query_id: None ,
53- log_level: "INFO" . to_string( ) ,
54- } ,
55- RemoteLogElement {
56- timestamp: 1741680446186596 ,
57- path: "test_path2" . to_string( ) ,
58- messages: r#"{"conf":"RpcClientConf { endpoints: [\"0.0.0.0:9191\"], username: \"root\", password: \"root\", tls_conf: None, timeout: Some(60s), auto_sync_interval: Some(60s), unhealthy_endpoint_evict_time: 120s }","message":"use remote meta"}"# . to_string( ) ,
59- cluster_id: "cluster_id" . to_string( ) ,
60- node_id: "node_id" . to_string( ) ,
61- query_id: Some ( "query_id" . to_string( ) ) ,
62- log_level: "DEBUG" . to_string( ) ,
63- } ,
64- ]
65- }
0 commit comments