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

package org.apache.ambari.server.state.svccomphost;

import java.text.MessageFormat;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import org.apache.ambari.server.AmbariException;
import org.apache.ambari.server.agent.AlertDefinitionCommand;
import org.apache.ambari.server.agent.stomp.HostLevelParamsHolder;
import org.apache.ambari.server.agent.stomp.TopologyHolder;
import org.apache.ambari.server.api.services.AmbariMetaInfo;
import org.apache.ambari.server.controller.AmbariManagementController;
import org.apache.ambari.server.controller.ServiceComponentHostRequest;
import org.apache.ambari.server.controller.ServiceComponentHostResponse;
import org.apache.ambari.server.controller.internal.DeleteHostComponentStatusMetaData;
import org.apache.ambari.server.events.AlertHashInvalidationEvent;
import org.apache.ambari.server.events.HostComponentUpdate;
import org.apache.ambari.server.events.HostComponentsUpdateEvent;
import org.apache.ambari.server.events.MaintenanceModeEvent;
import org.apache.ambari.server.events.ServiceComponentInstalledEvent;
import org.apache.ambari.server.events.ServiceComponentUninstalledEvent;
import org.apache.ambari.server.events.StaleConfigsUpdateEvent;
import org.apache.ambari.server.events.TopologyUpdateEvent;
import org.apache.ambari.server.events.publishers.AmbariEventPublisher;
import org.apache.ambari.server.events.publishers.STOMPUpdatePublisher;
import org.apache.ambari.server.orm.dao.HostComponentDesiredStateDAO;
import org.apache.ambari.server.orm.dao.HostComponentStateDAO;
import org.apache.ambari.server.orm.dao.HostDAO;
import org.apache.ambari.server.orm.dao.HostVersionDAO;
import org.apache.ambari.server.orm.dao.RepositoryVersionDAO;
import org.apache.ambari.server.orm.dao.ServiceComponentDesiredStateDAO;
import org.apache.ambari.server.orm.dao.StackDAO;
import org.apache.ambari.server.orm.entities.HostComponentDesiredStateEntity;
import org.apache.ambari.server.orm.entities.HostComponentStateEntity;
import org.apache.ambari.server.orm.entities.HostEntity;
import org.apache.ambari.server.orm.entities.HostVersionEntity;
import org.apache.ambari.server.orm.entities.RepositoryVersionEntity;
import org.apache.ambari.server.orm.entities.ServiceComponentDesiredStateEntity;
import org.apache.ambari.server.orm.entities.StackEntity;
import org.apache.ambari.server.stack.upgrade.RepositoryVersionHelper;
import org.apache.ambari.server.state.BlueprintProvisioningState;
import org.apache.ambari.server.state.Cluster;
import org.apache.ambari.server.state.Clusters;
import org.apache.ambari.server.state.ComponentInfo;
import org.apache.ambari.server.state.ConfigHelper;
import org.apache.ambari.server.state.DesiredConfig;
import org.apache.ambari.server.state.Host;
import org.apache.ambari.server.state.HostComponentAdminState;
import org.apache.ambari.server.state.HostConfig;
import org.apache.ambari.server.state.HostState;
import org.apache.ambari.server.state.MaintenanceState;
import org.apache.ambari.server.state.RepositoryVersionState;
import org.apache.ambari.server.state.ServiceComponent;
import org.apache.ambari.server.state.ServiceComponentHost;
import org.apache.ambari.server.state.ServiceComponentHostEvent;
import org.apache.ambari.server.state.ServiceComponentHostEventType;
import org.apache.ambari.server.state.StackId;
import org.apache.ambari.server.state.StackInfo;
import org.apache.ambari.server.state.State;
import org.apache.ambari.server.state.UpgradeState;
import org.apache.ambari.server.state.alert.AlertDefinitionHash;
import org.apache.ambari.server.state.fsm.InvalidStateTransitionException;
import org.apache.ambari.server.state.fsm.SingleArcTransition;
import org.apache.ambari.server.state.fsm.StateMachine;
import org.apache.ambari.server.state.fsm.StateMachineFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.collect.ImmutableList;
import com.google.common.util.concurrent.Striped;
import com.google.inject.Inject;
import com.google.inject.Provider;
import com.google.inject.assistedinject.Assisted;
import com.google.inject.assistedinject.AssistedInject;
import com.google.inject.persist.Transactional;

public class ServiceComponentHostImpl implements ServiceComponentHost {

  private static final Logger LOG =
      LoggerFactory.getLogger(ServiceComponentHostImpl.class);

