forked from apache/airflow
/
log_reader.py
119 lines (102 loc) · 4.78 KB
/
log_reader.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements. See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership. The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied. See the License for the
# specific language governing permissions and limitations
# under the License.
import logging
from typing import Dict, Iterator, List, Optional, Tuple
from airflow.compat.functools import cached_property
from airflow.configuration import conf
from airflow.models import TaskInstance
from airflow.utils.helpers import render_log_filename
from airflow.utils.log.logging_mixin import ExternalLoggingMixin
class TaskLogReader:
"""Task log reader"""
def read_log_chunks(
self, ti: TaskInstance, try_number: Optional[int], metadata
) -> Tuple[List[Tuple[Tuple[str, str]]], Dict[str, str]]:
"""
Reads chunks of Task Instance logs.
:param ti: The taskInstance
:type ti: TaskInstance
:param try_number: If provided, logs for the given try will be returned.
Otherwise, logs from all attempts are returned.
:type try_number: Optional[int]
:param metadata: A dictionary containing information about how to read the task log
:type metadata: dict
:rtype: Tuple[List[Tuple[Tuple[str, str]]], Dict[str, str]]
The following is an example of how to use this method to read log:
.. code-block:: python
logs, metadata = task_log_reader.read_log_chunks(ti, try_number, metadata)
logs = logs[0] if try_number is not None else logs
where task_log_reader is an instance of TaskLogReader. The metadata will always
contain information about the task log which can enable you read logs to the
end.
"""
logs, metadatas = self.log_handler.read(ti, try_number, metadata=metadata)
metadata = metadatas[0]
return logs, metadata
def read_log_stream(self, ti: TaskInstance, try_number: Optional[int], metadata: dict) -> Iterator[str]:
"""
Used to continuously read log to the end
:param ti: The Task Instance
:type ti: TaskInstance
:param try_number: the task try number
:type try_number: Optional[int]
:param metadata: A dictionary containing information about how to read the task log
:type metadata: dict
:rtype: Iterator[str]
"""
if try_number is None:
next_try = ti.next_try_number
try_numbers = list(range(1, next_try))
else:
try_numbers = [try_number]
for current_try_number in try_numbers:
metadata.pop('end_of_log', None)
metadata.pop('max_offset', None)
metadata.pop('offset', None)
while 'end_of_log' not in metadata or not metadata['end_of_log']:
logs, metadata = self.read_log_chunks(ti, current_try_number, metadata)
for host, log in logs[0]:
yield "\n".join([host, log]) + "\n"
@cached_property
def log_handler(self):
"""Log handler, which is configured to read logs."""
logger = logging.getLogger('airflow.task')
task_log_reader = conf.get('logging', 'task_log_reader')
handler = next((handler for handler in logger.handlers if handler.name == task_log_reader), None)
return handler
@property
def supports_read(self):
"""Checks if a read operation is supported by a current log handler."""
return hasattr(self.log_handler, 'read')
@property
def supports_external_link(self):
"""Check if the logging handler supports external links (e.g. to Elasticsearch, Stackdriver, etc)."""
return isinstance(self.log_handler, ExternalLoggingMixin)
def render_log_filename(self, ti: TaskInstance, try_number: Optional[int] = None):
"""
Renders the log attachment filename
:param ti: The task instance
:type ti: TaskInstance
:param try_number: The task try number
:type try_number: Optional[int]
:rtype: str
"""
filename_template = conf.get('logging', 'LOG_FILENAME_TEMPLATE')
attachment_filename = render_log_filename(
ti=ti, try_number="all" if try_number is None else try_number, filename_template=filename_template
)
return attachment_filename