add slow request sampler

core/src/main/java/com/linecorp/armeria/common/util/AndSampler.java

@@ -0,0 +1,37 @@
+/*
+ * Copyright 2023 LINE Corporation
+ *
+ * LINE Corporation 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:
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ */
+package com.linecorp.armeria.common.util;
+
+/**
+ * Sample if both of the samplers sample.
+ */
+final class AndSampler<T> implements Sampler<T> {
+
+    private final Sampler<T> left;
+    private final Sampler<T> right;
+
+    AndSampler(Sampler<T> left, Sampler<T> right) {
+        this.left = left;
+        this.right = right;
+    }
+
+    @Override
+    public boolean isSampled(T t) {
+        // Assign the variables otherwise the short-circuiting will cause sampler to not be used.
+        final boolean leftSampled = left.isSampled(t);
+        final boolean rightSampled = right.isSampled(t);
+        return leftSampled && rightSampled;
+    }
+
+    @Override
+    public String toString() {
+        return left + " and " + right;
+    }
+}

core/src/main/java/com/linecorp/armeria/common/util/OrSampler.java

@@ -0,0 +1,37 @@
+/*
+ * Copyright 2023 LINE Corporation
+ *
+ * LINE Corporation 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:
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ */
+package com.linecorp.armeria.common.util;
+
+/**
+ * Sample if one of the samplers samples.
+ */
+final class OrSampler<T> implements Sampler<T> {
+
+    private final Sampler<T> left;
+    private final Sampler<T> right;
+
+    OrSampler(Sampler<T> left, Sampler<T> right) {
+        this.left = left;
+        this.right = right;
+    }
+
+    @Override
+    public boolean isSampled(T t) {
+        // Assign the variables otherwise the short-circuiting will cause sampler to not be used.
+        final boolean leftSampled = left.isSampled(t);
+        final boolean rightSampled = right.isSampled(t);
+        return leftSampled || rightSampled;
+    }
+
+    @Override
+    public String toString() {
+        return left + " or " + right;
+    }
+}

core/src/main/java/com/linecorp/armeria/common/util/Sampler.java

@@ -29,6 +29,8 @@
  */
 package com.linecorp.armeria.common.util;
 
