[Author Prev][Author Next][Thread Prev][Thread Next][Author Index][Thread Index]

[tor-commits] [stem/master] CollecTor publication time filtering



commit 6c07fe554ef6a7d40b196d314101153eacfb41af
Author: Damian Johnson <atagar@xxxxxxxxxxxxxx>
Date:   Fri Dec 27 15:13:44 2019 -0800

    CollecTor publication time filtering
    
    When I first wrote this module I played fast and lose with the 'start' and
    'end' parameters, guessing relevance purely based on filenames.
    
    When Karsten added timestamps to the index he better defined the relevant
    timestamp to be a descriptor's publication, which everything except
    microdescriptors contains.
    
    Interestingly, archives can contain publications both before and after its
    filename date. For example...
    
      recent/relay-descriptors/server-descriptors/2019-12-27-22-04-59-server-descriptors
    
      Old filename derived timestamps:
        start: 2019-12-27 22:04:59
        end: 2019-12-27 23:04:59
    
      Index's publication timpestamps:
        start: 2019-12-27 20:30:00
        end: 2019-12-27 22:45:00
    
    If the file was created at 22:04 how does it contain something published at
    22:45?
    
    Regardless, now that the index contains publication times for our purposes
    filenames dates are moot. Our 'start' and 'end' arguments provide the subset
    of archives that reside within the given publication range.
    
    For example, the following downloads descriptors that were published up to two
    hours ago...
    
      recent = datetime.datetime.utcnow() - datetime.timedelta(minutes = 120)
      descriptors = stem.descriptor.collector.get_server_descriptors(start = recent)
    
    If we make this more sophisticated we can demonstrate how many
    descriptors we pull from each archive...
    
      import datetime
      import stem.descriptor.collector
    
      collector = stem.descriptor.collector.get_instance()
      recent = datetime.datetime.utcnow() - datetime.timedelta(minutes = 120)
    
      # This effectively does the same thing as get_server_descriptors(),
      # but in a way we can also determine the full counts.
    
      for f in collector.files('server-descriptor', start = recent):
        all_desc = list(f.read())
        recent_desc = list(f.read(start = recent))
    
        print('%s (%s => %s)' % (f.path, f.start, f.end))
        print('  %i of %i descriptors were published recently' % (len(recent_desc), len(all_desc)))
    
      # Download them again, but through our more prevalently used
      # get_server_descriptors() method.
    
      print('\nIn total there are %i server descriptors published recently' % len(list(collector.get_server_descriptors(start = recent))))
    
      ----------------------------------------------------------------------
    
      % python demo.py
    
      recent/relay-descriptors/server-descriptors/2019-12-27-21-04-59-server-descriptors (2019-12-27 17:59:00 => 2019-12-27 22:13:00)
        3 of 817 descriptors were published recently
      recent/relay-descriptors/server-descriptors/2019-12-27-22-04-59-server-descriptors (2019-12-27 20:30:00 => 2019-12-27 22:45:00)
        297 of 776 descriptors were published recently
      recent/relay-descriptors/server-descriptors/2019-12-27-23-04-59-server-descriptors (2019-12-27 21:49:00 => 2019-12-27 23:01:00)
        800 of 800 descriptors were published recently
    
      In total there are 1100 server descriptors published recently
---
 stem/descriptor/collector.py       | 86 ++++++++++++++++++++++----------------
 test/integ/descriptor/collector.py | 32 +++++++-------
 test/unit/descriptor/collector.py  |  1 +
 3 files changed, 69 insertions(+), 50 deletions(-)

diff --git a/stem/descriptor/collector.py b/stem/descriptor/collector.py
index 05826c49..3ee0e1aa 100644
--- a/stem/descriptor/collector.py
+++ b/stem/descriptor/collector.py
@@ -181,10 +181,10 @@ class File(object):
   :var int size: size of the file
   :var str sha256: file's sha256 checksum
 
