Changeset 5446

Timestamp:

Oct 15, 2010, 1:59:00 PM (12 years ago)

Author:

Nicklas Nordborg

Message:

References #1527: Improve remote control in job agents

This should provide everything we need. The real test is implementing #1528 so I'll keep this open until it is known to work.

Location:

trunk

Files:

• config/dist/jobagent.properties (modified) (1 diff)
• doc/src/docbook/appendix/jobagent.properties.xml (modified) (1 diff)
• src/clients/jobagent/net/sf/basedb/clients/jobagent/Agent.java (modified) (12 diffs)
• src/clients/jobagent/net/sf/basedb/clients/jobagent/AgentController.java (modified) (2 diffs)
• src/clients/jobagent/net/sf/basedb/clients/jobagent/handlers/AbstractCustomRequestHandler.java (added)
• src/clients/jobagent/net/sf/basedb/clients/jobagent/handlers/CustomRequestHandler.java (added)
• src/clients/jobagent/net/sf/basedb/clients/jobagent/handlers/DefaultRequestHandler.java (modified) (4 diffs)
• src/clients/jobagent/net/sf/basedb/clients/jobagent/handlers/MultiProtocolRequestHandler.java (added)
• src/core/net/sf/basedb/core/signal/SocketSignalTransporter.java (modified) (1 diff)
• src/core/net/sf/basedb/util/jobagent/JobAgentConnection.java (modified) (1 diff)
• src/test/TestJobAgent.java (modified) (7 diffs)

trunk/config/dist/jobagent.properties

@@ -54 +54 @@
 agent.allowremote.start=true
 agent.allowremote.pause=true
+
+# ==============================
+# Custom remote control handlers
+# ==============================
+
+## Handlers for custom remote control can be registered with one or more
+## settings like the one below. Replace 'foo' with the actual name of the 
+## protocol. The class must implement the CustumRequestHandler
+## interface. Requests are sent to the handler on the agent's remote control
+## port: foo://more-custom-data .....
+# agent.request-handler.foo = <class-name>
 
 

trunk/doc/src/docbook/appendix/jobagent.properties.xml

@@ -187 +187 @@
     </variablelist>
 
+  </simplesect>
+
+  <simplesect id="appendix.jobagent.properties.request">
+    <title>Custom request handlers</title>
+
+    <variablelist>
+    <varlistentry>
+      <term><property>agent.request-handler.*</property></term>
+      <listitem>
+        <para>
+          Optional. One or more entries for custom remote control handlers.
+          The * should be replaced with the name of the protocol and the
+          value should be the name of a class implementing the
+          <interfacename docapi="net.sf.basedb.clients.jobagent.handlers">CustomRequestHandler</interfacename>
+          interface. Requests can then be sent to the agent's remote control port on
+          the form: <code>foo://custom-data....</code>.
+        </para>
+      </listitem>
+    </varlistentry>
+    </variablelist>
+    
   </simplesect>
 

trunk/src/clients/jobagent/net/sf/basedb/clients/jobagent/Agent.java

@@ -34 +34 @@
 
 import net.sf.basedb.clients.jobagent.executors.ProcessJobExecutor;
+import net.sf.basedb.clients.jobagent.handlers.CustomRequestHandler;
 import net.sf.basedb.clients.jobagent.handlers.DefaultRequestHandler;
+import net.sf.basedb.clients.jobagent.handlers.MultiProtocolRequestHandler;
+import net.sf.basedb.clients.jobagent.handlers.SignalRequestHandler;
 import net.sf.basedb.core.Application;
 import net.sf.basedb.core.DbControl;
@@ -133 +136 @@
   </tr>
   <tr>
+    <td>agent.request-handler.*</td>
+    <td>-</td>
+    <td>
+      One or more entries that can be used to register custom remote control request handlers.
+      The * should be replaced with the name of the protocol which means that requests
+      should take the form: protocol://custom-data....
+      <p>
+      The value is the name of a class that implements the {@link CustomRequestHandler}
+      interface. The implementation must provide a public no-argument constructor. The
+      implementation must be thread-safe and be able to handle multiple requests at the
+      time.
+    </td>
+  </tr>
+  <tr>
     <td>agent.executor.class</td>
     <td>{@link ProcessJobExecutor}</td>
@@ -236 +253 @@
   private final boolean allowRemotePause;
   private final boolean allowRemoteStart;
