/*
 * 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.slider.server.appmaster.state;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.yarn.api.records.Container;
import org.apache.hadoop.yarn.api.records.ContainerId;
import org.apache.hadoop.yarn.api.records.ContainerStatus;
import org.apache.hadoop.yarn.api.records.FinalApplicationStatus;
import org.apache.hadoop.yarn.api.records.NodeId;
import org.apache.hadoop.yarn.api.records.NodeReport;
import org.apache.hadoop.yarn.api.records.Priority;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.api.records.impl.pb.ContainerPBImpl;
import org.apache.hadoop.yarn.client.api.AMRMClient;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.exceptions.YarnRuntimeException;
import org.apache.slider.api.ClusterDescription;
import org.apache.slider.api.ClusterDescriptionKeys;
import org.apache.slider.api.ClusterDescriptionOperations;
import org.apache.slider.api.ClusterNode;
import org.apache.slider.api.InternalKeys;
import org.apache.slider.api.ResourceKeys;
import org.apache.slider.api.RoleKeys;
import org.apache.slider.api.StatusKeys;
import org.apache.slider.api.proto.Messages;
import org.apache.slider.common.SliderExitCodes;
import org.apache.slider.common.SliderKeys;
import org.apache.slider.common.tools.ConfigHelper;
import org.apache.slider.common.tools.SliderUtils;
import org.apache.slider.core.conf.AggregateConf;
import org.apache.slider.core.conf.ConfTree;
import org.apache.slider.core.conf.ConfTreeOperations;
import org.apache.slider.core.conf.MapOperations;
import org.apache.slider.core.exceptions.BadClusterStateException;
import org.apache.slider.core.exceptions.BadConfigException;
import org.apache.slider.core.exceptions.ErrorStrings;
import org.apache.slider.core.exceptions.NoSuchNodeException;
import org.apache.slider.core.exceptions.SliderInternalStateException;
import org.apache.slider.core.exceptions.TriggerClusterTeardownException;
import org.apache.slider.providers.ProviderRole;
import org.apache.slider.server.appmaster.operations.AbstractRMOperation;
import org.apache.slider.server.appmaster.operations.CancelRequestOperation;
import org.apache.slider.server.appmaster.operations.ContainerReleaseOperation;
import org.apache.slider.server.appmaster.operations.ContainerRequestOperation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.ListIterator;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicInteger;

import static org.apache.slider.api.ResourceKeys.*;
import static org.apache.slider.api.RoleKeys.*;


/**
 * The model of all the ongoing state of a Slider AM.
 *
 * concurrency rules: any method which begins with <i>build</i>
 * is not synchronized and intended to be used during
 * initialization.
 */
public class AppState {
  protected static final Logger log =
    LoggerFactory.getLogger(AppState.class);
  
  private final AbstractRecordFactory recordFactory;

  /**
   * Flag set to indicate the application is live -this only happens
   * after the buildInstance operation
   */
  private boolean applicationLive = false;

  /**
   * The definition of the instance. Flexing updates the resources section
   This is used as a synchronization point on activities that update
   the CD, and also to update some of the structures that
   feed in to the CD
   */
  private AggregateConf instanceDefinition;
  
  private long snapshotTime;
  private AggregateConf instanceDefinitionSnapshot;

  /**
   * snapshot of resources as of last update time
   */
  private ConfTreeOperations resourcesSnapshot;
  private ConfTreeOperations appConfSnapshot;
  private ConfTreeOperations internalsSnapshot;
  
  /**
   * This is the status, the live model
   */
  private ClusterDescription clusterStatus = new ClusterDescription();

  /**
   * Metadata provided by the AM for use in filling in status requests
   */
  private Map<String, String> applicationInfo;
  
  /**
   * Client properties created via the provider -static for the life
   * of the application
   */
  private Map<String, String> clientProperties = new HashMap<String, String>();

  /**
   The cluster description published to callers
   This is used as a synchronization point on activities that update
   the CD, and also to update some of the structures that
   feed in to the CD
   */
  private ClusterDescription clusterSpec = new ClusterDescription();

  private final Map<Integer, RoleStatus> roleStatusMap =
    new ConcurrentHashMap<Integer, RoleStatus>();

  private final Map<String, ProviderRole> roles =
    new ConcurrentHashMap<String, ProviderRole>();

  private final Map<Integer, ProviderRole> rolePriorityMap = 
    new ConcurrentHashMap<Integer, ProviderRole>();

  /**
   * The master node.
   */
  private RoleInstance appMasterNode;

  /**
   * Hash map of the containers we have. This includes things that have
   * been allocated but are not live; it is a superset of the live list
   */
  private final ConcurrentMap<ContainerId, RoleInstance> ownedContainers =
    new ConcurrentHashMap<ContainerId, RoleInstance>();

  /**
   * Hash map of the containers we have released, but we
   * are still awaiting acknowledgements on. Any failure of these
   * containers is treated as a successful outcome
   */
  private final ConcurrentMap<ContainerId, Container> containersBeingReleased =
    new ConcurrentHashMap<ContainerId, Container>();
  
  /**
   * Counter for completed containers ( complete denotes successful or failed )
   */
  private final AtomicInteger completedContainerCount = new AtomicInteger();

  /**
   *   Count of failed containers

   */
  private final AtomicInteger failedContainerCount = new AtomicInteger();

  /**
   * # of started containers
   */
  private final AtomicInteger startedContainers = new AtomicInteger();

  /**
   * # of containers that failed to start 
   */
  private final AtomicInteger startFailedContainers = new AtomicInteger();

  /**
   * Track the number of surplus containers received and discarded
   */
  private final AtomicInteger surplusContainers = new AtomicInteger();


  /**
   * Map of requested nodes. This records the command used to start it,
   * resources, etc. When container started callback is received,
   * the node is promoted from here to the containerMap
   */
  private final Map<ContainerId, RoleInstance> startingNodes =
    new ConcurrentHashMap<ContainerId, RoleInstance>();

  /**
   * List of completed nodes. This isn't kept in the CD as it gets too
   * big for the RPC responses. Indeed, we should think about how deep to get this
   */
  private final Map<ContainerId, RoleInstance> completedNodes
    = new ConcurrentHashMap<ContainerId, RoleInstance>();

  /**
   * Nodes that failed to start.
   * Again, kept out of the CD
   */
  private final Map<ContainerId, RoleInstance> failedNodes =
    new ConcurrentHashMap<ContainerId, RoleInstance>();

  /**
   * Nodes that came assigned to a role above that
   * which were asked for -this appears to happen
   */
  private final Set<ContainerId> surplusNodes = new HashSet<ContainerId>();

  /**
   * Map of containerID -> cluster nodes, for status reports.
   * Access to this should be synchronized on the clusterDescription
   */
  private final Map<ContainerId, RoleInstance> liveNodes =
    new ConcurrentHashMap<ContainerId, RoleInstance>();
  private final AtomicInteger completionOfNodeNotInLiveListEvent =
    new AtomicInteger();
  private final AtomicInteger completionOfUnknownContainerEvent =
    new AtomicInteger();


  /**
   * Record of the max no. of cores allowed in this cluster
   */
  private int containerMaxCores;

  /**
   * limit container memory
   */
  private int containerMaxMemory;
  
  private RoleHistory roleHistory;
  private Configuration publishedProviderConf;
  private long startTimeThreshold;
  
  private int failureThreshold = 10;
  
  private String logServerURL = "";
  
  private ContainerReleaseSelector containerReleaseSelector;

  public AppState(AbstractRecordFactory recordFactory) {
    this.recordFactory = recordFactory;
  }

  public int getFailedCountainerCount() {
    return failedContainerCount.get();
  }

  /**
   * Increment the count and return the new value
   * @return the latest failed container count
   */
  public int incFailedCountainerCount() {
    return failedContainerCount.incrementAndGet();
  }

