001package io.prometheus.metrics.core.metrics;
002
003import io.prometheus.metrics.annotations.StableApi;
004import java.lang.reflect.Array;
005import java.util.concurrent.TimeUnit;
006import java.util.function.LongSupplier;
007import java.util.function.ObjDoubleConsumer;
008import java.util.function.Supplier;
009
010/**
011 * Maintains a ring buffer of T to implement a sliding time window.
012 *
013 * <p>This is used to maintain a sliding window of {@link CKMSQuantiles} for {@link Summary}
014 * metrics.
015 *
016 * <p>It is implemented in a generic way so that 3rd party libraries can use it for implementing
017 * sliding windows.
018 *
019 * <p><b>Thread Safety:</b> This class uses coarse-grained {@code synchronized} methods for
020 * simplicity and correctness. All public methods ({@link #current()} and {@link #observe(double)})
021 * are synchronized, which ensures thread-safe access to the ring buffer and rotation logic.
022 *
023 * <p><b>Performance Note:</b> The synchronized approach may cause contention under high-frequency
024 * observations.
025 *
026 * <p>However, given that Summary metrics are less commonly used (Histogram is generally preferred),
027 * and the observation frequency is typically lower than Counter increments, the current
028 * implementation provides an acceptable trade-off between simplicity and performance.
029 */
030@StableApi
031public class SlidingWindow<T> {
032
033  private final Supplier<T> constructor;
034  private final ObjDoubleConsumer<T> observeFunction;
035  private final T[] ringBuffer;
036  private int currentBucket;
037  private long lastRotateTimestampMillis;
038  private final long durationBetweenRotatesMillis;
039  private final LongSupplier currentTimeMillis;
040
041  /**
042   * Example: If the {@code maxAgeSeconds} is 60 and {@code ageBuckets} is 3, then 3 instances of
043   * {@code T} are maintained and the sliding window moves to the next instance of T every 20
044   * seconds.
045   *
046   * @param clazz type of T
047   * @param constructor for creating a new instance of T as the old one gets evicted
048   * @param observeFunction for observing a value (e.g. calling {@code t.observe(value)}
049   * @param maxAgeSeconds after this amount of time an instance of T gets evicted.
050   * @param ageBuckets number of age buckets.
051   */
052  public SlidingWindow(
053      Class<T> clazz,
054      Supplier<T> constructor,
055      ObjDoubleConsumer<T> observeFunction,
056      long maxAgeSeconds,
057      int ageBuckets) {
058    this(clazz, constructor, observeFunction, maxAgeSeconds, ageBuckets, System::currentTimeMillis);
059  }
060
061  // VisibleForTesting
062  @SuppressWarnings("unchecked")
063  SlidingWindow(
064      Class<T> clazz,
065      Supplier<T> constructor,
066      ObjDoubleConsumer<T> observeFunction,
067      long maxAgeSeconds,
068      int ageBuckets,
069      LongSupplier currentTimeMillis) {
070    this.constructor = constructor;
071    this.observeFunction = observeFunction;
072    this.ringBuffer = (T[]) Array.newInstance(clazz, ageBuckets);
073    for (int i = 0; i < ringBuffer.length; i++) {
074      this.ringBuffer[i] = constructor.get();
075    }
076    this.currentBucket = 0;
077    this.lastRotateTimestampMillis = currentTimeMillis.getAsLong();
078    this.durationBetweenRotatesMillis = TimeUnit.SECONDS.toMillis(maxAgeSeconds) / ageBuckets;
079    this.currentTimeMillis = currentTimeMillis;
080  }
081
082  /** Get the currently active instance of {@code T}. */
083  public synchronized T current() {
084    return rotate();
085  }
086
087  /** Observe a value. */
088  public synchronized void observe(double value) {
089    rotate();
090    for (T t : ringBuffer) {
091      observeFunction.accept(t, value);
092    }
093  }
094
095  private T rotate() {
096    long timeSinceLastRotateMillis = currentTimeMillis.getAsLong() - lastRotateTimestampMillis;
097    while (timeSinceLastRotateMillis > durationBetweenRotatesMillis) {
098      ringBuffer[currentBucket] = constructor.get();
099      if (++currentBucket >= ringBuffer.length) {
100        currentBucket = 0;
101      }
102      timeSinceLastRotateMillis -= durationBetweenRotatesMillis;
103      lastRotateTimestampMillis += durationBetweenRotatesMillis;
104    }
105    return ringBuffer[currentBucket];
106  }
107}