Chiebot-Mirror
/
grpc
mirror of https://github.com/grpc/grpc.git
8
0
Fork 0
Code Issues Projects Releases Wiki Activity

client channel: record call completion even if recv_trailing_metadata was not started (#29198)

Browse Source 
* client channel: record call completion even if recv_trailing_metadata was not started

* add test

* add test for opencensus filter

* remove unnecessary code
reviewable/pr29296/r1
Mark D. Roth 3 years ago committed by GitHub
parent 8f3cd544cd
commit eded3b6bc3
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
10 changed files with 186 additions and 76 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 48
      src/core/ext/filters/client_channel/client_channel.cc
  2. 2
      src/core/ext/filters/client_channel/client_channel.h
  3. 5
      src/core/lib/channel/call_tracer.h
  4. 28
      src/cpp/ext/filters/census/client_filter.cc
  5. 26
      src/cpp/ext/filters/census/open_census_call_tracer.h
  6. 1
      test/core/util/test_lb_policies.cc
  7. 1
      test/core/util/test_lb_policies.h
  8. 105
      test/cpp/end2end/client_lb_end2end_test.cc
  9. 1
      test/cpp/ext/filters/census/BUILD
  10. 45
      test/cpp/ext/filters/census/stats_plugin_end2end_test.cc

48
src/core/ext/filters/client_channel/client_channel.cc
Unescape View File

@ -2427,6 +2427,7 @@ class ClientChannel::LoadBalancedCall::Metadata
explicit Metadata(grpc_metadata_batch* batch) : batch_(batch) {}
void Add(absl::string_view key, absl::string_view value) override {
if (batch_ == nullptr) return;
// Gross, egregious hack to support legacy grpclb behavior.
// TODO(ctiller): Use a promise context for this once that plumbing is done.
if (key == GrpcLbClientStatsMetadata::key()) {
@ -2447,6 +2448,7 @@ class ClientChannel::LoadBalancedCall::Metadata
std::vector<std::pair<std::string, std::string>> TestOnlyCopyToVector()
override {
if (batch_ == nullptr) return {};
Encoder encoder;
batch_->Encode(&encoder);
return encoder.Take();
@ -2454,6 +2456,7 @@ class ClientChannel::LoadBalancedCall::Metadata
absl::optional<absl::string_view> Lookup(absl::string_view key,
std::string* buffer) const override {
if (batch_ == nullptr) return absl::nullopt;
return batch_->GetStringValue(key, buffer);
}
@ -2524,7 +2527,8 @@ class ClientChannel::LoadBalancedCall::BackendMetricAccessor
: lb_call_(lb_call) {}
const BackendMetricData* GetBackendMetricData() override {
if (lb_call_->backend_metric_data_ == nullptr) {
if (lb_call_->backend_metric_data_ == nullptr &&
lb_call_->recv_trailing_metadata_ != nullptr) {
if (const auto* md = lb_call_->recv_trailing_metadata_->get_pointer(
XEndpointLoadMetricsBinMetadata())) {
lb_call_->backend_metric_data_ =
@ -2594,6 +2598,12 @@ ClientChannel::LoadBalancedCall::~LoadBalancedCall() {
}
void ClientChannel::LoadBalancedCall::Orphan() {
// If the recv_trailing_metadata op was never started, then notify
// about call completion here, as best we can. We assume status
// CANCELLED in this case.
if (recv_trailing_metadata_ == nullptr) {
RecordCallCompletion(absl::CancelledError("call cancelled"));
}
// Compute latency and report it to the tracer.
if (call_attempt_tracer_ != nullptr) {
gpr_timespec latency =
@ -2905,22 +2915,7 @@ void ClientChannel::LoadBalancedCall::RecvTrailingMetadataReady(
status = absl::Status(static_cast<absl::StatusCode>(code), message);
}
}
// If we have a tracer, notify it.
if (self->call_attempt_tracer_ != nullptr) {
self->call_attempt_tracer_->RecordReceivedTrailingMetadata(
status, self->recv_trailing_metadata_,
*self->transport_stream_stats_);
}
// If the LB policy requested a callback for trailing metadata, invoke
// the callback.
if (self->lb_subchannel_call_tracker_ != nullptr) {
Metadata trailing_metadata(self->recv_trailing_metadata_);
BackendMetricAccessor backend_metric_accessor(self);
LoadBalancingPolicy::SubchannelCallTrackerInterface::FinishArgs args = {
status, &trailing_metadata, &backend_metric_accessor};
self->lb_subchannel_call_tracker_->Finish(args);
self->lb_subchannel_call_tracker_.reset();
}
self->RecordCallCompletion(status);
}
// Chain to original callback.
if (self->failure_error_ != GRPC_ERROR_NONE) {
@ -2933,6 +2928,25 @@ void ClientChannel::LoadBalancedCall::RecvTrailingMetadataReady(
error);
}
void ClientChannel::LoadBalancedCall::RecordCallCompletion(
absl::Status status) {
// If we have a tracer, notify it.
if (call_attempt_tracer_ != nullptr) {
call_attempt_tracer_->RecordReceivedTrailingMetadata(
status, recv_trailing_metadata_, transport_stream_stats_);
}
// If the LB policy requested a callback for trailing metadata, invoke
// the callback.
if (lb_subchannel_call_tracker_ != nullptr) {
Metadata trailing_metadata(recv_trailing_metadata_);
BackendMetricAccessor backend_metric_accessor(this);
LoadBalancingPolicy::SubchannelCallTrackerInterface::FinishArgs args = {
status, &trailing_metadata, &backend_metric_accessor};
lb_subchannel_call_tracker_->Finish(args);
lb_subchannel_call_tracker_.reset();
}
}
void ClientChannel::LoadBalancedCall::CreateSubchannelCall() {
SubchannelCall::Args call_args = {
std::move(connected_subchannel_), pollent_, path_.Ref(), /*start_time=*/0,

2
src/core/ext/filters/client_channel/client_channel.h
Unescape View File

@ -433,6 +433,8 @@ class ClientChannel::LoadBalancedCall
static void RecvMessageReady(void* arg, grpc_error_handle error);
static void RecvTrailingMetadataReady(void* arg, grpc_error_handle error);
void RecordCallCompletion(absl::Status status);
void CreateSubchannelCall();
// Invoked when a pick is completed, on both success or failure.
static void PickDone(void* arg, grpc_error_handle error);

5
src/core/lib/channel/call_tracer.h
Unescape View File

@ -59,9 +59,12 @@ class CallTracer {
virtual void RecordReceivedInitialMetadata(
grpc_metadata_batch* recv_initial_metadata, uint32_t flags) = 0;
virtual void RecordReceivedMessage(const ByteStream& recv_message) = 0;
// If the call was cancelled before the recv_trailing_metadata op
// was started, recv_trailing_metadata and transport_stream_stats
// will be null.
virtual void RecordReceivedTrailingMetadata(
absl::Status status, grpc_metadata_batch* recv_trailing_metadata,
const grpc_transport_stream_stats& transport_stream_stats) = 0;
const grpc_transport_stream_stats* transport_stream_stats) = 0;
virtual void RecordCancel(grpc_error_handle cancel_error) = 0;
// Should be the last API call to the object. Once invoked, the tracer
// library is free to destroy the object.

28
src/cpp/ext/filters/census/client_filter.cc
Unescape View File

@ -95,7 +95,7 @@ OpenCensusCallTracer::OpenCensusCallAttemptTracer::OpenCensusCallAttemptTracer(
void OpenCensusCallTracer::OpenCensusCallAttemptTracer::
RecordSendInitialMetadata(grpc_metadata_batch* send_initial_metadata,
uint32_t /* flags */) {
uint32_t /*flags*/) {
char tracing_buf[kMaxTraceContextLen];
size_t tracing_len = TraceContextSerialize(context_.Context(), tracing_buf,
kMaxTraceContextLen);
@ -114,12 +114,12 @@ void OpenCensusCallTracer::OpenCensusCallAttemptTracer::
}
void OpenCensusCallTracer::OpenCensusCallAttemptTracer::RecordSendMessage(
const grpc_core::ByteStream& /* send_message */) {
const grpc_core::ByteStream& /*send_message*/) {
++sent_message_count_;
}
void OpenCensusCallTracer::OpenCensusCallAttemptTracer::RecordReceivedMessage(
const grpc_core::ByteStream& /* recv_message */) {
const grpc_core::ByteStream& /*recv_message*/) {
++recv_message_count_;
}
@ -140,21 +140,25 @@ void FilterTrailingMetadata(grpc_metadata_batch* b, uint64_t* elapsed_time) {
void OpenCensusCallTracer::OpenCensusCallAttemptTracer::
RecordReceivedTrailingMetadata(
absl::Status status, grpc_metadata_batch* recv_trailing_metadata,
const grpc_transport_stream_stats& transport_stream_stats) {
FilterTrailingMetadata(recv_trailing_metadata, &elapsed_time_);
const uint64_t request_size = transport_stream_stats.outgoing.data_bytes;
const uint64_t response_size = transport_stream_stats.incoming.data_bytes;
const grpc_transport_stream_stats* transport_stream_stats) {
status_code_ = status.code();
if (recv_trailing_metadata == nullptr || transport_stream_stats == nullptr) {
return;
}
uint64_t elapsed_time = 0;
FilterTrailingMetadata(recv_trailing_metadata, &elapsed_time);
std::vector<std::pair<opencensus::tags::TagKey, std::string>> tags =
context_.tags().tags();
tags.emplace_back(ClientMethodTagKey(), std::string(parent_->method_));
status_code_ = status.code();
std::string final_status = absl::StatusCodeToString(status_code_);
tags.emplace_back(ClientStatusTagKey(), final_status);
::opencensus::stats::Record(
{{RpcClientSentBytesPerRpc(), static_cast<double>(request_size)},
{RpcClientReceivedBytesPerRpc(), static_cast<double>(response_size)},
{{RpcClientSentBytesPerRpc(),
static_cast<double>(transport_stream_stats->outgoing.data_bytes)},
{RpcClientReceivedBytesPerRpc(),
static_cast<double>(transport_stream_stats->incoming.data_bytes)},
{RpcClientServerLatency(),
ToDoubleMilliseconds(absl::Nanoseconds(elapsed_time_))}},
ToDoubleMilliseconds(absl::Nanoseconds(elapsed_time))}},
tags);
}
@ -165,7 +169,7 @@ void OpenCensusCallTracer::OpenCensusCallAttemptTracer::RecordCancel(
}
void OpenCensusCallTracer::OpenCensusCallAttemptTracer::RecordEnd(
const gpr_timespec& /* latency */) {
const gpr_timespec& /*latency*/) {
double latency_ms = absl::ToDoubleMilliseconds(absl::Now() - start_time_);
std::vector<std::pair<opencensus::tags::TagKey, std::string>> tags =
context_.tags().tags();

26
src/cpp/ext/filters/census/open_census_call_tracer.h
Unescape View File

@ -33,25 +33,23 @@ class OpenCensusCallTracer : public grpc_core::CallTracer {
OpenCensusCallAttemptTracer(OpenCensusCallTracer* parent,
uint64_t attempt_num, bool is_transparent_retry,
bool arena_allocated);
void RecordSendInitialMetadata(
grpc_metadata_batch* /* send_initial_metadata */,
uint32_t /* flags */) override;
void RecordOnDoneSendInitialMetadata(gpr_atm* /* peer_string */) override {}
void RecordSendInitialMetadata(grpc_metadata_batch* send_initial_metadata,
uint32_t /*flags*/) override;
void RecordOnDoneSendInitialMetadata(gpr_atm* /*peer_string*/) override {}
void RecordSendTrailingMetadata(
grpc_metadata_batch* /* send_trailing_metadata */) override {}
grpc_metadata_batch* /*send_trailing_metadata*/) override {}
void RecordSendMessage(
const grpc_core::ByteStream& /* send_message */) override;
const grpc_core::ByteStream& /*send_message*/) override;
void RecordReceivedInitialMetadata(
grpc_metadata_batch* /* recv_initial_metadata */,
uint32_t /* flags */) override {}
grpc_metadata_batch* /*recv_initial_metadata*/,
uint32_t /*flags*/) override {}
void RecordReceivedMessage(
const grpc_core::ByteStream& /* recv_message */) override;
const grpc_core::ByteStream& /*recv_message*/) override;
void RecordReceivedTrailingMetadata(
absl::Status /* status */, grpc_metadata_batch* recv_trailing_metadata,
const grpc_transport_stream_stats& /* transport_stream_stats */)
override;
absl::Status status, grpc_metadata_batch* recv_trailing_metadata,
const grpc_transport_stream_stats* transport_stream_stats) override;
void RecordCancel(grpc_error_handle cancel_error) override;
void RecordEnd(const gpr_timespec& /* latency */) override;
void RecordEnd(const gpr_timespec& /*latency*/) override;
CensusContext* context() { return &context_; }
@ -65,8 +63,6 @@ class OpenCensusCallTracer : public grpc_core::CallTracer {
CensusContext context_;
// Start time (for measuring latency).
absl::Time start_time_;
// Server elapsed time in nanoseconds.
uint64_t elapsed_time_ = 0;
// Number of messages in this RPC.
uint64_t recv_message_count_ = 0;
uint64_t sent_message_count_ = 0;

1
test/core/util/test_lb_policies.cc
Unescape View File

@ -281,6 +281,7 @@ class InterceptRecvTrailingMetadataLoadBalancingPolicy
void Finish(FinishArgs args) override {
TrailingMetadataArgsSeen args_seen;
args_seen.status = args.status;
args_seen.backend_metric_data =
args.backend_metric_accessor->GetBackendMetricData();
args_seen.metadata = args.trailing_metadata->TestOnlyCopyToVector();

1
test/core/util/test_lb_policies.h
Unescape View File

@ -36,6 +36,7 @@ void RegisterTestPickArgsLoadBalancingPolicy(
TestPickArgsCallback cb, const char* delegate_policy_name = "pick_first");
struct TrailingMetadataArgsSeen {
absl::Status status;
const LoadBalancingPolicy::BackendMetricAccessor::BackendMetricData*
backend_metric_data;
MetadataVector metadata;

105
test/cpp/end2end/client_lb_end2end_test.cc
Unescape View File

@ -78,6 +78,8 @@ namespace grpc {
namespace testing {
namespace {
constexpr char kRequestMessage[] = "Live long and prosper.";
gpr_atm g_connection_delay_ms;
void tcp_client_connect_with_delay(grpc_closure* closure, grpc_endpoint** ep,
@ -232,7 +234,6 @@ class ClientLbEnd2endTest : public ::testing::Test {
protected:
ClientLbEnd2endTest()
: server_host_("localhost"),
kRequestMessage_("Live long and prosper."),
creds_(new SecureChannelCredentials(
grpc_fake_transport_security_credentials_create())) {}
@ -316,21 +317,22 @@ class ClientLbEnd2endTest : public ::testing::Test {
bool SendRpc(
const std::unique_ptr<grpc::testing::EchoTestService::Stub>& stub,
EchoResponse* response = nullptr, int timeout_ms = 1000,
Status* result = nullptr, bool wait_for_ready = false) {
const bool local_response = (response == nullptr);
if (local_response) response = new EchoResponse;
EchoRequest request;
request.set_message(kRequestMessage_);
request.mutable_param()->set_echo_metadata(true);
Status* result = nullptr, bool wait_for_ready = false,
EchoRequest* request = nullptr) {
EchoResponse local_response;
if (response == nullptr) response = &local_response;
EchoRequest local_request;
if (request == nullptr) request = &local_request;
request->set_message(kRequestMessage);
request->mutable_param()->set_echo_metadata(true);
ClientContext context;
context.set_deadline(grpc_timeout_milliseconds_to_deadline(timeout_ms));
if (wait_for_ready) context.set_wait_for_ready(true);
context.AddMetadata("foo", "1");
context.AddMetadata("bar", "2");
context.AddMetadata("baz", "3");
Status status = stub->Echo(&context, request, response);
Status status = stub->Echo(&context, *request, response);
if (result != nullptr) *result = status;
if (local_response) delete response;
return status.ok();
}
@ -345,7 +347,7 @@ class ClientLbEnd2endTest : public ::testing::Test {
<< "\n"
<< "Error: " << status.error_message() << " "
<< status.error_details();
ASSERT_EQ(response.message(), kRequestMessage_)
ASSERT_EQ(response.message(), kRequestMessage)
<< "From " << location.file() << ":" << location.line();
if (!success) abort();
}
@ -483,7 +485,6 @@ class ClientLbEnd2endTest : public ::testing::Test {
const std::string server_host_;
std::vector<std::unique_ptr<ServerData>> servers_;
const std::string kRequestMessage_;
std::shared_ptr<ChannelCredentials> creds_;
bool ipv6_only_ = false;
};
@ -1832,14 +1833,19 @@ class ClientLbInterceptTrailingMetadataTest : public ClientLbEnd2endTest {
return trailers_intercepted_;
}
const grpc_core::MetadataVector& trailing_metadata() {
absl::Status last_status() {
grpc::internal::MutexLock lock(&mu_);
return last_status_;
}
grpc_core::MetadataVector trailing_metadata() {
grpc::internal::MutexLock lock(&mu_);
return trailing_metadata_;
return std::move(trailing_metadata_);
}
const xds::data::orca::v3::OrcaLoadReport* backend_load_report() {
std::unique_ptr<xds::data::orca::v3::OrcaLoadReport> backend_load_report() {
grpc::internal::MutexLock lock(&mu_);
return load_report_.get();
return std::move(load_report_);
}
private:
@ -1848,6 +1854,7 @@ class ClientLbInterceptTrailingMetadataTest : public ClientLbEnd2endTest {
const auto* backend_metric_data = args_seen.backend_metric_data;
ClientLbInterceptTrailingMetadataTest* self = current_test_instance_;
grpc::internal::MutexLock lock(&self->mu_);
self->last_status_ = args_seen.status;
self->trailers_intercepted_++;
self->trailing_metadata_ = args_seen.metadata;
if (backend_metric_data != nullptr) {
@ -1872,6 +1879,7 @@ class ClientLbInterceptTrailingMetadataTest : public ClientLbEnd2endTest {
static ClientLbInterceptTrailingMetadataTest* current_test_instance_;
grpc::internal::Mutex mu_;
int trailers_intercepted_ = 0;
absl::Status last_status_;
grpc_core::MetadataVector trailing_metadata_;
std::unique_ptr<xds::data::orca::v3::OrcaLoadReport> load_report_;
};
@ -1879,13 +1887,74 @@ class ClientLbInterceptTrailingMetadataTest : public ClientLbEnd2endTest {
ClientLbInterceptTrailingMetadataTest*
ClientLbInterceptTrailingMetadataTest::current_test_instance_ = nullptr;
TEST_F(ClientLbInterceptTrailingMetadataTest, StatusOk) {
StartServers(1);
auto response_generator = BuildResolverResponseGenerator();
auto channel =
BuildChannel("intercept_trailing_metadata_lb", response_generator);
auto stub = BuildStub(channel);
response_generator.SetNextResolution(GetServersPorts());
// Send an OK RPC.
CheckRpcSendOk(stub, DEBUG_LOCATION);
// Check LB policy name for the channel.
EXPECT_EQ("intercept_trailing_metadata_lb",
channel->GetLoadBalancingPolicyName());
EXPECT_EQ(1, trailers_intercepted());
EXPECT_EQ(absl::OkStatus(), last_status());
}
TEST_F(ClientLbInterceptTrailingMetadataTest, StatusFailed) {
StartServers(1);
auto response_generator = BuildResolverResponseGenerator();
auto channel =
BuildChannel("intercept_trailing_metadata_lb", response_generator);
auto stub = BuildStub(channel);
response_generator.SetNextResolution(GetServersPorts());
EchoRequest request;
auto* expected_error = request.mutable_param()->mutable_expected_error();
expected_error->set_code(GRPC_STATUS_PERMISSION_DENIED);
expected_error->set_error_message("bummer, man");
Status status;
SendRpc(stub, /*response=*/nullptr, /*timeout_ms=*/1000, &status,
/*wait_for_ready=*/false, &request);
EXPECT_EQ(status.error_code(), GRPC_STATUS_PERMISSION_DENIED);
EXPECT_EQ(status.error_message(), "bummer, man");
absl::Status status_seen_by_lb = last_status();
EXPECT_EQ(status_seen_by_lb.code(), absl::StatusCode::kPermissionDenied);
EXPECT_EQ(status_seen_by_lb.message(), "bummer, man");
}
TEST_F(ClientLbInterceptTrailingMetadataTest,
StatusCancelledWithoutStartingRecvTrailingMetadata) {
StartServers(1);
auto response_generator = BuildResolverResponseGenerator();
auto channel =
BuildChannel("intercept_trailing_metadata_lb", response_generator);
response_generator.SetNextResolution(GetServersPorts());
auto stub = BuildStub(channel);
{
// Start a stream (sends initial metadata) and then cancel without
// calling Finish().
ClientContext ctx;
auto stream = stub->BidiStream(&ctx);
ctx.TryCancel();
}
// Check status seen by LB policy.
EXPECT_EQ(1, trailers_intercepted());
absl::Status status_seen_by_lb = last_status();
EXPECT_EQ(status_seen_by_lb.code(), absl::StatusCode::kCancelled);
EXPECT_EQ(status_seen_by_lb.message(), "call cancelled");
}
TEST_F(ClientLbInterceptTrailingMetadataTest, InterceptsRetriesDisabled) {
const int kNumServers = 1;
const int kNumRpcs = 10;
StartServers(kNumServers);
auto response_generator = BuildResolverResponseGenerator();
auto channel =
BuildChannel("intercept_trailing_metadata_lb", response_generator);
ChannelArguments channel_args;
channel_args.SetInt(GRPC_ARG_ENABLE_RETRIES, 0);
auto channel = BuildChannel("intercept_trailing_metadata_lb",
response_generator, channel_args);
auto stub = BuildStub(channel);
response_generator.SetNextResolution(GetServersPorts());
for (size_t i = 0; i < kNumRpcs; ++i) {
@ -1971,7 +2040,7 @@ TEST_F(ClientLbInterceptTrailingMetadataTest, BackendMetricData) {
response_generator.SetNextResolution(GetServersPorts());
for (size_t i = 0; i < kNumRpcs; ++i) {
CheckRpcSendOk(stub, DEBUG_LOCATION);
auto* actual = backend_load_report();
auto actual = backend_load_report();
ASSERT_NE(actual, nullptr);
// TODO(roth): Change this to use EqualsProto() once that becomes
// available in OSS.

1
test/cpp/ext/filters/census/BUILD
Unescape View File

@ -37,6 +37,7 @@ grpc_cc_test(
"//:grpc_opencensus_plugin",
"//src/proto/grpc/testing:echo_proto",
"//test/core/util:grpc_test_util",
"//test/cpp/end2end:test_service_impl",
"//test/cpp/util:test_config",
"//test/cpp/util:test_util",
],

45
test/cpp/ext/filters/census/stats_plugin_end2end_test.cc
Unescape View File

@ -37,6 +37,7 @@
#include "src/cpp/ext/filters/census/grpc_plugin.h"
#include "src/proto/grpc/testing/echo.grpc.pb.h"
#include "test/core/util/test_config.h"
#include "test/cpp/end2end/test_service_impl.h"
namespace grpc {
namespace testing {
@ -54,12 +55,25 @@ const auto TEST_TAG_KEY = TagKey::Register("my_key");
const auto TEST_TAG_VALUE = "my_value";
const char* kExpectedTraceIdKey = "expected_trace_id";
class EchoServer final : public EchoTestService::Service {
grpc::Status Echo(grpc::ServerContext* context, const EchoRequest* request,
EchoResponse* response) override {
class EchoServer final : public TestServiceImpl {
Status Echo(ServerContext* context, const EchoRequest* request,
EchoResponse* response) override {
CheckMetadata(context);
return TestServiceImpl::Echo(context, request, response);
}
Status BidiStream(
ServerContext* context,
ServerReaderWriter<EchoResponse, EchoRequest>* stream) override {
CheckMetadata(context);
return TestServiceImpl::BidiStream(context, stream);
}
private:
void CheckMetadata(ServerContext* context) {
for (const auto& metadata : context->client_metadata()) {
if (metadata.first == kExpectedTraceIdKey) {
EXPECT_EQ(metadata.second, reinterpret_cast<const grpc::CensusContext*>(
EXPECT_EQ(metadata.second, reinterpret_cast<const CensusContext*>(
context->census_context())
->Span()
.context()
@ -68,14 +82,6 @@ class EchoServer final : public EchoTestService::Service {
break;
}
}
if (request->param().expected_error().code() == 0) {
response->set_message(request->message());
return grpc::Status::OK;
} else {
return grpc::Status(static_cast<grpc::StatusCode>(
request->param().expected_error().code()),
"");
}
}
};
@ -374,6 +380,20 @@ TEST_F(StatsPluginEnd2EndTest, CompletedRpcs) {
::testing::UnorderedElementsAre(::testing::Pair(
::testing::ElementsAre(server_method_name_, "OK"), i + 1)));
}
// Client should see calls that are cancelled without calling Finish().
{
ClientContext ctx;
auto stream = stub_->BidiStream(&ctx);
ctx.TryCancel();
}
absl::SleepFor(absl::Milliseconds(500));
TestUtils::Flush();
EXPECT_THAT(client_completed_rpcs_view.GetData().int_data(),
::testing::Contains(::testing::Pair(
::testing::ElementsAre(
"grpc.testing.EchoTestService/BidiStream", "CANCELLED"),
1)));
}
TEST_F(StatsPluginEnd2EndTest, RequestReceivedMessagesPerRpc) {
@ -472,7 +492,6 @@ TEST_F(StatsPluginEnd2EndTest, TestRetryStatsWithAdditionalRetries) {
ClientTransparentRetriesCumulative());
View client_retry_delay_per_call_view(ClientRetryDelayPerCallCumulative());
ChannelArguments args;
args.SetInt(GRPC_ARG_ENABLE_RETRIES, 1);
args.SetString(GRPC_ARG_SERVICE_CONFIG,
"{\n"
" \"methodConfig\": [ {\n"

Write Preview
Loading…
Cancel
Save