SONAR-8798 use TransportClient to check node is operational

package org.elasticsearch.client:transport and most of its dependencies into sonar-application.jar; do not open Elasticsearch's http port by default

8 files changed, 169 insertions, 42 deletions

diff --git a/server/sonar-process-monitor/pom.xml b/server/sonar-process-monitor/pom.xml
index e9b5b1fdcce..fd12a5dcd95 100644
--- a/server/sonar-process-monitor/pom.xml
+++ b/server/sonar-process-monitor/pom.xml
@@ -47,6 +47,15 @@
       <groupId>com.google.guava</groupId>
       <artifactId>guava</artifactId>
     </dependency>
+    <dependency>
+      <groupId>org.elasticsearch.client</groupId>
+      <artifactId>transport</artifactId>
+    </dependency>
+    <dependency>
+      <groupId>org.apache.logging.log4j</groupId>
+      <artifactId>log4j-api</artifactId>
+      <version>2.6.2</version>
+    </dependency>
 
     <dependency>
       <groupId>com.google.code.findbugs</groupId>
diff --git a/server/sonar-process-monitor/src/main/java/org/sonar/application/SchedulerImpl.java b/server/sonar-process-monitor/src/main/java/org/sonar/application/SchedulerImpl.java
index 3c95aa7f311..a62c0d36eae 100644
--- a/server/sonar-process-monitor/src/main/java/org/sonar/application/SchedulerImpl.java
+++ b/server/sonar-process-monitor/src/main/java/org/sonar/application/SchedulerImpl.java
@@ -107,7 +107,7 @@ public class SchedulerImpl implements Scheduler, ProcessEventListener, ProcessLi
   private void tryToStartEs() {
     SQProcess process = processesById.get(ProcessId.ELASTICSEARCH);
     if (process != null) {
-      tryToStartEsProcess(process, commandFactory::createEsCommand);
+      tryToStartEsProcess(process, () -> commandFactory.createEsCommand(settings));
     }
   }
 
diff --git a/server/sonar-process-monitor/src/main/java/org/sonar/application/process/CommandFactory.java b/server/sonar-process-monitor/src/main/java/org/sonar/application/process/CommandFactory.java
index d893b9f7c54..0e8f9a768b5 100644
--- a/server/sonar-process-monitor/src/main/java/org/sonar/application/process/CommandFactory.java
+++ b/server/sonar-process-monitor/src/main/java/org/sonar/application/process/CommandFactory.java
@@ -19,9 +19,11 @@
  */
 package org.sonar.application.process;
 
+import org.sonar.application.config.AppSettings;
+
 public interface CommandFactory {
 
-  EsCommand createEsCommand();
+  EsCommand createEsCommand(AppSettings settings);
 
   JavaCommand createWebCommand(boolean leader);
 
diff --git a/server/sonar-process-monitor/src/main/java/org/sonar/application/process/CommandFactoryImpl.java b/server/sonar-process-monitor/src/main/java/org/sonar/application/process/CommandFactoryImpl.java
index 1404e4c2d0e..3b358126445 100644
--- a/server/sonar-process-monitor/src/main/java/org/sonar/application/process/CommandFactoryImpl.java
+++ b/server/sonar-process-monitor/src/main/java/org/sonar/application/process/CommandFactoryImpl.java
@@ -52,22 +52,24 @@ public class CommandFactoryImpl implements CommandFactory {
   }
 
   @Override
-  public EsCommand createEsCommand() {
-    File homeDir = settings.getProps().nonNullValueAsFile(ProcessProperties.PATH_HOME);
+  public EsCommand createEsCommand(AppSettings settings) {
+    File homeDir = this.settings.getProps().nonNullValueAsFile(ProcessProperties.PATH_HOME);
     File executable = new File(homeDir, getExecutable());
     if (!executable.exists()) {
       throw new IllegalStateException("Cannot find elasticsearch binary");
     }
 
-    Map<String, String> settingsMap = new EsSettings(settings.getProps()).build();
+    Map<String, String> settingsMap = new EsSettings(this.settings.getProps()).build();
 
     EsCommand res = new EsCommand(ProcessId.ELASTICSEARCH)
       .setWorkDir(executable.getParentFile().getParentFile())
       .setExecutable(executable)
-      .setArguments(settings.getProps().rawProperties())
+      .setArguments(this.settings.getProps().rawProperties())
+      .setClusterName(settingsMap.get("cluster.name"))
+      .setHost(settingsMap.get("network.host"))
       // TODO add argument to specify log4j configuration file
       // TODO add argument to specify yaml configuration file
-      .setUrl("http://" + settingsMap.get("http.host") + ":" + settingsMap.get("http.port"));
+      .setPort(Integer.valueOf(settingsMap.get("transport.tcp.port")));
 
     settingsMap.entrySet().stream()
       .filter(entry -> !"path.home".equals(entry.getKey()))
diff --git a/server/sonar-process-monitor/src/main/java/org/sonar/application/process/EsCommand.java b/server/sonar-process-monitor/src/main/java/org/sonar/application/process/EsCommand.java
index 684df33df11..0eed2d2aebd 100644
--- a/server/sonar-process-monitor/src/main/java/org/sonar/application/process/EsCommand.java
+++ b/server/sonar-process-monitor/src/main/java/org/sonar/application/process/EsCommand.java
@@ -26,7 +26,9 @@ import org.sonar.process.ProcessId;
 
 public class EsCommand extends AbstractCommand<EsCommand> {
   private File executable;
-  private String url;
+  private String clusterName;
+  private String host;
+  private int port;
   private List<String> esOptions = new ArrayList<>();
 
   public EsCommand(ProcessId id) {
@@ -42,12 +44,30 @@ public class EsCommand extends AbstractCommand<EsCommand> {
     return this;
   }
 
-  public String getUrl() {
-    return url;
+  public String getClusterName() {
+    return clusterName;
   }
 
-  public EsCommand setUrl(String url) {
-    this.url = url;
+  public EsCommand setClusterName(String clusterName) {
+    this.clusterName = clusterName;
+    return this;
+  }
+
+  public String getHost() {
+    return host;
+  }
+
+  public EsCommand setHost(String host) {
+    this.host = host;
+    return this;
+  }
+
+  public int getPort() {
+    return port;
+  }
+
+  public EsCommand setPort(int port) {
+    this.port = port;
     return this;
   }
 
diff --git a/server/sonar-process-monitor/src/main/java/org/sonar/application/process/EsProcessMonitor.java b/server/sonar-process-monitor/src/main/java/org/sonar/application/process/EsProcessMonitor.java
index 66aa95d91a7..d5b39eac99d 100644
--- a/server/sonar-process-monitor/src/main/java/org/sonar/application/process/EsProcessMonitor.java
+++ b/server/sonar-process-monitor/src/main/java/org/sonar/application/process/EsProcessMonitor.java
@@ -19,16 +19,38 @@
  */
 package org.sonar.application.process;
 
-import java.io.IOException;
-import java.net.ConnectException;
+import com.google.common.net.HostAndPort;
+import io.netty.util.ThreadDeathWatcher;
+import io.netty.util.concurrent.GlobalEventExecutor;
+import java.net.InetAddress;
 import java.net.MalformedURLException;
-import java.net.URL;
-import java.net.URLConnection;
+import java.net.UnknownHostException;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
-import org.apache.commons.io.IOUtils;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
+import org.elasticsearch.action.admin.cluster.health.ClusterHealthRequest;
+import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
+import org.elasticsearch.client.transport.NoNodeAvailableException;
+import org.elasticsearch.client.transport.TransportClient;
+import org.elasticsearch.cluster.health.ClusterHealthStatus;
+import org.elasticsearch.common.network.NetworkModule;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.transport.InetSocketTransportAddress;
+import org.elasticsearch.common.transport.TransportAddress;
+import org.elasticsearch.transport.Netty4Plugin;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import static java.util.Collections.singletonList;
+import static java.util.Collections.unmodifiableList;
+import static org.elasticsearch.common.unit.TimeValue.timeValueSeconds;
+import static org.sonar.application.process.EsProcessMonitor.Status.CONNECTION_REFUSED;
+import static org.sonar.application.process.EsProcessMonitor.Status.GREEN;
+import static org.sonar.application.process.EsProcessMonitor.Status.KO;
+import static org.sonar.application.process.EsProcessMonitor.Status.RED;
+import static org.sonar.application.process.EsProcessMonitor.Status.YELLOW;
+
 public class EsProcessMonitor extends AbstractProcessMonitor {
   private static final Logger LOG = LoggerFactory.getLogger(EsProcessMonitor.class);
   private static final int WAIT_FOR_UP_DELAY_IN_MILLIS = 100;
@@ -36,11 +58,12 @@ public class EsProcessMonitor extends AbstractProcessMonitor {
 
   private final AtomicBoolean nodeUp = new AtomicBoolean(false);
   private final AtomicBoolean nodeOperational = new AtomicBoolean(false);
-  private final URL healthCheckURL;
+  private final EsCommand esCommand;
+  private AtomicReference<TransportClient> transportClient = new AtomicReference<>(null);
 
-  public EsProcessMonitor(Process process, String url) throws MalformedURLException {
+  public EsProcessMonitor(Process process, EsCommand esCommand) throws MalformedURLException {
     super(process);
-    this.healthCheckURL = new URL(url + "/_cluster/health?wait_for_status=yellow&timeout=30s");
+    this.esCommand = esCommand;
   }
 
   @Override
@@ -49,14 +72,17 @@ public class EsProcessMonitor extends AbstractProcessMonitor {
       return true;
     }
 
+    boolean flag = false;
     try {
-      boolean flag = checkOperational();
-      if (flag) {
-        nodeOperational.set(true);
-      }
+      flag = checkOperational();
     } catch (InterruptedException e) {
       LOG.trace("Interrupted while checking ES node is operational", e);
       Thread.currentThread().interrupt();
+    } finally {
+      if (flag) {
+        transportClient.set(null);
+        nodeOperational.set(true);
+      }
     }
     return nodeOperational.get();
   }
@@ -73,30 +99,97 @@ public class EsProcessMonitor extends AbstractProcessMonitor {
         status = checkStatus();
       }
     } while (!nodeUp.get() && i < WAIT_FOR_UP_TIMEOUT);
-    return status == Status.YELLOW || status == Status.GREEN;
+    return status == YELLOW || status == GREEN;
+  }
+
+  static class MinimalTransportClient extends TransportClient {
+
+    MinimalTransportClient(Settings settings) {
+      super(settings, Settings.EMPTY, unmodifiableList(singletonList(Netty4Plugin.class)));
+    }
+
+    @Override
+    public void close() {
+      super.close();
+      if (NetworkModule.TRANSPORT_TYPE_SETTING.exists(settings) == false
+          || NetworkModule.TRANSPORT_TYPE_SETTING.get(settings).equals(Netty4Plugin.NETTY_TRANSPORT_NAME)) {
+        try {
+          GlobalEventExecutor.INSTANCE.awaitInactivity(5, TimeUnit.SECONDS);
+        } catch (InterruptedException e) {
+          Thread.currentThread().interrupt();
+        }
+        try {
+          ThreadDeathWatcher.awaitInactivity(5, TimeUnit.SECONDS);
+        } catch (InterruptedException e) {
+          Thread.currentThread().interrupt();
+        }
+      }
+    }
+
   }
 
   private Status checkStatus() {
     try {
-      URLConnection urlConnection = healthCheckURL.openConnection();
-      urlConnection.connect();
-      String response = IOUtils.toString(urlConnection.getInputStream());
-      if (response.contains("\"status\":\"green\"")) {
-        return Status.GREEN;
-      } else if (response.contains("\"status\":\"yellow\"")) {
-        return Status.YELLOW;
-      } else if (response.contains("\"status\":\"red\"")) {
-        return Status.RED;
+      ClusterHealthResponse response = getTransportClient().admin().cluster()
+        .health(new ClusterHealthRequest().waitForStatus(ClusterHealthStatus.YELLOW).timeout(timeValueSeconds(30)))
+        .actionGet();
+      if (response.getStatus() == ClusterHealthStatus.GREEN) {
+        return GREEN;
+      }
+      if (response.getStatus() == ClusterHealthStatus.YELLOW) {
+        return YELLOW;
       }
-      return Status.KO;
-    } catch (ConnectException e) {
-      return Status.CONNECTION_REFUSED;
-    } catch (IOException e) {
-      LOG.error("Unexpected error occurred while checking ES node status using WebService API", e);
-      return Status.KO;
+      if (response.getStatus() == ClusterHealthStatus.RED) {
+        return RED;
+      }
+      return KO;
+    } catch (NoNodeAvailableException e) {
+      return CONNECTION_REFUSED;
+    } catch (Exception e) {
+      LOG.error("Failed to check status", e);
+      return KO;
+    }
+  }
+
+  private TransportClient getTransportClient() {
+    TransportClient res = this.transportClient.get();
+    if (res == null) {
+      res = buildTransportClient();
+      if (this.transportClient.compareAndSet(null, res)) {
+        return res;
+      }
+      return this.transportClient.get();
+    }
+    return res;
+  }
+
+  private TransportClient buildTransportClient() {
+    org.elasticsearch.common.settings.Settings.Builder esSettings = org.elasticsearch.common.settings.Settings.builder();
+
+    // mandatory property defined by bootstrap process
+    esSettings.put("cluster.name", esCommand.getClusterName());
+
+    TransportClient nativeClient = new MinimalTransportClient(esSettings.build());
+    HostAndPort host = HostAndPort.fromParts(esCommand.getHost(), esCommand.getPort());
+    addHostToClient(host, nativeClient);
+    if (LOG.isDebugEnabled()) {
+      LOG.debug("Connected to Elasticsearch node: [{}]", displayedAddresses(nativeClient));
+    }
+    return nativeClient;
+  }
+
+  private static void addHostToClient(HostAndPort host, TransportClient client) {
+    try {
+      client.addTransportAddress(new InetSocketTransportAddress(InetAddress.getByName(host.getHostText()), host.getPortOrDefault(9001)));
+    } catch (UnknownHostException e) {
+      throw new IllegalStateException("Can not resolve host [" + host + "]", e);
     }
   }
 
+  private static String displayedAddresses(TransportClient nativeClient) {
+    return nativeClient.transportAddresses().stream().map(TransportAddress::toString).collect(Collectors.joining(", "));
+  }
+
   enum Status {
     CONNECTION_REFUSED, KO, RED, YELLOW, GREEN
   }
diff --git a/server/sonar-process-monitor/src/main/java/org/sonar/application/process/ProcessLauncherImpl.java b/server/sonar-process-monitor/src/main/java/org/sonar/application/process/ProcessLauncherImpl.java
index d297b8a02a6..c0d5ef2acda 100644
--- a/server/sonar-process-monitor/src/main/java/org/sonar/application/process/ProcessLauncherImpl.java
+++ b/server/sonar-process-monitor/src/main/java/org/sonar/application/process/ProcessLauncherImpl.java
@@ -71,7 +71,7 @@ public class ProcessLauncherImpl implements ProcessLauncher {
 
       process = processBuilder.start();
 
-      return new EsProcessMonitor(process, esCommand.getUrl());
+      return new EsProcessMonitor(process, esCommand);
     } catch (Exception e) {
       // just in case
       if (process != null) {
diff --git a/server/sonar-process-monitor/src/test/java/org/sonar/application/SchedulerImplTest.java b/server/sonar-process-monitor/src/test/java/org/sonar/application/SchedulerImplTest.java
index a457afe3a4e..d8335fcdf93 100644
--- a/server/sonar-process-monitor/src/test/java/org/sonar/application/SchedulerImplTest.java
+++ b/server/sonar-process-monitor/src/test/java/org/sonar/application/SchedulerImplTest.java
@@ -34,6 +34,7 @@ import org.junit.rules.ExpectedException;
 import org.junit.rules.TestRule;
 import org.junit.rules.Timeout;
 import org.mockito.Mockito;
+import org.sonar.application.config.AppSettings;
 import org.sonar.application.config.TestAppSettings;
 import org.sonar.application.process.AbstractCommand;
 import org.sonar.application.process.CommandFactory;
@@ -309,7 +310,7 @@ public class SchedulerImplTest {
 
   private static class TestCommandFactory implements CommandFactory {
     @Override
-    public EsCommand createEsCommand() {
+    public EsCommand createEsCommand(AppSettings settings) {
       return ES_COMMAND;
     }