add labels affinity feature

Author: larrylian

reps/2022-08-31-actor-affinity-apis.md

@@ -36,8 +36,8 @@ Yes, this will be a complement to ray core's ability to flexibly schedule actors
 
 Actor affinity/anti-affinity schedule API Design
 1. Scheduling Strategy adds an ActorAffinitySchedulingStrategy. 
-2. This strategy consists of several ActorAffinityMatchExpressions.
-3. ActorAffinityMatchExpression has 4 match types which are IN/NOT_IN/EXISTS/DOES_NOT_EXIST
+2. This strategy consists of several LabelMatchExpressions.
+3. LabelMatchExpression has 4 match types which are IN/NOT_IN/EXISTS/DOES_NOT_EXIST
 
 Use Case | ActorAffinityOperator
 -- | --
@@ -72,10 +72,10 @@ SchedulingStrategyT = Union[None, str,
                             ActorAffinitySchedulingStrategy]
 
 class ActorAffinitySchedulingStrategy:
-    def __init__(self, match_expressions: List[ActorAffinityMatchExpression]):
+    def __init__(self, match_expressions: List[LabelMatchExpression]):
         self.match_expressions = match_expressions
 
-class ActorAffinityMatchExpression:
+class LabelMatchExpression:
     """An expression used to represent an actor's affinity.
     Attributes:
         key: the key of label
@@ -84,7 +84,7 @@ class ActorAffinityMatchExpression:
         values: a list of label value
         soft: ...
     """
