filip/Stalwart
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

Improved tracing (closes #180 closes #417 closes #376 closes #418 closes #517)

Browse Source
This commit is contained in:
mdecimus
2024-07-30 16:12:34 +02:00
parent a45eb50231
commit d29d21692e
65 changed files with 2005 additions and 1112 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
crates/common/Cargo.toml
View File

@@ -38,13 +38,10 @@ x509-parser = "0.16.0"
pem = "3.0"
chrono = { version = "0.4", features = ["serde"] }
hyper = { version = "1.0.1", features = ["server", "http1", "http2"] }
tracing-subscriber = { version = "0.3", features = ["env-filter"] }
tracing-appender = "0.2"
tracing-opentelemetry = "0.23.0"
opentelemetry = { version = "0.22.0" }
opentelemetry_sdk = { version = "0.22.1", features = ["rt-tokio"] }
opentelemetry-otlp = { version = "0.15.0", features = ["http-proto", "reqwest-client"] }
opentelemetry-semantic-conventions = { version = "0.14.0" }
#opentelemetry = { version = "0.22.0" }
#opentelemetry_sdk = { version = "0.22.1", features = ["rt-tokio"] }
#opentelemetry-otlp = { version = "0.15.0", features = ["http-proto", "reqwest-client"] }
#opentelemetry-semantic-conventions = { version = "0.14.0" }
imagesize = "0.13"
sha1 = "0.10"
sha2 = "0.10.6"

131
crates/common/src/config/network.rs
View File

@@ -45,134 +45,3 @@ impl Network {
network
}
}
/*
impl Webhooks {
pub fn parse(config: &mut Config) -> Self {
let mut hooks = Webhooks {
events: Default::default(),
hooks: Default::default(),
};
for id in config
.sub_keys("webhook", ".url")
.map(|s| s.to_string())
.collect::<Vec<_>>()
{
if let Some(webhook) = parse_webhook(config, &id) {
hooks.events.extend(&webhook.events);
hooks.hooks.insert(webhook.id, webhook.into());
}
}
hooks
}
}
fn parse_webhook(config: &mut Config, id: &str) -> Option<Webhook> {
let mut headers = HeaderMap::new();
for (header, value) in config
.values(("webhook", id, "headers"))
.map(|(_, v)| {
if let Some((k, v)) = v.split_once(':') {
Ok((
HeaderName::from_str(k.trim()).map_err(|err| {
format!("Invalid header found in property \"webhook.{id}.headers\": {err}",)
})?,
HeaderValue::from_str(v.trim()).map_err(|err| {
format!("Invalid header found in property \"webhook.{id}.headers\": {err}",)
})?,
))
} else {
Err(format!(
"Invalid header found in property \"webhook.{id}.headers\": {v}",
))
}
})
.collect::<Result<Vec<(HeaderName, HeaderValue)>, String>>()
.map_err(|e| config.new_parse_error(("webhook", id, "headers"), e))
.unwrap_or_default()
{
headers.insert(header, value);
}
headers.insert(CONTENT_TYPE, "application/json".parse().unwrap());
if let (Some(name), Some(secret)) = (
config.value(("webhook", id, "auth.username")),
config.value(("webhook", id, "auth.secret")),
) {
headers.insert(
AUTHORIZATION,
format!("Basic {}", STANDARD.encode(format!("{}:{}", name, secret)))
.parse()
.unwrap(),
);
}
// Parse webhook events
let mut events = AHashSet::new();
let mut parse_errors = Vec::new();
for (_, value) in config.values(("webhook", id, "events")) {
match WebhookType::from_str(value) {
Ok(event) => {
events.insert(event);
}
Err(err) => {
parse_errors.push(err);
}
}
}
if !parse_errors.is_empty() {
config.new_parse_error(
("webhook", id, "events"),
format!("Invalid webhook events: {}", parse_errors.join(", ")),
);
}
let url = config.value_require(("webhook", id, "url"))?.to_string();
Some(Webhook {
id: xxhash_rust::xxh3::xxh3_64(url.as_bytes()),
url,
timeout: config
.property_or_default(("webhook", id, "timeout"), "30s")
.unwrap_or_else(|| Duration::from_secs(30)),
tls_allow_invalid_certs: config
.property_or_default(("webhook", id, "allow-invalid-certs"), "false")
.unwrap_or_default(),
headers,
key: config
.value(("webhook", id, "signature-key"))
.unwrap_or_default()
.to_string(),
throttle: config
.property_or_default(("webhook", id, "throttle"), "1s")
.unwrap_or_else(|| Duration::from_secs(1)),
events,
})
}
impl FromStr for WebhookType {
type Err = String;
fn from_str(s: &str) -> Result<Self, Self::Err> {
match s {
"auth.success" => Ok(Self::AuthSuccess),
"auth.failure" => Ok(Self::AuthFailure),
"auth.banned" => Ok(Self::AuthBanned),
"auth.error" => Ok(Self::AuthError),
"message.accepted" => Ok(Self::MessageAccepted),
"message.rejected" => Ok(Self::MessageRejected),
"message.appended" => Ok(Self::MessageAppended),
"account.over-quota" => Ok(Self::AccountOverQuota),
"dsn" => Ok(Self::DSN),
"double-bounce" => Ok(Self::DoubleBounce),
"report.incoming.dmarc" => Ok(Self::IncomingDmarcReport),
"report.incoming.tls" => Ok(Self::IncomingTlsReport),
"report.incoming.arf" => Ok(Self::IncomingArfReport),
"report.outgoing" => Ok(Self::OutgoingReport),
_ => Err(s.to_string()),
}
}
}
*/

