Add integration tests for SOFA RPC Bolt: happy path, distributed trace, server error

Author: ValentinZakharov

dd-java-agent/instrumentation/sofarpc/sofarpc-5.0/src/test/groovy/datadog/trace/instrumentation/sofarpc/SofaRpcTest.groovy

@@ -9,20 +9,33 @@ import datadog.trace.agent.test.InstrumentationSpecification
 import datadog.trace.bootstrap.instrumentation.api.Tags
 import spock.lang.Shared
 
+import static datadog.trace.agent.test.utils.TraceUtils.basicSpan
+import static datadog.trace.agent.test.utils.TraceUtils.runUnderTrace
+
 class SofaRpcTest extends InstrumentationSpecification {
 
   @Shared
   int port = 12201
 
+  @Shared
+  int errorPort = 12202
+
   @Shared
   ProviderBootstrap providerBootstrap
 
+  @Shared
+  ProviderBootstrap errorProviderBootstrap
+
   @Shared
   GreeterService greeterService
 
+  @Shared
+  FaultyService faultyService
+
   def setupSpec() {
     ApplicationConfig appConfig = new ApplicationConfig().setAppName("test-server")
 
+    // Happy-path server: Bolt on port 12201
     ServerConfig serverConfig =
       new ServerConfig()
       .setProtocol("bolt")
@@ -39,61 +52,154 @@ class SofaRpcTest extends InstrumentationSpecification {
 
     providerBootstrap = providerConfig.export()
 
-    ConsumerConfig<GreeterService> consumerConfig =
+    greeterService =
       new ConsumerConfig<GreeterService>()
       .setApplication(new ApplicationConfig().setAppName("test-client"))
       .setInterfaceId(GreeterService.name)
       .setDirectUrl("bolt://127.0.0.1:${port}")
       .setProtocol("bolt")
       .setRegister(false)
       .setSubscribe(false)
+      .refer()
+
+    // Error-path server: Bolt on port 12202, separate interface to avoid registry conflict
+    ServerConfig errorServerConfig =
+      new ServerConfig()
+      .setProtocol("bolt")
+      .setHost("127.0.0.1")
+      .setPort(errorPort)
+
+    ProviderConfig<FaultyService> errorProviderConfig =
+      new ProviderConfig<FaultyService>()
+      .setApplication(appConfig)
+      .setInterfaceId(FaultyService.name)
+      .setRef(new FaultyServiceImpl())
+      .setServer(errorServerConfig)
+      .setRegister(false)
 
-    greeterService = consumerConfig.refer()
+    errorProviderBootstrap = errorProviderConfig.export()
+
+    faultyService =
+      new ConsumerConfig<FaultyService>()
+      .setApplication(new ApplicationConfig().setAppName("test-client"))
+      .setInterfaceId(FaultyService.name)
+      .setDirectUrl("bolt://127.0.0.1:${errorPort}")
+      .setProtocol("bolt")
+      .setRegister(false)
+      .setSubscribe(false)
+      .refer()
   }
 
   def cleanupSpec() {
     providerBootstrap?.unExport()
+    errorProviderBootstrap?.unExport()
   }
 
   def "client and server spans created for synchronous Bolt RPC call"() {
     setup:
     String serviceUniqueName = GreeterService.name + ":1.0"
 
     when:
-    String reply = greeterService.sayHello("World")
+    // runUnderTrace gives the client trace 2 spans (caller + sofarpc.request), making it
+    // unambiguously distinguishable from the 1-span server trace in assertTraces below.
+    String reply = runUnderTrace("caller") { greeterService.sayHello("World") }
 
     then:
     reply == "Hello, World"
 
     and:
-    // Client and server are in the same JVM but use real TCP sockets (Bolt),
-    // so each side produces its own local trace entry.
     assertTraces(2) {
-      trace(1) {
+      // trace(0): client side — 2 spans [caller, sofarpc.request(client)]
+      trace(2) {
+        basicSpan(it, "caller")
         span {
           operationName "sofarpc.request"
           resourceName "${serviceUniqueName}/sayHello"
           spanType "rpc"
           errored false
+          childOf span(0)
           tags {
             "$Tags.RPC_SERVICE" serviceUniqueName
+            "rpc.system" "sofarpc"
+            "sofarpc.protocol" "bolt"
             "component" "sofarpc-client"
             "span.kind" "client"
+            peerServiceFrom(Tags.RPC_SERVICE)
             defaultTags()
           }
         }
       }
+      // trace(1): server side — 1 span [sofarpc.request(server)], child of client span
       trace(1) {
         span {
           operationName "sofarpc.request"
           resourceName "${serviceUniqueName}/sayHello"
           spanType "rpc"
           errored false
+          childOf trace(0).get(1)
           tags {
             "$Tags.RPC_SERVICE" serviceUniqueName
+            "rpc.system" "sofarpc"
+            "sofarpc.protocol" "bolt"
             "component" "sofarpc-server"
             "span.kind" "server"
-            defaultTags(true) // distributed root span — parent context propagated via Bolt headers
+            defaultTags(true)
+          }
+        }
+      }
+    }
+  }
+
+  def "server error is marked on server span"() {
+    setup:
+    String serviceUniqueName = FaultyService.name + ":1.0"
+
+    when:
+    // SOFA RPC Bolt propagates server exceptions back to the client as a SofaRpcException.
+    // The client-side AbstractCluster.invoke() returns the SofaResponse to the proxy layer,
+    // which then throws — after our instrumentation's OnMethodExit has already closed the scope.
+    // So the CLIENT span is not errored; only the SERVER span reflects the error.
+    faultyService.fail()
+
+    then:
+    thrown(Exception)
+
+    and:
+    assertTraces(2) {
+      // Traces sorted by root-span start time. Client span starts first (initiates the call),
+      // so the client trace is trace(0) and the server trace is trace(1).
+      trace(1) {
+        span {
+          operationName "sofarpc.request"
+          resourceName "${serviceUniqueName}/fail"
+          spanType "rpc"
+          errored false
+          tags {
+            "$Tags.RPC_SERVICE" serviceUniqueName
+            "rpc.system" "sofarpc"
+            "sofarpc.protocol" "bolt"
+            "component" "sofarpc-client"
+            "span.kind" "client"
+            peerServiceFrom(Tags.RPC_SERVICE)
+            defaultTags()
+          }
+        }
+      }
+      trace(1) {
+        span {
+          operationName "sofarpc.request"
+          resourceName "${serviceUniqueName}/fail"
+          spanType "rpc"
+          errored true
+          childOf trace(0).get(0)
+          tags {
+            "$Tags.RPC_SERVICE" serviceUniqueName
+            "rpc.system" "sofarpc"
+            "sofarpc.protocol" "bolt"
+            "component" "sofarpc-server"
+            "span.kind" "server"
+            "error.message" { String }
+            defaultTags(true)
           }
         }
       }
@@ -110,4 +216,17 @@ class SofaRpcTest extends InstrumentationSpecification {
       return "Hello, ${name}"
     }
   }
+
+  interface FaultyService {
+    // Non-void return type: SOFA RPC Bolt throws SofaRpcException on client side
+    // when the server returns an error response, which is what we verify in the test.
+    String fail()
+  }
+
+  static class FaultyServiceImpl implements FaultyService {
+    @Override
+    String fail() {
+      throw new IllegalStateException("something went wrong")
+    }
+  }
 }