Differential D5950 Diff 21361 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 from 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, ...)

"""

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 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 124 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)

# 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).

days = from_position_offset_to_days(visit_stats_d["next_position_offset"])

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

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.

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

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…

# FIXME: or is this here that we use visit_scheduler_queue_position.position for

# the given visit type to initialize such value?

current_position = (

visit_stats_d["next_visit_queue_position"]

if visit_stats_d.get("next_visit_queue_position")

else utcnow()

)

visit_stats_d["next_visit_queue_position"] = current_position + visit_interval

scheduler.origin_visit_stats_upsert(

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

)