Managing and monitoring the nodes through 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:

• it 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 notifications forwarding mechanism automatically adapts to node connecting and disconnecting from the grid, 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 node forwarding MBean

The forwarding mechanism relies extensively on a JPPF driver MBean, which exposes the NodeForwardingMBean interface. All the methods in this interface take at least a NodeSelector parameter to specify which set of nodes they apply to. Each method will invoke a corresponding method of a node MBean on each of the selected nodes. The method invoked for each node has the same signature as the NodeForwardingMBean method, minus the NodeSelector.

The results of invoking the method are then grouped into an instance of ResultsMap<String, R>, which maps the UUID of each selected node to the result of invoking the MBean method on this node. Each invocation result is an instance of InvocationResult<R>, which wraps either the actual invocation result or an eventual exception that may have occurred. The parametrized type R is the type of result that is returned by the MBean method called on each node. For MBean methods that return void, R will be instantiated as the Void type and the returned vlaue will always be null.

To illustrate these concepts, the following example retrieves the state of all the nodes in a single method call:

JMXDriverConnectionWrapper driver = ...;
// get a proxy to the NodeForwardingMBean
NodeForwardingMBean forwarder = driver.getForwarder();
// get the state of all the nodes
ResultsMap<String, JPPFNodeState> results = forwarder.state(NodeSelector.ALL_NODES);
// iterate on all the entries in the results map
for (Map.Entry<String, InvocationResult<JPPFNodeState>> entry: results.entrySet()) {
  String uuid = entry.getKey();
  InvocationResult<JPPFNodeState> result = entry.getValue();
  if (result.isException()) {
    // process the exception
    System.out.println("node " + uuid + " returned an exception: " + result.exception());
  } else {
    // process the result
    System.out.println("node " + uuid + " returned " + result.result());
  }
}

1.1 Reflective invocations

NodeForwardingMBean provides a number of generic methods for reflective invocations, which work with JPPF built-in node MBeans as well as with custom node MBeans:

public interface NodeForwardingMBean extends NotificationEmitter, ForwardingNotficationEmitter {
  // Invoke a method on the specified MBean of the selected nodes
  <E> ResultsMap<String, E> forwardInvoke(NodeSelector selector, String mbeanName,
    String methodName, Object[] params, String[] signature) throws Exception;

  // Convenience method to invoke an MBean method that has no parameter
  <E> ResultsMap<String, E> forwardInvoke(NodeSelector selector, String mbeanName,
    String methodName) throws Exception;

  // Get the value of an attribute of the specified MBean for each selected node
  <E> ResultsMap<String, E> forwardGetAttribute(NodeSelector selector, String mbeanName,
    String attribute) throws Exception;

  // Set the value of an attribute of the specified MBean on the selected nodes
  ResultsMap<String, Void> forwardSetAttribute(NodeSelector selector, String mbeanName,
    String attribute, Object value) throws Exception;
}

Example usage:

JMXDriverConnectionWrapper driver = ...;
NodeForwardingMBean forwarder = driver.getForwarder();

// get the state of all the nodes
ResultsMap<String, JPPFNodeState> results1 = forwarder.forwardInvoke(NodeSelector.ALL_NODES,
  JPPFNodeAdminMBean.MBEAN_NAME, "state");

// update the thread pool size to 4 for all the nodes
ResultsMap<String, JPPFNodeState> results2 = forwarder.forwardInvoke(NodeSelector.ALL_NODES,
  JPPFNodeAdminMBean.MBEAN_NAME, "updateThreadPoolSize",
  new Integer[] { 4 }, new String[] { Integer.class.getName() });

// get the total number of executed tasks for all the nodes
ResultsMap<String, Integer> results3 = forwarder.forwardGetAttribute(NodeSelector.ALL_NODES,
  JPPFNodeTaskMonitorMBean.MBEAN_NAME, "TotalTasksExecuted");

// set the class loader delegation model on all the nodes
ResultsMap<String, Void> results4 = forwarder.forwardSetAttribute(NodeSelector.ALL_NODES,
  JPPFNodeAdminMBean.MBEAN_NAME, "DelegationModel", DelegationModel.PARENT_FIRST);

1.2 Other convenience methods

NodeForwardingMBean also provides some convenience methods to forward requests to the JPPFNodeAdminMBean, JPPFNodeProvisioningMBean and DiagnosticsMBean node MBeans. These methods have a signature similar to the corresponding methods in the node MBeans, with two differences:

• they have an additional NodeSelector parameter
• instead of returning a single value, they return a ResultsMap<String, R> associating a return value with each sleected node

The main advantage of these methods is that they are much less cumbersome than reflective invocations.

For example, the following code retrieves a thread dump for all the selected nodes, which is equivalent to calling DiagnosticsMBean.threadDump() on each of the nodes:

JMXDriverConnectionWrapper driver = ...;
NodeForwardingMBean forwarder = driver.getForwarder();
// get a thread dump for each node
ResultsMap<String, ThreadDump> results = forwarder.threadDump(NodeSelector.ALL_NODES);

2 Registering JMX notification listeners

Contrary to other MBeans, the registration and unregistration of notification listers which forward notifications from selected nodes is not done with the addNotificationListener(...) and removeNotificationListener(...) methods in NodeForwardingMBean.

Instead, you must use the following methods exposed by JMXDriverConnectionWrapper:

public class JMXDriverConnectionWrapper extends JMXConnectionWrapper
  implements JPPFDriverAdminMBean {
  // Register a notification listener that receives notifications
  // from the specified MBean on the selected nodes
  public String registerForwardingNotificationListener(NodeSelector selector, String mBeanName,
    NotificationListener listener, NotificationFilter filter, Object handback) throws Exception

  // Unregister a previously registered forwarding notification listener
  public void unregisterForwardingNotificationListener(String listenerID) throws Exception

  // Uneregister all previously registered forwarding notification listeners
  public List<String> unregisterAllForwardingNotificationListeners() throws Exception
}

Note that registerForwardingNotificationListener() returns a string that represents the id of the registered listener. This id is unique for a given JMXDriverConnectionWrapper instance, and is the parameter that must be provided to the unregisterForwardingNotificationListener(String) method.

The notifications received by the listener are of type JPPFNodeForwardingNotification which, in addition to carrying the original notification emitted by the node, also encapsulates information to identify the originating node and MBean. The class JPPFNodeForwardingNotification is defined as follows:

public class JPPFNodeForwardingNotification extends Notification {
  // Get the actual notification forwarded from the node
  public Notification getNotification()

  // Get the uuid of the originating node
  public String getNodeUuid()

  // Get the name of the originating MBean in the node
  public String getMBeanName()
}

The following example registers a listener which receives otifications from the JPPFNodeTaskMonitorMBean on all the nodes with a t least 4 CPUs:

JMXDriverConnectionWrapper driver = ...

// listener definition
NotificationListener listener = (notif, handback) -> {
  // cast the forwarding notification to its actual type
  JPPFNodeForwardingNotification forwardingNotif = (JPPFNodeForwardingNotification) notif;
  // cast the original notification to its actual type
  TaskExecutionNotification taskNotif = 
    (TaskExecutionNotification) forwardingNotif.getNotification();
  // process the notification
  if (taskNotif.isUserNotification()) {
    System.out.println("received the task message " + taskNotif.getUserData() + 
      " from the node " + forwardingNotif.getNodeUuid());
  }
};

// select the nodes with at least 4 cpus
NodeSelector selector = new ExecutionPolicySelector(new AtLeast("availableProcessors", 4));
// register the listener and capture its listenerId
String listenerId = driver.registerForwardingNotificationListener(
  selector, JPPFNodeTaskMonitorMBean.MBEAN_NAME, listener, null, null);

// ... execute tasks ...
...

// when done, unregister the listener
driver.unregisterForwardingNotificationListener(listenerId);

Instead of a standard NotificationListener, you can use the ForwardingNotificationListener interface, defined as:

public interface ForwardingNotificationListener extends NotificationListener {
  // Called when a JPPFNodeForwardingNotification occurs
  void handleNotification(JPPFNodeForwardingNotification notification, Object handback);

  @Override
  default void handleNotification(Notification notification, Object handback) { ... }
}

It provides a convenient way to avoid casting the received notification to JPPFNodeForwardingNotification. Using this listener interface, the listener defintion in the example above becomes:

ForwardingNotificationListener listener = (notif, handback) -> {
  // cast the original notification to its actual type
  TaskExecutionNotification taskNotif = (TaskExecutionNotification) notif.getNotification();
  // process the notification
  if (taskNotif.isUserNotification()) {
    System.out.println("received the task message " + taskNotif.getUserData() + 
      " from the node " + notif.getNodeUuid());
  }
};

