[HDDS-1201] Reporting corrupted containers info to SCM

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/interfaces/Container.java

@@ -154,6 +154,8 @@ ContainerReplicaProto getContainerReport()
 
   /**
    * check and report the structural integrity of the container.
+   * @return true if the integrity checks pass
+   *         false otherwise
    */
-  void check() throws StorageContainerException;
+  boolean check();
 }

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/states/endpoint/HeartbeatEndpointTask.java

@@ -135,8 +135,12 @@ public EndpointStateMachine.EndPointStates call() throws Exception {
       addReports(requestBuilder);
       addContainerActions(requestBuilder);
       addPipelineActions(requestBuilder);
+      SCMHeartbeatRequestProto request = requestBuilder.build();
+      if (LOG.isDebugEnabled()) {
+        LOG.debug("Sending heartbeat message :: {}", request.toString());
+      }
       SCMHeartbeatResponseProto reponse = rpcEndpoint.getEndPoint()
-          .sendHeartbeat(requestBuilder.build());
+          .sendHeartbeat(request);
       processResponse(reponse, datanodeDetailsProto);
       rpcEndpoint.setLastSuccessfulHeartbeat(ZonedDateTime.now());
       rpcEndpoint.zeroMissedCount();

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueContainer.java

@@ -648,7 +648,7 @@ public File getContainerDBFile() {
   /**
    * run integrity checks on the Container metadata.
    */
-  public void check() throws StorageContainerException {
+  public boolean check() {
     ContainerCheckLevel level = ContainerCheckLevel.NO_CHECK;
     long containerId = containerData.getContainerID();
 
@@ -671,14 +671,12 @@ public void check() throws StorageContainerException {
           containerData.getState());
       break;
     default:
-      throw new StorageContainerException(
-          "Invalid Container state found for Container : " + containerData
-              .getContainerID(), INVALID_CONTAINER_STATE);
+      break;
     }
 
     if (level == ContainerCheckLevel.NO_CHECK) {
       LOG.debug("Skipping integrity checks for Container Id : {}", containerId);
-      return;
+      return true;
     }
 
     KeyValueContainerCheck checker =
@@ -687,17 +685,11 @@ public void check() throws StorageContainerException {
 
     switch (level) {
     case FAST_CHECK:
-      checker.fastCheck();
-      break;
+      return checker.fastCheck();
     case FULL_CHECK:
-      checker.fullCheck();
-      break;
-    case NO_CHECK:
-      LOG.debug("Skipping integrity checks for Container Id : {}", containerId);
-      break;
+      return checker.fullCheck();
     default:
-      // we should not be here at all, scuttle the ship!
-      Preconditions.checkNotNull(0, "Invalid Containercheck level");
+      return true;
     }
   }
 

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueContainerCheck.java

@@ -72,37 +72,22 @@ public KeyValueContainerCheck(String metadataPath, Configuration conf,
    * These checks do not look inside the metadata files.
    * Applicable for OPEN containers.
    *
-   * @return true : corruption detected, false : no corruption.
+   * @return true : integrity checks pass, false : otherwise.
    */
   public boolean fastCheck() {
-    boolean corruption = false;
+    LOG.info("Running basic checks for container {};", containerID);
+    boolean valid = false;
     try {
-      basicChecks();
+      loadContainerData();
+      checkLayout();
+      checkContainerFile();
+      valid = true;
 
     } catch (IOException e) {
       handleCorruption(e);
-      corruption = true;
     }
 
-    return corruption;
-  }
-
-  /**
-   * Checks :
-   * 1. check directory layout
-   * 2. check container file
-   *
-   * @return void
-   */
-
-  private void basicChecks() throws IOException {
-
-    LOG.trace("Running basic checks for container {};", containerID);
-
-    loadContainerData();
-
-    checkLayout();
-    checkContainerFile();
+    return valid;
   }
 
   /**
@@ -114,21 +99,22 @@ private void basicChecks() throws IOException {
    * <p>
    * fullCheck is a superset of fastCheck
    *
-   * @return true : corruption detected, false : no corruption.
+   * @return true : integrity checks pass, false : otherwise.
    */
   public boolean fullCheck() {
-    boolean corruption = false;
+    boolean valid = false;
 
     try {
-      basicChecks();
-      checkBlockDB();
-
+      valid = fastCheck();
+      if (valid) {
+        checkBlockDB();
+      }
     } catch (IOException e) {
       handleCorruption(e);
-      corruption = true;
+      valid = false;
     }
 
-    return corruption;
+    return valid;
   }
 
   /**

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueHandler.java

@@ -895,8 +895,17 @@ public void markContainerForClose(Container container)
   public void markContainerUnhealthy(Container container)
       throws IOException {
     if (container.getContainerState() != State.UNHEALTHY) {
-      container.markContainerUnhealthy();
-      sendICR(container);
+      try {
+        container.markContainerUnhealthy();
+      } catch (IOException ex) {
+        // explicitly catch IOException here since the this operation
+        // will fail if the Rocksdb metadata is corrupted.
+        long id = container.getContainerData().getContainerID();
+        LOG.warn("Unexpected error while marking container "
+                +id+ " as unhealthy", ex);
+      } finally {
+        sendICR(container);
+      }
     }
   }
 

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/ozoneimpl/ContainerController.java

@@ -73,6 +73,18 @@ public void markContainerForClose(final long containerId)
     }
   }
 
+  /**
+   * Marks the container as UNHEALTHY.
+   *
+   * @param containerId Id of the container to update
+   * @throws IOException in case of exception
+   */
+  public void markContainerUnhealthy(final long containerId)
+          throws IOException {
+    Container container = containerSet.getContainer(containerId);
+    getHandler(container).markContainerUnhealthy(container);
+  }
+
   /**
    * Returns the container report.
    *

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/ozoneimpl/ContainerScrubber.java

@@ -18,13 +18,14 @@
 
 package org.apache.hadoop.ozone.container.ozoneimpl;
 
+import com.google.common.annotations.VisibleForTesting;
 import org.apache.commons.net.ntp.TimeStamp;
 import org.apache.hadoop.hdds.conf.OzoneConfiguration;
-import org.apache.hadoop.hdds.scm.container.common.helpers.StorageContainerException;
 import org.apache.hadoop.ozone.container.common.interfaces.Container;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.IOException;
 import java.util.Iterator;
 
 /**
@@ -56,11 +57,7 @@ public ContainerScrubber(OzoneConfiguration conf,
     LOG.info("Background ContainerScrubber starting up");
     while (true) {
 
-      try {
-        scrub();
-      } catch (StorageContainerException e) {
-        LOG.error("Scrubber encountered StorageContainerException.");
-      }
+      scrub();
 
       if (this.halt) {
         break; // stop and exit if requested
@@ -129,27 +126,33 @@ private void throttleScrubber(TimeStamp startTime) {
     }
   }
 
-  private void scrub() throws StorageContainerException {
+  private void scrub() {
     Iterator<Container> containerIt = controller.getContainers();
     long count = 0;
 
     while (containerIt.hasNext() && !halt) {
       TimeStamp startTime = new TimeStamp(System.currentTimeMillis());
       Container container = containerIt.next();
-
       try {
-        container.check();
-      } catch (StorageContainerException e) {
-        LOG.error("Error unexpected exception {} for Container {}", e,
-            container.getContainerData().getContainerID());
-        container.markContainerUnhealthy();
-        // XXX Action required here
+        scrub(container);
+      } catch (IOException e) {
+        LOG.info("Unexpected error while scrubbing container {}",
+                container.getContainerData().getContainerID());
       }
+
       count++;
 
       throttleScrubber(startTime);
     }
 
     LOG.debug("iterator ran integrity checks on {} containers", count);
   }
+
+  @VisibleForTesting
+  public void scrub(Container container) throws IOException {
+    if (!container.check()) {
+      controller.markContainerUnhealthy(
+              container.getContainerData().getContainerID());
+    }
+  }
 }

hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/TestKeyValueContainerCheck.java

@@ -55,7 +55,6 @@
 
 import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_METADATA_STORE_IMPL_LEVELDB;
 import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_METADATA_STORE_IMPL_ROCKSDB;
-import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
 
 /**
@@ -101,7 +100,7 @@ public TestKeyValueContainerCheck(String metadataImpl) {
     int deletedBlocks = 1;
     int normalBlocks = 3;
     int chunksPerBlock = 4;
-    boolean corruption = false;
+    boolean valid = false;
 
     // test Closed Container
     createContainerWithBlocks(containerID, normalBlocks, deletedBlocks, 65536,
@@ -115,14 +114,14 @@ public TestKeyValueContainerCheck(String metadataImpl) {
             containerID);
 
     // first run checks on a Open Container
-    corruption = kvCheck.fastCheck();
-    assertFalse(corruption);
+    valid = kvCheck.fastCheck();
+    assertTrue(valid);
 
     container.close();
 
     // next run checks on a Closed Container
-    corruption = kvCheck.fullCheck();
-    assertFalse(corruption);
+    valid = kvCheck.fullCheck();
+    assertTrue(valid);
   }
 
   /**

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/IncrementalContainerReportHandler.java

@@ -47,6 +47,8 @@ public IncrementalContainerReportHandler(
   @Override
   public void onMessage(final IncrementalContainerReportFromDatanode report,
                         final EventPublisher publisher) {
+    LOG.debug("Processing incremental container report from data node {}",
+            report.getDatanodeDetails().getUuid());
 
     for (ContainerReplicaProto replicaProto :
         report.getReport().getReportList()) {