Changeset 2643 for trunk/src/clients/jobagent

Timestamp:

Sep 15, 2006, 4:08:12 PM (16 years ago)

Author:

Nicklas Nordborg

Message:

References #351: External job server usage

In-process job executor is now working. Work has started on the out-of-process job executor.
Added sendMessage and stackTrace property to job for improved error reporting.

Location:

trunk/src/clients/jobagent

Files:

• jobagent.properties.in (modified) (1 diff)
• jobagent.sh (modified) (1 diff)
• net/sf/basedb/clients/jobagent/Agent.java (modified) (3 diffs)
• net/sf/basedb/clients/jobagent/JobExecutor.java (modified) (1 diff)
• net/sf/basedb/clients/jobagent/JobRunner.java (modified) (4 diffs)
• net/sf/basedb/clients/jobagent/executors/DummyJobExecutor.java (modified) (2 diffs)
• net/sf/basedb/clients/jobagent/executors/ProcessJobExecutor.java (modified) (3 diffs)
• net/sf/basedb/clients/jobagent/executors/ThreadJobExecutor.java (modified) (4 diffs)

trunk/src/clients/jobagent/jobagent.properties.in

@@ -62 +62 @@
 # =======================
 
-## The name of the executor class that is responsible for starting the job
+# The name of the executor class that is responsible for starting the job
 ## The default is ProcessJobExecutor which starts job in a separate process
 ## The class must implement the net.sf.basedb.clients.JobExecutor interface
+
+## Executor that executes the jobs in a separate process (recommended)
 agent.executor.class=net.sf.basedb.clients.jobagent.executors.ProcessJobExecutor
+
+## Executor that executes jobs in the same process as the job agent (not recommended)
+# agent.executor.class=net.sf.basedb.clients.jobagent.executors.ThreadJobExecutor
+
+## Executor useful for debugging purposes
+# agent.executor.class=net.sf.basedb.clients.jobagent.executors.DummyJobExecutor
 
 ## Initialisation parameters to the job exector. This string will be passed

trunk/src/clients/jobagent/jobagent.sh

@@ -30 +30 @@
 
 # Execute JobAgent
-java -server -Xmx40M -Xms40M -cp $CP net.sf.basedb.clients.jobagent.AgentController $* &
+java -server -Xmx50M -Xms50M -cp $CP net.sf.basedb.clients.jobagent.AgentController $* &

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

@@ -291 +291 @@
   public Agent(Properties properties)
   {
+    if (log.isDebugEnabled())
+    {
+      log.debug("Creating new agent");
+      for (Map.Entry<Object, Object> entry : properties.entrySet())
+      {
+        Object value = entry.getValue();
+        Object key = entry.getKey();
+        if ("agent.password".equals(key))
+        {
+          value = "******************";
+        }
+        log.debug("Property [" + key + " = " + value + "]");
+      }
+    }
+    
     // BASE settings
     this.login = properties.getProperty("agent.user");
@@ -504 +519 @@
       jobExecutor = createJobExecutor();
       jobQueueChecker = createJobQueueChecker();
+      
+      // Start the BASE application in a new thread so we can return quickly
+      Thread t = new Thread(
+        new Runnable()
+        {
+          public void run()
+          {
+            SessionControl sc = null;
+            try
+            {
+              sc = getSessionControl();
+            }
+            catch (Throwable t)
+            {}
+            if (sc == null || !sc.isLoggedIn()) 
+            {
+              stop();
+            }
+          }
+        }
+      );
+      t.start();
     }
   }