  private final ReadWriteLock readWriteLock = new ReentrantReadWriteLock();
  private final Lock writeLock = readWriteLock.writeLock();

  private final ServiceComponent serviceComponent;

  private final Host host;

  private final HostComponentStateDAO hostComponentStateDAO;

  private final HostComponentDesiredStateDAO hostComponentDesiredStateDAO;

  private final HostDAO hostDAO;

  @Inject
  private RepositoryVersionDAO repositoryVersionDAO;

  @Inject
  private HostVersionDAO hostVersionDAO;

  private final ServiceComponentDesiredStateDAO serviceComponentDesiredStateDAO;

  private final Clusters clusters;

  @Inject
  private ConfigHelper helper;

  @Inject
  private AmbariMetaInfo ambariMetaInfo;

  @Inject
  private RepositoryVersionHelper repositoryVersionHelper;

  @Inject
  STOMPUpdatePublisher STOMPUpdatePublisher;

  @Inject
  private Provider<TopologyHolder> m_topologyHolder;

  @Inject
  private Provider<HostLevelParamsHolder> m_hostLevelParamsHolder;

  /**
   * Used for creating commands to send to the agents when alert definitions are
   * added as the result of a service install.
   */
  @Inject
  private AlertDefinitionHash alertDefinitionHash;

  /**
   * Used for topology event creation updates
   */
  @Inject
  private Provider<AmbariManagementController> controller;

  /**
   * Used to publish events relating to service CRUD operations.
   */
  private final AmbariEventPublisher eventPublisher;

  /**
   * Data access object for stack.
   */
  private final StackDAO stackDAO;

  /**
   * The desired component state entity id.
   */
  private final Long desiredStateEntityId;
  /**
   * Cache the generated id for host component state for fast lookups.
   */
  private final Long hostComponentStateId;

  private long lastOpStartTime;
  private long lastOpEndTime;
  private long lastOpLastUpdateTime;
  private AtomicReference<MaintenanceState> maintenanceState = new AtomicReference<>();

  private ConcurrentMap<String, HostConfig> actualConfigs = new ConcurrentHashMap<>();
  private ImmutableList<Map<String, String>> processes = ImmutableList.of();

  /**
   * Used for preventing multiple components on the same host from trying to
   * recalculate versions concurrently.
   */
  private static final Striped<Lock> HOST_VERSION_LOCK = Striped.lazyWeakLock(20);

  /**
   * The name of the host (which should never, ever change)
   */
  private final String hostName;

  private static final StateMachineFactory
  <ServiceComponentHostImpl, State,
  ServiceComponentHostEventType, ServiceComponentHostEvent>
    daemonStateMachineFactory
      = new StateMachineFactory<ServiceComponentHostImpl,
          State, ServiceComponentHostEventType,
          ServiceComponentHostEvent>
          (State.INIT)

  // define the state machine of a HostServiceComponent for runnable
  // components

  .addTransition(State.INIT,
      State.INSTALLING,
      ServiceComponentHostEventType.HOST_SVCCOMP_INSTALL,
      new ServiceComponentHostOpStartedTransition())

  .addTransition(State.INSTALLING,
      State.INSTALLED,
      ServiceComponentHostEventType.HOST_SVCCOMP_OP_SUCCEEDED,
      new ServiceComponentHostOpCompletedTransition())

  .addTransition(State.INSTALLING,
      State.INSTALLED,
      ServiceComponentHostEventType.HOST_SVCCOMP_OP_SUCCEEDED,
      new AlertDefinitionCommandTransition())

  .addTransition(State.INSTALLED,
      State.INSTALLED,
      ServiceComponentHostEventType.HOST_SVCCOMP_OP_SUCCEEDED,
      new ServiceComponentHostOpCompletedTransition())

  // Allow transition on abort
  .addTransition(State.INSTALLED,
      State.INSTALLED,
      ServiceComponentHostEventType.HOST_SVCCOMP_OP_FAILED,
      new ServiceComponentHostOpCompletedTransition())

  .addTransition(State.INSTALLING,
      State.INSTALLING,
      ServiceComponentHostEventType.HOST_SVCCOMP_OP_IN_PROGRESS,
      new ServiceComponentHostOpInProgressTransition())

  .addTransition(State.INSTALLING,
      State.INSTALLING,
      ServiceComponentHostEventType.HOST_SVCCOMP_INSTALL,
      new ServiceComponentHostOpStartedTransition())

