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

First kinda working POC of docker pull tracking

Browse Source
This commit is contained in:
nexgen_mirrors
2025-09-04 15:05:40 +09:30
parent 8b47047eca
commit 5dbba98282
6 changed files with 390 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/controller/api/workload_container_routes.py
+32 -2
View File
@@ -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()
app/models/models.py
+30 -18
View File
@@ -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:
docs/major todo items.ms
+1 -1
View File
@@ -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
migrations/versions/689bc1fd6f1d_add_pull_progress_fields_to_workloads.py
+46
View File
@@ -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 ###
websocket_server/events/docker.py
+31 -14
View File
@@ -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}",
worker/worker_tasks/container.py
+250 -1
View File
@@ -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: