- tor-commits - lists.torproject.org

[sbws/master] chg: scanner: Remove not needed thread loop
by juga＠torproject.org 21 Mar '19

21 Mar '19

commit 6fc0c16fc756f9cf4820b94d60b26c59b350a125 Author: juga0 <juga(a)riseup.net> Date: Mon Mar 4 19:00:32 2019 +0000 chg: scanner: Remove not needed thread loop The pool can queue all the relays to be measured, but this inner loop was not adding more worker threads to the pool unit it finishs the current threads. Instead, wait for all the worker threads in the pool to finish or try to get the status after waiting enough. Closes: #28864, #28865. Bugfix v1.0.0 --- sbws/core/scanner.py | 116 ++++++++++++++++++++++++++++++++++----------------- 1 file changed, 77 insertions(+), 39 deletions(-) diff --git a/sbws/core/scanner.py b/sbws/core/scanner.py index 2e88b53..3d0924d 100644 --- a/sbws/core/scanner.py +++ b/sbws/core/scanner.py @@ -443,28 +443,21 @@ def main_loop(args, conf, controller, relay_list, circuit_builder, result_dump, instead ``result_putter_error``, which logs the error and complete immediately. - Before iterating over the next relay, it waits (non blocking, since it - happens in the main thread) until one of the ``max_pending_results`` - threads has finished. - - This is not needed, since otherwise async_result will queue the relays to - measure in order and won't start reusing a thread to measure a relay until - other thread has finished. But it makes the logic a bit more sequential. - - Before the outer loop iterates, it also waits (again non blocking) that all - the ``Results`` are ready. + Before the outer loop iterates, it waits (non blocking) that all + the ``Results`` are ready calling ``wait_for_results``. This avoid to start measuring the same relay which might still being measured. """ - pending_results = [] # Set the time to wait for a thread to finish as the half of an HTTP # request timeout. - time_to_sleep = conf.getfloat('general', 'http_timeout') / 2 # Do not start a new loop if sbws is stopping. while not settings.end_event.is_set(): log.debug("Starting a new measurement loop.") num_relays = 0 + # Since loop might finish before pending_results is 0 due waiting too + # long, set it here and not outside the loop. + pending_results = [] loop_tstart = time.time() for target in relay_prioritizer.best_priority(): # Don't start measuring a relay if sbws is stopping. @@ -479,14 +472,14 @@ def main_loop(args, conf, controller, relay_list, circuit_builder, result_dump, [args, conf, destinations, circuit_builder, relay_list, target], {}, callback, callback_err) pending_results.append(async_result) - # Instead of letting apply_async to queue the relays in order until - # a thread has finished, wait here until a thread has finished. - while len(pending_results) >= max_pending_results: - # sleep is non-blocking since happens in the main process. - time.sleep(time_to_sleep) - pending_results = [r for r in pending_results if not r.ready()] - - wait_for_results(time_to_sleep, pending_results) + + # After the for has finished, the pool has queued all the relays + # and pending_results has the list of all the AsyncResults. + # It could also be obtained with pool._cache, which contains + # a dictionary with AsyncResults as items. + num_relays_to_measure = len(pending_results) + wait_for_results(num_relays_to_measure, pending_results) + loop_tstop = time.time() loop_tdelta = (loop_tstop - loop_tstart) / 60 log.debug("Measured %s relays in %s minutes", num_relays, loop_tdelta) @@ -497,27 +490,71 @@ def main_loop(args, conf, controller, relay_list, circuit_builder, result_dump, stop_threads(signal.SIGTERM, None) -def wait_for_results(time_to_sleep, pending_results): - time_waiting = 0 - while (len(pending_results) > 0 - and time_waiting <= TIMEOUT_MEASUREMENTS): - log.debug("Number of pending measurement threads %s after " - "a prioritization loop.", len(pending_results)) - time.sleep(time_to_sleep) - time_waiting += time_to_sleep +def wait_for_results(num_relays_to_measure, pending_results): + """Wait for the pool to finish and log progress. + + While there are relays being measured, just log the progress + and sleep :const:`~sbws.globals.TIMEOUT_MEASUREMENTS` (3mins), + which is aproximately the time it can take to measure a relay in + the worst case. + + When there has not been any relay measured in ``TIMEOUT_MEASUREMENTS`` + and there are still relays pending to be measured, it means there is no + progress and call :func:`~sbws.core.scanner.force_get_results`. + + This can happen in the case of a bug that makes either + :func:`~sbws.core.scanner.measure_relay`, + :func:`~sbws.core.scanner.result_putter` (callback) and/or + :func:`~sbws.core.scanner.result_putter_error` (callback error) stall. + + .. note:: in a future refactor, this could be simpler by: + + 1. Initializing the pool at the begingging of each loop + 2. Callling :meth:`~Pool.close`; :meth:`~Pool.join` after + :meth:`~Pool.apply_async`, + to ensure no new jobs are added until the pool has finished with all + the ones in the queue. + + As currently, there would be still two cases when the pool could stall: + + 1. There's an exception in ``measure_relay`` and another in + ``callback_err`` + 2. There's an exception ``callback``. + + This could also be simpler by not having callback and callback error in + ``apply_async`` and instead just calling callback with the + ``pending_results``. + + (callback could be also simpler by not having a thread and queue and + just storing to disk, since the time to write to disk is way smaller + than the time to request over the network.) + """ + num_last_measured = 1 + while num_last_measured > 0: + log.info("Pending measurements: %s out of %s: ", + len(pending_results), num_relays_to_measure) + time.sleep(TIMEOUT_MEASUREMENTS) + old_pending_results = pending_results pending_results = [r for r in pending_results if not r.ready()] - if time_waiting > TIMEOUT_MEASUREMENTS: - dumpstacks() + num_last_measured = len(old_pending_results) - len(pending_results) + if len(pending_results) > 0: + force_get_results(pending_results) def force_get_results(pending_results): - """Try to get either the result or an exception, which is logged. + """Try to get either the result or an exception, which gets logged. - Get call by ``wait_for_results`` when the time waiting was already long. - To get either the Result or an exception, call `get` with timeout. + It is call by :func:`~sbws.core.scanner.wait_for_results` when + the time waiting for the results was long. + + To get either the :class:`~sbws.lib.resultdump.Result` or an exception, + call :meth:`~AsyncResult.get` with timeout. Timeout is low since we already waited. - `get` is not call before, because it blocks and the callbacks are not call. + + ``get`` is not call before, because it blocks and the callbacks + are not call. """ + log.debug("Forcing get") for r in pending_results: try: result = r.get(timeout=0.1) @@ -527,13 +564,14 @@ def force_get_results(pending_results): # been processed yet except TimeoutError: log.warning("A result was not stored, it was not ready.") - # If the result raised an exception, get returns it, - # so, log any exception so that it can be fixed. - # This should not happen, since callback_err would have been call first + # If the result raised an exception, `get` returns it, + # then log any exception so that it can be fixed. + # This should not happen, since `callback_err` would have been call + # first. except Exception as e: log.critical(FILLUP_TICKET_MSG) - log.exception("exception %s", e) - # if the exception happened in the threads: + # If the exception happened in the threads, `log.exception` does + # not have the traceback. log.warning("traceback %s", traceback.print_exception(type(e), e, e.__traceback__))

