test: ✨ add intensive logging and opentelemetry

backend/Cargo.toml

@@ -6,7 +6,7 @@ edition = "2021"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [workspace]
-members = ["entity","migration"]
+members = ["entity", "migration"]
 
 [dependencies]
 log = "0.4.18"
@@ -29,22 +29,28 @@ leaky-bucket = "1.0.1"
 hickory-resolver = "0.24.0"
 crossbeam-queue = "0.3.8"
 dashmap = "5.5.3"
+tracing-opentelemetry = "0.22.0"
+opentelemetry_sdk = { version = "0.21.1", features = ["rt-tokio"] }
+opentelemetry = "0.21.0"
+opentelemetry-stdout = { version = "0.2.0", features = ["metrics"] }
+opentelemetry-semantic-conventions = "0.13.0"
+opentelemetry-otlp = { version = "0.14.0", features = ["metrics"] }
 
 [dependencies.tokio-stream]
 version = "0.1.14"
-features =["sync"]
+features = ["sync"]
 
 [dependencies.tracing-subscriber]
 version = "0.3.17"
-features =["json"]
+features = ["json"]
 
 [dependencies.tracing]
 version = "0.1.40"
 features = ["async-await", "log"]
 
 [dependencies.tokio]
 version = "1.34.0"
-features =["macros", "rt-multi-thread", "full","time"]
+features = ["macros", "rt-multi-thread", "full", "time"]
 
 [dependencies.sea-orm]
 version = "0.12.2"
@@ -53,7 +59,7 @@ features = [
     "macros",
     "mock",
     "sqlx-sqlite",
-    "with-chrono"
+    "with-chrono",
 ]
 
 [dependencies.serde]
@@ -66,7 +72,7 @@ features = ["transport", "channel", "codegen", "tls", "prost"]
 
 [dependencies.spin]
 version = "0.9.8"
-features = ["mutex","spin_mutex","rwlock"]
+features = ["mutex", "spin_mutex", "rwlock"]
 
 [build-dependencies]
 tonic-build = "0.10.2"

backend/src/controller/judger/route/mod.rs

@@ -106,6 +106,11 @@ async fn discover<I: Routable + Send>(
     config: JudgerConfig,
     router: Weak<Router>,
 ) -> Result<(), Error> {
+    let span = tracing::span!(
+        tracing::Level::DEBUG,
+        "service discover",
+        name = config.name
+    );
     let mut instance = I::new(config)?;
     loop {
         match instance.discover().await {
@@ -131,10 +136,14 @@ async fn discover<I: Routable + Send>(
                     }
                 }
             }
-            RouteStatus::Wait(dur) => tokio::time::sleep(dur).await,
+            RouteStatus::Wait(dur) => {
+                log::trace!("Service Discovery halt for {} seconds", dur.as_secs());
+                tokio::time::sleep(dur).await
+            }
             _ => break,
         }
     }
+    drop(span);
     Ok(())
 }
 
@@ -197,6 +206,7 @@ pub struct Upstream {
     healthy: AtomicIsize,
     clients: SegQueue<JudgerIntercept>,
     connection: ConnectionDetail,
+    _live_span: tracing::Span,
 }
 
 impl Upstream {
@@ -217,6 +227,8 @@ impl Upstream {
             result.push((uuid, lang));
         }
 
+        let live_span = tracing::span!(tracing::Level::INFO, "judger livetime", uri = detail.uri);
+
         let clients = SegQueue::default();
         clients.push(client);
 
@@ -225,6 +237,7 @@ impl Upstream {
                 healthy: AtomicIsize::new(HEALTHY_THRESHOLD),
                 clients,
                 connection: detail,
+                _live_span: live_span,
             }),
             result,
         ))

backend/src/endpoint/submit.rs

@@ -162,8 +162,6 @@ impl SubmitSet for Arc<Server> {
                 .ok_or(Error::NotInDB("user"))?;
         }
 
-        // if problem
-
         let submit = SubmitBuilder::default()
             .code(req.info.code)
             .lang(lang)

backend/src/endpoint/user.rs

