Repository: skypilot-org/skypilot
Pull request: Catch resource demands (accelerators) inconsistent with instance type.
Fixes #154.

Tested
- new unit tests
- `sky gpunode -t p3.2xlarge --gpus=K80 --cloud aws` correctly fails with the new error
- `sky gpunode -t p3.2xlarge --gpus=V100 --cloud aws` works
- `sky gpunode -t p3dn.24xlarge --cloud aws` works

Problem statement:
accelerators_to_hourly_cost prevents gpunode with user-specified instance type
```
❯ sky gpunode -t p3dn.24xlarge --cloud aws
I 01-08 14:18:15 optimizer.py:206] Defaulting estimated time to 1 hr. Call Task.set_time_estimator() to override.
Traceback (most recent call last):
  File "/Users/gautam/miniconda3/envs/sky/bin/sky", line 33, in <module>
    sys.exit(load_entry_point('sky', 'console_scripts', 'sky')())
  File "/Users/gautam/miniconda3/envs/sky/lib/python3.8/site-packages/click/core.py", line 1128, in __call__
    return self.main(*args, **kwargs)
  File "/Users/gautam/miniconda3/envs/sky/lib/python3.8/site-packages/click/core.py", line 1053, in main
    rv = self.invoke(ctx)
  File "/Users/gautam/miniconda3/envs/sky/lib/python3.8/site-packages/click/core.py", line 1659, in invoke
    return _process_result(sub_ctx.command.invoke(sub_ctx))
  File "/Users/gautam/miniconda3/envs/sky/lib/python3.8/site-packages/click/core.py", line 1395, in invoke
    return ctx.invoke(self.callback, **ctx.params)
  File "/Users/gautam/miniconda3/envs/sky/lib/python3.8/site-packages/click/core.py", line 754, in invoke
    return __callback(*args, **kwargs)
  File "/Users/gautam/Desktop/workbench/sky/prototype/sky/cli.py", line 669, in gpunode
    _create_and_ssh_into_node(
  File "/Users/gautam/Desktop/workbench/sky/prototype/sky/cli.py", line 211, in _create_and_ssh_into_node
    dag = sky.optimize(dag)
  File "/Users/gautam/Desktop/workbench/sky/prototype/sky/optimizer.py", line 76, in optimize
    optimized_dag, unused_best_plan = Optimizer._optimize_cost(
  File "/Users/gautam/Desktop/workbench/sky/prototype/sky/optimizer.py", line 229, in _optimize_cost
    estimated_cost = resources.get_cost(estimated_runtime)
  File "/Users/gautam/Desktop/workbench/sky/prototype/sky/resources.py", line 102, in get_cost
    hourly_cost += self.cloud.accelerators_to_hourly_cost(
  File "/Users/gautam/Desktop/workbench/sky/prototype/sky/clouds/cloud.py", line 77, in accelerators_to_hourly_cost
    raise NotImplementedError
NotImplementedError
```

Hints from issue discussion:

Commit message: Fine tune a few things

Patch to review:
diff --git a/prototype/examples/huggingface_glue_imdb_app.yaml b/prototype/examples/huggingface_glue_imdb_app.yaml
index 1e1ff257e9e..1d32d77c4d9 100644
--- a/prototype/examples/huggingface_glue_imdb_app.yaml
+++ b/prototype/examples/huggingface_glue_imdb_app.yaml
@@ -25,5 +25,5 @@ run: |
     --per_device_train_batch_size 32 \
     --learning_rate 2e-5 \
     --max_steps 50 \
-    --output_dir /tmp/imdb/ \
+    --output_dir /tmp/imdb/ --overwrite_output_dir \
     --fp16
diff --git a/prototype/examples/multi_echo.py b/prototype/examples/multi_echo.py
index d1343c221c5..223956795eb 100644
--- a/prototype/examples/multi_echo.py
+++ b/prototype/examples/multi_echo.py
@@ -6,7 +6,7 @@
     task = sky.Task().set_resources(cluster_resources)
 # `detach_run` will only detach the `run` command. The provision and `setup` are
 # still blocking.
-sky.launch(dag, cluster_name='multi_echo', detach_run=True)
+sky.launch(dag, cluster_name='multi-echo', detach_run=True)
 
 # Run the multiple tasks.
 for i in range(16):
@@ -14,4 +14,4 @@
         task = sky.Task(run=f'echo {i}; sleep 15')
         resources = sky.Resources(accelerators={'K80': 0.1})
         task.set_resources(resources)
