HDDS-4404. Datanode can go OOM when a Recon or SCM Server is very slow in processing reports

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

@@ -39,8 +39,13 @@
 import com.google.common.annotations.VisibleForTesting;
 import org.apache.hadoop.hdds.conf.ConfigurationSource;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.CommandStatus.Status;
+import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.CommandStatusReportsProto;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ContainerAction;
+import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.ContainerReportsProto;
+import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.IncrementalContainerReportProto;
+import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.NodeReportProto;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.PipelineAction;
+import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.PipelineReportsProto;
 import org.apache.hadoop.hdds.protocol.proto.StorageContainerDatanodeProtocolProtos.SCMCommandProto;
 import org.apache.hadoop.ozone.container.common.states.DatanodeState;
 import org.apache.hadoop.ozone.container.common.states.datanode.InitDatanodeState;
@@ -72,6 +77,11 @@ public class StateContext {
   private final AtomicLong stateExecutionCount;
   private final ConfigurationSource conf;
   private final Set<InetSocketAddress> endpoints;
+  // Only keeps the latest Container/Node/PipelineReport
+  private GeneratedMessage containerReport;
+  private GeneratedMessage nodeReport;
+  private GeneratedMessage pipelineReport;
+  // CommandStatusReport and IncrementalContainerReport queued in the map below
   private final Map<InetSocketAddress, List<GeneratedMessage>> reports;
   private final Map<InetSocketAddress, Queue<ContainerAction>> containerActions;
   private final Map<InetSocketAddress, Queue<PipelineAction>> pipelineActions;
@@ -80,6 +90,17 @@ public class StateContext {
   private boolean shutdownGracefully = false;
   private final AtomicLong threadPoolNotAvailableCount;
 
+  private static final String containerReportsProtoName =
+      ContainerReportsProto.getDescriptor().getFullName();
+  private static final String nodeReportProtoName =
+      NodeReportProto.getDescriptor().getFullName();
+  private static final String pipelineReportsProtoName =
+      PipelineReportsProto.getDescriptor().getFullName();
+  private static final String commandStatusReportsProtoName =
+      CommandStatusReportsProto.getDescriptor().getFullName();
+  private static final String incrementalContainerReportProtoName =
+      IncrementalContainerReportProto.getDescriptor().getFullName();
+
   /**
    * Starting with a 2 sec heartbeat frequency which will be updated to the
    * real HB frequency after scm registration. With this method the
@@ -103,6 +124,12 @@ public StateContext(ConfigurationSource conf,
     commandQueue = new LinkedList<>();
     cmdStatusMap = new ConcurrentHashMap<>();
     reports = new HashMap<>();
+    // TODO: Even better, is there a way to initialize those as
+    //  empty GeneratedMessage? In protobuf 3 there is Empty.Builder, not in 2?
+    containerReport = null;
+    nodeReport = null;
+    pipelineReport = null;
+
     endpoints = new HashSet<>();
     containerActions = new HashMap<>();
     pipelineActions = new HashMap<>();
@@ -190,16 +217,32 @@ void setShutdownGracefully() {
   public boolean getShutdownOnError() {
     return shutdownOnError;
   }
+
   /**
    * Adds the report to report queue.
    *
    * @param report report to be added
    */
   public void addReport(GeneratedMessage report) {
     if (report != null) {
-      synchronized (reports) {
-        for (InetSocketAddress endpoint : endpoints) {
-          reports.get(endpoint).add(report);
+      final String reportType = report.getDescriptorForType().getFullName();
+      for (InetSocketAddress endpoint : endpoints) {
+        // Check report type
+        if (reportType.equals(containerReportsProtoName)) {
+          containerReport = report;
+        } else if (reportType.equals(nodeReportProtoName)) {
+          nodeReport = report;
+        } else if (reportType.equals(pipelineReportsProtoName)) {
+          pipelineReport = report;
+        } else if (reportType.equals(commandStatusReportsProtoName) ||
+            reportType.equals(incrementalContainerReportProtoName)) {
+          // report type is CommandStatusReports or IncrementalContainerReport
+          synchronized (reports) {
+            reports.get(endpoint).add(report);
+          }
+        } else {
+          throw new IllegalArgumentException(
+              "Unidentified report message type: " + reportType);
         }
       }
     }
@@ -241,6 +284,15 @@ public List<GeneratedMessage> getAllAvailableReports(
   public List<GeneratedMessage> getReports(InetSocketAddress endpoint,
                                            int maxLimit) {
     List<GeneratedMessage> reportsToReturn = new LinkedList<>();
+    if (containerReport != null) {
+      reportsToReturn.add(containerReport);
+    }
+    if (nodeReport != null) {
+      reportsToReturn.add(nodeReport);
+    }
+    if (pipelineReport != null) {
+      reportsToReturn.add(pipelineReport);
+    }
     synchronized (reports) {
       List<GeneratedMessage> reportsForEndpoint = reports.get(endpoint);
       if (reportsForEndpoint != null) {
@@ -583,4 +635,16 @@ public void addEndpoint(InetSocketAddress endpoint) {
       this.reports.put(endpoint, new LinkedList<>());
     }
   }
+
+  public GeneratedMessage getContainerReport() {
+    return containerReport;
+  }
+
+  public GeneratedMessage getNodeReport() {
+    return nodeReport;
+  }
+
+  public GeneratedMessage getPipelineReport() {
+    return pipelineReport;
+  }
 }

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

@@ -141,15 +141,15 @@ public EndpointStateMachine.EndPointStates call() throws Exception {
       if (LOG.isDebugEnabled()) {
         LOG.debug("Sending heartbeat message :: {}", request.toString());
       }
-      SCMHeartbeatResponseProto reponse = rpcEndpoint.getEndPoint()
+      SCMHeartbeatResponseProto response = rpcEndpoint.getEndPoint()
           .sendHeartbeat(request);
-      processResponse(reponse, datanodeDetailsProto);
+      processResponse(response, datanodeDetailsProto);
       rpcEndpoint.setLastSuccessfulHeartbeat(ZonedDateTime.now());
       rpcEndpoint.zeroMissedCount();
     } catch (IOException ex) {
+      Preconditions.checkState(requestBuilder != null);
       // put back the reports which failed to be sent
       putBackReports(requestBuilder);
-
       rpcEndpoint.logIfNeeded(ex);
     } finally {
       rpcEndpoint.unlock();
@@ -160,12 +160,9 @@ public EndpointStateMachine.EndPointStates call() throws Exception {
   // TODO: Make it generic.
   private void putBackReports(SCMHeartbeatRequestProto.Builder requestBuilder) {
     List<GeneratedMessage> reports = new LinkedList<>();
-    if (requestBuilder.hasContainerReport()) {
-      reports.add(requestBuilder.getContainerReport());
-    }
-    if (requestBuilder.hasNodeReport()) {
-      reports.add(requestBuilder.getNodeReport());
-    }
+    // We only put back CommandStatusReports and IncrementalContainerReport
+    // because those are incremental. Container/Node/PipelineReport are
+    // accumulative so we can keep only the latest of each.
     if (requestBuilder.getCommandStatusReportsCount() != 0) {
       reports.addAll(requestBuilder.getCommandStatusReportsList());
     }
@@ -184,15 +181,30 @@ private void addReports(SCMHeartbeatRequestProto.Builder requestBuilder) {
     for (GeneratedMessage report :
         context.getAllAvailableReports(rpcEndpoint.getAddress())) {
       String reportName = report.getDescriptorForType().getFullName();
+      // Example reportName = hadoop.hdds.NodeReportProto
+
       for (Descriptors.FieldDescriptor descriptor :
           SCMHeartbeatRequestProto.getDescriptor().getFields()) {
+
         String heartbeatFieldName = descriptor.getMessageType().getFullName();
+        // Possible heartbeatFieldName =
+        //  hadoop.hdds.DatanodeDetailsProto
+        //  hadoop.hdds.NodeReportProto
+        //  hadoop.hdds.ContainerReportsProto
+        //  hadoop.hdds.IncrementalContainerReportProto
+        //  hadoop.hdds.CommandStatusReportsProto
+        //  hadoop.hdds.ContainerActionsProto
+        //  hadoop.hdds.PipelineActionsProto
+        //  hadoop.hdds.PipelineReportsProto
         if (heartbeatFieldName.equals(reportName)) {
           if (descriptor.isRepeated()) {
             requestBuilder.addRepeatedField(descriptor, report);
           } else {
             requestBuilder.setField(descriptor, report);
           }
+          // TODO: We can exit loop early here since we have a match already,
+          //  right? Double check.
+          break;
         }
       }
     }