hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/node/TestNodeManager.java
@Test
  public void testScmNodeReportUpdate() throws IOException,
      InterruptedException, TimeoutException {
    OzoneConfiguration conf = getConf();
    final int heartbeatCount = 5;
    final int nodeCount = 1;
    final int interval = 100;

    conf.setTimeDuration(OZONE_SCM_HEARTBEAT_PROCESS_INTERVAL, interval,
        MILLISECONDS);
    conf.setTimeDuration(OZONE_SCM_HEARTBEAT_INTERVAL, 1, SECONDS);
    conf.setTimeDuration(OZONE_SCM_STALENODE_INTERVAL, 3, SECONDS);
    conf.setTimeDuration(OZONE_SCM_DEADNODE_INTERVAL, 6, SECONDS);

    try (SCMNodeManager nodeManager = createNodeManager(conf)) {
      DatanodeDetails datanodeDetails = TestUtils.getDatanodeDetails(
          nodeManager);
      final long capacity = 2000;
      final long usedPerHeartbeat = 100;

      for (int x = 0; x < heartbeatCount; x++) {
        SCMNodeReport.Builder nrb = SCMNodeReport.newBuilder();
        SCMStorageReport.Builder srb = SCMStorageReport.newBuilder();
        srb.setStorageUuid(UUID.randomUUID().toString());
        srb.setCapacity(capacity).setScmUsed(x * usedPerHeartbeat)
            .setRemaining(capacity - x * usedPerHeartbeat).build();
        nrb.addStorageReport(srb);

        nodeManager.sendHeartbeat(
            datanodeDetails.getProtoBufMessage(), nrb.build(), reportState);
        Thread.sleep(100);
      }

      final long expectedScmUsed = usedPerHeartbeat * (heartbeatCount - 1);
      final long expectedRemaining = capacity - expectedScmUsed;

      GenericTestUtils.waitFor(
          () -> nodeManager.getStats().getScmUsed().get() == expectedScmUsed,
          100, 4 * 1000);

      long foundCapacity = nodeManager.getStats().getCapacity().get();
      assertEquals(capacity, foundCapacity);

      long foundScmUsed = nodeManager.getStats().getScmUsed().get();
      assertEquals(expectedScmUsed, foundScmUsed);

      long foundRemaining = nodeManager.getStats().getRemaining().get();
      assertEquals(expectedRemaining, foundRemaining);

      // Test NodeManager#getNodeStats
      assertEquals(nodeCount, nodeManager.getNodeStats().size());
      long nodeCapacity = nodeManager.getNodeStat(datanodeDetails).get()
          .getCapacity().get();
      assertEquals(capacity, nodeCapacity);

      foundScmUsed = nodeManager.getNodeStat(datanodeDetails).get().getScmUsed()
          .get();
      assertEquals(expectedScmUsed, foundScmUsed);

      foundRemaining = nodeManager.getNodeStat(datanodeDetails).get()
          .getRemaining().get();
      assertEquals(expectedRemaining, foundRemaining);

      // Compare the result from
      // NodeManager#getNodeStats and NodeManager#getNodeStat
      SCMNodeStat stat1 = nodeManager.getNodeStats().
          get(datanodeDetails.getUuid());
      SCMNodeStat stat2 = nodeManager.getNodeStat(datanodeDetails).get();
      assertEquals(stat1, stat2);

      // Wait up to 4s so that the node becomes stale
      // Verify the usage info should be unchanged.
      GenericTestUtils.waitFor(
          () -> nodeManager.getNodeCount(STALE) == 1, 100,
          4 * 1000);
      assertEquals(nodeCount, nodeManager.getNodeStats().size());

      foundCapacity = nodeManager.getNodeStat(datanodeDetails).get()
          .getCapacity().get();
      assertEquals(capacity, foundCapacity);
      foundScmUsed = nodeManager.getNodeStat(datanodeDetails).get()
          .getScmUsed().get();
      assertEquals(expectedScmUsed, foundScmUsed);

      foundRemaining = nodeManager.getNodeStat(datanodeDetails).get().
          getRemaining().get();
      assertEquals(expectedRemaining, foundRemaining);

      // Wait up to 4 more seconds so the node becomes dead
      // Verify usage info should be updated.
      GenericTestUtils.waitFor(
          () -> nodeManager.getNodeCount(DEAD) == 1, 100,
          4 * 1000);

      assertEquals(0, nodeManager.getNodeStats().size());
      foundCapacity = nodeManager.getStats().getCapacity().get();
      assertEquals(0, foundCapacity);

      foundScmUsed = nodeManager.getStats().getScmUsed().get();
      assertEquals(0, foundScmUsed);

      foundRemaining = nodeManager.getStats().getRemaining().get();
      assertEquals(0, foundRemaining);

      // Send a new report to bring the dead node back to healthy
      SCMNodeReport.Builder nrb = SCMNodeReport.newBuilder();
      SCMStorageReport.Builder srb = SCMStorageReport.newBuilder();
      srb.setStorageUuid(UUID.randomUUID().toString());
      srb.setCapacity(capacity).setScmUsed(expectedScmUsed)
          .setRemaining(expectedRemaining).build();
      nrb.addStorageReport(srb);
      nodeManager.sendHeartbeat(
          datanodeDetails.getProtoBufMessage(), nrb.build(), reportState);

      // Wait up to 5 seconds so that the dead node becomes healthy
      // Verify usage info should be updated.
      GenericTestUtils.waitFor(
          () -> nodeManager.getNodeCount(HEALTHY) == 1,
          100, 5 * 1000);
      GenericTestUtils.waitFor(
          () -> nodeManager.getStats().getScmUsed().get() == expectedScmUsed,
          100, 4 * 1000);
      assertEquals(nodeCount, nodeManager.getNodeStats().size());
      foundCapacity = nodeManager.getNodeStat(datanodeDetails).get()
          .getCapacity().get();
      assertEquals(capacity, foundCapacity);
      foundScmUsed = nodeManager.getNodeStat(datanodeDetails).get().getScmUsed()
          .get();
      assertEquals(expectedScmUsed, foundScmUsed);
      foundRemaining = nodeManager.getNodeStat(datanodeDetails).get()
          .getRemaining().get();
      assertEquals(expectedRemaining, foundRemaining);
    }
  }