-    sky.exec(dag, cluster_name='multi_echo', detach_run=True)
+    sky.exec(dag, cluster_name='multi-echo', detach_run=True)
diff --git a/prototype/sky/backends/cloud_vm_ray_backend.py b/prototype/sky/backends/cloud_vm_ray_backend.py
index 3925de9f828..0761961e952 100644
--- a/prototype/sky/backends/cloud_vm_ray_backend.py
+++ b/prototype/sky/backends/cloud_vm_ray_backend.py
@@ -1233,7 +1233,7 @@ def _exec_code_on_head(
                         f'{backend_utils.BOLD}sky logs -c {name} {job_id}'
                         f'{backend_utils.RESET_BOLD}'
                         '\nTo view the job queue:\t'
-                        f'{backend_utils.BOLD}sky queue -c {name}'
+                        f'{backend_utils.BOLD}sky queue {name}'
                         f'{backend_utils.RESET_BOLD}')
 
     def _add_job(self, handle: ResourceHandle) -> int:
@@ -1384,11 +1384,11 @@ def post_execute(self, handle: ResourceHandle, teardown: bool) -> None:
                         '\nTo submit a job:'
                         f'\t\t{backend_utils.BOLD}sky exec -c {name} yaml_file'
                         f'{backend_utils.RESET_BOLD}'
-                        '\nTo teardown the cluster:'
-                        f'\t{backend_utils.BOLD}sky down {name}'
-                        f'{backend_utils.RESET_BOLD}'
                         '\nTo stop the cluster:'
                         f'\t{backend_utils.BOLD}sky stop {name}'
+                        f'{backend_utils.RESET_BOLD}'
+                        '\nTo teardown the cluster:'
+                        f'\t{backend_utils.BOLD}sky down {name}'
                         f'{backend_utils.RESET_BOLD}')
             if handle.tpu_delete_script is not None:
                 logger.info('Tip: `sky down` will delete launched TPU(s) too.')
diff --git a/prototype/sky/cli.py b/prototype/sky/cli.py
index 81f3822c0b6..057fdf23d80 100644
--- a/prototype/sky/cli.py
+++ b/prototype/sky/cli.py
@@ -217,6 +217,10 @@ def _create_and_ssh_into_node(
     task = dag.tasks[0]
     backend.register_info(dag=dag)
 
+    # FIXME(gautam): typing the following seq of cmds:
+    #   sky gpunode -t p3.2xlarge --gpus=V100 --cloud aws
+    #   sky gpunode -t n1-standard-8 --gpus=V100 --cloud gcp
+    # makes the second cmd log into the first VM.  It should error out.
     handle = global_user_state.get_handle_from_cluster_name(cluster_name)
     if handle is None or handle.head_ip is None:
         # head_ip would be None if previous provisioning failed.
@@ -246,10 +250,10 @@ def _create_and_ssh_into_node(
     else:
         option = f' -c {cluster_name}'
     click.secho(f'sky {node_type}{option}', bold=True)
-    click.echo('To tear down the node:\t', nl=False)
-    click.secho(f'sky down {cluster_name}', bold=True)
     click.echo('To stop the node:\t', nl=False)
     click.secho(f'sky stop {cluster_name}', bold=True)
+    click.echo('To tear down the node:\t', nl=False)
+    click.secho(f'sky down {cluster_name}', bold=True)
 
 
 class _NaturalOrderGroup(click.Group):
@@ -463,7 +467,7 @@ def queue(cluster: Optional[str], skip_finished: bool, all_users: bool):
               help='Name of the existing cluster to find the job.')
 @click.argument('job_id', required=True, type=str)
 def logs(cluster: str, job_id: str):
-    """Tailing the log of a job."""
+    """Tail the log of a job."""
     # TODO: Add an option for downloading logs.
     cluster_name = cluster
     backend = backends.CloudVmRayBackend()
@@ -494,7 +498,8 @@ def logs(cluster: str, job_id: str):
               help='Cancel all jobs.')
 @click.argument('jobs', required=False, type=int, nargs=-1)
 def cancel(cluster: str, all: bool, jobs: List[int]):  # pylint: disable=redefined-builtin
-    """Cancel the jobs on a cluster."""
+    """Cancel job(s)."""
+    # FIXME(zhwu): should not allow SUCCEEDED jobs to be turned into CANCELLED.
     if len(jobs) == 0 and not all:
         raise click.UsageError(
             'sky cancel requires either a job id '
@@ -510,7 +515,8 @@ def cancel(cluster: str, all: bool, jobs: List[int]):  # pylint: disable=redefin
                     fg='yellow')
         jobs = None
     else:
