Merge lp:~blake-rouse/maas/fix-1680278 into lp:~maas-committers/maas/trunk

[Blake Rouse]

Update the last_ping for the script_set in its own transaction. Only update if the timestamp is later in time and only the latest message in the queue for that node. Uses select_for_update to remove the chance on conflicts.

[Blake Rouse (blake-rouse)]

Still need to do the unit tests. But I ran this with 9 VMs on my machine (which makes it out) and all deployed successfully with no issues or tracebacks.

[Lee Trager (ltrager)]

Two comments below but it looks good!

[Lee Trager (ltrager)]

LGTM! Tested by running testing on 3 machines at once on my laptop. Everything completed successfully and no exception in the logs.

review: Approve

=== modified file 'src/metadataserver/api_twisted.py'
--- src/metadataserver/api_twisted.py	2017-03-29 13:58:22 +0000
+++ src/metadataserver/api_twisted.py	2017-04-13 20:41:06 +0000
@@ -9,13 +9,16 @@
 from datetime import datetime
 import json
 
+from django.db import DatabaseError
 from maasserver.api.utils import extract_oauth_key_from_auth_header
 from maasserver.enum import (
     NODE_STATUS,
     NODE_TYPE,
 )
+from maasserver.models.timestampedmodel import now
 from maasserver.utils.orm import (
     in_transaction,
+    make_serialization_failure,
     transactional,
     TransactionManagementError,
 )
@@ -25,7 +28,10 @@
     add_event_to_node_event_log,
     process_file,
 )
-from metadataserver.models import NodeKey
+from metadataserver.models import (
+    NodeKey,
+    ScriptSet,
+)
 from provisioningserver.logger import LegacyLogger
 from provisioningserver.utils.twisted import deferred
 from twisted.application.internet import TimerService
@@ -225,7 +231,10 @@
                 "outside of a transaction.")
         else:
             # Here we're in a database thread, with a database connection.
-            for message in messages:
+            # We only save the last_ping off the last message in the
+            # list of messages. This removes the number of database saves
+            # required.
+            for idx, message in enumerate(messages):
                 try:
                     self._processMessage(node, message)
                 except:
@@ -233,6 +242,47 @@
                         None,
                         "Failed to process message "
                         "for node: %s" % node.hostname)
+                if idx == len(messages) - 1:
+                    try:
+                        self._updateLastPing(node, message)
+                    except:
+                        log.err(
+                            None,
+                            "Failed to update last ping "
+                            "for node: %s" % node.hostname)
+
+    @transactional
+    def _updateLastPing(self, node, message):
+        """
+        Update the last ping in any status which uses a script_set whenever a
+        node in that status contacts us.
+        """
+        script_set_statuses = {
+            NODE_STATUS.COMMISSIONING: 'current_commissioning_script_set_id',
+            NODE_STATUS.TESTING: 'current_testing_script_set_id',
+            NODE_STATUS.DEPLOYING: 'current_installation_script_set_id',
+        }
+        script_set_property = script_set_statuses.get(node.status)
+        if script_set_property is not None:
+            script_set_id = getattr(node, script_set_property)
+            if script_set_id is not None:
+                try:
+                    script_set = ScriptSet.objects.select_for_update(
+                        nowait=True).get(id=script_set_id)
+                except ScriptSet.DoesNotExist:
+                    # Wierd that it would be deleted, but let not cause a
+                    # stack trace for this error.
+                    pass
+                except DatabaseError:
+                    # select_for_update(nowait=True) failed instantly. Raise
+                    # error so @transactional will retry the whole operation.
+                    raise make_serialization_failure()
+                else:
+                    current_time = now()
+                    if (script_set.last_ping is None or
+                            current_time > script_set.last_ping):
+                        script_set.last_ping = current_time
+                        script_set.save(update_fields=['last_ping'])
 
     @transactional
     def _processMessage(self, node, message):
@@ -278,18 +328,6 @@
         for script_result, args in results.items():
             script_result.store_result(**args)
 
-        # Update the last ping in any status which uses a script_set whenever a
-        # node in that status contacts us.
-        script_set_statuses = {
-            NODE_STATUS.COMMISSIONING: node.current_commissioning_script_set,
-            NODE_STATUS.TESTING: node.current_testing_script_set,
-            NODE_STATUS.DEPLOYING: node.current_installation_script_set,
-        }
-        script_set = script_set_statuses.get(node.status)
-        if script_set is not None:
-            script_set.last_ping = message['timestamp']
-            script_set.save()
-
         # At the end of a top-level event, we change the node status.
         save_node = False
         if self._is_top_level(activity_name) and event_type == 'finish':
@@ -342,16 +380,24 @@
         """Top-level events do not have slashes in their names."""
         return '/' not in activity_name
 
-    @transactional
     def _processMessageNow(self, authorization, message):
