/*
 * 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.events.listeners.alerts;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.locks.Lock;

import org.apache.ambari.server.AmbariException;
import org.apache.ambari.server.EagerSingleton;
import org.apache.ambari.server.api.query.render.AlertSummaryGroupedRenderer;
import org.apache.ambari.server.configuration.Configuration;
import org.apache.ambari.server.controller.MaintenanceStateHelper;
import org.apache.ambari.server.controller.RootComponent;
import org.apache.ambari.server.controller.RootService;
import org.apache.ambari.server.events.AlertEvent;
import org.apache.ambari.server.events.AlertReceivedEvent;
import org.apache.ambari.server.events.AlertStateChangeEvent;
import org.apache.ambari.server.events.AlertUpdateEvent;
import org.apache.ambari.server.events.InitialAlertEvent;
import org.apache.ambari.server.events.publishers.AlertEventPublisher;
import org.apache.ambari.server.events.publishers.STOMPUpdatePublisher;
import org.apache.ambari.server.orm.RequiresSession;
import org.apache.ambari.server.orm.dao.AlertDefinitionDAO;
import org.apache.ambari.server.orm.dao.AlertsDAO;
import org.apache.ambari.server.orm.entities.AlertCurrentEntity;
import org.apache.ambari.server.orm.entities.AlertDefinitionEntity;
import org.apache.ambari.server.orm.entities.AlertHistoryEntity;
import org.apache.ambari.server.state.Alert;
import org.apache.ambari.server.state.AlertFirmness;
import org.apache.ambari.server.state.AlertState;
import org.apache.ambari.server.state.Cluster;
import org.apache.ambari.server.state.Clusters;
import org.apache.ambari.server.state.ConfigHelper;
import org.apache.ambari.server.state.Host;
import org.apache.ambari.server.state.MaintenanceState;
import org.apache.ambari.server.state.alert.AlertHelper;
import org.apache.ambari.server.state.alert.SourceType;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.math.NumberUtils;
import org.eclipse.jetty.util.StringUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.eventbus.AllowConcurrentEvents;
import com.google.common.eventbus.Subscribe;
import com.google.common.util.concurrent.Striped;
import com.google.inject.Inject;
import com.google.inject.Provider;
import com.google.inject.Singleton;

/**
 * The {@link AlertReceivedListener} class handles {@link AlertReceivedEvent}
 * and updates the appropriate DAOs. It may also fire new
 * {@link AlertStateChangeEvent} when an {@link AlertState} change is detected.
 */
@Singleton
@EagerSingleton
public class AlertReceivedListener {
  /**
   * Logger.
   */
  private static final Logger LOG = LoggerFactory.getLogger(AlertReceivedListener.class);

  @Inject
  Configuration m_configuration;

  @Inject
  AlertsDAO m_alertsDao;

  @Inject
  AlertDefinitionDAO m_definitionDao;

  /**
   * Used for looking up whether an alert has a valid service/component/host
   */
  @Inject
  private Provider<Clusters> m_clusters;

  @Inject
  private STOMPUpdatePublisher STOMPUpdatePublisher;

  /**
   * Used to calculate the maintenance state of new alerts being created.
   * Consider the case where you have disabled alerts for a component in MM.
   * This means that there are no current alerts in the system since disabling
   * them removes all current instances. New alerts being created for the
   * component in MM must reflect the correct MM.
   */
  @Inject
  private Provider<MaintenanceStateHelper> m_maintenanceStateHelper;

  @Inject
  private AlertHelper alertHelper;

  /**
   * Receives and publishes {@link AlertEvent} instances.
   */
  private AlertEventPublisher m_alertEventPublisher;

  /**
   * Used for ensuring that creation of {@link AlertCurrentEntity} instances has fine-grain
   * locks to prevent duplicates.
   */
  private Striped<Lock> creationLocks = Striped.lazyWeakLock(100);

