April 2013 - tor-commits - lists.torproject.org

[ooni-probe/master] Set the default collector via the command line option
by art＠torproject.org 30 Apr '13

30 Apr '13

commit b5c99e327a9c2ef1a2f2b9aa8d41d82a4dcc63d1 Author: Arturo Filastò <art(a)fuffa.org> Date: Thu Feb 28 20:47:05 2013 +0100 Set the default collector via the command line option --- collector | 1 - ooni/oonicli.py | 11 +---------- 2 files changed, 1 insertions(+), 11 deletions(-) diff --git a/collector b/collector deleted file mode 100644 index 0b260a1..0000000 --- a/collector +++ /dev/null @@ -1 +0,0 @@ -httpo://nkvphnp3p6agi5qq.onion diff --git a/ooni/oonicli.py b/ooni/oonicli.py index 93bf8bf..af06ad2 100644 --- a/ooni/oonicli.py +++ b/ooni/oonicli.py @@ -29,8 +29,7 @@ class Options(usage.Options): " files listed on the command line") optFlags = [["help", "h"], - ["resume", "r"], - ["no-default-reporter", "n"]] + ["resume", "r"]] optParameters = [["reportfile", "o", None, "report file name"], ["testdeck", "i", None, @@ -154,14 +153,6 @@ def runWithDirector(): log.msg("for example: ooniprobe -c httpo://nkvphnp3p6agi5qq.onion") sys.exit(1) - # Select one of the baked-in reporters unless the user has requested otherwise - if not global_options['no-default-reporter']: - with open('collector') as f: - reporter_url = random.choice(f.readlines()) - reporter_url = reporter_url.split('#')[0].strip() - oonib_reporter = OONIBReporter(test_details, reporter_url) - reporters.append(oonib_reporter) - log.debug("adding callback for startNetTest") d.addCallback(director.startNetTest, net_test_loader, reporters) d.addCallback(shutdown)

1 0

[ooni-probe/master] Fix typo in test deck before_i_commit
by art＠torproject.org 30 Apr '13

30 Apr '13

commit 7d0f96dcf6ab2ad9f110e6ebffb2af967e129904 Author: Arturo Filastò <art(a)fuffa.org> Date: Fri Mar 1 11:31:42 2013 +0100 Fix typo in test deck before_i_commit --- decks/before_i_commit.testdeck | 2 +- 1 files changed, 1 insertions(+), 1 deletions(-) diff --git a/decks/before_i_commit.testdeck b/decks/before_i_commit.testdeck index ebbd53a..63bac62 100644 --- a/decks/before_i_commit.testdeck +++ b/decks/before_i_commit.testdeck @@ -23,7 +23,7 @@ logfile: reports/before_i_commit.log pcapfile: null reportfile: reports/http_host.yamloo - subargs: [-b, 'http://93.95.227.200', -c, 'http://nkvphnp3p6agi5qq.onion', -f, example_inputs/http_host_file.txt] + subargs: [-b, 'http://93.95.227.200', -f, example_inputs/http_host_file.txt] test_file: nettests/manipulation/http_host.py - options: no-default-reporter: false

1 0

[ooni-probe/master] Remove now superfluos runner unittest
by art＠torproject.org 30 Apr '13

30 Apr '13

commit 52fe425a0c66a0ec2af96ec0b6c44662b68e3b18 Author: Arturo Filastò <art(a)fuffa.org> Date: Wed Feb 27 17:11:31 2013 +0100 Remove now superfluos runner unittest

1 0

[ooni-probe/master] Add more debugging to managers
by art＠torproject.org 30 Apr '13

30 Apr '13

