Repository: PrefectHQ/prefect
Pull request: Warn when work queues paused when starting a worker or agent
## Examples
### Workers
#### Starting a worker normally when at least one work queue is unpaused does not warn
```bash
❯ prefect worker start -p local-pool
Discovered type 'process' for work pool 'local-pool'.
Worker 'ProcessWorker 8fdf8693-a018-4634-a677-73f6f11e7678' started!
```
#### Starting a worker when all work queues are paused warns
```bash
❯ prefect worker start -p local-pool
All work queues in the work pool 'local-pool' are currently paused. This worker will not execute any flow runs until the work queues are unpaused.
Discovered type 'process' for work pool 'local-pool'.
Worker 'ProcessWorker 3fe94382-bd3e-48ce-9fd4-c0af6e9113cd' started!
```
#### Starting a worker with a specified work queue that is paused warns
```bash
❯ prefect worker start -p local-pool -q high-priority
Specified work queue(s) in the work pool 'local-pool' are currently paused. This worker will not execute any flow runs until the work queues are unpaused.
Discovered type 'process' for work pool 'local-pool'.
Worker 'ProcessWorker 41417088-1c14-4eca-92a5-158c9bbedbab' started!
```
#### Starting a worker with specified work queues where one is paused does not warn
```bash
❯ prefect worker start -p local-pool -q high-priority -q low-priority
Discovered type 'process' for work pool 'local-pool'.
Worker 'ProcessWorker 1a9981ca-1ae3-46e4-9961-5331b2555873' started!
```
------
### Agents
#### Starting an agent when `default` work queue is paused warns
```bash
❯ prefect agent start -p default-agent-pool
WARNING: The 'agent' command group has been deprecated. It will not be available after Sep 2024. Use `prefect worker start` instead. Refer to the upgrade guide for more information: https://docs.prefect.io/latest/guides/upgrade-guide-agents-to-workers/.
The 'default' work queue in the work pool 'default-agent-pool' is currently paused. This agent will not execute any flow runs until the work queue(s) are unpaused.
Starting v2.18.1+32.g79d96c678d agent connected to https://api.prefect.cloud/api/accounts/43a83e5a-b23e-47cf-9652-5c3dc511f7e1/workspaces/75241259-81a4-4879-92aa-fc3f1df3e6e4...

  ___ ___ ___ ___ ___ ___ _____     _   ___ ___ _  _ _____
 | _ \ _ \ __| __| __/ __|_   _|   /_\ / __| __| \| |_   _|
 |  _/   / _|| _|| _| (__  | |    / _ \ (_ | _|| .` | | |
 |_| |_|_\___|_| |___\___| |_|   /_/ \_\___|___|_|\_| |_|


