Enable streaming log (apache#54445)

Author: guan404ming

airflow-core/src/airflow/ui/src/pages/TaskInstance/Logs/Logs.test.tsx

@@ -98,6 +98,6 @@ describe("Task log grouping", () => {
 
     fireEvent.click(collapseItem);
 
-    expect(screen.getByText(/Marking task as SUCCESS/iu)).toBeVisible();
+    await waitFor(() => expect(screen.queryByText(/Marking task as SUCCESS/iu)).toBeVisible());
   }, 10_000);
 });

airflow-core/src/airflow/ui/src/queries/useLogs.tsx

@@ -26,6 +26,7 @@ import type { TaskInstanceResponse, TaskInstancesLogResponse } from "openapi/req
 import { renderStructuredLog } from "src/components/renderStructuredLog";
 import { isStatePending, useAutoRefresh } from "src/utils";
 import { getTaskInstanceLink } from "src/utils/links";
+import { parseStreamingLogContent } from "src/utils/logs";
 
 type Props = {
   accept?: "*/*" | "application/json" | "application/x-ndjson";
@@ -173,7 +174,7 @@ const parseLogs = ({
 
 export const useLogs = (
   {
-    accept = "application/json",
+    accept = "application/x-ndjson",
     dagId,
     expanded,
     logLevelFilters,
@@ -209,7 +210,7 @@ export const useLogs = (
   );
 
   const parsedData = parseLogs({
-    data: data?.content ?? [],
+    data: parseStreamingLogContent(data),
     expanded,
     logLevelFilters,
     showSource,

airflow-core/src/airflow/ui/src/utils/logs.ts

@@ -22,6 +22,8 @@
 /* eslint-disable perfectionist/sort-objects */
 import { createListCollection } from "@chakra-ui/react";
 
+import type { TaskInstancesLogResponse } from "openapi/requests/types.gen";
+
 export enum LogLevel {
   DEBUG = "debug",
   INFO = "info",
@@ -51,3 +53,22 @@ export const logLevelOptions = createListCollection<{
     { label: LogLevel.CRITICAL.toUpperCase(), value: LogLevel.CRITICAL },
   ],
 });
+
+export const parseStreamingLogContent = (
+  data: TaskInstancesLogResponse | undefined,
+): TaskInstancesLogResponse["content"] => {
+  if (!data?.content) {
+    const content = data as unknown as string;
+
+    try {
+      return content
+        .split("\n")
+        .filter((line) => line.trim() !== "")
+        .map((line) => JSON.parse(line) as string);
+    } catch {
+      return [];
+    }
+  }
+
+  return data.content;
+};