#166: added interrupt handling

agent/org.linkedin.glu.agent-api/src/main/groovy/org/linkedin/glu/agent/api/Agent.groovy

@@ -315,6 +315,13 @@ public interface Agent
    */
   def streamCommandResults(def args) throws NoSuchCommandException, AgentException
 
+  /**
+   * Interrupts the command.
+   *
+   * @param args.id the id of the command to interrupt
+   * @return <code>true</code> if the command was interrupted properly or <code>false</code> if
+   * there was no such command or already completed
+   */
   boolean interruptCommand(args) throws AgentException
 
   /********************************************************************

agent/org.linkedin.glu.agent-impl/src/main/groovy/org/linkedin/glu/agent/impl/command/CommandManagerImpl.groovy

@@ -139,7 +139,7 @@ public class CommandManagerImpl implements CommandManager
 
     try
     {
-      def res = commandExecution.waitForCompletion(args.timeout)
+      def res = commandExecution.getExitValue(args.timeout)
       commandExecution.log.info("waitForCommand(${GluGroovyCollectionUtils.xorMap(args, ['id'])}): ${res}")
       return res
     }
@@ -173,7 +173,7 @@ public class CommandManagerImpl implements CommandManager
     if(commandExecution)
     {
       res = commandExecution.interruptExecution()
-      commandExecution.log.info("interruptCommand(${GluGroovyCollectionUtils.xorMap(args, ['id'])})")
+      commandExecution.log.info("interruptCommand(${GluGroovyCollectionUtils.xorMap(args, ['id'])}): ${res}")
     }
     else
     {

commands/org.linkedin.glu.commands-impl/src/main/groovy/org/linkedin/glu/commands/impl/CommandExecution.groovy

@@ -23,6 +23,7 @@ import java.util.concurrent.ExecutorService
 import org.linkedin.glu.groovy.utils.concurrent.FutureTaskExecution
 import org.slf4j.Logger
 import org.linkedin.groovy.util.config.Config
+import java.util.concurrent.TimeoutException
 
 /**
  * @author yan@pongasoft.com */
@@ -78,15 +79,18 @@ public class CommandExecution<T>
     completionTime > 0
   }
 
-  def waitForCompletion(def timeout)
+  void waitForCompletion(def timeout)
   {
-    try
+    if(!isCompleted())
     {
-      futureExecution.get(timeout)
-    }
-    catch(ExecutionException e)
-    {
-      throw e.cause
+      try
+      {
+        futureExecution.get(timeout)
+      }
+      catch(ExecutionException e)
+      {
+        // ok we just want to wait until the command completes but no more than the timeout
+      }
     }
   }
 
@@ -98,25 +102,67 @@ public class CommandExecution<T>
   def getExitValue()
   {
     if(isCompleted())
+    {
+      try
+      {
+        futureExecution.get()
+      }
+      catch(ExecutionException e)
+      {
+        throw e.cause
+      }
+    }
+    else
+      return null
+  }
+
+  def getExitValue(timeout)
+  {
     try
     {
-      futureExecution.get()
+      futureExecution.get(timeout)
     }
     catch(ExecutionException e)
     {
       throw e.cause
     }
   }
 
-  def getExitValue(timeout)
+  /**
+   * Completion value either return the result of the call if succeeded or the exception
+   * if an exception was thrown. Does not throw an exception! Does not wait!
+   */
+  def getCompletionValue()
+  {
+    if(isCompleted())
+    {
+      try
+      {
+        futureExecution.get()
+      }
+      catch(ExecutionException e)
+      {
+        e.cause
+      }
+    }
+    else
+      return null
+  }
+
+  /**
+   * Completion value either return the result of the call if succeeded or the exception
+   * if an exception was thrown. Throws only the <code>TimeoutException</code> if cannot get
+   * a result in the timeout provided
+   */
+  def getCompletionValue(timeout) throws TimeoutException
   {
     try
     {
       futureExecution.get(timeout)
     }
     catch(ExecutionException e)
     {
-      throw e.cause
+      e.cause
     }
   }
 

orchestration/org.linkedin.glu.orchestration-engine/src/main/groovy/org/linkedin/glu/orchestration/engine/agents/AgentsService.groovy

@@ -92,4 +92,14 @@ interface AgentsService
                            args,
                            Closure commandResultProcessor)
 
+  /**
+   * Interrupts the command.
+   *
+   * @param args.id the id of the command to interrupt
+   * @return <code>true</code> if the command was interrupted properly or <code>false</code> if
+   * there was no such command or already completed
+   */
+  boolean interruptCommand(Fabric fabric,
+                           String agentName,
+                           args)
 }

orchestration/org.linkedin.glu.orchestration-engine/src/main/groovy/org/linkedin/glu/orchestration/engine/agents/AgentsServiceImpl.groovy

@@ -248,6 +248,14 @@ class AgentsServiceImpl implements AgentsService, AgentURIProvider
     }
   }
 
+  @Override
+  boolean interruptCommand(Fabric fabric, String agentName, def args)
+  {
+    withRemoteAgent(fabric, agentName) { Agent agent ->
+      agent.interruptCommand(args)
+    } as boolean
+  }
+
   /**
    * Create the system entry for the given agent and mountpoint.
    */

