Repository: skypilot-org/skypilot
Pull request: [Autodown] Support for autodown
Closes #954.

Please refer to the following tests for the design of the CLI.
Main changes:
1. Add an option `--terminate` for`sky launch` and `sky start`
2. Add an option `--terminate` for `sky autostop`
3. `sky.launch` replaces the `teardown` option with `terminate`.


Tested:
- [x] `sky launch -c test-autodown -i 1 --terminate`; `sky status -r`
- [x] `sky cpunode -c test-autodown --cloud gcp`; `sky autostop  --terminate test-autodown`; `sky status -r`
- [x] `sky cpunode -c test-autodown`; `sky stop test-autodown`; `sky start test-autodown -i 1 --terminate`; `sky status -r`
- [x] `tests/run_smoke_tests.sh test_autodown`
- [x] `sky launch -c test-autodown -i 2 --terminate`; `sky status`
```
NAME             LAUNCHED        RESOURCES             STATUS   AUTOSTOP      COMMAND                         
test-autodown    a few secs ago  1x AWS(m6i.2xlarge)   UP       2 min (down)  sky launch -c test-autodown...  
smoke-test-zhwu  4 days ago      1x GCP(n1-highmem-8)  STOPPED  -             sky start smoke-test-zhwu      
```

Problem statement:
Design auto-down, similar to auto-stop
This came up during a user conversation. The user has been using auto-stopping, and he'd also like an auto-down equivalent. 

To design this, we need to think about
- Do we provide both CLIs, e.g., `sky autostop` and `sky autodown` co-exist.  In code as well.
- Alternatively, we can keep one code path, but provide a `auto_down_or_stop: bool`.
- Think about the existing `-i` option in `sky launch`.  Should we rename it to `sky launch [--autostop n] / [--autodown n]`?

Hints from issue discussion:

Commit message: fix comment

Patch to review:
diff --git a/sky/backends/cloud_vm_ray_backend.py b/sky/backends/cloud_vm_ray_backend.py
index 4043051196c..2c3caaf0765 100644
--- a/sky/backends/cloud_vm_ray_backend.py
+++ b/sky/backends/cloud_vm_ray_backend.py
@@ -2608,10 +2608,11 @@ def post_teardown_cleanup(self,
     def set_autostop(self,
                      handle: ResourceHandle,
                      idle_minutes_to_autostop: Optional[int],
+                     terminate: bool = True,
                      stream_logs: bool = True) -> None:
         if idle_minutes_to_autostop is not None:
             code = autostop_lib.AutostopCodeGen.set_autostop(
-                idle_minutes_to_autostop, self.NAME)
+                idle_minutes_to_autostop, self.NAME, terminate)
             returncode, _, stderr = self.run_on_head(handle,
                                                      code,
                                                      require_outputs=True,
diff --git a/sky/cli.py b/sky/cli.py
index 0f36ad71cf2..88237d8fb37 100644
--- a/sky/cli.py
+++ b/sky/cli.py
@@ -604,6 +604,7 @@ def _launch_with_confirm(
     detach_run: bool,
     no_confirm: bool = False,
     idle_minutes_to_autostop: Optional[int] = None,
+    terminate: bool = False,
     retry_until_up: bool = False,
     no_setup: bool = False,
     node_type: Optional[str] = None,
@@ -654,6 +655,7 @@ def _launch_with_confirm(
             detach_run=detach_run,
             backend=backend,
             idle_minutes_to_autostop=idle_minutes_to_autostop,
+            terminate=terminate,
             retry_until_up=retry_until_up,
             no_setup=no_setup,
         )
@@ -992,6 +994,15 @@ def cli():
           'Setting this flag is equivalent to '
           'running ``sky launch -d ...`` and then ``sky autostop -i <minutes>``'
           '. If not set, the cluster will not be auto-stopped.'))