  /**
   * Constructor.
   *
   * @param publisher
   */
  @Inject
  public AlertReceivedListener(AlertEventPublisher publisher) {
    m_alertEventPublisher = publisher;
    m_alertEventPublisher.register(this);
  }

  /**
   * Adds an alert. Checks for a new state before creating a new history record.
   *
   * @param event
   *          the event to handle.
   */
  @Subscribe
  @AllowConcurrentEvents
  @RequiresSession
  public void onAlertEvent(AlertReceivedEvent event) throws AmbariException {
    if (LOG.isDebugEnabled()) {
      LOG.debug(event.toString());
    }

    // process the list of alerts inside of a single transaction to prevent too
    // many transactions/commits
    List<Alert> alerts = event.getAlerts();

    // these can be wrapped in their own transaction
    List<AlertCurrentEntity> toMerge = new ArrayList<>();
    List<AlertCurrentEntity> toCreateHistoryAndMerge = new ArrayList<>();

    List<AlertEvent> alertEvents = new ArrayList<>(20);
    Map<Long, Map<String, AlertSummaryGroupedRenderer.AlertDefinitionSummary>> alertUpdates = new HashMap<>();

    for (Alert alert : alerts) {
      Long clusterId = alert.getClusterId();
      if (clusterId == null) {
        // check event
        clusterId = event.getClusterId();
      }

      AlertDefinitionEntity definition = m_definitionDao.findByName(clusterId, alert.getName());

      if (null == definition) {
        LOG.warn(
          "Received an alert for {} which is a definition that does not exist in cluster id={}",
          alert.getName(), clusterId);

        continue;
      }

      alert.setComponent(definition.getComponentName());
      alert.setLabel(definition.getComponentName());
      alert.setService(definition.getServiceName());

      // jobs that were running when a service/component/host was changed
      // which invalidate the alert should not be reported
      if (!isValid(alert)) {
        continue;
      }
      // it's possible that a definition which is disabled will still have a
      // running alert returned; this will ensure we don't record it
      if (!definition.getEnabled()) {
        LOG.debug(
          "Received an alert for {} which is disabled. No more alerts should be received for this definition.",
          alert.getName());

        continue;
      }

      updateAlertDetails(alert, definition);

      // jobs that were running when a service/component/host was changed
      // which invalidate the alert should not be reported
      if (!isValid(alert)) {
        continue;
      }

      AlertCurrentEntity current;
      AlertState alertState = alert.getState();

      // attempt to lookup the current alert
      current = getCurrentEntity(clusterId, alert, definition);

      // if it doesn't exist then we must create it, ensuring that two or more
      // aren't created from other threads
      if( null == current ){

        // if there is no current alert and the state is skipped, then simply
        // skip over this one as there is nothing to update in the databse
        if (alertState == AlertState.SKIPPED) {
          continue;
        }

        // create a key out of the cluster/definition name/host (possibly null)
        int key = Objects.hash(clusterId, alert.getName(), alert.getHostName());
        Lock lock = creationLocks.get(key);
        lock.lock();

        // attempt to lookup the current alert again to ensure that a previous
        // thread didn't already create it
        try {
          // if it's not null anymore, then there's no work to do here
          current = getCurrentEntity(clusterId, alert, definition);
          if( null != current ) {
            continue;
          }

          // the current alert is still null, so go through and create it
          AlertHistoryEntity history = createHistory(clusterId, definition, alert);

          // this new alert must reflect the correct MM state for the
          // service/component/host
          MaintenanceState maintenanceState = getMaintenanceState(alert, clusterId);

          current = new AlertCurrentEntity();
          current.setMaintenanceState(maintenanceState);
          current.setAlertHistory(history);
          current.setLatestTimestamp(alert.getTimestamp());
          current.setOriginalTimestamp(alert.getTimestamp());
          clearStaleAlerts(alert.getHostName(), definition.getDefinitionId());

          // brand new alert instances being received are always HARD
          current.setFirmness(AlertFirmness.HARD);

          m_alertsDao.create(current);

          // create the event to fire later
          alertEvents.add(new InitialAlertEvent(clusterId, alert, current));

          if (!alertUpdates.containsKey(clusterId)) {
            alertUpdates.put(clusterId, new HashMap<>());
          }
          Map<String, AlertSummaryGroupedRenderer.AlertDefinitionSummary> summaries = alertUpdates.get(clusterId);

          AlertSummaryGroupedRenderer.updateSummary(summaries, definition.getDefinitionId(),
              definition.getDefinitionName(), alertState, alert.getTimestamp(), maintenanceState, alert.getText());
        } finally {
          // release the lock for this alert
          lock.unlock();
        }
      } else if (alertState == current.getAlertHistory().getAlertState()
          || alertState == AlertState.SKIPPED) {

        // update the timestamp no matter what
        current.setLatestTimestamp(alert.getTimestamp());
        clearStaleAlerts(alert.getHostName(), definition.getDefinitionId());

        // only update some fields if the alert isn't SKIPPED
        if (alertState != AlertState.SKIPPED) {
          current.setLatestText(alert.getText());

          // ++ the occurrences (should be safe enough since we should ever only
          // be handling unique alert events concurrently
          long occurrences = current.getOccurrences() + 1;
          current.setOccurrences(occurrences);

          // ensure that if we've met the repeat tolerance and the alert is
          // still SOFT, then we transition it to HARD - we also need to fire an
          // event
          AlertFirmness firmness = current.getFirmness();
          int repeatTolerance = getRepeatTolerance(definition, clusterId);
          if (firmness == AlertFirmness.SOFT && occurrences >= repeatTolerance) {
            current.setFirmness(AlertFirmness.HARD);

            // create the event to fire later
            AlertStateChangeEvent stateChangedEvent = new AlertStateChangeEvent(clusterId, alert,
                current, alertState, firmness);

            alertEvents.add(stateChangedEvent);
          }
        }

        // some special cases for SKIPPED alerts
        if (alertState == AlertState.SKIPPED) {
          // set the text on a SKIPPED alert IFF it's not blank; a blank text
          // field means that the alert doesn't want to change the existing text
          String alertText = alert.getText();
          if (StringUtils.isNotBlank(alertText)) {
            current.setLatestText(alertText);
          }
        }

        // store the entity for merging later
        toMerge.add(current);
      } else {
        if (LOG.isDebugEnabled()) {
          LOG.debug(
            "Alert State Changed: CurrentId {}, CurrentTimestamp {}, HistoryId {}, HistoryState {}",
            current.getAlertId(), current.getLatestTimestamp(),
            current.getAlertHistory().getAlertId(),
            current.getAlertHistory().getAlertState());
        }

        AlertHistoryEntity oldHistory = current.getAlertHistory();
        AlertState oldState = oldHistory.getAlertState();
        AlertFirmness oldFirmness = current.getFirmness();

        // insert history, update current
        AlertHistoryEntity history = createHistory(clusterId,
          oldHistory.getAlertDefinition(), alert);

        current.setLatestTimestamp(alert.getTimestamp());
        current.setOriginalTimestamp(alert.getTimestamp());
        current.setLatestText(alert.getText());

        clearStaleAlerts(alert.getHostName(), definition.getDefinitionId());

        current.setAlertHistory(history);

        // figure out how to set the occurrences correctly
        switch (alertState) {
          // an OK state always resets, regardless of what the old one was
          case OK:
            current.setOccurrences(1);
            break;
          case CRITICAL:
          case SKIPPED:
          case UNKNOWN:
          case WARNING:
            // OK -> non-OK is a reset
            if (oldState == AlertState.OK) {
              current.setOccurrences(1);
            } else {
              // non-OK -> non-OK is a continuation
              current.setOccurrences(current.getOccurrences() + 1);
            }
            break;
          default:
            break;
        }

        // set the firmness of the new alert state based on the state, type,
        // occurrences, and repeat tolerance
        AlertFirmness firmness = calculateFirmnessForStateChange(clusterId, definition,
            alertState, current.getOccurrences());

        current.setFirmness(firmness);

        // store the entity for merging later
        toCreateHistoryAndMerge.add(current);

        // create the event to fire later
        alertEvents.add(new AlertStateChangeEvent(clusterId, alert, current, oldState, oldFirmness));

        // create alert update to fire event to UI
        MaintenanceState maintenanceState = getMaintenanceState(alert, clusterId);

        if (!alertUpdates.containsKey(clusterId)) {
          alertUpdates.put(clusterId, new HashMap<>());
        }
        Map<String, AlertSummaryGroupedRenderer.AlertDefinitionSummary> summaries = alertUpdates.get(clusterId);

        AlertSummaryGroupedRenderer.updateSummary(summaries, definition.getDefinitionId(),
            definition.getDefinitionName(), alertState, alert.getTimestamp(), maintenanceState, alert.getText());
      }
    }

    // invokes the EntityManager create/merge on various entities in a single
    // transaction
    m_alertsDao.saveEntities(toMerge, toCreateHistoryAndMerge);

    // broadcast events
    for (AlertEvent eventToFire : alertEvents) {
      m_alertEventPublisher.publish(eventToFire);
    }
    if (!alertUpdates.isEmpty()) {
      STOMPUpdatePublisher.publish(new AlertUpdateEvent(alertUpdates));
    }
  }

  private void clearStaleAlerts(String hostName, Long definitionId) throws AmbariException {
    if (StringUtil.isNotBlank(hostName)) {
      Host host = m_clusters.get().getHosts().stream().filter(h -> h.getHostName().equals(hostName))
          .findFirst().orElse(null);
      if (host != null) {
        alertHelper.clearStaleAlert(host.getHostId(), definitionId);
      }
    } else {
      alertHelper.clearStaleAlert(definitionId);
    }
  }

  private void updateAlertDetails(Alert alert, AlertDefinitionEntity definition) {
    if (alert.getService() == null) {
      alert.setService(definition.getServiceName());
    }
    if (alert.getComponent() == null) {
      alert.setComponent(definition.getComponentName());
    }
  }

  private MaintenanceState getMaintenanceState(Alert alert, Long clusterId) {
    MaintenanceState maintenanceState = MaintenanceState.OFF;
    try {
      maintenanceState = m_maintenanceStateHelper.get().getEffectiveState(clusterId, alert);
    } catch (Exception exception) {
      LOG.error("Unable to determine the maintenance mode state for {}, defaulting to OFF",
          alert, exception);
    }
    return maintenanceState;
  }

  /**
   * Gets the {@link AlertCurrentEntity} which cooresponds to the new alert being received, if any.
   *
   * @param clusterId the ID of the cluster.
   * @param alert the alert being received (not {@code null}).
   * @param definition  the {@link AlertDefinitionEntity} for the alert being received (not {@code null}).
   * @return  the existing current alert or {@code null} for none.
   */
  private AlertCurrentEntity getCurrentEntity(long clusterId, Alert alert, AlertDefinitionEntity definition){
    if (StringUtils.isBlank(alert.getHostName()) || definition.isHostIgnored()) {
      return m_alertsDao.findCurrentByNameNoHost(clusterId, alert.getName());
    } else {
      return m_alertsDao.findCurrentByHostAndName(clusterId, alert.getHostName(),
        alert.getName());
    }
  }

  /**
   * Gets whether the specified alert is valid for its reported cluster,
   * service, component, and host. This method is necessary for the following
   * cases
   * <ul>
   * <li>A service/component is removed, but an alert queued for reporting is
   * received after that event.</li>
   * <li>A host is removed from the cluster but the agent is still running and
   * reporting</li>
   * <li>A cluster is renamed</li>
   * </ul>
   *
   * @param alert
   *          the alert.
   * @return {@code true} if the alert is for a valid combination of
   *         cluster/service/component/host.
   */
  private boolean isValid(Alert alert) {
    Long clusterId = alert.getClusterId();
    String serviceName = alert.getService();
    String componentName = alert.getComponent();
    String hostName = alert.getHostName();

    // AMBARI/AMBARI_SERVER is always a valid service/component combination
    String ambariServiceName = RootService.AMBARI.name();
    String ambariServerComponentName = RootComponent.AMBARI_SERVER.name();
    String ambariAgentComponentName = RootComponent.AMBARI_AGENT.name();
    if (ambariServiceName.equals(serviceName) && ambariServerComponentName.equals(componentName)) {
      return true;
    }

    // if the alert is not bound to a cluster, then it's most likely a
    // host alert and is always valid as long as the host exists
    Clusters clusters = m_clusters.get();
    if (clusterId == null) {
      // no cluster, no host; return true out of respect for the unknown alert
      if (StringUtils.isBlank(hostName)) {
        return true;
      }

      // if a host is reported, it must be registered to some cluster somewhere
      if (!clusters.hostExists(hostName)) {
        LOG.error("Unable to process alert {} for an invalid host named {}",
            alert.getName(), hostName);
        return false;
      }

      // no cluster, valid host; return true
      return true;
    }

    // at this point the following criteria is guaranteed, so get the cluster
    // - a cluster exists
    // - this is not for AMBARI_SERVER component
    final Cluster cluster;
    try {
      cluster = clusters.getCluster(clusterId);
      if (null == cluster) {
        LOG.error("Unable to process alert {} for cluster id={}", alert.getName(), clusterId);
        return false;
      }
    } catch (AmbariException ambariException) {
      String msg = String.format("Unable to process alert %s for cluster id=%s", alert.getName(), clusterId);
      LOG.error(msg, ambariException);
      return false;
    }

    // at this point the following criteria is guaranteed
    // - a cluster exists
    // - this is not for AMBARI_SERVER component
    //
    // if the alert is for AMBARI/AMBARI_AGENT, then it's valid IFF
    // the agent's host is still a part of the reported cluster
    if (ambariServiceName.equals(serviceName) && ambariAgentComponentName.equals(componentName)) {
      // agents MUST report a hostname
      if (StringUtils.isBlank(hostName) || !clusters.hostExists(hostName)
          || !clusters.isHostMappedToCluster(clusterId, hostName)) {
        LOG.warn(
            "Unable to process alert {} for cluster {} and host {} because the host is not a part of the cluster.",
            alert.getName(), hostName);

        return false;
      }

      // AMBARI/AMBARI_AGENT and valid host; return true
      return true;
    }

    // at this point the following criteria is guaranteed
    // - a cluster exists
    // - not for the AMBARI service
    if (StringUtils.isNotBlank(hostName)) {
      // if valid hostname
      if (!clusters.hostExists(hostName)) {
        LOG.warn("Unable to process alert {} for an invalid host named {}",
            alert.getName(), hostName);
        return false;
      }

      if (!cluster.getServices().containsKey(serviceName)) {
        LOG.warn("Unable to process alert {} for an invalid service named {}",
            alert.getName(), serviceName);

        return false;
      }

      // if the alert is for a host/component then verify that the component
      // is actually installed on that host
      if (null != componentName &&
          !cluster.getHosts(serviceName, componentName).contains(hostName)) {
        LOG.warn(
            "Unable to process alert {} for an invalid service {} and component {} on host {}",
            alert.getName(), serviceName, componentName, hostName);
        return false;
      }
    }

    return true;
  }

  /**
   * Convenience method to create a new historical alert.
   *
   * @param clusterId
   *          the cluster id
   * @param definition
   *          the definition
   * @param alert
   *          the alert data
   * @return the new history record
   */
  private AlertHistoryEntity createHistory(long clusterId,
      AlertDefinitionEntity definition, Alert alert) {
    AlertHistoryEntity history = new AlertHistoryEntity();
    history.setAlertDefinition(definition);
    history.setAlertDefinitionId(definition.getDefinitionId());
    history.setAlertLabel(definition.getLabel());
    history.setAlertInstance(alert.getInstance());
    history.setAlertState(alert.getState());
    history.setAlertText(alert.getText());
    history.setAlertTimestamp(alert.getTimestamp());
    history.setClusterId(clusterId);
    history.setComponentName(alert.getComponent());
    history.setServiceName(alert.getService());

    // only set a host for the history item if the alert definition says to
    if (definition.isHostIgnored()) {
      history.setHostName(null);
    } else {
      history.setHostName(alert.getHostName());
    }

    return history;
  }

  /**
   * Gets the firmness for an {@link AlertCurrentEntity}. The following rules
   * apply:
   * <ul>
   * <li>If an alert is {@link AlertState#OK}, then the firmness is always
   * {@link AlertFirmness#HARD}.</li>
   * <li>If an alert is {@link SourceType#AGGREGATE}, then the firmness is
   * always {@link AlertFirmness#HARD}.</li>
   * <li>Otherwise, the firmness will be {@link AlertFirmness#SOFT} unless the
   * repeat tolerance has been met.</li>
   * </ul>
   *
   * @param definition
   *          the definition to read any repeat tolerance overrides from.
   * @param state
   *          the state of the {@link AlertCurrentEntity}.
   * @param occurrences the
   *          occurrences of the alert in the current state (used for
   *          calculation firmness when moving between non-OK states)
   * @return
   */
  private AlertFirmness calculateFirmnessForStateChange(Long clusterId, AlertDefinitionEntity definition,
      AlertState state, long occurrences) {
    // OK is always HARD since the alert has fulfilled the conditions
    if (state == AlertState.OK) {
      return AlertFirmness.HARD;
    }

    // aggregate alerts are always HARD since they only react to HARD alerts
    if (definition.getSourceType() == SourceType.AGGREGATE) {
      return AlertFirmness.HARD;
    }

    int tolerance = getRepeatTolerance(definition, clusterId);
    if (tolerance <= 1) {
      return AlertFirmness.HARD;
    }

    if (tolerance <= occurrences) {
      return AlertFirmness.HARD;
    }

    return AlertFirmness.SOFT;
  }

  /**
   * Gets the repeat tolerance value for the specified definition. This method
   * will return the override from the definition if
   * {@link AlertDefinitionEntity#isRepeatToleranceEnabled()} is {@code true}.
   * Otherwise, it uses {@link ConfigHelper#CLUSTER_ENV_ALERT_REPEAT_TOLERANCE},
   * defaulting to {@code 1} if not found.
   *
   * @param definition the definition (not {@code null}).
   * @param clusterId the ID of the cluster (not {@code null}).
   * @return the repeat tolerance for the alert
   */
  private int getRepeatTolerance(AlertDefinitionEntity definition, Long clusterId) {

    // if the definition overrides the global value, then use that
    if (definition.isRepeatToleranceEnabled()) {
      return definition.getRepeatTolerance();
    }

    int repeatTolerance = 1;
    try {
      Cluster cluster = m_clusters.get().getCluster(clusterId);
      String value = cluster.getClusterProperty(ConfigHelper.CLUSTER_ENV_ALERT_REPEAT_TOLERANCE, "1");
      repeatTolerance = NumberUtils.toInt(value, 1);
    } catch (AmbariException ambariException) {
      String msg = String.format("Unable to read %s/%s from cluster %s, defaulting to 1",
        ConfigHelper.CLUSTER_ENV, ConfigHelper.CLUSTER_ENV_ALERT_REPEAT_TOLERANCE, clusterId
      );
      LOG.warn(msg, ambariException);
    }

    return repeatTolerance;
  }
}
