This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 pushed a commit to branch v3-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 8744696c21e857b84e5f3628650531c07c78cd31
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Fri Oct 30 21:27:25 2026 +0530
[v3-4-test] Honor Airflow logging levels in TypeScript SDK (#76005) (#76030)
* Honor Airflow logging levels in TypeScript SDK
* Harden TS SDK log level filtering
- Keep the helper method's level when args carry a `level` field, so a
non-string value can no longer crash `send()`.
- Filter levels outside LogLevel as INFO instead of letting them bypass
the threshold.
- Filter by the channel's own name, matching the `[name]` event prefix,
rather than a caller-supplied `logger` field.
- Report an invalid `logging_level` instead of silently using INFO.
- Drop FATAL so accepted names match structlog's NAME_TO_LEVEL.
- Extract namespace_levels parsing and resolve each channel's threshold
once at construction.
* Restore FATAL log level alias in TS SDK
Airflow's configure_logging adds a `fatal` alias to NAME_TO_LEVEL before
parsing logging_level and namespace_levels, so Python accepts FATAL.
Removing it made the TS SDK reject a setting the Python side honors.
* Address review: rename threshold to logLevel and trim comments
---------
(cherry picked from commit 5cc2413925ae71ecd5951195463d9df78081b55c)
Co-authored-by: Shivam Rastogi <[email protected]>
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
ts-sdk/src/coordinator/log-channel.ts | 94 ++++++++++++++-
ts-sdk/tests/coordinator/log-channel.test.ts | 173 +++++++++++++++++++++++++++
2 files changed, 261 insertions(+), 6 deletions(-)
diff --git a/ts-sdk/src/coordinator/log-channel.ts
b/ts-sdk/src/coordinator/log-channel.ts
index 9dd85482c01..cb411a25e02 100644
--- a/ts-sdk/src/coordinator/log-channel.ts
+++ b/ts-sdk/src/coordinator/log-channel.ts
@@ -49,28 +49,106 @@ const DEFAULT_LOGGER_NAME = "ts-sdk";
const CLOSE_FLUSH_TIMEOUT_MS = 3_000;
+// Level names match Airflow's Python parser: structlog's NAME_TO_LEVEL plus
the `fatal` alias
+// that configure_logging adds.
+const LOG_LEVELS = new Map([
+ ["NOTSET", 0],
+ ["DEBUG", 10],
+ ["INFO", 20],
+ ["WARN", 30],
+ ["WARNING", 30],
+ ["ERROR", 40],
+ ["EXCEPTION", 40],
+ ["CRITICAL", 50],
+ ["FATAL", 50],
+]);
+
+const DEFAULT_LOG_LEVEL = LOG_LEVELS.get("INFO")!;
+
+function parseLogLevel(value: string | undefined): number | undefined {
+ return LOG_LEVELS.get(value?.trim().toUpperCase() ?? "");
+}
+
+/** Parse `namespace_levels`: `<logger>=<level>` pairs separated by whitespace
and/or commas.
+ * Later duplicates win; malformed entries are returned so the caller can
report them. */
+function parseNamespaceLevels(value: string | undefined): {
+ levels: Map<string, number>;
+ invalid: string[];
+} {
+ const levels = new Map<string, number>();
+ const invalid: string[] = [];
+ for (const entry of (value ?? "").split(/[\s,]+/).filter(Boolean)) {
+ const separator = entry.indexOf("=");
+ const level = parseLogLevel(entry.slice(separator + 1));
+ if (separator <= 0 || level === undefined) {
+ invalid.push(entry);
+ } else {
+ levels.set(entry.slice(0, separator), level);
+ }
+ }
+ return { levels, invalid };
+}
+
+/** Threshold for `name`: the most specific configured dotted ancestor, else
the global level. */
+function resolveLogLevel(shared: LogChannelState, name: string): number {
+ let namespace = name;
+ while (namespace) {
+ const level = shared.namespaceLevels.get(namespace);
+ if (level !== undefined) return level;
+ const separator = namespace.lastIndexOf(".");
+ if (separator < 0) break;
+ namespace = namespace.slice(0, separator);
+ }
+ return shared.level;
+}
+
interface LogChannelState {
sock: Socket | null;
connected: boolean;
closed: boolean;
buffer: string[];
+ level: number;
+ namespaceLevels: ReadonlyMap<string, number>;
}
export class LogChannel {
private readonly shared: LogChannelState;
private readonly name: string;
private readonly isRoot: boolean;
+ private readonly logLevel: number;
private constructor(shared: LogChannelState, name: string, isRoot: boolean) {
this.shared = shared;
this.name = name;
this.isRoot = isRoot;
+ this.logLevel = resolveLogLevel(shared, name);
}
/** Create a root channel with no socket yet; records are buffered
* until {@link LogChannel#connect} flushes them. */
static createBuffered(name: string = DEFAULT_LOGGER_NAME): LogChannel {
- return new LogChannel({ sock: null, connected: false, closed: false,
buffer: [] }, name, true);
+ const rawLevel = process.env["AIRFLOW__LOGGING__LOGGING_LEVEL"];
+ const level = parseLogLevel(rawLevel);
+ const namespaceLevels =
parseNamespaceLevels(process.env["AIRFLOW__LOGGING__NAMESPACE_LEVELS"]);
+ const root = new LogChannel(
+ {
+ sock: null,
+ connected: false,
+ closed: false,
+ buffer: [],
+ level: level ?? DEFAULT_LOG_LEVEL,
+ namespaceLevels: namespaceLevels.levels,
+ },
+ name,
+ true,
+ );
+ if (level === undefined && rawLevel?.trim()) {
+ root.error("Ignoring invalid logging_level; using INFO", {
logging_level: rawLevel });
+ }
+ for (const entry of namespaceLevels.invalid) {
+ root.error("Ignoring invalid namespace_levels entry", { entry });
+ }
+ return root;
}
/** Connect the socket and flush buffered records, in order. Root only,
@@ -116,13 +194,17 @@ export class LogChannel {
}
send(
- record: Omit<LogRecord, "timestamp" | "logger"> & {
+ record: Pick<LogRecord, "event" | "level"> & {
timestamp?: string;
logger?: string;
+ [key: string]: unknown;
},
): void {
// Drop late records after the log socket has closed.
if (this.shared.closed) return;
+ const level =
+ typeof record.level === "string" ?
LOG_LEVELS.get(record.level.toUpperCase()) : undefined;
+ if ((level ?? DEFAULT_LOG_LEVEL) < this.logLevel) return;
// Prepend the logger name to the event message so it surfaces in
// the Airflow UI task log view, which renders the message text but
// hides the `logger` JSON field. The field is still emitted for
@@ -148,19 +230,19 @@ export class LogChannel {
}
debug(event: string, args: Record<string, unknown> = {}): void {
- this.send({ event, level: "debug", ...args });
+ this.send({ event, ...args, level: "debug" });
}
info(event: string, args: Record<string, unknown> = {}): void {
- this.send({ event, level: "info", ...args });
+ this.send({ event, ...args, level: "info" });
}
warning(event: string, args: Record<string, unknown> = {}): void {
- this.send({ event, level: "warning", ...args });
+ this.send({ event, ...args, level: "warning" });
}
error(event: string, args: Record<string, unknown> = {}): void {
- this.send({ event, level: "error", ...args });
+ this.send({ event, ...args, level: "error" });
}
async close(): Promise<void> {
diff --git a/ts-sdk/tests/coordinator/log-channel.test.ts
b/ts-sdk/tests/coordinator/log-channel.test.ts
index f454b718cfc..6b7b6e3a251 100644
--- a/ts-sdk/tests/coordinator/log-channel.test.ts
+++ b/ts-sdk/tests/coordinator/log-channel.test.ts
@@ -62,11 +62,181 @@ describe("LogChannel", () => {
let fx: Fixture;
beforeEach(async () => {
+ vi.stubEnv("AIRFLOW__LOGGING__LOGGING_LEVEL", undefined);
+ vi.stubEnv("AIRFLOW__LOGGING__NAMESPACE_LEVELS", "");
fx = await makeServer();
});
afterEach(async () => {
fx.server.close();
+ vi.unstubAllEnvs();
+ });
+
+ it.each([
+ [undefined, ["info", "warning", "error"]],
+ ["", ["info", "warning", "error"]],
+ // Invalid values fall back to INFO and report themselves as the first
record.
+ ["VERBOSE", ["error", "info", "warning", "error"]],
+ ["INFO", ["info", "warning", "error"]],
+ [" debug ", ["debug", "info", "warning", "error"]],
+ ["WARNING", ["warning", "error"]],
+ ["WARN", ["warning", "error"]],
+ ["ERROR", ["error"]],
+ ["EXCEPTION", ["error"]],
+ ["CRITICAL", []],
+ ["FATAL", []],
+ ["NOTSET", ["debug", "info", "warning", "error"]],
+ ] as const)("filters socket records with logging_level=%s", async (level,
expected) => {
+ vi.stubEnv("AIRFLOW__LOGGING__LOGGING_LEVEL", level);
+ const root = await connectChannel(`127.0.0.1:${fx.port}`);
+ root.debug("debug");
+ root.info("info");
+ root.warning("warning");
+ root.error("error");
+ await root.close();
+ await fx.sockClosed;
+ expect(readRecords(fx.received).map((record) =>
record["level"])).toEqual(expected);
+ });
+
+ it("uses the nearest dotted namespace override and the last duplicate
entry", async () => {
+ vi.stubEnv("AIRFLOW__LOGGING__LOGGING_LEVEL", "ERROR");
+ vi.stubEnv(
+ "AIRFLOW__LOGGING__NAMESPACE_LEVELS",
+ " ts-sdk=WARNING, ts-sdk.comm=ERROR
ts-sdk.comm=INFO,ts-sdk.comm.deep=DEBUG ",
+ );
+ const root = await connectChannel(`127.0.0.1:${fx.port}`);
+ root.info("filtered at root");
+ root.warning("root warning");
+ root.child("comm").debug("filtered at comm");
+ root.child("comm").info("comm info");
+ root.child("comm").child("deep").child("child").debug("deep debug");
+ root.child("commander").info("partial namespace must not match");
+ root.child("commander").warning("inherited root warning");
+ await root.close();
+ await fx.sockClosed;
+ expect(readRecords(fx.received).map((record) => record["event"])).toEqual([
+ "[ts-sdk] root warning",
+ "[ts-sdk.comm] comm info",
+ "[ts-sdk.comm.deep.child] deep debug",
+ "[ts-sdk.commander] inherited root warning",
+ ]);
+ });
+
+ it("keeps valid namespace overrides and reports invalid entries once", async
() => {
+ vi.stubEnv(
+ "AIRFLOW__LOGGING__NAMESPACE_LEVELS",
+ "broken,=DEBUG,ts-sdk.comm=VERBOSE,ts-sdk.client=debug",
+ );
+ const root = await connectChannel(`127.0.0.1:${fx.port}`);
+ root.child("comm").debug("invalid override falls back to INFO");
+ root.child("client").debug("valid override");
+ root.child("client").debug("second record");
+ await root.close();
+ await fx.sockClosed;
+ const records = readRecords(fx.received);
+ expect(records.filter((record) => record["level"] === "error").map((r) =>
r["entry"])).toEqual([
+ "broken",
+ "=DEBUG",
+ "ts-sdk.comm=VERBOSE",
+ ]);
+ expect(records.filter((record) => record["level"] === "debug").map((r) =>
r["event"])).toEqual([
+ "[ts-sdk.client] valid override",
+ "[ts-sdk.client] second record",
+ ]);
+ });
+
+ it("filters before buffering and shares the initial configuration with later
children", async () => {
+ const root = LogChannel.createBuffered();
+ root.debug("filtered before connect");
+ root.info("buffered info");
+ vi.stubEnv("AIRFLOW__LOGGING__LOGGING_LEVEL", "DEBUG");
+ vi.stubEnv("AIRFLOW__LOGGING__NAMESPACE_LEVELS", "ts-sdk.comm=DEBUG");
+ root.child("comm").debug("filtered despite later environment change");
+ await root.connect(`127.0.0.1:${fx.port}`);
+ root.child("comm").info("connected info");
+ await root.close();
+ await fx.sockClosed;
+ expect(readRecords(fx.received).map((record) => record["event"])).toEqual([
+ "[ts-sdk] buffered info",
+ "[ts-sdk.comm] connected info",
+ ]);
+ });
+
+ it("filters records on stderr fallback as well as the socket", async () => {
+ const write = vi.spyOn(process.stderr, "write").mockImplementation(() =>
true);
+ const root = LogChannel.createBuffered();
+ try {
+ root.debug("filtered before connect");
+ root.info("buffered info");
+ await root.close();
+ expect(write.mock.calls).toHaveLength(1);
+ expect(JSON.parse(String(write.mock.calls[0]?.[0]))).toMatchObject({
+ event: "[ts-sdk] buffered info",
+ level: "info",
+ });
+ } finally {
+ write.mockRestore();
+ }
+ });
+
+ it("reports an invalid logging_level once", async () => {
+ vi.stubEnv("AIRFLOW__LOGGING__LOGGING_LEVEL", "TRACE");
+ const root = await connectChannel(`127.0.0.1:${fx.port}`);
+ root.debug("filtered at INFO");
+ await root.close();
+ await fx.sockClosed;
+ expect(readRecords(fx.received)).toMatchObject([
+ {
+ level: "error",
+ event: "[ts-sdk] Ignoring invalid logging_level; using INFO",
+ logging_level: "TRACE",
+ },
+ ]);
+ });
+
+ it("keeps the method's level when args carry a level field", async () => {
+ vi.stubEnv("AIRFLOW__LOGGING__LOGGING_LEVEL", "WARNING");
+ const root = await connectChannel(`127.0.0.1:${fx.port}`);
+ root.info("numeric level", { level: 3 });
+ root.debug("level escalation", { level: "error" });
+ root.error("level demotion", { level: "debug" });
+ await root.close();
+ await fx.sockClosed;
+ expect(readRecords(fx.received)).toMatchObject([
+ { event: "[ts-sdk] level demotion", level: "error" },
+ ]);
+ });
+
+ it("treats a level outside LogLevel as INFO instead of bypassing the
filter", async () => {
+ vi.stubEnv("AIRFLOW__LOGGING__LOGGING_LEVEL", "WARNING");
+ const root = await connectChannel(`127.0.0.1:${fx.port}`);
+ root.send({ event: "unknown", level: "trace" as never });
+ root.send({ event: "non-string", level: 7 as never });
+ root.warning("kept");
+ await root.close();
+ await fx.sockClosed;
+ expect(readRecords(fx.received).map((record) =>
record["event"])).toEqual(["[ts-sdk] kept"]);
+ });
+
+ it("filters by the channel name even when the record overrides logger",
async () => {
+ vi.stubEnv("AIRFLOW__LOGGING__NAMESPACE_LEVELS", "ts-sdk.client=DEBUG");
+ const root = await connectChannel(`127.0.0.1:${fx.port}`);
+ root.child("comm").debug("override does not unlock DEBUG", { logger:
"ts-sdk.client" });
+ root.child("comm").info("non-string logger", { logger: 123 });
+ await root.close();
+ await fx.sockClosed;
+ expect(readRecords(fx.received)).toMatchObject([
+ { event: "[ts-sdk.comm] non-string logger", logger: 123 },
+ ]);
+ });
+
+ it("applies namespace overrides to the invalid-entry report itself", async
() => {
+ vi.stubEnv("AIRFLOW__LOGGING__NAMESPACE_LEVELS", "broken ts-sdk=CRITICAL");
+ const root = await connectChannel(`127.0.0.1:${fx.port}`);
+ root.error("filtered at CRITICAL");
+ await root.close();
+ await fx.sockClosed;
+ expect(readRecords(fx.received)).toEqual([]);
});
it("defaults logger name to 'ts-sdk' and auto-stamps timestamp", async () =>
{
@@ -104,6 +274,7 @@ describe("LogChannel", () => {
});
it("child() creates a hierarchical sibling sharing the socket", async () => {
+ vi.stubEnv("AIRFLOW__LOGGING__LOGGING_LEVEL", "DEBUG");
const root = await connectChannel(`127.0.0.1:${fx.port}`);
const comm = root.child("comm");
const client = root.child("client");
@@ -156,6 +327,7 @@ describe("LogChannel", () => {
});
it("buffers records emitted before connect and flushes them in order on
connect", async () => {
+ vi.stubEnv("AIRFLOW__LOGGING__LOGGING_LEVEL", "DEBUG");
const root = LogChannel.createBuffered();
const child = root.child("runtime");
child.debug("connecting", { attempt: 1 });
@@ -284,6 +456,7 @@ describe("LogChannel", () => {
});
it("warns once and falls back to stderr after an unexpected disconnect",
async () => {
+ vi.stubEnv("AIRFLOW__LOGGING__LOGGING_LEVEL", "DEBUG");
const serverSocks: net.Socket[] = [];
const received: Buffer[] = [];
const server = net.createServer((sock) => {