1 0

[sbws/master] new: v3bwfile: Add KeyValues to monitor relays
by juga＠torproject.org 21 Mar '19

21 Mar '19

commit 013cbb7247d531511672e26abbd69a4177014a6b Author: juga0 <juga(a)riseup.net> Date: Tue Feb 26 15:33:58 2019 +0000 new: v3bwfile: Add KeyValues to monitor relays that are not measured or not reported in the bandwidth file. Part of #29591. --- sbws/lib/v3bwfile.py | 95 ++++++++++++++++++++++++++++++++++++++++++++++++---- 1 file changed, 89 insertions(+), 6 deletions(-) diff --git a/sbws/lib/v3bwfile.py b/sbws/lib/v3bwfile.py index 37475d2..d6c545a 100644 --- a/sbws/lib/v3bwfile.py +++ b/sbws/lib/v3bwfile.py @@ -26,20 +26,71 @@ log = logging.getLogger(__name__) LINE_SEP = '\n' KEYVALUE_SEP_V1 = '=' KEYVALUE_SEP_V2 = ' ' + +# NOTE: in a future refactor make make all the KeyValues be a dictionary +# with their type, so that it's more similar to stem parser. + +# Header KeyValues +# ================= # List of the extra KeyValues accepted by the class EXTRA_ARG_KEYVALUES = ['software', 'software_version', 'file_created', 'earliest_bandwidth', 'generator_started', 'scanner_country', 'destinations_countries'] +# number_eligible_relays is the number that ends in the bandwidth file +# ie, have not been excluded by one of the filters in 4. below +# They should be call recent_measurement_included_count to be congruent +# with the other KeyValues. STATS_KEYVALUES = ['number_eligible_relays', 'minimum_number_eligible_relays', 'number_consensus_relays', 'percent_eligible_relays', 'minimum_percent_eligible_relays'] -KEYVALUES_INT = STATS_KEYVALUES +# Added in #29591 +BW_HEADER_KEYVALUES_MONITOR = [ + # 1.1 header: the number of different consensuses, that sbws has seen, + # since the last 5 days + 'recent_consensus_count', + # 2.4 Number of times a priority list has been created + 'recent_priority_list_count', + # 2.5 Number of relays that there were in a priority list + # [50, number of relays in the network * 0.05] + 'recent_priority_relay_count', + # 3.6 header: the number of times that sbws has tried to measure any relay, + # since the last 5 days + # This would be the number of times a relays were in a priority list + 'recent_measurement_attempt_count', + # 3.7 header: the number of times that sbws has tried to measure any relay, + # since the last 5 days, but it didn't work + # This should be the number of attempts - number of ResultSuccess - + # something else we don't know yet + # So far is the number of ResultError + 'recent_measurement_failure_count', + # The number of success results should be: + # the number of attempts - the number of failures + # 4.6 header: the number of successful results, created in the last 5 days, + # that were excluded by a filter + # This is the sum of the following 3 + not success results + # 'recent_measurement_exclusion_count', + 'recent_measurement_exclusion_not_distanciated_count', + 'recent_measurement_exclusion_not_recent_count', + 'recent_measurement_exclusion_not_min_num_count', +] +BANDWIDTH_HEADER_KEY_VALUES_INIT = \ + ['earliest_bandwidth', 'generator_started', + 'scanner_country', 'destinations_countries']\ + + STATS_KEYVALUES \ + + BW_HEADER_KEYVALUES_MONITOR + +KEYVALUES_INT = STATS_KEYVALUES + BW_HEADER_KEYVALUES_MONITOR # List of all unordered KeyValues currently being used to generate the file UNORDERED_KEYVALUES = EXTRA_ARG_KEYVALUES + STATS_KEYVALUES + \ - ['latest_bandwidth'] + ['latest_bandwidth'] + \ + BW_HEADER_KEYVALUES_MONITOR # List of all the KeyValues currently being used to generate the file ALL_KEYVALUES = ['version'] + UNORDERED_KEYVALUES + TERMINATOR = '=====' + +# Bandwidth Lines KeyValues +# ========================= # Num header lines in v1.X.X using all the KeyValues NUM_LINES_HEADER_V1 = len(ALL_KEYVALUES) + 2 LINE_TERMINATOR = TERMINATOR + LINE_SEP @@ -50,14 +101,46 @@ BW_KEYVALUE_SEP_V1 = ' ' BW_KEYVALUES_BASIC = ['node_id', 'bw'] BW_KEYVALUES_FILE = BW_KEYVALUES_BASIC + \ ['master_key_ed25519', 'nick', 'rtt', 'time', - 'success', 'error_stream', 'error_circ', 'error_misc'] + 'success', 'error_stream', 'error_circ', 'error_misc', + # Added in #292951 + 'error_second_relay', 'error_destination'] BW_KEYVALUES_EXTRA_BWS = ['bw_median', 'bw_mean', 'desc_bw_avg', 'desc_bw_bur', 'desc_bw_obs_last', 'desc_bw_obs_mean', 'consensus_bandwidth', 'consensus_bandwidth_is_unmeasured'] -BW_KEYVALUES_EXTRA = BW_KEYVALUES_FILE + BW_KEYVALUES_EXTRA_BWS + +# Added in #292951 +BANDWIDTH_LINE_KEY_VALUES_MONITOR = [ + # 1.2 relay: the number of different consensuses, that sbws has seen, + # since the last 5 days, that have this relay + 'relay_in_recent_consensus_count', + # 3.8 relay: the number of times that sbws has tried to measure + # this relay, since the last 5 days + # This would be the number of times a relay was in a priority list (2.6) + # since once it gets measured, it either returns ResultError, + # ResultSuccess or something else happened that we don't know yet + 'relay_recent_measurement_attempt_count', + # 3.9 relay: the number of times that sbws has tried to measure + # this relay, since the last 5 days, but it didn't work + # This should be the number of attempts - number of ResultSuccess - + # something else we don't know yet + # So far is the number of ResultError + 'relay_recent_measurement_failure_count', + # The number of success results should be: + # the number of attempts - the number of failures + # 4.8 relay: the number of successful results, created in the last 5 days, + # that were excluded by a rule, for this relay + # This would be the sum of the following 3 + the number of not success + 'relay_recent_measurement_exclusion_count', + 'relay_recent_measurement_exclusion_not_distanciated', + 'relay_recent_measurement_exclusion_not_recent_count', + 'relay_recent_measurement_exclusion_not_min_num_count', +] +BW_KEYVALUES_EXTRA = BW_KEYVALUES_FILE + BW_KEYVALUES_EXTRA_BWS \ + + BANDWIDTH_LINE_KEY_VALUES_MONITOR BW_KEYVALUES_INT = ['bw', 'rtt', 'success', 'error_stream', - 'error_circ', 'error_misc'] + BW_KEYVALUES_EXTRA_BWS + 'error_circ', 'error_misc'] + BW_KEYVALUES_EXTRA_BWS \ + + BANDWIDTH_LINE_KEY_VALUES_MONITOR BW_KEYVALUES = BW_KEYVALUES_BASIC + BW_KEYVALUES_EXTRA @@ -133,7 +216,7 @@ class V3BWHeader(object): # same as timestamp self.latest_bandwidth = unixts_to_isodt_str(timestamp) [setattr(self, k, v) for k, v in kwargs.items() - if k in EXTRA_ARG_KEYVALUES] + if k in BANDWIDTH_HEADER_KEY_VALUES_INIT] def __str__(self): if self.version.startswith('1.'):

