feat: efm2

common/lib/connection_plugin_chain_builder.ts

@@ -39,6 +39,7 @@ 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 { ConfigurationProfile } from "./profile/configuration_profile";
+import { HostMonitoring2PluginFactory } from "./plugins/efm2/host_monitoring2_plugin_factory";
 
 /*
   Type alias used for plugin factory sorting. It holds a reference to a plugin
@@ -59,6 +60,7 @@ export class ConnectionPluginChainBuilder {
     ["readWriteSplitting", { factory: ReadWriteSplittingPluginFactory, weight: 600 }],
     ["failover", { factory: FailoverPluginFactory, weight: 700 }],
     ["efm", { factory: HostMonitoringPluginFactory, weight: 800 }],
+    ["efm2", { factory: HostMonitoring2PluginFactory, weight: 810 }],
     ["fastestResponseStrategy", { factory: FastestResponseStrategyPluginFactory, weight: 900 }],
     ["limitless", { factory: LimitlessConnectionPluginFactory, weight: 950 }],
     ["iam", { factory: IamAuthenticationPluginFactory, weight: 1000 }],

common/lib/plugins/efm/host_monitoring_connection_plugin.ts

@@ -121,6 +121,7 @@
     } finally {
       if (monitorContext != null) {
         await this.monitorService.stopMonitoring(monitorContext);
+        logger.debug(Messages.get("HostMonitoringConnectionPlugin.monitoringDeactivated", methodName));
 
         if (monitorContext.isHostUnhealthy) {
           const monitoringHostInfo = await this.getMonitoringHostInfo();
@@ -148,12 +149,12 @@
     return result;
   }
 
-  private throwUnableToIdentifyConnection(host: HostInfo | null, provider: HostListProvider | null): never {
+  private throwUnableToIdentifyConnection(host: HostInfo | null): never {
     throw new AwsWrapperError(
       Messages.get(
         "HostMonitoringConnectionPlugin.unableToIdentifyConnection",
         host != null ? host.host : "unknown host",
         provider != null ? provider.getHostProviderType() : "unknown provider"
       )
     );
   }
@@ -163,17 +164,17 @@
       this.monitoringHostInfo = this.pluginService.getCurrentHostInfo();
       const provider: HostListProvider | null = this.pluginService.getHostListProvider();
       if (this.monitoringHostInfo == null) {
-        this.throwUnableToIdentifyConnection(null, provider);
+        this.throwUnableToIdentifyConnection(null);
       }
       const rdsUrlType: RdsUrlType = this.rdsUtils.identifyRdsType(this.monitoringHostInfo.url);
 
       try {
         if (rdsUrlType.isRdsCluster) {
-          logger.debug("Monitoring host info is associated with a cluster endpoint, plugin needs to identify the cluster connection");
+          logger.debug(Messages.get("HostMonitoringConnectionPlugin.identifyClusterConnection"));
           this.monitoringHostInfo = await this.pluginService.identifyConnection(this.pluginService.getCurrentClient().targetClient!);
           if (this.monitoringHostInfo == null) {
             const host: HostInfo | null = this.pluginService.getCurrentHostInfo();
-            this.throwUnableToIdentifyConnection(host, provider);
+            this.throwUnableToIdentifyConnection(host);
           }
           await this.pluginService.fillAliases(this.pluginService.getCurrentClient().targetClient!, this.monitoringHostInfo);
         }

common/lib/plugins/efm/monitor_connection_context.ts

@@ -77,7 +77,7 @@ export class MonitorConnectionContext {
       this.telemetryAbortedConnectionCounter.inc();
     } catch (error: any) {
       // ignore
-      logger.debug(Messages.get("MonitorConnectionContext.exceptionAbortingConnection", error.message));
+      logger.debug(Messages.get("MonitorConnectionContext.errorAbortingConnection", error.message));
     }
     this.abortedConnectionCounter++;
   }
@@ -127,7 +127,6 @@ export class MonitorConnectionContext {
 
       const invalidHostDurationNano: number = statusCheckEndNano - this.invalidHostStartTimeNano;
       const maxInvalidHostDurationMillis: number = this.failureDetectionIntervalMillis * Math.max(0, this.failureDetectionCount);
-
       if (this.failureCount >= this.failureDetectionCount || invalidHostDurationNano >= maxInvalidHostDurationMillis * 1_000_000) {
         logger.debug(Messages.get("MonitorConnectionContext.hostDead", hostName));
         this.isHostUnhealthy = true;

common/lib/plugins/efm2/host_monitoring2_connection_plugin.ts

@@ -0,0 +1,175 @@
+/*
+  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 { PluginService } from "../../plugin_service";
+import { HostChangeOptions } from "../../host_change_options";
+import { HostInfo } from "../../host_info";
+import { OldConnectionSuggestionAction } from "../../old_connection_suggestion_action";
+import { RdsUtils } from "../../utils/rds_utils";
+import { AbstractConnectionPlugin } from "../../abstract_connection_plugin";
+import { RdsUrlType } from "../../utils/rds_url_type";
+import { WrapperProperties } from "../../wrapper_property";
+import { MonitorConnectionContext } from "./monitor_connection_context";
+import { logger, uniqueId } from "../../../logutils";
+import { Messages } from "../../utils/messages";
+import { MonitorService, MonitorServiceImpl } from "./monitor_service";
+import { AwsWrapperError } from "../../utils/errors";
+import { HostListProvider } from "../../host_list_provider/host_list_provider";
+import { CanReleaseResources } from "../../can_release_resources";
+import { SubscribedMethodHelper } from "../../utils/subscribed_method_helper";
+import { ClientWrapper } from "../../client_wrapper";
+
+export class HostMonitoring2ConnectionPlugin extends AbstractConnectionPlugin implements CanReleaseResources {
+  id: string = uniqueId("_efm2Plugin");
+  private readonly properties: Map<string, any>;
+  private pluginService: PluginService;
+  private rdsUtils: RdsUtils;
+  private monitoringHostInfo: HostInfo | null = null;
+  private monitorService: MonitorService;
+
+  constructor(pluginService: PluginService, properties: Map<string, any>, rdsUtils: RdsUtils = new RdsUtils(), monitorService?: MonitorServiceImpl) {
+    super();
+    this.pluginService = pluginService;
+    this.properties = properties;
+    this.rdsUtils = rdsUtils;
+    this.monitorService = monitorService ?? new MonitorServiceImpl(pluginService);
+  }
+
+  getSubscribedMethods(): Set<string> {
+    return new Set<string>(["*"]);
+  }
+
+  connect(
+    hostInfo: HostInfo,
+    props: Map<string, any>,
+    isInitialConnection: boolean,
+    connectFunc: () => Promise<ClientWrapper>
+  ): Promise<ClientWrapper> {
+    return this.connectInternal(hostInfo, connectFunc);
+  }
+
+  forceConnect(
+    hostInfo: HostInfo,
+    props: Map<string, any>,
+    isInitialConnection: boolean,
+    forceConnectFunc: () => Promise<ClientWrapper>
+  ): Promise<ClientWrapper> {
+    return this.connectInternal(hostInfo, forceConnectFunc);
+  }
+
+  private async connectInternal(hostInfo: HostInfo, connectFunc: () => Promise<ClientWrapper>): Promise<ClientWrapper> {
+    const targetClient = await connectFunc();
+    if (targetClient != null) {
+      const type: RdsUrlType = this.rdsUtils.identifyRdsType(hostInfo.host);
+      if (type.isRdsCluster) {
+        hostInfo.resetAliases();
+        await this.pluginService.fillAliases(targetClient, hostInfo);
+      }
+    }
+    return targetClient;
+  }
+
+  async execute<T>(methodName: string, methodFunc: () => Promise<T>, methodArgs: any): Promise<T> {
+    const isEnabled: boolean = WrapperProperties.FAILURE_DETECTION_ENABLED.get(this.properties) as boolean;
+
+    if (!isEnabled || !SubscribedMethodHelper.NETWORK_BOUND_METHODS.includes(methodName)) {
+      return methodFunc();
+    }
+
+    const failureDetectionTimeMillis: number = WrapperProperties.FAILURE_DETECTION_TIME_MS.get(this.properties) as number;
+    const failureDetectionIntervalMillis: number = WrapperProperties.FAILURE_DETECTION_INTERVAL_MS.get(this.properties) as number;
+    const failureDetectionCount: number = WrapperProperties.FAILURE_DETECTION_COUNT.get(this.properties) as number;
+
+    let result: T;
+    let monitorContext: MonitorConnectionContext | null = null;
+
+    try {
+      logger.debug(Messages.get("HostMonitoringConnectionPlugin.activatedMonitoring", methodName));
+      const monitoringHostInfo: HostInfo = await this.getMonitoringHostInfo();
+
+      monitorContext = await this.monitorService.startMonitoring(
+        this.pluginService.getCurrentClient().targetClient,
+        monitoringHostInfo,
+        this.properties,
+        failureDetectionTimeMillis,
+        failureDetectionIntervalMillis,
+        failureDetectionCount
+      );
+
+      result = await methodFunc();
+    } finally {
+      if (monitorContext != null) {
+        await this.monitorService.stopMonitoring(monitorContext, this.pluginService.getCurrentClient().targetClient);
+
+        logger.debug(Messages.get("HostMonitoringConnectionPlugin.monitoringDeactivated", methodName));
+      }
+    }
+
+    return result;
+  }
+
+  private throwUnableToIdentifyConnection(host: HostInfo | null): never {
+    const provider: HostListProvider | null = this.pluginService.getHostListProvider();
+    throw new AwsWrapperError(
+      Messages.get(
+        "HostMonitoringConnectionPlugin.unableToIdentifyConnection",
+        host != null ? host.host : "unknown host",
+        provider != null ? provider.getHostProviderType() : "unknown provider"
+      )
+    );
+  }
+
+  async getMonitoringHostInfo(): Promise<HostInfo> {
+    if (this.monitoringHostInfo == null) {
+      this.monitoringHostInfo = this.pluginService.getCurrentHostInfo();
+      if (this.monitoringHostInfo == null) {
+        this.throwUnableToIdentifyConnection(null);
+      }
+      const rdsUrlType: RdsUrlType = this.rdsUtils.identifyRdsType(this.monitoringHostInfo.url);
+
+      try {
+        if (rdsUrlType.isRdsCluster) {
+          logger.debug(Messages.get("HostMonitoringConnectionPlugin.identifyClusterConnection"));
+          this.monitoringHostInfo = await this.pluginService.identifyConnection(this.pluginService.getCurrentClient().targetClient!);
+          if (this.monitoringHostInfo == null) {
+            const host: HostInfo | null = this.pluginService.getCurrentHostInfo();
+            this.throwUnableToIdentifyConnection(host);
+          }
+          await this.pluginService.fillAliases(this.pluginService.getCurrentClient().targetClient!, this.monitoringHostInfo);
+        }
+      } catch (error: any) {
+        if (!(error instanceof AwsWrapperError)) {
+          logger.debug(Messages.get("HostMonitoringConnectionPlugin.errorIdentifyingConnection", error.message));
+        }
+        throw error;
+      }
+    }
+
+    return this.monitoringHostInfo;
+  }
+
+  async notifyConnectionChanged(changes: Set<HostChangeOptions>): Promise<OldConnectionSuggestionAction> {
+    if (changes.has(HostChangeOptions.HOSTNAME) || changes.has(HostChangeOptions.HOST_CHANGED)) {
+      // Reset monitoring host info since the associated connection has changed.
+      this.monitoringHostInfo = null;
+    }
+    return OldConnectionSuggestionAction.NO_OPINION;
+  }
+
+  async releaseResources(): Promise<void> {
+    return this.monitorService.releaseResources();
+  }
+}

common/lib/plugins/efm2/host_monitoring2_plugin_factory.ts

@@ -0,0 +1,37 @@
+/*
+  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 { ConnectionPluginFactory } from "../../plugin_factory";
+import { PluginService } from "../../plugin_service";
+import { ConnectionPlugin } from "../../connection_plugin";
+import { RdsUtils } from "../../utils/rds_utils";
+import { AwsWrapperError } from "../../utils/errors";
+import { Messages } from "../../utils/messages";
+
+export class HostMonitoring2PluginFactory extends ConnectionPluginFactory {
+  private static hostMonitoring2Plugin: any;
+
+  async getInstance(pluginService: PluginService, properties: Map<string, any>): Promise<ConnectionPlugin> {
+    try {
+      if (!HostMonitoring2PluginFactory.hostMonitoring2Plugin) {
+        HostMonitoring2PluginFactory.hostMonitoring2Plugin = await import("./host_monitoring2_connection_plugin");
+      }
+      return new HostMonitoring2PluginFactory.hostMonitoring2Plugin.HostMonitoring2ConnectionPlugin(pluginService, properties, new RdsUtils());
+    } catch (error: any) {
+      throw new AwsWrapperError(Messages.get("ConnectionPluginChainBuilder.errorImportingPlugin", error.message, "HostMonitoringPlugin"));
+    }
+  }
+}