1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
// Copyright (C) 2026 Ordinary Labs, LLC.
//
// SPDX-License-Identifier: AGPL-3.0-only
use crate::server::OrdinaryAppRouter;
use async_compression::tokio::write::ZlibDecoder;
use axum::Router;
use axum::http::StatusCode;
use axum::routing::post;
use bytes::Bytes;
use ordinary_config::OrdinaryConfig;
use ordinary_utils::json::JsonValuable;
use serde::{Deserialize, Serialize};
use std::collections::BTreeMap;
use std::sync::Arc;
use tokio::io::AsyncWriteExt;
#[derive(Serialize, Deserialize, Debug)]
pub(crate) struct ClientEvent {
/// timestamp
pub ts: String,
/// timezone
pub tz: String,
/// log level
#[serde(skip_serializing)]
pub lvl: String,
/// version
pub v: String,
/// correlation id
pub cid: String,
/// memory id
pub mid: String,
/// path
pub p: String,
#[serde(skip_serializing_if = "Option::is_none")]
#[serde(default)]
/// query
pub q: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
#[serde(default)]
/// fragment
pub f: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
#[serde(default)]
/// flags
pub flg: Option<BTreeMap<String, String>>,
#[serde(skip_serializing_if = "Option::is_none")]
#[serde(default)]
/// fields
pub fld: Option<BTreeMap<String, String>>,
#[serde(skip_serializing)]
/// message
pub msg: String,
}
pub fn setup_routes(config: &Arc<OrdinaryConfig>) -> Option<OrdinaryAppRouter> {
if config.client_events == Some(true) {
let router = Router::new().route(
"/.ordinary/v1/events",
post(|body: Bytes| async move {
let mut decoder = ZlibDecoder::new(Vec::new());
if let Err(err) = decoder.write_all(body.as_ref()).await {
tracing::error!(%err);
return StatusCode::INTERNAL_SERVER_ERROR;
}
if let Err(err) = decoder.shutdown().await {
tracing::error!(%err);
return StatusCode::INTERNAL_SERVER_ERROR;
}
let mut events_slice = decoder.into_inner();
let events: Vec<ClientEvent> = match simd_json::from_slice(&mut events_slice) {
Ok(parsed) => parsed,
Err(err) => {
tracing::error!(%err);
return StatusCode::INTERNAL_SERVER_ERROR;
}
};
let client_span = tracing::info_span!("client");
client_span.in_scope(|| {
for event in events {
#[cfg(tracing_unstable)]
{
let level = event.lvl.as_str();
let message = event.msg.as_str();
if let Ok(value) = serde_json::to_value(&event) {
let event = JsonValuable(value);
match level {
"TRACE" => tracing::trace!(
evt = tracing::field::valuable(&event),
"{message}"
),
"DEBUG" => tracing::debug!(
evt = tracing::field::valuable(&event),
"{message}"
),
"INFO" => {
tracing::info!(
evt = tracing::field::valuable(&event),
"{message}"
);
}
"WARN" => {
tracing::warn!(
evt = tracing::field::valuable(&event),
"{message}"
);
}
_ => tracing::error!(
evt = tracing::field::valuable(&event),
"{message}"
),
}
}
}
#[cfg(not(tracing_unstable))]
{
match event.lvl.as_ref() {
"TRACE" => tracing::trace!(evt = debug(&event)),
"DEBUG" => tracing::debug!(evt = debug(&event)),
"INFO" => tracing::info!(evt = debug(&event)),
"WARN" => tracing::warn!(evt = debug(&event)),
_ => tracing::error!(evt = debug(&event)),
}
}
}
});
StatusCode::OK
}),
);
return Some(router);
}
None
}