From 0d4ffd5173db0adc9dc1eb7df6295f3b149f33cb Mon Sep 17 00:00:00 2001 From: Lucian <943071074@qq.com> Date: Wed, 22 May 2024 17:08:41 +0800 Subject: [PATCH] Release 650 (#351) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * 提供API查询session链接数 * fix node forbidden bug * support top sub/pub query api * interface app cleaner * sessionNode renew add weight * default push opened * connection load balance support p2c * update version & fix & add ut * fix ut & git uint test add dump * set unit test timeout * registry-client compatible old version server * fix p2c client bug * fix log & update version --- .github/workflows/unit-test.yml | 7 ++ VERSION | 2 +- client/all/pom.xml | 2 +- client/api/pom.xml | 2 +- client/impl/pom.xml | 2 +- .../client/provider/DefaultServerManager.java | 10 +- .../client/provider/DefaultServerNode.java | 19 +++ .../client/remoting/ClientConnection.java | 17 ++- .../registry/client/remoting/ServerNode.java | 7 ++ .../provider/DefaultServerManagerTest.java | 36 +++++- client/log/pom.xml | 2 +- client/pom.xml | 2 +- core/pom.xml | 2 +- pom.xml | 2 +- server/common/model/pom.xml | 2 +- .../model/constants/ValueConstants.java | 6 + .../model/metaserver/nodes/SessionNode.java | 10 +- .../model/metaserver/nodes/NodeTest.java | 6 +- .../metaserver/rpc/DataCenterNodesTest.java | 3 +- server/common/pom.xml | 2 +- server/common/util/pom.xml | 2 +- server/distribution/all/pom.xml | 2 +- server/distribution/pom.xml | 2 +- server/pom.xml | 2 +- server/remoting/api/pom.xml | 2 +- server/remoting/bolt/pom.xml | 2 +- server/remoting/http/pom.xml | 2 +- server/remoting/pom.xml | 2 +- server/server/data/pom.xml | 2 +- server/server/integration/pom.xml | 2 +- server/server/meta/pom.xml | 2 +- .../bootstrap/config/MetaServerConfig.java | 2 + .../config/MetaServerConfigBean.java | 11 ++ .../cleaner/InterfaceAppsIndexCleaner.java | 116 +++++++++++++++++- .../filter/DefaultForbiddenServerManager.java | 12 ++ .../session/DefaultSessionServerManager.java | 7 ++ .../meta/resource/MetaCenterResource.java | 30 +++++ .../InterfaceAppsIndexCleanerTest.java | 63 ++++++++++ .../DefaultSessionServerManagerTest.java | 32 ++++- .../impl/DefaultCurrentDcMetaServerTest.java | 4 +- .../DefaultSessionServerServiceTest.java | 18 +-- .../meta/resource/MetaCenterResourceTest.java | 12 +- .../resource/RegistryCoreOpsResourceTest.java | 22 ++++ server/server/pom.xml | 2 +- server/server/session/pom.xml | 2 +- .../node/service/MetaServerServiceImpl.java | 22 +++- .../resource/SessionDigestResource.java | 36 ++++-- .../session/resource/SessionOpenResource.java | 41 +++++++ .../AbstractSessionServerTestBase.java | 5 + .../service/MetaServerServiceImplTest.java | 2 + .../providedata/FetchStopPushServiceTest.java | 2 +- server/server/shared/pom.xml | 2 +- .../meta/AbstractMetaServerService.java | 13 ++ .../server/shared/meta/MetaServerService.java | 3 + .../providedata/BaseStopPushService.java | 2 +- .../shared/meta/MetaServerServiceTest.java | 11 +- .../server/shared/util/NodeUtilsTest.java | 4 +- server/store/api/pom.xml | 2 +- .../repository/InterfaceAppsRepository.java | 3 + server/store/jdbc/pom.xml | 2 +- .../jdbc/mapper/InterfaceAppsIndexMapper.java | 6 + .../impl/InterfaceAppsJdbcRepository.java | 12 +- .../h2-mapper/interface_apps_index.xml | 8 ++ .../resources/mapper/interface_apps_index.xml | 7 ++ .../impl/InterfaceAppsJdbcRepositoryTest.java | 11 +- server/store/jraft/pom.xml | 2 +- .../impl/InterfaceAppsRaftRepository.java | 6 + server/store/pom.xml | 2 +- test/pom.xml | 4 +- .../InterfaceAppsJdbcRepositoryTest.java | 6 +- 70 files changed, 622 insertions(+), 88 deletions(-) diff --git a/.github/workflows/unit-test.yml b/.github/workflows/unit-test.yml index 219efa905..70fb20213 100644 --- a/.github/workflows/unit-test.yml +++ b/.github/workflows/unit-test.yml @@ -13,6 +13,7 @@ jobs: build: runs-on: ubuntu-latest + timeout-minutes: 15 steps: - uses: actions/checkout@v2 @@ -23,6 +24,12 @@ jobs: - name: Unit Testt run: mvn compile -B && mvn clean test -DisSkipIntegrationTest=true "-Dtest.logging.level=ERROR" --fail-at-end --batch-mode + - name: Upload heap dump + if: always() + uses: actions/upload-artifact@v2 + with: + name: heap-dump + path: /tmp/*.hprof - name: Publish Test Report if: ${{ always() }} uses: ScaCap/action-surefire-report@v1 diff --git a/VERSION b/VERSION index 19b860c18..a194c18e8 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -6.4.0 +6.5.1 diff --git a/client/all/pom.xml b/client/all/pom.xml index 9fe1ac90a..13f069eee 100644 --- a/client/all/pom.xml +++ b/client/all/pom.xml @@ -6,7 +6,7 @@ com.alipay.sofa registry-client-all - 6.4.0 + 6.5.1 ${project.groupId}:${project.artifactId} http://github.com/alipay/sofa-registry diff --git a/client/api/pom.xml b/client/api/pom.xml index 9ad4b1665..095f6ec4a 100644 --- a/client/api/pom.xml +++ b/client/api/pom.xml @@ -5,7 +5,7 @@ com.alipay.sofa registry-client-parent - 6.4.0 + 6.5.1 ../pom.xml 4.0.0 diff --git a/client/impl/pom.xml b/client/impl/pom.xml index 0ed77e6df..a8adff1b2 100644 --- a/client/impl/pom.xml +++ b/client/impl/pom.xml @@ -5,7 +5,7 @@ com.alipay.sofa registry-client-parent - 6.4.0 + 6.5.1 ../pom.xml 4.0.0 diff --git a/client/impl/src/main/java/com/alipay/sofa/registry/client/provider/DefaultServerManager.java b/client/impl/src/main/java/com/alipay/sofa/registry/client/provider/DefaultServerManager.java index 050e46c3a..af93d27ff 100644 --- a/client/impl/src/main/java/com/alipay/sofa/registry/client/provider/DefaultServerManager.java +++ b/client/impl/src/main/java/com/alipay/sofa/registry/client/provider/DefaultServerManager.java @@ -90,7 +90,7 @@ public ServerNode random() { private void syncServerList() { String url = String.format( - "http://%s:%d/api/servers/query", + "http://%s:%d/api/servers/queryWithWeight", config.getRegistryEndpoint(), config.getRegistryEndpointPort()); Map params = new HashMap(); params.put("env", config.getEnv()); @@ -100,6 +100,14 @@ private void syncServerList() { params.put("instanceId", config.getInstanceId()); try { String result = HttpClientUtils.get(url, params, config); + if (null == result) { + // when registry not support query with weight , go back + url = + String.format( + "http://%s:%d/api/servers/query", + config.getRegistryEndpoint(), config.getRegistryEndpointPort()); + result = HttpClientUtils.get(url, params, config); + } if (null != result) { String[] servers = result.split(";"); Set tempNodes = new HashSet(); diff --git a/client/impl/src/main/java/com/alipay/sofa/registry/client/provider/DefaultServerNode.java b/client/impl/src/main/java/com/alipay/sofa/registry/client/provider/DefaultServerNode.java index 886631ff0..2c71ea8c0 100644 --- a/client/impl/src/main/java/com/alipay/sofa/registry/client/provider/DefaultServerNode.java +++ b/client/impl/src/main/java/com/alipay/sofa/registry/client/provider/DefaultServerNode.java @@ -17,6 +17,7 @@ package com.alipay.sofa.registry.client.provider; import com.alipay.sofa.registry.client.remoting.ServerNode; +import com.alipay.sofa.registry.client.util.StringUtils; import java.util.Properties; /** @@ -35,6 +36,8 @@ public class DefaultServerNode implements ServerNode { private Properties properties; + private static final String WEIGHT_KEY = "weight"; + /** * Instantiates a new Default server node. * @@ -80,6 +83,22 @@ public String getUrl() { return url; } + @Override + public int getWeight() { + if (null == getProperties()) { + return 0; + } + String weightStr = getProperties().getProperty(WEIGHT_KEY); + if (StringUtils.isBlank(weightStr)) { + return 0; + } + try { + return Integer.parseInt(weightStr); + } catch (NumberFormatException e) { + return 0; + } + } + /** * Gets properties. * diff --git a/client/impl/src/main/java/com/alipay/sofa/registry/client/remoting/ClientConnection.java b/client/impl/src/main/java/com/alipay/sofa/registry/client/remoting/ClientConnection.java index 7d4e5ea6e..fd0a6e99c 100644 --- a/client/impl/src/main/java/com/alipay/sofa/registry/client/remoting/ClientConnection.java +++ b/client/impl/src/main/java/com/alipay/sofa/registry/client/remoting/ClientConnection.java @@ -125,12 +125,21 @@ private boolean connect() { List serverNodes = new ArrayList(serverManager.getServerList()); // shuffle server list to make server connections as discrete as possible Collections.shuffle(serverNodes); - for (ServerNode serverNode : serverNodes) { + int choosed = 0; + for (int i = 0; i < serverNodes.size(); i++) { try { - connection = connect(serverNode); + // Power of Two Choices + if (serverNodes.size() > 1) { + choosed = + serverNodes.get(i).getWeight() + > serverNodes.get((i + 1) % serverNodes.size()).getWeight() + ? (i + 1) % serverNodes.size() + : i; + } + connection = connect(serverNodes.get(choosed)); if (null != connection && connection.isFine()) { resetRegister(); - LOGGER.info("[Connect] Successfully connected to server: {}", serverNode); + LOGGER.info("[Connect] Successfully connected to server: {}", serverNodes.get(choosed)); break; } else { recycle(connection); @@ -138,7 +147,7 @@ private boolean connect() { Thread.sleep(random.nextInt(RECONNECTING_DELAY)); } catch (Exception e) { - LOGGER.error("[Connect] Failed trying connect to {}", serverNode, e); + LOGGER.error("[Connect] Failed trying connect to {}", serverNodes.get(choosed), e); } } diff --git a/client/impl/src/main/java/com/alipay/sofa/registry/client/remoting/ServerNode.java b/client/impl/src/main/java/com/alipay/sofa/registry/client/remoting/ServerNode.java index 360baa2b0..44af30052 100644 --- a/client/impl/src/main/java/com/alipay/sofa/registry/client/remoting/ServerNode.java +++ b/client/impl/src/main/java/com/alipay/sofa/registry/client/remoting/ServerNode.java @@ -53,4 +53,11 @@ public interface ServerNode { * @return the url */ String getUrl(); + + /** + * Gets Weight + * + * @return the weight + */ + int getWeight(); } diff --git a/client/impl/src/test/java/com/alipay/sofa/registry/client/provider/DefaultServerManagerTest.java b/client/impl/src/test/java/com/alipay/sofa/registry/client/provider/DefaultServerManagerTest.java index 1e3e0aff1..d7d9fb8da 100644 --- a/client/impl/src/test/java/com/alipay/sofa/registry/client/provider/DefaultServerManagerTest.java +++ b/client/impl/src/test/java/com/alipay/sofa/registry/client/provider/DefaultServerManagerTest.java @@ -17,9 +17,7 @@ package com.alipay.sofa.registry.client.provider; import static org.junit.Assert.assertNotNull; -import static org.mockito.Matchers.any; -import static org.mockito.Matchers.anyMapOf; -import static org.mockito.Matchers.anyString; +import static org.mockito.Matchers.*; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.times; import static org.mockito.Mockito.when; @@ -67,4 +65,36 @@ public void initServerList() throws Exception { // verify PowerMockito.verifyStatic(times(4)); } + + @Test + public void initServerListCompatible() throws Exception { + // given + PowerMockito.mockStatic(HttpClientUtils.class); + RegistryClientConfig config = mock(RegistryClientConfig.class); + + // when + when(config.getSyncConfigRetryInterval()).thenReturn(100); + when(HttpClientUtils.get( + eq("http://null:0/api/servers/queryWithWeight"), + anyMapOf(String.class, String.class), + any(RegistryClientConfig.class))) + .thenReturn(null); + when(HttpClientUtils.get( + eq("http://null:0/api/servers/query"), + anyMapOf(String.class, String.class), + any(RegistryClientConfig.class))) + .thenReturn("127.0.0.1:9600;127.0.0.2:9600"); + + // then + ServerManager serverManager = new DefaultServerManager(config); + + List serverList = serverManager.getServerList(); + + assertNotNull(serverList); + + Thread.sleep(450); + + // verify + PowerMockito.verifyStatic(times(4)); + } } diff --git a/client/log/pom.xml b/client/log/pom.xml index 9d4e7ecfe..abc15d408 100644 --- a/client/log/pom.xml +++ b/client/log/pom.xml @@ -5,7 +5,7 @@ com.alipay.sofa registry-client-parent - 6.4.0 + 6.5.1 ../pom.xml 4.0.0 diff --git a/client/pom.xml b/client/pom.xml index 3603baeee..017d3e42b 100644 --- a/client/pom.xml +++ b/client/pom.xml @@ -7,7 +7,7 @@ com.alipay.sofa registry-parent - 6.4.0 + 6.5.1 ../pom.xml diff --git a/core/pom.xml b/core/pom.xml index 0b65bf7de..e603a981d 100644 --- a/core/pom.xml +++ b/core/pom.xml @@ -5,7 +5,7 @@ com.alipay.sofa registry-parent - 6.4.0 + 6.5.1 ../pom.xml 4.0.0 diff --git a/pom.xml b/pom.xml index 5475a4250..eb80ead87 100644 --- a/pom.xml +++ b/pom.xml @@ -6,7 +6,7 @@ 4.0.0 com.alipay.sofa registry-parent - 6.4.0 + 6.5.1 pom diff --git a/server/common/model/pom.xml b/server/common/model/pom.xml index 37d009f36..70b62ae39 100644 --- a/server/common/model/pom.xml +++ b/server/common/model/pom.xml @@ -5,7 +5,7 @@ com.alipay.sofa registry-common - 6.4.0 + 6.5.1 ../pom.xml 4.0.0 diff --git a/server/common/model/src/main/java/com/alipay/sofa/registry/common/model/constants/ValueConstants.java b/server/common/model/src/main/java/com/alipay/sofa/registry/common/model/constants/ValueConstants.java index 01b04a7eb..40c627cd1 100644 --- a/server/common/model/src/main/java/com/alipay/sofa/registry/common/model/constants/ValueConstants.java +++ b/server/common/model/src/main/java/com/alipay/sofa/registry/common/model/constants/ValueConstants.java @@ -138,6 +138,12 @@ public class ValueConstants { "app_revision.cleaner.enabled", SESSION_PROVIDE_DATA_INSTANCE_ID, SESSION_PROVIDE_DATA_GROUP); + + public static final String INTERFACE_APP_CLEANER_ENABLED_DATA_ID = + DataInfo.toDataInfoId( + "interface_app.cleaner.enabled", + SESSION_PROVIDE_DATA_INSTANCE_ID, + SESSION_PROVIDE_DATA_GROUP); public static final String COMPRESS_PUSH_SWITCH_DATA_ID = DataInfo.toDataInfoId( "compress.push.switch", SESSION_PROVIDE_DATA_INSTANCE_ID, SESSION_PROVIDE_DATA_GROUP); diff --git a/server/common/model/src/main/java/com/alipay/sofa/registry/common/model/metaserver/nodes/SessionNode.java b/server/common/model/src/main/java/com/alipay/sofa/registry/common/model/metaserver/nodes/SessionNode.java index b204a4c98..3032f937a 100644 --- a/server/common/model/src/main/java/com/alipay/sofa/registry/common/model/metaserver/nodes/SessionNode.java +++ b/server/common/model/src/main/java/com/alipay/sofa/registry/common/model/metaserver/nodes/SessionNode.java @@ -27,6 +27,9 @@ public class SessionNode extends AbstractNode { private final ProcessId processId; + // session weight for client conn load balance + private int weight; + /** * constructor * @@ -34,9 +37,10 @@ public class SessionNode extends AbstractNode { * @param regionId regionId * @param processId processId */ - public SessionNode(URL nodeUrl, String regionId, ProcessId processId) { + public SessionNode(URL nodeUrl, String regionId, ProcessId processId, int weight) { super(null, nodeUrl, regionId); this.processId = processId; + this.weight = weight; } @Override @@ -61,6 +65,10 @@ public ProcessId getProcessId() { return processId; } + public int getWeight() { + return weight; + } + /** * Hash code int. * diff --git a/server/common/model/src/test/java/com/alipay/sofa/registry/common/model/metaserver/nodes/NodeTest.java b/server/common/model/src/test/java/com/alipay/sofa/registry/common/model/metaserver/nodes/NodeTest.java index aac125cb8..798271220 100644 --- a/server/common/model/src/test/java/com/alipay/sofa/registry/common/model/metaserver/nodes/NodeTest.java +++ b/server/common/model/src/test/java/com/alipay/sofa/registry/common/model/metaserver/nodes/NodeTest.java @@ -51,9 +51,9 @@ public void testDataNode() { public void testSessionNode() { ProcessId processId1 = new ProcessId("test", 1, 2, 3); ProcessId processId2 = new ProcessId("test1", 1, 2, 3); - SessionNode node1 = new SessionNode(url1, region, processId1); - SessionNode node2 = new SessionNode(url2, region, processId2); - SessionNode node3 = new SessionNode(url1, region, processId2); + SessionNode node1 = new SessionNode(url1, region, processId1, 0); + SessionNode node2 = new SessionNode(url2, region, processId2, 0); + SessionNode node3 = new SessionNode(url1, region, processId2, 0); Assert.assertEquals(node1, node3); Assert.assertEquals(node1.hashCode(), node3.hashCode()); Assert.assertEquals(node1.toString(), node3.toString()); diff --git a/server/common/model/src/test/java/com/alipay/sofa/registry/common/model/metaserver/rpc/DataCenterNodesTest.java b/server/common/model/src/test/java/com/alipay/sofa/registry/common/model/metaserver/rpc/DataCenterNodesTest.java index c341b2700..beb1df6dc 100644 --- a/server/common/model/src/test/java/com/alipay/sofa/registry/common/model/metaserver/rpc/DataCenterNodesTest.java +++ b/server/common/model/src/test/java/com/alipay/sofa/registry/common/model/metaserver/rpc/DataCenterNodesTest.java @@ -31,7 +31,8 @@ public void test() { ProcessId processId1 = new ProcessId("test", 1, 2, 3); final String dataId = "testDataId"; DataCenterNodes request = new DataCenterNodes(Node.NodeType.CLIENT, 10, dataId); - SessionNode sessionNode = new SessionNode(new URL("192.168.1.1", 8888), "testZone", processId1); + SessionNode sessionNode = + new SessionNode(new URL("192.168.1.1", 8888), "testZone", processId1, 0); request.setNodes(Collections.singletonMap("testKey", sessionNode)); Assert.assertEquals(request.getDataCenterId(), dataId); Assert.assertEquals(request.getVersion(), 10); diff --git a/server/common/pom.xml b/server/common/pom.xml index 8eb18249e..7f4b2b223 100644 --- a/server/common/pom.xml +++ b/server/common/pom.xml @@ -5,7 +5,7 @@ com.alipay.sofa registry-server-parent - 6.4.0 + 6.5.1 ../pom.xml 4.0.0 diff --git a/server/common/util/pom.xml b/server/common/util/pom.xml index e4a24c30f..1bc605861 100644 --- a/server/common/util/pom.xml +++ b/server/common/util/pom.xml @@ -5,7 +5,7 @@ com.alipay.sofa registry-common - 6.4.0 + 6.5.1 ../pom.xml 4.0.0 diff --git a/server/distribution/all/pom.xml b/server/distribution/all/pom.xml index e82ceef53..c7752134d 100644 --- a/server/distribution/all/pom.xml +++ b/server/distribution/all/pom.xml @@ -5,7 +5,7 @@ com.alipay.sofa registry-distribution - 6.4.0 + 6.5.1 ../pom.xml 4.0.0 diff --git a/server/distribution/pom.xml b/server/distribution/pom.xml index d39cbf7f3..04fa6c0e4 100644 --- a/server/distribution/pom.xml +++ b/server/distribution/pom.xml @@ -6,7 +6,7 @@ com.alipay.sofa registry-server-parent - 6.4.0 + 6.5.1 ../pom.xml 4.0.0 diff --git a/server/pom.xml b/server/pom.xml index df4297d44..0430287e4 100644 --- a/server/pom.xml +++ b/server/pom.xml @@ -7,7 +7,7 @@ com.alipay.sofa registry-parent - 6.4.0 + 6.5.1 ../pom.xml diff --git a/server/remoting/api/pom.xml b/server/remoting/api/pom.xml index 15144170f..1e50e6cfe 100644 --- a/server/remoting/api/pom.xml +++ b/server/remoting/api/pom.xml @@ -5,7 +5,7 @@ com.alipay.sofa registry-remoting - 6.4.0 + 6.5.1 ../pom.xml 4.0.0 diff --git a/server/remoting/bolt/pom.xml b/server/remoting/bolt/pom.xml index fa651f1cf..052a84abd 100644 --- a/server/remoting/bolt/pom.xml +++ b/server/remoting/bolt/pom.xml @@ -5,7 +5,7 @@ com.alipay.sofa registry-remoting - 6.4.0 + 6.5.1 ../pom.xml 4.0.0 diff --git a/server/remoting/http/pom.xml b/server/remoting/http/pom.xml index ef6b5401a..ef3b0b985 100644 --- a/server/remoting/http/pom.xml +++ b/server/remoting/http/pom.xml @@ -5,7 +5,7 @@ com.alipay.sofa registry-remoting - 6.4.0 + 6.5.1 ../pom.xml 4.0.0 diff --git a/server/remoting/pom.xml b/server/remoting/pom.xml index 00002d834..a0087f215 100644 --- a/server/remoting/pom.xml +++ b/server/remoting/pom.xml @@ -5,7 +5,7 @@ com.alipay.sofa registry-server-parent - 6.4.0 + 6.5.1 ../pom.xml 4.0.0 diff --git a/server/server/data/pom.xml b/server/server/data/pom.xml index 2e0144410..1a74cd232 100644 --- a/server/server/data/pom.xml +++ b/server/server/data/pom.xml @@ -5,7 +5,7 @@ com.alipay.sofa registry-server - 6.4.0 + 6.5.1 ../pom.xml 4.0.0 diff --git a/server/server/integration/pom.xml b/server/server/integration/pom.xml index 0cd3ac0b5..3942c391e 100644 --- a/server/server/integration/pom.xml +++ b/server/server/integration/pom.xml @@ -5,7 +5,7 @@ com.alipay.sofa registry-server - 6.4.0 + 6.5.1 ../pom.xml 4.0.0 diff --git a/server/server/meta/pom.xml b/server/server/meta/pom.xml index c3f15d62b..6aa0815a7 100644 --- a/server/server/meta/pom.xml +++ b/server/server/meta/pom.xml @@ -5,7 +5,7 @@ com.alipay.sofa registry-server - 6.4.0 + 6.5.1 ../pom.xml 4.0.0 diff --git a/server/server/meta/src/main/java/com/alipay/sofa/registry/server/meta/bootstrap/config/MetaServerConfig.java b/server/server/meta/src/main/java/com/alipay/sofa/registry/server/meta/bootstrap/config/MetaServerConfig.java index 573a54e3d..954400af1 100644 --- a/server/server/meta/src/main/java/com/alipay/sofa/registry/server/meta/bootstrap/config/MetaServerConfig.java +++ b/server/server/meta/src/main/java/com/alipay/sofa/registry/server/meta/bootstrap/config/MetaServerConfig.java @@ -79,5 +79,7 @@ public interface MetaServerConfig { int getAppRevisionMaxRemove(); + int getInterfaceMaxRemove(); + int getAppRevisionCountAlarmThreshold(); } diff --git a/server/server/meta/src/main/java/com/alipay/sofa/registry/server/meta/bootstrap/config/MetaServerConfigBean.java b/server/server/meta/src/main/java/com/alipay/sofa/registry/server/meta/bootstrap/config/MetaServerConfigBean.java index 94b0fa703..54cac38c8 100644 --- a/server/server/meta/src/main/java/com/alipay/sofa/registry/server/meta/bootstrap/config/MetaServerConfigBean.java +++ b/server/server/meta/src/main/java/com/alipay/sofa/registry/server/meta/bootstrap/config/MetaServerConfigBean.java @@ -74,6 +74,8 @@ public class MetaServerConfigBean implements MetaServerConfig { private int clientManagerExpireDays = 1; private int appRevisionMaxRemove = 2000; + + private int interfaceMaxRemove = 2000; private int appRevisionCountAlarmThreshold = 20; private long metaLeaderWarmupMillis = @@ -518,9 +520,18 @@ public int getAppRevisionMaxRemove() { return appRevisionMaxRemove; } + @Override + public int getInterfaceMaxRemove() { + return interfaceMaxRemove; + } + public void setAppRevisionMaxRemove(int appRevisionMaxRemove) { this.appRevisionMaxRemove = appRevisionMaxRemove; } + + public void setInterfaceMaxRemove(int interfaceMaxRemove) { + this.interfaceMaxRemove = interfaceMaxRemove; + } /** * Setter method for property clientManagerCleanSecs. * diff --git a/server/server/meta/src/main/java/com/alipay/sofa/registry/server/meta/cleaner/InterfaceAppsIndexCleaner.java b/server/server/meta/src/main/java/com/alipay/sofa/registry/server/meta/cleaner/InterfaceAppsIndexCleaner.java index c67bdef68..f71f01a35 100644 --- a/server/server/meta/src/main/java/com/alipay/sofa/registry/server/meta/cleaner/InterfaceAppsIndexCleaner.java +++ b/server/server/meta/src/main/java/com/alipay/sofa/registry/server/meta/cleaner/InterfaceAppsIndexCleaner.java @@ -16,29 +16,46 @@ */ package com.alipay.sofa.registry.server.meta.cleaner; +import com.alipay.sofa.registry.cache.ConsecutiveSuccess; +import com.alipay.sofa.registry.common.model.console.PersistenceData; +import com.alipay.sofa.registry.common.model.console.PersistenceDataBuilder; +import com.alipay.sofa.registry.common.model.constants.ValueConstants; import com.alipay.sofa.registry.common.model.store.AppRevision; import com.alipay.sofa.registry.jdbc.config.MetadataConfig; import com.alipay.sofa.registry.log.Logger; import com.alipay.sofa.registry.log.LoggerFactory; import com.alipay.sofa.registry.server.meta.MetaLeaderService; +import com.alipay.sofa.registry.server.meta.bootstrap.config.MetaServerConfig; +import com.alipay.sofa.registry.server.meta.provide.data.ProvideDataService; +import com.alipay.sofa.registry.store.api.DBResponse; +import com.alipay.sofa.registry.store.api.OperationStatus; +import com.alipay.sofa.registry.store.api.date.DateNowRepository; import com.alipay.sofa.registry.store.api.repository.AppRevisionRepository; import com.alipay.sofa.registry.store.api.repository.InterfaceAppsRepository; import com.alipay.sofa.registry.util.ConcurrentUtils; +import com.alipay.sofa.registry.util.StringFormatter; import com.alipay.sofa.registry.util.WakeUpLoopRunnable; import com.google.common.collect.Maps; import java.util.*; import java.util.concurrent.TimeUnit; +import javax.annotation.PostConstruct; import org.glassfish.jersey.internal.guava.Sets; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.ApplicationListener; import org.springframework.context.event.ContextRefreshedEvent; -public class InterfaceAppsIndexCleaner implements ApplicationListener { +public class InterfaceAppsIndexCleaner + implements MetaLeaderService.MetaLeaderElectorListener, + ApplicationListener { private static final Logger LOG = LoggerFactory.getLogger("METADATA-EXCHANGE", "[InterfaceApps]"); + private static final boolean DEFAULT_ENABLED = false; + final Renewer renewer = new Renewer(); + final Cleaner cleaner = new Cleaner(); + @Autowired AppRevisionRepository appRevisionRepository; @Autowired InterfaceAppsRepository interfaceAppsRepository; @@ -47,6 +64,22 @@ public class InterfaceAppsIndexCleaner implements ApplicationListener ret = + provideDataService.queryProvideData(ValueConstants.INTERFACE_APP_CLEANER_ENABLED_DATA_ID); + if (ret.getOperationStatus() == OperationStatus.SUCCESS) { + PersistenceData data = ret.getEntity(); + return Boolean.parseBoolean(data.getData()); + } + return DEFAULT_ENABLED; + } + + Date dateBeforeNow(int minutes) { + Calendar calendar = Calendar.getInstance(); + calendar.setTime(dateNowRepository.getNow()); + calendar.add(Calendar.MINUTE, -minutes); + return calendar.getTime(); + } + + void cleanup() { + if (!metaLeaderService.amILeader()) { + return; + } + if (!consecutiveSuccess.check()) { + return; + } + int count = + interfaceAppsRepository.cleanDeleted( + dateBeforeNow(metadataConfig.getInterfaceAppsIndexRenewIntervalMinutes() * 15), + metaServerConfig.getInterfaceMaxRemove()); + if (count > 0) { + LOG.info("clean up {} interface app", count); + } + } + public void startRenew() { renewer.wakeup(); } + public void startCleaner() { + cleaner.wakeup(); + } + + public void setEnabled(boolean enabled) { + PersistenceData persistenceData = + PersistenceDataBuilder.createPersistenceData( + ValueConstants.INTERFACE_APP_CLEANER_ENABLED_DATA_ID, Boolean.toString(enabled)); + try { + provideDataService.saveProvideData(persistenceData); + } catch (Exception e) { + LOG.error("set interface app cleaner failed: ", e); + throw new RuntimeException( + StringFormatter.format("set interface app cleaner failed: {}", e.getMessage())); + } + } + + @Override + public void becomeLeader() { + consecutiveSuccess.clear(); + } + + @Override + public void loseLeader() { + consecutiveSuccess.clear(); + } + final class Renewer extends WakeUpLoopRunnable { @Override public int getWaitingMillis() { @@ -122,4 +220,20 @@ public void runUnthrowable() { renew(); } } + + final class Cleaner extends WakeUpLoopRunnable { + @Override + public int getWaitingMillis() { + int base = metadataConfig.getInterfaceAppsIndexRenewIntervalMinutes() * 1000 * 60; + return (int) (base + Math.random() * base); + } + + @Override + public void runUnthrowable() { + if (!isEnabled()) { + return; + } + cleanup(); + } + } } diff --git a/server/server/meta/src/main/java/com/alipay/sofa/registry/server/meta/lease/filter/DefaultForbiddenServerManager.java b/server/server/meta/src/main/java/com/alipay/sofa/registry/server/meta/lease/filter/DefaultForbiddenServerManager.java index 7f40ddd0b..52448dd73 100644 --- a/server/server/meta/src/main/java/com/alipay/sofa/registry/server/meta/lease/filter/DefaultForbiddenServerManager.java +++ b/server/server/meta/src/main/java/com/alipay/sofa/registry/server/meta/lease/filter/DefaultForbiddenServerManager.java @@ -26,6 +26,9 @@ import com.alipay.sofa.registry.common.model.metaserver.NodeServerOperateInfo; import com.alipay.sofa.registry.common.model.metaserver.OperationInfo; import com.alipay.sofa.registry.common.model.metaserver.blacklist.RegistryForbiddenServerRequest; +import com.alipay.sofa.registry.common.model.metaserver.nodes.DataNode; +import com.alipay.sofa.registry.common.model.metaserver.nodes.MetaNode; +import com.alipay.sofa.registry.common.model.metaserver.nodes.SessionNode; import com.alipay.sofa.registry.exception.SofaRegistryRuntimeException; import com.alipay.sofa.registry.log.Logger; import com.alipay.sofa.registry.log.LoggerFactory; @@ -91,6 +94,15 @@ public boolean removeFromBlacklist(RegistryForbiddenServerRequest request) { @Override public boolean allowSelect(Lease lease) { ForbiddenServer servers = getForbiddenServers(); + if (lease.getRenewal() instanceof SessionNode) { + return !servers.contains(NodeType.SESSION, lease.getRenewal().getNodeUrl().getIpAddress()); + } + if (lease.getRenewal() instanceof DataNode) { + return !servers.contains(NodeType.DATA, lease.getRenewal().getNodeUrl().getIpAddress()); + } + if (lease.getRenewal() instanceof MetaNode) { + return !servers.contains(NodeType.META, lease.getRenewal().getNodeUrl().getIpAddress()); + } return !servers.contains(lease.getRenewal().getNodeUrl().getIpAddress()); } diff --git a/server/server/meta/src/main/java/com/alipay/sofa/registry/server/meta/lease/session/DefaultSessionServerManager.java b/server/server/meta/src/main/java/com/alipay/sofa/registry/server/meta/lease/session/DefaultSessionServerManager.java index 0bb25a1a3..2dfc1825e 100644 --- a/server/server/meta/src/main/java/com/alipay/sofa/registry/server/meta/lease/session/DefaultSessionServerManager.java +++ b/server/server/meta/src/main/java/com/alipay/sofa/registry/server/meta/lease/session/DefaultSessionServerManager.java @@ -102,6 +102,13 @@ public boolean renew(SessionNode renewal, int duration) { notifyObservers(new NodeModified<>(lease.getRenewal(), renewal)); return false; } else { + // replace the session node, as it has changed weight already + if (renewal.getProcessId() != null + && lease != null + && lease.getRenewal() != null + && renewal.getWeight() != lease.getRenewal().getWeight()) { + lease.setRenewal(renewal); + } return super.renew(renewal, duration); } } diff --git a/server/server/meta/src/main/java/com/alipay/sofa/registry/server/meta/resource/MetaCenterResource.java b/server/server/meta/src/main/java/com/alipay/sofa/registry/server/meta/resource/MetaCenterResource.java index 8d09a3849..393961522 100644 --- a/server/server/meta/src/main/java/com/alipay/sofa/registry/server/meta/resource/MetaCenterResource.java +++ b/server/server/meta/src/main/java/com/alipay/sofa/registry/server/meta/resource/MetaCenterResource.java @@ -62,6 +62,16 @@ public Result interfaceAppsIndexRenew() { return result; } + @PUT + @Path("interfaceAppsIndex/clean") + @Produces(MediaType.APPLICATION_JSON) + public Result interfaceAppsIndexClean() { + Result result = new Result(); + interfaceAppsIndexCleaner.startCleaner(); + result.setSuccess(true); + return result; + } + @PUT @Path("appRevisionCleaner/switch") @Produces(MediaType.APPLICATION_JSON) @@ -77,6 +87,21 @@ public Result appRevisionCleanerEnable(@FormParam("enabled") boolean enabled) { return result; } + @PUT + @Path("interfaceAppsCleaner/switch") + @Produces(MediaType.APPLICATION_JSON) + public Result interfaceAppsCleanerEnable(@FormParam("enabled") boolean enabled) { + Result result = new Result(); + try { + interfaceAppsIndexCleaner.setEnabled(enabled); + result.setSuccess(true); + } catch (Exception e) { + result.setSuccess(false); + result.setMessage(e.getMessage()); + } + return result; + } + MetaCenterResource setInterfaceAppsIndexCleaner(InterfaceAppsIndexCleaner cleaner) { interfaceAppsIndexCleaner = cleaner; return this; @@ -125,4 +150,9 @@ public MetaCenterResource setProvideDataService(ProvideDataService provideDataSe this.provideDataService = provideDataService; return this; } + + @VisibleForTesting + public void setAppRevisionCleaner(AppRevisionCleaner appRevisionCleaner) { + this.appRevisionCleaner = appRevisionCleaner; + } } diff --git a/server/server/meta/src/test/java/com/alipay/sofa/registry/server/meta/cleaner/InterfaceAppsIndexCleanerTest.java b/server/server/meta/src/test/java/com/alipay/sofa/registry/server/meta/cleaner/InterfaceAppsIndexCleanerTest.java index af1958fda..11bc511ec 100644 --- a/server/server/meta/src/test/java/com/alipay/sofa/registry/server/meta/cleaner/InterfaceAppsIndexCleanerTest.java +++ b/server/server/meta/src/test/java/com/alipay/sofa/registry/server/meta/cleaner/InterfaceAppsIndexCleanerTest.java @@ -18,13 +18,23 @@ import static org.mockito.Mockito.*; +import com.alipay.sofa.registry.cache.ConsecutiveSuccess; +import com.alipay.sofa.registry.common.model.console.PersistenceDataBuilder; +import com.alipay.sofa.registry.common.model.constants.ValueConstants; import com.alipay.sofa.registry.jdbc.config.MetadataConfig; import com.alipay.sofa.registry.jdbc.domain.AppRevisionDomain; import com.alipay.sofa.registry.jdbc.domain.InterfaceAppsIndexDomain; +import com.alipay.sofa.registry.jdbc.repository.impl.DateNowJdbcRepository; import com.alipay.sofa.registry.server.meta.AbstractMetaServerTestBase; +import com.alipay.sofa.registry.server.meta.bootstrap.config.MetaServerConfigBean; +import com.alipay.sofa.registry.server.meta.provide.data.DefaultProvideDataService; +import com.alipay.sofa.registry.store.api.DBResponse; +import com.alipay.sofa.registry.store.api.OperationStatus; import com.alipay.sofa.registry.store.api.repository.AppRevisionRepository; import com.alipay.sofa.registry.store.api.repository.InterfaceAppsRepository; +import java.util.Date; import org.assertj.core.util.Lists; +import org.junit.After; import org.junit.Before; import org.junit.Test; @@ -35,11 +45,32 @@ public class InterfaceAppsIndexCleanerTest extends AbstractMetaServerTestBase { public void beforeTest() throws Exception { makeMetaLeader(); interfaceAppsIndexCleaner = new InterfaceAppsIndexCleaner(metaLeaderService); + interfaceAppsIndexCleaner.dateNowRepository = mock(DateNowJdbcRepository.class); interfaceAppsIndexCleaner.appRevisionRepository = mock(AppRevisionRepository.class); interfaceAppsIndexCleaner.interfaceAppsRepository = mock(InterfaceAppsRepository.class); interfaceAppsIndexCleaner.metadataConfig = mock(MetadataConfig.class); + interfaceAppsIndexCleaner.consecutiveSuccess = new ConsecutiveSuccess(2, 10000); + interfaceAppsIndexCleaner.provideDataService = mock(DefaultProvideDataService.class); + when(interfaceAppsIndexCleaner.metadataConfig.getInterfaceAppsIndexRenewIntervalMinutes()) .thenReturn(10000); + interfaceAppsIndexCleaner.metaServerConfig = new MetaServerConfigBean(commonConfig); + + doReturn( + new DBResponse<>( + PersistenceDataBuilder.createPersistenceData( + ValueConstants.INTERFACE_APP_CLEANER_ENABLED_DATA_ID, "true"), + OperationStatus.SUCCESS)) + .when(interfaceAppsIndexCleaner.provideDataService) + .queryProvideData(anyString()); + + doReturn(true).when(interfaceAppsIndexCleaner.provideDataService).saveProvideData(any()); + } + + @After + public void afterTest() { + interfaceAppsIndexCleaner.cleaner.close(); + interfaceAppsIndexCleaner.renewer.close(); } @Test @@ -72,4 +103,36 @@ public void testClean() { mocked.renew(); mocked.renew(); } + + @Test + public void testCleanInterface() throws Exception { + InterfaceAppsIndexCleaner mocked = spy(interfaceAppsIndexCleaner); + doReturn(1).when(mocked.interfaceAppsRepository).cleanDeleted(any(), anyInt()); + doReturn(new Date()).when(mocked.dateNowRepository).getNow(); + mocked.renew(); + mocked.renew(); + mocked.cleanup(); + makeMetaNonLeader(); + mocked.cleanup(); + } + + @Test + public void testEnable() throws Exception { + InterfaceAppsIndexCleaner mocked = spy(interfaceAppsIndexCleaner); + doReturn(1).when(mocked.interfaceAppsRepository).cleanDeleted(any(), anyInt()); + doReturn(new Date()).when(mocked.dateNowRepository).getNow(); + mocked.setEnabled(true); + mocked.startRenew(); + mocked.cleaner.getWaitingMillis(); + mocked.cleaner.runUnthrowable(); + mocked.startCleaner(); + } + + @Test(expected = RuntimeException.class) + public void testEnableThrowEx() { + doThrow(new RuntimeException()) + .when(interfaceAppsIndexCleaner.provideDataService) + .saveProvideData(any()); + interfaceAppsIndexCleaner.setEnabled(true); + } } diff --git a/server/server/meta/src/test/java/com/alipay/sofa/registry/server/meta/lease/session/DefaultSessionServerManagerTest.java b/server/server/meta/src/test/java/com/alipay/sofa/registry/server/meta/lease/session/DefaultSessionServerManagerTest.java index 46461b071..eae799374 100644 --- a/server/server/meta/src/test/java/com/alipay/sofa/registry/server/meta/lease/session/DefaultSessionServerManagerTest.java +++ b/server/server/meta/src/test/java/com/alipay/sofa/registry/server/meta/lease/session/DefaultSessionServerManagerTest.java @@ -19,6 +19,7 @@ import static org.mockito.Mockito.when; import com.alipay.sofa.registry.common.model.ProcessId; +import com.alipay.sofa.registry.common.model.metaserver.Lease; import com.alipay.sofa.registry.common.model.metaserver.inter.heartbeat.HeartbeatRequest; import com.alipay.sofa.registry.common.model.metaserver.nodes.SessionNode; import com.alipay.sofa.registry.common.model.slot.SlotConfig; @@ -66,7 +67,7 @@ public void afterDefaultSessionManagerTest() throws Exception { public void testGetEpoch() throws TimeoutException, InterruptedException { Assert.assertEquals(0, sessionManager.getEpoch()); sessionManager.renew( - new SessionNode(randomURL(randomIp()), getDc(), ServerEnv.PROCESS_ID), 1000); + new SessionNode(randomURL(randomIp()), getDc(), ServerEnv.PROCESS_ID, 0), 1000); waitConditionUntilTimeOut(() -> sessionManager.getEpoch() > 0, 100); Assert.assertNotEquals(0, sessionManager.getEpoch()); } @@ -82,7 +83,8 @@ public void testRenew() throws Exception { String ip = randomIp(); long timestamp = System.currentTimeMillis(); SessionNode sessionNode = - new SessionNode(randomURL(ip), getDc(), new ProcessId(ip, timestamp, 1, random.nextInt())); + new SessionNode( + randomURL(ip), getDc(), new ProcessId(ip, timestamp, 1, random.nextInt()), 0); NotifyObserversCounter counter = new NotifyObserversCounter(); sessionManager.addObserver(counter); @@ -100,7 +102,8 @@ public void testRenew() throws Exception { new SessionNode( sessionNode.getNodeUrl(), getDc(), - new ProcessId(sessionNode.getIp(), timestamp, 2, random.nextInt())); + new ProcessId(sessionNode.getIp(), timestamp, 2, random.nextInt()), + 0); Assert.assertFalse(sessionManager.renew(sessionNode2, 1)); Assert.assertEquals(2, counter.getCounter()); } @@ -109,7 +112,7 @@ public void testRenew() throws Exception { public void testSessionServerManagerRefreshEpochOnlyOnceWhenNewRegistered() throws TimeoutException, InterruptedException { makeMetaLeader(); - SessionNode node = new SessionNode(randomURL(randomIp()), getDc(), ServerEnv.PROCESS_ID); + SessionNode node = new SessionNode(randomURL(randomIp()), getDc(), ServerEnv.PROCESS_ID, 0); sessionManager.renew(node, 1000); Assert.assertEquals(1, sessionManager.getSessionServerMetaInfo().getClusterMembers().size()); } @@ -130,7 +133,7 @@ public void testCancel() { protected List randomSessionNodes(int num) { List result = Lists.newArrayList(); for (int i = 0; i < num; i++) { - result.add(new SessionNode(randomURL(randomIp()), getDc(), ServerEnv.PROCESS_ID)); + result.add(new SessionNode(randomURL(randomIp()), getDc(), ServerEnv.PROCESS_ID, 0)); } return result; } @@ -141,7 +144,8 @@ public void testOnHeartbeat() { for (SessionNode sessionNode : sessionNodes) { sessionManager.renew(sessionNode, 1000); } - SessionNode sessionNode = new SessionNode(new URL(randomIp()), getDc(), ServerEnv.PROCESS_ID); + SessionNode sessionNode = + new SessionNode(new URL(randomIp()), getDc(), ServerEnv.PROCESS_ID, 0); sessionManager.renew(sessionNode, 1000); SlotTable slotTable = randomSlotTable(); @@ -157,4 +161,20 @@ public void testOnHeartbeat() { .setSlotTable(slotTable)); Assert.assertNotEquals(SlotTable.INIT, slotManager.getSlotTable()); } + + @Test + public void testWeightChange() { + List sessionNodes = randomSessionNodes(10); + for (SessionNode sessionNode : sessionNodes) { + sessionManager.register(new Lease<>(sessionNode, 10000)); + } + SessionNode sessionNode = + new SessionNode( + sessionNodes.get(0).getNodeUrl(), + sessionNodes.get(0).getRegionId(), + sessionNodes.get(0).getProcessId(), + 1); + sessionManager.renew(sessionNode, 1000); + Assert.assertEquals(10, sessionManager.getSessionServerMetaInfo().getClusterMembers().size()); + } } diff --git a/server/server/meta/src/test/java/com/alipay/sofa/registry/server/meta/metaserver/impl/DefaultCurrentDcMetaServerTest.java b/server/server/meta/src/test/java/com/alipay/sofa/registry/server/meta/metaserver/impl/DefaultCurrentDcMetaServerTest.java index f1272a31b..4017abd86 100644 --- a/server/server/meta/src/test/java/com/alipay/sofa/registry/server/meta/metaserver/impl/DefaultCurrentDcMetaServerTest.java +++ b/server/server/meta/src/test/java/com/alipay/sofa/registry/server/meta/metaserver/impl/DefaultCurrentDcMetaServerTest.java @@ -82,8 +82,8 @@ public void testGetSessionServers() { new VersionedList<>( DatumVersionUtil.nextId(), Lists.newArrayList( - new SessionNode(randomURL(), getDc(), ServerEnv.PROCESS_ID), - new SessionNode(randomURL(), getDc(), ServerEnv.PROCESS_ID)))); + new SessionNode(randomURL(), getDc(), ServerEnv.PROCESS_ID, 0), + new SessionNode(randomURL(), getDc(), ServerEnv.PROCESS_ID, 0)))); Assert.assertEquals( 2, metaServer.getSessionServerManager().getSessionServerMetaInfo().getClusterMembers().size()); diff --git a/server/server/meta/src/test/java/com/alipay/sofa/registry/server/meta/remoting/session/DefaultSessionServerServiceTest.java b/server/server/meta/src/test/java/com/alipay/sofa/registry/server/meta/remoting/session/DefaultSessionServerServiceTest.java index bfb8219d3..9e05ea93e 100644 --- a/server/server/meta/src/test/java/com/alipay/sofa/registry/server/meta/remoting/session/DefaultSessionServerServiceTest.java +++ b/server/server/meta/src/test/java/com/alipay/sofa/registry/server/meta/remoting/session/DefaultSessionServerServiceTest.java @@ -91,9 +91,9 @@ public Object answer(InvocationOnMock invocationOnMock) throws Throwable { new VersionedList<>( DatumVersionUtil.nextId(), Lists.newArrayList( - new SessionNode(randomURL(ip1), getDc(), ServerEnv.PROCESS_ID), - new SessionNode(randomURL(ip2), getDc(), ServerEnv.PROCESS_ID), - new SessionNode(randomURL(randomIp()), getDc(), ServerEnv.PROCESS_ID)))); + new SessionNode(randomURL(ip1), getDc(), ServerEnv.PROCESS_ID, 0), + new SessionNode(randomURL(ip2), getDc(), ServerEnv.PROCESS_ID, 0), + new SessionNode(randomURL(randomIp()), getDc(), ServerEnv.PROCESS_ID, 0)))); notifier.notifyProvideDataChange( new ProvideDataChangeEvent(ValueConstants.BLACK_LIST_DATA_ID, System.currentTimeMillis())); Thread.sleep(100); @@ -114,9 +114,9 @@ public void testBoltResponsePositive() throws InterruptedException, RequestExcep new VersionedList<>( DatumVersionUtil.nextId(), Lists.newArrayList( - new SessionNode(randomURL(ip1), getDc(), ServerEnv.PROCESS_ID), - new SessionNode(randomURL(ip2), getDc(), ServerEnv.PROCESS_ID), - new SessionNode(randomURL(randomIp()), getDc(), ServerEnv.PROCESS_ID)))); + new SessionNode(randomURL(ip1), getDc(), ServerEnv.PROCESS_ID, 0), + new SessionNode(randomURL(ip2), getDc(), ServerEnv.PROCESS_ID, 0), + new SessionNode(randomURL(randomIp()), getDc(), ServerEnv.PROCESS_ID, 0)))); Client client2 = spy(getRpcClient(scheduled, 10, "Response")); SessionNodeExchanger otherNodeExchanger = mock(SessionNodeExchanger.class); when(otherNodeExchanger.request(any(Request.class))) @@ -154,9 +154,9 @@ public void testBroadcastInvoke() throws Exception { new VersionedList<>( DatumVersionUtil.nextId(), Lists.newArrayList( - new SessionNode(randomURL(ip1), getDc(), ServerEnv.PROCESS_ID), - new SessionNode(randomURL(ip2), getDc(), ServerEnv.PROCESS_ID), - new SessionNode(randomURL(randomIp()), getDc(), ServerEnv.PROCESS_ID)))); + new SessionNode(randomURL(ip1), getDc(), ServerEnv.PROCESS_ID, 0), + new SessionNode(randomURL(ip2), getDc(), ServerEnv.PROCESS_ID, 0), + new SessionNode(randomURL(randomIp()), getDc(), ServerEnv.PROCESS_ID, 0)))); SessionNodeExchanger otherNodeExchanger = mock(SessionNodeExchanger.class); when(otherNodeExchanger.request(any(Request.class))).thenReturn(Object::new); diff --git a/server/server/meta/src/test/java/com/alipay/sofa/registry/server/meta/resource/MetaCenterResourceTest.java b/server/server/meta/src/test/java/com/alipay/sofa/registry/server/meta/resource/MetaCenterResourceTest.java index 4c9b3d7ac..7d4b72a2c 100644 --- a/server/server/meta/src/test/java/com/alipay/sofa/registry/server/meta/resource/MetaCenterResourceTest.java +++ b/server/server/meta/src/test/java/com/alipay/sofa/registry/server/meta/resource/MetaCenterResourceTest.java @@ -21,6 +21,7 @@ import com.alipay.sofa.registry.core.model.Result; import com.alipay.sofa.registry.jdbc.convertor.AppRevisionDomainConvertor; import com.alipay.sofa.registry.server.meta.AbstractMetaServerTestBase; +import com.alipay.sofa.registry.server.meta.cleaner.InterfaceAppsIndexCleaner; import com.alipay.sofa.registry.server.meta.provide.data.DefaultProvideDataNotifier; import com.alipay.sofa.registry.server.meta.provide.data.ProvideDataService; import org.junit.Assert; @@ -31,16 +32,19 @@ public class MetaCenterResourceTest { private MetaCenterResource metaCenterResource; private DefaultProvideDataNotifier dataNotifier; + private InterfaceAppsIndexCleaner interfaceAppsIndexCleaner; private ProvideDataService provideDataService = spy(new AbstractMetaServerTestBase.InMemoryProvideDataRepo()); @Before public void before() { dataNotifier = mock(DefaultProvideDataNotifier.class); + interfaceAppsIndexCleaner = mock(InterfaceAppsIndexCleaner.class); metaCenterResource = new MetaCenterResource() .setProvideDataNotifier(dataNotifier) - .setProvideDataService(provideDataService); + .setProvideDataService(provideDataService) + .setInterfaceAppsIndexCleaner(interfaceAppsIndexCleaner); } @Test @@ -56,4 +60,10 @@ public void testSetAppRevisionWriteSwitch() { new AppRevisionDomainConvertor.EnableConfig(false, true)); Assert.assertFalse(ret.isSuccess()); } + + @Test + public void testInterfaceStartCleaner() { + Result ret = metaCenterResource.interfaceAppsIndexClean(); + Assert.assertTrue(ret.isSuccess()); + } } diff --git a/server/server/meta/src/test/java/com/alipay/sofa/registry/server/meta/resource/RegistryCoreOpsResourceTest.java b/server/server/meta/src/test/java/com/alipay/sofa/registry/server/meta/resource/RegistryCoreOpsResourceTest.java index 31d81fa34..6da1a7033 100644 --- a/server/server/meta/src/test/java/com/alipay/sofa/registry/server/meta/resource/RegistryCoreOpsResourceTest.java +++ b/server/server/meta/src/test/java/com/alipay/sofa/registry/server/meta/resource/RegistryCoreOpsResourceTest.java @@ -21,8 +21,13 @@ import static org.mockito.Mockito.when; import com.alipay.sofa.registry.common.model.CommonResponse; +import com.alipay.sofa.registry.common.model.ProcessId; import com.alipay.sofa.registry.common.model.constants.ValueConstants; import com.alipay.sofa.registry.common.model.metaserver.Lease; +import com.alipay.sofa.registry.common.model.metaserver.nodes.DataNode; +import com.alipay.sofa.registry.common.model.metaserver.nodes.MetaNode; +import com.alipay.sofa.registry.common.model.metaserver.nodes.SessionNode; +import com.alipay.sofa.registry.common.model.store.URL; import com.alipay.sofa.registry.exception.SofaRegistryRuntimeException; import com.alipay.sofa.registry.server.meta.AbstractMetaServerTestBase; import com.alipay.sofa.registry.server.meta.lease.filter.DefaultForbiddenServerManager; @@ -119,4 +124,21 @@ public void testRejoinServerGroup() { Assert.assertTrue( registryForbiddenServerManager.allowSelect(new Lease<>(new SimpleNode("127.0.0.1"), 100))); } + + @Test + public void testDistinguishServerGroup() { + ProcessId processId1 = new ProcessId("test", 1, 2, 3); + CommonResponse response = resource.kickoffServer("testCell", "DATA", "127.0.0.1"); + Assert.assertTrue(response.isSuccess()); + Assert.assertTrue( + registryForbiddenServerManager.allowSelect( + new Lease<>( + new SessionNode(new URL("127.0.0.1", 8888), "regionId", processId1, 0), 100))); + Assert.assertFalse( + registryForbiddenServerManager.allowSelect( + new Lease<>(new DataNode(new URL("127.0.0.1", 8888), "regionId"), 100))); + Assert.assertTrue( + registryForbiddenServerManager.allowSelect( + new Lease<>(new MetaNode(new URL("127.0.0.1", 8888), "regionId"), 100))); + } } diff --git a/server/server/pom.xml b/server/server/pom.xml index 4fd7f419e..58c859097 100644 --- a/server/server/pom.xml +++ b/server/server/pom.xml @@ -5,7 +5,7 @@ com.alipay.sofa registry-server-parent - 6.4.0 + 6.5.1 ../pom.xml 4.0.0 diff --git a/server/server/session/pom.xml b/server/server/session/pom.xml index 3329609a8..9c7195071 100644 --- a/server/server/session/pom.xml +++ b/server/server/session/pom.xml @@ -5,7 +5,7 @@ com.alipay.sofa registry-server - 6.4.0 + 6.5.1 ../pom.xml 4.0.0 diff --git a/server/server/session/src/main/java/com/alipay/sofa/registry/server/session/node/service/MetaServerServiceImpl.java b/server/server/session/src/main/java/com/alipay/sofa/registry/server/session/node/service/MetaServerServiceImpl.java index 1cf3b8808..52dc5a5bc 100644 --- a/server/server/session/src/main/java/com/alipay/sofa/registry/server/session/node/service/MetaServerServiceImpl.java +++ b/server/server/session/src/main/java/com/alipay/sofa/registry/server/session/node/service/MetaServerServiceImpl.java @@ -24,6 +24,8 @@ import com.alipay.sofa.registry.common.model.slot.SlotConfig; import com.alipay.sofa.registry.common.model.slot.SlotTable; import com.alipay.sofa.registry.common.model.store.URL; +import com.alipay.sofa.registry.remoting.Server; +import com.alipay.sofa.registry.remoting.exchange.Exchange; import com.alipay.sofa.registry.server.session.bootstrap.SessionServerConfig; import com.alipay.sofa.registry.server.session.multi.cluster.DataCenterMetadataCache; import com.alipay.sofa.registry.server.session.remoting.DataNodeExchanger; @@ -51,6 +53,8 @@ public class MetaServerServiceImpl extends AbstractMetaServerService getDataInfoIdList() { return ret; } + @GET + @Path("{type}/getDataInfoIdListGroupByCount") + @Produces(MediaType.APPLICATION_JSON) + public List> getDataInfoIdListGroupByCount( + final @PathParam("type") String type) { + List> list = new ArrayList<>(); + String inputType = type.toUpperCase(); + switch (inputType) { + case PUB: + sessionDataStore + .getDatas() + .forEach((key, value) -> list.add(new AbstractMap.SimpleEntry<>(key, value.size()))); + break; + case WAT: + sessionWatchers + .getDatas() + .forEach((key, value) -> list.add(new AbstractMap.SimpleEntry<>(key, value.size()))); + break; + default: + sessionInterests + .getDatas() + .forEach((key, value) -> list.add(new AbstractMap.SimpleEntry<>(key, value.size()))); + break; + } + list.sort(Collections.reverseOrder(Map.Entry.comparingByValue())); + return list; + } + @GET @Path("checkSumDataInfoIdList") @Produces(MediaType.APPLICATION_JSON) diff --git a/server/server/session/src/main/java/com/alipay/sofa/registry/server/session/resource/SessionOpenResource.java b/server/server/session/src/main/java/com/alipay/sofa/registry/server/session/resource/SessionOpenResource.java index 212e9db82..aeec04264 100644 --- a/server/server/session/src/main/java/com/alipay/sofa/registry/server/session/resource/SessionOpenResource.java +++ b/server/server/session/src/main/java/com/alipay/sofa/registry/server/session/resource/SessionOpenResource.java @@ -16,12 +16,15 @@ */ package com.alipay.sofa.registry.server.session.resource; +import com.alipay.sofa.registry.common.model.metaserver.nodes.SessionNode; import com.alipay.sofa.registry.common.model.slot.Slot; +import com.alipay.sofa.registry.remoting.exchange.Exchange; import com.alipay.sofa.registry.server.session.bootstrap.SessionServerConfig; import com.alipay.sofa.registry.server.session.slot.SlotTableCache; import com.alipay.sofa.registry.server.shared.meta.MetaServerService; import com.alipay.sofa.registry.util.ParaCheckUtil; import com.google.common.base.Joiner; +import java.util.ArrayList; import java.util.List; import java.util.stream.Collectors; import javax.ws.rs.*; @@ -42,6 +45,8 @@ public class SessionOpenResource { @Autowired private SlotTableCache slotTableCache; + @Autowired private Exchange boltExchange; + @GET @Path("query.json") @Produces(MediaType.APPLICATION_JSON) @@ -63,6 +68,27 @@ public String getSessionServerList(@QueryParam("zone") String zone) { return Joiner.on(";").join(getSessionServerListJson(zone)); } + @GET + @Path("queryWithWeight") + @Produces(MediaType.TEXT_PLAIN) + public String getSessionServerListWithConnNum(@QueryParam("zone") String zone) { + if (StringUtils.isBlank(zone)) { + zone = sessionServerConfig.getSessionServerRegion(); + } + + if (StringUtils.isNotBlank(zone)) { + zone = zone.toUpperCase(); + } + return Joiner.on(";").join(getSessionServersWithConnNum(zone)); + } + + @GET + @Path("connectionNum") + @Produces(MediaType.TEXT_PLAIN) + public int getCurrentSessionConnNum() { + return boltExchange.getServer(sessionServerConfig.getServerPort()).getChannels().size(); + } + @GET @Path("alive") public String checkAlive() { @@ -91,6 +117,21 @@ private List getSessionServers(String zone) { return serverList; } + private List getSessionServersWithConnNum(String zone) { + List serverList = metaNodeService.getSessionNodeWithConnNumList(zone); + List serverWithConnNumList = new ArrayList<>(); + serverList.forEach( + item -> { + serverWithConnNumList.add( + item.getNodeUrl().getIpAddress() + + ":" + + sessionServerConfig.getServerPort() + + "?weight=" + + item.getWeight()); + }); + return serverWithConnNumList; + } + @GET @Path("slot") @Produces(MediaType.APPLICATION_JSON) diff --git a/server/server/session/src/test/java/com/alipay/sofa/registry/server/session/AbstractSessionServerTestBase.java b/server/server/session/src/test/java/com/alipay/sofa/registry/server/session/AbstractSessionServerTestBase.java index 1667ae6be..bc6f26abb 100644 --- a/server/server/session/src/test/java/com/alipay/sofa/registry/server/session/AbstractSessionServerTestBase.java +++ b/server/server/session/src/test/java/com/alipay/sofa/registry/server/session/AbstractSessionServerTestBase.java @@ -883,6 +883,11 @@ public synchronized void register(String appName, Set interfaceNames) { @Override public void renew(String interfaceName, String appName) {} + @Override + public int cleanDeleted(Date beforeTime, int limit) { + return 0; + } + @Override public void startSynced() {} diff --git a/server/server/session/src/test/java/com/alipay/sofa/registry/server/session/node/service/MetaServerServiceImplTest.java b/server/server/session/src/test/java/com/alipay/sofa/registry/server/session/node/service/MetaServerServiceImplTest.java index f42ed49fd..1905335d3 100644 --- a/server/server/session/src/test/java/com/alipay/sofa/registry/server/session/node/service/MetaServerServiceImplTest.java +++ b/server/server/session/src/test/java/com/alipay/sofa/registry/server/session/node/service/MetaServerServiceImplTest.java @@ -25,6 +25,7 @@ import com.alipay.sofa.registry.common.model.slot.SlotTable; import com.alipay.sofa.registry.log.Logger; import com.alipay.sofa.registry.log.LoggerFactory; +import com.alipay.sofa.registry.remoting.bolt.exchange.BoltExchange; import com.alipay.sofa.registry.server.session.TestUtils; import com.alipay.sofa.registry.server.session.bootstrap.SessionServerConfigBean; import com.alipay.sofa.registry.server.session.remoting.DataNodeExchanger; @@ -137,6 +138,7 @@ private void init() { sessionServerConfigBean = TestUtils.newSessionConfig("testDc"); impl.setSessionServerConfig(sessionServerConfigBean); + impl.setBoltExchange(new BoltExchange()); Assert.assertEquals( impl.getRenewIntervalSecs(), sessionServerConfigBean.getSchedulerHeartbeatIntervalSecs()); slotTableCache = new SlotTableCacheImpl(); diff --git a/server/server/session/src/test/java/com/alipay/sofa/registry/server/session/providedata/FetchStopPushServiceTest.java b/server/server/session/src/test/java/com/alipay/sofa/registry/server/session/providedata/FetchStopPushServiceTest.java index 6443c2996..70595e6d1 100644 --- a/server/server/session/src/test/java/com/alipay/sofa/registry/server/session/providedata/FetchStopPushServiceTest.java +++ b/server/server/session/src/test/java/com/alipay/sofa/registry/server/session/providedata/FetchStopPushServiceTest.java @@ -37,7 +37,7 @@ public void beforeFetchStopPushServiceTest() { @Test public void test() { - Assert.assertTrue(isStopPushSwitch()); + Assert.assertFalse(isStopPushSwitch()); Assert.assertTrue(doProcess(storage.get(), new StopPushStorage(2L, false))); Assert.assertEquals(isStopPushSwitch(), false); diff --git a/server/server/shared/pom.xml b/server/server/shared/pom.xml index e54583827..9debc42e0 100644 --- a/server/server/shared/pom.xml +++ b/server/server/shared/pom.xml @@ -5,7 +5,7 @@ registry-server com.alipay.sofa - 6.4.0 + 6.5.1 4.0.0 diff --git a/server/server/shared/src/main/java/com/alipay/sofa/registry/server/shared/meta/AbstractMetaServerService.java b/server/server/shared/src/main/java/com/alipay/sofa/registry/server/shared/meta/AbstractMetaServerService.java index 597f5643a..a08330f5b 100644 --- a/server/server/shared/src/main/java/com/alipay/sofa/registry/server/shared/meta/AbstractMetaServerService.java +++ b/server/server/shared/src/main/java/com/alipay/sofa/registry/server/shared/meta/AbstractMetaServerService.java @@ -366,6 +366,19 @@ public List getSessionServerList(String zonename) { return serverList; } + public List getSessionNodeWithConnNumList(String zonename) { + List serverList = new ArrayList<>(); + for (SessionNode sessionNode : getSessionNodes().values()) { + if (StringUtils.isBlank(zonename) || zonename.equals(sessionNode.getRegionId())) { + URL url = sessionNode.getNodeUrl(); + if (url != null) { + serverList.add(sessionNode); + } + } + } + return serverList; + } + @Override public long getSessionServerEpoch() { return state.sessionServerEpoch; diff --git a/server/server/shared/src/main/java/com/alipay/sofa/registry/server/shared/meta/MetaServerService.java b/server/server/shared/src/main/java/com/alipay/sofa/registry/server/shared/meta/MetaServerService.java index 44255f00d..0fa96a7a9 100644 --- a/server/server/shared/src/main/java/com/alipay/sofa/registry/server/shared/meta/MetaServerService.java +++ b/server/server/shared/src/main/java/com/alipay/sofa/registry/server/shared/meta/MetaServerService.java @@ -20,6 +20,7 @@ import com.alipay.sofa.registry.common.model.metaserver.FetchSystemPropertyResult; import com.alipay.sofa.registry.common.model.metaserver.ProvideData; import com.alipay.sofa.registry.common.model.metaserver.SlotTableChangeEvent; +import com.alipay.sofa.registry.common.model.metaserver.nodes.SessionNode; import com.alipay.sofa.registry.common.model.slot.SlotTableStatusResponse; import java.util.List; import java.util.Map; @@ -72,6 +73,8 @@ public interface MetaServerService { */ List getSessionServerList(String zonename); + List getSessionNodeWithConnNumList(String zonename); + Set getSessionProcessIds(); /** * Gets get data server list. diff --git a/server/server/shared/src/main/java/com/alipay/sofa/registry/server/shared/providedata/BaseStopPushService.java b/server/server/shared/src/main/java/com/alipay/sofa/registry/server/shared/providedata/BaseStopPushService.java index a80c0bf66..d2180afb0 100644 --- a/server/server/shared/src/main/java/com/alipay/sofa/registry/server/shared/providedata/BaseStopPushService.java +++ b/server/server/shared/src/main/java/com/alipay/sofa/registry/server/shared/providedata/BaseStopPushService.java @@ -32,7 +32,7 @@ public abstract class BaseStopPushService @Autowired private ProvideDataRepository provideDataRepository; - private static final StopPushStorage INIT = new StopPushStorage(INIT_VERSION, true); + private static final StopPushStorage INIT = new StopPushStorage(INIT_VERSION, false); public BaseStopPushService() { // default value is stop.push diff --git a/server/server/shared/src/test/java/com/alipay/sofa/registry/server/shared/meta/MetaServerServiceTest.java b/server/server/shared/src/test/java/com/alipay/sofa/registry/server/shared/meta/MetaServerServiceTest.java index 81dc95172..20baf09d8 100644 --- a/server/server/shared/src/test/java/com/alipay/sofa/registry/server/shared/meta/MetaServerServiceTest.java +++ b/server/server/shared/src/test/java/com/alipay/sofa/registry/server/shared/meta/MetaServerServiceTest.java @@ -121,9 +121,9 @@ public void testHandleHeartbeatResp() { new VersionedList( 1, Lists.newArrayList( - new SessionNode(new URL("192.168.1.2"), "zoneA", ServerEnv.PROCESS_ID), + new SessionNode(new URL("192.168.1.2"), "zoneA", ServerEnv.PROCESS_ID, 0), new SessionNode( - new URL("192.168.1.3"), "zoneB", new ProcessId("test", 1, 1, 1)))), + new URL("192.168.1.3"), "zoneB", new ProcessId("test", 1, 1, 1), 0))), "test", 100, Collections.emptyMap()); @@ -152,6 +152,9 @@ public void testHandleHeartbeatResp() { Assert.assertTrue(zones.contains("192.168.1.2")); Assert.assertTrue(zones.contains("192.168.1.3")); + List nodes = mockServerService.getSessionNodeWithConnNumList(""); + Assert.assertEquals(nodes.size(), 2); + Assert.assertEquals(2, mockServerService.getSessionProcessIds().size()); Assert.assertTrue(mockServerService.getSessionProcessIds().contains(ServerEnv.PROCESS_ID)); zones = mockServerService.getSessionServerList("zoneC"); @@ -235,9 +238,9 @@ public void testRenewNode() { new VersionedList( 1, Lists.newArrayList( - new SessionNode(new URL("192.168.1.2"), "zoneA", ServerEnv.PROCESS_ID), + new SessionNode(new URL("192.168.1.2"), "zoneA", ServerEnv.PROCESS_ID, 0), new SessionNode( - new URL("192.168.1.3"), "zoneB", new ProcessId("test", 1, 1, 1)))), + new URL("192.168.1.3"), "zoneB", new ProcessId("test", 1, 1, 1), 0))), "test", 100, Collections.emptyMap()); diff --git a/server/server/shared/src/test/java/com/alipay/sofa/registry/server/shared/util/NodeUtilsTest.java b/server/server/shared/src/test/java/com/alipay/sofa/registry/server/shared/util/NodeUtilsTest.java index cb93c6f57..4adad5ee5 100644 --- a/server/server/shared/src/test/java/com/alipay/sofa/registry/server/shared/util/NodeUtilsTest.java +++ b/server/server/shared/src/test/java/com/alipay/sofa/registry/server/shared/util/NodeUtilsTest.java @@ -28,8 +28,8 @@ public class NodeUtilsTest { @Test public void test() { Assert.assertTrue(NodeUtils.transferNodeToIpList(Collections.EMPTY_LIST).isEmpty()); - SessionNode node1 = new SessionNode(new URL("xx", 12), "test", null); - SessionNode node2 = new SessionNode(new URL("xyz", 34), "test", null); + SessionNode node1 = new SessionNode(new URL("xx", 12), "test", null, 0); + SessionNode node2 = new SessionNode(new URL("xyz", 34), "test", null, 0); List list = NodeUtils.transferNodeToIpList(Lists.newArrayList(node1, node2)); Assert.assertEquals(list.get(0), "xx"); Assert.assertEquals(list.get(1), "xyz"); diff --git a/server/store/api/pom.xml b/server/store/api/pom.xml index 37009f207..525e29879 100644 --- a/server/store/api/pom.xml +++ b/server/store/api/pom.xml @@ -5,7 +5,7 @@ com.alipay.sofa registry-store - 6.4.0 + 6.5.1 ../pom.xml 4.0.0 diff --git a/server/store/api/src/main/java/com/alipay/sofa/registry/store/api/repository/InterfaceAppsRepository.java b/server/store/api/src/main/java/com/alipay/sofa/registry/store/api/repository/InterfaceAppsRepository.java index e55b8a349..dc585d786 100644 --- a/server/store/api/src/main/java/com/alipay/sofa/registry/store/api/repository/InterfaceAppsRepository.java +++ b/server/store/api/src/main/java/com/alipay/sofa/registry/store/api/repository/InterfaceAppsRepository.java @@ -18,6 +18,7 @@ import com.alipay.sofa.registry.common.model.appmeta.InterfaceMapping; import com.alipay.sofa.registry.store.api.multi.MultiDataCenterListener; +import java.util.Date; import java.util.Map; import java.util.Set; @@ -39,6 +40,8 @@ public interface InterfaceAppsRepository extends MultiDataCenterListener { void renew(String interfaceName, String appName); + int cleanDeleted(Date beforeTime, int limit); + void startSynced(); void waitSynced(); diff --git a/server/store/jdbc/pom.xml b/server/store/jdbc/pom.xml index 6eb600e15..9c938c781 100644 --- a/server/store/jdbc/pom.xml +++ b/server/store/jdbc/pom.xml @@ -6,7 +6,7 @@ com.alipay.sofa registry-store - 6.4.0 + 6.5.1 ../pom.xml diff --git a/server/store/jdbc/src/main/java/com/alipay/sofa/registry/jdbc/mapper/InterfaceAppsIndexMapper.java b/server/store/jdbc/src/main/java/com/alipay/sofa/registry/jdbc/mapper/InterfaceAppsIndexMapper.java index 1875b5739..6cc7dbc45 100644 --- a/server/store/jdbc/src/main/java/com/alipay/sofa/registry/jdbc/mapper/InterfaceAppsIndexMapper.java +++ b/server/store/jdbc/src/main/java/com/alipay/sofa/registry/jdbc/mapper/InterfaceAppsIndexMapper.java @@ -17,6 +17,7 @@ package com.alipay.sofa.registry.jdbc.mapper; import com.alipay.sofa.registry.jdbc.domain.InterfaceAppsIndexDomain; +import java.util.Date; import java.util.List; import java.util.Set; import org.apache.ibatis.annotations.Param; @@ -55,4 +56,9 @@ List queryLargeThan( @Param("dataCenters") Set dataCenters, @Param("maxId") long maxId, @Param("limitCount") int limitCount); + + int cleanDeleted( + @Param("dataCenter") String dataCenter, + @Param("beforeTime") Date beforeTime, + @Param("limit") int limit); } diff --git a/server/store/jdbc/src/main/java/com/alipay/sofa/registry/jdbc/repository/impl/InterfaceAppsJdbcRepository.java b/server/store/jdbc/src/main/java/com/alipay/sofa/registry/jdbc/repository/impl/InterfaceAppsJdbcRepository.java index 75a342f21..b38b8faee 100644 --- a/server/store/jdbc/src/main/java/com/alipay/sofa/registry/jdbc/repository/impl/InterfaceAppsJdbcRepository.java +++ b/server/store/jdbc/src/main/java/com/alipay/sofa/registry/jdbc/repository/impl/InterfaceAppsJdbcRepository.java @@ -104,6 +104,12 @@ public void renew(String interfaceName, String appName) { defaultCommonConfig.getDefaultClusterId(), interfaceName, appName)); } + @Override + public int cleanDeleted(Date beforeTime, int limit) { + return interfaceAppsIndexMapper.cleanDeleted( + defaultCommonConfig.getDefaultClusterId(), beforeTime, limit); + } + @Override public void startSynced() { ParaCheckUtil.checkNotEmpty(dataCenters, "dataCenters"); @@ -219,8 +225,10 @@ protected void preList(InterfaceAppsIndexContainer newContainer) { for (Entry entry : currentMappings.entrySet()) { InterfaceMapping newMapping = newMappings.get(entry.getKey()); InterfaceMapping currentMapping = entry.getValue(); - if (newMapping == null - || newMapping.getNanosVersion() < currentMapping.getNanosVersion() + if (newMapping == null) { + continue; + } + if (newMapping.getNanosVersion() < currentMapping.getNanosVersion() || (newMapping.getNanosVersion() == currentMapping.getNanosVersion() && !newMapping.getApps().equals(currentMapping.getApps()))) { if (conflictCallback != null) { diff --git a/server/store/jdbc/src/main/resources/h2-mapper/interface_apps_index.xml b/server/store/jdbc/src/main/resources/h2-mapper/interface_apps_index.xml index 662c1ff66..5a7842657 100644 --- a/server/store/jdbc/src/main/resources/h2-mapper/interface_apps_index.xml +++ b/server/store/jdbc/src/main/resources/h2-mapper/interface_apps_index.xml @@ -45,4 +45,12 @@ order by id limit #{limitCount} + + + + + diff --git a/server/store/jdbc/src/main/resources/mapper/interface_apps_index.xml b/server/store/jdbc/src/main/resources/mapper/interface_apps_index.xml index b9e7f7661..75efa0bfb 100644 --- a/server/store/jdbc/src/main/resources/mapper/interface_apps_index.xml +++ b/server/store/jdbc/src/main/resources/mapper/interface_apps_index.xml @@ -41,4 +41,11 @@ order by id limit #{limitCount} + + + + diff --git a/server/store/jdbc/src/test/java/com/alipay/sofa/registry/jdbc/repository/impl/InterfaceAppsJdbcRepositoryTest.java b/server/store/jdbc/src/test/java/com/alipay/sofa/registry/jdbc/repository/impl/InterfaceAppsJdbcRepositoryTest.java index 7a0e133e1..6ab098641 100644 --- a/server/store/jdbc/src/test/java/com/alipay/sofa/registry/jdbc/repository/impl/InterfaceAppsJdbcRepositoryTest.java +++ b/server/store/jdbc/src/test/java/com/alipay/sofa/registry/jdbc/repository/impl/InterfaceAppsJdbcRepositoryTest.java @@ -32,10 +32,7 @@ import com.alipay.sofa.registry.util.TimestampUtil; import com.google.common.collect.Lists; import com.google.common.collect.Sets; -import java.util.ArrayList; -import java.util.Collections; -import java.util.List; -import java.util.Set; +import java.util.*; import java.util.concurrent.atomic.AtomicInteger; import org.junit.Assert; import org.junit.Before; @@ -131,4 +128,10 @@ public void testRuntimeException() { impl.register(app1, Collections.singleton(service)); } } + + @Test + public void testCleaned() { + interfaceAppsJdbcRepository.renew("testInterfaceName", "testApp"); + Assert.assertEquals(0, interfaceAppsJdbcRepository.cleanDeleted(new Date(), 0)); + } } diff --git a/server/store/jraft/pom.xml b/server/store/jraft/pom.xml index 1ffbb8f36..7db147208 100644 --- a/server/store/jraft/pom.xml +++ b/server/store/jraft/pom.xml @@ -6,7 +6,7 @@ com.alipay.sofa registry-store - 6.4.0 + 6.5.1 ../pom.xml diff --git a/server/store/jraft/src/main/java/com/alipay/sofa/registry/jraft/repository/impl/InterfaceAppsRaftRepository.java b/server/store/jraft/src/main/java/com/alipay/sofa/registry/jraft/repository/impl/InterfaceAppsRaftRepository.java index ddfbddfcd..08dd48113 100644 --- a/server/store/jraft/src/main/java/com/alipay/sofa/registry/jraft/repository/impl/InterfaceAppsRaftRepository.java +++ b/server/store/jraft/src/main/java/com/alipay/sofa/registry/jraft/repository/impl/InterfaceAppsRaftRepository.java @@ -20,6 +20,7 @@ import com.alipay.sofa.registry.log.Logger; import com.alipay.sofa.registry.log.LoggerFactory; import com.alipay.sofa.registry.store.api.repository.InterfaceAppsRepository; +import java.util.Date; import java.util.Map; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; @@ -52,6 +53,11 @@ public void register(String appName, Set interfaceNames) { @Override public void renew(String interfaceName, String appName) {} + @Override + public int cleanDeleted(Date beforeTime, int limit) { + return 0; + } + @Override public void startSynced() {} diff --git a/server/store/pom.xml b/server/store/pom.xml index f491cf621..8a1e076a8 100644 --- a/server/store/pom.xml +++ b/server/store/pom.xml @@ -5,7 +5,7 @@ com.alipay.sofa registry-server-parent - 6.4.0 + 6.5.1 ../pom.xml 4.0.0 diff --git a/test/pom.xml b/test/pom.xml index 2f9d3153d..18c76c50b 100644 --- a/test/pom.xml +++ b/test/pom.xml @@ -5,7 +5,7 @@ com.alipay.sofa registry-parent - 6.4.0 + 6.5.1 ../pom.xml 4.0.0 @@ -122,7 +122,7 @@ none:none org.junit:com.springsource.org.junit - @{argLine} -Xmx4096m -XX:PermSize=256m -XX:MaxPermSize=512m -Dfile.encoding=UTF8 + @{argLine} -Xmx6144m -XX:PermSize=256m -XX:MaxPermSize=512m -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=/tmp -Dfile.encoding=UTF8 **/*Test.java diff --git a/test/src/test/java/com/alipay/sofa/registry/test/metadata/InterfaceAppsJdbcRepositoryTest.java b/test/src/test/java/com/alipay/sofa/registry/test/metadata/InterfaceAppsJdbcRepositoryTest.java index d491b92b4..61b6db2d3 100644 --- a/test/src/test/java/com/alipay/sofa/registry/test/metadata/InterfaceAppsJdbcRepositoryTest.java +++ b/test/src/test/java/com/alipay/sofa/registry/test/metadata/InterfaceAppsJdbcRepositoryTest.java @@ -20,10 +20,7 @@ import com.alipay.sofa.registry.server.session.metadata.MetadataCacheRegistry; import com.alipay.sofa.registry.store.api.repository.InterfaceAppsRepository; import com.alipay.sofa.registry.test.BaseIntegrationTest; -import java.util.ArrayList; -import java.util.Collections; -import java.util.HashSet; -import java.util.List; +import java.util.*; import org.junit.Assert; import org.junit.Before; import org.junit.Test; @@ -82,5 +79,6 @@ public void batchSaveTest() throws InterruptedException { Assert.assertEquals(1, appNames.getApps().size()); Assert.assertTrue(appNames.getApps().contains(app)); } + interfaceAppsJdbcRepository.cleanDeleted(new Date(), 1); } }