Github Action for GlobusComputeExecutor (#3619)

* Support for testing GlobusComputeExecutor in a github action
* Adding shared_fs and staging_required tags to tests
* Adding GlobusComputeExecutor test config

.github/workflows/gce_test.yaml

@@ -0,0 +1,103 @@
+name: GlobusComputeExecutor tests
+
+on:
+  pull_request:
+    types:
+      - opened
+      - synchronize
+
+jobs:
+  main-test-suite:
+    strategy:
+      matrix:
+        python-version: ["3.11"]
+    runs-on: ubuntu-20.04
+    timeout-minutes: 60
+
+    steps:
+    - uses: actions/checkout@master
+
+    - name: Set up Python ${{ matrix.python-version }}
+      uses: actions/setup-python@v4
+      with:
+        python-version: ${{ matrix.python-version }}
+
+    - name: Collect Job Information
+      id: job-info
+      run: |
+        echo "Python Version: ${{ matrix.python-version }}" >> ci_job_info.txt
+        echo "CI Triggering Event: ${{ github.event_name }}" >> ci_job_info.txt
+        echo "Triggering Git Ref: ${{ github.ref }}" >> ci_job_info.txt
+        echo "Triggering Git SHA: ${{ github.sha }}" >> ci_job_info.txt
+        echo "Workflow Run: ${{ github.run_number }}" >> ci_job_info.txt
+        echo "Workflow Attempt: ${{ github.run_attempt }}" >> ci_job_info.txt
+        as_ascii="$(echo "${{ github.ref_name }}" | perl -pe "s/[^A-z0-9-]+/-/g; s/^-+|-+\$//g; s/--+/-/g;")"
+        echo "as-ascii=$as_ascii" >> $GITHUB_OUTPUT
+
+    - name: Non-requirements based install
+      run: |
+        # libpython3.5: make workqueue binary installer happy
+        # mpich: required by radical executor
+        sudo apt-get update -q
+        sudo apt-get install -qy libpython3.5 mpich
+
+    - name: setup virtual env
+      run: |
+        make virtualenv
+        source .venv/bin/activate
+
+    - name: make deps clean_coverage
+      run: |
+        source .venv/bin/activate
+        make deps
+        make clean_coverage
+        
+        # Temporary fix, until changes make it into compute releases
+        git clone -b configure_tasks_working_dir https://github.com/globus/globus-compute.git        
+        pip3 install globus-compute/compute_sdk globus-compute/compute_endpoint
+
+    - name: start globus_compute_endpoint
+      env:
+        GLOBUS_COMPUTE_CLIENT_ID: ${{ secrets.GLOBUS_COMPUTE_CLIENT_ID }}
+        GLOBUS_COMPUTE_CLIENT_SECRET: ${{ secrets.GLOBUS_COMPUTE_SECRET_KEY }}
+      run: |
+        source /home/runner/work/parsl/parsl/.venv/bin/activate
+        globus-compute-endpoint configure default
+        which globus-compute-endpoint
+        python3 -c "import globus_compute_sdk; print(globus_compute_sdk.__version__)"
+        python3 -c "import globus_compute_endpoint; print(globus_compute_endpoint.__version__)"
+        cat << EOF > /home/runner/.globus_compute/default/config.yaml
+        engine:
+          type: ThreadPoolEngine
+          max_workers: 4
+          working_dir: /home/runner/.globus_compute/default/tasks_working_dir
+        EOF
+        cat /home/runner/.globus_compute/default/config.yaml
+        mkdir ~/.globus_compute/default/tasks_working_dir
+        globus-compute-endpoint start default
+        globus-compute-endpoint list
+    - name: make test
+      env:
+        GLOBUS_COMPUTE_CLIENT_ID: ${{ secrets.GLOBUS_COMPUTE_CLIENT_ID }}
+        GLOBUS_COMPUTE_CLIENT_SECRET: ${{ secrets.GLOBUS_COMPUTE_SECRET_KEY }}
+      run: |
+        source .venv/bin/activate
+        export GLOBUS_COMPUTE_ENDPOINT=$(globus-compute-endpoint list | grep default | cut -c 3-38)      
+        echo "GLOBUS_COMPUTE_ENDPOINT = $GLOBUS_COMPUTE_ENDPOINT"
+
+        # temporary; until test-matrixification
+        export PARSL_TEST_PRESERVE_NUM_RUNS=7
+
+        make gce_test
+        ln -s .pytest/parsltest-current test_runinfo
+
+    - name: Archive runinfo logs
+      if: ${{ always() }}
+      uses: actions/upload-artifact@v4
+      with:
+        name: runinfo-${{ matrix.python-version }}-${{ steps.job-info.outputs.as-ascii }}-${{ github.sha }}
+        path: |
+          runinfo/
+          .pytest/
+          ci_job_info.txt
+        compression-level: 9

Makefile

@@ -51,6 +51,10 @@ clean_coverage:
 mypy: ## run mypy checks
 	MYPYPATH=$(CWD)/mypy-stubs mypy parsl/
 
+.PHONY: gce_test
+gce_test: ## Run tests with GlobusComputeExecutor
+	pytest -v -k "not shared_fs and not issue_3620 and not staging_required" --config parsl/tests/configs/globus_compute.py parsl/tests/ --random-order --durations 10
+
 .PHONY: local_thread_test
 local_thread_test: ## run all tests with local_thread config
 	pytest parsl/tests/ -k "not cleannet" --config parsl/tests/configs/local_threads.py --random-order --durations 10

docs/reference.rst

@@ -77,6 +77,7 @@ Executors
     parsl.executors.taskvine.TaskVineExecutor
     parsl.executors.FluxExecutor
     parsl.executors.radical.RadicalPilotExecutor
+    parsl.executors.globus_compute.GlobusComputeExecutor
 
 Manager Selectors
 =================

parsl/executors/globus_compute.py

@@ -2,16 +2,13 @@
 from concurrent.futures import Future
 from typing import Any, Callable, Dict, Optional, Union
 
-import typeguard
-
 from parsl.errors import OptionalModuleMissing
 from parsl.executors.base import ParslExecutor
 from parsl.utils import RepresentationMixin
 
 UUID_LIKE_T = Union[uuid.UUID, str]
 
 
-
 class GlobusComputeExecutor(ParslExecutor, RepresentationMixin):
     """ GlobusComputeExecutor enables remote execution on Globus Compute endpoints
 
@@ -25,15 +22,14 @@ def __init__(
             self,
             endpoint_id: Optional[UUID_LIKE_T] = None,
             task_group_id: Optional[UUID_LIKE_T] = None,
-            resource_specification: Optional[dict[str, Any]] = None,
-            user_endpoint_config: Optional[dict[str, Any]] = None,
+            resource_specification: Optional[Dict[str, Any]] = None,
+            user_endpoint_config: Optional[Dict[str, Any]] = None,
             label: str = "GlobusComputeExecutor",
             batch_size: int = 128,
             amqp_port: Optional[int] = None,
             **kwargs,
-        ):
+    ):
         """
-
         Parameters
         ----------
 
@@ -141,5 +137,3 @@ def shutdown(self, wait=True, *, cancel_futures=False):
             Tasks cannot be cancelled once they are registered.
         """
         return self._executor.shutdown()
-
-

parsl/tests/configs/globus_compute.py

@@ -0,0 +1,18 @@
+import os
+
+from parsl.config import Config
+from parsl.executors import GlobusComputeExecutor
+
+
+def fresh_config():
+
+    endpoint_id = os.environ["GLOBUS_COMPUTE_ENDPOINT"]
+
+    return Config(
+        executors=[
+            GlobusComputeExecutor(
+                label="globus_compute",
+                endpoint_id=endpoint_id
+            )
+        ]
+    )

parsl/tests/conftest.py

@@ -167,6 +167,10 @@ def pytest_configure(config):
         'markers',
         'shared_fs: Marks tests that require a shared_fs between the workers are the test client'
     )
+    config.addinivalue_line(
+        'markers',
+        'issue_3620: Marks tests that do not work correctly on GlobusComputeExecutor (ref: issue 3620)'
+    )
 
 
 @pytest.fixture(autouse=True, scope='session')

parsl/tests/test_bash_apps/test_stdout.py

@@ -34,6 +34,7 @@ def echo_to_streams(msg, stderr=None, stdout=None):
 ]
 
 
+@pytest.mark.shared_fs
 @pytest.mark.parametrize('spec', speclist, ids=testids)
 def test_bad_stdout_specs(spec):
     """Testing bad stdout spec cases"""

parsl/tests/test_docs/test_workflow1.py

@@ -22,6 +22,7 @@ def save(message, outputs=[]):
     return 'echo {m} &> {o}'.format(m=message, o=outputs[0])
 
 
+@pytest.mark.shared_fs
 @pytest.mark.staging_required
 def test_procedural(N=2):
     """Procedural workflow example from docs on

parsl/tests/test_error_handling/test_resource_spec.py

@@ -1,3 +1,5 @@
+import pytest
+
 import parsl
 from parsl.app.app import python_app
 from parsl.executors import WorkQueueExecutor
@@ -11,6 +13,7 @@ def double(x, parsl_resource_specification={}):
     return x * 2
 
 
+@pytest.mark.issue_3620
 def test_resource(n=2):
     executors = parsl.dfk().executors
     executor = None

parsl/tests/test_staging/test_docs_1.py

@@ -11,7 +11,6 @@ def convert(inputs=[], outputs=[]):
             out.write(content.upper())
 
 
-@pytest.mark.cleannet
 @pytest.mark.staging_required
 def test():
     # create an remote Parsl file

test-requirements.txt

@@ -12,6 +12,7 @@ types-python-dateutil
 types-requests
 types-paramiko
 mpi4py
+globus-compute-sdk>=2.27.1
 
 # sqlalchemy is needed for typechecking, so it's here
 # as well as at runtime for optional monitoring execution