1 0

[sbws/master] chg: docs: !minor Replace anchor link changed in upstream
by juga＠torproject.org 21 Mar '19

21 Mar '19

commit 9c72ea2f90552869cb0be89e0886e75ea97d4054 Author: juga0 <juga(a)riseup.net> Date: Fri Mar 8 15:25:58 2019 +0000 chg: docs: !minor Replace anchor link changed in upstream --- CONTRIBUTING.rst | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/CONTRIBUTING.rst b/CONTRIBUTING.rst index 72a0e78..15da287 100644 --- a/CONTRIBUTING.rst +++ b/CONTRIBUTING.rst @@ -159,7 +159,7 @@ When a PR is being reviewed, new changes might be needed: - If the change does not modify a previous change, create new commits and push. - If the change modifies a previous change and it's small, - `git commit fixup <https://git-scm.com/docs/git-commit#git-commit---fixupltcommitgt>`_ + `git commit fixup <https://git-scm.com/docs/git-commit#Documentation/git-commit.txt---fixupltc…>`_ should be used. When it is agreed that the PR is ready, create a new branch named ``mybranch_02`` and run:

1 0

[sbws/master] new: relaylist: Store relays' consensus timestamps
by juga＠torproject.org 21 Mar '19

21 Mar '19

commit 417ebfa90429d3531e8a5310cf58dff4fc2ff158 Author: juga0 <juga(a)riseup.net> Date: Fri Feb 15 08:39:05 2019 +0000 new: relaylist: Store relays' consensus timestamps for each relay and for the list of relays when fetching new network statuses. To count the number of times a relay was in a consensus and how many consensuses have been seen. In the test network router status entries don't have document attribute. Part of #28566 --- sbws/core/scanner.py | 6 +- sbws/globals.py | 10 +++ sbws/lib/relaylist.py | 178 ++++++++++++++++++++++++++++++++++++++++++++++---- 3 files changed, 179 insertions(+), 15 deletions(-) diff --git a/sbws/core/scanner.py b/sbws/core/scanner.py index 6842372..503f9cc 100644 --- a/sbws/core/scanner.py +++ b/sbws/core/scanner.py @@ -624,8 +624,10 @@ def run_speedtest(args, conf): # Call only once to initialize http_headers settings.init_http_headers(conf.get('scanner', 'nickname'), state['uuid'], str(controller.get_version())) - - rl = RelayList(args, conf, controller) + # To do not have to pass args and conf to RelayList, pass an extra + # argument with the data_period + measurements_period = conf.getint('general', 'data_period') + rl = RelayList(args, conf, controller, measurements_period) cb = CB(args, conf, controller, rl) rd = ResultDump(args, conf) rp = RelayPrioritizer(args, conf, rl, rd) diff --git a/sbws/globals.py b/sbws/globals.py index abd4791..015e857 100644 --- a/sbws/globals.py +++ b/sbws/globals.py @@ -84,6 +84,16 @@ MAX_BW_DIFF_PERC = 50 # Tor already accept lines of any size, but leaving the limit anyway. BW_LINE_SIZE = 1022 +# RelayList, ResultDump, v3bwfile +# For how many seconds in the past the relays and measurements data is keep/ +# considered valid. +# This is currently set by default in config.default.ini as ``date_period``, +# and used in ResultDump and v3bwfile. +# In a future refactor, constants in config.default.ini should be moved here, +# or calculated in settings, so that there's no need to pass the configuration +# to all the functions. +MEASUREMENTS_PERIOD = 5 * 24 * 60 * 60 + # Metadata to send in every requests, so that data servers can know which # scanners are using them. # In Requests these keys are case insensitive. diff --git a/sbws/lib/relaylist.py b/sbws/lib/relaylist.py index b296c06..bee0a14 100644 --- a/sbws/lib/relaylist.py +++ b/sbws/lib/relaylist.py @@ -1,16 +1,52 @@ +import copy +from datetime import datetime, timedelta + from stem.descriptor.router_status_entry import RouterStatusEntryV3 from stem.descriptor.server_descriptor import ServerDescriptor from stem import Flag, DescriptorUnavailable, ControllerError import random -import time import logging from threading import Lock +from ..globals import MEASUREMENTS_PERIOD + log = logging.getLogger(__name__) +def remove_old_consensus_timestamps( + consensus_timestamps, measurements_period=MEASUREMENTS_PERIOD): + """ + Remove the consensus timestamps that are older than period for which + the measurements are keep from a list of consensus_timestamps. + + :param list consensus_timestamps: + :param int measurements_period: + :returns list: a new list of ``consensus_timestamps`` + """ + oldest_date = datetime.utcnow() - timedelta(measurements_period) + new_consensus_timestamps = \ + [t for t in consensus_timestamps if t >= oldest_date] + return new_consensus_timestamps + + +def valid_after_from_network_statuses(network_statuses): + """Obtain the consensus Valid-After datetime from the ``document`` + attribute of a ``stem.descriptor.RouterStatusEntryV3``. + + :param list network_statuses: + returns datetime: + """ + for ns in network_statuses: + document = getattr(ns, 'document', None) + if document: + valid_after = getattr(document, 'valid_after', None) + if valid_after: + return valid_after + return datetime.utcnow().replace(microsecond=0) + + class Relay: - def __init__(self, fp, cont, ns=None, desc=None): + def __init__(self, fp, cont, ns=None, desc=None, timestamp=None): ''' Given a relay fingerprint, fetch all the information about a relay that sbws currently needs and store it in this class. Acts as an abstraction @@ -18,6 +54,9 @@ class Relay: :param str fp: fingerprint of the relay. :param cont: active and valid stem Tor controller connection + + :param datatime timestamp: the timestamp of a consensus + (RouterStatusEntryV3) from which this relay has been obtained. ''' assert isinstance(fp, str) assert len(fp) == 40 @@ -38,6 +77,8 @@ class Relay: self._desc = cont.get_server_descriptor(fp, default=None) except (DescriptorUnavailable, ControllerError) as e: log.exception("Exception trying to get desc %s", e) + self._consensus_timestamps = [] + self._add_consensus_timestamp(timestamp) def _from_desc(self, attr): if not self._desc: @@ -107,6 +148,63 @@ class Relay: return None return key.rstrip('=') + @property + def consensus_valid_after(self): + """Obtain the consensus Valid-After from the document of this relay + network status. + """ + network_status_document = self._from_ns('document') + if network_status_document: + return getattr(network_status_document, 'valid_after', None) + return None + + @property + def last_consensus_timestamp(self): + if len(self._consensus_timestamps) >= 1: + return self._consensus_timestamps[-1] + return None + + def _add_consensus_timestamp(self, timestamp=None): + """Add the consensus timestamp in which this relay is present. + """ + # It is possible to access to the relay's consensensus Valid-After + if self.consensus_valid_after is not None: + # The consensus timestamp list was initialized. + if self.last_consensus_timestamp is not None: + # Valid-After is more recent than the most recent stored + # consensus timestamp. + if self.consensus_valid_after > self.last_consensus_timestamp: + # Add Valid-After + self._consensus_timestamps.append( + self.consensus_valid_after + ) + # The consensus timestamp list was not initialized. + else: + # Add Valid-After + self._consensus_timestamps.append(self.consensus_valid_after) + # If there was already a list the timestamp arg is more recent than + # the most recent timestamp stored, + elif (self.last_consensus_timestamp is not None + and timestamp > self.last_consensus_timestamp): + # Add the arg timestamp. + self._consensus_timestamps.append(timestamp) + # In any other case + else: + # Add the current datetime + self._consensus_timestamps.append( + datetime.utcnow().replace(microsecond=0)) + + def _remove_old_consensus_timestamps( + self, measurements_period=MEASUREMENTS_PERIOD): + self._consensus_timestamps = \ + remove_old_consensus_timestamps( + copy.deepcopy(self._consensus_timestamps, measurements_period) + ) + + def update_consensus_timestamps(self, timestamp=None): + self._add_consensus_timestamp(timestamp) + self._remove_old_consensus_timestamps() + def can_exit_to_port(self, port): """ Returns True if the relay has an exit policy and the policy accepts @@ -129,16 +227,39 @@ class RelayList: transparently in the background. Provides useful interfaces for getting only relays of a certain type. ''' - REFRESH_INTERVAL = 300 # seconds - def __init__(self, args, conf, controller): + def __init__(self, args, conf, controller, + measurements_period=MEASUREMENTS_PERIOD): self._controller = controller self.rng = random.SystemRandom() self._refresh_lock = Lock() + # To track all the consensus seen. + self._consensus_timestamps = [] + # Initialize so that there's no error trying to access to it. + # In future refactor, change to a dictionary, where the keys are + # the relays' fingerprint. + self._relays = [] + # The period of time for which the measurements are keep. + self._measurements_period = measurements_period self._refresh() def _need_refresh(self): - return time.time() >= self._last_refresh + self.REFRESH_INTERVAL + # New consensuses happen every hour. + return datetime.utcnow() >= \ + self.last_consensus_timestamp + timedelta(seconds=60*60) + + @property + def last_consensus_timestamp(self): + """Returns the datetime when the last consensus was obtained.""" + if (getattr(self, "_consensus_timestamps") + and self._consensus_timestamps): + return self._consensus_timestamps[-1] + # If the object was not created from __init__, it won't have + # consensus_timestamps attribute or it might be empty. + # In this case force new update. + # Anytime more than 1h in the past will be old. + self._consensus_timestamps = [] + return datetime.utcnow() - timedelta(seconds=60*61) @property def relays(self): @@ -197,19 +318,50 @@ class RelayList: def _relays_without_flag(self, flag): return [r for r in self.relays if flag not in r.flags] + def _remove_old_consensus_timestamps(self): + self._consensus_timestamps = remove_old_consensus_timestamps( + copy.deepcopy(self._consensus_timestamps), + self._measurements_period + ) + def _init_relays(self): + """Returns a new list of relays that are in the current consensus. + And update the consensus timestamp list with the current one. + + """ c = self._controller - try: - relays = [Relay(ns.fingerprint, c, ns=ns) - for ns in c.get_network_statuses()] - except ControllerError as e: - log.exception("Exception trying to init relays %s", e) - return [] - return relays + # This will get router statuses from this Tor cache, might not be + # updated with the network. + # Change to stem.descriptor.remote in future refactor. + network_statuses = c.get_network_statuses() + new_relays_dict = dict([(r.fingerprint, r) for r in network_statuses]) + + # Find the timestamp of the last consensus. + timestamp = valid_after_from_network_statuses(network_statuses) + self._consensus_timestamps.append(timestamp) + self._remove_old_consensus_timestamps() + # Update the relays that were in the previous consensus with the + # new timestamp + new_relays = [] + relays = copy.deepcopy(self._relays) + for r in relays: + if r.fingerprint in new_relays_dict.keys(): + r.update_consensus_timestamps(timestamp) + new_relays_dict.pop(r.fingerprint) + new_relays.append(r) + + # Add the relays that were not in the previous consensus + # If there was an relay in some older previous consensus, + # it won't get stored, so its previous consensuses are lost, + # but probably this is fine for now to don't make it more complicated. + for fp, ns in new_relays_dict.items(): + r = Relay(ns.fingerprint, c, ns=ns, timestamp=timestamp) + new_relays.append(r) + return new_relays def _refresh(self): + # Set a new list of relays. self._relays = self._init_relays() - self._last_refresh = time.time() def exits_not_bad_allowing_port(self, port): return [r for r in self.exits

