risingwave_rt/
logger.rs

1// Copyright 2023 RisingWave Labs
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::borrow::Cow;
16use std::env;
17use std::path::PathBuf;
18
19use either::Either;
20use fastrace_opentelemetry::OpenTelemetryReporter;
21use opentelemetry::InstrumentationScope;
22use opentelemetry::trace::TracerProvider;
23use opentelemetry_otlp::SpanExporter;
24use opentelemetry_sdk::Resource;
25use opentelemetry_sdk::trace::TracerProviderBuilder;
26use risingwave_common::util::deployment::Deployment;
27use risingwave_common::util::env_var::env_var_is_true;
28use risingwave_common::util::query_log::*;
29use risingwave_common::util::tracing::layer::set_toggle_otel_layer_fn;
30use thiserror_ext::AsReport;
31use tracing::level_filters::LevelFilter as Level;
32use tracing_subscriber::filter::Targets;
33use tracing_subscriber::fmt::FormatFields;
34use tracing_subscriber::fmt::format::DefaultFields;
35use tracing_subscriber::fmt::time::OffsetTime;
36use tracing_subscriber::layer::SubscriberExt;
37use tracing_subscriber::prelude::*;
38use tracing_subscriber::{EnvFilter, filter, reload};
39
40/// Parse a comma-separated list of `key=value` pairs into `OpenTelemetry` `KeyValue` attributes.
41///
42/// - Splits by comma to get pairs.
43/// - Each pair must contain exactly one `=`.
44/// - Trims whitespace around key and value.
45/// - Ignores invalid pairs but logs a warning.
46/// - Returns an empty vec for empty or unset input.
47fn parse_extra_tracing_attributes(input: &str) -> Vec<opentelemetry::KeyValue> {
48    use opentelemetry::KeyValue;
49
50    if input.trim().is_empty() {
51        return vec![];
52    }
53
54    input
55        .split(',')
56        .filter_map(|pair| {
57            let pair = pair.trim();
58            if pair.is_empty() {
59                return None;
60            }
61            let parts: Vec<&str> = pair.splitn(2, '=').collect();
62            if parts.len() != 2 {
63                eprintln!("warning: ignoring invalid tracing attribute pair (no '='): {pair:?}");
64                return None;
65            }
66            let key = parts[0].trim();
67            let value = parts[1].trim();
68            if key.is_empty() {
69                eprintln!("warning: ignoring tracing attribute with empty key: {pair:?}");
70                return None;
71            }
72            Some(KeyValue::new(key.to_owned(), value.to_owned()))
73        })
74        .collect()
75}
76
77pub struct LoggerSettings {
78    /// The name of the service. Used to identify the service in distributed tracing.
79    name: String,
80    /// Enable tokio console output.
81    enable_tokio_console: bool,
82    /// Enable colorful output in console.
83    colorful: bool,
84    /// Output to `stderr` instead of `stdout`.
85    stderr: bool,
86    /// Whether to include thread name in the log.
87    with_thread_name: bool,
88    /// Override target settings.
89    targets: Vec<(String, tracing::metadata::LevelFilter)>,
90    /// Override the default level.
91    default_level: Option<tracing::metadata::LevelFilter>,
92    /// The endpoint of the tracing collector in OTLP gRPC protocol.
93    tracing_endpoint: Option<String>,
94    /// Extra key/value attributes to attach to `OpenTelemetry` trace resources.
95    extra_tracing_attributes: Vec<opentelemetry::KeyValue>,
96}
97
98impl Default for LoggerSettings {
99    fn default() -> Self {
100        Self::new("risingwave")
101    }
102}
103
104impl LoggerSettings {
105    /// Create a new logger settings from the given command-line options.
106    ///
107    /// If env var `RW_TRACING_ENDPOINT` is not set, the meta address will be used
108    /// as the default tracing endpoint, which means that the embedded tracing
109    /// collector will be used.
110    pub fn from_opts<O: risingwave_common::opts::Opts>(opts: &O) -> Self {
111        let mut settings = Self::new(O::name());
112        if settings.tracing_endpoint.is_none() // no explicit endpoint
113            && let Some(addr) = opts.meta_addr().exactly_one()
114        // meta address is valid
115        {
116            // Use embedded collector in the meta service.
117            // TODO: when there's multiple meta nodes for high availability, we may send
118            // to a wrong node here.
119            settings.tracing_endpoint = Some(addr.to_string());
120        }
121        settings
122    }
123
124    /// Create a new logger settings with the given service name.
125    pub fn new(name: impl Into<String>) -> Self {
126        Self {
127            name: name.into(),
128            enable_tokio_console: false,
129            colorful: console::colors_enabled_stderr() && console::colors_enabled(),
130            stderr: false,
131            with_thread_name: false,
132            targets: vec![],
133            default_level: None,
134            tracing_endpoint: std::env::var("RW_TRACING_ENDPOINT").ok(),
135            extra_tracing_attributes: std::env::var("RW_TRACING_EXTRA_ATTRIBUTES")
136                .ok()
137                .map(|v| parse_extra_tracing_attributes(&v))
138                .unwrap_or_default(),
139        }
140    }
141
142    /// Enable tokio console output.
143    pub fn tokio_console(mut self, enabled: bool) -> Self {
144        self.enable_tokio_console = enabled;
145        self
146    }
147
148    /// Output to `stderr` instead of `stdout`.
149    pub fn stderr(mut self, enabled: bool) -> Self {
150        self.stderr = enabled;
151        self
152    }
153
154    /// Whether to include thread name in the log.
155    pub fn with_thread_name(mut self, enabled: bool) -> Self {
156        self.with_thread_name = enabled;
157        self
158    }
159
160    /// Overrides the default target settings.
161    pub fn with_target(
162        mut self,
163        target: impl Into<String>,
164        level: impl Into<tracing::metadata::LevelFilter>,
165    ) -> Self {
166        self.targets.push((target.into(), level.into()));
167        self
168    }
169
170    /// Overrides the default level.
171    pub fn with_default(mut self, level: impl Into<tracing::metadata::LevelFilter>) -> Self {
172        self.default_level = Some(level.into());
173        self
174    }
175
176    /// Overrides the tracing endpoint.
177    pub fn with_tracing_endpoint(mut self, endpoint: impl Into<String>) -> Self {
178        self.tracing_endpoint = Some(endpoint.into());
179        self
180    }
181}
182
183/// Create a filter that disables all events or spans.
184fn disabled_filter() -> filter::Targets {
185    filter::Targets::new()
186}
187
188/// Init logger for RisingWave binaries.
189///
190/// ## Environment variables to configure logger dynamically
191///
192/// ### `RUST_LOG`
193///
194/// Overrides default level and tracing targets of the fmt layer (formatting and
195/// logging to `stdout` or `stderr`).
196///
197/// Note that only verbosity levels below or equal to `DEBUG` are effective in
198/// release builds.
199///
200/// e.g.,
201/// ```bash
202/// RUST_LOG="info,risingwave_stream=debug,events=debug"
203/// ```
204///
205/// ### `RW_QUERY_LOG_PATH`
206///
207/// Configures the path to generate query log.
208///
209/// If it is set,
210/// - Dump logs of all SQLs, i.e., tracing target [`PGWIRE_QUERY_LOG`] to
211///   `RW_QUERY_LOG_PATH/query.log`.
212/// - Dump slow queries, i.e., tracing target [`PGWIRE_SLOW_QUERY_LOG`] to
213///   `RW_QUERY_LOG_PATH/slow_query.log`.
214///
215/// Note:
216/// To enable query log in the fmt layer (slow query is included by default), set
217/// ```bash
218/// RUST_LOG="pgwire_query_log=info"
219/// ```
220///
221/// `RW_QUERY_LOG_TRUNCATE_LEN` configures the max length of the SQLs logged in the query log,
222/// to avoid the log file growing too large. The default value is 1024 in production.
223///
224/// ### `ENABLE_PRETTY_LOG`
225///
226/// If it is set to `true`, enable pretty log output, which contains line numbers and prints spans in multiple lines.
227/// This can be helpful for development and debugging.
228///
229/// Hint: Also turn off other uninteresting logs to make the most of the pretty log.
230/// e.g.,
231/// ```bash
232/// RUST_LOG="risingwave_storage::hummock::event_handler=off,batch_execute=off,risingwave_batch::task=off" ENABLE_PRETTY_LOG=true risedev d
233/// ```
234pub fn init_risingwave_logger(settings: LoggerSettings) {
235    let deployment = Deployment::current();
236
237    // Default timer for logging with local time offset.
238    let default_timer = OffsetTime::local_rfc_3339().unwrap_or_else(|e| {
239        println!(
240            "failed to get local time offset, falling back to UTC: {}",
241            e.as_report()
242        );
243        OffsetTime::new(
244            time::UtcOffset::UTC,
245            time::format_description::well_known::Rfc3339,
246        )
247    });
248
249    // Default filter for logging to stdout and tracing.
250    let default_filter = {
251        let mut filter = filter::Targets::new();
252
253        // Configure levels for some RisingWave crates. Can still be overridden by `RUST_LOG`.
254        // Other RisingWave crates like `stream` and `storage` will follow the default level.
255        filter = filter
256            .with_target("auto_schema_change", Level::INFO)
257            .with_target("risingwave_sqlparser", Level::INFO)
258            .with_target("risingwave_connector_node", Level::INFO)
259            .with_target("pgwire", Level::INFO)
260            .with_target(PGWIRE_QUERY_LOG, Level::OFF)
261            // debug-purposed events are disabled unless `RUST_LOG` overrides
262            .with_target("events", Level::OFF);
263
264        // Configure levels for external crates. Can still be overridden by `RUST_LOG`.
265        // Other external crates will follow the default level.
266        filter = filter
267            .with_target("foyer", Level::INFO)
268            .with_target("aws", Level::INFO)
269            .with_target("aws_config", Level::WARN)
270            .with_target("aws_endpoint", Level::WARN)
271            .with_target("aws_credential_types::cache::lazy_caching", Level::WARN)
272            .with_target("hyper", Level::WARN)
273            .with_target("h2", Level::WARN)
274            .with_target("tower", Level::WARN)
275            .with_target("tonic", Level::WARN)
276            .with_target("isahc", Level::WARN)
277            .with_target("console_subscriber", Level::WARN)
278            .with_target("reqwest", Level::WARN)
279            .with_target("sled", Level::INFO)
280            .with_target("cranelift", Level::INFO)
281            .with_target("wasmtime", Level::INFO)
282            .with_target("sqlx", Level::WARN)
283            .with_target("opendal", Level::INFO)
284            .with_target("reqsign", Level::INFO)
285            .with_target("jni", Level::INFO)
286            .with_target("async_nats", Level::WARN);
287
288        // For all other crates, apply default level depending on the deployment and `debug_assertions` flag.
289        let default_level = match deployment {
290            Deployment::Ci => Level::INFO,
291            _ => {
292                if cfg!(debug_assertions) {
293                    Level::DEBUG
294                } else {
295                    Level::INFO
296                }
297            }
298        };
299        filter = filter.with_default(default_level);
300
301        // Overrides from settings.
302        filter = filter.with_targets(settings.targets);
303        if let Some(default_level) = settings.default_level {
304            filter = filter.with_default(default_level);
305        }
306
307        // Overrides from env var.
308        if let Ok(rust_log) = std::env::var(EnvFilter::DEFAULT_ENV)
309            && !rust_log.is_empty()
310        {
311            let rust_log_targets: Targets = rust_log.parse().expect("failed to parse `RUST_LOG`");
312            if let Some(default_level) = rust_log_targets.default_level() {
313                filter = filter.with_default(default_level);
314            }
315            filter = filter.with_targets(rust_log_targets)
316        };
317
318        filter
319    };
320
321    let mut layers = vec![];
322
323    // fmt layer (formatting and logging to `stdout` or `stderr`)
324    {
325        let fmt_layer = tracing_subscriber::fmt::layer()
326            .with_thread_names(settings.with_thread_name)
327            .with_timer(default_timer.clone())
328            .with_ansi(settings.colorful)
329            .with_writer(move || {
330                if settings.stderr {
331                    Either::Left(std::io::stderr())
332                } else {
333                    Either::Right(std::io::stdout())
334                }
335            });
336
337        let fmt_layer = match deployment {
338            Deployment::Ci => fmt_layer.compact().boxed(),
339            Deployment::Cloud => fmt_layer
340                .json()
341                .map_event_format(|e| e.with_current_span(false)) // avoid duplication as there's a span list field
342                .boxed(),
343            Deployment::Other => {
344                if env_var_is_true("ENABLE_PRETTY_LOG") {
345                    fmt_layer.pretty().boxed()
346                } else {
347                    fmt_layer.boxed()
348                }
349            }
350        };
351
352        layers.push(
353            fmt_layer
354                .with_filter(default_filter.clone().with_target("rw_tracing", Level::OFF)) // filter-out tracing-only events
355                .boxed(),
356        );
357    };
358
359    // If `RW_QUERY_LOG_PATH` env var is set to a directory, turn on query log files.
360    let query_log_path = std::env::var("RW_QUERY_LOG_PATH");
361    if let Ok(query_log_path) = query_log_path {
362        let query_log_path = PathBuf::from(query_log_path);
363        std::fs::create_dir_all(query_log_path.clone()).unwrap_or_else(|e| {
364            panic!(
365                "failed to create directory '{}' for query log: {}",
366                query_log_path.display(),
367                e.as_report(),
368            )
369        });
370
371        /// Newtype wrapper for `DefaultFields`.
372        ///
373        /// `fmt::Layer` will share the same `FormattedFields` extension for spans across
374        /// different layers, as long as the type of `N: FormatFields` is the same. This
375        /// will cause several problems:
376        ///
377        /// - `with_ansi(false)` does not take effect and it will follow the settings of
378        ///   the primary fmt layer installed above.
379        /// - `Span::record` will update the same `FormattedFields` multiple times,
380        ///   leading to duplicated fields.
381        ///
382        /// As a workaround, we use a newtype wrapper here to get a different type id.
383        /// The const generic parameter `SLOW` is further used to distinguish between the
384        /// query log and the slow query log.
385        #[derive(Default)]
386        struct FmtFields<const SLOW: bool>(DefaultFields);
387
388        impl<'writer, const SLOW: bool> FormatFields<'writer> for FmtFields<SLOW> {
389            fn format_fields<R: tracing_subscriber::field::RecordFields>(
390                &self,
391                writer: tracing_subscriber::fmt::format::Writer<'writer>,
392                fields: R,
393            ) -> std::fmt::Result {
394                self.0.format_fields(writer, fields)
395            }
396        }
397
398        for (file_name, target, is_slow) in [
399            ("query.log", PGWIRE_QUERY_LOG, false),
400            ("slow_query.log", PGWIRE_SLOW_QUERY_LOG, true),
401        ] {
402            let path = query_log_path.join(file_name);
403
404            let file = std::fs::OpenOptions::new()
405                .create(true)
406                .write(true)
407                .truncate(true)
408                .open(&path)
409                .unwrap_or_else(|e| {
410                    panic!("failed to create `{}`: {}", path.display(), e.as_report(),)
411                });
412
413            let layer = tracing_subscriber::fmt::layer()
414                .with_ansi(false)
415                .with_level(false)
416                .with_file(false)
417                .with_target(false)
418                .with_timer(default_timer.clone())
419                .with_thread_names(true)
420                .with_thread_ids(true)
421                .with_writer(file);
422
423            let layer = match is_slow {
424                true => layer.fmt_fields(FmtFields::<true>::default()).boxed(),
425                false => layer.fmt_fields(FmtFields::<false>::default()).boxed(),
426            };
427
428            let layer = layer.with_filter(
429                filter::Targets::new()
430                    // Root span must be enabled to provide common info like the SQL query.
431                    .with_target(PGWIRE_ROOT_SPAN_TARGET, Level::INFO)
432                    .with_target(target, Level::INFO),
433            );
434
435            layers.push(layer.boxed());
436        }
437    }
438
439    if settings.enable_tokio_console {
440        let (console_layer, server) = console_subscriber::ConsoleLayer::builder()
441            .with_default_env()
442            .build();
443        let console_layer = console_layer.with_filter(
444            filter::Targets::new()
445                .with_target("tokio", Level::TRACE)
446                .with_target("runtime", Level::TRACE),
447        );
448        layers.push(console_layer.boxed());
449        std::thread::spawn(|| {
450            tokio::runtime::Builder::new_current_thread()
451                .enable_all()
452                .build()
453                .unwrap()
454                .block_on(async move {
455                    println!("serving console subscriber");
456                    server.serve().await.unwrap();
457                });
458        });
459    };
460
461    // Tracing layer
462    #[cfg(not(madsim))]
463    if let Some(endpoint) = settings.tracing_endpoint {
464        println!("opentelemetry tracing will be exported to `{endpoint}` if enabled");
465
466        use opentelemetry::KeyValue;
467        use opentelemetry_otlp::WithExportConfig;
468        use opentelemetry_semantic_conventions::resource;
469
470        let id = format!(
471            "{}-{}",
472            hostname::get()
473                .ok()
474                .and_then(|o| o.into_string().ok())
475                .unwrap_or_default(),
476            std::process::id()
477        );
478
479        let extra_attributes = &settings.extra_tracing_attributes;
480        if !extra_attributes.is_empty() {
481            println!(
482                "extra tracing resource attributes: {:?}",
483                extra_attributes
484                    .iter()
485                    .map(|kv| format!("{}={}", kv.key, kv.value))
486                    .collect::<Vec<_>>()
487            );
488        }
489
490        let (otel_tracer, exporter) = {
491            let runtime = tokio::runtime::Builder::new_multi_thread()
492                .enable_all()
493                .thread_name("rw-otel")
494                .worker_threads(2)
495                .build()
496                .unwrap();
497            let runtime = Box::leak(Box::new(runtime));
498
499            // Installing the exporter requires a tokio runtime.
500            let _entered = runtime.enter();
501
502            // TODO(bugen): better service name
503            // https://github.com/jaegertracing/jaeger-ui/issues/336
504            let service_name = format!("{}-{}", settings.name, id);
505
506            let mut resource_attrs = vec![
507                KeyValue::new(resource::SERVICE_NAME, service_name.clone()),
508                KeyValue::new(resource::SERVICE_INSTANCE_ID, id.clone()),
509                KeyValue::new(resource::SERVICE_VERSION, env!("CARGO_PKG_VERSION")),
510                KeyValue::new(resource::PROCESS_PID, std::process::id().to_string()),
511            ];
512            resource_attrs.extend(extra_attributes.iter().cloned());
513
514            let otel_tracer = TracerProviderBuilder::default()
515                .with_batch_exporter(
516                    SpanExporter::builder()
517                        .with_tonic()
518                        .with_endpoint(&endpoint)
519                        .build()
520                        .unwrap(),
521                )
522                .with_resource(Resource::builder().with_attributes(resource_attrs).build())
523                .build()
524                .tracer(service_name);
525
526            let exporter = SpanExporter::builder()
527                .with_tonic()
528                .with_endpoint(&endpoint)
529                .with_protocol(opentelemetry_otlp::Protocol::Grpc)
530                .with_timeout(opentelemetry_otlp::OTEL_EXPORTER_OTLP_TIMEOUT_DEFAULT)
531                .build()
532                .unwrap();
533
534            (otel_tracer, exporter)
535        };
536
537        // Disable by filtering out all events or spans by default.
538        //
539        // It'll be enabled with `toggle_otel_layer` based on the system parameter `enable_tracing` later.
540        let (reload_filter, reload_handle) = reload::Layer::new(disabled_filter());
541
542        set_toggle_otel_layer_fn(move |enabled: bool| {
543            let result = reload_handle.modify(|f| {
544                *f = if enabled {
545                    default_filter.clone()
546                } else {
547                    disabled_filter()
548                }
549            });
550
551            match result {
552                Ok(_) => tracing::info!(
553                    "opentelemetry tracing {}",
554                    if enabled { "enabled" } else { "disabled" },
555                ),
556
557                Err(error) => tracing::error!(
558                    error = %error.as_report(),
559                    "failed to {} opentelemetry tracing",
560                    if enabled { "enable" } else { "disable" },
561                ),
562            }
563        });
564
565        let layer = tracing_opentelemetry::layer()
566            .with_tracer(otel_tracer)
567            .with_filter(reload_filter);
568
569        layers.push(layer.boxed());
570
571        // The reporter is used by fastrace in foyer for dynamically tail-based tracing.
572        //
573        // Code here only setup the OpenTelemetry reporter. To enable/disable the function, please use risectl.
574        //
575        // e.g.
576        //
577        // ```bash
578        // risectl hummock tiered-cache-tracing -h
579        // ```
580        let mut fastrace_resource_attrs: Vec<opentelemetry::KeyValue> =
581            vec![opentelemetry::KeyValue::new(
582                opentelemetry_semantic_conventions::resource::SERVICE_NAME,
583                format!("fastrace-{id}"),
584            )];
585        fastrace_resource_attrs.extend(extra_attributes.iter().cloned());
586
587        let reporter = OpenTelemetryReporter::new(
588            exporter,
589            Cow::Owned(
590                Resource::builder()
591                    .with_attributes(fastrace_resource_attrs)
592                    .build(),
593            ),
594            InstrumentationScope::builder("opentelemetry-instrumentation-foyer").build(),
595        );
596        fastrace::set_reporter(reporter, fastrace::collector::Config::default());
597        tracing::info!("opentelemetry exporter for fastrace is set at {endpoint}");
598    }
599
600    tracing_subscriber::registry().with(layers).init();
601    // TODO: add file-appender tracing subscriber in the future
602}
603
604#[cfg(test)]
605mod tests {
606    use super::*;
607
608    #[test]
609    fn test_parse_extra_tracing_attributes_normal() {
610        let attrs =
611            parse_extra_tracing_attributes("cluster=prod,region=us-east-1,namespace=terry-dev");
612        assert_eq!(attrs.len(), 3);
613        assert_eq!(attrs[0].key.as_str(), "cluster");
614        assert_eq!(attrs[0].value.as_str(), "prod");
615        assert_eq!(attrs[1].key.as_str(), "region");
616        assert_eq!(attrs[1].value.as_str(), "us-east-1");
617        assert_eq!(attrs[2].key.as_str(), "namespace");
618        assert_eq!(attrs[2].value.as_str(), "terry-dev");
619    }
620
621    #[test]
622    fn test_parse_extra_tracing_attributes_whitespace() {
623        let attrs = parse_extra_tracing_attributes("  key1 = val1 , key2=val2  ");
624        assert_eq!(attrs.len(), 2);
625        assert_eq!(attrs[0].key.as_str(), "key1");
626        assert_eq!(attrs[0].value.as_str(), "val1");
627        assert_eq!(attrs[1].key.as_str(), "key2");
628        assert_eq!(attrs[1].value.as_str(), "val2");
629    }
630
631    #[test]
632    fn test_parse_extra_tracing_attributes_empty() {
633        assert!(parse_extra_tracing_attributes("").is_empty());
634        assert!(parse_extra_tracing_attributes("  ").is_empty());
635    }
636
637    #[test]
638    fn test_parse_extra_tracing_attributes_invalid_pairs() {
639        // No '=' sign — should be skipped
640        let attrs = parse_extra_tracing_attributes("good=value,badpair,also_good=123");
641        assert_eq!(attrs.len(), 2);
642        assert_eq!(attrs[0].key.as_str(), "good");
643        assert_eq!(attrs[1].key.as_str(), "also_good");
644    }
645
646    #[test]
647    fn test_parse_extra_tracing_attributes_empty_key() {
648        // "=value" has empty key — should be skipped
649        let attrs = parse_extra_tracing_attributes("=value,key=val");
650        assert_eq!(attrs.len(), 1);
651        assert_eq!(attrs[0].key.as_str(), "key");
652    }
653
654    #[test]
655    fn test_parse_extra_tracing_attributes_value_with_equals() {
656        // value itself contains '=' — only first '=' is the delimiter
657        let attrs = parse_extra_tracing_attributes("expr=a=b");
658        assert_eq!(attrs.len(), 1);
659        assert_eq!(attrs[0].key.as_str(), "expr");
660        assert_eq!(attrs[0].value.as_str(), "a=b");
661    }
662
663    #[test]
664    fn test_parse_extra_tracing_attributes_trailing_comma() {
665        let attrs = parse_extra_tracing_attributes("k1=v1,k2=v2,");
666        assert_eq!(attrs.len(), 2);
667    }
668
669    #[test]
670    fn test_parse_extra_tracing_attributes_empty_value() {
671        let attrs = parse_extra_tracing_attributes("key=");
672        assert_eq!(attrs.len(), 1);
673        assert_eq!(attrs[0].key.as_str(), "key");
674        assert_eq!(attrs[0].value.as_str(), "");
675    }
676}
risingwave_rt/logger.rs

risingwave_rt/
logger.rs
