jmug/stoatchat
1
0
Fork 1
mirror of https://github.com/stoatchat/stoatchat.git synced 2026-07-14 05:26:59 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: add otel logging, integrate with tracing

Browse Source
This commit is contained in:
izzy
2026-01-24 01:27:32 +00:00
parent e896582a5c
commit 97511059a4
12 changed files with 134 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.mise/tasks/docker/start
View File

@@ -2,4 +2,4 @@
#MISE description="Start Docker containers"
set -e
docker compose up -d
docker compose up -d "$@"

2
.mise/tasks/docker/stop
View File

@@ -2,4 +2,4 @@
#MISE description="Stop Docker containers"
set -e
docker compose down
docker compose down "$@"

12
README.md
View File

@@ -109,6 +109,18 @@ If you'd like to change anything, create a `Revolt.overrides.toml` file and spec
> ports: !override
> - "14072:5672"
> - "14672:15672"
>
> victoria-metrics:
> ports: !override
> - "18428:8428"
>
> victoria-logs:
> ports: !override
> - "19428:9428"
>
> victoria-traces:
> ports: !override
> - "14428:10428"
> ```
>
> And corresponding Revolt configuration:

24
compose.yml
View File

@@ -1,3 +1,5 @@
name: stoat-backend
services:
# Redis
redis:
@@ -69,9 +71,29 @@ services:
MAILDEV_INCOMING_USER: smtp
MAILDEV_INCOMING_PASS: smtp
# LiveKit
livekit:
image: ghcr.io/stoatchat/livekit-server:v1.9.9
command: --config /etc/livekit.yml
network_mode: "host"
volumes:
- ./livekit.yml:/etc/livekit.yml
- ./livekit.yml:/etc/livekit.yml
# VM
victoria-metrics:
image: victoriametrics/victoria-metrics:latest
ports:
- 8428:8428
# VL
victoria-logs:
image: docker.io/victoriametrics/victoria-logs:v1.43.1
ports:
- 9428:9428
command: -storageDataPath=victoria-logs-data
# VT
victoria-traces:
image: docker.io/victoriametrics/victoria-traces:latest
ports:
- 10428:10428

1
crates/core/config/src/lib.rs
View File

@@ -471,7 +471,6 @@ pub async fn setup_logging(release: &'static str, dsn: String) -> Option<sentry:
std::env::set_var("ROCKET_ADDRESS", "0.0.0.0");
}
pretty_env_logger::init();
log::info!("Starting {release}");
if dsn.is_empty() {

24
crates/services/autumn/Cargo.toml
View File

@@ -39,10 +39,6 @@ serde = { version = "1.0", features = ["derive"] }
# Async runtime
tokio = { version = "1.0", features = ["full"] }
# Logging
tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["env-filter"] }
# Core crates
revolt-files = { version = "0.9.4", path = "../../core/files" }
revolt-config = { version = "0.9.4", path = "../../core/config" }
@@ -53,15 +49,25 @@ revolt-result = { version = "0.9.4", path = "../../core/result", features = [
"utoipa",
"axum",
] }
revolt-ratelimits = { version = "0.9.4", path = "../../core/ratelimits", features = ["axum"] }
revolt-ratelimits = { version = "0.9.4", path = "../../core/ratelimits", features = [
"axum",
] }
# Axum / web server
tempfile = "3.12.0"
axum-macros = "0.4.1"
axum_typed_multipart = "0.12.1"
axum = { version = "0.7.5", features = ["multipart"] }
tower-http = { version = "0.5.2", features = ["cors"] }
axum-macros = { workspace = true }
axum_typed_multipart = { workspace = true }
axum = { workspace = true }
tower-http = { workspace = true }
# OpenAPI & documentation generation
utoipa-scalar = { version = "0.1.0", features = ["axum"] }
utoipa = { version = "4.2.3", features = ["axum_extras", "ulid"] }
# OpenTelemetry
tracing = { workspace = true }
tracing-subscriber = { workspace = true }
opentelemetry = { workspace = true }
opentelemetry_sdk = { workspace = true }
opentelemetry-otlp = { workspace = true }
opentelemetry-appender-tracing = { workspace = true }

46
crates/services/autumn/src/api.rs
View File

@@ -27,7 +27,6 @@ use utoipa::ToSchema;
use crate::{exif::strip_metadata, metadata::generate_metadata, mime_type::determine_mime_type, AppState};
/// Build the API router
pub async fn router() -> Router<AppState> {
let config = config().await;
@@ -52,8 +51,6 @@ pub async fn router() -> Router<AppState> {
}
lazy_static! {
/// Short-lived file cache to allow us to populate different CDN regions without increasing bandwidth to S3 provider
/// Uploads will also be stored here to prevent immediately queued downloads from doing the entire round-trip
static ref S3_CACHE: moka::future::Cache<String, Result<Vec<u8>>> = moka::future::Cache::builder()
.weigher(|_key, value: &Result<Vec<u8>>| -> u32 {
std::mem::size_of::<Result<Vec<u8>>>() as u32 + if let Ok(vec) = value {
@@ -70,7 +67,6 @@ lazy_static! {
.build();
}
/// Retrieve hash information and file data by given hash
async fn retrieve_file_by_hash(hash: &FileHash) -> Result<Vec<u8>> {
if let Some(data) = S3_CACHE.get(&hash.id).await {
data
@@ -81,17 +77,15 @@ async fn retrieve_file_by_hash(hash: &FileHash) -> Result<Vec<u8>> {
}
}
/// Successful root response
#[derive(Serialize, Debug, ToSchema)]
pub struct RootResponse {
autumn: &'static str,
version: &'static str,
}
/// Capture crate version from Cargo
static CRATE_VERSION: &str = env!("CARGO_PKG_VERSION");
/// Root response from service
/// Get information about the service
#[utoipa::path(
get,
path = "/",
@@ -106,7 +100,6 @@ async fn root() -> Json<RootResponse> {
})
}
/// Empty handler for OPTIONS routes
async fn options() {}
/// Available tags to upload to
@@ -170,28 +163,19 @@ async fn upload_file(
Path(tag): Path<Tag>,
TypedMultipart(UploadPayload { mut file }): TypedMultipart<UploadPayload>,
) -> Result<Json<UploadResponse>> {
// Fetch configuration
let now = Instant::now();
let config = config().await;
// Keep track of processing time
let now = Instant::now();
// Extract the filename, or give it a generic name
let filename = file.metadata.file_name.unwrap_or("unnamed-file".to_owned());
// Load file to memory
let mut buf = Vec::<u8>::new();
report_internal_error!(file.contents.read_to_end(&mut buf))?;
// Take note of original file size
let original_file_size = buf.len();
// Ensure the file is not empty
if original_file_size < config.files.limit.min_file_size {
return Err(create_error!(FileTooSmall));
}
// Get user's file upload limits
let limits = user.limits().await;
let size_limit = *limits
.file_upload_size_limit
@@ -202,24 +186,19 @@ async fn upload_file(
return Err(create_error!(FileTooLarge { max: size_limit }));
}
// Generate sha256 hash
let original_hash = {
let mut hasher = sha2::Sha256::new();
hasher.update(&buf);
hasher.finalize()
};
// Generate an ID for this file
let id = if matches!(tag, Tag::emojis) {
ulid::Ulid::new().to_string()
} else {
nanoid::nanoid!(42)
};
// Determine the mime type for the file
let mime_type = determine_mime_type(&mut file.contents, &buf, &filename);
// Check blocklist for mime type
if config
.files
.blocked_mime_types
@@ -229,15 +208,11 @@ async fn upload_file(
return Err(create_error!(FileTypeNotAllowed));
}
// Determine metadata for the file
let metadata = generate_metadata(&file.contents, mime_type);
// Block non-images for non-attachment uploads
if !matches!(tag, Tag::attachments) && !matches!(metadata, Metadata::Image { .. }) {
return Err(create_error!(FileTypeNotAllowed));
}
// Find an existing hash and use that if possible
let file_hash_exists = if let Ok(file_hash) = db
.fetch_attachment_hash(&format!("{original_hash:02x}"))
.await
@@ -260,10 +235,8 @@ async fn upload_file(
false
};
// Strip metadata
let (buf, metadata) = strip_metadata(file.contents, buf, metadata, mime_type).await?;
// Virus scan files if ClamAV is configured
if matches!(metadata, Metadata::File)
&& (config.files.scan_mime_types.is_empty()
|| config.files.scan_mime_types.iter().any(|v| v == mime_type))
@@ -272,7 +245,6 @@ async fn upload_file(
return Err(create_error!(InternalError));
}
// Print file information for debug purposes
let new_file_size = buf.len() + AUTHENTICATION_TAG_SIZE_BYTES;
let processed_hash = {
let mut hasher = sha2::Sha256::new();
@@ -284,7 +256,6 @@ async fn upload_file(
tracing::info!("Received file {filename}\nOriginal hash: {original_hash:02x}\nOriginal size: {original_file_size} bytes\nMime type: {mime_type}\nMetadata: {metadata:?}\nProcessed file size: {new_file_size} bytes ({:.2}%).\nProcessed hash: {processed_hash:02x}\nProcessing took {time_to_process:?}", process_ratio * 100.0);
// Create hash entry in database
let file_hash = FileHash {
id: format!("{original_hash:02x}"),
processed_hash: format!("{processed_hash:02x}"),
@@ -300,21 +271,17 @@ async fn upload_file(
size: new_file_size as isize,
};
// Add attachment hash if it doesn't exist
if !file_hash_exists {
db.insert_attachment_hash(&file_hash).await?;
}
// Upload the file to S3 and commit nonce to database
let upload_start = Instant::now();
let nonce = upload_to_s3(&file_hash.bucket_id, &file_hash.id, &buf).await?;
db.set_attachment_hash_nonce(&file_hash.id, &nonce).await?;
// Debug information
let time_to_upload = Instant::now() - upload_start;
tracing::info!("Took {time_to_upload:?} to upload {new_file_size} bytes to S3.");
// Finally, create the file and return its ID
let tag: &'static str = tag.into();
db.insert_attachment(&file_hash.into_file(id.clone(), tag.to_owned(), filename, user.id))
.await?;
@@ -362,12 +329,10 @@ async fn fetch_preview(
let tag_str: &'static str = tag.clone().into();
let file = db.fetch_attachment(tag_str, &file_id).await?;
// Ignore deleted files
if file.deleted.is_some_and(|v| v) {
return Err(create_error!(NotFound));
}
// Ignore files that haven't been attached
if file.used_for.is_none() {
return Err(create_error!(NotFound));
}
@@ -376,7 +341,7 @@ async fn fetch_preview(
let is_animated = hash.content_type == "image/gif"; // TODO: extract this data from files
// Only process image files and don't process GIFs if not avatar or icon
// Process GIFs if avatar or icon
if !matches!(hash.metadata, Metadata::Image { .. })
|| (is_animated && !matches!(tag, Tag::avatars | Tag::icons))
{
@@ -385,10 +350,8 @@ async fn fetch_preview(
);
}
// Original image data
let data = retrieve_file_by_hash(&hash).await?;
// Read image and create thumbnail
let data = create_thumbnail(
decode_image(&mut Cursor::new(data), &file.content_type)?,
tag_str,
@@ -430,17 +393,14 @@ async fn fetch_file(
let tag: &'static str = tag.clone().into();
let file = db.fetch_attachment(tag, &file_id).await?;
// Ignore deleted files
if file.deleted.is_some_and(|v| v) {
return Err(create_error!(NotFound));
}
// Ignore files that haven't been attached
if file.used_for.is_none() {
return Err(create_error!(NotFound));
}
// Ensure filename is correct
if file_name != file.filename {
if file_name == "original" {
return Ok(

2
crates/services/autumn/src/clamav.rs
View File

@@ -3,7 +3,6 @@ use std::time::Duration;
use revolt_config::{config, report_internal_error};
use revolt_result::Result;
/// Initialise ClamAV
pub async fn init() {
let config = config().await;
@@ -32,7 +31,6 @@ pub async fn init() {
}
}
/// Scan for malware
pub async fn is_malware(buf: &[u8]) -> Result<bool> {
let config = config().await;
if config.files.clamd_host.is_empty() {

35
crates/services/autumn/src/exif.rs
View File

@@ -8,7 +8,6 @@ use revolt_result::{create_error, Result};
use tempfile::NamedTempFile;
use tokio::process::Command;
/// Strip EXIF data from given file and produce new file and metadata
pub async fn strip_metadata(
file: NamedTempFile,
buf: Vec<u8>,
@@ -37,20 +36,16 @@ pub async fn strip_metadata(
// }
// Apply orientation manually & strip all other EXIF data
"image/jpeg" | "image/png" | "image/avif" | "image/tiff" => {
// Create a reader
let mut cursor = Cursor::new(buf);
// Decode the image
let image = report_internal_error!(report_internal_error!(ImageReader::new(
&mut cursor
)
.with_guessed_format())?
.decode());
// Reset read position
cursor.set_position(0);
// Extract orientation data
let exif_reader = Reader::new();
let rotation = match exif_reader.read_from_container(&mut cursor) {
Ok(exif) => match exif.get_field(exif::Tag::Orientation, exif::In::PRIMARY) {
@@ -60,12 +55,10 @@ pub async fn strip_metadata(
_ => 0,
};
// Create a buffer to write to
let mut bytes: Vec<u8> = Vec::new();
let mut writer = Cursor::new(&mut bytes);
// Apply the EXIF rotation
// See https://jdhao.github.io/2019/07/31/image_rotation_exif_info/
// https://jdhao.github.io/2019/07/31/image_rotation_exif_info/
report_internal_error!(match &rotation {
2 => image?.fliph(),
3 => image?.rotate180(),
@@ -87,7 +80,6 @@ pub async fn strip_metadata(
},
))?;
// Calculate dimensions after rotation.
let (width, height) = match &rotation {
2 | 4 | 5 | 7 => (*height, *width),
_ => (*width, *height),
@@ -95,16 +87,13 @@ pub async fn strip_metadata(
Ok((bytes, Metadata::Image { width, height }))
}
// JXLs store EXIF data but we don't have the ability to write them
// TODO: JXLs store EXIF data but we don't have the ability to write them
"image/jxl" => Ok((buf, metadata)),
// All other images that cannot store EXIF data
// assume all other images that cannot store EXIF data
_ => Ok((buf, metadata)),
},
// Use ffmpeg to copy video stream and probe new metadata
Metadata::Video { .. } => match mime {
// Strip EXIF data by copying video stream
"video/mp4" | "video/webm" | "video/quicktime" => {
// Pick the correct file format for ffmpeg
let ext = match mime {
"video/mp4" => "mp4",
"video/webm" => "webm",
@@ -112,30 +101,22 @@ pub async fn strip_metadata(
_ => unreachable!(),
};
// Temporary output file
let mut out_file = report_internal_error!(NamedTempFile::new())?;
// Process the file with ffmpeg
report_internal_error!(
Command::new("ffmpeg")
.args([
// Overwrite the temporary file
"-y",
// Read original uploaded file
"-i",
file.path().to_str().ok_or(create_error!(InternalError))?,
// Strip any metadata
"-map_metadata",
"-map_metadata", // strip metadata
"-1",
// Copy video / audio data to new file
"-c:v",
"-c:v", // just copy the streams
"copy",
"-c:a",
"copy",
// Select correct file format
"-f",
ext,
// Save to new temporary file
out_file
.path()
.to_str()
@@ -145,19 +126,17 @@ pub async fn strip_metadata(
.await
)?;
// Probe the file again
let metadata = crate::metadata::generate_metadata(&out_file, mime);
// Read the file from disk
let mut buf = Vec::<u8>::new();
report_internal_error!(out_file.read_to_end(&mut buf))?;
Ok((buf, metadata))
}
// Assume all other video formats cannot store EXIF data
// assume all other video formats cannot store EXIF data
_ => Ok((buf, metadata)),
},
// all other file types don't store EXIF data
// assume all other file types don't store EXIF data
_ => Ok((buf, metadata)),
}
}

80
crates/services/autumn/src/main.rs
View File

@@ -6,6 +6,7 @@ use axum_macros::FromRef;
use revolt_database::{Database, DatabaseInfo};
use revolt_ratelimits::axum as ratelimiter;
use tokio::net::TcpListener;
use tower_http::trace::TraceLayer;
use utoipa::{
openapi::security::{ApiKey, ApiKeyValue, SecurityScheme},
Modify, OpenApi,
@@ -27,13 +28,32 @@ struct AppState {
#[tokio::main]
async fn main() -> Result<(), std::io::Error> {
// Configure logging and environment
revolt_config::configure!(files);
let logger_provider = init_logs();
// Wait for ClamAV
let otel_layer = OpenTelemetryTracingBridge::new(&logger_provider);
let filter_otel = EnvFilter::new("info")
.add_directive("hyper=off".parse().unwrap())
.add_directive("tonic=off".parse().unwrap())
.add_directive("h2=off".parse().unwrap())
.add_directive("reqwest=off".parse().unwrap());
let otel_layer = otel_layer.with_filter(filter_otel);
let filter_fmt = EnvFilter::new("info");
let fmt_layer = tracing_subscriber::fmt::layer()
.with_thread_names(true)
.with_filter(filter_fmt);
tracing_subscriber::registry()
.with(otel_layer)
.with(fmt_layer)
.init();
revolt_config::configure!(files);
clamav::init().await;
// Configure API schema
#[derive(OpenApi)]
#[openapi(
modifiers(&SecurityAddon),
@@ -76,7 +96,6 @@ async fn main() -> Result<(), std::io::Error> {
}
}
// Connect to the database
let db = DatabaseInfo::Auto.connect().await.unwrap();
let ratelimits = ratelimiter::RatelimitStorage::new(ratelimits::AutumnRatelimits);
@@ -85,7 +104,6 @@ async fn main() -> Result<(), std::io::Error> {
ratelimit_storage: ratelimits,
};
// Configure Axum and router
let app = Router::new()
.merge(Scalar::with_url("/scalar", ApiDoc::openapi()))
.nest("/", api::router().await)
@@ -94,10 +112,56 @@ async fn main() -> Result<(), std::io::Error> {
state.clone(),
ratelimiter::ratelimit_middleware,
))
.layer(TraceLayer::new_for_http())
.with_state(state);
// Configure TCP listener and bind
let address = SocketAddr::from((Ipv4Addr::UNSPECIFIED, 14704));
let listener = TcpListener::bind(&address).await?;
axum::serve(listener, app.into_make_service()).await
axum::serve(listener, app.into_make_service()).await?;
if let Err(e) = logger_provider.shutdown() {
panic!("logger provider failed to shut down");
}
Ok(())
}
use opentelemetry::trace::{TraceContextExt, Tracer};
use opentelemetry::KeyValue;
use opentelemetry::{global, InstrumentationScope};
use opentelemetry_appender_tracing::layer::OpenTelemetryTracingBridge;
use opentelemetry_otlp::{LogExporter, MetricExporter, Protocol, SpanExporter, WithExportConfig};
use opentelemetry_sdk::logs::SdkLoggerProvider;
use opentelemetry_sdk::metrics::SdkMeterProvider;
use opentelemetry_sdk::trace::SdkTracerProvider;
use opentelemetry_sdk::Resource;
use std::error::Error;
use std::sync::OnceLock;
use tracing::info;
use tracing_subscriber::prelude::*;
use tracing_subscriber::EnvFilter;
fn get_resource() -> Resource {
static RESOURCE: OnceLock<Resource> = OnceLock::new();
RESOURCE
.get_or_init(|| {
Resource::builder()
.with_service_name("basic-otlp-example-grpc")
.build()
})
.clone()
}
fn init_logs() -> SdkLoggerProvider {
let exporter = LogExporter::builder()
.with_http()
.with_endpoint("http://localhost:19428/insert/opentelemetry/v1/logs")
.with_protocol(Protocol::HttpBinary)
.build()
.expect("Failed to create log exporter");
SdkLoggerProvider::builder()
.with_resource(get_resource())
.with_batch_exporter(exporter)
.build()
}

2
crates/services/autumn/src/metadata.rs
View File

@@ -19,7 +19,6 @@ static SUPPORTED_IMAGE_MIME: [&str; 9] = [
"image/webp",
];
/// Generate metadata from file, using mime type as a hint
pub fn generate_metadata(f: &NamedTempFile, mime_type: &str) -> Metadata {
if SUPPORTED_IMAGE_MIME.contains(&mime_type) {
image_size(f)
@@ -44,7 +43,6 @@ pub fn generate_metadata(f: &NamedTempFile, mime_type: &str) -> Metadata {
}
}
/// Subroutine to ensure data isn't corrupted
pub fn validate_from_metadata(
reader: Cursor<Vec<u8>>,
metadata: Metadata,

4
crates/services/autumn/src/mime_type.rs
View File

@@ -1,15 +1,12 @@
use tempfile::NamedTempFile;
/// Determine the mime type of the given temporary file and filename
pub fn determine_mime_type(f: &mut NamedTempFile, buf: &[u8], file_name: &str) -> &'static str {
// Force certain extensions into particular mime types
if file_name.to_lowercase().ends_with(".apk") {
return "application/vnd.android.package-archive";
} else if file_name.to_lowercase().ends_with(".exe") {
return "application/vnd.microsoft.portable-executable";
}
// Use magic signatures to determine mime type
let kind = infer::get_from_path(f.path()).expect("file read successfully");
let mime_type = if let Some(kind) = kind {
kind.mime_type()
@@ -17,7 +14,6 @@ pub fn determine_mime_type(f: &mut NamedTempFile, buf: &[u8], file_name: &str) -
"application/octet-stream"
};
// See if the file is actually just plain Unicode/ASCII text
if mime_type == "application/octet-stream" && simdutf8::basic::from_utf8(buf).is_ok() {
if file_name.to_lowercase().ends_with(".svg") {
return "image/svg+xml";