-  :var datetime start: beginning of the time range descriptors are for,
-    **None** if this cannot be determined
-  :var datetime end: ending of the time range descriptors are for,
-    **None** if this cannot be determined
+  :var datetime start: first publication within the file, **None** if this
+    cannot be determined
+  :var datetime end: last publication within the file, **None** if this cannot
+    be determined
   :var datetime last_modified: when the file was last modified
   """
 
@@ -206,7 +206,7 @@ class File(object):
     else:
       self.start, self.end = File._guess_time_range(path)
 
-  def read(self, directory = None, descriptor_type = None, document_handler = DocumentHandler.ENTRIES, timeout = None, retries = 3):
+  def read(self, directory = None, descriptor_type = None, start = None, end = None, document_handler = DocumentHandler.ENTRIES, timeout = None, retries = 3):
     """
     Provides descriptors from this archive. Descriptors are downloaded or read
     from disk as follows...
@@ -229,6 +229,8 @@ class File(object):
     :param str descriptor_type: `descriptor type
       <https://metrics.torproject.org/collector.html#data-formats>`_, this is
       guessed if not provided
+    :param datetime.datetime start: publication time to begin with
+    :param datetime.datetime end: publication time to end with
     :param stem.descriptor.__init__.DocumentHandler document_handler: method in
       which to parse a :class:`~stem.descriptor.networkstatus.NetworkStatusDocument`
     :param int timeout: timeout when connection becomes idle, no timeout
@@ -267,7 +269,7 @@ class File(object):
 
         tmp_directory = tempfile.mkdtemp()
 
-        for desc in self.read(tmp_directory, descriptor_type, document_handler, timeout, retries):
+        for desc in self.read(tmp_directory, descriptor_type, start, end, document_handler, timeout, retries):
           yield desc
 
         shutil.rmtree(tmp_directory)
@@ -281,6 +283,17 @@ class File(object):
 
     for desc in stem.descriptor.parse_file(path, document_handler = document_handler):
       if descriptor_type is None or descriptor_type.startswith(desc.type_annotation().name):
+        # TODO: This can filter server and extrainfo times, but other
+        # descriptor types may use other attribute names.
+
+        published = getattr(desc, 'published', None)
+
+        if published:
+          if start and published < start:
+            continue
+          elif end and published > end:
+            continue
+
         yield desc
 
   def download(self, directory, decompress = True, timeout = None, retries = 3, overwrite = False):
@@ -405,8 +418,8 @@ class CollecTor(object):
     Provides server descriptors published during the given time range, sorted
     oldest to newest.
 
-    :param datetime.datetime start: time range to begin with
-    :param datetime.datetime end: time range to end with
+    :param datetime.datetime start: publication time to begin with
+    :param datetime.datetime end: publication time to end with
     :param str cache_to: directory to cache archives into, if an archive is
       available here it is not downloaded
     :param bool bridge: standard descriptors if **False**, bridge if **True**
@@ -424,7 +437,7 @@ class CollecTor(object):
     desc_type = 'server-descriptor' if not bridge else 'bridge-server-descriptor'
 
     for f in self.files(desc_type, start, end):
-      for desc in f.read(cache_to, desc_type, timeout = timeout, retries = retries):
+      for desc in f.read(cache_to, desc_type, start, end, timeout = timeout, retries = retries):
         yield desc
 
   def get_extrainfo_descriptors(self, start = None, end = None, cache_to = None, bridge = False, timeout = None, retries = 3):
@@ -432,8 +445,8 @@ class CollecTor(object):
     Provides extrainfo descriptors published during the given time range,
     sorted oldest to newest.
 
-    :param datetime.datetime start: time range to begin with
-    :param datetime.datetime end: time range to end with
+    :param datetime.datetime start: publication time to begin with
+    :param datetime.datetime end: publication time to end with
     :param str cache_to: directory to cache archives into, if an archive is
       available here it is not downloaded
     :param bool bridge: standard descriptors if **False**, bridge if **True**
@@ -451,13 +464,13 @@ class CollecTor(object):
     desc_type = 'extra-info' if not bridge else 'bridge-extra-info'
 
     for f in self.files(desc_type, start, end):
