Revamp observability

Cargo.toml

@@ -14,6 +14,7 @@ exclude = ["xtask"]
 members = [
     "crates/kitsune-cache",
     "crates/kitsune-captcha",
+    "crates/kitsune-config",
     "crates/kitsune-core",
     "crates/kitsune-db",
     "crates/kitsune-email",
@@ -22,6 +23,7 @@ members = [
     "crates/kitsune-http-signatures",
     "crates/kitsune-language",
     "crates/kitsune-messaging",
+    "crates/kitsune-observability",
     "crates/kitsune-retry-policies",
     "crates/kitsune-search",
     "crates/kitsune-storage",
@@ -41,4 +43,10 @@ edition = "2021"
 version = "0.0.1-pre.3"
 
 [patch.crates-io]
+# Patch the opentelemetry crate versions to get rid of a few tonic dependencies
+opentelemetry_api = { git = "https://github.com/open-telemetry/opentelemetry-rust.git", rev = "41e8d63652b323866c03981b4b2ca62b9b8d6d44" }
+opentelemetry-http = { git = "https://github.com/open-telemetry/opentelemetry-rust.git", rev = "41e8d63652b323866c03981b4b2ca62b9b8d6d44" }
+opentelemetry-otlp = { git = "https://github.com/open-telemetry/opentelemetry-rust.git", rev = "41e8d63652b323866c03981b4b2ca62b9b8d6d44" }
+opentelemetry_sdk = { git = "https://github.com/open-telemetry/opentelemetry-rust.git", rev = "41e8d63652b323866c03981b4b2ca62b9b8d6d44" }
+
 redis = { git = "https://github.com/aumetra/redis-rs.git", rev = "3c4ee09d432a69e1d87d66dcba14c519467c9b81" }

config.docker.toml

@@ -27,7 +27,6 @@ frontend-dir = "./kitsune-fe/dist"
 max-upload-size = 5242880          # 5MB
 media-proxy-enabled = false
 port = 5000
-prometheus-port = 9000
 request-timeout-secs = 60
 
 [search]
@@ -39,4 +38,4 @@ upload-dir = "./uploads"
 
 [url]
 scheme = "http"
-domain = "localhost:5000"
+domain = "localhost:5000"

config.example.toml

@@ -27,7 +27,6 @@ frontend-dir = "./kitsune-fe/dist"
 max-upload-size = 5242880          # 5MB
 media-proxy-enabled = false
 port = 5000
-prometheus-port = 9000
 request-timeout-secs = 60
 
 [search]

crates/kitsune-config/Cargo.toml

@@ -0,0 +1,11 @@
+[package]
+name = "kitsune-config"
+edition.workspace = true
+version.workspace = true
+
+[dependencies]
+eyre = "0.6.8"
+serde = { version = "1.0.189", features = ["derive"] }
+smol_str = { version = "0.2.0", features = ["serde"] }
+tokio = { version = "1.33.0", features = ["fs"] }
+toml = { version = "0.8.2", default-features = false, features = ["parse"] }

crates/kitsune-core/src/config.rs → crates/kitsune-config/src/lib.rs

@@ -117,6 +117,12 @@ pub struct MeiliSearchConfiguration {
     pub api_key: SmolStr,
 }
 
+#[derive(Clone, Deserialize, Serialize)]
+#[serde(rename_all = "kebab-case")]
+pub struct OpenTelemetryConfiguration {
+    pub http_endpoint: String,
+}
+
 #[derive(Clone, Deserialize, Serialize)]
 #[serde(rename_all = "kebab-case", tag = "type")]
 pub enum SearchConfiguration {
@@ -133,7 +139,6 @@ pub struct ServerConfiguration {
     pub media_proxy_enabled: bool,
     pub oidc: Option<OidcConfiguration>,
     pub port: u16,
-    pub prometheus_port: u16,
     pub request_timeout_secs: u64,
 }
 
@@ -179,6 +184,7 @@ pub struct Configuration {
     pub instance: InstanceConfiguration,
     pub job_queue: JobQueueConfiguration,
     pub messaging: MessagingConfiguration,
+    pub opentelemetry: Option<OpenTelemetryConfiguration>,
     pub server: ServerConfiguration,
     pub search: SearchConfiguration,
     pub storage: StorageConfiguration,

crates/kitsune-core/Cargo.toml

@@ -10,7 +10,9 @@ async-recursion = "1.0.5"
 async-stream = "0.3.5"
 async-trait = "0.1.74"
 athena = { path = "../../lib/athena" }
-autometrics = { version = "0.6.0", default-features = false }
+autometrics = { version = "0.6.0", default-features = false, features = [
+    "metrics",
+] }
 base64-simd = "0.8.0"
 bytes = "1.5.0"
 const_format = "0.2.32"
@@ -34,6 +36,7 @@ img-parts = "0.3.0"
 iso8601-timestamp = "0.2.12"
 kitsune-cache = { path = "../kitsune-cache" }
 kitsune-captcha = { path = "../kitsune-captcha" }
+kitsune-config = { path = "../kitsune-config" }
 kitsune-db = { path = "../kitsune-db" }
 kitsune-email = { path = "../kitsune-email" }
 kitsune-embed = { path = "../kitsune-embed" }
@@ -65,7 +68,6 @@ speedy-uuid = { path = "../../lib/speedy-uuid", features = ["diesel"] }
 thiserror = "1.0.50"
 time = "0.3.30"
 tokio = { version = "1.33.0", features = ["macros", "rt"] }
-toml = { version = "0.8.2", default-features = false, features = ["parse"] }
 tracing = "0.1.40"
 typed-builder = "0.18.0"
 url = "2.4.1"
@@ -75,7 +77,6 @@ zxcvbn = { version = "2.2.2", default-features = false }
 default = []
 mastodon-api = []
 meilisearch = ["kitsune-search/meilisearch"]
-metrics = ["autometrics/metrics"]
 
 [build-dependencies]
 vergen = { version = "8.2.5", features = ["build", "git", "gitcl"] }

crates/kitsune-core/src/activitypub/fetcher.rs

@@ -324,7 +324,6 @@ mod test {
     use super::MAX_FETCH_DEPTH;
     use crate::{
         activitypub::Fetcher,
-        config::FederationFilterConfiguration,
         error::{ApiError, Error},
         service::federation_filter::FederationFilterService,
         webfinger::Webfinger,
@@ -335,6 +334,7 @@ mod test {
     use hyper::{Body, Request, Response, StatusCode, Uri};
     use iso8601_timestamp::Timestamp;
     use kitsune_cache::NoopCache;
+    use kitsune_config::FederationFilterConfiguration;
     use kitsune_db::{model::account::Account, schema::accounts};
     use kitsune_http_client::Client;
     use kitsune_search::NoopSearchService;

crates/kitsune-core/src/lib.rs

@@ -14,7 +14,6 @@ extern crate tracing;
 
 pub mod activitypub;
 pub mod blocking;
-pub mod config;
 pub mod consts;
 pub mod error;
 pub mod event;
@@ -29,10 +28,6 @@ pub mod webfinger;
 
 use self::{
     activitypub::Fetcher,
-    config::{
-        CacheConfiguration, CaptchaConfiguration, Configuration, EmailConfiguration,
-        MessagingConfiguration, SearchConfiguration, StorageConfiguration,
-    },
     job::KitsuneContextRepo,
     resolve::PostResolver,
     service::{
@@ -48,6 +43,10 @@ use athena::JobQueue;
 use eyre::Context;
 use kitsune_cache::{ArcCache, InMemoryCache, NoopCache, RedisCache};
 use kitsune_captcha::{hcaptcha::Captcha as HCaptcha, mcaptcha::Captcha as MCaptcha, Captcha};
+use kitsune_config::{
+    CacheConfiguration, CaptchaConfiguration, Configuration, EmailConfiguration,
+    MessagingConfiguration, SearchConfiguration, StorageConfiguration,
+};
 use kitsune_db::PgPool;
 use kitsune_email::{
     lettre::{message::Mailbox, AsyncSmtpTransport, Tokio1Executor},

crates/kitsune-core/src/resolve/post.rs

@@ -78,7 +78,6 @@ mod test {
     use super::PostResolver;
     use crate::{
         activitypub::Fetcher,
-        config::FederationFilterConfiguration,
         job::KitsuneContextRepo,
         service::{
             account::AccountService, attachment::AttachmentService,
@@ -92,6 +91,7 @@ mod test {
     use diesel_async::RunQueryDsl;
     use hyper::{Body, Request, Response};
     use kitsune_cache::NoopCache;
+    use kitsune_config::FederationFilterConfiguration;
     use kitsune_db::{model::account::Account, schema::accounts};
     use kitsune_http_client::Client;
     use kitsune_search::NoopSearchService;

crates/kitsune-core/src/service/federation_filter.rs

@@ -1,5 +1,6 @@
-use crate::{config::FederationFilterConfiguration, error::FederationFilterError};
+use crate::error::FederationFilterError;
 use globset::{Glob, GlobSet, GlobSetBuilder};
+use kitsune_config::FederationFilterConfiguration;
 use kitsune_type::ap::{actor::Actor, Activity, Object};
 use std::sync::Arc;
 use url::Url;

crates/kitsune-http-client/src/lib.rs

@@ -336,6 +336,12 @@ pub struct Response {
 }
 
 impl Response {
+    /// Convert the response into its inner `hyper` representation
+    #[must_use]
+    pub fn into_inner(self) -> HyperResponse<BoxBody<Bytes, BoxError>> {
+        self.inner
+    }
+
     /// Read the body into a `Bytes`
     ///
     /// # Errors

crates/kitsune-observability/Cargo.toml

@@ -0,0 +1,29 @@
+[package]
+name = "kitsune-observability"
+edition.workspace = true
+version.workspace = true
+
+[dependencies]
+async-trait = "0.1.74"
+eyre = "0.6.8"
+hyper = { version = "0.14.27", default-features = false }
+kitsune-config = { path = "../kitsune-config" }
+kitsune-http-client = { path = "../kitsune-http-client" }
+metrics = "0.21.1"
+metrics-opentelemetry = { git = "https://github.com/aumetra/metrics-opentelemetry.git", rev = "7c3176266c215bb9a7cbc31b3c32f75a22824928" }
+metrics-tracing-context = "0.14.0"
+metrics-util = "0.15.1"
+opentelemetry = { version = "0.20.0", default-features = false, features = [
+    "rt-tokio",
+    "trace",
+] }
+opentelemetry-http = "0.9.0"
+opentelemetry-otlp = { version = "0.13.0", default-features = false, features = [
+    "http-proto",
+    "metrics",
+    "trace",
+] }
+tracing = "0.1.40"
+tracing-error = "0.2.0"
+tracing-opentelemetry = { version = "0.21.0", default-features = false }
+tracing-subscriber = "0.3.17"

crates/kitsune-observability/src/lib.rs

@@ -0,0 +1,116 @@
+use async_trait::async_trait;
+use eyre::Context;
+use hyper::body::Body;
+use kitsune_config::Configuration;
+use metrics_opentelemetry::OpenTelemetryRecorder;
+use metrics_tracing_context::{MetricsLayer, TracingContextLayer};
+use metrics_util::layers::Layer as _;
+use opentelemetry::{
+    metrics::{noop::NoopMeterProvider, Meter, MeterProvider},
+    runtime::Tokio,
+    trace::{noop::NoopTracer, Tracer},
+};
+use opentelemetry_http::{Bytes, HttpClient, HttpError, Request, Response};
+use opentelemetry_otlp::WithExportConfig;
+use std::{env, fmt};
+use tracing_error::ErrorLayer;
+use tracing_opentelemetry::{OpenTelemetryLayer, PreSampledTracer};
+use tracing_subscriber::{
+    filter::{LevelFilter, Targets},
+    layer::SubscriberExt,
+    Layer as _, Registry,
+};
+
+#[derive(Clone)]
+struct HttpClientAdapter {
+    inner: kitsune_http_client::Client,
+}
+
+impl fmt::Debug for HttpClientAdapter {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "HttpClientAdapter")
+    }
+}
+
+#[async_trait]
+impl HttpClient for HttpClientAdapter {
+    async fn send(&self, request: Request<Vec<u8>>) -> Result<Response<Bytes>, HttpError> {
+        let (parts, body) = request.into_parts();
+        let body = Body::from(body);
+        let request = Request::from_parts(parts, body);
+
+        let response = self.inner.execute(request).await?.into_inner();
+
+        let (parts, body) = response.into_parts();
+        let body = hyper::body::to_bytes(body).await?;
+
+        Ok(Response::from_parts(parts, body))
+    }
+}
+
+fn initialise_logging<T>(tracer: T) -> eyre::Result<()>
+where
+    T: Tracer + PreSampledTracer + Send + Sync + 'static,
+{
+    let env_filter = env::var("RUST_LOG")
+        .map_err(eyre::Report::from)
+        .and_then(|targets| targets.parse().context("Failed to parse RUST_LOG value"))
+        .unwrap_or_else(|_| Targets::default().with_default(LevelFilter::INFO));
+
+    let subscriber = Registry::default()
+        .with(tracing_subscriber::fmt::layer().with_filter(env_filter))
+        .with(ErrorLayer::default())
+        .with(OpenTelemetryLayer::new(tracer));
+
+    let subscriber = subscriber.with(MetricsLayer::new());
+
+    tracing::subscriber::set_global_default(subscriber)
+        .context("Couldn't install the global tracing subscriber")?;
+
+    Ok(())
+}
+
+fn initialise_metrics(meter: Meter) -> eyre::Result<()> {
+    let recorder = TracingContextLayer::all().layer(OpenTelemetryRecorder::new(meter));
+    metrics::set_boxed_recorder(Box::new(recorder))
+        .context("Couldn't install the global metrics recorder")?;
+
+    Ok(())
+}
+
+pub fn initialise(app_name: &'static str, config: &Configuration) -> eyre::Result<()> {
+    if let Some(ref opentelemetry_config) = config.opentelemetry {
+        let http_client = HttpClientAdapter {
+            inner: kitsune_http_client::Client::default(),
+        };
+
+        let tracer = opentelemetry_otlp::new_pipeline()
+            .tracing()
+            .with_exporter(
+                opentelemetry_otlp::new_exporter()
+                    .http()
+                    .with_endpoint(opentelemetry_config.http_endpoint.as_str())
+                    .with_http_client(http_client.clone()),
+            )
+            .install_batch(Tokio)?;
+
+        initialise_logging(tracer)?;
+
+        let meter_provider = opentelemetry_otlp::new_pipeline()
+            .metrics(Tokio)
+            .with_exporter(
+                opentelemetry_otlp::new_exporter()
+                    .http()
+                    .with_endpoint(opentelemetry_config.http_endpoint.as_str())
+                    .with_http_client(http_client.clone()),
+            )
+            .build()?;
+
+        initialise_metrics(meter_provider.meter(app_name))?;
+    } else {
+        initialise_logging(NoopTracer::new())?;
+        initialise_metrics(NoopMeterProvider::new().meter(app_name))?;
+    }
+
+    Ok(())
+}

kitsune-job-runner/Cargo.toml

@@ -8,8 +8,10 @@ athena = { path = "../lib/athena" }
 clap = { version = "4.4.6", features = ["derive"] }
 color-eyre = "0.6.2"
 deadpool-redis = "0.13.0"
+kitsune-config = { path = "../crates/kitsune-config" }
 kitsune-core = { path = "../crates/kitsune-core" }
 kitsune-db = { path = "../crates/kitsune-db" }
+kitsune-observability = { path = "../crates/kitsune-observability" }
 kitsune-retry-policies = { path = "../crates/kitsune-retry-policies" }
 mimalloc = "0.1.39"
 tokio = { version = "1.33.0", features = ["full"] }

kitsune-job-runner/src/lib.rs

@@ -2,9 +2,9 @@
 extern crate tracing;
 
 use athena::JobQueue;
+use kitsune_config::JobQueueConfiguration;
 use kitsune_core::{
     activitypub::Deliverer,
-    config::JobQueueConfiguration,
     job::{JobRunnerContext, KitsuneContextRepo},
     state::State as CoreState,
 };

kitsune-job-runner/src/main.rs

@@ -1,6 +1,7 @@
 use clap::Parser;
 use color_eyre::eyre;
-use kitsune_core::{config::Configuration, consts::VERSION};
+use kitsune_config::Configuration;
+use kitsune_core::consts::VERSION;
 use std::path::PathBuf;
 use tokio::fs;
 
@@ -24,6 +25,8 @@ async fn main() -> eyre::Result<()> {
     let raw_config = fs::read_to_string(args.config).await?;
     let config: Configuration = toml::from_str(&raw_config)?;
 
+    kitsune_observability::initialise(env!("CARGO_PKG_NAME"), &config)?;
+
     let db_pool = kitsune_db::connect(
         &config.database.url,
         config.database.max_connections as usize,