merlimat closed pull request #1862: Fix REST APIs provided by Pulsar proxy
URL: https://github.com/apache/incubator-pulsar/pull/1862
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git
a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/AdminProxyHandler.java
b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/AdminProxyHandler.java
index dce91264b6..deb17d8066 100644
---
a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/AdminProxyHandler.java
+++
b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/AdminProxyHandler.java
@@ -18,29 +18,41 @@
*/
package org.apache.pulsar.proxy.server;
+import static org.apache.commons.lang3.StringUtils.isBlank;
+
import java.io.IOException;
+import java.net.URI;
import java.security.cert.X509Certificate;
import java.util.Objects;
+
import javax.net.ssl.SSLContext;
import javax.servlet.ServletException;
+import javax.servlet.http.HttpServletRequest;
+
import org.apache.pulsar.client.api.Authentication;
import org.apache.pulsar.client.api.AuthenticationDataProvider;
import org.apache.pulsar.client.api.AuthenticationFactory;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.common.util.SecurityUtility;
+import org.apache.pulsar.policies.data.loadbalancer.ServiceLookupData;
import org.eclipse.jetty.client.HttpClient;
import org.eclipse.jetty.proxy.AsyncProxyServlet;
import org.eclipse.jetty.util.ssl.SslContextFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-class AdminProxyHandler extends AsyncProxyServlet.Transparent {
+class AdminProxyHandler extends AsyncProxyServlet {
private static final Logger LOG =
LoggerFactory.getLogger(AdminProxyHandler.class);
private final ProxyConfiguration config;
+ private final BrokerDiscoveryProvider discoveryProvider;
+ private final String brokerWebServiceUrl;
- AdminProxyHandler(ProxyConfiguration config) {
+ AdminProxyHandler(ProxyConfiguration config, BrokerDiscoveryProvider
discoveryProvider) {
this.config = config;
+ this.discoveryProvider = discoveryProvider;
+ this.brokerWebServiceUrl = config.isTlsEnabledWithBroker() ?
config.getBrokerWebServiceURLTLS()
+ : config.getBrokerWebServiceURL();
}
@Override
@@ -103,4 +115,51 @@ protected HttpClient newHttpClient() {
// return an unauthenticated client, every request will fail.
return new HttpClient();
}
+
+ @Override
+ protected String rewriteTarget(HttpServletRequest request) {
+ StringBuilder url = new StringBuilder();
+
+ if (isBlank(brokerWebServiceUrl)) {
+ try {
+ ServiceLookupData availableBroker =
discoveryProvider.nextBroker();
+
+ if (config.isTlsEnabledWithBroker()) {
+ url.append(availableBroker.getWebServiceUrlTls());
+ } else {
+ url.append(availableBroker.getWebServiceUrl());
+ }
+
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("[{}:{}] Selected active broker is {}",
request.getRemoteAddr(), request.getRemotePort(),
+ url.toString());
+ }
+ } catch (Exception e) {
+ LOG.warn("[{}:{}] Failed to get next active broker {}",
request.getRemoteAddr(),
+ request.getRemotePort(), e.getMessage(), e);
+ return null;
+ }
+ } else {
+ url.append(brokerWebServiceUrl);
+ }
+
+ if (url.lastIndexOf("/") == url.length() - 1) {
+ url.deleteCharAt(url.lastIndexOf("/"));
+ }
+ url.append(request.getRequestURI());
+
+ String query = request.getQueryString();
+ if (query != null) {
+ url.append("?").append(query);
+ }
+
+ URI rewrittenUrl = URI.create(url.toString()).normalize();
+
+ if (!validateDestination(rewrittenUrl.getHost(),
rewrittenUrl.getPort())) {
+ return null;
+ }
+
+ return rewrittenUrl.toString();
+ }
+
}
diff --git
a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConfiguration.java
b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConfiguration.java
index dcd9580bda..4d027867cf 100644
---
a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConfiguration.java
+++
b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyConfiguration.java
@@ -44,6 +44,10 @@
private String brokerServiceURL;
private String brokerServiceURLTLS;
+ // These settings are unnecessary if `zookeeperServers` is specified
+ private String brokerWebServiceURL;
+ private String brokerWebServiceURLTLS;
+
// Port to use to server binary-proto request
private int servicePort = 6650;
// Port to use to server binary-proto-tls request
@@ -138,6 +142,22 @@ public void setBrokerServiceURL(String
discoveryServiceURL) {
this.brokerServiceURL = discoveryServiceURL;
}
+ public String getBrokerWebServiceURL() {
+ return brokerWebServiceURL;
+ }
+
+ public void setBrokerWebServiceURL(String brokerWebServiceURL) {
+ this.brokerWebServiceURL = brokerWebServiceURL;
+ }
+
+ public String getBrokerWebServiceURLTLS() {
+ return brokerWebServiceURLTLS;
+ }
+
+ public void setBrokerWebServiceURLTLS(String brokerWebServiceURLTLS) {
+ this.brokerWebServiceURLTLS = brokerWebServiceURLTLS;
+ }
+
public String getZookeeperServers() {
return zookeeperServers;
}
diff --git
a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyServiceStarter.java
b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyServiceStarter.java
index 77fcd77048..75adb161a3 100644
---
a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyServiceStarter.java
+++
b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/ProxyServiceStarter.java
@@ -105,7 +105,13 @@ public ProxyServiceStarter(String[] args) throws Exception
{
if ((isBlank(config.getBrokerServiceURL()) &&
isBlank(config.getBrokerServiceURLTLS()))
|| config.isAuthorizationEnabled()) {
checkArgument(!isEmpty(config.getZookeeperServers()),
"zookeeperServers must be provided");
- checkArgument(!isEmpty(config.getConfigurationStoreServers()),
"configurationStoreServers must be provided");
+ checkArgument(!isEmpty(config.getConfigurationStoreServers()),
+ "configurationStoreServers must be provided");
+ }
+
+ if ((!config.isTlsEnabledWithBroker() &&
isBlank(config.getBrokerWebServiceURL()))
+ || (config.isTlsEnabledWithBroker() &&
isBlank(config.getBrokerWebServiceURLTLS()))) {
+ checkArgument(!isEmpty(config.getZookeeperServers()),
"zookeeperServers must be provided");
}
java.security.Security.addProvider(new
org.bouncycastle.jce.provider.BouncyCastleProvider());
@@ -132,12 +138,11 @@ public ProxyServiceStarter(String[] args) throws
Exception {
server.addRestResources("/", VipStatus.class.getPackage().getName(),
VipStatus.ATTRIBUTE_STATUS_FILE_PATH,
config.getStatusFilePath());
- AdminProxyHandler adminProxyHandler = new AdminProxyHandler(config);
+ AdminProxyHandler adminProxyHandler = new AdminProxyHandler(config,
proxyService.getDiscoveryProvider());
ServletHolder servletHolder = new ServletHolder(adminProxyHandler);
servletHolder.setInitParameter("preserveHost", "true");
- servletHolder.setInitParameter("proxyTo",
config.getBrokerServiceURL());
- server.addServlet("/admin/*", servletHolder);
- server.addServlet("/lookup/*", servletHolder);
+ server.addServlet("/admin", servletHolder);
+ server.addServlet("/lookup", servletHolder);
// start web-service
server.start();
diff --git
a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/WebServer.java
b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/WebServer.java
index f7a4f28af1..e840b4299e 100644
--- a/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/WebServer.java
+++ b/pulsar-proxy/src/main/java/org/apache/pulsar/proxy/server/WebServer.java
@@ -100,13 +100,14 @@ public URI getServiceUri() {
return this.server.getURI();
}
- public void addServlet(String path, ServletHolder servletHolder) {
- addServlet(path, servletHolder, Collections.emptyList());
+ public void addServlet(String basePath, ServletHolder servletHolder) {
+ addServlet(basePath, servletHolder, Collections.emptyList());
}
- public void addServlet(String path, ServletHolder servletHolder,
List<Pair<String, Object>> attributes) {
+ public void addServlet(String basePath, ServletHolder servletHolder,
List<Pair<String, Object>> attributes) {
ServletContextHandler context = new
ServletContextHandler(ServletContextHandler.SESSIONS);
- context.addServlet(servletHolder, path);
+ context.setContextPath(basePath);
+ context.addServlet(servletHolder, "/*");
for (Pair<String, Object> attribute : attributes) {
context.setAttribute(attribute.getLeft(), attribute.getRight());
}
diff --git
a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/AuthedAdminProxyHandlerTest.java
b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/AuthedAdminProxyHandlerTest.java
index ec217cf64a..4d32a58231 100644
---
a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/AuthedAdminProxyHandlerTest.java
+++
b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/AuthedAdminProxyHandlerTest.java
@@ -19,18 +19,23 @@
package org.apache.pulsar.proxy.server;
import static org.mockito.Matchers.anyObject;
+import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.spy;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
+
import javax.servlet.http.HttpServletRequest;
+
import org.apache.bookkeeper.test.PortManager;
import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
import org.apache.pulsar.broker.authentication.AuthenticationProviderTls;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.impl.auth.AuthenticationTls;
+import org.apache.pulsar.policies.data.loadbalancer.LoadManagerReport;
+import org.apache.pulsar.policies.data.loadbalancer.LoadReport;
import org.eclipse.jetty.servlet.ServletHolder;
import org.mockito.Mockito;
import org.testng.Assert;
@@ -50,6 +55,7 @@
private final String configClusterName = "test";
private ProxyConfiguration proxyConfig = new ProxyConfiguration();
private WebServer webServer;
+ private BrokerDiscoveryProvider discoveryProvider;
private AdminProxyWrapper adminProxyHandler;
@BeforeMethod
@@ -109,12 +115,15 @@ protected void setup() throws Exception {
webServer = new WebServer(proxyConfig);
- adminProxyHandler = new AdminProxyWrapper(proxyConfig);
+ discoveryProvider = spy(new BrokerDiscoveryProvider(proxyConfig,
mockZooKeeperClientFactory));
+ LoadManagerReport report = new LoadReport(brokerUrl.toString(),
brokerUrlTls.toString(), null, null);
+ doReturn(report).when(discoveryProvider).nextBroker();
+
+ adminProxyHandler = new AdminProxyWrapper(proxyConfig,
discoveryProvider);
ServletHolder servletHolder = new ServletHolder(adminProxyHandler);
servletHolder.setInitParameter("preserveHost", "true");
- servletHolder.setInitParameter("proxyTo", brokerUrlTls.toString());
- webServer.addServlet("/admin/*", servletHolder);
- webServer.addServlet("/lookup/*", servletHolder);
+ webServer.addServlet("/admin", servletHolder);
+ webServer.addServlet("/lookup", servletHolder);
// start web-service
webServer.start();
@@ -142,20 +151,21 @@ public void testAuthenticatedProxy() throws Exception {
List<String> activeBrokers =
admin.brokers().getActiveBrokers(configClusterName);
Assert.assertEquals(activeBrokers.size(), 1);
- Assert.assertTrue(adminProxyHandler.rewriteCalled);
+ Assert.assertEquals(adminProxyHandler.rewrittenUrl,
String.format("%s/admin/v2/brokers/%s",
+ brokerUrlTls.toString(), configClusterName));
}
static class AdminProxyWrapper extends AdminProxyHandler {
- boolean rewriteCalled = false;
+ String rewrittenUrl;
- AdminProxyWrapper(ProxyConfiguration config) {
- super(config);
+ AdminProxyWrapper(ProxyConfiguration config, BrokerDiscoveryProvider
discoveryProvider) {
+ super(config, discoveryProvider);
}
@Override
protected String rewriteTarget(HttpServletRequest clientRequest) {
- rewriteCalled = true;
- return super.rewriteTarget(clientRequest);
+ rewrittenUrl = super.rewriteTarget(clientRequest);
+ return rewrittenUrl;
}
}
diff --git
a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/UnauthedAdminProxyHandlerTest.java
b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/UnauthedAdminProxyHandlerTest.java
index 2c90d10494..5e099ce70b 100644
---
a/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/UnauthedAdminProxyHandlerTest.java
+++
b/pulsar-proxy/src/test/java/org/apache/pulsar/proxy/server/UnauthedAdminProxyHandlerTest.java
@@ -18,15 +18,26 @@
*/
package org.apache.pulsar.proxy.server;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.spy;
+
import java.util.HashSet;
import java.util.List;
import java.util.Set;
+
import javax.servlet.http.HttpServletRequest;
import javax.validation.constraints.AssertTrue;
+import javax.ws.rs.client.Client;
+import javax.ws.rs.client.ClientBuilder;
+import javax.ws.rs.client.WebTarget;
+
import org.apache.bookkeeper.test.PortManager;
import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.common.configuration.VipStatus;
import org.eclipse.jetty.servlet.ServletHolder;
+import org.glassfish.jersey.client.ClientConfig;
+import org.glassfish.jersey.logging.LoggingFeature;
import org.testng.Assert;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
@@ -34,9 +45,11 @@
public class UnauthedAdminProxyHandlerTest extends MockedPulsarServiceBaseTest
{
private final String DUMMY_VALUE = "DUMMY_VALUE";
+ private final String STATUS_FILE_PATH =
"./src/test/resources/vip_status.html";
private ProxyConfiguration proxyConfig = new ProxyConfiguration();
- private AdminProxyWrapper adminProxyHandler;
private WebServer webServer;
+ private BrokerDiscoveryProvider discoveryProvider;
+ private AdminProxyWrapper adminProxyHandler;
@Override
@BeforeClass
@@ -54,19 +67,22 @@ protected void setup() throws Exception {
// start proxy service
proxyConfig.setServicePort(PortManager.nextFreePort());
proxyConfig.setWebServicePort(PortManager.nextFreePort());
- proxyConfig.setBrokerServiceURL(brokerUrl.toString());
-
+ proxyConfig.setBrokerWebServiceURL(brokerUrl.toString());
+ proxyConfig.setStatusFilePath(STATUS_FILE_PATH);
proxyConfig.setZookeeperServers(DUMMY_VALUE);
proxyConfig.setGlobalZookeeperServers(DUMMY_VALUE);
webServer = new WebServer(proxyConfig);
- adminProxyHandler = new AdminProxyWrapper(proxyConfig);
+ discoveryProvider = spy(new BrokerDiscoveryProvider(proxyConfig,
mockZooKeeperClientFactory));
+ adminProxyHandler = new AdminProxyWrapper(proxyConfig,
discoveryProvider);
ServletHolder servletHolder = new ServletHolder(adminProxyHandler);
servletHolder.setInitParameter("preserveHost", "true");
- servletHolder.setInitParameter("proxyTo", brokerUrl.toString());
- webServer.addServlet("/admin/*", servletHolder);
- webServer.addServlet("/lookup/*", servletHolder);
+ webServer.addServlet("/admin", servletHolder);
+ webServer.addServlet("/lookup", servletHolder);
+
+ webServer.addRestResources("/", VipStatus.class.getPackage().getName(),
+ VipStatus.ATTRIBUTE_STATUS_FILE_PATH,
proxyConfig.getStatusFilePath());
// start web-service
webServer.start();
@@ -87,22 +103,33 @@ public void testUnauthenticatedProxy() throws Exception {
.build();
List<String> activeBrokers =
admin.brokers().getActiveBrokers(configClusterName);
Assert.assertEquals(activeBrokers.size(), 1);
- Assert.assertTrue(adminProxyHandler.rewriteCalled);
+ Assert.assertEquals(adminProxyHandler.rewrittenUrl,
String.format("%s/admin/v2/brokers/%s",
+ brokerUrl.toString(), configClusterName));
+ }
+
+ @Test
+ public void testVipStatus() throws Exception {
+ Client client = ClientBuilder.newClient(new
ClientConfig().register(LoggingFeature.class));
+ WebTarget webTarget = client.target("http://127.0.0.1:"; +
proxyConfig.getWebServicePort())
+ .path("/status.html");
+ String response = webTarget.request().get(String.class);
+ Assert.assertEquals(response, "OK");
+ client.close();
}
static class AdminProxyWrapper extends AdminProxyHandler {
- boolean rewriteCalled = false;
+ String rewrittenUrl;
- AdminProxyWrapper(ProxyConfiguration config) {
- super(config);
+ AdminProxyWrapper(ProxyConfiguration config, BrokerDiscoveryProvider
discoveryProvider) {
+ super(config, discoveryProvider);
}
@Override
protected String rewriteTarget(HttpServletRequest clientRequest) {
- rewriteCalled = true;
- return super.rewriteTarget(clientRequest);
+ rewrittenUrl = super.rewriteTarget(clientRequest);
+ return rewrittenUrl;
}
}
-}
\ No newline at end of file
+}
diff --git a/pulsar-proxy/src/test/resources/vip_status.html
b/pulsar-proxy/src/test/resources/vip_status.html
new file mode 100644
index 0000000000..a0eb9fbf8c
--- /dev/null
+++ b/pulsar-proxy/src/test/resources/vip_status.html
@@ -0,0 +1,20 @@
+<!--
+
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF 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.
+
+-->
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
