Merge pull request #16626 from ywelsch/fix/http-publish-port

Add additional fallback to http.publish_port and restrict fallback to transport.publish_port

core/src/main/java/org/elasticsearch/common/settings/ClusterSettings.java

@@ -219,6 +219,9 @@ public final class ClusterSettings extends AbstractScopedSettings {
                     HttpTransportSettings.SETTING_HTTP_DETAILED_ERRORS_ENABLED,
                     HttpTransportSettings.SETTING_PIPELINING,
                     HttpTransportSettings.SETTING_CORS_ALLOW_ORIGIN,
+                    HttpTransportSettings.SETTING_HTTP_HOST,
+                    HttpTransportSettings.SETTING_HTTP_PUBLISH_HOST,
+                    HttpTransportSettings.SETTING_HTTP_BIND_HOST,
                     HttpTransportSettings.SETTING_HTTP_PORT,
                     HttpTransportSettings.SETTING_HTTP_PUBLISH_PORT,
                     HttpTransportSettings.SETTING_PIPELINING_MAX_EVENTS,

core/src/main/java/org/elasticsearch/http/HttpTransportSettings.java

@@ -47,7 +47,7 @@ public final class HttpTransportSettings {
     public static final Setting<List<String>> SETTING_HTTP_BIND_HOST = listSetting("http.bind_host", SETTING_HTTP_HOST, s -> s, false, Scope.CLUSTER);
 
     public static final Setting<PortsRange> SETTING_HTTP_PORT = new Setting<PortsRange>("http.port", "9200-9300", PortsRange::new, false, Scope.CLUSTER);
-    public static final Setting<Integer> SETTING_HTTP_PUBLISH_PORT = Setting.intSetting("http.publish_port", 0, 0, false, Scope.CLUSTER);
+    public static final Setting<Integer> SETTING_HTTP_PUBLISH_PORT = Setting.intSetting("http.publish_port", -1, -1, false, Scope.CLUSTER);
     public static final Setting<Boolean> SETTING_HTTP_DETAILED_ERRORS_ENABLED = Setting.boolSetting("http.detailed_errors.enabled", true, false, Scope.CLUSTER);
     public static final Setting<ByteSizeValue> SETTING_HTTP_MAX_CONTENT_LENGTH = Setting.byteSizeSetting("http.max_content_length", new ByteSizeValue(100, ByteSizeUnit.MB), false, Scope.CLUSTER) ;
     public static final Setting<ByteSizeValue> SETTING_HTTP_MAX_CHUNK_SIZE = Setting.byteSizeSetting("http.max_chunk_size", new ByteSizeValue(8, ByteSizeUnit.KB), false, Scope.CLUSTER) ;

core/src/main/java/org/elasticsearch/http/netty/NettyHttpServerTransport.java

@@ -19,6 +19,8 @@
 
 package org.elasticsearch.http.netty;
 
+import com.carrotsearch.hppc.IntHashSet;
+import com.carrotsearch.hppc.IntSet;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.component.AbstractLifecycleComponent;
 import org.elasticsearch.common.inject.Inject;
@@ -192,8 +194,6 @@ public class NettyHttpServerTransport extends AbstractLifecycleComponent<HttpSer
     protected final boolean detailedErrorsEnabled;
     protected final ThreadPool threadPool;
 
-    protected int publishPort;
-
     protected final boolean tcpNoDelay;
     protected final boolean tcpKeepAlive;
     protected final boolean reuseAddress;
@@ -237,7 +237,6 @@ public class NettyHttpServerTransport extends AbstractLifecycleComponent<HttpSer
         this.port = SETTING_HTTP_PORT.get(settings);
         this.bindHosts = SETTING_HTTP_BIND_HOST.get(settings).toArray(Strings.EMPTY_ARRAY);
         this.publishHosts = SETTING_HTTP_PUBLISH_HOST.get(settings).toArray(Strings.EMPTY_ARRAY);
-        this.publishPort = SETTING_HTTP_PUBLISH_PORT.get(settings);
         this.tcpNoDelay = SETTING_HTTP_TCP_NO_DELAY.get(settings);
         this.tcpKeepAlive = SETTING_HTTP_TCP_KEEP_ALIVE.get(settings);
         this.reuseAddress = SETTING_HTTP_TCP_REUSE_ADDRESS.get(settings);
@@ -312,7 +311,10 @@ public class NettyHttpServerTransport extends AbstractLifecycleComponent<HttpSer
         serverBootstrap.setOption("child.receiveBufferSizePredictorFactory", receiveBufferSizePredictorFactory);
         serverBootstrap.setOption("reuseAddress", reuseAddress);
         serverBootstrap.setOption("child.reuseAddress", reuseAddress);
+        this.boundAddress = createBoundHttpAddress();
+    }
 
+    private BoundTransportAddress createBoundHttpAddress() {
         // Bind and start to accept incoming connections.
         InetAddress hostAddresses[];
         try {
@@ -333,7 +335,16 @@ public class NettyHttpServerTransport extends AbstractLifecycleComponent<HttpSer
             throw new BindTransportException("Failed to resolve publish address", e);
         }
 
-        if (0 == publishPort) {
+        final int publishPort = resolvePublishPort(settings, boundAddresses, publishInetAddress);
+        final InetSocketAddress publishAddress = new InetSocketAddress(publishInetAddress, publishPort);
+        return new BoundTransportAddress(boundAddresses.toArray(new TransportAddress[boundAddresses.size()]), new InetSocketTransportAddress(publishAddress));
+    }
+
+    // package private for tests
+    static int resolvePublishPort(Settings settings, List<InetSocketTransportAddress> boundAddresses, InetAddress publishInetAddress) {
+        int publishPort = SETTING_HTTP_PUBLISH_PORT.get(settings);
+
+        if (publishPort < 0) {
             for (InetSocketTransportAddress boundAddress : boundAddresses) {
                 InetAddress boundInetAddress = boundAddress.address().getAddress();
                 if (boundInetAddress.isAnyLocalAddress() || boundInetAddress.equals(publishInetAddress)) {
@@ -343,13 +354,23 @@ public class NettyHttpServerTransport extends AbstractLifecycleComponent<HttpSer
             }
         }
 
-        if (0 == publishPort) {
+        // if no matching boundAddress found, check if there is a unique port for all bound addresses
-            throw new BindHttpException("Publish address [" + publishInetAddress + "] does not match any of the bound addresses [" + boundAddresses + "]");
+        if (publishPort < 0) {
+            final IntSet ports = new IntHashSet();
+            for (InetSocketTransportAddress boundAddress : boundAddresses) {
+                ports.add(boundAddress.getPort());
+            }
+            if (ports.size() == 1) {
+                publishPort = ports.iterator().next().value;
+            }
         }
 
-        final InetSocketAddress publishAddress = new InetSocketAddress(publishInetAddress, publishPort);
+        if (publishPort < 0) {
-        ;
+            throw new BindHttpException("Failed to auto-resolve http publish port, multiple bound addresses " + boundAddresses +
-        this.boundAddress = new BoundTransportAddress(boundAddresses.toArray(new TransportAddress[boundAddresses.size()]), new InetSocketTransportAddress(publishAddress));
+                " with distinct ports and none of them matched the publish address (" + publishInetAddress + "). " +
+                "Please specify a unique port by setting " + SETTING_HTTP_PORT.getKey() + " or " + SETTING_HTTP_PUBLISH_PORT.getKey());
+        }
+        return publishPort;
     }
 
     private CorsConfig buildCorsConfig(Settings settings) {

core/src/main/java/org/elasticsearch/transport/BindTransportException.java

@@ -35,4 +35,8 @@ public class BindTransportException extends TransportException {
     public BindTransportException(String message, Throwable cause) {
         super(message, cause);
     }
-}
+
+    public BindTransportException(String message) {
+        super(message);
+    }
+}

core/src/main/java/org/elasticsearch/transport/netty/NettyTransport.java

@@ -19,6 +19,8 @@
 
 package org.elasticsearch.transport.netty;
 
+import com.carrotsearch.hppc.IntHashSet;
+import com.carrotsearch.hppc.IntSet;
 import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.Version;
 import org.elasticsearch.cluster.node.DiscoveryNode;
@@ -535,8 +537,16 @@ public class NettyTransport extends AbstractLifecycleComponent<Transport> implem
             throw new BindTransportException("Failed to resolve publish address", e);
         }
 
+        final int publishPort = resolvePublishPort(name, settings, profileSettings, boundAddresses, publishInetAddress);
+        final TransportAddress publishAddress = new InetSocketTransportAddress(new InetSocketAddress(publishInetAddress, publishPort));
+        return new BoundTransportAddress(transportBoundAddresses, publishAddress);
+    }
+
+    // package private for tests
+    static int resolvePublishPort(String profileName, Settings settings, Settings profileSettings, List<InetSocketAddress> boundAddresses,
+                               InetAddress publishInetAddress) {
         int publishPort;
-        if (TransportSettings.DEFAULT_PROFILE.equals(name)) {
+        if (TransportSettings.DEFAULT_PROFILE.equals(profileName)) {
             publishPort = TransportSettings.PUBLISH_PORT.get(settings);
         } else {
             publishPort = profileSettings.getAsInt("publish_port", -1);
@@ -553,17 +563,25 @@ public class NettyTransport extends AbstractLifecycleComponent<Transport> implem
             }
         }
 
-        // if port still not matches, just take port of first bound address
+        // if no matching boundAddress found, check if there is a unique port for all bound addresses
         if (publishPort < 0) {
-            // TODO: In case of DEFAULT_PROFILE we should probably fail here, as publish address does not match any bound address
+            final IntSet ports = new IntHashSet();
-            // In case of a custom profile, we might use the publish address of the default profile
+            for (InetSocketAddress boundAddress : boundAddresses) {
-            publishPort = boundAddresses.get(0).getPort();
+                ports.add(boundAddress.getPort());
-            logger.warn("Publish port not found by matching publish address [{}] to bound addresses [{}], "
+            }
-                    + "falling back to port [{}] of first bound address", publishInetAddress, boundAddresses, publishPort);
+            if (ports.size() == 1) {
+                publishPort = ports.iterator().next().value;
+            }
         }
 
-        final TransportAddress publishAddress = new InetSocketTransportAddress(new InetSocketAddress(publishInetAddress, publishPort));
+        if (publishPort < 0) {
-        return new BoundTransportAddress(transportBoundAddresses, publishAddress);
+            String profileExplanation = TransportSettings.DEFAULT_PROFILE.equals(profileName) ? "" : " for profile " + profileName;
+            throw new BindTransportException("Failed to auto-resolve publish port" + profileExplanation + ", multiple bound addresses " +
+                boundAddresses + " with distinct ports and none of them matched the publish address (" + publishInetAddress + "). " +
+                "Please specify a unique port by setting " + TransportSettings.PORT.getKey() + " or " +
+                TransportSettings.PUBLISH_PORT.getKey());
+        }
+        return publishPort;
     }
 
     private void createServerBootstrap(String name, Settings settings) {

core/src/test/java/org/elasticsearch/http/netty/HttpPublishPortIT.java

@@ -1,59 +0,0 @@
-/*
- * Licensed to Elasticsearch under one or more contributor
- * license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright
- * ownership. Elasticsearch licenses this file to you under
- * the Apache License, Version 2.0 (the "License"); you may
- * not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *    http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied.  See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-package org.elasticsearch.http.netty;
-
-import org.elasticsearch.action.admin.cluster.node.info.NodeInfo;
-import org.elasticsearch.action.admin.cluster.node.info.NodesInfoResponse;
-import org.elasticsearch.common.network.NetworkModule;
-import org.elasticsearch.common.settings.Settings;
-import org.elasticsearch.common.transport.BoundTransportAddress;
-import org.elasticsearch.common.transport.InetSocketTransportAddress;
-import org.elasticsearch.http.HttpTransportSettings;
-import org.elasticsearch.test.ESIntegTestCase;
-import org.elasticsearch.test.ESIntegTestCase.ClusterScope;
-import org.elasticsearch.test.ESIntegTestCase.Scope;
-
-import static org.hamcrest.CoreMatchers.is;
-import static org.hamcrest.Matchers.arrayWithSize;
-import static org.hamcrest.Matchers.greaterThanOrEqualTo;
-import static org.hamcrest.Matchers.instanceOf;
-
-@ClusterScope(scope = Scope.SUITE, numDataNodes = 1)
-public class HttpPublishPortIT extends ESIntegTestCase {
-    @Override
-    protected Settings nodeSettings(int nodeOrdinal) {
-        return Settings.settingsBuilder()
-                .put(super.nodeSettings(nodeOrdinal))
-                .put(NetworkModule.HTTP_ENABLED.getKey(), true)
-                .put(HttpTransportSettings.SETTING_HTTP_PUBLISH_PORT.getKey(), 9080)
-                .build();
-    }
-
-    public void testHttpPublishPort() throws Exception {
-        NodesInfoResponse response = client().admin().cluster().prepareNodesInfo().clear().setHttp(true).get();
-        assertThat(response.getNodes(), arrayWithSize(greaterThanOrEqualTo(1)));
-        NodeInfo nodeInfo = response.getNodes()[0];
-
-        BoundTransportAddress address = nodeInfo.getHttp().address();
-        assertThat(address.publishAddress(), instanceOf(InetSocketTransportAddress.class));
-
-        InetSocketTransportAddress publishAddress = (InetSocketTransportAddress) address.publishAddress();
-        assertThat(publishAddress.address().getPort(), is(9080));
-    }
-}

core/src/test/java/org/elasticsearch/http/netty/NettyHttpPublishPortTests.java

@@ -0,0 +1,91 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.http.netty;
+
+import org.elasticsearch.common.network.NetworkUtils;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.transport.InetSocketTransportAddress;
+import org.elasticsearch.http.BindHttpException;
+import org.elasticsearch.http.HttpTransportSettings;
+import org.elasticsearch.test.ESTestCase;
+
+import java.net.UnknownHostException;
+import java.util.ArrayList;
+import java.util.List;
+
+import static java.net.InetAddress.getByName;
+import static java.util.Arrays.asList;
+import static org.elasticsearch.http.netty.NettyHttpServerTransport.resolvePublishPort;
+import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.equalTo;
+
+public class NettyHttpPublishPortTests extends ESTestCase {
+
+    public void testHttpPublishPort() throws Exception {
+        int boundPort = randomIntBetween(9000, 9100);
+        int otherBoundPort = randomIntBetween(9200, 9300);
+
+        int publishPort = resolvePublishPort(Settings.builder().put(HttpTransportSettings.SETTING_HTTP_PUBLISH_PORT.getKey(), 9080).build(),
+            randomAddresses(), getByName("127.0.0.2"));
+        assertThat("Publish port should be explicitly set to 9080", publishPort, equalTo(9080));
+
+        publishPort = resolvePublishPort(Settings.EMPTY, asList(address("127.0.0.1", boundPort), address("127.0.0.2", otherBoundPort)),
+            getByName("127.0.0.1"));
+        assertThat("Publish port should be derived from matched address", publishPort, equalTo(boundPort));
+
+        publishPort = resolvePublishPort(Settings.EMPTY, asList(address("127.0.0.1", boundPort), address("127.0.0.2", boundPort)),
+            getByName("127.0.0.3"));
+        assertThat("Publish port should be derived from unique port of bound addresses", publishPort, equalTo(boundPort));
+
+        try {
+            resolvePublishPort(Settings.EMPTY, asList(address("127.0.0.1", boundPort), address("127.0.0.2", otherBoundPort)),
+                getByName("127.0.0.3"));
+            fail("Expected BindHttpException as publish_port not specified and non-unique port of bound addresses");
+        } catch (BindHttpException e) {
+            assertThat(e.getMessage(), containsString("Failed to auto-resolve http publish port"));
+        }
+
+        publishPort = resolvePublishPort(Settings.EMPTY, asList(address("0.0.0.0", boundPort), address("127.0.0.2", otherBoundPort)),
+            getByName("127.0.0.1"));
+        assertThat("Publish port should be derived from matching wildcard address", publishPort, equalTo(boundPort));
+
+        if (NetworkUtils.SUPPORTS_V6) {
+            publishPort = resolvePublishPort(Settings.EMPTY, asList(address("0.0.0.0", boundPort), address("127.0.0.2", otherBoundPort)),
+                getByName("::1"));
+            assertThat("Publish port should be derived from matching wildcard address", publishPort, equalTo(boundPort));
+        }
+    }
+
+    private InetSocketTransportAddress address(String host, int port) throws UnknownHostException {
+        return new InetSocketTransportAddress(getByName(host), port);
+    }
+
+    private InetSocketTransportAddress randomAddress() throws UnknownHostException {
+        return address("127.0.0." + randomIntBetween(1, 100), randomIntBetween(9200, 9300));
+    }
+
+    private List<InetSocketTransportAddress> randomAddresses() throws UnknownHostException {
+        List<InetSocketTransportAddress> addresses = new ArrayList<>();
+        for (int i = 0; i < randomIntBetween(1, 5); i++) {
+            addresses.add(randomAddress());
+        }
+        return addresses;
+    }
+}

core/src/test/java/org/elasticsearch/transport/netty/NettyPublishPortTests.java

@@ -0,0 +1,110 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.transport.netty;
+
+import org.elasticsearch.common.network.NetworkUtils;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.transport.BindTransportException;
+import org.elasticsearch.transport.TransportSettings;
+
+import java.net.InetSocketAddress;
+import java.net.UnknownHostException;
+import java.util.ArrayList;
+import java.util.List;
+
+import static java.net.InetAddress.getByName;
+import static java.util.Arrays.asList;
+import static org.elasticsearch.transport.netty.NettyTransport.resolvePublishPort;
+import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.equalTo;
+
+public class NettyPublishPortTests extends ESTestCase {
+
+    public void testPublishPort() throws Exception {
+        int boundPort = randomIntBetween(9000, 9100);
+        int otherBoundPort = randomIntBetween(9200, 9300);
+
+        boolean useProfile = randomBoolean();
+        final String profile;
+        final Settings settings;
+        final Settings profileSettings;
+        if (useProfile) {
+            profile = "some_profile";
+            settings = randomBoolean() ? Settings.EMPTY : Settings.builder().put(TransportSettings.PUBLISH_PORT.getKey(), 9081).build();
+            profileSettings = Settings.builder().put("publish_port", 9080).build();
+        } else {
+            profile = TransportSettings.DEFAULT_PROFILE;
+            settings = Settings.builder().put(TransportSettings.PUBLISH_PORT.getKey(), 9081).build();
+            profileSettings = randomBoolean() ? Settings.EMPTY : Settings.builder().put("publish_port", 9080).build();;
+        }
+
+        int publishPort = resolvePublishPort(profile, settings, profileSettings,
+            randomAddresses(), getByName("127.0.0.2"));
+        assertThat("Publish port should be explicitly set", publishPort, equalTo(useProfile ? 9080 : 9081));
+
+        publishPort = resolvePublishPort(profile, Settings.EMPTY, Settings.EMPTY,
+            asList(address("127.0.0.1", boundPort), address("127.0.0.2", otherBoundPort)),
+            getByName("127.0.0.1"));
+        assertThat("Publish port should be derived from matched address", publishPort, equalTo(boundPort));
+
+        publishPort = resolvePublishPort(profile, Settings.EMPTY, Settings.EMPTY,
+            asList(address("127.0.0.1", boundPort), address("127.0.0.2", boundPort)),
+            getByName("127.0.0.3"));
+        assertThat("Publish port should be derived from unique port of bound addresses", publishPort, equalTo(boundPort));
+
+        try {
+            resolvePublishPort(profile, Settings.EMPTY, Settings.EMPTY,
+                asList(address("127.0.0.1", boundPort), address("127.0.0.2", otherBoundPort)),
+                getByName("127.0.0.3"));
+            fail("Expected BindTransportException as publish_port not specified and non-unique port of bound addresses");
+        } catch (BindTransportException e) {
+            assertThat(e.getMessage(), containsString("Failed to auto-resolve publish port"));
+        }
+
+        publishPort = resolvePublishPort(profile, Settings.EMPTY, Settings.EMPTY,
+            asList(address("0.0.0.0", boundPort), address("127.0.0.2", otherBoundPort)),
+            getByName("127.0.0.1"));
+        assertThat("Publish port should be derived from matching wildcard address", publishPort, equalTo(boundPort));
+
+        if (NetworkUtils.SUPPORTS_V6) {
+            publishPort = resolvePublishPort(profile, Settings.EMPTY, Settings.EMPTY,
+                asList(address("0.0.0.0", boundPort), address("127.0.0.2", otherBoundPort)),
+                getByName("::1"));
+            assertThat("Publish port should be derived from matching wildcard address", publishPort, equalTo(boundPort));
+        }
+    }
+
+    private InetSocketAddress address(String host, int port) throws UnknownHostException {
+        return new InetSocketAddress(getByName(host), port);
+    }
+
+    private InetSocketAddress randomAddress() throws UnknownHostException {
+        return address("127.0.0." + randomIntBetween(1, 100), randomIntBetween(9200, 9300));
+    }
+
+    private List<InetSocketAddress> randomAddresses() throws UnknownHostException {
+        List<InetSocketAddress> addresses = new ArrayList<>();
+        for (int i = 0; i < randomIntBetween(1, 5); i++) {
+            addresses.add(randomAddress());
+        }
+        return addresses;
+    }
+}