  public int getStartFailedCountainerCount() {
    return startFailedContainers.get();
  }

  /**
   * Increment the count and return the new value
   * @return the latest failed container count
   */
  public int incStartedCountainerCount() {
    return startedContainers.incrementAndGet();
  }

  public int getStartedCountainerCount() {
    return startedContainers.get();
  }

  /**
   * Increment the count and return the new value
   * @return the latest failed container count
   */
  public int incStartFailedCountainerCount() {
    return startFailedContainers.incrementAndGet();
  }

  
  public AtomicInteger getStartFailedContainers() {
    return startFailedContainers;
  }

  public AtomicInteger getCompletionOfNodeNotInLiveListEvent() {
    return completionOfNodeNotInLiveListEvent;
  }

  public AtomicInteger getCompletionOfUnknownContainerEvent() {
    return completionOfUnknownContainerEvent;
  }


  public Map<Integer, RoleStatus> getRoleStatusMap() {
    return roleStatusMap;
  }
  
  protected Map<String, ProviderRole> getRoleMap() {
    return roles;
  }

  public Map<Integer, ProviderRole> getRolePriorityMap() {
    return rolePriorityMap;
  }

  private Map<ContainerId, RoleInstance> getStartingNodes() {
    return startingNodes;
  }

  private Map<ContainerId, RoleInstance> getCompletedNodes() {
    return completedNodes;
  }


  public Map<ContainerId, RoleInstance> getFailedNodes() {
    return failedNodes;
  }


  public Map<ContainerId, RoleInstance> getLiveNodes() {
    return liveNodes;
  }

  /**
   * Get the desired cluster state
   * @return the specification of the cluter
   */
  public ClusterDescription getClusterSpec() {
    return clusterSpec;
  }


  public ClusterDescription getClusterStatus() {
    return clusterStatus;
  }

  @VisibleForTesting
  protected void setClusterStatus(ClusterDescription clusterDesc) {
    this.clusterStatus = clusterDesc;
  }

  /**
   * Set the instance definition -this also builds the (now obsolete)
   * cluster specification from it.
   * 
   * Important: this is for early binding and must not be used after the build
   * operation is complete. 
   * @param definition
   * @throws BadConfigException
   */
  public synchronized void updateInstanceDefinition(AggregateConf definition) throws
                                                                              BadConfigException,
                                                                              IOException {
    this.instanceDefinition = definition;
    onInstanceDefinitionUpdated();
  }

  public synchronized AggregateConf getInstanceDefinition() {
    return instanceDefinition;
  }

  /**
   * Get the role history of the application
   * @return the role history
   */
  @VisibleForTesting
  public RoleHistory getRoleHistory() {
    return roleHistory;
  }

  /**
   * Get the path used for history files
   * @return the directory used for history files
   */
  @VisibleForTesting
  public Path getHistoryPath() {
    return roleHistory.getHistoryPath();
  }

  /**
   * Set the container limits -the max that can be asked for,
   * which are used when the "max" values are requested
   * @param maxMemory maximum memory
   * @param maxCores maximum cores
   */
  public void setContainerLimits(int maxMemory, int maxCores) {
    containerMaxCores = maxCores;
    containerMaxMemory = maxMemory;
  }


  public ConfTreeOperations getResourcesSnapshot() {
    return resourcesSnapshot;
  }


  public ConfTreeOperations getAppConfSnapshot() {
    return appConfSnapshot;
  }


  public ConfTreeOperations getInternalsSnapshot() {
    return internalsSnapshot;
  }


  public boolean isApplicationLive() {
    return applicationLive;
  }


  public long getSnapshotTime() {
    return snapshotTime;
  }


  public AggregateConf getInstanceDefinitionSnapshot() {
    return instanceDefinitionSnapshot;
  }

  /**
   * Build up the application state
   * @param instanceDefinition definition of the applicatin instance
   * @param appmasterConfig
   * @param publishedProviderConf any configuration info to be published by a provider
   * @param providerRoles roles offered by a provider
   * @param fs filesystem
   * @param historyDir directory containing history files
   * @param liveContainers list of live containers supplied on an AM restart
   * @param applicationInfo
   * @param releaseSelector
   */
  public synchronized void buildInstance(AggregateConf instanceDefinition,
      Configuration appmasterConfig,
      Configuration publishedProviderConf,
      List<ProviderRole> providerRoles,
      FileSystem fs,
      Path historyDir,
      List<Container> liveContainers,
      Map<String, String> applicationInfo,
      SimpleReleaseSelector releaseSelector)
      throws  BadClusterStateException, BadConfigException, IOException {
    Preconditions.checkArgument(instanceDefinition != null);
    Preconditions.checkArgument(releaseSelector != null);

    this.publishedProviderConf = publishedProviderConf;
    this.applicationInfo = applicationInfo != null ? applicationInfo
                                                   : new HashMap<String, String>();

    clientProperties = new HashMap<String, String>();
    containerReleaseSelector = releaseSelector;


    Set<String> confKeys = ConfigHelper.sortedConfigKeys(publishedProviderConf);

    //  Add the -site configuration properties
    for (String key : confKeys) {
      String val = publishedProviderConf.get(key);
      clientProperties.put(key, val);
    }

    // set the cluster specification (once its dependency the client properties
    // is out the way
    updateInstanceDefinition(instanceDefinition);

    //build the initial role list
    for (ProviderRole providerRole : providerRoles) {
      buildRole(providerRole);
    }

    ConfTreeOperations resources =
        instanceDefinition.getResourceOperations();

    Set<String> roleNames = resources.getComponentNames();
    for (String name : roleNames) {
      if (!roles.containsKey(name)) {
        // this is a new value
        log.info("Adding role {}", name);
        MapOperations resComponent = resources.getComponent(name);
        ProviderRole dynamicRole =
            createDynamicProviderRole(name, resComponent);
        buildRole(dynamicRole);
        providerRoles.add(dynamicRole);
      }
    }
    //then pick up the requirements
    buildRoleRequirementsFromResources();


    //set the livespan
    MapOperations globalResOpts =
        instanceDefinition.getResourceOperations().getGlobalOptions();
    
    startTimeThreshold = globalResOpts.getOptionInt(
        InternalKeys.INTERNAL_CONTAINER_FAILURE_SHORTLIFE,
        InternalKeys.DEFAULT_INTERNAL_CONTAINER_FAILURE_SHORTLIFE);

    failureThreshold = globalResOpts.getOptionInt(
        ResourceKeys.CONTAINER_FAILURE_THRESHOLD,
        ResourceKeys.DEFAULT_CONTAINER_FAILURE_THRESHOLD);
    initClusterStatus();


    // add the roles
    roleHistory = new RoleHistory(providerRoles);
    roleHistory.onStart(fs, historyDir);

    //rebuild any live containers
    rebuildModelFromRestart(liveContainers);

    // any am config options to pick up

    logServerURL = appmasterConfig.get(YarnConfiguration.YARN_LOG_SERVER_URL, "");
    //mark as live
    applicationLive = true;
  }

  public void initClusterStatus() {
    //copy into cluster status. 
    ClusterDescription status = ClusterDescription.copy(clusterSpec);
    status.state = ClusterDescription.STATE_CREATED;
    MapOperations infoOps = new MapOperations("info", status.info);
    infoOps.mergeWithoutOverwrite(applicationInfo);
    SliderUtils.addBuildInfo(infoOps, "status");

    long now = now();
    status.setInfoTime(StatusKeys.INFO_LIVE_TIME_HUMAN,
                              StatusKeys.INFO_LIVE_TIME_MILLIS,
                              now);
    SliderUtils.setInfoTime(infoOps,
        StatusKeys.INFO_LIVE_TIME_HUMAN,
        StatusKeys.INFO_LIVE_TIME_MILLIS,
        now);
    if (0 == status.createTime) {
      status.createTime = now;
      SliderUtils.setInfoTime(infoOps,
          StatusKeys.INFO_CREATE_TIME_HUMAN,
          StatusKeys.INFO_CREATE_TIME_MILLIS,
          now);
    }
    status.state = ClusterDescription.STATE_LIVE;

      //set the app state to this status
    setClusterStatus(status);
  }

