Merge lp:~blake-rouse/maas/simplestreams-file-handler-connectionwrapper into lp:~maas-committers/maas/trunk

[Blake Rouse]

Replaced TransactionWrapper with ConnectionWrapper.

TransactionWrapper wrapped a LargeFileObject in its own transaction, when reading data into StreamingHttpResponse. This works in theory but causes the main connection for django to be in a transaction, and when a new request is received the Transaction middleware will fail to initialize do to the connection already being in a transaction.

ConnectionWrapper wraps the LargeFileObject in its own connection to the database. This allows the large object to use its own transaction, and not cause any issues with incoming requests to django.

[Jeroen T. Vermeulen (jtv)]

Looks like a sensible change — I even wonder how much need there is for a transaction at all in this case, but I don't see it doing any harm either. I'll have to take some of the Django mechanics on faith (and on its test coverage, obviously). It does make me wonder if we make the same provision when writing the same large object?

I'm approving, but with some notes for small things that I think need fixing.

.

Can you make it clear that _set_up is designed to be idempotent for lazy initialisation? If I had to maintain it under pressure I might miss that, I could imagine myself missing that and breaking the idempotency.

(Pedantic note #1: the docstring for _set_up uses “Setup,” the noun, as a verb.)

.

(Pedantic note #2: the leading comment for TestConnectionWrapper.make_file_for_client looks like the sentence got jumbled.)

.

In test_download_calls__get_new_connection:

        # The connection is not made until first access of the content. Read
        # the content so it called.
        b''.join(response.streaming_content)

Well commented, but mind the typo. :) Our usual way to force the full iteration is to use list():

        list(response.streaming_content)

The same snippet also occurs in test_download_connection_is_not_same_as_django_connections. Short as it is, it may be worth extracting into a function. That way it's a "thing" and you don't need to explain it in the tests, apart from the name and the docstring.

.

In test_download_connection_is_not_same_as_django_connections:

        # This is very import test. The connection that is used by the wrapper
        # cannot be the same as the connection using for all other webrequests,
        # as this will cause transactional errors to occur.

Could you add just a bit more background to those transactional errors? The "to occur" part is just filler to substitute for an actual reason IMHO: you can leave it out without losing anything — except then it becomes more obvious that something is missing.

(Pedantic note #3: I could have sworn that first sentence was written in Asia!)

review: Approve

=== modified file 'src/maasserver/bootresources.py'
--- src/maasserver/bootresources.py	2014-08-26 17:50:29 +0000
+++ src/maasserver/bootresources.py	2014-08-28 13:11:20 +0000
@@ -26,8 +26,10 @@
 from crochet import run_in_reactor
 from django.db import (
     close_old_connections,
+    connections,
     transaction,
     )
+from django.db.utils import load_backend
 from django.http import (
     Http404,
     HttpResponse,
@@ -102,39 +104,63 @@
             arches=['amd64'], subarches=['*'], labels=['release'])
 
 
-class TransactionWrapper:
-    """Wraps `LargeObjectFile` in transaction, so `StreamingHttpResponse`
-    can be used. Once the stream is done, then the transaction is
-    closed.
+class ConnectionWrapper:
+    """Wraps `LargeObjectFile` in a new database connection.
+
+    `StreamingHttpResponse` runs outside of django context, so connection
+    that is not shared by django is needed.
+
+    A new database connection is made at the start of the interation and is
+    closed upon close of wrapper.
     """
 
-    def __init__(self, largeobject):
+    def __init__(self, largeobject, alias="default"):
         self.largeobject = largeobject
-        self._atomic = None
+        self.alias = alias
+        self._connection = None
         self._stream = None
 
+    def _get_new_connection(self):
+        """Create new database connection."""
+        db = connections.databases[self.alias]
+        backend = load_backend(db['ENGINE'])
+        return backend.DatabaseWrapper(db, self.alias)
+
+    def _set_up(self):
+        """Sets up the connection and stream.
+
+        This uses lazy initialisation because it is called each time
+        `next` is called.
+        """
+        if self._connection is None:
+            self._connection = self._get_new_connection()
+            self._connection.connect()
+            self._connection.enter_transaction_management()
+            self._connection.set_autocommit(False)
+        if self._stream is None:
+            self._stream = self.largeobject.open(
+                'rb', connection=self._connection)
+
     def __iter__(self):
         return self
 
     def next(self):
-        if self._atomic is None:
-            self._atomic = transaction.atomic()
-            self._atomic.__enter__()
-        if self._stream is None:
-            self._stream = self.largeobject.open('rb')
+        self._set_up()
         data = self._stream.read(self.largeobject.block_size)
         if len(data) == 0:
             raise StopIteration
         return data
 
     def close(self):
+        """Close the connection and stream."""
         if self._stream is not None:
             self._stream.close()
             self._stream = None
-        if self._atomic is not None:
-            self._atomic.__exit__()
-            self._atomic = None
-        close_old_connections()
+        if self._connection is not None:
+            self._connection.commit()
+            self._connection.leave_transaction_management()
+            self._connection.close()
+            self._connection = None
 
 
 class SimpleStreamsHandler:
@@ -297,7 +323,7 @@
         except BootResourceFile.DoesNotExist:
             raise Http404()
         response = StreamingHttpResponse(
-            TransactionWrapper(rfile.largefile.content),
+            ConnectionWrapper(rfile.largefile.content),
             content_type='application/octet-stream')
         response['Content-Length'] = rfile.largefile.total_size
         return response

=== modified file 'src/maasserver/fields.py'
--- src/maasserver/fields.py	2014-08-13 21:49:35 +0000
+++ src/maasserver/fields.py	2014-08-28 13:11:20 +0000
@@ -444,9 +444,11 @@
     def __iter__(self):
         return self
 
-    def open(self, mode="rwb", new_file=None, using="default"):
+    def open(self, mode="rwb", new_file=None, using="default",
+             connection=None):
         """Opens the internal large object instance."""
-        connection = connections[using]
+        if connection is None:
+            connection = connections[using]
         self._lobject = connection.connection.lobject(
             self.oid, mode, 0, new_file)
         self.oid = self._lobject.oid

=== modified file 'src/maasserver/tests/test_bootresources.py'
--- src/maasserver/tests/test_bootresources.py	2014-08-27 12:26:41 +0000
+++ src/maasserver/tests/test_bootresources.py	2014-08-28 13:11:20 +0000
@@ -21,7 +21,10 @@
 from StringIO import StringIO
 
 from django.core.urlresolvers import reverse
-from django.db import transaction
+from django.db import (
+    connections,
+    transaction,
+    )
 from django.http import StreamingHttpResponse
 from django.test.client import Client
 from maasserver import bootresources
@@ -435,25 +438,23 @@
         version = resource_set.version
         resource_file = resource_set.files.order_by('?')[0]
         filename = resource_file.filename
-        with resource_file.largefile.content.open('rb') as stream:
-            content = stream.read()
         response = self.get_file_client(
             os, arch, subarch, series, version, filename)
         self.assertIsInstance(response, StreamingHttpResponse)
-        self.assertEqual(content, b''.join(response.streaming_content))
-
-
-class TestTransactionWrapper(MAASTestCase):
-    """Tests the use of StreamingHttpResponse(TransactionWrapper(stream)).
+
+
+class TestConnectionWrapper(TransactionTestCase):
+    """Tests the use of StreamingHttpResponse(ConnectionWrapper(stream)).
 
     We do not run this inside of `MAASServerTestCase` as that wraps a
-    transaction around each test. This removes that behavior so we can
-    test that the transaction is remaining open for all of the content.
+    transaction around each test. Since a new connection is created to return
+    the actual content, the transaction to create the data needs be committed.
     """
 
-    def test_download(self):
-        # Do the setup inside of a transaction, as we are running in a test
-        # that doesn't enable transactions per test.
+    def make_file_for_client(self):
+        # Set up the database information inside of a transaction. This is
+        # done so the information is committed. As the new connection needs
+        # to be able to access the data.
         with transaction.atomic():
             os = factory.make_name('os')
             series = factory.make_name('series')
@@ -477,26 +478,81 @@
             largefile = factory.make_large_file(content=content, size=size)
             factory.make_boot_resource_file(
                 resource_set, largefile, filename=filename, filetype=filetype)
-
-        # Outside of the transaction, we run the actual test. The client will
-        # run inside of its own transaction, but once the streaming response
-        # is returned that transaction will be closed.
-        client = Client()
-        response = client.get(
-            reverse(
-                'simplestreams_file_handler', kwargs={
-                    'os': os,
-                    'arch': arch,
-                    'subarch': subarch,
-                    'series': series,
-                    'version': version,
-                    'filename': filename,
-                    }))
-
-        # If TransactionWrapper does not work, then a ProgramError will be
-        # thrown. If it works then content will match.
-        self.assertEqual(content, b''.join(response.streaming_content))
-        self.assertTrue(largefile.content.closed)
+        return content, reverse(
+            'simplestreams_file_handler', kwargs={
+                'os': os,
+                'arch': arch,
+                'subarch': subarch,
+                'series': series,
+                'version': version,
+                'filename': filename,
+                })
+
+    def read_response(self, response):
+        """Read the streaming_content from the response.
+
+        :rtype: bytes
+        """
+        return b''.join(response.streaming_content)
+
+    def test_download_calls__get_new_connection(self):
+        content, url = self.make_file_for_client()
+        mock_get_new_connection = self.patch(
+            bootresources.ConnectionWrapper, '_get_new_connection')
+
+        client = Client()
+        response = client.get(url)
+        self.read_response(response)
+        self.assertThat(mock_get_new_connection, MockCalledOnceWith())
+
+    def test_download_connection_is_not_same_as_django_connections(self):
+        content, url = self.make_file_for_client()
+
+        class AssertConnectionWrapper(bootresources.ConnectionWrapper):
+
+            def _set_up(self):
+                super(AssertConnectionWrapper, self)._set_up()
+                # Capture the created connection
+                AssertConnectionWrapper.connection = self._connection
+
+            def close(self):
+                # Close the stream, but we don't want to close the
+                # connection as the test is testing that the connection is
+                # not the same as the connection django is using for other
+                # webrequests.
+                if self._stream is not None:
+                    self._stream.close()
+                    self._stream = None
+                self._connection = None
+
+        self.patch(
+            bootresources, 'ConnectionWrapper', AssertConnectionWrapper)
+
+        client = Client()
+        response = client.get(url)
+        self.read_response(response)
+
+        # Add cleanup to close the connection, since this was removed from
+        # AssertConnectionWrapper.close method.
+        def close():
+            conn = AssertConnectionWrapper.connection
+            conn.commit()
+            conn.leave_transaction_management()
+            conn.close()
+        self.addCleanup(close)
+
+        # The connection that is used by the wrapper cannot be the same as the
+        # connection be using for all other webrequests. Without this
+        # seperate the transactional middleware will fail to initialize,
+        # because the the connection will already be in a transaction.
+        #
+        # Note: cannot test if DatabaseWrapper != DatabaseWrapper, as it will
+        # report true, because the __eq__ operator only checks if the aliases
+        # are the same. This is checking the underlying connection is
+        # different, which is the important part.
+        self.assertNotEqual(
+            connections["default"].connection,
+            AssertConnectionWrapper.connection.connection)
 
 
 def make_product():