diff --git a/common/lib/connection_plugin_chain_builder.ts b/common/lib/connection_plugin_chain_builder.ts index 38ec0a4f1..1f4fd2777 100644 --- a/common/lib/connection_plugin_chain_builder.ts +++ b/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"; diff --git a/common/lib/plugins/efm/base/connection_context.ts b/common/lib/plugins/efm/base/connection_context.ts new file mode 100644 index 000000000..bfcaffd56 --- /dev/null +++ b/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; + updateConnectionStatus(hostName: string, statusCheckStartTimeNano: number, statusCheckEndTimeNano: number, isValid: boolean): Promise; +} + +export class ConnectionContextImpl implements ConnectionContext { + readonly failureDetectionIntervalMillis: number; + readonly failureDetectionCount: number; + readonly expectedActiveMonitoringStartTimeNano: number; + + private readonly failureDetectionTimeMillis: number; + private readonly connectionToAbortRef: WeakRef; + 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 { + 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 { + 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 { + 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)); + } +} diff --git a/common/lib/plugins/efm/base/host_monitor.ts b/common/lib/plugins/efm/base/host_monitor.ts new file mode 100644 index 000000000..9cebcd58e --- /dev/null +++ b/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; + + releaseResources(): Promise; +} + +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; + private readonly hostInfo: HostInfo; + private readonly monitorDisposalTimeMillis: number; + private contexts: ConnectionContext[] = []; + + private contextLastUsedTimestampNano: number; + private monitoringClient: ClientWrapper | null = null; + private delayTimeoutId: ReturnType | undefined; + private sleepTimeoutId: ReturnType | undefined; + + constructor(pluginService: PluginService, hostInfo: HostInfo, properties: Map, 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 { + 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((resolve) => { + this.delayTimeoutId = setTimeout(resolve, delayMillis); + }); + } else { + if (getCurrentTimeNano() - this.contextLastUsedTimestampNano >= this.monitorDisposalTimeMillis * 1_000_000) { + break; + } + await new Promise((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 { + this.contexts.length = 0; + await this.closeMonitoringClient(); + } + + async releaseResources(): Promise { + clearTimeout(this.delayTimeoutId); + clearTimeout(this.sleepTimeoutId); + await this.stop(); + } + + protected async checkConnectionStatus(): Promise { + 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 { + if (this.monitoringClient) { + try { + await this.pluginService.abortTargetClient(this.monitoringClient); + } catch { + // ignore + } + this.monitoringClient = null; + } + } +} diff --git a/common/lib/plugins/efm/base/host_monitor_service.ts b/common/lib/plugins/efm/base/host_monitor_service.ts new file mode 100644 index 000000000..1c976a179 --- /dev/null +++ b/common/lib/plugins/efm/base/host_monitor_service.ts @@ -0,0 +1,117 @@ +/* + 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, ConnectionContextImpl } from "./connection_context"; +import { HostMonitor, HostMonitorImpl } from "./host_monitor"; +import { HostInfo } from "../../../host_info"; +import { ClientWrapper } from "../../../client_wrapper"; +import { WrapperProperties } from "../../../wrapper_property"; +import { AwsWrapperError } from "../../../utils/errors"; +import { Messages } from "../../../utils/messages"; +import { MonitorService } from "../../../utils/monitoring/monitor_service"; +import { MonitorInitializer } from "../../../utils/monitoring/monitor"; +import { TelemetryCounter } from "../../../utils/telemetry/telemetry_counter"; +import { FullServicesContainer } from "../../../utils/full_services_container"; + +export interface HostMonitorService { + startMonitoring( + connectionToAbort: ClientWrapper, + hostInfo: HostInfo, + properties: Map, + failureDetectionTimeMillis: number, + failureDetectionIntervalMillis: number, + failureDetectionCount: number + ): Promise; + + stopMonitoring(context: ConnectionContext): void; + + releaseResources(): Promise; +} + +export class HostMonitorServiceImpl implements HostMonitorService { + private static readonly MONITOR_DISPOSAL_TIME_NS = BigInt(10 * 60 * 1_000_000_000); // 10 minutes + private static readonly INACTIVE_TIMEOUT_NS = BigInt(3 * 60 * 1_000_000_000); // 3 minutes + + private readonly servicesContainer: FullServicesContainer; + private readonly coreMonitorService: MonitorService; + private readonly abortedConnectionsCounter: TelemetryCounter; + + constructor(servicesContainer: FullServicesContainer) { + this.servicesContainer = servicesContainer; + this.coreMonitorService = servicesContainer.monitorService; + const telemetryFactory = servicesContainer.telemetryFactory; + this.abortedConnectionsCounter = telemetryFactory.createCounter("efm.connections.aborted"); + + this.coreMonitorService.registerMonitorTypeIfAbsent( + HostMonitorImpl, + HostMonitorServiceImpl.MONITOR_DISPOSAL_TIME_NS, + HostMonitorServiceImpl.INACTIVE_TIMEOUT_NS, + new Set(), + undefined + ); + } + + async startMonitoring( + connectionToAbort: ClientWrapper, + hostInfo: HostInfo, + properties: Map, + failureDetectionTimeMillis: number, + failureDetectionIntervalMillis: number, + failureDetectionCount: number + ): Promise { + const monitorKey = hostInfo.hostId || hostInfo.host; + + const monitor = await this.getMonitor(monitorKey, hostInfo, properties); + + if (!monitor) { + throw new AwsWrapperError(Messages.get("MonitorService.startMonitoringNullMonitor", hostInfo.host)); + } + + const context = new ConnectionContextImpl( + connectionToAbort, + failureDetectionTimeMillis, + failureDetectionIntervalMillis, + failureDetectionCount, + this.abortedConnectionsCounter + ); + + monitor.startMonitoring(context); + return context; + } + + stopMonitoring(context: ConnectionContext): void { + context.setInactive(); + } + + private async getMonitor(monitorKey: string, hostInfo: HostInfo, properties: Map): Promise { + const monitorDisposalTimeMillis = WrapperProperties.MONITOR_DISPOSAL_TIME_MS.get(properties); + + const initializer: MonitorInitializer = { + createMonitor: (_servicesContainer) => + new HostMonitorImpl(this.servicesContainer.pluginService, hostInfo, properties, monitorDisposalTimeMillis) + }; + + return await this.coreMonitorService.runIfAbsent(HostMonitorImpl, monitorKey, this.servicesContainer, properties, initializer); + } + + async releaseResources(): Promise { + await this.coreMonitorService.stopAndRemoveMonitors(HostMonitorImpl); + } + + static async clearMonitors(monitorService: MonitorService): Promise { + await monitorService.stopAndRemoveMonitors(HostMonitorImpl); + } +} diff --git a/common/lib/plugins/efm/monitor.ts b/common/lib/plugins/efm/monitor.ts deleted file mode 100644 index 3094427f6..000000000 --- a/common/lib/plugins/efm/monitor.ts +++ /dev/null @@ -1,290 +0,0 @@ -/* - 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 { MonitorConnectionContext } from "./monitor_connection_context"; -import { HostInfo } from "../../host_info"; -import { PluginService } from "../../plugin_service"; -import { logger } from "../../../logutils"; -import { Messages } from "../../utils/messages"; -import { ClientWrapper } from "../../client_wrapper"; -import { getCurrentTimeNano, sleep } from "../../utils/utils"; -import { TelemetryFactory } from "../../utils/telemetry/telemetry_factory"; -import { TelemetryCounter } from "../../utils/telemetry/telemetry_counter"; -import { TelemetryTraceLevel } from "../../utils/telemetry/telemetry_trace_level"; -import { WrapperProperties } from "../../wrapper_property"; - -export interface Monitor { - startMonitoring(context: MonitorConnectionContext): void; - - stopMonitoring(context: MonitorConnectionContext): void; - - clearContexts(): void; - - isStopped(): boolean; - - run(): Promise; - - releaseResources(): Promise; - - endMonitoringClient(): Promise; -} - -class ConnectionStatus { - isValid: boolean; - elapsedTimeNano: number; - - constructor(isValid: boolean, elapsedTimeNano: number) { - this.isValid = isValid; - this.elapsedTimeNano = elapsedTimeNano; - } -} - -export class MonitorImpl implements Monitor { - private readonly SLEEP_WHEN_INACTIVE_MILLIS: number = 100; - private readonly MIN_CONNECTION_CHECK_TIMEOUT_MILLIS: number = 3000; - - private readonly activeContexts: MonitorConnectionContext[] = []; - private readonly newContexts: MonitorConnectionContext[] = []; - - private readonly pluginService: PluginService; - private readonly properties: Map; - private readonly hostInfo: HostInfo; - private readonly monitorDisposalTimeMillis: number; - private readonly telemetryFactory: TelemetryFactory; - private readonly instanceInvalidCounter: TelemetryCounter; - private contextLastUsedTimestampNanos: number; - private started = false; - private stopped: boolean = false; - private cancelled: boolean = false; - private monitoringClient: ClientWrapper | null = null; - private delayMillisTimeoutId: any; - private sleepWhenInactiveTimeoutId: any; - - constructor(pluginService: PluginService, hostInfo: HostInfo, properties: Map, monitorDisposalTimeMillis: number) { - this.pluginService = pluginService; - this.telemetryFactory = this.pluginService.getTelemetryFactory(); - this.properties = properties; - this.hostInfo = hostInfo; - this.monitorDisposalTimeMillis = monitorDisposalTimeMillis; - this.contextLastUsedTimestampNanos = getCurrentTimeNano(); - const instanceId = this.hostInfo.hostId ?? this.hostInfo.host; - this.instanceInvalidCounter = this.telemetryFactory.createCounter(`efm.hostUnhealthy.count.${instanceId}`); - } - - startRun() { - this.run(); - this.started = true; - } - - startMonitoring(context: MonitorConnectionContext): void { - if (this.stopped) { - logger.warn(Messages.get("MonitorImpl.monitorIsStopped", this.hostInfo.host)); - } - - const currentTimeNanos: number = getCurrentTimeNano(); - context.startMonitorTimeNano = currentTimeNanos; - this.contextLastUsedTimestampNanos = currentTimeNanos; - this.newContexts.push(context); - if (!this.started) { - this.startRun(); - } - } - - stopMonitoring(context: MonitorConnectionContext): void { - if (context == null) { - logger.warn(Messages.get("MonitorImpl.contextNullWarning")); - return; - } - - context.isActiveContext = false; - this.contextLastUsedTimestampNanos = getCurrentTimeNano(); - } - - async run(): Promise { - logger.debug(Messages.get("MonitorImpl.startMonitoring", this.hostInfo.host)); - - try { - while (!this.cancelled) { - try { - let newMonitorContext: MonitorConnectionContext | undefined; - let firstAddedNewMonitorContext: MonitorConnectionContext | null = null; - const currentTimeNano: number = getCurrentTimeNano(); - while ((newMonitorContext = this.newContexts?.shift()) != null) { - if (firstAddedNewMonitorContext === newMonitorContext) { - this.newContexts.push(newMonitorContext); - - break; - } - - if (newMonitorContext.isActiveContext) { - if (newMonitorContext.expectedActiveMonitoringStartTimeNano > currentTimeNano) { - this.newContexts.push(newMonitorContext); - - firstAddedNewMonitorContext = firstAddedNewMonitorContext ?? newMonitorContext; - } else { - this.activeContexts.push(newMonitorContext); - } - } - } - - if (this.activeContexts.length > 0) { - this.contextLastUsedTimestampNanos = getCurrentTimeNano(); - - const statusCheckStartTimeNanos: number = getCurrentTimeNano(); - this.contextLastUsedTimestampNanos = statusCheckStartTimeNanos; - - const status: ConnectionStatus = await this.checkConnectionStatus(); - let delayMillis: number = -1; - - let monitorContext: MonitorConnectionContext | undefined; - let firstAddedMonitorContext: MonitorConnectionContext | null = null; - - while ((monitorContext = this.activeContexts?.shift()) != null) { - // If context is already invalid, just skip it. - if (!monitorContext.isActiveContext) { - continue; - } - - if (firstAddedMonitorContext == monitorContext) { - // This context is already processed by this loop. - // Add it to the array and exit this loop. - - this.activeContexts?.push(monitorContext); - break; - } - - // Otherwise, process this context. - await monitorContext.updateConnectionStatus( - this.hostInfo.url, - statusCheckStartTimeNanos, - statusCheckStartTimeNanos + status.elapsedTimeNano, - status.isValid - ); - - if (monitorContext.isActiveContext && !monitorContext.isHostUnhealthy) { - this.activeContexts?.push(monitorContext); - - if (firstAddedMonitorContext == null) { - firstAddedMonitorContext = monitorContext; - } - - if (delayMillis === -1 || delayMillis > monitorContext.failureDetectionIntervalMillis) { - delayMillis = monitorContext.failureDetectionIntervalMillis; - } - } - } - - if (delayMillis === -1) { - // No active contexts. - delayMillis = this.SLEEP_WHEN_INACTIVE_MILLIS; - } else { - delayMillis -= Math.round(status.elapsedTimeNano / 1_000_000); - // Check for minimum delay between host health check; - if (delayMillis <= 0) { - delayMillis = this.MIN_CONNECTION_CHECK_TIMEOUT_MILLIS; - } - } - - await new Promise((resolve) => { - this.delayMillisTimeoutId = setTimeout(resolve, delayMillis); - }); - } else { - if (getCurrentTimeNano() - this.contextLastUsedTimestampNanos >= this.monitorDisposalTimeMillis * 1_000_000) { - break; - } - await new Promise((resolve) => { - this.sleepWhenInactiveTimeoutId = setTimeout(resolve, this.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)); - } finally { - this.stopped = true; - await this.endMonitoringClient(); - } - - logger.debug(Messages.get("MonitorImpl.stopMonitoring", this.hostInfo.host)); - } - - /** - * Check the status of the monitored server by sending a ping. - * - * @return whether the server is still alive and the elapsed time spent checking. - */ - async checkConnectionStatus(): Promise { - 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 { - const clientIsValid = this.monitoringClient && (await this.pluginService.isClientValid(this.monitoringClient)); - - if (this.monitoringClient !== null && clientIsValid) { - return Promise.resolve(new ConnectionStatus(clientIsValid, getCurrentTimeNano() - startNanos)); - } - - await this.endMonitoringClient(); - // Open a new connection. - const monitoringConnProperties: Map = 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); - } - - logger.debug(`Opening a monitoring connection to ${this.hostInfo.url}`); - this.monitoringClient = await this.pluginService.forceConnect(this.hostInfo, monitoringConnProperties); - logger.debug(`Successfully opened monitoring connection to ${this.monitoringClient.id} - ${this.hostInfo.url}`); - return Promise.resolve(new ConnectionStatus(true, getCurrentTimeNano() - startNanos)); - } catch (error: any) { - this.instanceInvalidCounter.inc(); - await this.endMonitoringClient(); - return Promise.resolve(new ConnectionStatus(false, getCurrentTimeNano() - startNanos)); - } - }); - } - - clearContexts(): void { - this.activeContexts.length = 0; - this.newContexts.length = 0; - } - - isStopped(): boolean { - return this.stopped || this.cancelled; - } - - async releaseResources() { - this.cancelled = true; - clearTimeout(this.delayMillisTimeoutId); - clearTimeout(this.sleepWhenInactiveTimeoutId); - await this.endMonitoringClient(); - // Allow time for monitor loop to close. - await sleep(500); - } - - async endMonitoringClient() { - if (this.monitoringClient) { - await this.pluginService.abortTargetClient(this.monitoringClient); - this.monitoringClient = null; - } - } -} diff --git a/common/lib/plugins/efm/monitor_connection_context.ts b/common/lib/plugins/efm/monitor_connection_context.ts deleted file mode 100644 index a68893228..000000000 --- a/common/lib/plugins/efm/monitor_connection_context.ts +++ /dev/null @@ -1,155 +0,0 @@ -/* - 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 { Monitor } from "./monitor"; -import { logger, uniqueId } from "../../../logutils"; -import { Messages } from "../../utils/messages"; -import { ClientWrapper } from "../../client_wrapper"; -import { sleep } from "../../utils/utils"; -import { AwsWrapperError } from "../../utils/errors"; -import { PluginService } from "../../plugin_service"; -import { TelemetryCounter } from "../../utils/telemetry/telemetry_counter"; - -export class MonitorConnectionContext { - readonly failureDetectionIntervalMillis: number; - private readonly failureDetectionTimeMillis: number; - private readonly failureDetectionCount: number; - private readonly telemetryAbortedConnectionCounter: TelemetryCounter; - readonly clientToAbort: ClientWrapper; - readonly monitor: Monitor; - readonly pluginService: PluginService; - - isActiveContext: boolean = true; - isHostUnhealthy: boolean = false; - startMonitorTimeNano: number = 0; - expectedActiveMonitoringStartTimeNano: number = 0; - private invalidHostStartTimeNano: number = 0; - private abortedConnectionCounter: number = 0; - failureCount: number = 0; - id: string = uniqueId("_monitorContext"); - - constructor( - monitor: Monitor, - clientToAbort: ClientWrapper, - failureDetectionTimeMillis: number, - failureDetectionIntervalMillis: number, - failureDetectionCount: number, - pluginService: PluginService, - telemetryAbortedConnectionCounter: TelemetryCounter - ) { - this.monitor = monitor; - this.clientToAbort = clientToAbort; - this.failureDetectionTimeMillis = failureDetectionTimeMillis; - this.failureDetectionIntervalMillis = failureDetectionIntervalMillis; - this.failureDetectionCount = failureDetectionCount; - this.pluginService = pluginService; - this.telemetryAbortedConnectionCounter = telemetryAbortedConnectionCounter; - } - - resetInvalidHostStartTimeNano(): void { - this.invalidHostStartTimeNano = 0; - } - - isInvalidHostStartTimeDefined(): boolean { - return this.invalidHostStartTimeNano > 0; - } - - async abortConnection(): Promise { - if (this.clientToAbort == null || !this.isActiveContext) { - return Promise.resolve(); - } - - try { - await this.pluginService.abortTargetClient(this.clientToAbort); - this.telemetryAbortedConnectionCounter.inc(); - } catch (error: any) { - // ignore - logger.debug(Messages.get("MonitorConnectionContext.errorAbortingConnection", error.message)); - } - this.abortedConnectionCounter++; - } - - /** - * Update whether the connection is still valid if the total elapsed time has passed the grace period. - * @param hostName A host name for logging purposes. - * @param statusCheckStartNano The time when connection status check started in nanos. - * @param statusCheckEndNano The time when connection status check ended in nanos. - * @param isValid Whether the connection is valid. - */ - async updateConnectionStatus(hostName: string, statusCheckStartNano: number, statusCheckEndNano: number, isValid: boolean): Promise { - if (!this.isActiveContext) { - return; - } - - const totalElapsedTimeNano: number = statusCheckEndNano - this.startMonitorTimeNano; - if (totalElapsedTimeNano > this.failureDetectionTimeMillis * 1000000) { - await this.setConnectionValid(hostName, isValid, statusCheckStartNano, statusCheckEndNano); - } - } - - /** - * Set whether the connection to the server is still valid based on the monitoring settings set in the {@link AwsClient}. - * - *