@@ -633 +670 @@
   public SessionControl getSessionControl()
   {
-    if (sc == null)
-    {
-      Application.start(false);
-      sc = Application.newSessionControl("net.sf.basedb.clients.jobagent", 
-        SocketUtil.getLocalHost().toString(), null);
-    }
-    if (!sc.isLoggedIn())
-    {
-      sc.login(login, password, null, false);
+    try
+    {
+      if (sc == null)
+      {
+        log.info("Starting BASE application");
+        Application.start(false);
+        sc = Application.newSessionControl("net.sf.basedb.clients.jobagent", 
+          SocketUtil.getLocalHost().toString(), null);
+      }
+      if (!sc.isLoggedIn())
+      {
+        log.info("Logging in as user: " + login);
+        sc.login(login, password, "Job agent running on host " + getServerName(), false);
+      }
+    }
+    catch (Throwable t)
+    {
+      log.error(t.getMessage(), t);
     }
     return sc;

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

@@ -62 +62 @@
     to the job executor to decide how to handle this. It may either decide
     to kill the job or let it continue as if nothing happened.
+    <p>
+    The job sent to this method has it's status set to {@link Job.Status#PREPARED}.
+    The implementation of this method must change the status to either 
+    {@link Job.Status#DONE} or {@link Job.Status#ERROR}. If the status hasn't changed
+    the job agent will set the status to signal an unknown error.
     
     @param sc A <code>SessionControl</code> where the owner of the job is

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

@@ -24 +24 @@
 package net.sf.basedb.clients.jobagent;
 
+import java.util.Arrays;
+
+import net.sf.basedb.core.DbControl;
+import net.sf.basedb.core.ItemModifiedException;
 import net.sf.basedb.core.Job;
+import net.sf.basedb.core.SessionControl;
 
 /**
@@ -45 +50 @@
   */
   private static final org.apache.log4j.Logger log = 
-    org.apache.log4j.LogManager.getLogger("net.sf.basedb.clients.jobagent.Agent");
+    org.apache.log4j.LogManager.getLogger("net.sf.basedb.clients.jobagent.JobRunner");
 
   private final Job job;
@@ -80 +85 @@
       return;
     }
+    DbControl dc = null;
+    SessionControl sc = null;
     try
     {
       log.info("Executing job " + job + " using executor " + jobExecutor);
-      jobExecutor.executeJob(agent.getImpersonatedSessionControl(job), agent, job, slotToUse);
+      
+      // Log in as the owner of the job
+      sc = agent.getImpersonatedSessionControl(job);
+      dc = sc.newDbControl();
+      
+      // Load a fresh copy of the job
+      Job j = Job.getById(dc, job.getId());
+      
+      // Status may have changed since the queue was checked
+      if (j.getStatus() != Job.Status.WAITING) return;
+      
+      // Change the status to Job.Status.PREPARED
+      j.setPrepared(agent.getServerName());
+      try
+      {
+        dc.commit();
+      }
+      catch (ItemModifiedException ex)
+      {
+        // Another thread or agent has started the job already, return 
+        // without starting the job
+        return;
+      }
+      
+      try
+      {
+        jobExecutor.executeJob(sc, agent, j, slotToUse);
+      }
+      catch (Throwable t)
+      {
+        // Something went wrong, update status to ERROR
+        dc = sc.newDbControl();
+        j = Job.getById(dc, j.getId());
+        j.doneError(t.getMessage(), Arrays.asList(t));
+        dc.commit();
+        throw t;
+      }
+      
+      // Check that that the status has changed from prepared
+      dc = sc.newDbControl();
+      j = Job.getById(dc, j.getId());
+      if (j.getStatus() == Job.Status.PREPARED)
+      {
+        j.doneError("Unknown error, job was never executed.");
+        log.error("Unknown error: Job " + job + " was never executed by executor " + jobExecutor);
+      }
+      else 
+      {
+        log.info("Done executing job: " + job + " with executor " + jobExecutor);
+      }
+      dc.commit();
     }
     catch (Throwable t)
@@ -92 +149 @@
     {
       agent.jobDone(job, slotToUse);
+      if (dc != null) dc.close();
+      if (sc != null && sc.isLoggedIn()) 
+      {
+        try
+        {
+          sc.logout();
+        }
+        catch (Throwable t)
+        {
+          log.error(t.getMessage(), t);
+        }
+      }
     }
   }

trunk/src/clients/jobagent/net/sf/basedb/clients/jobagent/executors/DummyJobExecutor.java

@@ -33 +33 @@
   This is dummy job executor implementation which just sets the status
   of each job to {@link net.sf.basedb.core.Job.Status#DONE} without actually
-  executin the job. This class is usful for debuggin the job agent application.
+  executing the job. This class is usful for debugging the job agent application.
 
   @author nicklas
@@ -71 +71 @@
       job.doneOk("Not really, but used for testing job agent");
       dc.commit();
+      log.info("Done executing: " + job);
     }
     finally

trunk/src/clients/jobagent/net/sf/basedb/clients/jobagent/executors/ProcessJobExecutor.java

@@ -24 +24 @@
 package net.sf.basedb.clients.jobagent.executors;
 
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.io.OutputStream;
+import java.io.PrintWriter;
+import java.io.Reader;
+import java.io.StringWriter;
+import java.io.Writer;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
 import net.sf.basedb.clients.jobagent.Agent;
 import net.sf.basedb.clients.jobagent.JobExecutor;
+import net.sf.basedb.core.DbControl;
 import net.sf.basedb.core.Job;
 import net.sf.basedb.core.SessionControl;
@@ -70 +84 @@
   {
     log.info("Executing job: " + job);
-    // TODO - start new process for executing the job
+    String classPath = System.getProperty("java.class.path");
+    
+    List<String> cmd = new ArrayList<String>(10);
+    cmd.add("java");
+    cmd.add("-server");
+    cmd.add("-cp");
+    cmd.add(classPath);
+    cmd.add("net.sf.basedb.clients.jobagent.executors.ThreadJobExecutor");
+    cmd.add(Integer.toString(job.getId()));
+
+    log.debug("Using class path: " + classPath);
+    ProcessBuilder builder = new ProcessBuilder(cmd);
+    builder.redirectErrorStream(true);
+    
+    Process process = null;
+    DbControl dc = null;
+    try
+    {
+      try
+      {
+        process = builder.start();
+      }
+      catch (Throwable t)
+      {
+        log.error("Error executing job: " + job, t);
+        dc = sc.newDbControl();
+        job = Job.getById(dc, job.getId());
+        job.doneError(t.getMessage(), Arrays.asList(t));
+        dc.commit();
+        return;
+      }
+      
+      // Set up threads for reading output
+      StringWriter result = new StringWriter();
+      Thread t = new Thread(new StreamRedirector(
+        new InputStreamReader(process.getInputStream()), result));
+      t.start();
+
+      try
+      {
+        log.info("Waiting for process to end");
+        int exitCode = process.waitFor();
+        if (exitCode != 0)
+        {
+          log.error("Process ended with exit code: " + exitCode);
+          log.error("Process output is: " + result.toString());
+          dc = sc.newDbControl();
+          job = Job.getById(dc, job.getId());
+          job.doneError("Job ended with exit code: " + exitCode, 
+            Arrays.asList(new Exception(result.toString())));
+          dc.commit();
+        }
+        else
+        {
+          log.info("Process ended");
+        }
+      }
+      catch (InterruptedException ex)
+      {
+        log.info("Job was interrupted: " + job, ex);
+        // Kill the process
+        process.destroy();
+        dc = sc.newDbControl();
+        job = Job.getById(dc, job.getId());
+        job.doneError(ex.getMessage(), Arrays.asList(ex));
+        dc.commit();
+      }
+    }
+    finally
+    {
+      if (dc != null) dc.close();
+    }
+    
   }
 
@@ -76 +162 @@
   {}
   // -------------------------------------------
+  
+  
+  public static class StreamRedirector
+    implements Runnable
+  {
+  
+    private final Reader in;
+    private final Writer out;
+    public StreamRedirector(Reader in, Writer out)
+    {
+      this.in = in;
+      this.out = out;
+    }
+    
+    public void run()
+    {
+      BufferedReader reader = new BufferedReader(in);
+      PrintWriter writer = new PrintWriter(out);
+      String line = null;
+      try
+      {
+        while ((line = reader.readLine()) != null)
+        {
+          writer.println(line);
+        }
+      }
+      catch (IOException ex)
+      {}
+    }
+  }
+  
 }

