February 2018 – Jason R. Ralph

This is a neat way to generate file names in a directory that match a specific pattern, I use this to generate a list of files exported out of hive to load into S3.

def find_files(directory, pattern):
    for root, dirs, files in os.walk(directory):
        for basename in sorted(files):
            if fnmatch.fnmatch(basename, pattern):
                filename = os.path.join(root, basename)
                yield filename

def find_files(directory, pattern):

for root, dirs, files in os.walk(directory):

for basename in sorted(files):

if fnmatch.fnmatch(basename, pattern):

filename = os.path.join(root, basename)

yield filename

local_dir = '/mnt/share/etl/date/'
for files in find_files(local_dir,'*.gz'):
    key = files[1:]
    try:
        awss3.upload(key,files)
        log_msg = ('uploading file: [{0}] to S3').format(files)
        log.write(log_msg)
    except Exception as e:
        log_msg = ('ERROR: {0} uploading file: [{0}] to S3').format(e,files)
        log.write(log_msg, 'error')

local_dir = '/mnt/share/etl/date/'

for files in find_files(local_dir,'*.gz'):

key = files[1:]

try:

awss3.upload(key,files)

log_msg = ('uploading file: [{0}] to S3').format(files)

log.write(log_msg)

except Exception as e:

log_msg = ('ERROR: {0} uploading file: [{0}] to S3').format(e,files)

log.write(log_msg, 'error')

I recently came across a tough to debug issue where I was calling a shell script from python using the subprocess module, this shell script called rsync, no matter what I would always run into a timeout situation. I fired up strace and noticed that the process was in a timeout state.

select(4, NULL, [3], [3], {60, 0}) = 0 (Timeout)

I looked at the subprocess documentation and apparently using pipes will fill the system pipe buffer.

Warning

This will deadlock when using stdout=PIPE and/or stderr=PIPE and the child process generates enough output to a pipe such that it blocks waiting for the OS pipe buffer to accept more data. Use communicate() to avoid that.

I was baffled, I finally took the approach to eliminate stderr and stdout and just check the return status of the command using run(). Here is what I finally came up with, and all was well.

stdbuf -oL -e0 /usr/local/bin/rsync --outbuf=N -avz

1	stdbuf -oL -e0 /usr/local/bin/rsync --outbuf=N -avz

def execute_jobs(cmd):
     try:
         logger.info('Start Command: [%s]' % cmd)
         subprocess.run(shlex.split(cmd), check=True)
         logger.info('Command Success: [%s]' % cmd)
     except subprocess.CalledProcessError as e:
         logger.critical('[%s] FATAL: Command failed with error [%s]' % (cmd,e))

def execute_jobs(cmd):

try:

logger.info('Start Command: [%s]' % cmd)

subprocess.run(shlex.split(cmd), check=True)

logger.info('Command Success: [%s]' % cmd)

except subprocess.CalledProcessError as e:

logger.critical('[%s] FATAL: Command failed with error [%s]' % (cmd,e))

Hope you find this and it helps you.

Jason R. Ralph

Linux All Day Everyday

Month: February 2018

Python Generator Find Files With Wildcard

Python3 Subprocess and Rsync Deadlock Strace Timeout