chore: refactor efm plugins

common/lib/connection_plugin_chain_builder.ts

@@ -31,7 +31,7 @@ import { StaleDnsPluginFactory } from "./plugins/stale_dns/stale_dns_plugin_fact
 import { FederatedAuthPluginFactory } from "./plugins/federated_auth/federated_auth_plugin_factory";
 import { ReadWriteSplittingPluginFactory } from "./plugins/read_write_splitting/read_write_splitting_plugin_factory";
 import { OktaAuthPluginFactory } from "./plugins/federated_auth/okta_auth_plugin_factory";
-import { HostMonitoringPluginFactory } from "./plugins/efm/host_monitoring_plugin_factory";
+import { HostMonitoringPluginFactory } from "./plugins/efm/v1/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 { ConnectionProviderManager } from "./connection_provider_manager";
@@ -41,7 +41,7 @@ import { LimitlessConnectionPluginFactory } from "./plugins/limitless/limitless_
 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 { HostMonitoring2PluginFactory } from "./plugins/efm/v2/host_monitoring2_plugin_factory";
 import { BlueGreenPluginFactory } from "./plugins/bluegreen/blue_green_plugin_factory";
 import { GlobalDbFailoverPluginFactory } from "./plugins/gdb_failover/global_db_failover_plugin_factory";
 import { FullServicesContainer } from "./utils/full_services_container";

common/lib/plugins/efm/base/connection_context.ts

