JamesBhattarai/3cloud-backend
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Added large container pull tracking to the worker client, websocket server and the api server

Browse Source
This commit is contained in:
nexgen_mirrors
2025-09-11 00:05:18 +09:30
parent 5dbba98282
commit ecd9dbfba0
13 changed files with 271 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/controller/api/workload_container_routes.py
+2 -2
View File
@@ -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:
app/models/models.py
+5 -5
View File
@@ -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
app/utils/create_workload_container.py
+15 -2
View File
@@ -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":
docs/launch_single_container.sh
+2 -2
View File
@@ -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
}
migrations/versions/32bbe8d31a98_add_pull_progress_fields_to_workloads.py
+38
View File
@@ -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 ###
migrations/versions/689bc1fd6f1d_add_pull_progress_fields_to_workloads.py
+4 -4
View File
@@ -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 ###
migrations/versions/8f3a2b1c4d5e_restore_pull_layers_columns.py
+36
View File
@@ -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 ###
websocket_server/events/docker.py
+1 -1
View File
@@ -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 = {
worker/dockerMonitor.py
+23 -4
View File
@@ -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}")
worker/libvirtMonitor.py
+18 -3
View File
@@ -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()
try:
if hasattr(self, 'manager'):
self.manager.shutdown()
except Exception as e:
logger.error(f"Error shutting down manager: {e}")
worker/main.py
+20 -3
View File
@@ -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
worker/workerClient.py
+20 -4
View File
@@ -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.")
worker/worker_tasks/container.py
+87 -44
View File
@@ -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)
})