Skip to content

Commit bc6c898

Browse files
songzhendongsongzhendong
committed
Address PR #139 review: Node-native gRPC lifecycle safety
- GRPCChannelManager: closed flag and stale connectivity callback guards - GRPCChannelManager: reportError notifies DISCONNECT on network errors - ServiceManagementClient: track status, clear client on disconnect, RPC deadlines - TraceSegmentServiceClient: single report timer/in-flight promise, flush safety - MeterSender: status gating and shared in-flight report promise - Remove unused GRPCStreamServiceStatus busy-wait class
1 parent 481b290 commit bc6c898

5 files changed

Lines changed: 179 additions & 167 deletions

File tree

‎src/agent/core/meter/MeterSender.ts‎

Lines changed: 62 additions & 52 deletions
Original file line numberDiff line numberDiff line change
@@ -34,10 +34,12 @@ const logReportError = throttled(logger, 'error', 30000);
3434

3535
/** Reports Node.js runtime metrics via gRPC MeterReportService (Go/Python-compatible pipeline). */
3636
export default class MeterSender implements BootService, GRPCChannelListener {
37-
private reporterClient!: MeterReportServiceClient;
37+
private status = GRPCChannelStatus.DISCONNECT;
38+
private reporterClient?: MeterReportServiceClient;
3839
private readonly buffer: RuntimeSnapshot[] = [];
3940
private collectTimer?: NodeJS.Timeout;
4041
private reportTimer?: NodeJS.Timeout;
42+
private reporting?: Promise<void>;
4143

4244
private collector!: RuntimeMetricsCollector;
4345

@@ -57,9 +59,8 @@ export default class MeterSender implements BootService, GRPCChannelListener {
5759
}
5860

5961
statusChanged(status: GRPCChannelStatus): void {
60-
if (status === GRPCChannelStatus.CONNECTED) {
61-
this.reporterClient = this.createReporterClient();
62-
}
62+
this.status = status;
63+
this.reporterClient = status === GRPCChannelStatus.CONNECTED ? this.createReporterClient() : undefined;
6364
}
6465

6566
private createReporterClient(): MeterReportServiceClient {
@@ -70,20 +71,15 @@ export default class MeterSender implements BootService, GRPCChannelListener {
7071
);
7172
}
7273

73-
get isConnected(): boolean {
74-
return ServiceManager.INSTANCE.findService(GRPCChannelManager)!.isConnected();
75-
}
76-
7774
private startTimers(): void {
7875
this.collectTimer = setInterval(
7976
() => this.collectSample(),
8077
config.runtimeMetricsCollectPeriod || 1000,
8178
) as NodeJS.Timeout;
8279
this.collectTimer.unref();
83-
this.reportTimer = setInterval(
84-
() => this.reportBufferedMetrics(),
85-
config.runtimeMetricsReportPeriod || 1000,
86-
) as NodeJS.Timeout;
80+
this.reportTimer = setInterval(() => {
81+
void this.reportBufferedMetrics();
82+
}, config.runtimeMetricsReportPeriod || 1000) as NodeJS.Timeout;
8783
this.reportTimer.unref();
8884
}
8985

@@ -95,56 +91,68 @@ export default class MeterSender implements BootService, GRPCChannelListener {
9591
this.buffer.push(this.collector.sample());
9692
}
9793

98-
private reportBufferedMetrics(callback?: () => void): void {
99-
try {
100-
if (this.buffer.length === 0 || !this.isConnected || !this.reporterClient) {
101-
if (callback) callback();
102-
return;
103-
}
94+
private reportBufferedMetrics(): Promise<void> {
95+
if (this.reporting) {
96+
return this.reporting;
97+
}
10498

105-
const snapshots = this.buffer.splice(0, this.buffer.length);
106-
const stream = this.reporterClient.collect(
107-
new grpc.Metadata(),
108-
{ deadline: Date.now() + (config.traceTimeout || 10000) },
109-
(error: grpc.ServiceError | null) => {
110-
if (error) {
111-
logReportError('Failed to report runtime meter data', error);
112-
ServiceManager.INSTANCE.findService(GRPCChannelManager)!.reportError(error);
113-
}
114-
if (callback) callback();
115-
},
116-
);
99+
this.reporting = this.doReportBufferedMetrics().finally(() => {
100+
this.reporting = undefined;
101+
});
102+
103+
return this.reporting;
104+
}
117105

106+
private doReportBufferedMetrics(): Promise<void> {
107+
return new Promise((resolve) => {
118108
try {
119-
let metadataWritten = false;
120-
const timestamp = Date.now();
121-
for (const snapshot of snapshots) {
122-
for (const meterData of this.collector.toMeterData(snapshot)) {
123-
if (!metadataWritten) {
124-
meterData
125-
.setService(config.serviceName!)
126-
.setServiceinstance(config.serviceInstance!)
127-
.setTimestamp(timestamp);
128-
metadataWritten = true;
109+
if (this.buffer.length === 0 || this.status !== GRPCChannelStatus.CONNECTED || !this.reporterClient) {
110+
resolve();
111+
return;
112+
}
113+
114+
const snapshots = this.buffer.splice(0, this.buffer.length);
115+
const stream = this.reporterClient.collect(
116+
new grpc.Metadata(),
117+
{ deadline: Date.now() + (config.traceTimeout || 10000) },
118+
(error: grpc.ServiceError | null) => {
119+
if (error) {
120+
logReportError('Failed to report runtime meter data', error);
121+
ServiceManager.INSTANCE.findService(GRPCChannelManager)!.reportError(error);
122+
}
123+
resolve();
124+
},
125+
);
126+
127+
try {
128+
let metadataWritten = false;
129+
const timestamp = Date.now();
130+
for (const snapshot of snapshots) {
131+
for (const meterData of this.collector.toMeterData(snapshot)) {
132+
if (!metadataWritten) {
133+
meterData
134+
.setService(config.serviceName!)
135+
.setServiceinstance(config.serviceInstance!)
136+
.setTimestamp(timestamp);
137+
metadataWritten = true;
138+
}
139+
stream.write(meterData);
129140
}
130-
stream.write(meterData);
131141
}
142+
} finally {
143+
stream.end();
132144
}
133-
} finally {
134-
stream.end();
145+
} catch (error) {
146+
logReportError('Failed to report runtime meter data', error);
147+
ServiceManager.INSTANCE.findService(GRPCChannelManager)!.reportError(error);
148+
resolve();
135149
}
136-
} catch (error) {
137-
logReportError('Failed to report runtime meter data', error);
138-
ServiceManager.INSTANCE.findService(GRPCChannelManager)!.reportError(error);
139-
if (callback) callback();
140-
}
150+
});
141151
}
142152

143153
flush(): Promise<any> | null {
144-
return new Promise((resolve) => {
145-
this.collectSample();
146-
this.reportBufferedMetrics(resolve);
147-
});
154+
this.collectSample();
155+
return this.reportBufferedMetrics();
148156
}
149157

150158
shutdown(): void {
@@ -156,6 +164,8 @@ export default class MeterSender implements BootService, GRPCChannelListener {
156164
clearInterval(this.reportTimer);
157165
this.reportTimer = undefined;
158166
}
167+
this.reporting = undefined;
168+
this.reporterClient = undefined;
159169
this.buffer.length = 0;
160170
this.collector.destroy();
161171
logger.info('MeterSender destroyed and resources cleaned up');

‎src/agent/core/remote/GRPCChannelManager.ts‎

Lines changed: 38 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,18 @@ import TLSChannelBuilder from './TLSChannelBuilder';
3232

3333
const logger = createLogger(__filename);
3434

35+
function isGrpcNetworkError(error: unknown): boolean {
36+
const code = (error as grpc.ServiceError | undefined)?.code;
37+
38+
return (
39+
code === grpc.status.UNAVAILABLE ||
40+
code === grpc.status.PERMISSION_DENIED ||
41+
code === grpc.status.UNAUTHENTICATED ||
42+
code === grpc.status.RESOURCE_EXHAUSTED ||
43+
code === grpc.status.UNKNOWN
44+
);
45+
}
46+
3547
/**
3648
* Shared gRPC channel manager (Java GRPCChannelManager skeleton).
3749
* V1: single address; V2 reserved: multi-address failover via reportError().
@@ -40,6 +52,7 @@ export default class GRPCChannelManager implements BootService {
4052
private managedChannel: GRPCChannel | null = null;
4153
private readonly listeners: GRPCChannelListener[] = [];
4254
private lastStatus: GRPCChannelStatus | null = null;
55+
private closed = false;
4356

4457
/** V1: first address when comma-separated; V2: failover selection. */
4558
resolveAddress(): string {
@@ -74,14 +87,21 @@ export default class GRPCChannelManager implements BootService {
7487
return Number.MAX_SAFE_INTEGER;
7588
}
7689

77-
/** V2 hook: network errors may trigger address failover. */
90+
/** Notify DISCONNECT on network errors so periodic work stops until grpc-js reconnects. */
7891
reportError(error: unknown): void {
79-
logger.debug('gRPC report error (multi-address failover reserved for V2): %s', error);
92+
if (!isGrpcNetworkError(error)) {
93+
logger.debug('gRPC report error (ignored): %s', error);
94+
return;
95+
}
96+
97+
logger.debug('gRPC network error, notify DISCONNECT: %s', error);
98+
this.notify(GRPCChannelStatus.DISCONNECT);
8099
}
81100

82101
prepare(): void {}
83102

84103
boot(): void {
104+
this.closed = false;
85105
const address = this.resolveAddress();
86106
const [host, portText] = address.split(':');
87107
const port = Number.parseInt(portText, 10);
@@ -103,16 +123,29 @@ export default class GRPCChannelManager implements BootService {
103123
onComplete(): void {}
104124

105125
shutdown(): void {
106-
this.managedChannel?.shutdownNow();
126+
this.closed = true;
127+
const managed = this.managedChannel;
107128
this.managedChannel = null;
129+
managed?.shutdownNow();
130+
this.notify(GRPCChannelStatus.DISCONNECT);
108131
}
109132

110133
private watchConnectivityState(): void {
111-
const channel = this.getChannel();
134+
const managed = this.managedChannel;
135+
if (this.closed || !managed) {
136+
return;
137+
}
138+
139+
const channel = managed.getChannel();
112140
const currentState = channel.getConnectivityState(true);
141+
113142
channel.watchConnectivityState(currentState, Infinity, (error) => {
143+
if (this.closed || this.managedChannel !== managed) {
144+
return;
145+
}
146+
114147
if (error) {
115-
logger.debug('Channel connectivity watch error: %s', error.message);
148+
logger.debug('Channel connectivity watch stopped: %s', error.message);
116149
return;
117150
}
118151

‎src/agent/core/remote/GRPCStreamServiceStatus.ts‎

Lines changed: 0 additions & 62 deletions
This file was deleted.

0 commit comments

Comments
 (0)