Author: bfoster
Date: Wed Dec 15 17:19:10 2010
New Revision: 1049645
URL: http://svn.apache.org/viewvc?rev=1049645&view=rev
Log:
- added the ability for Resource Manager to dynamically add/remove nodes and
modify queues at run-time
--------------------
OODT-78
Modified:
oodt/trunk/CHANGES.txt
oodt/trunk/resource/src/main/java/org/apache/oodt/cas/resource/monitor/AssignmentMonitor.java
oodt/trunk/resource/src/main/java/org/apache/oodt/cas/resource/monitor/AssignmentMonitorFactory.java
oodt/trunk/resource/src/main/java/org/apache/oodt/cas/resource/monitor/Monitor.java
oodt/trunk/resource/src/main/java/org/apache/oodt/cas/resource/system/XmlRpcResourceManager.java
oodt/trunk/resource/src/main/java/org/apache/oodt/cas/resource/system/XmlRpcResourceManagerClient.java
oodt/trunk/resource/src/main/java/org/apache/oodt/cas/resource/util/GenericResourceManagerObjectFactory.java
oodt/trunk/resource/src/test/org/apache/oodt/cas/resource/monitor/TestAssignmentMonitor.java
Modified: oodt/trunk/CHANGES.txt
URL:
http://svn.apache.org/viewvc/oodt/trunk/CHANGES.txt?rev=1049645&r1=1049644&r2=1049645&view=diff
==============================================================================
--- oodt/trunk/CHANGES.txt (original)
+++ oodt/trunk/CHANGES.txt Wed Dec 15 17:19:10 2010
@@ -4,6 +4,8 @@ Apache OODT Change Log
Release 0.2 (Current Development)
--------------------------------------------
+* OODT-78 Ability for Resource Manager to dynamically add/remove nodes and
modify queues at run-time (bfoster)
+
* OODT-79 LRUScheduler removes a job from the JobQueue, but adds it back if
can't schedule it, which (in the
JobStack impl) causes JobRepo to create duplicate copies of the same JobSpec
with different JobIds (bfoster)
Modified:
oodt/trunk/resource/src/main/java/org/apache/oodt/cas/resource/monitor/AssignmentMonitor.java
URL:
http://svn.apache.org/viewvc/oodt/trunk/resource/src/main/java/org/apache/oodt/cas/resource/monitor/AssignmentMonitor.java?rev=1049645&r1=1049644&r2=1049645&view=diff
==============================================================================
---
oodt/trunk/resource/src/main/java/org/apache/oodt/cas/resource/monitor/AssignmentMonitor.java
(original)
+++
oodt/trunk/resource/src/main/java/org/apache/oodt/cas/resource/monitor/AssignmentMonitor.java
Wed Dec 15 17:19:10 2010
@@ -19,34 +19,20 @@
package org.apache.oodt.cas.resource.monitor;
//JDK imports
-import java.io.File;
-import java.io.FileFilter;
-import java.io.FileInputStream;
-import java.io.FileNotFoundException;
-import java.net.URI;
-import java.net.URISyntaxException;
import java.net.URL;
-import java.util.Arrays;
import java.util.HashMap;
-import java.util.Iterator;
import java.util.List;
-import java.util.Set;
import java.util.Vector;
-import java.util.logging.Level;
import java.util.logging.Logger;
-import org.w3c.dom.Document;
-import org.w3c.dom.NodeList;
-import org.w3c.dom.Element;
//OODT imports
-import org.apache.oodt.commons.xml.XMLUtils;
-import org.apache.oodt.cas.resource.util.XmlStructFactory;
import org.apache.oodt.cas.resource.structs.ResourceNode;
import org.apache.oodt.cas.resource.structs.exceptions.MonitorException;
/**
*
* @author woollard
+ * @author bfoster
* @version $Revision$
*
* <p>
@@ -58,49 +44,34 @@ import org.apache.oodt.cas.resource.stru
*/
public class AssignmentMonitor implements Monitor {
- private List nodesHomeUris = null;
-
/* our log stream */
private static Logger LOG = Logger.getLogger(AssignmentMonitor.class
.getName());
/* our nodes map */
- private static HashMap nodesMap = new HashMap();
+ private static HashMap<String, ResourceNode> nodesMap;
/* our load map */
- private static HashMap loadMap = new HashMap();
-
- private static FileFilter nodesXmlFilter = new FileFilter() {
- public boolean accept(File pathname) {
- return pathname.isFile()
- && pathname.toString().endsWith("nodes.xml");
- }
- };
+ private static HashMap<String, Integer> loadMap;
- public AssignmentMonitor(List uris) {
- nodesHomeUris = uris;
- nodesMap = loadNodeInfo(nodesHomeUris);
-
- loadMap = new HashMap();
-
- Set nodeIds = nodesMap.keySet();
- Iterator It = nodeIds.iterator();
- while (It.hasNext()) {
- String node = (String) It.next();
- loadMap.put(node, new Integer(0));
+ public AssignmentMonitor(List<ResourceNode> nodes) {
+ nodesMap = new HashMap<String, ResourceNode>();
+ loadMap = new HashMap<String, Integer>();
+
+ for (ResourceNode node : nodes) {
+ nodesMap.put(node.getNodeId(), node);
+ loadMap.put(node.getNodeId(), new Integer(0));
}
}
/*
* (non-Javadoc)
*
- * @see org.apache.oodt.cas.resource.monitor.Monitor#assignLoad(
- * org.apache.oodt.cas.resource.structs.ResourceNode, int)
+ * @see gov.nasa.jpl.oodt.cas.resource.monitor.Monitor#assignLoad(
+ * gov.nasa.jpl.oodt.cas.resource.structs.ResourceNode, int)
*/
public boolean assignLoad(ResourceNode node, int loadValue)
throws MonitorException {
- // ResourceNode resource = (ResourceNode)
- // nodesMap.get(node.getNodeId());
int loadCap = node.getCapacity();
int curLoad = ((Integer) loadMap.get(node.getNodeId())).intValue();
@@ -111,7 +82,6 @@ public class AssignmentMonitor implement
} else {
return false;
}
-
}
public boolean reduceLoad(ResourceNode node, int loadValue)
@@ -128,7 +98,7 @@ public class AssignmentMonitor implement
/*
* (non-Javadoc)
*
- * @see
org.apache.oodt.cas.resource.monitor.Monitor#getLoad(org.apache.oodt.cas.resource.structs.ResourceNode)
+ * @see
gov.nasa.jpl.oodt.cas.resource.monitor.Monitor#getLoad(gov.nasa.jpl.oodt.cas.resource.structs.ResourceNode)
*/
public int getLoad(ResourceNode node) throws MonitorException {
ResourceNode resource = (ResourceNode) nodesMap.get(node.getNodeId());
@@ -139,16 +109,16 @@ public class AssignmentMonitor implement
/*
* (non-Javadoc)
*
- * @see org.apache.oodt.cas.resource.monitor.Monitor#getNodes()
+ * @see gov.nasa.jpl.oodt.cas.resource.monitor.Monitor#getNodes()
*/
- public List getNodes() throws MonitorException {
- return Arrays.asList(nodesMap.values().toArray());
+ public List<ResourceNode> getNodes() throws MonitorException {
+ return new Vector<ResourceNode>(nodesMap.values());
}
/*
* (non-Javadoc)
*
- * @see
org.apache.oodt.cas.resource.monitor.Monitor#getNodeById(java.lang.String)
+ * @see
gov.nasa.jpl.oodt.cas.resource.monitor.Monitor#getNodeById(java.lang.String)
*/
public ResourceNode getNodeById(String nodeId) throws MonitorException {
return (ResourceNode) nodesMap.get(nodeId);
@@ -157,11 +127,11 @@ public class AssignmentMonitor implement
/*
* (non-Javadoc)
*
- * @see
org.apache.oodt.cas.resource.monitor.Monitor#getNodeByURL(java.net.URL)
+ * @see
gov.nasa.jpl.oodt.cas.resource.monitor.Monitor#getNodeByURL(java.net.URL)
*/
public ResourceNode getNodeByURL(URL ipAddr) throws MonitorException {
ResourceNode targetResource = null;
- Vector nodes = (Vector) this.getNodes();
+ List<ResourceNode> nodes = this.getNodes();
for (int i = 0; i < nodes.size(); i++) {
if (((ResourceNode) nodes.get(i)).getIpAddr() == ipAddr) {
targetResource = (ResourceNode) nodes.get(i);
@@ -171,68 +141,15 @@ public class AssignmentMonitor implement
return targetResource;
}
- private HashMap loadNodeInfo(List dirUris) {
-
- HashMap resources = new HashMap();
-
- if (dirUris != null && dirUris.size() > 0) {
- for (Iterator i = dirUris.iterator(); i.hasNext();) {
- String dirUri = (String) i.next();
-
- try {
- File nodesDir = new File(new URI(dirUri));
- if (nodesDir.isDirectory()) {
-
- String nodesDirStr = nodesDir.getAbsolutePath();
-
- if (!nodesDirStr.endsWith("/")) {
- nodesDirStr += "/";
- }
-
- // get all the workflow xml files
- File[] nodesFiles = nodesDir.listFiles(nodesXmlFilter);
-
- for (int j = 0; j < nodesFiles.length; j++) {
-
- String nodesXmlFile = nodesFiles[j]
- .getAbsolutePath();
- Document nodesRoot = null;
- try {
- nodesRoot = XMLUtils
- .getDocumentRoot(new FileInputStream(
- nodesFiles[j]));
- } catch (FileNotFoundException e) {
- e.printStackTrace();
- return null;
- }
-
- NodeList nodeList = nodesRoot
- .getElementsByTagName("node");
-
- if (nodeList != null && nodeList.getLength() > 0) {
- for (int k = 0; k < nodeList.getLength(); k++)
{
- ResourceNode resource = XmlStructFactory
- .getNodes((Element) nodeList
- .item(k));
- resources.put(resource.getNodeId(),
- resource);
- }
- }
- }
- }
- } catch (URISyntaxException e) {
- e.printStackTrace();
- LOG
- .log(
- Level.WARNING,
- "DirUri: "
- + dirUri
- + " is not a directory: skipping
node loading for it.");
- }
- }
- }
+ public void addNode(ResourceNode node) throws MonitorException {
+ nodesMap.put(node.getNodeId(), node);
+ if (!loadMap.containsKey(node.getNodeId()))
+ loadMap.put(node.getNodeId(), new Integer(0));
+ }
- return resources;
+ public void removeNodeById(String nodeId) throws MonitorException {
+ nodesMap.remove(nodeId);
+ loadMap.remove(nodeId);
}
}
Modified:
oodt/trunk/resource/src/main/java/org/apache/oodt/cas/resource/monitor/AssignmentMonitorFactory.java
URL:
http://svn.apache.org/viewvc/oodt/trunk/resource/src/main/java/org/apache/oodt/cas/resource/monitor/AssignmentMonitorFactory.java?rev=1049645&r1=1049644&r2=1049645&view=diff
==============================================================================
---
oodt/trunk/resource/src/main/java/org/apache/oodt/cas/resource/monitor/AssignmentMonitorFactory.java
(original)
+++
oodt/trunk/resource/src/main/java/org/apache/oodt/cas/resource/monitor/AssignmentMonitorFactory.java
Wed Dec 15 17:19:10 2010
@@ -19,14 +19,16 @@
package org.apache.oodt.cas.resource.monitor;
//OODT imports
-import org.apache.oodt.cas.metadata.util.PathUtils;
+import org.apache.oodt.cas.resource.noderepo.XmlNodeRepositoryFactory;
+import org.apache.oodt.cas.resource.util.GenericResourceManagerObjectFactory;
//JDK imports
-import java.util.Arrays;
-import java.util.List;
+import java.util.logging.Level;
+import java.util.logging.Logger;
/**
* @author woollard
+ * @author bfoster
* @version $Revision$
*
* <p>
@@ -36,29 +38,28 @@ import java.util.List;
*/
public class AssignmentMonitorFactory implements MonitorFactory {
- private List nodesDirList;
+ private static final Logger LOG = Logger
+ .getLogger(AssignmentMonitorFactory.class.getName());
- public AssignmentMonitorFactory() {
- String nodesDirUris = System
-
.getProperty("org.apache.oodt.cas.resource.monitor.nodes.dirs");
-
- if (nodesDirUris != null) {
- /* do env var replacement */
- nodesDirUris =
PathUtils.replaceEnvVariables(nodesDirUris);
- String[] dirUris = nodesDirUris.split(",");
- nodesDirList = Arrays.asList(dirUris);
- }
- }
-
- /* (non-Javadoc)
- * @see
org.apache.oodt.cas.resource.monitor.MonitorFactory#createMonitor()
- */
- public Monitor createMonitor() {
- if (nodesDirList != null) {
- return new AssignmentMonitor(nodesDirList);
- } else {
- return null;
- }
- }
+ /*
+ * (non-Javadoc)
+ *
+ * @see
+ * gov.nasa.jpl.oodt.cas.resource.monitor.MonitorFactory#createMonitor()
+ */
+ public AssignmentMonitor createMonitor() {
+ try {
+ String nodeRepoFactoryStr = System.getProperty(
+ "gov.nasa.jpl.oodt.cas.resource.nodes.repo.factory",
+ XmlNodeRepositoryFactory.class.getCanonicalName());
+ return new AssignmentMonitor(GenericResourceManagerObjectFactory
+ .getNodeRepositoryFromFactory(nodeRepoFactoryStr)
+ .loadNodes());
+ } catch (Exception e) {
+ LOG.log(Level.SEVERE, "Failed to create Assignment Monitor : "
+ + e.getMessage(), e);
+ return null;
+ }
+ }
}
Modified:
oodt/trunk/resource/src/main/java/org/apache/oodt/cas/resource/monitor/Monitor.java
URL:
http://svn.apache.org/viewvc/oodt/trunk/resource/src/main/java/org/apache/oodt/cas/resource/monitor/Monitor.java?rev=1049645&r1=1049644&r2=1049645&view=diff
==============================================================================
---
oodt/trunk/resource/src/main/java/org/apache/oodt/cas/resource/monitor/Monitor.java
(original)
+++
oodt/trunk/resource/src/main/java/org/apache/oodt/cas/resource/monitor/Monitor.java
Wed Dec 15 17:19:10 2010
@@ -29,6 +29,7 @@ import org.apache.oodt.cas.resource.stru
/**
*
* @author woollard
+ * @author bfoster
* @version $Revision$
*
* <p>
@@ -104,4 +105,22 @@ public interface Monitor {
*/
public boolean assignLoad(ResourceNode node, int loadValue)
throws MonitorException;
+
+ /**
+ * Adds a new {...@link ResourceNode} for this {...@link Monitor} to
manage (if
+ * node already exist, then should perform update)
+ *
+ * @param node
+ * The new {...@link ResourceNode} to manage
+ */
+ public void addNode(ResourceNode node) throws MonitorException;
+
+ /**
+ * Remove {...@link ResourceNode} from this {...@link Monitor}
+ *
+ * @param nodeId
+ * The id of the {...@link ResourceNode} to remove
+ */
+ public void removeNodeById(String nodeId) throws MonitorException;
+
}
Modified:
oodt/trunk/resource/src/main/java/org/apache/oodt/cas/resource/system/XmlRpcResourceManager.java
URL:
http://svn.apache.org/viewvc/oodt/trunk/resource/src/main/java/org/apache/oodt/cas/resource/system/XmlRpcResourceManager.java?rev=1049645&r1=1049644&r2=1049645&view=diff
==============================================================================
---
oodt/trunk/resource/src/main/java/org/apache/oodt/cas/resource/system/XmlRpcResourceManager.java
(original)
+++
oodt/trunk/resource/src/main/java/org/apache/oodt/cas/resource/system/XmlRpcResourceManager.java
Wed Dec 15 17:19:10 2010
@@ -28,6 +28,7 @@ import org.apache.oodt.cas.resource.stru
import org.apache.oodt.cas.resource.structs.exceptions.JobQueueException;
import org.apache.oodt.cas.resource.structs.exceptions.JobRepositoryException;
import org.apache.oodt.cas.resource.structs.exceptions.MonitorException;
+import org.apache.oodt.cas.resource.structs.exceptions.QueueManagerException;
import org.apache.oodt.cas.resource.structs.exceptions.SchedulerException;
import org.apache.oodt.cas.resource.util.GenericResourceManagerObjectFactory;
import org.apache.oodt.cas.resource.util.XmlRpcStructFactory;
@@ -271,6 +272,48 @@ public class XmlRpcResourceManager {
return execNode;
}
+ public List<String> getQueues() throws QueueManagerException {
+ return new Vector<String>(this.scheduler.getQueueManager().getQueues());
+ }
+
+ public boolean addQueue(String queueName) throws QueueManagerException {
+ this.scheduler.getQueueManager().addQueue(queueName);
+ return true;
+ }
+
+ public boolean removeQueue(String queueName) throws QueueManagerException {
+ this.scheduler.getQueueManager().removeQueue(queueName);
+ return true;
+ }
+
+ public boolean addNode(Hashtable hashNode) throws MonitorException {
+
this.scheduler.getMonitor().addNode(XmlRpcStructFactory.getResourceNodeFromXmlRpc(hashNode));
+ return true;
+ }
+
+ public boolean removeNode(String nodeId) throws MonitorException {
+ this.scheduler.getMonitor().removeNodeById(nodeId);
+ return true;
+ }
+
+ public boolean addNodeToQueue(String nodeId, String queueName) throws
QueueManagerException {
+ this.scheduler.getQueueManager().addNodeToQueue(nodeId, queueName);
+ return true;
+ }
+
+ public boolean removeNodeFromQueue(String nodeId, String queueName) throws
QueueManagerException {
+ this.scheduler.getQueueManager().removeNodeFromQueue(nodeId, queueName);
+ return true;
+ }
+
+ public List<String> getNodesInQueue(String queueName) throws
QueueManagerException {
+ return new
Vector<String>(this.scheduler.getQueueManager().getNodes(queueName));
+ }
+
+ public List<String> getQueuesWithNode(String nodeId) throws
QueueManagerException {
+ return new
Vector<String>(this.scheduler.getQueueManager().getQueues(nodeId));
+ }
+
public static void main(String[] args) throws Exception {
int portNum = -1;
String usage = "XmlRpcResourceManager --portNum <port number for xml
rpc service>\n";
Modified:
oodt/trunk/resource/src/main/java/org/apache/oodt/cas/resource/system/XmlRpcResourceManagerClient.java
URL:
http://svn.apache.org/viewvc/oodt/trunk/resource/src/main/java/org/apache/oodt/cas/resource/system/XmlRpcResourceManagerClient.java?rev=1049645&r1=1049644&r2=1049645&view=diff
==============================================================================
---
oodt/trunk/resource/src/main/java/org/apache/oodt/cas/resource/system/XmlRpcResourceManagerClient.java
(original)
+++
oodt/trunk/resource/src/main/java/org/apache/oodt/cas/resource/system/XmlRpcResourceManagerClient.java
Wed Dec 15 17:19:10 2010
@@ -31,6 +31,7 @@ import org.apache.oodt.cas.resource.stru
import org.apache.oodt.cas.resource.structs.exceptions.JobExecutionException;
import org.apache.oodt.cas.resource.structs.exceptions.JobRepositoryException;
import org.apache.oodt.cas.resource.structs.exceptions.MonitorException;
+import org.apache.oodt.cas.resource.structs.exceptions.QueueManagerException;
import org.apache.oodt.cas.resource.util.XmlRpcStructFactory;
//JDK imports
@@ -116,6 +117,15 @@ public class XmlRpcResourceManagerClient
String getNodeByIdOperation = "--getNodeById --nodeId <node id>\n";
String getNodesOperation = "--getNodes\n";
+ String getQueuesOperation = "--getQueues\n";
+ String addNodeOperation = "--addNode --nodeId <node id> --ipAddr <url>
--capacity <max load>\n";
+ String removeNodeOperation = "--removeNode --nodeId <node id>\n";
+ String addQueueOperation = "--addQueue --queueName <queue name>\n";
+ String removeQueueOperation = "--removeQueue --queueName <queue
name>\n";
+ String addNodeToQueueOperation = "--addNodeToQueue --nodeId <node id>
--queueName <queue name>\n";
+ String getNodesInQueueOperation = "--getNodesInQueue --queueName
<queue name>\n";
+ String getQueuesWithNodeOperation = "--getQueuesWithNode --nodeId
<node id>\n";
+ String removeNodeFromQueueOperation = "--removeNodeFromQueue --nodeId
<node id> --queueName <queue name>\n";
String submitJobOperation = "--submitJob --def <job def file> --input
<job input constructor>\n";
String submitJobRemoteOperation = "--submitJob --def <job def file>
--input <job input constructor> --url <url>\n";
String getJobInfoOperation = "--getJobInfo --id <job id>\n";
@@ -164,6 +174,167 @@ public class XmlRpcResourceManagerClient
}
}
+ }else if (operation.equals("--getQueues")) {
+ List<String> queueNames = client.getQueues();
+ System.out.println("Queues:");
+ for (String queueName : queueNames)
+ System.out.println(" - " + queueName);
+ System.out.println();
+
+ }else if (operation.equals("--addNode")) {
+ String nodeId = null;
+ String nodeUrl = null;
+ String capacity = null;
+
+ for (int i = 4; i < args.length; i++) {
+ if (args[i].equals("--nodeId")) {
+ nodeId = args[++i];
+ }else if (args[i].equals("--ipAddr")) {
+ nodeUrl = args[++i];
+ }else if (args[i].equals("--capacity")) {
+ capacity = args[++i];
+ }
+ }
+
+ if (nodeId == null || nodeUrl == null || capacity == null) {
+ System.err.println(addNodeOperation);
+ System.exit(1);
+ }
+
+ client.addNode(new ResourceNode(nodeId, new URL(nodeUrl),
Integer.parseInt(capacity)));
+ System.out.println("Successfully added node!");
+
+ }else if (operation.equals("--removeNode")) {
+ String nodeId = null;
+
+ for (int i = 4; i < args.length; i++) {
+ if (args[i].equals("--nodeId")) {
+ nodeId = args[++i];
+ }
+ }
+
+ if (nodeId == null) {
+ System.err.println(removeNodeOperation);
+ System.exit(1);
+ }
+
+ client.removeNode(nodeId);
+ System.out.println("Successfully removed node!");
+
+ }else if (operation.equals("--addQueue")) {
+ String queueName = null;
+
+ for (int i = 4; i < args.length; i++) {
+ if (args[i].equals("--queueName")) {
+ queueName = args[++i];
+ }
+ }
+
+ if (queueName == null) {
+ System.err.println(addQueueOperation);
+ System.exit(1);
+ }
+
+ client.addQueue(queueName);
+ System.out.println("Successfully added queue!");
+
+ }else if (operation.equals("--removeQueue")) {
+ String queueName = null;
+
+ for (int i = 4; i < args.length; i++) {
+ if (args[i].equals("--queueName")) {
+ queueName = args[++i];
+ }
+ }
+
+ if (queueName == null) {
+ System.err.println(removeQueueOperation);
+ System.exit(1);
+ }
+
+ client.removeQueue(queueName);
+ System.out.println("Successfully removed queue!");
+
+ }else if (operation.equals("--addNodeToQueue")) {
+ String nodeId = null;
+ String queueName = null;
+
+ for (int i = 4; i < args.length; i++) {
+ if (args[i].equals("--nodeId")) {
+ nodeId = args[++i];
+ }else if (args[i].equals("--queueName")) {
+ queueName = args[++i];
+ }
+ }
+
+ if (nodeId == null || queueName == null) {
+ System.err.println(addNodeToQueueOperation);
+ System.exit(1);
+ }
+
+ client.addNodeToQueue(nodeId, queueName);
+ System.out.println("Successfully added node to queue!");
+
+ }else if (operation.equals("--removeNodeFromQueue")) {
+ String nodeId = null;
+ String queueName = null;
+
+ for (int i = 4; i < args.length; i++) {
+ if (args[i].equals("--nodeId")) {
+ nodeId = args[++i];
+ }else if (args[i].equals("--queueName")) {
+ queueName = args[++i];
+ }
+ }
+
+ if (nodeId == null || queueName == null) {
+ System.err.println(removeNodeFromQueueOperation);
+ System.exit(1);
+ }
+
+ client.removeNodeFromQueue(nodeId, queueName);
+ System.out.println("Successfully removed node from queue!");
+
+ }else if (operation.equals("--getNodesInQueue")) {
+ String queueName = null;
+
+ for (int i = 4; i < args.length; i++) {
+ if (args[i].equals("--queueName")) {
+ queueName = args[++i];
+ }
+ }
+
+ if (queueName == null) {
+ System.err.println(getNodesInQueueOperation);
+ System.exit(1);
+ }
+
+ List<String> nodeIds = client.getNodesInQueue(queueName);
+ System.out.println("Nodes in Queue '" + queueName + "':");
+ for (String nodeId : nodeIds)
+ System.out.println(" - " + nodeId);
+ System.out.println();
+
+ }else if (operation.equals("--getQueuesWithNode")) {
+ String nodeId = null;
+
+ for (int i = 4; i < args.length; i++) {
+ if (args[i].equals("--nodeId")) {
+ nodeId = args[++i];
+ }
+ }
+
+ if (nodeId == null) {
+ System.err.println(getQueuesWithNodeOperation);
+ System.exit(1);
+ }
+
+ List<String> queueNames = client.getQueuesWithNode(nodeId);
+ System.out.println("Queues with node '" + nodeId + "':");
+ for (String queueName : queueNames)
+ System.out.println(" - " + queueName);
+ System.out.println();
+
} else if (operation.equals("--getExecNode")) {
String jobId = null;
@@ -454,6 +625,146 @@ public class XmlRpcResourceManagerClient
this.resMgrUrl = resMgrUrl;
}
+ /**
+ * Creates a queue with the given name
+ * @param queueName The name of the queue to be created
+ * @throws QueueManagerException on any error
+ */
+ public void addQueue(String queueName) throws QueueManagerException {
+ try {
+ Vector<Object> argList = new Vector<Object>();
+ argList.add(queueName);
+ client.execute("resourcemgr.addQueue", argList);
+ }catch (Exception e) {
+ throw new QueueManagerException(e.getMessage(), e);
+ }
+ }
+
+ /**
+ * Removes the queue with the given name
+ * @param queueName The name of the queue to be removed
+ * @throws QueueManagerException on any error
+ */
+ public void removeQueue(String queueName) throws QueueManagerException {
+ try {
+ Vector<Object> argList = new Vector<Object>();
+ argList.add(queueName);
+ client.execute("resourcemgr.removeQueue", argList);
+ }catch (Exception e) {
+ throw new QueueManagerException(e.getMessage(), e);
+ }
+ }
+
+ /**
+ * Adds a node
+ * @param node The node to be added
+ * @throws MonitorException on any error
+ */
+ public void addNode(ResourceNode node) throws MonitorException {
+ try {
+ Vector<Object> argList = new Vector<Object>();
+ argList.add(XmlRpcStructFactory.getXmlRpcResourceNode(node));
+ client.execute("resourcemgr.addNode", argList);
+ }catch (Exception e) {
+ throw new MonitorException(e.getMessage(), e);
+ }
+ }
+
+ /**
+ * Removes the node with the given id
+ * @param nodeId The id of the node to be removed
+ * @throws MonitorException on any error
+ */
+ public void removeNode(String nodeId) throws MonitorException {
+ try {
+ Vector<Object> argList = new Vector<Object>();
+ argList.add(nodeId);
+ client.execute("resourcemgr.removeNode", argList);
+ }catch (Exception e) {
+ throw new MonitorException(e.getMessage(), e);
+ }
+ }
+
+ /**
+ * Addes the node with given id to the queue with the given name
+ * @param nodeId The id of the node to be added to the given queueName
+ * @param queueName The name of the queue to add the given node
+ * @throws QueueManagerException on any error
+ */
+ public void addNodeToQueue(String nodeId, String queueName) throws
QueueManagerException {
+ try {
+ Vector<Object> argList = new Vector<Object>();
+ argList.add(nodeId);
+ argList.add(queueName);
+ client.execute("resourcemgr.addNodeToQueue", argList);
+ }catch (Exception e) {
+ throw new QueueManagerException(e.getMessage(), e);
+ }
+ }
+
+ /**
+ * Remove the node with the given id from the queue with the given name
+ * @param nodeId The id of the node to be remove from the given queueName
+ * @param queueName The name of the queue from which to remove the given
node
+ * @throws QueueManagerException on any error
+ */
+ public void removeNodeFromQueue(String nodeId, String queueName) throws
QueueManagerException {
+ try {
+ Vector<Object> argList = new Vector<Object>();
+ argList.add(nodeId);
+ argList.add(queueName);
+ client.execute("resourcemgr.removeNodeFromQueue", argList);
+ }catch (Exception e) {
+ throw new QueueManagerException(e.getMessage(), e);
+ }
+ }
+
+ /**
+ * Gets a list of currently supported queue names
+ * @return A list of currently supported queue names
+ * @throws QueueManagerException on any error
+ */
+ public List<String> getQueues() throws QueueManagerException {
+ try {
+ Vector<Object> argList = new Vector<Object>();
+ return (List<String>) client.execute("resourcemgr.getQueues",
argList);
+ }catch (Exception e) {
+ throw new QueueManagerException(e.getMessage(), e);
+ }
+ }
+
+ /**
+ * Gets a list of ids of the nodes in the given queue
+ * @param queueName The name of the queue to get node ids from
+ * @return List of node ids in the given queueName
+ * @throws QueueManagerException on any error
+ */
+ public List<String> getNodesInQueue(String queueName) throws
QueueManagerException {
+ try {
+ Vector<Object> argList = new Vector<Object>();
+ argList.add(queueName);
+ return (List<String>)
client.execute("resourcemgr.getNodesInQueue", argList);
+ }catch (Exception e) {
+ throw new QueueManagerException(e.getMessage(), e);
+ }
+ }
+
+ /**
+ * Gets a list of queues which contain the node with the given nodeId
+ * @param nodeId The id of the node to get queues it belongs to
+ * @return List of queues which contain the give node
+ * @throws QueueManagerException on any error
+ */
+ public List<String> getQueuesWithNode(String nodeId) throws
QueueManagerException {
+ try {
+ Vector<Object> argList = new Vector<Object>();
+ argList.add(nodeId);
+ return (List<String>)
client.execute("resourcemgr.getQueuesWithNode", argList);
+ }catch (Exception e) {
+ throw new QueueManagerException(e.getMessage(), e);
+ }
+ }
+
private static String getReadableJobStatus(String status) {
if (status.equals(JobStatus.COMPLETE)) {
return "COMPLETE";
Modified:
oodt/trunk/resource/src/main/java/org/apache/oodt/cas/resource/util/GenericResourceManagerObjectFactory.java
URL:
http://svn.apache.org/viewvc/oodt/trunk/resource/src/main/java/org/apache/oodt/cas/resource/util/GenericResourceManagerObjectFactory.java?rev=1049645&r1=1049644&r2=1049645&view=diff
==============================================================================
---
oodt/trunk/resource/src/main/java/org/apache/oodt/cas/resource/util/GenericResourceManagerObjectFactory.java
(original)
+++
oodt/trunk/resource/src/main/java/org/apache/oodt/cas/resource/util/GenericResourceManagerObjectFactory.java
Wed Dec 15 17:19:10 2010
@@ -31,6 +31,8 @@ import org.apache.oodt.cas.resource.jobr
import org.apache.oodt.cas.resource.jobrepo.JobRepositoryFactory;
import org.apache.oodt.cas.resource.monitor.Monitor;
import org.apache.oodt.cas.resource.monitor.MonitorFactory;
+import org.apache.oodt.cas.resource.noderepo.NodeRepository;
+import org.apache.oodt.cas.resource.noderepo.NodeRepositoryFactory;
import org.apache.oodt.cas.resource.queuerepo.QueueRepository;
import org.apache.oodt.cas.resource.queuerepo.QueueRepositoryFactory;
import org.apache.oodt.cas.resource.scheduler.Scheduler;
@@ -160,6 +162,43 @@ public final class GenericResourceManage
}
/**
+ * Creates a new {...@link NodeRepository} implementation from the given
+ * {...@link QueueRepositoryFactory} class name.
+ *
+ * @param serviceFactory
+ * The class name of the {...@link NodeRepositoryFactory} to use to
create new
+ * {...@link QueueRepository}s.
+ * @return A new implementation of a {...@link NodeRepository}.
+ */
+ public static NodeRepository getNodeRepositoryFromFactory(String
nodeRepositoryFactory) {
+ Class clazz = null;
+ NodeRepositoryFactory factory = null;
+
+ try {
+ clazz = Class.forName(nodeRepositoryFactory);
+ factory = (NodeRepositoryFactory) clazz.newInstance();
+ return factory.createNodeRepository();
+ } catch (ClassNotFoundException e) {
+ e.printStackTrace();
+ LOG.log(Level.WARNING,
+ "ClassNotFoundException when loading node repository factory class "
+ + nodeRepositoryFactory + " Message: " + e.getMessage());
+ } catch (InstantiationException e) {
+ e.printStackTrace();
+ LOG.log(Level.WARNING,
+ "InstantiationException when loading node repository factory class "
+ + nodeRepositoryFactory + " Message: " + e.getMessage());
+ } catch (IllegalAccessException e) {
+ e.printStackTrace();
+ LOG.log(Level.WARNING,
+ "IllegalAccessException when loading node repository factory class "
+ + nodeRepositoryFactory + " Message: " + e.getMessage());
+ }
+
+ return null;
+ }
+
+ /**
* Creates a new {...@link JobQueue} implementation from the given
* {...@link JobQueueFactory} class name.
*
Modified:
oodt/trunk/resource/src/test/org/apache/oodt/cas/resource/monitor/TestAssignmentMonitor.java
URL:
http://svn.apache.org/viewvc/oodt/trunk/resource/src/test/org/apache/oodt/cas/resource/monitor/TestAssignmentMonitor.java?rev=1049645&r1=1049644&r2=1049645&view=diff
==============================================================================
---
oodt/trunk/resource/src/test/org/apache/oodt/cas/resource/monitor/TestAssignmentMonitor.java
(original)
+++
oodt/trunk/resource/src/test/org/apache/oodt/cas/resource/monitor/TestAssignmentMonitor.java
Wed Dec 15 17:19:10 2010
@@ -24,6 +24,8 @@ import org.apache.oodt.cas.resource.stru
//JDK imports
import java.io.File;
+import java.net.MalformedURLException;
+import java.net.URL;
import java.util.Iterator;
import java.util.List;
import java.util.Vector;
@@ -33,6 +35,7 @@ import junit.framework.TestCase;
/**
* @author mattmann
+ * @author bfoster
* @version $Revision$
*
* <p>
@@ -44,9 +47,11 @@ public class TestAssignmentMonitor exten
private AssignmentMonitor assgnMon = null;
protected void setUp() {
- List uris = new Vector();
- uris.add(new File("./src/main/resources/examples").toURI().toString());
- assgnMon = new AssignmentMonitor(uris);
+ System.setProperty("gov.nasa.jpl.oodt.cas.resource.nodes.dirs",
+ "file:"
+ + new File("./src/main/resources/examples")
+ .getAbsolutePath());
+ assgnMon = new AssignmentMonitorFactory().createMonitor();
}
public void testGetNodes() {
@@ -102,4 +107,31 @@ public class TestAssignmentMonitor exten
assertTrue(hasNode1);
}
+ public void testNodeModification() throws MonitorException,
+ MalformedURLException {
+ List<ResourceNode> nodes = new Vector<ResourceNode>(this.assgnMon
+ .getNodes());
+ ResourceNode test1 = new ResourceNode("Test1", new URL(
+ "http://localhost:1111";), 9);
+ ResourceNode test2 = new ResourceNode("Test2", new URL(
+ "http://localhost:2222";), 9);
+ ResourceNode test3 = new ResourceNode("Test3", new URL(
+ "http://localhost:3333";), 9);
+ this.assgnMon.addNode(test1);
+ nodes.add(test1);
+ this.assgnMon.addNode(test2);
+ nodes.add(test2);
+ this.assgnMon.addNode(test3);
+ nodes.add(test3);
+
+ assertTrue(nodes.containsAll(this.assgnMon.getNodes())
+ && this.assgnMon.getNodes().containsAll(nodes));
+
+ this.assgnMon.removeNodeById(test1.getNodeId());
+ nodes.remove(test1);
+
+ assertTrue(nodes.containsAll(this.assgnMon.getNodes())
+ && this.assgnMon.getNodes().containsAll(nodes));
+ }
+
}
