Refactor gRPC runtime services and add Node.js runtime metrics (#139)
Refactor gRPC service lifecycle and add Node.js runtime metrics
This PR replaces the old per-client gRPC protocol wiring with a ServiceManager-based lifecycle and a shared grpc-js channel manager, using the Java agent core service model as structural reference while keeping Node.js-specific async behavior.
It adds Node.js runtime metric collection through MeterReportService, including process CPU, heap, RSS, external memory, and heap limit meters, with bounded buffering, configurable collection/report intervals, and deprecated runtime metric config aliases.
The gRPC lifecycle now includes explicit shutdown guards, stale callback protection, DISCONNECT handling, per-call deadlines, throttled error logging, and controlled report/flush scheduling to avoid agent-side crashes, duplicate timers, or unnecessary RPC work while disconnected.
Verified with lint, build, focused config/runtime/channel tests, and manual lifecycle repros for shutdown, late grpc-js callbacks, reconnect state, and destroy/start config behavior.
diff --git a/README.md b/README.md
index 2a31b3e..50586c5 100644
--- a/README.md
+++ b/README.md
@@ -72,10 +72,31 @@
| `SW_AWS_SQS_CHECK_BODY` | Incoming SQS messages check inside the body for trace ID in order to allow linking outgoing SNS messages to incoming SQS. | `false` |
| `SW_AGENT_MAX_BUFFER_SIZE` | The maximum buffer size before sending the segment data to backend | `'1000'` |
| `SW_AGENT_TRACE_TIMEOUT` | The timeout for trace requests to backend services | `'10000'` |
+| `SW_AGENT_NODEJS_RUNTIME_METRICS_REPORTER_ACTIVE` | Whether to report Node.js runtime metrics through MeterReportService (default: collect 1s, report 1s) | `true` |
+| `SW_AGENT_NODEJS_RUNTIME_METRICS_COLLECT_PERIOD` | Runtime metric sample interval in milliseconds | `1000` |
+| `SW_AGENT_NODEJS_RUNTIME_METRICS_REPORT_PERIOD` | Runtime metric report interval in milliseconds (aligned with Java JVM metrics upload interval) | `1000` |
+| `SW_AGENT_NODEJS_RUNTIME_METRICS_BUFFER_SIZE` | Maximum buffered runtime metric samples before dropping oldest | `600` |
+
+Legacy env names `SW_AGENT_RUNTIME_METRICS_*`, `SW_AGENT_NVM_METRICS_*` and `SW_AGENT_NVM_JVM_*` are still accepted as deprecated aliases.
Note that the various ignore options like `SW_IGNORE_SUFFIX`, `SW_TRACE_IGNORE_PATH` and `SW_HTTP_IGNORE_METHOD` as well as endpoints which are not recorded due to exceeding `SW_AGENT_MAX_BUFFER_SIZE` all propagate their ignored status downstream to any other endpoints they may call. If that endpoint is running the Node Skywalking agent then regardless of its ignore settings it will not be recorded since its upstream parent was not recorded. This allows the elimination of entire trees of endpoints you are not interested in as well as eliminating partial traces if a span in the chain is ignored but calls out to other endpoints which are recorded as children of ROOT instead of the actual parent.
+## Node.js Runtime Metrics
+
+The agent reports six process-level meters (`instance_nodejs_*`) via `MeterReportService` by default (collect 1s, report 1s). Set `SW_AGENT_NODEJS_RUNTIME_METRICS_REPORTER_ACTIVE=false` to disable. Process CPU combines `process.cpuUsage()` user + system, normalized by logical CPU count (0–100%).
+
+| Node.js source | Meter name | Notes |
+| :--- | :--- | :--- |
+| `process.cpuUsage()` user + system | `instance_nodejs_process_cpu` | % |
+| `process.memoryUsage().heapUsed` | `instance_nodejs_heap_used` | bytes |
+| `process.memoryUsage().heapTotal` | `instance_nodejs_heap_total` | bytes |
+| `v8.getHeapStatistics().heap_size_limit` | `instance_nodejs_heap_limit` | bytes |
+| `process.memoryUsage().rss` | `instance_nodejs_rss` | bytes |
+| `process.memoryUsage().external` | `instance_nodejs_external_memory` | bytes |
+
+Custom business metrics are not available through a public API; use [OpenTelemetry metrics](https://skywalking.apache.org/docs/main/latest/en/setup/backend/opentelemetry-receiver/) if you need those.
+
## Supported Libraries
Some built-in plugins support automatic instrumentation of NodeJS libraries, the complete list is as follows:
diff --git a/src/agent/protocol/grpc/clients/Client.ts b/src/agent/core/boot/BootService.ts
similarity index 79%
copy from src/agent/protocol/grpc/clients/Client.ts
copy to src/agent/core/boot/BootService.ts
index 8d8fd06..34a53d5 100644
--- a/src/agent/protocol/grpc/clients/Client.ts
+++ b/src/agent/core/boot/BootService.ts
@@ -17,12 +17,15 @@
*
*/
-export default interface Client {
- readonly isConnected: boolean;
+/** Boot lifecycle contract (Java {@code BootService}). */
+export default interface BootService {
+ prepare(): void;
- start(): void;
+ boot(): void;
- flush(): Promise<any> | null;
+ onComplete(): void;
- destroy?(): void;
+ shutdown(): void;
+
+ priority(): number;
}
diff --git a/src/agent/core/boot/ServiceManager.ts b/src/agent/core/boot/ServiceManager.ts
new file mode 100644
index 0000000..e5dd18d
--- /dev/null
+++ b/src/agent/core/boot/ServiceManager.ts
@@ -0,0 +1,133 @@
+/*!
+ *
+ * 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 config from '../../../config/AgentConfig';
+import { createLogger } from '../../../logging';
+import BootService from './BootService';
+import MeterSender from '../meter/MeterSender';
+import GRPCChannelManager from '../remote/GRPCChannelManager';
+import ServiceManagementClient from '../remote/ServiceManagementClient';
+import TraceSegmentServiceClient from '../remote/TraceSegmentServiceClient';
+
+const logger = createLogger(__filename);
+
+/** Service registry and boot orchestrator (Java {@code ServiceManager}). */
+class ServiceManager {
+ private static readonly instance = new ServiceManager();
+
+ private bootedServices = new Map<new (...args: never[]) => BootService, BootService>();
+
+ private booted = false;
+
+ static get INSTANCE(): ServiceManager {
+ return ServiceManager.instance;
+ }
+
+ boot(): void {
+ if (this.booted) {
+ return;
+ }
+
+ this.loadServices();
+ const services = this.sortByPriority(true);
+
+ for (const service of services) {
+ try {
+ service.prepare();
+ } catch (error) {
+ logger.error('ServiceManager prepare failed: %s', error);
+ }
+ }
+
+ for (const service of services) {
+ try {
+ service.boot();
+ } catch (error) {
+ logger.error('ServiceManager boot failed: %s', error);
+ }
+ }
+
+ for (const service of services) {
+ try {
+ service.onComplete();
+ } catch (error) {
+ logger.error('ServiceManager onComplete failed: %s', error);
+ }
+ }
+
+ this.booted = true;
+ }
+
+ shutdown(): void {
+ for (const service of this.sortByPriority(false)) {
+ try {
+ service.shutdown();
+ } catch (error) {
+ logger.error('ServiceManager shutdown failed: %s', error);
+ }
+ }
+ this.bootedServices.clear();
+ this.booted = false;
+ }
+
+ flush(): Promise<unknown> | null {
+ const traceFlush = this.findService(TraceSegmentServiceClient)?.flush() ?? null;
+ const meterSender = this.findService(MeterSender);
+ const meterFlush = meterSender?.flush() ?? null;
+
+ if (!traceFlush && !meterFlush) {
+ return null;
+ }
+ if (!traceFlush) {
+ return meterFlush;
+ }
+ if (!meterFlush) {
+ return traceFlush;
+ }
+
+ return Promise.all([traceFlush, meterFlush]).then(() => null);
+ }
+
+ findService<T extends BootService>(serviceClass: new (...args: never[]) => T): T | undefined {
+ return this.bootedServices.get(serviceClass) as T | undefined;
+ }
+
+ private register<T extends BootService>(serviceClass: new (...args: never[]) => T, service: T): void {
+ this.bootedServices.set(serviceClass, service);
+ }
+
+ private loadServices(): void {
+ this.register(GRPCChannelManager, new GRPCChannelManager());
+ this.register(TraceSegmentServiceClient, new TraceSegmentServiceClient());
+ this.register(ServiceManagementClient, new ServiceManagementClient());
+ if (config.runtimeMetricsReporterActive) {
+ this.register(MeterSender, new MeterSender());
+ }
+ }
+
+ private sortByPriority(ascending: boolean): BootService[] {
+ const services = Array.from(this.bootedServices.values());
+ services.sort((left, right) =>
+ ascending ? left.priority() - right.priority() : right.priority() - left.priority(),
+ );
+ return services;
+ }
+}
+
+export default ServiceManager;
diff --git a/src/agent/core/meter/MeterSender.ts b/src/agent/core/meter/MeterSender.ts
new file mode 100644
index 0000000..b66d370
--- /dev/null
+++ b/src/agent/core/meter/MeterSender.ts
@@ -0,0 +1,222 @@
+/*!
+ *
+ * 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 config from '../../../config/AgentConfig';
+import * as grpc from '@grpc/grpc-js';
+import { createLogger, throttled } from '../../../logging';
+import { MeterReportServiceClient } from '../../../proto/language-agent/Meter_grpc_pb';
+import BootService from '../boot/BootService';
+import ServiceManager from '../boot/ServiceManager';
+import RuntimeMetricsCollector from './RuntimeMetricsCollector';
+import { RuntimeSnapshot } from './RuntimeSampler';
+import GRPCChannelManager from '../remote/GRPCChannelManager';
+import { GRPCChannelListener } from '../remote/GRPCChannelListener';
+import { GRPCChannelStatus } from '../remote/GRPCChannelStatus';
+
+const logger = createLogger(__filename);
+const logReportError = throttled(logger, 'error', 30000);
+
+/** Reports Node.js runtime metrics via gRPC MeterReportService (Go/Python-compatible pipeline). */
+export default class MeterSender implements BootService, GRPCChannelListener {
+ private closed = false;
+ private channelManager?: GRPCChannelManager;
+ private status = GRPCChannelStatus.DISCONNECT;
+ private reporterClient?: MeterReportServiceClient;
+ private readonly buffer: RuntimeSnapshot[] = [];
+ private collectTimer?: NodeJS.Timeout;
+ private reportTimer?: NodeJS.Timeout;
+ private reporting?: Promise<void>;
+
+ private collector!: RuntimeMetricsCollector;
+
+ prepare(): void {
+ this.collector = new RuntimeMetricsCollector();
+ this.channelManager = ServiceManager.INSTANCE.findService(GRPCChannelManager);
+ this.channelManager?.addChannelListener(this);
+ }
+
+ boot(): void {
+ if (this.collectTimer || this.reportTimer) {
+ logger.warn('MeterSender timers already scheduled; skipping duplicate boot.');
+ return;
+ }
+
+ this.startTimers();
+ }
+
+ onComplete(): void {}
+
+ priority(): number {
+ return 0;
+ }
+
+ statusChanged(status: GRPCChannelStatus): void {
+ this.status = status;
+ this.reporterClient = status === GRPCChannelStatus.CONNECTED ? this.createReporterClient() : undefined;
+ }
+
+ private createReporterClient(): MeterReportServiceClient | undefined {
+ if (!this.channelManager) {
+ return undefined;
+ }
+
+ return new MeterReportServiceClient(
+ config.collectorAddress,
+ grpc.credentials.createInsecure(),
+ this.channelManager.getClientOptions(),
+ );
+ }
+
+ private startTimers(): void {
+ this.collectTimer = setInterval(() => {
+ if (this.closed) {
+ return;
+ }
+ this.collectSample();
+ }, config.runtimeMetricsCollectPeriod || 1000) as NodeJS.Timeout;
+ this.collectTimer.unref();
+ this.reportTimer = setInterval(() => {
+ if (this.closed) {
+ return;
+ }
+ void this.reportBufferedMetrics();
+ }, config.runtimeMetricsReportPeriod || 1000) as NodeJS.Timeout;
+ this.reportTimer.unref();
+ }
+
+ private collectSample(): void {
+ const maxBufferSize = config.runtimeMetricsBufferSize || 600;
+ if (this.buffer.length >= maxBufferSize) {
+ this.buffer.shift();
+ }
+ this.buffer.push(this.collector.sample());
+ }
+
+ private reportBufferedMetrics(): Promise<void> {
+ if (this.closed) {
+ return Promise.resolve();
+ }
+
+ if (this.reporting) {
+ return this.reporting;
+ }
+
+ this.reporting = this.doReportBufferedMetrics().finally(() => {
+ this.reporting = undefined;
+ });
+
+ return this.reporting;
+ }
+
+ private doReportBufferedMetrics(): Promise<void> {
+ return new Promise((resolve) => {
+ try {
+ if (this.closed) {
+ resolve();
+ return;
+ }
+
+ if (this.buffer.length === 0 || this.status !== GRPCChannelStatus.CONNECTED || !this.reporterClient) {
+ resolve();
+ return;
+ }
+
+ if (!config.serviceName || !config.serviceInstance) {
+ resolve();
+ return;
+ }
+
+ const snapshots = this.buffer.splice(0, this.buffer.length);
+ const stream = this.reporterClient.collect(
+ new grpc.Metadata(),
+ { deadline: Date.now() + (config.traceTimeout || 10000) },
+ (error: grpc.ServiceError | null) => {
+ if (error) {
+ logReportError('Failed to report runtime meter data', error);
+ this.reportGrpcError(error);
+ }
+ resolve();
+ },
+ );
+
+ try {
+ let metadataWritten = false;
+ const timestamp = Date.now();
+ for (const snapshot of snapshots) {
+ for (const meterData of this.collector.toMeterData(snapshot)) {
+ if (!metadataWritten) {
+ meterData
+ .setService(config.serviceName)
+ .setServiceinstance(config.serviceInstance)
+ .setTimestamp(timestamp);
+ metadataWritten = true;
+ }
+ stream.write(meterData);
+ }
+ }
+ } finally {
+ try {
+ stream.end();
+ } catch (error) {
+ logReportError('Failed to end meter collect stream', error);
+ resolve();
+ }
+ }
+ } catch (error) {
+ logReportError('Failed to report runtime meter data', error);
+ this.reportGrpcError(error);
+ resolve();
+ }
+ });
+ }
+
+ private reportGrpcError(error: unknown): void {
+ if (this.closed) {
+ return;
+ }
+
+ this.channelManager?.reportError(error);
+ }
+
+ flush(): Promise<any> | null {
+ if (this.closed) {
+ return null;
+ }
+ this.collectSample();
+ return this.reportBufferedMetrics();
+ }
+
+ shutdown(): void {
+ this.closed = true;
+ if (this.collectTimer) {
+ clearInterval(this.collectTimer);
+ this.collectTimer = undefined;
+ }
+ if (this.reportTimer) {
+ clearInterval(this.reportTimer);
+ this.reportTimer = undefined;
+ }
+ this.reporting = undefined;
+ this.reporterClient = undefined;
+ this.buffer.length = 0;
+ this.collector.destroy();
+ this.channelManager = undefined;
+ logger.info('MeterSender destroyed and resources cleaned up');
+ }
+}
diff --git a/src/agent/core/meter/RuntimeMetricsCollector.ts b/src/agent/core/meter/RuntimeMetricsCollector.ts
new file mode 100644
index 0000000..0d59a29
--- /dev/null
+++ b/src/agent/core/meter/RuntimeMetricsCollector.ts
@@ -0,0 +1,49 @@
+/*!
+ *
+ * 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 { MeterData, MeterSingleValue } from '../../../proto/language-agent/Meter_pb';
+import RuntimeSampler, { RuntimeSnapshot } from './RuntimeSampler';
+
+/** Maps Node.js runtime samples into MeterReportService single-value meters (instance_nodejs_*). */
+export default class RuntimeMetricsCollector {
+ private readonly sampler = new RuntimeSampler();
+
+ sample(): RuntimeSnapshot {
+ return this.sampler.sample();
+ }
+
+ toMeterData(snapshot: RuntimeSnapshot): MeterData[] {
+ const gauges: Array<[string, number]> = [
+ ['instance_nodejs_process_cpu', snapshot.cpuUserPercent + snapshot.cpuSystemPercent],
+ ['instance_nodejs_heap_used', snapshot.heapUsed],
+ ['instance_nodejs_heap_total', snapshot.heapTotal],
+ ['instance_nodejs_heap_limit', snapshot.heapSizeLimit],
+ ['instance_nodejs_rss', snapshot.rss],
+ ['instance_nodejs_external_memory', snapshot.external],
+ ];
+
+ return gauges.map(([name, value]) =>
+ new MeterData().setSinglevalue(new MeterSingleValue().setName(name).setValue(value)),
+ );
+ }
+
+ destroy(): void {
+ this.sampler.destroy();
+ }
+}
diff --git a/src/agent/core/meter/RuntimeSampler.ts b/src/agent/core/meter/RuntimeSampler.ts
new file mode 100644
index 0000000..c7ac80e
--- /dev/null
+++ b/src/agent/core/meter/RuntimeSampler.ts
@@ -0,0 +1,65 @@
+/*!
+ *
+ * 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 os from 'os';
+import v8 from 'v8';
+
+export type RuntimeSnapshot = {
+ heapUsed: number;
+ heapTotal: number;
+ heapSizeLimit: number;
+ rss: number;
+ external: number;
+ cpuUserPercent: number;
+ cpuSystemPercent: number;
+};
+
+export default class RuntimeSampler {
+ private readonly logicalCpuCount = Math.max(1, os.cpus().length);
+ private lastCpuUsage = process.cpuUsage();
+ private lastCpuTimestamp = process.hrtime.bigint();
+
+ sample(): RuntimeSnapshot {
+ const memory = process.memoryUsage();
+ const heapStats = v8.getHeapStatistics();
+ const cpuUsage = process.cpuUsage(this.lastCpuUsage);
+ const now = process.hrtime.bigint();
+ const elapsedMicros = Number(now - this.lastCpuTimestamp) / 1000;
+ this.lastCpuUsage = process.cpuUsage();
+ this.lastCpuTimestamp = now;
+
+ const cpuScale = elapsedMicros > 0 ? 100 / elapsedMicros / this.logicalCpuCount : 0;
+ const cpuUserPercent = cpuUsage.user * cpuScale;
+ const cpuSystemPercent = cpuUsage.system * cpuScale;
+
+ return {
+ heapUsed: memory.heapUsed,
+ heapTotal: memory.heapTotal,
+ heapSizeLimit: heapStats.heap_size_limit,
+ rss: memory.rss,
+ external: memory.external,
+ cpuUserPercent,
+ cpuSystemPercent,
+ };
+ }
+
+ destroy(): void {
+ // no-op: kept for symmetry with start/stop lifecycle
+ }
+}
diff --git a/src/agent/protocol/grpc/AuthInterceptor.ts b/src/agent/core/remote/AgentIDDecorator.ts
similarity index 61%
copy from src/agent/protocol/grpc/AuthInterceptor.ts
copy to src/agent/core/remote/AgentIDDecorator.ts
index 0f5ae4d..6f77406 100644
--- a/src/agent/protocol/grpc/AuthInterceptor.ts
+++ b/src/agent/core/remote/AgentIDDecorator.ts
@@ -18,12 +18,19 @@
*/
import * as grpc from '@grpc/grpc-js';
-import config from '../../../config/AgentConfig';
+import * as packageInfo from '../../../../package.json';
+import ChannelDecorator from './ChannelDecorator';
-export default function AuthInterceptor() {
- const mata = new grpc.Metadata();
- if (config.authorization) {
- mata.add('Authentication', config.authorization);
+/** Add agent version to every RPC (Java AgentIDDecorator). */
+export default class AgentIDDecorator implements ChannelDecorator {
+ build(): grpc.Interceptor {
+ return (options, nextCall) => {
+ return new grpc.InterceptingCall(nextCall(options), {
+ start: (metadata, listener, next) => {
+ metadata.set('Agent-Version', packageInfo.version);
+ next(metadata, listener);
+ },
+ });
+ };
}
- return mata;
}
diff --git a/src/agent/protocol/grpc/AuthInterceptor.ts b/src/agent/core/remote/AuthenticationDecorator.ts
similarity index 60%
copy from src/agent/protocol/grpc/AuthInterceptor.ts
copy to src/agent/core/remote/AuthenticationDecorator.ts
index 0f5ae4d..b0d138f 100644
--- a/src/agent/protocol/grpc/AuthInterceptor.ts
+++ b/src/agent/core/remote/AuthenticationDecorator.ts
@@ -19,11 +19,20 @@
import * as grpc from '@grpc/grpc-js';
import config from '../../../config/AgentConfig';
+import ChannelDecorator from './ChannelDecorator';
-export default function AuthInterceptor() {
- const mata = new grpc.Metadata();
- if (config.authorization) {
- mata.add('Authentication', config.authorization);
+/** Active authentication header by SW_AGENT_AUTHENTICATION (Java AuthenticationDecorator). */
+export default class AuthenticationDecorator implements ChannelDecorator {
+ build(): grpc.Interceptor {
+ return (options, nextCall) => {
+ return new grpc.InterceptingCall(nextCall(options), {
+ start: (metadata, listener, next) => {
+ if (config.authorization) {
+ metadata.set('Authentication', config.authorization);
+ }
+ next(metadata, listener);
+ },
+ });
+ };
}
- return mata;
}
diff --git a/src/agent/protocol/grpc/AuthInterceptor.ts b/src/agent/core/remote/ChannelBuilder.ts
similarity index 72%
copy from src/agent/protocol/grpc/AuthInterceptor.ts
copy to src/agent/core/remote/ChannelBuilder.ts
index 0f5ae4d..9a2d7c9 100644
--- a/src/agent/protocol/grpc/AuthInterceptor.ts
+++ b/src/agent/core/remote/ChannelBuilder.ts
@@ -18,12 +18,13 @@
*/
import * as grpc from '@grpc/grpc-js';
-import config from '../../../config/AgentConfig';
-export default function AuthInterceptor() {
- const mata = new grpc.Metadata();
- if (config.authorization) {
- mata.add('Authentication', config.authorization);
- }
- return mata;
+/** Mutable state passed through ChannelBuilder chain (Java ManagedChannelBuilder equivalent). */
+export interface ChannelBuildContext {
+ credentials: grpc.ChannelCredentials;
+ options: grpc.ChannelOptions;
+}
+
+export default interface ChannelBuilder {
+ build(context: ChannelBuildContext): ChannelBuildContext;
}
diff --git a/src/agent/protocol/grpc/AuthInterceptor.ts b/src/agent/core/remote/ChannelDecorator.ts
similarity index 77%
rename from src/agent/protocol/grpc/AuthInterceptor.ts
rename to src/agent/core/remote/ChannelDecorator.ts
index 0f5ae4d..c99319d 100644
--- a/src/agent/protocol/grpc/AuthInterceptor.ts
+++ b/src/agent/core/remote/ChannelDecorator.ts
@@ -18,12 +18,11 @@
*/
import * as grpc from '@grpc/grpc-js';
-import config from '../../../config/AgentConfig';
-export default function AuthInterceptor() {
- const mata = new grpc.Metadata();
- if (config.authorization) {
- mata.add('Authentication', config.authorization);
- }
- return mata;
+/**
+ * Decorates the gRPC channel (Java ChannelDecorator equivalent).
+ * Node uses client interceptors because @grpc/grpc-js has no ClientInterceptors.intercept().
+ */
+export default interface ChannelDecorator {
+ build(): grpc.Interceptor;
}
diff --git a/src/agent/core/remote/GRPCChannel.ts b/src/agent/core/remote/GRPCChannel.ts
new file mode 100644
index 0000000..4eae558
--- /dev/null
+++ b/src/agent/core/remote/GRPCChannel.ts
@@ -0,0 +1,100 @@
+/*!
+ *
+ * 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 * as grpc from '@grpc/grpc-js';
+import { ClientOptions, connectivityState } from '@grpc/grpc-js';
+import ChannelBuilder, { ChannelBuildContext } from './ChannelBuilder';
+import ChannelDecorator from './ChannelDecorator';
+
+export default class GRPCChannel {
+ private readonly originChannel: grpc.Channel;
+ private readonly interceptors: grpc.Interceptor[];
+
+ private constructor(host: string, port: number, channelBuilders: ChannelBuilder[], decorators: ChannelDecorator[]) {
+ let context: ChannelBuildContext = {
+ credentials: grpc.credentials.createInsecure(),
+ options: {},
+ };
+
+ for (const builder of channelBuilders) {
+ context = builder.build(context);
+ }
+
+ this.originChannel = new grpc.Channel(`${host}:${port}`, context.credentials, context.options);
+ this.interceptors = decorators.map((decorator) => decorator.build());
+ }
+
+ static create(
+ host: string,
+ port: number,
+ channelBuilders: ChannelBuilder[],
+ decorators: ChannelDecorator[],
+ ): GRPCChannel {
+ return new GRPCChannel(host, port, channelBuilders, decorators);
+ }
+
+ static newBuilder(host: string, port: number): GRPCChannelBuilder {
+ return new GRPCChannelBuilder(host, port);
+ }
+
+ getChannel(): grpc.Channel {
+ return this.originChannel;
+ }
+
+ getClientOptions(): ClientOptions {
+ return {
+ channelOverride: this.originChannel,
+ interceptors: this.interceptors,
+ };
+ }
+
+ isConnected(requestConnection = false): boolean {
+ return this.originChannel.getConnectivityState(requestConnection) === connectivityState.READY;
+ }
+
+ shutdownNow(): void {
+ this.originChannel.close();
+ }
+}
+
+class GRPCChannelBuilder {
+ private readonly host: string;
+ private readonly port: number;
+ private readonly channelBuilders: ChannelBuilder[] = [];
+ private readonly decorators: ChannelDecorator[] = [];
+
+ constructor(host: string, port: number) {
+ this.host = host;
+ this.port = port;
+ }
+
+ addManagedChannelBuilder(builder: ChannelBuilder): this {
+ this.channelBuilders.push(builder);
+ return this;
+ }
+
+ addChannelDecorator(decorator: ChannelDecorator): this {
+ this.decorators.push(decorator);
+ return this;
+ }
+
+ build(): GRPCChannel {
+ return GRPCChannel.create(this.host, this.port, this.channelBuilders, this.decorators);
+ }
+}
diff --git a/src/agent/protocol/grpc/clients/Client.ts b/src/agent/core/remote/GRPCChannelListener.ts
similarity index 84%
copy from src/agent/protocol/grpc/clients/Client.ts
copy to src/agent/core/remote/GRPCChannelListener.ts
index 8d8fd06..34febae 100644
--- a/src/agent/protocol/grpc/clients/Client.ts
+++ b/src/agent/core/remote/GRPCChannelListener.ts
@@ -17,12 +17,8 @@
*
*/
-export default interface Client {
- readonly isConnected: boolean;
+import { GRPCChannelStatus } from './GRPCChannelStatus';
- start(): void;
-
- flush(): Promise<any> | null;
-
- destroy?(): void;
+export interface GRPCChannelListener {
+ statusChanged(status: GRPCChannelStatus): void;
}
diff --git a/src/agent/core/remote/GRPCChannelManager.ts b/src/agent/core/remote/GRPCChannelManager.ts
new file mode 100644
index 0000000..5eb2fed
--- /dev/null
+++ b/src/agent/core/remote/GRPCChannelManager.ts
@@ -0,0 +1,203 @@
+/*!
+ *
+ * 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 * as grpc from '@grpc/grpc-js';
+import { ClientOptions } from '@grpc/grpc-js';
+import config from '../../../config/AgentConfig';
+import { createLogger } from '../../../logging';
+import AgentIDDecorator from './AgentIDDecorator';
+import AuthenticationDecorator from './AuthenticationDecorator';
+import GRPCChannel from './GRPCChannel';
+import { GRPCChannelListener } from './GRPCChannelListener';
+import { GRPCChannelStatus } from './GRPCChannelStatus';
+import BootService from '../boot/BootService';
+import StandardChannelBuilder from './StandardChannelBuilder';
+import TLSChannelBuilder from './TLSChannelBuilder';
+
+const logger = createLogger(__filename);
+
+function isGrpcNetworkError(error: unknown): boolean {
+ const code = (error as grpc.ServiceError | undefined)?.code;
+
+ return (
+ code === grpc.status.UNAVAILABLE ||
+ code === grpc.status.PERMISSION_DENIED ||
+ code === grpc.status.UNAUTHENTICATED ||
+ code === grpc.status.RESOURCE_EXHAUSTED ||
+ code === grpc.status.UNKNOWN
+ );
+}
+
+/**
+ * Shared gRPC channel manager (Java GRPCChannelManager skeleton).
+ * V1: single address; V2 reserved: multi-address failover via reportError().
+ */
+export default class GRPCChannelManager implements BootService {
+ private managedChannel: GRPCChannel | null = null;
+ private readonly listeners: GRPCChannelListener[] = [];
+ private lastStatus: GRPCChannelStatus | null = null;
+ private closed = false;
+
+ /** V1: first address when comma-separated; V2: failover selection. */
+ resolveAddress(): string {
+ const raw = config.collectorAddress ?? '';
+ const first = raw.split(',')[0]?.trim();
+ if (!first) {
+ throw new Error('collectorAddress is not configured');
+ }
+ return first;
+ }
+
+ getChannel(): grpc.Channel {
+ if (!this.managedChannel) {
+ throw new Error('gRPC channel is not available');
+ }
+
+ return this.managedChannel.getChannel();
+ }
+
+ getClientOptions(): ClientOptions {
+ if (!this.managedChannel) {
+ throw new Error('gRPC channel is not available');
+ }
+
+ return this.managedChannel.getClientOptions();
+ }
+
+ isConnected(): boolean {
+ return this.managedChannel?.isConnected(true) ?? false;
+ }
+
+ addChannelListener(listener: GRPCChannelListener): void {
+ this.listeners.push(listener);
+ if (this.lastStatus !== null) {
+ listener.statusChanged(this.lastStatus);
+ }
+ }
+
+ priority(): number {
+ return Number.MAX_SAFE_INTEGER;
+ }
+
+ /** Align local status with grpc-js connectivity; avoid permanent DISCONNECT while channel stays READY. */
+ reportError(error: unknown): void {
+ if (!isGrpcNetworkError(error)) {
+ logger.debug('gRPC report error (ignored): %s', error);
+ return;
+ }
+
+ const managed = this.managedChannel;
+ if (!managed || this.closed) {
+ this.notify(GRPCChannelStatus.DISCONNECT);
+ return;
+ }
+
+ if (managed.isConnected(false)) {
+ logger.debug('gRPC network error but channel still connected: %s', error);
+ this.notify(GRPCChannelStatus.CONNECTED);
+ return;
+ }
+
+ logger.debug('gRPC network error, notify DISCONNECT: %s', error);
+ this.notify(GRPCChannelStatus.DISCONNECT);
+ }
+
+ prepare(): void {}
+
+ boot(): void {
+ this.closed = false;
+ const address = this.resolveAddress();
+ const [host, portText] = address.split(':');
+ const port = Number.parseInt(portText, 10);
+
+ if (!host || Number.isNaN(port)) {
+ throw new Error(`Invalid collector address: ${address}`);
+ }
+
+ this.managedChannel = GRPCChannel.newBuilder(host, port)
+ .addManagedChannelBuilder(new StandardChannelBuilder())
+ .addManagedChannelBuilder(new TLSChannelBuilder())
+ .addChannelDecorator(new AgentIDDecorator())
+ .addChannelDecorator(new AuthenticationDecorator())
+ .build();
+
+ this.watchConnectivityState();
+ this.notifyCurrentConnectivityState(true);
+ }
+
+ onComplete(): void {}
+
+ shutdown(): void {
+ this.closed = true;
+ const managed = this.managedChannel;
+ this.managedChannel = null;
+ managed?.shutdownNow();
+ this.notify(GRPCChannelStatus.DISCONNECT);
+ this.listeners.length = 0;
+ }
+
+ private watchConnectivityState(): void {
+ const managed = this.managedChannel;
+ if (this.closed || !managed) {
+ return;
+ }
+
+ const channel = managed.getChannel();
+ const currentState = channel.getConnectivityState(true);
+
+ channel.watchConnectivityState(currentState, Infinity, (error) => {
+ if (this.closed || this.managedChannel !== managed) {
+ return;
+ }
+
+ if (error) {
+ logger.debug('Channel connectivity watch stopped: %s', error.message);
+ return;
+ }
+
+ this.notifyCurrentConnectivityState(false);
+ this.watchConnectivityState();
+ });
+ }
+
+ private notifyCurrentConnectivityState(requestConnection: boolean): void {
+ const managed = this.managedChannel;
+ if (this.closed || !managed) {
+ return;
+ }
+
+ const channel = managed.getChannel();
+ const ready = channel.getConnectivityState(requestConnection) === grpc.connectivityState.READY;
+ this.notify(ready ? GRPCChannelStatus.CONNECTED : GRPCChannelStatus.DISCONNECT);
+ }
+
+ private notify(status: GRPCChannelStatus): void {
+ if (this.lastStatus === status) {
+ return;
+ }
+ this.lastStatus = status;
+ for (const listener of this.listeners) {
+ try {
+ listener.statusChanged(status);
+ } catch (err) {
+ logger.error('GRPCChannelListener failed: %s', err);
+ }
+ }
+ }
+}
diff --git a/src/agent/protocol/grpc/clients/Client.ts b/src/agent/core/remote/GRPCChannelStatus.ts
similarity index 84%
rename from src/agent/protocol/grpc/clients/Client.ts
rename to src/agent/core/remote/GRPCChannelStatus.ts
index 8d8fd06..5200d58 100644
--- a/src/agent/protocol/grpc/clients/Client.ts
+++ b/src/agent/core/remote/GRPCChannelStatus.ts
@@ -17,12 +17,7 @@
*
*/
-export default interface Client {
- readonly isConnected: boolean;
-
- start(): void;
-
- flush(): Promise<any> | null;
-
- destroy?(): void;
+export enum GRPCChannelStatus {
+ CONNECTED = 'CONNECTED',
+ DISCONNECT = 'DISCONNECT',
}
diff --git a/src/agent/core/remote/ServiceManagementClient.ts b/src/agent/core/remote/ServiceManagementClient.ts
new file mode 100644
index 0000000..a558caf
--- /dev/null
+++ b/src/agent/core/remote/ServiceManagementClient.ts
@@ -0,0 +1,164 @@
+/*!
+ *
+ * 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 * as grpc from '@grpc/grpc-js';
+import * as os from 'os';
+import * as packageInfo from '../../../../package.json';
+import config from '../../../config/AgentConfig';
+import { createLogger, throttled } from '../../../logging';
+import BootService from '../boot/BootService';
+import ServiceManager from '../boot/ServiceManager';
+import { ManagementServiceClient } from '../../../proto/management/Management_grpc_pb';
+import { InstancePingPkg, InstanceProperties } from '../../../proto/management/Management_pb';
+import { KeyStringValuePair } from '../../../proto/common/Common_pb';
+import GRPCChannelManager from './GRPCChannelManager';
+import { GRPCChannelListener } from './GRPCChannelListener';
+import { GRPCChannelStatus } from './GRPCChannelStatus';
+
+const logger = createLogger(__filename);
+const logHeartbeatError = throttled(logger, 'error', 30000);
+
+export default class ServiceManagementClient implements BootService, GRPCChannelListener {
+ private closed = false;
+ private channelManager?: GRPCChannelManager;
+ private status = GRPCChannelStatus.DISCONNECT;
+ private managementServiceClient?: ManagementServiceClient;
+ private heartbeatTimer?: NodeJS.Timeout;
+ private keepAlivePkg?: InstancePingPkg;
+ private instanceProperties?: InstanceProperties;
+ private sendPropertiesCounter = 0;
+
+ /** Same default as Java Config.Collector.PROPERTIES_REPORT_PERIOD_FACTOR (10). */
+ private static readonly PROPERTIES_REPORT_PERIOD_FACTOR = 10;
+
+ prepare(): void {
+ this.channelManager = ServiceManager.INSTANCE.findService(GRPCChannelManager);
+ this.channelManager?.addChannelListener(this);
+ }
+
+ boot(): void {
+ if (this.heartbeatTimer) {
+ logger.warn(`
+ The heartbeat timer has already been scheduled,
+ this may be a potential bug, please consider reporting
+ this to ${packageInfo.bugs.url}
+ `);
+ return;
+ }
+
+ this.keepAlivePkg = new InstancePingPkg().setService(config.serviceName).setServiceinstance(config.serviceInstance);
+
+ this.instanceProperties = new InstanceProperties()
+ .setService(config.serviceName)
+ .setServiceinstance(config.serviceInstance)
+ .setPropertiesList([
+ new KeyStringValuePair().setKey('language').setValue('NodeJS'),
+ new KeyStringValuePair().setKey('OS Name').setValue(os.platform()),
+ new KeyStringValuePair().setKey('hostname').setValue(os.hostname()),
+ new KeyStringValuePair().setKey('Process No.').setValue(`${process.pid}`),
+ ]);
+
+ this.heartbeatTimer = setInterval(() => this.sendHeartbeat(), 20000) as NodeJS.Timeout;
+ this.heartbeatTimer.unref();
+ }
+
+ onComplete(): void {}
+
+ shutdown(): void {
+ this.closed = true;
+ if (this.heartbeatTimer) {
+ clearInterval(this.heartbeatTimer);
+ this.heartbeatTimer = undefined;
+ }
+ this.managementServiceClient = undefined;
+ this.channelManager = undefined;
+ logger.info('ServiceManagementClient destroyed and resources cleaned up');
+ }
+
+ priority(): number {
+ return 0;
+ }
+
+ statusChanged(status: GRPCChannelStatus): void {
+ this.status = status;
+ this.managementServiceClient = status === GRPCChannelStatus.CONNECTED ? this.createManagementClient() : undefined;
+ }
+
+ private sendHeartbeat(): void {
+ if (
+ this.closed ||
+ this.status !== GRPCChannelStatus.CONNECTED ||
+ !this.managementServiceClient ||
+ !this.instanceProperties ||
+ !this.keepAlivePkg
+ ) {
+ return;
+ }
+
+ const options = { deadline: Date.now() + config.traceTimeout };
+ const reportProperties =
+ Math.abs(this.sendPropertiesCounter++) % ServiceManagementClient.PROPERTIES_REPORT_PERIOD_FACTOR === 0;
+
+ if (reportProperties) {
+ this.managementServiceClient.reportInstanceProperties(
+ this.instanceProperties,
+ new grpc.Metadata(),
+ options,
+ (error) => {
+ if (error) {
+ logHeartbeatError('Failed to send instance properties', error);
+ this.reportGrpcError(error);
+ }
+ },
+ );
+ return;
+ }
+
+ this.managementServiceClient.keepAlive(this.keepAlivePkg, new grpc.Metadata(), options, (error) => {
+ if (error) {
+ logHeartbeatError('Failed to send heartbeat', error);
+ this.reportGrpcError(error);
+ }
+ });
+ }
+
+ private createManagementClient(): ManagementServiceClient | undefined {
+ if (!this.channelManager) {
+ return undefined;
+ }
+
+ return new ManagementServiceClient(
+ config.collectorAddress,
+ grpc.credentials.createInsecure(),
+ this.channelManager.getClientOptions(),
+ );
+ }
+ private reportGrpcError(error: unknown): void {
+ if (this.closed) {
+ return;
+ }
+
+ this.channelManager?.reportError(error);
+ }
+
+ flush(): Promise<unknown> | null {
+ logger.warn('ServiceManagementClient does not need flush().');
+ return null;
+ }
+}
diff --git a/src/agent/protocol/grpc/AuthInterceptor.ts b/src/agent/core/remote/StandardChannelBuilder.ts
similarity index 61%
copy from src/agent/protocol/grpc/AuthInterceptor.ts
copy to src/agent/core/remote/StandardChannelBuilder.ts
index 0f5ae4d..ae37f24 100644
--- a/src/agent/protocol/grpc/AuthInterceptor.ts
+++ b/src/agent/core/remote/StandardChannelBuilder.ts
@@ -18,12 +18,19 @@
*/
import * as grpc from '@grpc/grpc-js';
-import config from '../../../config/AgentConfig';
+import ChannelBuilder, { ChannelBuildContext } from './ChannelBuilder';
-export default function AuthInterceptor() {
- const mata = new grpc.Metadata();
- if (config.authorization) {
- mata.add('Authentication', config.authorization);
+const MAX_INBOUND_MESSAGE_SIZE = 1024 * 1024 * 50;
+
+export default class StandardChannelBuilder implements ChannelBuilder {
+ build(context: ChannelBuildContext): ChannelBuildContext {
+ return {
+ credentials: grpc.credentials.createInsecure(),
+ options: {
+ ...context.options,
+ 'grpc.max_receive_message_length': MAX_INBOUND_MESSAGE_SIZE,
+ 'grpc.max_send_message_length': MAX_INBOUND_MESSAGE_SIZE,
+ },
+ };
}
- return mata;
}
diff --git a/src/agent/protocol/grpc/AuthInterceptor.ts b/src/agent/core/remote/TLSChannelBuilder.ts
similarity index 66%
copy from src/agent/protocol/grpc/AuthInterceptor.ts
copy to src/agent/core/remote/TLSChannelBuilder.ts
index 0f5ae4d..d3d7e51 100644
--- a/src/agent/protocol/grpc/AuthInterceptor.ts
+++ b/src/agent/core/remote/TLSChannelBuilder.ts
@@ -19,11 +19,17 @@
import * as grpc from '@grpc/grpc-js';
import config from '../../../config/AgentConfig';
+import ChannelBuilder, { ChannelBuildContext } from './ChannelBuilder';
-export default function AuthInterceptor() {
- const mata = new grpc.Metadata();
- if (config.authorization) {
- mata.add('Authentication', config.authorization);
+/** When SW_AGENT_SECURE=true, upgrade channel credentials to TLS (Java TLSChannelBuilder simplified). */
+export default class TLSChannelBuilder implements ChannelBuilder {
+ build(context: ChannelBuildContext): ChannelBuildContext {
+ if (config.secure) {
+ return {
+ ...context,
+ credentials: grpc.credentials.createSsl(),
+ };
+ }
+ return context;
}
- return mata;
}
diff --git a/src/agent/core/remote/TraceSegmentServiceClient.ts b/src/agent/core/remote/TraceSegmentServiceClient.ts
new file mode 100644
index 0000000..3f1812d
--- /dev/null
+++ b/src/agent/core/remote/TraceSegmentServiceClient.ts
@@ -0,0 +1,231 @@
+/*!
+ *
+ * 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 config from '../../../config/AgentConfig';
+import * as grpc from '@grpc/grpc-js';
+import { createLogger, throttled } from '../../../logging';
+import BootService from '../boot/BootService';
+import ServiceManager from '../boot/ServiceManager';
+import { TraceSegmentReportServiceClient } from '../../../proto/language-agent/Tracing_grpc_pb';
+import { emitter } from '../../../lib/EventEmitter';
+import Segment from '../../../trace/context/Segment';
+import GRPCChannelManager from './GRPCChannelManager';
+import { GRPCChannelListener } from './GRPCChannelListener';
+import { GRPCChannelStatus } from './GRPCChannelStatus';
+
+const logger = createLogger(__filename);
+const logReportError = throttled(logger, 'error', 30000);
+const logBufferFull = throttled(logger, 'warn', 30000);
+
+export default class TraceSegmentServiceClient implements BootService, GRPCChannelListener {
+ private closed = false;
+ private channelManager?: GRPCChannelManager;
+ private status = GRPCChannelStatus.DISCONNECT;
+ private reporterClient?: TraceSegmentReportServiceClient;
+ private readonly buffer: Segment[] = [];
+ private timeout?: NodeJS.Timeout;
+ private reporting?: Promise<void>;
+ private segmentFinishedListener?: (segment: Segment) => void;
+
+ prepare(): void {
+ this.channelManager = ServiceManager.INSTANCE.findService(GRPCChannelManager);
+ this.channelManager?.addChannelListener(this);
+
+ if (this.segmentFinishedListener) {
+ emitter.off('segment-finished', this.segmentFinishedListener);
+ }
+
+ this.segmentFinishedListener = (segment: Segment) => {
+ if (this.closed) {
+ return;
+ }
+
+ if (this.buffer.length >= config.maxBufferSize) {
+ logBufferFull(
+ `Trace buffer reached maximum size (${config.maxBufferSize}); discarding oldest segments. The collector at ${config.collectorAddress} is likely unreachable.`,
+ );
+ this.buffer.shift();
+ }
+
+ this.buffer.push(segment);
+ this.timeout?.ref();
+ };
+
+ emitter.on('segment-finished', this.segmentFinishedListener);
+ }
+
+ boot(): void {
+ this.scheduleNextReport();
+ }
+
+ onComplete(): void {}
+
+ shutdown(): void {
+ this.closed = true;
+ if (this.segmentFinishedListener) {
+ emitter.off('segment-finished', this.segmentFinishedListener);
+ }
+
+ if (this.timeout) {
+ clearTimeout(this.timeout);
+ this.timeout = undefined;
+ }
+
+ this.reporting = undefined;
+ this.reporterClient = undefined;
+ this.buffer.length = 0;
+ this.channelManager = undefined;
+ logger.info('TraceSegmentServiceClient destroyed and resources cleaned up');
+ }
+
+ priority(): number {
+ return 0;
+ }
+
+ statusChanged(status: GRPCChannelStatus): void {
+ this.status = status;
+ this.reporterClient = status === GRPCChannelStatus.CONNECTED ? this.createReporterClient() : undefined;
+ }
+
+ private createReporterClient(): TraceSegmentReportServiceClient | undefined {
+ if (!this.channelManager) {
+ return undefined;
+ }
+
+ return new TraceSegmentReportServiceClient(
+ config.collectorAddress,
+ grpc.credentials.createInsecure(),
+ this.channelManager.getClientOptions(),
+ );
+ }
+
+ private scheduleNextReport(): void {
+ if (this.closed || this.timeout) {
+ return;
+ }
+
+ this.timeout = setTimeout(() => {
+ this.timeout = undefined;
+ if (this.closed) {
+ return;
+ }
+ void this.reportOnce().finally(() => this.scheduleNextReport());
+ }, 1000) as unknown as NodeJS.Timeout;
+ this.timeout.unref();
+ }
+
+ private reportOnce(): Promise<void> {
+ if (this.closed) {
+ return Promise.resolve();
+ }
+
+ if (this.reporting) {
+ return this.reporting;
+ }
+
+ this.reporting = this.doReport().finally(() => {
+ this.reporting = undefined;
+ });
+
+ return this.reporting;
+ }
+
+ private doReport(): Promise<void> {
+ return new Promise((resolve) => {
+ if (this.closed) {
+ resolve();
+ return;
+ }
+
+ emitter.emit('segments-sent');
+
+ if (this.buffer.length === 0) {
+ resolve();
+ return;
+ }
+
+ if (this.status !== GRPCChannelStatus.CONNECTED || !this.reporterClient) {
+ resolve();
+ return;
+ }
+
+ let stream: ReturnType<TraceSegmentReportServiceClient['collect']> | undefined;
+ try {
+ stream = this.reporterClient.collect(
+ new grpc.Metadata(),
+ { deadline: Date.now() + config.traceTimeout },
+ (error) => {
+ if (error) {
+ logReportError('Failed to report trace data', error);
+ this.reportGrpcError(error);
+ }
+ resolve();
+ },
+ );
+
+ for (const segment of this.buffer) {
+ if (segment) {
+ if (logger._isDebugEnabled) {
+ logger.debug('Sending segment ', { segment });
+ }
+ stream.write(segment.transform());
+ }
+ }
+ } catch (error) {
+ logReportError('Failed to report trace data', error);
+ this.reportGrpcError(error);
+ resolve();
+ } finally {
+ this.buffer.length = 0;
+ try {
+ stream?.end();
+ } catch (error) {
+ logReportError('Failed to end trace collect stream', error);
+ resolve();
+ }
+ }
+ });
+ }
+
+ private reportGrpcError(error: unknown): void {
+ if (this.closed) {
+ return;
+ }
+
+ this.channelManager?.reportError(error);
+ }
+
+ flush(): Promise<unknown> | null {
+ if (this.closed) {
+ return null;
+ }
+
+ if (this.timeout) {
+ clearTimeout(this.timeout);
+ this.timeout = undefined;
+ }
+
+ if (this.buffer.length === 0) {
+ this.scheduleNextReport();
+ return null;
+ }
+
+ return this.reportOnce().finally(() => this.scheduleNextReport());
+ }
+}
diff --git a/src/agent/protocol/Protocol.ts b/src/agent/protocol/Protocol.ts
deleted file mode 100644
index 062318a..0000000
--- a/src/agent/protocol/Protocol.ts
+++ /dev/null
@@ -1,33 +0,0 @@
-/*!
- *
- * 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.
- *
- */
-
-/**
- * The transport protocol between the agent and the backend (OAP).
- */
-export default interface Protocol {
- isConnected: boolean;
-
- heartbeat(): this;
-
- report(): this;
-
- flush(): Promise<any> | null;
-
- destroy?(): void;
-}
diff --git a/src/agent/protocol/grpc/GrpcProtocol.ts b/src/agent/protocol/grpc/GrpcProtocol.ts
deleted file mode 100644
index f151141..0000000
--- a/src/agent/protocol/grpc/GrpcProtocol.ts
+++ /dev/null
@@ -1,56 +0,0 @@
-/*!
- *
- * 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 Protocol from '../../../agent/protocol/Protocol';
-import HeartbeatClient from '../../../agent/protocol/grpc/clients/HeartbeatClient';
-import TraceReportClient from '../../../agent/protocol/grpc/clients/TraceReportClient';
-
-export default class GrpcProtocol implements Protocol {
- private readonly heartbeatClient: HeartbeatClient;
- private readonly traceReportClient: TraceReportClient;
-
- constructor() {
- this.heartbeatClient = new HeartbeatClient();
- this.traceReportClient = new TraceReportClient();
- }
-
- get isConnected(): boolean {
- return this.heartbeatClient.isConnected && this.traceReportClient.isConnected;
- }
-
- heartbeat(): this {
- this.heartbeatClient.start();
- return this;
- }
-
- report(): this {
- this.traceReportClient.start();
- return this;
- }
-
- flush(): Promise<any> | null {
- return this.traceReportClient.flush();
- }
-
- destroy(): void {
- // Clean up both clients to prevent memory leaks
- this.heartbeatClient.destroy?.();
- this.traceReportClient.destroy?.();
- }
-}
diff --git a/src/agent/protocol/grpc/SegmentObjectAdapter.ts b/src/agent/protocol/grpc/SegmentObjectAdapter.ts
deleted file mode 100644
index 6559f20..0000000
--- a/src/agent/protocol/grpc/SegmentObjectAdapter.ts
+++ /dev/null
@@ -1,74 +0,0 @@
-/*!
- *
- * 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 config from '../../../config/AgentConfig';
-import { KeyStringValuePair } from '../../../proto/common/Common_pb';
-import Segment from '../../../trace/context/Segment';
-import { Log, RefType, SegmentObject, SegmentReference, SpanObject } from '../../../proto/language-agent/Tracing_pb';
-
-/**
- * An adapter that adapts {@link Segment} objects to gRPC object {@link SegmentObject}.
- */
-export default class SegmentObjectAdapter extends SegmentObject {
- constructor(segment: Segment) {
- super();
- super
- .setService(config.serviceName)
- .setServiceinstance(config.serviceInstance)
- .setTraceid(segment.relatedTraces[0].toString())
- .setTracesegmentid(segment.segmentId.toString())
- .setSpansList(
- segment.spans.map((span) =>
- new SpanObject()
- .setSpanid(span.id)
- .setParentspanid(span.parentId)
- .setStarttime(span.startTime)
- .setEndtime(span.endTime)
- .setOperationname(span.operation)
- .setPeer(span.peer)
- .setSpantype(span.type)
- .setSpanlayer(span.layer)
- .setComponentid(span.component.id)
- .setIserror(span.errored)
- .setLogsList(
- span.logs.map((log) =>
- new Log()
- .setTime(log.timestamp)
- .setDataList(
- log.items.map((logItem) => new KeyStringValuePair().setKey(logItem.key).setValue(logItem.val)),
- ),
- ),
- )
- .setTagsList(span.tags.map((tag) => new KeyStringValuePair().setKey(tag.key).setValue(tag.val)))
- .setRefsList(
- span.refs.map((ref) =>
- new SegmentReference()
- .setReftype(RefType.CROSSPROCESS)
- .setTraceid(ref.traceId.toString())
- .setParenttracesegmentid(ref.segmentId.toString())
- .setParentspanid(ref.spanId)
- .setParentservice(ref.service)
- .setParentserviceinstance(ref.serviceInstance)
- .setNetworkaddressusedatpeer(ref.clientAddress),
- ),
- ),
- ),
- );
- }
-}
diff --git a/src/agent/protocol/grpc/clients/HeartbeatClient.ts b/src/agent/protocol/grpc/clients/HeartbeatClient.ts
deleted file mode 100755
index 2c101a6..0000000
--- a/src/agent/protocol/grpc/clients/HeartbeatClient.ts
+++ /dev/null
@@ -1,103 +0,0 @@
-/*!
- *
- * 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 * as grpc from '@grpc/grpc-js';
-import { connectivityState } from '@grpc/grpc-js';
-
-import * as packageInfo from '../../../../../package.json';
-import { createLogger, throttled } from '../../../../logging';
-import Client from './Client';
-import { ManagementServiceClient } from '../../../../proto/management/Management_grpc_pb';
-import AuthInterceptor from '../AuthInterceptor';
-import { InstancePingPkg, InstanceProperties } from '../../../../proto/management/Management_pb';
-import config from '../../../../config/AgentConfig';
-import { KeyStringValuePair } from '../../../../proto/common/Common_pb';
-import * as os from 'os';
-
-const logger = createLogger(__filename);
-const logHeartbeatError = throttled(logger, 'error', 30000);
-
-export default class HeartbeatClient implements Client {
- private readonly managementServiceClient: ManagementServiceClient;
- private heartbeatTimer?: NodeJS.Timeout;
-
- constructor() {
- this.managementServiceClient = new ManagementServiceClient(
- config.collectorAddress,
- config.secure ? grpc.credentials.createSsl() : grpc.credentials.createInsecure(),
- );
- }
-
- get isConnected(): boolean {
- return this.managementServiceClient.getChannel().getConnectivityState(true) === connectivityState.READY;
- }
-
- start() {
- if (this.heartbeatTimer) {
- logger.warn(`
- The heartbeat timer has already been scheduled,
- this may be a potential bug, please consider reporting
- this to ${packageInfo.bugs.url}
- `);
- return;
- }
-
- const keepAlivePkg = new InstancePingPkg()
- .setService(config.serviceName)
- .setServiceinstance(config.serviceInstance);
-
- const instanceProperties = new InstanceProperties()
- .setService(config.serviceName)
- .setServiceinstance(config.serviceInstance)
- .setPropertiesList([
- new KeyStringValuePair().setKey('language').setValue('NodeJS'),
- new KeyStringValuePair().setKey('OS Name').setValue(os.platform()),
- new KeyStringValuePair().setKey('hostname').setValue(os.hostname()),
- new KeyStringValuePair().setKey('Process No.').setValue(`${process.pid}`),
- ]);
-
- this.heartbeatTimer = setInterval(() => {
- this.managementServiceClient.reportInstanceProperties(instanceProperties, AuthInterceptor(), (error, _) => {
- if (error) {
- logHeartbeatError('Failed to send heartbeat', error);
- }
- });
- this.managementServiceClient.keepAlive(keepAlivePkg, AuthInterceptor(), (error, _) => {
- if (error) {
- logHeartbeatError('Failed to send heartbeat', error);
- }
- });
- }, 20000).unref();
- }
-
- flush(): Promise<any> | null {
- logger.warn('HeartbeatClient does not need flush().');
- return null;
- }
-
- destroy(): void {
- // Clear heartbeat timer to prevent memory leak
- if (this.heartbeatTimer) {
- clearInterval(this.heartbeatTimer);
- this.heartbeatTimer = undefined;
- }
-
- logger.info('HeartbeatClient destroyed and resources cleaned up');
- }
-}
diff --git a/src/agent/protocol/grpc/clients/TraceReportClient.ts b/src/agent/protocol/grpc/clients/TraceReportClient.ts
deleted file mode 100755
index 4f5cfe7..0000000
--- a/src/agent/protocol/grpc/clients/TraceReportClient.ts
+++ /dev/null
@@ -1,152 +0,0 @@
-/*!
- *
- * 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 config from '../../../../config/AgentConfig';
-import * as grpc from '@grpc/grpc-js';
-import { connectivityState } from '@grpc/grpc-js';
-import { createLogger, throttled } from '../../../../logging';
-import Client from './Client';
-import { TraceSegmentReportServiceClient } from '../../../../proto/language-agent/Tracing_grpc_pb';
-import AuthInterceptor from '../AuthInterceptor';
-import SegmentObjectAdapter from '../SegmentObjectAdapter';
-import { emitter } from '../../../../lib/EventEmitter';
-import Segment from '../../../../trace/context/Segment';
-
-const logger = createLogger(__filename);
-const logReportError = throttled(logger, 'error', 30000);
-const logBufferFull = throttled(logger, 'warn', 30000);
-
-export default class TraceReportClient implements Client {
- private readonly reporterClient: TraceSegmentReportServiceClient;
- private readonly buffer: Segment[] = [];
- private timeout?: NodeJS.Timeout;
- private segmentFinishedListener: (segment: Segment) => void;
-
- constructor() {
- this.reporterClient = new TraceSegmentReportServiceClient(
- config.collectorAddress,
- config.secure ? grpc.credentials.createSsl() : grpc.credentials.createInsecure(),
- );
-
- // Store listener reference for cleanup
- this.segmentFinishedListener = (segment: Segment) => {
- // Limit buffer size to prevent memory leak during network issues
- if (this.buffer.length >= config.maxBufferSize) {
- logBufferFull(
- `Trace buffer reached maximum size (${config.maxBufferSize}); discarding oldest segments. The collector at ${config.collectorAddress} is likely unreachable.`,
- );
- this.buffer.shift(); // Remove oldest segment
- }
-
- this.buffer.push(segment);
- this.timeout?.ref();
- };
-
- emitter.on('segment-finished', this.segmentFinishedListener);
- }
-
- get isConnected(): boolean {
- return this.reporterClient?.getChannel().getConnectivityState(true) === connectivityState.READY;
- }
-
- private reportFunction(callback?: any) {
- emitter.emit('segments-sent'); // reset limiter in SpanContext
-
- try {
- if (this.buffer.length === 0) {
- if (callback) callback();
-
- return;
- }
-
- // Collector unreachable: keep the (bounded) buffer and let gRPC reconnect with its own exponential
- // backoff, instead of failing a stream every tick and logging an error storm that exhausts the heap.
- // The channel keeps trying to connect because `isConnected` polls getConnectivityState(true).
- if (!this.isConnected) {
- if (callback) callback();
-
- return;
- }
-
- const stream = this.reporterClient.collect(
- AuthInterceptor(),
- { deadline: Date.now() + config.traceTimeout },
- (error, _) => {
- if (error) {
- logReportError('Failed to report trace data', error);
- }
-
- if (callback) callback();
- },
- );
-
- try {
- for (const segment of this.buffer) {
- if (segment) {
- if (logger._isDebugEnabled) {
- logger.debug('Sending segment ', { segment });
- }
-
- stream.write(new SegmentObjectAdapter(segment));
- }
- }
- } finally {
- this.buffer.length = 0;
- }
-
- stream.end();
- } finally {
- this.timeout = setTimeout(this.reportFunction.bind(this), 1000).unref();
- }
- }
-
- start() {
- this.timeout = setTimeout(this.reportFunction.bind(this), 1000).unref();
- }
-
- flush(): Promise<any> | null {
- // This function explicitly returns null instead of a resolved Promise in case of nothing to flush so that in this
- // case passing control back to the event loop can be avoided. Even a resolved Promise will run other things in
- // the event loop when it is awaited and before it continues.
-
- return this.buffer.length === 0
- ? null
- : new Promise((resolve) => {
- this.reportFunction(resolve);
- });
- }
-
- destroy(): void {
- // Clean up event listener to prevent memory leak
- if (this.segmentFinishedListener) {
- emitter.off('segment-finished', this.segmentFinishedListener);
- }
-
- // Clear timeout
- if (this.timeout) {
- clearTimeout(this.timeout);
- this.timeout = undefined;
- }
-
- // Clear buffer
- this.buffer.length = 0;
-
- logger.info('TraceReportClient destroyed and resources cleaned up');
- }
-}
diff --git a/src/config/AgentConfig.ts b/src/config/AgentConfig.ts
index 2ad5905..54d8a34 100644
--- a/src/config/AgentConfig.ts
+++ b/src/config/AgentConfig.ts
@@ -43,9 +43,120 @@
reIgnoreOperation?: RegExp;
reHttpIgnoreMethod?: RegExp;
traceTimeout?: number;
+ runtimeMetricsReporterActive?: boolean;
+ runtimeMetricsCollectPeriod?: number;
+ runtimeMetricsReportPeriod?: number;
+ runtimeMetricsBufferSize?: number;
+ /** @deprecated use runtimeMetricsReporterActive */
+ nvmMetricsReporterActive?: boolean;
+ /** @deprecated use runtimeMetricsCollectPeriod */
+ nvmMetricsCollectPeriod?: number;
+ /** @deprecated use runtimeMetricsReportPeriod */
+ nvmMetricsReportPeriod?: number;
+ /** @deprecated use runtimeMetricsBufferSize */
+ nvmMetricsBufferSize?: number;
+ /** @deprecated use runtimeMetricsReporterActive */
+ nvmJvmReporterActive?: boolean;
+ /** @deprecated use runtimeMetricsCollectPeriod */
+ nvmJvmMetricsCollectPeriod?: number;
+ /** @deprecated use runtimeMetricsReportPeriod */
+ nvmJvmMetricsReportPeriod?: number;
+ /** @deprecated use runtimeMetricsBufferSize */
+ nvmJvmMetricsBufferSize?: number;
};
-export function finalizeConfig(config: AgentConfig): void {
+export function normalizeDeprecatedRuntimeMetricOptions(options: AgentConfig): AgentConfig {
+ const normalized = { ...options };
+
+ if (normalized.runtimeMetricsReporterActive === undefined) {
+ const reporterActive = normalized.nvmMetricsReporterActive ?? normalized.nvmJvmReporterActive;
+ if (reporterActive !== undefined) {
+ normalized.runtimeMetricsReporterActive = reporterActive;
+ }
+ }
+ delete normalized.nvmMetricsReporterActive;
+ delete normalized.nvmJvmReporterActive;
+
+ if (normalized.runtimeMetricsCollectPeriod === undefined) {
+ const collectPeriod = normalized.nvmMetricsCollectPeriod ?? normalized.nvmJvmMetricsCollectPeriod;
+ if (collectPeriod !== undefined) {
+ normalized.runtimeMetricsCollectPeriod = collectPeriod;
+ }
+ }
+ delete normalized.nvmMetricsCollectPeriod;
+ delete normalized.nvmJvmMetricsCollectPeriod;
+
+ if (normalized.runtimeMetricsReportPeriod === undefined) {
+ const reportPeriod = normalized.nvmMetricsReportPeriod ?? normalized.nvmJvmMetricsReportPeriod;
+ if (reportPeriod !== undefined) {
+ normalized.runtimeMetricsReportPeriod = reportPeriod;
+ }
+ }
+ delete normalized.nvmMetricsReportPeriod;
+ delete normalized.nvmJvmMetricsReportPeriod;
+
+ if (normalized.runtimeMetricsBufferSize === undefined) {
+ const bufferSize = normalized.nvmMetricsBufferSize ?? normalized.nvmJvmMetricsBufferSize;
+ if (bufferSize !== undefined) {
+ normalized.runtimeMetricsBufferSize = bufferSize;
+ }
+ }
+ delete normalized.nvmMetricsBufferSize;
+ delete normalized.nvmJvmMetricsBufferSize;
+
+ return normalized;
+}
+
+function clearDeprecatedRuntimeMetricFields(config: AgentConfig): void {
+ delete config.nvmMetricsReporterActive;
+ delete config.nvmJvmReporterActive;
+ delete config.nvmMetricsCollectPeriod;
+ delete config.nvmJvmMetricsCollectPeriod;
+ delete config.nvmMetricsReportPeriod;
+ delete config.nvmJvmMetricsReportPeriod;
+ delete config.nvmMetricsBufferSize;
+ delete config.nvmJvmMetricsBufferSize;
+}
+
+function applyDeprecatedRuntimeMetricConfig(config: AgentConfig, options: AgentConfig = {}): void {
+ if (options.runtimeMetricsReporterActive === undefined) {
+ if (config.nvmMetricsReporterActive !== undefined) {
+ config.runtimeMetricsReporterActive = config.nvmMetricsReporterActive;
+ } else if (config.nvmJvmReporterActive !== undefined) {
+ config.runtimeMetricsReporterActive = config.nvmJvmReporterActive;
+ }
+ }
+
+ if (options.runtimeMetricsCollectPeriod === undefined) {
+ if (config.nvmMetricsCollectPeriod !== undefined) {
+ config.runtimeMetricsCollectPeriod = config.nvmMetricsCollectPeriod;
+ } else if (config.nvmJvmMetricsCollectPeriod !== undefined) {
+ config.runtimeMetricsCollectPeriod = config.nvmJvmMetricsCollectPeriod;
+ }
+ }
+
+ if (options.runtimeMetricsReportPeriod === undefined) {
+ if (config.nvmMetricsReportPeriod !== undefined) {
+ config.runtimeMetricsReportPeriod = config.nvmMetricsReportPeriod;
+ } else if (config.nvmJvmMetricsReportPeriod !== undefined) {
+ config.runtimeMetricsReportPeriod = config.nvmJvmMetricsReportPeriod;
+ }
+ }
+
+ if (options.runtimeMetricsBufferSize === undefined) {
+ if (config.nvmMetricsBufferSize !== undefined) {
+ config.runtimeMetricsBufferSize = config.nvmMetricsBufferSize;
+ } else if (config.nvmJvmMetricsBufferSize !== undefined) {
+ config.runtimeMetricsBufferSize = config.nvmJvmMetricsBufferSize;
+ }
+ }
+
+ clearDeprecatedRuntimeMetricFields(config);
+}
+
+export function finalizeConfig(config: AgentConfig, options: AgentConfig = {}): void {
+ applyDeprecatedRuntimeMetricConfig(config, options);
+
const escapeRegExp = (s: string) => s.replace(/([.*+?^=!:${}()|\[\]\/\\])/g, '\\$1');
config.reDisablePlugins = RegExp(
@@ -140,6 +251,44 @@
traceTimeout: ((n) => (Number.isSafeInteger(n) && n > 0 ? n : 10 * 1000))(
Number.parseInt(process.env.SW_AGENT_TRACE_TIMEOUT ?? '', 10),
),
+ runtimeMetricsReporterActive: ((): boolean => {
+ const configured =
+ process.env.SW_AGENT_NODEJS_RUNTIME_METRICS_REPORTER_ACTIVE ??
+ process.env.SW_AGENT_RUNTIME_METRICS_REPORTER_ACTIVE ??
+ process.env.SW_AGENT_NVM_METRICS_REPORTER_ACTIVE ??
+ process.env.SW_AGENT_NVM_JVM_REPORTER_ACTIVE;
+ return configured?.toLowerCase() !== 'false';
+ })(),
+ runtimeMetricsCollectPeriod: ((n) => (Number.isSafeInteger(n) && n > 0 ? n : 1000))(
+ Number.parseInt(
+ process.env.SW_AGENT_NODEJS_RUNTIME_METRICS_COLLECT_PERIOD ??
+ process.env.SW_AGENT_RUNTIME_METRICS_COLLECT_PERIOD ??
+ process.env.SW_AGENT_NVM_METRICS_COLLECT_PERIOD ??
+ process.env.SW_AGENT_NVM_JVM_METRICS_COLLECT_PERIOD ??
+ '',
+ 10,
+ ),
+ ),
+ runtimeMetricsReportPeriod: ((n) => (Number.isSafeInteger(n) && n > 0 ? n : 1000))(
+ Number.parseInt(
+ process.env.SW_AGENT_NODEJS_RUNTIME_METRICS_REPORT_PERIOD ??
+ process.env.SW_AGENT_RUNTIME_METRICS_REPORT_PERIOD ??
+ process.env.SW_AGENT_NVM_METRICS_REPORT_PERIOD ??
+ process.env.SW_AGENT_NVM_JVM_METRICS_REPORT_PERIOD ??
+ '',
+ 10,
+ ),
+ ),
+ runtimeMetricsBufferSize: ((n) => (Number.isSafeInteger(n) && n > 0 ? n : 600))(
+ Number.parseInt(
+ process.env.SW_AGENT_NODEJS_RUNTIME_METRICS_BUFFER_SIZE ??
+ process.env.SW_AGENT_RUNTIME_METRICS_BUFFER_SIZE ??
+ process.env.SW_AGENT_NVM_METRICS_BUFFER_SIZE ??
+ process.env.SW_AGENT_NVM_JVM_METRICS_BUFFER_SIZE ??
+ '',
+ 10,
+ ),
+ ),
};
export default _config;
diff --git a/src/index.ts b/src/index.ts
index ae2c494..33d67a2 100644
--- a/src/index.ts
+++ b/src/index.ts
@@ -17,9 +17,8 @@
*
*/
-import config, { AgentConfig, finalizeConfig } from './config/AgentConfig';
-import Protocol from './agent/protocol/Protocol';
-import GrpcProtocol from './agent/protocol/grpc/GrpcProtocol';
+import config, { AgentConfig, finalizeConfig, normalizeDeprecatedRuntimeMetricOptions } from './config/AgentConfig';
+import ServiceManager from './agent/core/boot/ServiceManager';
import { createLogger } from './logging';
import PluginInstaller from './core/PluginInstaller';
import SpanContext from './trace/context/SpanContext';
@@ -28,7 +27,6 @@
class Agent {
private started = false;
- private protocol: Protocol | null = null;
start(options: AgentConfig = {}): void {
if (process.env.SW_DISABLE === 'true') {
@@ -41,49 +39,45 @@
return;
}
- Object.assign(config, options);
- finalizeConfig(config);
+ Object.assign(config, normalizeDeprecatedRuntimeMetricOptions(options));
+ finalizeConfig(config, options);
logger.debug('Starting SkyWalking agent');
new PluginInstaller().install();
- this.protocol = new GrpcProtocol().heartbeat().report();
+ ServiceManager.INSTANCE.boot();
this.started = true;
}
flush(): Promise<any> | null {
- if (this.protocol === null) {
+ if (!this.started) {
logger.warn('Trying to flush() SkyWalking agent which is not started.');
return null;
}
- const spanContextFlush = SpanContext.flush(); // if there are spans which haven't finished then wait for them
- const protocol = this.protocol;
-
- if (!spanContextFlush) return protocol.flush();
+ const spanContextFlush = SpanContext.flush();
+ if (!spanContextFlush) {
+ return ServiceManager.INSTANCE.flush();
+ }
return new Promise((resolve) => {
spanContextFlush.then(() => {
- const protocolFlush = protocol.flush();
-
- if (!protocolFlush) resolve(null);
- else protocolFlush.then(() => resolve(null));
+ const serviceFlush = ServiceManager.INSTANCE.flush();
+ if (!serviceFlush) resolve(null);
+ else serviceFlush.then(() => resolve(null));
});
});
}
destroy(): void {
- if (this.protocol === null) {
+ if (!this.started) {
logger.warn('Trying to destroy() SkyWalking agent which is not started.');
return;
}
logger.info('Destroying SkyWalking agent and cleaning up resources');
-
- // Clean up protocol resources
- this.protocol.destroy?.();
- this.protocol = null;
+ ServiceManager.INSTANCE.shutdown();
this.started = false;
}
}
diff --git a/src/trace/context/Segment.ts b/src/trace/context/Segment.ts
index 9c8a7fd..7d97414 100644
--- a/src/trace/context/Segment.ts
+++ b/src/trace/context/Segment.ts
@@ -21,6 +21,9 @@
import ID from '../../trace/ID';
import NewID from '../../trace/NewID';
import SegmentRef from '../../trace/context/SegmentRef';
+import config from '../../config/AgentConfig';
+import { KeyStringValuePair } from '../../proto/common/Common_pb';
+import { Log, RefType, SegmentObject, SegmentReference, SpanObject } from '../../proto/language-agent/Tracing_pb';
export default class Segment {
segmentId = new ID();
@@ -48,4 +51,50 @@
return this;
}
+
+ /** Convert to gRPC SegmentObject (Java TraceSegment.transform). */
+ transform(): SegmentObject {
+ return new SegmentObject()
+ .setService(config.serviceName)
+ .setServiceinstance(config.serviceInstance)
+ .setTraceid(this.relatedTraces[0].toString())
+ .setTracesegmentid(this.segmentId.toString())
+ .setSpansList(
+ this.spans.map((span) =>
+ new SpanObject()
+ .setSpanid(span.id)
+ .setParentspanid(span.parentId)
+ .setStarttime(span.startTime)
+ .setEndtime(span.endTime)
+ .setOperationname(span.operation)
+ .setPeer(span.peer)
+ .setSpantype(span.type)
+ .setSpanlayer(span.layer)
+ .setComponentid(span.component.id)
+ .setIserror(span.errored)
+ .setLogsList(
+ span.logs.map((log) =>
+ new Log()
+ .setTime(log.timestamp)
+ .setDataList(
+ log.items.map((logItem) => new KeyStringValuePair().setKey(logItem.key).setValue(logItem.val)),
+ ),
+ ),
+ )
+ .setTagsList(span.tags.map((tag) => new KeyStringValuePair().setKey(tag.key).setValue(tag.val)))
+ .setRefsList(
+ span.refs.map((ref) =>
+ new SegmentReference()
+ .setReftype(RefType.CROSSPROCESS)
+ .setTraceid(ref.traceId.toString())
+ .setParenttracesegmentid(ref.segmentId.toString())
+ .setParentspanid(ref.spanId)
+ .setParentservice(ref.service)
+ .setParentserviceinstance(ref.serviceInstance)
+ .setNetworkaddressusedatpeer(ref.clientAddress),
+ ),
+ ),
+ ),
+ );
+ }
}
diff --git a/tests/config/AgentConfig.test.ts b/tests/config/AgentConfig.test.ts
new file mode 100644
index 0000000..08b7fd3
--- /dev/null
+++ b/tests/config/AgentConfig.test.ts
@@ -0,0 +1,170 @@
+/*!
+ *
+ * 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.
+ *
+ */
+
+/* eslint-env jest */
+
+const registeredServiceNames = new Set<string>();
+
+jest.mock('../../src/agent/core/boot/ServiceManager', () => ({
+ __esModule: true,
+ default: {
+ INSTANCE: {
+ boot: jest.fn(() => {
+ registeredServiceNames.clear();
+ registeredServiceNames.add('GRPCChannelManager');
+ registeredServiceNames.add('TraceSegmentServiceClient');
+ registeredServiceNames.add('ServiceManagementClient');
+ const { default: agentConfig } = jest.requireActual('../../src/config/AgentConfig') as {
+ default: { runtimeMetricsReporterActive?: boolean };
+ };
+ if (agentConfig.runtimeMetricsReporterActive) {
+ registeredServiceNames.add('MeterSender');
+ }
+ }),
+ shutdown: jest.fn(() => {
+ registeredServiceNames.clear();
+ }),
+ flush: jest.fn(),
+ findService: jest.fn((serviceClass: { name: string }) =>
+ registeredServiceNames.has(serviceClass.name) ? {} : undefined,
+ ),
+ },
+ },
+}));
+
+jest.mock('../../src/core/PluginInstaller', () => ({
+ __esModule: true,
+ default: jest.fn().mockImplementation(() => ({
+ install: jest.fn(),
+ })),
+}));
+
+import agent, { config } from '../../src/index';
+import ServiceManager from '../../src/agent/core/boot/ServiceManager';
+import MeterSender from '../../src/agent/core/meter/MeterSender';
+import { AgentConfig, normalizeDeprecatedRuntimeMetricOptions } from '../../src/config/AgentConfig';
+
+function resetRuntimeMetricConfig(): void {
+ const mutableConfig = config as AgentConfig;
+ mutableConfig.runtimeMetricsReporterActive = true;
+ mutableConfig.runtimeMetricsCollectPeriod = 1000;
+ mutableConfig.runtimeMetricsReportPeriod = 1000;
+ mutableConfig.runtimeMetricsBufferSize = 600;
+ delete mutableConfig.nvmMetricsReporterActive;
+ delete mutableConfig.nvmJvmReporterActive;
+ delete mutableConfig.nvmMetricsCollectPeriod;
+ delete mutableConfig.nvmJvmMetricsCollectPeriod;
+ delete mutableConfig.nvmMetricsReportPeriod;
+ delete mutableConfig.nvmJvmMetricsReportPeriod;
+ delete mutableConfig.nvmMetricsBufferSize;
+ delete mutableConfig.nvmJvmMetricsBufferSize;
+}
+
+describe('AgentConfig deprecated runtime metric options (unit)', () => {
+ afterEach(() => {
+ agent.destroy();
+ resetRuntimeMetricConfig();
+ });
+
+ it('maps deprecated programmatic aliases before merge', () => {
+ const normalized = normalizeDeprecatedRuntimeMetricOptions({
+ nvmMetricsReporterActive: false,
+ nvmMetricsCollectPeriod: 2000,
+ nvmMetricsReportPeriod: 3000,
+ nvmMetricsBufferSize: 42,
+ });
+
+ expect(normalized.runtimeMetricsReporterActive).toBe(false);
+ expect(normalized.runtimeMetricsCollectPeriod).toBe(2000);
+ expect(normalized.runtimeMetricsReportPeriod).toBe(3000);
+ expect(normalized.runtimeMetricsBufferSize).toBe(42);
+ });
+
+ it('maps nvmJvm deprecated aliases before merge', () => {
+ const normalized = normalizeDeprecatedRuntimeMetricOptions({
+ nvmJvmReporterActive: false,
+ nvmJvmMetricsCollectPeriod: 2222,
+ nvmJvmMetricsReportPeriod: 3333,
+ nvmJvmMetricsBufferSize: 44,
+ });
+
+ expect(normalized.runtimeMetricsReporterActive).toBe(false);
+ expect(normalized.runtimeMetricsCollectPeriod).toBe(2222);
+ expect(normalized.runtimeMetricsReportPeriod).toBe(3333);
+ expect(normalized.runtimeMetricsBufferSize).toBe(44);
+ });
+
+ it('keeps explicit canonical options over deprecated aliases', () => {
+ const normalized = normalizeDeprecatedRuntimeMetricOptions({
+ runtimeMetricsReporterActive: true,
+ nvmMetricsReporterActive: false,
+ });
+
+ expect(normalized.runtimeMetricsReporterActive).toBe(true);
+ expect(normalized.nvmMetricsReporterActive).toBeUndefined();
+ expect(normalized.nvmJvmReporterActive).toBeUndefined();
+ });
+
+ it('disables runtime metrics when agent.start receives nvmMetrics alias', () => {
+ agent.start({ nvmMetricsReporterActive: false });
+
+ expect(config.runtimeMetricsReporterActive).toBe(false);
+ expect(ServiceManager.INSTANCE.findService(MeterSender)).toBeUndefined();
+ });
+
+ it('disables runtime metrics when agent.start receives nvmJvm alias', () => {
+ agent.start({ nvmJvmReporterActive: false });
+
+ expect(config.runtimeMetricsReporterActive).toBe(false);
+ expect(ServiceManager.INSTANCE.findService(MeterSender)).toBeUndefined();
+ });
+
+ it('disables runtime metrics when deprecated alias is set on exported config', () => {
+ (config as AgentConfig).nvmMetricsReporterActive = false;
+
+ agent.start();
+
+ expect(config.runtimeMetricsReporterActive).toBe(false);
+ expect(ServiceManager.INSTANCE.findService(MeterSender)).toBeUndefined();
+ });
+ it('maps deprecated aliases without leaving stale alias keys on normalized options', () => {
+ const normalized = normalizeDeprecatedRuntimeMetricOptions({
+ nvmMetricsReporterActive: false,
+ });
+
+ expect(normalized.runtimeMetricsReporterActive).toBe(false);
+ expect(normalized.nvmMetricsReporterActive).toBeUndefined();
+ expect(normalized.nvmJvmReporterActive).toBeUndefined();
+ });
+
+ it('re-enables runtime metrics after destroy/start with canonical option', () => {
+ agent.start({ nvmMetricsReporterActive: false });
+
+ expect(config.runtimeMetricsReporterActive).toBe(false);
+ expect((config as AgentConfig).nvmMetricsReporterActive).toBeUndefined();
+
+ agent.destroy();
+
+ agent.start({ runtimeMetricsReporterActive: true });
+
+ expect(config.runtimeMetricsReporterActive).toBe(true);
+ expect((config as AgentConfig).nvmMetricsReporterActive).toBeUndefined();
+ expect(ServiceManager.INSTANCE.findService(MeterSender)).toBeDefined();
+ });
+});
diff --git a/tests/plugins/express/expected.data.yaml b/tests/plugins/express/expected.data.yaml
index 891f130..afddb26 100644
--- a/tests/plugins/express/expected.data.yaml
+++ b/tests/plugins/express/expected.data.yaml
@@ -119,3 +119,59 @@
spanType: Exit
peer: server:5000
skipAnalysis: false
+
+meterItems:
+ - serviceName: server
+ meterSize: 6
+ meters:
+ - meterId:
+ name: instance_nodejs_process_cpu
+ tags: []
+ singleValue: gt 0
+ - meterId:
+ name: instance_nodejs_heap_used
+ tags: []
+ singleValue: gt 0
+ - meterId:
+ name: instance_nodejs_heap_total
+ tags: []
+ singleValue: gt 0
+ - meterId:
+ name: instance_nodejs_heap_limit
+ tags: []
+ singleValue: gt 0
+ - meterId:
+ name: instance_nodejs_rss
+ tags: []
+ singleValue: gt 0
+ - meterId:
+ name: instance_nodejs_external_memory
+ tags: []
+ singleValue: gt 0
+ - serviceName: client
+ meterSize: 6
+ meters:
+ - meterId:
+ name: instance_nodejs_process_cpu
+ tags: []
+ singleValue: gt 0
+ - meterId:
+ name: instance_nodejs_heap_used
+ tags: []
+ singleValue: gt 0
+ - meterId:
+ name: instance_nodejs_heap_total
+ tags: []
+ singleValue: gt 0
+ - meterId:
+ name: instance_nodejs_heap_limit
+ tags: []
+ singleValue: gt 0
+ - meterId:
+ name: instance_nodejs_rss
+ tags: []
+ singleValue: gt 0
+ - meterId:
+ name: instance_nodejs_external_memory
+ tags: []
+ singleValue: gt 0
diff --git a/tests/remote/GRPCChannelManager.test.ts b/tests/remote/GRPCChannelManager.test.ts
new file mode 100644
index 0000000..ad6f16e
--- /dev/null
+++ b/tests/remote/GRPCChannelManager.test.ts
@@ -0,0 +1,85 @@
+/*!
+ *
+ * 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.
+ *
+ */
+
+/* eslint-env jest */
+
+import * as grpc from '@grpc/grpc-js';
+import GRPCChannelManager from '../../src/agent/core/remote/GRPCChannelManager';
+import { GRPCChannelStatus } from '../../src/agent/core/remote/GRPCChannelStatus';
+
+const mockShutdownNow = jest.fn();
+const mockGetConnectivityState = jest.fn();
+const mockWatchConnectivityState = jest.fn();
+
+jest.mock('../../src/agent/core/remote/GRPCChannel', () => ({
+ __esModule: true,
+ default: {
+ newBuilder: jest.fn(() => ({
+ addManagedChannelBuilder: jest.fn().mockReturnThis(),
+ addChannelDecorator: jest.fn().mockReturnThis(),
+ build: jest.fn(() => ({
+ getChannel: () => ({
+ getConnectivityState: mockGetConnectivityState,
+ watchConnectivityState: mockWatchConnectivityState,
+ }),
+ getClientOptions: () => ({ channelOverride: {} }),
+ isConnected: jest.fn(() => true),
+ shutdownNow: mockShutdownNow,
+ })),
+ })),
+ },
+}));
+
+jest.mock('../../src/config/AgentConfig', () => ({
+ __esModule: true,
+ default: {
+ collectorAddress: '127.0.0.1:11800',
+ },
+}));
+
+describe('GRPCChannelManager initial connectivity', () => {
+ beforeEach(() => {
+ jest.clearAllMocks();
+ mockWatchConnectivityState.mockImplementation(() => undefined);
+ });
+
+ it('notifies CONNECTED when channel is already READY at boot', () => {
+ mockGetConnectivityState.mockReturnValue(grpc.connectivityState.READY);
+
+ const listener = { statusChanged: jest.fn() };
+ const manager = new GRPCChannelManager();
+
+ manager.addChannelListener(listener);
+ manager.boot();
+
+ expect(listener.statusChanged).toHaveBeenCalledWith(GRPCChannelStatus.CONNECTED);
+ });
+
+ it('notifies DISCONNECT when channel is not READY at boot', () => {
+ mockGetConnectivityState.mockReturnValue(grpc.connectivityState.CONNECTING);
+
+ const listener = { statusChanged: jest.fn() };
+ const manager = new GRPCChannelManager();
+
+ manager.addChannelListener(listener);
+ manager.boot();
+
+ expect(listener.statusChanged).toHaveBeenCalledWith(GRPCChannelStatus.DISCONNECT);
+ });
+});
diff --git a/tests/runtime/RuntimeMetricsCollector.test.ts b/tests/runtime/RuntimeMetricsCollector.test.ts
new file mode 100644
index 0000000..35b7833
--- /dev/null
+++ b/tests/runtime/RuntimeMetricsCollector.test.ts
@@ -0,0 +1,57 @@
+/*!
+ *
+ * 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.
+ *
+ */
+
+/* eslint-env jest */
+
+import RuntimeMetricsCollector from '../../src/agent/core/meter/RuntimeMetricsCollector';
+
+describe('RuntimeMetricsCollector', () => {
+ let collector: RuntimeMetricsCollector;
+
+ beforeEach(() => {
+ collector = new RuntimeMetricsCollector();
+ });
+
+ afterEach(() => {
+ collector.destroy();
+ });
+
+ it('maps Node.js runtime data into nodejs meter fields', () => {
+ const snapshot = collector.sample();
+ const meters = collector.toMeterData(snapshot);
+ const names = meters.map((meter) => meter.getSinglevalue()?.getName());
+
+ expect(names).toEqual(
+ expect.arrayContaining([
+ 'instance_nodejs_process_cpu',
+ 'instance_nodejs_heap_used',
+ 'instance_nodejs_heap_total',
+ 'instance_nodejs_heap_limit',
+ 'instance_nodejs_rss',
+ 'instance_nodejs_external_memory',
+ ]),
+ );
+
+ expect(names).toHaveLength(6);
+
+ for (const meter of meters) {
+ expect(meter.getSinglevalue()?.getValue()).toBeGreaterThanOrEqual(0);
+ }
+ });
+});
