Testing/manager

.github/workflows/push-pr_workflow.yml

@@ -10,7 +10,7 @@ jobs:
 
     steps:
     - name: Checkout code
-      uses: actions/checkout@v2
+      uses: actions/checkout@v4
       with:
         fetch-depth: 0 # Checkout the whole history, in case the target is way far behind
 
@@ -40,14 +40,14 @@ jobs:
       MAX_COMPLEXITY: 15
 
     steps:
-    - uses: actions/checkout@v2
+    - uses: actions/checkout@v4
     - name: Set up Python
-      uses: actions/setup-python@v2
+      uses: actions/setup-python@v5
       with:
         python-version: '3.x'
 
     - name: Check cache
-      uses: actions/cache@v2
+      uses: actions/cache@v4
       with:
         path: ~/.cache/pip
         key: ${{ hashFiles('requirements/release.txt') }}-${{ hashFiles('requirements/dev.txt') }}
@@ -95,14 +95,14 @@ jobs:
         python-version: ['3.7', '3.8', '3.9', '3.10', '3.11']
 
     steps:
-    - uses: actions/checkout@v2
+    - uses: actions/checkout@v4
     - name: Set up Python ${{ matrix.python-version }}
-      uses: actions/setup-python@v2
+      uses: actions/setup-python@v5
       with:
         python-version: ${{ matrix.python-version }}
 
     - name: Check cache
-      uses: actions/cache@v2
+      uses: actions/cache@v4
       with:
         path: ${{ env.pythonLocation }}
         key: ${{ env.pythonLocation }}-${{ hashFiles('requirements/release.txt') }}-${{ hashFiles('requirements/dev.txt') }}
@@ -112,8 +112,7 @@ jobs:
         python3 -m pip install --upgrade pip
         if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
         pip3 install -r requirements/dev.txt
-        pip freeze
-
+
     - name: Install singularity
       run: |
         sudo apt-get update && sudo apt-get install -y \
@@ -153,6 +152,73 @@ jobs:
       run: |
         python3 tests/integration/run_tests.py --verbose --local
 
+  Integration-tests:
+    runs-on: ubuntu-latest
+    env: 
+      GO_VERSION: 1.18.1
+      SINGULARITY_VERSION: 3.9.9
+      OS: linux
+      ARCH: amd64
+
+    strategy:
+      matrix:
+        python-version: ['3.7', '3.8', '3.9', '3.10', '3.11']
+
+    steps:
+    - uses: actions/checkout@v4
+    - name: Set up Python ${{ matrix.python-version }}
+      uses: actions/setup-python@v5
+      with:
+        python-version: ${{ matrix.python-version }}
+
+    - name: Check cache
+      uses: actions/cache@v4
+      with:
+        path: ${{ env.pythonLocation }}
+        key: ${{ env.pythonLocation }}-${{ hashFiles('requirements/release.txt') }}-${{ hashFiles('requirements/dev.txt') }}
+
+    - name: Install dependencies
+      run: |
+        python3 -m pip install --upgrade pip
+        if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
+        pip3 install -r requirements/dev.txt
+
+    - name: Install merlin
+      run: |
+        pip3 install -e .
+        merlin config
+
+    - name: Install singularity
+      run: |
+        sudo apt-get update && sudo apt-get install -y \
+          build-essential \
+          libssl-dev \
+          uuid-dev \
+          libgpgme11-dev \
+          squashfs-tools \
+          libseccomp-dev \
+          pkg-config
+        wget https://go.dev/dl/go$GO_VERSION.$OS-$ARCH.tar.gz
+        sudo tar -C /usr/local -xzf go$GO_VERSION.$OS-$ARCH.tar.gz
+        rm go$GO_VERSION.$OS-$ARCH.tar.gz
+        export PATH=$PATH:/usr/local/go/bin
+        wget https://github.com/sylabs/singularity/releases/download/v$SINGULARITY_VERSION/singularity-ce-$SINGULARITY_VERSION.tar.gz
+        tar -xzf singularity-ce-$SINGULARITY_VERSION.tar.gz
+        cd singularity-ce-$SINGULARITY_VERSION
+        ./mconfig && \
+          make -C ./builddir && \
+          sudo make -C ./builddir install
+
+    - name: Install CLI task dependencies generated from the 'feature demo' workflow
+      run: |
+        merlin example feature_demo
+        pip3 install -r feature_demo/requirements.txt
+
+    # TODO remove the --ignore statement once those tests are fixed
+    - name: Run integration test suite for distributed tests
+      run: |
+        pytest --ignore tests/integration/test_celeryadapter.py tests/integration/
+
   Distributed-test-suite:
     runs-on: ubuntu-latest
     services:

CHANGELOG.md

@@ -6,6 +6,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 ### Added
+- Merlin manager capability to monitor celery workers.
 - Several new unit tests for the following subdirectories:
   - `merlin/common/`
   - `merlin/config/`
@@ -21,6 +22,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Split the `start_server` and `config_server` functions of `merlin/server/server_commands.py` into multiple functions to make testing easier
 - Split the `create_server_config` function of `merlin/server/server_config.py` into two functions to make testing easier
 - Combined `set_snapshot_seconds` and `set_snapshot_changes` methods of `RedisConfig` into one method `set_snapshot`
+- Moved stop-workers and query-workers integration tests to pytest tests
 
 ## [1.12.2b1]
 ### Added

merlin/celery.py

@@ -114,9 +114,11 @@ def route_for_task(name, args, kwargs, options, task=None, **kw):  # pylint: dis
     BROKER_URI = None
     RESULTS_BACKEND_URI = None
 
