common_telemetry/
logging.rs

1// Copyright 2023 Greptime Team
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
15//! logging stuffs, inspired by databend
16use std::collections::HashMap;
17use std::env;
18use std::io::IsTerminal;
19use std::sync::{Arc, Mutex, Once, RwLock};
20use std::time::Duration;
21
22use common_base::serde::empty_string_as_default;
23use once_cell::sync::{Lazy, OnceCell};
24use opentelemetry::trace::TracerProvider;
25use opentelemetry::{KeyValue, global};
26use opentelemetry_otlp::{Protocol, SpanExporter, WithExportConfig, WithHttpConfig};
27use opentelemetry_sdk::propagation::TraceContextPropagator;
28use opentelemetry_sdk::trace::{Sampler, Tracer};
29use opentelemetry_semantic_conventions::resource;
30use serde::{Deserialize, Serialize};
31use tracing::callsite;
32use tracing::metadata::LevelFilter;
33use tracing_appender::non_blocking::WorkerGuard;
34use tracing_appender::rolling::{RollingFileAppender, Rotation};
35use tracing_log::LogTracer;
36use tracing_subscriber::filter::{FilterFn, Targets};
37use tracing_subscriber::fmt::Layer;
38use tracing_subscriber::layer::{Layered, SubscriberExt};
39use tracing_subscriber::prelude::*;
40use tracing_subscriber::{EnvFilter, Registry, filter};
41
42use crate::tracing_sampler::{TracingSampleOptions, create_sampler};
43
44/// The default endpoint when use gRPC exporter protocol.
45pub const DEFAULT_OTLP_GRPC_ENDPOINT: &str = "http://localhost:4317";
46
47/// The default endpoint when use HTTP exporter protocol.
48pub const DEFAULT_OTLP_HTTP_ENDPOINT: &str = "http://localhost:4318/v1/traces";
49
50/// The default logs directory.
51pub const DEFAULT_LOGGING_DIR: &str = "logs";
52
53/// Handle for reloading log level
54pub static LOG_RELOAD_HANDLE: OnceCell<tracing_subscriber::reload::Handle<Targets, Registry>> =
55    OnceCell::new();
56
57type DynSubscriber = Layered<tracing_subscriber::reload::Layer<Targets, Registry>, Registry>;
58type OtelTraceLayer = tracing_opentelemetry::OpenTelemetryLayer<DynSubscriber, Tracer>;
59
60#[derive(Clone)]
61pub struct TraceReloadHandle {
62    inner: Arc<RwLock<Option<OtelTraceLayer>>>,
63}
64
65impl TraceReloadHandle {
66    fn new(inner: Arc<RwLock<Option<OtelTraceLayer>>>) -> Self {
67        Self { inner }
68    }
69
70    pub fn reload(&self, new_layer: Option<OtelTraceLayer>) {
71        let mut guard = self.inner.write().unwrap();
72        *guard = new_layer;
73        drop(guard);
74
75        callsite::rebuild_interest_cache();
76    }
77}
78
79/// A tracing layer that can be dynamically reloaded.
80///
81/// Mostly copied from [`tracing_subscriber::reload::Layer`].
82struct TraceLayer {
83    inner: Arc<RwLock<Option<OtelTraceLayer>>>,
84}
85
86impl TraceLayer {
87    fn new(initial: Option<OtelTraceLayer>) -> (Self, TraceReloadHandle) {
88        let inner = Arc::new(RwLock::new(initial));
89        (
90            Self {
91                inner: inner.clone(),
92            },
93            TraceReloadHandle::new(inner),
94        )
95    }
96
97    fn with_layer<R>(&self, f: impl FnOnce(&OtelTraceLayer) -> R) -> Option<R> {
98        self.inner
99            .read()
100            .ok()
101            .and_then(|guard| guard.as_ref().map(f))
102    }
103
104    fn with_layer_mut<R>(&self, f: impl FnOnce(&mut OtelTraceLayer) -> R) -> Option<R> {
105        self.inner
106            .write()
107            .ok()
108            .and_then(|mut guard| guard.as_mut().map(f))
109    }
110}
111
112impl tracing_subscriber::Layer<DynSubscriber> for TraceLayer {
113    fn on_register_dispatch(&self, subscriber: &tracing::Dispatch) {
114        let _ = self.with_layer(|layer| layer.on_register_dispatch(subscriber));
115    }
116
117    fn on_layer(&mut self, subscriber: &mut DynSubscriber) {
118        let _ = self.with_layer_mut(|layer| layer.on_layer(subscriber));
119    }
120
121    fn register_callsite(
122        &self,
123        metadata: &'static tracing::Metadata<'static>,
124    ) -> tracing::subscriber::Interest {
125        self.with_layer(|layer| layer.register_callsite(metadata))
126            .unwrap_or_else(tracing::subscriber::Interest::always)
127    }
128
129    fn enabled(
130        &self,
131        metadata: &tracing::Metadata<'_>,
132        ctx: tracing_subscriber::layer::Context<'_, DynSubscriber>,
133    ) -> bool {
134        self.with_layer(|layer| layer.enabled(metadata, ctx))
135            .unwrap_or(true)
136    }
137
138    fn on_new_span(
139        &self,
140        attrs: &tracing::span::Attributes<'_>,
141        id: &tracing::span::Id,
142        ctx: tracing_subscriber::layer::Context<'_, DynSubscriber>,
143    ) {
144        let _ = self.with_layer(|layer| layer.on_new_span(attrs, id, ctx));
145    }
146
147    fn max_level_hint(&self) -> Option<LevelFilter> {
148        self.with_layer(|layer| layer.max_level_hint()).flatten()
149    }
150
151    fn on_record(
152        &self,
153        span: &tracing::span::Id,
154        values: &tracing::span::Record<'_>,
155        ctx: tracing_subscriber::layer::Context<'_, DynSubscriber>,
156    ) {
157        let _ = self.with_layer(|layer| layer.on_record(span, values, ctx));
158    }
159
160    fn on_follows_from(
161        &self,
162        span: &tracing::span::Id,
163        follows: &tracing::span::Id,
164        ctx: tracing_subscriber::layer::Context<'_, DynSubscriber>,
165    ) {
166        let _ = self.with_layer(|layer| layer.on_follows_from(span, follows, ctx));
167    }
168
169    fn event_enabled(
170        &self,
171        event: &tracing::Event<'_>,
172        ctx: tracing_subscriber::layer::Context<'_, DynSubscriber>,
173    ) -> bool {
174        self.with_layer(|layer| layer.event_enabled(event, ctx))
175            .unwrap_or(true)
176    }
177
178    fn on_event(
179        &self,
180        event: &tracing::Event<'_>,
181        ctx: tracing_subscriber::layer::Context<'_, DynSubscriber>,
182    ) {
183        let _ = self.with_layer(|layer| layer.on_event(event, ctx));
184    }
185
186    fn on_enter(
187        &self,
188        id: &tracing::span::Id,
189        ctx: tracing_subscriber::layer::Context<'_, DynSubscriber>,
190    ) {
191        let _ = self.with_layer(|layer| layer.on_enter(id, ctx));
192    }
193
194    fn on_exit(
195        &self,
196        id: &tracing::span::Id,
197        ctx: tracing_subscriber::layer::Context<'_, DynSubscriber>,
198    ) {
199        let _ = self.with_layer(|layer| layer.on_exit(id, ctx));
200    }
201
202    fn on_close(
203        &self,
204        id: tracing::span::Id,
205        ctx: tracing_subscriber::layer::Context<'_, DynSubscriber>,
206    ) {
207        let _ = self.with_layer(|layer| layer.on_close(id, ctx));
208    }
209
210    fn on_id_change(
211        &self,
212        old: &tracing::span::Id,
213        new: &tracing::span::Id,
214        ctx: tracing_subscriber::layer::Context<'_, DynSubscriber>,
215    ) {
216        let _ = self.with_layer(|layer| layer.on_id_change(old, new, ctx));
217    }
218
219    unsafe fn downcast_raw(&self, id: std::any::TypeId) -> Option<*const ()> {
220        self.inner.read().ok().and_then(|guard| {
221            guard
222                .as_ref()
223                .and_then(|layer| unsafe { layer.downcast_raw(id) })
224        })
225    }
226}
227
228/// Handle for reloading trace level
229pub static TRACE_RELOAD_HANDLE: OnceCell<TraceReloadHandle> = OnceCell::new();
230
231static TRACER: OnceCell<Mutex<TraceState>> = OnceCell::new();
232
233#[derive(Debug)]
234enum TraceState {
235    Ready(Tracer),
236    Deferred(TraceContext),
237}
238
239/// The logging options that used to initialize the logger.
240#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
241#[serde(default)]
242pub struct LoggingOptions {
243    /// The directory to store log files. If not set, logs will be written to stdout.
244    pub dir: String,
245
246    /// The log level that can be one of "trace", "debug", "info", "warn", "error". Default is "info".
247    pub level: Option<String>,
248
249    /// The log format that can be one of "json" or "text". Default is "text".
250    #[serde(default, deserialize_with = "empty_string_as_default")]
251    pub log_format: LogFormat,
252
253    /// The maximum number of log files set by default.
254    pub max_log_files: usize,
255
256    /// Whether to append logs to stdout. Default is true.
257    pub append_stdout: bool,
258
259    /// Whether to write logs to files in `dir`. Default is true.
260    pub enable_file_logging: bool,
261
262    /// Whether to enable tracing with OTLP. Default is false.
263    pub enable_otlp_tracing: bool,
264
265    /// The endpoint of OTLP.
266    pub otlp_endpoint: Option<String>,
267
268    /// The tracing sample ratio.
269    pub tracing_sample_ratio: Option<TracingSampleOptions>,
270
271    /// The protocol of OTLP export.
272    pub otlp_export_protocol: Option<OtlpExportProtocol>,
273
274    /// Additional HTTP headers for OTLP exporter.
275    #[serde(skip_serializing_if = "HashMap::is_empty")]
276    pub otlp_headers: HashMap<String, String>,
277
278    /// Whether to enable per-region metrics.
279    pub enable_per_region_metrics: bool,
280}
281
282/// The protocol of OTLP export.
283#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
284#[serde(rename_all = "snake_case")]
285pub enum OtlpExportProtocol {
286    /// GRPC protocol.
287    Grpc,
288
289    /// HTTP protocol with binary protobuf.
290    Http,
291}
292
293/// The options of slow query.
294#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
295#[serde(default)]
296pub struct SlowQueryOptions {
297    /// Whether to enable slow query log.
298    pub enable: bool,
299
300    /// The record type of slow queries.
301    #[serde(deserialize_with = "empty_string_as_default")]
302    pub record_type: SlowQueriesRecordType,
303
304    /// The threshold of slow queries.
305    #[serde(with = "humantime_serde")]
306    pub threshold: Duration,
307
308    /// The sample ratio of slow queries.
309    pub sample_ratio: f64,
310
311    /// The table TTL of `slow_queries` system table. Default is "90d".
312    /// It's used when `record_type` is `SystemTable`.
313    #[serde(with = "humantime_serde")]
314    pub ttl: Duration,
315}
316
317impl Default for SlowQueryOptions {
318    fn default() -> Self {
319        Self {
320            enable: true,
321            record_type: SlowQueriesRecordType::SystemTable,
322            threshold: Duration::from_secs(30),
323            sample_ratio: 1.0,
324            ttl: Duration::from_days(90),
325        }
326    }
327}
328
329#[derive(Clone, Debug, Serialize, Deserialize, Copy, PartialEq, Default)]
330#[serde(rename_all = "snake_case")]
331pub enum SlowQueriesRecordType {
332    /// Record the slow query in the system table.
333    #[default]
334    SystemTable,
335    /// Record the slow query in a specific logs file.
336    Log,
337}
338
339#[derive(Clone, Debug, Copy, PartialEq, Eq, Serialize, Deserialize, Default)]
340#[serde(rename_all = "snake_case")]
341pub enum LogFormat {
342    Json,
343    #[default]
344    Text,
345}
346
347#[derive(Clone, Debug)]
348struct TraceContext {
349    app_name: String,
350    node_id: String,
351    logging_opts: LoggingOptions,
352}
353
354impl Default for LoggingOptions {
355    fn default() -> Self {
356        Self {
357            // The directory path will be configured at application startup, typically using the data home directory as a base.
358            dir: "".to_string(),
359            level: None,
360            log_format: LogFormat::Text,
361            enable_otlp_tracing: false,
362            otlp_endpoint: None,
363            tracing_sample_ratio: None,
364            append_stdout: true,
365            enable_file_logging: true,
366            // Rotation hourly, 24 files per day, keeps info log files of 30 days
367            max_log_files: 720,
368            otlp_export_protocol: None,
369            otlp_headers: HashMap::new(),
370            enable_per_region_metrics: false,
371        }
372    }
373}
374
375#[derive(Default, Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
376pub struct TracingOptions {
377    #[cfg(feature = "tokio-console")]
378    pub tokio_console_addr: Option<String>,
379}
380
381/// Init tracing for unittest.
382/// Write logs to file `unittest`.
383pub fn init_default_ut_logging() {
384    static START: Once = Once::new();
385
386    START.call_once(|| {
387        let mut g = GLOBAL_UT_LOG_GUARD.as_ref().lock().unwrap();
388
389        // When running in Github's actions, env "UNITTEST_LOG_DIR" is set to a directory other
390        // than "/tmp".
391        // This is to fix the problem that the "/tmp" disk space of action runner's is small,
392        // if we write testing logs in it, actions would fail due to disk out of space error.
393        let dir =
394            env::var("UNITTEST_LOG_DIR").unwrap_or_else(|_| "/tmp/__unittest_logs".to_string());
395
396        let level = env::var("UNITTEST_LOG_LEVEL").unwrap_or_else(|_|
397            "debug,hyper=warn,tower=warn,datafusion=warn,reqwest=warn,sqlparser=warn,h2=info,opendal=info,rskafka=info".to_string()
398        );
399        let opts = LoggingOptions {
400            dir: dir.clone(),
401            level: Some(level),
402            ..Default::default()
403        };
404        *g = Some(init_global_logging(
405            "unittest",
406            &opts,
407            &TracingOptions::default(),
408            None,
409            None,
410        ));
411
412        crate::info!("logs dir = {}", dir);
413    });
414}
415
416static GLOBAL_UT_LOG_GUARD: Lazy<Arc<Mutex<Option<Vec<WorkerGuard>>>>> =
417    Lazy::new(|| Arc::new(Mutex::new(None)));
418
419const DEFAULT_LOG_TARGETS: &str = "info";
420
421#[allow(clippy::print_stdout)]
422pub fn init_global_logging(
423    app_name: &str,
424    opts: &LoggingOptions,
425    tracing_opts: &TracingOptions,
426    node_id: Option<String>,
427    slow_query_opts: Option<&SlowQueryOptions>,
428) -> Vec<WorkerGuard> {
429    static START: Once = Once::new();
430    let mut guards = vec![];
431    let node_id = node_id.unwrap_or_else(|| "none".to_string());
432
433    START.call_once(|| {
434        // Enable log compatible layer to convert log record to tracing span.
435        LogTracer::init().expect("log tracer must be valid");
436
437        // Configure the stdout logging layer.
438        let stdout_logging_layer = if opts.append_stdout {
439            let (writer, guard) = tracing_appender::non_blocking(std::io::stdout());
440            guards.push(guard);
441
442            if opts.log_format == LogFormat::Json {
443                Some(
444                    Layer::new()
445                        .json()
446                        .with_writer(writer)
447                        .with_ansi(std::io::stdout().is_terminal())
448                        .boxed(),
449                )
450            } else {
451                Some(
452                    Layer::new()
453                        .with_writer(writer)
454                        .with_ansi(std::io::stdout().is_terminal())
455                        .boxed(),
456                )
457            }
458        } else {
459            None
460        };
461
462        let file_logging_enabled = opts.enable_file_logging && !opts.dir.is_empty();
463
464        // Configure the file logging layer with rolling policy.
465        let file_logging_layer = if file_logging_enabled {
466            let rolling_appender = RollingFileAppender::builder()
467                .rotation(Rotation::HOURLY)
468                .filename_prefix("greptimedb")
469                .max_log_files(opts.max_log_files)
470                .build(&opts.dir)
471                .unwrap_or_else(|e| {
472                    panic!(
473                        "initializing rolling file appender at {} failed: {}",
474                        &opts.dir, e
475                    )
476                });
477            let (writer, guard) = tracing_appender::non_blocking(rolling_appender);
478            guards.push(guard);
479
480            if opts.log_format == LogFormat::Json {
481                Some(
482                    Layer::new()
483                        .json()
484                        .with_writer(writer)
485                        .with_ansi(false)
486                        .boxed(),
487                )
488            } else {
489                Some(Layer::new().with_writer(writer).with_ansi(false).boxed())
490            }
491        } else {
492            None
493        };
494
495        // Configure the error file logging layer with rolling policy.
496        let err_file_logging_layer = if file_logging_enabled {
497            let rolling_appender = RollingFileAppender::builder()
498                .rotation(Rotation::HOURLY)
499                .filename_prefix("greptimedb-err")
500                .max_log_files(opts.max_log_files)
501                .build(&opts.dir)
502                .unwrap_or_else(|e| {
503                    panic!(
504                        "initializing rolling file appender at {} failed: {}",
505                        &opts.dir, e
506                    )
507                });
508            let (writer, guard) = tracing_appender::non_blocking(rolling_appender);
509            guards.push(guard);
510
511            if opts.log_format == LogFormat::Json {
512                Some(
513                    Layer::new()
514                        .json()
515                        .with_writer(writer)
516                        .with_ansi(false)
517                        .with_filter(filter::LevelFilter::ERROR)
518                        .boxed(),
519                )
520            } else {
521                Some(
522                    Layer::new()
523                        .with_writer(writer)
524                        .with_ansi(false)
525                        .with_filter(filter::LevelFilter::ERROR)
526                        .boxed(),
527                )
528            }
529        } else {
530            None
531        };
532
533        let slow_query_logging_layer = build_slow_query_logger(opts, slow_query_opts, &mut guards);
534
535        // resolve log level settings from:
536        // - options from command line or config files
537        // - environment variable: RUST_LOG
538        // - default settings
539        let filter = opts
540            .level
541            .as_deref()
542            .or(env::var(EnvFilter::DEFAULT_ENV).ok().as_deref())
543            .unwrap_or(DEFAULT_LOG_TARGETS)
544            .parse::<filter::Targets>()
545            .expect("error parsing log level string");
546
547        let (dyn_filter, reload_handle) = tracing_subscriber::reload::Layer::new(filter.clone());
548
549        LOG_RELOAD_HANDLE
550            .set(reload_handle)
551            .expect("reload handle already set, maybe init_global_logging get called twice?");
552
553        let mut initial_tracer = None;
554        let trace_state = if opts.enable_otlp_tracing {
555            let tracer = create_tracer(app_name, &node_id, opts);
556            initial_tracer = Some(tracer.clone());
557            TraceState::Ready(tracer)
558        } else {
559            TraceState::Deferred(TraceContext {
560                app_name: app_name.to_string(),
561                node_id: node_id.clone(),
562                logging_opts: opts.clone(),
563            })
564        };
565
566        TRACER
567            .set(Mutex::new(trace_state))
568            .expect("trace state already initialized");
569
570        let initial_trace_layer = initial_tracer
571            .as_ref()
572            .map(|tracer| tracing_opentelemetry::layer().with_tracer(tracer.clone()));
573
574        let (dyn_trace_layer, trace_reload_handle) = TraceLayer::new(initial_trace_layer);
575
576        TRACE_RELOAD_HANDLE
577            .set(trace_reload_handle)
578            .unwrap_or_else(|_| panic!("failed to set trace reload handle"));
579
580        // Must enable 'tokio_unstable' cfg to use this feature.
581        // For example: `RUSTFLAGS="--cfg tokio_unstable" cargo run -F common-telemetry/console -- standalone start`
582        #[cfg(feature = "tokio-console")]
583        let subscriber = {
584            let tokio_console_layer =
585                if let Some(tokio_console_addr) = &tracing_opts.tokio_console_addr {
586                    let addr: std::net::SocketAddr = tokio_console_addr.parse().unwrap_or_else(|e| {
587                    panic!("Invalid binding address '{tokio_console_addr}' for tokio-console: {e}");
588                });
589                    println!("tokio-console listening on {addr}");
590
591                    Some(
592                        console_subscriber::ConsoleLayer::builder()
593                            .server_addr(addr)
594                            .spawn(),
595                    )
596                } else {
597                    None
598                };
599
600            Registry::default()
601                .with(dyn_filter)
602                .with(dyn_trace_layer)
603                .with(tokio_console_layer)
604                .with(stdout_logging_layer)
605                .with(file_logging_layer)
606                .with(err_file_logging_layer)
607                .with(slow_query_logging_layer)
608        };
609
610        // consume the `tracing_opts` to avoid "unused" warnings.
611        let _ = tracing_opts;
612
613        #[cfg(not(feature = "tokio-console"))]
614        let subscriber = Registry::default()
615            .with(dyn_filter)
616            .with(dyn_trace_layer)
617            .with(stdout_logging_layer)
618            .with(file_logging_layer)
619            .with(err_file_logging_layer)
620            .with(slow_query_logging_layer);
621
622        global::set_text_map_propagator(TraceContextPropagator::new());
623
624        tracing::subscriber::set_global_default(subscriber)
625            .expect("error setting global tracing subscriber");
626    });
627
628    guards
629}
630
631fn create_tracer(app_name: &str, node_id: &str, opts: &LoggingOptions) -> Tracer {
632    let sampler = opts
633        .tracing_sample_ratio
634        .as_ref()
635        .map(create_sampler)
636        .map(Sampler::ParentBased)
637        .unwrap_or(Sampler::ParentBased(Box::new(Sampler::AlwaysOn)));
638
639    let resource = opentelemetry_sdk::Resource::builder_empty()
640        .with_attributes([
641            KeyValue::new(resource::SERVICE_NAME, app_name.to_string()),
642            KeyValue::new(resource::SERVICE_INSTANCE_ID, node_id.to_string()),
643            KeyValue::new(resource::SERVICE_VERSION, common_version::version()),
644            KeyValue::new(resource::PROCESS_PID, std::process::id().to_string()),
645        ])
646        .build();
647
648    opentelemetry_sdk::trace::SdkTracerProvider::builder()
649        .with_batch_exporter(build_otlp_exporter(opts))
650        .with_sampler(sampler)
651        .with_resource(resource)
652        .build()
653        .tracer("greptimedb")
654}
655
656/// Ensure that the OTLP tracer has been constructed, building it lazily if needed.
657pub fn get_or_init_tracer() -> Result<Tracer, &'static str> {
658    let state = TRACER.get().ok_or("trace state is not initialized")?;
659    let mut guard = state.lock().expect("trace state lock poisoned");
660
661    match &mut *guard {
662        TraceState::Ready(tracer) => Ok(tracer.clone()),
663        TraceState::Deferred(context) => {
664            let tracer = create_tracer(&context.app_name, &context.node_id, &context.logging_opts);
665            *guard = TraceState::Ready(tracer.clone());
666            Ok(tracer)
667        }
668    }
669}
670
671fn build_otlp_exporter(opts: &LoggingOptions) -> SpanExporter {
672    let protocol = opts
673        .otlp_export_protocol
674        .clone()
675        .unwrap_or(OtlpExportProtocol::Http);
676
677    let endpoint = opts
678        .otlp_endpoint
679        .as_ref()
680        .map(|e| {
681            if e.starts_with("http") {
682                e.clone()
683            } else {
684                format!("http://{}", e)
685            }
686        })
687        .unwrap_or_else(|| match protocol {
688            OtlpExportProtocol::Grpc => DEFAULT_OTLP_GRPC_ENDPOINT.to_string(),
689            OtlpExportProtocol::Http => DEFAULT_OTLP_HTTP_ENDPOINT.to_string(),
690        });
691
692    match protocol {
693        OtlpExportProtocol::Grpc => SpanExporter::builder()
694            .with_tonic()
695            .with_endpoint(endpoint)
696            .build()
697            .expect("Failed to create OTLP gRPC exporter "),
698
699        OtlpExportProtocol::Http => SpanExporter::builder()
700            .with_http()
701            .with_endpoint(endpoint)
702            .with_protocol(Protocol::HttpBinary)
703            .with_headers(opts.otlp_headers.clone())
704            .build()
705            .expect("Failed to create OTLP HTTP exporter "),
706    }
707}
708
709fn build_slow_query_logger<S>(
710    opts: &LoggingOptions,
711    slow_query_opts: Option<&SlowQueryOptions>,
712    guards: &mut Vec<WorkerGuard>,
713) -> Option<Box<dyn tracing_subscriber::Layer<S> + Send + Sync + 'static>>
714where
715    S: tracing::Subscriber
716        + Send
717        + 'static
718        + for<'span> tracing_subscriber::registry::LookupSpan<'span>,
719{
720    if let Some(slow_query_opts) = slow_query_opts {
721        if opts.enable_file_logging
722            && !opts.dir.is_empty()
723            && slow_query_opts.enable
724            && slow_query_opts.record_type == SlowQueriesRecordType::Log
725        {
726            let rolling_appender = RollingFileAppender::builder()
727                .rotation(Rotation::HOURLY)
728                .filename_prefix("greptimedb-slow-queries")
729                .max_log_files(opts.max_log_files)
730                .build(&opts.dir)
731                .unwrap_or_else(|e| {
732                    panic!(
733                        "initializing rolling file appender at {} failed: {}",
734                        &opts.dir, e
735                    )
736                });
737            let (writer, guard) = tracing_appender::non_blocking(rolling_appender);
738            guards.push(guard);
739
740            // Only logs if the field contains "slow".
741            let slow_query_filter = FilterFn::new(|metadata| {
742                metadata
743                    .fields()
744                    .iter()
745                    .any(|field| field.name().contains("slow"))
746            });
747
748            if opts.log_format == LogFormat::Json {
749                Some(
750                    Layer::new()
751                        .json()
752                        .with_writer(writer)
753                        .with_ansi(false)
754                        .with_filter(slow_query_filter)
755                        .boxed(),
756                )
757            } else {
758                Some(
759                    Layer::new()
760                        .with_writer(writer)
761                        .with_ansi(false)
762                        .with_filter(slow_query_filter)
763                        .boxed(),
764                )
765            }
766        } else {
767            None
768        }
769    } else {
770        None
771    }
772}
773
774#[cfg(test)]
775mod tests {
776    use super::*;
777
778    #[test]
779    fn test_logging_options_deserialization_default() {
780        let json = r#"{}"#;
781        let opts: LoggingOptions = serde_json::from_str(json).unwrap();
782
783        assert_eq!(opts.log_format, LogFormat::Text);
784        assert_eq!(opts.dir, "");
785        assert_eq!(opts.level, None);
786        assert!(opts.append_stdout);
787        assert!(opts.enable_file_logging);
788    }
789
790    #[test]
791    fn test_logging_options_deserialization_enable_file_logging() {
792        let json = r#"{"enable_file_logging": false}"#;
793        let opts: LoggingOptions = serde_json::from_str(json).unwrap();
794
795        assert!(!opts.enable_file_logging);
796    }
797
798    #[test]
799    fn test_logging_options_deserialization_empty_log_format() {
800        let json = r#"{"log_format": ""}"#;
801        let opts: LoggingOptions = serde_json::from_str(json).unwrap();
802
803        // Empty string should use default (Text)
804        assert_eq!(opts.log_format, LogFormat::Text);
805    }
806
807    #[test]
808    fn test_logging_options_deserialization_valid_log_format() {
809        let json_format = r#"{"log_format": "json"}"#;
810        let opts: LoggingOptions = serde_json::from_str(json_format).unwrap();
811        assert_eq!(opts.log_format, LogFormat::Json);
812
813        let text_format = r#"{"log_format": "text"}"#;
814        let opts: LoggingOptions = serde_json::from_str(text_format).unwrap();
815        assert_eq!(opts.log_format, LogFormat::Text);
816    }
817
818    #[test]
819    fn test_logging_options_deserialization_missing_log_format() {
820        let json = r#"{"dir": "/tmp/logs"}"#;
821        let opts: LoggingOptions = serde_json::from_str(json).unwrap();
822
823        // Missing log_format should use default (Text)
824        assert_eq!(opts.log_format, LogFormat::Text);
825        assert_eq!(opts.dir, "/tmp/logs");
826    }
827
828    #[test]
829    fn test_slow_query_options_deserialization_default() {
830        let json = r#"{"enable": true, "threshold": "30s"}"#;
831        let opts: SlowQueryOptions = serde_json::from_str(json).unwrap();
832
833        assert_eq!(opts.record_type, SlowQueriesRecordType::SystemTable);
834        assert!(opts.enable);
835    }
836
837    #[test]
838    fn test_slow_query_options_deserialization_empty_record_type() {
839        let json = r#"{"enable": true, "record_type": "", "threshold": "30s"}"#;
840        let opts: SlowQueryOptions = serde_json::from_str(json).unwrap();
841
842        // Empty string should use default (SystemTable)
843        assert_eq!(opts.record_type, SlowQueriesRecordType::SystemTable);
844        assert!(opts.enable);
845    }
846
847    #[test]
848    fn test_slow_query_options_deserialization_valid_record_type() {
849        let system_table_json =
850            r#"{"enable": true, "record_type": "system_table", "threshold": "30s"}"#;
851        let opts: SlowQueryOptions = serde_json::from_str(system_table_json).unwrap();
852        assert_eq!(opts.record_type, SlowQueriesRecordType::SystemTable);
853
854        let log_json = r#"{"enable": true, "record_type": "log", "threshold": "30s"}"#;
855        let opts: SlowQueryOptions = serde_json::from_str(log_json).unwrap();
856        assert_eq!(opts.record_type, SlowQueriesRecordType::Log);
857    }
858
859    #[test]
860    fn test_slow_query_options_deserialization_missing_record_type() {
861        let json = r#"{"enable": false, "threshold": "30s"}"#;
862        let opts: SlowQueryOptions = serde_json::from_str(json).unwrap();
863
864        // Missing record_type should use default (SystemTable)
865        assert_eq!(opts.record_type, SlowQueriesRecordType::SystemTable);
866        assert!(!opts.enable);
867    }
868
869    #[test]
870    fn test_otlp_export_protocol_deserialization_valid_values() {
871        let grpc_json = r#""grpc""#;
872        let protocol: OtlpExportProtocol = serde_json::from_str(grpc_json).unwrap();
873        assert_eq!(protocol, OtlpExportProtocol::Grpc);
874
875        let http_json = r#""http""#;
876        let protocol: OtlpExportProtocol = serde_json::from_str(http_json).unwrap();
877        assert_eq!(protocol, OtlpExportProtocol::Http);
878    }
879
880    #[test]
881    fn test_logging_options_partial_eq_all_fields() {
882        let base = LoggingOptions::default();
883
884        let mut log_format = base.clone();
885        log_format.log_format = LogFormat::Json;
886        assert_ne!(base, log_format);
887
888        let mut max_log_files = base.clone();
889        max_log_files.max_log_files += 1;
890        assert_ne!(base, max_log_files);
891
892        let mut otlp_export_protocol = base.clone();
893        otlp_export_protocol.otlp_export_protocol = Some(OtlpExportProtocol::Http);
894        assert_ne!(base, otlp_export_protocol);
895
896        let mut otlp_headers = base.clone();
897        otlp_headers
898            .otlp_headers
899            .insert("key".to_string(), "value".to_string());
900        assert_ne!(base, otlp_headers);
901    }
902}
common_telemetry/logging.rs

common_telemetry/
logging.rs