1 0

[sbws/master] new: resultdump: Store relay consensus count
by juga＠torproject.org 21 Mar '19

21 Mar '19

commit f2d231ef1325653967ba34c67fc95a17fe9b7c8f Author: juga0 <juga(a)riseup.net> Date: Fri Feb 8 09:13:12 2019 +0000 new: resultdump: Store relay consensus count for every result that is stored. Part of #28566. --- sbws/lib/resultdump.py | 32 ++++++++++++++++++++++++-------- tests/unit/conftest.py | 4 +++- 2 files changed, 27 insertions(+), 9 deletions(-) diff --git a/sbws/lib/resultdump.py b/sbws/lib/resultdump.py index 39aeda4..c081b61 100644 --- a/sbws/lib/resultdump.py +++ b/sbws/lib/resultdump.py @@ -204,7 +204,8 @@ class Result: def __init__(self, fingerprint, nickname, address, master_key_ed25519, average_bandwidth=None, burst_bandwidth=None, observed_bandwidth=None, consensus_bandwidth=None, - consensus_bandwidth_is_unmeasured=None): + consensus_bandwidth_is_unmeasured=None, + consensus_count=None): self.fingerprint = fingerprint self.nickname = nickname self.address = address @@ -215,15 +216,19 @@ class Result: self.consensus_bandwidth = consensus_bandwidth self.consensus_bandwidth_is_unmeasured = \ consensus_bandwidth_is_unmeasured + # The number of times the relay was in a consensus. + self.consensus_count = consensus_count - def __init__(self, relay, circ, dest_url, scanner_nick, t=None): + def __init__(self, relay, circ, dest_url, scanner_nick, t=None, + consensus_count=None): self._relay = Result.Relay(relay.fingerprint, relay.nickname, relay.address, relay.master_key_ed25519, relay.average_bandwidth, relay.burst_bandwidth, relay.observed_bandwidth, relay.consensus_bandwidth, - relay.consensus_bandwidth_is_unmeasured) + relay.consensus_bandwidth_is_unmeasured, + relay.consensus_count) self._circ = circ self._dest_url = dest_url self._scanner = scanner_nick @@ -270,6 +275,11 @@ class Result: return self._relay.master_key_ed25519 @property + def consensus_count(self): + """Number of times the relay was in a consensus.""" + return self._relay.consensus_count + + @property def circ(self): return self._circ @@ -301,6 +311,7 @@ class Result: 'type': self.type, 'scanner': self.scanner, 'version': self.version, + 'consensus_count': self.consensus_count, } @staticmethod @@ -368,7 +379,8 @@ class ResultError(Result): return ResultError( Result.Relay( d['fingerprint'], d['nickname'], d['address'], - d['master_key_ed25519']), + d['master_key_ed25519'], + consensus_count=d.get('consensus_count', None)), d['circ'], d['dest_url'], d['scanner'], msg=d['msg'], t=d['time']) @@ -409,7 +421,8 @@ class ResultErrorCircuit(ResultError): return ResultErrorCircuit( Result.Relay( d['fingerprint'], d['nickname'], d['address'], - d['master_key_ed25519']), + d['master_key_ed25519'], + consensus_count=d.get('consensus_count', None)), d['circ'], d['dest_url'], d['scanner'], msg=d['msg'], t=d['time']) @@ -432,7 +445,8 @@ class ResultErrorStream(ResultError): return ResultErrorStream( Result.Relay( d['fingerprint'], d['nickname'], d['address'], - d['master_key_ed25519']), + d['master_key_ed25519'], + consensus_count=d.get('consensus_count', None)), d['circ'], d['dest_url'], d['scanner'], msg=d['msg'], t=d['time']) @@ -468,7 +482,8 @@ class ResultErrorAuth(ResultError): return ResultErrorAuth( Result.Relay( d['fingerprint'], d['nickname'], d['address'], - d['master_key_ed25519']), + d['master_key_ed25519'], + consensus_count=d.get('consensus_count', None)), d['circ'], d['dest_url'], d['scanner'], msg=d['msg'], t=d['time']) @@ -505,7 +520,8 @@ class ResultSuccess(Result): d['master_key_ed25519'], d['relay_average_bandwidth'], d.get('relay_burst_bandwidth'), d['relay_observed_bandwidth'], d.get('consensus_bandwidth'), - d.get('consensus_bandwidth_is_unmeasured')), + d.get('consensus_bandwidth_is_unmeasured'), + consensus_count=d.get('consensus_count', None)), d['circ'], d['dest_url'], d['scanner'], t=d['time']) diff --git a/tests/unit/conftest.py b/tests/unit/conftest.py index ada1b9b..f5d3d05 100644 --- a/tests/unit/conftest.py +++ b/tests/unit/conftest.py @@ -71,7 +71,8 @@ UNMEASURED = False RELAY1 = Result.Relay(FP1, NICK1, IP1, ED25519, average_bandwidth=AVG_BW, burst_bandwidth=BUR_BW, observed_bandwidth=OBS_BW, consensus_bandwidth=BW, - consensus_bandwidth_is_unmeasured=UNMEASURED) + consensus_bandwidth_is_unmeasured=UNMEASURED, + consensus_count=2) RELAY2 = Result.Relay(FP2, NICK2, IP2, ED25519) RESULT = Result(RELAY1, CIRC12, DEST_URL, SCANNER, t=TIME1) @@ -95,6 +96,7 @@ RELAY_DICT = { "relay_observed_bandwidth": OBS_BW, "consensus_bandwidth": BW, "consensus_bandwidth_is_unmeasured": UNMEASURED, + "consensus_count": 2, } BASE_RESULT_NO_RELAY_DICT = {

