feat: blue green support (#498)

Author: karenc-bq

common/lib/aws_client.ts

@@ -177,6 +177,10 @@ export abstract class AwsClient extends EventEmitter {
 
   abstract rollback(): Promise<any>;
 
+  unwrapPlugin<T>(iface: new (...args: any[]) => T): T | null {
+    return this.pluginManager.unwrapPlugin(iface);
+  }
+
   async isValid(): Promise<boolean> {
     if (!this.targetClient) {
       return Promise.resolve(false);

common/lib/connection_plugin_chain_builder.ts

@@ -33,19 +33,16 @@ import { ReadWriteSplittingPluginFactory } from "./plugins/read_write_splitting_
 import { OktaAuthPluginFactory } from "./plugins/federated_auth/okta_auth_plugin_factory";
 import { HostMonitoringPluginFactory } from "./plugins/efm/host_monitoring_plugin_factory";
 import { AuroraInitialConnectionStrategyFactory } from "./plugins/aurora_initial_connection_strategy_plugin_factory";
-import {
-  AuroraConnectionTrackerPluginFactory
-} from "./plugins/connection_tracker/aurora_connection_tracker_plugin_factory";
+import { AuroraConnectionTrackerPluginFactory } from "./plugins/connection_tracker/aurora_connection_tracker_plugin_factory";
 import { ConnectionProviderManager } from "./connection_provider_manager";
 import { DeveloperConnectionPluginFactory } from "./plugins/dev/developer_connection_plugin_factory";
 import { ConnectionPluginFactory } from "./plugin_factory";
 import { LimitlessConnectionPluginFactory } from "./plugins/limitless/limitless_connection_plugin_factory";
-import {
-  FastestResponseStrategyPluginFactory
-} from "./plugins/strategy/fastest_response/fastest_respose_strategy_plugin_factory";
+import { FastestResponseStrategyPluginFactory } from "./plugins/strategy/fastest_response/fastest_respose_strategy_plugin_factory";
 import { CustomEndpointPluginFactory } from "./plugins/custom_endpoint/custom_endpoint_plugin_factory";
 import { ConfigurationProfile } from "./profile/configuration_profile";
 import { HostMonitoring2PluginFactory } from "./plugins/efm2/host_monitoring2_plugin_factory";
+import { BlueGreenPluginFactory } from "./plugins/bluegreen/blue_green_plugin_factory";
 
 /*
   Type alias used for plugin factory sorting. It holds a reference to a plugin
@@ -64,6 +61,7 @@ export class ConnectionPluginChainBuilder {
     ["initialConnection", { factory: AuroraInitialConnectionStrategyFactory, weight: 390 }],
     ["auroraConnectionTracker", { factory: AuroraConnectionTrackerPluginFactory, weight: 400 }],
     ["staleDns", { factory: StaleDnsPluginFactory, weight: 500 }],
+    ["bg", { factory: BlueGreenPluginFactory, weight: 550 }],
     ["readWriteSplitting", { factory: ReadWriteSplittingPluginFactory, weight: 600 }],
     ["failover", { factory: FailoverPluginFactory, weight: 700 }],
     ["failover2", { factory: Failover2PluginFactory, weight: 710 }],
@@ -84,6 +82,7 @@ export class ConnectionPluginChainBuilder {
     [AuroraInitialConnectionStrategyFactory, 390],
     [AuroraConnectionTrackerPluginFactory, 400],
     [StaleDnsPluginFactory, 500],
+    [BlueGreenPluginFactory, 550],
     [ReadWriteSplittingPluginFactory, 600],
     [FailoverPluginFactory, 700],
     [Failover2PluginFactory, 710],

common/lib/database_dialect/blue_green_dialect.ts

@@ -0,0 +1,58 @@
+/*
+  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.
+*/
+
+import { ClientWrapper } from "../client_wrapper";
+
+export class BlueGreenResult {
+  private readonly _version: string;
+  private readonly _endpoint: string;
+  private readonly _port: number;
+  private readonly _role: string;
+  private readonly _status: string;
+
+  constructor(version: string, endpoint: string, port: number, role: string, status: string) {
+    this._version = version;
+    this._endpoint = endpoint;
+    this._port = port;
+    this._role = role;
+    this._status = status;
+  }
+
+  get version(): string {
+    return this._version;
+  }
+
+  get endpoint(): string {
+    return this._endpoint;
+  }
+
+  get port(): number {
+    return this._port;
+  }
+
+  get role(): string {
+    return this._role;
+  }
+
+  get status(): string {
+    return this._status;
+  }
+}
+
+export interface BlueGreenDialect {
+  isBlueGreenStatusAvailable(clientWrapper: ClientWrapper): Promise<boolean>;
+  getBlueGreenStatus(clientWrapper: ClientWrapper): Promise<BlueGreenResult[] | null>;
+}

common/lib/driver_connection_provider.ts

@@ -27,7 +27,6 @@ import { promisify } from "util";
 import { lookup } from "dns";
 import { PluginService } from "./plugin_service";
 import { logger } from "../logutils";
-import { maskProperties } from "./utils/utils";
 import { ClientWrapper } from "./client_wrapper";
 import { RoundRobinHostSelector } from "./round_robin_host_selector";
 import { DriverDialect } from "./driver_dialect/driver_dialect";
@@ -88,14 +87,7 @@ export class DriverConnectionProvider implements ConnectionProvider {
       const fixedHost: string = this.rdsUtils.removeGreenInstancePrefix(hostInfo.host);
       resultProps.set(WrapperProperties.HOST.name, fixedHost);
 
-      logger.info(
-        "Connecting to " +
-          fixedHost +
-          " after correcting the hostname from " +
-          originalHost +
-          "\nwith properties: \n" +
-          JSON.stringify(Object.fromEntries(maskProperties(resultProps)))
-      );
+      logger.info("Connecting to " + fixedHost + " after correcting the hostname from " + originalHost);
 
       resultTargetClient = driverDialect.connect(hostInfo, resultProps);
     }

common/lib/error_handler.ts

@@ -21,6 +21,8 @@ export interface ErrorHandler {
 
   isNetworkError(e: Error): boolean;
 
+  isSyntaxError(e: Error): boolean;
+
   /**
    * Checks whether there has been an unexpected error emitted and if the error is a type of login error.
    */

common/lib/plugin_manager.ts

@@ -413,10 +413,32 @@ export class PluginManager {
     throw new AwsWrapperError(Messages.get("PluginManager.unableToRetrievePlugin"));
   }
 
+  isPluginInUse(plugin: any): boolean {
+    for (const p of this._plugins) {
+      if (p instanceof plugin) {
+        return true;
+      }
+    }
+    return false;
+  }
+
   static registerPlugin(pluginCode: string, pluginFactory: typeof ConnectionPluginFactory) {
     ConnectionPluginChainBuilder.PLUGIN_FACTORIES.set(pluginCode, {
       factory: pluginFactory,
       weight: ConnectionPluginChainBuilder.WEIGHT_RELATIVE_TO_PRIOR_PLUGIN
     });
   }
+
+  unwrapPlugin<T>(iface: new (...args: any[]) => T): T | null {
+    if (!this._plugins) {
+      return null;
+    }
+
+    for (const p of this._plugins) {
+      if (p instanceof iface) {
+        return p as any;
+      }
+    }
+    return null;
+  }
 }

common/lib/plugin_service.ts

@@ -140,6 +140,16 @@ export interface PluginService extends ErrorHandler {
   getTelemetryFactory(): TelemetryFactory;
 
   setAllowedAndBlockedHosts(allowedAndBlockedHosts: AllowedAndBlockedHosts): void;
+
+  setStatus<T>(clazz: any, status: T | null, clusterBound: boolean): void;
+
+  setStatus<T>(clazz: any, status: T | null, key: string): void;
+
+  getStatus<T>(clazz: any, clusterBound: boolean): T;
+
+  getStatus<T>(clazz: any, key: string): T;
+
+  isPluginInUse(plugin: any): boolean;
 }
 
 export class PluginServiceImpl implements PluginService, HostListProviderService {
@@ -159,6 +169,8 @@ export class PluginServiceImpl implements PluginService, HostListProviderService
   protected static readonly hostAvailabilityExpiringCache: CacheMap<string, HostAvailability> = new CacheMap<string, HostAvailability>();
   readonly props: Map<string, any>;
   private allowedAndBlockedHosts: AllowedAndBlockedHosts | null = null;
+  protected static readonly statusesExpiringCache: CacheMap<string, any> = new CacheMap();
+  protected static readonly DEFAULT_STATUS_CACHE_EXPIRE_NANO: number = 3_600_000_000_000; // 60 minutes
 
   constructor(
     container: PluginServiceManagerContainer,
@@ -686,6 +698,10 @@ export class PluginServiceImpl implements PluginService, HostListProviderService
     return this.getDialect().getErrorHandler().isNetworkError(e);
   }
 
+  isSyntaxError(e: Error): boolean {
+    return this.getDialect().getErrorHandler().isSyntaxError(e);
+  }
+
   hasLoginError(): boolean {
     return this.getDialect().getErrorHandler().hasLoginError();
   }
@@ -713,4 +729,53 @@ export class PluginServiceImpl implements PluginService, HostListProviderService
   static clearHostAvailabilityCache(): void {
     PluginServiceImpl.hostAvailabilityExpiringCache.clear();
   }
+
+  getStatus<T>(clazz: any, clusterBound: boolean): T;
+  getStatus<T>(clazz: any, key: string): T;
+  getStatus<T>(clazz: any, clusterBound: boolean | string): T {
+    if (typeof clusterBound === "string") {
+      return <T>PluginServiceImpl.statusesExpiringCache.get(this.getStatusCacheKey(clazz, clusterBound));
+    }
+    let clusterId: string = null;
+    if (clusterBound) {
+      try {
+        clusterId = this._hostListProvider.getClusterId();
+      } catch (e) {
+        // Do nothing
+      }
+    }
+    return this.getStatus(clazz, clusterId);
+  }
+
+  protected getStatusCacheKey<T>(clazz: T, key: string): string {
+    return `${!key ? "" : key.trim().toLowerCase()}::${clazz.toString()}`;
+  }
+
+  setStatus<T>(clazz: any, status: T | null, clusterBound: boolean): void;
+  setStatus<T>(clazz: any, status: T | null, key: string): void;
+  setStatus<T>(clazz: any, status: T, clusterBound: boolean | string): void {
+    if (typeof clusterBound === "string") {
+      const cacheKey: string = this.getStatusCacheKey(clazz, clusterBound);
+      if (!status) {
+        PluginServiceImpl.statusesExpiringCache.delete(cacheKey);
+      } else {
+        PluginServiceImpl.statusesExpiringCache.put(cacheKey, status, PluginServiceImpl.DEFAULT_STATUS_CACHE_EXPIRE_NANO);
+      }
+      return;
+    }
+
+    let clusterId: string | null = null;
+    if (clusterBound) {
+      try {
+        clusterId = this._hostListProvider.getClusterId();
+      } catch (e) {
+        // Do nothing
+      }
+    }
+    this.setStatus(clazz, status, clusterId);
+  }
+
+  isPluginInUse(plugin: any) {
+    return this.pluginServiceManagerContainer.pluginManager!.isPluginInUse(plugin);
+  }
 }