Difference between revisions of "Executor"

Latest revision as of 12:01, 22 May 2015

The Executor acts as a container for gCube tasks, i.e. functionally unconstrained bodies of code that lack a network interface but can be dynamically deployed into the service and executed through its interface. In particular, gCube tasks are designed, packaged, and deployed as plugins of the Executor service.

An instance of the Executor publishes descriptive information about the co-deployed tasks, can execute them on demand on behalf of clients, and can inform clients about the state of their execution. Clients may interact with the Executor service through a library of high-level facilities that subsumes standard service stubs to simplify the discovery of service instances and the execution of tasks available in those instances.

Design

Like all services that can be dynamically extended with plugins, the Executor has a plugin manager that accepts requests to register or deregister plugins of gCube tasks. The requests are not issued by service clients, however. They are issued by the Deployer in response to the availability of tasks in the infrastructure. The manager persists plugin profiles to autonomically re-register them at each container restarts.

Clients interact with either one of two port-types, the Engine and the Task.

The Engine port-type is the point of contact for clients that wish to launch the execution of registered tasks. The port-type is stateful, in that it maintains descriptions of the available tasks in a single stateful resource, the engine. The engine is created at service startup, when it subscribes with the plugin manager to be notified of plugin registration and de-registration events. It is then bound to the port-type into a WS-Resource accessible to clients via the implied resource pattern of WSRF. The task descriptions are modelled as a single, multi-valued Resource Property (RP) of the WS-Resource and published at regular intervals in all the scopes of the service instance. Task descriptions include the name of the task, a textual description for it, a set of arbitrary-valued properties, prototypical examples of the task inputs, and prototypical examples of the task outputs. The precise definition of the RP and the signature of the launch operation can be found in the WSDL of the Engine port-type.

The Task port-type is the the point of contacts for clients that wish to monitor the execution of tasks. The port-type is stateful, in that it maintains information about the execution of tasks in dedicated task resources. Task resources are created by the engine when tasks are launched, and are bound to the port-type into WS-Resources available via the implied resource pattern of WSRF. The execution state is modelled as RPs of the WS-Resources and published in all the scopes of the WS-Resources at regular intervals. RPs include the start time, inputs, and current state of the execution, as well as the logs, outputs and errors produced by the task. The port-type exposes a single operation to stop a running task (which is designed to be stopped) as well as standard operations of the gCube Notification Provider.

Task resources interact with running tasks by injecting them with a task runtime object in which they can find what they need to consume (e.g. inputs) and place what they need to produce (e.g. logs, outputs, errors). Task resources also inject tasks with a logger that redirects transparently to the resources all the logs produced by the tasks. It is through the runtime that task resources can publish the current state of execution in RPs of WS-Resources.

Finally, special treatment is given to scheduled tasks, i.e. tasks that execute at fixed intervals indefinitely or as long as certain conditions are verified. First, their task resources are persisted and the execution of the task resumed after a container startup (it would be unsound to restart non scheduled tasks). Second, their task resources subscribe with the plugin manager to be notified if the plugin of the task is deregistered; in this case, the task is stopped at the end of the current schedule (this is not generally possible with non scheduled tasks).

Sample Usage

The Executor service provides a client library to simplify the following procedures:

• discover service instances that can execute the target task. This requires interaction with the Information System.
• launch the execution of the task with one the discovered instances. This requires interaction with the Engine port-type of the Executor.
• monitor the execution of the running task. This requires interaction with the Task port-type of the Executor.

Launching Tasks

ExecutorCalls can be used to launch tasks. Clients create one with the name of the task and the scope in which the task should execute:

String name;
GCUBEScope scope;
...
call = new ExecutorCall(name,scope);

In a secure infrastructure, they will also provide a security manager:

GCUBESecurityManager secmanager,
.....
call = new ExecutorCall(name,scope, secmanager);

Clients may now launch tasks as follows:

TaskCall task = call. launch();

where the TaskCall returned by launch allows clients to act upon the running task, as we discuss below.

Best Effort and Known Endpoints

At launch time, an ExecutorCall will transparently discover Executor instances that can execute the target task and ask each of them to execute the task, until one succeeds. Clients can set task properties to refine discovery:

String propertyName, propertyValue;
...
call.setTaskProperty(propertyName,propertyValue);

