GT4 远程执行客户机快速入门
您是一位软件 开发 人员，正在努力地把公司的应用程序移植到一个崭新的分布式环境中。您的组织已经为网格计算这一令人兴奋的新领域提供了一些资源，但是您却不知道接下来应该做些什么。最终期限很快就要到了，惶恐的气氛逐渐在您的团队中蔓延开来。您需要做的
您是一位软件
开发人员，正在努力地把公司的应用程序移植到一个崭新的分布式环境中。您的组织已经为网格计算这一令人兴奋的新领域提供了一些资源，但是您却不知道接下来应该做些什么。最终期限很快就要到了，惶恐的气氛逐渐在您的团队中蔓延开来。您需要做的事情是制定一个快速路线图，将组织的应用程序部署或移植到计算或数据网格中。Globus 一直以来都在致力于提供有关 Globus Toolkit V4.0（GT4）的文档，但您现在需要的是
开发文档。Globus 正在创建一些
开发文档
Web 站点来解决这个问题。同时，在本文中，我们给出了一些资源管理领域中的开发技巧，目的在于帮助您使用
Web services Grid Resource Allocation and Management（WS-GRAM）服务快速启用网格应用程序。
 |
OGSA-GRAM 和 WS-GRAM 之间的可移植性问题
GT4 中的 WS-GRAM 与 GT3（OGSA-MMJFS）中实现的 GRAM 接口并不兼容，不过大部分接口的名字都是相同的：例如 GramJobListener、GramJob 等。实际上，这个自定义 GRAM 客户机的大部分代码都与 GT3 的对应部分非常类似，是构建在 GT3 的代码基础之上的。GRAM 在 GT4 中作为 Web Services Resource Framework（WSRF）服务实现；而在 GT3 中，它是一个 OGSA 服务。 |
|
全新的 WS-GRAM
Globus GRAM 服务为多种类型的作业调度程序提供了安全提交作业的功能。为支持 XML 加密和签名，WS-GRAM 彻底重新进行了构建。它使用数字证书在客户机和服务器之间发送安全的 XML SOAP 消息。实际上，GT4 就是一个安全 Web 服务容器。它与传统容器之间的区别就是在于可跨多个 SOAP（Simple Object Aclearcase/" target="_blank" >ccess Protocol）消息（也称为 WSRF）维护状态。
下面的技巧展示了如何创建一个自定义 Java™ 技术程序向远程主机提交 WS-GRAM 作业。
技巧 1：监听作业状态的变化
要监听作业状态的变化，就需要实现 org.globus.exec.client.GramJobListener 接口，它与其 GT3 的对应部分并不兼容（顺便提一下，二者的接口名是相同的）。这种技术如下所示。
清单 1. WS-GRAM 客户机基础实现
/**
* 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;
|
WS-GRAM 中的基本对象是 GramJob。它用来以 XML 格式提交作业描述文件。作业描述中定义了可执行程序、参数、标准输入/输出/错误、阶段文件（staging file）以及其他一些选项。下面是一个简单的作业描述文件。
清单 2. 简单的作业描述 XML 文件
<!--
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>
|
清单 2 中的作业文件将输出远程主机环境，并将 stdout/stderr 的内容存储到用户的主目录中（分别使用 stdout 和 stderr 的名字）。如 fileCleanup 段所示，在作业完成之后，这些文件就会被删除。如果 cleanup 段被注释掉了，那么对此作业的后续调用就会产生一些副作用：将输出附加到同名文件之后。因此建议您在每次作业调用之后清除文件。
技巧 2：接收到状态变化时应该做些什么？
要接收作业状态的变化，我们必须要实现 stateChanged 回调函数（请参见清单 3）。
清单 3. 接收状态变化
/**
* 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());
}
}
}
}
|
接收到 stateChanged 回调时，必须检查作业状态，并相应地通知正在等待的线程。上面的子程序会输出作业状态，通知正在等待的线程，然后适当地释放资源。
技巧 3：作业设置
在这个步骤中，我们需要正确设置作业参数。例如，如果指定了一个作业文件，那么就会使用这个文件的内容创建一个 GramJob 对象。还可设置作业超时。另外还必须定义授权和消息保护之类的安全参数。授权可以是用户、主机（默认）或身份标识。消息保护可以是加密或 XML 签名。其他选项包括委托、持续时间以及终止时间（请参见清单 4）。设置好这些参数后，即可提交作业了。
清单 4. 作业参数的设置
/**
* 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);
}
|
技巧 4：作业处理
创建好对象并设置好其参数后，我们就可以对作业进行处理了。在这个步骤中需要考虑的问题如下：
- 加载一个有效的代理证书 —— 可以使用默认的用户代理，也可以动态创建一个自定义代理。
- 创建一个惟一的作业标识符 —— Axis 工厂 org.apache.axis.components.uuid.UUIDGenFactory 可用于生成惟一的字符串标识符。
- 如果这个作业是以非批处理模式发送的，那么其父类就应该为作业状态变化添加一个监听程序。
- 作业必须提交给工厂端点。
- 如果作业的提交模式为非批处理，那么客户机必须要等待作业完成，并相应地销毁作业资源（请参见清单 5）。
清单 5. 作业处理
/**
* 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);
}
}
}
|
技巧 5：等待作业完成（非批处理模式）
如果提交模式是非批处理，那么主线程就必须等待作业完成，并将状态信息返回给客户机。这是通过检查作业的状态并通过循环判断作业终止原因来实现的。作业终止的原因可能是作业故障（通常是在服务器端产生的错误）、客户机超时或未知错误（请参见清单 6）。
清单 6. 等待作业完成
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;
}
}
}
}
|
准备好测试了？
最后，我做好了一切准备，可以进行简单的测试了。按清单 7 设置参数，即可测试这个自定义 WS-GRAM 客户机。运行输出结果如清单 8 所示。
清单 7. WS-GRAM 客户机测试
// 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();
}
|
清单 8. 清单 7 运行的输出结果
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
|
|
回页首 |
|
故障诊断和调试技巧
下面是用来检索作业输出结果、诊断常见错误和调试代码的一些技巧。
技巧 1：检索作业输出结果
WS-GRAM Java API 目前不能检索作业的输出结果。例如，如果我们在一台远程主机上运行 /bin/hostname 命令，就会得到这个作业的一个惟一标识符，但此 API 不能将该命令的输出结果回显到控制台。C-API 可胜任此类任务。考虑下面的命令。
清单 9. C-API 命令
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.
|
WS-GRAM C-client 提交清单 2 中给出的作业描述文件，并返回或将输出结果回显到客户机控制台上。WS-GRAM Java API 目前无法处理此类任务。
Globus 尚未在当前版本的工具包中实现将 Java 输出流回显到客户机上的功能的原因不明。或许是由于时间和资源不足。
技巧 2：故障诊断
下面让我们来看看创建自定义 WS-GRAM 客户机时遇到的两个常见错误，以及排除这些错误的方法。
清单 10. 第一个常见错误
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)
...
|
这个错误是由于 Axis Web 服务客户机不能找到安全服务 HTTPS 的协议定义而导致的。要排除这个错误，在启动脚本中添加如下 VM 参数即可：
-Daxis.ClientConfigFile=[GLOBUS_LOCATION]/client-config.wsdd
|
清单 11. 第二个常见错误
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)
...
|
这个错误是因 WSRF 代码无法找到 GT4 配置目录和其他必需组件的位置而导致的。要排除这个错误，添加如下 VM 参数即可：
-DGLOBUS_LOCATION=[GLOBUS_LOCATION]
|
技巧 3：启用调试
如果您遇到了无法确定的未知错误，请在 log4j.properties 文件中启用调试，如下所示。要使其生效，log4j.properties 文件应该在您的 classpath 目录中。然后应可看到大量调试消息，帮助您判断到底发生了什么情况。
清单 12. 在 log4j.properties 中启用调试
# 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
|
|
回页首 |
|
结束语
GRAM 服务为那些有权访问网格环境中作业宿主资源的用户提供了将作业安全地提交到多种类型的作业调度程序的功能。有效代理的存在是作业提交的必要条件。所有的 GRAM 作业提交选项都可以通过嵌入请求文档输入而透明地进行支持。实际上，作业启动是通过提交客户端向 GRAM 服务所提供的作业描述来实现的。
本文是一篇 WS-GRAM 快速入门，旨在帮助您开发可用于 Web 服务、Web 应用程序或您的自定义作业提交程序的客户机程序。
|
回页首 |
|
下载
| 描述 | 名字 | 大小 | 下载方法 |
| Source code |
gr-wsgramclient.java |
14KB |
HTTP |
原文转自：http://www.ltesting.net
- 评论列表（网友评论仅供网友表达个人看法，并不表明本站同意其观点或证实其描述）
-
| 
模板
教程
环境
性能
功能
管理