+@click.option(
+    '--terminate',
+    default=False,
+    is_flag=True,
+    required=False,
+    help=(
+        'Terminate the cluster after execution. If --idle-minutes-to-autostop '
+        'is set, the cluster will be torn down after the idle time.'),
+)
 @click.option(
     '--retry-until-up',
     '-r',
@@ -1033,6 +1044,7 @@ def launch(
     env: List[Dict[str, str]],
     disk_size: Optional[int],
     idle_minutes_to_autostop: Optional[int],
+    terminate: bool,
     retry_until_up: bool,
     yes: bool,
     no_setup: bool,
@@ -1083,6 +1095,7 @@ def launch(
         detach_run=detach_run,
         no_confirm=yes,
         idle_minutes_to_autostop=idle_minutes_to_autostop,
+        terminate=terminate,
         retry_until_up=retry_until_up,
         no_setup=no_setup,
         is_local_cloud=onprem_utils.check_if_local_cloud(cluster))
@@ -1499,6 +1512,12 @@ def stop(
               is_flag=True,
               required=False,
               help='Cancel the auto-stopping.')
+@click.option(
+    '--terminate',
+    default=False,
+    is_flag=True,
+    required=False,
+    help='Terminate the cluster instead of stopping it, when auto-stopping.')
 @click.option('--yes',
               '-y',
               is_flag=True,
@@ -1511,6 +1530,7 @@ def autostop(
     all: Optional[bool],  # pylint: disable=redefined-builtin
     idle_minutes: Optional[int],
     cancel: bool,  # pylint: disable=redefined-outer-name
+    terminate: bool,
     yes: bool,
 ):
     """Schedule or cancel auto-stopping for cluster(s).
@@ -1548,7 +1568,7 @@ def autostop(
         idle_minutes = 5
     _terminate_or_stop_clusters(clusters,
                                 apply_to_all=all,
-                                terminate=False,
+                                terminate=terminate,
                                 no_confirm=yes,
                                 idle_minutes_to_autostop=idle_minutes)
 
@@ -1584,6 +1604,15 @@ def autostop(
           'Setting this flag is equivalent to '
           'running ``sky launch -d ...`` and then ``sky autostop -i <minutes>``'
           '. If not set, the cluster will not be auto-stopped.'))
+@click.option(
+    '--terminate',
+    default=False,
+    is_flag=True,
+    required=False,
+    help=(
+        'Terminate the cluster after execution. If --idle-minutes-to-autostop '
+        'is set, the cluster will be torn down after the idle time.'),
+)
 @click.option(
     '--retry-until-up',
     '-r',
@@ -1595,7 +1624,8 @@ def autostop(
 @usage_lib.entrypoint
 # pylint: disable=redefined-builtin
 def start(clusters: Tuple[str], all: bool, yes: bool,
-          idle_minutes_to_autostop: int, retry_until_up: bool):
+          idle_minutes_to_autostop: Optional[int], terminate: bool,
+          retry_until_up: bool):
     """Restart cluster(s).
 
     If a cluster is previously stopped (status is STOPPED) or failed in
@@ -1711,7 +1741,10 @@ def start(clusters: Tuple[str], all: bool, yes: bool,
 
     for name in to_start:
         try:
-            core.start(name, idle_minutes_to_autostop, retry_until_up)
+            core.start(name,
+                       idle_minutes_to_autostop,
+                       retry_until_up,
+                       terminate=terminate)
         except exceptions.NotSupportedError as e:
             click.echo(str(e))
         click.secho(f'Cluster {name} started.', fg='green')
@@ -1796,8 +1829,6 @@ def _terminate_or_stop_clusters(
     name is explicitly and uniquely specified (not via glob) and purge is set
     to True.
     """
-    assert idle_minutes_to_autostop is None or not terminate, (
-        idle_minutes_to_autostop, terminate)
     command = 'down' if terminate else 'stop'
     if not names and apply_to_all is None:
         raise click.UsageError(
@@ -1807,7 +1838,8 @@ def _terminate_or_stop_clusters(
     operation = 'Terminating' if terminate else 'Stopping'
     if idle_minutes_to_autostop is not None:
         verb = 'Scheduling' if idle_minutes_to_autostop >= 0 else 'Cancelling'
-        operation = f'{verb} auto-stop on'
+        down_str = ' (terminate)' if terminate else ''
+        operation = f'{verb} auto-stop{down_str} on'
 
     if len(names) > 0:
         reserved_clusters = [
@@ -1902,7 +1934,7 @@ def _terminate_or_stop(name: str):
         success_progress = False
         if idle_minutes_to_autostop is not None:
             try:
-                core.autostop(name, idle_minutes_to_autostop)
+                core.autostop(name, idle_minutes_to_autostop, terminate)
             except (exceptions.NotSupportedError,
                     exceptions.ClusterNotUpError) as e:
                 message = str(e)
diff --git a/sky/core.py b/sky/core.py
index 87d2a9846ac..00774983c46 100644
--- a/sky/core.py
+++ b/sky/core.py
@@ -53,7 +53,8 @@ def status(all: bool, refresh: bool) -> List[Dict[str, Any]]:
 
 def _start(cluster_name: str,
            idle_minutes_to_autostop: Optional[int] = None,
-           retry_until_up: bool = False) -> backends.Backend.ResourceHandle:
+           retry_until_up: bool = False,
+           terminate: bool = False) -> backends.Backend.ResourceHandle:
 
     cluster_status, handle = backend_utils.refresh_cluster_status_handle(
         cluster_name)
@@ -79,13 +80,16 @@ def _start(cluster_name: str,
                                cluster_name=cluster_name,
                                retry_until_up=retry_until_up)
     if idle_minutes_to_autostop is not None:
-        backend.set_autostop(handle, idle_minutes_to_autostop)
+        backend.set_autostop(handle,
+                             idle_minutes_to_autostop,
+                             terminate=terminate)
     return handle
 
 
 def start(cluster_name: str,
           idle_minutes_to_autostop: Optional[int] = None,
-          retry_until_up: bool = False):
+          retry_until_up: bool = False,
+          terminate: bool = False):
     """Start the cluster.
 
     Please refer to the sky.cli.start for the document.
@@ -93,7 +97,7 @@ def start(cluster_name: str,
     Raises:
         sky.exceptions.NotSupportedError: the cluster is not supported.
     """
-    _start(cluster_name, idle_minutes_to_autostop, retry_until_up)
+    _start(cluster_name, idle_minutes_to_autostop, retry_until_up, terminate)
 
 
 def stop(cluster_name: str, purge: bool = False):
@@ -156,7 +160,9 @@ def down(cluster_name: str, purge: bool = False):
     backend.teardown(handle, terminate=True, purge=purge)
 
 
-def autostop(cluster_name: str, idle_minutes_to_autostop: int):
+def autostop(cluster_name: str,
+             idle_minutes_to_autostop: int,
+             terminate: bool = False):
     """Set the autostop time of the cluster.
 
     Please refer to the sky.cli.autostop for the document.
@@ -170,7 +176,8 @@ def autostop(cluster_name: str, idle_minutes_to_autostop: int):
         sky.exceptions.ClusterNotUpError: the cluster is not UP.
     """
     verb = 'Scheduling' if idle_minutes_to_autostop >= 0 else 'Cancelling'
-    operation = f'{verb} auto-stop on'
+    down_str = ' (tear down)' if terminate else ''
+    operation = f'{verb} auto-stop{down_str} on'
     if cluster_name in backend_utils.SKY_RESERVED_CLUSTER_NAMES:
         raise exceptions.NotSupportedError(
             f'{operation} sky reserved cluster {cluster_name!r} '
@@ -202,7 +209,7 @@ def autostop(cluster_name: str, idle_minutes_to_autostop: int):
                 f'{colorama.Style.RESET_ALL}'
                 '\n  Auto-stop can only be set/unset for '
                 f'{global_user_state.ClusterStatus.UP.value} clusters.')
-    backend.set_autostop(handle, idle_minutes_to_autostop)
+    backend.set_autostop(handle, idle_minutes_to_autostop, terminate)
 
 
 # ==================
diff --git a/sky/execution.py b/sky/execution.py
index da60737feae..b87eea0e44e 100644
--- a/sky/execution.py
+++ b/sky/execution.py
@@ -93,13 +93,13 @@ class Stage(enum.Enum):
     SETUP = enum.auto()
     PRE_EXEC = enum.auto()
     EXEC = enum.auto()
-    TEARDOWN = enum.auto()
+    TERMINATE = enum.auto()
 
 
 def _execute(
     dag: sky.Dag,
     dryrun: bool = False,
-    teardown: bool = False,
+    terminate: bool = False,
     stream_logs: bool = True,
     handle: Any = None,
     backend: Optional[backends.Backend] = None,
@@ -120,8 +120,9 @@ def _execute(
       dag: sky.Dag.
       dryrun: bool; if True, only print the provision info (e.g., cluster
         yaml).
-      teardown: bool; whether to teardown the launched resources after
-        execution.
+      terminate: bool; whether to terminate the launched resources after
+        execution. If idle_minutes_to_autostop is set, the cluster will be
+        terminated after the specified idle time.
       stream_logs: bool; whether to stream all tasks' outputs to the client.
       handle: Any; if provided, execution will use an existing backend cluster
         handle instead of provisioning a new one.
@@ -214,7 +215,9 @@ def _execute(
 
         if stages is None or Stage.PRE_EXEC in stages:
             if idle_minutes_to_autostop is not None:
-                backend.set_autostop(handle, idle_minutes_to_autostop)
+                backend.set_autostop(handle,
+                                     idle_minutes_to_autostop,
+                                     terminate=terminate)
 
         if stages is None or Stage.EXEC in stages:
             try:
@@ -222,10 +225,10 @@ def _execute(
                 backend.execute(handle, task, detach_run)
             finally:
                 # Enables post_execute() to be run after KeyboardInterrupt.
-                backend.post_execute(handle, teardown)
+                backend.post_execute(handle, terminate)
 
-        if stages is None or Stage.TEARDOWN in stages:
-            if teardown:
+        if stages is None or Stage.TERMINATE in stages:
+            if terminate and idle_minutes_to_autostop is None:
                 backend.teardown_ephemeral_storage(task)
                 backend.teardown(handle, terminate=True)
     finally:
@@ -253,7 +256,7 @@ def launch(
     retry_until_up: bool = False,
     idle_minutes_to_autostop: Optional[int] = None,
     dryrun: bool = False,
-    teardown: bool = False,
+    terminate: bool = False,
     stream_logs: bool = True,
     backend: Optional[backends.Backend] = None,
     optimize_target: OptimizeTarget = OptimizeTarget.COST,
@@ -290,7 +293,7 @@ def launch(
     _execute(
         dag=dag,
         dryrun=dryrun,
-        teardown=teardown,
+        terminate=terminate,
         stream_logs=stream_logs,
         handle=None,
         backend=backend,
@@ -308,7 +311,7 @@ def exec(  # pylint: disable=redefined-builtin
     dag: sky.Dag,
     cluster_name: str,
     dryrun: bool = False,
-    teardown: bool = False,
+    terminate: bool = False,
     stream_logs: bool = True,
     backend: Optional[backends.Backend] = None,
     optimize_target: OptimizeTarget = OptimizeTarget.COST,
@@ -327,7 +330,7 @@ def exec(  # pylint: disable=redefined-builtin
                              'Use `sky status` to check the status.')
     _execute(dag=dag,
              dryrun=dryrun,
-             teardown=teardown,
+             terminate=terminate,
              stream_logs=stream_logs,
              handle=handle,
              backend=backend,
diff --git a/sky/skylet/autostop_lib.py b/sky/skylet/autostop_lib.py
index 73c74f3f510..80a990dd23f 100644
--- a/sky/skylet/autostop_lib.py
+++ b/sky/skylet/autostop_lib.py
@@ -10,14 +10,23 @@
 
 
 class AutostopConfig:
+    """Autostop configuration."""
 
-    def __init__(self, autostop_idle_minutes: int, boot_time: int,
-                 backend: Optional[str]):
+    def __init__(self,
+                 autostop_idle_minutes: int,
+                 boot_time: int,
+                 backend: Optional[str],
+                 terminate: bool = False):
         assert autostop_idle_minutes < 0 or backend is not None, (
             autostop_idle_minutes, backend)
         self.autostop_idle_minutes = autostop_idle_minutes
         self.boot_time = boot_time
         self.backend = backend
+        self.terminate = terminate
+
+    def __set_state__(self, state: dict):
+        state.setdefault('terminate', False)
+        self.__dict__.update(state)
 
 
 def get_autostop_config() -> Optional[AutostopConfig]:
@@ -27,9 +36,11 @@ def get_autostop_config() -> Optional[AutostopConfig]:
     return pickle.loads(config_str)
 
 
-def set_autostop(idle_minutes: int, backend: Optional[str]) -> None:
+def set_autostop(idle_minutes: int, backend: Optional[str],
+                 terminate: bool) -> None:
     boot_time = psutil.boot_time()
-    autostop_config = AutostopConfig(idle_minutes, boot_time, backend)
+    autostop_config = AutostopConfig(idle_minutes, boot_time, backend,
+                                     terminate)
     configs.set_config(AUTOSTOP_CONFIG_KEY, pickle.dumps(autostop_config))
 
 
@@ -43,9 +54,11 @@ class AutostopCodeGen:
     _PREFIX = ['from sky.skylet import autostop_lib']
 
     @classmethod
-    def set_autostop(cls, idle_minutes: int, backend: str) -> str:
+    def set_autostop(cls, idle_minutes: int, backend: str,
+                     terminate: bool) -> str:
         code = [
-            f'autostop_lib.set_autostop({idle_minutes}, {backend!r})',
+            f'autostop_lib.set_autostop({idle_minutes}, {backend!r},'
+            f' {terminate})',
         ]
         return cls._build(code)
 
diff --git a/sky/skylet/events.py b/sky/skylet/events.py
index 7f5fbaba17a..df9102670f3 100644
--- a/sky/skylet/events.py
+++ b/sky/skylet/events.py
@@ -119,7 +119,8 @@ def _run(self):
     def _stop_cluster(self, autostop_config):
         if (autostop_config.backend ==
                 cloud_vm_ray_backend.CloudVmRayBackend.NAME):
-            self._replace_yaml_for_stopping(self.ray_yaml_path)
+            self._replace_yaml_for_stopping(self.ray_yaml_path,
+                                            autostop_config.terminate)
             # `ray up` is required to reset the upscaling speed and min/max
             # workers. Otherwise, `ray down --workers-only` will continuously
             # scale down and up.
@@ -135,13 +136,18 @@ def _stop_cluster(self, autostop_config):
         else:
             raise NotImplementedError
 
-    def _replace_yaml_for_stopping(self, yaml_path: str):
+    def _replace_yaml_for_stopping(self, yaml_path: str, terminate: bool):
         with open(yaml_path, 'r') as f:
             yaml_str = f.read()
         # Update the number of workers to 0.
         yaml_str = self._NUM_WORKER_PATTERN.sub(r'\g<1>_workers: 0', yaml_str)
         yaml_str = self._UPSCALING_PATTERN.sub(r'upscaling_speed: 0', yaml_str)
-        yaml_str = self._CATCH_NODES.sub(r'cache_stopped_nodes: true', yaml_str)
+        if terminate:
+            yaml_str = self._CATCH_NODES.sub(r'cache_stopped_nodes: false',
+                                             yaml_str)
+        else:
+            yaml_str = self._CATCH_NODES.sub(r'cache_stopped_nodes: true',
+                                             yaml_str)
         config = yaml.safe_load(yaml_str)
         # Set the private key with the existed key on the remote instance.
         config['auth']['ssh_private_key'] = '~/ray_bootstrap_key.pem'


Code review comments:
diff_hunk: @@ -2608,10 +2608,11 @@ def post_teardown_cleanup(self,
     def set_autostop(self,
                      handle: ResourceHandle,
                      idle_minutes_to_autostop: Optional[int],
+                     terminate: bool = True,
path: sky/backends/cloud_vm_ray_backend.py
text: @user1:
Should this default to True or False?

@author:
Good catch! Changed to False.

diff_hunk: @@ -1499,6 +1512,12 @@ def stop(
               is_flag=True,
               required=False,
               help='Cancel the auto-stopping.')
+@click.option(
+    '--terminate',
+    default=False,
+    is_flag=True,
+    required=False,
+    help='Terminate the cluster instead of stopping it, when auto-stopping.')
path: sky/cli.py
text: @user1:
nit: `, when auto-stopping` -> `(i.e., autodown rather than autostop)`?

@author:
Done. Thanks!

diff_hunk: @@ -992,6 +994,15 @@ def cli():
           'Setting this flag is equivalent to '
           'running ``sky launch -d ...`` and then ``sky autostop -i <minutes>``'
           '. If not set, the cluster will not be auto-stopped.'))
+@click.option(
+    '--terminate',
+    default=False,
+    is_flag=True,
+    required=False,
+    help=(
+        'Terminate the cluster after execution. If --idle-minutes-to-autostop '
path: sky/cli.py
text: @user1:

See comment about abnormal job exits

diff_hunk: @@ -120,8 +120,9 @@ def _execute(
       dag: sky.Dag.
       dryrun: bool; if True, only print the provision info (e.g., cluster
         yaml).
-      teardown: bool; whether to teardown the launched resources after
-        execution.
+      terminate: bool; whether to terminate the launched resources after
+        execution. If idle_minutes_to_autostop is set, the cluster will be
path: sky/execution.py
text: @user1:
nit: after the execution of the task exits (successfully or abnormally)

I believe it's cleaned up after abnormal job exits, just double-checking.

@author:
Yes, it will clean up after the job fails, but if the cluster fails at the setup/filemount. 
During adding this I found that it previously will not terminate the cluster when the user ctrl-c'ed during the logging for the execution. I just moved the teardown into the `finally: ...` block to make sure the clean up will happen after the ctrl-c. Wdyt?

diff_hunk: @@ -992,6 +994,15 @@ def cli():
           'Setting this flag is equivalent to '
           'running ``sky launch -d ...`` and then ``sky autostop -i <minutes>``'
           '. If not set, the cluster will not be auto-stopped.'))
+@click.option(
+    '--terminate',
path: sky/cli.py
text: @user1:
How do you feel about the alternative of `--rm`? (E.g., use this name in user interfaces; in the backend we can keep `terminate`)

Pros:
- Popularized by docker: https://docs.docker.com/engine/reference/run/#clean-up---rm
- For some reason `--terminate` to me makes sense only for serverful settings. I can imagine in the future `sky launch` may gain the ability to launch serverless jobs. In that case, "--terminate a job after it exits" is a bit confusing imo.
  - Feel free to push back as (1) this seems a weak argument (2) `--rm` may not be that much better.
- Faster to type.

---

Another alternative is `--down`.

Pros:
- Matches `sky down`.
- Matches the colloquial "autodown".
- Faster to type.

@author:
I think it would be better to keep this consistent with the option in `sky.launch`, so that the user do not need to learn another set of function parameters when using the programetic API. I feel that the `rm` can be a bit confusing when appear in the `sky.launch(..., rm=True, ...)`. Do you think replacing both the `--terminate` and `sky.launch(..., terminate=True, ...)` with `--down` and `sky.launch(..., down=True, ...)` will be better?

@user2:
My 2c: - `--down` >> `--terminate` >> `--rm`

`--down` gets highest preference because its consistent with `sky down`. I'm a little apprehensive of `--terminate` mainly because it adds a new word `terminate` to user vocabulary, which has the same effect as `sky down`. Since `down` is a fundamental operation, I think it's a good idea to maintain consistency here.  

@user1:
`--down` sounds great.

Minor: I see that some user-facing help strings get "terminate" replaced with "tear down". That seems okay but the former is much more common in cloud providers parlance (e.g., AWS console).

@author:
Good point! I found we are using both `tear down` and `terminate` in our docs. I would prefer `tear down`, since it is closer to the `--down` or `sky down`. I went through the docstr's, and tried to make them more consistent, but still keep some of the `terminate`. PTAL. : )

diff_hunk: @@ -1807,7 +1838,8 @@ def _terminate_or_stop_clusters(
     operation = 'Terminating' if terminate else 'Stopping'
     if idle_minutes_to_autostop is not None:
         verb = 'Scheduling' if idle_minutes_to_autostop >= 0 else 'Cancelling'
-        operation = f'{verb} auto-stop on'
+        down_str = ' (terminate)' if terminate else ''
+        operation = f'{verb} auto-stop{down_str} on'
path: sky/cli.py
text: @user1:
Wdyt about
```
> sky autostop --terminate
Scheduling auto-stop (terminate) on ..
```
vs.
```
> sky autostop --down / --rm
Scheduling auto-down on ..
```
? The last code block suggests `--down` may be better than `--rm`.

@author:
I think `auto-down` can be a good option. Updated the log.

diff_hunk: @@ -1595,7 +1624,8 @@ def autostop(
 @usage_lib.entrypoint
 # pylint: disable=redefined-builtin
 def start(clusters: Tuple[str], all: bool, yes: bool,
-          idle_minutes_to_autostop: int, retry_until_up: bool):
+          idle_minutes_to_autostop: Optional[int], terminate: bool,
path: sky/cli.py
text: @user1:
`sky start --terminate myclus` seems non-intuitive. IIUC it will restart the cluster then immediately shut it down. Wdyt about (1) leaving this flag out of this command; (2) keep this; (3) enforce if --terminate is set, -i must be set?

@author:
This is a good point! I prefer the option 3. Just updated code. Wdyt?

@user1:
Sounds great.