-        click.secho(f'Cancelling jobs {jobs} on cluster {cluster} ...',
+        jobs_str = ', '.join(map(str, jobs))
+        click.secho(f'Cancelling jobs ({jobs_str}) on cluster {cluster} ...',
                     fg='yellow')
 
     codegen = backend_utils.JobLibCodeGen()
@@ -522,41 +528,6 @@ def cancel(cluster: str, all: bool, jobs: List[int]):  # pylint: disable=redefin
     backend.run_on_head(handle, code, stream_logs=False)
 
 
-@cli.command()
-@click.argument('clusters', nargs=-1, required=False)
-@click.option('--all',
-              '-a',
-              default=None,
-              is_flag=True,
-              help='Tear down all existing clusters.')
-def down(
-        clusters: Tuple[str],
-        all: Optional[bool],  # pylint: disable=redefined-builtin
-):
-    """Tear down cluster(s).
-
-    CLUSTER is the name of the cluster to tear down.  If both CLUSTER and --all
-    are supplied, the latter takes precedence.
-
-    Accelerators (e.g., TPU) that are part of the cluster will be deleted too.
-
-    Examples:
-
-      \b
-      # Tear down a specific cluster.
-      sky down cluster_name
-
-      \b
-      # Tear down multiple clusters.
-      sky down cluster1 cluster2
-
-      \b
-      # Tear down all existing clusters.
-      sky down -a
-    """
-    _terminate_or_stop_clusters(clusters, apply_to_all=all, terminate=True)
-
-
 @cli.command()
 @click.argument('clusters', nargs=-1, required=False)
 @click.option('--all',
@@ -687,6 +658,41 @@ def _filter(name, all_clusters):
         click.secho(f'Cluster {name} started.', fg='green')
 
 
+@cli.command()
+@click.argument('clusters', nargs=-1, required=False)
+@click.option('--all',
+              '-a',
+              default=None,
+              is_flag=True,
+              help='Tear down all existing clusters.')
+def down(
+        clusters: Tuple[str],
+        all: Optional[bool],  # pylint: disable=redefined-builtin
+):
+    """Tear down cluster(s).
+
+    CLUSTER is the name of the cluster to tear down.  If both CLUSTER and --all
+    are supplied, the latter takes precedence.
+
+    Accelerators (e.g., TPU) that are part of the cluster will be deleted too.
+
+    Examples:
+
+      \b
+      # Tear down a specific cluster.
+      sky down cluster_name
+
+      \b
+      # Tear down multiple clusters.
+      sky down cluster1 cluster2
+
+      \b
+      # Tear down all existing clusters.
+      sky down -a
+    """
+    _terminate_or_stop_clusters(clusters, apply_to_all=all, terminate=True)
+
+
 def _terminate_or_stop_clusters(names: Tuple[str], apply_to_all: Optional[bool],
                                 terminate: bool) -> None:
     """Terminates or stops a cluster (or all clusters)."""
@@ -780,10 +786,12 @@ def gpunode(cluster: str, port_forward: Optional[List[int]],
             f'Cloud \'{cloud}\' is not supported. ' + \
             f'Supported clouds: {list(task_lib.CLOUD_REGISTRY.keys())}'
         )
-    if gpus is None:
-        gpus = {'K80': 1}
-    else:
+    if gpus is not None:
         gpus = _parse_accelerator_options(gpus)
+    elif instance_type is None:
+        # Use this request if both gpus and instance_type are not specified.
+        gpus = {'K80': 1}
+
     resources = sky.Resources(cloud=cloud_provider,
                               instance_type=instance_type,
                               accelerators=gpus,
diff --git a/prototype/sky/resources.py b/prototype/sky/resources.py
index d6bff4d23b9..3798594807d 100644
--- a/prototype/sky/resources.py
+++ b/prototype/sky/resources.py
@@ -61,6 +61,8 @@ def __init__(
         self.accelerator_args = accelerator_args
         self.use_spot = use_spot
 
+        self._clear_redundant_accelerators_field()
+
     def __repr__(self) -> str:
         accelerators = ''
         accelerator_args = ''
@@ -77,6 +79,29 @@ def __repr__(self) -> str:
     def is_launchable(self) -> bool:
         return self.cloud is not None and self.instance_type is not None
 
+    def _clear_redundant_accelerators_field(self) -> None:
+        """Try-clears any specified accelerators field for non-GCP."""
+        # GCP must have 'self.accelerators' because it attaches accelerators to
+        # regular VMs.  For other clouds, this field may be
+        # reduandant/inconsistent with the specified instance_type.
+        if self.is_launchable() and not isinstance(self.cloud, clouds.GCP):
+            acc_requested = self.accelerators
+            if acc_requested is None:
+                return
+            acc_from_instance_type = (
+                self.cloud.get_accelerators_from_instance_type(
+                    self.instance_type))
+            if acc_from_instance_type != acc_requested:
+                raise ValueError(
+                    'Infeasible resource demands found:\n'
+                    f'  Instance type requested: {self.instance_type}\n'
+                    f'  Accelerators for {self.instance_type}: '
+                    f'{acc_from_instance_type}\n'
+                    f'  Accelerators requested: {acc_requested}\n'
+                    f'To fix: either only specify instance_type, or change '
+                    'the accelerators field to be consistent.')
+            self.accelerators = None
+
     def get_accelerators(self) -> Optional[Dict[str, int]]:
         """Returns the accelerators field directly or by inferring.
 
diff --git a/prototype/tests/test_optimizer_dryruns.py b/prototype/tests/test_optimizer_dryruns.py
index 10bc6082848..83e08e3088a 100644
--- a/prototype/tests/test_optimizer_dryruns.py
+++ b/prototype/tests/test_optimizer_dryruns.py
@@ -9,7 +9,6 @@ def _test_resources(resources):
         train = sky.Task('train')
         train.set_resources({resources})
     sky.launch(dag, dryrun=True)
-    assert True
 
 
 def test_resources_aws():
@@ -50,3 +49,30 @@ def test_partial_v100():
     _test_resources(
         sky.Resources(clouds.AWS(), accelerators='V100', use_spot=True))
     _test_resources(sky.Resources(clouds.AWS(), accelerators={'V100': 8}))
+
+
+def test_instance_type_mistmatches_accelerators():
+    bad_instance_and_accs = [
+        # Actual: V100
+        ('p3.2xlarge', 'K80'),
+        # Actual: None
+        ('m4.2xlarge', 'V100'),
+    ]
+    for instance, acc in bad_instance_and_accs:
+        with pytest.raises(ValueError) as e:
+            _test_resources(
+                sky.Resources(clouds.AWS(),
+                              instance_type=instance,
+                              accelerators=acc))
+        assert 'Infeasible resource demands found' in str(e.value)
+
+
+def test_instance_type_matches_accelerators():
+    _test_resources(
+        sky.Resources(clouds.AWS(),
+                      instance_type='p3.2xlarge',
+                      accelerators='V100'))
+    _test_resources(
+        sky.Resources(clouds.GCP(),
+                      instance_type='n1-standard-2',
+                      accelerators='V100'))


Code review comments:
diff_hunk: @@ -77,6 +79,29 @@ def __repr__(self) -> str:
     def is_launchable(self) -> bool:
         return self.cloud is not None and self.instance_type is not None
 
+    def _clear_redundant_accelerators_field(self) -> None:
+        """Try-clears any specified accelerators field for non-GCP."""
+        # GCP must have 'self.accelerators' because it attaches accelerators to
+        # regular VMs.  For other clouds, this field may be
+        # reduandant/inconsistent with the specified instance_type.
+        if self.is_launchable() and not isinstance(self.cloud, clouds.GCP):
+            acc_requested = self.accelerators
+            if acc_requested is None:
+                return
+            acc_from_instance_type = (
+                self.cloud.get_accelerators_from_instance_type(
+                    self.instance_type))
+            if acc_from_instance_type != acc_requested:
path: prototype/sky/resources.py
text: @user1:
Here is a case where the user needs to set different accelerators to instance_type. The task (as shown below) must be run on `p2.16xlarge` due to the peak memory limitation, but the user does not want to reserve all the GPUs, so that she can submit other small jobs to that cluster as well. 
```yaml
# task.yaml
resources:
  cloud: aws
  instance_type: p3.16xlarge
  accelerators:
    V100: 1
setup: ...
run: ...
```

One way is to force the user to have an additional `cluster.yaml` to have the correct instance_type vs accelerators, but an additional file may be bothering? Or we can only check the `acc_requested.keys()[0] ==  self.accelerators.keys()[0]` and `acc_requested.values()[0] <= self.accelerators.values()[0]`.

@author:
Great catch!  I think part of why this is hard to spot is because the same `resources:` field is used for both the cluster shape, and the task demand.  

Pushed the second fix that you proposed and verified this yaml can launch (and that the codegen'd Ray program has the correct `{V100:1}` demand).