22
33import asyncio
44import json
5+ import logging
6+ import os
57import shutil
68import signal
79import sys
1921from centml .sdk .api import get_centml_client
2022from centml .sdk .config import settings
2123
24+ _log = logging .getLogger ("centml.cli.shell" )
25+
26+
27+ def _setup_debug_log ():
28+ """Configure file-based debug logging (stdout unusable in raw mode)."""
29+ log_path = os .environ .get (
30+ "CENTML_SHELL_DEBUG_LOG" , "/tmp/centml_shell_debug.log"
31+ )
32+ handler = logging .FileHandler (log_path , mode = "w" )
33+ handler .setFormatter (
34+ logging .Formatter ("%(asctime)s.%(msecs)03d %(message)s" , datefmt = "%H:%M:%S" )
35+ )
36+ _log .addHandler (handler )
37+ _log .setLevel (logging .DEBUG )
38+ _log .debug ("=== shell debug log started (pid=%d) ===" , os .getpid ())
39+
2240
2341# ---------------------------------------------------------------------------
2442# pyte screen renderer -- converts pyte's in-memory screen buffer to ANSI
@@ -224,36 +242,58 @@ async def _forward_io(ws, screen, stream, shutdown):
224242
225243 async def _read_ws ():
226244 nonlocal exit_code , should_reconnect
245+ _log .debug ("[read_ws] started" )
246+ msg_count = 0
227247 try :
228248 async for raw_msg in ws :
249+ msg_count += 1
229250 msg = json .loads (raw_msg )
251+ keys = list (msg .keys ())
252+ data_snippet = ""
253+ if msg .get ("data" ):
254+ data_snippet = repr (msg ["data" ][:120 ])
255+ _log .debug (
256+ "[read_ws] msg#%d keys=%s data=%s" ,
257+ msg_count , keys , data_snippet ,
258+ )
230259 if msg .get ("data" ):
231260 # pyte expects \r\n; remote PTY may send bare \n
232261 # (same as xterm.js ``convertEol: true``).
233262 stream .feed (msg ["data" ].replace ("\n " , "\r \n " ))
234263 _render_dirty (screen , sys .stdout .buffer )
235264 elif msg .get ("error" ):
265+ _log .debug ("[read_ws] error: %s" , msg ["error" ])
236266 stream .feed (f"Error: { msg ['error' ]} \r \n " )
237267 _render_dirty (screen , sys .stdout .buffer )
238268 # ArgoCD sends {"Code": ...} as a reconnect signal (token
239269 # refresh), not a shell exit code. Mirror ArgoCD UI behavior:
240270 # disconnect and reconnect with a fresh token.
241271 if "Code" in msg :
272+ _log .debug (
273+ "[read_ws] got Code=%s, setting should_reconnect=True" ,
274+ msg ["Code" ],
275+ )
242276 should_reconnect = True
243277 return
244- except websockets .ConnectionClosed :
278+ _log .debug ("[read_ws] ws iterator exhausted after %d msgs" , msg_count )
279+ except websockets .ConnectionClosed as exc :
280+ _log .debug (
281+ "[read_ws] ConnectionClosed after %d msgs: %s" , msg_count , exc ,
282+ )
245283 # Backend proxy may not send a clean close frame when
246284 # ArgoCD disconnects after the remote shell exits.
247285 return
248286
249287 async def _read_stdin ():
288+ _log .debug ("[read_stdin] started" )
250289 read_queue = asyncio .Queue ()
251290
252291 def _on_stdin_ready ():
253292 data = sys .stdin .buffer .read1 (4096 )
254293 if data :
255294 read_queue .put_nowait (data )
256295 else :
296+ _log .debug ("[read_stdin] stdin EOF" )
257297 stdin_closed .set ()
258298
259299 loop .add_reader (stdin_fd , _on_stdin_ready )
@@ -263,6 +303,7 @@ def _on_stdin_ready():
263303 data = await asyncio .wait_for (read_queue .get (), timeout = 0.5 )
264304 except asyncio .TimeoutError :
265305 continue
306+ _log .debug ("[read_stdin] sending %d bytes: %s" , len (data ), repr (data [:40 ]))
266307 try :
267308 await ws .send (
268309 json .dumps (
@@ -275,20 +316,31 @@ def _on_stdin_ready():
275316 )
276317 )
277318 except websockets .ConnectionClosed :
319+ _log .debug ("[read_stdin] ws closed on send" )
278320 return
321+ _log .debug (
322+ "[read_stdin] loop exited: stdin_closed=%s shutdown=%s" ,
323+ stdin_closed .is_set (), shutdown .is_set (),
324+ )
279325 finally :
280326 loop .remove_reader (stdin_fd )
281327
282328 async def _watch_shutdown ():
283329 while not shutdown .is_set ():
284330 await asyncio .sleep (0.2 )
285331
286- tasks = [
287- asyncio .create_task (_read_ws ()),
288- asyncio .create_task (_read_stdin ()),
289- asyncio .create_task (_watch_shutdown ()),
290- ]
332+ _log .debug ("[forward_io] creating tasks" )
333+ task_ws = asyncio .create_task (_read_ws ())
334+ task_stdin = asyncio .create_task (_read_stdin ())
335+ task_shutdown = asyncio .create_task (_watch_shutdown ())
336+ tasks = [task_ws , task_stdin , task_shutdown ]
337+ task_names = {id (task_ws ): "read_ws" , id (task_stdin ): "read_stdin" , id (task_shutdown ): "watch_shutdown" }
338+
291339 done , pending = await asyncio .wait (tasks , return_when = asyncio .FIRST_COMPLETED )
340+ done_names = [task_names [id (t )] for t in done ]
341+ pending_names = [task_names [id (t )] for t in pending ]
342+ _log .debug ("[forward_io] first completed: done=%s pending=%s" , done_names , pending_names )
343+
292344 for t in pending :
293345 t .cancel ()
294346 for t in pending :
@@ -298,7 +350,12 @@ async def _watch_shutdown():
298350 pass
299351 for t in done :
300352 if t .exception () is not None :
353+ _log .debug ("[forward_io] task exception: %s" , t .exception ())
301354 raise t .exception ()
355+ _log .debug (
356+ "[forward_io] returning exit_code=%d should_reconnect=%s" ,
357+ exit_code , should_reconnect ,
358+ )
302359 return (exit_code , should_reconnect )
303360
304361
@@ -350,12 +407,17 @@ def _send_resize():
350407
351408 try :
352409 last_code_time = - 999.0
410+ attempt = 0
353411 while True :
412+ attempt += 1
413+ _log .debug ("[session] attempt #%d, fetching token" , attempt )
354414 token = get_token_fn ()
355415 headers = {"Authorization" : f"Bearer { token } " }
416+ _log .debug ("[session] connecting to %s" , ws_url )
356417 async with websockets .connect (
357418 ws_url , additional_headers = headers , close_timeout = 2
358419 ) as ws :
420+ _log .debug ("[session] connected, sending resize %dx%d" , cols , rows )
359421 _ws_ref [0 ] = ws
360422 await ws .send (
361423 json .dumps (
@@ -366,15 +428,27 @@ def _send_resize():
366428 ws , screen , stream , shutdown
367429 )
368430 _ws_ref [0 ] = None
431+ _log .debug (
432+ "[session] forward_io returned exit_code=%d reconnect=%s shutdown=%s" ,
433+ exit_code , should_reconnect , shutdown .is_set (),
434+ )
369435 if not should_reconnect or shutdown .is_set ():
436+ _log .debug ("[session] exiting with code %d" , exit_code )
370437 return exit_code
371438 # If we got two Code signals within 3 seconds the shell
372439 # has genuinely exited (reconnect just opened a new
373440 # session that immediately closed). Stop reconnecting.
374441 now = time .monotonic ()
375- if now - last_code_time < 3.0 :
442+ gap = now - last_code_time
443+ _log .debug (
444+ "[session] reconnect: gap=%.2fs last_code_time=%.2f now=%.2f" ,
445+ gap , last_code_time , now ,
446+ )
447+ if gap < 3.0 :
448+ _log .debug ("[session] rapid Code, exiting with 0" )
376449 return 0
377450 last_code_time = now
451+ _log .debug ("[session] will reconnect" )
378452 finally :
379453 loop .remove_signal_handler (signal .SIGWINCH )
380454 loop .remove_signal_handler (signal .SIGTERM )
@@ -444,9 +518,16 @@ async def _exec_session(ws_url, token, command):
444518 buffer = ""
445519 is_capturing = False
446520 is_done = False
521+ msg_count = 0
447522 try :
448523 async for raw_msg in ws :
524+ msg_count += 1
449525 msg = json .loads (raw_msg )
526+ keys = list (msg .keys ())
527+ _log .debug (
528+ "[exec] msg#%d keys=%s data_len=%d" ,
529+ msg_count , keys , len (msg .get ("data" , "" )),
530+ )
450531 if msg .get ("data" ):
451532 buffer += msg ["data" ]
452533 while "\n " in buffer :
@@ -455,6 +536,7 @@ async def _exec_session(ws_url, token, command):
455536 line_stream , line_screen , line .rstrip ("\r " )
456537 )
457538 if _BEGIN_MARKER in clean :
539+ _log .debug ("[exec] BEGIN marker found" )
458540 is_capturing = True
459541 continue
460542 if _END_MARKER in clean :
@@ -464,21 +546,26 @@ async def _exec_session(ws_url, token, command):
464546 exit_code = int (parts [1 ].strip ())
465547 except ValueError :
466548 pass
549+ _log .debug ("[exec] END marker, exit_code=%d" , exit_code )
467550 is_done = True
468551 break
469552 if is_capturing :
470553 sys .stdout .write (line + "\n " )
471554 sys .stdout .flush ()
472555 elif msg .get ("error" ):
556+ _log .debug ("[exec] error: %s" , msg ["error" ])
473557 sys .stderr .write (f"Error: { msg ['error' ]} \n " )
474558 return 1
475559 if "Code" in msg and not is_done :
560+ _log .debug ("[exec] Code=%s before done, returning 1" , msg ["Code" ])
476561 sys .stderr .write ("Connection interrupted, please retry.\n " )
477562 return 1
478563 if is_done :
564+ _log .debug ("[exec] done, breaking" )
479565 break
480- except websockets .ConnectionClosed :
481- pass
566+ except websockets .ConnectionClosed as exc :
567+ _log .debug ("[exec] ConnectionClosed: %s" , exc )
568+ _log .debug ("[exec] returning exit_code=%d" , exit_code )
482569 return exit_code
483570
484571
@@ -496,6 +583,7 @@ async def _exec_session(ws_url, token, command):
496583)
497584@handle_exception
498585def shell (deployment_id , pod , shell_type ):
586+ _setup_debug_log ()
499587 if not sys .stdin .isatty ():
500588 raise click .ClickException ("Interactive shell requires a terminal (TTY)" )
501589
@@ -525,6 +613,7 @@ def shell(deployment_id, pod, shell_type):
525613)
526614@handle_exception
527615def exec_cmd (deployment_id , command , pod , shell_type ):
616+ _setup_debug_log ()
528617 with get_centml_client () as cclient :
529618 pod_name = _resolve_pod (cclient , deployment_id , pod )
530619
0 commit comments