3 Node forwarding static proxies

Instead of using the cumbersome NodeForwardingMBean interface to forward JMX requests and notifications to the nodes, you can also make use of the more convenient MBean forwarding proxy classes in the org.jppf.management.forwarding.generated package. These proxy classes are modeled after the node MBean interfaces for which they implement the forwarding mechainsm, with the following differences:

1) The class name is suffixed with "Forwarder". For instance, the forwarder for the JPPFNodeAdminMBean MBean resolves to JPPFNodeAdminMBeanForwarder.

2) Each method in the forwarding proxy has the same name as the equivalent method in the single-node MBean, but its signature and return type are different:

• the method takes an additional NodeSelector as first parameter
• the return type is a ResultsMap<String, R>, where R is the return type of the equivalent single-node method.

For example the following single-node method:

public interface JPPFNodeAdminMBean extends JPPFAdminMBean {
  JPPFNodeState state() throws Exception;
}

is transformed into:

public class JPPFNodeAdminMBeanForwarder extends AbstractMBeanForwarder {
  public ResultsMap<String, JPPFNodeState> state(NodeSelector selector) throws Exception { ... }
}

3) All forwarding proxy classes inherit from AbstractMBeanForwarder, defined as follows:

public class AbstractMBeanForwarder {
  // Initialize this proxy with the specified driver jmx connection and mbean name 
  public AbstractMBeanForwarder(JMXDriverConnectionWrapper jmx, String mbeanName) throws Exception

  // Invoke a method on the specified MBean of the selected nodes attached to the driver
  public <E> ResultsMap<String, E> invoke(NodeSelector selector, String methodName,
    Object[] params, String[] signature) throws Exception

  // Convenience method to invoke an MBean method that has no parameter
  public <E> ResultsMap<String, E> invoke(NodeSelector selector, String methodName)
    throws Exception

  // Get the value of an attribute of the specified MBean for each selected node
  public <E> ResultsMap<String, E> getAttribute(NodeSelector selector, String attribute)
    throws Exception

  // Set the value of an attribute of the specified MBean on the selected nodes
  public <E> ResultsMap<String, Void> setAttribute(NodeSelector selector,
    String attribute, E value) throws Exception

  // Add a listener to this MBean for all selected nodes, without filter or handback object
  public String addNotificationListener(NodeSelector selector, NotificationListener listener)
    throws Exception

  // Add a listener to this MBean for all selected nodes
  public String addNotificationListener(NodeSelector selector, NotificationListener listener,
    NotificationFilter filter, Object handback) throws Exception

  // Remove a listener from this MBean 
  public void removeNotificationListener(String listenerID) throws Exception

  // Remove all listeners registered from this MBean forwarder
  public void removeAllNotificationListeners() throws Exception
}

As we can see, AbstractMBeanForwarder has built-in methods for reflective fowarding invocations and for notification listeners registration.

4) A forwarding proxy can be created either directly, using its constructor, or with the getMBeanForwarder() helper method in JMXDriverConnectionWrapper.

The following example illustrates how the forwarding proxies can be used:

JMXDriverConnectionWrapper driver = ...;

// instantiate via JMXDriverConnectionWrapper.getMBeanForwarder()
JPPFNodeAdminMBeanForwarder adminForwarder = driver.getMBeanForwarder(JPPFNodeAdminMBean.class);
// get the states of all nodes
ResultsMap<String, JPPFNodeState> states = adminForwarder.state(NodeSelector.ALL_NODES);

// instantiate directly via the proxy's constructor
JPPFNodeTaskMonitorMBeanForwarder taskForwarder = new JPPFNodeTaskMonitorMBeanForwarder(driver);
// get the number of executed tasks for all the nodes
ResultsMap<String, Integer> nbTasks = taskForwarder.getTotalTasksExecuted(NodeSelector.ALL_NODES);

// register a listener that receives notfications from the tasks in all selected nodes
ForwardingNotificationListener listener =
  (notif, handback) -> System.out.println("got notification: " + notif);
// select the nodes with at least 4 cpus
NodeSelector selector = new ExecutionPolicySelector(new AtLeast("availableProcessors", 4));
// listen to tasks notifications from the selected nodes
String listenerId = taskForwarder.addNotificationListener(selector, listener);

// ... execute tasks
...

// when done, unregister the listener
taskForwarder.removeNotificationListener(listenerId);