shivaam commented on code in PR #69302:
URL: https://github.com/apache/airflow/pull/69302#discussion_r3555077813
##########
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
Review Comment:
I think we can remove the reference to GO from the comments as they will
quickly become obsolete. You can put a comment directly on the diff or in the
pr description in this case.
##########
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:
When we reconnect the socket, the airflow coordinator side needs to accept
the reconnection again. Can you please verify this part?
##########
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 {
Review Comment:
I prefer this to be called LogChannelState as it includes both the socket
and the buffer.
--
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]
