Adding a readiness check before using services in tests

buildSrc/src/main/groovy/org/elasticsearch/hadoop/gradle/fixture/hadoop/HadoopClusterFormationTasks.groovy

@@ -485,12 +485,17 @@ class HadoopClusterFormationTasks {
                         resourceexists {
                             file(file: instance.pidFile.toString())
                         }
+                        or {
+                            instance.readinessCheckSocketAddress == null
+                            socket(server: instance.readinessCheckSocketAddress.hostName, port: instance.readinessCheckSocketAddress.port)
+                        }
                     }
                 }
             }
             // Timed out waiting for pidfiles or failures
             if (project.ant.properties.containsKey("failed${name}".toString())) {
-                waitFailed(project, instance, project.logger, "Failed to start hadoop cluster: timed out after ${waitSeconds} seconds")
+                waitFailed(project, instance, project.logger, "Failed to start ${instance.config.roleDescriptor.roleName()}: timed out " +
+                        "after ${waitSeconds} seconds")
             }
 
             // Check to see if there were any failed markers

buildSrc/src/main/groovy/org/elasticsearch/hadoop/gradle/fixture/hadoop/InstanceInfo.groovy

@@ -63,6 +63,8 @@ class InstanceInfo {
     /** the pid file the node will use */
     File pidFile
 
+    InetSocketAddress readinessCheckSocketAddress
+
     /** service home dir */
     File homeDir
 
@@ -151,6 +153,7 @@ class InstanceInfo {
         pidFile = new File(baseDir, config.getServiceDescriptor().pidFileName(config))
         homeDir = new File(baseDir, config.getServiceDescriptor().homeDirName(config))
         confDir = new File(homeDir, config.getServiceDescriptor().confDirName(config))
+        readinessCheckSocketAddress = config.serviceDescriptor.readinessCheckHostAndPort(config)
 
         configFiles = config.getServiceDescriptor().configFiles(config).collect { name -> new File(confDir, name) }
         configContents = config.getServiceDescriptor().collectConfigFilesContents(config)

buildSrc/src/main/groovy/org/elasticsearch/hadoop/gradle/fixture/hadoop/ServiceDescriptor.groovy

@@ -149,4 +149,11 @@ interface ServiceDescriptor {
      * home directory. If you need to perform set up operations with Gradle tasks, use configureSetupTasks.
      */
     Map<String, Object[]> defaultSetupCommands(InstanceConfiguration configuration)
+
+    /**
+     * This is a host/port that can be used to check if an instance of the service is ready to be used.
+     * @param configuration The configuration of the instance we want to get the socket address for
+     * @return An InetSocketAddress that can be used to check if the instance whose configuration is given is ready to be used
+     */
+    InetSocketAddress readinessCheckHostAndPort(InstanceConfiguration configuration)
 }

buildSrc/src/main/groovy/org/elasticsearch/hadoop/gradle/fixture/hadoop/services/HadoopServiceDescriptor.groovy

@@ -207,9 +207,9 @@ class HadoopServiceDescriptor implements ServiceDescriptor {
         } else if (RESOURCEMANAGER.equals(role)) {
             FileSettings yarnSite = configFileContents.get('yarn-site.xml')
             if ('HTTPS_ONLY' == yarnSite.get('yarn.http.policy')) {
-                return "https://${yarnSite.getOrDefault('yarn.resourcemanager.webapp.address', 'localhost:8090')}"
+                return "https://${yarnSite.getOrDefault('yarn.resourcemanager.webapp.https.address', 'localhost:8090')}"
             } else {
-                return "http://${yarnSite.getOrDefault('yarn.resourcemanager.webapp.https.address', 'localhost:8088')}"
+                return "http://${yarnSite.getOrDefault('yarn.resourcemanager.webapp.address', 'localhost:8088')}"
             }
         } else if (NODEMANAGER.equals(role)) {
             FileSettings yarnSite = configFileContents.get('yarn-site.xml')
@@ -227,6 +227,48 @@ class HadoopServiceDescriptor implements ServiceDescriptor {
         throw new UnsupportedOperationException("Unknown instance [${role.roleName()}]")
     }
 
+    @Override
+    InetSocketAddress readinessCheckHostAndPort(InstanceConfiguration configuration) {
+        RoleDescriptor role = configuration.roleDescriptor
+        SettingsContainer settingsContainer = configuration.getSettingsContainer()
+        if (NAMENODE.equals(role)) {
+            FileSettings hdfsSite = settingsContainer.getFile('hdfs-site.xml')
+            if ('HTTPS_ONLY' == hdfsSite.get('dfs.http.policy')) {
+                return getInetAddress(hdfsSite.getOrDefault('dfs.namenode.https-address', 'localhost:50470'))
+            } else {
+                return getInetAddress(hdfsSite.getOrDefault('dfs.namenode.http-address', 'localhost:50070'))
+            }
+        } else if (DATANODE.equals(role)) {
+            FileSettings hdfsSite = settingsContainer.getFile('hdfs-site.xml')
+            if ('HTTPS_ONLY' == hdfsSite.get('dfs.http.policy')) {
+                return getInetAddress(hdfsSite.getOrDefault('dfs.datanode.https-address', 'localhost:50475'))
+            } else {
+                return getInetAddress(hdfsSite.getOrDefault('dfs.datanode.http-address', 'localhost:50075'))
+            }
+        } else if (RESOURCEMANAGER.equals(role)) {
+            FileSettings yarnSite = settingsContainer.getFile('yarn-site.xml')
+            if ('HTTPS_ONLY' == yarnSite.get('yarn.http.policy')) {
+                return getInetAddress(yarnSite.getOrDefault('yarn.resourcemanager.webapp.https.address', 'localhost:8090'))
+            } else {
+                return getInetAddress(yarnSite.getOrDefault('yarn.resourcemanager.webapp.address', 'localhost:8088'))
+            }
+        } else if (NODEMANAGER.equals(role)) {
+            FileSettings yarnSite = settingsContainer.getFile('yarn-site.xml')
+            return getInetAddress(yarnSite.getOrDefault('yarn.nodemanager.webapp.address', 'localhost:8042'))
+        } else if (HISTORYSERVER.equals(role)) {
+            FileSettings mapredSite = settingsContainer.getFile('mapred-site.xml')
+            return getInetAddress(mapredSite.getOrDefault('mapreduce.jobhistory.webapp.address', 'localhost:19888'))
+        } else if (GATEWAY.equals(role)) {
+            return null // No web interface for Gateway
+        }
+        throw new UnsupportedOperationException("Unknown instance [${role.roleName()}]")
+    }
+
+    private static InetSocketAddress getInetAddress(String hostAndPort) {
+        String[] hostAndPortArray = hostAndPort.split(":")
+        return new InetSocketAddress(hostAndPortArray[0], Integer.valueOf(hostAndPortArray[1]))
+    }
+
     @Override
     List<String> startCommand(InstanceConfiguration configuration) {
         RoleDescriptor role = configuration.roleDescriptor

buildSrc/src/main/groovy/org/elasticsearch/hadoop/gradle/fixture/hadoop/services/HiveServiceDescriptor.groovy

@@ -27,6 +27,7 @@ import org.elasticsearch.hadoop.gradle.fixture.hadoop.RoleDescriptor
 import org.elasticsearch.hadoop.gradle.fixture.hadoop.ServiceDescriptor
 import org.elasticsearch.hadoop.gradle.fixture.hadoop.conf.InstanceConfiguration
 import org.elasticsearch.hadoop.gradle.fixture.hadoop.conf.ServiceConfiguration
+import org.elasticsearch.hadoop.gradle.fixture.hadoop.conf.SettingsContainer
 import org.gradle.api.file.DuplicatesStrategy
 import org.gradle.api.tasks.Copy
 import org.gradle.api.tasks.Delete
@@ -124,6 +125,17 @@ class HiveServiceDescriptor implements ServiceDescriptor {
         throw new UnsupportedOperationException("Unknown instance [${configuration.roleDescriptor.roleName()}]")
     }
 
+    @Override
+    InetSocketAddress readinessCheckHostAndPort(InstanceConfiguration configuration) {
+        if (HIVESERVER.equals(configuration.roleDescriptor)) {
+            FileSettings fileSettings = configuration.getSettingsContainer().getFile('hive-site.xml')
+            String host = fileSettings.getOrDefault('hive.server2.thrift.bind.host', 'localhost')
+            String port = fileSettings.getOrDefault('hive.server2.thrift.port', '10000')
+            return new InetSocketAddress(host, Integer.valueOf(port))
+        }
+        throw new UnsupportedOperationException("Unknown instance [${configuration.roleDescriptor.roleName()}]")
+    }
+
     @Override
     List<String> startCommand(InstanceConfiguration configuration) {
         // We specify the hive root logger to print to console via the hiveconf override.

buildSrc/src/main/groovy/org/elasticsearch/hadoop/gradle/fixture/hadoop/services/PigServiceDescriptor.groovy

@@ -119,6 +119,14 @@ class PigServiceDescriptor implements ServiceDescriptor {
         throw new UnsupportedOperationException("Unknown instance [${configuration.roleDescriptor.roleName()}]")
     }
 
+    @Override
+    InetSocketAddress readinessCheckHostAndPort(InstanceConfiguration configuration) {
+        if (GATEWAY.equals(configuration.roleDescriptor)) {
+            return null
+        }
+        throw new UnsupportedOperationException("Unknown instance [${configuration.roleDescriptor.roleName()}]")
+    }
+
     @Override
     List<String> startCommand(InstanceConfiguration configuration) {
         return ['']

buildSrc/src/main/groovy/org/elasticsearch/hadoop/gradle/fixture/hadoop/services/SparkYarnServiceDescriptor.groovy

@@ -127,6 +127,14 @@ class SparkYarnServiceDescriptor implements ServiceDescriptor {
         throw new UnsupportedOperationException("Unknown instance [${configuration.roleDescriptor.roleName()}]")
     }
 
+    @Override
+    InetSocketAddress readinessCheckHostAndPort(InstanceConfiguration configuration) {
+        if (GATEWAY.equals(configuration.roleDescriptor)) {
+            return null
+        }
+        throw new UnsupportedOperationException("Unknown instance [${configuration.roleDescriptor.roleName()}]")
+    }
+
     @Override
     List<String> startCommand(InstanceConfiguration configuration) {
         // No start command for gateway services