@@ -0,0 +1,147 @@
+/*
+  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";
+import { TelemetryCounter } from "../../../utils/telemetry/telemetry_counter";
+import { logger } from "../../../../logutils";
+import { Messages } from "../../../utils/messages";
+import { getCurrentTimeNano } from "../../../utils/utils";
+
+/**
+ * Monitoring context for each connection. This contains each connection's criteria for whether a
+ * host should be considered unhealthy. The context is shared between the main task and the
+ * monitor task.
+ */
+export interface ConnectionContext {
+  readonly failureDetectionIntervalMillis: number;
+  readonly failureDetectionCount: number;
+  readonly expectedActiveMonitoringStartTimeNano: number;
+
+  isActiveContext(): boolean;
+  isHostUnhealthy(): boolean;
+  setInactive(): void;
+  abortConnection(): Promise<void>;
+  updateConnectionStatus(hostName: string, statusCheckStartTimeNano: number, statusCheckEndTimeNano: number, isValid: boolean): Promise<void>;
+}
+
+export class ConnectionContextImpl implements ConnectionContext {
+  readonly failureDetectionIntervalMillis: number;
+  readonly failureDetectionCount: number;
+  readonly expectedActiveMonitoringStartTimeNano: number;
+
+  private readonly failureDetectionTimeMillis: number;
+  private readonly connectionToAbortRef: WeakRef<ClientWrapper>;
+  private readonly abortedConnectionsCounter: TelemetryCounter;
+  private readonly startMonitorTimeNano: number;
+
+  private _activeContext: boolean = true;
+  private _hostUnhealthy: boolean = false;
+  private invalidHostStartTimeNano: number = 0;
+  private failureCount: number = 0;
+
+  constructor(
+    connectionToAbort: ClientWrapper,
+    failureDetectionTimeMillis: number,
+    failureDetectionIntervalMillis: number,
+    failureDetectionCount: number,
+    abortedConnectionsCounter: TelemetryCounter
+  ) {
+    this.connectionToAbortRef = new WeakRef(connectionToAbort);
+    this.failureDetectionTimeMillis = failureDetectionTimeMillis;
+    this.failureDetectionIntervalMillis = failureDetectionIntervalMillis;
+    this.failureDetectionCount = failureDetectionCount;
+    this.abortedConnectionsCounter = abortedConnectionsCounter;
+    this.startMonitorTimeNano = getCurrentTimeNano();
+    this.expectedActiveMonitoringStartTimeNano = this.startMonitorTimeNano + this.failureDetectionTimeMillis * 1_000_000;
+  }
+
+  isActiveContext(): boolean {
+    return this._activeContext;
+  }
+
+  isHostUnhealthy(): boolean {
+    return this._hostUnhealthy;
+  }
+
+  setInactive(): void {
+    this._activeContext = false;
+  }
+
+  async abortConnection(): Promise<void> {
+    const connectionToAbort = this.connectionToAbortRef.deref();
+    if (connectionToAbort == null || !this._activeContext) {
+      return;
+    }
+
+    try {
+      await connectionToAbort.abort();
+      this.abortedConnectionsCounter.inc();
+    } catch (error: any) {
+      // ignore
+      logger.debug(Messages.get("MonitorConnectionContext.errorAbortingConnection", error.message));
+    }
+  }
+
+  /**
+   * Update whether the connection is still valid if the total elapsed time has passed the
+   * grace period.
+   */
+  async updateConnectionStatus(hostName: string, statusCheckStartTimeNano: number, statusCheckEndTimeNano: number, isValid: boolean): Promise<void> {
+    if (!this._activeContext) {
+      return;
+    }
+
+    const totalElapsedTimeNano = statusCheckEndTimeNano - this.startMonitorTimeNano;
+
+    if (totalElapsedTimeNano > this.failureDetectionTimeMillis * 1_000_000) {
+      await this.setConnectionValid(hostName, isValid, statusCheckStartTimeNano, statusCheckEndTimeNano);
+    }
+  }
+
+  private async setConnectionValid(
+    hostName: string,
+    connectionValid: boolean,
+    statusCheckStartNano: number,
+    statusCheckEndNano: number
+  ): Promise<void> {
+    if (!connectionValid) {
+      this.failureCount++;
+
+      if (this.invalidHostStartTimeNano === 0) {
+        this.invalidHostStartTimeNano = statusCheckStartNano;
+      }
+
+      const invalidHostDurationNano = statusCheckEndNano - this.invalidHostStartTimeNano;
+      const maxInvalidHostDurationNano = this.failureDetectionIntervalMillis * Math.max(0, this.failureDetectionCount) * 1_000_000;
+
+      if (invalidHostDurationNano >= maxInvalidHostDurationNano) {
+        logger.debug(Messages.get("MonitorConnectionContext.hostDead", hostName));
+        this._hostUnhealthy = true;
+        await this.abortConnection();
+        return;
+      }
+
+      logger.debug(Messages.get("MonitorConnectionContext.hostNotResponding", hostName));
+      return;
+    }
+
+    this.failureCount = 0;
+    this.invalidHostStartTimeNano = 0;
+    this._hostUnhealthy = false;
+
+    logger.debug(Messages.get("MonitorConnectionContext.hostAlive", hostName));
+  }
+}

common/lib/plugins/efm/base/host_monitor.ts

