Avoid reuse of stopped monitor threads (#362)

Author: congoamz

aws_advanced_python_wrapper/host_monitoring_plugin.py

@@ -112,6 +112,9 @@ IamPlugin.IsNoneOrEmpty=[IamPlugin] Property "{}" is None or empty.
 LogUtils.Topology=[LogUtils] Topology {}
 
 Monitor.ContextNone=[Monitor] Parameter 'context' should not evaluate to None.
+Monitor.ExceptionInMonitorLoop=[Monitor] Continuing monitoring after an unhandled exception was thrown in the monitoring thread for host '{}'.
+Monitor.StoppingMonitorUnhandledException=[Monitor] Stopping thread after an unhandled exception was thrown in the monitoring thread for host '{}'.
+Monitor.InterruptedException=[Monitor] Monitoring thread for host '{}' was interrupted.
 Monitor.OpenedMonitorConnection=[Monitor] Opened a monitoring connection to '{}'.
 Monitor.OpeningMonitorConnection=[Monitor] Opening a monitoring connection to '{}'.
 

aws_advanced_python_wrapper/resources/aws_advanced_python_wrapper_messages.properties

@@ -26,38 +26,38 @@ class Logger:
     def __init__(self, name: str):
         self.logger = getLogger(name)
 
-    def debug(self, msg, *args):
+    def debug(self, msg, *args, **kwargs):
         if not self.logger.isEnabledFor(logging.DEBUG):
             return
 
         if args is not None and len(args) > 0:
-            self.logger.debug(Messages.get_formatted(msg, *args))
+            self.logger.debug(Messages.get_formatted(msg, *args), **kwargs)
         else:
             try:
-                self.logger.debug(Messages.get(msg))
+                self.logger.debug(Messages.get(msg), **kwargs)
             except NotInResourceBundleError:
-                self.logger.debug(msg)
+                self.logger.debug(msg, **kwargs)
 
-    def error(self, msg, *args):
+    def error(self, msg, *args, **kwargs):
         if not self.logger.isEnabledFor(logging.ERROR):
             return
 
         if args is not None and len(args) > 0:
-            self.logger.error(Messages.get_formatted(msg, *args))
+            self.logger.error(Messages.get_formatted(msg, *args), **kwargs)
         else:
             try:
-                self.logger.error(Messages.get(msg))
+                self.logger.error(Messages.get(msg), **kwargs)
             except NotInResourceBundleError:
-                self.logger.error(msg)
+                self.logger.error(msg, **kwargs)
 
-    def warning(self, msg, *args):
+    def warning(self, msg, *args, **kwargs):
         if not self.logger.isEnabledFor(logging.WARNING):
             return
 
         if args is not None and len(args) > 0:
-            self.logger.warning(Messages.get_formatted(msg, *args))
+            self.logger.warning(Messages.get_formatted(msg, *args), **kwargs)
         else:
             try:
-                self.logger.warning(Messages.get(msg))
+                self.logger.warning(Messages.get(msg), **kwargs)
             except NotInResourceBundleError:
-                self.logger.warning(msg)
+                self.logger.warning(msg, **kwargs)

aws_advanced_python_wrapper/utils/log.py

@@ -156,7 +156,7 @@ class WrapperProperties:
     MONITOR_DISPOSAL_TIME_MS = WrapperProperty(
         "monitor_disposal_time_ms",
         "Interval in milliseconds after which a monitor should be considered inactive and marked for disposal.",
-        60_000)
+        600_000)  # 10 minutes
 
     # Failover
     ENABLE_FAILOVER = WrapperProperty(

aws_advanced_python_wrapper/utils/properties.py

@@ -71,24 +71,12 @@ def mock_plugin_service(mocker, mock_conn, mock_driver_dialect):
 
 
 @pytest.fixture
-def mock_monitor_service(mocker):
-    service = mocker.MagicMock()
-    container = MonitoringThreadContainer()
-
-    def release_monitor(monitor_to_release):
-        container.release_monitor(monitor_to_release)
-
-    service.notify_unused.side_effect = release_monitor
-    return service
-
-
-@pytest.fixture
-def monitor(mock_plugin_service, mock_monitor_service, host_info, props):
+def monitor(mock_plugin_service, host_info, props):
     return Monitor(
         mock_plugin_service,
         host_info,
         props,
-        mock_monitor_service)
+        MonitoringThreadContainer())
 
 
 @pytest.fixture(autouse=True)
@@ -138,7 +126,6 @@ def test_run_host_available(
         monitor,
         host_info,
         monitoring_conn_props,
-        mock_monitor_service,
         mock_plugin_service,
         mock_conn,
         mock_driver_dialect):
