RouterFsck { public void fsck() { final long startTime = Time.monotonicNow(); try { String warnMsg = "Now FSCK to DFSRouter is unstable feature. " + "There may be incompatible changes between releases."; LOG.warn(warnMsg); out.println(warnMsg); String msg = "Federated FSCK started by " + UserGroupInformation.getCurrentUser() + " from " + remoteAddress + " at " + new Date(); LOG.info(msg); out.println(msg); StateStoreService stateStore = router.getStateStore(); MembershipStore membership = stateStore.getRegisteredRecordStore(MembershipStore.class); GetNamenodeRegistrationsRequest request = GetNamenodeRegistrationsRequest.newInstance(); GetNamenodeRegistrationsResponse response = membership.getNamenodeRegistrations(request); List<MembershipState> memberships = response.getNamenodeMemberships(); Collections.sort(memberships); for (MembershipState nn : memberships) { if (nn.getState() == FederationNamenodeServiceState.ACTIVE) { try { String webAddress = nn.getWebAddress(); out.write("Checking " + nn + " at " + webAddress + "\n"); remoteFsck(nn); } catch (IOException ioe) { out.println("Cannot query " + nn + ": " + ioe.getMessage() + "\n"); } } } out.println("Federated FSCK ended at " + new Date() + " in " + (Time.monotonicNow() - startTime + " milliseconds")); } catch (Exception e) { String errMsg = "Fsck " + e.getMessage(); LOG.warn(errMsg, e); out.println("Federated FSCK ended at " + new Date() + " in " + (Time.monotonicNow() - startTime + " milliseconds")); out.println(e.getMessage()); out.print("\n\n" + errMsg); } finally { out.close(); } } RouterFsck(Router router, Map<String, String[]> pmap,
                    PrintWriter out, InetAddress remoteAddress); void fsck(); static final Logger LOG; }
@Test public void testFsck() throws Exception { MountTable addEntry = MountTable.newInstance("/testdir", Collections.singletonMap("ns0", "/testdir")); assertTrue(addMountTable(addEntry)); addEntry = MountTable.newInstance("/testdir2", Collections.singletonMap("ns1", "/testdir2")); assertTrue(addMountTable(addEntry)); routerFs.createNewFile(new Path("/testdir/testfile")); routerFs.createNewFile(new Path("/testdir2/testfile2")); routerFs.createNewFile(new Path("/testdir2/testfile3")); routerFs.createNewFile(new Path("/testdir2/testfile4")); try (CloseableHttpClient httpClient = HttpClients.createDefault()) { HttpGet httpGet = new HttpGet("http: ":" + webAddress.getPort() + "/fsck"); try (CloseableHttpResponse httpResponse = httpClient.execute(httpGet)) { assertEquals(HttpStatus.SC_OK, httpResponse.getStatusLine().getStatusCode()); String out = EntityUtils.toString( httpResponse.getEntity(), StandardCharsets.UTF_8); LOG.info(out); assertTrue(out.contains("Federated FSCK started")); assertTrue(out.contains("Total files:\t1")); assertTrue(out.contains("Total files:\t3")); assertTrue(out.contains("Federated FSCK ended")); int nnCount = 0; for (MembershipState nn : memberships) { if (nn.getState() == FederationNamenodeServiceState.ACTIVE) { assertTrue(out.contains( "Checking " + nn + " at " + nn.getWebAddress() + "\n")); nnCount++; } } assertEquals(2, nnCount); } httpGet = new HttpGet("http: ":" + webAddress.getPort() + "/fsck?path=/testdir"); try (CloseableHttpResponse httpResponse = httpClient.execute(httpGet)) { assertEquals(HttpStatus.SC_OK, httpResponse.getStatusLine().getStatusCode()); String out = EntityUtils.toString( httpResponse.getEntity(), StandardCharsets.UTF_8); LOG.info(out); assertTrue(out.contains("Federated FSCK started")); assertTrue(out.contains("Total files:\t1")); assertFalse(out.contains("Total files:\t3")); assertTrue(out.contains("Federated FSCK ended")); int nnCount = 0; for (MembershipState nn : memberships) { if (nn.getState() == FederationNamenodeServiceState.ACTIVE) { assertTrue(out.contains( "Checking " + nn + " at " + nn.getWebAddress() + "\n")); nnCount++; } } assertEquals(2, nnCount); } } }