[Nightly test] Bring back the old way of running commands. (#22209)

Bring back the old way of running commands for non-k8s tests.

This also fixes the regression from many_drivers.py

release/e2e.py

@@ -479,7 +479,15 @@ class S3SyncSessionController(SessionController):
         )
         global_command_runner.wait_command(cid)
 
-    def push(self, session_name, source, target):
+    def push(
+        self,
+        session_name: str,
+        source: Optional[str],
+        target: Optional[str],
+        config: Optional[str],
+        all_nodes: bool,
+        no_warning: bool = False,
+    ):
         if source is None and target is None:
             self._push_local_dir(session_name)
             return
@@ -1153,6 +1161,7 @@ def create_and_wait_for_session(
     session_url = anyscale_session_url(
         project_id=GLOBAL_CONFIG["ANYSCALE_PROJECT"], session_id=session_id
     )
+    logger.info(f"URL: {session_url}")
     logger.info(f"Link to session: {_format_link(session_url)}")
 
     result = sdk.start_session(session_id, start_session_options={})
@@ -1215,6 +1224,7 @@ def run_session_command(
     session_url = anyscale_session_url(
         project_id=GLOBAL_CONFIG["ANYSCALE_PROJECT"], session_id=session_id
     )
+    logger.info(f"URL: {session_url}")
     logger.info(f"Link to session: {_format_link(session_url)}")
     result_queue.put(State(state_str, time.time(), None))
     result = sdk.create_session_command(
@@ -1496,7 +1506,11 @@ def run_test_config(
         cli_token=GLOBAL_CONFIG["ANYSCALE_CLI_TOKEN"],
         host=GLOBAL_CONFIG["ANYSCALE_HOST"],
     )
-    session_controller = S3SyncSessionController(sdk, result_queue)
+    on_k8s = test_config["cluster"].get("compute_on_k8s")
+    if on_k8s:
+        session_controller = S3SyncSessionController(sdk, result_queue)
+    else:
+        session_controller = SessionController()
 
     cloud_id = test_config["cluster"].get("cloud_id", None)
     cloud_name = test_config["cluster"].get("cloud_name", None)
@@ -1775,6 +1789,8 @@ def run_test_config(
                     session_name=session_name,
                     source=None,
                     target=None,
+                    config=None,
+                    all_nodes=False,
                 )
 
                 logger.info("Syncing test state to session...")
@@ -1782,6 +1798,8 @@ def run_test_config(
                     session_name=session_name,
                     source=test_state_file,
                     target=state_json,
+                    config=None,
+                    all_nodes=False,
                 )
 
                 session_url = anyscale_session_url(

release/long_running_tests/workloads/many_drivers.py

@@ -105,9 +105,19 @@ parser = argparse.ArgumentParser(prog="Many Drivers long running tests")
 parser.add_argument(
     "--iteration-num", type=int, help="How many iterations to run", required=False
 )
+parser.add_argument(
+    "--smoke-test",
+    action="store_true",
+    help="Whether or not the test is smoke test.",
+    default=False,
+)
 args = parser.parse_args()
+
+iteration_num = args.iteration_num
+if args.smoke_test:
+    iteration_num = 400
 while True:
-    if args.iteration_num is not None and args.iteration_num < iteration:
+    if iteration_num is not None and iteration_num < iteration:
         break
     # Wait for a driver to finish and start a new driver.
     [ready_id], running_ids = ray.wait(running_ids, num_returns=1)