Remove arcwrapper and associated logic.

Author: albu-diku

mig/cgi-bin/arcresources.py

@@ -51,7 +51,7 @@
 from mig.shared.gridscript import clean_grid_stdin, \
     remove_jobrequest_pending_files, check_mrsl_files, requeue_job, \
     server_cleanup, load_queue, save_queue, load_schedule_cache, \
-    save_schedule_cache, arc_job_status, clean_arc_job
+    save_schedule_cache
 from mig.shared.notification import notify_user_thread
 from mig.shared.resadm import atomic_resource_exe_restart, put_exe_pgid
 from mig.shared.vgrid import job_fits_res_vgrid, validated_vgrid_list
@@ -167,38 +167,6 @@ def time_out_jobs(stop_event):
                         send_message_to_grid_script(grid_script_msg,
                                                     logger, configuration)
 
-                    elif job['UNIQUE_RESOURCE_NAME'] == 'ARC':
-                        if not configuration.arc_clusters:
-                            logger.error('ARC backend disabled - ignore %s' %
-                                         job)
-                            continue
-                        jobstatus = arc_job_status(job, configuration, logger)
-
-                        # take action if the job is failed or killed.
-                        # No action for a finished job, since other
-                        # machinery will be at work to update it
-
-                        if jobstatus in ['FINISHED', 'FAILED', 'KILLED']:
-                            logger.debug(
-                                'discovered %s job %s, clean it on the server'
-                                % (jobstatus, job['JOB_ID']))
-                            if jobstatus in ['FAILED', 'KILLED']:
-                                msg = '(failed inside ARC)'
-                            else:
-                                msg = None
-                            exec_job = executing_queue.dequeue_job_by_id(
-                                job['JOB_ID'])
-                            if exec_job:
-                                # job was still there, clean up here
-                                # (otherwise, someone else picked it up in
-                                # the meantime)
-                                clean_arc_job(exec_job, jobstatus, msg,
-                                              configuration, logger, False)
-                        else:
-                            logger.debug(
-                                'Status %s for ARC job %s, no action required'
-                                % (jobstatus, job['JOB_ID']))
-
     except Exception as err:
         logger.error('time_out_jobs: unexpected exception: %s' % err)
     logger.info('time_out_jobs: time out thread terminating')
@@ -446,66 +414,6 @@ def graceful_shutdown():
         dict_userjob['OWNER'] = user_id
         dict_userjob['MIGRATE_COUNT'] = "0"
 
-        # ARC jobs: directly submit, and put in executing_queue
-        if dict_userjob['JOBTYPE'] == 'arc':
-            if not configuration.arc_clusters:
-                logger.error('ARC backend disabled - ignore %s' %
-                             dict_userjob)
-                continue
-            logger.debug('ARC Job')
-            (arc_job, msg) = jobscriptgenerator.create_arc_job(
-                dict_userjob, configuration, logger)
-            if not arc_job:
-                # something has gone wrong
-                logger.error('Job NOT submitted (%s)' % msg)
-                # discard this job (as FAILED, including message)
-                # see gridscript::requeue_job for how to do this...
-
-                dict_userjob['STATUS'] = 'FAILED'
-                dict_userjob['FAILED_TIMESTAMP'] = time.gmtime()
-                # and create an execution history (basically empty)
-                hist = (
-                    {'QUEUED_TIMESTAMP': dict_userjob['QUEUED_TIMESTAMP'],
-                     'EXECUTING_TIMESTAMP': dict_userjob['FAILED_TIMESTAMP'],
-                     'FAILED_TIMESTAMP': dict_userjob['FAILED_TIMESTAMP'],
-                     'FAILED_MESSAGE': ('ARC Submission failed: %s' % msg),
-                     'UNIQUE_RESOURCE_NAME': 'ARC', })
-                dict_userjob['EXECUTION_HISTORY'] = [hist]
-
-                # should also notify the user (if requested)
-                # not implented for this branch.
-
-            else:
-                # all fine, job is now in some ARC queue
-                logger.debug('Job submitted (%s,%s)' %
-                             (arc_job['SESSIONID'], arc_job['ARCID']))
-                # set some job fields for job status retrieval, and
-                # put in exec.queue for job status queries and timeout
-                dict_userjob['SESSIONID'] = arc_job['SESSIONID']
-                # abuse these two fields,
-                # expected by timeout thread to be there anyway
-                dict_userjob['UNIQUE_RESOURCE_NAME'] = 'ARC'
-                dict_userjob['EXE'] = arc_job['ARCID']
-
-                # this one is used by the timeout thread as well
-                # We put in a wild guess, 10 minutes. Perhaps not enough
-                dict_userjob['EXECUTION_DELAY'] = 600
-
-                # set to executing even though it is kind-of wrong...
-                dict_userjob['STATUS'] = 'EXECUTING'
-                dict_userjob['EXECUTING_TIMESTAMP'] = time.gmtime()
-                executing_queue.enqueue_job(dict_userjob,
-                                            executing_queue.queue_length())
-
-            # Either way, save the job mrsl.
-            # Status is EXECUTING or FAILED
-            pickle(dict_userjob, file_userjob, logger)
-
-            # go on with scheduling loop (do not use scheduler magic below)
-            continue
-
-        # following: non-ARC code
-
         # put job in queue
 
         job_queue.enqueue_job(dict_userjob, job_queue.queue_length())
