You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "guan404ming (via GitHub)" <gi...@apache.org> on 2026/07/03 06:40:59 UTC
[PR] Buffer TypeScript SDK logs across log socket disconnects [airflow]
guan404ming opened a new pull request, #69302:
URL: https://github.com/apache/airflow/pull/69302
<!-- SPDX-License-Identifier: Apache-2.0
https://www.apache.org/licenses/LICENSE-2.0 -->
<!--
Thank you for contributing!
Please provide above a brief description of the changes made in this pull request.
Write a good git commit message following this guide: http://chris.beams.io/posts/git-commit/
Please make sure that your code changes are covered with tests.
And in case of new features or big changes remember to adjust the documentation.
For user-facing UI changes, please attach before/after screenshots (or a short
screen recording) so reviewers can assess the visual impact.
Feel free to ping (in general) for the review if you do not see reaction for a few days
(72 Hours is the minimum reaction time you can expect from volunteers) - we sometimes miss notifications.
In case of an existing issue, reference it using one of the following:
* closes: #ISSUE
* related: #ISSUE
-->
## Related Issue
closes: #69292
## Why
Logs emitted while the log socket is disconnected are silently lost, unlike the Java and Go SDK runtimes which buffer and flush them.
## How
- Buffer serialized log records while the socket is down and flush them in order on reconnect
- Reconnect automatically every 250ms after an unexpected socket close, stopping on close()
- Share socket state between the root logger and its children so all observe the same connection
---
##### Was generative AI tooling used to co-author this PR?
<!--
If generative AI tooling has been used in the process of authoring this PR, please
change below checkbox to `[X]` followed by the name of the tool, uncomment the "Generated-by".
-->
- [ ] Yes (please specify the tool below)
<!--
Generated-by: [Tool Name] following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
-->
---
* Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)** for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
* For fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
* When adding dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
* For significant user-facing changes create newsfragment: `{pr_number}.significant.rst`, in [airflow-core/newsfragments](https://github.com/apache/airflow/tree/main/airflow-core/newsfragments). You can add this file in a follow-up commit after the PR is created so you know the PR number.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Buffer TypeScript SDK logs across log socket disconnects [airflow]
Posted by "guan404ming (via GitHub)" <gi...@apache.org>.
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Send TS SDK logs to stderr on socket disconnect [airflow]
Posted by "shivaam (via GitHub)" <gi...@apache.org>.
shivaam commented on code in PR #69302:
URL: https://github.com/apache/airflow/pull/69302#discussion_r3597997826
##########
ts-sdk/src/coordinator/log-channel.ts:
##########
@@ -47,31 +47,45 @@ export interface LogRecord {
const DEFAULT_LOGGER_NAME = "ts-sdk";
+interface LogChannelState {
+ sock: Socket;
+ connected: boolean;
+ closed: boolean;
+}
+
Review Comment:
I agree with Guan-ming we dont need 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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Buffer TypeScript SDK logs across log socket disconnects [airflow]
Posted by "guan404ming (via GitHub)" <gi...@apache.org>.
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Send TS SDK logs to stderr on socket disconnect [airflow]
Posted by "guan404ming (via GitHub)" <gi...@apache.org>.
guan404ming commented on code in PR #69302:
URL: https://github.com/apache/airflow/pull/69302#discussion_r3595950192
##########
ts-sdk/src/coordinator/log-channel.ts:
##########
@@ -47,31 +47,45 @@ export interface LogRecord {
const DEFAULT_LOGGER_NAME = "ts-sdk";
+interface LogChannelState {
+ sock: Socket;
+ connected: boolean;
+ closed: boolean;
+}
+
Review Comment:
Thanks for the suggestion. One question though `startCoordinator()` connects the log socket first, and a `LogChannel` only exists after connect(), so I think that there may be no record can be emitted pre-connect. The only pre-connect output is genuine failures, where stderr at ERROR seems semantically right. Feel free to correct me if I miss any cases, thanks!
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Send TS SDK logs to stderr on socket disconnect [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #69302:
URL: https://github.com/apache/airflow/pull/69302#discussion_r3608252730
##########
ts-sdk/src/coordinator/log-channel.ts:
##########
@@ -47,31 +47,45 @@ export interface LogRecord {
const DEFAULT_LOGGER_NAME = "ts-sdk";
+interface LogChannelState {
+ sock: Socket;
+ connected: boolean;
+ closed: boolean;
+}
+
Review Comment:
Sorry for the late reply:
The purpose of the buffer is different. Take the following sequence as example:
1. TS subprocess started.
2. TS execution time logger can still logs somethings at this moment! (e.g. trying to connect to coordinator `--comms=127:<port> ...` ..
3. TS subprocess really connected to the coordinator.
4. All the logs from TS logger will forward to `--log` socket after connected.
The buffer I mentioned handle the stage 2 where the TS logger can still log before connected to the coordinator log socket. We should "buffer" the logs at stage 2 and pass though to the log socket once connected.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Send TS SDK logs to stderr on socket disconnect [airflow]
Posted by "guan404ming (via GitHub)" <gi...@apache.org>.
guan404ming commented on code in PR #69302:
URL: https://github.com/apache/airflow/pull/69302#discussion_r3608505527
##########
ts-sdk/src/coordinator/log-channel.ts:
##########
@@ -120,8 +142,20 @@ export class LogChannel {
async close(): Promise<void> {
if (!this.isRoot) return;
+ this.shared.closed = true;
+ if (!this.shared.connected) {
+ this.shared.sock.destroy();
+ return;
+ }
return new Promise((resolve) => {
- this.sock.end(() => resolve());
+ const timer = setTimeout(() => {
+ this.shared.sock.destroy();
+ resolve();
+ }, CLOSE_FLUSH_TIMEOUT_MS);
Review Comment:
Neat one, thanks for the suggestion. Just updated.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Send TS SDK logs to stderr on socket disconnect [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #69302:
URL: https://github.com/apache/airflow/pull/69302#discussion_r3608252730
##########
ts-sdk/src/coordinator/log-channel.ts:
##########
@@ -47,31 +47,45 @@ export interface LogRecord {
const DEFAULT_LOGGER_NAME = "ts-sdk";
+interface LogChannelState {
+ sock: Socket;
+ connected: boolean;
+ closed: boolean;
+}
+
Review Comment:
Sorry for the late reply:
The purpose of the buffer is different. No matter the final task state is success or fail will need to handle the following scenario. (Both Java and Go SDK handle this.)
Here's the lifecycle of the Lang-SDK subprocess.
1. TS subprocess started (by the `popen` on coordinator side)
2. TS execution time logger can still logs somethings at this moment! (e.g. trying to connect to coordinator `--comms=127:<port> ...` ..
3. TS subprocess really connected to the coordinator.
4. All the logs from TS logger will forward to `--log` socket after connected.
The buffer I mentioned handle the stage 2 where the TS logger can still log before connected to the coordinator log socket. We should "buffer" the logs at stage 2 and pass though to the log socket once connected.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Buffer TypeScript SDK logs across log socket disconnects [airflow]
Posted by "shivaam (via GitHub)" <gi...@apache.org>.
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Send TS SDK logs to stderr on socket disconnect [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #69302:
URL: https://github.com/apache/airflow/pull/69302#discussion_r3608385908
##########
ts-sdk/src/coordinator/log-channel.ts:
##########
@@ -99,7 +114,12 @@ export class LogChannel {
event: `[${this.name}] ${record.event}`,
timestamp: record.timestamp ?? new Date().toISOString(),
});
- this.sock.write(Buffer.from(line + "\n", "utf8"));
+ const payload = line + "\n";
+ if (!this.shared.connected) {
+ process.stderr.write(payload);
+ return;
+ }
+ this.shared.sock.write(Buffer.from(payload, "utf8"));
Review Comment:
If the peer dies but the error/close hasn't been processed, `connected` is still true, so `shared.sock.write()` runs on a dead socket and the record is lost. Maybe also check `shared.sock.writable` before writing and fall back to stderr if false?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Send TS SDK logs to stderr on socket disconnect [airflow]
Posted by "guan404ming (via GitHub)" <gi...@apache.org>.
guan404ming commented on code in PR #69302:
URL: https://github.com/apache/airflow/pull/69302#discussion_r3608457252
##########
ts-sdk/src/coordinator/log-channel.ts:
##########
@@ -120,8 +140,13 @@ export class LogChannel {
async close(): Promise<void> {
if (!this.isRoot) return;
+ this.shared.closed = true;
+ if (!this.shared.connected) {
+ this.shared.sock.destroy();
+ return;
+ }
return new Promise((resolve) => {
- this.sock.end(() => resolve());
+ this.shared.sock.end(() => resolve());
Review Comment:
Thanks for catching this. Added a 3s timeout racing the graceful end() then on timeout the socket is destroyed and close() resolves.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Send TS SDK logs to stderr on socket disconnect [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 merged PR #69302:
URL: https://github.com/apache/airflow/pull/69302
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Send TypeScript SDK logs to stderr when the log socket disconnects [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #69302:
URL: https://github.com/apache/airflow/pull/69302#discussion_r3568422995
##########
ts-sdk/src/coordinator/log-channel.ts:
##########
@@ -47,31 +47,45 @@ export interface LogRecord {
const DEFAULT_LOGGER_NAME = "ts-sdk";
+interface LogChannelState {
+ sock: Socket;
+ connected: boolean;
+ closed: boolean;
+}
+
export class LogChannel {
- private readonly sock: Socket;
+ private readonly shared: LogChannelState;
private readonly name: string;
private readonly isRoot: boolean;
- private constructor(sock: Socket, name: string, isRoot: boolean) {
- this.sock = sock;
+ private constructor(shared: LogChannelState, 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: LogChannelState = {
+ sock: await connectTcp(addr),
+ connected: true,
+ closed: false,
+ };
+ shared.sock.on("error", (err) => {
+ process.stderr.write(`[${name}] log socket error: ${err.message}\n`);
+ });
Review Comment:
Caught by Claude:
The `"error"` handler only writes to stderr; only the later `"close"` handler flips `shared.connected = false`. Node fires `"close"` a tick or so after `"error"` (empirically ~0.05ms later in a local repro), and in that gap a `send()` call still takes the "connected" branch and writes to an already-destroyed socket — the record is silently dropped instead of falling back to stderr as intended. Setting the flag in the `error` handler too closes the gap:
```suggestion
shared.sock.on("error", (err) => {
shared.connected = false;
process.stderr.write(`[${name}] log socket error: ${err.message}\n`);
});
```
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Send TS SDK logs to stderr on socket disconnect [airflow]
Posted by "guan404ming (via GitHub)" <gi...@apache.org>.
guan404ming commented on code in PR #69302:
URL: https://github.com/apache/airflow/pull/69302#discussion_r3595780911
##########
ts-sdk/src/coordinator/log-channel.ts:
##########
@@ -47,31 +47,45 @@ export interface LogRecord {
const DEFAULT_LOGGER_NAME = "ts-sdk";
+interface LogChannelState {
+ sock: Socket;
+ connected: boolean;
+ closed: boolean;
+}
+
export class LogChannel {
- private readonly sock: Socket;
+ private readonly shared: LogChannelState;
private readonly name: string;
private readonly isRoot: boolean;
- private constructor(sock: Socket, name: string, isRoot: boolean) {
- this.sock = sock;
+ private constructor(shared: LogChannelState, 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: LogChannelState = {
+ sock: await connectTcp(addr),
+ connected: true,
+ closed: false,
+ };
+ shared.sock.on("error", (err) => {
+ process.stderr.write(`[${name}] log socket error: ${err.message}\n`);
+ });
Review Comment:
Great catch, that is really neat.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Send TS SDK logs to stderr on socket disconnect [airflow]
Posted by "guan404ming (via GitHub)" <gi...@apache.org>.
guan404ming commented on PR #69302:
URL: https://github.com/apache/airflow/pull/69302#issuecomment-4992605133
> Continue on https://github.com/apache/airflow/pull/69696#issuecomment-4947946260 this discussion. IIUC, It seems the current PR already refactor as what you mentioned in above comment right?
Yes, this one has been refactored as what we discussed there!
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Send TS SDK logs to stderr on socket disconnect [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #69302:
URL: https://github.com/apache/airflow/pull/69302#discussion_r3608381650
##########
ts-sdk/src/coordinator/log-channel.ts:
##########
@@ -47,31 +47,46 @@ export interface LogRecord {
const DEFAULT_LOGGER_NAME = "ts-sdk";
+interface LogChannelState {
+ sock: Socket;
+ connected: boolean;
+ closed: boolean;
+}
+
export class LogChannel {
- private readonly sock: Socket;
+ private readonly shared: LogChannelState;
private readonly name: string;
private readonly isRoot: boolean;
- private constructor(sock: Socket, name: string, isRoot: boolean) {
- this.sock = sock;
+ private constructor(shared: LogChannelState, 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: LogChannelState = {
+ sock: await connectTcp(addr),
+ connected: true,
+ closed: false,
+ };
+ shared.sock.on("error", (err) => {
+ shared.connected = false;
+ process.stderr.write(`[${name}] log socket error: ${err.message}\n`);
+ });
+ shared.sock.on("close", () => {
+ if (shared.closed) return;
+ shared.connected = false;
+ process.stderr.write(`[${name}] log socket closed unexpectedly; further logs go to stderr\n`);
+ });
Review Comment:
These two tend to fire together in errors like EPIPE. Maybe add a `if (shared.connected)` (or `closed`) check so only one of the error messages are written in this case?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Send TS SDK logs to stderr on socket disconnect [airflow]
Posted by "guan404ming (via GitHub)" <gi...@apache.org>.
guan404ming commented on code in PR #69302:
URL: https://github.com/apache/airflow/pull/69302#discussion_r3608363125
##########
ts-sdk/src/coordinator/log-channel.ts:
##########
@@ -47,31 +47,45 @@ export interface LogRecord {
const DEFAULT_LOGGER_NAME = "ts-sdk";
+interface LogChannelState {
+ sock: Socket;
+ connected: boolean;
+ closed: boolean;
+}
+
Review Comment:
That makes sense to me to add it in stage 2. I could help add in a follow up!
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Send TS SDK logs to stderr on socket disconnect [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #69302:
URL: https://github.com/apache/airflow/pull/69302#discussion_r3608497146
##########
ts-sdk/src/coordinator/log-channel.ts:
##########
@@ -120,8 +142,20 @@ export class LogChannel {
async close(): Promise<void> {
if (!this.isRoot) return;
+ this.shared.closed = true;
+ if (!this.shared.connected) {
+ this.shared.sock.destroy();
+ return;
+ }
return new Promise((resolve) => {
- this.sock.end(() => resolve());
+ const timer = setTimeout(() => {
+ this.shared.sock.destroy();
+ resolve();
+ }, CLOSE_FLUSH_TIMEOUT_MS);
Review Comment:
We can add a `timer.unref()` to avoid the process not being closed before the timer fires. (Since we’re just cleaning up, the timer does not need to fire if the process is dying anyway.) Not a big deal.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Send TS SDK logs to stderr on socket disconnect [airflow]
Posted by "guan404ming (via GitHub)" <gi...@apache.org>.
guan404ming commented on code in PR #69302:
URL: https://github.com/apache/airflow/pull/69302#discussion_r3608456024
##########
ts-sdk/src/coordinator/log-channel.ts:
##########
@@ -99,7 +114,12 @@ export class LogChannel {
event: `[${this.name}] ${record.event}`,
timestamp: record.timestamp ?? new Date().toISOString(),
});
- this.sock.write(Buffer.from(line + "\n", "utf8"));
+ const payload = line + "\n";
+ if (!this.shared.connected) {
+ process.stderr.write(payload);
+ return;
+ }
+ this.shared.sock.write(Buffer.from(payload, "utf8"));
Review Comment:
Good point, I do not cover the case when implementation. `send()` now also checks `sock.writable` and falls back to stderr in that window. Test added.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Send TS SDK logs to stderr on socket disconnect [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #69302:
URL: https://github.com/apache/airflow/pull/69302#discussion_r3608461856
##########
ts-sdk/src/coordinator/log-channel.ts:
##########
@@ -47,31 +47,45 @@ export interface LogRecord {
const DEFAULT_LOGGER_NAME = "ts-sdk";
+interface LogChannelState {
+ sock: Socket;
+ connected: boolean;
+ closed: boolean;
+}
+
Review Comment:
Thanks! Just FYI that please feel free to reference the diagram in https://github.com/apache/airflow/pull/70076 if my above description is not clear enough. Thanks.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Send TS SDK logs to stderr on socket disconnect [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #69302:
URL: https://github.com/apache/airflow/pull/69302#discussion_r3608252730
##########
ts-sdk/src/coordinator/log-channel.ts:
##########
@@ -47,31 +47,45 @@ export interface LogRecord {
const DEFAULT_LOGGER_NAME = "ts-sdk";
+interface LogChannelState {
+ sock: Socket;
+ connected: boolean;
+ closed: boolean;
+}
+
Review Comment:
Sorry for the late reply!
The purpose of the buffer is different. No matter the final task state is success or fail will need to handle the following scenario. (Both Java and Go SDK handle this.)
Here's the lifecycle of the Lang-SDK subprocess.
1. TS subprocess started (by the `popen` on coordinator side)
2. TS execution time logger can still logs somethings at this moment! (e.g. trying to connect to coordinator `--comms=127:<port> ...` ..
3. TS subprocess really connected to the coordinator.
4. All the logs from TS logger will forward to `--log` socket after connected.
The buffer I mentioned handle the stage 2 where the TS logger can still log before connected to the coordinator log socket. We should "buffer" the logs at stage 2 and pass though to the log socket once connected.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Send TS SDK logs to stderr on socket disconnect [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #69302:
URL: https://github.com/apache/airflow/pull/69302#discussion_r3608393065
##########
ts-sdk/src/coordinator/log-channel.ts:
##########
@@ -120,8 +140,13 @@ export class LogChannel {
async close(): Promise<void> {
if (!this.isRoot) return;
+ this.shared.closed = true;
+ if (!this.shared.connected) {
+ this.shared.sock.destroy();
+ return;
+ }
return new Promise((resolve) => {
- this.sock.end(() => resolve());
+ this.shared.sock.end(() => resolve());
Review Comment:
This may never resolve if the graceful `end()` never completes its flush (e.g. the socket errors). Maybe we can add a timeout here to ensure things always continue on?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Send TS SDK logs to stderr on socket disconnect [airflow]
Posted by "guan404ming (via GitHub)" <gi...@apache.org>.
guan404ming commented on PR #69302:
URL: https://github.com/apache/airflow/pull/69302#issuecomment-5011863005
@jason810496 would you want to take another look before merge? thanks!
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Send TS SDK logs to stderr on socket disconnect [airflow]
Posted by "guan404ming (via GitHub)" <gi...@apache.org>.
guan404ming commented on code in PR #69302:
URL: https://github.com/apache/airflow/pull/69302#discussion_r3608453676
##########
ts-sdk/src/coordinator/log-channel.ts:
##########
@@ -47,31 +47,46 @@ export interface LogRecord {
const DEFAULT_LOGGER_NAME = "ts-sdk";
+interface LogChannelState {
+ sock: Socket;
+ connected: boolean;
+ closed: boolean;
+}
+
export class LogChannel {
- private readonly sock: Socket;
+ private readonly shared: LogChannelState;
private readonly name: string;
private readonly isRoot: boolean;
- private constructor(sock: Socket, name: string, isRoot: boolean) {
- this.sock = sock;
+ private constructor(shared: LogChannelState, 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: LogChannelState = {
+ sock: await connectTcp(addr),
+ connected: true,
+ closed: false,
+ };
+ shared.sock.on("error", (err) => {
+ shared.connected = false;
+ process.stderr.write(`[${name}] log socket error: ${err.message}\n`);
+ });
+ shared.sock.on("close", () => {
+ if (shared.closed) return;
+ shared.connected = false;
+ process.stderr.write(`[${name}] log socket closed unexpectedly; further logs go to stderr\n`);
+ });
Review Comment:
Right, on EPIPE both fire and we'd write two messages. Just Added a `!shared.connected` guard to the close handler, so when the error handler has already reported (and flipped connected), the close handler stays quiet.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Send TS SDK logs to stderr on socket disconnect [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #69302:
URL: https://github.com/apache/airflow/pull/69302#discussion_r3608252730
##########
ts-sdk/src/coordinator/log-channel.ts:
##########
@@ -47,31 +47,45 @@ export interface LogRecord {
const DEFAULT_LOGGER_NAME = "ts-sdk";
+interface LogChannelState {
+ sock: Socket;
+ connected: boolean;
+ closed: boolean;
+}
+
Review Comment:
Sorry for the late reply:
The purpose of the buffer is different. Take the following sequence as example:
1. TS subprocess started (by the `popen` on coordinator side)
2. TS execution time logger can still logs somethings at this moment! (e.g. trying to connect to coordinator `--comms=127:<port> ...` ..
3. TS subprocess really connected to the coordinator.
4. All the logs from TS logger will forward to `--log` socket after connected.
The buffer I mentioned handle the stage 2 where the TS logger can still log before connected to the coordinator log socket. We should "buffer" the logs at stage 2 and pass though to the log socket once connected.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Send TS SDK logs to stderr on socket disconnect [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #69302:
URL: https://github.com/apache/airflow/pull/69302#discussion_r3608381650
##########
ts-sdk/src/coordinator/log-channel.ts:
##########
@@ -47,31 +47,46 @@ export interface LogRecord {
const DEFAULT_LOGGER_NAME = "ts-sdk";
+interface LogChannelState {
+ sock: Socket;
+ connected: boolean;
+ closed: boolean;
+}
+
export class LogChannel {
- private readonly sock: Socket;
+ private readonly shared: LogChannelState;
private readonly name: string;
private readonly isRoot: boolean;
- private constructor(sock: Socket, name: string, isRoot: boolean) {
- this.sock = sock;
+ private constructor(shared: LogChannelState, 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: LogChannelState = {
+ sock: await connectTcp(addr),
+ connected: true,
+ closed: false,
+ };
+ shared.sock.on("error", (err) => {
+ shared.connected = false;
+ process.stderr.write(`[${name}] log socket error: ${err.message}\n`);
+ });
+ shared.sock.on("close", () => {
+ if (shared.closed) return;
+ shared.connected = false;
+ process.stderr.write(`[${name}] log socket closed unexpectedly; further logs go to stderr\n`);
+ });
Review Comment:
These two tend to fire together in errors like EPIPE. Maybe add a `if (shared.connected)` check so only one of the error messages are written in this case?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Buffer TypeScript SDK logs across log socket disconnects [airflow]
Posted by "guan404ming (via GitHub)" <gi...@apache.org>.
guan404ming commented on code in PR #69302:
URL: https://github.com/apache/airflow/pull/69302#discussion_r3557666963
##########
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:
Fixed. Dropped the Go SDK reference and kept only the description of the buffering behavior itself.
##########
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:
Fixed, renamed to `LogChannelState`.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Send TypeScript SDK logs to stderr when the log socket disconnects [airflow]
Posted by "jason810496 (via GitHub)" <gi...@apache.org>.
jason810496 commented on code in PR #69302:
URL: https://github.com/apache/airflow/pull/69302#discussion_r3568286598
##########
ts-sdk/src/coordinator/log-channel.ts:
##########
@@ -47,31 +47,45 @@ export interface LogRecord {
const DEFAULT_LOGGER_NAME = "ts-sdk";
+interface LogChannelState {
+ sock: Socket;
+ connected: boolean;
+ closed: boolean;
+}
+
Review Comment:
I think we can keep the logging buffer part. Perhaps in the follow-up, we can keep the scope as-is for this PR to only focus on the redirecting the logs to stderr once receive any fatal error.
More specific, the follow-up will target the logging that TS runtime produces before the socket connects as buffer then send to the log socket once both side connect.
##########
ts-sdk/src/coordinator/log-channel.ts:
##########
@@ -47,31 +47,45 @@ export interface LogRecord {
const DEFAULT_LOGGER_NAME = "ts-sdk";
+interface LogChannelState {
+ sock: Socket;
+ connected: boolean;
+ closed: boolean;
+}
+
export class LogChannel {
- private readonly sock: Socket;
+ private readonly shared: LogChannelState;
private readonly name: string;
private readonly isRoot: boolean;
- private constructor(sock: Socket, name: string, isRoot: boolean) {
- this.sock = sock;
+ private constructor(shared: LogChannelState, 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: LogChannelState = {
+ sock: await connectTcp(addr),
+ connected: true,
+ closed: false,
+ };
+ shared.sock.on("error", (err) => {
+ process.stderr.write(`[${name}] log socket error: ${err.message}\n`);
+ });
Review Comment:
The `"error"` handler only writes to stderr; only the later `"close"` handler flips `shared.connected = false`. Node fires `"close"` a tick or so after `"error"` (empirically ~0.05ms later in a local repro), and in that gap a `send()` call still takes the "connected" branch and writes to an already-destroyed socket — the record is silently dropped instead of falling back to stderr as intended. Setting the flag in the `error` handler too closes the gap:
```suggestion
shared.sock.on("error", (err) => {
shared.connected = false;
process.stderr.write(`[${name}] log socket error: ${err.message}\n`);
});
```
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org