pipe a file to a tcp socket and send to other host.

file_pipe_socket.py

def __runner(server_id, host, port, filename):
    LOGGER.debug('start %s', os.getpid())
    LOGGER.debug('start ppid %s', os.getppid())
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.settimeout(1.0)
    server_address = (host, port)
    try:
        sock.connect(server_address)
    except socket.error as err:
        return {'retcode': err.errno, 'stdout': '', 'stderr': str(err)}

    sockfile = sock.makefile('w')
    sock.sendall(server_id+'\n')
    sock.sendall(filename+'\n')
    subp = subprocess.Popen(['tail', '-f', filename], stdout=sockfile, stderr=sockfile)
    time.sleep(0.01)
    subp.poll()

    if subp.returncode is not None:
        err = subp.stderr.read()
        LOGGER.info('tail %s %s %s', subp.returncode, err, port)
        return {'retcode': subp.returncode, 'stdout': subp.stdout.read(), 'stderr': subp.stderr.read()}


    # init state ok
    # daemon -> return ok for salt master
    #        -> fork watchdog
    try:
        watchdog = os.fork()
    except OSError:
        LOGGER.exception(err)
        return {'retcode': err.errno, 'stdout': '', 'stderr': err.strerror}

    if watchdog:
        return {'retcode': 0, 'stdout': 'open file ok\nconnect server ok\n', 'stderr': ''}

    stderr = ''
    interval = 1.0
    LOGGER.debug('start interval loop')
    while interval <= TIMEOUT:
        try:
            info = sockfile.readline()
            LOGGER.debug(info)
            if info == '':
                break
            interval = 1.0
            time.sleep(1.0)
        except socket.timeout:
            interval += 1.0
            continue
        except Exception as err:
            stderr = 'unknown error {0}\n'.format(err)
            break

    # clean up and exit
    ret = stderr if stderr else 'timeout'
    subp.kill()
    sockfile.write(ret)
    sockfile.close()
    sock.close()
    exit(0)

def __new_daemon(server_id, host, port, filename):

    try:
        # [P]salt-minion.executor -> exit(0)
        #                         -> daemon fork
        pid = os.fork()
    except OSError as err:
        LOGGER.exception(err)
        return {'retcode': err.errno, 'stdout': '', 'stderr': err.strerror}

    if pid > 0:
        sys.exit(0)

    os.setsid()
    os.umask(0)

    try:
        # [P]daemon -> exit(0)
        #           -> runner
        runner_pid = os.fork()
    except OSError as err:
        LOGGER.exception(err)
        return {'retcode': err.errno, 'stdout': '', 'stderr': err.strerror}

    if runner_pid > 0:
        sys.exit(0)

    return __runner(server_id, host, port, filename)


def tail(host, port, filename):
    LOGGER.info('tail -f %s %s %s', filename, host, port)
    if '..' in filename:
        return {'retcode': 400, 'stdout': '', 'stderr': 'filename illegal'}

    if filename[-4:] != '.log':
        return {'retcode': 400, 'stdout': '', 'stderr': 'filename illegal'}

    if filename[:10] != '/*********':
        return {'retcode': 400, 'stdout': '', 'stderr': 'filename illegal'}
    server_id = __grains__.get('id', 'unknown')
    ret = __new_daemon(server_id, host, port, filename)
    return ret