StorageLocationChecker { public List<StorageLocation> check( final Configuration conf, final Collection<StorageLocation> dataDirs) throws InterruptedException, IOException { final HashMap<StorageLocation, Boolean> goodLocations = new LinkedHashMap<>(); final Set<StorageLocation> failedLocations = new HashSet<>(); final Map<StorageLocation, ListenableFuture<VolumeCheckResult>> futures = Maps.newHashMap(); final LocalFileSystem localFS = FileSystem.getLocal(conf); final CheckContext context = new CheckContext(localFS, expectedPermission); for (StorageLocation location : dataDirs) { goodLocations.put(location, true); Optional<ListenableFuture<VolumeCheckResult>> olf = delegateChecker.schedule(location, context); if (olf.isPresent()) { futures.put(location, olf.get()); } } if (maxVolumeFailuresTolerated >= dataDirs.size()) { throw new HadoopIllegalArgumentException("Invalid value configured for " + DFS_DATANODE_FAILED_VOLUMES_TOLERATED_KEY + " - " + maxVolumeFailuresTolerated + ". Value configured is >= " + "to the number of configured volumes (" + dataDirs.size() + ")."); } final long checkStartTimeMs = timer.monotonicNow(); for (Map.Entry<StorageLocation, ListenableFuture<VolumeCheckResult>> entry : futures.entrySet()) { final long waitSoFarMs = (timer.monotonicNow() - checkStartTimeMs); final long timeLeftMs = Math.max(0, maxAllowedTimeForCheckMs - waitSoFarMs); final StorageLocation location = entry.getKey(); try { final VolumeCheckResult result = entry.getValue().get(timeLeftMs, TimeUnit.MILLISECONDS); switch (result) { case HEALTHY: break; case DEGRADED: LOG.warn("StorageLocation {} appears to be degraded.", location); break; case FAILED: LOG.warn("StorageLocation {} detected as failed.", location); failedLocations.add(location); goodLocations.remove(location); break; default: LOG.error("Unexpected health check result {} for StorageLocation {}", result, location); } } catch (ExecutionException|TimeoutException e) { LOG.warn("Exception checking StorageLocation " + location, e.getCause()); failedLocations.add(location); goodLocations.remove(location); } } if (maxVolumeFailuresTolerated == DataNode.MAX_VOLUME_FAILURE_TOLERATED_LIMIT) { if (dataDirs.size() == failedLocations.size()) { throw new DiskErrorException("Too many failed volumes - " + "current valid volumes: " + goodLocations.size() + ", volumes configured: " + dataDirs.size() + ", volumes failed: " + failedLocations.size() + ", volume failures tolerated: " + maxVolumeFailuresTolerated); } } else { if (failedLocations.size() > maxVolumeFailuresTolerated) { throw new DiskErrorException("Too many failed volumes - " + "current valid volumes: " + goodLocations.size() + ", volumes configured: " + dataDirs.size() + ", volumes failed: " + failedLocations.size() + ", volume failures tolerated: " + maxVolumeFailuresTolerated); } } if (goodLocations.size() == 0) { throw new DiskErrorException("All directories in " + DFS_DATANODE_DATA_DIR_KEY + " are invalid: " + failedLocations); } return new ArrayList<>(goodLocations.keySet()); } StorageLocationChecker(Configuration conf, Timer timer); List<StorageLocation> check(
      final Configuration conf,
      final Collection<StorageLocation> dataDirs); void shutdownAndWait(int gracePeriod, TimeUnit timeUnit); static final Logger LOG; }
@Test(timeout=30000) public void testAllLocationsHealthy() throws Exception { final List<StorageLocation> locations = makeMockLocations(HEALTHY, HEALTHY, HEALTHY); final Configuration conf = new HdfsConfiguration(); conf.setInt(DFS_DATANODE_FAILED_VOLUMES_TOLERATED_KEY, 0); StorageLocationChecker checker = new StorageLocationChecker(conf, new FakeTimer()); List<StorageLocation> filteredLocations = checker.check(conf, locations); assertThat(filteredLocations.size(), is(3)); for (StorageLocation location : locations) { verify(location).check(any(StorageLocation.CheckContext.class)); } }
@Test(timeout=30000) public void testFailedLocationsBelowThreshold() throws Exception { final List<StorageLocation> locations = makeMockLocations(HEALTHY, HEALTHY, FAILED); final Configuration conf = new HdfsConfiguration(); conf.setInt(DFS_DATANODE_FAILED_VOLUMES_TOLERATED_KEY, 1); StorageLocationChecker checker = new StorageLocationChecker(conf, new FakeTimer()); List<StorageLocation> filteredLocations = checker.check(conf, locations); assertThat(filteredLocations.size(), is(2)); }
@Test(timeout=30000) public void testFailedLocationsAboveThreshold() throws Exception { final List<StorageLocation> locations = makeMockLocations(HEALTHY, FAILED, FAILED); final Configuration conf = new HdfsConfiguration(); conf.setInt(DFS_DATANODE_FAILED_VOLUMES_TOLERATED_KEY, 1); thrown.expect(IOException.class); thrown.expectMessage("Too many failed volumes - current valid volumes: 1," + " volumes configured: 3, volumes failed: 2, volume failures" + " tolerated: 1"); StorageLocationChecker checker = new StorageLocationChecker(conf, new FakeTimer()); checker.check(conf, locations); }
@Test(timeout=30000) public void testBadConfiguration() throws Exception { final List<StorageLocation> locations = makeMockLocations(HEALTHY, HEALTHY, HEALTHY); final Configuration conf = new HdfsConfiguration(); conf.setInt(DFS_DATANODE_FAILED_VOLUMES_TOLERATED_KEY, 3); thrown.expect(HadoopIllegalArgumentException.class); thrown.expectMessage("Invalid value configured"); StorageLocationChecker checker = new StorageLocationChecker(conf, new FakeTimer()); checker.check(conf, locations); }
@Test public void testInvalidConfigurationValues() throws Exception { final List<StorageLocation> locations = makeMockLocations(HEALTHY, HEALTHY, HEALTHY); Configuration conf = new HdfsConfiguration(); conf.setInt(DFS_DATANODE_FAILED_VOLUMES_TOLERATED_KEY, 4); intercept(HadoopIllegalArgumentException.class, "Invalid value configured for dfs.datanode.failed.volumes.tolerated" + " - 4. Value configured is >= to the " + "number of configured volumes (3).", () -> new StorageLocationChecker(conf, new FakeTimer()).check(conf, locations)); conf.unset(DFS_DATANODE_FAILED_VOLUMES_TOLERATED_KEY); conf.setInt(DFS_DATANODE_DISK_CHECK_TIMEOUT_KEY, 0); intercept(HadoopIllegalArgumentException.class, "Invalid value configured for dfs.datanode.disk.check.timeout" + " - 0 (should be > 0)", () -> new StorageLocationChecker(conf, new FakeTimer())); conf.unset(DFS_DATANODE_DISK_CHECK_TIMEOUT_KEY); conf.setInt(DFS_DATANODE_FAILED_VOLUMES_TOLERATED_KEY, -2); intercept(HadoopIllegalArgumentException.class, "Invalid value configured for dfs.datanode.failed.volumes.tolerated" + " - -2 should be greater than or equal to -1", () -> new StorageLocationChecker(conf, new FakeTimer())); }