Merge lp:~facundo/ubuntuone-client/stable-ping-it-baby into lp:ubuntuone-client/stable-1-4

[Facundo Batista]

Client periodically pings the server to check connection.

Client periodically pings the server to check connection.

Tests included. Also added a small helper in the MementoHandler that is
useful for debugging while coding.

Note that I also had to fix an ugly monkeypatching that didn't let me log in debug.

Also note that I hardcoded the ping period. Maybe in the future we would want for it to be configurable, but I wanted to reduce the footprint to ease the SRU. I chose 10 minutes as a balance between time between pings and server load (having 500k connected clients to 16 servers, one ping each 10 mins will be 52 pings per second per server).

[Guillermo Gonzalez (verterok)]

looks good.

review: Approve

[Lucio Torre (lucio.torre)]

looks good

review: Approve

=== modified file 'contrib/testing/testcase.py'
--- contrib/testing/testcase.py	2010-09-07 16:02:04 +0000
+++ contrib/testing/testcase.py	2010-09-24 18:05:58 +0000
@@ -407,6 +407,7 @@
         """ Create the instance, and add a records attribute. """
         logging.Handler.__init__(self, *args, **kwargs)
         self.records = []
+        self.debug = False
 
     def emit(self, record):
         """ Just add the record to self.records. """
@@ -417,6 +418,10 @@
         for rec in self.records:
             if rec.levelno == level and all(m in rec.message for m in msgs):
                 return True
+        if self.debug:
+            recorded =  [(logging.getLevelName(r.levelno), r.message)
+                         for r in self.records]
+            print "Memento messages:", recorded
         return False
 
     def check_debug(self, *msgs):

