guan404ming commented on code in PR #69302:
URL: https://github.com/apache/airflow/pull/69302#discussion_r3557883928
##########
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:
Just opened a follow up pr to fix that. Please take another look, thanks!
https://github.com/apache/airflow/pull/69696/
--
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]