commit 069582c7d14f74851985acec06d999cd2a3b0452 Author: Arturo Filastò <art(a)fuffa.org> Date: Thu Feb 28 20:46:14 2013 +0100 Add more debugging to managers --- ooni/managers.py | 12 ++++++++++-- 1 files changed, 10 insertions(+), 2 deletions(-) diff --git a/ooni/managers.py b/ooni/managers.py index 011e3b0..17b430d 100644 --- a/ooni/managers.py +++ b/ooni/managers.py @@ -1,5 +1,6 @@ import itertools from twisted.internet import defer +from ooni.utils import log def makeIterable(item): """ @@ -28,6 +29,9 @@ class TaskManager(object): The has failed to complete, we append it to the end of the task chain to be re-run once all the currently scheduled tasks have run. """ + log.err("Task %s has failed" % task) + log.exception(failure) + self._active_tasks.remove(task) self.failures.append((failure, task)) @@ -93,6 +97,8 @@ class TaskManager(object): Takes as argument a single task or a task iterable and appends it to the task generator queue. """ + log.debug("Starting this task %s" % repr(task_or_task_iterator)) + iterable = makeIterable(task_or_task_iterator) self._tasks = itertools.chain(self._tasks, iterable) @@ -138,7 +144,8 @@ class MeasurementManager(TaskManager): concurrency = 10 def succeeded(self, result, measurement): - pass + log.debug("Successfully performed measurement %s" % measurement) + log.debug(result) def failed(self, failure, measurement): pass @@ -149,7 +156,8 @@ class ReportEntryManager(TaskManager): concurrency = 20 def succeeded(self, result, task): - pass + log.debug("Successfully performed report %s" % task) + log.debug(result) def failed(self, failure, task): pass

1 0

[ooni-probe/master] Add option to disable the default collector
by art＠torproject.org 30 Apr '13

30 Apr '13

commit 9a30d020ce6bbfdbfc3e56dcc2b89cb932bfdb6f Author: Arturo Filastò <art(a)fuffa.org> Date: Sat Mar 9 03:13:50 2013 +0100 Add option to disable the default collector * More robust startup process --- ooni/errors.py | 2 ++ ooni/oonicli.py | 36 ++++++++++++++++++++++++++---------- 2 files changed, 28 insertions(+), 10 deletions(-) diff --git a/ooni/errors.py b/ooni/errors.py index ac05463..1be42c3 100644 --- a/ooni/errors.py +++ b/ooni/errors.py @@ -147,3 +147,5 @@ class GeoIPDataFilesNotFound(Exception): pass class NoMoreReporters(Exception): pass +class TorNotRunning(Exception): + pass diff --git a/ooni/oonicli.py b/ooni/oonicli.py index 06aa20c..3eae499 100644 --- a/ooni/oonicli.py +++ b/ooni/oonicli.py @@ -10,7 +10,7 @@ from twisted.internet import reactor from twisted.python import usage from twisted.python.util import spewer -from ooni.errors import InvalidOONIBCollectorAddress +from ooni import errors from ooni import config from ooni.director import Director @@ -29,7 +29,9 @@ class Options(usage.Options): " files listed on the command line") optFlags = [["help", "h"], - ["resume", "r"]] + ["resume", "r"], + ["no-collector", "n"] + ] optParameters = [["reportfile", "o", None, "report file name"], ["testdeck", "i", None, @@ -102,6 +104,9 @@ def runWithDirector(): # contains (test_cases, options, cmd_line_options) test_list = [] + if global_options['no-collector']: + log.msg("Not reporting using a collector") + global_options.pop('collector') if global_options['testdeck']: test_deck = yaml.safe_load(open(global_options['testdeck'])) @@ -131,32 +136,43 @@ def runWithDirector(): def director_startup_failed(failure): log.err("Failed to start the director") - log.exception(failure) reactor.stop() # Wait until director has started up (including bootstrapping Tor) before adding tess def post_director_start(_): for net_test_loader in test_list: + collector = global_options['collector'] test_details = net_test_loader.testDetails yaml_reporter = YAMLReporter(test_details) reporters = [yaml_reporter] - if global_options['collector']: + if collector.startswith('httpo') \ + and not config.tor_state: + log.err("Tor does not appear to be running") + log.err("Reporting with the collector %s is not possible") + log.msg("Try with a different collector or disable collector reporting with -n") + raise errors.TorNotRunning + elif collector: + log.msg("Reporting using collector: %s" % + collector) try: oonib_reporter = OONIBReporter(test_details, - global_options['collector']) + collector) reporters.append(oonib_reporter) - except InvalidOONIBCollectorAddress: + except errors.InvalidOONIBCollectorAddress, e: log.err("Invalid format for oonib collector address.") log.msg("Should be in the format http://<collector_address>:<port>") log.msg("for example: ooniprobe -c httpo://nkvphnp3p6agi5qq.onion") - sys.exit(1) + raise e log.debug("adding callback for startNetTest") d.addCallback(director.startNetTest, net_test_loader, reporters) - d.addCallback(shutdown) + d.addBoth(shutdown) - d.addCallback(post_director_start) - d.addErrback(director_startup_failed) + def start(): + d.addCallback(post_director_start) + d.addErrback(director_startup_failed) + + reactor.callWhenRunning(start) reactor.run()

1 0

[ooni-probe/master] Parametrize task timeout and retry count
by art＠torproject.org 30 Apr '13

30 Apr '13

commit 3dc0795a36db25898bafd4c5f0b25f864781f1a6 Author: Arturo Filastò <art(a)fuffa.org> Date: Thu Apr 11 11:58:29 2013 +0200 Parametrize task timeout and retry count --- ooni/director.py | 3 ++- ooni/managers.py | 12 ++++++------ ooni/tasks.py | 5 +++++ ooniprobe.conf.sample | 12 ++++++++++++ 4 files changed, 25 insertions(+), 7 deletions(-) diff --git a/ooni/director.py b/ooni/director.py index bba5eb3..ec4e461 100644 --- a/ooni/director.py +++ b/ooni/director.py @@ -157,7 +157,7 @@ class Director(object): self.failedMeasurements += 1 self.failures.append((failure, measurement)) - return failure + return None def reporterFailed(self, failure, net_test): """ @@ -174,6 +174,7 @@ class Director(object): def netTestDone(self, result, net_test): self.activeNetTests.remove(net_test) + @defer.inlineCallbacks def startNetTest(self, _, net_test_loader, reporters): """ Create the Report for the NetTest and start the report NetTest. diff --git a/ooni/managers.py b/ooni/managers.py index 17b430d..d54c169 100644 --- a/ooni/managers.py +++ b/ooni/managers.py @@ -1,6 +1,8 @@ import itertools + from twisted.internet import defer from ooni.utils import log +from ooni import config def makeIterable(item): """ @@ -139,9 +141,8 @@ class MeasurementManager(TaskManager): NetTest on the contrary is aware of the typology of measurements that it is dispatching as they are logically grouped by test file. """ - # XXX tweak these values - retries = 2 - concurrency = 10 + retries = config.advanced.measuement_retries + concurrency = config.advanced.measurement_concurrency def succeeded(self, result, measurement): log.debug("Successfully performed measurement %s" % measurement) @@ -151,9 +152,8 @@ class MeasurementManager(TaskManager): pass class ReportEntryManager(TaskManager): - # XXX tweak these values - retries = 3 - concurrency = 20 + retries = config.advanced.reporting_retries + concurrency = config.advanced.reporting_concurrency def succeeded(self, result, task): log.debug("Successfully performed report %s" % task) diff --git a/ooni/tasks.py b/ooni/tasks.py index c17a5c1..59a908d 100644 --- a/ooni/tasks.py +++ b/ooni/tasks.py @@ -1,5 +1,6 @@ import time +from ooni import config from twisted.internet import defer, reactor class BaseTask(object): @@ -90,6 +91,8 @@ class TaskWithTimeout(BaseTask): return BaseTask.start(self) class Measurement(TaskWithTimeout): + timeout = config.advanced.measurement_timeout + def __init__(self, test_class, test_method, test_input): """ test_class: @@ -127,6 +130,8 @@ class Measurement(TaskWithTimeout): return d class ReportEntry(TaskWithTimeout): + timeout = config.advanced.reporting_timeout + def __init__(self, reporter, measurement): self.reporter = reporter self.measurement = measurement diff --git a/ooniprobe.conf.sample b/ooniprobe.conf.sample index b488216..aeb96c2 100644 --- a/ooniprobe.conf.sample +++ b/ooniprobe.conf.sample @@ -33,6 +33,18 @@ advanced: # If you do not specify start_tor, you will have to have Tor running and # explicitly set the control port and SOCKS port start_tor: true + # After how many seconds we should give up on a particular measurement + measurement_timeout: 30 + # After how many retries we should give up on a measurement + measurement_retry: 2 + # How many measurments to perform concurrently + measurement_concurrency = 100 + # After how may seconds we should give up reporting + reporting_timeout: 30 + # After how many retries to give up on reporting + reporting_retry: 3 + # How many reports to perform concurrently + reporting_concurrency: 20 tor: #socks_port: 9050 #control_port: 9051