  /**
   * Build a dynamic provider role
   * @param name name of role
   * @return a new provider role
   * @throws BadConfigException bad configuration
   */
  @VisibleForTesting
  public ProviderRole createDynamicProviderRole(String name,
                                                MapOperations component) throws
                                                        BadConfigException {
    String priOpt = component.getMandatoryOption(ResourceKeys.COMPONENT_PRIORITY);
    int priority = SliderUtils.parseAndValidate("value of " + name + " " +
        ResourceKeys.COMPONENT_PRIORITY,
        priOpt, 0, 1, -1);
    log.info("Role {} assigned priority {}", name, priority);
    String placementOpt = component.getOption(
      ResourceKeys.COMPONENT_PLACEMENT_POLICY, "0");
    int placement = SliderUtils.parseAndValidate("value of " + name + " " +
        ResourceKeys.COMPONENT_PLACEMENT_POLICY,
        placementOpt, 0, 0, -1);
    return new ProviderRole(name, priority, placement);
  }


  /**
   * Actions to perform when an instance definition is updated
   * Currently: resolve the configuration
   *  updated the cluster spec derivative
   * @throws BadConfigException
   */
  private synchronized void onInstanceDefinitionUpdated() throws
                                                          BadConfigException,
                                                          IOException {
    instanceDefinition.resolve();

    // force in the AM desired state values
    ConfTreeOperations resources =
        instanceDefinition.getResourceOperations();
    if (resources.getComponent(SliderKeys.COMPONENT_AM) != null) {
      resources.setComponentOpt(
          SliderKeys.COMPONENT_AM, ResourceKeys.COMPONENT_INSTANCES, "1");
    }

    //note the time 
    snapshotTime = now();
    //snapshot all three sectons
    resourcesSnapshot =
      ConfTreeOperations.fromInstance(instanceDefinition.getResources());
    appConfSnapshot =
      ConfTreeOperations.fromInstance(instanceDefinition.getAppConf());
    internalsSnapshot =
      ConfTreeOperations.fromInstance(instanceDefinition.getInternal());
    //build a new aggregate from the snapshots
    instanceDefinitionSnapshot = new AggregateConf(resourcesSnapshot.confTree,
                                                   appConfSnapshot.confTree,
                                                   internalsSnapshot.confTree);
    instanceDefinitionSnapshot.setName(instanceDefinition.getName());

    clusterSpec =
      ClusterDescriptionOperations.buildFromInstanceDefinition(
        instanceDefinition);
    

//     Add the -site configuration properties
    for (Map.Entry<String, String> prop : clientProperties.entrySet()) {
      clusterSpec.clientProperties.put(prop.getKey(), prop.getValue());
    }
    
  }
  
  /**
   * The resource configuration is updated -review and update state.
   * @param resources updated resources specification
   * @return a list of any dynamically added provider roles
   * (purely for testing purposes)
   */
  public synchronized List<ProviderRole> updateResourceDefinitions(ConfTree resources)
      throws BadConfigException, IOException {
    log.debug("Updating resources to {}", resources);
    
    instanceDefinition.setResources(resources);
    onInstanceDefinitionUpdated();
 
    //propagate the role table
    Map<String, Map<String, String>> updated = resources.components;
    getClusterStatus().roles = SliderUtils.deepClone(updated);
    getClusterStatus().updateTime = now();
    return buildRoleRequirementsFromResources();
  }

  /**
   * build the role requirements from the cluster specification
   * @return a list of any dynamically added provider roles
   */
  private List<ProviderRole> buildRoleRequirementsFromResources() throws BadConfigException {

    List<ProviderRole> newRoles = new ArrayList<ProviderRole>(0);
    
    //now update every role's desired count.
    //if there are no instance values, that role count goes to zero

    ConfTreeOperations resources =
        instanceDefinition.getResourceOperations();

    // Add all the existing roles
    for (RoleStatus roleStatus : getRoleStatusMap().values()) {
      if (roleStatus.getExcludeFromFlexing()) {
        // skip inflexible roles, e.g AM itself
        continue;
      }
      int currentDesired = roleStatus.getDesired();
      String role = roleStatus.getName();
      MapOperations comp =
          resources.getComponent(role);
      int desiredInstanceCount = getDesiredInstanceCount(resources, role);
      if (desiredInstanceCount == 0) {
        log.info("Role {} has 0 instances specified", role);
      }
      if (currentDesired != desiredInstanceCount) {
        log.info("Role {} flexed from {} to {}", role, currentDesired,
            desiredInstanceCount);
        roleStatus.setDesired(desiredInstanceCount);
      }
    }

    //now the dynamic ones. Iterate through the the cluster spec and
    //add any role status entries not in the role status
    Set<String> roleNames = resources.getComponentNames();
    for (String name : roleNames) {
      if (!roles.containsKey(name)) {
        // this is a new value
        log.info("Adding new role {}", name);
        MapOperations component = resources.getComponent(name);
        ProviderRole dynamicRole = createDynamicProviderRole(name,
            component);
        RoleStatus roleStatus = buildRole(dynamicRole);
        roleStatus.setDesired(getDesiredInstanceCount(resources, name));
        log.info("New role {}", roleStatus);
        roleHistory.addNewProviderRole(dynamicRole);
        newRoles.add(dynamicRole);
      }
    }
    return newRoles;
  }

  /**
   * Get the desired instance count of a role, rejecting negative values
   * @param resources resource map
   * @param role role name
   * @return the instance count
   * @throws BadConfigException if the count is negative
   */
  private int getDesiredInstanceCount(ConfTreeOperations resources,
      String role) throws BadConfigException {
    int desiredInstanceCount =
      resources.getComponentOptInt(role, ResourceKeys.COMPONENT_INSTANCES, 0);

    if (desiredInstanceCount < 0) {
      log.error("Role {} has negative desired instances : {}", role,
          desiredInstanceCount);
      throw new BadConfigException(
          "Negative instance count (%) requested for component %s",
          desiredInstanceCount, role);
    }
    return desiredInstanceCount;
  }

  /**
   * Add knowledge of a role.
   * This is a build-time operation that is not synchronized, and
   * should be used while setting up the system state -before servicing
   * requests.
   * @param providerRole role to add
   * @return the role status built up
   * @throws BadConfigException if a role of that priority already exists
   */
  public RoleStatus buildRole(ProviderRole providerRole) throws BadConfigException {
    //build role status map
    int priority = providerRole.id;
    if (roleStatusMap.containsKey(priority)) {
      throw new BadConfigException("Duplicate Provider Key: %s and %s",
                                   providerRole,
                                   roleStatusMap.get(priority));
    }
    RoleStatus roleStatus = new RoleStatus(providerRole);
    roleStatusMap.put(priority, roleStatus);
    roles.put(providerRole.name, providerRole);
    rolePriorityMap.put(priority, providerRole);
    return roleStatus;
  }

  /**
   * build up the special master node, which lives
   * in the live node set but has a lifecycle bonded to the AM
   * @param containerId the AM master
   * @param host hostname
   * @param amPort port
   * @param nodeHttpAddress http address: may be null
   */
  public void buildAppMasterNode(ContainerId containerId,
                                 String host,
                                 int amPort,
                                 String nodeHttpAddress) {
    Container container = new ContainerPBImpl();
    container.setId(containerId);
    NodeId nodeId = NodeId.newInstance(host, amPort);
    container.setNodeId(nodeId);
    container.setNodeHttpAddress(nodeHttpAddress);
    RoleInstance am = new RoleInstance(container);
    am.role = SliderKeys.COMPONENT_AM;
    appMasterNode = am;
    //it is also added to the set of live nodes
    getLiveNodes().put(containerId, am);
    
    // patch up the role status
    RoleStatus roleStatus = roleStatusMap.get(
        (SliderKeys.ROLE_AM_PRIORITY_INDEX));
    roleStatus.setDesired(1);
    roleStatus.incActual();
    roleStatus.incStarted();
  }