@@ -0,0 +1,230 @@
+/*
+  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 { ConnectionContext } from "./connection_context";
+import { HostInfo } from "../../../host_info";
+import { PluginService } from "../../../plugin_service";
+import { ClientWrapper } from "../../../client_wrapper";
+import { WrapperProperties } from "../../../wrapper_property";
+import { logger } from "../../../../logutils";
+import { Messages } from "../../../utils/messages";
+import { getCurrentTimeNano } from "../../../utils/utils";
+import { getTimeInNanos } from "../../../utils/utils";
+import { TelemetryCounter } from "../../../utils/telemetry/telemetry_counter";
+import { TelemetryFactory } from "../../../utils/telemetry/telemetry_factory";
+import { TelemetryTraceLevel } from "../../../utils/telemetry/telemetry_trace_level";
+import { AbstractMonitor, MonitorState } from "../../../utils/monitoring/monitor";
+
+export interface HostMonitor {
+  startMonitoring(context: ConnectionContext): void;
+
+  stopMonitoring(context: ConnectionContext): void;
+
+  clearContexts(): void;
+
+  isStopped(): boolean;
+
+  run(): Promise<void>;
+
+  releaseResources(): Promise<void>;
+}
+
+type ConnectionStatus = [isValid: boolean, elapsedTimeNano: number];
+
+export class HostMonitorImpl extends AbstractMonitor implements HostMonitor {
+  private static readonly SLEEP_WHEN_INACTIVE_MILLIS = 100;
+  private static readonly MIN_CONNECTION_CHECK_TIMEOUT_MILLIS = 3000;
+  private static readonly MONITOR_TERMINATION_TIMEOUT_SEC = 30;
+
+  private readonly pluginService: PluginService;
+  private readonly telemetryFactory: TelemetryFactory;
+  private readonly hostInvalidCounter: TelemetryCounter;
+  private readonly properties: Map<string, any>;
+  private readonly hostInfo: HostInfo;
+  private readonly monitorDisposalTimeMillis: number;
+  private contexts: ConnectionContext[] = [];
+
+  private contextLastUsedTimestampNano: number;
+  private monitoringClient: ClientWrapper | null = null;
+  private delayTimeoutId: ReturnType<typeof setTimeout> | undefined;
+  private sleepTimeoutId: ReturnType<typeof setTimeout> | undefined;
+
+  constructor(pluginService: PluginService, hostInfo: HostInfo, properties: Map<string, any>, monitorDisposalTimeMillis: number) {
+    super(HostMonitorImpl.MONITOR_TERMINATION_TIMEOUT_SEC);
+    this.pluginService = pluginService;
+    this.telemetryFactory = this.pluginService.getTelemetryFactory();
+    this.hostInfo = hostInfo;
+    this.properties = properties;
+    this.monitorDisposalTimeMillis = monitorDisposalTimeMillis;
+    this.hostInvalidCounter = this.telemetryFactory.createCounter(`efm.nodeUnhealthy.count.${hostInfo.hostId || hostInfo.host}`);
+    this.contextLastUsedTimestampNano = getCurrentTimeNano();
+  }
+
+  startMonitoring(context: ConnectionContext): void {
+    if (this._stop) {
+      logger.warn(Messages.get("MonitorImpl.monitorIsStopped", this.hostInfo.host));
+    }
+
+    this.contextLastUsedTimestampNano = getCurrentTimeNano();
+    this.lastActivityTimestampNanos = getTimeInNanos();
+    this.contexts.push(context);
+  }
+
+  stopMonitoring(context: ConnectionContext): void {
+    if (!context) {
+      logger.warn(Messages.get("MonitorImpl.contextNullWarning"));
+      return;
+    }
+
+    context.setInactive();
+    this.contextLastUsedTimestampNano = getCurrentTimeNano();
+    this.lastActivityTimestampNanos = getTimeInNanos();
+  }
+
+  clearContexts(): void {
+    this.contexts.length = 0;
+  }
+
+  isStopped(): boolean {
+    return this._stop || this.state === MonitorState.STOPPED;
+  }
+
+  canDispose(): boolean {
+    return this.contexts.length === 0;
+  }
+
+  async monitor(): Promise<void> {
+    logger.debug(Messages.get("MonitorImpl.startMonitoring", this.hostInfo.host));
+
+    try {
+      while (!this._stop) {
+        try {
+          this.lastActivityTimestampNanos = getTimeInNanos();
+
+          const activeContexts = this.contexts.filter((ctx) => ctx.isActiveContext());
+
+          if (activeContexts.length > 0) {
+            this.contextLastUsedTimestampNano = getCurrentTimeNano();
+
+            const statusCheckStartTimeNano = getCurrentTimeNano();
+            const [isValid, elapsedTimeNano] = await this.checkConnectionStatus();
+
+            let delayMillis = -1;
+
+            for (const context of activeContexts) {
+              if (!context.isActiveContext()) {
+                continue;
+              }
+
+              await context.updateConnectionStatus(this.hostInfo.url, statusCheckStartTimeNano, statusCheckStartTimeNano + elapsedTimeNano, isValid);
+
+              if (
+                context.isActiveContext() &&
+                !context.isHostUnhealthy() &&
+                statusCheckStartTimeNano >= context.expectedActiveMonitoringStartTimeNano
+              ) {
+                if (delayMillis === -1 || delayMillis > context.failureDetectionIntervalMillis) {
+                  delayMillis = context.failureDetectionIntervalMillis;
+                }
+              }
+            }
+
+            this.contexts = this.contexts.filter((ctx) => ctx.isActiveContext() && !ctx.isHostUnhealthy());
+
+            if (delayMillis === -1) {
+              delayMillis = HostMonitorImpl.SLEEP_WHEN_INACTIVE_MILLIS;
+            } else {
+              delayMillis -= Math.round(elapsedTimeNano / 1_000_000);
+              if (delayMillis <= 0) {
+                delayMillis = HostMonitorImpl.MIN_CONNECTION_CHECK_TIMEOUT_MILLIS;
+              }
+            }
+
+            await new Promise<void>((resolve) => {
+              this.delayTimeoutId = setTimeout(resolve, delayMillis);
+            });
+          } else {
+            if (getCurrentTimeNano() - this.contextLastUsedTimestampNano >= this.monitorDisposalTimeMillis * 1_000_000) {
+              break;
+            }
+            await new Promise<void>((resolve) => {
+              this.sleepTimeoutId = setTimeout(resolve, HostMonitorImpl.SLEEP_WHEN_INACTIVE_MILLIS);
+            });
+          }
+        } catch (error: any) {
+          logger.debug(Messages.get("MonitorImpl.errorDuringMonitoringContinue", error.message));
+        }
+      }
+    } catch (error: any) {
+      logger.debug(Messages.get("MonitorImpl.errorDuringMonitoringStop", error.message));
+    }
+
+    logger.debug(Messages.get("MonitorImpl.stopMonitoring", this.hostInfo.host));
+  }
+
+  async close(): Promise<void> {
+    this.contexts.length = 0;
+    await this.closeMonitoringClient();
+  }
+
+  async releaseResources(): Promise<void> {
+    clearTimeout(this.delayTimeoutId);
+    clearTimeout(this.sleepTimeoutId);
+    await this.stop();
+  }
+
+  protected async checkConnectionStatus(): Promise<ConnectionStatus> {
+    const connectContext = this.telemetryFactory.openTelemetryContext("Connection status check", TelemetryTraceLevel.FORCE_TOP_LEVEL);
+    connectContext.setAttribute("url", this.hostInfo.host);
+    return await connectContext.start(async () => {
+      const startNanos = getCurrentTimeNano();
+      try {
+        if (this.monitoringClient != null && (await this.pluginService.isClientValid(this.monitoringClient))) {
+          return [true, getCurrentTimeNano() - startNanos];
+        }
+
+        await this.closeMonitoringClient();
+
+        const monitoringConnProperties = new Map(this.properties);
+        for (const key of monitoringConnProperties.keys()) {
+          if (!key.startsWith(WrapperProperties.MONITORING_PROPERTY_PREFIX)) {
+            continue;
+          }
+          monitoringConnProperties.set(key.substring(WrapperProperties.MONITORING_PROPERTY_PREFIX.length), this.properties.get(key));
+          monitoringConnProperties.delete(key);
+        }
+
+        this.monitoringClient = await this.pluginService.forceConnect(this.hostInfo, monitoringConnProperties);
+        return [true, getCurrentTimeNano() - startNanos];
+      } catch (error: any) {
+        this.hostInvalidCounter.inc();
+        await this.closeMonitoringClient();
+        return [false, getCurrentTimeNano() - startNanos];
+      }
+    });
+  }
+
+  private async closeMonitoringClient(): Promise<void> {
+    if (this.monitoringClient) {
+      try {
+        await this.pluginService.abortTargetClient(this.monitoringClient);
+      } catch {
+        // ignore
+      }
+      this.monitoringClient = null;
+    }
+  }
+}