1 0

[sbws/master] new: relaylist: Count consensus timestamps
by juga＠torproject.org 21 Mar '19

21 Mar '19

commit 021716b7791bb32caa509681d0cce08258673686 Author: juga0 <juga(a)riseup.net> Date: Fri Feb 15 13:51:19 2019 +0000 new: relaylist: Count consensus timestamps that have been seen for a relay and for the relay list. Part of #28566. --- sbws/lib/relaylist.py | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/sbws/lib/relaylist.py b/sbws/lib/relaylist.py index bee0a14..4cac543 100644 --- a/sbws/lib/relaylist.py +++ b/sbws/lib/relaylist.py @@ -205,6 +205,11 @@ class Relay: self._add_consensus_timestamp(timestamp) self._remove_old_consensus_timestamps() + @property + def consensus_count(self): + """Number of times the relay was in a conensus.""" + return len(self._consensus_timestamps) + def can_exit_to_port(self, port): """ Returns True if the relay has an exit policy and the policy accepts @@ -363,6 +368,11 @@ class RelayList: # Set a new list of relays. self._relays = self._init_relays() + @property + def consensus_count(self): + """Number of times a new consensus was obtained.""" + return len(self._consensus_timestamps) + def exits_not_bad_allowing_port(self, port): return [r for r in self.exits if r.is_exit_not_bad_allowing_port(port)]

