From 48961e8a2ccc53aa2f72be366b617b9ad2a9af80 Mon Sep 17 00:00:00 2001 From: Stanley Cheung Date: Thu, 25 Jan 2024 16:10:27 -0800 Subject: [PATCH] [PSM Interop] Add a payload to xds interop client when sending RPCs (#35545) When testing CSM Observability, we discovered that the c++ xds interop client is not sending any payload with the `UnaryCall` RPCs so most of the metrics will have a value of 0. Adding a payload to the xds interop client here. We need this fix so that we can verify that the metrics are recording the right number of bytes being sent / received. So we need a non-trivial payload to be sent with the `UnaryCall` RPC between the xds interop client and server. Closes #35545 COPYBARA_INTEGRATE_REVIEW=https://github.com/grpc/grpc/pull/35545 from stanley-cheung:xds-client-payload 11be4e6f4fda8a7d6267ec43b861a82d7fa0c4b8 PiperOrigin-RevId: 601596246 --- test/cpp/interop/xds_interop_client.cc | 59 +++++++++++++++++++++- test/cpp/interop/xds_interop_server_lib.cc | 7 ++- 2 files changed, 63 insertions(+), 3 deletions(-) diff --git a/test/cpp/interop/xds_interop_client.cc b/test/cpp/interop/xds_interop_client.cc index 55084235f3e..5c2eac7aa81 100644 --- a/test/cpp/interop/xds_interop_client.cc +++ b/test/cpp/interop/xds_interop_client.cc @@ -67,6 +67,12 @@ ABSL_FLAG(int32_t, stats_port, 50052, "Port to expose peer distribution stats service."); ABSL_FLAG(std::string, rpc, "UnaryCall", "a comma separated list of rpc methods."); +ABSL_FLAG(int32_t, request_payload_size, 0, + "Set the SimpleRequest.payload.body to a string of repeated 0 (zero) " + "ASCII characters of the given size in bytes."); +ABSL_FLAG(int32_t, response_payload_size, 0, + "Ask the server to respond with SimpleResponse.payload.body of the " + "given length (may not be implemented on the server)."); ABSL_FLAG(std::string, metadata, "", "metadata to send with the RPC."); ABSL_FLAG(std::string, expect_status, "OK", "RPC status for the test RPC to be considered successful"); @@ -117,6 +123,9 @@ struct RpcConfig { ClientConfigureRequest::RpcType type; std::vector> metadata; int timeout_sec = 0; + std::string request_payload; + int request_payload_size = 0; + int response_payload_size = 0; }; struct RpcConfigurationsQueue { // A queue of RPC configurations detailing how RPCs should be sent. @@ -154,11 +163,17 @@ class TestClient { std::chrono::system_clock::now() + std::chrono::seconds(INT_MAX); } } + SimpleRequest request; + request.set_response_size(config.response_payload_size); + if (config.request_payload_size > 0) { + request.mutable_payload()->set_body(config.request_payload.c_str(), + config.request_payload_size); + } call->context.set_deadline(deadline); call->result.saved_request_id = saved_request_id; call->result.rpc_type = ClientConfigureRequest::UNARY_CALL; - call->simple_response_reader = stub_->PrepareAsyncUnaryCall( - &call->context, SimpleRequest::default_instance(), &cq_); + call->simple_response_reader = + stub_->PrepareAsyncUnaryCall(&call->context, request, &cq_); call->simple_response_reader->StartCall(); call->simple_response_reader->Finish(&call->result.simple_response, &call->result.status, call); @@ -324,6 +339,10 @@ class XdsUpdateClientConfigureServiceImpl metadata_map[data.type()].push_back({data.key(), data.value()}); } std::vector configs; + int request_payload_size = absl::GetFlag(FLAGS_request_payload_size); + int response_payload_size = absl::GetFlag(FLAGS_response_payload_size); + GPR_ASSERT(request_payload_size >= 0); + GPR_ASSERT(response_payload_size >= 0); for (const auto& rpc : request->types()) { RpcConfig config; config.timeout_sec = request->timeout_sec(); @@ -332,6 +351,22 @@ class XdsUpdateClientConfigureServiceImpl if (metadata_iter != metadata_map.end()) { config.metadata = metadata_iter->second; } + if (request_payload_size > 0 && + config.type == ClientConfigureRequest::EMPTY_CALL) { + gpr_log(GPR_ERROR, + "request_payload_size should not be set " + "for EMPTY_CALL"); + } + if (response_payload_size > 0 && + config.type == ClientConfigureRequest::EMPTY_CALL) { + gpr_log(GPR_ERROR, + "response_payload_size should not be set " + "for EMPTY_CALL"); + } + config.request_payload_size = request_payload_size; + std::string payload(config.request_payload_size, '0'); + config.request_payload = payload; + config.response_payload_size = response_payload_size; configs.push_back(std::move(config)); } { @@ -459,6 +494,10 @@ void BuildRpcConfigsFromFlags(RpcConfigurationsQueue* rpc_configs_queue) { std::vector configs; std::vector rpc_methods = absl::StrSplit(absl::GetFlag(FLAGS_rpc), ',', absl::SkipEmpty()); + int request_payload_size = absl::GetFlag(FLAGS_request_payload_size); + int response_payload_size = absl::GetFlag(FLAGS_response_payload_size); + GPR_ASSERT(request_payload_size >= 0); + GPR_ASSERT(response_payload_size >= 0); for (const std::string& rpc_method : rpc_methods) { RpcConfig config; if (rpc_method == "EmptyCall") { @@ -472,6 +511,22 @@ void BuildRpcConfigsFromFlags(RpcConfigurationsQueue* rpc_configs_queue) { if (metadata_iter != metadata_map.end()) { config.metadata = metadata_iter->second; } + if (request_payload_size > 0 && + config.type == ClientConfigureRequest::EMPTY_CALL) { + gpr_log(GPR_ERROR, + "request_payload_size should not be set " + "for EMPTY_CALL"); + } + if (response_payload_size > 0 && + config.type == ClientConfigureRequest::EMPTY_CALL) { + gpr_log(GPR_ERROR, + "response_payload_size should not be set " + "for EMPTY_CALL"); + } + config.request_payload_size = request_payload_size; + std::string payload(config.request_payload_size, '0'); + config.request_payload = payload; + config.response_payload_size = response_payload_size; configs.push_back(std::move(config)); } { diff --git a/test/cpp/interop/xds_interop_server_lib.cc b/test/cpp/interop/xds_interop_server_lib.cc index 8508b749f7a..12a26b16497 100644 --- a/test/cpp/interop/xds_interop_server_lib.cc +++ b/test/cpp/interop/xds_interop_server_lib.cc @@ -81,7 +81,7 @@ class TestServiceImpl : public TestService::Service { absl::string_view server_id) : hostname_(hostname), server_id_(server_id) {} - Status UnaryCall(ServerContext* context, const SimpleRequest* /*request*/, + Status UnaryCall(ServerContext* context, const SimpleRequest* request, SimpleResponse* response) override { response->set_server_id(server_id_); for (const auto& rpc_behavior : GetRpcBehaviorMetadata(context)) { @@ -91,6 +91,11 @@ class TestServiceImpl : public TestService::Service { return *maybe_status; } } + if (request->response_size() > 0) { + std::string payload(request->response_size(), '0'); + response->mutable_payload()->set_body(payload.c_str(), + request->response_size()); + } response->set_hostname(hostname_); context->AddInitialMetadata("hostname", hostname_); return Status::OK;