  .addTransition(State.INSTALLING,
      State.INSTALL_FAILED,
      ServiceComponentHostEventType.HOST_SVCCOMP_OP_FAILED,
      new ServiceComponentHostOpCompletedTransition())

  .addTransition(State.INSTALL_FAILED,
      State.INSTALLING,
      ServiceComponentHostEventType.HOST_SVCCOMP_OP_RESTART,
      new ServiceComponentHostOpStartedTransition())

  .addTransition(State.INSTALL_FAILED,
      State.INSTALLING,
      ServiceComponentHostEventType.HOST_SVCCOMP_INSTALL,
      new ServiceComponentHostOpStartedTransition())

  // Allow transition on abort
  .addTransition(State.INSTALL_FAILED,
      State.INSTALL_FAILED,
      ServiceComponentHostEventType.HOST_SVCCOMP_OP_FAILED,
      new ServiceComponentHostOpCompletedTransition())

  .addTransition(State.INSTALLED,
      State.STARTING,
      ServiceComponentHostEventType.HOST_SVCCOMP_START,
      new ServiceComponentHostOpStartedTransition())

  .addTransition(State.INSTALLED,
      State.UNINSTALLING,
      ServiceComponentHostEventType.HOST_SVCCOMP_UNINSTALL,
      new ServiceComponentHostOpStartedTransition())

  .addTransition(State.INSTALLED,
      State.INSTALLING,
      ServiceComponentHostEventType.HOST_SVCCOMP_INSTALL,
      new ServiceComponentHostOpStartedTransition())

  .addTransition(State.INSTALLED,
      State.STOPPING,
      ServiceComponentHostEventType.HOST_SVCCOMP_STOP,
      new ServiceComponentHostOpStartedTransition())

  .addTransition(State.INSTALLED,
      State.UPGRADING,
      ServiceComponentHostEventType.HOST_SVCCOMP_UPGRADE,
      new ServiceComponentHostOpStartedTransition())

  .addTransition(State.INSTALLED,
      State.INSTALLED,
      ServiceComponentHostEventType.HOST_SVCCOMP_OP_IN_PROGRESS,
      new ServiceComponentHostOpInProgressTransition())

  .addTransition(State.INSTALLED,
      State.STARTED,
      ServiceComponentHostEventType.HOST_SVCCOMP_STARTED,
      new ServiceComponentHostOpCompletedTransition())

  .addTransition(State.INSTALLED,
      State.INSTALLED,
      ServiceComponentHostEventType.HOST_SVCCOMP_STOPPED,
      new ServiceComponentHostOpCompletedTransition())

  .addTransition(State.STARTING,
      State.STARTING,
      ServiceComponentHostEventType.HOST_SVCCOMP_OP_IN_PROGRESS,
      new ServiceComponentHostOpInProgressTransition())

  .addTransition(State.STARTING,
      State.STARTING,
      ServiceComponentHostEventType.HOST_SVCCOMP_START,
      new ServiceComponentHostOpStartedTransition())

  .addTransition(State.STARTING,
      State.STARTED,
      ServiceComponentHostEventType.HOST_SVCCOMP_STARTED,
      new ServiceComponentHostOpCompletedTransition())

  .addTransition(State.STARTING,
      State.INSTALLED,
      ServiceComponentHostEventType.HOST_SVCCOMP_OP_FAILED,
      new ServiceComponentHostOpCompletedTransition())

  .addTransition(State.INSTALLED,
      State.STARTING,
      ServiceComponentHostEventType.HOST_SVCCOMP_OP_RESTART,
      new ServiceComponentHostOpStartedTransition())

  .addTransition(State.STARTED,
      State.STARTED,
      ServiceComponentHostEventType.HOST_SVCCOMP_STARTED,
      new ServiceComponentHostOpCompletedTransition())

  .addTransition(State.STARTED,
      State.STOPPING,
      ServiceComponentHostEventType.HOST_SVCCOMP_STOP,
      new ServiceComponentHostOpStartedTransition())

  .addTransition(State.STARTED,
      State.STARTED,
      ServiceComponentHostEventType.HOST_SVCCOMP_OP_IN_PROGRESS,
      new ServiceComponentHostOpInProgressTransition())

  // Allow transition on abort
  .addTransition(State.STARTED,
      State.STARTED,
      ServiceComponentHostEventType.HOST_SVCCOMP_OP_FAILED,
      new ServiceComponentHostOpCompletedTransition())