30
crates/common/src/config/server/listener.rs
View File

@@ -34,15 +34,15 @@ use super::{
impl Servers {
pub fn parse(config: &mut Config) -> Self {
// Parse ACME managers
let mut servers = Servers::default();
// Create sessionId generator
let id_generator = Arc::new(
config
.property::<u64>("cluster.node-id")
.map(SnowflakeIdGenerator::with_node_id)
.unwrap_or_default(),
);
let mut servers = Servers {
span_id_gen: Arc::new(
config
.property::<u64>("cluster.node-id")
.map(SnowflakeIdGenerator::with_node_id)
.unwrap_or_default(),
),
..Default::default()
};
// Parse servers
for id in config
@@ -50,17 +50,12 @@ impl Servers {
.map(|s| s.to_string())
.collect::<Vec<_>>()
{
servers.parse_server(config, id, id_generator.clone());
servers.parse_server(config, id);
}
servers
}
fn parse_server(
&mut self,
config: &mut Config,
id_: String,
id_generator: Arc<SnowflakeIdGenerator>,
) {
fn parse_server(&mut self, config: &mut Config, id_: String) {
// Parse protocol
let id = id_.as_str();
let protocol =
@@ -197,6 +192,7 @@ impl Servers {
proxy_networks.push(network);
}
let span_id_gen = self.span_id_gen.clone();
self.servers.push(Server {
max_connections: config
.property_or_else(
@@ -209,7 +205,7 @@ impl Servers {
protocol,
listeners,
proxy_networks,
id_generator,
span_id_gen,
});
}

3
crates/common/src/config/server/mod.rs
View File

@@ -20,6 +20,7 @@ pub mod tls;
pub struct Servers {
pub servers: Vec<Server>,
pub tcp_acceptors: AHashMap<String, TcpAcceptor>,
pub span_id_gen: Arc<SnowflakeIdGenerator>,
}
#[derive(Debug, Default)]
@@ -29,7 +30,7 @@ pub struct Server {
pub listeners: Vec<Listener>,
pub proxy_networks: Vec<IpAddrMask>,
pub max_connections: u64,
pub id_generator: Arc<SnowflakeIdGenerator>,
pub span_id_gen: Arc<SnowflakeIdGenerator>,
}
#[derive(Debug)]

403
crates/common/src/config/tracers.rs
View File

@@ -4,52 +4,105 @@
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use std::{collections::HashMap, str::FromStr};
use std::{str::FromStr, time::Duration};
use opentelemetry_otlp::{HttpExporterBuilder, TonicExporterBuilder, WithExportConfig};
use tracing_appender::rolling::RollingFileAppender;
use trc::Level;
use ahash::{AHashMap, AHashSet};
use base64::{engine::general_purpose::STANDARD, Engine};
use hyper::{
header::{HeaderName, HeaderValue, AUTHORIZATION, CONTENT_TYPE},
HeaderMap,
};
use trc::{subscriber::Interests, EventType, Level};
use utils::config::Config;
#[derive(Debug)]
pub enum Tracer {
Stdout {
id: String,
level: Level,
ansi: bool,
},
Log {
id: String,
level: Level,
appender: RollingFileAppender,
ansi: bool,
},
Journal {
id: String,
level: Level,
},
Otel {
id: String,
level: Level,
tracer: OtelTracer,
},
pub struct Tracer {
pub id: String,
pub interests: Interests,
pub typ: TracerType,
pub lossy: bool,
}
#[derive(Debug)]
pub enum OtelTracer {
Gprc(TonicExporterBuilder),
Http(HttpExporterBuilder),
pub enum TracerType {
Console(ConsoleTracer),
Log(LogTracer),
Otel(OtelTracer),
Webhook(WebhookTracer),
Journal,
}
#[derive(Debug)]
pub struct ConsoleTracer {
pub ansi: bool,
pub multiline: bool,
pub buffered: bool,
}
#[derive(Debug)]
pub struct LogTracer {
pub path: String,
pub prefix: String,
pub rotate: RotationStrategy,
pub ansi: bool,
pub multiline: bool,
}
#[derive(Debug)]
pub struct OtelTracer {
pub endpoint: String,
pub headers: AHashMap<String, String>,
pub is_http: bool,
}
#[derive(Debug)]
pub struct WebhookTracer {
pub url: String,
pub key: String,
pub timeout: Duration,
pub throttle: Duration,
pub tls_allow_invalid_certs: bool,
pub headers: HeaderMap,
}
#[derive(Debug)]
pub enum RotationStrategy {
Daily,
Hourly,
Minutely,
Never,
}
#[derive(Debug)]
pub struct Tracers {
pub global_interests: Interests,
pub custom_levels: AHashMap<EventType, Level>,
pub tracers: Vec<Tracer>,
}
impl Tracers {
pub fn parse(config: &mut Config) -> Self {
let mut tracers = Vec::new();
// Parse custom logging levels
let mut custom_levels = AHashMap::new();
for event_name in config
.sub_keys("tracing.level", "")
.map(|s| s.to_string())
.collect::<Vec<_>>()
{
if let Some(event_type) =
config.try_parse_value::<EventType>(("tracing.level", &event_name), &event_name)
{
if let Some(level) =
config.property_require::<Level>(("tracing.level", &event_name))
{
custom_levels.insert(event_type, level);
}
}
}
// Parse tracers
let mut tracers: Vec<Tracer> = Vec::new();
let mut global_interests = Interests::default();
for tracer_id in config
.sub_keys("tracer", ".type")
.map(|s| s.to_string())
@@ -65,16 +118,8 @@ impl Tracers {
continue;
}
// Parse level
let level = Level::from_str(config.value(("tracer", id, "level")).unwrap_or("info"))
.map_err(|err| {
config.new_parse_error(
("tracer", id, "level"),
format!("Invalid log level: {err}"),
)
})
.unwrap_or(Level::Info);
match config
// Parse tracer
let typ = match config
.value(("tracer", id, "type"))
.unwrap_or_default()
.to_string()
@@ -85,61 +130,65 @@ impl Tracers {
.value_require(("tracer", id, "path"))
.map(|s| s.to_string())
{
let prefix = config.value(("tracer", id, "prefix")).unwrap_or("stalwart");
let appender =
match config.value(("tracer", id, "rotate")).unwrap_or("daily") {
"daily" => tracing_appender::rolling::daily(path, prefix),
"hourly" => tracing_appender::rolling::hourly(path, prefix),
"minutely" => tracing_appender::rolling::minutely(path, prefix),
"never" => tracing_appender::rolling::never(path, prefix),
TracerType::Log(LogTracer {
path,
prefix: config
.value(("tracer", id, "prefix"))
.unwrap_or("stalwart")
.to_string(),
rotate: match config.value(("tracer", id, "rotate")).unwrap_or("daily")
{
"daily" => RotationStrategy::Daily,
"hourly" => RotationStrategy::Hourly,
"minutely" => RotationStrategy::Minutely,
"never" => RotationStrategy::Never,
rotate => {
let appender = tracing_appender::rolling::daily(path, prefix);
let err = format!("Invalid rotate value: {rotate}");
let err = format!("Invalid rotation strategy: {rotate}");
config.new_parse_error(("tracer", id, "rotate"), err);
appender
RotationStrategy::Daily
}
};
tracers.push(Tracer::Log {
id: id.to_string(),
level,
appender,
},
ansi: config
.property_or_default(("tracer", id, "ansi"), "false")
.unwrap_or(false),
});
multiline: config
.property_or_default(("tracer", id, "multiline"), "false")
.unwrap_or(false),
})
} else {
continue;
}
}
"stdout" => {
tracers.push(Tracer::Stdout {
id: id.to_string(),
level,
ansi: config
.property_or_default(("tracer", id, "ansi"), "true")
.unwrap_or(true),
});
}
"console" | "stdout" | "stderr" => TracerType::Console(ConsoleTracer {
ansi: config
.property_or_default(("tracer", id, "ansi"), "true")
.unwrap_or(true),
multiline: config
.property_or_default(("tracer", id, "multiline"), "true")
.unwrap_or(true),
buffered: config
.property_or_default(("tracer", id, "buffered"), "true")
.unwrap_or(true),
}),
"otel" | "open-telemetry" => {
match config
.value_require(("tracer", id, "transport"))
.unwrap_or_default()
{
"gprc" => {
let mut exporter = opentelemetry_otlp::new_exporter().tonic();
if let Some(endpoint) = config.value(("tracer", id, "endpoint")) {
exporter = exporter.with_endpoint(endpoint);
}
tracers.push(Tracer::Otel {
id: id.to_string(),
level,
tracer: OtelTracer::Gprc(exporter),
});
}
"gprc" => TracerType::Otel(OtelTracer {
endpoint: config
.value(("tracer", id, "endpoint"))
.unwrap_or_default()
.to_string(),
headers: Default::default(),
is_http: false,
}),
"http" => {
if let Some(endpoint) = config
.value_require(("tracer", id, "endpoint"))
.map(|s| s.to_string())
{
let mut headers = HashMap::new();
let mut headers = AHashMap::new();
let mut err = None;
for (_, value) in config.values(("tracer", id, "headers")) {
if let Some((key, value)) = value.split_once(':') {
@@ -157,38 +206,31 @@ impl Tracers {
config.new_parse_error(("tracer", id, "headers"), err);
}
let mut exporter = opentelemetry_otlp::new_exporter()
.http()
.with_endpoint(endpoint);
if !headers.is_empty() {
exporter = exporter.with_headers(headers);
}
tracers.push(Tracer::Otel {
id: id.to_string(),
level,
tracer: OtelTracer::Http(exporter),
});
TracerType::Otel(OtelTracer {
endpoint,
headers,
is_http: true,
})
} else {
continue;
}
}
"" => {}
transport => {
let err = format!("Invalid transport: {transport}");
config.new_parse_error(("tracer", id, "transport"), err);
continue;
}
}
}
"journal" => {
if !tracers.iter().any(|t| matches!(t, Tracer::Journal { .. })) {
tracers.push(Tracer::Journal {
id: id.to_string(),
level,
});
if !tracers.iter().any(|t| matches!(t.typ, TracerType::Journal)) {
TracerType::Journal
} else {
config.new_build_error(
("tracer", id, "type"),
"Only one journal tracer is allowed".to_string(),
);
continue;
}
}
unknown => {
@@ -196,10 +238,181 @@ impl Tracers {
("tracer", id, "type"),
format!("Unknown tracer type: {unknown}"),
);
continue;
}
};
// Create tracer
let mut tracer = Tracer {
id: id.to_string(),
interests: Default::default(),
lossy: config
.property_or_default(("tracer", id, "lossy"), "false")
.unwrap_or(false),
typ,
};
// Parse level
let level = Level::from_str(config.value(("tracer", id, "level")).unwrap_or("info"))
.map_err(|err| {
config.new_parse_error(
("tracer", id, "level"),
format!("Invalid log level: {err}"),
)
})
.unwrap_or(Level::Info);
// Parse disabled events
let mut disabled_events = AHashSet::new();
for (_, event_type) in config.properties::<EventType>(("tracer", id, "disabled-events"))
{
disabled_events.insert(event_type);
}
// Build interests lists
for event_type in EventType::variants() {
if !disabled_events.contains(&event_type) {
let event_level = custom_levels
.get(&event_type)
.copied()
.unwrap_or(event_type.level());
if event_level <= level {
tracer.interests.set(event_type);
global_interests.set(event_type);
}
}
}
if !tracer.interests.is_empty() {
tracers.push(tracer);
} else {
config.new_build_warning(("tracer", "id"), "No events enabled for tracer");
}
}
Tracers { tracers }
// Parse webhooks
for id in config
.sub_keys("webhook", ".url")
.map(|s| s.to_string())
.collect::<Vec<_>>()
{
if let Some(webhook) = parse_webhook(config, &id, &mut global_interests) {
tracers.push(webhook);
}
}
// Add default tracer if none were found
if tracers.is_empty() {
for event_type in EventType::variants() {
let event_level = custom_levels
.get(&event_type)
.copied()
.unwrap_or(event_type.level());
if event_level <= Level::Info {
global_interests.set(event_type);
}
}
tracers.push(Tracer {
id: "default".to_string(),
interests: global_interests.clone(),
typ: TracerType::Console(ConsoleTracer {
ansi: true,
multiline: true,
buffered: true,
}),
lossy: false,
});
}
Tracers {
tracers,
global_interests,
custom_levels,
}
}
}
fn parse_webhook(
config: &mut Config,
id: &str,
global_interests: &mut Interests,
) -> Option<Tracer> {
let mut headers = HeaderMap::new();
for (header, value) in config
.values(("webhook", id, "headers"))
.map(|(_, v)| {
if let Some((k, v)) = v.split_once(':') {
Ok((
HeaderName::from_str(k.trim()).map_err(|err| {
format!("Invalid header found in property \"webhook.{id}.headers\": {err}",)
})?,
HeaderValue::from_str(v.trim()).map_err(|err| {
format!("Invalid header found in property \"webhook.{id}.headers\": {err}",)
})?,
))
} else {
Err(format!(
"Invalid header found in property \"webhook.{id}.headers\": {v}",
))
}
})
.collect::<Result<Vec<(HeaderName, HeaderValue)>, String>>()
.map_err(|e| config.new_parse_error(("webhook", id, "headers"), e))
.unwrap_or_default()
{
headers.insert(header, value);
}
headers.insert(CONTENT_TYPE, "application/json".parse().unwrap());
if let (Some(name), Some(secret)) = (
config.value(("webhook", id, "auth.username")),
config.value(("webhook", id, "auth.secret")),
) {
headers.insert(
AUTHORIZATION,
format!("Basic {}", STANDARD.encode(format!("{}:{}", name, secret)))
.parse()
.unwrap(),
);
}
// Build tracer
let mut tracer = Tracer {
id: id.to_string(),
interests: Default::default(),
lossy: config
.property_or_default(("webhook", id, "lossy"), "false")
.unwrap_or(false),
typ: TracerType::Webhook(WebhookTracer {
url: config.value_require(("webhook", id, "url"))?.to_string(),
timeout: config
.property_or_default(("webhook", id, "timeout"), "30s")
.unwrap_or_else(|| Duration::from_secs(30)),
tls_allow_invalid_certs: config
.property_or_default(("webhook", id, "allow-invalid-certs"), "false")
.unwrap_or_default(),
headers,
key: config
.value(("webhook", id, "signature-key"))
.unwrap_or_default()
.to_string(),
throttle: config
.property_or_default(("webhook", id, "throttle"), "1s")
.unwrap_or_else(|| Duration::from_secs(1)),
}),
};
// Parse webhook events
for (_, event_type) in config.properties::<EventType>(("webhook", id, "events")) {
tracer.interests.set(event_type);
global_interests.set(event_type);
}
if !tracer.interests.is_empty() {
Some(tracer)
} else {
config.new_build_warning(("webhook", id), "No events enabled for webhook");
None
}
}

2
crates/common/src/listener/listen.rs
View File

@@ -46,7 +46,7 @@ impl Server {
limiter: ConcurrencyLimiter::new(self.max_connections),
acceptor,
shutdown_rx,
id_generator: self.id_generator,
span_id_gen: self.span_id_gen,
});
let is_tls = matches!(instance.acceptor, TcpAcceptor::Tls { implicit, .. } if implicit);
let is_https = is_tls && self.protocol == ServerProtocol::Http;

8
crates/common/src/listener/mod.rs
View File

@@ -37,7 +37,7 @@ pub struct ServerInstance {
pub limiter: ConcurrencyLimiter,
pub proxy_networks: Vec<IpAddrMask>,
pub shutdown_rx: watch::Receiver<bool>,
pub id_generator: Arc<SnowflakeIdGenerator>,
pub span_id_gen: Arc<SnowflakeIdGenerator>,
}
#[derive(Default)]
@@ -109,7 +109,7 @@ pub trait SessionManager: Sync + Send + 'static + Clone {
Ok(stream) => {
// Generate sessionId
session.session_id =
session.instance.id_generator.generate().unwrap_or_default();
session.instance.span_id_gen.generate().unwrap_or_default();
session_id = session.session_id;
trc::event!(
@@ -151,7 +151,7 @@ pub trait SessionManager: Sync + Send + 'static + Clone {
TcpAcceptorResult::Plain(stream) => {
// Generate sessionId
session.session_id =
session.instance.id_generator.generate().unwrap_or_default();
session.instance.span_id_gen.generate().unwrap_or_default();
session_id = session.session_id;
trc::event!(
@@ -170,7 +170,7 @@ pub trait SessionManager: Sync + Send + 'static + Clone {
}
} else {
// Generate sessionId
session.session_id = session.instance.id_generator.generate().unwrap_or_default();
session.session_id = session.instance.span_id_gen.generate().unwrap_or_default();
session_id = session.session_id;
trc::event!(

5
crates/common/src/manager/boot.rs
View File

@@ -12,7 +12,6 @@ use store::{
rand::{distributions::Alphanumeric, thread_rng, Rng},
Stores,
};
use tracing_appender::non_blocking::WorkerGuard;
use utils::{
config::{Config, ConfigKey},
failed, UnwrapFailure,
@@ -32,7 +31,6 @@ pub struct BootManager {
pub config: Config,
pub core: SharedCore,
pub servers: Servers,
pub guards: Option<Vec<WorkerGuard>>,
}
const HELP: &str = r#"Stalwart Mail Server
@@ -164,7 +162,7 @@ impl BootManager {
}
// Enable tracing
let guards = Tracers::parse(&mut config).enable(&mut config);
Tracers::parse(&mut config).enable();
match import_export {
ImportExport::None => {
@@ -323,7 +321,6 @@ impl BootManager {
BootManager {
core,
guards,
config,
servers,
}

6
crates/common/src/manager/reload.rs
View File

@@ -23,6 +23,7 @@ use super::config::{ConfigManager, Patterns};
pub struct ReloadResult {
pub config: Config,
pub new_core: Option<Core>,
pub tracers: Option<Tracers>,
}
impl Core {
@@ -75,6 +76,7 @@ impl Core {
Ok(ReloadResult {
config,
new_core: core.into(),
tracers: None,
})
}
@@ -82,7 +84,7 @@ impl Core {
let mut config = self.storage.config.build_config("").await?;
// Parse tracers
Tracers::parse(&mut config);
let tracers = Tracers::parse(&mut config);
// Load stores
let mut stores = Stores {
@@ -136,6 +138,7 @@ impl Core {
ReloadResult {
config,
new_core: core.into(),
tracers: tracers.into(),
}
} else {
config.into()
@@ -148,6 +151,7 @@ impl From<Config> for ReloadResult {
Self {
config,
new_core: None,
tracers: None,
}
}
}

141
crates/common/src/tracing/log.rs Normal file
View File

@@ -0,0 +1,141 @@
/*
* SPDX-FileCopyrightText: 2020 Stalwart Labs Ltd <hello@stalw.art>
*
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use std::{path::PathBuf, time::SystemTime};
use crate::config::tracers::{LogTracer, RotationStrategy};
use mail_parser::DateTime;
use tokio::{
fs::{File, OpenOptions},
io::BufWriter,
};
use trc::{fmt::FmtWriter, subscriber::SubscriberBuilder, ServerEvent};
pub(crate) fn spawn_log_tracer(builder: SubscriberBuilder, settings: LogTracer) {
let mut tx = builder.register();
tokio::spawn(async move {
if let Some(writer) = settings.build_writer().await {
let mut buf = FmtWriter::new(writer)
.with_ansi(settings.ansi)
.with_multiline(settings.multiline);
let mut roatation_timestamp = settings.next_rotation();
while let Some(events) = tx.recv().await {
for event in events {
// Check if we need to rotate the log file
if roatation_timestamp != 0 && event.inner.timestamp > roatation_timestamp {
if let Err(err) = buf.flush().await {
trc::event!(
Server(ServerEvent::TracingError),
Reason = err.to_string(),
Details = "Failed to flush log buffer"
);
}
if let Some(writer) = settings.build_writer().await {
buf.update_writer(writer);
roatation_timestamp = settings.next_rotation();
} else {
return;
};
}
if let Err(err) = buf.write(&event).await {
trc::event!(
Server(ServerEvent::TracingError),
Reason = err.to_string(),
Details = "Failed to write event to log"
);
return;
}
}
if let Err(err) = buf.flush().await {
trc::event!(
Server(ServerEvent::TracingError),
Reason = err.to_string(),
Details = "Failed to flush log buffer"
);
}
}
}
});
}
impl LogTracer {
pub async fn build_writer(&self) -> Option<BufWriter<File>> {
let now = DateTime::from_timestamp(
SystemTime::now()
.duration_since(SystemTime::UNIX_EPOCH)
.map_or(0, |d| d.as_secs()) as i64,
);
let file_name = match self.rotate {
RotationStrategy::Daily => {
format!(
"{}.{:04}-{:02}-{:02}",
self.prefix, now.year, now.month, now.day
)
}
RotationStrategy::Hourly => {
format!(
"{}.{:04}-{:02}-{:02}T{:02}",
self.prefix, now.year, now.month, now.day, now.hour
)
}
RotationStrategy::Minutely => {
format!(
"{}.{:04}-{:02}-{:02}T{:02}:{:02}",
self.prefix, now.year, now.month, now.day, now.hour, now.minute
)
}
RotationStrategy::Never => self.prefix.clone(),
};
let path = PathBuf::from(&self.path).join(file_name);
match OpenOptions::new()
.create(true)
.append(true)
.open(&path)
.await
{
Ok(writer) => Some(BufWriter::new(writer)),
Err(err) => {
trc::event!(
Server(ServerEvent::TracingError),
Details = "Failed to create log file",
Path = path.to_string_lossy().into_owned(),
Reason = err.to_string(),
);
None
}
}
}
pub fn next_rotation(&self) -> u64 {
let mut now = DateTime::from_timestamp(
SystemTime::now()
.duration_since(SystemTime::UNIX_EPOCH)
.map_or(0, |d| d.as_secs()) as i64,
);
now.second = 0;
match self.rotate {
RotationStrategy::Daily => {
now.hour = 0;
now.minute = 0;
now.to_timestamp() as u64 + 86400
}
RotationStrategy::Hourly => {
now.minute = 0;
now.to_timestamp() as u64 + 3600
}
RotationStrategy::Minutely => now.to_timestamp() as u64 + 60,
RotationStrategy::Never => 0,
}
}
}

116
crates/common/src/tracing/mod.rs
View File

@@ -4,43 +4,102 @@
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
pub mod log;
pub mod stdout;
//pub mod webhook;
use opentelemetry::KeyValue;
use opentelemetry_sdk::{
trace::{self, Sampler},
Resource,
};
use opentelemetry_semantic_conventions::resource::{SERVICE_NAME, SERVICE_VERSION};
use stdout::spawn_stdout_tracer;
use tracing_appender::non_blocking::WorkerGuard;
use tracing_subscriber::{
layer::SubscriberExt, util::SubscriberInitExt, EnvFilter, Layer, Registry,
};
use trc::subscriber::SubscriberBuilder;
use utils::config::Config;
use log::spawn_log_tracer;
use stdout::spawn_console_tracer;
use trc::{collector::Collector, subscriber::SubscriberBuilder};
use crate::config::tracers::{OtelTracer, Tracer, Tracers};
use crate::config::tracers::{ConsoleTracer, TracerType, Tracers};
impl Tracer {
pub fn spawn(self) {
match self {
Tracer::Stdout { id, level, ansi } => {
spawn_stdout_tracer(SubscriberBuilder::new(id).with_level(level), ansi);
impl Tracers {
pub fn enable(self) {
// Spawn tracers
for tracer in self.tracers {
tracer.typ.spawn(
SubscriberBuilder::new(tracer.id)
.with_interests(tracer.interests)
.with_lossy(tracer.lossy),
);
}
// Update global collector
Collector::set_interests(self.global_interests);
Collector::update_custom_levels(self.custom_levels);
Collector::reload();
}
pub fn update(self) {
// Remove tracers that are no longer active
let active_subscribers = Collector::get_subscribers();
for subscribed_id in &active_subscribers {
if !self
.tracers
.iter()
.any(|tracer| tracer.id == *subscribed_id)
{
Collector::remove_subscriber(subscribed_id.clone());
}
Tracer::Log {
id,
level,
appender,
ansi,
} => todo!(),
Tracer::Journal { id, level } => todo!(),
Tracer::Otel { id, level, tracer } => todo!(),
}
// Activate new tracers or update existing ones
for tracer in self.tracers {
if active_subscribers.contains(&tracer.id) {
Collector::update_subscriber(tracer.id, tracer.interests, tracer.lossy);
} else {
tracer.typ.spawn(
SubscriberBuilder::new(tracer.id)
.with_interests(tracer.interests)
.with_lossy(tracer.lossy),
);
}
}
// Update global collector
Collector::set_interests(self.global_interests);
Collector::update_custom_levels(self.custom_levels);
Collector::reload();
}
#[cfg(feature = "test_mode")]
pub fn test_tracer(level: trc::Level) {
let mut interests = trc::subscriber::Interests::default();
for event in trc::EventType::variants() {
if event.level() <= level {
interests.set(event);
}
}
spawn_console_tracer(
SubscriberBuilder::new("stdout".to_string())
.with_interests(interests.clone())
.with_lossy(false),
ConsoleTracer {
ansi: true,
multiline: false,
buffered: true,
},
);
Collector::set_interests(interests);
Collector::reload();
}
}
impl TracerType {
pub fn spawn(self, builder: SubscriberBuilder) {
match self {
TracerType::Console(settings) => spawn_console_tracer(builder, settings),
TracerType::Log(settings) => spawn_log_tracer(builder, settings),
TracerType::Otel(_) => todo!(),
TracerType::Webhook(_) => todo!(),
TracerType::Journal => todo!(),
}
}
}
/*
impl Tracers {
pub fn enable(self, config: &mut Config) -> Option<Vec<WorkerGuard>> {
let mut layers: Option<Box<dyn Layer<Registry> + Sync + Send>> = None;
@@ -155,3 +214,4 @@ impl Tracers {
}
}
}
*/

82
crates/common/src/tracing/stdout.rs
View File

@@ -4,15 +4,91 @@
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-SEL
*/
use trc::{subscriber::SubscriberBuilder, Level};
use std::{
io::{stderr, Error},
pin::Pin,
task::{Context, Poll},
};
pub(crate) fn spawn_stdout_tracer(builder: SubscriberBuilder, ansi: bool) {
use crate::config::tracers::ConsoleTracer;
use std::io::Write;
use tokio::io::AsyncWrite;
use trc::{fmt::FmtWriter, subscriber::SubscriberBuilder};
pub(crate) fn spawn_console_tracer(builder: SubscriberBuilder, settings: ConsoleTracer) {
let mut tx = builder.register();
tokio::spawn(async move {
let mut buf = FmtWriter::new(StdErrWriter::default())
.with_ansi(settings.ansi)
.with_multiline(settings.multiline);
while let Some(events) = tx.recv().await {
for event in events {
eprintln!("{}", event);
let _ = buf.write(&event).await;
if !settings.buffered {
let _ = buf.flush().await;
}
}
if settings.buffered {
let _ = buf.flush().await;
}
}
});
}
const BUFFER_CAPACITY: usize = 4096;
pub struct StdErrWriter {
buffer: Vec<u8>,
}
impl AsyncWrite for StdErrWriter {
fn poll_write(
mut self: Pin<&mut Self>,
_: &mut Context<'_>,
bytes: &[u8],
) -> Poll<Result<usize, Error>> {
let bytes_len = bytes.len();
let buffer_len = self.buffer.len();
if buffer_len + bytes_len < BUFFER_CAPACITY {
self.buffer.extend_from_slice(bytes);
Poll::Ready(Ok(bytes_len))
} else if bytes_len > BUFFER_CAPACITY {
let result = stderr()
.write_all(&self.buffer)
.and_then(|_| stderr().write_all(bytes));
self.buffer.clear();
Poll::Ready(result.map(|_| bytes_len))
} else {
let result = stderr().write_all(&self.buffer);
self.buffer.clear();
self.buffer.extend_from_slice(bytes);
Poll::Ready(result.map(|_| bytes_len))
}
}
fn poll_flush(mut self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Result<(), Error>> {
Poll::Ready(if !self.buffer.is_empty() {
let result = stderr().write_all(&self.buffer);
self.buffer.clear();
result
} else {
Ok(())
})
}
fn poll_shutdown(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Result<(), Error>> {
Poll::Ready(Ok(()))
}
}
impl Default for StdErrWriter {
fn default() -> Self {
Self {
buffer: Vec::with_capacity(BUFFER_CAPACITY),
}
}
}