This is an automated email from the ASF dual-hosted git repository.
RongtongJin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/rocketmq-clients.git
The following commit(s) were added to refs/heads/master by this push:
new 8fd7226c Feat/nodejs metrics export (#1387)
8fd7226c is described below
commit 8fd7226cd3fd64b30c3afa2a7386e8058de7f41b
Author: zhaohai <[email protected]>
AuthorDate: Fri Oct 9 10:45:00 2026 +0800
Feat/nodejs metrics export (#1387)
* feat(nodejs): support NOTIFY_UNSUBSCRIBE_LITE_COMMAND and message
interceptor hooks
- BaseClient: add onNotifyUnsubscribeLiteCommand hook (warn by default,
overridden by lite push consumers)
- TelemetrySession: dispatch NOTIFY_UNSUBSCRIBE_LITE_COMMAND (was falling
through to the unknown-command branch)
- LitePushConsumerImpl: override the command handler with the
endpoints-aware
signature and forward to the lite subscription manager
- BaseClient: drop the commented-out recover-orphaned-transaction reply
(warn only, no reply)
- New hook package: MessageHookPoints, MessageHookPointsStatus,
Attribute/AttributeKey, MessageInterceptor(+Context/Impl),
CompositedMessageInterceptor and InflightRequestCountInterceptor; clients
expose addMessageInterceptor and doBefore/doAfter, configurable via
BaseClientOptions#messageInterceptor
- Wire hook points: SEND (producer), RECEIVE/ACK/CHANGE_INVISIBLE_DURATION
(consumer), CONSUME (consume task); push consumers register the built-in
inflight request count interceptor
- Tests: cover telemetry dispatch and interceptor chain ordering, error
isolation, attribute propagation and inflight counting
* feat(nodejs): wire COMMIT_TRANSACTION / ROLLBACK_TRANSACTION /
FORWARD_TO_DLQ hook points
Complete the remaining 3 of the 8 MessageHookPoints:
- K-2 Producer#endTransaction fires COMMIT_TRANSACTION /
ROLLBACK_TRANSACTION
(selected by resolution === COMMIT), hook status OK/ERROR taken from the
response Code or the thrown error.
- K-3 Consumer#forwardMessageToDeadLetterQueueViaRpc fires FORWARD_TO_DLQ
around every DLQ RPC attempt. ProcessQueue now forwards the MessageView
so interceptors observe the real message.
- Add 4 regression tests (transaction hooks, DLQ hook, error paths);
offline suite 119/119, integration 16/16.
* feat(nodejs): add metrics export pipeline over OpenTelemetry / OTLP gRPC
Add src/metrics/*:
- ClientMeterManager rebuilds an OpenTelemetry SdkMeterProvider on every
settings command, exporting the four RocketMQ histograms (send cost
time, delivery latency, await time, process time) with explicit bucket
boundaries, plus consumer cached-message gauges. Every export request is
signed through the client's MQv2-HMAC-SHA1 metadata provider.
- MessageMeterInterceptor records the histograms at the SEND / RECEIVE /
CONSUME hook points via the interceptor chain.
- PushConsumerGaugeObserver aggregates cached messages/bytes per topic.
- BaseClient wires the manager (created in the constructor, reset from
onSettingsCommand, shut down on client shutdown).
---
nodejs/package-lock.json | 211 ++++++++++++
nodejs/package.json | 6 +-
nodejs/src/client/BaseClient.ts | 97 +++++-
nodejs/src/consumer/ConsumeService.ts | 11 +-
nodejs/src/consumer/ConsumeTask.ts | 46 ++-
nodejs/src/consumer/Consumer.ts | 82 ++++-
nodejs/src/consumer/FifoConsumeService.ts | 6 +-
nodejs/src/consumer/LiteFifoConsumeService.ts | 6 +-
nodejs/src/consumer/LitePushConsumerImpl.ts | 11 +-
nodejs/src/consumer/LiteStandardConsumeService.ts | 6 +-
nodejs/src/consumer/ProcessQueue.ts | 9 +
nodejs/src/consumer/PushConsumer.ts | 36 ++-
nodejs/src/consumer/PushConsumerGaugeObserver.ts | 60 ++++
nodejs/src/consumer/StandardConsumeService.ts | 6 +-
nodejs/src/{index.ts => hook/Attribute.ts} | 26 +-
nodejs/src/{index.ts => hook/AttributeKey.ts} | 32 +-
nodejs/src/hook/CompositedMessageInterceptor.ts | 98 ++++++
nodejs/src/hook/InflightRequestCountInterceptor.ts | 44 +++
nodejs/src/hook/MessageHookPoints.ts | 55 ++++
.../{index.ts => hook/MessageHookPointsStatus.ts} | 15 +-
.../ConsumeTask.ts => hook/MessageInterceptor.ts} | 34 +-
.../MessageInterceptorContext.ts} | 34 +-
nodejs/src/hook/MessageInterceptorContextImpl.ts | 68 ++++
nodejs/src/hook/MessageMeterInterceptor.ts | 162 ++++++++++
nodejs/src/{ => hook}/index.ts | 17 +-
nodejs/src/index.ts | 1 +
nodejs/src/metrics/ClientMeter.ts | 81 +++++
nodejs/src/metrics/ClientMeterManager.ts | 145 +++++++++
.../{index.ts => metrics/EmptyGaugeObserver.ts} | 26 +-
nodejs/src/{index.ts => metrics/GaugeEnum.ts} | 23 +-
nodejs/src/{index.ts => metrics/GaugeObserver.ts} | 22 +-
nodejs/src/metrics/HistogramEnum.ts | 69 ++++
.../src/{index.ts => metrics/InvocationStatus.ts} | 18 +-
nodejs/src/metrics/Metric.ts | 42 +++
nodejs/src/{index.ts => metrics/MetricLabels.ts} | 21 +-
nodejs/src/{ => metrics}/index.ts | 16 +-
nodejs/src/producer/Producer.ts | 43 ++-
nodejs/test/hook.test.ts | 276 ++++++++++++++++
nodejs/test/metrics.test.ts | 352 +++++++++++++++++++++
39 files changed, 2159 insertions(+), 154 deletions(-)
diff --git a/nodejs/package-lock.json b/nodejs/package-lock.json
index 91bd72f5..547ba624 100644
--- a/nodejs/package-lock.json
+++ b/nodejs/package-lock.json
@@ -11,6 +11,10 @@
"dependencies": {
"@grpc/grpc-js": "^1.9.1",
"@node-rs/crc32": "^1.7.2",
+ "@opentelemetry/api": "^1.9.1",
+ "@opentelemetry/exporter-metrics-otlp-grpc": "^0.57.2",
+ "@opentelemetry/resources": "^1.30.1",
+ "@opentelemetry/sdk-metrics": "^1.30.1",
"address": "^1.2.2",
"egg-logger": "^3.4.0",
"google-protobuf": "^3.21.2",
@@ -1271,6 +1275,213 @@
"node": ">= 8"
}
},
+ "node_modules/@opentelemetry/api": {
+ "version": "1.9.1",
+ "resolved":
"https://registry.npmjs.org/@opentelemetry/api/-/api-1.9.1.tgz";,
+ "integrity":
"sha512-gLyJlPHPZYdAk1JENA9LeHejZe1Ti77/pTeFm/nMXmQH/HFZlcS/O2XJB+L8fkbrNSqhdtlvjBVjxwUYanNH5Q==",
+ "license": "Apache-2.0",
+ "engines": {
+ "node": ">=8.0.0"
+ }
+ },
+ "node_modules/@opentelemetry/api-logs": {
+ "version": "0.57.2",
+ "resolved":
"https://registry.npmjs.org/@opentelemetry/api-logs/-/api-logs-0.57.2.tgz";,
+ "integrity":
"sha512-uIX52NnTM0iBh84MShlpouI7UKqkZ7MrUszTmaypHBu4r7NofznSnQRfJ+uUeDtQDj6w8eFGg5KBLDAwAPz1+A==",
+ "license": "Apache-2.0",
+ "dependencies": {
+ "@opentelemetry/api": "^1.3.0"
+ },
+ "engines": {
+ "node": ">=14"
+ }
+ },
+ "node_modules/@opentelemetry/core": {
+ "version": "1.30.1",
+ "resolved":
"https://registry.npmjs.org/@opentelemetry/core/-/core-1.30.1.tgz";,
+ "integrity":
"sha512-OOCM2C/QIURhJMuKaekP3TRBxBKxG/TWWA0TL2J6nXUtDnuCtccy49LUJF8xPFXMX+0LMcxFpCo8M9cGY1W6rQ==",
+ "license": "Apache-2.0",
+ "dependencies": {
+ "@opentelemetry/semantic-conventions": "1.28.0"
+ },
+ "engines": {
+ "node": ">=14"
+ },
+ "peerDependencies": {
+ "@opentelemetry/api": ">=1.0.0 <1.10.0"
+ }
+ },
+ "node_modules/@opentelemetry/exporter-metrics-otlp-grpc": {
+ "version": "0.57.2",
+ "resolved":
"https://registry.npmjs.org/@opentelemetry/exporter-metrics-otlp-grpc/-/exporter-metrics-otlp-grpc-0.57.2.tgz";,
+ "integrity":
"sha512-r70B8yKR41F0EC443b5CGB4rUaOMm99I5N75QQt6sHKxYDzSEc6gm48Diz1CI1biwa5tDPznpylTrywO/pT7qw==",
+ "license": "Apache-2.0",
+ "dependencies": {
+ "@grpc/grpc-js": "^1.7.1",
+ "@opentelemetry/core": "1.30.1",
+ "@opentelemetry/exporter-metrics-otlp-http": "0.57.2",
+ "@opentelemetry/otlp-exporter-base": "0.57.2",
+ "@opentelemetry/otlp-grpc-exporter-base": "0.57.2",
+ "@opentelemetry/otlp-transformer": "0.57.2",
+ "@opentelemetry/resources": "1.30.1",
+ "@opentelemetry/sdk-metrics": "1.30.1"
+ },
+ "engines": {
+ "node": ">=14"
+ },
+ "peerDependencies": {
+ "@opentelemetry/api": "^1.3.0"
+ }
+ },
+ "node_modules/@opentelemetry/exporter-metrics-otlp-http": {
+ "version": "0.57.2",
+ "resolved":
"https://registry.npmjs.org/@opentelemetry/exporter-metrics-otlp-http/-/exporter-metrics-otlp-http-0.57.2.tgz";,
+ "integrity":
"sha512-ttb9+4iKw04IMubjm3t0EZsYRNWr3kg44uUuzfo9CaccYlOh8cDooe4QObDUkvx9d5qQUrbEckhrWKfJnKhemA==",
+ "license": "Apache-2.0",
+ "dependencies": {
+ "@opentelemetry/core": "1.30.1",
+ "@opentelemetry/otlp-exporter-base": "0.57.2",
+ "@opentelemetry/otlp-transformer": "0.57.2",
+ "@opentelemetry/resources": "1.30.1",
+ "@opentelemetry/sdk-metrics": "1.30.1"
+ },
+ "engines": {
+ "node": ">=14"
+ },
+ "peerDependencies": {
+ "@opentelemetry/api": "^1.3.0"
+ }
+ },
+ "node_modules/@opentelemetry/otlp-exporter-base": {
+ "version": "0.57.2",
+ "resolved":
"https://registry.npmjs.org/@opentelemetry/otlp-exporter-base/-/otlp-exporter-base-0.57.2.tgz";,
+ "integrity":
"sha512-XdxEzL23Urhidyebg5E6jZoaiW5ygP/mRjxLHixogbqwDy2Faduzb5N0o/Oi+XTIJu+iyxXdVORjXax+Qgfxag==",
+ "license": "Apache-2.0",
+ "dependencies": {
+ "@opentelemetry/core": "1.30.1",
+ "@opentelemetry/otlp-transformer": "0.57.2"
+ },
+ "engines": {
+ "node": ">=14"
+ },
+ "peerDependencies": {
+ "@opentelemetry/api": "^1.3.0"
+ }
+ },
+ "node_modules/@opentelemetry/otlp-grpc-exporter-base": {
+ "version": "0.57.2",
+ "resolved":
"https://registry.npmjs.org/@opentelemetry/otlp-grpc-exporter-base/-/otlp-grpc-exporter-base-0.57.2.tgz";,
+ "integrity":
"sha512-USn173KTWy0saqqRB5yU9xUZ2xdgb1Rdu5IosJnm9aV4hMTuFFRTUsQxbgc24QxpCHeoKzzCSnS/JzdV0oM2iQ==",
+ "license": "Apache-2.0",
+ "dependencies": {
+ "@grpc/grpc-js": "^1.7.1",
+ "@opentelemetry/core": "1.30.1",
+ "@opentelemetry/otlp-exporter-base": "0.57.2",
+ "@opentelemetry/otlp-transformer": "0.57.2"
+ },
+ "engines": {
+ "node": ">=14"
+ },
+ "peerDependencies": {
+ "@opentelemetry/api": "^1.3.0"
+ }
+ },
+ "node_modules/@opentelemetry/otlp-transformer": {
+ "version": "0.57.2",
+ "resolved":
"https://registry.npmjs.org/@opentelemetry/otlp-transformer/-/otlp-transformer-0.57.2.tgz";,
+ "integrity":
"sha512-48IIRj49gbQVK52jYsw70+Jv+JbahT8BqT2Th7C4H7RCM9d0gZ5sgNPoMpWldmfjvIsSgiGJtjfk9MeZvjhoig==",
+ "license": "Apache-2.0",
+ "dependencies": {
+ "@opentelemetry/api-logs": "0.57.2",
+ "@opentelemetry/core": "1.30.1",
+ "@opentelemetry/resources": "1.30.1",
+ "@opentelemetry/sdk-logs": "0.57.2",
+ "@opentelemetry/sdk-metrics": "1.30.1",
+ "@opentelemetry/sdk-trace-base": "1.30.1",
+ "protobufjs": "^7.3.0"
+ },
+ "engines": {
+ "node": ">=14"
+ },
+ "peerDependencies": {
+ "@opentelemetry/api": "^1.3.0"
+ }
+ },
+ "node_modules/@opentelemetry/resources": {
+ "version": "1.30.1",
+ "resolved":
"https://registry.npmjs.org/@opentelemetry/resources/-/resources-1.30.1.tgz";,
+ "integrity":
"sha512-5UxZqiAgLYGFjS4s9qm5mBVo433u+dSPUFWVWXmLAD4wB65oMCoXaJP1KJa9DIYYMeHu3z4BZcStG3LC593cWA==",
+ "license": "Apache-2.0",
+ "dependencies": {
+ "@opentelemetry/core": "1.30.1",
+ "@opentelemetry/semantic-conventions": "1.28.0"
+ },
+ "engines": {
+ "node": ">=14"
+ },
+ "peerDependencies": {
+ "@opentelemetry/api": ">=1.0.0 <1.10.0"
+ }
+ },
+ "node_modules/@opentelemetry/sdk-logs": {
+ "version": "0.57.2",
+ "resolved":
"https://registry.npmjs.org/@opentelemetry/sdk-logs/-/sdk-logs-0.57.2.tgz";,
+ "integrity":
"sha512-TXFHJ5c+BKggWbdEQ/inpgIzEmS2BGQowLE9UhsMd7YYlUfBQJ4uax0VF/B5NYigdM/75OoJGhAV3upEhK+3gg==",
+ "license": "Apache-2.0",
+ "dependencies": {
+ "@opentelemetry/api-logs": "0.57.2",
+ "@opentelemetry/core": "1.30.1",
+ "@opentelemetry/resources": "1.30.1"
+ },
+ "engines": {
+ "node": ">=14"
+ },
+ "peerDependencies": {
+ "@opentelemetry/api": ">=1.4.0 <1.10.0"
+ }
+ },
+ "node_modules/@opentelemetry/sdk-metrics": {
+ "version": "1.30.1",
+ "resolved":
"https://registry.npmjs.org/@opentelemetry/sdk-metrics/-/sdk-metrics-1.30.1.tgz";,
+ "integrity":
"sha512-q9zcZ0Okl8jRgmy7eNW3Ku1XSgg3sDLa5evHZpCwjspw7E8Is4K/haRPDJrBcX3YSn/Y7gUvFnByNYEKQNbNog==",
+ "license": "Apache-2.0",
+ "dependencies": {
+ "@opentelemetry/core": "1.30.1",
+ "@opentelemetry/resources": "1.30.1"
+ },
+ "engines": {
+ "node": ">=14"
+ },
+ "peerDependencies": {
+ "@opentelemetry/api": ">=1.3.0 <1.10.0"
+ }
+ },
+ "node_modules/@opentelemetry/sdk-trace-base": {
+ "version": "1.30.1",
+ "resolved":
"https://registry.npmjs.org/@opentelemetry/sdk-trace-base/-/sdk-trace-base-1.30.1.tgz";,
+ "integrity":
"sha512-jVPgBbH1gCy2Lb7X0AVQ8XAfgg0pJ4nvl8/IiQA6nxOsPvS+0zMJaFSs2ltXe0J6C8dqjcnpyqINDJmU30+uOg==",
+ "license": "Apache-2.0",
+ "dependencies": {
+ "@opentelemetry/core": "1.30.1",
+ "@opentelemetry/resources": "1.30.1",
+ "@opentelemetry/semantic-conventions": "1.28.0"
+ },
+ "engines": {
+ "node": ">=14"
+ },
+ "peerDependencies": {
+ "@opentelemetry/api": ">=1.0.0 <1.10.0"
+ }
+ },
+ "node_modules/@opentelemetry/semantic-conventions": {
+ "version": "1.28.0",
+ "resolved":
"https://registry.npmjs.org/@opentelemetry/semantic-conventions/-/semantic-conventions-1.28.0.tgz";,
+ "integrity":
"sha512-lp4qAiMTD4sNWW4DbKLBkfiMZ4jbAboJIGOQr5DvciMRI494OapieI9qiODpOt0XBr1LjIDy1xAGAnVs5supTA==",
+ "license": "Apache-2.0",
+ "engines": {
+ "node": ">=14"
+ }
+ },
"node_modules/@pkgjs/parseargs": {
"version": "0.11.0",
"resolved":
"https://registry.npmjs.org/@pkgjs/parseargs/-/parseargs-0.11.0.tgz";,
diff --git a/nodejs/package.json b/nodejs/package.json
index 9e9337f9..96ec1a69 100644
--- a/nodejs/package.json
+++ b/nodejs/package.json
@@ -43,13 +43,17 @@
"egg-bin": "^6.4.2",
"eslint": "^8.48.0",
"eslint-config-egg": "^12.2.1",
- "grpc-tools": "^1.12.4",
"grpc_tools_node_protoc_ts": "^5.3.3",
+ "grpc-tools": "^1.12.4",
"typescript": "^5.2.2"
},
"dependencies": {
"@grpc/grpc-js": "^1.9.1",
"@node-rs/crc32": "^1.7.2",
+ "@opentelemetry/api": "^1.9.1",
+ "@opentelemetry/exporter-metrics-otlp-grpc": "^0.57.2",
+ "@opentelemetry/resources": "^1.30.1",
+ "@opentelemetry/sdk-metrics": "^1.30.1",
"address": "^1.2.2",
"egg-logger": "^3.4.0",
"google-protobuf": "^3.21.2",
diff --git a/nodejs/src/client/BaseClient.ts b/nodejs/src/client/BaseClient.ts
index d2b0e852..d8cdce85 100644
--- a/nodejs/src/client/BaseClient.ts
+++ b/nodejs/src/client/BaseClient.ts
@@ -40,6 +40,13 @@ import {
import { createResource, getRequestDateTime, sign } from '../util';
import { TopicRouteData, Endpoints } from '../route';
import { ClientException, NotFoundException, StatusChecker } from
'../exception';
+import {
+ CompositedMessageInterceptor,
+ GeneralMessage,
+ MessageInterceptor,
+ MessageInterceptorContext,
+ MessageMeterInterceptor,
+} from '../hook';
import { Settings } from './Settings';
import { UserAgent } from './UserAgent';
import { ILogger, getDefaultLogger } from './Logger';
@@ -47,6 +54,8 @@ import { SessionCredentials } from './SessionCredentials';
import { RpcClientManager } from './RpcClientManager';
import { TelemetrySession } from './TelemetrySession';
import { ClientId } from './ClientId';
+import { ClientMeterManager } from '../metrics';
+import { Metric } from '../metrics/Metric';
const debug = debuglog('rocketmq-client-nodejs:client:BaseClient');
@@ -70,6 +79,11 @@ export interface BaseClientOptions {
requestTimeout?: number;
logger?: ILogger;
topics?: string[];
+ /**
+ * Message interceptor(s) invoked around send/receive/consume/ack hook
points,
+ * mirroring the Java client's MessageInterceptor chain.
+ */
+ messageInterceptor?: MessageInterceptor | MessageInterceptor[];
}
/**
@@ -95,6 +109,8 @@ export abstract class BaseClient {
protected readonly logger: ILogger;
protected readonly rpcClientManager: RpcClientManager;
readonly #telemetrySessions = new Map<string, TelemetrySession>();
+ readonly #compositedMessageInterceptor = new CompositedMessageInterceptor();
+ protected readonly clientMeterManager: ClientMeterManager;
#startupResolve?: () => void;
#startupReject?: (err: Error) => void;
#timers: NodeJS.Timeout[] = [];
@@ -139,6 +155,12 @@ export abstract class BaseClient {
// Default request timeout is 3000ms
this.requestTimeout = options.requestTimeout ?? 3000;
this.rpcClientManager = new RpcClientManager(this, this.logger);
+ // Wire the metrics pipeline: a dedicated meter manager plus an interceptor
+ // that records the four RocketMQ histograms around the
SEND/RECEIVE/CONSUME
+ // hook points. Mirrors Java's ClientImpl#initClientMeterManager.
+ this.clientMeterManager = new ClientMeterManager(this.clientId, () =>
this.getRequestMetadata(), this.logger);
+ this.addMessageInterceptor(
+ new MessageMeterInterceptor(this.clientMeterManager, this.clientId, ()
=> this.getConsumerGroup()));
if (options.topics) {
for (const topic of options.topics) {
if (topic && topic.trim().length > 0) {
@@ -146,6 +168,14 @@ export abstract class BaseClient {
}
}
}
+ // Register message interceptors provided by the user (allowed before
startup only).
+ if (options.messageInterceptor) {
+ const interceptors = Array.isArray(options.messageInterceptor)
+ ? options.messageInterceptor : [ options.messageInterceptor ];
+ for (const interceptor of interceptors) {
+ this.addMessageInterceptor(interceptor);
+ }
+ }
}
/**
@@ -276,6 +306,10 @@ export abstract class BaseClient {
// 4. Close RPC connections
this.rpcClientManager.close();
+ // 4.5 Close the metrics export pipeline (flushes pending exports and
releases
+ // the OTLP/gRPC channel). Mirrors ClientImpl#shutdown's meter manager
close.
+ await this.clientMeterManager.shutdown();
+
// 5. Clear caches
this.topicRouteCache.clear();
this.inflightRouteFutures.clear();
@@ -513,6 +547,15 @@ export abstract class BaseClient {
return metadata;
}
+ /**
+ * Consumer group of this client, used as a metric attribute for consumer
+ * histograms/gauges. Producers have no consumer group, so the base
+ * implementation returns undefined; Consumer overrides this.
+ */
+ getConsumerGroup(): string | undefined {
+ return undefined;
+ }
+
protected abstract getSettings(): Settings;
/**
@@ -561,20 +604,19 @@ export abstract class BaseClient {
onSettingsCommand(_endpoints: Endpoints, settings: SettingsPB) {
this.logger.info('Received settings command, clientId=%s, settings=%j',
this.clientId, settings.toObject());
- // final Metric metric = new Metric(settings.getMetric());
- // clientMeterManager.reset(metric);
+ const metric = new Metric(settings.getMetric());
+ this.clientMeterManager.reset(metric);
this.getSettings().sync(settings);
this.logger.info('Sync settings=%j, clientId=%s', this.getSettings(),
this.clientId);
this.#startupResolve && this.#startupResolve();
}
onRecoverOrphanedTransactionCommand(_endpoints: Endpoints, command:
RecoverOrphanedTransactionCommand) {
+ // The base client does not support transactions; subclasses (e.g.
Producer) override
+ // this method to recover orphaned transactional messages. Mirroring the
Java client
+ // (ClientImpl#onRecoverOrphanedTransactionCommand), no reply is sent back
to remote.
this.logger.warn('Ignore orphaned transaction recovery command from
remote, which is not expected, clientId=%s, command=%j',
this.clientId, command.toObject());
- // const telemetryCommand = new TelemetryCommand();
- // telemetryCommand.setStatus(new Status().setCode(Code.NOT_IMPLEMENTED));
- // telemetryCommand.setRecoverOrphanedTransactionCommand(new
RecoverOrphanedTransactionCommand());
- // this.telemetry(endpoints, telemetryCommand);
}
onVerifyMessageCommand(endpoints: Endpoints, command: VerifyMessageCommand) {
@@ -589,6 +631,13 @@ export abstract class BaseClient {
this.telemetry(endpoints, telemetryCommand);
}
+ /**
+ * Handle the notify-unsubscribe-lite command sent by remote when a lite
topic
+ * subscription must be dropped (e.g. quota violation or administrative
action).
+ *
+ * The base client ignores it, mirroring Java
ClientImpl#onNotifyUnsubscribeLiteCommand;
+ * lite consumers override it to drive their lite subscription manager.
+ */
onNotifyUnsubscribeLiteCommand(endpoints: Endpoints, command:
NotifyUnsubscribeLiteCommand) {
this.logger.warn('Ignore notify unsubscribe lite command from remote,
which is not expected, clientId=%s, endpoints=%s, command=%j',
this.clientId, endpoints.facade, command.toObject());
@@ -652,6 +701,42 @@ export abstract class BaseClient {
return this.endpoints;
}
+ /**
+ * Add a message interceptor to the client. Interceptors can only be
registered
+ * before the client is running, mirroring Java
ClientImpl#addMessageInterceptor.
+ */
+ addMessageInterceptor(messageInterceptor: MessageInterceptor) {
+ if (!this.isRunning()) {
+ this.#compositedMessageInterceptor.addInterceptor(messageInterceptor);
+ }
+ }
+
+ /**
+ * Invoke the interceptor chain before a hook point is executed.
+ */
+ doBefore(context: MessageInterceptorContext, generalMessages:
GeneralMessage[]) {
+ try {
+ this.#compositedMessageInterceptor.doBefore(context, generalMessages);
+ } catch (t) {
+ // Should never reach here, the composite interceptor guards each
interceptor.
+ this.logger.error('[Bug] Exception raised while handling messages,
clientId=%s, error=%s',
+ this.clientId, t);
+ }
+ }
+
+ /**
+ * Invoke the interceptor chain after a hook point is executed.
+ */
+ doAfter(context: MessageInterceptorContext, generalMessages:
GeneralMessage[]) {
+ try {
+ this.#compositedMessageInterceptor.doAfter(context, generalMessages);
+ } catch (t) {
+ // Should never reach here, the composite interceptor guards each
interceptor.
+ this.logger.error('[Bug] Exception raised while handling messages,
clientId=%s, error=%s',
+ this.clientId, t);
+ }
+ }
+
/**
* Get the RPC client manager.
*
diff --git a/nodejs/src/consumer/ConsumeService.ts
b/nodejs/src/consumer/ConsumeService.ts
index a241eaff..c46b2783 100644
--- a/nodejs/src/consumer/ConsumeService.ts
+++ b/nodejs/src/consumer/ConsumeService.ts
@@ -15,6 +15,7 @@
* limitations under the License.
*/
import { MessageView } from '../message';
+import { MessageInterceptor } from '../hook';
import { ConsumeResult } from './ConsumeResult';
import { ConsumeTask } from './ConsumeTask';
import { MessageListener } from './MessageListener';
@@ -23,12 +24,17 @@ import type { ProcessQueue } from './ProcessQueue';
export abstract class ConsumeService {
protected readonly clientId: string;
readonly #messageListener: MessageListener;
+ readonly #messageInterceptor?: MessageInterceptor;
+ readonly #consumerGroup?: string;
#aborted = false;
#pendingTimers: Set<NodeJS.Timeout> = new Set();
- constructor(clientId: string, messageListener: MessageListener) {
+ constructor(clientId: string, messageListener: MessageListener,
+ messageInterceptor?: MessageInterceptor, consumerGroup?: string) {
this.clientId = clientId;
this.#messageListener = messageListener;
+ this.#messageInterceptor = messageInterceptor;
+ this.#consumerGroup = consumerGroup;
}
abstract consume(pq: ProcessQueue, messageViews: MessageView[]): void;
@@ -37,7 +43,8 @@ export abstract class ConsumeService {
if (this.#aborted) {
return ConsumeResult.FAILURE;
}
- const task = new ConsumeTask(this.clientId, this.#messageListener,
messageView);
+ const task = new ConsumeTask(this.clientId, this.#messageListener,
messageView,
+ this.#messageInterceptor, this.#consumerGroup);
if (delay <= 0) {
return this.#executeTask(task);
}
diff --git a/nodejs/src/consumer/ConsumeTask.ts
b/nodejs/src/consumer/ConsumeTask.ts
index b502e7f0..ee95151f 100644
--- a/nodejs/src/consumer/ConsumeTask.ts
+++ b/nodejs/src/consumer/ConsumeTask.ts
@@ -16,24 +16,64 @@
*/
import { MessageView } from '../message';
+import {
+ Attribute,
+ AttributeKey,
+ MessageHookPoints,
+ MessageHookPointsStatus,
+ MessageInterceptor,
+ MessageInterceptorContextImpl,
+} from '../hook';
import { ConsumeResult } from './ConsumeResult';
import { MessageListener } from './MessageListener';
+// Context keys exposed to interceptors, mirroring the Java ConsumeTask keys.
+export const REMOTE_ADDR_CONTEXT_KEY =
AttributeKey.create<string>('remote_address');
+export const MESSAGE_VIEW_CONTEXT_KEY =
AttributeKey.create<MessageView>('message_view');
+export const CONSUMER_GROUP_CONTEXT_KEY =
AttributeKey.create<string>('consumer_group');
+export const CONSUME_ERROR_CONTEXT_KEY =
AttributeKey.create<unknown>('consume_error');
+
export class ConsumeTask {
readonly #messageListener: MessageListener;
readonly #messageView: MessageView;
+ readonly #consumerGroup?: string;
+ readonly #messageInterceptor?: MessageInterceptor;
- constructor(_clientId: string, messageListener: MessageListener,
messageView: MessageView) {
+ constructor(_clientId: string, messageListener: MessageListener,
messageView: MessageView,
+ messageInterceptor?: MessageInterceptor, consumerGroup?: string) {
this.#messageListener = messageListener;
this.#messageView = messageView;
+ this.#messageInterceptor = messageInterceptor;
+ this.#consumerGroup = consumerGroup;
}
async call(): Promise<ConsumeResult> {
+ const generalMessages = [ this.#messageView ];
+ let context = new MessageInterceptorContextImpl(MessageHookPoints.CONSUME);
+ context.putAttribute(MESSAGE_VIEW_CONTEXT_KEY,
Attribute.create(this.#messageView));
+ if (this.#consumerGroup) {
+ context.putAttribute(CONSUMER_GROUP_CONTEXT_KEY,
Attribute.create(this.#consumerGroup));
+ }
+ let throwable: unknown;
+ let consumeResult: ConsumeResult;
+ if (this.#messageInterceptor) {
+ this.#messageInterceptor.doBefore(context, generalMessages);
+ }
try {
- return await this.#messageListener.consume(this.#messageView);
+ consumeResult = await this.#messageListener.consume(this.#messageView);
} catch (e) {
// Message listener raised an exception while consuming messages
- return ConsumeResult.FAILURE;
+ throwable = e;
+ consumeResult = ConsumeResult.FAILURE;
+ }
+ const status = consumeResult === ConsumeResult.SUCCESS ?
MessageHookPointsStatus.OK : MessageHookPointsStatus.ERROR;
+ context = MessageInterceptorContextImpl.withStatus(context, status);
+ if (throwable) {
+ context.putAttribute(CONSUME_ERROR_CONTEXT_KEY,
Attribute.create(throwable));
+ }
+ if (this.#messageInterceptor) {
+ this.#messageInterceptor.doAfter(context, generalMessages);
}
+ return consumeResult;
}
}
diff --git a/nodejs/src/consumer/Consumer.ts b/nodejs/src/consumer/Consumer.ts
index 23a4330a..f312fe4f 100644
--- a/nodejs/src/consumer/Consumer.ts
+++ b/nodejs/src/consumer/Consumer.ts
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-import { Message, Status } from '../../proto/apache/rocketmq/v2/definition_pb';
+import { Code, Message, Status } from
'../../proto/apache/rocketmq/v2/definition_pb';
import {
AckMessageRequest,
ChangeInvisibleDurationRequest,
@@ -27,6 +27,12 @@ import { MessageQueue } from '../route';
import { StatusChecker } from '../exception';
import { BaseClient, BaseClientOptions } from '../client';
import { createDuration, createResource } from '../util';
+import {
+ GeneralMessage,
+ MessageHookPoints,
+ MessageHookPointsStatus,
+ MessageInterceptorContextImpl,
+} from '../hook';
import { FilterExpression } from './FilterExpression';
export interface ConsumerOptions extends BaseClientOptions {
@@ -91,10 +97,18 @@ export abstract class Consumer extends BaseClient {
}
const messages = messageList.map(message => new MessageView(message, mq,
transportDeliveryTimestamp));
+ // RECEIVE hook point, mirroring Java ProcessQueueImpl.
+ const context = new
MessageInterceptorContextImpl(MessageHookPoints.RECEIVE,
MessageHookPointsStatus.OK);
+ this.doBefore(context, messages);
+ this.doAfter(MessageInterceptorContextImpl.withStatus(context,
MessageHookPointsStatus.OK), messages);
return messages;
} catch (err) {
this.logger.error('Failed to receive messages, topic=%s, endpoints=%s,
clientId=%s, error=%s',
request.getMessageQueue()?.getTopic()?.getName(), endpoints, (this as
any).clientId, err);
+ // RECEIVE hook point with error status.
+ const context = new
MessageInterceptorContextImpl(MessageHookPoints.RECEIVE);
+ this.doBefore(context, []);
+ this.doAfter(MessageInterceptorContextImpl.withStatus(context,
MessageHookPointsStatus.ERROR), []);
throw err;
}
}
@@ -110,11 +124,23 @@ export abstract class Consumer extends BaseClient {
if (this.isLiteConsumer() && messageView.liteTopic) {
entry.setLiteTopic(messageView.liteTopic);
}
- const res = await this.rpcClientManager.ackMessage(endpoints, request,
this.requestTimeout);
- // FIXME: handle fail ack
- const response = res.toObject();
- StatusChecker.check(response.status);
- return response.entriesList;
+ // ACK hook point, mirroring Java ConsumerImpl#ackMessage.
+ const context = new MessageInterceptorContextImpl(MessageHookPoints.ACK);
+ const generalMessages = [ messageView ];
+ this.doBefore(context, generalMessages);
+ let hookStatus = MessageHookPointsStatus.OK;
+ try {
+ const res = await this.rpcClientManager.ackMessage(endpoints, request,
this.requestTimeout);
+ // FIXME: handle fail ack
+ const response = res.toObject();
+ hookStatus = response.status?.code === Code.OK ?
MessageHookPointsStatus.OK : MessageHookPointsStatus.ERROR;
+ StatusChecker.check(response.status);
+ this.doAfter(MessageInterceptorContextImpl.withStatus(context,
hookStatus), generalMessages);
+ return response.entriesList;
+ } catch (err) {
+ this.doAfter(MessageInterceptorContextImpl.withStatus(context,
MessageHookPointsStatus.ERROR), generalMessages);
+ throw err;
+ }
}
protected async invisibleDuration(messageView: MessageView,
invisibleDuration: number) {
@@ -128,10 +154,22 @@ export abstract class Consumer extends BaseClient {
request.setLiteTopic(messageView.liteTopic);
}
- const res = await
this.rpcClientManager.changeInvisibleDuration(messageView.endpoints, request,
this.requestTimeout);
- const response = res.toObject();
- StatusChecker.check(response.status);
- return response.receiptHandle;
+ // CHANGE_INVISIBLE_DURATION hook point, mirroring Java
ConsumerImpl#changeInvisibleDuration.
+ const context = new
MessageInterceptorContextImpl(MessageHookPoints.CHANGE_INVISIBLE_DURATION);
+ const generalMessages = [ messageView ];
+ this.doBefore(context, generalMessages);
+ let hookStatus = MessageHookPointsStatus.OK;
+ try {
+ const res = await
this.rpcClientManager.changeInvisibleDuration(messageView.endpoints, request,
this.requestTimeout);
+ const response = res.toObject();
+ hookStatus = response.status?.code === Code.OK ?
MessageHookPointsStatus.OK : MessageHookPointsStatus.ERROR;
+ StatusChecker.check(response.status);
+ this.doAfter(MessageInterceptorContextImpl.withStatus(context,
hookStatus), generalMessages);
+ return response.receiptHandle;
+ } catch (err) {
+ this.doAfter(MessageInterceptorContextImpl.withStatus(context,
MessageHookPointsStatus.ERROR), generalMessages);
+ throw err;
+ }
}
/**
@@ -147,9 +185,27 @@ export abstract class Consumer extends BaseClient {
return res;
}
- async forwardMessageToDeadLetterQueueViaRpc(endpoints: any, request: any,
timeout: number) {
- const res = await
this.rpcClientManager.forwardMessageToDeadLetterQueue(endpoints, request,
timeout);
- return res;
+ /**
+ * Forward the message to the dead letter queue, mirroring Java
+ * PushConsumerImpl#forwardMessageToDeadLetterQueue. The FORWARD_TO_DLQ hook
+ * point is triggered around every RPC attempt.
+ */
+ async forwardMessageToDeadLetterQueueViaRpc(endpoints: any, request: any,
timeout: number,
+ messageView: MessageView) {
+ // FORWARD_TO_DLQ hook point, mirroring Java
PushConsumerImpl#forwardMessageToDeadLetterQueue.
+ const context = new
MessageInterceptorContextImpl(MessageHookPoints.FORWARD_TO_DLQ);
+ const generalMessages: GeneralMessage[] = [ messageView ];
+ this.doBefore(context, generalMessages);
+ try {
+ const response = await
this.rpcClientManager.forwardMessageToDeadLetterQueue(endpoints, request,
timeout);
+ const hookStatus = response.getStatus()?.getCode() === Code.OK ?
+ MessageHookPointsStatus.OK : MessageHookPointsStatus.ERROR;
+ this.doAfter(MessageInterceptorContextImpl.withStatus(context,
hookStatus), generalMessages);
+ return response;
+ } catch (err) {
+ this.doAfter(MessageInterceptorContextImpl.withStatus(context,
MessageHookPointsStatus.ERROR), generalMessages);
+ throw err;
+ }
}
/**
diff --git a/nodejs/src/consumer/FifoConsumeService.ts
b/nodejs/src/consumer/FifoConsumeService.ts
index f7bfae74..fe2f7a94 100644
--- a/nodejs/src/consumer/FifoConsumeService.ts
+++ b/nodejs/src/consumer/FifoConsumeService.ts
@@ -16,6 +16,7 @@
*/
import { MessageView } from '../message';
+import { MessageInterceptor } from '../hook';
import { ConsumeService } from './ConsumeService';
import { MessageListener } from './MessageListener';
import type { ProcessQueue } from './ProcessQueue';
@@ -25,8 +26,9 @@ export class FifoConsumeService extends ConsumeService {
readonly #enableFifoConsumeAccelerator: boolean;
readonly #logger: ILogger;
- constructor(clientId: string, messageListener: MessageListener,
enableFifoConsumeAccelerator?: boolean) {
- super(clientId, messageListener);
+ constructor(clientId: string, messageListener: MessageListener,
enableFifoConsumeAccelerator?: boolean,
+ messageInterceptor?: MessageInterceptor, consumerGroup?: string) {
+ super(clientId, messageListener, messageInterceptor, consumerGroup);
this.#enableFifoConsumeAccelerator = enableFifoConsumeAccelerator ?? false;
this.#logger = getDefaultLogger();
}
diff --git a/nodejs/src/consumer/LiteFifoConsumeService.ts
b/nodejs/src/consumer/LiteFifoConsumeService.ts
index 29f39ab3..3923d074 100644
--- a/nodejs/src/consumer/LiteFifoConsumeService.ts
+++ b/nodejs/src/consumer/LiteFifoConsumeService.ts
@@ -16,6 +16,7 @@
*/
import { MessageView } from '../message';
+import { MessageInterceptor } from '../hook';
import { ConsumeResultSuspend } from './ConsumeResult';
import { FifoConsumeService } from './FifoConsumeService';
import { MessageListener } from './MessageListener';
@@ -37,8 +38,9 @@ import { ILogger, getDefaultLogger } from '../client/Logger';
export class LiteFifoConsumeService extends FifoConsumeService {
readonly #logger: ILogger;
- constructor(clientId: string, messageListener: MessageListener,
enableFifoConsumeAccelerator?: boolean) {
- super(clientId, messageListener, enableFifoConsumeAccelerator);
+ constructor(clientId: string, messageListener: MessageListener,
enableFifoConsumeAccelerator?: boolean,
+ messageInterceptor?: MessageInterceptor, consumerGroup?: string) {
+ super(clientId, messageListener, enableFifoConsumeAccelerator,
messageInterceptor, consumerGroup);
this.#logger = getDefaultLogger();
}
diff --git a/nodejs/src/consumer/LitePushConsumerImpl.ts
b/nodejs/src/consumer/LitePushConsumerImpl.ts
index 7693cb89..86f8e27f 100644
--- a/nodejs/src/consumer/LitePushConsumerImpl.ts
+++ b/nodejs/src/consumer/LitePushConsumerImpl.ts
@@ -95,10 +95,11 @@ export class LitePushConsumerImpl extends PushConsumer
implements LitePushConsum
'Create lite FIFO consume service, consumerGroup=%s, clientId=%s,
enableFifoConsumeAccelerator=%s',
this.consumerGroup, this.clientId, this.#enableFifoConsumeAccelerator,
);
- return new LiteFifoConsumeService(this.clientId, this.#messageListener,
this.#enableFifoConsumeAccelerator);
+ return new LiteFifoConsumeService(this.clientId, this.#messageListener,
this.#enableFifoConsumeAccelerator,
+ this, this.consumerGroup);
}
this.logger.info('Create lite standard consume service, consumerGroup=%s,
clientId=%s', this.consumerGroup, this.clientId);
- return new LiteStandardConsumeService(this.clientId,
this.#messageListener);
+ return new LiteStandardConsumeService(this.clientId,
this.#messageListener, this, this.consumerGroup);
}
/**
@@ -218,10 +219,12 @@ export class LitePushConsumerImpl extends PushConsumer
implements LitePushConsum
* <p>This method is called when the server sends a notification to
unsubscribe
* from a lite topic, typically due to quota violations or administrative
actions.</p>
*
- * @param _endpoints - The server endpoints
+ * @param endpoints - The remote endpoints which sent the command
* @param command - The unsubscribe command from the server
*/
- onNotifyUnsubscribeLiteCommand(_endpoints: Endpoints, command:
NotifyUnsubscribeLiteCommand) {
+ onNotifyUnsubscribeLiteCommand(endpoints: Endpoints, command:
NotifyUnsubscribeLiteCommand) {
+ this.logger.info('Received notify unsubscribe lite command, liteTopic=%s,
endpoints=%s, clientId=%s',
+ command.getLiteTopic(), endpoints.facade, this.clientId);
this.liteSubscriptionManager.onNotifyUnsubscribeLiteCommand(command);
}
diff --git a/nodejs/src/consumer/LiteStandardConsumeService.ts
b/nodejs/src/consumer/LiteStandardConsumeService.ts
index f3d93ced..5f41f1f0 100644
--- a/nodejs/src/consumer/LiteStandardConsumeService.ts
+++ b/nodejs/src/consumer/LiteStandardConsumeService.ts
@@ -16,6 +16,7 @@
*/
import { MessageView } from '../message';
+import { MessageInterceptor } from '../hook';
import { ConsumeService } from './ConsumeService';
import { MessageListener } from './MessageListener';
import type { ProcessQueue } from './ProcessQueue';
@@ -31,8 +32,9 @@ import { ILogger, getDefaultLogger } from '../client/Logger';
export class LiteStandardConsumeService extends ConsumeService {
readonly #logger: ILogger;
- constructor(clientId: string, messageListener: MessageListener) {
- super(clientId, messageListener);
+ constructor(clientId: string, messageListener: MessageListener,
+ messageInterceptor?: MessageInterceptor, consumerGroup?: string) {
+ super(clientId, messageListener, messageInterceptor, consumerGroup);
this.#logger = getDefaultLogger();
}
diff --git a/nodejs/src/consumer/ProcessQueue.ts
b/nodejs/src/consumer/ProcessQueue.ts
index e4e2c7fd..6a0fde13 100644
--- a/nodejs/src/consumer/ProcessQueue.ts
+++ b/nodejs/src/consumer/ProcessQueue.ts
@@ -53,6 +53,14 @@ export class ProcessQueue {
return this.#mq;
}
+ /**
+ * Topic of the underlying message queue, used as a metric attribute when
+ * aggregating the consumer gauges.
+ */
+ get topic(): string {
+ return this.#mq.topic.name;
+ }
+
drop(): void {
this.#dropped = true;
}
@@ -326,6 +334,7 @@ export class ProcessQueue {
endpoints,
this.#consumer.wrapForwardMessageToDeadLetterQueueRequest(messageView),
this.#consumer.requestTimeoutValue,
+ messageView,
);
const status = response.getStatus();
if (!status) {
diff --git a/nodejs/src/consumer/PushConsumer.ts
b/nodejs/src/consumer/PushConsumer.ts
index f680425c..301d0bc1 100644
--- a/nodejs/src/consumer/PushConsumer.ts
+++ b/nodejs/src/consumer/PushConsumer.ts
@@ -38,9 +38,11 @@ import { ConsumeService } from './ConsumeService';
import { StandardConsumeService } from './StandardConsumeService';
import { FifoConsumeService } from './FifoConsumeService';
import { ProcessQueue } from './ProcessQueue';
+import { PushConsumerGaugeObserver } from './PushConsumerGaugeObserver';
import { Assignment } from './Assignment';
import { Assignments } from './Assignments';
import { MessageListener } from './MessageListener';
+import { InflightRequestCountInterceptor } from '../hook';
const ASSIGNMENT_SCAN_SCHEDULE_DELAY = 1000;
const ASSIGNMENT_SCAN_SCHEDULE_PERIOD = 5000;
@@ -71,6 +73,7 @@ export class PushConsumer extends Consumer {
readonly #enableFifoConsumeAccelerator: boolean;
readonly #processQueueTable = new Map<string /* mq key */, { mq:
MessageQueue; pq: ProcessQueue }>();
readonly #metrics = new ConsumeMetrics();
+ readonly #inflightRequestCountInterceptor: InflightRequestCountInterceptor;
#consumeService!: ConsumeService;
#scanAssignmentTimer?: NodeJS.Timeout;
#inflightReceiveRequestCount = 0;
@@ -92,6 +95,16 @@ export class PushConsumer extends Consumer {
this.#maxCacheMessageSizeInBytes = options.maxCacheMessageSizeInBytes ??
64 * 1024 * 1024;
this.#enableFifoConsumeAccelerator = options.enableFifoConsumeAccelerator
?? false;
+ // Built-in interceptor tracking in-flight receive requests, mirroring
Java PushConsumerImpl.
+ this.#inflightRequestCountInterceptor = new
InflightRequestCountInterceptor();
+ this.addMessageInterceptor(this.#inflightRequestCountInterceptor);
+
+ // Register the consumer gauge observer so cached message count/bytes are
+ // exported to the broker. The provider reads the process-queue table
lazily
+ // at scrape time, so registering here (before any queue exists) is safe.
+ this.clientMeterManager.setGaugeObserver(
+ new PushConsumerGaugeObserver(() => this.processQueues(), this.clientId,
this.consumerGroup));
+
this.#pushSubscriptionSettings = new PushSubscriptionSettings(
options.namespace, this.clientId, this.getClientType(), this.endpoints,
this.consumerGroup, this.requestTimeout, this.#subscriptionExpressions,
@@ -230,11 +243,22 @@ export class PushConsumer extends Consumer {
// No-op for push consumer; assignments are queried separately
}
+ /**
+ * Get the count of in-flight receive requests tracked by the built-in
+ * {@link InflightRequestCountInterceptor}, mirroring Java PushConsumerImpl.
+ */
+ protected getInflightReceiveRequestCountFromInterceptor(): number {
+ return
this.#inflightRequestCountInterceptor.getInflightReceiveRequestCount();
+ }
+
protected createConsumeService(): ConsumeService {
+ // `this` implements MessageInterceptor (doBefore/doAfter delegating to the
+ // composited interceptor chain), mirroring Java PushConsumerImpl.
if (this.#pushSubscriptionSettings.isFifo()) {
- return new FifoConsumeService(this.clientId, this.#messageListener,
this.#enableFifoConsumeAccelerator);
+ return new FifoConsumeService(this.clientId, this.#messageListener,
this.#enableFifoConsumeAccelerator,
+ this, this.consumerGroup);
}
- return new StandardConsumeService(this.clientId, this.#messageListener);
+ return new StandardConsumeService(this.clientId, this.#messageListener,
this, this.consumerGroup);
}
async subscribe(topic: string, filterExpression: FilterExpression) {
@@ -410,6 +434,14 @@ export class PushConsumer extends Consumer {
return this.#processQueueTable.size;
}
+ /**
+ * All live process queues, used by the consumer gauge observer to aggregate
+ * cached message count/bytes. Reads the private table lazily at scrape time.
+ */
+ processQueues(): ProcessQueue[] {
+ return [ ...this.#processQueueTable.values() ].map(entry => entry.pq);
+ }
+
cacheMessageBytesThresholdPerQueue(): number {
const size = this.getQueueSize();
if (size <= 0) return 0;
diff --git a/nodejs/src/consumer/PushConsumerGaugeObserver.ts
b/nodejs/src/consumer/PushConsumerGaugeObserver.ts
new file mode 100644
index 00000000..c7a32be2
--- /dev/null
+++ b/nodejs/src/consumer/PushConsumerGaugeObserver.ts
@@ -0,0 +1,60 @@
+/**
+ * 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 } from '@opentelemetry/api';
+import { ProcessQueue } from './ProcessQueue';
+import { GaugeEnum } from '../metrics/GaugeEnum';
+import { GaugeObserver } from '../metrics/GaugeObserver';
+import { MetricLabels } from '../metrics/MetricLabels';
+
+/**
+ * Aggregates the per-process-queue cache statistics of a push consumer into
the
+ * two RocketMQ consumer gauges, mirroring the GaugeObserver implementation
held
+ * by org.apache.rocketmq.client.java.impl.consumer.PushConsumerImpl.
+ */
+export class PushConsumerGaugeObserver implements GaugeObserver {
+ constructor(
+ private readonly processQueuesProvider: () => ProcessQueue[],
+ private readonly clientId: string,
+ private readonly consumerGroup: string,
+ ) {
+ }
+
+ getGauges(): GaugeEnum[] {
+ return [ GaugeEnum.CONSUMER_CACHED_MESSAGES,
GaugeEnum.CONSUMER_CACHED_BYTES ];
+ }
+
+ getValues(gauge: GaugeEnum): Map<Attributes, number> {
+ const result = new Map<Attributes, number>();
+ const byTopic = new Map<string, number>();
+ for (const pq of this.processQueuesProvider()) {
+ const increment = gauge === GaugeEnum.CONSUMER_CACHED_MESSAGES
+ ? pq.cachedMessagesCount()
+ : pq.cachedMessageBytes();
+ byTopic.set(pq.topic, (byTopic.get(pq.topic) ?? 0) + increment);
+ }
+ for (const [ topic, value ] of byTopic) {
+ const attributes: Attributes = {
+ [MetricLabels.TOPIC]: topic,
+ [MetricLabels.CLIENT_ID]: this.clientId,
+ [MetricLabels.CONSUMER_GROUP]: this.consumerGroup,
+ };
+ result.set(attributes, value);
+ }
+ return result;
+ }
+}
diff --git a/nodejs/src/consumer/StandardConsumeService.ts
b/nodejs/src/consumer/StandardConsumeService.ts
index bd507902..1a40a754 100644
--- a/nodejs/src/consumer/StandardConsumeService.ts
+++ b/nodejs/src/consumer/StandardConsumeService.ts
@@ -16,6 +16,7 @@
*/
import { MessageView } from '../message';
+import { MessageInterceptor } from '../hook';
import { ConsumeService } from './ConsumeService';
import { MessageListener } from './MessageListener';
import type { ProcessQueue } from './ProcessQueue';
@@ -25,8 +26,9 @@ export class StandardConsumeService extends ConsumeService {
private readonly logger = getDefaultLogger();
// eslint-disable-next-line @typescript-eslint/no-useless-constructor
- constructor(clientId: string, messageListener: MessageListener) {
- super(clientId, messageListener);
+ constructor(clientId: string, messageListener: MessageListener,
+ messageInterceptor?: MessageInterceptor, consumerGroup?: string) {
+ super(clientId, messageListener, messageInterceptor, consumerGroup);
}
consume(pq: ProcessQueue, messageViews: MessageView[]): void {
diff --git a/nodejs/src/index.ts b/nodejs/src/hook/Attribute.ts
similarity index 67%
copy from nodejs/src/index.ts
copy to nodejs/src/hook/Attribute.ts
index 28ec5f3b..c79e48a8 100644
--- a/nodejs/src/index.ts
+++ b/nodejs/src/hook/Attribute.ts
@@ -15,10 +15,22 @@
* limitations under the License.
*/
-export * from './consumer';
-export * from './exception';
-export * from './message';
-export * from './producer';
-export * from './retry';
-export * from './route';
-export * from './client';
+/**
+ * Immutable value holder shared through a {@link MessageInterceptorContext},
+ * mirroring org.apache.rocketmq.client.java.hook.Attribute.
+ */
+export class Attribute<T> {
+ readonly #value: T;
+
+ private constructor(value: T) {
+ this.#value = value;
+ }
+
+ static create<T>(value: T): Attribute<T> {
+ return new Attribute(value);
+ }
+
+ get(): T {
+ return this.#value;
+ }
+}
diff --git a/nodejs/src/index.ts b/nodejs/src/hook/AttributeKey.ts
similarity index 57%
copy from nodejs/src/index.ts
copy to nodejs/src/hook/AttributeKey.ts
index 28ec5f3b..64e11b20 100644
--- a/nodejs/src/index.ts
+++ b/nodejs/src/hook/AttributeKey.ts
@@ -15,10 +15,28 @@
* limitations under the License.
*/
-export * from './consumer';
-export * from './exception';
-export * from './message';
-export * from './producer';
-export * from './retry';
-export * from './route';
-export * from './client';
+/**
+ * Typed key of a context attribute, mirroring
+ * org.apache.rocketmq.client.java.hook.AttributeKey.
+ *
+ * The type parameter is only used for compile-time typing of the associated
value.
+ */
+export class AttributeKey<T> {
+ readonly name: string;
+ /**
+ * Never assigned at runtime; it only carries the value type for
compile-time typing.
+ */
+ readonly typeHint?: T;
+
+ constructor(name: string) {
+ this.name = name;
+ }
+
+ static create<T>(name: string): AttributeKey<T> {
+ return new AttributeKey<T>(name);
+ }
+
+ toString() {
+ return `AttributeKey(${this.name})`;
+ }
+}
diff --git a/nodejs/src/hook/CompositedMessageInterceptor.ts
b/nodejs/src/hook/CompositedMessageInterceptor.ts
new file mode 100644
index 00000000..a884d0bc
--- /dev/null
+++ b/nodejs/src/hook/CompositedMessageInterceptor.ts
@@ -0,0 +1,98 @@
+/**
+ * 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 { Attribute } from './Attribute';
+import { AttributeKey } from './AttributeKey';
+import { MessageHookPointsStatus } from './MessageHookPointsStatus';
+import { MessageInterceptorContext } from './MessageInterceptorContext';
+import { MessageInterceptorContextImpl } from
'./MessageInterceptorContextImpl';
+import { GeneralMessage, MessageInterceptor } from './MessageInterceptor';
+
+// Key under which per-interceptor attributes are stashed between doBefore and
doAfter.
+const INTERCEPTOR_ATTRIBUTES_KEY = AttributeKey
+ .create<Map<number, Map<AttributeKey<any>,
Attribute<any>>>>('composited_interceptor_attributes');
+
+/**
+ * Composite interceptor which delegates to a chain of interceptors: doBefore
runs
+ * in registration order while doAfter runs in reverse order, mirroring
+ * org.apache.rocketmq.client.java.hook.CompositedMessageInterceptor.
+ */
+export class CompositedMessageInterceptor implements MessageInterceptor {
+ readonly #interceptors: MessageInterceptor[] = [];
+
+ constructor(interceptors: MessageInterceptor[] = []) {
+ this.#interceptors.push(...interceptors);
+ }
+
+ addInterceptor(interceptor: MessageInterceptor) {
+ this.#interceptors.push(interceptor);
+ }
+
+ get size() {
+ return this.#interceptors.length;
+ }
+
+ doBefore(context0: MessageInterceptorContext, messages: GeneralMessage[]) {
+ const attributeMap = new Map<number, Map<AttributeKey<any>,
Attribute<any>>>();
+ for (let index = 0; index < this.#interceptors.length; index++) {
+ const context = new
MessageInterceptorContextImpl(context0.getMessageHookPoints(),
context0.getStatus());
+ // Share the attributes accumulated by the caller and previous
interceptors.
+ for (const [ key, value ] of context0.getAttributes()) {
+ context.putAttribute(key, value);
+ }
+ try {
+ this.#interceptors[index].doBefore(context, messages);
+ } catch (t) {
+ // Interceptors must never break the message pipeline.
+ // eslint-disable-next-line no-console
+ console.error(
+ '[CompositedMessageInterceptor] Exception raised while handling
messages before hook point=%s, error=%s',
+ context0.getMessageHookPoints(), t);
+ }
+ attributeMap.set(index, context.getAttributes());
+ }
+ context0.putAttribute(INTERCEPTOR_ATTRIBUTES_KEY,
Attribute.create(attributeMap));
+ // Propagate attributes collected by the first interceptor to the caller
context.
+ const firstAttributes = attributeMap.get(0);
+ if (firstAttributes) {
+ for (const [ key, value ] of firstAttributes) {
+ context0.putAttribute(key, value);
+ }
+ }
+ }
+
+ doAfter(context0: MessageInterceptorContext, messages: GeneralMessage[]) {
+ const attributeMap =
context0.getAttribute(INTERCEPTOR_ATTRIBUTES_KEY)?.get()
+ ?? new Map<number, Map<AttributeKey<any>, Attribute<any>>>();
+ for (let index = this.#interceptors.length - 1; index >= 0; index--) {
+ const attributes = attributeMap.get(index);
+ const context = new MessageInterceptorContextImpl(
+ context0.getMessageHookPoints(), context0.getStatus() ??
MessageHookPointsStatus.OK, attributes);
+ for (const [ key, value ] of context0.getAttributes()) {
+ context.putAttribute(key, value);
+ }
+ try {
+ this.#interceptors[index].doAfter(context, messages);
+ } catch (t) {
+ // eslint-disable-next-line no-console
+ console.error(
+ '[CompositedMessageInterceptor] Exception raised while handling
messages after hook point=%s, error=%s',
+ context0.getMessageHookPoints(), t);
+ }
+ }
+ }
+}
diff --git a/nodejs/src/hook/InflightRequestCountInterceptor.ts
b/nodejs/src/hook/InflightRequestCountInterceptor.ts
new file mode 100644
index 00000000..c508532c
--- /dev/null
+++ b/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) {
+ this.#inflightReceiveRequestCount++;
+ }
+ }
+
+ doAfter(context: MessageInterceptorContext, _messages: GeneralMessage[]) {
+ if (context.getMessageHookPoints() === MessageHookPoints.RECEIVE) {
+ this.#inflightReceiveRequestCount--;
+ }
+ }
+
+ getInflightReceiveRequestCount(): number {
+ return this.#inflightReceiveRequestCount;
+ }
+}
diff --git a/nodejs/src/hook/MessageHookPoints.ts
b/nodejs/src/hook/MessageHookPoints.ts
new file mode 100644
index 00000000..693572ad
--- /dev/null
+++ b/nodejs/src/hook/MessageHookPoints.ts
@@ -0,0 +1,55 @@
+/**
+ * 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.
+ */
+
+/**
+ * Hook points at which {@link MessageInterceptor}s are invoked, mirroring
+ * org.apache.rocketmq.client.java.hook.MessageHookPoints.
+ */
+export enum MessageHookPoints {
+ /**
+ * The hook point of message publishing.
+ */
+ SEND = 'SEND',
+ /**
+ * The hook point of message reception.
+ */
+ RECEIVE = 'RECEIVE',
+ /**
+ * The hook point of message consumption.
+ */
+ CONSUME = 'CONSUME',
+ /**
+ * The hook point of message ack acknowledgement.
+ */
+ ACK = 'ACK',
+ /**
+ * The hook point of message changing invisible duration.
+ */
+ CHANGE_INVISIBLE_DURATION = 'CHANGE_INVISIBLE_DURATION',
+ /**
+ * The hook point of message transaction commit.
+ */
+ COMMIT_TRANSACTION = 'COMMIT_TRANSACTION',
+ /**
+ * The hook point of message transaction rollback.
+ */
+ ROLLBACK_TRANSACTION = 'ROLLBACK_TRANSACTION',
+ /**
+ * The hook point of forwarding message to DLQ.
+ */
+ FORWARD_TO_DLQ = 'FORWARD_TO_DLQ',
+}
diff --git a/nodejs/src/index.ts b/nodejs/src/hook/MessageHookPointsStatus.ts
similarity index 80%
copy from nodejs/src/index.ts
copy to nodejs/src/hook/MessageHookPointsStatus.ts
index 28ec5f3b..71dba35c 100644
--- a/nodejs/src/index.ts
+++ b/nodejs/src/hook/MessageHookPointsStatus.ts
@@ -15,10 +15,11 @@
* limitations under the License.
*/
-export * from './consumer';
-export * from './exception';
-export * from './message';
-export * from './producer';
-export * from './retry';
-export * from './route';
-export * from './client';
+/**
+ * Status of a hook point invocation, mirroring
+ * org.apache.rocketmq.client.java.hook.MessageHookPointsStatus.
+ */
+export enum MessageHookPointsStatus {
+ OK = 'OK',
+ ERROR = 'ERROR',
+}
diff --git a/nodejs/src/consumer/ConsumeTask.ts
b/nodejs/src/hook/MessageInterceptor.ts
similarity index 52%
copy from nodejs/src/consumer/ConsumeTask.ts
copy to nodejs/src/hook/MessageInterceptor.ts
index b502e7f0..92f385a9 100644
--- a/nodejs/src/consumer/ConsumeTask.ts
+++ b/nodejs/src/hook/MessageInterceptor.ts
@@ -15,25 +15,21 @@
* limitations under the License.
*/
-import { MessageView } from '../message';
-import { ConsumeResult } from './ConsumeResult';
-import { MessageListener } from './MessageListener';
+import { MessageView } from '../message/MessageView';
+import { MessageInterceptorContext } from './MessageInterceptorContext';
-export class ConsumeTask {
- readonly #messageListener: MessageListener;
- readonly #messageView: MessageView;
-
- constructor(_clientId: string, messageListener: MessageListener,
messageView: MessageView) {
- this.#messageListener = messageListener;
- this.#messageView = messageView;
- }
+/**
+ * Message passed to interceptors: either an outgoing message or a received
one,
+ * mirroring org.apache.rocketmq.client.java.message.GeneralMessage.
+ */
+export type GeneralMessage = MessageView | { topic: string };
- async call(): Promise<ConsumeResult> {
- try {
- return await this.#messageListener.consume(this.#messageView);
- } catch (e) {
- // Message listener raised an exception while consuming messages
- return ConsumeResult.FAILURE;
- }
- }
+/**
+ * Interface for intercepting ingoing/outgoing message before/after they are
+ * dispatched by client, mirroring
+ * org.apache.rocketmq.client.java.hook.MessageInterceptor.
+ */
+export interface MessageInterceptor {
+ doBefore(context: MessageInterceptorContext, messages: GeneralMessage[]):
void;
+ doAfter(context: MessageInterceptorContext, messages: GeneralMessage[]):
void;
}
diff --git a/nodejs/src/consumer/ConsumeTask.ts
b/nodejs/src/hook/MessageInterceptorContext.ts
similarity index 53%
copy from nodejs/src/consumer/ConsumeTask.ts
copy to nodejs/src/hook/MessageInterceptorContext.ts
index b502e7f0..73bcd820 100644
--- a/nodejs/src/consumer/ConsumeTask.ts
+++ b/nodejs/src/hook/MessageInterceptorContext.ts
@@ -15,25 +15,19 @@
* limitations under the License.
*/
-import { MessageView } from '../message';
-import { ConsumeResult } from './ConsumeResult';
-import { MessageListener } from './MessageListener';
+import { Attribute } from './Attribute';
+import { AttributeKey } from './AttributeKey';
+import { MessageHookPoints } from './MessageHookPoints';
+import { MessageHookPointsStatus } from './MessageHookPointsStatus';
-export class ConsumeTask {
- readonly #messageListener: MessageListener;
- readonly #messageView: MessageView;
-
- constructor(_clientId: string, messageListener: MessageListener,
messageView: MessageView) {
- this.#messageListener = messageListener;
- this.#messageView = messageView;
- }
-
- async call(): Promise<ConsumeResult> {
- try {
- return await this.#messageListener.consume(this.#messageView);
- } catch (e) {
- // Message listener raised an exception while consuming messages
- return ConsumeResult.FAILURE;
- }
- }
+/**
+ * Context carried through the message interceptor chain, mirroring
+ * org.apache.rocketmq.client.java.hook.MessageInterceptorContext.
+ */
+export interface MessageInterceptorContext {
+ getMessageHookPoints(): MessageHookPoints;
+ getStatus(): MessageHookPointsStatus | undefined;
+ putAttribute<T>(key: AttributeKey<T>, value: Attribute<T>): void;
+ getAttribute<T>(key: AttributeKey<T>): Attribute<T> | undefined;
+ getAttributes(): Map<AttributeKey<any>, Attribute<any>>;
}
diff --git a/nodejs/src/hook/MessageInterceptorContextImpl.ts
b/nodejs/src/hook/MessageInterceptorContextImpl.ts
new file mode 100644
index 00000000..250ce105
--- /dev/null
+++ b/nodejs/src/hook/MessageInterceptorContextImpl.ts
@@ -0,0 +1,68 @@
+/**
+ * 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 { Attribute } from './Attribute';
+import { AttributeKey } from './AttributeKey';
+import { MessageHookPoints } from './MessageHookPoints';
+import { MessageHookPointsStatus } from './MessageHookPointsStatus';
+import { MessageInterceptorContext } from './MessageInterceptorContext';
+
+/**
+ * Default {@link MessageInterceptorContext} implementation, mirroring
+ * org.apache.rocketmq.client.java.hook.MessageInterceptorContextImpl.
+ */
+export class MessageInterceptorContextImpl implements
MessageInterceptorContext {
+ readonly #messageHookPoints: MessageHookPoints;
+ readonly #status: MessageHookPointsStatus | undefined;
+ readonly #attributes: Map<AttributeKey<any>, Attribute<any>>;
+
+ constructor(messageHookPoints: MessageHookPoints,
+ status?: MessageHookPointsStatus,
+ attributes?: Map<AttributeKey<any>, Attribute<any>>) {
+ this.#messageHookPoints = messageHookPoints;
+ this.#status = status;
+ this.#attributes = new Map(attributes ?? []);
+ }
+
+ /**
+ * Derive a new context from an existing one with the given status, e.g.
once the
+ * outcome of the intercepted operation is known.
+ */
+ static withStatus(context: MessageInterceptorContext, status:
MessageHookPointsStatus) {
+ return new MessageInterceptorContextImpl(context.getMessageHookPoints(),
status, context.getAttributes());
+ }
+
+ getMessageHookPoints(): MessageHookPoints {
+ return this.#messageHookPoints;
+ }
+
+ getStatus(): MessageHookPointsStatus | undefined {
+ return this.#status;
+ }
+
+ putAttribute<T>(key: AttributeKey<T>, value: Attribute<T>) {
+ this.#attributes.set(key, value);
+ }
+
+ getAttribute<T>(key: AttributeKey<T>): Attribute<T> | undefined {
+ return this.#attributes.get(key) as Attribute<T> | undefined;
+ }
+
+ getAttributes(): Map<AttributeKey<any>, Attribute<any>> {
+ return this.#attributes;
+ }
+}
diff --git a/nodejs/src/hook/MessageMeterInterceptor.ts
b/nodejs/src/hook/MessageMeterInterceptor.ts
new file mode 100644
index 00000000..42a59259
--- /dev/null
+++ b/nodejs/src/hook/MessageMeterInterceptor.ts
@@ -0,0 +1,162 @@
+/**
+ * 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 } from '@opentelemetry/api';
+import { Attribute, AttributeKey, MessageHookPoints, MessageHookPointsStatus,
MessageInterceptor, MessageInterceptorContext } from './';
+import { GeneralMessage } from './MessageInterceptor';
+import { MetricLabels } from '../metrics/MetricLabels';
+import { InvocationStatus } from '../metrics/InvocationStatus';
+import { HistogramEnum } from '../metrics/HistogramEnum';
+import { ClientMeterManager } from '../metrics/ClientMeterManager';
+
+/**
+ * Minimal message shape required to compute metric attributes / latencies.
+ * Covers both the producer's PublishingMessage (cast to GeneralMessage) and
the
+ * consumer's MessageView (which implements GeneralMessage).
+ */
+type MetricMessage = {
+ topic: string;
+ transportDeliveryTimestamp?: Date;
+ decodeTimestamp?: Date;
+};
+
+/**
+ * Records the four RocketMQ histograms around the SEND / RECEIVE / CONSUME
hook
+ * points, mirroring
org.apache.rocketmq.client.java.metrics.MessageMeterInterceptor.
+ *
+ * Unlike Java (which wires an AuthInterceptor + IpNameResolverFactory onto the
+ * export channel), the Node client signs each export request via a per-call
+ * metadata provider (see {@link ClientMeterManager}), so no auth plumbing is
+ * needed here.
+ */
+export class MessageMeterInterceptor implements MessageInterceptor {
+ static readonly SEND_STOPWATCH_KEY =
AttributeKey.create<number>('send_stopwatch');
+ static readonly CONSUME_STOPWATCH_KEY =
AttributeKey.create<number>('consume_stopwatch');
+
+ constructor(
+ private readonly meterManager: ClientMeterManager,
+ private readonly clientId: string,
+ private readonly getConsumerGroup: () => string | undefined,
+ ) {
+ }
+
+ doBefore(context: MessageInterceptorContext, messages: GeneralMessage[]):
void {
+ if (!this.meterManager.isEnabled()) {
+ return;
+ }
+ switch (context.getMessageHookPoints()) {
+ case MessageHookPoints.SEND:
+ context.putAttribute(MessageMeterInterceptor.SEND_STOPWATCH_KEY,
Attribute.create(Date.now()));
+ break;
+ case MessageHookPoints.CONSUME: {
+ const consumerGroup = this.getConsumerGroup();
+ if (!consumerGroup) {
+ break;
+ }
+ const message = messages[0] as MetricMessage | undefined;
+ if (!message || !message.decodeTimestamp) {
+ break;
+ }
+ const latency = Date.now() - message.decodeTimestamp.getTime();
+ const attributes = this.#consumerAttributes(message.topic,
consumerGroup);
+ this.meterManager.record(HistogramEnum.AWAIT_TIME, attributes,
latency);
+ context.putAttribute(MessageMeterInterceptor.CONSUME_STOPWATCH_KEY,
Attribute.create(Date.now()));
+ break;
+ }
+ default:
+ break;
+ }
+ }
+
+ doAfter(context: MessageInterceptorContext, messages: GeneralMessage[]):
void {
+ if (!this.meterManager.isEnabled()) {
+ return;
+ }
+ const status = context.getStatus();
+ const invocationStatus = status === MessageHookPointsStatus.OK ?
InvocationStatus.SUCCESS : InvocationStatus.FAILURE;
+ switch (context.getMessageHookPoints()) {
+ case MessageHookPoints.SEND: {
+ const attr =
context.getAttribute(MessageMeterInterceptor.SEND_STOPWATCH_KEY);
+ if (!attr) {
+ break;
+ }
+ const message = messages[0] as MetricMessage | undefined;
+ if (!message) {
+ break;
+ }
+ const cost = Date.now() - attr.get();
+ const attributes: Attributes = {
+ [MetricLabels.TOPIC]: message.topic,
+ [MetricLabels.CLIENT_ID]: this.clientId,
+ [MetricLabels.INVOCATION_STATUS]: invocationStatus.name,
+ };
+ this.meterManager.record(HistogramEnum.SEND_COST_TIME, attributes,
cost);
+ break;
+ }
+ case MessageHookPoints.RECEIVE: {
+ const consumerGroup = this.getConsumerGroup();
+ if (!consumerGroup) {
+ break;
+ }
+ const message = messages[0] as MetricMessage | undefined;
+ if (!message || !message.transportDeliveryTimestamp) {
+ break;
+ }
+ const latency = Date.now() -
message.transportDeliveryTimestamp.getTime();
+ if (latency < 0) {
+ break;
+ }
+ const attributes = this.#consumerAttributes(message.topic,
consumerGroup);
+ this.meterManager.record(HistogramEnum.DELIVERY_LATENCY, attributes,
latency);
+ break;
+ }
+ case MessageHookPoints.CONSUME: {
+ const consumerGroup = this.getConsumerGroup();
+ if (!consumerGroup) {
+ break;
+ }
+ const attr =
context.getAttribute(MessageMeterInterceptor.CONSUME_STOPWATCH_KEY);
+ if (!attr) {
+ break;
+ }
+ const message = messages[0] as MetricMessage | undefined;
+ if (!message) {
+ break;
+ }
+ const cost = Date.now() - attr.get();
+ const attributes: Attributes = {
+ [MetricLabels.TOPIC]: message.topic,
+ [MetricLabels.CLIENT_ID]: this.clientId,
+ [MetricLabels.CONSUMER_GROUP]: consumerGroup,
+ [MetricLabels.INVOCATION_STATUS]: invocationStatus.name,
+ };
+ this.meterManager.record(HistogramEnum.PROCESS_TIME, attributes, cost);
+ break;
+ }
+ default:
+ break;
+ }
+ }
+
+ #consumerAttributes(topic: string, consumerGroup: string): Attributes {
+ return {
+ [MetricLabels.TOPIC]: topic,
+ [MetricLabels.CLIENT_ID]: this.clientId,
+ [MetricLabels.CONSUMER_GROUP]: consumerGroup,
+ };
+ }
+}
diff --git a/nodejs/src/index.ts b/nodejs/src/hook/index.ts
similarity index 65%
copy from nodejs/src/index.ts
copy to nodejs/src/hook/index.ts
index 28ec5f3b..225a5e15 100644
--- a/nodejs/src/index.ts
+++ b/nodejs/src/hook/index.ts
@@ -15,10 +15,13 @@
* limitations under the License.
*/
-export * from './consumer';
-export * from './exception';
-export * from './message';
-export * from './producer';
-export * from './retry';
-export * from './route';
-export * from './client';
+export * from './Attribute';
+export * from './AttributeKey';
+export * from './CompositedMessageInterceptor';
+export * from './InflightRequestCountInterceptor';
+export * from './MessageHookPoints';
+export * from './MessageHookPointsStatus';
+export * from './MessageInterceptor';
+export * from './MessageInterceptorContext';
+export * from './MessageInterceptorContextImpl';
+export * from './MessageMeterInterceptor';
diff --git a/nodejs/src/index.ts b/nodejs/src/index.ts
index 28ec5f3b..32e0b114 100644
--- a/nodejs/src/index.ts
+++ b/nodejs/src/index.ts
@@ -17,6 +17,7 @@
export * from './consumer';
export * from './exception';
+export * from './hook';
export * from './message';
export * from './producer';
export * from './retry';
diff --git a/nodejs/src/metrics/ClientMeter.ts
b/nodejs/src/metrics/ClientMeter.ts
new file mode 100644
index 00000000..3645729d
--- /dev/null
+++ b/nodejs/src/metrics/ClientMeter.ts
@@ -0,0 +1,81 @@
+/**
+ * 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, Histogram, Meter } from '@opentelemetry/api';
+import { MeterProvider as SdkMeterProvider } from '@opentelemetry/sdk-metrics';
+import { Endpoints } from '../route';
+import { HistogramEnum } from './HistogramEnum';
+import { Metric } from './Metric';
+
+/**
+ * Wraps an OpenTelemetry {@link Meter} and records the four RocketMQ
histograms.
+ * When metrics are disabled it is a no-op instance, mirroring
+ * org.apache.rocketmq.client.java.metrics.ClientMeter.
+ */
+export class ClientMeter {
+ readonly enabled: boolean;
+ private readonly meter?: Meter;
+ private readonly endpoints?: Endpoints;
+ private readonly provider?: SdkMeterProvider;
+ private readonly histogramMap = new Map<HistogramEnum, Histogram>();
+
+ constructor(meter: Meter | null, endpoints: Endpoints | null, provider:
SdkMeterProvider | null) {
+ this.enabled = meter != null && endpoints != null && provider != null;
+ this.meter = meter ?? undefined;
+ this.endpoints = endpoints ?? undefined;
+ this.provider = provider ?? undefined;
+ }
+
+ static disabledInstance(): ClientMeter {
+ return new ClientMeter(null, null, null);
+ }
+
+ record(histogramEnum: HistogramEnum, attributes: Attributes, value: number) {
+ if (!this.enabled || !this.meter) {
+ return;
+ }
+ let histogram = this.histogramMap.get(histogramEnum);
+ if (!histogram) {
+ histogram = this.meter.createHistogram(histogramEnum);
+ this.histogramMap.set(histogramEnum, histogram);
+ }
+ histogram.record(value, attributes);
+ }
+
+ async shutdown() {
+ if (!this.enabled || !this.provider) {
+ return;
+ }
+ try {
+ await this.provider.shutdown();
+ } catch (t) {
+ // best-effort shutdown, ignore
+ }
+ }
+
+ /**
+ * Returns true if the supplied metric switch is already satisfied by this
+ * meter (so a rebuild is unnecessary), mirroring ClientMeter#satisfy.
+ */
+ satisfy(metric: Metric): boolean {
+ if (this.enabled && metric.on && this.endpoints != null &&
metric.endpoints != null
+ && this.endpoints.equals(metric.endpoints)) {
+ return true;
+ }
+ return !this.enabled && !metric.on;
+ }
+}
diff --git a/nodejs/src/metrics/ClientMeterManager.ts
b/nodejs/src/metrics/ClientMeterManager.ts
new file mode 100644
index 00000000..51c5231f
--- /dev/null
+++ b/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;
+ // Use the first address as the OTLP/gRPC target. The exporter's URL is
+ // normalized to "host:port" internally, which gRPC-js resolves through
+ // its default DNS resolver (a single IP works fine; the custom `ip`
+ // resolver is used for the primary RPC channels).
+ const address = endpoints.addressesList[0];
+ const url = `http://${address.host}:${address.port}`;
+ const exporter = new OTLPMetricExporter({
+ url,
+ credentials: ChannelCredentials.createInsecure(),
+ timeoutMillis: METRIC_EXPORTER_RPC_TIMEOUT,
+ // Per-request signing mirrors the Java client's AuthInterceptor. The
+ // d.ts types this field as `Metadata`, but the runtime (and the
+ // exporter's own merge logic in otlp-grpc-exporter-base) expects a
+ // function returning freshly-signed Metadata.
+ metadata: (() => this.metadataProvider()) as unknown as Metadata,
+ });
+
+ const reader = new PeriodicExportingMetricReader({
+ exporter,
+ exportIntervalMillis: METRIC_READER_INTERVAL,
+ exportTimeoutMillis: METRIC_EXPORTER_RPC_TIMEOUT,
+ });
+
+ const views: View[] = [
+ new View(buildHistogramView(HistogramEnum.SEND_COST_TIME)),
+ new View(buildHistogramView(HistogramEnum.DELIVERY_LATENCY)),
+ new View(buildHistogramView(HistogramEnum.AWAIT_TIME)),
+ new View(buildHistogramView(HistogramEnum.PROCESS_TIME)),
+ ];
+
+ const provider: SdkMeterProvider = new SdkMeterProvider({
+ resource: Resource.empty(),
+ readers: [ reader ],
+ views,
+ });
+
+ const meter: Meter = provider.getMeter(METRIC_INSTRUMENTATION_NAME);
+
+ const existedClientMeter = this.clientMeter;
+ this.clientMeter = new ClientMeter(meter, endpoints, provider);
+ await existedClientMeter.shutdown();
+ this.logger.info('Metrics is on, endpoints=%s, clientId=%s', endpoints,
this.clientId);
+
+ const gauges = this.gaugeObserver.getGauges();
+ for (const gauge of gauges) {
+ meter.createObservableGauge(gauge).addCallback(observableResult => {
+ const values = this.gaugeObserver.getValues(gauge);
+ for (const [ attributes, value ] of values) {
+ observableResult.observe(value, attributes);
+ }
+ });
+ }
+ } catch (t) {
+ this.logger.error('Exception raised when resetting message meter,
clientId=%s, error=%s', this.clientId, t);
+ }
+ }
+}
diff --git a/nodejs/src/index.ts b/nodejs/src/metrics/EmptyGaugeObserver.ts
similarity index 59%
copy from nodejs/src/index.ts
copy to nodejs/src/metrics/EmptyGaugeObserver.ts
index 28ec5f3b..f9b4ce6b 100644
--- a/nodejs/src/index.ts
+++ b/nodejs/src/metrics/EmptyGaugeObserver.ts
@@ -15,10 +15,22 @@
* limitations under the License.
*/
-export * from './consumer';
-export * from './exception';
-export * from './message';
-export * from './producer';
-export * from './retry';
-export * from './route';
-export * from './client';
+import { Attributes } from '@opentelemetry/api';
+import { GaugeEnum } from './GaugeEnum';
+import { GaugeObserver } from './GaugeObserver';
+
+/**
+ * Default no-op gauge observer used before a consumer registers a real one,
+ * mirroring org.apache.rocketmq.client.java.metrics.EmptyGaugeObserver.
+ */
+export class EmptyGaugeObserver implements GaugeObserver {
+ static readonly EMPTY: EmptyGaugeObserver = new EmptyGaugeObserver();
+
+ getGauges(): GaugeEnum[] {
+ return [];
+ }
+
+ getValues(): Map<Attributes, number> {
+ return new Map();
+ }
+}
diff --git a/nodejs/src/index.ts b/nodejs/src/metrics/GaugeEnum.ts
similarity index 62%
copy from nodejs/src/index.ts
copy to nodejs/src/metrics/GaugeEnum.ts
index 28ec5f3b..ee3c757f 100644
--- a/nodejs/src/index.ts
+++ b/nodejs/src/metrics/GaugeEnum.ts
@@ -15,10 +15,19 @@
* limitations under the License.
*/
-export * from './consumer';
-export * from './exception';
-export * from './message';
-export * from './producer';
-export * from './retry';
-export * from './route';
-export * from './client';
+/**
+ * The two consumer gauges exported by the RocketMQ client, mirroring
+ * org.apache.rocketmq.client.java.metrics.GaugeEnum.
+ */
+export enum GaugeEnum {
+ /**
+ * Cached message count of push consumer.
+ * Labels: topic, client_id, consumer_group.
+ */
+ CONSUMER_CACHED_MESSAGES = 'rocketmq_consumer_cached_messages',
+ /**
+ * Cached message bytes of push consumer.
+ * Labels: topic, client_id, consumer_group.
+ */
+ CONSUMER_CACHED_BYTES = 'rocketmq_consumer_cached_bytes',
+}
diff --git a/nodejs/src/index.ts b/nodejs/src/metrics/GaugeObserver.ts
similarity index 61%
copy from nodejs/src/index.ts
copy to nodejs/src/metrics/GaugeObserver.ts
index 28ec5f3b..8ad2ea4e 100644
--- a/nodejs/src/index.ts
+++ b/nodejs/src/metrics/GaugeObserver.ts
@@ -15,10 +15,18 @@
* limitations under the License.
*/
-export * from './consumer';
-export * from './exception';
-export * from './message';
-export * from './producer';
-export * from './retry';
-export * from './route';
-export * from './client';
+import { Attributes } from '@opentelemetry/api';
+import { GaugeEnum } from './GaugeEnum';
+
+/**
+ * Supplies the live values observed by the periodic metric reader for each
+ * consumer gauge, mirroring
org.apache.rocketmq.client.java.metrics.GaugeObserver.
+ *
+ * The map key is the attribute set (topic / client_id / consumer_group) and
the
+ * value is the observed gauge reading.
+ */
+export interface GaugeObserver {
+ getGauges(): GaugeEnum[];
+
+ getValues(gauge: GaugeEnum): Map<Attributes, number>;
+}
diff --git a/nodejs/src/metrics/HistogramEnum.ts
b/nodejs/src/metrics/HistogramEnum.ts
new file mode 100644
index 00000000..c5de21bb
--- /dev/null
+++ b/nodejs/src/metrics/HistogramEnum.ts
@@ -0,0 +1,69 @@
+/**
+ * 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 { ExplicitBucketHistogramAggregation, InstrumentType, ViewOptions }
from '@opentelemetry/sdk-metrics';
+
+/**
+ * The four histograms exported by the RocketMQ client, mirroring
+ * org.apache.rocketmq.client.java.metrics.HistogramEnum. The bucket
+ * boundaries (in milliseconds) are copied verbatim from the Java client.
+ */
+export enum HistogramEnum {
+ /**
+ * Cost time of successful api calls of message publishing.
+ * Labels: topic, client_id, invocation_status.
+ */
+ SEND_COST_TIME = 'rocketmq_send_cost_time',
+ /**
+ * Latency of message delivery from remote.
+ * Labels: topic, client_id, consumer_group.
+ */
+ DELIVERY_LATENCY = 'rocketmq_delivery_latency',
+ /**
+ * Await time of message consumption.
+ * Labels: topic, client_id, consumer_group.
+ */
+ AWAIT_TIME = 'rocketmq_await_time',
+ /**
+ * Process time of message consumption.
+ * Labels: topic, client_id, consumer_group, invocation_status.
+ */
+ PROCESS_TIME = 'rocketmq_process_time',
+}
+
+/**
+ * Explicit bucket boundaries (milliseconds) for each histogram, identical to
+ * the Java client's Aggregation.explicitBucketHistogram(...) calls.
+ */
+export const HISTOGRAM_BUCKETS: Record<HistogramEnum, number[]> = {
+ [HistogramEnum.SEND_COST_TIME]: [ 1.0, 5.0, 10.0, 20.0, 50.0, 200.0, 500.0 ],
+ [HistogramEnum.DELIVERY_LATENCY]: [ 1.0, 5.0, 10.0, 20.0, 50.0, 200.0, 500.0
],
+ [HistogramEnum.AWAIT_TIME]: [ 1.0, 5.0, 20.0, 100.0, 1000.0, 5 * 1000.0, 10
* 1000.0 ],
+ [HistogramEnum.PROCESS_TIME]: [ 1.0, 5.0, 10.0, 100.0, 1000.0, 10 * 1000.0,
60 * 1000.0 ],
+};
+
+/**
+ * Build the OpenTelemetry view that forces the given histogram onto the exact
+ * explicit-bucket aggregation used by the Java client.
+ */
+export function buildHistogramView(histogramEnum: HistogramEnum): ViewOptions {
+ return {
+ instrumentName: histogramEnum,
+ instrumentType: InstrumentType.HISTOGRAM,
+ aggregation: new
ExplicitBucketHistogramAggregation(HISTOGRAM_BUCKETS[histogramEnum]),
+ };
+}
diff --git a/nodejs/src/index.ts b/nodejs/src/metrics/InvocationStatus.ts
similarity index 69%
copy from nodejs/src/index.ts
copy to nodejs/src/metrics/InvocationStatus.ts
index 28ec5f3b..f57ca1a2 100644
--- a/nodejs/src/index.ts
+++ b/nodejs/src/metrics/InvocationStatus.ts
@@ -15,10 +15,14 @@
* limitations under the License.
*/
-export * from './consumer';
-export * from './exception';
-export * from './message';
-export * from './producer';
-export * from './retry';
-export * from './route';
-export * from './client';
+/**
+ * Outcome of an instrumented operation (send / consume), mirroring
+ * org.apache.rocketmq.client.java.metrics.InvocationStatus.
+ */
+export class InvocationStatus {
+ static readonly SUCCESS = new InvocationStatus('success');
+ static readonly FAILURE = new InvocationStatus('failure');
+
+ private constructor(readonly name: string) {
+ }
+}
diff --git a/nodejs/src/metrics/Metric.ts b/nodejs/src/metrics/Metric.ts
new file mode 100644
index 00000000..8eadb580
--- /dev/null
+++ b/nodejs/src/metrics/Metric.ts
@@ -0,0 +1,42 @@
+/**
+ * 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 { Endpoints } from '../route';
+import { Metric as MetricPB } from
'../../proto/apache/rocketmq/v2/definition_pb';
+
+/**
+ * Local view of the metric switch received from the broker via the telemetry
+ * settings command, mirroring org.apache.rocketmq.client.java.metrics.Metric.
+ *
+ * Metrics are only considered enabled when both the `on` flag is set AND a
+ * valid endpoints is present.
+ */
+export class Metric {
+ readonly endpoints: Endpoints | null;
+ readonly on: boolean;
+
+ constructor(metric: MetricPB | undefined | null) {
+ if (metric && metric.hasEndpoints() && metric.getOn()) {
+ // Endpoints expects the plain AsObject shape, so unwrap the proto
message.
+ this.endpoints = new Endpoints(metric.getEndpoints()!.toObject());
+ this.on = true;
+ } else {
+ this.endpoints = null;
+ this.on = false;
+ }
+ }
+}
diff --git a/nodejs/src/index.ts b/nodejs/src/metrics/MetricLabels.ts
similarity index 65%
copy from nodejs/src/index.ts
copy to nodejs/src/metrics/MetricLabels.ts
index 28ec5f3b..44480a1e 100644
--- a/nodejs/src/index.ts
+++ b/nodejs/src/metrics/MetricLabels.ts
@@ -15,10 +15,17 @@
* limitations under the License.
*/
-export * from './consumer';
-export * from './exception';
-export * from './message';
-export * from './producer';
-export * from './retry';
-export * from './route';
-export * from './client';
+/**
+ * Canonical attribute (label) keys attached to every RocketMQ metric,
+ * mirroring org.apache.rocketmq.client.java.metrics.MetricLabels.
+ */
+export class MetricLabels {
+ static readonly TOPIC = 'topic';
+ static readonly CLIENT_ID = 'client_id';
+ static readonly CONSUMER_GROUP = 'consumer_group';
+ static readonly INVOCATION_STATUS = 'invocation_status';
+
+ private constructor() {
+ // constant holder
+ }
+}
diff --git a/nodejs/src/index.ts b/nodejs/src/metrics/index.ts
similarity index 73%
copy from nodejs/src/index.ts
copy to nodejs/src/metrics/index.ts
index 28ec5f3b..a39d22e2 100644
--- a/nodejs/src/index.ts
+++ b/nodejs/src/metrics/index.ts
@@ -15,10 +15,12 @@
* limitations under the License.
*/
-export * from './consumer';
-export * from './exception';
-export * from './message';
-export * from './producer';
-export * from './retry';
-export * from './route';
-export * from './client';
+export * from './MetricLabels';
+export * from './InvocationStatus';
+export * from './HistogramEnum';
+export * from './GaugeEnum';
+export * from './GaugeObserver';
+export * from './EmptyGaugeObserver';
+export * from './Metric';
+export * from './ClientMeter';
+export * from './ClientMeterManager';
diff --git a/nodejs/src/producer/Producer.ts b/nodejs/src/producer/Producer.ts
index d582d703..04a9bd4e 100644
--- a/nodejs/src/producer/Producer.ts
+++ b/nodejs/src/producer/Producer.ts
@@ -50,6 +50,12 @@ import { Transaction } from './Transaction';
import { createResource } from '../util';
import { RecallReceipt } from './RecallReceipt';
import { RecallMessageRequest } from
'../../proto/apache/rocketmq/v2/service_pb';
+import {
+ GeneralMessage,
+ MessageHookPoints,
+ MessageHookPointsStatus,
+ MessageInterceptorContextImpl,
+} from '../hook';
export interface ProducerOptions extends BaseClientOptions {
topic?: string | string[];
@@ -114,11 +120,25 @@ export class Producer extends BaseClient {
.setTopic(createResource(message.topic).setResourceNamespace(this.namespace))
.setResolution(resolution)
.setSource(source);
- const response = await this.rpcClientManager.endTransaction(endpoints,
request, this.requestTimeout);
- StatusChecker.check(response.getStatus()?.toObject());
-
- this.logger.debug?.('End transaction successfully, messageId=%s,
transactionId=%s, resolution=%s, source=%s, clientId=%s',
- messageId, transactionId, resolutionStr, sourceStr, this.clientId);
+ // COMMIT_TRANSACTION / ROLLBACK_TRANSACTION hook point, mirroring Java
ProducerImpl#endTransaction.
+ const messageHookPoints = resolution === TransactionResolution.COMMIT ?
+ MessageHookPoints.COMMIT_TRANSACTION :
MessageHookPoints.ROLLBACK_TRANSACTION;
+ const context = new MessageInterceptorContextImpl(messageHookPoints);
+ const generalMessages: GeneralMessage[] = [ message ];
+ this.doBefore(context, generalMessages);
+ try {
+ const response = await this.rpcClientManager.endTransaction(endpoints,
request, this.requestTimeout);
+ const statusObj = response.getStatus()?.toObject();
+ const hookStatus = statusObj?.code === Code.OK ?
MessageHookPointsStatus.OK : MessageHookPointsStatus.ERROR;
+ StatusChecker.check(statusObj);
+ this.doAfter(MessageInterceptorContextImpl.withStatus(context,
hookStatus), generalMessages);
+
+ this.logger.debug?.('End transaction successfully, messageId=%s,
transactionId=%s, resolution=%s, source=%s, clientId=%s',
+ messageId, transactionId, resolutionStr, sourceStr, this.clientId);
+ } catch (err) {
+ this.doAfter(MessageInterceptorContextImpl.withStatus(context,
MessageHookPointsStatus.ERROR), generalMessages);
+ throw err;
+ }
}
async onRecoverOrphanedTransactionCommand(endpoints: Endpoints, command:
RecoverOrphanedTransactionCommand) {
@@ -241,7 +261,18 @@ export class Producer extends BaseClient {
// Prepare the candidate message queue(s) for retry-sending in advance.
const candidates = messageGroup ? [
loadBalancer.takeMessageQueueByMessageGroup(messageGroup) ] :
this.#takeMessageQueues(loadBalancer);
- return await this.#send0(topic, messageType, candidates, pubMessages, 1);
+ // SEND hook point, mirroring Java ProducerImpl#send.
+ const context = new MessageInterceptorContextImpl(MessageHookPoints.SEND);
+ const generalMessages = pubMessages as unknown as GeneralMessage[];
+ this.doBefore(context, generalMessages);
+ try {
+ const receipts = await this.#send0(topic, messageType, candidates,
pubMessages, 1);
+ this.doAfter(MessageInterceptorContextImpl.withStatus(context,
MessageHookPointsStatus.OK), generalMessages);
+ return receipts;
+ } catch (err) {
+ this.doAfter(MessageInterceptorContextImpl.withStatus(context,
MessageHookPointsStatus.ERROR), generalMessages);
+ throw err;
+ }
}
#wrapSendMessageRequest(pubMessages: PublishingMessage[], mq: MessageQueue) {
diff --git a/nodejs/test/hook.test.ts b/nodejs/test/hook.test.ts
new file mode 100644
index 00000000..c6e72039
--- /dev/null
+++ b/nodejs/test/hook.test.ts
@@ -0,0 +1,276 @@
+/**
+ * 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.
+ */
+
+/**
+ * Tests for the message interceptor / hook framework and the
+ * NOTIFY_UNSUBSCRIBE_LITE_COMMAND telemetry dispatch:
+ * - Telemetry command dispatch vs Java (J-1)
+ * - Message interceptor chain (J-3)
+ * - Transaction hook points (K-2)
+ * - FORWARD_TO_DLQ hook point (K-3)
+ */
+
+import { describe, it } from 'node:test';
+import * as assert from 'node:assert';
+import { Endpoints } from '../src/route';
+import { Producer } from '../src/producer';
+import { Consumer } from '../src/consumer';
+import { Message } from '../src/message';
+import { Settings } from '../src/client';
+import {
+ Attribute,
+ AttributeKey,
+ CompositedMessageInterceptor,
+ InflightRequestCountInterceptor,
+ MessageHookPoints,
+ MessageHookPointsStatus,
+ MessageInterceptor,
+ MessageInterceptorContextImpl,
+} from '../src/hook';
+import { Code, TransactionResolution } from
'../proto/apache/rocketmq/v2/definition_pb';
+import { Status } from '../proto/apache/rocketmq/v2/definition_pb';
+import {
+ HeartbeatRequest,
+ NotifyClientTerminationRequest,
+} from '../proto/apache/rocketmq/v2/service_pb';
+
+function statusOf(code: Code): Status.AsObject {
+ return { code, message: 'mock message', requestId: '' } as unknown as
Status.AsObject;
+}
+
+describe('Telemetry command dispatch vs Java (J-1)', () => {
+ it('should dispatch NOTIFY_UNSUBSCRIBE_LITE_COMMAND to the client', () => {
+ const { EventEmitter } = require('node:events');
+ const { TelemetryCommand, NotifyUnsubscribeLiteCommand } =
require('../proto/apache/rocketmq/v2/service_pb');
+ const { TelemetrySession } = require('../src/client/TelemetrySession');
+
+ const received: Array<{ liteTopic: string; facade: string }> = [];
+ const stream = new EventEmitter();
+ stream.write = () => true;
+ stream.end = () => undefined;
+ stream.removeAllListeners = () => stream;
+
+ const fakeClient = {
+ clientId: 'fake-client-id',
+ createTelemetryStream: () => stream,
+ onNotifyUnsubscribeLiteCommand(endpoints: any, command: any) {
+ received.push({ liteTopic: command.getLiteTopic(), facade:
endpoints.facade });
+ },
+ };
+ new TelemetrySession(fakeClient, new Endpoints('127.0.0.1:8081'),
+ { info() {}, warn() {}, error() {}, debug() {} });
+
+ const telemetryCommand = new TelemetryCommand();
+ telemetryCommand.setNotifyUnsubscribeLiteCommand(new
NotifyUnsubscribeLiteCommand().setLiteTopic('lite-topic-1'));
+ stream.emit('data', telemetryCommand);
+
+ assert.strictEqual(received.length, 1);
+ assert.strictEqual(received[0].liteTopic, 'lite-topic-1');
+ assert.strictEqual(received[0].facade, '127.0.0.1:8081');
+ });
+});
+
+describe('Transaction hook points vs Java (K-2)', () => {
+ function createProducer(events: string[]) {
+ const producer = new Producer({
+ endpoints: '127.0.0.1:8081',
+ namespace: '',
+ topics: [ 'TopicTestForTransaction' ],
+ messageInterceptor: {
+ doBefore: context => {
+
events.push(`before:${MessageHookPoints[context.getMessageHookPoints()]}`);
+ },
+ doAfter: context => {
+
events.push(`after:${MessageHookPoints[context.getMessageHookPoints()]}:${MessageHookPointsStatus[context.getStatus()]}`);
+ },
+ },
+ } as any);
+ return producer;
+ }
+
+ function stubEndTransaction(producer: Producer, status: Status.AsObject) {
+ (producer as any).rpcClientManager = {
+ endTransaction: async () => ({
+ getStatus: () => ({
+ getCode: () => status.code,
+ toObject: () => status,
+ }),
+ }),
+ };
+ }
+
+ it('should trigger COMMIT_TRANSACTION on commit and ROLLBACK_TRANSACTION on
rollback', async () => {
+ const commitEvents: string[] = [];
+ const producer = createProducer(commitEvents);
+ stubEndTransaction(producer, statusOf(Code.OK));
+ await producer.endTransaction(new Endpoints('127.0.0.1:8081'),
+ new Message({ topic: 'TopicTestForTransaction', body:
Buffer.from('body') }),
+ 'message-id', 'transaction-id', TransactionResolution.COMMIT);
+ assert.deepStrictEqual(commitEvents, [
+ 'before:COMMIT_TRANSACTION', 'after:COMMIT_TRANSACTION:OK',
+ ]);
+
+ const rollbackEvents: string[] = [];
+ const rollbackProducer = createProducer(rollbackEvents);
+ stubEndTransaction(rollbackProducer, statusOf(Code.OK));
+ await rollbackProducer.endTransaction(new Endpoints('127.0.0.1:8081'),
+ new Message({ topic: 'TopicTestForTransaction', body:
Buffer.from('body') }),
+ 'message-id', 'transaction-id', TransactionResolution.ROLLBACK);
+ assert.deepStrictEqual(rollbackEvents, [
+ 'before:ROLLBACK_TRANSACTION', 'after:ROLLBACK_TRANSACTION:OK',
+ ]);
+ });
+
+ it('should mark the transaction hook as ERROR when the RPC fails', async ()
=> {
+ const events: string[] = [];
+ const producer = createProducer(events);
+ (producer as any).rpcClientManager = {
+ endTransaction: async () => {
+ throw new Error('end transaction failed');
+ },
+ };
+ await assert.rejects(async () => {
+ await producer.endTransaction(new Endpoints('127.0.0.1:8081'),
+ new Message({ topic: 'TopicTestForTransaction', body:
Buffer.from('body') }),
+ 'message-id', 'transaction-id', TransactionResolution.COMMIT);
+ });
+ assert.deepStrictEqual(events, [
+ 'before:COMMIT_TRANSACTION', 'after:COMMIT_TRANSACTION:ERROR',
+ ]);
+ });
+});
+
+describe('FORWARD_TO_DLQ hook point vs Java (K-3)', () => {
+ class HookTestConsumer extends Consumer {
+ protected getSettings(): Settings {
+ return {} as Settings;
+ }
+ protected wrapHeartbeatRequest(): HeartbeatRequest {
+ return new HeartbeatRequest();
+ }
+ protected wrapNotifyClientTerminationRequest():
NotifyClientTerminationRequest {
+ return new NotifyClientTerminationRequest();
+ }
+ }
+
+ function createConsumer(events: string[]) {
+ return new HookTestConsumer({
+ endpoints: '127.0.0.1:8081',
+ namespace: '',
+ consumerGroup: 'TestGroup',
+ messageInterceptor: {
+ doBefore: context => {
+
events.push(`before:${MessageHookPoints[context.getMessageHookPoints()]}`);
+ },
+ doAfter: context => {
+
events.push(`after:${MessageHookPoints[context.getMessageHookPoints()]}:${MessageHookPointsStatus[context.getStatus()]}`);
+ },
+ },
+ } as any);
+ }
+
+ it('should trigger FORWARD_TO_DLQ with the status carried out of the
response', async () => {
+ const okEvents: string[] = [];
+ const okConsumer = createConsumer(okEvents);
+ (okConsumer as any).rpcClientManager = {
+ forwardMessageToDeadLetterQueue: async () => ({ getStatus: () => ({
getCode: () => Code.OK }) }),
+ };
+ await okConsumer.forwardMessageToDeadLetterQueueViaRpc(new
Endpoints('127.0.0.1:8081'), {}, 3000,
+ { topic: 'TopicTest' } as any);
+ assert.deepStrictEqual(okEvents, [ 'before:FORWARD_TO_DLQ',
'after:FORWARD_TO_DLQ:OK' ]);
+
+ const errEvents: string[] = [];
+ const errConsumer = createConsumer(errEvents);
+ (errConsumer as any).rpcClientManager = {
+ forwardMessageToDeadLetterQueue: async () => ({ getStatus: () => ({
getCode: () => Code.INTERNAL_ERROR }) }),
+ };
+ await errConsumer.forwardMessageToDeadLetterQueueViaRpc(new
Endpoints('127.0.0.1:8081'), {}, 3000,
+ { topic: 'TopicTest' } as any);
+ assert.deepStrictEqual(errEvents, [ 'before:FORWARD_TO_DLQ',
'after:FORWARD_TO_DLQ:ERROR' ]);
+ });
+
+ it('should mark FORWARD_TO_DLQ as ERROR when the RPC throws', async () => {
+ const events: string[] = [];
+ const consumer = createConsumer(events);
+ (consumer as any).rpcClientManager = {
+ forwardMessageToDeadLetterQueue: async () => {
+ throw new Error('forward failed');
+ },
+ };
+ await assert.rejects(async () => {
+ await consumer.forwardMessageToDeadLetterQueueViaRpc(new
Endpoints('127.0.0.1:8081'), {}, 3000,
+ { topic: 'TopicTest' } as any);
+ });
+ assert.deepStrictEqual(events, [ 'before:FORWARD_TO_DLQ',
'after:FORWARD_TO_DLQ:ERROR' ]);
+ });
+});
+
+describe('Message interceptor chain (J-3)', () => {
+ it('should run doBefore in order and doAfter in reverse order', () => {
+ const calls: string[] = [];
+ const mk = (name: string): MessageInterceptor => ({
+ doBefore: () => { calls.push('before-' + name); },
+ doAfter: () => { calls.push('after-' + name); },
+ });
+ const composited = new CompositedMessageInterceptor([ mk('a'), mk('b'),
mk('c') ]);
+ const context = new MessageInterceptorContextImpl(MessageHookPoints.SEND);
+ composited.doBefore(context, []);
+ composited.doAfter(MessageInterceptorContextImpl.withStatus(context,
MessageHookPointsStatus.OK), []);
+ assert.deepStrictEqual(calls, [
+ 'before-a', 'before-b', 'before-c', 'after-c', 'after-b', 'after-a',
+ ]);
+ });
+
+ it('should not break the pipeline when an interceptor throws', () => {
+ const calls: string[] = [];
+ const composited = new CompositedMessageInterceptor([
+ { doBefore: () => { calls.push('before-throwing'); }, doAfter: () => {
calls.push('after-throwing'); } },
+ { doBefore: () => { throw new Error('boom'); }, doAfter: () => { throw
new Error('boom'); } },
+ { doBefore: () => { calls.push('before-ok'); }, doAfter: () => {
calls.push('after-ok'); } },
+ ]);
+ const context = new
MessageInterceptorContextImpl(MessageHookPoints.CONSUME);
+ composited.doBefore(context, []);
+ composited.doAfter(MessageInterceptorContextImpl.withStatus(context,
MessageHookPointsStatus.ERROR), []);
+ assert.deepStrictEqual(calls, [ 'before-throwing', 'before-ok',
'after-ok', 'after-throwing' ]);
+ });
+
+ it('should carry attributes between doBefore and doAfter', () => {
+ const key = AttributeKey.create<string>('k');
+ let seen: string | undefined;
+ const composited = new CompositedMessageInterceptor([
+ {
+ doBefore: context => { context.putAttribute(key,
Attribute.create('v')); },
+ doAfter: context => { seen = context.getAttribute(key)?.get(); },
+ },
+ ]);
+ const context = new
MessageInterceptorContextImpl(MessageHookPoints.RECEIVE);
+ composited.doBefore(context, []);
+ composited.doAfter(MessageInterceptorContextImpl.withStatus(context,
MessageHookPointsStatus.OK), []);
+ assert.strictEqual(seen, 'v');
+ });
+
+ it('should count inflight receive requests via
InflightRequestCountInterceptor', () => {
+ const interceptor = new InflightRequestCountInterceptor();
+ const composited = new CompositedMessageInterceptor([ interceptor ]);
+ const context = new
MessageInterceptorContextImpl(MessageHookPoints.RECEIVE);
+ composited.doBefore(context, []);
+ composited.doBefore(context, []);
+ assert.strictEqual(interceptor.getInflightReceiveRequestCount(), 2);
+ composited.doAfter(MessageInterceptorContextImpl.withStatus(context,
MessageHookPointsStatus.OK), []);
+ assert.strictEqual(interceptor.getInflightReceiveRequestCount(), 1);
+ });
+});
diff --git a/nodejs/test/metrics.test.ts b/nodejs/test/metrics.test.ts
new file mode 100644
index 00000000..7a7c7cf4
--- /dev/null
+++ b/nodejs/test/metrics.test.ts
@@ -0,0 +1,352 @@
+/**
+ * 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.
+ */
+
+/**
+ * Regression tests for the metrics export pipeline, aligned with the Java
+ * client's org.apache.rocketmq.client.java.metrics package:
+ * - Metric switch parsing (on flag + endpoints)
+ * - Histogram bucket definitions / OTel view construction
+ * - ClientMeter / ClientMeterManager enable-disable lifecycle
+ * - MessageMeterInterceptor recording at the SEND/RECEIVE/CONSUME hook points
+ * - PushConsumerGaugeObserver aggregation of cached message count/bytes
+ */
+
+import { describe, it } from 'node:test';
+import * as assert from 'node:assert';
+import { Metadata } from '@grpc/grpc-js';
+import { Attributes } from '@opentelemetry/api';
+import { InstrumentType } from '@opentelemetry/sdk-metrics';
+import {
+ Endpoints as EndpointsPB,
+ AddressScheme,
+ Metric as MetricPB,
+} from '../proto/apache/rocketmq/v2/definition_pb';
+import { Producer } from '../src/producer';
+import { ProcessQueue } from '../src/consumer/ProcessQueue';
+import { PushConsumerGaugeObserver } from
'../src/consumer/PushConsumerGaugeObserver';
+import { Metric } from '../src/metrics/Metric';
+import { ClientMeter } from '../src/metrics/ClientMeter';
+import { ClientMeterManager } from '../src/metrics/ClientMeterManager';
+import { EmptyGaugeObserver } from '../src/metrics/EmptyGaugeObserver';
+import { GaugeEnum } from '../src/metrics/GaugeEnum';
+import {
+ HistogramEnum,
+ HISTOGRAM_BUCKETS,
+ buildHistogramView,
+} from '../src/metrics/HistogramEnum';
+import { InvocationStatus } from '../src/metrics/InvocationStatus';
+import { MetricLabels } from '../src/metrics/MetricLabels';
+import {
+ MessageHookPoints,
+ MessageHookPointsStatus,
+ MessageInterceptorContextImpl,
+ MessageMeterInterceptor,
+} from '../src/hook';
+
+const silentLogger = {
+ info() { /* noop */ },
+ warn() { /* noop */ },
+ error() { /* noop */ },
+ debug() { /* noop */ },
+};
+
+function buildMetricPb(on: boolean, host = '127.0.0.1', port = 10811):
MetricPB {
+ const endpointsPb = new EndpointsPB();
+ endpointsPb.setScheme(AddressScheme.IPV4);
+ endpointsPb.addAddresses().setHost(host).setPort(port);
+ return new MetricPB().setOn(on).setEndpoints(endpointsPb);
+}
+
+/**
+ * Captures every histogram record instead of exporting it, so the interceptor
+ * can be exercised without standing up the OTLP pipeline.
+ */
+class RecordingMeterManager {
+ readonly records: Array<{ histogramEnum: HistogramEnum; attributes:
Attributes; value: number }> = [];
+ enabled = true;
+
+ isEnabled(): boolean {
+ return this.enabled;
+ }
+
+ record(histogramEnum: HistogramEnum, attributes: Attributes, value: number):
void {
+ this.records.push({ histogramEnum, attributes, value });
+ }
+
+ setGaugeObserver(): void {
+ // not exercised by the interceptor
+ }
+
+ async shutdown(): Promise<void> {
+ // noop
+ }
+}
+
+function asManager(recorder: RecordingMeterManager): ClientMeterManager {
+ return recorder as unknown as ClientMeterManager;
+}
+
+describe('Metric switch parsing (vs Java metrics.Metric)', () => {
+ it('should be enabled when on=true and endpoints are present', () => {
+ const metric = new Metric(buildMetricPb(true, '127.0.0.1', 10811));
+ assert.strictEqual(metric.on, true);
+ assert.ok(metric.endpoints, 'expected endpoints to be parsed');
+ assert.deepStrictEqual(metric.endpoints!.addressesList, [{ host:
'127.0.0.1', port: 10811 }]);
+ });
+
+ it('should be disabled when the on flag is false', () => {
+ const metric = new Metric(buildMetricPb(false));
+ assert.strictEqual(metric.on, false);
+ assert.strictEqual(metric.endpoints, null);
+ });
+
+ it('should be disabled when the metric is absent', () => {
+ const metric = new Metric(undefined);
+ assert.strictEqual(metric.on, false);
+ assert.strictEqual(metric.endpoints, null);
+ });
+});
+
+describe('Histogram buckets and views', () => {
+ it('should define explicit buckets for all four histograms', () => {
+ const histograms = Object.values(HistogramEnum);
+ assert.strictEqual(histograms.length, 4);
+ for (const histogram of histograms) {
+ assert.ok(HISTOGRAM_BUCKETS[histogram].length > 0,
+ `expected buckets for ${histogram}`);
+ }
+ });
+
+ it('should build a view targeting the histogram instrument', () => {
+ const view = buildHistogramView(HistogramEnum.SEND_COST_TIME);
+ assert.strictEqual(view.instrumentName, HistogramEnum.SEND_COST_TIME);
+ assert.strictEqual(view.instrumentType, InstrumentType.HISTOGRAM);
+ assert.ok(view.aggregation, 'expected an explicit bucket aggregation');
+ });
+});
+
+describe('ClientMeter lifecycle', () => {
+ it('should be a no-op when disabled', () => {
+ const meter = ClientMeter.disabledInstance();
+ assert.strictEqual(meter.enabled, false);
+ // must not throw even though no underlying meter exists
+ meter.record(HistogramEnum.SEND_COST_TIME, {}, 1);
+ });
+
+ it('should only be satisfied by a matching switch', () => {
+ const meter = ClientMeter.disabledInstance();
+ assert.strictEqual(meter.satisfy(new Metric(undefined)), true);
+ assert.strictEqual(meter.satisfy(new Metric(buildMetricPb(true))), false);
+ });
+});
+
+describe('ClientMeterManager reset', () => {
+ it('should keep metrics disabled while the switch is off', async () => {
+ const manager = new ClientMeterManager('client-1', () => new Metadata(),
silentLogger);
+ assert.strictEqual(manager.isEnabled(), false);
+ await manager.reset(new Metric(buildMetricPb(false)));
+ assert.strictEqual(manager.isEnabled(), false);
+ // idempotent: a second reset must not throw
+ await manager.reset(new Metric(undefined));
+ assert.strictEqual(manager.isEnabled(), false);
+ await manager.shutdown();
+ });
+
+ it('should build the OTLP pipeline when the switch is on', async () => {
+ const manager = new ClientMeterManager('client-2', () => new Metadata(),
silentLogger);
+ try {
+ // 127.0.0.1:65535 is a closed port: the pipeline is built eagerly, while
+ // the periodic export (60s) never fires during the test.
+ await manager.reset(new Metric(buildMetricPb(true, '127.0.0.1', 65535)));
+ assert.strictEqual(manager.isEnabled(), true, 'expected the meter to be
enabled');
+ } finally {
+ await manager.shutdown();
+ }
+ });
+});
+
+describe('MessageMeterInterceptor (SEND/RECEIVE/CONSUME)', () => {
+ it('should record send cost time on the SEND hook point', () => {
+ const recorder = new RecordingMeterManager();
+ const interceptor = new MessageMeterInterceptor(asManager(recorder),
'client-1', () => undefined);
+ const context = new MessageInterceptorContextImpl(MessageHookPoints.SEND);
+ const message = { topic: 'TopicSend' } as any;
+ interceptor.doBefore(context, [ message ]);
+ interceptor.doAfter(
+ MessageInterceptorContextImpl.withStatus(context,
MessageHookPointsStatus.OK), [ message ]);
+
+ assert.strictEqual(recorder.records.length, 1);
+ const record = recorder.records[0];
+ assert.strictEqual(record.histogramEnum, HistogramEnum.SEND_COST_TIME);
+ assert.strictEqual(record.attributes[MetricLabels.TOPIC], 'TopicSend');
+ assert.strictEqual(record.attributes[MetricLabels.CLIENT_ID], 'client-1');
+ assert.strictEqual(record.attributes[MetricLabels.INVOCATION_STATUS],
InvocationStatus.SUCCESS.name);
+ assert.ok(record.value >= 0, 'expected a non-negative cost');
+ });
+
+ it('should mark the send as failed when the hook status is ERROR', () => {
+ const recorder = new RecordingMeterManager();
+ const interceptor = new MessageMeterInterceptor(asManager(recorder),
'client-1', () => undefined);
+ const context = new MessageInterceptorContextImpl(MessageHookPoints.SEND);
+ const message = { topic: 'TopicSend' } as any;
+ interceptor.doBefore(context, [ message ]);
+ interceptor.doAfter(
+ MessageInterceptorContextImpl.withStatus(context,
MessageHookPointsStatus.ERROR), [ message ]);
+
+ assert.strictEqual(recorder.records.length, 1);
+
assert.strictEqual(recorder.records[0].attributes[MetricLabels.INVOCATION_STATUS],
+ InvocationStatus.FAILURE.name);
+ });
+
+ it('should record delivery latency on the RECEIVE hook point', () => {
+ const recorder = new RecordingMeterManager();
+ const interceptor = new MessageMeterInterceptor(asManager(recorder),
'client-1', () => 'group-1');
+ const context = new
MessageInterceptorContextImpl(MessageHookPoints.RECEIVE);
+ const message = { topic: 'TopicRecv', transportDeliveryTimestamp: new
Date(Date.now() - 120) } as any;
+ interceptor.doBefore(context, [ message ]);
+ interceptor.doAfter(
+ MessageInterceptorContextImpl.withStatus(context,
MessageHookPointsStatus.OK), [ message ]);
+
+ assert.strictEqual(recorder.records.length, 1);
+ const record = recorder.records[0];
+ assert.strictEqual(record.histogramEnum, HistogramEnum.DELIVERY_LATENCY);
+ assert.strictEqual(record.attributes[MetricLabels.TOPIC], 'TopicRecv');
+ assert.strictEqual(record.attributes[MetricLabels.CONSUMER_GROUP],
'group-1');
+ // 120ms elapsed between the delivery timestamp and now
+ assert.ok(record.value >= 100 && record.value < 5000, `unexpected latency
${record.value}`);
+ });
+
+ it('should skip delivery latency when the timestamp is in the future', () =>
{
+ const recorder = new RecordingMeterManager();
+ const interceptor = new MessageMeterInterceptor(asManager(recorder),
'client-1', () => 'group-1');
+ const context = new
MessageInterceptorContextImpl(MessageHookPoints.RECEIVE);
+ const message = { topic: 'TopicRecv', transportDeliveryTimestamp: new
Date(Date.now() + 60_000) } as any;
+ interceptor.doBefore(context, [ message ]);
+ interceptor.doAfter(
+ MessageInterceptorContextImpl.withStatus(context,
MessageHookPointsStatus.OK), [ message ]);
+
+ assert.strictEqual(recorder.records.length, 0, 'negative latency must not
be recorded');
+ });
+
+ it('should record await time and process time on the CONSUME hook point', ()
=> {
+ const recorder = new RecordingMeterManager();
+ const interceptor = new MessageMeterInterceptor(asManager(recorder),
'client-1', () => 'group-1');
+ const context = new
MessageInterceptorContextImpl(MessageHookPoints.CONSUME);
+ const message = { topic: 'TopicConsume', decodeTimestamp: new
Date(Date.now() - 200) } as any;
+ interceptor.doBefore(context, [ message ]);
+ interceptor.doAfter(
+ MessageInterceptorContextImpl.withStatus(context,
MessageHookPointsStatus.OK), [ message ]);
+
+ assert.strictEqual(recorder.records.length, 2);
+ assert.strictEqual(recorder.records[0].histogramEnum,
HistogramEnum.AWAIT_TIME);
+ assert.ok(recorder.records[0].value >= 200, `unexpected await time
${recorder.records[0].value}`);
+ assert.strictEqual(recorder.records[1].histogramEnum,
HistogramEnum.PROCESS_TIME);
+
assert.strictEqual(recorder.records[1].attributes[MetricLabels.CONSUMER_GROUP],
'group-1');
+
assert.strictEqual(recorder.records[1].attributes[MetricLabels.INVOCATION_STATUS],
+ InvocationStatus.SUCCESS.name);
+ });
+
+ it('should record nothing when metrics are disabled', () => {
+ const recorder = new RecordingMeterManager();
+ recorder.enabled = false;
+ const interceptor = new MessageMeterInterceptor(asManager(recorder),
'client-1', () => 'group-1');
+ const context = new MessageInterceptorContextImpl(MessageHookPoints.SEND);
+ const message = { topic: 'TopicSend' } as any;
+ interceptor.doBefore(context, [ message ]);
+ interceptor.doAfter(
+ MessageInterceptorContextImpl.withStatus(context,
MessageHookPointsStatus.OK), [ message ]);
+ assert.strictEqual(recorder.records.length, 0);
+ });
+
+ it('should record nothing on CONSUME when the client has no consumer group',
() => {
+ const recorder = new RecordingMeterManager();
+ const interceptor = new MessageMeterInterceptor(asManager(recorder),
'client-1', () => undefined);
+ const context = new
MessageInterceptorContextImpl(MessageHookPoints.CONSUME);
+ const message = { topic: 'TopicConsume', decodeTimestamp: new
Date(Date.now() - 50) } as any;
+ interceptor.doBefore(context, [ message ]);
+ interceptor.doAfter(
+ MessageInterceptorContextImpl.withStatus(context,
MessageHookPointsStatus.OK), [ message ]);
+ assert.strictEqual(recorder.records.length, 0);
+ });
+});
+
+describe('Consumer gauge observers', () => {
+ function fakeProcessQueue(topic: string, count: number, bytes: number):
ProcessQueue {
+ return {
+ topic,
+ cachedMessagesCount: () => count,
+ cachedMessageBytes: () => bytes,
+ } as unknown as ProcessQueue;
+ }
+
+ function toEntries(values: Map<Attributes, number>) {
+ return Array.from(values.entries())
+ .map(([ attributes, value ]) => ({
+ topic: attributes[MetricLabels.TOPIC],
+ group: attributes[MetricLabels.CONSUMER_GROUP],
+ clientId: attributes[MetricLabels.CLIENT_ID],
+ value,
+ }))
+ .sort((a, b) => String(a.topic).localeCompare(String(b.topic)));
+ }
+
+ it('should aggregate cached messages and bytes per topic', () => {
+ const observer = new PushConsumerGaugeObserver(
+ () => [
+ fakeProcessQueue('TopicA', 3, 300),
+ fakeProcessQueue('TopicA', 2, 200),
+ fakeProcessQueue('TopicB', 5, 500),
+ ],
+ 'client-1', 'group-1');
+
+ assert.deepStrictEqual(observer.getGauges(),
+ [ GaugeEnum.CONSUMER_CACHED_MESSAGES, GaugeEnum.CONSUMER_CACHED_BYTES ]);
+
+
assert.deepStrictEqual(toEntries(observer.getValues(GaugeEnum.CONSUMER_CACHED_MESSAGES)),
[
+ { topic: 'TopicA', group: 'group-1', clientId: 'client-1', value: 5 },
+ { topic: 'TopicB', group: 'group-1', clientId: 'client-1', value: 5 },
+ ]);
+
assert.deepStrictEqual(toEntries(observer.getValues(GaugeEnum.CONSUMER_CACHED_BYTES)),
[
+ { topic: 'TopicA', group: 'group-1', clientId: 'client-1', value: 500 },
+ { topic: 'TopicB', group: 'group-1', clientId: 'client-1', value: 500 },
+ ]);
+ });
+
+ it('should return an empty map when no process queue is assigned', () => {
+ const observer = new PushConsumerGaugeObserver(() => [], 'client-1',
'group-1');
+
assert.strictEqual(observer.getValues(GaugeEnum.CONSUMER_CACHED_MESSAGES).size,
0);
+
assert.strictEqual(observer.getValues(GaugeEnum.CONSUMER_CACHED_BYTES).size, 0);
+ });
+
+ it('should expose no gauges through the empty observer', () => {
+ assert.deepStrictEqual(EmptyGaugeObserver.EMPTY.getGauges(), []);
+ assert.strictEqual(EmptyGaugeObserver.EMPTY.getValues().size, 0);
+ });
+});
+
+describe('BaseClient metrics wiring', () => {
+ it('should create the meter manager during construction', () => {
+ const producer = new Producer({
+ endpoints: '127.0.0.1:8081',
+ namespace: '',
+ topics: [ 'TopicTest' ],
+ });
+ const manager = (producer as any).clientMeterManager;
+ assert.ok(manager instanceof ClientMeterManager, 'expected a
ClientMeterManager to be created');
+ assert.strictEqual(manager.isEnabled(), false, 'metrics start disabled
until the settings command');
+ });
+});