+import static java.util.Objects.requireNonNull;
+
 /**
  * Sampler is responsible for deciding if a particular trace should be "sampled", i.e. whether the
  * overhead of tracing will occur and/or if a trace will be reported to the collection tier.
@@ -42,6 +44,76 @@
  */
 @FunctionalInterface
 public interface Sampler<T> {
+
+    /**
+     * Returns a sampler that applies logical or operator to both samplers decisions.
+     */
+    default Sampler<T> or(Sampler<T> other) {
+        return new OrSampler<>(this, requireNonNull(other, "other"));
+    }
+
+    /**
+     * Returns a sampler that applies logical and operator to both samplers decisions.
+     */
+    default Sampler<T> and(Sampler<T> other) {
+        return new AndSampler<>(this, requireNonNull(other, "other"));
+    }
+
+    /**
+     * Returns a sampler that applies logical not operator to the sampler decision.
+     */
+    default Sampler<T> not() {
+        return object -> !isSampled(object);
+    }
+
+    /**
+     * Returns a sampler that returns {@code true} if the value is greater than the given value.
+     */
+    static <T extends Comparable<T>> Sampler<T> greaterThan(T val) {
+        requireNonNull(val, "val");
+        return object -> object.compareTo(val) > 0;
+    }
+
+    /**
+     * Returns a sampler that returns {@code true} if the value is less than or equal to the given value.
+     */
+    static <T extends Comparable<T>> Sampler<T> greaterThanOrEqual(T val) {
+        requireNonNull(val, "val");
+        return object -> object.compareTo(val) >= 0;
+    }
+
+    /**
+     * Returns a sampler that returns {@code true} if the value is less than the given value.
+     */
+    static <T extends Comparable<T>> Sampler<T> lessThan(T val) {
+        requireNonNull(val, "val");
+        return object -> object.compareTo(val) < 0;
+    }
+
+    /**
+     * Returns a sampler that returns {@code true} if the value is less than or equal to the given value.
+     */
+    static <T extends Comparable<T>> Sampler<T> lessThanOrEqual(T val) {
+        requireNonNull(val, "val");
+        return object -> object.compareTo(val) <= 0;
+    }
+
+    /**
+     * Returns a sampler that returns {@code true} if the value is equal to the given value.
+     */
+    static <T extends Comparable<T>> Sampler<T> equal(T val) {
+        requireNonNull(val, "val");
+        return object -> object.compareTo(val) == 0;
+    }
+
+    /**
+     * Returns a sampler that returns {@code true} if the value is inside the percentile distribution
+     * in the given time window.
+     */
+    static Sampler<Long> percentile(float percentile, long windowMilliseconds) {
+        return new TimeWindowPercentileSampler(percentile, windowMilliseconds);
+    }
+
     /**
      * Returns a probabilistic sampler which samples at the specified {@code probability}
      * between {@code 0.0} and {@code 1.0}.

core/src/main/java/com/linecorp/armeria/common/util/TimeWindowPercentileSampler.java

@@ -0,0 +1,107 @@
+/*
+ * Copyright 2023 LINE Corporation
+ *
+ * LINE Corporation 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:
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ */
+package com.linecorp.armeria.common.util;
+
+import java.time.Duration;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.MoreObjects;
+
+import com.linecorp.armeria.common.metric.MoreMeters;
+
+import io.micrometer.core.instrument.Clock;
+import io.micrometer.core.instrument.distribution.DistributionStatisticConfig;
+import io.micrometer.core.instrument.distribution.HistogramSnapshot;
+import io.micrometer.core.instrument.distribution.TimeWindowPercentileHistogram;
+
+/**
+ * Sample if the value is less than the percentile of the values in the last window.
+ */
+final class TimeWindowPercentileSampler implements Sampler<Long> {
+
+    private final float percentile;
+    private final long windowLengthMillis;
+    private final TimeWindowPercentileHistogram histogram;
+    private final long snapshotUpdateNanos;
+    private static final long DEFAULT_SNAPSHOT_UPDATE_NANOS = TimeUnit.SECONDS.toNanos(1);
+    private long lastSnapshotNanos;
+    private HistogramSnapshot histogramSnapshot;
+
+    private final Clock clock;
+    private final AtomicReference<Boolean> isTakingSnapshot = new AtomicReference<>(false);
+
+    TimeWindowPercentileSampler(float percentile, long windowLengthMillis) {
+        this(percentile, windowLengthMillis, Clock.SYSTEM, DEFAULT_SNAPSHOT_UPDATE_NANOS);
+    }
+
+    @VisibleForTesting
+    TimeWindowPercentileSampler(float percentile, long windowLengthMillis, Clock clock,
+                                long snapshotUpdateNanos) {
+        this.percentile = percentile;
+        this.windowLengthMillis = windowLengthMillis;
+
+        final DistributionStatisticConfig distributionStatisticConfig =
+                DistributionStatisticConfig.builder()
+                                           .percentilesHistogram(false)
+                                           .percentiles(percentile)
+                                           .expiry(Duration.ofMillis(windowLengthMillis))
+                                           .build()
+                                           .merge(MoreMeters.distributionStatisticConfig());
+        this.histogram = new TimeWindowPercentileHistogram(clock, distributionStatisticConfig, true);
+        this.snapshotUpdateNanos = snapshotUpdateNanos;
+        this.histogramSnapshot = histogram.takeSnapshot(0, 0, 0);
+        this.clock = clock;
+        this.lastSnapshotNanos = clock.monotonicTime();
+    }
+
+    @VisibleForTesting
+    static TimeWindowPercentileSampler create(float percentile, long windowLengthMillis,
+                                              long snapshotUpdateNanos) {
+        return new TimeWindowPercentileSampler(percentile, windowLengthMillis, Clock.SYSTEM,
+                                               snapshotUpdateNanos);
+    }
+
+    @Override
+    public boolean isSampled(Long t) {
+        histogram.recordLong(t);
+
+        System.out.println("lastSnapshotNanos: " + lastSnapshotNanos);
+        System.out.println("snapshotUpdateNanos: " + snapshotUpdateNanos);
+        System.out.println("clock.monotonicTime(): " + clock.monotonicTime());
+
+        if (lastSnapshotNanos + snapshotUpdateNanos <= clock.monotonicTime()) {
+            if (isTakingSnapshot.compareAndSet(false, true)) {
+                // Two threads reach here back to back. Make sure snapshot is not taken very recently before
+                // we acquired the lock.
+                if (lastSnapshotNanos + snapshotUpdateNanos <= clock.monotonicTime()) {
+                    System.out.println("Taking snapshot");
+                    histogramSnapshot = histogram.takeSnapshot(0, 0, 0);
+                    lastSnapshotNanos = clock.monotonicTime();
+                    isTakingSnapshot.set(false);
+                }
+            }
+        }
+
+        final Double percentileValue = histogramSnapshot.percentileValues()[0].value();
+        return t >= percentileValue.longValue();
+    }
+
+    @Override
+    public String toString() {
+        return MoreObjects.toStringHelper(this)
+                          .omitNullValues()
+                          .add("percentile", percentile)
+                          .add("windowLengthMillis", windowLengthMillis)
+                          .add("snapshotUpdateNanos", snapshotUpdateNanos)
+                          .toString();
+    }
+}

core/src/main/java/com/linecorp/armeria/server/annotation/decorator/LoggingDecorator.java

@@ -83,6 +83,28 @@
      */
     float failureSamplingRate() default -1.0f;
 
+    /**
+     * Sample the requests if they are slower than the {@code slowRequestSamplingPercentile} percent
+     * of the requests.
+     */
+    float slowRequestSamplingPercentile() default -1.0f;
+
+    /**
+     * Slow request percentiles are calculated over the last {@code slowRequestSamplingWindowMillis}.
+     */
+    long slowRequestSamplingWindowMillis() default 10 * 60 * 1000;
+
+    /**
+     * Don't sample the requests if they are faster than the {@code slowRequestSamplingLowerBoundMillis()}.
+     * Should be used with {@link #slowRequestSamplingUpperBoundMillis()}.
+     */
+    long slowRequestSamplingLowerBoundMillis() default 0L;
+
+    /**
+     * Always sample the requests if they are slower than the {@code slowRequestSamplingUpperBoundMillis()}.
+     */
+    long slowRequestSamplingUpperBoundMillis() default Long.MAX_VALUE;
+
     /**
      * The order of decoration, where a {@link Decorator} of lower value will be applied first.
      */

core/src/main/java/com/linecorp/armeria/server/annotation/decorator/LoggingDecoratorFactoryFunction.java

@@ -21,6 +21,7 @@
 import com.linecorp.armeria.server.HttpService;
 import com.linecorp.armeria.server.annotation.DecoratorFactoryFunction;
 import com.linecorp.armeria.server.logging.LoggingService;
+import com.linecorp.armeria.server.logging.LoggingServiceBuilder;
 
 /**
  * A factory which creates a {@link LoggingService} decorator.
@@ -38,15 +39,26 @@ public final class LoggingDecoratorFactoryFunction implements DecoratorFactoryFu
         final float failureSamplingRate =
                 parameter.failureSamplingRate() >= 0.0f ? parameter.failureSamplingRate()
                                                         : parameter.samplingRate();
-        return LoggingService.builder()
-                             .logWriter(LogWriter.builder()
-                                                 .requestLogLevel(parameter.requestLogLevel())
-                                                 .successfulResponseLogLevel(
-                                                         parameter.successfulResponseLogLevel())
-                                                 .failureResponseLogLevel(parameter.failureResponseLogLevel())
-                                                 .build())
-                             .successSamplingRate(successSamplingRate)
-                             .failureSamplingRate(failureSamplingRate)
-                             .newDecorator();
+        final LogWriter logWriter = LogWriter.builder()
+                                             .requestLogLevel(parameter.requestLogLevel())
+                                             .successfulResponseLogLevel(parameter.successfulResponseLogLevel())
+                                             .failureResponseLogLevel(parameter.failureResponseLogLevel())
+                                             .build();
+        final LoggingServiceBuilder builder
+                = LoggingService.builder()
+                                .logWriter(logWriter)
+                                .successSamplingRate(successSamplingRate)
+                                .failureSamplingRate(failureSamplingRate);
+
+        if (
+                parameter.slowRequestSamplingPercentile() >= 0.0f ||
+                parameter.slowRequestSamplingUpperBoundMillis() >= 0
+        ) {
+            builder.slowRequestSamplingPercentile(parameter.slowRequestSamplingPercentile(),
+                                                  parameter.slowRequestSamplingWindowMillis(),
+                                                  parameter.slowRequestSamplingLowerBoundMillis(),
+                                                  parameter.slowRequestSamplingUpperBoundMillis());
+        }
+        return builder.newDecorator();
     }
 }

core/src/main/java/com/linecorp/armeria/server/logging/LoggingService.java

@@ -61,17 +61,23 @@ public static LoggingServiceBuilder builder() {
 
     LoggingService(HttpService delegate, LogWriter logWriter,
                    Sampler<? super ServiceRequestContext> successSampler,
-                   Sampler<? super ServiceRequestContext> failureSampler) {
+                   Sampler<? super ServiceRequestContext> failureSampler,
+                   Sampler<Long> slowRequestSampler) {
         super(requireNonNull(delegate, "delegate"));
         this.logWriter = requireNonNull(logWriter, "logWriter");
         requireNonNull(successSampler, "successSampler");
         requireNonNull(failureSampler, "failureSampler");
         sampler = requestLog -> {
             final ServiceRequestContext ctx = (ServiceRequestContext) requestLog.context();
+            final boolean isSlow = slowRequestSampler.isSampled(requestLog.totalDurationNanos());
+            final boolean successOrFailure;
             if (ctx.config().successFunction().isSuccess(ctx, requestLog)) {
-                return successSampler.isSampled(ctx);
+                successOrFailure = successSampler.isSampled(ctx);
+            } else {
+                successOrFailure = failureSampler.isSampled(ctx);
             }
-            return failureSampler.isSampled(ctx);
+
+            return successOrFailure || isSlow;
         };
     }