Merge branch 'feat/mymdc_run_check' into 'master'

[webservice] Check run migration status using MyMDC

See merge request detectors/pycalibration!553

tests/test_webservice.py

-import os
 import sys
 from pathlib import Path
 from unittest import mock
-from webservice.messages import MigrationError
 
 import pytest
-from testpath import MockCommand
 
 sys.path.insert(0, Path(__file__).parent / 'webservice')
 import webservice  # noqa: import not at top of file
+from webservice.messages import MigrationError  # noqa: import not at top
 from webservice.webservice import (  # noqa: import not at top of file
     check_files,
     merge,
@@ -79,58 +77,57 @@ def test_parse_config():
 
 @pytest.mark.asyncio
 @pytest.mark.parametrize(
-    "xattr_list, xattr_get, expected_result",
+    "mock_json, expected_result",
     [
-        (["user.status"], b"offline", True),
-        (["user.status"], b"tape", True),
-        (["user.status"], b"dCache", True),
+        ({'runs': [{'repositories': {'DESY_DCACHE_RAW_CC': 'whatever_its_a_dict'}}]},  # noqa
+         ['DESY_DCACHE_RAW_CC']),
+
+        ({'runs': [{'repositories': {'XFEL_GPFS_OFFLINE_RAW_CC': 'whatever_its_a_dict'}}]},  # noqa
+         ['XFEL_GPFS_OFFLINE_RAW_CC']),
     ],
 )
-async def test_wait_on_transfer(xattr_list, xattr_get, expected_result, tmp_path):
-    with mock.patch.object(os, "listxattr", lambda path: xattr_list):
-        with mock.patch.object(os, "getxattr", lambda path, attr: xattr_get):
-            res = await wait_on_transfer(
-                str(tmp_path),
-                max_tries_completion=1,
-                max_tries_attributes=1,
-                sleep_attributes=1,
-            )
-            assert res is expected_result
+async def test_wait_on_transfer(mock_json, expected_result):
+    response = mock.Mock()
+    response.status_code = 200
+    response.json = mock.Mock(return_value=mock_json)
+    client = mock.Mock()
+    client.get_runs_by_proposal_number_api = mock.Mock(
+        return_value=response)
+
+    res = await wait_on_transfer(client, 1, 1, 1, 1)
+    assert res == expected_result
+
 
 @pytest.mark.asyncio
 @pytest.mark.parametrize(
-    "xattr_list, xattr_get, exception_match",
+    "mock_json, status_code, exception_class, exception_match",
     [
-        ([], b"", r"FAILED:.*user.status.*"),
-        (["user.status"], b"notmigrated2d2", r"FAILED:.*notmigratedr2d2.*"),
-        (["user.status"], b"foobar", r"FAILED:.*unknown.*"),
+        ([],
+         200,
+         ValueError,
+         r"FAILED:.*empty response.*"),
+        ([],
+         404,
+         ValueError,
+         r"FAILED:.*MyMDC.*404.*"),
+        ({'runs': [{'repositories': {}}]},
+         200,
+         MigrationError,
+         r"FAILED:.*migration issue.*"),
     ],
 )
 async def test_wait_on_transfer_exceptions(
-    xattr_list, xattr_get, exception_match, tmp_path
+    mock_json, status_code, exception_class, exception_match
 ):
-    with mock.patch.object(os, "listxattr", lambda path: xattr_list):
-        with mock.patch.object(os, "getxattr", lambda path, attr: xattr_get):
-            with pytest.raises(MigrationError, match=exception_match):
-                await wait_on_transfer(
-                    str(tmp_path),
-                    max_tries_completion=1,
-                    max_tries_attributes=1,
-                    sleep_attributes=1,
-                )
-
-
-@pytest.mark.asyncio
-async def test_wait_on_transfer_timeout(tmp_path):
-    with mock.patch.object(os, "listxattr", lambda path: ["user.status"]):
-        with mock.patch.object(os, "getxattr", lambda path, attr: b"migration_in_progress"):
-            with pytest.raises(MigrationError, match=r"FAILED:.*progress.*"):
-                await wait_on_transfer(
-                    str(tmp_path),
-                    max_tries_completion=1,
-                    max_tries_attributes=1,
-                    sleep_attributes=1,
-                )
+    response = mock.Mock()
+    response.status_code = status_code
+    response.json = mock.Mock(return_value=mock_json)
+    client = mock.Mock()
+    client.get_runs_by_proposal_number_api = mock.Mock(
+        return_value=response)
+
+    with pytest.raises(exception_class, match=exception_match):
+        await wait_on_transfer(client, 1, 1, 1, 1)
 
 
 @pytest.mark.asyncio

webservice/messages.py

@@ -17,35 +17,15 @@ class Errors:
 
 
 class MigrationError(Exception):
-    @classmethod
-    def no_user_status_xattr(cls, run_path: str):
-        return cls(
-            f"FAILED: migration issue for run `{run_path}`: `user.status` xattr not "
-            f"present after 5 minutes, migration may have failed, manually restarting "
-            f"migration may help"
-        )
-
-    @classmethod
-    def migration_failed(cls, run_path: str):
-        return cls(
-            f"FAILED: migration issue for run `{run_path}`: migration marked as failed "
-            f"(`run user.status` set to `notmigratedr2d2`), manually restarting "
-            f"migration may help"
-        )
 
     @classmethod
-    def unknown_user_status(cls, run_path: str, status: str):
+    def timeout(cls, run: int, proposal: int, time: str):
         return cls(
-            f"FAILED: migration issue for run `{run_path}`: run has an unknown "
-            f"`user.status` xattr `{status}`, manually restarting migration may help"
+            f"FAILED: migration issue for proposal {proposal} run `{run}`: "
+            f"migration still incompleted after {time}, "
+            "manually restarting migration may help"
         )
 
-    @classmethod
-    def timeout(cls, run_path: str, time: str):
-        return cls(
-            f"FAILED: migration issue for run `{run_path}`: migration still marked as"
-            f"in progress after {time}, manually restarting migration may help"
-        )
 
 class MDC:
     MIGRATION_TIMEOUT = "Timeout waiting for migration. Contact it-support@xfel.eu"

webservice/webservice.py

@@ -30,12 +30,12 @@ from kafka import KafkaProducer
 from kafka.errors import KafkaError
 from metadata_client.metadata_client import MetadataClient
 
-from .config import webservice as config
-
 try:
-    from .messages import MDC, Errors, Success, MigrationError
+    from .config import webservice as config
+    from .messages import MDC, Errors, MigrationError, Success
 except ImportError:
-    from messages import MDC, Errors, Success, MigrationError
+    from config import webservice as config
+    from messages import MDC, Errors, MigrationError, Success
 
 
 def init_job_db(config):
@@ -534,108 +534,89 @@ async def run_action(job_db, cmd, mode, proposal, run, rid) -> str:
 
 
 async def wait_on_transfer(
-    run_path: str,
-    max_tries_completion: int = 300,
-    max_tries_attributes: int = 5,
+    mdc,
+    run: int,
+    proposal: int,
+    max_tries: int = 300,
     sleep_completion: int = 10,
-    sleep_attributes: int = 60,
-) -> bool:
-    """
-    Wait on data files to be transferred to Maxwell
-
-    :param run_path: Folder, which contains data files migrated to Maxwell
-    :param max_tries_completion: Maximum number of tries to check for migration completed attribute
-    :param max_tries_attributes: Maximum number of tries to check for migration attributes being present
-    :param sleep_completion: Sleep time between checks for migration completed attribute
-    :param sleep_attributes: Sleep time between checks for migration attributes being present
-    :return: True if files are transferred
+) -> List[str]:
+    """Query MyMDC to get run migration status.
+
+    This coro queries MyMDC to get the storage spaces where the data is
+    available from (known as repositories.)
+    If the data is available on offline or dCache storage, this coro will
+    return. Else, it will loop until available.
+
+    If the data is not avalailable after all retries, a MigrationError will be
+    raised.
+    A ValueError can also be raised if a repository is not supported in this
+    webservice.
+
+    :param run: the run id
+    :param proposal: the proposal id
+    :param max_tries: maximum number of tries to check for migration completed
+    :param sleep_completion: sleep time in seconds between checks
+    :return repos: a list of repositories in which the data is available
+    :raise MigrationError: when the data is not on offline or dCache within the
+                           maximum max_tries * sleep_completion time
+    :raise ValueError: if the reply from MyMDC is not 200 ok or no data was
+                       returned.
     """
-    # TODO: Make use of MyMDC to request whether the run has been copied.
-    # It is not sufficient to know that the files are on disk, but also to
-    # check the copy is finished (ie. that the files are complete).
-    if 'pnfs' in os.path.realpath(run_path):
-        return True
-
-    tries_for_completion = 0
-    tries_for_attributes = 0
-
-    # FIXME: if not kafka, then do event-driven, no sleep
-    # wait until folder gets created
-    while not os.path.exists(run_path):
-        if tries_for_completion > max_tries_completion:
-            return False
-        tries_for_completion += 1
-        await asyncio.sleep(sleep_completion)
+    loop = get_event_loop()
+    for iteration in range(max_tries):
+        response = await loop.run_in_executor(
+            None,
+            mdc.get_runs_by_proposal_number_api,
+            proposal,
+            run
+        )
 
-    # FIXME: if not kafka, then do event-driven, no sleep
-    # wait until files are migrated
-    while True:
-        #  TODO: add test case for migration issues/missing attribute handling
-        if "user.status" not in os.listxattr(run_path):
-            if tries_for_attributes >= max_tries_attributes:
-                logging.critical(
-                    "`status` attribute missing after max tries for migration reached. "
-                    "Migration may have failed, try triggering migration manually again."
-                )
-                #  TODO: automatically re-trigger migration...?
-                raise MigrationError.no_user_status_xattr(run_path)
+        if response.status_code != 200:
+            raise ValueError("FAILED: MyMDC replied with "
+                             f"{response.status_code}: {response.reason}")
 
-            tries_for_attributes += 1
-            logging.warning(
-                f"`status` attribute missing, migration may have failed, on attempt "
-                f"{tries_for_attributes}/{max_tries_attributes}"
-            )
+        if not response.json():
+            raise ValueError("FAILED: MyMDC replied empty response for "
+                             f"proposal {proposal}, run {run}")
 
-            #  Wait for a longer time if the attributes are missing
-            await asyncio.sleep(sleep_attributes)
-            continue
+        run_details, = response.json()['runs']
+        repositories = list(run_details['repositories'].keys())
 
-        user_status = os.getxattr(run_path, "user.status").decode().lower()
-        logging.debug(f"{run_path}: {user_status=}")  # TODO: use `trace` instead of `debug` w/ loguru
-        if user_status in ["dcache", "tape", "offline"]:
-            return True
-        elif user_status == "notmigrated2d2":
-            logging.critical(f"Migration failed for {run_path}")
-            raise MigrationError.migration_failed(run_path)
-        elif user_status != "migration_in_progress":
-            logging.critical("Unknown status: {user_status}")
-            if tries_for_completion > max_tries_completion:
-                raise MigrationError.unknown_user_status(run_path, user_status)
-
-        if tries_for_completion > max_tries_completion:
-            raise MigrationError.timeout(
-                run_path, f"{tries_for_completion*sleep_completion}s"
-            )
+        for identifier in repositories:
+            if any(s in identifier for s in ('DCACHE', 'OFFLINE')):
+                return repositories
 
-        tries_for_completion += 1
+        logging.info(f"Proposal {proposal} run {run} not migrated yet. "
+                     f"Will try again ({iteration}/{max_tries})")
         await asyncio.sleep(sleep_completion)
 
+    raise MigrationError.timeout(run, proposal,
+                                 f"{max_tries*sleep_completion}s")
+
 
 async def wait_transfers(
-        wait_runs: List[str], in_folder: str, proposal: str
+    mdc: MetadataClient,
+    runs: List[int],
+    proposal: int
 ) -> bool:
     """Wait for multiple runs to be transferred to Maxwell.
 
-    :param wait_runs: Run numbers to wait for
-    :param in_folder: Proposal raw directory containing runs
+    :param runs: Run numbers to wait for
     :param proposal: Proposal number
     :return: True if all runs transferred, false on timeout
     """
-    logging.debug(f"Waiting for: proposal {proposal}, runs {wait_runs}")
+    logging.debug(f"Waiting for: proposal {proposal}, runs {runs}")
 
-    # FIXME: this loop should be an asyncio.gather
-    for runnr in wait_runs:
-        rpath = "{}/r{:04d}/".format(in_folder, int(runnr))
-        transfer_complete = await wait_on_transfer(rpath)
-        if not transfer_complete:
-            logging.error(
-                Errors.TRANSFER_EVAL_FAILED.format(proposal, runnr)
-            )
-            return False
+    coros = [wait_on_transfer(mdc, run, proposal) for run in runs]
+    try:
+        await asyncio.gather(*coros)
+    except (MigrationError, ValueError):
+        logging.error(
+            Errors.TRANSFER_EVAL_FAILED.format(proposal, runs), exc_info=True
+        )
+        return False
 
-    logging.info(
-        "Transfer complete: proposal %s, runs %s", proposal, wait_runs
-    )
+    logging.info("Transfer complete: proposal %s, runs %s", proposal, runs)
     return True
 
 
@@ -881,7 +862,7 @@ class ActionsServer:
             await update_mdc_status(self.mdc, 'correct', rid, queued_msg)
             try:
                 transfer_complete = await wait_transfers(
-                    [runnr], in_folder, proposal
+                    self.mdc, [runnr], proposal
                 )
                 if not transfer_complete:
                     # Timed out
@@ -915,12 +896,8 @@ class ActionsServer:
                             thisconf["priority"] = str(priority)
 
                         detectors[karabo_id] = thisconf
-                copy_file_set = copy_file_list.difference(corr_file_list)  # noqa
+                copy_file_set = copy_file_list.difference(corr_file_list)
                 asyncio.ensure_future(copy_untouched_files(copy_file_set))
-            except MigrationError as e:
-                logging.error("Migration issue", exc_info=e)
-                await update_mdc_status(self.mdc, 'correct', rid, str(e))
-                return
             except Exception as corr_e:
                 logging.error("Error during correction", exc_info=corr_e)
                 await update_mdc_status(self.mdc, 'correct', rid,
@@ -1008,19 +985,12 @@ class ActionsServer:
         async def _continue():
             """Runs in the background after we reply to the 'dark_request' request"""
             await update_mdc_status(self.mdc, 'dark_request', rid, queued_msg)
-            try:
-                transfer_complete = await wait_transfers(
-                    runs, in_folder, proposal
+            transfer_complete = await wait_transfers(self.mdc, runs, proposal)
+            if not transfer_complete:
+                # Timed out
+                await update_mdc_status(
+                    self.mdc, 'dark_request', rid, MDC.MIGRATION_TIMEOUT
                 )
-                if not transfer_complete:
-                    # Timed out
-                    await update_mdc_status(
-                        self.mdc, 'dark_request', rid, MDC.MIGRATION_TIMEOUT
-                    )
-                    return
-            except MigrationError as e:
-                logging.error("Migration issue", exc_info=e)
-                await update_mdc_status(self.mdc, 'dark_request', rid, str(e))
                 return
 
             # Notebooks require one or three runs, depending on the