[autoscaler] Make NodeLaunchException serializable (ray-project#31256)

In order to make node launch exceptions serializable, drop the optional stack trace which isn't serializable.

Co-authored-by: Alex <[email protected]>
Signed-off-by: tmynn <[email protected]>

python/ray/autoscaler/node_launch_exception.py

@@ -27,3 +27,11 @@ def __init__(
         self.category = category
         self.description = description
         self.src_exc_info = src_exc_info
+
+    def __reduce__(self):
+        # NOTE: Since tracebacks can't be pickled, we'll drop the optional
+        # traceback if we have to serialize this object.
+        return (
+            self.__class__,
+            (self.category, self.description, None),
+        )

python/ray/tests/test_autoscaler_e2e.py

@@ -2,11 +2,13 @@
 from ray.autoscaler._private.constants import AUTOSCALER_METRIC_PORT
 
 import ray
+import sys
 from ray._private.test_utils import (
     wait_for_condition,
     get_metric_check_condition,
 )
 from ray.cluster_utils import AutoscalingCluster
+from ray.autoscaler.node_launch_exception import NodeLaunchException
 
 
 def test_ray_status_e2e(shutdown_only):
@@ -111,8 +113,26 @@ def ping(self):
         cluster.shutdown()
 
 
+def test_node_launch_exception_serialization(shutdown_only):
+    ray.init(num_cpus=1)
+
+    exc_info = None
+    try:
+        raise Exception("Test exception.")
+    except Exception:
+        exc_info = sys.exc_info()
+    assert exc_info is not None
+
+    exc = NodeLaunchException("cat", "desc", exc_info)
+
+    after_serialization = ray.get(ray.put(exc))
+
+    assert after_serialization.category == exc.category
+    assert after_serialization.description == exc.description
+    assert after_serialization.src_exc_info is None
+
+
 if __name__ == "__main__":
-    import sys
     import os
     import pytest