-        try:
-            node = NodeKey.objects.get_node_for_key(authorization)
-        except NodeKey.DoesNotExist:
-            # The node that should get this message has already had its owner
-            # cleared or changed and this message cannot be saved.
-            return None
+        # This should be called in a non-reactor thread with a pre-existing
+        # connection (e.g. via deferToDatabase).
+        if in_transaction():
+            raise TransactionManagementError(
+                "_processMessageNow must be called from "
+                "outside of a transaction.")
         else:
-            return self._processMessage(node, message)
+            try:
+                node = transactional(NodeKey.objects.get_node_for_key)(
+                    authorization)
+            except NodeKey.DoesNotExist:
+                # The node that should get this message has already had its
+                # owner cleared or changed and this message cannot be saved.
+                return None
+            else:
+                self._processMessage(node, message)
+                self._updateLastPing(node, message)
 
     @deferred
     def queueMessage(self, authorization, message):
 
=== modified file 'src/metadataserver/tests/test_api_twisted.py'
--- src/metadataserver/tests/test_api_twisted.py	2017-03-29 13:58:22 +0000
+++ src/metadataserver/tests/test_api_twisted.py	2017-04-13 20:41:06 +0000
@@ -11,6 +11,7 @@
 from io import BytesIO
 import json
 from unittest.mock import (
+    call,
     Mock,
     sentinel,
 )
@@ -35,6 +36,7 @@
 from maasserver.utils.threads import deferToDatabase
 from maastesting.matchers import (
     MockCalledOnceWith,
+    MockCallsMatch,
     MockNotCalled,
 )
 from maastesting.testcase import MAASTestCase
@@ -229,20 +231,37 @@
 
     @wait_for_reactor
     @inlineCallbacks
-    def test__processMessages_calls_processMessage(self):
+    def test__processMessageNow_fails_when_in_transaction(self):
+        worker = StatusWorkerService(sentinel.dbtasks)
+        with ExpectedException(TransactionManagementError):
+            yield deferToDatabase(
+                transactional(worker._processMessageNow),
+                sentinel.node, sentinel.message)
+
+    @wait_for_reactor
+    @inlineCallbacks
+    def test__processMessages_calls_processMessage_and_updateLastPing(self):
         worker = StatusWorkerService(sentinel.dbtasks)
         mock_processMessage = self.patch(worker, "_processMessage")
+        mock_updateLastPing = self.patch(worker, "_updateLastPing")
         yield deferToDatabase(
-            worker._processMessages, sentinel.node, [sentinel.message])
+            worker._processMessages, sentinel.node,
+            [sentinel.message1, sentinel.message2])
         self.assertThat(
             mock_processMessage,
-            MockCalledOnceWith(sentinel.node, sentinel.message))
+            MockCallsMatch(
+                call(sentinel.node, sentinel.message1),
+                call(sentinel.node, sentinel.message2)))
+        self.assertThat(
+            mock_updateLastPing,
+            MockCalledOnceWith(sentinel.node, sentinel.message2))
 
     @wait_for_reactor
     @inlineCallbacks
     def test_queueMessages_processes_top_level_message_instantly(self):
         worker = StatusWorkerService(sentinel.dbtasks)
         mock_processMessage = self.patch(worker, "_processMessage")
+        mock_updateLastPing = self.patch(worker, "_updateLastPing")
         message = self.make_message()
         message['event_type'] = 'finish'
         nodes_with_tokens = yield deferToDatabase(self.make_nodes_with_tokens)
@@ -251,6 +270,9 @@
         self.assertThat(
             mock_processMessage,
             MockCalledOnceWith(node, message))
+        self.assertThat(
+            mock_updateLastPing,
+            MockCalledOnceWith(node, message))
 
     @wait_for_reactor
     @inlineCallbacks
@@ -313,6 +335,10 @@
         worker = StatusWorkerService(sentinel.dbtasks)
         worker._processMessage(node, payload)
 
+    def updateLastPing(self, node, payload):
+        worker = StatusWorkerService(sentinel.dbtasks)
+        worker._updateLastPing(node, payload)
+
     def test_status_installation_result_does_not_affect_other_node(self):
         node1 = factory.make_Node(status=NODE_STATUS.DEPLOYING)
         node2 = factory.make_Node(status=NODE_STATUS.DEPLOYING)
@@ -792,7 +818,7 @@
                 break
         self.assertEqual(content, script_result.stdout)
 
-    def test_status_updates_script_status_last_ping(self):
+    def test_updateLastPing_updates_script_status_last_ping(self):
         nodes = {
             status: factory.make_Node(
                 status=status, with_empty_script_sets=True)
@@ -810,7 +836,7 @@
                 'description': 'testing',
                 'timestamp': datetime.utcnow(),
             }
-            self.processMessage(node, payload)
+            self.updateLastPing(node, payload)
             script_set_statuses = {
                 NODE_STATUS.COMMISSIONING: (
                     node.current_commissioning_script_set),