rocketmq_observability/
log_filter.rs

1// Copyright 2023 The RocketMQ Rust Authors
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::ffi::OsString;
16use std::fmt;
17use std::sync::Arc;
18use std::sync::Mutex;
19use std::sync::RwLock;
20
21use serde::Deserialize;
22use serde::Serialize;
23use tracing_subscriber::EnvFilter;
24
25use crate::ObservabilityError;
26use crate::ReloadConfig;
27
28pub const DEFAULT_LOG_FILTER: &str = "info";
29
30/// Reads the process-level `RUST_LOG` override without silently discarding a
31/// non-Unicode value.
32pub fn read_rust_log() -> Result<Option<String>, ObservabilityError> {
33    std::env::var_os("RUST_LOG").map(decode_environment_filter).transpose()
34}
35
36fn decode_environment_filter(value: OsString) -> Result<String, ObservabilityError> {
37    value
38        .into_string()
39        .map_err(|_| ObservabilityError::invalid_config("RUST_LOG contains non-Unicode data"))
40}
41
42#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
43#[serde(rename_all = "snake_case")]
44pub enum LogFilterSource {
45    Runtime,
46    Cli,
47    Env,
48    Config,
49    Default,
50}
51
52impl fmt::Display for LogFilterSource {
53    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
54        formatter.write_str(match self {
55            Self::Runtime => "runtime",
56            Self::Cli => "cli",
57            Self::Env => "env",
58            Self::Config => "config",
59            Self::Default => "default",
60        })
61    }
62}
63
64impl LogFilterSource {
65    pub const fn as_str(self) -> &'static str {
66        match self {
67            Self::Runtime => "runtime",
68            Self::Cli => "cli",
69            Self::Env => "env",
70            Self::Config => "config",
71            Self::Default => "default",
72        }
73    }
74}
75
76#[derive(Debug, Clone, Copy, Default)]
77pub struct LogFilterInputs<'a> {
78    pub runtime: Option<&'a str>,
79    pub cli: Option<&'a str>,
80    pub environment: Option<&'a str>,
81    pub config: Option<&'a str>,
82    pub legacy_config: Option<&'a str>,
83}
84
85#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
86pub struct ResolvedLogFilter {
87    filter: String,
88    source: LogFilterSource,
89}
90
91impl ResolvedLogFilter {
92    pub fn filter(&self) -> &str {
93        &self.filter
94    }
95
96    pub const fn source(&self) -> LogFilterSource {
97        self.source
98    }
99}
100
101#[derive(Debug, Clone, PartialEq, Eq)]
102pub struct LogFilterReloadRequest {
103    filter: String,
104}
105
106impl LogFilterReloadRequest {
107    pub fn new(filter: impl Into<String>) -> Self {
108        Self { filter: filter.into() }
109    }
110
111    pub fn filter(&self) -> &str {
112        &self.filter
113    }
114}
115
116type ReloadOperation = dyn Fn(EnvFilter) -> Result<(), ObservabilityError> + Send + Sync + 'static;
117
118struct LogFilterHandleInner {
119    current: RwLock<ResolvedLogFilter>,
120    reload: Box<ReloadOperation>,
121    operation_lock: Mutex<()>,
122    metrics: crate::metrics::log_filter::LogFilterMetrics,
123}
124
125#[derive(Clone)]
126pub struct LogFilterHandle {
127    inner: Arc<LogFilterHandleInner>,
128}
129
130impl LogFilterHandle {
131    pub(crate) fn new<F>(
132        initial: ResolvedLogFilter,
133        metrics: crate::metrics::log_filter::LogFilterMetrics,
134        reload: F,
135    ) -> Self
136    where
137        F: Fn(EnvFilter) -> Result<(), ObservabilityError> + Send + Sync + 'static,
138    {
139        Self {
140            inner: Arc::new(LogFilterHandleInner {
141                metrics,
142                current: RwLock::new(initial),
143                reload: Box::new(reload),
144                operation_lock: Mutex::new(()),
145            }),
146        }
147    }
148
149    pub fn current(&self) -> ResolvedLogFilter {
150        self.inner
151            .current
152            .read()
153            .unwrap_or_else(|poisoned| poisoned.into_inner())
154            .clone()
155    }
156
157    pub fn reload(&self, request: LogFilterReloadRequest) -> Result<ResolvedLogFilter, ObservabilityError> {
158        let resolved = LogFilterResolver::resolve(LogFilterInputs {
159            runtime: Some(request.filter()),
160            ..LogFilterInputs::default()
161        });
162        let resolved = match resolved {
163            Ok(resolved) => resolved,
164            Err(error) => {
165                self.inner.metrics.record_reload(false, LogFilterSource::Runtime);
166                return Err(error);
167            }
168        };
169        self.apply_resolved(resolved)
170    }
171
172    /// Restores a previously resolved startup baseline, including its original
173    /// source metadata.
174    pub fn restore(&self, baseline: &ResolvedLogFilter) -> Result<ResolvedLogFilter, ObservabilityError> {
175        self.apply_resolved(baseline.clone())
176    }
177
178    fn apply_resolved(&self, resolved: ResolvedLogFilter) -> Result<ResolvedLogFilter, ObservabilityError> {
179        let filter = EnvFilter::try_new(resolved.filter())
180            .map_err(|error| ObservabilityError::invalid_log_filter(resolved.filter(), error))?;
181        let _operation = self
182            .inner
183            .operation_lock
184            .lock()
185            .unwrap_or_else(|poisoned| poisoned.into_inner());
186        let previous = self.current();
187        if let Err(error) = (self.inner.reload)(filter) {
188            self.inner.metrics.record_reload(false, resolved.source());
189            return Err(error);
190        }
191        *self
192            .inner
193            .current
194            .write()
195            .unwrap_or_else(|poisoned| poisoned.into_inner()) = resolved.clone();
196        self.inner.metrics.record_reload(true, resolved.source());
197        self.inner
198            .metrics
199            .set_active(Some(previous.source()), resolved.source());
200        Ok(resolved)
201    }
202
203    /// Records the absolute expiry timestamp of a temporary filter override.
204    pub fn set_expiry_timestamp(&self, timestamp_seconds: u64) {
205        self.inner.metrics.set_expiry_timestamp(timestamp_seconds);
206    }
207
208    /// Records a failed durable audit write.
209    pub fn record_audit_failure(&self) {
210        self.inner.metrics.record_audit_failure();
211    }
212
213    /// Records a failed automatic restore after an override expires.
214    pub fn record_auto_restore_failure(&self) {
215        self.inner.metrics.record_auto_restore_failure();
216    }
217
218    /// Records a failed rollback to the startup filter.
219    pub fn record_rollback_failure(&self) {
220        self.inner.metrics.record_rollback_failure();
221    }
222}
223
224#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
225#[serde(default)]
226pub struct LoggingOverrides {
227    pub logging: LoggingOverrideConfig,
228    #[serde(rename = "logFilter", alias = "log_filter")]
229    pub log_filter: Option<String>,
230}
231
232#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
233#[serde(default)]
234pub struct LoggingOverrideConfig {
235    pub filter: Option<String>,
236    pub reload: ReloadConfig,
237}
238
239pub struct LogFilterResolver;
240
241impl LogFilterResolver {
242    pub fn resolve(inputs: LogFilterInputs<'_>) -> Result<ResolvedLogFilter, ObservabilityError> {
243        let config = normalized_config_filter(inputs.config, inputs.legacy_config)?;
244        let (candidate, source) = inputs
245            .runtime
246            .map(|value| (value, LogFilterSource::Runtime))
247            .or_else(|| inputs.cli.map(|value| (value, LogFilterSource::Cli)))
248            .or_else(|| inputs.environment.map(|value| (value, LogFilterSource::Env)))
249            .or_else(|| config.map(|value| (value, LogFilterSource::Config)))
250            .unwrap_or((DEFAULT_LOG_FILTER, LogFilterSource::Default));
251        let candidate = candidate.trim();
252        if candidate.is_empty() {
253            return Err(ObservabilityError::invalid_log_filter_detail(
254                candidate,
255                "log filter must not be blank",
256            ));
257        }
258
259        EnvFilter::try_new(candidate).map_err(|error| ObservabilityError::invalid_log_filter(candidate, error))?;
260        Ok(ResolvedLogFilter {
261            filter: candidate.to_string(),
262            source,
263        })
264    }
265}
266
267fn normalized_config_filter<'a>(
268    config: Option<&'a str>,
269    legacy_config: Option<&'a str>,
270) -> Result<Option<&'a str>, ObservabilityError> {
271    match (config, legacy_config) {
272        (Some(config), Some(legacy)) if config.trim() != legacy.trim() => Err(ObservabilityError::invalid_config(
273            "logging.filter conflicts with the Java-compatible logFilter value",
274        )),
275        (Some(config), _) => Ok(Some(config)),
276        (None, Some(legacy)) => Ok(Some(legacy)),
277        (None, None) => Ok(None),
278    }
279}
280
281#[cfg(all(test, unix))]
282mod tests {
283    use super::*;
284
285    #[test]
286    fn non_unicode_environment_filter_is_rejected() {
287        use std::os::unix::ffi::OsStringExt;
288
289        let error = decode_environment_filter(OsString::from_vec(vec![0xff]))
290            .expect_err("non-Unicode environment values must fail closed");
291
292        assert_eq!(error.code(), rocketmq_error::OBSERVABILITY_CONFIGURATION_INVALID.code());
293    }
294}
rocketmq_observability/log_filter.rs

rocketmq_observability/
log_filter.rs
