This is an automated email from the ASF dual-hosted git repository.
Aias00 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shenyu.git
The following commit(s) were added to refs/heads/master by this push:
new 6a70424535 fix(zookeeper): align instance registration and discovery
paths (#7446)
6a70424535 is described below
commit 6a70424535d9360ca8e891ff48d2089af99b8c62
Author: lymerin <[email protected]>
AuthorDate: Thu Oct 8 14:36:42 2026 +0800
fix(zookeeper): align instance registration and discovery paths (#7446)
Fixes #6525
---
shenyu-admin/pom.xml | 7 +
.../admin/discovery/APDiscoveryProcessor.java | 2 +-
.../discovery/AbstractDiscoveryProcessor.java | 12 +-
.../admin/discovery/DefaultDiscoveryProcessor.java | 2 +-
.../DiscoveryDataChangedEventSyncListener.java | 3 +
.../shenyu/admin/transfer/DiscoveryTransfer.java | 1 +
.../mappers/discovery-upstream-sqlmap.xml | 2 +-
.../ZookeeperClientDiscoveryIntegrationTest.java | 272 +++++++++++++++++++++
.../core/register/InstanceRegisterListener.java | 12 +-
.../src/main/resources/application.yml | 3 +-
.../ZookeeperInstanceRegisterRepository.java | 25 +-
.../ZookeeperInstanceRegisterRepositoryTest.java | 68 +++++-
12 files changed, 386 insertions(+), 23 deletions(-)
diff --git a/shenyu-admin/pom.xml b/shenyu-admin/pom.xml
index cd0806b1ec..333671776d 100644
--- a/shenyu-admin/pom.xml
+++ b/shenyu-admin/pom.xml
@@ -275,6 +275,13 @@
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.shenyu</groupId>
+ <artifactId>shenyu-client-core</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+
<dependency>
<groupId>com.squareup.okhttp3</groupId>
<artifactId>okhttp</artifactId>
diff --git
a/shenyu-admin/src/main/java/org/apache/shenyu/admin/discovery/APDiscoveryProcessor.java
b/shenyu-admin/src/main/java/org/apache/shenyu/admin/discovery/APDiscoveryProcessor.java
index 6533f94f8f..4dd086d574 100644
---
a/shenyu-admin/src/main/java/org/apache/shenyu/admin/discovery/APDiscoveryProcessor.java
+++
b/shenyu-admin/src/main/java/org/apache/shenyu/admin/discovery/APDiscoveryProcessor.java
@@ -73,7 +73,7 @@ public class APDiscoveryProcessor extends
AbstractDiscoveryProcessor {
}
});
cacheKey.add(key);
- super.addChangedEventListener(discoveryHandlerDTO.getDiscoveryId(),
discoveryDataChangedEventListener);
+ super.addChangedEventListener(discoveryHandlerDTO.getDiscoveryId(),
key, discoveryDataChangedEventListener);
DataChangedEvent dataChangedEvent = new
DataChangedEvent(ConfigGroupEnum.PROXY_SELECTOR, DataEventTypeEnum.CREATE,
Collections.singletonList(DiscoveryTransfer.INSTANCE.mapToData(proxySelectorDTO)));
publishEvent(dataChangedEvent);
diff --git
a/shenyu-admin/src/main/java/org/apache/shenyu/admin/discovery/AbstractDiscoveryProcessor.java
b/shenyu-admin/src/main/java/org/apache/shenyu/admin/discovery/AbstractDiscoveryProcessor.java
index 9606b3518c..578bcb85d7 100644
---
a/shenyu-admin/src/main/java/org/apache/shenyu/admin/discovery/AbstractDiscoveryProcessor.java
+++
b/shenyu-admin/src/main/java/org/apache/shenyu/admin/discovery/AbstractDiscoveryProcessor.java
@@ -258,7 +258,7 @@ public abstract class AbstractDiscoveryProcessor implements
DiscoveryProcessor,
* @param proxySelectorDTO proxySelectorDTO
*/
public void addDiscoverySyncDataListener(final DiscoveryHandlerDTO
discoveryHandlerDTO, final ProxySelectorDTO proxySelectorDTO) {
- final DataChangedEventListener changedEventListener =
this.getChangedEventListener(discoveryHandlerDTO.getDiscoveryId());
+ final DataChangedEventListener changedEventListener =
this.getChangedEventListener(discoveryHandlerDTO.getDiscoveryId(),
buildProxySelectorKey(discoveryHandlerDTO.getListenerNode()));
if (Objects.nonNull(changedEventListener)) {
DiscoverySyncData discoverySyncData = new DiscoverySyncData();
discoverySyncData.setPluginName(proxySelectorDTO.getPluginName());
@@ -300,20 +300,22 @@ public abstract class AbstractDiscoveryProcessor
implements DiscoveryProcessor,
* addChangedEventListener.
*
* @param discoveryId discoveryId
+ * @param key watched service key
* @param dataChangedEventListener dataChangedEventListener
*/
- public void addChangedEventListener(final String discoveryId, final
DataChangedEventListener dataChangedEventListener) {
- this.dataChangedEventListenerMap.put(discoveryId,
dataChangedEventListener);
+ public void addChangedEventListener(final String discoveryId, final String
key, final DataChangedEventListener dataChangedEventListener) {
+ this.dataChangedEventListenerMap.put(discoveryId + ":" + key,
dataChangedEventListener);
}
/**
* getChangedEventListener.
*
* @param discoveryId discoveryId
+ * @param key watched service key
* @return {@link DataChangedEventListener}
*/
- public DataChangedEventListener getChangedEventListener(final String
discoveryId) {
- return this.dataChangedEventListenerMap.get(discoveryId);
+ public DataChangedEventListener getChangedEventListener(final String
discoveryId, final String key) {
+ return this.dataChangedEventListenerMap.get(discoveryId + ":" + key);
}
/**
diff --git
a/shenyu-admin/src/main/java/org/apache/shenyu/admin/discovery/DefaultDiscoveryProcessor.java
b/shenyu-admin/src/main/java/org/apache/shenyu/admin/discovery/DefaultDiscoveryProcessor.java
index 533f116989..c6e68fbe0b 100644
---
a/shenyu-admin/src/main/java/org/apache/shenyu/admin/discovery/DefaultDiscoveryProcessor.java
+++
b/shenyu-admin/src/main/java/org/apache/shenyu/admin/discovery/DefaultDiscoveryProcessor.java
@@ -78,7 +78,7 @@ public class DefaultDiscoveryProcessor extends
AbstractDiscoveryProcessor {
}
});
cacheKey.add(key);
- super.addChangedEventListener(discoveryHandlerDTO.getDiscoveryId(),
discoveryDataChangedEventListener);
+ super.addChangedEventListener(discoveryHandlerDTO.getDiscoveryId(),
key, discoveryDataChangedEventListener);
DataChangedEvent dataChangedEvent = new
DataChangedEvent(ConfigGroupEnum.PROXY_SELECTOR, DataEventTypeEnum.CREATE,
Collections.singletonList(DiscoveryTransfer.INSTANCE.mapToData(proxySelectorDTO)));
publishEvent(dataChangedEvent);
diff --git
a/shenyu-admin/src/main/java/org/apache/shenyu/admin/discovery/DiscoveryDataChangedEventSyncListener.java
b/shenyu-admin/src/main/java/org/apache/shenyu/admin/discovery/DiscoveryDataChangedEventSyncListener.java
index d3ba6f0916..2b7b7c5b93 100644
---
a/shenyu-admin/src/main/java/org/apache/shenyu/admin/discovery/DiscoveryDataChangedEventSyncListener.java
+++
b/shenyu-admin/src/main/java/org/apache/shenyu/admin/discovery/DiscoveryDataChangedEventSyncListener.java
@@ -117,6 +117,7 @@ public class DiscoveryDataChangedEventSyncListener
implements DataChangedEventLi
case UPDATED:
upstreamDataList.stream().map(DiscoveryTransfer.INSTANCE::mapToDo).forEach(discoveryUpstreamDO
-> {
discoveryUpstreamDO.setDiscoveryHandlerId(discoveryHandlerId);
+ discoveryUpstreamDO.setDateUpdated(new
Timestamp(System.currentTimeMillis()));
int effect =
discoveryUpstreamMapper.updateDiscoveryHandlerIdAndUrl(discoveryUpstreamDO);
LOG.info("[DiscoveryDataChangedEventSyncListener] UPDATE
Upstream {}, effect = {} ", discoveryUpstreamDO.getUpstreamUrl(), effect);
});
@@ -132,6 +133,8 @@ public class DiscoveryDataChangedEventSyncListener
implements DataChangedEventLi
default:
throw new
IllegalStateException("DiscoveryDataChangedEventSyncListener find
IllegalState");
}
+
syncData.setUpstreamDataList(discoveryUpstreamMapper.selectByDiscoveryHandlerId(discoveryHandlerId).stream()
+
.map(DiscoveryTransfer.INSTANCE::mapToData).collect(Collectors.toList()));
DataChangedEvent dataChangedEvent = new
DataChangedEvent(ConfigGroupEnum.DISCOVER_UPSTREAM, DataEventTypeEnum.UPDATE,
Collections.singletonList(syncData));
eventPublisher.publishEvent(dataChangedEvent);
}
diff --git
a/shenyu-admin/src/main/java/org/apache/shenyu/admin/transfer/DiscoveryTransfer.java
b/shenyu-admin/src/main/java/org/apache/shenyu/admin/transfer/DiscoveryTransfer.java
index 91f854cd5f..5ba4bb9903 100644
---
a/shenyu-admin/src/main/java/org/apache/shenyu/admin/transfer/DiscoveryTransfer.java
+++
b/shenyu-admin/src/main/java/org/apache/shenyu/admin/transfer/DiscoveryTransfer.java
@@ -185,6 +185,7 @@ public enum DiscoveryTransfer {
return Optional.ofNullable(discoveryUpstreamDO).map(data -> {
DiscoveryUpstreamData discoveryUpstreamData = new
DiscoveryUpstreamData();
discoveryUpstreamData.setId(data.getId());
+ discoveryUpstreamData.setNamespaceId(data.getNamespaceId());
discoveryUpstreamData.setProtocol(data.getProtocol());
discoveryUpstreamData.setUrl(data.getUpstreamUrl());
discoveryUpstreamData.setStatus(data.getUpstreamStatus());
diff --git
a/shenyu-admin/src/main/resources/mappers/discovery-upstream-sqlmap.xml
b/shenyu-admin/src/main/resources/mappers/discovery-upstream-sqlmap.xml
index b79eb049de..ce46ef1a3c 100644
--- a/shenyu-admin/src/main/resources/mappers/discovery-upstream-sqlmap.xml
+++ b/shenyu-admin/src/main/resources/mappers/discovery-upstream-sqlmap.xml
@@ -123,7 +123,7 @@
weight=#{weight,jdbcType=INTEGER},
props=#{props,jdbcType=VARCHAR},
date_updated=#{dateUpdated, jdbcType=TIMESTAMP}
- WHERE discovery_handler_id=#{id, jdbcType=VARCHAR} and upstream_url
=#{upstreamUrl}
+ WHERE discovery_handler_id=#{discoveryHandlerId, jdbcType=VARCHAR} and
upstream_url =#{upstreamUrl}
</update>
<select id="selectAll" resultMap="BaseResultMap">
diff --git
a/shenyu-admin/src/test/java/org/apache/shenyu/admin/discovery/ZookeeperClientDiscoveryIntegrationTest.java
b/shenyu-admin/src/test/java/org/apache/shenyu/admin/discovery/ZookeeperClientDiscoveryIntegrationTest.java
new file mode 100644
index 0000000000..8b29361b4b
--- /dev/null
+++
b/shenyu-admin/src/test/java/org/apache/shenyu/admin/discovery/ZookeeperClientDiscoveryIntegrationTest.java
@@ -0,0 +1,272 @@
+/*
+ * 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.shenyu.admin.discovery;
+
+import com.google.gson.JsonObject;
+import jakarta.annotation.Resource;
+import org.apache.curator.test.TestingServer;
+import org.apache.shenyu.admin.AbstractSpringIntegrationTest;
+import org.apache.shenyu.admin.mapper.DiscoveryMapper;
+import org.apache.shenyu.admin.model.entity.DiscoveryDO;
+import
org.apache.shenyu.client.core.register.ClientDiscoveryConfigRefreshedEventListener;
+import org.apache.shenyu.client.core.register.ClientRegisterConfig;
+import org.apache.shenyu.client.core.register.ClientRegisterConfigImpl;
+import org.apache.shenyu.client.core.register.InstanceRegisterListener;
+import org.apache.shenyu.common.constant.Constants;
+import org.apache.shenyu.common.dto.DiscoverySyncData;
+import org.apache.shenyu.common.dto.DiscoveryUpstreamData;
+import org.apache.shenyu.common.enums.ConfigGroupEnum;
+import org.apache.shenyu.common.enums.PluginEnum;
+import org.apache.shenyu.common.enums.RpcTypeEnum;
+import org.apache.shenyu.common.utils.GsonUtils;
+import org.apache.shenyu.register.client.http.HttpClientRegisterRepository;
+import org.apache.shenyu.register.common.config.ShenyuClientConfig;
+import org.apache.shenyu.register.common.config.ShenyuDiscoveryConfig;
+import org.apache.shenyu.register.common.config.ShenyuRegisterCenterConfig;
+import org.apache.shenyu.registry.api.ShenyuInstanceRegisterRepository;
+import org.apache.shenyu.registry.api.entity.InstanceEntity;
+import org.apache.shenyu.registry.api.path.InstancePathConstants;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.test.web.server.LocalServerPort;
+import org.springframework.context.support.GenericApplicationContext;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.test.context.TestPropertySource;
+import org.springframework.test.util.ReflectionTestUtils;
+
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.WebSocket;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.CompletionStage;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Registers through the real client HTTP sender, Admin and ZooKeeper.
+ */
+@TestPropertySource(properties = "shenyu.sync.websocket.token=zk-6525-sync")
+public class ZookeeperClientDiscoveryIntegrationTest extends
AbstractSpringIntegrationTest {
+
+ @LocalServerPort
+ private int adminPort;
+
+ @Resource
+ private JdbcTemplate jdbcTemplate;
+
+ @Resource
+ private DiscoveryProcessorHolder processorHolder;
+
+ @Resource
+ private DiscoveryMapper discoveryMapper;
+
+ private final List<GenericApplicationContext> clients = new ArrayList<>();
+
+ private final List<DiscoverySyncData> snapshots = new
CopyOnWriteArrayList<>();
+
+ private WebSocket websocket;
+
+ @AfterEach
+ public void cleanup() {
+ if (Objects.nonNull(websocket)) {
+ websocket.abort();
+ }
+ clients.forEach(this::stopClient);
+ List<DiscoveryDO> discoveries = registeredDiscoveries();
+ discoveries.forEach(discovery ->
processorHolder.chooseProcessor("zookeeper").removeDiscovery(discovery));
+ jdbcTemplate.update("DELETE FROM discovery_upstream WHERE
discovery_handler_id IN "
+ + "(SELECT discovery_handler_id FROM discovery_rel WHERE
selector_id LIKE 'zk-6525-%')");
+ jdbcTemplate.update("DELETE FROM discovery_handler WHERE id IN (SELECT
discovery_handler_id FROM discovery_rel WHERE selector_id LIKE 'zk-6525-%')");
+ jdbcTemplate.update("DELETE FROM discovery_rel WHERE selector_id LIKE
'zk-6525-%'");
+ discoveries.forEach(discovery -> jdbcTemplate.update("DELETE FROM
discovery WHERE id = ?", discovery.getId()));
+ jdbcTemplate.update("DELETE FROM selector WHERE id LIKE 'zk-6525-%'");
+ }
+
+ @Test
+ public void testClientRegistrationAndInstanceChanges() throws Exception {
+ websocket =
HttpClient.newHttpClient().newWebSocketBuilder().header(Constants.SHENYU_NAMESPACE_ID,
Constants.SYS_DEFAULT_NAMESPACE_ID)
+ .header(Constants.X_SHENYU_SYNC_TOKEN, "zk-6525-sync")
+ .header(Constants.CLIENT_PORT_NAME, "19194")
+ .buildAsync(URI.create("ws://127.0.0.1:" + adminPort +
"/websocket"), new WebSocket.Listener() {
+ private final StringBuilder message = new StringBuilder();
+
+ @Override
+ public CompletionStage<?> onText(final WebSocket socket,
final CharSequence data, final boolean last) {
+ message.append(data);
+ if (last) {
+ JsonObject frame =
GsonUtils.getInstance().fromJson(message.toString(), JsonObject.class);
+ if
(ConfigGroupEnum.DISCOVER_UPSTREAM.name().equals(frame.get("groupType").getAsString()))
{
+
snapshots.addAll(GsonUtils.getInstance().fromList(frame.get("data").toString(),
DiscoverySyncData.class));
+ }
+ message.setLength(0);
+ }
+ socket.request(1);
+ return null;
+ }
+ }).join();
+ try (TestingServer server = new TestingServer()) {
+ createSelector("a");
+ createSelector("b");
+ GenericApplicationContext first = startClient(server, "a", 19195,
"/shenyu/discovery/http_example");
+ awaitUpstreams("a", List.of("127.0.0.1:19195"));
+
assertEquals(InstancePathConstants.buildInstanceParentPath("zk-6525-a"),
listenerNode("a"));
+ ShenyuDiscoveryConfig config =
first.getBean(ShenyuDiscoveryConfig.class);
+ assertNull(config.getProps().getProperty("watchPath"));
+ ShenyuInstanceRegisterRepository reader = reader();
+ assertTrue(reader.serviceExists(config.getRegisterPath()));
+ assertTrue(reader.serviceExists("zk-6525-a"));
+ assertEquals(1,
reader.selectInstances(config.getRegisterPath()).size());
+ assertFalse(reader.serviceExists("/shenyu/discovery"));
+
assertFalse(reader.serviceExists("/shenyu/discovery/http_example"));
+
assertTrue(reader.selectInstances("/shenyu/discovery/http_example").isEmpty());
+ final GenericApplicationContext other = startClient(server, "b",
19196, null);
+ awaitUpstreams("b", List.of("127.0.0.1:19196"));
+
assertEquals(InstancePathConstants.buildInstanceParentPath("zk-6525-b"),
listenerNode("b"));
+ final GenericApplicationContext second = startClient(server, "a",
19197, null);
+ awaitUpstreams("a", List.of("127.0.0.1:19195", "127.0.0.1:19197"));
+ assertSyncSnapshot("a", List.of("127.0.0.1:19195",
"127.0.0.1:19197"));
+ InstanceEntity update =
InstanceEntity.builder().appName("zk-6525-a").host("127.0.0.1").port(19197).build();
+ update.setStatus(0);
+ update.setWeight(80);
+ writer(second).persistInstance(update);
+ await().atMost(Duration.ofSeconds(15)).untilAsserted(() ->
assertEquals(80,
+ jdbcTemplate.queryForObject("SELECT weight FROM
discovery_upstream WHERE upstream_url = '127.0.0.1:19197'", Integer.class)));
+ assertSyncPublished("a", "127.0.0.1:19197", 80);
+ // Discovery configuration registration is drained by the Admin
subscriber every two seconds.
+ Thread.sleep(3000);
+ final GenericApplicationContext otherSecond = startClient(server,
"b", 19198, null);
+ awaitUpstreams("b", List.of("127.0.0.1:19196", "127.0.0.1:19198"));
+ awaitUpstreams("a", List.of("127.0.0.1:19195", "127.0.0.1:19197"));
+ stopClient(otherSecond);
+ awaitUpstreams("b", List.of("127.0.0.1:19196"));
+ stopClient(second);
+ awaitUpstreams("a", List.of("127.0.0.1:19195"));
+ assertSyncSnapshot("a", List.of("127.0.0.1:19195"));
+ awaitUpstreams("b", List.of("127.0.0.1:19196"));
+ stopClient(first);
+ awaitUpstreams("a", List.of());
+ assertSyncSnapshot("a", List.of());
+ awaitUpstreams("b", List.of("127.0.0.1:19196"));
+ stopClient(other);
+ awaitUpstreams("b", List.of());
+ registeredDiscoveries().forEach(discovery ->
processorHolder.chooseProcessor("zookeeper").removeDiscovery(discovery));
+ }
+ }
+
+ private GenericApplicationContext startClient(final TestingServer server,
final String app, final int port, final String registerPath) {
+ GenericApplicationContext context = new GenericApplicationContext();
+ clients.add(context);
+ ShenyuClientConfig client = new ShenyuClientConfig();
+ client.setNamespace(Constants.SYS_DEFAULT_NAMESPACE_ID);
+ ShenyuClientConfig.ClientPropertiesConfig properties = new
ShenyuClientConfig.ClientPropertiesConfig();
+ properties.getProps().setProperty("contextPath", "/zk-6525-" + app);
+ properties.getProps().setProperty("appName", "zk-6525-" + app);
+ properties.getProps().setProperty("host", "127.0.0.1");
+ properties.getProps().setProperty("port", String.valueOf(port));
+ client.getClient().put("http", properties);
+ final ClientRegisterConfig register = new
ClientRegisterConfigImpl(client, RpcTypeEnum.HTTP, context,
context.getEnvironment());
+ ShenyuDiscoveryConfig discovery = new ShenyuDiscoveryConfig();
+ discovery.setType("zookeeper");
+ discovery.setServerList(server.getConnectString());
+ discovery.setRegisterPath(registerPath);
+ discovery.getProps().setProperty("name", register.getAppName());
+ DiscoveryUpstreamData upstream = new DiscoveryUpstreamData();
+ upstream.setUrl(register.getHost() + ":" + register.getPort());
+ upstream.setProtocol("http://";);
+ upstream.setStatus(0);
+ upstream.setWeight(50);
+ ShenyuRegisterCenterConfig admin = new ShenyuRegisterCenterConfig();
+ admin.setServerLists("http://127.0.0.1:"; + adminPort);
+ admin.getProps().setProperty("username", "admin");
+ admin.getProps().setProperty("password", "123456");
+ HttpClientRegisterRepository sender = new
HttpClientRegisterRepository(admin);
+ context.registerBean(ShenyuDiscoveryConfig.class, () -> discovery);
+ context.registerBean(InstanceRegisterListener.class, () -> new
InstanceRegisterListener(upstream, discovery));
+ context.registerBean(ClientDiscoveryConfigRefreshedEventListener.class,
+ () -> new
ClientDiscoveryConfigRefreshedEventListener(discovery, sender, register,
PluginEnum.DIVIDE, client));
+ context.refresh();
+ return context;
+ }
+
+ private void stopClient(final GenericApplicationContext context) {
+ if (!context.isActive()) {
+ return;
+ }
+ InstanceRegisterListener listener =
context.getBean(InstanceRegisterListener.class);
+ writer(context).close();
+ ReflectionTestUtils.setField(listener, "discoveryService", null);
+ context.close();
+ }
+
+ private ShenyuInstanceRegisterRepository writer(final
GenericApplicationContext context) {
+ return (ShenyuInstanceRegisterRepository)
ReflectionTestUtils.getField(context.getBean(InstanceRegisterListener.class),
"discoveryService");
+ }
+
+ private ShenyuInstanceRegisterRepository reader() {
+ String id = jdbcTemplate.queryForObject("SELECT discovery_id FROM
discovery_handler WHERE listener_node = ?", String.class, listenerNode("a"));
+ return ((AbstractDiscoveryProcessor)
processorHolder.chooseProcessor("zookeeper")).getShenyuDiscoveryService(id);
+ }
+
+ private List<DiscoveryDO> registeredDiscoveries() {
+ return jdbcTemplate.queryForList("SELECT DISTINCT h.discovery_id FROM
discovery_handler h JOIN discovery_rel r ON r.discovery_handler_id = h.id "
+ + "WHERE r.selector_id LIKE 'zk-6525-%'",
String.class).stream().map(discoveryMapper::selectById).toList();
+ }
+
+ private void createSelector(final String app) {
+ jdbcTemplate.update("INSERT INTO selector (id, plugin_id,
selector_name, match_mode, selector_type, sort_code, "
+ + "enabled, loged, continued, match_restful, namespace_id) "
+ + "VALUES (?, '5', ?, 0, 0, 0, 1, 0, 0, 0, ?)", "zk-6525-" +
app, "/zk-6525-" + app, Constants.SYS_DEFAULT_NAMESPACE_ID);
+ }
+
+ private String listenerNode(final String app) {
+ return jdbcTemplate.queryForObject("SELECT listener_node FROM
discovery_handler h JOIN discovery_rel r ON r.discovery_handler_id = h.id WHERE
r.selector_id = ?",
+ String.class, "zk-6525-" + app);
+ }
+
+ private void awaitUpstreams(final String app, final List<String> expected)
{
+ await().atMost(Duration.ofSeconds(15)).untilAsserted(() ->
assertEquals(expected,
+ jdbcTemplate.queryForList("SELECT upstream_url FROM
discovery_upstream u JOIN discovery_rel r ON r.discovery_handler_id =
u.discovery_handler_id "
+ + "WHERE r.selector_id = ? ORDER BY upstream_url",
String.class, "zk-6525-" + app)));
+ }
+
+ private void assertSyncPublished(final String app, final String url, final
int weight) {
+ await().atMost(Duration.ofSeconds(15)).untilAsserted(() ->
assertTrue(snapshots.stream()
+ .filter(data -> ("zk-6525-" +
app).equals(data.getSelectorId()))
+ .flatMap(data ->
data.getUpstreamDataList().stream()).anyMatch(upstream ->
url.equals(upstream.getUrl()) && upstream.getWeight() == weight)));
+ }
+
+ private void assertSyncSnapshot(final String app, final List<String>
expected) {
+ await().atMost(Duration.ofSeconds(15)).untilAsserted(() -> {
+ List<DiscoverySyncData> appSnapshots = snapshots.stream()
+ .filter(data -> ("zk-6525-" +
app).equals(data.getSelectorId())).toList();
+ assertFalse(appSnapshots.isEmpty());
+ DiscoverySyncData snapshot = appSnapshots.get(appSnapshots.size()
- 1);
+ assertEquals(expected,
snapshot.getUpstreamDataList().stream().map(DiscoveryUpstreamData::getUrl).sorted().toList());
+ assertEquals(Constants.SYS_DEFAULT_NAMESPACE_ID,
snapshot.getNamespaceId());
+ snapshot.getUpstreamDataList().forEach(upstream ->
assertEquals(snapshot.getNamespaceId(), upstream.getNamespaceId()));
+ });
+ }
+}
diff --git
a/shenyu-client/shenyu-client-core/src/main/java/org/apache/shenyu/client/core/register/InstanceRegisterListener.java
b/shenyu-client/shenyu-client-core/src/main/java/org/apache/shenyu/client/core/register/InstanceRegisterListener.java
index b3d0fc076c..35e03eddae 100644
---
a/shenyu-client/shenyu-client-core/src/main/java/org/apache/shenyu/client/core/register/InstanceRegisterListener.java
+++
b/shenyu-client/shenyu-client-core/src/main/java/org/apache/shenyu/client/core/register/InstanceRegisterListener.java
@@ -24,6 +24,7 @@ import
org.apache.shenyu.register.common.config.ShenyuDiscoveryConfig;
import org.apache.shenyu.registry.api.ShenyuInstanceRegisterRepository;
import org.apache.shenyu.registry.api.config.RegisterConfig;
import org.apache.shenyu.registry.api.entity.InstanceEntity;
+import org.apache.shenyu.registry.api.path.InstancePathConstants;
import org.apache.shenyu.spi.ExtensionLoader;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -50,6 +51,8 @@ public class InstanceRegisterListener implements
ApplicationListener<ContextRefr
private final RegisterConfig discoveryConfig;
+ private final ShenyuDiscoveryConfig shenyuDiscoveryConfig;
+
private ShenyuInstanceRegisterRepository discoveryService;
private final String path;
@@ -61,6 +64,7 @@ public class InstanceRegisterListener implements
ApplicationListener<ContextRefr
this.discoveryConfig.setServerLists(shenyuDiscoveryConfig.getServerList());
this.discoveryConfig.setRegisterType(shenyuDiscoveryConfig.getType());
this.discoveryConfig.setProps(Optional.ofNullable(shenyuDiscoveryConfig.getProps()).orElse(new
Properties()));
+ this.shenyuDiscoveryConfig = shenyuDiscoveryConfig;
this.path = shenyuDiscoveryConfig.getRegisterPath();
Runtime.getRuntime().addShutdownHook(new Thread(() -> {
LOGGER.info("unregister upstream server by jvm runtime hook");
@@ -77,7 +81,10 @@ public class InstanceRegisterListener implements
ApplicationListener<ContextRefr
return;
}
this.discoveryService =
ExtensionLoader.getExtensionLoader(ShenyuInstanceRegisterRepository.class).getJoin(discoveryConfig.getRegisterType());
- discoveryConfig.getProps().put("watchPath", path);
+ boolean zookeeper =
StringUtils.equalsIgnoreCase(discoveryConfig.getRegisterType(), "zookeeper");
+ if (!zookeeper) {
+ discoveryConfig.getProps().put("watchPath", path);
+ }
discoveryService.init(discoveryConfig);
InstanceEntity instance = new InstanceEntity();
instance.setStatus(currentInstanceUpstream.getStatus());
@@ -87,6 +94,9 @@ public class InstanceRegisterListener implements
ApplicationListener<ContextRefr
instance.setHost(uri.getHost());
instance.setAppName(discoveryConfig.getProps().getProperty("name"));
discoveryService.persistInstance(instance);
+ if (zookeeper) {
+
shenyuDiscoveryConfig.setRegisterPath(InstancePathConstants.buildInstanceParentPath(instance.getAppName()));
+ }
LOGGER.info("shenyu register into ShenyuDiscoveryService {}
success", discoveryConfig.getRegisterType());
} catch (Exception e) {
LOGGER.error("shenyu register into ShenyuDiscoveryService {} type
find error", discoveryConfig.getRegisterType(), e);
diff --git
a/shenyu-examples/shenyu-examples-http/src/main/resources/application.yml
b/shenyu-examples/shenyu-examples-http/src/main/resources/application.yml
index 549455dfb1..902cacfe46 100644
--- a/shenyu-examples/shenyu-examples-http/src/main/resources/application.yml
+++ b/shenyu-examples/shenyu-examples-http/src/main/resources/application.yml
@@ -35,11 +35,12 @@ shenyu:
# port: 8189
# discovery:
# enable: true
+# register: true
# protocol: http:// # https://
# type: zookeeper
# serverList: 127.0.0.1:2181
-# registerPath: /shenyu/discovery/http_example
# props:
+# name: ${shenyu.client.http.props.appName}
# baseSleepTimeMilliseconds: 1000
# maxRetries: 4
# maxSleepTimeMilliseconds: 5000
diff --git
a/shenyu-registry/shenyu-registry-zookeeper/src/main/java/org/apache/shenyu/registry/zookeeper/ZookeeperInstanceRegisterRepository.java
b/shenyu-registry/shenyu-registry-zookeeper/src/main/java/org/apache/shenyu/registry/zookeeper/ZookeeperInstanceRegisterRepository.java
index 4c87896cb4..da913eccdb 100644
---
a/shenyu-registry/shenyu-registry-zookeeper/src/main/java/org/apache/shenyu/registry/zookeeper/ZookeeperInstanceRegisterRepository.java
+++
b/shenyu-registry/shenyu-registry-zookeeper/src/main/java/org/apache/shenyu/registry/zookeeper/ZookeeperInstanceRegisterRepository.java
@@ -19,6 +19,7 @@ package org.apache.shenyu.registry.zookeeper;
import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.Multimap;
+import com.google.gson.JsonElement;
import com.google.gson.JsonObject;
import org.apache.commons.lang3.StringUtils;
import org.apache.curator.framework.api.CuratorWatcher;
@@ -64,8 +65,6 @@ public class ZookeeperInstanceRegisterRepository implements
ShenyuInstanceRegist
private ZookeeperClient client;
- private String watchPath;
-
private final Map<String, String> nodeDataMap = new ConcurrentHashMap<>();
private final Multimap<String, CuratorCache> cacheMap =
ArrayListMultimap.create();
@@ -81,8 +80,6 @@ public class ZookeeperInstanceRegisterRepository implements
ShenyuInstanceRegist
int baseSleepTime =
Integer.parseInt(props.getProperty("baseSleepTime", "1000"));
int maxRetries = Integer.parseInt(props.getProperty("maxRetries",
"3"));
int maxSleepTime = Integer.parseInt(props.getProperty("maxSleepTime",
String.valueOf(Integer.MAX_VALUE)));
- watchPath = props.getProperty("watchPath", null);
-
ZookeeperConfig zkConfig = ZookeeperConfig.builder()
.url(config.getServerLists())
.baseSleepTimeMilliseconds(baseSleepTime)
@@ -129,8 +126,7 @@ public class ZookeeperInstanceRegisterRepository implements
ShenyuInstanceRegist
@Override
public List<InstanceEntity> selectInstances(final String selectKey) {
try {
- final String watchKey = StringUtils.isNotBlank(watchPath)
- ? InstancePathConstants.buildRealNode(watchPath,
selectKey) : InstancePathConstants.buildInstanceParentPath(selectKey);
+ final String watchKey = buildInstancePath(selectKey);
final Function<List<String>, List<InstanceEntity>>
getInstanceRegisterFun = childrenList -> childrenList.stream().map(childPath ->
{
String instanceRegisterJsonStr =
client.get(InstancePathConstants.buildRealNode(watchKey, childPath));
InstanceEntity instanceEntity =
GsonUtils.getInstance().fromJson(instanceRegisterJsonStr, InstanceEntity.class);
@@ -153,7 +149,7 @@ public class ZookeeperInstanceRegisterRepository implements
ShenyuInstanceRegist
@Override
public void process(final WatchedEvent event) {
try {
- String path = Objects.isNull(event.getPath()) ?
selectKey : event.getPath();
+ String path = Objects.isNull(event.getPath()) ?
watchKey : event.getPath();
List<String> childrenList =
StringUtils.isNotBlank(path) ? client.subscribeChildrenChanges(path, this)
: Collections.emptyList();
watcherInstanceRegisterMap.put(selectKey,
getInstanceRegisterFun.apply(childrenList));
@@ -176,7 +172,7 @@ public class ZookeeperInstanceRegisterRepository implements
ShenyuInstanceRegist
@Override
public boolean serviceExists(final String key) {
try {
- return Objects.nonNull(client.get(key));
+ return client.isExist(buildInstancePath(key));
} catch (Exception e) {
throw new ShenyuException(e);
}
@@ -185,7 +181,7 @@ public class ZookeeperInstanceRegisterRepository implements
ShenyuInstanceRegist
@Override
public void watchInstances(final String key, final ChangedEventListener
listener) {
try {
- CuratorCache treeCache = client.addCache(key, (type, oldData,
data) -> {
+ CuratorCache treeCache = client.addCache(buildInstancePath(key),
(type, oldData, data) -> {
// Curator delivers NODE_DELETED with a null new ChildData and
the deleted node in oldData
ChildData changedNode = CuratorCacheListener.Type.NODE_DELETED
== type ? oldData : data;
if (!Objects.nonNull(changedNode) ||
!Objects.nonNull(changedNode.getData())) {
@@ -200,6 +196,13 @@ public class ZookeeperInstanceRegisterRepository
implements ShenyuInstanceRegist
LOGGER.info("zookeeper registry watch Ignore non-ephemeral
node changes path {}", currentPath);
return;
}
+ if
(currentPath.startsWith(InstancePathConstants.buildInstanceParentPath() + "/"))
{
+ JsonObject instanceData =
GsonUtils.getInstance().fromJson(currentData, JsonObject.class);
+ JsonElement host = instanceData.get("host");
+ String address = Objects.nonNull(host) &&
!host.isJsonNull() ? host.getAsString() :
instanceData.get("address").getAsString();
+ instanceData.addProperty("url", address + Constants.COLONS
+ instanceData.get("port").getAsInt());
+ currentData = GsonUtils.getInstance().toJson(instanceData);
+ }
switch (type) {
case NODE_CREATED:
listener.onEvent(currentPath, currentData,
ChangedEventListener.Event.ADDED);
@@ -255,6 +258,10 @@ public class ZookeeperInstanceRegisterRepository
implements ShenyuInstanceRegist
}
}
+ private String buildInstancePath(final String key) {
+ return key.startsWith("/") ? key :
InstancePathConstants.buildInstanceParentPath(key);
+ }
+
private String buildInstanceNodeName(final InstanceEntity instance) {
String host = instance.getHost();
int port = instance.getPort();
diff --git
a/shenyu-registry/shenyu-registry-zookeeper/src/test/java/org/apache/shenyu/registry/zookeeper/ZookeeperInstanceRegisterRepositoryTest.java
b/shenyu-registry/shenyu-registry-zookeeper/src/test/java/org/apache/shenyu/registry/zookeeper/ZookeeperInstanceRegisterRepositoryTest.java
index 4740c779b0..e2105e326c 100644
---
a/shenyu-registry/shenyu-registry-zookeeper/src/test/java/org/apache/shenyu/registry/zookeeper/ZookeeperInstanceRegisterRepositoryTest.java
+++
b/shenyu-registry/shenyu-registry-zookeeper/src/test/java/org/apache/shenyu/registry/zookeeper/ZookeeperInstanceRegisterRepositoryTest.java
@@ -18,28 +18,39 @@
package org.apache.shenyu.registry.zookeeper;
import org.apache.curator.framework.CuratorFramework;
+import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.CuratorWatcher;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.CuratorCacheListener;
import org.apache.curator.framework.listen.Listenable;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.framework.state.ConnectionStateListener;
+import org.apache.curator.retry.ExponentialBackoffRetry;
+import org.apache.curator.test.TestingServer;
+import org.apache.shenyu.common.dto.DiscoveryUpstreamData;
import org.apache.shenyu.common.utils.GsonUtils;
import org.apache.shenyu.registry.api.event.ChangedEventListener;
import org.apache.shenyu.infra.zookeeper.client.ZookeeperClient;
import org.apache.shenyu.registry.api.config.RegisterConfig;
import org.apache.shenyu.registry.api.entity.InstanceEntity;
import org.apache.shenyu.registry.api.path.InstancePathConstants;
+import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.junit.jupiter.api.Test;
import org.mockito.MockedConstruction;
+import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
+import java.util.Map;
import java.util.Properties;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyString;
@@ -50,6 +61,48 @@ import static org.mockito.Mockito.when;
public final class ZookeeperInstanceRegisterRepositoryTest {
+ @Test
+ public void testSelectAndWatchNativeInstances() throws Exception {
+ try (TestingServer server = new TestingServer();
+ CuratorFramework writer =
CuratorFrameworkFactory.newClient(server.getConnectString(), new
ExponentialBackoffRetry(100, 3))) {
+ writer.start();
+ assertTrue(writer.blockUntilConnected(5, TimeUnit.SECONDS));
+ InstanceEntity instance = new InstanceEntity("native-test",
"127.0.0.1", 9195);
+ instance.setWeight(50);
+ String parent =
InstancePathConstants.buildInstanceParentPath(instance.getAppName());
+ String path = parent + "/127.0.0.1:9195";
+ String json = GsonUtils.getInstance().toJson(instance);
+
writer.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL).forPath(path,
json.getBytes(StandardCharsets.UTF_8));
+ ZookeeperInstanceRegisterRepository repository = new
ZookeeperInstanceRegisterRepository();
+ try {
+ repository.init(new RegisterConfig("zookeeper",
server.getConnectString(), new Properties()));
+ List<InstanceEntity> instances =
repository.selectInstances(parent);
+ assertEquals(1, instances.size());
+ assertEquals("http://127.0.0.1:9195";,
instances.get(0).getUri().toString());
+ BlockingQueue<Map.Entry<ChangedEventListener.Event, String>>
events = new LinkedBlockingQueue<>();
+ repository.watchInstances(parent, (key, value, event) ->
events.add(Map.entry(event, value)));
+ assertInstanceEvent(events, ChangedEventListener.Event.ADDED,
50);
+ instance.setWeight(80);
+ writer.setData().forPath(path,
GsonUtils.getInstance().toJson(instance).getBytes(StandardCharsets.UTF_8));
+ assertInstanceEvent(events,
ChangedEventListener.Event.UPDATED, 80);
+ writer.delete().forPath(path);
+ assertInstanceEvent(events,
ChangedEventListener.Event.DELETED, 80);
+ } finally {
+ repository.close();
+ }
+ }
+ }
+
+ private void assertInstanceEvent(final
BlockingQueue<Map.Entry<ChangedEventListener.Event, String>> events,
+ final ChangedEventListener.Event
expected, final int weight) throws InterruptedException {
+ Map.Entry<ChangedEventListener.Event, String> event = events.poll(5,
TimeUnit.SECONDS);
+ assertNotNull(event);
+ assertEquals(expected, event.getKey());
+ DiscoveryUpstreamData upstream =
GsonUtils.getInstance().fromJson(event.getValue(), DiscoveryUpstreamData.class);
+ assertEquals("127.0.0.1:9195", upstream.getUrl());
+ assertEquals(weight, upstream.getWeight());
+ }
+
@Test
public void testZookeeperInstanceRegisterRepository() {
final Listenable listenable = mock(Listenable.class);
@@ -103,10 +156,10 @@ public final class
ZookeeperInstanceRegisterRepositoryTest {
final Properties configProps = config.getProps();
configProps.setProperty("digest", "digest");
repository.init(config);
- String selectKey = InstancePathConstants.buildInstanceParentPath();
+ String selectKey = data.getAppName();
assertEquals(1, repository.selectInstances(selectKey).size());
WatchedEvent mockEvent = mock(WatchedEvent.class);
- when(mockEvent.getPath()).thenReturn(selectKey);
+ when(mockEvent.getPath()).thenReturn(null);
hasInstance[0] = false;
watcherArr[0].process(mockEvent);
assertTrue(repository.selectInstances(selectKey).isEmpty());
@@ -135,10 +188,17 @@ public final class
ZookeeperInstanceRegisterRepositoryTest {
org.mockito.Mockito.verify(client).addCache(org.mockito.ArgumentMatchers.eq("/shenyu/register/instance"),
captor.capture());
org.apache.zookeeper.data.Stat stat = new
org.apache.zookeeper.data.Stat();
stat.setEphemeralOwner(1L);
- ChildData deletedNode = new
ChildData("/shenyu/register/instance/app-host-9195", stat,
"instance-data".getBytes(java.nio.charset.StandardCharsets.UTF_8));
+ String path = InstancePathConstants.buildInstanceParentPath("app")
+ "/host:9195";
+ String instanceData =
"{\"appName\":\"app\",\"host\":\"host\",\"port\":9195,\"status\":0,\"weight\":50}";
+ ChildData deletedNode = new ChildData(path, stat,
instanceData.getBytes(java.nio.charset.StandardCharsets.UTF_8));
// Curator delivers NODE_DELETED with a null new ChildData and the
node in oldData
captor.getValue().event(CuratorCacheListener.Type.NODE_DELETED,
deletedNode, null);
-
org.mockito.Mockito.verify(changedEventListener).onEvent("/shenyu/register/instance/app-host-9195",
"instance-data", ChangedEventListener.Event.DELETED);
+ org.mockito.ArgumentCaptor<String> value =
org.mockito.ArgumentCaptor.forClass(String.class);
+
org.mockito.Mockito.verify(changedEventListener).onEvent(org.mockito.ArgumentMatchers.eq(path),
value.capture(),
org.mockito.ArgumentMatchers.eq(ChangedEventListener.Event.DELETED));
+ org.apache.shenyu.common.dto.DiscoveryUpstreamData upstream =
GsonUtils.getInstance().fromJson(value.getValue(),
org.apache.shenyu.common.dto.DiscoveryUpstreamData.class);
+ assertEquals("host:9195", upstream.getUrl());
+ assertEquals(50, upstream.getWeight());
+ assertEquals(0, upstream.getStatus());
}
}