=== modified file 'tests/syncdaemon/test_action_queue.py'
--- tests/syncdaemon/test_action_queue.py	2010-09-21 14:31:36 +0000
+++ tests/syncdaemon/test_action_queue.py	2010-09-24 18:05:58 +0000
@@ -49,7 +49,7 @@
 from ubuntuone.syncdaemon import states
 from ubuntuone.syncdaemon.action_queue import (
     ActionQueue, ActionQueueCommand, ChangePublicAccess, CreateUDF,
-    DeleteVolume, Download, ListVolumes,
+    DeleteVolume, Download, ListVolumes, ActionQueueProtocol,
     NoisyRequestQueue, RequestQueue, UploadProgressWrapper, Upload,
     CreateShare, GetPublicFiles, GetDelta, GetDeltaFromScratch,
     TRANSFER_PROGRESS_THRESHOLD, Unlink, Move, MakeFile, MakeDir,
@@ -3677,3 +3677,95 @@
         self.assertTrue(self.handler.check_note('Session ID: %r' %
                                                 str(request.session_id)))
 
+
+class ActionQueueProtocolTests(TwistedTestCase):
+    """Test the ACQ class."""
+
+    def setUp(self):
+        """Set up."""
+        # create an AQP and put a factory to it
+        self.aqp = ActionQueueProtocol()
+        obj = Mocker().mock()
+        obj.event_queue.push('SYS_CONNECTION_MADE')
+        self.aqp.factory = obj
+
+        # set up the logger
+        self.handler = MementoHandler()
+        self.handler.setLevel(logging.DEBUG)
+        self.aqp.log.addHandler(self.handler)
+
+    def tearDown(self):
+        """Tear down."""
+        self.aqp.log.removeHandler(self.handler)
+        task = self.aqp._looping_ping
+        if task is not None and task.running:
+            task.stop()
+
+    def test_connection_made(self):
+        """Connection is made."""
+        mocker = Mocker()
+        obj = mocker.mock()
+        obj.event_queue.push('SYS_CONNECTION_MADE')
+        self.aqp.factory = obj
+
+        # test
+        with mocker:
+            self.aqp.connectionMade()
+        self.assertTrue(self.handler.check_info('Connection made.'))
+        self.assertFalse(self.aqp._looping_ping is None)
+
+    def test_connection_lost(self):
+        """Connection is lost."""
+        self.aqp.connectionLost('foo')
+        self.assertTrue(self.handler.check_info(
+                        'Connection lost, reason: foo.'))
+        self.assertTrue(self.aqp._looping_ping is None)
+
+    def test_ping_connection_made_twice(self):
+        """If connection made is called twice, don't create two tasks."""
+        self.aqp.connectionMade()
+        task1 = self.aqp._looping_ping
+        self.aqp.connectionMade()
+        task2 = self.aqp._looping_ping
+        self.assertTrue(task1 is task2)
+
+    def test_ping_connection_lost_twice(self):
+        """If connection lost is called twice, don't stop None."""
+        self.aqp.connectionMade()
+        self.assertFalse(self.aqp._looping_ping is None)
+        self.aqp.connectionLost('reason')
+        self.assertTrue(self.aqp._looping_ping is None)
+        self.aqp.connectionLost('reason')
+        self.assertTrue(self.aqp._looping_ping is None)
+
+    @defer.inlineCallbacks
+    def test_ping_task_calls_ping(self):
+        """The task will call the _do_ping method."""
+        self.aqp._ping_delay = .1
+        deferred = defer.Deferred()
+
+        def fake_ping():
+            """Stop the loop and trigger the deferred test."""
+            self.aqp._looping_ping.stop()
+            deferred.callback(True)
+
+        self.aqp._do_ping = fake_ping
+        self.aqp.connectionMade()
+        yield deferred
+
+    def test_ping_do_ping(self):
+        """Ping and log."""
+        # mock the request
+        mocker = Mocker()
+        req = mocker.mock()
+        req.rtt
+        mocker.result(1.123123)
+        self.aqp.ping = lambda: defer.succeed(req)
+
+        # ping will be called, and req accessed, otherwise mocker will complain
+        with mocker:
+            self.aqp._do_ping()
+
+        # check also the log
+        self.assertTrue(self.handler.check_debug('Ping! rtt: 1.123 segs'))
+

=== modified file 'ubuntuone/syncdaemon/action_queue.py'
--- ubuntuone/syncdaemon/action_queue.py	2010-09-21 14:31:36 +0000
+++ ubuntuone/syncdaemon/action_queue.py	2010-09-24 18:05:58 +0000
@@ -43,7 +43,7 @@
 from urlparse import urljoin
 
 from zope.interface import implements
-from twisted.internet import reactor, defer, threads
+from twisted.internet import reactor, defer, threads, task
 from twisted.internet import error as twisted_errors
 from twisted.names import client as dns_client
 from twisted.python.failure import Failure, DefaultException
@@ -120,48 +120,48 @@
         self.log = logging.getLogger('ubuntuone.SyncDaemon.StorageClient')
         # configure the handler level to be < than DEBUG
         self.log.setLevel(TRACE)
-        self.log.debug = partial(self.log.log, TRACE)
+        self.log_trace = partial(self.log.log, TRACE)
 
     def processMessage(self, message):
         """Wrapper that logs the message and result."""
         # don't log the full message if it's of type BYTES
         if message.type == protocol_pb2.Message.BYTES:
-            self.log.debug('start - processMessage: id: %s, type: %s',
+            self.log_trace('start - processMessage: id: %s, type: %s',
                            message.id, message.type)
         else:
-            self.log.debug('start - processMessage: %s',
+            self.log_trace('start - processMessage: %s',
                           str(message).replace("\n", " "))
         if message.id in self.requests:
             req = self.requests[message.id]
             req.deferred.addCallbacks(self.log_success, self.log_error)
         result = ThrottlingStorageClient.processMessage(self, message)
-        self.log.debug('end - processMessage: id: %s - result: %s',
+        self.log_trace('end - processMessage: id: %s - result: %s',
                        message.id, result)
         return result
 
     def log_error(self, failure):
         """Logging errback for requests."""
-        self.log.debug('request error: %s', failure)
+        self.log_trace('request error: %s', failure)
         return failure
 
     def log_success(self, result):
         """Logging callback for requests."""
-        self.log.debug('request finished: %s', result)
+        self.log_trace('request finished: %s', result)
         if getattr(result, '__dict__', None):
-            self.log.debug('result.__dict__: %s', result.__dict__)
+            self.log_trace('result.__dict__: %s', result.__dict__)
         return result
 
     def sendMessage(self, message):
         """Wrapper that logs the message and result."""
         # don't log the full message if it's of type BYTES
         if message.type == protocol_pb2.Message.BYTES:
-            self.log.debug('start - sendMessage: id: %s, type: %s',
+            self.log_trace('start - sendMessage: id: %s, type: %s',
                            message.id, message.type)
         else:
-            self.log.debug('start - sendMessage: %s',
+            self.log_trace('start - sendMessage: %s',
                           str(message).replace("\n", " "))
         result = ThrottlingStorageClient.sendMessage(self, message)
-        self.log.debug('end - sendMessage: id: %s', message.id)
+        self.log_trace('end - sendMessage: id: %s', message.id)
         return result
 
 
@@ -169,15 +169,29 @@
     """This is the Action Queue version of the StorageClient protocol."""
 
     factory = None
+    _looping_ping = None
+    _ping_delay = 600  # 10 minutes
 
     def connectionMade(self):
         """A new connection was made."""
         self.log.info('Connection made.')
         self.factory.event_queue.push('SYS_CONNECTION_MADE')
+        if self._looping_ping is None:
+            self._looping_ping = task.LoopingCall(self._do_ping)
+            self._looping_ping.start(self._ping_delay, now=False)
 
     def connectionLost(self, reason):
         """The connection was lost."""
         self.log.info('Connection lost, reason: %s.', reason)
+        if self._looping_ping is not None:
+            self._looping_ping.stop()
+            self._looping_ping = None
+
+    @defer.inlineCallbacks
+    def _do_ping(self):
+        """Ping the server just to use the network."""
+        req = yield self.ping()
+        self.log.debug("Ping! rtt: %.3f segs", req.rtt)
 
 
 class Marker(str):