diff --git a/.mise/tasks/docker/start b/.mise/tasks/docker/start index cfcb0a9a..fcef730d 100755 --- a/.mise/tasks/docker/start +++ b/.mise/tasks/docker/start @@ -2,4 +2,4 @@ #MISE description="Start Docker containers" set -e -docker compose up -d +docker compose up -d "$@" diff --git a/.mise/tasks/docker/stop b/.mise/tasks/docker/stop index 51d39d61..5500c220 100755 --- a/.mise/tasks/docker/stop +++ b/.mise/tasks/docker/stop @@ -2,4 +2,4 @@ #MISE description="Stop Docker containers" set -e -docker compose down +docker compose down "$@" diff --git a/README.md b/README.md index 05baca7e..14605969 100644 --- a/README.md +++ b/README.md @@ -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: diff --git a/compose.yml b/compose.yml index e1e84b6f..e215efcf 100644 --- a/compose.yml +++ b/compose.yml @@ -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 \ No newline at end of file + - ./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 diff --git a/crates/core/config/src/lib.rs b/crates/core/config/src/lib.rs index 35c93b5a..065658dc 100644 --- a/crates/core/config/src/lib.rs +++ b/crates/core/config/src/lib.rs @@ -471,7 +471,6 @@ pub async fn setup_logging(release: &'static str, dsn: String) -> Option Router { let config = config().await; @@ -52,8 +51,6 @@ pub async fn router() -> Router { } 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>> = moka::future::Cache::builder() .weigher(|_key, value: &Result>| -> u32 { std::mem::size_of::>>() 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> { if let Some(data) = S3_CACHE.get(&hash.id).await { data @@ -81,17 +77,15 @@ async fn retrieve_file_by_hash(hash: &FileHash) -> Result> { } } -/// 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 { }) } -/// Empty handler for OPTIONS routes async fn options() {} /// Available tags to upload to @@ -170,28 +163,19 @@ async fn upload_file( Path(tag): Path, TypedMultipart(UploadPayload { mut file }): TypedMultipart, ) -> Result> { - // 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::::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( diff --git a/crates/services/autumn/src/clamav.rs b/crates/services/autumn/src/clamav.rs index 2994fdbd..c71ad03d 100644 --- a/crates/services/autumn/src/clamav.rs +++ b/crates/services/autumn/src/clamav.rs @@ -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 { let config = config().await; if config.files.clamd_host.is_empty() { diff --git a/crates/services/autumn/src/exif.rs b/crates/services/autumn/src/exif.rs index 23be9ac7..f87c679b 100644 --- a/crates/services/autumn/src/exif.rs +++ b/crates/services/autumn/src/exif.rs @@ -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, @@ -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 = 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::::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)), } } diff --git a/crates/services/autumn/src/main.rs b/crates/services/autumn/src/main.rs index cb936e0c..438d95df 100644 --- a/crates/services/autumn/src/main.rs +++ b/crates/services/autumn/src/main.rs @@ -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 = 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() } diff --git a/crates/services/autumn/src/metadata.rs b/crates/services/autumn/src/metadata.rs index 72b610e9..0375f5f9 100644 --- a/crates/services/autumn/src/metadata.rs +++ b/crates/services/autumn/src/metadata.rs @@ -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>, metadata: Metadata, diff --git a/crates/services/autumn/src/mime_type.rs b/crates/services/autumn/src/mime_type.rs index 8ad76351..41a71e50 100644 --- a/crates/services/autumn/src/mime_type.rs +++ b/crates/services/autumn/src/mime_type.rs @@ -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";