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

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.lang.Thread.UncaughtExceptionHandler;
import java.lang.reflect.Type;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.apache.ambari.server.AmbariService;
import org.apache.ambari.server.configuration.Configuration;
import org.apache.ambari.server.controller.jmx.JMXMetricHolder;
import org.apache.ambari.server.controller.utilities.ScalingThreadPoolExecutor;
import org.apache.ambari.server.controller.utilities.StreamProvider;
import org.apache.commons.io.IOUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.fasterxml.jackson.core.JsonParser;
import com.fasterxml.jackson.databind.MapperFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.ObjectReader;
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import com.google.common.collect.Sets;
import com.google.common.util.concurrent.AbstractService;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.google.gson.Gson;
import com.google.gson.reflect.TypeToken;
import com.google.gson.stream.JsonReader;
import com.google.inject.Inject;

/**
 * The {@link MetricsRetrievalService} is used as a headless, autonomous service
 * which encapsulates:
 * <ul>
 * <li>An {@link ExecutorService} for fulfilling remote metric URL requests
 * <li>A cache for JMX metrics
 * <li>A cache for REST metrics
 * </ul>
 *
 * Classes can inject an instance of this service in order to gain access to its
 * caches and request mechanism.
 * <p/>
 * Callers must submit a request to the service in order to reach out and pull
 * in remote metric data. Otherwise, the cache will never be populated. On the
 * first usage of this service, the cache will always be empty. On every
 * subsequent request, the data from the prior invocation of
 * {@link #submitRequest(MetricSourceType, StreamProvider, String)} will be available.
 * <p/>
 * Metric data is cached temporarily and is controlled by
 * {@link Configuration#getMetricsServiceCacheTimeout()}.
 * <p/>
 * In order to control throttling requests to the same endpoint,
 * {@link Configuration#isMetricsServiceRequestTTLCacheEnabled()} can be enabled
 * to allow for a fixed interval of time to pass between requests.
 */
@AmbariService
public class MetricsRetrievalService extends AbstractService {

  /**
   * The type of web service hosting the metrics.
   */
  public enum MetricSourceType {
    /**
     * JMX
     */
    JMX,

    /**
     * REST
     */
    REST
  }

  /**
   * Logger.
   */
  private static final Logger LOG = LoggerFactory.getLogger(MetricsRetrievalService.class);

  /**
   * The timeout for exceptions which are caught and then cached to prevent log
   * spamming.
   *
   * @see #s_exceptionCache
   */
  private static final int EXCEPTION_CACHE_TIMEOUT_MINUTES = 20;

  /**
   * Exceptions from this service should not SPAM the logs; so cache exceptions
   * and log once every {@vale #EXCEPTION_CACHE_TIMEOUT_MINUTES} minutes.
   */
  private static final Cache<String, Throwable> s_exceptionCache = CacheBuilder.newBuilder().expireAfterWrite(
      EXCEPTION_CACHE_TIMEOUT_MINUTES, TimeUnit.MINUTES).build();

  /**
   * Configuration.
   */
  @Inject
  private Configuration m_configuration;

  /**
   * Used for reading REST JSON responses.
   */
  @Inject
  private Gson m_gson;

  /**
   * A cache of URL to parsed JMX beans
   */
  private Cache<String, JMXMetricHolder> m_jmxCache;

  /**
   * A cache of URL to parsed REST data.
   */
  private Cache<String, Map<String, String>> m_restCache;

  /**
   * The {@link Executor} which will handle all of the requests to load remote
   * metrics from URLs.
   */
  private ThreadPoolExecutor m_threadPoolExecutor;

  /**
   * Used to parse remote JMX JSON into a {@link Map}.
   */
  private final ObjectReader m_jmxObjectReader;

  /**
   * A thread-safe collection of all of the URL endpoints queued for processing.
   * This helps prevent the same endpoint from being queued multiple times.
   */
  private final Set<String> m_queuedUrls = Sets.newConcurrentHashSet();

  /**
   * An evicting cache which ensures that multiple requests for the same
   * endpoint are not executed back-to-back. When enabled, a fixed period of
   * time must pass before this service will make requests to a previously
   * retrieved endpoint.
   * <p/>
   * If this cache is not enabled, then it will be {@code null}.
   * <p/>
   * For simplicity, this is a cache of URL to URL.
   */
  private Cache<String, String> m_ttlUrlCache;


  /**
   * The size of the worker queue (used for logged warnings about size).
   */
  private int m_queueMaximumSize;

  /**
   * Constructor.
   *
   */
  public MetricsRetrievalService() {
    ObjectMapper jmxObjectMapper = new ObjectMapper();
    jmxObjectMapper.configure(MapperFeature.USE_ANNOTATIONS, false);
    jmxObjectMapper.configure(JsonParser.Feature.ALLOW_NON_NUMERIC_NUMBERS, true);
    m_jmxObjectReader = jmxObjectMapper.reader(JMXMetricHolder.class);
  }

  /**
   * {@inheritDoc}
   */
  @Override
  protected void doStart() {
    // initialize the caches
    int jmxCacheExpirationMinutes = m_configuration.getMetricsServiceCacheTimeout();
    m_jmxCache = CacheBuilder.newBuilder().expireAfterWrite(jmxCacheExpirationMinutes,
        TimeUnit.MINUTES).build();

    m_restCache = CacheBuilder.newBuilder().expireAfterWrite(jmxCacheExpirationMinutes,
        TimeUnit.MINUTES).build();

    // enable the TTL cache if configured; otherwise leave it as null
    int ttlSeconds = m_configuration.getMetricsServiceRequestTTL();
    boolean ttlCacheEnabled = m_configuration.isMetricsServiceRequestTTLCacheEnabled();
    if (ttlCacheEnabled) {
      m_ttlUrlCache = CacheBuilder.newBuilder().expireAfterWrite(ttlSeconds,
          TimeUnit.SECONDS).build();
    }

    // iniitalize the executor service
    int corePoolSize = m_configuration.getMetricsServiceThreadPoolCoreSize();
    int maxPoolSize = m_configuration.getMetricsServiceThreadPoolMaxSize();
    m_queueMaximumSize = m_configuration.getMetricsServiceWorkerQueueSize();
    int threadPriority = m_configuration.getMetricsServiceThreadPriority();
    m_threadPoolExecutor = new ScalingThreadPoolExecutor(corePoolSize, maxPoolSize, 30,
        TimeUnit.SECONDS, m_queueMaximumSize);

    m_threadPoolExecutor.allowCoreThreadTimeOut(true);
    m_threadPoolExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.DiscardOldestPolicy());

    ThreadFactory threadFactory = new ThreadFactoryBuilder().setDaemon(true).setNameFormat(
        "ambari-metrics-retrieval-service-thread-%d").setPriority(
            threadPriority).setUncaughtExceptionHandler(
            new MetricRunnableExceptionHandler()).build();

    m_threadPoolExecutor.setThreadFactory(threadFactory);

    LOG.info(
        "Initializing the Metrics Retrieval Service with core={}, max={}, workerQueue={}, threadPriority={}",
        corePoolSize, maxPoolSize, m_queueMaximumSize, threadPriority);

    if (ttlCacheEnabled) {
      LOG.info("Metrics Retrieval Service request TTL cache is enabled and set to {} seconds",
          ttlSeconds);
    }
    notifyStarted();
  }

  /**
   * Testing method for setting a synchronous {@link ThreadPoolExecutor}.
   *
   * @param threadPoolExecutor
   */
  public void setThreadPoolExecutor(ThreadPoolExecutor threadPoolExecutor) {
    m_threadPoolExecutor = threadPoolExecutor;
  }

  /**
   * {@inheritDoc}
   */
  @Override
  protected void doStop() {
    m_jmxCache.invalidateAll();
    m_restCache.invalidateAll();

    if (null != m_ttlUrlCache) {
      m_ttlUrlCache.invalidateAll();
    }

    m_queuedUrls.clear();
    m_threadPoolExecutor.shutdownNow();
    notifyStopped();
  }

  /**
   * Submit a {@link Runnable} for execution which retrieves metric data from
   * the supplied endpoint. This will run inside of an {@link ExecutorService}
   * to retrieve metric data from a URL endpoint and parse the result into a
   * cached value.
   * <p/>
   * Once metric data is retrieved it is cached. Data in the cache can be
   * retrieved via {@link #getCachedJMXMetric(String)} or
   * {@link #getCachedRESTMetric(String)}, depending on the type of metric
   * requested.
   * <p/>
   * Callers need not worry about invoking this multiple times for the same URL
   * endpoint. A single endpoint will only be enqueued once regardless of how
   * many times this method is called until it has been fully retrieved and
   * parsed. If the last endpoint request was too recent, then this method will
   * opt to not make another call until the TTL period expires.
   *
   * @param type
   *          the type of service hosting the metric (not {@code null}).
   * @param streamProvider
   *          the {@link StreamProvider} to use to read from the remote
   *          endpoint.
   * @param url
   *          the URL to read from
   *
   * @see #getCachedJMXMetric(String)
   */
  public void submitRequest(MetricSourceType type, StreamProvider streamProvider, String url) {
    // check to ensure that the request isn't already queued
    if (m_queuedUrls.contains(url)) {
      return;
    }

    // check to ensure that the request wasn't made too recently
    if (null != m_ttlUrlCache && null != m_ttlUrlCache.getIfPresent(url)) {
      return;
    }

    // log warnings if the queue size seems to be rather large
    BlockingQueue<Runnable> queue = m_threadPoolExecutor.getQueue();
    int queueSize = queue.size();
    if (queueSize > Math.floor(0.9f * m_queueMaximumSize)) {
      LOG.warn("The worker queue contains {} work items and is at {}% of capacity", queueSize,
          ((float) queueSize / m_queueMaximumSize) * 100);
    }

    // enqueue this URL
    m_queuedUrls.add(url);

    Runnable runnable = null;
    switch (type) {
      case JMX:
        runnable = new JMXRunnable(m_jmxCache, m_queuedUrls, m_ttlUrlCache, m_jmxObjectReader,
            streamProvider, url);
        break;
      case REST:
        runnable = new RESTRunnable(m_restCache, m_queuedUrls, m_ttlUrlCache, m_gson,
            streamProvider, url);
        break;
      default:
        LOG.warn("Unable to retrieve metrics for the unknown type {}", type);
        break;
    }

    if (null != runnable) {
      m_threadPoolExecutor.execute(runnable);
    }
  }

  /**
   * Gets a cached JMX metric in the form of a {@link JMXMetricHolder}. If there
   * is no metric data cached for the given URL, then {@code null} is returned.
   * <p/>
   * The onky way this cache is populated is by requesting the data to be loaded
   * asynchronously via
   * {@link #submitRequest(MetricSourceType, StreamProvider, String)} with the
   * {@link MetricSourceType#JMX} type.
   *
   * @param jmxUrl
   *          the URL to retrieve cached data for (not {@code null}).
   * @return the metric, or {@code null} if none.
   */
  public JMXMetricHolder getCachedJMXMetric(String jmxUrl) {
    return m_jmxCache.getIfPresent(jmxUrl);
  }

  /**
   * Gets a cached REST metric in the form of a {@link Map}. If there is no
   * metric data cached for the given URL, then {@code null} is returned.
   * <p/>
   * The only way this cache is populated is by requesting the data to be loaded
   * asynchronously via
   * {@link #submitRequest(MetricSourceType, StreamProvider, String)} with the
   * {@link MetricSourceType#REST} type.
   *
   * @param restUrl
   *          the URL to retrieve cached data for (not {@code null}).
   * @return the metric, or {@code null} if none.
   */
  public Map<String, String> getCachedRESTMetric(String restUrl) {
    return m_restCache.getIfPresent(restUrl);
  }

  /**
   * Encapsulates the common logic for all metric {@link Runnable} instances.
   */
  private static abstract class MetricRunnable implements Runnable {

    /**
     * An initialized stream provider to read the remote endpoint.
     */
    protected final StreamProvider m_streamProvider;

    /**
     * A fully-qualified URL to read from.
     */
    protected final String m_url;

    /**
     * The URLs which have been requested but not yet read.
     */
    private final Set<String> m_queuedUrls;

    /**
     * An evicting cache used to control whether a request for a metric can be
     * made or if it is too soon after the last request.
     */
    private final Cache<String, String> m_ttlUrlCache;

    /**
     * Constructor.
     *
     * @param streamProvider
     *          the stream provider to read the URL with
     * @param url
     *          the URL endpoint to read data from (JMX or REST)
     * @param queuedUrls
     *          the URLs which are currently waiting to be processed. This
     *          method will remove the specified URL from this {@link Set} when
     *          it completes (successful or not).
     * @param ttlUrlCache
     *          an evicting cache which is used to determine if a request for a
     *          metric is too soon after the last request, or {@code null} if
     *          requests can be made sequentially without any separation.
     */
    private MetricRunnable(StreamProvider streamProvider, String url, Set<String> queuedUrls,
        Cache<String, String> ttlUrlCache) {
      m_streamProvider = streamProvider;
      m_url = url;
      m_queuedUrls = queuedUrls;
      m_ttlUrlCache = ttlUrlCache;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public final void run() {

      // provide some profiling
      long startTime = 0;
      long endTime = 0;
      boolean isDebugEnabled = LOG.isDebugEnabled();
      if (isDebugEnabled) {
        startTime = System.currentTimeMillis();
      }

      InputStream inputStream = null;

      try {
        if (isDebugEnabled) {
          endTime = System.currentTimeMillis();
          LOG.debug("Loading metric JSON from {} took {}ms", m_url, (endTime - startTime));
        }

        // read the stream and process it
        inputStream = m_streamProvider.readFrom(m_url);
        processInputStreamAndCacheResult(inputStream);

        // cache the URL, but only after successful parsing of the response
        if (null != m_ttlUrlCache) {
          m_ttlUrlCache.put(m_url, m_url);
        }
      } catch (IOException exception)
      {
        LOG.debug("Removing cached values for url {}", m_url);
        // need to ensure old values are removed because they could be not valid if the state have changed.
        removeCachedMetricsForCurrentURL();
        logException(exception, m_url);
      } catch (Exception exception) {
        logException(exception, m_url);
      } finally {
        IOUtils.closeQuietly(inputStream);

        // remove this URL from the list of queued URLs to ensure it will be
        // requested again
        m_queuedUrls.remove(m_url);
      }
    }

    /**
     * Removes metric values for current URL from cache.
     */
    protected abstract void removeCachedMetricsForCurrentURL();

    /**
     * Reads data from the specified {@link InputStream} and processes that into
     * a cachable value. The value will then be cached by this method.
     *
     * @param inputStream
     * @throws Exception
     */
    protected abstract void processInputStreamAndCacheResult(InputStream inputStream)
        throws Exception;

    /**
     * Logs the exception for the URL exactly once and caches the fact that the
     * exception was logged. This is to prevent an endpoint being down from
     * spamming the logs.
     *
     * @param throwable
     *          the exception to log (not {@code null}).
     * @param url
     *          the URL associated with the exception (not {@code null}).
     */
    final void logException(Throwable throwable, String url) {
      String cacheKey = buildCacheKey(throwable, url);
      if (null == s_exceptionCache.getIfPresent(cacheKey)) {
        // cache it and log it
        s_exceptionCache.put(cacheKey, throwable);
        LOG.error(
            "Unable to retrieve metrics from {}. Subsequent failures will be suppressed from the log for {} minutes.",
            url, EXCEPTION_CACHE_TIMEOUT_MINUTES, throwable);
      }
    }

    /**
     * Builds a unique cache key for the combination of {@link Throwable} and
     * {@link String} URL.
     *
     * @param throwable
     * @param url
     * @return the key, such as <code>IOException-http://www.server.com/jmx</code>.
     */
    private String buildCacheKey(Throwable throwable, String url) {
      if (null == throwable || null == url) {
        return "";
      }

      String throwableName = throwable.getClass().getSimpleName();
      return throwableName + "-" + url;
    }

  }

  /**
   * A {@link Runnable} used to retrieve JMX data from a remote URL endpoint.
   * There is no need for a {@link Callable} here since the
   * {@link MetricsRetrievalService} doesn't care about when the value returns or
   * whether an exception is thrown.
   */
  private static final class JMXRunnable extends MetricRunnable {

    private final ObjectReader m_jmxObjectReader;
    private final Cache<String, JMXMetricHolder> m_cache;

    /**
     * Constructor.
     *
     * @param cache
     * @param queuedUrls
     * @param ttlUrlCache
     * @param jmxObjectReader
     * @param streamProvider
     * @param jmxUrl
     */
    private JMXRunnable(Cache<String, JMXMetricHolder> cache, Set<String> queuedUrls,
        Cache<String, String> ttlUrlCache, ObjectReader jmxObjectReader,
        StreamProvider streamProvider, String jmxUrl) {
      super(streamProvider, jmxUrl, queuedUrls, ttlUrlCache);
      m_cache = cache;
      m_jmxObjectReader = jmxObjectReader;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    protected void removeCachedMetricsForCurrentURL() {
      m_cache.invalidate(m_url);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    protected void processInputStreamAndCacheResult(InputStream inputStream) throws Exception {
      JMXMetricHolder jmxMetricHolder = m_jmxObjectReader.readValue(inputStream);
      m_cache.put(m_url, jmxMetricHolder);
    }
  }

  /**
   * A {@link Runnable} used to retrieve REST data from a remote URL endpoint.
   * There is no need for a {@link Callable} here since the
   * {@link MetricsRetrievalService} doesn't care about when the value returns
   * or whether an exception is thrown.
   */
  private static final class RESTRunnable extends MetricRunnable {

    private final Gson m_gson;
    private final Cache<String, Map<String, String>> m_cache;

    /**
     * Constructor.
     *
     * @param cache
     * @param queuedUrls
     * @param ttlUrlCache
     * @param gson
     * @param streamProvider
     * @param restUrl
     */
    private RESTRunnable(Cache<String, Map<String, String>> cache, Set<String> queuedUrls,
        Cache<String, String> ttlUrlCache, Gson gson, StreamProvider streamProvider,
        String restUrl) {
      super(streamProvider, restUrl, queuedUrls, ttlUrlCache);
      m_cache = cache;
      m_gson = gson;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    protected void removeCachedMetricsForCurrentURL() {
      m_cache.invalidate(m_url);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    protected void processInputStreamAndCacheResult(InputStream inputStream) throws Exception {
      Type type = new TypeToken<Map<Object, Object>>() {}.getType();

      JsonReader jsonReader = new JsonReader(
          new BufferedReader(new InputStreamReader(inputStream)));

      Map<String, String> jsonMap = m_gson.fromJson(jsonReader, type);
      m_cache.put(m_url, jsonMap);
    }
  }

  /**
   * A default exception handler.
   */
  private static final class MetricRunnableExceptionHandler implements UncaughtExceptionHandler {

    /**
     * {@inheritDoc}
     */
    @Override
    public void uncaughtException(Thread t, Throwable e) {
      LOG.error("Asynchronous metric retrieval encountered an exception with thread {}", t, e);
    }
  }
}