These monitoring settings include: - * - *

    - *
  • {@code failureDetectionInterval} - *
  • {@code failureDetectionTime} - *
  • {@code failureDetectionCount} - *
- * - * @param hostName A host name for logging purposes. - * @param connectionValid Boolean indicating whether the server is still responsive. - * @param statusCheckStartNano The time when connection status check started in nanos. - * @param statusCheckEndNano The time when connection status check ended in nanos. - * @protected - */ - async setConnectionValid(hostName: string, connectionValid: boolean, statusCheckStartNano: number, statusCheckEndNano: number) { - if (!connectionValid) { - this.failureCount++; - - if (!this.isInvalidHostStartTimeDefined()) { - this.invalidHostStartTimeNano = statusCheckStartNano; - } - - 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; - await this.abortConnection(); - return; - } - - logger.debug(Messages.get("MonitorConnectionContext.hostNotResponding", hostName)); - return; - } - - this.failureCount = 0; - this.resetInvalidHostStartTimeNano(); - this.isHostUnhealthy = false; - - logger.debug(Messages.get("MonitorConnectionContext.hostAlive", hostName)); - } - - async trackHealthStatus() { - while (!this.isHostUnhealthy && this.isActiveContext) { - await sleep(100); - } - - throw new AwsWrapperError("trackHealthStatus stopped"); - } -} diff --git a/common/lib/plugins/efm/monitor_service.ts b/common/lib/plugins/efm/monitor_service.ts deleted file mode 100644 index c81639f52..000000000 --- a/common/lib/plugins/efm/monitor_service.ts +++ /dev/null @@ -1,174 +0,0 @@ -/* - 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 { MonitorConnectionContext } from "./monitor_connection_context"; -import { HostInfo } from "../../host_info"; -import { AwsWrapperError, IllegalArgumentError } from "../../utils/errors"; -import { Monitor, MonitorImpl } from "./monitor"; -import { WrapperProperties } from "../../wrapper_property"; -import { PluginService } from "../../plugin_service"; -import { Messages } from "../../utils/messages"; -import { SlidingExpirationCacheWithCleanupTask } from "../../utils/sliding_expiration_cache_with_cleanup_task"; -import { ClientWrapper } from "../../client_wrapper"; - -export interface MonitorService { - startMonitoring( - clientToAbort: ClientWrapper, - hostKeys: Set, - hostInfo: HostInfo, - properties: Map, - failureDetectionTimeMillis: number, - failureDetectionIntervalMillis: number, - failureDetectionCount: number - ): Promise; - - stopMonitoring(context: MonitorConnectionContext): Promise; - - stopMonitoringForAllConnections(hostKeys: Set): void; - - releaseResources(): Promise; -} - -export class MonitorServiceImpl implements MonitorService { - private static readonly CACHE_CLEANUP_NANOS = BigInt(60_000_000_000); - protected static readonly monitors: SlidingExpirationCacheWithCleanupTask = new SlidingExpirationCacheWithCleanupTask( - MonitorServiceImpl.CACHE_CLEANUP_NANOS, - undefined, - async (monitor: Monitor) => { - await monitor.releaseResources(); - }, - "efm/MonitorServiceImpl.monitors" - ); - private readonly pluginService: PluginService; - private cachedMonitorHostKeys: Set | undefined; - private cachedMonitorRef: WeakRef | undefined; - monitorSupplier = (pluginService: PluginService, hostInfo: HostInfo, properties: Map, monitorDisposalTimeMillis: number) => - new MonitorImpl(pluginService, hostInfo, properties, monitorDisposalTimeMillis); - - constructor(pluginService: PluginService) { - this.pluginService = pluginService; - } - - async startMonitoring( - clientToAbort: ClientWrapper, - hostKeys: Set, - hostInfo: HostInfo, - properties: Map, - failureDetectionTimeMillis: number, - failureDetectionIntervalMillis: number, - failureDetectionCount: number - ): Promise { - if (hostKeys.size === 0) { - throw new IllegalArgumentError(Messages.get("MonitorService.emptyAliasSet", hostInfo.host)); - } - - let monitor: Monitor | null = this.cachedMonitorRef?.deref() ?? null; - - if (!monitor || (monitor && monitor.isStopped()) || this.cachedMonitorHostKeys?.size === 0 || this.cachedMonitorHostKeys !== hostKeys) { - monitor = await this.getMonitor(hostKeys, hostInfo, properties); - if (monitor) { - this.cachedMonitorRef = new WeakRef(monitor); - this.cachedMonitorHostKeys = hostKeys; - } - } - - const telemetryFactory = this.pluginService.getTelemetryFactory(); - const abortedConnectionsCounter = telemetryFactory.createCounter("efm.connections.aborted"); - - if (monitor) { - const context = new MonitorConnectionContext( - monitor, - clientToAbort, - failureDetectionTimeMillis, - failureDetectionIntervalMillis, - failureDetectionCount, - this.pluginService, - abortedConnectionsCounter - ); - monitor.startMonitoring(context); - return context; - } - - throw new AwsWrapperError(Messages.get("MonitorService.startMonitoringNullMonitor", hostInfo.host)); - } - - async stopMonitoring(context: MonitorConnectionContext) { - context.monitor.stopMonitoring(context); - await context.monitor.endMonitoringClient(); - } - - stopMonitoringForAllConnections(hostKeys: Set) { - let monitor: Monitor; - for (const hostKey of hostKeys) { - monitor = MonitorServiceImpl.monitors.get(hostKey); - if (monitor) { - monitor.clearContexts(); - return; - } - } - } - - async getMonitor(hostKeys: Set, hostInfo: HostInfo, properties: Map): Promise { - let monitor: Monitor; - let anyHostKey: string; - for (const hostKey of hostKeys) { - monitor = MonitorServiceImpl.monitors.get(hostKey); - anyHostKey = hostKey; - if (monitor) { - break; - } - } - - const cacheExpirationNanos = BigInt(WrapperProperties.MONITOR_DISPOSAL_TIME_MS.get(properties) * 1_000_000); - if (anyHostKey && (!monitor || (monitor && monitor.isStopped()))) { - monitor = MonitorServiceImpl.monitors.computeIfAbsent( - anyHostKey, - () => this.monitorSupplier(this.pluginService, hostInfo, properties, WrapperProperties.MONITOR_DISPOSAL_TIME_MS.get(properties)), - cacheExpirationNanos - ); - - if (monitor && monitor.isStopped()) { - await monitor.releaseResources(); - MonitorServiceImpl.monitors.remove(anyHostKey); - monitor = this.monitorSupplier(this.pluginService, hostInfo, properties, WrapperProperties.MONITOR_DISPOSAL_TIME_MS.get(properties)); - } - } - - if (monitor) { - this.populateMonitorMap(hostKeys, monitor, cacheExpirationNanos); - return monitor; - } - - return null; - } - - private populateMonitorMap(hostKeys: Set, monitor: Monitor, cacheExpirationNanos: bigint) { - for (const hostKey of hostKeys) { - MonitorServiceImpl.monitors.putIfAbsent(hostKey, monitor, cacheExpirationNanos); - } - } - - async releaseResources() { - await MonitorServiceImpl.monitors.clear(); - this.cachedMonitorHostKeys = undefined; - this.cachedMonitorRef = undefined; - } - - // Used for performance testing. - static async clearMonitors() { - await MonitorServiceImpl.monitors.clear(); - } -} diff --git a/common/lib/plugins/efm/host_monitoring_connection_plugin.ts b/common/lib/plugins/efm/v1/host_monitoring_connection_plugin.ts similarity index 64% rename from common/lib/plugins/efm/host_monitoring_connection_plugin.ts rename to common/lib/plugins/efm/v1/host_monitoring_connection_plugin.ts index 59ad986fb..3c7548d24 100644 --- a/common/lib/plugins/efm/host_monitoring_connection_plugin.ts +++ b/common/lib/plugins/efm/v1/host_monitoring_connection_plugin.ts @@ -14,37 +14,42 @@ limitations under the License. */ -import { HostInfo, AwsWrapperError, UnavailableHostError, HostAvailability } from "../../"; -import { PluginService } from "../../plugin_service"; -import { HostChangeOptions } from "../../host_change_options"; -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 { 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"; +import { HostInfo } from "../../../host_info"; +import { AwsWrapperError, UnavailableHostError } from "../../../utils/errors"; +import { PluginService } from "../../../plugin_service"; +import { HostChangeOptions } from "../../../host_change_options"; +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 { ConnectionContext } from "../base/connection_context"; +import { HostMonitorService, HostMonitorServiceImpl } from "../base/host_monitor_service"; +import { logger, uniqueId } from "../../../../logutils"; +import { Messages } from "../../../utils/messages"; +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"; +import { FullServicesContainer } from "../../../utils/full_services_container"; +import { sleep } from "../../../utils/utils"; export class HostMonitoringConnectionPlugin extends AbstractConnectionPlugin implements CanReleaseResources { + private static readonly UNHEALTHY_STATE = Symbol("unhealthy"); + id: string = uniqueId("_efmPlugin"); - private readonly properties: Map; - private pluginService: PluginService; - private rdsUtils: RdsUtils; + protected readonly properties: Map; + protected readonly pluginService: PluginService; + protected readonly rdsUtils: RdsUtils; + protected readonly monitorService: HostMonitorService; private monitoringHostInfo: HostInfo | null = null; - private monitorService: MonitorService; - constructor(pluginService: PluginService, properties: Map, rdsUtils: RdsUtils = new RdsUtils(), monitorService?: MonitorServiceImpl) { + constructor(servicesContainer: FullServicesContainer, properties: Map, monitorService?: HostMonitorService) { super(); - this.pluginService = pluginService; + this.pluginService = servicesContainer.pluginService; this.properties = properties; - this.rdsUtils = rdsUtils; - this.monitorService = monitorService ?? new MonitorServiceImpl(pluginService); + this.rdsUtils = new RdsUtils(); + this.monitorService = monitorService ?? new HostMonitorServiceImpl(servicesContainer); } getSubscribedMethods(): Set { @@ -78,20 +83,19 @@ export class HostMonitoringConnectionPlugin extends AbstractConnectionPlugin imp 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; + const failureDetectionTimeMillis: number = WrapperProperties.FAILURE_DETECTION_TIME_MS.get(this.properties); + const failureDetectionIntervalMillis: number = WrapperProperties.FAILURE_DETECTION_INTERVAL_MS.get(this.properties); + const failureDetectionCount: number = WrapperProperties.FAILURE_DETECTION_COUNT.get(this.properties); let result: T; - let monitorContext: MonitorConnectionContext | null = null; + let context: ConnectionContext | null = null; try { logger.debug(Messages.get("HostMonitoringConnectionPlugin.activatedMonitoring", methodName)); const monitoringHostInfo: HostInfo = await this.getMonitoringHostInfo(); - monitorContext = await this.monitorService.startMonitoring( + context = await this.monitorService.startMonitoring( this.pluginService.getCurrentClient().targetClient, - monitoringHostInfo.allAliases, monitoringHostInfo, this.properties, failureDetectionTimeMillis, @@ -99,19 +103,21 @@ export class HostMonitoringConnectionPlugin extends AbstractConnectionPlugin imp failureDetectionCount ); - result = await Promise.race([monitorContext.trackHealthStatus(), methodFunc()]) - .then((result: any) => { - return result; - }) - .catch((error: any) => { - throw error; + const methodPromise = methodFunc(); + const raceResult = await Promise.race([this.waitForUnhealthy(context), methodPromise]); + if (raceResult === HostMonitoringConnectionPlugin.UNHEALTHY_STATE) { + methodPromise.catch(() => { + // Attach a no-op rejection handler to prevent it from throwing an unhandled promise rejection error when waitForUnhealthy times out first. }); + throw new AwsWrapperError("Host monitoring detected unhealthy host"); + } + result = raceResult as T; } finally { - if (monitorContext != null) { - await this.monitorService.stopMonitoring(monitorContext); + if (context != null) { + this.monitorService.stopMonitoring(context); logger.debug(Messages.get("HostMonitoringConnectionPlugin.monitoringDeactivated", methodName)); - if (monitorContext.isHostUnhealthy) { + if (context.isHostUnhealthy()) { const monitoringHostInfo = await this.getMonitoringHostInfo(); const targetClient = this.pluginService.getCurrentClient().targetClient; let isClientValid = false; @@ -135,6 +141,13 @@ export class HostMonitoringConnectionPlugin extends AbstractConnectionPlugin imp return result; } + private async waitForUnhealthy(context: ConnectionContext): Promise { + while (!context.isHostUnhealthy() && context.isActiveContext()) { + await sleep(100); + } + return HostMonitoringConnectionPlugin.UNHEALTHY_STATE; + } + private throwUnableToIdentifyConnection(host: HostInfo | null): never { const provider: HostListProvider | null = this.pluginService.getHostListProvider(); throw new AwsWrapperError( @@ -167,6 +180,7 @@ export class HostMonitoringConnectionPlugin extends AbstractConnectionPlugin imp const host: HostInfo | null = this.pluginService.getCurrentHostInfo(); this.throwUnableToIdentifyConnection(host); } + // Update identified HostInfo for the current connection await this.pluginService.setCurrentClient(this.pluginService.getCurrentClient().targetClient!, this.monitoringHostInfo); } @@ -180,15 +194,9 @@ export class HostMonitoringConnectionPlugin extends AbstractConnectionPlugin imp } async notifyConnectionChanged(changes: Set): Promise { - if (changes.has(HostChangeOptions.WENT_DOWN) || changes.has(HostChangeOptions.HOST_DELETED)) { - const aliases = (await this.getMonitoringHostInfo())?.allAliases; - if (aliases && aliases.size > 0) { - this.monitorService.stopMonitoringForAllConnections(aliases); - } + if (changes.has(HostChangeOptions.HOSTNAME) || changes.has(HostChangeOptions.HOST_CHANGED)) { + this.monitoringHostInfo = null; } - - // Reset monitoring host info since the associated connection has changed. - this.monitoringHostInfo = null; return OldConnectionSuggestionAction.NO_OPINION; } diff --git a/common/lib/plugins/efm/host_monitoring_plugin_factory.ts b/common/lib/plugins/efm/v1/host_monitoring_plugin_factory.ts similarity index 72% rename from common/lib/plugins/efm/host_monitoring_plugin_factory.ts rename to common/lib/plugins/efm/v1/host_monitoring_plugin_factory.ts index d6b430954..49f3d6284 100644 --- a/common/lib/plugins/efm/host_monitoring_plugin_factory.ts +++ b/common/lib/plugins/efm/v1/host_monitoring_plugin_factory.ts @@ -14,12 +14,11 @@ limitations under the License. */ -import { ConnectionPluginFactory } from "../../plugin_factory"; -import { ConnectionPlugin } from "../../connection_plugin"; -import { RdsUtils } from "../../utils/rds_utils"; -import { AwsWrapperError } from "../../utils/errors"; -import { Messages } from "../../utils/messages"; -import { FullServicesContainer } from "../../utils/full_services_container"; +import { ConnectionPluginFactory } from "../../../plugin_factory"; +import { ConnectionPlugin } from "../../../connection_plugin"; +import { AwsWrapperError } from "../../../utils/errors"; +import { Messages } from "../../../utils/messages"; +import { FullServicesContainer } from "../../../utils/full_services_container"; export class HostMonitoringPluginFactory extends ConnectionPluginFactory { private static hostMonitoringPlugin: any; @@ -29,11 +28,7 @@ export class HostMonitoringPluginFactory extends ConnectionPluginFactory { if (!HostMonitoringPluginFactory.hostMonitoringPlugin) { HostMonitoringPluginFactory.hostMonitoringPlugin = await import("./host_monitoring_connection_plugin"); } - return new HostMonitoringPluginFactory.hostMonitoringPlugin.HostMonitoringConnectionPlugin( - servicesContainer.pluginService, - properties, - new RdsUtils() - ); + return new HostMonitoringPluginFactory.hostMonitoringPlugin.HostMonitoringConnectionPlugin(servicesContainer, properties); } catch (error: any) { throw new AwsWrapperError(Messages.get("ConnectionPluginChainBuilder.errorImportingPlugin", error.message, "HostMonitoringPlugin")); } diff --git a/common/lib/plugins/efm/v2/host_monitoring2_connection_plugin.ts b/common/lib/plugins/efm/v2/host_monitoring2_connection_plugin.ts new file mode 100644 index 000000000..3ca1194dd --- /dev/null +++ b/common/lib/plugins/efm/v2/host_monitoring2_connection_plugin.ts @@ -0,0 +1,70 @@ +/* + 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 { WrapperProperties } from "../../../wrapper_property"; +import { ConnectionContext } from "../base/connection_context"; +import { HostMonitorService, HostMonitorServiceImpl } from "../base/host_monitor_service"; +import { HostMonitoringConnectionPlugin } from "../v1/host_monitoring_connection_plugin"; +import { logger, uniqueId } from "../../../../logutils"; +import { Messages } from "../../../utils/messages"; +import { SubscribedMethodHelper } from "../../../utils/subscribed_method_helper"; +import { FullServicesContainer } from "../../../utils/full_services_container"; + +export class HostMonitoring2ConnectionPlugin extends HostMonitoringConnectionPlugin { + id: string = uniqueId("_efm2Plugin"); + + constructor(servicesContainer: FullServicesContainer, properties: Map, monitorService?: HostMonitorService) { + super(servicesContainer, properties, monitorService ?? new HostMonitorServiceImpl(servicesContainer)); + } + + async execute(methodName: string, methodFunc: () => Promise, methodArgs: any): Promise { + const isEnabled: boolean = WrapperProperties.FAILURE_DETECTION_ENABLED.get(this.properties); + + if (!isEnabled || !SubscribedMethodHelper.NETWORK_BOUND_METHODS.includes(methodName)) { + return methodFunc(); + } + + const failureDetectionTimeMillis: number = WrapperProperties.FAILURE_DETECTION_TIME_MS.get(this.properties); + const failureDetectionIntervalMillis: number = WrapperProperties.FAILURE_DETECTION_INTERVAL_MS.get(this.properties); + const failureDetectionCount: number = WrapperProperties.FAILURE_DETECTION_COUNT.get(this.properties); + + let result: T; + let context: ConnectionContext | null = null; + + try { + logger.debug(Messages.get("HostMonitoringConnectionPlugin.activatedMonitoring", methodName)); + const monitoringHostInfo = await this.getMonitoringHostInfo(); + + context = await this.monitorService.startMonitoring( + this.pluginService.getCurrentClient().targetClient, + monitoringHostInfo, + this.properties, + failureDetectionTimeMillis, + failureDetectionIntervalMillis, + failureDetectionCount + ); + + result = await methodFunc(); + } finally { + if (context != null) { + this.monitorService.stopMonitoring(context); + logger.debug(Messages.get("HostMonitoringConnectionPlugin.monitoringDeactivated", methodName)); + } + } + + return result; + } +} diff --git a/common/lib/plugins/efm2/host_monitoring2_plugin_factory.ts b/common/lib/plugins/efm/v2/host_monitoring2_plugin_factory.ts similarity index 72% rename from common/lib/plugins/efm2/host_monitoring2_plugin_factory.ts rename to common/lib/plugins/efm/v2/host_monitoring2_plugin_factory.ts index 0ac0319f4..e21e4c8be 100644 --- a/common/lib/plugins/efm2/host_monitoring2_plugin_factory.ts +++ b/common/lib/plugins/efm/v2/host_monitoring2_plugin_factory.ts @@ -14,12 +14,11 @@ limitations under the License. */ -import { ConnectionPluginFactory } from "../../plugin_factory"; -import { ConnectionPlugin } from "../../connection_plugin"; -import { RdsUtils } from "../../utils/rds_utils"; -import { AwsWrapperError } from "../../utils/errors"; -import { Messages } from "../../utils/messages"; -import { FullServicesContainer } from "../../utils/full_services_container"; +import { ConnectionPluginFactory } from "../../../plugin_factory"; +import { ConnectionPlugin } from "../../../connection_plugin"; +import { AwsWrapperError } from "../../../utils/errors"; +import { Messages } from "../../../utils/messages"; +import { FullServicesContainer } from "../../../utils/full_services_container"; export class HostMonitoring2PluginFactory extends ConnectionPluginFactory { private static hostMonitoring2Plugin: any; @@ -29,11 +28,7 @@ export class HostMonitoring2PluginFactory extends ConnectionPluginFactory { if (!HostMonitoring2PluginFactory.hostMonitoring2Plugin) { HostMonitoring2PluginFactory.hostMonitoring2Plugin = await import("./host_monitoring2_connection_plugin"); } - return new HostMonitoring2PluginFactory.hostMonitoring2Plugin.HostMonitoring2ConnectionPlugin( - servicesContainer.pluginService, - properties, - new RdsUtils() - ); + return new HostMonitoring2PluginFactory.hostMonitoring2Plugin.HostMonitoring2ConnectionPlugin(servicesContainer, properties); } catch (error: any) { throw new AwsWrapperError(Messages.get("ConnectionPluginChainBuilder.errorImportingPlugin", error.message, "HostMonitoringPlugin")); } diff --git a/common/lib/plugins/efm2/host_monitoring2_connection_plugin.ts b/common/lib/plugins/efm2/host_monitoring2_connection_plugin.ts deleted file mode 100644 index 036b6dcd3..000000000 --- a/common/lib/plugins/efm2/host_monitoring2_connection_plugin.ts +++ /dev/null @@ -1,163 +0,0 @@ -/* - 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 "../../"; -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; - private pluginService: PluginService; - private rdsUtils: RdsUtils; - private monitoringHostInfo: HostInfo | null = null; - private monitorService: MonitorService; - - constructor(pluginService: PluginService, properties: Map, rdsUtils: RdsUtils = new RdsUtils(), monitorService?: MonitorServiceImpl) { - super(); - this.pluginService = pluginService; - this.properties = properties; - this.rdsUtils = rdsUtils; - this.monitorService = monitorService ?? new MonitorServiceImpl(pluginService); - } - - getSubscribedMethods(): Set { - return new Set(["*"]); - } - - async connect( - hostInfo: HostInfo, - props: Map, - isInitialConnection: boolean, - connectFunc: () => Promise - ): Promise { - 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(methodName: string, methodFunc: () => Promise, methodArgs: any): Promise { - const isEnabled: boolean = WrapperProperties.FAILURE_DETECTION_ENABLED.get(this.properties); - - if (!isEnabled || !SubscribedMethodHelper.NETWORK_BOUND_METHODS.includes(methodName)) { - return methodFunc(); - } - - const failureDetectionTimeMillis: number = WrapperProperties.FAILURE_DETECTION_TIME_MS.get(this.properties); - const failureDetectionIntervalMillis: number = WrapperProperties.FAILURE_DETECTION_INTERVAL_MS.get(this.properties); - const failureDetectionCount: number = WrapperProperties.FAILURE_DETECTION_COUNT.get(this.properties); - - 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 { - if (this.monitoringHostInfo) { - return this.monitoringHostInfo; - } - 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): Promise { - 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 { - await this.monitorService.releaseResources(); - } -} diff --git a/common/lib/plugins/efm2/monitor.ts b/common/lib/plugins/efm2/monitor.ts deleted file mode 100644 index f1d72a61a..000000000 --- a/common/lib/plugins/efm2/monitor.ts +++ /dev/null @@ -1,319 +0,0 @@ -/* - 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 { MonitorConnectionContext } from "./monitor_connection_context"; -import { HostInfo } from "../../host_info"; -import { PluginService } from "../../plugin_service"; -import { logger } from "../../../logutils"; -import { Messages } from "../../utils/messages"; -import { ClientWrapper } from "../../client_wrapper"; -import { getCurrentTimeNano, sleep } from "../../utils/utils"; -import { TelemetryFactory } from "../../utils/telemetry/telemetry_factory"; -import { TelemetryCounter } from "../../utils/telemetry/telemetry_counter"; -import { TelemetryTraceLevel } from "../../utils/telemetry/telemetry_trace_level"; -import { HostAvailability } from "../../host_availability/host_availability"; -import { MapUtils } from "../../utils/map_utils"; -import { WrapperProperties } from "../../wrapper_property"; - -export interface Monitor { - startMonitoring(context: MonitorConnectionContext): void; - - run(): Promise; - - canDispose(): boolean; - - endMonitoringClient(): Promise; - - releaseResources(): Promise; -} - -export class MonitorImpl implements Monitor { - private static readonly TASK_SLEEP_MILLIS: number = 100; - private activeContexts: WeakRef[] = []; - static newContexts: Map>> = new Map>>(); - private readonly pluginService: PluginService; - private readonly telemetryFactory: TelemetryFactory; - private readonly properties: Map; - private readonly hostInfo: HostInfo; - private stopped: boolean = false; - - private monitoringClient: ClientWrapper | null = null; - - private readonly failureDetectionTimeNano: number; - private readonly failureDetectionIntervalNanos: number; - private readonly failureDetectionCount: number; - - private invalidHostStartTimeNano: number; - private failureCount: number; - private hostUnhealthy: boolean = false; - private readonly abortedConnectionsCounter: TelemetryCounter; - private delayMillisTimeoutId: any; - private sleepWhenHostHealthyTimeoutId: any; - - constructor( - pluginService: PluginService, - hostInfo: HostInfo, - properties: Map, - failureDetectionTimeMillis: number, - failureDetectionIntervalMillis: number, - failureDetectionCount: number, - abortedConnectionsCounter: TelemetryCounter - ) { - this.pluginService = pluginService; - this.telemetryFactory = this.pluginService.getTelemetryFactory(); - this.hostInfo = hostInfo; - this.properties = properties; - this.failureDetectionTimeNano = failureDetectionTimeMillis * 1000000; - this.failureDetectionIntervalNanos = failureDetectionIntervalMillis * 1000000; - this.failureDetectionCount = failureDetectionCount; - this.abortedConnectionsCounter = abortedConnectionsCounter; - - const hostId: string = this.hostInfo.hostId ?? this.hostInfo.host; - - this.telemetryFactory.createGauge(`efm2.newContexts.size.${hostId}`, () => MonitorImpl.newContexts.size === Number.MAX_SAFE_INTEGER); - - this.telemetryFactory.createGauge(`efm2.activeContexts.size.${hostId}`, () => this.activeContexts.length === Number.MAX_SAFE_INTEGER); - - this.telemetryFactory.createGauge(`efm2.hostHealthy.${hostId}`, () => (this.hostUnhealthy ? 0 : 1)); - - Promise.race([this.newContextRun(), this.run()]).finally(() => { - this.stopped = true; - }); - } - - canDispose(): boolean { - return this.activeContexts.length === 0 && MonitorImpl.newContexts.size === 0; - } - - startMonitoring(context: MonitorConnectionContext): void { - if (this.isStopped()) { - logger.warn(Messages.get("MonitorImpl.monitorIsStopped", this.hostInfo.host)); - } - - const startMonitorTimeNano = getCurrentTimeNano() + this.failureDetectionTimeNano; - const connectionQueue = MapUtils.computeIfAbsent( - MonitorImpl.newContexts, - startMonitorTimeNano, - () => new Array>() - ); - connectionQueue.push(new WeakRef(context)); - } - - async newContextRun(): Promise { - logger.debug(Messages.get("MonitorImpl.startMonitoringTaskNewContext", this.hostInfo.host)); - - try { - while (!this.isStopped()) { - const currentTimeNanos = getCurrentTimeNano(); - // Get entries with key (that is a time in nanos) less than current time. - const processedKeys: number[] = new Array(); - for (const [key, val] of MonitorImpl.newContexts.entries()) { - if (key < currentTimeNanos) { - const queue: Array> = val; - - processedKeys.push(key); - // Each value of found entry is a queue of monitoring contexts awaiting active monitoring. - // Add all contexts to an active monitoring contexts queue. - // Ignore disposed contexts. - let monitorContextRef: WeakRef | undefined; - - while ((monitorContextRef = queue?.shift()) != null) { - const monitorContext: MonitorConnectionContext = monitorContextRef?.deref() ?? null; - if (monitorContext && monitorContext.isActive()) { - this.activeContexts.push(monitorContextRef); - } - } - } - } - processedKeys.forEach((key) => { - MonitorImpl.newContexts.delete(key); - }); - await sleep(1000); - } - return; - } catch (err) { - // do nothing, exit task - } - logger.debug(Messages.get("MonitorImpl.stopMonitoringTaskNewContext", this.hostInfo.host)); - } - - async run(): Promise { - logger.debug(Messages.get("MonitorImpl.startMonitoring", this.hostInfo.host)); - - try { - while (!this.isStopped()) { - try { - if (this.activeContexts.length === 0 && !this.hostUnhealthy) { - await new Promise((resolve) => { - this.delayMillisTimeoutId = setTimeout(resolve, MonitorImpl.TASK_SLEEP_MILLIS); - }); - continue; - } - - const statusCheckStartTimeNanos: number = getCurrentTimeNano(); - const isValid = await this.checkConnectionStatus(); - const statusCheckEndTimeNanos: number = getCurrentTimeNano(); - - await this.updateHostHealthStatus(isValid, statusCheckStartTimeNanos, statusCheckEndTimeNanos); - - if (this.hostUnhealthy) { - this.pluginService.setAvailability(this.hostInfo.aliases, HostAvailability.NOT_AVAILABLE); - } - const tmpActiveContexts: WeakRef[] = []; - - let monitorContextRef: WeakRef | undefined; - - while ((monitorContextRef = this.activeContexts?.shift()) != null) { - if (this.isStopped()) { - break; - } - - const monitorContext: MonitorConnectionContext = monitorContextRef?.deref() ?? null; - - if (!monitorContext) { - continue; - } - - if (this.hostUnhealthy) { - // Kill connection - monitorContext.setHostUnhealthy(true); - const clientToAbort = monitorContext.getClient(); - - monitorContext.setInactive(); - if (clientToAbort != null) { - await this.endMonitoringClient(clientToAbort); - - this.abortedConnectionsCounter.inc(); - } - } else if (monitorContext && monitorContext.isActive()) { - tmpActiveContexts.push(monitorContextRef); - } - } - - // activeContexts is empty now and tmpActiveContexts contains all yet active contexts - // Add active contexts back to the queue. - this.activeContexts.push(...tmpActiveContexts); - - const delayMillis = (this.failureDetectionIntervalNanos - (statusCheckEndTimeNanos - statusCheckStartTimeNanos)) / 1000000; - - await new Promise((resolve) => { - this.delayMillisTimeoutId = setTimeout( - resolve, - delayMillis < MonitorImpl.TASK_SLEEP_MILLIS ? MonitorImpl.TASK_SLEEP_MILLIS : delayMillis - ); - }); - } catch (error: any) { - logger.debug(Messages.get("MonitorImpl.errorDuringMonitoringContinue", error.message)); - } - } - } catch (error: any) { - logger.debug(Messages.get("MonitorImpl.errorDuringMonitoringStop", error.message)); - } finally { - await this.endMonitoringClient(); - } - - logger.debug(Messages.get("MonitorImpl.stopMonitoring", this.hostInfo.host)); - } - - /** - * Check the status of the monitored server by sending a ping. - * - * @return whether the server is still alive and the elapsed time spent checking. - */ - async checkConnectionStatus(): Promise { - const connectContext = this.telemetryFactory.openTelemetryContext("Connection status check", TelemetryTraceLevel.FORCE_TOP_LEVEL); - connectContext.setAttribute("url", this.hostInfo.host); - try { - if (!(await this.pluginService.isClientValid(this.monitoringClient))) { - // Open a new connection. - const monitoringConnProperties: Map = 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); - } - - logger.debug(`Opening a monitoring connection to ${this.hostInfo.url}`); - this.monitoringClient = await this.pluginService.forceConnect(this.hostInfo, monitoringConnProperties); - logger.debug(`Successfully opened monitoring connection to ${this.monitoringClient.id} - ${this.hostInfo.url}`); - return true; - } - return true; - } catch (error: any) { - return false; - } - } - - isStopped(): boolean { - return this.stopped; - } - - updateHostHealthStatus(connectionValid: boolean, statusCheckStartNano: number, statusCheckEndNano: number): Promise { - if (!connectionValid) { - this.failureCount++; - - if (this.invalidHostStartTimeNano === 0) { - this.invalidHostStartTimeNano = statusCheckStartNano; - } - - const invalidHostDurationNano = statusCheckEndNano - this.invalidHostStartTimeNano; - const maxInvalidHostDurationNano = this.failureDetectionIntervalNanos * Math.max(0, this.failureDetectionCount - 1); - - if (this.failureCount >= this.failureDetectionCount || invalidHostDurationNano >= maxInvalidHostDurationNano) { - logger.debug(Messages.get("MonitorConnectionContext.hostDead", this.hostInfo.host)); - this.hostUnhealthy = true; - return Promise.resolve(); - } - logger.debug(Messages.get("MonitorConnectionContext.hostNotResponding", this.hostInfo.host)); - return Promise.resolve(); - } - - if (this.failureCount > 0) { - // Host is back alive - logger.debug(Messages.get("MonitorConnectionContext.hostAlive", this.hostInfo.host)); - } - this.failureCount = 0; - this.invalidHostStartTimeNano = 0; - this.hostUnhealthy = false; - } - - async releaseResources() { - this.stopped = true; - clearTimeout(this.delayMillisTimeoutId); - clearTimeout(this.sleepWhenHostHealthyTimeoutId); - this.activeContexts = null; - await this.endMonitoringClient(); - // Allow time for monitor loop to close. - await sleep(500); - } - - async endMonitoringClient(clientToAbort?: ClientWrapper) { - try { - if (clientToAbort) { - await this.pluginService.abortTargetClient(clientToAbort); - } else if (this.monitoringClient) { - await this.pluginService.abortTargetClient(this.monitoringClient); - this.monitoringClient = null; - } - this.stopped = true; - } catch (error: any) { - // ignore - logger.debug(Messages.get("MonitorConnectionContext.errorAbortingConnection", error.message)); - } - } -} diff --git a/common/lib/plugins/efm2/monitor_connection_context.ts b/common/lib/plugins/efm2/monitor_connection_context.ts deleted file mode 100644 index 156b41963..000000000 --- a/common/lib/plugins/efm2/monitor_connection_context.ts +++ /dev/null @@ -1,59 +0,0 @@ -/* - 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 { uniqueId } from "../../../logutils"; -import { ClientWrapper } from "../../client_wrapper"; - -/** - * Monitoring context for each connection. This contains each connection's criteria for whether a - * server should be considered unhealthy. The context is shared between the main task and the monitor task. - */ -export class MonitorConnectionContext { - private clientToAbortRef: WeakRef | undefined; - isHostUnhealthy: boolean = false; - id: string = uniqueId("_monitorContext"); - - /** - * Constructor. - * - * @param clientToAbort A reference to the connection associated with this context that will be aborted. - */ - constructor(clientToAbort: ClientWrapper) { - if (clientToAbort) { - this.clientToAbortRef = new WeakRef(clientToAbort); - } - } - - setHostUnhealthy(hostUnhealthy: boolean) { - this.isHostUnhealthy = hostUnhealthy; - } - - shouldAbort(): boolean { - return this.isHostUnhealthy && this.clientToAbortRef != null; - } - - setInactive(): void { - this.clientToAbortRef = null; - } - - getClient(): ClientWrapper | null { - return this.clientToAbortRef?.deref() ?? null; - } - - isActive() { - return !!this.clientToAbortRef?.deref(); - } -} diff --git a/common/lib/plugins/efm2/monitor_service.ts b/common/lib/plugins/efm2/monitor_service.ts deleted file mode 100644 index 40a788178..000000000 --- a/common/lib/plugins/efm2/monitor_service.ts +++ /dev/null @@ -1,157 +0,0 @@ -/* - 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 { MonitorConnectionContext } from "./monitor_connection_context"; -import { HostInfo } from "../../host_info"; -import { AwsWrapperError } from "../../utils/errors"; -import { Monitor, MonitorImpl } from "./monitor"; -import { WrapperProperties } from "../../wrapper_property"; -import { PluginService } from "../../plugin_service"; -import { Messages } from "../../utils/messages"; -import { TelemetryCounter } from "../../utils/telemetry/telemetry_counter"; -import { TelemetryFactory } from "../../utils/telemetry/telemetry_factory"; -import { ClientWrapper } from "../../client_wrapper"; -import { logger } from "../../../logutils"; -import { SlidingExpirationCacheWithCleanupTask } from "../../utils/sliding_expiration_cache_with_cleanup_task"; - -export interface MonitorService { - startMonitoring( - clientToAbort: ClientWrapper, - hostInfo: HostInfo, - properties: Map, - failureDetectionTimeMillis: number, - failureDetectionIntervalMillis: number, - failureDetectionCount: number - ): Promise; - - /** - * Stop monitoring for a connection represented by the given {@link MonitorConnectionContext}. - * Removes the context from the {@link MonitorImpl}. - * - * @param context The {@link MonitorConnectionContext} representing a connection. - * @param clientToAbort A reference to the connection associated with this context that will be aborted. - */ - stopMonitoring(context: MonitorConnectionContext, clientToAbort: ClientWrapper): Promise; - - releaseResources(): Promise; -} - -export class MonitorServiceImpl implements MonitorService { - private static readonly CACHE_CLEANUP_NANOS = BigInt(60_000_000_000); - - protected static readonly monitors: SlidingExpirationCacheWithCleanupTask = new SlidingExpirationCacheWithCleanupTask( - MonitorServiceImpl.CACHE_CLEANUP_NANOS, - (monitor: Monitor) => monitor.canDispose(), - async (monitor: Monitor) => { - { - try { - await monitor.releaseResources(); - } catch (error) { - // ignore - } - } - }, - "efm2/MonitorServiceImpl.monitors" - ); - private readonly pluginService: PluginService; - private telemetryFactory: TelemetryFactory; - private readonly abortedConnectionsCounter: TelemetryCounter; - monitorSupplier = ( - pluginService: PluginService, - hostInfo: HostInfo, - properties: Map, - failureDetectionTimeMillis: number, - failureDetectionIntervalMillis: number, - failureDetectionCount: number, - abortedConnectionsCounter: TelemetryCounter - ) => - new MonitorImpl( - pluginService, - hostInfo, - properties, - failureDetectionTimeMillis, - failureDetectionIntervalMillis, - failureDetectionCount, - abortedConnectionsCounter - ); - - constructor(pluginService: PluginService) { - this.pluginService = pluginService; - this.telemetryFactory = pluginService.getTelemetryFactory(); - this.abortedConnectionsCounter = this.telemetryFactory.createCounter("efm2.connections.aborted"); - } - - async startMonitoring( - clientToAbort: ClientWrapper, - hostInfo: HostInfo, - properties: Map, - failureDetectionTimeMillis: number, - failureDetectionIntervalMillis: number, - failureDetectionCount: number - ): Promise { - const monitor = await this.getMonitor(hostInfo, properties, failureDetectionTimeMillis, failureDetectionIntervalMillis, failureDetectionCount); - - if (monitor) { - const context = new MonitorConnectionContext(clientToAbort); - monitor.startMonitoring(context); - return context; - } - - throw new AwsWrapperError(Messages.get("MonitorService.startMonitoringNullMonitor", hostInfo.host)); - } - - async stopMonitoring(context: MonitorConnectionContext, clientToAbort: ClientWrapper): Promise { - context.setInactive(); - if (context.shouldAbort()) { - try { - await clientToAbort.abort(); - this.abortedConnectionsCounter.inc(); - } catch (error) { - // ignore - logger.debug(Messages.get("MonitorConnectionContext.errorAbortingConnection", error.message)); - } - } - } - - async getMonitor( - hostInfo: HostInfo, - properties: Map, - failureDetectionTimeMillis: number, - failureDetectionIntervalMillis: number, - failureDetectionCount: number - ): Promise { - const monitorKey: string = `${failureDetectionTimeMillis.toString()} ${failureDetectionIntervalMillis.toString()} ${failureDetectionCount.toString()} ${hostInfo.host}`; - const cacheExpirationNanos = BigInt(WrapperProperties.MONITOR_DISPOSAL_TIME_MS.get(properties) * 1_000_000); - return MonitorServiceImpl.monitors.computeIfAbsent( - monitorKey, - () => - this.monitorSupplier( - this.pluginService, - hostInfo, - properties, - failureDetectionTimeMillis, - failureDetectionIntervalMillis, - failureDetectionCount, - this.abortedConnectionsCounter - ), - cacheExpirationNanos - ); - } - - async releaseResources() { - await MonitorServiceImpl.monitors.clear(); - } -} diff --git a/common/lib/profile/driver_configuration_profiles.ts b/common/lib/profile/driver_configuration_profiles.ts index 958d5dead..a9e2ffb62 100644 --- a/common/lib/profile/driver_configuration_profiles.ts +++ b/common/lib/profile/driver_configuration_profiles.ts @@ -17,7 +17,7 @@ import { ConfigurationProfile } from "./configuration_profile"; import { ConfigurationProfilePresetCodes } from "./configuration_profile_codes"; import { WrapperProperties } from "../wrapper_property"; -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 { ReadWriteSplittingPluginFactory } from "../plugins/read_write_splitting/read_write_splitting_plugin_factory"; diff --git a/common/lib/utils/monitoring/monitor.ts b/common/lib/utils/monitoring/monitor.ts index 81dadea43..5bf93ffcd 100644 --- a/common/lib/utils/monitoring/monitor.ts +++ b/common/lib/utils/monitoring/monitor.ts @@ -78,7 +78,7 @@ export abstract class AbstractMonitor implements Monitor { this.monitorPromise = this.run(); } - protected async run(): Promise { + async run(): Promise { try { this.state = MonitorState.RUNNING; this.lastActivityTimestampNanos = BigInt(Date.now() * 1_000_000); diff --git a/index.ts b/index.ts index 6292539cd..74a840bed 100644 --- a/index.ts +++ b/index.ts @@ -30,8 +30,8 @@ export { DefaultPlugin } from "./common/lib/plugins/default_plugin"; export { ReadWriteSplittingPlugin } from "./common/lib/plugins/read_write_splitting/read_write_splitting_plugin"; export { FailoverPlugin } from "./common/lib/plugins/failover/failover_plugin"; export { Failover2Plugin } from "./common/lib/plugins/failover2/failover2_plugin"; -export { HostMonitoringConnectionPlugin } from "./common/lib/plugins/efm/host_monitoring_connection_plugin"; -export { HostMonitoring2ConnectionPlugin } from "./common/lib/plugins/efm2/host_monitoring2_connection_plugin"; +export { HostMonitoringConnectionPlugin } from "./common/lib/plugins/efm/v1/host_monitoring_connection_plugin"; +export { HostMonitoring2ConnectionPlugin } from "./common/lib/plugins/efm/v2/host_monitoring2_connection_plugin"; export { DeveloperConnectionPlugin } from "./common/lib/plugins/dev/developer_connection_plugin"; export { BlueGreenPlugin } from "./common/lib/plugins/bluegreen/blue_green_plugin"; export { AuroraConnectionTrackerPlugin } from "./common/lib/plugins/connection_tracker/aurora_connection_tracker_plugin"; diff --git a/tests/integration/container/tests/performance.test.ts b/tests/integration/container/tests/performance.test.ts index f15d1268e..43d10ac1a 100644 --- a/tests/integration/container/tests/performance.test.ts +++ b/tests/integration/container/tests/performance.test.ts @@ -25,7 +25,7 @@ import { WrapperProperties, PluginManager } from "../../../../index"; import { features } from "./config"; import { PerfStat } from "./utils/perf_stat"; import { PerfTestUtility } from "./utils/perf_util"; -import { MonitorServiceImpl } from "../../../../common/lib/plugins/efm/monitor_service"; +import { HostMonitorServiceImpl as MonitorServiceImpl } from "../../../../common/lib/plugins/efm/base/host_monitor_service"; const itIf = features.includes(TestEnvironmentFeatures.FAILOVER_SUPPORTED) && diff --git a/tests/unit/host_monitoring_plugin.test.ts b/tests/unit/host_monitoring_plugin.test.ts index 02b70dd81..88c7c012c 100644 --- a/tests/unit/host_monitoring_plugin.test.ts +++ b/tests/unit/host_monitoring_plugin.test.ts @@ -1,12 +1,12 @@ /* 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. @@ -15,15 +15,12 @@ */ import { WrapperProperties } from "../../common/lib/wrapper_property"; -import { HostMonitoringConnectionPlugin } from "../../common/lib/plugins/efm/host_monitoring_connection_plugin"; -import { DatabaseDialect } from "../../common/lib/database_dialect/database_dialect"; -import { PgDatabaseDialect } from "../../pg/lib/dialect/pg_database_dialect"; +import { HostMonitoringConnectionPlugin } from "../../common/lib/plugins/efm/v1/host_monitoring_connection_plugin"; import { anything, instance, mock, reset, verify, when } from "ts-mockito"; -import { RdsUtils } from "../../common/lib/utils/rds_utils"; -import { MonitorServiceImpl } from "../../common/lib/plugins/efm/monitor_service"; +import { HostMonitorServiceImpl } from "../../common/lib/plugins/efm/base/host_monitor_service"; import { PluginServiceImpl } from "../../common/lib/plugin_service"; -import { MonitorConnectionContext } from "../../common/lib/plugins/efm/monitor_connection_context"; -import { AwsWrapperError, HostAvailability, HostInfo } from "../../common/lib"; +import { ConnectionContext, ConnectionContextImpl } from "../../common/lib/plugins/efm/base/connection_context"; +import { AwsWrapperError, HostInfo } from "../../common/lib"; import { RdsUrlType } from "../../common/lib/utils/rds_url_type"; import { HostChangeOptions } from "../../common/lib/host_change_options"; import { OldConnectionSuggestionAction } from "../../common/lib/old_connection_suggestion_action"; @@ -31,6 +28,8 @@ import { Messages } from "../../common/lib/utils/messages"; import { AwsPGClient } from "../../pg/lib"; import { MySQLClientWrapper } from "../../common/lib/mysql_client_wrapper"; import { MySQL2DriverDialect } from "../../mysql/lib/dialect/mysql2_driver_dialect"; +import { FullServicesContainer } from "../../common/lib/utils/full_services_container"; +import { NullTelemetryFactory } from "../../common/lib/utils/telemetry/null_telemetry_factory"; const FAILURE_DETECTION_TIME = 10; const FAILURE_DETECTION_INTERVAL = 100; @@ -38,12 +37,10 @@ const FAILURE_DETECTION_COUNT = 5; const MONITOR_METHOD_NAME = "query"; const NO_MONITOR_METHOD_NAME = "end"; -const mockDialect: DatabaseDialect = mock(PgDatabaseDialect); -const mockMonitorService: MonitorServiceImpl = mock(MonitorServiceImpl); -const mockMonitorConnectionContext: MonitorConnectionContext = mock(MonitorConnectionContext); +const mockMonitorService: HostMonitorServiceImpl = mock(HostMonitorServiceImpl); +const mockConnectionContext: ConnectionContextImpl = mock(ConnectionContextImpl); const mockPluginService: PluginServiceImpl = mock(PluginServiceImpl); const mockClient: AwsPGClient = mock(AwsPGClient); -const mockRdsUtils = mock(RdsUtils); const mockHostInfo1 = mock(HostInfo); const mockHostInfo2 = mock(HostInfo); @@ -59,18 +56,19 @@ function incrementQueryCounter() { const mockClientWrapper = new MySQLClientWrapper(undefined, mock(HostInfo), new Map(), new MySQL2DriverDialect()); function initDefaultMockReturns() { - when(mockDialect.getHostAliasQuery()).thenReturn("any"); - when(mockMonitorService.startMonitoring(anything(), anything(), anything(), anything(), anything(), anything(), anything())).thenResolve( - instance(mockMonitorConnectionContext) + when(mockMonitorService.startMonitoring(anything(), anything(), anything(), anything(), anything(), anything())).thenResolve( + instance(mockConnectionContext) ); + when(mockConnectionContext.isActiveContext()).thenReturn(true); + when(mockConnectionContext.isHostUnhealthy()).thenReturn(false); when(mockHostInfo1.host).thenReturn("host"); when(mockHostInfo1.port).thenReturn(1234); when(mockHostInfo2.host).thenReturn("host"); when(mockHostInfo2.port).thenReturn(1234); when(mockPluginService.getCurrentHostInfo()).thenReturn(instance(mockHostInfo1)); + when(mockPluginService.getRoutedHostInfo()).thenReturn(instance(mockHostInfo1)); when(mockPluginService.getCurrentClient()).thenReturn(instance(mockClient)); when(mockClient.targetClient).thenReturn(mockClientWrapper); - when(mockRdsUtils.identifyRdsType(anything())).thenReturn(RdsUrlType.RDS_INSTANCE); properties.set(WrapperProperties.FAILURE_DETECTION_ENABLED.name, true); properties.set(WrapperProperties.FAILURE_DETECTION_TIME_MS.name, FAILURE_DETECTION_TIME); @@ -79,13 +77,18 @@ function initDefaultMockReturns() { } function initializePlugin() { - plugin = new HostMonitoringConnectionPlugin(instance(mockPluginService), properties, instance(mockRdsUtils), instance(mockMonitorService)); + const servicesContainer = { + pluginService: instance(mockPluginService), + telemetryFactory: new NullTelemetryFactory() + } as unknown as FullServicesContainer; + + plugin = new HostMonitoringConnectionPlugin(servicesContainer, properties, instance(mockMonitorService)); } describe("host monitoring plugin test", () => { beforeEach(() => { reset(mockMonitorService); - reset(mockMonitorConnectionContext); + reset(mockConnectionContext); reset(mockPluginService); reset(mockClient); @@ -99,7 +102,7 @@ describe("host monitoring plugin test", () => { initializePlugin(); await plugin.execute(MONITOR_METHOD_NAME, incrementQueryCounter, {}); - verify(mockMonitorService.startMonitoring(anything(), anything(), anything(), anything(), anything(), anything(), anything())).never(); + verify(mockMonitorService.startMonitoring(anything(), anything(), anything(), anything(), anything(), anything())).never(); verify(mockMonitorService.stopMonitoring(anything())).never(); expect(queryCounter).toBe(1); }); @@ -108,7 +111,7 @@ describe("host monitoring plugin test", () => { initializePlugin(); await plugin.execute(NO_MONITOR_METHOD_NAME, incrementQueryCounter, {}); - verify(mockMonitorService.startMonitoring(anything(), anything(), anything(), anything(), anything(), anything(), anything())).never(); + verify(mockMonitorService.startMonitoring(anything(), anything(), anything(), anything(), anything(), anything())).never(); verify(mockMonitorService.stopMonitoring(anything())).never(); expect(queryCounter).toBe(1); }); @@ -117,7 +120,7 @@ describe("host monitoring plugin test", () => { initializePlugin(); await plugin.execute(MONITOR_METHOD_NAME, incrementQueryCounter, {}); - verify(mockMonitorService.startMonitoring(anything(), anything(), anything(), anything(), anything(), anything(), anything())).once(); + verify(mockMonitorService.startMonitoring(anything(), anything(), anything(), anything(), anything(), anything())).once(); verify(mockMonitorService.stopMonitoring(anything())).once(); expect(queryCounter).toBe(1); }); @@ -126,55 +129,25 @@ describe("host monitoring plugin test", () => { initializePlugin(); const expectedError = new AwsWrapperError("Error thrown during isClientValid"); - when(mockMonitorConnectionContext.isHostUnhealthy).thenReturn(true); + when(mockConnectionContext.isHostUnhealthy()).thenReturn(true); when(mockPluginService.isClientValid(mockClientWrapper)).thenThrow(expectedError); await expect(plugin.execute(MONITOR_METHOD_NAME, incrementQueryCounter, {})).rejects.toThrow(expectedError); }); - it("execute cleanup when abort connection throws error", async () => { + it("execute cleanup when connection is invalid throws unavailable host error", async () => { initializePlugin(); when(mockPluginService.isClientValid(mockClientWrapper)).thenResolve(false); - when(mockMonitorConnectionContext.isHostUnhealthy).thenReturn(true); + when(mockConnectionContext.isHostUnhealthy()).thenReturn(true); const expectedError = new AwsWrapperError(Messages.get("HostMonitoringConnectionPlugin.unavailableHost", "host")); await expect(plugin.execute(MONITOR_METHOD_NAME, incrementQueryCounter, {})).rejects.toThrow(expectedError); - verify(mockPluginService.setAvailability(anything(), HostAvailability.NOT_AVAILABLE)).once(); }); - it("test connect", async () => { + it("notify connection changed resets monitoring host info", async () => { initializePlugin(); - when(mockRdsUtils.identifyRdsType(anything())).thenReturn(RdsUrlType.RDS_WRITER_CLUSTER); - await plugin.connect(instance(mockHostInfo1), properties, true, () => { - return Promise.resolve(mockClientWrapper); - }); - verify(mockPluginService.fillAliases(anything(), anything())).once(); + const result = await plugin.notifyConnectionChanged(new Set([HostChangeOptions.HOSTNAME])); + expect(result).toBe(OldConnectionSuggestionAction.NO_OPINION); }); - - it.each([[HostChangeOptions.WENT_DOWN], [HostChangeOptions.HOST_DELETED]])( - "notify connection changed when node went down", - async (options: HostChangeOptions) => { - initializePlugin(); - - await plugin.execute(MONITOR_METHOD_NAME, incrementQueryCounter, {}); - - const aliases1 = new Set(["alias1", "alias2"]); - const aliases2 = new Set(["alias3", "alias4"]); - - when(mockHostInfo1.allAliases).thenReturn(aliases1); - when(mockHostInfo2.allAliases).thenReturn(aliases2); - when(mockPluginService.getCurrentHostInfo()).thenReturn(instance(mockHostInfo1)); - - expect(await plugin.notifyConnectionChanged(new Set([options]))).toBe(OldConnectionSuggestionAction.NO_OPINION); - // NodeKeys should contain {"alias1", "alias2"}. - verify(mockMonitorService.stopMonitoringForAllConnections(aliases1)).once(); - - when(mockPluginService.getCurrentHostInfo()).thenReturn(instance(mockHostInfo2)); - expect(await plugin.notifyConnectionChanged(new Set([options]))).toBe(OldConnectionSuggestionAction.NO_OPINION); - // NotifyConnectionChanged should reset the monitoringHostSpec. - // NodeKeys should contain {"alias3", "alias4"} - verify(mockMonitorService.stopMonitoringForAllConnections(aliases2)).once(); - } - ); }); diff --git a/tests/unit/monitor_connection_context.test.ts b/tests/unit/monitor_connection_context.test.ts index 47e5eab01..3f11b3de3 100644 --- a/tests/unit/monitor_connection_context.test.ts +++ b/tests/unit/monitor_connection_context.test.ts @@ -1,12 +1,12 @@ /* 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. @@ -14,113 +14,131 @@ limitations under the License. */ -import { instance, mock, spy, verify } from "ts-mockito"; -import { MonitorConnectionContext } from "../../common/lib/plugins/efm/monitor_connection_context"; -import { MonitorImpl } from "../../common/lib/plugins/efm/monitor"; -import { PluginServiceImpl } from "../../common/lib/plugin_service"; +import { mock } from "ts-mockito"; +import { ConnectionContextImpl } from "../../common/lib/plugins/efm/base/connection_context"; import { NullTelemetryFactory } from "../../common/lib/utils/telemetry/null_telemetry_factory"; import { MySQLClientWrapper } from "../../common/lib/mysql_client_wrapper"; +import { HostInfo } from "../../common/lib"; +import { MySQL2DriverDialect } from "../../mysql/lib/dialect/mysql2_driver_dialect"; +import { getCurrentTimeNano } from "../../common/lib/utils/utils"; -const mockPluginService: PluginServiceImpl = mock(PluginServiceImpl); -const mockMonitor = mock(MonitorImpl); -const mockTargetClient = { - end() { - throw new Error("close"); - } -}; -const mockClientWrapper = mock(MySQLClientWrapper); +const mockClientWrapper = new MySQLClientWrapper(undefined, mock(HostInfo), new Map(), new MySQL2DriverDialect()); const FAILURE_DETECTION_TIME_MILLIS = 10; const FAILURE_DETECTION_INTERVAL_MILLIS = 100; const FAILURE_DETECTION_COUNT = 3; const VALIDATION_INTERVAL_MILLIS = 50; -let context: MonitorConnectionContext; +let context: ConnectionContextImpl; -describe("monitor connection context test", () => { +describe("connection context test", () => { beforeEach(() => { - context = new MonitorConnectionContext( - instance(mockMonitor), - null, + context = new ConnectionContextImpl( + mockClientWrapper, FAILURE_DETECTION_TIME_MILLIS, FAILURE_DETECTION_INTERVAL_MILLIS, FAILURE_DETECTION_COUNT, - instance(mockPluginService), new NullTelemetryFactory().createCounter("counter") ); }); it("isHostUnhealthy with valid connection", async () => { - const currentTimeNano = Date.now(); - await context.setConnectionValid("test-node", true, currentTimeNano, currentTimeNano); - expect(context.isHostUnhealthy).toBe(false); - expect(context.failureCount).toBe(0); + const ctx = new ConnectionContextImpl( + mockClientWrapper, + FAILURE_DETECTION_TIME_MILLIS, + FAILURE_DETECTION_INTERVAL_MILLIS, + FAILURE_DETECTION_COUNT, + new NullTelemetryFactory().createCounter("counter") + ); + // Timestamps must exceed the grace period relative to context's startMonitorTimeNano (hrtime-based) + const checkStart = getCurrentTimeNano() + (FAILURE_DETECTION_TIME_MILLIS + 100) * 1_000_000; + const checkEnd = checkStart + 1_000_000; + await ctx.updateConnectionStatus("test-node", checkStart, checkEnd, true); + expect(ctx.isHostUnhealthy()).toBe(false); }); - it("isHostUnhealthy with invalid connection", async () => { - const currentTimeNano = Date.now(); - await context.setConnectionValid("test-node", false, currentTimeNano, currentTimeNano); - expect(context.isHostUnhealthy).toBe(false); - expect(context.failureCount).toBe(1); + it("isHostUnhealthy with invalid connection - single failure", async () => { + const ctx = new ConnectionContextImpl( + mockClientWrapper, + FAILURE_DETECTION_TIME_MILLIS, + FAILURE_DETECTION_INTERVAL_MILLIS, + FAILURE_DETECTION_COUNT, + new NullTelemetryFactory().createCounter("counter") + ); + const checkStart = getCurrentTimeNano() + (FAILURE_DETECTION_TIME_MILLIS + 100) * 1_000_000; + const checkEnd = checkStart + 1_000_000; + await ctx.updateConnectionStatus("test-node", checkStart, checkEnd, false); + // Single failure doesn't exceed threshold + expect(ctx.isHostUnhealthy()).toBe(false); }); - it("isHostUnhealthy exceeds failure detection count", async () => { - let currentTimeNano = Date.now(); - context.failureCount = 0; - context.resetInvalidHostStartTimeNano(); + it("isHostUnhealthy exceeds failure detection threshold", async () => { + const ctx = new ConnectionContextImpl( + mockClientWrapper, + FAILURE_DETECTION_TIME_MILLIS, + FAILURE_DETECTION_INTERVAL_MILLIS, + FAILURE_DETECTION_COUNT, + new NullTelemetryFactory().createCounter("counter") + ); - for (let i = 0; i < 5; i++) { - const statusCheckStartTime = currentTimeNano; - const statusCheckEndTime = currentTimeNano + VALIDATION_INTERVAL_MILLIS * 1_000_000; + // Use timestamps far enough in the future to exceed the grace period (failureDetectionTimeMillis). + // The context's startMonitorTimeNano is hrtime-based (~now), so we simulate checks starting after the grace period. + const futureStartNano = getCurrentTimeNano() + (FAILURE_DETECTION_TIME_MILLIS + 100) * 1_000_000; - await context.setConnectionValid("test-node", false, statusCheckStartTime, statusCheckEndTime); - if (i >= FAILURE_DETECTION_COUNT - 1) { - expect(context.isHostUnhealthy).toBe(true); - } else { - expect(context.isHostUnhealthy).toBe(false); - } + // maxInvalidHostDuration = failureDetectionIntervalMillis * failureDetectionCount = 100 * 3 = 300ms + // Each check spans FAILURE_DETECTION_INTERVAL_MILLIS (100ms), so after 3 checks the duration exceeds threshold. + for (let i = 0; i < FAILURE_DETECTION_COUNT + 2; i++) { + const checkStart = futureStartNano + i * FAILURE_DETECTION_INTERVAL_MILLIS * 1_000_000; + const checkEnd = checkStart + FAILURE_DETECTION_INTERVAL_MILLIS * 1_000_000; + + await ctx.updateConnectionStatus("test-node", checkStart, checkEnd, false); - currentTimeNano += VALIDATION_INTERVAL_MILLIS * 1_000_000; + if (ctx.isHostUnhealthy()) { + break; + } } - const statusCheckStartTime = currentTimeNano; - const statusCheckEndTime = currentTimeNano + VALIDATION_INTERVAL_MILLIS * 1_000_000; - await context.setConnectionValid("test-node", false, statusCheckStartTime, statusCheckEndTime); - expect(context.isHostUnhealthy).toBe(true); + expect(ctx.isHostUnhealthy()).toBe(true); }); - it.each([[true], [false]])("updateConnectionStatus", async (isValid: boolean) => { - const currentTimeNano = Date.now(); - const statusCheckStartTime = Date.now() - FAILURE_DETECTION_TIME_MILLIS * 1_000_000; + it("updateConnectionStatus skips within grace period", async () => { + // Use a context with a long grace period + const ctx = new ConnectionContextImpl( + mockClientWrapper, + 30000, // 30s grace + FAILURE_DETECTION_INTERVAL_MILLIS, + FAILURE_DETECTION_COUNT, + new NullTelemetryFactory().createCounter("counter") + ); - const contextSpy = spy(context); - await context.updateConnectionStatus("test-node", statusCheckStartTime, currentTimeNano, isValid); + // Call with a time that's within the grace period (elapsed < failureDetectionTime) + const now = getCurrentTimeNano(); + await ctx.updateConnectionStatus("test-node", now, now + 1_000_000, false); - verify(contextSpy.setConnectionValid("test-node", isValid, statusCheckStartTime, currentTimeNano)).once(); + // Should not have detected unhealthy since we're within grace period + expect(ctx.isHostUnhealthy()).toBe(false); }); - it("updateConnectionStatus - inactive context", async () => { - const currentTimeNano = Date.now(); - const statusCheckStartTime = Date.now() - 1000; - context.isActiveContext = false; + it("updateConnectionStatus skips for inactive context", async () => { + context.setInactive(); + expect(context.isActiveContext()).toBe(false); - const contextSpy = spy(context); - await context.updateConnectionStatus("test-node", statusCheckStartTime, currentTimeNano, true); + const now = getCurrentTimeNano(); + await context.updateConnectionStatus("test-node", now, now + 100_000_000, false); - verify(contextSpy.setConnectionValid("test-node", true, statusCheckStartTime, currentTimeNano)).never(); + // Should remain not-unhealthy since the context is inactive + expect(context.isHostUnhealthy()).toBe(false); }); - it("abort client ignores error", async () => { - context = new MonitorConnectionContext( - instance(mockMonitor), - mockClientWrapper, - FAILURE_DETECTION_TIME_MILLIS, - FAILURE_DETECTION_INTERVAL_MILLIS, - FAILURE_DETECTION_COUNT, - instance(mockPluginService), - new NullTelemetryFactory().createCounter("counter") - ); + it("setInactive marks context inactive", () => { + expect(context.isActiveContext()).toBe(true); + context.setInactive(); + expect(context.isActiveContext()).toBe(false); + }); + it("abortConnection does nothing for inactive context", async () => { + context.setInactive(); + // Should not throw await context.abortConnection(); }); }); diff --git a/tests/unit/monitor_impl.test.ts b/tests/unit/monitor_impl.test.ts index 8ba1116cc..2ba18316f 100644 --- a/tests/unit/monitor_impl.test.ts +++ b/tests/unit/monitor_impl.test.ts @@ -1,12 +1,12 @@ /* 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. @@ -14,61 +14,56 @@ limitations under the License. */ -import { MonitorImpl } from "../../common/lib/plugins/efm/monitor"; -import { anything, instance, mock, reset, spy, verify, when } from "ts-mockito"; +import { HostMonitorImpl } from "../../common/lib/plugins/efm/base/host_monitor"; +import { ConnectionContextImpl } from "../../common/lib/plugins/efm/base/connection_context"; +import { anything, instance, mock, reset, when } from "ts-mockito"; import { PluginService, PluginServiceImpl } from "../../common/lib/plugin_service"; import { HostInfo } from "../../common/lib"; -import { AwsClient } from "../../common/lib/aws_client"; -import { MonitorConnectionContext } from "../../common/lib/plugins/efm/monitor_connection_context"; import { sleep } from "../../common/lib/utils/utils"; import { ClientWrapper } from "../../common/lib/client_wrapper"; import { NullTelemetryFactory } from "../../common/lib/utils/telemetry/null_telemetry_factory"; import { MySQLClientWrapper } from "../../common/lib/mysql_client_wrapper"; import { MySQL2DriverDialect } from "../../mysql/lib/dialect/mysql2_driver_dialect"; -class MonitorImplTest extends MonitorImpl { +class MonitorImplTest extends HostMonitorImpl { constructor(pluginService: PluginService, hostInfo: HostInfo, properties: Map, monitorDisposalTimeMillis: number) { super(pluginService, hostInfo, properties, monitorDisposalTimeMillis); } - override startRun() { - // do nothing + override checkConnectionStatus() { + // Exposes the protected checkConnectionStatus method for testing purposes. + return super.checkConnectionStatus(); } } const mockPluginService = mock(PluginServiceImpl); const mockHostInfo = mock(HostInfo); -const mockClient = mock(AwsClient); const mockClientWrapper: ClientWrapper = new MySQLClientWrapper(undefined, mock(HostInfo), new Map(), new MySQL2DriverDialect()); const SHORT_INTERVAL_MILLIS = 30; const properties = new Map(); -let monitor: MonitorImpl; -let monitorSpy: MonitorImpl; +let monitor: MonitorImplTest; describe("monitor impl test", () => { beforeEach(() => { reset(mockPluginService); - when(mockPluginService.getCurrentClient()).thenReturn(instance(mockClient)); when(mockPluginService.forceConnect(anything(), anything())).thenResolve(mockClientWrapper); when(mockPluginService.getTelemetryFactory()).thenReturn(new NullTelemetryFactory()); monitor = new MonitorImplTest(instance(mockPluginService), instance(mockHostInfo), properties, 0); - monitorSpy = spy(monitor); }); - afterEach(() => { - monitor.releaseResources(); + afterEach(async () => { + await monitor.releaseResources(); }); it("is client healthy with no existing client", async () => { - const status = await monitor.checkConnectionStatus(); + const [isValid, elapsedTimeNano] = await monitor.checkConnectionStatus(); - verify(mockPluginService.forceConnect(anything(), anything())).once(); - expect(status.isValid).toBe(true); - expect(status.elapsedTimeNano).toBeGreaterThanOrEqual(0); + expect(isValid).toBe(true); + expect(elapsedTimeNano).toBeGreaterThanOrEqual(0); }); it("is client healthy with existing client", async () => { @@ -77,13 +72,11 @@ describe("monitor impl test", () => { // Start up a monitoring client. await monitor.checkConnectionStatus(); - const status1 = await monitor.checkConnectionStatus(); - expect(status1.isValid).toBe(true); + const [isValid1] = await monitor.checkConnectionStatus(); + expect(isValid1).toBe(true); - const status2 = await monitor.checkConnectionStatus(); - expect(status2.isValid).toBe(true); - - verify(mockPluginService.isClientValid(mockClientWrapper)).twice(); + const [isValid2] = await monitor.checkConnectionStatus(); + expect(isValid2).toBe(true); }); it("is client healthy with error", async () => { @@ -92,31 +85,23 @@ describe("monitor impl test", () => { // Start up a monitoring client. await monitor.checkConnectionStatus(); - const status = await monitor.checkConnectionStatus(); - expect(status.isValid).toBe(false); - expect(status.elapsedTimeNano).toBeGreaterThanOrEqual(0); + const [isValid, elapsedTimeNano] = await monitor.checkConnectionStatus(); + expect(isValid).toBe(false); + expect(elapsedTimeNano).toBeGreaterThanOrEqual(0); }); it("run without context", async () => { - // Should end by itself. + // Should end by itself (monitorDisposalTimeMillis = 0). await monitor.run(); - verify(monitorSpy.checkConnectionStatus()).never(); }); it("run with context", async () => { - const monitorContextInstance = new MonitorConnectionContext( - monitor, - mockClientWrapper, - 30000, - 5000, - 3, - instance(mockPluginService), - new NullTelemetryFactory().createCounter("name") - ); - monitor.startMonitoring(monitorContextInstance); - // Should end by itself. + when(mockPluginService.isClientValid(anything())).thenResolve(true); + + const context = new ConnectionContextImpl(mockClientWrapper, 30000, 5000, 3, new NullTelemetryFactory().createCounter("counter")); + monitor.startMonitoring(context); monitor.run(); await sleep(SHORT_INTERVAL_MILLIS); - monitor.stopMonitoring(monitorContextInstance); + monitor.stopMonitoring(context); }); }); diff --git a/tests/unit/monitor_service_impl.test.ts b/tests/unit/monitor_service_impl.test.ts index 676d8c2bd..4093196c3 100644 --- a/tests/unit/monitor_service_impl.test.ts +++ b/tests/unit/monitor_service_impl.test.ts @@ -1,12 +1,12 @@ /* 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. @@ -14,203 +14,168 @@ limitations under the License. */ -import { anything, capture, instance, mock, reset, verify, when } from "ts-mockito"; -import { MonitorImpl } from "../../common/lib/plugins/efm/monitor"; -import { MonitorServiceImpl } from "../../common/lib/plugins/efm/monitor_service"; -import { PluginService, PluginServiceImpl } from "../../common/lib/plugin_service"; +import { instance, mock, reset, when } from "ts-mockito"; +import { HostMonitorServiceImpl } from "../../common/lib/plugins/efm/base/host_monitor_service"; +import { HostMonitorImpl } from "../../common/lib/plugins/efm/base/host_monitor"; +import { PluginServiceImpl } from "../../common/lib/plugin_service"; import { HostInfo, HostInfoBuilder } from "../../common/lib"; import { SimpleHostAvailabilityStrategy } from "../../common/lib/host_availability/simple_host_availability_strategy"; import { NullTelemetryFactory } from "../../common/lib/utils/telemetry/null_telemetry_factory"; import { MySQLClientWrapper } from "../../common/lib/mysql_client_wrapper"; - -class MonitorImplTest extends MonitorImpl { - constructor(pluginService: PluginService, hostInfo: HostInfo, properties: Map, monitorDisposalTimeMillis: number) { - super(pluginService, hostInfo, properties, monitorDisposalTimeMillis); - } - - override startRun() { - // do nothing - } -} - -const mockPluginService = mock(PluginServiceImpl); -const mockMonitorA = mock(MonitorImpl); -const mockMonitorB = mock(MonitorImpl); -const mockHostInfo = mock(HostInfo); -const mockClientWrapper = mock(MySQLClientWrapper); +import { MySQL2DriverDialect } from "../../mysql/lib/dialect/mysql2_driver_dialect"; +import { MonitorServiceImpl } from "../../common/lib/utils/monitoring/monitor_service"; +import { FullServicesContainer } from "../../common/lib/utils/full_services_container"; +import { BatchingEventPublisher } from "../../common/lib/utils/events/batching_event_publisher"; +import { WrapperProperties } from "../../common/lib/wrapper_property"; const FAILURE_DETECTION_TIME_MILLIS = 10; const FAILURE_DETECTION_INTERVAL_MILLIS = 100; const FAILURE_DETECTION_COUNT = 3; -const NODE_KEYS = new Set(["any"]); -const properties = new Map(); -let monitorService: MonitorServiceImpl; +const mockPluginService = mock(PluginServiceImpl); + +const properties: Map = new Map(); +properties.set(WrapperProperties.MONITOR_DISPOSAL_TIME_MS.name, 600000); + +let monitorService: HostMonitorServiceImpl; +let coreMonitorService: MonitorServiceImpl; +let servicesContainer: FullServicesContainer; +let eventPublisher: BatchingEventPublisher; describe("monitor service impl test", () => { beforeEach(() => { reset(mockPluginService); - reset(mockMonitorA); - when(mockPluginService.getTelemetryFactory()).thenReturn(new NullTelemetryFactory()); - monitorService = new MonitorServiceImpl(instance(mockPluginService)); - monitorService.monitorSupplier = () => new MonitorImplTest(instance(mockPluginService), instance(mockHostInfo), properties, 0); + const telemetryFactory = new NullTelemetryFactory(); + when(mockPluginService.getTelemetryFactory()).thenReturn(telemetryFactory); + when(mockPluginService.isClientValid(undefined)).thenResolve(false); + + eventPublisher = new BatchingEventPublisher(60_000); + coreMonitorService = new MonitorServiceImpl(eventPublisher); + + servicesContainer = { + pluginService: instance(mockPluginService), + monitorService: coreMonitorService, + telemetryFactory: telemetryFactory + } as unknown as FullServicesContainer; + + monitorService = new HostMonitorServiceImpl(servicesContainer); }); afterEach(async () => { await monitorService.releaseResources(); + await coreMonitorService.releaseResources(); + eventPublisher.releaseResources(); }); - it("start monitoring", async () => { - monitorService.monitorSupplier = () => instance(mockMonitorA); + it("start monitoring creates context", async () => { + const hostInfo = new HostInfoBuilder({ host: "test-host", hostAvailabilityStrategy: new SimpleHostAvailabilityStrategy() }).build(); + const mockClientWrapper = new MySQLClientWrapper(undefined, mock(HostInfo), new Map(), new MySQL2DriverDialect()); - await monitorService.startMonitoring( + const context = await monitorService.startMonitoring( mockClientWrapper, - NODE_KEYS, - instance(mockHostInfo), - new Map(), + hostInfo, + properties, FAILURE_DETECTION_TIME_MILLIS, FAILURE_DETECTION_INTERVAL_MILLIS, FAILURE_DETECTION_COUNT ); - const arg = capture(mockMonitorA.startMonitoring).last(); - expect(arg).toBeDefined(); - expect(arg).not.toBeNull(); - }); - it("start monitoring called multiple times", async () => { - monitorService.monitorSupplier = () => instance(mockMonitorA); - - const runs = 5; - - for (let i = 0; i < runs; i++) { - await monitorService.startMonitoring( - mockClientWrapper, - NODE_KEYS, - instance(mockHostInfo), - new Map(), - FAILURE_DETECTION_TIME_MILLIS, - FAILURE_DETECTION_INTERVAL_MILLIS, - FAILURE_DETECTION_COUNT - ); - const arg = capture(mockMonitorA.startMonitoring).last(); - expect(arg).toBeDefined(); - expect(arg).not.toBeNull(); - } + expect(context).toBeDefined(); + expect(context).not.toBeNull(); + expect(context.isActiveContext()).toBe(true); + expect(context.isHostUnhealthy()).toBe(false); }); - it("start and stop monitoring", async () => { - monitorService.monitorSupplier = () => instance(mockMonitorA); + it("stop monitoring sets context inactive", async () => { + const hostInfo = new HostInfoBuilder({ host: "test-host", hostAvailabilityStrategy: new SimpleHostAvailabilityStrategy() }).build(); + const mockClientWrapper = new MySQLClientWrapper(undefined, mock(HostInfo), new Map(), new MySQL2DriverDialect()); const context = await monitorService.startMonitoring( mockClientWrapper, - NODE_KEYS, - instance(mockHostInfo), - new Map(), + hostInfo, + properties, FAILURE_DETECTION_TIME_MILLIS, FAILURE_DETECTION_INTERVAL_MILLIS, FAILURE_DETECTION_COUNT ); - await monitorService.stopMonitoring(context); - const arg = capture(mockMonitorA.startMonitoring).last(); - expect(arg[0]).toBe(context); - verify(mockMonitorA.stopMonitoring(anything())).once(); + monitorService.stopMonitoring(context); + expect(context.isActiveContext()).toBe(false); }); - it("stop monitoring called twice", async () => { - monitorService.monitorSupplier = () => instance(mockMonitorA); + it("start monitoring reuses monitor for same host", async () => { + const hostInfo = new HostInfoBuilder({ host: "test-host", hostAvailabilityStrategy: new SimpleHostAvailabilityStrategy() }).build(); + const mockClientWrapper = new MySQLClientWrapper(undefined, mock(HostInfo), new Map(), new MySQL2DriverDialect()); - const context = await monitorService.startMonitoring( + await monitorService.startMonitoring( mockClientWrapper, - NODE_KEYS, - instance(mockHostInfo), - new Map(), + hostInfo, + properties, FAILURE_DETECTION_TIME_MILLIS, FAILURE_DETECTION_INTERVAL_MILLIS, FAILURE_DETECTION_COUNT ); - await monitorService.stopMonitoring(context); - const arg = capture(mockMonitorA.startMonitoring).last(); - expect(arg[0]).toBe(context); + const monitor1 = coreMonitorService.get(HostMonitorImpl, hostInfo.host); - await monitorService.stopMonitoring(context); - verify(mockMonitorA.stopMonitoring(anything())).twice(); - }); + await monitorService.startMonitoring( + mockClientWrapper, + hostInfo, + properties, + FAILURE_DETECTION_TIME_MILLIS, + FAILURE_DETECTION_INTERVAL_MILLIS, + FAILURE_DETECTION_COUNT + ); - it("stop monitoring for all connections with invalid node keys", async () => { - monitorService.stopMonitoringForAllConnections(new Set()); - monitorService.stopMonitoringForAllConnections(new Set(["foo"])); + const monitor2 = coreMonitorService.get(HostMonitorImpl, hostInfo.host); + expect(monitor2).toBe(monitor1); }); - it("stop monitoring for all connections", async () => { - const keysA = new Set(["monitorA"]); - const keysB = new Set(["monitorB"]); + it("start monitoring creates separate monitors for different hosts", async () => { + const hostInfoA = new HostInfoBuilder({ host: "host-a", hostAvailabilityStrategy: new SimpleHostAvailabilityStrategy() }).build(); + const hostInfoB = new HostInfoBuilder({ host: "host-b", hostAvailabilityStrategy: new SimpleHostAvailabilityStrategy() }).build(); + const mockClientWrapper = new MySQLClientWrapper(undefined, mock(HostInfo), new Map(), new MySQL2DriverDialect()); - monitorService.monitorSupplier = () => instance(mockMonitorA); - await monitorService.getMonitor( - keysA, - new HostInfoBuilder({ host: "test", hostAvailabilityStrategy: new SimpleHostAvailabilityStrategy() }).build(), - properties + await monitorService.startMonitoring( + mockClientWrapper, + hostInfoA, + properties, + FAILURE_DETECTION_TIME_MILLIS, + FAILURE_DETECTION_INTERVAL_MILLIS, + FAILURE_DETECTION_COUNT ); - monitorService.monitorSupplier = () => instance(mockMonitorB); - await monitorService.getMonitor( - keysB, - new HostInfoBuilder({ host: "test", hostAvailabilityStrategy: new SimpleHostAvailabilityStrategy() }).build(), - properties + await monitorService.startMonitoring( + mockClientWrapper, + hostInfoB, + properties, + FAILURE_DETECTION_TIME_MILLIS, + FAILURE_DETECTION_INTERVAL_MILLIS, + FAILURE_DETECTION_COUNT ); - monitorService.stopMonitoringForAllConnections(keysA); - verify(mockMonitorA.clearContexts()).once(); - - monitorService.stopMonitoringForAllConnections(keysB); - verify(mockMonitorB.clearContexts()).once(); + const monitorA = coreMonitorService.get(HostMonitorImpl, hostInfoA.host); + const monitorB = coreMonitorService.get(HostMonitorImpl, hostInfoB.host); + expect(monitorA).not.toBe(monitorB); }); - it("getMonitor called with multiple hosts in keys", async () => { - const keysA = new Set(["host1.domain", "host2.domain"]); - const keysB = new Set(["host2.domain"]); - - const monitorOne = await monitorService.getMonitor(keysA, instance(mockHostInfo), properties); - expect(monitorOne).not.toBeNull(); - - const monitorOneDupe = await monitorService.getMonitor(keysB, instance(mockHostInfo), properties); - expect(monitorOneDupe).toBe(monitorOne); - }); - - it("getMonitor called with different host keys", async () => { - const keys = new Set(["hostNEW.domain"]); - - const monitorOne = await monitorService.getMonitor(keys, instance(mockHostInfo), properties); - expect(monitorOne).not.toBeNull(); + it("release resources clears monitors", async () => { + const hostInfo = new HostInfoBuilder({ host: "test-host", hostAvailabilityStrategy: new SimpleHostAvailabilityStrategy() }).build(); + const mockClientWrapper = new MySQLClientWrapper(undefined, mock(HostInfo), new Map(), new MySQL2DriverDialect()); - const monitorOneDupe = await monitorService.getMonitor(keys, instance(mockHostInfo), properties); - expect(monitorOneDupe).toBe(monitorOne); - - const monitorTwo = await monitorService.getMonitor(NODE_KEYS, instance(mockHostInfo), properties); - expect(monitorTwo).not.toBeNull(); - expect(monitorTwo).not.toBe(monitorOne); - }); - - it("getMonitor called with same keys in different sets", async () => { - const keysA = new Set(["hostA"]); - const keysB = new Set(["hostA", "hostB"]); - const keysC = new Set(["hostB"]); - - const monitorOne = await monitorService.getMonitor(keysA, instance(mockHostInfo), properties); - expect(monitorOne).not.toBeNull(); - - const monitorOneDupe = await monitorService.getMonitor(keysB, instance(mockHostInfo), properties); - expect(monitorOneDupe).toBe(monitorOne); + await monitorService.startMonitoring( + mockClientWrapper, + hostInfo, + properties, + FAILURE_DETECTION_TIME_MILLIS, + FAILURE_DETECTION_INTERVAL_MILLIS, + FAILURE_DETECTION_COUNT + ); - const monitorOneDupeTwo = await monitorService.getMonitor(keysC, instance(mockHostInfo), properties); - expect(monitorOneDupeTwo).toBe(monitorOne); - }); + await monitorService.releaseResources(); - it("startMonitoring with no host keys", async () => { - const keys: Set = new Set(); - expect(await monitorService.getMonitor(keys, instance(mockHostInfo), properties)).toBeNull(); + const monitor = coreMonitorService.get(HostMonitorImpl, hostInfo.host); + expect(monitor).toBeNull(); }); });