/**
 * 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.hadoop.yarn.service.component;

import org.apache.hadoop.yarn.api.records.Container;
import org.apache.hadoop.yarn.api.records.ContainerStatus;
import org.apache.hadoop.yarn.api.records.ExecutionType;
import org.apache.hadoop.yarn.api.records.ExecutionTypeRequest;
import org.apache.hadoop.yarn.api.records.Priority;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.api.records.ResourceSizing;
import org.apache.hadoop.yarn.api.records.SchedulingRequest;
import org.apache.hadoop.yarn.api.resource.PlacementConstraint;
import org.apache.hadoop.yarn.api.resource.PlacementConstraint.TargetExpression;
import org.apache.hadoop.yarn.api.resource.PlacementConstraints;
import org.apache.hadoop.yarn.api.resource.PlacementConstraints.PlacementTargets;
import org.apache.hadoop.yarn.client.api.AMRMClient.ContainerRequest;
import org.apache.hadoop.yarn.client.api.async.AMRMClientAsync;
import org.apache.hadoop.yarn.event.AsyncDispatcher;
import org.apache.hadoop.yarn.event.EventHandler;
import org.apache.hadoop.yarn.service.ContainerFailureTracker;
import org.apache.hadoop.yarn.service.ServiceContext;
import org.apache.hadoop.yarn.service.ServiceMaster;
import org.apache.hadoop.yarn.service.ServiceMetrics;
import org.apache.hadoop.yarn.service.ServiceScheduler;
import org.apache.hadoop.yarn.service.api.records.PlacementPolicy;
import org.apache.hadoop.yarn.service.api.records.ResourceInformation;
import org.apache.hadoop.yarn.service.api.records.ServiceState;
import org.apache.hadoop.yarn.service.component.instance.ComponentInstance;
import org.apache.hadoop.yarn.service.component.instance.ComponentInstanceEvent;
import org.apache.hadoop.yarn.service.component.instance.ComponentInstanceId;
import org.apache.hadoop.yarn.service.monitor.probe.MonitorUtils;
import org.apache.hadoop.yarn.service.monitor.probe.Probe;
import org.apache.hadoop.yarn.service.provider.ProviderUtils;
import org.apache.hadoop.yarn.service.utils.ServiceUtils;
import org.apache.hadoop.yarn.state.InvalidStateTransitionException;
import org.apache.hadoop.yarn.state.MultipleArcTransition;
import org.apache.hadoop.yarn.state.SingleArcTransition;
import org.apache.hadoop.yarn.state.StateMachine;
import org.apache.hadoop.yarn.state.StateMachineFactory;
import org.apache.hadoop.yarn.util.Apps;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.text.MessageFormat;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.EnumSet;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import static org.apache.hadoop.yarn.api.records.ContainerExitStatus.*;
import static org.apache.hadoop.yarn.service.api.ServiceApiConstants.*;
import static org.apache.hadoop.yarn.service.component.ComponentEventType.*;
import static org.apache.hadoop.yarn.service.component.ComponentState.*;
import static org.apache.hadoop.yarn.service.component.instance.ComponentInstanceEventType.*;
import static org.apache.hadoop.yarn.service.conf.YarnServiceConf.CONTAINER_FAILURE_THRESHOLD;

public class Component implements EventHandler<ComponentEvent> {
  private static final Logger LOG = LoggerFactory.getLogger(Component.class);

  private org.apache.hadoop.yarn.service.api.records.Component componentSpec;
  private long allocateId;
  private Priority priority;
  private ServiceMetrics componentMetrics;
  private ServiceScheduler scheduler;
  private ServiceContext context;
  private AMRMClientAsync<ContainerRequest> amrmClient;
  private AtomicLong instanceIdCounter = new AtomicLong();
  private Map<String, ComponentInstance> compInstances =
      new ConcurrentHashMap<>();
  // component instances to be assigned with a container
  private List<ComponentInstance> pendingInstances =
      Collections.synchronizedList(new LinkedList<>());
  private ContainerFailureTracker failureTracker;
  private Probe probe;
  private final ReentrantReadWriteLock.ReadLock readLock;
  private final ReentrantReadWriteLock.WriteLock writeLock;
  public int maxContainerFailurePerComp;
  // The number of containers failed since last reset. This excludes preempted,
  // disk_failed containers etc. This will be reset to 0 periodically.
  public AtomicInteger currentContainerFailure = new AtomicInteger(0);

  private StateMachine<ComponentState, ComponentEventType, ComponentEvent>
      stateMachine;
  private AsyncDispatcher dispatcher;
  private static final StateMachineFactory<Component, ComponentState, ComponentEventType, ComponentEvent>
      stateMachineFactory =
      new StateMachineFactory<Component, ComponentState, ComponentEventType, ComponentEvent>(
          INIT)
           // INIT will only got to FLEXING
          .addTransition(INIT, EnumSet.of(STABLE, FLEXING, INIT),
              FLEX, new FlexComponentTransition())
          // container recovered on AM restart
          .addTransition(INIT, INIT, CONTAINER_RECOVERED,
              new ContainerRecoveredTransition())

          // container recovered in AM heartbeat
          .addTransition(FLEXING, FLEXING, CONTAINER_RECOVERED,
              new ContainerRecoveredTransition())

          // container allocated by RM
          .addTransition(FLEXING, FLEXING, CONTAINER_ALLOCATED,
              new ContainerAllocatedTransition())
          // container launched on NM
          .addTransition(FLEXING, EnumSet.of(STABLE, FLEXING),
              CONTAINER_STARTED, new ContainerStartedTransition())
          // container failed while flexing
          .addTransition(FLEXING, FLEXING, CONTAINER_COMPLETED,
              new ContainerCompletedTransition())
          // Flex while previous flex is still in progress
          .addTransition(FLEXING, EnumSet.of(FLEXING, STABLE), FLEX,
              new FlexComponentTransition())

          // container failed while stable
          .addTransition(STABLE, FLEXING, CONTAINER_COMPLETED,
              new ContainerCompletedTransition())
          // Ignore surplus container
          .addTransition(STABLE, STABLE, CONTAINER_ALLOCATED,
              new ContainerAllocatedTransition())
          // Flex by user
          // For flex up, go to FLEXING state
          // For flex down, go to STABLE state
          .addTransition(STABLE, EnumSet.of(STABLE, FLEXING),
              FLEX, new FlexComponentTransition())
          .addTransition(STABLE, UPGRADING, UPGRADE,
              new ComponentNeedsUpgradeTransition())
          .addTransition(FLEXING, UPGRADING, UPGRADE,
              new ComponentNeedsUpgradeTransition())
          .addTransition(UPGRADING, UPGRADING, UPGRADE,
              new ComponentNeedsUpgradeTransition())
          .installTopology();

  public Component(
      org.apache.hadoop.yarn.service.api.records.Component component,
      long allocateId, ServiceContext context) {
    this.allocateId = allocateId;
    this.priority = Priority.newInstance((int) allocateId);
    this.componentSpec = component;
    componentMetrics = ServiceMetrics.register(component.getName(),
        "Metrics for component " + component.getName());
    componentMetrics
        .tag("type", "Metrics type [component or service]", "component");
    this.scheduler = context.scheduler;
    this.context = context;
    amrmClient = scheduler.getAmRMClient();
    ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
    this.readLock = lock.readLock();
    this.writeLock = lock.writeLock();
    this.stateMachine = stateMachineFactory.make(this);
    dispatcher = scheduler.getDispatcher();
    failureTracker =
        new ContainerFailureTracker(context, this);
    probe = MonitorUtils.getProbe(componentSpec.getReadinessCheck());
    maxContainerFailurePerComp = componentSpec.getConfiguration()
        .getPropertyInt(CONTAINER_FAILURE_THRESHOLD, 10);
    createNumCompInstances(component.getNumberOfContainers());
    setDesiredContainers(component.getNumberOfContainers().intValue());
  }

  private void createNumCompInstances(long count) {
    for (int i = 0; i < count; i++) {
      createOneCompInstance();
    }
  }

  private void createOneCompInstance() {
    ComponentInstanceId id =
        new ComponentInstanceId(instanceIdCounter.getAndIncrement(),
            componentSpec.getName());
    ComponentInstance instance = new ComponentInstance(this, id);
    compInstances.put(instance.getCompInstanceName(), instance);
    pendingInstances.add(instance);
  }

  private static class FlexComponentTransition implements
      MultipleArcTransition<Component, ComponentEvent, ComponentState> {
    // For flex up, go to FLEXING state
    // For flex down, go to STABLE state
    @Override
    public ComponentState transition(Component component,
        ComponentEvent event) {
      component.setDesiredContainers((int)event.getDesired());
      if (!component.areDependenciesReady()) {
        LOG.info("[FLEX COMPONENT {}]: Flex deferred because dependencies not"
            + " satisfied.", component.getName());
        return component.getState();
      }
      if (component.getState() == INIT) {
        // This happens on init
        LOG.info("[INIT COMPONENT " + component.getName() + "]: " + event
            .getDesired() + " instances.");
        component.requestContainers(component.pendingInstances.size());
        return checkIfStable(component);
      }
      long before = component.getComponentSpec().getNumberOfContainers();
      long delta = event.getDesired() - before;
      component.getComponentSpec().setNumberOfContainers(event.getDesired());
      if (delta > 0) {
        // Scale up
        LOG.info("[FLEX UP COMPONENT " + component.getName() + "]: scaling up from "
                + before + " to " + event.getDesired());
        component.requestContainers(delta);
        component.createNumCompInstances(delta);
        component.componentSpec.setState(
            org.apache.hadoop.yarn.service.api.records.ComponentState.FLEXING);
        component.getScheduler().getApp().setState(ServiceState.STARTED);
        return FLEXING;
      } else if (delta < 0){
        delta = 0 - delta;
        // scale down
        LOG.info("[FLEX DOWN COMPONENT " + component.getName()
            + "]: scaling down from " + before + " to " + event.getDesired());
        List<ComponentInstance> list =
            new ArrayList<>(component.getAllComponentInstances());

        // sort in Most recent -> oldest order, destroy most recent ones.
        list.sort(Collections.reverseOrder());
        for (int i = 0; i < delta; i++) {
          ComponentInstance instance = list.get(i);
          // remove the instance
          component.compInstances.remove(instance.getCompInstanceName());
          component.pendingInstances.remove(instance);
          // decrement id counter
          component.instanceIdCounter.decrementAndGet();
          instance.destroy();
        }
        checkAndUpdateComponentState(component, false);
        return STABLE;
      } else {
        LOG.info("[FLEX COMPONENT " + component.getName() + "]: already has " +
            event.getDesired() + " instances, ignoring");
        return STABLE;
      }
    }
  }

  private static class ContainerAllocatedTransition extends BaseTransition {
    @Override
    public void transition(Component component, ComponentEvent event) {
      component.assignContainerToCompInstance(event.getContainer());
    }
  }

  private static class ContainerRecoveredTransition extends BaseTransition {
    @Override
    public void transition(Component component, ComponentEvent event) {
      ComponentInstance instance = event.getInstance();
      Container container = event.getContainer();
      if (instance == null) {
        LOG.info("[COMPONENT {}]: Trying to recover {} but event did not " +
                "specify component instance",
            component.getName(), container.getId());
        component.releaseContainer(container);
        return;
      }

      component.pendingInstances.remove(instance);
      instance.setContainer(container);
      ProviderUtils.initCompInstanceDir(component.getContext().fs, instance);
      component.getScheduler().addLiveCompInstance(container.getId(), instance);
      LOG.info("[COMPONENT {}]: Recovered {} for component instance {} on " +
              "host {}, num pending component instances reduced to {} ",
          component.getName(), container.getId(),
          instance.getCompInstanceName(), container.getNodeId(),
          component.pendingInstances.size());
      component.dispatcher.getEventHandler().handle(
          new ComponentInstanceEvent(container.getId(), START));
    }
  }

  private static class ContainerStartedTransition implements
      MultipleArcTransition<Component,ComponentEvent,ComponentState> {

    @Override public ComponentState transition(Component component,
        ComponentEvent event) {
      component.dispatcher.getEventHandler().handle(
          new ComponentInstanceEvent(event.getContainerId(), START));
      return checkIfStable(component);
    }
  }

  private static ComponentState checkIfStable(Component component) {
    // if desired == running
    if (component.componentMetrics.containersReady.value() == component
        .getComponentSpec().getNumberOfContainers()) {
      component.componentSpec.setState(
          org.apache.hadoop.yarn.service.api.records.ComponentState.STABLE);
      return STABLE;
    } else {
      component.componentSpec.setState(
          org.apache.hadoop.yarn.service.api.records.ComponentState.FLEXING);
      return FLEXING;
    }
  }

  // This method should be called whenever there is an increment or decrement
  // of a READY state container of a component
  public static synchronized void checkAndUpdateComponentState(
      Component component, boolean isIncrement) {
    org.apache.hadoop.yarn.service.api.records.ComponentState curState =
        component.componentSpec.getState();
    if (isIncrement) {
      // check if all containers are in READY state
      if (component.componentMetrics.containersReady
          .value() == component.componentMetrics.containersDesired.value()) {
        component.componentSpec.setState(
            org.apache.hadoop.yarn.service.api.records.ComponentState.STABLE);
        if (curState != component.componentSpec.getState()) {
          LOG.info("[COMPONENT {}] state changed from {} -> {}",
              component.componentSpec.getName(), curState,
              component.componentSpec.getState());
        }
        // component state change will trigger re-check of service state
        ServiceMaster.checkAndUpdateServiceState(component.scheduler,
            isIncrement);
      }
    } else {
      // container moving out of READY state could be because of FLEX down so
      // still need to verify the count before changing the component state
      if (component.componentMetrics.containersReady
          .value() < component.componentMetrics.containersDesired.value()) {
        component.componentSpec.setState(
            org.apache.hadoop.yarn.service.api.records.ComponentState.FLEXING);
        if (curState != component.componentSpec.getState()) {
          LOG.info("[COMPONENT {}] state changed from {} -> {}",
              component.componentSpec.getName(), curState,
              component.componentSpec.getState());
        }
        // component state change will trigger re-check of service state
        ServiceMaster.checkAndUpdateServiceState(component.scheduler,
            isIncrement);
      }
    }
  }

  private static class ContainerCompletedTransition extends BaseTransition {
    @Override
    public void transition(Component component, ComponentEvent event) {
      component.updateMetrics(event.getStatus());
      component.dispatcher.getEventHandler().handle(
          new ComponentInstanceEvent(event.getStatus().getContainerId(),
              STOP).setStatus(event.getStatus()));
      component.componentSpec.setState(
          org.apache.hadoop.yarn.service.api.records.ComponentState.FLEXING);
      component.getScheduler().getApp().setState(ServiceState.STARTED);
    }
  }

  private static class ComponentNeedsUpgradeTransition extends BaseTransition {
    @Override
    public void transition(Component component, ComponentEvent event) {
      component.componentSpec.setState(org.apache.hadoop.yarn.service.api.
          records.ComponentState.NEEDS_UPGRADE);
    }
  }

  public void removePendingInstance(ComponentInstance instance) {
    pendingInstances.remove(instance);
  }

  public void reInsertPendingInstance(ComponentInstance instance) {
    pendingInstances.add(instance);
  }

  private void releaseContainer(Container container) {
    scheduler.getAmRMClient().releaseAssignedContainer(container.getId());
    componentMetrics.surplusContainers.incr();
    scheduler.getServiceMetrics().surplusContainers.incr();
  }

  private void assignContainerToCompInstance(Container container) {
    if (pendingInstances.size() == 0) {
      LOG.info(
          "[COMPONENT {}]: No pending component instance left, release surplus container {}",
          getName(), container.getId());
      releaseContainer(container);
      return;
    }
    ComponentInstance instance = pendingInstances.remove(0);
    LOG.info(
        "[COMPONENT {}]: {} allocated, num pending component instances reduced to {}",
        getName(), container.getId(), pendingInstances.size());
    instance.setContainer(container);
    scheduler.addLiveCompInstance(container.getId(), instance);
    LOG.info(
        "[COMPONENT {}]: Assigned {} to component instance {} and launch on host {} ",
        getName(), container.getId(), instance.getCompInstanceName(),
        container.getNodeId());
    scheduler.getContainerLaunchService()
        .launchCompInstance(scheduler.getApp(), instance, container);
  }

  @SuppressWarnings({ "unchecked" })
  public void requestContainers(long count) {
    LOG.info("[COMPONENT {}] Requesting for {} container(s)",
        componentSpec.getName(), count);
    org.apache.hadoop.yarn.service.api.records.Resource componentResource =
        componentSpec.getResource();

    Resource resource = Resource.newInstance(componentResource.calcMemoryMB(),
        componentResource.getCpus());

    if (componentResource.getAdditional() != null) {
      for (Map.Entry<String, ResourceInformation> entry : componentResource
          .getAdditional().entrySet()) {

        String resourceName = entry.getKey();

        // Avoid setting memory/cpu under "additional"
        if (resourceName.equals(
            org.apache.hadoop.yarn.api.records.ResourceInformation.MEMORY_URI)
            || resourceName.equals(
            org.apache.hadoop.yarn.api.records.ResourceInformation.VCORES_URI)) {
          LOG.warn("Please set memory/vcore in the main section of resource, "
              + "ignoring this entry=" + resourceName);
          continue;
        }

        ResourceInformation specInfo = entry.getValue();
        org.apache.hadoop.yarn.api.records.ResourceInformation ri =
            org.apache.hadoop.yarn.api.records.ResourceInformation.newInstance(
                entry.getKey(),
                specInfo.getUnit(),
                specInfo.getValue());
        resource.setResourceInformation(resourceName, ri);
      }
    }

    if (!scheduler.hasAtLeastOnePlacementConstraint()) {
      for (int i = 0; i < count; i++) {
        ContainerRequest request = ContainerRequest.newBuilder()
            .capability(resource).priority(priority)
            .allocationRequestId(allocateId).relaxLocality(true).build();
        LOG.info("[COMPONENT {}] Submitting container request : {}",
            componentSpec.getName(), request);
        amrmClient.addContainerRequest(request);
      }
    } else {
      // Schedule placement requests. Validation of non-null target tags and
      // that they refer to existing component names are already done. So, no
      // need to validate here.
      PlacementPolicy placementPolicy = componentSpec.getPlacementPolicy();
      Collection<SchedulingRequest> schedulingRequests = new HashSet<>();
      // We prepare an AND-ed composite constraint to be the final composite
      // constraint. If placement expressions are specified to create advanced
      // composite constraints then this AND-ed composite constraint is not
      // used.
      PlacementConstraint finalConstraint = null;
      for (org.apache.hadoop.yarn.service.api.records.PlacementConstraint
          yarnServiceConstraint : placementPolicy.getConstraints()) {
        List<TargetExpression> targetExpressions = new ArrayList<>();
        // Currently only intra-application allocation tags are supported.
        if (!yarnServiceConstraint.getTargetTags().isEmpty()) {
          targetExpressions.add(PlacementTargets.allocationTagToIntraApp(
              yarnServiceConstraint.getTargetTags().toArray(new String[0])));
        }
        // Add all node attributes
        for (Map.Entry<String, List<String>> attribute : yarnServiceConstraint
            .getNodeAttributes().entrySet()) {
          targetExpressions.add(PlacementTargets.nodeAttribute(
              attribute.getKey(), attribute.getValue().toArray(new String[0])));
        }
        // Add all node partitions
        if (!yarnServiceConstraint.getNodePartitions().isEmpty()) {
          targetExpressions
              .add(PlacementTargets.nodePartition(yarnServiceConstraint
                  .getNodePartitions().toArray(new String[0])));
        }
        PlacementConstraint constraint = null;
        switch (yarnServiceConstraint.getType()) {
        case AFFINITY:
          constraint = PlacementConstraints
              .targetIn(yarnServiceConstraint.getScope().getValue(),
                  targetExpressions.toArray(new TargetExpression[0]))
              .build();
          break;
        case ANTI_AFFINITY:
          constraint = PlacementConstraints
              .targetNotIn(yarnServiceConstraint.getScope().getValue(),
                  targetExpressions.toArray(new TargetExpression[0]))
              .build();
          break;
        case AFFINITY_WITH_CARDINALITY:
          constraint = PlacementConstraints.targetCardinality(
              yarnServiceConstraint.getScope().name().toLowerCase(),
              yarnServiceConstraint.getMinCardinality() == null ? 0
                  : yarnServiceConstraint.getMinCardinality().intValue(),
              yarnServiceConstraint.getMaxCardinality() == null
                  ? Integer.MAX_VALUE
                  : yarnServiceConstraint.getMaxCardinality().intValue(),
              targetExpressions.toArray(new TargetExpression[0])).build();
          break;
        }
        // The default AND-ed final composite constraint
        if (finalConstraint != null) {
          finalConstraint = PlacementConstraints
              .and(constraint.getConstraintExpr(),
                  finalConstraint.getConstraintExpr())
              .build();
        } else {
          finalConstraint = constraint;
        }
        LOG.debug("[COMPONENT {}] Placement constraint: {}",
            componentSpec.getName(), constraint.getConstraintExpr().toString());
      }
      ResourceSizing resourceSizing = ResourceSizing.newInstance((int) count,
          resource);
      LOG.debug("[COMPONENT {}] Resource sizing: {}", componentSpec.getName(),
          resourceSizing);
      SchedulingRequest request = SchedulingRequest.newBuilder()
          .priority(priority).allocationRequestId(allocateId)
          .allocationTags(Collections.singleton(componentSpec.getName()))
          .executionType(
              ExecutionTypeRequest.newInstance(ExecutionType.GUARANTEED, true))
          .placementConstraintExpression(finalConstraint)
          .resourceSizing(resourceSizing).build();
      LOG.info("[COMPONENT {}] Submitting scheduling request: {}",
          componentSpec.getName(), request);
      schedulingRequests.add(request);
      amrmClient.addSchedulingRequests(schedulingRequests);
    }
  }

  private void setDesiredContainers(int n) {
    int delta = n - scheduler.getServiceMetrics().containersDesired.value();
    if (delta > 0) {
      scheduler.getServiceMetrics().containersDesired.incr(delta);
    } else {
      scheduler.getServiceMetrics().containersDesired.decr(delta);
    }
    componentMetrics.containersDesired.set(n);
  }


  private void updateMetrics(ContainerStatus status) {
    switch (status.getExitStatus()) {
    case SUCCESS:
      componentMetrics.containersSucceeded.incr();
      scheduler.getServiceMetrics().containersSucceeded.incr();
      return;
    case PREEMPTED:
      componentMetrics.containersPreempted.incr();
      scheduler.getServiceMetrics().containersPreempted.incr();
      break;
    case DISKS_FAILED:
      componentMetrics.containersDiskFailure.incr();
      scheduler.getServiceMetrics().containersDiskFailure.incr();
      break;
    default:
      break;
    }

    // containersFailed include preempted, disks_failed etc.
    componentMetrics.containersFailed.incr();
    scheduler.getServiceMetrics().containersFailed.incr();

    if (Apps.shouldCountTowardsNodeBlacklisting(status.getExitStatus())) {
      String host = scheduler.getLiveInstances().get(status.getContainerId())
          .getNodeId().getHost();
      failureTracker.incNodeFailure(host);
      currentContainerFailure.getAndIncrement() ;
    }
  }

  public boolean areDependenciesReady() {
    List<String> dependencies = componentSpec.getDependencies();
    if (ServiceUtils.isEmpty(dependencies)) {
      return true;
    }
    for (String dependency : dependencies) {
      Component dependentComponent =
          scheduler.getAllComponents().get(dependency);
      if (dependentComponent == null) {
        LOG.error("Couldn't find dependency {} for {} (should never happen)",
            dependency, getName());
        continue;
      }
      if (dependentComponent.getNumReadyInstances() < dependentComponent
          .getNumDesiredInstances()) {
        LOG.info("[COMPONENT {}]: Dependency {} not satisfied, only {} of {}"
                + " instances are ready.", getName(), dependency,
            dependentComponent.getNumReadyInstances(),
            dependentComponent.getNumDesiredInstances());
        return false;
      }
    }
    return true;
  }

  public Map<String, String> getDependencyHostIpTokens() {
    Map<String, String> tokens = new HashMap<>();
    List<String> dependencies = componentSpec.getDependencies();
    if (ServiceUtils.isEmpty(dependencies)) {
      return tokens;
    }
    for (String dependency : dependencies) {
      Collection<ComponentInstance> instances = scheduler.getAllComponents()
          .get(dependency).getAllComponentInstances();
      for (ComponentInstance instance : instances) {
        if (instance.getContainerStatus() == null) {
          continue;
        }
        if (ServiceUtils.isEmpty(instance.getContainerStatus().getIPs()) ||
            ServiceUtils.isUnset(instance.getContainerStatus().getHost())) {
          continue;
        }
        String ip = instance.getContainerStatus().getIPs().get(0);
        String host = instance.getContainerStatus().getHost();
        tokens.put(String.format(COMPONENT_INSTANCE_IP,
            instance.getCompInstanceName().toUpperCase()), ip);
        tokens.put(String.format(COMPONENT_INSTANCE_HOST,
            instance.getCompInstanceName().toUpperCase()), host);
      }
    }
    return tokens;
  }

  public void incRunningContainers() {
    componentMetrics.containersRunning.incr();
    scheduler.getServiceMetrics().containersRunning.incr();
  }

  public void decRunningContainers() {
    componentMetrics.containersRunning.decr();
    scheduler.getServiceMetrics().containersRunning.decr();
  }

  public void incContainersReady() {
    componentMetrics.containersReady.incr();
    scheduler.getServiceMetrics().containersReady.incr();
    checkAndUpdateComponentState(this, true);
  }

  public void decContainersReady() {
    componentMetrics.containersReady.decr();
    scheduler.getServiceMetrics().containersReady.decr();
    checkAndUpdateComponentState(this, false);
  }

  public int getNumReadyInstances() {
    return componentMetrics.containersReady.value();
  }

  public int getNumRunningInstances() {
    return componentMetrics.containersRunning.value();
  }

  public int getNumDesiredInstances() {
    return componentMetrics.containersDesired.value();
  }

  public ComponentInstance getComponentInstance(String componentInstanceName) {
    return compInstances.get(componentInstanceName);
  }

  public Collection<ComponentInstance> getAllComponentInstances() {
    return compInstances.values();
  }

  public org.apache.hadoop.yarn.service.api.records.Component getComponentSpec() {
    return this.componentSpec;
  }

  public void resetCompFailureCount() {
    LOG.info("[COMPONENT {}]: Reset container failure count from {} to 0.",
        getName(), currentContainerFailure.get());
    currentContainerFailure.set(0);
    failureTracker.resetContainerFailures();
  }

  public Probe getProbe() {
    return probe;
  }

  public Priority getPriority() {
    return priority;
  }

  public long getAllocateId() {
    return allocateId;
  }

  public String getName () {
    return componentSpec.getName();
  }

  public ComponentState getState() {
    this.readLock.lock();

    try {
      return this.stateMachine.getCurrentState();
    } finally {
      this.readLock.unlock();
    }
  }
  public ServiceScheduler getScheduler() {
    return scheduler;
  }

  @Override
  public void handle(ComponentEvent event) {
    try {
      writeLock.lock();
      ComponentState oldState = getState();
      try {
        stateMachine.doTransition(event.getType(), event);
      } catch (InvalidStateTransitionException e) {
        LOG.error(MessageFormat.format("[COMPONENT {0}]: Invalid event {1} at {2}",
            componentSpec.getName(), event.getType(), oldState), e);
      }
      if (oldState != getState()) {
        LOG.info("[COMPONENT {}] Transitioned from {} to {} on {} event.",
            componentSpec.getName(), oldState, getState(), event.getType());
      }
    } finally {
      writeLock.unlock();
    }
  }

  private static class BaseTransition implements
      SingleArcTransition<Component, ComponentEvent> {

    @Override public void transition(Component component,
        ComponentEvent event) {
    }
  }

  public ServiceContext getContext() {
    return context;
  }

  // Only for testing
  public List<ComponentInstance> getPendingInstances() {
    return pendingInstances;
  }
}