Spaces:
Runtime error
Runtime error
File size: 3,996 Bytes
36334a9 e20f133 36334a9 e20f133 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 |
import os
import time
import requests
from typing import Optional, Tuple
from ui.statusui import StatusUI
from checks.health_check import check_model_endpoint
from agents.model import load_huggingface_model
def wake_up_endpoint(
endpoint_uri: str,
ui,
max_wait: int = 300,
initial_delay: float = 3.0,
backoff_factor: float = 1.5,
max_retry_delay: float = 10.0
) -> Tuple[bool, Optional[str]]:
"""
Poll the endpoint until it responds OK or timeout.
Args:
endpoint_uri: The endpoint URL to monitor
ui: UI object for status updates
max_wait: Maximum total wait time in seconds (minimum 60s enforced)
initial_delay: Initial delay between retries in seconds
backoff_factor: Multiplier for exponential backoff
max_retry_delay: Maximum delay between retries in seconds
Returns:
Tuple of (success: bool, error_message: Optional[str])
"""
# Configuration validation
max_wait = max(max_wait, 60)
current_delay = min(initial_delay, max_retry_delay)
# Prepare request components
headers = {}
if hf_token := os.environ.get("HF_TOKEN"):
headers["Authorization"] = f"Bearer {hf_token}"
payload = {"inputs": "ping"}
timeout = min(5, current_delay * 0.8) # Ensure timeout is less than delay
start_time = time.time()
last_status = None
while (time.time() - start_time) < max_wait:
try:
# Log attempt
if endpoint_uri != last_status:
ui.append(f"Pinging endpoint: {endpoint_uri}")
last_status = endpoint_uri
# Make the request
response = requests.post(
endpoint_uri,
headers=headers,
json=payload,
timeout=timeout
)
if response.ok:
ui.append("✅ Endpoint is awake and responsive")
return True, None
# Handle specific HTTP status codes
if response.status_code in {503, 504}:
status_msg = f"Endpoint warming up (HTTP {response.status_code})"
else:
status_msg = f"Unexpected response (HTTP {response.status_code})"
ui.append(f"{status_msg}, retrying in {current_delay:.1f}s...")
except requests.exceptions.RequestException as e:
ui.append(f"Connection error ({type(e).__name__}), retrying in {current_delay:.1f}s...")
# Wait before next attempt with exponential backoff
time.sleep(current_delay)
current_delay = min(current_delay * backoff_factor, max_retry_delay)
timeout = min(5, current_delay * 0.8)
# Timeout reached
error_msg = f"❌ Timed out after {max_wait}s waiting for endpoint"
ui.append(error_msg)
return False, error_msg
def run_status_checks():
"""Run all status checks and return endpoint URI if successful"""
ui = StatusUI("Content Agent Status Checks")
ui.launch()
ui.append("Starting prechecks...")
ui.append("Checking endpoint..")
endpoint_uri = load_huggingface_model() # Get the URI for the endpoint
ui.append(endpoint_uri)
# Wake it up before health check
wake_up_successful = wake_up_endpoint(endpoint_uri, ui)
success, error_msg = wake_up_endpoint(endpoint_uri, ui)
i
if not success:
ui.append("Warning: Could not wake up the endpoint. Exiting.")
else:
ui.append("✅ End point responded OK.")
is_healthy, status_info = check_model_endpoint(endpoint_uri) # Test the endpoint
if not is_healthy:
from checks.failed_check import create_failed_gradio_ui
interface = create_failed_gradio_ui(status_info)
interface.launch(show_error=True, share=True)
return None
return endpoint_uri |