Add counters for partitions

concurrency-limits-core/build.gradle

@@ -6,7 +6,9 @@ sourceCompatibility = JavaVersion.VERSION_1_8
 
 dependencies {
     compile "org.slf4j:slf4j-api:1.7.+"
-
+
+    testCompile project(":concurrency-limits-spectator")
+
     testCompile 'junit:junit-dep:4.10'
     testCompile "org.slf4j:slf4j-log4j12:1.7.+"
 

concurrency-limits-core/src/main/java/com/netflix/concurrency/limits/MetricIds.java

@@ -21,6 +21,7 @@
 public final class MetricIds {
     public static final String LIMIT_NAME = "limit";
     public static final String CALL_NAME = "call";
+    public static final String PARTITIONED_CALL_NAME = "call.partition";
     public static final String INFLIGHT_NAME = "inflight";
     public static final String PARTITION_LIMIT_NAME = "limit.partition";
     public static final String MIN_RTT_NAME = "min_rtt";

concurrency-limits-core/src/main/java/com/netflix/concurrency/limits/limiter/AbstractPartitionedLimiter.java

@@ -111,6 +111,10 @@ static class Partition {
         private int busy = 0;
         private long backoffMillis = 0;
         private MetricRegistry.SampleListener inflightDistribution;
+        private MetricRegistry.Counter successCounter;
+        private MetricRegistry.Counter droppedCounter;
+        private MetricRegistry.Counter ignoredCounter;
+        private MetricRegistry.Counter rejectedCounter;
 
         Partition(String name) {
             this.name = name;
@@ -140,7 +144,6 @@ boolean isLimitExceeded() {
         void acquire() {
             busy++;
             inflightDistribution.addSample(busy);
-
         }
 
         void release() {
@@ -162,6 +165,11 @@ public int getInflight() {
         void createMetrics(MetricRegistry registry) {
             this.inflightDistribution = registry.distribution(MetricIds.INFLIGHT_NAME, PARTITION_TAG_NAME, name);
             registry.gauge(MetricIds.PARTITION_LIMIT_NAME, this::getLimit, PARTITION_TAG_NAME, name);
+
+            this.successCounter = registry.counter(MetricIds.PARTITIONED_CALL_NAME, PARTITION_TAG_NAME, name, STATUS_TAG, "success");
+            this.droppedCounter = registry.counter(MetricIds.PARTITIONED_CALL_NAME, PARTITION_TAG_NAME, name, STATUS_TAG, "dropped");
+            this.ignoredCounter = registry.counter(MetricIds.PARTITIONED_CALL_NAME, PARTITION_TAG_NAME, name, STATUS_TAG, "ignored");
+            this.rejectedCounter =registry.counter(MetricIds.PARTITIONED_CALL_NAME, PARTITION_TAG_NAME, name, STATUS_TAG, "rejected");
         }
 
         @Override
@@ -228,6 +236,7 @@ public Optional<Listener> acquire(ContextT context) {
                     }
                 }
 
+                partition.rejectedCounter.increment();
                 return createRejectedListener();
             }
 
@@ -237,18 +246,21 @@ public Optional<Listener> acquire(ContextT context) {
                 @Override
                 public void onSuccess() {
                     listener.onSuccess();
+                    partition.successCounter.increment();
                     releasePartition(partition);
                 }
 
                 @Override
                 public void onIgnore() {
                     listener.onIgnore();
+                    partition.ignoredCounter.increment();
                     releasePartition(partition);
                 }
 
                 @Override
                 public void onDropped() {
                     listener.onDropped();
+                    partition.droppedCounter.increment();
                     releasePartition(partition);
                 }
             });

concurrency-limits-core/src/test/java/com/netflix/concurrency/limits/limiter/AbstractPartitionedLimiterTest.java

@@ -1,15 +1,26 @@
 package com.netflix.concurrency.limits.limiter;
 
 import com.netflix.concurrency.limits.Limiter;
+import com.netflix.concurrency.limits.MetricIds;
+import com.netflix.concurrency.limits.MetricRegistry;
 import com.netflix.concurrency.limits.limit.FixedLimit;
 import com.netflix.concurrency.limits.limit.SettableLimit;
+import com.netflix.concurrency.limits.spectator.SpectatorMetricRegistry;
+import com.netflix.spectator.api.DefaultRegistry;
+import java.util.concurrent.TimeUnit;
 import org.junit.Assert;
 import org.junit.Test;
 
 import java.util.Optional;
 import java.util.function.Function;
 
 public class AbstractPartitionedLimiterTest {
+
+  private static final String REGISTRY_ID = "partitioned.limiter.test";
+  private DefaultRegistry registry = new DefaultRegistry();
+  private MetricRegistry metricRegistry =
+      new SpectatorMetricRegistry(registry, registry.createId(REGISTRY_ID));
+
     public static class TestPartitionedLimiter extends AbstractPartitionedLimiter<String> {
         public static class Builder extends AbstractPartitionedLimiter.Builder<Builder, String> {
             @Override
@@ -47,14 +58,16 @@ public void useExcessCapacityUntilTotalLimit() {
                 .partition("batch", 0.3)
                 .partition("live", 0.7)
                 .limit(FixedLimit.of(10))
+                .metricRegistry(metricRegistry)
                 .build();
 
         for (int i = 0; i < 10; i++) {
             Assert.assertTrue(limiter.acquire("batch").isPresent());
             Assert.assertEquals(i+1, limiter.getPartition("batch").getInflight());
         }
 
-        Assert.assertFalse(limiter.acquire("batch").isPresent());
+      Assert.assertFalse(limiter.acquire("batch").isPresent());
+      Assert.assertEquals(1, getCount("batch", "rejected"));
     }
 
     @Test
@@ -64,6 +77,7 @@ public void exceedTotalLimitForUnusedBin() {
                 .partition("batch", 0.3)
                 .partition("live", 0.7)
                 .limit(FixedLimit.of(10))
+                .metricRegistry(metricRegistry)
                 .build();
 
         for (int i = 0; i < 10; i++) {
@@ -72,6 +86,7 @@ public void exceedTotalLimitForUnusedBin() {
         }
 
         Assert.assertFalse(limiter.acquire("batch").isPresent());
+        Assert.assertEquals(1, getCount("batch", "rejected"));
 
         for (int i = 0; i < 7; i++) {
             Assert.assertTrue(limiter.acquire("live").isPresent());
@@ -88,6 +103,7 @@ public void rejectOnceAllLimitsReached() {
                 .partition("batch", 0.3)
                 .partition("live", 0.7)
                 .limit(FixedLimit.of(10))
+                .metricRegistry(metricRegistry)
                 .build();
 
         for (int i = 0; i < 3; i++) {
@@ -103,7 +119,9 @@ public void rejectOnceAllLimitsReached() {
         }
 
         Assert.assertFalse(limiter.acquire("batch").isPresent());
+        Assert.assertEquals(1, getCount("batch", "rejected"));
         Assert.assertFalse(limiter.acquire("live").isPresent());
+        Assert.assertEquals(1, getCount("live", "rejected"));
     }
 
     @Test
@@ -156,4 +174,39 @@ public void setLimitReservesBusy() {
         Assert.assertEquals(1, limiter.getPartition("batch").getInflight());
         Assert.assertEquals(1, limiter.getInflight());
     }
+
+    @Test
+    public void listenerUpdatesCounters() {
+      AbstractPartitionedLimiter<String> limiter = (AbstractPartitionedLimiter<String>) TestPartitionedLimiter.newBuilder()
+          .partitionResolver(Function.identity())
+          .partition("batch", 0.3)
+          .partition("live", 0.7)
+          .limit(FixedLimit.of(10))
+          .metricRegistry(metricRegistry)
+          .build();
+
+      final Optional<Limiter.Listener> optionalListener = limiter.acquire("batch");
+      Assert.assertTrue(optionalListener.isPresent());
+      Limiter.Listener listener = optionalListener.get();
+
+      listener.onSuccess();
+      Assert.assertEquals(1, getCount("batch", "success"));
+
+      listener.onDropped();
+      Assert.assertEquals(1, getCount("batch", "dropped"));
+
+      listener.onIgnore();
+      Assert.assertEquals(1, getCount("batch", "ignored"));
+    }
+
+    public long getCount(String partition, String status) {
+      try {
+        TimeUnit.SECONDS.sleep(1);
+
+        String name = REGISTRY_ID + "." + MetricIds.PARTITIONED_CALL_NAME;
+        return registry.counter(name, "partition", partition, "status", status).count();
+      } catch (InterruptedException e) {
+        throw new RuntimeException(e);
+      }
+    }
 }