Merge ~ack/maas:postgres-notifier-1817484 into maas:master

[Alberto Donato]

LP: #1817484 - add keepalive for Postgres connections and connect/disconnect events for the listener

[Newell Jensen (newell-jensen)]

LGTM syntax wise. Did not test it myself.

review: Approve

[MAAS Lander (maas-lander)]

UNIT TESTS
-b postgres-notifier-1817484 lp:~ack/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: SUCCESS
COMMIT: b705ee247aafa8ba6d3b891a5c5e1ab6490d8d90

review: Approve

[MAAS Lander (maas-lander)]

UNIT TESTS
-b postgres-notifier-1817484 lp:~ack/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: FAILED
LOG: http://maas-ci-jenkins.internal:8080/job/maas/job/branch-tester/5231/console
COMMIT: 4c1b3df86c1d00efeb63a5c816078e7417099467

review: Needs Fixing

[Blake Rouse (blake-rouse)]

Looks good.

review: Approve

[MAAS Lander (maas-lander)]

LANDING
-b postgres-notifier-1817484 lp:~ack/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: FAILED BUILD
LOG: http://maas-ci-jenkins.internal:8080/job/maas/job/branch-tester/5236/consoleText

[MAAS Lander (maas-lander)]

LANDING
-b postgres-notifier-1817484 lp:~ack/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: FAILED BUILD
LOG: http://maas-ci-jenkins.internal:8080/job/maas/job/branch-tester/5239/consoleText

[MAAS Lander (maas-lander)]

UNIT TESTS
-b postgres-notifier-1817484 lp:~ack/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: FAILED
LOG: http://maas-ci-jenkins.internal:8080/job/maas/job/branch-tester/5240/console
COMMIT: fc8d35a8cbb3dff7b175e2ffc9833082702b9e9b

review: Needs Fixing

[MAAS Lander (maas-lander)]

LANDING
-b postgres-notifier-1817484 lp:~ack/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: FAILED BUILD
LOG: http://maas-ci-jenkins.internal:8080/job/maas/job/branch-tester/5248/consoleText

[MAAS Lander (maas-lander)]

UNIT TESTS
-b postgres-notifier-1817484 lp:~ack/maas/+git/maas into -b master lp:~maas-committers/maas

STATUS: FAILED
LOG: http://maas-ci-jenkins.internal:8080/job/maas/job/branch-tester/5249/console
COMMIT: 2c7acda9b451d0adf21a5e8ae01a49b29ba28c9e

review: Needs Fixing

diff --git a/src/maasserver/config.py b/src/maasserver/config.py
index 0b613bd..1eabe0d 100644
--- a/src/maasserver/config.py
+++ b/src/maasserver/config.py
@@ -58,6 +58,22 @@ class RegionConfiguration(Configuration, metaclass=RegionConfigurationMeta):
         "database_conn_max_age",
         "The lifetime of a database connection, in seconds.",
         Int(if_missing=(5 * 60), accept_python=False, min=0))
+    database_keepalive = ConfigurationOption(
+        "database_keepalive",
+        "Whether keepalive for database connections is enabled.",
+        StringBool(if_missing=True))
+    database_keepalive_idle = ConfigurationOption(
+        "database_keepalive_idle",
+        "Time (in seconds) after which keepalives will be started.",
+        Int(if_missing=15))
+    database_keepalive_interval = ConfigurationOption(
+        "database_keepalive_interval",
+        "Interval (in seconds) between keepaliveds.",
+        Int(if_missing=15))
+    database_keepalive_count = ConfigurationOption(
+        "database_keepalive_count",
+        "Number of keeaplives that can be lost before connection is reset.",
+        Int(if_missing=2))
 
     # Worker options.
     num_workers = ConfigurationOption(
diff --git a/src/maasserver/djangosettings/settings.py b/src/maasserver/djangosettings/settings.py
index 28a0308..172cbaf 100644
--- a/src/maasserver/djangosettings/settings.py
+++ b/src/maasserver/djangosettings/settings.py
@@ -115,6 +115,12 @@ try:
                 'HOST': config.database_host,
                 'PORT': str(config.database_port),
                 'CONN_MAX_AGE': config.database_conn_max_age,
+                'OPTIONS': {
+                    'keepalives': int(config.database_keepalive),
+                    'keepalives_idle': config.database_keepalive_idle,
+                    'keepalives_interval': config.database_keepalive_interval,
+                    'keepalives_count': config.database_keepalive_count
+                }
             }
         }
         DEBUG = config.debug
diff --git a/src/maasserver/listener.py b/src/maasserver/listener.py
index 1e4d669..e30b1c9 100644
--- a/src/maasserver/listener.py
+++ b/src/maasserver/listener.py
@@ -15,6 +15,7 @@ from errno import ENOENT
 from django.db import connections
 from django.db.utils import load_backend
 from provisioningserver.utils.enum import map_enum
