Copilot commented on code in PR #139: URL: https://github.com/apache/skywalking-nodejs/pull/139#discussion_r3487348748
########## src/agent/core/remote/GRPCChannelManager.ts: ########## @@ -0,0 +1,192 @@ +/*! + * + * 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(); + } + + 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) => { Review Comment: GRPCChannelManager never notifies listeners of the initial connectivity state. If the channel is already READY when watchConnectivityState() is set up, no state change occurs and remote clients may never create their stubs (status stays DISCONNECT). Also, passing Infinity as the watch deadline is not a documented/portable value for `@grpc/grpc-js` deadlines and may cause the watch to stop unexpectedly. ########## 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(), + ); Review Comment: createReporterClient() uses config.collectorAddress, which may contain multiple comma-separated backends. Even with channelOverride, grpc-js can validate/parse the target and reject comma-separated values. Use GRPCChannelManager.resolveAddress() so the meter client always uses the same single backend as the shared channel. ########## 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(), + ); Review Comment: createReporterClient() uses config.collectorAddress, which may contain multiple comma-separated backends. When channelOverride is used this string can still be validated/parsed by grpc-js, and a comma-separated target can be rejected. Use the same resolved single address as GRPCChannelManager to avoid invalid targets and keep v1 behavior consistent. ########## 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); + } + } Review Comment: MeterSender only sets service/serviceInstance/timestamp on the first MeterData written to the stream. Subsequent MeterData messages are sent without these fields, which can make the batch partially invalid/unattributable on the backend. Set these fields on every MeterData before writing. ########## 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(), + ); Review Comment: createManagementClient() uses config.collectorAddress, which may include comma-separated backends. grpc-js may reject such a target string even when channelOverride is provided. Prefer the single resolved address from GRPCChannelManager to avoid invalid targets and keep the v1 "first address" semantics consistent. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