-      for desc in f.read(cache_to, desc_type, timeout = timeout, retries = retries):
+      for desc in f.read(cache_to, desc_type, start, end, timeout = timeout, retries = retries):
         yield desc
 
   def get_microdescriptors(self, start = None, end = None, cache_to = None, timeout = None, retries = 3):
     """
-    Provides microdescriptors published during the given time range,
-    sorted oldest to newest. Unlike server/extrainfo descriptors,
+    Provides microdescriptors estimated to be published during the given time
+    range, sorted oldest to newest. Unlike server/extrainfo descriptors,
     microdescriptors change very infrequently...
 
     ::
@@ -466,10 +479,11 @@ class CollecTor(object):
       about once per week." -dir-spec section 3.3
 
     CollecTor archives only contain microdescriptors that *change*, so hourly
-    tarballs often contain very few.
+    tarballs often contain very few. Microdescriptors also do not contain
+    their publication timestamp, so this is estimated.
 
-    :param datetime.datetime start: time range to begin with
-    :param datetime.datetime end: time range to end with
+    :param datetime.datetime start: publication time to begin with
+    :param datetime.datetime end: publication time to end with
     :param str cache_to: directory to cache archives into, if an archive is
       available here it is not downloaded
     :param int timeout: timeout for downloading each individual archive when
@@ -484,7 +498,7 @@ class CollecTor(object):
     """
 
     for f in self.files('microdescriptor', start, end):
-      for desc in f.read(cache_to, 'microdescriptor', timeout = timeout, retries = retries):
+      for desc in f.read(cache_to, 'microdescriptor', start, end, timeout = timeout, retries = retries):
         yield desc
 
   def get_consensus(self, start = None, end = None, cache_to = None, document_handler = DocumentHandler.ENTRIES, version = 3, microdescriptor = False, bridge = False, timeout = None, retries = 3):
@@ -492,8 +506,8 @@ class CollecTor(object):
     Provides consensus router status entries published during the given time
     range, sorted oldest to newest.
 
-    :param datetime.datetime start: time range to begin with
-    :param datetime.datetime end: time range to end with
+    :param datetime.datetime start: publication time to begin with
+    :param datetime.datetime end: publication time to end with
     :param str cache_to: directory to cache archives into, if an archive is
       available here it is not downloaded
     :param stem.descriptor.__init__.DocumentHandler document_handler: method in
@@ -528,7 +542,7 @@ class CollecTor(object):
         raise ValueError('Only v2 and v3 router status entries are available (not version %s)' % version)
 
     for f in self.files(desc_type, start, end):
-      for desc in f.read(cache_to, desc_type, document_handler, timeout = timeout, retries = retries):
+      for desc in f.read(cache_to, desc_type, start, end, document_handler, timeout = timeout, retries = retries):
         yield desc
 
   def get_key_certificates(self, start = None, end = None, cache_to = None, timeout = None, retries = 3):
@@ -536,8 +550,8 @@ class CollecTor(object):
     Directory authority key certificates for the given time range,
     sorted oldest to newest.
 
-    :param datetime.datetime start: time range to begin with
-    :param datetime.datetime end: time range to end with
+    :param datetime.datetime start: publication time to begin with
+    :param datetime.datetime end: publication time to end with
     :param str cache_to: directory to cache archives into, if an archive is
       available here it is not downloaded
     :param int timeout: timeout for downloading each individual archive when
@@ -552,7 +566,7 @@ class CollecTor(object):
     """
 
     for f in self.files('dir-key-certificate-3', start, end):
-      for desc in f.read(cache_to, 'dir-key-certificate-3', timeout = timeout, retries = retries):
+      for desc in f.read(cache_to, 'dir-key-certificate-3', start, end, timeout = timeout, retries = retries):
         yield desc
 
   def get_bandwidth_files(self, start = None, end = None, cache_to = None, timeout = None, retries = 3):
@@ -560,8 +574,8 @@ class CollecTor(object):
     Bandwidth authority heuristics for the given time range, sorted oldest to
     newest.
 
-    :param datetime.datetime start: time range to begin with
-    :param datetime.datetime end: time range to end with
+    :param datetime.datetime start: publication time to begin with
+    :param datetime.datetime end: publication time to end with
     :param str cache_to: directory to cache archives into, if an archive is
       available here it is not downloaded
     :param int timeout: timeout for downloading each individual archive when
@@ -576,7 +590,7 @@ class CollecTor(object):
     """
 
     for f in self.files('bandwidth-file', start, end):