+  private final Map<String, Class<? extends CustomRequestHandler>> customRequestHandlerClasses;
   
   // Job execution settings
@@ -243 +261 @@
   private InetAddress serverAddress;
   private JobAgentServerConnection server;
-  private RequestHandler requestHandler;
+  private MultiProtocolRequestHandler requestHandler;
   private AgentSignalReceiver signalReceiver;
   private JobExecutor jobExecutor;
@@ -306 +324 @@
       }
     }
-    
+
     // BASE settings
     this.login = properties.getProperty("agent.user");
@@ -320 +338 @@
     this.allowRemotePause = Values.getBoolean(properties.getProperty("agent.allowremote.pause"), true);
     this.allowRemoteStart = Values.getBoolean(properties.getProperty("agent.allowremote.start"), true);
+    
+    // Additional request handlers
+    this.customRequestHandlerClasses = getCustomRequestHandlerClasses();
     
     // Job execution settings
@@ -464 +485 @@
   }
   
+  @SuppressWarnings("unchecked")
+  private Map<String, Class<? extends CustomRequestHandler>> getCustomRequestHandlerClasses()
+  {
+    Map<String, Class<? extends CustomRequestHandler>> handlers = new HashMap<String, Class<? extends CustomRequestHandler>>();
+    for (String property : properties.stringPropertyNames())
+    {
+      if (property.startsWith("agent.request-handler."))
+      {
+        String protocol = property.substring("agent.request-handler.".length());
+        String className = properties.getProperty(property);
+        try
+        {
+          Class<? extends CustomRequestHandler> clazz = (Class<? extends CustomRequestHandler>)Class.forName(className);
+          if (!CustomRequestHandler.class.isAssignableFrom(clazz))
+          {
+            log.warn("Class '" + className + "' doesn't implement the CustomRequestHandler interface, ignored!");
+          }
+          else
+          {
+            log.info("Loaded custom request handler for protocol '" + protocol + "': " + clazz.getName());
+            handlers.put(protocol, clazz);
+          }
+        }
+        catch (Throwable t)
+        {
+          log.warn("Class " + className + " not found, ignored!");
+        }
+      }
+    }
+    return handlers;
+  }
+  
+  private void registerCustomRequestHandlers(MultiProtocolRequestHandler master)
+  {
+    Map<Class<? extends CustomRequestHandler>, CustomRequestHandler> created = 
+      new HashMap<Class<? extends CustomRequestHandler>, CustomRequestHandler>();
+    for (Map.Entry<String, Class<? extends CustomRequestHandler>> entry : customRequestHandlerClasses.entrySet())
+    {
+      String protocol = entry.getKey();
+      Class<? extends CustomRequestHandler> clazz = entry.getValue();
+      CustomRequestHandler handler = created.get(clazz);
+      if (handler == null)
+      {
+        try
+        {
+          handler = clazz.newInstance();
+          handler.init(this);
+          created.put(clazz, handler);
+          log.info("Created request handler for protocol '" + protocol + ": " + clazz.getName());
+        }
+        catch (Throwable t)
+        {
+          log.error("Could not create request handler for protocol '" + protocol + ": " + clazz.getName(), t);
+        }
+      }
+      master.registerProtocols(handler, protocol);
+    }
+  }
+  
   /**
     Get the class object for the configured job executor. If the 
@@ -520 +600 @@
     Note! The listener service is started in a separate thread and this method
     returns as soon as the network connections are set up.
-    
-    @param requestHandler A {@link RequestHandler} that handles the 
+    <p>
+    Note! The default handler supplied as an argument is used as a fallback 
+    handler for unregistered protocols. Additional request handlers can be
+    set up by calling {@link #registerRequestHandler(RequestHandler, String...)}.
+    
+    @param defaultHandler A {@link RequestHandler} that handles the 
       incoming requsts, or null to use the {@link DefaultRequestHandler}
     @throws IOException If there is an error when starting the service
   */
-  public synchronized void service(RequestHandler requestHandler)
+  public synchronized void service(RequestHandler defaultHandler)
     throws IOException
   {
@@ -533 +617 @@
       this.signalReceiver = new AgentSignalReceiver(this);
       this.signalReceiver.init(null);
-      this.requestHandler = requestHandler == null ? new DefaultRequestHandler(this) : requestHandler;
+      if (defaultHandler == null) defaultHandler = new DefaultRequestHandler(this);
+      this.requestHandler = new MultiProtocolRequestHandler(this, defaultHandler);
+      registerRequestHandler(new SignalRequestHandler(this), "signal");
+      registerCustomRequestHandlers(requestHandler);
       this.server = new JobAgentServerConnection(port, this.requestHandler, logServer);
       server.open();
@@ -687 +774 @@
     }
     return allow;
