feat: efm v2 support

# Conflicts:
#	aws_advanced_python_wrapper/plugin_service.py
#	aws_advanced_python_wrapper/utils/atomic.py

# Conflicts:
#	docs/using-the-python-driver/UsingThePythonDriver.md

Author: JuanLeee

aws_advanced_python_wrapper/host_monitoring_plugin.py

@@ -24,7 +24,6 @@
 
 from concurrent.futures import (Executor, Future, ThreadPoolExecutor,
                                 TimeoutError)
-from copy import copy
 from dataclasses import dataclass
 from queue import Queue
 from threading import Event, Lock, RLock
@@ -43,6 +42,7 @@
 from aws_advanced_python_wrapper.utils.notifications import (
     ConnectionEvent, HostEvent, OldConnectionSuggestedAction)
 from aws_advanced_python_wrapper.utils.properties import (Properties,
+                                                          PropertiesUtils,
                                                           WrapperProperties)
 from aws_advanced_python_wrapper.utils.rdsutils import RdsUtils
 from aws_advanced_python_wrapper.utils.telemetry.telemetry import (
@@ -335,7 +335,6 @@ class Monitor:
     _DEFAULT_CONNECT_TIMEOUT_SEC = 10
     _INACTIVE_SLEEP_MS = 100
     _MIN_HOST_CHECK_TIMEOUT_MS = 3000
-    _MONITORING_PROPERTY_PREFIX = "monitoring-"
     _QUERY = "SELECT 1"
 
     def __init__(
@@ -511,19 +510,15 @@ def _check_host_status(self, host_check_timeout_ms: int) -> HostStatus:
         try:
             driver_dialect = self._plugin_service.driver_dialect
             if self._monitoring_conn is None or driver_dialect.is_closed(self._monitoring_conn):
-                props_copy: Properties = copy(self._props)
-                for key, value in self._props.items():
-                    if key.startswith(Monitor._MONITORING_PROPERTY_PREFIX):
-                        props_copy[key[len(Monitor._MONITORING_PROPERTY_PREFIX):len(key)]] = value
-                        props_copy.pop(key, None)
+                monitoring_properties: Properties = PropertiesUtils.create_monitoring_properties(self._props)
 
                 # Set a default connect timeout if the user hasn't configured one
-                if props_copy.get(WrapperProperties.CONNECT_TIMEOUT_SEC.name, None) is None:
-                    props_copy[WrapperProperties.CONNECT_TIMEOUT_SEC.name] = Monitor._DEFAULT_CONNECT_TIMEOUT_SEC
+                if monitoring_properties.get(WrapperProperties.CONNECT_TIMEOUT_SEC.name, None) is None:
+                    monitoring_properties[WrapperProperties.CONNECT_TIMEOUT_SEC.name] = Monitor._DEFAULT_CONNECT_TIMEOUT_SEC
 
                 logger.debug("Monitor.OpeningMonitorConnection", self._host_info.url)
                 start_ns = perf_counter_ns()
-                self._monitoring_conn = self._plugin_service.force_connect(self._host_info, props_copy, None)
+                self._monitoring_conn = self._plugin_service.force_connect(self._host_info, monitoring_properties, None)
                 logger.debug("Monitor.OpenedMonitorConnection", self._host_info.url)
                 return Monitor.HostStatus(True, perf_counter_ns() - start_ns)
 

aws_advanced_python_wrapper/host_monitoring_v2_plugin.py

@@ -75,6 +75,8 @@
     HostListProviderService, StaticHostListProvider)
 from aws_advanced_python_wrapper.host_monitoring_plugin import \
     HostMonitoringPluginFactory
+from aws_advanced_python_wrapper.host_monitoring_v2_plugin import \
+    HostMonitoringV2PluginFactory
 from aws_advanced_python_wrapper.hostinfo import HostInfo, HostRole
 from aws_advanced_python_wrapper.iam_plugin import IamAuthPluginFactory
 from aws_advanced_python_wrapper.plugin import CanReleaseResources
@@ -755,6 +757,7 @@ class PluginManager(CanReleaseResources):
         "aws_secrets_manager": AwsSecretsManagerPluginFactory,
         "aurora_connection_tracker": AuroraConnectionTrackerPluginFactory,
         "host_monitoring": HostMonitoringPluginFactory,
+        "host_monitoring_v2": HostMonitoringV2PluginFactory,
         "failover": FailoverPluginFactory,
         "read_write_splitting": ReadWriteSplittingPluginFactory,
         "fastest_response_strategy": FastestResponseStrategyPluginFactory,
@@ -783,6 +786,7 @@ class PluginManager(CanReleaseResources):
         ReadWriteSplittingPluginFactory: 300,
         FailoverPluginFactory: 400,
         HostMonitoringPluginFactory: 500,
+        HostMonitoringV2PluginFactory: 510,
         BlueGreenPluginFactory: 550,
         FastestResponseStrategyPluginFactory: 600,
         IamAuthPluginFactory: 700,

aws_advanced_python_wrapper/plugin_service.py

@@ -176,6 +176,28 @@ HostMonitoringPlugin.ConfigurationNotSupported=[HostMonitoringPlugin] Aborting c
 HostMonitoringPlugin.UnableToIdentifyConnection=[HostMonitoringPlugin] Unable to identify the connected database instance: '{}', please ensure the correct host list provider is specified. The host list provider in use is: '{}'.
 HostMonitoringPlugin.UnavailableHost=[HostMonitoringPlugin] Host '{}' is unavailable.
 
+HostMonitoringV2Plugin.ActivatedMonitoring=[HostMonitoringV2Plugin] Executing method '{}', monitoring is activated.
+HostMonitoringV2Plugin.ClusterEndpointHostInfo=[HostMonitoringV2Plugin] The HostInfo to monitor is associated with a cluster endpoint. The plugin will attempt to identify the connected database instance.
+HostMonitoringV2Plugin.ErrorIdentifyingConnection=[HostMonitoringV2Plugin] An error occurred while identifying the connection database instance: '{}'.
+HostMonitoringV2Plugin.MonitoringDeactivated=[HostMonitoringV2Plugin] Monitoring deactivated for method '{}'.
+HostMonitoringV2Plugin.ConnectionNone=[HostMonitoringV2Plugin] Attempted to execute method '{}' but the current connection is None.
+HostMonitoringV2Plugin.HostInfoNone=[HostMonitoringV2Plugin] Could not find HostInfo to monitor for the current connection.
+HostMonitoringV2Plugin.ConfigurationNotSupported=[HostMonitoringV2Plugin] Aborting connections from a separate thread is not supported for the detected driver dialect: '{}'. The EFM V2 plugin requires this feature to be supported.
+HostMonitoringV2Plugin.UnableToIdentifyConnection=[HostMonitoringV2Plugin] Unable to identify the connected database instance: '{}', please ensure the correct host list provider is specified. The host list provider in use is: '{}'.
+HostMonitorV2.ExceptionDuringMonitoringStop=[HostMonitorV2] Stopping monitoring after unhandled exception was thrown in monitoring thread for host '{}'. Exception: '{}'
+HostMonitorV2.MonitorIsStopped=[HostMonitorV2] Monitoring was already stopped for host '{}'.
+HostMonitorV2.StartMonitoringThreadNewContext=[HostMonitorV2] Start monitoring thread for checking new contexts for '{}'.
+HostMonitorV2.StopMonitoringThreadNewContext=[HostMonitorV2] Stop monitoring thread for checking new contexts for '{}'.
+HostMonitorV2.StartMonitoringThread=[HostMonitorV2] Start monitoring thread for '{}'.
+HostMonitorV2.OpeningMonitoringConnection=[HostMonitorV2] Opening a monitoring connection to '{}'
+HostMonitorV2.OpenedMonitoringConnection=[HostMonitorV2] Opened monitoring connection: '{}'
+HostMonitorV2.ExceptionAbortingConnection=[HostMonitorV2] Exception while aborting connection: '{}'
+HostMonitorV2.HostDead=[HostMonitorV2] Host '{}' is *dead*.
+HostMonitorV2.HostNotResponding=[HostMonitorV2] Host '{}' is not *responding* '{}'.
+HostMonitorV2.HostAlive=[HostMonitorV2] Host '{}' is *alive*.
+HostMonitorV2.StopMonitoringThread=[HostMonitorV2]  Stop monitoring thread for '{}'.
+MonitorServiceV2.ExceptionAbortingConnection=[MonitorServiceV2] Exception during aborting connection: '{}'
+
 HostSelector.NoEligibleHost=[HostSelector] No Eligible Hosts Found.
 HostSelector.NoHostsMatchingRole=[HostSelector] No hosts were found matching the requested role: '{}'.
 
@@ -193,15 +215,15 @@ LimitlessPlugin.UnsupportedDialectOrDatabase=[LimitlessPlugin] Unsupported diale
 
 LimitlessQueryHelper.UnsupportedDialectOrDatabase=[LimitlessQueryHelper] Unsupported dialect '{}' encountered. Please ensure JDBC connection parameters are correct, and refer to the documentation to ensure that the connecting database is compatible with the Limitless Connection Plugin.
 
-LimitlessRouterMonitor.errorDuringMonitoringStop=[LimitlessRouterMonitor] Stopping monitoring after unhandled error was thrown in Limitless Router Monitoring thread for node {}. Error: {}
-LimitlessRouterMonitor.InterruptedErrorDuringMonitoring=[LimitlessRouterMonitor] Limitless Router Monitoring thread for node {} was interrupted.
+LimitlessRouterMonitor.errorDuringMonitoringStop=[LimitlessRouterMonitor] Stopping monitoring after unhandled error was thrown in Limitless Router Monitoring thread for host {}. Error: {}
+LimitlessRouterMonitor.InterruptedErrorDuringMonitoring=[LimitlessRouterMonitor] Limitless Router Monitoring thread for host {} was interrupted.
 LimitlessRouterMonitor.InvalidQuery=[LimitlessRouterMonitor] Limitless Connection Plugin has encountered an error obtaining Limitless Router endpoints. Please ensure that you are connecting to an Aurora Limitless Database Shard Group Endpoint URL.
 LimitlessRouterMonitor.InvalidRouterLoad=[LimitlessRouterMonitor] Invalid load metric value of '{}' from the transaction router query aurora_limitless_router_endpoints() for transaction router '{}'. The load metric value must be a decimal value between 0 and 1. Host weight be assigned a default weight of 1.
 LimitlessRouterMonitor.GetNetworkTimeoutError=[LimitlessRouterMonitor] An error occurred while getting the connection network timeout: {}
 LimitlessRouterMonitor.OpeningConnection=[LimitlessRouterMonitor] Opening Limitless Router Monitor connection to '{}'.
 LimitlessRouterMonitor.OpenedConnection=[LimitlessRouterMonitor] Opened Limitless Router Monitor connection: {}.
-LimitlessRouterMonitor.Running=[LimitlessRouterMonitor] Limitless Router Monitor thread running on node {}.
-LimitlessRouterMonitor.Stopped=[LimitlessRouterMonitor] Limitless Router Monitor thread stopped on node {}.
+LimitlessRouterMonitor.Running=[LimitlessRouterMonitor] Limitless Router Monitor thread running on host {}.
+LimitlessRouterMonitor.Stopped=[LimitlessRouterMonitor] Limitless Router Monitor thread stopped on host {}.
 
 LimitlessRouterService.ConnectWithHost=[LimitlessRouterService] Connecting to host {}.
 LimitlessRouterService.ErrorClosingMonitor=[LimitlessRouterService] An error occurred while closing Limitless Router Monitor: {}

aws_advanced_python_wrapper/resources/aws_advanced_python_wrapper_messages.properties

@@ -13,6 +13,9 @@
 #  limitations under the License.
 
 from threading import Lock
+from typing import Generic, TypeVar
+
+T = TypeVar('T')
 
 
 class AtomicInt:
@@ -59,3 +62,31 @@ def compare_and_set(self, expected_value: int, new_value: int) -> bool:
                 self._value = new_value
                 return True
             return False
+
+
+class AtomicBoolean:
+    def __init__(self, initial_value: bool):
+        self._value: bool = initial_value
+        self._lock: Lock = Lock()
+
+    def get(self) -> bool:
+        with self._lock:
+            return self._value
+
+    def set(self, value: bool) -> None:
+        with self._lock:
+            self._value = value
+
+
+class AtomicReference(Generic[T]):
+    def __init__(self, initial_value: T):
+        self._value: T = initial_value
+        self._lock: Lock = Lock()
+
+    def get(self) -> T:
+        with self._lock:
+            return self._value
+
+    def set(self, new_value: T) -> None:
+        with self._lock:
+            self._value = new_value

aws_advanced_python_wrapper/utils/atomic.py

@@ -11,7 +11,7 @@
 #  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 #  See the License for the specific language governing permissions and
 #  limitations under the License.
-
+import copy
 from typing import Any, Dict, Optional
 from urllib.parse import unquote
 
@@ -73,7 +73,7 @@ def set(self, props: Properties, value: Any):
 
 
 class WrapperProperties:
-    DEFAULT_PLUGINS = "aurora_connection_tracker,failover,host_monitoring"
+    DEFAULT_PLUGINS = "aurora_connection_tracker,failover,host_monitoring_v2"
     _DEFAULT_TOKEN_EXPIRATION_SEC = 15 * 60
 
     PROFILE_NAME = WrapperProperty("profile_name", "Driver configuration profile name", None)
@@ -464,6 +464,8 @@ class WrapperProperties:
 
 
 class PropertiesUtils:
+    _MONITORING_PROPERTY_PREFIX = "monitoring-"
+
     @staticmethod
     def parse_properties(conn_info: str, **kwargs: Any) -> Properties:
         if conn_info.startswith("postgresql://") or conn_info.startswith("postgres://"):
@@ -582,7 +584,7 @@ def remove_wrapper_props(props: Properties):
                 if attr_val.name not in persisting_properties:
                     props.pop(attr_val.name, None)
 
-        monitor_prop_keys = [key for key in props if key.startswith("monitoring-")]
+        monitor_prop_keys = [key for key in props if key.startswith(PropertiesUtils._MONITORING_PROPERTY_PREFIX)]
         for key in monitor_prop_keys:
             props.pop(key, None)
 
@@ -610,3 +612,12 @@ def mask_properties(props: Properties) -> Properties:
             masked_properties[WrapperProperties.PASSWORD.name] = "***"
 
         return masked_properties
+
+    @staticmethod
+    def create_monitoring_properties(props: Properties) -> Properties:
+        monitoring_properties = copy.deepcopy(props)
+        for property_key in monitoring_properties.keys():
+            if property_key.startswith(PropertiesUtils._MONITORING_PROPERTY_PREFIX):
+                monitoring_properties[property_key[len(PropertiesUtils._MONITORING_PROPERTY_PREFIX):]] = \
+                    monitoring_properties[property_key]
+        return monitoring_properties

aws_advanced_python_wrapper/utils/properties.py

@@ -64,7 +64,7 @@ The AWS Advanced Python Driver has several built-in plugins that are available t
 | Plugin name                                                                                            | Plugin Code                 | Database Compatibility   | Description                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                       | Additional Required Dependencies                                     |
 |--------------------------------------------------------------------------------------------------------|-----------------------------|--------------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|----------------------------------------------------------------------|
 | [Failover Connection Plugin](./using-plugins/UsingTheFailoverPlugin.md)                                | `failover`                  | Aurora                   | Enables the failover functionality supported by Amazon Aurora clusters. Prevents opening a wrong connection to an old writer host dues to stale DNS after failover event. This plugin is enabled by default.                                                                                                                                                                                                                                                                                                      | None                                                                 |                             
-| [Host Monitoring Connection Plugin](./using-plugins/UsingTheHostMonitoringPlugin.md)                   | `host_monitoring`           | Aurora                   | Enables enhanced host connection failure monitoring, allowing faster failure detection rates. This plugin is enabled by default.                                                                                                                                                                                                                                                                                                                                                                                  | None                                                                 |
+| [Host Monitoring Connection Plugin](./using-plugins/UsingTheHostMonitoringPlugin.md)                   | `host_monitoring_v2` or `host_monitoring`         | Aurora                   | Enables enhanced host connection failure monitoring, allowing faster failure detection rates. This plugin is enabled by default.                                                                                                                                                                                                                                                                                                                                                                                  | None                                                                 |
 | [IAM Authentication Connection Plugin](./using-plugins/UsingTheIamAuthenticationPlugin.md)             | `iam`                       | Any database             | Enables users to connect to their Amazon Aurora clusters using AWS Identity and Access Management (IAM).                                                                                                                                                                                                                                                                                                                                                                                                          | [Boto3 - AWS SDK for Python](https://aws.amazon.com/sdk-for-python/) |
 | [AWS Secrets Manager Connection Plugin](./using-plugins/UsingTheAwsSecretsManagerPlugin.md)            | `aws_secrets_manager`       | Any database             | Enables fetching database credentials from the AWS Secrets Manager service.                                                                                                                                                                                                                                                                                                                                                                                                                                       | [Boto3 - AWS SDK for Python](https://aws.amazon.com/sdk-for-python/) |
 | [Federated Authentication Connection Plugin](./using-plugins/UsingTheFederatedAuthenticationPlugin.md) | `federated_auth`            | Any database             | Enables users to authenticate via Federated Identity and then database access via IAM.                                                                                                                                                                                                                                                                                                                                                                                                                            | [Boto3 - AWS SDK for Python](https://aws.amazon.com/sdk-for-python/) |

docs/using-the-python-driver/UsingThePythonDriver.md

@@ -13,7 +13,7 @@ The plugin can be loaded by adding the plugin code `fastest_response_strategy` t
 
 ```python
 params = {
-    "plugins": "read_write_splitting,fastest_response_strategy,failover,host_monitoring",
+    "plugins": "read_write_splitting,fastest_response_strategy,failover,host_monitoring_v2",
     "reader_response_strategy": "fastest_response"
     # Add other connection properties below...
 }