This is an automated email from the ASF dual-hosted git repository.
Aias00 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shenyu.git
The following commit(s) were added to refs/heads/master by this push:
new b4e115367f feat(agent-gateway): trace logical MCP tool calls with
application telemetry (#7622)
b4e115367f is described below
commit b4e115367fd57adc8aba332d655b1009932ba394
Author: BobSong <[email protected]>
AuthorDate: Thu Oct 15 11:21:05 2026 +0800
feat(agent-gateway): trace logical MCP tool calls with application
telemetry (#7622)
Co-authored-by: BobSong-dev <[email protected]>
---
.../shenyu-plugin-agent-gateway/README.md | 46 ++++
.../gateway/protocol/AgentMcpCallObservation.java | 73 +++++-
.../agent/gateway/protocol/AgentMcpCallTracer.java | 55 +++++
.../gateway/protocol/AgentMcpHttpHandler.java | 15 +-
.../gateway/protocol/AgentMcpCallTracerTest.java | 185 ++++++++++++++++
.../pom.xml | 15 ++
.../gateway/AgentGatewayPluginConfiguration.java | 38 +++-
.../gateway/OpenTelemetryAgentMcpCallTracer.java | 77 +++++++
.../AgentGatewayPluginConfigurationTest.java | 52 +++++
.../OpenTelemetryAgentMcpCallTracerTest.java | 244 +++++++++++++++++++++
.../gateway/OpenTelemetryAgentMcpOtlpTest.java | 86 ++++++++
11 files changed, 870 insertions(+), 16 deletions(-)
diff --git a/shenyu-plugin/shenyu-plugin-agent-gateway/README.md
b/shenyu-plugin/shenyu-plugin-agent-gateway/README.md
index f49bc9222f..762b1951d3 100644
--- a/shenyu-plugin/shenyu-plugin-agent-gateway/README.md
+++ b/shenyu-plugin/shenyu-plugin-agent-gateway/README.md
@@ -508,3 +508,49 @@ It is not arbitrary free-text redaction or
prompt-injection prevention. Operator
must enumerate the fields to remove and use compatible tools; unsupported
results
fail closed rather than being described as safely redacted. The original schema
is never rewritten or weakened to make a deletion pass.
+
+## Optional OpenTelemetry call spans
+
+The Agent Gateway starter can trace logical MCP tool calls when
+`shenyu.plugins.agent.gateway.tracing.enabled=true`. Tracing is disabled by
+default. Add the OpenTelemetry API/SDK to the gateway application and supply an
+unambiguous `OpenTelemetry` Spring bean. The starter uses only the API; it does
+not create, register globally, configure or close an SDK/exporter. Explicitly
+enabling tracing without a usable bridge fails startup. A custom
+`AgentMcpCallTracer` bean is an explicit alternative instrumentation opt-in.
+
+Each authenticated, parsed `tools/call` creates one INTERNAL span named
+`shenyu.agent_gateway.tools.call`. It ends once after response writing
completes,
+fails or is cancelled, not when the provider merely computes a result.
Discovery,
+listing, invalid requests and rejection before parsing do not create logical
+call spans. Existing Metrics observations remain independent.
+
+The only span attribute added by this bridge is `shenyu.mcp.outcome`, with the
+existing bounded values `success`, `tool_error`, `rejected`, `server_error`,
+`timeout` and `cancelled`. Business/server errors and timeouts set ERROR
status;
+other outcomes retain UNSET. No exception event, message or stack trace is
+recorded. It does not collect tool names, RPC/request ids, identities,
parameters,
+results, credential headers or endpoint URLs. Application SDK resource
attributes,
+processors, sampling and other instrumentations remain the operator's
responsibility.
+
+To link a call to a trusted application tracing parent, explicitly provide an
+`io.opentelemetry.context.Context` in Reactor Context keyed by that class:
+
+```java
+return chain.filter(exchange).contextWrite(context ->
+ context.put(io.opentelemetry.context.Context.class, trustedParent));
+```
+
+The bridge copies only the parent span identity, not baggage or arbitrary
context
+data. It puts the child span context under the same key for tool execution, so
+`Mono.deferContextual` can access it across scheduler changes. It never reads a
+thread-local/global parent, installs a long-lived Scope, extracts caller
tracing
+headers or injects outbound HTTP headers. Without an explicit Reactor parent,
+the call gets a root span; automatic Java-agent/Micrometer parent integration
is
+not implied. No SDK-wide Reactor hooks are installed.
+
+For OTLP export, configure the application-owned SDK's exporter/processor,
trusted
+collector endpoint, TLS, credentials and shutdown/flush policy in the
application.
+Merely setting the gateway tracing flag or supplying a no-op API does not
export
+telemetry. There is no separate gateway retry, Token accounting or new metrics
+exporter in this feature. Errors in the observation bridge do not fail tool
calls.
diff --git
a/shenyu-plugin/shenyu-plugin-agent-gateway/src/main/java/org/apache/shenyu/plugin/agent/gateway/protocol/AgentMcpCallObservation.java
b/shenyu-plugin/shenyu-plugin-agent-gateway/src/main/java/org/apache/shenyu/plugin/agent/gateway/protocol/AgentMcpCallObservation.java
index ab997a904f..4247687b44 100644
---
a/shenyu-plugin/shenyu-plugin-agent-gateway/src/main/java/org/apache/shenyu/plugin/agent/gateway/protocol/AgentMcpCallObservation.java
+++
b/shenyu-plugin/shenyu-plugin-agent-gateway/src/main/java/org/apache/shenyu/plugin/agent/gateway/protocol/AgentMcpCallObservation.java
@@ -23,11 +23,14 @@ import
org.apache.shenyu.common.metrics.AgentMcpCallObserver.Outcome;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.publisher.SignalType;
+import reactor.util.context.Context;
+import reactor.util.context.ContextView;
import java.util.Objects;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
/**
* One subscription's bounded observation state, independent of RPC ids and
sessions.
@@ -38,7 +41,15 @@ final class AgentMcpCallObservation {
private final AgentMcpCallObserver observer;
- private final AtomicBoolean recorded = new AtomicBoolean();
+ private final AgentMcpCallTracer tracer;
+
+ private final ContextView parentContext;
+
+ private volatile AgentMcpCallTracer.Trace trace;
+
+ private final AtomicReference<Outcome> completed = new AtomicReference<>();
+
+ private final AtomicBoolean traceRecorded = new AtomicBoolean();
private volatile long started;
@@ -47,16 +58,45 @@ final class AgentMcpCallObservation {
private volatile Outcome outcome = Outcome.SERVER_ERROR;
AgentMcpCallObservation(final AgentMcpCallObserver observer) {
+ this(observer, null, Context.empty());
+ }
+
+ AgentMcpCallObservation(final AgentMcpCallObserver observer, final
AgentMcpCallTracer tracer, final ContextView parentContext) {
this.observer = observer;
+ this.tracer = tracer;
+ this.parentContext = parentContext;
}
void parsed(final AgentMcpRequest request) {
- if (Objects.nonNull(observer) &&
"tools/call".equals(request.getMethod())) {
+ if (Objects.nonNull(completed.get())) {
+ return;
+ }
+ if ((Objects.nonNull(observer) || Objects.nonNull(tracer)) &&
"tools/call".equals(request.getMethod())) {
started = System.nanoTime();
call = true;
+ if (Objects.nonNull(tracer)) {
+ try {
+ trace = tracer.start(parentContext);
+ // Cancellation may finish the request while the
application creates its span.
+ finishTrace();
+ } catch (RuntimeException error) {
+ LOG.debug("MCP call tracing callback failed");
+ }
+ }
}
}
+ Context traceContext(final Context context) {
+ if (Objects.nonNull(trace)) {
+ try {
+ return Objects.requireNonNull(trace.attach(context), "trace
context");
+ } catch (RuntimeException error) {
+ LOG.debug("MCP call tracing callback failed");
+ }
+ }
+ return context;
+ }
+
void response(final ObjectNode response) {
outcome = response.path("result").path("isError").asBoolean(false) ?
Outcome.TOOL_ERROR : Outcome.SUCCESS;
}
@@ -72,16 +112,31 @@ final class AgentMcpCallObservation {
}
void finish(final SignalType signal) {
- if (!call || !recorded.compareAndSet(false, true)) {
+ Outcome terminal = signal == SignalType.CANCEL ? Outcome.CANCELLED :
outcome;
+ if (!completed.compareAndSet(null, terminal) || !call) {
return;
}
- Outcome terminal = signal == SignalType.CANCEL ? Outcome.CANCELLED :
outcome;
long millis = TimeUnit.NANOSECONDS.toMillis(Math.max(0,
System.nanoTime() - started));
- try {
- observer.record(terminal, millis);
- } catch (RuntimeException error) {
- // Observability must not fail a call or log potentially sensitive
callback errors.
- LOG.debug("MCP call statistics callback failed");
+ if (Objects.nonNull(observer)) {
+ try {
+ observer.record(terminal, millis);
+ } catch (RuntimeException error) {
+ // Observability must not fail a call or log potentially
sensitive callback errors.
+ LOG.debug("MCP call statistics callback failed");
+ }
+ }
+ finishTrace();
+ }
+
+ private void finishTrace() {
+ AgentMcpCallTracer.Trace current = trace;
+ Outcome terminal = completed.get();
+ if (Objects.nonNull(current) && Objects.nonNull(terminal) &&
traceRecorded.compareAndSet(false, true)) {
+ try {
+ current.finish(terminal);
+ } catch (RuntimeException error) {
+ LOG.debug("MCP call tracing callback failed");
+ }
}
}
}
diff --git
a/shenyu-plugin/shenyu-plugin-agent-gateway/src/main/java/org/apache/shenyu/plugin/agent/gateway/protocol/AgentMcpCallTracer.java
b/shenyu-plugin/shenyu-plugin-agent-gateway/src/main/java/org/apache/shenyu/plugin/agent/gateway/protocol/AgentMcpCallTracer.java
new file mode 100644
index 0000000000..2dcf81fdc3
--- /dev/null
+++
b/shenyu-plugin/shenyu-plugin-agent-gateway/src/main/java/org/apache/shenyu/plugin/agent/gateway/protocol/AgentMcpCallTracer.java
@@ -0,0 +1,55 @@
+/*
+ * 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.agent.gateway.protocol;
+
+import org.apache.shenyu.common.metrics.AgentMcpCallObserver.Outcome;
+import reactor.util.context.Context;
+import reactor.util.context.ContextView;
+
+/**
+ * Optional tracing bridge; each parsed tools/call owns one trace until its
response terminates.
+ */
+@FunctionalInterface
+public interface AgentMcpCallTracer {
+
+ /**
+ * Start one subscription's trace without receiving request data or
identifiers.
+ * @param parentContext subscription-local context
+ * @return exclusively owned trace
+ */
+ Trace start(ContextView parentContext);
+
+ /**
+ * A request-local trace, independent of RPC ids and shared remote clients.
+ */
+ interface Trace {
+
+ /**
+ * Attach explicit execution context without installing a thread-local
scope.
+ * @param context downstream context
+ * @return execution context
+ */
+ Context attach(Context context);
+
+ /**
+ * Finish after execution and response writing, including cancellation.
+ * @param outcome bounded logical terminal outcome
+ */
+ void finish(Outcome outcome);
+ }
+}
diff --git
a/shenyu-plugin/shenyu-plugin-agent-gateway/src/main/java/org/apache/shenyu/plugin/agent/gateway/protocol/AgentMcpHttpHandler.java
b/shenyu-plugin/shenyu-plugin-agent-gateway/src/main/java/org/apache/shenyu/plugin/agent/gateway/protocol/AgentMcpHttpHandler.java
index aa95fe9b95..2ba90c9bdd 100644
---
a/shenyu-plugin/shenyu-plugin-agent-gateway/src/main/java/org/apache/shenyu/plugin/agent/gateway/protocol/AgentMcpHttpHandler.java
+++
b/shenyu-plugin/shenyu-plugin-agent-gateway/src/main/java/org/apache/shenyu/plugin/agent/gateway/protocol/AgentMcpHttpHandler.java
@@ -59,13 +59,20 @@ public final class AgentMcpHttpHandler {
private final AgentMcpSecurityResolver securityResolver;
+ private final AgentMcpCallTracer tracer;
+
private final AgentMcpRequestParser parser = new AgentMcpRequestParser();
private final ObjectMapper mapper = new ObjectMapper();
public AgentMcpHttpHandler(final AgentMcpDispatcher dispatcher, final
AgentMcpSecurityResolver securityResolver) {
+ this(dispatcher, securityResolver, null);
+ }
+
+ public AgentMcpHttpHandler(final AgentMcpDispatcher dispatcher, final
AgentMcpSecurityResolver securityResolver, final AgentMcpCallTracer tracer) {
this.dispatcher = Objects.requireNonNull(dispatcher, "dispatcher");
this.securityResolver = Objects.requireNonNull(securityResolver,
"securityResolver");
+ this.tracer = tracer;
}
/**
@@ -79,10 +86,11 @@ public final class AgentMcpHttpHandler {
*/
public Mono<Void> handle(final ServerWebExchange exchange, final
AgentGatewayMcpConfig config,
final AgentTrafficContext traffic, final long
configurationVersion) {
- return Mono.defer(() -> {
+ return Mono.deferContextual(parentContext -> {
final Instant deadline =
Instant.now().plusMillis(config.getTimeoutMs());
Object callback =
exchange.getAttribute(Constants.METRICS_AGENT_MCP_CALL);
- final AgentMcpCallObservation observation = new
AgentMcpCallObservation(callback instanceof AgentMcpCallObserver observer ?
observer : null);
+ final AgentMcpCallObservation observation = new
AgentMcpCallObservation(callback instanceof AgentMcpCallObserver observer ?
observer : null,
+ tracer, parentContext);
AtomicReference<JsonNode> rpcId = new AtomicReference<>();
return Mono.fromRunnable(() -> preflight(exchange, config))
.then(Mono.defer(() -> securityResolver.resolve(exchange))
@@ -92,7 +100,8 @@ public final class AgentMcpHttpHandler {
rpcId.set(request.getId());
observation.parsed(request);
return dispatcher.dispatch(request, () -> new
AgentMcpExecutionContext(traffic.getRequestId(), identity.getSubject(),
- traffic.getRuleId(),
configurationVersion, config.getAllowedTools(), identity.getToolGrants(),
deadline, config.getResultFilters()));
+ traffic.getRuleId(),
configurationVersion, config.getAllowedTools(), identity.getToolGrants(),
deadline, config.getResultFilters()))
+
.contextWrite(observation::traceContext);
}))
.flatMap(response -> {
observation.response(response);
diff --git
a/shenyu-plugin/shenyu-plugin-agent-gateway/src/test/java/org/apache/shenyu/plugin/agent/gateway/protocol/AgentMcpCallTracerTest.java
b/shenyu-plugin/shenyu-plugin-agent-gateway/src/test/java/org/apache/shenyu/plugin/agent/gateway/protocol/AgentMcpCallTracerTest.java
new file mode 100644
index 0000000000..137cc52872
--- /dev/null
+++
b/shenyu-plugin/shenyu-plugin-agent-gateway/src/test/java/org/apache/shenyu/plugin/agent/gateway/protocol/AgentMcpCallTracerTest.java
@@ -0,0 +1,185 @@
+/*
+ * 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.agent.gateway.protocol;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.fasterxml.jackson.databind.node.TextNode;
+import org.apache.shenyu.common.metrics.AgentMcpCallObserver;
+import org.apache.shenyu.common.metrics.AgentMcpCallObserver.Outcome;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import reactor.core.publisher.SignalType;
+import reactor.util.context.Context;
+
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+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.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class AgentMcpCallTracerTest {
+
+ @Test
+ void parsingAfterTerminationCannotStartANewTrace() {
+ AtomicInteger starts = new AtomicInteger();
+ AtomicInteger metrics = new AtomicInteger();
+ AgentMcpCallObservation observation = new
AgentMcpCallObservation((outcome, millis) -> metrics.incrementAndGet(), ignored
-> {
+ starts.incrementAndGet();
+ return mock(AgentMcpCallTracer.Trace.class);
+ }, Context.empty());
+ observation.finish(SignalType.CANCEL);
+ observation.parsed(request("tools/call"));
+ observation.finish(SignalType.ON_COMPLETE);
+ assertEquals(0, starts.get());
+ assertEquals(0, metrics.get());
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void aSpanCreatedAfterRequestTerminationIsStillFinishedOnce(final boolean
cancel) throws Exception {
+ CountDownLatch entered = new CountDownLatch(1);
+ CountDownLatch release = new CountDownLatch(1);
+ var executor = Executors.newSingleThreadExecutor();
+ AgentMcpCallTracer.Trace trace = mock(AgentMcpCallTracer.Trace.class);
+ AtomicInteger metrics = new AtomicInteger();
+ AgentMcpCallTracer tracer = ignored -> {
+ entered.countDown();
+ try {
+ if (!release.await(5, TimeUnit.SECONDS)) {
+ throw new IllegalStateException("Fixture release timed
out");
+ }
+ } catch (InterruptedException error) {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException("Fixture interrupted", error);
+ }
+ return trace;
+ };
+ AgentMcpCallObservation observation = new
AgentMcpCallObservation((outcome, millis) -> metrics.incrementAndGet(), tracer,
Context.empty());
+ try {
+ final var parsed = executor.submit(() ->
observation.parsed(request("tools/call")));
+ assertTrue(entered.await(5, TimeUnit.SECONDS));
+ observation.failure(new TimeoutException("private-timeout"));
+ observation.finish(cancel ? SignalType.CANCEL :
SignalType.ON_COMPLETE);
+ release.countDown();
+ parsed.get(5, TimeUnit.SECONDS);
+ observation.finish(SignalType.ON_ERROR);
+ verify(trace).finish(cancel ? Outcome.CANCELLED : Outcome.TIMEOUT);
+ assertEquals(1, metrics.get());
+ } finally {
+ release.countDown();
+ executor.shutdownNow();
+ }
+ }
+
+ @Test
+ void tracingWithoutMetricsStillEndsExactlyOnce() {
+ AgentMcpCallTracer.Trace trace = mock(AgentMcpCallTracer.Trace.class);
+ AgentMcpCallObservation observation = new
AgentMcpCallObservation(null, ignored -> trace, Context.empty());
+ observation.parsed(request("tools/call"));
+ observation.finish(SignalType.CANCEL);
+ observation.finish(SignalType.ON_COMPLETE);
+ verify(trace).finish(Outcome.CANCELLED);
+ }
+
+ @ParameterizedTest
+ @ValueSource(ints = {0, 1, 2})
+ void tracingFailuresDoNotPreventMetricsOrChangeContext(final int stage) {
+ AtomicInteger metrics = new AtomicInteger();
+ AgentMcpCallTracer.Trace trace = new AgentMcpCallTracer.Trace() {
+ @Override
+ public Context attach(final Context context) {
+ if (stage == 1) {
+ throw new IllegalStateException("private-attach-error");
+ }
+ return context;
+ }
+
+ @Override
+ public void finish(final Outcome outcome) {
+ if (stage == 2) {
+ throw new IllegalStateException("private-finish-error");
+ }
+ }
+ };
+ AgentMcpCallTracer tracer = ignored -> {
+ if (stage == 0) {
+ throw new IllegalStateException("private-start-error");
+ }
+ return trace;
+ };
+ AgentMcpCallObservation observation = new
AgentMcpCallObservation((outcome, millis) -> metrics.incrementAndGet(), tracer,
Context.empty());
+ observation.parsed(request("tools/call"));
+ Context context = Context.of("private-identity", "private-value");
+ assertSame(context, observation.traceContext(context));
+ observation.finish(SignalType.ON_COMPLETE);
+ assertEquals(1, metrics.get());
+ }
+
+ @Test
+ void metricsFailureDoesNotPreventTraceCompletion() {
+ AgentMcpCallTracer.Trace trace = mock(AgentMcpCallTracer.Trace.class);
+ AgentMcpCallObserver observer = (outcome, millis) -> {
+ throw new IllegalStateException("private-observer-error");
+ };
+ AgentMcpCallObservation observation = new
AgentMcpCallObservation(observer, ignored -> trace, Context.empty());
+ observation.parsed(request("tools/call"));
+ observation.finish(SignalType.ON_COMPLETE);
+ verify(trace).finish(Outcome.SERVER_ERROR);
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"tools/list", "server/discover"})
+ void discoveryDoesNotStartACallTrace(final String method) {
+ AtomicInteger starts = new AtomicInteger();
+ AgentMcpCallObservation observation = new
AgentMcpCallObservation(null, ignored -> {
+ starts.incrementAndGet();
+ return mock(AgentMcpCallTracer.Trace.class);
+ }, Context.empty());
+ observation.parsed(request(method));
+ observation.finish(SignalType.ON_COMPLETE);
+ assertEquals(0, starts.get());
+ }
+
+ @Test
+ void invalidTraceContextCannotReplaceExecutionContextWithNull() {
+ AgentMcpCallTracer.Trace trace = mock(AgentMcpCallTracer.Trace.class);
+ when(trace.attach(any())).thenReturn(null);
+ AgentMcpCallObservation observation = new
AgentMcpCallObservation(null, ignored -> trace, Context.empty());
+ observation.parsed(request("tools/call"));
+ Context context = Context.empty();
+ assertSame(context, observation.traceContext(context));
+ observation.finish(SignalType.CANCEL);
+ verify(trace).finish(Outcome.CANCELLED);
+ }
+
+ private AgentMcpRequest request(final String method) {
+ ObjectNode params = new ObjectMapper().createObjectNode();
+
params.putObject("_meta").putObject("io.modelcontextprotocol/clientCapabilities");
+ return new AgentMcpRequest(TextNode.valueOf("same-id"), method,
params);
+ }
+}
diff --git
a/shenyu-spring-boot-starter/shenyu-spring-boot-starter-plugin/shenyu-spring-boot-starter-plugin-agent-gateway/pom.xml
b/shenyu-spring-boot-starter/shenyu-spring-boot-starter-plugin/shenyu-spring-boot-starter-plugin-agent-gateway/pom.xml
index 1ba6f72faa..3ce2ba86b8 100644
---
a/shenyu-spring-boot-starter/shenyu-spring-boot-starter-plugin/shenyu-spring-boot-starter-plugin-agent-gateway/pom.xml
+++
b/shenyu-spring-boot-starter/shenyu-spring-boot-starter-plugin/shenyu-spring-boot-starter-plugin-agent-gateway/pom.xml
@@ -26,6 +26,21 @@
<artifactId>shenyu-spring-boot-starter-plugin-agent-gateway</artifactId>
<dependencies>
+ <dependency>
+ <groupId>io.opentelemetry</groupId>
+ <artifactId>opentelemetry-api</artifactId>
+ <optional>true</optional>
+ </dependency>
+ <dependency>
+ <groupId>io.opentelemetry</groupId>
+ <artifactId>opentelemetry-sdk-testing</artifactId>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>io.opentelemetry</groupId>
+ <artifactId>opentelemetry-exporter-otlp</artifactId>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.apache.shenyu</groupId>
<artifactId>shenyu-plugin-agent-gateway</artifactId>
diff --git
a/shenyu-spring-boot-starter/shenyu-spring-boot-starter-plugin/shenyu-spring-boot-starter-plugin-agent-gateway/src/main/java/org/apache/shenyu/springboot/starter/plugin/agent/gateway/AgentGatewayPluginConfiguration.java
b/shenyu-spring-boot-starter/shenyu-spring-boot-starter-plugin/shenyu-spring-boot-starter-plugin-agent-gateway/src/main/java/org/apache/shenyu/springboot/starter/plugin/agent/gateway/AgentGatewayPluginConfiguration.java
index 0475fb0273..4ab469170e 100644
---
a/shenyu-spring-boot-starter/shenyu-spring-boot-starter-plugin/shenyu-spring-boot-starter-plugin-agent-gateway/src/main/java/org/apache/shenyu/springboot/starter/plugin/agent/gateway/AgentGatewayPluginConfiguration.java
+++
b/shenyu-spring-boot-starter/shenyu-spring-boot-starter-plugin/shenyu-spring-boot-starter-plugin-agent-gateway/src/main/java/org/apache/shenyu/springboot/starter/plugin/agent/gateway/AgentGatewayPluginConfiguration.java
@@ -17,6 +17,8 @@
package org.apache.shenyu.springboot.starter.plugin.agent.gateway;
+import io.opentelemetry.api.OpenTelemetry;
+import org.apache.shenyu.plugin.agent.gateway.protocol.AgentMcpCallTracer;
import org.apache.shenyu.plugin.agent.gateway.AgentGatewayPlugin;
import org.apache.shenyu.plugin.agent.gateway.AgentGatewayConstants;
import
org.apache.shenyu.plugin.agent.gateway.handler.AgentGatewayPluginDataHandler;
@@ -29,6 +31,7 @@ import
org.apache.shenyu.plugin.agent.gateway.tool.AgentToolRegistry;
import org.apache.shenyu.plugin.api.ShenyuPlugin;
import org.apache.shenyu.plugin.base.handler.PluginDataHandler;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.beans.factory.ObjectProvider;
@@ -56,19 +59,26 @@ public class AgentGatewayPluginConfiguration {
* @param providers local providers
* @param resolvers trusted identity adapters
* @param catalogs optional, unambiguous managed remote catalog
+ * @param tracers optional tracing bridge
+ * @param environment deployment-controlled tracing opt-in
* @return the plugin
*/
@Bean
public ShenyuPlugin agentGatewayPlugin(final
ObjectProvider<AgentToolProvider> providers, final
ObjectProvider<AgentMcpSecurityResolver> resolvers,
- final
ObjectProvider<AgentMcpRemoteCatalog> catalogs) {
- return createPlugin(providers, resolvers, catalogs.getIfAvailable());
+ final
ObjectProvider<AgentMcpRemoteCatalog> catalogs, final
ObjectProvider<AgentMcpCallTracer> tracers,
+ final Environment environment) {
+ AgentMcpCallTracer tracer = tracers.getIfAvailable();
+ if
(environment.getProperty("shenyu.plugins.agent.gateway.tracing.enabled",
Boolean.class, false) && java.util.Objects.isNull(tracer)) {
+ throw new IllegalStateException("Enabled MCP tracing requires an
application tracing bridge");
+ }
+ return createPlugin(providers, resolvers, catalogs.getIfAvailable(),
tracer);
}
private ShenyuPlugin createPlugin(final ObjectProvider<AgentToolProvider>
providers, final ObjectProvider<AgentMcpSecurityResolver> resolvers,
- final AgentMcpRemoteCatalog catalog) {
+ final AgentMcpRemoteCatalog catalog,
final AgentMcpCallTracer tracer) {
AgentToolRegistry registry = new
AgentToolRegistry(providers.orderedStream().toList());
AgentMcpSecurityResolver resolver = resolvers.getIfAvailable(() ->
exchange -> Mono.empty());
- return new AgentGatewayPlugin(new AgentMcpHttpHandler(new
AgentMcpDispatcher(registry, "shenyu-agent-gateway",
AgentGatewayConstants.MCP_SERVER_VERSION, catalog), resolver));
+ return new AgentGatewayPlugin(new AgentMcpHttpHandler(new
AgentMcpDispatcher(registry, "shenyu-agent-gateway",
AgentGatewayConstants.MCP_SERVER_VERSION, catalog), resolver, tracer));
}
/**
@@ -101,4 +111,24 @@ public class AgentGatewayPluginConfiguration {
Path.of(environment.getRequiredProperty("shenyu.plugins.agent.gateway.aggregation.credentials-directory"))));
return new ManagedRemoteMcpCatalog(allowed,
tools.orderedStream().map(AgentToolProvider::getName).collect(Collectors.toUnmodifiableSet()),
resolver);
}
+
+ /**
+ * Optional API bridge; SDK, exporters and their lifecycle belong to the
application.
+ */
+ @Configuration(proxyBeanMethods = false)
+ @ConditionalOnClass(OpenTelemetry.class)
+ @ConditionalOnProperty(value =
"shenyu.plugins.agent.gateway.tracing.enabled", havingValue = "true")
+ static class TracingConfiguration {
+
+ /**
+ * Use the explicitly supplied application telemetry, never a second
SDK or global singleton.
+ * @param telemetry application-owned telemetry
+ * @return request-local tracing bridge
+ */
+ @Bean
+ @ConditionalOnMissingBean(AgentMcpCallTracer.class)
+ AgentMcpCallTracer agentGatewayCallTracer(final OpenTelemetry
telemetry) {
+ return new OpenTelemetryAgentMcpCallTracer(telemetry);
+ }
+ }
}
diff --git
a/shenyu-spring-boot-starter/shenyu-spring-boot-starter-plugin/shenyu-spring-boot-starter-plugin-agent-gateway/src/main/java/org/apache/shenyu/springboot/starter/plugin/agent/gateway/OpenTelemetryAgentMcpCallTracer.java
b/shenyu-spring-boot-starter/shenyu-spring-boot-starter-plugin/shenyu-spring-boot-starter-plugin-agent-gateway/src/main/java/org/apache/shenyu/springboot/starter/plugin/agent/gateway/OpenTelemetryAgentMcpCallTracer.java
new file mode 100644
index 0000000000..ccc5cc37c3
--- /dev/null
+++
b/shenyu-spring-boot-starter/shenyu-spring-boot-starter-plugin/shenyu-spring-boot-starter-plugin-agent-gateway/src/main/java/org/apache/shenyu/springboot/starter/plugin/agent/gateway/OpenTelemetryAgentMcpCallTracer.java
@@ -0,0 +1,77 @@
+/*
+ * 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.springboot.starter.plugin.agent.gateway;
+
+import io.opentelemetry.api.OpenTelemetry;
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.api.trace.SpanKind;
+import io.opentelemetry.api.trace.StatusCode;
+import io.opentelemetry.api.trace.Tracer;
+import org.apache.shenyu.common.metrics.AgentMcpCallObserver.Outcome;
+import org.apache.shenyu.plugin.agent.gateway.protocol.AgentMcpCallTracer;
+import reactor.util.context.Context;
+import reactor.util.context.ContextView;
+
+import java.util.Locale;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * Bounded MCP spans using explicitly propagated Reactor context, without
thread-local scopes.
+ */
+final class OpenTelemetryAgentMcpCallTracer implements AgentMcpCallTracer {
+
+ private final Tracer tracer;
+
+ OpenTelemetryAgentMcpCallTracer(final OpenTelemetry telemetry) {
+ tracer = telemetry.getTracer("org.apache.shenyu.agent.gateway");
+ }
+
+ @Override
+ public Trace start(final ContextView parentContext) {
+ Object value =
parentContext.getOrDefault(io.opentelemetry.context.Context.class, null);
+ io.opentelemetry.context.Context parent =
io.opentelemetry.context.Context.root();
+ if (value instanceof io.opentelemetry.context.Context context) {
+ // Keep only the parent span identity, never baggage or unrelated
application data.
+ parent =
parent.with(Span.wrap(Span.fromContext(context).getSpanContext()));
+ }
+ Span span =
tracer.spanBuilder("shenyu.agent_gateway.tools.call").setSpanKind(SpanKind.INTERNAL).setParent(parent).startSpan();
+ io.opentelemetry.context.Context execution =
io.opentelemetry.context.Context.root().with(span);
+ AtomicBoolean ended = new AtomicBoolean();
+ return new Trace() {
+ @Override
+ public Context attach(final Context context) {
+ return context.put(io.opentelemetry.context.Context.class,
execution);
+ }
+
+ @Override
+ public void finish(final Outcome outcome) {
+ if (!ended.compareAndSet(false, true)) {
+ return;
+ }
+ try {
+ span.setAttribute("shenyu.mcp.outcome",
outcome.name().toLowerCase(Locale.ROOT));
+ if (outcome == Outcome.TOOL_ERROR || outcome ==
Outcome.SERVER_ERROR || outcome == Outcome.TIMEOUT) {
+ span.setStatus(StatusCode.ERROR);
+ }
+ } finally {
+ span.end();
+ }
+ }
+ };
+ }
+}
diff --git
a/shenyu-spring-boot-starter/shenyu-spring-boot-starter-plugin/shenyu-spring-boot-starter-plugin-agent-gateway/src/test/java/org/apache/shenyu/springboot/starter/plugin/agent/gateway/AgentGatewayPluginConfigurationTest.java
b/shenyu-spring-boot-starter/shenyu-spring-boot-starter-plugin/shenyu-spring-boot-starter-plugin-agent-gateway/src/test/java/org/apache/shenyu/springboot/starter/plugin/agent/gateway/AgentGatewayPluginConfigurationTest.java
index 0a667ffef5..1385178896 100644
---
a/shenyu-spring-boot-starter/shenyu-spring-boot-starter-plugin/shenyu-spring-boot-starter-plugin-agent-gateway/src/test/java/org/apache/shenyu/springboot/starter/plugin/agent/gateway/AgentGatewayPluginConfigurationTest.java
+++
b/shenyu-spring-boot-starter/shenyu-spring-boot-starter-plugin/shenyu-spring-boot-starter-plugin-agent-gateway/src/test/java/org/apache/shenyu/springboot/starter/plugin/agent/gateway/AgentGatewayPluginConfigurationTest.java
@@ -20,6 +20,9 @@ package
org.apache.shenyu.springboot.starter.plugin.agent.gateway;
import org.apache.shenyu.plugin.agent.gateway.AgentGatewayPlugin;
import org.apache.shenyu.plugin.agent.gateway.AgentTrafficContext;
import org.apache.shenyu.plugin.agent.gateway.protocol.AgentMcpHttpHandler;
+import org.apache.shenyu.plugin.agent.gateway.protocol.AgentMcpCallTracer;
+import io.opentelemetry.api.OpenTelemetry;
+import org.springframework.boot.test.context.FilteredClassLoader;
import org.apache.shenyu.plugin.agent.gateway.security.AgentMcpIdentity;
import
org.apache.shenyu.plugin.agent.gateway.security.AgentMcpSecurityResolver;
import org.apache.shenyu.plugin.agent.gateway.tool.AgentToolProvider;
@@ -59,6 +62,55 @@ class AgentGatewayPluginConfigurationTest {
});
}
+ @Test
+ void tracingIsOffByDefaultEvenWithApplicationTelemetry() {
+
contextRunner.withPropertyValues("shenyu.plugins.agent.gateway.enabled=true")
+ .withBean(OpenTelemetry.class, OpenTelemetry::noop)
+ .run(context ->
assertFalse(context.containsBean("agentGatewayCallTracer")));
+ }
+
+ @Test
+ void enabledTracingUsesApplicationTelemetry() {
+
contextRunner.withPropertyValues("shenyu.plugins.agent.gateway.enabled=true",
"shenyu.plugins.agent.gateway.tracing.enabled=true")
+ .withBean(OpenTelemetry.class, OpenTelemetry::noop)
+ .run(context -> {
+ assertTrue(context.getBean(AgentMcpCallTracer.class)
instanceof OpenTelemetryAgentMcpCallTracer);
+ AgentMcpHttpHandler handler = (AgentMcpHttpHandler)
ReflectionTestUtils.getField(context.getBean(AgentGatewayPlugin.class),
"mcpHandler");
+ assertEquals(context.getBean(AgentMcpCallTracer.class),
ReflectionTestUtils.getField(handler, "tracer"));
+ });
+ }
+
+ @Test
+ void enabledTracingRequiresApplicationTelemetry() {
+
contextRunner.withPropertyValues("shenyu.plugins.agent.gateway.enabled=true",
"shenyu.plugins.agent.gateway.tracing.enabled=true")
+ .run(context -> assertNotNull(context.getStartupFailure()));
+ }
+
+ @Test
+ void optionalTelemetryApiIsNotRequiredWhenDisabled() {
+ contextRunner.withClassLoader(new
FilteredClassLoader("io.opentelemetry"))
+
.withPropertyValues("shenyu.plugins.agent.gateway.enabled=true")
+ .run(context -> {
+ assertNotNull(context.getBean(AgentGatewayPlugin.class));
+
assertFalse(context.containsBean("agentGatewayCallTracer"));
+ });
+ }
+
+ @Test
+ void explicitlyEnabledTracingCannotSilentlyDisappearWithoutApi() {
+ contextRunner.withClassLoader(new
FilteredClassLoader("io.opentelemetry"))
+
.withPropertyValues("shenyu.plugins.agent.gateway.enabled=true",
"shenyu.plugins.agent.gateway.tracing.enabled=true")
+ .run(context -> assertNotNull(context.getStartupFailure()));
+ }
+
+ @Test
+ void customTracingBridgeDoesNotRequireAnotherSdk() {
+ AgentMcpCallTracer tracer = ignored ->
mock(AgentMcpCallTracer.Trace.class);
+
contextRunner.withPropertyValues("shenyu.plugins.agent.gateway.enabled=true",
"shenyu.plugins.agent.gateway.tracing.enabled=true")
+ .withBean(AgentMcpCallTracer.class, () -> tracer)
+ .run(context -> assertEquals(tracer,
context.getBean(AgentMcpCallTracer.class)));
+ }
+
@Test
void shouldAssemblePluginWhenEnabled() {
contextRunner.withPropertyValues("shenyu.plugins.agent.gateway.enabled=true")
diff --git
a/shenyu-spring-boot-starter/shenyu-spring-boot-starter-plugin/shenyu-spring-boot-starter-plugin-agent-gateway/src/test/java/org/apache/shenyu/springboot/starter/plugin/agent/gateway/OpenTelemetryAgentMcpCallTracerTest.java
b/shenyu-spring-boot-starter/shenyu-spring-boot-starter-plugin/shenyu-spring-boot-starter-plugin-agent-gateway/src/test/java/org/apache/shenyu/springboot/starter/plugin/agent/gateway/OpenTelemetryAgentMcpCallTracerTest.java
new file mode 100644
index 0000000000..99e15b9811
--- /dev/null
+++
b/shenyu-spring-boot-starter/shenyu-spring-boot-starter-plugin/shenyu-spring-boot-starter-plugin-agent-gateway/src/test/java/org/apache/shenyu/springboot/starter/plugin/agent/gateway/OpenTelemetryAgentMcpCallTracerTest.java
@@ -0,0 +1,244 @@
+/*
+ * 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.springboot.starter.plugin.agent.gateway;
+
+import com.google.gson.JsonObject;
+import com.google.gson.JsonParser;
+import io.opentelemetry.api.baggage.Baggage;
+import io.opentelemetry.api.common.AttributeKey;
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.api.trace.SpanKind;
+import io.opentelemetry.api.trace.StatusCode;
+import io.opentelemetry.context.Context;
+import io.opentelemetry.context.Scope;
+import io.opentelemetry.sdk.OpenTelemetrySdk;
+import io.opentelemetry.sdk.testing.exporter.InMemorySpanExporter;
+import io.opentelemetry.sdk.trace.SdkTracerProvider;
+import io.opentelemetry.sdk.trace.data.SpanData;
+import io.opentelemetry.sdk.trace.export.SimpleSpanProcessor;
+import org.apache.shenyu.common.dto.AgentGatewayMcpConfig;
+import org.apache.shenyu.common.metrics.AgentMcpCallObserver.Outcome;
+import org.apache.shenyu.plugin.agent.gateway.AgentTrafficContext;
+import org.apache.shenyu.plugin.agent.gateway.protocol.AgentMcpCallTracer;
+import org.apache.shenyu.plugin.agent.gateway.protocol.AgentMcpDispatcher;
+import org.apache.shenyu.plugin.agent.gateway.protocol.AgentMcpHttpHandler;
+import org.apache.shenyu.plugin.agent.gateway.security.AgentMcpIdentity;
+import org.apache.shenyu.plugin.agent.gateway.tool.AgentToolExecutionException;
+import org.apache.shenyu.plugin.agent.gateway.tool.AgentToolProvider;
+import org.apache.shenyu.plugin.agent.gateway.tool.AgentToolRegistry;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.springframework.core.io.buffer.DataBufferUtils;
+import org.springframework.mock.http.server.reactive.MockServerHttpRequest;
+import org.springframework.mock.web.server.MockServerWebExchange;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Schedulers;
+import reactor.test.StepVerifier;
+
+import java.time.Duration;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.awaitility.Awaitility.await;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+class OpenTelemetryAgentMcpCallTracerTest {
+
+ @ParameterizedTest
+ @CsvSource({"json,SUCCESS", "sse,SUCCESS", "json,TOOL_ERROR",
"sse,TOOL_ERROR", "json,SERVER_ERROR", "sse,SERVER_ERROR", "json,REJECTED",
"sse,REJECTED"})
+ void exportsOnlyBoundedTerminalData(final String mode, final Outcome
expected) {
+ try (Fixture fixture = new Fixture()) {
+ Mono<JsonObject> action = switch (expected) {
+ case TOOL_ERROR -> Mono.error(new
AgentToolExecutionException("private-business-error"));
+ case SERVER_ERROR -> Mono.error(new
IllegalStateException("private-server-error"));
+ default ->
Mono.just(JsonParser.parseString("{\"private-result\":\"secret\"}").getAsJsonObject());
+ };
+ MockServerWebExchange exchange = exchange("tools/call");
+ StepVerifier.create(fixture.handler(action, expected ==
Outcome.REJECTED ? Set.of() : Set.of("read"))
+ .handle(exchange, config(mode, 30000), traffic(),
1)).verifyComplete();
+ SpanData span = fixture.single(expected);
+ assertEquals(SpanKind.INTERNAL, span.getKind());
+ assertEquals(expected == Outcome.TOOL_ERROR || expected ==
Outcome.SERVER_ERROR ? StatusCode.ERROR : StatusCode.UNSET,
span.getStatus().getStatusCode());
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"json", "sse"})
+ void deadlineEndsSpanOnceAndCancelsItsExecution(final String mode) {
+ try (Fixture fixture = new Fixture()) {
+ AtomicBoolean cancelled = new AtomicBoolean();
+ MockServerWebExchange exchange = exchange("tools/call");
+ AgentMcpHttpHandler handler =
fixture.handler(Mono.<JsonObject>never().doOnCancel(() -> cancelled.set(true)),
Set.of("read"));
+ StepVerifier.withVirtualTime(() -> handler.handle(exchange,
config(mode, 100), traffic(), 1))
+ .thenAwait(Duration.ofMillis(101)).verifyComplete();
+ assertEquals(504, exchange.getResponse().getStatusCode().value());
+ assertTrue(cancelled.get());
+ assertEquals(StatusCode.ERROR,
fixture.single(Outcome.TIMEOUT).getStatus().getStatusCode());
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"json", "sse"})
+ void responseWritingMustTerminateBeforeSpanEnds(final String mode) {
+ try (Fixture fixture = new Fixture()) {
+ MockServerWebExchange exchange = exchange("tools/call");
+ exchange.getResponse().setWriteHandler(buffers ->
buffers.doOnNext(DataBufferUtils::release).then(Mono.never()));
+ StepVerifier.create(fixture.handler(Mono.just(new JsonObject()),
Set.of("read")).handle(exchange, config(mode, 30000), traffic(), 1))
+ .then(() ->
assertTrue(fixture.exporter.getFinishedSpanItems().isEmpty())).thenCancel().verify();
+ fixture.single(Outcome.CANCELLED);
+ }
+ }
+
+ @Test
+ void reactorParentAndChildStayPrivateAcrossThirtyTwoSameIdCalls() {
+ try (Fixture fixture = new Fixture()) {
+ List<String> parents = new CopyOnWriteArrayList<>();
+ Map<String, String> children = new ConcurrentHashMap<>();
+ Mono<JsonObject> action = Mono.deferContextual(context -> {
+ Context execution = context.get(Context.class);
+ assertTrue(Baggage.fromContext(execution).isEmpty());
+
children.put(Span.fromContext(execution).getSpanContext().getSpanId(),
context.get("expected-parent"));
+ return Mono.just(new JsonObject());
+ }).subscribeOn(Schedulers.parallel());
+ AgentMcpHttpHandler handler = fixture.handler(action,
Set.of("read"));
+ StepVerifier.create(Flux.range(0, 32).flatMap(index -> {
+ Span parent =
fixture.sdk.getTracer("test-parent").spanBuilder("parent").startSpan();
+ parents.add(parent.getSpanContext().getSpanId());
+ Context context = Baggage.builder().put("private-subject",
"secret").build().storeInContext(Context.root().with(parent));
+ return handler.handle(exchange("tools/call"), config("json",
30000), traffic(), index)
+ .contextWrite(value -> value.put(Context.class,
context).put("expected-parent", parent.getSpanContext().getSpanId()))
+ .doFinally(signal -> parent.end());
+ }, 32)).verifyComplete();
+ await().atMost(Duration.ofSeconds(3)).until(() ->
fixture.exporter.getFinishedSpanItems().size() == 64);
+ List<SpanData> calls =
fixture.exporter.getFinishedSpanItems().stream().filter(span ->
"shenyu.agent_gateway.tools.call".equals(span.getName())).toList();
+ assertEquals(32, calls.size());
+ assertEquals(32, children.size());
+ assertEquals(Set.copyOf(parents),
calls.stream().map(SpanData::getParentSpanId).collect(java.util.stream.Collectors.toSet()));
+ assertEquals(children.keySet(),
calls.stream().map(SpanData::getSpanId).collect(java.util.stream.Collectors.toSet()));
+ calls.forEach(span -> assertEquals(children.get(span.getSpanId()),
span.getParentSpanId()));
+ assertFalse(Span.current().getSpanContext().isValid());
+ }
+ }
+
+ @Test
+ void doesNotInheritThreadLocalParentOrEndIt() {
+ try (Fixture fixture = new Fixture()) {
+ Span unrelated =
fixture.sdk.getTracer("unrelated").spanBuilder("unrelated").startSpan();
+ try (Scope ignored = unrelated.makeCurrent()) {
+ AgentMcpCallTracer.Trace trace =
fixture.tracer.start(reactor.util.context.Context.empty());
+ trace.finish(Outcome.SUCCESS);
+ trace.finish(Outcome.TIMEOUT);
+
assertFalse(fixture.single(Outcome.SUCCESS).getParentSpanContext().isValid());
+ assertEquals(unrelated.getSpanContext(),
Span.current().getSpanContext());
+ } finally {
+ unrelated.end();
+ }
+ }
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"tools/list", "server/discover"})
+ void discoveryDoesNotExportLogicalCallSpans(final String method) {
+ try (Fixture fixture = new Fixture()) {
+ StepVerifier.create(fixture.handler(Mono.just(new JsonObject()),
Set.of("read"))
+ .handle(exchange(method), config("json", 30000),
traffic(), 1)).verifyComplete();
+ assertTrue(fixture.exporter.getFinishedSpanItems().isEmpty());
+ }
+ }
+
+ @Test
+ void cancellingBeforeParsingDoesNotCreateASpan() {
+ try (Fixture fixture = new Fixture()) {
+ StepVerifier.create(fixture.handler(Mono.never(), Set.of("read"))
+ .handle(exchange("tools/call"), config("json", 30000),
traffic(), 1)).thenCancel().verify();
+ assertTrue(fixture.exporter.getFinishedSpanItems().isEmpty());
+ }
+ }
+
+ private static MockServerWebExchange exchange(final String method) {
+ MockServerHttpRequest.BodyBuilder builder =
MockServerHttpRequest.post("/agent").header("Content-Type", "application/json")
+ .header("Accept", "application/json,
text/event-stream").header("MCP-Protocol-Version",
"2026-07-28").header("Mcp-Method", method);
+ if ("tools/call".equals(method)) {
+ builder.header("Mcp-Name", "read");
+ }
+ return
MockServerWebExchange.from(builder.body("{\"jsonrpc\":\"2.0\",\"id\":\"same-id\",\"method\":\""
+ method + "\","
+ +
"\"params\":{\"name\":\"read\",\"_meta\":{\"io.modelcontextprotocol/protocolVersion\":\"2026-07-28\","
+ + "\"io.modelcontextprotocol/clientCapabilities\":{}}}}"));
+ }
+
+ private static AgentGatewayMcpConfig config(final String mode, final int
timeout) {
+ return
AgentGatewayMcpConfig.parse(JsonParser.parseString("{\"allowedTools\":[\"read\"],\"responseMode\":\""
+ mode
+ + "\",\"timeoutMs\":" + timeout + "}").getAsJsonObject(),
true);
+ }
+
+ private static AgentTrafficContext traffic() {
+ return new AgentTrafficContext("private-id", "mcp",
"private-selector", "private-rule");
+ }
+
+ private static final class Fixture implements AutoCloseable {
+
+ private final InMemorySpanExporter exporter =
InMemorySpanExporter.create();
+
+ private final SdkTracerProvider provider =
SdkTracerProvider.builder().addSpanProcessor(SimpleSpanProcessor.create(exporter)).build();
+
+ private final OpenTelemetrySdk sdk =
OpenTelemetrySdk.builder().setTracerProvider(provider).build();
+
+ private final OpenTelemetryAgentMcpCallTracer tracer = new
OpenTelemetryAgentMcpCallTracer(sdk);
+
+ private AgentMcpHttpHandler handler(final Mono<JsonObject> action,
final Set<String> grants) {
+ AgentToolProvider tool = mock(AgentToolProvider.class);
+ when(tool.getName()).thenReturn("read");
+ when(tool.getDescription()).thenReturn("private-description");
+
when(tool.getInputSchema()).thenReturn(JsonParser.parseString("{\"type\":\"object\"}").getAsJsonObject());
+ when(tool.getRequiredClientCapabilities()).thenReturn(new
JsonObject());
+ when(tool.invoke(any())).thenReturn(action);
+ return new AgentMcpHttpHandler(new AgentMcpDispatcher(new
AgentToolRegistry(List.of(tool)), "test", "1"),
+ ignored -> Mono.just(new
AgentMcpIdentity("private-subject", grants)), tracer);
+ }
+
+ private SpanData single(final Outcome outcome) {
+ List<SpanData> spans = exporter.getFinishedSpanItems();
+ assertEquals(1, spans.size());
+ SpanData span = spans.get(0);
+ assertEquals("shenyu.agent_gateway.tools.call", span.getName());
+ assertEquals(1, span.getAttributes().size());
+ assertEquals(outcome.name().toLowerCase(java.util.Locale.ROOT),
span.getAttributes().get(AttributeKey.stringKey("shenyu.mcp.outcome")));
+ assertTrue(span.getEvents().isEmpty());
+ assertTrue(span.getLinks().isEmpty());
+ assertTrue(span.getEndEpochNanos() >= span.getStartEpochNanos());
+ return span;
+ }
+
+ @Override
+ public void close() {
+ sdk.close();
+ }
+ }
+}
diff --git
a/shenyu-spring-boot-starter/shenyu-spring-boot-starter-plugin/shenyu-spring-boot-starter-plugin-agent-gateway/src/test/java/org/apache/shenyu/springboot/starter/plugin/agent/gateway/OpenTelemetryAgentMcpOtlpTest.java
b/shenyu-spring-boot-starter/shenyu-spring-boot-starter-plugin/shenyu-spring-boot-starter-plugin-agent-gateway/src/test/java/org/apache/shenyu/springboot/starter/plugin/agent/gateway/OpenTelemetryAgentMcpOtlpTest.java
new file mode 100644
index 0000000000..0dfca132cc
--- /dev/null
+++
b/shenyu-spring-boot-starter/shenyu-spring-boot-starter-plugin/shenyu-spring-boot-starter-plugin-agent-gateway/src/test/java/org/apache/shenyu/springboot/starter/plugin/agent/gateway/OpenTelemetryAgentMcpOtlpTest.java
@@ -0,0 +1,86 @@
+/*
+ * 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.springboot.starter.plugin.agent.gateway;
+
+import com.sun.net.httpserver.HttpServer;
+import io.opentelemetry.api.OpenTelemetry;
+import io.opentelemetry.exporter.otlp.http.trace.OtlpHttpSpanExporter;
+import io.opentelemetry.sdk.OpenTelemetrySdk;
+import io.opentelemetry.sdk.resources.Resource;
+import io.opentelemetry.sdk.trace.SdkTracerProvider;
+import io.opentelemetry.sdk.trace.export.SimpleSpanProcessor;
+import org.apache.shenyu.common.metrics.AgentMcpCallObserver.Outcome;
+import org.apache.shenyu.plugin.agent.gateway.protocol.AgentMcpCallTracer;
+import org.junit.jupiter.api.Test;
+import reactor.util.context.Context;
+
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class OpenTelemetryAgentMcpOtlpTest {
+
+ @Test
+ void applicationOwnedSdkExportsOnceOverRealOtlpHttp() throws Exception {
+ AtomicReference<byte[]> payload = new AtomicReference<>();
+ AtomicReference<String> contentType = new AtomicReference<>();
+ AtomicReference<String> method = new AtomicReference<>();
+ AtomicInteger requests = new AtomicInteger();
+ HttpServer receiver = HttpServer.create(new
InetSocketAddress("127.0.0.1", 0), 0);
+ receiver.createContext("/v1/traces", exchange -> {
+ requests.incrementAndGet();
+ method.set(exchange.getRequestMethod());
+
contentType.set(exchange.getRequestHeaders().getFirst("Content-Type"));
+ payload.set(exchange.getRequestBody().readAllBytes());
+ exchange.getResponseHeaders().set("Content-Type",
"application/x-protobuf");
+ exchange.sendResponseHeaders(200, -1);
+ exchange.close();
+ });
+ receiver.start();
+ try {
+ OtlpHttpSpanExporter exporter =
OtlpHttpSpanExporter.builder().setTimeout(Duration.ofSeconds(3))
+ .setEndpoint("http://127.0.0.1:"; +
receiver.getAddress().getPort() + "/v1/traces").build();
+ try (SdkTracerProvider provider =
SdkTracerProvider.builder().setResource(Resource.empty())
+
.addSpanProcessor(SimpleSpanProcessor.create(exporter)).build()) {
+ OpenTelemetry telemetry =
OpenTelemetrySdk.builder().setTracerProvider(provider).build();
+ AgentMcpCallTracer.Trace trace = new
OpenTelemetryAgentMcpCallTracer(telemetry).start(Context.of("private-subject",
"secret-value"));
+ trace.finish(Outcome.SUCCESS);
+ trace.finish(Outcome.SERVER_ERROR);
+ assertTrue(provider.forceFlush().join(5,
TimeUnit.SECONDS).isSuccess());
+ assertEquals(1, requests.get());
+ assertEquals("POST", method.get());
+ assertEquals("application/x-protobuf", contentType.get());
+ String wire = new String(payload.get(),
StandardCharsets.UTF_8);
+ assertTrue(wire.contains("shenyu.agent_gateway.tools.call"));
+ assertTrue(wire.contains("shenyu.mcp.outcome"));
+ assertTrue(wire.contains("success"));
+ assertFalse(wire.contains("private-subject"));
+ assertFalse(wire.contains("secret-value"));
+ }
+ } finally {
+ receiver.stop(0);
+ }
+ }
+}
