agave-jobs/jobs-core/src/main/java/org/iplantc/service/jobs/queue/actions/StagingAction.java
/**
     * Stages the user-supplied {@link Job#getInputs()} to the {@link Job#getWorkPath()} on the {@link ExecutionSystem}.
     * If an input, system, or permissions are absent and the job cannot continue without them, then a @link {JobDependencyException}
     * will be thrown and no retries will be made. Otherwise, any failures will result in the job following the normal
     * retry policy.
     *
     * @throws SystemUnavailableException when one of the input or execution systems is not available.
     * @throws JobException when a non-critical failure occurs such as a connection failure, db lookup failure, etc.
     * @throws JobDependencyException when a critical job dependency is absent and the job cannot continue. This will
     *   effectively cancel any further attempts.
     * @throws ClosedByInterruptException when the action is interrupted by an external process
     */
Stages the user-supplied  Job#getInputs() to the  Job#getWorkPath() on the  ExecutionSystem.
If an input, system, or permissions are absent and the job cannot continue without them, then a @link {JobDependencyException}
will be thrown and no retries will be made. Otherwise, any failures will result in the job following the normal
retry policy.

@throws SystemUnavailableException when one of the input or execution systems is not available.
@throws JobException when a non-critical failure occurs such as a connection failure, db lookup failure, etc.
@throws JobDependencyException when a critical job dependency is absent and the job cannot continue. This will
effectively cancel any further attempts.
@throws ClosedByInterruptException when the action is interrupted by an external process
Stages the user - supplied Job#getInputs () to the Job#getWorkPath () on the ExecutionSystem . If an input system or permissions are absent and the job cannot continue without them then a @link { JobDependencyException } will be thrown and no retries will be made . Otherwise any failures will result in the job following the normal retry policy . @throws SystemUnavailableException when one of the input or execution systems is not available . @throws JobException when a non - critical failure occurs such as a connection failure db lookup failure etc . @throws JobDependencyException when a critical job dependency is absent and the job cannot continue . This will effectively cancel any further attempts . @throws ClosedByInterruptException when the action is interrupted by an external process
public void run() throws SystemUnavailableException, JobException, ClosedByInterruptException, JobDependencyException {
        RemoteDataClient jobInputRemoteDataClient = null;
        RemoteDataClient jobExecutionSystemRemoteDataClient = null;
        try {
            ExecutionSystem executionSystem = getExecutionSystem();
            jobExecutionSystemRemoteDataClient = getRemoteDataClientForExecutionSystem(executionSystem);
            String remoteJobWorkPath = calculateRemoteJobPath(executionSystem);
            createJobRemoteWorkPath(executionSystem, jobExecutionSystemRemoteDataClient, remoteJobWorkPath);
            getJob().setWorkPath(remoteJobWorkPath);
            log.debug("Beginning staging inputs for job " + getJob().getUuid() + " to " +
                    executionSystem.getSystemId() + ":" + getJob().getWorkPath());
            JobManager.updateStatus(getJob(), JobStatusType.PROCESSING_INPUTS);
            Map<String, String[]> jobInputMap = JobManager.getJobInputMap(getJob());
            for (String inputKey : jobInputMap.keySet()) {
                checkStopped();
                String[] rawInputValues = jobInputMap.get(inputKey);
                URI singleRawInputUri = null;
                RemoteSystem jobInputSystem = null;
                for (String singleRawInputValue : rawInputValues) {
                    checkStopped();
                    singleRawInputUri = new URI(singleRawInputValue);
                    String remoteJobInputPath = null;
                    if (ApiUriUtil.isInternalURI(singleRawInputUri)) {
                        jobInputSystem = ApiUriUtil.getRemoteSystem(getJob().getOwner(), singleRawInputUri);
                        jobInputRemoteDataClient = getRemoteDataClientForInputSystem(jobInputSystem, singleRawInputValue);
                        remoteJobInputPath = getRemoteJobInputPathIfJobOwnerHasPermission(singleRawInputUri, jobInputSystem, jobInputRemoteDataClient);
                    } else {
                        jobInputRemoteDataClient = getRemoteDataClientForUri(singleRawInputUri);
                        remoteJobInputPath = singleRawInputUri.getPath();
                    }
                    String destPath = Paths.get(remoteJobWorkPath).resolve(FilenameUtils.getName(remoteJobInputPath)).toString();
                    checkStopped();
                    try {
                        if (!isJobInputAlreadyTransferred(singleRawInputValue, destPath, jobExecutionSystemRemoteDataClient, remoteJobInputPath, jobInputRemoteDataClient)) {
                            transferJobInput(jobInputRemoteDataClient, jobExecutionSystemRemoteDataClient, singleRawInputValue, remoteJobInputPath, destPath);
                        }
                    } finally {
                        try { jobExecutionSystemRemoteDataClient.disconnect(); } catch (Exception ignored) {}
                        try { jobInputRemoteDataClient.disconnect(); } catch (Exception ignored) {}
                    }
                }
            }
            log.debug("Completed staging inputs for job " + getJob().getUuid() + " to " +
                    executionSystem.getSystemId() + ":" + getJob().getWorkPath());
        }
        catch (JobException|ClosedByInterruptException e) {
            throw e;
        }
        catch (SystemUnknownException | MissingDataException e) {
            log.error(e.getMessage());
            throw new JobDependencyException(e.getMessage(), e);
        }
        catch (PermissionException e) {
            String message = "User lacks permissions to access input for job " + getJob().getUuid() + ". " +
                    e.getMessage();
            log.error(message);
            throw new JobDependencyException(message, e);
        }
        catch (AuthenticationException e) {
            log.error("Unable to authenticate to stage input file for job " + getJob().getUuid() + ". " +
                    e.getMessage());
            throw new JobDependencyException(e.getMessage(), e);
        }
        catch (URISyntaxException e) {
            String message = "Invalid input url, " + e.getInput() + " provided as input for job " +
                    getJob().getUuid();
            log.error(message);
            throw new JobDependencyException(message, e);
        }
        catch (AgaveNamespaceException e) {
            String message = "Invalid url format provided as input for job " +
                    getJob().getUuid() + ". " + e.getMessage();
            log.error(message);
            throw new JobDependencyException(message, e);
        }
        catch (NotImplementedException e) {
            log.error("Invalid input protocol provided for job " + getJob().getUuid());
            throw new JobDependencyException(e.getMessage());
        }
        catch (JobInputStagingException e) {
            log.error("Failed to create work directory for job " + getJob().getUuid(), e);
            throw new JobException(e.getMessage(), e);
        }
        catch (TransferException | RemoteDataException e) {
            log.error(e.getMessage());
            throw new JobException(e.getMessage(), e);
        }
        catch (Throwable e) {
            log.error("Failed to stage input for job " + getJob().getUuid(), e);
            throw new JobException(e.getMessage(), e);
        }
        finally {
            if (jobInputRemoteDataClient != null) jobInputRemoteDataClient.disconnect();
            if (jobExecutionSystemRemoteDataClient != null) jobExecutionSystemRemoteDataClient.disconnect();
        }
        if (!isStopped()) {
            if (getJob().getStatus() == JobStatusType.STAGING_INPUTS) {
                updateJobStatus(JobStatusType.STAGED, JobStatusType.STAGED.getDescription());
                log.debug("Completed staging inputs for job " + getJob().getUuid() + " " + getJob().getName());
            }
        }
    }
