def METHOD_NAME(
        self,
        task: 'task_lib.Task',
        to_provision: Optional['resources.Resources'],
        dryrun: bool,
        stream_logs: bool,
        cluster_name: Optional[str] = None,
        retry_until_up: bool = False) -> Optional[_ResourceHandleType]:
    if cluster_name is None:
        cluster_name = sky.backends.backend_utils.generate_cluster_name()
    usage_lib.record_cluster_name_for_current_operation(cluster_name)
    usage_lib.messages.usage.update_actual_task(task)
    return self._provision(task, to_provision, dryrun, stream_logs,
                           cluster_name, retry_until_up)