Client can also bypass discovery altogether if they know the Engine endpoint of an Executor instance that can execute the task:

String hostname, port;
...
call.setEndpoint(hostname,port);

or, directly:

EndpointReferenceType epr;
...
call.setEndpointReference(epr);

Task Inputs

Clients may also pass inputs to tasks that expect them:

Map<String,Object> inputs;
...
TaskCall task = call.launch(inputs);

Clients may also use non-primitive input types that are specific to the task:

String input2name = ....
MyType input2value = ...
inputs.put(input2Name, input2Value);

In this case, however, clients must register with the MappingRegistry the type mapping required to serialise MyType instances (a type mapping is a correspondence between a type and its serialisation on the wire). As type mappings are explicitly provided by the context of the plugin of the target task (see below), e.g. an instance of MyPluginContext, clients can conveniently register them as follows:

MyPluginContext pcontext = new MyPluginContext();
MappingRegistry.register(context.getTypeMappings().toArray(new TypeMapping[0]));

Do also note that clients of tasks that expect specific types have an explicit dependency on task implementations.

Stopping and Monitoring Tasks

TaskCalls can be used to interact with running tasks. Clients that use ExecutorCalls obtain a TaskCall when they launch a task, but any client may create one for any running task. TaskCalls are instantiated exactly like ExecutorCalls, i.e. with a name, a scope, and an optional security manager. They follow the same best-effort strategy and offer the same facilities to customise discovery (cf. setTaskProperty()) or bypass it altogether (cf. setEndpointReference()).

Clients may use TaskCalls to obtain a local proxy of the running task:

TaskCall task;
....
TaskProxy proxy = task.getProxy();
System.out.format("Task invoked started at %Tc with %s state",proxy.getStartTime(),proxy.getState());

The proxy's methods mirror the RPs of the WS-Resource that models the execution of the target task. However, they execute against a local cache of the RP values and do not engage the remote WS-Resource. The cache is populated when the proxy is created but may be explicitly synchronised thereafter:

proxy.synchronize();

Clients may also subscribe for changes to the running task, such as a new execution state or a new output:

TaskMonitor monitor;
....
task.subscribe(monitor);

TaskMonitor is an abstract class that defines callbacks for TaskMonitor.TaskTopics. Clients subclass TaskMonitor to implement the callbacks of interest, typically with an anonymous class:

TaskMonitor monitor = new TaskMonitor() {
      public void onStateChange(TaskProxy proxy) throws Exception {
        if (proxy.getState().equals(State.Failed.INSTANCE.toString())) { //state values are the string conversion of handler's states
	  System.out.println("task has failed with error "+proxy.getError());
        }
        else if (proxy.getState().equals(State.Done.INSTANCE.toString())) {
          System.out.println("task has completed with: "+proxy.getOutput().get("endresult")));
        }
        else logger.info("task has moved to status "+state);
     }
 
     public void onOutputChange(TaskProxy proxy) { 
        if (proxy.getOutput().containsKey("endresult"))
	logger.info("output message is "+proxy.getOutput().get("endresult")));
     }
};

The callbacks yield a task proxy that has been synchronised and thus reflect the change. Here the monitor subscribes to both topics and implements both callbacks. Optionally, a monitor may limit its interest to a single TaskTopic by passing it to the constructor of TaskMonitor:

TaskMonitor monitor = new TaskMonitor(TaskMonitor.STATECHANGE) {
      public void onStateChange(TaskProxy proxy) throws Exception {...}
};

Finally, clients may attempt to stop a task:

task.stop();

The task must be designed for this. An GCUBEUnrecoverableException indicates that this is not the case, while a GCUBERetrySameException shows that even though the task can be stopped the attempt failed.

Plugin Development

Executor plugins may have arbitrary size and dependencies but must include the following components:

• an implementation of the ExecutorTask interface which embodies the task;

• a subclass of the ExecutorPluginContext class which provides information about the task;

• a profile that binds the plugin to the Executor service.

Tasks

ExecutorTask is the interface required of all gCube Tasks. It is a tagging interface defined by the Executor service:

public interface ExecutorTask extends Lifetime<TaskRuntime> {}

where Lifetime is a parametric interface of the Handler Framework. The framework is part of gCore and offers a rich set of features for the development of sophisticated ExecutorTask. We assume familiarity with it in the following.

