Skip to content
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

Fix timestamp parse failure for k8s executor pod tailing #31175

Merged
merged 6 commits into from
May 11, 2023
Merged

Fix timestamp parse failure for k8s executor pod tailing #31175

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
00bffad
Fix timestamp parse failure for k8s executor pod tailing
dstandish May 9, 2023
679af50
fix test
dstandish May 10, 2023
d015efb
use existing function
dstandish May 10, 2023
bbcc7b3
fix tests
dstandish May 10, 2023
f6b84e5
fix test actually
dstandish May 10, 2023
a727104
fix test actually actually
dstandish May 10, 2023
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: 5 additions & 4 deletions airflow/executors/kubernetes_executor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,7 +48,7 @@
from airflow.kubernetes.pod_generator import PodGenerator
from airflow.models.taskinstance import TaskInstance
from airflow.utils.event_scheduler import EventScheduler
from airflow.utils.log.logging_mixin import LoggingMixin
from airflow.utils.log.logging_mixin import LoggingMixin, remove_escape_codes
from airflow.utils.session import NEW_SESSION, provide_session
from airflow.utils.state import State, TaskInstanceState

Expand Down Expand Up @@ -794,7 +794,7 @@ def get_task_log(self, ti: TaskInstance, try_number: int) -> tuple[list[str], li

client = get_kube_client()

messages.append(f"Trying to get logs (last 100 lines) from worker pod {ti.hostname}")
messages.append(f"Attempting to fetch logs from pod {ti.hostname} through kube API")
selector = PodGenerator.build_selector_for_k8s_executor_pod(
dag_id=ti.dag_id,
task_id=ti.task_id,
Expand All @@ -820,9 +820,10 @@ def get_task_log(self, ti: TaskInstance, try_number: int) -> tuple[list[str], li
tail_lines=100,
_preload_content=False,
)

for line in res:
log.append(line.decode())
log.append(remove_escape_codes(line.decode()))
if log:
messages.append("Found logs through kube API")
except Exception as e:
messages.append(f"Reading from k8s pod logs failed: {str(e)}")
return messages, ["\n".join(log)]
Expand Down
2 changes: 1 addition & 1 deletion airflow/utils/log/file_task_handler.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -315,7 +315,7 @@ def _read(
if response:
executor_messages, executor_logs = response
if executor_messages:
messages_list.extend(messages_list)
messages_list.extend(executor_messages)
if not (remote_logs and ti.state not in State.unfinished):
# when finished, if we have remote logs, no need to check local
worker_log_full_path = Path(self.local_base, worker_log_rel_path)
Expand Down
6 changes: 5 additions & 1 deletion tests/executors/test_kubernetes_executor.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1140,7 +1140,11 @@ def test_get_task_log(self, mock_get_kube_client, create_task_instance_of_operat
messages, logs = executor.get_task_log(ti=ti, try_number=1)

mock_kube_client.read_namespaced_pod_log.assert_called_once()
assert "Trying to get logs (last 100 lines) from worker pod " in messages
expected_messages = [
"Attempting to fetch logs from pod through kube API",
"Found logs through kube API",
]
assert messages == expected_messages
assert logs[0] == "a_\nb_\nc_"

mock_kube_client.reset_mock()
Expand Down