From abcc4f306ccb9b993749f7bca83f9950c0116dc3 Mon Sep 17 00:00:00 2001 From: Cory Hawkvelt Date: Tue, 18 Mar 2025 00:01:27 +1030 Subject: [PATCH] VM status updates when booted --- app/controller/api/workload_vm_routes.py | 65 +++++++++-- requirements.txt | 4 +- streamlit_server/views/virtual_machines.py | 26 ++++- websocket_server.py | 50 ++++---- worker_tasks/libvirt.py | 126 ++++++++++++++++++++- 5 files changed, 234 insertions(+), 37 deletions(-) diff --git a/app/controller/api/workload_vm_routes.py b/app/controller/api/workload_vm_routes.py index 0aa91c5..a81d022 100644 --- a/app/controller/api/workload_vm_routes.py +++ b/app/controller/api/workload_vm_routes.py @@ -250,7 +250,7 @@ def validate_payload(payload): return sanitized_payload @api_bp.route('/workloads/virtual_machines', methods=['POST']) -def add_VirtualMachine_workload1(): +def add_VirtualMachine_workload(): request_data = request.json logger.debug(request_data) @@ -357,13 +357,64 @@ def add_VirtualMachine_workload1(): @api_bp.route('/workloads/virtual_machines/status_update/', methods=['PUT']) def edit_VirtualMachine_workload(virtual_machine_id): data = request.json - new_status = data.get('new_status') + + # Validate new_status + if data.get("new_status") not in ["started", "deleted"]: + error_message = "Invalid status. Must be 'started' or 'deleted'." + logger.error(error_message) + return jsonify({"success": False, "message": error_message}), 400 + + # Validate system_vm_id and worker_id as UUIDs + try: + worker_id = uuid.UUID(data.get("worker_id")) + except (ValueError, TypeError) as e: + error_message = "Invalid UUID format for worker_id." + logger.error(f"{error_message} Error: {str(e)}") + return jsonify({"success": False, "message": error_message}), 400 + + # Validate virtual_machine_id as UUID + try: + virtual_machine_id = uuid.UUID(virtual_machine_id) + except (ValueError, TypeError) as e: + error_message = "Invalid UUID format for virtual_machine_id." + logger.error(f"{error_message} Error: {str(e)}") + return jsonify({"success": False, "message": error_message}), 400 + + # Validate timestamp as ISO format + try: + datetime.fromisoformat(data.get("timestamp")) + except (ValueError, TypeError) as e: + error_message = "Invalid ISO timestamp format." + logger.error(f"{error_message} Error: {str(e)}") + return jsonify({"success": False, "message": error_message}), 400 + + # Fetch the virtual machine from the database + try: + virtual_machine = Workload.query.filter_by(id=virtual_machine_id, workload_type="VirtualMachine").first_or_404() + except Exception as e: + error_message = f"Failed to fetch virtual machine with ID {virtual_machine_id}." + logger.error(f"{error_message} Error: {str(e)}") + return jsonify({"success": False, "message": error_message}), 404 + + # Check if the VM is actually on the reported worker + if virtual_machine.workload_host_id != worker_id: + error_message = f"Workload host in DB for VM ID {virtual_machine_id} is {virtual_machine.workload_host_id}, but was reported from {worker_id}. Ignoring." + logger.error(error_message) + return jsonify({"success": False, "message": error_message}), 400 + + # Update the status of the virtual machine + try: + new_status = data.get('new_status') + logger.info(f"Updating VirtualMachine ID: {virtual_machine_id} to new status {new_status}") + virtual_machine.status = new_status + db.session.commit() + return jsonify({"success": True, "message": "Status updated successfully."}), 200 + except Exception as e: + error_message = f"Failed to update status for VirtualMachine ID: {virtual_machine_id}." + logger.error(f"{error_message} Error: {str(e)}") + db.session.rollback() + return jsonify({"success": False, "message": error_message}), 500 - logger.info(f"Updating VirtualMachine ID:{virtual_machine_id} to new status {new_status}") - workload = Workload.query.filter_by(id=uuid.UUID(virtual_machine_id), workload_type="VirtualMachine").first_or_404() - workload.status = new_status - db.session.commit() - return jsonify(success=True) @api_bp.route('/workloads/virtual_machines/', methods=['GET']) def get_VirtualMachine_workload(workload_id): diff --git a/requirements.txt b/requirements.txt index 1f03eda..e6fa5cd 100644 --- a/requirements.txt +++ b/requirements.txt @@ -10,4 +10,6 @@ flask_sqlalchemy flask_migrate websocket-client streamlit -libvirt-python \ No newline at end of file +libvirt-python +colorlog +docker \ No newline at end of file diff --git a/streamlit_server/views/virtual_machines.py b/streamlit_server/views/virtual_machines.py index a3109e2..49c3e0d 100644 --- a/streamlit_server/views/virtual_machines.py +++ b/streamlit_server/views/virtual_machines.py @@ -132,7 +132,7 @@ def render(): st.session_state.delete_confirmation_shown = False st.session_state.delete_confirmed = False filtered_VirtualMachines=[] - + # Display the VMs for VirtualMachine in filtered_VirtualMachines: with st.container(): @@ -144,13 +144,33 @@ def render(): try: VirtualMachine_name = json.loads(launch_params).get('docker_image') except json.JSONDecodeError: - print("LP not vald json") + print("LP not valid json") VirtualMachine_name = None else: print("lp empty") VirtualMachine_name = None st.markdown(f"### {VirtualMachine['name']}") - st.text(f"Status: {VirtualMachine['status']}") + + # Create a grey box for the status + status = VirtualMachine['status'] + if status == "error": + status_color = "red" + elif status == "started": + status_color = "green" + elif status == "pending-allocated": + status_color = "orange" + else: + status_color = "blue" + + st.markdown( + f""" +
+ Status: {status} +
+ """, + unsafe_allow_html=True + ) + st.text(f"VDC ID: {VirtualMachine['vdc_id']}") st.text(f"Image: {VirtualMachine_name}") with col2: diff --git a/websocket_server.py b/websocket_server.py index f626be2..4adc062 100755 --- a/websocket_server.py +++ b/websocket_server.py @@ -505,7 +505,9 @@ def handle_libvirt_event(data): libvirt_vm_id = details.get("libvirt_vm_id") system_vm_id = details.get("system_vm_id") - timestamp = details.get("timestamp") + unix_timestamp = details.get("timestamp") + timestamp=datetime.utcfromtimestamp(unix_timestamp).isoformat() + # Log the extracted data logger.info(f"Received libvirt event from worker {worker_id}:") @@ -517,6 +519,28 @@ def handle_libvirt_event(data): # Optionally, log the full details for debugging purposes logger.debug(f"Full event details: {data}") + if event_type=="libvirt_started": + #Send the update to the API Server + # system_vm_id + # worker_id + # state="started" + # timestamp + URL=f"/workloads/virtual_machines/status_update/{system_vm_id}" + logger.debug(f"Sending task update payload to API server for VM ID {system_vm_id}") + + payload = { + "new_status": "started", + "system_vm_id": system_vm_id, + "worker_id": worker_id, + "timestamp": timestamp + } + + headers = {"Content-Type": "application/json"} + websocket_server_response = requests.put(f"{api_server_url}/workloads/virtual_machines/status_update/{system_vm_id}", data=json.dumps(payload), headers=headers) + logger.debug( websocket_server_response) + logger.info("Update complete") + + except Exception as e: logger.error(f"Error processing libvirt event: {e}") @@ -614,30 +638,6 @@ def create_task(): except Exception as e: logger.error(f"Error creating task: {e}") return jsonify({"error": "Failed to create task"}), 500 - - # # # Example request structure - # payload_text = { - # "worker_id": "234", - # "task_type": "container", - # "job_details": { - # "tenancyID": "tenant123a", - # "containers": [ - # { - # "container_id": "af47632e-43b7-4474-82fa-475a24fe88d5", - # "docker_image": "nginx", - # "cpu_shares": 1, - # "mem_limit": 128, - # "container_name": "web-server1", - # "command": ["nginx", "-g", "daemon off;"], - # "working_dir": "/usr/share/nginx/html" - # } - # ] - # } - # } - - - - if __name__ == "__main__": diff --git a/worker_tasks/libvirt.py b/worker_tasks/libvirt.py index f09fdbd..7bb5400 100644 --- a/worker_tasks/libvirt.py +++ b/worker_tasks/libvirt.py @@ -8,6 +8,12 @@ from logging import Logger import xml.etree.ElementTree as ET from xml.dom import minidom +import requests +import hashlib +import gzip +import shutil + + class LibvirtVirtualMachineTask: def __init__(self, params, config, logger): """ @@ -583,4 +589,122 @@ class LibvirtVirtualMachineTask: )) # Return True if there are differences - return len(diff) > 0 \ No newline at end of file + return len(diff) > 0 + + + + + # # Example usage + # json_input = ''' + # { + # "url": "https://example.com/path/to/image.jpg.gz", + # "md5": "d41d8cd98f00b204e9800998ecf8427e", + # "compressed": true + # } + # ''' + # logger.info("Starting process_image function...") + # result = process_image(json_input) + # logger.info(f"Result: {result}") + def _process_image(self, json_string): + # Parse the JSON input + self.logger.info("Processing image - Parsing JSON input...") + try: + data = json.loads(json_string) + url = data['url'] + expected_md5 = data['md5'] + is_compressed = data.get('compressed', False) # Default to False if not provided + self.logger.debug(f"URL: {url}, Expected MD5: {expected_md5}, Compressed: {is_compressed}") + except json.JSONDecodeError as e: + self.logger.error(f"Invalid JSON input: {e}") + return f"Error: Invalid JSON input. {e}" + except KeyError as e: + self.logger.error(f"Missing required key in JSON: {e}") + return f"Error: Missing required key in JSON. {e}" + + # Extract the final file name from the URL + file_name = os.path.basename(url) + if is_compressed: + file_name = file_name.rstrip('.gz') # Remove .gz extension if compressed + md5_file_name = f"{file_name}.md5" + self.logger.debug(f"File name: {file_name}, MD5 file name: {md5_file_name}") + + # Check if the image file exists locally + if os.path.exists(file_name): + self.logger.info(f"Image file '{file_name}' exists locally.") + # Check if the MD5 file exists + if os.path.exists(md5_file_name): + self.logger.info(f"MD5 file '{md5_file_name}' exists.") + with open(md5_file_name, 'r') as md5_file: + local_md5 = md5_file.read().strip() + self.logger.debug(f"Local MD5: {local_md5}, Expected MD5: {expected_md5}") + + # Compare the local MD5 with the expected MD5 + if local_md5 == expected_md5: + self.logger.info("MD5 matches. No further action needed.") + return True + else: + self.logger.warning("MD5 mismatch. Deleting local files and downloading again.") + os.remove(file_name) + os.remove(md5_file_name) + else: + self.logger.warning(f"MD5 file '{md5_file_name}' missing. Deleting local image and downloading again.") + os.remove(file_name) + else: + self.logger.info(f"Image file '{file_name}' does not exist locally. Downloading...") + + # Download the image from the URL using streaming + temp_file_name = f"{file_name}.tmp" + try: + self.logger.info(f"Downloading image from URL: {url}") + with requests.get(url, stream=True) as response: + response.raise_for_status() # Raise an error for bad status codes + with open(temp_file_name, 'wb') as temp_file: + for chunk in response.iter_content(chunk_size=8192): + temp_file.write(chunk) + self.logger.info(f"Download complete. Temporary file saved as '{temp_file_name}'.") + + # If the image is compressed, uncompress it + if is_compressed: + self.logger.info(f"Uncompressing the image '{temp_file_name}'...") + with gzip.open(temp_file_name, 'rb') as f_in: + with open(file_name, 'wb') as f_out: + shutil.copyfileobj(f_in, f_out) + self.logger.info(f"Uncompressed image saved as '{file_name}'.") + os.remove(temp_file_name) # Delete the temporary compressed file + else: + os.rename(temp_file_name, file_name) # Rename the temporary file to the final name + self.logger.info(f"Renamed temporary file '{temp_file_name}' to '{file_name}'.") + + except requests.exceptions.RequestException as e: + self.logger.error(f"Failed to download the image. Error: {e}") + if os.path.exists(temp_file_name): + os.remove(temp_file_name) + return f"Error: Failed to download the image. {e}" + + # Calculate the MD5 hash of the downloaded image + self.logger.info(f"Calculating MD5 hash for '{file_name}'...") + md5_hash = hashlib.md5() + with open(file_name, 'rb') as image_file: + for chunk in iter(lambda: image_file.read(4096), b""): + md5_hash.update(chunk) + actual_md5 = md5_hash.hexdigest() + self.logger.debug(f"Calculated MD5: {actual_md5}, Expected MD5: {expected_md5}") + + # Compare the calculated MD5 with the expected MD5 + if actual_md5 == expected_md5: + self.logger.info("MD5 matches. Saving MD5 hash to file...") + with open(md5_file_name, 'w') as md5_file: + md5_file.write(actual_md5) + self.logger.info(f"MD5 hash saved to '{md5_file_name}'.") + return True + else: + self.logger.error(f"MD5 mismatch. Expected {expected_md5}, got {actual_md5}.") + # Delete the downloaded image and MD5 file if they exist + if os.path.exists(file_name): + os.remove(file_name) + self.logger.info(f"Deleted file '{file_name}'.") + if os.path.exists(md5_file_name): + os.remove(md5_file_name) + self.logger.info(f"Deleted file '{md5_file_name}'.") + return f"Error: MD5 mismatch. Expected {expected_md5}, got {actual_md5}." + \ No newline at end of file