public void run ( ) throws SystemUnavailableException , JobException , ClosedByInterruptException , JobDependencyException { RemoteDataClient jobInputRemoteDataClient = null ; RemoteDataClient jobExecutionSystemRemoteDataClient = null ; try { ExecutionSystem executionSystem = getExecutionSystem ( ) ; jobExecutionSystemRemoteDataClient = getRemoteDataClientForExecutionSystem ( executionSystem ) ; String remoteJobWorkPath = calculateRemoteJobPath ( executionSystem ) ; createJobRemoteWorkPath ( executionSystem , jobExecutionSystemRemoteDataClient , remoteJobWorkPath ) ; getJob ( ) . setWorkPath ( remoteJobWorkPath ) ; log . debug ( "Beginning staging inputs for job " + getJob ( ) . getUuid ( ) + " to " + executionSystem . getSystemId ( ) + ":" + getJob ( ) . getWorkPath ( ) ) ; JobManager . updateStatus ( getJob ( ) , JobStatusType . PROCESSING_INPUTS ) ; Map < String , String [ ] > jobInputMap = JobManager . getJobInputMap ( getJob ( ) ) ; for ( String inputKey : jobInputMap . keySet ( ) ) { checkStopped ( ) ; String [ ] rawInputValues = jobInputMap . get ( inputKey ) ; URI singleRawInputUri = null ; RemoteSystem jobInputSystem = null ; for ( String singleRawInputValue : rawInputValues ) { checkStopped ( ) ; singleRawInputUri = new URI ( singleRawInputValue ) ; String remoteJobInputPath = null ; if ( ApiUriUtil . isInternalURI ( singleRawInputUri ) ) { jobInputSystem = ApiUriUtil . getRemoteSystem ( getJob ( ) . getOwner ( ) , singleRawInputUri ) ; jobInputRemoteDataClient = getRemoteDataClientForInputSystem ( jobInputSystem , singleRawInputValue ) ; remoteJobInputPath = getRemoteJobInputPathIfJobOwnerHasPermission ( singleRawInputUri , jobInputSystem , jobInputRemoteDataClient ) ; } else { jobInputRemoteDataClient = getRemoteDataClientForUri ( singleRawInputUri ) ; remoteJobInputPath = singleRawInputUri . getPath ( ) ; } String destPath = Paths . get ( remoteJobWorkPath ) . resolve ( FilenameUtils . getName ( remoteJobInputPath ) ) . toString ( ) ; checkStopped ( ) ; try { if ( ! isJobInputAlreadyTransferred ( singleRawInputValue , destPath , jobExecutionSystemRemoteDataClient , remoteJobInputPath , jobInputRemoteDataClient ) ) { transferJobInput ( jobInputRemoteDataClient , jobExecutionSystemRemoteDataClient , singleRawInputValue , remoteJobInputPath , destPath ) ; } } finally { try { jobExecutionSystemRemoteDataClient . disconnect ( ) ; } catch ( Exception ignored ) { } try { jobInputRemoteDataClient . disconnect ( ) ; } catch ( Exception ignored ) { } } } } log . debug ( "Completed staging inputs for job " + getJob ( ) . getUuid ( ) + " to " + executionSystem . getSystemId ( ) + ":" + getJob ( ) . getWorkPath ( ) ) ; } catch ( JobException | ClosedByInterruptException e ) { throw e ; } catch ( SystemUnknownException | MissingDataException e ) { log . error ( e . getMessage ( ) ) ; throw new JobDependencyException ( e . getMessage ( ) , e ) ; } catch ( PermissionException e ) { String message = "User lacks permissions to access input for job " + getJob ( ) . getUuid ( ) + ". " + e . getMessage ( ) ; log . error ( message ) ; throw new JobDependencyException ( message , e ) ; } catch ( AuthenticationException e ) { log . error ( "Unable to authenticate to stage input file for job " + getJob ( ) . getUuid ( ) + ". " + e . getMessage ( ) ) ; throw new JobDependencyException ( e . getMessage ( ) , e ) ; } catch ( URISyntaxException e ) { String message = "Invalid input url, " + e . getInput ( ) + " provided as input for job " + getJob ( ) . getUuid ( ) ; log . error ( message ) ; throw new JobDependencyException ( message , e ) ; } catch ( AgaveNamespaceException e ) { String message = "Invalid url format provided as input for job " + getJob ( ) . getUuid ( ) + ". " + e . getMessage ( ) ; log . error ( message ) ; throw new JobDependencyException ( message , e ) ; } catch ( NotImplementedException e ) { log . error ( "Invalid input protocol provided for job " + getJob ( ) . getUuid ( ) ) ; throw new JobDependencyException ( e . getMessage ( ) ) ; } catch ( JobInputStagingException e ) { log . error ( "Failed to create work directory for job " + getJob ( ) . getUuid ( ) , e ) ; throw new JobException ( e . getMessage ( ) , e ) ; } catch ( TransferException | RemoteDataException e ) { log . error ( e . getMessage ( ) ) ; throw new JobException ( e . getMessage ( ) , e ) ; } catch ( Throwable e ) { log . error ( "Failed to stage input for job " + getJob ( ) . getUuid ( ) , e ) ; throw new JobException ( e . getMessage ( ) , e ) ; } finally { if ( jobInputRemoteDataClient != null ) jobInputRemoteDataClient . disconnect ( ) ; if ( jobExecutionSystemRemoteDataClient != null ) jobExecutionSystemRemoteDataClient . disconnect ( ) ; } if ( ! isStopped ( ) ) { if ( getJob ( ) . getStatus ( ) == JobStatusType . STAGING_INPUTS ) { updateJobStatus ( JobStatusType . STAGED , JobStatusType . STAGED . getDescription ( ) ) ; log . debug ( "Completed staging inputs for job " + getJob ( ) . getUuid ( ) + " " + getJob ( ) . getName ( ) ) ; } } }
Stages the user-supplied {@link Job#getInputs()} to the {@link Job#getWorkPath()} on the {@link ExecutionSystem}.
Stages the user - supplied { @link Job#getInputs () } to the { @link Job#getWorkPath () } on the { @link ExecutionSystem } .
// copy to remote execution work directory // calculate the job work directory. we do this here in the event the execution system // root, home, work, or scratch directories change between job request time and  input staging. // we need a way to parallelize this task. Ideally we'd just throw each input // file to the staging queue and let 'em rip // Get a well-formed map of user-supplied + default + hidden/required inputs for the job // Inputs are stored as a JSON object with the Job table, so we use this convenience // method to make it less nasty to work with // each job input corresponds to a SoftwareInput, which may have multiple values, so we iterate over // each job input, processing all values for that input in turn. // circuit breaker // inputs can have multiple values provided to them by the user and/or their defaults. // here we iterate over each input value, staging them to the remote system. // ensure it's a valid URI we can pass for parsing. // URL will be handled differently depending on the scheme and whether it points to an internal // API resource. We check that first before proceeding. // get system for an input URI representing an internal or agave url // get a client to the remote system for the input // We handle non-internal URL as standard URL. We parse based on URI schema and generate // a client dynamically based on URI components. // for non-internal URL, the client will use "/" as the rootdir and homedir, so the given path // will be treated as absolute regardless. // figure out the agave relative path of the job input on the execution system // circuit breaker // see if we can skip this transfer due to prior success // finally ok to make the remote transfer // close connections every time since we reuse these RDC. // these will be self-describing, non-fatal exceptions // status should have been updated in job object if anything was // staged