From 522e5c5634b51f5792e966e260d8dd65b5c596ed Mon Sep 17 00:00:00 2001 From: "Wiktor A. Falkowski" Date: Tue, 20 Oct 2026 19:03:15 +0000 Subject: [PATCH] fix(vps): report stable creation progress --- server_provisioning.py | 154 ++++++++++++++++++++++++++--------------- websocket_service.py | 129 ++++++++++++++++++++++++---------- 2 files changed, 193 insertions(+), 90 deletions(-) diff --git a/server_provisioning.py b/server_provisioning.py index 4fceec2..9db6688 100644 --- a/server_provisioning.py +++ b/server_provisioning.py @@ -786,13 +786,16 @@ def create_vps_server(config: Dict, user_id: str, server_id: str, namespace: str memory_request = f"{ram_gb * 0.5}Gi" if config['image'] == 'windows': - return create_windows_vps(config, user_id, server_id, namespace, image_config, ports, storage_size, cpu_cores, memory_limit, memory_request) + return create_windows_vps( + config, user_id, server_id, namespace, image_config, ports, + storage_size, cpu_cores, memory_limit, memory_request, progress_callback + ) # Create DataVolume data_volume_name = f"{server_id}-rootdisk" if progress_callback: - progress_callback('Initializing', 0, 'Preparing VPS creation...') + progress_callback('Preparing storage', 3, 'Preparing the VPS system disk.') data_volume_body = { 'apiVersion': 'cdi.kubevirt.io/v1beta1', @@ -978,11 +981,13 @@ final_message: "AnchorHost VPS ready! User: {image_config['user']}, Password: {i if dv_created and k8s_manager.connected: logger.info("⏳ Waiting for DataVolume to be ready...") if progress_callback: - progress_callback('Pending', 0, 'Creating DataVolume...') + progress_callback('Allocating storage', 5, 'Allocating storage for the system image.') max_wait = 600 - wait_interval = 5 # Check more frequently for better progress updates + wait_interval = 5 elapsed = 0 + latest_progress = 12 + image_ready = False while elapsed < max_wait: try: @@ -998,67 +1003,75 @@ final_message: "AnchorHost VPS ready! User: {image_config['user']}, Password: {i phase = status.get('phase', '') progress = status.get('progress', '') - # Extract progress percentage if available - progress_percent = 0 + # CDI reports the actual image download percentage. + progress_percent = None if progress: - # Progress format is usually "XX.X%" or "XX.XX%" try: progress_percent = float(progress.replace('%', '')) except (ValueError, AttributeError): pass - # Map phase to progress percentage if not available - if progress_percent == 0: - phase_progress_map = { - 'Pending': 5, - 'PVCBound': 10, - 'ImportScheduled': 15, - 'ImportInProgress': 50, - 'Succeeded': 100 - } - progress_percent = phase_progress_map.get(phase, 0) - - # Create progress message - phase_messages = { - 'Pending': 'Waiting for storage to be allocated...', - 'PVCBound': 'Storage allocated, preparing import...', - 'ImportScheduled': 'Import scheduled, starting download...', - 'ImportInProgress': f'Downloading image... ({progress if progress else "in progress"})', - 'Succeeded': 'Image download completed!', - 'Failed': 'Image download failed' + phase_progress_map = { + 'Pending': 5, + 'PVCBound': 9, + 'ImportScheduled': 12, + 'Succeeded': 84 } - message = phase_messages.get(phase, f'Processing... ({phase})') + if phase == 'ImportInProgress' and progress_percent is not None: + progress_percent = 12 + max(0, min(100, progress_percent)) * 0.72 + else: + progress_percent = phase_progress_map.get(phase, latest_progress) + latest_progress = max(latest_progress, int(progress_percent)) + + phase_messages = { + 'Pending': ('Allocating storage', 'Waiting for the storage volume.'), + 'PVCBound': ('Preparing image download', 'Storage is ready. Preparing the system image.'), + 'ImportScheduled': ('Starting image download', 'The system image download has been scheduled.'), + 'ImportInProgress': ('Downloading system image', f'Downloading the system image ({progress or "in progress"}).'), + 'Succeeded': ('System image ready', 'The system image is ready.') + } + progress_phase, message = phase_messages.get( + phase, ('Preparing system image', 'Waiting for the system image importer.') + ) - # Send progress update if progress_callback: - progress_callback(phase, progress_percent, message) + progress_callback(progress_phase, latest_progress, message) if phase == 'Succeeded': - logger.info("✅ DataVolume ready!") + logger.info("VPS system image is ready") + image_ready = True break elif phase == 'Failed': error_message = status.get('conditions', [{}])[0].get('message', 'Unknown error') if status.get('conditions') else 'Unknown error' - logger.error(f"❌ DataVolume failed: {error_message}") + logger.error(f"VPS system image import failed: {error_message}") if progress_callback: - progress_callback('Failed', 0, f'Image download failed: {error_message}') - break + progress_callback('Failed', latest_progress, f'System image download failed: {error_message}') + raise RuntimeError(f'System image download failed: {error_message}') except Exception as e: - logger.error(f"⚠ Error checking DataVolume: {e}") + if isinstance(e, RuntimeError): + raise + logger.warning(f"Could not read VPS image import status: {e}") if progress_callback: - progress_callback('Error', 0, f'Error checking progress: {str(e)}') + progress_callback( + 'Downloading system image', latest_progress, + 'Waiting for the image download status to update.' + ) + if image_ready: + break time.sleep(wait_interval) elapsed += wait_interval if elapsed >= max_wait: - logger.warning(f"⚠ DataVolume wait timeout after {max_wait}s") + logger.warning(f"VPS image import did not finish within {max_wait}s") if progress_callback: - progress_callback('Timeout', 0, 'Image download timed out, proceeding anyway...') + progress_callback('Failed', latest_progress, 'The system image download timed out.') + raise TimeoutError('System image download timed out; the virtual machine was not started.') # Create VM if progress_callback: - progress_callback('CreatingVM', 90, 'Creating virtual machine...') + progress_callback('Starting virtual machine', 88, 'Creating the virtual machine resource.') try: custom_objects_api.create_namespaced_custom_object( @@ -1068,15 +1081,16 @@ final_message: "AnchorHost VPS ready! User: {image_config['user']}, Password: {i plural="virtualmachines", body=vm_body ) - logger.info(f"✅ Created VirtualMachine {server_id}") - if progress_callback: - progress_callback('Completed', 100, 'VPS created successfully!') + logger.info(f"Created VirtualMachine {server_id}") except Exception as e: logger.error(f"❌ Failed to create VM: {e}") if progress_callback: progress_callback('Error', 0, f'Failed to create VM: {str(e)}') raise Exception(f"Failed to create VM: {str(e)}") + if progress_callback: + progress_callback('Configuring network', 92, 'Setting up the VPS network and ports.') + # Create Service with FIXED port handling if ports: service_ports = [] @@ -1123,17 +1137,14 @@ final_message: "AnchorHost VPS ready! User: {image_config['user']}, Password: {i ports.extend(actual_ports) except Exception as e: - logger.error(f"⚠ Failed to create service: {e}") - - # Background status sync - if k8s_manager.connected: - threading.Thread( - target=wait_and_sync_vm_status, - args=(server_id, user_id, namespace), - daemon=True - ).start() + logger.error(f"Failed to create VPS network service: {e}") + if progress_callback: + progress_callback('Failed', 92, 'Could not create the VPS network service.') + raise RuntimeError(f'Failed to create VPS network service: {e}') from e # Create server record + if progress_callback: + progress_callback('Registering VPS', 97, 'Saving the VPS to your server list.') server_data = { 'id': server_id, 'name': config['name'], @@ -1151,7 +1162,18 @@ final_message: "AnchorHost VPS ready! User: {image_config['user']}, Password: {i 'k8s_name': server_id } - return create_server_record(server_data) + server = create_server_record(server_data) + if not server: + raise RuntimeError('VPS resources were created, but the server record could not be saved.') + + if k8s_manager.connected: + threading.Thread( + target=wait_and_sync_vm_status, + args=(server_id, user_id, namespace), + daemon=True + ).start() + + return server def wait_and_sync_vm_status(server_id: str, user_id: str, namespace: str): """Background VM status sync""" @@ -1240,9 +1262,12 @@ def wait_and_sync_pod_status(server_id: str, user_id: str, namespace: str): sync_server_status_from_k8s(server_id, user_id) def create_windows_vps(config: Dict, user_id: str, server_id: str, namespace: str, image_config: Dict, - ports: List, storage_size: str, cpu_cores: int, memory_limit: str, memory_request: str) -> Dict: + ports: List, storage_size: str, cpu_cores: int, memory_limit: str, memory_request: str, + progress_callback=None) -> Dict: """Create Windows VPS""" pvc_name = f"{server_id}-rootdisk" + if progress_callback: + progress_callback('Preparing storage', 5, 'Preparing storage for the Windows VPS.') if not k8s_manager.create_pvc(namespace, pvc_name, storage_size): raise Exception("Failed to create PVC") @@ -1281,6 +1306,8 @@ def create_windows_vps(config: Dict, user_id: str, server_id: str, namespace: st } } + if progress_callback: + progress_callback('Starting virtual machine', 88, 'Creating the Windows virtual machine resource.') custom_objects_api.create_namespaced_custom_object( group="kubevirt.io", version="v1", @@ -1289,6 +1316,9 @@ def create_windows_vps(config: Dict, user_id: str, server_id: str, namespace: st body=vm_body ) + if progress_callback: + progress_callback('Configuring network', 92, 'Setting up the Windows VPS network and ports.') + # Create Service with FIXED port handling if ports: service_ports = [] @@ -1334,7 +1364,10 @@ def create_windows_vps(config: Dict, user_id: str, server_id: str, namespace: st ports.extend(actual_ports) except Exception as e: - logger.error(f"⚠ Failed to create service: {e}") + logger.error(f"Failed to create Windows VPS network service: {e}") + if progress_callback: + progress_callback('Failed', 92, 'Could not create the VPS network service.') + raise RuntimeError(f'Failed to create Windows VPS network service: {e}') from e server_data = { 'id': server_id, @@ -1353,7 +1386,20 @@ def create_windows_vps(config: Dict, user_id: str, server_id: str, namespace: st 'k8s_name': server_id } - return create_server_record(server_data) + if progress_callback: + progress_callback('Registering VPS', 97, 'Saving the VPS to your server list.') + server = create_server_record(server_data) + if not server: + raise RuntimeError('VPS resources were created, but the server record could not be saved.') + + if k8s_manager.connected: + threading.Thread( + target=wait_and_sync_vm_status, + args=(server_id, user_id, namespace), + daemon=True + ).start() + + return server def create_website_server(config: Dict, user_id: str, server_id: str, namespace: str) -> Dict: """Create Website Server with FIXED port handling""" diff --git a/websocket_service.py b/websocket_service.py index c32d2e7..a3cbda3 100644 --- a/websocket_service.py +++ b/websocket_service.py @@ -48,6 +48,17 @@ import websockets __all__ = ['DateTimeEncoder', 'attach_to_game_console', 'ensure_node_shell_pod', 'execute_minecraft_save_all', 'handle_2fa_disable', 'handle_2fa_enable', 'handle_2fa_login', 'handle_2fa_setup', 'handle_add_credits', 'handle_authenticate', 'handle_change_server_version', 'handle_check_demo_eligibility', 'handle_convert_demo_to_paid', 'handle_copy_path', 'handle_create_backup', 'handle_create_custom_domain', 'handle_create_directory', 'handle_create_domain', 'handle_create_server', 'handle_create_ticket', 'handle_delete_backup', 'handle_delete_custom_domain', 'handle_delete_domain', 'handle_delete_path', 'handle_delete_server', 'handle_download_file', 'handle_exec_command', 'handle_expose_port', 'handle_game_console_command', 'handle_game_console_stream', 'handle_generate_sftp', 'handle_get_backups', 'handle_get_credits', 'handle_get_minecraft_settings', 'handle_get_server_creation_progress', 'handle_get_server_logs', 'handle_get_server_metrics', 'handle_get_servers', 'handle_get_ticket', 'handle_get_ticket_stats', 'handle_get_tickets', 'handle_link_domain', 'handle_list_custom_domains', 'handle_list_domains', 'handle_list_files', 'handle_login', 'handle_move_path', 'handle_node_terminal_connect', 'handle_node_terminal_disconnect', 'handle_node_terminal_input', 'handle_pack_zip', 'handle_read_file', 'handle_register', 'handle_remove_port', 'handle_restart_server', 'handle_restore_backup', 'handle_send_ticket_message', 'handle_start_chunked_upload', 'handle_start_server', 'handle_stop_server', 'handle_sync_server_status', 'handle_unlink_domain', 'handle_unpack_zip', 'handle_update_minecraft_settings', 'handle_update_ticket_status', 'handle_upload_file', 'handle_upload_file_chunk', 'handle_verify_custom_domain', 'handle_vps_terminal_connect', 'handle_vps_terminal_disconnect', 'handle_vps_terminal_input', 'handle_websocket', 'handle_write_file', 'notify_admins', 'notify_ticket_participants', 'parse_ls_output', 'send_message', 'send_verification_email', 'start_websocket_server', 'stream_game_console_output', 'stream_metrics', 'sync_user_domains_with_dns'] +_server_creation_progress_by_user: Dict[str, Dict] = {} +_server_creation_progress_lock = threading.Lock() + + +def _set_server_creation_progress(user_id: str, progress: Dict) -> Dict: + """Keep the latest creation status available after a WebSocket reconnect.""" + snapshot = {**progress, 'timestamp': time.time()} + with _server_creation_progress_lock: + _server_creation_progress_by_user[str(user_id)] = snapshot + return snapshot + async def handle_create_server(client_id, data): @@ -58,6 +69,14 @@ async def handle_create_server(client_id, data): return user_id = client_info['user_id'] + server_type = data.get('serverType', '') + _set_server_creation_progress(user_id, { + 'phase': 'Preparing', + 'progress': 1, + 'message': 'Checking capacity and preparing your server.', + 'serverType': server_type, + 'state': 'active' + }) try: # Check if this is a demo server request @@ -67,6 +86,18 @@ async def handle_create_server(client_id, data): server_name = data.get('name', '').strip() is_valid, error_msg = validate_server_name(server_name) if not is_valid: + failed = _set_server_creation_progress(user_id, { + 'phase': 'Failed', + 'progress': 1, + 'message': error_msg, + 'serverType': server_type, + 'state': 'failed' + }) + await send_message(client_id, {'type': 'server_creation_progress', **failed}) + await send_message(client_id, { + 'type': 'server_creation_failed', + 'error': error_msg + }) await send_message(client_id, { 'type': 'error', 'error': error_msg @@ -80,7 +111,15 @@ async def handle_create_server(client_id, data): capacity_check = validate_server_capacity(data) if not capacity_check['success']: + failed = _set_server_creation_progress(user_id, { + 'phase': 'Capacity check', + 'progress': 1, + 'message': capacity_check['message'], + 'serverType': server_type, + 'state': 'failed' + }) # Send detailed capacity error + await send_message(client_id, {'type': 'server_creation_progress', **failed}) await send_message(client_id, { 'type': 'capacity_error', 'error': capacity_check['message'], @@ -90,43 +129,48 @@ async def handle_create_server(client_id, data): }) return - # Capacity check passed, proceed with server creation - # Create progress callback that sends websocket messages + # Capacity check passed, proceed with server creation. loop = asyncio.get_event_loop() def progress_callback(phase, progress, message): - """Thread-safe progress callback that sends websocket messages""" + """Store and forward progress updates from the provisioning thread.""" try: - # Store progress in client info for reconnection recovery - if client_id in connected_clients: - connected_clients[client_id]['server_creation_progress'] = { - 'phase': phase, - 'progress': progress, - 'message': message, - 'timestamp': time.time() - } + with _server_creation_progress_lock: + previous = dict(_server_creation_progress_by_user.get(str(user_id), {})) + state = 'failed' if phase in ('Failed', 'Error', 'Timeout') else 'active' + snapshot = _set_server_creation_progress(user_id, { + 'phase': phase, + 'progress': max(int(previous.get('progress', 0)), max(0, min(99, int(progress)))), + 'message': message, + 'serverType': server_type, + 'state': state + }) - # Use run_coroutine_threadsafe to send async message from executor thread future = asyncio.run_coroutine_threadsafe( send_message(client_id, { 'type': 'server_creation_progress', - 'phase': phase, - 'progress': progress, - 'message': message + **snapshot }), loop ) - # Don't wait for completion to avoid blocking - # The future will complete asynchronously + future.result(timeout=2) except Exception as e: logger.error(f"Error sending progress update: {e}") - # Run blocking create_server in executor with progress callback + # Run blocking provisioning outside the event loop. server = await loop.run_in_executor(None, create_server, data, user_id, progress_callback) - # Clear server creation progress on completion - if client_id in connected_clients: - connected_clients[client_id].pop('server_creation_progress', None) + completed = _set_server_creation_progress(user_id, { + 'phase': 'Provisioned', + 'progress': 100, + 'message': 'Server resources are ready. The VPS may take a little longer to boot.', + 'serverType': server_type, + 'state': 'complete' + }) + await send_message(client_id, { + 'type': 'server_creation_progress', + **completed + }) # Get user credits user = get_user_by_id(user_id) @@ -145,10 +189,22 @@ async def handle_create_server(client_id, data): }) except Exception as e: logger.error(f"Error creating server: {e}", exc_info=True) - - # Clear server creation progress on error - if client_id in connected_clients: - connected_clients[client_id].pop('server_creation_progress', None) + previous = _server_creation_progress_by_user.get(str(user_id), {}) + failed = _set_server_creation_progress(user_id, { + 'phase': 'Failed', + 'progress': previous.get('progress', 0), + 'message': str(e), + 'serverType': data.get('serverType', ''), + 'state': 'failed' + }) + await send_message(client_id, { + 'type': 'server_creation_progress', + **failed + }) + await send_message(client_id, { + 'type': 'server_creation_failed', + 'error': str(e) + }) # Check if it's a capacity-related error error_msg = str(e) @@ -1269,25 +1325,25 @@ async def handle_get_server_creation_progress(client_id, data): await send_message(client_id, {'type': 'error', 'error': 'Not authenticated'}) return - # Get stored progress if available - progress = client_info.get('server_creation_progress') + user_id = str(client_info['user_id']) + with _server_creation_progress_lock: + progress = _server_creation_progress_by_user.get(user_id) if progress: - # Check if progress is recent (within last 5 minutes) - if time.time() - progress.get('timestamp', 0) < 300: + expiry = 1800 if progress.get('state') == 'active' else 600 + if time.time() - progress.get('timestamp', 0) < expiry: await send_message(client_id, { 'type': 'server_creation_progress', - 'phase': progress.get('phase', ''), - 'progress': progress.get('progress', 0), - 'message': progress.get('message', 'Creating server...') + **progress }) else: - # Progress is stale, clear it - client_info.pop('server_creation_progress', None) + with _server_creation_progress_lock: + _server_creation_progress_by_user.pop(user_id, None) await send_message(client_id, { 'type': 'server_creation_progress', 'phase': 'Unknown', 'progress': 0, - 'message': 'No active server creation found' + 'message': 'No recent server creation status is available.', + 'state': 'unknown' }) else: # No progress stored @@ -1295,7 +1351,8 @@ async def handle_get_server_creation_progress(client_id, data): 'type': 'server_creation_progress', 'phase': 'Unknown', 'progress': 0, - 'message': 'No active server creation found' + 'message': 'No recent server creation status is available.', + 'state': 'unknown' }) async def handle_create_domain(client_id, data):