/*
 * 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.agent;


import static org.apache.ambari.server.controller.KerberosHelperImpl.CHECK_KEYTABS;
import static org.apache.ambari.server.controller.KerberosHelperImpl.SET_KEYTAB;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;

import org.apache.ambari.server.AmbariException;
import org.apache.ambari.server.Role;
import org.apache.ambari.server.RoleCommand;
import org.apache.ambari.server.ServiceComponentHostNotFoundException;
import org.apache.ambari.server.ServiceComponentNotFoundException;
import org.apache.ambari.server.ServiceNotFoundException;
import org.apache.ambari.server.actionmanager.ActionManager;
import org.apache.ambari.server.actionmanager.HostRoleCommand;
import org.apache.ambari.server.actionmanager.HostRoleStatus;
import org.apache.ambari.server.agent.ExecutionCommand.KeyNames;
import org.apache.ambari.server.agent.stomp.dto.ComponentVersionReport;
import org.apache.ambari.server.agent.stomp.dto.ComponentVersionReports;
import org.apache.ambari.server.api.services.AmbariMetaInfo;
import org.apache.ambari.server.controller.MaintenanceStateHelper;
import org.apache.ambari.server.events.ActionFinalReportReceivedEvent;
import org.apache.ambari.server.events.AlertEvent;
import org.apache.ambari.server.events.AlertReceivedEvent;
import org.apache.ambari.server.events.HostComponentVersionAdvertisedEvent;
import org.apache.ambari.server.events.publishers.AlertEventPublisher;
import org.apache.ambari.server.events.publishers.AmbariEventPublisher;
import org.apache.ambari.server.events.publishers.VersionEventPublisher;
import org.apache.ambari.server.metadata.ActionMetadata;
import org.apache.ambari.server.orm.dao.KerberosKeytabDAO;
import org.apache.ambari.server.orm.dao.KerberosKeytabPrincipalDAO;
import org.apache.ambari.server.orm.entities.KerberosKeytabPrincipalEntity;
import org.apache.ambari.server.stack.upgrade.Direction;
import org.apache.ambari.server.state.Alert;
import org.apache.ambari.server.state.Cluster;
import org.apache.ambari.server.state.Clusters;
import org.apache.ambari.server.state.Host;
import org.apache.ambari.server.state.HostHealthStatus;
import org.apache.ambari.server.state.HostState;
import org.apache.ambari.server.state.Service;
import org.apache.ambari.server.state.ServiceComponent;
import org.apache.ambari.server.state.ServiceComponentHost;
import org.apache.ambari.server.state.UpgradeState;
import org.apache.ambari.server.state.fsm.InvalidStateTransitionException;
import org.apache.ambari.server.state.host.HostStatusUpdatesReceivedEvent;
import org.apache.ambari.server.state.scheduler.RequestExecution;
import org.apache.ambari.server.state.svccomphost.ServiceComponentHostOpFailedEvent;
import org.apache.ambari.server.state.svccomphost.ServiceComponentHostOpInProgressEvent;
import org.apache.ambari.server.state.svccomphost.ServiceComponentHostOpSucceededEvent;
import org.apache.ambari.server.state.svccomphost.ServiceComponentHostStartedEvent;
import org.apache.ambari.server.state.svccomphost.ServiceComponentHostStoppedEvent;
import org.apache.ambari.spi.upgrade.UpgradeType;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.util.concurrent.AbstractService;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.google.gson.Gson;
import com.google.gson.JsonSyntaxException;
import com.google.gson.annotations.SerializedName;
import com.google.inject.Inject;
import com.google.inject.Injector;

/**
 * HeartbeatProcessor class is used for bulk processing data retrieved from agents in background
 */
public class HeartbeatProcessor extends AbstractService{
  private static final Logger LOG = LoggerFactory.getLogger(HeartbeatProcessor.class);

  private ScheduledExecutorService executor;

  private ConcurrentLinkedQueue<HeartBeat> heartBeatsQueue = new ConcurrentLinkedQueue<>();

  private volatile boolean shouldRun = true;

  //TODO rewrite to correlate with heartbeat frequency, hardcoded in agent as of now
  private long delay = 5000;
  private long period = 1000;

  private int poolSize = 1;

  private Clusters clusterFsm;
  private HeartbeatMonitor heartbeatMonitor;
  private Injector injector;
  private ActionManager actionManager;

  /**
   * Publishes {@link AlertEvent} instances.
   */
  @Inject
  AlertEventPublisher alertEventPublisher;

  @Inject
  AmbariEventPublisher ambariEventPublisher;

  @Inject
  VersionEventPublisher versionEventPublisher;

  @Inject
  ActionMetadata actionMetadata;

  @Inject
  MaintenanceStateHelper maintenanceStateHelper;

  @Inject
  AmbariMetaInfo ambariMetaInfo;

  @Inject
  KerberosKeytabPrincipalDAO kerberosKeytabPrincipalDAO;

  @Inject
  KerberosKeytabDAO kerberosKeytabDAO;

  @Inject
  Gson gson;

  @Inject
  public HeartbeatProcessor(Clusters clusterFsm, ActionManager am, HeartbeatMonitor heartbeatMonitor,
                            Injector injector) {
    injector.injectMembers(this);

    this.injector = injector;
    this.heartbeatMonitor = heartbeatMonitor;
    this.clusterFsm = clusterFsm;
    actionManager = am;
    ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("ambari-heartbeat-processor-%d").build();
    executor = Executors.newScheduledThreadPool(poolSize, threadFactory);
  }

  @Override
  protected void doStart() {
    LOG.info("**** Starting heartbeats processing threads ****");
    for (int i = 0; i < poolSize; i++) {
      executor.scheduleAtFixedRate(new HeartbeatProcessingTask(), delay, period, TimeUnit.MILLISECONDS);
    }
  }

  @Override
  protected void doStop() {
    LOG.info("**** Stopping heartbeats processing threads ****");
    shouldRun = false;
    executor.shutdown();
  }

  public void addHeartbeat(HeartBeat heartBeat) {
    heartBeatsQueue.add(heartBeat);
  }

  private HeartBeat pollHeartbeat() {
    return heartBeatsQueue.poll();
  }

  /**
   * Processing task to be scheduled for execution
   */
  private class HeartbeatProcessingTask implements Runnable {

    @Override
    public void run() {
      while (shouldRun) {
        try {
          HeartBeat heartbeat = pollHeartbeat();
          if (heartbeat == null) {
            break;
          }
          processHeartbeat(heartbeat);
        } catch (Exception e) {
          LOG.error("Exception received while processing heartbeat", e);
        } catch (Throwable throwable) {
          //catch everything to prevent task suppression
          LOG.error("ERROR: ", throwable);
        }


      }
    }
  }

  /**
   * Incapsulates logic for processing data from agent heartbeat
   *
   * @param heartbeat Agent heartbeat object
   * @throws AmbariException
   */
  public void processHeartbeat(HeartBeat heartbeat) throws AmbariException {
    long now = System.currentTimeMillis();

    processAlerts(heartbeat);

    //process status reports before command reports to prevent status override immediately after task finish
    processStatusReports(heartbeat);
    processCommandReports(heartbeat, now);
    //host status calculation are based on task and status reports, should be performed last
    processHostStatus(heartbeat);
  }


  /**
   * Extracts all of the {@link Alert}s from the heartbeat and fires
   * {@link AlertEvent}s for each one. If there is a problem looking up the
   * cluster, then alerts will not be processed.
   *
   * @param heartbeat the heartbeat to process.
   */
  protected void processAlerts(HeartBeat heartbeat) {
    if (heartbeat != null) {
      processAlerts(heartbeat.getHostname(), heartbeat.getAlerts());
    }
  }

  public void processAlerts(String hostname, List<Alert> alerts) {
    if (alerts != null && !alerts.isEmpty()) {
      for (Alert alert : alerts) {
        if (alert.getHostName() == null) {
          alert.setHostName(hostname);
        }
      }
      AlertEvent event = new AlertReceivedEvent(alerts);
      alertEventPublisher.publish(event);

    }
  }

  /**
   * Update host status basing on components statuses
   *
   * @param heartbeat heartbeat to process
   * @throws AmbariException
   */
  protected void processHostStatus(HeartBeat heartbeat) throws AmbariException {
    processHostStatus(heartbeat.getComponentStatus(), heartbeat.getReports(), heartbeat.getHostname());
  }