orchestration/org.linkedin.glu.orchestration-engine/src/main/groovy/org/linkedin/glu/orchestration/engine/agents/AuditedAgentsService.groovy

@@ -115,4 +115,11 @@ public class AuditedAgentsService implements AgentsService
                             res?.id?.toString())
     }
   }
+
+  @Override
+  boolean interruptCommand(Fabric fabric, String agentName, def args)
+  {
+    auditLogService.audit('agent.interruptCommand', "${agentName} / ${fabric.name} / ${args.id}")
+    agentsService.interruptCommand(fabric, agentName, args)
+  }
 }

orchestration/org.linkedin.glu.orchestration-engine/src/main/groovy/org/linkedin/glu/orchestration/engine/commands/CommandExecutionStorage.groovy

@@ -41,6 +41,14 @@ public interface CommandExecutionStorage
                                   Long stderrTotalBytesCount,
                                   String exitValue)
 
+  DbCommandExecution endExecution(String commandId,
+                                  long endTime,
+                                  byte[] stdoutFirstBytes,
+                                  Long stdoutTotalBytesCount,
+                                  byte[] stderrFirstBytes,
+                                  Long stderrTotalBytesCount,
+                                  Throwable exception)
+
   DbCommandExecution findCommandExecution(String fabric, String commandId)
 
   Map findCommandExecutions(String fabric, String agent, def params)

orchestration/org.linkedin.glu.orchestration-engine/src/main/groovy/org/linkedin/glu/orchestration/engine/commands/CommandExecutionStorageImpl.groovy

@@ -21,6 +21,8 @@ import org.slf4j.Logger
 import org.slf4j.LoggerFactory
 import org.linkedin.util.annotations.Initializable
 import org.linkedin.glu.groovy.utils.collections.GluGroovyCollectionUtils
+import org.linkedin.util.lang.MemorySize
+import org.linkedin.glu.utils.io.LimitedOutputStream
 
 /**
  * @author yan@pongasoft.com */
@@ -32,6 +34,9 @@ public class CommandExecutionStorageImpl implements CommandExecutionStorage
   @Initializable
   int maxResults = 25
 
+  @Initializable
+  MemorySize stackTraceMaxSize = MemorySize.parse('255')
+
   @Override
   DbCommandExecution startExecution(String fabric,
                                     String agent,
@@ -71,6 +76,25 @@ public class CommandExecutionStorageImpl implements CommandExecutionStorage
                                   byte[] stderrFirstBytes,
                                   Long stderrTotalBytesCount,
                                   String exitValue)
+  {
+    endExecution(commandId,
+                 endTime,
+                 stdoutFirstBytes,
+                 stdoutTotalBytesCount,
+                 stderrFirstBytes,
+                 stderrTotalBytesCount,
+                 exitValue,
+                 false)
+  }
+
+  DbCommandExecution endExecution(String commandId,
+                                  long endTime,
+                                  byte[] stdoutFirstBytes,
+                                  Long stdoutTotalBytesCount,
+                                  byte[] stderrFirstBytes,
+                                  Long stderrTotalBytesCount,
+                                  String exitValue,
+                                  boolean isException)
   {
     DbCommandExecution.withTransaction {
       DbCommandExecution execution = DbCommandExecution.findByCommandId(commandId)
@@ -86,6 +110,7 @@ public class CommandExecutionStorageImpl implements CommandExecutionStorage
         execution.stderrFirstBytes = stderrFirstBytes
         execution.stderrTotalBytesCount = stderrTotalBytesCount
         execution.exitValue = exitValue
+        execution.isException = isException
 
         if(!execution.save())
         {
@@ -96,6 +121,29 @@ public class CommandExecutionStorageImpl implements CommandExecutionStorage
     }
   }
 
+  @Override
+  DbCommandExecution endExecution(String commandId,
+                                  long endTime,
+                                  byte[] stdoutFirstBytes,
+                                  Long stdoutTotalBytesCount,
+                                  byte[] stderrFirstBytes,
+                                  Long stderrTotalBytesCount,
+                                  Throwable exception)
+  {
+    def baos = new ByteArrayOutputStream()
+
+    def os = new PrintStream(new LimitedOutputStream(baos, stackTraceMaxSize))
+    os.withStream { exception.printStackTrace(it) }
+    endExecution(commandId,
+                 endTime,
+                 stdoutFirstBytes,
+                 stdoutTotalBytesCount,
+                 stderrFirstBytes,
+                 stderrTotalBytesCount,
+                 new String(baos.toByteArray()),
+                 true)
+  }
+
   @Override
   DbCommandExecution findCommandExecution(String fabric, String commandId)
   {

orchestration/org.linkedin.glu.orchestration-engine/src/main/groovy/org/linkedin/glu/orchestration/engine/commands/CommandsService.groovy

@@ -59,4 +59,15 @@ public interface CommandsService
   DbCommandExecution findCommandExecution(Fabric fabric, String commandId)
 
   Map findCommandExecutions(Fabric fabric, String agentName, def params)
+
+  /**
+   * Interrupts the command.
+   *
+   * @param args.id the id of the command to interrupt
+   * @return <code>true</code> if the command was interrupted properly or <code>false</code> if
+   * there was no such command or already completed
+   */
+  boolean interruptCommand(Fabric fabric,
+                           String agentName,
+                           String commandId)
 }