#!/usr/bin/env python3 # import sys import os from os import path import argparse import time import datetime import glob import re import random import stat import subprocess try: import yaml assert yaml.__version__ > '5.1' except: sys.exit( 'package "yaml" is required but not installed! Run\n' + 'python -m pip install yaml --user\n' + 'to install it.' ) if __name__ == '__main__': sys.path.append( os.path.abspath( os.path.join( os.path.dirname( path.realpath(__file__)), '..', '..' ))) import palm import palm.config import palm.system def ENV( directory ): resolved_name = os.environ['HOME'] directory = directory.replace('$HOME', resolved_name) return directory def palm_shell( command, capture_output=False, ignore_error=False ): return palm.system.shell( command, None, capture_output, ignore_error ) def palm_mkdir( directory ): if not path.isdir( directory ): try: os.makedirs(directory) print('') print('*** creating directory:' ) print(' ' + directory ) except: print('') print('+++ can\'t create directory:' ) print(' ' + directory ) raise def find_filename(filename, append = False, file_cycnum ='', file_ext = ''): maxcycle = 0 filelist = glob.glob(filename+'[.]*') if len(glob.glob(filename)) == 1: filelist.append(glob.glob(filename)[0]) for file in filelist: # filename without path (i.e. after the last "/") basename = path.basename(file) # check if there is an extension and remove it ext = path.splitext(basename)[1] ext=ext.replace('.','') if ext ==file_ext: basename = path.splitext(basename)[0] # check for an existing cycle number cycle = path.splitext(basename)[1] cycle=cycle.replace('.','') if len(re.findall('^[0-9]+$',cycle)) == 1: icycle = int(cycle) + 1 else: icycle = 1 if icycle > maxcycle: maxcycle = icycle # SET THE CYCLE NUMBER # IN CASE OF FILE-APPEND, IT MUST BE THE HIGHEST EXISTING CYCLE NUMBER if append == True: if maxcycle > 0: maxcycle = maxcycle -1 cycnum = maxcycle file_local = filename # ADD CYCLE NUMBER TO FILENAME # IN APPEND MODE, FILES KEEP THEIR CURRENT CYCLE NUMBER if append == False: # SET RUN NUMBER AS CYCLE NUMBER, IF THERE IS NOT A CONFLICT # WITH AN EXISTING CYCLE NUMBER if file_cycnum: if int(file_cycnum) >= cycnum: cycnum = int(file_cycnum) else: if int(file_cycnum) > 0: print("\n--- INFORMATIVE: The following file cannot get a unified cycle number", flush = True) file_local = file_local + '.'+str(cycnum).zfill(3) if file_ext.strip() != '': file_local = file_local + '.' + file_ext return file_local def main( prog=None, argv=None ): try: # sys.path.append( os.path.abspath( os.path.join( os.path.dirname(__file__), '..'))) parser = argparse.ArgumentParser(add_help=True) parser.add_argument('--pref', metavar='', dest='yamlfile', type=str, action='store', help="yaml file configuration file with information on data output" ) parser.add_argument('--log', metavar='', dest='job_protocol_file', type=str, help="full path to job protocol file on remote host") parser.add_argument('-d',metavar='', dest='localpath', type=str, help="local job catalog") parser.add_argument('-f',metavar='', dest='frequency', type=int, nargs='?', default=1800 , help="frequency of monitoring") parser.add_argument('-i',metavar='', dest='ip', type=str, help="ip of remote host") parser.add_argument('-k',metavar='', dest='ssh_key', type=str, nargs='?', default='' , help="ssh key for ssh key authentication") parser.add_argument('-p',metavar='', dest='port', type=str, nargs='?', default='', help="port for ssh/scp on remote host") parser.add_argument('-r',metavar='', dest='restart_file', type=str, help="full path to restart file on remote host") parser.add_argument('-u',metavar='', dest='user', type=str, help="username of remote host") args = parser.parse_args() ssh_key = ' -i '+args.ssh_key if args.ssh_key else '' ssh_portopt = ' -p '+args.port if args.port else '' scp_portopt = ' -P '+args.port if args.port else '' ssh_command = 'ssh -q ' + ssh_key + ssh_portopt + ' ' scp_command = 'scp -q ' + ssh_key + scp_portopt + ' ' address = args.user+'@'+args.ip print('\n*** start of monitoring file on remote host:\n '+ args.yamlfile, flush = True) myCR = ControlRemote(localpath = args.localpath, yamlfile = args.yamlfile, job_protocol_file = args.job_protocol_file, ssh_command = ssh_command, scp_command = scp_command, address = address) myCR.watch(frequency = args.frequency) myCR.transfer() myCR.restart(file = args.restart_file) print('\n*** watchdog_remote finished!') except Exception as exception: print('start of watchdog_remote failed!') print(str(exception)) sys.exit(1) class ControlRemote: def __init__(self, localpath, yamlfile, job_protocol_file, ssh_command, scp_command, address): self.localpath = localpath self.yamlfile = yamlfile self.local_yamlfile = path.join(self.localpath,path.basename(self.yamlfile)) self.job_protocol_file = job_protocol_file self.ssh_command = ssh_command self.scp_command = scp_command self.address = address #YAML FILE PROCESSING: def transfer(self): try: #COPY YAML FILE TO LOCAL JOB CATALOG palm_mkdir(os.path.expanduser(self.localpath)) command = self.scp_command + self.address + ':'+self.yamlfile + ' ' + self.localpath print(' ',command, flush = True) palm_shell(command) #COPY JOB PROTOCOL FILE TO LOCAL JOB CATALOG file_construct = path.splitext(path.basename(self.job_protocol_file)) filename = file_construct[0] if len(file_construct) ==2: id_number = file_construct[1].replace('.','_') else: id_number = '' command = self.scp_command + self.address + ':' + self.job_protocol_file + ' ' + path.join(self.localpath,filename) print(' ',command, flush = True) palm_shell(command) #FILE WITH COMMAND FOR CALLING WATHCODG_REMOTE cmd_file = path.join(self.localpath, 'cmd_watchdog_' + filename + id_number) #READ DATA FROM YAML FILE with open(self.local_yamlfile, "r") as stream: try: data = yaml.safe_load(stream) except yaml.YAMLError as exc: print( str(exc) ) print( '+++ reading yaml failed...' ) raise #TRANSFER OUTPUT DATA FROM REMOTE HOST TO LOCAL HOST file_transfer_problems = False for i in range(len(data)): data[i]['path_local'] = ENV(data[i]['path_local']) #FIND CYCLE NUMBER OF FILE ON LOCAL HOST if data[i]['catalog'] == 'file': filename = path.join(data[i]['path_local'],data[i]['file_local']) target = find_filename(filename, append = data[i]['append'], file_cycnum =data[i]['cyclenum'], file_ext = data[i]['ext']) else: # directory target = path.join(data[i]['path_local'],data[i]['file_local']) #SAVE REMOTE FILE OR DIRECTORY TO LOCAL HOST if data[i]['append'] == False: if data[i]['catalog'] == 'dir': scp_command = self.scp_command + ' -r ' exists = os.path.isdir(target) if exists: print('\n*** directory is already existing:', flush = True) print(' '+ target, flush = True) print(' Remote Directory: ',flush=True) print(' ' + path.join(data[i]['path_remote'],data[i]['file_remote']), flush = True) print(' is not stored at local host!', flush = True) file_transfer_problems = True continue #shutil.rmtree(target, ignore_errors=True) else: scp_command = self.scp_command palm_mkdir(os.path.expanduser(data[i]['path_local'])) command = scp_command + self.address +':'+ path.join(data[i]['path_remote'],data[i]['file_remote']) + ' ' + target+ '; echo $?' print('\n ',command, flush = True) status = palm_shell(command, capture_output = True) if status != '0': file_transfer_problems = True print('+++ file transfer problems', flush = True) print(' scp return code: ',status, flush = True) #APPEND FILE else: random_nr = random.randint(0,32767) target_tmp = path.join(data[i]['path_local'],data[i]['file_local']) + '_' + str(random_nr) command = scp_command + self.address +':'+ path.join(data[i]['path_remote'],data[i]['file_remote']) + ' ' + target_tmp+ '; echo $?' print('\n ',command, flush = True) status = palm_shell(command, capture_output = True) if status != '0': file_transfer_problems = True print('+++ file transfer problems', flush = True) print(' scp return code: ',status, flush = True) else: exists = path.isfile(target) if exists: palm_shell('cat ' + target_tmp + ' >> ' + target) print(' file successfully appended to', flush = True) print(' '+ target, flush = True) os.remove(target_tmp) else: print('+++ file-append failed', flush = True) file_transfer_problems = True #DELETE OUTPUT DATA AND CONFIGURATION FILE ON REMOTE HOST if not file_transfer_problems: print('\n*** Output files could be backed up to local computer.') print('\n*** Output files are now deleted from the remote host.\n') for i in range(len(data)): if data[i]['catalog'] == 'file': command = self.ssh_command + self.address + ' "rm ' + path.join(data[i]['path_remote'],data[i]['file_remote']) + '" ; exit $?\n' else: command = self.ssh_command + self.address + ' "rm -rf' + path.join(data[i]['path_remote'],data[i]['file_remote']) + '" ; exit $?\n' print(' ',command, flush = True) palm_shell(command) command = self.ssh_command + self.address + ' "rm ' + self.yamlfile + '" ; exit $?\n' print(' ',command, flush = True) palm_shell(command) print(' remove ', self.local_yamlfile) os.remove(self.local_yamlfile) print(' remove ', cmd_file) os.remove(cmd_file) else: print('\n+++ Output files could not be backed up to local computer. They are still on the remote host.') except Exception as exception: print('\n+++ processing of yaml file failed!') print(str(exception)) sys.exit(1) #CHECK IF A RESTART RUN HAS TO BE STARTED AND START IT def restart(self, file): try: #CHECK IF A RESTART FILE EXISTS ON REMOTE HOST print('\n check for restart file:', flush = True) print(' '+file, flush = True) command = self.ssh_command + self.address + ' "[[ -f ' + file + ' ]] && echo restart file available || echo restart file not found" 2>&1; exit $?\n' status = palm_shell(command,capture_output=True) if status == 'restart file available': print('\n\n*** restart file "'+ file + '"\n'\ + ' on remote host found!\n', flush = True) #COPY RESTART FILE TO LOCAL HOST local_file = path.join(self.localpath, path.basename(file)) command = self.scp_command + self.address + ':'+file + ' ' + local_file + '; echo $?' print(' ',command, flush = True) status = palm_shell(command, capture_output = True) if status != '0': print('+++ transfer problems with retart file', flush = True) print(' scp return code: ',status, flush = True) print('+++ palm restart run cannot be launched!') else: os.chmod(local_file, stat.S_IXUSR |stat.S_IRUSR ) print('*** start restart run via: ', flush = True) command = local_file + " > palmrun_restart.log 2>&1" print(' '+command, flush = True) subprocess.Popen([command], stdin = None, stdout = None, stderr = None, close_fds = True, shell = True) time.sleep(30) print('*** script to start the restart run is now deleted:', flush = True) #DELETE FILE ON REMOTE HOST command = self.ssh_command + self.address + ' "rm -rf ' + file + '" ; exit $?\n' print(' ',command, flush = True) palm_shell(command) #DELETE LOCAL_FILE ON LOCAL HOST print(' remove ', local_file) os.remove(local_file) else: print(' restart file not found: palm finished!', flush = True) except Exception as exception: print('\n+++ check for restart run / start a restart run failed!') print(str(exception)) sys.exit(1) #MONITORING YAML FILE ON REMOTE HOST def watch(self, frequency): try: while True: calltime = datetime.datetime.now( datetime.timezone.utc ).strftime('%a %b %d %H:%M:%S UTC %Y') print('>>> check:', calltime, flush = True) #check if fike is already created on remote host command = self.ssh_command + self.address + ' "[[ -f ' + self.yamlfile + ' ]] && echo jobfile available || echo jobfile not found" 2>&1; exit $?\n' status = palm_shell(command,capture_output=True) if status == 'jobfile available': print('\n\n*** jobfile "'+ path.basename(self.yamlfile) + '"\n'\ + ' on remote host found!\n', flush = True) break time.sleep(frequency) except KeyboardInterrupt: print("\n +++ watchdog interrupted") sys.exit(1) if __name__ == "__main__": sys.exit( main() )