@Test public void verifyTotalRequestMetricIsNotReportedWhenMetricsAreNotIncremented() throws TException { final long totalRequests = TEST_END_BLOCK - TEST_START_BLOCK; assertEquals(metrics.getCounters().get(BookKeeperMetrics.CacheMetric.TOTAL_REQUEST_COUNT.getMetricName()).getCount(), 0); CacheStatusRequest request = new CacheStatusRequest(TEST_REMOTE_PATH, TEST_FILE_LENGTH, TEST_LAST_MODIFIED, TEST_START_BLOCK, TEST_END_BLOCK).setClusterType(ClusterType.TEST_CLUSTER_MANAGER.ordinal()); bookKeeper.getCacheStatus(request); assertEquals(metrics.getCounters().get(BookKeeperMetrics.CacheMetric.TOTAL_REQUEST_COUNT.getMetricName()).getCount(), 0); }
@Override public CacheStatusResponse getCacheStatus(CacheStatusRequest request) throws TException { try { initializeClusterManager(request); } catch (ClusterManagerInitilizationException ex) { log.error("Not able to initialize ClusterManager for cluster type : " + ClusterType.findByValue(request.getClusterType()) + " with Exception : " + ex); return null; } String currentNodeName = clusterManager.getCurrentNodeName(); Map<Long, String> blockSplits = new HashMap<>(); long blockNumber = 0; long fileLength = request.getFileLength(); String remotePath = request.getRemotePath(); long lastModified = request.getLastModified(); long startBlock = request.getStartBlock(); long endBlock = request.getEndBlock(); boolean incrMetrics = request.isIncrMetrics(); for (long i = 0; i < fileLength; i = i + splitSize) { long end = i + splitSize; if (end > fileLength) { end = fileLength; } String key = remotePath + i + end; String nodeAddress = clusterManager.locateKey(key); blockSplits.put(blockNumber, nodeAddress); blockNumber++; } FileMetadata md; try { md = fileMetadataCache.get(remotePath, () -> new FileMetadata( remotePath, fileLength, lastModified, 0, conf, generationNumberCache, fileAccessedBloomFilter)); if (isInvalidationRequired(md.getLastModified(), lastModified)) { invalidateFileMetadata(remotePath); md = fileMetadataCache.get(remotePath, () -> new FileMetadata( remotePath, fileLength, lastModified, 0, conf, generationNumberCache, fileAccessedBloomFilter)); } } catch (ExecutionException e) { log.error(String.format("Could not fetch Metadata for %s", remotePath), e); throw new TException(e); } endBlock = setCorrectEndBlock(endBlock, fileLength, remotePath); List<BlockLocation> blockLocations = new ArrayList<>((int) (endBlock - startBlock)); int blockSize = CacheConfig.getBlockSize(conf); int totalRequests = 0; int cacheRequests = 0; int remoteRequests = 0; int nonLocalRequests = 0; try { for (long blockNum = startBlock; blockNum < endBlock; blockNum++) { totalRequests++; long split = (blockNum * blockSize) / splitSize; if (!blockSplits.get(split).equals(currentNodeName)) { blockLocations.add(new BlockLocation(Location.NON_LOCAL, blockSplits.get(split))); nonLocalRequests++; } else { if (md.isBlockCached(blockNum)) { blockLocations.add(new BlockLocation(Location.CACHED, blockSplits.get(split))); cacheRequests++; } else { blockLocations.add(new BlockLocation(Location.LOCAL, blockSplits.get(split))); remoteRequests++; } } } } catch (IOException e) { throw new TException(e); } if (request.isIncrMetrics() && !isValidatingCachingBehavior(remotePath)) { totalRequestCount.inc(totalRequests); nonlocalRequestCount.inc(nonLocalRequests); cacheRequestCount.inc(cacheRequests); remoteRequestCount.inc(remoteRequests); } return new CacheStatusResponse(blockLocations, md.getGenerationNumber()); }
BookKeeper implements BookKeeperService.Iface { @Override public CacheStatusResponse getCacheStatus(CacheStatusRequest request) throws TException { try { initializeClusterManager(request); } catch (ClusterManagerInitilizationException ex) { log.error("Not able to initialize ClusterManager for cluster type : " + ClusterType.findByValue(request.getClusterType()) + " with Exception : " + ex); return null; } String currentNodeName = clusterManager.getCurrentNodeName(); Map<Long, String> blockSplits = new HashMap<>(); long blockNumber = 0; long fileLength = request.getFileLength(); String remotePath = request.getRemotePath(); long lastModified = request.getLastModified(); long startBlock = request.getStartBlock(); long endBlock = request.getEndBlock(); boolean incrMetrics = request.isIncrMetrics(); for (long i = 0; i < fileLength; i = i + splitSize) { long end = i + splitSize; if (end > fileLength) { end = fileLength; } String key = remotePath + i + end; String nodeAddress = clusterManager.locateKey(key); blockSplits.put(blockNumber, nodeAddress); blockNumber++; } FileMetadata md; try { md = fileMetadataCache.get(remotePath, () -> new FileMetadata( remotePath, fileLength, lastModified, 0, conf, generationNumberCache, fileAccessedBloomFilter)); if (isInvalidationRequired(md.getLastModified(), lastModified)) { invalidateFileMetadata(remotePath); md = fileMetadataCache.get(remotePath, () -> new FileMetadata( remotePath, fileLength, lastModified, 0, conf, generationNumberCache, fileAccessedBloomFilter)); } } catch (ExecutionException e) { log.error(String.format("Could not fetch Metadata for %s", remotePath), e); throw new TException(e); } endBlock = setCorrectEndBlock(endBlock, fileLength, remotePath); List<BlockLocation> blockLocations = new ArrayList<>((int) (endBlock - startBlock)); int blockSize = CacheConfig.getBlockSize(conf); int totalRequests = 0; int cacheRequests = 0; int remoteRequests = 0; int nonLocalRequests = 0; try { for (long blockNum = startBlock; blockNum < endBlock; blockNum++) { totalRequests++; long split = (blockNum * blockSize) / splitSize; if (!blockSplits.get(split).equals(currentNodeName)) { blockLocations.add(new BlockLocation(Location.NON_LOCAL, blockSplits.get(split))); nonLocalRequests++; } else { if (md.isBlockCached(blockNum)) { blockLocations.add(new BlockLocation(Location.CACHED, blockSplits.get(split))); cacheRequests++; } else { blockLocations.add(new BlockLocation(Location.LOCAL, blockSplits.get(split))); remoteRequests++; } } } } catch (IOException e) { throw new TException(e); } if (request.isIncrMetrics() && !isValidatingCachingBehavior(remotePath)) { totalRequestCount.inc(totalRequests); nonlocalRequestCount.inc(nonLocalRequests); cacheRequestCount.inc(cacheRequests); remoteRequestCount.inc(remoteRequests); } return new CacheStatusResponse(blockLocations, md.getGenerationNumber()); } }
BookKeeper implements BookKeeperService.Iface { @Override public CacheStatusResponse getCacheStatus(CacheStatusRequest request) throws TException { try { initializeClusterManager(request); } catch (ClusterManagerInitilizationException ex) { log.error("Not able to initialize ClusterManager for cluster type : " + ClusterType.findByValue(request.getClusterType()) + " with Exception : " + ex); return null; } String currentNodeName = clusterManager.getCurrentNodeName(); Map<Long, String> blockSplits = new HashMap<>(); long blockNumber = 0; long fileLength = request.getFileLength(); String remotePath = request.getRemotePath(); long lastModified = request.getLastModified(); long startBlock = request.getStartBlock(); long endBlock = request.getEndBlock(); boolean incrMetrics = request.isIncrMetrics(); for (long i = 0; i < fileLength; i = i + splitSize) { long end = i + splitSize; if (end > fileLength) { end = fileLength; } String key = remotePath + i + end; String nodeAddress = clusterManager.locateKey(key); blockSplits.put(blockNumber, nodeAddress); blockNumber++; } FileMetadata md; try { md = fileMetadataCache.get(remotePath, () -> new FileMetadata( remotePath, fileLength, lastModified, 0, conf, generationNumberCache, fileAccessedBloomFilter)); if (isInvalidationRequired(md.getLastModified(), lastModified)) { invalidateFileMetadata(remotePath); md = fileMetadataCache.get(remotePath, () -> new FileMetadata( remotePath, fileLength, lastModified, 0, conf, generationNumberCache, fileAccessedBloomFilter)); } } catch (ExecutionException e) { log.error(String.format("Could not fetch Metadata for %s", remotePath), e); throw new TException(e); } endBlock = setCorrectEndBlock(endBlock, fileLength, remotePath); List<BlockLocation> blockLocations = new ArrayList<>((int) (endBlock - startBlock)); int blockSize = CacheConfig.getBlockSize(conf); int totalRequests = 0; int cacheRequests = 0; int remoteRequests = 0; int nonLocalRequests = 0; try { for (long blockNum = startBlock; blockNum < endBlock; blockNum++) { totalRequests++; long split = (blockNum * blockSize) / splitSize; if (!blockSplits.get(split).equals(currentNodeName)) { blockLocations.add(new BlockLocation(Location.NON_LOCAL, blockSplits.get(split))); nonLocalRequests++; } else { if (md.isBlockCached(blockNum)) { blockLocations.add(new BlockLocation(Location.CACHED, blockSplits.get(split))); cacheRequests++; } else { blockLocations.add(new BlockLocation(Location.LOCAL, blockSplits.get(split))); remoteRequests++; } } } } catch (IOException e) { throw new TException(e); } if (request.isIncrMetrics() && !isValidatingCachingBehavior(remotePath)) { totalRequestCount.inc(totalRequests); nonlocalRequestCount.inc(nonLocalRequests); cacheRequestCount.inc(cacheRequests); remoteRequestCount.inc(remoteRequests); } return new CacheStatusResponse(blockLocations, md.getGenerationNumber()); } BookKeeper(Configuration conf, BookKeeperMetrics bookKeeperMetrics); @VisibleForTesting  BookKeeper(Configuration conf, BookKeeperMetrics bookKeeperMetrics, Ticker ticker); }
BookKeeper implements BookKeeperService.Iface { @Override public CacheStatusResponse getCacheStatus(CacheStatusRequest request) throws TException { try { initializeClusterManager(request); } catch (ClusterManagerInitilizationException ex) { log.error("Not able to initialize ClusterManager for cluster type : " + ClusterType.findByValue(request.getClusterType()) + " with Exception : " + ex); return null; } String currentNodeName = clusterManager.getCurrentNodeName(); Map<Long, String> blockSplits = new HashMap<>(); long blockNumber = 0; long fileLength = request.getFileLength(); String remotePath = request.getRemotePath(); long lastModified = request.getLastModified(); long startBlock = request.getStartBlock(); long endBlock = request.getEndBlock(); boolean incrMetrics = request.isIncrMetrics(); for (long i = 0; i < fileLength; i = i + splitSize) { long end = i + splitSize; if (end > fileLength) { end = fileLength; } String key = remotePath + i + end; String nodeAddress = clusterManager.locateKey(key); blockSplits.put(blockNumber, nodeAddress); blockNumber++; } FileMetadata md; try { md = fileMetadataCache.get(remotePath, () -> new FileMetadata( remotePath, fileLength, lastModified, 0, conf, generationNumberCache, fileAccessedBloomFilter)); if (isInvalidationRequired(md.getLastModified(), lastModified)) { invalidateFileMetadata(remotePath); md = fileMetadataCache.get(remotePath, () -> new FileMetadata( remotePath, fileLength, lastModified, 0, conf, generationNumberCache, fileAccessedBloomFilter)); } } catch (ExecutionException e) { log.error(String.format("Could not fetch Metadata for %s", remotePath), e); throw new TException(e); } endBlock = setCorrectEndBlock(endBlock, fileLength, remotePath); List<BlockLocation> blockLocations = new ArrayList<>((int) (endBlock - startBlock)); int blockSize = CacheConfig.getBlockSize(conf); int totalRequests = 0; int cacheRequests = 0; int remoteRequests = 0; int nonLocalRequests = 0; try { for (long blockNum = startBlock; blockNum < endBlock; blockNum++) { totalRequests++; long split = (blockNum * blockSize) / splitSize; if (!blockSplits.get(split).equals(currentNodeName)) { blockLocations.add(new BlockLocation(Location.NON_LOCAL, blockSplits.get(split))); nonLocalRequests++; } else { if (md.isBlockCached(blockNum)) { blockLocations.add(new BlockLocation(Location.CACHED, blockSplits.get(split))); cacheRequests++; } else { blockLocations.add(new BlockLocation(Location.LOCAL, blockSplits.get(split))); remoteRequests++; } } } } catch (IOException e) { throw new TException(e); } if (request.isIncrMetrics() && !isValidatingCachingBehavior(remotePath)) { totalRequestCount.inc(totalRequests); nonlocalRequestCount.inc(nonLocalRequests); cacheRequestCount.inc(cacheRequests); remoteRequestCount.inc(remoteRequests); } return new CacheStatusResponse(blockLocations, md.getGenerationNumber()); } BookKeeper(Configuration conf, BookKeeperMetrics bookKeeperMetrics); @VisibleForTesting  BookKeeper(Configuration conf, BookKeeperMetrics bookKeeperMetrics, Ticker ticker); @Override boolean isBookKeeperAlive(); @Override CacheStatusResponse getCacheStatus(CacheStatusRequest request); boolean isInitialized(); @VisibleForTesting void initializeClusterManager(int clusterType); @VisibleForTesting ClusterManager getClusterManagerInstance(ClusterType clusterType, Configuration config); @Override void setAllCached(String remotePath, long fileLength, long lastModified, long startBlock, long endBlock, int generationNumber); @VisibleForTesting long getTotalCacheWeight(); @Override Map<String, Double> getCacheMetrics(); @Override Map<String, Long> getReadRequestChainStats(); @Override ReadResponse readData(String remotePath, long offset, int length, long fileSize, long lastModified, int clusterType); @Override FileInfo getFileInfo(String remotePath); @VisibleForTesting FileMetadata getFileMetadata(String key); @Override void invalidateFileMetadata(String key); }
BookKeeper implements BookKeeperService.Iface { @Override public CacheStatusResponse getCacheStatus(CacheStatusRequest request) throws TException { try { initializeClusterManager(request); } catch (ClusterManagerInitilizationException ex) { log.error("Not able to initialize ClusterManager for cluster type : " + ClusterType.findByValue(request.getClusterType()) + " with Exception : " + ex); return null; } String currentNodeName = clusterManager.getCurrentNodeName(); Map<Long, String> blockSplits = new HashMap<>(); long blockNumber = 0; long fileLength = request.getFileLength(); String remotePath = request.getRemotePath(); long lastModified = request.getLastModified(); long startBlock = request.getStartBlock(); long endBlock = request.getEndBlock(); boolean incrMetrics = request.isIncrMetrics(); for (long i = 0; i < fileLength; i = i + splitSize) { long end = i + splitSize; if (end > fileLength) { end = fileLength; } String key = remotePath + i + end; String nodeAddress = clusterManager.locateKey(key); blockSplits.put(blockNumber, nodeAddress); blockNumber++; } FileMetadata md; try { md = fileMetadataCache.get(remotePath, () -> new FileMetadata( remotePath, fileLength, lastModified, 0, conf, generationNumberCache, fileAccessedBloomFilter)); if (isInvalidationRequired(md.getLastModified(), lastModified)) { invalidateFileMetadata(remotePath); md = fileMetadataCache.get(remotePath, () -> new FileMetadata( remotePath, fileLength, lastModified, 0, conf, generationNumberCache, fileAccessedBloomFilter)); } } catch (ExecutionException e) { log.error(String.format("Could not fetch Metadata for %s", remotePath), e); throw new TException(e); } endBlock = setCorrectEndBlock(endBlock, fileLength, remotePath); List<BlockLocation> blockLocations = new ArrayList<>((int) (endBlock - startBlock)); int blockSize = CacheConfig.getBlockSize(conf); int totalRequests = 0; int cacheRequests = 0; int remoteRequests = 0; int nonLocalRequests = 0; try { for (long blockNum = startBlock; blockNum < endBlock; blockNum++) { totalRequests++; long split = (blockNum * blockSize) / splitSize; if (!blockSplits.get(split).equals(currentNodeName)) { blockLocations.add(new BlockLocation(Location.NON_LOCAL, blockSplits.get(split))); nonLocalRequests++; } else { if (md.isBlockCached(blockNum)) { blockLocations.add(new BlockLocation(Location.CACHED, blockSplits.get(split))); cacheRequests++; } else { blockLocations.add(new BlockLocation(Location.LOCAL, blockSplits.get(split))); remoteRequests++; } } } } catch (IOException e) { throw new TException(e); } if (request.isIncrMetrics() && !isValidatingCachingBehavior(remotePath)) { totalRequestCount.inc(totalRequests); nonlocalRequestCount.inc(nonLocalRequests); cacheRequestCount.inc(cacheRequests); remoteRequestCount.inc(remoteRequests); } return new CacheStatusResponse(blockLocations, md.getGenerationNumber()); } BookKeeper(Configuration conf, BookKeeperMetrics bookKeeperMetrics); @VisibleForTesting  BookKeeper(Configuration conf, BookKeeperMetrics bookKeeperMetrics, Ticker ticker); @Override boolean isBookKeeperAlive(); @Override CacheStatusResponse getCacheStatus(CacheStatusRequest request); boolean isInitialized(); @VisibleForTesting void initializeClusterManager(int clusterType); @VisibleForTesting ClusterManager getClusterManagerInstance(ClusterType clusterType, Configuration config); @Override void setAllCached(String remotePath, long fileLength, long lastModified, long startBlock, long endBlock, int generationNumber); @VisibleForTesting long getTotalCacheWeight(); @Override Map<String, Double> getCacheMetrics(); @Override Map<String, Long> getReadRequestChainStats(); @Override ReadResponse readData(String remotePath, long offset, int length, long fileSize, long lastModified, int clusterType); @Override FileInfo getFileInfo(String remotePath); @VisibleForTesting FileMetadata getFileMetadata(String key); @Override void invalidateFileMetadata(String key);  }