Skip to content

grpc_log_util

drunc.grpc_testing_tools.grpc_log_util

Functions

stderr_observer(log_file_name)

Redirect stderr to a log file for process output capture.

Parameters:

Name Type Description Default
log_file_name

Path to log file for stderr redirection

 required
Source code in drunc/grpc_testing_tools/grpc_log_util.py 
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
def stderr_observer(log_file_name):
    """
    Redirect stderr to a log file for process output capture.

    Args:
        log_file_name: Path to log file for stderr redirection
    """
    r, w = os.pipe()
    stderr_fd = sys.stderr.fileno()
    os.dup2(w, stderr_fd)
    os.close(w)

    def reader():
        log_handle = open(log_file_name, "a", buffering=1)
        with os.fdopen(r) as pipe:
            for line in pipe:
                log_handle.write(f"{line.strip()}\n")
                log_handle.flush()
        log_handle.close()

    reader_thread = threading.Thread(target=reader, daemon=True)
    reader_thread.start()

stdout_observer(log_file_name)

Redirect stdout to a log file for process output capture.

Parameters:

Name Type Description Default
log_file_name

Path to log file for stdout redirection

 required
Source code in drunc/grpc_testing_tools/grpc_log_util.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
def stdout_observer(log_file_name):
    """
    Redirect stdout to a log file for process output capture.

    Args:
        log_file_name: Path to log file for stdout redirection
    """
    r, w = os.pipe()
    stdout_fd = sys.stdout.fileno()
    os.dup2(w, stdout_fd)
    os.close(w)

    def reader():
        log_handle = open(log_file_name, "a", buffering=1)
        with os.fdopen(r) as pipe:
            for line in pipe:
                log_handle.write(f"{line.strip()}\n")
                log_handle.flush()
        log_handle.close()

    reader_thread = threading.Thread(target=reader, daemon=True)
    reader_thread.start()