1 0

[sbws/master] new: tests: Add test to compare consensus timestamp
by juga＠torproject.org 21 Mar '19

21 Mar '19

commit bd2eea778599d5a66a1ba997f05235f82fbebf4b Author: juga0 <juga(a)riseup.net> Date: Fri Feb 15 13:54:21 2019 +0000 new: tests: Add test to compare consensus timestamp Part of #28566. --- tests/integration/lib/test_relaylist.py | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/tests/integration/lib/test_relaylist.py b/tests/integration/lib/test_relaylist.py index a1a5efb..b3d97c2 100644 --- a/tests/integration/lib/test_relaylist.py +++ b/tests/integration/lib/test_relaylist.py @@ -15,3 +15,8 @@ def test_relay_properties(persistent_launch_tor): assert relay.address == '127.10.0.1' assert relay.master_key_ed25519 == \ 'wLglSEw9/DHfpNrlrqjVRSnGLVWfnm0vYxkryH4aT6Q' + + +def test_relay_list_last_consensus_timestamp(rl): + assert rl.last_consensus_timestamp == \ + rl._relays[0].last_consensus_timestamp

1 0

[sbws/master] new: v3bwfile: Include relay consensus count
by juga＠torproject.org 21 Mar '19

21 Mar '19

commit 6047da5c4a9b55139f66a8974e20bb0c81a95d4b Author: juga0 <juga(a)riseup.net> Date: Fri Feb 8 09:18:14 2019 +0000 new: v3bwfile: Include relay consensus count Include a KeyValue in each bandwidth line with the number of times the relay has been in a consensus in the past days. Part of #28566. --- sbws/lib/v3bwfile.py | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/sbws/lib/v3bwfile.py b/sbws/lib/v3bwfile.py index d6c545a..d827029 100644 --- a/sbws/lib/v3bwfile.py +++ b/sbws/lib/v3bwfile.py @@ -404,6 +404,11 @@ class V3BWLine(object): kwargs['master_key_ed25519'] = results[0].master_key_ed25519 kwargs['time'] = cls.last_time_from_results(results) kwargs.update(cls.result_types_from_results(results)) + consensuses_count = [r.consensus_count for r in results + if getattr(r, 'consensus_count', None)] + if consensuses_count: + consensus_count = max(consensuses_count) + kwargs['relay_in_recent_consensus_count'] = consensus_count success_results = [r for r in results if isinstance(r, ResultSuccess)] if not success_results:

