Log Clustering

This script computes clusters of similar log lines from the provided log files. It uses the drain3 library to extract templates from log lines. It can filter log lines based on a regex pattern and display the clusters.

compute_margin_for_display(max_number)

Compute the margin for displaying numbers.

Parameters:

Name	Type	Description	Default
max_number	int	The maximum number to display.	required

Source code in src/gool/log_clustering.py

def compute_margin_for_display(max_number: int) -> int:
    """
    Compute the margin for displaying numbers.

    Args:
        max_number (int): The maximum number to display.
    """
    if max_number <= 0:
        return 1
    b10 = log10(max_number)
    margin = ceil(log10(max_number)) if b10 != int(b10) else int(b10 + 1)
    return margin

create_config_file_if_needed(home_config)

Create default config file if does not exists.

Parameters:

Name	Type	Description	Default
home_config	Path	path of the default config file	required

Source code in src/gool/log_clustering.py

def create_config_file_if_needed(home_config: Path) -> None:
    """
    Create default config file if does not exists.

    Args:
        home_config (Path): path of the default config file
    """
    if not home_config.exists():
        logging.info(f"Deploying default config to {home_config}")
        home_config.parent.mkdir(parents=True, exist_ok=True)
        source = resources.files("gool").joinpath("drain3.ini")

        with resources.as_file(source) as default_path:
            shutil.copy(default_path, home_config)

create_file_line_generators(logfile_paths, progress)

Create progress tasks for all files and return a chained generator yielding lines from all files.

Parameters:

Name	Type	Description	Default
logfile_paths	tuple[Path, ...]	Paths to the log files.	required
progress	Progress	The progress instance to track file processing.	required

Returns:

Name	Type	Description
Generator	chain	A single generator yielding lines from all files.

Source code in src/gool/log_clustering.py

def create_file_line_generators(
    logfile_paths: tuple[pathlib.Path, ...],
    progress: Progress,
) -> itertools.chain:
    """
    Create progress tasks for all files and return a chained generator yielding lines from all files.

    Args:
        logfile_paths (tuple[pathlib.Path, ...]): Paths to the log files.
        progress (Progress): The progress instance to track file processing.

    Returns:
        Generator: A single generator yielding lines from all files.
    """
    generators = []
    for logfile_path in logfile_paths:
        sample_nb_lines = 20000
        number_of_lines = estimate_lines(logfile_path, sample_nb_lines)
        task_id = progress.add_task(f"{pathlib.Path(logfile_path).name}", total=number_of_lines)

        def line_generator(path: pathlib.Path, tid: TaskID, nb_lines: int) -> Generator[str, None, None]:
            """Generator that yields lines from a file and updates progress."""
            with open(path, encoding="utf-8", errors="surrogateescape") as f:
                for line in f:
                    progress.update(tid, advance=1)
                    yield line
            progress.update(tid, completed=nb_lines)
            progress.stop_task(tid)

        generators.append(line_generator(logfile_path, task_id, number_of_lines))

    # Chain all generators into one
    return itertools.chain(*generators)

display_clusters(template_miner, order_by='count', raw=False, table_title='Log Clusters')

Display all clusters in a table with 3 columns: Count - Char Size (KB) - Template.

Parameters:

Name	Type	Description	Default
template_miner	LogsMiner	The logs miner which has run.	required
order_by	str	How to order clusters: "count", "size", or "template". Defaults to "count".	'count'
raw	bool	display the data without rich if true.	False
table_title	str	title of the table displayed on top.	'Log Clusters'

Source code in src/gool/log_clustering.py

