Skip to content

Remove readiness cache #419

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
timmarkhuff wants to merge 7 commits into
base: main
Choose a base branch
from
Open

Remove readiness cache #419

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
0aa4178
checkpoint
May 27, 2026
f04457e
removing the readiness cache
May 27, 2026
49a740a
Automatically reformatting code with black and isort
May 27, 2026
20e652e
changing script name and adding test
May 27, 2026
e1683e1
Merge branch 'tim/fix-readiness-bug' of github.com:groundlight/edge-e…
May 27, 2026
80e6d8d
removing unneeded test
May 27, 2026
39d1858
cleaning up comments
May 27, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion app/api/routes/edge_detector_readiness.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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()

Expand All @@ -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}
248 changes: 127 additions & 121 deletions app/api/routes/image_queries.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down
57 changes: 26 additions & 31 deletions app/core/edge_inference.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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}
Expand All @@ -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}")
Expand Down Expand Up @@ -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:
"""
Expand Down
Loading
Loading