  protected void processHostStatus(List<ComponentStatus> componentStatuses, List<CommandReport> reports,
                                   String hostName) throws AmbariException {

    Host host = clusterFsm.getHost(hostName);
    HostHealthStatus.HealthStatus healthStatus = host.getHealthStatus().getHealthStatus();

    if (!healthStatus.equals(HostHealthStatus.HealthStatus.UNKNOWN)) {

      //Host status info could be calculated only if agent returned statuses in heartbeat
      //Or, if a command is executed that can change component status
      boolean calculateHostStatus = false;
      Long clusterId = null;
      if (CollectionUtils.isNotEmpty(componentStatuses)) {
        calculateHostStatus = true;
        clusterId = componentStatuses.stream()
            .findFirst()
            .map(ComponentStatus::getClusterId)
            .orElse(null);
      }

      if (!calculateHostStatus && CollectionUtils.isNotEmpty(reports)) {
        for (CommandReport report : reports) {
          if (RoleCommand.ACTIONEXECUTE.toString().equals(report.getRoleCommand())) {
            continue;
          }

          String service = report.getServiceName();
          if (actionMetadata.getActions(service.toLowerCase()).contains(report.getRole())) {
            continue;
          }
          if (report.getStatus().equals("COMPLETED")) {
            calculateHostStatus = true;
            clusterId = Long.parseLong(report.getClusterId());
            break;
          }
        }
      }

      if (calculateHostStatus) {
        host.calculateHostStatus(clusterId);
      }

      //If host doesn't belong to any cluster
      if ((clusterFsm.getClustersForHost(host.getHostName())).size() == 0) {
        healthStatus = HostHealthStatus.HealthStatus.HEALTHY;
        host.setStatus(healthStatus.name());
      }
    }
  }

  /**
   * Process reports of tasks executed on agents
   *
   * @param heartbeat heartbeat to process
   * @param now       cached current time
   * @throws AmbariException
   */
  protected void processCommandReports(HeartBeat heartbeat, long now) throws AmbariException {
    processCommandReports(heartbeat.getReports(), heartbeat.getHostname(), now);
  }

