HDDS-1329. Update documentation for Ozone-0.4.0 release. Contributed By Ajay Kumar.

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/NodeHealthScriptRunner.java

@@ -249,7 +249,7 @@ public boolean isHealthy() {
   }
 
   /**
-   * Sets if the node is healhty or not considering disks' health also.
+   * Sets if the node is healthy or not considering disks' health also.
    * 
    * @param isHealthy
    *          if or not node is healthy

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

@@ -33,6 +33,8 @@
 import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
 import org.apache.hadoop.hdds.security.exception.SCMSecurityException;
 import org.apache.hadoop.hdds.security.x509.SecurityConfig;
+import org.apache.hadoop.hdds.tracing.GrpcClientInterceptor;
+import org.apache.hadoop.hdds.tracing.TracingUtil;
 import org.apache.hadoop.ozone.OzoneConfigKeys;
 import org.apache.hadoop.ozone.OzoneConsts;
 import org.apache.hadoop.security.UserGroupInformation;
@@ -136,7 +138,8 @@ private void connectToDatanode(DatanodeDetails dn, String encodedToken)
     NettyChannelBuilder channelBuilder = NettyChannelBuilder.forAddress(dn
             .getIpAddress(), port).usePlaintext()
             .maxInboundMessageSize(OzoneConsts.OZONE_SCM_CHUNK_MAX_SIZE)
-            .intercept(new ClientCredentialInterceptor(userName, encodedToken));
+            .intercept(new ClientCredentialInterceptor(userName, encodedToken),
+                new GrpcClientInterceptor());
     if (secConfig.isGrpcTlsEnabled()) {
       File trustCertCollectionFile = secConfig.getTrustStoreFile();
       File privateKeyFile = secConfig.getClientPrivateKeyFile();
@@ -204,7 +207,7 @@ public ContainerCommandResponseProto sendCommand(
       ContainerCommandRequestProto request) throws IOException {
     try {
       XceiverClientReply reply;
-      reply = sendCommandWithRetry(request, null);
+      reply = sendCommandWithTraceIDAndRetry(request, null);
       ContainerCommandResponseProto responseProto = reply.getResponse().get();
       return responseProto;
     } catch (ExecutionException | InterruptedException e) {
@@ -217,7 +220,21 @@ public XceiverClientReply sendCommand(
       ContainerCommandRequestProto request, List<DatanodeDetails> excludeDns)
       throws IOException {
     Preconditions.checkState(HddsUtils.isReadOnly(request));
-    return sendCommandWithRetry(request, excludeDns);
+    return sendCommandWithTraceIDAndRetry(request, excludeDns);
+  }
+
+  private XceiverClientReply sendCommandWithTraceIDAndRetry(
+      ContainerCommandRequestProto request, List<DatanodeDetails> excludeDns)
+      throws IOException {
+    try (Scope scope = GlobalTracer.get()
+        .buildSpan("XceiverClientGrpc." + request.getCmdType().name())
+        .startActive(true)) {
+      ContainerCommandRequestProto finalPayload =
+          ContainerCommandRequestProto.newBuilder(request)
+              .setTraceID(TracingUtil.exportCurrentSpan())
+              .build();
+      return sendCommandWithRetry(finalPayload, excludeDns);
+    }
   }
 
   private XceiverClientReply sendCommandWithRetry(
@@ -253,7 +270,7 @@ private XceiverClientReply sendCommandWithRetry(
           break;
         }
       } catch (ExecutionException | InterruptedException e) {
-        LOG.warn("Failed to execute command " + request + " on datanode " + dn
+        LOG.debug("Failed to execute command " + request + " on datanode " + dn
             .getUuidString(), e);
         if (Status.fromThrowable(e.getCause()).getCode()
             == Status.UNAUTHENTICATED.getCode()) {

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

@@ -17,6 +17,7 @@
  */
 package org.apache.hadoop.hdds.scm;
 
+import com.google.common.annotations.VisibleForTesting;
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
 import org.apache.hadoop.metrics2.MetricsSystem;
