From ecd9dbfba04754c988b480b467de56973f280a9a Mon Sep 17 00:00:00 2001 From: Cory Hawkvelt Date: Thu, 11 Sep 2025 00:05:18 +0930 Subject: [PATCH] Added large container pull tracking to the worker client, websocket server and the api server --- .../api/workload_container_routes.py | 4 +- app/models/models.py | 10 +- app/utils/create_workload_container.py | 17 ++- docs/launch_single_container.sh | 4 +- ...8_add_pull_progress_fields_to_workloads.py | 38 +++++ ...d_add_pull_progress_fields_to_workloads.py | 8 +- ...f3a2b1c4d5e_restore_pull_layers_columns.py | 36 +++++ websocket_server/events/docker.py | 2 +- worker/dockerMonitor.py | 27 +++- worker/libvirtMonitor.py | 21 ++- worker/main.py | 23 ++- worker/workerClient.py | 24 +++- worker/worker_tasks/container.py | 131 ++++++++++++------ 13 files changed, 271 insertions(+), 74 deletions(-) create mode 100644 migrations/versions/32bbe8d31a98_add_pull_progress_fields_to_workloads.py create mode 100644 migrations/versions/8f3a2b1c4d5e_restore_pull_layers_columns.py diff --git a/app/controller/api/workload_container_routes.py b/app/controller/api/workload_container_routes.py index f449f53..07265a7 100644 --- a/app/controller/api/workload_container_routes.py +++ b/app/controller/api/workload_container_routes.py @@ -439,9 +439,9 @@ def update_container_workload(system_container_id): # 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) + container.pull_layers_completed = pull_progress.get("layers_completed", 0) + container.pull_layers_total = pull_progress.get("layers_total", 0) # Set pull start time if not already set if container.pull_start_time is None: diff --git a/app/models/models.py b/app/models/models.py index 08d12d4..87dc404 100644 --- a/app/models/models.py +++ b/app/models/models.py @@ -528,9 +528,9 @@ class Workload(BaseModel): # 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_layers_completed = Column(Integer, default=0) + pull_layers_total = 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) @@ -607,7 +607,7 @@ class Workload(BaseModel): None: ["pending", "pending-allocation", "error", "pulling"], # Initial states "pending": ["provisioning", "error", "pulling"], "provisioning": ["running", "error", "pulling"], - "running": ["stopping", "error","pending-deleted","dead","stopped", "pulling"], + "running": ["stopping", "error","pending-deleted","dead","stopped", "pulling","host_failed"], "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": ["*"], @@ -620,8 +620,8 @@ class Workload(BaseModel): "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"], + "host_failed": ["*"], + "error": ["*"], "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 diff --git a/app/utils/create_workload_container.py b/app/utils/create_workload_container.py index aa131e2..9510fbc 100644 --- a/app/utils/create_workload_container.py +++ b/app/utils/create_workload_container.py @@ -379,7 +379,10 @@ def allocate_and_dispatch(pod_id: str) -> None: except Exception: launch_params = {} - for port_mapping in launch_params.get("ports", []): + _ports=launch_params.get("ports", []) + logger.debug(f"Ports for this container {_ports}") + for port_mapping in _ports: + logger.debug(f"Allocating port {port_mapping}") allocate_ports_for_container( container=container, host=host, @@ -499,7 +502,13 @@ def build_pod_payload(pod: ContainerPod) -> Dict: continue # Skip deleted or missing workloads else: logger.debug(f"including container {w.id}") - launch_params = json.loads(w.launch_params) + + # For NSController, use the updated launch_params that include port mappings + if w.workload_type == "NSController": + launch_params = ns_lp + else: + launch_params = json.loads(w.launch_params) + if ":" not in launch_params["docker_image"]: launch_params["docker_image"] += ":latest" @@ -510,6 +519,10 @@ def build_pod_payload(pod: ContainerPod) -> Dict: "workload_type": w.workload_type, "desired_state": "running", } + + # Include ports for NSController + if w.workload_type == "NSController" and "ports" in launch_params: + cont["ports"] = launch_params["ports"] # Check if this is a special lifecycle operation (restart) if w.status == "pending-restart": diff --git a/docs/launch_single_container.sh b/docs/launch_single_container.sh index fd4bb15..d509ee7 100644 --- a/docs/launch_single_container.sh +++ b/docs/launch_single_container.sh @@ -2,11 +2,11 @@ curl -X POST http://192.168.50.124:5000/api/workloads/containers -H "Content-T "virtual_data_center": "bad48639-d2f6-4bba-8667-53e410a2eeb9", "containers": [ { - "docker_image": "joke_container", + "docker_image": "nginx", "container_name": "very-important-webserver", "ports": [ { - "internal": 9090, + "internal": 80, "external": 9090, "use_dns": false } diff --git a/migrations/versions/32bbe8d31a98_add_pull_progress_fields_to_workloads.py b/migrations/versions/32bbe8d31a98_add_pull_progress_fields_to_workloads.py new file mode 100644 index 0000000..acd5e63 --- /dev/null +++ b/migrations/versions/32bbe8d31a98_add_pull_progress_fields_to_workloads.py @@ -0,0 +1,38 @@ +"""Add pull progress fields to workloads + +Revision ID: 32bbe8d31a98 +Revises: 689bc1fd6f1d +Create Date: 2025-09-04 15:38:17.018892 + +""" +from alembic import op +import sqlalchemy as sa +from sqlalchemy.dialects import mysql + +# revision identifiers, used by Alembic. +revision = '32bbe8d31a98' +down_revision = '689bc1fd6f1d' +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_layers_completed', sa.Integer(), nullable=True)) + batch_op.add_column(sa.Column('pull_layers_total', sa.Integer(), nullable=True)) + batch_op.drop_column('pull_progress_percentage') + batch_op.drop_column('pull_eta_seconds') + + # ### 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.add_column(sa.Column('pull_eta_seconds', mysql.INTEGER(display_width=11), autoincrement=False, nullable=True)) + batch_op.add_column(sa.Column('pull_progress_percentage', mysql.FLOAT(), nullable=True)) + batch_op.drop_column('pull_layers_total') + batch_op.drop_column('pull_layers_completed') + + # ### end Alembic commands ### diff --git a/migrations/versions/689bc1fd6f1d_add_pull_progress_fields_to_workloads.py b/migrations/versions/689bc1fd6f1d_add_pull_progress_fields_to_workloads.py index 578535f..055ac17 100644 --- a/migrations/versions/689bc1fd6f1d_add_pull_progress_fields_to_workloads.py +++ b/migrations/versions/689bc1fd6f1d_add_pull_progress_fields_to_workloads.py @@ -20,9 +20,9 @@ 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_layers_completed', sa.Integer(), nullable=True)) + batch_op.add_column(sa.Column('pull_layers_total', 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)) @@ -38,9 +38,9 @@ def downgrade(): 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_layers_total') + batch_op.drop_column('pull_layers_completed') 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/migrations/versions/8f3a2b1c4d5e_restore_pull_layers_columns.py b/migrations/versions/8f3a2b1c4d5e_restore_pull_layers_columns.py new file mode 100644 index 0000000..d0fa125 --- /dev/null +++ b/migrations/versions/8f3a2b1c4d5e_restore_pull_layers_columns.py @@ -0,0 +1,36 @@ +"""Restore pull layers columns that were incorrectly dropped + +Revision ID: 8f3a2b1c4d5e +Revises: 32bbe8d31a98 +Create Date: 2025-09-09 15:31:00.000000 + +""" +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision = '8f3a2b1c4d5e' +down_revision = '32bbe8d31a98' +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_layers_completed', sa.Integer(), nullable=True)) + batch_op.add_column(sa.Column('pull_layers_total', sa.Integer(), 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_layers_total') + batch_op.drop_column('pull_layers_completed') + + + # ### end Alembic commands ### \ No newline at end of file diff --git a/websocket_server/events/docker.py b/websocket_server/events/docker.py index cc18a83..88a43cc 100644 --- a/websocket_server/events/docker.py +++ b/websocket_server/events/docker.py @@ -68,7 +68,7 @@ def register_socketio_handlers(socketio): "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}%") + logger.info(f"Processing pull progress update for container {system_container_id}: {pull_progress.get('bytes_downloaded', 0)} bytes, {pull_progress.get('layers_completed', 0)}/{pull_progress.get('layers_total', 0)} layers") else: # Handle standard status events new_status_map = { diff --git a/worker/dockerMonitor.py b/worker/dockerMonitor.py index 2fe9231..cf008f1 100644 --- a/worker/dockerMonitor.py +++ b/worker/dockerMonitor.py @@ -23,6 +23,14 @@ class DockerMonitor: self.manager = multiprocessing.Manager() self.blacklisted_containers = self.manager.list() self.stop_event = self.manager.Event() + + def __del__(self): + """Cleanup manager resources when object is deleted""" + try: + if hasattr(self, 'manager'): + self.manager.shutdown() + except Exception as e: + logger.error(f"Error during DockerMonitor cleanup: {e}") def add_to_blacklist(self, container_id): """Add a container ID to the blacklist.""" @@ -103,8 +111,19 @@ class DockerMonitor: def stop(self): """Stop the Docker monitor""" - self.stop_event.set() - self.docker_client.close() + try: + self.stop_event.set() + except Exception as e: + logger.error(f"Error setting stop event: {e}") + + try: + self.docker_client.close() + except Exception as e: + logger.error(f"Error closing Docker client: {e}") + # Clean up the manager - if hasattr(self, 'manager'): - self.manager.shutdown() + try: + if hasattr(self, 'manager'): + self.manager.shutdown() + except Exception as e: + logger.error(f"Error shutting down manager: {e}") diff --git a/worker/libvirtMonitor.py b/worker/libvirtMonitor.py index 7a78520..1bb7327 100644 --- a/worker/libvirtMonitor.py +++ b/worker/libvirtMonitor.py @@ -13,6 +13,14 @@ class LibvirtMonitor: self.event_impl = None self.manager = multiprocessing.Manager() self.stop_event = self.manager.Event() + + def __del__(self): + """Cleanup manager resources when object is deleted""" + try: + if hasattr(self, 'manager'): + self.manager.shutdown() + except Exception as e: + logger.error(f"Error during LibvirtMonitor cleanup: {e}") def connect_libvirt(self): """Connect to libvirt""" @@ -220,7 +228,14 @@ class LibvirtMonitor: def stop(self): """Stop the libvirt monitor""" - self.stop_event.set() + try: + self.stop_event.set() + except Exception as e: + logger.error(f"Error setting stop event: {e}") + # Clean up the manager - if hasattr(self, 'manager'): - self.manager.shutdown() \ No newline at end of file + try: + if hasattr(self, 'manager'): + self.manager.shutdown() + except Exception as e: + logger.error(f"Error shutting down manager: {e}") \ No newline at end of file diff --git a/worker/main.py b/worker/main.py index 1e11226..7b89fe0 100644 --- a/worker/main.py +++ b/worker/main.py @@ -237,6 +237,8 @@ def worker_process_func(event_queue, docker_monitor=None, libvirt_monitor=None): worker = WorkerClient(event_queue=event_queue, docker_monitor=docker_monitor, libvirt_monitor=libvirt_monitor) try: await worker.start() + except KeyboardInterrupt: + logger.info("Worker process interrupted.") finally: worker.stop() @@ -353,14 +355,29 @@ def main(): # Handle graceful shutdown def signal_handler(sig, frame): logger.info("Shutting down...") - for monitor in monitors: - monitor.stop() + # First, stop the worker process gracefully for process in processes: - process.join(timeout=5) if process.is_alive(): + logger.info(f"Terminating worker process {process.pid}") process.terminate() + # Give processes a chance to terminate gracefully + for process in processes: + process.join(timeout=3) + if process.is_alive(): + logger.info(f"Force killing worker process {process.pid}") + process.kill() + process.join(timeout=1) + + # Then stop monitors + for monitor in monitors: + try: + monitor.stop() + except Exception as e: + logger.error(f"Error stopping monitor: {e}") + + logger.info("Shutdown complete") sys.exit(0) # Register signal handlers diff --git a/worker/workerClient.py b/worker/workerClient.py index 4fa90b6..395a91f 100644 --- a/worker/workerClient.py +++ b/worker/workerClient.py @@ -106,10 +106,16 @@ class WorkerClient: self.monitor_processes.clear() if self.docker_monitor: logger.info(f"Stopping Docker Monitor") - self.docker_monitor.stop() + try: + self.docker_monitor.stop() + except Exception as e: + logger.error(f"Error stopping Docker monitor: {e}") if self.libvirt_monitor: logger.info(f"Stopping Libvirt Monitor") - self.libvirt_monitor.stop() + try: + self.libvirt_monitor.stop() + except Exception as e: + logger.error(f"Error stopping Libvirt monitor: {e}") async def handle_worker_ping(self, data): """ @@ -132,7 +138,7 @@ class WorkerClient: logger.error(f"Connection failed: {data}") self.joined_server = False - async def on_disconnect(self): + async def on_disconnect(self, *args): logger.info("Disconnected from the server.") self.joined_server = False self.monitors_started = False @@ -299,6 +305,11 @@ class WorkerClient: "action": event_data.get("details", {}).get("Action"), **({"error": event_data.get("error")} if event_data.get("error") else {}) } + + # Include pull_progress data for pull_progress events + if event_type == "pull_progress" and "pull_progress" in event_data: + logger.debug("Sengin pull progreesss info") + payload["details"]["pull_progress"] = event_data["pull_progress"] # Emit the event to the server await self.sio.emit("docker_event", payload) @@ -339,9 +350,14 @@ class WorkerClient: def stop(self): """Stop the worker client.""" + logger.info("Stopping worker client...") self.running = False self.monitors_started = False - self.stop_monitors() + try: + self.stop_monitors() + except Exception as e: + logger.error(f"Error stopping monitors: {e}") + logger.info("Worker client stopped.") diff --git a/worker/worker_tasks/container.py b/worker/worker_tasks/container.py index ddf881d..3f3e6d5 100644 --- a/worker/worker_tasks/container.py +++ b/worker/worker_tasks/container.py @@ -68,24 +68,75 @@ class ContainerTask: if 'storage' in container and container['storage']: self._process_storage_volumes(container['storage']) - # Find the NSController container ID for networking - nscontroller_container_id = None - for container_spec in container_specs: - if container_spec.get("workload_type") == "NSController": - nscontroller_container_id = container_spec["container_id"] - break + nscontroller_specs: list[dict] = [] + regular_container_specs: list[dict] = [] - # Process all containers in unified loop for container_spec in container_specs: if container_spec.get("workload_type") == "NSController": - # Handle NSController - self.logger.debug("Launching NSController") - self.ensure_container(container_spec, is_nscontroller=True) + nscontroller_specs.append(container_spec) else: - # Handle regular containers - set nscontroller_container_id for networking - if nscontroller_container_id: - container_spec["nscontroller_container_id"] = nscontroller_container_id - self.ensure_container(container_spec) + regular_container_specs.append(container_spec) + + nscontroller_container_id: str | None = None + if len(nscontroller_specs) == 0: + self.logger.info("No NSController containers found in this payload.") + elif len(nscontroller_specs) == 1: + nscontroller_container_id = nscontroller_specs[0].get("container_id") + self.logger.info( + "Identified single NSController container_id=%s to initialize networking.", + nscontroller_container_id, + ) + else: + # Multiple NSControllers present—log a warning and pick the first deterministically. + nscontroller_container_id = nscontroller_specs[0].get("container_id") + self.logger.warning( + "Multiple NSController containers detected (%d). Proceeding with the first container_id=%s.", + len(nscontroller_specs), + nscontroller_container_id, + ) + + # ---- launch NSController(s) first (no injection of nscontroller_container_id) + for ns_container_spec in nscontroller_specs: + self.logger.debug( + "Launching NSController container_id=%s", + ns_container_spec.get("container_id"), + ) + self.ensure_container(ns_container_spec, is_nscontroller=True) + + # ---- launch regular containers with networking pointer if available + for regular_container_spec in regular_container_specs: + if nscontroller_container_id: + regular_container_spec["nscontroller_container_id"] = nscontroller_container_id + self.logger.debug( + "Injecting nscontroller_container_id=%s into container_id=%s", + nscontroller_container_id, + regular_container_spec.get("container_id"), + ) + self.logger.debug( + "Launching regular container_id=%s", + regular_container_spec.get("container_id"), + ) + self.ensure_container(regular_container_spec) + + + # # Find the NSController container ID for networking + # nscontroller_container_id = None + # for container_spec in container_specs: + # if container_spec.get("workload_type") == "NSController": + # nscontroller_container_id = container_spec["container_id"] + # break + + # # Process all containers in unified loop + # for container_spec in container_specs: + # if container_spec.get("workload_type") == "NSController": + # # Handle NSController + # self.logger.debug("Launching NSController") + # self.ensure_container(container_spec, is_nscontroller=True) + # else: + # # Handle regular containers - set nscontroller_container_id for networking + # if nscontroller_container_id: + # container_spec["nscontroller_container_id"] = nscontroller_container_id + # self.ensure_container(container_spec) # Set success to False if there were any launch failures response_payload["success"] = len(self.launch_failures) == 0 @@ -192,7 +243,6 @@ class ContainerTask: "bytes_downloaded": 0, "layers_completed": 0, "total_layers": 0, - "progress_percentage": 0.0, "eta_seconds": None, "error": None } @@ -238,28 +288,30 @@ class ContainerTask: '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}%") + self.logger.debug(f"Sent pull progress event for {container_id}: {progress_data['bytes_downloaded']} bytes, {progress_data['layers_completed']}/{progress_data['layers_total']} layers") 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.""" + """Sum current bytes and count completed layers.""" overall_current_bytes = 0 - overall_total_bytes = 0 - any_totals_known = False + layers_completed = 0 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) + status = latest_event.get("status") - if not any_totals_known: - return None, None - return overall_current_bytes, overall_total_bytes + if current_bytes is not None: + overall_current_bytes += int(current_bytes) + + # Count as completed if status indicates completion or current == total + if status in ["Download complete", "Pull complete"] or \ + (current_bytes is not None and total_bytes is not None and current_bytes == total_bytes): + layers_completed += 1 + + return overall_current_bytes, layers_completed def human_readable_bytes(byte_count) -> str: """Convert a byte count into a human-readable string.""" @@ -280,21 +332,14 @@ class ContainerTask: 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 = "? / ?" + overall_current_bytes, layers_completed = compute_overall_progress() # Update pull result - pull_result["progress_percentage"] = overall_percentage pull_result["bytes_downloaded"] = overall_current_bytes or 0 + pull_result["layers_completed"] = layers_completed 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)}") + self.logger.debug(f"Pull progress: {human_readable_bytes(overall_current_bytes or 0)} downloaded - Layers: {layers_completed}/{len(layer_events_by_id)}") # Start the pull operation self.logger.debug(f"Pulling image {image_name} with initial progress tracking: {enable_progress_tracking}") @@ -332,13 +377,11 @@ class ContainerTask: log_progress_snapshot() # Send progress event - overall_current_bytes, overall_total_bytes = compute_overall_progress() - progress_percentage = pull_result["progress_percentage"] + overall_current_bytes, layers_completed = compute_overall_progress() send_progress_event({ - "progress_percentage": progress_percentage, "bytes_downloaded": overall_current_bytes or 0, - "total_bytes": overall_total_bytes, + "layers_completed": layers_completed, "layers_total": len(layer_events_by_id), "status": "pulling" }) @@ -351,9 +394,9 @@ class ContainerTask: # 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, + "layers_completed": pull_result["layers_completed"], + "layers_total": pull_result["total_layers"], "status": "completed" }) @@ -368,9 +411,9 @@ class ContainerTask: # 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, + "layers_completed": pull_result["layers_completed"], + "layers_total": pull_result["total_layers"], "status": "failed", "error": str(e) })