content: /*
   * 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.controller.metrics.timeline.cache;
  
  import java.io.IOException;
  import java.net.URISyntaxException;
  import java.util.Date;
  import java.util.Map;
  import java.util.TreeMap;
  
  import org.apache.ambari.server.configuration.ComponentSSLConfiguration;
  import org.apache.ambari.server.configuration.Configuration;
  import org.apache.ambari.server.controller.internal.URLStreamProvider;
  import org.apache.ambari.server.controller.metrics.timeline.MetricsRequestHelper;
  import org.apache.ambari.server.controller.spi.TemporalInfo;
  import org.apache.hadoop.metrics2.sink.timeline.Precision;
  import org.apache.hadoop.metrics2.sink.timeline.TimelineMetric;
  import org.apache.hadoop.metrics2.sink.timeline.TimelineMetrics;
  import org.apache.http.client.utils.URIBuilder;
  import org.ehcache.spi.loaderwriter.BulkCacheLoadingException;
  import org.ehcache.spi.loaderwriter.BulkCacheWritingException;
  import org.ehcache.spi.loaderwriter.CacheLoaderWriter;
  import org.ehcache.spi.loaderwriter.CacheWritingException;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import com.google.inject.Inject;
  import com.google.inject.Singleton;
  
  
  @Singleton
  public class TimelineMetricCacheEntryFactory implements CacheLoaderWriter<TimelineAppMetricCacheKey, TimelineMetricsCacheValue> {
    private final static Logger LOG = LoggerFactory.getLogger(TimelineMetricCacheEntryFactory.class);
    // Not declared final to ease unit test code and allow streamProvider
    // injection
    private MetricsRequestHelper requestHelperForGets;
    private MetricsRequestHelper requestHelperForUpdates;
    private final Long BUFFER_TIME_DIFF_CATCHUP_INTERVAL;
  
    @Inject
    public TimelineMetricCacheEntryFactory(Configuration configuration) {
      // Longer timeout for first cache miss
      requestHelperForGets = new MetricsRequestHelper(new URLStreamProvider(
        configuration.getMetricsRequestConnectTimeoutMillis(),
        configuration.getMetricsRequestReadTimeoutMillis(),
        ComponentSSLConfiguration.instance()));
  
      // Timeout setting different from first request timeout
      // Allows stale data to be returned at the behest of performance.
      requestHelperForUpdates = new MetricsRequestHelper(new URLStreamProvider(
        configuration.getMetricsRequestConnectTimeoutMillis(),
        configuration.getMetricsRequestIntervalReadTimeoutMillis(),
        ComponentSSLConfiguration.instance()));
  
      BUFFER_TIME_DIFF_CATCHUP_INTERVAL = configuration.getMetricRequestBufferTimeCatchupInterval();
    }
  
    /**
     * This method is called on a get element from cache call when key is not
     * found in cache, returns a value for the key to be cached.
     *
     * @param key @org.apache.ambari.server.controller.metrics.timeline.cache.TimelineAppMetricCacheKey
     * @return @org.apache.hadoop.metrics2.sink.timeline.TimelineMetrics
     * @throws Exception
     */
    @Override
    public TimelineMetricsCacheValue load(TimelineAppMetricCacheKey key) throws Exception {
      LOG.debug("Creating cache entry since none exists, key = {}", key);
      TimelineAppMetricCacheKey metricCacheKey = key;
  
      TimelineMetrics timelineMetrics = null;
      try {
        URIBuilder uriBuilder = new URIBuilder(metricCacheKey.getSpec());
        timelineMetrics = requestHelperForGets.fetchTimelineMetrics(uriBuilder,
          metricCacheKey.getTemporalInfo().getStartTimeMillis(),
          metricCacheKey.getTemporalInfo().getEndTimeMillis());
      } catch (URISyntaxException e) {
        LOG.debug("Caught URISyntaxException on fetching metrics. {}", e.getMessage());
        throw new RuntimeException(e);
      } catch (IOException io) {
        LOG.debug("Caught IOException on fetching metrics. {}", io.getMessage());
        throw io;
      }
  
      TimelineMetricsCacheValue value = null;
      if (timelineMetrics != null && !timelineMetrics.getMetrics().isEmpty()) {
        value = new TimelineMetricsCacheValue(
          metricCacheKey.getTemporalInfo().getStartTime(),
          metricCacheKey.getTemporalInfo().getEndTime(),
          timelineMetrics, // Null or empty should prompt a refresh
          Precision.getPrecision(metricCacheKey.getTemporalInfo().getStartTimeMillis(),
            metricCacheKey.getTemporalInfo().getEndTimeMillis()) //Initial Precision
        );
        LOG.debug("Created cache entry: {}", value);
      }
      return value;
    }
  
    /**
     * Called on a get call for existing values in the cache,
     * the necessary locking code is present in the get call and this call
     * should update the value of the cache entry before returning.
     *
     * @param key @org.apache.ambari.server.controller.metrics.timeline.cache.TimelineAppMetricCacheKey
     * @param value @org.apache.hadoop.metrics2.sink.timeline.TimelineMetrics
     * @throws Exception
     */
    @Override
    public void write(TimelineAppMetricCacheKey key, TimelineMetricsCacheValue value) throws Exception {
      TimelineAppMetricCacheKey metricCacheKey = key;
      TimelineMetricsCacheValue existingMetrics = value;
  
      LOG.debug("Updating cache entry, key: {}, with value = {}", key, value);
  
      Long existingSeriesStartTime = existingMetrics.getStartTime();
      Long existingSeriesEndTime = existingMetrics.getEndTime();
  
      TemporalInfo newTemporalInfo = metricCacheKey.getTemporalInfo();
      Long requestedStartTime = newTemporalInfo.getStartTimeMillis();
      Long requestedEndTime = newTemporalInfo.getEndTimeMillis();
  
      // Calculate new start and end times
      URIBuilder uriBuilder = new URIBuilder(metricCacheKey.getSpec());
  
      Precision requestedPrecision = Precision.getPrecision(requestedStartTime, requestedEndTime);
      Precision currentPrecision = existingMetrics.getPrecision();
  
      Long newStartTime = null;
      Long newEndTime = null;
      if(!requestedPrecision.equals(currentPrecision)) {
        // Ignore cache entry. Get the entire data from the AMS and update the cache.
        LOG.debug("Precision changed from {} to {}", currentPrecision, requestedPrecision);
        newStartTime = requestedStartTime;
        newEndTime = requestedEndTime;
      } else {
        //Get only the metric values for the delta period from the cache.
        LOG.debug("No change in precision {}", currentPrecision);
        newStartTime = getRefreshRequestStartTime(existingSeriesStartTime,
            existingSeriesEndTime, requestedStartTime);
        newEndTime = getRefreshRequestEndTime(existingSeriesStartTime,
            existingSeriesEndTime, requestedEndTime);
      }
  
      // Cover complete overlap scenario
      // time axis: |-------- exSt ----- reqSt ------ reqEnd ----- exEnd ---------|
      if (newEndTime > newStartTime &&
         !((newStartTime.equals(existingSeriesStartTime) &&
         newEndTime.equals(existingSeriesEndTime)) && requestedPrecision.equals(currentPrecision)) ) {
  
        LOG.debug("Existing cached timeseries startTime = {}, endTime = {}",
          new Date(getMillisecondsTime(existingSeriesStartTime)), new Date(getMillisecondsTime(existingSeriesEndTime)));
  
        LOG.debug("Requested timeseries startTime = {}, endTime = {}",
          new Date(getMillisecondsTime(newStartTime)), new Date(getMillisecondsTime(newEndTime)));
  
        // Update spec with new start and end time
        uriBuilder.setParameter("startTime", String.valueOf(newStartTime));
        uriBuilder.setParameter("endTime", String.valueOf(newEndTime));
        uriBuilder.setParameter("precision",requestedPrecision.toString());
  
        try {
          TimelineMetrics newTimeSeries = requestHelperForUpdates.fetchTimelineMetrics(uriBuilder, newStartTime, newEndTime);
  
          // Update existing time series with new values
          updateTimelineMetricsInCache(newTimeSeries, existingMetrics,
            getMillisecondsTime(requestedStartTime),
            getMillisecondsTime(requestedEndTime), !currentPrecision.equals(requestedPrecision));
  
          // Replace old boundary values
          existingMetrics.setStartTime(requestedStartTime);
          existingMetrics.setEndTime(requestedEndTime);
          existingMetrics.setPrecision(requestedPrecision);
  
        } catch (IOException io) {
          if (LOG.isDebugEnabled()) {
            LOG.debug("Exception retrieving metrics.", io);
          }
          throw io;
        }
      } else {
        LOG.debug("Skip updating cache with new startTime = {}, new endTime = {}",
          new Date(getMillisecondsTime(newStartTime)), new Date(getMillisecondsTime(newEndTime)));
      }
    }
  
    /**
     * Update cache with new timeseries data
     */
    protected void updateTimelineMetricsInCache(TimelineMetrics newMetrics,
        TimelineMetricsCacheValue timelineMetricsCacheValue,
        Long requestedStartTime, Long requestedEndTime, boolean removeAll) {
      TimelineMetrics existingTimelineMetrics = timelineMetricsCacheValue.getTimelineMetrics();
  
      // Remove values that do not fit before adding new data
      updateExistingMetricValues(existingTimelineMetrics, requestedStartTime,
        requestedEndTime, removeAll);
  
      if (newMetrics != null && !newMetrics.getMetrics().isEmpty()) {
        for (TimelineMetric timelineMetric : newMetrics.getMetrics()) {
          if (LOG.isTraceEnabled()) {
            TreeMap<Long, Double> sortedMetrics = new TreeMap<>(timelineMetric.getMetricValues());
  
            LOG.trace("New metric: {} # {}, startTime = {}, endTime = {}",
              timelineMetric.getMetricName(), timelineMetric.getMetricValues().size(), sortedMetrics.firstKey(), sortedMetrics.lastKey());
          }
  
          TimelineMetric existingMetric = null;
  
          for (TimelineMetric metric : existingTimelineMetrics.getMetrics()) {
            if (metric.equalsExceptTime(timelineMetric)) {
              existingMetric = metric;
            }
          }
  
          if (existingMetric != null) {
            // Add new ones
            existingMetric.getMetricValues().putAll(timelineMetric.getMetricValues());
  
            if (LOG.isTraceEnabled()) {
              TreeMap<Long, Double> sortedMetrics = new TreeMap<>(existingMetric.getMetricValues());
              LOG.trace("Merged metric: {}, Final size: {}, startTime = {}, endTime = {}",
                timelineMetric.getMetricName(), existingMetric.getMetricValues().size(), sortedMetrics.firstKey(), sortedMetrics.lastKey());
            }
          } else {
            existingTimelineMetrics.getMetrics().add(timelineMetric);
          }
        }
      }
    }
  
    // Remove out of band data from the cache
    private void updateExistingMetricValues(TimelineMetrics existingMetrics,
        Long requestedStartTime, Long requestedEndTime, boolean removeAll) {
      for (TimelineMetric existingMetric : existingMetrics.getMetrics()) {
        if (removeAll) {
          existingMetric.setMetricValues(new TreeMap<>());
        } else {
          TreeMap<Long, Double> existingMetricValues = existingMetric.getMetricValues();
          LOG.trace("Existing metric: {} # {}", existingMetric.getMetricName(), existingMetricValues.size());
  
          // Retain only the values that are within the [requestStartTime, requestedEndTime] window
          existingMetricValues.headMap(requestedStartTime,false).clear();
          existingMetricValues.tailMap(requestedEndTime, false).clear();
        }
      }
    }
  
    // Scenario: Regular graph updates
    // time axis: |-------- exSt ----- reqSt ------ exEnd ----- reqEnd ---------|
    // Scenario: Selective graph updates
    // time axis: |-------- exSt ----- exEnd ------ reqSt ----- reqEnd ---------|
    // Scenario: Extended time window
    // time axis: |-------- reSt ----- exSt ------- extEnd ---- reqEnd ---------|
    protected Long getRefreshRequestStartTime(Long existingSeriesStartTime,
        Long existingSeriesEndTime, Long requestedStartTime) {
      Long diff = requestedStartTime - existingSeriesEndTime;
      Long startTime = requestedStartTime;
  
      if (diff < 0 && requestedStartTime > existingSeriesStartTime) {
        // Regular graph updates
        // Overlapping timeseries data refresh only new part
        // Account for missing data on the trailing edge due to buffering
        startTime = getTimeShiftedStartTime(existingSeriesEndTime);
      }
  
      LOG.trace("Requesting timeseries data with new startTime = {}", new Date(getMillisecondsTime(startTime)));
  
      return startTime;
    }
  
    // Scenario: Regular graph updates
    // time axis: |-------- exSt ----- reqSt ------ exEnd ----- reqEnd ---------|
    // Scenario: Old data request /w overlap
    // time axis: |-------- reqSt ----- exSt ------ reqEnd ----- extEnd --------|
    // Scenario: Very Old data request /wo overlap
    // time axis: |-------- reqSt ----- reqEnd ------ exSt ----- extEnd --------|
    protected Long getRefreshRequestEndTime(Long existingSeriesStartTime,
        Long existingSeriesEndTime, Long requestedEndTime) {
      Long endTime = requestedEndTime;
      Long diff = requestedEndTime - existingSeriesEndTime;
      if (diff < 0 && requestedEndTime > existingSeriesStartTime) {
        // End time overlaps existing timeseries
        // Get only older data that might not be in the cache
        endTime = existingSeriesStartTime;
      }
  
      LOG.trace("Requesting timeseries data with new endTime = {}", new Date(getMillisecondsTime(endTime)));
      return endTime;
    }
  
    /**
     * Time shift by a constant taking into account Epoch vs millis
     */
    private long getTimeShiftedStartTime(long startTime) {
      if (startTime < 9999999999l) {
        // Epoch time
        return startTime - (BUFFER_TIME_DIFF_CATCHUP_INTERVAL / 1000);
      } else {
        return startTime - BUFFER_TIME_DIFF_CATCHUP_INTERVAL;
      }
    }
  
    private long getMillisecondsTime(long time) {
      if (time < 9999999999l) {
        return time * 1000;
      } else {
        return time;
      }
    }
    @Override
    public void delete(TimelineAppMetricCacheKey key) throws CacheWritingException {
      // no need to implement.
    }
  
    @Override
    public Map<TimelineAppMetricCacheKey, TimelineMetricsCacheValue> loadAll(Iterable<? extends TimelineAppMetricCacheKey> keys) throws BulkCacheLoadingException, Exception {
  //    no need to implement.
      return null;
    }
  
    @Override
    public void writeAll(Iterable<? extends Map.Entry<? extends TimelineAppMetricCacheKey, ? extends TimelineMetricsCacheValue>> entries) throws BulkCacheWritingException, Exception {
  //    no need to implement.
    }
  
    @Override
    public void deleteAll(Iterable<? extends TimelineAppMetricCacheKey> keys) throws BulkCacheWritingException, Exception {
  //    no need to implement.
    }
  }
  