ambari-server/src/main/java/org/apache/ambari/server/serveraction/ServerActionExecutor.java - ambari - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */

 package org.apache.ambari.server.serveraction;

 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;

 import org.apache.ambari.server.AmbariException;
 import org.apache.ambari.server.Role;
 import org.apache.ambari.server.StaticallyInject;
 import org.apache.ambari.server.actionmanager.ActionDBAccessor;
 import org.apache.ambari.server.actionmanager.ExecutionCommandWrapper;
 import org.apache.ambari.server.actionmanager.HostRoleCommand;
 import org.apache.ambari.server.actionmanager.HostRoleStatus;
 import org.apache.ambari.server.actionmanager.Request;
 import org.apache.ambari.server.agent.CommandReport;
 import org.apache.ambari.server.agent.ExecutionCommand;
 import org.apache.ambari.server.configuration.Configuration;
 import org.apache.ambari.server.controller.internal.CalculatedStatus;
 import org.apache.ambari.server.security.authorization.internal.InternalAuthenticationToken;
 import org.apache.ambari.server.utils.StageUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 import com.google.inject.Inject;
 import com.google.inject.Injector;
 import org.springframework.security.core.context.SecurityContextHolder;

 /**
  * Server Action Executor used to execute server-side actions (or tasks)
  * <p/>
  * The ServerActionExecutor executes in its own thread, polling for AMBARI_SERVER_ACTION
  * HostRoleCommands queued for execution.  It is expected that this thread is managed by the
  * ActionScheduler such that it is started when the ActionScheduler is started and stopped when the
  * ActionScheduler is stopped.
  */
 @StaticallyInject
 public class ServerActionExecutor {

   private final static Logger LOG = LoggerFactory.getLogger(ServerActionExecutor.class);
   private final static Long DEFAULT_EXECUTION_TIMEOUT_MS = 1000L * 60 * 5;
   private final static Long POLLING_TIMEOUT_MS = 1000L * 5;

   /**
    * An injector to use to inject objects into ServerAction instances.
    */
   @Inject
   private static Injector injector;

   /**
    * The Ambari configuration used to obtain configured details such as the default server-side action
    * timeout.
    */
   @Inject
   private static Configuration configuration;

   /**
    * Maps request IDs to "blackboards" of shared data.
    * <p/>
    * This map is not synchronized, so any access to it should synchronize on
    * requestSharedDataMap object
    */
   private final Map<Long, ConcurrentMap<String, Object>> requestSharedDataMap =
       new HashMap<Long, ConcurrentMap<String, Object>>();

   /**
    * The hostname of the (Ambari) server.
    * <p/>
    * This hostname is cached so that cycles are spent querying for it more than once.
    */
   private final String serverHostName;

   /**
    * Database accessor to query and update the database of action commands.
    */
   private final ActionDBAccessor db;

   /**
    * Internal locking object used to manage access to activeAwakeRequest.
    */
   private final Object wakeupSyncObject = new Object();

   /**
    * Timeout (in milliseconds) used to throttle polling of database for new action commands.
    */
   private final long sleepTimeMS;

   /**
    * Flag used to help keep thing moving in the event an "awake" request was encountered while busy
    * handing an action.
    */
   private boolean activeAwakeRequest = false;

   /**
    * A reference to the Thread handling the work for this ServerActionExecutor
    */
   private Thread executorThread = null;

   /**
    * Statically initialize the Injector
    * <p/>
    * This should only be used for unit tests.
    *
    * @param injector the Injector to (manually) statically inject
    */
   public static void init(Injector injector) {
     ServerActionExecutor.injector = injector;
   }

   /**
    * Creates a new ServerActionExecutor
    *
    * @param db          the ActionDBAccessor to use to read and update tasks
    * @param sleepTimeMS the time (in milliseconds) to wait between polling the database for more tasks
    */
   public ServerActionExecutor(ActionDBAccessor db, long sleepTimeMS) {
     serverHostName = StageUtils.getHostName();
     this.db = db;
     this.sleepTimeMS = (sleepTimeMS < 1) ? POLLING_TIMEOUT_MS : sleepTimeMS;
   }

   /**
    * Starts this ServerActionExecutor's main thread.
    */
   public void start() {
     LOG.info("Starting Server Action Executor thread...");
     executorThread = new Thread(new Runnable() {

       @Override
       public void run() {
         while (!Thread.interrupted()) {
           try {
             synchronized (wakeupSyncObject) {
               if (!activeAwakeRequest) {
                 wakeupSyncObject.wait(sleepTimeMS);
               }
               activeAwakeRequest = false;
             }

             doWork();
           } catch (InterruptedException e) {
             LOG.warn("Server Action Executor thread interrupted, starting to shutdown...");
             break;
           }
         }

         LOG.info("Server Action Executor thread shutting down...");
       }
     }, "Server Action Executor");
     executorThread.start();

     if (executorThread.isAlive()) {
       LOG.info("Server Action Executor thread started.");
     }
   }

   /**
    * Attempts to stop this ServerActionExecutor's main thread.
    */
   public void stop() {
     LOG.info("Stopping Server Action Executor thread...");

     if (executorThread != null) {
       executorThread.interrupt();

       // Wait for about 60 seconds for the thread to stop
       for (int i = 0; i < 120; i++) {
         try {
           executorThread.join(500);
         } catch (InterruptedException e) {
           // Ignore this...
         }

         if (!executorThread.isAlive()) {
           break;
         }
       }

       if (!executorThread.isAlive()) {
         executorThread = null;
       }
     }

     if (executorThread == null) {
       LOG.info("Server Action Executor thread stopped.");
     } else {
       LOG.warn("Server Action Executor thread hasn't stopped, giving up waiting.");
     }
   }

   /**
    * Attempts to force this ServerActionExecutor to wake up and do work.
    * <p/>
    * Should be called from another thread when we want scheduler to
    * make a run ASAP (for example, to process desired configs of SCHs).
    * The method is guaranteed to return quickly.
    */
   public void awake() {
     synchronized (wakeupSyncObject) {
       activeAwakeRequest = true;
       wakeupSyncObject.notify();
     }
   }

   /**
    * Returns a Map to be used to share data among server actions within a given request context.
    *
    * @param requestId a long identifying the id of the relevant request
    * @return a ConcurrentMap of "shared" data
    */
   private ConcurrentMap<String, Object> getRequestSharedDataContext(long requestId) {
     synchronized (requestSharedDataMap) {
       ConcurrentMap<String, Object> map = requestSharedDataMap.get(requestId);

       if (map == null) {
         map = new ConcurrentHashMap<String, Object>();
         requestSharedDataMap.put(requestId, map);
       }

       return map;
     }
   }

   /**
    * Cleans up orphaned shared data Maps due to completed or failed request
    * contexts. We are unable to use {@link Request#getStatus()} since this field
    * is not populated in the database but, instead, calculated in realtime.
    */
   private void cleanRequestShareDataContexts() {
     // if the cache is empty, do nothing
     if (requestSharedDataMap.isEmpty()) {
       return;
     }

     try {
       // for every item in the map, get the request and check its status
       synchronized (requestSharedDataMap) {
         Set<Long> requestIds = requestSharedDataMap.keySet();
         List<Request> requests = db.getRequests(requestIds);
         for (Request request : requests) {
           // calcuate the status from the stages and then remove from the map if
           // necessary
           CalculatedStatus calculatedStatus = CalculatedStatus.statusFromStages(
               request.getStages());

           // calcuate the status of the request
           HostRoleStatus status = calculatedStatus.getStatus();

           // remove the request from the map if the request is COMPLETED or
           // FAILED
           switch (status) {
             case FAILED:
             case COMPLETED:
               requestSharedDataMap.remove(request.getRequestId());
               break;
             default:
               break;
           }
         }
       }
     } catch (Exception exception) {
       LOG.warn("Unable to clear the server-side action request cache", exception);
     }
   }

   /**
    * A helper method to create CommandReports indicating the action/task is in progress
    *
    * @return a new CommandReport
    */
   private CommandReport createInProgressReport() {
     CommandReport commandReport = new CommandReport();
     commandReport.setStatus(HostRoleStatus.IN_PROGRESS.toString());
     commandReport.setStdErr("");
     commandReport.setStdOut("");
     return commandReport;
   }

   /**
    * A helper method to create CommandReports indicating the action/task had timed out
    *
    * @return a new CommandReport
    */
   private CommandReport createTimedOutReport() {
     CommandReport commandReport = new CommandReport();
     commandReport.setStatus(HostRoleStatus.TIMEDOUT.toString());
     commandReport.setStdErr("");
     commandReport.setStdOut("");
     return commandReport;
   }

   /**
    * A helper method to create CommandReports indicating the action/task has had an error
    *
    * @param message a String containing the error message to report
    * @return a new CommandReport
    */
   private CommandReport createErrorReport(String message) {
     CommandReport commandReport = new CommandReport();
     commandReport.setStatus(HostRoleStatus.FAILED.toString());
     commandReport.setExitCode(1);
     commandReport.setStdOut("Server action failed");
     commandReport.setStdErr((message == null) ? "Server action failed" : message);
     return commandReport;
   }

   /**
    * Stores the status of the task/action
    * <p/>
    * If the command report is not specified (null), an error report will be created.
    *
    * @param hostRoleCommand  the HostRoleCommand for the relevant task
    * @param executionCommand the ExecutionCommand for the relevant task
    * @param commandReport    the CommandReport to store
    */
   private void updateHostRoleState(HostRoleCommand hostRoleCommand, ExecutionCommand executionCommand,
                                    CommandReport commandReport) {
     if (commandReport == null) {
       commandReport = createErrorReport("Unknown error condition");
     }

     db.updateHostRoleState(null, hostRoleCommand.getRequestId(),
         hostRoleCommand.getStageId(), executionCommand.getRole(), commandReport);
   }

   /**
    * Determine what the timeout for this action/task should be.
    * <p/>
    * If the timeout value is not set in the command parameter map (under the key
    * ExecutionCommand.KeyNames.COMMAND_TIMEOUT or "command_timeout", the default timeout value will
    * be used.  It is expected that the timeout value stored in the command parameter map (if any) is
    * in seconds.
    *
    * @param executionCommand the ExecutionCommand for the relevant task
    * @return a long declaring the action/task's timeout
    */
   private long determineTimeout(ExecutionCommand executionCommand) {
     Map<String, String> params = executionCommand.getCommandParams();
     String paramsTimeout = (params == null) ? null : params.get(ExecutionCommand.KeyNames.COMMAND_TIMEOUT);
     Long timeout;

     try {
       timeout = (paramsTimeout == null)
           ? null
           : (Long.parseLong(paramsTimeout) * 1000L); // Convert seconds to milliseconds
     } catch (NumberFormatException e) {
       timeout = null;
     }

     // If a configured timeout value is not set for the command, attempt to get the configured
     // default
     if (timeout == null) {
       Integer defaultTimeoutSeconds = configuration.getDefaultServerTaskTimeout();
       if (defaultTimeoutSeconds != null) {
         timeout = defaultTimeoutSeconds * 1000L; // convert seconds to milliseconds
       }
     }

     // If a timeout value was not determined, return the hard-coded timeout value, else return the
     // determined timeout value.
     return (timeout == null)
         ? DEFAULT_EXECUTION_TIMEOUT_MS
         : ((timeout < 0) ? 0 : timeout);
   }

   /**
    * Execute the logic to handle each task in the queue in the order in which it was queued.
    * <p/>
    * A single task is executed at one time, allowing for a specified (ExecutionCommand.KeyNames.COMMAND_TIMEOUT)
    * or the default timeout for it to complete before considering the task timed out.
    *
    * @throws InterruptedException
    */
   public void doWork() throws InterruptedException {
     List<HostRoleCommand> tasks = db.getTasksByRoleAndStatus(Role.AMBARI_SERVER_ACTION.name(),
       HostRoleStatus.QUEUED);

     if ((tasks != null) && !tasks.isEmpty()) {
       for (HostRoleCommand task : tasks) {
         Long taskId = task.getTaskId();

         LOG.debug("Processing task #{}", taskId);

         if (task.getStatus() == HostRoleStatus.QUEUED) {
           ExecutionCommandWrapper executionWrapper = task.getExecutionCommandWrapper();

           if (executionWrapper != null) {
             ExecutionCommand executionCommand = executionWrapper.getExecutionCommand();

             if (executionCommand != null) {
               // For now, execute only one task at a time. This may change in the future in the
               // event it is discovered that this is a bottleneck. Since this implementation may
               // change, it should be noted from outside of this class, that there is no expectation
               // that tasks will be processed in order or serially.
               Worker worker = new Worker(task, executionCommand);
               Thread workerThread = new Thread(worker, String.format("Server Action Executor Worker %s", taskId));
               Long timeout = determineTimeout(executionCommand);

               updateHostRoleState(task, executionCommand, createInProgressReport());

               LOG.debug("Starting Server Action Executor Worker thread for task #{}.", taskId);
               workerThread.start();

               try {
                 workerThread.join(timeout);
               } catch (InterruptedException e) {
                 // Make sure the workerThread is interrupted as well.
                 workerThread.interrupt();
                 throw e;
               }

               if (workerThread.isAlive()) {
                 LOG.debug("Server Action Executor Worker thread for task #{} timed out - it failed to complete within {} ms.",
                     taskId, timeout);
                 workerThread.interrupt();
                 updateHostRoleState(task, executionCommand, createTimedOutReport());
               } else {
                 LOG.debug("Server Action Executor Worker thread for task #{} exited on its own.", taskId);
                 updateHostRoleState(task, executionCommand, worker.getCommandReport());
               }
             } else {
               LOG.warn("Task #{} failed to produce an ExecutionCommand, skipping.", taskId);
             }
           } else {
             LOG.warn("Task #{} failed to produce an ExecutionCommandWrapper, skipping.", taskId);
           }
         } else {
           LOG.warn("Queued task #{} is expected to have a status of {} but has a status of {}, skipping.",
               taskId, HostRoleStatus.QUEUED, task.getStatus());
         }
       }
     }

     cleanRequestShareDataContexts();
   }

   /**
    * Internal class to execute a unit of work in its own thread
    */
   private class Worker implements Runnable {
     /**
      * The task id of the relevant task
      */
     private final Long taskId;

     /**
      * The HostRoleCommand data used by this Worker to execute the task
      */
     private final HostRoleCommand hostRoleCommand;

     /**
      * The ExecutionCommand data used by this Worker to execute the task
      */
     private final ExecutionCommand executionCommand;

     /**
      * The resulting CommandReport used by the caller to update the status of the relevant task
      */
     private CommandReport commandReport = null;

     @Override
     public void run() {
       try {
         LOG.debug("Executing task #{}", taskId);

         // Set an internal administrator user to be the authenticated user in the event an
         // authorization check is needed while performing a server-side action.
         InternalAuthenticationToken authentication = new InternalAuthenticationToken("server_action_executor");
         authentication.setAuthenticated(true);
         SecurityContextHolder.getContext().setAuthentication(authentication);

         commandReport = execute(hostRoleCommand, executionCommand);

         LOG.debug("Task #{} completed execution with status of {}",
             taskId, (commandReport == null) ? "UNKNOWN" : commandReport.getStatus());
       } catch (Throwable t) {
         LOG.warn("Task #{} failed to complete execution due to thrown exception: {}:{}",
             taskId, t.getClass().getName(), t.getLocalizedMessage(), t);

         commandReport = createErrorReport(t.getLocalizedMessage());
       }
     }

     /**
      * Returns the resulting CommandReport
      *
      * @return a CommandReport
      */
     public CommandReport getCommandReport() {
       return commandReport;
     }

     /**
      * Attempts to execute the task specified using data from the supplied HostRoleCommand and
      * ExecutionCommand.
      * <p/>
      * Retrieves the role parameters from the supplied ExecutionCommand and queries it for the
      * "ACTON_NAME" property.  The returned String is expected to be the classname of a ServerAction
      * implementation.  If so, an instance of the implementation class is created and executed
      * yielding a CommandReport to (eventually) return back to the parent thread.
      *
      * @param hostRoleCommand  The HostRoleCommand the HostRoleCommand for the relevant task
      * @param executionCommand the ExecutionCommand for the relevant task
      * @return the resulting CommandReport
      * @throws AmbariException
      * @throws InterruptedException
      */
     private CommandReport execute(HostRoleCommand hostRoleCommand, ExecutionCommand executionCommand)
         throws AmbariException, InterruptedException {

       if (hostRoleCommand == null) {
         throw new AmbariException("Missing HostRoleCommand data");
       } else if (executionCommand == null) {
         throw new AmbariException("Missing ExecutionCommand data");
       } else {
         Map<String, String> roleParams = executionCommand.getRoleParams();

         if (roleParams == null) {
           throw new AmbariException("Missing RoleParams data");
         } else {
           String actionClassname = roleParams.get(ServerAction.ACTION_NAME);

           if (actionClassname == null) {
             throw new AmbariException("Missing action classname for server action");
           } else {
             ServerAction action = createServerAction(actionClassname);

             if (action == null) {
               throw new AmbariException("Failed to create server action: " + actionClassname);
             } else {
               // Set properties on the action:
               action.setExecutionCommand(executionCommand);
               action.setHostRoleCommand(hostRoleCommand);

               return action.execute(getRequestSharedDataContext(hostRoleCommand.getRequestId()));
             }
           }
         }
       }
     }

     /**
      * Attempts to create an instance of the ServerAction class implementation specified in
      * classname.
      *
      * @param classname a String declaring the classname of the ServerAction class to instantiate
      * @return the instantiated ServerAction implementation
      * @throws AmbariException
      */
     private ServerAction createServerAction(String classname) throws AmbariException {
       try {
         Class<?> actionClass = Class.forName(classname);

         if (actionClass == null) {
           throw new AmbariException("Unable to load server action class: " + classname);
         } else {
           Class<? extends ServerAction> serverActionClass = actionClass.asSubclass(ServerAction.class);

           if (serverActionClass == null) {
             throw new AmbariException("Unable to execute server action class, invalid type: " + classname);
           } else {
             return injector.getInstance(serverActionClass);
           }
         }
       } catch (ClassNotFoundException e) {
         throw new AmbariException("Unable to load server action class: " + classname, e);
       }
     }

     /**
      * Constructs a new Worker used to execute a task
      *
      * @param hostRoleCommand  the HostRoleCommand for the relevant task
      * @param executionCommand the ExecutionCommand for the relevant task
      */
     private Worker(HostRoleCommand hostRoleCommand, ExecutionCommand executionCommand) {
       taskId = hostRoleCommand.getTaskId();
       this.hostRoleCommand = hostRoleCommand;
       this.executionCommand = executionCommand;
     }
   }
 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/

	package org.apache.ambari.server.serveraction;

	import java.util.HashMap;
	import java.util.List;
	import java.util.Map;
	import java.util.Set;
	import java.util.concurrent.ConcurrentHashMap;
	import java.util.concurrent.ConcurrentMap;

	import org.apache.ambari.server.AmbariException;
	import org.apache.ambari.server.Role;
	import org.apache.ambari.server.StaticallyInject;
	import org.apache.ambari.server.actionmanager.ActionDBAccessor;
	import org.apache.ambari.server.actionmanager.ExecutionCommandWrapper;
	import org.apache.ambari.server.actionmanager.HostRoleCommand;
	import org.apache.ambari.server.actionmanager.HostRoleStatus;
	import org.apache.ambari.server.actionmanager.Request;
	import org.apache.ambari.server.agent.CommandReport;
	import org.apache.ambari.server.agent.ExecutionCommand;
	import org.apache.ambari.server.configuration.Configuration;
	import org.apache.ambari.server.controller.internal.CalculatedStatus;
	import org.apache.ambari.server.security.authorization.internal.InternalAuthenticationToken;
	import org.apache.ambari.server.utils.StageUtils;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	import com.google.inject.Inject;
	import com.google.inject.Injector;
	import org.springframework.security.core.context.SecurityContextHolder;

	/**
	* Server Action Executor used to execute server-side actions (or tasks)
	* <p/>
	* The ServerActionExecutor executes in its own thread, polling for AMBARI_SERVER_ACTION
	* HostRoleCommands queued for execution. It is expected that this thread is managed by the
	* ActionScheduler such that it is started when the ActionScheduler is started and stopped when the
	* ActionScheduler is stopped.
	*/
	@StaticallyInject
	public class ServerActionExecutor {

	private final static Logger LOG = LoggerFactory.getLogger(ServerActionExecutor.class);
	private final static Long DEFAULT_EXECUTION_TIMEOUT_MS = 1000L * 60 * 5;
	private final static Long POLLING_TIMEOUT_MS = 1000L * 5;

	/**
	* An injector to use to inject objects into ServerAction instances.
	*/
	@Inject
	private static Injector injector;

	/**
	* The Ambari configuration used to obtain configured details such as the default server-side action
	* timeout.
	*/
	@Inject
	private static Configuration configuration;

	/**
	* Maps request IDs to "blackboards" of shared data.
	* <p/>
	* This map is not synchronized, so any access to it should synchronize on
	* requestSharedDataMap object
	*/
	private final Map<Long, ConcurrentMap<String, Object>> requestSharedDataMap =
	new HashMap<Long, ConcurrentMap<String, Object>>();

	/**
	* The hostname of the (Ambari) server.
	* <p/>
	* This hostname is cached so that cycles are spent querying for it more than once.
	*/
	private final String serverHostName;

	/**
	* Database accessor to query and update the database of action commands.
	*/
	private final ActionDBAccessor db;

	/**
	* Internal locking object used to manage access to activeAwakeRequest.
	*/
	private final Object wakeupSyncObject = new Object();

	/**
	* Timeout (in milliseconds) used to throttle polling of database for new action commands.
	*/
	private final long sleepTimeMS;

	/**
	* Flag used to help keep thing moving in the event an "awake" request was encountered while busy
	* handing an action.
	*/
	private boolean activeAwakeRequest = false;

	/**
	* A reference to the Thread handling the work for this ServerActionExecutor
	*/
	private Thread executorThread = null;

	/**
	* Statically initialize the Injector
	* <p/>
	* This should only be used for unit tests.
	*
	* @param injector the Injector to (manually) statically inject
	*/
	public static void init(Injector injector) {
	ServerActionExecutor.injector = injector;
	}

	/**
	* Creates a new ServerActionExecutor
	*
	* @param db the ActionDBAccessor to use to read and update tasks
	* @param sleepTimeMS the time (in milliseconds) to wait between polling the database for more tasks
	*/
	public ServerActionExecutor(ActionDBAccessor db, long sleepTimeMS) {
	serverHostName = StageUtils.getHostName();
	this.db = db;
	this.sleepTimeMS = (sleepTimeMS < 1) ? POLLING_TIMEOUT_MS : sleepTimeMS;
	}

	/**
	* Starts this ServerActionExecutor's main thread.
	*/
	public void start() {
	LOG.info("Starting Server Action Executor thread...");
	executorThread = new Thread(new Runnable() {

	@Override
	public void run() {
	while (!Thread.interrupted()) {
	try {
	synchronized (wakeupSyncObject) {
	if (!activeAwakeRequest) {
	wakeupSyncObject.wait(sleepTimeMS);
	}
	activeAwakeRequest = false;
	}

	doWork();
	} catch (InterruptedException e) {
	LOG.warn("Server Action Executor thread interrupted, starting to shutdown...");
	break;
	}
	}

	LOG.info("Server Action Executor thread shutting down...");
	}
	}, "Server Action Executor");
	executorThread.start();

	if (executorThread.isAlive()) {
	LOG.info("Server Action Executor thread started.");
	}
	}

	/**
	* Attempts to stop this ServerActionExecutor's main thread.
	*/
	public void stop() {
	LOG.info("Stopping Server Action Executor thread...");

	if (executorThread != null) {
	executorThread.interrupt();

	// Wait for about 60 seconds for the thread to stop
	for (int i = 0; i < 120; i++) {
	try {
	executorThread.join(500);
	} catch (InterruptedException e) {
	// Ignore this...
	}

	if (!executorThread.isAlive()) {
	break;
	}
	}

	if (!executorThread.isAlive()) {
	executorThread = null;
	}
	}

	if (executorThread == null) {
	LOG.info("Server Action Executor thread stopped.");
	} else {
	LOG.warn("Server Action Executor thread hasn't stopped, giving up waiting.");
	}
	}

	/**
	* Attempts to force this ServerActionExecutor to wake up and do work.
	* <p/>
	* Should be called from another thread when we want scheduler to
	* make a run ASAP (for example, to process desired configs of SCHs).
	* The method is guaranteed to return quickly.
	*/
	public void awake() {
	synchronized (wakeupSyncObject) {
	activeAwakeRequest = true;
	wakeupSyncObject.notify();
	}
	}

	/**
	* Returns a Map to be used to share data among server actions within a given request context.
	*
	* @param requestId a long identifying the id of the relevant request
	* @return a ConcurrentMap of "shared" data
	*/
	private ConcurrentMap<String, Object> getRequestSharedDataContext(long requestId) {
	synchronized (requestSharedDataMap) {
	ConcurrentMap<String, Object> map = requestSharedDataMap.get(requestId);

	if (map == null) {
	map = new ConcurrentHashMap<String, Object>();
	requestSharedDataMap.put(requestId, map);
	}

	return map;
	}
	}

	/**
	* Cleans up orphaned shared data Maps due to completed or failed request
	* contexts. We are unable to use {@link Request#getStatus()} since this field
	* is not populated in the database but, instead, calculated in realtime.
	*/
	private void cleanRequestShareDataContexts() {
	// if the cache is empty, do nothing
	if (requestSharedDataMap.isEmpty()) {
	return;
	}

	try {
	// for every item in the map, get the request and check its status
	synchronized (requestSharedDataMap) {
	Set<Long> requestIds = requestSharedDataMap.keySet();
	List<Request> requests = db.getRequests(requestIds);
	for (Request request : requests) {
	// calcuate the status from the stages and then remove from the map if
	// necessary
	CalculatedStatus calculatedStatus = CalculatedStatus.statusFromStages(
	request.getStages());

	// calcuate the status of the request
	HostRoleStatus status = calculatedStatus.getStatus();

	// remove the request from the map if the request is COMPLETED or
	// FAILED
	switch (status) {
	case FAILED:
	case COMPLETED:
	requestSharedDataMap.remove(request.getRequestId());
	break;
	default:
	break;
	}
	}
	}
	} catch (Exception exception) {
	LOG.warn("Unable to clear the server-side action request cache", exception);
	}
	}

	/**
	* A helper method to create CommandReports indicating the action/task is in progress
	*
	* @return a new CommandReport
	*/
	private CommandReport createInProgressReport() {
	CommandReport commandReport = new CommandReport();
	commandReport.setStatus(HostRoleStatus.IN_PROGRESS.toString());
	commandReport.setStdErr("");
	commandReport.setStdOut("");
	return commandReport;
	}

	/**
	* A helper method to create CommandReports indicating the action/task had timed out
	*
	* @return a new CommandReport
	*/
	private CommandReport createTimedOutReport() {
	CommandReport commandReport = new CommandReport();
	commandReport.setStatus(HostRoleStatus.TIMEDOUT.toString());
	commandReport.setStdErr("");
	commandReport.setStdOut("");
	return commandReport;
	}

	/**
	* A helper method to create CommandReports indicating the action/task has had an error
	*
	* @param message a String containing the error message to report
	* @return a new CommandReport
	*/
	private CommandReport createErrorReport(String message) {
	CommandReport commandReport = new CommandReport();
	commandReport.setStatus(HostRoleStatus.FAILED.toString());
	commandReport.setExitCode(1);
	commandReport.setStdOut("Server action failed");
	commandReport.setStdErr((message == null) ? "Server action failed" : message);
	return commandReport;
	}

	/**
	* Stores the status of the task/action
	* <p/>
	* If the command report is not specified (null), an error report will be created.
	*
	* @param hostRoleCommand the HostRoleCommand for the relevant task
	* @param executionCommand the ExecutionCommand for the relevant task
	* @param commandReport the CommandReport to store
	*/
	private void updateHostRoleState(HostRoleCommand hostRoleCommand, ExecutionCommand executionCommand,
	CommandReport commandReport) {
	if (commandReport == null) {
	commandReport = createErrorReport("Unknown error condition");
	}

	db.updateHostRoleState(null, hostRoleCommand.getRequestId(),
	hostRoleCommand.getStageId(), executionCommand.getRole(), commandReport);
	}

	/**
	* Determine what the timeout for this action/task should be.
	* <p/>
	* If the timeout value is not set in the command parameter map (under the key
	* ExecutionCommand.KeyNames.COMMAND_TIMEOUT or "command_timeout", the default timeout value will
	* be used. It is expected that the timeout value stored in the command parameter map (if any) is
	* in seconds.
	*
	* @param executionCommand the ExecutionCommand for the relevant task
	* @return a long declaring the action/task's timeout
	*/
	private long determineTimeout(ExecutionCommand executionCommand) {
	Map<String, String> params = executionCommand.getCommandParams();
	String paramsTimeout = (params == null) ? null : params.get(ExecutionCommand.KeyNames.COMMAND_TIMEOUT);
	Long timeout;

	try {
	timeout = (paramsTimeout == null)
	? null
	: (Long.parseLong(paramsTimeout) * 1000L); // Convert seconds to milliseconds
	} catch (NumberFormatException e) {
	timeout = null;
	}

	// If a configured timeout value is not set for the command, attempt to get the configured
	// default
	if (timeout == null) {
	Integer defaultTimeoutSeconds = configuration.getDefaultServerTaskTimeout();
	if (defaultTimeoutSeconds != null) {
	timeout = defaultTimeoutSeconds * 1000L; // convert seconds to milliseconds
	}
	}

	// If a timeout value was not determined, return the hard-coded timeout value, else return the
	// determined timeout value.
	return (timeout == null)
	? DEFAULT_EXECUTION_TIMEOUT_MS
	: ((timeout < 0) ? 0 : timeout);
	}

	/**
	* Execute the logic to handle each task in the queue in the order in which it was queued.
	* <p/>
	* A single task is executed at one time, allowing for a specified (ExecutionCommand.KeyNames.COMMAND_TIMEOUT)
	* or the default timeout for it to complete before considering the task timed out.
	*
	* @throws InterruptedException
	*/
	public void doWork() throws InterruptedException {
	List<HostRoleCommand> tasks = db.getTasksByRoleAndStatus(Role.AMBARI_SERVER_ACTION.name(),
	HostRoleStatus.QUEUED);

	if ((tasks != null) && !tasks.isEmpty()) {
	for (HostRoleCommand task : tasks) {
	Long taskId = task.getTaskId();

	LOG.debug("Processing task #{}", taskId);

	if (task.getStatus() == HostRoleStatus.QUEUED) {
	ExecutionCommandWrapper executionWrapper = task.getExecutionCommandWrapper();

	if (executionWrapper != null) {
	ExecutionCommand executionCommand = executionWrapper.getExecutionCommand();

	if (executionCommand != null) {
	// For now, execute only one task at a time. This may change in the future in the
	// event it is discovered that this is a bottleneck. Since this implementation may
	// change, it should be noted from outside of this class, that there is no expectation
	// that tasks will be processed in order or serially.
	Worker worker = new Worker(task, executionCommand);
	Thread workerThread = new Thread(worker, String.format("Server Action Executor Worker %s", taskId));
	Long timeout = determineTimeout(executionCommand);

	updateHostRoleState(task, executionCommand, createInProgressReport());

	LOG.debug("Starting Server Action Executor Worker thread for task #{}.", taskId);
	workerThread.start();

	try {
	workerThread.join(timeout);
	} catch (InterruptedException e) {
	// Make sure the workerThread is interrupted as well.
	workerThread.interrupt();
	throw e;
	}

	if (workerThread.isAlive()) {
	LOG.debug("Server Action Executor Worker thread for task #{} timed out - it failed to complete within {} ms.",
	taskId, timeout);
	workerThread.interrupt();
	updateHostRoleState(task, executionCommand, createTimedOutReport());
	} else {
	LOG.debug("Server Action Executor Worker thread for task #{} exited on its own.", taskId);
	updateHostRoleState(task, executionCommand, worker.getCommandReport());
	}
	} else {
	LOG.warn("Task #{} failed to produce an ExecutionCommand, skipping.", taskId);
	}
	} else {
	LOG.warn("Task #{} failed to produce an ExecutionCommandWrapper, skipping.", taskId);
	}
	} else {
	LOG.warn("Queued task #{} is expected to have a status of {} but has a status of {}, skipping.",
	taskId, HostRoleStatus.QUEUED, task.getStatus());
	}
	}
	}

	cleanRequestShareDataContexts();
	}

	/**
	* Internal class to execute a unit of work in its own thread
	*/
	private class Worker implements Runnable {
	/**
	* The task id of the relevant task
	*/
	private final Long taskId;

	/**
	* The HostRoleCommand data used by this Worker to execute the task
	*/
	private final HostRoleCommand hostRoleCommand;

	/**
	* The ExecutionCommand data used by this Worker to execute the task
	*/
	private final ExecutionCommand executionCommand;

	/**
	* The resulting CommandReport used by the caller to update the status of the relevant task
	*/
	private CommandReport commandReport = null;

	@Override
	public void run() {
	try {
	LOG.debug("Executing task #{}", taskId);

	// Set an internal administrator user to be the authenticated user in the event an
	// authorization check is needed while performing a server-side action.
	InternalAuthenticationToken authentication = new InternalAuthenticationToken("server_action_executor");
	authentication.setAuthenticated(true);
	SecurityContextHolder.getContext().setAuthentication(authentication);

	commandReport = execute(hostRoleCommand, executionCommand);

	LOG.debug("Task #{} completed execution with status of {}",
	taskId, (commandReport == null) ? "UNKNOWN" : commandReport.getStatus());
	} catch (Throwable t) {
	LOG.warn("Task #{} failed to complete execution due to thrown exception: {}:{}",
	taskId, t.getClass().getName(), t.getLocalizedMessage(), t);

	commandReport = createErrorReport(t.getLocalizedMessage());
	}
	}

	/**
	* Returns the resulting CommandReport
	*
	* @return a CommandReport
	*/
	public CommandReport getCommandReport() {
	return commandReport;
	}

	/**
	* Attempts to execute the task specified using data from the supplied HostRoleCommand and
	* ExecutionCommand.
	* <p/>
	* Retrieves the role parameters from the supplied ExecutionCommand and queries it for the
	* "ACTON_NAME" property. The returned String is expected to be the classname of a ServerAction
	* implementation. If so, an instance of the implementation class is created and executed
	* yielding a CommandReport to (eventually) return back to the parent thread.
	*
	* @param hostRoleCommand The HostRoleCommand the HostRoleCommand for the relevant task
	* @param executionCommand the ExecutionCommand for the relevant task
	* @return the resulting CommandReport
	* @throws AmbariException
	* @throws InterruptedException
	*/
	private CommandReport execute(HostRoleCommand hostRoleCommand, ExecutionCommand executionCommand)
	throws AmbariException, InterruptedException {

	if (hostRoleCommand == null) {
	throw new AmbariException("Missing HostRoleCommand data");
	} else if (executionCommand == null) {
	throw new AmbariException("Missing ExecutionCommand data");
	} else {
	Map<String, String> roleParams = executionCommand.getRoleParams();

	if (roleParams == null) {
	throw new AmbariException("Missing RoleParams data");
	} else {
	String actionClassname = roleParams.get(ServerAction.ACTION_NAME);

	if (actionClassname == null) {
	throw new AmbariException("Missing action classname for server action");
	} else {
	ServerAction action = createServerAction(actionClassname);

	if (action == null) {
	throw new AmbariException("Failed to create server action: " + actionClassname);
	} else {
	// Set properties on the action:
	action.setExecutionCommand(executionCommand);
	action.setHostRoleCommand(hostRoleCommand);

	return action.execute(getRequestSharedDataContext(hostRoleCommand.getRequestId()));
	}
	}
	}
	}
	}

	/**
	* Attempts to create an instance of the ServerAction class implementation specified in
	* classname.
	*
	* @param classname a String declaring the classname of the ServerAction class to instantiate
	* @return the instantiated ServerAction implementation
	* @throws AmbariException
	*/
	private ServerAction createServerAction(String classname) throws AmbariException {
	try {
	Class<?> actionClass = Class.forName(classname);

	if (actionClass == null) {
	throw new AmbariException("Unable to load server action class: " + classname);
	} else {
	Class<? extends ServerAction> serverActionClass = actionClass.asSubclass(ServerAction.class);

	if (serverActionClass == null) {
	throw new AmbariException("Unable to execute server action class, invalid type: " + classname);
	} else {
	return injector.getInstance(serverActionClass);
	}
	}
	} catch (ClassNotFoundException e) {
	throw new AmbariException("Unable to load server action class: " + classname, e);
	}
	}

	/**
	* Constructs a new Worker used to execute a task
	*
	* @param hostRoleCommand the HostRoleCommand for the relevant task
	* @param executionCommand the ExecutionCommand for the relevant task
	*/
	private Worker(HostRoleCommand hostRoleCommand, ExecutionCommand executionCommand) {
	taskId = hostRoleCommand.getTaskId();
	this.hostRoleCommand = hostRoleCommand;
	this.executionCommand = executionCommand;
	}
	}
	}