  .addTransition(State.STARTED,
      State.INSTALLED,
      ServiceComponentHostEventType.HOST_SVCCOMP_STOPPED,
      new ServiceComponentHostOpCompletedTransition())

  .addTransition(State.STOPPING,
      State.STOPPING,
      ServiceComponentHostEventType.HOST_SVCCOMP_OP_IN_PROGRESS,
      new ServiceComponentHostOpInProgressTransition())

  .addTransition(State.STOPPING,
      State.INSTALLED,
      ServiceComponentHostEventType.HOST_SVCCOMP_STOPPED,
      new ServiceComponentHostOpCompletedTransition())

  .addTransition(State.STOPPING,
      State.STARTED,
      ServiceComponentHostEventType.HOST_SVCCOMP_OP_FAILED,
      new ServiceComponentHostOpCompletedTransition())

  .addTransition(State.STARTED,
      State.STOPPING,
      ServiceComponentHostEventType.HOST_SVCCOMP_OP_RESTART,
      new ServiceComponentHostOpStartedTransition())

  .addTransition(State.UNINSTALLING,
      State.UNINSTALLING,
      ServiceComponentHostEventType.HOST_SVCCOMP_OP_IN_PROGRESS,
      new ServiceComponentHostOpInProgressTransition())

  .addTransition(State.UNINSTALLING,
      State.UNINSTALLED,
      ServiceComponentHostEventType.HOST_SVCCOMP_OP_SUCCEEDED,
      new ServiceComponentHostOpCompletedTransition())

  .addTransition(State.UNINSTALLING,
      State.UNINSTALLING,
      ServiceComponentHostEventType.HOST_SVCCOMP_OP_FAILED,
      new ServiceComponentHostOpCompletedTransition())

  .addTransition(State.UPGRADING,
      State.UPGRADING,
      ServiceComponentHostEventType.HOST_SVCCOMP_OP_IN_PROGRESS,
      new ServiceComponentHostOpInProgressTransition())

  .addTransition(State.UPGRADING,
      State.INSTALLED,
      ServiceComponentHostEventType.HOST_SVCCOMP_OP_SUCCEEDED,
      new ServiceComponentHostOpCompletedTransition())

  .addTransition(State.UPGRADING,
      State.UPGRADING,
      ServiceComponentHostEventType.HOST_SVCCOMP_OP_FAILED,
      new ServiceComponentHostOpCompletedTransition())

  .addTransition(State.UPGRADING,
      State.UPGRADING,
      ServiceComponentHostEventType.HOST_SVCCOMP_UPGRADE,
      new ServiceComponentHostOpStartedTransition())

  .addTransition(State.UPGRADING,
      State.UPGRADING,
      ServiceComponentHostEventType.HOST_SVCCOMP_UPGRADE,
      new ServiceComponentHostOpInProgressTransition())

  .addTransition(State.UNINSTALLING,
      State.UNINSTALLING,
      ServiceComponentHostEventType.HOST_SVCCOMP_OP_RESTART,
      new ServiceComponentHostOpStartedTransition())

  .addTransition(State.UNINSTALLING,
      State.UNINSTALLING,
      ServiceComponentHostEventType.HOST_SVCCOMP_UNINSTALL,
      new ServiceComponentHostOpStartedTransition())

  .addTransition(State.UNINSTALLED,
      State.INSTALLING,
      ServiceComponentHostEventType.HOST_SVCCOMP_INSTALL,
      new ServiceComponentHostOpStartedTransition())

  .addTransition(State.UNINSTALLED,
      State.WIPING_OUT,
      ServiceComponentHostEventType.HOST_SVCCOMP_WIPEOUT,
      new ServiceComponentHostOpStartedTransition())

  .addTransition(State.WIPING_OUT,
      State.WIPING_OUT,
      ServiceComponentHostEventType.HOST_SVCCOMP_OP_IN_PROGRESS,
      new ServiceComponentHostOpInProgressTransition())

  .addTransition(State.WIPING_OUT,
      State.INIT,
      ServiceComponentHostEventType.HOST_SVCCOMP_OP_SUCCEEDED,
      new ServiceComponentHostOpCompletedTransition())

  .addTransition(State.WIPING_OUT,
      State.WIPING_OUT,
      ServiceComponentHostEventType.HOST_SVCCOMP_OP_FAILED,
      new ServiceComponentHostOpCompletedTransition())

  .addTransition(State.WIPING_OUT,
      State.WIPING_OUT,
      ServiceComponentHostEventType.HOST_SVCCOMP_OP_RESTART,
      new ServiceComponentHostOpStartedTransition())

