package io.prometheus.metrics.core.metrics;

import io.prometheus.metrics.annotations.StableApi;
import io.prometheus.metrics.config.ExemplarsProperties;
import io.prometheus.metrics.config.MetricsProperties;
import io.prometheus.metrics.config.PrometheusProperties;
import io.prometheus.metrics.core.datapoints.DistributionDataPoint;
import io.prometheus.metrics.core.exemplars.ExemplarSampler;
import io.prometheus.metrics.core.exemplars.ExemplarSamplerConfig;
import io.prometheus.metrics.core.util.Scheduler;
import io.prometheus.metrics.model.registry.MetricType;
import io.prometheus.metrics.model.snapshots.ClassicHistogramBuckets;
import io.prometheus.metrics.model.snapshots.Exemplars;
import io.prometheus.metrics.model.snapshots.HistogramSnapshot;
import io.prometheus.metrics.model.snapshots.Labels;
import io.prometheus.metrics.model.snapshots.NativeHistogramBuckets;
import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.SortedSet;
import java.util.TreeSet;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.DoubleAdder;
import java.util.concurrent.atomic.LongAdder;
import java.util.function.Supplier;
import javax.annotation.Nullable;

/**
 * Histogram metric. Example usage:
 *
 * <pre>{@code
 * Histogram histogram = Histogram.builder()
 *         .name("http_request_duration_seconds")
 *         .help("HTTP request service time in seconds")
 *         .unit(SECONDS)
 *         .labelNames("method", "path", "status_code")
 *         .register();
 *
 * long start = System.nanoTime();
 * // do something
 * histogram.labelValues("GET", "/", "200").observe(Unit.nanosToSeconds(System.nanoTime() - start));
 * }</pre>
 *
 * Prometheus supports two internal representations of histograms:
 *
 * <ol>
 *   <li><i>Classic Histograms</i> have a fixed number of buckets with fixed bucket boundaries.
 *   <li><i>Native Histograms</i> have an infinite number of buckets with a dynamic resolution.
 *       Prometheus native histograms are the same as OpenTelemetry's exponential histograms.
 * </ol>
 *
 * By default, a histogram maintains both representations, i.e. the example above will maintain a
 * classic histogram representation with Prometheus' default bucket boundaries as well as native
 * histogram representation. Which representation is used depends on the exposition format, i.e.
 * which content type the Prometheus server accepts when scraping. Exposition format "Text" exposes
 * the classic histogram, exposition format "Protobuf" exposes both representations. This is great
 * for migrating from classic histograms to native histograms.
 *
 * <p>If you want the classic representation only, use {@link Histogram.Builder#classicOnly}. If you
 * want the native representation only, use {@link Histogram.Builder#nativeOnly}.
 */