-    def __init__(self, key: str, operator: ActorAffinityOperator,
+    def __init__(self, key: str, operator: LabelMatchOperator,
                  values: List[str], soft: bool):
         self.key = key
         self.operator = operator
@@ -106,7 +106,7 @@ Step 2: Set actor affinity strategy.
 1. The target actor is expected to be scheduled with the actors whose label key is "location" and value in ["dc-1"].
 ```python
 match_expressions = [
-    ActorAffinityMatchExpression("location", ActorAffinityOperator.IN, ["dc_1"], False)
+    LabelMatchExpression("location", LabelMatchOperator.IN, ["dc_1"], False)
 ]
 actor_affinity_strategy = ActorAffinitySchedulingStrategy(match_expressions)
 actor = Actor.options(scheduling_strategy = actor_affinity_strategy).remote()
@@ -116,7 +116,7 @@ actor = Actor.options(scheduling_strategy = actor_affinity_strategy).remote()
 with the actors whose label key is "location" and value in ["dc-1"].
 ```python
 match_expressions = [
- ActorAffinityMatchExpression("location", ActorAffinityOperator.NOT_IN, ["dc_1"], False)
+ LabelMatchExpression("location", LabelMatchOperator.NOT_IN, ["dc_1"], False)
 ]
 actor_affinity_strategy = ActorAffinitySchedulingStrategy(match_expressions)
 actor = Actor.options(scheduling_strategy = actor_affinity_strategy).remote()
@@ -125,7 +125,7 @@ actor = Actor.options(scheduling_strategy = actor_affinity_strategy).remote()
 3. The target actor is expected to be scheduled with the actors whose label key exists "location".
 ```python
 match_expressions = [
- ActorAffinityMatchExpression("location", ActorAffinityOperator.EXISTS, [], False)
+ LabelMatchExpression("location", LabelMatchOperator.EXISTS, [], False)
 ]
 actor_affinity_strategy = ActorAffinitySchedulingStrategy(match_expressions)
 actor = Actor.options(scheduling_strategy = actor_affinity_strategy).remote()
@@ -134,7 +134,7 @@ actor = Actor.options(scheduling_strategy = actor_affinity_strategy).remote()
 4. The target actor is not expected to be scheduled with the actors whose label key exists "location".
 ```python
 match_expressions = [
- ActorAffinityMatchExpression("location", ActorAffinityOperator.DOES_NOT_EXIST, [], False)
+ LabelMatchExpression("location", ActorAffinityOperator.DOES_NOT_EXIST, [], False)
 ]
 actor_affinity_strategy = ActorAffinitySchedulingStrategy(match_expressions)
 actor = Actor.options(scheduling_strategy = actor_affinity_strategy).remote()
@@ -143,8 +143,8 @@ actor = Actor.options(scheduling_strategy = actor_affinity_strategy).remote()
 5. You can also set multiple expressions at the same time, and multiple expressions need to be satisfied when scheduling.
 ```python
 match_expressions = [
- ActorAffinityMatchExpression("location", ActorAffinityOperator.DOES_NOT_EXIST, [], False),
- ActorAffinityMatchExpression("version", ActorAffinityOperator.EXISTS, [], False)
+ LabelMatchExpression("location", LabelMatchOperator.DOES_NOT_EXIST, [], False),
+ LabelMatchExpression("version", LabelMatchOperator.EXISTS, [], False)
 ]
 actor_affinity_strategy = ActorAffinitySchedulingStrategy(match_expressions)
 actor = Actor.options(scheduling_strategy = actor_affinity_strategy).remote()
@@ -181,70 +181,70 @@ Set the labels for this actor API
 Actor affinity scheduling strategy API
 ```java
 public class ActorAffinitySchedulingStrategy implements SchedulingStrategy {
-  private ActorAffinitySchedulingStrategy(List<ActorAffinityMatchExpression> expressions) {
+  private ActorAffinitySchedulingStrategy(List<LabelMatchExpression> expressions) {
 }
 
-public class ActorAffinityMatchExpression {
+public class LabelMatchExpression {
   private String key;
-  private ActorAffinityOperator operator;
+  private LabelMatchOperator operator;
   private List<String> values;
   private boolean isSoft;
 
   /**
    * Returns an affinity expression to indicate that the target actor is expected to be scheduled
    * with the actors whose label meets one of the composed key and values. eg:
-   * ActorAffinityMatchExpression.in("location", new ArrayList<>() {{ add("dc-1");}}, false).
+   * LabelMatchExpression.in("location", new ArrayList<>() {{ add("dc-1");}}, false).
    *
    * @param key The key of label.
    * @param values A list of label values.
    * @param isSoft If true, the actor will be scheduled even there's no matched actor.
-   * @return ActorAffinityMatchExpression.
+   * @return LabelMatchExpression.
    */
-  public static ActorAffinityMatchExpression in(String key, List<String> values, boolean isSoft) {
-    return new ActorAffinityMatchExpression(key, ActorAffinityOperator.IN, values, isSoft);
+  public static LabelMatchExpression in(String key, List<String> values, boolean isSoft) {
+    return new LabelMatchExpression(key, LabelMatchOperator.IN, values, isSoft);
   }
 
   /**
    * Returns an affinity expression to indicate that the target actor is not expected to be
    * scheduled with the actors whose label meets one of the composed key and values. eg:
-   * ActorAffinityMatchExpression.notIn( "location", new ArrayList<>() {{ add("dc-1");}}, false).
+   * LabelMatchExpression.notIn( "location", new ArrayList<>() {{ add("dc-1");}}, false).
    *
    * @param key The key of label.
    * @param values A list of label values.
    * @param isSoft If true, the actor will be scheduled even there's no matched actor.
-   * @return ActorAffinityMatchExpression.
+   * @return LabelMatchExpression.
    */
-  public static ActorAffinityMatchExpression notIn(
+  public static LabelMatchExpression notIn(
       String key, List<String> values, boolean isSoft) {
-    return new ActorAffinityMatchExpression(key, ActorAffinityOperator.NOT_IN, values, isSoft);
+    return new LabelMatchExpression(key, LabelMatchOperator.NOT_IN, values, isSoft);
   }
 
   /**
    * Returns an affinity expression to indicate that the target actor is expected to be scheduled
    * with the actors whose labels exists the specified key. eg:
-   * ActorAffinityMatchExpression.exists("location", false).
+   * LabelMatchExpression.exists("location", false).
    *
    * @param key The key of label.
    * @param isSoft If true, the actor will be scheduled even there's no matched actor.
-   * @return ActorAffinityMatchExpression.
+   * @return LabelMatchExpression.
    */
-  public static ActorAffinityMatchExpression exists(String key, boolean isSoft) {
-    return new ActorAffinityMatchExpression(
-        key, ActorAffinityOperator.EXISTS, new ArrayList<String>(), isSoft);
+  public static LabelMatchExpression exists(String key, boolean isSoft) {
+    return new LabelMatchExpression(
+        key, LabelMatchOperator.EXISTS, new ArrayList<String>(), isSoft);
   }
 
   /**
    * Returns an affinity expression to indicate that the target actor is not expected to be
    * scheduled with the actors whose labels exists the specified key. eg:
-   * ActorAffinityMatchExpression.doesNotExist("location", false).
+   * LabelMatchExpression.doesNotExist("location", false).
    *
    * @param key The key of label.
    * @param isSoft If true, the actor will be scheduled even there's no matched actor.
-   * @return ActorAffinityMatchExpression.
+   * @return LabelMatchExpression.
    */
-  public static ActorAffinityMatchExpression doesNotExist(String key, boolean isSoft) {
-    return new ActorAffinityMatchExpression(
-        key, ActorAffinityOperator.DOES_NOT_EXIST, new ArrayList<String>(), isSoft);
+  public static LabelMatchExpression doesNotExist(String key, boolean isSoft) {
+    return new LabelMatchExpression(
+        key, LabelMatchOperator.DOES_NOT_EXIST, new ArrayList<String>(), isSoft);
   }
 
 }
@@ -282,7 +282,7 @@ locationValues.add("dc_1");
 locationValues.add("dc_2");
 ActorAffinitySchedulingStrategy schedulingStrategy =
     new ActorAffinitySchedulingStrategy.Builder()
-        .addExpression(ActorAffinityMatchExpression.in("location", locationValues, false))
+        .addExpression(LabelMatchExpression.in("location", locationValues, false))
         .build();
 ActorHandle<Counter> actor2 =
     Ray.actor(Counter::new, 1).setSchedulingStrategy(schedulingStrategy).remote();
@@ -294,7 +294,7 @@ List<String> values = new ArrayList<>();
 values.add("dc-1");
 ActorAffinitySchedulingStrategy schedulingStrategyNotIn =
     new ActorAffinitySchedulingStrategy.Builder()
-        .addExpression(ActorAffinityMatchExpression.notIn("location", values, false))
+        .addExpression(LabelMatchExpression.notIn("location", values, false))
         .build();
 ActorHandle<Counter> actor3 =
     Ray.actor(Counter::new, 1).setSchedulingStrategy(schedulingStrategyNotIn).remote();
@@ -304,7 +304,7 @@ ActorHandle<Counter> actor3 =
 ```java
 ActorAffinitySchedulingStrategy schedulingStrategyExists =
     new ActorAffinitySchedulingStrategy.Builder()
-        .addExpression(ActorAffinityMatchExpression.exists("version", false))
+        .addExpression(LabelMatchExpression.exists("version", false))
         .build();
 ActorHandle<Counter> actor4 =
     Ray.actor(Counter::new, 1).setSchedulingStrategy(schedulingStrategyExists).remote();
@@ -315,7 +315,7 @@ Assert.assertEquals(actor4.task(Counter::getValue).remote().get(10000), Integer.
 ```java
 ActorAffinitySchedulingStrategy schedulingStrategyDoesNotExist =
     new ActorAffinitySchedulingStrategy.Builder()
-        .addExpression(ActorAffinityMatchExpression.doesNotExist("version", false))
+        .addExpression(LabelMatchExpression.doesNotExist("version", false))
         .build();
 ActorHandle<Counter> actor5 =
     Ray.actor(Counter::new, 1).setSchedulingStrategy(schedulingStrategyDoesNotExist).remote();
@@ -325,8 +325,8 @@ ActorHandle<Counter> actor5 =
 ```java
 ActorAffinitySchedulingStrategy schedulingStrategy =
     new ActorAffinitySchedulingStrategy.Builder()
-        .addExpression(ActorAffinityMatchExpression.doesNotExist("version", false))
-        .addExpression(ActorAffinityMatchExpression.Exists("location", false))
+        .addExpression(LabelMatchExpression.doesNotExist("version", false))
+        .addExpression(LabelMatchExpression.Exists("location", false))
         .build();
 ActorHandle<Counter> actor6 =
     Ray.actor(Counter::new, 1).setSchedulingStrategy(schedulingStrategy).remote();
@@ -412,10 +412,10 @@ message ResourcesData {
   // heartbeat enabled.
   bool resources_available_changed = 3;
 
-  // Map<key, Map<value, reference_count>> Actors scheduled to this node and actor labels information
-  repeat Map<string, Map<string, int>> actor_labels = 15
+  // Map<label_type, Map<namespace, Map<label_key, label_value>>> Actors/Tasks/Nodes labels information
+  repeat Map<string, Map<string, Map<string, string>>> labels = 15
   // Whether the actors of this node is changed.
-  bool actor_labels_changed = 16,
+  bool labels_changed = 16,
 }
 
 
@@ -465,7 +465,7 @@ Actor scheduling flowchart：
 ![Actor scheduling flowchart](https://user-images.githubusercontent.com/11072802/202128385-f72609c5-308d-4210-84ff-bf3ba6df381c.png)
 
 Node Resources synchronization mechanism:
-![Node Resources synchronization mechanism](https://user-images.githubusercontent.com/11072802/202128406-b4745e6e-3565-41a2-bfe3-78843379bf09.png)
+![Node Resources synchronization mechanism](https://user-images.githubusercontent.com/11072802/203783157-fad67f25-b046-49ac-b201-b54942073823.png)
 
 4. Scheduling optimization through ActorLabels  
 Now any node raylet has ActorLabels information for all nodes. 
@@ -491,16 +491,6 @@ class GcsLabelManager {
 }
 ```
 
-<b> VS.  putting Labels into the custom resource solution  <b>  
-<b>Advantages:<b>
-1. Compared with the scheme of putting Labels in the custom resource. This scheme can also reuse the resource synchronization mechanism. Then it won't destroy the concept of coustrom resouce.
-2. The Label index table of all nodes can be constructed from the ActorLabels information of each node. If you use Custom Resource, you can't build.
-3. If the Custom Resouces scheme is used, the accuracy of custom resouces scheduling will be affected. The current scheme is completely independent of existing scheduling policies and resources and will not affect them. The code is also more concise.
-
-
-<b>DisAdvantages:<b>
-1. The interface for resource reporting and updating needs to be adapted to ActorLabels in ResouceData.
-
 <b>Issue<b>
 1. Because there must be a delay in resource synchronization under raylet scheduling. So if actor affinity is Soft semantics, there will be inaccurate scheduling.
 
@@ -523,8 +513,40 @@ This solution is to learn the PodAffinity/NodeAffinity features of K8s。
 https://kubernetes.io/docs/concepts/scheduling-eviction/assign-pod-node/#affinity-and-anti-affinity
 
 ### what's the alternative to achieve the same goal?
-1、Ray 
-Now ray placement group can achieve the same goal. But PG is too heavy and complicated to be user friendly
+<b> Option 2: putting Labels into the custom resource solution  <b>  
+```
+class ResourceRequest {
+    absl::flat_hash_map<ResourceID, FixedPoint> resources_;
+}
+
+Add Actor/Task/Node labels to resources_ as custom resources.
+eg:  
+{
+    "CPU": 16,
+    "memory": xx,
+    "custom_resources": xx,
+    "actor_labels_key@value": 1,
+    "task_labels_key@value": 1,
+    "node_labels_key@value": 1,
+}
+```
+If you put labels into custom_resources, you need to do the following adaptation:
+1. Compared with custom_resources, labels need to add a specific prefix to distinguish them from custom_resources.
+2. The key and value of Labels need to be concatenated with special characters (@).
+3. When using Labels to build a Labels index table, you need to parse the resources key.
+
+<b>DisAdvantages:<b>
+1. Labels unlike cpu resource these are numeric types. Compared with the above scheme. This will destroy the concept of coustrom resouce.
+2. Actor and Task are isolated by namespace. It is difficult to isolate through namespace if adding custom_resource.
+2. The Label index table of all nodes can be constructed from the ActorLabels information of each node. If you use Custom Resource, this requires parsing the resource_key and doing a lot of string splitting which will cost performance.
+3. If custom_resource happens to be the same as the spliced string of labels. Then it will affect the correctness of scheduling.
+
+
+<b>Advantages:<b>
+1. The interface for resource reporting and updating don't modify.
+
+
+
 ## Compatibility, Deprecation, and Migration Plan
 
 ## Test Plan and Acceptance Criteria
@@ -536,7 +558,7 @@ All APIs will be fully unit tested. All specifications in this documentation wil
 Later, if necessary, you can extend the semantics of "OR" by adding "is_or_semantics" to ActorAffinitySchedulingStrategy.
 ```
 class ActorAffinitySchedulingStrategy:
-    def __init__(self, match_expressions: List[ActorAffinityMatchExpression], is_or_semantics = false):
+    def __init__(self, match_expressions: List[LabelMatchExpression], is_or_semantics = false):
         self.match_expressions = match_expressions
         self.is_or_semantics = 
 ```