diff --git a/tests/api_service/CMakeLists.txt b/tests/api_service/CMakeLists.txt index d5bd61b3ef..b785b0767f 100644 --- a/tests/api_service/CMakeLists.txt +++ b/tests/api_service/CMakeLists.txt @@ -48,6 +48,17 @@ cc_test( "${_api_json_test_env}" ) +cc_test( + NAME + api_error_test + SRCS + api_error_test.cpp + DEPS + api_service + GTest::gtest_main + nlohmann_json::nlohmann_json +) + cc_test( NAME completion_json_parser_test diff --git a/tests/api_service/api_error_test.cpp b/tests/api_service/api_error_test.cpp new file mode 100644 index 0000000000..b0d5017a02 --- /dev/null +++ b/tests/api_service/api_error_test.cpp @@ -0,0 +1,236 @@ +/* Copyright 2025-2026 The xLLM Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + https://github.com/jd-opensource/xllm/blob/main/LICENSE + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +==============================================================================*/ + +#include "api_service/api_error.h" + +#include +#include + +#include +#include +#include + +#include "anthropic.pb.h" +#include "api_service/non_stream_call.h" +#include "api_service/stream_call.h" +#include "api_service/xllm_metrics.h" +#include "chat.pb.h" +#include "completion.pb.h" +#include "core/util/closure_guard.h" + +namespace xllm { +namespace { + +class NoopClosure final : public google::protobuf::Closure { + public: + void Run() override { ++run_count_; } + + int run_count() const { return run_count_; } + + private: + int run_count_ = 0; +}; + +TEST(ApiErrorTest, OpenAIErrorReplacesInvalidUtf8) { + const std::string invalid_message("invalid-\xff", 9); + + std::string body; + EXPECT_NO_THROW(body = api_service::make_openai_error_json( + StatusCode::INVALID_ARGUMENT, invalid_message)); + + const auto parsed = nlohmann::json::parse(body); + EXPECT_EQ(parsed["error"]["message"], "invalid-\xef\xbf\xbd"); + EXPECT_EQ(parsed["error"]["type"], "invalid_request_error"); +} + +TEST(ApiErrorTest, AnthropicErrorReplacesInvalidUtf8) { + const std::string invalid_message("invalid-\xff", 9); + + std::string body; + EXPECT_NO_THROW(body = api_service::make_anthropic_error_json( + StatusCode::INVALID_ARGUMENT, invalid_message)); + + const auto parsed = nlohmann::json::parse(body); + EXPECT_EQ(parsed["type"], "error"); + EXPECT_EQ(parsed["error"]["message"], "invalid-\xef\xbf\xbd"); +} + +TEST(XllmMetricsTest, UsesHttpStatusForStructuredErrors) { + brpc::Controller success; + success.http_response().set_status_code(200); + EXPECT_FALSE(is_failed_request(&success)); + + brpc::Controller client_error; + client_error.http_response().set_status_code(400); + EXPECT_TRUE(is_failed_request(&client_error)); + + brpc::Controller server_error; + server_error.http_response().set_status_code(500); + EXPECT_TRUE(is_failed_request(&server_error)); +} + +TEST(CallTest, RequestIdHeaderTakesPrecedenceOverBody) { + brpc::Controller controller; + controller.http_request().SetHeader("x-request-id", "header-id"); + proto::CompletionRequest request; + request.set_x_request_id("body-id"); + proto::CompletionResponse response; + NoopClosure done; + + NonStreamCall call( + &controller, &done, &request, &response, true); + + EXPECT_EQ(call.x_request_id(), "header-id"); + ASSERT_NE(controller.http_response().GetHeader("x-request-id"), nullptr); + EXPECT_EQ(*controller.http_response().GetHeader("x-request-id"), "header-id"); +} + +TEST(CallTest, RequestIdFallsBackToBody) { + brpc::Controller controller; + proto::CompletionRequest request; + request.set_x_request_id("body-id"); + proto::CompletionResponse response; + NoopClosure done; + + NonStreamCall call( + &controller, &done, &request, &response, true); + + EXPECT_EQ(call.x_request_id(), "body-id"); + ASSERT_NE(controller.http_response().GetHeader("x-request-id"), nullptr); + EXPECT_EQ(*controller.http_response().GetHeader("x-request-id"), "body-id"); +} + +TEST(CallTest, RequestIdIsGeneratedWhenHeaderAndBodyAreMissing) { + brpc::Controller controller; + proto::CompletionRequest request; + proto::CompletionResponse response; + NoopClosure done; + + NonStreamCall call( + &controller, &done, &request, &response, true); + + EXPECT_FALSE(call.x_request_id().empty()); + EXPECT_EQ(call.x_request_id().find("req-"), 0); + ASSERT_NE(controller.http_response().GetHeader("x-request-id"), nullptr); + EXPECT_EQ(*controller.http_response().GetHeader("x-request-id"), + call.x_request_id()); +} + +TEST(CallTest, ExtractsRequestIdFromAllSupportedBodyTypes) { + proto::CompletionRequest completion; + completion.set_x_request_id("completion-id"); + EXPECT_EQ(request_body_x_request_id(&completion), "completion-id"); + + proto::ChatRequest chat; + chat.set_x_request_id("chat-id"); + EXPECT_EQ(request_body_x_request_id(&chat), "chat-id"); + + proto::AnthropicMessagesRequest anthropic; + anthropic.set_x_request_id("anthropic-id"); + EXPECT_EQ(request_body_x_request_id(&anthropic), "anthropic-id"); +} + +TEST(CallTest, TypedNonStreamErrorUsesBrpcFailure) { + brpc::Controller controller; + proto::CompletionRequest request; + proto::CompletionResponse response; + NoopClosure done; + + { + NonStreamCall call( + &controller, &done, &request, &response, true); + EXPECT_TRUE(call.finish_with_error(StatusCode::INVALID_ARGUMENT, + "typed request failed")); + EXPECT_TRUE(controller.Failed()); + EXPECT_EQ(controller.ErrorText(), "typed request failed"); + } + + EXPECT_EQ(done.run_count(), 1); +} + +TEST(CallTest, TypedStreamCallErrorUsesBrpcFailure) { + brpc::Controller controller; + proto::CompletionRequest request; + request.set_stream(false); + proto::CompletionResponse response; + NoopClosure done; + + { + StreamCall call( + &controller, &done, &request, &response, true); + EXPECT_TRUE(call.finish_with_error(StatusCode::UNAVAILABLE, + "typed stream call failed")); + EXPECT_TRUE(controller.Failed()); + EXPECT_EQ(controller.ErrorText(), "typed stream call failed"); + } + + EXPECT_EQ(done.run_count(), 1); +} + +TEST(CallTest, HttpErrorKeepsStructuredOpenAIResponse) { + brpc::Controller controller; + proto::CompletionRequest request; + proto::CompletionResponse response; + NoopClosure done; + + { + NonStreamCall call( + &controller, &done, &request, &response, true, true); + EXPECT_TRUE(call.finish_with_error(StatusCode::INVALID_ARGUMENT, + "invalid request")); + EXPECT_FALSE(controller.Failed()); + EXPECT_EQ(controller.http_response().status_code(), 400); + EXPECT_EQ(controller.response_attachment().to_string(), + api_service::make_openai_error_json(StatusCode::INVALID_ARGUMENT, + "invalid request")); + } + + EXPECT_EQ(done.run_count(), 1); +} + +TEST(CallTest, AsyncCompletionRunsAfterFinalErrorState) { + brpc::Controller controller; + proto::CompletionRequest request; + proto::CompletionResponse response; + NoopClosure done; + int completion_count = 0; + bool observed_failure = false; + ClosureGuard done_guard( + &done, + [](void* /*unused*/) {}, + [&](void* completion_context) { + ++completion_count; + ASSERT_NE(completion_context, nullptr); + observed_failure = + static_cast(completion_context) + ->failed; + }); + + auto async_closure = done_guard.release_for_async(); + EXPECT_EQ(completion_count, 0); + { + NonStreamCall call( + &controller, std::move(async_closure), &request, &response, true, true); + call.finish_with_error(StatusCode::UNKNOWN, "async request failed"); + EXPECT_EQ(completion_count, 0); + } + + EXPECT_EQ(completion_count, 1); + EXPECT_TRUE(observed_failure); + EXPECT_EQ(done.run_count(), 1); +} + +} // namespace +} // namespace xllm diff --git a/tests/core/framework/request/CMakeLists.txt b/tests/core/framework/request/CMakeLists.txt index a1129f8e42..5568957931 100644 --- a/tests/core/framework/request/CMakeLists.txt +++ b/tests/core/framework/request/CMakeLists.txt @@ -38,6 +38,8 @@ cc_test( SRCS stopping_checker_test.cpp DEPS + :block + :prefix_cache :request GTest::gtest_main ) diff --git a/tests/core/util/CMakeLists.txt b/tests/core/util/CMakeLists.txt index 49cba0f30f..ceeea40d84 100644 --- a/tests/core/util/CMakeLists.txt +++ b/tests/core/util/CMakeLists.txt @@ -9,6 +9,7 @@ cc_test( http_downloader_test.cpp suffix_decoding_cache_test.cpp threadpool_test.cpp + verbose_trace_logger_test.cpp vocab_validation_test.cpp DEPS util diff --git a/tests/core/util/verbose_trace_logger_test.cpp b/tests/core/util/verbose_trace_logger_test.cpp new file mode 100644 index 0000000000..9c17e019a3 --- /dev/null +++ b/tests/core/util/verbose_trace_logger_test.cpp @@ -0,0 +1,53 @@ +/* Copyright 2026 The xLLM Authors. All Rights Reserved. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + https://github.com/jd-opensource/xllm/blob/main/LICENSE + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +==============================================================================*/ + +#include "core/util/verbose_trace_logger.h" + +#include + +namespace xllm { +namespace { + +TEST(VerboseTraceLogPathTest, SingleRankPreservesConfiguredPath) { + EXPECT_EQ(resolve_verbose_trace_log_path( + "log/verbose_trace.log", /*nnodes=*/1, /*node_rank=*/0), + "log/verbose_trace.log"); +} + +TEST(VerboseTraceLogPathTest, MultiRankInsertsRankBeforeLastExtension) { + EXPECT_EQ(resolve_verbose_trace_log_path( + "log.v1/verbose.trace.log", /*nnodes=*/4, /*node_rank=*/2), + "log.v1/verbose.trace_rank_2.log"); +} + +TEST(VerboseTraceLogPathTest, MultiRankSuffixesRankZero) { + EXPECT_EQ(resolve_verbose_trace_log_path( + "log/verbose_trace.log", /*nnodes=*/4, /*node_rank=*/0), + "log/verbose_trace_rank_0.log"); +} + +TEST(VerboseTraceLogPathTest, MultiRankAppendsRankWithoutExtension) { + EXPECT_EQ(resolve_verbose_trace_log_path( + "log/verbose_trace", /*nnodes=*/4, /*node_rank=*/3), + "log/verbose_trace_rank_3"); +} + +TEST(VerboseTraceLogPathTest, EmptyPathRemainsEmpty) { + EXPECT_TRUE(resolve_verbose_trace_log_path("", /*nnodes=*/4, /*node_rank=*/1) + .empty()); +} + +} // namespace +} // namespace xllm diff --git a/xllm/CMakeLists.txt b/xllm/CMakeLists.txt index 2451510e8c..a27599b4dc 100644 --- a/xllm/CMakeLists.txt +++ b/xllm/CMakeLists.txt @@ -83,6 +83,7 @@ else() :config :xllm_server :master + :util absl::strings Boost::serialization gflags::gflags diff --git a/xllm/api_service/CMakeLists.txt b/xllm/api_service/CMakeLists.txt index 63691c2c56..60c00d91e3 100644 --- a/xllm/api_service/CMakeLists.txt +++ b/xllm/api_service/CMakeLists.txt @@ -7,6 +7,7 @@ cc_library( api_service.h api_service_impl.h call.h + api_error.h chat_json_parser.h completion_json_parser.h completion_service_impl.h @@ -37,6 +38,7 @@ cc_library( completion_json_parser.cpp service_impl_factory.cpp call.cpp + api_error.cpp completion_service_impl.cpp rec_completion_service_impl.cpp chat_service_impl.cpp diff --git a/xllm/api_service/anthropic_service_impl.cpp b/xllm/api_service/anthropic_service_impl.cpp index 6474d2eb26..89d6c8f740 100644 --- a/xllm/api_service/anthropic_service_impl.cpp +++ b/xllm/api_service/anthropic_service_impl.cpp @@ -658,7 +658,7 @@ void AnthropicServiceImpl::process_async_impl( // Build request parameters RequestParams request_params( - rpc_request, call->get_x_request_id(), call->get_x_request_time()); + rpc_request, call->x_request_id(), call->x_request_time()); // Build messages std::vector messages = build_messages(rpc_request); diff --git a/xllm/api_service/api_error.cpp b/xllm/api_service/api_error.cpp new file mode 100644 index 0000000000..252a40b504 --- /dev/null +++ b/xllm/api_service/api_error.cpp @@ -0,0 +1,250 @@ +/* Copyright 2025-2026 The xLLM Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + https://github.com/jd-opensource/xllm/blob/main/LICENSE + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +==============================================================================*/ + +#include "api_service/api_error.h" + +#include + +#include +#include +#include + +#include "core/common/instance_name.h" +#include "core/util/uuid.h" +#include "core/util/verbose_trace_logger.h" + +namespace xllm { +namespace api_service { + +namespace { + +// HTTP status codes used for error responses. Kept local to avoid pulling in +// brpc's http_status_code.h across the api_service headers. +constexpr int32_t kHttpBadRequest = 400; +constexpr int32_t kHttpClientClosedRequest = 499; +constexpr int32_t kHttpInternalServerError = 500; +constexpr int32_t kHttpServiceUnavailable = 503; +constexpr int32_t kHttpGatewayTimeout = 504; +constexpr int32_t kHttpTooManyRequests = 429; + +} // namespace + +int32_t status_code_to_http_status(StatusCode code) { + switch (code) { + case StatusCode::OK: + return 200; + case StatusCode::CANCELLED: + return kHttpClientClosedRequest; + case StatusCode::INVALID_ARGUMENT: + return kHttpBadRequest; + case StatusCode::DEADLINE_EXCEEDED: + return kHttpGatewayTimeout; + case StatusCode::RESOURCE_EXHAUSTED: + return kHttpTooManyRequests; + case StatusCode::UNAVAILABLE: + return kHttpServiceUnavailable; + case StatusCode::UNKNOWN: + default: + return kHttpInternalServerError; + } +} + +const char* status_code_to_openai_error_type(StatusCode code) { + switch (code) { + case StatusCode::INVALID_ARGUMENT: + return "invalid_request_error"; + case StatusCode::RESOURCE_EXHAUSTED: + return "rate_limit_error"; + case StatusCode::UNAVAILABLE: + return "service_unavailable_error"; + case StatusCode::DEADLINE_EXCEEDED: + return "timeout_error"; + case StatusCode::CANCELLED: + return "request_cancelled"; + case StatusCode::OK: + case StatusCode::UNKNOWN: + default: + return "internal_error"; + } +} + +const char* status_code_to_anthropic_error_type(StatusCode code) { + switch (code) { + case StatusCode::INVALID_ARGUMENT: + return "invalid_request_error"; + case StatusCode::RESOURCE_EXHAUSTED: + return "rate_limit_error"; + case StatusCode::UNAVAILABLE: + return "overloaded_error"; + case StatusCode::DEADLINE_EXCEEDED: + return "timeout_error"; + case StatusCode::CANCELLED: + return "request_cancelled"; + case StatusCode::OK: + case StatusCode::UNKNOWN: + default: + return "api_error"; + } +} + +const char* status_code_to_string(StatusCode code) { + switch (code) { + case StatusCode::OK: + return "ok"; + case StatusCode::CANCELLED: + return "cancelled"; + case StatusCode::INVALID_ARGUMENT: + return "invalid_argument"; + case StatusCode::DEADLINE_EXCEEDED: + return "deadline_exceeded"; + case StatusCode::RESOURCE_EXHAUSTED: + return "resource_exhausted"; + case StatusCode::UNAVAILABLE: + return "unavailable"; + case StatusCode::UNKNOWN: + default: + return "unknown"; + } +} + +std::string make_openai_error_json(StatusCode code, std::string_view message) { + nlohmann::json body; + auto& error = body["error"]; + error["message"] = message; + error["type"] = status_code_to_openai_error_type(code); + error["code"] = status_code_to_string(code); + error["param"] = nullptr; + return body.dump(-1, ' ', false, nlohmann::json::error_handler_t::replace); +} + +std::string make_anthropic_error_json(StatusCode code, + std::string_view message) { + nlohmann::json body; + body["type"] = "error"; + auto& error = body["error"]; + error["type"] = status_code_to_anthropic_error_type(code); + error["message"] = message; + return body.dump(-1, ' ', false, nlohmann::json::error_handler_t::replace); +} + +std::string get_header_x_request_id(const brpc::Controller* controller) { + if (controller == nullptr) { + return ""; + } + const std::string* request_id = + controller->http_request().GetHeader("x-request-id"); + if (request_id != nullptr) { + return *request_id; + } + const std::string* ms_request_id = + controller->http_request().GetHeader("x-ms-client-request-id"); + if (ms_request_id != nullptr) { + return *ms_request_id; + } + return ""; +} + +std::string generate_x_request_id() { + thread_local ShortUUID short_uuid; + return "req-" + InstanceName::name()->get_name_hash() + "-" + + short_uuid.random(); +} + +std::string ensure_x_request_id(const brpc::Controller* controller) { + std::string x_request_id = get_header_x_request_id(controller); + if (x_request_id.empty()) { + x_request_id = generate_x_request_id(); + } + return x_request_id; +} + +void log_request_error(StatusCode code, + std::string_view message, + std::string_view x_request_id, + bool write_verbose_trace) { + std::string_view id = x_request_id.empty() ? "-" : x_request_id; + LOG(ERROR) << "[x-request-id=" << id + << "] request failed (http=" << status_code_to_http_status(code) + << ", status=" << status_code_to_string(code) << "): " << message; + if (!write_verbose_trace) { + return; + } + // Mirror the failure to the asynchronous verbose trace log (no-op when the + // feature is disabled). This keeps the detailed off-hot-path record in sync + // with every error response we emit, regardless of the entry point. + XLLM_VERBOSE_TRACE() << "event=request_error x-request-id=" << id + << " http=" << status_code_to_http_status(code) + << " status=" << status_code_to_string(code) + << " message=" << message; +} + +namespace { + +void write_error_body(brpc::Controller* controller, + StatusCode code, + std::string_view message, + std::string_view x_request_id, + bool write_verbose_trace, + const std::string& body) { + const int32_t http_status = status_code_to_http_status(code); + log_request_error(code, message, x_request_id, write_verbose_trace); + + controller->http_response().set_content_type("application/json"); + controller->http_response().set_status_code(http_status); + if (!x_request_id.empty()) { + controller->http_response().SetHeader("x-request-id", + std::string(x_request_id)); + } + // Drop any partially serialized payload before emitting the error body. + controller->response_attachment().clear(); + controller->response_attachment().append(body); +} + +} // namespace + +void write_openai_error(brpc::Controller* controller, + StatusCode code, + std::string_view message, + std::string_view x_request_id, + bool write_verbose_trace) { + if (controller == nullptr) { + return; + } + write_error_body(controller, + code, + message, + x_request_id, + write_verbose_trace, + make_openai_error_json(code, message)); +} + +void write_anthropic_error(brpc::Controller* controller, + StatusCode code, + std::string_view message, + std::string_view x_request_id, + bool write_verbose_trace) { + if (controller == nullptr) { + return; + } + write_error_body(controller, + code, + message, + x_request_id, + write_verbose_trace, + make_anthropic_error_json(code, message)); +} + +} // namespace api_service +} // namespace xllm diff --git a/xllm/api_service/api_error.h b/xllm/api_service/api_error.h new file mode 100644 index 0000000000..9ced929ef3 --- /dev/null +++ b/xllm/api_service/api_error.h @@ -0,0 +1,92 @@ +/* Copyright 2025-2026 The xLLM Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + https://github.com/jd-opensource/xllm/blob/main/LICENSE + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +==============================================================================*/ + +#pragma once + +#include + +#include +#include + +#include "core/common/types.h" + +namespace xllm { +namespace api_service { + +// Map an internal StatusCode to the corresponding HTTP status code. +int32_t status_code_to_http_status(StatusCode code); + +// Map an internal StatusCode to an OpenAI-style error "type" string +// (e.g. "invalid_request_error", "rate_limit_error"). +const char* status_code_to_openai_error_type(StatusCode code); + +// Map an internal StatusCode to an Anthropic-style error "type" string. +const char* status_code_to_anthropic_error_type(StatusCode code); + +// Stable machine-readable short code for a StatusCode (e.g. +// "invalid_argument"). +const char* status_code_to_string(StatusCode code); + +// Build an OpenAI-compatible error body: +// {"error":{"message":...,"type":...,"code":...,"param":null}} +std::string make_openai_error_json(StatusCode code, std::string_view message); + +// Build an Anthropic-compatible error body: +// {"type":"error","error":{"type":...,"message":...}} +std::string make_anthropic_error_json(StatusCode code, + std::string_view message); + +// Extract the client-supplied x-request-id from an HTTP controller, looking up +// "x-request-id" then "x-ms-client-request-id". Returns an empty string when +// neither header is present. +std::string get_header_x_request_id(const brpc::Controller* controller); + +// Generate a server-side request-scoped x-request-id, formatted like the +// engine's internal request ids ("req--") so it is +// globally unique and visually consistent with them. +std::string generate_x_request_id(); + +// Resolve the x-request-id for a request: the client-supplied value when +// present, otherwise a freshly generated one. Guarantees a non-empty result so +// early failures (before a Call exists) can still be correlated. +std::string ensure_x_request_id(const brpc::Controller* controller); + +// Log a request failure together with its x-request-id, without touching the +// response. Used by streaming paths that have already committed a 200 header +// and can no longer change the status code or body envelope. +void log_request_error(StatusCode code, + std::string_view message, + std::string_view x_request_id, + bool write_verbose_trace = true); + +// Finish a non-stream HTTP request with a structured OpenAI-style error: sets +// the HTTP status code, writes the JSON error body, and logs the failure with +// the x-request-id. Does NOT call SetFailed, which would discard the custom +// body and force a generic 500 status. +void write_openai_error(brpc::Controller* controller, + StatusCode code, + std::string_view message, + std::string_view x_request_id, + bool write_verbose_trace = true); + +// Same as write_openai_error but emits the Anthropic error envelope. +void write_anthropic_error(brpc::Controller* controller, + StatusCode code, + std::string_view message, + std::string_view x_request_id, + bool write_verbose_trace = true); + +} // namespace api_service +} // namespace xllm diff --git a/xllm/api_service/api_service.cpp b/xllm/api_service/api_service.cpp index 77630fa267..8676043993 100644 --- a/xllm/api_service/api_service.cpp +++ b/xllm/api_service/api_service.cpp @@ -22,6 +22,7 @@ limitations under the License. #include +#include "api_service/api_error.h" #include "api_service/chat_json_parser.h" #include "api_service/completion_json_parser.h" #include "api_service/service_impl_factory.h" @@ -40,6 +41,7 @@ limitations under the License. #include "core/framework/config/distributed_config.h" #include "core/framework/config/profile_config.h" #include "core/util/closure_guard.h" +#include "core/util/verbose_trace_logger.h" #include "embedding.pb.h" #include "image_generation.pb.h" #include "models.pb.h" @@ -62,6 +64,76 @@ google::protobuf::Arena* GetArenaWithCheck( const char* kSampleNotSupportedError = "/v1/sample is only supported for LLM"; +void complete_request_metric(::google::protobuf::RpcController* controller, + void* completion_context) { + if (completion_context == nullptr) { + request_out_metric(static_cast(controller)); + return; + } + const auto* completion_status = + static_cast(completion_context); + request_out_metric(completion_status->failed); +} + +// Upper bound on the raw request body captured into the verbose trace log, so a +// pathologically large payload cannot blow up the off-hot-path record. +constexpr size_t kVerboseBodyLimit = 8192; + +// Emit the failing request's full context (x-request-id, endpoint, raw body) to +// the asynchronous verbose trace log. Guarded by enabled() so the body string +// is only materialized when the feature is on. +void verbose_log_request_failure(brpc::Controller* ctrl, + StatusCode code, + const std::string& message, + const std::string& x_request_id) { + if (!VerboseTraceLogger::get_instance().enabled()) { + return; + } + std::string body = ctrl->request_attachment().to_string(); + if (body.size() > kVerboseBodyLimit) { + body.resize(kVerboseBodyLimit); + body.append("...(truncated)"); + } + // request_body is arbitrary, fully untrusted user input (it may contain + // spaces, braces, quotes, or newlines), so it is emitted last and on its own + // line: everything from the trailing "request_body=" newline up to the next + // timestamped entry is the verbatim body. This keeps it copy-pasteable with + // no escaping and no delimiter the body itself could collide with; the + // millisecond timestamp prefix on the following entry marks where it ends. + XLLM_VERBOSE_TRACE() << "event=request_failed_context x-request-id=" + << x_request_id << " http=" + << api_service::status_code_to_http_status(code) + << " path=" << ctrl->http_request().uri().path() + << " status=" << api_service::status_code_to_string(code) + << " reason=" << message << " request_body=\n" + << body; +} + +// Finish an HTTP request with a structured OpenAI-style error body plus an +// x-request-id annotated log line. Centralizes the early-failure path shared by +// the HTTP entrypoints (JSON parse failure, missing model, disabled feature, +// ...). These early failures happen before a Call exists, so the x-request-id +// is resolved here (client header or freshly generated) for the log line. +void fail_http_request(brpc::Controller* ctrl, + StatusCode code, + const std::string& message) { + const std::string x_request_id = api_service::ensure_x_request_id(ctrl); + verbose_log_request_failure(ctrl, code, message, x_request_id); + api_service::write_openai_error( + ctrl, code, message, x_request_id, false /*write_verbose_trace*/); +} + +// Anthropic HTTP entrypoints surface failures using the Anthropic error +// envelope instead of the OpenAI one. +void fail_anthropic_request(brpc::Controller* ctrl, + StatusCode code, + const std::string& message) { + const std::string x_request_id = api_service::ensure_x_request_id(ctrl); + verbose_log_request_failure(ctrl, code, message, x_request_id); + api_service::write_anthropic_error( + ctrl, code, message, x_request_id, false /*write_verbose_trace*/); +} + // Shared dispatch for brpc-typed (non-Http) APIs that follow the standard // service_impl -> process_async pattern. Performs argument validation, // instantiates the appropriate Call wrapper from the typed proto request and @@ -80,8 +152,8 @@ void process_typed_brpc_request(std::unique_ptr& service_impl, xllm::ClosureGuard done_guard( done, [](void* /*unused*/) { request_in_metric(nullptr); }, - [controller](void* /*unused*/) { - request_out_metric(static_cast(controller)); + [controller](void* completion_context) { + complete_request_metric(controller, completion_context); }); if (!request || !response || !controller) { LOG(ERROR) << "brpc request | respose | controller is null"; @@ -102,7 +174,7 @@ void process_typed_brpc_request(std::unique_ptr& service_impl, // from it. We cast away constness so the Call can hold a non-const pointer. auto req_pb = const_cast(request); std::shared_ptr call = std::make_shared( - ctrl, done_guard.release(), req_pb, response, arena != nullptr); + ctrl, done_guard.release_for_async(), req_pb, response, arena != nullptr); service_impl->process_async(call); } @@ -152,8 +224,8 @@ void APIService::Completions(::google::protobuf::RpcController* controller, xllm::ClosureGuard done_guard( done, [](void* /*unused*/) { request_in_metric(nullptr); }, - [controller](void* /*unused*/) { - request_out_metric(static_cast(controller)); + [controller](void* completion_context) { + complete_request_metric(controller, completion_context); }); if (!request || !response || !controller) { LOG(ERROR) << "brpc request | respose | controller is null."; @@ -167,7 +239,7 @@ void APIService::Completions(::google::protobuf::RpcController* controller, auto arena = GetArenaWithCheck(response); std::shared_ptr call = std::make_shared( ctrl, - done_guard.release(), + done_guard.release_for_async(), const_cast(request), response, arena != nullptr); @@ -182,8 +254,8 @@ void APIService::CompletionsHttp(::google::protobuf::RpcController* controller, xllm::ClosureGuard done_guard( done, [](void* /*unused*/) { request_in_metric(nullptr); }, - [controller](void* /*unused*/) { - request_out_metric(static_cast(controller)); + [controller](void* completion_context) { + complete_request_metric(controller, completion_context); }); if (!request || !response || !controller) { LOG(ERROR) << "brpc request | respose | controller is null"; @@ -201,9 +273,8 @@ void APIService::CompletionsHttp(::google::protobuf::RpcController* controller, auto [preprocess_status, processed_json] = preprocess_completion_prompt(ctrl->request_attachment().to_string()); if (!preprocess_status.ok()) { - ctrl->SetFailed(preprocess_status.message()); - LOG(ERROR) << "completion prompt preprocessing failed: " - << preprocess_status.message(); + fail_http_request( + ctrl, preprocess_status.code(), preprocess_status.message()); return; } @@ -212,13 +283,17 @@ void APIService::CompletionsHttp(::google::protobuf::RpcController* controller, auto st = json2pb::JsonToProtoMessage(processed_json, req_pb, options, &error); if (!st) { - ctrl->SetFailed(error); - LOG(ERROR) << "parse json to proto failed: " << error; + fail_http_request(ctrl, StatusCode::INVALID_ARGUMENT, error); return; } - std::shared_ptr call = std::make_shared( - ctrl, done_guard.release(), req_pb, resp_pb, arena != nullptr); + std::shared_ptr call = + std::make_shared(ctrl, + done_guard.release_for_async(), + req_pb, + resp_pb, + arena != nullptr, + true); if (completion_service_impl_) { completion_service_impl_->process_async(call); } else if (rec_completion_service_impl_) { @@ -233,8 +308,8 @@ void APIService::Sample(::google::protobuf::RpcController* controller, xllm::ClosureGuard done_guard( done, [](void* /*unused*/) { request_in_metric(nullptr); }, - [controller](void* /*unused*/) { - request_out_metric(static_cast(controller)); + [controller](void* completion_context) { + complete_request_metric(controller, completion_context); }); if (!request || !response || !controller) { LOG(ERROR) << "brpc request | respose | controller is null."; @@ -261,8 +336,8 @@ void APIService::SampleHttp(::google::protobuf::RpcController* controller, xllm::ClosureGuard done_guard( done, [](void* /*unused*/) { request_in_metric(nullptr); }, - [controller](void* /*unused*/) { - request_out_metric(static_cast(controller)); + [controller](void* completion_context) { + complete_request_metric(controller, completion_context); }); if (!request || !response || !controller) { LOG(ERROR) << "brpc request | respose | controller is null"; @@ -271,7 +346,8 @@ void APIService::SampleHttp(::google::protobuf::RpcController* controller, auto ctrl = reinterpret_cast(controller); if (!sample_service_impl_) { - ctrl->SetFailed(kSampleNotSupportedError); + fail_http_request( + ctrl, StatusCode::INVALID_ARGUMENT, kSampleNotSupportedError); return; } @@ -287,13 +363,17 @@ void APIService::SampleHttp(::google::protobuf::RpcController* controller, butil::IOBufAsZeroCopyInputStream iobuf_stream(buf); auto st = json2pb::JsonToProtoMessage(&iobuf_stream, req_pb, options, &error); if (!st) { - ctrl->SetFailed(error); - LOG(ERROR) << "parse json to proto failed: " << error; + fail_http_request(ctrl, StatusCode::INVALID_ARGUMENT, error); return; } - std::shared_ptr call = std::make_shared( - ctrl, done_guard.release(), req_pb, resp_pb, arena != nullptr); + std::shared_ptr call = + std::make_shared(ctrl, + done_guard.release_for_async(), + req_pb, + resp_pb, + arena != nullptr, + true); sample_service_impl_->process_async(call); } @@ -334,7 +414,9 @@ void chat_completions_http_impl(std::unique_ptr& service, auto content_len = get_json_content_length(ctrl); if (content_len == (size_t)-1L) { - ctrl->SetFailed("Content-Length header is missing."); + fail_http_request(ctrl, + StatusCode::INVALID_ARGUMENT, + "Content-Length header is missing."); return; } @@ -344,9 +426,8 @@ void chat_completions_http_impl(std::unique_ptr& service, auto [preprocess_status, processed_json] = chat_json_parser.preprocess(std::move(attachment)); if (!preprocess_status.ok()) { - ctrl->SetFailed(preprocess_status.message()); - LOG(ERROR) << "Complex message preprocessing failed: " - << preprocess_status.message(); + fail_http_request( + ctrl, preprocess_status.code(), preprocess_status.message()); return; } @@ -355,13 +436,16 @@ void chat_completions_http_impl(std::unique_ptr& service, auto status = google::protobuf::util::JsonStringToMessage( processed_json, req_pb, options); if (!status.ok()) { - ctrl->SetFailed(status.ToString()); - LOG(ERROR) << "parse json to proto failed: " << status.ToString(); + fail_http_request(ctrl, StatusCode::INVALID_ARGUMENT, status.ToString()); return; } - auto call = std::make_shared( - ctrl, guard.release(), req_pb, resp_pb, arena != nullptr /*use_arena*/); + auto call = std::make_shared(ctrl, + guard.release_for_async(), + req_pb, + resp_pb, + arena != nullptr /*use_arena*/, + true /*is_http_request*/); service->process_async(call); } @@ -405,8 +489,8 @@ void APIService::ChatCompletions(::google::protobuf::RpcController* controller, xllm::ClosureGuard done_guard( done, [](void* /*unused*/) { request_in_metric(nullptr); }, - [controller](void* /*unused*/) { - request_out_metric(static_cast(controller)); + [controller](void* completion_context) { + complete_request_metric(controller, completion_context); }); if (!request || !response || !controller) { LOG(ERROR) << "brpc request | respose | controller is null"; @@ -427,8 +511,8 @@ void APIService::ChatCompletionsHttp( xllm::ClosureGuard done_guard( done, [](void* /*unused*/) { request_in_metric(nullptr); }, - [controller](void* /*unused*/) { - request_out_metric(static_cast(controller)); + [controller](void* completion_context) { + complete_request_metric(controller, completion_context); }); if (!request || !response || !controller) { LOG(ERROR) << "brpc request | respose | controller is null"; @@ -451,8 +535,8 @@ void APIService::Embeddings(::google::protobuf::RpcController* controller, xllm::ClosureGuard done_guard( done, [](void* /*unused*/) { request_in_metric(nullptr); }, - [controller](void* /*unused*/) { - request_out_metric(static_cast(controller)); + [controller](void* completion_context) { + complete_request_metric(controller, completion_context); }); if (!request || !response || !controller) { LOG(ERROR) << "brpc request | respose | controller is null"; @@ -478,7 +562,7 @@ void APIService::Embeddings(::google::protobuf::RpcController* controller, } std::shared_ptr call = std::make_shared( - ctrl, done_guard.release(), req_pb, response, arena != nullptr); + ctrl, done_guard.release_for_async(), req_pb, response, arena != nullptr); embedding_service_impl_->process_async(call); } @@ -492,8 +576,8 @@ void handle_embedding_request(std::unique_ptr& embedding_service_impl_, xllm::ClosureGuard done_guard( done, [](void* /*unused*/) { request_in_metric(nullptr); }, - [controller](void* /*unused*/) { - request_out_metric(static_cast(controller)); + [controller](void* completion_context) { + complete_request_metric(controller, completion_context); }); if (!request || !response || !controller) { LOG(ERROR) << "brpc request | respose | controller is null"; @@ -514,8 +598,7 @@ void handle_embedding_request(std::unique_ptr& embedding_service_impl_, butil::IOBufAsZeroCopyInputStream iobuf_stream(buf); auto st = json2pb::JsonToProtoMessage(&iobuf_stream, req_pb, options, &error); if (!st) { - ctrl->SetFailed(error); - LOG(ERROR) << "parse json to proto failed: " << error; + fail_http_request(ctrl, StatusCode::INVALID_ARGUMENT, error); return; } @@ -524,8 +607,13 @@ void handle_embedding_request(std::unique_ptr& embedding_service_impl_, req_pb->set_encoding_format("float"); } - std::shared_ptr call = std::make_shared( - ctrl, done_guard.release(), req_pb, resp_pb, arena != nullptr); + std::shared_ptr call = + std::make_shared(ctrl, + done_guard.release_for_async(), + req_pb, + resp_pb, + arena != nullptr, + true); embedding_service_impl_->process_async(call); } } // namespace @@ -564,8 +652,8 @@ void APIService::ImageGenerationHttp( xllm::ClosureGuard done_guard( done, [](void* /*unused*/) { request_in_metric(nullptr); }, - [controller](void* /*unused*/) { - request_out_metric(static_cast(controller)); + [controller](void* completion_context) { + complete_request_metric(controller, completion_context); }); if (!request || !response || !controller) { LOG(ERROR) << "brpc request | respose | controller is null"; @@ -587,13 +675,16 @@ void APIService::ImageGenerationHttp( butil::IOBufAsZeroCopyInputStream iobuf_stream(buf); auto st = json2pb::JsonToProtoMessage(&iobuf_stream, req_pb, options, &error); if (!st) { - ctrl->SetFailed(error); - LOG(ERROR) << "parse json to proto failed: " << error; + fail_http_request(ctrl, StatusCode::INVALID_ARGUMENT, error); return; } std::shared_ptr call = - std::make_shared( - ctrl, done_guard.release(), req_pb, resp_pb, arena != nullptr); + std::make_shared(ctrl, + done_guard.release_for_async(), + req_pb, + resp_pb, + arena != nullptr, + true); image_generation_service_impl_->process_async(call); } @@ -618,8 +709,8 @@ void APIService::AudioGenerationHttp( xllm::ClosureGuard done_guard( done, [](void* /*unused*/) { request_in_metric(nullptr); }, - [controller](void* /*unused*/) { - request_out_metric(static_cast(controller)); + [controller](void* completion_context) { + complete_request_metric(controller, completion_context); }); if (!request || !response || !controller) { LOG(ERROR) << "brpc request | response | controller is null"; @@ -641,13 +732,16 @@ void APIService::AudioGenerationHttp( butil::IOBufAsZeroCopyInputStream iobuf_stream(buf); bool st = json2pb::JsonToProtoMessage(&iobuf_stream, req_pb, options, &error); if (!st) { - ctrl->SetFailed(error); - LOG(ERROR) << "parse json to proto failed: " << error; + fail_http_request(ctrl, StatusCode::INVALID_ARGUMENT, error); return; } std::shared_ptr call = - std::make_shared( - ctrl, done_guard.release(), req_pb, resp_pb, arena != nullptr); + std::make_shared(ctrl, + done_guard.release_for_async(), + req_pb, + resp_pb, + arena != nullptr, + true); audio_generation_service_impl_->process_async(call); } @@ -672,8 +766,8 @@ void APIService::TextGenerationHttp( xllm::ClosureGuard done_guard( done, [](void* /*unused*/) { request_in_metric(nullptr); }, - [controller](void* /*unused*/) { - request_out_metric(static_cast(controller)); + [controller](void* completion_context) { + complete_request_metric(controller, completion_context); }); if (!request || !response || !controller) { LOG(ERROR) << "brpc request | response | controller is null"; @@ -695,13 +789,16 @@ void APIService::TextGenerationHttp( butil::IOBufAsZeroCopyInputStream iobuf_stream(buf); bool st = json2pb::JsonToProtoMessage(&iobuf_stream, req_pb, options, &error); if (!st) { - ctrl->SetFailed(error); - LOG(ERROR) << "parse json to proto failed: " << error; + fail_http_request(ctrl, StatusCode::INVALID_ARGUMENT, error); return; } std::shared_ptr call = - std::make_shared( - ctrl, done_guard.release(), req_pb, resp_pb, arena != nullptr); + std::make_shared(ctrl, + done_guard.release_for_async(), + req_pb, + resp_pb, + arena != nullptr, + true); text_generation_service_impl_->process_async(call); } @@ -726,8 +823,8 @@ void APIService::VideoGenerationHttp( xllm::ClosureGuard done_guard( done, [](void* /*unused*/) { request_in_metric(nullptr); }, - [controller](void* /*unused*/) { - request_out_metric(static_cast(controller)); + [controller](void* completion_context) { + complete_request_metric(controller, completion_context); }); if (!request || !response || !controller) { LOG(ERROR) << "brpc request | respose | controller is null"; @@ -749,13 +846,16 @@ void APIService::VideoGenerationHttp( butil::IOBufAsZeroCopyInputStream iobuf_stream(buf); auto st = json2pb::JsonToProtoMessage(&iobuf_stream, req_pb, options, &error); if (!st) { - ctrl->SetFailed(error); - LOG(ERROR) << "parse json to proto failed: " << error; + fail_http_request(ctrl, StatusCode::INVALID_ARGUMENT, error); return; } std::shared_ptr call = - std::make_shared( - ctrl, done_guard.release(), req_pb, resp_pb, arena != nullptr); + std::make_shared(ctrl, + done_guard.release_for_async(), + req_pb, + resp_pb, + arena != nullptr, + true); video_generation_service_impl_->process_async(call); } @@ -774,8 +874,8 @@ void APIService::RerankHttp(::google::protobuf::RpcController* controller, xllm::ClosureGuard done_guard( done, [](void* /*unused*/) { request_in_metric(nullptr); }, - [controller](void* /*unused*/) { - request_out_metric(static_cast(controller)); + [controller](void* completion_context) { + complete_request_metric(controller, completion_context); }); if (!request || !response || !controller) { LOG(ERROR) << "brpc request | respose | controller is null"; @@ -795,13 +895,17 @@ void APIService::RerankHttp(::google::protobuf::RpcController* controller, butil::IOBufAsZeroCopyInputStream iobuf_stream(buf); auto st = json2pb::JsonToProtoMessage(&iobuf_stream, req_pb, options, &error); if (!st) { - ctrl->SetFailed(error); - LOG(ERROR) << "parse json to proto failed: " << error; + fail_http_request(ctrl, StatusCode::INVALID_ARGUMENT, error); return; } - std::shared_ptr call = std::make_shared( - ctrl, done_guard.release(), req_pb, resp_pb, arena != nullptr); + std::shared_ptr call = + std::make_shared(ctrl, + done_guard.release_for_async(), + req_pb, + resp_pb, + arena != nullptr, + true); rerank_service_impl_->process_async(call); } @@ -840,20 +944,20 @@ void APIService::ModelsHttp(::google::protobuf::RpcController* controller, auto resp_pb = google::protobuf::Arena::CreateMessage(arena); + auto ctrl = reinterpret_cast(controller); bool st_models = models_service_impl_->list_models(nullptr, resp_pb); if (!st_models) { - LOG(ERROR) << "list models failed."; + fail_http_request(ctrl, StatusCode::UNKNOWN, "list models failed."); return; } - auto ctrl = reinterpret_cast(controller); json2pb::Pb2JsonOptions json_options; json_options.bytes_to_base64 = false; std::string err_msg; butil::IOBufAsZeroCopyOutputStream json_output(&ctrl->response_attachment()); if (!json2pb::ProtoMessageToJson( *resp_pb, &json_output, json_options, &err_msg)) { - LOG(ERROR) << "proto to json failed"; + fail_http_request(ctrl, StatusCode::UNKNOWN, err_msg); return; } } @@ -893,7 +997,9 @@ void handle_anthropic_messages(std::unique_ptr& service, auto content_len = get_json_content_length(ctrl); if (content_len == (size_t)-1L) { - ctrl->SetFailed("Content-Length header is missing."); + fail_anthropic_request(ctrl, + StatusCode::INVALID_ARGUMENT, + "Content-Length header is missing."); return; } std::string attachment; @@ -902,9 +1008,8 @@ void handle_anthropic_messages(std::unique_ptr& service, auto [preprocess_status, processed_json] = ChatJsonParser::anthropic().preprocess(std::move(attachment)); if (!preprocess_status.ok()) { - ctrl->SetFailed(preprocess_status.message()); - LOG(ERROR) << "Anthropic JSON preprocessing failed: " - << preprocess_status.message(); + fail_anthropic_request( + ctrl, preprocess_status.code(), preprocess_status.message()); return; } @@ -913,13 +1018,17 @@ void handle_anthropic_messages(std::unique_ptr& service, auto status = google::protobuf::util::JsonStringToMessage( processed_json, req_pb, options); if (!status.ok()) { - ctrl->SetFailed(status.ToString()); - LOG(ERROR) << "parse json to proto failed: " << status.ToString(); + fail_anthropic_request( + ctrl, StatusCode::INVALID_ARGUMENT, status.ToString()); return; } - auto call = std::make_shared( - ctrl, guard.release(), req_pb, resp_pb, arena != nullptr /*use_arena*/); + auto call = std::make_shared(ctrl, + guard.release_for_async(), + req_pb, + resp_pb, + arena != nullptr /*use_arena*/, + true /*is_http_request*/); service->process_async(call); } @@ -934,8 +1043,8 @@ void APIService::AnthropicMessagesHttp( xllm::ClosureGuard done_guard( done, [](void* /*unused*/) { request_in_metric(nullptr); }, - [controller](void* /*unused*/) { - request_out_metric(static_cast(controller)); + [controller](void* completion_context) { + complete_request_metric(controller, completion_context); }); if (!request || !response || !controller) { @@ -949,8 +1058,10 @@ void APIService::AnthropicMessagesHttp( handle_anthropic_messages( anthropic_service_impl_, done_guard, ctrl, request, response); } else { - ctrl->SetFailed("Anthropic messages API is only supported for LLM engine"); - LOG(ERROR) << "Anthropic messages API is only supported for LLM engine"; + fail_anthropic_request( + ctrl, + StatusCode::INVALID_ARGUMENT, + "Anthropic messages API is only supported for LLM engine"); } } @@ -1083,15 +1194,13 @@ void APIService::ForkMasterHttp(::google::protobuf::RpcController* controller, butil::IOBufAsZeroCopyInputStream iobuf_stream(buf); auto st = json2pb::JsonToProtoMessage(&iobuf_stream, req_pb, options, &error); if (!st) { - ctrl->SetFailed(error); - LOG(ERROR) << "parse json to proto failed: " << error; + fail_http_request(ctrl, StatusCode::INVALID_ARGUMENT, error); return; } std::string error_message; if (!do_fork_master(*req_pb, &error_message)) { - LOG(ERROR) << "fork_master failed: " << error_message; - ctrl->SetFailed(error_message); + fail_http_request(ctrl, StatusCode::UNKNOWN, error_message); } } @@ -1180,14 +1289,13 @@ void APIService::SleepHttp(::google::protobuf::RpcController* controller, butil::IOBufAsZeroCopyInputStream iobuf_stream(buf); auto st = json2pb::JsonToProtoMessage(&iobuf_stream, req_pb, options, &error); if (!st) { - ctrl->SetFailed(error); - LOG(ERROR) << "parse json to proto failed: " << error; + fail_http_request(ctrl, StatusCode::INVALID_ARGUMENT, error); return; } std::string error_message; if (!do_sleep(*req_pb, &error_message)) { - ctrl->SetFailed(error_message); + fail_http_request(ctrl, StatusCode::UNKNOWN, error_message); } // Success: return HTTP 200 with empty body } @@ -1289,14 +1397,13 @@ void APIService::WakeupHttp(::google::protobuf::RpcController* controller, butil::IOBufAsZeroCopyInputStream iobuf_stream(buf); auto st = json2pb::JsonToProtoMessage(&iobuf_stream, req_pb, options, &error); if (!st) { - ctrl->SetFailed(error); - LOG(ERROR) << "parse json to proto failed: " << error; + fail_http_request(ctrl, StatusCode::INVALID_ARGUMENT, error); return; } std::string error_message; if (!do_wakeup(*req_pb, &error_message)) { - ctrl->SetFailed(error_message); + fail_http_request(ctrl, StatusCode::UNKNOWN, error_message); } // Success: return HTTP 200 with empty body } @@ -1314,23 +1421,21 @@ void APIService::StartProfileHttp(::google::protobuf::RpcController* controller, auto ctrl = reinterpret_cast(controller); if (!ProfileConfig::get_instance().enable_online_profile()) { - LOG(ERROR) << "Profiling is disabled. Start the server with " - "--enable_online_profile=true to use /start_profile."; - ctrl->SetFailed( - "Profiling is disabled. Start the server with " - "--enable_online_profile=true."); + fail_http_request(ctrl, + StatusCode::UNAVAILABLE, + "Profiling is disabled. Start the server with " + "--enable_online_profile=true."); return; } if (master_ == nullptr) { - LOG(ERROR) << "No master available to start profiling."; - ctrl->SetFailed("No master available to start profiling."); + fail_http_request( + ctrl, StatusCode::UNKNOWN, "No master available to start profiling."); return; } LOG(INFO) << "Starting profiler."; if (!master_->start_profile()) { - LOG(ERROR) << "Failed to start profiler."; - ctrl->SetFailed("Failed to start profiler."); + fail_http_request(ctrl, StatusCode::UNKNOWN, "Failed to start profiler."); return; } LOG(INFO) << "Profiler started."; @@ -1350,23 +1455,21 @@ void APIService::StopProfileHttp(::google::protobuf::RpcController* controller, auto ctrl = reinterpret_cast(controller); if (!ProfileConfig::get_instance().enable_online_profile()) { - LOG(ERROR) << "Profiling is disabled. Start the server with " - "--enable_online_profile=true to use /stop_profile."; - ctrl->SetFailed( - "Profiling is disabled. Start the server with " - "--enable_online_profile=true."); + fail_http_request(ctrl, + StatusCode::UNAVAILABLE, + "Profiling is disabled. Start the server with " + "--enable_online_profile=true."); return; } if (master_ == nullptr) { - LOG(ERROR) << "No master available to stop profiling."; - ctrl->SetFailed("No master available to stop profiling."); + fail_http_request( + ctrl, StatusCode::UNKNOWN, "No master available to stop profiling."); return; } LOG(INFO) << "Stopping profiler."; if (!master_->stop_profile()) { - LOG(ERROR) << "Failed to stop profiler."; - ctrl->SetFailed("Failed to stop profiler."); + fail_http_request(ctrl, StatusCode::UNKNOWN, "Failed to stop profiler."); return; } LOG(INFO) << "Profiler stopped."; @@ -1417,15 +1520,15 @@ void APIService::LinkP2PHttp(::google::protobuf::RpcController* controller, butil::IOBufAsZeroCopyInputStream iobuf_stream(buf); auto st = json2pb::JsonToProtoMessage(&iobuf_stream, req_pb, options, &error); if (!st) { - ctrl->SetFailed(error); - LOG(ERROR) << "parse json to proto failed: " << error; + fail_http_request(ctrl, StatusCode::INVALID_ARGUMENT, error); return; } Master* master = get_model_master(req_pb->model_id()); if (master == nullptr) { LOG(ERROR) << "Master for model " << req_pb->model_id() << " not found"; - ctrl->SetFailed("Master for model not found"); + fail_http_request( + ctrl, StatusCode::INVALID_ARGUMENT, "Master for model not found"); return; } bool status = master->link_p2p( @@ -1487,15 +1590,15 @@ void APIService::UnlinkP2PHttp(::google::protobuf::RpcController* controller, butil::IOBufAsZeroCopyInputStream iobuf_stream(buf); auto st = json2pb::JsonToProtoMessage(&iobuf_stream, req_pb, options, &error); if (!st) { - ctrl->SetFailed(error); - LOG(ERROR) << "parse json to proto failed: " << error; + fail_http_request(ctrl, StatusCode::INVALID_ARGUMENT, error); return; } Master* master = get_model_master(req_pb->model_id()); if (master == nullptr) { LOG(ERROR) << "Master for model " << req_pb->model_id() << " not found"; - ctrl->SetFailed("Master for model not found"); + fail_http_request( + ctrl, StatusCode::INVALID_ARGUMENT, "Master for model not found"); return; } bool status = master->unlink_p2p( @@ -1568,8 +1671,7 @@ void APIService::PauseHttp(::google::protobuf::RpcController* controller, auto st = json2pb::JsonToProtoMessage(&iobuf_stream, req_pb, options, &error); if (!st) { - ctrl->SetFailed(error); - LOG(ERROR) << "PauseHttp: parse json to proto failed: " << error; + fail_http_request(ctrl, StatusCode::INVALID_ARGUMENT, error); return; } } diff --git a/xllm/api_service/audio_generation_service_impl.cpp b/xllm/api_service/audio_generation_service_impl.cpp index 4cf26d8566..2588ed7273 100644 --- a/xllm/api_service/audio_generation_service_impl.cpp +++ b/xllm/api_service/audio_generation_service_impl.cpp @@ -72,7 +72,7 @@ void AudioGenerationServiceImpl::process_async_impl( } DiTRequestParams request_params( - rpc_request, call->get_x_request_id(), call->get_x_request_time()); + rpc_request, call->x_request_id(), call->x_request_time()); std::string saved_request_id = request_params.request_id; master_->handle_request( diff --git a/xllm/api_service/call.cpp b/xllm/api_service/call.cpp index 0938493f52..b060031592 100644 --- a/xllm/api_service/call.cpp +++ b/xllm/api_service/call.cpp @@ -15,20 +15,30 @@ limitations under the License. #include "call.h" +#include "api_service/api_error.h" #include "core/common/constants.h" +#include "core/util/verbose_trace_logger.h" namespace xllm { -Call::Call(brpc::Controller* controller) : controller_(controller) { init(); } +Call::Call(brpc::Controller* controller, + std::string body_x_request_id, + bool is_http_request, + CompletionCallback completion_callback) + : controller_(controller), + is_http_request_(is_http_request), + completion_callback_(std::move(completion_callback)) { + init(std::move(body_x_request_id)); +} -void Call::init() { - if (controller_->http_request().GetHeader("x-request-id")) { - x_request_id_ = *controller_->http_request().GetHeader("x-request-id"); - } else if (controller_->http_request().GetHeader("x-ms-client-request-id")) { - x_request_id_ = - *controller_->http_request().GetHeader("x-ms-client-request-id"); +void Call::complete_request() { + if (completion_callback_) { + auto callback = std::move(completion_callback_); + callback(&completion_status_); } +} +void Call::init(std::string body_x_request_id) { if (controller_->http_request().GetHeader("x-request-time")) { x_request_time_ = *controller_->http_request().GetHeader("x-request-time"); } else if (controller_->http_request().GetHeader("x-request-timems")) { @@ -36,6 +46,20 @@ void Call::init() { *controller_->http_request().GetHeader("x-request-timems"); } + // Resolve the request-scoped x-request-id once, here, so logs, the response + // header and the engine use the same value. + x_request_id_ = api_service::get_header_x_request_id(controller_); + if (x_request_id_.empty()) { + x_request_id_ = std::move(body_x_request_id); + } + if (x_request_id_.empty()) { + x_request_id_ = api_service::generate_x_request_id(); + } + controller_->http_response().SetHeader("x-request-id", x_request_id_); + XLLM_VERBOSE_TRACE() << "event=request_received x-request-id=" + << x_request_id_ + << " path=" << controller_->http_request().uri().path(); + init_request_payload(); } diff --git a/xllm/api_service/call.h b/xllm/api_service/call.h index dc934729f6..3088c0e5d5 100644 --- a/xllm/api_service/call.h +++ b/xllm/api_service/call.h @@ -17,17 +17,45 @@ limitations under the License. #include +#include #include +#include namespace xllm { +template +std::string request_body_x_request_id(const Request* request) { + if constexpr (requires(const Request& value) { + value.has_x_request_id(); + value.x_request_id(); + }) { + if (request != nullptr && request->has_x_request_id()) { + return request->x_request_id(); + } + } + return ""; +} + class Call { public: - Call(brpc::Controller* controller); + struct CompletionStatus { + bool failed = false; + }; + + using CompletionCallback = std::function; + + Call(brpc::Controller* controller, + std::string body_x_request_id = "", + bool is_http_request = false, + CompletionCallback completion_callback = {}); virtual ~Call() = default; - std::string get_x_request_id() { return x_request_id_; } - std::string get_x_request_time() { return x_request_time_; } + const std::string& x_request_time() const { return x_request_time_; } + + // The request-scoped x-request-id: the client header when present, then the + // request body value, otherwise a server-generated id. Always non-empty + // after construction and shared across logs, the response and the engine. + const std::string& x_request_id() const { return x_request_id_; } std::string take_request_payload() { return std::move(request_payload_); } void init_request_payload(); @@ -35,13 +63,19 @@ class Call { virtual bool is_disconnected() const = 0; protected: - void init(); + void init(std::string body_x_request_id); + void complete_request(); + void mark_request_failed() { completion_status_.failed = true; } + bool is_http_request() const { return is_http_request_; } protected: brpc::Controller* controller_; + bool is_http_request_ = false; + CompletionCallback completion_callback_; + CompletionStatus completion_status_; - std::string x_request_id_; std::string x_request_time_; + std::string x_request_id_; std::string request_payload_; }; diff --git a/xllm/api_service/chat_service_impl.cpp b/xllm/api_service/chat_service_impl.cpp index 876ecb99de..c9852092d2 100644 --- a/xllm/api_service/chat_service_impl.cpp +++ b/xllm/api_service/chat_service_impl.cpp @@ -525,7 +525,7 @@ void ChatServiceImpl::process_rec_chat_request(std::shared_ptr call) { } RequestParams request_params( - rpc_request, call->get_x_request_id(), call->get_x_request_time()); + rpc_request, call->x_request_id(), call->x_request_time()); // Build messages from rpc request (inline code, not extracted to helper) std::vector messages; @@ -755,7 +755,7 @@ void ChatServiceImpl::process_async_impl(std::shared_ptr call) { } RequestParams request_params( - rpc_request, call->get_x_request_id(), call->get_x_request_time()); + rpc_request, call->x_request_id(), call->x_request_time()); std::vector messages; messages.reserve(rpc_request.messages_size()); for (const auto& message : rpc_request.messages()) { @@ -911,7 +911,7 @@ void MMChatServiceImpl::process_async_impl(std::shared_ptr call) { } RequestParams request_params( - rpc_request, call->get_x_request_id(), call->get_x_request_time()); + rpc_request, call->x_request_id(), call->x_request_time()); std::vector messages; if (!mm_service_utils::build_messages( diff --git a/xllm/api_service/completion_service_impl.cpp b/xllm/api_service/completion_service_impl.cpp index c3b7d2395b..5e128e67f5 100644 --- a/xllm/api_service/completion_service_impl.cpp +++ b/xllm/api_service/completion_service_impl.cpp @@ -272,7 +272,7 @@ void CompletionServiceImpl::process_async_impl( } RequestParams request_params( - rpc_request, call->get_x_request_id(), call->get_x_request_time()); + rpc_request, call->x_request_id(), call->x_request_time()); bool include_usage = false; if (rpc_request.has_stream_options()) { include_usage = rpc_request.stream_options().include_usage(); diff --git a/xllm/api_service/embedding_service_impl.cpp b/xllm/api_service/embedding_service_impl.cpp index 7fb9bc7b93..8780aaacad 100644 --- a/xllm/api_service/embedding_service_impl.cpp +++ b/xllm/api_service/embedding_service_impl.cpp @@ -101,7 +101,7 @@ void EmbeddingServiceImpl::process_async_impl( // create RequestParams for embeddings request // set is_embeddings and max_tokens = 1 to control engine step once. RequestParams request_params( - rpc_request, call->get_x_request_id(), call->get_x_request_time()); + rpc_request, call->x_request_id(), call->x_request_time()); // TODO only support input_str for now auto& input = rpc_request.input(); @@ -150,7 +150,7 @@ void MMEmbeddingServiceImpl::process_async_impl( // create RequestParams for embeddings request // set is_embeddings and max_tokens = 1 to control engine step once. RequestParams request_params( - rpc_request, call->get_x_request_id(), call->get_x_request_time()); + rpc_request, call->x_request_id(), call->x_request_time()); auto& req_messages = rpc_request.messages(); diff --git a/xllm/api_service/image_generation_service_impl.cpp b/xllm/api_service/image_generation_service_impl.cpp index 01ffb67080..ff217c8fe0 100644 --- a/xllm/api_service/image_generation_service_impl.cpp +++ b/xllm/api_service/image_generation_service_impl.cpp @@ -89,7 +89,7 @@ void ImageGenerationServiceImpl::process_async_impl( // create DiTRequestParams for image generation request DiTRequestParams request_params( - rpc_request, call->get_x_request_id(), call->get_x_request_time()); + rpc_request, call->x_request_id(), call->x_request_time()); auto saved_request_id = request_params.request_id; // schedule the request diff --git a/xllm/api_service/non_stream_call.h b/xllm/api_service/non_stream_call.h index d1fe96bac5..12e48bf638 100644 --- a/xllm/api_service/non_stream_call.h +++ b/xllm/api_service/non_stream_call.h @@ -25,8 +25,11 @@ limitations under the License. #include #include +#include "api_service/api_error.h" #include "call.h" #include "core/common/types.h" +#include "core/util/closure_guard.h" +#include "core/util/verbose_trace_logger.h" namespace xllm { @@ -39,20 +42,32 @@ class NonStreamCall : public Call { ::google::protobuf::Closure* done, Request* request, Response* response, - bool use_arena = false) - : Call(controller), - done_(done), - request_(request), - response_(response), - use_arena_(use_arena) { - controller_->http_response().set_content_type("application/json"); + bool use_arena = false, + bool is_http_request = false) + : NonStreamCall(controller, + done, + request, + response, + use_arena, + is_http_request, + {}) {} - json_options_.bytes_to_base64 = false; - json_options_.jsonify_empty_array = true; - json_options_.always_print_primitive_fields = true; - } + NonStreamCall(brpc::Controller* controller, + ClosureGuard::AsyncClosure async_closure, + Request* request, + Response* response, + bool use_arena = false, + bool is_http_request = false) + : NonStreamCall(controller, + async_closure.done, + request, + response, + use_arena, + is_http_request, + std::move(async_closure.after_done)) {} ~NonStreamCall() override { + complete_request(); done_->Run(); if (!use_arena_) { @@ -61,6 +76,30 @@ class NonStreamCall : public Call { } } + private: + NonStreamCall(brpc::Controller* controller, + ::google::protobuf::Closure* done, + Request* request, + Response* response, + bool use_arena, + bool is_http_request, + CompletionCallback completion_callback) + : Call(controller, + request_body_x_request_id(request), + is_http_request, + std::move(completion_callback)), + done_(done), + request_(request), + response_(response), + use_arena_(use_arena) { + controller_->http_response().set_content_type("application/json"); + + json_options_.bytes_to_base64 = false; + json_options_.jsonify_empty_array = true; + json_options_.always_print_primitive_fields = true; + } + + public: void set_bytes_to_base64(bool bytes_to_base64) { json_options_.bytes_to_base64 = bytes_to_base64; } @@ -72,9 +111,12 @@ class NonStreamCall : public Call { std::string err_msg; if (!json2pb::ProtoMessageToJson( response, &json_output, json_options_, &err_msg)) { - return finish_with_error(StatusCode::UNKNOWN, err_msg); + finish_with_error(StatusCode::UNKNOWN, err_msg); + return false; } + XLLM_VERBOSE_TRACE() << "event=request_completed http=200 x-request-id=" + << x_request_id(); return true; } @@ -102,7 +144,13 @@ class NonStreamCall : public Call { // For non stream response bool finish_with_error(const StatusCode& code, const std::string& error_message) { - controller_->SetFailed(error_message); + mark_request_failed(); + if (is_http_request()) { + api_service::write_openai_error( + controller_, code, error_message, x_request_id()); + } else { + controller_->SetFailed(error_message); + } return true; } diff --git a/xllm/api_service/qwen3_rerank_service_impl.cpp b/xllm/api_service/qwen3_rerank_service_impl.cpp index b0a440ccac..7c4bebe5c6 100644 --- a/xllm/api_service/qwen3_rerank_service_impl.cpp +++ b/xllm/api_service/qwen3_rerank_service_impl.cpp @@ -48,7 +48,7 @@ void Qwen3RerankServiceImpl::process_async_impl( } RequestParams request_params( - rpc_request, call->get_x_request_id(), call->get_x_request_time()); + rpc_request, call->x_request_id(), call->x_request_time()); std::vector sps(documents.size(), request_params); auto request_id = request_params.request_id; diff --git a/xllm/api_service/rec_completion_service_impl.cpp b/xllm/api_service/rec_completion_service_impl.cpp index a44d30be08..daded465e6 100644 --- a/xllm/api_service/rec_completion_service_impl.cpp +++ b/xllm/api_service/rec_completion_service_impl.cpp @@ -292,7 +292,7 @@ void RecCompletionServiceImpl::process_async_impl( Timer request_timer; RequestParams request_params( - rpc_request, call->get_x_request_id(), call->get_x_request_time()); + rpc_request, call->x_request_id(), call->x_request_time()); if (::xllm::RecConfig::get_instance().enable_output_sku_logprobs()) { request_params.logprobs = true; } diff --git a/xllm/api_service/rerank_service_impl.cpp b/xllm/api_service/rerank_service_impl.cpp index 9789f6e8dd..ef8c007e31 100644 --- a/xllm/api_service/rerank_service_impl.cpp +++ b/xllm/api_service/rerank_service_impl.cpp @@ -98,7 +98,7 @@ void RerankServiceImpl::process_async_impl(std::shared_ptr call) { documents.emplace_back(rpc_request.query()); RequestParams request_params( - rpc_request, call->get_x_request_id(), call->get_x_request_time()); + rpc_request, call->x_request_id(), call->x_request_time()); std::vector sps(documents.size(), request_params); auto request_id = request_params.request_id; diff --git a/xllm/api_service/sample_service_impl.cpp b/xllm/api_service/sample_service_impl.cpp index eeaf67d716..57f48ce47a 100644 --- a/xllm/api_service/sample_service_impl.cpp +++ b/xllm/api_service/sample_service_impl.cpp @@ -403,6 +403,7 @@ void SampleServiceImpl::process_async_impl(std::shared_ptr call) { "Failed to build sample selector runtime mapping"); return; } + request_params.set_x_request_id_if_absent(call->x_request_id()); if (request_params.sample_slots.empty()) { if (!sample_service_internal::build_empty_response( diff --git a/xllm/api_service/stream_call.h b/xllm/api_service/stream_call.h index 993faf0642..1262c2909c 100644 --- a/xllm/api_service/stream_call.h +++ b/xllm/api_service/stream_call.h @@ -27,8 +27,11 @@ limitations under the License. #include "anthropic.pb.h" #include "api_service/anthropic_json.h" +#include "api_service/api_error.h" #include "api_service/call.h" #include "core/common/types.h" +#include "core/util/closure_guard.h" +#include "core/util/verbose_trace_logger.h" namespace xllm { @@ -42,8 +45,54 @@ class StreamCall : public Call { ::google::protobuf::Closure* done, Request* request, Response* response, - bool use_arena = false) - : Call(controller), + bool use_arena = false, + bool is_http_request = false) + : StreamCall(controller, + done, + request, + response, + use_arena, + is_http_request, + {}) {} + + StreamCall(brpc::Controller* controller, + ClosureGuard::AsyncClosure async_closure, + Request* request, + Response* response, + bool use_arena = false, + bool is_http_request = false) + : StreamCall(controller, + async_closure.done, + request, + response, + use_arena, + is_http_request, + std::move(async_closure.after_done)) {} + + ~StreamCall() override { + complete_request(); + // For non stream response, call brpc done Run + if (!stream_) { + done_->Run(); + } + if (!use_arena_) { + delete request_; + delete response_; + } + } + + private: + StreamCall(brpc::Controller* controller, + ::google::protobuf::Closure* done, + Request* request, + Response* response, + bool use_arena, + bool is_http_request, + CompletionCallback completion_callback) + : Call(controller, + request_body_x_request_id(request), + is_http_request, + std::move(completion_callback)), done_(done), request_(request), response_(response), @@ -69,17 +118,7 @@ class StreamCall : public Call { json_options_.jsonify_empty_array = true; } - ~StreamCall() override { - // For non stream response, call brpc done Run - if (!stream_) { - done_->Run(); - } - if (!use_arena_) { - delete request_; - delete response_; - } - } - + public: bool write_and_finish(Response& response) { butil::IOBufAsZeroCopyOutputStream json_output( &controller_->response_attachment()); @@ -88,18 +127,28 @@ class StreamCall : public Call { response, &json_output, json_options_, &err_msg)) { return finish_with_error(StatusCode::UNKNOWN, err_msg); } + XLLM_VERBOSE_TRACE() << "event=request_completed http=200 x-request-id=" + << x_request_id(); return true; } - bool finish_with_error(const StatusCode& code, - const std::string& error_message) { - if (!stream_) { + virtual bool finish_with_error(const StatusCode& code, + const std::string& error_message) { + mark_request_failed(); + if (!is_http_request()) { controller_->SetFailed(error_message); - + } else if (!stream_) { + api_service::write_openai_error( + controller_, code, error_message, x_request_id()); } else { + // The stream has already committed a 200 header, so the error can only be + // surfaced as a terminal SSE event carrying the OpenAI error envelope. io_buf_.clear(); - io_buf_.append(error_message); + io_buf_.append("data: "); + io_buf_.append(api_service::make_openai_error_json(code, error_message)); + io_buf_.append("\n\n"); pa_->Write(io_buf_); + api_service::log_request_error(code, error_message, x_request_id()); } return true; @@ -128,6 +177,8 @@ class StreamCall : public Call { io_buf_.append("data: [DONE]\n\n"); pa_->Write(io_buf_); + XLLM_VERBOSE_TRACE() << "event=request_completed http=200 x-request-id=" + << x_request_id(); return true; } @@ -170,15 +221,52 @@ class AnthropicCall : public StreamCall(controller, done, request, response, - use_arena) {} + use_arena, + is_http_request) {} + + AnthropicCall(brpc::Controller* controller, + ClosureGuard::AsyncClosure async_closure, + proto::AnthropicMessagesRequest* request, + proto::AnthropicMessagesResponse* response, + bool use_arena = false, + bool is_http_request = false) + : StreamCall(controller, + std::move(async_closure), + request, + response, + use_arena, + is_http_request) {} - ~AnthropicCall() {} + ~AnthropicCall() override = default; + + // Anthropic uses its own error envelope: a JSON object + // {"type":"error","error":{"type":...,"message":...}} for non-stream and an + // `error` SSE event for streaming responses. + bool finish_with_error(const StatusCode& code, + const std::string& error_message) override { + this->mark_request_failed(); + if (!this->stream_) { + api_service::write_anthropic_error( + this->controller_, code, error_message, this->x_request_id()); + } else { + this->io_buf_.clear(); + this->io_buf_.append("event: error\ndata: "); + this->io_buf_.append( + api_service::make_anthropic_error_json(code, error_message)); + this->io_buf_.append("\n\n"); + this->pa_->Write(this->io_buf_); + api_service::log_request_error(code, error_message, this->x_request_id()); + } + return true; + } bool write_and_finish(proto::AnthropicMessagesResponse& response) { std::string json; @@ -188,6 +276,8 @@ class AnthropicCall : public StreamCallfinish_with_error(StatusCode::UNKNOWN, err_msg); } this->controller_->response_attachment().append(json); + XLLM_VERBOSE_TRACE() << "event=request_completed http=200 x-request-id=" + << this->x_request_id(); return true; } diff --git a/xllm/api_service/text_generation_service_impl.cpp b/xllm/api_service/text_generation_service_impl.cpp index bd28d4e442..6190a84c6b 100644 --- a/xllm/api_service/text_generation_service_impl.cpp +++ b/xllm/api_service/text_generation_service_impl.cpp @@ -75,7 +75,7 @@ void TextGenerationServiceImpl::process_async_impl( } DiTRequestParams request_params( - rpc_request, call->get_x_request_id(), call->get_x_request_time()); + rpc_request, call->x_request_id(), call->x_request_time()); const std::string saved_request_id = request_params.request_id; master_->handle_request( diff --git a/xllm/api_service/video_generation_service_impl.cpp b/xllm/api_service/video_generation_service_impl.cpp index 25359aed5d..76841c5107 100644 --- a/xllm/api_service/video_generation_service_impl.cpp +++ b/xllm/api_service/video_generation_service_impl.cpp @@ -77,7 +77,7 @@ void VideoGenerationServiceImpl::process_async_impl( } DiTRequestParams request_params( - rpc_request, call->get_x_request_id(), call->get_x_request_time()); + rpc_request, call->x_request_id(), call->x_request_time()); std::string saved_request_id = request_params.request_id; master_->handle_request( diff --git a/xllm/api_service/xllm_metrics.h b/xllm/api_service/xllm_metrics.h index b8c1f0e6d1..c2e2ec4302 100644 --- a/xllm/api_service/xllm_metrics.h +++ b/xllm/api_service/xllm_metrics.h @@ -13,6 +13,8 @@ See the License for the specific language governing permissions and limitations under the License. ==============================================================================*/ +#pragma once + #include #include @@ -20,6 +22,10 @@ limitations under the License. namespace xllm { +static bool is_failed_request(const brpc::Controller* ctrl) { + return ctrl->Failed() || ctrl->http_response().status_code() >= 400; +} + static void request_in_metric(void* context) { COUNTER_INC(server_request_in_total); } @@ -31,7 +37,7 @@ static void request_out_metric(void* context) { return; } - if (!ctrl->Failed()) { + if (!is_failed_request(ctrl)) { COUNTER_INC(server_request_total_ok); } else { COUNTER_INC(server_request_total_fail); @@ -41,6 +47,14 @@ static void request_out_metric(void* context) { } } +static void request_out_metric(bool failed) { + if (failed) { + COUNTER_INC(server_request_total_fail); + } else { + COUNTER_INC(server_request_total_ok); + } +} + static void device_info_metric() { // TODO: get cpu device info GAUGE_SET(xllm_cpu_num, 0); diff --git a/xllm/core/common/global_flags.h b/xllm/core/common/global_flags.h index 45b2f9f1aa..8a9aad4e93 100755 --- a/xllm/core/common/global_flags.h +++ b/xllm/core/common/global_flags.h @@ -402,3 +402,12 @@ DECLARE_bool(enable_aclnn_swiglu); DECLARE_bool(use_cpp_chat_template); DECLARE_int32(health_check_interval_ms); + +// --- verbose trace logging config --- +DECLARE_bool(enable_verbose_trace_log); + +DECLARE_string(verbose_trace_log_path); + +DECLARE_int32(verbose_trace_log_max_size_mb); + +DECLARE_int32(verbose_trace_log_max_files); diff --git a/xllm/core/framework/config/service_config.cpp b/xllm/core/framework/config/service_config.cpp index 8f80d6ff44..9ed220a695 100644 --- a/xllm/core/framework/config/service_config.cpp +++ b/xllm/core/framework/config/service_config.cpp @@ -55,6 +55,35 @@ DEFINE_int32(health_check_interval_ms, 3000, "Worker health check interval in milliseconds."); +DEFINE_bool(enable_verbose_trace_log, + false, + "Enable asynchronous verbose request-trace logging to a file. When " + "enabled, failed-request context and request lifecycle events are " + "written off the serving hot path by a background thread."); + +DEFINE_string( + verbose_trace_log_path, + "log/verbose_trace.log", + "Base file path for the verbose request-trace log. When nnodes is greater " + "than 1, _rank_ is inserted before the final extension (for " + "example, log/verbose_trace_rank_2.log). Single-rank serving uses this " + "path " + "unchanged. Used only when verbose trace logging is enabled. Defaults to " + "log/verbose_trace.log, so it sits alongside node_.log."); + +DEFINE_int32(verbose_trace_log_max_size_mb, + 1024, + "Max size in MiB of a single verbose request-trace log file " + "before it is rotated. Defaults to 1024 (1 GiB)."); + +DEFINE_int32( + verbose_trace_log_max_files, + 1000, + "Max number of verbose request-trace log files to keep per rank, including " + "the active file. Each rank rotates independently and removes its oldest " + "file once this many exist; this is not a deployment-wide limit. Defaults " + "to 1000."); + namespace xllm { void ServiceConfig::from_flags() { @@ -68,6 +97,10 @@ void ServiceConfig::from_flags() { XLLM_CONFIG_ASSIGN_FROM_FLAG(num_request_handling_threads); XLLM_CONFIG_ASSIGN_FROM_FLAG(num_response_handling_threads); XLLM_CONFIG_ASSIGN_FROM_FLAG(health_check_interval_ms); + XLLM_CONFIG_ASSIGN_FROM_FLAG(enable_verbose_trace_log); + XLLM_CONFIG_ASSIGN_FROM_FLAG(verbose_trace_log_path); + XLLM_CONFIG_ASSIGN_FROM_FLAG(verbose_trace_log_max_size_mb); + XLLM_CONFIG_ASSIGN_FROM_FLAG(verbose_trace_log_max_files); } void ServiceConfig::from_json(const JsonReader& json) { @@ -82,6 +115,10 @@ void ServiceConfig::from_json(const JsonReader& json) { XLLM_CONFIG_ASSIGN_FROM_JSON(num_request_handling_threads); XLLM_CONFIG_ASSIGN_FROM_JSON(num_response_handling_threads); XLLM_CONFIG_ASSIGN_FROM_JSON(health_check_interval_ms); + XLLM_CONFIG_ASSIGN_FROM_JSON(enable_verbose_trace_log); + XLLM_CONFIG_ASSIGN_FROM_JSON(verbose_trace_log_path); + XLLM_CONFIG_ASSIGN_FROM_JSON(verbose_trace_log_max_size_mb); + XLLM_CONFIG_ASSIGN_FROM_JSON(verbose_trace_log_max_files); } void ServiceConfig::append_config_json( @@ -106,6 +143,14 @@ void ServiceConfig::append_config_json( config_json, default_config, num_response_handling_threads); APPEND_CONFIG_JSON_VALUE_IF_NOT_DEFAULT( config_json, default_config, health_check_interval_ms); + APPEND_CONFIG_JSON_VALUE_IF_NOT_DEFAULT( + config_json, default_config, enable_verbose_trace_log); + APPEND_CONFIG_JSON_VALUE_IF_NOT_DEFAULT( + config_json, default_config, verbose_trace_log_path); + APPEND_CONFIG_JSON_VALUE_IF_NOT_DEFAULT( + config_json, default_config, verbose_trace_log_max_size_mb); + APPEND_CONFIG_JSON_VALUE_IF_NOT_DEFAULT( + config_json, default_config, verbose_trace_log_max_files); } ServiceConfig& ServiceConfig::get_instance() { diff --git a/xllm/core/framework/config/service_config.h b/xllm/core/framework/config/service_config.h index 0c2882a060..9f6f8ac7a6 100644 --- a/xllm/core/framework/config/service_config.h +++ b/xllm/core/framework/config/service_config.h @@ -50,7 +50,11 @@ class ServiceConfig final { "max_concurrent_requests", "num_request_handling_threads", "num_response_handling_threads", - "health_check_interval_ms"}}; + "health_check_interval_ms", + "enable_verbose_trace_log", + "verbose_trace_log_path", + "verbose_trace_log_max_size_mb", + "verbose_trace_log_max_files"}}; return kOptionCategory; } @@ -73,6 +77,14 @@ class ServiceConfig final { PROPERTY(int32_t, num_response_handling_threads) = 4; PROPERTY(int32_t, health_check_interval_ms) = 3000; + + PROPERTY(bool, enable_verbose_trace_log) = false; + + PROPERTY(std::string, verbose_trace_log_path); + + PROPERTY(int32_t, verbose_trace_log_max_size_mb) = 1024; + + PROPERTY(int32_t, verbose_trace_log_max_files) = 1000; }; } // namespace xllm diff --git a/xllm/core/framework/prefix_cache/CMakeLists.txt b/xllm/core/framework/prefix_cache/CMakeLists.txt index a85ec9e129..559a88c93b 100644 --- a/xllm/core/framework/prefix_cache/CMakeLists.txt +++ b/xllm/core/framework/prefix_cache/CMakeLists.txt @@ -17,7 +17,7 @@ cc_library( DEPS $<$:torch_npu> $<$:graph> - :request + :multimodal :common glog::glog Boost::serialization diff --git a/xllm/core/framework/request/dit_request_params.h b/xllm/core/framework/request/dit_request_params.h index bc87a75731..5a4369aaa3 100644 --- a/xllm/core/framework/request/dit_request_params.h +++ b/xllm/core/framework/request/dit_request_params.h @@ -32,6 +32,16 @@ struct DiTRequestParams { const std::string& x_rid, const std::string& x_rtime); + // When neither the x-request-id header nor the request body supplied a client + // request id, fall back to `fallback` (the Call's generated trace id) so the + // logs, the verbose trace and the echoed response header all share one id. + // No-op when an id is already present. + void set_x_request_id_if_absent(const std::string& fallback) { + if (x_request_id.empty()) { + x_request_id = fallback; + } + } + bool verify_params(DiTOutputCallback callback) const; // request id diff --git a/xllm/core/framework/request/request_params.h b/xllm/core/framework/request/request_params.h index 5ddfdf8bbf..ef3a0554a7 100644 --- a/xllm/core/framework/request/request_params.h +++ b/xllm/core/framework/request/request_params.h @@ -62,6 +62,16 @@ struct RequestParams { const std::string& x_rid, const std::string& x_rtime); + // When neither the x-request-id header nor the request body supplied a client + // request id, fall back to `fallback` (the Call's generated trace id) so the + // logs, the verbose trace and the echoed response header all share one id. + // No-op when an id is already present. + void set_x_request_id_if_absent(const std::string& fallback) { + if (x_request_id.empty()) { + x_request_id = fallback; + } + } + bool verify_params(OutputCallback callback) const; // request id diff --git a/xllm/core/util/CMakeLists.txt b/xllm/core/util/CMakeLists.txt index ea8fc467ec..2bc1c7ffb9 100644 --- a/xllm/core/util/CMakeLists.txt +++ b/xllm/core/util/CMakeLists.txt @@ -34,6 +34,7 @@ cc_library( type_traits.h utils.h uuid.h + verbose_trace_logger.h shared_memory_manager.h SRCS cpu_affinity.cpp @@ -53,6 +54,7 @@ cc_library( timer.cpp utils.cpp uuid.cpp + verbose_trace_logger.cpp shared_memory_manager.cpp DEPS Folly::folly diff --git a/xllm/core/util/closure_guard.h b/xllm/core/util/closure_guard.h index 8f0305f919..076f4a0236 100644 --- a/xllm/core/util/closure_guard.h +++ b/xllm/core/util/closure_guard.h @@ -18,6 +18,7 @@ limitations under the License. #include #include +#include #include "butil/macros.h" @@ -26,6 +27,13 @@ namespace xllm { // RAII: Call Run() of the closure on destruction. class ClosureGuard { public: + using DoneCallback = std::function; + + struct AsyncClosure { + google::protobuf::Closure* done = nullptr; + DoneCallback after_done; + }; + ClosureGuard() : _done(nullptr), _before_done([](void*) {}), _after_done([](void*) {}) {} @@ -62,6 +70,16 @@ class ClosureGuard { return prev_done; } + // Transfer both the closure and its completion callback to an asynchronous + // owner. The callback must be run by that owner when the operation actually + // completes, before running done. + AsyncClosure release_for_async() { + AsyncClosure released{_done, std::move(_after_done)}; + _done = nullptr; + _after_done = [](void*) {}; + return released; + } + // True if no closure inside. bool empty() const { return _done == nullptr; } diff --git a/xllm/core/util/verbose_trace_logger.cpp b/xllm/core/util/verbose_trace_logger.cpp new file mode 100644 index 0000000000..a698f877c8 --- /dev/null +++ b/xllm/core/util/verbose_trace_logger.cpp @@ -0,0 +1,157 @@ +/* Copyright 2025-2026 The xLLM Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + https://github.com/jd-opensource/xllm/blob/main/LICENSE + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +==============================================================================*/ + +#include "core/util/verbose_trace_logger.h" + +#include +#include +#include +#include + +#include +#include +#include + +namespace xllm { + +namespace { + +// Background queue capacity (entries) and writer-thread count for the async +// logger. One writer keeps file output strictly ordered. +constexpr size_t kQueueSize = 8192; +constexpr size_t kWriterThreads = 1; + +constexpr int64_t kBytesPerMiB = 1024 * 1024; + +} // namespace + +std::string resolve_verbose_trace_log_path(const std::string& configured_path, + int32_t nnodes, + int32_t node_rank) { + if (configured_path.empty() || nnodes <= 1) { + return configured_path; + } + + const std::filesystem::path path(configured_path); + const std::string extension = path.extension().string(); + const size_t extension_position = configured_path.size() - extension.size(); + std::string resolved_path = configured_path; + resolved_path.insert(extension_position, + "_rank_" + std::to_string(node_rank)); + return resolved_path; +} + +VerboseTraceLogger& VerboseTraceLogger::get_instance() { + static VerboseTraceLogger instance; + return instance; +} + +VerboseTraceLogger::~VerboseTraceLogger() { shutdown(); } + +void VerboseTraceLogger::initialize(bool enabled, + const std::string& file_path, + int32_t max_size_mb, + int32_t max_files) { + if (!enabled) { + LOG(INFO) << "Verbose trace logging is disabled."; + return; + } + if (enabled_.load(std::memory_order_relaxed)) { + LOG(WARNING) << "Verbose trace logger already initialized; ignoring."; + return; + } + if (file_path.empty()) { + LOG(ERROR) << "Verbose trace logging enabled but no file path provided; " + "verbose trace logging stays disabled."; + return; + } + + std::error_code ec; + const std::filesystem::path path(file_path); + if (path.has_parent_path()) { + std::filesystem::create_directories(path.parent_path(), ec); + if (ec) { + LOG(ERROR) << "Failed to create directory for verbose trace log " + << file_path << ": " << ec.message() + << "; verbose trace logging stays disabled."; + return; + } + } + + // Guard against non-positive configuration that spdlog would reject. + const size_t max_size_bytes = + static_cast(std::max(1, max_size_mb)) * kBytesPerMiB; + const size_t total_files = static_cast(std::max(1, max_files)); + // spdlog's max_files parameter counts rotated backups and excludes the + // active file. The service option is the total on-disk file limit. + const size_t backup_files = total_files - 1; + + try { + thread_pool_ = std::make_shared( + kQueueSize, kWriterThreads); + auto sink = std::make_shared( + file_path, max_size_bytes, backup_files); + logger_ = std::make_shared( + "verbose_trace", + std::move(sink), + thread_pool_, + // Never block the serving hot path: drop the oldest queued entry if the + // background writer cannot keep up. + spdlog::async_overflow_policy::overrun_oldest); + // Prefix each entry with a millisecond wall-clock timestamp; the entry body + // is emitted verbatim and spdlog appends the trailing newline. + logger_->set_pattern("%Y-%m-%d %H:%M:%S.%e %v"); + logger_->set_level(spdlog::level::info); + // Flush each entry so a live tail and post-crash inspection see it + // promptly; this runs on the background thread, off the serving hot path. + logger_->flush_on(spdlog::level::info); + } catch (const std::exception& e) { + LOG(ERROR) << "Failed to open verbose trace log file " << file_path << ": " + << e.what() << "; verbose trace logging stays disabled."; + logger_.reset(); + thread_pool_.reset(); + return; + } + + enabled_.store(true, std::memory_order_release); + LOG(INFO) << "Verbose trace logging enabled, writing to " << file_path + << " (rotating at " << max_size_mb << " MiB, keeping up to " + << total_files << " files)."; +} + +void VerboseTraceLogger::log(const std::string& entry) { + if (!enabled_.load(std::memory_order_relaxed)) { + return; + } + // spdlog copies the message into its async ring buffer internally, so passing + // by const-ref is safe (the thread-local source buffer can be reused + // immediately after this returns). + logger_->info(entry); +} + +void VerboseTraceLogger::shutdown() { + if (!enabled_.exchange(false, std::memory_order_acq_rel)) { + return; + } + if (logger_) { + logger_->flush(); + logger_.reset(); + } + // Destroying the thread pool drains any queued entries, then joins the + // background writer. + thread_pool_.reset(); +} + +} // namespace xllm diff --git a/xllm/core/util/verbose_trace_logger.h b/xllm/core/util/verbose_trace_logger.h new file mode 100644 index 0000000000..83bba2221c --- /dev/null +++ b/xllm/core/util/verbose_trace_logger.h @@ -0,0 +1,138 @@ +/* Copyright 2025-2026 The xLLM Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + https://github.com/jd-opensource/xllm/blob/main/LICENSE + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +==============================================================================*/ + +#pragma once + +#include +#include +#include +#include +#include +#include + +namespace spdlog { +class logger; +namespace details { +class thread_pool; +} // namespace details +} // namespace spdlog + +namespace xllm { + +[[nodiscard]] std::string resolve_verbose_trace_log_path( + const std::string& configured_path, + int32_t nnodes, + int32_t node_rank); + +// Asynchronous request-trace logger. +// +// When enabled, serving threads format an entry into a thread-local buffer and +// hand it to spdlog's async queue with a single non-blocking memcpy (no file +// I/O, no heap allocation after the first call per thread). A dedicated +// background thread drains the queue and appends entries to a rotating file. +// When disabled, the only cost is one relaxed atomic load. +class VerboseTraceLogger final { + public: + static VerboseTraceLogger& get_instance(); + + VerboseTraceLogger(const VerboseTraceLogger&) = delete; + VerboseTraceLogger& operator=(const VerboseTraceLogger&) = delete; + + void initialize(bool enabled, + const std::string& file_path, + int32_t max_size_mb, + int32_t max_files); + + bool enabled() const { return enabled_.load(std::memory_order_relaxed); } + + void log(const std::string& entry); + + void shutdown(); + + private: + VerboseTraceLogger() = default; + ~VerboseTraceLogger(); + + std::atomic enabled_{false}; + std::shared_ptr thread_pool_; + std::shared_ptr logger_; +}; + +namespace detail { + +// Lightweight message builder that accumulates into a thread-local pre- +// allocated string buffer. Eliminates std::ostringstream overhead (no locale, +// no virtual dispatch, no per-call heap allocation after the first invocation +// on each thread). On destruction the buffer content is handed to spdlog's +// async queue (one memcpy into the ring buffer). +class VerboseTraceMessage final { + public: + VerboseTraceMessage() { buf().clear(); } + + VerboseTraceMessage& operator<<(std::string_view val) { + buf().append(val); + return *this; + } + + VerboseTraceMessage& operator<<(int32_t val) { + char tmp[12]; + auto [ptr, ec] = std::to_chars(tmp, tmp + sizeof(tmp), val); + buf().append(tmp, static_cast(ptr - tmp)); + return *this; + } + + VerboseTraceMessage& operator<<(int64_t val) { + char tmp[24]; + auto [ptr, ec] = std::to_chars(tmp, tmp + sizeof(tmp), val); + buf().append(tmp, static_cast(ptr - tmp)); + return *this; + } + + VerboseTraceMessage& operator<<(uint64_t val) { + char tmp[24]; + auto [ptr, ec] = std::to_chars(tmp, tmp + sizeof(tmp), val); + buf().append(tmp, static_cast(ptr - tmp)); + return *this; + } + + ~VerboseTraceMessage() { VerboseTraceLogger::get_instance().log(buf()); } + + private: + static constexpr size_t kDefaultCapacity = 512; + + static std::string& buf() { + thread_local std::string tl_buf = [] { + std::string s; + s.reserve(kDefaultCapacity); + return s; + }(); + return tl_buf; + } +}; + +class VerboseTraceVoidify final { + public: + void operator&(const VerboseTraceMessage&) {} +}; + +} // namespace detail + +#define XLLM_VERBOSE_TRACE() \ + !::xllm::VerboseTraceLogger::get_instance().enabled() \ + ? (void)0 \ + : ::xllm::detail::VerboseTraceVoidify() & \ + ::xllm::detail::VerboseTraceMessage() + +} // namespace xllm diff --git a/xllm/xllm.cpp b/xllm/xllm.cpp index 24bd4937a5..ae67100b7e 100644 --- a/xllm/xllm.cpp +++ b/xllm/xllm.cpp @@ -57,6 +57,7 @@ limitations under the License. #include "core/platform/device_name_utils.h" #include "core/util/net.h" #include "core/util/utils.h" +#include "core/util/verbose_trace_logger.h" #include "function_call/function_call_parser.h" #include "parser/reasoning_parser.h" #include "server/xllm_server_registry.h" @@ -503,6 +504,19 @@ int main(int argc, char** argv) { google::InitGoogleLogging("xllm"); initialize_configs(); + const ServiceConfig& service_config = ServiceConfig::get_instance(); + const DistributedConfig& distributed_config = + DistributedConfig::get_instance(); + const std::string verbose_trace_log_path = + resolve_verbose_trace_log_path(service_config.verbose_trace_log_path(), + distributed_config.nnodes(), + distributed_config.node_rank()); + VerboseTraceLogger::get_instance().initialize( + service_config.enable_verbose_trace_log(), + verbose_trace_log_path, + service_config.verbose_trace_log_max_size_mb(), + service_config.verbose_trace_log_max_files()); + // Check if model path is provided if (::xllm::ModelConfig::get_instance().model().empty()) { HelpFormatter::print_error("--model flag is required");