  /**
   * Note that the master node has been launched,
   * though it isn't considered live until any forked
   * processes are running. It is NOT registered with
   * the role history -the container is incomplete
   * and it will just cause confusion
   */
  public void noteAMLaunched() {
    getLiveNodes().put(appMasterNode.getContainerId(), appMasterNode);
  }

  /**
   * AM declares ourselves live in the cluster description.
   * This is meant to be triggered from the callback
   * indicating the spawned process is up and running.
   */
  public void noteAMLive() {
    appMasterNode.state = ClusterDescription.STATE_LIVE;
  }

  public RoleInstance getAppMasterNode() {
    return appMasterNode;
  }


  public RoleStatus lookupRoleStatus(int key) {
    RoleStatus rs = getRoleStatusMap().get(key);
    if (rs == null) {
      throw new RuntimeException("Cannot find role for role ID " + key);
    }
    return rs;
  }


  public RoleStatus lookupRoleStatus(Container c) throws YarnRuntimeException {
    return lookupRoleStatus(ContainerPriority.extractRole(c));
  }

  /**
   * Get a clone of the role status list. Concurrent events may mean this
   * list (or indeed, some of the role status entries) may be inconsistent
   * @return a snapshot of the role status entries
   */
  public List<RoleStatus> cloneRoleStatusList() {
    Collection<RoleStatus> statuses = roleStatusMap.values();
    List<RoleStatus> statusList = new ArrayList<RoleStatus>(statuses.size());
    try {
      for (RoleStatus status : statuses) {
        statusList.add((RoleStatus)(status.clone()));
      }
    } catch (CloneNotSupportedException e) {
      log.warn("Unexpected cloning failure: {}", e, e);
    }
    return statusList;
  }


  public RoleStatus lookupRoleStatus(String name) throws YarnRuntimeException {
    ProviderRole providerRole = roles.get(name);
    if (providerRole == null) {
      throw new YarnRuntimeException("Unknown role " + name);
    }
    return lookupRoleStatus(providerRole.id);
  }


  /**
   * Clone the list of active (==owned) containers
   * @return the list of role instances representing all owned containers
   */
  public synchronized List<RoleInstance> cloneOwnedContainerList() {
    Collection<RoleInstance> values = ownedContainers.values();
    return new ArrayList<RoleInstance>(values);
  }

  /**
   * Get the number of active (==owned) containers
   * @return
   */
  public int getNumOwnedContainers() {
    return ownedContainers.size();
  }
  
  /**
   * Look up an active container: any container that the AM has, even
   * if it is not currently running/live
   */
  public RoleInstance getOwnedContainer(ContainerId id) {
    return ownedContainers.get(id);
  }

  /**
   * Remove an owned container
   * @param id container ID
   * @return the instance removed
   */
  private RoleInstance removeOwnedContainer(ContainerId id) {
    return ownedContainers.remove(id);
  }

  /**
   * set/update an owned container
   * @param id container ID
   * @param instance
   * @return
   */
  private RoleInstance putOwnedContainer(ContainerId id,
      RoleInstance instance) {
    return ownedContainers.put(id, instance);
  }

  /**
   * Clone the live container list. This is synchronized.
   * @return a snapshot of the live node list
   */
  public synchronized List<RoleInstance> cloneLiveContainerInfoList() {
    List<RoleInstance> allRoleInstances;
    Collection<RoleInstance> values = getLiveNodes().values();
    allRoleInstances = new ArrayList<RoleInstance>(values);
    return allRoleInstances;
  }

  /**
   * Lookup live instance by string value of container ID
   * @param containerId container ID as a string
   * @return the role instance for that container
   * @throws NoSuchNodeException if it does not exist
   */
  public synchronized RoleInstance getLiveInstanceByContainerID(String containerId)
      throws NoSuchNodeException {
    Collection<RoleInstance> nodes = getLiveNodes().values();
    return findNodeInCollection(containerId, nodes);
  }

  /**
   * Lookup owned instance by string value of container ID
   * @param containerId container ID as a string
   * @return the role instance for that container
   * @throws NoSuchNodeException if it does not exist
   */
  public synchronized RoleInstance getOwnedInstanceByContainerID(String containerId)
      throws NoSuchNodeException {
    Collection<RoleInstance> nodes = ownedContainers.values();
    return findNodeInCollection(containerId, nodes);
  }

  
  /**
   * Iterate through a collection of role instances to find one with a
   * specific (string) container ID
   * @param containerId container ID as a string
   * @param nodes collection
   * @return 
   * @throws NoSuchNodeException if there was no match
   */
  private RoleInstance findNodeInCollection(String containerId,
      Collection<RoleInstance> nodes) throws NoSuchNodeException {
    RoleInstance found = null;
    for (RoleInstance node : nodes) {
      if (containerId.equals(node.id)) {
        found = node;
        break;
      }
    }
    if (found != null) {
      return found;
    } else {
      //at this point: no node
      throw new NoSuchNodeException(containerId);
    }
  }


  public synchronized List<RoleInstance> getLiveInstancesByContainerIDs(
    Collection<String> containerIDs) {
    //first, a hashmap of those containerIDs is built up
    Set<String> uuidSet = new HashSet<String>(containerIDs);
    List<RoleInstance> nodes = new ArrayList<RoleInstance>(uuidSet.size());
    Collection<RoleInstance> clusterNodes = getLiveNodes().values();

    for (RoleInstance node : clusterNodes) {
      if (uuidSet.contains(node.id)) {
        nodes.add(node);
      }
    }
    //at this point: a possibly empty list of nodes
    return nodes;
  }

  /**
   * Enum all nodes by role. 
   * @param role role, or "" for all roles
   * @return a list of nodes, may be empty
   */
  public synchronized List<RoleInstance> enumLiveNodesInRole(String role) {
    List<RoleInstance> nodes = new ArrayList<RoleInstance>();
    Collection<RoleInstance> allRoleInstances = getLiveNodes().values();
    for (RoleInstance node : allRoleInstances) {
      if (role.isEmpty() || role.equals(node.role)) {
        nodes.add(node);
      }
    }
    return nodes;
  }

 
  /**
   * enum nodes by role ID, from either the active or live node list
   * @param roleId role the container must be in
   * @param active flag to indicate "use active list" rather than the smaller
   * "live" list
   * @return a list of nodes, may be empty
   */
  public synchronized List<RoleInstance> enumNodesWithRoleId(int roleId,
      boolean active) {
    List<RoleInstance> nodes = new ArrayList<RoleInstance>();
    Collection<RoleInstance> allRoleInstances;
    allRoleInstances = active ? ownedContainers.values() : liveNodes.values();
    for (RoleInstance node : allRoleInstances) {
      if (node.roleId == roleId) {
        nodes.add(node);
      }
    }
    return nodes;
  }


  /**
   * Build an instance map.
   * @return the map of Role name to list of role instances
   */
  private synchronized Map<String, List<String>> createRoleToInstanceMap() {
    Map<String, List<String>> map = new HashMap<String, List<String>>();
    for (RoleInstance node : getLiveNodes().values()) {
      List<String> containers = map.get(node.role);
      if (containers == null) {
        containers = new ArrayList<String>();
        map.put(node.role, containers);
      }
      containers.add(node.id);
    }
    return map;
  }
  /**
   * Build an instance map to send over the wire
   * @return the map of Role name to list of Cluster Nodes, ready
   */
  private synchronized Map<String, Map<String, ClusterNode>> createRoleToClusterNodeMap() {
    Map<String, Map<String, ClusterNode>> map = new HashMap<String, Map<String, ClusterNode>>();
    for (RoleInstance node : getLiveNodes().values()) {
      
      Map<String, ClusterNode> containers = map.get(node.role);
      if (containers == null) {
        containers = new HashMap<String, ClusterNode>();
        map.put(node.role, containers);
      }
      Messages.RoleInstanceState pbuf = node.toProtobuf();
      ClusterNode clusterNode = ClusterNode.fromProtobuf(pbuf);
      containers.put(clusterNode.name, clusterNode);
    }
    return map;
  }

