feat: bg implementation

Author: JuanLeee

AwsWrapperDataProvider.Plugin.SecretsManager/SecretsManager/SecretsManagerAuthPlugin.cs

@@ -1,3 +1,4 @@
+
 // Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
 //
 // Licensed under the Apache License, Version 2.0 (the "License").

AwsWrapperDataProvider.sln

@@ -41,6 +41,8 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "AwsWrapperDataProviderExamp
 EndProject
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "AwsWrapperDataProvider.Performance.Tests", "AwsWrapperDataProvider.Performance.Tests\AwsWrapperDataProvider.Performance.Tests.csproj", "{1A00C448-9549-4995-9180-C70C908963A5}"
 EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "AwsWrapperDataProvider.Plugin.BlueGreenConnection.Tests", "AwsWrapperDataProvider.Plugin.BlueGreenConnection.Tests\AwsWrapperDataProvider.Plugin.BlueGreenConnection.Tests.csproj", "{304AEAED-F4A9-4B95-9F4D-8183CB47E4EB}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -99,6 +101,10 @@ Global
 		{D87E0507-5E62-4852-BDFC-CB9E167A85FB}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{D87E0507-5E62-4852-BDFC-CB9E167A85FB}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{D87E0507-5E62-4852-BDFC-CB9E167A85FB}.Release|Any CPU.Build.0 = Release|Any CPU
+		{304AEAED-F4A9-4B95-9F4D-8183CB47E4EB}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{304AEAED-F4A9-4B95-9F4D-8183CB47E4EB}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{304AEAED-F4A9-4B95-9F4D-8183CB47E4EB}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{304AEAED-F4A9-4B95-9F4D-8183CB47E4EB}.Release|Any CPU.Build.0 = Release|Any CPU
 		{C3E1A7B2-8D4F-4E90-B2A1-5F6E7D8C9B0A}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
 		{C3E1A7B2-8D4F-4E90-B2A1-5F6E7D8C9B0A}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{C3E1A7B2-8D4F-4E90-B2A1-5F6E7D8C9B0A}.Release|Any CPU.ActiveCfg = Release|Any CPU

AwsWrapperDataProvider/AwsWrapperConnection.cs

@@ -39,7 +39,7 @@ public class AwsWrapperConnection : DbConnection, IWrapper
 
     private bool deferredInitialization = false;
 
-    internal ConnectionPluginManager? PluginManager { get; private set; }
+    public ConnectionPluginManager? PluginManager { get; private set; }
 
     internal Dictionary<string, string>? ConnectionProperties { get; private set; }
 

AwsWrapperDataProvider/Driver/ConnectionPluginManager.cs

@@ -20,14 +20,17 @@
 using AwsWrapperDataProvider.Driver.HostInfo;
 using AwsWrapperDataProvider.Driver.HostListProviders;
 using AwsWrapperDataProvider.Driver.Plugins;
+using AwsWrapperDataProvider.Driver.Utils;
 using AwsWrapperDataProvider.Properties;
