Merge pull request #152 from OJ-lab/slhmy/refactor

Refactor judger code

.github/workflows/docker-publish-server.yml → .github/workflows/docker-publish-judger.yml

@@ -17,7 +17,7 @@ env:
   # Use docker.io for Docker Hub if empty
   REGISTRY: ghcr.io
   # github.repository as <account>/<repo>
-  IMAGE_NAME: ${{ github.repository }}/judger-server
+  IMAGE_NAME: ${{ github.repository }}/judger
 
 
 jobs:
@@ -71,7 +71,7 @@ jobs:
         id: build-and-push
         uses: docker/build-push-action@v5
         with:
-          file: docker/judger-server.dockerfile
+          file: docker/judger.dockerfile
           context: .
           push: ${{ github.event_name != 'pull_request' }}
           tags: ${{ steps.meta.outputs.tags }}

data/.gitignore

@@ -1,2 +1 @@
-rclone-minio.conf
-rclone-problem-package/
+problem-package/

data/rclone-minio.conf.example → data/default-rclone.conf

@@ -2,8 +2,8 @@
 type = s3
 provider = Minio
 env_auth = false
-access_key_id = YOUR_ACCESS_KEY
-secret_access_key = YOUR_SECRET_KEY
+access_key_id = minio-root-user
+secret_access_key = minio-root-password
 endpoint = http://127.0.0.1:9000
 location_constraint =
 acl = private

docker/judger-server.dockerfile → docker/judger.dockerfile

@@ -5,20 +5,20 @@ COPY judger /usr/src/judger
 WORKDIR /usr/src/judger
 
 RUN apt update && apt install -y libseccomp-dev gcc
-RUN cargo build --bin judger-server --release
+RUN cargo build --bin judger --release
 
 
 FROM ubuntu:latest
 
 RUN apt update && apt install -y libseccomp-dev gcc g++ curl unzip
-COPY --from=build /usr/src/judger/target/release/judger-server /usr/local/bin/judger-server
+COPY --from=build /usr/src/judger/target/release/judger /usr/local/bin/judger
 
 RUN curl https://rclone.org/install.sh | bash
 
 RUN mkdir /workspace
 WORKDIR /workspace
-COPY data/dev-problem-package /workspace/data/dev-problem-package
-RUN mkdir /workspace/data/rclone-problem-package
+COPY data/default-rclone.conf /workspace/data/default-rclone.conf
+RUN mkdir /workspace/data/problem-package
 
 ENV RUST_LOG=DEBUG
 EXPOSE 8000

judger/Cargo.toml

@@ -43,12 +43,4 @@ chrono = { version = "0.4", features = ["serde"] }
 anyhow = "1"
 thiserror = "1"
 
-uuid = { version = "1.4", features = ["serde", "v4"] }
-
-[[bin]]
-name ="judger-cli"
-path ="src/cli/main.rs"
-
-[[bin]]
-name ="judger-server"
-path ="src/server/main.rs"
+uuid = { version = "1.4", features = ["serde", "v4"] }

judger/src/server/client/httpclient.rs → judger/src/agent/http.rs

@@ -8,9 +8,7 @@ impl HttpClient {
         let client = reqwest::Client::new();
         Self { client, base_url }
     }
-    pub fn _get(&self, path: String) -> reqwest::RequestBuilder {
-        self.client.get(format!("{}{}", self.base_url, path))
-    }
+
     pub fn post(&self, path: String) -> reqwest::RequestBuilder {
         self.client.post(format!("{}{}", self.base_url, path))
     }

judger/src/agent/mod.rs

@@ -0,0 +1,3 @@
+pub mod http;
+pub mod platform;
+pub mod rclone;

judger/src/agent/platform/mod.rs

