Differential D5950 Diff 21508 swh/scheduler/journal_client.py

Changeset View

Standalone View

View Options

swh/scheduler/journal_client.py

# See the AUTHORS file at the top-level directory of this distribution

# License: GNU General Public License version 3, or any later version

# See top-level LICENSE file for more information

from datetime import datetime

from datetime import datetime, timedelta

import random

from typing import Dict, List, Optional, Tuple

import attr

from swh.scheduler.interface import SchedulerInterface

from swh.scheduler.model import OriginVisitStats

from swh.scheduler.utils import utcnow

msg_type = "origin_visit_status"

def max_date(*dates: Optional[datetime]) -> datetime:

"""Return the max date of given (possibly None) dates

At least one date must be not None.

Show All 10 Lines

def update_next_position_offset(visit_stats: Dict, increment: int) -> None:

resulting value must be a positive integer.

"""

visit_stats["next_position_offset"] = max(

0, visit_stats["next_position_offset"] + increment

)

def from_position_offset_to_days(position_offset: int) -> int:

"""Compute position offset to interval in days.

- index 0 and 1: interval 1 day

- index 2, 3 and 4: interval 2 days

- index 5 and up: interval `4^(n-4)` days for n in (4, 16, 64, 256, 1024, ...)

Args:

position_offset: The actual position offset for a given visit stats

Returns:

The offset as an interval in number of days

"""

assert position_offset >= 0

if position_offset < 2:

result = 1

elif position_offset < 5:

result = 2

else:

result = 4 ** (position_offset - 4)

return result

def next_visit_queue_position(

queue_position_per_visit_type: Dict, visit_stats: Dict

) -> datetime:

"""Compute the next visit queue position for the given visit_stats.

Args:

- queue_position_per_visit_type: The global state of the queue per visit type

- visit_stats: The actual visit information to compute the next position for

Returns:

The actual next visit queue position for that visit stats

"""

# We set the next visit target to its current value + the new visit interval

# multiplied by a random fudge factor (picked in the -/+ 10% range) to avoid

# scheduling burst

days = from_position_offset_to_days(visit_stats["next_position_offset"])

random_fudge_factor = random.uniform(-0.1, 0.1)

visit_interval = timedelta(days=days * (1 + random_fudge_factor))

# Use the current queue position per visit type as starting position if none is

# already set

default_queue_position = queue_position_per_visit_type.get(

visit_stats["visit_type"], utcnow()

)

current_position = (

visit_stats["next_visit_queue_position"]

if visit_stats.get("next_visit_queue_position")

else default_queue_position

)

return current_position + visit_interval

def process_journal_objects(

messages: Dict[str, List[Dict]], *, scheduler: SchedulerInterface

) -> None:

"""Read messages from origin_visit_status journal topic to update "origin_visit_stats"

information on (origin, visit_type). The goal is to compute visit stats information

per origin and visit_type: last_eventful, last_uneventful, last_failed,

last_notfound, last_snapshot, ...

▲ Show 20 Lines • Show All 44 Lines • ▼ Show 20 Lines

origin_visit_stats: Dict[Tuple[str, str], Dict] = {

)

}

# Use the default values from the model object

empty_object = {

field.name: field.default if field.default != attr.NOTHING else None

for field in attr.fields(OriginVisitStats)

}

# Retrieve the global queue state

queue_position_per_visit_type = scheduler.visit_scheduler_queue_position_get()

for msg_dict in interesting_messages:

origin = msg_dict["origin"]

visit_type = msg_dict["type"]

pk = origin, visit_type

if pk not in origin_visit_stats:

origin_visit_stats[pk] = {

**empty_object,

"url": origin,

▲ Show 20 Lines • Show All 64 Lines • ▼ Show 20 Lines

for msg_dict in interesting_messages:

# populating the last_uneventful message

continue

else:

# uneventful event

visit_stats_d["last_uneventful"] = current_status_date

# Visit this origin less often in the future

update_next_position_offset(visit_stats_d, 1)

visit_stats_d["next_visit_queue_position"] = next_visit_queue_position(

queue_position_per_visit_type, visit_stats_d

)

ardumontAuthorUnsubmitted

Done

days = from_position_offset_to_days(visit_stats_d["next_position_offset"])

- random_fudge_factor = random.choice(range(-10, 11)) / 100

+ random_fudge_factor = random.uniform(-0.1, 0.1)

visit_interval = timedelta(days=days * (1 + random_fudge_factor))

use random.uniform!

ardumont: use random.uniform!

ardumontAuthorUnsubmitted

Done

great stuff [1]:

In [16]: while True: print(random.uniform(-0.1, 0.1)) ; sleep(0.1)
-0.07802568520265081
-0.012047862685772293
-0.09515824425588425
-0.09383355154670565
-0.07567022279491248
-0.028994578286562933
0.032746260794542265
0.09695343564361572
0.03365570420242092
-0.03822556466045879
-0.08169951622895931
-0.09698416654743738
-0.0011827698844187884
-0.036335877650609
0.05447047809839983
-0.05956846622376569
0.04799241406841087

[1] https://docs.python.org/3/library/random.html#random.uniform

ardumont: great stuff [1]: ``` In [16]: while True: print(random.uniform(-0.1, 0.1)) ; sleep(0.1) -0.

scheduler.origin_visit_stats_upsert(

ardumontAuthorUnsubmitted

Done

ensure days is ok with float.

ardumont: ensure days is ok with float.

ardumontAuthorUnsubmitted

Done

It was meant in regards to timedelta.

It is ok:

In [1]: from datetime import datetime, timedelta

In [2]: timedelta(days=0.1)
Out[2]: datetime.timedelta(seconds=8640)

In [3]: timedelta(days=1)
Out[3]: datetime.timedelta(days=1)

In [5]: timedelta(seconds=86400)
Out[5]: datetime.timedelta(days=1)

ardumont: It was meant in regards to timedelta. It is ok: ``` In [1]: from datetime import datetime…

OriginVisitStats(**ovs) for ovs in origin_visit_stats.values()

)