HealthCountsStream extends BucketedRollingCounterStream<HystrixCommandCompletion, long[], HystrixCommandMetrics.HealthCounts> { public static HealthCountsStream getInstance(HystrixCommandKey commandKey, HystrixCommandProperties properties) { final int healthCountBucketSizeInMs = properties.metricsHealthSnapshotIntervalInMilliseconds().get(); if (healthCountBucketSizeInMs == 0) { throw new RuntimeException("You have set the bucket size to 0ms. Please set a positive number, so that the metric stream can be properly consumed"); } final int numHealthCountBuckets = properties.metricsRollingStatisticalWindowInMilliseconds().get() / healthCountBucketSizeInMs; return getInstance(commandKey, numHealthCountBuckets, healthCountBucketSizeInMs); } private  HealthCountsStream(final HystrixCommandKey commandKey, final int numBuckets, final int bucketSizeInMs,
                               Func2<long[], HystrixCommandCompletion, long[]> reduceCommandCompletion); static HealthCountsStream getInstance(HystrixCommandKey commandKey, HystrixCommandProperties properties); static HealthCountsStream getInstance(HystrixCommandKey commandKey, int numBuckets, int bucketSizeInMs); static void reset(); static void removeByKey(HystrixCommandKey key);  }
@Test public void testEmptyStreamProducesZeros() { HystrixCommandKey key = HystrixCommandKey.Factory.asKey("CMD-Health-A"); stream = HealthCountsStream.getInstance(key, 10, 100); final CountDownLatch latch = new CountDownLatch(1); stream.observe().take(10).subscribe(getSubscriber(latch)); try { assertTrue(latch.await(10000, TimeUnit.MILLISECONDS)); } catch (InterruptedException ex) { fail("Interrupted ex"); } System.out.println("ReqLog : " + HystrixRequestLog.getCurrentRequest().getExecutedCommandsAsString()); assertEquals(0L, stream.getLatest().getErrorCount()); assertEquals(0L, stream.getLatest().getTotalRequests()); }
@Test public void testSingleSuccess() { HystrixCommandKey key = HystrixCommandKey.Factory.asKey("CMD-Health-B"); stream = HealthCountsStream.getInstance(key, 10, 100); final CountDownLatch latch = new CountDownLatch(1); stream.observe().take(10).subscribe(getSubscriber(latch)); CommandStreamTest.Command cmd = CommandStreamTest.Command.from(groupKey, key, HystrixEventType.SUCCESS, 20); cmd.observe(); try { assertTrue(latch.await(10000, TimeUnit.MILLISECONDS)); } catch (InterruptedException ex) { fail("Interrupted ex"); } System.out.println("ReqLog : " + HystrixRequestLog.getCurrentRequest().getExecutedCommandsAsString()); assertEquals(0L, stream.getLatest().getErrorCount()); assertEquals(1L, stream.getLatest().getTotalRequests()); }
@Test public void testSingleFailure() { HystrixCommandKey key = HystrixCommandKey.Factory.asKey("CMD-Health-C"); stream = HealthCountsStream.getInstance(key, 10, 100); final CountDownLatch latch = new CountDownLatch(1); stream.observe().take(10).subscribe(getSubscriber(latch)); CommandStreamTest.Command cmd = CommandStreamTest.Command.from(groupKey, key, HystrixEventType.FAILURE, 20); cmd.observe(); try { assertTrue(latch.await(10000, TimeUnit.MILLISECONDS)); } catch (InterruptedException ex) { fail("Interrupted ex"); } System.out.println("ReqLog : " + HystrixRequestLog.getCurrentRequest().getExecutedCommandsAsString()); assertEquals(1L, stream.getLatest().getErrorCount()); assertEquals(1L, stream.getLatest().getTotalRequests()); }
@Test public void testSingleTimeout() { HystrixCommandKey key = HystrixCommandKey.Factory.asKey("CMD-Health-D"); stream = HealthCountsStream.getInstance(key, 10, 100); final CountDownLatch latch = new CountDownLatch(1); stream.observe().take(10).subscribe(getSubscriber(latch)); CommandStreamTest.Command cmd = CommandStreamTest.Command.from(groupKey, key, HystrixEventType.TIMEOUT); cmd.observe(); try { assertTrue(latch.await(10000, TimeUnit.MILLISECONDS)); } catch (InterruptedException ex) { fail("Interrupted ex"); } System.out.println("ReqLog : " + HystrixRequestLog.getCurrentRequest().getExecutedCommandsAsString()); assertEquals(1L, stream.getLatest().getErrorCount()); assertEquals(1L, stream.getLatest().getTotalRequests()); }
@Test public void testSingleBadRequest() { HystrixCommandKey key = HystrixCommandKey.Factory.asKey("CMD-Health-E"); stream = HealthCountsStream.getInstance(key, 10, 100); final CountDownLatch latch = new CountDownLatch(1); stream.observe().take(10).subscribe(getSubscriber(latch)); CommandStreamTest.Command cmd = CommandStreamTest.Command.from(groupKey, key, HystrixEventType.BAD_REQUEST); cmd.observe(); try { assertTrue(latch.await(10000, TimeUnit.MILLISECONDS)); } catch (InterruptedException ex) { fail("Interrupted ex"); } System.out.println("ReqLog : " + HystrixRequestLog.getCurrentRequest().getExecutedCommandsAsString()); assertEquals(0L, stream.getLatest().getErrorCount()); assertEquals(0L, stream.getLatest().getTotalRequests()); }
@Test public void testRequestFromCache() { HystrixCommandKey key = HystrixCommandKey.Factory.asKey("CMD-Health-F"); stream = HealthCountsStream.getInstance(key, 10, 100); final CountDownLatch latch = new CountDownLatch(1); stream.observe().take(10).subscribe(getSubscriber(latch)); CommandStreamTest.Command cmd1 = CommandStreamTest.Command.from(groupKey, key, HystrixEventType.SUCCESS, 20); CommandStreamTest.Command cmd2 = CommandStreamTest.Command.from(groupKey, key, HystrixEventType.RESPONSE_FROM_CACHE); CommandStreamTest.Command cmd3 = CommandStreamTest.Command.from(groupKey, key, HystrixEventType.RESPONSE_FROM_CACHE); cmd1.observe(); cmd2.observe(); cmd3.observe(); try { assertTrue(latch.await(10000, TimeUnit.MILLISECONDS)); } catch (InterruptedException ex) { fail("Interrupted ex"); } System.out.println("ReqLog : " + HystrixRequestLog.getCurrentRequest().getExecutedCommandsAsString()); assertEquals(0L, stream.getLatest().getErrorCount()); assertEquals(1L, stream.getLatest().getTotalRequests()); }
@Test public void testShortCircuited() { HystrixCommandKey key = HystrixCommandKey.Factory.asKey("CMD-Health-G"); stream = HealthCountsStream.getInstance(key, 10, 100); final CountDownLatch latch = new CountDownLatch(1); stream.observe().take(10).subscribe(getSubscriber(latch)); CommandStreamTest.Command failure1 = CommandStreamTest.Command.from(groupKey, key, HystrixEventType.FAILURE, 20); CommandStreamTest.Command failure2 = CommandStreamTest.Command.from(groupKey, key, HystrixEventType.FAILURE, 20); CommandStreamTest.Command failure3 = CommandStreamTest.Command.from(groupKey, key, HystrixEventType.FAILURE, 20); CommandStreamTest.Command shortCircuit1 = CommandStreamTest.Command.from(groupKey, key, HystrixEventType.SUCCESS); CommandStreamTest.Command shortCircuit2 = CommandStreamTest.Command.from(groupKey, key, HystrixEventType.SUCCESS); failure1.observe(); failure2.observe(); failure3.observe(); try { Thread.sleep(500); } catch (InterruptedException ie) { fail(ie.getMessage()); } shortCircuit1.observe(); shortCircuit2.observe(); try { assertTrue(latch.await(10000, TimeUnit.MILLISECONDS)); } catch (InterruptedException ex) { fail("Interrupted ex"); } assertTrue(shortCircuit1.isResponseShortCircuited()); assertTrue(shortCircuit2.isResponseShortCircuited()); System.out.println("ReqLog : " + HystrixRequestLog.getCurrentRequest().getExecutedCommandsAsString()); assertEquals(3L, stream.getLatest().getErrorCount()); assertEquals(3L, stream.getLatest().getTotalRequests()); }
@Test public void testSemaphoreRejected() { HystrixCommandKey key = HystrixCommandKey.Factory.asKey("CMD-Health-H"); stream = HealthCountsStream.getInstance(key, 10, 100); final CountDownLatch latch = new CountDownLatch(1); stream.observe().take(10).subscribe(getSubscriber(latch)); List<Command> saturators = new ArrayList<Command>(); for (int i = 0; i < 10; i++) { saturators.add(CommandStreamTest.Command.from(groupKey, key, HystrixEventType.SUCCESS, 400, HystrixCommandProperties.ExecutionIsolationStrategy.SEMAPHORE)); } CommandStreamTest.Command rejected1 = CommandStreamTest.Command.from(groupKey, key, HystrixEventType.SUCCESS, 0, HystrixCommandProperties.ExecutionIsolationStrategy.SEMAPHORE); CommandStreamTest.Command rejected2 = CommandStreamTest.Command.from(groupKey, key, HystrixEventType.SUCCESS, 0, HystrixCommandProperties.ExecutionIsolationStrategy.SEMAPHORE); for (final CommandStreamTest.Command saturator : saturators) { new Thread(new HystrixContextRunnable(new Runnable() { @Override public void run() { saturator.observe(); } })).start(); } try { Thread.sleep(100); } catch (InterruptedException ie) { fail(ie.getMessage()); } rejected1.observe(); rejected2.observe(); try { assertTrue(latch.await(10000, TimeUnit.MILLISECONDS)); } catch (InterruptedException ex) { fail("Interrupted ex"); } System.out.println("ReqLog : " + HystrixRequestLog.getCurrentRequest().getExecutedCommandsAsString()); assertTrue(rejected1.isResponseSemaphoreRejected()); assertTrue(rejected2.isResponseSemaphoreRejected()); assertEquals(2L, stream.getLatest().getErrorCount()); assertEquals(12L, stream.getLatest().getTotalRequests()); }
@Test public void testThreadPoolRejected() { HystrixCommandKey key = HystrixCommandKey.Factory.asKey("CMD-Health-I"); stream = HealthCountsStream.getInstance(key, 10, 100); final CountDownLatch latch = new CountDownLatch(1); stream.observe().take(10).subscribe(getSubscriber(latch)); List<CommandStreamTest.Command> saturators = new ArrayList<CommandStreamTest.Command>(); for (int i = 0; i < 10; i++) { saturators.add(CommandStreamTest.Command.from(groupKey, key, HystrixEventType.SUCCESS, 400)); } CommandStreamTest.Command rejected1 = CommandStreamTest.Command.from(groupKey, key, HystrixEventType.SUCCESS, 0); CommandStreamTest.Command rejected2 = CommandStreamTest.Command.from(groupKey, key, HystrixEventType.SUCCESS, 0); for (final CommandStreamTest.Command saturator : saturators) { saturator.observe(); } try { Thread.sleep(100); } catch (InterruptedException ie) { fail(ie.getMessage()); } rejected1.observe(); rejected2.observe(); try { assertTrue(latch.await(10000, TimeUnit.MILLISECONDS)); } catch (InterruptedException ex) { fail("Interrupted ex"); } System.out.println("ReqLog : " + HystrixRequestLog.getCurrentRequest().getExecutedCommandsAsString()); assertTrue(rejected1.isResponseThreadPoolRejected()); assertTrue(rejected2.isResponseThreadPoolRejected()); assertEquals(2L, stream.getLatest().getErrorCount()); assertEquals(12L, stream.getLatest().getTotalRequests()); }
@Test public void testFallbackFailure() { HystrixCommandKey key = HystrixCommandKey.Factory.asKey("CMD-Health-J"); stream = HealthCountsStream.getInstance(key, 10, 100); final CountDownLatch latch = new CountDownLatch(1); stream.observe().take(10).subscribe(getSubscriber(latch)); CommandStreamTest.Command cmd = CommandStreamTest.Command.from(groupKey, key, HystrixEventType.FAILURE, 20, HystrixEventType.FALLBACK_FAILURE); cmd.observe(); try { assertTrue(latch.await(10000, TimeUnit.MILLISECONDS)); } catch (InterruptedException ex) { fail("Interrupted ex"); } System.out.println("ReqLog : " + HystrixRequestLog.getCurrentRequest().getExecutedCommandsAsString()); assertEquals(1L, stream.getLatest().getErrorCount()); assertEquals(1L, stream.getLatest().getTotalRequests()); }
@Test public void testFallbackMissing() { HystrixCommandKey key = HystrixCommandKey.Factory.asKey("CMD-Health-K"); stream = HealthCountsStream.getInstance(key, 10, 100); final CountDownLatch latch = new CountDownLatch(1); stream.observe().take(10).subscribe(getSubscriber(latch)); CommandStreamTest.Command cmd = CommandStreamTest.Command.from(groupKey, key, HystrixEventType.FAILURE, 20, HystrixEventType.FALLBACK_MISSING); cmd.observe(); try { assertTrue(latch.await(10000, TimeUnit.MILLISECONDS)); } catch (InterruptedException ex) { fail("Interrupted ex"); } assertEquals(1L, stream.getLatest().getErrorCount()); assertEquals(1L, stream.getLatest().getTotalRequests()); }
@Test public void testFallbackRejection() { HystrixCommandKey key = HystrixCommandKey.Factory.asKey("CMD-Health-L"); stream = HealthCountsStream.getInstance(key, 10, 100); final CountDownLatch latch = new CountDownLatch(1); stream.observe().take(10).subscribe(getSubscriber(latch)); List<CommandStreamTest.Command> fallbackSaturators = new ArrayList<CommandStreamTest.Command>(); for (int i = 0; i < 5; i++) { fallbackSaturators.add(CommandStreamTest.Command.from(groupKey, key, HystrixEventType.FAILURE, 20, HystrixEventType.FALLBACK_SUCCESS, 400)); } CommandStreamTest.Command rejection1 = CommandStreamTest.Command.from(groupKey, key, HystrixEventType.FAILURE, 20, HystrixEventType.FALLBACK_SUCCESS, 0); CommandStreamTest.Command rejection2 = CommandStreamTest.Command.from(groupKey, key, HystrixEventType.FAILURE, 20, HystrixEventType.FALLBACK_SUCCESS, 0); for (CommandStreamTest.Command saturator: fallbackSaturators) { saturator.observe(); } try { Thread.sleep(70); } catch (InterruptedException ex) { fail(ex.getMessage()); } rejection1.observe(); rejection2.observe(); try { assertTrue(latch.await(10000, TimeUnit.MILLISECONDS)); } catch (InterruptedException ex) { fail("Interrupted ex"); } System.out.println("ReqLog : " + HystrixRequestLog.getCurrentRequest().getExecutedCommandsAsString()); assertEquals(7L, stream.getLatest().getErrorCount()); assertEquals(7L, stream.getLatest().getTotalRequests()); }
@Test public void testMultipleEventsOverTimeGetStoredAndAgeOut() { HystrixCommandKey key = HystrixCommandKey.Factory.asKey("CMD-Health-M"); stream = HealthCountsStream.getInstance(key, 10, 100); final CountDownLatch latch = new CountDownLatch(1); stream.observe().take(30).subscribe(getSubscriber(latch)); CommandStreamTest.Command cmd1 = CommandStreamTest.Command.from(groupKey, key, HystrixEventType.SUCCESS, 20); CommandStreamTest.Command cmd2 = CommandStreamTest.Command.from(groupKey, key, HystrixEventType.FAILURE, 10); cmd1.observe(); cmd2.observe(); try { assertTrue(latch.await(10000, TimeUnit.MILLISECONDS)); } catch (InterruptedException ex) { fail("Interrupted ex"); } System.out.println("ReqLog : " + HystrixRequestLog.getCurrentRequest().getExecutedCommandsAsString()); assertEquals(0L, stream.getLatest().getErrorCount()); assertEquals(0L, stream.getLatest().getTotalRequests()); }
@Test public void testSharedSourceStream() throws InterruptedException { HystrixCommandKey key = HystrixCommandKey.Factory.asKey("CMD-Health-N"); stream = HealthCountsStream.getInstance(key, 10, 100); final CountDownLatch latch = new CountDownLatch(1); final AtomicBoolean allEqual = new AtomicBoolean(false); Observable<HystrixCommandMetrics.HealthCounts> o1 = stream .observe() .take(10) .observeOn(Schedulers.computation()); Observable<HystrixCommandMetrics.HealthCounts> o2 = stream .observe() .take(10) .observeOn(Schedulers.computation()); Observable<Boolean> zipped = Observable.zip(o1, o2, new Func2<HystrixCommandMetrics.HealthCounts, HystrixCommandMetrics.HealthCounts, Boolean>() { @Override public Boolean call(HystrixCommandMetrics.HealthCounts healthCounts, HystrixCommandMetrics.HealthCounts healthCounts2) { return healthCounts == healthCounts2; } }); Observable < Boolean > reduced = zipped.reduce(true, new Func2<Boolean, Boolean, Boolean>() { @Override public Boolean call(Boolean a, Boolean b) { return a && b; } }); reduced.subscribe(new Subscriber<Boolean>() { @Override public void onCompleted() { System.out.println(System.currentTimeMillis() + " : " + Thread.currentThread().getName() + " Reduced OnCompleted"); latch.countDown(); } @Override public void onError(Throwable e) { System.out.println(System.currentTimeMillis() + " : " + Thread.currentThread().getName() + " Reduced OnError : " + e); e.printStackTrace(); latch.countDown(); } @Override public void onNext(Boolean b) { System.out.println(System.currentTimeMillis() + " : " + Thread.currentThread().getName() + " Reduced OnNext : " + b); allEqual.set(b); } }); for (int i = 0; i < 10; i++) { HystrixCommand<Integer> cmd = CommandStreamTest.Command.from(groupKey, key, HystrixEventType.SUCCESS, 20); cmd.execute(); } assertTrue(latch.await(10000, TimeUnit.MILLISECONDS)); assertTrue(allEqual.get()); }
@Test public void testTwoSubscribersOneUnsubscribes() throws Exception { HystrixCommandKey key = HystrixCommandKey.Factory.asKey("CMD-Health-O"); stream = HealthCountsStream.getInstance(key, 10, 100); final CountDownLatch latch1 = new CountDownLatch(1); final CountDownLatch latch2 = new CountDownLatch(1); final AtomicInteger healthCounts1 = new AtomicInteger(0); final AtomicInteger healthCounts2 = new AtomicInteger(0); Subscription s1 = stream .observe() .take(10) .observeOn(Schedulers.computation()) .doOnUnsubscribe(new Action0() { @Override public void call() { latch1.countDown(); } }) .subscribe(new Subscriber<HystrixCommandMetrics.HealthCounts>() { @Override public void onCompleted() { System.out.println(System.currentTimeMillis() + " : " + Thread.currentThread().getName() + " : Health 1 OnCompleted"); latch1.countDown(); } @Override public void onError(Throwable e) { System.out.println(System.currentTimeMillis() + " : " + Thread.currentThread().getName() + " : Health 1 OnError : " + e); latch1.countDown(); } @Override public void onNext(HystrixCommandMetrics.HealthCounts healthCounts) { System.out.println(System.currentTimeMillis() + " : " + Thread.currentThread().getName() + " : Health 1 OnNext : " + healthCounts); healthCounts1.incrementAndGet(); } }); Subscription s2 = stream .observe() .take(10) .observeOn(Schedulers.computation()) .doOnUnsubscribe(new Action0() { @Override public void call() { latch2.countDown(); } }) .subscribe(new Subscriber<HystrixCommandMetrics.HealthCounts>() { @Override public void onCompleted() { System.out.println(System.currentTimeMillis() + " : " + Thread.currentThread().getName() + " : Health 2 OnCompleted"); latch2.countDown(); } @Override public void onError(Throwable e) { System.out.println(System.currentTimeMillis() + " : " + Thread.currentThread().getName() + " : Health 2 OnError : " + e); latch2.countDown(); } @Override public void onNext(HystrixCommandMetrics.HealthCounts healthCounts) { System.out.println(System.currentTimeMillis() + " : " + Thread.currentThread().getName() + " : Health 2 OnNext : " + healthCounts + " : " + healthCounts2.get()); healthCounts2.incrementAndGet(); } }); for (int i = 0; i < 10; i++) { HystrixCommand<Integer> cmd = CommandStreamTest.Command.from(groupKey, key, HystrixEventType.SUCCESS, 20); cmd.execute(); if (i == 5) { s1.unsubscribe(); } } assertTrue(stream.isSourceCurrentlySubscribed()); assertTrue(latch1.await(10000, TimeUnit.MILLISECONDS)); assertTrue(latch2.await(10000, TimeUnit.MILLISECONDS)); System.out.println("s1 got : " + healthCounts1.get() + ", s2 got : " + healthCounts2.get()); assertTrue("s1 got data", healthCounts1.get() > 0); assertTrue("s2 got data", healthCounts2.get() > 0); assertTrue("s1 got less data than s2", healthCounts2.get() > healthCounts1.get()); }