Upgrade deps and dump version to 0.6.0 (#18)

* chore(deps): cargo

* wip: use axum & cargo fmt

* wip: remove `once_cell`

* wip: upgrade opentelemetry

ref: open-telemetry/opentelemetry-rust#1000
ref: open-telemetry/opentelemetry-rust#2221
ref: open-telemetry/opentelemetry-rust#2085

* chore: cargo clippy

* chore(deps): upgrade go

* version: upgrade to 0.6.0

* chore(deps): use rkyv

* chore(deps): update

* chore(deps): update

* wip: split mods into crates

* wip: remove unused dependencies

* chore(ci): upgrade rust version in Dockerfile

* fix(ci): update cargo install path

* chore: tidy up main.rs

* fix(test): review snaps & fix tests

* chore(deps): update

* fix(cgroup): early stop when use `map_while`

* chore: init logger before setup cgroup

* chore(deps): ebpf version restore to v0.12.3

* chore(lazylock): use `LazyLock::force` to init cache

* fix(cache): remove unnecessary async

* chore(config): make default functions const

Cargo.toml

@@ -1,66 +1,76 @@
-[package]
-name = "seele"
-version = "0.2.0"
+[workspace]
+members = [ "crates/*" ]
+resolver = "2"
+
+[workspace.package]
 edition = "2021"
+version = "0.6.0"
 
 [profile.release]
+lto = "thin"
 opt-level = 3
 strip = true
-lto = true
 
-[dependencies]
+[workspace.dependencies]
 anyhow = "1.0"
 async-recursion = "1.1"
 async-trait = "0.1"
+axum = "0.8"
 base64 = "0.22"
-bincode = "2.0.0-rc.3"
-bytes = "1.7"
-chrono = { version = "0.4", features = ["serde"] }
-config = "0.14"
+bytes = "1.9"
+chrono = { version = "0.4", features = [ "serde" ] }
+config = "0.15"
 dbus = "0.9"
 duct = "0.13"
 either = "1.12"
 ellipse = "0.2"
 futures-util = "0.3"
-http = "1.1"
-http-cache = { version = "0.19", default-features = false, features = ["manager-moka"] }
-http-cache-reqwest = "0.14"
-indexmap = { version = "2.6", features = ["serde"] }
+http = "1.2"
+http-cache = { version = "0.20", default-features = false, features = [
+    "manager-moka"
+] }
+http-cache-reqwest = "0.15"
+indexmap = { version = "2.7", features = [ "serde" ] }
+insta = { version = "1.42", features = [ "glob", "redactions", "ron" ] }
 lapin = "2.5"
-libcgroups = "0.4"
-moka = { version = "0.12", features = ["future", "sync"] }
-nano-id = { version = "0.4", features = ["base62"] }
-nix = { version = "0.29", features = ["mount", "hostname", "signal"] }
+libcgroups = "0.5"
+map-macro = "0.3"
+moka = { version = "0.12", features = [ "future", "sync" ] }
+nano-id = { version = "0.4", features = [ "base62" ] }
+nix = { version = "0.29", features = [ "hostname", "mount", "signal" ] }
 num_cpus = "1"
-once_cell = "1"
+opentelemetry = "0.27"
+opentelemetry-otlp = "0.27"
+opentelemetry_sdk = { version = "0.27", features = [ "rt-tokio" ] }
+quick-js = { version = "0.4", features = [ "patched" ] }
+rand = { version = "0.9" }
 regex = "1"
-quick-js = { version = "0.4", features = ["patched"] }
-reqwest = { version = "0.12", features = ["stream", "multipart"] }
-reqwest-middleware = "0.3"
+reqwest = { version = "0.12", features = [ "multipart", "stream" ] }
+reqwest-middleware = "0.4"
 ring-channel = "0.12"
-serde = { version = "1.0", features = ["derive", "rc"] }
-serde_json = { version = "1.0", features = ["preserve_order"] }
+rkyv = "0.8"
+serde = { version = "1.0", features = [ "derive", "rc" ] }
+serde_json = { version = "1.0", features = [ "preserve_order" ] }
 serde_yaml = "0.9"
 sha2 = "0.10"
 shell-words = "1.1"
 systemd = "0.10"
 thread_local = "1.1"
-tokio = { version = "1", features = ["full"] }
+tokio = { version = "1", features = [ "full" ] }
 tokio-graceful-shutdown = "0.15"
-tokio-util = { version = "0.7", features = ["io"] }
+tokio-util = { version = "0.7", features = [ "io" ] }
 tracing = "0.1"
+tracing-opentelemetry = "0.28"
 tracing-subscriber = "0.3"
 triggered = "0.1"
-url = { version = "2.5", features = ["serde"] }
+url = { version = "2.5", features = [ "serde" ] }
 uzers = "0.12"
 
-# locked dependencies because of the api changes in the newer versions
-tracing-opentelemetry = "=0.18.0"
-opentelemetry = { version = "=0.18.0", features = ["rt-tokio", "metrics", "trace"] }
-opentelemetry-otlp = { version = "=0.11.0", features = ["metrics"] }
-hyper = { version = "0", features = ["server", "stream"] }
+# local dependencies
 
-[dev-dependencies]
-insta = { version = "1.40", features = ["glob", "redactions", "ron"] }
-map-macro = "0.3"
-rand = { version = "0.8" }
+seele_cgroup = { path = "crates/seele-cgroup" }
+seele_composer = { path = "crates/seele-composer" }
+seele_config = { path = "crates/seele-config" }
+seele_exchange = { path = "crates/seele-exchange" }
+seele_shared = { path = "crates/seele-shared" }
+seele_worker = { path = "crates/seele-worker" }

Dockerfile

@@ -8,14 +8,14 @@ RUN go mod download && go mod verify
 COPY runj/ ./
 RUN make build
 
-FROM rust:1.82-slim-bookworm AS builder
+FROM rust:1.84-slim-bookworm AS builder
 RUN apt update -qq && \
     DEBIAN_FRONTEND=noninteractive apt install -qqy --no-install-recommends pkg-config libdbus-1-dev libsystemd-dev protobuf-compiler libssl-dev patch
 ENV COMMIT_TAG=$GIT_NAME
 ENV COMMIT_SHA=$GIT_SHA
 WORKDIR /usr/src/seele
 COPY . .
-RUN cargo install --path .
+RUN cargo install --path crates/seele
 
 FROM bitnami/minideb:bookworm AS runtime
 WORKDIR /etc/seele

Dockerfile.dev

@@ -5,7 +5,7 @@ RUN go mod download && go mod verify
 COPY runj/ ./
 RUN make build
 
-FROM lukemathwalker/cargo-chef:latest-rust-1.82-slim-bookworm AS chef
+FROM lukemathwalker/cargo-chef:latest-rust-1.84-slim-bookworm AS chef
 RUN apt update -qq && \
     DEBIAN_FRONTEND=noninteractive apt install -qqy --no-install-recommends pkg-config libdbus-1-dev libsystemd-dev protobuf-compiler libssl-dev patch
 WORKDIR /app

crates/seele-cgroup/Cargo.toml

@@ -0,0 +1,15 @@
+[package]
+edition.workspace = true
+name = "seele_cgroup"
+version.workspace = true
+
+[dependencies]
+anyhow = { workspace = true }
+dbus = { workspace = true }
+libcgroups = { workspace = true }
+tracing = { workspace = true }
+
+# local dependencies
+
+seele_config = { workspace = true }
+seele_shared = { workspace = true }

src/cgroup/mod.rs → crates/seele-cgroup/src/lib.rs

@@ -3,18 +3,16 @@ use std::{
     io::{BufRead, BufReader},
     path::PathBuf,
     process,
+    sync::LazyLock,
 };
 
-use anyhow::{bail, Context, Result};
-use libcgroups::common::{get_cgroup_setup, read_cgroup_file, write_cgroup_file_str, CgroupSetup};
-use once_cell::sync::Lazy;
+use anyhow::{Context, Result, bail};
+use libcgroups::common::{CgroupSetup, get_cgroup_setup, read_cgroup_file, write_cgroup_file_str};
+use seele_config as conf;
+use seele_shared as shared;
 use tracing::debug;
 
 pub use self::utils::*;
-use crate::{
-    conf::{self, SeeleWorkMode},
-    shared,
-};
 
 #[rustfmt::skip]
 #[allow(clippy::all)]
@@ -26,17 +24,18 @@ mod utils;
 
 const MANDATORY_CONTROLLERS: &str = "+cpu +cpuset +memory +io +pids";
 
-pub static CGROUP_PATH: Lazy<PathBuf> = Lazy::new(|| match &conf::CONFIG.work_mode {
-    SeeleWorkMode::Bare => {
+pub static CGROUP_PATH: LazyLock<PathBuf> = LazyLock::new(|| match &conf::CONFIG.work_mode {
+    conf::SeeleWorkMode::Bare => {
         systemd::create_and_enter_cgroup().expect("Error entering cgroup scope cgroup")
     }
     _ => utils::check_and_get_self_cgroup().expect("Error getting process' cgroup path"),
 });
 
-pub static CGROUP_MAIN_SCOPE_PATH: Lazy<PathBuf> = Lazy::new(|| CGROUP_PATH.join("main.scope"));
+pub static CGROUP_MAIN_SCOPE_PATH: LazyLock<PathBuf> =
+    LazyLock::new(|| CGROUP_PATH.join("main.scope"));
 
-pub static CGROUP_CONTAINER_SLICE_PATH: Lazy<PathBuf> =
-    Lazy::new(|| CGROUP_PATH.join("container.slice"));
+pub static CGROUP_CONTAINER_SLICE_PATH: LazyLock<PathBuf> =
+    LazyLock::new(|| CGROUP_PATH.join("container.slice"));
 
 #[inline]
 pub fn check_cgroup_setup() -> Result<()> {
@@ -97,16 +96,16 @@ pub fn bind_application_threads() -> Result<()> {
 
     let pids = {
         let content = read_cgroup_file(CGROUP_MAIN_SCOPE_PATH.join("cgroup.threads"))?;
-        let pids = BufReader::new(content.as_bytes())
+        let pids: Vec<u32> = BufReader::new(content.as_bytes())
             .lines()
-            .flatten()
+            .map_while(Result::ok)
             .filter_map(|line| {
                 let line = line.trim();
                 if *shared::TINI_PRESENTS && line == "1" {
-                    return None;
+                    None
+                } else {
+                    Some(line.parse::<u32>().with_context(|| format!("Error parsing line: {line}")))
                 }
-
-                Some(line.parse::<u32>().with_context(|| format!("Error parsing line: {line}")))
             })
             .collect::<Result<Vec<_>>>()?;
 

src/cgroup/systemd.rs → crates/seele-cgroup/src/systemd.rs

@@ -1,6 +1,6 @@
 use std::{path::PathBuf, process, time::Duration};
 
-use anyhow::{bail, Context, Result};
+use anyhow::{Context, Result, bail};
 use dbus::{
     arg::{RefArg, Variant},
     blocking::{Connection, Proxy},

src/cgroup/utils.rs → crates/seele-cgroup/src/utils.rs

@@ -1,7 +1,7 @@
 use std::{fs, path::PathBuf};
 
-use anyhow::{bail, Context, Result};
-use libcgroups::common::{read_cgroup_file, DEFAULT_CGROUP_ROOT};
+use anyhow::{Result, bail};
+use libcgroups::common::{DEFAULT_CGROUP_ROOT, read_cgroup_file};
 
 pub fn check_and_get_self_cgroup() -> Result<PathBuf> {
     let content = fs::read_to_string("/proc/thread-self/cgroup")?;
@@ -31,5 +31,9 @@ pub fn check_and_get_self_cgroup() -> Result<PathBuf> {
 pub fn get_self_cpuset_cpu() -> Result<i64> {
     let path = check_and_get_self_cgroup()?;
     let content = read_cgroup_file(path.join("cpuset.cpus"))?;
-    content.trim().parse().with_context(|| format!("Unexpected cpuset.cpus content: {}", content))
+    tracing::debug!("cpuset.cpus path: {:?}", path.join("cpuset.cpus"));
+    match content.trim().parse() {
+        Ok(cpu) => Ok(cpu),
+        Err(_err) => bail!("Error parsing cpuset.cpus content: {:?}", content),
+    }
 }

crates/seele-composer/Cargo.toml

@@ -0,0 +1,33 @@
+[package]
+edition.workspace = true
+name = "seele_composer"
+version.workspace = true
+
+[dependencies]
+anyhow = { workspace = true }
+async-recursion = { workspace = true }
+chrono = { workspace = true }
+either = { workspace = true }
+ellipse = { workspace = true }
+futures-util = { workspace = true }
+opentelemetry = { workspace = true }
+quick-js = { workspace = true }
+reqwest = { workspace = true }
+ring-channel = { workspace = true }
+serde = { workspace = true }
+serde_json = { workspace = true }
+serde_yaml = { workspace = true }
+tokio = { workspace = true }
+tokio-graceful-shutdown = { workspace = true }
+tracing = { workspace = true }
+
+# local dependencies
+
+seele_config = { workspace = true }
+seele_shared = { workspace = true }
+seele_worker = { workspace = true }
+
+[dev-dependencies]
+insta = { workspace = true }
+map-macro = { workspace = true }
+rand = { workspace = true }

src/composer/execute.rs → crates/seele-composer/src/execute.rs

@@ -1,20 +1,20 @@
 use std::{iter, path::PathBuf, sync::Arc};
 
-use anyhow::{bail, Context, Result};
+use anyhow::{Context, Result, bail};
 use async_recursion::async_recursion;
 use either::Either;
 use futures_util::future;
 use ring_channel::RingSender;
 use tokio::{
-    sync::{oneshot, Mutex},
+    sync::{Mutex, oneshot},
     time::Instant,
 };
-use tracing::{debug, instrument, Span};
+use tracing::{Span, debug, instrument};
 
 use super::predicate;
 use crate::{
-    composer::report::apply_embeds_config,
-    entities::{
+    report::apply_embeds_config,
+    shared::entities::{
         ActionTaskConfig, ParallelFailedReport, ParallelSuccessReport, ParallelTaskConfig,
         SequenceFailedReport, SequenceSuccessReport, Submission, SubmissionReportUploadConfig,
         TaskConfig, TaskConfigExt, TaskEmbeds, TaskFailedReport, TaskNode, TaskNodeExt,
@@ -293,9 +293,9 @@ mod tests {
     use tokio::{runtime::Builder, sync::mpsc};
 
     use crate::{
-        composer::resolve::resolve_submission,
         entities::{ActionReport, ActionSuccessReport, ActionSuccessReportExt},
-        worker::{action, WorkerQueueItem},
+        resolve::resolve_submission,
+        worker::WorkerQueueItem,
     };
 
     #[test]
@@ -325,9 +325,9 @@ mod tests {
                         .send(Ok(ActionReport::Success(ActionSuccessReport {
                             run_at: Utc::now(),
                             time_elapsed_ms: 0,
-                            ext: ActionSuccessReportExt::Noop(action::noop::ExecutionReport {
-                                test: 0,
-                            }),
+                            ext: ActionSuccessReportExt::Noop(
+                                crate::shared::entities::noop::ExecutionReport { test: 0 },
+                            ),
                         })))
                         .unwrap();
                 }

src/composer/mod.rs → crates/seele-composer/src/lib.rs

@@ -1,24 +1,28 @@
 use std::{num::NonZeroUsize, sync::Arc};
 
-use anyhow::{bail, Context, Result};
+use anyhow::{Context, Result, bail};
 use chrono::Utc;
 use ellipse::Ellipse;
 use futures_util::StreamExt;
-use opentelemetry::{Context as OpenTelemetryCtx, KeyValue};
+use opentelemetry::KeyValue;
 use ring_channel::{RingReceiver, RingSender};
+use seele_config as conf;
+use seele_shared as shared;
+use seele_shared::entities;
+use seele_worker as worker;
 use tokio::{
     fs,
     sync::mpsc::{self, error::TryRecvError},
     time::Instant,
 };
 use tokio_graceful_shutdown::{FutureExt, SubsystemHandle};
-use tracing::{debug, error, field, instrument, Span};
+use tracing::{Span, debug, error, field, instrument};
 
 pub use self::signal::*;
 use crate::{
-    composer::{report::apply_uploads_config, reporter::execute_reporter},
-    conf,
     entities::{Submission, SubmissionConfig},
+    report::apply_uploads_config,
+    reporter::execute_reporter,
     shared::metrics,
     worker::WorkerQueueTx,
 };
@@ -88,11 +92,9 @@ async fn handle_submission(
         let end = Instant::now();
         end.duration_since(begin).as_secs_f64()
     };
-    metrics::SUBMISSION_HANDLING_HISTOGRAM.record(
-        &OpenTelemetryCtx::current(),
-        duration,
-        &[KeyValue::new(SUBMISSION_STATUS, signal_type)],
-    );
+
+    metrics::SUBMISSION_HANDLING_HISTOGRAM
+        .record(duration, &[KeyValue::new(SUBMISSION_STATUS, signal_type)]);
 }
 
 async fn do_handle_submission(