  protected void processCommandReports(List<CommandReport> reports, String hostName, Long now)
      throws AmbariException {

    // Cache HostRoleCommand entities because we will need them few times
    List<Long> taskIds = new ArrayList<>();
    for (CommandReport report : reports) {
      taskIds.add(report.getTaskId());
    }
    Map<Long, HostRoleCommand> commands = actionManager.getTasksMap(taskIds);

    for (CommandReport report : reports) {

      Long clusterId = Long.parseLong(report.getClusterId());

      LOG.debug("Received command report: {}", report);

      // get this locally; don't touch the database
      Host host = clusterFsm.getHost(hostName);
      if (host == null) {
        LOG.error("Received a command report and was unable to retrieve Host for hostname = " + hostName);
        continue;
      }

      // Send event for final command reports for actions
      if (RoleCommand.valueOf(report.getRoleCommand()) == RoleCommand.ACTIONEXECUTE &&
          HostRoleStatus.valueOf(report.getStatus()).isCompletedState()) {
        ActionFinalReportReceivedEvent event = new ActionFinalReportReceivedEvent(
            clusterId, hostName, report, false);
        ambariEventPublisher.publish(event);
      }

      // Fetch HostRoleCommand that corresponds to a given task ID
      HostRoleCommand hostRoleCommand = commands.get(report.getTaskId());
      if (hostRoleCommand == null) {
        LOG.warn("Can't fetch HostRoleCommand with taskId = " + report.getTaskId());
      } else {
        // Skip sending events for command reports for ABORTed commands
        if (hostRoleCommand.getStatus() == HostRoleStatus.ABORTED) {
          continue;
        }
        if (hostRoleCommand.getStatus() == HostRoleStatus.QUEUED &&
            report.getStatus().equals("IN_PROGRESS")) {
          hostRoleCommand.setStartTime(now);

          // Because the task may be retried several times, set the original start time only once.
          if (hostRoleCommand.getOriginalStartTime() == -1) {
            hostRoleCommand.setOriginalStartTime(now);
          }
        }
      }

      // If the report indicates the keytab file was successfully transferred to a host or removed
      // from a host, record this for future reference
      if (Service.Type.KERBEROS.name().equalsIgnoreCase(report.getServiceName()) &&
          Role.KERBEROS_CLIENT.name().equalsIgnoreCase(report.getRole()) &&
          RoleCommand.CUSTOM_COMMAND.name().equalsIgnoreCase(report.getRoleCommand()) &&
          RequestExecution.Status.COMPLETED.name().equalsIgnoreCase(report.getStatus())) {

        String customCommand = report.getCustomCommand();

        if (SET_KEYTAB.equalsIgnoreCase(customCommand)) {
          WriteKeytabsStructuredOut writeKeytabsStructuredOut;
          try {
            writeKeytabsStructuredOut = gson.fromJson(report.getStructuredOut(), WriteKeytabsStructuredOut.class);
          } catch (JsonSyntaxException ex) {
            //Json structure was incorrect do nothing, pass this data further for processing
            writeKeytabsStructuredOut = null;
          }

          if (writeKeytabsStructuredOut != null) {
            Map<String, String> keytabs = writeKeytabsStructuredOut.getKeytabs();
            if (keytabs != null) {
              for (Map.Entry<String, String> entry : keytabs.entrySet()) {
                String keytabPath = entry.getValue();
                for (KerberosKeytabPrincipalEntity kkpe : kerberosKeytabPrincipalDAO.findByHostAndKeytab(host.getHostId(), keytabPath)) {
                  kkpe.setDistributed(true);
                  kerberosKeytabPrincipalDAO.merge(kkpe);
                }
              }
            }
          }
        } else if (CHECK_KEYTABS.equalsIgnoreCase(customCommand)) {
          ListKeytabsStructuredOut structuredOut = gson.fromJson(report.getStructuredOut(), ListKeytabsStructuredOut.class);
          for (MissingKeytab each : structuredOut.missingKeytabs) {
            LOG.info("Missing principal: {} for keytab: {} on host: {}", each.principal, each.keytabFilePath, hostName);
            KerberosKeytabPrincipalEntity kkpe = kerberosKeytabPrincipalDAO.findByHostKeytabAndPrincipal(host.getHostId(), each.keytabFilePath, each.principal);
            kkpe.setDistributed(false);
            kerberosKeytabPrincipalDAO.merge(kkpe);
          }
        }
      }

      //pass custom START, STOP and RESTART
      if (RoleCommand.ACTIONEXECUTE.toString().equals(report.getRoleCommand()) ||
          (RoleCommand.CUSTOM_COMMAND.toString().equals(report.getRoleCommand()) &&
              !("RESTART".equals(report.getCustomCommand()) ||
                  "START".equals(report.getCustomCommand()) ||
                  "STOP".equals(report.getCustomCommand())))) {
        continue;
      }

      Cluster cl = clusterFsm.getCluster(Long.parseLong(report.getClusterId()));
      String service = report.getServiceName();
      if (service == null || service.isEmpty()) {
        throw new AmbariException("Invalid command report, service: " + service);
      }
      if (actionMetadata.getActions(service.toLowerCase()).contains(report.getRole())) {
        LOG.debug("{} is an action - skip component lookup", report.getRole());
      } else {
        try {
          Service svc = cl.getService(service);
          ServiceComponent svcComp = svc.getServiceComponent(report.getRole());
          ServiceComponentHost scHost = svcComp.getServiceComponentHost(hostName);
          String schName = scHost.getServiceComponentName();

          if (report.getStatus().equals(HostRoleStatus.COMPLETED.toString())) {

            // Reading component version if it is present
            if (StringUtils.isNotBlank(report.getStructuredOut())
                && !StringUtils.equals("{}", report.getStructuredOut())) {
              ComponentVersionStructuredOut structuredOutput = null;
              try {
                structuredOutput = gson.fromJson(report.getStructuredOut(), ComponentVersionStructuredOut.class);
              } catch (JsonSyntaxException ex) {
                //Json structure for component version was incorrect
                //do nothing, pass this data further for processing
              }

              String newVersion = structuredOutput == null ? null : structuredOutput.version;
              Long repoVersionId = structuredOutput == null ? null : structuredOutput.repositoryVersionId;

              HostComponentVersionAdvertisedEvent event = new HostComponentVersionAdvertisedEvent(
                  cl, scHost, newVersion, repoVersionId);

              versionEventPublisher.publish(event);
            }

            if (!scHost.getState().equals(org.apache.ambari.server.state.State.UPGRADING) &&
                (report.getRoleCommand().equals(RoleCommand.START.toString()) ||
                (report.getRoleCommand().equals(RoleCommand.CUSTOM_COMMAND.toString()) &&
                    ("START".equals(report.getCustomCommand()) ||
                        "RESTART".equals(report.getCustomCommand()))))) {
              scHost.setRestartRequired(false);
            }

            // Necessary for resetting clients stale configs after starting service
            if ((RoleCommand.INSTALL.toString().equals(report.getRoleCommand()) ||
                (RoleCommand.CUSTOM_COMMAND.toString().equals(report.getRoleCommand()) &&
                    "INSTALL".equals(report.getCustomCommand()))) && svcComp.isClientComponent()){
              scHost.setRestartRequired(false);
            }

            if (RoleCommand.CUSTOM_COMMAND.toString().equals(report.getRoleCommand()) &&
                !("START".equals(report.getCustomCommand()) ||
                    "STOP".equals(report.getCustomCommand()))) {
              //do not affect states for custom commands except START and STOP
              //lets status commands to be responsible for this
              continue;
            }

            if (RoleCommand.START.toString().equals(report.getRoleCommand()) ||
                (RoleCommand.CUSTOM_COMMAND.toString().equals(report.getRoleCommand()) &&
                    "START".equals(report.getCustomCommand()))) {
              scHost.handleEvent(new ServiceComponentHostStartedEvent(schName,
                  hostName, now));
              scHost.setRestartRequired(false);
            } else if (RoleCommand.STOP.toString().equals(report.getRoleCommand()) ||
                (RoleCommand.CUSTOM_COMMAND.toString().equals(report.getRoleCommand()) &&
                    "STOP".equals(report.getCustomCommand()))) {
              scHost.handleEvent(new ServiceComponentHostStoppedEvent(schName,
                  hostName, now));
            } else {
              scHost.handleEvent(new ServiceComponentHostOpSucceededEvent(schName,
                  hostName, now));
            }
          } else if (report.getStatus().equals("FAILED")) {

            if (StringUtils.isNotBlank(report.getStructuredOut())) {
              try {
                ComponentVersionStructuredOut structuredOutput = gson.fromJson(report.getStructuredOut(), ComponentVersionStructuredOut.class);

                if (null != structuredOutput.upgradeDirection) {
                  scHost.setUpgradeState(UpgradeState.FAILED);
                }
              } catch (JsonSyntaxException ex) {
                LOG.warn("Structured output was found, but not parseable: {}", report.getStructuredOut());
              }
            }

            LOG.error("Operation failed - may be retried. Service component host: "
                + schName + ", host: " + hostName + " Action id " + report.getActionId() + " and taskId " + report.getTaskId());
            if (actionManager.isInProgressCommand(report)) {
              scHost.handleEvent(new ServiceComponentHostOpFailedEvent
                  (schName, hostName, now));
            } else {
              LOG.info("Received report for a command that is no longer active. " + report);
            }
          } else if (report.getStatus().equals("IN_PROGRESS")) {
            scHost.handleEvent(new ServiceComponentHostOpInProgressEvent(schName,
                hostName, now));
          }
        } catch (ServiceComponentNotFoundException scnex) {
          LOG.warn("Service component not found ", scnex);
        } catch (InvalidStateTransitionException ex) {
          if (LOG.isDebugEnabled()) {
            LOG.warn("State machine exception.", ex);
          } else {
            LOG.warn("State machine exception. " + ex.getMessage());
          }
        }
      }
    }

    //Update state machines from reports
    actionManager.processTaskResponse(hostName, reports, commands);
  }

