RollingCommandLatencyDistributionStream extends RollingDistributionStream<HystrixCommandCompletion> { public static RollingCommandLatencyDistributionStream getInstance(HystrixCommandKey commandKey, HystrixCommandProperties properties) { final int percentileMetricWindow = properties.metricsRollingPercentileWindowInMilliseconds().get(); final int numPercentileBuckets = properties.metricsRollingPercentileWindowBuckets().get(); final int percentileBucketSizeInMs = percentileMetricWindow / numPercentileBuckets; return getInstance(commandKey, numPercentileBuckets, percentileBucketSizeInMs); } private  RollingCommandLatencyDistributionStream(HystrixCommandKey commandKey, int numPercentileBuckets, int percentileBucketSizeInMs); static RollingCommandLatencyDistributionStream getInstance(HystrixCommandKey commandKey, HystrixCommandProperties properties); static RollingCommandLatencyDistributionStream getInstance(HystrixCommandKey commandKey, int numBuckets, int bucketSizeInMs); static void reset();  }
@Test public void testEmptyStreamProducesEmptyDistributions() { HystrixCommandKey key = HystrixCommandKey.Factory.asKey("CMD-Latency-A"); stream = RollingCommandLatencyDistributionStream.getInstance(key, 10, 100); stream.startCachingStreamValuesIfUnstarted(); final CountDownLatch latch = new CountDownLatch(1); stream.observe().take(10).subscribe(new Subscriber<CachedValuesHistogram>() { @Override public void onCompleted() { latch.countDown(); } @Override public void onError(Throwable e) { fail(e.getMessage()); } @Override public void onNext(CachedValuesHistogram distribution) { System.out.println("OnNext @ " + System.currentTimeMillis()); assertEquals(0, distribution.getTotalCount()); } }); try { assertTrue(latch.await(10000, TimeUnit.MILLISECONDS)); } catch (InterruptedException ex) { fail("Interrupted ex"); } assertEquals(0, stream.getLatest().getTotalCount()); }
@Test public void testSingleBucketGetsStored() { HystrixCommandKey key = HystrixCommandKey.Factory.asKey("CMD-Latency-B"); stream = RollingCommandLatencyDistributionStream.getInstance(key, 10, 100); stream.startCachingStreamValuesIfUnstarted(); final CountDownLatch latch = new CountDownLatch(1); stream.observe().take(10).subscribe(new Subscriber<CachedValuesHistogram>() { @Override public void onCompleted() { latch.countDown(); } @Override public void onError(Throwable e) { fail(e.getMessage()); } @Override public void onNext(CachedValuesHistogram distribution) { System.out.println(System.currentTimeMillis() + " : " + Thread.currentThread().getName() + " Received distribution with count : " + distribution.getTotalCount() + " and mean : " + distribution.getMean()); if (distribution.getTotalCount() == 1) { assertBetween(10, 50, (int) distribution.getMean()); } else if (distribution.getTotalCount() == 2) { assertBetween(300, 400, (int) distribution.getMean()); } } }); Command cmd1 = Command.from(groupKey, key, HystrixEventType.SUCCESS, 10); Command cmd2 = Command.from(groupKey, key, HystrixEventType.TIMEOUT); cmd1.observe(); cmd2.observe(); try { assertTrue(latch.await(10000, TimeUnit.MILLISECONDS)); } catch (InterruptedException ex) { fail("Interrupted ex"); } assertBetween(150, 400, stream.getLatestMean()); assertBetween(10, 50, stream.getLatestPercentile(0.0)); assertBetween(300, 800, stream.getLatestPercentile(100.0)); }
@Test public void testSingleBucketWithMultipleEventTypes() { HystrixCommandKey key = HystrixCommandKey.Factory.asKey("CMD-Latency-C"); stream = RollingCommandLatencyDistributionStream.getInstance(key, 10, 100); stream.startCachingStreamValuesIfUnstarted(); final CountDownLatch latch = new CountDownLatch(1); stream.observe().take(10).subscribe(new Subscriber<CachedValuesHistogram>() { @Override public void onCompleted() { latch.countDown(); } @Override public void onError(Throwable e) { fail(e.getMessage()); } @Override public void onNext(CachedValuesHistogram distribution) { System.out.println(System.currentTimeMillis() + " : " + Thread.currentThread().getName() + " Received distribution with count : " + distribution.getTotalCount() + " and mean : " + distribution.getMean()); if (distribution.getTotalCount() < 4 && distribution.getTotalCount() > 0) { assertBetween(10, 50, (int) distribution.getMean()); } else if (distribution.getTotalCount() == 4){ assertBetween(150, 250, (int) distribution.getMean()); } } }); Command cmd1 = Command.from(groupKey, key, HystrixEventType.SUCCESS, 10); Command cmd2 = Command.from(groupKey, key, HystrixEventType.TIMEOUT); Command cmd3 = Command.from(groupKey, key, HystrixEventType.FAILURE, 30); Command cmd4 = Command.from(groupKey, key, HystrixEventType.BAD_REQUEST, 40); cmd1.observe(); cmd2.observe(); cmd3.observe(); cmd4.observe(); try { assertTrue(latch.await(10000, TimeUnit.MILLISECONDS)); } catch (InterruptedException ex) { fail("Interrupted ex"); } assertBetween(150, 350, stream.getLatestMean()); assertBetween(10, 40, stream.getLatestPercentile(0.0)); assertBetween(600, 800, stream.getLatestPercentile(100.0)); }
@Test public void testShortCircuitedCommandDoesNotGetLatencyTracked() { HystrixCommandKey key = HystrixCommandKey.Factory.asKey("CMD-Latency-D"); stream = RollingCommandLatencyDistributionStream.getInstance(key, 10, 100); stream.startCachingStreamValuesIfUnstarted(); List<Command> commands = new ArrayList<Command>(); for (int i = 0; i < 3; i++) { commands.add(Command.from(groupKey, key, HystrixEventType.FAILURE, 0)); } final CountDownLatch latch = new CountDownLatch(1); stream.observe().take(10).subscribe(new Subscriber<CachedValuesHistogram>() { @Override public void onCompleted() { latch.countDown(); } @Override public void onError(Throwable e) { fail(e.getMessage()); } @Override public void onNext(CachedValuesHistogram distribution) { System.out.println(System.currentTimeMillis() + " : " + Thread.currentThread().getName() + " Received distribution with count : " + distribution.getTotalCount() + " and mean : " + distribution.getMean()); assertBetween(0, 30, (int) distribution.getMean()); } }); for (Command cmd: commands) { cmd.observe(); } Command shortCircuit = Command.from(groupKey, key, HystrixEventType.SUCCESS); try { Thread.sleep(200); shortCircuit.observe(); } catch (InterruptedException ie) { fail(ie.getMessage()); } try { assertTrue(latch.await(10000, TimeUnit.MILLISECONDS)); } catch (InterruptedException ex) { fail("Interrupted ex"); } assertEquals(3, stream.getLatest().getTotalCount()); assertBetween(0, 30, stream.getLatestMean()); System.out.println("ReqLog : " + HystrixRequestLog.getCurrentRequest().getExecutedCommandsAsString()); assertTrue(shortCircuit.isResponseShortCircuited()); }
@Test public void testThreadPoolRejectedCommandDoesNotGetLatencyTracked() { HystrixCommandKey key = HystrixCommandKey.Factory.asKey("CMD-Latency-E"); stream = RollingCommandLatencyDistributionStream.getInstance(key, 10, 100); stream.startCachingStreamValuesIfUnstarted(); List<Command> commands = new ArrayList<Command>(); for (int i = 0; i < 10; i++) { commands.add(Command.from(groupKey, key, HystrixEventType.SUCCESS, 200)); } final CountDownLatch latch = new CountDownLatch(1); stream.observe().take(10).subscribe(new Subscriber<CachedValuesHistogram>() { @Override public void onCompleted() { latch.countDown(); } @Override public void onError(Throwable e) { fail(e.getMessage()); } @Override public void onNext(CachedValuesHistogram distribution) { System.out.println(System.currentTimeMillis() + " : " + Thread.currentThread().getName() + " Received distribution with count : " + distribution.getTotalCount() + " and mean : " + distribution.getMean()); } }); for (Command cmd: commands) { cmd.observe(); } Command threadPoolRejected = Command.from(groupKey, key, HystrixEventType.SUCCESS); try { Thread.sleep(40); threadPoolRejected.observe(); } catch (InterruptedException ie) { fail(ie.getMessage()); } try { assertTrue(latch.await(10000, TimeUnit.MILLISECONDS)); } catch (InterruptedException ex) { fail("Interrupted ex"); } assertEquals(10, stream.getLatest().getTotalCount()); assertBetween(200, 250, stream.getLatestMean()); System.out.println("ReqLog : " + HystrixRequestLog.getCurrentRequest().getExecutedCommandsAsString()); assertTrue(threadPoolRejected.isResponseThreadPoolRejected()); }
@Test public void testSemaphoreRejectedCommandDoesNotGetLatencyTracked() { HystrixCommandKey key = HystrixCommandKey.Factory.asKey("CMD-Latency-F"); stream = RollingCommandLatencyDistributionStream.getInstance(key, 10, 100); stream.startCachingStreamValuesIfUnstarted(); List<Command> commands = new ArrayList<Command>(); for (int i = 0; i < 10; i++) { commands.add(Command.from(groupKey, key, HystrixEventType.SUCCESS, 200, HystrixCommandProperties.ExecutionIsolationStrategy.SEMAPHORE)); } final CountDownLatch latch = new CountDownLatch(1); stream.observe().take(10).subscribe(new Subscriber<CachedValuesHistogram>() { @Override public void onCompleted() { latch.countDown(); } @Override public void onError(Throwable e) { fail(e.getMessage()); } @Override public void onNext(CachedValuesHistogram distribution) { System.out.println(System.currentTimeMillis() + " : " + Thread.currentThread().getName() + " Received distribution with count : " + distribution.getTotalCount() + " and mean : " + distribution.getMean()); if (distribution.getTotalCount() > 0) { assertBetween(200, 250, (int) distribution.getMean()); } } }); for (final Command cmd: commands) { new Thread(new HystrixContextRunnable(new Runnable() { @Override public void run() { cmd.observe(); } })).start(); } Command semaphoreRejected = Command.from(groupKey, key, HystrixEventType.SUCCESS); try { Thread.sleep(40); semaphoreRejected.observe(); } catch (InterruptedException ie) { fail(ie.getMessage()); } try { assertTrue(latch.await(10000, TimeUnit.MILLISECONDS)); } catch (InterruptedException ex) { fail("Interrupted ex"); } assertEquals(10, stream.getLatest().getTotalCount()); assertBetween(200, 250, stream.getLatestMean()); System.out.println("ReqLog : " + HystrixRequestLog.getCurrentRequest().getExecutedCommandsAsString()); assertTrue(semaphoreRejected.isResponseSemaphoreRejected()); }
@Test public void testResponseFromCacheDoesNotGetLatencyTracked() { HystrixCommandKey key = HystrixCommandKey.Factory.asKey("CMD-Latency-G"); stream = RollingCommandLatencyDistributionStream.getInstance(key, 10, 100); stream.startCachingStreamValuesIfUnstarted(); List<Command> commands = Command.getCommandsWithResponseFromCache(groupKey, key); final CountDownLatch latch = new CountDownLatch(1); stream.observe().take(10).subscribe(new Subscriber<CachedValuesHistogram>() { @Override public void onCompleted() { latch.countDown(); } @Override public void onError(Throwable e) { fail(e.getMessage()); } @Override public void onNext(CachedValuesHistogram distribution) { System.out.println(System.currentTimeMillis() + " : " + Thread.currentThread().getName() + " Received distribution with count : " + distribution.getTotalCount() + " and mean : " + distribution.getMean()); assertTrue(distribution.getTotalCount() <= 1); } }); for (Command cmd: commands) { cmd.observe(); } try { assertTrue(latch.await(10000, TimeUnit.MILLISECONDS)); } catch (InterruptedException ex) { fail("Interrupted ex"); } assertEquals(1, stream.getLatest().getTotalCount()); assertBetween(0, 30, stream.getLatestMean()); System.out.println("ReqLog : " + HystrixRequestLog.getCurrentRequest().getExecutedCommandsAsString()); }
@Test public void testMultipleBucketsBothGetStored() { HystrixCommandKey key = HystrixCommandKey.Factory.asKey("CMD-Latency-H"); stream = RollingCommandLatencyDistributionStream.getInstance(key, 10, 100); stream.startCachingStreamValuesIfUnstarted(); final CountDownLatch latch = new CountDownLatch(1); stream.observe().take(10).subscribe(new Subscriber<CachedValuesHistogram>() { @Override public void onCompleted() { latch.countDown(); } @Override public void onError(Throwable e) { fail(e.getMessage()); } @Override public void onNext(CachedValuesHistogram distribution) { System.out.println(System.currentTimeMillis() + " : " + Thread.currentThread().getName() + " Received distribution with count : " + distribution.getTotalCount() + " and mean : " + distribution.getMean()); if (distribution.getTotalCount() == 2) { assertBetween(55, 90, (int) distribution.getMean()); } if (distribution.getTotalCount() == 5) { assertEquals(60, 90, (long) distribution.getMean()); } } }); Command cmd1 = Command.from(groupKey, key, HystrixEventType.SUCCESS, 10); Command cmd2 = Command.from(groupKey, key, HystrixEventType.FAILURE, 100); cmd1.observe(); cmd2.observe(); try { Thread.sleep(500); } catch (InterruptedException ie) { fail("Interrupted ex"); } Command cmd3 = Command.from(groupKey, key, HystrixEventType.SUCCESS, 60); Command cmd4 = Command.from(groupKey, key, HystrixEventType.SUCCESS, 60); Command cmd5 = Command.from(groupKey, key, HystrixEventType.SUCCESS, 70); cmd3.observe(); cmd4.observe(); cmd5.observe(); try { assertTrue(latch.await(10000, TimeUnit.MILLISECONDS)); } catch (InterruptedException ex) { fail("Interrupted ex"); } assertBetween(55, 90, stream.getLatestMean()); assertBetween(10, 50, stream.getLatestPercentile(0.0)); assertBetween(100, 150, stream.getLatestPercentile(100.0)); }
@Test public void testMultipleBucketsBothGetStoredAndThenAgeOut() { HystrixCommandKey key = HystrixCommandKey.Factory.asKey("CMD-Latency-I"); stream = RollingCommandLatencyDistributionStream.getInstance(key, 10, 100); stream.startCachingStreamValuesIfUnstarted(); final CountDownLatch latch = new CountDownLatch(1); stream.observe().take(30).subscribe(new Subscriber<CachedValuesHistogram>() { @Override public void onCompleted() { latch.countDown(); } @Override public void onError(Throwable e) { fail(e.getMessage()); } @Override public void onNext(CachedValuesHistogram distribution) { System.out.println(System.currentTimeMillis() + " : " + Thread.currentThread().getName() + " Received distribution with count : " + distribution.getTotalCount() + " and mean : " + distribution.getMean()); if (distribution.getTotalCount() == 2) { assertBetween(55, 90, (int) distribution.getMean()); } if (distribution.getTotalCount() == 5) { assertEquals(60, 90, (long) distribution.getMean()); } } }); Command cmd1 = Command.from(groupKey, key, HystrixEventType.SUCCESS, 10); Command cmd2 = Command.from(groupKey, key, HystrixEventType.FAILURE, 100); cmd1.observe(); cmd2.observe(); try { Thread.sleep(500); } catch (InterruptedException ie) { fail("Interrupted ex"); } Command cmd3 = Command.from(groupKey, key, HystrixEventType.SUCCESS, 60); Command cmd4 = Command.from(groupKey, key, HystrixEventType.SUCCESS, 60); Command cmd5 = Command.from(groupKey, key, HystrixEventType.SUCCESS, 70); cmd3.observe(); cmd4.observe(); cmd5.observe(); try { assertTrue(latch.await(10000, TimeUnit.MILLISECONDS)); } catch (InterruptedException ex) { fail("Interrupted ex"); } assertEquals(0, stream.getLatest().getTotalCount()); }