/**
   * Test single node stat update based on nodereport from different heartbeat
   * status (healthy, stale and dead).
   * @throws IOException
   * @throws InterruptedException
   * @throws TimeoutException
   */
Test single node stat update based on nodereport from different heartbeat
status (healthy, stale and dead).
Test single node stat update based on nodereport from different heartbeat status ( healthy stale and dead ) .
@Test
  public void testScmNodeReportUpdate() throws IOException,
      InterruptedException, TimeoutException {
    OzoneConfiguration conf = getConf();
    final int heartbeatCount = 5;
    final int nodeCount = 1;
    final int interval = 100;
    conf.setTimeDuration(OZONE_SCM_HEARTBEAT_PROCESS_INTERVAL, interval,
        MILLISECONDS);
    conf.setTimeDuration(OZONE_SCM_HEARTBEAT_INTERVAL, 1, SECONDS);
    conf.setTimeDuration(OZONE_SCM_STALENODE_INTERVAL, 3, SECONDS);
    conf.setTimeDuration(OZONE_SCM_DEADNODE_INTERVAL, 6, SECONDS);
    try (SCMNodeManager nodeManager = createNodeManager(conf)) {
      DatanodeDetails datanodeDetails = TestUtils.getDatanodeDetails(
          nodeManager);
      final long capacity = 2000;
      final long usedPerHeartbeat = 100;
      for (int x = 0; x < heartbeatCount; x++) {
        SCMNodeReport.Builder nrb = SCMNodeReport.newBuilder();
        SCMStorageReport.Builder srb = SCMStorageReport.newBuilder();
        srb.setStorageUuid(UUID.randomUUID().toString());
        srb.setCapacity(capacity).setScmUsed(x * usedPerHeartbeat)
            .setRemaining(capacity - x * usedPerHeartbeat).build();
        nrb.addStorageReport(srb);
        nodeManager.sendHeartbeat(
            datanodeDetails.getProtoBufMessage(), nrb.build(), reportState);
        Thread.sleep(100);
      }
      final long expectedScmUsed = usedPerHeartbeat * (heartbeatCount - 1);
      final long expectedRemaining = capacity - expectedScmUsed;
      GenericTestUtils.waitFor(
          () -> nodeManager.getStats().getScmUsed().get() == expectedScmUsed,
          100, 4 * 1000);
      long foundCapacity = nodeManager.getStats().getCapacity().get();
      assertEquals(capacity, foundCapacity);
      long foundScmUsed = nodeManager.getStats().getScmUsed().get();
      assertEquals(expectedScmUsed, foundScmUsed);
      long foundRemaining = nodeManager.getStats().getRemaining().get();
      assertEquals(expectedRemaining, foundRemaining);
      assertEquals(nodeCount, nodeManager.getNodeStats().size());
      long nodeCapacity = nodeManager.getNodeStat(datanodeDetails).get()
          .getCapacity().get();
      assertEquals(capacity, nodeCapacity);
      foundScmUsed = nodeManager.getNodeStat(datanodeDetails).get().getScmUsed()
          .get();
      assertEquals(expectedScmUsed, foundScmUsed);
      foundRemaining = nodeManager.getNodeStat(datanodeDetails).get()
          .getRemaining().get();
      assertEquals(expectedRemaining, foundRemaining);
      SCMNodeStat stat1 = nodeManager.getNodeStats().
          get(datanodeDetails.getUuid());
      SCMNodeStat stat2 = nodeManager.getNodeStat(datanodeDetails).get();
      assertEquals(stat1, stat2);
      GenericTestUtils.waitFor(
          () -> nodeManager.getNodeCount(STALE) == 1, 100,
          4 * 1000);
      assertEquals(nodeCount, nodeManager.getNodeStats().size());
      foundCapacity = nodeManager.getNodeStat(datanodeDetails).get()
          .getCapacity().get();
      assertEquals(capacity, foundCapacity);
      foundScmUsed = nodeManager.getNodeStat(datanodeDetails).get()
          .getScmUsed().get();
      assertEquals(expectedScmUsed, foundScmUsed);
      foundRemaining = nodeManager.getNodeStat(datanodeDetails).get().
          getRemaining().get();
      assertEquals(expectedRemaining, foundRemaining);
      GenericTestUtils.waitFor(
          () -> nodeManager.getNodeCount(DEAD) == 1, 100,
          4 * 1000);
      assertEquals(0, nodeManager.getNodeStats().size());
      foundCapacity = nodeManager.getStats().getCapacity().get();
      assertEquals(0, foundCapacity);
      foundScmUsed = nodeManager.getStats().getScmUsed().get();
      assertEquals(0, foundScmUsed);
      foundRemaining = nodeManager.getStats().getRemaining().get();
      assertEquals(0, foundRemaining);
      SCMNodeReport.Builder nrb = SCMNodeReport.newBuilder();
      SCMStorageReport.Builder srb = SCMStorageReport.newBuilder();
      srb.setStorageUuid(UUID.randomUUID().toString());
      srb.setCapacity(capacity).setScmUsed(expectedScmUsed)
          .setRemaining(expectedRemaining).build();
      nrb.addStorageReport(srb);
      nodeManager.sendHeartbeat(
          datanodeDetails.getProtoBufMessage(), nrb.build(), reportState);
      GenericTestUtils.waitFor(
          () -> nodeManager.getNodeCount(HEALTHY) == 1,
          100, 5 * 1000);
      GenericTestUtils.waitFor(
          () -> nodeManager.getStats().getScmUsed().get() == expectedScmUsed,
          100, 4 * 1000);
      assertEquals(nodeCount, nodeManager.getNodeStats().size());
      foundCapacity = nodeManager.getNodeStat(datanodeDetails).get()
          .getCapacity().get();
      assertEquals(capacity, foundCapacity);
      foundScmUsed = nodeManager.getNodeStat(datanodeDetails).get().getScmUsed()
          .get();
      assertEquals(expectedScmUsed, foundScmUsed);
      foundRemaining = nodeManager.getNodeStat(datanodeDetails).get()
          .getRemaining().get();
      assertEquals(expectedRemaining, foundRemaining);
    }
  }
