fix and enable repository-hdfs secure tests

Due to recent changes done for converting `repository-hdfs` to test
clusters (elastic#41252), the `integTestSecure*` tasks did not depend on
`secureHdfsFixture` which when running would fail as the fixture
would not be available. This commit adds the dependency of fixture
to the task.

The `secureHdfsFixture` is an AntFixture which is spawned process.
Internally it waits for 30 seconds for the resources to be made available.
For my local machine it took almost 45 seconds to be available so I have
added the wait time as an input to the AntFixture defaults to 30 seconds
 and set it to 60 seconds in case of secure hdfs fixture.

Another problem while running the `secureHdfsFixture` where it would fail
due to port not being privileged port (i.e system port, port < 1024).
By default datanode address key tries to find a free port and this would
later on fail in case it is running in a secure mode. To address this
in case of secure mode we find free port below 1024 and set it in the
config. The config `DFSConfigKeys.IGNORE_SECURE_PORTS_FOR_TESTING_KEY` is set to
`true` but it did not help.
https://fisheye.apache.org/browse/~br=branch-2.8.1/hadoop/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/SecureDataNodeStarter.java?hb=true#to140

The integ test for secure hdfs were disabled for long time and so
the changes done in elastic#42090 to fix the tests are also done in this commit.

buildSrc/src/main/groovy/org/elasticsearch/gradle/test/AntFixture.groovy

@@ -58,6 +58,9 @@ public class AntFixture extends AntTask implements Fixture {
     @Input
     boolean useShell = false
 
+    @Input
+    int maxWaitInSeconds = 30
+
     /**
      * A flag to indicate whether the fixture should be run in the foreground, or spawned.
      * It is protected so subclasses can override (eg RunTask).
@@ -128,7 +131,7 @@ public class AntFixture extends AntTask implements Fixture {
 
         String failedProp = "failed${name}"
         // first wait for resources, or the failure marker from the wrapper script
-        ant.waitfor(maxwait: '30', maxwaitunit: 'second', checkevery: '500', checkeveryunit: 'millisecond', timeoutproperty: failedProp) {
+        ant.waitfor(maxwait: maxWaitInSeconds, maxwaitunit: 'second', checkevery: '500', checkeveryunit: 'millisecond', timeoutproperty: failedProp) {
             or {
                 resourceexists {
                     file(file: failureMarker.toString())

plugins/repository-hdfs/build.gradle

@@ -91,13 +91,13 @@ for (String fixtureName : ['hdfsFixture', 'haHdfsFixture', 'secureHdfsFixture',
     dependsOn project.configurations.hdfsFixture, project(':test:fixtures:krb5kdc-fixture').tasks.postProcessFixture
     executable = new File(project.runtimeJavaHome, 'bin/java')
     env 'CLASSPATH', "${ -> project.configurations.hdfsFixture.asPath }"
+    maxWaitInSeconds 60
     onlyIf {  project(':test:fixtures:krb5kdc-fixture').buildFixture.enabled }
     waitCondition = { fixture, ant ->
       // the hdfs.MiniHDFS fixture writes the ports file when
       // it's ready, so we can just wait for the file to exist
       return fixture.portsFile.exists()
     }
-
     final List<String> miniHDFSArgs = []
 
     // If it's a secure fixture, then depend on Kerberos Fixture and principals + add the krb5conf to the JVM options
@@ -125,7 +125,7 @@ for (String fixtureName : ['hdfsFixture', 'haHdfsFixture', 'secureHdfsFixture',
   }
 }
 
-Set disabledIntegTestTaskNames = ['integTestSecure', 'integTestSecureHa']
+Set disabledIntegTestTaskNames = []
 
 for (String integTestTaskName : ['integTestHa', 'integTestSecure', 'integTestSecureHa']) {
   task "${integTestTaskName}"(type: RestIntegTestTask) {
@@ -136,10 +136,13 @@ for (String integTestTaskName : ['integTestHa', 'integTestSecure', 'integTestSec
       enabled = false;
     }
 
+    if (integTestTaskName.contains("Secure")) {
+      dependsOn secureHdfsFixture
+    }
+
     runner {
       if (integTestTaskName.contains("Secure")) {
         if (disabledIntegTestTaskNames.contains(integTestTaskName) == false) {
-            dependsOn secureHdfsFixture
             nonInputProperties.systemProperty "test.krb5.principal.es", "elasticsearch@${realm}"
             nonInputProperties.systemProperty "test.krb5.principal.hdfs", "hdfs/hdfs.build.elastic.co@${realm}"
             jvmArgs "-Djava.security.krb5.conf=${krb5conf}"

plugins/repository-hdfs/src/test/resources/rest-api-spec/test/secure_hdfs_repository/30_snapshot_get.yml

@@ -48,17 +48,17 @@
         repository: test_snapshot_get_repository
         snapshot: test_snapshot_get
 
-  - length: { snapshots: 1 }
-  - match: { snapshots.0.snapshot : test_snapshot_get }
+  - length: { responses.0.snapshots: 1 }
+  - match: { responses.0.snapshots.0.snapshot : test_snapshot_get }
 
   # List snapshot info
   - do:
       snapshot.get:
         repository: test_snapshot_get_repository
         snapshot: "*"
 
-  - length: { snapshots: 1 }
-  - match: { snapshots.0.snapshot : test_snapshot_get }
+  - length: { responses.0.snapshots: 1 }
+  - match: { responses.0.snapshots.0.snapshot : test_snapshot_get }
 
   # Remove our snapshot
   - do: 

plugins/repository-hdfs/src/test/resources/rest-api-spec/test/secure_hdfs_repository/30_snapshot_readonly.yml

@@ -23,7 +23,7 @@
         repository: test_snapshot_repository_ro
         snapshot: "_all"
 
-  - length: { snapshots: 1 }
+  - length: { responses.0.snapshots: 1 }
 
   # Remove our repository
   - do:

test/fixtures/hdfs-fixture/src/main/java/hdfs/MiniHDFS.java

@@ -34,7 +34,9 @@
 import org.apache.hadoop.security.UserGroupInformation;
 
 import java.io.File;
+import java.io.IOException;
 import java.lang.management.ManagementFactory;
+import java.net.ServerSocket;
 import java.net.URL;
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
@@ -44,6 +46,7 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
+import java.util.Random;
 
 /**
  * MiniHDFS test fixture. There is a CLI tool, but here we can
@@ -94,11 +97,16 @@ public static void main(String[] args) throws Exception {
             cfg.set(DFSConfigKeys.DFS_BLOCK_ACCESS_TOKEN_ENABLE_KEY, "true");
             cfg.set(DFSConfigKeys.IGNORE_SECURE_PORTS_FOR_TESTING_KEY, "true");
             cfg.set(DFSConfigKeys.DFS_ENCRYPT_DATA_TRANSFER_KEY, "true");
+            // If we ask port to be allocated automatically, this fails in case of secure hdfs setup
+            // it needs port to be privileged. See org.apache.hadoop.hdfs.server.datanode.SecureDataNodeStarter
+            cfg.set(DFSConfigKeys.DFS_DATANODE_ADDRESS_KEY, "0.0.0.0:" + getFreeSocketPort(secure));
+            cfg.set(DFSConfigKeys.DFS_DATANODE_HTTP_ADDRESS_KEY, "0.0.0.0:" + getFreeSocketPort(secure));
         }
 
         UserGroupInformation.setConfiguration(cfg);
 
         MiniDFSCluster.Builder builder = new MiniDFSCluster.Builder(cfg);
+        builder.checkDataNodeAddrConfig(true);
         if (secure) {
             builder.nameNodePort(9998);
         } else {
@@ -173,6 +181,35 @@ public static void main(String[] args) throws Exception {
         tmp = Files.createTempFile(baseDir, null, null);
         Files.write(tmp, portFileContent.getBytes(StandardCharsets.UTF_8));
         Files.move(tmp, baseDir.resolve(PORT_FILE_NAME), StandardCopyOption.ATOMIC_MOVE);
+
+        while(true) {
+            Thread.sleep(2000);
+        }
     }
 
+
+    /**
+     * Return a free port number. There is no guarantee it will remain free, so
+     * it should be used immediately.
+     * @param secure if {@code true} then returns free port less than 1024.
+     * @returns a free port if found or else returns 0
+     */
+    private static int getFreeSocketPort(boolean secure) {
+        int port = 0;
+        int tries = 0;
+        do {
+            try {
+                port = secure ? new Random().nextInt(1024) : 0;
+                ServerSocket s = new ServerSocket(port);
+                s.setReuseAddress(true);
+                port = s.getLocalPort();
+                s.close();
+                return port;
+            } catch (IOException e) {
+                // Could not get a free port, continue
+            }
+            tries++;
+        } while (tries < 100 && (port == 0 && !secure) || (secure && port > 1024));
+        return port;
+    }
 }