diff --git a/docs/TELEMETRY.md b/docs/TELEMETRY.md index 1dfcda77a..161dc9749 100644 --- a/docs/TELEMETRY.md +++ b/docs/TELEMETRY.md @@ -266,6 +266,69 @@ model/provider names, attachment metadata, filenames, tool names, token usage, URLs, error messages, or local chat log text. No chat log category is added to the OpenTelemetry log allowlist. +## Local MCP server lifecycle + +The local MCP HTTP server exports transport lifecycle diagnostics for both +MCP-only mode and gateway-enabled local MCP: + +- lifecycle traces: `openclaw.mcp.server.start` and + `openclaw.mcp.server.stop` +- request trace: `openclaw.mcp.server.request` +- lifecycle counter: `openclaw.mcp.server.lifecycle.operations` +- request counter: `openclaw.mcp.server.requests` +- listener-error counter: `openclaw.mcp.server.listener.errors` +- request-duration histogram: `openclaw.mcp.server.request.duration` + +Lifecycle operations are emitted once per real start or stop attempt. Concurrent +or repeated successful starts do not create duplicate operations. Repeated stop +and disposal calls share the first stop operation. The stop span covers listener +stop and in-flight handler drain. Listener close happens later during resource +disposal, so a close failure increments the listener-error counter but does not +retroactively change the completed stop result. + +Request duration is post-accept handling time. It starts after +`HttpListener.GetContextAsync` returns, includes handler-limiter admission, body +read, bridge dispatch, and response delivery, and ends at successful delivery or +rejection. It does not include time spent in the HTTP.sys backlog before a +context is accepted. + +The MCP request span is an independent transport-level root. A `tools/call` +request also creates the existing independent `openclaw.node.tool.invoke` root, +which measures node dispatch through response delivery. The overlap is +intentional: the MCP request signal diagnoses local HTTP transport behavior, +while node-tool telemetry diagnoses command execution. A successfully delivered +JSON-RPC error envelope is a successful MCP transport request; response text is +never parsed to classify telemetry. + +Reviewed MCP attributes are: + +- `openclaw.mcp.server.operation`: `start` or `stop` +- `openclaw.mcp.server.request.kind`: `probe`, `json_rpc`, or `other` +- `openclaw.outcome`: `success`, `failure`, or `canceled` +- `openclaw.error.category`: `none`, `listener_start`, `listener_accept`, + `listener_stop`, `listener_close`, `authentication_failed`, `busy`, `timeout`, + `shutdown`, `drain_timeout`, `invalid_request`, `transport_failure`, or + `internal_failure` +- `error.type`: exception type on spans only + +Metrics always include the finite error category, including `none` on success, +and never include exception types. Authentication rejection, handler saturation, +invalid HTTP requests, request deadlines, shutdown cancellation, response +delivery failures, and unexpected internal failures are classified through typed +control flow. Deadline and shutdown cancellation use first-wins attribution, so +a later shutdown cannot turn a timeout into cancellation and a later deadline +cannot turn shutdown into timeout. Each source records its cause before +canceling the request token, so request handling cannot observe cancellation +before attribution. If multiple failures occur during stop, the first failure +owns the lifecycle result; later listener failures remain visible through the +listener-error counter. + +MCP server telemetry never exports listener ports or endpoints, local or remote +addresses, HTTP scheme or version, headers, user agents, content lengths, bearer +tokens, request or response bodies, JSON-RPC methods or IDs, tool names, command +arguments, or error and exception messages. Existing detailed MCP logs remain +local. No MCP category is added to the OpenTelemetry log allowlist. + ## Windows node tool calls Gateway `node.invoke` and local MCP `tools/call` share one node-side telemetry diff --git a/src/OpenClaw.Shared/Mcp/McpHttpServer.cs b/src/OpenClaw.Shared/Mcp/McpHttpServer.cs index 3b9ec2fca..9f22701aa 100644 --- a/src/OpenClaw.Shared/Mcp/McpHttpServer.cs +++ b/src/OpenClaw.Shared/Mcp/McpHttpServer.cs @@ -6,6 +6,7 @@ using System.Text; using System.Threading; using System.Threading.Tasks; +using OpenClaw.Shared.Telemetry; namespace OpenClaw.Shared.Mcp; @@ -36,7 +37,7 @@ namespace OpenClaw.Shared.Mcp; /// but anything sandboxed away from %APPDATA%\OpenClawTray\ cannot. /// /// Stability defenses (CR-003/CR-005): -/// - Per-request hard deadline (RequestTimeoutMs) bounds body-read and +/// - Per-request hard deadline bounds body-read and /// bridge dispatch so a slow or hung client cannot pin a handler slot /// forever. /// - Active handler tasks are tracked so Stop/Dispose can drain in-flight @@ -45,26 +46,11 @@ namespace OpenClaw.Shared.Mcp; public sealed class McpHttpServer : IDisposable, IAsyncDisposable { private const long MaxRequestBodyBytes = 4L * 1024 * 1024; // 4 MiB - // 16 leaves headroom for parallel tool callers (e.g. an editor + Claude - // Desktop + a CLI script) without making each connection cheap enough to - // become a DoS lever — request size cap + per-handler timeout still bound - // memory. Bumped from 8 after queue-stall reports under multi-IDE use. - private const int MaxConcurrentHandlers = 16; - // Sized to cover the longest legitimate capability: screen.record up to - // 300s plus encoding + serialization. Earlier 90s deadline silently abort - // ed valid recording requests while the OS capture pipeline kept running - // unobserved (unified review H10). Cancellation now flows through the - // capability via INodeCapability.ExecuteAsync(NodeInvokeRequest, CT) so - // tools that opt in actually stop the underlying work. - private const int RequestTimeoutMs = 360_000; - // How long Dispose waits for in-flight handlers to drain before forcing - // tear-down. Past this point handlers may observe disposed services. - private static readonly TimeSpan DrainTimeout = TimeSpan.FromSeconds(5); - private readonly McpToolBridge _bridge; private readonly int _port; private readonly IOpenClawLogger _logger; - private readonly HttpListener _listener; + private readonly IMcpHttpListener _listener; + private readonly McpHttpServerOptions _options; private readonly Action _writeText; /// /// Required bearer token for HTTP requests. Empty/null disables auth (the @@ -73,9 +59,10 @@ public sealed class McpHttpServer : IDisposable, IAsyncDisposable /// private string? _authToken; private readonly CancellationTokenSource _cts = new(); - private readonly SemaphoreSlim _handlerLimiter = new(MaxConcurrentHandlers, MaxConcurrentHandlers); + private readonly SemaphoreSlim _handlerLimiter; private readonly object _activeLock = new(); private readonly HashSet _activeHandlers = new(); + private readonly object _startLock = new(); private readonly object _shutdownLock = new(); private Task? _acceptLoop; private Task? _stopTask; @@ -87,7 +74,14 @@ public sealed class McpHttpServer : IDisposable, IAsyncDisposable public string Endpoint => $"http://127.0.0.1:{_port}/"; public McpHttpServer(McpToolBridge bridge, int port, IOpenClawLogger logger, string? authToken = null) - : this(bridge, port, logger, authToken, WriteText) + : this( + bridge, + port, + logger, + authToken, + WriteText, + McpHttpServerOptions.Default, + new SystemMcpHttpListener(port)) { } @@ -97,25 +91,119 @@ internal McpHttpServer( IOpenClawLogger logger, string? authToken, Action writeText) + : this( + bridge, + port, + logger, + authToken, + writeText, + McpHttpServerOptions.Default, + new SystemMcpHttpListener(port)) + { + } + + internal McpHttpServer( + McpToolBridge bridge, + int port, + IOpenClawLogger logger, + string? authToken, + McpHttpServerOptions options) + : this( + bridge, + port, + logger, + authToken, + WriteText, + options, + new SystemMcpHttpListener(port)) + { + } + + internal McpHttpServer( + McpToolBridge bridge, + int port, + IOpenClawLogger logger, + string? authToken, + McpHttpServerOptions options, + IMcpHttpListener listener) + : this( + bridge, + port, + logger, + authToken, + WriteText, + options, + listener) + { + } + + internal McpHttpServer( + McpToolBridge bridge, + int port, + IOpenClawLogger logger, + string? authToken, + Action writeText, + McpHttpServerOptions options, + IMcpHttpListener? listener = null) { _bridge = bridge ?? throw new ArgumentNullException(nameof(bridge)); _logger = logger ?? throw new ArgumentNullException(nameof(logger)); _writeText = writeText ?? throw new ArgumentNullException(nameof(writeText)); + _options = options ?? throw new ArgumentNullException(nameof(options)); _port = port; _authToken = string.IsNullOrEmpty(authToken) ? null : authToken; - _listener = new HttpListener(); + _listener = listener ?? new SystemMcpHttpListener(port); + _handlerLimiter = new SemaphoreSlim( + _options.MaxConcurrentHandlers, + _options.MaxConcurrentHandlers); // Loopback binding — not reachable from other machines. Use only the // numeric host on Windows so non-elevated startup does not require a // separate netsh http urlacl reservation for http://localhost:port/. - _listener.Prefixes.Add($"http://127.0.0.1:{port}/"); } public void Start() { - if (_listener.IsListening) return; - _listener.Start(); - _acceptLoop = Task.Run(() => AcceptLoopAsync(_cts.Token)); - _logger.Info($"[MCP] HTTP server listening on {Endpoint}"); + lock (_startLock) + { + if (_listener.IsListening) + return; + + using var telemetry = McpServerTelemetry.StartLifecycle(McpServerOperation.Start); + try + { + _listener.Start(); + } + catch (Exception ex) + { + McpServerTelemetry.RecordListenerError(McpServerErrorCategory.ListenerStart); + telemetry.Complete( + McpServerOutcome.Failure, + McpServerErrorCategory.ListenerStart, + ex.GetType()); + throw; + } + + try + { + _acceptLoop = Task.Run(() => AcceptLoopAsync(_cts.Token)); + _logger.Info($"[MCP] HTTP server listening on {Endpoint}"); + telemetry.Complete(McpServerOutcome.Success); + } + catch (Exception ex) + { + telemetry.Complete( + McpServerOutcome.Failure, + McpServerErrorCategory.InternalFailure, + ex.GetType()); + try { _listener.Stop(); } + catch (Exception stopEx) + { + McpServerTelemetry.RecordListenerError(McpServerErrorCategory.ListenerStop); + _logger.Debug($"[MCP] Listener cleanup after start failure threw: {stopEx.Message}"); + } + throw; + } + } } public void UpdateAuthToken(string? authToken) @@ -143,49 +231,172 @@ private async Task AcceptLoopAsync(CancellationToken ct) catch (Exception ex) { if (ct.IsCancellationRequested) break; + McpServerTelemetry.RecordListenerError(McpServerErrorCategory.ListenerAccept); _logger.Error("[MCP] Accept failed", ex); continue; } - // Defensive: even though the prefix is loopback-only, double-check. - if (!IPAddress.IsLoopback(ctx.Request.RemoteEndPoint.Address)) + McpServerRequestOperation? requestTelemetry = McpServerTelemetry.StartRequest( + GetRequestKind(ctx.Request.HttpMethod)); + var handlerSlotAcquired = false; + try { - Reject(ctx, HttpStatusCode.Forbidden, "loopback only"); - continue; - } + // Defensive: even though the prefix is loopback-only, double-check. + if (!IPAddress.IsLoopback(ctx.Request.RemoteEndPoint.Address)) + { + CompleteRejection( + requestTelemetry, + ctx, + HttpStatusCode.Forbidden, + "loopback only", + McpServerOutcome.Failure, + McpServerErrorCategory.InvalidRequest); + continue; + } + + // Cap concurrent handlers — a misbehaving local client can otherwise + // pin every threadpool thread on long-running screen/camera calls. + // Wait briefly so transient handoff spikes can succeed without + // introducing unbounded queueing. + _options.HandlerAdmissionStarted?.Invoke(); + if (!await _handlerLimiter + .WaitAsync(_options.HandlerAdmissionTimeout, ct) + .ConfigureAwait(false)) + { + CompleteRejection( + requestTelemetry, + ctx, + HttpStatusCode.ServiceUnavailable, + "server busy", + McpServerOutcome.Failure, + McpServerErrorCategory.Busy); + continue; + } + handlerSlotAcquired = true; - // Cap concurrent handlers — a misbehaving local client can otherwise - // pin every threadpool thread on long-running screen/camera calls. - // Wait briefly: a slot freed during typical request handoff is well - // under 50ms, so a small queue here turns transient spikes into - // success rather than 503s without inviting unbounded queueing. - if (!await _handlerLimiter.WaitAsync(50, ct).ConfigureAwait(false)) + // NOTE: do not pass `ct` to Task.Run. If the token is cancelled + // between WaitAsync returning and the delegate starting, Task.Run + // skips the delegate and the finally never runs — leaking a + // semaphore slot. Let the delegate observe cancellation itself. + var handlerTelemetry = requestTelemetry; + var handlerTask = Task.Run(() => RunHandlerAsync(ctx, handlerTelemetry)); + TrackHandler(handlerTask); + requestTelemetry = null; + handlerSlotAcquired = false; + } + catch (OperationCanceledException) when (ct.IsCancellationRequested) { - Reject(ctx, (HttpStatusCode)503, "server busy"); - continue; + requestTelemetry?.Complete( + McpServerOutcome.Canceled, + McpServerErrorCategory.Shutdown); + break; + } + catch (Exception ex) + { + requestTelemetry?.Complete( + McpServerOutcome.Failure, + McpServerErrorCategory.InternalFailure, + ex.GetType()); + _logger.Error("[MCP] Failed to admit request", ex); + Reject(ctx, HttpStatusCode.InternalServerError, "internal error"); + } + finally + { + if (handlerSlotAcquired) + { + try { _handlerLimiter.Release(); } + catch (ObjectDisposedException) { } + } + requestTelemetry?.Dispose(); } - - // NOTE: do not pass `ct` to Task.Run. If the token is cancelled - // between WaitAsync returning and the delegate starting, Task.Run - // skips the delegate and the finally never runs — leaking a - // semaphore slot. Let the delegate observe cancellation itself. - var handlerTask = Task.Run(() => RunHandlerAsync(ctx)); - TrackHandler(handlerTask); } } - private async Task RunHandlerAsync(HttpListenerContext ctx) + private async Task RunHandlerAsync( + HttpListenerContext ctx, + McpServerRequestOperation requestTelemetry) { - // Per-request linked CTS: server shutdown OR per-request deadline. - // The bridge call observes this so a wedged tool cannot pin the slot. - using var requestCts = CancellationTokenSource.CreateLinkedTokenSource(_cts.Token); - requestCts.CancelAfter(RequestTimeoutMs); + var cancellationState = new McpRequestCancellationState(); + CancellationTokenSource? timeoutCts = null; + CancellationTokenSource? requestCts = null; + CancellationTokenRegistration shutdownRegistration = default; + CancellationTokenRegistration timeoutRegistration = default; + CancellationToken shutdownCancellation = default; + CancellationToken timeoutCancellation = default; try { - await HandleAsync(ctx, requestCts.Token).ConfigureAwait(false); + _options.HandlerExecutionStarting?.Invoke(); + timeoutCts = new CancellationTokenSource(); + shutdownCancellation = _cts.Token; + timeoutCancellation = timeoutCts.Token; + requestCts = new CancellationTokenSource(); + var requestCancellation = requestCts; + shutdownRegistration = shutdownCancellation.Register( + static state => + { + var propagation = (McpRequestCancellationPropagation)state!; + PropagateCancellation( + propagation.State, + propagation.RequestCancellation, + propagation.Cause); + }, + new McpRequestCancellationPropagation( + cancellationState, + requestCancellation, + McpRequestCancellationCause.Shutdown)); + timeoutRegistration = timeoutCancellation.Register( + static state => + { + var propagation = (McpRequestCancellationPropagation)state!; + PropagateCancellation( + propagation.State, + propagation.RequestCancellation, + propagation.Cause); + }, + new McpRequestCancellationPropagation( + cancellationState, + requestCancellation, + McpRequestCancellationCause.Timeout)); + timeoutCts.CancelAfter(_options.RequestTimeout); + + var result = await HandleAsync( + ctx, + requestCts.Token, + cancellationState, + timeoutCancellation, + shutdownCancellation) + .ConfigureAwait(false); + requestTelemetry.Complete(result.Outcome, result.ErrorCategory, result.ErrorType); + } + catch (OperationCanceledException) when (requestCts?.IsCancellationRequested == true) + { + var result = GetCancellationResult( + cancellationState, + timeoutCancellation, + shutdownCancellation); + requestTelemetry.Complete(result.Outcome, result.ErrorCategory); + } + catch (ObjectDisposedException) when (Volatile.Read(ref _disposed) != 0) + { + requestTelemetry.Complete( + McpServerOutcome.Canceled, + McpServerErrorCategory.Shutdown); + } + catch (Exception ex) + { + requestTelemetry.Complete( + McpServerOutcome.Failure, + McpServerErrorCategory.InternalFailure, + ex.GetType()); + _logger.Error("[MCP] Request handler failed outside transport handling", ex); } finally { + timeoutRegistration.Dispose(); + shutdownRegistration.Dispose(); + requestCts?.Dispose(); + timeoutCts?.Dispose(); + requestTelemetry.Dispose(); // Defensive: if Dispose has already disposed the limiter, swallow. // Without this guard, a handler racing with shutdown can throw // ObjectDisposedException into an unobserved task, which surfaces @@ -211,7 +422,12 @@ private void TrackHandler(Task task) }, CancellationToken.None, TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default); } - private async Task HandleAsync(HttpListenerContext ctx, CancellationToken ct) + private async Task HandleAsync( + HttpListenerContext ctx, + CancellationToken ct, + McpRequestCancellationState cancellationState, + CancellationToken timeoutCancellation, + CancellationToken shutdownCancellation) { // Snapshot the auth token once. UpdateAuthToken can rotate _authToken // on another thread, and reading the field separately for the null-test @@ -228,8 +444,12 @@ private async Task HandleAsync(HttpListenerContext ctx, CancellationToken ct) var origin = ctx.Request.Headers["Origin"]; if (!string.IsNullOrEmpty(origin)) { - Reject(ctx, HttpStatusCode.Forbidden, "origin not allowed"); - return; + return RejectRequest( + ctx, + HttpStatusCode.Forbidden, + "origin not allowed", + McpServerOutcome.Failure, + McpServerErrorCategory.InvalidRequest); } // Belt-and-suspenders: a browser may strip Origin (e.g. via a // privacy extension) but still send Sec-Fetch-Site / Sec-Fetch-Mode @@ -239,16 +459,24 @@ private async Task HandleAsync(HttpListenerContext ctx, CancellationToken ct) !string.IsNullOrEmpty(ctx.Request.Headers["Sec-Fetch-Mode"]) || !string.IsNullOrEmpty(ctx.Request.Headers["Referer"])) { - Reject(ctx, HttpStatusCode.Forbidden, "browser context not allowed"); - return; + return RejectRequest( + ctx, + HttpStatusCode.Forbidden, + "browser context not allowed", + McpServerOutcome.Failure, + McpServerErrorCategory.InvalidRequest); } // Host header must match our loopback bind. Defends against DNS // rebinding pivots that route a public hostname to 127.0.0.1. if (!IsHostAllowed(ctx.Request.Headers["Host"])) { - Reject(ctx, HttpStatusCode.Forbidden, "host not allowed"); - return; + return RejectRequest( + ctx, + HttpStatusCode.Forbidden, + "host not allowed", + McpServerOutcome.Failure, + McpServerErrorCategory.InvalidRequest); } // Bearer-token check. Defends against untrusted local processes @@ -259,8 +487,12 @@ private async Task HandleAsync(HttpListenerContext ctx, CancellationToken ct) // so alternate verbs cannot bypass the configured token gate. if (authToken != null && !IsAuthorized(authToken, ctx.Request.Headers["Authorization"])) { - Reject(ctx, HttpStatusCode.Unauthorized, "missing or invalid bearer token"); - return; + return RejectRequest( + ctx, + HttpStatusCode.Unauthorized, + "missing or invalid bearer token", + McpServerOutcome.Failure, + McpServerErrorCategory.AuthenticationFailed); } if (ctx.Request.HttpMethod == "GET") @@ -269,13 +501,17 @@ private async Task HandleAsync(HttpListenerContext ctx, CancellationToken ct) // from a curl/browser without hitting the JSON-RPC endpoint. _writeText(ctx.Response, HttpStatusCode.OK, $"OpenClaw MCP server. POST JSON-RPC to {Endpoint}", "text/plain"); - return; + return McpRequestResult.Success; } if (ctx.Request.HttpMethod != "POST") { - Reject(ctx, HttpStatusCode.MethodNotAllowed, "POST only"); - return; + return RejectRequest( + ctx, + HttpStatusCode.MethodNotAllowed, + "POST only", + McpServerOutcome.Failure, + McpServerErrorCategory.InvalidRequest); } // Force application/json on POST. Combined with the Origin check, @@ -286,33 +522,52 @@ private async Task HandleAsync(HttpListenerContext ctx, CancellationToken ct) var contentTypeBase = (semi >= 0 ? contentType.Substring(0, semi) : contentType).Trim(); if (!string.Equals(contentTypeBase, "application/json", StringComparison.OrdinalIgnoreCase)) { - Reject(ctx, HttpStatusCode.UnsupportedMediaType, "application/json required"); - return; + return RejectRequest( + ctx, + HttpStatusCode.UnsupportedMediaType, + "application/json required", + McpServerOutcome.Failure, + McpServerErrorCategory.InvalidRequest); } // Reject bodies that exceed our cap *before* reading them — a // multi-GB POST would otherwise OOM the tray. if (ctx.Request.ContentLength64 > MaxRequestBodyBytes) { - Reject(ctx, HttpStatusCode.RequestEntityTooLarge, "request body too large"); - return; + return RejectRequest( + ctx, + HttpStatusCode.RequestEntityTooLarge, + "request body too large", + McpServerOutcome.Failure, + McpServerErrorCategory.InvalidRequest); } string body; try { - body = await ReadBodyAsync(ctx.Request, MaxRequestBodyBytes, ct).ConfigureAwait(false); + body = await (_options.RequestBodyReader ?? ReadBodyAsync)( + ctx.Request, + MaxRequestBodyBytes, + ct) + .ConfigureAwait(false); } catch (InvalidDataException) { - Reject(ctx, HttpStatusCode.RequestEntityTooLarge, "request body too large"); - return; + return RejectRequest( + ctx, + HttpStatusCode.RequestEntityTooLarge, + "request body too large", + McpServerOutcome.Failure, + McpServerErrorCategory.InvalidRequest); } catch (OperationCanceledException) when (ct.IsCancellationRequested) { // Slow-body or stuck client — free the slot rather than blocking forever. - Reject(ctx, HttpStatusCode.RequestTimeout, "request timed out"); - return; + return RejectCancellation( + ctx, + cancellationState, + timeoutCancellation, + shutdownCancellation); } try @@ -323,8 +578,11 @@ private async Task HandleAsync(HttpListenerContext ctx, CancellationToken ct) } catch (OperationCanceledException) when (ct.IsCancellationRequested) { - Reject(ctx, HttpStatusCode.RequestTimeout, "request timed out"); - return; + return RejectCancellation( + ctx, + cancellationState, + timeoutCancellation, + shutdownCancellation); } if (transportResponse.Body == null) @@ -333,22 +591,150 @@ private async Task HandleAsync(HttpListenerContext ctx, CancellationToken ct) ctx.Response.StatusCode = (int)HttpStatusCode.NoContent; ctx.Response.Close(); transportResponse.CompleteDelivery(); - return; + return McpRequestResult.Success; } _writeText(ctx.Response, HttpStatusCode.OK, transportResponse.Body, "application/json"); transportResponse.CompleteDelivery(); + return McpRequestResult.Success; } catch (Exception ex) { transportResponse?.CompleteDelivery(ex.GetType()); _logger.Error("[MCP] Request failed", ex); - // Response may already be partially written or closed; swallow. - try { Reject(ctx, HttpStatusCode.InternalServerError, "internal error"); } - catch (Exception rejEx) { _logger.Debug($"[MCP] Reject after handler error failed (response already disposed?): {rejEx.Message}"); } + Reject(ctx, HttpStatusCode.InternalServerError, "internal error"); + + if (ct.IsCancellationRequested) + { + return GetCancellationResult( + cancellationState, + timeoutCancellation, + shutdownCancellation, + ex.GetType()); + } + + return new McpRequestResult( + McpServerOutcome.Failure, + IsTransportException(ex) + ? McpServerErrorCategory.TransportFailure + : McpServerErrorCategory.InternalFailure, + ex.GetType()); } } + private McpRequestResult RejectCancellation( + HttpListenerContext ctx, + McpRequestCancellationState cancellationState, + CancellationToken timeoutCancellation, + CancellationToken shutdownCancellation) + { + var result = GetCancellationResult( + cancellationState, + timeoutCancellation, + shutdownCancellation); + var writeError = Reject(ctx, HttpStatusCode.RequestTimeout, "request timed out"); + return writeError == null + ? result + : new McpRequestResult( + McpServerOutcome.Failure, + McpServerErrorCategory.TransportFailure, + writeError); + } + + private McpRequestResult RejectRequest( + HttpListenerContext ctx, + HttpStatusCode status, + string reason, + McpServerOutcome outcome, + McpServerErrorCategory errorCategory) + { + var writeError = Reject(ctx, status, reason); + return writeError == null + ? new McpRequestResult(outcome, errorCategory) + : new McpRequestResult( + McpServerOutcome.Failure, + McpServerErrorCategory.TransportFailure, + writeError); + } + + private void CompleteRejection( + McpServerRequestOperation telemetry, + HttpListenerContext ctx, + HttpStatusCode status, + string reason, + McpServerOutcome outcome, + McpServerErrorCategory errorCategory) + { + var result = RejectRequest(ctx, status, reason, outcome, errorCategory); + telemetry.Complete(result.Outcome, result.ErrorCategory, result.ErrorType); + } + + private static McpServerRequestKind GetRequestKind(string method) => + method switch + { + "GET" => McpServerRequestKind.Probe, + "POST" => McpServerRequestKind.JsonRpc, + _ => McpServerRequestKind.Other + }; + + private static McpRequestResult GetCancellationResult( + McpRequestCancellationState cancellationState, + CancellationToken timeoutCancellation, + CancellationToken shutdownCancellation, + Type? errorType = null) => + ResolveCancellationCause( + cancellationState, + timeoutCancellation, + shutdownCancellation) switch + { + McpRequestCancellationCause.Shutdown => new McpRequestResult( + McpServerOutcome.Canceled, + McpServerErrorCategory.Shutdown, + errorType), + McpRequestCancellationCause.Timeout => new McpRequestResult( + McpServerOutcome.Failure, + McpServerErrorCategory.Timeout, + errorType), + _ => new McpRequestResult( + McpServerOutcome.Failure, + McpServerErrorCategory.InternalFailure, + errorType) + }; + + internal static McpRequestCancellationCause ResolveCancellationCause( + McpRequestCancellationState cancellationState, + CancellationToken timeoutCancellation, + CancellationToken shutdownCancellation) + { + var cause = cancellationState.Cause; + if (cause != McpRequestCancellationCause.None) + return cause; + + var timeoutRequested = timeoutCancellation.IsCancellationRequested; + var shutdownRequested = shutdownCancellation.IsCancellationRequested; + if (timeoutRequested == shutdownRequested) + return McpRequestCancellationCause.None; + + cancellationState.TrySet( + timeoutRequested + ? McpRequestCancellationCause.Timeout + : McpRequestCancellationCause.Shutdown); + + return cancellationState.Cause; + } + + internal static void PropagateCancellation( + McpRequestCancellationState cancellationState, + CancellationTokenSource requestCancellation, + McpRequestCancellationCause cause) + { + cancellationState.TrySet(cause); + requestCancellation.Cancel(); + } + + private static bool IsTransportException(Exception exception) => + exception is IOException or HttpListenerException or ObjectDisposedException; + private static bool IsAuthorized(string authToken, string? authHeader) { if (string.IsNullOrEmpty(authHeader)) return false; @@ -399,7 +785,9 @@ private static async Task ReadBodyAsync(HttpListenerRequest request, lon long total = 0; while (true) { - var n = await request.InputStream.ReadAsync(buffer.AsMemory(0, buffer.Length), ct).ConfigureAwait(false); + var n = await request.InputStream + .ReadAsync(buffer.AsMemory(0, buffer.Length), ct) + .ConfigureAwait(false); if (n <= 0) break; total += n; if (total > maxBytes) throw new InvalidDataException("request body exceeds cap"); @@ -413,9 +801,13 @@ private static async Task ReadBodyAsync(HttpListenerRequest request, lon } } - private void Reject(HttpListenerContext ctx, HttpStatusCode status, string reason) + private Type? Reject(HttpListenerContext ctx, HttpStatusCode status, string reason) { - try { _writeText(ctx.Response, status, reason, "text/plain"); } + try + { + _writeText(ctx.Response, status, reason, "text/plain"); + return null; + } catch (Exception ex) { // Response may already be disposed; a failed write means the client @@ -423,6 +815,7 @@ private void Reject(HttpListenerContext ctx, HttpStatusCode status, string reaso // outside a catch block, so emit a Trace breadcrumb here rather than // relying on a (non-existent) outer log. System.Diagnostics.Trace.WriteLine($"McpHttpServer.Reject: failed to write {(int)status} '{reason}': {ex.GetType().Name}: {ex.Message}"); + return ex.GetType(); } } @@ -454,30 +847,83 @@ public Task StopAsync(TimeSpan drainTimeout) private async Task StopCoreAsync(TimeSpan drainTimeout) { - try { _cts.Cancel(); } - catch (Exception ex) { _logger.Debug($"[MCP] StopCore cts.Cancel threw: {ex.Message}"); } - try { if (_listener.IsListening) _listener.Stop(); } - catch (Exception ex) { _logger.Debug($"[MCP] StopCore listener.Stop threw: {ex.Message}"); } + using var telemetry = McpServerTelemetry.StartLifecycle(McpServerOperation.Stop); + var outcome = McpServerOutcome.Success; + var errorCategory = McpServerErrorCategory.None; + Type? errorType = null; - // Snapshot before awaiting — handlers remove themselves on completion, - // and we don't want enumeration to race the continuation. - Task[] toAwait; - lock (_activeLock) { toAwait = new Task[_activeHandlers.Count]; _activeHandlers.CopyTo(toAwait); } - - var allHandlers = Task.WhenAll(toAwait); - var deadline = Task.Delay(drainTimeout); - var winner = await Task.WhenAny(allHandlers, deadline).ConfigureAwait(false); - if (winner == deadline && toAwait.Length > 0) + void SetFailure(McpServerErrorCategory category, Type? type = null) { - int still; - lock (_activeLock) { still = _activeHandlers.Count; } - _logger.Warn($"[MCP] Drain timeout ({drainTimeout.TotalSeconds:F1}s); {still} handler(s) still running"); + if (outcome != McpServerOutcome.Success) + return; + + outcome = McpServerOutcome.Failure; + errorCategory = category; + errorType = type; } - if (_acceptLoop != null) + try + { + try { _cts.Cancel(); } + catch (Exception ex) + { + SetFailure(McpServerErrorCategory.InternalFailure, ex.GetType()); + _logger.Debug($"[MCP] StopCore cts.Cancel threw: {ex.Message}"); + } + + try + { + if (_listener.IsListening) + _listener.Stop(); + } + catch (Exception ex) + { + SetFailure(McpServerErrorCategory.ListenerStop, ex.GetType()); + McpServerTelemetry.RecordListenerError(McpServerErrorCategory.ListenerStop); + _logger.Debug($"[MCP] StopCore listener.Stop threw: {ex.Message}"); + } + + // Snapshot before awaiting — handlers remove themselves on completion, + // and we don't want enumeration to race the continuation. + Task[] toAwait; + lock (_activeLock) + { + toAwait = new Task[_activeHandlers.Count]; + _activeHandlers.CopyTo(toAwait); + } + + var allHandlers = Task.WhenAll(toAwait); + var deadline = Task.Delay(drainTimeout); + var winner = await Task.WhenAny(allHandlers, deadline).ConfigureAwait(false); + if (winner == deadline && toAwait.Length > 0) + { + SetFailure(McpServerErrorCategory.DrainTimeout); + int still; + lock (_activeLock) { still = _activeHandlers.Count; } + _logger.Warn($"[MCP] Drain timeout ({drainTimeout.TotalSeconds:F1}s); {still} handler(s) still running"); + } + + if (_acceptLoop != null) + { + try + { + await Task.WhenAny(_acceptLoop, Task.Delay(TimeSpan.FromSeconds(1))) + .ConfigureAwait(false); + } + catch (Exception ex) + { + _logger.Debug($"[MCP] Accept loop final await threw (loop may have errored): {ex.Message}"); + } + } + } + catch (Exception ex) { - try { await Task.WhenAny(_acceptLoop, Task.Delay(TimeSpan.FromSeconds(1))).ConfigureAwait(false); } - catch (Exception ex) { _logger.Debug($"[MCP] Accept loop final await threw (loop may have errored): {ex.Message}"); } + SetFailure(McpServerErrorCategory.InternalFailure, ex.GetType()); + throw; + } + finally + { + telemetry.Complete(outcome, errorCategory, errorType); } } @@ -511,7 +957,7 @@ private async Task DisposeCoreAsync() { try { - await StopAsync(DrainTimeout).ConfigureAwait(false); + await StopAsync(_options.DrainTimeout).ConfigureAwait(false); } catch (Exception ex) { @@ -537,7 +983,11 @@ private void DisposeResources() } try { _listener.Close(); } - catch (Exception ex) { _logger.Debug($"[MCP] listener.Close during dispose threw: {ex.Message}"); } + catch (Exception ex) + { + McpServerTelemetry.RecordListenerError(McpServerErrorCategory.ListenerClose); + _logger.Debug($"[MCP] listener.Close during dispose threw: {ex.Message}"); + } _cts.Dispose(); _handlerLimiter.Dispose(); } @@ -565,4 +1015,37 @@ private void ObserveBackgroundFault(Task task, string message) TaskScheduler.Default); } } + + private readonly record struct McpRequestResult( + McpServerOutcome Outcome, + McpServerErrorCategory ErrorCategory, + Type? ErrorType = null) + { + public static McpRequestResult Success { get; } = new( + McpServerOutcome.Success, + McpServerErrorCategory.None); + } + + internal enum McpRequestCancellationCause + { + None, + Timeout, + Shutdown + } + + internal sealed class McpRequestCancellationState + { + private int _cause; + + public McpRequestCancellationCause Cause => + (McpRequestCancellationCause)Volatile.Read(ref _cause); + + public void TrySet(McpRequestCancellationCause cause) => + Interlocked.CompareExchange(ref _cause, (int)cause, (int)McpRequestCancellationCause.None); + } + + private sealed record McpRequestCancellationPropagation( + McpRequestCancellationState State, + CancellationTokenSource RequestCancellation, + McpRequestCancellationCause Cause); } diff --git a/src/OpenClaw.Shared/Mcp/McpHttpServerOptions.cs b/src/OpenClaw.Shared/Mcp/McpHttpServerOptions.cs new file mode 100644 index 000000000..c6b93f645 --- /dev/null +++ b/src/OpenClaw.Shared/Mcp/McpHttpServerOptions.cs @@ -0,0 +1,72 @@ +using System.Net; + +namespace OpenClaw.Shared.Mcp; + +internal sealed class McpHttpServerOptions +{ + public static McpHttpServerOptions Default { get; } = new( + maxConcurrentHandlers: 16, + handlerAdmissionTimeout: TimeSpan.FromMilliseconds(50), + requestTimeout: TimeSpan.FromMinutes(6), + drainTimeout: TimeSpan.FromSeconds(5)); + + public McpHttpServerOptions( + int maxConcurrentHandlers, + TimeSpan handlerAdmissionTimeout, + TimeSpan requestTimeout, + TimeSpan drainTimeout, + Action? handlerAdmissionStarted = null, + Func>? requestBodyReader = null, + Action? handlerExecutionStarting = null) + { + if (maxConcurrentHandlers <= 0) + throw new ArgumentOutOfRangeException(nameof(maxConcurrentHandlers)); + if (handlerAdmissionTimeout < TimeSpan.Zero) + throw new ArgumentOutOfRangeException(nameof(handlerAdmissionTimeout)); + if (requestTimeout <= TimeSpan.Zero) + throw new ArgumentOutOfRangeException(nameof(requestTimeout)); + if (drainTimeout < TimeSpan.Zero) + throw new ArgumentOutOfRangeException(nameof(drainTimeout)); + + MaxConcurrentHandlers = maxConcurrentHandlers; + HandlerAdmissionTimeout = handlerAdmissionTimeout; + RequestTimeout = requestTimeout; + DrainTimeout = drainTimeout; + HandlerAdmissionStarted = handlerAdmissionStarted; + RequestBodyReader = requestBodyReader; + HandlerExecutionStarting = handlerExecutionStarting; + } + + public int MaxConcurrentHandlers { get; } + public TimeSpan HandlerAdmissionTimeout { get; } + public TimeSpan RequestTimeout { get; } + public TimeSpan DrainTimeout { get; } + public Action? HandlerAdmissionStarted { get; } + public Func>? RequestBodyReader { get; } + public Action? HandlerExecutionStarting { get; } +} + +internal interface IMcpHttpListener +{ + bool IsListening { get; } + void Start(); + Task GetContextAsync(); + void Stop(); + void Close(); +} + +internal sealed class SystemMcpHttpListener : IMcpHttpListener +{ + private readonly HttpListener _listener = new(); + + public SystemMcpHttpListener(int port) + { + _listener.Prefixes.Add($"http://127.0.0.1:{port}/"); + } + + public bool IsListening => _listener.IsListening; + public void Start() => _listener.Start(); + public Task GetContextAsync() => _listener.GetContextAsync(); + public void Stop() => _listener.Stop(); + public void Close() => _listener.Close(); +} diff --git a/src/OpenClaw.Shared/Telemetry/McpServerTelemetry.cs b/src/OpenClaw.Shared/Telemetry/McpServerTelemetry.cs new file mode 100644 index 000000000..18a48435d --- /dev/null +++ b/src/OpenClaw.Shared/Telemetry/McpServerTelemetry.cs @@ -0,0 +1,324 @@ +using System.Diagnostics; +using System.Diagnostics.Metrics; + +namespace OpenClaw.Shared.Telemetry; + +public enum McpServerOperation +{ + Start, + Stop +} + +public enum McpServerRequestKind +{ + Probe, + JsonRpc, + Other +} + +public enum McpServerOutcome +{ + Success, + Failure, + Canceled +} + +public enum McpServerErrorCategory +{ + None, + ListenerStart, + ListenerAccept, + ListenerStop, + ListenerClose, + AuthenticationFailed, + Busy, + Timeout, + Shutdown, + DrainTimeout, + InvalidRequest, + TransportFailure, + InternalFailure +} + +public sealed record McpServerTelemetryCompletion( + McpServerOutcome Outcome, + McpServerErrorCategory ErrorCategory, + string? ErrorType, + double DurationMilliseconds); + +public static class McpServerTelemetry +{ + public const string StartSpanName = "openclaw.mcp.server.start"; + public const string StopSpanName = "openclaw.mcp.server.stop"; + public const string RequestSpanName = "openclaw.mcp.server.request"; + public const string LifecycleOperationsMetricName = "openclaw.mcp.server.lifecycle.operations"; + public const string RequestsMetricName = "openclaw.mcp.server.requests"; + public const string ListenerErrorsMetricName = "openclaw.mcp.server.listener.errors"; + public const string RequestDurationMetricName = "openclaw.mcp.server.request.duration"; + public const string OperationTag = "openclaw.mcp.server.operation"; + public const string RequestKindTag = "openclaw.mcp.server.request.kind"; + + private static readonly Counter LifecycleOperations = OpenClawTelemetry.CreateCounter( + LifecycleOperationsMetricName, + unit: "{operation}", + description: "Number of local MCP server lifecycle operations."); + private static readonly Counter Requests = OpenClawTelemetry.CreateCounter( + RequestsMetricName, + unit: "{request}", + description: "Number of local MCP HTTP requests handled."); + private static readonly Counter ListenerErrors = OpenClawTelemetry.CreateCounter( + ListenerErrorsMetricName, + unit: "{error}", + description: "Number of local MCP HTTP listener errors."); + private static readonly Histogram RequestDuration = OpenClawTelemetry.CreateHistogram( + RequestDurationMetricName, + unit: "ms", + description: "Post-accept local MCP HTTP request handling duration."); + + public static McpServerLifecycleOperation StartLifecycle(McpServerOperation operation) => + new(operation, LifecycleOperations); + + public static McpServerRequestOperation StartRequest(McpServerRequestKind kind) => + new(kind, Requests, RequestDuration); + + public static void RecordListenerError(McpServerErrorCategory errorCategory) + { + if (errorCategory is not ( + McpServerErrorCategory.ListenerStart or + McpServerErrorCategory.ListenerAccept or + McpServerErrorCategory.ListenerStop or + McpServerErrorCategory.ListenerClose)) + { + throw new ArgumentOutOfRangeException( + nameof(errorCategory), + errorCategory, + "Listener errors require a listener error category."); + } + + OpenClawTelemetry.Add( + ListenerErrors, + tags: + [ + OpenClawTelemetryTag.String( + OpenClawTelemetryTagKey.ErrorCategory, + errorCategory.ToTelemetryValue()) + ]); + } + + internal static void ApplyTerminalTags( + Activity? activity, + McpServerOutcome outcome, + McpServerErrorCategory errorCategory, + Type? errorType) + { + if (activity == null) + return; + + activity.SetTag(OpenClawTelemetryTagKey.Outcome.ToTelemetryName(), outcome.ToTelemetryValue()); + if (errorCategory != McpServerErrorCategory.None) + { + activity.SetTag( + OpenClawTelemetryTagKey.ErrorCategory.ToTelemetryName(), + errorCategory.ToTelemetryValue()); + } + + if (errorType != null) + activity.SetTag(OpenClawTelemetryTagKey.ErrorType.ToTelemetryName(), errorType.FullName); + + activity.SetStatus(outcome switch + { + McpServerOutcome.Success => ActivityStatusCode.Ok, + McpServerOutcome.Failure => ActivityStatusCode.Error, + McpServerOutcome.Canceled => ActivityStatusCode.Unset, + _ => throw new ArgumentOutOfRangeException( + nameof(outcome), + outcome, + "Unknown MCP server outcome.") + }, outcome == McpServerOutcome.Failure ? errorType?.Name : null); + } + + internal static OpenClawTelemetryTag[] CreateMetricTags( + string dimensionTag, + string dimensionValue, + McpServerOutcome outcome, + McpServerErrorCategory errorCategory) => + [ + OpenClawTelemetryTag.String(dimensionTag, dimensionValue), + OpenClawTelemetryTag.String(OpenClawTelemetryTagKey.Outcome, outcome.ToTelemetryValue()), + OpenClawTelemetryTag.String( + OpenClawTelemetryTagKey.ErrorCategory, + errorCategory.ToTelemetryValue()) + ]; +} + +public sealed class McpServerLifecycleOperation : IDisposable +{ + private readonly Activity? _activity; + private readonly Stopwatch _stopwatch = Stopwatch.StartNew(); + private readonly Counter _operations; + private readonly McpServerOperation _operation; + private int _completed; + + internal McpServerLifecycleOperation( + McpServerOperation operation, + Counter operations) + { + _operation = operation; + _operations = operations; + _activity = OpenClawTelemetry.StartDetachedActivity( + operation == McpServerOperation.Start + ? McpServerTelemetry.StartSpanName + : McpServerTelemetry.StopSpanName, + default(ActivityContext), + [ + OpenClawTelemetryTag.String( + McpServerTelemetry.OperationTag, + operation.ToTelemetryValue()) + ]); + } + + public McpServerTelemetryCompletion? Complete( + McpServerOutcome outcome, + McpServerErrorCategory errorCategory = McpServerErrorCategory.None, + Type? errorType = null) + { + if (Interlocked.Exchange(ref _completed, 1) != 0) + return null; + + _stopwatch.Stop(); + McpServerTelemetry.ApplyTerminalTags(_activity, outcome, errorCategory, errorType); + OpenClawTelemetry.Add( + _operations, + tags: McpServerTelemetry.CreateMetricTags( + McpServerTelemetry.OperationTag, + _operation.ToTelemetryValue(), + outcome, + errorCategory)); + OpenClawTelemetry.StopDetachedActivity(_activity); + return new McpServerTelemetryCompletion( + outcome, + errorCategory, + errorType?.FullName, + _stopwatch.Elapsed.TotalMilliseconds); + } + + public void Dispose() => + Complete(McpServerOutcome.Canceled, McpServerErrorCategory.InternalFailure); +} + +public sealed class McpServerRequestOperation : IDisposable +{ + private readonly Activity? _activity; + private readonly Stopwatch _stopwatch = Stopwatch.StartNew(); + private readonly Counter _requests; + private readonly Histogram _duration; + private readonly McpServerRequestKind _kind; + private int _completed; + + internal McpServerRequestOperation( + McpServerRequestKind kind, + Counter requests, + Histogram duration) + { + _kind = kind; + _requests = requests; + _duration = duration; + _activity = OpenClawTelemetry.StartDetachedActivity( + McpServerTelemetry.RequestSpanName, + default(ActivityContext), + [ + OpenClawTelemetryTag.String( + McpServerTelemetry.RequestKindTag, + kind.ToTelemetryValue()) + ], + System.Diagnostics.ActivityKind.Server); + } + + public McpServerTelemetryCompletion? Complete( + McpServerOutcome outcome, + McpServerErrorCategory errorCategory = McpServerErrorCategory.None, + Type? errorType = null) + { + if (Interlocked.Exchange(ref _completed, 1) != 0) + return null; + + _stopwatch.Stop(); + McpServerTelemetry.ApplyTerminalTags(_activity, outcome, errorCategory, errorType); + var tags = McpServerTelemetry.CreateMetricTags( + McpServerTelemetry.RequestKindTag, + _kind.ToTelemetryValue(), + outcome, + errorCategory); + OpenClawTelemetry.Add(_requests, tags: tags); + OpenClawTelemetry.Record(_duration, _stopwatch.Elapsed.TotalMilliseconds, tags); + OpenClawTelemetry.StopDetachedActivity(_activity); + return new McpServerTelemetryCompletion( + outcome, + errorCategory, + errorType?.FullName, + _stopwatch.Elapsed.TotalMilliseconds); + } + + public void Dispose() => + Complete(McpServerOutcome.Canceled, McpServerErrorCategory.InternalFailure); +} + +public static class McpServerTelemetryValues +{ + public static string ToTelemetryValue(this McpServerOperation value) => + value switch + { + McpServerOperation.Start => "start", + McpServerOperation.Stop => "stop", + _ => throw new ArgumentOutOfRangeException( + nameof(value), + value, + "Unknown MCP server operation.") + }; + + public static string ToTelemetryValue(this McpServerRequestKind value) => + value switch + { + McpServerRequestKind.Probe => "probe", + McpServerRequestKind.JsonRpc => "json_rpc", + McpServerRequestKind.Other => "other", + _ => throw new ArgumentOutOfRangeException( + nameof(value), + value, + "Unknown MCP server request kind.") + }; + + public static string ToTelemetryValue(this McpServerOutcome value) => + value switch + { + McpServerOutcome.Success => "success", + McpServerOutcome.Failure => "failure", + McpServerOutcome.Canceled => "canceled", + _ => throw new ArgumentOutOfRangeException( + nameof(value), + value, + "Unknown MCP server outcome.") + }; + + public static string ToTelemetryValue(this McpServerErrorCategory value) => + value switch + { + McpServerErrorCategory.None => "none", + McpServerErrorCategory.ListenerStart => "listener_start", + McpServerErrorCategory.ListenerAccept => "listener_accept", + McpServerErrorCategory.ListenerStop => "listener_stop", + McpServerErrorCategory.ListenerClose => "listener_close", + McpServerErrorCategory.AuthenticationFailed => "authentication_failed", + McpServerErrorCategory.Busy => "busy", + McpServerErrorCategory.Timeout => "timeout", + McpServerErrorCategory.Shutdown => "shutdown", + McpServerErrorCategory.DrainTimeout => "drain_timeout", + McpServerErrorCategory.InvalidRequest => "invalid_request", + McpServerErrorCategory.TransportFailure => "transport_failure", + McpServerErrorCategory.InternalFailure => "internal_failure", + _ => throw new ArgumentOutOfRangeException( + nameof(value), + value, + "Unknown MCP server error category.") + }; +} diff --git a/tests/OpenClaw.Shared.Tests/McpHttpServerTelemetryTests.cs b/tests/OpenClaw.Shared.Tests/McpHttpServerTelemetryTests.cs new file mode 100644 index 000000000..fa9701e29 --- /dev/null +++ b/tests/OpenClaw.Shared.Tests/McpHttpServerTelemetryTests.cs @@ -0,0 +1,729 @@ +using System.Collections.Concurrent; +using System.Diagnostics; +using System.Diagnostics.Metrics; +using System.Net; +using System.Net.Http; +using System.Net.Sockets; +using System.Text; +using OpenClaw.Shared.Mcp; +using OpenClaw.Shared.Telemetry; +using OpenClaw.Shared.Tests.Telemetry; + +namespace OpenClaw.Shared.Tests; + +[Collection(McpServerTelemetryCollection.Name)] +public sealed class McpHttpServerTelemetryTests +{ + [Fact] + public async Task ConcurrentStartAndRepeatedStop_EmitLifecycleExactlyOnce() + { + using var metrics = new MetricCollector(); + var listener = new TestMcpHttpListener(blockStart: true); + await using var server = CreateServer(listener); + + var firstStart = Task.Run(server.Start); + await listener.StartEntered.Task.WaitAsync(TimeSpan.FromSeconds(2)); + var secondStart = Task.Run(server.Start); + listener.ReleaseStart(); + await Task.WhenAll(firstStart, secondStart); + + await server.StopAsync(TimeSpan.FromSeconds(1)); + await server.StopAsync(TimeSpan.FromSeconds(1)); + + Assert.Equal(1, listener.StartCount); + Assert.Equal(1, listener.StopCount); + await EventuallyAsync(() => + FindMeasurements(metrics, McpServerTelemetry.LifecycleOperationsMetricName).Count == 2); + Assert.Single( + FindMeasurements(metrics, McpServerTelemetry.LifecycleOperationsMetricName), + measurement => HasTags( + measurement.Tags, + (McpServerTelemetry.OperationTag, "start"), + (OpenClawTelemetryTagKey.Outcome.ToTelemetryName(), "success"))); + Assert.Single( + FindMeasurements(metrics, McpServerTelemetry.LifecycleOperationsMetricName), + measurement => HasTags( + measurement.Tags, + (McpServerTelemetry.OperationTag, "stop"), + (OpenClawTelemetryTagKey.Outcome.ToTelemetryName(), "success"))); + } + + [Fact] + public async Task ListenerFailures_AreFiniteAndCloseDoesNotOverwriteStop() + { + using var activities = new ActivityCollector(); + using var metrics = new MetricCollector(); + var listener = new TestMcpHttpListener + { + AcceptErrorOnce = new HttpListenerException(995), + StopError = new HttpListenerException(87), + CloseError = new InvalidOperationException("close failed") + }; + var server = CreateServer(listener); + + server.Start(); + await EventuallyAsync(() => HasListenerError(metrics, "listener_accept")); + await server.DisposeAsync(); + + await EventuallyAsync(() => + HasListenerError(metrics, "listener_stop") && + HasListenerError(metrics, "listener_close")); + var stop = Assert.Single( + activities.Stopped, + activity => activity.OperationName == McpServerTelemetry.StopSpanName); + Assert.Equal( + "listener_stop", + stop.GetTagItem(OpenClawTelemetryTagKey.ErrorCategory.ToTelemetryName())); + Assert.DoesNotContain( + activities.Stopped, + activity => + activity.OperationName == McpServerTelemetry.StopSpanName && + Equals( + activity.GetTagItem(OpenClawTelemetryTagKey.ErrorCategory.ToTelemetryName()), + "listener_close")); + } + + [Fact] + public async Task ListenerStartFailure_IsRecordedAndRethrown() + { + using var activities = new ActivityCollector(); + using var metrics = new MetricCollector(); + var expected = new HttpListenerException(5); + var listener = new TestMcpHttpListener { StartError = expected }; + var server = CreateServer(listener); + + var thrown = Assert.Throws(server.Start); + + Assert.Same(expected, thrown); + Assert.True(HasListenerError(metrics, "listener_start")); + var start = Assert.Single( + activities.Stopped, + activity => activity.OperationName == McpServerTelemetry.StartSpanName); + Assert.Equal(ActivityStatusCode.Error, start.Status); + Assert.Equal( + "listener_start", + start.GetTagItem(OpenClawTelemetryTagKey.ErrorCategory.ToTelemetryName())); + Assert.Equal( + typeof(HttpListenerException).FullName, + start.GetTagItem(OpenClawTelemetryTagKey.ErrorType.ToTelemetryName())); + await server.DisposeAsync(); + } + + [Fact] + public async Task ListenerStopFailure_TakesPrecedenceOverDrainTimeout() + { + using var activities = new ActivityCollector(); + using var metrics = new MetricCollector(); + var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var capability = new GatedCapability(release); + var port = FreePort(); + var bridge = new McpToolBridge(() => new INodeCapability[] { capability }); + var options = CreateOptions(); + var listener = new ThrowAfterStopListener(new SystemMcpHttpListener(port)); + await using var server = new McpHttpServer( + bridge, + port, + NullLogger.Instance, + authToken: null, + options, + listener); + server.Start(); + using var http = new HttpClient { BaseAddress = new Uri(server.Endpoint) }; + var request = http.PostAsync( + "", + Json("""{"jsonrpc":"2.0","id":1,"method":"tools/call","params":{"name":"gate.wait"}}""")); + await capability.Entered.WaitAsync(TimeSpan.FromSeconds(2)); + + await server.StopAsync(TimeSpan.Zero); + release.TrySetResult(true); + await IgnoreRequestFailureAsync(request); + + var stop = Assert.Single( + activities.Stopped, + activity => activity.OperationName == McpServerTelemetry.StopSpanName); + Assert.Equal( + "listener_stop", + stop.GetTagItem(OpenClawTelemetryTagKey.ErrorCategory.ToTelemetryName())); + Assert.True(HasListenerError(metrics, "listener_stop")); + Assert.Single( + FindMeasurements(metrics, McpServerTelemetry.LifecycleOperationsMetricName), + measurement => HasTags( + measurement.Tags, + (McpServerTelemetry.OperationTag, "stop"), + (OpenClawTelemetryTagKey.ErrorCategory.ToTelemetryName(), "listener_stop"))); + } + + [Fact] + public async Task AuthFailureAndMalformedJson_UseTransportOutcomes() + { + using var metrics = new MetricCollector(); + var port = FreePort(); + var authToken = Guid.Empty.ToString("N"); + var bridge = new McpToolBridge(() => Array.Empty()); + await using var server = new McpHttpServer( + bridge, + port, + NullLogger.Instance, + authToken); + server.Start(); + using var http = new HttpClient { BaseAddress = new Uri(server.Endpoint) }; + + using var unauthorized = await http.PostAsync( + "", + Json("""{"jsonrpc":"2.0","id":1,"method":"ping"}""")); + Assert.Equal(HttpStatusCode.Unauthorized, unauthorized.StatusCode); + + http.DefaultRequestHeaders.Authorization = + new System.Net.Http.Headers.AuthenticationHeaderValue("Bearer", authToken); + using var malformed = await http.PostAsync("", Json("{")); + Assert.Equal(HttpStatusCode.OK, malformed.StatusCode); + + await EventuallyAsync(() => + FindMeasurements(metrics, McpServerTelemetry.RequestsMetricName).Count >= 2); + Assert.Single( + FindMeasurements(metrics, McpServerTelemetry.RequestsMetricName), + measurement => HasTags( + measurement.Tags, + (OpenClawTelemetryTagKey.Outcome.ToTelemetryName(), "failure"), + (OpenClawTelemetryTagKey.ErrorCategory.ToTelemetryName(), "authentication_failed"))); + Assert.Single( + FindMeasurements(metrics, McpServerTelemetry.RequestsMetricName), + measurement => HasTags( + measurement.Tags, + (OpenClawTelemetryTagKey.Outcome.ToTelemetryName(), "success"), + (OpenClawTelemetryTagKey.ErrorCategory.ToTelemetryName(), "none"))); + await AssertRequestDurationsAsync(metrics, expectedCount: 2); + } + + [Fact] + public async Task HandlerSaturation_RecordsBusy() + { + using var metrics = new MetricCollector(); + var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var capability = new GatedCapability(release); + var port = FreePort(); + var bridge = new McpToolBridge(() => new INodeCapability[] { capability }); + var options = CreateOptions( + maxConcurrentHandlers: 1, + handlerAdmissionTimeout: TimeSpan.FromMilliseconds(20)); + await using var server = new McpHttpServer( + bridge, + port, + NullLogger.Instance, + authToken: null, + options); + server.Start(); + using var http = new HttpClient { BaseAddress = new Uri(server.Endpoint) }; + + var first = http.PostAsync( + "", + Json("""{"jsonrpc":"2.0","id":1,"method":"tools/call","params":{"name":"gate.wait"}}""")); + await capability.Entered.WaitAsync(TimeSpan.FromSeconds(2)); + using var busy = await http.GetAsync(""); + + Assert.Equal(HttpStatusCode.ServiceUnavailable, busy.StatusCode); + release.TrySetResult(true); + using var firstResponse = await first; + await EventuallyAsync(() => HasRequestCategory(metrics, "busy")); + await AssertRequestDurationsAsync(metrics, expectedCount: 2); + } + + [Fact] + public async Task ShutdownWhileWaitingForHandler_RecordsShutdownNotBusyOrTimeout() + { + using var metrics = new MetricCollector(); + var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var capability = new GatedCapability(release); + var secondAdmission = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var admissions = 0; + var port = FreePort(); + var bridge = new McpToolBridge(() => new INodeCapability[] { capability }); + var options = CreateOptions( + maxConcurrentHandlers: 1, + handlerAdmissionTimeout: TimeSpan.FromSeconds(10), + handlerAdmissionStarted: () => + { + if (Interlocked.Increment(ref admissions) == 2) + secondAdmission.TrySetResult(true); + }); + await using var server = new McpHttpServer( + bridge, + port, + NullLogger.Instance, + authToken: null, + options); + server.Start(); + using var http = new HttpClient { BaseAddress = new Uri(server.Endpoint) }; + + var first = http.PostAsync( + "", + Json("""{"jsonrpc":"2.0","id":1,"method":"tools/call","params":{"name":"gate.wait"}}""")); + await capability.Entered.WaitAsync(TimeSpan.FromSeconds(2)); + var waiting = http.GetAsync(""); + await secondAdmission.Task.WaitAsync(TimeSpan.FromSeconds(2)); + + await server.StopAsync(TimeSpan.FromMilliseconds(20)); + release.TrySetResult(true); + await IgnoreRequestFailureAsync(first); + await IgnoreRequestFailureAsync(waiting); + + await EventuallyAsync(() => HasRequestCategory(metrics, "shutdown")); + Assert.False(HasRequestCategory(metrics, "busy")); + Assert.False(HasRequestCategory(metrics, "timeout")); + } + + [Fact] + public void CancellationCause_FallsBackOnlyWhenOneSourceCanceled() + { + using var timeout = new CancellationTokenSource(); + using var shutdown = new CancellationTokenSource(); + var timeoutState = new McpHttpServer.McpRequestCancellationState(); + + timeout.Cancel(); + + Assert.Equal( + McpHttpServer.McpRequestCancellationCause.Timeout, + McpHttpServer.ResolveCancellationCause( + timeoutState, + timeout.Token, + shutdown.Token)); + + shutdown.Cancel(); + + Assert.Equal( + McpHttpServer.McpRequestCancellationCause.Timeout, + McpHttpServer.ResolveCancellationCause( + timeoutState, + timeout.Token, + shutdown.Token)); + + using var shutdownOnly = new CancellationTokenSource(); + var shutdownState = new McpHttpServer.McpRequestCancellationState(); + shutdownOnly.Cancel(); + + Assert.Equal( + McpHttpServer.McpRequestCancellationCause.Shutdown, + McpHttpServer.ResolveCancellationCause( + shutdownState, + CancellationToken.None, + shutdownOnly.Token)); + + var ambiguousState = new McpHttpServer.McpRequestCancellationState(); + + Assert.Equal( + McpHttpServer.McpRequestCancellationCause.None, + McpHttpServer.ResolveCancellationCause( + ambiguousState, + timeout.Token, + shutdown.Token)); + } + + [Fact] + public void CancellationPropagation_RecordsFirstCauseBeforeCancelingRequest() + { + using var requestCancellation = new CancellationTokenSource(); + var state = new McpHttpServer.McpRequestCancellationState(); + var observedCause = McpHttpServer.McpRequestCancellationCause.None; + using var registration = requestCancellation.Token.Register( + () => observedCause = state.Cause); + + McpHttpServer.PropagateCancellation( + state, + requestCancellation, + McpHttpServer.McpRequestCancellationCause.Timeout); + state.TrySet(McpHttpServer.McpRequestCancellationCause.Shutdown); + + Assert.True(requestCancellation.IsCancellationRequested); + Assert.Equal(McpHttpServer.McpRequestCancellationCause.Timeout, observedCause); + Assert.Equal(McpHttpServer.McpRequestCancellationCause.Timeout, state.Cause); + } + + [Fact] + public async Task DisposeAfterDrainTimeout_BeforeHandlerSetup_CompletesShutdownAndReleasesSlot() + { + using var metrics = new MetricCollector(); + using var executionRelease = new ManualResetEventSlim(false); + var executionEntered = new TaskCompletionSource( + TaskCreationOptions.RunContinuationsAsynchronously); + var port = FreePort(); + var bridge = new McpToolBridge(() => Array.Empty()); + var options = new McpHttpServerOptions( + maxConcurrentHandlers: 1, + handlerAdmissionTimeout: TimeSpan.FromMilliseconds(50), + requestTimeout: TimeSpan.FromSeconds(5), + drainTimeout: TimeSpan.Zero, + handlerExecutionStarting: () => + { + executionEntered.TrySetResult(true); + executionRelease.Wait(); + }); + var server = new McpHttpServer( + bridge, + port, + NullLogger.Instance, + authToken: null, + options); + server.Start(); + using var http = new HttpClient { BaseAddress = new Uri(server.Endpoint) }; + var request = http.GetAsync(""); + await executionEntered.Task.WaitAsync(TimeSpan.FromSeconds(2)); + + await server.DisposeAsync(); + executionRelease.Set(); + await IgnoreRequestFailureAsync(request); + + await EventuallyAsync(() => HasRequestCategory(metrics, "shutdown")); + Assert.False(HasRequestCategory(metrics, "internal_failure")); + } + + [Fact] + public async Task SlowBodyDeadline_RecordsTimeout() + { + using var metrics = new MetricCollector(); + var port = FreePort(); + var bridge = new McpToolBridge(() => Array.Empty()); + var options = CreateOptions( + requestTimeout: TimeSpan.FromMilliseconds(50), + requestBodyReader: static async (_, _, cancellationToken) => + { + await Task.Delay(Timeout.InfiniteTimeSpan, cancellationToken); + return ""; + }); + await using var server = new McpHttpServer( + bridge, + port, + NullLogger.Instance, + authToken: null, + options); + server.Start(); + using var http = new HttpClient { BaseAddress = new Uri(server.Endpoint) }; + + using var response = await http.PostAsync( + "", + Json("""{"jsonrpc":"2.0","id":1,"method":"ping"}""")); + + Assert.Equal(HttpStatusCode.RequestTimeout, response.StatusCode); + await EventuallyAsync(() => HasRequestCategory(metrics, "timeout")); + await AssertRequestDurationsAsync(metrics, expectedCount: 1); + } + + [Fact] + public async Task ResponseWriteFailure_RecordsTransportFailure() + { + using var metrics = new MetricCollector(); + var port = FreePort(); + var bridge = new McpToolBridge(() => Array.Empty()); + await using var server = new McpHttpServer( + bridge, + port, + NullLogger.Instance, + authToken: null, + static (response, _, _, _) => + { + response.Close(); + throw new IOException("simulated response write failure"); + }); + server.Start(); + using var http = new HttpClient { BaseAddress = new Uri(server.Endpoint) }; + + await IgnoreRequestFailureAsync(http.GetAsync("")); + + await EventuallyAsync(() => HasRequestCategory(metrics, "transport_failure")); + await AssertRequestDurationsAsync(metrics, expectedCount: 1); + } + + private static McpHttpServer CreateServer(TestMcpHttpListener listener) + { + var bridge = new McpToolBridge(() => Array.Empty()); + return new McpHttpServer( + bridge, + 8765, + NullLogger.Instance, + authToken: null, + static (_, _, _, _) => { }, + McpHttpServerOptions.Default, + listener); + } + + private static McpHttpServerOptions CreateOptions( + int maxConcurrentHandlers = 16, + TimeSpan? handlerAdmissionTimeout = null, + TimeSpan? requestTimeout = null, + Action? handlerAdmissionStarted = null, + Func>? requestBodyReader = null, + Action? handlerExecutionStarting = null) => + new( + maxConcurrentHandlers, + handlerAdmissionTimeout ?? TimeSpan.FromMilliseconds(50), + requestTimeout ?? TimeSpan.FromSeconds(5), + TimeSpan.FromSeconds(1), + handlerAdmissionStarted, + requestBodyReader, + handlerExecutionStarting); + + private static StringContent Json(string body) => + new(body, Encoding.UTF8, "application/json"); + + private static int FreePort() + { + var listener = new TcpListener(IPAddress.Loopback, 0); + listener.Start(); + var port = ((IPEndPoint)listener.LocalEndpoint).Port; + listener.Stop(); + return port; + } + + private static async Task IgnoreRequestFailureAsync(Task request) + { + try + { + using var response = await request; + } + catch (HttpRequestException) + { + } + catch (TaskCanceledException) + { + } + } + + private static async Task EventuallyAsync(Func condition) + { + var deadline = Stopwatch.StartNew(); + while (!condition()) + { + Assert.True( + deadline.Elapsed < TimeSpan.FromSeconds(2), + "Timed out waiting for telemetry."); + await Task.Delay(10); + } + } + + private static bool HasListenerError(MetricCollector metrics, string category) => + FindMeasurements(metrics, McpServerTelemetry.ListenerErrorsMetricName) + .Any(measurement => HasTags( + measurement.Tags, + (OpenClawTelemetryTagKey.ErrorCategory.ToTelemetryName(), category))); + + private static bool HasRequestCategory(MetricCollector metrics, string category) => + FindMeasurements(metrics, McpServerTelemetry.RequestsMetricName) + .Any(measurement => HasTags( + measurement.Tags, + (OpenClawTelemetryTagKey.ErrorCategory.ToTelemetryName(), category))); + + private static List> FindMeasurements( + MetricCollector metrics, + string name) => + metrics.LongMeasurements.Where(item => item.Name == name).ToList(); + + private static async Task AssertRequestDurationsAsync( + MetricCollector metrics, + int expectedCount) + { + await EventuallyAsync(() => + FindMeasurements(metrics, McpServerTelemetry.RequestsMetricName).Count == expectedCount && + FindDurationMeasurements(metrics).Count == expectedCount); + + var requests = FindMeasurements(metrics, McpServerTelemetry.RequestsMetricName); + var durations = FindDurationMeasurements(metrics); + Assert.All(durations, measurement => Assert.True(measurement.Value >= 0)); + Assert.Equal( + requests.Select(measurement => TagSignature(measurement.Tags)).Order(), + durations.Select(measurement => TagSignature(measurement.Tags)).Order()); + } + + private static List> FindDurationMeasurements( + MetricCollector metrics) => + metrics.DoubleMeasurements + .Where(item => item.Name == McpServerTelemetry.RequestDurationMetricName) + .ToList(); + + private static string TagSignature( + IReadOnlyCollection> tags) => + string.Join( + "|", + tags.OrderBy(tag => tag.Key) + .Select(tag => $"{tag.Key}={tag.Value}")); + + private static bool HasTags( + IReadOnlyCollection> actual, + params (string Key, object? Value)[] expected) => + expected.All(pair => + actual.Any(tag => tag.Key == pair.Key && Equals(tag.Value, pair.Value))); + + private sealed class GatedCapability : INodeCapability + { + private readonly TaskCompletionSource _release; + private readonly TaskCompletionSource _entered = + new(TaskCreationOptions.RunContinuationsAsynchronously); + + public GatedCapability(TaskCompletionSource release) + { + _release = release; + } + + public string Category => "gate"; + public IReadOnlyList Commands => ["gate.wait"]; + public Task Entered => _entered.Task; + public bool CanHandle(string command) => command == "gate.wait"; + + public async Task ExecuteAsync(NodeInvokeRequest request) + { + _entered.TrySetResult(true); + await _release.Task.ConfigureAwait(false); + return new NodeInvokeResponse { Ok = true }; + } + } + + private sealed class TestMcpHttpListener : IMcpHttpListener + { + private readonly TaskCompletionSource _pendingAccept = + new(TaskCreationOptions.RunContinuationsAsynchronously); + private readonly ManualResetEventSlim _startRelease; + private int _acceptCount; + + public TestMcpHttpListener(bool blockStart = false) + { + _startRelease = new ManualResetEventSlim(!blockStart); + } + + public bool IsListening { get; private set; } + public Exception? StartError { get; init; } + public Exception? AcceptErrorOnce { get; init; } + public Exception? StopError { get; init; } + public Exception? CloseError { get; init; } + public int StartCount { get; private set; } + public int StopCount { get; private set; } + public TaskCompletionSource StartEntered { get; } = + new(TaskCreationOptions.RunContinuationsAsynchronously); + + public void Start() + { + StartCount++; + StartEntered.TrySetResult(true); + _startRelease.Wait(); + if (StartError != null) + throw StartError; + IsListening = true; + } + + public Task GetContextAsync() + { + if (Interlocked.Increment(ref _acceptCount) == 1 && AcceptErrorOnce != null) + return Task.FromException(AcceptErrorOnce); + return _pendingAccept.Task; + } + + public void Stop() + { + StopCount++; + IsListening = false; + _pendingAccept.TrySetException(new HttpListenerException(995)); + if (StopError != null) + throw StopError; + } + + public void Close() + { + IsListening = false; + _pendingAccept.TrySetException(new ObjectDisposedException(nameof(TestMcpHttpListener))); + _startRelease.Dispose(); + if (CloseError != null) + throw CloseError; + } + + public void ReleaseStart() => _startRelease.Set(); + } + + private sealed class ThrowAfterStopListener : IMcpHttpListener + { + private readonly IMcpHttpListener _inner; + + public ThrowAfterStopListener(IMcpHttpListener inner) + { + _inner = inner; + } + + public bool IsListening => _inner.IsListening; + public void Start() => _inner.Start(); + public Task GetContextAsync() => _inner.GetContextAsync(); + + public void Stop() + { + _inner.Stop(); + throw new HttpListenerException(87); + } + + public void Close() => _inner.Close(); + } + + private sealed class ActivityCollector : IDisposable + { + private readonly ActivityListener _listener; + + public ActivityCollector() + { + _listener = new ActivityListener + { + ShouldListenTo = source => + source.Name == OpenClawActivitySourceName.OpenClaw.ToTelemetryName(), + Sample = (ref ActivityCreationOptions options) => + options.Name.StartsWith("openclaw.mcp.server.", StringComparison.Ordinal) + ? ActivitySamplingResult.AllDataAndRecorded + : ActivitySamplingResult.None, + ActivityStopped = activity => + { + if (activity.OperationName.StartsWith( + "openclaw.mcp.server.", + StringComparison.Ordinal)) + { + Stopped.Enqueue(activity); + } + } + }; + ActivitySource.AddActivityListener(_listener); + } + + public ConcurrentQueue Stopped { get; } = new(); + + public void Dispose() => _listener.Dispose(); + } + + private sealed class MetricCollector : IDisposable + { + private readonly MeterListener _listener = new(); + + public MetricCollector() + { + _listener.InstrumentPublished = (instrument, listener) => + { + if (instrument.Meter.Name == OpenClawMeterName.OpenClaw.ToTelemetryName() && + instrument.Name.StartsWith("openclaw.mcp.server.", StringComparison.Ordinal)) + { + listener.EnableMeasurementEvents(instrument); + } + }; + _listener.SetMeasurementEventCallback((instrument, measurement, tags, _) => + LongMeasurements.Add(new MetricMeasurement( + instrument.Name, + measurement, + tags.ToArray()))); + _listener.SetMeasurementEventCallback((instrument, measurement, tags, _) => + DoubleMeasurements.Add(new MetricMeasurement( + instrument.Name, + measurement, + tags.ToArray()))); + _listener.Start(); + } + + public ConcurrentBag> LongMeasurements { get; } = new(); + public ConcurrentBag> DoubleMeasurements { get; } = new(); + + public void Dispose() => _listener.Dispose(); + } + + private sealed record MetricMeasurement( + string Name, + T Value, + KeyValuePair[] Tags); +} diff --git a/tests/OpenClaw.Shared.Tests/McpHttpServerTests.cs b/tests/OpenClaw.Shared.Tests/McpHttpServerTests.cs index b7bf2a108..82b33ef15 100644 --- a/tests/OpenClaw.Shared.Tests/McpHttpServerTests.cs +++ b/tests/OpenClaw.Shared.Tests/McpHttpServerTests.cs @@ -10,6 +10,7 @@ using OpenClaw.Shared; using OpenClaw.Shared.Mcp; using OpenClaw.Shared.Telemetry; +using OpenClaw.Shared.Tests.Telemetry; using Xunit; namespace OpenClaw.Shared.Tests; @@ -19,6 +20,7 @@ namespace OpenClaw.Shared.Tests; /// boots the server on an ephemeral port so they can run in parallel and we /// don't collide with the production 8765. /// +[Collection(McpServerTelemetryCollection.Name)] public class McpHttpServerTests { private sealed class FakeCapability : INodeCapability diff --git a/tests/OpenClaw.Shared.Tests/Telemetry/McpServerTelemetryTests.cs b/tests/OpenClaw.Shared.Tests/Telemetry/McpServerTelemetryTests.cs new file mode 100644 index 000000000..e61e22b55 --- /dev/null +++ b/tests/OpenClaw.Shared.Tests/Telemetry/McpServerTelemetryTests.cs @@ -0,0 +1,265 @@ +using System.Collections.Concurrent; +using System.Diagnostics; +using System.Diagnostics.Metrics; +using OpenClaw.Shared.Telemetry; + +namespace OpenClaw.Shared.Tests.Telemetry; + +[CollectionDefinition(Name, DisableParallelization = true)] +public sealed class McpServerTelemetryCollection +{ + public const string Name = "MCP server telemetry"; +} + +[Collection(McpServerTelemetryCollection.Name)] +public sealed class McpServerTelemetryTests +{ + [Fact] + public void ConstantsAndFiniteValues_AreStable() + { + Assert.Equal("openclaw.mcp.server.start", McpServerTelemetry.StartSpanName); + Assert.Equal("openclaw.mcp.server.stop", McpServerTelemetry.StopSpanName); + Assert.Equal("openclaw.mcp.server.request", McpServerTelemetry.RequestSpanName); + Assert.Equal( + "openclaw.mcp.server.lifecycle.operations", + McpServerTelemetry.LifecycleOperationsMetricName); + Assert.Equal("openclaw.mcp.server.requests", McpServerTelemetry.RequestsMetricName); + Assert.Equal( + "openclaw.mcp.server.listener.errors", + McpServerTelemetry.ListenerErrorsMetricName); + Assert.Equal( + "openclaw.mcp.server.request.duration", + McpServerTelemetry.RequestDurationMetricName); + Assert.Equal("start", McpServerOperation.Start.ToTelemetryValue()); + Assert.Equal("stop", McpServerOperation.Stop.ToTelemetryValue()); + Assert.Equal("probe", McpServerRequestKind.Probe.ToTelemetryValue()); + Assert.Equal("json_rpc", McpServerRequestKind.JsonRpc.ToTelemetryValue()); + Assert.Equal("authentication_failed", McpServerErrorCategory.AuthenticationFailed.ToTelemetryValue()); + Assert.Equal("drain_timeout", McpServerErrorCategory.DrainTimeout.ToTelemetryValue()); + } + + [Fact] + public void FiniteValueMappings_RejectUnknownEnumValues() + { + Assert.Throws( + () => ((McpServerOperation)(-1)).ToTelemetryValue()); + Assert.Throws( + () => ((McpServerRequestKind)(-1)).ToTelemetryValue()); + Assert.Throws( + () => ((McpServerOutcome)(-1)).ToTelemetryValue()); + Assert.Throws( + () => ((McpServerErrorCategory)(-1)).ToTelemetryValue()); + } + + [Fact] + public void Lifecycle_CreatesDetachedRoot_AndCompletesExactlyOnce() + { + using var activities = new ActivityCollector(); + using var metrics = new MetricCollector(); + using var ambient = new Activity("ambient").Start(); + using var operation = McpServerTelemetry.StartLifecycle(McpServerOperation.Start); + + Assert.Same(ambient, Activity.Current); + + var completion = operation.Complete(McpServerOutcome.Success); + var duplicate = operation.Complete( + McpServerOutcome.Failure, + McpServerErrorCategory.InternalFailure); + + Assert.NotNull(completion); + Assert.Null(duplicate); + Assert.True(completion.DurationMilliseconds >= 0); + Assert.Same(ambient, Activity.Current); + + var activity = Assert.Single(activities.Stopped); + Assert.Equal(McpServerTelemetry.StartSpanName, activity.OperationName); + Assert.Equal(default, activity.ParentSpanId); + Assert.Equal(ActivityStatusCode.Ok, activity.Status); + Assert.Equal( + "start", + activity.GetTagItem(McpServerTelemetry.OperationTag)); + Assert.Equal( + "success", + activity.GetTagItem(OpenClawTelemetryTagKey.Outcome.ToTelemetryName())); + Assert.Null(activity.GetTagItem(OpenClawTelemetryTagKey.ErrorCategory.ToTelemetryName())); + + var measurement = Assert.Single( + metrics.LongMeasurements, + item => item.Name == McpServerTelemetry.LifecycleOperationsMetricName); + Assert.Equal(1, measurement.Value); + AssertTags( + measurement.Tags, + (McpServerTelemetry.OperationTag, "start"), + (OpenClawTelemetryTagKey.Outcome.ToTelemetryName(), "success"), + (OpenClawTelemetryTagKey.ErrorCategory.ToTelemetryName(), "none")); + } + + [Fact] + public void Request_RecordsDurationAndOnlyReviewedTags() + { + using var activities = new ActivityCollector(); + using var metrics = new MetricCollector(); + using var operation = McpServerTelemetry.StartRequest(McpServerRequestKind.JsonRpc); + + var completion = operation.Complete( + McpServerOutcome.Failure, + McpServerErrorCategory.AuthenticationFailed, + typeof(UnauthorizedAccessException)); + + Assert.NotNull(completion); + Assert.True(completion.DurationMilliseconds >= 0); + + var activity = Assert.Single(activities.Stopped); + Assert.Equal(System.Diagnostics.ActivityKind.Server, activity.Kind); + Assert.Equal(ActivityStatusCode.Error, activity.Status); + AssertTags( + activity.TagObjects.ToArray(), + (McpServerTelemetry.RequestKindTag, "json_rpc"), + (OpenClawTelemetryTagKey.Outcome.ToTelemetryName(), "failure"), + (OpenClawTelemetryTagKey.ErrorCategory.ToTelemetryName(), "authentication_failed"), + (OpenClawTelemetryTagKey.ErrorType.ToTelemetryName(), typeof(UnauthorizedAccessException).FullName)); + + var count = Assert.Single( + metrics.LongMeasurements, + item => item.Name == McpServerTelemetry.RequestsMetricName); + var duration = Assert.Single( + metrics.DoubleMeasurements, + item => item.Name == McpServerTelemetry.RequestDurationMetricName); + Assert.Equal(1, count.Value); + Assert.True(duration.Value >= 0); + AssertTags( + count.Tags, + (McpServerTelemetry.RequestKindTag, "json_rpc"), + (OpenClawTelemetryTagKey.Outcome.ToTelemetryName(), "failure"), + (OpenClawTelemetryTagKey.ErrorCategory.ToTelemetryName(), "authentication_failed")); + Assert.Equal( + count.Tags.OrderBy(tag => tag.Key), + duration.Tags.OrderBy(tag => tag.Key)); + Assert.DoesNotContain( + count.Tags, + tag => tag.Key == OpenClawTelemetryTagKey.ErrorType.ToTelemetryName()); + + var reviewedKeys = new HashSet(StringComparer.Ordinal) + { + McpServerTelemetry.RequestKindTag, + OpenClawTelemetryTagKey.Outcome.ToTelemetryName(), + OpenClawTelemetryTagKey.ErrorCategory.ToTelemetryName(), + OpenClawTelemetryTagKey.ErrorType.ToTelemetryName() + }; + Assert.All(activity.TagObjects, tag => Assert.Contains(tag.Key, reviewedKeys)); + Assert.All(count.Tags, tag => Assert.Contains(tag.Key, reviewedKeys)); + Assert.All(duration.Tags, tag => Assert.Contains(tag.Key, reviewedKeys)); + } + + [Fact] + public void ListenerError_RecordsOnlyFiniteCategory() + { + using var metrics = new MetricCollector(); + + McpServerTelemetry.RecordListenerError(McpServerErrorCategory.ListenerAccept); + + var measurement = Assert.Single( + metrics.LongMeasurements, + item => item.Name == McpServerTelemetry.ListenerErrorsMetricName); + AssertTags( + measurement.Tags, + (OpenClawTelemetryTagKey.ErrorCategory.ToTelemetryName(), "listener_accept")); + Assert.Throws( + () => McpServerTelemetry.RecordListenerError(McpServerErrorCategory.Timeout)); + } + + [Fact] + public void NoListeners_OperationsRemainBehaviorSafe() + { + using var lifecycle = McpServerTelemetry.StartLifecycle(McpServerOperation.Stop); + using var request = McpServerTelemetry.StartRequest(McpServerRequestKind.Other); + + Assert.NotNull(lifecycle.Complete(McpServerOutcome.Success)); + Assert.NotNull(request.Complete( + McpServerOutcome.Canceled, + McpServerErrorCategory.Shutdown)); + } + + private static void AssertTags( + IReadOnlyCollection> actual, + params (string Key, object? Value)[] expected) + { + Assert.Equal(expected.Length, actual.Count); + foreach (var (key, value) in expected) + { + Assert.Contains( + actual, + tag => tag.Key == key && Equals(tag.Value, value)); + } + } + + private sealed class ActivityCollector : IDisposable + { + private readonly ActivityListener _listener; + + public ActivityCollector() + { + _listener = new ActivityListener + { + ShouldListenTo = source => + source.Name == OpenClawActivitySourceName.OpenClaw.ToTelemetryName(), + Sample = (ref ActivityCreationOptions options) => + options.Name.StartsWith("openclaw.mcp.server.", StringComparison.Ordinal) + ? ActivitySamplingResult.AllDataAndRecorded + : ActivitySamplingResult.None, + ActivityStopped = activity => + { + if (activity.OperationName.StartsWith( + "openclaw.mcp.server.", + StringComparison.Ordinal)) + { + Stopped.Enqueue(activity); + } + } + }; + ActivitySource.AddActivityListener(_listener); + } + + public ConcurrentQueue Stopped { get; } = new(); + + public void Dispose() => _listener.Dispose(); + } + + private sealed class MetricCollector : IDisposable + { + private readonly MeterListener _listener = new(); + + public MetricCollector() + { + _listener.InstrumentPublished = (instrument, listener) => + { + if (instrument.Meter.Name == OpenClawMeterName.OpenClaw.ToTelemetryName() && + instrument.Name.StartsWith("openclaw.mcp.server.", StringComparison.Ordinal)) + { + listener.EnableMeasurementEvents(instrument); + } + }; + _listener.SetMeasurementEventCallback((instrument, measurement, tags, _) => + LongMeasurements.Add(new MetricMeasurement( + instrument.Name, + measurement, + tags.ToArray()))); + _listener.SetMeasurementEventCallback((instrument, measurement, tags, _) => + DoubleMeasurements.Add(new MetricMeasurement( + instrument.Name, + measurement, + tags.ToArray()))); + _listener.Start(); + } + + public ConcurrentBag> LongMeasurements { get; } = new(); + public ConcurrentBag> DoubleMeasurements { get; } = new(); + + public void Dispose() => _listener.Dispose(); + } + + private sealed record MetricMeasurement( + string Name, + T Value, + KeyValuePair[] Tags); +}