@@ -0,0 +1,99 @@
+use super::http::HttpClient;
+use judge_core::{compiler::Language, judge::result::JudgeResultInfo};
+
+pub struct PlatformClient {
+    client: HttpClient,
+}
+
+impl PlatformClient {
+    pub fn new(base_url: String) -> Self {
+        Self {
+            client: HttpClient::new(base_url),
+        }
+    }
+
+    pub async fn pick_task(&self) -> Result<JudgeTask, anyhow::Error> {
+        pick_task(&self.client).await
+    }
+
+    pub async fn report_task(
+        &self,
+        stream_id: &str,
+        results: Vec<JudgeResultInfo>,
+    ) -> Result<(), anyhow::Error> {
+        report_task(&self.client, stream_id, results).await
+    }
+}
+
+#[derive(Deserialize, Debug)]
+pub struct JudgeTask {
+    #[serde(rename = "submissionUID")]
+    pub submission_uid: String,
+    #[serde(rename = "problemSlug")]
+    pub problem_slug: String,
+    pub code: String,
+    pub language: Language,
+    #[serde(rename = "redisStreamID")]
+    pub redis_stream_id: String,
+}
+#[derive(Serialize)]
+struct PickTaskBody {
+    consumer: String,
+}
+#[derive(Deserialize, Debug)]
+struct PickTaskResponse {
+    task: JudgeTask,
+}
+
+async fn pick_task(client: &HttpClient) -> Result<JudgeTask, anyhow::Error> {
+    let pick_url = "/api/v1/judge/task/pick";
+    let body = PickTaskBody {
+        consumer: "".to_string(),
+    };
+    let response = client.post(pick_url.to_string()).json(&body).send().await?;
+
+    match response.status() {
+        reqwest::StatusCode::OK => Ok(response.json::<PickTaskResponse>().await?.task),
+        _ => Err(anyhow::anyhow!("Queue is empty")),
+    }
+}
+
+#[derive(Serialize)]
+struct ReportTaskBody {
+    consumer: String,
+    stream_id: String,
+    verdict_json: String,
+}
+#[derive(Deserialize, Debug)]
+struct ReportTaskResponse {
+    message: String,
+}
+
+async fn report_task(
+    client: &HttpClient,
+    stream_id: &str,
+    results: Vec<JudgeResultInfo>,
+) -> Result<(), anyhow::Error> {
+    let report_url = "/api/v1/judge/task/report";
+    let body = ReportTaskBody {
+        consumer: "".to_string(),
+        stream_id: stream_id.to_owned(),
+        verdict_json: serde_json::to_string(&results).unwrap(),
+    };
+    let response = client
+        .post(report_url.to_string())
+        .json(&body)
+        .send()
+        .await?;
+
+    match response.status() {
+        reqwest::StatusCode::OK => {
+            log::debug!(
+                "Report message: {:?}",
+                response.json::<ReportTaskResponse>().await?.message
+            );
+            Ok(())
+        }
+        _ => Err(anyhow::anyhow!("Report Failed")),
+    }
+}

judger/src/agent/rclone.rs

@@ -0,0 +1,41 @@
+use anyhow::{self, Error};
+use std::path::{Path, PathBuf};
+use std::process::Command;
+
+pub struct RcloneClient {
+    config_path: PathBuf,
+}
+
+impl RcloneClient {
+    pub fn new(config_path: PathBuf) -> Self {
+        Self { config_path }
+    }
+
+    pub fn is_avaliable(&self) -> bool {
+        let status = Command::new("rclone")
+            .arg("--config")
+            .arg(format!("{}", self.config_path.to_string_lossy()))
+            .arg("ls")
+            .arg("minio:")
+            .status()
+            .expect("Failed to rclone");
+
+        status.success()
+    }
+
+    pub fn sync_bucket(&self, bucket_name: &str, target_dir: &Path) -> Result<(), Error> {
+        let status = Command::new("rclone")
+            .arg("--config")
+            .arg(format!("{}", self.config_path.to_string_lossy()))
+            .arg("sync")
+            .arg(format!("minio:{}", bucket_name))
+            .arg(format!("{}", target_dir.to_string_lossy()))
+            .status()
+            .expect("Failed to rclone");
+        if status.success() {
+            Ok(())
+        } else {
+            Err(anyhow::anyhow!("rclone sync failed, please check config."))
+        }
+    }
+}

judger/src/server/environment/mod.rs → judger/src/env/mod.rs

@@ -3,23 +3,30 @@ use std::path::PathBuf;
 use structopt::StructOpt;
 
 #[derive(StructOpt, Debug, Clone)]
-#[structopt(name = "judge-server")]
+#[structopt(name = "judger")]
 pub struct JudgeServerOpt {
+    /// For loading Opt from .env file
     #[structopt(long)]
     pub env_path: Option<String>,
 
     /// Port to listen to
     #[structopt(env = "PORT", default_value = "8080")]
     pub port: u16,
 
-    #[structopt(long, default_value = "data/dev-problem-package")]
+    // TODO: make rclone optional
+    #[structopt(long, default_value = "data/default-rclone.conf")]
+    pub rclone_config: PathBuf,
+    #[structopt(long, default_value = "oj-lab-problem-package")]
+    pub problem_package_bucket: String,
+    /// Where to store problem package
+    #[structopt(long, default_value = "data/problem-package")]
     pub problem_package_dir: PathBuf,
 
-    #[structopt(env = "BASE_URL", default_value = "http://localhost:8080/api/v1/judge")]
-    pub base_url: String,
-
-    #[structopt(env = "INTERVAL", default_value = "10")]
-    pub interval: i32,
+    #[structopt(env = "PLATFORM_URI", default_value = "http://localhost:8080/")]
+    pub platform_uri: String,
+    /// Interval to fetch task in seconds
+    #[structopt(env = "FETCH_TASK_INTERVAL", default_value = "10")]
+    pub fetch_task_interval: u64,
 }
 
 pub fn load_option() -> JudgeServerOpt {

judger/src/server/error.rs → judger/src/error.rs

@@ -2,7 +2,6 @@
 
 use actix_web::{HttpResponse, ResponseError};
 use judge_core::error::JudgeCoreError;
-use judger::service::error::JudgeServiceError;
 
 #[derive(Debug, thiserror::Error)]
 pub enum ServiceError {
@@ -23,7 +22,6 @@ pub enum ServiceError {
 #[derive(Debug)]
 pub enum ClientError {
     InternalError(anyhow::Error),
-    PackageError(JudgeServiceError),
 }
 
 #[derive(Serialize)]