EIC code displayed by LXR master/​pilot2/​pilot/​control/​job.py	Source navigation Diff markup Identifier search General search
	Version: master test Revert   Change

File indexing completed on 2026-04-10 08:39:15

 #!/usr/bin/env python
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
 # You may obtain a copy of the License at
 # http://www.apache.org/licenses/LICENSE-2.0
 #
 # Authors:
 # - Mario Lassnig, mario.lassnig@cern.ch, 2016-2017
 # - Daniel Drizhuk, d.drizhuk@gmail.com, 2017
 # - Paul Nilsson, paul.nilsson@cern.ch, 2017-2021
 # - Wen Guan, wen.guan@cern.ch, 2018
 
 from __future__ import print_function  # Python 2
 
 import os
 import time
 import hashlib
 import random
 import socket
 import logging
 
 try:
     import Queue as queue  # noqa: N813
 #except ModuleNotFoundError:  # Python 3
 except Exception:
     import queue  # Python 3
 
 from json import dumps  #, loads
 from re import findall
 from glob import glob
 
 from pilot.common.errorcodes import ErrorCodes
 from pilot.common.exception import ExcThread, PilotException  #, JobAlreadyRunning
 from pilot.info import infosys, JobData, InfoService, JobInfoProvider
 from pilot.util import https
 from pilot.util.auxiliary import get_batchsystem_jobid, get_job_scheduler_id, get_pilot_id, \
     set_pilot_state, get_pilot_state, check_for_final_server_update, pilot_version_banner, is_virtual_machine, \
     is_python3, show_memory_usage, has_instruction_sets, locate_core_file, get_display_info
 from pilot.util.config import config
 from pilot.util.common import should_abort, was_pilot_killed
 from pilot.util.constants import PILOT_MULTIJOB_START_TIME, PILOT_PRE_GETJOB, PILOT_POST_GETJOB, PILOT_KILL_SIGNAL, LOG_TRANSFER_NOT_DONE, \
     LOG_TRANSFER_IN_PROGRESS, LOG_TRANSFER_DONE, LOG_TRANSFER_FAILED, SERVER_UPDATE_TROUBLE, SERVER_UPDATE_FINAL, \
     SERVER_UPDATE_UPDATING, SERVER_UPDATE_NOT_DONE
 from pilot.util.container import execute
 from pilot.util.filehandling import find_text_files, tail, is_json, copy, remove, write_json, establish_logging, write_file, \
     create_symlink
 from pilot.util.harvester import request_new_jobs, remove_job_request_file, parse_job_definition_file, \
     is_harvester_mode, get_worker_attributes_file, publish_job_report, publish_work_report, get_event_status_file, \
     publish_stageout_files
 from pilot.util.jobmetrics import get_job_metrics
 from pilot.util.math import mean
 from pilot.util.middleware import containerise_general_command
 from pilot.util.monitoring import job_monitor_tasks, check_local_space
 from pilot.util.monitoringtime import MonitoringTime
 from pilot.util.processes import cleanup, threads_aborted, kill_process, kill_processes
 from pilot.util.proxy import get_distinguished_name
 from pilot.util.queuehandling import scan_for_jobs, put_in_queue, queue_report, purge_queue
 from pilot.util.timing import add_to_pilot_timing, timing_report, get_postgetjob_time, get_time_since, time_stamp
 from pilot.util.workernode import get_disk_space, collect_workernode_info, get_node_name, get_cpu_model
 
 logger = logging.getLogger(__name__)
 errors = ErrorCodes()
 
 
 def control(queues, traces, args):
     """
     Main function of job control.
 
     :param queues: internal queues for job handling.
     :param traces: tuple containing internal pilot states.
     :param args: Pilot arguments (e.g. containing queue name, queuedata dictionary, etc).
     :return:
     """
 
     targets = {'validate': validate, 'retrieve': retrieve, 'create_data_payload': create_data_payload,
                'queue_monitor': queue_monitor, 'job_monitor': job_monitor, 'fast_job_monitor': fast_job_monitor}
     threads = [ExcThread(bucket=queue.Queue(), target=target, kwargs={'queues': queues, 'traces': traces, 'args': args},
                          name=name) for name, target in list(targets.items())]  # Python 2/3
 
     [thread.start() for thread in threads]
 
     # if an exception is thrown, the graceful_stop will be set by the ExcThread class run() function
     while not args.graceful_stop.is_set():
         for thread in threads:
             bucket = thread.get_bucket()
             try:
                 exc = bucket.get(block=False)
             except queue.Empty:
                 pass
             else:
                 exc_type, exc_obj, exc_trace = exc
                 logger.warning("thread \'%s\' received an exception from bucket: %s", thread.name, exc_obj)
 
                 # deal with the exception
                 # ..
 
             thread.join(0.1)
             time.sleep(0.1)
 
         time.sleep(0.5)
 
     logger.debug('job control ending since graceful_stop has been set')
     if args.abort_job.is_set():
         if traces.pilot['command'] == 'aborting':
             logger.warning('jobs are aborting')
         elif traces.pilot['command'] == 'abort':
             logger.warning('job control detected a set abort_job (due to a kill signal)')
             traces.pilot['command'] = 'aborting'
 
             # find all running jobs and stop them, find all jobs in queues relevant to this module
             #abort_jobs_in_queues(queues, args.signal)
 
     # proceed to set the job_aborted flag?
     if threads_aborted():
         logger.debug('will proceed to set job_aborted')
         args.job_aborted.set()
     else:
         logger.debug('will not set job_aborted yet')
 
     logger.debug('[job] control thread has finished')
     # test kill signal during end of generic workflow
     #import signal
     #os.kill(os.getpid(), signal.SIGBUS)
 
 
 def _validate_job(job):
     """
     Verify job parameters for specific problems.
 
     :param job: job object.
     :return: Boolean.
     """
 
     pilot_user = os.environ.get('PILOT_USER', 'generic').lower()
     user = __import__('pilot.user.%s.common' % pilot_user, globals(), locals(), [pilot_user], 0)  # Python 2/3
     container = __import__('pilot.user.%s.container' % pilot_user, globals(), locals(), [user], 0)  # Python 2/3
 
     # should a container be used for the payload?
     try:
         kwargs = {'job': job}
         job.usecontainer = container.do_use_container(**kwargs)
     except Exception as error:
         logger.warning('exception caught: %s', error)
 
     return True if user.verify_job(job) else False
 
 
 def verify_error_code(job):
     """
     Make sure an error code is properly set.
     This makes sure that job.piloterrorcode is always set for a failed/holding job, that not only
     job.piloterrorcodes are set but not job.piloterrorcode. This function also negates the sign of the error code
     and sets job state 'holding' (instead of 'failed') if the error is found to be recoverable by a later job (user
     jobs only).
 
     :param job: job object.
     :return:
     """
 
     if job.piloterrorcode == 0 and len(job.piloterrorcodes) > 0:
         logger.warning('piloterrorcode set to first piloterrorcodes list entry: %s', str(job.piloterrorcodes))
         job.piloterrorcode = job.piloterrorcodes[0]
 
     if job.piloterrorcode != 0 and job.is_analysis():
         if errors.is_recoverable(code=job.piloterrorcode):
             job.piloterrorcode = -abs(job.piloterrorcode)
             job.state = 'failed'
             logger.info('failed user job is recoverable (error code=%s)', job.piloterrorcode)
         else:
             logger.info('failed user job is not recoverable')
     else:
         logger.info('verified error code')
 
 
 def get_proper_state(job, state):
     """
     Return a proper job state to send to server.
     This function should only return 'starting', 'running', 'finished', 'holding' or 'failed'.
     If the internal job.serverstate is not yet set, it means it is the first server update, ie 'starting' should be
     sent.
 
     :param job: job object.
     :param state: internal pilot state (string).
     :return: valid server state (string).
     """
 
     if job.serverstate == "finished" or job.serverstate == "failed":
         pass
     elif job.serverstate == "" and state != "finished" and state != "failed":
         job.serverstate = 'starting'
     elif state == "finished" or state == "failed" or state == "holding":
         job.serverstate = state
     else:
         job.serverstate = 'running'
 
     return job.serverstate
 
 
 def publish_harvester_reports(state, args, data, job, final):
     """
     Publish all reports needed by Harvester.
 
     :param state: job state (string).
     :param args: pilot args object.
     :param data: data structure for server update (dictionary).
     :param job: job object.
     :param final: is this the final update? (Boolean).
     :return: True if successful, False otherwise (Boolean).
     """
 
     # write part of the heartbeat message to worker attributes files needed by Harvester
     path = get_worker_attributes_file(args)
 
     # add jobStatus (state) for Harvester
     data['jobStatus'] = state
 
     # publish work report
     if not publish_work_report(data, path):
         logger.debug('failed to write to workerAttributesFile %s', path)
         return False
 
     # check if we are in final state then write out information for output files
     if final:
         # Use the job information to write Harvester event_status.dump file
         event_status_file = get_event_status_file(args)
         if publish_stageout_files(job, event_status_file):
             logger.debug('wrote log and output files to file %s', event_status_file)
         else:
             logger.warning('could not write log and output files to file %s', event_status_file)
             return False
 
         # publish job report
         _path = os.path.join(job.workdir, config.Payload.jobreport)
         if os.path.exists(_path):
             if publish_job_report(job, args, config.Payload.jobreport):
                 logger.debug('wrote job report file')
                 return True
             else:
                 logger.warning('failed to write job report file')
                 return False
     else:
         logger.info('finished writing various report files in Harvester mode')
 
     return True
 
 
 def write_heartbeat_to_file(data):
     """
     Write heartbeat dictionary to file.
     This is only done when server updates are not wanted.
 
     :param data: server data (dictionary).
     :return: True if successful, False otherwise (Boolean).
     """
 
     path = os.path.join(os.environ.get('PILOT_HOME'), config.Pilot.heartbeat_message)
     if write_json(path, data):
         logger.debug('heartbeat dictionary: %s', data)
         logger.debug('wrote heartbeat to file %s', path)
         return True
     else:
         return False
 
 
 def send_state(job, args, state, xml=None, metadata=None, test_tobekilled=False):
     """
     Update the server (send heartbeat message).
     Interpret and handle any server instructions arriving with the updateJob back channel.
 
     :param job: job object.
     :param args: Pilot arguments (e.g. containing queue name, queuedata dictionary, etc).
     :param state: job state (string).
     :param xml: optional metadata xml (string).
     :param metadata: job report metadata read as a string.
     :param test_tobekilled: emulate a tobekilled command (boolean).
     :return: boolean (True if successful, False otherwise).
     """
 
     state = get_proper_state(job, state)
 
     # should the pilot make any server updates?
     if not args.update_server:
         logger.info('pilot will not update the server (heartbeat message will be written to file)')
     tag = 'sending' if args.update_server else 'writing'
 
     if state == 'finished' or state == 'failed' or state == 'holding':
         final = True
         os.environ['SERVER_UPDATE'] = SERVER_UPDATE_UPDATING
         logger.info('job %s has %s - %s final server update', job.jobid, state, tag)
 
         # make sure that job.state is 'failed' if there's a set error code
         if job.piloterrorcode or job.piloterrorcodes:
             logger.warning('making sure that job.state is set to failed since a pilot error code is set')
             state = 'failed'
             job.state = state
         # make sure an error code is properly set
         elif state != 'finished':
             verify_error_code(job)
     else:
         final = False
         logger.info('job %s has state \'%s\' - %s heartbeat', job.jobid, state, tag)
 
     # build the data structure needed for getJob, updateJob
     data = get_data_structure(job, state, args, xml=xml, metadata=metadata)
 
     # write the heartbeat message to file if the server is not to be updated by the pilot (Nordugrid mode)
     if not args.update_server:
         logger.debug('is_harvester_mode(args) : {0}'.format(is_harvester_mode(args)))
         # if in harvester mode write to files required by harvester
         if is_harvester_mode(args):
             return publish_harvester_reports(state, args, data, job, final)
         else:
             # store the file in the main workdir
             return write_heartbeat_to_file(data)
 
     try:
         if config.Pilot.pandajob == 'real':
             time_before = int(time.time())
             max_attempts = 10
             attempt = 0
             done = False
             while attempt < max_attempts and not done:
                 logger.info('job update attempt %d/%d', attempt + 1, max_attempts)
 
                 # get the URL for the PanDA server from pilot options or from config
                 pandaserver = get_panda_server(args.url, args.port)
 
                 res = https.request('{pandaserver}/server/panda/updateJob'.format(pandaserver=pandaserver), data=data)
                 if res is not None:
                     done = True
                 attempt += 1
 
             time_after = int(time.time())
             logger.info('server updateJob request completed in %ds for job %s', time_after - time_before, job.jobid)
             logger.info("server responded with: res = %s", str(res))
 
             show_memory_usage()
 
             if res is not None:
                 # does the server update contain any backchannel information? if so, update the job object
                 handle_backchannel_command(res, job, args, test_tobekilled=test_tobekilled)
 
                 if final:
                     os.environ['SERVER_UPDATE'] = SERVER_UPDATE_FINAL
                     logger.debug('set SERVER_UPDATE=SERVER_UPDATE_FINAL')
                 return True
         else:
             logger.info('skipping job update for fake test job')
             return True
 
     except Exception as error:
         logger.warning('exception caught while sending https request: %s', error)
         logger.warning('possibly offending data: %s', data)
 
     if final:
         os.environ['SERVER_UPDATE'] = SERVER_UPDATE_TROUBLE
         logger.debug('set SERVER_UPDATE=SERVER_UPDATE_TROUBLE')
 
     return False
 
 
 def get_job_status_from_server(job_id, url, port):
     """
     Return the current status of job <jobId> from the dispatcher.
     typical dispatcher response: 'status=finished&StatusCode=0'
     StatusCode  0: succeeded
                10: time-out
                20: general error
                30: failed
     In the case of time-out, the dispatcher will be asked one more time after 10 s.
 
     :param job_id: PanDA job id (int).
     :param url: PanDA server URL (string).
     :param port: PanDA server port (int).
     :return: status (string; e.g. holding), attempt_nr (int), status_code (int)
     """
 
     status = 'unknown'
     attempt_nr = 0
     status_code = 0
     if config.Pilot.pandajob == 'fake':
         return status, attempt_nr, status_code
 
     data = {}
     data['ids'] = job_id
 
     # get the URL for the PanDA server from pilot options or from config
     pandaserver = get_panda_server(url, port)
 
     # ask dispatcher about lost job status
     trial = 1
     max_trials = 2
 
     while trial <= max_trials:
         try:
             # open connection
             ret = https.request('{pandaserver}/server/panda/getStatus'.format(pandaserver=pandaserver), data=data)
             response = ret[1]
             logger.info("response: %s", str(response))
             if response:
                 try:
                     # decode the response
                     # eg. var = ['status=notfound', 'attemptNr=0', 'StatusCode=0']
                     # = response
 
                     status = response['status']  # e.g. 'holding'
                     attempt_nr = int(response['attemptNr'])  # e.g. '0'
                     status_code = int(response['StatusCode'])  # e.g. '0'
                 except Exception as error:
                     logger.warning(
                         "exception: dispatcher did not return allowed values: %s, %s", str(ret), error)
                     status = "unknown"
                     attempt_nr = -1
                     status_code = 20
                 else:
                     logger.debug('server job status=%s, attempt_nr=%d, status_code=%d', status, attempt_nr, status_code)
             else:
                 logger.warning("dispatcher did not return allowed values: %s", str(ret))
                 status = "unknown"
                 attempt_nr = -1
                 status_code = 20
         except Exception as error:
             logger.warning("could not interpret job status from dispatcher: %s", error)
             status = 'unknown'
             attempt_nr = -1
             status_code = -1
             break
         else:
             if status_code == 0:  # success
                 break
             elif status_code == 10:  # time-out
                 trial += 1
                 time.sleep(10)
                 continue
             elif status_code == 20:  # other error
                 if ret[0] == 13056 or ret[0] == '13056':
                     logger.warning("wrong certificate used with curl operation? (encountered error 13056)")
                 break
             else:  # general error
                 break
 
     return status, attempt_nr, status_code
 
 
 def get_panda_server(url, port):
     """
     Get the URL for the PanDA server.
 
     :param url: URL string, if set in pilot option (port not included).
     :param port: port number, if set in pilot option (int).
     :return: full URL (either from pilot options or from config file)
     """
 
     if url.startswith('https://'):
         url = url.replace('https://', '')
 
     if url != '' and port != 0:
         pandaserver = '%s:%s' % (url, port) if ":" not in url else url
     else:
         pandaserver = config.Pilot.pandaserver
 
     if not pandaserver.startswith('http'):
         pandaserver = 'https://' + pandaserver
 
     # add randomization for PanDA server
     default = 'pandaserver.cern.ch'
     if default in pandaserver:
         rnd = random.choice([socket.getfqdn(vv) for vv in set([v[-1][0] for v in socket.getaddrinfo(default, 25443, socket.AF_INET)])])
         pandaserver = pandaserver.replace(default, rnd)
         logger.debug('updated %s to %s', default, pandaserver)
 
     return pandaserver
 
 
 def get_debug_command(cmd):
     """
     Identify and filter the given debug command.
 
     Note: only a single command will be allowed from a predefined list: tail, ls, gdb, ps, du.
 
     :param cmd: raw debug command from job definition (string).
     :return: debug_mode (Boolean, True if command is deemed ok), debug_command (string).
     """
 
     debug_mode = False
     debug_command = ""
 
     allowed_commands = ['tail', 'ls', 'ps', 'gdb', 'du']
     forbidden_commands = ['rm']
 
     # remove any 'debug,' command that the server might send redundantly
     if ',' in cmd and 'debug' in cmd:
         cmd = cmd.replace('debug,', '').replace(',debug', '')
     try:
         tmp = cmd.split(' ')
         com = tmp[0]
     except Exception as error:
         logger.warning('failed to identify debug command: %s', error)
     else:
         if com not in allowed_commands:
             logger.warning('command=%s is not in the list of allowed commands: %s', com, str(allowed_commands))
         elif ';' in cmd or '&#59' in cmd:
             logger.warning('debug command cannot contain \';\': \'%s\'', cmd)
         elif com in forbidden_commands:
             logger.warning('command=%s is not allowed', com)
         else:
             debug_mode = True
             debug_command = cmd
     return debug_mode, debug_command
 
 
 def handle_backchannel_command(res, job, args, test_tobekilled=False):
     """
     Does the server update contain any backchannel information? if so, update the job object.
 
     :param res: server response (dictionary).
     :param job: job object.
     :param args: pilot args object.
     :param test_tobekilled: emulate a tobekilled command (boolean).
     :return:
     """
 
     if test_tobekilled:
         logger.info('faking a \'tobekilled\' command')
         res['command'] = 'tobekilled'
 
     if 'command' in res and res.get('command') != 'NULL':
         # warning: server might return comma-separated string, 'debug,tobekilled'
         cmd = res.get('command')
         # is it a 'command options'-type? debug_command=tail .., ls .., gdb .., ps .., du ..
         if ' ' in cmd and 'tobekilled' not in cmd:
             try:
                 job.debug, job.debug_command = get_debug_command(cmd)
             except Exception as error:
                 logger.debug('exception caught in get_debug_command(): %s', error)
         elif 'tobekilled' in cmd:
             logger.info('pilot received a panda server signal to kill job %s at %s', job.jobid, time_stamp())
             set_pilot_state(job=job, state="failed")
             job.piloterrorcodes, job.piloterrordiags = errors.add_error_code(errors.PANDAKILL)
             if job.pid:
                 logger.debug('killing payload process')
                 kill_process(job.pid)
             else:
                 logger.debug('no pid to kill')
             args.abort_job.set()
         elif 'softkill' in cmd:
             logger.info('pilot received a panda server signal to softkill job %s at %s', job.jobid, time_stamp())
             # event service kill instruction
             job.debug_command = 'softkill'
         elif 'debug' in cmd:
             logger.info('pilot received a command to turn on standard debug mode from the server')
             job.debug = True
             job.debug_command = 'debug'
         elif 'debugoff' in cmd:
             logger.info('pilot received a command to turn off debug mode from the server')
             job.debug = False
             job.debug_command = 'debugoff'
         else:
             logger.warning('received unknown server command via backchannel: %s', cmd)
 
     # for testing debug mode
     # job.debug = True
     # job.debug_command = 'du -sk'
     # job.debug_command = 'tail -30 payload.stdout'
     # job.debug_command = 'ls -ltr workDir'  # not really tested
     # job.debug_command = 'ls -ltr %s' % job.workdir
     # job.debug_command = 'ps -ef'
     # job.debug_command = 'ps axo pid,ppid,pgid,args'
     # job.debug_command = 'gdb --pid % -ex \'generate-core-file\''
 
 
 def add_data_structure_ids(data, version_tag):
     """
     Add pilot, batch and scheduler ids to the data structure for getJob, updateJob.
 
     :param data: data structure (dict).
     :return: updated data structure (dict).
     """
 
     schedulerid = get_job_scheduler_id()
     if schedulerid:
         data['schedulerID'] = schedulerid
 
     pilotid = get_pilot_id()
     if pilotid:
         pilotversion = os.environ.get('PILOT_VERSION')
 
         # report the batch system job id, if available
         batchsystem_type, batchsystem_id = get_batchsystem_jobid()
 
         if batchsystem_type:
             data['pilotID'] = "%s|%s|%s|%s" % (pilotid, batchsystem_type, version_tag, pilotversion)
             data['batchID'] = batchsystem_id
         else:
             data['pilotID'] = "%s|%s|%s" % (pilotid, version_tag, pilotversion)
 
     return data
 
 
 def get_data_structure(job, state, args, xml=None, metadata=None):
     """
     Build the data structure needed for getJob, updateJob.
 
     :param job: job object.
     :param state: state of the job (string).
     :param args:
     :param xml: optional XML string.
     :param metadata: job report metadata read as a string.
     :return: data structure (dictionary).
     """
 
     data = {'jobId': job.jobid,
             'state': state,
             'timestamp': time_stamp(),
             'siteName': os.environ.get('PILOT_SITENAME'),  # args.site,
             'node': get_node_name(),
             'attemptNr': job.attemptnr}
 
     # add pilot, batch and scheduler ids to the data structure
     data = add_data_structure_ids(data, args.version_tag)
 
     starttime = get_postgetjob_time(job.jobid, args)
     if starttime:
         data['startTime'] = starttime
 
     job_metrics = get_job_metrics(job)
     if job_metrics:
         data['jobMetrics'] = job_metrics
 
     if xml is not None:
         data['xml'] = xml
     if metadata is not None:
         data['metaData'] = metadata
 
     # in debug mode, also send a tail of the latest log file touched by the payload
     if job.debug:
         data['stdout'] = process_debug_mode(job)
 
     # add the core count
     if job.corecount and job.corecount != 'null' and job.corecount != 'NULL':
         data['coreCount'] = job.corecount
         #data['coreCount'] = mean(job.corecounts) if job.corecounts else job.corecount
     if job.corecounts:
         _mean = mean(job.corecounts)
         logger.info('mean actualcorecount: %f', _mean)
         data['meanCoreCount'] = _mean
 
     # get the number of events, should report in heartbeat in case of preempted.
     if job.nevents != 0:
         data['nEvents'] = job.nevents
         logger.info("total number of processed events: %d (read)", job.nevents)
     else:
         logger.info("payload/TRF did not report the number of read events")
 
     # get the CU consumption time
     constime = get_cpu_consumption_time(job.cpuconsumptiontime)
     if constime and constime != -1:
         data['cpuConsumptionTime'] = constime
         data['cpuConversionFactor'] = job.cpuconversionfactor
     data['cpuConsumptionUnit'] = job.cpuconsumptionunit + "+" + get_cpu_model()
 
     instruction_sets = has_instruction_sets(['AVX2'])
     product, vendor = get_display_info()
     if instruction_sets:
         if 'cpuConsumptionUnit' in data:
             data['cpuConsumptionUnit'] += '+' + instruction_sets
         else:
             data['cpuConsumptionUnit'] = instruction_sets
         if product and vendor:
             logger.debug('cpuConsumptionUnit: could have added: product=%s, vendor=%s', product, vendor)
 
     # add memory information if available
     add_memory_info(data, job.workdir, name=job.memorymonitor)
     if state == 'finished' or state == 'failed':
         add_timing_and_extracts(data, job, state, args)
         add_error_codes(data, job)
 
     return data
 
 
 def process_debug_mode(job):
     """
     Handle debug mode - preprocess debug command, get the output and kill the payload in case of gdb.
 
     :param job: job object.
     :return: stdout from debug command (string).
     """
 
     # for gdb commands, use the proper gdb version (the system one may be too old)
     if job.debug_command.startswith('gdb '):
         pilot_user = os.environ.get('PILOT_USER', 'generic').lower()
         user = __import__('pilot.user.%s.common' % pilot_user, globals(), locals(), [pilot_user], 0)  # Python 2/3
         user.preprocess_debug_command(job)
 
     stdout = get_debug_stdout(job)
     if stdout:
         # in case gdb was successfully used, the payload can now be killed
         if job.debug_command.startswith('gdb ') and job.pid:
             job.piloterrorcodes, job.piloterrordiags = errors.add_error_code(errors.PANDAKILL,
                                                                              msg='payload was killed after gdb produced requested core file')
             logger.debug('will proceed to kill payload processes')
             kill_processes(job.pid)
 
     return stdout
 
 
 def get_debug_stdout(job):
     """
     Return the requested output from a given debug command.
 
     :param job: job object.
     :return: output (string).
     """
 
     if job.debug_command == 'debug':
         return get_payload_log_tail(job.workdir)
     elif 'tail ' in job.debug_command:
         return get_requested_log_tail(job.debug_command, job.workdir)
     elif 'ls ' in job.debug_command:
         return get_ls(job.debug_command, job.workdir)
     elif 'ps ' in job.debug_command or 'gdb ' in job.debug_command:
         return get_general_command_stdout(job)
     else:
         # general command, execute and return output
         _, stdout, _ = execute(job.debug_command)
         logger.info('debug_command: %s:\n\n%s\n', job.debug_command, stdout)
         return stdout
 
 
 def get_general_command_stdout(job):
     """
     Return the output from the requested debug command.
 
     :param job: job object.
     :return: output (string).
     """
 
     stdout = ''
 
     # for gdb, we might have to process the debug command (e.g. to identify the proper pid to debug)
     if 'gdb ' in job.debug_command and '--pid %' in job.debug_command:
         pilot_user = os.environ.get('PILOT_USER', 'generic').lower()
         user = __import__('pilot.user.%s.common' % pilot_user, globals(), locals(), [pilot_user], 0)  # Python 2/3
         job.debug_command = user.process_debug_command(job.debug_command, job.jobid)
 
     if job.debug_command:
         _containerisation = False  # set this with some logic instead - not used for now
         if _containerisation:
             try:
                 containerise_general_command(job, job.infosys.queuedata.container_options,
                                              label='general',
                                              container_type='container')
             except PilotException as error:
                 logger.warning('general containerisation threw a pilot exception: %s', error)
             except Exception as error:
                 logger.warning('general containerisation threw an exception: %s', error)
         else:
             _, stdout, stderr = execute(job.debug_command)
             logger.debug("%s (stdout):\n\n%s\n\n", job.debug_command, stdout)
             logger.debug("%s (stderr):\n\n%s\n\n", job.debug_command, stderr)
 
         # in case a core file was produced, locate it
         path = locate_core_file(cmd=job.debug_command) if 'gdb ' in job.debug_command else ''
         if path:
             # copy it to the working directory (so it will be saved in the log)
             try:
                 copy(path, job.workdir)
             except Exception:
                 pass
 
     return stdout
 
 
 def get_ls(debug_command, workdir):
     """
     Return the requested ls debug command.
 
     :param debug_command: full debug command (string).
     :param workdir: job work directory (string).
     :return: output (string).
     """
 
     items = debug_command.split(' ')
     # cmd = items[0]
     options = ' '.join(items[1:])
     path = options.split(' ')[-1] if ' ' in options else options
     if path.startswith('-'):
         path = '.'
     finalpath = os.path.join(workdir, path)
     debug_command = debug_command.replace(path, finalpath)
 
     _, stdout, _ = execute(debug_command)
     logger.debug("%s:\n\n%s\n\n", debug_command, stdout)
 
     return stdout
 
 
 def get_requested_log_tail(debug_command, workdir):
     """
     Return the tail of the requested debug log.
 
     Examples
       tail workdir/tmp.stdout* <- pilot finds the requested log file in the specified relative path
       tail log.RAWtoALL <- pilot finds the requested log file
 
     :param debug_command: full debug command (string).
     :param workdir: job work directory (string).
     :return: output (string).
     """
 
     _tail = ""
     items = debug_command.split(' ')
     cmd = items[0]
     options = ' '.join(items[1:])
     logger.debug('debug command: %s', cmd)
     logger.debug('debug options: %s', options)
 
     # assume that the path is the last of the options; <some option> <some path>
     path = options.split(' ')[-1] if ' ' in options else options
     fullpath = os.path.join(workdir, path)
 
     # find all files with the given pattern and pick the latest updated file (if several)
     files = glob(fullpath)
     if files:
         logger.info('files found: %s', str(files))
         _tail = get_latest_log_tail(files)
     else:
         logger.warning('did not find \'%s\' in path %s', path, fullpath)
 
     if _tail:
         logger.debug('tail =\n\n%s\n\n', _tail)
 
     return _tail
 
 
 def add_error_codes(data, job):
     """
     Add error codes to data structure.
 
     :param data: data dictionary.
     :param job: job object.
     :return:
     """
 
     # error codes
     pilot_error_code = job.piloterrorcode
     pilot_error_codes = job.piloterrorcodes
     if pilot_error_codes != []:
         logger.warning('pilotErrorCodes = %s (will report primary/first error code)', str(pilot_error_codes))
         data['pilotErrorCode'] = pilot_error_codes[0]
     else:
         data['pilotErrorCode'] = pilot_error_code
 
     # add error info
     pilot_error_diag = job.piloterrordiag
     pilot_error_diags = job.piloterrordiags
     if pilot_error_diags != []:
         logger.warning('pilotErrorDiags = %s (will report primary/first error diag)', str(pilot_error_diags))
         data['pilotErrorDiag'] = pilot_error_diags[0]
     else:
         data['pilotErrorDiag'] = pilot_error_diag
     data['transExitCode'] = job.transexitcode
     data['exeErrorCode'] = job.exeerrorcode
     data['exeErrorDiag'] = job.exeerrordiag
 
 
 def get_cpu_consumption_time(cpuconsumptiontime):
     """
     Get the CPU consumption time.
     The function makes sure that the value exists and is within allowed limits (< 10^9).
 
     :param cpuconsumptiontime: CPU consumption time (int/None).
     :return: properly set CPU consumption time (int/None).
     """
 
     constime = None
 
     try:
         constime = int(cpuconsumptiontime)
     except Exception:
         constime = None
     if constime and constime > 10 ** 9:
         logger.warning("unrealistic cpuconsumptiontime: %d (reset to -1)", constime)
         constime = -1
 
     return constime
 
 
 def add_timing_and_extracts(data, job, state, args):
     """
     Add timing info and log extracts to data structure for a completed job (finished or failed) to be sent to server.
     Note: this function updates the data dictionary.
 
     :param data: data structure (dictionary).
     :param job: job object.
     :param state: state of the job (string).
     :param args: pilot args.
     :return:
     """
 
     time_getjob, time_stagein, time_payload, time_stageout, time_total_setup = timing_report(job.jobid, args)
     data['pilotTiming'] = "%s|%s|%s|%s|%s" % \
                           (time_getjob, time_stagein, time_payload, time_stageout, time_total_setup)
 
     # add log extracts (for failed/holding jobs or for jobs with outbound connections)
     extracts = ""
     if state == 'failed' or state == 'holding':
         pilot_user = os.environ.get('PILOT_USER', 'generic').lower()
         user = __import__('pilot.user.%s.diagnose' % pilot_user, globals(), locals(), [pilot_user], 0)  # Python 2/3
         extracts = user.get_log_extracts(job, state)
         if extracts != "":
             logger.warning('\nXXXXXXXXXXXXXXXXXXXXX[begin log extracts]\n%s\nXXXXXXXXXXXXXXXXXXXXX[end log extracts]', extracts)
     data['pilotLog'] = extracts[:1024]
     data['endTime'] = time.time()
 
 
 def add_memory_info(data, workdir, name=""):
     """
     Add memory information (if available) to the data structure that will be sent to the server with job updates
     Note: this function updates the data dictionary.
 
     :param data: data structure (dictionary).
     :param workdir: working directory of the job (string).
     :param name: name of memory monitor (string).
     :return:
     """
 
     pilot_user = os.environ.get('PILOT_USER', 'generic').lower()
     utilities = __import__('pilot.user.%s.utilities' % pilot_user, globals(), locals(), [pilot_user], 0)  # Python 2/3
     try:
         utility_node = utilities.get_memory_monitor_info(workdir, name=name)
         data.update(utility_node)
     except Exception as error:
         logger.info('memory information not available: %s', error)
 
 
 def remove_pilot_logs_from_list(list_of_files):
     """
     Remove any pilot logs from the list of last updated files.
 
     :param list_of_files: list of last updated files (list).
     :return: list of files (list).
     """
 
     # note: better to move experiment specific files to user area
 
     # ignore the pilot log files
     try:
         to_be_removed = [config.Pilot.pilotlog, config.Pilot.stageinlog, config.Pilot.stageoutlog,
                          config.Pilot.timing_file, config.Pilot.remotefileverification_dictionary,
                          config.Pilot.remotefileverification_log, config.Pilot.base_trace_report,
                          config.Container.container_script, config.Container.release_setup,
                          config.Container.stagein_status_dictionary, config.Container.stagein_replica_dictionary,
                          'eventLoopHeartBeat.txt', 'memory_monitor_output.txt', 'memory_monitor_summary.json_snapshot']
     except Exception as error:
         logger.warning('exception caught: %s', error)
         to_be_removed = []
 
     new_list_of_files = []
     for filename in list_of_files:
         if os.path.basename(filename) not in to_be_removed and '/pilot/' not in filename and 'prmon' not in filename:
             new_list_of_files.append(filename)
 
     return new_list_of_files
 
 
 def get_payload_log_tail(workdir):
     """
     Return the tail of the payload stdout or its latest updated log file.
 
     :param workdir: job work directory (string).
     :return: tail of stdout (string).
     """
 
     # find the latest updated log file
     # list_of_files = get_list_of_log_files()
     # find the latest updated text file
     list_of_files = find_text_files()
     list_of_files = remove_pilot_logs_from_list(list_of_files)
 
     if not list_of_files:
         logger.info('no log files were found (will use default %s)', config.Payload.payloadstdout)
         list_of_files = [os.path.join(workdir, config.Payload.payloadstdout)]
 
     return get_latest_log_tail(list_of_files)
 
 
 def get_latest_log_tail(files):
     """
     Get the tail of the latest updated file from the given file list.
 
     :param files: files (list).
     """
 
     stdout_tail = ""
 
     try:
         latest_file = max(files, key=os.path.getmtime)
         logger.info('tail of file %s will be added to heartbeat', latest_file)
 
         # now get the tail of the found log file and protect against potentially large tails
         stdout_tail = latest_file + "\n" + tail(latest_file)
         stdout_tail = stdout_tail[-2048:]
     except OSError as exc:
         logger.warning('failed to get payload stdout tail: %s', exc)
 
     return stdout_tail
 
 
 def validate(queues, traces, args):
     """
     Perform validation of job.
 
     :param queues: queues object.
     :param traces: traces object.
     :param args: args object.
     :return:
     """
 
     while not args.graceful_stop.is_set():
         time.sleep(0.5)
         try:
             job = queues.jobs.get(block=True, timeout=1)
         except queue.Empty:
             continue
 
         traces.pilot['nr_jobs'] += 1
 
         # set the environmental variable for the task id
         os.environ['PanDA_TaskID'] = str(job.taskid)
         logger.info('processing PanDA job %s from task %s', job.jobid, job.taskid)
 
         if _validate_job(job):
 
             # Define a new parent group
             os.setpgrp()
 
             job_dir = os.path.join(args.mainworkdir, 'PanDA_Pilot-%s' % job.jobid)
             logger.debug('creating job working directory: %s', job_dir)
             try:
                 os.mkdir(job_dir)
                 os.chmod(job_dir, 0o770)
                 job.workdir = job_dir
             except Exception as error:
                 logger.debug('cannot create working directory: %s', error)
                 traces.pilot['error_code'] = errors.MKDIR
                 job.piloterrorcodes, job.piloterrordiags = errors.add_error_code(traces.pilot['error_code'])
                 job.piloterrordiag = error
                 put_in_queue(job, queues.failed_jobs)
                 break
             else:
                 create_k8_link(job_dir)
 
 #            try:
 #                # stream the job object to file
 #                job_dict = job.to_json()
 #                write_json(os.path.join(job.workdir, 'job.json'), job_dict)
 #            except Exception as error:
 #                logger.debug('exception caught: %s', error)
 #            else:
 #                try:
 #                    _job_dict = read_json(os.path.join(job.workdir, 'job.json'))
 #                    job_dict = loads(_job_dict)
 #                    _job = JobData(job_dict, use_kmap=False)
 #                except Exception as error:
 #                    logger.warning('exception caught: %s', error)
 
             create_symlink(from_path='../%s' % config.Pilot.pilotlog, to_path=os.path.join(job_dir, config.Pilot.pilotlog))
 
             # pre-cleanup
             pilot_user = os.environ.get('PILOT_USER', 'generic').lower()
             utilities = __import__('pilot.user.%s.utilities' % pilot_user, globals(), locals(), [pilot_user], 0)  # Python 2/3
             try:
                 utilities.precleanup()
             except Exception as error:
                 logger.warning('exception caught: %s', error)
 
             # store the PanDA job id for the wrapper to pick up
             store_jobid(job.jobid, args.sourcedir)
 
             # run the delayed space check now
             delayed_space_check(queues, traces, args, job)
 
             # make sure that ctypes is available (needed at the end by orphan killer)
             verify_ctypes(queues, job)
         else:
             logger.debug('Failed to validate job=%s', job.jobid)
             put_in_queue(job, queues.failed_jobs)
 
     # proceed to set the job_aborted flag?
     if threads_aborted():
         logger.debug('will proceed to set job_aborted')
         args.job_aborted.set()
     else:
         logger.debug('will not set job_aborted yet')
 
     logger.debug('[job] validate thread has finished')
 
 
 def verify_ctypes(queues, job):
     """
     Verify ctypes and make sure all subprocess are parented.
 
     :param queues: queues object.
     :param job: job object.
     :return:
     """
 
     try:
         import ctypes
     # except ModuleNotFoundError as error:  # Python 3
     except Exception as error:
         diagnostics = 'ctypes python module could not be imported: %s' % error
         logger.warning(diagnostics)
         #job.piloterrorcodes, job.piloterrordiags = errors.add_error_code(errors.NOCTYPES, msg=diagnostics)
         #logger.debug('Failed to validate job=%s', job.jobid)
         #put_in_queue(job, queues.failed_jobs)
     else:
         logger.debug('ctypes python module imported')
 
         # make sure all children are parented by the pilot
         # specifically, this will include any 'orphans', i.e. if the pilot kills all subprocesses at the end,
         # 'orphans' will be included (orphans seem like the wrong name)
         libc = ctypes.CDLL('libc.so.6')
         pr_set_child_subreaper = 36
         libc.prctl(pr_set_child_subreaper, 1)
         logger.debug('all child subprocesses will be parented')
 
 
 def delayed_space_check(queues, traces, args, job):
     """
     Run the delayed space check if necessary.
 
     :param queues: queues object.
     :param traces: traces object.
     :param args: args object.
     :param job: job object.
     :return:
     """
 
     proceed_with_local_space_check = True if (args.harvester_submitmode.lower() == 'push' and args.update_server) else False
     if proceed_with_local_space_check:
         logger.debug('pilot will now perform delayed space check')
         exit_code, diagnostics = check_local_space()
         if exit_code != 0:
             traces.pilot['error_code'] = errors.NOLOCALSPACE
             # set the corresponding error code
             job.piloterrorcodes, job.piloterrordiags = errors.add_error_code(errors.NOLOCALSPACE, msg=diagnostics)
             logger.debug('Failed to validate job=%s', job.jobid)
             put_in_queue(job, queues.failed_jobs)
         else:
             put_in_queue(job, queues.validated_jobs)
     else:
         put_in_queue(job, queues.validated_jobs)
 
 
 def create_k8_link(job_dir):
     """
     Create a soft link to the payload workdir on Kubernetes if SHARED_DIR exists.
 
     :param job_dir: payload workdir (string).
     """
 
     shared_dir = os.environ.get('SHARED_DIR', None)
     if shared_dir:
         #create_symlink(from_path=os.path.join(shared_dir, 'payload_workdir'), to_path=job_dir)
         create_symlink(from_path=job_dir, to_path=os.path.join(shared_dir, 'payload_workdir'))
     else:
         logger.debug('will not create symlink in SHARED_DIR')
 
 
 def store_jobid(jobid, init_dir):
     """
     Store the PanDA job id in a file that can be picked up by the wrapper for other reporting.
 
     :param jobid: job id (int).
     :param init_dir: pilot init dir (string).
     :return:
     """
 
     try:
         path = os.path.join(os.path.join(init_dir, 'pilot2'), config.Pilot.jobid_file)
         path = path.replace('pilot2/pilot2', 'pilot2')  # dirty fix for bad paths
         mode = 'a' if os.path.exists(path) else 'w'
         logger.debug('path=%s  mode=%s', path, mode)
         write_file(path, "%s\n" % str(jobid), mode=mode, mute=False)
     except Exception as error:
         logger.warning('exception caught while trying to store job id: %s', error)
 
 
 def create_data_payload(queues, traces, args):
     """
     Get a Job object from the "validated_jobs" queue.
 
     If the job has defined input files, move the Job object to the "data_in" queue and put the internal pilot state to
     "stagein". In case there are no input files, place the Job object in the "finished_data_in" queue. For either case,
     the thread also places the Job object in the "payloads" queue (another thread will retrieve it and wait for any
     stage-in to finish).
 
     :param queues: internal queues for job handling.
     :param traces: tuple containing internal pilot states.
     :param args: Pilot arguments (e.g. containing queue name, queuedata dictionary, etc).
     :return:
     """
 
     while not args.graceful_stop.is_set():
         time.sleep(0.5)
         try:
             job = queues.validated_jobs.get(block=True, timeout=1)
         except queue.Empty:
             continue
 
         if job.indata:
             # if the job has input data, put the job object in the data_in queue which will trigger stage-in
             set_pilot_state(job=job, state='stagein')
             put_in_queue(job, queues.data_in)
 
         else:
             # if the job does not have any input data, then pretend that stage-in has finished and put the job
             # in the finished_data_in queue
             put_in_queue(job, queues.finished_data_in)
 
         put_in_queue(job, queues.payloads)
 
     # proceed to set the job_aborted flag?
     if threads_aborted():
         logger.debug('will proceed to set job_aborted')
         args.job_aborted.set()
     else:
         logger.debug('will not set job_aborted yet')
 
     logger.debug('[job] create_data_payload thread has finished')
 
 
 def get_task_id():
     """
     Return the task id for the current job.
     Note: currently the implementation uses an environmental variable to store this number (PanDA_TaskID).
 
     :return: task id (string). Returns empty string in case of error.
     """
 
     if "PanDA_TaskID" in os.environ:
         taskid = os.environ["PanDA_TaskID"]
     else:
         logger.warning('PanDA_TaskID not set in environment')
         taskid = ""
 
     return taskid
 
 
 def get_job_label(args):
     """
     Return a proper job label.
     The function returns a job label that corresponds to the actual pilot version, ie if the pilot is a development
     version (ptest or rc_test2) or production version (managed or user).
     Example: -i RC -> job_label = rc_test2.
     NOTE: it should be enough to only use the job label, -j rc_test2 (and not specify -i RC at all).
 
     :param args: pilot args object.
     :return: job_label (string).
     """
 
     # PQ status
     status = infosys.queuedata.status
 
     if args.version_tag == 'RC' and args.job_label == 'rc_test2':
         job_label = 'rc_test2'
     elif args.version_tag == 'RC' and args.job_label == 'ptest':
         job_label = args.job_label
     elif args.version_tag == 'RCM' and args.job_label == 'ptest':
         job_label = 'rcm_test2'
     elif args.version_tag == 'ALRB':
         job_label = 'rc_alrb'
     elif status == 'test' and args.job_label != 'ptest':
         logger.warning('PQ status set to test - will use job label / prodSourceLabel test')
         job_label = 'test'
     else:
         job_label = args.job_label
 
     return job_label
 
 
 def get_dispatcher_dictionary(args):
     """
     Return a dictionary with required fields for the dispatcher getJob operation.
 
     The dictionary should contain the following fields: siteName, computingElement (queue name),
     prodSourceLabel (e.g. user, test, ptest), diskSpace (available disk space for a job in MB),
     workingGroup, countryGroup, cpu (float), mem (float) and node (worker node name).
 
     workingGroup, countryGroup and allowOtherCountry
     we add a new pilot setting allowOtherCountry=True to be used in conjunction with countryGroup=us for
     US pilots. With these settings, the Panda server will produce the desired behaviour of dedicated X% of
     the resource exclusively (so long as jobs are available) to countryGroup=us jobs. When allowOtherCountry=false
     this maintains the behavior relied on by current users of the countryGroup mechanism -- to NOT allow
     the resource to be used outside the privileged group under any circumstances.
 
     :param args: arguments (e.g. containing queue name, queuedata dictionary, etc).
     :returns: dictionary prepared for the dispatcher getJob operation.
     """
 
     _diskspace = get_disk_space(infosys.queuedata)
 
     _mem, _cpu, _disk = collect_workernode_info()
 
     _nodename = get_node_name()
 
     # override for RC dev pilots
     job_label = get_job_label(args)
 
     data = {
         'siteName': infosys.queuedata.resource,  # next: remove redundant '-r' option of pilot.py
         'computingElement': args.queue,
         'prodSourceLabel': job_label,
         'diskSpace': _diskspace,
         'workingGroup': args.working_group,
         'cpu': _cpu,
         'mem': _mem,
         'node': _nodename
     }
 
     if args.jobtype != "":
         data['jobType'] = args.jobtype
 
     if args.allow_other_country != "":
         data['allowOtherCountry'] = args.allow_other_country
 
     if args.country_group != "":
         data['countryGroup'] = args.country_group
 
     if args.job_label == 'self':
         dn = get_distinguished_name()
         data['prodUserID'] = dn
 
     taskid = get_task_id()
     if taskid != "" and args.allow_same_user:
         data['taskID'] = taskid
         logger.info("will download a new job belonging to task id: %s", data['taskID'])
 
     if args.resource_type != "":
         data['resourceType'] = args.resource_type
 
     # add harvester fields
     if 'HARVESTER_ID' in os.environ:
         data['harvester_id'] = os.environ.get('HARVESTER_ID')
     if 'HARVESTER_WORKER_ID' in os.environ:
         data['worker_id'] = os.environ.get('HARVESTER_WORKER_ID')
 
 #    instruction_sets = has_instruction_sets(['AVX', 'AVX2'])
 #    if instruction_sets:
 #        data['cpuConsumptionUnit'] = instruction_sets
 
     return data
 
 
 def proceed_with_getjob(timefloor, starttime, jobnumber, getjob_requests, max_getjob_requests, update_server, submitmode, harvester, verify_proxy, traces):
     """
     Can we proceed with getJob?
     We may not proceed if we have run out of time (timefloor limit), if the proxy is too short, if disk space is too
     small or if we have already proceed enough jobs.
 
     :param timefloor: timefloor limit (s) (int).
     :param starttime: start time of retrieve() (s) (int).
     :param jobnumber: number of downloaded jobs (int).
     :param getjob_requests: number of getjob requests (int).
     :param update_server: should pilot update server? (Boolean).
     :param submitmode: Harvester submit mode, PULL or PUSH (string).
     :param harvester: True if Harvester is used, False otherwise. Affects the max number of getjob reads (from file) (Boolean).
     :param verify_proxy: True if the proxy should be verified. False otherwise (Boolean).
     :param traces: traces object (to be able to propagate a proxy error all the way back to the wrapper).
     :return: True if pilot should proceed with getJob (Boolean).
     """
 
     # use for testing thread exceptions. the exception will be picked up by ExcThread run() and caught in job.control()
     # raise NoLocalSpace('testing exception from proceed_with_getjob')
 
     #timefloor = 600
     currenttime = time.time()
 
     pilot_user = os.environ.get('PILOT_USER', 'generic').lower()
     common = __import__('pilot.user.%s.common' % pilot_user, globals(), locals(), [pilot_user], 0)  # Python 2/3
     if not common.allow_timefloor(submitmode):
         timefloor = 0
 
     # should the proxy be verified?
     if verify_proxy:
         userproxy = __import__('pilot.user.%s.proxy' % pilot_user, globals(), locals(), [pilot_user], 0)  # Python 2/3
 
         # is the proxy still valid?
         exit_code, diagnostics = userproxy.verify_proxy()
         if traces.pilot['error_code'] == 0:  # careful so we don't overwrite another error code
             traces.pilot['error_code'] = exit_code
         if exit_code == errors.NOPROXY or exit_code == errors.NOVOMSPROXY:
             logger.warning(diagnostics)
             return False
 
     # is there enough local space to run a job?
     # note: do not run this test at this point if submit mode=PUSH and we are in truePilot mode on ARC
     # (available local space will in this case be checked after the job definition has been read from file, so the
     # pilot can report the error with a server update)
     proceed_with_local_space_check = False if (submitmode.lower() == 'push' and update_server) else True
     if proceed_with_local_space_check:
         exit_code, diagnostics = check_local_space()
         if exit_code != 0:
             traces.pilot['error_code'] = errors.NOLOCALSPACE
             return False
     else:
         logger.debug('pilot will delay local space check until after job definition has been read from file')
 
     maximum_getjob_requests = 60 if harvester else max_getjob_requests  # 1 s apart (if harvester)
     if getjob_requests > int(maximum_getjob_requests):
         logger.warning('reached maximum number of getjob requests (%s) -- will abort pilot', maximum_getjob_requests)
         # use singleton:
         # instruct the pilot to wrap up quickly
         os.environ['PILOT_WRAP_UP'] = 'QUICKLY'
         return False
 
     if timefloor == 0 and jobnumber > 0:
         logger.warning("since timefloor is set to 0, pilot was only allowed to run one job")
         # use singleton:
         # instruct the pilot to wrap up quickly
         os.environ['PILOT_WRAP_UP'] = 'QUICKLY'
         return False
 
     if (currenttime - starttime > timefloor) and jobnumber > 0:
         logger.warning("the pilot has run out of time (timefloor=%d has been passed)", timefloor)
         # use singleton:
         # instruct the pilot to wrap up quickly
         os.environ['PILOT_WRAP_UP'] = 'QUICKLY'
         return False
 
     # timefloor not relevant for the first job
     if jobnumber > 0:
         logger.info('since timefloor=%d s and only %d s has passed since launch, pilot can run another job', timefloor, currenttime - starttime)
 
     if harvester and jobnumber > 0:
         # unless it's the first job (which is preplaced in the init dir), instruct Harvester to place another job
         # in the init dir
         logger.info('asking Harvester for another job')
         request_new_jobs()
 
     if os.environ.get('SERVER_UPDATE', '') == SERVER_UPDATE_UPDATING:
         logger.info('still updating previous job, will not ask for a new job yet')
         return False
 
     os.environ['SERVER_UPDATE'] = SERVER_UPDATE_NOT_DONE
     return True
 
 
 def getjob_server_command(url, port):
     """
     Prepare the getJob server command.
 
     :param url: PanDA server URL (string)
     :param port: PanDA server port
     :return: full server command (URL string)
     """
 
     if url != "":
         port_pattern = '.:([0-9]+)'
         if not findall(port_pattern, url):
             url = url + ':%s' % port
         else:
             logger.debug('URL already contains port: %s', url)
     else:
         url = config.Pilot.pandaserver
     if url == "":
         logger.fatal('PanDA server url not set (either as pilot option or in config file)')
     elif not url.startswith("http"):
         url = 'https://' + url
         logger.warning('detected missing protocol in server url (added)')
 
     return '{pandaserver}/server/panda/getJob'.format(pandaserver=url)
 
 
 def get_job_definition_from_file(path, harvester):
     """
     Get a job definition from a pre-placed file.
     In Harvester mode, also remove any existing job request files since it is no longer needed/wanted.
 
     :param path: path to job definition file.
     :param harvester: True if Harvester is being used (determined from args.harvester), otherwise False
     :return: job definition dictionary.
     """
 
     # remove any existing Harvester job request files (silent in non-Harvester mode) and read the JSON
     if harvester:
         remove_job_request_file()
         if is_json(path):
             job_definition_list = parse_job_definition_file(path)
             if not job_definition_list:
                 logger.warning('no jobs were found in Harvester job definitions file: %s', path)
                 return {}
             else:
                 # remove the job definition file from the original location, place a renamed copy in the pilot dir
                 new_path = os.path.join(os.environ.get('PILOT_HOME'), 'job_definition.json')
                 copy(path, new_path)
                 remove(path)
 
                 # note: the pilot can only handle one job at the time from Harvester
                 return job_definition_list[0]
 
     # old style
     res = {}
     with open(path, 'r') as jobdatafile:
         response = jobdatafile.read()
         if len(response) == 0:
             logger.fatal('encountered empty job definition file: %s', path)
             res = None  # this is a fatal error, no point in continuing as the file will not be replaced
         else:
             # parse response message
             # logger.debug('%s:\n\n%s\n\n', path, response)
             try:
                 from urlparse import parse_qsl  # Python 2
             # except ModuleNotFoundError:  # Python 3
             except Exception:
                 from urllib.parse import parse_qsl  # Python 3
             datalist = parse_qsl(response, keep_blank_values=True)
 
             # convert to dictionary
             for data in datalist:
                 res[data[0]] = data[1]
 
     if os.path.exists(path):
         remove(path)
 
     return res
 
 
 def get_job_definition_from_server(args):
     """
     Get a job definition from a server.
 
     :param args: Pilot arguments (e.g. containing queue name, queuedata dictionary, etc).
     :return: job definition dictionary.
     """
 
     res = {}
 
     # get the job dispatcher dictionary
     data = get_dispatcher_dictionary(args)
 
     cmd = getjob_server_command(args.url, args.port)
     if cmd != "":
         logger.info('executing server command: %s', cmd)
         res = https.request(cmd, data=data)
 
     return res
 
 
 def locate_job_definition(args):
     """
     Locate the job definition file among standard locations.
 
     :param args: Pilot arguments (e.g. containing queue name, queuedata dictionary, etc).
     :return: path (string).
     """
 
     if args.harvester_datadir:
         paths = [os.path.join(args.harvester_datadir, config.Pilot.pandajobdata)]
     else:
         paths = [os.path.join("%s/.." % args.sourcedir, config.Pilot.pandajobdata),
                  os.path.join(args.sourcedir, config.Pilot.pandajobdata),
                  os.path.join(os.environ['PILOT_WORK_DIR'], config.Pilot.pandajobdata)]
 
     if args.harvester_workdir:
         paths.append(os.path.join(args.harvester_workdir, config.Harvester.pandajob_file))
     if 'HARVESTER_WORKDIR' in os.environ:
         paths.append(os.path.join(os.environ['HARVESTER_WORKDIR'], config.Harvester.pandajob_file))
 
     path = ""
     for _path in paths:
         if os.path.exists(_path):
             path = _path
             break
 
     if path == "":
         logger.info('did not find any local job definition file')
 
     return path
 
 
 def get_job_definition(args):
     """
     Get a job definition from a source (server or pre-placed local file).
 
     :param args: Pilot arguments (e.g. containing queue name, queuedata dictionary, etc).
     :return: job definition dictionary.
     """
 
     res = {}
     path = locate_job_definition(args)
 
     # should we run a normal 'real' job or a 'fake' job?
     if config.Pilot.pandajob == 'fake':
         logger.info('will use a fake PanDA job')
         res = get_fake_job()
     elif os.path.exists(path):
         logger.info('will read job definition from file %s', path)
         res = get_job_definition_from_file(path, args.harvester)
     else:
         if args.harvester and args.harvester_submitmode.lower() == 'push':
             pass  # local job definition file not found (go to sleep)
         else:
             logger.info('will download job definition from server')
             res = get_job_definition_from_server(args)
 
     return res
 
 
 def now():
     """
     Return the current epoch as a UTF-8 encoded string.
     :return: current time as encoded string
     """
     return str(time.time()).encode('utf-8')
 
 
 def get_fake_job(input=True):
     """
     Return a job definition for internal pilot testing.
     Note: this function is only used for testing purposes. The job definitions below are ATLAS specific.
 
     :param input: Boolean, set to False if no input files are wanted
     :return: job definition (dictionary).
     """
 
     res = None
 
     # create hashes
     hash = hashlib.md5()
     hash.update(now())
     log_guid = hash.hexdigest()
     hash.update(now())
     guid = hash.hexdigest()
     hash.update(now())
     job_name = hash.hexdigest()
 
     if config.Pilot.testjobtype == 'production':
         logger.info('creating fake test production job definition')
         res = {'jobsetID': 'NULL',
                'logGUID': log_guid,
                'cmtConfig': 'x86_64-slc6-gcc48-opt',
                'prodDBlocks': 'user.mlassnig:user.mlassnig.pilot.test.single.hits',
                'dispatchDBlockTokenForOut': 'NULL,NULL',
                'destinationDBlockToken': 'NULL,NULL',
                'destinationSE': 'AGLT2_TEST',
                'realDatasets': job_name,
                'prodUserID': 'no_one',
                'GUID': guid,
                'realDatasetsIn': 'user.mlassnig:user.mlassnig.pilot.test.single.hits',
                'nSent': 0,
                'cloud': 'US',
                'StatusCode': 0,
                'homepackage': 'AtlasProduction/20.1.4.14',
                'inFiles': 'HITS.06828093._000096.pool.root.1',
                'processingType': 'pilot-ptest',
                'ddmEndPointOut': 'UTA_SWT2_DATADISK,UTA_SWT2_DATADISK',
                'fsize': '94834717',
                'fileDestinationSE': 'AGLT2_TEST,AGLT2_TEST',
                'scopeOut': 'panda',
                'minRamCount': 0,
                'jobDefinitionID': 7932,
                'maxWalltime': 'NULL',
                'scopeLog': 'panda',
                'transformation': 'Reco_tf.py',
                'maxDiskCount': 0,
                'coreCount': 1,
                'prodDBlockToken': 'NULL',
                'transferType': 'NULL',
                'destinationDblock': job_name,
                'dispatchDBlockToken': 'NULL',
                'jobPars': '--maxEvents=1 --inputHITSFile HITS.06828093._000096.pool.root.1 --outputRDOFile RDO_%s.root' % job_name,
                'attemptNr': 0,
                'swRelease': 'Atlas-20.1.4',
                'nucleus': 'NULL',
                'maxCpuCount': 0,
                'outFiles': 'RDO_%s.root,%s.job.log.tgz' % (job_name, job_name),
                'currentPriority': 1000,
                'scopeIn': 'mc15_13TeV',
                'PandaID': '0',
                'sourceSite': 'NULL',
                'dispatchDblock': 'NULL',
                'prodSourceLabel': 'ptest',
                'checksum': 'ad:5d000974',
                'jobName': job_name,
                'ddmEndPointIn': 'UTA_SWT2_DATADISK',
                'taskID': 'NULL',
                'logFile': '%s.job.log.tgz' % job_name}
     elif config.Pilot.testjobtype == 'user':
         logger.info('creating fake test user job definition')
         res = {'jobsetID': 'NULL',
                'logGUID': log_guid,
                'cmtConfig': 'x86_64-slc6-gcc49-opt',
                'prodDBlocks': 'data15_13TeV:data15_13TeV.00276336.physics_Main.merge.AOD.r7562_p2521_tid07709524_00',
                'dispatchDBlockTokenForOut': 'NULL,NULL',
                'destinationDBlockToken': 'NULL,NULL',
                'destinationSE': 'ANALY_SWT2_CPB',
                'realDatasets': job_name,
                'prodUserID': 'None',
                'GUID': guid,
                'realDatasetsIn': 'data15_13TeV:data15_13TeV.00276336.physics_Main.merge.AOD.r7562_p2521_tid07709524_00',
                'nSent': '0',
                'cloud': 'US',
                'StatusCode': 0,
                'homepackage': 'AnalysisTransforms-AtlasDerivation_20.7.6.4',
                'inFiles': 'AOD.07709524._000050.pool.root.1',
                'processingType': 'pilot-ptest',
                'ddmEndPointOut': 'SWT2_CPB_SCRATCHDISK,SWT2_CPB_SCRATCHDISK',
                'fsize': '1564780952',
                'fileDestinationSE': 'ANALY_SWT2_CPB,ANALY_SWT2_CPB',
                'scopeOut': 'user.gangarbt',
                'minRamCount': '0',
                'jobDefinitionID': '9445',
                'maxWalltime': 'NULL',
                'scopeLog': 'user.gangarbt',
                'transformation': 'http://pandaserver.cern.ch:25080/trf/user/runAthena-00-00-11',
                'maxDiskCount': '0',
                'coreCount': '1',
                'prodDBlockToken': 'NULL',
                'transferType': 'NULL',
                'destinationDblock': job_name,
                'dispatchDBlockToken': 'NULL',
                'jobPars': '-a sources.20115461.derivation.tgz -r ./ -j "Reco_tf.py '
                           '--inputAODFile AOD.07709524._000050.pool.root.1 --outputDAODFile test.pool.root '
                           '--reductionConf HIGG3D1" -i "[\'AOD.07709524._000050.pool.root.1\']" -m "[]" -n "[]" --trf'
                           ' --useLocalIO --accessmode=copy -o '
                           '"{\'IROOT\': [(\'DAOD_HIGG3D1.test.pool.root\', \'%s.root\')]}" '
                           '--sourceURL https://aipanda012.cern.ch:25443' % (job_name),
                'attemptNr': '0',
                'swRelease': 'Atlas-20.7.6',
                'nucleus': 'NULL',
                'maxCpuCount': '0',
                'outFiles': '%s.root,%s.job.log.tgz' % (job_name, job_name),
                'currentPriority': '1000',
                'scopeIn': 'data15_13TeV',
                'PandaID': '0',
                'sourceSite': 'NULL',
                'dispatchDblock': 'data15_13TeV:data15_13TeV.00276336.physics_Main.merge.AOD.r7562_p2521_tid07709524_00',
                'prodSourceLabel': 'ptest',
                'checksum': 'ad:b11f45a7',
                'jobName': job_name,
                'ddmEndPointIn': 'SWT2_CPB_SCRATCHDISK',
                'taskID': 'NULL',
                'logFile': '%s.job.log.tgz' % job_name}
     else:
         logger.warning('unknown test job type: %s', config.Pilot.testjobtype)
 
     if res:
         if not input:
             res['inFiles'] = 'NULL'
             res['GUID'] = 'NULL'
             res['scopeIn'] = 'NULL'
             res['fsize'] = 'NULL'
             res['realDatasetsIn'] = 'NULL'
             res['checksum'] = 'NULL'
 
         if config.Pilot.testtransfertype == "NULL" or config.Pilot.testtransfertype == 'direct':
             res['transferType'] = config.Pilot.testtransfertype
         else:
             logger.warning('unknown test transfer type: %s (ignored)', config.Pilot.testtransfertype)
 
         if config.Pilot.testjobcommand == 'sleep':
             res['transformation'] = 'sleep'
             res['jobPars'] = '1'
             res['inFiles'] = ''
             res['outFiles'] = ''
 
         # convert to unicode for Python 2
         try:  # in case some later version of Python 3 has problems using u'' (seems ok with 3.7 at least)
             if not is_python3():
                 _res = {}
                 for entry in res:
                     if type(res[entry]) is str:
                         _res[u'%s' % entry] = u'%s' % res[entry]
                     else:
                         _res[u'%s' % entry] = res[entry]
                 res = _res
         except Exception:
             pass
     return res
 
 
 def get_job_retrieval_delay(harvester):
     """
     Return the proper delay between job retrieval attempts.
     In Harvester mode, the pilot will look once per second for a job definition file.
 
     :param harvester: True if Harvester is being used (determined from args.harvester), otherwise False
     :return: sleep (s)
     """
 
     return 1 if harvester else 60
 
 
 def retrieve(queues, traces, args):  # noqa: C901
     """
     Retrieve all jobs from a source.
 
     The job definition is a json dictionary that is either present in the launch
     directory (preplaced) or downloaded from a server specified by `args.url`.
 
     The function retrieves the job definition from the proper source and places
     it in the `queues.jobs` queue.
 
     WARNING: this function is nearly too complex. Be careful with adding more lines as flake8 will fail it.
 
     :param queues: internal queues for job handling.
     :param traces: tuple containing internal pilot states.
     :param args: Pilot arguments (e.g. containing queue name, queuedata dictionary, etc).
     :raises PilotException: if create_job fails (e.g. because queuedata could not be downloaded).
     :return:
     """
 
     timefloor = infosys.queuedata.timefloor
     starttime = time.time()
 
     jobnumber = 0  # number of downloaded jobs
     getjob_requests = 0
     getjob_failures = 0
     print_node_info()
 
     while not args.graceful_stop.is_set():
 
         time.sleep(0.5)
         getjob_requests += 1
 
         if not proceed_with_getjob(timefloor, starttime, jobnumber, getjob_requests, args.getjob_requests,
                                    args.update_server, args.harvester_submitmode, args.harvester, args.verify_proxy, traces):
             # do not set graceful stop if pilot has not finished sending the final job update
             # i.e. wait until SERVER_UPDATE is DONE_FINAL
             check_for_final_server_update(args.update_server)
             args.graceful_stop.set()
             break
 
         # store time stamp
         time_pre_getjob = time.time()
 
         # get a job definition from a source (file or server)
         res = get_job_definition(args)
         logger.info('job definition = %s', str(res))
 
         if res is None:
             logger.fatal('fatal error in job download loop - cannot continue')
             # do not set graceful stop if pilot has not finished sending the final job update
             # i.e. wait until SERVER_UPDATE is DONE_FINAL
             check_for_final_server_update(args.update_server)
             args.graceful_stop.set()
             break
 
         if not res:
             getjob_failures += 1
             if getjob_failures >= args.getjob_failures:
                 logger.warning('did not get a job -- max number of job request failures reached: %d', getjob_failures)
                 args.graceful_stop.set()
                 break
 
             delay = get_job_retrieval_delay(args.harvester)
             if not args.harvester:
                 logger.warning('did not get a job -- sleep %d s and repeat', delay)
             for _ in range(delay):
                 if args.graceful_stop.is_set():
                     break
                 time.sleep(1)
         else:
             # it seems the PanDA server returns StatusCode as an int, but the aCT returns it as a string
             # note: StatusCode keyword is not available in job definition files from Harvester (not needed)
             if 'StatusCode' in res and res['StatusCode'] != '0' and res['StatusCode'] != 0:
                 getjob_failures += 1
                 if getjob_failures >= args.getjob_failures:
                     logger.warning('did not get a job -- max number of job request failures reached: %d',
                                    getjob_failures)
                     args.graceful_stop.set()
                     break
 
                 logger.warning('did not get a job -- sleep 60s and repeat -- status: %s', res['StatusCode'])
                 for i in range(60):
                     if args.graceful_stop.is_set():
                         break
                     time.sleep(1)
             else:
                 # create the job object out of the raw dispatcher job dictionary
                 try:
                     job = create_job(res, args.queue)
                 except PilotException as error:
                     raise error
                 #else:
                     # verify the job status on the server
                     #try:
                     #    job_status, job_attempt_nr, job_status_code = get_job_status_from_server(job.jobid, args.url, args.port)
                     #    if job_status == "running":
                     #        pilot_error_diag = "job %s is already running elsewhere - aborting" % job.jobid
                     #        logger.warning(pilot_error_diag)
                     #        raise JobAlreadyRunning(pilot_error_diag)
                     #except Exception as error:
                     #    logger.warning("%s", error)
                 # write time stamps to pilot timing file
                 # note: PILOT_POST_GETJOB corresponds to START_TIME in Pilot 1
                 add_to_pilot_timing(job.jobid, PILOT_PRE_GETJOB, time_pre_getjob, args)
                 add_to_pilot_timing(job.jobid, PILOT_POST_GETJOB, time.time(), args)
 
                 # add the job definition to the jobs queue and increase the job counter,
                 # and wait until the job has finished
                 put_in_queue(job, queues.jobs)
 
                 jobnumber += 1
                 while not args.graceful_stop.is_set():
                     if has_job_completed(queues, args):
                         # purge queue(s) that retains job object
                         purge_queue(queues.finished_data_in)
 
                         args.job_aborted.clear()
                         args.abort_job.clear()
                         logger.info('ready for new job')
 
                         # re-establish logging
                         logging.info('pilot has finished for previous job - re-establishing logging')
                         logging.handlers = []
                         logging.shutdown()
                         establish_logging(debug=args.debug, nopilotlog=args.nopilotlog)
                         pilot_version_banner()
                         getjob_requests = 0
                         add_to_pilot_timing('1', PILOT_MULTIJOB_START_TIME, time.time(), args)
                         break
                     time.sleep(0.5)
 
     # proceed to set the job_aborted flag?
     if threads_aborted():
         logger.debug('will proceed to set job_aborted')
         args.job_aborted.set()
     else:
         logger.debug('will not set job_aborted yet')
 
     logger.debug('[job] retrieve thread has finished')
 
 
 def print_node_info():
     """
     Print information about the local node to the log.
 
     :return:
     """
 
     if is_virtual_machine():
         logger.info("pilot is running in a virtual machine")
     else:
         logger.info("pilot is not running in a virtual machine")
 
 
 def create_job(dispatcher_response, queue):
     """
     Create a job object out of the dispatcher response.
 
     :param dispatcher_response: raw job dictionary from the dispatcher.
     :param queue: queue name (string).
     :return: job object
     """
 
     # initialize (job specific) InfoService instance
 
     job = JobData(dispatcher_response)
 
     jobinfosys = InfoService()
     jobinfosys.init(queue, infosys.confinfo, infosys.extinfo, JobInfoProvider(job))
     job.init(infosys)
 
     #job.workdir = os.getcwd()
 
     logger.info('received job: %s (sleep until the job has finished)', job.jobid)
     logger.info('job details: \n%s', job)
 
     # payload environment wants the PANDAID to be set, also used below
     os.environ['PANDAID'] = job.jobid
 
     return job
 
 
 def has_job_completed(queues, args):
     """
     Has the current job completed (finished or failed)?
     Note: the job object was extracted from monitored_payloads queue before this function was called.
 
     :param queues: Pilot queues object.
     :return: True is the payload has finished or failed
     """
 
     # check if the job has finished
     try:
         job = queues.completed_jobs.get(block=True, timeout=1)
     except queue.Empty:
         # logger.info("(job still running)")
         pass
     else:
         make_job_report(job)
         cmd = 'ls -lF %s' % os.environ.get('PILOT_HOME')
         logger.debug('%s:\n', cmd)
         _, stdout, _ = execute(cmd)
         logger.debug(stdout)
 
         queue_report(queues)
         job.reset_errors()
         logger.info("job %s has completed (purged errors)", job.jobid)
 
         # cleanup of any remaining processes
         if job.pid:
             job.zombies.append(job.pid)
         cleanup(job, args)
 
         return True
 
     # is there anything in the finished_jobs queue?
     #finished_queue_snapshot = list(queues.finished_jobs.queue)
     #peek = [obj for obj in finished_queue_snapshot if jobid == obj.jobid]
     #if peek:
     #    logger.info("job %s has completed (finished)", jobid)
     #    return True
 
     # is there anything in the failed_jobs queue?
     #failed_queue_snapshot = list(queues.failed_jobs.queue)
     #peek = [obj for obj in failed_queue_snapshot if jobid == obj.jobid]
     #if peek:
     #    logger.info("job %s has completed (failed)", jobid)
     #    return True
 
     return False
 
 
 def get_job_from_queue(queues, state):
     """
     Check if the job has finished or failed and if so return it.
 
     :param queues: pilot queues.
     :param state: job state (e.g. finished/failed) (string).
     :return: job object.
     """
     try:
         if state == "finished":
             job = queues.finished_jobs.get(block=True, timeout=1)
         elif state == "failed":
             job = queues.failed_jobs.get(block=True, timeout=1)
         else:
             job = None
     except queue.Empty:
         # logger.info("(job still running)")
         job = None
     else:
         # make sure that state=failed
         set_pilot_state(job=job, state=state)
         logger.info("job %s has state=%s", job.jobid, job.state)
 
     return job
 
 
 def is_queue_empty(queues, queue):
     """
     Check if the given queue is empty (without pulling).
 
     :param queues: pilot queues object.
     :param queue: queue name (string).
     :return: True if queue is empty, False otherwise
     """
 
     status = False
     if queue in queues._fields:
         _queue = getattr(queues, queue)
         jobs = list(_queue.queue)
         if len(jobs) > 0:
             logger.info('queue %s not empty: found %d job(s)', queue, len(jobs))
         else:
             logger.info('queue %s is empty', queue)
             status = True
     else:
         logger.warning('queue %s not present in %s', queue, queues._fields)
 
     return status
 
 
 def order_log_transfer(queues, job):
     """
     Order a log transfer for a failed job.
 
     :param queues: pilot queues object.
     :param job: job object.
     :return:
     """
 
     # add the job object to the data_out queue to have it staged out
     job.stageout = 'log'  # only stage-out log file
     #set_pilot_state(job=job, state='stageout')
     put_in_queue(job, queues.data_out)
 
     logger.debug('job added to data_out queue')
 
     # wait for the log transfer to finish
     n = 0
     nmax = 60
     while n < nmax:
         # refresh the log_transfer since it might have changed
         log_transfer = job.get_status('LOG_TRANSFER')
         logger.info('waiting for log transfer to finish (#%d/#%d): %s', n + 1, nmax, log_transfer)
         if is_queue_empty(queues, 'data_out') and \
                 (log_transfer == LOG_TRANSFER_DONE or log_transfer == LOG_TRANSFER_FAILED):  # set in data component
             logger.info('stage-out of log has completed')
             break
         else:
             if log_transfer == LOG_TRANSFER_IN_PROGRESS:  # set in data component, job object is singleton
                 logger.info('log transfer is in progress')
             time.sleep(2)
             n += 1
 
     logger.info('proceeding with server update (n=%d)', n)
 
 
 def wait_for_aborted_job_stageout(args, queues, job):
     """
     Wait for stage-out to finish for aborted job.
 
     :param args: pilot args object.
     :param queues: pilot queues object.
     :param job: job object.
     :return:
     """
 
     # if the pilot received a kill signal, how much time has passed since the signal was intercepted?
     try:
         time_since_kill = get_time_since('1', PILOT_KILL_SIGNAL, args)
         was_killed = was_pilot_killed(args.timing)
         if was_killed:
             logger.info('%d s passed since kill signal was intercepted - make sure that stage-out has finished', time_since_kill)
     except Exception as error:
         logger.warning('exception caught: %s', error)
         time_since_kill = 60
     else:
         if time_since_kill > 60 or time_since_kill < 0:  # fail-safe
             logger.warning('reset time_since_kill to 60 since value is out of allowed limits')
             time_since_kill = 60
 
     # if stage-out has not finished, we need to wait (less than two minutes or the batch system will issue
     # a hard SIGKILL)
     max_wait_time = 2 * 60 - time_since_kill - 5
     logger.debug('using max_wait_time = %d s', max_wait_time)
     t0 = time.time()
     while time.time() - t0 < max_wait_time:
         if job in queues.finished_data_out.queue or job in queues.failed_data_out.queue:
             logger.info('stage-out has finished, proceed with final server update')
             break
         else:
             time.sleep(0.5)
 
     logger.info('proceeding with final server update')
 
 
 def get_job_status(job, key):
     """
     Wrapper function around job.get_status().
     If key = 'LOG_TRANSFER' but job object is not defined, the function will return value = LOG_TRANSFER_NOT_DONE.
 
     :param job: job object.
     :param key: key name (string).
     :return: value (string).
     """
 
     value = ""
     if job:
         value = job.get_status(key)
     else:
         if key == 'LOG_TRANSFER':
             value = LOG_TRANSFER_NOT_DONE
 
     return value
 
 
 def queue_monitor(queues, traces, args):  # noqa: C901
     """
     Monitoring of queues.
     This function monitors queue activity, specifically if a job has finished or failed and then reports to the server.
 
     :param queues: internal queues for job handling.
     :param traces: tuple containing internal pilot states.
     :param args: Pilot arguments (e.g. containing queue name, queuedata dictionary, etc).
     :return:
     """
 
     # scan queues until at least one queue has a job object. abort if it takes too long time
     if not scan_for_jobs(queues):
         logger.warning('queues are still empty of jobs - will begin queue monitoring anyway')
 
     job = None
     while True:  # will abort when graceful_stop has been set or if enough time has passed after kill signal
         time.sleep(1)
 
         if traces.pilot['command'] == 'abort':
             logger.warning('job queue monitor received an abort instruction')
             args.graceful_stop.set()
 
         # abort in case graceful_stop has been set, and less than 30 s has passed since MAXTIME was reached (if set)
         # (abort at the end of the loop)
         abort_thread = should_abort(args, label='job:queue_monitor')
         if abort_thread and os.environ.get('PILOT_WRAP_UP', '') == 'NORMAL':
             pause_queue_monitor(20)
 
         # check if the job has finished
         imax = 20
         i = 0
         while i < imax and os.environ.get('PILOT_WRAP_UP', '') == 'NORMAL':
             job = get_finished_or_failed_job(args, queues)
             if job:
                 logger.debug('returned job has state=%s', job.state)
                 #if job.state == 'failed':
                 #    logger.warning('will abort failed job (should prepare for final server update)')
                 break
             i += 1
             state = get_pilot_state()  # the job object is not available, but the state is also kept in PILOT_JOB_STATE
             if state != 'stage-out':
                 # logger.info("no need to wait since job state=\'%s\'", state)
                 break
             pause_queue_monitor(1) if not abort_thread else pause_queue_monitor(10)
 
         # job has not been defined if it's still running
         if not job and not abort_thread:
             continue
 
         completed_jobids = queues.completed_jobids.queue if queues.completed_jobids else []
         if job and job.jobid not in completed_jobids:
             logger.info("preparing for final server update for job %s in state=\'%s\'", job.jobid, job.state)
 
             if args.job_aborted.is_set():
                 # wait for stage-out to finish for aborted job
                 wait_for_aborted_job_stageout(args, queues, job)
 
             # send final server update
             update_server(job, args)
 
             # we can now stop monitoring this job, so remove it from the monitored_payloads queue and add it to the
             # completed_jobs queue which will tell retrieve() that it can download another job
             try:
                 _job = queues.monitored_payloads.get(block=True, timeout=1)
             except queue.Empty:
                 logger.warning('failed to dequeue job: queue is empty (did job fail before job monitor started?)')
                 make_job_report(job)
             else:
                 logger.debug('job %s was dequeued from the monitored payloads queue', _job.jobid)
                 # now ready for the next job (or quit)
                 put_in_queue(job.jobid, queues.completed_jobids)
 
                 put_in_queue(job, queues.completed_jobs)
                 del _job
                 logger.debug('tmp job object deleted')
 
         if abort_thread:
             break
 
     # proceed to set the job_aborted flag?
     if threads_aborted():
         logger.debug('will proceed to set job_aborted')
         args.job_aborted.set()
     else:
         logger.debug('will not set job_aborted yet')
 
     logger.debug('[job] queue monitor thread has finished')
 
 
 def update_server(job, args):
     """
     Update the server (wrapper for send_state() that also prepares the metadata).
 
     :param job: job object.
     :param args: pilot args object.
     :return:
     """
 
     # user specific actions
     pilot_user = os.environ.get('PILOT_USER', 'generic').lower()
     user = __import__('pilot.user.%s.common' % pilot_user, globals(), locals(), [pilot_user], 0)  # Python 2/3
     metadata = user.get_metadata(job.workdir)
     try:
         user.update_server(job)
     except Exception as error:
         logger.warning('exception caught in update_server(): %s', error)
     if job.fileinfo:
         send_state(job, args, job.state, xml=dumps(job.fileinfo), metadata=metadata)
     else:
         send_state(job, args, job.state, metadata=metadata)
 
 
 def pause_queue_monitor(delay):
     """
     Pause the queue monitor to let log transfer complete.
     Note: this function should use globally available object. Use sleep for now.
     :param delay: sleep time in seconds (int).
     :return:
     """
 
     logger.warning('since job:queue_monitor is responsible for sending job updates, we sleep for %d s', delay)
     time.sleep(delay)
 
 
 def get_finished_or_failed_job(args, queues):
     """
     Check if the job has either finished or failed and if so return it.
     If failed, order a log transfer. If the job is in state 'failed' and abort_job is set, set job_aborted.
 
     :param args: pilot args object.
     :param queues: pilot queues object.
     :return: job object.
     """
 
     job = get_job_from_queue(queues, "finished")
     if job:
         # logger.debug('get_finished_or_failed_job: job has finished')
         pass
     else:
         # logger.debug('check_job: job has not finished')
         job = get_job_from_queue(queues, "failed")
         if job:
             logger.debug('get_finished_or_failed_job: job has failed')
             job.state = 'failed'
             args.job_aborted.set()
 
             # get the current log transfer status
             log_transfer = get_job_status(job, 'LOG_TRANSFER')
             if log_transfer == LOG_TRANSFER_NOT_DONE:
                 # order a log transfer for a failed job
                 order_log_transfer(queues, job)
 
     # check if the job has failed
     if job and job.state == 'failed':
         # set job_aborted in case of kill signals
         if args.abort_job.is_set():
             logger.warning('queue monitor detected a set abort_job (due to a kill signal)')
             # do not set graceful stop if pilot has not finished sending the final job update
             # i.e. wait until SERVER_UPDATE is DONE_FINAL
             #check_for_final_server_update(args.update_server)
             #args.job_aborted.set()
 
     return job
 
 
 def get_heartbeat_period(debug=False):
     """
     Return the proper heartbeat period, as determined by normal or debug mode.
     In normal mode, the heartbeat period is 30*60 s, while in debug mode it is 5*60 s. Both values are defined in the
     config file.
 
     :param debug: Boolean, True for debug mode. False otherwise.
     :return: heartbeat period (int).
     """
 
     try:
         return int(config.Pilot.heartbeat if not debug else config.Pilot.debug_heartbeat)
     except Exception as error:
         logger.warning('bad config data for heartbeat period: %s (will use default 1800 s)', error)
         return 1800
 
 
 def check_for_abort_job(args, caller=''):
     """
     Check if args.abort_job.is_set().
 
     :param args: Pilot arguments (e.g. containing queue name, queuedata dictionary, etc).
     :param caller: function name of caller (string).
     :return: Boolean, True if args_job.is_set()
     """
     abort_job = False
     if args.abort_job.is_set():
         logger.warning('%s detected an abort_job request (signal=%s)', caller, args.signal)
         logger.warning('in case pilot is running more than one job, all jobs will be aborted')
         abort_job = True
 
     return abort_job
 
 
 def interceptor(queues, traces, args):
     """
     MOVE THIS TO INTERCEPTOR.PY; TEMPLATE FOR THREADS
 
     :param queues: internal queues for job handling.
     :param traces: tuple containing internal pilot states.
     :param args: Pilot arguments (e.g. containing queue name, queuedata dictionary, etc).
     :return:
     """
 
     # overall loop counter (ignoring the fact that more than one job may be running)
     n = 0
     while not args.graceful_stop.is_set():
         time.sleep(0.1)
 
         # abort in case graceful_stop has been set, and less than 30 s has passed since MAXTIME was reached (if set)
         # (abort at the end of the loop)
         abort = should_abort(args, label='job:interceptor')
 
         # check for any abort_job requests
         abort_job = check_for_abort_job(args, caller='interceptor')
         if not abort_job:
             # peek at the jobs in the validated_jobs queue and send the running ones to the heartbeat function
             jobs = queues.monitored_payloads.queue
             if jobs:
                 for _ in range(len(jobs)):
                     logger.info('interceptor loop %d: looking for communication file', n)
             time.sleep(30)
 
         n += 1
 
         if abort or abort_job:
             break
 
     # proceed to set the job_aborted flag?
     if threads_aborted():
         logger.debug('will proceed to set job_aborted')
         args.job_aborted.set()
     else:
         logger.debug('will not set job_aborted yet')
 
     logger.debug('[job] interceptor thread has finished')
 
 
 def fast_monitor_tasks(job):
     """
     Perform user specific fast monitoring tasks.
 
     :param job: job object.
     :return: exit code (int).
     """
 
     exit_code = 0
 
     pilot_user = os.environ.get('PILOT_USER', 'generic').lower()
     user = __import__('pilot.user.%s.monitoring' % pilot_user, globals(), locals(), [pilot_user], 0)  # Python 2/3
     try:
         exit_code = user.fast_monitor_tasks(job)
     except Exception as exc:
         logger.warning('caught exception: %s', exc)
 
     return exit_code
 
 
 def fast_job_monitor(queues, traces, args):
     """
     Fast monitoring of job parameters.
 
     This function can be used for monitoring processes below the one minute threshold of the normal job_monitor thread.
 
     :param queues: internal queues for job handling.
     :param traces: tuple containing internal pilot states.
     :param args: Pilot arguments (e.g. containing queue name, queuedata dictionary, etc).
     :return:
     """
 
     # peeking and current time; peeking_time gets updated if and when jobs are being monitored, update_time is only
     # used for sending the heartbeat and is updated after a server update
     #peeking_time = int(time.time())
     #update_time = peeking_time
 
     # end thread immediately, unless fast monitoring is required
     if not args.use_realtime_logging:
         logger.warning('fast monitoring not required - ending thread')
         return
 
     while not args.graceful_stop.is_set():
         time.sleep(10)
 
         # abort in case graceful_stop has been set, and less than 30 s has passed since MAXTIME was reached (if set)
         # (abort at the end of the loop)
         abort = should_abort(args, label='job:fast_job_monitor')
         if abort:
             break
 
         if traces.pilot.get('command') == 'abort':
             logger.warning('fast job monitor received an abort command')
             break
 
         # check for any abort_job requests
         abort_job = check_for_abort_job(args, caller='fast job monitor')
         if abort_job:
             break
         else:
             # peek at the jobs in the validated_jobs queue and send the running ones to the heartbeat function
             jobs = queues.monitored_payloads.queue
             if jobs:
                 for i in range(len(jobs)):
                     #current_id = jobs[i].jobid
                     if jobs[i].state == 'finished' or jobs[i].state == 'failed':
                         logger.info('will abort job monitoring soon since job state=%s (job is still in queue)', jobs[i].state)
                         break
 
                 # perform the monitoring tasks
                 exit_code = fast_monitor_tasks(jobs[i])
                 if exit_code:
                     logger.debug('fast monitoring reported an error: %d', exit_code)
 
     # proceed to set the job_aborted flag?
     if threads_aborted():
         logger.debug('will proceed to set job_aborted')
         args.job_aborted.set()
     else:
         logger.debug('will not set job_aborted yet')
 
     logger.debug('[job] fast job monitor thread has finished')
 
 
 def job_monitor(queues, traces, args):  # noqa: C901
     """
     Monitoring of job parameters.
     This function monitors certain job parameters, such as job looping, at various time intervals. The main loop
     is executed once a minute, while individual verifications may be executed at any time interval (>= 1 minute). E.g.
     looping jobs are checked once per ten minutes (default) and the heartbeat is send once per 30 minutes. Memory
     usage is checked once a minute.
 
     :param queues: internal queues for job handling.
     :param traces: tuple containing internal pilot states.
     :param args: Pilot arguments (e.g. containing queue name, queuedata dictionary, etc).
     :return:
     """
 
     # initialize the monitoring time object
     mt = MonitoringTime()
 
     # peeking and current time; peeking_time gets updated if and when jobs are being monitored, update_time is only
     # used for sending the heartbeat and is updated after a server update
     peeking_time = int(time.time())
     update_time = peeking_time
 
     # overall loop counter (ignoring the fact that more than one job may be running)
     n = 0
     while not args.graceful_stop.is_set():
         time.sleep(0.5)
 
         # abort in case graceful_stop has been set, and less than 30 s has passed since MAXTIME was reached (if set)
         # (abort at the end of the loop)
         abort = should_abort(args, label='job:job_monitor')
 
         if traces.pilot.get('command') == 'abort':
             logger.warning('job monitor received an abort command')
 
         # check for any abort_job requests
         abort_job = check_for_abort_job(args, caller='job monitor')
         if not abort_job:
             if not queues.current_data_in.empty():
                 # make sure to send heartbeat regularly if stage-in takes a long time
                 jobs = queues.current_data_in.queue
                 if jobs:
                     for i in range(len(jobs)):
                         # send heartbeat if it is time (note that the heartbeat function might update the job object, e.g.
                         # by turning on debug mode, ie we need to get the heartbeat period in case it has changed)
                         update_time = send_heartbeat_if_time(jobs[i], args, update_time)
 
                         # note: when sending a state change to the server, the server might respond with 'tobekilled'
                         try:
                             jobs[i]
                         except Exception as error:
                             logger.warning('detected stale jobs[i] object in job_monitor: %s', error)
                         else:
                             if jobs[i].state == 'failed':
                                 logger.warning('job state is \'failed\' - order log transfer and abort job_monitor() (1)')
                                 jobs[i].stageout = 'log'  # only stage-out log file
                                 put_in_queue(jobs[i], queues.data_out)
 
                     # sleep for a while if stage-in has not completed
                     time.sleep(1)
                     continue
             elif queues.finished_data_in.empty():
                 # sleep for a while if stage-in has not completed
                 time.sleep(1)
                 continue
 
             time.sleep(60)
 
         # peek at the jobs in the validated_jobs queue and send the running ones to the heartbeat function
         jobs = queues.monitored_payloads.queue
         if jobs:
             # update the peeking time
             peeking_time = int(time.time())
             for i in range(len(jobs)):
                 current_id = jobs[i].jobid
                 logger.info('monitor loop #%d: job %d:%s is in state \'%s\'', n, i, current_id, jobs[i].state)
                 if jobs[i].state == 'finished' or jobs[i].state == 'failed':
                     logger.info('will abort job monitoring soon since job state=%s (job is still in queue)', jobs[i].state)
                     break
 
                 # perform the monitoring tasks
                 exit_code, diagnostics = job_monitor_tasks(jobs[i], mt, args)
                 if exit_code != 0:
                     if exit_code == errors.NOVOMSPROXY:
                         logger.warning('VOMS proxy has expired - keep monitoring job')
                     elif exit_code == errors.KILLPAYLOAD:
                         jobs[i].piloterrorcodes, jobs[i].piloterrordiags = errors.add_error_code(exit_code)
                         logger.debug('killing payload process')
                         kill_process(jobs[i].pid)
                         break
                     else:
                         try:
                             fail_monitored_job(jobs[i], exit_code, diagnostics, queues, traces)
                         except Exception as error:
                             logger.warning('(1) exception caught: %s (job id=%s)', error, current_id)
                         break
 
                 # run this check again in case job_monitor_tasks() takes a long time to finish (and the job object
                 # has expired in the mean time)
                 try:
                     _job = jobs[i]
                 except Exception:
                     logger.info('aborting job monitoring since job object (job id=%s) has expired', current_id)
                     break
 
                 # send heartbeat if it is time (note that the heartbeat function might update the job object, e.g.
                 # by turning on debug mode, ie we need to get the heartbeat period in case it has changed)
                 try:
                     update_time = send_heartbeat_if_time(_job, args, update_time)
                 except Exception as error:
                     logger.warning('(2) exception caught: %s (job id=%s)', error, current_id)
                     break
                 else:
                     # note: when sending a state change to the server, the server might respond with 'tobekilled'
                     if _job.state == 'failed':
                         logger.warning('job state is \'failed\' - order log transfer and abort job_monitor() (2)')
                         _job.stageout = 'log'  # only stage-out log file
                         put_in_queue(_job, queues.data_out)
                         abort = True
                         break
 
         elif os.environ.get('PILOT_JOB_STATE') == 'stagein':
             logger.info('job monitoring is waiting for stage-in to finish')
         else:
             # check the waiting time in the job monitor. set global graceful_stop if necessary
             check_job_monitor_waiting_time(args, peeking_time, abort_override=abort_job)
 
         n += 1
 
         if abort or abort_job:
             break
 
     # proceed to set the job_aborted flag?
     if threads_aborted():
         logger.debug('will proceed to set job_aborted')
         args.job_aborted.set()
     else:
         logger.debug('will not set job_aborted yet')
 
     logger.debug('[job] job monitor thread has finished')
 
 
 def send_heartbeat_if_time(job, args, update_time):
     """
     Send a heartbeat to the server if it is time to do so.
 
     :param job: job object.
     :param args: args object.
     :param update_time: last update time (from time.time()).
     :return: possibly updated update_time (from time.time()).
     """
 
     if int(time.time()) - update_time >= get_heartbeat_period(job.debug):
         if job.serverstate != 'finished' and job.serverstate != 'failed':
             send_state(job, args, 'running')
             update_time = int(time.time())
 
     return update_time
 
 
 def check_job_monitor_waiting_time(args, peeking_time, abort_override=False):
     """
     Check the waiting time in the job monitor.
     Set global graceful_stop if necessary.
 
     :param args: args object.
     :param peeking_time: time when monitored_payloads queue was peeked into (int).
     :return:
     """
 
     waiting_time = int(time.time()) - peeking_time
     msg = 'no jobs in monitored_payloads queue (waited for %d s)' % waiting_time
     if waiting_time > 60 * 60:
         abort = True
         msg += ' - aborting'
     else:
         abort = False
     if logger:
         logger.warning(msg)
     else:
         print(msg)
     if abort or abort_override:
         # do not set graceful stop if pilot has not finished sending the final job update
         # i.e. wait until SERVER_UPDATE is DONE_FINAL
         check_for_final_server_update(args.update_server)
         args.graceful_stop.set()
 
 
 def fail_monitored_job(job, exit_code, diagnostics, queues, traces):
     """
     Fail a monitored job.
 
     :param job: job object
     :param exit_code: exit code from job_monitor_tasks (int).
     :param diagnostics: pilot error diagnostics (string).
     :param queues: queues object.
     :param traces: traces object.
     :return:
     """
 
     set_pilot_state(job=job, state="failed")
     job.piloterrorcodes, job.piloterrordiags = errors.add_error_code(exit_code, msg=diagnostics)
     job.piloterrordiag = diagnostics
     traces.pilot['error_code'] = exit_code
     put_in_queue(job, queues.failed_payloads)
     logger.info('aborting job monitoring since job state=%s', job.state)
 
 
 def make_job_report(job):
     """
     Make a summary report for the given job.
     This function is called when the job has completed.
 
     :param job: job object.
     :return:
     """
 
     logger.info('')
     logger.info('job summary report')
     logger.info('--------------------------------------------------')
     logger.info('PanDA job id: %s', job.jobid)
     logger.info('task id: %s', job.taskid)
     n = len(job.piloterrorcodes)
     if n > 0:
         for i in range(n):
             logger.info('error %d/%d: %s: %s', i + 1, n, job.piloterrorcodes[i], job.piloterrordiags[i])
     else:
         logger.info('errors: (none)')
     if job.piloterrorcode != 0:
         logger.info('pilot error code: %d', job.piloterrorcode)
         logger.info('pilot error diag: %s', job.piloterrordiag)
     info = ""
     for key in job.status:
         info += key + " = " + job.status[key] + " "
     logger.info('status: %s', info)
     s = ""
     if job.is_analysis() and job.state != 'finished':
         s = '(user job is recoverable)' if errors.is_recoverable(code=job.piloterrorcode) else '(user job is not recoverable)'
     logger.info('pilot state: %s %s', job.state, s)
     logger.info('transexitcode: %d', job.transexitcode)
     logger.info('exeerrorcode: %d', job.exeerrorcode)
     logger.info('exeerrordiag: %s', job.exeerrordiag)
     logger.info('exitcode: %d', job.exitcode)
     logger.info('exitmsg: %s', job.exitmsg)
     logger.info('cpuconsumptiontime: %d %s', job.cpuconsumptiontime, job.cpuconsumptionunit)
     logger.info('nevents: %d', job.nevents)
     logger.info('neventsw: %d', job.neventsw)
     logger.info('pid: %s', job.pid)
     logger.info('pgrp: %s', str(job.pgrp))
     logger.info('corecount: %d', job.corecount)
     logger.info('event service: %s', str(job.is_eventservice))
     logger.info('sizes: %s', str(job.sizes))
     logger.info('--------------------------------------------------')
     logger.info('')