  .addTransition(State.WIPING_OUT,
      State.WIPING_OUT,
      ServiceComponentHostEventType.HOST_SVCCOMP_WIPEOUT,
      new ServiceComponentHostOpStartedTransition())

  .addTransition(State.INSTALLED,
      State.DISABLED,
      ServiceComponentHostEventType.HOST_SVCCOMP_DISABLE,
      new ServiceComponentHostOpCompletedTransition())

  .addTransition(State.DISABLED,
      State.DISABLED,
      ServiceComponentHostEventType.HOST_SVCCOMP_DISABLE,
      new ServiceComponentHostOpCompletedTransition())

  .addTransition(State.UNKNOWN,
      State.DISABLED,
      ServiceComponentHostEventType.HOST_SVCCOMP_DISABLE,
      new ServiceComponentHostOpCompletedTransition())

  .addTransition(State.UNKNOWN,
      State.UNKNOWN,
      ServiceComponentHostEventType.HOST_SVCCOMP_OP_IN_PROGRESS,
      new ServiceComponentHostOpCompletedTransition())

  .addTransition(State.INSTALL_FAILED,
      State.DISABLED,
      ServiceComponentHostEventType.HOST_SVCCOMP_DISABLE,
      new ServiceComponentHostOpCompletedTransition())

  .addTransition(State.DISABLED,
      State.INSTALLED,
      ServiceComponentHostEventType.HOST_SVCCOMP_RESTORE,
      new ServiceComponentHostOpCompletedTransition())

     .installTopology();

  private static final StateMachineFactory
  <ServiceComponentHostImpl, State,
  ServiceComponentHostEventType, ServiceComponentHostEvent>
    clientStateMachineFactory
      = new StateMachineFactory<ServiceComponentHostImpl,
          State, ServiceComponentHostEventType,
          ServiceComponentHostEvent>
          (State.INIT)

  // define the state machine of a HostServiceComponent for client only
  // components

     .addTransition(State.INIT,
         State.INSTALLING,
         ServiceComponentHostEventType.HOST_SVCCOMP_INSTALL,
         new ServiceComponentHostOpStartedTransition())

     .addTransition(State.INSTALLING,
         State.INSTALLED,
         ServiceComponentHostEventType.HOST_SVCCOMP_OP_SUCCEEDED,
         new ServiceComponentHostOpCompletedTransition())

     .addTransition(State.INSTALLING,
         State.INSTALLING,
         ServiceComponentHostEventType.HOST_SVCCOMP_INSTALL,
         new ServiceComponentHostOpStartedTransition())

     .addTransition(State.INSTALLING,
         State.INSTALLING,
         ServiceComponentHostEventType.HOST_SVCCOMP_OP_IN_PROGRESS,
         new ServiceComponentHostOpInProgressTransition())

     .addTransition(State.INSTALLED,
         State.INSTALLED,
         ServiceComponentHostEventType.HOST_SVCCOMP_OP_IN_PROGRESS,
         new ServiceComponentHostOpInProgressTransition())

     // Allow transition on abort
     .addTransition(State.INSTALLED,
         State.INSTALLED,
         ServiceComponentHostEventType.HOST_SVCCOMP_OP_FAILED,
         new ServiceComponentHostOpCompletedTransition())

     .addTransition(State.INSTALLING,
         State.INSTALL_FAILED,
         ServiceComponentHostEventType.HOST_SVCCOMP_OP_FAILED,
         new ServiceComponentHostOpCompletedTransition())

     .addTransition(State.INSTALL_FAILED,
         State.INSTALLING,
         ServiceComponentHostEventType.HOST_SVCCOMP_OP_RESTART,
         new ServiceComponentHostOpStartedTransition())
     .addTransition(State.INSTALL_FAILED,
         State.INSTALLING,
         ServiceComponentHostEventType.HOST_SVCCOMP_INSTALL,
         new ServiceComponentHostOpStartedTransition())
      // Allow transition on abort
     .addTransition(State.INSTALL_FAILED, State.INSTALL_FAILED,
         ServiceComponentHostEventType.HOST_SVCCOMP_OP_FAILED,
         new ServiceComponentHostOpCompletedTransition())

     .addTransition(State.INSTALL_FAILED,
         State.INSTALL_FAILED,
         ServiceComponentHostEventType.HOST_SVCCOMP_OP_IN_PROGRESS,
         new ServiceComponentHostOpInProgressTransition())