  /**
   * Process reports of status commands
   *
   * @param heartbeat heartbeat to process
   * @throws AmbariException
   */
  protected void processStatusReports(HeartBeat heartbeat) throws AmbariException {
    processStatusReports(heartbeat.getComponentStatus(), heartbeat.getHostname());
  }

  /**
   * Process reports of components versions
   * @throws AmbariException
   */
  public void processVersionReports(ComponentVersionReports versionReports, String hostname) throws AmbariException {
    Set<Cluster> clusters = clusterFsm.getClustersForHost(hostname);
    for (Cluster cl : clusters) {
      for (Map.Entry<String, List<ComponentVersionReport>> status : versionReports
          .getComponentVersionReports().entrySet()) {
        if (Long.valueOf(status.getKey()).equals(cl.getClusterId())) {
          for (ComponentVersionReport versionReport : status.getValue()) {
            try {
              Service svc = cl.getService(versionReport.getServiceName());

              String componentName = versionReport.getComponentName();
              if (svc.getServiceComponents().containsKey(componentName)) {
                ServiceComponent svcComp = svc.getServiceComponent(
                    componentName);
                ServiceComponentHost scHost = svcComp.getServiceComponentHost(
                    hostname);

                String version = versionReport.getVersion();

                HostComponentVersionAdvertisedEvent event = new HostComponentVersionAdvertisedEvent(cl,
                    scHost, version);
                versionEventPublisher.publish(event);
              }
            } catch (ServiceNotFoundException e) {
              LOG.warn("Received a version report for a non-initialized"
                  + " service"
                  + ", clusterId=" + versionReport.getClusterId()
                  + ", serviceName=" + versionReport.getServiceName());
              continue;
            } catch (ServiceComponentNotFoundException e) {
              LOG.warn("Received a version report for a non-initialized"
                  + " servicecomponent"
                  + ", clusterId=" + versionReport.getClusterId()
                  + ", serviceName=" + versionReport.getServiceName()
                  + ", componentName=" + versionReport.getComponentName());
              continue;
            } catch (ServiceComponentHostNotFoundException e) {
              LOG.warn("Received a version report for a non-initialized"
                  + " hostcomponent"
                  + ", clusterId=" + versionReport.getClusterId()
                  + ", serviceName=" + versionReport.getServiceName()
                  + ", componentName=" + versionReport.getComponentName()
                  + ", hostname=" + hostname);
              continue;
            } catch (RuntimeException e) {
              LOG.warn("Received a version report with invalid payload"
                  + " service"
                  + ", clusterId=" + versionReport.getClusterId()
                  + ", serviceName=" + versionReport.getServiceName()
                  + ", componentName=" + versionReport.getComponentName()
                  + ", hostname=" + hostname
                  + ", error=" + e.getMessage());
              continue;
            }
          }
        }
      }
    }
  }