def display_clusters(
    template_miner: LogsMiner, order_by: str = "count", raw: bool = False, table_title: str = "Log Clusters"
) -> None:
    """
    Display all clusters in a table with 3 columns: Count - Char Size (KB) - Template.

    Args:
        template_miner (LogsMiner): The logs miner which has run.
        order_by (str): How to order clusters: "count", "size", or "template". Defaults to "count".
        raw (bool): display the data without rich if true.
        table_title (str): title of the table displayed on top.
    """
    clusters_data = template_miner.get_cluster_data()

    if not clusters_data:
        return

    if order_by == "size":
        clusters_data.sort(key=lambda x: x.char_size, reverse=True)
    elif order_by == "template":
        clusters_data.sort(key=lambda x: x.template)
    else:
        clusters_data.sort(key=lambda x: x.count, reverse=True)

    if raw:
        console.print(table_title)
        count_margin = compute_margin_for_display(max(cluster.count for cluster in clusters_data))
        size_margin = compute_margin_for_display(max(cluster.char_size for cluster in clusters_data) // KB_FACTOR)
        for cluster in clusters_data:
            pattern = surrogate_non_printable(cluster.template)
            count_str = f"{cluster.count}"
            size_kb = cluster.char_size // KB_FACTOR
            size_str = f"{size_kb}"
            console.print(
                f"{count_str:>{count_margin}} - {size_str:>{size_margin}} - {pattern}",
                soft_wrap=True,
                markup=False,
            )
    else:
        table = Table(title=table_title, highlight=True)
        table.add_column("Count", justify="right", style="cyan", no_wrap=True)
        table.add_column("Char Size (KB)", justify="right", style="magenta", no_wrap=True)
        table.add_column("Template", justify="left")

        for cluster in clusters_data:
            pattern = surrogate_non_printable(cluster.template)
            count_str = f"{cluster.count:,}".replace(",", " ")
            size_kb = cluster.char_size // KB_FACTOR
            size_str = f"{size_kb:,}".replace(",", " ")
            table.add_row(count_str, size_str, pattern)

        # Print the table
        console.print(table, markup=False)

display_diff_baseline(missing, added, common, raw=False, display_common=False)

Display the difference between baseline clusters and current clusters.

Parameters:

Name	Type	Description	Default
missing	list[str]	Clusters present in baseline but missing in current.	required
added	list[str]	Clusters present in current but missing in baseline.	required
common	list[str]	Clusters present in both baseline and current.	required
raw	bool	If True, display in plain text format for bash processing. Defaults to False.	False
display_common	bool	If True, also display common clusters. Defaults to False.	False

Source code in src/gool/log_clustering.py

def display_diff_baseline(
    missing: list[str], added: list[str], common: list[str], raw: bool = False, display_common: bool = False
) -> None:
    """
    Display the difference between baseline clusters and current clusters.

    Args:
        missing (list[str]): Clusters present in baseline but missing in current.
        added (list[str]): Clusters present in current but missing in baseline.
        common (list[str]): Clusters present in both baseline and current.
        raw (bool): If True, display in plain text format for bash processing. Defaults to False.
        display_common (bool): If True, also display common clusters. Defaults to False.
    """

    def display_data_raw(list_data: list[str], header: str) -> None:
        console.print(f"{header}:")
        for item in list_data:
            console.print(f"{item}", soft_wrap=True, markup=False)
        print()

    def display_data_rich(list_data: list[str], header: str) -> None:
        table = Table(title=header, highlight=True)
        min_width_empty_tab = 100
        table.min_width = min_width_empty_tab
        table.add_column("Template", justify="left")
        for item in list_data:
            pattern = surrogate_non_printable(item)
            table.add_row(pattern)
        console.print(table, markup=False)
        print()

    def display_data(raw: bool, list_data: list[str], header: str) -> None:
        if raw:
            display_data_raw(list_data, header)
        else:
            display_data_rich(list_data, header)

    display_data(raw, missing, "Missing /baseline")
    display_data(raw, added, "Added /baseline")
    if display_common:
        display_data(raw, common, "Common with baseline")

estimate_lines(path, nb_sample_lines=1000)

Estimate total lines based on file size and sample average. Args: path (pathlib.Path): Path to the log file. nb_sample_lines (int): Number of lines to sample for average calculation.

Source code in src/gool/log_clustering.py

def estimate_lines(path: pathlib.Path, nb_sample_lines: int = 1000) -> int:
    """
    Estimate total lines based on file size and sample average.
    Args:
        path (pathlib.Path): Path to the log file.
        nb_sample_lines (int): Number of lines to sample for average calculation.
    """
    file_size = path.stat().st_size
    if file_size == 0:
        return 0
    # Sample first 'sample_lines' to get avg bytes per line
    avg_bytes_per_line = 0
    with open(path, "rb") as f:  # Use binary for accurate byte counting
        for i, line in enumerate(f):
            if i >= nb_sample_lines:
                break
            avg_bytes_per_line += len(line)
    if nb_sample_lines > 0:
        avg_bytes_per_line //= nb_sample_lines
    # Estimate total lines
    estimated = int(file_size / avg_bytes_per_line) if avg_bytes_per_line > 0 else 0
    return max(estimated, 1)

main(args)

Extract the cluster templates from the provided log files.

Parameters:

Name	Type	Description	Default
args	Arguments	paths and options for the log extractor	required

Returns:

Name	Type	Description
int	int	0 if everything went well

Source code in src/gool/log_clustering.py

def main(args: Arguments) -> int:
    """
    Extract the cluster templates from the provided log files.

    Args:
        args (Arguments): paths and options for the log extractor

    Returns:
        int: 0 if everything went well
    """
    if args.cfg_file == HOME_CFG_FILE:
        create_config_file_if_needed(HOME_CFG_FILE)
    drain3_cfg_file = args.cfg_file
    similarity_threshold = args.similarity_threshold
    tree_depth = args.tree_depth
    time_pattern = re.compile(args.time_pattern) if args.time_pattern else None
    filter_regexp = re.compile(args.filter) if args.filter else None
    time_format = args.time_format
    time_min = datetime.strptime(args.time_min, time_format) if args.time_min else None
    time_max = datetime.strptime(args.time_max, time_format) if args.time_max else None

    logging.info("Running clustering.")
    runner, total_nb_lines = _create_and_run_miner(
        drain3_cfg_file,
        similarity_threshold,
        tree_depth,
        args.logfile_paths,
        filter_regexp=filter_regexp,
        time_pattern_regexp=time_pattern,
        time_format=time_format,
        time_min=time_min,
        time_max=time_max,
        unordered_time=args.unordered_time,
    )

    if args.baseline:
        base_filter_regexp = re.compile(args.base_filter) if args.base_filter else filter_regexp
        base_time_min = datetime.strptime(args.base_time_min, time_format) if args.base_time_min else time_min
        base_time_max = datetime.strptime(args.base_time_max, time_format) if args.base_time_max else time_max
        logging.info("Running baseline clustering.")
        bl_runner, bl_total_nb_lines = _create_and_run_miner(
            drain3_cfg_file,
            similarity_threshold,
            tree_depth,
            args.baseline,
            filter_regexp=base_filter_regexp,
            time_pattern_regexp=time_pattern,
            time_format=time_format,
            time_min=base_time_min,
            time_max=base_time_max,
            unordered_time=args.unordered_time,
        )
        sanity_check(bl_runner.get_total_nb_lines_clusters(), bl_total_nb_lines)
        missing, added, common = LogsMiner.diff_baseline(bl_runner, runner)
        display_diff_baseline(
            missing,
            added,
            common,
            raw=args.raw,
            display_common=args.display_common,
        )
        if args.display_common:
            display_results(bl_runner, args.raw, args.lex_order, args.size_order, "Baseline Log Clusters")

    display_results(runner, args.raw, args.lex_order, args.size_order, "Log Clusters")
    total_nb_lines_clusters = runner.get_total_nb_lines_clusters()
    return sanity_check(total_nb_lines_clusters, total_nb_lines)

main_cli()

Main entry point for the CLI.

Returns:

Name	Type	Description
int	int	Exit code.

Source code in src/gool/log_clustering.py

def main_cli() -> int:
    """
    Main entry point for the CLI.

    Returns:
        int: Exit code.
    """
    try:
        cfg = tyro.cli(Arguments)
        if cfg.version:
            print(__version__)
            return ErrorCode.NO_ERROR.value
        return main(cfg)
    except KeyboardInterrupt:
        error_console.print("\n[red]Process interrupted by user[/red]")
        return ErrorCode.IO_ERROR.value

sanity_check(total_nb_lines_clusters, total_nb_lines)

Check if the total number of lines in clusters matches the processed lines.

Source code in src/gool/log_clustering.py

def sanity_check(total_nb_lines_clusters: int, total_nb_lines: int) -> int:
    """Check if the total number of lines in clusters matches the processed lines."""
    result = 0
    if total_nb_lines_clusters != total_nb_lines:
        logging.error(
            "The number of lines in the clusters (%d) does "
            "not match the total number of lines processed (%d)."
            "Maybe you should increase [DRAIN]/max_clusters parameter.",
            total_nb_lines_clusters,
            total_nb_lines,
        )
        result = 1
    return result

surrogate_non_printable(s)

Surrogate non-printable characters from a string.

Parameters:

Name	Type	Description	Default
s	str	The input string.	required

Returns: str: The cleaned string.

Source code in src/gool/log_clustering.py

def surrogate_non_printable(s: str) -> str:
    """
    Surrogate non-printable characters from a string.

    Args:
        s (str): The input string.
    Returns:
        str: The cleaned string.
    """
    return s.encode("utf-8", errors="surrogateescape").decode("utf-8")