Refactor judging components

Cargo.toml

@@ -36,6 +36,6 @@ opt-level = 0
 [workspace]
 # TODO: add other as they are fixed
 members=[ "src/devtool", "src/gen-api-client", "src/cli", "src/client",
-    "src/problem-loader", "src/invoker", "src/dist-files-generator",
-     "src/dist-builder", "src/svaluer", "src/invoker-api", "src/pps/api",
+    "src/problem-loader", "src/judge", "src/dist-files-generator",
+     "src/dist-builder", "src/svaluer", "src/judging-apis", "src/invoker", "src/pps/api",
      "src/pps/cli", "src/pps/server" ]

k8s/jjs/templates/invoker.yaml → k8s/jjs/templates/judge.yaml

@@ -1,7 +1,7 @@
 apiVersion: rbac.authorization.k8s.io/v1
 kind: Role
 metadata:
-  name: invoker
+  name: judge
 rules:
   - apiGroups: [""]
     resources: ["configmaps"]
@@ -10,27 +10,27 @@ rules:
 apiVersion: v1
 kind: ServiceAccount
 metadata:
-  name: invoker
+  name: judge
 ---
 apiVersion: rbac.authorization.k8s.io/v1
 kind: RoleBinding
 metadata:
-  name: invoker
+  name: judge
 subjects:
   - kind: ServiceAccount
-    name: invoker
+    name: judge
     apiGroup: ""
 roleRef:
   kind: Role
-  name: invoker
+  name: judge
   apiGroup: rbac.authorization.k8s.io
 ---
 apiVersion: apps/v1
 kind: Deployment
 metadata:
-  name: invoker
+  name: judge
   labels:
-    app: invoker
+    app: judge
 # {{- if .Values.dev.kubeScore  }}
   annotations:
     kube-score/ignore: container-security-context
@@ -39,21 +39,21 @@ spec:
   replicas: 1
   selector:
     matchLabels:
-      app: invoker
+      app: judge
   template:
     metadata:
       labels:
-        app: invoker
+        app: judge
     spec:
-      serviceAccountName: invoker
+      serviceAccountName: judge
       containers:
-        - name: invoker
+        - name: judge
           env: 
             - name: RUST_LOG
-              value: info,invoker=trace,problem_loader=trace,puller=trace
+              value: info,judger=trace,problem_loader=trace,puller=trace
             - name: JJS_AUTH_DATA_INLINE
               value: '{"endpoint": "http://apiserver:1779/", "auth": {"byToken": {"token": "Dev::root"}}}'
-          image: "{{ .Values.image.repositoryPrefix }}invoker:{{ .Values.image.tag }}"
+          image: "{{ .Values.image.repositoryPrefix }}judge:{{ .Values.image.tag }}"
           imagePullPolicy: {{ .Values.image.pullPolicy }}
           securityContext: 
             privileged: true
@@ -69,7 +69,7 @@ spec:
 apiVersion: v1
 kind: Service
 metadata:
-  name: invoker
+  name: judge
 spec:
   type: ClusterIP
   ports:
@@ -78,4 +78,4 @@ spec:
       protocol: TCP
       name: http
   selector:
-    app: invoker
+    app: judge

k8s/jjs/templates/network-policy.yaml

@@ -16,11 +16,11 @@ spec:
 apiVersion: networking.k8s.io/v1
 kind: NetworkPolicy
 metadata:
-  name: builtin-invoker
+  name: builtin-judge
 spec:
   podSelector:
     matchLabels:
-      app: invoker
+      app: judge
   policyTypes:
   - Ingress
   ingress: []

src/dist-builder/src/main.rs

