Page MenuHomeSoftware Heritage
Differential D5956 Diff 21407 swh/scheduler/tests/test_scheduler.py

Changeset View

Standalone View

View Options

swh/scheduler/tests/test_scheduler.py

# Copyright (C) 2017-2021 The Software Heritage developers # Copyright (C) 2017-2021 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution # See the AUTHORS file at the top-level directory of this distribution
# License: GNU General Public License version 3, or any later version # License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information # See top-level LICENSE file for more information
from collections import defaultdict from collections import defaultdict
import copy import copy
import datetime import datetime
from datetime import timedelta
import inspect import inspect
import random import random
from typing import Any, Dict, List, Optional, Tuple from typing import Any, Dict, List, Optional, Tuple
import uuid import uuid
import attr import attr
from psycopg2.extras import execute_values from psycopg2.extras import execute_values
import pytest import pytest
from swh.model.hashutil import hash_to_bytes from swh.model.hashutil import hash_to_bytes
from swh.scheduler.exc import SchedulerException, StaleData, UnknownPolicy from swh.scheduler.exc import SchedulerException, StaleData, UnknownPolicy
from swh.scheduler.interface import ListedOriginPageToken, SchedulerInterface from swh.scheduler.interface import (
SCHEDULING_POLICY_ORIGINS_NO_LAST_UPDATE,
ListedOriginPageToken,
SchedulerInterface,
)
from swh.scheduler.model import ListedOrigin, OriginVisitStats, SchedulerMetrics from swh.scheduler.model import ListedOrigin, OriginVisitStats, SchedulerMetrics
from swh.scheduler.utils import utcnow from swh.scheduler.utils import utcnow
from .common import ( from .common import (
LISTERS, LISTERS,
TASK_TYPES, TASK_TYPES,
TEMPLATES, TEMPLATES,
tasks_from_template, tasks_from_template,
tasks_with_priority_from_template, tasks_with_priority_from_template,
) )
ONEDAY = datetime.timedelta(days=1) ONEDAY = timedelta(days=1)
NUM_PRIORITY_TASKS = {None: 100, "high": 60, "normal": 30, "low": 20} NUM_PRIORITY_TASKS = {None: 100, "high": 60, "normal": 30, "low": 20}
def subdict(d, keys=None, excl=()): def subdict(d, keys=None, excl=()):
if keys is None: if keys is None:
keys = [k for k in d.keys()] keys = [k for k in d.keys()]
return {k: d[k] for k in keys if k not in excl} return {k: d[k] for k in keys if k not in excl}
▲ Show 20 Lines • Show All 773 Lines • ▼ Show 20 Lines ):
after = self._check_grab_next_visit_basic( after = self._check_grab_next_visit_basic(
swh_scheduler, visit_type, policy, expected, **kwargs swh_scheduler, visit_type, policy, expected, **kwargs
) )
# But a week, later, they should # But a week, later, they should
ret = swh_scheduler.grab_next_visits( ret = swh_scheduler.grab_next_visits(
visit_type=visit_type, visit_type=visit_type,
count=len(expected) + 1, count=len(expected) + 1,
policy=policy, policy=policy,
timestamp=after + datetime.timedelta(days=7), timestamp=after + timedelta(days=7),
) )
# We need to sort them because their 'last_scheduled' field is updated to # We need to sort them because their 'last_scheduled' field is updated to
# exactly the same value, so the order is not deterministic # exactly the same value, so the order is not deterministic
assert sorted(ret) == sorted( assert sorted(ret) == sorted(
expected expected
), "grab_next_visits didn't reschedule visits after a week" ), "grab_next_visits didn't reschedule visits after a week"
def _prepare_oldest_scheduled_first_origins( def _prepare_oldest_scheduled_first_origins(
Show All 9 Lines ):
OriginVisitStats( OriginVisitStats(
url=origin.url, url=origin.url,
visit_type=origin.visit_type, visit_type=origin.visit_type,
last_snapshot=None, last_snapshot=None,
last_eventful=None, last_eventful=None,
last_uneventful=None, last_uneventful=None,
last_failed=None, last_failed=None,
last_notfound=None, last_notfound=None,
last_scheduled=base_date - datetime.timedelta(seconds=i), last_scheduled=base_date - timedelta(seconds=i),
) )
for i, origin in enumerate(origins[1:]) for i, origin in enumerate(origins[1:])
] ]
swh_scheduler.origin_visit_stats_upsert(visit_stats) swh_scheduler.origin_visit_stats_upsert(visit_stats)
# We expect to retrieve the origin with a NULL last_scheduled # We expect to retrieve the origin with a NULL last_scheduled
# as well as those with the oldest values (i.e. the last ones), in order. # as well as those with the oldest values (i.e. the last ones), in order.
expected = [origins[0]] + origins[1:][::-1] expected = [origins[0]] + origins[1:][::-1]
▲ Show 20 Lines • Show All 41 Lines • ▼ Show 20 Lines ):
last_eventful=None, last_eventful=None,
last_uneventful=None, last_uneventful=None,
**ovs_args, **ovs_args,
) )
for i, origin in enumerate(origins) for i, origin in enumerate(origins)
] ]
swh_scheduler.origin_visit_stats_upsert(visit_stats) swh_scheduler.origin_visit_stats_upsert(visit_stats)
cooldown_td = datetime.timedelta(days=cooldown) cooldown_td = timedelta(days=cooldown)
cooldown_args = { cooldown_args = {
"scheduled_cooldown": None, "scheduled_cooldown": None,
"failed_cooldown": None, "failed_cooldown": None,
"notfound_cooldown": None, "notfound_cooldown": None,
} }
cooldown_args[f"{which_cooldown}_cooldown"] = cooldown_td cooldown_args[f"{which_cooldown}_cooldown"] = cooldown_td
ret = swh_scheduler.grab_next_visits( ret = swh_scheduler.grab_next_visits(
visit_type=visit_type, visit_type=visit_type,
count=len(expected) + 1, count=len(expected) + 1,
policy="oldest_scheduled_first", policy="oldest_scheduled_first",
timestamp=after + cooldown_td - datetime.timedelta(seconds=1), timestamp=after + cooldown_td - timedelta(seconds=1),
**cooldown_args, **cooldown_args,
) )
assert ret == [], f"{which_cooldown}_cooldown ignored" assert ret == [], f"{which_cooldown}_cooldown ignored"
ret = swh_scheduler.grab_next_visits( ret = swh_scheduler.grab_next_visits(
visit_type=visit_type, visit_type=visit_type,
count=len(expected) + 1, count=len(expected) + 1,
policy="oldest_scheduled_first", policy="oldest_scheduled_first",
timestamp=after + cooldown_td + datetime.timedelta(seconds=1), timestamp=after + cooldown_td + timedelta(seconds=1),
**cooldown_args, **cooldown_args,
) )
assert sorted(ret) == sorted( assert sorted(ret) == sorted(
expected expected
), "grab_next_visits didn't reschedule visits after the configured cooldown" ), "grab_next_visits didn't reschedule visits after the configured cooldown"
def test_grab_next_visits_never_visited_oldest_update_first( def test_grab_next_visits_never_visited_oldest_update_first(
self, swh_scheduler, listed_origins_by_type, self, swh_scheduler, listed_origins_by_type,
): ):
visit_type, origins = self._grab_next_visits_setup( visit_type, origins = self._grab_next_visits_setup(
swh_scheduler, listed_origins_by_type swh_scheduler, listed_origins_by_type
) )
# Update known origins with a `last_update` field that we control # Update known origins with a `last_update` field that we control
base_date = datetime.datetime(2020, 1, 1, 0, 0, 0, tzinfo=datetime.timezone.utc) base_date = datetime.datetime(2020, 1, 1, 0, 0, 0, tzinfo=datetime.timezone.utc)
updated_origins = [ updated_origins = [
attr.evolve(origin, last_update=base_date - datetime.timedelta(seconds=i)) attr.evolve(origin, last_update=base_date - timedelta(seconds=i))
for i, origin in enumerate(origins) for i, origin in enumerate(origins)
] ]
updated_origins = swh_scheduler.record_listed_origins(updated_origins) updated_origins = swh_scheduler.record_listed_origins(updated_origins)
# We expect to retrieve origins with the oldest update date, that is # We expect to retrieve origins with the oldest update date, that is
# origins at the end of our updated_origins list. # origins at the end of our updated_origins list.
expected_origins = sorted(updated_origins, key=lambda o: o.last_update) expected_origins = sorted(updated_origins, key=lambda o: o.last_update)
Show All 9 Linesclass TestScheduler:
): ):
visit_type, origins = self._grab_next_visits_setup( visit_type, origins = self._grab_next_visits_setup(
swh_scheduler, listed_origins_by_type swh_scheduler, listed_origins_by_type
) )
# Update known origins with a `last_update` field that we control # Update known origins with a `last_update` field that we control
base_date = datetime.datetime(2020, 1, 1, 0, 0, 0, tzinfo=datetime.timezone.utc) base_date = datetime.datetime(2020, 1, 1, 0, 0, 0, tzinfo=datetime.timezone.utc)
updated_origins = [ updated_origins = [
attr.evolve(origin, last_update=base_date - datetime.timedelta(seconds=i)) attr.evolve(origin, last_update=base_date - timedelta(seconds=i))
for i, origin in enumerate(origins) for i, origin in enumerate(origins)
] ]
updated_origins = swh_scheduler.record_listed_origins(updated_origins) updated_origins = swh_scheduler.record_listed_origins(updated_origins)
# Update the visit stats with a known visit at a controlled date for # Update the visit stats with a known visit at a controlled date for
# half the origins. Pick the date in the middle of the # half the origins. Pick the date in the middle of the
# updated_origins' `last_update` range # updated_origins' `last_update` range
visit_date = updated_origins[len(updated_origins) // 2].last_update visit_date = updated_origins[len(updated_origins) // 2].last_update
Show All 38 Lines def test_grab_next_visits_underflow(self, swh_scheduler, listed_origins_by_type):
swh_scheduler.record_listed_origins(origins) swh_scheduler.record_listed_origins(origins)
ret = swh_scheduler.grab_next_visits( ret = swh_scheduler.grab_next_visits(
visit_type, len(origins) + 2, policy="oldest_scheduled_first" visit_type, len(origins) + 2, policy="oldest_scheduled_first"
) )
assert len(ret) == 5 assert len(ret) == 5
def test_grab_next_visits_no_last_update(
self, swh_scheduler, listed_origins_by_type
):
"""Check that grab_next_visits retrieve tasks without last update information"""
visit_type = next(iter(listed_origins_by_type))
origins = []
for origin in listed_origins_by_type[visit_type]:
origins.append(
attr.evolve(origin, last_update=None)
) # void the last update
assert len(origins) > 0
swh_scheduler.record_listed_origins(origins[:900])
# As visit_scheduler_queue_position state and no origin_visit_stats are prepared
# yet
actual_state = swh_scheduler.visit_scheduler_queue_position_get()
assert actual_state == {}
ret = swh_scheduler.grab_next_visits(
visit_type, len(origins), policy=SCHEDULING_POLICY_ORIGINS_NO_LAST_UPDATE,
)
# Nothing will get fetched
assert len(ret) == 0
ardumontUnsubmitted
Done
Inline Actions
assert len(ret) == 0
- state_position = utcnow() - timedelta(days=-1)
+ state_position = utcnow() - timedelta(days=1)
# Prepare the visit_scheduler_queue_position state for the visit type concerned

¯\_(ツ)_/¯

ardumont: ¯\_(ツ)_/¯
state_position = utcnow() - timedelta(days=1)
# Prepare the visit_scheduler_queue_position state for the visit type concerned
# here
swh_scheduler.visit_scheduler_queue_position_set(visit_type, state_position)
ret = swh_scheduler.grab_next_visits(
visit_type, len(origins), policy=SCHEDULING_POLICY_ORIGINS_NO_LAST_UPDATE,
)
# No origin visit stats so nothing get scheduled
assert len(ret) == 0
relevant_origin_index = 900
# Simulate the journal client updated information in the right range (1 day)
visit_stats = [
OriginVisitStats(
url=origin.url,
visit_type=origin.visit_type,
last_eventful=utcnow(),
last_uneventful=None,
last_failed=None,
last_notfound=None,
next_visit_queue_position=state_position
+ timedelta(days=random.choice([d * 0.1 for d in range(0, 10)])),
)
for origin in origins[:relevant_origin_index] # are relevant for test
] + [
# those visit statuses whose queue position is "far" in the future gets
# ignored
OriginVisitStats(
url=origin.url,
visit_type=origin.visit_type,
last_eventful=utcnow(),
last_uneventful=None,
last_failed=None,
last_notfound=None,
next_visit_queue_position=state_position
+ timedelta(days=random.choice(range(2, 10))),
)
for origin in origins[relevant_origin_index:] # will get ignored
]
swh_scheduler.origin_visit_stats_upsert(visit_stats)
actual_visits = swh_scheduler.grab_next_visits(
visit_type, len(origins), policy=SCHEDULING_POLICY_ORIGINS_NO_LAST_UPDATE,
)
assert len(actual_visits) == len(origins[:900])
visit_stats = swh_scheduler.origin_visit_stats_get(
(o.url, o.visit_type) for o in actual_visits
)
assert len(visit_stats) == len(actual_visits)
for visit_stat, origin in zip(visit_stats, actual_visits):
assert origin.url == visit_stat.url
assert origin.visit_type == visit_stat.visit_type
assert origin.last_update is None
assert (
state_position <= visit_stat.next_visit_queue_position
and visit_stat.next_visit_queue_position
< state_position + timedelta(days=1)
)
def test_grab_next_visits_unknown_policy(self, swh_scheduler): def test_grab_next_visits_unknown_policy(self, swh_scheduler):
unknown_policy = "non_existing_policy" unknown_policy = "non_existing_policy"
NUM_RESULTS = 5 NUM_RESULTS = 5
with pytest.raises(UnknownPolicy, match=unknown_policy): with pytest.raises(UnknownPolicy, match=unknown_policy):
swh_scheduler.grab_next_visits("type", NUM_RESULTS, policy=unknown_policy) swh_scheduler.grab_next_visits("type", NUM_RESULTS, policy=unknown_policy)
def _create_task_types(self, scheduler): def _create_task_types(self, scheduler):
for tt in TASK_TYPES.values(): for tt in TASK_TYPES.values():
▲ Show 20 Lines • Show All 247 Lines • ▼ Show 20 Lines def test_metrics_origins_with_pending_changes(self, swh_scheduler, listed_origins):
# respect to the "last update" time for the origin # respect to the "last update" time for the origin
visited_origin = listed_origins[0] visited_origin = listed_origins[0]
assert visited_origin.last_update is not None assert visited_origin.last_update is not None
swh_scheduler.origin_visit_stats_upsert( swh_scheduler.origin_visit_stats_upsert(
[ [
OriginVisitStats( OriginVisitStats(
url=visited_origin.url, url=visited_origin.url,
visit_type=visited_origin.visit_type, visit_type=visited_origin.visit_type,
last_eventful=visited_origin.last_update last_eventful=visited_origin.last_update - timedelta(days=1),
- datetime.timedelta(days=1),
last_uneventful=None, last_uneventful=None,
last_failed=None, last_failed=None,
last_notfound=None, last_notfound=None,
last_snapshot=hash_to_bytes( last_snapshot=hash_to_bytes(
"d81cc0710eb6cf9efd5b920a8453e1e07157b6cd" "d81cc0710eb6cf9efd5b920a8453e1e07157b6cd"
), ),
), ),
] ]
Show All 29 Lines
def test_update_metrics_twice(self, swh_scheduler, listed_origins): def test_update_metrics_twice(self, swh_scheduler, listed_origins):
swh_scheduler.record_listed_origins(listed_origins) swh_scheduler.record_listed_origins(listed_origins)
ts = utcnow() ts = utcnow()
ret = swh_scheduler.update_metrics(timestamp=ts) ret = swh_scheduler.update_metrics(timestamp=ts)
assert all(metric.last_update == ts for metric in ret) assert all(metric.last_update == ts for metric in ret)
second_ts = ts + datetime.timedelta(seconds=1) second_ts = ts + timedelta(seconds=1)
ret = swh_scheduler.update_metrics(timestamp=second_ts) ret = swh_scheduler.update_metrics(timestamp=second_ts)
assert all(metric.last_update == second_ts for metric in ret) assert all(metric.last_update == second_ts for metric in ret)
def test_get_metrics(self, swh_scheduler, listed_origins): def test_get_metrics(self, swh_scheduler, listed_origins):
swh_scheduler.record_listed_origins(listed_origins) swh_scheduler.record_listed_origins(listed_origins)
updated = swh_scheduler.update_metrics() updated = swh_scheduler.update_metrics()
retrieved = swh_scheduler.get_metrics() retrieved = swh_scheduler.get_metrics()
Show All 29 Lines
About · Developer information · Legal