+from provisioningserver.utils.events import EventGroup
 from provisioningserver.utils.twisted import (
     callOut,
     suppress,
@@ -98,6 +99,7 @@ class PostgresListenerService(Service, object):
         self.disconnecting = None
         self.registeredChannels = False
         self.log = Logger(__name__, self)
+        self.events = EventGroup("connected", "disconnected")
 
     def startService(self):
         """Start the listener."""
@@ -300,6 +302,7 @@ class PostgresListenerService(Service, object):
             def connect(interval=self.HANDLE_NOTIFY_DELAY):
                 d = deferToThread(self.startConnection)
                 d.addCallback(callOut, deferToThread, self.registerChannels)
+                d.addCallback(callOut, self.events.connected.fire)
                 d.addCallback(callOut, self.startReading)
                 d.addCallback(callOut, self.runHandleNotify, interval)
                 # On failure ensure that the database connection is stopped.
@@ -350,6 +353,7 @@ class PostgresListenerService(Service, object):
             self.log.failure("Connection lost.", reason)
         if self.autoReconnect:
             reactor.callLater(3, self.tryConnection)
+        self.events.disconnected.fire(reason)
 
     def registerChannel(self, channel):
         """Register the channel."""
diff --git a/src/maasserver/management/commands/tests/test_config.py b/src/maasserver/management/commands/tests/test_config.py
index 63dc5ea..e738265 100644
--- a/src/maasserver/management/commands/tests/test_config.py
+++ b/src/maasserver/management/commands/tests/test_config.py
@@ -116,11 +116,16 @@ class TestConfigurationSet(MAASTestCase):
         # Set the option to a random value.
         if self.option == "database_port":
             value = factory.pick_port()
-        elif self.option == "database_conn_max_age":
+        elif self.option in (
+                "database_conn_max_age", "database_keepalive_count",
+                "database_keepalive_interval",
+                "database_keepalive_idle"):
             value = random.randint(0, 60)
         elif self.option == "num_workers":
             value = random.randint(1, 16)
-        elif self.option in ["debug", "debug_queries", "debug_http"]:
+        elif self.option in [
+                "debug", "debug_queries", "debug_http",
+                "database_keepalive"]:
             value = random.choice(['true', 'false'])
         else:
             value = factory.make_name("foobar")
diff --git a/src/maasserver/region_controller.py b/src/maasserver/region_controller.py
index 4087a73..9a41c72 100644
--- a/src/maasserver/region_controller.py
+++ b/src/maasserver/region_controller.py
@@ -119,16 +119,15 @@ class RegionControllerService(Service):
         self.postgresListener.register("sys_dns", self.markDNSForUpdate)
         self.postgresListener.register("sys_proxy", self.markProxyForUpdate)
         self.postgresListener.register("sys_rbac", self.markRBACForUpdate)
-
-        # Update DNS and proxy on first start.
-        self.markDNSForUpdate(None, None)
-        self.markProxyForUpdate(None, None)
-        self.markRBACForUpdate(None, None)
+        self.postgresListener.events.connected.registerHandler(
+            self.markAllForUpdate)
 
     @asynchronous(timeout=FOREVER)
     def stopService(self):
         """Close the controller."""
         super(RegionControllerService, self).stopService()
+        self.postgresListener.events.connected.unregisterHandler(
+            self.markAllForUpdate)
         self.postgresListener.unregister("sys_dns", self.markDNSForUpdate)
         self.postgresListener.unregister("sys_proxy", self.markProxyForUpdate)
         self.postgresListener.unregister("sys_rbac", self.markRBACForUpdate)
@@ -137,6 +136,11 @@ class RegionControllerService(Service):
             self.processing.stop()
             return d
 
+    def markAllForUpdate(self):
+        self.markDNSForUpdate(None, None)
+        self.markProxyForUpdate(None, None)
+        self.markRBACForUpdate(None, None)
+
     def markDNSForUpdate(self, channel, message):
         """Called when the `sys_dns` message is received."""
         self.needsDNSUpdate = True
diff --git a/src/maasserver/tests/test_config.py b/src/maasserver/tests/test_config.py
index 19edf83..1878e10 100644
--- a/src/maasserver/tests/test_config.py
+++ b/src/maasserver/tests/test_config.py
@@ -74,6 +74,10 @@ class TestRegionConfigurationDatabaseOptions(MAASTestCase):
         "database_user": "maas",
         "database_pass": "",
         "database_conn_max_age": 5 * 60,
+        "database_keepalive": True,
+        "database_keepalive_idle": 15,
+        "database_keepalive_interval": 15,
+        "database_keepalive_count": 2
     }
 
     scenarios = tuple(
@@ -89,6 +93,8 @@ class TestRegionConfigurationDatabaseOptions(MAASTestCase):
         config = RegionConfiguration({})
         if isinstance(getattr(config, self.option), str):
             example_value = factory.make_name(self.option)
+        elif self.option == 'database_keepalive':
+            example_value = random.choice(['true', 'false'])
         else:
             example_value = factory.pick_port()
         # Argument values will most often be passed in from the command-line,
@@ -96,7 +102,12 @@ class TestRegionConfigurationDatabaseOptions(MAASTestCase):
         setattr(config, self.option, str(example_value))
         self.assertEqual(example_value, getattr(config, self.option))
         # It's also stored in the configuration database.
-        self.assertEqual({self.option: example_value}, config.store)
+        expected_value = example_value
+        if example_value == 'true':
+            expected_value = True
+        elif expected_value == 'false':
+            expected_value = False
+        self.assertEqual({self.option: expected_value}, config.store)
 
 
 class TestRegionConfigurationWorkerOptions(MAASTestCase):
diff --git a/src/maasserver/tests/test_listener.py b/src/maasserver/tests/test_listener.py
index e58118c..fdbbb10 100644
--- a/src/maasserver/tests/test_listener.py
+++ b/src/maasserver/tests/test_listener.py
@@ -111,6 +111,23 @@ class TestPostgresListenerService(MAASServerTestCase):
 
     @wait_for_reactor
     @inlineCallbacks
+    def test_event_on_connection(self):
+        listener = PostgresListenerService()
+        start_calls = []
+        stop_calls = []
+        listener.events.connected.registerHandler(
+            lambda: start_calls.append(True))
+        listener.events.disconnected.registerHandler(
+            lambda reason: stop_calls.append(reason))
+        yield listener.startService()
+        self.assertEqual(len(start_calls), 1)
+        yield listener.stopService()
+        self.assertEqual(len(stop_calls), 1)
+        [failure] = stop_calls
+        self.assertIsInstance(failure.value, error.ConnectionDone)
+
+    @wait_for_reactor
+    @inlineCallbacks
     def test__handles_missing_system_handler_on_notification(self):
         # Captured notifications from the database will go here.
         notices = DeferredQueue()
diff --git a/src/maasserver/tests/test_region_controller.py b/src/maasserver/tests/test_region_controller.py
index 6b67be8..2c109db 100644
--- a/src/maasserver/tests/test_region_controller.py
+++ b/src/maasserver/tests/test_region_controller.py
@@ -43,6 +43,7 @@ from maastesting.matchers import (
     MockCallsMatch,
     MockNotCalled,
 )
+from provisioningserver.utils.events import Event
 from testtools import ExpectedException
 from testtools.matchers import MatchesStructure
 from twisted.internet import reactor
@@ -92,37 +93,18 @@ class TestRegionControllerService(MAASServerTestCase):
                 call("sys_proxy", service.markProxyForUpdate),
                 call("sys_rbac", service.markRBACForUpdate)))
 
