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

fixing reconcile

Browse Source
This commit is contained in:
nexgen_mirrors
2025-11-17 13:12:52 +10:30
parent e97e295be6
commit 3cb2a0f2fb
4 changed files with 41 additions and 392 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/controller/api/workload_container_routes.py
+1 -1
View File
@@ -2,7 +2,7 @@
import uuid
from flask import json, request
from app import app, db, logger
from app.models.models import Workload, WorkloadHostPortMapping, VirtualDataCenter, ContainerPod, CloudflareDNSRecord, CloudflareTunnel, WorkloadRequest
from app.models.models import Workload, VirtualDataCenter, ContainerPod, WorkloadRequest
from datetime import datetime
from app.controller import api_bp
from sqlalchemy import or_
worker/MISSING_CONTAINER_FIX.md
-93
View File
@@ -1,93 +0,0 @@
# Fix for Missing Container Reporting Edge Case
## Problem Description
There was an edge case in the worker component where the server thought a container was running, but it didn't exist on the host. In this scenario, no update was sent to the API because the host saw no delta change.
This happened because the container status reporting functionality only reported statuses for containers that actually existed on the host, but didn't handle containers that were expected to exist according to the server but were missing on the host.
## Solution
The solution was to enhance the `send_full_statuses_for_host` method in the `ContainerTask` class to:
1. Track which containers actually exist on the host
2. Compare this with the list of containers that should exist according to the server
3. Report any missing containers as "died" to the server
## Implementation Details
### Changes to `worker_tasks/container.py`
The `send_full_statuses_for_host` method was modified to:
1. Create a set of container IDs that actually exist on the host:
```python
existing_container_ids = set()
```
2. When iterating through existing containers, add their IDs to this set:
```python
existing_container_ids.add(system_id)
```
3. After processing existing containers, identify missing containers when scope is "pod":
```python
if scope == "pod" and pod_payload:
# Get the list of container IDs that should exist
server_container_ids = set()
containers_data = pod_payload.get("containers", [])
for container_data in containers_data:
container_id = container_data.get("container_id")
if container_id:
server_container_ids.add(container_id)
# Find missing containers
missing_container_ids = server_container_ids - existing_container_ids
```
4. For each missing container, create and send a "die" event to the server:
```python
for missing_container_id in missing_container_ids:
# Create event data for the missing container
event_data = {
'source': 'docker',
'event_type': 'die', # Report as 'die' since it's missing
'container_id': missing_container_id,
'system_container_id': missing_container_id,
'container_name': container_name,
'timestamp': time.time(),
'details': {
'status': 'die',
'id': missing_container_id,
# ... other details
},
}
# Put event in the queue for the worker client
if self.docker_monitor and hasattr(self.docker_monitor, 'event_queue'):
self.docker_monitor.event_queue.put(event_data)
```
### Changes to `workerClient.py`
The `run_reconciliation` method was updated to use the enhanced `send_full_statuses_for_host` function:
```python
# Use the enhanced send_full_statuses_for_host function which now handles missing containers
container_task = ContainerTask(logger=logger, docker_monitor=self.docker_monitor)
container_task.send_full_statuses_for_host(workloads_data["job_details"], scope="pod")
```
## Testing
A test script `test_missing_container_reporting.py` was updated to verify the fix. The test:
1. Mocks a Docker client that returns no containers (simulating missing containers)
2. Mocks the `fetch_container_workloads_for_host` method to return expected containers
3. Verifies that "die" events are correctly sent for missing containers
The test confirms that the fix works as expected.
## Impact
This fix ensures that the server is properly notified when containers that should be running are missing from the host, allowing the server to update its state accordingly and maintain consistency between the server and worker.
worker/MISSING_CONTAINER_FIX_SUMMARY.md
-78
View File
@@ -1,78 +0,0 @@
# Missing Container Reporting Fix - Implementation Summary
## Problem
There was an edge case in the worker component where the server thought a container was running, but it didn't exist on the host. In this scenario, no update was sent to the API because the host saw no delta change.
## Solution
Enhanced the `send_full_statuses_for_host` method in the `ContainerTask` class to detect and report containers that are expected to exist according to the server but are missing on the host.
## Changes Made
### 1. Enhanced `send_full_statuses_for_host` method in `worker_tasks/container.py`
The method was modified to:
- Track which containers actually exist on the host
- Compare this with the list of containers that should exist according to the server
- Report any missing containers as "died" to the server
Key changes:
```python
# Create a set of container IDs that actually exist on the host
existing_container_ids = set()
# When iterating through existing containers, add their IDs to this set
existing_container_ids.add(system_id)
# After processing existing containers, identify missing containers when scope is "pod"
if scope == "pod" and pod_payload:
# Get the list of container IDs that should exist
server_container_ids = set()
containers_data = pod_payload.get("containers", [])
for container_data in containers_data:
container_id = container_data.get("container_id")
if container_id:
server_container_ids.add(container_id)
# Find missing containers
missing_container_ids = server_container_ids - existing_container_ids
# Report each missing container
for missing_container_id in missing_container_ids:
# Create and send "die" event for missing container
```
### 2. Updated `run_reconciliation` method in `workerClient.py`
The method was updated to use the enhanced functionality:
```python
# Use the enhanced send_full_statuses_for_host function which now handles missing containers
container_task = ContainerTask(logger=logger, docker_monitor=self.docker_monitor)
container_task.send_full_statuses_for_host(workloads_data["job_details"], scope="pod")
```
### 3. Updated test script `test_missing_container_reporting.py`
The test was updated to:
- Properly mock the DockerMonitor with event_queue attribute
- Use the new approach for testing missing container reporting
### 4. Updated documentation
- `MISSING_CONTAINER_FIX.md` was updated to reflect the new implementation approach
- Added this summary document
## Testing
The fix was verified with `test_missing_container_reporting.py` which:
1. Mocks a Docker client that returns no containers (simulating missing containers)
2. Mocks the `fetch_container_workloads_for_host` method to return expected containers
3. Verifies that "die" events are correctly sent for missing containers
Test results confirm that:
- Missing containers are correctly detected
- "die" events are sent for each missing container
- Event data structure is correct
## Impact
This fix ensures that the server is properly notified when containers that should be running are missing from the host, allowing the server to update its state accordingly and maintain consistency between the server and worker.
worker/workerClient.py
+40 -220
View File
@@ -187,33 +187,6 @@ class WorkerClient:
# Run container reconciliation on both first join and reconnects
asyncio.create_task(self.run_reconciliation())
self.first_join_completed = True
async def run_reconciliation(self):
"""
Run the reconciliation process after a short delay to ensure
the connection is fully established.
"""
# Wait a bit for the connection to be fully established
await asyncio.sleep(2)
# Run container reconciliation
success = await self.reconcile_containers()
if success:
logger.info("Container reconciliation completed successfully on worker join")
else:
logger.error("Container reconciliation failed on worker join")
# Always report current container statuses to update the server
try:
# Fetch container workloads from the API
workloads_data = await self.fetch_container_workloads_for_host()
if workloads_data is not None:
# Use the enhanced send_full_statuses_for_host function which now handles missing containers
container_task = ContainerTask(logger=logger, docker_monitor=self.docker_monitor)
container_task.send_full_statuses_for_host(workloads_data["job_details"], scope="pod")
except Exception as e:
logger.error(f"Error reporting container statuses: {str(e)}")
async def on_join_reject(self, data):
logger.error("Join rejected.")
@@ -401,14 +374,50 @@ class WorkerClient:
if not self.event_queue:
return
# Dictionary to store the most recent event for each container and status combination
pending_events = {}
# Timer to ensure events are sent even if no new events arrive
last_send_time = time.time()
send_interval = 1.0 # Send events at least every second
while self.running:
try:
# Non-blocking check for events
if not self.event_queue.empty():
events_processed = 0
while not self.event_queue.empty() and events_processed < 100: # Process up to 100 events at once
event = self.event_queue.get(block=False)
logger.debug(f"Processing event from queue: {event}")
await self.send_event(event)
await asyncio.sleep(0.1) # Small delay to prevent CPU thrashing
events_processed += 1
# Create a key for deduplication based on container and event type
container_id = event.get('system_container_id') or event.get('container_id')
event_type = event.get('event_type')
source = event.get('source')
if container_id and event_type and source:
# Create a unique key for this container and event type combination
key = f"{source}:{container_id}:{event_type}"
# Store the most recent event for this key
pending_events[key] = event
logger.debug(f"Added/updated pending event: {key}")
else:
# If we can't create a key, send the event immediately
logger.debug(f"Processing event immediately (no key): {event}")
await self.send_event(event)
# Send pending events if we have any or if enough time has passed
current_time = time.time()
if pending_events and (len(pending_events) >= 10 or (current_time - last_send_time) >= send_interval):
logger.debug(f"Sending {len(pending_events)} pending events")
for key, event in list(pending_events.items()):
logger.debug(f"Processing event from queue: {event}")
await self.send_event(event)
del pending_events[key]
last_send_time = current_time
# Small delay to prevent CPU thrashing
await asyncio.sleep(0.01)
except Exception as e:
logger.error(f"Error in event processing: {e}")
@@ -607,192 +616,3 @@ class WorkerClient:
if session:
session.close()
logger.info(f"[VNC] Stopped session for {request_id}")
async def fetch_container_workloads_for_host(self):
"""
Fetch all container workloads for this host from the API.
Returns the response in pod-update format.
"""
import requests
from settings import settings
try:
# Get API base URL from settings
api_base_url = settings.get_value("API_BASE_URL")
worker_id = self.worker_id
# Construct the URL for the new endpoint
url = f"{api_base_url}/api/workload_hosts/{worker_id}/container_workloads"
logger.info(f"Fetching container workloads from {url}")
# Make the HTTP request
response = requests.get(url)
# Check if the request was successful
if response.status_code == 200:
response_data = response.json()
# Check if the response has the standard envelope format
if isinstance(response_data, dict) and 'success' in response_data:
if response_data['success']:
# Return the data from the 'data' field if it exists
return response_data.get('data')
else:
# Handle error responses
error_msg = response_data.get('message', 'Unknown error')
error_details = response_data.get('error', {})
logger.error(f"API Error: {error_msg}")
if error_details:
logger.error(f"Error details: {error_details}")
return None
else:
# Handle responses that don't follow the standard format
return response_data
else:
logger.error(f"Failed to fetch container workloads. Status code: {response.status_code}")
logger.error(f"Response: {response.text}")
return None
except requests.exceptions.RequestException as e:
logger.error(f"API Error when fetching container workloads: {str(e)}")
return None
except Exception as e:
logger.error(f"Unexpected error when fetching container workloads: {str(e)}")
return None
async def reconcile_containers(self):
"""
Reconcile containers by fetching the desired state from the API and ensuring
the local Docker state matches it.
"""
logger.info("Starting container reconciliation process")
# Fetch container workloads from the API
workloads_data = await self.fetch_container_workloads_for_host()
logger.debug(workloads_data)
if workloads_data is None:
logger.error("Failed to fetch container workloads for reconciliation")
return False
try:
# Process the workloads using the existing ContainerTask.handle_pod_update method
container_task = ContainerTask(logger=logger, docker_monitor=self.docker_monitor)
result = container_task.handle_pod_update_with_reconciliation(workloads_data["job_details"])
if result and result.get("success"):
logger.info("Container reconciliation completed successfully")
# Handle any launch failures
if result.get("launch_failures"):
for failure in result["launch_failures"]:
logger.error(f"Container launch failure: {failure['id']} - {failure['error']}")
# Add to event queue so it gets reported to the server
if self.event_queue:
event_data = {
"source": "docker",
"event_type": "launch_failed",
"container_id": failure["id"],
"system_container_id": failure["id"],
"container_name": failure["id"],
"timestamp": time.time(),
"error": failure["error"]
}
self.event_queue.put(event_data)
logger.debug(f"Added Docker event to queue: launch_failed for container {failure['id']}.")
else:
logger.error("Container reconciliation failed")
if result and result.get("launch_failures"):
for failure in result["launch_failures"]:
logger.error(f"Container launch failure: {failure['id']} - {failure['error']}")
return False
except Exception as e:
logger.error(f"Error during container reconciliation: {str(e)}")
return False
# Now perform cleanup of containers that are managed by us but not in the server response
await self.cleanup_unwanted_containers(workloads_data)
return True
async def cleanup_unwanted_containers(self, workloads_data):
"""
Compare running Docker containers with server response and terminate those
with "managed_by" flag set to the worker's UUID but not in server response.
"""
logger.info("Starting container cleanup process")
try:
import docker
# Get Docker client
docker_client = docker.from_env()
# Get the list of container IDs from the server response
server_container_ids = set()
containers_data = workloads_data.get("job_details", {}).get("containers", [])
for container_data in containers_data:
container_id = container_data.get("container_id")
if container_id:
server_container_ids.add(container_id)
logger.debug(f"Server container IDs: {server_container_ids}")
# Get all running containers
all_containers = docker_client.containers.list(all=True)
# Track how many containers we terminate
terminated_count = 0
# Check each container
for container in all_containers:
# Check if container has the managed_by label
labels = container.labels
managed_by = labels.get("managed_by")
system_container_id = labels.get("system_container_id")
logger.debug(f"Checking container {container.name} (ID: {container.short_id}) - managed_by: {managed_by}, system_id: {system_container_id}")
# If container is managed by us but not in server response, terminate it
if managed_by == settings.get_value("WORKER_ID") and system_container_id and system_container_id not in server_container_ids:
logger.info(f"Terminating container {container.name} (ID: {container.short_id}) as it's managed by us but not in server response")
try:
# Stop and remove the container
container.stop(timeout=10)
container.remove(force=True)
logger.info(f"Successfully terminated container {container.name}")
terminated_count += 1
# Report termination to server if we have an event queue
if self.event_queue:
event_data = {
"source": "docker",
"event_type": "terminated",
"container_id": container.short_id,
"system_container_id": system_container_id,
"container_name": container.name,
"timestamp": time.time(),
"details": {
"reason": "not_in_server_response"
}
}
self.event_queue.put(event_data)
logger.debug(f"Added Docker event to queue: terminated for container {container.short_id}.")
except Exception as e:
logger.error(f"Failed to terminate container {container.name}: {str(e)}")
# Report failure to server if we have an event queue
if self.event_queue:
event_data = {
"source": "docker",
"event_type": "terminate_failed",
"container_id": container.short_id,
"system_container_id": system_container_id,
"container_name": container.name,
"timestamp": time.time(),
"error": str(e)
}
self.event_queue.put(event_data)
logger.debug(f"Added Docker event to queue: terminate_failed for container {container.short_id}.")
logger.info(f"Container cleanup completed. Terminated {terminated_count} containers.")
except Exception as e:
logger.error(f"Error during container cleanup: {str(e)}")