    .addTransition(State.INSTALLED,
         State.INSTALLED,
         ServiceComponentHostEventType.HOST_SVCCOMP_OP_SUCCEEDED,
         new ServiceComponentHostOpCompletedTransition())
     .addTransition(State.INSTALLED,
         State.UNINSTALLING,
         ServiceComponentHostEventType.HOST_SVCCOMP_UNINSTALL,
         new ServiceComponentHostOpStartedTransition())
     .addTransition(State.INSTALLED,
         State.INSTALLING,
         ServiceComponentHostEventType.HOST_SVCCOMP_INSTALL,
         new ServiceComponentHostOpStartedTransition())
    .addTransition(State.INSTALLED,
         State.UPGRADING,
         ServiceComponentHostEventType.HOST_SVCCOMP_UPGRADE,
         new ServiceComponentHostOpStartedTransition())

     .addTransition(State.UPGRADING,
         State.UPGRADING,
         ServiceComponentHostEventType.HOST_SVCCOMP_OP_IN_PROGRESS,
         new ServiceComponentHostOpInProgressTransition())
     .addTransition(State.UPGRADING,
         State.INSTALLED,
         ServiceComponentHostEventType.HOST_SVCCOMP_OP_SUCCEEDED,
         new ServiceComponentHostOpCompletedTransition())
     .addTransition(State.UPGRADING,
         State.UPGRADING,
         ServiceComponentHostEventType.HOST_SVCCOMP_OP_FAILED,
         new ServiceComponentHostOpCompletedTransition())
     .addTransition(State.UPGRADING,
         State.UPGRADING,
         ServiceComponentHostEventType.HOST_SVCCOMP_UPGRADE,
         new ServiceComponentHostOpStartedTransition())
     .addTransition(State.UPGRADING,
         State.UPGRADING,
         ServiceComponentHostEventType.HOST_SVCCOMP_UPGRADE,
         new ServiceComponentHostOpInProgressTransition())

     .addTransition(State.UNINSTALLING,
         State.UNINSTALLING,
         ServiceComponentHostEventType.HOST_SVCCOMP_OP_IN_PROGRESS,
         new ServiceComponentHostOpInProgressTransition())
     .addTransition(State.UNINSTALLING,
         State.UNINSTALLED,
         ServiceComponentHostEventType.HOST_SVCCOMP_OP_SUCCEEDED,
         new ServiceComponentHostOpCompletedTransition())
     .addTransition(State.UNINSTALLING,
         State.UNINSTALLING,
         ServiceComponentHostEventType.HOST_SVCCOMP_OP_FAILED,
         new ServiceComponentHostOpCompletedTransition())

     .addTransition(State.UNINSTALLING,
         State.UNINSTALLING,
         ServiceComponentHostEventType.HOST_SVCCOMP_OP_RESTART,
         new ServiceComponentHostOpStartedTransition())
     .addTransition(State.UNINSTALLING,
         State.UNINSTALLING,
         ServiceComponentHostEventType.HOST_SVCCOMP_UNINSTALL,
         new ServiceComponentHostOpStartedTransition())

     .addTransition(State.UNINSTALLED,
         State.INSTALLING,
         ServiceComponentHostEventType.HOST_SVCCOMP_INSTALL,
         new ServiceComponentHostOpStartedTransition())

     .addTransition(State.UNINSTALLED,
         State.WIPING_OUT,
         ServiceComponentHostEventType.HOST_SVCCOMP_WIPEOUT,
         new ServiceComponentHostOpStartedTransition())

     .addTransition(State.WIPING_OUT,
         State.WIPING_OUT,
         ServiceComponentHostEventType.HOST_SVCCOMP_OP_IN_PROGRESS,
         new ServiceComponentHostOpInProgressTransition())
     .addTransition(State.WIPING_OUT,
         State.INIT,
         ServiceComponentHostEventType.HOST_SVCCOMP_OP_SUCCEEDED,
         new ServiceComponentHostOpCompletedTransition())
     .addTransition(State.WIPING_OUT,
         State.WIPING_OUT,
         ServiceComponentHostEventType.HOST_SVCCOMP_OP_FAILED,
         new ServiceComponentHostOpCompletedTransition())

     .addTransition(State.WIPING_OUT,
         State.WIPING_OUT,
         ServiceComponentHostEventType.HOST_SVCCOMP_OP_RESTART,
         new ServiceComponentHostOpStartedTransition())
     .addTransition(State.WIPING_OUT,
         State.WIPING_OUT,
         ServiceComponentHostEventType.HOST_SVCCOMP_WIPEOUT,
         new ServiceComponentHostOpStartedTransition())