1 0

[sbws/master] new: relaylist: Store total consensus count
by juga＠torproject.org 21 Mar '19

21 Mar '19

commit 1387f91ce469f2c54dfdd067cfd2cd0b3b9099a1 Author: juga0 <juga(a)riseup.net> Date: Fri Feb 15 16:06:53 2019 +0000 new: relaylist: Store total consensus count to the state file. To know the total number of consensus seen in the last days. Part of #28566. --- sbws/core/scanner.py | 2 +- sbws/lib/relaylist.py | 10 +++++++++- 2 files changed, 10 insertions(+), 2 deletions(-) diff --git a/sbws/core/scanner.py b/sbws/core/scanner.py index 503f9cc..73d0b0e 100644 --- a/sbws/core/scanner.py +++ b/sbws/core/scanner.py @@ -627,7 +627,7 @@ def run_speedtest(args, conf): # To do not have to pass args and conf to RelayList, pass an extra # argument with the data_period measurements_period = conf.getint('general', 'data_period') - rl = RelayList(args, conf, controller, measurements_period) + rl = RelayList(args, conf, controller, measurements_period, state) cb = CB(args, conf, controller, rl) rd = ResultDump(args, conf) rp = RelayPrioritizer(args, conf, rl, rd) diff --git a/sbws/lib/relaylist.py b/sbws/lib/relaylist.py index 4cac543..3857718 100644 --- a/sbws/lib/relaylist.py +++ b/sbws/lib/relaylist.py @@ -234,7 +234,7 @@ class RelayList: ''' def __init__(self, args, conf, controller, - measurements_period=MEASUREMENTS_PERIOD): + measurements_period=MEASUREMENTS_PERIOD, state=None): self._controller = controller self.rng = random.SystemRandom() self._refresh_lock = Lock() @@ -246,6 +246,7 @@ class RelayList: self._relays = [] # The period of time for which the measurements are keep. self._measurements_period = measurements_period + self._state = state self._refresh() def _need_refresh(self): @@ -368,6 +369,13 @@ class RelayList: # Set a new list of relays. self._relays = self._init_relays() + log.info("Number of consensuses obtained in the last %s days: %s.", + int(self._measurements_period / 24 / 60 / 60), + self.consensus_count) + # NOTE: blocking, writes to file! + if self._state is not None: + self._state['consensus_count'] = self.consensus_count + @property def consensus_count(self): """Number of times a new consensus was obtained."""