1 0

[ooni-probe/master] Make the starting of ooniprobe via cmd line more robust
by art＠torproject.org 30 Apr '13

30 Apr '13

commit 7d593af7aebd3e96b7702b83ba7f0fbe2b58c141 Author: Arturo Filastò <art(a)fuffa.org> Date: Thu Apr 11 11:58:59 2013 +0200 Make the starting of ooniprobe via cmd line more robust --- ooni/oonicli.py | 25 +++++++++++++++---------- 1 files changed, 15 insertions(+), 10 deletions(-) diff --git a/ooni/oonicli.py b/ooni/oonicli.py index 3eae499..0bc5364 100644 --- a/ooni/oonicli.py +++ b/ooni/oonicli.py @@ -81,6 +81,7 @@ def parseOptions(): try: cmd_line_options.parseOptions() except usage.UsageError, ue: + print cmd_line_options.getUsage() raise SystemExit, "%s: %s" % (sys.argv[0], ue) return dict(cmd_line_options) @@ -106,7 +107,7 @@ def runWithDirector(): test_list = [] if global_options['no-collector']: log.msg("Not reporting using a collector") - global_options.pop('collector') + global_options['collector'] = None if global_options['testdeck']: test_deck = yaml.safe_load(open(global_options['testdeck'])) @@ -136,6 +137,17 @@ def runWithDirector(): def director_startup_failed(failure): log.err("Failed to start the director") + r = failure.trap(errors.TorNotRunning, + errors.InvalidOONIBCollectorAddress) + if r == errors.TorNotRunning: + log.err("Tor does not appear to be running") + log.err("Reporting with the collector %s is not possible" % + global_options['collector']) + log.msg("Try with a different collector or disable collector reporting with -n") + elif r == errors.InvalidOONIBCollectorAddress: + log.err("Invalid format for oonib collector address.") + log.msg("Should be in the format http://<collector_address>:<port>") + log.msg("for example: ooniprobe -c httpo://nkvphnp3p6agi5qq.onion") reactor.stop() # Wait until director has started up (including bootstrapping Tor) before adding tess @@ -147,23 +159,16 @@ def runWithDirector(): yaml_reporter = YAMLReporter(test_details) reporters = [yaml_reporter] - if collector.startswith('httpo') \ + if collector and collector.startswith('httpo') \ and not config.tor_state: - log.err("Tor does not appear to be running") - log.err("Reporting with the collector %s is not possible") - log.msg("Try with a different collector or disable collector reporting with -n") raise errors.TorNotRunning elif collector: - log.msg("Reporting using collector: %s" % - collector) + log.msg("Reporting using collector: %s" % collector) try: oonib_reporter = OONIBReporter(test_details, collector) reporters.append(oonib_reporter) except errors.InvalidOONIBCollectorAddress, e: - log.err("Invalid format for oonib collector address.") - log.msg("Should be in the format http://<collector_address>:<port>") - log.msg("for example: ooniprobe -c httpo://nkvphnp3p6agi5qq.onion") raise e log.debug("adding callback for startNetTest")

