Skip to content

Logs Miner

ExceptionLineAfterMaxTime

Bases: Exception

Raised when a log line has a time after the specified maximum time.

Source code in src/gool/logs_miner.py 
14
15
16
17
class ExceptionLineAfterMaxTime(Exception):
    """Raised when a log line has a time after the specified maximum time."""

    pass

LogsMiner

Source code in src/gool/logs_miner.py 
 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
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
class LogsMiner:
    def __init__(
        self,
        drain3_cfg_file: Path,
        similarity_threshold: float | None,
        tree_depth: int | None,
        *,
        filter_regexp: re.Pattern[str] | None = None,
        time_pattern_regexp: re.Pattern[str] | None = None,
        time_format: str | None = None,
        time_min: datetime | None = None,
        time_max: datetime | None = None,
        unordered_time: bool = False,
    ):
        """
        Initialize the LogsMiner.

        Args:
            drain3_cfg_file (Path): Path to drain3 cfg file.
            similarity_threshold (float): parameter between 0 and 1 to characterize proximity of lines inside a cluster.
            tree_depth (int): size max of the drain3 tree.
            filter_regexp (re.Pattern[str] | None): Regex pattern to filter log lines.
            time_pattern_regexp (re.Pattern[str] | None): Regex pattern to extract time from log lines.
            time_format (str | None): The format of the time extracted from log lines.
            time_min (datetime | None): Minimum time for log lines to be processed.
            time_max (datetime | None): Maximum time for log lines to be processed.
            unordered_time (bool): If True, log lines with time after time_max will not raise an exception and will be processed.
        """
        drain3_config = create_drain3_cfg(drain3_cfg_file, similarity_threshold, tree_depth)
        self.template_miner = TemplateMiner(config=drain3_config)
        self.filter_regexp = filter_regexp
        self.time_pattern_regexp = time_pattern_regexp
        self.time_format = time_format
        self.time_min = time_min
        self.time_max = time_max
        self.unordered_time = unordered_time
        self.cluster_char_sizes: Counter = Counter()
        # misc attr
        self.time_warn_displayed = False

    def get_cluster_data(self) -> list[ClusterResult]:
        """
        Get the clusters templates and their sizes

        Returns:
            list[ClusterResult]: all the clusters template
        """
        clusters_data = [
            ClusterResult(
                count=cluster.size,
                char_size=self.cluster_char_sizes[cluster.cluster_id],
                template=cluster.get_template(),
            )
            for cluster in self.template_miner.drain.clusters
        ]
        return clusters_data

    @staticmethod
    def diff_baseline(baseline_miner: "LogsMiner", miner: "LogsMiner") -> tuple[list[str], list[str], list[str]]:
        """
        Diff the clusters of the miner with the baseline miner and print the differences.

        Args:
            baseline_miner (TemplateMiner): The baseline template miner.
            miner (TemplateMiner): The template miner to compare with the baseline.
        Returns:
            tuple[set[str], set[str], set[str]]: A tuple containing three sets:
                - missing_from_baseline: Clusters present in the baseline but missing in the miner.
                - added_from_baseline: Clusters present in the miner but missing in the baseline.
                - common: Clusters present in both the baseline and the miner.
        """
        baseline_clusters = {cluster.get_template() for cluster in baseline_miner.template_miner.drain.clusters}
        clusters = {cluster.get_template() for cluster in miner.template_miner.drain.clusters}

        missing_vs_baseline = baseline_clusters - clusters
        added_vs_baseline = clusters - baseline_clusters
        common = baseline_clusters & clusters

        return sorted(missing_vs_baseline), sorted(added_vs_baseline), sorted(common)

    def get_total_nb_lines_clusters(self) -> int:
        """
        Get the total number of lines in all clusters.

        Returns:
            int: The total number of lines in all clusters.
        """
        return sum(cluster.size for cluster in self.template_miner.drain.clusters)

    def _extract_time_from_line(self, line: str) -> datetime | None:
        if not self.time_pattern_regexp or not self.time_format:
            return None
        match = self.time_pattern_regexp.match(line)
        if match:
            time_str = match.group(1)
            try:
                return datetime.strptime(time_str, self.time_format)
            except ValueError:
                if not self.time_warn_displayed:
                    logging.warning(
                        "Failed to parse time '%s' in line: %s. This warning is only displayed once", time_str, line
                    )
                    self.time_warn_displayed = True
                return None
        else:
            logging.warning("Time pattern not found in line: %s", line)
            return None

    def _check_line_out_time_range(
        self,
        line: str,
    ) -> bool:
        """
        Check if the time extracted from a log line is within the specified time range.

        Args:
            line (str): The log line to check.

        Returns:
            bool: True if the line should be skipped (time is out of range), False otherwise

        throws:
            ExceptionLineAfterMaxTime: If the line's time is after the specified maximum time
        """
        if self.time_pattern_regexp is None or self.time_format is None:
            return False
        line_time = self._extract_time_from_line(line)
        if line_time is None:
            return True
        if self.time_min and line_time < self.time_min:
            return True
        if self.time_max and line_time > self.time_max:
            if self.unordered_time:
                return True
            raise ExceptionLineAfterMaxTime()
        return False

    def add_log_lines_to_miner(
        self,
        line_generator: Iterator[str],
    ) -> int:
        """
        Add log lines from a generator to the template miner.

        Args:
            line_generator (Generator): A generator yielding log lines.

        Returns:
            int: The number of lines added and cluster sizes counter.
        """
        start_time = time.perf_counter()
        total_nb_lines = 0

        for line in line_generator:
            if not line:
                continue
            if self.filter_regexp and not self.filter_regexp.match(line):
                continue
            try:
                if self._check_line_out_time_range(line):
                    continue
            except ExceptionLineAfterMaxTime:
                logging.info(
                    "Reached a log line with time after the specified maximum time. Stopping processing further lines."
                )
                break
            total_nb_lines += 1
            result = self.template_miner.add_log_message(line)
            self.cluster_char_sizes[result["cluster_id"]] += len(line)
        end_time = time.perf_counter()
        elapsed_time = end_time - start_time
        logging.info(
            "Processed %d lines in %.2f seconds (%.2f lines/second).",
            total_nb_lines,
            elapsed_time,
            total_nb_lines / elapsed_time if elapsed_time > 0 else 0,
        )
        return total_nb_lines

