Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
6 changes: 6 additions & 0 deletions cpp/include/cuopt/grpc/cython_grpc_client.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -130,6 +130,12 @@ class grpc_python_client_t {

bool connect(std::string& error_out);

/**
* Short CheckStatus probe (default 5s deadline). False if the server
* does not answer; error_out is populated from get_last_error().
*/
bool ping(std::string& error_out, int timeout_seconds = 5);

grpc_submit_result_t submit(
cuopt::mathematical_optimization::io::data_model_view_t<int, double>* data_model,
cuopt::mathematical_optimization::solver_settings_t<int, double>* settings,
Expand Down
9 changes: 9 additions & 0 deletions cpp/src/grpc/client/cython_grpc_client.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -97,6 +97,15 @@ bool grpc_python_client_t::connect(std::string& error_out)
return true;
}

bool grpc_python_client_t::ping(std::string& error_out, int timeout_seconds)
{
if (!impl_->client.ping(timeout_seconds)) {
error_out = impl_->client.get_last_error();
return false;
}
return true;
}

grpc_submit_result_t grpc_python_client_t::submit(
cuopt::mathematical_optimization::io::data_model_view_t<int, double>* data_model,
cuopt::mathematical_optimization::solver_settings_t<int, double>* settings,
Expand Down
52 changes: 33 additions & 19 deletions cpp/src/grpc/client/grpc_client.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -228,25 +228,11 @@ bool grpc_client_t::connect()
"[grpc_client] Connecting to " << config_.server_address
<< (config_.enable_tls ? " (TLS)" : ""));

// Verify connectivity with a lightweight RPC probe. Channel-level checks like
// WaitForConnected are unreliable (gRPC lazy connection on localhost can
// report READY even without a server). A real RPC with a deadline is the
// only reliable way to confirm the server is reachable.
{
grpc::ClientContext probe_ctx;
probe_ctx.set_deadline(std::chrono::system_clock::now() + std::chrono::seconds(5));
cuopt::remote::StatusRequest probe_req;
probe_req.set_job_id("__connection_probe__");
cuopt::remote::StatusResponse probe_resp;
auto probe_status = impl_->stub->CheckStatus(&probe_ctx, probe_req, &probe_resp);

auto code = probe_status.error_code();
if (code != grpc::StatusCode::OK && code != grpc::StatusCode::NOT_FOUND) {
last_error_ = "Failed to connect to server at " + config_.server_address + " (" +
probe_status.error_message() + ")";
GRPC_CLIENT_DEBUG_LOG(config_, "[grpc_client] Connection failed: " << last_error_);
return false;
}
if (!ping(5)) {
last_error_ =
"Failed to connect to server at " + config_.server_address + " (" + last_error_ + ")";
GRPC_CLIENT_DEBUG_LOG(config_, "[grpc_client] Connection failed: " << last_error_);
return false;
}

GRPC_CLIENT_DEBUG_LOG(config_,
Expand All @@ -264,6 +250,34 @@ bool grpc_client_t::is_connected() const
return state == GRPC_CHANNEL_READY || state == GRPC_CHANNEL_IDLE;
}

bool grpc_client_t::ping(int timeout_seconds)
{
// Channel-level READY is unreliable; a real RPC with a deadline is the
// only way to confirm the server is reachable.
if (!impl_->stub) {
last_error_ = "Not connected to server";
return false;
}

if (timeout_seconds <= 0) { timeout_seconds = 5; }

grpc::ClientContext probe_ctx;
probe_ctx.set_deadline(std::chrono::system_clock::now() + std::chrono::seconds(timeout_seconds));
cuopt::remote::StatusRequest probe_req;
probe_req.set_job_id("__connection_probe__");
cuopt::remote::StatusResponse probe_resp;
auto probe_status = impl_->stub->CheckStatus(&probe_ctx, probe_req, &probe_resp);

auto code = probe_status.error_code();
if (code != grpc::StatusCode::OK && code != grpc::StatusCode::NOT_FOUND) {
last_error_ = probe_status.error_message().empty() ? "gRPC server health check failed"
: probe_status.error_message();
GRPC_CLIENT_DEBUG_LOG(config_, "[grpc_client] ping failed: " << last_error_);
return false;
}
return true;
}

void grpc_client_t::start_log_streaming(const std::string& job_id)
{
if (!config_.stream_logs || !config_.log_callback) return;
Expand Down
8 changes: 8 additions & 0 deletions cpp/src/grpc/client/grpc_client.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -276,6 +276,14 @@ class grpc_client_t {
*/
bool is_connected() const;

/**
* @brief Probe the server with a short CheckStatus RPC.
*
* Same check used at connect time: OK or NOT_FOUND means the process
* answered. Other gRPC codes (UNAVAILABLE, DEADLINE_EXCEEDED, ...) fail.
*/
bool ping(int timeout_seconds = 5);

/**
* @brief Solve an LP problem remotely
*
Expand Down
26 changes: 26 additions & 0 deletions cpp/tests/linear_programming/grpc/grpc_client_test.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -369,6 +369,32 @@ class GrpcClientTest : public ::testing::Test {
// CheckStatus Tests
// =============================================================================

TEST_F(GrpcClientTest, Ping_Success_NotFound)
{
EXPECT_CALL(*mock_stub_, CheckStatus(_, _, _))
.WillOnce([](grpc::ClientContext*,
const cuopt::remote::StatusRequest& req,
cuopt::remote::StatusResponse* resp) {
EXPECT_EQ(req.job_id(), "__connection_probe__");
resp->set_job_status(cuopt::remote::NOT_FOUND);
return grpc::Status::OK;
});

EXPECT_TRUE(client_->ping(5));
}

TEST_F(GrpcClientTest, Ping_Failure_Unavailable)
{
EXPECT_CALL(*mock_stub_, CheckStatus(_, _, _))
.WillOnce([](grpc::ClientContext*,
const cuopt::remote::StatusRequest&,
cuopt::remote::StatusResponse*) {
return grpc::Status(grpc::StatusCode::UNAVAILABLE, "down");
});

EXPECT_FALSE(client_->ping(5));
}

TEST_F(GrpcClientTest, CheckStatus_Success_Completed)
{
// Setup mock to return COMPLETED status
Expand Down
1 change: 1 addition & 0 deletions python/cuopt/cuopt/grpc/client/grpc_client.pxd
Original file line number Diff line number Diff line change
Expand Up @@ -196,6 +196,7 @@ cdef extern from "cuopt/grpc/cython_grpc_client.hpp" namespace "cuopt::cython":
const grpc_python_client_connect_options_t& options,
) except +
bint connect(string& error_out) except +
bint ping(string& error_out, int timeout_seconds) except + nogil
string last_error()

# Shared job control
Expand Down
18 changes: 18 additions & 0 deletions python/cuopt/cuopt/grpc/client/grpc_client.pyx
Original file line number Diff line number Diff line change
Expand Up @@ -301,6 +301,24 @@ cdef class Client:
if not self._client.get().connect(error_out):
raise GrpcError(error_out.decode("utf-8"))

def ping(self, timeout_seconds=5):
"""
Probe ``cuopt_grpc_server`` with a short CheckStatus RPC.

Raises :class:`GrpcError` if the server does not answer before
``timeout_seconds`` (default 5). Used by the HTTP proxy health
endpoints so Kubernetes can restart a combined proxy+gRPC container.
"""
cdef string error_out
cdef int timeout = int(timeout_seconds)
cdef bint ok
if timeout <= 0:
timeout = 5
with nogil:
ok = self._client.get().ping(error_out, timeout)
if not ok:
raise GrpcError(error_out.decode("utf-8") or "gRPC ping failed")

def _spawn_client(self):
"""Create a sibling connection with the same host/port/TLS settings."""
return Client(self._host, self._port, tls=self._tls)
Expand Down
40 changes: 39 additions & 1 deletion python/cuopt_server/cuopt_server/proxy_webserver.py
Original file line number Diff line number Diff line change
Expand Up @@ -343,7 +343,44 @@ def _deserialize_convert_submit(
@app.get("/v2/health/ready", responses=HealthResponse)
@app.get("/v2/health/live", responses=HealthResponse)
def health():
return {"status": "RUNNING", "version": app.version}
try:
_require_grpc_healthy()
return {"status": "RUNNING", "version": app.version}
except HTTPException as e:
return http_exception_handler(e)


def _grpc_backend_ok():
"""Return (True, '') if cuopt_grpc_server answered a short probe."""
client = _grpc_client
if client is None:
return False, "gRPC client is not connected"
try:
ping = getattr(client, "ping", None)
if callable(ping):
ping(5)
else:
client.status("__connection_probe__")
return True, ""
except Exception as e:
text = str(e).strip() or type(e).__name__
return False, text


def _require_grpc_healthy():
ok, msg = _grpc_backend_ok()
if not ok:
logging.error("ERROR : %s", msg)
raise HTTPException(
status_code=500,
detail=(
"Status : Broken\n"
"The gRPC backend is unavailable. "
"This process cannot serve solves until "
"cuopt_grpc_server is healthy.\n"
"ERROR : gRPC backend unavailable"
),
)


@app.get(
Expand Down Expand Up @@ -721,6 +758,7 @@ async def postrequest(
warnings = check_client_version(client_version)

try:
await asyncio.to_thread(_require_grpc_healthy)
accept = _resolve_accept(
accept, ctype if ctype != mime_pickle else mime_json
)
Expand Down
42 changes: 42 additions & 0 deletions python/cuopt_server/cuopt_server/tests/test_grpc_http_proxy.py
Original file line number Diff line number Diff line change
Expand Up @@ -133,6 +133,10 @@ def __init__(self):
self._incumbents = {}
self._logs = {}

def ping(self, timeout_seconds=5):
if getattr(self, "unhealthy", False):
raise RuntimeError("gRPC server unavailable")

def submit(self, problem, settings, enable_incumbents=None):
job_id = str(uuid.uuid4())
self.jobs[job_id] = FakeJobStatus.COMPLETED
Expand Down Expand Up @@ -176,6 +180,9 @@ def incumbents(self, job_id, from_index=0):

@pytest.fixture(scope="module")
def proxy_server():
reset_proxy_state()
set_grpc_client(FakeClient())
set_max_request_size(1024 * 1024 * 1024)
port = _free_port()
server = _Uvicorn(
uvicorn.Config(
Expand Down Expand Up @@ -204,6 +211,7 @@ def proxy_server():
yield url
server.should_exit = True
thread.join(timeout=5)
reset_proxy_state()


@pytest.fixture
Expand Down Expand Up @@ -303,6 +311,40 @@ def test_health(proxy):
assert "version" in body


def test_health_fails_when_grpc_is_down(proxy):
url, fake = proxy
fake.unhealthy = True
res = requests.get(url + "/cuopt/health")
assert res.status_code == 500
body = res.json()
assert "Broken" in body["error"] or "Broken" in str(body)
assert "gRPC" in body["error"] or "unavailable" in body["error"].lower()
assert "gRPC server unavailable" not in body["error"]


def test_health_fails_without_grpc_client(proxy_server):
url = proxy_server
reset_proxy_state()
try:
res = requests.get(url + "/cuopt/health")
assert res.status_code == 500
assert "gRPC" in res.json()["error"]
finally:
reset_proxy_state()


def test_submit_rejected_when_grpc_unhealthy(proxy):
url, fake = proxy
fake.unhealthy = True
res = requests.post(
url + "/cuopt/request",
headers={"CLIENT-VERSION": "custom"},
json=_lp(),
)
assert res.status_code == 500
assert fake.submitted == []


def test_submit_status_result_delete(proxy):
url, fake = proxy
lp = _lp()
Expand Down
Loading