Changeset 6432

Timestamp:

Mar 14, 2014, 8:23:38 AM (10 years ago)

Author:

Nicklas Nordborg

Message:

References #1796: Improve support for jobs running on external servers

Added an extension point (net.sf.basedb.core.signal.job) and ExtensionSignalTransporter that uses this extension point to send signals to jobs. The web interface has been updated to send 'ABORT' and 'STATUS' signals also to jobs in the 'WAITING' state.

The 'WAITING' state is somewhat redefined so that external jobs can use this state to indicate that the job has actually been queued. This means that jobs are not put in 'WAITING' state as soon as parameters has been added to it, but Job.setScheduled() must be called first. This change probably affect a lot of the test code only (must check this), but not regular usage with the request/response api.

Added externalId property to job so that external jobs have a place to store an identifier.

Location:

trunk

Files:

• src/core/core-extensions.xml (modified) (1 diff)
• src/core/net/sf/basedb/core/Install.java (modified) (1 diff)
• src/core/net/sf/basedb/core/Job.java (modified) (9 diffs)
• src/core/net/sf/basedb/core/Update.java (modified) (2 diffs)
• src/core/net/sf/basedb/core/data/JobData.java (modified) (1 diff)
• src/core/net/sf/basedb/core/signal/ExtensionSignalTransporter.java (added)
• src/core/net/sf/basedb/core/signal/LocalSignalReceiver.java (modified) (1 diff)
• src/core/net/sf/basedb/core/signal/Signal.java (modified) (1 diff)
• src/core/net/sf/basedb/core/signal/SignalHandler.java (modified) (2 diffs)
• www/views/jobs/ajax.jsp (modified) (1 diff)
• www/views/jobs/index.jsp (modified) (1 diff)
• www/views/jobs/view_job.jsp (modified) (1 diff)

trunk/src/core/core-extensions.xml

@@ -152 +152 @@
   </extension-point>
   
+  <extension-point 
+    id="net.sf.basedb.core.signal.job">
+    <action-class>net.sf.basedb.core.signal.SignalHandler</action-class>
+    <name>Job signal handler</name>
+    <description>
+      Extension point for sending signals to jobs. Intended for jobs 
+      running on external servers outside of direct control for BASE.
+      The ClientContext is populated with the current SessionControl and
+      Job item, and the "signal-uri" attribute is set to the URI registered
+      with Job.setSignalTransporter. It is recommended that implementations
+      return quickly. If the action takes a long time the handler should spawn
+      a new thread.
+    </description>
+  </extension-point>
+  
   <extension 
     id="net.sf.basedb.core.uri.http-connection-manager"

trunk/src/core/net/sf/basedb/core/Install.java

@@ -118 +118 @@
     method.
   */
