[Serve] Make serialized_policy_def and policy private (ray-projec…

…t#43291)

Make serialized_policy_def and policy private so users know not to use them.

---------

Signed-off-by: Gene Su <[email protected]>

python/ray/_private/pydantic_compat.py

@@ -21,6 +21,7 @@
     NonNegativeInt = None
     PositiveFloat = None
     PositiveInt = None
+    PrivateAttr = None
     StrictInt = None
     ValidationError = None
     root_validator = None
@@ -41,6 +42,7 @@
         NonNegativeInt,
         PositiveFloat,
         PositiveInt,
+        PrivateAttr,
         StrictInt,
         ValidationError,
         root_validator,
@@ -60,6 +62,7 @@ def is_subclass_of_base_model(obj):
         NonNegativeInt,
         PositiveFloat,
         PositiveInt,
+        PrivateAttr,
         StrictInt,
         ValidationError,
         root_validator,

python/ray/serve/config.py

@@ -10,6 +10,7 @@
     NonNegativeInt,
     PositiveFloat,
     PositiveInt,
+    PrivateAttr,
     validator,
 )
 from ray._private.utils import import_attr
@@ -63,10 +64,10 @@ class AutoscalingConfig(BaseModel):
     upscale_delay_s: NonNegativeFloat = 30.0
 
     # Cloudpickled policy definition.
-    serialized_policy_def: bytes = b""
+    _serialized_policy_def: bytes = PrivateAttr(default=b"")
 
     # Custom autoscaling config. Defaults to the request-based autoscaler.
-    policy: Union[str, Callable] = DEFAULT_AUTOSCALING_POLICY
+    _policy: Union[str, Callable] = PrivateAttr(default=DEFAULT_AUTOSCALING_POLICY)
 
     @validator("max_replicas", always=True)
     def replicas_settings_valid(cls, max_replicas, values):
@@ -92,13 +93,18 @@ def replicas_settings_valid(cls, max_replicas, values):
 
         return max_replicas
 
-    @validator("policy", always=True)
-    def serialize_policy(cls, policy, values) -> str:
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+        self.serialize_policy()
+
+    def serialize_policy(self) -> None:
         """Serialize policy with cloudpickle.
 
         Import the policy if it's passed in as a string import path. Then cloudpickle
         the policy and set `serialized_policy_def` if it's empty.
         """
+        values = self.dict()
+        policy = values.get("_policy")
         if isinstance(policy, Callable):
             policy = f"{policy.__module__}.{policy.__name__}"
 
@@ -108,10 +114,9 @@ def serialize_policy(cls, policy, values) -> str:
         policy_path = policy
         policy = import_attr(policy)
 
-        if not values.get("serialized_policy_def"):
-            values["serialized_policy_def"] = cloudpickle.dumps(policy)
-
-        return policy_path
+        if not values.get("_serialized_policy_def"):
+            self._serialized_policy_def = cloudpickle.dumps(policy)
+        self._policy = policy_path
 
     @classmethod
     def default(cls):
@@ -121,7 +126,7 @@ def default(cls):
 
     def get_policy(self) -> Callable:
         """Deserialize policy from cloudpickled bytes."""
-        return cloudpickle.loads(self.serialized_policy_def)
+        return cloudpickle.loads(self._serialized_policy_def)
 
     def get_upscale_smoothing_factor(self) -> PositiveFloat:
         return self.upscale_smoothing_factor or self.smoothing_factor

python/ray/serve/schema.py

@@ -1125,7 +1125,7 @@ def _get_user_facing_json_serializable_dict(
                     and "autoscaling_config" in deployment["deployment_config"]
                 ):
                     deployment["deployment_config"]["autoscaling_config"].pop(
-                        "serialized_policy_def", None
+                        "_serialized_policy_def", None
                     )
 
         return values

python/ray/serve/tests/test_controller.py

@@ -88,10 +88,10 @@ def test_get_serve_instance_details_json_serializable(serve_instance, policy):
     autoscaling_config = {
         "min_replicas": 1,
         "max_replicas": 10,
-        "policy": policy,
+        "_policy": policy,
     }
     if policy is None:
-        autoscaling_config.pop("policy")
+        autoscaling_config.pop("_policy")
 
     @serve.deployment(autoscaling_config=autoscaling_config)
     def autoscaling_app():
@@ -111,11 +111,6 @@ def autoscaling_app():
         controller.get_deployment_details.remote("default", "autoscaling_app")
     )
     replica = deployment_details.replicas[0]
