[REST] Add prefixed URI path

Allow a custom path to be specified when accessing the cluster.

close elastic#642

docs/src/reference/asciidoc/core/configuration.adoc

@@ -105,6 +105,10 @@ List of {es} nodes to connect to. When using {es} remotely, _do_ set this option
 `es.port` (default 9200)::
 Default HTTP/REST port used for connecting to {es} - this setting is applied to the nodes in `es.nodes` that do not have any port specified.
 
+added[2.2]
+`es.nodes.path.prefix` (default empty)::
+Prefix to add to _all_ requests made to {es}. Useful in environments where the cluster is proxied/routed under a certain path. For example, if the cluster is located at +someaddress:someport/custom/path/prefix+, one would set +es.nodes.path.prefix+ to +/custom/path/prefix+.
+
 [float]
 ==== Querying
 `es.query` (default none)::

mr/src/itest/java/org/elasticsearch/hadoop/integration/rest/ssl/SSLTests.java

@@ -34,13 +34,14 @@
 import org.junit.Test;
 import org.junit.rules.ExternalResource;
 
-import static org.junit.Assert.*;
+import static org.junit.Assert.assertEquals;
 
 import static org.elasticsearch.hadoop.cfg.ConfigurationOptions.*;
 
 public class SSLTests {
 
     private static int SSL_PORT;
+    private final String PREFIX = "prefixed_path";
 
     @ClassRule
     public static ExternalResource SSL_SERVER = new ExternalResource() {
@@ -75,6 +76,7 @@ public void setup() {
         cfg.setProperty(ES_NET_SSL_CERT_ALLOW_SELF_SIGNED, "true");
         cfg.setProperty(ES_NET_SSL_TRUST_STORE_LOCATION, "ssl/client.jks");
         cfg.setProperty(ES_NET_SSL_TRUST_STORE_PASS, "testpass");
+        cfg.setProperty(ES_NODES_PATH_PREFIX, PREFIX);
 
         transport = new CommonsHttpTransport(cfg.copy(), "localhost");
     }
@@ -92,6 +94,6 @@ public void testBasicSSLHandshake() throws Exception {
 
         Response execute = transport.execute(req);
         String content = IOUtils.asString(execute.body());
-        assertEquals(path, content);
+        assertEquals("/" + PREFIX + path, content);
     }
 }

mr/src/main/java/org/elasticsearch/hadoop/cfg/ConfigurationOptions.java

@@ -41,6 +41,10 @@ public interface ConfigurationOptions {
     String ES_PORT = "es.port";
     String ES_PORT_DEFAULT = "9200";
 
+    /** Elasticsearch prefix **/
+    String ES_NODES_PATH_PREFIX = "es.nodes.path.prefix";
+    String ES_NODES_PATH_PREFIX_DEFAULT = "";
+
     /** Elasticsearch index */
     String ES_RESOURCE = "es.resource";
     String ES_RESOURCE_READ = "es.resource.read";

mr/src/main/java/org/elasticsearch/hadoop/cfg/Settings.java

@@ -33,7 +33,7 @@
 import org.elasticsearch.hadoop.util.unit.TimeValue;
 
 import static org.elasticsearch.hadoop.cfg.ConfigurationOptions.*;
-import static org.elasticsearch.hadoop.cfg.InternalConfigurationOptions.*;
+import static org.elasticsearch.hadoop.cfg.InternalConfigurationOptions.INTERNAL_ES_TARGET_FIELDS;
 
 /**
  * Holder class containing the various configuration bits used by ElasticSearch Hadoop. Handles internally the fall back to defaults when looking for undefined, optional settings.
@@ -60,6 +60,10 @@ public boolean getNodesDiscovery() {
         return Booleans.parseBoolean(getProperty(ES_NODES_DISCOVERY), !getNodesWANOnly());
     }
 
+    public String getNodesPathPrefix() {
+        return getProperty(ES_NODES_PATH_PREFIX, ES_NODES_PATH_PREFIX_DEFAULT);
+    }
+
     public boolean getNodesDataOnly() {
         // by default, if not set, return a value compatible with the WAN setting (see above)
         return Booleans.parseBoolean(getProperty(ES_NODES_DATA_ONLY), !getNodesWANOnly() && !getNodesClientOnly());

mr/src/main/java/org/elasticsearch/hadoop/rest/commonshttp/CommonsHttpTransport.java

@@ -85,6 +85,8 @@ public class CommonsHttpTransport implements Transport, StatsAware {
     private HttpConnection conn;
     private String proxyInfo = "";
     private final String httpInfo;
+    private final boolean sslEnabled;
+    private final String pathPrefix;
     private final Settings settings;
 
     private static class ResponseInputStream extends DelegatingInputStream implements ReusableInputStream {
@@ -155,6 +157,10 @@ public void close() {
     public CommonsHttpTransport(Settings settings, String host) {
         this.settings = settings;
         httpInfo = host;
+        sslEnabled = settings.getNetworkSSLEnabled();
+
+        String pathPref = settings.getNodesPathPrefix();
+        pathPrefix = (StringUtils.hasText(pathPref) ? addLeadingSlashIfNeeded(StringUtils.trimWhitespace(pathPref)) : StringUtils.trimWhitespace(pathPref));
 
         HttpClientParams params = new HttpClientParams();
         params.setParameter(HttpMethodParams.RETRY_HANDLER, new DefaultHttpMethodRetryHandler(
@@ -180,7 +186,7 @@ public boolean retryMethod(HttpMethod method, IOException exception, int executi
         hostConfig = (HostConfiguration) authSettings[0];
 
         try {
-            hostConfig.setHost(new URI(escapeUri(host, settings.getNetworkSSLEnabled()), false));
+            hostConfig.setHost(new URI(escapeUri(host, sslEnabled), false));
         } catch (IOException ex) {
             throw new EsHadoopTransportException("Invalid target URI " + host, ex);
         }
@@ -200,7 +206,7 @@ public boolean retryMethod(HttpMethod method, IOException exception, int executi
     }
 
     private HostConfiguration setupSSLIfNeeded(Settings settings, HostConfiguration hostConfig) {
-        if (!settings.getNetworkSSLEnabled()) {
+        if (!sslEnabled) {
             return hostConfig;
         }
 
@@ -336,8 +342,8 @@ private HostConfiguration setupSocksProxy(Settings settings, HostConfiguration h
             // switch protocol
             // due to how HttpCommons work internally this dance is best to be kept as is
             //
-            String schema = settings.getNetworkSSLEnabled() ? "https" : "http";
-            int port = settings.getNetworkSSLEnabled() ? 443 : 80;
+            String schema = sslEnabled ? "https" : "http";
+            int port = sslEnabled ? 443 : 80;
             SocksSocketFactory socketFactory = new SocksSocketFactory(proxyHost, proxyPort, proxyUser, proxyPass);
             replaceProtocol(hostConfig, socketFactory, schema, port);
         }
@@ -393,10 +399,12 @@ public Response execute(Request request) throws IOException {
 
         CharSequence uri = request.uri();
         if (StringUtils.hasText(uri)) {
-            http.setURI(new URI(escapeUri(uri.toString(), settings.getNetworkSSLEnabled()), false));
+            http.setURI(new URI(escapeUri(uri.toString(), sslEnabled), false));
         }
+
         // NB: initialize the path _after_ the URI otherwise the path gets reset to /
-        http.setPath(prefixPath(request.path().toString()));
+        // add node prefix (if specified)
+        http.setPath(pathPrefix + addLeadingSlashIfNeeded(request.path().toString()));
 
         try {
             // validate new URI
@@ -467,7 +475,7 @@ private static String escapeUri(String uri, boolean ssl) {
         return escaped.contains("://") ? escaped : (ssl ? "https://" : "http://") + escaped;
     }
 
-    private static String prefixPath(String string) {
+    private static String addLeadingSlashIfNeeded(String string) {
         return string.startsWith("/") ? string : "/" + string;
     }