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 5961490ff5 refactor(agent-gateway): clarify MCP call metrics
observation (#7538)
5961490ff5 is described below
commit 5961490ff5930139e92c4fe84f0f7f1f011aecf7
Author: BobSong <[email protected]>
AuthorDate: Sun Oct 11 16:45:23 2026 +0800
refactor(agent-gateway): clarify MCP call metrics observation (#7538)
Co-authored-by: BobSong-dev <[email protected]>
---
.../gateway/protocol/AgentMcpCallObservation.java | 4 +-
.../gateway/protocol/AgentMcpHttpHandler.java | 11 +++--
.../protocol/AgentMcpCallObservationTest.java | 51 ++++++++++++++++++++--
.../shenyu/plugin/metrics/MetricsPlugin.java | 1 +
.../metrics/reporter/MetricsReporterTest.java | 19 +++++---
5 files changed, 73 insertions(+), 13 deletions(-)
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 2e47c4354d..ab997a904f 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
@@ -46,8 +46,8 @@ final class AgentMcpCallObservation {
private volatile Outcome outcome = Outcome.SERVER_ERROR;
- AgentMcpCallObservation(final Object observer) {
- this.observer = observer instanceof AgentMcpCallObserver ?
(AgentMcpCallObserver) observer : null;
+ AgentMcpCallObservation(final AgentMcpCallObserver observer) {
+ this.observer = observer;
}
void parsed(final AgentMcpRequest request) {
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 b0edc084dc..354d1765de 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
@@ -22,6 +22,7 @@ import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
import org.apache.shenyu.common.constant.Constants;
import org.apache.shenyu.common.dto.AgentGatewayMcpConfig;
+import org.apache.shenyu.common.metrics.AgentMcpCallObserver;
import org.apache.shenyu.plugin.agent.gateway.AgentTrafficContext;
import
org.apache.shenyu.plugin.agent.gateway.security.AgentMcpSecurityResolver;
import org.springframework.core.io.buffer.DataBuffer;
@@ -80,7 +81,8 @@ public final class AgentMcpHttpHandler {
final AgentTrafficContext traffic, final long
configurationVersion) {
return Mono.defer(() -> {
final Instant deadline =
Instant.now().plusMillis(config.getTimeoutMs());
- final AgentMcpCallObservation observation = new
AgentMcpCallObservation(exchange.getAttribute(Constants.METRICS_AGENT_MCP_CALL));
+ Object callback =
exchange.getAttribute(Constants.METRICS_AGENT_MCP_CALL);
+ final AgentMcpCallObservation observation = new
AgentMcpCallObservation(callback instanceof AgentMcpCallObserver observer ?
observer : null);
AtomicReference<JsonNode> rpcId = new AtomicReference<>();
return Mono.fromRunnable(() -> preflight(exchange, config))
.then(Mono.defer(() -> securityResolver.resolve(exchange))
@@ -97,8 +99,11 @@ public final class AgentMcpHttpHandler {
return write(exchange, response, config,
"sse".equals(config.getResponseMode()), 200);
})
.timeout(Duration.ofMillis(config.getTimeoutMs()))
- .doOnError(observation::failure)
- .onErrorResume(error -> handleFailure(exchange, config,
rpcId.get(), error))
+ .onErrorResume(error -> {
+ observation.failure(error);
+ return handleFailure(exchange, config, rpcId.get(),
error);
+ })
+ // Recovery writes can fail independently of the execution
error classified above.
.doOnError(observation::failure)
.doFinally(observation::finish);
});
diff --git
a/shenyu-plugin/shenyu-plugin-agent-gateway/src/test/java/org/apache/shenyu/plugin/agent/gateway/protocol/AgentMcpCallObservationTest.java
b/shenyu-plugin/shenyu-plugin-agent-gateway/src/test/java/org/apache/shenyu/plugin/agent/gateway/protocol/AgentMcpCallObservationTest.java
index dbde556e0f..ccad4dad40 100644
---
a/shenyu-plugin/shenyu-plugin-agent-gateway/src/test/java/org/apache/shenyu/plugin/agent/gateway/protocol/AgentMcpCallObservationTest.java
+++
b/shenyu-plugin/shenyu-plugin-agent-gateway/src/test/java/org/apache/shenyu/plugin/agent/gateway/protocol/AgentMcpCallObservationTest.java
@@ -158,6 +158,48 @@ class AgentMcpCallObservationTest {
observer.assertOnly(Outcome.SERVER_ERROR);
}
+ @ParameterizedTest
+ @ValueSource(strings = {"json", "sse"})
+ void errorResponseWriteFailureOverridesOriginalRejection(final String
mode) {
+ Recording observer = new Recording();
+ AtomicInteger invocations = new AtomicInteger();
+ MockServerWebExchange exchange = exchange("tools/call", observer);
+ exchange.getResponse().setWriteHandler(buffers ->
buffers.doOnNext(DataBufferUtils::release)
+ .then(Mono.error(new
IllegalStateException("private-error-response-write"))));
+ AgentMcpHttpHandler handler = handler(tool(Mono.defer(() -> {
+ invocations.incrementAndGet();
+ return Mono.just(new JsonObject());
+ })), Set.of());
+ StepVerifier.create(handler.handle(exchange, config(mode, 30000),
traffic, 1))
+ .expectError(IllegalStateException.class).verify();
+ observer.assertOnly(Outcome.SERVER_ERROR);
+ assertEquals(0, invocations.get());
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"json", "sse"})
+ void errorResponseWriteTimeoutOverridesOriginalRejection(final String
mode) {
+ Recording observer = new Recording();
+ MockServerWebExchange exchange = exchange("tools/call", observer);
+ exchange.getResponse().setWriteHandler(buffers ->
buffers.doOnNext(DataBufferUtils::release).then(Mono.never()));
+ AgentMcpHttpHandler handler = handler(tool(Mono.just(new
JsonObject())), Set.of());
+ StepVerifier.withVirtualTime(() -> handler.handle(exchange,
config(mode, 100), traffic, 1))
+ .then(() -> assertTrue(observer.outcomes.isEmpty()))
+
.thenAwait(Duration.ofMillis(101)).expectError(TimeoutException.class).verify();
+ observer.assertOnly(Outcome.TIMEOUT);
+ }
+
+ @ParameterizedTest
+ @ValueSource(strings = {"json", "sse"})
+ void cancellingErrorResponseWriteOverridesOriginalRejection(final String
mode) {
+ Recording observer = new Recording();
+ MockServerWebExchange exchange = exchange("tools/call", observer);
+ exchange.getResponse().setWriteHandler(buffers ->
buffers.doOnNext(DataBufferUtils::release).then(Mono.never()));
+ StepVerifier.create(handler(tool(Mono.just(new JsonObject())),
Set.of()).handle(exchange, config(mode, 30000), traffic, 1))
+ .then(() ->
assertTrue(observer.outcomes.isEmpty())).thenCancel().verify();
+ observer.assertOnly(Outcome.CANCELLED);
+ }
+
@ParameterizedTest
@ValueSource(strings = {"json", "sse"})
void recordsResponseLimitFailureWithoutLeakingSuccess(final String mode) {
@@ -274,10 +316,13 @@ class AgentMcpCallObservationTest {
observer.assertOnly(Outcome.CANCELLED);
}
- @Test
- void absentOrWrongTypeCallbackDoesNotRequireMetricsPlugin() {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ void absentOrWrongTypeCallbackDoesNotRequireMetricsPlugin(final boolean
wrongType) {
MockServerWebExchange exchange = exchange("tools/call", null);
- exchange.getAttributes().put(Constants.METRICS_AGENT_MCP_CALL,
"unusable-callback");
+ if (wrongType) {
+ exchange.getAttributes().put(Constants.METRICS_AGENT_MCP_CALL,
"unusable-callback");
+ }
StepVerifier.create(handler(tool(Mono.just(new JsonObject())),
Set.of("read"))
.handle(exchange, config("json", 30000), traffic,
1)).verifyComplete();
assertEquals(200, exchange.getResponse().getStatusCode().value());
diff --git
a/shenyu-plugin/shenyu-plugin-metrics/src/main/java/org/apache/shenyu/plugin/metrics/MetricsPlugin.java
b/shenyu-plugin/shenyu-plugin-metrics/src/main/java/org/apache/shenyu/plugin/metrics/MetricsPlugin.java
index a6527250a6..e20f28c951 100644
---
a/shenyu-plugin/shenyu-plugin-metrics/src/main/java/org/apache/shenyu/plugin/metrics/MetricsPlugin.java
+++
b/shenyu-plugin/shenyu-plugin-metrics/src/main/java/org/apache/shenyu/plugin/metrics/MetricsPlugin.java
@@ -58,6 +58,7 @@ public class MetricsPlugin implements ShenyuPlugin {
}
private void setMetricsCallbacks(final ServerWebExchange exchange) {
+ // Metrics must run before Agent Gateway so its per-subscription
observation can capture this callback.
exchange.getAttributes().put(Constants.METRICS_AGENT_MCP_CALL,
(AgentMcpCallObserver) (outcome, millis) -> {
String[] labels = {outcome.name().toLowerCase(Locale.ROOT)};
MetricsReporter.counterIncrement(LabelNames.AGENT_MCP_CALLS_TOTAL,
labels);
diff --git
a/shenyu-plugin/shenyu-plugin-metrics/src/test/java/org/apache/shenyu/plugin/metrics/reporter/MetricsReporterTest.java
b/shenyu-plugin/shenyu-plugin-metrics/src/test/java/org/apache/shenyu/plugin/metrics/reporter/MetricsReporterTest.java
index 023635a2d9..057aaa77d0 100644
---
a/shenyu-plugin/shenyu-plugin-metrics/src/test/java/org/apache/shenyu/plugin/metrics/reporter/MetricsReporterTest.java
+++
b/shenyu-plugin/shenyu-plugin-metrics/src/test/java/org/apache/shenyu/plugin/metrics/reporter/MetricsReporterTest.java
@@ -22,6 +22,7 @@ import io.prometheus.client.Gauge;
import io.prometheus.client.Histogram;
import org.apache.shenyu.common.utils.ReflectUtils;
import org.apache.shenyu.plugin.metrics.config.Metric;
+import org.apache.shenyu.plugin.metrics.constant.LabelNames;
import org.apache.shenyu.plugin.metrics.enums.MetricType;
import org.apache.shenyu.plugin.metrics.prometheus.PrometheusMetricsRegister;
import org.apache.shenyu.plugin.metrics.spi.MetricsRegister;
@@ -58,9 +59,15 @@ public final class MetricsReporterTest {
@Test
public void testRegister() throws Exception {
Map<String, Counter> map1 = getPrivateField(metricsRegister,
"COUNTER_MAP", Map.class);
- Assertions.assertEquals(map1.size(), 10);
+ for (String name : List.of(LabelNames.REQUEST_TOTAL,
LabelNames.REQUEST_TYPE_TOTAL, LabelNames.REQUEST_THROW_TOTAL,
+ LabelNames.AGENT_MCP_CALLS_TOTAL,
LabelNames.SENTINEL_REQUEST_RESTRICT_TOTAL,
LabelNames.SENTINEL_REQUEST_CIRCUITBREAKER_TOTAL,
+ LabelNames.RESILIENCE4J_REQUEST_RESTRICT_TOTAL,
LabelNames.RESILIENCE4J_REQUEST_CIRCUITBREAKER_TOTAL,
+ LabelNames.HYSTRIX_REQUEST_CIRCUITBREAKER_TOTAL,
LabelNames.RATELIMITER_REQUEST_RESTRICT_TOTAL)) {
+ Assertions.assertNotNull(map1.get(name), name);
+ }
Map<String, Histogram> map2 = getPrivateField(metricsRegister,
"HISTOGRAM_MAP", Map.class);
- Assertions.assertEquals(map2.size(), 4);
+ Assertions.assertNotNull(map2.get(LabelNames.EXECUTE_LATENCY_NAME));
+ Assertions.assertNotNull(map2.get(LabelNames.AGENT_MCP_CALL_LATENCY));
List<String> labels = new ArrayList<>();
labels.add("shenyu_request_total");
Collection<Metric> metrics = new ArrayList<>();
@@ -69,11 +76,13 @@ public final class MetricsReporterTest {
metrics.add(new Metric(MetricType.HISTOGRAM, "name3", DOCUMENT,
labels));
MetricsReporter.registerMetrics(metrics);
Map<String, Counter> map3 = getPrivateField(metricsRegister,
"COUNTER_MAP", Map.class);
- Assertions.assertEquals(map3.size(), 11);
+ Assertions.assertNotNull(map3.get("name1"));
+ Assertions.assertFalse(map3.containsKey("name2"));
Map<String, Histogram> map4 = getPrivateField(metricsRegister,
"HISTOGRAM_MAP", Map.class);
- Assertions.assertEquals(map4.size(), 5);
+ Assertions.assertNotNull(map4.get("name3"));
+ Assertions.assertFalse(map4.containsKey("name2"));
Map<String, Gauge> map5 = getPrivateField(metricsRegister,
"GAUGE_MAP", Map.class);
- Assertions.assertEquals(map5.size(), 3);
+ Assertions.assertNotNull(map5.get("name2"));
}
@Test