@@ -1296,22 +1204,6 @@ def graceful_shutdown():
             msg += \
                 ', but job is being executed by %s:%s, ignoring result.'\
                 % (job_dict['UNIQUE_RESOURCE_NAME'], job_dict['EXE'])
-        elif job_dict['UNIQUE_RESOURCE_NAME'] == 'ARC':
-            if not configuration.arc_clusters:
-                logger.error('ARC backend disabled - ignore %s' %
-                             job_dict)
-                continue
-            msg += (', which is an ARC job (ID %s).' % job_dict['EXE'])
-
-            # remove from the executing queue
-            executing_queue.dequeue_job_by_id(job_id)
-
-            # job status has been checked by put script already
-            # we need to clean up the job remainder (links, queue, and ARC
-            # side)
-            clean_arc_job(job_dict, 'FINISHED', None,
-                          configuration, logger, False)
-            msg += 'ARC job completed'
 
         else:
 
@@ -1455,26 +1347,6 @@ def graceful_shutdown():
                     'Cancel job: Could not get job_dict for executing job')
                 continue
 
-            # special treatment of ARC jobs: delete two links and cancel job
-            # in ARC
-            if unique_resource_name == 'ARC':
-                if not configuration.arc_clusters:
-                    logger.error('ARC backend disabled - ignore %s' %
-                                 job_dict)
-                    continue
-
-                # remove from the executing queue
-                executing_queue.dequeue_job_by_id(job_id)
-
-                # job status has been set by the cancel request already, but
-                # we need to kill the ARC job, or clean it (if already
-                # finished), and clean up the job remainder links
-                clean_arc_job(job_dict, 'CANCELED', None,
-                              configuration, logger, True)
-
-                logger.debug('ARC job completed')
-                continue
-
             if not server_cleanup(
                 job_dict['SESSIONID'],
                 job_dict['IOSESSIONID'],
@@ -1539,26 +1411,6 @@ def graceful_shutdown():
 
         job_dict = executing_queue.get_job_by_id(jobid)
 
-        # special treatment of ARC jobs: delete two links and
-        # clean job in ARC system, do not retry.
-        if job_dict and unique_resource_name == 'ARC':
-            if not configuration.arc_clusters:
-                logger.error('ARC backend disabled - ignore %s' %
-                             job_dict)
-                continue
-
-            # remove from the executing queue
-            executing_queue.dequeue_job_by_id(jobid)
-
-            # job status has been set by the cancel request already, but
-            # we need to kill the ARC job, or clean it (if already finished),
-            # and clean up the job remainder links
-            clean_arc_job(job_dict, 'FAILED', 'Job timed out',
-                          configuration, logger, True)
-
-            logger.debug('ARC job timed out, removed')
-            continue
-
         # Execution information is removed from job_dict in
         # requeue_job - save here
 

mig/server/grid_script.py

@@ -47,7 +47,6 @@
 
 try:
     from mig.shared import mrsltoxrsl
-    from mig.shared import arcwrapper
 except Exception as exc:
     # Ignore errors and let it crash if ARC is enabled without the lib
     pass
@@ -441,130 +440,6 @@ def create_job_script(
     return (job_dict, 'OK')
 
 
-def create_arc_job(
-    job,
-    configuration,
-    logger,
-):
-    """Analog to create_job_script for ARC jobs:
-    Creates symLinks for receiving result files, translates job dict to ARC
-    xrsl, and stores resulting job script (xrsl + sh script) for submitting.
-
-    We do _not_ create a separate job_dict with copies and SESSIONID inside,
-    as opposed to create_job_script, all we need is the link from 
-    webserver_home / sessionID into the user's home directory 
-    ("job_output/job['JOB_ID']" is added to the result upload URLs in the 
-    translation). 
-
-    Returns message (ARC job ID if no error) and sessionid (None if error)
-    """
-
-    if not configuration.arc_clusters:
-        return (None, 'No ARC support!')
-    if not job['JOBTYPE'] == 'arc':
-        return (None, 'Error. This is not an ARC job')
-
-    # Deep copy job for local changes
-    job_dict = deepcopy(job)
-    # Finally expand reserved job variables like +JOBID+ and +JOBNAME+
-    job_dict = expand_variables(job_dict)
-    # ... no more changes to job_dict from here on
-    client_id = "%(USER_CERT)s" % job_dict
-
-    # we do not want to see empty jobs here. Test as done in create_job_script.
-    if client_id == configuration.empty_job_name:
-        return (None, 'Error. empty job for ARC?')
-
-    # generate random session ID:
-    sessionid = hexlify(os.urandom(session_id_bytes))
-    logger.debug('session ID (for creating links): %s' % sessionid)
-
-    client_dir = client_id_dir(client_id)
-
-    # make symbolic links inside webserver_home:
-    #
-    # we need: link to owner's dir. to receive results,
-    #          job mRSL inside sessid_to_mrsl_link_home
-    linklist = [(configuration.user_home + client_dir,
-                 configuration.webserver_home + sessionid),
-                (configuration.mrsl_files_dir + client_dir + '/' +
-                 "%(JOB_ID)s" % job_dict + '.mRSL',
-                 configuration.sessid_to_mrsl_link_home + sessionid + '.mRSL')
-                ]
-
-    for (dest, loc) in linklist:
-        make_symlink(dest, loc, logger)
-
-    # the translation generates an xRSL object which specifies to execute
-    # a shell script with script_name. If sessionid != None, results will
-    # be uploaded to sid_redirect/sessionid/job_output/job_id
-
-    try:
-        (xrsl, script, script_name) = mrsltoxrsl.translate(job_dict, sessionid)
-        logger.debug('translated to xRSL: %s' % xrsl)
-        logger.debug('script:\n %s' % script)
-
-    except Exception as err:
-        # error during translation, pass a message
-        logger.error('Error during xRSL translation: %s' % err.__str__())
-        return (None, err.__str__())
-
-        # we submit directly from here (the other version above does
-        # copyFileToResource and gen_job_script generates all files)
-
-    # we have to put the generated script somewhere..., and submit from there.
-    # inputfiles are given by the user as relative paths from his home,
-    # so we should use that location (and clean up afterwards).
-
-    # write script (to user home)
-    user_home = os.path.join(configuration.user_home, client_dir)
-    script_path = os.path.abspath(os.path.join(user_home, script_name))
-    write_file(script, script_path, logger)
-
-    os.chdir(user_home)
-
-    try:
-        logger.debug('submitting job to ARC')
-        session = arcwrapper.Ui(user_home)
-        arc_job_ids = session.submit(xrsl)
-
-        # if no exception occurred, we are done:
-
-        job_dict['ARCID'] = arc_job_ids[0]
-        job_dict['SESSIONID'] = sessionid
-
-        msg = 'OK'
-        result = job_dict
-
-    # when errors occurred, pass a message to the caller.
-    except arcwrapper.ARCWrapperError as err:
-        msg = err.what()
-        result = None  # unsuccessful
-    except arcwrapper.NoProxyError as err:
-        msg = 'No Proxy found: %s' % err.what()
-        result = None  # unsuccessful
-    except Exception as err:
-        msg = err.__str__()
-        result = None  # unsuccessful
-
-    # always remove the generated script
-    os.remove(script_name)
-    # and remove the created links immediately if failed
-    if not result:
-        for (_, link) in linklist:
-            os.remove(link)
-        logger.error('Unsuccessful ARC job submission: %s' % msg)
-    else:
-        logger.debug('submitted to ARC as job %s' % msg)
-    return (result, msg)
-
-    # errors are handled inside grid_script. For ARC jobs, set status = FAILED
-    # on errors, and include the message
-    # One potential error is that the proxy is invalid,
-    # which should be checked inside the parser, before informing
-    # grid_script about the new job.
-
-
 def gen_job_script(
     job_dictionary,
     resource_config,