@StableApi
public class Histogram extends StatefulMetric<DistributionDataPoint, Histogram.DataPoint>
    implements DistributionDataPoint {

  // nativeSchema == CLASSIC_HISTOGRAM indicates that this is a classic histogram only.
  private static final int CLASSIC_HISTOGRAM = Integer.MIN_VALUE;

  // NATIVE_BOUNDS is used to look up the native bucket index depending on the current schema.
  private static final double[][] NATIVE_BOUNDS;

  @Nullable private final ExemplarSamplerConfig exemplarSamplerConfig;
  @Nullable private final Supplier<Labels> exemplarLabelsSupplier;

  // Upper bounds for the classic histogram buckets. Contains at least +Inf.
  // An empty array indicates that this is a native histogram only.
  private final double[] classicUpperBounds;

  // The schema defines the resolution of the native histogram.
  // Schema is Prometheus terminology, in OpenTelemetry it's named "scale".
  // The formula for the bucket boundaries at position "index" is:
  //
  // base := base = (2^(2^-scale))
  // lowerBound := base^(index-1)
  // upperBound := base^(index)
  //
  // Note that this is off-by-one compared to OpenTelemetry.
  //
  // Example: With schema 0 the bucket boundaries are ... 1/16, 1/8, 1/4, 1/2, 1, 2, 4, 8, 16, ...
  // Each increment in schema doubles the number of buckets.
  //
  // The initialNativeSchema is the schema we start with. The histogram will automatically scale
  // down
  // if the number of native histogram buckets exceeds nativeMaxBuckets.
  private final int nativeInitialSchema; // integer in [-4, 8]

  // Native histogram buckets get smaller and smaller the closer they get to zero.
  // To avoid wasting a lot of buckets for observations fluctuating around zero, we consider all
  // values in [-zeroThreshold, +zeroThreshold] to be equal to zero.
  //
  // The zeroThreshold is initialized with minZeroThreshold, and will grow up to maxZeroThreshold if
  // the number of native histogram buckets exceeds nativeMaxBuckets.
  private final double nativeMinZeroThreshold;
  private final double nativeMaxZeroThreshold;

  // When the number of native histogram buckets becomes larger than nativeMaxBuckets,
  // an attempt is made to reduce the number of buckets:
  // (1) Reset if the last reset is longer than the reset duration ago
  // (2) Increase the zero bucket width if it's smaller than nativeMaxZeroThreshold
  // (3) Decrease the nativeSchema, i.e. merge pairs of neighboring buckets into one
  private final int nativeMaxBuckets;

  // If the number of native histogram buckets exceeds nativeMaxBuckets,
  // the histogram may reset (all values set to zero) after nativeResetDurationSeconds is expired.
  private final long nativeResetDurationSeconds; // 0 indicates no reset

  private Histogram(Histogram.Builder builder, PrometheusProperties prometheusProperties) {
    super(builder);
    MetricsProperties[] properties = getMetricProperties(builder, prometheusProperties);
    nativeInitialSchema =
        getConfigProperty(
            properties,
            props -> {
              if (Boolean.TRUE.equals(props.getHistogramClassicOnly())) {
                return CLASSIC_HISTOGRAM;
              } else {
                return props.getHistogramNativeInitialSchema();
              }
            });
    classicUpperBounds =
        getConfigProperty(
            properties,
            props -> {
              if (Boolean.TRUE.equals(props.getHistogramNativeOnly())) {
                return new double[] {};
              } else if (props.getHistogramClassicUpperBounds() != null) {
                SortedSet<Double> upperBounds =
                    new TreeSet<>(props.getHistogramClassicUpperBounds());
                upperBounds.add(Double.POSITIVE_INFINITY);
                double[] result = new double[upperBounds.size()];
                int i = 0;
                for (double upperBound : upperBounds) {
                  result[i++] = upperBound;
                }
                return result;
              } else {
                return null;
              }
            });
    double max =
        getConfigProperty(properties, MetricsProperties::getHistogramNativeMaxZeroThreshold);
    double min =
        getConfigProperty(properties, MetricsProperties::getHistogramNativeMinZeroThreshold);
    nativeMaxZeroThreshold =
        max == Builder.DEFAULT_NATIVE_MAX_ZERO_THRESHOLD && min > max ? min : max;
    nativeMinZeroThreshold = Math.min(min, nativeMaxZeroThreshold);
    nativeMaxBuckets =
        getConfigProperty(properties, MetricsProperties::getHistogramNativeMaxNumberOfBuckets);
    nativeResetDurationSeconds =
        getConfigProperty(properties, MetricsProperties::getHistogramNativeResetDurationSeconds);
    boolean exemplarsEnabled =
        getConfigProperty(properties, MetricsProperties::getExemplarsEnabled);
    ExemplarsProperties exemplarsProperties = prometheusProperties.getExemplarProperties();
    if (exemplarsEnabled) {
      exemplarSamplerConfig =
          classicUpperBounds.length == 0
              ? new ExemplarSamplerConfig(exemplarsProperties, 4)
              : new ExemplarSamplerConfig(exemplarsProperties, classicUpperBounds);
    } else {
      exemplarSamplerConfig = null;
    }
    exemplarLabelsSupplier = builder.exemplarLabelsSupplier;
  }

  @Override
  public double getSum() {
    return getNoLabels().getSum();
  }

  @Override
  public long getCount() {
    return getNoLabels().getCount();
  }

  @Override
  public void observe(double amount) {
    getNoLabels().observe(amount);
  }

  @Override
  public void observeWithExemplar(double amount, Labels labels) {
    getNoLabels().observeWithExemplar(amount, labels);
  }

  public class DataPoint implements DistributionDataPoint {
    private final LongAdder[] classicBuckets;
    private final ConcurrentHashMap<Integer, LongAdder> nativeBucketsForPositiveValues =
        new ConcurrentHashMap<>();
    private final ConcurrentHashMap<Integer, LongAdder> nativeBucketsForNegativeValues =
        new ConcurrentHashMap<>();
    private final LongAdder nativeZeroCount = new LongAdder();
    private final LongAdder count = new LongAdder();
    private final DoubleAdder sum = new DoubleAdder();
    private volatile int nativeSchema =
        nativeInitialSchema; // integer in [-4, 8] or CLASSIC_HISTOGRAM
    private volatile double nativeZeroThreshold = Histogram.this.nativeMinZeroThreshold;
    private volatile long createdTimeMillis = System.currentTimeMillis();
    private final Buffer buffer = new Buffer();
    private volatile boolean resetDurationExpired = false;
    @Nullable private final ExemplarSampler exemplarSampler;

    private DataPoint() {
      if (exemplarSamplerConfig != null) {
        exemplarSampler = new ExemplarSampler(exemplarSamplerConfig, null, exemplarLabelsSupplier);
      } else {
        exemplarSampler = null;
      }
      classicBuckets = new LongAdder[classicUpperBounds.length];
      for (int i = 0; i < classicUpperBounds.length; i++) {
        classicBuckets[i] = new LongAdder();
      }
      maybeScheduleNextReset();
    }

    @Override
    public double getSum() {
      return sum.sum();
    }

    @Override
    public long getCount() {
      return count.sum();
    }

    @Override
    public void observe(double value) {
      if (Double.isNaN(value)) {
        // See https://github.com/prometheus/client_golang/issues/1275 on ignoring NaN observations.
        return;
      }
      if (!buffer.append(value)) {
        doObserve(value, false);
      }
      if (exemplarSampler != null) {
        exemplarSampler.observe(value);
      }
    }

    @Override
    public void observeWithExemplar(double value, Labels labels) {
      if (Double.isNaN(value)) {
        // See https://github.com/prometheus/client_golang/issues/1275 on ignoring NaN observations.
        return;
      }
      if (!buffer.append(value)) {
        doObserve(value, false);
      }
      if (exemplarSampler != null) {
        exemplarSampler.observeWithExemplar(value, labels);
      }
    }

    private void doObserve(double value, boolean fromBuffer) {
      // classicUpperBounds is an empty array if this is a native histogram only.
      for (int i = 0; i < classicUpperBounds.length; ++i) {
        // The last bucket is +Inf, so we always increment.
        if (value <= classicUpperBounds[i]) {
          classicBuckets[i].add(1);
          break;
        }
      }
      boolean nativeBucketCreated = false;
      if (Histogram.this.nativeInitialSchema != CLASSIC_HISTOGRAM) {
        if (value > nativeZeroThreshold) {
          nativeBucketCreated = addToNativeBucket(value, nativeBucketsForPositiveValues);
        } else if (value < -nativeZeroThreshold) {
          nativeBucketCreated = addToNativeBucket(-value, nativeBucketsForNegativeValues);
        } else {
          nativeZeroCount.add(1);
        }
      }
      sum.add(value);
      count
          .increment(); // must be the last step, because count is used to signal that the operation
      // is complete.
      if (!fromBuffer) {
        // maybeResetOrScaleDown will switch to the buffer,
        // which won't work if we are currently still processing observations from the buffer.
        // The reason is that before switching to the buffer we wait for all pending observations to
        // be counted.
        // If we do this while still applying observations from the buffer, the pending observations
        // from
        // the buffer will never be counted, and the buffer.run() method will wait forever.
        maybeResetOrScaleDown(value, nativeBucketCreated);
      }
    }

    private HistogramSnapshot.HistogramDataPointSnapshot collect(Labels labels) {
      Exemplars exemplars = exemplarSampler != null ? exemplarSampler.collect() : Exemplars.EMPTY;
      return buffer.run(
          expectedCount -> count.sum() == expectedCount,
          () -> {
            if (classicUpperBounds.length == 0) {
              // native only
              return new HistogramSnapshot.HistogramDataPointSnapshot(
                  nativeSchema,
                  nativeZeroCount.sum(),
                  nativeZeroThreshold,
                  toBucketList(nativeBucketsForPositiveValues),
                  toBucketList(nativeBucketsForNegativeValues),
                  sum.sum(),
                  labels,
                  exemplars,
                  createdTimeMillis);
            } else if (Histogram.this.nativeInitialSchema == CLASSIC_HISTOGRAM) {
              // classic only
              return new HistogramSnapshot.HistogramDataPointSnapshot(
                  ClassicHistogramBuckets.of(classicUpperBounds, classicBuckets),
                  sum.sum(),
                  labels,
                  exemplars,
                  createdTimeMillis);
            } else {
              // hybrid: classic and native
              return new HistogramSnapshot.HistogramDataPointSnapshot(
                  ClassicHistogramBuckets.of(classicUpperBounds, classicBuckets),
                  nativeSchema,
                  nativeZeroCount.sum(),
                  nativeZeroThreshold,
                  toBucketList(nativeBucketsForPositiveValues),
                  toBucketList(nativeBucketsForNegativeValues),
                  sum.sum(),
                  labels,
                  exemplars,
                  createdTimeMillis);
            }
          },
          v -> doObserve(v, true));
    }

    private boolean addToNativeBucket(double value, ConcurrentHashMap<Integer, LongAdder> buckets) {
      boolean newBucketCreated = false;
      int bucketIndex;
      if (Double.isInfinite(value)) {
        bucketIndex = findBucketIndex(Double.MAX_VALUE) + 1;
      } else {
        bucketIndex = findBucketIndex(value);
      }
      LongAdder bucketCount = buckets.get(bucketIndex);
      if (bucketCount == null) {
        LongAdder newBucketCount = new LongAdder();
        LongAdder existingBucketCount = buckets.putIfAbsent(bucketIndex, newBucketCount);
        if (existingBucketCount == null) {
          newBucketCreated = true;
          bucketCount = newBucketCount;
        } else {
          bucketCount = existingBucketCount;
        }
      }
      bucketCount.increment();
      return newBucketCreated;
    }

    private int findBucketIndex(double value) {
      // Preconditions:
      // Double.isNan(value) is false;
      // Double.isInfinite(value) is false;
      // value > 0
      // ---
      // The following is a naive implementation of C's frexp() function.
      // Performance can be improved by using the internal Bit representation of floating point
      // numbers.
      // More info on the Bit representation of floating point numbers:
      // https://stackoverflow.com/questions/8341395/what-is-a-subnormal-floating-point-number
      // Result: value == frac * 2^exp where frac in [0.5, 1).
      double frac = value;
      int exp = 0;
      while (frac < 0.5) {
        frac *= 2.0;
        exp--;
      }
      while (frac >= 1.0) {
        frac /= 2.0;
        exp++;
      }
      // end of frexp()

      if (nativeSchema >= 1) {
        return findIndex(NATIVE_BOUNDS[nativeSchema - 1], frac)
            + (exp - 1) * NATIVE_BOUNDS[nativeSchema - 1].length;
      } else {
        int bucketIndex = exp;
        if (frac == 0.5) {
          bucketIndex--;
        }
        int offset = (1 << -nativeSchema) - 1;
        bucketIndex = (bucketIndex + offset) >> -nativeSchema;
        return bucketIndex;
      }
    }

    private int findIndex(double[] bounds, double frac) {
      // The following is the equivalent of golang's sort.SearchFloat64s(bounds, frac)
      // See https://pkg.go.dev/sort#SearchFloat64s
      int first = 0;
      int last = bounds.length - 1;
      while (first <= last) {
        int mid = (first + last) / 2;
        if (bounds[mid] == frac) {
          return mid;
        } else if (bounds[mid] < frac) {
          first = mid + 1;
        } else {
          last = mid - 1;
        }
      }
      return last + 1;
    }

    /**
     * Makes sure that the number of native buckets does not exceed nativeMaxBuckets.
     *
     * <ul>
     *   <li>If the histogram has already been scaled down (nativeSchema < initialSchema) reset
     *       after resetIntervalExpired to get back to the original schema.
     *   <li>If a new bucket was created and we now exceed nativeMaxBuckets run maybeScaleDown() to
     *       scale down
     * </ul>
     */
    private void maybeResetOrScaleDown(double value, boolean nativeBucketCreated) {
      AtomicBoolean wasReset = new AtomicBoolean(false);
      if (resetDurationExpired && nativeSchema < nativeInitialSchema) {
        // If nativeSchema < initialNativeSchema the histogram has been scaled down.
        // So if resetDurationExpired we will reset it to restore the original native schema.
        buffer.run(
            expectedCount -> count.sum() == expectedCount,
            () -> {
              if (maybeReset()) {
                wasReset.set(true);
              }
              return null;
            },
            v -> doObserve(v, true));
      } else if (nativeBucketCreated) {
        // If a new bucket was created we need to check if nativeMaxBuckets is exceeded
        // and scale down if so.
        maybeScaleDown(wasReset);
      }
      if (wasReset.get()) {
        // We just discarded the newly observed value. Observe it again.
        if (!buffer.append(value)) {
          doObserve(value, true);
        }
      }
    }

    private void maybeScaleDown(AtomicBoolean wasReset) {
      if (nativeMaxBuckets == 0 || nativeSchema == -4) {
        return;
      }
      int numberOfBuckets =
          nativeBucketsForPositiveValues.size() + nativeBucketsForNegativeValues.size();
      if (numberOfBuckets <= nativeMaxBuckets) {
        return;
      }
      buffer.run(
          expectedCount -> count.sum() == expectedCount,
          () -> {
            // Now we are in the synchronized block while new observations go into the buffer.
            // Check again if we need to limit the bucket size, because another thread might
            // have limited it in the meantime.
            int numBuckets =
                nativeBucketsForPositiveValues.size() + nativeBucketsForNegativeValues.size();
            if (numBuckets <= nativeMaxBuckets || nativeSchema == -4) {
              return null;
            }
            if (maybeReset()) {
              wasReset.set(true);
              return null;
            }
            if (maybeWidenZeroBucket()) {
              return null;
            }
            doubleBucketWidth();
            return null;
          },
          v -> doObserve(v, true));
    }

    // maybeReset is called in the synchronized block while new observations go into the buffer.
    private boolean maybeReset() {
      if (!resetDurationExpired) {
        return false;
      }
      resetDurationExpired = false;
      buffer.reset();
      nativeBucketsForPositiveValues.clear();
      nativeBucketsForNegativeValues.clear();
      nativeZeroCount.reset();
      count.reset();
      sum.reset();
      for (LongAdder classicBucket : classicBuckets) {
        classicBucket.reset();
      }
      nativeZeroThreshold = nativeMinZeroThreshold;
      nativeSchema = Histogram.this.nativeInitialSchema;
      createdTimeMillis = System.currentTimeMillis();
      if (exemplarSampler != null) {
        exemplarSampler.reset();
      }
      maybeScheduleNextReset();
      return true;
    }

    // maybeWidenZeroBucket is called in the synchronized block while new observations go into the
    // buffer.
    private boolean maybeWidenZeroBucket() {
      if (nativeZeroThreshold >= nativeMaxZeroThreshold) {
        return false;
      }
      int smallestIndex = findSmallestIndex(nativeBucketsForPositiveValues);
      int smallestNegativeIndex = findSmallestIndex(nativeBucketsForNegativeValues);
      if (smallestNegativeIndex < smallestIndex) {
        smallestIndex = smallestNegativeIndex;
      }
      if (smallestIndex == Integer.MAX_VALUE) {
        return false;
      }
      double newZeroThreshold = nativeBucketIndexToUpperBound(nativeSchema, smallestIndex);
      if (newZeroThreshold > nativeMaxZeroThreshold) {
        return false;
      }
      mergeWithZeroBucket(smallestIndex, nativeBucketsForPositiveValues);
      mergeWithZeroBucket(smallestIndex, nativeBucketsForNegativeValues);
      nativeZeroThreshold = newZeroThreshold;
      return true;
    }

    private void mergeWithZeroBucket(int index, Map<Integer, LongAdder> buckets) {
      LongAdder count = buckets.remove(index);
      if (count != null) {
        nativeZeroCount.add(count.sum());
      }
    }

    private double nativeBucketIndexToUpperBound(int schema, int index) {
      double result = calcUpperBound(schema, index);
      if (Double.isInfinite(result)) {
        // The last bucket boundary should always be MAX_VALUE, so that the +Inf bucket counts only
        // actual +Inf observations.
        // However, MAX_VALUE is not a natural bucket boundary, so we introduce MAX_VALUE
        // as an artificial boundary before +Inf.
        double previousBucketBoundary = calcUpperBound(schema, index - 1);
        if (Double.isFinite(previousBucketBoundary) && previousBucketBoundary < Double.MAX_VALUE) {
          return Double.MAX_VALUE;
        }
      }
      return result;
    }

    private double calcUpperBound(int schema, int index) {
      // The actual formula is:
      // ---
      // base := 2^(2^-schema);
      // upperBound := base^index;
      // ---
      // The following implementation reduces the numerical error for index > 0.
      // It's not very efficient. We should refactor and use an algorithm as in client_golang's
      // getLe()
      double factor = 1.0;
      while (index > 0) {
        if (index % 2 == 0) {
          index /= 2;
          schema -= 1;
        } else {
          index -= 1;
          factor *= Math.pow(2, Math.pow(2, -schema));
        }
      }
      return factor * Math.pow(2, index * Math.pow(2, -schema));
    }

    private int findSmallestIndex(Map<Integer, LongAdder> nativeBuckets) {
      int result = Integer.MAX_VALUE;
      for (int key : nativeBuckets.keySet()) {
        if (key < result) {
          result = key;
        }
      }
      return result;
    }

    // doubleBucketWidth is called in the synchronized block while new observations go into the
    // buffer.
    @SuppressWarnings("NonAtomicVolatileUpdate")
    private void doubleBucketWidth() {
      doubleBucketWidth(nativeBucketsForPositiveValues);
      doubleBucketWidth(nativeBucketsForNegativeValues);
      nativeSchema--;
    }

    private void doubleBucketWidth(Map<Integer, LongAdder> buckets) {
      int[] keys = new int[buckets.size()];
      long[] values = new long[keys.length];
      int i = 0;
      for (Map.Entry<Integer, LongAdder> entry : buckets.entrySet()) {
        keys[i] = entry.getKey();
        values[i] = entry.getValue().sum();
        i++;
      }
      buckets.clear();
      for (i = 0; i < keys.length; i++) {
        int index = (keys[i] > 0 ? keys[i] + 1 : keys[i]) / 2;
        LongAdder count = buckets.computeIfAbsent(index, k -> new LongAdder());
        count.add(values[i]);
      }
    }

    private NativeHistogramBuckets toBucketList(ConcurrentHashMap<Integer, LongAdder> map) {
      int[] bucketIndexes = new int[map.size()];
      long[] counts = new long[map.size()];
      int i = 0;
      for (Map.Entry<Integer, LongAdder> entry : map.entrySet()) {
        bucketIndexes[i] = entry.getKey();
        counts[i] = entry.getValue().sum();
        i++;
      }
      return NativeHistogramBuckets.of(bucketIndexes, counts);
    }

    @SuppressWarnings("FutureReturnValueIgnored")
    private void maybeScheduleNextReset() {
      if (nativeResetDurationSeconds > 0) {
        Scheduler.schedule(
            () -> resetDurationExpired = true, nativeResetDurationSeconds, TimeUnit.SECONDS);
      }
    }
  }

  @Override
  public HistogramSnapshot collect() {
    return (HistogramSnapshot) super.collect();
  }

  @Override
  protected HistogramSnapshot collect(List<Labels> labels, List<DataPoint> metricData) {
    List<HistogramSnapshot.HistogramDataPointSnapshot> data = new ArrayList<>(labels.size());
    for (int i = 0; i < labels.size(); i++) {
      data.add(metricData.get(i).collect(labels.get(i)));
    }
    return new HistogramSnapshot(metadata, data);
  }

  /**
   * @deprecated Use {@link #getMetricFamilyDescriptor()} instead.
   */
  @Override
  @Deprecated
  @SuppressWarnings("InlineMeSuggester")
  public MetricType getMetricType() {
    return MetricType.HISTOGRAM;
  }

  @Override
  protected DataPoint newDataPoint() {
    return new DataPoint();
  }

  static {
    // See bounds in client_golang's histogram implementation.
    NATIVE_BOUNDS = new double[8][];
    for (int schema = 1; schema <= 8; schema++) {
      NATIVE_BOUNDS[schema - 1] = new double[1 << schema];
      NATIVE_BOUNDS[schema - 1][0] = 0.5;
      // https://github.com/open-telemetry/opentelemetry-proto/blob/main/opentelemetry/proto/metrics/v1/metrics.proto#L501
      double base = Math.pow(2, Math.pow(2, -schema));
      for (int i = 1; i < NATIVE_BOUNDS[schema - 1].length; i++) {
        if (i % 2 == 0 && schema > 1) {
          // Use previously calculated value for increased precision, see comment in client_golang's
          // implementation.
          NATIVE_BOUNDS[schema - 1][i] = NATIVE_BOUNDS[schema - 2][i / 2];
        } else {
          NATIVE_BOUNDS[schema - 1][i] = NATIVE_BOUNDS[schema - 1][i - 1] * base;
        }
      }
    }
  }

  public static Builder builder() {
    return new Builder(PrometheusProperties.get());
  }

  public static Builder builder(PrometheusProperties config) {
    return new Builder(config);
  }

  public static class Builder extends StatefulMetric.Builder<Histogram.Builder, Histogram> {

    @SuppressWarnings("MutablePublicArray")
    public static final double[] DEFAULT_CLASSIC_UPPER_BOUNDS =
        new double[] {.005, .01, .025, .05, .1, .25, .5, 1, 2.5, 5, 10};

    private static final double DEFAULT_NATIVE_MIN_ZERO_THRESHOLD = Math.pow(2.0, -128);
    private static final double DEFAULT_NATIVE_MAX_ZERO_THRESHOLD = Math.pow(2.0, -128);
    private static final int DEFAULT_NATIVE_INITIAL_SCHEMA = 5;
    private static final int DEFAULT_NATIVE_MAX_NUMBER_OF_BUCKETS = 160;
    private static final long DEFAULT_NATIVE_RESET_DURATION_SECONDS = 0; // 0 means no reset

    @Nullable private Boolean nativeOnly;
    @Nullable private Boolean classicOnly;
    @Nullable private double[] classicUpperBounds;
    @Nullable private Integer nativeInitialSchema;
    @Nullable private Double nativeMaxZeroThreshold;
    @Nullable private Double nativeMinZeroThreshold;
    @Nullable private Integer nativeMaxNumberOfBuckets;
    @Nullable private Long nativeResetDurationSeconds;

    @Override
    public Histogram build() {
      return new Histogram(this, properties);
    }

    @Override
    protected MetricsProperties toProperties() {
      MetricsProperties.Builder builder = MetricsProperties.builder();
      if (classicUpperBounds != null) {
        builder.histogramClassicUpperBounds(classicUpperBounds);
      }
      return builder
          .exemplarsEnabled(exemplarsEnabled)
          .histogramNativeOnly(nativeOnly)
          .histogramClassicOnly(classicOnly)
          .histogramNativeInitialSchema(nativeInitialSchema)
          .histogramNativeMinZeroThreshold(nativeMinZeroThreshold)
          .histogramNativeMaxZeroThreshold(nativeMaxZeroThreshold)
          .histogramNativeMaxNumberOfBuckets(nativeMaxNumberOfBuckets)
          .histogramNativeResetDurationSeconds(nativeResetDurationSeconds)
          .build();
    }

    /** Default properties for histogram metrics. */
    @Override
    public MetricsProperties getDefaultProperties() {
      return MetricsProperties.builder()
          .exemplarsEnabled(true)
          .histogramNativeOnly(false)
          .histogramClassicOnly(false)
          .histogramClassicUpperBounds(DEFAULT_CLASSIC_UPPER_BOUNDS)
          .histogramNativeInitialSchema(DEFAULT_NATIVE_INITIAL_SCHEMA)
          .histogramNativeMinZeroThreshold(DEFAULT_NATIVE_MIN_ZERO_THRESHOLD)
          .histogramNativeMaxZeroThreshold(DEFAULT_NATIVE_MAX_ZERO_THRESHOLD)
          .histogramNativeMaxNumberOfBuckets(DEFAULT_NATIVE_MAX_NUMBER_OF_BUCKETS)
          .histogramNativeResetDurationSeconds(DEFAULT_NATIVE_RESET_DURATION_SECONDS)
          .build();
    }

    private Builder(PrometheusProperties config) {
      super(Collections.singletonList("le"), config);
    }

    /**
     * Use the native histogram representation only, i.e. don't maintain classic histogram buckets.
     * See {@link Histogram} for more info.
     */
    public Builder nativeOnly() {
      if (Boolean.TRUE.equals(classicOnly)) {
        throw new IllegalArgumentException("Cannot call nativeOnly() after calling classicOnly().");
      }
      nativeOnly = true;
      return this;
    }

    /**
     * Use the classic histogram representation only, i.e. don't maintain native histogram buckets.
     * See {@link Histogram} for more info.
     */
    public Builder classicOnly() {
      if (Boolean.TRUE.equals(nativeOnly)) {
        throw new IllegalArgumentException("Cannot call classicOnly() after calling nativeOnly().");
      }
      classicOnly = true;
      return this;
    }

    /**
     * Set the upper bounds for the classic histogram buckets. Default is {@link
     * Builder#DEFAULT_CLASSIC_UPPER_BOUNDS}. If the +Inf bucket is missing it will be added. If
     * upperBounds contains duplicates the duplicates will be removed.
     */
    public Builder classicUpperBounds(double... upperBounds) {
      this.classicUpperBounds = upperBounds;
      for (double bound : upperBounds) {
        if (Double.isNaN(bound)) {
          throw new IllegalArgumentException("Cannot use NaN as upper bound for a histogram");
        }
      }
      return this;
    }

    /**
     * Create classic histogram buckets with linear bucket boundaries.
     *
     * <p>Example: {@code classicLinearUpperBounds(1.0, 0.5, 10)} creates bucket boundaries {@code
     * [[1.0, 1.5, 2.0, 2.5, 3.0, 3.5, 4.0, 4.5, 5.0, 5.5]}.
     *
     * @param start is the first bucket boundary
     * @param width is the width of each bucket
     * @param count is the total number of buckets, including start
     */
    public Builder classicLinearUpperBounds(double start, double width, int count) {
      this.classicUpperBounds = new double[count];
      // Use BigDecimal to avoid weird bucket boundaries like 0.7000000000000001.
      BigDecimal s = new BigDecimal(Double.toString(start));
      BigDecimal w = new BigDecimal(Double.toString(width));
      for (int i = 0; i < count; i++) {
        classicUpperBounds[i] = s.add(w.multiply(new BigDecimal(i))).doubleValue();
      }
      return this;
    }

    /**
     * Create classic histogram buckets with exponential boundaries.
     *
     * <p>Example: {@code classicExponentialUpperBounds(1.0, 2.0, 10)} creates bucket boundaries
     * {@code [1.0, 2.0, 4.0, 8.0, 16.0, 32.0, 64.0, 128.0, 256.0, 512.0]}
     *
     * @param start is the first bucket boundary
     * @param factor growth factor
     * @param count total number of buckets, including start
     */
    public Builder classicExponentialUpperBounds(double start, double factor, int count) {
      classicUpperBounds = new double[count];
      for (int i = 0; i < count; i++) {
        classicUpperBounds[i] = start * Math.pow(factor, i);
      }
      return this;
    }

    /**
     * The schema is a number in [-4, 8] defining the resolution of the native histogram. Default is
     * {@link Builder#DEFAULT_NATIVE_INITIAL_SCHEMA}.
     *
     * <p>The higher the schema, the finer the resolution. Schema is Prometheus terminology. In
     * OpenTelemetry it's called "scale".
     *
     * <p>Note that the schema for a histogram may be automatically decreased at runtime if the
     * number of native histogram buckets exceeds {@link #nativeMaxNumberOfBuckets(int)}.
     *
     * <p>The following table shows:
     *
     * <ul>
     *   <li>factor: The growth factor for bucket boundaries, i.e. next bucket boundary = growth
     *       factor * previous bucket boundary.
     *   <li>max quantile error: The maximum error for quantiles calculated using the Prometheus
     *       histogram_quantile() function, relative to the observed value, assuming harmonic mean.
     * </ul>
     *
     * <table border="1">
     *     <caption>max quantile errors for different growth factors</caption>
     *     <tr>
     *         <td>schema</td><td>factor</td><td>max quantile error</td>
     *     </tr>
     *     <tr>
     *         <td>-4</td><td>65.536</td><td>99%</td>
     *     </tr>
     *     <tr>
     *         <td>-3</td><td>256</td><td>99%</td>
     *     </tr>
     *     <tr>
     *         <td>-2</td><td>16</td><td>88%</td>
     *     </tr>
     *     <tr>
     *         <td>-1</td><td>4</td><td>60%</td>
     *     </tr>
     *     <tr>
     *         <td>0</td><td>2</td><td>33%</td>
     *     </tr>
     *     <tr>
     *         <td>1</td><td>1.4142...</td><td>17%</td>
     *     </tr>
     *     <tr>
     *         <td>2</td><td>1.1892...</td><td>9%</td>
     *     </tr>
     *     <tr>
     *         <td>3</td><td>1.1090...</td><td>4%</td>
     *     </tr>
     *     <tr>
     *         <td>4</td><td>1.0442...</td><td>2%</td>
     *     </tr>
     *     <tr>
     *         <td>5</td><td>1.0218...</td><td>1%</td>
     *     </tr>
     *     <tr>
     *         <td>6</td><td>1.0108...</td><td>0.5%</td>
     *     </tr>
     *     <tr>
     *         <td>7</td><td>1.0054...</td><td>0.3%</td>
     *     </tr>
     *     <tr>
     *         <td>8</td><td>1.0027...</td><td>0.1%</td>
     *     </tr>
     * </table>
     */
    public Builder nativeInitialSchema(int nativeSchema) {
      if (nativeSchema < -4 || nativeSchema > 8) {
        throw new IllegalArgumentException(
            "Unsupported native histogram schema "
                + nativeSchema
                + ": expecting -4 <= schema <= 8.");
      }
      this.nativeInitialSchema = nativeSchema;
      return this;
    }

    /**
     * Native histogram buckets get smaller and smaller the closer they get to zero. To avoid
     * wasting a lot of buckets for observations fluctuating around zero, we consider all values in
     * [-zeroThreshold, +zeroThreshold] to be equal to zero.
     *
     * <p>The zeroThreshold is initialized with minZeroThreshold, and will grow up to
     * maxZeroThreshold if the number of native histogram buckets exceeds nativeMaxBuckets.
     *
     * <p>Default is {@link Builder#DEFAULT_NATIVE_MAX_NUMBER_OF_BUCKETS}.
     */
    public Builder nativeMaxZeroThreshold(double nativeMaxZeroThreshold) {
      if (nativeMaxZeroThreshold < 0) {
        throw new IllegalArgumentException(
            "Illegal native max zero threshold " + nativeMaxZeroThreshold + ": must be >= 0");
      }
      this.nativeMaxZeroThreshold = nativeMaxZeroThreshold;
      return this;
    }

    /**
     * Native histogram buckets get smaller and smaller the closer they get to zero. To avoid
     * wasting a lot of buckets for observations fluctuating around zero, we consider all values in
     * [-zeroThreshold, +zeroThreshold] to be equal to zero.
     *
     * <p>The zeroThreshold is initialized with minZeroThreshold, and will grow up to
     * maxZeroThreshold if the number of native histogram buckets exceeds nativeMaxBuckets.
     *
     * <p>Default is {@link Builder#DEFAULT_NATIVE_MIN_ZERO_THRESHOLD}.
     */
    public Builder nativeMinZeroThreshold(double nativeMinZeroThreshold) {
      if (nativeMinZeroThreshold < 0) {
        throw new IllegalArgumentException(
            "Illegal native min zero threshold " + nativeMinZeroThreshold + ": must be >= 0");
      }
      this.nativeMinZeroThreshold = nativeMinZeroThreshold;
      return this;
    }

    /**
     * Limit the number of native buckets.
     *
     * <p>If the number of native buckets exceeds the maximum, the {@link #nativeInitialSchema(int)}
     * is decreased, i.e. the resolution of the histogram is decreased to reduce the number of
     * buckets.
     *
     * <p>Default is {@link Builder#DEFAULT_NATIVE_MAX_NUMBER_OF_BUCKETS}.
     */
    public Builder nativeMaxNumberOfBuckets(int nativeMaxBuckets) {
      this.nativeMaxNumberOfBuckets = nativeMaxBuckets;
      return this;
    }

    /**
     * If the histogram needed to be scaled down because {@link #nativeMaxNumberOfBuckets(int)} was
     * exceeded, reset the histogram after a certain time interval to go back to the original {@link
     * #nativeInitialSchema(int)}.
     *
     * <p>Reset means all values are set to zero. A good value might be 24h or 7d.
     *
     * <p>Default is no reset.
     */
    public Builder nativeResetDuration(long duration, TimeUnit unit) {
      if (duration <= 0) {
        throw new IllegalArgumentException(duration + ": value > 0 expected");
      }
      long seconds = unit.toSeconds(duration);
      if (seconds == 0) {
        throw new IllegalArgumentException(
            duration
                + " "
                + unit
                + ": duration must be at least 1 second. Sub-second durations are not supported.");
      }
      nativeResetDurationSeconds = seconds;
      return this;
    }

    @Override
    protected Builder self() {
      return this;
    }
  }
}