+using Microsoft.Extensions.Logging;
 
 namespace AwsWrapperDataProvider.Driver;
 
 public class ConnectionPluginManager
 {
     private readonly Dictionary<string, Delegate> pluginChainDelegates = [];
-    protected IList<IConnectionPlugin> plugins = [];
+    public IList<IConnectionPlugin> Plugins = [];
+    public string[] ActivePluginCodes = [];
     protected IConnectionProvider defaultConnProvider;
     protected IConnectionProvider? effectiveConnProvider;
     protected ConfigurationProfile? configurationProfile;
@@ -38,6 +41,8 @@ public class ConnectionPluginManager
     private const string ConnectMethod = "DbConnection.Open";
     private const string ForceConnectMethod = "DbConnection.ForceOpen";
     private const string InitHostMethod = "initHostProvider";
+    
+    private static readonly ILogger<ConnectionPluginManager> Logger = LoggerUtils.GetLogger<ConnectionPluginManager>();
 
     private delegate Task<T> PluginPipelineDelegate<T>(IConnectionPlugin plugin, ADONetDelegate<T> methodFunc);
 
@@ -74,7 +79,7 @@ public ConnectionPluginManager(
     {
         this.defaultConnProvider = defaultConnectionProvider;
         this.effectiveConnProvider = effectiveConnectionProvider;
-        this.plugins = plugins;
+        this.Plugins = plugins;
         this.ConnectionWrapper = connection;
     }
 
@@ -84,12 +89,13 @@ public void InitConnectionPluginChain(
     {
         this.pluginService = pluginService;
         ConnectionPluginChainBuilder pluginChainBuilder = new();
-        this.plugins = pluginChainBuilder.GetPlugins(
+        this.Plugins = pluginChainBuilder.GetPlugins(
             this.pluginService,
             this.defaultConnProvider,
             this.effectiveConnProvider,
             props,
             this.configurationProfile);
+        this.ActivePluginCodes = pluginChainBuilder.GetPluginCodes(this.pluginService, props);
     }
 
     private async Task<T> ExecuteWithSubscribedPlugins<T>(
@@ -104,7 +110,8 @@ private async Task<T> ExecuteWithSubscribedPlugins<T>(
         if (!this.pluginChainDelegates.TryGetValue(methodName, out Delegate? del))
         {
             del = this.MakePluginChainDelegate<T>(methodName);
-            this.pluginChainDelegates.Add(methodName, del);
+            Logger.LogDebug("Created plugin chain delegate for method {Delegate}", del);
+            this.pluginChainDelegates[methodName] = del;
         }
 
         if (del is not PluginChainADONetDelegate<T> pluginChainDelegate)
@@ -129,16 +136,24 @@ private async Task<T> ExecuteWithSubscribedPlugins<T>(
 
     private PluginChainADONetDelegate<T> MakePluginChainDelegate<T>(string methodName)
     {
-        PluginChainADONetDelegate<T>? pluginChainDelegate = null;
+        var pluginsInChain = new List<string>();
 
-        for (int i = this.plugins.Count - 1; i >= 0; i--)
+        PluginChainADONetDelegate<T>? pluginChainDelegate = null;
+        Logger.LogDebug("MakePluginChainDelegate {Plugins}", this.Plugins);
+        for (int i = this.Plugins.Count - 1; i >= 0; i--)
         {
-            IConnectionPlugin plugin = this.plugins[i];
+            IConnectionPlugin plugin = this.Plugins[i];
             IReadOnlySet<string> subscribedMethods = plugin.SubscribedMethods;
             bool isSubscribed = subscribedMethods.Contains(AllMethods) || subscribedMethods.Contains(methodName);
 
             if (isSubscribed)
             {
+                string pluginName = plugin.GetType().Name;
+                pluginsInChain.Add(pluginName);
+
+                Console.WriteLine($"  [{pluginsInChain.Count}] {pluginName}");
+                Logger.LogDebug("  Added plugin to chain: {PluginName}", pluginName);
+
                 if (pluginChainDelegate == null)
                 {
                     // DefaultConnectionPlugin always terminates the list of plugins
@@ -166,6 +181,7 @@ public virtual Task<T> Execute<T>(
         ADONetDelegate<T> methodFunc,
         params object[] methodArgs)
     {
+        Logger.LogTrace("ConnectionPluginManager Execute");
         return this.ExecuteWithSubscribedPlugins(
             methodName,
             (plugin, methodFunc) => plugin.Execute(methodInvokeOn, methodName, methodFunc, methodArgs),
@@ -248,4 +264,9 @@ public virtual bool AcceptsStrategy(string strategy)
     {
         return this.defaultConnProvider.AcceptsStrategy(strategy);
     }
+
+    public bool IsPluginActive(string pluginCode)
+    {
+        return this.ActivePluginCodes.Contains(pluginCode);
+    }
 }

AwsWrapperDataProvider/Driver/Dialects/AuroraMySqlDialect.cs

@@ -21,7 +21,7 @@
 
 namespace AwsWrapperDataProvider.Driver.Dialects;
 
-public class AuroraMySqlDialect : MySqlDialect
+public class AuroraMySqlDialect : MySqlDialect, IBlueGreenDialect
 {
     private static readonly ILogger<AuroraMySqlDialect> Logger = LoggerUtils.GetLogger<AuroraMySqlDialect>();
 
@@ -34,11 +34,16 @@ public class AuroraMySqlDialect : MySqlDialect
 
     private static readonly string NodeIdQuery = "SELECT @@aurora_server_id, @@aurora_server_id";
 
-    internal static readonly string IsDialectQuery = "SHOW VARIABLES LIKE 'aurora_version'";
+    internal static readonly string 
+        IsDialectQuery = "SHOW VARIABLES LIKE 'aurora_version'";
 
     private static readonly string IsWriterQuery = "SELECT SERVER_ID FROM information_schema.replica_host_status "
         + "WHERE SESSION_ID = 'MASTER_SESSION_ID' AND SERVER_ID = @@aurora_server_id";
 
+    private static readonly string AuroraMySqlBgTopologyExistsQuery = "SELECT 1 AS tmp FROM information_schema.tables WHERE table_schema = 'mysql' AND table_name = 'rds_topology'";
+
+    protected static readonly string AuroraMySqlBgStatusQuery = "SELECT * FROM mysql.rds_topology";
+
     public override IList<Type> DialectUpdateCandidates { get; } = [
         typeof(RdsMultiAzDbClusterMySqlDialect),
     ];
@@ -54,7 +59,7 @@ public override async Task<bool> IsDialect(DbConnection connection)
         }
         catch (Exception ex) when (this.ExceptionHandler.IsSyntaxError(ex))
         {
-            // Syntax error - expected when querying against incorrect dialect
+            Logger.LogTrace(ex, Resources.Error_CantCheckDialect_Syntax, nameof(RdsPgDialect));
         }
         catch (Exception ex)
         {
@@ -85,4 +90,14 @@ private HostListProviderSupplier GetHostListProviderSupplier()
                     NodeIdQuery,
                     IsReaderQuery);
     }
+
+    public async Task<bool> IsBlueGreenStatusAvailable(DbConnection connection)
+    {
+        return await DialectUtils.CheckExistenceQueries(connection, this.ExceptionHandler, Logger, AuroraMySqlBgTopologyExistsQuery);
+    }
+
+    public string GetBlueGreenStatusQuery()
+    {
+        return AuroraMySqlBgStatusQuery;
+    }
 }

AwsWrapperDataProvider/Driver/Dialects/AuroraPgDialect.cs

@@ -48,6 +48,10 @@ public class AuroraPgDialect : PgDialect, IAuroraLimitlessDialect
     private static readonly string IsWriterQuery = "SELECT SERVER_ID FROM pg_catalog.aurora_replica_status() "
         + "WHERE SESSION_ID OPERATOR(pg_catalog.=) 'MASTER_SESSION_ID' AND SERVER_ID OPERATOR(pg_catalog.=) aurora_db_instance_identifier()";
 
+    protected static readonly string AuroraPostgreSqlBgTopologyExistsQuery = "SELECT 'pg_catalog.get_blue_green_fast_switchover_metadata'::regproc";
+
+    protected static readonly string AuroraPostgreSqlBgStatusQuery = "SELECT * FROM pg_catalog.get_blue_green_fast_switchover_metadata('aws_wrapper_dotnet_data_provider_wrapper')";
+
     public override IList<Type> DialectUpdateCandidates { get; } = [
         typeof(RdsMultiAzDbClusterPgDialect),
     ];
@@ -83,7 +87,7 @@ public override async Task<bool> IsDialect(DbConnection connection)
         }
         catch (Exception ex) when (this.ExceptionHandler.IsSyntaxError(ex))
         {
-            // Syntax error - expected when querying against incorrect dialect
+            Logger.LogTrace(ex, Resources.Error_CantCheckDialect_Syntax, nameof(RdsPgDialect));
         }
         catch (Exception ex)
         {
@@ -115,5 +119,15 @@ private HostListProviderSupplier GetHostListProviderSupplier()
                     IsReaderQuery);
     }
 
+    public async Task<bool> IsBlueGreenStatusAvailable(DbConnection connection)
+    {
+        return await DialectUtils.CheckExistenceQueries(connection, this.ExceptionHandler, Logger, AuroraPostgreSqlBgTopologyExistsQuery);
+    }
+
+    public string GetBlueGreenStatusQuery()
+    {
+        return AuroraPostgreSqlBgStatusQuery;
+    }
+
     public string LimitlessRouterEndpointQuery { get => "SELECT router_endpoint, load from aurora_limitless_router_endpoints()"; }
 }

AwsWrapperDataProvider/Driver/Dialects/DialectUtils.cs

@@ -0,0 +1,49 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+// 
+// Licensed under the Apache License, Version 2.0 (the "License").
+// You may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+// 
+// http://www.apache.org/licenses/LICENSE-2.0
+// 
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// 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.
+
+using System.Data.Common;
+using AwsWrapperDataProvider.Driver.Exceptions;
+using AwsWrapperDataProvider.Properties;
+using Microsoft.Extensions.Logging;
+
+namespace AwsWrapperDataProvider.Driver.Dialects;
+
+public static class DialectUtils
+{
+    public static async Task<bool> CheckExistenceQueries(DbConnection connection, IExceptionHandler exceptionHandler, ILogger logger, string query)
+    {
+        try
+        {
+            await using var command = connection.CreateCommand();
+            command.CommandText = query;
+            await using var reader = await command.ExecuteReaderAsync();
+            return await reader.ReadAsync();
+        }
+        catch (Exception ex) when (exceptionHandler.IsSyntaxError(ex))
+        {
+            // Syntax error - expected when querying against incorrect dialect
+        }
+        catch (Exception ex)
+        {
+            logger.LogTrace(ex, Resources.Error_CantCheckDialect, nameof(AuroraMySqlDialect));
+        }
+
+        return false;
+    }
+
+    public static bool IsBlueGreenConnectionDialect(IDialect dialect)
+    {
+        return dialect is IBlueGreenDialect;
+    }
+}

AwsWrapperDataProvider/Driver/Dialects/IBlueGreenDialect.cs

@@ -0,0 +1,24 @@
+// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License").
+// You may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// 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.
+
+using System.Data.Common;
+
+namespace AwsWrapperDataProvider.Driver.Dialects;
+
+public interface IBlueGreenDialect
+{
+    Task<bool> IsBlueGreenStatusAvailable(DbConnection connection);
+
+    string GetBlueGreenStatusQuery();
+}

AwsWrapperDataProvider/Driver/Dialects/RdsMultiAzDbClusterPgDialect.cs

@@ -32,7 +32,7 @@ public class RdsMultiAzDbClusterPgDialect : PgDialect
     private static readonly ILogger<RdsMultiAzDbClusterPgDialect> Logger = LoggerUtils.GetLogger<RdsMultiAzDbClusterPgDialect>();
 
     private static readonly string TopologyQuery =
-        $"SELECT id, endpoint, port FROM rds_tools.show_topology('aws_dotnet_driver-{DriverVersion}')";
+        $"SELECT id, endpoint, port FROM rds_tools.show_topology('aws_wrapper_dotnet_data_provider_wrapper-{DriverVersion}')";
 
     private static readonly string FetchWriterNodeQuery =
         "SELECT multi_az_db_cluster_source_dbi_resource_id FROM rds_tools.multi_az_db_cluster_source_dbi_resource_id()"

AwsWrapperDataProvider/Driver/Dialects/RdsMySqlDialect.cs

@@ -19,8 +19,13 @@
 
 namespace AwsWrapperDataProvider.Driver.Dialects;
 
-public class RdsMySqlDialect : MySqlDialect
+public class RdsMySqlDialect : MySqlDialect, IBlueGreenDialect
 {
+    protected static readonly string RdsMySqlTopologyTableExistsQuery = "SELECT 1 AS tmp FROM information_schema.tables WHERE" +
+                                                                        " table_schema = 'mysql' AND table_name = 'rds_topology'";
+
+    protected static readonly string RdsMySqlBgStatusQuery = "SELECT * FROM mysql.rds_topology";
+
     private static readonly ILogger<RdsMySqlDialect> Logger = LoggerUtils.GetLogger<RdsMySqlDialect>();
 
     public override IList<Type> DialectUpdateCandidates { get; } =
@@ -56,7 +61,7 @@ public override async Task<bool> IsDialect(DbConnection conn)
         }
         catch (Exception ex) when (this.ExceptionHandler.IsSyntaxError(ex))
         {
-            // Syntax error - expected when querying against incorrect dialect
+            Logger.LogTrace(ex, Resources.Error_CantCheckDialect_Syntax, nameof(RdsPgDialect));
         }
         catch (Exception ex)
         {
@@ -65,4 +70,14 @@ public override async Task<bool> IsDialect(DbConnection conn)
 
         return false;
     }
+
+    public async Task<bool> IsBlueGreenStatusAvailable(DbConnection connection)
+    {
+        return await DialectUtils.CheckExistenceQueries(connection, this.ExceptionHandler, Logger, RdsMySqlTopologyTableExistsQuery);
+    }
+
+    public string GetBlueGreenStatusQuery()
+    {
+        return RdsMySqlBgStatusQuery;
+    }
 }