guan404ming commented on code in PR #69302:
URL: https://github.com/apache/airflow/pull/69302#discussion_r3564661410
##########
ts-sdk/src/coordinator/log-channel.ts:
##########
@@ -47,31 +47,90 @@ export interface LogRecord {
const DEFAULT_LOGGER_NAME = "ts-sdk";
+export const RECONNECT_DELAY_MS = 250;
+
+// Socket state shared by the root logger and all its children, so a
+// disconnect observed by one is a disconnect for all. Mirrors the Go
+// SDK's SocketLogHandler: records emitted while disconnected are
+// buffered (already serialized) and flushed in order on reconnect.
+interface SharedSocket {
+ sock: Socket;
+ addr: string;
+ connected: boolean;
+ closed: boolean;
+ buffered: Buffer[];
+ reconnectTimer: NodeJS.Timeout | null;
+}
+
export class LogChannel {
- private readonly sock: Socket;
+ private readonly shared: SharedSocket;
private readonly name: string;
private readonly isRoot: boolean;
- private constructor(sock: Socket, name: string, isRoot: boolean) {
- this.sock = sock;
+ private constructor(shared: SharedSocket, name: string, isRoot: boolean) {
+ this.shared = shared;
this.name = name;
this.isRoot = isRoot;
- if (isRoot) {
- sock.on("error", (err) => {
- process.stderr.write(`[${this.name}] log socket error:
${err.message}\n`);
- });
- }
}
static async connect(addr: string, name: string = DEFAULT_LOGGER_NAME):
Promise<LogChannel> {
- return new LogChannel(await connectTcp(addr), name, true);
+ const shared: SharedSocket = {
+ sock: await connectTcp(addr),
+ addr,
+ connected: true,
+ closed: false,
+ buffered: [],
+ reconnectTimer: null,
+ };
+ const channel = new LogChannel(shared, name, true);
+ channel.watchSocket(shared.sock);
+ return channel;
}
/** Create a sibling logger that shares the underlying socket but
* carries a hierarchical name (`parent.suffix`). Only the root
* owns the socket — children's `close()` is a no-op. */
child(suffix: string): LogChannel {
- return new LogChannel(this.sock, `${this.name}.${suffix}`, false);
+ return new LogChannel(this.shared, `${this.name}.${suffix}`, false);
+ }
+
+ private watchSocket(sock: Socket): void {
+ sock.on("error", (err) => {
+ process.stderr.write(`[${this.name}] log socket error:
${err.message}\n`);
+ });
+ sock.on("close", () => {
+ if (this.shared.closed || this.shared.sock !== sock) return;
+ this.shared.connected = false;
+ this.scheduleReconnect();
Review Comment:
Agreed with the concerns here and in #69696. I think there's nothing to
buffer before the initial connection, and reconnecting doesn't help much for a
same-host socket. Reworking this PR to treat a disconnect as fatal and surface
it via stderr instead.
--
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]
