HDDS-7419. Integrate the GetKeyInfo API to OFS

README.md

@@ -8,7 +8,7 @@ Ozone is a scalable, redundant, and distributed object store for Hadoop and Clou
  * SCALABLE: Ozone is designed to scale to tens of billions of files and blocks and, in the future, even more.
  * CONSISTENT: Ozone is a strongly consistent object store. This consistency is achieved by using protocols like RAFT.
  * CLOUD-NATIVE: Ozone is designed to work well in containerized environments like YARN and Kubernetes.
- * SECURE: Ozone integrates with Kerberos infrastructure for access control and supports TDE and on-wire encryption.
+ * SECURE: Ozone integrates with Kerberos infrastructure for authentication, supports native ACLs and integrates with Ranger for access control and supports TDE and on-wire encryption.
  * HIGHLY AVAILABLE: Ozone is a fully replicated system that is designed to survive multiple failures.
 
 ## Documentation

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

@@ -39,6 +39,8 @@ public class XceiverClientMetrics {
 
   private @Metric MutableCounterLong pendingOps;
   private @Metric MutableCounterLong totalOps;
+  private @Metric MutableCounterLong ecReconstructionTotal;
+  private @Metric MutableCounterLong ecReconstructionFailsTotal;
   private MutableCounterLong[] pendingOpsArray;
   private MutableCounterLong[] opsArray;
   private MutableRate[] containerOpsLatency;
@@ -100,6 +102,14 @@ public long getPendingContainerOpCountMetrics(ContainerProtos.Type type) {
     return pendingOpsArray[type.ordinal()].value();
   }
 
+  public void incECReconstructionTotal() {
+    ecReconstructionTotal.incr();
+  }
+
+  public void incECReconstructionFailsTotal() {
+    ecReconstructionFailsTotal.incr();
+  }
+
   @VisibleForTesting
   public long getTotalOpCount() {
     return totalOps.value();

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

@@ -46,6 +46,7 @@
 import org.apache.hadoop.security.token.Token;
 
 import com.google.common.annotations.VisibleForTesting;
+import org.apache.ratis.thirdparty.io.grpc.Status;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -140,18 +141,22 @@ public synchronized void initialize() throws IOException {
     IOException catchEx = null;
     do {
       try {
-        // If refresh returns new pipeline, retry with it.
-        // If we get IOException due to connectivity issue,
-        // retry according to retry policy.
         chunks = getChunkInfos();
         break;
+        // If we get a StorageContainerException or an IOException due to
+        // datanodes are not reachable, refresh to get the latest pipeline
+        // info and retry.
+        // Otherwise, just retry according to the retry policy.
       } catch (SCMSecurityException ex) {
         throw ex;
       } catch (StorageContainerException ex) {
         refreshPipeline(ex);
         catchEx = ex;
       } catch (IOException ex) {
         LOG.debug("Retry to get chunk info fail", ex);
+        if (isConnectivityIssue(ex)) {
+          refreshPipeline(ex);
+        }
         catchEx = ex;
       }
     } while (shouldRetryRead(catchEx));
@@ -187,19 +192,19 @@ public synchronized void initialize() throws IOException {
     }
   }
 
+  /**
+   * Check if this exception is because datanodes are not reachable.
+   */
+  private boolean isConnectivityIssue(IOException ex) {
+    return Status.fromThrowable(ex).getCode() == Status.UNAVAILABLE.getCode();
+  }
+
   private void refreshPipeline(IOException cause) throws IOException {
     LOG.info("Unable to read information for block {} from pipeline {}: {}",
         blockID, pipeline.getId(), cause.getMessage());
     if (refreshPipelineFunction != null) {
       LOG.debug("Re-fetching pipeline for block {}", blockID);
-      Pipeline newPipeline = refreshPipelineFunction.apply(blockID);
-      if (newPipeline == null || newPipeline.sameDatanodes(pipeline)) {
-        LOG.warn("No new pipeline for block {}", blockID);
-        throw cause;
-      } else {
-        LOG.debug("New pipeline got for block {}", blockID);
-        this.pipeline = newPipeline;
-      }
+      this.pipeline = refreshPipelineFunction.apply(blockID);
     } else {
       throw cause;
     }
@@ -301,21 +306,27 @@ protected synchronized int readWithStrategy(ByteReaderStrategy strategy)
       int numBytesRead;
       try {
         numBytesRead = strategy.readFromBlock(current, numBytesToRead);
-        retries = 0; // reset retries after successful read
+        retries = 0;
+        // If we get a StorageContainerException or an IOException due to
+        // datanodes are not reachable, refresh to get the latest pipeline
+        // info and retry.
+        // Otherwise, just retry according to the retry policy.
+      } catch (SCMSecurityException ex) {
+        throw ex;
       } catch (StorageContainerException e) {
         if (shouldRetryRead(e)) {
           handleReadError(e);
           continue;
         } else {
           throw e;
         }
-      } catch (SCMSecurityException ex) {
-        throw ex;
       } catch (IOException ex) {
-        // We got a IOException which might be due
-        // to DN down or connectivity issue.
         if (shouldRetryRead(ex)) {
-          current.releaseClient();
+          if (isConnectivityIssue(ex)) {
+            handleReadError(ex);
+          } else {
+            current.releaseClient();
+          }
           continue;
         } else {
           throw ex;

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

@@ -32,7 +32,6 @@
 import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos.ReadChunkResponseProto;
 import org.apache.hadoop.hdds.scm.XceiverClientFactory;
 import org.apache.hadoop.hdds.scm.XceiverClientSpi;
-import org.apache.hadoop.hdds.scm.container.common.helpers.StorageContainerException;
 import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
 import org.apache.hadoop.ozone.common.Checksum;
 import org.apache.hadoop.ozone.common.ChecksumData;
@@ -425,20 +424,12 @@ protected ByteBuffer[] readChunk(ChunkInfo readChunkInfo)
       throws IOException {
     ReadChunkResponseProto readChunkResponse;
 
-    try {
-      List<CheckedBiFunction> validators =
-          ContainerProtocolCalls.getValidatorList();
-      validators.add(validator);
+    List<CheckedBiFunction> validators =
+        ContainerProtocolCalls.getValidatorList();
+    validators.add(validator);
 
-      readChunkResponse = ContainerProtocolCalls.readChunk(xceiverClient,
-          readChunkInfo, blockID, validators, token);
-
-    } catch (IOException e) {
-      if (e instanceof StorageContainerException) {
-        throw e;
-      }
-      throw new IOException("Unexpected OzoneException: " + e.toString(), e);
-    }
+    readChunkResponse = ContainerProtocolCalls.readChunk(xceiverClient,
+        readChunkInfo, blockID, validators, token);
 
     if (readChunkResponse.hasData()) {
       return readChunkResponse.getData().asReadOnlyByteBufferList()

hadoop-hdds/client/src/main/java/org/apache/hadoop/ozone/client/io/ECBlockInputStreamProxy.java

@@ -21,6 +21,7 @@
 import org.apache.hadoop.hdds.client.ECReplicationConfig;
 import org.apache.hadoop.hdds.protocol.DatanodeDetails;
 import org.apache.hadoop.hdds.scm.XceiverClientFactory;
+import org.apache.hadoop.hdds.scm.XceiverClientManager;
 import org.apache.hadoop.hdds.scm.pipeline.Pipeline;
 import org.apache.hadoop.hdds.scm.storage.BlockExtendedInputStream;
 import org.apache.hadoop.hdds.scm.storage.BlockLocationInfo;
@@ -117,6 +118,10 @@ private synchronized void setReaderType() {
   }
 
   private void createBlockReader() {
+    if (reconstructionReader) {
+      XceiverClientManager.getXceiverClientMetrics()
+          .incECReconstructionTotal();
+    }
     blockReader = ecBlockInputStreamFactory.create(reconstructionReader,
         failedLocations, repConfig, blockInfo, verifyChecksum,
         xceiverClientFactory, refreshFunction);
@@ -162,6 +167,8 @@ public synchronized int read(ByteBuffer buf) throws IOException {
         // If we get an error from the reconstruction reader, there
         // is nothing left to try. It will re-try until it has insufficient
         // locations internally, so if an error comes here, just re-throw it.
+        XceiverClientManager.getXceiverClientMetrics()
+            .incECReconstructionFailsTotal();
         throw e;
       }
       if (e instanceof BadDataLocationException) {