Skip to content

[v2-11-test] Resolve OOM When Reading Large Logs in Webserver#45914

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
jason810496 wants to merge 11 commits into from
Closed

[v2-11-test] Resolve OOM When Reading Large Logs in Webserver #45914

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
7fe9b77
Bump trove-classifiers from 2025.4.11.15 to 2025.4.28.22 (#49973)
dependabot[bot] Apr 29, 2025
ab176bf
[v2-10-test] Resolve OOM When Reading Large Logs in Webserver
jason810496 Jan 22, 2025
3226233
Fix test decorator
jason810496 Jan 22, 2025
680a713
Fix test
jason810496 Jan 22, 2025
48debfe
Fix test_log_handler, test_log_reader
jason810496 Jan 23, 2025
54386ec
Fix scheduler_job_runner
jason810496 Jan 23, 2025
c6d4e5e
Fix log_handler
jason810496 Jan 23, 2025
7e39433
Fix test_log_endpoint test
jason810496 Jan 30, 2025
5797341
fixup! Add type hint
jason810496 Feb 19, 2025
e4e5a85
fixup! remove unnecessary type hint
jason810496 Feb 20, 2025
6cef899
Fix dedupe for logs with same timestamp
jason810496 Apr 30, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 6 additions & 3 deletions airflow/api_connexion/endpoints/log_endpoint.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -115,11 +115,14 @@ def get_log(
# return_type would be either the above two or None
logs: Any
if return_type == "application/json" or return_type is None: # default
logs, metadata = task_log_reader.read_log_chunks(ti, task_try_number, metadata)
logs = logs[0] if task_try_number is not None else logs
hosts, log_streams, metadata = task_log_reader.read_log_chunks(ti, task_try_number, metadata)
host = f"{hosts[0] or ''}\n"
Comment thread
jason810496 marked this conversation as resolved.
logs = log_streams[0] if task_try_number is not None else log_streams
# we must have token here, so we can safely ignore it
token = URLSafeSerializer(key).dumps(metadata) # type: ignore[assignment]
return logs_schema.dump(LogResponseObject(continuation_token=token, content=logs))
return logs_schema.dump(
LogResponseObject(continuation_token=token, content=host + "\n".join(log for log in logs))
)
# text/plain. Stream
logs = task_log_reader.read_log_stream(ti, task_try_number, metadata)

Expand Down
3 changes: 2 additions & 1 deletion airflow/executors/base_executor.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -52,6 +52,7 @@
from airflow.executors.executor_utils import ExecutorName
from airflow.models.taskinstance import TaskInstance
from airflow.models.taskinstancekey import TaskInstanceKey
from airflow.utils.log.file_task_handler import _CompatibleLogSourceType

# Command to execute - list of strings
# the first element is always "airflow".
Expand DownExpand Up@@ -528,7 +529,7 @@ def execute_async(
"""
raise NotImplementedError()

def get_task_log(self, ti: TaskInstance, try_number: int) -> tuple[list[str], list[str]]:
def get_task_log(self, ti: TaskInstance, try_number: int) -> _CompatibleLogSourceType:
"""
Return the task logs.

Expand Down
15 changes: 8 additions & 7 deletions airflow/jobs/scheduler_job_runner.py
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -872,16 +872,17 @@ def _process_executor_events(self, executor: BaseExecutor, session: Session) ->

task_log_reader = TaskLogReader()
if task_log_reader.supports_read:
metadata: dict[str, Any] = {}
logs, metadata = task_log_reader.read_log_chunks(ti, ti.try_number, metadata)
if ti.hostname in dict(logs[0]):
message = str(dict(logs[0])[ti.hostname]).replace("\\n", "\n")
hosts, log_streams, metadata = task_log_reader.read_log_chunks(
ti, ti.try_number, None
)
if ti.hostname in hosts:
message = str(next(iter(log_streams[0])))
while metadata["end_of_log"] is False:
logs, metadata = task_log_reader.read_log_chunks(
hosts, log_streams, metadata = task_log_reader.read_log_chunks(

@kaxilkaxilApr 30, 2025

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Since this is a big change, I am afraid to include it in 2.11!

For example, in this case where the return type of task_log_reader.read_log_chunks has changed.

@dstandishdstandishApr 30, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@kaxil it seems he is intending this to go to 3.0 only
#45914 (comment)

@dstandishdstandishApr 30, 2025

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yeah we should not spend time adding meaningful changes to 2.x

@jason810496jason810496May 1, 2025

Copy link
Copy Markdown
MemberAuthor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

For example, in this case where the return type of task_log_reader.read_log_chunks has changed.

I think changing the return type of read_log_chunks or other internal methods isn't a problem, since the user-facing interface is the REST API. As long as the REST API response remains consistent, users won’t be affected or need to be aware of the refactor.

ti, ti.try_number - 1, metadata
)
if ti.hostname in dict(logs[0]):
message = message + str(dict(logs[0])[ti.hostname]).replace("\\n", "\n")
if ti.hostname in hosts:
message = message + str(next(iter(log_streams[0])))
if span.is_recording():
span.add_event(
name="task_log",
Expand Down
Loading
, 'i'); if (__m === '*' || __re.test(location.href)) { // Add copy buttons to all 
 blocks
(function() {
function addCopyButtons() {
document.querySelectorAll('pre code').forEach(function(codeBlock) {
if (codeBlock.parentElement.hasAttribute('data-copy-added')) return;
codeBlock.parentElement.setAttribute('data-copy-added', 'true');
var btn = document.createElement('button');
btn.textContent = 'Copy';
btn.style.cssText = 'position:absolute;top:4px;right:4px;padding:2px 8px;font-size:11px;background:#4ecdc4;border:none;border-radius:4px;color:#1a1a2e;cursor:pointer;opacity:0.7;transition:opacity 0.2s;';
btn.onmouseover = function() { this.style.opacity = '1'; };
btn.onmouseout = function() { this.style.opacity = '0.7'; };
btn.onclick = function() {
navigator.clipboard.writeText(codeBlock.textContent).then(function() {
btn.textContent = 'Copied!';
setTimeout(function() { btn.textContent = 'Copy'; }, 1500);
});
};
codeBlock.parentElement.style.position = 'relative';
codeBlock.parentElement.appendChild(btn);
});
}
addCopyButtons();
// Re-run on dynamic content
var observer = new MutationObserver(addCopyButtons);
observer.observe(document.body, { childList: true, subtree: true });
})();
}
} catch(__e) { console.warn('[Userscript:Add Copy Buttons to Code Blocks]', __e); }
})();
(function(){
try {
var __m = "github.com";
var __re = new RegExp('^' + "github\\.com" + '
[v2-11-test] Resolve OOM When Reading Large Logs in Webserver by jason810496 · Pull Request #45914 · apache/airflow · GitHub
Skip to content
[v2-11-test] Resolve OOM When Reading Large Logs in Webserver#45914

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Closed

jason810496
wants to merge
11
commits into

from

 Closed


[v2-11-test] Resolve OOM When Reading Large Logs in Webserver
#45914
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


11 commits


Select commit
Hold shift + click to select a range
7fe9b77

Bump trove-classifiers from 2025.4.11.15 to 2025.4.28.22 (#49973)


dependabot[bot] Apr 29, 2025
ab176bf

[v2-10-test] Resolve OOM When Reading Large Logs in Webserver


jason810496 Jan 22, 2025
3226233

Fix test decorator


jason810496 Jan 22, 2025
680a713

Fix test


jason810496 Jan 22, 2025
48debfe

Fix test_log_handler, test_log_reader


jason810496 Jan 23, 2025
54386ec

Fix scheduler_job_runner


jason810496 Jan 23, 2025
c6d4e5e

Fix log_handler


jason810496 Jan 23, 2025
7e39433

Fix test_log_endpoint test


jason810496 Jan 30, 2025
5797341

fixup! Add type hint


jason810496 Feb 19, 2025
e4e5a85

fixup! remove unnecessary type hint


jason810496 Feb 20, 2025
6cef899

Fix dedupe for logs with same timestamp


jason810496 Apr 30, 2025

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


9 changes: 6 additions & 3 deletions
airflow/api_connexion/endpoints/log_endpoint.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -115,11 +115,14 @@ def get_log(
 # return_type would be either the above two or None
 logs: Any
 if return_type == "application/json" or return_type is None: # default
 logs, metadata = task_log_reader.read_log_chunks(ti, task_try_number, metadata)
 logs = logs[0] if task_try_number is not None else logs
 hosts, log_streams, metadata = task_log_reader.read_log_chunks(ti, task_try_number, metadata)
 host = f"{hosts[0] or ''}\n"
Comment thread
jason810496 marked this conversation as resolved.

 logs = log_streams[0] if task_try_number is not None else log_streams
 # we must have token here, so we can safely ignore it
 token = URLSafeSerializer(key).dumps(metadata) # type: ignore[assignment]
 return logs_schema.dump(LogResponseObject(continuation_token=token, content=logs))
 return logs_schema.dump(
 LogResponseObject(continuation_token=token, content=host + "\n".join(log for log in logs))
 )
 # text/plain. Stream
 logs = task_log_reader.read_log_stream(ti, task_try_number, metadata)

Expand Down

3 changes: 2 additions & 1 deletion
airflow/executors/base_executor.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -52,6 +52,7 @@
 from airflow.executors.executor_utils import ExecutorName
 from airflow.models.taskinstance import TaskInstance
 from airflow.models.taskinstancekey import TaskInstanceKey
 from airflow.utils.log.file_task_handler import _CompatibleLogSourceType

 # Command to execute - list of strings
 # the first element is always "airflow".
Expand DownExpand Up@@ -528,7 +529,7 @@ def execute_async(
 """
 raise NotImplementedError()

 def get_task_log(self, ti: TaskInstance, try_number: int) -> tuple[list[str], list[str]]:
 def get_task_log(self, ti: TaskInstance, try_number: int) -> _CompatibleLogSourceType:
 """
 Return the task logs.

Expand Down

15 changes: 8 additions & 7 deletions
airflow/jobs/scheduler_job_runner.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -872,16 +872,17 @@ def _process_executor_events(self, executor: BaseExecutor, session: Session) ->

 task_log_reader = TaskLogReader()
 if task_log_reader.supports_read:
 metadata: dict[str, Any] = {}
 logs, metadata = task_log_reader.read_log_chunks(ti, ti.try_number, metadata)
 if ti.hostname in dict(logs[0]):
 message = str(dict(logs[0])[ti.hostname]).replace("\\n", "\n")
 hosts, log_streams, metadata = task_log_reader.read_log_chunks(
 ti, ti.try_number, None
 )
 if ti.hostname in hosts:
 message = str(next(iter(log_streams[0])))
 while metadata["end_of_log"] is False:
logs, metadata = task_log_reader.read_log_chunks(
hosts, log_streams, metadata = task_log_reader.read_log_chunks(
@kaxilkaxilApr 30, 2025

Copy link


Copy Markdown

Member
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Since this is a big change, I am afraid to include it in 2.11!
For example, in this case where the return type of task_log_reader.read_log_chunks has changed.
@dstandishdstandishApr 30, 2025

Copy link


Copy Markdown

Contributor
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@kaxil it seems he is intending this to go to 3.0 only
#45914 (comment)
@dstandishdstandishApr 30, 2025

Copy link


Copy Markdown

Contributor
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yeah we should not spend time adding meaningful changes to 2.x
@jason810496jason810496May 1, 2025

Copy link


Copy Markdown

MemberAuthor
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

For example, in this case where the return type of task_log_reader.read_log_chunks has changed.
I think changing the return type of read_log_chunks or other internal methods isn't a problem, since the user-facing interface is the REST API. As long as the REST API response remains consistent, users won’t be affected or need to be aware of the refactor.
 ti, ti.try_number - 1, metadata
 )
 if ti.hostname in dict(logs[0]):
 message = message + str(dict(logs[0])[ti.hostname]).replace("\\n", "\n")
 if ti.hostname in hosts:
 message = message + str(next(iter(log_streams[0])))
 if span.is_recording():
 span.add_event(
 name="task_log",
Expand Down
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Force GitHub README to respect dark mode
(function() {
var style = document.createElement('style');
style.textContent = '
.markdown-body {
color-scheme: dark light;
}
.markdown-body pre { background: #161b22 !important; }
.markdown-body code { background: rgba(110, 118, 129, 0.4) !important; }
.markdown-body table th, .markdown-body table td { border-color: #30363d !important; }
.markdown-body img { background: #0d1117; }
.markdown-body blockquote { border-left-color: #8b949e; }
.markdown-body hr { border-color: #30363d; }
';
document.head.appendChild(style);
})();
}
} catch(__e) { console.warn('[Userscript:GitHub Dark Mode README Fix]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '
[v2-11-test] Resolve OOM When Reading Large Logs in Webserver by jason810496 · Pull Request #45914 · apache/airflow · GitHub
Skip to content
[v2-11-test] Resolve OOM When Reading Large Logs in Webserver#45914

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Closed

jason810496
wants to merge
11
commits into

from

 Closed


[v2-11-test] Resolve OOM When Reading Large Logs in Webserver
#45914
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


11 commits


Select commit
Hold shift + click to select a range
7fe9b77

Bump trove-classifiers from 2025.4.11.15 to 2025.4.28.22 (#49973)


dependabot[bot] Apr 29, 2025
ab176bf

[v2-10-test] Resolve OOM When Reading Large Logs in Webserver


jason810496 Jan 22, 2025
3226233

Fix test decorator


jason810496 Jan 22, 2025
680a713

Fix test


jason810496 Jan 22, 2025
48debfe

Fix test_log_handler, test_log_reader


jason810496 Jan 23, 2025
54386ec

Fix scheduler_job_runner


jason810496 Jan 23, 2025
c6d4e5e

Fix log_handler


jason810496 Jan 23, 2025
7e39433

Fix test_log_endpoint test


jason810496 Jan 30, 2025
5797341

fixup! Add type hint


jason810496 Feb 19, 2025
e4e5a85

fixup! remove unnecessary type hint


jason810496 Feb 20, 2025
6cef899

Fix dedupe for logs with same timestamp


jason810496 Apr 30, 2025

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


9 changes: 6 additions & 3 deletions
airflow/api_connexion/endpoints/log_endpoint.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -115,11 +115,14 @@ def get_log(
 # return_type would be either the above two or None
 logs: Any
 if return_type == "application/json" or return_type is None: # default
 logs, metadata = task_log_reader.read_log_chunks(ti, task_try_number, metadata)
 logs = logs[0] if task_try_number is not None else logs
 hosts, log_streams, metadata = task_log_reader.read_log_chunks(ti, task_try_number, metadata)
 host = f"{hosts[0] or ''}\n"
Comment thread
jason810496 marked this conversation as resolved.

 logs = log_streams[0] if task_try_number is not None else log_streams
 # we must have token here, so we can safely ignore it
 token = URLSafeSerializer(key).dumps(metadata) # type: ignore[assignment]
 return logs_schema.dump(LogResponseObject(continuation_token=token, content=logs))
 return logs_schema.dump(
 LogResponseObject(continuation_token=token, content=host + "\n".join(log for log in logs))
 )
 # text/plain. Stream
 logs = task_log_reader.read_log_stream(ti, task_try_number, metadata)

Expand Down

3 changes: 2 additions & 1 deletion
airflow/executors/base_executor.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -52,6 +52,7 @@
 from airflow.executors.executor_utils import ExecutorName
 from airflow.models.taskinstance import TaskInstance
 from airflow.models.taskinstancekey import TaskInstanceKey
 from airflow.utils.log.file_task_handler import _CompatibleLogSourceType

 # Command to execute - list of strings
 # the first element is always "airflow".
Expand DownExpand Up@@ -528,7 +529,7 @@ def execute_async(
 """
 raise NotImplementedError()

 def get_task_log(self, ti: TaskInstance, try_number: int) -> tuple[list[str], list[str]]:
 def get_task_log(self, ti: TaskInstance, try_number: int) -> _CompatibleLogSourceType:
 """
 Return the task logs.

Expand Down

15 changes: 8 additions & 7 deletions
airflow/jobs/scheduler_job_runner.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -872,16 +872,17 @@ def _process_executor_events(self, executor: BaseExecutor, session: Session) ->

 task_log_reader = TaskLogReader()
 if task_log_reader.supports_read:
 metadata: dict[str, Any] = {}
 logs, metadata = task_log_reader.read_log_chunks(ti, ti.try_number, metadata)
 if ti.hostname in dict(logs[0]):
 message = str(dict(logs[0])[ti.hostname]).replace("\\n", "\n")
 hosts, log_streams, metadata = task_log_reader.read_log_chunks(
 ti, ti.try_number, None
 )
 if ti.hostname in hosts:
 message = str(next(iter(log_streams[0])))
 while metadata["end_of_log"] is False:
logs, metadata = task_log_reader.read_log_chunks(
hosts, log_streams, metadata = task_log_reader.read_log_chunks(
@kaxilkaxilApr 30, 2025

Copy link


Copy Markdown

Member
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Since this is a big change, I am afraid to include it in 2.11!
For example, in this case where the return type of task_log_reader.read_log_chunks has changed.
@dstandishdstandishApr 30, 2025

Copy link


Copy Markdown

Contributor
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@kaxil it seems he is intending this to go to 3.0 only
#45914 (comment)
@dstandishdstandishApr 30, 2025

Copy link


Copy Markdown

Contributor
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yeah we should not spend time adding meaningful changes to 2.x
@jason810496jason810496May 1, 2025

Copy link


Copy Markdown

MemberAuthor
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

For example, in this case where the return type of task_log_reader.read_log_chunks has changed.
I think changing the return type of read_log_chunks or other internal methods isn't a problem, since the user-facing interface is the REST API. As long as the REST API response remains consistent, users won’t be affected or need to be aware of the refactor.
 ti, ti.try_number - 1, metadata
 )
 if ti.hostname in dict(logs[0]):
 message = message + str(dict(logs[0])[ti.hostname]).replace("\\n", "\n")
 if ti.hostname in hosts:
 message = message + str(next(iter(log_streams[0])))
 if span.is_recording():
 span.add_event(
 name="task_log",
Expand Down
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Highlight search terms from Google/DuckDuckGo/Bing referrer
(function() {
var ref = document.referrer;
var terms = [];
if (ref.includes('google.com') || ref.includes('duckduckgo.com') || ref.includes('bing.com')) {
var url = new URL(ref);
var q = url.searchParams.get('q') || url.searchParams.get('p');
if (q) {
terms = q.split(/\s+/).filter(function(t) { return t.length > 2; });
}
}
if (terms.length === 0) return;
var style = document.createElement('style');
style.textContent = '.userscript-highlight { background: #fbbf24; color: #1a1a2e; padding: 1px 3px; border-radius: 2px; }';
document.head.appendChild(style);
function highlight(node) {
if (node.nodeType === 3) { // text node
var text = node.textContent;
var found = false;
terms.forEach(function(term) {
var regex = new RegExp('(' + term.replace(/[.*+?^${}()|[\]\\]/g, '\\') + ')', 'gi');
if (regex.test(text)) {
found = true;
var frag = document.createDocumentFragment();
var parts = text.split(regex);
parts.forEach(function(part, i) {
if (i % 2 === 0) {
frag.appendChild(document.createTextNode(part));
} else {
var span = document.createElement('span');
span.className = 'userscript-highlight';
span.textContent = part;
frag.appendChild(span);
}
});
node.parentNode.replaceChild(frag, node);
}
});
} else if (node.nodeType === 1 && node.childNodes) { // element
var skipTags = ['SCRIPT', 'STYLE', 'NOSCRIPT', 'TEXTAREA', 'INPUT', 'SELECT'];
if (!skipTags.includes(node.tagName)) {
Array.from(node.childNodes).forEach(highlight);
}
}
}
highlight(document.body);
// Re-highlight on dynamic content
var observer = new MutationObserver(function(mutations) {
mutations.forEach(function(m) {
m.addedNodes.forEach(function(node) {
if (node.nodeType === 1 || node.nodeType === 3) highlight(node);
});
});
});
observer.observe(document.body, { childList: true, subtree: true });
})();
}
} catch(__e) { console.warn('[Userscript:Highlight Search Terms]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '
[v2-11-test] Resolve OOM When Reading Large Logs in Webserver by jason810496 · Pull Request #45914 · apache/airflow · GitHub
Skip to content
[v2-11-test] Resolve OOM When Reading Large Logs in Webserver#45914

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Closed

jason810496
wants to merge
11
commits into

from

 Closed


[v2-11-test] Resolve OOM When Reading Large Logs in Webserver
#45914
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


11 commits


Select commit
Hold shift + click to select a range
7fe9b77

Bump trove-classifiers from 2025.4.11.15 to 2025.4.28.22 (#49973)


dependabot[bot] Apr 29, 2025
ab176bf

[v2-10-test] Resolve OOM When Reading Large Logs in Webserver


jason810496 Jan 22, 2025
3226233

Fix test decorator


jason810496 Jan 22, 2025
680a713

Fix test


jason810496 Jan 22, 2025
48debfe

Fix test_log_handler, test_log_reader


jason810496 Jan 23, 2025
54386ec

Fix scheduler_job_runner


jason810496 Jan 23, 2025
c6d4e5e

Fix log_handler


jason810496 Jan 23, 2025
7e39433

Fix test_log_endpoint test


jason810496 Jan 30, 2025
5797341

fixup! Add type hint


jason810496 Feb 19, 2025
e4e5a85

fixup! remove unnecessary type hint


jason810496 Feb 20, 2025
6cef899

Fix dedupe for logs with same timestamp


jason810496 Apr 30, 2025

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


9 changes: 6 additions & 3 deletions
airflow/api_connexion/endpoints/log_endpoint.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -115,11 +115,14 @@ def get_log(
 # return_type would be either the above two or None
 logs: Any
 if return_type == "application/json" or return_type is None: # default
 logs, metadata = task_log_reader.read_log_chunks(ti, task_try_number, metadata)
 logs = logs[0] if task_try_number is not None else logs
 hosts, log_streams, metadata = task_log_reader.read_log_chunks(ti, task_try_number, metadata)
 host = f"{hosts[0] or ''}\n"
Comment thread
jason810496 marked this conversation as resolved.

 logs = log_streams[0] if task_try_number is not None else log_streams
 # we must have token here, so we can safely ignore it
 token = URLSafeSerializer(key).dumps(metadata) # type: ignore[assignment]
 return logs_schema.dump(LogResponseObject(continuation_token=token, content=logs))
 return logs_schema.dump(
 LogResponseObject(continuation_token=token, content=host + "\n".join(log for log in logs))
 )
 # text/plain. Stream
 logs = task_log_reader.read_log_stream(ti, task_try_number, metadata)

Expand Down

3 changes: 2 additions & 1 deletion
airflow/executors/base_executor.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -52,6 +52,7 @@
 from airflow.executors.executor_utils import ExecutorName
 from airflow.models.taskinstance import TaskInstance
 from airflow.models.taskinstancekey import TaskInstanceKey
 from airflow.utils.log.file_task_handler import _CompatibleLogSourceType

 # Command to execute - list of strings
 # the first element is always "airflow".
Expand DownExpand Up@@ -528,7 +529,7 @@ def execute_async(
 """
 raise NotImplementedError()

 def get_task_log(self, ti: TaskInstance, try_number: int) -> tuple[list[str], list[str]]:
 def get_task_log(self, ti: TaskInstance, try_number: int) -> _CompatibleLogSourceType:
 """
 Return the task logs.

Expand Down

15 changes: 8 additions & 7 deletions
airflow/jobs/scheduler_job_runner.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -872,16 +872,17 @@ def _process_executor_events(self, executor: BaseExecutor, session: Session) ->

 task_log_reader = TaskLogReader()
 if task_log_reader.supports_read:
 metadata: dict[str, Any] = {}
 logs, metadata = task_log_reader.read_log_chunks(ti, ti.try_number, metadata)
 if ti.hostname in dict(logs[0]):
 message = str(dict(logs[0])[ti.hostname]).replace("\\n", "\n")
 hosts, log_streams, metadata = task_log_reader.read_log_chunks(
 ti, ti.try_number, None
 )
 if ti.hostname in hosts:
 message = str(next(iter(log_streams[0])))
 while metadata["end_of_log"] is False:
logs, metadata = task_log_reader.read_log_chunks(
hosts, log_streams, metadata = task_log_reader.read_log_chunks(
@kaxilkaxilApr 30, 2025

Copy link


Copy Markdown

Member
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Since this is a big change, I am afraid to include it in 2.11!
For example, in this case where the return type of task_log_reader.read_log_chunks has changed.
@dstandishdstandishApr 30, 2025

Copy link


Copy Markdown

Contributor
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@kaxil it seems he is intending this to go to 3.0 only
#45914 (comment)
@dstandishdstandishApr 30, 2025

Copy link


Copy Markdown

Contributor
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yeah we should not spend time adding meaningful changes to 2.x
@jason810496jason810496May 1, 2025

Copy link


Copy Markdown

MemberAuthor
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

For example, in this case where the return type of task_log_reader.read_log_chunks has changed.
I think changing the return type of read_log_chunks or other internal methods isn't a problem, since the user-facing interface is the REST API. As long as the REST API response remains consistent, users won’t be affected or need to be aware of the refactor.
 ti, ti.try_number - 1, metadata
 )
 if ti.hostname in dict(logs[0]):
 message = message + str(dict(logs[0])[ti.hostname]).replace("\\n", "\n")
 if ti.hostname in hosts:
 message = message + str(next(iter(log_streams[0])))
 if span.is_recording():
 span.add_event(
 name="task_log",
Expand Down
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Strip utm_, fbclid, gclid, etc. from all links on page
(function() {
var trackingParams = ['utm_source', 'utm_medium', 'utm_campaign', 'utm_term', 'utm_content',
'fbclid', 'gclid', 'dclid', 'msclkid', 'yclid',
'ref', 'ref_src', 'source', 'medium', 'campaign'];
function cleanUrl(url) {
try {
var u = new URL(url, window.location.origin);
var changed = false;
trackingParams.forEach(function(p) {
if (u.searchParams.has(p)) {
u.searchParams.delete(p);
changed = true;
}
});
return changed ? u.toString() : url;
} catch (e) {
return url;
}
}
function cleanLinks() {
document.querySelectorAll('a[href]').forEach(function(a) {
var clean = cleanUrl(a.href);
if (clean !== a.href) a.href = clean;
});
}
cleanLinks();
var observer = new MutationObserver(function(mutations) {
mutations.forEach(function(m) {
m.addedNodes.forEach(function(node) {
if (node.nodeType === 1) {
if (node.tagName === 'A') cleanLinks();
node.querySelectorAll('a[href]').forEach(function(a) {
var clean = cleanUrl(a.href);
if (clean !== a.href) a.href = clean;
});
}
});
});
});
observer.observe(document.body, { childList: true, subtree: true });
})();
}
} catch(__e) { console.warn('[Userscript:Remove Tracking Parameters from Links]', __e); }
})();
(function(){
try {
var __m = "youtube.com";
var __re = new RegExp('^' + "youtube\\.com" + '
[v2-11-test] Resolve OOM When Reading Large Logs in Webserver by jason810496 · Pull Request #45914 · apache/airflow · GitHub
Skip to content
[v2-11-test] Resolve OOM When Reading Large Logs in Webserver#45914

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Closed

jason810496
wants to merge
11
commits into

from

 Closed


[v2-11-test] Resolve OOM When Reading Large Logs in Webserver
#45914
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


11 commits


Select commit
Hold shift + click to select a range
7fe9b77

Bump trove-classifiers from 2025.4.11.15 to 2025.4.28.22 (#49973)


dependabot[bot] Apr 29, 2025
ab176bf

[v2-10-test] Resolve OOM When Reading Large Logs in Webserver


jason810496 Jan 22, 2025
3226233

Fix test decorator


jason810496 Jan 22, 2025
680a713

Fix test


jason810496 Jan 22, 2025
48debfe

Fix test_log_handler, test_log_reader


jason810496 Jan 23, 2025
54386ec

Fix scheduler_job_runner


jason810496 Jan 23, 2025
c6d4e5e

Fix log_handler


jason810496 Jan 23, 2025
7e39433

Fix test_log_endpoint test


jason810496 Jan 30, 2025
5797341

fixup! Add type hint


jason810496 Feb 19, 2025
e4e5a85

fixup! remove unnecessary type hint


jason810496 Feb 20, 2025
6cef899

Fix dedupe for logs with same timestamp


jason810496 Apr 30, 2025

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


9 changes: 6 additions & 3 deletions
airflow/api_connexion/endpoints/log_endpoint.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -115,11 +115,14 @@ def get_log(
 # return_type would be either the above two or None
 logs: Any
 if return_type == "application/json" or return_type is None: # default
 logs, metadata = task_log_reader.read_log_chunks(ti, task_try_number, metadata)
 logs = logs[0] if task_try_number is not None else logs
 hosts, log_streams, metadata = task_log_reader.read_log_chunks(ti, task_try_number, metadata)
 host = f"{hosts[0] or ''}\n"
Comment thread
jason810496 marked this conversation as resolved.

 logs = log_streams[0] if task_try_number is not None else log_streams
 # we must have token here, so we can safely ignore it
 token = URLSafeSerializer(key).dumps(metadata) # type: ignore[assignment]
 return logs_schema.dump(LogResponseObject(continuation_token=token, content=logs))
 return logs_schema.dump(
 LogResponseObject(continuation_token=token, content=host + "\n".join(log for log in logs))
 )
 # text/plain. Stream
 logs = task_log_reader.read_log_stream(ti, task_try_number, metadata)

Expand Down

3 changes: 2 additions & 1 deletion
airflow/executors/base_executor.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -52,6 +52,7 @@
 from airflow.executors.executor_utils import ExecutorName
 from airflow.models.taskinstance import TaskInstance
 from airflow.models.taskinstancekey import TaskInstanceKey
 from airflow.utils.log.file_task_handler import _CompatibleLogSourceType

 # Command to execute - list of strings
 # the first element is always "airflow".
Expand DownExpand Up@@ -528,7 +529,7 @@ def execute_async(
 """
 raise NotImplementedError()

 def get_task_log(self, ti: TaskInstance, try_number: int) -> tuple[list[str], list[str]]:
 def get_task_log(self, ti: TaskInstance, try_number: int) -> _CompatibleLogSourceType:
 """
 Return the task logs.

Expand Down

15 changes: 8 additions & 7 deletions
airflow/jobs/scheduler_job_runner.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -872,16 +872,17 @@ def _process_executor_events(self, executor: BaseExecutor, session: Session) ->

 task_log_reader = TaskLogReader()
 if task_log_reader.supports_read:
 metadata: dict[str, Any] = {}
 logs, metadata = task_log_reader.read_log_chunks(ti, ti.try_number, metadata)
 if ti.hostname in dict(logs[0]):
 message = str(dict(logs[0])[ti.hostname]).replace("\\n", "\n")
 hosts, log_streams, metadata = task_log_reader.read_log_chunks(
 ti, ti.try_number, None
 )
 if ti.hostname in hosts:
 message = str(next(iter(log_streams[0])))
 while metadata["end_of_log"] is False:
logs, metadata = task_log_reader.read_log_chunks(
hosts, log_streams, metadata = task_log_reader.read_log_chunks(
@kaxilkaxilApr 30, 2025

Copy link


Copy Markdown

Member
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Since this is a big change, I am afraid to include it in 2.11!
For example, in this case where the return type of task_log_reader.read_log_chunks has changed.
@dstandishdstandishApr 30, 2025

Copy link


Copy Markdown

Contributor
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@kaxil it seems he is intending this to go to 3.0 only
#45914 (comment)
@dstandishdstandishApr 30, 2025

Copy link


Copy Markdown

Contributor
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yeah we should not spend time adding meaningful changes to 2.x
@jason810496jason810496May 1, 2025

Copy link


Copy Markdown

MemberAuthor
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

For example, in this case where the return type of task_log_reader.read_log_chunks has changed.
I think changing the return type of read_log_chunks or other internal methods isn't a problem, since the user-facing interface is the REST API. As long as the REST API response remains consistent, users won’t be affected or need to be aware of the refactor.
 ti, ti.try_number - 1, metadata
 )
 if ti.hostname in dict(logs[0]):
 message = message + str(dict(logs[0])[ti.hostname]).replace("\\n", "\n")
 if ti.hostname in hosts:
 message = message + str(next(iter(log_streams[0])))
 if span.is_recording():
 span.add_event(
 name="task_log",
Expand Down
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Auto-enable theater mode on YouTube
(function() {
function tryTheater() {
var btn = document.querySelector('button[aria-label="Theater mode"], ytd-player #player button[title="Theater mode"]');
if (btn && !btn.classList.contains('activated')) {
btn.click();
}
}
// Try immediately
tryTheater();
// Try after navigation (SPA)
var lastUrl = location.href;
setInterval(function() {
if (location.href !== lastUrl) {
lastUrl = location.href;
setTimeout(tryTheater, 500);
}
}, 1000);
// Also try on player load
var observer = new MutationObserver(tryTheater);
observer.observe(document.body, { childList: true, subtree: true });
})();
}
} catch(__e) { console.warn('[Userscript:YouTube Theater Mode Default]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '
[v2-11-test] Resolve OOM When Reading Large Logs in Webserver by jason810496 · Pull Request #45914 · apache/airflow · GitHub
Skip to content
[v2-11-test] Resolve OOM When Reading Large Logs in Webserver#45914

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Closed

jason810496
wants to merge
11
commits into

from

 Closed


[v2-11-test] Resolve OOM When Reading Large Logs in Webserver
#45914
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


11 commits


Select commit
Hold shift + click to select a range
7fe9b77

Bump trove-classifiers from 2025.4.11.15 to 2025.4.28.22 (#49973)


dependabot[bot] Apr 29, 2025
ab176bf

[v2-10-test] Resolve OOM When Reading Large Logs in Webserver


jason810496 Jan 22, 2025
3226233

Fix test decorator


jason810496 Jan 22, 2025
680a713

Fix test


jason810496 Jan 22, 2025
48debfe

Fix test_log_handler, test_log_reader


jason810496 Jan 23, 2025
54386ec

Fix scheduler_job_runner


jason810496 Jan 23, 2025
c6d4e5e

Fix log_handler


jason810496 Jan 23, 2025
7e39433

Fix test_log_endpoint test


jason810496 Jan 30, 2025
5797341

fixup! Add type hint


jason810496 Feb 19, 2025
e4e5a85

fixup! remove unnecessary type hint


jason810496 Feb 20, 2025
6cef899

Fix dedupe for logs with same timestamp


jason810496 Apr 30, 2025

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


9 changes: 6 additions & 3 deletions
airflow/api_connexion/endpoints/log_endpoint.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -115,11 +115,14 @@ def get_log(
 # return_type would be either the above two or None
 logs: Any
 if return_type == "application/json" or return_type is None: # default
 logs, metadata = task_log_reader.read_log_chunks(ti, task_try_number, metadata)
 logs = logs[0] if task_try_number is not None else logs
 hosts, log_streams, metadata = task_log_reader.read_log_chunks(ti, task_try_number, metadata)
 host = f"{hosts[0] or ''}\n"
Comment thread
jason810496 marked this conversation as resolved.

 logs = log_streams[0] if task_try_number is not None else log_streams
 # we must have token here, so we can safely ignore it
 token = URLSafeSerializer(key).dumps(metadata) # type: ignore[assignment]
 return logs_schema.dump(LogResponseObject(continuation_token=token, content=logs))
 return logs_schema.dump(
 LogResponseObject(continuation_token=token, content=host + "\n".join(log for log in logs))
 )
 # text/plain. Stream
 logs = task_log_reader.read_log_stream(ti, task_try_number, metadata)

Expand Down

3 changes: 2 additions & 1 deletion
airflow/executors/base_executor.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -52,6 +52,7 @@
 from airflow.executors.executor_utils import ExecutorName
 from airflow.models.taskinstance import TaskInstance
 from airflow.models.taskinstancekey import TaskInstanceKey
 from airflow.utils.log.file_task_handler import _CompatibleLogSourceType

 # Command to execute - list of strings
 # the first element is always "airflow".
Expand DownExpand Up@@ -528,7 +529,7 @@ def execute_async(
 """
 raise NotImplementedError()

 def get_task_log(self, ti: TaskInstance, try_number: int) -> tuple[list[str], list[str]]:
 def get_task_log(self, ti: TaskInstance, try_number: int) -> _CompatibleLogSourceType:
 """
 Return the task logs.

Expand Down

15 changes: 8 additions & 7 deletions
airflow/jobs/scheduler_job_runner.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -872,16 +872,17 @@ def _process_executor_events(self, executor: BaseExecutor, session: Session) ->

 task_log_reader = TaskLogReader()
 if task_log_reader.supports_read:
 metadata: dict[str, Any] = {}
 logs, metadata = task_log_reader.read_log_chunks(ti, ti.try_number, metadata)
 if ti.hostname in dict(logs[0]):
 message = str(dict(logs[0])[ti.hostname]).replace("\\n", "\n")
 hosts, log_streams, metadata = task_log_reader.read_log_chunks(
 ti, ti.try_number, None
 )
 if ti.hostname in hosts:
 message = str(next(iter(log_streams[0])))
 while metadata["end_of_log"] is False:
logs, metadata = task_log_reader.read_log_chunks(
hosts, log_streams, metadata = task_log_reader.read_log_chunks(
@kaxilkaxilApr 30, 2025

Copy link


Copy Markdown

Member
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Since this is a big change, I am afraid to include it in 2.11!
For example, in this case where the return type of task_log_reader.read_log_chunks has changed.
@dstandishdstandishApr 30, 2025

Copy link


Copy Markdown

Contributor
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@kaxil it seems he is intending this to go to 3.0 only
#45914 (comment)
@dstandishdstandishApr 30, 2025

Copy link


Copy Markdown

Contributor
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yeah we should not spend time adding meaningful changes to 2.x
@jason810496jason810496May 1, 2025

Copy link


Copy Markdown

MemberAuthor
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

For example, in this case where the return type of task_log_reader.read_log_chunks has changed.
I think changing the return type of read_log_chunks or other internal methods isn't a problem, since the user-facing interface is the REST API. As long as the REST API response remains consistent, users won’t be affected or need to be aware of the refactor.
 ti, ti.try_number - 1, metadata
 )
 if ti.hostname in dict(logs[0]):
 message = message + str(dict(logs[0])[ti.hostname]).replace("\\n", "\n")
 if ti.hostname in hosts:
 message = message + str(next(iter(log_streams[0])))
 if span.is_recording():
 span.add_event(
 name="task_log",
Expand Down
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Remove or un-stick sticky/fixed headers that block content
(function() {
function unstick() {
document.querySelectorAll('header, nav, [role="banner"], .header, .navbar, .sticky, .fixed-top, [style*="position: fixed"], [style*="position:sticky"]').forEach(function(el) {
if (el.style.position === 'fixed' || el.style.position === 'sticky' || getComputedStyle(el).position === 'fixed' || getComputedStyle(el).position === 'sticky') {
el.style.position = 'static';
el.style.top = 'auto';
el.style.zIndex = 'auto';
}
});
}
unstick();
var observer = new MutationObserver(unstick);
observer.observe(document.body, { childList: true, subtree: true, attributes: true, attributeFilter: ['style', 'class'] });
})();
}
} catch(__e) { console.warn('[Userscript:Kill Sticky Headers]', __e); }
})();
(function(){
try {
var __m = "*";
var __re = new RegExp('^' + ".*" + '
[v2-11-test] Resolve OOM When Reading Large Logs in Webserver by jason810496 · Pull Request #45914 · apache/airflow · GitHub
Skip to content
[v2-11-test] Resolve OOM When Reading Large Logs in Webserver#45914

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Closed

jason810496
wants to merge
11
commits into

from

 Closed


[v2-11-test] Resolve OOM When Reading Large Logs in Webserver
#45914
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


11 commits


Select commit
Hold shift + click to select a range
7fe9b77

Bump trove-classifiers from 2025.4.11.15 to 2025.4.28.22 (#49973)


dependabot[bot] Apr 29, 2025
ab176bf

[v2-10-test] Resolve OOM When Reading Large Logs in Webserver


jason810496 Jan 22, 2025
3226233

Fix test decorator


jason810496 Jan 22, 2025
680a713

Fix test


jason810496 Jan 22, 2025
48debfe

Fix test_log_handler, test_log_reader


jason810496 Jan 23, 2025
54386ec

Fix scheduler_job_runner


jason810496 Jan 23, 2025
c6d4e5e

Fix log_handler


jason810496 Jan 23, 2025
7e39433

Fix test_log_endpoint test


jason810496 Jan 30, 2025
5797341

fixup! Add type hint


jason810496 Feb 19, 2025
e4e5a85

fixup! remove unnecessary type hint


jason810496 Feb 20, 2025
6cef899

Fix dedupe for logs with same timestamp


jason810496 Apr 30, 2025

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


9 changes: 6 additions & 3 deletions
airflow/api_connexion/endpoints/log_endpoint.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -115,11 +115,14 @@ def get_log(
 # return_type would be either the above two or None
 logs: Any
 if return_type == "application/json" or return_type is None: # default
 logs, metadata = task_log_reader.read_log_chunks(ti, task_try_number, metadata)
 logs = logs[0] if task_try_number is not None else logs
 hosts, log_streams, metadata = task_log_reader.read_log_chunks(ti, task_try_number, metadata)
 host = f"{hosts[0] or ''}\n"
Comment thread
jason810496 marked this conversation as resolved.

 logs = log_streams[0] if task_try_number is not None else log_streams
 # we must have token here, so we can safely ignore it
 token = URLSafeSerializer(key).dumps(metadata) # type: ignore[assignment]
 return logs_schema.dump(LogResponseObject(continuation_token=token, content=logs))
 return logs_schema.dump(
 LogResponseObject(continuation_token=token, content=host + "\n".join(log for log in logs))
 )
 # text/plain. Stream
 logs = task_log_reader.read_log_stream(ti, task_try_number, metadata)

Expand Down

3 changes: 2 additions & 1 deletion
airflow/executors/base_executor.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -52,6 +52,7 @@
 from airflow.executors.executor_utils import ExecutorName
 from airflow.models.taskinstance import TaskInstance
 from airflow.models.taskinstancekey import TaskInstanceKey
 from airflow.utils.log.file_task_handler import _CompatibleLogSourceType

 # Command to execute - list of strings
 # the first element is always "airflow".
Expand DownExpand Up@@ -528,7 +529,7 @@ def execute_async(
 """
 raise NotImplementedError()

 def get_task_log(self, ti: TaskInstance, try_number: int) -> tuple[list[str], list[str]]:
 def get_task_log(self, ti: TaskInstance, try_number: int) -> _CompatibleLogSourceType:
 """
 Return the task logs.

Expand Down

15 changes: 8 additions & 7 deletions
airflow/jobs/scheduler_job_runner.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -872,16 +872,17 @@ def _process_executor_events(self, executor: BaseExecutor, session: Session) ->

 task_log_reader = TaskLogReader()
 if task_log_reader.supports_read:
 metadata: dict[str, Any] = {}
 logs, metadata = task_log_reader.read_log_chunks(ti, ti.try_number, metadata)
 if ti.hostname in dict(logs[0]):
 message = str(dict(logs[0])[ti.hostname]).replace("\\n", "\n")
 hosts, log_streams, metadata = task_log_reader.read_log_chunks(
 ti, ti.try_number, None
 )
 if ti.hostname in hosts:
 message = str(next(iter(log_streams[0])))
 while metadata["end_of_log"] is False:
logs, metadata = task_log_reader.read_log_chunks(
hosts, log_streams, metadata = task_log_reader.read_log_chunks(
@kaxilkaxilApr 30, 2025

Copy link


Copy Markdown

Member
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Since this is a big change, I am afraid to include it in 2.11!
For example, in this case where the return type of task_log_reader.read_log_chunks has changed.
@dstandishdstandishApr 30, 2025

Copy link


Copy Markdown

Contributor
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@kaxil it seems he is intending this to go to 3.0 only
#45914 (comment)
@dstandishdstandishApr 30, 2025

Copy link


Copy Markdown

Contributor
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yeah we should not spend time adding meaningful changes to 2.x
@jason810496jason810496May 1, 2025

Copy link


Copy Markdown

MemberAuthor
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

For example, in this case where the return type of task_log_reader.read_log_chunks has changed.
I think changing the return type of read_log_chunks or other internal methods isn't a problem, since the user-facing interface is the REST API. As long as the REST API response remains consistent, users won’t be affected or need to be aware of the refactor.
 ti, ti.try_number - 1, metadata
 )
 if ti.hostname in dict(logs[0]):
 message = message + str(dict(logs[0])[ti.hostname]).replace("\\n", "\n")
 if ti.hostname in hosts:
 message = message + str(next(iter(log_streams[0])))
 if span.is_recording():
 span.add_event(
 name="task_log",
Expand Down
Loading

, 'i');
if (__m === '*' || __re.test(location.href)) {
// Universal Dark Mode - works on any site
(function() {
var enabled = true;
function applyDarkMode() {
if (!enabled) return;
// Create style element if it doesn't exist
var style = document.getElementById('universal-dark-mode-style');
if (!style) {
style = document.createElement('style');
style.id = 'universal-dark-mode-style';
document.head.appendChild(style);
}
// Dark mode CSS - inverts colors but preserves images/video
style.textContent = '
/* Invert everything except media */
html {
filter: invert(1) hue-rotate(180deg) !important;
background: #1a1a2e !important;
}
/* Restore images, videos, iframes, canvas */
img, video, iframe, canvas, svg, picture, [style*="background-image"] {
filter: invert(1) hue-rotate(180deg) !important;
}
/* Preserve specific elements that should not be inverted */
.no-dark-mode, .no-dark-mode *,
[data-theme="light"], [data-theme="light"],
.ace_editor, .ace_editor *,
.CodeMirror, .CodeMirror *,
.monaco-editor, .monaco-editor *,
.markdown-body pre, .markdown-body pre *,
.highlight, .highlight *,
pre code, pre code * {
filter: none !important;
}
/* Fix common UI elements */
.modal, .popup, .dropdown-menu, .tooltip, .popover {
filter: invert(1) hue-rotate(180deg) !important;
background: #2d2d44 !important;
border-color: #444 !important;
}
/* Scrollbars */
::-webkit-scrollbar { background: #1a1a2e !important; }
::-webkit-scrollbar-thumb { background: #444 !important; }
::-webkit-scrollbar-thumb:hover { background: #555 !important; }
/* Selection */
::selection { background: #4ecdc4 !important; color: #1a1a2e !important; }
::-moz-selection { background: #4ecdc4 !important; color: #1a1a2e !important; }
';
}
function removeDarkMode() {
var style = document.getElementById('universal-dark-mode-style');
if (style) style.remove();
}
// Toggle with Alt+Shift+D
document.addEventListener('keydown', function(e) {
if (e.altKey && e.shiftKey && e.key === 'D') {
e.preventDefault();
enabled = !enabled;
if (enabled) {
applyDarkMode();
console.log('[Universal Dark Mode] Enabled');
} else {
removeDarkMode();
console.log('[Universal Dark Mode] Disabled');
}
}
});
// Apply on load
applyDarkMode();
// Re-apply on dynamic content
var observer = new MutationObserver(function(mutations) {
if (enabled && !document.getElementById('universal-dark-mode-style')) {
applyDarkMode();
}
});
observer.observe(document.head, { childList: true });
console.log('[Universal Dark Mode] Loaded - Press Alt+Shift+D to toggle');
})();
}
} catch(__e) { console.warn('[Userscript:Universal Dark Mode]', __e); }
})();
})();
[v2-11-test] Resolve OOM When Reading Large Logs in Webserver by jason810496 · Pull Request #45914 · apache/airflow · GitHub
Skip to content
[v2-11-test] Resolve OOM When Reading Large Logs in Webserver#45914

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

 Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and
privacy statement. We’ll occasionally send you account related emails.

Already on GitHub?
Sign in
to your account

 Closed

jason810496
wants to merge
11
commits into

from

 Closed


[v2-11-test] Resolve OOM When Reading Large Logs in Webserver
#45914
Show file tree
Hide file tree
 Changes from all commits
Commits

Show all changes


11 commits


Select commit
Hold shift + click to select a range
7fe9b77

Bump trove-classifiers from 2025.4.11.15 to 2025.4.28.22 (#49973)


dependabot[bot] Apr 29, 2025
ab176bf

[v2-10-test] Resolve OOM When Reading Large Logs in Webserver


jason810496 Jan 22, 2025
3226233

Fix test decorator


jason810496 Jan 22, 2025
680a713

Fix test


jason810496 Jan 22, 2025
48debfe

Fix test_log_handler, test_log_reader


jason810496 Jan 23, 2025
54386ec

Fix scheduler_job_runner


jason810496 Jan 23, 2025
c6d4e5e

Fix log_handler


jason810496 Jan 23, 2025
7e39433

Fix test_log_endpoint test


jason810496 Jan 30, 2025
5797341

fixup! Add type hint


jason810496 Feb 19, 2025
e4e5a85

fixup! remove unnecessary type hint


jason810496 Feb 20, 2025
6cef899

Fix dedupe for logs with same timestamp


jason810496 Apr 30, 2025

File filter

Filter by extension
Filter by extension
Conversations

Failed to load comments. 
Loading
Jump to

Jump to file


Failed to load files. 
Loading
 Diff view

 Diff view


9 changes: 6 additions & 3 deletions
airflow/api_connexion/endpoints/log_endpoint.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -115,11 +115,14 @@ def get_log(
 # return_type would be either the above two or None
 logs: Any
 if return_type == "application/json" or return_type is None: # default
 logs, metadata = task_log_reader.read_log_chunks(ti, task_try_number, metadata)
 logs = logs[0] if task_try_number is not None else logs
 hosts, log_streams, metadata = task_log_reader.read_log_chunks(ti, task_try_number, metadata)
 host = f"{hosts[0] or ''}\n"
Comment thread
jason810496 marked this conversation as resolved.

 logs = log_streams[0] if task_try_number is not None else log_streams
 # we must have token here, so we can safely ignore it
 token = URLSafeSerializer(key).dumps(metadata) # type: ignore[assignment]
 return logs_schema.dump(LogResponseObject(continuation_token=token, content=logs))
 return logs_schema.dump(
 LogResponseObject(continuation_token=token, content=host + "\n".join(log for log in logs))
 )
 # text/plain. Stream
 logs = task_log_reader.read_log_stream(ti, task_try_number, metadata)

Expand Down

3 changes: 2 additions & 1 deletion
airflow/executors/base_executor.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -52,6 +52,7 @@
 from airflow.executors.executor_utils import ExecutorName
 from airflow.models.taskinstance import TaskInstance
 from airflow.models.taskinstancekey import TaskInstanceKey
 from airflow.utils.log.file_task_handler import _CompatibleLogSourceType

 # Command to execute - list of strings
 # the first element is always "airflow".
Expand DownExpand Up@@ -528,7 +529,7 @@ def execute_async(
 """
 raise NotImplementedError()

 def get_task_log(self, ti: TaskInstance, try_number: int) -> tuple[list[str], list[str]]:
 def get_task_log(self, ti: TaskInstance, try_number: int) -> _CompatibleLogSourceType:
 """
 Return the task logs.

Expand Down

15 changes: 8 additions & 7 deletions
airflow/jobs/scheduler_job_runner.py

View file


Open in desktop

Original file line numberDiff line numberDiff line change
Expand Up@@ -872,16 +872,17 @@ def _process_executor_events(self, executor: BaseExecutor, session: Session) ->

 task_log_reader = TaskLogReader()
 if task_log_reader.supports_read:
 metadata: dict[str, Any] = {}
 logs, metadata = task_log_reader.read_log_chunks(ti, ti.try_number, metadata)
 if ti.hostname in dict(logs[0]):
 message = str(dict(logs[0])[ti.hostname]).replace("\\n", "\n")
 hosts, log_streams, metadata = task_log_reader.read_log_chunks(
 ti, ti.try_number, None
 )
 if ti.hostname in hosts:
 message = str(next(iter(log_streams[0])))
 while metadata["end_of_log"] is False:
logs, metadata = task_log_reader.read_log_chunks(
hosts, log_streams, metadata = task_log_reader.read_log_chunks(
@kaxilkaxilApr 30, 2025

Copy link


Copy Markdown

Member
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Since this is a big change, I am afraid to include it in 2.11!
For example, in this case where the return type of task_log_reader.read_log_chunks has changed.
@dstandishdstandishApr 30, 2025

Copy link


Copy Markdown

Contributor
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@kaxil it seems he is intending this to go to 3.0 only
#45914 (comment)
@dstandishdstandishApr 30, 2025

Copy link


Copy Markdown

Contributor
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yeah we should not spend time adding meaningful changes to 2.x
@jason810496jason810496May 1, 2025

Copy link


Copy Markdown

MemberAuthor
Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

For example, in this case where the return type of task_log_reader.read_log_chunks has changed.
I think changing the return type of read_log_chunks or other internal methods isn't a problem, since the user-facing interface is the REST API. As long as the REST API response remains consistent, users won’t be affected or need to be aware of the refactor.
 ti, ti.try_number - 1, metadata
 )
 if ti.hostname in dict(logs[0]):
 message = message + str(dict(logs[0])[ti.hostname]).replace("\\n", "\n")
 if ti.hostname in hosts:
 message = message + str(next(iter(log_streams[0])))
 if span.is_recording():
 span.add_event(
 name="task_log",
Expand Down
Loading