feat(scheduler): support resource requirement in priority plugin

ci/flame-cluster-benchmark.yaml

@@ -2,10 +2,9 @@
 cluster:
   name: flame
   endpoint: "https://flame-session-manager:8080"
-  slot: "cpu=1,mem=1g"
+  resreq: "cpu=1,mem=1g"
   policies:
     - priority
-    - fairshare
     - gang
   storage: none
   schedule_interval: 100

ci/flame-cluster-docker.yaml

@@ -2,10 +2,9 @@
 cluster:
   name: flame
   endpoint: "https://flame-session-manager:8080"
-  slot: "cpu=1,mem=1g"
+  resreq: "cpu=1,mem=1g"
   policies:
     - priority
-    - fairshare
     - gang
   storage: fs://data/
   schedule_interval: 500           # Scheduler loop interval in milliseconds (default: 500)

ci/flame-cluster-local.yaml

@@ -2,10 +2,9 @@
 cluster:
   name: flame
   endpoint: "http://127.0.0.1:30080"
-  slot: "cpu=1,mem=1g"
+  resreq: "cpu=1,mem=1g"
   policies:
     - priority
-    - fairshare
     - gang
   storage: fs://data/
   schedule_interval: 100           # Scheduler loop interval in milliseconds (default: 500)

ci/flame-cluster.yaml

@@ -2,7 +2,7 @@
 cluster:
   name: flame
   endpoint: "https://flame-session-manager:8080"
-  slot: "cpu=1,mem=1g"
+  resreq: "cpu=1,mem=2g"
   policies:
     - priority
     - gang

common/src/apis/from_rpc.rs

@@ -149,7 +149,6 @@ impl TryFrom<(rpc::Application, rpc::Session)> for SessionContext {
         Ok(SessionContext {
             session_id: metadata.id,
             application,
-            slots: spec.slots,
             common_data: spec.common_data.map(CommonData::from),
         })
     }

common/src/apis/session.rs

@@ -94,10 +94,10 @@ impl Session {
                 self.id, self.application, attr.application
             )));
         }
-        if self.slots != attr.slots {
+        if self.resreq != attr.resreq {
             return Err(FlameError::InvalidConfig(format!(
-                "session <{}> spec mismatch: slots differs (expected {}, got {})",
-                self.id, self.slots, attr.slots
+                "session <{}> spec mismatch: resreq differs (expected {:?}, got {:?})",
+                self.id, self.resreq, attr.resreq
             )));
         }
         if self.min_instances != attr.min_instances {

common/src/apis/to_rpc.rs

@@ -196,13 +196,12 @@ impl From<&Session> for rpc::Session {
             }),
             spec: Some(rpc::SessionSpec {
                 application: ssn.application.clone(),
-                slots: ssn.slots,
                 common_data: ssn.common_data.clone().map(CommonData::into),
                 min_instances: ssn.min_instances,
                 max_instances: ssn.max_instances,
                 batch_size: ssn.batch_size,
                 priority: ssn.priority,
-                resreq: None,
+                resreq: ssn.resreq.clone().map(rpc::ResourceRequirement::from),
             }),
             status: Some(status),
         }