test/cpp/ext/filters/census/stats_plugin_end2end_test.cc

   1 /*
   2  *
   3  * Copyright 2018 gRPC authors.
   4  *
   5  * Licensed under the Apache License, Version 2.0 (the "License");
   6  * you may not use this file except in compliance with the License.
   7  * You may obtain a copy of the License at
   8  *
   9  *     http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  * Unless required by applicable law or agreed to in writing, software
  12  * distributed under the License is distributed on an "AS IS" BASIS,
  13  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14  * See the License for the specific language governing permissions and
  15  * limitations under the License.
  16  *
  17  */
  18
  19 #include <string>
  20 #include <thread>  // NOLINT
  21 #include <vector>
  22
  23 #include "absl/strings/str_cat.h"
  24 #include "absl/strings/string_view.h"
  25 #include "gmock/gmock.h"
  26 #include "gtest/gtest.h"
  27 #include "include/grpc++/grpc++.h"
  28 #include "include/grpcpp/opencensus.h"
  29 #include "opencensus/stats/stats.h"
  30 #include "opencensus/stats/tag_key.h"
  31 #include "opencensus/stats/testing/test_utils.h"
  32 #include "opencensus/tags/tag_map.h"
  33 #include "opencensus/tags/with_tag_map.h"
  34 #include "src/cpp/ext/filters/census/grpc_plugin.h"
  35 #include "src/proto/grpc/testing/echo.grpc.pb.h"
  36 #include "test/core/util/test_config.h"
  37
  38 namespace grpc {
  39 namespace testing {
  40 namespace {
  41
  42 using ::opencensus::stats::Aggregation;
  43 using ::opencensus::stats::Distribution;
  44 using ::opencensus::stats::View;
  45 using ::opencensus::stats::ViewDescriptor;
  46 using ::opencensus::stats::testing::TestUtils;
  47 using ::opencensus::tags::TagKey;
  48 using ::opencensus::tags::TagMap;
  49 using ::opencensus::tags::WithTagMap;
  50
  51 static const auto TEST_TAG_KEY = TagKey::Register("my_key");
  52 static const auto TEST_TAG_VALUE = "my_value";
  53
  54 class EchoServer final : public EchoTestService::Service {
  55   ::grpc::Status Echo(::grpc::ServerContext* context,
  56                       const EchoRequest* request,
  57                       EchoResponse* response) override {
  58     if (request->param().expected_error().code() == 0) {
  59       response->set_message(request->message());
  60       return ::grpc::Status::OK;
  61     } else {
  62       return ::grpc::Status(static_cast<::grpc::StatusCode>(
  63                                 request->param().expected_error().code()),
  64                             "");
  65     }
  66   }
  67 };
  68
  69 class StatsPluginEnd2EndTest : public ::testing::Test {
  70  protected:
  71   static void SetUpTestCase() { RegisterOpenCensusPlugin(); }
  72
  73   void SetUp() {
  74     // Set up a synchronous server on a different thread to avoid the asynch
  75     // interface.
  76     ::grpc::ServerBuilder builder;
  77     int port;
  78     // Use IPv4 here because it's less flaky than IPv6 ("[::]:0") on Travis.
  79     builder.AddListeningPort("0.0.0.0:0", ::grpc::InsecureServerCredentials(),
  80                              &port);
  81     builder.RegisterService(&service_);
  82     server_ = builder.BuildAndStart();
  83     ASSERT_NE(nullptr, server_);
  84     ASSERT_NE(0, port);
  85     server_address_ = absl::StrCat("0.0.0.0:", port);
  86     server_thread_ = std::thread(&StatsPluginEnd2EndTest::RunServerLoop, this);
  87
  88     stub_ = EchoTestService::NewStub(::grpc::CreateChannel(
  89         server_address_, ::grpc::InsecureChannelCredentials()));
  90   }
  91
  92   void TearDown() {
  93     server_->Shutdown();
  94     server_thread_.join();
  95   }
  96
  97   void RunServerLoop() { server_->Wait(); }
  98
  99   const std::string client_method_name_ = "grpc.testing.EchoTestService/Echo";
 100   const std::string server_method_name_ = "grpc.testing.EchoTestService/Echo";
 101
 102   std::string server_address_;
 103   EchoServer service_;
 104   std::unique_ptr<grpc::Server> server_;
 105   std::thread server_thread_;
 106
 107   std::unique_ptr<EchoTestService::Stub> stub_;
 108 };
 109
 110 TEST_F(StatsPluginEnd2EndTest, ErrorCount) {
 111   const auto client_method_descriptor =
 112       ViewDescriptor()
 113           .set_measure(kRpcClientRoundtripLatencyMeasureName)
 114           .set_name("client_method")
 115           .set_aggregation(Aggregation::Count())
 116           .add_column(ClientMethodTagKey())
 117           .add_column(TEST_TAG_KEY);
 118   View client_method_view(client_method_descriptor);
 119   const auto server_method_descriptor =
 120       ViewDescriptor()
 121           .set_measure(kRpcServerServerLatencyMeasureName)
 122           .set_name("server_method")
 123           .set_aggregation(Aggregation::Count())
 124           .add_column(ServerMethodTagKey());
 125   //.add_column(TEST_TAG_KEY);
 126   View server_method_view(server_method_descriptor);
 127
 128   const auto client_status_descriptor =
 129       ViewDescriptor()
 130           .set_measure(kRpcClientRoundtripLatencyMeasureName)
 131           .set_name("client_status")
 132           .set_aggregation(Aggregation::Count())
 133           .add_column(ClientStatusTagKey())
 134           .add_column(TEST_TAG_KEY);
 135   View client_status_view(client_status_descriptor);
 136   const auto server_status_descriptor =
 137       ViewDescriptor()
 138           .set_measure(kRpcServerServerLatencyMeasureName)
 139           .set_name("server_status")
 140           .set_aggregation(Aggregation::Count())
 141           .add_column(ServerStatusTagKey());
 142   View server_status_view(server_status_descriptor);
 143
 144   // Cover all valid statuses.
 145   for (int i = 0; i <= 16; ++i) {
 146     EchoRequest request;
 147     request.set_message("foo");
 148     request.mutable_param()->mutable_expected_error()->set_code(i);
 149     EchoResponse response;
 150     ::grpc::ClientContext context;
 151     {
 152       WithTagMap tags({{TEST_TAG_KEY, TEST_TAG_VALUE}});
 153       ::grpc::Status status = stub_->Echo(&context, request, &response);
 154     }
 155   }
 156   absl::SleepFor(absl::Milliseconds(500));
 157   TestUtils::Flush();
 158
 159   // Client side views can be tagged with custom tags.
 160   EXPECT_THAT(
 161       client_method_view.GetData().int_data(),
 162       ::testing::UnorderedElementsAre(::testing::Pair(
 163           ::testing::ElementsAre(client_method_name_, TEST_TAG_VALUE), 17)));
 164   // TODO: Implement server view tagging with custom tags.
 165   EXPECT_THAT(server_method_view.GetData().int_data(),
 166               ::testing::UnorderedElementsAre(::testing::Pair(
 167                   ::testing::ElementsAre(server_method_name_), 17)));
 168
 169   // Client side views can be tagged with custom tags.
 170   auto client_tags = {
 171       ::testing::Pair(::testing::ElementsAre("OK", TEST_TAG_VALUE), 1),
 172       ::testing::Pair(::testing::ElementsAre("CANCELLED", TEST_TAG_VALUE), 1),
 173       ::testing::Pair(::testing::ElementsAre("UNKNOWN", TEST_TAG_VALUE), 1),
 174       ::testing::Pair(
 175           ::testing::ElementsAre("INVALID_ARGUMENT", TEST_TAG_VALUE), 1),
 176       ::testing::Pair(
 177           ::testing::ElementsAre("DEADLINE_EXCEEDED", TEST_TAG_VALUE), 1),
 178       ::testing::Pair(::testing::ElementsAre("NOT_FOUND", TEST_TAG_VALUE), 1),
 179       ::testing::Pair(::testing::ElementsAre("ALREADY_EXISTS", TEST_TAG_VALUE),
 180                       1),
 181       ::testing::Pair(
 182           ::testing::ElementsAre("PERMISSION_DENIED", TEST_TAG_VALUE), 1),
 183       ::testing::Pair(::testing::ElementsAre("UNAUTHENTICATED", TEST_TAG_VALUE),
 184                       1),
 185       ::testing::Pair(
 186           ::testing::ElementsAre("RESOURCE_EXHAUSTED", TEST_TAG_VALUE), 1),
 187       ::testing::Pair(
 188           ::testing::ElementsAre("FAILED_PRECONDITION", TEST_TAG_VALUE), 1),
 189       ::testing::Pair(::testing::ElementsAre("ABORTED", TEST_TAG_VALUE), 1),
 190       ::testing::Pair(::testing::ElementsAre("OUT_OF_RANGE", TEST_TAG_VALUE),
 191                       1),
 192       ::testing::Pair(::testing::ElementsAre("UNIMPLEMENTED", TEST_TAG_VALUE),
 193                       1),
 194       ::testing::Pair(::testing::ElementsAre("INTERNAL", TEST_TAG_VALUE), 1),
 195       ::testing::Pair(::testing::ElementsAre("UNAVAILABLE", TEST_TAG_VALUE), 1),
 196       ::testing::Pair(::testing::ElementsAre("DATA_LOSS", TEST_TAG_VALUE), 1),
 197   };
 198
 199   // TODO: Implement server view tagging with custom tags.
 200   auto server_tags = {
 201       ::testing::Pair(::testing::ElementsAre("OK"), 1),
 202       ::testing::Pair(::testing::ElementsAre("CANCELLED"), 1),
 203       ::testing::Pair(::testing::ElementsAre("UNKNOWN"), 1),
 204       ::testing::Pair(::testing::ElementsAre("INVALID_ARGUMENT"), 1),
 205       ::testing::Pair(::testing::ElementsAre("DEADLINE_EXCEEDED"), 1),
 206       ::testing::Pair(::testing::ElementsAre("NOT_FOUND"), 1),
 207       ::testing::Pair(::testing::ElementsAre("ALREADY_EXISTS"), 1),
 208       ::testing::Pair(::testing::ElementsAre("PERMISSION_DENIED"), 1),
 209       ::testing::Pair(::testing::ElementsAre("UNAUTHENTICATED"), 1),
 210       ::testing::Pair(::testing::ElementsAre("RESOURCE_EXHAUSTED"), 1),
 211       ::testing::Pair(::testing::ElementsAre("FAILED_PRECONDITION"), 1),
 212       ::testing::Pair(::testing::ElementsAre("ABORTED"), 1),
 213       ::testing::Pair(::testing::ElementsAre("OUT_OF_RANGE"), 1),
 214       ::testing::Pair(::testing::ElementsAre("UNIMPLEMENTED"), 1),
 215       ::testing::Pair(::testing::ElementsAre("INTERNAL"), 1),
 216       ::testing::Pair(::testing::ElementsAre("UNAVAILABLE"), 1),
 217       ::testing::Pair(::testing::ElementsAre("DATA_LOSS"), 1),
 218   };
 219
 220   EXPECT_THAT(client_status_view.GetData().int_data(),
 221               ::testing::UnorderedElementsAreArray(client_tags));
 222   EXPECT_THAT(server_status_view.GetData().int_data(),
 223               ::testing::UnorderedElementsAreArray(server_tags));
 224 }
 225
 226 TEST_F(StatsPluginEnd2EndTest, RequestReceivedBytesPerRpc) {
 227   View client_sent_bytes_per_rpc_view(ClientSentBytesPerRpcCumulative());
 228   View client_received_bytes_per_rpc_view(
 229       ClientReceivedBytesPerRpcCumulative());
 230   View server_sent_bytes_per_rpc_view(ServerSentBytesPerRpcCumulative());
 231   View server_received_bytes_per_rpc_view(
 232       ServerReceivedBytesPerRpcCumulative());
 233
 234   {
 235     EchoRequest request;
 236     request.set_message("foo");
 237     EchoResponse response;
 238     ::grpc::ClientContext context;
 239     ::grpc::Status status = stub_->Echo(&context, request, &response);
 240     ASSERT_TRUE(status.ok());
 241     EXPECT_EQ("foo", response.message());
 242   }
 243   absl::SleepFor(absl::Milliseconds(500));
 244   TestUtils::Flush();
 245
 246   EXPECT_THAT(client_received_bytes_per_rpc_view.GetData().distribution_data(),
 247               ::testing::UnorderedElementsAre(::testing::Pair(
 248                   ::testing::ElementsAre(client_method_name_),
 249                   ::testing::AllOf(::testing::Property(&Distribution::count, 1),
 250                                    ::testing::Property(&Distribution::mean,
 251                                                        ::testing::Gt(0.0))))));
 252   EXPECT_THAT(client_sent_bytes_per_rpc_view.GetData().distribution_data(),
 253               ::testing::UnorderedElementsAre(::testing::Pair(
 254                   ::testing::ElementsAre(client_method_name_),
 255                   ::testing::AllOf(::testing::Property(&Distribution::count, 1),
 256                                    ::testing::Property(&Distribution::mean,
 257                                                        ::testing::Gt(0.0))))));
 258   EXPECT_THAT(server_received_bytes_per_rpc_view.GetData().distribution_data(),
 259               ::testing::UnorderedElementsAre(::testing::Pair(
 260                   ::testing::ElementsAre(server_method_name_),
 261                   ::testing::AllOf(::testing::Property(&Distribution::count, 1),
 262                                    ::testing::Property(&Distribution::mean,
 263                                                        ::testing::Gt(0.0))))));
 264   EXPECT_THAT(server_sent_bytes_per_rpc_view.GetData().distribution_data(),
 265               ::testing::UnorderedElementsAre(::testing::Pair(
 266                   ::testing::ElementsAre(server_method_name_),
 267                   ::testing::AllOf(::testing::Property(&Distribution::count, 1),
 268                                    ::testing::Property(&Distribution::mean,
 269                                                        ::testing::Gt(0.0))))));
 270 }
 271
 272 TEST_F(StatsPluginEnd2EndTest, Latency) {
 273   View client_latency_view(ClientRoundtripLatencyCumulative());
 274   View client_server_latency_view(ClientServerLatencyCumulative());
 275   View server_server_latency_view(ServerServerLatencyCumulative());
 276
 277   const absl::Time start_time = absl::Now();
 278   {
 279     EchoRequest request;
 280     request.set_message("foo");
 281     EchoResponse response;
 282     ::grpc::ClientContext context;
 283     ::grpc::Status status = stub_->Echo(&context, request, &response);
 284     ASSERT_TRUE(status.ok());
 285     EXPECT_EQ("foo", response.message());
 286   }
 287   // We do not know exact latency/elapsed time, but we know it is less than the
 288   // entire time spent making the RPC.
 289   const double max_time = absl::ToDoubleMilliseconds(absl::Now() - start_time);
 290
 291   absl::SleepFor(absl::Milliseconds(500));
 292   TestUtils::Flush();
 293
 294   EXPECT_THAT(
 295       client_latency_view.GetData().distribution_data(),
 296       ::testing::UnorderedElementsAre(::testing::Pair(
 297           ::testing::ElementsAre(client_method_name_),
 298           ::testing::AllOf(
 299               ::testing::Property(&Distribution::count, 1),
 300               ::testing::Property(&Distribution::mean, ::testing::Gt(0.0)),
 301               ::testing::Property(&Distribution::mean,
 302                                   ::testing::Lt(max_time))))));
 303
 304   // Elapsed time is a subinterval of total latency.
 305   const auto client_latency = client_latency_view.GetData()
 306                                   .distribution_data()
 307                                   .find({client_method_name_})
 308                                   ->second.mean();
 309   EXPECT_THAT(
 310       client_server_latency_view.GetData().distribution_data(),
 311       ::testing::UnorderedElementsAre(::testing::Pair(
 312           ::testing::ElementsAre(client_method_name_),
 313           ::testing::AllOf(
 314               ::testing::Property(&Distribution::count, 1),
 315               ::testing::Property(&Distribution::mean, ::testing::Gt(0.0)),
 316               ::testing::Property(&Distribution::mean,
 317                                   ::testing::Lt(client_latency))))));
 318
 319   // client server elapsed time should be the same value propagated to the
 320   // client.
 321   const auto client_elapsed_time = client_server_latency_view.GetData()
 322                                        .distribution_data()
 323                                        .find({client_method_name_})
 324                                        ->second.mean();
 325   EXPECT_THAT(
 326       server_server_latency_view.GetData().distribution_data(),
 327       ::testing::UnorderedElementsAre(::testing::Pair(
 328           ::testing::ElementsAre(server_method_name_),
 329           ::testing::AllOf(
 330               ::testing::Property(&Distribution::count, 1),
 331               ::testing::Property(&Distribution::mean,
 332                                   ::testing::DoubleEq(client_elapsed_time))))));
 333 }
 334
 335 TEST_F(StatsPluginEnd2EndTest, CompletedRpcs) {
 336   View client_completed_rpcs_view(ClientCompletedRpcsCumulative());
 337   View server_completed_rpcs_view(ServerCompletedRpcsCumulative());
 338
 339   EchoRequest request;
 340   request.set_message("foo");
 341   EchoResponse response;
 342   const int count = 5;
 343   for (int i = 0; i < count; ++i) {
 344     {
 345       ::grpc::ClientContext context;
 346       ::grpc::Status status = stub_->Echo(&context, request, &response);
 347       ASSERT_TRUE(status.ok());
 348       EXPECT_EQ("foo", response.message());
 349     }
 350     absl::SleepFor(absl::Milliseconds(500));
 351     TestUtils::Flush();
 352
 353     EXPECT_THAT(client_completed_rpcs_view.GetData().int_data(),
 354                 ::testing::UnorderedElementsAre(::testing::Pair(
 355                     ::testing::ElementsAre(client_method_name_, "OK"), i + 1)));
 356     EXPECT_THAT(server_completed_rpcs_view.GetData().int_data(),
 357                 ::testing::UnorderedElementsAre(::testing::Pair(
 358                     ::testing::ElementsAre(server_method_name_, "OK"), i + 1)));
 359   }
 360 }
 361
 362 TEST_F(StatsPluginEnd2EndTest, RequestReceivedMessagesPerRpc) {
 363   // TODO: Use streaming RPCs.
 364   View client_received_messages_per_rpc_view(
 365       ClientSentMessagesPerRpcCumulative());
 366   View client_sent_messages_per_rpc_view(
 367       ClientReceivedMessagesPerRpcCumulative());
 368   View server_received_messages_per_rpc_view(
 369       ServerSentMessagesPerRpcCumulative());
 370   View server_sent_messages_per_rpc_view(
 371       ServerReceivedMessagesPerRpcCumulative());
 372
 373   EchoRequest request;
 374   request.set_message("foo");
 375   EchoResponse response;
 376   const int count = 5;
 377   for (int i = 0; i < count; ++i) {
 378     {
 379       ::grpc::ClientContext context;
 380       ::grpc::Status status = stub_->Echo(&context, request, &response);
 381       ASSERT_TRUE(status.ok());
 382       EXPECT_EQ("foo", response.message());
 383     }
 384     absl::SleepFor(absl::Milliseconds(500));
 385     TestUtils::Flush();
 386
 387     EXPECT_THAT(
 388         client_received_messages_per_rpc_view.GetData().distribution_data(),
 389         ::testing::UnorderedElementsAre(::testing::Pair(
 390             ::testing::ElementsAre(client_method_name_),
 391             ::testing::AllOf(::testing::Property(&Distribution::count, i + 1),
 392                              ::testing::Property(&Distribution::mean,
 393                                                  ::testing::DoubleEq(1.0))))));
 394     EXPECT_THAT(
 395         client_sent_messages_per_rpc_view.GetData().distribution_data(),
 396         ::testing::UnorderedElementsAre(::testing::Pair(
 397             ::testing::ElementsAre(client_method_name_),
 398             ::testing::AllOf(::testing::Property(&Distribution::count, i + 1),
 399                              ::testing::Property(&Distribution::mean,
 400                                                  ::testing::DoubleEq(1.0))))));
 401     EXPECT_THAT(
 402         server_received_messages_per_rpc_view.GetData().distribution_data(),
 403         ::testing::UnorderedElementsAre(::testing::Pair(
 404             ::testing::ElementsAre(server_method_name_),
 405             ::testing::AllOf(::testing::Property(&Distribution::count, i + 1),
 406                              ::testing::Property(&Distribution::mean,
 407                                                  ::testing::DoubleEq(1.0))))));
 408     EXPECT_THAT(
 409         server_sent_messages_per_rpc_view.GetData().distribution_data(),
 410         ::testing::UnorderedElementsAre(::testing::Pair(
 411             ::testing::ElementsAre(server_method_name_),
 412             ::testing::AllOf(::testing::Property(&Distribution::count, i + 1),
 413                              ::testing::Property(&Distribution::mean,
 414                                                  ::testing::DoubleEq(1.0))))));
 415   }
 416 }
 417
 418 }  // namespace
 419 }  // namespace testing
 420 }  // namespace grpc
 421
 422 int main(int argc, char** argv) {
 423   grpc::testing::TestEnvironment env(argc, argv);
 424   ::testing::InitGoogleTest(&argc, argv);
 425   return RUN_ALL_TESTS();
 426 }