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

import static org.apache.ambari.server.controller.internal.ProvisionAction.INSTALL_AND_START;
import static org.apache.ambari.server.controller.internal.ProvisionAction.INSTALL_ONLY;
import static org.apache.ambari.server.controller.internal.ProvisionAction.START_ONLY;

import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.ambari.server.actionmanager.HostRoleCommand;
import org.apache.ambari.server.actionmanager.HostRoleStatus;
import org.apache.ambari.server.api.predicate.InvalidQueryException;
import org.apache.ambari.server.api.predicate.PredicateCompiler;
import org.apache.ambari.server.controller.internal.HostResourceProvider;
import org.apache.ambari.server.controller.internal.ResourceImpl;
import org.apache.ambari.server.controller.internal.Stack;
import org.apache.ambari.server.controller.spi.Predicate;
import org.apache.ambari.server.controller.spi.Resource;
import org.apache.ambari.server.orm.entities.HostRoleCommandEntity;
import org.apache.ambari.server.orm.entities.TopologyHostRequestEntity;
import org.apache.ambari.server.orm.entities.TopologyHostTaskEntity;
import org.apache.ambari.server.orm.entities.TopologyLogicalTaskEntity;
import org.apache.ambari.server.state.Host;
import org.apache.ambari.server.topology.tasks.InstallHostTask;
import org.apache.ambari.server.topology.tasks.PersistHostResourcesTask;
import org.apache.ambari.server.topology.tasks.RegisterWithConfigGroupTask;
import org.apache.ambari.server.topology.tasks.StartHostTask;
import org.apache.ambari.server.topology.tasks.TopologyHostTask;
import org.apache.ambari.server.topology.tasks.TopologyTask;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.base.Optional;
import com.google.common.base.Throwables;

/**
 * Represents a set of requests to a single host such as install, start, etc.
 */
public class HostRequest implements Comparable<HostRequest> {

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

  private long requestId;
  private String blueprint;
  private HostGroup hostGroup;
  private String hostgroupName;
  private Predicate predicate;
  private String hostname = null;
  private long clusterId;
  private boolean containsMaster;
  private final long id;
  private boolean isOutstanding = true;
  private final boolean skipFailure;
  private HostRoleStatus status = HostRoleStatus.PENDING;
  private String statusMessage;

  private Map<TopologyTask, Map<String, Long>> logicalTaskMap = new HashMap<>();

  Map<Long, HostRoleCommand> logicalTasks = new HashMap<>();

  // logical task id -> physical tasks
  private Map<Long, Long> physicalTasks = new ConcurrentHashMap<>();

  private List<TopologyHostTask> topologyTasks = new ArrayList<>();

  private ClusterTopology topology;

  private static PredicateCompiler predicateCompiler = new PredicateCompiler();

  public HostRequest(long requestId, long id, long clusterId, String hostname, String blueprintName,
                     HostGroup hostGroup, Predicate predicate, ClusterTopology topology, boolean skipFailure) {
    this.requestId = requestId;
    this.id = id;
    this.clusterId = clusterId;
    blueprint = blueprintName;
    this.hostGroup = hostGroup;
    hostgroupName = hostGroup.getName();
    this.predicate = predicate;
    containsMaster = hostGroup.containsMasterComponent();
    this.topology = topology;
    this.skipFailure = skipFailure;
    createTasks(this.skipFailure);
    LOG.info("HostRequest: Created request for host: " +
        (hostname == null ? "Host Assignment Pending" : hostname));
  }

  /**
   * Only to be used when replaying persisted requests upon server startup.
   *
   * @param requestId    logical request id
   * @param id           host request id
   * @param predicate    host predicate
   * @param topology     cluster topology
   * @param entity       host request entity
   */
  public HostRequest(long requestId, long id, String predicate,
                     ClusterTopology topology, TopologyHostRequestEntity entity, boolean skipFailure) {

    this.requestId = requestId;
    this.id = id;
    clusterId = topology.getClusterId();
    blueprint = topology.getBlueprint().getName();
    hostgroupName = entity.getTopologyHostGroupEntity().getName();
    hostGroup = topology.getBlueprint().getHostGroup(hostgroupName);
    hostname = entity.getHostName();
    setStatus(entity.getStatus());
    statusMessage = entity.getStatusMessage();
    this.predicate = toPredicate(predicate);
    containsMaster = hostGroup.containsMasterComponent();
    this.topology = topology;
    this.skipFailure = skipFailure;

    createTasksForReplay(entity);

    //todo: we may be able to simplify by just checking hostname
    isOutstanding = hostname == null || !topology.getAmbariContext().
        isHostRegisteredWithCluster(clusterId, hostname);

    LOG.info("HostRequest: Successfully recovered host request for host: " +
        (hostname == null ? "Host Assignment Pending" : hostname));
  }