1 0

[ooni-probe/master] Pass the IP address of the probe instead of an object
by art＠torproject.org 30 Apr '13

30 Apr '13

commit a5fe45c7ef4bcfd09c9653826a423de0519a6d4d Author: Arturo Filastò <art(a)fuffa.org> Date: Thu Apr 11 12:15:28 2013 +0200 Pass the IP address of the probe instead of an object --- ooni/nettest.py | 4 ++-- ooni/utils/geodata.py | 1 + 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/ooni/nettest.py b/ooni/nettest.py index 0260ca2..41d7f9f 100644 --- a/ooni/nettest.py +++ b/ooni/nettest.py @@ -7,7 +7,7 @@ from twisted.python import usage, reflect from ooni import geoip from ooni.tasks import Measurement -from ooni.utils import log, checkForRoot +from ooni.utils import log, checkForRoot, geodata from ooni import config from ooni import otime @@ -40,7 +40,7 @@ class NetTestLoader(object): config.privacy.includecity): log.msg("We will include some geo data in the report") try: - client_geodata = geodata.IPToLocation(config.probe_ip) + client_geodata = geodata.IPToLocation(config.probe_ip.address) except e.GeoIPDataFilesNotFound: log.err("Unable to find the geoip data files") client_geodata = {'city': None, 'countrycode': None, 'asn': None} diff --git a/ooni/utils/geodata.py b/ooni/utils/geodata.py index 56ce05c..2acfdb0 100644 --- a/ooni/utils/geodata.py +++ b/ooni/utils/geodata.py @@ -14,6 +14,7 @@ except ImportError: log.err("Unable to import pygeoip. We will not be able to run geo IP related measurements") def IPToLocation(ipaddr): + log.debug("looking up %s" % ipaddr) city_file = os.path.join(config.advanced.geoip_data_dir, 'GeoLiteCity.dat') country_file = os.path.join(config.advanced.geoip_data_dir, 'GeoIP.dat') asn_file = os.path.join(config.advanced.geoip_data_dir, 'GeoIPASNum.dat')

