diff --git a/.github/workflows/e2e-k8s.yml b/.github/workflows/e2e-k8s.yml index c0b6775f8b49..bfd0e1f3834a 100644 --- a/.github/workflows/e2e-k8s.yml +++ b/.github/workflows/e2e-k8s.yml @@ -63,6 +63,8 @@ jobs: submodules: true - name: Test conditional disk cleanup run: bash .github/scripts/free-disk-space-test.sh + - name: Test local HTTP upstream fixture and Compose wiring + run: python3 -B shenyu-e2e/shenyu-e2e-case/compose/fixtures/test_httpbin.py - uses: ./actions/paths-filter id: filter with: diff --git a/shenyu-e2e/shenyu-e2e-case/compose/fixtures/httpbin.py b/shenyu-e2e/shenyu-e2e-case/compose/fixtures/httpbin.py new file mode 100644 index 000000000000..08c291145c4c --- /dev/null +++ b/shenyu-e2e/shenyu-e2e-case/compose/fixtures/httpbin.py @@ -0,0 +1,57 @@ +# 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. + +"""Local upstream for the Divide URI and HTTP method routing scenarios.""" + +import json +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from urllib.parse import parse_qs, urlsplit + + +class HttpbinHandler(BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" + + def respond(self): + path = urlsplit(self.path) + data = self.rfile.read(int(self.headers.get("Content-Length", "0"))) + status = 200 + if path.path == "/status/200": + body = b"" + elif path.path == "/anything" or path.path.startswith("/anything/"): + body = json.dumps({ + "method": self.command, + "url": self.path, + "args": parse_qs(path.query), + "data": data.decode("utf-8"), + }).encode("utf-8") + else: + status = 404 + body = b'{"error":"unknown fixture endpoint"}' + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + if self.command != "HEAD": + self.wfile.write(body) + + do_GET = respond + do_POST = respond + do_PUT = respond + do_DELETE = respond + do_HEAD = respond + + +if __name__ == "__main__": + ThreadingHTTPServer(("0.0.0.0", 80), HttpbinHandler).serve_forever() diff --git a/shenyu-e2e/shenyu-e2e-case/compose/fixtures/httpbin.yml b/shenyu-e2e/shenyu-e2e-case/compose/fixtures/httpbin.yml new file mode 100644 index 000000000000..762d031d120b --- /dev/null +++ b/shenyu-e2e/shenyu-e2e-case/compose/fixtures/httpbin.yml @@ -0,0 +1,31 @@ +# 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. + +services: + httpbin: + image: python:3.12-alpine + command: ["python", "-B", "-u", "/opt/httpbin.py"] + volumes: + - ./httpbin.py:/opt/httpbin.py:ro + healthcheck: + test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://127.0.0.1/anything', timeout=2)"] + interval: 2s + timeout: 3s + retries: 15 + networks: + shenyu: + # Existing routing scenarios resolve their upstream on this test network. + aliases: + - httpbin.org diff --git a/shenyu-e2e/shenyu-e2e-case/compose/fixtures/test_httpbin.py b/shenyu-e2e/shenyu-e2e-case/compose/fixtures/test_httpbin.py new file mode 100644 index 000000000000..ec0741a1a6a8 --- /dev/null +++ b/shenyu-e2e/shenyu-e2e-case/compose/fixtures/test_httpbin.py @@ -0,0 +1,97 @@ +# 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. + +"""Exercise the upstream over HTTP and validate every consuming Compose file.""" + +import http.client +import json +from pathlib import Path +import subprocess +import threading +import unittest + +from httpbin import HttpbinHandler, ThreadingHTTPServer + + +class HttpbinTest(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.server = ThreadingHTTPServer(("127.0.0.1", 0), HttpbinHandler) + cls.thread = threading.Thread(target=cls.server.serve_forever) + cls.thread.start() + + @classmethod + def tearDownClass(cls): + cls.server.shutdown() + cls.server.server_close() + cls.thread.join() + + def setUp(self): + self.connection = http.client.HTTPConnection(*self.server.server_address, timeout=3) + + def tearDown(self): + self.connection.close() + + def test_echo_methods_paths_and_request_bodies(self): + for method in ("GET", "POST", "PUT", "DELETE"): + for path in ("/anything", "/anything/xx", "/anything/xx/yy", "/anything/xxx"): + with self.subTest(method=method, path=path): + self.connection.request(method, path + "?id=23", body="request body") + response = self.connection.getresponse() + self.assertEqual(200, response.status) + body = json.loads(response.read()) + self.assertEqual(method, body["method"]) + self.assertEqual(path + "?id=23", body["url"]) + self.assertEqual({"id": ["23"]}, body["args"]) + self.assertEqual("request body", body["data"]) + + def test_status_200_has_empty_body_for_every_method(self): + for method in ("GET", "POST", "PUT", "DELETE", "HEAD"): + with self.subTest(method=method): + self.connection.request(method, "/status/200") + response = self.connection.getresponse() + self.assertEqual(200, response.status) + self.assertEqual(b"", response.read()) + + def test_unknown_path_is_not_a_success(self): + self.connection.request("GET", "/not-a-fixture") + response = self.connection.getresponse() + self.assertEqual(404, response.status) + response.read() + + def test_compose_uses_healthy_local_upstream(self): + directory = Path(__file__).resolve().parent + compose = directory.parent + files = sorted((compose / "storage").glob("shenyu-storage-*.yml")) + files += sorted((compose / "sync").glob("shenyu-sync-websocket-cluster-*.yml")) + self.assertEqual(6, len(files)) + for path in files: + with self.subTest(compose=path.name): + result = subprocess.run( + ["docker", "compose", "-f", str(path), "config", "--format", "json"], + check=True, capture_output=True, text=True, + ) + services = json.loads(result.stdout)["services"] + fixture = services["httpbin"] + self.assertIn("httpbin.org", fixture["networks"]["shenyu"]["aliases"]) + self.assertEqual("service_healthy", services["shenyu-bootstrap"]["depends_on"]["httpbin"]["condition"]) + mount = next(v for v in fixture["volumes"] if v["target"] == "/opt/httpbin.py") + self.assertEqual(directory / "httpbin.py", Path(mount["source"])) + self.assertTrue(mount["read_only"]) + self.assertIn("/anything", fixture["healthcheck"]["test"][-1]) + + +if __name__ == "__main__": + unittest.main() diff --git a/shenyu-e2e/shenyu-e2e-case/compose/storage/shenyu-storage-h2.yml b/shenyu-e2e/shenyu-e2e-case/compose/storage/shenyu-storage-h2.yml index c35ccd1dbecb..a722a0501f57 100644 --- a/shenyu-e2e/shenyu-e2e-case/compose/storage/shenyu-storage-h2.yml +++ b/shenyu-e2e/shenyu-e2e-case/compose/storage/shenyu-storage-h2.yml @@ -17,6 +17,10 @@ version: '3.9' services: + httpbin: + extends: + file: ../fixtures/httpbin.yml + service: httpbin shenyu-admin: image: apache/shenyu-admin:latest @@ -63,6 +67,8 @@ services: start_period: 30s restart: always depends_on: + httpbin: + condition: service_healthy shenyu-admin: condition: service_healthy networks: diff --git a/shenyu-e2e/shenyu-e2e-case/compose/storage/shenyu-storage-mysql.yml b/shenyu-e2e/shenyu-e2e-case/compose/storage/shenyu-storage-mysql.yml index 06f0b40476ec..2141c9d9a8ec 100644 --- a/shenyu-e2e/shenyu-e2e-case/compose/storage/shenyu-storage-mysql.yml +++ b/shenyu-e2e/shenyu-e2e-case/compose/storage/shenyu-storage-mysql.yml @@ -17,6 +17,10 @@ version: '3.9' services: + httpbin: + extends: + file: ../fixtures/httpbin.yml + service: httpbin shenyu-mysql: image: mysql:8.0 container_name: shenyu-mysql @@ -98,6 +102,8 @@ services: start_period: 30s restart: always depends_on: + httpbin: + condition: service_healthy shenyu-admin: condition: service_healthy networks: diff --git a/shenyu-e2e/shenyu-e2e-case/compose/storage/shenyu-storage-opengauss.yml b/shenyu-e2e/shenyu-e2e-case/compose/storage/shenyu-storage-opengauss.yml index d0c3b6622836..53f62e6303da 100644 --- a/shenyu-e2e/shenyu-e2e-case/compose/storage/shenyu-storage-opengauss.yml +++ b/shenyu-e2e/shenyu-e2e-case/compose/storage/shenyu-storage-opengauss.yml @@ -17,6 +17,10 @@ version: '3.9' services: + httpbin: + extends: + file: ../fixtures/httpbin.yml + service: httpbin shenyu-opengauss: image: enmotech/opengauss-lite:5.0.1 container_name: shenyu-opengauss @@ -97,6 +101,8 @@ services: start_period: 30s restart: always depends_on: + httpbin: + condition: service_healthy shenyu-admin: condition: service_healthy networks: diff --git a/shenyu-e2e/shenyu-e2e-case/compose/storage/shenyu-storage-postgres.yml b/shenyu-e2e/shenyu-e2e-case/compose/storage/shenyu-storage-postgres.yml index 29e3e87acae0..147d78b699dc 100644 --- a/shenyu-e2e/shenyu-e2e-case/compose/storage/shenyu-storage-postgres.yml +++ b/shenyu-e2e/shenyu-e2e-case/compose/storage/shenyu-storage-postgres.yml @@ -17,6 +17,10 @@ version: '3.9' services: + httpbin: + extends: + file: ../fixtures/httpbin.yml + service: httpbin shenyu-postgres: image: postgres:14 container_name: shenyu-postgres @@ -97,6 +101,8 @@ services: start_period: 30s restart: always depends_on: + httpbin: + condition: service_healthy shenyu-admin: condition: service_healthy networks: diff --git a/shenyu-e2e/shenyu-e2e-case/compose/sync/shenyu-sync-websocket-cluster-jdbc.yml b/shenyu-e2e/shenyu-e2e-case/compose/sync/shenyu-sync-websocket-cluster-jdbc.yml index 85ac6bf35c80..7142d5d8d52b 100644 --- a/shenyu-e2e/shenyu-e2e-case/compose/sync/shenyu-sync-websocket-cluster-jdbc.yml +++ b/shenyu-e2e/shenyu-e2e-case/compose/sync/shenyu-sync-websocket-cluster-jdbc.yml @@ -17,6 +17,10 @@ version: '3.9' services: + httpbin: + extends: + file: ../fixtures/httpbin.yml + service: httpbin shenyu-mysql: image: mysql:8.0 container_name: shenyu-mysql @@ -133,6 +137,8 @@ services: start_period: 30s restart: always depends_on: + httpbin: + condition: service_healthy shenyu-admin-1: condition: service_healthy shenyu-admin-2: diff --git a/shenyu-e2e/shenyu-e2e-case/compose/sync/shenyu-sync-websocket-cluster-zookeeper.yml b/shenyu-e2e/shenyu-e2e-case/compose/sync/shenyu-sync-websocket-cluster-zookeeper.yml index 58d3d41b5b6b..79dea728bef4 100644 --- a/shenyu-e2e/shenyu-e2e-case/compose/sync/shenyu-sync-websocket-cluster-zookeeper.yml +++ b/shenyu-e2e/shenyu-e2e-case/compose/sync/shenyu-sync-websocket-cluster-zookeeper.yml @@ -17,6 +17,10 @@ version: '3.9' services: + httpbin: + extends: + file: ../fixtures/httpbin.yml + service: httpbin shenyu-zookeeper-cluster: image: zookeeper:3.9 container_name: shenyu-zookeeper-cluster @@ -158,6 +162,8 @@ services: start_period: 30s restart: always depends_on: + httpbin: + condition: service_healthy shenyu-admin-1: condition: service_healthy shenyu-admin-2: diff --git a/shenyu-e2e/shenyu-e2e-case/shenyu-e2e-case-logging-rocketmq/src/test/java/org/apache/shenyu/e2e/testcase/logging/rocketmq/DividePluginCases.java b/shenyu-e2e/shenyu-e2e-case/shenyu-e2e-case-logging-rocketmq/src/test/java/org/apache/shenyu/e2e/testcase/logging/rocketmq/DividePluginCases.java index 87ce0e4a9643..a0cb0fdd6a41 100644 --- a/shenyu-e2e/shenyu-e2e-case/shenyu-e2e-case-logging-rocketmq/src/test/java/org/apache/shenyu/e2e/testcase/logging/rocketmq/DividePluginCases.java +++ b/shenyu-e2e/shenyu-e2e-case/shenyu-e2e-case-logging-rocketmq/src/test/java/org/apache/shenyu/e2e/testcase/logging/rocketmq/DividePluginCases.java @@ -19,10 +19,11 @@ import com.google.common.collect.Lists; import io.restassured.http.Method; -import org.apache.commons.collections.CollectionUtils; -import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer; -import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; -import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently; +import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer; +import org.apache.rocketmq.client.exception.MQClientException; +import org.apache.rocketmq.common.consumer.ConsumeFromWhere; +import org.apache.rocketmq.common.message.MessageQueue; +import org.apache.rocketmq.common.protocol.ResponseCode; import org.apache.shenyu.e2e.engine.scenario.ShenYuScenarioProvider; import org.apache.shenyu.e2e.engine.scenario.specification.ScenarioSpec; import org.apache.shenyu.e2e.engine.scenario.specification.ShenYuBeforeEachSpec; @@ -35,8 +36,12 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.nio.charset.StandardCharsets; import java.time.Duration; +import java.util.Collection; import java.util.List; +import java.util.Objects; +import java.util.UUID; import java.util.concurrent.atomic.AtomicBoolean; import static org.apache.shenyu.e2e.engine.scenario.function.HttpCheckers.exists; @@ -55,7 +60,7 @@ public class DividePluginCases implements ShenYuScenarioProvider { private static final String TEST = "/http/order/findById?id=123"; - private static final Duration LOG_CONSUME_TIMEOUT = Duration.ofSeconds(30); + private static final Duration LOG_CONSUME_TIMEOUT = Duration.ofSeconds(60); private static final Logger LOG = LoggerFactory.getLogger(DividePluginCases.class); @@ -79,6 +84,42 @@ private ShenYuScenarioSpec testDivideHello() { .build(); } + static void awaitRequestLog(final DefaultLitePullConsumer consumer, final String endpoint) { + AtomicBoolean assigned = new AtomicBoolean(); + await().alias("RocketMQ access log for " + endpoint) + .atMost(LOG_CONSUME_TIMEOUT).pollInterval(Duration.ofMillis(200)) + .until(() -> { + if (!assigned.get()) { + Collection queues; + try { + queues = consumer.fetchMessageQueues(TOPIC); + } catch (MQClientException ex) { + Throwable cause = ex; + while (Objects.nonNull(cause)) { + if (cause instanceof MQClientException && ((MQClientException) cause).getResponseCode() == ResponseCode.TOPIC_NOT_EXIST) { + return false; + } + cause = cause.getCause(); + } + throw ex; + } + if (queues.isEmpty()) { + return false; + } + consumer.assign(queues); + for (MessageQueue queue : queues) { + consumer.seekToBegin(queue); + } + assigned.set(true); + } + return consumer.poll(200).stream().anyMatch(message -> { + String body = new String(message.getBody(), StandardCharsets.UTF_8); + LOG.info("RocketMQ access log: {}", body); + return body.contains(endpoint); + }); + }); + } + private ShenYuScenarioSpec testRocketMQHello() { return ShenYuScenarioSpec.builder() .name("testRocketMQHello") @@ -102,27 +143,17 @@ private ShenYuScenarioSpec testRocketMQHello() { .caseSpec( ShenYuCaseSpec.builder() .add(request -> { - AtomicBoolean isLog = new AtomicBoolean(false); - DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(CONSUMERGROUP); + String requestId = UUID.randomUUID().toString(); + String endpoint = "/http/order/findById?id=23&e2e=" + requestId; + DefaultLitePullConsumer consumer = new DefaultLitePullConsumer(CONSUMERGROUP + "-" + requestId); try { consumer.setNamesrvAddr(NAMESERVER); - consumer.subscribe(TOPIC, "*"); - consumer.registerMessageListener((MessageListenerConcurrently) (msgs, consumeConcurrentlyContext) -> { - LOG.info("Msg:{}", msgs); - if (CollectionUtils.isNotEmpty(msgs)) { - msgs.forEach(e -> { - if (new String(e.getBody()).contains("/http/order/findById?id=23")) { - isLog.set(true); - } - }); - } - return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; - }); + consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET); + consumer.setAutoCommit(false); consumer.start(); LOG.info("RocketMQ consumer started"); - request.request(Method.GET, "/http/order/findById?id=23"); - await().atMost(LOG_CONSUME_TIMEOUT).untilTrue(isLog); - LOG.info("isLog.get():{}", isLog.get()); + request.request(Method.GET, endpoint).then().statusCode(200); + awaitRequestLog(consumer, endpoint); } catch (Exception e) { LOG.error("error", e); Assertions.fail("Failed to consume RocketMQ access log", e); diff --git a/shenyu-e2e/shenyu-e2e-case/shenyu-e2e-case-logging-rocketmq/src/test/java/org/apache/shenyu/e2e/testcase/logging/rocketmq/RocketMQLogConsumptionTest.java b/shenyu-e2e/shenyu-e2e-case/shenyu-e2e-case-logging-rocketmq/src/test/java/org/apache/shenyu/e2e/testcase/logging/rocketmq/RocketMQLogConsumptionTest.java new file mode 100644 index 000000000000..d620ca3c1f09 --- /dev/null +++ b/shenyu-e2e/shenyu-e2e-case/shenyu-e2e-case-logging-rocketmq/src/test/java/org/apache/shenyu/e2e/testcase/logging/rocketmq/RocketMQLogConsumptionTest.java @@ -0,0 +1,94 @@ +/* + * 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.e2e.testcase.logging.rocketmq; + +import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer; +import org.apache.rocketmq.client.exception.MQClientException; +import org.apache.rocketmq.common.message.MessageExt; +import org.apache.rocketmq.common.message.MessageQueue; +import org.apache.rocketmq.common.protocol.ResponseCode; +import org.junit.jupiter.api.Test; + +import java.nio.charset.StandardCharsets; +import java.util.Collection; +import java.util.List; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class RocketMQLogConsumptionTest { + + @Test + void rejectsNonTransientBrokerErrors() { + DefaultLitePullConsumer consumer = new DefaultLitePullConsumer("denied-log-fixture") { + @Override + public Collection fetchMessageQueues(final String topic) throws MQClientException { + throw new MQClientException(ResponseCode.NO_PERMISSION, "access denied"); + } + }; + + MQClientException exception = assertThrows(MQClientException.class, () -> DividePluginCases.awaitRequestLog(consumer, "/current-request")); + assertEquals(ResponseCode.NO_PERMISSION, exception.getResponseCode()); + } + + @Test + void waitsForTopicCreationAndRejectsUnrelatedLogs() { + AtomicInteger discoveries = new AtomicInteger(); + AtomicInteger polls = new AtomicInteger(); + AtomicInteger seeks = new AtomicInteger(); + MessageQueue queue = new MessageQueue("shenyu-access-logging", "broker", 0); + String endpoint = "/http/order/findById?id=23&e2e=current-request"; + DefaultLitePullConsumer consumer = new DefaultLitePullConsumer("log-fixture") { + @Override + public Collection fetchMessageQueues(final String topic) throws MQClientException { + if (discoveries.incrementAndGet() == 1) { + throw new MQClientException("topic route pending", new MQClientException(ResponseCode.TOPIC_NOT_EXIST, "topic is not published yet")); + } + return List.of(queue); + } + + @Override + public void assign(final Collection queues) { + assertEquals(List.of(queue), queues); + } + + @Override + public void seekToBegin(final MessageQueue assignedQueue) { + assertEquals(queue, assignedQueue); + seeks.incrementAndGet(); + } + + @Override + public List poll(final long timeout) { + assertTrue(seeks.get() > 0); + MessageExt message = new MessageExt(); + String uri = polls.incrementAndGet() == 1 ? "/http/order/findById?id=23&e2e=old-request" : endpoint; + message.setBody(("{\"requestUri\":\"http://gateway" + uri + "\"}").getBytes(StandardCharsets.UTF_8)); + return List.of(message); + } + }; + + DividePluginCases.awaitRequestLog(consumer, endpoint); + + assertEquals(2, discoveries.get()); + assertEquals(1, seeks.get()); + assertEquals(2, polls.get()); + } +} diff --git a/shenyu-e2e/shenyu-e2e-case/shenyu-e2e-case-spring-cloud/pom.xml b/shenyu-e2e/shenyu-e2e-case/shenyu-e2e-case-spring-cloud/pom.xml index 4ff46f9ed993..ef1d944bd275 100644 --- a/shenyu-e2e/shenyu-e2e-case/shenyu-e2e-case-spring-cloud/pom.xml +++ b/shenyu-e2e/shenyu-e2e-case/shenyu-e2e-case-spring-cloud/pom.xml @@ -28,4 +28,13 @@ shenyu-e2e-case-spring-cloud + + + org.awaitility + awaitility + ${awaitility.version} + test + + + diff --git a/shenyu-e2e/shenyu-e2e-case/shenyu-e2e-case-spring-cloud/src/test/java/org/apache/shenyu/e2e/testcase/springcloud/SpringCloudPluginTest.java b/shenyu-e2e/shenyu-e2e-case/shenyu-e2e-case-spring-cloud/src/test/java/org/apache/shenyu/e2e/testcase/springcloud/SpringCloudPluginTest.java index d2847f639e25..5166c418eda5 100644 --- a/shenyu-e2e/shenyu-e2e-case/shenyu-e2e-case-spring-cloud/src/test/java/org/apache/shenyu/e2e/testcase/springcloud/SpringCloudPluginTest.java +++ b/shenyu-e2e/shenyu-e2e-case/shenyu-e2e-case-spring-cloud/src/test/java/org/apache/shenyu/e2e/testcase/springcloud/SpringCloudPluginTest.java @@ -20,7 +20,6 @@ import org.apache.shenyu.e2e.client.WaitDataSync; import org.apache.shenyu.e2e.client.admin.AdminClient; import org.apache.shenyu.e2e.client.gateway.GatewayClient; -import org.apache.shenyu.e2e.constant.Constants; import org.apache.shenyu.e2e.engine.annotation.ShenYuScenario; import org.apache.shenyu.e2e.engine.annotation.ShenYuTest; import org.apache.shenyu.e2e.engine.scenario.specification.CaseSpec; @@ -31,9 +30,11 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.util.HashMap; +import java.time.Duration; import java.util.List; -import java.util.Map; + +import static org.awaitility.Awaitility.await; +import static org.hamcrest.Matchers.equalTo; /** * Testing spring-cloud plugin. @@ -74,17 +75,20 @@ public void setup(final AdminClient adminClient, final GatewayClient gatewayClie List selectorCache = gatewayClient.getSelectorCache(); LOG.info("selectorCache: {}", selectorCache); - Map reqBody = new HashMap<>(); - reqBody.put("pluginId", "5"); - reqBody.put("name", "divide"); - reqBody.put("enabled", "true"); - reqBody.put("role", "Proxy"); - reqBody.put("sort", "200"); - reqBody.put("config", "{\"enabled\":true}"); - reqBody.put("namespaceId", Constants.SYS_DEFAULT_NAMESPACE_NAMESPACE_ID); - adminClient.changePluginStatus("1801816010882822185", reqBody); WaitDataSync.waitGatewayPluginUse(gatewayClient, "org.apache.shenyu.plugin.divide.DividePlugin"); + waitForRegisteredRoute(gatewayClient); + } + static void waitForRegisteredRoute(final GatewayClient gatewayClient) { + // Selector/rule synchronization can complete before Eureka publishes the upstream. + await().alias("Spring Cloud discovery upstream routes through the gateway") + .atMost(Duration.ofSeconds(90)) + .pollInterval(Duration.ofMillis(200)) + .untilAsserted(() -> gatewayClient.getHttpRequesterSupplier().get() + .get("/springcloud/order/path/123/hahah") + .then().statusCode(200) + .body("id", equalTo("123")) + .body("name", equalTo("hello world spring cloud restful: hahah"))); } @ShenYuScenario(provider = SpringCloudPluginCases.class) diff --git a/shenyu-e2e/shenyu-e2e-case/shenyu-e2e-case-spring-cloud/src/test/java/org/apache/shenyu/e2e/testcase/springcloud/SpringCloudRouteReadinessTest.java b/shenyu-e2e/shenyu-e2e-case/shenyu-e2e-case-spring-cloud/src/test/java/org/apache/shenyu/e2e/testcase/springcloud/SpringCloudRouteReadinessTest.java new file mode 100644 index 000000000000..ee1adb461280 --- /dev/null +++ b/shenyu-e2e/shenyu-e2e-case/shenyu-e2e-case-spring-cloud/src/test/java/org/apache/shenyu/e2e/testcase/springcloud/SpringCloudRouteReadinessTest.java @@ -0,0 +1,57 @@ +/* + * 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.e2e.testcase.springcloud; + +import com.sun.net.httpserver.HttpServer; +import org.apache.shenyu.e2e.client.gateway.GatewayClient; +import org.junit.jupiter.api.Test; + +import java.net.InetSocketAddress; +import java.nio.charset.StandardCharsets; +import java.util.Properties; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.junit.jupiter.api.Assertions.assertTrue; + +class SpringCloudRouteReadinessTest { + + @Test + void waitsForDiscoveredUpstreamAndExpectedResponse() throws Exception { + AtomicInteger requests = new AtomicInteger(); + HttpServer server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); + server.createContext("/springcloud/order/path/123/hahah", exchange -> { + int attempt = requests.incrementAndGet(); + byte[] body = (attempt < 3 ? "{\"code\":-107,\"message\":\"upstream not ready\"}" + : "{\"id\":\"123\",\"name\":\"hello world spring cloud restful: hahah\"}").getBytes(StandardCharsets.UTF_8); + exchange.getResponseHeaders().set("Content-Type", "application/json"); + exchange.sendResponseHeaders(attempt == 1 ? 503 : 200, body.length); + try (var output = exchange.getResponseBody()) { + output.write(body); + } + }); + server.start(); + try { + GatewayClient gateway = new GatewayClient("readiness-test", "gateway", + "http://127.0.0.1:" + server.getAddress().getPort(), new Properties()); + SpringCloudPluginTest.waitForRegisteredRoute(gateway); + assertTrue(requests.get() >= 3); + } finally { + server.stop(0); + } + } +} diff --git a/shenyu-e2e/shenyu-e2e-engine/src/test/java/org/apache/shenyu/e2e/engine/function/WaitForHelperTest.java b/shenyu-e2e/shenyu-e2e-engine/src/test/java/org/apache/shenyu/e2e/engine/function/WaitForHelperTest.java index 10220adb54ac..9f07f4a5bd8a 100644 --- a/shenyu-e2e/shenyu-e2e-engine/src/test/java/org/apache/shenyu/e2e/engine/function/WaitForHelperTest.java +++ b/shenyu-e2e/shenyu-e2e-engine/src/test/java/org/apache/shenyu/e2e/engine/function/WaitForHelperTest.java @@ -17,13 +17,23 @@ package org.apache.shenyu.e2e.engine.function; -import org.apache.shenyu.e2e.engine.scenario.function.WaitForHelper; +import com.sun.net.httpserver.HttpExchange; +import com.sun.net.httpserver.HttpServer; import io.restassured.builder.ResponseSpecBuilder; import io.restassured.http.Method; +import org.apache.shenyu.e2e.engine.scenario.function.WaitForHelper; +import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import java.io.IOException; +import java.io.OutputStream; +import java.net.InetSocketAddress; +import java.nio.charset.StandardCharsets; import java.time.Duration; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import static io.restassured.RestAssured.given; @@ -31,26 +41,73 @@ import static org.hamcrest.CoreMatchers.equalTo; public class WaitForHelperTest { - + + private final CountDownLatch delayedRequest = new CountDownLatch(1); + + private final CountDownLatch releaseResponse = new CountDownLatch(1); + + private HttpServer server; + + @BeforeEach + void setUp() throws IOException { + server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0); + server.createContext("/get", exchange -> writeJson(exchange, "{\"url\":\"" + exchange.getRequestURI() + "\"}")); + server.createContext("/delay/2", exchange -> { + delayedRequest.countDown(); + try { + releaseResponse.await(); + writeJson(exchange, "{\"delay\":2}"); + } catch (InterruptedException ex) { + Thread.currentThread().interrupt(); + exchange.close(); + } + }); + server.start(); + // Initialize the HTTP client before measuring the helper's short timeout. + given().baseUri(baseUri()).get("/get").then().statusCode(200); + } + + @AfterEach + void tearDown() { + releaseResponse.countDown(); + server.stop(0); + } + @Test void testSuccess() throws TimeoutException { new WaitForHelper().waitFor( - () -> given().baseUri("https://postman-echo.com").when(), + () -> given().baseUri(baseUri()).when(), Method.GET, "/get?foo1=bar1&foo2=bar2", new ResponseSpecBuilder().expectBody("url", containsString("/get")).build() ); } - + @Test - void testTimeout() { + void testTimeout() throws InterruptedException { Assertions.assertThrows(TimeoutException.class, () -> new WaitForHelper(10, Duration.ofSeconds(1), Duration.ofMillis(500)).waitFor( - () -> given().baseUri("https://postman-echo.com").when(), + () -> given().baseUri(baseUri()).when(), Method.GET, "/delay/2", new ResponseSpecBuilder().expectBody("delay", equalTo(2)).build() ) ); + Assertions.assertTrue(delayedRequest.await(2, TimeUnit.SECONDS)); + } + + private String baseUri() { + return "http://127.0.0.1:" + server.getAddress().getPort(); + } + + private static void writeJson(final HttpExchange exchange, final String body) throws IOException { + byte[] bytes = body.getBytes(StandardCharsets.UTF_8); + exchange.getResponseHeaders().set("Content-Type", "application/json"); + exchange.sendResponseHeaders(200, bytes.length); + try (OutputStream output = exchange.getResponseBody()) { + output.write(bytes); + } finally { + exchange.close(); + } } } diff --git a/shenyu-sync-data-center/shenyu-sync-data-websocket/src/main/java/org/apache/shenyu/plugin/sync/data/websocket/WebsocketSyncDataService.java b/shenyu-sync-data-center/shenyu-sync-data-websocket/src/main/java/org/apache/shenyu/plugin/sync/data/websocket/WebsocketSyncDataService.java index 4bab04bf7500..c2f32418a2f9 100644 --- a/shenyu-sync-data-center/shenyu-sync-data-websocket/src/main/java/org/apache/shenyu/plugin/sync/data/websocket/WebsocketSyncDataService.java +++ b/shenyu-sync-data-center/shenyu-sync-data-websocket/src/main/java/org/apache/shenyu/plugin/sync/data/websocket/WebsocketSyncDataService.java @@ -43,9 +43,11 @@ import java.net.URI; import java.util.HashMap; import java.util.Iterator; +import java.util.LinkedHashSet; import java.util.List; import java.util.Map; import java.util.Objects; +import java.util.Set; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; @@ -150,6 +152,7 @@ private synchronized void masterCheck() { clients.add(createClient(url)); } } + Set masterUrls = new LinkedHashSet<>(); Iterator iterator = clients.iterator(); while (iterator.hasNext()) { ShenyuWebsocketClient websocketClient = iterator.next(); @@ -166,11 +169,42 @@ private synchronized void masterCheck() { return; } - if (!websocketClient.isConnectedToMaster()) { + // An open socket may still be waiting for its running-mode reply. + if (Objects.equals(runningMode, RunningModeEnum.CLUSTER.name()) && !websocketClient.isConnectedToMaster()) { + URI masterUri = parseMasterUri(websocketClient.getMasterUrl()); + if (Objects.nonNull(masterUri) && !masterUri.equals(websocketClient.getURI())) { + masterUrls.add(masterUri); + } websocketClient.nowClose(); iterator.remove(); } } + for (URI masterUri : masterUrls) { + if (clients.stream().noneMatch(client -> masterUri.equals(client.getURI()))) { + try { + clients.add(createClient(masterUri.toString())); + } catch (RuntimeException ex) { + LOG.warn("websocket connection to advertised master [{}] failed", masterUri, ex); + } + } + } + } + + private URI parseMasterUri(final String masterUrl) { + if (StringUtils.isBlank(masterUrl)) { + return null; + } + try { + URI uri = URI.create(masterUrl); + if (("ws".equalsIgnoreCase(uri.getScheme()) || "wss".equalsIgnoreCase(uri.getScheme())) && StringUtils.isNotBlank(uri.getHost())) { + return uri; + } + } catch (IllegalArgumentException ex) { + LOG.warn("Ignore invalid websocket master address [{}]", masterUrl, ex); + return null; + } + LOG.warn("Ignore invalid websocket master address [{}]", masterUrl); + return null; } @Override diff --git a/shenyu-sync-data-center/shenyu-sync-data-websocket/src/main/java/org/apache/shenyu/plugin/sync/data/websocket/client/ShenyuWebsocketClient.java b/shenyu-sync-data-center/shenyu-sync-data-websocket/src/main/java/org/apache/shenyu/plugin/sync/data/websocket/client/ShenyuWebsocketClient.java index 64522afd2286..678586d773d0 100644 --- a/shenyu-sync-data-center/shenyu-sync-data-websocket/src/main/java/org/apache/shenyu/plugin/sync/data/websocket/client/ShenyuWebsocketClient.java +++ b/shenyu-sync-data-center/shenyu-sync-data-websocket/src/main/java/org/apache/shenyu/plugin/sync/data/websocket/client/ShenyuWebsocketClient.java @@ -103,9 +103,9 @@ public final class ShenyuWebsocketClient extends WebSocketClient { private TimerTask timerTask; - private String runningMode; + private volatile String runningMode; - private String masterUrl; + private volatile String masterUrl; private volatile boolean isConnectedToMaster; @@ -263,6 +263,9 @@ public boolean connectBlocking() { @Override public void onOpen(final ServerHandshake serverHandshake) { + runningMode = null; + masterUrl = null; + isConnectedToMaster = false; LOG.info("websocket connection server[{}] is opened, sending sync msg", this.getURI().toString()); send(DataEventTypeEnum.RUNNING_MODE.name()); if (!alreadySync) { @@ -292,11 +295,12 @@ public void onMessage(final String result) { if (Objects.equals(WebsocketSyncFrame.EVENT_TYPE, eventType) && Objects.nonNull(initialSyncState)) { initialSyncState.accept(GsonUtils.getInstance().fromJson(result, WebsocketSyncFrame.class), this::handleResult); } else if (Objects.equals(DataEventTypeEnum.RUNNING_MODE.name(), eventType)) { - this.runningMode = String.valueOf(jsonToMap.get(RunningModeConstants.RUNNING_MODE)); - if (!Objects.equals(RunningModeEnum.STANDALONE.name(), runningMode)) { + String mode = String.valueOf(jsonToMap.get(RunningModeConstants.RUNNING_MODE)); + if (!Objects.equals(RunningModeEnum.STANDALONE.name(), mode)) { this.masterUrl = String.valueOf(jsonToMap.get(RunningModeConstants.MASTER_URL)); this.isConnectedToMaster = Boolean.TRUE.equals(jsonToMap.get(RunningModeConstants.IS_MASTER)); } + this.runningMode = mode; } else if (Objects.nonNull(initialSyncState)) { initialSyncState.applyIncremental(() -> handleResult(result)); } else { diff --git a/shenyu-sync-data-center/shenyu-sync-data-websocket/src/test/java/org/apache/shenyu/plugin/sync/data/websocket/WebsocketMasterRedirectTest.java b/shenyu-sync-data-center/shenyu-sync-data-websocket/src/test/java/org/apache/shenyu/plugin/sync/data/websocket/WebsocketMasterRedirectTest.java new file mode 100644 index 000000000000..922085d44e7a --- /dev/null +++ b/shenyu-sync-data-center/shenyu-sync-data-websocket/src/test/java/org/apache/shenyu/plugin/sync/data/websocket/WebsocketMasterRedirectTest.java @@ -0,0 +1,310 @@ +/* + * 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.plugin.sync.data.websocket; + +import org.apache.shenyu.common.config.ShenyuConfig; +import org.apache.shenyu.common.constant.Constants; +import org.apache.shenyu.common.timer.Timer; +import org.apache.shenyu.common.timer.WheelTimerFactory; +import org.apache.shenyu.common.utils.GsonUtils; +import org.apache.shenyu.plugin.sync.data.websocket.client.ShenyuWebsocketClient; +import org.apache.shenyu.plugin.sync.data.websocket.config.WebsocketConfig; +import org.apache.shenyu.sync.data.api.PluginDataSubscriber; +import org.java_websocket.WebSocket; +import org.java_websocket.handshake.ClientHandshake; +import org.java_websocket.server.WebSocketServer; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.NullAndEmptySource; +import org.junit.jupiter.params.provider.ValueSource; +import org.mockito.MockedConstruction; +import org.mockito.MockedStatic; +import org.springframework.boot.autoconfigure.web.ServerProperties; + +import java.lang.reflect.Field; +import java.net.InetSocketAddress; +import java.net.URI; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.awaitility.Awaitility.await; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.mockConstruction; +import static org.mockito.Mockito.mockStatic; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * Exercises master discovery through real sockets and manually triggered checks. + */ +class WebsocketMasterRedirectTest { + + private final List servers = new ArrayList<>(); + + private MockedStatic timers; + + private WebsocketSyncDataService service; + + private final CountDownLatch applied = new CountDownLatch(1); + + @BeforeEach + void setUp() { + timers = mockStatic(WheelTimerFactory.class); + timers.when(WheelTimerFactory::newWheelTimer).thenReturn(mock(Timer.class)); + timers.when(WheelTimerFactory::getSharedTimer).thenReturn(mock(Timer.class)); + } + + @AfterEach + void tearDown() throws InterruptedException { + try { + if (Objects.nonNull(service)) { + service.close(); + } + for (TestServer server : servers) { + server.stop(1000); + } + } finally { + timers.close(); + } + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + void testRedirectsFollowerConnectionsToOneMaster(final boolean multipleFollowers) throws Exception { + TestServer master = startServer(true); + TestServer follower = startServer(false); + follower.masterUrl = master.url(); + final List seeds = new ArrayList<>(Collections.singletonList(follower.url())); + if (multipleFollowers) { + TestServer other = startServer(false); + other.masterUrl = master.url(); + seeds.add(other.url()); + } + createService(seeds); + awaitModes(); + + check(); + + assertTrue(applied.await(5, TimeUnit.SECONDS), "Configuration should arrive from the advertised master"); + await().atMost(Duration.ofSeconds(5)).until(() -> clients().size() == 1 && clients().get(0).isConnectedToMaster()); + assertEquals(URI.create(master.url()), clients().get(0).getURI()); + assertEquals(1, master.opens.get()); + assertTrue(follower.closed.await(5, TimeUnit.SECONDS)); + assertEquals("test-token", master.handshake.getFieldValue(Constants.X_SHENYU_SYNC_TOKEN)); + assertEquals("https://gateway.example", master.handshake.getFieldValue("Origin")); + assertEquals("cluster-test", master.handshake.getFieldValue(Constants.SHENYU_NAMESPACE_ID)); + assertEquals("9195", master.handshake.getFieldValue(Constants.CLIENT_PORT_NAME)); + assertEquals(seeds, service.getWebsocketConfig().getUrls()); + check(); + assertEquals(1, master.opens.get()); + assertEquals(1, follower.opens.get()); + } + + @Test + void testKeepsConnectionWhileRunningModeReplyIsPending() throws Exception { + TestServer server = startServer(true); + server.replyMode = false; + createService(Collections.singletonList(server.url())); + ShenyuWebsocketClient client = clients().get(0); + + check(); + + assertEquals(1, clients().size()); + assertSame(client, clients().get(0)); + assertTrue(client.isOpen()); + server.sendMode(); + awaitModes(); + check(); + assertSame(client, clients().get(0)); + } + + @ParameterizedTest + @NullAndEmptySource + @ValueSource(strings = {"null", "/websocket", "http://admin/websocket", "ws://[invalid"}) + void testInvalidMasterAddressFallsBackToSeeds(final String address) throws Exception { + createService(Collections.emptyList()); + ShenyuWebsocketClient follower = mock(ShenyuWebsocketClient.class); + when(follower.isOpen()).thenReturn(true); + when(follower.getRunningMode()).thenReturn("CLUSTER"); + when(follower.getMasterUrl()).thenReturn(address); + clients().add(follower); + try (MockedConstruction constructed = mockConstruction(ShenyuWebsocketClient.class)) { + check(); + assertTrue(clients().isEmpty()); + assertTrue(constructed.constructed().isEmpty()); + verify(follower).nowClose(); + } + } + + @Test + void testReusesExistingMasterConnection() throws Exception { + createService(Collections.emptyList()); + ShenyuWebsocketClient follower = mock(ShenyuWebsocketClient.class); + when(follower.isOpen()).thenReturn(true); + when(follower.getRunningMode()).thenReturn("CLUSTER"); + when(follower.getMasterUrl()).thenReturn("ws://master:9095/websocket"); + ShenyuWebsocketClient master = mock(ShenyuWebsocketClient.class); + when(master.isOpen()).thenReturn(true); + when(master.getRunningMode()).thenReturn("CLUSTER"); + when(master.isConnectedToMaster()).thenReturn(true); + when(master.getURI()).thenReturn(URI.create("ws://master:9095/websocket")); + clients().add(follower); + clients().add(master); + try (MockedConstruction constructed = mockConstruction(ShenyuWebsocketClient.class)) { + check(); + assertEquals(Collections.singletonList(master), clients()); + assertTrue(constructed.constructed().isEmpty()); + } + } + + @Test + void testClosedRedirectConnectionReturnsToConfiguredSeed() throws Exception { + createService(Collections.emptyList()); + service.getWebsocketConfig().setUrls(Collections.singletonList("ws://seed:9095/websocket")); + ShenyuWebsocketClient failedMaster = mock(ShenyuWebsocketClient.class); + clients().add(failedMaster); + check(); + verify(failedMaster).nowClose(); + try (MockedConstruction constructed = mockConstruction(ShenyuWebsocketClient.class, (client, context) -> + assertEquals(URI.create("ws://seed:9095/websocket"), context.arguments().get(0)))) { + check(); + assertEquals(1, constructed.constructed().size()); + } + } + + private void createService(final List seeds) { + WebsocketConfig config = new WebsocketConfig(); + config.setUrls(seeds); + config.setToken("test-token"); + config.setAllowOrigin("https://gateway.example"); + ShenyuConfig shenyu = new ShenyuConfig(); + shenyu.setNamespace("cluster-test"); + ServerProperties properties = new ServerProperties(); + properties.setPort(9195); + PluginDataSubscriber subscriber = mock(PluginDataSubscriber.class); + doAnswer(invocation -> { + applied.countDown(); + return null; + }).when(subscriber).onSubscribe(any()); + service = new WebsocketSyncDataService(config, shenyu, subscriber, Collections.emptyList(), Collections.emptyList(), + Collections.emptyList(), Collections.emptyList(), Collections.emptyList(), properties); + } + + private TestServer startServer(final boolean master) throws InterruptedException { + TestServer server = new TestServer(master); + servers.add(server); + server.start(); + assertTrue(server.started.await(5, TimeUnit.SECONDS)); + return server; + } + + private void awaitModes() { + await().atMost(Duration.ofSeconds(5)).until(() -> clients().stream().allMatch(client -> Objects.nonNull(client.getRunningMode()))); + } + + @SuppressWarnings("unchecked") + private List clients() throws ReflectiveOperationException { + Field field = WebsocketSyncDataService.class.getDeclaredField("clients"); + field.setAccessible(true); + return (List) field.get(service); + } + + private void check() throws ReflectiveOperationException { + java.lang.reflect.Method method = WebsocketSyncDataService.class.getDeclaredMethod("masterCheck"); + method.setAccessible(true); + method.invoke(service); + } + + private static final class TestServer extends WebSocketServer { + + private final boolean master; + + private final CountDownLatch started = new CountDownLatch(1); + + private final CountDownLatch closed = new CountDownLatch(1); + + private final AtomicInteger opens = new AtomicInteger(); + + private volatile String masterUrl; + + private volatile boolean replyMode = true; + + private volatile WebSocket connection; + + private volatile ClientHandshake handshake; + + private TestServer(final boolean master) { + super(new InetSocketAddress("127.0.0.1", 0)); + this.master = master; + } + + private String url() { + return "ws://127.0.0.1:" + getPort() + "/websocket"; + } + + private void sendMode() { + connection.send(GsonUtils.getInstance().toJson(Map.of("eventType", "RUNNING_MODE", "runningMode", "CLUSTER", + "isMaster", master, "masterUrl", master ? url() : masterUrl))); + } + + @Override + public void onOpen(final WebSocket socket, final ClientHandshake clientHandshake) { + connection = socket; + handshake = clientHandshake; + opens.incrementAndGet(); + } + + @Override + public void onClose(final WebSocket socket, final int code, final String reason, final boolean remote) { + closed.countDown(); + } + + @Override + public void onMessage(final WebSocket socket, final String message) { + if ("RUNNING_MODE".equals(message) && replyMode) { + sendMode(); + } else if ("MYSELF".equals(message) && master) { + socket.send("{\"groupType\":\"PLUGIN\",\"eventType\":\"CREATE\",\"data\":[{\"id\":\"divide\",\"name\":\"divide\",\"enabled\":true}]}"); + } + } + + @Override + public void onError(final WebSocket socket, final Exception exception) { + throw new IllegalStateException(exception); + } + + @Override + public void onStart() { + started.countDown(); + } + } +} diff --git a/shenyu-sync-data-center/shenyu-sync-data-websocket/src/test/java/org/apache/shenyu/plugin/sync/data/websocket/client/ShenyuWebsocketClientTest.java b/shenyu-sync-data-center/shenyu-sync-data-websocket/src/test/java/org/apache/shenyu/plugin/sync/data/websocket/client/ShenyuWebsocketClientTest.java index deb8eeee9cf2..7c51d9f92b18 100644 --- a/shenyu-sync-data-center/shenyu-sync-data-websocket/src/test/java/org/apache/shenyu/plugin/sync/data/websocket/client/ShenyuWebsocketClientTest.java +++ b/shenyu-sync-data-center/shenyu-sync-data-websocket/src/test/java/org/apache/shenyu/plugin/sync/data/websocket/client/ShenyuWebsocketClientTest.java @@ -118,6 +118,20 @@ public void testOnOpen() { verify(shenyuWebsocketClient).send(DataEventTypeEnum.MYSELF.name()); } + @Test + void testOnOpenClearsPreviousRunningMode() { + ShenyuWebsocketClient client = spy(shenyuWebsocketClient); + doNothing().when(client).send(anyString()); + client.onMessage("{\"eventType\":\"RUNNING_MODE\",\"runningMode\":\"CLUSTER\",\"isMaster\":true,\"masterUrl\":\"ws://old-master/websocket\"}"); + assertTrue(client.isConnectedToMaster()); + + client.onOpen(mock(ServerHandshake.class)); + + Assertions.assertNull(client.getRunningMode()); + Assertions.assertNull(client.getMasterUrl()); + assertFalse(client.isConnectedToMaster()); + } + @Test public void testOnMessage() { doNothing().when(pluginDataSubscriber).onPluginRefresh(any());