-  public static final int NEW_SCHEMA_VERSION = Integer.valueOf(115).intValue();
+  public static final int NEW_SCHEMA_VERSION = Integer.valueOf(116).intValue();
   
   public static synchronized int createTables(SchemaGenerator.Mode mode, ProgressReporter progress,

trunk/src/core/net/sf/basedb/core/Job.java

@@ -29 +29 @@
 import net.sf.basedb.core.data.PluginConfigurationData;
 import net.sf.basedb.core.data.PluginDefinitionData;
-
 import net.sf.basedb.core.plugin.Plugin;
 import net.sf.basedb.core.plugin.InteractivePlugin;
@@ -36 +35 @@
 import net.sf.basedb.core.plugin.Response;
 import net.sf.basedb.core.plugin.Plugin.MainType;
+import net.sf.basedb.core.signal.ExtensionSignalTransporter;
 import net.sf.basedb.core.signal.Signal;
 import net.sf.basedb.core.signal.SignalTransporter;
+import net.sf.basedb.util.extensions.ClientContext;
 
 import java.util.Collection;
@@ -396 +397 @@
 
   /**
+    The maximum allowed length of the external ID.
+    @since 3.3
+  */
+  public static final int MAX_EXTERNAL_ID_LENGTH = JobData.MAX_EXTERNAL_ID_LENGTH;
+
+  /**
+    Get the external ID of a job running on an external server.
+    @since 3.3
+  */
+  public String getExternalId()
+  {
+    return getData().getExternalId();
+  }
+  /**
+    Set the external ID.
+    @since 3.3
+  */
+  public void setExternalId(String externalId)
+  {
+    checkPermission(Permission.WRITE);
+    getData().setExternalId(StringUtil.setNullableString(externalId, "externalId", MAX_EXTERNAL_ID_LENGTH));
+  }
+
+  /**
     Get the type of job, ie. if it is a job executed by a plugin
     or not.
@@ -836 +861 @@
     Note 3! This method always returns the same transporter instance if
     it is called multiple times.
+    <p>
+    Note 4! If the signal transporter is a {@link ExtensionSignalTransporter} 
+    the extension point is "net.sf.basedb.core.signal.job". The client
+    context is populated with a {@link SessionControl} and the current item
+    is this {@link Job}. The information in the job should be considered read-only,
+    if the signal handler wants to update information it must create a new
+    {@link DbControl} from the session control and re-load the job.
     
     @return An initialised signal transporter object, or null
@@ -858 +890 @@
           SignalTransporter transporter = (SignalTransporter)Class.forName(transporterClass).newInstance();
           transporter.init(initParams);
+          
+          if (transporter instanceof ExtensionSignalTransporter)
+          {
+            ExtensionSignalTransporter xtTransporter = (ExtensionSignalTransporter)transporter;
+            ClientContext context = new ClientContext(getSessionControl(), this);
+            xtTransporter.setContext(context);
+            xtTransporter.setExtensionPointId("net.sf.basedb.core.signal.job");
+          }
+          
           signalTransporter = new JobSignalTransporter(transporter);
         }
@@ -870 +911 @@
   
   /**
+    Request that the underlying job updates the status information about
+    itself. This method is intended to be used for external jobs that
+    are not BASE plug-ins (which can report progress through the plug-in API).
+
+    This method sends a {@link Signal#STATUS} signal to the 
+    {@link SignalTransporter} that has been registered for the job. 
+    If there is no signal transporter this method does nothing. 
+    
+    The signal transporter should deliver the signal as quickly as possible
+    and then return. The actual progress update may take place at a later
+    time when it is convenient to do so for the job.
+    
+    @return TRUE if a {@link Signal#STATUS} could be sent, FALSE if not
+    @since 3.3
+  */
+  public boolean requestStatusUpdate()
+  {
+    boolean signalSent = false;
+    if (hasSignalTransporter())
+    {
+      try
+      {
+        SignalTransporter transporter = getSignalTransporter();
+        Collection<Signal> supported = transporter.getSupportedSignals();
+        if (supported == null || supported.contains(Signal.STATUS))
+        {
+          transporter.send(Signal.STATUS);
+          signalSent = true;
+        }
+      }
+      catch (RuntimeException ex)
+      {
+        ex.printStackTrace();
+      }
+    }
+    return signalSent;
+  }
+
+  
+  /**
     Get the date and time the job ended.
     @return A date, or null if the job hasn't ended
@@ -879 +960 @@
 
   /**
+    Set the job's status to {@link Status#WAITING}. This status is
+    used to signal that a job has been fully configured and is ready
+    for execution.
+    
+    @param server Name of the server which this job will executed on
+    @param agent The job agent that is going to execute the
+      job or null if it is executed by something else, eg. the
+      internal job queue
+    @throws PermissionDeniedException If the logged in user doesn't have write permission.
+    @since 3.3
+  */
+  public void setScheduled(String server, JobAgent agent)
+    throws PermissionDeniedException
+  {
+    checkPermission(Permission.WRITE);
+    Job.Status status = getStatus();
+    if (status != Status.UNCONFIGURED) 
+    {
+      throw new PermissionDeniedException("Can't schedule a job with status '"+getStatus()+"': "+toString());
+    }
+    JobData data = getData();
+    data.setScheduled(new Date());
+    data.setStatus(Status.WAITING.getValue());
+    data.setServer(StringUtil.setNullableString(server, "server", MAX_SERVER_LENGTH));
+    getData().setJobAgentId(agent == null ? null : agent.getId());
+  }
+  
+  /**
     Set the job's status to {@link Status#PREPARED}. This status is
-    used to signal that a job is about to be executed, but the actual excecution
+    used to signal that a job is about to be executed, but the actual execution
     hasn't started yet. This status prevents two job agents from trying to
     start the same job twice.
@@ -1350 +1459 @@
       throw new PermissionDeniedException("Cannot configure an executing job: "+toString());
     }
-    getData().setStatus(Status.WAITING.getValue());
-    getData().setScheduled(new Date());
+    //getData().setStatus(Status.WAITING.getValue());
+    //getData().setScheduled(new Date());
     setParameterValuesInternal(name, label, description, parameterType, Arrays.asList(value), true);
   }
@@ -1402 +1511 @@
       throw new PermissionDeniedException("Cannot configure an executing job: "+toString());
     }
-    getData().setStatus(Status.WAITING.getValue());
-    getData().setScheduled(new Date());
+    //getData().setStatus(Status.WAITING.getValue());
+    //getData().setScheduled(new Date());
     setParameterValuesInternal(name, label, description, parameterType, values, true);
   }

trunk/src/core/net/sf/basedb/core/Update.java

@@ -208 +208 @@
     </td>
   </tr>
+  
+  <tr>
+    <td>116</td>
+    <td>
+      Added {@link JobData#getExternalId()}. No special database update is needed. Only increase 
+      the schema version.
+    </td>
+  </tr>
   </table>
 
@@ -340 +348 @@
       }
       
-      if (schemaVersion < 115)
-      {
-        if (progress != null) progress.display((int)(progress_current), "--Updating schema version: " + schemaVersion + " -> 115...");
-        // Schemaversion 115 only updates the version number
-        schemaVersion = setSchemaVersionInTransaction(session, 115);
-        progress_current += progress_step;
+      if (schemaVersion < 116)
+      {
+        if (progress != null) progress.display((int)(progress_current), "--Updating schema version: " + schemaVersion + " -> 116...");
+        // Schemaversion 115 and 116 only updates the version number
+        schemaVersion = setSchemaVersionInTransaction(session, 116);
+        progress_current += 2 * progress_step;
       }
 

trunk/src/core/net/sf/basedb/core/data/JobData.java

@@ -105 +105 @@
   }
   
+  /**
+    The maximum allowed length of the external ID.
+    @since 3.3
+  */
+  public static final int MAX_EXTERNAL_ID_LENGTH = 255;
+  private String externalId;
+  /**
+    External ID of a job running on an external system.
+    @hibernate.property column="`external_id`" type="string" length="255" not-null="false"
+    @since 3.3
+  */
+  public String getExternalId()
+  {
+    return externalId;
+  }
+  public void setExternalId(String externalId)
+  {
+    this.externalId = externalId;
+  }
+  
   private PluginDefinitionData pluginDefinition;
   /**

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

@@ -114 +114 @@
   {
     logger.info("Receiving signal " + signal.getId() + " to " + handlerId);
-    SignalHandler handler = super.getSignalHandler(handlerId);
+    SignalHandler handler = getSignalHandler(handlerId);
     if (handler != null) handler.handleSignal(signal);
   }

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

@@ -68 +68 @@
     );
 
+  /**
+    The STATUS signal, that requests that the receiver should
+    update the status on the system. This signal is typically sent 
+    to a job managed by an external job manager when someone
+    on the BASE side want to have an update about the progress.
+    @since 3.3
+  */
+  public static final Signal STATUS = 
+    registerSignal("STATUS", "Status",
+      "Sending this signal indicates that the system wants a status " +
+      "update from the receiver. Exactly what this means is up to the sender " +
+      "and reciever to agree on. Typically, this signal can be sent to external " +
+      "job managers when someone on the BASE side wants to have an update about the progress."
+    );
   
   /**

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

@@ -24 +24 @@
 import java.util.Collection;
 
+import net.sf.basedb.util.extensions.Action;
+
 /**
   A signal handler is an object that is able to receive and act on signals.
@@ -41 +43 @@
 */
 public interface SignalHandler
+  extends Action
 {
   /**
-    Handle the given signal. 
+    Handle the given signal. It is recommended that signals are handled
+    as quickly as possible. If there is a risk that it may take a long time
+    it is recommended to spawn a new thread and handle the signal asynchronously.
     @param signal The signal to handle
     @throws UnsupportedSignalException If the signal is not supported

trunk/www/views/jobs/ajax.jsp

@@ -51 +51 @@
     dc = sc.newDbControl();
     Job job = Job.getById(dc, itemId);
+    job.requestStatusUpdate();
     json.put("id", job.getId());
     json.put("jobStatus", job.getStatus().name());

trunk/www/views/jobs/index.jsp

@@ -263 +263 @@
     dc = sc.newDbControl();
     Job job = Job.getById(dc, cc.getId());
-    if (job.getStatus() == Job.Status.WAITING)
-    {
-      job.doneError("Aborted by user");
+    SignalTransporter signalTransporter = job.getSignalTransporter();
+    if (signalTransporter == null)
+    {
+      if (job.getStatus() == Job.Status.WAITING)
+      {
+        job.doneError("Aborted by user");
+      }
     }
     else
     {
-      SignalTransporter signalTransporter = job.getSignalTransporter();
-      if (signalTransporter != null) signalTransporter.send(Signal.ABORT);
+      signalTransporter.send(Signal.ABORT);
     }
     dc.commit();

trunk/www/views/jobs/view_job.jsp

@@ -162 +162 @@
   }
   
-  // Check if the plug-in supports the "Abort" signal
-  boolean supportsAbort = status == Job.Status.WAITING && writePermission;
-  if (status == Job.Status.EXECUTING && writePermission)
+  // Check if the job can be aborted
+  boolean supportsAbort = false; 
+  
+  // The status must be executable or EXECUTING
+  if ((status == Job.Status.EXECUTING || status.isExecutable()) && writePermission)
   {
+    // Check if a signal-transporter exists
     try
     {
       SignalTransporter signalTransporter = job.getSignalTransporter();
-      Collection<Signal> supportedSignals = signalTransporter != null ? 
-        signalTransporter.getSupportedSignals() : null;
-      supportsAbort = signalTransporter != null && 
-        (supportedSignals == null || supportedSignals.contains(Signal.ABORT));
+      if (signalTransporter == null)
+      {
+        // The job is only abortable if it is waiting
+        supportsAbort = status == Job.Status.WAITING;
+      }
+      else
+      {
+        // Otherwise we must check with the signal transporter
+        Collection<Signal> supportedSignals = signalTransporter.getSupportedSignals();
+        supportsAbort = supportedSignals == null || supportedSignals.contains(Signal.ABORT);
+      }
     }
     catch (Exception ex)
     {}
   }
+  if (autoUpdate) job.requestStatusUpdate();
   %>
   <base:page type="popup" title="<%=title%>" id="view-page">