Agent started! Looking for work from work pool 'default-agent-pool'...
```
#### Starting an agent when `default` work queue is paused but we've specified additional work queue to use does not warn
```bash
❯ prefect agent start -p default-agent-pool -q other-work-queue -q default
WARNING: The 'agent' command group has been deprecated. It will not be available after Sep 2024. Use `prefect worker start` instead. Refer to the upgrade guide for more information: https://docs.prefect.io/latest/guides/upgrade-guide-agents-to-workers/.
Starting v2.18.1+32.g79d96c678d agent connected to https://api.prefect.cloud/api/accounts/43a83e5a-b23e-47cf-9652-5c3dc511f7e1/workspaces/75241259-81a4-4879-92aa-fc3f1df3e6e4...

  ___ ___ ___ ___ ___ ___ _____     _   ___ ___ _  _ _____
 | _ \ _ \ __| __| __/ __|_   _|   /_\ / __| __| \| |_   _|
 |  _/   / _|| _|| _| (__  | |    / _ \ (_ | _|| .` | | |
 |_| |_|_\___|_| |___\___| |_|   /_/ \_\___|___|_|\_| |_|


Agent started! Looking for work from work pool 'default-agent-pool'...
```


Problem statement:
Add a warning message when the default work queue is paused
### First check

- [X] I added a descriptive title to this issue.
- [X] I used the GitHub search to find a similar request and didn't find it.
- [X] I searched the Prefect documentation for this feature.

### Prefect Version

2.x

### Describe the current behavior

If I pause my default work queue and the try to run an agent, there's no warning that the work queue is paused.  I just spent a while debugging the issue before looking in the UI and realized the queue was paused. 

### Describe the proposed behavior

If I run prefect agent start -p testt and the default work queue in the testt work pool is paused. I get an warning message saying the default work queue is paused.  (We had this for work queues and it was very helpful). 

### Example Use

I create a work pool in the UI (e.g. name `testt`) and pause my default work queue.  Later I start my (local) agent using the cli (`prefect agent start -p testt`) and try to kick off a run from a deployment.  Because the work queue is paused, no runs are picked up but I do get a message in the CLI so I know to go unpause my work queue. 

### Additional context

_No response_

Hints from issue discussion:
@zhen0 

> I get an error message saying the default work queue is paused.

A warning or an error? Seems like the agent should probably continue to run.
@billpalombi could you triage this?
Hmm, this is tricky. The agent (and workers) will successfully find the work pool, but I don't think it has any way of knowing whether that work pool or its queues are paused or not. Lets keep this issue open to reflect the UX issue, but circle back to it once we've rolled out the forthcoming agent enhancements (workers).
This issue is stale because it has been open 30 days with no activity. To keep this issue open remove stale label or comment.
This issue was closed because it has been stale for 14 days with no activity. If this issue is important or you have more to add feel free to re-open it.

Commit message: update docstring

Patch to review:
diff --git a/src/prefect/cli/agent.py b/src/prefect/cli/agent.py
index 7f7022b428bb..e5061a845d0c 100644
--- a/src/prefect/cli/agent.py
+++ b/src/prefect/cli/agent.py
@@ -4,7 +4,7 @@
 
 import os
 from functools import partial
-from typing import List
+from typing import List, Optional
 from uuid import UUID
 
 import anyio
@@ -16,6 +16,7 @@
 from prefect.cli._utilities import exit_with_error
 from prefect.cli.root import app
 from prefect.client import get_client
+from prefect.client.schemas.filters import WorkQueueFilter, WorkQueueFilterName
 from prefect.exceptions import ObjectNotFound
 from prefect.settings import (
     PREFECT_AGENT_PREFETCH_SECONDS,
@@ -121,6 +122,25 @@ async def start(
             ),
             style="blue",
         )
+    if work_pool_name:
+        is_queues_paused = await _check_work_queues_paused(
+            work_pool_name=work_pool_name,
+            work_queues=work_queues,
+        )
+        if is_queues_paused:
+            queue_scope = (
+                "The 'default' work queue"
+                if not work_queues
+                else "Specified work queue(s)"
+            )
+            app.console.print(
+                (
+                    f"{queue_scope} in the work pool {work_pool_name!r} is currently"
+                    " paused. This agent will not execute any flow runs until the work"
+                    " queue(s) are unpaused."
+                ),
+                style="yellow",
+            )
 
     if not work_queues and not tags and not work_queue_prefix and not work_pool_name:
         exit_with_error("No work queues provided!", style="red")
@@ -225,3 +245,31 @@ async def start(
             )
 
     app.console.print("Agent stopped!")
+
+
+async def _check_work_queues_paused(
+    work_pool_name: str, work_queues: Optional[List[str]]
+) -> bool:
+    """
+    Check if the default work queue in the work pool is paused. If work queues are specified,
+    only those work queues are checked.
+
+    Args:
+        - work_pool_name (str): the name of the work pool to check
+        - work_queues (Optional[List[str]]): the names of the work queues to check
+
+    Returns:
+        - bool: True if work queues are paused, False otherwise
+    """
+    work_queues_list = work_queues or ["default"]
+    try:
+        work_queues_filter = WorkQueueFilter(
+            name=WorkQueueFilterName(any_=work_queues_list)
+        )
+        async with get_client() as client:
+            wqs = await client.read_work_queues(
+                work_pool_name=work_pool_name, work_queue_filter=work_queues_filter
+            )
+            return all(queue.is_paused for queue in wqs) if wqs else False
+    except ObjectNotFound:
+        return False
diff --git a/src/prefect/cli/worker.py b/src/prefect/cli/worker.py
index e260794c61aa..6f1f1ca5445e 100644
--- a/src/prefect/cli/worker.py
+++ b/src/prefect/cli/worker.py
@@ -14,6 +14,7 @@
 from prefect.cli.root import app, is_interactive
 from prefect.client.collections import get_collections_metadata_client
 from prefect.client.orchestration import get_client
+from prefect.client.schemas.filters import WorkQueueFilter, WorkQueueFilterName
 from prefect.exceptions import ObjectNotFound
 from prefect.plugins import load_prefect_collections
 from prefect.settings import (
@@ -126,6 +127,23 @@ async def start(
             style="yellow",
         )
 
+    is_queues_paused = await _check_work_queues_paused(
+        work_pool_name,
+        work_queues,
+    )
+    if is_queues_paused:
+        queue_scope = (
+            "All work queues" if not work_queues else "Specified work queue(s)"
+        )
+        app.console.print(
+            (
+                f"{queue_scope} in the work pool {work_pool_name!r} are currently"
+                " paused. This worker will not execute any flow runs until the work"
+                " queues are unpaused."
+            ),
+            style="yellow",
+        )
+
     worker_cls = await _get_worker_class(worker_type, work_pool_name, install_policy)
 
     if worker_cls is None:
@@ -222,6 +240,35 @@ async def _check_work_pool_paused(work_pool_name: str) -> bool:
         return False
 
 
+async def _check_work_queues_paused(
+    work_pool_name: str, work_queues: Optional[List[str]]
+) -> bool:
+    """
+    Check if all work queues in the work pool are paused. If work queues are specified,
+    only those work queues are checked.
+
+    Args:
+        - work_pool_name (str): the name of the work pool to check
+        - work_queues (Optional[List[str]]): the names of the work queues to check
+
+    Returns:
+        - bool: True if work queues are paused, False otherwise
+    """
+    try:
+        work_queues_filter = (
+            WorkQueueFilter(name=WorkQueueFilterName(any_=work_queues))
+            if work_queues
+            else []
+        )
+        async with get_client() as client:
+            wqs = await client.read_work_queues(
+                work_pool_name=work_pool_name, work_queue_filter=work_queues_filter
+            )
+            return all(queue.is_paused for queue in wqs) if wqs else False
+    except ObjectNotFound:
+        return False
+
+
 async def _retrieve_worker_type_from_pool(work_pool_name: Optional[str] = None) -> str:
     try:
         async with get_client() as client:
diff --git a/tests/cli/test_worker.py b/tests/cli/test_worker.py
index f10e88d27004..6ed3e48b071a 100644
--- a/tests/cli/test_worker.py
+++ b/tests/cli/test_worker.py
@@ -212,6 +212,90 @@ def test_start_worker_with_work_queue_names(monkeypatch, process_work_pool):
     )
 
 
+@pytest.mark.usefixtures("use_hosted_api_server")
+def test_start_worker_with_specified_work_queues_paused(monkeypatch, process_work_pool):
+    mock_worker = MagicMock()
+    monkeypatch.setattr(prefect.cli.worker, "lookup_type", lambda x, y: mock_worker)
+
+    invoke_and_assert(
+        command=[
+            "work-queue",
+            "pause",
+            "default",
+            "--pool",
+            process_work_pool.name,
+        ],
+        expected_code=0,
+        expected_output_contains=[
+            f"Work queue 'default' in work pool {process_work_pool.name!r} paused"
+        ],
+    )
+
+    invoke_and_assert(
+        command=[
+            "worker",
+            "start",
+            "-p",
+            process_work_pool.name,
+            "--work-queue",
+            "default",
+            "--run-once",
+        ],
+        expected_code=0,
+        expected_output_contains=[
+            f"Specified work queue(s) in the work pool {process_work_pool.name!r} are currently paused.",
+        ],
+    )
+
+    mock_worker.assert_called_once_with(
+        name=None,
+        work_pool_name=process_work_pool.name,
+        work_queues=["default"],
+        prefetch_seconds=ANY,
+        limit=None,
+        heartbeat_interval_seconds=30,
+        base_job_template=None,
+    )
+
+
+@pytest.mark.usefixtures("use_hosted_api_server")
+def test_start_worker_with_all_work_queues_paused(monkeypatch, process_work_pool):
+    mock_worker = MagicMock()
+    monkeypatch.setattr(prefect.cli.worker, "lookup_type", lambda x, y: mock_worker)
+
+    invoke_and_assert(
+        command=[
+            "work-queue",
+            "pause",
+            "default",
+            "--pool",
+            process_work_pool.name,
+        ],
+        expected_code=0,
+        expected_output_contains=[
+            f"Work queue 'default' in work pool {process_work_pool.name!r} paused"
+        ],
+    )
+
+    invoke_and_assert(
+        command=["worker", "start", "-p", process_work_pool.name, "--run-once"],
+        expected_code=0,
+        expected_output_contains=[
+            f"All work queues in the work pool {process_work_pool.name!r} are currently paused.",
+        ],
+    )
+
+    mock_worker.assert_called_once_with(
+        name=None,
+        work_pool_name=process_work_pool.name,
+        work_queues=None,
+        prefetch_seconds=ANY,
+        limit=None,
+        heartbeat_interval_seconds=30,
+        base_job_template=None,
+    )
+
+
 @pytest.mark.usefixtures("use_hosted_api_server")
 def test_start_worker_with_prefetch_seconds(monkeypatch):
     mock_worker = MagicMock()


Code review comments:
diff_hunk: @@ -222,6 +240,35 @@ async def _check_work_pool_paused(work_pool_name: str) -> bool:
         return False
 
 
+async def _check_work_queues_paused(
+    work_pool_name: str, work_queues: Optional[List[str]]
+) -> bool:
+    """
+    Check if all work queues in the work pool are paused. If work queues are specified,
+    only those work queues are checked.
+
+    Args:
+        - work_pool_name (str): the name of the work pool to check
+        - work_queues (Optional[List[str]]): the names of the work queues to check
+
+    Returns:
+        - bool: True if work queues are paused, False otherwise
+    """
+    try:
+        work_queues_filter = (
+            WorkQueueFilter(name=WorkQueueFilterName(any_=work_queues))
+            if work_queues
+            else []
path: src/prefect/cli/worker.py
text: @user1:
hmm should this be `None`?
```suggestion
            else None
```

@author:
Ah yeah, good call, I was thinking the `else []` was for `any_` but yeah, it's for `work_queues_filter` so that makes sense.