feat: Introduce Celery tasks overview, enhance pod detail fetching with resource metrics and API timeouts, and improve error handling.

Author: dimittal

agent/tasks.py

@@ -3,16 +3,21 @@
 import requests
 from celery import shared_task
 from django.conf import settings
+from requests.exceptions import RequestException, Timeout
 
 from utils.time_utils import current_epoch_timestamp
 
 logger = logging.getLogger(__name__)
 
+# Timeout constants (in seconds)
+K8S_API_TIMEOUT = 30
+HTTP_REQUEST_TIMEOUT = 30
+
 
 def get_pod_details(namespace='drdroid'):
     # Import kubernetes here to avoid loading at module import time
     from kubernetes import client, config
-    
+
     try:
         # Try to load in-cluster config first, then local config
         try:
@@ -21,13 +26,36 @@ def get_pod_details(namespace='drdroid'):
             config.load_kube_config()
 
         v1 = client.CoreV1Api()
-        
-        # Get all pods in the namespace
-        pods = v1.list_namespaced_pod(namespace)
-        events = v1.list_namespaced_event(namespace)
-        
+
+        # Get all pods in the namespace with timeout
+        pods = v1.list_namespaced_pod(namespace, _request_timeout=K8S_API_TIMEOUT)
+        events = v1.list_namespaced_event(namespace, _request_timeout=K8S_API_TIMEOUT)
+
+        # Get pod metrics for CPU/memory usage
+        pod_metrics = {}
+        try:
+            custom_api = client.CustomObjectsApi()
+            metrics = custom_api.list_namespaced_custom_object(
+                group="metrics.k8s.io",
+                version="v1beta1",
+                namespace=namespace,
+                plural="pods",
+                _request_timeout=K8S_API_TIMEOUT
+            )
+            for item in metrics.get('items', []):
+                pod_name = item['metadata']['name']
+                container_metrics = {}
+                for container in item.get('containers', []):
+                    container_metrics[container['name']] = {
+                        'cpu': container['usage'].get('cpu', 'unknown'),
+                        'memory': container['usage'].get('memory', 'unknown')
+                    }
+                pod_metrics[pod_name] = container_metrics
+        except Exception as e:
+            logger.warning(f"Could not fetch pod metrics (metrics-server may not be installed): {e}")
+
         pod_data = []
-        
+
         for pod in pods.items:
             # Extract container statuses
             container_statuses = []
@@ -40,7 +68,7 @@ def get_pod_details(namespace='drdroid'):
                         state = "terminated"
                     elif container.state.waiting:
                         state = "waiting"
-                        
+
                     last_state = "unknown"
                     if container.last_state.running:
                         last_state = "running"
@@ -49,13 +77,21 @@ def get_pod_details(namespace='drdroid'):
                     elif container.last_state.waiting:
                         last_state = "waiting"
 
-                    container_statuses.append({
+                    container_info = {
                         "name": container.name,
                         "state": state,
                         "last_state": last_state,
                         "ready": container.ready,
                         "restart_count": container.restart_count
-                    })
+                    }
+
+                    # Add resource usage if available
+                    if pod.metadata.name in pod_metrics:
+                        container_usage = pod_metrics[pod.metadata.name].get(container.name, {})
+                        container_info["cpu_usage"] = container_usage.get('cpu', 'unknown')
+                        container_info["memory_usage"] = container_usage.get('memory', 'unknown')
+
+                    container_statuses.append(container_info)
 
             # Extract events for this pod
             pod_events = []
@@ -75,38 +111,49 @@ def get_pod_details(namespace='drdroid'):
                 "containers": container_statuses,
                 "events": pod_events
             })
-            
+
         return pod_data
 
     except Exception as e:
-        logger.error(f"Error fetching pod details: {e}")
+        logger.error(f"Error fetching pod details: {e}", exc_info=True)
         return []
 
 
-@shared_task(max_retries=3, default_retry_delay=10)
+@shared_task
 def send_ping_to_drd_cloud():
     drd_cloud_host = settings.DRD_CLOUD_API_HOST
     drd_cloud_api_token = settings.DRD_CLOUD_API_TOKEN
     commit_hash = settings.VPC_AGENT_COMMIT_HASH
     current_epoch = current_epoch_timestamp()
 
-    # Get pod details
-    pod_details = get_pod_details()
-
-    # Establish reachability with DRD Cloud
-    payload = {
-        'commit_hash': commit_hash,
-        'pods': pod_details
-    }
-
-    response = requests.post(f'{drd_cloud_host}/connectors/proxy/ping',
-                        headers={'Authorization': f'Bearer {drd_cloud_api_token}'},
-                        json=payload)
-
-    if response.status_code != 200:
-        logger.error(f'Failed to connect to DRD Cloud at {current_epoch} with code: {response.status_code} '
-                     f'and response {response.text}')
+    try:
+        # Get pod details
+        pod_details = get_pod_details()
+
+        # Establish reachability with DRD Cloud
+        payload = {
+            'commit_hash': commit_hash,
+            'pods': pod_details
+        }
+
+        response = requests.post(
+            f'{drd_cloud_host}/connectors/proxy/ping',
+            headers={'Authorization': f'Bearer {drd_cloud_api_token}'},
+            json=payload,
+            timeout=HTTP_REQUEST_TIMEOUT
+        )
+
+        if response.status_code != 200:
+            logger.error(f'Failed to connect to DRD Cloud at {current_epoch} with code: {response.status_code} '
+                         f'and response {response.text}')
+            return False
+        else:
+            logger.info(f'Successfully connected to DRD Cloud at {current_epoch}')
+        return True
+
+    except (RequestException, Timeout) as e:
+        logger.error(f'Request error while pinging DRD Cloud at {current_epoch}: {str(e)}')
+        return False
+    except Exception as e:
+        logger.error(f'Unexpected error while pinging DRD Cloud at {current_epoch}: {str(e)}')
         return False
-    else:
-        logger.info(f'Successfully connected to DRD Cloud at {current_epoch}')
-    return True