diff --git a/python/cuopt_server/cuopt_server/cuopt_proxy.py b/python/cuopt_server/cuopt_server/cuopt_proxy.py index 709c5dd5c7..26a7f1e999 100644 --- a/python/cuopt_server/cuopt_server/cuopt_proxy.py +++ b/python/cuopt_server/cuopt_server/cuopt_proxy.py @@ -16,9 +16,14 @@ import cuopt_server.utils.settings as settings from cuopt_server._version import __version__ -from cuopt_server.utils.logutil import message_init - -log_fmt = "%(asctime)s.%(msecs)03d %(levelname)s %(message)s" +from cuopt_server.utils.logutil import ( + get_ncaid, + get_requestid, + get_solverid, + message_init, +) + +log_fmt = "%(ncaid)s%(requestid)s%(asctime)s.%(msecs)03d %(levelname)s %(message)s%(solverid)s" # noqa date_fmt = "%Y-%m-%d %H:%M:%S" @@ -165,6 +170,22 @@ def _configure_logging(args: argparse.Namespace) -> None: handlers=handlers, force=True, ) + log_factory = logging.getLogRecordFactory() + + def record_factory(*args, **kwargs): + record = log_factory(*args, **kwargs) + record.ncaid = get_ncaid() + record.requestid = get_requestid() + record.solverid = get_solverid() + if record.ncaid: + record.ncaid = f"NCA_ID={record.ncaid} " + if record.requestid: + record.requestid = f"NVCF_REQID={record.requestid} " + if record.solverid: + record.solverid = f" (GPU {record.solverid})" + return record + + logging.setLogRecordFactory(record_factory) def main(argv: Sequence[str] | None = None) -> None: diff --git a/python/cuopt_server/cuopt_server/proxy_webserver.py b/python/cuopt_server/cuopt_server/proxy_webserver.py index fd0ab33a71..0b24aac473 100644 --- a/python/cuopt_server/cuopt_server/proxy_webserver.py +++ b/python/cuopt_server/cuopt_server/proxy_webserver.py @@ -4,6 +4,7 @@ """FastAPI app for the gRPC-backed HTTP proxy (LP/MILP/VRP).""" import asyncio +import contextlib import logging import os import threading @@ -36,14 +37,19 @@ IncumbentSolutionResponse, LogResponse, LogResponseModel, + ManagedRequestResponse, RequestResponse, RequestStatusModel, SolutionResponse, ValidationErrorResponse, + cuoptDataInternal, + cuoptdataschema, lp_example_data, lp_msgpack_example_data, lp_zlib_example_data, lpschema, + managed_lp_example_data, + managed_vrp_example_data, ) from cuopt_server.utils.exceptions import ( exception_handler, @@ -87,6 +93,7 @@ validate_file_path, write_result_file, ) +from cuopt_server.utils.logutil import message, set_ncaid, set_requestid from cuopt_server.utils.routing.conversion import ( create_data_model as create_routing_data_model, create_solver as create_routing_solver, @@ -116,6 +123,13 @@ _incumbent_locks = {} _max_request_size = 1024 * 1024 * 1024 +# Managed POST /cuopt/cuopt polls job status from the event loop rather than +# calling Client.wait, which would hold a thread-pool worker for the whole +# solve and starve other requests, including the health check. Back off up to +# _STATUS_POLL_MAX so long solves do not poll thousands of times. +_STATUS_POLL_MIN = 0.05 +_STATUS_POLL_MAX = 1.0 + _ROUTING_KEYS = { "cost_matrix_data", "task_data", @@ -553,6 +567,31 @@ def _deserialize_convert_submit( data = load_optimization_file(file_path, warnings) else: data = deserialize(ctype, buf) + return _convert_and_submit( + data, + warnings, + validation_only, + incumbent_solutions, + solver_logs, + accept, + result_file, + warmstart_id, + initial_ids, + ) + + +def _convert_and_submit( + data, + warnings, + validation_only, + incumbent_solutions, + solver_logs, + accept, + result_file, + warmstart_id, + initial_ids, +): + """Convert a decoded problem body and submit it over gRPC.""" if _looks_like_routing(data): initials = _collect_vrp_initials(initial_ids) data_model, solver_settings, vehicle_ids, task_ids = _prepare_vrp( @@ -589,6 +628,7 @@ def _deserialize_convert_submit( ) return job_id job_id = get_grpc_routing_client().submit(data_model, solver_settings) + logging.info(message(f"sent VRP job {job_id} to gRPC")) _store_job( job_id, { @@ -643,6 +683,7 @@ def _deserialize_convert_submit( solver_settings, enable_incumbents=incumbents_enabled, ) + logging.info(message(f"sent LP job {job_id} to gRPC")) _store_job( job_id, { @@ -904,6 +945,50 @@ def deleterequest( return encode(exception_handler(e), accept) +def _result_envelope(job_id, meta, kind, req_id="", cache_warmstart=False): + """Build the legacy solution envelope for a finished job. + + Returns ``(None, [], [])`` when the result is not available yet. + + ``cache_warmstart`` populates the warmstart cache from an LP solution so a + later GET of the warmstart route does not have to refetch and reparse the + result. Callers that release the job before returning leave it off. + """ + result_kind, sol = _result_for_job(job_id, meta, kind) + if sol is None: + return None, [], [] + notes = [] + warnings = [] if meta is None else list(meta.get("warnings") or []) + solve_time = 0 + if result_kind == "vrp": + inner = routing_solution_to_http( + sol, + vehicle_ids=None if meta is None else meta.get("vehicle_ids"), + task_ids=None if meta is None else meta.get("task_ids"), + ) + if inner.get("status") == 1: + notes.append(sol.get("status_message") or "") + notes = [n for n in notes if n] + else: + if cache_warmstart: + _store_warmstart(job_id, _warmstart_dict_from_sol(sol)) + inner = solution_to_http(sol, include_warmstart=False) + try: + notes.append(sol.get_termination_reason()) + except Exception: + pass + if inner.get("solution"): + solve_time = inner["solution"].get("solver_time") or 0 + envelope = make_response( + {"solver_response": inner}, + warnings=warnings, + notes=notes, + reqId=req_id, + total_solve_time=solve_time, + ) + return envelope, warnings, notes + + @app.get( "/cuopt/solution/{id}/warmstart", include_in_schema=False, @@ -955,44 +1040,11 @@ def getsolution( status_code=409, detail=f"job {id} {_status_name(status).lower()}", ) - result_kind, sol = _result_for_job(id, meta, kind) - if sol is None: + envelope, warnings, notes = _result_envelope( + id, meta, kind, req_id=id, cache_warmstart=True + ) + if envelope is None: return encode({"reqId": id}, accept) - notes = [] - warnings = [] if meta is None else list(meta.get("warnings") or []) - if result_kind == "vrp": - inner = routing_solution_to_http( - sol, - vehicle_ids=None if meta is None else meta.get("vehicle_ids"), - task_ids=None if meta is None else meta.get("task_ids"), - ) - if inner.get("status") == 1: - notes.append(sol.get("status_message") or "") - solve_time = 0 - envelope = make_response( - {"solver_response": inner}, - warnings=warnings, - notes=[n for n in notes if n], - reqId=id, - total_solve_time=solve_time, - ) - else: - _store_warmstart(id, _warmstart_dict_from_sol(sol)) - inner = solution_to_http(sol, include_warmstart=False) - try: - notes.append(sol.get_termination_reason()) - except Exception: - pass - solve_time = 0 - if inner.get("solution"): - solve_time = inner["solution"].get("solver_time") or 0 - envelope = make_response( - {"solver_response": inner}, - warnings=warnings, - notes=notes, - reqId=id, - total_solve_time=solve_time, - ) resultdir, maxresult, mode = settings.get_result_dir() result_file = "" if meta is None else meta.get("result_file") or "" if result_file and resultdir: @@ -1039,11 +1091,210 @@ def getrequest( return encode(exception_handler(e), accept) +def _submit_managed_job(ctype, buf, accept): + """Validate and submit a POST /cuopt/cuopt body, returning the job id.""" + body = deserialize(ctype, buf) + try: + wrapper = cuoptDataInternal.parse_obj(body) + except (RequestValidationError, ValidationError): + raise + except Exception as e: + raise HTTPException( + status_code=422, + detail="unable to validate optimization data stream, %s" % str(e), + ) + + if wrapper.data is None: + # NVCF asset files are the only way legacy allowed a null body here + raise HTTPException( + status_code=422, + detail="data is required, NVCF assets are not supported", + ) + + warnings = check_client_version(wrapper.client_version or "") + validation_only = (wrapper.action or "").endswith("Validator") + + job_id = _convert_and_submit( + wrapper.data, + warnings, + validation_only, + False, + False, + accept, + "", + "", + None, + ) + logging.info(message(f"submitted job {job_id}")) + return job_id + + +async def _poll_job_status(job_id): + """Return the terminal gRPC status, yielding the loop while job runs.""" + client = get_grpc_client() + interval = _STATUS_POLL_MIN + while True: + status = await asyncio.to_thread(client.status, job_id) + if not _is_status(status, "QUEUED", "PROCESSING"): + return status + await asyncio.sleep(interval) + interval = min(interval * 2, _STATUS_POLL_MAX) + + +def _release_managed_job(job_id): + try: + get_grpc_client().delete(job_id) + except Exception: + logging.warning(f"could not delete job {job_id}", exc_info=True) + _pop_job(job_id) + + +async def _submit_wait_solution(ctype, buf, accept): + """Submit, wait, and return a solution for POST /cuopt/cuopt. + + The managed endpoint is stateless: the gRPC job and the proxy metadata + are released before returning, so there is nothing left to poll or + delete afterwards. + """ + job_id = await asyncio.to_thread(_submit_managed_job, ctype, buf, accept) + meta = _get_job(job_id) + if meta is not None and meta.get("validation_only"): + envelope = dict(meta["validation_result"]) + envelope.pop("reqId", None) + _pop_job(job_id) + logging.info(message(f"validation-only job {job_id} complete")) + return envelope + + kind = None if meta is None else meta.get("kind") + try: + logging.info(message(f"waiting for job {job_id}")) + try: + status = await _poll_job_status(job_id) + except Exception: + logging.error( + message(f"gRPC wait failed for job {job_id}"), + exc_info=True, + ) + raise + status_name = _status_name(status) + logging.info(message(f"gRPC status for job {job_id} is {status_name}")) + if not _is_status(status, "COMPLETED"): + logging.error( + message( + f"gRPC job {job_id} finished unsuccessfully: {status_name}" + ) + ) + raise HTTPException( + status_code=409, + detail=f"job {status_name.lower()}", + ) + envelope, _warnings, _notes = await asyncio.to_thread( + _result_envelope, job_id, meta, kind + ) + if envelope is None: + logging.error(message(f"gRPC job {job_id} returned no solution")) + raise HTTPException( + status_code=500, detail="solver returned no solution" + ) + logging.info(message(f"received result for job {job_id}")) + logging.info({"cuopt_complete": status_name}) + return envelope + finally: + # shielded so a client disconnect mid-solve still frees the job + cleanup = asyncio.ensure_future( + asyncio.to_thread(_release_managed_job, job_id) + ) + with contextlib.suppress(asyncio.CancelledError): + await asyncio.shield(cleanup) + + @app.post( "/cuopt/cuopt", + description=( + "Note: This is for the managed service, and users will never call " + "this API directly. Takes all the data and options at once, solves " + "any type of cuOpt problem and returns the result. If you are " + "self-hosting cuOpt, use /cuopt/request instead." + ), + include_in_schema=False, + summary="Managed Service Endpoint", + responses=ManagedRequestResponse, + openapi_extra={ + "requestBody": { + "content": { + "application/json": { + "schema": cuoptdataschema, + "examples": { + "VRP request": {"value": managed_vrp_example_data}, + "LP request": {"value": managed_lp_example_data}, + }, + }, + }, + "required": True, + } + }, ) -async def post_cuopt_sync(): - _not_implemented("POST /cuopt/cuopt") +async def cuopt( + request: Request, + accept: str = Header(default="application/json"), + content_type: str = Header(default="application/json"), + content_length: int = Header(default=0), + nvcf_ncaid: str = Header(default=""), + nvcf_reqid: str = Header(default=""), +): + ctype = content_type + if accept in mime_wild: + accept = mime_json + + try: + await asyncio.to_thread(_require_grpc_healthy) + set_ncaid(nvcf_ncaid) + set_requestid(nvcf_reqid) + + # msgpack is allowed for local testing; NVCF itself only uses json + if ctype not in [mime_json, mime_msgpack]: + raise HTTPException( + status_code=415, + detail=f"Unsupported Content-Type value {ctype}, " + f"supported values are {[mime_json, mime_msgpack]}", + ) + if accept not in [mime_json, mime_msgpack]: + raise HTTPException( + status_code=415, + detail=f"Unsupported Accept value {accept}, " + f"supported values are {[mime_json, mime_msgpack]}", + ) + + sz = int(content_length) + if sz < 0: + raise HTTPException( + status_code=422, detail="Content-Length must be non-negative" + ) + if sz > _max_request_size: + raise HTTPException( + status_code=413, + detail=( + f"Content-Length exceeds maximum of " + f"{_max_request_size} bytes" + ), + ) + if sz == 0: + raise HTTPException(status_code=422, detail="Data length is zero") + + buf = bytearray(sz) + await get_data(buf, request) + + envelope = await _submit_wait_solution(ctype, buf, accept) + return encode(envelope, accept, job_result=True) + + except (RequestValidationError, ValidationError) as e: + return encode(validation_exception_handler(e), accept) + + except HTTPException as e: + return encode(http_exception_handler(e), accept) + + except Exception as e: + return encode(exception_handler(e), accept) @app.post( diff --git a/python/cuopt_server/cuopt_server/tests/test_grpc_http_proxy.py b/python/cuopt_server/cuopt_server/tests/test_grpc_http_proxy.py index 9a2fe7659f..0e9440d6f0 100644 --- a/python/cuopt_server/cuopt_server/tests/test_grpc_http_proxy.py +++ b/python/cuopt_server/cuopt_server/tests/test_grpc_http_proxy.py @@ -2,10 +2,12 @@ # SPDX-License-Identifier: Apache-2.0 import json +import logging import socket import threading import time import uuid +from concurrent.futures import ThreadPoolExecutor from types import SimpleNamespace import pytest @@ -181,6 +183,10 @@ def __init__(self): self.submitted = [] self.cancelled = [] self.deleted = [] + self.waits = [] + self.status_calls = [] + self.result_calls = [] + self.pending_statuses = 0 self._incumbents = {} self._logs = {} @@ -202,9 +208,18 @@ def submit(self, problem, settings, enable_incumbents=None): return job_id def status(self, job_id): + self.status_calls.append(job_id) + if self.pending_statuses > 0: + self.pending_statuses -= 1 + return FakeJobStatus.PROCESSING + return self.jobs.get(job_id, FakeJobStatus.NOT_FOUND) + + def wait(self, job_id, timeout=None): + self.waits.append(timeout) return self.jobs.get(job_id, FakeJobStatus.NOT_FOUND) def result(self, job_id, variable_names=None): + self.result_calls.append(job_id) status = self.jobs.get(job_id) if status in (FakeJobStatus.FAILED, FakeJobStatus.CANCELLED): raise RuntimeError(f"job {status.name.lower()}") @@ -531,6 +546,27 @@ def test_warmstart_get_and_reuse(proxy): assert list(ws.current_primal_solution) == [0.1, 0.2] +def test_getsolution_caches_warmstart(proxy): + import cuopt_server.proxy_webserver as pw + + url, fake = proxy + req_id = requests.post( + url + "/cuopt/request", + headers={"CLIENT-VERSION": "custom"}, + json=_lp(), + ).json()["reqId"] + + assert requests.get(url + f"/cuopt/solution/{req_id}").status_code == 200 + assert pw._cached_warmstart(req_id) is not None + + # Cached by the solution GET, so the warmstart route serves it without + # refetching the result over gRPC. + calls = len(fake.result_calls) + warm = requests.get(url + f"/cuopt/solution/{req_id}/warmstart") + assert warm.status_code == 200, warm.text + assert fake.result_calls[calls:] == [] + + def test_warmstart_missing_id_is_404(proxy): url, _ = proxy missing = str(uuid.uuid4()) @@ -858,10 +894,249 @@ def test_vrp_solution_after_sidecar_lost(proxy): assert "vehicle_data" in sol.json()["response"]["solver_response"] -def test_post_solution_and_sync_are_501(proxy): +def test_sync_cuopt_lp(proxy): + url, fake = proxy + res = requests.post( + url + "/cuopt/cuopt", + json={ + "action": "cuOpt_LP", + "data": _lp(), + "client_version": "custom", + }, + ) + assert res.status_code == 200, res.text + body = res.json() + assert "reqId" not in body + assert body["response"]["solver_response"]["status"] == "Optimal" + # the managed path is stateless: no job or result is left behind + job_id = fake.submitted[0]["id"] + assert job_id in fake.deleted + # the endpoint polls status instead of blocking a thread in Client.wait + assert fake.waits == [] + import cuopt_server.proxy_webserver as pw + + with pw._jobs_lock: + assert job_id not in pw._jobs + + +def test_sync_cuopt_logs_job_lifecycle(proxy, caplog): + url, fake = proxy + with caplog.at_level(logging.INFO): + res = requests.post( + url + "/cuopt/cuopt", + json={ + "action": "cuOpt_LP", + "data": _lp(), + "client_version": "custom", + }, + ) + assert res.status_code == 200, res.text + job_id = fake.submitted[0]["id"] + text = caplog.text + assert f"sent LP job {job_id} to gRPC" in text + assert f"submitted job {job_id}" in text + assert f"waiting for job {job_id}" in text + assert f"gRPC status for job {job_id} is COMPLETED" in text + assert f"received result for job {job_id}" in text + + +def test_sync_cuopt_vrp(proxy): + url, fake = proxy + res = requests.post( + url + "/cuopt/cuopt", + json={ + "action": "cuOpt_OptimizedRouting", + "data": _vrp(), + "client_version": "custom", + }, + ) + assert res.status_code == 200, res.text + body = res.json() + assert "reqId" not in body + solver_response = body["response"]["solver_response"] + assert solver_response["vehicle_data"]["veh-1"]["task_id"] == ["A"] + assert fake.routing.submitted[0]["id"] in fake.deleted + + +def test_sync_cuopt_validator_does_not_submit(proxy): + url, fake = proxy + res = requests.post( + url + "/cuopt/cuopt", + json={ + "action": "cuOpt_LPValidator", + "data": _lp(), + "client_version": "custom", + }, + ) + assert res.status_code == 200, res.text + body = res.json() + assert "reqId" not in body + assert body["notes"] == ["Input is valid"] + assert fake.submitted == [] + + +def test_sync_cuopt_logs_nvcf_ids(proxy): + from cuopt_server.utils.logutil import get_ncaid, get_requestid + + url, _ = proxy + seen = {} + + async def _capture(ctype, buf, accept): + seen["ncaid"] = get_ncaid() + seen["reqid"] = get_requestid() + return make_response({"solver_response": {"status": 0}}) + + import cuopt_server.proxy_webserver as pw + + original = pw._submit_wait_solution + pw._submit_wait_solution = _capture + try: + res = requests.post( + url + "/cuopt/cuopt", + headers={"NVCF-NCAID": "nca-1", "NVCF-REQID": "req-1"}, + json={"action": "cuOpt_LP", "data": _lp()}, + ) + finally: + pw._submit_wait_solution = original + assert res.status_code == 200, res.text + assert seen == {"ncaid": "nca-1", "reqid": "req-1"} + + +def test_sync_cuopt_log_records_include_nvcf_ids(proxy, caplog): + from cuopt_server.utils.logutil import ( + get_ncaid, + get_requestid, + get_solverid, + ) + + url, fake = proxy + previous = logging.getLogRecordFactory() + + def record_factory(*args, **kwargs): + record = previous(*args, **kwargs) + record.ncaid = get_ncaid() + record.requestid = get_requestid() + record.solverid = get_solverid() + if record.ncaid: + record.ncaid = f"NCA_ID={record.ncaid} " + if record.requestid: + record.requestid = f"NVCF_REQID={record.requestid} " + if record.solverid: + record.solverid = f" (GPU {record.solverid})" + return record + + logging.setLogRecordFactory(record_factory) + try: + with caplog.at_level(logging.INFO): + res = requests.post( + url + "/cuopt/cuopt", + headers={"NVCF-NCAID": "nca-1", "NVCF-REQID": "req-1"}, + json={ + "action": "cuOpt_LP", + "data": _lp(), + "client_version": "custom", + }, + ) + finally: + logging.setLogRecordFactory(previous) + assert res.status_code == 200, res.text + job_id = fake.submitted[0]["id"] + markers = ( + f"sent LP job {job_id} to gRPC", + f"submitted job {job_id}", + f"waiting for job {job_id}", + f"received result for job {job_id}", + ) + lifecycle = [ + rec + for rec in caplog.records + if any(marker in rec.getMessage() for marker in markers) + ] + assert lifecycle, caplog.text + for rec in lifecycle: + assert rec.ncaid == "NCA_ID=nca-1 " + assert rec.requestid == "NVCF_REQID=req-1 " + + +def test_sync_cuopt_rejects_zlib_content_type(proxy): + import zlib + + url, _ = proxy + payload = zlib.compress( + json.dumps({"action": "cuOpt_LP", "data": _lp()}).encode() + ) + res = requests.post( + url + "/cuopt/cuopt", + headers={"Content-Type": mime_zlib}, + data=payload, + ) + assert res.status_code == 415, res.text + + +def test_sync_cuopt_requires_wrapped_data(proxy): + url, _ = proxy + res = requests.post(url + "/cuopt/cuopt", json=_lp()) + assert res.status_code == 422, res.text + + +def test_sync_cuopt_polls_until_job_is_done(proxy): + url, fake = proxy + fake.pending_statuses = 3 + res = requests.post( + url + "/cuopt/cuopt", + json={ + "action": "cuOpt_LP", + "data": _lp(), + "client_version": "custom", + }, + ) + assert res.status_code == 200, res.text + assert res.json()["response"]["solver_response"]["status"] == "Optimal" + job_id = fake.submitted[0]["id"] + assert fake.status_calls.count(job_id) >= 4 + assert fake.waits == [] + assert job_id in fake.deleted + + +def test_sync_cuopt_health_is_served_during_solve(proxy): + """A pending solve must not hold the thread pool that health needs.""" + url, fake = proxy + fake.pending_statuses = 5 + + with ThreadPoolExecutor(max_workers=1) as pool: + solve = pool.submit( + requests.post, + url + "/cuopt/cuopt", + json={ + "action": "cuOpt_LP", + "data": _lp(), + "client_version": "custom", + }, + timeout=8, + ) + deadline = time.monotonic() + 5 + while fake.pending_statuses > 3: + if time.monotonic() > deadline: + pytest.fail("solve never started polling") + time.sleep(0.01) + health = requests.get(url + "/cuopt/health", timeout=5) + assert health.status_code == 200, health.text + assert solve.result(timeout=10).status_code == 200 + + +def test_sync_cuopt_rejects_null_data(proxy): + url, _ = proxy + res = requests.post( + url + "/cuopt/cuopt", + json={"action": "cuOpt_LP", "data": None}, + ) + assert res.status_code == 422, res.text + assert "NVCF assets" in res.json()["error"] + + +def test_post_solution_and_wildcards_are_501(proxy): url, _ = proxy assert requests.post(url + "/cuopt/solution", json={}).status_code == 501 - assert requests.post(url + "/cuopt/cuopt", json={}).status_code == 501 assert requests.delete(url + "/cuopt/request/*").status_code == 501 assert ( requests.delete(