-    @wait_for_reactor
-    @inlineCallbacks
-    def test_startService_calls_markDNSForUpdate(self):
-        listener = MagicMock()
-        service = self.make_service(listener)
-        mock_markDNSForUpdate = self.patch(service, "markDNSForUpdate")
-        service.startService()
-        yield service.processingDefer
-        self.assertThat(mock_markDNSForUpdate, MockCalledOnceWith(None, None))
-
-    @wait_for_reactor
-    @inlineCallbacks
-    def test_startService_calls_markProxyForUpdate(self):
+    def test_startService_markAllForUpdate_on_connect(self):
         listener = MagicMock()
+        listener.events.connected = Event()
         service = self.make_service(listener)
-        mock_markProxyForUpdate = self.patch(service, "markProxyForUpdate")
+        mock_mark_dns_for_update = self.patch(service, 'markDNSForUpdate')
+        mock_mark_rbac_for_update = self.patch(service, 'markRBACForUpdate')
+        mock_mark_proxy_for_update = self.patch(service, 'markProxyForUpdate')
         service.startService()
-        yield service.processingDefer
-        self.assertThat(
-            mock_markProxyForUpdate, MockCalledOnceWith(None, None))
-
-    @wait_for_reactor
-    @inlineCallbacks
-    def test_startService_calls_markRBACForUpdate(self):
-        listener = MagicMock()
-        service = self.make_service(listener)
-        mock_markRBACForUpdate = self.patch(service, "markRBACForUpdate")
-        service.startService()
-        yield service.processingDefer
-        self.assertThat(
-            mock_markRBACForUpdate, MockCalledOnceWith(None, None))
+        service.postgresListener.events.connected.fire()
+        mock_mark_dns_for_update.assert_called_once()
+        mock_mark_rbac_for_update.assert_called_once()
+        mock_mark_proxy_for_update.assert_called_once()
 
     def test_stopService_calls_unregister_on_the_listener(self):
         listener = MagicMock()