/*
 * Copyright DataStax, Inc.
 *
 * Licensed 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 com.datastax.oss.driver.internal.core.metrics;

import com.codahale.metrics.Reservoir;
import com.codahale.metrics.Snapshot;
import java.io.OutputStream;
import java.time.Duration;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import net.jcip.annotations.GuardedBy;
import net.jcip.annotations.ThreadSafe;
import org.HdrHistogram.Histogram;
import org.HdrHistogram.Recorder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

A reservoir implementation backed by the HdrHistogram library.
It uses a Recorder to capture snapshots at a configurable interval: calls to update(long) are recorded in a "live" histogram, while getSnapshot() is based on a "cached", read-only histogram. Each time the cached histogram becomes older than the interval, the two histograms are switched (therefore statistics won't be available during the first interval after initialization, since we don't have a cached histogram yet). 
Note that this class does not implement size(). 
See Also:
HdrHistogram/**
 * A reservoir implementation backed by the HdrHistogram library.
 *
 * <p>It uses a {@link Recorder} to capture snapshots at a configurable interval: calls to {@link
 * #update(long)} are recorded in a "live" histogram, while {@link #getSnapshot()} is based on a
 * "cached", read-only histogram. Each time the cached histogram becomes older than the interval,
 * the two histograms are switched (therefore statistics won't be available during the first
 * interval after initialization, since we don't have a cached histogram yet).
 *
 * <p>Note that this class does not implement {@link #size()}.
 *
 * @see <a href="http://hdrhistogram.github.io/HdrHistogram/">HdrHistogram</a>
 */
@ThreadSafe
public class HdrReservoir implements Reservoir {

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

  private final String logPrefix;
  private final Recorder recorder;
  private final long refreshIntervalNanos;

  // The lock only orchestrates `getSnapshot()` calls; `update()` is fed directly to the recorder,
  // which is lock-free. `getSnapshot()` calls are comparatively rare, so locking is not a
  // bottleneck.
  private final ReadWriteLock cacheLock = new ReentrantReadWriteLock();

  @GuardedBy("cacheLock")
  private Histogram cachedHistogram;

  @GuardedBy("cacheLock")
  private long cachedHistogramTimestampNanos;

  @GuardedBy("cacheLock")
  private Snapshot cachedSnapshot;

  public HdrReservoir(
      Duration highestTrackableLatency,
      int numberOfSignificantValueDigits,
      Duration refreshInterval,
      String logPrefix) {
    this.logPrefix = logPrefix;
    // The Reservoir interface is supposed to be agnostic to the unit. However, the Metrics library
    // heavily leans towards nanoseconds (for example, Timer feeds nanoseconds to update(); JmxTimer
    // assumes that the snapshot results are in nanoseconds).
    // In our case, microseconds are precise enough for request metrics, and we don't want to waste
    // space unnecessarily. So we simply use microseconds for our internal storage, and do the
    // conversion when needed.
    this.recorder =
        new Recorder(highestTrackableLatency.toNanos() / 1000, numberOfSignificantValueDigits);
    this.refreshIntervalNanos = refreshInterval.toNanos();
    this.cachedHistogramTimestampNanos = System.nanoTime();
    this.cachedSnapshot = EMPTY_SNAPSHOT;
  }

  @Override
  public void update(long value) {
    try {
      recorder.recordValue(value / 1000);
    } catch (ArrayIndexOutOfBoundsException e) {
      LOG.warn("[{}] Recorded value ({}) is out of bounds, discarding", logPrefix, value);
    }
  }

  
Not implemented: this reservoir implementation is intended for use with a Histogram, which doesn't use this method. 
(original description: {@inheritDoc})
/**
   * <em>Not implemented</em>: this reservoir implementation is intended for use with a {@link
   * com.codahale.metrics.Histogram}, which doesn't use this method.
   *
   * <p>(original description: {@inheritDoc})
   */
  @Override
  public int size() {
    throw new UnsupportedOperationException("HdrReservoir does not implement size()");
  }

  
{@inheritDoc}
Note that the snapshots returned from this method do not implement Snapshot.getValues() nor Snapshot.dump(OutputStream). In addition, due to the way that internal data structures are recycled, you should not hold onto a snapshot for more than the refresh interval; one way to ensure this is to never cache the result of this method. /**
   * {@inheritDoc}
   *
   * <p>Note that the snapshots returned from this method do not implement {@link
   * Snapshot#getValues()} nor {@link Snapshot#dump(OutputStream)}. In addition, due to the way that
   * internal data structures are recycled, you should not hold onto a snapshot for more than the
   * refresh interval; one way to ensure this is to never cache the result of this method.
   */
  @Override
  public Snapshot getSnapshot() {
    long now = System.nanoTime();

    cacheLock.readLock().lock();
    try {
      if (now - cachedHistogramTimestampNanos < refreshIntervalNanos) {
        return cachedSnapshot;
      }
    } finally {
      cacheLock.readLock().unlock();
    }

    cacheLock.writeLock().lock();
    try {
      // Might have raced with another writer => re-check the timestamp
      if (now - cachedHistogramTimestampNanos >= refreshIntervalNanos) {
        LOG.debug("Cached snapshot is too old, refreshing");
        cachedHistogram = recorder.getIntervalHistogram(cachedHistogram);
        cachedSnapshot = new HdrSnapshot(cachedHistogram);
        cachedHistogramTimestampNanos = now;
      }
      return cachedSnapshot;
    } finally {
      cacheLock.writeLock().unlock();
    }
  }