  void markHostRequestFailed(HostRoleStatus status, Throwable cause, PersistedState persistedState) {
    String errorMessage = StringUtils.substringBefore(Throwables.getRootCause(cause).getMessage(), "\n");
    LOG.info("HostRequest: marking host request {} for {} as {} due to {}", id, hostname, status, errorMessage);
    abortPendingTasks();
    setStatus(status);
    setStatusMessage(errorMessage);
    persistedState.setHostRequestStatus(id, status, errorMessage);
  }

  //todo: synchronization
  public synchronized HostOfferResponse offer(Host host) {
    if (!isOutstanding) {
      return HostOfferResponse.DECLINED_DUE_TO_DONE;
    }
    if (matchesHost(host)) {
      isOutstanding = false;
      hostname = host.getHostName();
      setHostOnTasks(host);
      return HostOfferResponse.createAcceptedResponse(id, hostGroup.getName(), topologyTasks);
    } else {
      return HostOfferResponse.DECLINED_DUE_TO_PREDICATE;
    }
  }

  public HostRoleStatus getStatus() {
    return status;
  }

  public void setStatus(HostRoleStatus status) {
    if (status != null) {
      this.status = status;
    }
  }

  public void setStatusMessage(String errorMessage) {
    this.statusMessage = errorMessage;
  }

  public Optional<String> getStatusMessage() {
    return Optional.fromNullable(statusMessage);
  }

  public void setHostName(String hostName) {
    hostname = hostName;
  }

  public long getRequestId() {
    return requestId;
  }

  public long getClusterId() {
    return clusterId;
  }
  public String getBlueprint() {
    return blueprint;
  }

  public HostGroup getHostGroup() {
    return hostGroup;
  }

  public String getHostgroupName() {
    return hostgroupName;
  }

  public Predicate getPredicate() {
    return predicate;
  }

  public boolean isCompleted() {
    return ! isOutstanding;
  }

  public boolean shouldSkipFailure() {
    return skipFailure;
  }

  private void createTasks(boolean skipFailure) {
    // high level topology tasks such as INSTALL, START, ...
    topologyTasks.add(new PersistHostResourcesTask(topology, this));
    topologyTasks.add(new RegisterWithConfigGroupTask(topology, this));

    InstallHostTask installTask = new InstallHostTask(topology, this, skipFailure);
    topologyTasks.add(installTask);
    logicalTaskMap.put(installTask, new HashMap<>());

    boolean skipStartTaskCreate = topology.getProvisionAction().equals(INSTALL_ONLY);
    boolean skipInstallTaskCreate = topology.getProvisionAction().equals(START_ONLY);

    StartHostTask startTask = null;
    if (!skipStartTaskCreate) {
      startTask = new StartHostTask(topology, this, skipFailure);
      topologyTasks.add(startTask);
      logicalTaskMap.put(startTask, new HashMap<>());
    } else {
      LOG.info("Skipping Start task creation since provision action = " + topology.getProvisionAction());
    }

    // lower level logical component level tasks which get mapped to physical tasks
    HostGroup hostGroup = getHostGroup();
    Collection<String> startOnlyComponents = hostGroup.getComponentNames(START_ONLY);
    Collection<String> installOnlyComponents = hostGroup.getComponentNames(INSTALL_ONLY);
    Collection<String> installAndStartComponents = hostGroup.getComponentNames(INSTALL_AND_START);

    for (String component : hostGroup.getComponentNames()) {
      if (component == null || component.equals("AMBARI_SERVER")) {
        LOG.info("Skipping component {} when creating request\n", component);
        continue;
      }

      String hostName = getHostName() != null ?
          getHostName() :
          "PENDING HOST ASSIGNMENT : HOSTGROUP=" + getHostgroupName();

      AmbariContext context = topology.getAmbariContext();
      Stack stack = hostGroup.getStack();

      // Skip INSTALL task in case server component is marked as START_ONLY, or the cluster provision_action is
      // START_ONLY, unless component is marked with INSTALL_ONLY or INSTALL_AND_START.
      if (startOnlyComponents.contains(component) || (skipInstallTaskCreate &&
        !installOnlyComponents.contains(component) && !installAndStartComponents.contains(component))
          && stack != null && !stack.getComponentInfo(component).isClient()) {
        LOG.info("Skipping create of INSTALL task for {} on {}.", component, hostName);
      } else {
        HostRoleCommand logicalInstallTask = context.createAmbariTask(
          getRequestId(), id, component, hostName, AmbariContext.TaskType.INSTALL, skipFailure);
        logicalTasks.put(logicalInstallTask.getTaskId(), logicalInstallTask);
        logicalTaskMap.get(installTask).put(component, logicalInstallTask.getTaskId());
      }

      // Skip START task if component is a client, or ir marked as INSTALL_ONLY or cluster provision_action is
      // INSTALL_ONLY
      if (installOnlyComponents.contains(component) || skipStartTaskCreate ||
        (stack != null && stack.getComponentInfo(component).isClient())) {
        LOG.info("Skipping create of START task for {} on {}.", component, hostName);
      } else {
        HostRoleCommand logicalStartTask = context.createAmbariTask(
            getRequestId(), id, component, hostName, AmbariContext.TaskType.START, skipFailure);
        logicalTasks.put(logicalStartTask.getTaskId(), logicalStartTask);
        logicalTaskMap.get(startTask).put(component, logicalStartTask.getTaskId());
      }
    }
  }