  /**
   * Process reports of status commands
   * @throws AmbariException
   */
  public void processStatusReports(List<ComponentStatus> componentStatuses, String hostname) throws AmbariException {
    Set<Cluster> clusters = clusterFsm.getClustersForHost(hostname);
    for (Cluster cl : clusters) {
      for (ComponentStatus status : componentStatuses) {
        if (status.getClusterId().equals(cl.getClusterId())) {
          try {
            Service svc = cl.getService(status.getServiceName());

            String componentName = status.getComponentName();
            if (svc.getServiceComponents().containsKey(componentName)) {
              ServiceComponent svcComp = svc.getServiceComponent(
                  componentName);
              ServiceComponentHost scHost = svcComp.getServiceComponentHost(
                  hostname);
              if (status.getStatus() != null) {
                org.apache.ambari.server.state.State prevState = scHost.getState();
                org.apache.ambari.server.state.State liveState =
                    org.apache.ambari.server.state.State.valueOf(org.apache.ambari.server.state.State.class,
                        status.getStatus());
                //ignore reports from status commands if component is in INIT or any "in progress" state
                if (prevState.equals(org.apache.ambari.server.state.State.INSTALLED)
                    || prevState.equals(org.apache.ambari.server.state.State.STARTED)
                    || prevState.equals(org.apache.ambari.server.state.State.UNKNOWN)) {
                  scHost.setState(liveState);
                  if (!prevState.equals(liveState)) {
                    LOG.info("State of service component " + componentName
                        + " of service " + status.getServiceName()
                        + " of cluster " + status.getClusterId()
                        + " has changed from " + prevState + " to " + liveState
                        + " at host " + hostname
                        + " according to STATUS_COMMAND report");
                  }
                }
              }

              Map<String, Object> extra = status.getExtra();
              if (null != extra && !extra.isEmpty()) {
                try {
                  if (extra.containsKey("processes")) {
                    @SuppressWarnings("unchecked")
                    List<Map<String, String>> list = (List<Map<String, String>>) extra.get("processes");
                    scHost.setProcesses(list);
                  }
                } catch (Exception e) {
                  LOG.error("Could not access extra JSON for " +
                      scHost.getServiceComponentName() + " from " +
                      scHost.getHostName() + ": " + status.getExtra() +
                      " (" + e.getMessage() + ")");
                }
              }

              heartbeatMonitor.getAgentRequests()
                  .setExecutionDetailsRequest(hostname, componentName, status.getSendExecCmdDet());
            } else {
              // TODO: What should be done otherwise?
            }
          } catch (ServiceNotFoundException e) {
            LOG.warn("Received a live status update for a non-initialized"
                + " service"
                + ", clusterId=" + status.getClusterId()
                + ", serviceName=" + status.getServiceName());
            // FIXME ignore invalid live update and continue for now?
            continue;
          } catch (ServiceComponentNotFoundException e) {
            LOG.warn("Received a live status update for a non-initialized"
                + " servicecomponent"
                + ", clusterId=" + status.getClusterId()
                + ", serviceName=" + status.getServiceName()
                + ", componentName=" + status.getComponentName());
            // FIXME ignore invalid live update and continue for now?
            continue;
          } catch (ServiceComponentHostNotFoundException e) {
            LOG.warn("Received a live status update for a non-initialized"
                + " service"
                + ", clusterId=" + status.getClusterId()
                + ", serviceName=" + status.getServiceName()
                + ", componentName=" + status.getComponentName()
                + ", hostname=" + hostname);
            // FIXME ignore invalid live update and continue for now?
            continue;
          } catch (RuntimeException e) {
            LOG.warn("Received a live status with invalid payload"
                + " service"
                + ", clusterId=" + status.getClusterId()
                + ", serviceName=" + status.getServiceName()
                + ", componentName=" + status.getComponentName()
                + ", hostname=" + hostname
                + ", error=" + e.getMessage());
            continue;
          }
        }
      }
    }

    Host host = clusterFsm.getHost(hostname);
    long now = System.currentTimeMillis();
    // If the host is waiting for component status updates, notify it
    if (componentStatuses.size() > 0
        && host.getState().equals(HostState.WAITING_FOR_HOST_STATUS_UPDATES)) {
      try {
        LOG.debug("Got component status updates for host {}", hostname);
        host.handleEvent(new HostStatusUpdatesReceivedEvent(hostname, now));
      } catch (InvalidStateTransitionException e) {
        LOG.warn("Failed to notify the host about component status updates for host {}", hostname, e);
      }
    }
  }