-    policy_path = "ray.serve.autoscaling_policy:default_autoscaling_policy"
-    if policy == default_autoscaling_policy:
-        policy_path = (
-            "ray.serve.autoscaling_policy.replica_queue_length_autoscaling_policy"
-        )
 
     expected_json = json.dumps(
         {
@@ -176,7 +171,6 @@ def autoscaling_app():
                                     "downscale_smoothing_factor": None,
                                     "downscale_delay_s": 600.0,
                                     "upscale_delay_s": 30.0,
-                                    "policy": policy_path,
                                 },
                                 "graceful_shutdown_wait_loop_s": 2.0,
                                 "graceful_shutdown_timeout_s": 20.0,
@@ -214,7 +208,7 @@ def autoscaling_app():
     application = details["applications"]["default"]
     deployment = application["deployments"]["autoscaling_app"]
     autoscaling_config = deployment["deployment_config"]["autoscaling_config"]
-    assert "serialized_policy_def" not in autoscaling_config
+    assert "_serialized_policy_def" not in autoscaling_config
 
 
 if __name__ == "__main__":

python/ray/serve/tests/test_deploy_2.py

@@ -343,7 +343,6 @@ async def __call__(self):
         "downscale_smoothing_factor": None,
         "smoothing_factor": 1.0,
         "initial_replicas": None,
-        "policy": "ray.serve.autoscaling_policy:default_autoscaling_policy",
     }
 
     for i in range(3):

python/ray/serve/tests/test_deploy_app.py

@@ -1287,7 +1287,6 @@ def test_num_replicas_auto(client: ServeControllerClient):
         "downscale_smoothing_factor": None,
         "smoothing_factor": 1.0,
         "initial_replicas": None,
-        "policy": "ray.serve.autoscaling_policy:default_autoscaling_policy",
     }
 
     h = serve.get_app_handle(SERVE_DEFAULT_APP_NAME)

python/ray/serve/tests/unit/test_config.py

@@ -576,17 +576,16 @@ def test_autoscaling_policy_serializations(policy):
     """
     autoscaling_config = AutoscalingConfig()
     if policy:
-        autoscaling_config = AutoscalingConfig(policy=policy)
+        autoscaling_config = AutoscalingConfig(_policy=policy)
 
     config = DeploymentConfig.from_default(autoscaling_config=autoscaling_config)
     deserialized_autoscaling_policy = DeploymentConfig.from_proto_bytes(
         config.to_proto_bytes()
     ).autoscaling_config.get_policy()
 
-    if policy is None:
-        assert deserialized_autoscaling_policy == default_autoscaling_policy
-    else:
-        assert deserialized_autoscaling_policy() == fake_policy_return_value
+    # Right now we don't allow modifying the autoscaling policy, so this will always
+    # be the default autoscaling policy
+    assert deserialized_autoscaling_policy == default_autoscaling_policy
 
 
 def test_autoscaling_policy_import_fails_for_non_existing_policy():
@@ -595,9 +594,9 @@ def test_autoscaling_policy_import_fails_for_non_existing_policy():
     This test will ensure non-existing policy will be caught. It can happen when we
     moved the default policy or when user pass in a non-existing policy.
     """
+    # Right now we don't allow modifying the autoscaling policy, so this will not fail
     policy = "i.dont.exist:fake_policy"
-    with pytest.raises(ModuleNotFoundError):
-        AutoscalingConfig(policy=policy)
+    AutoscalingConfig(_policy=policy)
 
 
 def test_default_autoscaling_policy_import_path():

python/ray/serve/tests/unit/test_schema.py

@@ -833,7 +833,7 @@ def test_serve_instance_details_is_json_serializable():
                             "name": "deployment1",
                             "autoscaling_config": {
                                 # Byte object will cause json serializable error
-                                "serialized_policy_def": serialized_policy_def
+                                "_serialized_policy_def": serialized_policy_def
                             },
                         },
                         "replicas": [],
@@ -880,7 +880,7 @@ def test_serve_instance_details_is_json_serializable():
     application = details["applications"]["app1"]
     deployment = application["deployments"]["deployment1"]
     autoscaling_config = deployment["deployment_config"]["autoscaling_config"]
-    assert "serialized_policy_def" not in autoscaling_config
+    assert "_serialized_policy_def" not in autoscaling_config
 
 
 if __name__ == "__main__":

src/ray/protobuf/serve.proto

@@ -62,11 +62,11 @@ message AutoscalingConfig {
   optional double downscale_smoothing_factor = 11;
 
   // The cloudpickled policy definition.
-  bytes serialized_policy_def = 12;
+  bytes _serialized_policy_def = 12;
 
   // The import path of the policy if user passed a string. Will be the concatenation
   // of the policy module and the policy name if user passed a callable.
-  string policy = 13;
+  string _policy = 13;
 }
 
 //[Begin] LOGGING CONFIG