  private void createTasksForReplay(TopologyHostRequestEntity entity) {
    topologyTasks.add(new PersistHostResourcesTask(topology, this));
    topologyTasks.add(new RegisterWithConfigGroupTask(topology, this));
    InstallHostTask installTask = new InstallHostTask(topology, this, skipFailure);
    topologyTasks.add(installTask);
    logicalTaskMap.put(installTask, new HashMap<>());

    boolean skipStartTaskCreate = topology.getProvisionAction().equals(INSTALL_ONLY);

    if (!skipStartTaskCreate) {
      StartHostTask startTask = new StartHostTask(topology, this, skipFailure);
      topologyTasks.add(startTask);
      logicalTaskMap.put(startTask, new HashMap<>());
    }

    AmbariContext ambariContext = topology.getAmbariContext();
    // lower level logical component level tasks which get mapped to physical tasks
    for (TopologyHostTaskEntity topologyTaskEntity : entity.getTopologyHostTaskEntities()) {
      TopologyTask.Type taskType = TopologyTask.Type.valueOf(topologyTaskEntity.getType());
      for (TopologyLogicalTaskEntity logicalTaskEntity : topologyTaskEntity.getTopologyLogicalTaskEntities()) {
        Long logicalTaskId = logicalTaskEntity.getId();
        String component = logicalTaskEntity.getComponentName();

        AmbariContext.TaskType logicalTaskType = getLogicalTaskType(taskType);
        HostRoleCommand task = ambariContext.createAmbariTask(logicalTaskId, getRequestId(), id,
            component, entity.getHostName(), logicalTaskType, skipFailure);

        logicalTasks.put(logicalTaskId, task);
        Long physicalTaskId = logicalTaskEntity.getPhysicalTaskId();
        if (physicalTaskId != null) {
          registerPhysicalTaskId(logicalTaskId, physicalTaskId);
        }

        //assumes only one task per type
        for (TopologyTask topologyTask : topologyTasks) {
          if (taskType == topologyTask.getType()) {
            logicalTaskMap.get(topologyTask).put(component, logicalTaskId);
          }
        }
      }
    }
  }

  private static AmbariContext.TaskType getLogicalTaskType(TopologyTask.Type topologyTaskType) {
    return topologyTaskType ==
        TopologyTask.Type.INSTALL ?
        AmbariContext.TaskType.INSTALL :
        AmbariContext.TaskType.START;
  }

  private void setHostOnTasks(Host host) {
    for (HostRoleCommand task : getLogicalTasks()) {
      task.setHost(host.getHostId(), host.getHostName());
    }
  }

  public List<TopologyHostTask> getTopologyTasks() {
    return topologyTasks;
  }

