Skip to content

grpc_log_file_manager

drunc.grpc_testing_tools.grpc_log_file_manager

Classes

LogFileManager()

Manager for process log files with automatic creation and cleanup.

Creates unique log files in /tmp for each process and ensures proper cleanup after test completion to prevent file system clutter.

Initialise log file manager with empty state.

Source code in drunc/grpc_testing_tools/grpc_log_file_manager.py 
28
29
30
31
32
def __init__(self):
    """Initialise log file manager with empty state."""
    self.log_files = []
    self.file_positions = {}
    self.temp_dir = None
Functions
check_for_errors()

Check all log files for gRPC error patterns.

Returns:

Type Description

Error details if found, None otherwise
Source code in drunc/grpc_testing_tools/grpc_log_file_manager.py 
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
def check_for_errors(self):
    """
    Check all log files for gRPC error patterns.

    Returns:
        Error details if found, None otherwise
    """
    for log_file in self.get_all_log_files():
        if not os.path.exists(log_file):
            continue

        # Read new content since last check
        try:
            with open(log_file, "r", encoding="utf-8", errors="ignore") as f:
                f.seek(self.file_positions[log_file])
                new_content = f.read()
                self.file_positions[log_file] = f.tell()
        except (IOError, OSError):
            # File may not be ready yet, continue monitoring
            continue

        if new_content:
            # Check new content for error patterns
            error_lines = self._scan_content_for_errors(new_content)
            if error_lines:
                # Store error details and signal detection
                self.detected_error = {
                    "file": log_file,
                    "lines": error_lines,
                }
                return self.detected_error
    return None
cleanup()

Remove all log files and temporary directory.

Source code in drunc/grpc_testing_tools/grpc_log_file_manager.py 
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
def cleanup(self):
    """Remove all log files and temporary directory."""
    for log_file in self.log_files:
        try:
            if os.path.exists(log_file):
                os.remove(log_file)
        except Exception as e:
            print(f"Warning: Could not remove log file {log_file}: {e}")

    if self.temp_dir and os.path.exists(self.temp_dir):
        try:
            os.rmdir(self.temp_dir)
        except Exception as e:
            print(f"Warning: Could not remove temp directory {self.temp_dir}: {e}")

    self.log_files.clear()
    self.file_positions.clear()
    self.temp_dir = None
create_log_file(process_name)

Create a unique log file for a process.

Parameters:

Name Type Description Default
process_name str

Name of the process requiring logging

 required

Returns:

Type Description
str

Full path to the created log file
Source code in drunc/grpc_testing_tools/grpc_log_file_manager.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
def create_log_file(self, process_name: str) -> str:
    """
    Create a unique log file for a process.

    Args:
        process_name: Name of the process requiring logging

    Returns:
        Full path to the created log file
    """
    if not self.temp_dir:
        # Create temporary directory for all test log files
        self.temp_dir = tempfile.mkdtemp(prefix="grpc_test_logs_")

    # Generate unique log file path with timestamp
    timestamp = int(time.time())
    log_file = os.path.join(self.temp_dir, f"{process_name}_{timestamp}.log")

    # Create empty log file
    Path(log_file).touch()
    self.log_files.append(log_file)
    self.file_positions[log_file] = 0

    return log_file
get_all_log_files()

Retrieve list of all created log files.

Returns:

Type Description
List[str]

List of absolute paths to all created log files
Source code in drunc/grpc_testing_tools/grpc_log_file_manager.py 
59
60
61
62
63
64
65
66
def get_all_log_files(self) -> List[str]:
    """
    Retrieve list of all created log files.

    Returns:
        List of absolute paths to all created log files
    """
    return self.log_files.copy()