RockteMQ-AI commented on code in PR #1368: URL: https://github.com/apache/rocketmq-clients/pull/1368#discussion_r3977203393
########## nodejs/src/metrics/ClientMeterManager.ts: ########## @@ -0,0 +1,145 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 { Attributes, Meter } from '@opentelemetry/api'; +import { + MeterProvider as SdkMeterProvider, + PeriodicExportingMetricReader, + View, +} from '@opentelemetry/sdk-metrics'; +import { Resource } from '@opentelemetry/resources'; +import { OTLPMetricExporter } from '@opentelemetry/exporter-metrics-otlp-grpc'; +import { ChannelCredentials, Metadata } from '@grpc/grpc-js'; +import { ILogger } from '../client/Logger'; +import { ClientMeter } from './ClientMeter'; +import { GaugeObserver } from './GaugeObserver'; +import { EmptyGaugeObserver } from './EmptyGaugeObserver'; +import { HistogramEnum, buildHistogramView } from './HistogramEnum'; +import { Metric } from './Metric'; + +const METRIC_EXPORTER_RPC_TIMEOUT = 5000; +const METRIC_READER_INTERVAL = 60000; +const METRIC_INSTRUMENTATION_NAME = 'org.apache.rocketmq.message'; + +/** + * Owns the OpenTelemetry metric pipeline for a single client. On every settings + * command it lazily (re)builds an SdkMeterProvider that exports the four + * RocketMQ histograms plus the consumer gauges to the broker-supplied metric + * endpoints over OTLP/gRPC, mirroring + * org.apache.rocketmq.client.java.metrics.ClientMeterManager. + */ +export class ClientMeterManager { + private readonly clientId: string; + private readonly metadataProvider: () => Metadata; + private readonly logger: ILogger; + private clientMeter: ClientMeter; + private gaugeObserver: GaugeObserver = EmptyGaugeObserver.EMPTY; + + constructor(clientId: string, metadataProvider: () => Metadata, logger: ILogger) { + this.clientId = clientId; + this.metadataProvider = metadataProvider; + this.logger = logger; + this.clientMeter = ClientMeter.disabledInstance(); + } + + setGaugeObserver(gaugeObserver: GaugeObserver) { + this.gaugeObserver = gaugeObserver; + } + + record(histogramEnum: HistogramEnum, attributes: Attributes, value: number) { + this.clientMeter.record(histogramEnum, attributes, value); + } + + isEnabled(): boolean { + return this.clientMeter.enabled; + } + + async shutdown() { + await this.clientMeter.shutdown(); + } + + async reset(metric: Metric): Promise<void> { + try { + if (this.clientMeter.satisfy(metric)) { + this.logger.info('Metric settings is satisfied by the current message meter, metric=%j, clientId=%s', + metric, this.clientId); + return; + } + if (!metric.on || !metric.endpoints) { + this.logger.info('Metric is off, clientId=%s', this.clientId); + await this.clientMeter.shutdown(); + this.clientMeter = ClientMeter.disabledInstance(); + return; + } + const endpoints = metric.endpoints; Review Comment: OTLP export uses `ChannelCredentials.createInsecure()` for the gRPC channel. This is fine for internal metrics collection but should be documented that metrics data is sent unencrypted. If metrics contain sensitive information, consider adding a configuration option for TLS. ########## nodejs/src/hook/InflightRequestCountInterceptor.ts: ########## @@ -0,0 +1,44 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You 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 { MessageHookPoints } from './MessageHookPoints'; +import { MessageInterceptorContext } from './MessageInterceptorContext'; +import { GeneralMessage, MessageInterceptor } from './MessageInterceptor'; + +/** + * Interceptor which counts the in-flight receive requests, mirroring + * org.apache.rocketmq.client.java.hook.InflightRequestCountInterceptor. + */ +export class InflightRequestCountInterceptor implements MessageInterceptor { + #inflightReceiveRequestCount = 0; + + doBefore(context: MessageInterceptorContext, _messages: GeneralMessage[]) { + if (context.getMessageHookPoints() === MessageHookPoints.RECEIVE) { Review Comment: The counter only decrements when `doAfter` is called with `RECEIVE` hook point. This is correct behavior but the asymmetry between `doBefore` (increments on RECEIVE) and `doAfter` (decrements on RECEIVE) could be confusing. Consider adding a comment clarifying that this interceptor only tracks RECEIVE operations, or add an assertion that both methods are called with the same hook point. ########## nodejs/src/client/TelemetrySession.ts: ########## @@ -144,21 +164,35 @@ export class TelemetrySession { } } + /** + * Schedule a telemetry stream renewal. The timer is tracked so that it can + * be cancelled on release(), and a pending timer doubles as a guard to + * prevent error/end events from scheduling duplicate reconnects. + */ + #scheduleRenewStream() { + if (this.#released || this.#reconnectTimer) { + return; + } + this.#reconnectTimer = setTimeout(() => { + this.#reconnectTimer = undefined; + if (this.#released) { Review Comment: Good fix for the timer leak — clearing `#sessionRefreshTimer` before setting a new one prevents accumulation. This was a subtle bug that could cause memory leaks in long-running processes. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
