[WIP] HDDS-15412. [DataNode] Disk volume-specific container replication thread pool

hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/GrpcClientMetricsInterceptor.java

@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hdds.scm;
+
+import org.apache.ratis.thirdparty.io.grpc.CallOptions;
+import org.apache.ratis.thirdparty.io.grpc.Channel;
+import org.apache.ratis.thirdparty.io.grpc.ClientCall;
+import org.apache.ratis.thirdparty.io.grpc.ClientInterceptor;
+import org.apache.ratis.thirdparty.io.grpc.ForwardingClientCall.SimpleForwardingClientCall;
+import org.apache.ratis.thirdparty.io.grpc.ForwardingClientCallListener.SimpleForwardingClientCallListener;
+import org.apache.ratis.thirdparty.io.grpc.Metadata;
+import org.apache.ratis.thirdparty.io.grpc.MethodDescriptor;
+import org.apache.ratis.thirdparty.io.grpc.Status;
+
+/**
+ * Interceptor to capture gRPC level metrics.
+ */
+public class GrpcClientMetricsInterceptor implements ClientInterceptor {
+
+  private final XceiverClientMetrics metrics;
+
+  public GrpcClientMetricsInterceptor(XceiverClientMetrics metrics) {
+    this.metrics = metrics;
+  }
+
+  @Override
+  public <ReqT, RespT> ClientCall<ReqT, RespT> interceptCall(
+      MethodDescriptor<ReqT, RespT> method, CallOptions callOptions,
+      Channel next) {
+
+    return new SimpleForwardingClientCall<ReqT, RespT>(
+        next.newCall(method, callOptions)) {
+
+      @Override
+      public void start(Listener<RespT> responseListener, Metadata headers) {
+        super.start(new SimpleForwardingClientCallListener<RespT>(responseListener) {
+          @Override
+          public void onClose(Status status, Metadata trailers) {
+            if (!status.isOk()) {
+              if (status.getCode() == Status.Code.UNAUTHENTICATED) {
+                metrics.incGrpcAuthenticationFailures();
+              } else if (status.getCode() == Status.Code.UNAVAILABLE ||
+                  status.getCode() == Status.Code.DEADLINE_EXCEEDED) {
+                metrics.incGrpcConnectionFailures();
+              }
+            }
+            super.onClose(status, trailers);
+          }
+        }, headers);
+      }
+    };
+  }
+}

hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/XceiverClientMetrics.java

