From 9fdab824a756fb6a7ae904a8fec4119cea47574f Mon Sep 17 00:00:00 2001 From: Xiaoyumuxi <3075514079@qq.com> Date: Mon, 14 Sep 2026 15:09:34 +0800 Subject: [PATCH] ci: redesign validation and extended integration workflows --- .github/workflows/benchmark.yml | 52 +++++ .github/workflows/ci.yml | 88 ++++++-- .github/workflows/integration.yml | 205 ++++++++++++++++++ go_client/main.go | 33 ++- python_client/client.py | 58 +++-- .../rpc/consumer/FullIntegrationTest.java | 97 +++++---- 6 files changed, 425 insertions(+), 108 deletions(-) create mode 100644 .github/workflows/benchmark.yml create mode 100644 .github/workflows/integration.yml diff --git a/.github/workflows/benchmark.yml b/.github/workflows/benchmark.yml new file mode 100644 index 0000000..372503b --- /dev/null +++ b/.github/workflows/benchmark.yml @@ -0,0 +1,52 @@ +name: Benchmark + +on: + workflow_dispatch: + schedule: + - cron: '0 18 * * 6' + +permissions: + contents: read + +concurrency: + group: benchmark-${{ github.ref }} + cancel-in-progress: true + +jobs: + serialization-benchmark: + name: Serialization Benchmark + runs-on: ubuntu-24.04 + timeout-minutes: 20 + + steps: + - name: Checkout code + uses: actions/checkout@v7 + + - name: Set up JDK 17 + uses: actions/setup-java@v6 + with: + java-version: '17' + distribution: 'temurin' + cache: 'maven' + + - name: Build benchmark module + run: mvn -B -ntp -pl rpc-benchmark -am clean package -DskipTests -Drpc.registry=local + + - name: Run JMH serialization benchmark + run: >- + java -jar rpc-benchmark/target/benchmarks.jar + '.*SerializationBenchmark.*' + -f 1 + -wi 1 + -i 2 + -rf json + -rff benchmark-results.json + + - name: Upload benchmark result + if: always() + uses: actions/upload-artifact@v7 + with: + name: serialization-benchmark + retention-days: 30 + if-no-files-found: error + path: benchmark-results.json diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 1bf5779..7a39192 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -1,14 +1,23 @@ -name: CI/CD Pipeline +name: CI on: push: - branches: [ main, master, develop ] + branches: [ main ] pull_request: - branches: [ main, master, develop ] + branches: [ main ] + +permissions: + contents: read + +concurrency: + group: ci-${{ github.workflow }}-${{ github.ref }} + cancel-in-progress: true jobs: - build-and-test: + build: + name: Build runs-on: ubuntu-24.04 + timeout-minutes: 10 steps: - name: Checkout code @@ -21,29 +30,43 @@ jobs: distribution: 'temurin' cache: 'maven' - - name: Build with Maven - run: mvn clean package -DskipTests + - name: Build all modules + run: mvn -B -ntp clean package -DskipTests -Drpc.registry=local - - name: Run unit tests - run: mvn test -pl rpc-core,rpc-transport-netty -am + unit-tests: + name: Unit Tests & Coverage + runs-on: ubuntu-24.04 + timeout-minutes: 10 + + steps: + - name: Checkout code + uses: actions/checkout@v7 - - name: Run integration tests - run: mvn test -pl rpc-consumer -am -Dtest=FullIntegrationTest + - name: Set up JDK 17 + uses: actions/setup-java@v6 + with: + java-version: '17' + distribution: 'temurin' + cache: 'maven' - - name: Check test coverage - run: mvn jacoco:report + - name: Run core and transport unit tests + run: mvn -B -ntp test -pl rpc-core,rpc-transport-netty -am -Drpc.registry=local - - name: Upload test results + - name: Upload unit test and coverage reports if: always() uses: actions/upload-artifact@v7 with: - name: test-results + name: unit-test-reports + retention-days: 7 + if-no-files-found: warn path: | **/target/surefire-reports/ **/target/site/jacoco/ - code-quality: + rpc-integration: + name: RPC Integration runs-on: ubuntu-24.04 + timeout-minutes: 10 steps: - name: Checkout code @@ -56,8 +79,35 @@ jobs: distribution: 'temurin' cache: 'maven' - - name: Run Maven verify - run: mvn verify -DskipTests + - name: Run local-registry Netty end-to-end test + run: >- + mvn -B -ntp test + -pl rpc-consumer -am + -Dtest=FullIntegrationTest + -Dsurefire.failIfNoSpecifiedTests=false + -Drpc.registry=local + -Drpc.protocol=netty + + - name: Upload integration test reports + if: always() + uses: actions/upload-artifact@v7 + with: + name: rpc-integration-reports + retention-days: 7 + if-no-files-found: warn + path: '**/target/surefire-reports/' + + ci-gate: + name: CI Gate + if: always() + needs: [ build, unit-tests, rpc-integration ] + runs-on: ubuntu-24.04 + timeout-minutes: 2 - - name: Check code style - run: mvn checkstyle:check || true + steps: + - name: Verify required checks + shell: bash + run: | + test "${{ needs.build.result }}" = "success" + test "${{ needs.unit-tests.result }}" = "success" + test "${{ needs.rpc-integration.result }}" = "success" diff --git a/.github/workflows/integration.yml b/.github/workflows/integration.yml new file mode 100644 index 0000000..a0ec440 --- /dev/null +++ b/.github/workflows/integration.yml @@ -0,0 +1,205 @@ +name: Extended Integration + +on: + workflow_dispatch: + schedule: + - cron: '0 18 * * 0' + +permissions: + contents: read + +concurrency: + group: extended-integration-${{ github.ref }} + cancel-in-progress: false + +jobs: + protocol-matrix: + name: Protocol - ${{ matrix.protocol }} + runs-on: ubuntu-24.04 + timeout-minutes: 12 + strategy: + fail-fast: false + matrix: + protocol: [ netty, http, http2, grpc ] + + steps: + - name: Checkout code + uses: actions/checkout@v7 + + - name: Set up JDK 17 + uses: actions/setup-java@v6 + with: + java-version: '17' + distribution: 'temurin' + cache: 'maven' + + - name: Run end-to-end protocol test + run: >- + mvn -B -ntp test + -pl rpc-consumer -am + -Dtest=FullIntegrationTest + -Dsurefire.failIfNoSpecifiedTests=false + -Drpc.registry=local + -Drpc.protocol=${{ matrix.protocol }} + + - name: Upload protocol test reports + if: always() + uses: actions/upload-artifact@v7 + with: + name: protocol-${{ matrix.protocol }}-reports + retention-days: 7 + if-no-files-found: warn + path: '**/target/surefire-reports/' + + nacos-integration: + name: Nacos Integration + runs-on: ubuntu-24.04 + timeout-minutes: 15 + + services: + nacos: + image: nacos/nacos-server:v2.5.4 + env: + MODE: standalone + NACOS_AUTH_ENABLE: 'false' + JVM_XMS: 256m + JVM_XMX: 256m + JVM_XMN: 128m + ports: + - 8848:8848 + - 9848:9848 + - 9849:9849 + + steps: + - name: Checkout code + uses: actions/checkout@v7 + + - name: Set up JDK 17 + uses: actions/setup-java@v6 + with: + java-version: '17' + distribution: 'temurin' + cache: 'maven' + + - name: Wait for Nacos + shell: bash + run: | + for i in {1..60}; do + if curl -fsS http://127.0.0.1:8848/nacos/v1/console/health/readiness >/dev/null 2>&1; then + exit 0 + fi + sleep 2 + done + echo 'Nacos did not become ready in time' + docker ps -a + container_id=$(docker ps -aq --filter ancestor=nacos/nacos-server:v2.5.4 | head -1) + if [ -n "$container_id" ]; then + docker logs "$container_id" || true + fi + exit 1 + + - name: Run real Nacos registry integration test + run: >- + mvn -B -ntp test + -pl rpc-consumer -am + -Dtest=FullIntegrationTest + -Dsurefire.failIfNoSpecifiedTests=false + -Drpc.registry=nacos + -Drpc.registry-address=127.0.0.1:8848 + -Drpc.protocol=netty + + - name: Upload Nacos integration reports + if: always() + uses: actions/upload-artifact@v7 + with: + name: nacos-integration-reports + retention-days: 7 + if-no-files-found: warn + path: '**/target/surefire-reports/' + + cross-language-grpc: + name: gRPC - Python & Go + runs-on: ubuntu-24.04 + timeout-minutes: 20 + + steps: + - name: Checkout code + uses: actions/checkout@v7 + + - name: Set up JDK 17 + uses: actions/setup-java@v6 + with: + java-version: '17' + distribution: 'temurin' + cache: 'maven' + + - name: Set up Python + uses: actions/setup-python@v7 + with: + python-version: '3.12' + + - name: Set up Go + uses: actions/setup-go@v7 + with: + go-version-file: go_client/go.mod + cache-dependency-path: go_client/go.sum + + - name: Install Java modules + run: mvn -B -ntp install -DskipTests -Drpc.registry=local + + - name: Install Python gRPC runtime + run: python -m pip install grpcio==1.76.0 protobuf==6.31.1 + + - name: Start Java gRPC provider + shell: bash + run: | + nohup mvn -B -ntp -pl rpc-provider \ + org.codehaus.mojo:exec-maven-plugin:3.5.0:java \ + -Dexec.mainClass=com.xiaoyu.rpc.provider.ProviderApp \ + -Drpc.registry=local \ + -Drpc.protocol=grpc \ + -Drpc.serializer=protobuf \ + > provider.log 2>&1 & + echo $! > provider.pid + + - name: Wait for Java provider + run: | + python - <<'PY' + import socket + import sys + import time + + for _ in range(60): + try: + with socket.create_connection(('127.0.0.1', 8080), timeout=1): + sys.exit(0) + except OSError: + time.sleep(0.5) + + print('Java gRPC provider did not become ready in time', file=sys.stderr) + sys.exit(1) + PY + + - name: Run official Python gRPC client + run: python python_client/client.py + + - name: Run Go gRPC client + working-directory: go_client + run: go run . + + - name: Stop Java provider + if: always() + shell: bash + run: | + if [ -f provider.pid ]; then + kill "$(cat provider.pid)" 2>/dev/null || true + fi + + - name: Upload provider log + if: always() + uses: actions/upload-artifact@v7 + with: + name: cross-language-provider-log + retention-days: 7 + if-no-files-found: warn + path: provider.log diff --git a/go_client/main.go b/go_client/main.go index 2f4b1b3..2ee7d8e 100644 --- a/go_client/main.go +++ b/go_client/main.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "log" + "strings" "time" "go_client/pb" @@ -15,7 +16,6 @@ import ( ) func main() { - // Connect to the Java gRPC server addr := "localhost:8080" conn, err := grpc.Dial(addr, grpc.WithTransportCredentials(insecure.NewCredentials())) if err != nil { @@ -25,14 +25,12 @@ func main() { client := pb.NewGrpcServiceClient(conn) - // Wrap the parameter in a Protobuf StringValue - param := wrapperspb.String( "World") + param := wrapperspb.String("World") paramBytes, err := proto.Marshal(param) if err != nil { log.Fatalf("failed to marshal param: %v", err) } - // Prepare the RPC request req := &pb.RpcRequest{ InterfaceName: "com.xiaoyu.rpc.api.HelloService", MethodName: "sayHello", @@ -41,29 +39,28 @@ func main() { RequestId: fmt.Sprintf("go-req-%d", time.Now().UnixNano()), } - // Set a timeout for the call - ctx, cancel := context.WithTimeout(context.Background(), time.Second*5) + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() - // Call the handle method fmt.Printf("Sending RpcRequest: interface=%s, method=%s, param=%s\n", req.InterfaceName, req.MethodName, "World") resp, err := client.Handle(ctx, req) if err != nil { log.Fatalf("could not call handle: %v", err) } + if resp.Message != "Success" { + log.Fatalf("RPC failed: %s", resp.Message) + } - fmt.Println("RpcResponse received:") - if resp.Message == "Success" { - resultVal := &wrapperspb.StringValue{} - if err := proto.Unmarshal(resp.Data, resultVal); err != nil { - log.Printf("failed to unmarshal data: %v", err) - fmt.Printf("Data (Raw): %v\n", resp.Data) - } else { - fmt.Printf("Data: %s\n", resultVal.Value) - } - } else { - fmt.Printf("Data (Raw): %v\n", resp.Data) + resultVal := &wrapperspb.StringValue{} + if err := proto.Unmarshal(resp.Data, resultVal); err != nil { + log.Fatalf("failed to unmarshal data: %v", err) + } + if !strings.Contains(resultVal.Value, "World") { + log.Fatalf("unexpected RPC result: %q", resultVal.Value) } + + fmt.Println("RpcResponse received:") + fmt.Printf("Data: %s\n", resultVal.Value) fmt.Printf("Message: %s\n", resp.Message) fmt.Printf("RequestID: %s\n", resp.RequestId) } diff --git a/python_client/client.py b/python_client/client.py index c4f8e3f..a0ea6f9 100644 --- a/python_client/client.py +++ b/python_client/client.py @@ -1,37 +1,49 @@ +import sys + import grpc +from google.protobuf import wrappers_pb2 + import rpc_meta_pb2 import rpc_meta_pb2_grpc -from google.protobuf import wrappers_pb2 + def run(): - # Connect to the Java gRPC server channel = grpc.insecure_channel('localhost:8080') stub = rpc_meta_pb2_grpc.GrpcServiceStub(channel) - # Wrap the parameter in a Protobuf StringValue - param = wrappers_pb2.StringValue(value="World") - param_bytes = param.SerializeToString() - + param = wrappers_pb2.StringValue(value='World') rpc_request = rpc_meta_pb2.RpcRequest( - interface_name="com.xiaoyu.rpc.api.HelloService", - method_name="sayHello", - param_types=["java.lang.String"], - parameters=[param_bytes] + interface_name='com.xiaoyu.rpc.api.HelloService', + method_name='sayHello', + param_types=['java.lang.String'], + parameters=[param.SerializeToString()], + request_id='python-ci-request', ) try: - response = stub.handle(rpc_request) - print("RpcResponse received:") - # The return value is also a StringValue serialized object - if response.message == "Success": - result_val = wrappers_pb2.StringValue() - result_val.ParseFromString(response.data) - print(f"Data: {result_val.value}") - else: - print(f"Data (Raw): {response.data}") - print(f"Message: {response.message}") - except grpc.RpcError as e: - print(f"gRPC Error: {e.code()} - {e.details()}") + response = stub.handle(rpc_request, timeout=5) + if response.message != 'Success': + raise RuntimeError(f'RPC failed: {response.message}') + + result = wrappers_pb2.StringValue() + result.ParseFromString(response.data) + if 'World' not in result.value: + raise RuntimeError(f'Unexpected RPC result: {result.value!r}') + + print('RpcResponse received:') + print(f'Data: {result.value}') + print(f'Message: {response.message}') + print(f'RequestID: {response.request_id}') + finally: + channel.close() + if __name__ == '__main__': - run() + try: + run() + except grpc.RpcError as exc: + print(f'gRPC Error: {exc.code()} - {exc.details()}', file=sys.stderr) + sys.exit(1) + except Exception as exc: + print(f'Client Error: {exc}', file=sys.stderr) + sys.exit(1) diff --git a/rpc-consumer/src/test/java/com/xiaoyu/rpc/consumer/FullIntegrationTest.java b/rpc-consumer/src/test/java/com/xiaoyu/rpc/consumer/FullIntegrationTest.java index cd38854..7994f4e 100644 --- a/rpc-consumer/src/test/java/com/xiaoyu/rpc/consumer/FullIntegrationTest.java +++ b/rpc-consumer/src/test/java/com/xiaoyu/rpc/consumer/FullIntegrationTest.java @@ -1,5 +1,6 @@ package com.xiaoyu.rpc.consumer; +import com.google.protobuf.ByteString; import com.xiaoyu.rpc.api.HelloService; import com.xiaoyu.rpc.common.serialization.Serializer; import com.xiaoyu.rpc.common.serialization.SerializerCode; @@ -7,20 +8,21 @@ import com.xiaoyu.rpc.core.client.RpcClient; import com.xiaoyu.rpc.core.config.RpcConfig; import com.xiaoyu.rpc.core.server.RpcServer; -import com.google.protobuf.ByteString; -import org.junit.jupiter.api.Assumptions; import org.junit.jupiter.api.Test; -import java.lang.reflect.Field; import java.net.InetAddress; +import java.net.InetSocketAddress; import java.net.ServerSocket; +import java.net.Socket; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; -import static org.junit.jupiter.api.Assertions.*; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.api.Assertions.fail; public class FullIntegrationTest { - // Make the implementation public for reflection access public static class HelloServiceImpl implements HelloService { @Override public String sayHello(String name) { @@ -29,63 +31,53 @@ public String sayHello(String name) { } @Test - public void testFullIntegration() throws InterruptedException { - // Use Local Registry to avoid external dependency - System.setProperty("rpc.registry", "local"); - System.setProperty("rpc.server-port", "9090"); // Use port 9090 - // Ensure we use KRYO or JSON/Hessian serializer that supports mundane Java - // classes - System.setProperty("rpc.serializer", "kryo"); + public void testFullIntegration() throws Exception { + String registry = System.getProperty("rpc.registry", "local"); + String protocol = System.getProperty("rpc.protocol", "netty"); + String serializerName = System.getProperty("rpc.serializer", "kryo"); + int port = findFreePort(); + + System.setProperty("rpc.registry", registry); + System.setProperty("rpc.server-host", "127.0.0.1"); + System.setProperty("rpc.server-port", String.valueOf(port)); + System.setProperty("rpc.serializer", serializerName); System.setProperty("rpc.transport", "netty"); + System.setProperty("rpc.protocol", protocol); - // 沙箱或受限环境无法监听端口时，跳过此集成测试，避免把环境问题算成代码失败 - Assumptions.assumeTrue(canBindLocalPort(9090), "No permission to bind local test port 9090"); - - // 集成测试固定走 netty，减少跨协议变量，确保该用例只验证端到端调用主链路 - forceConfig("protocol", "netty"); - - // Start Server in a thread + AtomicReference serverFailure = new AtomicReference<>(); Thread serverThread = new Thread(() -> { try { RpcServer server = new RpcServer(); - // Explicitly register the service (Demonstrating the new API) server.register(HelloService.class, new HelloServiceImpl()); server.start(); - } catch (Exception e) { - e.printStackTrace(); + } catch (Throwable e) { + serverFailure.set(e); } - }); + }, "rpc-integration-server"); serverThread.setDaemon(true); serverThread.start(); - Thread.sleep(1500); // Wait for server start + waitForServer(port, serverFailure); try { - System.out.println("Starting Client..."); RpcClient rpcClient = new RpcClient(); Serializer serializer = SerializerCode.getSerializerByCode(RpcConfig.getInstance().getSerializerCode()); - System.out.println(">>> First Call"); String result1 = (String) rpcClient .sendRequest(buildRequest("World1", serializer), String.class) .get(5, TimeUnit.SECONDS); - System.out.println("Result1: " + result1); - System.out.println(">>> Second Call (Should reuse connection)"); String result2 = (String) rpcClient .sendRequest(buildRequest("World2", serializer), String.class) .get(5, TimeUnit.SECONDS); - System.out.println("Result2: " + result2); assertNotNull(result1, "Result1 should not be null"); - assertNotNull(result1, "Result2 should not be null"); + assertNotNull(result2, "Result2 should not be null"); assertTrue(result1.contains("World1"), "Result1 should contain World1"); assertTrue(result2.contains("World2"), "Result2 should contain World2"); - - System.out.println("Test Passed!"); - } catch (Exception e) { - e.printStackTrace(); - fail("Test failed with exception: " + e.getMessage()); + } finally { + System.clearProperty("rpc.server-host"); + System.clearProperty("rpc.server-port"); } } @@ -99,22 +91,31 @@ private static RpcRequest buildRequest(String name, Serializer serializer) { .build(); } - private static boolean canBindLocalPort(int port) { - try (ServerSocket ignored = new ServerSocket(port, 1, InetAddress.getByName("127.0.0.1"))) { - return true; - } catch (Exception e) { - return false; + private static int findFreePort() throws Exception { + try (ServerSocket socket = new ServerSocket(0, 1, InetAddress.getByName("127.0.0.1"))) { + return socket.getLocalPort(); } } - private static void forceConfig(String fieldName, Object value) { - try { - RpcConfig config = RpcConfig.getInstance(); - Field field = RpcConfig.class.getDeclaredField(fieldName); - field.setAccessible(true); - field.set(config, value); - } catch (Exception e) { - throw new RuntimeException("Failed to force RpcConfig field: " + fieldName, e); + private static void waitForServer(int port, AtomicReference serverFailure) throws Exception { + long deadlineNanos = System.nanoTime() + TimeUnit.SECONDS.toNanos(8); + Throwable lastFailure = null; + + while (System.nanoTime() < deadlineNanos) { + Throwable startupFailure = serverFailure.get(); + if (startupFailure != null) { + fail("RPC server failed during startup", startupFailure); + } + + try (Socket socket = new Socket()) { + socket.connect(new InetSocketAddress("127.0.0.1", port), 200); + return; + } catch (Exception e) { + lastFailure = e; + Thread.sleep(100); + } } + + fail("RPC server did not become ready on port " + port, lastFailure); } }