diff --git a/openframe-gateway-service-core/src/main/java/com/openframe/gateway/config/ws/ProxySessionCleanupWebSocketClient.java b/openframe-gateway-service-core/src/main/java/com/openframe/gateway/config/ws/ProxySessionCleanupWebSocketClient.java index 80cfd8924..247346ed8 100644 --- a/openframe-gateway-service-core/src/main/java/com/openframe/gateway/config/ws/ProxySessionCleanupWebSocketClient.java +++ b/openframe-gateway-service-core/src/main/java/com/openframe/gateway/config/ws/ProxySessionCleanupWebSocketClient.java @@ -25,12 +25,22 @@ public class ProxySessionCleanupWebSocketClient implements WebSocketClient { @Override public Mono execute(URI url, WebSocketHandler handler) { - return delegate.execute(url, wrapHandler(url, handler)); + return instrumentUpstream(url, delegate.execute(url, wrapHandler(url, handler))); } @Override public Mono execute(URI url, HttpHeaders headers, WebSocketHandler handler) { - return delegate.execute(url, headers, wrapHandler(url, handler)); + return instrumentUpstream(url, delegate.execute(url, headers, wrapHandler(url, handler))); + } + + // Log the upstream proxy outcome: a connect failure/timeout to a dead upstream (e.g. meshcentral or NATS) is otherwise invisible here — the inbound upgrade just never completes and the agent sees "No HTTP response". + private Mono instrumentUpstream(URI url, Mono proxy) { + boolean debugPath = loggingProperties.isDebugPath(url.getPath()); + // Host+path only: the upstream URI can carry forwarded query params (e.g. an agent key), which must not be logged. + String safeUrl = url.getScheme() + "://" + url.getHost() + (url.getPort() < 0 ? "" : ":" + url.getPort()) + url.getPath(); + return proxy + .doOnSubscribe(s -> { if (debugPath) { log.debug("ws proxy: connecting to upstream {}", safeUrl); } }) + .doOnError(ex -> log.warn("ws proxy: upstream connect/relay failed for {}: {}", safeUrl, ex.toString())); } private WebSocketHandler wrapHandler(URI targetUrl, WebSocketHandler handler) { diff --git a/openframe-gateway-service-core/src/main/java/com/openframe/gateway/config/ws/ToolWebSocketProxyUrlFilter.java b/openframe-gateway-service-core/src/main/java/com/openframe/gateway/config/ws/ToolWebSocketProxyUrlFilter.java index 19bf7d1a5..4454bce37 100644 --- a/openframe-gateway-service-core/src/main/java/com/openframe/gateway/config/ws/ToolWebSocketProxyUrlFilter.java +++ b/openframe-gateway-service-core/src/main/java/com/openframe/gateway/config/ws/ToolWebSocketProxyUrlFilter.java @@ -1,5 +1,7 @@ package com.openframe.gateway.config.ws; +import com.github.benmanes.caffeine.cache.Cache; +import com.github.benmanes.caffeine.cache.Caffeine; import com.openframe.data.document.tool.IntegratedTool; import com.openframe.data.reactive.repository.tool.ReactiveIntegratedToolRepository; import com.openframe.data.service.TenantIdProvider; @@ -15,12 +17,15 @@ import org.springframework.core.Ordered; import org.springframework.http.HttpHeaders; import org.springframework.http.HttpStatus; +import org.springframework.http.HttpStatusCode; import org.springframework.http.server.reactive.ServerHttpRequest; +import org.springframework.http.server.reactive.ServerHttpResponse; import org.springframework.web.server.ResponseStatusException; import org.springframework.web.server.ServerWebExchange; import reactor.core.publisher.Mono; import java.net.URI; +import java.time.Duration; @RequiredArgsConstructor @Slf4j @@ -28,9 +33,18 @@ public abstract class ToolWebSocketProxyUrlFilter implements GatewayFilter, Orde public static final String ORIGINAL_AUTHORIZATION_ATTR = "originalAuthorization"; + // Bounds the per-upgrade tool lookup so a stalled Mongo query fails fast instead of letting the WS upgrade hang to the gateway response-timeout. + private static final Duration LOOKUP_TIMEOUT = Duration.ofSeconds(5); + private final ReactiveIntegratedToolRepository toolRepository; private final ToolUpstreamResolverRegistry upstreamRegistry; + // Cache resolved tools briefly so an agent reconnect storm doesn't put a Mongo lookup on every WS upgrade (configs change rarely). + private final Cache toolCache = Caffeine.newBuilder() + .maximumSize(10_000) + .expireAfterWrite(Duration.ofSeconds(60)) + .build(); + /** * Shared multi-tenant routing mode. When true, a tool lookup with no resolved tenant must NOT * fall back to an unscoped query. Defaults false so single-tenant / OSS pods keep prior behavior. @@ -68,7 +82,17 @@ public Mono filter(ServerWebExchange exchange, GatewayFilterChain chain) { ServerWebExchange mutatedExchange = mutateExchange(exchange, tool); return chain.filter(mutatedExchange); - }); + }) + // Without this, a lookup error/timeout left the WS upgrade hanging with no response written, so the agent saw "No HTTP response" and retried forever; fail fast with a real status instead. + .onErrorResume(ex -> abortUpgrade(exchange, toolId, ex)); + } + + private Mono abortUpgrade(ServerWebExchange exchange, String toolId, Throwable ex) { + HttpStatusCode status = (ex instanceof ResponseStatusException rse) ? rse.getStatusCode() : HttpStatus.SERVICE_UNAVAILABLE; + log.warn("Tool websocket upgrade aborted (tool={}, status={}): {}", toolId, status, ex.toString()); + ServerHttpResponse response = exchange.getResponse(); + response.setStatusCode(status); + return response.setComplete(); } protected ServerWebExchange mutateExchange(ServerWebExchange exchange, IntegratedTool tool) { @@ -82,6 +106,14 @@ protected ServerWebExchange mutateExchange(ServerWebExchange exchange, Integrate * headers are never read; the unscoped {@code findByKey} is correct (one tenant only). */ private Mono getTool(String toolId, ServerHttpRequest request) { + String cacheKey = tenantRoutingEnabled + ? (TenantRoutingHeaders.tenantId(request) + "::" + toolId) + : toolId; + IntegratedTool cached = toolCache.getIfPresent(cacheKey); + if (cached != null) { + return Mono.just(cached); + } + Mono lookup = tenantRoutingEnabled ? toolRepository.findByTenantIdAndKey(TenantRoutingHeaders.tenantId(request), toolId) : toolRepository.findByKey(toolId); @@ -94,7 +126,10 @@ private Mono getTool(String toolId, ServerHttpRequest request) { return Mono.error(new IllegalArgumentException("Tool " + tool.getName() + " is not enabled")); } return Mono.just(tool); - }); + }) + .timeout(LOOKUP_TIMEOUT) + // Only successful (present + enabled) tools are cached; not-found / not-enabled stay uncached so they recover immediately once fixed. + .doOnNext(tool -> toolCache.put(cacheKey, tool)); } protected abstract String getRequestToolId(String path);