  /**
   * This class is used for mapping json of structured output for keytab distribution actions.
   */
  private static class WriteKeytabsStructuredOut {
    @SerializedName("keytabs")
    private Map<String, String> keytabs;

    @SerializedName("removedKeytabs")
    private Map<String, String> removedKeytabs;

    public Map<String, String> getKeytabs() {
      return keytabs;
    }

    public void setKeytabs(Map<String, String> keytabs) {
      this.keytabs = keytabs;
    }

    public Map<String, String> getRemovedKeytabs() {
      return removedKeytabs;
    }

    public void setRemovedKeytabs(Map<String, String> removedKeytabs) {
      this.removedKeytabs = removedKeytabs;
    }
  }

  private static class ListKeytabsStructuredOut {
    @SerializedName("missing_keytabs")
    private final List<MissingKeytab> missingKeytabs;

    public ListKeytabsStructuredOut(List<MissingKeytab> missingKeytabs) {
      this.missingKeytabs = missingKeytabs;
    }
  }

  private static class MissingKeytab {
    @SerializedName("principal")
    private final String principal;
    @SerializedName("keytab_file_path")
    private final String keytabFilePath;

    public MissingKeytab(String principal, String keytabFilePath) {
      this.principal = principal;
      this.keytabFilePath = keytabFilePath;
    }
  }

  /**
   * This class is used for mapping json of structured output for component START action.
   */
  private static class ComponentVersionStructuredOut {
    @SerializedName("version")
    private String version;

    @SerializedName("upgrade_type")
    private UpgradeType upgradeType = null;