__init__(drain3_cfg_file, similarity_threshold, tree_depth, *, filter_regexp=None, time_pattern_regexp=None, time_format=None, time_min=None, time_max=None, unordered_time=False)

Initialize the LogsMiner.

Parameters:

Name Type Description Default
drain3_cfg_file Path

Path to drain3 cfg file.

 required
similarity_threshold float

parameter between 0 and 1 to characterize proximity of lines inside a cluster.

 required
tree_depth int

size max of the drain3 tree.

 required
filter_regexp Pattern[str] | None

Regex pattern to filter log lines.

 None
time_pattern_regexp Pattern[str] | None

Regex pattern to extract time from log lines.

 None
time_format str | None

The format of the time extracted from log lines.

 None
time_min datetime | None

Minimum time for log lines to be processed.

 None
time_max datetime | None

Maximum time for log lines to be processed.

 None
unordered_time bool

If True, log lines with time after time_max will not raise an exception and will be processed.

 False
Source code in src/gool/logs_miner.py 
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
def __init__(
    self,
    drain3_cfg_file: Path,
    similarity_threshold: float | None,
    tree_depth: int | None,
    *,
    filter_regexp: re.Pattern[str] | None = None,
    time_pattern_regexp: re.Pattern[str] | None = None,
    time_format: str | None = None,
    time_min: datetime | None = None,
    time_max: datetime | None = None,
    unordered_time: bool = False,
):
    """
    Initialize the LogsMiner.

    Args:
        drain3_cfg_file (Path): Path to drain3 cfg file.
        similarity_threshold (float): parameter between 0 and 1 to characterize proximity of lines inside a cluster.
        tree_depth (int): size max of the drain3 tree.
        filter_regexp (re.Pattern[str] | None): Regex pattern to filter log lines.
        time_pattern_regexp (re.Pattern[str] | None): Regex pattern to extract time from log lines.
        time_format (str | None): The format of the time extracted from log lines.
        time_min (datetime | None): Minimum time for log lines to be processed.
        time_max (datetime | None): Maximum time for log lines to be processed.
        unordered_time (bool): If True, log lines with time after time_max will not raise an exception and will be processed.
    """
    drain3_config = create_drain3_cfg(drain3_cfg_file, similarity_threshold, tree_depth)
    self.template_miner = TemplateMiner(config=drain3_config)
    self.filter_regexp = filter_regexp
    self.time_pattern_regexp = time_pattern_regexp
    self.time_format = time_format
    self.time_min = time_min
    self.time_max = time_max
    self.unordered_time = unordered_time
    self.cluster_char_sizes: Counter = Counter()
    # misc attr
    self.time_warn_displayed = False

add_log_lines_to_miner(line_generator)

Add log lines from a generator to the template miner.

Parameters:

Name Type Description Default
line_generator Generator

A generator yielding log lines.

 required

Returns:

Name Type Description
int int