     .installTopology();

  private final StateMachine<State,
      ServiceComponentHostEventType, ServiceComponentHostEvent> stateMachine;

  static class ServiceComponentHostOpCompletedTransition
     implements SingleArcTransition<ServiceComponentHostImpl,
         ServiceComponentHostEvent> {

    @Override
    public void transition(ServiceComponentHostImpl impl,
        ServiceComponentHostEvent event) {
      // TODO Audit logs
      impl.updateLastOpInfo(event.getType(), event.getOpTimestamp());

    }
  }

  /**
   * The {@link AlertDefinitionCommandTransition} is used to capture the
   * transition from {@link State#INSTALLING} to {@link State#INSTALLED} so that
   * the host affected will have new {@link AlertDefinitionCommand}s pushed to
   * it.
   */
  static class AlertDefinitionCommandTransition implements
      SingleArcTransition<ServiceComponentHostImpl, ServiceComponentHostEvent> {

    @Override
    public void transition(ServiceComponentHostImpl impl,
        ServiceComponentHostEvent event) {
      if (event.getType() != ServiceComponentHostEventType.HOST_SVCCOMP_OP_SUCCEEDED) {
        return;
      }

      // invalidate the host
      String hostName = impl.getHostName();
      impl.alertDefinitionHash.invalidate(impl.getClusterName(), hostName);

      // publish the event
      AlertHashInvalidationEvent hashInvalidationEvent = new AlertHashInvalidationEvent(
          impl.getClusterId(), Collections.singletonList(hostName));

      impl.eventPublisher.publish(hashInvalidationEvent);
      impl.updateLastOpInfo(event.getType(), event.getOpTimestamp());

    }
  }

  static class ServiceComponentHostOpStartedTransition
    implements SingleArcTransition<ServiceComponentHostImpl,
        ServiceComponentHostEvent> {

    @Override
    public void transition(ServiceComponentHostImpl impl,
        ServiceComponentHostEvent event) {
      // TODO Audit logs
      // FIXME handle restartOp event
      impl.updateLastOpInfo(event.getType(), event.getOpTimestamp());
      if (event.getType() ==
          ServiceComponentHostEventType.HOST_SVCCOMP_INSTALL) {
        ServiceComponentHostInstallEvent e =
            (ServiceComponentHostInstallEvent) event;
        if (LOG.isDebugEnabled()) {
          LOG.debug("Updating live stack version during INSTALL event, new stack version={}", e.getStackId());
        }
      }
    }
  }

  static class ServiceComponentHostOpInProgressTransition
    implements SingleArcTransition<ServiceComponentHostImpl,
        ServiceComponentHostEvent> {

    @Override
    public void transition(ServiceComponentHostImpl impl,
        ServiceComponentHostEvent event) {
      // TODO Audit logs
      impl.updateLastOpInfo(event.getType(), event.getOpTimestamp());
    }
  }

  private void resetLastOpInfo() {
    setLastOpStartTime(-1);
    setLastOpLastUpdateTime(-1);
    setLastOpEndTime(-1);
  }

  private void updateLastOpInfo(ServiceComponentHostEventType eventType,
      long time) {
    try {
      writeLock.lock();
      switch (eventType) {
        case HOST_SVCCOMP_INSTALL:
        case HOST_SVCCOMP_START:
        case HOST_SVCCOMP_STOP:
        case HOST_SVCCOMP_UNINSTALL:
        case HOST_SVCCOMP_WIPEOUT:
        case HOST_SVCCOMP_OP_RESTART:
          resetLastOpInfo();
          setLastOpStartTime(time);
          break;
        case HOST_SVCCOMP_OP_FAILED:
        case HOST_SVCCOMP_OP_SUCCEEDED:
        case HOST_SVCCOMP_STOPPED:
        case HOST_SVCCOMP_STARTED:
          setLastOpLastUpdateTime(time);
          setLastOpEndTime(time);
          break;
        case HOST_SVCCOMP_OP_IN_PROGRESS:
          setLastOpLastUpdateTime(time);
          break;
      }
    } finally {
      writeLock.unlock();
    }
  }