+  }
+  
+  /**
+    Register a request handler for one or more protocols. See
+    {@link MultiProtocolRequestHandler} for more information.
+    Calls to this method is ignored if the job agent has no listener
+    service. See {@link #service(RequestHandler)}.
+    
+    @param handler The request handler
+    @param protocols The name of the protocols the handler should handle
+    @since 2.16
+  */
+  public void registerRequestHandler(RequestHandler handler, String... protocols)
+  {
+    if (this.requestHandler == null) return;
+    if (handler == null) throw new NullPointerException("handler");
+    requestHandler.registerProtocols(handler, protocols);
+  }
+
+  /**
+    Unregister one or more protocols. See
+    {@link MultiProtocolRequestHandler} for more information.
+    Calls to this method is ignored if the job agent has no listener
+    service. See {@link #service(RequestHandler)}.
+    
+    @param protocols The name of the protocols the handler should handle
+    @since 2.16
+  */
+  public void unregisterRequestHandler(String... protocols)
+  {
+    if (this.requestHandler == null) return;
+    requestHandler.unregisterProtocols(protocols);
   }
   
@@ -858 +977 @@
       server = null;
     }
+    if (requestHandler != null)
+    {
+      requestHandler.close();
+      requestHandler = null;
+    }
   }
   
@@ -937 +1061 @@
     // Send SHUTDOWN/ABORT to all jobs, that support signals
     signalReceiver.close(closeTimeout);
+    signalReceiver = null;
+    unregisterRequestHandler("signal");
   }
   

trunk/src/clients/jobagent/net/sf/basedb/clients/jobagent/AgentController.java

@@ -29 +29 @@
 import java.util.Properties;
 
+import net.sf.basedb.clients.jobagent.handlers.DefaultRequestHandler;
 import net.sf.basedb.core.DbControl;
 import net.sf.basedb.core.ItemNotFoundException;
@@ -238 +239 @@
         log.info("Creating a new job agent on port " + port);
         Agent agent = createAgent(properties);
-        agent.service(null); // Start listening for incoming connections
+        agent.service(new DefaultRequestHandler(agent)); // Start listening for incoming connections
       }
       else

trunk/src/clients/jobagent/net/sf/basedb/clients/jobagent/handlers/DefaultRequestHandler.java

@@ -67 +67 @@
     registerHandler(new StopRequestHandler(agent), "stop");
     registerHandler(new PauseRequestHandler(agent), "pause");
-    if (agent.getSignalReceiver() != null)
-    {
-      registerHandler(new SignalRequestHandler(agent), "signal");
-    }
   }
   
@@ -80 +76 @@
     }
   }
-  
+  /*
+    From the RequestHandler interface
+    ---------------------------------
+  */
+  @Override
   public String handleCmd(Socket incoming, String cmd)
   {
@@ -91 +91 @@
     String answer = null;
     RequestHandler handler = commandHandlers.get(cmd);
-    if (handler == null && cmd != null && cmd.startsWith("signal://"))
-    {
-      handler = commandHandlers.get("signal");
-    }
-    if (!agent.isAllowedControl(remote, cmd))
-    {
-      answer = "FAILED Permission denied: cmd=" + cmd + "; host=" + remote.toString();
-    }
-    else if (handler == null)
+    if (handler == null)
     {
       answer = "FAILED Unknown command: " + cmd;
@@ -113 +105 @@
     return answer;
   }
+  // --------------------------------------------------
 }

trunk/src/core/net/sf/basedb/core/signal/SocketSignalTransporter.java

@@ -71 +71 @@
       s = new Socket(host, port);
       SocketUtil.send(s, message, true);
+      SocketUtil.read(s, true); // Consume any response
     }
     catch (Exception ex)

trunk/src/core/net/sf/basedb/util/jobagent/JobAgentConnection.java

