fix and enable repository-hdfs secure tests

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,17 @@ for (String integTestTaskName : ['integTestHa', 'integTestSecure', 'integTestSec
       enabled = false;
     }
 
+    if (integTestTaskName.contains("Secure")) {
+      if (integTestTaskName.contains("Ha")) {
+        dependsOn secureHaHdfsFixture
+      } else {
+        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

@@ -32,9 +32,14 @@
 import org.apache.hadoop.hdfs.MiniDFSNNTopology;
 import org.apache.hadoop.hdfs.server.namenode.ha.HATestUtil;
 import org.apache.hadoop.security.UserGroupInformation;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.io.File;
+import java.io.IOException;
 import java.lang.management.ManagementFactory;
+import java.net.InetSocketAddress;
+import java.net.ServerSocket;
 import java.net.URL;
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
@@ -44,12 +49,15 @@
 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
  * easily properly setup logging, avoid parsing JSON, etc.
  */
 public class MiniHDFS {
+    public static final Logger LOG =
+        LoggerFactory.getLogger(MiniHDFS.class.getName());
 
     private static String PORT_FILE_NAME = "ports";
     private static String PID_FILE_NAME = "pid";
@@ -96,9 +104,19 @@ public static void main(String[] args) throws Exception {
             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_IPC_ADDRESS_KEY, "0.0.0.0:" + getFreeSocketPort(secure));
+        cfg.set(DFSConfigKeys.DFS_DATANODE_HTTP_ADDRESS_KEY, "0.0.0.0:" + getFreeSocketPort(secure));
+        LOG.info("datanode address : " + cfg.get(DFSConfigKeys.DFS_DATANODE_ADDRESS_KEY));
+        LOG.info("datanode ipc address : " + cfg.get(DFSConfigKeys.DFS_DATANODE_IPC_ADDRESS_KEY));
+        LOG.info("datanode http address : " + cfg.get(DFSConfigKeys.DFS_DATANODE_HTTP_ADDRESS_KEY));
+
         UserGroupInformation.setConfiguration(cfg);
 
         MiniDFSCluster.Builder builder = new MiniDFSCluster.Builder(cfg);
+        builder.checkDataNodeAddrConfig(true);
         if (secure) {
             builder.nameNodePort(9998);
         } else {
@@ -173,6 +191,36 @@ 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 noOfTries = 0;
+        do {
+            try {
+                port = secure ? new Random().nextInt(1024) : 0;
+                ServerSocket s = new ServerSocket();
+                s.setReuseAddress(true);
+                s.bind(new InetSocketAddress(port));
+                port = s.getLocalPort();
+                s.close();
+                return port;
+            } catch (IOException e) {
+                // Could not get a free port, continue
+            }
+            noOfTries++;
+        } while (noOfTries < 100 && (port == 0) || (secure && port > 1024));
+        return port;
+    }
 }