From aba5fb12faf9c911372268bab20cc4503982317a Mon Sep 17 00:00:00 2001 From: Dobromir Popov Date: Tue, 7 Jul 2026 18:12:57 +0200 Subject: [PATCH] Log node request processing so operators can see live activity in the console. --- packages/node/meshnet_node/relay_bridge.py | 3 ++ packages/node/meshnet_node/torch_server.py | 55 ++++++++++++++++++++-- packages/tracker/meshnet_tracker/server.py | 2 + tests/test_real_model_backend.py | 11 ++++- 4 files changed, 65 insertions(+), 6 deletions(-) diff --git a/packages/node/meshnet_node/relay_bridge.py b/packages/node/meshnet_node/relay_bridge.py index 721c7f1..a66a513 100644 --- a/packages/node/meshnet_node/relay_bridge.py +++ b/packages/node/meshnet_node/relay_bridge.py @@ -164,6 +164,9 @@ class RelayHttpBridge: path = str(payload.get("path") or "/") headers = payload.get("headers") if isinstance(payload.get("headers"), dict) else {} + req_suffix = f" request_id={request_id}" if request_id else "" + print(f" [node] relay {method} {path}{req_suffix}", flush=True) + # body_base64 carries binary data (e.g. bfloat16 activation tensors) safely. # Fallback to text "body" for backward-compat with non-binary requests. body_b64 = payload.get("body_base64") diff --git a/packages/node/meshnet_node/torch_server.py b/packages/node/meshnet_node/torch_server.py index 398ad6e..7d615ec 100644 --- a/packages/node/meshnet_node/torch_server.py +++ b/packages/node/meshnet_node/torch_server.py @@ -113,6 +113,10 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler): def log_message(self, fmt, *args): # noqa: suppress request logs in tests pass + def _request_log_suffix(self) -> str: + req_id = self.headers.get("X-Meshnet-Request-Id") or self.headers.get("X-Request-Id") + return f" request_id={req_id}" if req_id else "" + def do_POST(self): server: _TorchHTTPServer = self.server # type: ignore[assignment] if self.path == "/forward": @@ -199,8 +203,18 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler): return server.forward_chunk_count += 1 - if int(self.headers.get("X-Meshnet-Hop-Index", "0")) > 0: + hop_index = int(self.headers.get("X-Meshnet-Hop-Index", "0")) + if hop_index > 0: server.received_activations = True + if chunk_index_value == 0: + shard_start = getattr(server.backend, "shard_start", "?") + shard_end = getattr(server.backend, "shard_end", "?") + print( + f" [node] forward hop={hop_index} " + f"layers={shard_start}-{shard_end} " + f"session={session[:8]}{self._request_log_suffix()}", + flush=True, + ) start_layer_header = self.headers.get("X-Meshnet-Start-Layer") start_layer = int(start_layer_header) if start_layer_header else None @@ -311,20 +325,53 @@ class _TorchHandler(http.server.BaseHTTPRequestHandler): temperature = float(body.get("temperature") or 1.0) top_p = float(body.get("top_p") or 1.0) + print( + f" [node] processing chat model={model_name!r} stream={stream} " + f"max_tokens={max_tokens}{self._request_log_suffix()}", + flush=True, + ) + # Fast path: this node owns the complete model — use HF generate() with KV cache. # Avoids the single-token-per-forward-pass limitation of the distributed path. if backend.is_head and backend.is_tail: + gen_started = time.monotonic() try: if stream: - self._stream_openai_response( - backend.generate_text_streaming(messages, max_tokens, temperature, top_p), - model_name, + token_count = 0 + + def _counting_stream(): + nonlocal token_count + for token_text in backend.generate_text_streaming( + messages, max_tokens, temperature, top_p, + ): + if token_text: + token_count += 1 + yield token_text + + self._stream_openai_response(_counting_stream(), model_name) + print( + f" [node] chat complete (stream) tokens={token_count} " + f"elapsed_s={time.monotonic() - gen_started:.1f}{self._request_log_suffix()}", + flush=True, ) else: text = backend.generate_text(messages, max_tokens, temperature, top_p) + completion_tokens = _backend_token_count( + backend, "count_text_tokens", text, fallback=len(text.split()) or 1, + ) + print( + f" [node] chat complete tokens={completion_tokens} " + f"elapsed_s={time.monotonic() - gen_started:.1f}{self._request_log_suffix()}", + flush=True, + ) self._send_openai_response(text, model_name, False, messages, backend=backend) except Exception as exc: self._record_failed_request() + print( + f" [node] chat failed after {time.monotonic() - gen_started:.1f}s: {exc}" + f"{self._request_log_suffix()}", + flush=True, + ) self._send_json(500, {"error": f"generation failed: {exc}"}) return diff --git a/packages/tracker/meshnet_tracker/server.py b/packages/tracker/meshnet_tracker/server.py index 1d7fbb7..1ea498d 100644 --- a/packages/tracker/meshnet_tracker/server.py +++ b/packages/tracker/meshnet_tracker/server.py @@ -3035,6 +3035,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): headers={ "Content-Type": "application/json", "X-Meshnet-Route": downstream_urls, + "X-Meshnet-Request-Id": request_id, }, method="POST", ) @@ -3046,6 +3047,7 @@ class _TrackerHandler(http.server.BaseHTTPRequestHandler): relay_headers = { "Content-Type": "application/json", "X-Meshnet-Route": downstream_urls, + "X-Meshnet-Request-Id": request_id, **({"Authorization": auth} if auth else {}), } diff --git a/tests/test_real_model_backend.py b/tests/test_real_model_backend.py index 4e4928b..656c55f 100644 --- a/tests/test_real_model_backend.py +++ b/tests/test_real_model_backend.py @@ -225,7 +225,7 @@ def test_tail_forward_returns_text_completion_from_binary_activations(): node.stop() -def test_full_model_chat_completion_uses_generation_not_single_token_decode(): +def test_full_model_chat_completion_uses_generation_not_single_token_decode(capsys): node = TorchNodeServer(backend=_FakeFullBackend()) port = node.start() try: @@ -237,7 +237,10 @@ def test_full_model_chat_completion_uses_generation_not_single_token_decode(): req = urllib.request.Request( f"http://127.0.0.1:{port}/v1/chat/completions", data=payload, - headers={"Content-Type": "application/json"}, + headers={ + "Content-Type": "application/json", + "X-Meshnet-Request-Id": "req-test-123", + }, method="POST", ) with urllib.request.urlopen(req, timeout=5) as resp: @@ -248,6 +251,10 @@ def test_full_model_chat_completion_uses_generation_not_single_token_decode(): finally: node.stop() + out = capsys.readouterr().out + assert " [node] processing chat model='fake-model' stream=False max_tokens=7 request_id=req-test-123" in out + assert " [node] chat complete tokens=1 elapsed_s=" in out + def test_pipeline_hop_logs_are_suppressed_without_debug(capsys): tail_backend = _FakePipelineTailBackend()