  /**
   * Notification called just before the NM is asked to 
   * start a container
   * @param container container to start
   * @param instance clusterNode structure
   */
  public void containerStartSubmitted(Container container,
                                      RoleInstance instance) {
    instance.state = ClusterDescription.STATE_SUBMITTED;
    instance.container = container;
    instance.createTime = now();
    getStartingNodes().put(container.getId(), instance);
    putOwnedContainer(container.getId(), instance);
    roleHistory.onContainerStartSubmitted(container, instance);
  }

  /**
   * Note that a container has been submitted for release; update internal state
   * and mark the associated ContainerInfo released field to indicate that
   * while it is still in the active list, it has been queued for release.
   *
   * @param container container
   * @throws SliderInternalStateException if there is no container of that ID
   * on the active list
   */
  public synchronized void containerReleaseSubmitted(Container container)
      throws SliderInternalStateException {
    ContainerId id = container.getId();
    //look up the container
    RoleInstance instance = getOwnedContainer(id);
    if (instance == null) {
      throw new SliderInternalStateException(
        "No active container with ID " + id);
    }
    //verify that it isn't already released
    if (containersBeingReleased.containsKey(id)) {
      throw new SliderInternalStateException(
        "Container %s already queued for release", id);
    }
    instance.released = true;
    containersBeingReleased.put(id, instance.container);
    RoleStatus role = lookupRoleStatus(instance.roleId);
    role.incReleasing();
    roleHistory.onContainerReleaseSubmitted(container);
  }


  /**
   * Set up the resource requirements with all that this role needs, 
   * then create the container request itself.
   * @param role role to ask an instance of
   * @param capability a resource to set up
   * @return the request for a new container
   */
  public AMRMClient.ContainerRequest buildContainerResourceAndRequest(
        RoleStatus role,
        Resource capability) {
    buildResourceRequirements(role, capability);
    String labelExpression = getLabelExpression(role);
    //get the role history to select a suitable node, if available
    AMRMClient.ContainerRequest containerRequest =
      createContainerRequest(role, capability, labelExpression);
    return  containerRequest;
  }

  /**
   * Create a container request.
   * Update internal state, such as the role request count
   * This is where role history information will be used for placement decisions -
   * @param role role
   * @param resource requirements
   * @param labelExpression label expression to satisfy
   * @return the container request to submit
   */
  public AMRMClient.ContainerRequest createContainerRequest(RoleStatus role,
                                                            Resource resource,
                                                            String labelExpression) {
    
    
    AMRMClient.ContainerRequest request;
    request = roleHistory.requestNode(role, resource, labelExpression);
    role.incRequested();

    return request;
  }


  /**
   * Get the value of a YARN requirement (cores, RAM, etc).
   * These are returned as integers, but there is special handling of the 
   * string {@link ResourceKeys#YARN_RESOURCE_MAX}, which triggers
   * the return of the maximum value.
   * @param name component to get from
   * @param option option name
   * @param defVal default value
   * @param maxVal value to return if the max val is requested
   * @return parsed value
   * @throws NumberFormatException if the role could not be parsed.
   */
  private int getResourceRequirement(ConfTreeOperations resources,
                                     String name,
                                     String option,
                                     int defVal,
                                     int maxVal) {
    
    String val = resources.getComponentOpt(name, option,
                                           Integer.toString(defVal));
    Integer intVal;
    if (ResourceKeys.YARN_RESOURCE_MAX.equals(val)) {
      intVal = maxVal;
    } else {
      intVal = Integer.decode(val);
    }
    return intVal;
  }

  
  /**
   * Build up the resource requirements for this role from the
   * cluster specification, including substituing max allowed values
   * if the specification asked for it.
   * @param role role
   * @param capability capability to set up
   */
  public void buildResourceRequirements(RoleStatus role, Resource capability) {
    // Set up resource requirements from role values
    String name = role.getName();
    ConfTreeOperations resources = getResourcesSnapshot();
    int cores = getResourceRequirement(resources,
                                       name,
                                       YARN_CORES,
                                       DEF_YARN_CORES,
                                       containerMaxCores);
    capability.setVirtualCores(cores);
    int ram = getResourceRequirement(resources, name,
                                     YARN_MEMORY,
                                     DEF_YARN_MEMORY,
                                     containerMaxMemory);
    capability.setMemory(ram);
  }

  /**
   * Extract the label expression for this role.
   * @param role role
   */
  public String getLabelExpression(RoleStatus role) {
    // Set up resource requirements from role values
    String name = role.getName();
    ConfTreeOperations resources = getResourcesSnapshot();
    return resources.getComponentOpt(name, YARN_LABEL_EXPRESSION, DEF_YARN_LABEL_EXPRESSION);
  }

  /**
   * add a launched container to the node map for status responses
   * @param container id
   * @param node node details
   */
  private void addLaunchedContainer(Container container, RoleInstance node) {
    node.container = container;
    if (node.role == null) {
      throw new RuntimeException(
        "Unknown role for node " + node);
    }
    getLiveNodes().put(node.getContainerId(), node);
    //tell role history
    roleHistory.onContainerStarted(container);
  }

  /**
   * container start event
   * @param containerId container that is to be started
   * @return the role instance, or null if there was a problem
   */
  public synchronized RoleInstance onNodeManagerContainerStarted(ContainerId containerId) {
    try {
      return innerOnNodeManagerContainerStarted(containerId);
    } catch (YarnRuntimeException e) {
      log.error("NodeManager callback on started container {} failed",
                containerId,
                e);
      return null;
    }
  }

   /**
   * container start event handler -throwing an exception on problems
   * @param containerId container that is to be started
   * @return the role instance
   * @throws RuntimeException on problems
   */
  @VisibleForTesting
  public RoleInstance innerOnNodeManagerContainerStarted(ContainerId containerId) {
    incStartedCountainerCount();
    RoleInstance instance = getOwnedContainer(containerId);
    if (instance == null) {
      //serious problem
      throw new YarnRuntimeException("Container not in active containers start "+
                containerId);
    }
    if (instance.role == null) {
      throw new YarnRuntimeException("Component instance has no instance name " +
                                     instance);
    }
    instance.startTime = now();
    RoleInstance starting = getStartingNodes().remove(containerId);
    if (null == starting) {
      throw new YarnRuntimeException(
        "Container "+ containerId +"%s is already started");
    }
    instance.state = ClusterDescription.STATE_LIVE;
    RoleStatus roleStatus = lookupRoleStatus(instance.roleId);
    roleStatus.incStarted();
    Container container = instance.container;
    addLaunchedContainer(container, instance);
    return instance;
  }

  /**
   * update the application state after a failure to start a container.
   * This is perhaps where blacklisting could be most useful: failure
   * to start a container is a sign of a more serious problem
   * than a later exit.
   *
   * -relayed from NMClientAsync.CallbackHandler 
   * @param containerId failing container
   * @param thrown what was thrown
   */
  public synchronized void onNodeManagerContainerStartFailed(ContainerId containerId,
                                                             Throwable thrown) {
    removeOwnedContainer(containerId);
    incFailedCountainerCount();
    incStartFailedCountainerCount();
    RoleInstance instance = getStartingNodes().remove(containerId);
    if (null != instance) {
      RoleStatus roleStatus = lookupRoleStatus(instance.roleId);
      String text;
      if (null != thrown) {
        text = SliderUtils.stringify(thrown);
      } else {
        text = "container start failure";
      }
      instance.diagnostics = text;
      roleStatus.noteFailed(true, null);
      getFailedNodes().put(containerId, instance);
      roleHistory.onNodeManagerContainerStartFailed(instance.container);
    }
  }

