Merge branch 'master' into HDDS-13224

Author: YutaLin

hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneSecurityUtil.java

@@ -24,6 +24,7 @@
 
 import java.io.File;
 import java.io.IOException;
+import java.net.Inet6Address;
 import java.net.InetAddress;
 import java.net.NetworkInterface;
 import java.nio.file.Path;
@@ -108,8 +109,8 @@ public static List<InetAddress> getValidInetsForCurrentHost()
           InetAddress addr = enumAdds.nextElement();
 
           String hostAddress = addr.getHostAddress();
-          if (!INVALID_IPS.contains(hostAddress)
-              && ipValidator.isValid(hostAddress)) {
+          if (!INVALID_IPS.contains(hostAddress) && ipValidator.isValid(hostAddress)
+              && !isScopedOrMaskingIPv6Address(addr)) {
             LOG.info("Adding ip:{},host:{}", hostAddress, addr.getHostName());
             hostIps.add(addr);
           } else {
@@ -122,6 +123,41 @@ public static List<InetAddress> getValidInetsForCurrentHost()
     return hostIps;
   }
 
+  /**
+   * Determines if the supplied address is an IPv6 address, with a defined scope-id and/or with a defined prefix length.
+   * <p>
+   * This method became necessary after Commons Validator was upgraded from 1.6 version to 1.10. In 1.10 version the
+   * IPv6 addresses with a scope-id and/or with a prefix specifier became valid IPv6 addresses, but as these features
+   * are changing the string representation to do not represent only the 16 octet that specifies the address, the
+   * string representation can not be used as it is as a SAN extension in X.509 anymore as in RFC-5280 this type of
+   * Subject Alternative Name is exactly 4 octets in case of an IPv4 address, and 16 octets in case of an IPv6 address.
+   * BouncyCastle does not have support to deal with these in an IPAddress typed GeneralName, so we need to keep the
+   * previous behaviour, and skip IPv6 addresses with a prefix length and/or a scope-id.
+   * <p>
+   * According to RFC-4007 and the InetAddress contract the scope-id is at the end of the address' strin
+   * representation, separated by a '%' character from the address.
+   * According to RFC-4632 there is a possibility to specify a prefix length at the end of the address to specify
+   * routing related information. RFC-4007 specifies the prefix length to come after the scope-id.
+   * <p>
+   *
+   * @param addr the InetAddress to check
+   * @return if the InetAddress is an IPv6 address and if so it contains a scope-id and/or a prefix length.
+   * @see <a href="https://datatracker.ietf.org/doc/html/rfc4007">RFC-4007 - Scoped IPv6 Addresses</a>
+   * @see <a href="https://datatracker.ietf.org/doc/html/rfc4632#section-5.1">RFC-4632 - CIDR addressing strategy -
+   *        prefix length</a>
+   * @see <a href="https://datatracker.ietf.org/doc/html/rfc5280#section-4.2.1.6">RFC-5280 - SAN description</a>
+   * @see <a href="https://issues.apache.org/jira/browse/VALIDATOR-445">VALIDATOR-445 - Commons Validator change</a>
+   * @see <a href="https://github.com/bcgit/bc-java/issues/2024">BouncyCastle issue discussion about scoped IPv6
+   *        addresses</a>
+   */
+  public static boolean isScopedOrMaskingIPv6Address(InetAddress addr) {
+    if (addr instanceof Inet6Address) {
+      String hostAddress = addr.getHostAddress();
+      return hostAddress.contains("/") || hostAddress.contains("%");
+    }
+    return false;
+  }
+
   /**
    * Convert list of string encoded certificates to list of X509Certificate.
    * @param pemEncodedCerts

…/volume/StorageVolumeScannerMetrics.java …lume/BackgroundVolumeScannerMetrics.javahadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/volume/StorageVolumeScannerMetrics.java renamed to hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/volume/BackgroundVolumeScannerMetrics.java hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/volume/StorageVolumeScannerMetrics.java renamed to hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/volume/BackgroundVolumeScannerMetrics.java

@@ -26,12 +26,12 @@
 import org.apache.hadoop.metrics2.lib.MutableGaugeLong;
 
 /**
- * This class captures the Storage Volume Scanner Metrics.
+ * This class captures the Background Storage Volume Scanner Metrics.
  **/
 @InterfaceAudience.Private
-@Metrics(about = "Storage Volume Scanner Metrics", context = "dfs")
-public class StorageVolumeScannerMetrics {
-  public static final String SOURCE_NAME = StorageVolumeScannerMetrics.class.getSimpleName();
+@Metrics(about = "Background Volume Scanner Metrics", context = "dfs")
+public class BackgroundVolumeScannerMetrics {
+  public static final String SOURCE_NAME = BackgroundVolumeScannerMetrics.class.getSimpleName();
 
   @Metric("number of volumes scanned in the last iteration")
   private MutableGaugeLong numVolumesScannedInLastIteration;
@@ -49,12 +49,12 @@ public class StorageVolumeScannerMetrics {
       "since the last iteration had not elapsed")
   private MutableCounterLong numIterationsSkipped;
 
-  public StorageVolumeScannerMetrics() {
+  public BackgroundVolumeScannerMetrics() {
   }
 
-  public static StorageVolumeScannerMetrics create() {
+  public static BackgroundVolumeScannerMetrics create() {
     MetricsSystem ms = DefaultMetricsSystem.instance();
-    return ms.register(SOURCE_NAME, "Storage Volume Scanner Metrics", new StorageVolumeScannerMetrics());
+    return ms.register(SOURCE_NAME, "Background Volume Scanner Metrics", new BackgroundVolumeScannerMetrics());
   }
 
   /**

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

@@ -65,7 +65,7 @@ public class StorageVolumeChecker {
 
   private AsyncChecker<Boolean, VolumeCheckResult> delegateChecker;
 
-  private final StorageVolumeScannerMetrics metrics;
+  private final BackgroundVolumeScannerMetrics metrics;
 
   /**
    * Max allowed time for a disk check in milliseconds. If the check
@@ -105,7 +105,7 @@ public class StorageVolumeChecker {
   public StorageVolumeChecker(ConfigurationSource conf, Timer timer,
       String threadNamePrefix) {
 
-    metrics = StorageVolumeScannerMetrics.create();
+    metrics = BackgroundVolumeScannerMetrics.create();
 
     this.timer = timer;
 
@@ -441,7 +441,7 @@ void setDelegateChecker(
   }
 
   @VisibleForTesting
-  public StorageVolumeScannerMetrics getMetrics() {
+  public BackgroundVolumeScannerMetrics getMetrics() {
     return metrics;
   }
 }

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

@@ -79,7 +79,7 @@ public void testPeriodicVolumeChecker(TestInfo testInfo) throws Exception {
     FakeTimer timer = new FakeTimer();
 
     StorageVolumeChecker volumeChecker = new StorageVolumeChecker(conf, timer, "");
-    StorageVolumeScannerMetrics metrics = volumeChecker.getMetrics();
+    BackgroundVolumeScannerMetrics metrics = volumeChecker.getMetrics();
 
     try {
       volumeChecker.registerVolumeSet(new ImmutableVolumeSet(makeVolumes(2, HEALTHY)));

hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/balancer/ContainerBalancer.java

@@ -308,16 +308,13 @@ private void startBalancingThread(int nextIterationIndex,
   }
 
   /**
-   * Validates balancer's state based on the specified expectedRunning.
+   * Validates balancer's eligibility based on SCM state.
    * Confirms SCM is leader-ready and out of safe mode.
    *
-   * @param expectedRunning true if ContainerBalancer is expected to be
-   *                        running, else false
    * @throws IllegalContainerBalancerStateException if SCM is not
-   * leader-ready, is in safe mode, or state does not match the specified
-   * expected state
+   * leader-ready or is in safe mode
    */
-  private void validateState(boolean expectedRunning)
+  private void validateEligibility()
       throws IllegalContainerBalancerStateException {
     if (!scmContext.isLeaderReady()) {
       LOG.warn("SCM is not leader ready");
@@ -328,6 +325,19 @@ private void validateState(boolean expectedRunning)
       LOG.warn("SCM is in safe mode");
       throw new IllegalContainerBalancerStateException("SCM is in safe mode");
     }
+  }
+
+  /**
+   * Validates balancer's state based on the specified expectedRunning.
+   *
+   * @param expectedRunning true if ContainerBalancer is expected to be
+   *                        running, else false
+   * @throws IllegalContainerBalancerStateException if state does not
+   * match the specified expected state
+   */
+  private void validateState(boolean expectedRunning)
+      throws IllegalContainerBalancerStateException {
+    validateEligibility();
     if (!expectedRunning && !canBalancerStart()) {
       throw new IllegalContainerBalancerStateException(
           "Expect ContainerBalancer as not running state" +
@@ -387,18 +397,22 @@ private static void blockTillTaskStop(Thread balancingThread) {
    */
   public void stopBalancer()
       throws IOException, IllegalContainerBalancerStateException {
-    Thread balancingThread;
+    Thread balancingThread = null;
     lock.lock();
     try {
-      validateState(true);
+      validateEligibility();
       saveConfiguration(config, false, 0);
-      LOG.info("Trying to stop ContainerBalancer service.");
-      task.stop();
-      balancingThread = currentBalancingThread;
+      if (isBalancerRunning()) {
+        LOG.info("Trying to stop ContainerBalancer service.");
+        task.stop();
+        balancingThread = currentBalancingThread;
+      }
     } finally {
       lock.unlock();
     }
-    blockTillTaskStop(balancingThread);
+    if (balancingThread != null) {
+      blockTillTaskStop(balancingThread);
+    }
   }
 
   public void saveConfiguration(ContainerBalancerConfiguration configuration,

hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/balancer/TestContainerBalancer.java

@@ -20,6 +20,7 @@
 import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_NODE_REPORT_INTERVAL;
 import static org.apache.hadoop.hdds.HddsConfigKeys.HDDS_SCM_WAIT_TIME_AFTER_SAFE_MODE_EXIT;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertSame;
@@ -128,6 +129,8 @@ public void testShouldRun() throws Exception {
 
   @Test
   public void testStartBalancerStop() throws Exception {
+    //stop should not throw an exception as it is idempotent
+    assertDoesNotThrow(() -> containerBalancer.stopBalancer());
     startBalancer(balancerConfiguration);
     assertThrows(IllegalContainerBalancerStateException.class,
         () -> containerBalancer.startBalancer(balancerConfiguration),
@@ -142,9 +145,9 @@ public void testStartBalancerStop() throws Exception {
     stopBalancer();
     assertSame(ContainerBalancerTask.Status.STOPPED, containerBalancer.getBalancerStatus());
 
-    assertThrows(Exception.class,
-        () -> containerBalancer.stopBalancer(),
-        "Exception should be thrown when stop again");
+    // If the balancer is already stopped, the stop command should do nothing
+    // and return successfully as stopBalancer is idempotent
+    assertDoesNotThrow(() -> containerBalancer.stopBalancer());
   }
 
   @Test

hadoop-ozone/freon/src/main/java/org/apache/hadoop/ozone/freon/BaseFreonGenerator.java

@@ -26,8 +26,6 @@
 import io.opentelemetry.api.trace.StatusCode;
 import java.io.IOException;
 import java.io.InputStream;
-import java.time.Duration;
-import java.time.Instant;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.concurrent.ExecutorService;
@@ -327,8 +325,7 @@ public void init() {
     LongSupplier supplier;
     if (duration != null) {
       maxValue = durationInSecond;
-      supplier = () -> Duration.between(
-          Instant.ofEpochMilli(startTime), Instant.now()).getSeconds();
+      supplier = () -> (Time.monotonicNow() - startTime) / 1000;
     } else {
       maxValue = testNo;
       supplier = () -> successCounter.get() + failureCounter.get();

hadoop-ozone/integration-test-s3/src/test/java/org/apache/hadoop/ozone/s3/awssdk/v1/AbstractS3SDKV1Tests.java

@@ -97,6 +97,7 @@
 import java.util.Map;
 import java.util.Set;
 import java.util.stream.Collectors;
+import java.util.stream.IntStream;
 import javax.xml.bind.DatatypeConverter;
 import org.apache.commons.io.IOUtils;
 import org.apache.commons.lang3.RandomStringUtils;
@@ -128,6 +129,8 @@
 import org.junit.jupiter.api.TestInstance;
 import org.junit.jupiter.api.TestMethodOrder;
 import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
 
 /**
  * This is an abstract class to test the AWS Java S3 SDK operations.
@@ -141,6 +144,9 @@
 @TestMethodOrder(MethodOrderer.MethodName.class)
 public abstract class AbstractS3SDKV1Tests extends OzoneTestBase {
 
+  // server-side limitation
+  private static final int MAX_UPLOADS_LIMIT = 1000;
+
   /**
    * There are still some unsupported S3 operations.
    * Current unsupported S3 operations (non-exhaustive):
@@ -758,6 +764,7 @@ public void testListMultipartUploads() {
     uploadIds.add(uploadId3);
 
     ListMultipartUploadsRequest listMultipartUploadsRequest = new ListMultipartUploadsRequest(bucketName);
+    listMultipartUploadsRequest.setMaxUploads(5000);
 
     MultipartUploadListing result = s3Client.listMultipartUploads(listMultipartUploadsRequest);
 
@@ -768,26 +775,31 @@ public void testListMultipartUploads() {
     assertEquals(uploadIds, listUploadIds);
   }
 
-  @Test
-  public void testListMultipartUploadsPagination() {
-    final String bucketName = getBucketName();
+  @ParameterizedTest
+  @ValueSource(ints = {10, 5000})
+  public void testListMultipartUploadsPagination(int requestedMaxUploads) {
+    final String bucketName = getBucketName() + "-" + requestedMaxUploads;
     final String multipartKeyPrefix = getKeyName("multipart");
 
     s3Client.createBucket(bucketName);
 
-    // Create 25 multipart uploads to test pagination
+    // Create multipart uploads to test pagination
     List<String> allKeys = new ArrayList<>();
     Map<String, String> keyToUploadId = new HashMap<>();
 
-    for (int i = 0; i < 25; i++) {
-      String key = String.format("%s-%03d", multipartKeyPrefix, i);
+    final int effectiveMaxUploads = Math.min(requestedMaxUploads, MAX_UPLOADS_LIMIT);
+    final int uploadsCreated = 2 * effectiveMaxUploads + 5;
+    final int expectedPages = uploadsCreated / effectiveMaxUploads + 1;
+
+    for (int i = 0; i < uploadsCreated; i++) {
+      String key = String.format("%s-%04d", multipartKeyPrefix, i);
       allKeys.add(key);
       String uploadId = initiateMultipartUpload(bucketName, key, null, null, null);
       keyToUploadId.put(key, uploadId);
     }
     Collections.sort(allKeys);
 
-    // Test pagination with maxUploads=10
+    // Test pagination
     Set<String> retrievedKeys = new HashSet<>();
     String keyMarker = null;
     String uploadIdMarker = null;
@@ -796,18 +808,19 @@ public void testListMultipartUploadsPagination() {
 
     do {
       ListMultipartUploadsRequest request = new ListMultipartUploadsRequest(bucketName)
-          .withMaxUploads(10)
+          .withMaxUploads(requestedMaxUploads)
           .withKeyMarker(keyMarker)
           .withUploadIdMarker(uploadIdMarker);
 
       MultipartUploadListing result = s3Client.listMultipartUploads(request);
+      pageCount++;
 
       // Verify page size
-      if (pageCount < 2) {
-        assertEquals(10, result.getMultipartUploads().size());
+      if (pageCount < expectedPages) {
+        assertEquals(effectiveMaxUploads, result.getMultipartUploads().size());
         assertTrue(result.isTruncated());
       } else {
-        assertEquals(5, result.getMultipartUploads().size());
+        assertEquals(uploadsCreated % effectiveMaxUploads, result.getMultipartUploads().size());
         assertFalse(result.isTruncated());
       }
 
@@ -822,7 +835,7 @@ public void testListMultipartUploadsPagination() {
       assertNull(result.getPrefix());
       assertEquals(result.getUploadIdMarker(), uploadIdMarker);
       assertEquals(result.getKeyMarker(), keyMarker);
-      assertEquals(result.getMaxUploads(), 10);
+      assertEquals(effectiveMaxUploads, result.getMaxUploads());
 
       // Verify next markers content
       if (result.isTruncated()) {
@@ -840,32 +853,29 @@ public void testListMultipartUploadsPagination() {
       uploadIdMarker = result.getNextUploadIdMarker();
 
       truncated = result.isTruncated();
-      pageCount++;
 
     } while (truncated);
 
     // Verify pagination results
-    assertEquals(3, pageCount, "Should have exactly 3 pages");
-    assertEquals(25, retrievedKeys.size(), "Should retrieve all uploads");
+    assertEquals(expectedPages, pageCount);
+    assertEquals(uploadsCreated, retrievedKeys.size(), "Should retrieve all uploads");
     assertEquals(
         allKeys,
         retrievedKeys.stream().sorted().collect(Collectors.toList()),
         "Retrieved keys should match expected keys in order");
 
     // Test with prefix
-    String prefix = multipartKeyPrefix + "-01";
+    String prefix = multipartKeyPrefix + "-001";
     ListMultipartUploadsRequest prefixRequest = new ListMultipartUploadsRequest(bucketName)
         .withPrefix(prefix);
 
     MultipartUploadListing prefixResult = s3Client.listMultipartUploads(prefixRequest);
 
     assertEquals(prefix, prefixResult.getPrefix());
     assertEquals(
-        Arrays.asList(multipartKeyPrefix + "-010", multipartKeyPrefix + "-011",
-            multipartKeyPrefix + "-012", multipartKeyPrefix + "-013",
-            multipartKeyPrefix + "-014", multipartKeyPrefix + "-015",
-            multipartKeyPrefix + "-016", multipartKeyPrefix + "-017",
-            multipartKeyPrefix + "-018", multipartKeyPrefix + "-019"),
+        IntStream.rangeClosed(0, 9)
+            .mapToObj(i -> prefix + i)
+            .collect(Collectors.toList()),
         prefixResult.getMultipartUploads().stream()
             .map(MultipartUpload::getKey)
             .collect(Collectors.toList()));