  public Collection<HostRoleCommand> getLogicalTasks() {
    // sync logical task state with physical tasks
    for (HostRoleCommand logicalTask : logicalTasks.values()) {
      // set host on command detail if it is set to null
      String commandDetail = logicalTask.getCommandDetail();
      if (commandDetail != null && commandDetail.contains("null") && hostname != null) {
        logicalTask.setCommandDetail(commandDetail.replace("null", hostname));
      }
      Long physicalTaskId = physicalTasks.get(logicalTask.getTaskId());
      if (physicalTaskId != null) {
        HostRoleCommand physicalTask = topology.getAmbariContext().getPhysicalTask(physicalTaskId);
        if (physicalTask != null) {
          logicalTask.setStatus(physicalTask.getStatus());
          logicalTask.setCommandDetail(physicalTask.getCommandDetail());
          logicalTask.setCustomCommandName(physicalTask.getCustomCommandName());
          //todo: once we retry on failures, start/end times could span multiple physical tasks
          logicalTask.setStartTime(physicalTask.getStartTime());
          logicalTask.setOriginalStartTime(physicalTask.getOriginalStartTime());
          logicalTask.setEndTime(physicalTask.getEndTime());
          logicalTask.setErrorLog(physicalTask.getErrorLog());
          logicalTask.setExitCode(physicalTask.getExitCode());
          logicalTask.setExecutionCommandWrapper(physicalTask.getExecutionCommandWrapper());
          //todo: may be handled at a higher level than physical task
          logicalTask.setLastAttemptTime(physicalTask.getLastAttemptTime());
          logicalTask.setOutputLog(physicalTask.getOutputLog());
          logicalTask.setStderr(physicalTask.getStderr());
          logicalTask.setStdout(physicalTask.getStdout());
          logicalTask.setStructuredOut(physicalTask.getStructuredOut());
        }
      }
      if (logicalTask.getStatus() == HostRoleStatus.PENDING && status != HostRoleStatus.PENDING) {
        logicalTask.setStatus(status);
      }
    }
    return logicalTasks.values();
  }

  public Map<String, Long> getLogicalTasksForTopologyTask(TopologyTask topologyTask) {
    return new HashMap<>(logicalTaskMap.get(topologyTask));
  }

  public HostRoleCommand getLogicalTask(long logicalTaskId) {
    return logicalTasks.get(logicalTaskId);
  }

  public Collection<HostRoleCommandEntity> getTaskEntities() {
    Collection<HostRoleCommandEntity> taskEntities = new ArrayList<>();
    for (HostRoleCommand task : logicalTasks.values()) {
      HostRoleCommandEntity entity = task.constructNewPersistenceEntity();
      // the above method doesn't set all of the fields for some unknown reason
      entity.setOutputLog(task.getOutputLog());
      entity.setErrorLog(task.errorLog);

      // set state from physical task
      Long physicalTaskId = physicalTasks.get(task.getTaskId());
      if (physicalTaskId != null) {
        HostRoleCommand physicalTask = topology.getAmbariContext().getPhysicalTask(physicalTaskId);
        if (physicalTask != null) {
          entity.setStatus(physicalTask.getStatus());
          entity.setCommandDetail(physicalTask.getCommandDetail());
          entity.setCustomCommandName(physicalTask.getCustomCommandName());
          //todo: once we retry on failures, start/end times could span multiple physical tasks
          entity.setStartTime(physicalTask.getStartTime());
          entity.setOriginalStartTime(physicalTask.getOriginalStartTime());
          entity.setEndTime(physicalTask.getEndTime());
          entity.setErrorLog(physicalTask.getErrorLog());
          entity.setExitcode(physicalTask.getExitCode());
          //todo: may be handled at a higher level than physical task
          entity.setLastAttemptTime(physicalTask.getLastAttemptTime());
          entity.setOutputLog(physicalTask.getOutputLog());
          entity.setStdError(physicalTask.getStderr().getBytes());
          entity.setStdOut(physicalTask.getStdout().getBytes());
          entity.setStructuredOut(physicalTask.getStructuredOut().getBytes());
        }
      }

      taskEntities.add(entity);
    }
    return taskEntities;
  }

  public boolean containsMaster() {
    return containsMaster;
  }

  public boolean matchesHost(Host host) {
    return (hostname != null) ?
        host.getHostName().equals(hostname) :
        predicate == null || predicate.evaluate(new HostResourceAdapter(host));
  }

  public String getHostName() {
    return hostname;
  }

  public long getId() {
    return id;
  }

  public long getStageId() {
    // stage id is same as host request id
    return getId();
  }

  public Long getPhysicalTaskId(long logicalTaskId) {
    return physicalTasks.get(logicalTaskId);
  }

  public Map<Long, Long> getPhysicalTaskMapping() {
    return new ConcurrentHashMap<>(physicalTasks);
  }

  //todo: since this is used to determine equality, using hashCode() isn't safe as it can return the same
  //todo: value for 2 unequal requests
  @Override
  public int compareTo(HostRequest other) {
    if (containsMaster()) {
      return other.containsMaster() ? hashCode() - other.hashCode() : -1;
    } else if (other.containsMaster()) {
      return 1;
    } else {
      return hashCode() - other.hashCode();
    }
  }

