[core] Nightly release test with cross AZ fault injection (#57579)

Signed-off-by: dayshah <dhyey2019@gmail.com>
Signed-off-by: joshlee <joshlee@anyscale.com>
Co-authored-by: dayshah <dhyey2019@gmail.com>

Author: Sparks0219

release/nightly_tests/dataset/cross_az_250_350_compute_aws.yaml

@@ -0,0 +1,19 @@
+cloud_id: {{env["ANYSCALE_CLOUD_ID"]}}
+region: us-west-2
+
+head_node_type:
+  name: head
+  instance_type: m5.2xlarge
+  resources:
+    CPU: 0
+    GPU: 0
+
+worker_node_types:
+  - name: worker
+    instance_type: m5.2xlarge
+    min_workers: 250
+    max_workers: 350
+
+flags:
+  enable_multi_az_serve: true
+  allow-cross-zone-autoscaling: true

release/nightly_tests/dataset/cross_az_250_350_compute_gce.yaml

@@ -0,0 +1,19 @@
+cloud_id: {{env["ANYSCALE_CLOUD_ID"]}}
+region: us-west1
+
+head_node_type:
+  name: head
+  instance_type: n2-standard-8
+  resources:
+    CPU: 0
+    GPU: 0
+
+worker_node_types:
+  - name: worker
+    instance_type: n2-standard-8
+    min_workers: 250
+    max_workers: 350
+
+flags:
+  enable_multi_az_serve: true
+  allow-cross-zone-autoscaling: true

release/nightly_tests/dataset/map_benchmark.py

@@ -69,6 +69,13 @@ def parse_args() -> argparse.Namespace:
             "output of the first run as input."
         ),
     )
+    parser.add_argument(
+        "--concurrency",
+        default=[1, 1024],
+        nargs=2,
+        type=int,
+        help="Concurrency to use with 'map_batches'.",
+    )
     return parser.parse_args()
 
 
@@ -80,7 +87,8 @@ def main(args: argparse.Namespace) -> None:
     path = f"s3://ray-benchmark-data/tpch/parquet/sf{args.sf}/lineitem"
     path = [path] * args.repeat_inputs
 
-    def apply_map_batches(ds, use_actors=False):
+    def apply_map_batches(ds):
+        use_actors = args.compute == "actors"
         if not use_actors:
             return ds.map_batches(
                 functools.partial(
@@ -100,7 +108,7 @@ def apply_map_batches(ds, use_actors=False):
                 fn_constructor_args=[model_ref, args.map_batches_sleep_ms],
                 batch_format=args.batch_format,
                 batch_size=args.batch_size,
-                concurrency=(1, 1024),
+                concurrency=tuple(args.concurrency),
             )
 
     def benchmark_fn():
@@ -111,10 +119,9 @@ def benchmark_fn():
         if args.api == "map":
             ds = ds.map(increment_row)
         elif args.api == "map_batches":
-            use_actors = args.compute == "actors"
-            ds = apply_map_batches(ds, use_actors)
+            ds = apply_map_batches(ds)
             if args.repeat_map_batches == "repeat":
-                ds = apply_map_batches(ds, use_actors)
+                ds = apply_map_batches(ds)
         elif args.api == "flat_map":
             ds = ds.flat_map(flat_increment_row)
 

release/release_data_tests.yaml

@@ -646,3 +646,46 @@
   run:
     timeout: 5400
     script: python tpch_q1.py --sf 100
+
+#################################################
+# Cross-AZ RPC fault tolerance test
+#################################################
+
+- name: "cross_az_map_batches_autoscaling"
+  frequency: nightly
+  env: gce
+
+  cluster:
+    cluster_compute: cross_az_250_350_compute_gce.yaml
+
+  run:
+    timeout: 10800
+    script: >
+      python map_benchmark.py --api map_batches --batch-format numpy
+      --compute actors --sf 1000 --repeat-inputs 1 --concurrency 1024 2048
+
+  variations:
+    - __suffix__: gce
+    - __suffix__: aws
+      env: aws
+      cluster:
+        cluster_compute: cross_az_250_350_compute_aws.yaml
+ # TODO(#58246): Enable these variations once RAY_testing_rpc_failure is supported.
+ #   - __suffix__: gce_failure_injection
+ #     cluster:
+ #       byod:
+ #         # RAY_testing_rpc_failure is used to inject RPC failures across all RPCs (*) with no limit (-1) on the number of total failures,
+ #         # 10% request failures, 10% response failures, 1 guaranteed request failure and 1 guaranteed response failure.
+ #         # RAY_testing_rpc_failure_avoid_intra_node_failures=1 is used to avoid injecting RPC failures within the same node.
+ #         runtime_env:
+ #           - RAY_testing_rpc_failure="*=-1:10:10:1:1"
+ #           - RAY_testing_rpc_failure_avoid_intra_node_failures=1
+ #       cluster_compute: cross_az_250_350_compute_gce.yaml
+ #   - __suffix__: aws_failure_injection
+ #     env: aws
+ #     cluster:
+ #       byod:
+ #         runtime_env:
+ #           - RAY_testing_rpc_failure="*=-1:10:10:1:1"
+ #           - RAY_testing_rpc_failure_avoid_intra_node_failures=1
+ #       cluster_compute: cross_az_250_350_compute_aws.yaml