  public synchronized void onNodesUpdated(List<NodeReport> updatedNodes) {
    roleHistory.onNodesUpdated(updatedNodes);
  }

  /**
   * Is a role short lived by the threshold set for this application
   * @param instance instance
   * @return true if the instance is considered short lived
   */
  @VisibleForTesting
  public boolean isShortLived(RoleInstance instance) {
    long time = now();
    long started = instance.startTime;
    boolean shortlived;
    if (started > 0) {
      long duration = time - started;
      shortlived = duration < (startTimeThreshold * 1000);
    } else {
      // never even saw a start event
      shortlived = true;
    }
    return shortlived;
  }

  /**
   * Current time in milliseconds. Made protected for
   * the option to override it in tests.
   * @return the current time.
   */
  protected long now() {
    return System.currentTimeMillis();
  }

  /**
   * This is a very small class to send a multiple result back from 
   * the completion operation
   */
  public static class NodeCompletionResult {
    public boolean surplusNode = false;
    public RoleInstance roleInstance;
    public boolean containerFailed;
    public boolean unknownNode = false;

  
    public String toString() {
      final StringBuilder sb =
        new StringBuilder("NodeCompletionResult{");
      sb.append("surplusNode=").append(surplusNode);
      sb.append(", roleInstance=").append(roleInstance);
      sb.append(", containerFailed=").append(containerFailed);
      sb.append(", unknownNode=").append(unknownNode);
      sb.append('}');
      return sb.toString();
    }
  }
  
  /**
   * handle completed node in the CD -move something from the live
   * server list to the completed server list
   * @param status the node that has just completed
   * @return NodeCompletionResult
   */
  public synchronized NodeCompletionResult onCompletedNode(ContainerStatus status) {
    ContainerId containerId = status.getContainerId();
    NodeCompletionResult result = new NodeCompletionResult();
    RoleInstance roleInstance;

    if (containersBeingReleased.containsKey(containerId)) {
      log.info("Container was queued for release : {}", containerId);
      Container container = containersBeingReleased.remove(containerId);
      RoleStatus roleStatus = lookupRoleStatus(container);
      int releasing = roleStatus.decReleasing();
      int actual = roleStatus.decActual();
      int completedCount = roleStatus.incCompleted();
      log.info("decrementing role count for role {} to {}; releasing={}, completed={}",
          roleStatus.getName(),
          actual,
          releasing,
          completedCount);
      roleHistory.onReleaseCompleted(container, true);

    } else if (surplusNodes.remove(containerId)) {
      //its a surplus one being purged
      result.surplusNode = true;
    } else {
      //a container has failed 
      result.containerFailed = true;
      roleInstance = removeOwnedContainer(containerId);
      if (roleInstance != null) {
        //it was active, move it to failed 
        incFailedCountainerCount();
        failedNodes.put(containerId, roleInstance);
      } else {
        // the container may have been noted as failed already, so look
        // it up
        roleInstance = failedNodes.get(containerId);
      }
      if (roleInstance != null) {
        int roleId = roleInstance.roleId;
        String rolename = roleInstance.role;
        log.info("Failed container in role[{}] : {}", roleId, rolename);
        try {
          RoleStatus roleStatus = lookupRoleStatus(roleId);
          roleStatus.decActual();
          boolean shortLived = isShortLived(roleInstance);
          String message;
          Container failedContainer = roleInstance.container;
          
          //build the failure message
          if (failedContainer != null) {
            String completedLogsUrl = getLogsURLForContainer(failedContainer);
            message = String.format("Failure %s on host %s: %s",
                roleInstance.getContainerId().toString(),
                failedContainer.getNodeId().getHost(),
                completedLogsUrl);
          } else {
            message = String.format("Failure %s", containerId);
          }
          int failed = roleStatus.noteFailed(shortLived, message);
          log.info("Current count of failed role[{}] {} =  {}",
              roleId, rolename, failed);
          if (failedContainer != null) {
            roleHistory.onFailedContainer(failedContainer, shortLived);
          }
          
        } catch (YarnRuntimeException e1) {
          log.error("Failed container of unknown role {}", roleId);
        }
      } else {
        //this isn't a known container.
        
        log.error("Notified of completed container {} that is not in the list" +
                  " of active or failed containers", containerId);
        completionOfUnknownContainerEvent.incrementAndGet();
        result.unknownNode = true;
      }
    }

    if (result.surplusNode) {
      //a surplus node
      return result;
    }

    //record the complete node's details; this pulls it from the livenode set 
    //remove the node
    ContainerId id = status.getContainerId();
    log.info("Removing node ID {}", id);
    RoleInstance node = getLiveNodes().remove(id);
    if (node != null) {
      node.state = ClusterDescription.STATE_DESTROYED;
      node.exitCode = status.getExitStatus();
      node.diagnostics = status.getDiagnostics();
      getCompletedNodes().put(id, node);
      result.roleInstance = node;
    } else {
      // not in the list
      log.warn("Received notification of completion of unknown node {}", id);
      completionOfNodeNotInLiveListEvent.incrementAndGet();

    }
    
    // and the active node list if present
    removeOwnedContainer(containerId);
    
    // finally, verify the node doesn't exist any more
    assert !containersBeingReleased.containsKey(
        containerId) : "container still in release queue";
    assert !getLiveNodes().containsKey(
        containerId) : " container still in live nodes";
    assert getOwnedContainer(containerId) ==
           null : "Container still in active container list";

    return result;
  }

  /**
   * Get the URL log for a container
   * @param c container
   * @return the URL or "" if it cannot be determined
   */
  protected String getLogsURLForContainer(Container c) {
    if (c==null) {
      return null;
    }
    String user = null;
    try {
      user = SliderUtils.getCurrentUser().getShortUserName();
    } catch (IOException ignored) {
    }
    String completedLogsUrl = "";
    String url = logServerURL;
    if (user != null && SliderUtils.isSet(url)) {
      completedLogsUrl = url
          + "/" + c.getNodeId() + "/" + c.getId() + "/ctx/" + user;
    }
    return completedLogsUrl;
  }


  /**
   * Return the percentage done that Slider is to have YARN display in its
   * Web UI
   * @return an number from 0 to 100
   */
  public synchronized float getApplicationProgressPercentage() {
    float percentage;
    int desired = 0;
    float actual = 0;
    for (RoleStatus role : getRoleStatusMap().values()) {
      desired += role.getDesired();
      actual += role.getActual();
    }
    if (desired == 0) {
      percentage = 100;
    } else {
      percentage = actual / desired;
    }
    return percentage;
  }


  public void refreshClusterStatus() {
    refreshClusterStatus(null);
  }
  
