diff --git a/app/api/routes/edge_detector_readiness.py b/app/api/routes/edge_detector_readiness.py index 2d9e96eec..db4e6b11a 100644 --- a/app/api/routes/edge_detector_readiness.py +++ b/app/api/routes/edge_detector_readiness.py @@ -2,6 +2,7 @@ from app.core.app_state import AppState, get_app_state from app.core.edge_config_manager import EdgeConfigManager +from app.core.edge_inference import check_inference_ready router = APIRouter() @@ -15,4 +16,4 @@ async def get_edge_detector_readiness(app_state: AppState = Depends(get_app_stat """ config = EdgeConfigManager.active() detector_ids = [d.detector_id for d in config.detectors] - return {did: {"ready": app_state.edge_inference_manager.inference_is_available(did)} for did in detector_ids} + return {did: {"ready": check_inference_ready(did, app_state.separate_oodd_inference)} for did in detector_ids} diff --git a/app/api/routes/image_queries.py b/app/api/routes/image_queries.py index 796969471..0fa001d4e 100644 --- a/app/api/routes/image_queries.py +++ b/app/api/routes/image_queries.py @@ -171,141 +171,147 @@ async def post_image_query( # noqa: PLR0913, PLR0915, PLR0912 # If human review is required, we should skip edge inference completely logger.debug("Received human_review=ALWAYS. Skipping edge inference.") record_activity_for_metrics(detector_id, activity_type="escalations") - elif app_state.edge_inference_manager.inference_is_available(detector_id=detector_id): - # -- Edge-model Inference -- - logger.debug(f"Local inference is available for {detector_id=}. Running inference...") - results = app_state.edge_inference_manager.run_inference( - detector_id=detector_id, image_bytes=image_bytes, content_type=content_type, mode=detector_metadata.mode - ) - ml_confidence = results["confidence"] - class_index = results["label"] - record_confidence_for_metrics(detector_id, ml_confidence, class_index=class_index) - - is_confident_enough = ml_confidence >= confidence_threshold - if not is_confident_enough: - record_activity_for_metrics(detector_id, activity_type="below_threshold_iqs", class_index=class_index) - - if return_edge_prediction or is_confident_enough: # Return the edge prediction - if return_edge_prediction: - logger.debug(f"Returning edge prediction without cloud escalation. {detector_id=}") - else: - logger.debug(f"Edge detector confidence sufficient. {detector_id=}") - - image_query = create_iq( - detector_id=detector_id, - mode=detector_metadata.mode, - mode_configuration=detector_metadata.mode_configuration, - result_value=results["label"], - confidence=ml_confidence, - confidence_threshold=confidence_threshold, - is_done_processing=True, - query=detector_metadata.query, - patience_time=patience_time, - rois=results["rois"], - text=results["text"], - mlb_key=results.get("mlb_key"), - oodd_mlb_key=results.get("oodd_mlb_key"), + else: + try: + results = app_state.edge_inference_manager.run_inference( + detector_id=detector_id, image_bytes=image_bytes, content_type=content_type, mode=detector_metadata.mode ) + except RuntimeError: + logger.debug(f"Edge inference failed for {detector_id=}, treating as unavailable.") + results = None + + if results is not None: + # -- Edge-model Inference -- + logger.debug(f"Local inference succeeded for {detector_id=}.") + ml_confidence = results["confidence"] + class_index = results["label"] + record_confidence_for_metrics(detector_id, ml_confidence, class_index=class_index) + + is_confident_enough = ml_confidence >= confidence_threshold + if not is_confident_enough: + record_activity_for_metrics(detector_id, activity_type="below_threshold_iqs", class_index=class_index) - # Skip cloud operations if escalation is disabled - if disable_cloud_escalation: - return image_query - - if is_confident_enough: # Audit confident edge predictions at the specified rate - if random.random() < edge_config.global_config.confident_audit_rate: - logger.debug( - f"Auditing confident edge prediction with confidence {ml_confidence} for detector {detector_id=}." - ) - record_activity_for_metrics(detector_id, activity_type="audits") - submit_iq_params = SubmitImageQueryParams( - patience_time=patience_time, - confidence_threshold=confidence_threshold, - human_review=human_review, - metadata=generate_metadata_dict(results=results, is_edge_audit=True), - image_query_id=image_query.id, # We give the cloud IQ the same ID as the returned edge IQ - ) - # We write to the queue synchronously because it should be fast. But this could be done as a - # background task if it becomes slow. - write_escalation_to_queue( - writer=app_state.queue_writer, - detector_id=detector_id, - image_bytes=image_bytes, - submit_iq_params=submit_iq_params, - request_id=request_id, - ) - - # We keep done_processing=True here for `image_query` because although we escalated the query for - # an audit, this is invisible to the user. From their perspective, this is the final answer. - - # Don't want to escalate to cloud again if we're already auditing the query + if return_edge_prediction or is_confident_enough: # Return the edge prediction + if return_edge_prediction: + logger.debug(f"Returning edge prediction without cloud escalation. {detector_id=}") + else: + logger.debug(f"Edge detector confidence sufficient. {detector_id=}") + + image_query = create_iq( + detector_id=detector_id, + mode=detector_metadata.mode, + mode_configuration=detector_metadata.mode_configuration, + result_value=results["label"], + confidence=ml_confidence, + confidence_threshold=confidence_threshold, + is_done_processing=True, + query=detector_metadata.query, + patience_time=patience_time, + rois=results["rois"], + text=results["text"], + mlb_key=results.get("mlb_key"), + oodd_mlb_key=results.get("oodd_mlb_key"), + ) + + # Skip cloud operations if escalation is disabled + if disable_cloud_escalation: return image_query - # Escalate after returning edge prediction if escalation is enabled and we have low confidence. - if not is_confident_enough: - # Only escalate if we haven't escalated on this detector too recently. - if app_state.edge_inference_manager.escalation_cooldown_complete(detector_id, edge_config): - logger.debug( - f"Escalating to cloud due to low confidence: {ml_confidence} < thresh={confidence_threshold}" - ) - record_activity_for_metrics(detector_id, activity_type="escalations", class_index=class_index) - submit_iq_params = SubmitImageQueryParams( - patience_time=patience_time, - confidence_threshold=confidence_threshold, - human_review=human_review, - metadata=generate_metadata_dict(results=results, is_edge_audit=False), - image_query_id=image_query.id, # We give the cloud IQ the same ID as the returned edge IQ - ) - # We write to the queue synchronously because it should be fast. But this could be done as a - # background task if it becomes slow. - write_escalation_to_queue( - writer=app_state.queue_writer, - detector_id=detector_id, - image_bytes=image_bytes, - submit_iq_params=submit_iq_params, - request_id=request_id, - ) - # Not done processing because the IQ in the cloud could get a better answer once escalated - image_query.done_processing = False - else: - logger.debug( - f"Not escalating to cloud due to rate limit on background cloud escalations: {detector_id=}" - ) + if is_confident_enough: # Audit confident edge predictions at the specified rate + if random.random() < edge_config.global_config.confident_audit_rate: + logger.debug( + f"Auditing confident edge prediction with confidence {ml_confidence} for detector {detector_id=}." + ) + record_activity_for_metrics(detector_id, activity_type="audits") + submit_iq_params = SubmitImageQueryParams( + patience_time=patience_time, + confidence_threshold=confidence_threshold, + human_review=human_review, + metadata=generate_metadata_dict(results=results, is_edge_audit=True), + image_query_id=image_query.id, # We give the cloud IQ the same ID as the returned edge IQ + ) + # We write to the queue synchronously because it should be fast. But this could be done as a + # background task if it becomes slow. + write_escalation_to_queue( + writer=app_state.queue_writer, + detector_id=detector_id, + image_bytes=image_bytes, + submit_iq_params=submit_iq_params, + request_id=request_id, + ) + + # We keep done_processing=True here for `image_query` because although we escalated the query for + # an audit, this is invisible to the user. From their perspective, this is the final answer. + + # Don't want to escalate to cloud again if we're already auditing the query + return image_query + + # Escalate after returning edge prediction if escalation is enabled and we have low confidence. + if not is_confident_enough: + # Only escalate if we haven't escalated on this detector too recently. + if app_state.edge_inference_manager.escalation_cooldown_complete(detector_id, edge_config): + logger.debug( + f"Escalating to cloud due to low confidence: {ml_confidence} < thresh={confidence_threshold}" + ) + record_activity_for_metrics(detector_id, activity_type="escalations", class_index=class_index) + submit_iq_params = SubmitImageQueryParams( + patience_time=patience_time, + confidence_threshold=confidence_threshold, + human_review=human_review, + metadata=generate_metadata_dict(results=results, is_edge_audit=False), + image_query_id=image_query.id, # We give the cloud IQ the same ID as the returned edge IQ + ) + # We write to the queue synchronously because it should be fast. But this could be done as a + # background task if it becomes slow. + write_escalation_to_queue( + writer=app_state.queue_writer, + detector_id=detector_id, + image_bytes=image_bytes, + submit_iq_params=submit_iq_params, + request_id=request_id, + ) + # Not done processing because the IQ in the cloud could get a better answer once escalated + image_query.done_processing = False + else: + logger.debug( + f"Not escalating to cloud due to rate limit on background cloud escalations: {detector_id=}" + ) - return image_query - else: - # -- Edge-inference is not available -- - # Create an edge-inference deployment record, which may be used to spin up an edge-inference server. - logger.debug(f"Local inference not available for {detector_id=}. Creating inference deployment record.") - api_token = gl.api_client.configuration.api_key["ApiToken"] - - primary_model_name = get_edge_inference_model_name(detector_id=detector_id, is_oodd=False) - app_state.db_manager.create_or_update_inference_deployment_record( - deployment={ - "model_name": primary_model_name, - "detector_id": detector_id, - "api_token": api_token, - "deployment_created": False, - } - ) + return image_query + else: + # -- Edge-inference is not available -- + # Create an edge-inference deployment record, which may be used to spin up an edge-inference server. + logger.debug(f"Local inference not available for {detector_id=}. Creating inference deployment record.") + api_token = gl.api_client.configuration.api_key["ApiToken"] - if app_state.separate_oodd_inference: - oodd_model_name = get_edge_inference_model_name(detector_id=detector_id, is_oodd=True) + primary_model_name = get_edge_inference_model_name(detector_id=detector_id, is_oodd=False) app_state.db_manager.create_or_update_inference_deployment_record( deployment={ - "model_name": oodd_model_name, + "model_name": primary_model_name, "detector_id": detector_id, "api_token": api_token, "deployment_created": False, } ) - if return_edge_prediction: - raise HTTPException( - status_code=status.HTTP_503_SERVICE_UNAVAILABLE, - detail=( - f"Edge predictions are required, but an edge-inference server is not available for {detector_id=}." - ), - ) + if app_state.separate_oodd_inference: + oodd_model_name = get_edge_inference_model_name(detector_id=detector_id, is_oodd=True) + app_state.db_manager.create_or_update_inference_deployment_record( + deployment={ + "model_name": oodd_model_name, + "detector_id": detector_id, + "api_token": api_token, + "deployment_created": False, + } + ) + + if return_edge_prediction: + raise HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + detail=( + f"Edge predictions are required, but an edge-inference server is not available for {detector_id=}." + ), + ) # Fall back to submitting the image to the cloud if disable_cloud_escalation: diff --git a/app/core/edge_inference.py b/app/core/edge_inference.py index c3bab3900..c2dfd3a7c 100644 --- a/app/core/edge_inference.py +++ b/app/core/edge_inference.py @@ -9,7 +9,6 @@ import requests import yaml -from cachetools import TTLCache, cached from fastapi import HTTPException, status from groundlight.edge import EdgeEndpointConfig, InferenceConfig from jinja2 import Template @@ -29,22 +28,36 @@ logger = logging.getLogger(__name__) -# Simple TTL cache for is_edge_inference_ready checks to avoid having to re-check every time a request is processed. -# This will be process-specific, so each edge-endpoint worker will have its own cache instance. -ttl_cache = TTLCache(maxsize=128, ttl=5) +# Short connect timeout for inference requests. If the K8s Service has no ready endpoints, +# connections will hang indefinitely without a timeout. The read timeout is left unbounded +# since inference latency varies widely across model sizes and hardware. +INFERENCE_CONNECT_TIMEOUT_SEC = 1.0 -@cached(ttl_cache) -def is_edge_inference_ready(inference_client_url: str) -> bool: - model_ready_url = f"http://{inference_client_url}/health/ready" +def _check_url_ready(url: str) -> bool: + """GET /health/ready on a single inference service URL and return True on HTTP 200.""" try: - response = requests.get(model_ready_url) + response = requests.get(f"http://{url}/health/ready", timeout=INFERENCE_CONNECT_TIMEOUT_SEC) return response.status_code == status.HTTP_200_OK - except requests.exceptions.RequestException as e: - logger.warning(f"Failed to connect to {model_ready_url}: {e}") + except requests.exceptions.RequestException: return False +def check_inference_ready(detector_id: str, separate_oodd_inference: bool) -> bool: + """Checks if inference pods for this detector are ready to serve. + + Returns True only when all required pods respond to /health/ready. + """ + primary_url = get_edge_inference_service_name(detector_id) + ":8000" + if not _check_url_ready(primary_url): + return False + if separate_oodd_inference: + oodd_url = get_edge_inference_service_name(detector_id, is_oodd=True) + ":8000" + if not _check_url_ready(oodd_url): + return False + return True + + def submit_image_for_inference(inference_client_url: str, image_bytes: bytes, content_type: str) -> dict: inference_url = f"http://{inference_client_url}/infer" headers = {"Content-Type": content_type} @@ -56,7 +69,9 @@ def submit_image_for_inference(inference_client_url: str, image_bytes: bytes, co headers["X-GL-Parent-Span-Id"] = span.span_id try: logger.debug(f"Submitting image for inference to {inference_url}") - response = requests.post(inference_url, data=image_bytes, headers=headers) + response = requests.post( + inference_url, data=image_bytes, headers=headers, timeout=(INFERENCE_CONNECT_TIMEOUT_SEC, None) + ) if response.status_code != status.HTTP_200_OK: logger.error(f"Inference server returned an error: {response.status_code} - {response.text}") raise RuntimeError(f"Inference server error: {response.status_code} - {response.text}") @@ -266,26 +281,6 @@ def __init__( self.separate_oodd_inference = separate_oodd_inference self.last_escalation_times: dict[str, float | None] = {} - @trace_span - def inference_is_available(self, detector_id: str) -> bool: - """Check whether inference pods for this detector are ready to serve.""" - primary_url = get_edge_inference_service_name(detector_id) + ":8000" - oodd_url = ( - get_edge_inference_service_name(detector_id, is_oodd=True) + ":8000" - if self.separate_oodd_inference - else None - ) - - ready = is_edge_inference_ready(primary_url) and ( - not self.separate_oodd_inference or is_edge_inference_ready(oodd_url) - ) - if not ready: - logger.debug( - f"Edge inference server and/or OODD inference server is not ready. {primary_url=}, {oodd_url=}" - ) - return False - return True - @trace_span def run_inference(self, detector_id: str, image_bytes: bytes, content_type: str, mode: ModeEnum) -> dict: """ diff --git a/app/profiling/dashboard.py b/app/profiling/dashboard.py index 5183d8433..697518302 100644 --- a/app/profiling/dashboard.py +++ b/app/profiling/dashboard.py @@ -65,7 +65,6 @@ def _(): "detector_config": "#BCBD22", "get_detector_metadata": "#EF553B", "refresh_detector_metadata_if_needed": "#D62728", - "inference_is_available": "#00CC96", "run_inference": "#3CB371", "_submit_primary_inference": "#AB63FA", "_submit_oodd_inference": "#FFA15A", diff --git a/load-testing/fresh_pod_readiness_test.py b/load-testing/fresh_pod_readiness_test.py new file mode 100644 index 000000000..283d49ef9 --- /dev/null +++ b/load-testing/fresh_pod_readiness_test.py @@ -0,0 +1,109 @@ +"""Test that inference works correctly immediately after a fresh pod rollout. + +Configures a placeholder detector first, waits for its pod to be ready, +then swaps to the real detector under test. This guarantees the real +detector's pod goes through a complete fresh rollout on every run: + - While the placeholder is active, any pre-existing pod for the real + detector is fully evicted. + - When configuration finishes applying for the real detector, the pod + has just become ready for the first time this run. + +Immediately after that second configuration is applied, the script hammers +the real detector with inference requests as fast as possible for +VICTORY_DURATION_SEC seconds. SDK and transport retries are both disabled +so any transient 5xx errors surface immediately. + +Declares victory if no errors occur; otherwise reports all failures and +exits with a non-zero status. +""" +import time + +from groundlight import Groundlight, ExperimentalApi, ApiException + +import groundlight_helpers as glh +import image_helpers as imgh + +VICTORY_DURATION_SEC = 10.0 + + +def main() -> None: + """Run the fresh-pod readiness check against the configured Edge Endpoint.""" + gl = ExperimentalApi() + glh.error_if_endpoint_is_cloud(gl) + gl_cloud = Groundlight(endpoint=glh.CLOUD_ENDPOINT_PROD) + + # Disable retries so any "no pod available" errors propagate immediately. + glh.disable_all_retries(gl) + + detector = glh.provision_detector( + gl_cloud, + "BINARY", + "Set Config Readiness Bug Repro", + group_name="Edge Endpoint Load Testing", + ) + placeholder_detector = glh.provision_detector( + gl_cloud, + "BINARY", + "Set Config Readiness Bug Repro Placeholder", + group_name="Edge Endpoint Load Testing", + ) + + # Configure the placeholder and wait for it to be ready. At that point + # the real detector's pod has been evicted (it's no longer in the config). + print("Configuring placeholder detector to evict any existing pod for the real detector...") + glh.configure_edge_endpoint(gl, placeholder_detector) + + # Now configure the real detector. The pod is guaranteed to be freshly + # created -- there is no carry-over from a prior run. + glh.configure_edge_endpoint(gl, detector) + + print( + "\nDetector configuration has been applied. " + f"Hammering detector for {VICTORY_DURATION_SEC}s with all retries disabled...\n" + ) + + test_start = time.time() + failures = [] + iteration = 0 + + while time.time() - test_start < VICTORY_DURATION_SEC: + image, _, _ = imgh.generate_random_binary_image() + try: + iq = gl.submit_image_query( + detector=detector, + image=image, + **glh.IQ_KWARGS_FOR_NO_ESCALATION, + ) + print( + f"[{iteration}] ok | label={iq.result.label.value} | " + f"confidence={iq.result.confidence:.2f} | from_edge={iq.result.from_edge}" + ) + except ApiException as e: + elapsed = time.time() - test_start + body_preview = str(getattr(e, "body", "") or "")[:300] + failures.append({ + "iteration": iteration, + "elapsed_sec": elapsed, + "status": e.status, + "body": body_preview, + }) + print( + f"[{iteration}] FAIL at t={elapsed:.2f}s | " + f"status={e.status} | body={body_preview}" + ) + iteration += 1 + + elapsed = time.time() - test_start + + if failures: + print( + f"\nBug reproduced: {len(failures)}/{iteration} requests failed " + f"over {elapsed:.1f}s." + ) + raise SystemExit(1) + + print(f"\nVictory! {iteration} requests over {elapsed:.1f}s with no errors.") + + +if __name__ == "__main__": + main() diff --git a/load-testing/groundlight_helpers.py b/load-testing/groundlight_helpers.py index 5e1114429..10610fd52 100644 --- a/load-testing/groundlight_helpers.py +++ b/load-testing/groundlight_helpers.py @@ -8,6 +8,7 @@ import requests import json import time +import types import yaml from tqdm import trange @@ -30,6 +31,18 @@ PRIMING_LABELS_PER_CLASS = 5 +def disable_all_retries(gl: ExperimentalApi) -> None: + """Disable all SDK and transport-level retries so 5xx errors surface immediately. + + This reaches into SDK internals: it strips the RequestsRetryDecorator from + call_api via __wrapped__ and sets urllib3 retries to 0. May break if the + SDK's internal retry machinery changes. + """ + gl.configuration.retries = 0 + api = gl.api_client + api.call_api = types.MethodType(api.call_api.__wrapped__, api) + + def hash_pipeline_config(pipeline_config: str) -> str: """Return a short deterministic hash of the pipeline config string.""" return hashlib.sha256(pipeline_config.encode()).hexdigest()[:12] diff --git a/load-testing/rollouts_under_inference_load.py b/load-testing/rollouts_under_inference_load.py index 4f949f472..3febb01de 100644 --- a/load-testing/rollouts_under_inference_load.py +++ b/load-testing/rollouts_under_inference_load.py @@ -1,7 +1,6 @@ from groundlight import Groundlight, ExperimentalApi, Detector import subprocess import threading -import types import time import groundlight_helpers as glh @@ -14,19 +13,6 @@ MIN_ROLLOUTS = 2 -def disable_transport_retries(gl: ExperimentalApi) -> None: - """Disable urllib3 transport-level retries.""" - gl.configuration.retries = 0 - - -def disable_sdk_retries(gl: ExperimentalApi) -> None: - """Monkeypatch the SDK's internal API client to remove the RequestsRetryDecorator - from call_api, so that 5xx errors propagate immediately without retries. - NOTE: This reaches into SDK internals and may break if the retry decorator changes.""" - api = gl.api_client - api.call_api = types.MethodType(api.call_api.__wrapped__, api) - - def get_max_inference_revision() -> int: """Get the highest revision number across all inferencemodel deployments.""" result = subprocess.run( @@ -61,8 +47,7 @@ def main(): gl_cloud = Groundlight(endpoint=glh.CLOUD_ENDPOINT_PROD) # Disable internal retries in the python-sdk so that this test surfaces all errors, even if transient - disable_transport_retries(gl) - disable_sdk_retries(gl) + glh.disable_all_retries(gl) detector = glh.provision_detector( gl_cloud, "BINARY", "Rollout Under Load Test", diff --git a/test/api/test_image_queries.py b/test/api/test_image_queries.py index fce748d28..5298ea5f6 100644 --- a/test/api/test_image_queries.py +++ b/test/api/test_image_queries.py @@ -114,33 +114,29 @@ def assert_not_escalated_to_gl(detector: Detector | None = None): def enable_edge_inference( *, edge_response: dict | None = None, assert_ran: bool = False, assert_didnt_run: bool = False ): - """ - Context manager to mock everything for supporting edge inference. Returns that edge_inference - is available (as if the inference deployment is ready), and enables/requires the user to specify - the exact contents of the edge_response. - # TODO: better mocking support for edge_response + """Context manager that mocks edge inference for tests. + + By default (no edge_response), run_inference raises RuntimeError to simulate no pod available. + When edge_response is provided, run_inference returns that response to simulate successful inference. """ if assert_ran and assert_didnt_run: raise ValueError("Conflicting assertions configured.") - edge_response = edge_response or {} mock_edge_inference_manager = mock.Mock() - mock_edge_inference_manager.inference_is_available.return_value = True - mock_edge_inference_manager.run_inference.return_value = edge_response - - # We need to inject the edge_inference_manager mock via `get_app_state`, so - # we need to mock AppState as well. It would be nicer if we had more loosely - # coupled dependency injection here. - mock_app_state = mock.Mock() - mock_app_state.edge_inference_manager = mock_edge_inference_manager + if edge_response is not None: + mock_edge_inference_manager.run_inference.return_value = edge_response + else: + mock_edge_inference_manager.run_inference.side_effect = RuntimeError("Edge inference not available") - with mock.patch("app.api.routes.image_queries.get_app_state") as mock_app_state: + # Patch get_app_state and wire in our mock EdgeInferenceManager. + with mock.patch("app.api.routes.image_queries.get_app_state") as mock_get_app_state: + mock_get_app_state.return_value.edge_inference_manager = mock_edge_inference_manager yield mock_edge_inference_manager - if assert_ran: # assert that a request was sent to the inference server - mock_edge_inference_manager.inference_is_available.assert_called_once() + if assert_ran: + mock_edge_inference_manager.run_inference.assert_called_once() if assert_didnt_run: - mock_edge_inference_manager.inference_is_available.assert_not_called() + mock_edge_inference_manager.run_inference.assert_not_called() # @@ -156,7 +152,7 @@ def test_post_image_query(test_client: TestClient, detector: Detector): with assert_escalated_to_gl( submitted_with={"confidence_threshold": threshold}, detector=detector, sdk_response=confident_cloud_iq ): - with enable_edge_inference(assert_didnt_run=True): # No inference deployments configured yet + with enable_edge_inference(): # No deployment: run_inference raises RuntimeError, falls back to cloud response = test_client.post( url, headers={"Content-Type": "image/jpeg"},