HDDS-12080: delete irrelevant RATIS/THREE pipeline's raft logs in case of DEAD datanode state

dev-support/rat/rat-exclusions.txt

@@ -47,6 +47,7 @@ src/test/resources/additionalfields.container
 src/test/resources/incorrect.checksum.container
 src/test/resources/incorrect.container
 src/test/resources/test.db.ini
+src/test/resources/metadata/**
 
 # hadoop-hdds/docs
 **/themes/ozonedoc/**

hadoop-hdds/container-service/dev-support/findbugsExcludeFile.xml

@@ -81,4 +81,16 @@
     <Class name="org.apache.hadoop.ozone.container.ozoneimpl.TestBackgroundContainerMetadataScanner"/>
     <Bug pattern="RU_INVOKE_RUN, RV_RETURN_VALUE_IGNORED_NO_SIDE_EFFECT" />
   </Match>
+  <Match>
+    <Bug pattern="NP_NULL_ON_SOME_PATH_FROM_RETURN_VALUE">
+      <Class name="org.apache.hadoop.ozone.container.ozoneimpl.TestOzoneContainer"/>
+      <Method name="testCleanUpMetadataDirInCaseOfDeadNodeState"/>
+    </Bug>
+  </Match>
+  <Match>
+    <Bug pattern="RV_RETURN_VALUE_IGNORED_NO_SIDE_EFFECT">
+      <Class name="org.apache.hadoop.ozone.container.ozoneimpl.TestOzoneContainer"/>
+      <Method name="testCleanUpMetadataDirInCaseOfDeadNodeState"/>
+    </Bug>
+  </Match>
 </FindBugsFilter>

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/statemachine/DatanodeStateMachine.java

@@ -181,7 +181,7 @@ public DatanodeStateMachine(HddsDatanodeService hddsDatanodeService,
     constructionLock.writeLock().lock();
     try {
       container = new OzoneContainer(hddsDatanodeService, this.datanodeDetails,
-          conf, context, certClient, secretKeyClient);
+          conf, context, certClient, secretKeyClient, connectionManager);
     } finally {
       constructionLock.writeLock().unlock();
     }

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

@@ -21,10 +21,13 @@
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.Maps;
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import org.apache.commons.io.FileUtils;
+import org.apache.hadoop.hdds.HddsUtils;
 import org.apache.hadoop.hdds.conf.ConfigurationSource;
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
 import org.apache.hadoop.hdds.protocol.DatanodeDetails.Port.Name;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ContainerType;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ContainerReplicaProto;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.IncrementalContainerReportProto;
@@ -49,6 +52,7 @@
 import org.apache.hadoop.ozone.container.common.interfaces.Handler;
 import org.apache.hadoop.ozone.container.common.report.IncrementalReportSender;
 import org.apache.hadoop.ozone.container.common.statemachine.DatanodeConfiguration;
+import org.apache.hadoop.ozone.container.common.statemachine.SCMConnectionManager;
 import org.apache.hadoop.ozone.container.common.statemachine.StateContext;
 import org.apache.hadoop.ozone.container.common.transport.server.XceiverServerGrpc;
 import org.apache.hadoop.ozone.container.common.transport.server.XceiverServerSpi;
@@ -73,14 +77,19 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.File;
 import java.io.IOException;
+import java.net.InetSocketAddress;
 import java.time.Duration;
 import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
@@ -135,6 +144,7 @@ public class OzoneContainer {
   private ScheduledExecutorService dbCompactionExecutorService;
 
   private final ContainerMetrics metrics;
+  private final SCMConnectionManager scmConnectionManager;
   private WitnessedContainerMetadataStore witnessedContainerMetadataStore;
 
   enum InitializingStatus {
@@ -151,12 +161,14 @@ enum InitializingStatus {
    * @throws IOException
    */
   public OzoneContainer(HddsDatanodeService hddsDatanodeService,
-      DatanodeDetails datanodeDetails, ConfigurationSource conf,
-      StateContext context, CertificateClient certClient,
-      SecretKeyVerifierClient secretKeyClient) throws IOException {
+                        DatanodeDetails datanodeDetails, ConfigurationSource conf,
+                        StateContext context, CertificateClient certClient,
+                        SecretKeyVerifierClient secretKeyClient,
+                        SCMConnectionManager scmConnectionManager) throws IOException {
     config = conf;
     this.datanodeDetails = datanodeDetails;
     this.context = context;
+    this.scmConnectionManager = scmConnectionManager;
     this.volumeChecker = new StorageVolumeChecker(conf, new Timer(),
         datanodeDetails.threadNamePrefix());
 
@@ -229,6 +241,8 @@ public OzoneContainer(HddsDatanodeService hddsDatanodeService,
      */
     controller = new ContainerController(containerSet, handlers);
 
+    cleanUpRatisMetadataDirectory();
+
     writeChannel = XceiverServerRatis.newXceiverServerRatis(hddsDatanodeService,
         datanodeDetails, config, hddsDispatcher, controller, certClient,
         context);
@@ -300,7 +314,7 @@ public OzoneContainer(HddsDatanodeService hddsDatanodeService,
   public OzoneContainer(
       DatanodeDetails datanodeDetails, ConfigurationSource conf,
       StateContext context) throws IOException {
-    this(null, datanodeDetails, conf, context, null, null);
+    this(null, datanodeDetails, conf, context, null, null, null);
   }
 
   public GrpcTlsConfig getTlsClientConfig() {
@@ -656,4 +670,58 @@ public void compactDb() {
     }
   }
 
+  public DatanodeDetails getDatanodeDetails() {
+    return datanodeDetails;
+  }
+
+  private void cleanUpRatisMetadataDirectory()
+      throws IOException {
+    if (scmConnectionManager != null) {
+      Collection<InetSocketAddress> scmAddressesForDatanodes;
+      try {
+        scmAddressesForDatanodes = HddsUtils.getSCMAddressForDatanodes(config);
+      } catch (IllegalArgumentException e) {
+        LOG.error("Failed to get SCM addresses for datanodes: {}", e.getMessage());
+        return;
+      }
+      for (InetSocketAddress scmAddress : scmAddressesForDatanodes) {
+        scmConnectionManager.addSCMServer(scmAddress, context.getThreadNamePrefix());
+        context.addEndpoint(scmAddress);
+      }
+
+      scmConnectionManager.getValues().stream()
+          .filter(endPoint -> !endPoint.isPassive())
+          .findFirst()
+          .ifPresent(rpcEndPoint -> {
+            try {
+              // Check the previous state of the datanode stored on the SCM side
+              HddsProtos.NodeState nodePreviousState = rpcEndPoint.getEndPoint()
+                  .getNodeState(datanodeDetails.getUuid());
+
+              if (HddsProtos.NodeState.DEAD.equals(nodePreviousState)) {
+                LOG.info("The node previous state is DEAD, let's clean up the RATIS/THREE pipelines");
+                // OK, the node was previously marked as DEAD, let's clean up the
+                // RATIS/THREE pipelines (aka raft-groups)
+                this.getMetaVolumeSet().getVolumeMap().forEach((key, value) ->
+                    Arrays.stream(Objects.requireNonNull(value.getStorageDir()
+                            // don't touch the directory with volume check info
+                            .listFiles((dir, name) -> !name.equals("tmp"))))
+                        .filter(File::isDirectory) // only directories
+                        .forEach(directory -> {
+                          try {
+                            FileUtils.deleteDirectory(directory);
+                            LOG.info("Delete directory: {}", directory.getAbsolutePath());
+                          } catch (IOException e) {
+                            LOG.warn("Failed to delete directory: {}", directory);
+                          }
+                        }));
+              }
+
+            } catch (IOException e) {
+              LOG.error(String.format("Failed to get datanode previous state with SCM: %s", e.getMessage()), e);
+            }
+          });
+    }
+  }
+
 }

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/protocol/StorageContainerDatanodeProtocol.java

@@ -18,6 +18,8 @@
 
 import org.apache.hadoop.hdds.annotation.InterfaceAudience;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos.ExtendedDatanodeDetailsProto;
+import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.NodeStateRequestProto;
+import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.NodeStateResponseProto;
 import org.apache.hadoop.hdds.protocol.proto
     .StorageContainerDatanodeProtocolProtos.PipelineReportsProto;
 import org.apache.hadoop.hdds.protocol.proto
@@ -89,4 +91,6 @@ SCMRegisteredResponseProto register(
       PipelineReportsProto pipelineReports,
       LayoutVersionProto layoutInfo) throws IOException;
 
+  NodeStateResponseProto getNodeState(NodeStateRequestProto request) throws IOException;
+
 }

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/protocolPB/StorageContainerDatanodeProtocolClientSideTranslatorPB.java

@@ -18,9 +18,12 @@
 
 import com.google.protobuf.RpcController;
 import com.google.protobuf.ServiceException;
+import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos
     .ExtendedDatanodeDetailsProto;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.LayoutVersionProto;
+import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.NodeStateRequestProto;
+import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.NodeStateResponseProto;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.PipelineReportsProto;
 import org.apache.hadoop.hdds.protocol.proto
     .StorageContainerDatanodeProtocolProtos.ContainerReportsProto;
@@ -50,6 +53,7 @@
 
 import java.io.Closeable;
 import java.io.IOException;
+import java.util.UUID;
 import java.util.function.Consumer;
 
 /**
@@ -178,4 +182,18 @@ public SCMRegisteredResponseProto register(
         (builder) -> builder.setRegisterRequest(req))
         .getRegisterResponse();
   }
+
+  public HddsProtos.NodeState getNodeState(UUID datanodeUuid) throws IOException {
+    NodeStateRequestProto request = NodeStateRequestProto.newBuilder()
+        .setDatanodeUUID(datanodeUuid.toString())
+        .build();
+    return getNodeState(request).getNodeState();
+
+  }
+
+  @Override
+  public NodeStateResponseProto getNodeState(NodeStateRequestProto request) throws IOException {
+    return submitRequest(Type.NodePreviousState, builder -> builder.setNodeStateRequest(request))
+        .getNodeStateResponse();
+  }
 }

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/protocolPB/StorageContainerDatanodeProtocolServerSideTranslatorPB.java

@@ -118,6 +118,13 @@ public SCMDatanodeResponse processMessage(SCMDatanodeRequest request)
             .setStatus(Status.OK)
             .setRegisterResponse(register(request.getRegisterRequest()))
             .build();
+      case NodePreviousState:
+        return SCMDatanodeResponse.newBuilder()
+            .setCmdType(cmdType)
+            .setStatus(Status.OK)
+            .setNodeStateResponse(
+                impl.getNodeState(request.getNodeStateRequest()))
+            .build();
       default:
         throw new ServiceException("Unknown command type: " + cmdType);
       }

hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/ScmTestMock.java

@@ -26,6 +26,8 @@
     .StorageContainerDatanodeProtocolProtos.ContainerReplicaProto;
 import org.apache.hadoop.hdds.protocol.proto
         .StorageContainerDatanodeProtocolProtos.ContainerReportsProto;
+import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.NodeStateRequestProto;
+import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.NodeStateResponseProto;
 import org.apache.hadoop.hdds.protocol.proto
         .StorageContainerDatanodeProtocolProtos.PipelineReportsProto;
 import org.apache.hadoop.hdds.protocol.proto
@@ -255,6 +257,12 @@ private void sleepIfNeeded() {
                 .SCMRegisteredResponseProto.ErrorCode.success).build();
   }
 
+  @Override
+  public NodeStateResponseProto getNodeState(
+      NodeStateRequestProto request) throws IOException {
+    return null;
+  }
+
   /**
    * Update nodeReport.
    * @param datanodeDetailsProto