In the terminology of the Handler Framework, an ExecutorTask is a handler that manages its own lifetime (so that the Executor can monitor it) and that handles objects of type TaskRuntime, the role of which we discuss below. Notice that, like any other handler, an ExecutorTask can subclass any of the generic handler types defined in the Handler Framework.

The following is possibly the simplest ExecutorTask, for which we subclass the generic GCUBEHandler. Note that the task honors the commitment to manage its own lifetime by invoking the method setState() of GCUBEHandler with the lifetime states that are predefined for handlers in the Handler Framework.

class HelloTask extends GCUBEHandler<TaskRuntime> implements ExecutorTask {
 
    public void run() throws Exception {
       this.setState(Running.INSTANCE);
       this.getLogger().trace("hello world");
       this.setState(Done.INSTANCE);
    }
 
}

What really differentiates a task from any other handler is the handled TaskRuntime object. As discussed above, the Executor injects it into the task before executing it and that the task can use it to retrieve any inputs that clients may pass to the Executor, to access the context of the plugin, and to produce outputs and errors. Here is an example:

class EchoTask extends GCUBEHandler<TaskRuntime> implements ExecutorTask {
 
    public void run() throws Exception {
       TaskRuntime r = this.getHandled();
       this.setState(Running.INSTANCE);
       this.setState(Suspended.INSTANCE);
       this.getLogger().info("pausing...");
       Thread.sleep(1000);
       if (Math.random()>.8) {
    	   r.throwException(new Exception("random problem"));
    	   this.setState(Failed.INSTANCE);
       }
       else  {
    	   r.addOutput("echo","hello "+r.getInput("clientname"));
           this.setState(Done.INSTANCE);
       }
    }
}

Here the task uses its runtime to throw exceptions (method throwException()), to produce outputs (method addOutput()) and to retrieve inputs by name (method getInput()). Consult the documentation of TaskRuntime for a complete list of available methods. Note also that the logs emitted by the task will be transparently intercepted by the Executor service, published in the scope where the task will be launched, and made available to clients that will monitor its execution.

In conclusion, developing an ExecutorTask is no different from developing any other handler. One must only make sure to honor the Lifetime interface, make use of the handled TaskRuntime object, and satisfy the following requirement:

Note: Tasks are instantiated by the Executor and thus must have have a zero-argument constructor.

Contexts

A plugin context is the entry point to the implementation of a plugin and exposes information that allows the Executor to manage it.

Part of this information is common to all plugins and is required by gCore to act on behalf of the Executor for the most generic aspects of plugin management. This includes:

• zero or more descriptive properties whereby the plugin may be identified by the clients of the Executor;
• zero or more type mappings that may be required to deserialise object types that are statically unknown to the Executor but must be exchanged on the wire between its clients and the plugin (cf. discussion above).

Other information is instead required specifically by the Executor, including:

• the implementation of ExecutorTask which embodies the task included in the plugin;
• zero or more sample inputs required by the task included in the plugin;
• zero or more sample outputs produced by the task included in the plugin;
• the time after which any trace of tasks that have failed or successfully completed will be removed.

GCUBEPluginContext is defined in gCore as the root of all plugin contexts. It defines a number of callbacks whereby gCore can obtain the generic information it requires. ExecutorPluginContext derives GCUBEPluginContext in the Executor and repeats the pattern for the information it specifically requires. Plugin developers must derive ExecutorPluginContext to implement the callbacks and return information specific to their own plugin.

To simplify the process, GCUBEPluginContext implements its callbacks in terms of empty data structures (e.g. a list of DescriptivePropertys and a list of TypeMappings). Plugin developers may then simply populate these structures in the constructors of their own contexts (e.g. by invoking addDescriptiveProperty() and addTypeMapping()). The ExecutorPluginContext does the same for its own callbacks (cf. addSampleOutput() and addSampleInput()). One callback, getTaskClass(), is left abstract and plugin developers must implement it and return the class the class that implements ExecutorTask in their plugin. Finally, developers may override the implementation of getTimeToLive() to override the default time-to-live of the tasks in their plugin. The following example illustrates:

public class MyPluginContext extends ExecutorPluginContext {
 
   QName name = new QName("...", "ComplexProperty");
 