@@ -174,6 +174,8 @@ impl UserSet for Arc<Server> {
 
         let mut model: ActiveModel = Default::default();
 
+        log::info!("creating user({})", req.info.username);
+
         fill_active_model!(model, req.info, username);
 
         let hash = self.crypto.hash(req.info.password.as_str()).into();

backend/src/init/logger.rs

@@ -1,8 +1,117 @@
+use opentelemetry::{global, KeyValue};
+use opentelemetry_sdk::{
+    metrics::{
+        reader::{DefaultAggregationSelector, DefaultTemporalitySelector},
+        MeterProvider, PeriodicReader,
+    },
+    runtime,
+    trace::{BatchConfig, RandomIdGenerator, Sampler, Tracer},
+    Resource,
+};
+use opentelemetry_semantic_conventions::{
+    resource::{DEPLOYMENT_ENVIRONMENT, SERVICE_NAME, SERVICE_VERSION},
+    SCHEMA_URL,
+};
 use tracing::Level;
+use tracing_opentelemetry::{MetricsLayer, OpenTelemetryLayer};
+use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
 
 use super::config::GlobalConfig;
 
-pub fn init(config: &GlobalConfig) {
+fn resource() -> Resource {
+    Resource::from_schema_url(
+        [
+            KeyValue::new(SERVICE_NAME, "mdoj-backend"),
+            KeyValue::new(SERVICE_VERSION, env!("CARGO_PKG_VERSION")),
+            #[cfg(debug_assertions)]
+            KeyValue::new(DEPLOYMENT_ENVIRONMENT, "develop"),
+            #[cfg(not(debug_assertions))]
+            KeyValue::new(DEPLOYMENT_ENVIRONMENT, "production"),
+        ],
+        SCHEMA_URL,
+    )
+}
+
+fn init_tracer() -> Tracer {
+    opentelemetry_otlp::new_pipeline()
+        .tracing()
+        .with_trace_config(
+            opentelemetry_sdk::trace::Config::default()
+                // Customize sampling strategy
+                .with_sampler(Sampler::ParentBased(Box::new(Sampler::TraceIdRatioBased(
+                    1.0,
+                ))))
+                // If export trace to AWS X-Ray, you can use XrayIdGenerator
+                .with_id_generator(RandomIdGenerator::default())
+                .with_resource(resource()),
+        )
+        .with_batch_config(BatchConfig::default())
+        .with_exporter(opentelemetry_otlp::new_exporter().tonic())
+        .install_batch(runtime::Tokio)
+        .unwrap()
+}
+
+fn init_meter_provider() -> MeterProvider {
+    #[cfg(debug_assertions)]
+    let exporter = opentelemetry_otlp::new_exporter()
+        .tonic()
+        .build_metrics_exporter(
+            Box::new(DefaultAggregationSelector::new()),
+            Box::new(DefaultTemporalitySelector::new()),
+        )
+        .unwrap();
+    #[cfg(debug_assertions)]
+    let otlp_reader = PeriodicReader::builder(exporter, runtime::Tokio)
+        .with_interval(std::time::Duration::from_secs(30))
+        .build();
+
+    // For debugging in development
+    let stdout_reader = PeriodicReader::builder(
+        opentelemetry_stdout::MetricsExporter::default(),
+        runtime::Tokio,
+    )
+    .build();
+
+    let meter_provider = MeterProvider::builder()
+        .with_resource(resource())
+        .with_reader(stdout_reader);
+
+    #[cfg(debug_assertions)]
+    let meter_provider = meter_provider.with_reader(otlp_reader);
+
+    let meter_provider = meter_provider.build();
+
+    global::set_meter_provider(meter_provider.clone());
+
+    meter_provider
+}
+
+pub struct OtelGuard {
+    meter_provider: MeterProvider,
+}
+
+impl Drop for OtelGuard {
+    fn drop(&mut self) {
+        if let Err(err) = self.meter_provider.shutdown() {
+            eprintln!("{err:?}");
+        }
+        opentelemetry::global::shutdown_tracer_provider();
+    }
+}
+fn init_tracing_subscriber(level: Level) -> OtelGuard {
+    let meter_provider = init_meter_provider();
+
+    tracing_subscriber::registry()
+        .with(tracing_subscriber::filter::LevelFilter::from_level(level))
+        .with(tracing_subscriber::fmt::layer())
+        .with(MetricsLayer::new(meter_provider.clone()))
+        .with(OpenTelemetryLayer::new(init_tracer()))
+        .init();
+
+    OtelGuard { meter_provider }
+}
+
+pub fn init(config: &GlobalConfig) -> OtelGuard {
     let level = match config.log_level {
         0 => Level::TRACE,
         1 => Level::DEBUG,
@@ -12,9 +121,5 @@ pub fn init(config: &GlobalConfig) {
         _ => Level::INFO,
     };
 
-    tracing_subscriber::fmt()
-        .json()
-        .with_max_level(level)
-        .with_current_span(false)
-        .init();
+    init_tracing_subscriber(level)
 }

backend/src/init/mod.rs

@@ -1,12 +1,12 @@
-use self::config::GlobalConfig;
+use self::{config::GlobalConfig, logger::OtelGuard};
 
 pub mod config;
 pub mod db;
 pub mod logger;
 
-pub async fn new() -> GlobalConfig {
+pub async fn new() -> (GlobalConfig, OtelGuard) {
     let config = config::init().await;
-    logger::init(&config);
+    let olp_guard = logger::init(&config);
     db::init(&config).await;
-    config
+    (config, olp_guard)
 }

backend/src/main.rs

@@ -7,7 +7,6 @@ pub mod server;
 
 #[tokio::main]
 async fn main() {
-    init::new().await;
-    log::info!("starting server");
-    server::Server::start().await;
+    let server = server::Server::new().await;
+    server.start().await;
 }

backend/src/server.rs

@@ -11,7 +11,11 @@ use crate::{
         testcase_set_server::TestcaseSetServer, token_set_server::TokenSetServer,
         user_set_server::UserSetServer,
     },
-    init::config::{self},
+    init::{
+        self,
+        config::{self, GlobalConfig},
+        logger::{self, OtelGuard},
+    },
 };
 
 const MAX_FRAME_SIZE: u32 = 1024 * 1024 * 8;
@@ -21,11 +25,18 @@ pub struct Server {
     pub submit: judger::SubmitController,
     pub dup: duplicate::DupController,
     pub crypto: crypto::CryptoController,
+    config: GlobalConfig,
+    identity: transport::Identity,
+    _otp_guard: OtelGuard,
 }
 
 impl Server {
-    pub async fn start() {
+    pub async fn new() -> Arc<Self> {
         let config = config::init().await;
+
+        init::db::init(&config).await;
+        let otp_guard = logger::init(&config);
+
         log::info!("Loading TLS certificate...");
         let cert = fs::read_to_string(&config.grpc.public_pem)
             .await
@@ -37,25 +48,30 @@ impl Server {
 
         log::info!("Constructing server...");
 
-        let server = Arc::new(Server {
+        Arc::new(Server {
             token: token::TokenController::new(),
             submit: judger::SubmitController::new(&config).await.unwrap(),
             dup: duplicate::DupController::default(),
             crypto: crypto::CryptoController::new(&config),
-        });
-
+            config,
+            identity,
+            _otp_guard: otp_guard,
+        })
+    }
+    pub async fn start(self: Arc<Self>) {
         transport::Server::builder()
-            .tls_config(transport::ServerTlsConfig::new().identity(identity))
+            // .accept_http1(true)
+            .tls_config(transport::ServerTlsConfig::new().identity(self.identity.clone()))
             .unwrap()
             .max_frame_size(Some(MAX_FRAME_SIZE))
-            .add_service(tonic_web::enable(ProblemSetServer::new(server.clone())))
-            .add_service(tonic_web::enable(EducationSetServer::new(server.clone())))
-            .add_service(tonic_web::enable(UserSetServer::new(server.clone())))
-            .add_service(tonic_web::enable(TokenSetServer::new(server.clone())))
-            .add_service(tonic_web::enable(ContestSetServer::new(server.clone())))
-            .add_service(tonic_web::enable(TestcaseSetServer::new(server.clone())))
-            .add_service(tonic_web::enable(SubmitSetServer::new(server)))
-            .serve(config.bind_address.parse().unwrap())
+            .add_service(tonic_web::enable(ProblemSetServer::new(self.clone())))
+            .add_service(tonic_web::enable(EducationSetServer::new(self.clone())))
+            .add_service(tonic_web::enable(UserSetServer::new(self.clone())))
+            .add_service(tonic_web::enable(TokenSetServer::new(self.clone())))
+            .add_service(tonic_web::enable(ContestSetServer::new(self.clone())))
+            .add_service(tonic_web::enable(TestcaseSetServer::new(self.clone())))
+            .add_service(tonic_web::enable(SubmitSetServer::new(self.clone())))
+            .serve(self.config.bind_address.clone().parse().unwrap())
             .await
             .unwrap();
     }

docker/production/docker-compose.yml

@@ -1,5 +1,8 @@
 version: '3.9'
 services:
+  # jaeger:
+  #   image: jaegertracing/all-in-one
+  #   ports:
   gateway: 
     container_name: mdoj-gateway
     build: ../../frontend