diff --git a/app/controller/api/workload_container_routes.py b/app/controller/api/workload_container_routes.py index 1ac87a1..f449f53 100644 --- a/app/controller/api/workload_container_routes.py +++ b/app/controller/api/workload_container_routes.py @@ -334,7 +334,7 @@ def update_container_workload(system_container_id): data = request.json logger.debug(data) # Validate new_status - valid_statuses = ["running", "deleted", "stopped", "dead", "launch_failed"] + valid_statuses = ["running", "deleted", "stopped", "dead", "launch_failed", "pulling"] new_status = data.get("new_status") if new_status not in valid_statuses: error_message = f"Invalid status {new_status}. Must be one of: {', '.join(valid_statuses)}" @@ -431,7 +431,37 @@ def update_container_workload(system_container_id): try: new_status = data.get('new_status') logger.info(f"Updating Container ID: {system_container_id} to new status {new_status}") - + + # Handle pull progress data if status is "pulling" + if new_status == "pulling" and "pull_progress" in data: + pull_progress = data["pull_progress"] + logger.debug(f"Updating pull progress for container {system_container_id}: {pull_progress}") + + # Update pull progress fields + container.pull_status = pull_progress.get("status", "pulling") + container.pull_progress_percentage = pull_progress.get("progress_percentage", 0.0) + container.pull_eta_seconds = pull_progress.get("eta_seconds") + container.pull_bytes_downloaded = pull_progress.get("bytes_downloaded", 0) + + # Set pull start time if not already set + if container.pull_start_time is None: + container.pull_start_time = datetime.utcnow() + + # Update pull source if provided + if "status" in pull_progress: + if pull_progress["status"] == "completed": + container.pull_source = pull_progress.get("source", "remote") + container.pull_end_time = datetime.utcnow() + if container.pull_start_time: + duration = (container.pull_end_time - container.pull_start_time).total_seconds() + container.pull_duration_seconds = duration + elif pull_progress["status"] == "failed": + container.pull_status = "failed" + container.pull_end_time = datetime.utcnow() + if container.pull_start_time: + duration = (container.pull_end_time - container.pull_start_time).total_seconds() + container.pull_duration_seconds = duration + container.set_status(new_status) if new_status == "deleted": container.soft_delete() diff --git a/app/models/models.py b/app/models/models.py index a0d4842..08d12d4 100644 --- a/app/models/models.py +++ b/app/models/models.py @@ -525,6 +525,17 @@ class Workload(BaseModel): launch_params = Column(LONGTEXT, nullable=True) time_to_provision = Column(Float, nullable=True) is_sidecar = Column(Boolean, default=False, nullable=False) + + # Pull progress tracking fields + pull_status = Column(String(50), nullable=True) # idle, pulling, completed, failed, cached + pull_progress_percentage = Column(Float, default=0.0) + pull_eta_seconds = Column(Integer, nullable=True) + pull_bytes_downloaded = Column(Integer, default=0) + pull_start_time = Column(DateTime, nullable=True) + pull_end_time = Column(DateTime, nullable=True) + pull_duration_seconds = Column(Float, nullable=True) + pull_source = Column(String(50), nullable=True) # 'remote', 'cached', 'local_registry' + # Private fields (convention with single underscore) _status = db.Column("status", db.String(50), nullable=True, default="new") # Relationship to VolumeWorkloadMapping @@ -593,26 +604,27 @@ class Workload(BaseModel): def _validate_status_change(self, old_status, new_status): """Internal validation for status changes""" valid_transitions = { - None: ["pending", "pending-allocation", "error"], # Initial states - "pending": ["provisioning", "error"], - "provisioning": ["running", "error"], - "running": ["stopping", "error","pending-deleted","dead","stopped"], - "pending-allocation": ["running", "error","allocated","pending-deleted","failed-to-spawn"], - "allocated":["pending-deleted","running","dead","stopped", "launch_failed","failed-to-spawn"], + None: ["pending", "pending-allocation", "error", "pulling"], # Initial states + "pending": ["provisioning", "error", "pulling"], + "provisioning": ["running", "error", "pulling"], + "running": ["stopping", "error","pending-deleted","dead","stopped", "pulling"], + "pending-allocation": ["running", "error","allocated","pending-deleted","failed-to-spawn", "pulling"], + "allocated":["pending-deleted","running","dead","stopped", "launch_failed","failed-to-spawn", "pulling"], "pending-allocated": ["*"], "dead": ["*"], - "new": ["*"], - "deleted": ["deleted","running"], - "pending-deleted": ["running","deleted","dead","stopped"], # A container can go from pending-deleted to running if there is a race between a deletion and a creation event - "stopped": ["deleted"], - "launch_failed":["deleted","running", "error","pending-deleted","dead","stopped"], - "pending-stop": ["deleted","running","stopped"], - "pending-start": ["deleted","running","stopped","launch_failed"], - "pending-restart": ["deleted","running","stopped","launch_failed"], - "host_failed": ["*"], - "error": ["*"], - "failed-allocation": ["*"], - "failed-to-spawn": ["deleted","running","pending-restart","error","dead","stopped"], + "new": ["*", "pulling"], + "deleted": ["deleted","running", "pulling"], + "pending-deleted": ["running","deleted","dead","stopped", "pulling"], # A container can go from pending-deleted to running if there is a race between a deletion and a creation event + "stopped": ["deleted", "pulling"], + "launch_failed":["deleted","running", "error","pending-deleted","dead","stopped", "pulling"], + "pending-stop": ["deleted","running","stopped", "pulling"], + "pending-start": ["deleted","running","stopped","launch_failed", "pulling"], + "pending-restart": ["deleted","running","stopped","launch_failed", "pulling"], + "host_failed": ["*", "pulling"], + "error": ["*", "pulling"], + "failed-allocation": ["*", "pulling"], + "failed-to-spawn": ["deleted","running","pending-restart","error","dead","stopped", "pulling"], + "pulling": ["running", "error", "failed", "launch_failed", "dead", "stopped", "deleted"], # Pulling can transition to success or failure states } if old_status not in valid_transitions: diff --git a/docs/major todo items.ms b/docs/major todo items.ms index b74ee79..663eb7e 100644 --- a/docs/major todo items.ms +++ b/docs/major todo items.ms @@ -19,7 +19,7 @@ On the storage dashbaord Figure out how to reuturn all pooled_utilisations=WorkloadResourceUsage.query.filter(workload.workload_host_id=workload_host_id).all() - +Whe a failed host comes back online, we need to unfail the workloads that are on it(if any), should also ensure they are not mid-migration to new hosts pod\containr stop\start\restart diff --git a/migrations/versions/689bc1fd6f1d_add_pull_progress_fields_to_workloads.py b/migrations/versions/689bc1fd6f1d_add_pull_progress_fields_to_workloads.py new file mode 100644 index 0000000..578535f --- /dev/null +++ b/migrations/versions/689bc1fd6f1d_add_pull_progress_fields_to_workloads.py @@ -0,0 +1,46 @@ +"""Add pull progress fields to workloads + +Revision ID: 689bc1fd6f1d +Revises: 5b46d01a4afa +Create Date: 2025-09-03 21:12:43.600402 + +""" +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision = '689bc1fd6f1d' +down_revision = '5b46d01a4afa' +branch_labels = None +depends_on = None + + +def upgrade(): + # ### commands auto generated by Alembic - please adjust! ### + with op.batch_alter_table('workloads', schema=None) as batch_op: + batch_op.add_column(sa.Column('pull_status', sa.String(length=50), nullable=True)) + batch_op.add_column(sa.Column('pull_progress_percentage', sa.Float(), nullable=True)) + batch_op.add_column(sa.Column('pull_eta_seconds', sa.Integer(), nullable=True)) + batch_op.add_column(sa.Column('pull_bytes_downloaded', sa.Integer(), nullable=True)) + batch_op.add_column(sa.Column('pull_start_time', sa.DateTime(), nullable=True)) + batch_op.add_column(sa.Column('pull_end_time', sa.DateTime(), nullable=True)) + batch_op.add_column(sa.Column('pull_duration_seconds', sa.Float(), nullable=True)) + batch_op.add_column(sa.Column('pull_source', sa.String(length=50), nullable=True)) + + # ### end Alembic commands ### + + +def downgrade(): + # ### commands auto generated by Alembic - please adjust! ### + with op.batch_alter_table('workloads', schema=None) as batch_op: + batch_op.drop_column('pull_source') + batch_op.drop_column('pull_duration_seconds') + batch_op.drop_column('pull_end_time') + batch_op.drop_column('pull_start_time') + batch_op.drop_column('pull_bytes_downloaded') + batch_op.drop_column('pull_eta_seconds') + batch_op.drop_column('pull_progress_percentage') + batch_op.drop_column('pull_status') + + # ### end Alembic commands ### diff --git a/websocket_server/events/docker.py b/websocket_server/events/docker.py index ecbee7c..cc18a83 100644 --- a/websocket_server/events/docker.py +++ b/websocket_server/events/docker.py @@ -52,22 +52,39 @@ def register_socketio_handlers(socketio): "docker_start", "docker_destroy", "docker_die", - "docker_launch_failed", + "docker_launch_failed", + "docker_pull_progress", ): action_URL = f"workloads/containers/status_update/{system_container_id}" - new_status_map = { - "docker_start": "running", - "docker_destroy": "deleted", - "docker_die": "dead", - "docker_launch_failed": "launch_failed", - } - payload = { - "new_status": new_status_map[event_type], - "system_container_id": system_container_id, - "worker_id": worker_id, - "timestamp": timestamp, - **({"error": error} if error else {}) - } + + if event_type == "docker_pull_progress": + # Handle pull progress events + pull_progress = details.get("pull_progress", {}) + payload = { + "new_status": "pulling", + "system_container_id": system_container_id, + "worker_id": worker_id, + "timestamp": timestamp, + "pull_progress": pull_progress, + **({"error": error} if error else {}) + } + logger.info(f"Processing pull progress update for container {system_container_id}: {pull_progress.get('progress_percentage', 0):.1f}%") + else: + # Handle standard status events + new_status_map = { + "docker_start": "running", + "docker_destroy": "deleted", + "docker_die": "dead", + "docker_launch_failed": "launch_failed", + } + payload = { + "new_status": new_status_map[event_type], + "system_container_id": system_container_id, + "worker_id": worker_id, + "timestamp": timestamp, + **({"error": error} if error else {}) + } + headers = {"Content-Type": "application/json"} websocket_server_response = requests.put( f"{api_server_url}/{action_URL}", diff --git a/worker/worker_tasks/container.py b/worker/worker_tasks/container.py index 049295b..ddf881d 100644 --- a/worker/worker_tasks/container.py +++ b/worker/worker_tasks/container.py @@ -5,7 +5,8 @@ import pty import os import subprocess import select -from typing import List +import threading +from typing import List, Dict, Any, Optional, Callable class ContainerTask: @@ -171,6 +172,218 @@ class ContainerTask: self.logger.info(f"Container {container_id} already running and correct.") return existing + def pull_image_with_progress(self, image_name: str, container_id: str, enable_progress_tracking: bool = False) -> Dict[str, Any]: + """ + Pull a Docker image with optional progress tracking. + + Args: + image_name: The Docker image name to pull + container_id: The system container ID for tracking + enable_progress_tracking: Whether to send detailed progress events + + Returns: + Dict with pull results including status, duration, bytes downloaded, etc. + """ + pull_start_time = time.time() + pull_result = { + "status": "idle", + "source": "unknown", + "duration_seconds": 0.0, + "bytes_downloaded": 0, + "layers_completed": 0, + "total_layers": 0, + "progress_percentage": 0.0, + "eta_seconds": None, + "error": None + } + + try: + self.logger.info(f"Starting pull for image {image_name} (container: {container_id})") + + # Check if image already exists locally + try: + self.docker_client.images.get(image_name) + pull_result["status"] = "completed" + pull_result["source"] = "cached" + pull_result["duration_seconds"] = time.time() - pull_start_time + self.logger.info(f"Image {image_name} already cached locally") + return pull_result + except docker.errors.ImageNotFound: + pass # Image not cached, proceed with pull + + # Set initial status + pull_result["status"] = "pulling" + pull_result["source"] = "remote" + + # Progress tracking variables + last_progress_update = time.time() + progress_tracking_enabled = enable_progress_tracking # Start with initial setting + layer_events_by_id = {} # Track latest event per layer + stop_progress = threading.Event() + + def send_progress_event(progress_data: Dict[str, Any]): + """Send progress update via WebSocket if tracking is enabled""" + if not progress_tracking_enabled: + return + + try: + # Send event through docker_monitor if available + if self.docker_monitor and hasattr(self.docker_monitor, 'event_queue'): + event_data = { + 'source': 'docker', + 'event_type': 'pull_progress', + 'container_id': container_id, + 'system_container_id': container_id, + 'timestamp': time.time(), + 'pull_progress': progress_data + } + self.docker_monitor.event_queue.put(event_data) + self.logger.debug(f"Sent pull progress event for {container_id}: {progress_data['progress_percentage']:.1f}%") + except Exception as e: + self.logger.warning(f"Failed to send progress event: {e}") + + def compute_overall_progress() -> tuple: + """Sum current and total bytes across layers where totals are known.""" + overall_current_bytes = 0 + overall_total_bytes = 0 + any_totals_known = False + + for latest_event in layer_events_by_id.values(): + progress_detail = latest_event.get("progressDetail", {}) + current_bytes = progress_detail.get("current") + total_bytes = progress_detail.get("total") + if current_bytes is not None and total_bytes is not None: + any_totals_known = True + overall_current_bytes += int(current_bytes) + overall_total_bytes += int(total_bytes) + + if not any_totals_known: + return None, None + return overall_current_bytes, overall_total_bytes + + def human_readable_bytes(byte_count) -> str: + """Convert a byte count into a human-readable string.""" + if byte_count is None: + return "?" + unit_labels = ["B", "KB", "MB", "GB", "TB"] + value = float(byte_count) + unit_index = 0 + while value >= 1024.0 and unit_index < len(unit_labels) - 1: + value /= 1024.0 + unit_index += 1 + return f"{value:.1f} {unit_labels[unit_index]}" + + def log_progress_snapshot(): + """Log a concise, rolling snapshot of pull progress.""" + status_counts = {} + for latest_event in layer_events_by_id.values(): + status_text = latest_event.get("status") or "Unknown" + status_counts[status_text] = status_counts.get(status_text, 0) + 1 + + overall_current_bytes, overall_total_bytes = compute_overall_progress() + + if overall_current_bytes is not None and overall_total_bytes: + overall_percentage = (overall_current_bytes / overall_total_bytes) * 100 + overall_amount_text = f"{human_readable_bytes(overall_current_bytes)} / {human_readable_bytes(overall_total_bytes)}" + else: + overall_percentage = 0.0 + overall_amount_text = "? / ?" + + # Update pull result + pull_result["progress_percentage"] = overall_percentage + pull_result["bytes_downloaded"] = overall_current_bytes or 0 + pull_result["total_layers"] = len(layer_events_by_id) + + self.logger.debug(f"Pull progress: {overall_percentage:.1f}% ({overall_amount_text}) - Layers: {len(layer_events_by_id)}") + + # Start the pull operation + self.logger.debug(f"Pulling image {image_name} with initial progress tracking: {enable_progress_tracking}") + + # Use the existing API client for progress tracking + pull_response = self.docker_client.api.pull(image_name, stream=True, decode=True) + + # Process the pull stream + for stream_event in pull_response: + if stop_progress.is_set(): + break + + current_time = time.time() + pull_duration = current_time - pull_start_time + + # Enable progress tracking if pull takes longer than 30 seconds + if not progress_tracking_enabled and pull_duration > 10.0: + self.logger.info(f"Pull taking longer than 30s ({pull_duration:.1f}s), enabling detailed progress tracking") + progress_tracking_enabled = True + + # Handle errors emitted by the engine + if "error" in stream_event: + error_message = stream_event.get("error") or "Unknown error" + self.logger.error(f"Pull error: {error_message}") + raise RuntimeError(error_message) + + # Track per-layer status + layer_identifier = stream_event.get("id") + if layer_identifier: + layer_events_by_id[layer_identifier] = stream_event + + # Periodic snapshot and event sending + if progress_tracking_enabled and (current_time - last_progress_update) >= 5.0: + last_progress_update = current_time + log_progress_snapshot() + + # Send progress event + overall_current_bytes, overall_total_bytes = compute_overall_progress() + progress_percentage = pull_result["progress_percentage"] + + send_progress_event({ + "progress_percentage": progress_percentage, + "bytes_downloaded": overall_current_bytes or 0, + "total_bytes": overall_total_bytes, + "layers_total": len(layer_events_by_id), + "status": "pulling" + }) + + # Pull completed successfully + pull_end_time = time.time() + pull_result["status"] = "completed" + pull_result["duration_seconds"] = pull_end_time - pull_start_time + + # Send final progress event if tracking was ever enabled + if progress_tracking_enabled: + send_progress_event({ + "progress_percentage": 100.0, + "bytes_downloaded": pull_result["bytes_downloaded"], + "eta_seconds": 0, + "status": "completed" + }) + + self.logger.info(f"Successfully pulled image {image_name} in {pull_result['duration_seconds']:.2f}s") + + except Exception as e: + pull_end_time = time.time() + pull_result["status"] = "failed" + pull_result["duration_seconds"] = pull_end_time - pull_start_time + pull_result["error"] = str(e) + + # Send failure event + if enable_progress_tracking: + send_progress_event({ + "progress_percentage": pull_result["progress_percentage"], + "bytes_downloaded": pull_result["bytes_downloaded"], + "eta_seconds": None, + "status": "failed", + "error": str(e) + }) + + self.logger.error(f"Failed to pull image {image_name}: {e}") + + finally: + # Clean up progress thread if it exists + if 'stop_progress' in locals(): + stop_progress.set() + + return pull_result + def launch_container(self, container_spec, is_nscontroller=False): """Launch a new container from spec. Returns the Container or None on failure.""" self.logger.debug("Launching container") @@ -180,6 +393,40 @@ class ContainerTask: networks = container_spec.get("networks", []) system_container_id = container_spec.get("container_id") + # Handle image pulling with progress tracking + if docker_image: + # Normalize image name + if ":" not in docker_image: + docker_image = f"{docker_image}:latest" + + try: + # Start pull with progress tracking - the method will handle the 30-second logic internally + self.logger.debug(f"Starting image pull for {docker_image}") + pull_result = self.pull_image_with_progress( + docker_image, + system_container_id, + enable_progress_tracking=False # Start without tracking, enable after 30s if needed + ) + + # Handle pull results + if pull_result["status"] == "failed": + self.logger.error(f"Failed to pull image {docker_image}: {pull_result.get('error')}") + self.launch_failures.append({ + "id": system_container_id, + "error": f"Image pull failed: {pull_result.get('error')}" + }) + return None + + self.logger.info(f"Image pull completed: {docker_image} ({pull_result['source']}) in {pull_result['duration_seconds']:.2f}s") + + except Exception as e: + self.logger.error(f"Error during image pull for {docker_image}: {e}") + self.launch_failures.append({ + "id": system_container_id, + "error": f"Image pull error: {str(e)}" + }) + return None + # Ports (only for NSControllers, normal containers use shared namespace) if is_nscontroller: self.logger.debug("is nscontroller") @@ -204,6 +451,8 @@ class ContainerTask: "labels": labels, "network_mode": network_mode, } + + # Note: Image is already pulled above, so containers.run() will use the local image if ports: container_kwargs["ports"] = ports if "command" in container_spec: