/*
 * 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
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * 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.controller.logging;

import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.apache.ambari.server.AmbariService;
import org.apache.ambari.server.configuration.Configuration;
import org.apache.ambari.server.controller.AmbariManagementController;
import org.apache.ambari.server.controller.AmbariServer;
import org.apache.commons.collections4.MapUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import com.google.common.util.concurrent.AbstractService;
import com.google.inject.Inject;
import com.google.inject.Injector;


/**
 * The {@link LogSearchDataRetrievalService} is an Ambari Service that
 *   is used by the Ambari LogSearch integration code to obtain response
 *   data from the LogSearch server.
 *
 * In order to improve the performance of the LogSearch integration layer in
 *   Ambari, this service implements the following:
 *
 *  <ul>
 *    <li>A cache for LogSearch data that typically is returned by the LogSearch REST API</li>
 *    <li>Implements the remote request for LogSearch data not found in the cache on a separate
 *        thread, which keeps the request from affecting the overall performance of the
 *        Ambari REST API</li>
 *  </ul>
 *
 *  As with other services annotated with {@link AmbariService}, this class may be
 *    injected in order to obtain cached access to the LogSearch responses.
 *
 *  Caches are initially empty in this implementation, and a remote request
 *    to the LogSearch server will be made upon the first request for a given
 *    response.
 *
 *
 */
@AmbariService
public class LogSearchDataRetrievalService extends AbstractService {

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

  /**
   * Maximum number of failed attempts that the LogSearch integration code will attempt for
   *   a given component before treating the component as failed and skipping the request.
   *
   */
  private static int MAX_RETRIES_FOR_FAILED_METADATA_REQUEST = 10;

  /**
   * Factory instance used to handle URL string generation requests on the
   *   main request thread.
   */
  @Inject
  private LoggingRequestHelperFactory loggingRequestHelperFactory;

  @Inject
  private Injector injector;

  @Inject
  private Configuration ambariServerConfiguration;

  /**
   * A Cache of host+component names to a set of log files associated with
   *  that Host/Component combination.  This data is retrieved from the
   *  LogSearch server, but cached here for better performance.
   */
  private Cache<String, Set<String>> logFileNameCache;

  /**
   * A Cache of host+component names to a generated URI that
   *  can be used to access the "tail" of a given log file.
   *
   * This data is generated by ambari-server, but cached here to
   *  avoid re-creating these strings upon multiple calls to the
   *  associated HostComponent resource.
   */
  private Cache<String, String> logFileTailURICache;

  /**
   * A set that maintains the current requests being made,
   *  but not yet completed.  This Set will be used to
   *  keep multiple requests from being made for the same
   *  host/component combination.
   *
   */
  private final Set<String> currentRequests = Sets.newConcurrentHashSet();

  /**
   * A map that maintains the set of failure counts for logging
   * metadata requests on a per-component basis.  This map should
   * be consulted prior to making a metadata request to the LogSearch
   * service.  If LogSearch has already returned an empty list for the given
   * component, or any other error has occurred for a certain number of attempts,
   * the request should not be attempted further.
   *
   */
  private final Map<String, AtomicInteger> componentRequestFailureCounts =
    Maps.newConcurrentMap();



  /**
   * Executor instance to be used to run REST queries against
   * the LogSearch service.
   */
  private Executor executor;

  @Override
  protected void doStart() {

    LOG.debug("Initializing caches");

    // obtain the max cache expire time from the ambari configuration
    final int maxTimeoutForCacheInHours =
      ambariServerConfiguration.getLogSearchMetadataCacheExpireTimeout();

    LOG.debug("Caches configured with a max expire timeout of {} hours.", maxTimeoutForCacheInHours);

    // initialize the log file name cache
    logFileNameCache = CacheBuilder.newBuilder().expireAfterWrite(maxTimeoutForCacheInHours, TimeUnit.HOURS).build();
    // initialize the log file tail URI cache
    logFileTailURICache = CacheBuilder.newBuilder().expireAfterWrite(maxTimeoutForCacheInHours, TimeUnit.HOURS).build();

    // initialize the Executor
    executor = Executors.newSingleThreadExecutor();
  }

  @Override
  protected void doStop() {
    LOG.debug("Invalidating LogSearch caches");
    // invalidate the caches
    logFileNameCache.invalidateAll();

    logFileTailURICache.invalidateAll();
  }

  /**
   * This method attempts to obtain the log file names for the specified component
   *   on the specified host.  A cache lookup is first attempted. If the cache does not contain
   *   this data, an asynchronous task is launched in order to make the REST request to
   *   the LogSearch server to obtain this data.
   *
   * Once the data is available in the cache, subsequent calls for a given Host/Component
   *   combination should return non-null.
   *
   * @param component the component name
   * @param host the host name
   * @param cluster the cluster name
   *
   * @return a Set<String> that includes the log file names associated with this Host/Component
   *         combination, or null if that object does not exist in the cache.
   */
  public Set<String> getLogFileNames(String component, String host, String cluster) {
    final String key = generateKey(component, host);

    // check cache for data
    Set<String> cacheResult = logFileNameCache.getIfPresent(key);

    if (cacheResult != null) {
      LOG.debug("LogFileNames result for key = {} found in cache", key);
      return cacheResult;
    } else {
      if (!componentRequestFailureCounts.containsKey(component) || componentRequestFailureCounts.get(component).get() < MAX_RETRIES_FOR_FAILED_METADATA_REQUEST) {
        // queue up a thread to create the LogSearch REST request to obtain this information
        if (currentRequests.contains(key)) {
          LOG.debug("LogFileNames request has been made for key = {}, but not completed yet", key);
        } else {
          LOG.debug("LogFileNames result for key = {} not in cache, queueing up remote request", key);
          // add request key to queue, to keep multiple copies of the same request from
          // being submitted
          currentRequests.add(key);
          startLogSearchFileNameRequest(host, component, cluster);
        }
      } else {
        LOG.debug("Too many failures occurred while attempting to obtain log file metadata for component = {}, Ambari will ignore this component for LogSearch Integration", component);
      }
    }

    return null;
  }

  public String getLogFileTailURI(String baseURI, String component, String host, String cluster) {
    String key = generateKey(component, host);

    String result = logFileTailURICache.getIfPresent(key);
    if (result != null) {
      // return cached result
      return result;
    } else {
      // create URI and add to cache before returning
      if (loggingRequestHelperFactory != null) {
        LoggingRequestHelper helper =
          loggingRequestHelperFactory.getHelper(getController(), cluster);

        if (helper != null) {
          String tailFileURI =
            helper.createLogFileTailURI(baseURI, component, host);

          if (tailFileURI != null) {
            logFileTailURICache.put(key, tailFileURI);
            return tailFileURI;
          }
        }
      } else {
        LOG.debug("LoggingRequestHelperFactory not set on the retrieval service, this probably indicates an error in setup of this service.");
      }
    }

    return null;
  }

  protected void setLoggingRequestHelperFactory(LoggingRequestHelperFactory loggingRequestHelperFactory) {
    this.loggingRequestHelperFactory = loggingRequestHelperFactory;
  }

  /**
   * Package-level setter to facilitate simpler unit testing
   *
   * @param injector
   */
  void setInjector(Injector injector) {
    this.injector = injector;
  }

  /**
   * This protected method provides a way for unit-tests to insert a
   * mock executor for simpler unit-testing.
   *
   * @param executor an Executor instance
   */
  protected void setExecutor(Executor executor) {
    this.executor = executor;
  }

  /**
   * Package-level setter to facilitate simpler unit testing
   *
   * @param ambariServerConfiguration
   */
  void setConfiguration(Configuration ambariServerConfiguration) {
    this.ambariServerConfiguration = ambariServerConfiguration;
  }

  /**
   * This protected method allows for simpler unit tests.
   *
   * @return the Set of current Requests that are not yet completed
   */
  protected Set<String> getCurrentRequests() {
    return currentRequests;
  }

  /**
   * This protected method allows for simpler unit tests.
   *
   * @return the Map of failure counts on a per-component basis
   */
  protected Map<String, AtomicInteger> getComponentRequestFailureCounts() {
    return componentRequestFailureCounts;
  }

  private void startLogSearchFileNameRequest(String host, String component, String cluster) {
    // Create a separate instance of LoggingRequestHelperFactory for
    // each task launched, since these tasks will occur on a separate thread
    // TODO: In a future patch, this should be refactored, to either remove the need
    // TODO: for the separate factory instance at the level of this class, or to make
    // TODO: the LoggingRequestHelperFactory implementation thread-safe, so that
    // TODO: a single factory instance can be shared across multiple threads safely
    executor.execute(new LogSearchFileNameRequestRunnable(host, component, cluster, logFileNameCache, currentRequests,
                                                          injector.getInstance(LoggingRequestHelperFactory.class), componentRequestFailureCounts));
  }

  private AmbariManagementController getController() {
    return AmbariServer.getController();
  }



  private static String generateKey(String component, String host) {
    return component + "+" + host;
  }


  /**
   * A {@link Runnable} used to make requests to the remote LogSearch server's
   *   REST API.
   *
   * This implementation will update a cache shared with the {@link LogSearchDataRetrievalService},
   *   which can then be used for subsequent requests for the same data.
   *
   */
  static class LogSearchFileNameRequestRunnable implements Runnable {

    private final String host;

    private final String component;

    private final String cluster;

    private final Set<String> currentRequests;

    private final Cache<String, Set<String>> logFileNameCache;

    private LoggingRequestHelperFactory loggingRequestHelperFactory;

    private final Map<String, AtomicInteger> componentRequestFailureCounts;

    private AmbariManagementController controller;

    LogSearchFileNameRequestRunnable(String host, String component, String cluster, Cache<String, Set<String>> logFileNameCache, Set<String> currentRequests, LoggingRequestHelperFactory loggingRequestHelperFactory,
                                     Map<String, AtomicInteger> componentRequestFailureCounts) {
      this(host, component, cluster, logFileNameCache, currentRequests, loggingRequestHelperFactory, componentRequestFailureCounts, AmbariServer.getController());
    }

    LogSearchFileNameRequestRunnable(String host, String component, String cluster, Cache<String, Set<String>> logFileNameCache, Set<String> currentRequests,
                                               LoggingRequestHelperFactory loggingRequestHelperFactory, Map<String, AtomicInteger> componentRequestFailureCounts, AmbariManagementController controller) {
      this.host  = host;
      this.component = component;
      this.cluster = cluster;
      this.logFileNameCache = logFileNameCache;
      this.currentRequests = currentRequests;
      this.loggingRequestHelperFactory = loggingRequestHelperFactory;
      this.componentRequestFailureCounts = componentRequestFailureCounts;
      this.controller = controller;
    }

    @Override
    public void run() {
      LOG.debug("LogSearchFileNameRequestRunnable: starting...");
      try {
        LoggingRequestHelper helper =
          loggingRequestHelperFactory.getHelper(controller, cluster);

        if (helper != null) {
          // make request to LogSearch service
          HostLogFilesResponse logFilesResponse = helper.sendGetLogFileNamesRequest(host);
          // update the cache if result is available
          if (logFilesResponse != null && MapUtils.isNotEmpty(logFilesResponse.getHostLogFiles())) {
            LOG.debug("LogSearchFileNameRequestRunnable: request was successful, updating cache");
            // update cache with returned result
            for (Map.Entry<String, List<String>> componentEntry : logFilesResponse.getHostLogFiles().entrySet()) {
              final String key = generateKey(componentEntry.getKey(), host);
              logFileNameCache.put(key, new HashSet<>(componentEntry.getValue()));
            }
          } else {
            LOG.debug("LogSearchFileNameRequestRunnable: remote request was not successful for component = {} on host ={}", component, host);
            if (!componentRequestFailureCounts.containsKey(component)) {
              componentRequestFailureCounts.put(component, new AtomicInteger());
            }

            // increment the failure count for this component
            componentRequestFailureCounts.get(component).incrementAndGet();
          }
        } else {
          LOG.debug("LogSearchFileNameRequestRunnable: request helper was null.  This may mean that LogSearch is not available, or could be a potential connection problem.");
        }
      } finally {
        // since request has completed (either successfully or not),
        // remove this host/component key from the current requests
        currentRequests.remove(generateKey(component, host));
      }
    }

    protected void setLoggingRequestHelperFactory(LoggingRequestHelperFactory loggingRequestHelperFactory) {
      this.loggingRequestHelperFactory = loggingRequestHelperFactory;
    }

    protected void setAmbariManagementController(AmbariManagementController controller) {
      this.controller = controller;
    }


  }


}
