You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by zi...@apache.org on 2023/01/17 08:22:16 UTC
[pulsar] branch branch-2.9 updated: [improve][websocket][branch-2.9] Add ping support (#19245)
This is an automated email from the ASF dual-hosted git repository.
zixuan pushed a commit to branch branch-2.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-2.9 by this push:
new 965fcac52b5 [improve][websocket][branch-2.9] Add ping support (#19245)
965fcac52b5 is described below
commit 965fcac52b5424029b2ab71e4157c5e136032e48
Author: Zixuan Liu <no...@gmail.com>
AuthorDate: Tue Jan 17 16:22:09 2023 +0800
[improve][websocket][branch-2.9] Add ping support (#19245)
Signed-off-by: Zixuan Liu <no...@gmail.com>
---
.../websocket/proxy/ProxyIdleTimeoutTest.java | 105 ++++++++++++++++++++
.../pulsar/websocket/proxy/ProxyPingTest.java | 107 +++++++++++++++++++++
.../pulsar/websocket/AbstractWebSocketHandler.java | 33 +++++++
.../apache/pulsar/websocket/WebSocketService.java | 5 +
.../service/WebSocketProxyConfiguration.java | 3 +
5 files changed, 253 insertions(+)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyIdleTimeoutTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyIdleTimeoutTest.java
new file mode 100644
index 00000000000..71f8c0b6d86
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyIdleTimeoutTest.java
@@ -0,0 +1,105 @@
+/**
+ * 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.
+ */
+package org.apache.pulsar.websocket.proxy;
+
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.apache.pulsar.broker.BrokerTestUtil.spyWithClassAndConstructorArgs;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doReturn;
+import java.net.URI;
+import java.util.Optional;
+import java.util.concurrent.Future;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.client.api.ProducerConsumerBase;
+import org.apache.pulsar.metadata.impl.ZKMetadataStore;
+import org.apache.pulsar.websocket.WebSocketService;
+import org.apache.pulsar.websocket.service.ProxyServer;
+import org.apache.pulsar.websocket.service.WebSocketProxyConfiguration;
+import org.apache.pulsar.websocket.service.WebSocketServiceStarter;
+import org.awaitility.Awaitility;
+import org.eclipse.jetty.websocket.api.Session;
+import org.eclipse.jetty.websocket.client.ClientUpgradeRequest;
+import org.eclipse.jetty.websocket.client.WebSocketClient;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+@Test(groups = "websocket")
+@Slf4j
+public class ProxyIdleTimeoutTest extends ProducerConsumerBase {
+ protected String methodName;
+ private ProxyServer proxyServer;
+ private WebSocketService service;
+
+ private static final int TIME_TO_CHECK_BACKLOG_QUOTA = 5;
+
+ @BeforeMethod
+ public void setup() throws Exception {
+ conf.setBacklogQuotaCheckIntervalInSeconds(TIME_TO_CHECK_BACKLOG_QUOTA);
+
+ super.internalSetup();
+ super.producerBaseSetup();
+
+ WebSocketProxyConfiguration config = new WebSocketProxyConfiguration();
+ config.setWebServicePort(Optional.of(0));
+ config.setClusterName("test");
+ config.setConfigurationStoreServers(GLOBAL_DUMMY_VALUE);
+ config.setWebSocketSessionIdleTimeoutMillis(3 * 1000);
+ service = spyWithClassAndConstructorArgs(WebSocketService.class, config);
+ doReturn(new ZKMetadataStore(mockZooKeeperGlobal)).when(service).createMetadataStore(anyString(), anyInt());
+ proxyServer = new ProxyServer(config);
+ WebSocketServiceStarter.start(proxyServer, service);
+ log.info("Proxy Server Started");
+ }
+
+ @AfterMethod(alwaysRun = true)
+ protected void cleanup() throws Exception {
+ super.internalCleanup();
+ if (service != null) {
+ service.close();
+ }
+ if (proxyServer != null) {
+ proxyServer.stop();
+ }
+ log.info("Finished Cleaning Up Test setup");
+ }
+
+ @Test
+ public void testIdleTimeout() throws Exception {
+ String producerUri = "ws://localhost:" + proxyServer.getListenPortHTTP().get() +
+ "/ws/v2/producer/persistent/my-property/my-ns/my-topic1/";
+
+ URI produceUri = URI.create(producerUri);
+ WebSocketClient produceClient = new WebSocketClient();
+ SimpleProducerSocket produceSocket = new SimpleProducerSocket();
+
+ try {
+ produceClient.start();
+ ClientUpgradeRequest produceRequest = new ClientUpgradeRequest();
+ Future<Session> producerFuture = produceClient.connect(produceSocket, produceUri, produceRequest);
+ assertThat(producerFuture).succeedsWithin(2, SECONDS);
+ Session session = producerFuture.get();
+ Awaitility.await().during(5, SECONDS).untilAsserted(() -> assertThat(session.isOpen()).isFalse());
+ } finally {
+ produceClient.stop();
+ }
+ }
+}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPingTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPingTest.java
new file mode 100644
index 00000000000..6e0dfa46cfd
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/websocket/proxy/ProxyPingTest.java
@@ -0,0 +1,107 @@
+/**
+ * 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.
+ */
+package org.apache.pulsar.websocket.proxy;
+
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.apache.pulsar.broker.BrokerTestUtil.spyWithClassAndConstructorArgs;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doReturn;
+import java.net.URI;
+import java.util.Optional;
+import java.util.concurrent.Future;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.client.api.ProducerConsumerBase;
+import org.apache.pulsar.metadata.impl.ZKMetadataStore;
+import org.apache.pulsar.websocket.WebSocketService;
+import org.apache.pulsar.websocket.service.ProxyServer;
+import org.apache.pulsar.websocket.service.WebSocketProxyConfiguration;
+import org.apache.pulsar.websocket.service.WebSocketServiceStarter;
+import org.awaitility.Awaitility;
+import org.eclipse.jetty.websocket.api.Session;
+import org.eclipse.jetty.websocket.client.ClientUpgradeRequest;
+import org.eclipse.jetty.websocket.client.WebSocketClient;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+@Test(groups = "websocket")
+@Slf4j
+public class ProxyPingTest extends ProducerConsumerBase {
+ protected String methodName;
+
+ private ProxyServer proxyServer;
+ private WebSocketService service;
+
+ private static final int TIME_TO_CHECK_BACKLOG_QUOTA = 5;
+
+ @BeforeMethod
+ public void setup() throws Exception {
+ conf.setBacklogQuotaCheckIntervalInSeconds(TIME_TO_CHECK_BACKLOG_QUOTA);
+
+ super.internalSetup();
+ super.producerBaseSetup();
+
+ WebSocketProxyConfiguration config = new WebSocketProxyConfiguration();
+ config.setWebServicePort(Optional.of(0));
+ config.setClusterName("test");
+ config.setConfigurationStoreServers(GLOBAL_DUMMY_VALUE);
+ config.setWebSocketSessionIdleTimeoutMillis(3 * 1000);
+ config.setWebSocketPingDurationSeconds(2);
+ service = spyWithClassAndConstructorArgs(WebSocketService.class, config);
+ doReturn(new ZKMetadataStore(mockZooKeeperGlobal)).when(service).createMetadataStore(anyString(), anyInt());
+ proxyServer = new ProxyServer(config);
+ WebSocketServiceStarter.start(proxyServer, service);
+ log.info("Proxy Server Started");
+ }
+
+ @AfterMethod(alwaysRun = true)
+ protected void cleanup() throws Exception {
+ super.internalCleanup();
+ if (service != null) {
+ service.close();
+ }
+ if (proxyServer != null) {
+ proxyServer.stop();
+ }
+ log.info("Finished Cleaning Up Test setup");
+ }
+
+ @Test
+ public void testPing() throws Exception {
+ String producerUri = "ws://localhost:" + proxyServer.getListenPortHTTP().get() +
+ "/ws/v2/producer/persistent/my-property/my-ns/my-topic1/";
+
+ URI produceUri = URI.create(producerUri);
+ WebSocketClient produceClient = new WebSocketClient();
+ SimpleProducerSocket produceSocket = new SimpleProducerSocket();
+
+ try {
+ produceClient.start();
+ ClientUpgradeRequest produceRequest = new ClientUpgradeRequest();
+ Future<Session> producerFuture = produceClient.connect(produceSocket, produceUri, produceRequest);
+ assertThat(producerFuture).succeedsWithin(2, SECONDS);
+ Session session = producerFuture.get();
+ Awaitility.await().during(5, SECONDS).untilAsserted(() -> assertThat(session.isOpen()).isTrue());
+ } finally {
+ produceClient.stop();
+ }
+ }
+}
diff --git a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/AbstractWebSocketHandler.java b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/AbstractWebSocketHandler.java
index 20829732689..335d5d2e010 100644
--- a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/AbstractWebSocketHandler.java
+++ b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/AbstractWebSocketHandler.java
@@ -18,7 +18,12 @@
*/
package org.apache.pulsar.websocket;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Splitter;
+import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
import org.apache.commons.lang3.StringUtils;
import org.apache.pulsar.broker.authentication.AuthenticationDataHttps;
import org.apache.pulsar.broker.authentication.AuthenticationDataSource;
@@ -38,6 +43,7 @@ import org.apache.pulsar.client.api.PulsarClientException.TopicTerminatedExcepti
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.util.Codec;
+import org.apache.pulsar.websocket.service.WebSocketProxyConfiguration;
import org.eclipse.jetty.websocket.api.Session;
import org.eclipse.jetty.websocket.api.WebSocketAdapter;
import org.eclipse.jetty.websocket.servlet.ServletUpgradeResponse;
@@ -62,6 +68,7 @@ public abstract class AbstractWebSocketHandler extends WebSocketAdapter implemen
protected final TopicName topic;
protected final Map<String, String> queryParams;
+ private ScheduledFuture<?> pingFuture;
public AbstractWebSocketHandler(WebSocketService service, HttpServletRequest request, ServletUpgradeResponse response) {
this.service = service;
@@ -74,6 +81,12 @@ public abstract class AbstractWebSocketHandler extends WebSocketAdapter implemen
});
}
+ private void closePingFuture() {
+ if (pingFuture != null && !pingFuture.isDone()) {
+ pingFuture.cancel(true);
+ }
+ }
+
protected boolean checkAuth(ServletUpgradeResponse response) {
String authRole = "<none>";
if (service.isAuthenticationEnabled()) {
@@ -154,6 +167,19 @@ public abstract class AbstractWebSocketHandler extends WebSocketAdapter implemen
@Override
public void onWebSocketConnect(Session session) {
super.onWebSocketConnect(session);
+ WebSocketProxyConfiguration webSocketProxyConfig = service.getWebSocketProxyConfig();
+ if (webSocketProxyConfig != null) {
+ int webSocketPingDurationSeconds = webSocketProxyConfig.getWebSocketPingDurationSeconds();
+ if (webSocketPingDurationSeconds > 0) {
+ pingFuture = service.getExecutor().scheduleAtFixedRate(() -> {
+ try {
+ session.getRemote().sendPing(ByteBuffer.wrap("PING".getBytes(StandardCharsets.UTF_8)));
+ } catch (IOException e) {
+ log.warn("[{}] WebSocket send ping", getSession().getRemoteAddress(), e);
+ }
+ }, webSocketPingDurationSeconds, webSocketPingDurationSeconds, TimeUnit.SECONDS);
+ }
+ }
log.info("[{}] New WebSocket session on topic {}", session.getRemoteAddress(), topic);
}
@@ -162,6 +188,7 @@ public abstract class AbstractWebSocketHandler extends WebSocketAdapter implemen
super.onWebSocketError(cause);
log.info("[{}] WebSocket error on topic {} : {}", getSession().getRemoteAddress(), topic, cause.getMessage());
try {
+ closePingFuture();
close();
} catch (IOException e) {
log.error("Failed in closing WebSocket session for topic {} with error: {}", topic, e.getMessage());
@@ -173,6 +200,7 @@ public abstract class AbstractWebSocketHandler extends WebSocketAdapter implemen
log.info("[{}] Closed WebSocket session on topic {}. status: {} - reason: {}", getSession().getRemoteAddress(),
topic, statusCode, reason);
try {
+ closePingFuture();
close();
} catch (IOException e) {
log.warn("[{}] Failed to close handler for topic {}. ", getSession().getRemoteAddress(), topic, e);
@@ -241,6 +269,11 @@ public abstract class AbstractWebSocketHandler extends WebSocketAdapter implemen
return TopicName.get(domain, namespace, name);
}
+ @VisibleForTesting
+ public ScheduledFuture<?> getPingFuture() {
+ return pingFuture;
+ }
+
protected abstract Boolean isAuthorized(String authRole, AuthenticationDataSource authenticationData) throws Exception;
private static final Logger log = LoggerFactory.getLogger(AbstractWebSocketHandler.class);
diff --git a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/WebSocketService.java b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/WebSocketService.java
index 7a0f19bad10..fe0eb01125c 100644
--- a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/WebSocketService.java
+++ b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/WebSocketService.java
@@ -28,6 +28,7 @@ import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import javax.servlet.ServletException;
import javax.websocket.DeploymentException;
+import lombok.Getter;
import org.apache.bookkeeper.common.util.OrderedScheduler;
import org.apache.pulsar.broker.PulsarServerException;
import org.apache.pulsar.broker.ServiceConfiguration;
@@ -69,6 +70,9 @@ public class WebSocketService implements Closeable {
private MetadataStoreExtended configMetadataStore;
private ServiceConfiguration config;
+ @Getter
+ private WebSocketProxyConfiguration webSocketProxyConfig;
+
private ClusterData localCluster;
private final ConcurrentOpenHashMap<String, ConcurrentOpenHashSet<ProducerHandler>> topicProducerMap;
private final ConcurrentOpenHashMap<String, ConcurrentOpenHashSet<ConsumerHandler>> topicConsumerMap;
@@ -77,6 +81,7 @@ public class WebSocketService implements Closeable {
public WebSocketService(WebSocketProxyConfiguration config) {
this(createClusterData(config), PulsarConfigurationLoader.convertFrom(config));
+ this.webSocketProxyConfig = config;
}
public WebSocketService(ClusterData localCluster, ServiceConfiguration config) {
diff --git a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/service/WebSocketProxyConfiguration.java b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/service/WebSocketProxyConfiguration.java
index deb8ab3e032..b05b0cc0e32 100644
--- a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/service/WebSocketProxyConfiguration.java
+++ b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/service/WebSocketProxyConfiguration.java
@@ -134,6 +134,9 @@ public class WebSocketProxyConfiguration implements PulsarConfiguration {
@FieldContext(doc = "When this parameter is not empty, unauthenticated users perform as anonymousUserRole")
private String anonymousUserRole = null;
+ @FieldContext(doc = "Interval of time to sending the ping to keep alive. This value greater than 0 means enabled")
+ private int webSocketPingDurationSeconds = -1;
+
/***** --- TLS --- ****/
@Deprecated
private boolean tlsEnabled = false;