@@ -193,7 +193,7 @@ fn make_rust_package_list() -> Vec<RustPackage> {
     add("pps-cli", "jjs-pps", Section::Tool);
     //add("userlist", "jjs-userlist", Section::Tool);
     add("cli", "jjs-cli", Section::Tool);
-    add("invoker", "jjs-invoker", Section::Daemon);
+    add("judge", "jjs-judge", Section::Daemon);
     add("svaluer", "jjs-svaluer", Section::Tool);
     /*add(
         "configure-toolchains",

src/invoker/Cargo.toml

@@ -1,49 +1,18 @@
-
 [package]
 name = "invoker"
 version = "0.1.0"
 authors = ["Mikail Bagishov <bagishov.mikail@yandex.ru>"]
 edition = "2018"
 
 [dependencies]
+judging-apis = { path = "../judging-apis" }
+rpc = { git = "https://github.com/jjs-dev/commons" }
 minion = {git = "https://github.com/jjs-dev/minion"}
-serde = { version = "1.0.117", features = ["derive"] }
-serde_json = "1.0.59"
-dotenv = "0.15.0"
-aho-corasick = "0.7.14"
-invoker-api = {path = "../invoker-api"}
-pom = {path = "../pom"}
-libc = "0.2.80"
+hyper = "0.13.8"
+tokio = { version = "0.2.22", features = ["macros", "io-std"] }
+anyhow = "1.0.34"
+futures-util = "0.3.7"
+tracing = "0.1.21"
 nix = "0.19.0"
-strum = "0.19.5"
-strum_macros = "0.19.4"
-chrono = "0.4.19"
-tempfile = "3.1.0"
-fs_extra = "1.2.0"
-base64 = "0.13.0"
-bitflags = "1.2.1"
-util = {path = "../util"}
-anyhow = "1.0.33"
-thiserror = "1.0.21"
-uuid = { version = "0.8.1", features = ["v5"] }
-problem-loader = {path = "../problem-loader"}
-tokio = { version = "0.2.22", features = ["rt-core", "process", "io-std", "macros", "fs", "sync"] }
-async-trait = "0.1.41"
-num_cpus = "1.13.0"
-serde_yaml = "0.8.14"
-openssl = "0.10.30"
-actix-web = { version = "3.2.0", features = ["openssl"], default-features = false }
-actix-rt = { version = "1.1.1", default-features = false }
 once_cell = "1.4.1"
-client = {path = "../client"}
-kube = { version = "0.43.0", optional = true }
-k8s-openapi = { version = "0.9.0", optional = true, features = ["v1_17"], default-features = false }
-puller = { git = "https://github.com/jjs-dev/commons" }
-tracing = "0.1.21"
-tracing-futures = "0.2.4"
-async-mpmc = { git = "https://github.com/jjs-dev/commons" }
-multiwake = { git = "https://github.com/jjs-dev/commons" }
-dkregistry = { git = "https://github.com/mikailbag/dkregistry-rs", branch = "all" }
-
-[features]
-k8s = ["kube", "k8s-openapi"]
+serde = { version = "1.0.117", features = ["derive"] }

src/invoker/src/config.rs

@@ -1,16 +1,10 @@
 use serde::{Deserialize, Serialize};
+use std::path::PathBuf;
 
 #[derive(Deserialize, Serialize, Debug)]
 #[serde(deny_unknown_fields)]
 #[serde(rename_all = "kebab-case")]
-pub struct InvokerConfig {
-    /// How many workers should be spawned
-    /// By default equal to processor count
-    #[serde(default)]
-    pub workers: Option<usize>,
-    /// API service config
-    #[serde(default)]
-    pub api: ApiSvcConfig,
+pub struct Config {
     /// If enabled, invoker will directly mount host filesystem instead of
     /// toolchain image.
     #[serde(default)]
@@ -23,36 +17,6 @@ pub struct InvokerConfig {
     /// As usual, all mounts will be no-suid and read-only.
     #[serde(default)]
     pub expose_host_dirs: Option<Vec<String>>,
-    /// Configures how invoker should resolve problems
-    pub problems: problem_loader::LoaderConfig,
-}
-
-#[derive(Deserialize, Serialize, Debug)]
-#[serde(deny_unknown_fields)]
-pub struct ApiSvcConfig {
-    /// Override bind IP
-    #[serde(default = "ApiSvcConfig::default_address")]
-    pub address: String,
-    /// Override bind port
-    #[serde(default = "ApiSvcConfig::default_port")]
-    pub port: u16,
-}
-
-impl ApiSvcConfig {
-    fn default_address() -> String {
-        "0.0.0.0".to_string()
-    }
-
-    fn default_port() -> u16 {
-        1789
-    }
-}
-
-impl Default for ApiSvcConfig {
-    fn default() -> Self {
-        ApiSvcConfig {
-            address: ApiSvcConfig::default_address(),
-            port: ApiSvcConfig::default_port(),
-        }
-    }
+    /// Directory which will contain temporary invocation data.
+    pub work_root: PathBuf,
 }

src/invoker/src/graph_exec.rs

@@ -0,0 +1,12 @@
+//! Interprets given request graph
+use judging_apis::invoke::InvokeRequest;
+
+pub struct Interpreter<'a> {
+    req: &'a InvokeRequest,
+}
+
+impl<'a> Interpreter<'a> {
+    pub fn new(req: &'a InvokeRequest) -> Self {
+        Interpreter { req }
+    }
+}

src/invoker/src/worker/invoke_util.rs → src/invoker/src/invoke_util.rs

@@ -1,10 +1,10 @@
-use crate::worker::{Command, LoweredJudgeRequest};
 use anyhow::Context;
+use judging_apis::invoke::{Command, EnvVarValue, InvokeRequest};
 use std::{
-    fs,
     path::{Path, PathBuf},
     time::Duration,
 };
+use tokio::fs;
 use tracing::{debug, error};
 
 pub(crate) struct Sandbox {
@@ -35,11 +35,12 @@ static DEFAULT_HOST_MOUNTS: once_cell::sync::Lazy<Vec<String>> = once_cell::sync
     ]
 });
 
-pub(crate) fn create_sandbox(
-    req: &LoweredJudgeRequest,
-    test_id: Option<u32>,
+pub(crate) async fn create_sandbox(
+    config: &crate::config::Config,
+    req: &InvokeRequest,
+    req_id: &str,
     backend: &dyn minion::erased::Backend,
-    config: &crate::config::InvokerConfig,
+    settings: judging_apis::invoke::Sandbox,
 ) -> anyhow::Result<Sandbox> {
     let mut shared_dirs = vec![];
     if config.host_toolchains {
@@ -58,10 +59,10 @@ pub(crate) fn create_sandbox(
         }
     } else {
         let toolchain_dir = &req.toolchain_dir;
-        let opt_items =
-            fs::read_dir(&toolchain_dir).context("failed to list toolchains sysroot")?;
-        for item in opt_items {
-            let item = item.context("failed to stat toolchains sysroot item")?;
+        let mut opt_items = fs::read_dir(&toolchain_dir)
+            .await
+            .context("failed to list toolchains sysroot")?;
+        while let Some(item) = opt_items.next_entry().await? {
             let name = item.file_name();
             let shared_dir = minion::SharedDir {
                 src: toolchain_dir.join(&name),
@@ -72,40 +73,39 @@ pub(crate) fn create_sandbox(
         }
     }
 
-    let limits = if let Some(test_id) = test_id {
-        req.problem.tests[(test_id - 1) as usize].limits
-    } else {
-        req.compile_limits
-    };
-    let out_dir = req.step_dir(test_id);
-    std::fs::create_dir_all(&out_dir).context("failed to create step directory")?;
+    let work_dir = config.work_root.join(req_id);
+    tokio::fs::create_dir_all(&work_dir)
+        .await
+        .context("failed to create working directory")?;
     let umount_path;
     #[cfg(target_os = "linux")]
     {
-        let quota = limits.work_dir_size();
+        let quota = settings.limits.work_dir_size();
         let quota = minion::linux::ext::Quota::bytes(quota);
-        minion::linux::ext::make_tmpfs(&out_dir.join("data"), quota)
+        minion::linux::ext::make_tmpfs(&work_dir.join("data"), quota)
             .context("failed to set size limit on shared directory")?;
-        umount_path = Some(out_dir.join("data"));
+        umount_path = Some(work_dir.join("data"));
     }
     #[cfg(not(target_os = "linux"))]
     {
         umount_path = None;
     }
     shared_dirs.push(minion::SharedDir {
-        src: out_dir.join("data"),
+        src: work_dir.join("data"),
         dest: PathBuf::from("/jjs"),
         kind: minion::SharedDirKind::Full,
     });
-    let cpu_time_limit = Duration::from_millis(limits.time() as u64);
-    let real_time_limit = Duration::from_millis(limits.time() * 3 as u64);
-    std::fs::create_dir(out_dir.join("root")).context("failed to create chroot dir")?;
+    let cpu_time_limit = Duration::from_millis(settings.limits.time() as u64);
+    let real_time_limit = Duration::from_millis(settings.limits.time() * 3 as u64);
+    tokio::fs::create_dir(work_dir.join("root"))
+        .await
+        .context("failed to create chroot dir")?;
     // TODO adjust integer types
     let sandbox_options = minion::SandboxOptions {
-        max_alive_process_count: limits.process_count() as _,
-        memory_limit: limits.memory() as _,
+        max_alive_process_count: settings.limits.process_count() as _,
+        memory_limit: settings.limits.memory() as _,
         exposed_paths: shared_dirs,
-        isolation_root: out_dir.join("root"),
+        isolation_root: work_dir.join("root"),
         cpu_time_limit,
         real_time_limit,
     };
@@ -125,17 +125,35 @@ pub(crate) fn log_execute_command(command_interp: &Command) {
 pub(crate) fn command_set_from_judge_req(cmd: &mut minion::Command, command: &Command) {
     cmd.path(&command.argv[0]);
     cmd.args(&command.argv[1..]);
-    cmd.envs(&command.env);
+    cmd.envs(
+        command
+            .env
+            .iter()
+            .map(|(name, value)| -> std::ffi::OsString {
+                match value {
+                    EnvVarValue::Plain(p) => format!("{}={}", name, p).into(),
+                    EnvVarValue::File(_) => unreachable!(),
+                }
+            }),
+    );
 }
 
-pub(crate) fn command_set_stdio(cmd: &mut minion::Command, stdout_path: &Path, stderr_path: &Path) {
-    let stdout_file = fs::File::create(stdout_path).expect("io error");
+pub(crate) async fn command_set_stdio(
+    cmd: &mut minion::Command,
+    stdout_path: &Path,
+    stderr_path: &Path,
+) {
+    let stdout_file = fs::File::create(stdout_path).await.expect("io error");
 
-    let stderr_file = fs::File::create(stderr_path).expect("io error");
+    let stderr_file = fs::File::create(stderr_path).await.expect("io error");
     // Safety: std::fs::File owns it's handle
     unsafe {
-        cmd.stdout(minion::OutputSpecification::handle_of(stdout_file));
+        cmd.stdout(minion::OutputSpecification::handle_of(
+            stdout_file.into_std().await,
+        ));
 
-        cmd.stderr(minion::OutputSpecification::handle_of(stderr_file));
+        cmd.stderr(minion::OutputSpecification::handle_of(
+            stderr_file.into_std().await,
+        ));
     }
 }