1 0

[sbws/master] new: v3bwfile: Include consensus count header
by juga＠torproject.org 21 Mar '19

21 Mar '19

commit 0bc03df5ee5e8d4ec3ead47d9a9761ea3c8b7978 Author: juga0 <juga(a)riseup.net> Date: Fri Feb 15 16:54:10 2019 +0000 new: v3bwfile: Include consensus count header The total number of consensuses seen in the last days. Part of #28566. --- sbws/lib/v3bwfile.py | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/sbws/lib/v3bwfile.py b/sbws/lib/v3bwfile.py index d827029..fa05fbc 100644 --- a/sbws/lib/v3bwfile.py +++ b/sbws/lib/v3bwfile.py @@ -229,7 +229,9 @@ class V3BWHeader(object): kwargs = dict() latest_bandwidth = cls.latest_bandwidth_from_results(results) earliest_bandwidth = cls.earliest_bandwidth_from_results(results) + # NOTE: Blocking, reads file generator_started = cls.generator_started_from_file(state_fpath) + consensus_count = cls.consensus_count_from_file(state_fpath) timestamp = str(latest_bandwidth) kwargs['latest_bandwidth'] = unixts_to_isodt_str(latest_bandwidth) kwargs['earliest_bandwidth'] = unixts_to_isodt_str(earliest_bandwidth) @@ -240,6 +242,8 @@ class V3BWHeader(object): kwargs['scanner_country'] = scanner_country if destinations_countries is not None: kwargs['destinations_countries'] = destinations_countries + if consensus_count is not None: + kwargs['recent_consensus_count'] = str(consensus_count) h = cls(timestamp, **kwargs) return h @@ -297,6 +301,14 @@ class V3BWHeader(object): return None @staticmethod + def consensus_count_from_file(state_fpath): + state = State(state_fpath) + if 'consensus_count' in state: + return state['consensus_count'] + else: + return None + + @staticmethod def latest_bandwidth_from_results(results): return round(max([r.time for fp in results for r in results[fp]]))

1 0