Nodes management and monitoring via the driver

JPPF provides support for forwarding JMX requests to the nodes, along with receiving notifications from them, via the JPPF driver's JMX server. Which nodes are impacted is determined by a user-provided node selector.

This brings two major benefits:

• this allows managing and monitoring the nodes in situations where the nodes are not reachable from the client, for instance when the client and nodes are on different networks or subnets
• the requests and notification forwarding mechanism automatically adapts to node connection and disconnection events, which means that if new nodes are started in the grid, they will be automatically enrolled in the forwarding mechanism, provided they match the node selector

1 The JPPFNodeForwardingMBean interface

All forwarding operations are performed by a special MBean in the driver, exposing the JPPFNodeForwardingMBean interface. The object name of this mbean is "org.jppf:name=provisioning,type=node", also the value of the constant JPPFNodeForwardingMBean.MBEAN_NAME.

As we can see in the Javadoc, all the mthods in this interface return a Map<String, Object>, where:

• the key represents the UUID of a node to which the request was forwarded
• the value is either the return value of the forwarded request for this specific node, or an Exception that was raised when executing the request for the node

We can also see that all methods of JPPFNodeForwardingMBean take a NodeSelector as parameter, specifying the nodes to which the request is forwarded. The methods forwardInvoke(), forwardGetAttribute() and forwardSetAttribute() are general methods applied to a specified MBean of each selected node. The other methods apply to predefined MBeans registered in the node.

For instance, you could get the states of selected nodes in two different ways, as follows:

NodeSelector selector = NodeSelector.ALL_NODES;
JPPFNodeForwardingMBean forwarder = ...;
Map<String, Object> result = new HashMap<>();
// the state of the nodes can be obtained generically by calling forwardInvoke()
result = forwarder.forwardInvoke(selector, JPPFNodeAdminMBean.MBEAN_NAME, "state");
// the state can also be obtained by calling state() without need for an mbean name
result = forwarder.state(selector);

2 Node selectors

All the node forwarding APIs make use of node selectors to conveniently specify which nodes they apply to. A node selector is an instance of the NodeSelector interface, defined as follows:

// Marker interface for selecting nodes
public interface NodeSelector extends Serializable {
  // Constant for a selector which accepts all nodes
  NodeSelector ALL_NODES = new AllNodesSelector();
  // Determine whether the node is accepted
  boolean accepts(JPPFManagementInfo nodeInfo);
}

The participation of a node is determined by the accepts() method, which takes a JPPFManagementInfo argument that represents information on the node as seen by the server. In particular, it provides access to the same node-specific JPPFSystemInformation object that is used by execution policies.

2.1 Selecting all the nodes

AllNodesSelector will select all the nodes currently attached to the server. Rather than creating instances of this class, you also use the predefined constant NodeSelector.ALL_NODES. It is defined as:

// Selects all nodes
public class AllNodesSelector implements NodeSelector {
  // Default constructor
  public AllNodesSelector()
}

Example usage:

// these two selectors are equivalent
NodeSelector selector1 = new AllNodesSelector();
NodeSelector selector2 = NodeSelector.ALL_NODES;

2.2 Execution policy selector

ExecutionPolicySelector uses an execution policy, such as can be set upon a JPPF job's SLA to perform the selection of the nodes. It is defined as follows:

// Selects nodes based on an execution policy
public class ExecutionPolicySelector implements NodeSelector {
  // Initialize this selector with an execution policy
  public ExecutionPolicySelector(final ExecutionPolicy policy)
  // Get the execution policy to use to select the nodes
  public ExecutionPolicy getPolicy()
}

Example usage:

// define a selector that selects all nodes with at least 4 cores
ExecutionPolicy policy = new AtLeast("availableProcessors", 4); 
NodeSelector selector = new ExecutionPolicySelector(policy);

2.3 UUID-based selector

UuidSelector will only select nodes whose UUID is part of the collection or array of specified UUIDs. This class is defined as:

// Selects nodes based on their uuids
public final static class UuidSelector implements NodeSelector {
  // Initialize this selector with a collection of node UUIDs
  public UuidSelector(final Collection<String> uuids)
  // Initialize this selector with an array of node UUIDs
  public UuidSelector(final String...uuids)
  // Get the collection of uuids of the nodes to select
  public Collection<String> getUuidList()
}

Note that the node selection dynamically adjusts to the JPPF grid topology, or in other words two distinct selections with the same selector instance may return a different set of nodes: when new nodes are added to the grid or existing nodes are terminated, these changes in the toplogy will be automatically taken into account by the selection mechanism.

Example usage:

// define a selector that selects all idle nodes at a given time
JMXDriverConnectionWrapper driver = ...;
// get the uuids of all idle nodes
Collection<JPPFManagementInfo> infos = driver.idleNodesInformation();
List<String> uuids = new ArrayList<>();
if (infos ! null) {
  for (JPPFManagementInfo info: infos)
    uuids.add(info.getUuid());
}
NodeSelector selector = new UuidSelector(uuids);

2.4 Scripted node selector

A ScriptedNodeSelector uses a script expression that evaluates to a boolean to determine which nodes are accepted. The script language must be accepted by the JVM and must conform to the JSR-223 / javax.script API specifications.

A scripted node selector is particularly useful when you need to implement a complex filtering logic but do not want to deploy the associated code in the driver's classpath.

ScriptedNodeSelector extends BaseScriptEvaluator and is defined as follows:

public class ScriptedNodeSelector extends BaseScriptEvaluator implements NodeSelector {
  // Initialize this selector with the specfied language and script
  public ScriptedNodeSelector(String language, String script)
  // Initialize this selector with the specfied language and script reader
  public ScriptedNodeSelector(String language, Reader scriptReader) throws IOException
  // Initialize this selector with the specfied language and script file
  public ScriptedNodeSelector(String language, File scriptFile) throws IOException
}