@@ -48,6 +48,8 @@ public class XceiverClientMetrics implements MetricsSource {
   private @Metric MutableCounterLong totalOps;
   private @Metric MutableCounterLong ecReconstructionTotal;
   private @Metric MutableCounterLong ecReconstructionFailsTotal;
+  private @Metric MutableCounterLong grpcAuthenticationFailures;
+  private @Metric MutableCounterLong grpcConnectionFailures;
   private EnumMap<ContainerProtos.Type, MutableCounterLong> pendingOpsArray;
   private EnumMap<ContainerProtos.Type, MutableCounterLong> opsArray;
   private EnumMap<ContainerProtos.Type, PerformanceMetrics> containerOpsLatency;
@@ -115,6 +117,14 @@ public void incECReconstructionFailsTotal() {
     ecReconstructionFailsTotal.incr();
   }
 
+  public void incGrpcAuthenticationFailures() {
+    grpcAuthenticationFailures.incr();
+  }
+
+  public void incGrpcConnectionFailures() {
+    grpcConnectionFailures.incr();
+  }
+
   @VisibleForTesting
   public long getTotalOpCount() {
     return totalOps.value();
@@ -144,6 +154,8 @@ public void getMetrics(MetricsCollector collector, boolean b) {
     totalOps.snapshot(recordBuilder, true);
     ecReconstructionTotal.snapshot(recordBuilder, true);
     ecReconstructionFailsTotal.snapshot(recordBuilder, true);
+    grpcAuthenticationFailures.snapshot(recordBuilder, true);
+    grpcConnectionFailures.snapshot(recordBuilder, true);
 
     for (ContainerProtos.Type type : ContainerProtos.Type.values()) {
       pendingOpsArray.get(type).snapshot(recordBuilder, b);

hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/TestXceiverClientMetricsUnit.java

@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.hdds.scm;
+
+import static org.apache.ozone.test.MetricsAsserts.assertCounter;
+import static org.apache.ozone.test.MetricsAsserts.getMetrics;
+
+import org.apache.hadoop.metrics2.MetricsRecordBuilder;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Unit test for XceiverClientMetrics.
+ */
+public class TestXceiverClientMetricsUnit {
+
+  @Test
+  public void testGrpcFailures() {
+    XceiverClientMetrics metrics = XceiverClientMetrics.create();
+    try {
+      metrics.incGrpcAuthenticationFailures();
+      metrics.incGrpcConnectionFailures();
+      metrics.incGrpcConnectionFailures();
+
+      MetricsRecordBuilder recordBuilder = getMetrics(XceiverClientMetrics.SOURCE_NAME);
+      assertCounter("GrpcAuthenticationFailures", 1L, recordBuilder);
+      assertCounter("GrpcConnectionFailures", 2L, recordBuilder);
+    } finally {
+      metrics.unRegister();
+    }
+  }
+}

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

@@ -195,6 +195,9 @@ public DatanodeStateMachine(HddsDatanodeService hddsDatanodeService,
     }
     nextHB = new AtomicLong(Time.monotonicNow());
 
+    ReplicationConfig replicationConfig =
+        conf.getObject(ReplicationConfig.class);
+
     ContainerImporter importer = new ContainerImporter(conf,
         container.getContainerSet(),
         container.getController(),
@@ -205,15 +208,14 @@ public DatanodeStateMachine(HddsDatanodeService hddsDatanodeService,
         importer,
         new SimpleContainerDownloader(conf, certClient));
     ContainerReplicator pushReplicator = new PushReplicator(conf,
-        new OnDemandContainerReplicationSource(container.getController()),
+        new OnDemandContainerReplicationSource(container.getController(),
+            replicationConfig),
         new GrpcContainerUploader(conf, certClient, container.getController())
     );
 
     pullReplicatorWithMetrics = new MeasuredReplicator(pullReplicator, "pull");
     pushReplicatorWithMetrics = new MeasuredReplicator(pushReplicator, "push");
 
-    ReplicationConfig replicationConfig =
-        conf.getObject(ReplicationConfig.class);
     supervisor = ReplicationSupervisor.newBuilder()
         .stateContext(context)
         .datanodeConfig(dnConf)

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/volume/HddsVolume.java

@@ -31,6 +31,7 @@
 import java.util.concurrent.ConcurrentSkipListSet;
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Function;
 import org.apache.commons.io.FileUtils;
@@ -109,6 +110,9 @@ public class HddsVolume extends StorageVolume {
   private final AtomicBoolean dbLoaded = new AtomicBoolean(false);
   private final AtomicBoolean dbLoadFailure = new AtomicBoolean(false);
 
+  private final AtomicInteger activeOutboundReplications =
+      new AtomicInteger(0);
+
   /**
    * Builder for HddsVolume.
    */
@@ -701,4 +705,16 @@ public void compactDb() {
       LOG.warn("compact rocksdb error in {}", dbFilePath, e);
     }
   }
+
+  public int incActiveOutboundReplications() {
+    return activeOutboundReplications.incrementAndGet();
+  }
+
+  public int decActiveOutboundReplications() {
+    return activeOutboundReplications.decrementAndGet();
+  }
+
+  public int getActiveOutboundReplications() {
+    return activeOutboundReplications.get();
+  }
 }

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/AbstractReplicationTask.java

@@ -34,7 +34,7 @@ public abstract class AbstractReplicationTask {
 
   private final long containerId;
 
-  private final Instant queued;
+  private Instant queued;
 
   private final long deadlineMsSinceEpoch;
 
@@ -78,6 +78,10 @@ public Instant getQueued() {
     return queued;
   }
 
+  public void updateQueuedTime() {
+    this.queued = Instant.now();
+  }
+
   public long getTerm() {
     return term;
   }

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/ContainerDownloader.java

@@ -18,6 +18,7 @@
 package org.apache.hadoop.ozone.container.replication;
 
 import java.io.Closeable;
+import java.io.IOException;
 import java.nio.file.Path;
 import java.util.List;
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
@@ -34,6 +35,6 @@ public interface ContainerDownloader extends Closeable {
 
   Path getContainerDataFromReplicas(long containerId,
       List<DatanodeDetails> sources, Path downloadDir,
-      CopyContainerCompression compression);
+      CopyContainerCompression compression) throws IOException;
 
 }

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/DownloadAndImportReplicator.java

@@ -23,6 +23,8 @@
 import java.util.List;
 import org.apache.hadoop.hdds.conf.ConfigurationSource;
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
+import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
+import org.apache.hadoop.hdds.scm.container.common.helpers.StorageContainerException;
 import org.apache.hadoop.ozone.container.common.impl.ContainerSet;
 import org.apache.hadoop.ozone.container.common.volume.HddsVolume;
 import org.apache.hadoop.ozone.container.replication.AbstractReplicationTask.Status;
@@ -96,8 +98,16 @@ public void replicate(ReplicationTask task) {
       LOG.info("Container {} is replicated successfully", containerID);
       task.setStatus(Status.DONE);
     } catch (IOException e) {
-      LOG.error("Container {} replication was unsuccessful.", containerID, e);
-      task.setStatus(Status.FAILED);
+      if (e instanceof StorageContainerException &&
+          ((StorageContainerException) e).getResult() ==
+              ContainerProtos.Result.REPLICATION_LIMIT_REACHED) {
+        LOG.info("Container {} replication will be retried as the " +
+            "replication limit was reached.", containerID);
+        task.setStatus(Status.QUEUED);
+      } else {
+        LOG.error("Container {} replication was unsuccessful.", containerID, e);
+        task.setStatus(Status.FAILED);
+      }
     } finally {
       if (targetVolume != null) {
         targetVolume.incCommittedBytes(-containerImporter.getDefaultReplicationSpace());

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/GrpcReplicationService.java

@@ -17,6 +17,7 @@
 
 package org.apache.hadoop.ozone.container.replication;
 
+import static org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Result.REPLICATION_LIMIT_REACHED;
 import static org.apache.hadoop.hdds.protocol.datanode.proto.IntraDatanodeProtocolServiceGrpc.getDownloadMethod;
 import static org.apache.hadoop.hdds.protocol.datanode.proto.IntraDatanodeProtocolServiceGrpc.getUploadMethod;
 import static org.apache.hadoop.ozone.container.replication.CopyContainerCompression.fromProto;
@@ -30,12 +31,14 @@
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.SendContainerRequest;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.SendContainerResponse;
 import org.apache.hadoop.hdds.protocol.datanode.proto.IntraDatanodeProtocolServiceGrpc;
+import org.apache.hadoop.hdds.scm.container.common.helpers.StorageContainerException;
 import org.apache.hadoop.hdds.utils.IOUtils;
 import org.apache.ratis.grpc.util.ZeroCopyMessageMarshaller;
 import org.apache.ratis.thirdparty.com.google.protobuf.MessageLite;
 import org.apache.ratis.thirdparty.io.grpc.MethodDescriptor;
 import org.apache.ratis.thirdparty.io.grpc.ServerCallHandler;
 import org.apache.ratis.thirdparty.io.grpc.ServerServiceDefinition;
+import org.apache.ratis.thirdparty.io.grpc.Status;
 import org.apache.ratis.thirdparty.io.grpc.stub.CallStreamObserver;
 import org.apache.ratis.thirdparty.io.grpc.stub.StreamObserver;
 import org.slf4j.Logger;
@@ -132,6 +135,15 @@ public void download(CopyContainerRequestProto request,
           (CallStreamObserver<CopyContainerResponseProto>) responseObserver,
           containerID, BUFFER_SIZE);
       source.copyData(containerID, outputStream, compression);
+    } catch (StorageContainerException e) {
+      if (e.getResult() == REPLICATION_LIMIT_REACHED) {
+        responseObserver.onError(Status.RESOURCE_EXHAUSTED
+            .withDescription(e.getMessage())
+            .asRuntimeException());
+      } else {
+        LOG.warn("Error streaming container {}", containerID, e);
+        responseObserver.onError(e);
+      }
     } catch (IOException e) {
       LOG.warn("Error streaming container {}", containerID, e);
       responseObserver.onError(e);

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/replication/OnDemandContainerReplicationSource.java

@@ -18,13 +18,17 @@
 package org.apache.hadoop.ozone.container.replication;
 
 import static org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Result.CONTAINER_NOT_FOUND;
+import static org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.Result.REPLICATION_LIMIT_REACHED;
 
 import java.io.IOException;
 import java.io.OutputStream;
 import org.apache.hadoop.hdds.scm.container.common.helpers.StorageContainerException;
 import org.apache.hadoop.ozone.container.common.interfaces.Container;
+import org.apache.hadoop.ozone.container.common.volume.HddsVolume;
 import org.apache.hadoop.ozone.container.keyvalue.TarContainerPacker;
 import org.apache.hadoop.ozone.container.ozoneimpl.ContainerController;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * A naive implementation of the replication source which creates a tar file
@@ -33,11 +37,17 @@
 public class OnDemandContainerReplicationSource
     implements ContainerReplicationSource {
 
+  private static final Logger LOG =
+      LoggerFactory.getLogger(OnDemandContainerReplicationSource.class);
+
   private final ContainerController controller;
+  private final ReplicationServer.ReplicationConfig config;
 
   public OnDemandContainerReplicationSource(
-      ContainerController controller) {
+      ContainerController controller,
+      ReplicationServer.ReplicationConfig config) {
     this.controller = controller;
+    this.config = config;
   }
 
   @Override
@@ -57,8 +67,27 @@ public void copyData(long containerId, OutputStream destination,
           " is not found.", CONTAINER_NOT_FOUND);
     }
 
-    controller.exportContainer(
-        container.getContainerType(), containerId, destination,
-        new TarContainerPacker(compression));
+    HddsVolume volume = (HddsVolume) container.getContainerData().getVolume();
+    if (volume != null) {
+      if (volume.getActiveOutboundReplications() >=
+          config.getVolumeOutboundLimit()) {
+        LOG.info("Volume {} has reached the maximum number of concurrent " +
+                "replication reads ({})", volume.getStorageID(),
+            config.getVolumeOutboundLimit());
+        throw new StorageContainerException("Volume " + volume.getStorageID() +
+            " has reached the maximum number of concurrent replication reads ("
+            + config.getVolumeOutboundLimit() + ")", REPLICATION_LIMIT_REACHED);
+      }
+      volume.incActiveOutboundReplications();
+    }
+    try {
+      controller.exportContainer(
+          container.getContainerType(), containerId, destination,
+          new TarContainerPacker(compression));
+    } finally {
+      if (volume != null) {
+        volume.decActiveOutboundReplications();
+      }
+    }
   }
 }