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
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194 | @dataclasses.dataclass
class Arguments:
"""
Use drain algorithm to cluster similar log lines
from the provided log files.
Arguments for the log clustering script :
"""
# logs files paths to process
logfile_paths: tyro.conf.Positional[tuple[pathlib.Path, ...]]
# baseline log files to compare with. If set, the clusters will be compared to baseline clusters.
baseline: Annotated[tuple[pathlib.Path, ...], tyro.conf.arg(aliases=["-b"])] = ()
# configuration file for the drain3 template miner. If not provided default file will be searched in user's home
# and created if not present.
cfg_file: Annotated[pathlib.Path, tyro.conf.arg(aliases=["-c"])] = HOME_CFG_FILE
# If set, filter input log lines which does not match the regex (re python module syntax).
# Example: '.*(\| Warning |\| Error ).*'
filter: Annotated[str, tyro.conf.arg(aliases=["-f"])] = ""
# applied on baseline if set instead of filter.
base_filter: str = ""
# Format code (for strptime) of the timestamp extracted of each line to convert string into datetime.
# If empty, the program will try with these formats : "%H:%M:%S.%f", "%H:%M:%S,%f" and "%H:%M:%S" to extract the timestamp.
# User must provide both time_format and time_pattern.
time_format: Annotated[str, tyro.conf.arg(aliases=["-T"])] = ""
# regexp pattern to extract timestamp of each line. Default pattern is r'(\d{2}:\d{2}:\d{2}(?:\.\d{1,9})?)'
# to extract time like "15:04:05.000" or "15:04:05".
time_pattern: Annotated[str, tyro.conf.arg(aliases=["-t"])] = ""
# timestamp used to filter log lines based on the timestamp extracted with time_pattern.
# Example: "09:04:05.000" will only process logs after this time.
time_min: Annotated[str, tyro.conf.arg(aliases=["-m"])] = ""
# timestamp used to filer baseline log lines if set instead of time_min.
base_time_min: str = ""
# timestamp used to filter log lines based on the time extracted with time_pattern.
# Example: "12:04:05.000" will only process logs before this time.
time_max: Annotated[str, tyro.conf.arg(aliases=["-M"])] = ""
# timestamp used to filer baseline log lines if set instead of time_max.
base_time_max: str = ""
# assume log lines are not ordered by timestamp, and do not stop processing
# after reaching a line with time after time_max.
unordered_time: Annotated[tyro.conf.FlagCreatePairsOff[bool], tyro.conf.arg(aliases=["-u"])] = False
# If set, all baseline clusters will also be displayed.
display_common: Annotated[tyro.conf.FlagCreatePairsOff[bool], tyro.conf.arg(aliases=["-C"])] = False
# If set, the clusters will be ordered lexicographically.
lex_order: Annotated[tyro.conf.FlagCreatePairsOff[bool], tyro.conf.arg(aliases=["-l"])] = False
# The clusters will be ordered by this total length. Where total length is the sum of all
# log lines lengths belonging to the cluster.
size_order: Annotated[tyro.conf.FlagCreatePairsOff[bool], tyro.conf.arg(aliases=["-z"])] = False
# Similarity threshold for the template miner to group lines together.
# A higher value will lead to create more clusters. Drain default value is 0.4.
similarity_threshold: Annotated[float | None, tyro.conf.arg(aliases=["-s"])] = None
# depth of the tree to build the templates miner,
# a higher value will lead to create more clusters.
# The higher the value, the more tokens of the log lines
# will be considered to build the clusters. Increase this value
# to make clustering rely on distant tokens. Drain default value is 4.
tree_depth: Annotated[int | None, tyro.conf.arg(aliases=["-d"])] = None
# If set, output clusters in plain text format without colors or table formatting,
# making it easy to process with bash tools like grep, awk, or cut.
raw: Annotated[tyro.conf.FlagCreatePairsOff[bool], tyro.conf.arg(aliases=["-r"])] = False
# return the version and exit
version: Annotated[tyro.conf.FlagCreatePairsOff[bool], tyro.conf.arg(aliases=["-v"])] = False
def _post_init_time_pattern_and_format(self) -> None:
if self.time_pattern:
try:
re.compile(self.time_pattern)
except re.error as e:
logging.critical("Invalid time regex pattern: %s. Error: %s", self.time_pattern, e)
sys.exit(ErrorCode.TIME_PATTERN_ERROR.value)
if not self.time_format:
error_message = "Time format is required when time pattern is provided."
logging.critical(error_message)
sys.exit(ErrorCode.TIME_PATTERN_ERROR.value)
if self.time_format and not self.time_pattern:
error_message = "Time pattern is required when time format is provided."
logging.critical(error_message)
sys.exit(ErrorCode.TIME_PATTERN_ERROR.value)
def _post_init_fallback_pattern_format(self) -> None:
if self.time_min or self.time_max:
time_value = self.time_min if self.time_min else self.time_max
if not self.time_format and not self.time_pattern:
logging.info(
"No time pattern or format provided. Trying to guess them from time value '%s'.",
time_value,
)
guessed_regexp_format = guess_time_regexp_and_format_code(time_value)
if not guessed_regexp_format:
logging.critical(
"Failed to guess time pattern and format from time value '%s'. Please provide a valid time pattern and format.",
time_value,
)
sys.exit(ErrorCode.TIME_PATTERN_ERROR.value)
self.time_pattern, self.time_format = guessed_regexp_format
logging.info(
"Guessed time pattern: '%s' and time format: '%s' from time value '%s'.",
self.time_pattern,
self.time_format,
time_value,
)
def _post_init_time(self, time_min: str, time_max: str) -> None:
if time_min:
try:
datetime.strptime(time_min, self.time_format)
except ValueError as e:
logging.critical("Invalid time_min format: %s. Error: %s", time_min, e)
sys.exit(ErrorCode.TIME_PATTERN_ERROR.value)
if time_max:
try:
datetime.strptime(time_max, self.time_format)
except ValueError as e:
logging.critical("Invalid time_max format: %s. Error: %s", time_max, e)
sys.exit(ErrorCode.TIME_PATTERN_ERROR.value)
if time_max and time_min and self.time_format:
time_min_dt = datetime.strptime(time_min, self.time_format)
time_max_dt = datetime.strptime(time_max, self.time_format)
if time_max_dt < time_min_dt:
logging.critical("time_max (%s) is before time_min (%s).", time_max, time_min)
sys.exit(ErrorCode.TIME_PATTERN_ERROR.value)
def _post_init_logfile_paths(self) -> None:
if self.logfile_paths is None or len(self.logfile_paths) == 0:
error_message = "No log files provided. Please specify at least one log file."
logging.critical(error_message)
sys.exit(ErrorCode.NO_LOG_FILES.value)
else:
for logfile_path in self.logfile_paths:
if not logfile_path.exists():
logging.critical("Log file not found: %s", logfile_path)
sys.exit(ErrorCode.FILE_NOT_FOUND.value)
if not logfile_path.is_file():
logging.critical("Not a file: %s", logfile_path)
sys.exit(ErrorCode.FILE_NOT_FOUND.value)
def _post_init_baseline(self) -> None:
if self.baseline:
for baseline_path in self.baseline:
if not baseline_path.exists():
logging.critical("Baseline file not found: %s", baseline_path)
sys.exit(ErrorCode.FILE_NOT_FOUND.value)
if not baseline_path.is_file():
logging.critical("Not a file: %s", baseline_path)
sys.exit(ErrorCode.FILE_NOT_FOUND.value)
def _post_init_filter(self, line_filter: str) -> None:
if line_filter:
try:
re.compile(line_filter)
except re.error as e:
logging.critical("Invalid regex pattern: %s. Error: %s", line_filter, e)
sys.exit(ErrorCode.INVALID_REGEX.value)
def __post_init__(self) -> None:
if self.version:
return
self._post_init_logfile_paths()
self._post_init_baseline()
if self.tree_depth and self.tree_depth < 3:
error_message = f"The tree depth is set to {self.tree_depth}. Minimum value is 3."
logging.critical(error_message)
sys.exit(ErrorCode.INVALID_TREE_DEPTH.value)
if self.filter:
try:
re.compile(self.filter)
except re.error as e:
logging.critical("Invalid regex pattern: %s. Error: %s", self.filter, e)
sys.exit(ErrorCode.INVALID_REGEX.value)
self._post_init_filter(self.filter)
self._post_init_filter(self.base_filter)
self._post_init_time_pattern_and_format()
self._post_init_fallback_pattern_format()
self._post_init_time(self.time_min, self.time_max)
self._post_init_time(self.base_time_min, self.base_time_max)
|