I've created an on-demand class that can be initiated to collected any new data written to file from start_recording call until stop_recording which close the connection and retrieve the collected data so far.

It uses in test cases for obtaining relevant logs during the time where a certain operation was performed in order to verify its success.

I'm currently using this class for tests in remote machine, but would be happy to hear for any idea for improvements, correctness, etc.

import time

import paramiko
import select

from virtual_machine import utils
from multiprocessing import Queue
import multiprocess

class background():
    def __init__(self, config_file):
        self.q = Queue(1000)
        #this methods implemented elsewhere and read a config file into dictionary
        self.config = utils.get_params(config_file)

    def linesplit(socket):
        buffer_string = socket.recv(4048).decode('utf-8')
        done = False
        while not done:
            if '\n' in buffer_string:
                (line, buffer_string) = buffer_string.split('\n', 1)
                yield line + "\n"
                more = socket.recv(4048)
                if not more:
                    done = True
                    buffer_string = buffer_string + more.decode('utf-8')
        if buffer_string:
            yield buffer_string

    def do_tail(self, log_file):
        data = ""
        client = paramiko.SSHClient()
        from os.path import expanduser
        home = expanduser("~")

        grep_pattern = "grep_filter"
        remote_command = 'tail -0 -f {} '.format(log_file)
        transport = client.get_transport()
        channel = transport.open_session()

        while 1:
                rl, _, _ = select.select([channel], [], [], 0.0)
                if len(rl) > 0:
                    print("ready to read")
                    for line in background.linesplit(channel):

            except (KeyboardInterrupt, SystemExit):
                print('got ctrl+c')


        return data

    def start_recording(self, log_file):
        q = Queue(100)
        self.p = multiprocess.Process(target=self.do_tail, args=(log_file,), daemon=True)

    def stop_recording(self):

        data = ""
        while not self.q.empty():
            data += self.q.get()

        return data

if __name__ == '__main__':
    x = background("config.conf")
    # doing some operation 
    data = x.stop_recording()

1 Answer 1


Your linesplit repeats a little bit of code; you could phrase it as

while True:
    buffer_bytes = socket.recv(4096)
    if len(buffer_bytes) == 0:
    buffer_string = buffer_bytes.decode('utf-8')
    *lines, buffer_string = buffer_string.split('\n')
    yield from lines

'tail -0 -f {} '.format(log_file) could be simply 'tail -0 -f ' + log_file.

background should be Background.

Consider adding some PEP484 type hints.

rl, _, _ = select ...

can be

rl, *_ = select ...

The code between paramiko.SSHClient() and client.close() has no close-guarantee protection. There are various options, including a try/finally, or making your current class (or a factored-out tail-specific class) a context manager.

  • \$\begingroup\$ Thanks for your comments. Btw, I've encountered where the log recording was lagging behind, and where it did, the interesting part already finished. I wonder if there's any kind of system wide lock that can block the start_recording from returning, until the subprocess reach the ready to read line. Can you recommend anything ? thanks ! \$\endgroup\$
    – Zohar81
    Jul 31, 2021 at 15:44

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Not the answer you're looking for? Browse other questions tagged or ask your own question.