  //todo: once we have logical tasks, move tracking of physical tasks there
  public void registerPhysicalTaskId(long logicalTaskId, long physicalTaskId) {
    physicalTasks.put(logicalTaskId, physicalTaskId);
    topology.getAmbariContext().getPersistedTopologyState().registerPhysicalTask(logicalTaskId, physicalTaskId);
    getLogicalTask(logicalTaskId).incrementAttemptCount();
  }

  public void abortPendingTasks() {
    for (HostRoleCommand command : getLogicalTasks()) {
      if (command.getStatus() == HostRoleStatus.PENDING) {
        command.setStatus(HostRoleStatus.ABORTED);
      }
    }
  }

  private Predicate toPredicate(String predicate) {
    Predicate compiledPredicate = null;
    try {
      if (predicate != null && ! predicate.isEmpty()) {
        compiledPredicate = predicateCompiler.compile(predicate);
      }
    } catch (InvalidQueryException e) {
      // log error and proceed without predicate
      LOG.error("Unable to compile predicate for host request: " + e, e);
    }
    return compiledPredicate;
  }

  private class HostResourceAdapter implements Resource {
    Resource hostResource;

    public HostResourceAdapter(Host host) {
      buildPropertyMap(host);
    }

    @Override
    public Object getPropertyValue(String id) {
      return hostResource.getPropertyValue(id);
    }

    @Override
    public Map<String, Map<String, Object>> getPropertiesMap() {
      return hostResource.getPropertiesMap();
    }

    @Override
    public Type getType() {
      return Type.Host;
    }

    @Override
    public void addCategory(String id) {
      // read only, nothing to do
    }

    @Override
    public void setProperty(String id, Object value) {
      // read only, nothing to do
    }

    private void buildPropertyMap(Host host) {
      hostResource = new ResourceImpl(Resource.Type.Host);

      hostResource.setProperty(HostResourceProvider.HOST_HOST_NAME_PROPERTY_ID,
          host.getHostName());
      hostResource.setProperty(HostResourceProvider.HOST_PUBLIC_NAME_PROPERTY_ID,
          host.getPublicHostName());
      hostResource.setProperty(HostResourceProvider.HOST_IP_PROPERTY_ID,
          host.getIPv4());
      hostResource.setProperty(HostResourceProvider.HOST_TOTAL_MEM_PROPERTY_ID,
          host.getTotalMemBytes());
      hostResource.setProperty(HostResourceProvider.HOST_CPU_COUNT_PROPERTY_ID,
          (long) host.getCpuCount());
      hostResource.setProperty(HostResourceProvider.HOST_PHYSICAL_CPU_COUNT_PROPERTY_ID,
          (long) host.getPhCpuCount());
      hostResource.setProperty(HostResourceProvider.HOST_OS_ARCH_PROPERTY_ID,
          host.getOsArch());
      hostResource.setProperty(HostResourceProvider.HOST_OS_TYPE_PROPERTY_ID,
          host.getOsType());
      hostResource.setProperty(HostResourceProvider.HOST_OS_FAMILY_PROPERTY_ID,
          host.getOsFamily());
      hostResource.setProperty(HostResourceProvider.HOST_RACK_INFO_PROPERTY_ID,
          host.getRackInfo());
      hostResource.setProperty(HostResourceProvider.HOST_LAST_HEARTBEAT_TIME_PROPERTY_ID,
          host.getLastHeartbeatTime());
      hostResource.setProperty(HostResourceProvider.HOST_LAST_AGENT_ENV_PROPERTY_ID,
          host.getLastAgentEnv());
      hostResource.setProperty(HostResourceProvider.HOST_LAST_REGISTRATION_TIME_PROPERTY_ID,
          host.getLastRegistrationTime());
      hostResource.setProperty(HostResourceProvider.HOST_HOST_STATUS_PROPERTY_ID,
          host.getStatus());
      hostResource.setProperty(HostResourceProvider.HOST_HOST_HEALTH_REPORT_PROPERTY_ID,
          host.getHealthStatus().getHealthReport());
      hostResource.setProperty(HostResourceProvider.HOST_DISK_INFO_PROPERTY_ID,
          host.getDisksInfo());
      hostResource.setProperty(HostResourceProvider.HOST_STATE_PROPERTY_ID,
          host.getState());
    }
  }
}
