wu-sheng commented on code in PR #141:
URL: https://github.com/apache/skywalking-nodejs/pull/141#discussion_r3530573666
##########
src/index.ts:
##########
@@ -39,26 +53,33 @@ class Agent {
return;
}
- Object.assign(config, normalizeDeprecatedRuntimeMetricOptions(options));
- finalizeConfig(config, options);
+ const normalizedOptions = normalizeDeprecatedRuntimeMetricOptions(options);
+ Object.assign(agentConfig, normalizedOptions);
+ finalizeConfig(agentConfig, normalizedOptions);
logger.debug('Starting SkyWalking agent');
- new PluginInstaller().install();
-
- ServiceManager.INSTANCE.boot();
- this.started = true;
+ try {
+ new PluginInstaller().install();
+ ServiceManager.INSTANCE.boot();
+ this.started = true;
+ bootstrapPromise = Promise.resolve();
+ } catch (error) {
+ const err = error instanceof Error ? error : new Error(String(error));
+ bootstrapPromise = Promise.reject(err);
Review Comment:
**[High] Bootstrap failure becomes an `unhandledRejection` that can crash
the host process.**
This try/catch exists precisely so an agent bootstrap failure only logs and
never takes down the host app. But `bootstrapPromise = Promise.reject(err)`
stores a rejected promise with **no attached handler**. `whenReady()` (L38) is
opt-in and most embedders never call it.
Since Node 15 the default is `--unhandled-rejections=throw`: a rejected
promise with no handler by the microtask checkpoint **terminates the process**.
So a plugin that fails to load → catch logs (intended) → unhandled rejection →
Node kills the host application, defeating the whole point of the catch.
(The `whenReady` unit test does `await expect(whenReady()).rejects...`,
which itself attaches a handler and masks this.)
Suggested fix — attach a no-op handler so only opt-in awaiters observe the
rejection:
```suggestion
bootstrapPromise = Promise.reject(err);
// Avoid an unhandledRejection crashing the host when whenReady() is
never awaited.
bootstrapPromise.catch(() => {});
```
##########
src/agent/core/remote/GRPCChannelManager.ts:
##########
@@ -95,82 +111,227 @@ export default class GRPCChannelManager implements
BootService {
return Number.MAX_SAFE_INTEGER;
}
- /** Align local status with grpc-js connectivity; avoid permanent DISCONNECT
while channel stays READY. */
+ /** Align with grpc-js connectivity; avoid permanent DISCONNECT while
channel stays READY. */
reportError(error: unknown): void {
+ if (isGrpcAuthError(error)) {
+ logger.warn('gRPC authentication rejected by OAP; fix
SW_AGENT_AUTHENTICATION (no reconnect): %s', String(error));
+ return;
+ }
if (!isGrpcNetworkError(error)) {
logger.debug('gRPC report error (ignored): %s', error);
return;
}
+ if (this.closed) {
+ return;
+ }
const managed = this.managedChannel;
- if (!managed || this.closed) {
+ if (!managed) {
+ logger.debug('gRPC network error without channel, schedule reconnect:
%s', error);
+ this.reconnect = true;
this.notify(GRPCChannelStatus.DISCONNECT);
return;
}
+ this.reconnect = true;
+
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);
+ logger.debug('gRPC network error, schedule reconnect: %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.grpcServers = parseStaticBackendAddresses(config.collectorAddress ??
'').map((target) => ({ target }));
+ if (this.grpcServers.length === 0) {
+ logger.error('Collector server addresses are not set.');
+ logger.error('Agent will not uplink any data.');
+ return;
}
-
- this.managedChannel = GRPCChannel.newBuilder(host, port)
- .addManagedChannelBuilder(new StandardChannelBuilder())
- .addManagedChannelBuilder(new TLSChannelBuilder())
- .addChannelDecorator(new AgentIDDecorator())
- .addChannelDecorator(new AuthenticationDecorator())
- .build();
-
- this.watchConnectivityState();
- this.notifyCurrentConnectivityState(true);
+ this.reconnect = true;
+ const intervalMs = (config.grpcChannelCheckInterval ?? 30) * 1000;
+ this.checkTimer = setInterval(() => {
+ void this.runCheck();
+ }, intervalMs);
+ this.checkTimer.unref();
+ void this.runCheck();
}
onComplete(): void {}
shutdown(): void {
this.closed = true;
+ if (this.checkTimer) {
+ clearInterval(this.checkTimer);
+ this.checkTimer = undefined;
+ }
+ this.watcherGeneration += 1;
const managed = this.managedChannel;
this.managedChannel = null;
managed?.shutdownNow();
this.notify(GRPCChannelStatus.DISCONNECT);
this.listeners.length = 0;
+ this.grpcServers = [];
+ this.selectedIdx = -1;
+ this.currentTarget = null;
+ this.reconnect = true;
+ clearTlsMaterialCacheOnShutdown();
+ }
+
+ /** Java GRPCChannelManager.run() — exposed for unit tests. */
+ async runCheck(): Promise<void> {
+ if (this.closed || this.checkInFlight) {
+ return;
+ }
+ this.checkInFlight = true;
+ try {
+ logger.debug('gRPC channel check running, reconnect: %s',
this.reconnect);
+
+ // grpc-js may keep READY after TCP teardown; probe before
reconnect-only early return.
+ if (this.managedChannel && !this.managedChannel.isConnected(false)) {
+ this.reconnect = true;
+ }
+
+ if (config.isResolveDnsPeriodically && this.reconnect) {
+ const staticEntries =
parseStaticBackendAddresses(config.collectorAddress ?? '');
+ this.grpcServers = await expandBackendAddresses(staticEntries, true);
+ if (this.closed) {
+ return;
+ }
+ } else if (this.grpcServers.length === 0) {
+ this.grpcServers = parseStaticBackendAddresses(config.collectorAddress
?? '').map((target) => ({ target }));
+ }
+
+ if (!this.reconnect || this.closed) {
+ return;
+ }
+
+ if (this.grpcServers.length === 0) {
+ logger.debug('No collector backend available. Wait %s seconds to
retry', config.grpcChannelCheckInterval ?? 30);
+ return;
+ }
+
+ // Java parity: rebuild channel only when random index != selectedIdx.
+ // Same index with a changed resolved ip:port does NOT trigger rebuild
(see Java GRPCChannelManager.run()).
+ const index = Math.abs(Math.floor(Math.random() *
this.grpcServers.length));
+ if (index !== this.selectedIdx) {
+ await this.switchToServer(this.grpcServers[index]!, index);
+ return;
+ }
+
+ const forceReconnect = ++this.reconnectCount >
(config.forceReconnectionPeriod ?? 1);
+ if (this.managedChannel?.isConnected(forceReconnect)) {
+ this.reconnectCount = 0;
+ this.reconnect = false;
+ this.notifyCurrentConnectivityState(true);
+ return;
+ }
+
+ // Same index but backend unreachable — rotate (needed for DNS-expanded
ip:port lists).
+ if (this.grpcServers.length > 1 && this.selectedIdx >= 0 &&
!this.managedChannel?.isConnected(false)) {
+ const nextIdx = (this.selectedIdx + 1) % this.grpcServers.length;
+ await this.switchToServer(this.grpcServers[nextIdx]!, nextIdx);
+ }
+ } catch (error) {
+ logger.error('gRPC channel check failed: %s', error);
+ } finally {
+ this.checkInFlight = false;
+ }
+ }
+
+ /** @internal test hook */
+ getReconnectStateForTest(): boolean {
+ return this.reconnect;
+ }
+
+ /** @internal test hook */
+ getGrpcServersForTest(): BackendTarget[] {
+ return this.grpcServers.map((entry) => ({ ...entry }));
+ }
+
+ /** @internal test hook */
+ getSelectedIdxForTest(): number {
+ return this.selectedIdx;
+ }
+
+ /** @internal test hook */
+ getReconnectCountForTest(): number {
+ return this.reconnectCount;
+ }
+
+ /** @internal test hook */
+ getLastStatusForTest(): GRPCChannelStatus | null {
+ return this.lastStatus;
+ }
+
+ /** @internal test hook */
+ hasCheckTimerForTest(): boolean {
+ return this.checkTimer !== undefined;
+ }
+
+ private async switchToServer(backend: BackendTarget, index: number):
Promise<void> {
+ if (this.closed) {
+ return;
+ }
+
+ const target = backend.target;
+ const { host, port: portText } = splitHostPort(target);
+ const port = Number.parseInt(portText, 10);
+ if (!host || Number.isNaN(port)) {
+ throw new Error(`Invalid collector address: ${target}`);
+ }
+
+ this.watcherGeneration += 1;
+ const previous = this.managedChannel;
+ this.managedChannel = null;
+ previous?.shutdownNow();
+
+ try {
+ // Java TLSChannelBuilder reads TLS files on every channel rebuild.
+ await preloadTlsMaterials();
Review Comment:
**[Medium] Shutdown race across the `preloadTlsMaterials()` await leaks a
never-closed channel into grpc-js's global channelz map.**
`switchToServer` checks `this.closed` only at L280 (pre-await). If
`shutdown()` runs during this `await` (it captures the already-nulled
`managedChannel`, so its `shutdownNow()` is a no-op, and sets `closed=true`),
execution resumes here and **unconditionally** builds + assigns a new
`GRPCChannel` (L300-305). Then `watchConnectivityState()` (L315) and
`notifyCurrentConnectivityState()` (L316) both early-return on `closed`, so the
channel is **never closed**.
`GRPCChannel`'s ctor calls `new grpc.Channel` → `registerChannelzChannel`
(`internal-channel.js:181`) which stores it in the **process-global**
`entityMaps.channel` map (`channelz.js:196`); `unregisterChannelzRef` runs only
in `close()`. → a GC-defeating leak that accumulates across boot/shutdown
cycles (serverless freeze, test teardown, agent re-init). Note `runCheck`'s
DNS-expand await *is* guarded (L205); this path is the gap. (The channel is
idle, so it's only the channelz registration that leaks — no socket/idle-timer.)
Suggested fix — re-check `closed` after the await, before publishing the new
channel:
```ts
await preloadTlsMaterials();
if (this.closed) { this.managedChannel?.shutdownNow(); this.managedChannel =
null; return; }
// ...build channel...
if (this.closed) { built.shutdownNow(); return; }
this.managedChannel = built;
```
##########
src/agent/core/remote/GRPCChannelManager.ts:
##########
@@ -95,82 +111,227 @@ export default class GRPCChannelManager implements
BootService {
return Number.MAX_SAFE_INTEGER;
}
- /** Align local status with grpc-js connectivity; avoid permanent DISCONNECT
while channel stays READY. */
+ /** Align with grpc-js connectivity; avoid permanent DISCONNECT while
channel stays READY. */
reportError(error: unknown): void {
+ if (isGrpcAuthError(error)) {
+ logger.warn('gRPC authentication rejected by OAP; fix
SW_AGENT_AUTHENTICATION (no reconnect): %s', String(error));
+ return;
+ }
if (!isGrpcNetworkError(error)) {
logger.debug('gRPC report error (ignored): %s', error);
return;
}
+ if (this.closed) {
+ return;
+ }
const managed = this.managedChannel;
- if (!managed || this.closed) {
+ if (!managed) {
+ logger.debug('gRPC network error without channel, schedule reconnect:
%s', error);
+ this.reconnect = true;
this.notify(GRPCChannelStatus.DISCONNECT);
return;
}
+ this.reconnect = true;
+
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);
+ logger.debug('gRPC network error, schedule reconnect: %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.grpcServers = parseStaticBackendAddresses(config.collectorAddress ??
'').map((target) => ({ target }));
+ if (this.grpcServers.length === 0) {
+ logger.error('Collector server addresses are not set.');
+ logger.error('Agent will not uplink any data.');
+ return;
}
-
- this.managedChannel = GRPCChannel.newBuilder(host, port)
- .addManagedChannelBuilder(new StandardChannelBuilder())
- .addManagedChannelBuilder(new TLSChannelBuilder())
- .addChannelDecorator(new AgentIDDecorator())
- .addChannelDecorator(new AuthenticationDecorator())
- .build();
-
- this.watchConnectivityState();
- this.notifyCurrentConnectivityState(true);
+ this.reconnect = true;
+ const intervalMs = (config.grpcChannelCheckInterval ?? 30) * 1000;
+ this.checkTimer = setInterval(() => {
+ void this.runCheck();
+ }, intervalMs);
+ this.checkTimer.unref();
+ void this.runCheck();
}
onComplete(): void {}
shutdown(): void {
this.closed = true;
+ if (this.checkTimer) {
+ clearInterval(this.checkTimer);
+ this.checkTimer = undefined;
+ }
+ this.watcherGeneration += 1;
const managed = this.managedChannel;
this.managedChannel = null;
managed?.shutdownNow();
this.notify(GRPCChannelStatus.DISCONNECT);
this.listeners.length = 0;
+ this.grpcServers = [];
+ this.selectedIdx = -1;
+ this.currentTarget = null;
+ this.reconnect = true;
+ clearTlsMaterialCacheOnShutdown();
+ }
+
+ /** Java GRPCChannelManager.run() — exposed for unit tests. */
+ async runCheck(): Promise<void> {
+ if (this.closed || this.checkInFlight) {
+ return;
+ }
+ this.checkInFlight = true;
+ try {
+ logger.debug('gRPC channel check running, reconnect: %s',
this.reconnect);
+
+ // grpc-js may keep READY after TCP teardown; probe before
reconnect-only early return.
+ if (this.managedChannel && !this.managedChannel.isConnected(false)) {
+ this.reconnect = true;
+ }
+
+ if (config.isResolveDnsPeriodically && this.reconnect) {
+ const staticEntries =
parseStaticBackendAddresses(config.collectorAddress ?? '');
+ this.grpcServers = await expandBackendAddresses(staticEntries, true);
+ if (this.closed) {
+ return;
+ }
+ } else if (this.grpcServers.length === 0) {
+ this.grpcServers = parseStaticBackendAddresses(config.collectorAddress
?? '').map((target) => ({ target }));
+ }
+
+ if (!this.reconnect || this.closed) {
+ return;
+ }
+
+ if (this.grpcServers.length === 0) {
+ logger.debug('No collector backend available. Wait %s seconds to
retry', config.grpcChannelCheckInterval ?? 30);
+ return;
+ }
+
+ // Java parity: rebuild channel only when random index != selectedIdx.
+ // Same index with a changed resolved ip:port does NOT trigger rebuild
(see Java GRPCChannelManager.run()).
+ const index = Math.abs(Math.floor(Math.random() *
this.grpcServers.length));
+ if (index !== this.selectedIdx) {
+ await this.switchToServer(this.grpcServers[index]!, index);
+ return;
+ }
+
+ const forceReconnect = ++this.reconnectCount >
(config.forceReconnectionPeriod ?? 1);
+ if (this.managedChannel?.isConnected(forceReconnect)) {
+ this.reconnectCount = 0;
+ this.reconnect = false;
+ this.notifyCurrentConnectivityState(true);
+ return;
+ }
+
+ // Same index but backend unreachable — rotate (needed for DNS-expanded
ip:port lists).
+ if (this.grpcServers.length > 1 && this.selectedIdx >= 0 &&
!this.managedChannel?.isConnected(false)) {
+ const nextIdx = (this.selectedIdx + 1) % this.grpcServers.length;
+ await this.switchToServer(this.grpcServers[nextIdx]!, nextIdx);
+ }
+ } catch (error) {
+ logger.error('gRPC channel check failed: %s', error);
+ } finally {
+ this.checkInFlight = false;
+ }
+ }
+
+ /** @internal test hook */
+ getReconnectStateForTest(): boolean {
+ return this.reconnect;
+ }
+
+ /** @internal test hook */
+ getGrpcServersForTest(): BackendTarget[] {
+ return this.grpcServers.map((entry) => ({ ...entry }));
+ }
+
+ /** @internal test hook */
+ getSelectedIdxForTest(): number {
+ return this.selectedIdx;
+ }
+
+ /** @internal test hook */
+ getReconnectCountForTest(): number {
+ return this.reconnectCount;
+ }
+
+ /** @internal test hook */
+ getLastStatusForTest(): GRPCChannelStatus | null {
+ return this.lastStatus;
+ }
+
+ /** @internal test hook */
+ hasCheckTimerForTest(): boolean {
+ return this.checkTimer !== undefined;
+ }
+
+ private async switchToServer(backend: BackendTarget, index: number):
Promise<void> {
+ if (this.closed) {
+ return;
+ }
+
+ const target = backend.target;
+ const { host, port: portText } = splitHostPort(target);
+ const port = Number.parseInt(portText, 10);
+ if (!host || Number.isNaN(port)) {
+ throw new Error(`Invalid collector address: ${target}`);
+ }
+
+ this.watcherGeneration += 1;
+ const previous = this.managedChannel;
+ this.managedChannel = null;
+ previous?.shutdownNow();
+
+ try {
+ // Java TLSChannelBuilder reads TLS files on every channel rebuild.
+ await preloadTlsMaterials();
+
+ this.managedChannel = GRPCChannel.newBuilder(host, port,
backend.tlsServerName)
+ .addManagedChannelBuilder(new StandardChannelBuilder())
+ .addManagedChannelBuilder(new TLSChannelBuilder())
+ .addChannelDecorator(new AgentIDDecorator())
+ .addChannelDecorator(new AuthenticationDecorator())
+ .build();
+ } catch (error) {
+ logger.error('Failed to build gRPC channel for target [%s]: %s', target,
error);
+ return;
+ }
+
+ this.selectedIdx = index;
+ this.currentTarget = backend.target;
+ this.reconnectCount = 0;
+ this.reconnect = false;
+ this.watchConnectivityState();
+ this.notifyCurrentConnectivityState(true);
}
private watchConnectivityState(): void {
const managed = this.managedChannel;
if (this.closed || !managed) {
return;
}
-
+ const generation = this.watcherGeneration;
const channel = managed.getChannel();
const currentState = channel.getConnectivityState(true);
-
- channel.watchConnectivityState(currentState, Infinity, (error) => {
- if (this.closed || this.managedChannel !== managed) {
+ channel.watchConnectivityState(currentState, Date.now() + 86_400_000,
(error) => {
Review Comment:
**[Medium] 24h finite deadline should be `Infinity` — this ref's a timer
that holds the event loop and stops connectivity watching after 24h.**
In grpc-js 1.14.4 `internal-channel.js:561` does `if (deadline !==
Infinity)` and, for any finite deadline, `internal-channel.js:568` creates a
`setTimeout(...)` **without `.unref()`**. Consequences:
1. A **ref'd 24h timer keeps the Node event loop alive**, undoing the
deliberate `this.checkTimer.unref()` at L163 (grpc-js otherwise
`session.unref()`s its sockets so a passive agent never pins the host).
2. At 24h the callback fires with `Error('Deadline passed without
connectivity state change')`. The callback below treats any error as terminal
and returns **without re-arming** — with a single backend, `runCheck` never
re-enters `switchToServer`, so event-driven connectivity notifications stop
permanently (only the 30s poll remains).
Passing `Infinity` matches Java's watch-forever semantics;
`internal-channel.js:561` then creates no timer. The existing
generation/`managedChannel` guards still handle teardown.
```suggestion
channel.watchConnectivityState(currentState, Infinity, (error) => {
```
##########
src/agent/core/remote/TlsMaterialCache.ts:
##########
@@ -0,0 +1,210 @@
+/*!
+ *
+ * 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 fs from 'fs';
+import { promises as fsPromises } from 'fs';
+import config from '../../../config/AgentConfig';
+import { getAgentPackagePath, isPathInsideAgentRoot, resolveAgentPath } from
'../boot/AgentPackagePath';
+import path from 'path';
+import { loadDecryptionKeyFromBuffer } from '../util/PrivateKeyUtil';
+import { createLogger } from '../../../logging';
+
+const logger = createLogger(__filename);
+
+/** Max TLS PEM/CA file size (256 KiB) to limit memory DoS. */
+const MAX_TLS_MATERIAL_BYTES = 256 * 1024;
+
+export type TlsMaterialSnapshot = {
+ rootCerts: Buffer | null;
+ privateKey: Buffer | null;
+ certChain: Buffer | null;
+};
+
+/** Last preload result consumed by {@code TLSChannelBuilder.build()} in the
same rebuild. */
+let cachedSnapshot: TlsMaterialSnapshot | null = null;
+
+/** Whether the trusted CA file existed on the last {@link
preloadTlsMaterials} call. */
+let caFileExistsOnLastPreload = false;
+
+/** Read a regular file; reject symlinks (B-1 path traversal via link). */
+async function readRegularFileBytes(
+ filePath: string,
+ options: { mustStayUnderAgentRoot?: boolean; requirePrivateKeyMode?: boolean
} = {},
+): Promise<Buffer | null> {
+ try {
+ const lstat = await fsPromises.lstat(filePath);
+ if (lstat.isSymbolicLink()) {
+ logger.warn('TLS path [%s] is a symbolic link; refusing to load',
filePath);
+ return null;
+ }
+ if (!lstat.isFile()) {
+ return null;
+ }
+ if (lstat.size > MAX_TLS_MATERIAL_BYTES) {
+ logger.warn('TLS path [%s] exceeds max size %d bytes; refusing to load',
filePath, MAX_TLS_MATERIAL_BYTES);
+ return null;
+ }
+ if (options.requirePrivateKeyMode && (lstat.mode & 0o077) !== 0) {
Review Comment:
**[Medium] `(lstat.mode & 0o077)` rejects every key on Windows → mTLS can
never initialize there.**
Correct on POSIX. On Windows, Node synthesizes `Stats.mode` from the
read-only attribute and mirrors the owner bits into group/other, so a normal
key reports `0o100666` → `0o666 & 0o077 = 0o066 ≠ 0` → **refused** (read-only
reports `0o100444` → `0o044 ≠ 0`, also refused). `readPrivateKeyIfExists` then
returns null, `preloadTlsMaterials` (L169-171) nulls both key+cert, and
`TLSChannelBuilder` throws `mTLS material unavailable` regardless of the real
NTFS ACL. The `O_NOFOLLOW ?? 0` handling one line down is already
platform-aware; this check isn't.
```suggestion
if (options.requirePrivateKeyMode && process.platform !== 'win32' &&
(lstat.mode & 0o077) !== 0) {
```
##########
src/agent/core/remote/BackendAddressResolver.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 { promises as dnsPromises } from 'dns';
+import net from 'net';
+import { createLogger } from '../../../logging';
+
+const logger = createLogger(__filename);
+
+/** Cap DNS expansion to avoid unbounded memory on hostile or misconfigured
resolvers. */
+export const MAX_DNS_LOOKUP_RECORDS = 64;
+
+/** Abort slow DNS lookups so channel checks do not stall the event loop. */
+export const DNS_LOOKUP_TIMEOUT_MS = 5_000;
+
+/** Resolved gRPC target with optional TLS SNI hostname (H-2 per-entry
mapping). */
+export interface BackendTarget {
+ target: string;
+ /** Hostname for grpc.ssl_target_name_override when target is a resolved IP.
*/
+ tlsServerName?: string;
+}
+
+export type DnsLookupFn = (
+ hostname: string,
+ options: { all: true; verbatim?: boolean },
+) => Promise<Array<{ address: string; family: number }>>;
+
+const defaultLookup: DnsLookupFn = (hostname, options) =>
+ dnsPromises.lookup(hostname, options) as ReturnType<DnsLookupFn>;
+
+export async function lookupBackendHostRecords(
+ hostname: string,
+ lookup: DnsLookupFn = defaultLookup,
+ timeoutMs: number = DNS_LOOKUP_TIMEOUT_MS,
+): Promise<Array<{ address: string; family: number }>> {
+ let timeoutHandle: NodeJS.Timeout | undefined;
+ try {
+ const records = await Promise.race([
+ lookup(hostname, { all: true, verbatim: true }),
+ new Promise<never>((_, reject) => {
+ timeoutHandle = setTimeout(() => reject(new Error('DNS lookup
timeout')), timeoutMs);
+ timeoutHandle.unref();
+ }),
+ ]);
+ if (records.length > MAX_DNS_LOOKUP_RECORDS) {
+ logger.warn('DNS returned %s records for %s; using first %s',
records.length, hostname, MAX_DNS_LOOKUP_RECORDS);
+ return records.slice(0, MAX_DNS_LOOKUP_RECORDS);
+ }
+ return records;
+ } finally {
+ if (timeoutHandle) {
+ clearTimeout(timeoutHandle);
+ }
+ }
+}
+
+/** Parse comma-separated backend entries (Java BACKEND_SERVICE split). */
+export function parseStaticBackendAddresses(raw: string): string[] {
+ return raw
+ .split(',')
+ .map((entry) => entry.trim())
+ .filter(Boolean)
+ .filter(isValidHostPortEntry);
+}
+
+function isValidHostPortEntry(entry: string): boolean {
+ try {
+ const { host, port } = splitHostPort(entry);
+ const portNum = Number.parseInt(port, 10);
Review Comment:
**[Low] `Number.parseInt` is prefix-lenient, so malformed ports pass
validation and produce an unconnectable target instead of being rejected.**
`Number.parseInt('11800x', 10) === 11800` passes the range check, but
`formatHostPort` (L114-119) emits the **raw** slice, so
`collectorAddress='oap:11800x'` becomes target `…:11800x` handed to grpc-js as
an invalid authority — and the clean "format error" debug log at L88 is
bypassed, so a config typo silently yields a dead channel. `'oap: 11800'` slips
through as port `' 11800'` too. (Java uses `Integer.parseInt`, which throws.)
Suggested fix — require an all-digits port and reuse the parsed integer for
the target:
```ts
if (!/^[0-9]+$/.test(port)) return false;
const portNum = Number.parseInt(port, 10);
return portNum > 0 && portNum <= 65535;
```
##########
src/agent/core/remote/BackendAddressResolver.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 { promises as dnsPromises } from 'dns';
+import net from 'net';
+import { createLogger } from '../../../logging';
+
+const logger = createLogger(__filename);
+
+/** Cap DNS expansion to avoid unbounded memory on hostile or misconfigured
resolvers. */
+export const MAX_DNS_LOOKUP_RECORDS = 64;
+
+/** Abort slow DNS lookups so channel checks do not stall the event loop. */
+export const DNS_LOOKUP_TIMEOUT_MS = 5_000;
+
+/** Resolved gRPC target with optional TLS SNI hostname (H-2 per-entry
mapping). */
+export interface BackendTarget {
+ target: string;
+ /** Hostname for grpc.ssl_target_name_override when target is a resolved IP.
*/
+ tlsServerName?: string;
+}
+
+export type DnsLookupFn = (
+ hostname: string,
+ options: { all: true; verbatim?: boolean },
+) => Promise<Array<{ address: string; family: number }>>;
+
+const defaultLookup: DnsLookupFn = (hostname, options) =>
+ dnsPromises.lookup(hostname, options) as ReturnType<DnsLookupFn>;
+
+export async function lookupBackendHostRecords(
+ hostname: string,
+ lookup: DnsLookupFn = defaultLookup,
+ timeoutMs: number = DNS_LOOKUP_TIMEOUT_MS,
+): Promise<Array<{ address: string; family: number }>> {
+ let timeoutHandle: NodeJS.Timeout | undefined;
+ try {
+ const records = await Promise.race([
+ lookup(hostname, { all: true, verbatim: true }),
+ new Promise<never>((_, reject) => {
+ timeoutHandle = setTimeout(() => reject(new Error('DNS lookup
timeout')), timeoutMs);
+ timeoutHandle.unref();
+ }),
+ ]);
+ if (records.length > MAX_DNS_LOOKUP_RECORDS) {
+ logger.warn('DNS returned %s records for %s; using first %s',
records.length, hostname, MAX_DNS_LOOKUP_RECORDS);
+ return records.slice(0, MAX_DNS_LOOKUP_RECORDS);
+ }
+ return records;
+ } finally {
+ if (timeoutHandle) {
+ clearTimeout(timeoutHandle);
+ }
+ }
+}
+
+/** Parse comma-separated backend entries (Java BACKEND_SERVICE split). */
+export function parseStaticBackendAddresses(raw: string): string[] {
+ return raw
+ .split(',')
+ .map((entry) => entry.trim())
+ .filter(Boolean)
+ .filter(isValidHostPortEntry);
+}
+
+function isValidHostPortEntry(entry: string): boolean {
+ try {
+ const { host, port } = splitHostPort(entry);
+ const portNum = Number.parseInt(port, 10);
+ return Boolean(host) && !Number.isNaN(portNum) && portNum > 0 && portNum
<= 65535;
+ } catch {
+ logger.debug('Service address [%s] format error. Expected host:port',
entry);
+ return false;
+ }
+}
+
+/** Split host:port, supporting bracketed IPv6 literals such as [::1]:11800. */
+export function splitHostPort(entry: string): { host: string; port: string } {
+ const trimmed = entry.trim();
+ if (!trimmed) {
+ throw new Error(`Invalid host:port entry: ${entry}`);
+ }
+ if (trimmed.startsWith('[')) {
+ const close = trimmed.indexOf(']');
+ if (close < 0 || trimmed[close + 1] !== ':') {
+ throw new Error(`Invalid host:port entry: ${entry}`);
+ }
+ return { host: trimmed.slice(1, close), port: trimmed.slice(close + 2) };
+ }
+ const lastColon = trimmed.lastIndexOf(':');
Review Comment:
**[Low] Bare unbracketed IPv6 literal is mis-parsed and passes validation.**
For `'::1'`: not bracketed, `lastIndexOf(':') === 1`, so `host=':'`,
`port='1'`; `isValidHostPortEntry` accepts it (host truthy, port 1 in range).
Downstream `net.isIP(':') === 0` so it's DNS-resolved (fails, silently dropped)
or emitted as bogus target `':1'`, and it's also counted as a hostname by
`countDistinctNonIpHostnames`. The bracketed branch already enforces "IPv6
needs brackets"; this branch has no such guard.
Suggested fix — in the non-bracketed branch, reject a host that still
contains a colon:
```ts
const host = trimmed.slice(0, lastColon);
if (host.includes(':')) throw new Error(`Invalid host:port entry: ${entry}`);
```
##########
src/agent/core/remote/BackendAddressResolver.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 { promises as dnsPromises } from 'dns';
+import net from 'net';
+import { createLogger } from '../../../logging';
+
+const logger = createLogger(__filename);
+
+/** Cap DNS expansion to avoid unbounded memory on hostile or misconfigured
resolvers. */
+export const MAX_DNS_LOOKUP_RECORDS = 64;
+
+/** Abort slow DNS lookups so channel checks do not stall the event loop. */
+export const DNS_LOOKUP_TIMEOUT_MS = 5_000;
+
+/** Resolved gRPC target with optional TLS SNI hostname (H-2 per-entry
mapping). */
+export interface BackendTarget {
+ target: string;
+ /** Hostname for grpc.ssl_target_name_override when target is a resolved IP.
*/
+ tlsServerName?: string;
+}
+
+export type DnsLookupFn = (
+ hostname: string,
+ options: { all: true; verbatim?: boolean },
+) => Promise<Array<{ address: string; family: number }>>;
+
+const defaultLookup: DnsLookupFn = (hostname, options) =>
+ dnsPromises.lookup(hostname, options) as ReturnType<DnsLookupFn>;
+
+export async function lookupBackendHostRecords(
+ hostname: string,
+ lookup: DnsLookupFn = defaultLookup,
+ timeoutMs: number = DNS_LOOKUP_TIMEOUT_MS,
+): Promise<Array<{ address: string; family: number }>> {
+ let timeoutHandle: NodeJS.Timeout | undefined;
+ try {
+ const records = await Promise.race([
+ lookup(hostname, { all: true, verbatim: true }),
+ new Promise<never>((_, reject) => {
+ timeoutHandle = setTimeout(() => reject(new Error('DNS lookup
timeout')), timeoutMs);
+ timeoutHandle.unref();
+ }),
+ ]);
+ if (records.length > MAX_DNS_LOOKUP_RECORDS) {
+ logger.warn('DNS returned %s records for %s; using first %s',
records.length, hostname, MAX_DNS_LOOKUP_RECORDS);
+ return records.slice(0, MAX_DNS_LOOKUP_RECORDS);
+ }
+ return records;
+ } finally {
+ if (timeoutHandle) {
+ clearTimeout(timeoutHandle);
+ }
+ }
+}
+
+/** Parse comma-separated backend entries (Java BACKEND_SERVICE split). */
+export function parseStaticBackendAddresses(raw: string): string[] {
+ return raw
+ .split(',')
+ .map((entry) => entry.trim())
+ .filter(Boolean)
+ .filter(isValidHostPortEntry);
+}
+
+function isValidHostPortEntry(entry: string): boolean {
+ try {
+ const { host, port } = splitHostPort(entry);
+ const portNum = Number.parseInt(port, 10);
+ return Boolean(host) && !Number.isNaN(portNum) && portNum > 0 && portNum
<= 65535;
+ } catch {
+ logger.debug('Service address [%s] format error. Expected host:port',
entry);
+ return false;
+ }
+}
+
+/** Split host:port, supporting bracketed IPv6 literals such as [::1]:11800. */
+export function splitHostPort(entry: string): { host: string; port: string } {
+ const trimmed = entry.trim();
+ if (!trimmed) {
+ throw new Error(`Invalid host:port entry: ${entry}`);
+ }
+ if (trimmed.startsWith('[')) {
+ const close = trimmed.indexOf(']');
+ if (close < 0 || trimmed[close + 1] !== ':') {
+ throw new Error(`Invalid host:port entry: ${entry}`);
+ }
+ return { host: trimmed.slice(1, close), port: trimmed.slice(close + 2) };
+ }
+ const lastColon = trimmed.lastIndexOf(':');
+ if (lastColon <= 0) {
+ throw new Error(`Invalid host:port entry: ${entry}`);
+ }
+ return { host: trimmed.slice(0, lastColon), port: trimmed.slice(lastColon +
1) };
+}
+
+/** Format host:port for grpc target strings; IPv6 hosts are bracketed. */
+export function formatHostPort(host: string, port: string | number): string {
+ const portText = String(port);
+ if (net.isIPv6(host)) {
+ return `[${host}]:${portText}`;
+ }
+ return `${host}:${portText}`;
+}
+
+export function isLiteralIp(host: string): boolean {
+ return net.isIP(host) !== 0;
+}
+
+/**
+ * Expand backend entries to targets (Java InetAddress.getAllByName).
+ * When resolveDns=false, returns static entries only.
+ * DNS-expanded IPs carry tlsServerName from the source hostname for grpc-js
SNI.
+ */
+export async function expandBackendAddresses(
+ entries: string[],
+ resolveDns: boolean,
+ lookup: DnsLookupFn = defaultLookup,
+): Promise<BackendTarget[]> {
+ const byTarget = new Map<string, BackendTarget>();
+
+ const addTarget = (item: BackendTarget): void => {
+ const existing = byTarget.get(item.target);
+ if (!existing) {
+ byTarget.set(item.target, item);
+ return;
+ }
+ if (existing.tlsServerName && item.tlsServerName && existing.tlsServerName
!== item.tlsServerName) {
+ logger.warn(
+ 'Target %s resolves from multiple hostnames (%s vs %s); TLS SNI uses
%s',
+ item.target,
+ existing.tlsServerName,
+ item.tlsServerName,
+ existing.tlsServerName,
+ );
+ }
+ };
+
+ for (const entry of entries) {
+ if (!isValidHostPortEntry(entry)) {
+ continue;
+ }
+ const { host, port } = splitHostPort(entry);
+
+ if (!resolveDns || isLiteralIp(host)) {
+ addTarget({ target: formatHostPort(host, port) });
+ continue;
+ }
+
+ try {
+ const records = await lookupBackendHostRecords(host, lookup);
+ for (const record of records) {
+ addTarget({
+ target: formatHostPort(record.address, port),
+ tlsServerName: isLiteralIp(record.address) ? host : undefined,
+ });
+ }
+ } catch (error) {
+ logger.error('Failed to resolve %s of backend service.', host, error);
+ }
+ }
+
+ return Array.from(byTarget.values());
+}
+
+/**
+ * Fallback TLS SNI when per-target tlsServerName is unavailable (legacy /
static IP lists).
+ * Prefer {@link BackendTarget.tlsServerName} from {@link
expandBackendAddresses}.
+ */
+
+/** Count distinct non-IP hostnames in collectorAddress (comma-separated). */
+export function countDistinctNonIpHostnames(raw: string): number {
+ const hosts = new Set<string>();
+ for (const entry of parseStaticBackendAddresses(raw)) {
+ const { host } = splitHostPort(entry);
+ if (host && !isLiteralIp(host)) {
+ hosts.add(host.toLowerCase());
+ }
+ }
+ return hosts.size;
+}
+
+let multiHostnameTlsWarned = false;
+
+/** @deprecated Multi-hostname TLS SNI is handled per {@link BackendTarget};
kept for API compat. */
+export function warnMultiHostnameTlsConstraint(_collectorAddress: string):
void {
+ // no-op: per-target tlsServerName mapping supersedes first-hostname-only SNI
+}
+
+/** @internal test hook */
+export function resetMultiHostnameTlsWarnForTest(): void {
+ multiHostnameTlsWarned = false;
+}
+
+export function deriveTlsServerNameForConnectHost(connectHost: string,
collectorAddress: string): string | undefined {
Review Comment:
**[Low] Fallback SNI returns the *first* hostname in the list regardless of
which backend IP is actually being dialed → wrong
`grpc.ssl_target_name_override`.**
`deriveTlsServerNameForConnectHost` ignores `connectHost` and returns the
first non-IP host. Consumed at `GRPCChannelManager.ts:81` and
`TLSChannelBuilder.ts:88-91`. With e.g.
`collectorAddress='10.0.0.5:11800,secure-oap:11800'`,
`SW_AGENT_IS_RESOLVE_DNS_PERIODICALLY=false`, TLS on → dialing `10.0.0.5`
borrows `secure-oap` as SNI → cert-name mismatch / handshake under the wrong
authority. (`countDistinctNonIpHostnames` at L188 was clearly added
anticipating this but isn't used here.)
Suggested fix: drop this deprecated fallback and rely on the per-target
`BackendTarget.tlsServerName` from `expandBackendAddresses`, or only apply it
when exactly one distinct hostname exists in the config.
##########
src/agent/core/util/PrivateKeyUtil.ts:
##########
@@ -0,0 +1,96 @@
+/*!
+ *
+ * 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 fs from 'fs';
+import { promises as fsPromises } from 'fs';
+
+const PKCS_1_PEM_HEADER = '-----BEGIN RSA PRIVATE KEY-----';
+const PKCS_1_PEM_FOOTER = '-----END RSA PRIVATE KEY-----';
+const PKCS_8_PEM_HEADER = '-----BEGIN PRIVATE KEY-----';
+const PKCS_8_PEM_FOOTER = '-----END PRIVATE KEY-----';
+
+/**
+ * Load a RSA private key from PEM bytes (PKCS#1 or PKCS#8).
+ * Aligned with Java {@code PrivateKeyUtil.loadDecryptionKey}.
+ */
+export function loadDecryptionKeyFromBuffer(keyDataBytes: Buffer): Buffer {
+ let keyDataString = keyDataBytes.toString('utf8');
+
+ if (keyDataString.includes(PKCS_1_PEM_HEADER)) {
+ keyDataString = keyDataString.replace(PKCS_1_PEM_HEADER, '');
+ keyDataString = keyDataString.replace(PKCS_1_PEM_FOOTER, '');
+ keyDataString = keyDataString.replace(/\n/g, '');
+ const pkcs1Bytes = Buffer.from(keyDataString, 'base64');
+ return readPkcs1PrivateKey(pkcs1Bytes);
+ }
+
+ return keyDataBytes;
+}
+
+/**
+ * Load a RSA private key from a file (PEM PKCS#1 or PKCS#8).
+ * Aligned with Java {@code PrivateKeyUtil.loadDecryptionKey}.
+ */
+/** Preferred for runtime TLS loading (non-blocking). */
+export async function loadDecryptionKeyAsync(keyFilePath: string):
Promise<Buffer> {
+ return loadDecryptionKeyFromBuffer(await fsPromises.readFile(keyFilePath));
+}
+
+/** Synchronous loader retained for unit tests and legacy callers. */
+export function loadDecryptionKey(keyFilePath: string): Buffer {
+ return loadDecryptionKeyFromBuffer(fs.readFileSync(keyFilePath));
+}
+
+/** Convert raw PKCS#1 bytes into PKCS#8 PEM (Java readPkcs1PrivateKey). */
+function readPkcs1PrivateKey(pkcs1Bytes: Buffer): Buffer {
Review Comment:
**[Low] Hand-rolled PKCS#1→PKCS#8 ASN.1 re-wrapping is unnecessary in Node
and a latent corruption risk.**
This is a faithful port of the Java agent, where `KeyFactory` needs PKCS#8.
Node/OpenSSL don't: `channel-credentials.js:69-72` passes `key` straight into
`tls.createSecureContext`, and `crypto.createPrivateKey` /
`tls.createSecureContext` accept a raw `-----BEGIN RSA PRIVATE KEY-----`
(PKCS#1) PEM directly. Two latent issues: the fixed `0x82` (2-byte) length
prefix has no `0x81`/`0x83` branch (harmless only because real RSA keys exceed
256B), and the naive strip (only exact BEGIN/END lines + `\n`) would fold
`Proc-Type`/`DEK-Info` header lines of an *encrypted* PKCS#1 key into the
base64 → corrupt DER → opaque handshake error. No live bug on the supported
unencrypted-RSA path.
Suggested fix: drop the conversion and return the bytes unchanged
(`loadDecryptionKeyFromBuffer` just returns `keyDataBytes`), or, if a canonical
form is desired, `crypto.createPrivateKey(pem).export({ type: 'pkcs8', format:
'pem' })`.
##########
src/agent/core/remote/TLSChannelBuilder.ts:
##########
@@ -19,17 +19,84 @@
import * as grpc from '@grpc/grpc-js';
import config from '../../../config/AgentConfig';
+import { createLogger } from '../../../logging';
import ChannelBuilder, { ChannelBuildContext } from './ChannelBuilder';
+import { deriveTlsServerNameForConnectHost } from './BackendAddressResolver';
+import { getTlsMaterials, isCaFileAvailableFromPreload } from
'./TlsMaterialCache';
-/** When SW_AGENT_SECURE=true, upgrade channel credentials to TLS (Java
TLSChannelBuilder simplified). */
+const logger = createLogger(__filename);
+
+/** {@code SW_AGENT_FORCE_TLS} or {@code SW_AGENT_SECURE} (Node extension of
Java force_tls). */
+function isTlsRequiredByConfig(): boolean {
+ return Boolean(config.forceTls || config.secure);
+}
+
+/**
+ * Java {@code TLSChannelBuilder}: {@code FORCE_TLS || caFileExists} enables
TLS negotiation.
+ * CA presence comes from async {@link preloadTlsMaterials} (no sync stat on
the event loop).
+ */
+function shouldUseTls(): boolean {
+ return isTlsRequiredByConfig() || Boolean(getTlsMaterials()?.rootCerts) ||
isCaFileAvailableFromPreload();
+}
+
+export function isTlsEnabled(): boolean {
+ return shouldUseTls();
+}
+
+/**
+ * If only ca.crt exists, start TLS. If cert, key and ca files exist, enable
mTLS.
+ * Aligned with Java {@code TLSChannelBuilder}. TLS files must be preloaded via
+ * {@link preloadTlsMaterials} before channel build when a CA file is present.
+ */
export default class TLSChannelBuilder implements ChannelBuilder {
build(context: ChannelBuildContext): ChannelBuildContext {
- if (config.secure) {
- return {
- ...context,
- credentials: grpc.credentials.createSsl(),
- };
+ if (!shouldUseTls()) {
+ return context;
}
- return context;
+
+ const materials = getTlsMaterials();
+ const rootCerts = materials?.rootCerts ?? null;
+ const privateKey = materials?.privateKey ?? null;
+ const certChain = materials?.certChain ?? null;
+
+ let credentials: grpc.ChannelCredentials;
+
+ if (rootCerts) {
+ if (Boolean(config.sslCertChainPath) && Boolean(config.sslKeyPath)) {
Review Comment:
**[Nit] Redundant `if/else` — both branches call `createSsl` with identical
args.**
L70 and L72 are byte-identical `grpc.credentials.createSsl(rootCerts,
privateKey, certChain)`; the branch differs only by the null-guard throw
(L66-69). Since `preloadTlsMaterials` invariantly nulls
`privateKey`/`certChain` as a pair, `channel-credentials.js`'s own
key-without-chain throw is already unreachable. Collapse to a single
`createSsl` call and keep the guard (or drop it). Not a bug — just a divergence
hazard if one branch is edited later.
##########
src/agent/core/remote/TraceSegmentServiceClient.ts:
##########
@@ -165,44 +166,89 @@ export default class TraceSegmentServiceClient implements
BootService, GRPCChann
return;
}
+ let snapshots: Segment[] = [];
+ let settled = false;
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();
- },
- );
+ let writesStarted = false;
- for (const segment of this.buffer) {
- if (segment) {
- if (logger._isDebugEnabled) {
- logger.debug('Sending segment ', { segment });
- }
- stream.write(segment.transform());
+ const settle = (error?: unknown): void => {
+ if (settled) {
+ return;
+ }
+ settled = true;
+ if (error != null) {
+ logReportError('Failed to report trace data', error);
+ this.reportGrpcError(error);
+ if (!this.closed && !writesStarted) {
+ this.requeueSegments(snapshots);
}
}
- } catch (error) {
- logReportError('Failed to report trace data', error);
- this.reportGrpcError(error);
resolve();
- } finally {
- this.buffer.length = 0;
+ };
+
+ void (async () => {
try {
- stream?.end();
+ snapshots = this.buffer.splice(0, this.buffer.length);
+
+ stream = this.reporterClient!.collect(
+ new grpc.Metadata(),
+ { deadline: grpcUpstreamDeadlineMs() },
+ (error, commands) => {
+ if (error) {
+ settle(error);
+ } else {
+
ServiceManager.INSTANCE.findService(CommandService)?.receiveCommand(commands);
+ settle();
+ }
+ },
+ );
+
+ for (const segment of snapshots) {
+ if (this.closed) {
+ break;
+ }
+ if (segment) {
+ if (logger._isDebugEnabled) {
+ logger.debug(
+ 'Sending segment traceId=%s segmentId=%s spanCount=%d',
+ segment.relatedTraces[0]?.toString?.() ?? 'unknown',
+ segment.segmentId.toString(),
+ segment.spans.length,
+ );
+ }
+ let writeAccepted = stream.write(segment.transform());
+ writesStarted = true;
+ while (writeAccepted === false && stream && !this.closed) {
+ await new Promise<void>((resolveDrain) =>
stream!.once('drain', resolveDrain));
Review Comment:
**[Low] `await once('drain')` can hang and leak on RPC failure under
send-buffer pressure.**
This awaits `stream.once('drain')` with no error/close/timeout fallback, and
the `!this.closed` guard is only re-checked *after* drain resolves — so even
`shutdown()` can't unwind it. In grpc-js 1.14.4, `resolving-call.js:199` always
wraps in `createRetryingCall`; on a terminal status `retrying-call.js`
`reportStatus` resets `writeBuffer=[]` **without** firing pending buffer
callbacks, so the Writable `_write` cb (`call.js`) never completes, `'drain'`
never fires, and the client-streaming path (`client.js`) doesn't emit
`'error'`/`'close'` on the writable. The unary callback settles the outer
promise, but this inner async IIFE stays suspended forever — leaking the drain
listener and the spliced segment batch. Needs the batch to exceed the ~1MB
per-RPC retry buffer AND a deadline/failure mid-stream, hence low.
Suggested fix — race the drain against terminal signals (or just drop
remaining writes on backpressure like `MeterSender` does):
```ts
await Promise.race([
new Promise((r) => stream!.once('drain', r)),
new Promise((r) => stream!.once('error', r)),
new Promise((r) => stream!.once('close', r)),
]);
if (this.closed) break;
```
##########
src/agent/core/meter/RuntimeSampler.ts:
##########
@@ -47,15 +55,30 @@ export default class RuntimeSampler {
const cpuScale = elapsedMicros > 0 ? 100 / elapsedMicros /
this.logicalCpuCount : 0;
const cpuUserPercent = cpuUsage.user * cpuScale;
const cpuSystemPercent = cpuUsage.system * cpuScale;
+ const heapSpaceDetail = config.runtimeMetricsHeapSpaceDetail !== false;
+ const heapSpaceUsed = (spaceName: string): number => {
+ if (!heapSpaceDetail) {
+ return 0;
+ }
+ const heapSpaces = v8.getHeapSpaceStatistics();
Review Comment:
**[Nit] `v8.getHeapSpaceStatistics()` runs twice per sample.**
This `heapSpaceUsed` closure calls `v8.getHeapSpaceStatistics()` on every
invocation, and it's invoked twice per `sample()` — L80 (`old_space`) and L81
(`new_space`) — rebuilding the full heap-space array both times. Fetch it once
per `sample()` and read both `space_used_size` values from the single result.
While here: the adjacent (pre-existing, not introduced by this PR) CPU logic
reads `process.cpuUsage()` twice — `process.cpuUsage(this.lastCpuUsage)` for
the delta, then `this.lastCpuUsage = process.cpuUsage()` as a fresh baseline —
so CPU used between the two reads is attributed to neither interval. Reading
once and reusing the value would be more accurate.
##########
src/agent/core/remote/BackendAddressResolver.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 { promises as dnsPromises } from 'dns';
+import net from 'net';
+import { createLogger } from '../../../logging';
+
+const logger = createLogger(__filename);
+
+/** Cap DNS expansion to avoid unbounded memory on hostile or misconfigured
resolvers. */
+export const MAX_DNS_LOOKUP_RECORDS = 64;
+
+/** Abort slow DNS lookups so channel checks do not stall the event loop. */
+export const DNS_LOOKUP_TIMEOUT_MS = 5_000;
+
+/** Resolved gRPC target with optional TLS SNI hostname (H-2 per-entry
mapping). */
+export interface BackendTarget {
+ target: string;
+ /** Hostname for grpc.ssl_target_name_override when target is a resolved IP.
*/
+ tlsServerName?: string;
+}
+
+export type DnsLookupFn = (
+ hostname: string,
+ options: { all: true; verbatim?: boolean },
+) => Promise<Array<{ address: string; family: number }>>;
+
+const defaultLookup: DnsLookupFn = (hostname, options) =>
+ dnsPromises.lookup(hostname, options) as ReturnType<DnsLookupFn>;
+
+export async function lookupBackendHostRecords(
+ hostname: string,
+ lookup: DnsLookupFn = defaultLookup,
+ timeoutMs: number = DNS_LOOKUP_TIMEOUT_MS,
+): Promise<Array<{ address: string; family: number }>> {
+ let timeoutHandle: NodeJS.Timeout | undefined;
+ try {
+ const records = await Promise.race([
+ lookup(hostname, { all: true, verbatim: true }),
+ new Promise<never>((_, reject) => {
+ timeoutHandle = setTimeout(() => reject(new Error('DNS lookup
timeout')), timeoutMs);
+ timeoutHandle.unref();
+ }),
+ ]);
+ if (records.length > MAX_DNS_LOOKUP_RECORDS) {
+ logger.warn('DNS returned %s records for %s; using first %s',
records.length, hostname, MAX_DNS_LOOKUP_RECORDS);
+ return records.slice(0, MAX_DNS_LOOKUP_RECORDS);
+ }
+ return records;
+ } finally {
+ if (timeoutHandle) {
+ clearTimeout(timeoutHandle);
+ }
+ }
+}
+
+/** Parse comma-separated backend entries (Java BACKEND_SERVICE split). */
+export function parseStaticBackendAddresses(raw: string): string[] {
+ return raw
+ .split(',')
+ .map((entry) => entry.trim())
+ .filter(Boolean)
+ .filter(isValidHostPortEntry);
+}
+
+function isValidHostPortEntry(entry: string): boolean {
+ try {
+ const { host, port } = splitHostPort(entry);
+ const portNum = Number.parseInt(port, 10);
+ return Boolean(host) && !Number.isNaN(portNum) && portNum > 0 && portNum
<= 65535;
+ } catch {
+ logger.debug('Service address [%s] format error. Expected host:port',
entry);
+ return false;
+ }
+}
+
+/** Split host:port, supporting bracketed IPv6 literals such as [::1]:11800. */
+export function splitHostPort(entry: string): { host: string; port: string } {
+ const trimmed = entry.trim();
+ if (!trimmed) {
+ throw new Error(`Invalid host:port entry: ${entry}`);
+ }
+ if (trimmed.startsWith('[')) {
+ const close = trimmed.indexOf(']');
+ if (close < 0 || trimmed[close + 1] !== ':') {
+ throw new Error(`Invalid host:port entry: ${entry}`);
+ }
+ return { host: trimmed.slice(1, close), port: trimmed.slice(close + 2) };
+ }
+ const lastColon = trimmed.lastIndexOf(':');
+ if (lastColon <= 0) {
+ throw new Error(`Invalid host:port entry: ${entry}`);
+ }
+ return { host: trimmed.slice(0, lastColon), port: trimmed.slice(lastColon +
1) };
+}
+
+/** Format host:port for grpc target strings; IPv6 hosts are bracketed. */
+export function formatHostPort(host: string, port: string | number): string {
+ const portText = String(port);
+ if (net.isIPv6(host)) {
+ return `[${host}]:${portText}`;
+ }
+ return `${host}:${portText}`;
+}
+
+export function isLiteralIp(host: string): boolean {
+ return net.isIP(host) !== 0;
+}
+
+/**
+ * Expand backend entries to targets (Java InetAddress.getAllByName).
+ * When resolveDns=false, returns static entries only.
+ * DNS-expanded IPs carry tlsServerName from the source hostname for grpc-js
SNI.
+ */
+export async function expandBackendAddresses(
+ entries: string[],
+ resolveDns: boolean,
+ lookup: DnsLookupFn = defaultLookup,
+): Promise<BackendTarget[]> {
+ const byTarget = new Map<string, BackendTarget>();
+
+ const addTarget = (item: BackendTarget): void => {
+ const existing = byTarget.get(item.target);
+ if (!existing) {
+ byTarget.set(item.target, item);
+ return;
+ }
+ if (existing.tlsServerName && item.tlsServerName && existing.tlsServerName
!== item.tlsServerName) {
+ logger.warn(
+ 'Target %s resolves from multiple hostnames (%s vs %s); TLS SNI uses
%s',
+ item.target,
+ existing.tlsServerName,
+ item.tlsServerName,
+ existing.tlsServerName,
+ );
+ }
+ };
+
+ for (const entry of entries) {
+ if (!isValidHostPortEntry(entry)) {
+ continue;
+ }
+ const { host, port } = splitHostPort(entry);
+
+ if (!resolveDns || isLiteralIp(host)) {
+ addTarget({ target: formatHostPort(host, port) });
+ continue;
+ }
+
+ try {
+ const records = await lookupBackendHostRecords(host, lookup);
+ for (const record of records) {
+ addTarget({
+ target: formatHostPort(record.address, port),
+ tlsServerName: isLiteralIp(record.address) ? host : undefined,
+ });
+ }
+ } catch (error) {
+ logger.error('Failed to resolve %s of backend service.', host, error);
+ }
+ }
+
+ return Array.from(byTarget.values());
+}
+
+/**
+ * Fallback TLS SNI when per-target tlsServerName is unavailable (legacy /
static IP lists).
+ * Prefer {@link BackendTarget.tlsServerName} from {@link
expandBackendAddresses}.
+ */
+
+/** Count distinct non-IP hostnames in collectorAddress (comma-separated). */
+export function countDistinctNonIpHostnames(raw: string): number {
+ const hosts = new Set<string>();
+ for (const entry of parseStaticBackendAddresses(raw)) {
+ const { host } = splitHostPort(entry);
+ if (host && !isLiteralIp(host)) {
+ hosts.add(host.toLowerCase());
+ }
+ }
+ return hosts.size;
+}
+
+let multiHostnameTlsWarned = false;
+
+/** @deprecated Multi-hostname TLS SNI is handled per {@link BackendTarget};
kept for API compat. */
+export function warnMultiHostnameTlsConstraint(_collectorAddress: string):
void {
Review Comment:
**[Nit] Dead leftovers from the pre-per-target-SNI design.**
`warnMultiHostnameTlsConstraint` is an exported `@deprecated` **no-op**,
`multiHostnameTlsWarned` (L199) is write-only (never read), and
`resetMultiHostnameTlsWarnForTest` exists only for a test that asserts the
no-op does nothing. A `warn…`-named function that no longer warns is a
readability trap. Suggest deleting all three (and the dangling doc-comment at
L182-185) now that per-target `tlsServerName` supersedes them.
##########
src/agent/core/remote/TlsMaterialCache.ts:
##########
@@ -0,0 +1,210 @@
+/*!
+ *
+ * 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 fs from 'fs';
+import { promises as fsPromises } from 'fs';
+import config from '../../../config/AgentConfig';
+import { getAgentPackagePath, isPathInsideAgentRoot, resolveAgentPath } from
'../boot/AgentPackagePath';
+import path from 'path';
+import { loadDecryptionKeyFromBuffer } from '../util/PrivateKeyUtil';
+import { createLogger } from '../../../logging';
+
+const logger = createLogger(__filename);
+
+/** Max TLS PEM/CA file size (256 KiB) to limit memory DoS. */
+const MAX_TLS_MATERIAL_BYTES = 256 * 1024;
+
+export type TlsMaterialSnapshot = {
+ rootCerts: Buffer | null;
+ privateKey: Buffer | null;
+ certChain: Buffer | null;
+};
+
+/** Last preload result consumed by {@code TLSChannelBuilder.build()} in the
same rebuild. */
+let cachedSnapshot: TlsMaterialSnapshot | null = null;
+
+/** Whether the trusted CA file existed on the last {@link
preloadTlsMaterials} call. */
+let caFileExistsOnLastPreload = false;
+
+/** Read a regular file; reject symlinks (B-1 path traversal via link). */
+async function readRegularFileBytes(
+ filePath: string,
+ options: { mustStayUnderAgentRoot?: boolean; requirePrivateKeyMode?: boolean
} = {},
+): Promise<Buffer | null> {
+ try {
+ const lstat = await fsPromises.lstat(filePath);
+ if (lstat.isSymbolicLink()) {
+ logger.warn('TLS path [%s] is a symbolic link; refusing to load',
filePath);
+ return null;
+ }
+ if (!lstat.isFile()) {
+ return null;
+ }
+ if (lstat.size > MAX_TLS_MATERIAL_BYTES) {
+ logger.warn('TLS path [%s] exceeds max size %d bytes; refusing to load',
filePath, MAX_TLS_MATERIAL_BYTES);
+ return null;
+ }
+ if (options.requirePrivateKeyMode && (lstat.mode & 0o077) !== 0) {
+ logger.warn(
+ 'Private key [%s] is group/world accessible (mode %o); refusing to
load',
+ filePath,
+ lstat.mode & 0o777,
+ );
+ return null;
+ }
+ const realPath = await fsPromises.realpath(filePath);
+ if (options.mustStayUnderAgentRoot && !isPathInsideAgentRoot(realPath)) {
+ logger.warn('TLS path [%s] resolves outside agent package [%s]; refusing
to load', filePath, realPath);
+ return null;
+ }
+ const openFlags = fs.constants.O_RDONLY | (fs.constants.O_NOFOLLOW ?? 0);
+ const handle = await fsPromises.open(filePath, openFlags);
+ try {
+ return await handle.readFile();
+ } finally {
+ await handle.close();
+ }
+ } catch {
+ return null;
+ }
+}
+
+async function isRegularFilePath(filePath: string): Promise<boolean> {
+ try {
+ const lstat = await fsPromises.lstat(filePath);
+ return lstat.isFile() && !lstat.isSymbolicLink();
+ } catch {
+ return false;
+ }
+}
+
+function resolveConfiguredPath(configuredPath: string | undefined): string |
undefined {
+ if (!configuredPath) {
+ return undefined;
+ }
+ try {
+ return resolveAgentPath(configuredPath);
+ } catch (error) {
+ logger.error('Invalid TLS path [%s]: %s', configuredPath, error);
+ return undefined;
+ }
+}
+
+async function readFileIfExists(configuredPath: string | undefined):
Promise<Buffer | null> {
+ const filePath = resolveConfiguredPath(configuredPath);
+ if (!filePath) {
+ return null;
+ }
+ const mustStay = !path.isAbsolute((configuredPath ?? '').trim());
+ return readRegularFileBytes(filePath, { mustStayUnderAgentRoot: mustStay });
+}
+
+async function readPrivateKeyIfExists(configuredPath: string | undefined):
Promise<Buffer | null> {
+ const filePath = resolveConfiguredPath(configuredPath);
+ if (!filePath) {
+ return null;
+ }
+ const mustStay = !path.isAbsolute((configuredPath ?? '').trim());
+ const keyBytes = await readRegularFileBytes(filePath, {
+ mustStayUnderAgentRoot: mustStay,
+ requirePrivateKeyMode: true,
+ });
+ if (!keyBytes) {
+ return null;
+ }
+ try {
+ return loadDecryptionKeyFromBuffer(keyBytes);
+ } catch {
+ return null;
+ }
+}
+
+/**
+ * Async-load TLS files before channel rebuild (Java {@code TLSChannelBuilder}
reads on every build).
+ * Reloads from disk on each call so rotated certificates are picked up after
failover/reconnect.
+ */
+async function probeCaFileExists(): Promise<boolean> {
+ const filePath = resolveConfiguredPath(config.sslTrustedCaPath);
+ if (!filePath) {
+ return false;
+ }
+ if (!(await isRegularFilePath(filePath))) {
+ return false;
+ }
+ if (!path.isAbsolute((config.sslTrustedCaPath ?? '').trim())) {
+ try {
+ const realPath = await fsPromises.realpath(filePath);
+ return isPathInsideAgentRoot(realPath);
+ } catch {
+ return false;
+ }
+ }
+ return true;
+}
+
+export async function preloadTlsMaterials(): Promise<TlsMaterialSnapshot> {
+ caFileExistsOnLastPreload = await probeCaFileExists();
+ const rootCerts = await readFileIfExists(config.sslTrustedCaPath);
+ let privateKey = await readPrivateKeyIfExists(config.sslKeyPath);
+ let certChain = await readFileIfExists(config.sslCertChainPath);
+
+ const certPathSet = Boolean(config.sslCertChainPath);
+ const keyPathSet = Boolean(config.sslKeyPath);
+ if (certPathSet && keyPathSet && (!privateKey || !certChain)) {
+ privateKey = null;
+ certChain = null;
+ } else if (certPathSet !== keyPathSet) {
+ privateKey = null;
+ certChain = null;
+ }
+
+ cachedSnapshot = { rootCerts, privateKey, certChain };
+ return cachedSnapshot;
+}
+
+export function getTlsMaterials(): TlsMaterialSnapshot | null {
+ return cachedSnapshot;
+}
+
+/** Whether trusted CA file existed during the last preload (replaces sync
stat in TLSChannelBuilder). */
+export function isCaFileAvailableFromPreload(): boolean {
+ return caFileExistsOnLastPreload;
+}
+
+/** @internal test hook — reset preload cache between tests. */
+function zeroSensitiveBuffers(snapshot: TlsMaterialSnapshot | null): void {
Review Comment:
**[Nit] Key-wipe is necessarily incomplete — worth a comment so it doesn't
imply full erasure.**
`zeroSensitiveBuffers` fills only `cachedSnapshot.privateKey`. By shutdown,
`createSsl → tls.createSecureContext` (`channel-credentials.js:69-72`) has
copied the key into a native OpenSSL `SecureContext` unreachable from JS, and
(on the PKCS#1 path) `keyDataBytes.toString('utf8')` created an immutable,
unwipeable string plus distinct intermediate buffers. So the wipe can't fully
erase the key. It's not a defect (it does what JS can), but a short comment
noting the OpenSSL-copy + immutable-string limitation would avoid a false sense
of erasure. (On the common PKCS#8 path the cached buffer *is* the file-read
buffer, so wiping it does help.)
--
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]