@ Test public void testScmNodeReportUpdate ( ) throws IOException , InterruptedException , TimeoutException { OzoneConfiguration conf = getConf ( ) ; final int heartbeatCount = 5 ; final int nodeCount = 1 ; final int interval = 100 ; conf . setTimeDuration ( OZONE_SCM_HEARTBEAT_PROCESS_INTERVAL , interval , MILLISECONDS ) ; conf . setTimeDuration ( OZONE_SCM_HEARTBEAT_INTERVAL , 1 , SECONDS ) ; conf . setTimeDuration ( OZONE_SCM_STALENODE_INTERVAL , 3 , SECONDS ) ; conf . setTimeDuration ( OZONE_SCM_DEADNODE_INTERVAL , 6 , SECONDS ) ; try ( SCMNodeManager nodeManager = createNodeManager ( conf ) ) { DatanodeDetails datanodeDetails = TestUtils . getDatanodeDetails ( nodeManager ) ; final long capacity = 2000 ; final long usedPerHeartbeat = 100 ; for ( int x = 0 ; x < heartbeatCount ; x ++ ) { SCMNodeReport . Builder nrb = SCMNodeReport . newBuilder ( ) ; SCMStorageReport . Builder srb = SCMStorageReport . newBuilder ( ) ; srb . setStorageUuid ( UUID . randomUUID ( ) . toString ( ) ) ; srb . setCapacity ( capacity ) . setScmUsed ( x * usedPerHeartbeat ) . setRemaining ( capacity - x * usedPerHeartbeat ) . build ( ) ; nrb . addStorageReport ( srb ) ; nodeManager . sendHeartbeat ( datanodeDetails . getProtoBufMessage ( ) , nrb . build ( ) , reportState ) ; Thread . sleep ( 100 ) ; } final long expectedScmUsed = usedPerHeartbeat * ( heartbeatCount - 1 ) ; final long expectedRemaining = capacity - expectedScmUsed ; GenericTestUtils . waitFor ( ( ) -> nodeManager . getStats ( ) . getScmUsed ( ) . get ( ) == expectedScmUsed , 100 , 4 * 1000 ) ; long foundCapacity = nodeManager . getStats ( ) . getCapacity ( ) . get ( ) ; assertEquals ( capacity , foundCapacity ) ; long foundScmUsed = nodeManager . getStats ( ) . getScmUsed ( ) . get ( ) ; assertEquals ( expectedScmUsed , foundScmUsed ) ; long foundRemaining = nodeManager . getStats ( ) . getRemaining ( ) . get ( ) ; assertEquals ( expectedRemaining , foundRemaining ) ; assertEquals ( nodeCount , nodeManager . getNodeStats ( ) . size ( ) ) ; long nodeCapacity = nodeManager . getNodeStat ( datanodeDetails ) . get ( ) . getCapacity ( ) . get ( ) ; assertEquals ( capacity , nodeCapacity ) ; foundScmUsed = nodeManager . getNodeStat ( datanodeDetails ) . get ( ) . getScmUsed ( ) . get ( ) ; assertEquals ( expectedScmUsed , foundScmUsed ) ; foundRemaining = nodeManager . getNodeStat ( datanodeDetails ) . get ( ) . getRemaining ( ) . get ( ) ; assertEquals ( expectedRemaining , foundRemaining ) ; SCMNodeStat stat1 = nodeManager . getNodeStats ( ) . get ( datanodeDetails . getUuid ( ) ) ; SCMNodeStat stat2 = nodeManager . getNodeStat ( datanodeDetails ) . get ( ) ; assertEquals ( stat1 , stat2 ) ; GenericTestUtils . waitFor ( ( ) -> nodeManager . getNodeCount ( STALE ) == 1 , 100 , 4 * 1000 ) ; assertEquals ( nodeCount , nodeManager . getNodeStats ( ) . size ( ) ) ; foundCapacity = nodeManager . getNodeStat ( datanodeDetails ) . get ( ) . getCapacity ( ) . get ( ) ; assertEquals ( capacity , foundCapacity ) ; foundScmUsed = nodeManager . getNodeStat ( datanodeDetails ) . get ( ) . getScmUsed ( ) . get ( ) ; assertEquals ( expectedScmUsed , foundScmUsed ) ; foundRemaining = nodeManager . getNodeStat ( datanodeDetails ) . get ( ) . getRemaining ( ) . get ( ) ; assertEquals ( expectedRemaining , foundRemaining ) ; GenericTestUtils . waitFor ( ( ) -> nodeManager . getNodeCount ( DEAD ) == 1 , 100 , 4 * 1000 ) ; assertEquals ( 0 , nodeManager . getNodeStats ( ) . size ( ) ) ; foundCapacity = nodeManager . getStats ( ) . getCapacity ( ) . get ( ) ; assertEquals ( 0 , foundCapacity ) ; foundScmUsed = nodeManager . getStats ( ) . getScmUsed ( ) . get ( ) ; assertEquals ( 0 , foundScmUsed ) ; foundRemaining = nodeManager . getStats ( ) . getRemaining ( ) . get ( ) ; assertEquals ( 0 , foundRemaining ) ; SCMNodeReport . Builder nrb = SCMNodeReport . newBuilder ( ) ; SCMStorageReport . Builder srb = SCMStorageReport . newBuilder ( ) ; srb . setStorageUuid ( UUID . randomUUID ( ) . toString ( ) ) ; srb . setCapacity ( capacity ) . setScmUsed ( expectedScmUsed ) . setRemaining ( expectedRemaining ) . build ( ) ; nrb . addStorageReport ( srb ) ; nodeManager . sendHeartbeat ( datanodeDetails . getProtoBufMessage ( ) , nrb . build ( ) , reportState ) ; GenericTestUtils . waitFor ( ( ) -> nodeManager . getNodeCount ( HEALTHY ) == 1 , 100 , 5 * 1000 ) ; GenericTestUtils . waitFor ( ( ) -> nodeManager . getStats ( ) . getScmUsed ( ) . get ( ) == expectedScmUsed , 100 , 4 * 1000 ) ; assertEquals ( nodeCount , nodeManager . getNodeStats ( ) . size ( ) ) ; foundCapacity = nodeManager . getNodeStat ( datanodeDetails ) . get ( ) . getCapacity ( ) . get ( ) ; assertEquals ( capacity , foundCapacity ) ; foundScmUsed = nodeManager . getNodeStat ( datanodeDetails ) . get ( ) . getScmUsed ( ) . get ( ) ; assertEquals ( expectedScmUsed , foundScmUsed ) ; foundRemaining = nodeManager . getNodeStat ( datanodeDetails ) . get ( ) . getRemaining ( ) . get ( ) ; assertEquals ( expectedRemaining , foundRemaining ) ; } }
Test single node stat update based on nodereport from different heartbeat
 status (healthy, stale and dead).
Test single node stat update based on nodereport from different heartbeat status ( healthy stale and dead ) .
// Test NodeManager#getNodeStats // Compare the result from // NodeManager#getNodeStats and NodeManager#getNodeStat // Wait up to 4s so that the node becomes stale // Verify the usage info should be unchanged. // Wait up to 4 more seconds so the node becomes dead // Verify usage info should be updated. // Send a new report to bring the dead node back to healthy // Wait up to 5 seconds so that the dead node becomes healthy // Verify usage info should be updated.