AbfsPerfTracker { public String getClientLatency() { if (!enabled) { return null; } Instant trackerStart = Instant.now(); String latencyDetails = queue.poll(); if (LOG.isDebugEnabled()) { Instant stop = Instant.now(); long elapsed = Duration.between(trackerStart, stop).toMillis(); LOG.debug("Dequeued latency info [{} ms]: {}", elapsed, latencyDetails); } return latencyDetails; } AbfsPerfTracker(String filesystemName, String accountName, AbfsConfiguration configuration); protected  AbfsPerfTracker(String filesystemName, String accountName, boolean enabled); void trackInfo(AbfsPerfInfo perfInfo); Instant getLatencyInstant(); String getClientLatency();  }
@Test public void verifyDisablingOfTracker() throws Exception { AbfsPerfTracker abfsPerfTracker = new AbfsPerfTracker(accountName, filesystemName, false); String latencyDetails = abfsPerfTracker.getClientLatency(); assertThat(latencyDetails).describedAs("AbfsPerfTracker should be empty").isNull(); try (AbfsPerfInfo tracker = new AbfsPerfInfo(abfsPerfTracker, "disablingCaller", "disablingCallee")) { AbfsHttpOperation op = new AbfsHttpOperation(url, "GET", new ArrayList<>()); tracker.registerResult(op).registerSuccess(true); } latencyDetails = abfsPerfTracker.getClientLatency(); assertThat(latencyDetails).describedAs("AbfsPerfTracker should return no record").isNull(); }
@Test public void verifyTrackingForSingletonLatencyRecords() throws Exception { final int numTasks = 100; AbfsPerfTracker abfsPerfTracker = new AbfsPerfTracker(accountName, filesystemName, true); String latencyDetails = abfsPerfTracker.getClientLatency(); assertThat(latencyDetails).describedAs("AbfsPerfTracker should be empty").isNull(); List<Callable<Integer>> tasks = new ArrayList<>(); AbfsHttpOperation httpOperation = new AbfsHttpOperation(url, "GET", new ArrayList<>()); for (int i = 0; i < numTasks; i++) { tasks.add(() -> { try (AbfsPerfInfo tracker = new AbfsPerfInfo(abfsPerfTracker, "oneOperationCaller", "oneOperationCallee")) { tracker.registerResult(httpOperation).registerSuccess(true); return 0; } }); } for (Future<Integer> fr: executorService.invokeAll(tasks)) { fr.get(); } for (int i = 0; i < numTasks; i++) { latencyDetails = abfsPerfTracker.getClientLatency(); assertThat(latencyDetails).describedAs("AbfsPerfTracker should return non-null record").isNotNull(); assertThat(latencyDetails).describedAs("Latency record should be in the correct format") .containsPattern("h=[^ ]* t=[^ ]* a=bogusFilesystemName c=bogusAccountName cr=oneOperationCaller" + " ce=oneOperationCallee r=Succeeded l=[0-9]+ s=0 e= ci=[^ ]* ri=[^ ]* bs=0 br=0 m=GET" + " u=http%3A%2F%2Fwww.microsoft.com%2FbogusFile"); } latencyDetails = abfsPerfTracker.getClientLatency(); assertThat(latencyDetails).describedAs("AbfsPerfTracker should return no record").isNull(); }
@Test public void verifyTrackingForAggregateLatencyRecords() throws Exception { final int numTasks = 100; AbfsPerfTracker abfsPerfTracker = new AbfsPerfTracker(accountName, filesystemName, true); String latencyDetails = abfsPerfTracker.getClientLatency(); assertThat(latencyDetails).describedAs("AbfsPerfTracker should be empty").isNull(); List<Callable<Integer>> tasks = new ArrayList<>(); AbfsHttpOperation httpOperation = new AbfsHttpOperation(url, "GET", new ArrayList<>()); for (int i = 0; i < numTasks; i++) { tasks.add(() -> { try (AbfsPerfInfo tracker = new AbfsPerfInfo(abfsPerfTracker, "oneOperationCaller", "oneOperationCallee")) { tracker.registerResult(httpOperation).registerSuccess(true) .registerAggregates(Instant.now(), TEST_AGGREGATE_COUNT); return 0; } }); } for (Future<Integer> fr: executorService.invokeAll(tasks)) { fr.get(); } for (int i = 0; i < numTasks; i++) { latencyDetails = abfsPerfTracker.getClientLatency(); assertThat(latencyDetails).describedAs("AbfsPerfTracker should return non-null record").isNotNull(); assertThat(latencyDetails).describedAs("Latency record should be in the correct format") .containsPattern("h=[^ ]* t=[^ ]* a=bogusFilesystemName c=bogusAccountName cr=oneOperationCaller" + " ce=oneOperationCallee r=Succeeded l=[0-9]+ ls=[0-9]+ lc=" + TEST_AGGREGATE_COUNT + " s=0 e= ci=[^ ]* ri=[^ ]* bs=0 br=0 m=GET u=http%3A%2F%2Fwww.microsoft.com%2FbogusFile"); } latencyDetails = abfsPerfTracker.getClientLatency(); assertThat(latencyDetails).describedAs("AbfsPerfTracker should return no record").isNull(); }
@Test public void verifyGettingLatencyRecordsIsCheapWhenDisabled() throws Exception { final double maxLatencyWhenDisabledMs = 1000; final double minLatencyWhenDisabledMs = 0; final long numTasks = 1000; long aggregateLatency = 0; AbfsPerfTracker abfsPerfTracker = new AbfsPerfTracker(accountName, filesystemName, false); List<Callable<Long>> tasks = new ArrayList<>(); for (int i = 0; i < numTasks; i++) { tasks.add(() -> { Instant startGet = Instant.now(); abfsPerfTracker.getClientLatency(); long latencyGet = Duration.between(startGet, Instant.now()).toMillis(); LOG.debug("Spent {} ms in retrieving latency record.", latencyGet); return latencyGet; }); } for (Future<Long> fr: executorService.invokeAll(tasks)) { aggregateLatency += fr.get(); } double averageRecordLatency = aggregateLatency / numTasks; assertThat(averageRecordLatency).describedAs("Average time for getting latency records should be bounded") .isBetween(minLatencyWhenDisabledMs, maxLatencyWhenDisabledMs); }
@Test public void verifyGettingLatencyRecordsIsCheapWhenEnabled() throws Exception { final double maxLatencyWhenDisabledMs = 5000; final double minLatencyWhenDisabledMs = 0; final long numTasks = 1000; long aggregateLatency = 0; AbfsPerfTracker abfsPerfTracker = new AbfsPerfTracker(accountName, filesystemName, true); List<Callable<Long>> tasks = new ArrayList<>(); for (int i = 0; i < numTasks; i++) { tasks.add(() -> { Instant startRecord = Instant.now(); abfsPerfTracker.getClientLatency(); long latencyRecord = Duration.between(startRecord, Instant.now()).toMillis(); LOG.debug("Spent {} ms in recording latency.", latencyRecord); return latencyRecord; }); } for (Future<Long> fr: executorService.invokeAll(tasks)) { aggregateLatency += fr.get(); } double averageRecordLatency = aggregateLatency / numTasks; assertThat(averageRecordLatency).describedAs("Average time for getting latency records should be bounded") .isBetween(minLatencyWhenDisabledMs, maxLatencyWhenDisabledMs); }