The script will be evaluated for each JPPFManagementInfo passed as input to the accepts() method of the selector. The node information can be accessed from the script using a predefined variable named "nodeInfo", as in this example:

JPPFDriverAdminMBean driverAdmin = ...;
NodeSelector selector;
// create a selector with a Javascript script which accepts only master nodes
selector = new ScriptedNodeSelector("javascript", "nodeInfo.isMasterNode();");
// or an equivalent Groovy script
selector = new ScriptedNodeSelector("groovy", "return nodeInfo.isMasterNode()");
// get the number of selected nodes
int nbMasterNodes = driverAdmin.nbNodes(selector);

2.5 Custom node selector

When a node selector requires a complex logic and its performance is critical, you can always write your own implementation of NodeSelector, as in this example:

public class CustomNodeSelector implements NodeSelector {
  @Override
  public boolean accepts(JPPFManagementInfo nodeInfo) {
    // retrieve the node's number of cores
    int n = nodeInfo.getSystemInformation().getRuntime().getInt("availableProcessors");
    // accept only slave nodes with at least 4 cores
    return nodeInfo.isSlaveNode() && (n >= 4);
  }
}

Note: since the node selector is evaluated by the driver, the code of its implementation, along with all the classes it depends on, must be deployed in the driver's classpath.

3 Forwarding management requests

The request forwarding mechanism is based on a built-in driver MBean: JPPFNodeForwardingMBean, which provides methods to invoke methods, or get or set attributes on remote node MBeans. Each of its methods requires a NodeSelector argument and an MBean name, to determine to which nodes, and which MBean in these nodes, the request will be performed. The return value is always a map of node UUIDs to the corresponding value returned by the request (if any) to the corresponding node. If an exception is raised when performing the request on a specific node, then that exception is returned in the map.

Here is an example:

JPPFClient client = ...;
JMXDriverConnectionWrapper driverJmx = client.awaitWorkingConnectionPool()
  .awaitWorkingJMXConnection();

JPPFNodeForwardingMBean proxy;
// get a proxy to the node forwarding MBean
proxy = driverJmx.getProxy(
  JPPFNodeForwardingMBean.MBEAN_NAME, JPPFNodeForwardingMBean.class);
// or, in a much less cumbersome way:
proxy = driverJmx.getNodeForwarder();

// this selector selects all nodes attached to the driver
NodeSelector selector = new NodeSelector.AllNodes();
// this selector selects all nodes that have more than 2 processors
ExecutionPolicy policy = new MoreThan("availableProcessors", 2);
NodeSelector selector2 = new NodeSelector.ExecutionPolicySelector(policy);

// invoke the state() method on the remote 'JPPFNodeAdminMBean' node MBeans
// note that the MBean name does not need to be stated explicitely
Map<String, Object> results = proxy.state(selector);
// this is an exact equivalent, explicitely stating the target MBean on the nodes:
String targetMBeanName = JPPFNodeAdminMBean.MBEAN_NAME;
Map<String, Object> results2 = proxy.forwardInvoke(selector, targetMBeanName, "state");

// handling the results
for (Map.Entry<String, Object> entry: results) {
  if (entry.getValue() instanceof Exception) {
    // handle the exception ...
  } else {
    JPPFNodeState state = (JPPFNodeState) entry.getValue();
    // handle the result ...
  }
}

4 Forwarding JMX notifications

JPPF provides a way to subscribe to notifications from a set of selected nodes, which differs from the one specified in the JMX API. This is due to the fact that the server-side mechanism for the registration of notification listeners is unspecified and thus provides no reliable way to override it.

To circumvent this difficulty, the registration of the notification listener is performed via the JMX client wrapper JMXDriverConnectionWrapper:

1) To add a notification listener, use registerForwardingNotificationListener(NodeSelector, String, NotificationListener, NotificationFilter, Object). This will register a notification listener for the specified MBean on each of the selected nodes. This method returns a listener ID which will be used to remove the notfication listener later on. Thus, the application must be careful to keep track of all registered listener IDs.

2) To remove a notification listener, use unregisterForwardingNotificationListener(String listenerID).

The notifications forwarded from the nodes are all wrapped into instances of JPPFNodeForwardingNotification. This class, which inherits from Notification, provides additional APIs to identify from which node and which MBean the notification was emitted.

The following code sample puts it all together:

JPPFClient client = ...;
JMXDriverConnectionWrapper driverJmx = client.awaitWorkingConnectionPool()
  .awaitWorkingJMXConnection();

// this selector selects all nodes attached to the driver
NodeSelector selector = new AllNodesSelector();

// create a JMX notification listener
NotificationListener myListener = new NotificationListener() {
  @Override
  public void handleNotification(Notification notification, Object handback) {
    JPPFNodeForwardingNotification wrapping =
      (JPPFNodeForwardingNotification) notification;
    System.out.println("received notification from nodeUuid=" + wrapping.getNodeUuid()
      + ", mBeanName=" + wrapping.getMBeanName());
    // get the actual notification sent by the node
    TaskExecutionNotification actualNotif =
      (TaskExecutionNotification) wrapping.getNotification();
    // handle the notification data ...
  }
}

// register the notification listener with the JPPFNodeTaskMonitorMBean
// on the selected nodes
String listenerID = driverJmx.registerForwardingNotificationListener(
  selector, JPPFNodeTaskMonitorMBean.MBEAN_NAME, listener, null, null);

// ... submit a JPPF job ...

// once the job has completed, unregister the notification listener
driverJmx.unregisterForwardingNotificationListener(listenerID);