  @AssistedInject
  public ServiceComponentHostImpl(@Assisted ServiceComponent serviceComponent,
      @Assisted String hostName, @Assisted ServiceComponentDesiredStateEntity serviceComponentDesiredStateEntity,
      Clusters clusters, StackDAO stackDAO, HostDAO hostDAO,
      ServiceComponentDesiredStateDAO serviceComponentDesiredStateDAO,
      HostComponentStateDAO hostComponentStateDAO,
      HostComponentDesiredStateDAO hostComponentDesiredStateDAO,
      AmbariEventPublisher eventPublisher) {

    this.serviceComponent = serviceComponent;
    this.hostName = hostName;
    this.clusters = clusters;
    this.stackDAO = stackDAO;
    this.hostDAO = hostDAO;
    this.serviceComponentDesiredStateDAO = serviceComponentDesiredStateDAO;
    this.hostComponentStateDAO = hostComponentStateDAO;
    this.hostComponentDesiredStateDAO = hostComponentDesiredStateDAO;
    this.eventPublisher = eventPublisher;

    if (serviceComponent.isClientComponent()) {
      stateMachine = clientStateMachineFactory.make(this);
    } else {
      stateMachine = daemonStateMachineFactory.make(this);
    }

    HostEntity hostEntity = null;
    try {
      host = clusters.getHost(hostName);
      hostEntity = hostDAO.findByName(hostName);
      if (hostEntity == null) {
        throw new AmbariException("Could not find host " + hostName);
      }
    } catch (AmbariException e) {
      LOG.error("Host '{}' was not found" + hostName);
      throw new RuntimeException(e);
    }

    StackId stackId = serviceComponent.getDesiredStackId();
    StackEntity stackEntity = stackDAO.find(stackId.getStackName(),
        stackId.getStackVersion());

    HostComponentStateEntity stateEntity = new HostComponentStateEntity();
    stateEntity.setClusterId(serviceComponent.getClusterId());
    stateEntity.setComponentName(serviceComponent.getName());
    stateEntity.setServiceName(serviceComponent.getServiceName());
    stateEntity.setVersion(State.UNKNOWN.toString());
    stateEntity.setHostEntity(hostEntity);
    stateEntity.setCurrentState(stateMachine.getCurrentState());
    stateEntity.setUpgradeState(UpgradeState.NONE);

    HostComponentDesiredStateEntity desiredStateEntity = new HostComponentDesiredStateEntity();
    desiredStateEntity.setClusterId(serviceComponent.getClusterId());
    desiredStateEntity.setComponentName(serviceComponent.getName());
    desiredStateEntity.setServiceName(serviceComponent.getServiceName());
    desiredStateEntity.setHostEntity(hostEntity);
    desiredStateEntity.setDesiredState(State.INIT);

    if(!serviceComponent.isMasterComponent() && !serviceComponent.isClientComponent()) {
      desiredStateEntity.setAdminState(HostComponentAdminState.INSERVICE);
    } else {
      desiredStateEntity.setAdminState(null);
    }

    persistEntities(hostEntity, stateEntity, desiredStateEntity, serviceComponentDesiredStateEntity);

    // publish the service component installed event
    ServiceComponentInstalledEvent event = new ServiceComponentInstalledEvent(getClusterId(),
        stackId.getStackName(), stackId.getStackVersion(), getServiceName(),
        getServiceComponentName(), getHostName(), isRecoveryEnabled(), serviceComponent.isMasterComponent());

    eventPublisher.publish(event);

    desiredStateEntityId = desiredStateEntity.getId();
    hostComponentStateId = stateEntity.getId();

    resetLastOpInfo();
  }
  @AssistedInject
  public ServiceComponentHostImpl(@Assisted ServiceComponent serviceComponent,
      @Assisted String hostName,
      Clusters clusters, StackDAO stackDAO, HostDAO hostDAO,
      ServiceComponentDesiredStateDAO serviceComponentDesiredStateDAO,
      HostComponentStateDAO hostComponentStateDAO,
      HostComponentDesiredStateDAO hostComponentDesiredStateDAO,
      AmbariEventPublisher eventPublisher) {
      this(serviceComponent, hostName, null, clusters, stackDAO, hostDAO,
          serviceComponentDesiredStateDAO, hostComponentStateDAO, hostComponentDesiredStateDAO, eventPublisher);
  }

  @AssistedInject
  public ServiceComponentHostImpl(@Assisted ServiceComponent serviceComponent,
      @Assisted HostComponentStateEntity stateEntity,
      @Assisted HostComponentDesiredStateEntity desiredStateEntity, Clusters clusters,
      StackDAO stackDAO, HostDAO hostDAO,
      ServiceComponentDesiredStateDAO serviceComponentDesiredStateDAO,
      HostComponentStateDAO hostComponentStateDAO,
