GT4 远程执行客户机快速入门

/**
 * A Custom GRAM Client for GT4:
 * 1) Based on the GlobusRun command from the GT4 WS-GRAM implementation.
 * 2) The GT4 WSRF libraries are required to compile this stuff, plus the
 * following VM arguments must be used:
 * 	-Daxis.ClientConfigFile=[GLOBUS_LOCATION]/client-config.wsdd
 *  	-DGLOBUS_LOCATION=[GLOBUS_LOCATION]
 * @author Vladimir Silva
 *
 */
public class GRAMClient
	// Listen for job status messages
	implements GramJobListener
{
    private static Log logger = LogFactory.getLog(GRAMClient.class.getName());

    // Amount of time to wait for job status changes
    private static final long STATE_CHANGE_BASE_TIMEOUT_MILLIS = 60000;

    /**
     * Job submission member variables.
     */
    private GramJob job;

    // completed if Done or Failed
    private boolean jobCompleted = false;
    // Batch runs will not wait for the job to complete
    private boolean batch;

    // Delegation
    private boolean limitedDelegation = true;
    private boolean delegationEnabled = true;

    // Don't print messages by default
    private boolean quiet = false;

    // proxy credential
    private String proxyPath = null;

    /**
     * Application error state.
     */
     private boolean noInterruptHandling = false;
     private boolean isInterrupted = true;
     private boolean normalApplicationEnd = false;

<!--
 Job to print the remote host environment
 For a full set of WS-GRAM options see the WS_GRAM developers page at:
   http://www.globus.org/toolkit/docs/4.0/execu-
   tion/wsgram/user-index.html#s-wsgram-user-simplejob
-->
<job>
    <executable>/usr/bin/env</executable>

    <!-- Job stdout/err will be stored in the globus user home directory -->
    <stdout>${GLOBUS_USER_HOME}/stdout</stdout>
    <stderr>${GLOBUS_USER_HOME}/stderr</stderr>

    <!-- Clean up std out/err from user's home -->
    <fileCleanUp>
        <deletion>
            <file>file:///${GLOBUS_USER_HOME}/stdout</file>
        </deletion>
        <deletion>
            <file>file:///${GLOBUS_USER_HOME}/stderr</file>
        </deletion>
    </fileCleanUp>
</job>

    /**
     * Callback as a GramJobListener.
     * Will not be called in batch mode.
     */
    public void stateChanged(GramJob job) {
        StateEnumeration jobState = job.getState();
        boolean holding = job.isHolding();
        printMessage("========== State Notification ==========");
        printJobState(jobState, holding);
        printMessage("========================================");

        synchronized (this) {
            if (   jobState.equals(StateEnumeration.Done)
                || jobState.equals(StateEnumeration.Failed)) {

                printMessage("Exit Code: "
                    + Integer.toString(job.getExitCode()));

                this.jobCompleted = true;
            }

            notifyAll();

            // if we a running an interactive job,
            // prevent a hold from hanging the client
            if ( holding && !batch) {
                logger.debug(
                    "Automatically releasing hold for interactive job");
                try {
                    job.release();
                } catch (Exception e) {
                   String errorMessage = "Unable to release job from hold";
                   logger.debug(errorMessage, e);
                   printError(errorMessage + " - " + e.getMessage());
                }
            }
        }
    }

/**
* Submit a WS-GRAM Job (GT4)
* @param factoryEndpoint Factory endpoint reference
* @param simpleJobCommandLine Executable (null to use a job file)
* @param rslFileJob XML file (null to use a command line)
* @param authorization Authorization: Host, Self, Identity
* @param xmlSecurity XML Sec: Encryption or signature
* @param batchMode Submission mode: batch will not wait for completion
* @param dryRunMode Used to parse RSL
* @param quiet Messages/NO messages
* @param duration Duration date
* @param terminationDate Termination date
* @param timeout  Job timeout (ms)
*/
private void submitRSL(EndpointReferenceType factoryEndpoint,
		String simpleJobCommandLine,
		File rslFile,
		Authorization authorization,
		Integer xmlSecurity,
		boolean batchMode,
		boolean dryRunMode,
		boolean quiet,
		Date duration,
		Date terminationDate,
		int timeout)
throws Exception
{
	this.quiet = quiet;
	this.batch = batchMode || dryRunMode; // in single job only.
	// In multi-job, -batch is not allowed. Dryrun is.

	if (batchMode) {
		printMessage("Warning: Will not wait for job completion, "
				+ "and will not destroy job service.");
	}

	// create a job object with the XML spec or a simple command
	if (rslFile != null) {
		try {
			this.job = new GramJob(rslFile);
		} catch (Exception e) {
			String errorMessage = "Unable to parse RSL from file "
					+ rslFile;
			logger.debug(errorMessage, e);
			throw new IOException\
			(errorMessage + " - " + e.getMessage());
		}
	}
	else {
		this.job = new GramJob(RSLHelper
				.makeSimpleJob(simpleJobCommandLine));
	}

	// set job options
	job.setTimeOut(timeout);
	job.setAuthorization(authorization);
	job.setMessageProtectionType(xmlSecurity);
	job.setDelegationEnabled(this.delegationEnabled);
	job.setDuration(duration);
	job.setTerminationTime(terminationDate);

	// submit here
	this.processJob(job, factoryEndpoint, batch);
}

/**
* Submit the GRAM Job
* @param job Job object (GramJob)
* @param factoryEndpoint Factory end point reference
* @param batch If false will wait for job to complete
* @throws Exception
*/
private void processJob(GramJob job,
		EndpointReferenceType factoryEndpoint,
		boolean batch)
throws Exception
{
	// load custom proxy (if any)
	if (proxyPath != null) {
		try {
		            ExtendedGSSManager manager = (ExtendedGSSManager)
                                           ExtendedGSSManager.getInstance();

			String handle = "X509_USER_PROXY=" + proxyPath.toString();

			GSSCredential proxy = manager.createCredential(handle
					.getBytes(),
					ExtendedGSSCredential.\
					IMPEXP_MECH_SPECIFIC,
					GSSCredential.DEFAULT_LIFETIME, null,
					GSSCredential.INITIATE_AND_ACCEPT);
			job.setCredentials(proxy);
		} catch (Exception e) {
			logger.debug("Exception while obtaining user proxy: ", e);
			printError("error obtaining user \
			proxy: " + e.getMessage());
			// don't exit, but resume using default proxy instead
		}
	}

	// Generate a Job ID
	UUIDGen uuidgen 	= UUIDGenFactory.getUUIDGen();
	String submissionID = "uuid:" + uuidgen.nextUUID();

	printMessage("Submission ID: " + submissionID);

	if (!batch) {
		job.addListener(this);
	}

	boolean submitted = false;
	int tries = 0;

	while (!submitted) {
		tries++;
		try {
			job.submit(factoryEndpoint, batch, this.limitedDelegation,
					submissionID);
			submitted = true;
		} catch (Exception e) {
			logger.debug("Exception while \
			submitting the job request: ", e);
			throw new IOException("Job request error: " + e);
		}
	}

	if (batch) {
		printMessage("CREATED MANAGED JOB SERVICE WITH HANDLE:");
		printMessage(job.getHandle());
	}

	if (logger.isDebugEnabled()) {
		long millis = System.currentTimeMillis();
		BigDecimal seconds = new BigDecimal(((double) millis) / 1000);
		seconds = seconds.setScale(3, BigDecimal.ROUND_HALF_DOWN);
		logger.debug("Submission time (secs) after: " + seconds.toString());
		logger.debug("Submission time in milliseconds: " + millis);
	}

	if (!batch) {
		printMessage("WAITING FOR JOB TO FINISH");

		waitForJobCompletion(STATE_CHANGE_BASE_TIMEOUT_MILLIS);

		try {
			this.destroyJob(this.job); // TEST
		} catch (Exception e) {
			printError("could not destroy");
		}

		if (this.job.getState().equals(StateEnumeration.Failed)) {
			printJobFault(this.job);
		}
	}
}

private synchronized void waitForJobCompletion(
		long maxWaitPerStateNotificationMillis)
throws Exception
{

	long durationToWait = maxWaitPerStateNotificationMillis;
	long startTime;
	StateEnumeration oldState = job.getState();

	// prints one more state initially (Unsubmitted)
	// but cost extra remote call for sure. Null test below instead
	while (!this.jobCompleted)
	{
		if (logger.isDebugEnabled()) {
			logger.debug("Job not completed - \
			waiting for state change "
				+ "(timeout before pulling: " + durationToWait
				+ " ms).");
		}

		startTime = System.currentTimeMillis(); // (re)set start time
		try {
			wait(durationToWait); // wait for a state change notif
		} catch (InterruptedException ie) {
			String errorMessage = \
			"interrupted thread waiting for job to finish";
			logger.debug(errorMessage, ie);
			printError(errorMessage); // no exiting...
		}

		// now let's determine what stopped the wait():

		StateEnumeration currentState = job.getState();
		// A) New job state change notification (good!)
		if (currentState != null && !currentState.equals(oldState)) {
		   oldState = currentState; // wait for next state notif
		   durationToWait = maxWaitPerStateNotificationMillis; // reset
		}
		else
		{
			long now = System.currentTimeMillis();
			long durationWaited = now - startTime;

			// B) Timeout when waiting for a notification (bad)
			if (durationWaited >= durationToWait) {
				if (logger.isWarnEnabled()) {
				   logger.warn("Did not receive \
				   any new notification of "
					+ "job state change \
					after a delay of "
					+ durationToWait + \
					" ms.\nPulling job state.");
				}
				// pull state from remote job and print the
				// state only if it is a new state
				//refreshJobStatus();
				job.refreshStatus();

				// binary exponential backoff
				durationToWait = 2 * durationToWait;
			}
			// C) Some other reason
			else {
				// wait but only for \
				remainder of timeout duration
				durationToWait = durationToWait - durationWaited;
			}
		}
	}
}

	// remote host
	String contact 		= "rtpmeta";

	// Factory type: Fork, Condor, PBS, LSF
	String factoryType	= ManagedJobFactoryConstants.FACTORY_TYPE.FORK;

	// Job XML
	File rslFile 		= new File("/tmp/simple.xml");

	// Default Security: Host authorization + XML encryption
	Authorization authz = HostAuthorization.getInstance();
	Integer xmlSecurity = Constants.ENCRYPTION;

	// Submission mode: batch = will not wait
	boolean batchMode 			= false;

	// a Simple command executable (if no job file)
	String simpleJobCommandLine = null;

	// Job timeout values: duration, termination times
	Date serviceDuration 		= null;
	Date serviceTermination 	= null;
	int timeout 			= GramJob.DEFAULT_TIMEOUT;


	try {
		GRAMClient gram = new GRAMClient();
		gram.submitRSL(getFactoryEPR(contact,factoryType)
				, simpleJobCommandLine, rslFile
				, authz, xmlSecurity
				, batchMode, false, false
				, serviceDuration, serviceTermination, timeout );

	} catch (Exception e) {
		e.printStackTrace();
	}

Submission ID: uuid:3eb57530-cfc4-11da-bb0b-dadfac0c5c05
WAITING FOR JOB TO FINISH
========== State Notification ==========
Job State: CleanUp
========================================
========== State Notification ==========
Job State: Active
========================================
========== State Notification ==========
Job State: Done
========================================
Exit Code: 0
DESTROYING JOB RESOURCE
JOB RESOURCE DESTROYED

globusrun-ws -submit -s
  -factory https://rtpmeta:8443/wsrf/services/ManagedJobFactoryService
  -f /tmp/simple.xml

Delegating user credentials...Done.
Submitting job...Done.
Job ID: uuid:272c8c7c-cfcf-11da-9061-00065ba4e6be
Termination time: 04/20/2006 18:05 GMT
Current job state: Active
Current job state: CleanUp-Hold
X509_USER_KEY=
X509_CERT_DIR=/etc/grid-security/certificates
LOGNAME=globus
GLOBUS_LOCATION=/opt/gt-4.0.1
HOME=/opt/home/globus
X509_USER_CERT=
GLOBUS_GRAM_JOB_HANDLE=https://134.67.144.3:8443/wsrf/services/
	ManagedExecutableJobService?272c8c7c-cfcf-11da-9061-00065ba4e6be
JAVA_HOME=/usr/java/j2sdk1.4.2_10/jre
X509_USER_PROXY=
Current job state: CleanUp
Current job state: Done
Destroying job...Done.
Cleaning up any delegated credentials...Done.

No client transport named 'https' found!
	at org.apache.axis.client.AxisClient.invoke(AxisClient.java:170)
	at org.apache.axis.client.Call.invokeEngine(Call.java:2727)
	at org.apache.axis.client.Call.invoke(Call.java:2710)
	at org.apache.axis.client.Call.invoke(Call.java:2386)
	at org.apache.axis.client.Call.invoke(Call.java:2309)
	...

-Daxis.ClientConfigFile=[GLOBUS_LOCATION]/client-config.wsdd

org.apache.axis.ConfigurationException:
Configuration file directory './etc' does not exist or is not
	a directory or is not readable.
 at org.apache.axis.configuration.DirProvider.{init}(DirProvider.java:73)
 at org.globus.wsrf.container.ServiceDispatcher.{init}(ServiceDispatcher.java:80)
 at org.globus.wsrf.container.GSIServiceDispatcher.{init}(GSIServiceContainer.java:63)
 at org.globus.wsrf.container.GSIServiceContainer.createServiceDispatcher
 (GSIServiceContainer.java:49)
 at org.globus.wsrf.container.ServiceContainer.start(ServiceContainer.java:226)
	...

	-DGLOBUS_LOCATION=[GLOBUS_LOCATION]

# WS-GRAM client
log4j.category.org.globus.exec.client=DEBUG
log4j.category.org.globus.exec.generated=DEBUG

# Custom WS-GRAM client
log4j.category.gram=DEBUG