  /**
   * Update the cluster description with anything interesting
   * @param providerStatus status from the provider for the cluster info section
   */
  public synchronized ClusterDescription refreshClusterStatus(Map<String, String> providerStatus) {
    ClusterDescription cd = getClusterStatus();
    long now = now();
    cd.setInfoTime(StatusKeys.INFO_STATUS_TIME_HUMAN,
                   StatusKeys.INFO_STATUS_TIME_MILLIS,
                   now);
    if (providerStatus != null) {
      for (Map.Entry<String, String> entry : providerStatus.entrySet()) {
        cd.setInfo(entry.getKey(),entry.getValue());
      }
    }
    MapOperations infoOps = new MapOperations("info", cd.info);
    infoOps.mergeWithoutOverwrite(applicationInfo);
    SliderUtils.addBuildInfo(infoOps, "status");
    cd.statistics = new HashMap<String, Map<String, Integer>>();

    // build the map of node -> container IDs
    Map<String, List<String>> instanceMap = createRoleToInstanceMap();
    cd.instances = instanceMap;
    
    //build the map of node -> containers
    Map<String, Map<String, ClusterNode>> clusterNodes =
      createRoleToClusterNodeMap();
    cd.status = new HashMap<String, Object>();
    cd.status.put(ClusterDescriptionKeys.KEY_CLUSTER_LIVE, clusterNodes);


    for (RoleStatus role : getRoleStatusMap().values()) {
      String rolename = role.getName();
      List<String> instances = instanceMap.get(rolename);
      int nodeCount = instances != null ? instances.size(): 0;
      cd.setRoleOpt(rolename, ResourceKeys.COMPONENT_INSTANCES,
                    role.getDesired());
      cd.setRoleOpt(rolename, RoleKeys.ROLE_ACTUAL_INSTANCES, nodeCount);
      cd.setRoleOpt(rolename, ROLE_REQUESTED_INSTANCES, role.getRequested());
      cd.setRoleOpt(rolename, ROLE_RELEASING_INSTANCES, role.getReleasing());
      cd.setRoleOpt(rolename, ROLE_FAILED_INSTANCES, role.getFailed());
      cd.setRoleOpt(rolename, ROLE_FAILED_STARTING_INSTANCES, role.getStartFailed());
      Map<String, Integer> stats = role.buildStatistics();
      cd.statistics.put(rolename, stats);
    }

    Map<String, Integer> sliderstats = new HashMap<String, Integer>();
    sliderstats.put(StatusKeys.STATISTICS_CONTAINERS_COMPLETED,
        completedContainerCount.get());
    sliderstats.put(StatusKeys.STATISTICS_CONTAINERS_FAILED,
        failedContainerCount.get());
    sliderstats.put(StatusKeys.STATISTICS_CONTAINERS_LIVE, liveNodes.size());
    sliderstats.put(StatusKeys.STATISTICS_CONTAINERS_STARTED,
        startedContainers.get());
    sliderstats.put(StatusKeys.STATISTICS_CONTAINERS_START_FAILED,
        startFailedContainers.get());
    sliderstats.put(StatusKeys.STATISTICS_CONTAINERS_SURPLUS,
        surplusContainers.get());
    sliderstats.put(StatusKeys.STATISTICS_CONTAINERS_UNKNOWN_COMPLETED,
        completionOfUnknownContainerEvent.get());
    cd.statistics.put(SliderKeys.COMPONENT_AM, sliderstats);
    return cd;
  }

  /**
   * Look at where the current node state is -and whether it should be changed
   */
  public synchronized List<AbstractRMOperation> reviewRequestAndReleaseNodes()
      throws SliderInternalStateException, TriggerClusterTeardownException {
    log.debug("in reviewRequestAndReleaseNodes()");
    List<AbstractRMOperation> allOperations = new ArrayList<AbstractRMOperation>();
    for (RoleStatus roleStatus : getRoleStatusMap().values()) {
      if (!roleStatus.getExcludeFromFlexing()) {
        List<AbstractRMOperation> operations = reviewOneRole(roleStatus);
        allOperations.addAll(operations);
      }
    }
    return allOperations;
  }

  /**
   * Check the failure threshold for a role
   * @param role role to examine
   * @throws TriggerClusterTeardownException if the role
   * has failed too many times
   */
  private void checkFailureThreshold(RoleStatus role)
      throws TriggerClusterTeardownException {
    int failures = role.getFailed();
    int threshold = getFailureThresholdForRole(role);
    log.debug("Failure count of component: {}: {}, threshold={}",
        role.getName(), failures, threshold);

    if (failures > threshold) {
      throw new TriggerClusterTeardownException(
        SliderExitCodes.EXIT_DEPLOYMENT_FAILED,
          FinalApplicationStatus.FAILED, ErrorStrings.E_UNSTABLE_CLUSTER +
        " - failed with component %s failing %d times (%d in startup);" +
        " threshold is %d - last failure: %s",
          role.getName(),
        role.getFailed(),
        role.getStartFailed(),
          threshold,
        role.getFailureMessage());
    }
  }

  /**
   * Get the failure threshold for a specific role, falling back to
   * the global one if not
   * @param roleStatus role
   * @return the threshold for failures
   */
  private int getFailureThresholdForRole(RoleStatus roleStatus) {
    ConfTreeOperations resources =
        instanceDefinition.getResourceOperations();
    return resources.getComponentOptInt(roleStatus.getName(),
        ResourceKeys.CONTAINER_FAILURE_THRESHOLD,
        failureThreshold);
  }
  
  /**
   * Reset the failure counts of all roles
   */
  public void resetFailureCounts() {
    for (RoleStatus roleStatus : getRoleStatusMap().values()) {
      int failed = roleStatus.resetFailed();
      log.debug("Resetting failure count of {}; was {}", roleStatus.getName(),
          failed);
    }
  }
  
  /**
   * Look at the allocation status of one role, and trigger add/release
   * actions if the number of desired role instances doesnt equal 
   * (actual+pending)
   * @param role role
   * @return a list of operations
   * @throws SliderInternalStateException if the operation reveals that
   * the internal state of the application is inconsistent.
   */
  @SuppressWarnings("SynchronizationOnLocalVariableOrMethodParameter")
  private List<AbstractRMOperation> reviewOneRole(RoleStatus role)
      throws SliderInternalStateException, TriggerClusterTeardownException {
    List<AbstractRMOperation> operations = new ArrayList<AbstractRMOperation>();
    int delta;
    String details;
    int expected;
    String name = role.getName();
    synchronized (role) {
      delta = role.getDelta();
      expected = role.getDesired();
    }

    log.info("Reviewing {} : expected {}", role, expected);
    checkFailureThreshold(role);
    
    if (expected < 0 ) {
      // negative value: fail
      throw new TriggerClusterTeardownException(
          SliderExitCodes.EXIT_DEPLOYMENT_FAILED,
          FinalApplicationStatus.FAILED,
          "Negative component count of %d desired for component %s",
          expected, role);
    }
    
    if (delta > 0) {
      log.info("{}: Asking for {} more nodes(s) for a total of {} ", name,
               delta, expected);
      //more workers needed than we have -ask for more
      for (int i = 0; i < delta; i++) {
        Resource capability = recordFactory.newResource();
        AMRMClient.ContainerRequest containerAsk =
          buildContainerResourceAndRequest(role, capability);
        log.info("Container ask is {} and label = {}", containerAsk,
            containerAsk.getNodeLabelExpression());
        int askMemory = containerAsk.getCapability().getMemory();
        if (askMemory > this.containerMaxMemory) {
          log.warn("Memory requested: {} > max of {}", askMemory, containerMaxMemory);
        }
        operations.add(new ContainerRequestOperation(containerAsk));
      }
    } else if (delta < 0) {
      log.info("{}: Asking for {} fewer node(s) for a total of {}", name,
               -delta,
               expected);
      //reduce the number expected (i.e. subtract the delta)

      //then pick some containers to kill
      int excess = -delta;

      // how many requests are outstanding
      int outstandingRequests = role.getRequested();
      if (outstandingRequests > 0) {
        // outstanding requests.
        int toCancel = Math.min(outstandingRequests, excess);
        Priority p1 =
            ContainerPriority.createPriority(role.getPriority(), true);
        Priority p2 =
            ContainerPriority.createPriority(role.getPriority(), false);
        operations.add(new CancelRequestOperation(p1, p2, toCancel));
        role.cancel(toCancel);
        excess -= toCancel;
        assert excess >= 0 : "Attempted to cancel too many requests";
        log.info("Submitted {} cancellations, leaving {} to release",
            toCancel, excess);
        if (excess == 0) {
          log.info("After cancelling requests, application is at desired size");
        }
      }


      // after the cancellation there may be no excess
      if (excess > 0) {
        // get the nodes to release
        int roleId = role.getKey();

        // enum all active nodes that aren't being released
        List<RoleInstance> containersToRelease = enumNodesWithRoleId(roleId, true);

        // cut all release-in-progress nodes
        ListIterator<RoleInstance> li = containersToRelease.listIterator();
        while (li.hasNext()) {
          RoleInstance next = li.next();
          if (next.released) {
            li.remove();
          }
        }

        // warn if the desired state can't be reaced
        int numberAvailableForRelease = containersToRelease.size();
        if (numberAvailableForRelease < excess) {
          log.warn("Not enough nodes to release, have {} and need {} more",
              numberAvailableForRelease,
              excess - numberAvailableForRelease);
        }

        // ask the release selector to sort the targets
        containersToRelease =  containerReleaseSelector.sortCandidates(
            roleId,
            containersToRelease,
            excess);

        //crop to the excess

        List<RoleInstance> finalCandidates = (excess < numberAvailableForRelease) 
            ? containersToRelease.subList(0, excess)
            : containersToRelease;


        // then build up a release operation, logging each container as released
        for (RoleInstance possible : finalCandidates) {
          log.debug("Targeting for release: {}", possible);
          containerReleaseSubmitted(possible.container);
          operations.add(new ContainerReleaseOperation(possible.getId()));       
        }
      }

    }

    return operations;
  }