@@ -159,7 +146,6 @@ def test_run_host_available(
 
     mock_plugin_service.force_connect.assert_called_once_with(host_info, monitoring_conn_props, None)
     mock_driver_dialect.abort_connection.assert_not_called()
-    mock_monitor_service.notify_unused.assert_called_once()
     mock_conn.close.assert_called_once()
     assert context._is_host_unavailable is False
     assert monitor._is_stopped.is_set()
@@ -172,28 +158,27 @@ def test_run_host_unavailable(
         monitor,
         host_info,
         monitoring_conn_props,
-        mock_monitor_service,
         mock_plugin_service,
         mock_conn,
         mock_driver_dialect):
     remove_delays()
     executor = ThreadPoolExecutor()
     context = MonitoringContext(monitor, mock_conn, mock_driver_dialect, 30, 10, 3)
 
-    mocker.patch("aws_advanced_python_wrapper.host_monitoring_plugin.Monitor._execute_conn_check", side_effect=TimeoutError())
+    mocker.patch(
+        "aws_advanced_python_wrapper.host_monitoring_plugin.Monitor._execute_conn_check", side_effect=TimeoutError())
     monitor.start_monitoring(context)
     future = executor.submit(monitor.run)
     wait([future], 3)
 
     mock_plugin_service.force_connect.assert_called_once_with(host_info, monitoring_conn_props, None)
     mock_driver_dialect.abort_connection.assert_called_once()
-    mock_monitor_service.notify_unused.assert_called_once()
     mock_conn.close.assert_called_once()
     assert context._is_host_unavailable is True
     assert monitor._is_stopped.is_set()
 
 
-def test_run__no_contexts(mocker, mock_monitor_service, monitor):
+def test_run__no_contexts(mocker, monitor):
     host_alias = "host-1"
     container = MonitoringThreadContainer()
     container._monitor_map.put_if_absent(host_alias, monitor)
@@ -209,7 +194,8 @@ def test_run__no_contexts(mocker, mock_monitor_service, monitor):
 
 def test_check_connection_status__valid_then_invalid(mocker, monitor):
     mock_execute_conn_check = mocker.patch(
-        "aws_advanced_python_wrapper.host_monitoring_plugin.Monitor._execute_conn_check", side_effect=[None, TimeoutError()])
+        "aws_advanced_python_wrapper.host_monitoring_plugin.Monitor._execute_conn_check",
+        side_effect=[None, TimeoutError()])
 
     status = monitor._check_host_status(30)  # Initiate a monitoring connection
     assert status.is_available
@@ -221,7 +207,8 @@ def test_check_connection_status__valid_then_invalid(mocker, monitor):
 
 
 def test_check_connection_status__conn_check_throws_exception(mocker, monitor):
-    mocker.patch("aws_advanced_python_wrapper.host_monitoring_plugin.Monitor._execute_conn_check", side_effect=Exception())
+    mocker.patch("aws_advanced_python_wrapper.host_monitoring_plugin.Monitor._execute_conn_check",
+                 side_effect=Exception())
 
     status = monitor._check_host_status(30)  # Initiate a monitoring connection
     assert status.is_available

tests/unit/test_monitor.py

@@ -14,6 +14,7 @@
 
 import psycopg
 import pytest