The number of lines added and cluster sizes counter.
Source code in src/gool/logs_miner.py 
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
def add_log_lines_to_miner(
    self,
    line_generator: Iterator[str],
) -> int:
    """
    Add log lines from a generator to the template miner.

    Args:
        line_generator (Generator): A generator yielding log lines.

    Returns:
        int: The number of lines added and cluster sizes counter.
    """
    start_time = time.perf_counter()
    total_nb_lines = 0

    for line in line_generator:
        if not line:
            continue
        if self.filter_regexp and not self.filter_regexp.match(line):
            continue
        try:
            if self._check_line_out_time_range(line):
                continue
        except ExceptionLineAfterMaxTime:
            logging.info(
                "Reached a log line with time after the specified maximum time. Stopping processing further lines."
            )
            break
        total_nb_lines += 1
        result = self.template_miner.add_log_message(line)
        self.cluster_char_sizes[result["cluster_id"]] += len(line)
    end_time = time.perf_counter()
    elapsed_time = end_time - start_time
    logging.info(
        "Processed %d lines in %.2f seconds (%.2f lines/second).",
        total_nb_lines,
        elapsed_time,
        total_nb_lines / elapsed_time if elapsed_time > 0 else 0,
    )
    return total_nb_lines

diff_baseline(baseline_miner, miner) staticmethod

Diff the clusters of the miner with the baseline miner and print the differences.

Parameters:

Name Type Description Default
baseline_miner TemplateMiner

The baseline template miner.

 required
miner TemplateMiner

The template miner to compare with the baseline.

 required

Returns: tuple[set[str], set[str], set[str]]: A tuple containing three sets: - missing_from_baseline: Clusters present in the baseline but missing in the miner. - added_from_baseline: Clusters present in the miner but missing in the baseline. - common: Clusters present in both the baseline and the miner.

Source code in src/gool/logs_miner.py 
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
@staticmethod
def diff_baseline(baseline_miner: "LogsMiner", miner: "LogsMiner") -> tuple[list[str], list[str], list[str]]:
    """
    Diff the clusters of the miner with the baseline miner and print the differences.

    Args:
        baseline_miner (TemplateMiner): The baseline template miner.
        miner (TemplateMiner): The template miner to compare with the baseline.
    Returns:
        tuple[set[str], set[str], set[str]]: A tuple containing three sets:
            - missing_from_baseline: Clusters present in the baseline but missing in the miner.
            - added_from_baseline: Clusters present in the miner but missing in the baseline.
            - common: Clusters present in both the baseline and the miner.
    """
    baseline_clusters = {cluster.get_template() for cluster in baseline_miner.template_miner.drain.clusters}
    clusters = {cluster.get_template() for cluster in miner.template_miner.drain.clusters}

    missing_vs_baseline = baseline_clusters - clusters
    added_vs_baseline = clusters - baseline_clusters
    common = baseline_clusters & clusters

    return sorted(missing_vs_baseline), sorted(added_vs_baseline), sorted(common)

get_cluster_data()

Get the clusters templates and their sizes

Returns:

Type Description
list[ClusterResult]

list[ClusterResult]: all the clusters template
Source code in src/gool/logs_miner.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
def get_cluster_data(self) -> list[ClusterResult]:
    """
    Get the clusters templates and their sizes

    Returns:
        list[ClusterResult]: all the clusters template
    """
    clusters_data = [
        ClusterResult(
            count=cluster.size,
            char_size=self.cluster_char_sizes[cluster.cluster_id],
            template=cluster.get_template(),
        )
        for cluster in self.template_miner.drain.clusters
    ]
    return clusters_data

get_total_nb_lines_clusters()

Get the total number of lines in all clusters.

Returns:

Name Type Description
int int

The total number of lines in all clusters.
Source code in src/gool/logs_miner.py 
127
128
129
130
131
132
133
134
def get_total_nb_lines_clusters(self) -> int:
    """
    Get the total number of lines in all clusters.

    Returns:
        int: The total number of lines in all clusters.
    """
    return sum(cluster.size for cluster in self.template_miner.drain.clusters)

create_drain3_cfg(cfg_file, similarity_threshold, tree_depth)

Create a default configuration for the drain3 template miner.

Returns:

Name Type Description
TemplateMinerConfig Any

A configuration object for the drain3 template miner.
Source code in src/gool/logs_miner.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
def create_drain3_cfg(cfg_file: Path, similarity_threshold: float | None, tree_depth: int | None) -> Any:
    """
    Create a default configuration for the drain3 template miner.

    Returns:
        TemplateMinerConfig: A configuration object for the drain3 template miner.
    """
    drain3_cfg = TemplateMinerConfig()
    if cfg_file.exists():
        logging.info("Loading configuration from %s", cfg_file)
        drain3_cfg.load(cfg_file)
    else:
        logging.warning(
            "Configuration file %s not found.",
            cfg_file,
        )

    if similarity_threshold:
        drain3_cfg.drain_sim_th = similarity_threshold
    if tree_depth:
        drain3_cfg.drain_depth = tree_depth
    return drain3_cfg