-      for desc in f.read(cache_to, 'bandwidth-file', timeout = timeout, retries = retries):
+      for desc in f.read(cache_to, 'bandwidth-file', start, end, timeout = timeout, retries = retries):
         yield desc
 
   def get_exit_lists(self, start = None, end = None, cache_to = None, timeout = None, retries = 3):
@@ -584,8 +598,8 @@ class CollecTor(object):
     `TorDNSEL exit lists <https://www.torproject.org/projects/tordnsel.html.en>`_
     for the given time range, sorted oldest to newest.
 
-    :param datetime.datetime start: time range to begin with
-    :param datetime.datetime end: time range to end with
+    :param datetime.datetime start: publication time to begin with
+    :param datetime.datetime end: publication time to end with
     :param str cache_to: directory to cache archives into, if an archive is
       available here it is not downloaded
     :param int timeout: timeout for downloading each individual archive when
@@ -600,7 +614,7 @@ class CollecTor(object):
     """
 
     for f in self.files('tordnsel', start, end):
-      for desc in f.read(cache_to, 'tordnsel', timeout = timeout, retries = retries):
+      for desc in f.read(cache_to, 'tordnsel', start, end, timeout = timeout, retries = retries):
         yield desc
 
   def index(self, compression = 'best'):
@@ -643,8 +657,8 @@ class CollecTor(object):
     Provides files CollecTor presently has, sorted oldest to newest.
 
     :param str descriptor_type: descriptor type or prefix to retrieve
-    :param datetime.datetime start: time range to begin with
-    :param datetime.datetime end: time range to end with
+    :param datetime.datetime start: publication time to begin with
+    :param datetime.datetime end: publication time to end with
 
     :returns: **list** of :class:`~stem.descriptor.collector.File`
 
@@ -662,10 +676,10 @@ class CollecTor(object):
     matches = []
 
     for f in self._cached_files:
-      if start and (f.start is None or f.start < start):
-        continue
-      elif end and (f.end is None or f.end > end):
-        continue
+      if start and (f.end is None or f.end < start):
+        continue  # only contains descriptors before time range
+      elif end and (f.start is None or f.start > end):
+        continue  # only contains descriptors after time range
 
       if descriptor_type is None or any([desc_type.startswith(descriptor_type) for desc_type in f.types]):
         matches.append(f)
diff --git a/test/integ/descriptor/collector.py b/test/integ/descriptor/collector.py
index ac33490d..3af25c29 100644
--- a/test/integ/descriptor/collector.py
+++ b/test/integ/descriptor/collector.py
@@ -12,7 +12,11 @@ import stem.descriptor.collector
 
 from stem.descriptor import Compression
 
-RECENT = datetime.datetime.utcnow() - datetime.timedelta(minutes = 60)
+# The latest hour may or may not be published, so testing against a time range
+# a little back.
+
+START = datetime.datetime.utcnow() - datetime.timedelta(minutes = 180)
+END = datetime.datetime.utcnow() - datetime.timedelta(minutes = 120)
 
 
 class TestCollector(unittest.TestCase):
@@ -39,23 +43,23 @@ class TestCollector(unittest.TestCase):
   @test.require.only_run_once
   @test.require.online
   def test_downloading_server_descriptors(self):
-    recent_descriptors = list(stem.descriptor.collector.get_server_descriptors(start = RECENT))
+    recent_descriptors = list(stem.descriptor.collector.get_server_descriptors(start = START, end = END))
 
-    if not (300 < len(recent_descriptors) < 800):
-      self.fail('Downloaded %i descriptors, expected 300-800' % len(recent_descriptors))  # 584 on 8/5/19
+    if not (400 < len(recent_descriptors) < 1200):
+      self.fail('Downloaded %i descriptors, expected 400-1200' % len(recent_descriptors))  # 803 on 12/27/19
 
   @test.require.only_run_once
   @test.require.online
   def test_downloading_extrainfo_descriptors(self):
-    recent_descriptors = list(stem.descriptor.collector.get_extrainfo_descriptors(start = RECENT))
+    recent_descriptors = list(stem.descriptor.collector.get_extrainfo_descriptors(start = START, end = END))
 
-    if not (300 < len(recent_descriptors) < 800):
-      self.fail('Downloaded %i descriptors, expected 300-800' % len(recent_descriptors))  # 583 on 8/7/19
+    if not (400 < len(recent_descriptors) < 1200):
+      self.fail('Downloaded %i descriptors, expected 400-1200' % len(recent_descriptors))  # 803 on 12/27/19
 
   @test.require.only_run_once
   @test.require.online
   def test_downloading_microdescriptors(self):
-    recent_descriptors = list(stem.descriptor.collector.get_microdescriptors(start = RECENT))
+    recent_descriptors = list(stem.descriptor.collector.get_microdescriptors(start = START, end = END))
 
     if not (10 < len(recent_descriptors) < 100):
       self.fail('Downloaded %i descriptors, expected 10-100' % len(recent_descriptors))  # 23 on 8/7/19
@@ -63,18 +67,18 @@ class TestCollector(unittest.TestCase):
   @test.require.only_run_once
   @test.require.online
   def test_downloading_consensus_v3(self):
-    recent_descriptors = list(stem.descriptor.collector.get_consensus(start = RECENT))
+    recent_descriptors = list(stem.descriptor.collector.get_consensus(start = START, end = END))
 
-    if not (3000 < len(recent_descriptors) < 10000):
-      self.fail('Downloaded %i descriptors, expected 3000-10000' % len(recent_descriptors))  # 6554 on 8/10/19
+    if not (100 < len(recent_descriptors) < 500):
+      self.fail('Downloaded %i descriptors, expected 100-500' % len(recent_descriptors))  # 316 on 12/27/19
 
   @test.require.only_run_once
   @test.require.online
   def test_downloading_consensus_micro(self):
-    recent_descriptors = list(stem.descriptor.collector.get_consensus(start = RECENT, microdescriptor = True))
+    recent_descriptors = list(stem.descriptor.collector.get_consensus(start = START, end = END, microdescriptor = True))
 
-    if not (3000 < len(recent_descriptors) < 10000):
-      self.fail('Downloaded %i descriptors, expected 3000-10000' % len(recent_descriptors))
+    if not (100 < len(recent_descriptors) < 500):
+      self.fail('Downloaded %i descriptors, expected 100-500' % len(recent_descriptors))  # 316 on 12/27/19
 
   def test_downloading_consensus_invalid_type(self):
     test_values = (
diff --git a/test/unit/descriptor/collector.py b/test/unit/descriptor/collector.py
index 7d80d572..b4ff7636 100644
--- a/test/unit/descriptor/collector.py
+++ b/test/unit/descriptor/collector.py
@@ -181,6 +181,7 @@ class TestCollector(unittest.TestCase):
     ], [f.path for f in collector.files(descriptor_type = 'server-descriptor', end = datetime.datetime(2007, 1, 1))])
 
     self.assertEqual([
+      'archive/relay-descriptors/server-descriptors/server-descriptors-2006-02.tar.xz',
       'archive/relay-descriptors/server-descriptors/server-descriptors-2006-03.tar.xz',
     ], [f.path for f in collector.files(descriptor_type = 'server-descriptor', start = datetime.datetime(2006, 2, 10), end = datetime.datetime(2007, 1, 1))])
 

_______________________________________________
tor-commits mailing list
tor-commits@xxxxxxxxxxxxxxxxxxxx
https://lists.torproject.org/cgi-bin/mailman/listinfo/tor-commits