   public MyPluginContext() {
	this.addProperty(
	   new DescriptiveProperty("A simple property","property1","value1"),
	   new DescriptiveProperty("A complex property","property2",new ComplexProperty("value2"))
        );
 
	this.addTypeMappings(new TypeMapping(ComplexProperty.class,name)));
 
	this.addSampleInput(new DescriptiveProperty("A complex input","parameter1",new ComplexProperty("sample")));
	this.addSampleOutput(new DescriptiveProperty("A complex output","endresult",new ComplexProperty("sample")));	
   }
 
 
   /**{@inheritDoc} */
   public Class<? extends ExecutorTask> getTaskClass() {return MyTask.class;}
 
   /**{@inheritDoc} */
   public int getTimeToLive(){return 10;} //seconds!
 
}

Here the context constructor is used to populate the inherited data structures with properties, type mappings, sample inputs and sample outputs. ComplexProperty is a type defined within the plugin and used for the value of one plugin property, the value of the single task input, and the value the single task output. A type mapping is defined for it, accordingly. The class of MyTask is returned from getTaskClass() and getTimeToLive() is overridden to return a value of 10 seconds. Consult the documentation of ExecutorPluginContext for the full list of available methods and for the details of their signatures.

Note:Plugin contexts are instantiated by gCore and this requires them to have a zero-argument constructor.

Plugin Profiles

Plugins, as any piece of software developed for the gCube platform, must be profiled in order to be properly manager in a gCube infrastructure.

Here is an example of plugin profile:

<?xml version="1.0" encoding="UTF-8"?>
<Resource xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
    <ID/>
    <Type>Service</Type>
    <Profile>
        <Description>A Sample Plugin</Description>
        <Class>Samples</Class>
        <Name>SamplePlugin</Name>
        <Version>1.0.0</Version>
        <Packages>
            <Plugin>
            	<Name>plugin</Name>
            	<Version>1.0.0</Version>
            	<TargetService>
            		<Service>
            			<Class>VREManagement</Class>
           			<Name>Executor</Name>
            			<Version>1.0.0</Version>
            		</Service>
            		<Package>Main</Package>
          		<Version>1.0.0</Version>
            	</TargetService>
                <EntryPoint>org.acme.sample.plugin.PluginContext</EntryPoint>
                <Files><File>org.acme.sample.plugin.jar</File></Files>
            </Plugin>
        </Packages>
    </Profile>
</Resource>

With respect to other software profiles, the two distinguish characteristics of this profile are:

• the declaration of the TargetService, i.e. the service that will load and run the plugin
• the EntryPoint, i.e. the class extending the PluginContext

Development Environment and Testing

This Eclipse project] may be used to kick-off plugin development and testing.

The project:

• assumes the definition of user libraries for gCore and plugin dependencies in the hosting workspace (cf. GCORELIBS and PLUGINDEPS). The developer must at least add to the latter dependencies to the implementation of the Executor and its stubs. * pre-defines a context, a task, and a profile for the plugin (cf. PluginContext, Task, and profile.xml). The developer can choose the physical placement of dependencies, though the project assumes them to be outside the service implementation (cf. build.properties).
• includes a customisation of the standard gCore buildfile that processes the types under the folder schema and output corresponding stub class under the package types, from which they can be imported in the rest of the implementation. As usual, configurable build properties are in build.properties.
• defines a mock of the TaskRuntime that a running instance of the Executor will inject into the task before running it (cf. MockRuntime). This allows most tasks to be tested without deploying them in a running instance of the Executor. The sample test included in the project exemplifies how to use these facilities (cf. Test). Here is the significant excerpt:

....
//load plugin profile as classpath resource
GCUBEService plugin = GHNContext.getImplementation(GCUBEService.class);
plugin.load(new InputStreamReader(Test.class.getResourceAsStream("profile.xml")));
 
MockRuntime runtime = new MockRuntime(plugin);
 
//prepare manually mock runtime as a running instance of the Executor would.
runtime.getInputs().put("parameter1", new ComplexType("testvalue"));
 
 
//prepare task as a running instance of the Executor would.
GCUBEScopeManager manager = new GCUBEScopeManagerImpl();
manager.setScope(GCUBEScope.getScope("/gcube/devsec")); //would use args normally
 
//launch task as a running instance of the Executor would.
Task task = new Task();
task.setScopeManager(manager);
task.setHandled(runtime);
task.run();