diff --git a/application/tests/gap_analysis_preload_test.py b/application/tests/gap_analysis_preload_test.py new file mode 100644 index 000000000..bb1ce7d4f --- /dev/null +++ b/application/tests/gap_analysis_preload_test.py @@ -0,0 +1,105 @@ +import unittest +from typing import Any, Callable, Dict, List +from unittest.mock import MagicMock, patch + +from application.utils import gap_analysis + + +def _response(status: int = 200, body: Any = None) -> MagicMock: + res = MagicMock() + res.status_code = status + res.json.return_value = body if body is not None else {} + return res + + +class PreloadTest(unittest.TestCase): + """preload() polls map_analysis until every pair is cached or given up on.""" + + def _run( + self, standards: List[str], answer: Callable[[str, int], MagicMock] + ) -> Dict[str, Any]: + polls: Dict[str, int] = {} + + def fake_get(url: str) -> MagicMock: + if url.endswith("/rest/v1/standards"): + return _response(body=standards) + pair = url.split("standard=", 1)[1].replace("&standard=", "->") + polls[pair] = polls.get(pair, 0) + 1 + return answer(pair, polls[pair]) + + with patch.object(gap_analysis.requests, "get", side_effect=fake_get), patch( + "application.utils.gap_analysis.time.sleep" + ) as sleep: + dropped = gap_analysis.preload(target_url="http://x") + return {"dropped": dropped, "polls": polls, "sleeps": sleep.call_count} + + def test_all_pairs_cached_needs_one_pass_and_no_sleep(self) -> None: + out = self._run(["A", "B"], lambda _p, _n: _response(body={"result": {"k": 1}})) + self.assertEqual([], out["dropped"]) + self.assertEqual({"A->B": 1, "B->A": 1}, out["polls"]) + self.assertEqual(0, out["sleeps"]) + + def test_finished_pair_is_not_polled_again(self) -> None: + def answer(pair: str, n: int) -> MagicMock: + if pair == "A->B" or n >= 3: + return _response(body={"result": {"k": 1}}) + return _response(body={"job_id": "job-1"}) + + out = self._run(["A", "B"], answer) + self.assertEqual(1, out["polls"]["A->B"]) + self.assertEqual(3, out["polls"]["B->A"]) + + def test_long_running_job_is_not_dropped(self) -> None: + # GA jobs may run for 36 hours, so polling the same job id is not a failure. + polls_while_running = gap_analysis.PRELOAD_MAX_FAILURES * 5 + + def answer(_pair: str, n: int) -> MagicMock: + if n > polls_while_running: + return _response(body={"result": {"k": 1}}) + return _response(body={"job_id": "same-job"}) + + out = self._run(["A", "B"], answer) + self.assertEqual([], out["dropped"]) + + def test_pair_that_keeps_returning_404_is_dropped(self) -> None: + out = self._run(["A", "B"], lambda _p, _n: _response(status=404)) + self.assertEqual(["A->B", "B->A"], sorted(out["dropped"])) + self.assertEqual(gap_analysis.PRELOAD_MAX_FAILURES, out["polls"]["A->B"]) + + def test_job_that_keeps_failing_and_requeueing_is_dropped(self) -> None: + # map_analysis enqueues a new job once the last one failed, so a changed + # job id means the previous job ended without producing a result. + out = self._run(["A", "B"], lambda p, n: _response(body={"job_id": f"{p}{n}"})) + self.assertEqual(["A->B", "B->A"], sorted(out["dropped"])) + # The first poll only records the job id, it is not a failure. + self.assertEqual(gap_analysis.PRELOAD_MAX_FAILURES + 1, out["polls"]["A->B"]) + + def test_scattered_errors_do_not_drop_a_running_job(self) -> None: + def answer(_pair: str, n: int) -> MagicMock: + if n > 200: + return _response(body={"result": {"k": 1}}) + if n % 20 == 0: + return _response(status=503) + return _response(body={"job_id": "same-job"}) + + out = self._run(["A", "B"], answer) + self.assertEqual([], out["dropped"]) + + def test_failures_are_counted_per_pair(self) -> None: + def answer(pair: str, _n: int) -> MagicMock: + if pair == "A->B": + return _response(status=404) + return _response(body={"result": {"k": 1}}) + + out = self._run(["A", "B"], answer) + self.assertEqual(["A->B"], out["dropped"]) + + def test_success_is_not_logged_when_pairs_were_dropped(self) -> None: + with self.assertLogs(gap_analysis.logger, level="INFO") as logs: + self._run(["A", "B"], lambda _p, _n: _response(status=404)) + self.assertFalse(any("preloaded successfully" in m for m in logs.output)) + self.assertTrue(any("A->B" in m and "ERROR" in m for m in logs.output)) + + +if __name__ == "__main__": + unittest.main() diff --git a/application/utils/gap_analysis.py b/application/utils/gap_analysis.py index 49810a278..a6ce54dd5 100644 --- a/application/utils/gap_analysis.py +++ b/application/utils/gap_analysis.py @@ -5,7 +5,7 @@ import os import requests import time -from typing import Any, Dict, List, Optional +from typing import Any, Dict, List, Optional, Tuple from application.utils import redis from flask import json as flask_json import json @@ -22,6 +22,8 @@ } GAP_ANALYSIS_TIMEOUT = "129600s" # 36 hours +PRELOAD_MAX_FAILURES = 10 +PRELOAD_POLL_SECONDS = 30 OPENCRE_STANDARD_NAME = "OpenCRE" OPENCRE_OVERLAP_LINK_TYPES = ( defs.LinkTypes.LinkedTo, @@ -400,42 +402,85 @@ def schedule(standards: List[str], database): return {"result": res.get("result") if isinstance(res, dict) else res} -def preload(target_url: str): - waiting = [] +def preload(target_url: str) -> List[str]: + """ + Poll map_analysis until every standard pair is cached or given up on. + + A pair still running under the same job id is left alone, since GA jobs may + run for GAP_ANALYSIS_TIMEOUT. A pair counts a failure on a non-200 response, + on a 200 with neither result nor job id, and when its job id changes, because + map_analysis enqueues a new job when the last one ended without a result (or + its in-flight key expired). Seeing the same job again clears the count. After + PRELOAD_MAX_FAILURES failures in a row a pair is dropped, so one bad pair + cannot block the run forever. + + Returns the dropped pairs as ``A->B`` strings. + """ standards_request = requests.get(f"{target_url}/rest/v1/standards") standards = standards_request.json() - def calculate_a_to_b(sa: str, sb: str) -> bool: + def poll_pair(sa: str, sb: str) -> Tuple[bool, Optional[str]]: + """Returns (cached, job_id).""" res = requests.get( f"{target_url}/rest/v1/map_analysis?standard={sa}&standard={sb}" ) if res.status_code != 200: logger.warning("%s->%s returned %s", sa, sb, res.status_code) - return False + return False, None tojson = res.json() if tojson.get("result"): - return True + return True, None if tojson.get("job_id"): logger.info("%s->%s waiting", sa, sb) - return False + return False, tojson["job_id"] logger.warning( "%s->%s returned 200 but has no 'result' or 'job_id' key", sa, sb ) - return False - - for sa in standards: - for sb in standards: - if sa == sb: + return False, None + + waiting = [(sa, sb) for sa in standards for sb in standards if sa != sb] + failures: Dict[Tuple[str, str], int] = {pair: 0 for pair in waiting} + last_job_ids: Dict[Tuple[str, str], str] = {} + dropped: List[str] = [] + + while waiting: + for pair in list(waiting): + cached, job_id = poll_pair(*pair) + if cached: + waiting.remove(pair) continue - waiting.append(f"{sa}->{sb}") - while len(waiting): - for sa in standards: - for sb in standards: - if sa == sb: + + previous_job_id = last_job_ids.get(pair) + if job_id: + last_job_ids[pair] = job_id + if previous_job_id is None or previous_job_id == job_id: + # A job that is still alive clears earlier blips, so ten + # scattered 503s cannot drop a healthy long-running job. + failures[pair] = 0 continue - if calculate_a_to_b(sa, sb): - waiting.remove(f"{sa}->{sb}") if f"{sa}->{sb}" in waiting else "" - logger.info("calculating %s gap analyses", len(waiting)) - time.sleep(30) - logger.info("map analysis preloaded successfully") + + failures[pair] += 1 + if failures[pair] >= PRELOAD_MAX_FAILURES: + logger.warning( + "%s->%s failed %s times, dropping it", + pair[0], + pair[1], + failures[pair], + ) + waiting.remove(pair) + dropped.append(f"{pair[0]}->{pair[1]}") + + if waiting: + logger.info("calculating %s gap analyses", len(waiting)) + time.sleep(PRELOAD_POLL_SECONDS) + + if dropped: + logger.error( + "map analysis preload gave up on %s pairs: %s", + len(dropped), + ", ".join(dropped), + ) + else: + logger.info("map analysis preloaded successfully") + return dropped