1 0

[ooni-probe/master] Make reporting more robust
by art＠torproject.org 30 Apr '13

30 Apr '13

commit f7bfd1198d5b13ad571838372f7316488a6fb3c4 Author: Arturo Filastò <art(a)fuffa.org> Date: Thu Apr 11 11:59:46 2013 +0200 Make reporting more robust Get rid of ridiculous logic that uses a deferred list, instead keep track of the report task via a single deferred. --- ooni/director.py | 6 +- ooni/reporter.py | 166 +++++++++++++++++++++++++++++------------------------- 2 files changed, 93 insertions(+), 79 deletions(-) diff --git a/ooni/director.py b/ooni/director.py index ec4e461..e963049 100644 --- a/ooni/director.py +++ b/ooni/director.py @@ -189,14 +189,16 @@ class Director(object): net_test = NetTest(net_test_loader, report) net_test.director = self - net_test.report.open() + + yield net_test.report.open() self.measurementManager.schedule(net_test.generateMeasurements()) self.activeNetTests.append(net_test) net_test.done.addBoth(self.netTestDone, net_test) net_test.done.addBoth(report.close) - return net_test.done + + yield net_test.done def startSniffing(self): """ Start sniffing with Scapy. Exits if required privileges (root) are not diff --git a/ooni/reporter.py b/ooni/reporter.py index a3811b7..5338f3a 100644 --- a/ooni/reporter.py +++ b/ooni/reporter.py @@ -26,11 +26,10 @@ except ImportError: log.err("Scapy is not installed.") -from ooni.errors import InvalidOONIBCollectorAddress -from ooni.errors import ReportNotCreated, ReportAlreadyClosed +from ooni import errors from ooni import otime -from ooni.utils import pushFilenameStack +from ooni.utils import geodata, pushFilenameStack from ooni.utils.net import BodyReceiver, StringProducer, userAgents from ooni import config @@ -119,7 +118,6 @@ def safe_dump(data, stream=None, **kw): class OReporter(object): def __init__(self, test_details): - self.created = defer.Deferred() self.testDetails = test_details def createReport(self): @@ -190,9 +188,9 @@ class YAMLReporter(OReporter): def _write(self, format_string, *args): if not self._stream: - raise ReportNotCreated + raise errors.ReportNotCreated if self._stream.closed: - raise ReportAlreadyClosed + raise errors.ReportAlreadyClosed s = str(format_string) assert isinstance(s, type('')) if args: @@ -225,8 +223,6 @@ class YAMLReporter(OReporter): self._writeln("###########################################") self.writeReportEntry(self.testDetails) - self.created.callback(self) - return defer.succeed(None) def finish(self): self._stream.close() @@ -260,7 +256,7 @@ class OONIBReporter(OReporter): regexp = '^(http|httpo):\/\/[a-zA-Z0-9\-\.]+(:\d+)?$' if not re.match(regexp, self.collectorAddress) or \ len(self.collectorAddress) < 30: - raise InvalidOONIBCollectorAddress + raise errors.InvalidOONIBCollectorAddress @defer.inlineCallbacks def writeReportEntry(self, entry): @@ -306,7 +302,7 @@ class OONIBReporter(OReporter): self.agent = Agent(reactor, sockshost="127.0.0.1", socksport=int(config.tor.socks_port)) except Exception, e: - yield defer.fail(e) + log.exception(e) url = self.collectorAddress + '/report' @@ -339,11 +335,16 @@ class OONIBReporter(OReporter): bodyProducer=bodyProducer) except ConnectionRefusedError: log.err("Connection to reporting backend failed (ConnectionRefusedError)") - self.created.errback(defer.fail(OONIBReportCreationError)) + raise OONIBReportCreationError + + except errors.HostUnreachable: + log.err("Host is not reachable (HostUnreachable error") + raise OONIBReportCreationError except Exception, e: + log.err("Failed to connect to reporter backend") log.exception(e) - yield defer.fail(OONIBReportCreationError) + raise OONIBReportCreationError # This is a little trix to allow us to unspool the response. We create # a deferred and call yield on it. @@ -355,13 +356,13 @@ class OONIBReporter(OReporter): try: parsed_response = json.loads(backend_response) except Exception, e: + log.err("Failed to parse collector response") log.exception(e) - yield defer.fail(e) + raise OONIBReportCreationError self.reportID = parsed_response['report_id'] self.backendVersion = parsed_response['backend_version'] log.debug("Created report with id %s" % parsed_response['report_id']) - self.created.callback(self) class ReportClosed(Exception): pass @@ -386,42 +387,39 @@ class Report(object): self.done = defer.Deferred() self.reportEntryManager = reportEntryManager + self._reporters_openned = 0 + self._reporters_written = 0 + self._reporters_closed = 0 + def open(self): """ This will create all the reports that need to be created and fires the created callback of the reporter whose report got created. """ - l = [] + all_openned = defer.Deferred() + for reporter in self.reporters[:]: - d = reporter.createReport() - d.addErrback(self.failedOpeningReport, reporter) - reporter.created.addErrback(self.failedOpeningReport, reporter) - l.append(reporter.created) - log.debug("Reporters created: %s" % l) - # Should we consume errors silently? - dl = defer.DeferredList(l) - return dl - def failedOpeningReport(self, failure, reporter): - """ - This errback get's called every time we fail to create a report. - By fail we mean that the number of retries has exceeded. - Once a report has failed to be created with a reporter we give up and - remove the reporter from the list of reporters to write to. - """ - log.err("Failed to open %s reporter, giving up..." % reporter) - log.err("Reporter %s failed, removing from report..." % reporter) - self.reporters.remove(reporter) - # Don't forward the exception unless there are no more reporters - if len(self.reporters) == 0: - log.err("Removed last reporter %s" % reporter) - raise NoMoreReporters + def report_created(result): + self._reporters_openned += 1 + if len(self.reporters) == self._reporters_openned: + all_openned.callback(self._reporters_openned) + + def report_failed(failure): + print "WE HAVE FAILED!" + try: + self.failedOpeningReport(failure, reporter) + except errors.NoMoreReporters, e: + all_openned.errback(defer.fail(e)) + + d = defer.maybeDeferred(reporter.createReport) + d.addErrback(report_failed) + d.addCallback(report_created) + + return all_openned def write(self, measurement): """ - This is a lazy call that will write to all the reporters by waiting on - them to be created. - Will return a deferred that will fire once the report for the specified measurement have been written to all the reporters. @@ -431,41 +429,45 @@ class Report(object): an instance of :class:ooni.tasks.Measurement Returns: - a deferred list that will fire once all the report entries have - been written. + a deferred that will fire once all the report entries have + been written or errbacks when no more reporters """ - l = [] + all_written = defer.Deferred() + for reporter in self.reporters[:]: - report_write_task = ReportEntry(reporter, measurement) - def scheduleWriteReportEntry(result): - self.reportEntryManager.schedule(report_write_task) - - # delay scheduling the task until after the report is created - log.debug("Adding report entry task %s" % report_write_task) - reporter.created.addCallback(scheduleWriteReportEntry) - - # if the write task fails n times, kill the reporter - report_write_task.done.addErrback(self.failedOpeningReport, reporter) - l.append(report_write_task.done) - - # XXX: This seems a bit fragile. - # failedOpeningReport will forward the errback if the remaining - # reporter has failed. If we fireOnOneErrback, this means that - # the caller of report.write is responsible for attaching an - # errback to the returned deferred and handle this case. That - # probably means stopping the net test. - - # Here, fireOnOneErrback means to call the deferredlists errback - # as soon as any of the deferreds return a failure. consumeErrors - # is used to prevent any uncaught failures from raising an - # exception. Alternately we could attach a logger to the errback - # of each deferred and it would have the same effect - - # Probably the better thing to do here would be to add a callback - # to the deferredlist that checks to see if any reporters are left - # and raise an exception if there are no remaining reporters - dl = defer.DeferredList(l,fireOnOneErrback=True, consumeErrors=True) - return dl + def report_written(result): + self._reporters_written += 1 + if len(self.reporters) == self._reporters_written: + all_written.callback(self._reporters_written) + + def report_failed(failure): + log.err("Failed writing report entry") + log.exception(failure) + + report_entry_task = ReportEntry(reporter, measurement) + self.reportEntryManager.schedule(report_entry_task) + + report_entry_task.done.addCallback(report_written) + report_entry_task.done.addErrback(report_failed) + + return all_written + + def failedOpeningReport(self, failure, reporter): + """ + This errback get's called every time we fail to create a report. + By fail we mean that the number of retries has exceeded. + Once a report has failed to be created with a reporter we give up and + remove the reporter from the list of reporters to write to. + """ + log.err("Failed to open %s reporter, giving up..." % reporter) + log.err("Reporter %s failed, removing from report..." % reporter) + log.exception(failure) + self.reporters.remove(reporter) + # Don't forward the exception unless there are no more reporters + if len(self.reporters) == 0: + log.err("Removed last reporter %s" % reporter) + raise errors.NoMoreReporters + return def close(self, _): """ @@ -476,10 +478,20 @@ class Report(object): all the reports have been closed. """ - l = [] + all_closed = defer.Deferred() + for reporter in self.reporters[:]: + def report_closed(result): + self._reporters_closed += 1 + if len(self.reporters) == self._reporters_closed: + all_closed.callback(self._reporters_closed) + + def report_failed(failure): + log.err("Failed closing report") + log.exception(failure) + d = defer.maybeDeferred(reporter.finish) - l.append(d) - dl = defer.DeferredList(l) - return dl + d.addCallback(report_closed) + d.addErrback(report_failed) + return all_closed

