Skip to content

query_workers

CLI module for querying active Merlin task server workers.

This module defines the QueryWorkersCommand class, which implements the query-workers subcommand for the Merlin CLI. The command allows users to inspect the state of connected workers on a task server (e.g., Celery), optionally filtering by queues or worker names.

QueryWorkersCommand

Bases: CommandEntryPoint

Handles query-workers CLI command for querying information about Merlin workers.

Methods:

Name Description
add_parser

Adds the query-workers command to the CLI parser.
process_command

Processes the CLI input and dispatches the appropriate action.
Source code in merlin/cli/commands/query_workers.py 
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
class QueryWorkersCommand(CommandEntryPoint):
    """
    Handles `query-workers` CLI command for querying information about Merlin workers.

    Methods:
        add_parser: Adds the `query-workers` command to the CLI parser.
        process_command: Processes the CLI input and dispatches the appropriate action.
    """

    def add_parser(self, subparsers: ArgumentParser):
        """
        Add the `query-workers` command parser to the CLI argument parser.

        Parameters:
            subparsers (ArgumentParser): The subparsers object to which the `query-workers` command parser will be added.
        """
        query: ArgumentParser = subparsers.add_parser("query-workers", help="List connected task server workers.")
        query.set_defaults(func=self.process_command)
        query.add_argument(
            "--task_server",
            type=str,
            default="celery",
            help="Task server type from which to query workers.\
                                Default: %(default)s",
        )
        query.add_argument(
            "--spec",
            type=str,
            default=None,
            help="Path to a Merlin YAML spec file from which to read worker names to query.",
        )
        query.add_argument("--queues", type=str, default=None, nargs="+", help="Specific queues to query workers from.")
        query.add_argument(
            "--workers",
            type=str,
            action="store",
            nargs="+",
            default=None,
            help="Specific logical worker names to query.",
        )
        format_default = "rich"
        query.add_argument(
            "-f",
            "--format",
            choices=worker_formatter_factory.list_available(),
            default=format_default,
            help=f"Output format. Default: {format_default}",
        )
        query.add_argument(
            "-l",
            "--local-db",
            action="store_true",
            help="Use the local Merlin database for querying workers.",
        )

    def process_command(self, args: Namespace):
        """
        CLI command for finding all workers.

        This function retrieves and queries the names of any active workers.
        If the `--spec` argument is included, only query the workers defined in the spec file.

        Args:
            args: Parsed command-line arguments, which may include:\n
                - `spec`: Path to the specification file.
                - `task_server`: Address of the task server to query.
                - `queues`: List of queue names to filter workers.
                - `workers`: List of specific worker names to query.
        """
        print(banner_small)

        if args.local_db:
            initialize_config(local_mode=True)

        worker_names = []
        if args.workers:
            worker_names.extend(args.workers)

        # Get the workers from the spec file if --spec provided
        spec = None
        if args.spec:
            spec_path = verify_filepath(args.spec)
            spec = MerlinSpec.load_specification(spec_path)
            worker_names.extend(spec.get_worker_names())
            for worker_name in worker_names:
                if "$" in worker_name:
                    LOG.warning(f"Worker '{worker_name}' is unexpanded. Target provenance spec instead?")
            LOG.debug(f"Searching for the following workers to stop based on the spec {args.spec}: {worker_names}")

        task_server = spec.merlin["resources"]["task_server"] if spec else args.task_server
        worker_handler = worker_handler_factory.create(task_server)
        worker_handler.query_workers(args.format, queues=args.queues, workers=worker_names, local_db=args.local_db)

add_parser(subparsers)

Add the query-workers command parser to the CLI argument parser.

Parameters:

Name Type Description Default
subparsers ArgumentParser

The subparsers object to which the query-workers command parser will be added.

 required
Source code in merlin/cli/commands/query_workers.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def add_parser(self, subparsers: ArgumentParser):
    """
    Add the `query-workers` command parser to the CLI argument parser.

    Parameters:
        subparsers (ArgumentParser): The subparsers object to which the `query-workers` command parser will be added.
    """
    query: ArgumentParser = subparsers.add_parser("query-workers", help="List connected task server workers.")
    query.set_defaults(func=self.process_command)
    query.add_argument(
        "--task_server",
        type=str,
        default="celery",
        help="Task server type from which to query workers.\
                            Default: %(default)s",
    )
    query.add_argument(
        "--spec",
        type=str,
        default=None,
        help="Path to a Merlin YAML spec file from which to read worker names to query.",
    )
    query.add_argument("--queues", type=str, default=None, nargs="+", help="Specific queues to query workers from.")
    query.add_argument(
        "--workers",
        type=str,
        action="store",
        nargs="+",
        default=None,
        help="Specific logical worker names to query.",
    )
    format_default = "rich"
    query.add_argument(
        "-f",
        "--format",
        choices=worker_formatter_factory.list_available(),
        default=format_default,
        help=f"Output format. Default: {format_default}",
    )
    query.add_argument(
        "-l",
        "--local-db",
        action="store_true",
        help="Use the local Merlin database for querying workers.",
    )

process_command(args)

CLI command for finding all workers.

This function retrieves and queries the names of any active workers. If the --spec argument is included, only query the workers defined in the spec file.

Parameters:

Name Type Description Default
args Namespace

Parsed command-line arguments, which may include:

  • spec: Path to the specification file.
  • task_server: Address of the task server to query.
  • queues: List of queue names to filter workers.
  • workers: List of specific worker names to query.
 required
Source code in merlin/cli/commands/query_workers.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
def process_command(self, args: Namespace):
    """
    CLI command for finding all workers.

    This function retrieves and queries the names of any active workers.
    If the `--spec` argument is included, only query the workers defined in the spec file.

    Args:
        args: Parsed command-line arguments, which may include:\n
            - `spec`: Path to the specification file.
            - `task_server`: Address of the task server to query.
            - `queues`: List of queue names to filter workers.
            - `workers`: List of specific worker names to query.
    """
    print(banner_small)

    if args.local_db:
        initialize_config(local_mode=True)

    worker_names = []
    if args.workers:
        worker_names.extend(args.workers)

    # Get the workers from the spec file if --spec provided
    spec = None
    if args.spec:
        spec_path = verify_filepath(args.spec)
        spec = MerlinSpec.load_specification(spec_path)
        worker_names.extend(spec.get_worker_names())
        for worker_name in worker_names:
            if "$" in worker_name:
                LOG.warning(f"Worker '{worker_name}' is unexpanded. Target provenance spec instead?")
        LOG.debug(f"Searching for the following workers to stop based on the spec {args.spec}: {worker_names}")

    task_server = spec.merlin["resources"]["task_server"] if spec else args.task_server
    worker_handler = worker_handler_factory.create(task_server)
    worker_handler.query_workers(args.format, queues=args.queues, workers=worker_names, local_db=args.local_db)