  private class HdrSnapshot extends Snapshot {

    private final Histogram histogram;
    private final double meanNanos;
    private final double stdDevNanos;

    private HdrSnapshot(Histogram histogram) {
      this.histogram = histogram;

      // Cache those values because they rely on HdrHistogram's internal iterators, which are not
      // safe if the snapshot is accessed by concurrent reporters.
      // In contrast, getMin(), getMax() and getValue() are safe.
      this.meanNanos = histogram.getMean() * 1000;
      this.stdDevNanos = histogram.getStdDeviation() * 1000;
    }

    @Override
    public double getValue(double quantile) {
      return histogram.getValueAtPercentile(quantile * 100) * 1000;
    }

    
Not implemented: this reservoir implementation is intended for use with a Histogram, which doesn't use this method. 
(original description: {@inheritDoc})
/**
     * <em>Not implemented</em>: this reservoir implementation is intended for use with a {@link
     * com.codahale.metrics.Histogram}, which doesn't use this method.
     *
     * <p>(original description: {@inheritDoc})
     */
    @Override
    public long[] getValues() {
      // This can be implemented, but we ran into issues when accessed by concurrent reporters
      // because HdrHistogram uses an unsafe shared iterator.
      // So throwing instead since this method should be seldom used anyway.
      throw new UnsupportedOperationException(
          "HdrReservoir's snapshots do not implement getValues()");
    }

    @Override
    public int size() {
      long longSize = histogram.getTotalCount();
      // The Metrics API requires an int. It's very unlikely that we get an overflow here, unless
      // the refresh interval is ridiculously high (at 10k requests/s, it would have to be more than
      // 59 hours). However handle gracefully just in case.
      int size;
      if (longSize > Integer.MAX_VALUE) {
        LOG.warn("[{}] Too many recorded values, truncating", logPrefix);
        size = Integer.MAX_VALUE;
      } else {
        size = (int) longSize;
      }
      return size;
    }

    @Override
    public long getMax() {
      return histogram.getMaxValue() * 1000;
    }

    @Override
    public double getMean() {
      return meanNanos;
    }

    @Override
    public long getMin() {
      return histogram.getMinValue() * 1000;
    }

    @Override
    public double getStdDev() {
      return stdDevNanos;
    }

    
Not implemented: this reservoir implementation is intended for use with a Histogram, which doesn't use this method. 
(original description: {@inheritDoc})
/**
     * <em>Not implemented</em>: this reservoir implementation is intended for use with a {@link
     * com.codahale.metrics.Histogram}, which doesn't use this method.
     *
     * <p>(original description: {@inheritDoc})
     */
    @Override
    public void dump(OutputStream output) {
      throw new UnsupportedOperationException("HdrReservoir's snapshots do not implement dump()");
    }
  }

  private static final Snapshot EMPTY_SNAPSHOT =
      new Snapshot() {
        @Override
        public double getValue(double quantile) {
          return 0;
        }

        @Override
        public long[] getValues() {
          return new long[0];
        }

        @Override
        public int size() {
          return 0;
        }

        @Override
        public long getMax() {
          return 0;
        }

        @Override
        public double getMean() {
          return 0;
        }

        @Override
        public long getMin() {
          return 0;
        }

        @Override
        public double getStdDev() {
          return 0;
        }

        @Override
        public void dump(OutputStream output) {
          // nothing to do
        }
      };
}