1 0

[ooni-probe/master] Final cleanup and bugfixing for task_manager
by art＠torproject.org 30 Apr '13

30 Apr '13

commit 5265969cd60f0bf3c5929d129c513ea1e42ab9f2 Author: Arturo Filastò <art(a)fuffa.org> Date: Thu Apr 11 15:35:31 2013 +0200 Final cleanup and bugfixing for task_manager --- ooni/errors.py | 17 +++++++++++++++++ ooni/nettest.py | 12 ------------ ooni/reporter.py | 38 +++++++++++++++++--------------------- ooni/runner.py | 2 +- 4 files changed, 35 insertions(+), 34 deletions(-) diff --git a/ooni/errors.py b/ooni/errors.py index 1be42c3..613dc9b 100644 --- a/ooni/errors.py +++ b/ooni/errors.py @@ -143,9 +143,26 @@ class InsufficientPrivileges(Exception): class ProbeIPUnknown(Exception): pass + class GeoIPDataFilesNotFound(Exception): pass + class NoMoreReporters(Exception): pass + class TorNotRunning(Exception): pass + +class OONIBReportError(Exception): + pass + +class OONIBReportUpdateError(OONIBReportError): + pass + +class OONIBReportCreationError(OONIBReportError): + pass + +class OONIBTestDetailsLookupError(OONIBReportError): + pass + + diff --git a/ooni/nettest.py b/ooni/nettest.py index 41d7f9f..f0cfa0d 100644 --- a/ooni/nettest.py +++ b/ooni/nettest.py @@ -294,19 +294,7 @@ class NetTest(object): The state for the NetTest is informed of the fact that another task has reached the done state. - - Args: - report_results: - is the list of tuples returned by the self.report.write - :class:twisted.internet.defer.DeferredList - - Returns: - the same deferred list results """ - for report_status, report_result in report_results: - if report_status == False: - self.director.reporterFailed(report_result, self) - self.state.taskDone() if len(self.report.reporters) == 0: diff --git a/ooni/reporter.py b/ooni/reporter.py index 5338f3a..e44db57 100644 --- a/ooni/reporter.py +++ b/ooni/reporter.py @@ -227,18 +227,6 @@ class YAMLReporter(OReporter): def finish(self): self._stream.close() -class OONIBReportError(Exception): - pass - -class OONIBReportUpdateError(OONIBReportError): - pass - -class OONIBReportCreationError(OONIBReportError): - pass - -class OONIBTestDetailsLookupError(OONIBReportError): - pass - class OONIBReporter(OReporter): def __init__(self, test_details, collector_address): self.collectorAddress = collector_address @@ -283,7 +271,7 @@ class OONIBReporter(OReporter): # XXX we must trap this in the runner and make sure to report the # data later. log.err("Error in writing report entry") - raise OONIBReportUpdateError + raise errors.OONIBReportUpdateError @defer.inlineCallbacks def createReport(self): @@ -335,16 +323,17 @@ class OONIBReporter(OReporter): bodyProducer=bodyProducer) except ConnectionRefusedError: log.err("Connection to reporting backend failed (ConnectionRefusedError)") - raise OONIBReportCreationError + #yield defer.fail(OONIBReportCreationError()) + raise errors.OONIBReportCreationError except errors.HostUnreachable: log.err("Host is not reachable (HostUnreachable error") - raise OONIBReportCreationError + raise errors.OONIBReportCreationError except Exception, e: log.err("Failed to connect to reporter backend") log.exception(e) - raise OONIBReportCreationError + raise errors.OONIBReportCreationError # This is a little trix to allow us to unspool the response. We create # a deferred and call yield on it. @@ -358,7 +347,7 @@ class OONIBReporter(OReporter): except Exception, e: log.err("Failed to parse collector response") log.exception(e) - raise OONIBReportCreationError + raise errors.OONIBReportCreationError self.reportID = parsed_response['report_id'] self.backendVersion = parsed_response['backend_version'] @@ -398,19 +387,24 @@ class Report(object): """ all_openned = defer.Deferred() + def are_all_openned(): + if len(self.reporters) == self._reporters_openned: + all_openned.callback(self._reporters_openned) + for reporter in self.reporters[:]: def report_created(result): + log.debug("Created report with %s" % reporter) self._reporters_openned += 1 - if len(self.reporters) == self._reporters_openned: - all_openned.callback(self._reporters_openned) + are_all_openned() def report_failed(failure): - print "WE HAVE FAILED!" try: self.failedOpeningReport(failure, reporter) except errors.NoMoreReporters, e: all_openned.errback(defer.fail(e)) + else: + are_all_openned() d = defer.maybeDeferred(reporter.createReport) d.addErrback(report_failed) @@ -433,6 +427,7 @@ class Report(object): been written or errbacks when no more reporters """ all_written = defer.Deferred() + self._reporters_written = 0 for reporter in self.reporters[:]: def report_written(result): @@ -459,9 +454,10 @@ class Report(object): Once a report has failed to be created with a reporter we give up and remove the reporter from the list of reporters to write to. """ + failure.trap(errors.OONIBReportError) log.err("Failed to open %s reporter, giving up..." % reporter) log.err("Reporter %s failed, removing from report..." % reporter) - log.exception(failure) + #log.exception(failure) self.reporters.remove(reporter) # Don't forward the exception unless there are no more reporters if len(self.reporters) == 0: diff --git a/ooni/runner.py b/ooni/runner.py index abdc256..1c28019 100644 --- a/ooni/runner.py +++ b/ooni/runner.py @@ -11,7 +11,7 @@ from txtorcon import TorConfig from txtorcon import TorState, launch_tor from ooni import config -from ooni.reporter import OONIBReporter, YAMLReporter, OONIBReportError +from ooni.reporter import OONIBReporter, YAMLReporter from ooni.inputunit import InputUnitFactory from ooni.nettest import NetTestCase, NoPostProcessor from ooni.utils import log, checkForRoot, pushFilenameStack

1 0