@@ -37,7 +38,9 @@ public class XceiverClientMetrics {
       .getSimpleName();
 
   private @Metric MutableCounterLong pendingOps;
+  private @Metric MutableCounterLong totalOps;
   private MutableCounterLong[] pendingOpsArray;
+  private MutableCounterLong[] opsArray;
   private MutableRate[] containerOpsLatency;
   private MetricsRegistry registry;
 
@@ -46,12 +49,17 @@ public XceiverClientMetrics() {
     this.registry = new MetricsRegistry(SOURCE_NAME);
 
     this.pendingOpsArray = new MutableCounterLong[numEnumEntries];
+    this.opsArray = new MutableCounterLong[numEnumEntries];
     this.containerOpsLatency = new MutableRate[numEnumEntries];
     for (int i = 0; i < numEnumEntries; i++) {
       pendingOpsArray[i] = registry.newCounter(
           "numPending" + ContainerProtos.Type.forNumber(i + 1),
           "number of pending" + ContainerProtos.Type.forNumber(i + 1) + " ops",
           (long) 0);
+      opsArray[i] = registry
+          .newCounter("opCount" + ContainerProtos.Type.forNumber(i + 1),
+              "number of" + ContainerProtos.Type.forNumber(i + 1) + " ops",
+              (long) 0);
 
       containerOpsLatency[i] = registry.newRate(
           ContainerProtos.Type.forNumber(i + 1) + "Latency",
@@ -68,6 +76,8 @@ public static XceiverClientMetrics create() {
 
   public void incrPendingContainerOpsMetrics(ContainerProtos.Type type) {
     pendingOps.incr();
+    totalOps.incr();
+    opsArray[type.ordinal()].incr();
     pendingOpsArray[type.ordinal()].incr();
   }
 
@@ -85,6 +95,16 @@ public long getContainerOpsMetrics(ContainerProtos.Type type) {
     return pendingOpsArray[type.ordinal()].value();
   }
 
+  @VisibleForTesting
+  public long getTotalOpCount() {
+    return totalOps.value();
+  }
+
+  @VisibleForTesting
+  public long getContainerOpCountMetrics(ContainerProtos.Type type) {
+    return opsArray[type.ordinal()].value();
+  }
+
   public void unRegister() {
     MetricsSystem ms = DefaultMetricsSystem.instance();
     ms.unregisterSource(SOURCE_NAME);

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

@@ -18,13 +18,15 @@
 
 package org.apache.hadoop.hdds.scm;
 
+import com.google.common.annotations.VisibleForTesting;
 import org.apache.hadoop.hdds.HddsUtils;
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
 import org.apache.hadoop.hdds.security.x509.SecurityConfig;
 
 import io.opentracing.Scope;
 import io.opentracing.util.GlobalTracer;
+import org.apache.hadoop.util.Time;
 import org.apache.ratis.grpc.GrpcTlsConfig;
 import org.apache.ratis.proto.RaftProtos;
 import org.apache.ratis.protocol.RaftRetryFailureException;
@@ -47,6 +49,7 @@
 import org.apache.ratis.rpc.RpcType;
 import org.apache.ratis.rpc.SupportedRpcType;
 import org.apache.ratis.thirdparty.com.google.protobuf.ByteString;
+import org.apache.ratis.util.TimeDuration;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -74,14 +77,16 @@ public static XceiverClientRatis newXceiverClientRatis(
     final String rpcType = ozoneConf
         .get(ScmConfigKeys.DFS_CONTAINER_RATIS_RPC_TYPE_KEY,
             ScmConfigKeys.DFS_CONTAINER_RATIS_RPC_TYPE_DEFAULT);
+    final TimeDuration clientRequestTimeout =
+        RatisHelper.getClientRequestTimeout(ozoneConf);
     final int maxOutstandingRequests =
         HddsClientUtils.getMaxOutstandingRequests(ozoneConf);
     final RetryPolicy retryPolicy = RatisHelper.createRetryPolicy(ozoneConf);
     final GrpcTlsConfig tlsConfig = RatisHelper.createTlsClientConfig(new
         SecurityConfig(ozoneConf));
     return new XceiverClientRatis(pipeline,
         SupportedRpcType.valueOfIgnoreCase(rpcType), maxOutstandingRequests,
-        retryPolicy, tlsConfig);
+        retryPolicy, tlsConfig, clientRequestTimeout);
   }
 
   private final Pipeline pipeline;
@@ -90,19 +95,22 @@ public static XceiverClientRatis newXceiverClientRatis(
   private final int maxOutstandingRequests;
   private final RetryPolicy retryPolicy;
   private final GrpcTlsConfig tlsConfig;
+  private final TimeDuration clientRequestTimeout;
 
   // Map to track commit index at every server
   private final ConcurrentHashMap<UUID, Long> commitInfoMap;
 
   // create a separate RaftClient for watchForCommit API
   private RaftClient watchClient;
 
+  private XceiverClientMetrics metrics;
+
   /**
    * Constructs a client.
    */
   private XceiverClientRatis(Pipeline pipeline, RpcType rpcType,
       int maxOutStandingChunks, RetryPolicy retryPolicy,
-      GrpcTlsConfig tlsConfig) {
+      GrpcTlsConfig tlsConfig, TimeDuration timeout) {
     super();
     this.pipeline = pipeline;
     this.rpcType = rpcType;
@@ -111,6 +119,8 @@ private XceiverClientRatis(Pipeline pipeline, RpcType rpcType,
     commitInfoMap = new ConcurrentHashMap<>();
     watchClient = null;
     this.tlsConfig = tlsConfig;
+    this.clientRequestTimeout = timeout;
+    metrics = XceiverClientManager.getXceiverClientMetrics();
   }
 
   private void updateCommitInfosMap(
@@ -160,7 +170,7 @@ public void connect() throws Exception {
     // requests to be handled by raft client
     if (!client.compareAndSet(null,
         RatisHelper.newRaftClient(rpcType, getPipeline(), retryPolicy,
-            maxOutstandingRequests, tlsConfig))) {
+            maxOutstandingRequests, tlsConfig, clientRequestTimeout))) {
       throw new IllegalStateException("Client is already connected.");
     }
   }
@@ -194,6 +204,12 @@ private RaftClient getClient() {
     return Objects.requireNonNull(client.get(), "client is null");
   }
 
+
+  @VisibleForTesting
+  public ConcurrentHashMap<UUID, Long> getCommitInfoMap() {
+    return commitInfoMap;
+  }
+
   private CompletableFuture<RaftClientReply> sendRequestAsync(
       ContainerCommandRequestProto request) {
     try (Scope scope = GlobalTracer.get()
@@ -243,7 +259,7 @@ public XceiverClientReply watchForCommit(long index, long timeout)
     if (watchClient == null) {
       watchClient =
           RatisHelper.newRaftClient(rpcType, getPipeline(), retryPolicy,
-              maxOutstandingRequests, tlsConfig);
+              maxOutstandingRequests, tlsConfig, clientRequestTimeout);
     }
     CompletableFuture<RaftClientReply> replyFuture = watchClient
         .sendWatchAsync(index, RaftProtos.ReplicationLevel.ALL_COMMITTED);
@@ -260,9 +276,9 @@ public XceiverClientReply watchForCommit(long index, long timeout)
       // TODO : need to remove the code to create the new RaftClient instance
       // here once the watch request bypassing sliding window in Raft Client
       // gets fixed.
-      watchClient = RatisHelper
-          .newRaftClient(rpcType, getPipeline(), retryPolicy,
-              maxOutstandingRequests, tlsConfig);
+      watchClient =
+          RatisHelper.newRaftClient(rpcType, getPipeline(), retryPolicy,
+              maxOutstandingRequests, tlsConfig, clientRequestTimeout);
       reply = watchClient
           .sendWatchAsync(index, RaftProtos.ReplicationLevel.MAJORITY_COMMITTED)
           .get(timeout, TimeUnit.MILLISECONDS);
@@ -296,47 +312,52 @@ public XceiverClientReply watchForCommit(long index, long timeout)
   public XceiverClientReply sendCommandAsync(
       ContainerCommandRequestProto request) {
     XceiverClientReply asyncReply = new XceiverClientReply(null);
+    long requestTime = Time.monotonicNowNanos();
     CompletableFuture<RaftClientReply> raftClientReply =
         sendRequestAsync(request);
+    metrics.incrPendingContainerOpsMetrics(request.getCmdType());
     CompletableFuture<ContainerCommandResponseProto> containerCommandResponse =
-        raftClientReply.whenComplete((reply, e) -> LOG.debug(
-            "received reply {} for request: cmdType={} containerID={}"
-                + " pipelineID={} traceID={} exception: {}", reply,
-            request.getCmdType(), request.getContainerID(),
-            request.getPipelineID(), request.getTraceID(), e))
-            .thenApply(reply -> {
-              try {
-                // we need to handle RaftRetryFailure Exception
-                RaftRetryFailureException raftRetryFailureException =
-                    reply.getRetryFailureException();
-                if (raftRetryFailureException != null) {
-                  // in case of raft retry failure, the raft client is
-                  // not able to connect to the leader hence the pipeline
-                  // can not be used but this instance of RaftClient will close
-                  // and refreshed again. In case the client cannot connect to
-                   // leader, getClient call will fail.
+        raftClientReply.whenComplete((reply, e) -> {
+          LOG.debug("received reply {} for request: cmdType={} containerID={}"
+                  + " pipelineID={} traceID={} exception: {}", reply,
+              request.getCmdType(), request.getContainerID(),
+              request.getPipelineID(), request.getTraceID(), e);
+          metrics.decrPendingContainerOpsMetrics(request.getCmdType());
+          metrics.addContainerOpsLatency(request.getCmdType(),
+              Time.monotonicNowNanos() - requestTime);
+        }).thenApply(reply -> {
+          try {
+            // we need to handle RaftRetryFailure Exception
+            RaftRetryFailureException raftRetryFailureException =
+                reply.getRetryFailureException();
+            if (raftRetryFailureException != null) {
+              // in case of raft retry failure, the raft client is
+              // not able to connect to the leader hence the pipeline
+              // can not be used but this instance of RaftClient will close
+              // and refreshed again. In case the client cannot connect to
+              // leader, getClient call will fail.
 
-                  // No need to set the failed Server ID here. Ozone client
-                  // will directly exclude this pipeline in next allocate block
-                  // to SCM as in this case, it is the raft client which is not
-                  // able to connect to leader in the pipeline, though the
-                  // pipeline can still be functional.
-                  throw new CompletionException(raftRetryFailureException);
-                }
-                ContainerCommandResponseProto response =
-                    ContainerCommandResponseProto
-                        .parseFrom(reply.getMessage().getContent());
-                UUID serverId = RatisHelper.toDatanodeId(reply.getReplierId());
-                if (response.getResult() == ContainerProtos.Result.SUCCESS) {
-                  updateCommitInfosMap(reply.getCommitInfos());
-                }
-                asyncReply.setLogIndex(reply.getLogIndex());
-                addDatanodetoReply(serverId, asyncReply);
-                return response;
-              } catch (InvalidProtocolBufferException e) {
-                throw new CompletionException(e);
-              }
-            });
+              // No need to set the failed Server ID here. Ozone client
+              // will directly exclude this pipeline in next allocate block
+              // to SCM as in this case, it is the raft client which is not
+              // able to connect to leader in the pipeline, though the
+              // pipeline can still be functional.
+              throw new CompletionException(raftRetryFailureException);
+            }
+            ContainerCommandResponseProto response =
+                ContainerCommandResponseProto
+                    .parseFrom(reply.getMessage().getContent());
+            UUID serverId = RatisHelper.toDatanodeId(reply.getReplierId());
+            if (response.getResult() == ContainerProtos.Result.SUCCESS) {
+              updateCommitInfosMap(reply.getCommitInfos());
+            }
+            asyncReply.setLogIndex(reply.getLogIndex());
+            addDatanodetoReply(serverId, asyncReply);
+            return response;
+          } catch (InvalidProtocolBufferException e) {
+            throw new CompletionException(e);
+          }
+        });
     asyncReply.setResponse(containerCommandResponse);
     return asyncReply;
   }