@@ -133 +133 @@
   
   /**
-    Convenience method for opening a new socket, sending a command and
-    return the answer.
+    Send a remote control command to the job agent.
     
     @param cmd The command to send
     @return The answer
     @throws IOException If there is an error
-   */
-  private String send(String cmd)
+    @since 2.16 (was private before that)
+  */
+  public String send(String cmd)
     throws IOException
   {

trunk/src/test/TestJobAgent.java

@@ -28 +28 @@
 
 import net.sf.basedb.clients.jobagent.Agent;
+import net.sf.basedb.clients.jobagent.handlers.AbstractCustomRequestHandler;
 import net.sf.basedb.clients.jobagent.handlers.DefaultRequestHandler;
 
@@ -84 +85 @@
     test_list_jobs(id, 1);
     
+    // Extra test: send custom control command
+    test_send_control_command(id, "foo://hello");
+    test_send_control_command(id2, "foo://world");
+    
     if (TestUtil.waitBeforeDelete()) TestUtil.waitForEnter();
     TestJob.test_delete(jobId);
@@ -415 +420 @@
   }
   
-  static void create_fake_jobagent(final int port, String externalId)
-  {
-    try
-    {
-      // The request handler for incoming requests
+  static void create_fake_jobagent(final int port, final String externalId)
+  {
+    try
+    {
       Properties p = new Properties();
       p.setProperty("agent.port", Integer.toString(port));
@@ -425 +429 @@
       p.setProperty("agent.password", TestUtil.getPassword());
       p.setProperty("agent.id", externalId);
+      p.setProperty("agent.request-handler.foo", FooRequestHandler.class.getName());
       Agent agent = new Agent(p);
       RequestHandler requestHandler = new DefaultRequestHandler(agent)
@@ -430 +435 @@
         public String handleCmd(Socket incoming, String cmd)
         {
-          if (!TestUtil.getSilent()) write("--Job agent received cmd: " + cmd + " (port="+port+")");
+          if (!TestUtil.getSilent()) write("--Job agent received cmd: " + cmd + " (id=" + externalId + "; port="+port+")");
           String answer = super.handleCmd(incoming, cmd);
           if (!TestUtil.getSilent()) write("--Job agent answers: " + answer);
@@ -437 +442 @@
       };
       agent.service(requestHandler); // NOTE! Creates new thread for the listener
-      write("--Create fake job agent OK (port " + port + ")");
+      write("--Create fake job agent OK (id=" + externalId + "; port " + port + ")");
     }
     catch (Throwable t)
     {
-      write("--Create fake job agent FAILED (port " + port + ")");
+      write("--Create fake job agent FAILED (id=" + externalId + "; port " + port + ")");
       t.printStackTrace();
       ok = false;
@@ -468 +473 @@
     }
   }
+  
+  static void test_send_control_command(int jobAgentId, String cmd)
+  {
+    if (jobAgentId == 0) return;
+    DbControl dc = null;
+    try
+    {
+      dc = TestUtil.getDbControl();
+      JobAgent j = JobAgent.getById(dc, jobAgentId);
+      JobAgentConnection conn = j.getConnection(null);
+      String answer = conn.send(cmd);
+      write("--Send custom command OK (" + cmd + " --> " + answer + ")");
+    }
+    catch (Throwable ex)
+    {
+      write("--Send custom command FAILED (" + cmd + ")");
+      ex.printStackTrace();
+      ok = false;
+    }
+    finally
+    {
+      if (dc != null) dc.close();
+    }
+  }
+
+  
+  public static class FooRequestHandler
+    extends AbstractCustomRequestHandler
+  {
+  
+    public FooRequestHandler()
+    {}
+
+    @Override
+    public void init(Agent agent)
+    {
+      super.init(agent);
+      if (!TestUtil.getSilent()) 
+      {
+        write("--Initializing 'foo' handler for job agent (id=" + agent.getId() + "; port=" + agent.getPort() + ")");
+      }
+    }
+    @Override
+    public void close()
+    {
+      Agent agent = getAgent();
+      int port = agent.getPort();
+      String id = agent.getId();
+      if (!TestUtil.getSilent()) 
+      {
+        write("--Closing 'foo' handler for job agent (id=" + id + "; port=" + port + ")");
+      }
+      super.close();
+    }
+    
+    @Override
+    public String handleCmd(Socket socket, String cmd)
+    {
+      Agent agent = getAgent();
+      int port = agent.getPort();
+      String id = agent.getId();
+      if (!TestUtil.getSilent())
+      {
+        write("--Job agent received cmd: " + cmd + " (id=" + id + "; port="+port+")");
+      }
+      return "OK " + cmd;
+    }
+    
+  }
+  
 }