+app_name = "merlin_test_app" if os.getenv("CELERY_ENV") == "test" else "merlin"
+
 # initialize app with essential properties
 app: Celery = patch_celery().Celery(
-    "merlin",
+    app_name,
     broker=BROKER_URI,
     backend=RESULTS_BACKEND_URI,
     broker_use_ssl=BROKER_SSL,

merlin/examples/dev_workflows/multiple_workers.yaml

@@ -46,11 +46,11 @@ merlin:
     resources:
         workers:
             step_1_merlin_test_worker:
-                args: -l INFO
+                args: -l INFO --concurrency 1
                 steps: [step_1]
             step_2_merlin_test_worker:
-                args: -l INFO
+                args: -l INFO --concurrency 1
                 steps: [step_2]
             other_merlin_test_worker:
-                args: -l INFO
+                args: -l INFO --concurrency 1
                 steps: [step_3, step_4]

merlin/main.py

@@ -57,6 +57,7 @@
 from merlin.server.server_commands import config_server, init_server, restart_server, start_server, status_server, stop_server
 from merlin.spec.expansion import RESERVED, get_spec_with_expansion
 from merlin.spec.specification import MerlinSpec
+from merlin.study.celerymanageradapter import run_manager, start_manager, stop_manager
 from merlin.study.status import DetailedStatus, Status
 from merlin.study.status_constants import VALID_RETURN_CODES, VALID_STATUS_FILTERS
 from merlin.study.status_renderers import status_renderer_factory
@@ -359,7 +360,7 @@ def stop_workers(args):
                 LOG.warning(f"Worker '{worker_name}' is unexpanded. Target provenance spec instead?")
 
     # Send stop command to router
-    router.stop_workers(args.task_server, worker_names, args.queues, args.workers)
+    router.stop_workers(args.task_server, worker_names, args.queues, args.workers, args.level.upper())
 
 
 def print_info(args):
@@ -400,6 +401,35 @@ def process_example(args: Namespace) -> None:
         setup_example(args.workflow, args.path)
 
 
+def process_manager(args: Namespace):
+    """
+    Process the command for managing the workers.
+
+    This function interprets the command provided in the `args` namespace and
+    executes the corresponding manager function. It supports three commands:
+    "run", "start", and "stop".
+
+    :param args: parsed CLI arguments
+    """
+    if args.command == "run":
+        run_manager(query_frequency=args.query_frequency, query_timeout=args.query_timeout, worker_timeout=args.worker_timeout)
+    elif args.command == "start":
+        try:
+            start_manager(
+                query_frequency=args.query_frequency, query_timeout=args.query_timeout, worker_timeout=args.worker_timeout
+            )
+            LOG.info("Manager started successfully.")
+        except Exception as e:
+            LOG.error(f"Unable to start manager.\n{e}")
+    elif args.command == "stop":
+        if stop_manager():
+            LOG.info("Manager stopped successfully.")
+        else:
+            LOG.error("Unable to stop manager.")
+    else:
+        print("Run manager with a command. Try 'merlin manager -h' for more details")
+
+
 def process_monitor(args):
     """
     CLI command to monitor merlin workers and queues to keep
@@ -898,6 +928,75 @@ def generate_worker_touching_parsers(subparsers: ArgumentParser) -> None:
         help="regex match for specific workers to stop",
     )
 
+    # merlin manager
+    manager: ArgumentParser = subparsers.add_parser(
+        "manager",
+        help="Watchdog application to manage workers",
+        description="A daemon process that helps to restart and communicate with workers while running.",
+        formatter_class=ArgumentDefaultsHelpFormatter,
+    )
+    manager.set_defaults(func=process_manager)
+
+    def add_manager_options(manager_parser: ArgumentParser):
+        """
+        Add shared options for manager subcommands.
+
+        The `manager run` and `manager start` subcommands have the same options.
+        Rather than writing duplicate code for these we'll use this function
+        to add the arguments to these subcommands.
+
+        :param manager_parser: The ArgumentParser object to add these options to
+        """
+        manager_parser.add_argument(
+            "-qf",
+            "--query_frequency",
+            action="store",
+            type=int,
+            default=60,
+            help="The frequency at which workers will be queried for response.",
+        )
+        manager_parser.add_argument(
+            "-qt",
+            "--query_timeout",
+            action="store",
+            type=float,
+            default=0.5,
+            help="The timeout for the query response that are sent to workers.",
+        )
+        manager_parser.add_argument(
+            "-wt",
+            "--worker_timeout",
+            action="store",
+            type=int,
+            default=180,
+            help="The sum total (query_frequency*tries) time before an attempt is made to restart worker.",
+        )
+
+    manager_commands: ArgumentParser = manager.add_subparsers(dest="command")
+    manager_run = manager_commands.add_parser(
+        "run",
+        help="Run the daemon process",
+        description="Run manager",
+        formatter_class=ArgumentDefaultsHelpFormatter,
+    )
+    add_manager_options(manager_run)
+    manager_run.set_defaults(func=process_manager)
+    manager_start = manager_commands.add_parser(
+        "start",
+        help="Start the daemon process",
+        description="Start manager",
+        formatter_class=ArgumentDefaultsHelpFormatter,
+    )
+    add_manager_options(manager_start)
+    manager_start.set_defaults(func=process_manager)
+    manager_stop = manager_commands.add_parser(
+        "stop",
+        help="Stop the daemon process",
+        description="Stop manager",
+        formatter_class=ArgumentDefaultsHelpFormatter,
+    )
+    manager_stop.set_defaults(func=process_manager)
+
     # merlin monitor
     monitor: ArgumentParser = subparsers.add_parser(
         "monitor",