+from _weakref import ref
 
 from aws_advanced_python_wrapper.errors import AwsWrapperError
 from aws_advanced_python_wrapper.host_monitoring_plugin import (
@@ -39,7 +40,9 @@ def mock_thread_container(mocker):
 
 @pytest.fixture
 def mock_monitor(mocker):
-    return mocker.MagicMock()
+    monitor = mocker.MagicMock()
+    monitor.is_stopped = False
+    return monitor
 
 
 @pytest.fixture
@@ -57,21 +60,22 @@ def thread_container(mock_executor):
 @pytest.fixture
 def monitor_service_mocked_container(mock_plugin_service, mock_thread_container):
     service = MonitorService(mock_plugin_service)
-    service._thread_container = mock_thread_container
+    service._monitor_container = mock_thread_container
     return service
 
 
 @pytest.fixture
 def monitor_service_with_container(mock_plugin_service, thread_container):
     service = MonitorService(mock_plugin_service)
-    service._thread_container = thread_container
+    service._monitor_container = thread_container
     return service
 
 
 @pytest.fixture(autouse=True)
 def setup_teardown(mocker, mock_thread_container, mock_plugin_service, mock_monitor):
     mock_thread_container.get_or_create_monitor.return_value = mock_monitor
-    mocker.patch("aws_advanced_python_wrapper.host_monitoring_plugin.MonitorService._create_monitor", return_value=mock_monitor)
+    mocker.patch(
+        "aws_advanced_python_wrapper.host_monitoring_plugin.MonitorService._create_monitor", return_value=mock_monitor)
 
     yield
 
@@ -91,7 +95,7 @@ def test_start_monitoring(
         mock_conn, aliases, HostInfo("instance-1"), Properties(), 5000, 1000, 3)
 
     mock_monitor.start_monitoring.assert_called_once()
-    assert mock_monitor == monitor_service_mocked_container._cached_monitor
+    assert mock_monitor == monitor_service_mocked_container._cached_monitor()
     assert aliases == monitor_service_mocked_container._cached_monitor_aliases
 
 
@@ -105,14 +109,14 @@ def test_start_monitoring__multiple_calls(monitor_service_with_container, mock_m
 
     assert num_calls == mock_monitor.start_monitoring.call_count
     mock_executor.submit.assert_called_once_with(mock_monitor.run)
-    assert mock_monitor == monitor_service_with_container._cached_monitor
+    assert mock_monitor == monitor_service_with_container._cached_monitor()
     assert aliases == monitor_service_with_container._cached_monitor_aliases
 
 
 def test_start_monitoring__cached_monitor(
         monitor_service_mocked_container, mock_plugin_service, mock_monitor, mock_conn, mock_thread_container):
     aliases = frozenset({"instance-1"})
-    monitor_service_mocked_container._cached_monitor = mock_monitor
+    monitor_service_mocked_container._cached_monitor = ref(mock_monitor)
     monitor_service_mocked_container._cached_monitor_aliases = aliases
 
     monitor_service_mocked_container.start_monitoring(
@@ -121,7 +125,7 @@ def test_start_monitoring__cached_monitor(
     mock_plugin_service.get_dialect.assert_not_called()
     mock_thread_container.get_or_create_monitor.assert_not_called()
     mock_monitor.start_monitoring.assert_called_once()
-    assert mock_monitor == monitor_service_mocked_container._cached_monitor
+    assert mock_monitor == monitor_service_mocked_container._cached_monitor()
     assert aliases == monitor_service_mocked_container._cached_monitor_aliases
 
 
@@ -156,21 +160,9 @@ def test_stop_monitoring_host_connections(mocker, monitor_service_with_container
     mock_monitor2 = mocker.MagicMock()
     thread_container.get_or_create_monitor(aliases1, lambda: mock_monitor1)
     thread_container.get_or_create_monitor(aliases2, lambda: mock_monitor2)
-    reset_resource_spy = mocker.spy(thread_container, 'reset_resource')
 
     monitor_service_with_container.stop_monitoring_host(aliases1)
     mock_monitor1.clear_contexts.assert_called_once()
-    reset_resource_spy.assert_called_once_with(mock_monitor1)
-    reset_resource_spy.reset_mock()
 
     monitor_service_with_container.stop_monitoring_host(aliases2)
     mock_monitor2.clear_contexts.assert_called_once()
-    reset_resource_spy.assert_called_once_with(mock_monitor2)
-
-
-def test_release_resources(mocker, monitor_service_mocked_container):
-    spy = mocker.spy(MonitoringThreadContainer, 'release_instance')
-
-    monitor_service_mocked_container.release_resources()
-    assert monitor_service_mocked_container._thread_container is None
-    spy.assert_called_once()

tests/unit/test_monitor_service.py

@@ -73,33 +73,15 @@ def release_container():
 
 def test_get_or_create_monitor__monitor_created(
         container, mock_monitor_supplier, mock_stopped_monitor, mock_monitor1, mock_executor, mock_future):
-    container._available_monitors.put(mock_stopped_monitor)
-    container._tasks_map.put_if_absent(mock_stopped_monitor, mock_future)
-
     result = container.get_or_create_monitor(frozenset({"alias-1", "alias-2"}), mock_monitor_supplier)
     assert mock_monitor1 == result
-    assert container._tasks_map.get(mock_stopped_monitor) is None
-    mock_future.cancel.assert_called_once()
+
     mock_monitor_supplier.assert_called_once()
     mock_executor.submit.assert_called_once_with(mock_monitor1.run)
     assert mock_monitor1 == container._monitor_map.get("alias-1")
     assert mock_monitor1 == container._monitor_map.get("alias-2")
 
 
-def test_get_or_create_monitor__from_available_monitors(
-        container, mock_monitor_supplier, mock_monitor1, mock_executor, mock_future):
-    container._available_monitors.put(mock_monitor1)
-    container._tasks_map.put_if_absent(mock_monitor1, mock_future)
-
-    result = container.get_or_create_monitor(frozenset({"alias-1", "alias-2"}), mock_monitor_supplier)
-    assert mock_monitor1 == result
-    mock_future.cancel.assert_not_called()
-    mock_monitor_supplier.assert_not_called()
-    mock_executor.submit.assert_not_called()
-    assert mock_monitor1 == container._monitor_map.get("alias-1")
-    assert mock_monitor1 == container._monitor_map.get("alias-2")
-
-
 def test_get_or_create_monitor__from_monitor_map(container, mock_monitor1):
     container._monitor_map.put_if_absent("alias-2", mock_monitor1)
 
@@ -161,16 +143,6 @@ def test_get_or_create_monitor__null_monitor(container, mock_monitor_supplier):
         container.get_or_create_monitor(frozenset({"alias-1"}), mock_monitor_supplier)
 
 
-def test_reset_resource(mock_monitor1, mock_monitor2, container):
-    container._monitor_map.put_if_absent("alias-1", mock_monitor1)
-    container._monitor_map.put_if_absent("alias-2", mock_monitor2)
-
-    container.reset_resource(mock_monitor2)
-    assert mock_monitor1 == container._monitor_map.get("alias-1")
-    assert container._monitor_map.get("alias-2") is None
-    assert mock_monitor2 == container._available_monitors.get()
-
-
 def test_release_monitor(mocker, mock_monitor1, mock_monitor2, container):
     container._monitor_map.put_if_absent("alias-1", mock_monitor1)
     container._monitor_map.put_if_absent("alias-2", mock_monitor2)
@@ -193,18 +165,13 @@ def test_release_instance(mocker, container, mock_monitor1, mock_future):
     container._tasks_map.put_if_absent(mock_monitor1, mock_future)
     mock_future.done.return_value = False
     mock_future.cancelled.return_value = False
-    spy = mocker.spy(container._instance, "_release_resources")
 
     container2 = MonitoringThreadContainer()
     assert container2 is container
-    assert 2 == container._usage_count.get()
 
     container2.release_instance()
-    spy.assert_not_called()
 
-    container.release_instance()
     assert 0 == len(container._monitor_map)
     assert 0 == len(container._tasks_map)
     mock_future.cancel.assert_called_once()
     assert MonitoringThreadContainer._instance is None
-    assert 0 == MonitoringThreadContainer._usage_count.get()

tests/unit/test_monitoring_thread_container.py

@@ -97,14 +97,15 @@ def counter():
 
 @pytest.fixture(autouse=True)
 def verify_concurrency(mock_monitor, mock_executor, mock_future, counter, concurrent_counter):
+    # The ThreadPoolExecutor may have been shut down by a previous test, so we'll need to recreate it here.
+    MonitoringThreadContainer._executor = ThreadPoolExecutor(thread_name_prefix="MonitoringThreadContainerExecutor")
     yield
 
     counter.set(0)
     assert concurrent_counter.get() > 0
     concurrent_counter.set(0)
 
-    while MonitoringThreadContainer._instance is not None:
-        MonitoringThreadContainer.release_instance()
+    MonitoringThreadContainer.release_instance()
 
 
 def test_start_monitoring__connections_to_different_hosts(
@@ -122,12 +123,13 @@ def test_start_monitoring__connections_to_different_hosts(
 
     try:
         mock_create_monitor = mocker.patch(
-            "aws_advanced_python_wrapper.host_monitoring_plugin.MonitorService._create_monitor", return_value=mock_monitor)
+            "aws_advanced_python_wrapper.host_monitoring_plugin.MonitorService._create_monitor",
+            return_value=mock_monitor)
         contexts = start_monitoring(num_conns, services, host_alias_list)
         expected_start_monitoring_calls = [mocker.call(context) for context in contexts]
         mock_monitor.start_monitoring.assert_has_calls(expected_start_monitoring_calls, True)
         assert num_conns == len(MonitoringThreadContainer()._monitor_map)
-        expected_create_monitor_calls = [mocker.call(host_info, props)] * num_conns
+        expected_create_monitor_calls = [mocker.call(host_info, props, MonitoringThreadContainer())] * num_conns
         mock_create_monitor.assert_has_calls(expected_create_monitor_calls)
     finally:
         release_resources(services)
@@ -148,12 +150,13 @@ def test_start_monitoring__connections_to_same_host(
 
     try:
         mock_create_monitor = mocker.patch(
-            "aws_advanced_python_wrapper.host_monitoring_plugin.MonitorService._create_monitor", return_value=mock_monitor)
+            "aws_advanced_python_wrapper.host_monitoring_plugin.MonitorService._create_monitor",
+            return_value=mock_monitor)
         contexts = start_monitoring(num_conns, services, host_alias_list)
         expected_start_monitoring_calls = [mocker.call(context) for context in contexts]
         mock_monitor.start_monitoring.assert_has_calls(expected_start_monitoring_calls, True)
         assert 1 == len(MonitoringThreadContainer()._monitor_map)
-        expected_create_monitor_calls = [mocker.call(host_info, props)]
+        expected_create_monitor_calls = [mocker.call(host_info, props, MonitoringThreadContainer())]
         mock_create_monitor.assert_has_calls(expected_create_monitor_calls)
     finally:
         release_resources(services)