trunk/src/clients/jobagent/net/sf/basedb/clients/jobagent/executors/ThreadJobExecutor.java

@@ -24 +24 @@
 package net.sf.basedb.clients.jobagent.executors;
 
+import java.util.Arrays;
+import java.util.List;
+
 import net.sf.basedb.clients.jobagent.Agent;
 import net.sf.basedb.clients.jobagent.JobExecutor;
+import net.sf.basedb.core.DbControl;
 import net.sf.basedb.core.Job;
+import net.sf.basedb.core.PluginExecutionRequest;
+import net.sf.basedb.core.PluginResponse;
 import net.sf.basedb.core.SessionControl;
+import net.sf.basedb.core.plugin.Response;
 
 /**
@@ -50 +57 @@
   */
   private static final org.apache.log4j.Logger log = 
-    org.apache.log4j.LogManager.getLogger("net.sf.basedb.clients.jobagent.executors.ProcessJobExecutor");
+    org.apache.log4j.LogManager.getLogger("net.sf.basedb.clients.jobagent.executors.ThreadJobExecutor");
   
   
@@ -66 +73 @@
   {
     log.info("Executing job: " + job);
-    // TODO - execute job
+    DbControl dc = null;
+    try
+    {
+      dc = sc.newDbControl();
+      
+      // Load a fresh copy of the job item
+      job = Job.getById(dc, job.getId());
+      PluginExecutionRequest exec = null;
+      
+      try
+      {
+        exec = job.execute(null, agent.getServerName());
+      }
+      catch (Throwable t)
+      {
+        //  The job could not be started, set error status and send message to owner
+        job.doneError(t.getMessage(), Arrays.asList(t));
+        log.error("Error executing job: " + job, t);
+      }
+      dc.commit();
+
+      PluginResponse response = exec.invoke();
+      if (response.getStatus() == Response.Status.ERROR)
+      {
+        // There was an error, job status and message is already sent to the user
+        log.error("Error executing job: " + job + "; " + response.getMessage());
+        List<? extends Throwable> errors = response.getErrorList();
+        if (errors != null)
+        {
+          for (Throwable t : errors)
+          {
+            log.error(t.getMessage(), t);
+          }
+        }
+      }
+      else
+      {
+        log.info("Done executing: " + job);
+      }
+    }
+    finally
+    {
+      if (dc != null) dc.close();
+    }
   }
 
@@ -72 +122 @@
   {}
   // -------------------------------------------
+  
+  public static void main(String[] args)
+  {
+    System.out.println("Running job with ID: " + args[0]);
+    System.exit(11);
+  }
+  
 }