  /**
   * Releases a container based on container id
   * @param containerId
   * @return
   * @throws SliderInternalStateException
   */
  public List<AbstractRMOperation> releaseContainer(ContainerId containerId)
      throws SliderInternalStateException {
    List<AbstractRMOperation> operations = new ArrayList<AbstractRMOperation>();
    List<RoleInstance> activeRoleInstances = cloneOwnedContainerList();
    for (RoleInstance role : activeRoleInstances) {
      if (role.container.getId().equals(containerId)) {
        containerReleaseSubmitted(role.container);
        operations.add(new ContainerReleaseOperation(role.getId()));
      }
    }

    return operations;
  }


  /**
   * Find a container running on a specific host -looking
   * into the node ID to determine this.
   *
   * @param node node
   * @param roleId role the container must be in
   * @return a container or null if there are no containers on this host
   * that can be released.
   */
  private RoleInstance findRoleInstanceOnHost(NodeInstance node, int roleId) {
    Collection<RoleInstance> targets = cloneOwnedContainerList();
    String hostname = node.hostname;
    for (RoleInstance ri : targets) {
      if (hostname.equals(RoleHistoryUtils.hostnameOf(ri.container))
                         && ri.roleId == roleId
        && containersBeingReleased.get(ri.getContainerId()) == null) {
        return ri;
      }
    }
    return null;
  }
  
  /**
   * Release all containers.
   * @return a list of operations to execute
   */
  public synchronized List<AbstractRMOperation> releaseAllContainers() {

    Collection<RoleInstance> targets = cloneOwnedContainerList();
    log.info("Releasing {} containers", targets.size());
    List<AbstractRMOperation> operations =
      new ArrayList<AbstractRMOperation>(targets.size());
    for (RoleInstance instance : targets) {
      Container possible = instance.container;
      ContainerId id = possible.getId();
      if (!instance.released) {
        String url = getLogsURLForContainer(possible);
        log.info("Releasing container. Log: " + url);
        try {
          containerReleaseSubmitted(possible);
        } catch (SliderInternalStateException e) {
          log.warn("when releasing container {} :", possible, e);
        }
        operations.add(new ContainerReleaseOperation(id));
      }
    }
    return operations;
  }

  /**
   * Event handler for allocated containers: builds up the lists
   * of assignment actions (what to run where), and possibly
   * a list of release operations
   * @param allocatedContainers the containers allocated
   * @param assignments the assignments of roles to containers
   * @param releaseOperations any release operations
   */
  public synchronized void onContainersAllocated(List<Container> allocatedContainers,
                                    List<ContainerAssignment> assignments,
                                    List<AbstractRMOperation> releaseOperations) {
    assignments.clear();
    releaseOperations.clear();
    List<Container> ordered = roleHistory.prepareAllocationList(allocatedContainers);
    for (Container container : ordered) {
      String containerHostInfo = container.getNodeId().getHost()
                                 + ":" +
                                 container.getNodeId().getPort();
      int allocated;
      int desired;
      //get the role
      ContainerId cid = container.getId();
      RoleStatus role = lookupRoleStatus(container);
      

      //dec requested count
      role.decRequested();
      //inc allocated count -this may need to be dropped in a moment,
      // but us needed to update the logic below
      allocated = role.incActual();

      //look for (race condition) where we get more back than we asked
      desired = role.getDesired();

      roleHistory.onContainerAllocated( container, desired, allocated );

      if (allocated > desired) {
        log.info("Discarding surplus container {} on {}", cid,
                 containerHostInfo);
        releaseOperations.add(new ContainerReleaseOperation(cid));
        //register as a surplus node
        surplusNodes.add(cid);
        surplusContainers.incrementAndGet();
        //and, as we aren't binding it to role, dec that role's actual count
        role.decActual();
      } else {

        String roleName = role.getName();
        log.info("Assigning role {} to container" +
                 " {}," +
                 " on {}:{},",
                 roleName,
                 cid,
                 container.getNodeId().getHost(),
                 container.getNodeId().getPort()
                );

        assignments.add(new ContainerAssignment(container, role));
        //add to the history
        roleHistory.onContainerAssigned(container);
      }
    }
  }

  /**
   * Get diagnostics info about containers
   */
  public String getContainerDiagnosticInfo() {
    StringBuilder builder = new StringBuilder();
    for (RoleStatus roleStatus : getRoleStatusMap().values()) {
      builder.append(roleStatus).append('\n');
    }
    return builder.toString();
  }

  /**
   * Event handler for the list of active containers on restart.
   * Sets the info key {@link StatusKeys#INFO_CONTAINERS_AM_RESTART}
   * to the size of the list passed down (and does not set it if none were)
   * @param liveContainers the containers allocated
   * @return true if a rebuild took place (even if size 0)
   * @throws RuntimeException on problems
   */
  private boolean rebuildModelFromRestart(List<Container> liveContainers) throws
                                                                          BadClusterStateException {
    if (liveContainers == null) {
      return false;
    }
    for (Container container : liveContainers) {
      addRestartedContainer(container);
    }
    clusterStatus.setInfo(StatusKeys.INFO_CONTAINERS_AM_RESTART,
                               Integer.toString(liveContainers.size()));
    return true;
  }

  /**
   * Add a restarted container by walking it through the create/submit/start
   * lifecycle, so building up the internal structures
   * @param container container that was running before the AM restarted
   * @throws RuntimeException on problems
   */
  private void addRestartedContainer(Container container) throws
                                                          BadClusterStateException {
    String containerHostInfo = container.getNodeId().getHost()
                               + ":" +
                               container.getNodeId().getPort();
    // get the container ID
    ContainerId cid = container.getId();
    
    // get the role
    int roleId = ContainerPriority.extractRole(container);
    RoleStatus role =
      lookupRoleStatus(roleId);
    // increment its count
    role.incActual();
    String roleName = role.getName();
    
    log.info("Rebuilding container {} in role {} on {},",
             cid,
             roleName,
             containerHostInfo);
    
    //update app state internal structures and maps

    RoleInstance instance = new RoleInstance(container);
    instance.command = roleName;
    instance.role = roleName;
    instance.roleId = roleId;
    instance.environment = new String[0];
    instance.container = container;
    instance.createTime = now();
    instance.state = ClusterDescription.STATE_LIVE;
    putOwnedContainer(cid, instance);
    //role history gets told
    roleHistory.onContainerAssigned(container);
    // pretend the container has just had its start actions submitted
    containerStartSubmitted(container, instance);
    // now pretend it has just started
    innerOnNodeManagerContainerStarted(cid);
  }
}