Merge pull request #4631 from sreecha/stress_test_logging

Add ability to control the verbosity of stress test logs

test/cpp/interop/interop_client.cc

@@ -38,19 +38,19 @@
 #include <fstream>
 #include <memory>
 
+#include <grpc++/channel.h>
+#include <grpc++/client_context.h>
+#include <grpc++/security/credentials.h>
 #include <grpc/grpc.h>
 #include <grpc/support/log.h>
 #include <grpc/support/string_util.h>
 #include <grpc/support/useful.h>
-#include <grpc++/channel.h>
-#include <grpc++/client_context.h>
-#include <grpc++/security/credentials.h>
 
 #include "src/core/transport/byte_stream.h"
 #include "test/cpp/interop/client_helper.h"
-#include "test/proto/test.grpc.pb.h"
 #include "test/proto/empty.grpc.pb.h"
 #include "test/proto/messages.grpc.pb.h"
+#include "test/proto/test.grpc.pb.h"
 
 namespace grpc {
 namespace testing {
@@ -145,13 +145,13 @@ void InteropClient::AssertOkOrPrintErrorStatus(const Status& s) {
   if (s.ok()) {
     return;
   }
-  gpr_log(GPR_INFO, "Error status code: %d, message: %s", s.error_code(),
+  gpr_log(GPR_ERROR, "Error status code: %d, message: %s", s.error_code(),
           s.error_message().c_str());
   GPR_ASSERT(0);
 }
 
 void InteropClient::DoEmpty() {
-  gpr_log(GPR_INFO, "Sending an empty rpc...");
+  gpr_log(GPR_DEBUG, "Sending an empty rpc...");
 
   Empty request = Empty::default_instance();
   Empty response = Empty::default_instance();
@@ -160,7 +160,7 @@ void InteropClient::DoEmpty() {
   Status s = serviceStub_.Get()->EmptyCall(&context, request, &response);
   AssertOkOrPrintErrorStatus(s);
 
-  gpr_log(GPR_INFO, "Empty rpc done.");
+  gpr_log(GPR_DEBUG, "Empty rpc done.");
 }
 
 void InteropClient::PerformLargeUnary(SimpleRequest* request,
@@ -208,7 +208,7 @@ void InteropClient::PerformLargeUnary(SimpleRequest* request,
 void InteropClient::DoComputeEngineCreds(
     const grpc::string& default_service_account,
     const grpc::string& oauth_scope) {
-  gpr_log(GPR_INFO,
+  gpr_log(GPR_DEBUG,
           "Sending a large unary rpc with compute engine credentials ...");
   SimpleRequest request;
   SimpleResponse response;
@@ -216,19 +216,19 @@ void InteropClient::DoComputeEngineCreds(
   request.set_fill_oauth_scope(true);
   request.set_response_type(PayloadType::COMPRESSABLE);
   PerformLargeUnary(&request, &response);
-  gpr_log(GPR_INFO, "Got username %s", response.username().c_str());
-  gpr_log(GPR_INFO, "Got oauth_scope %s", response.oauth_scope().c_str());
+  gpr_log(GPR_DEBUG, "Got username %s", response.username().c_str());
+  gpr_log(GPR_DEBUG, "Got oauth_scope %s", response.oauth_scope().c_str());
   GPR_ASSERT(!response.username().empty());
   GPR_ASSERT(response.username().c_str() == default_service_account);
   GPR_ASSERT(!response.oauth_scope().empty());
   const char* oauth_scope_str = response.oauth_scope().c_str();
   GPR_ASSERT(oauth_scope.find(oauth_scope_str) != grpc::string::npos);
-  gpr_log(GPR_INFO, "Large unary with compute engine creds done.");
+  gpr_log(GPR_DEBUG, "Large unary with compute engine creds done.");
 }
 
 void InteropClient::DoOauth2AuthToken(const grpc::string& username,
                                       const grpc::string& oauth_scope) {
-  gpr_log(GPR_INFO,
+  gpr_log(GPR_DEBUG,
           "Sending a unary rpc with raw oauth2 access token credentials ...");
   SimpleRequest request;
   SimpleResponse response;
@@ -245,11 +245,11 @@ void InteropClient::DoOauth2AuthToken(const grpc::string& username,
   GPR_ASSERT(username == response.username());
   const char* oauth_scope_str = response.oauth_scope().c_str();
   GPR_ASSERT(oauth_scope.find(oauth_scope_str) != grpc::string::npos);
-  gpr_log(GPR_INFO, "Unary with oauth2 access token credentials done.");
+  gpr_log(GPR_DEBUG, "Unary with oauth2 access token credentials done.");
 }
 
 void InteropClient::DoPerRpcCreds(const grpc::string& json_key) {
-  gpr_log(GPR_INFO, "Sending a unary rpc with per-rpc JWT access token ...");
+  gpr_log(GPR_DEBUG, "Sending a unary rpc with per-rpc JWT access token ...");
   SimpleRequest request;
   SimpleResponse response;
   request.set_fill_username(true);
@@ -266,11 +266,12 @@ void InteropClient::DoPerRpcCreds(const grpc::string& json_key) {
   AssertOkOrPrintErrorStatus(s);
   GPR_ASSERT(!response.username().empty());
   GPR_ASSERT(json_key.find(response.username()) != grpc::string::npos);
-  gpr_log(GPR_INFO, "Unary with per-rpc JWT access token done.");
+  gpr_log(GPR_DEBUG, "Unary with per-rpc JWT access token done.");
 }
 
 void InteropClient::DoJwtTokenCreds(const grpc::string& username) {
-  gpr_log(GPR_INFO, "Sending a large unary rpc with JWT token credentials ...");
+  gpr_log(GPR_DEBUG,
+          "Sending a large unary rpc with JWT token credentials ...");
   SimpleRequest request;
   SimpleResponse response;
   request.set_fill_username(true);
@@ -278,16 +279,16 @@ void InteropClient::DoJwtTokenCreds(const grpc::string& username) {
   PerformLargeUnary(&request, &response);
   GPR_ASSERT(!response.username().empty());
   GPR_ASSERT(username.find(response.username()) != grpc::string::npos);
-  gpr_log(GPR_INFO, "Large unary with JWT token creds done.");
+  gpr_log(GPR_DEBUG, "Large unary with JWT token creds done.");
 }
 
 void InteropClient::DoLargeUnary() {
-  gpr_log(GPR_INFO, "Sending a large unary rpc...");
+  gpr_log(GPR_DEBUG, "Sending a large unary rpc...");
   SimpleRequest request;
   SimpleResponse response;
   request.set_response_type(PayloadType::COMPRESSABLE);
   PerformLargeUnary(&request, &response);
-  gpr_log(GPR_INFO, "Large unary done.");
+  gpr_log(GPR_DEBUG, "Large unary done.");
 }
 
 void InteropClient::DoLargeCompressedUnary() {
@@ -300,20 +301,21 @@ void InteropClient::DoLargeCompressedUnary() {
                    CompressionType_Name(compression_types[j]).c_str(),
                    PayloadType_Name(payload_types[i]).c_str());
 
-      gpr_log(GPR_INFO, "Sending a large compressed unary rpc %s.", log_suffix);
+      gpr_log(GPR_DEBUG, "Sending a large compressed unary rpc %s.",
+              log_suffix);
       SimpleRequest request;
       SimpleResponse response;
       request.set_response_type(payload_types[i]);
       request.set_response_compression(compression_types[j]);
       PerformLargeUnary(&request, &response, CompressionChecks);
-      gpr_log(GPR_INFO, "Large compressed unary done %s.", log_suffix);
+      gpr_log(GPR_DEBUG, "Large compressed unary done %s.", log_suffix);
       gpr_free(log_suffix);
     }
   }
 }
 
 void InteropClient::DoRequestStreaming() {
-  gpr_log(GPR_INFO, "Sending request steaming rpc ...");
+  gpr_log(GPR_DEBUG, "Sending request steaming rpc ...");
 
   ClientContext context;
   StreamingInputCallRequest request;
@@ -334,11 +336,11 @@ void InteropClient::DoRequestStreaming() {
 
   GPR_ASSERT(response.aggregated_payload_size() == aggregated_payload_size);
   AssertOkOrPrintErrorStatus(s);
-  gpr_log(GPR_INFO, "Request streaming done.");
+  gpr_log(GPR_DEBUG, "Request streaming done.");
 }
 
 void InteropClient::DoResponseStreaming() {
-  gpr_log(GPR_INFO, "Receiving response steaming rpc ...");
+  gpr_log(GPR_DEBUG, "Receiving response steaming rpc ...");
 
   ClientContext context;
   StreamingOutputCallRequest request;
@@ -359,7 +361,7 @@ void InteropClient::DoResponseStreaming() {
   GPR_ASSERT(response_stream_sizes.size() == i);
   Status s = stream->Finish();
   AssertOkOrPrintErrorStatus(s);
-  gpr_log(GPR_INFO, "Response streaming done.");
+  gpr_log(GPR_DEBUG, "Response streaming done.");
 }
 
 void InteropClient::DoResponseCompressedStreaming() {
@@ -376,7 +378,7 @@ void InteropClient::DoResponseCompressedStreaming() {
                    CompressionType_Name(compression_types[j]).c_str(),
                    PayloadType_Name(payload_types[i]).c_str());
 
-      gpr_log(GPR_INFO, "Receiving response steaming rpc %s.", log_suffix);
+      gpr_log(GPR_DEBUG, "Receiving response steaming rpc %s.", log_suffix);
 
       request.set_response_type(payload_types[i]);
       request.set_response_compression(compression_types[j]);
@@ -434,14 +436,14 @@ void InteropClient::DoResponseCompressedStreaming() {
       Status s = stream->Finish();
 
       AssertOkOrPrintErrorStatus(s);
-      gpr_log(GPR_INFO, "Response streaming done %s.", log_suffix);
+      gpr_log(GPR_DEBUG, "Response streaming done %s.", log_suffix);
       gpr_free(log_suffix);
     }
   }
 }
 
 void InteropClient::DoResponseStreamingWithSlowConsumer() {
-  gpr_log(GPR_INFO, "Receiving response steaming rpc with slow consumer ...");
+  gpr_log(GPR_DEBUG, "Receiving response steaming rpc with slow consumer ...");
 
   ClientContext context;
   StreamingOutputCallRequest request;
@@ -458,7 +460,7 @@ void InteropClient::DoResponseStreamingWithSlowConsumer() {
   while (stream->Read(&response)) {
     GPR_ASSERT(response.payload().body() ==
                grpc::string(kResponseMessageSize, '\0'));
-    gpr_log(GPR_INFO, "received message %d", i);
+    gpr_log(GPR_DEBUG, "received message %d", i);
     usleep(kReceiveDelayMilliSeconds * 1000);
     ++i;
   }
@@ -466,11 +468,11 @@ void InteropClient::DoResponseStreamingWithSlowConsumer() {
   Status s = stream->Finish();
 
   AssertOkOrPrintErrorStatus(s);
-  gpr_log(GPR_INFO, "Response streaming done.");
+  gpr_log(GPR_DEBUG, "Response streaming done.");
 }
 
 void InteropClient::DoHalfDuplex() {
-  gpr_log(GPR_INFO, "Sending half-duplex streaming rpc ...");
+  gpr_log(GPR_DEBUG, "Sending half-duplex streaming rpc ...");
 
   ClientContext context;
   std::unique_ptr<ClientReaderWriter<StreamingOutputCallRequest,
@@ -495,11 +497,11 @@ void InteropClient::DoHalfDuplex() {
   GPR_ASSERT(response_stream_sizes.size() == i);
   Status s = stream->Finish();
   AssertOkOrPrintErrorStatus(s);
-  gpr_log(GPR_INFO, "Half-duplex streaming rpc done.");
+  gpr_log(GPR_DEBUG, "Half-duplex streaming rpc done.");
 }
 
 void InteropClient::DoPingPong() {
-  gpr_log(GPR_INFO, "Sending Ping Pong streaming rpc ...");
+  gpr_log(GPR_DEBUG, "Sending Ping Pong streaming rpc ...");
 
   ClientContext context;
   std::unique_ptr<ClientReaderWriter<StreamingOutputCallRequest,
@@ -524,11 +526,11 @@ void InteropClient::DoPingPong() {
   GPR_ASSERT(!stream->Read(&response));
   Status s = stream->Finish();
   AssertOkOrPrintErrorStatus(s);
-  gpr_log(GPR_INFO, "Ping pong streaming done.");
+  gpr_log(GPR_DEBUG, "Ping pong streaming done.");
 }
 
 void InteropClient::DoCancelAfterBegin() {
-  gpr_log(GPR_INFO, "Sending request steaming rpc ...");
+  gpr_log(GPR_DEBUG, "Sending request steaming rpc ...");
 
   ClientContext context;
   StreamingInputCallRequest request;
@@ -537,15 +539,15 @@ void InteropClient::DoCancelAfterBegin() {
   std::unique_ptr<ClientWriter<StreamingInputCallRequest>> stream(
       serviceStub_.Get()->StreamingInputCall(&context, &response));
 
-  gpr_log(GPR_INFO, "Trying to cancel...");
+  gpr_log(GPR_DEBUG, "Trying to cancel...");
   context.TryCancel();
   Status s = stream->Finish();
   GPR_ASSERT(s.error_code() == StatusCode::CANCELLED);
-  gpr_log(GPR_INFO, "Canceling streaming done.");
+  gpr_log(GPR_DEBUG, "Canceling streaming done.");
 }
 
 void InteropClient::DoCancelAfterFirstResponse() {
-  gpr_log(GPR_INFO, "Sending Ping Pong streaming rpc ...");
+  gpr_log(GPR_DEBUG, "Sending Ping Pong streaming rpc ...");
 
   ClientContext context;
   std::unique_ptr<ClientReaderWriter<StreamingOutputCallRequest,
@@ -561,15 +563,16 @@ void InteropClient::DoCancelAfterFirstResponse() {
   GPR_ASSERT(stream->Write(request));
   GPR_ASSERT(stream->Read(&response));
   GPR_ASSERT(response.payload().body() == grpc::string(31415, '\0'));
-  gpr_log(GPR_INFO, "Trying to cancel...");
+  gpr_log(GPR_DEBUG, "Trying to cancel...");
   context.TryCancel();
 
   Status s = stream->Finish();
-  gpr_log(GPR_INFO, "Canceling pingpong streaming done.");
+  gpr_log(GPR_DEBUG, "Canceling pingpong streaming done.");
 }
 
 void InteropClient::DoTimeoutOnSleepingServer() {
-  gpr_log(GPR_INFO, "Sending Ping Pong streaming rpc with a short deadline...");
+  gpr_log(GPR_DEBUG,
+          "Sending Ping Pong streaming rpc with a short deadline...");
 
   ClientContext context;
   std::chrono::system_clock::time_point deadline =
@@ -585,11 +588,11 @@ void InteropClient::DoTimeoutOnSleepingServer() {
 
   Status s = stream->Finish();
   GPR_ASSERT(s.error_code() == StatusCode::DEADLINE_EXCEEDED);
-  gpr_log(GPR_INFO, "Pingpong streaming timeout done.");
+  gpr_log(GPR_DEBUG, "Pingpong streaming timeout done.");
 }
 
 void InteropClient::DoEmptyStream() {
-  gpr_log(GPR_INFO, "Starting empty_stream.");
+  gpr_log(GPR_DEBUG, "Starting empty_stream.");
 
   ClientContext context;
   std::unique_ptr<ClientReaderWriter<StreamingOutputCallRequest,
@@ -600,11 +603,12 @@ void InteropClient::DoEmptyStream() {
   GPR_ASSERT(stream->Read(&response) == false);
   Status s = stream->Finish();
   AssertOkOrPrintErrorStatus(s);
-  gpr_log(GPR_INFO, "empty_stream done.");
+  gpr_log(GPR_DEBUG, "empty_stream done.");
 }
 
 void InteropClient::DoStatusWithMessage() {
-  gpr_log(GPR_INFO, "Sending RPC with a request for status code 2 and message");
+  gpr_log(GPR_DEBUG,
+          "Sending RPC with a request for status code 2 and message");
 
   ClientContext context;
   SimpleRequest request;
@@ -618,7 +622,7 @@ void InteropClient::DoStatusWithMessage() {
 
   GPR_ASSERT(s.error_code() == grpc::StatusCode::UNKNOWN);
   GPR_ASSERT(s.error_message() == test_msg);
-  gpr_log(GPR_INFO, "Done testing Status and Message");
+  gpr_log(GPR_DEBUG, "Done testing Status and Message");
 }
 
 }  // namespace testing

test/cpp/interop/stress_interop_client.cc

@@ -112,7 +112,7 @@ void StressTestInteropClient::MainLoop(std::shared_ptr<Gauge> qps_gauge) {
          gpr_time_cmp(gpr_now(GPR_CLOCK_REALTIME), test_end_time) < 0) {
     // Select the test case to execute based on the weights and execute it
     TestCaseType test_case = test_selector_.GetNextTest();
-    gpr_log(GPR_INFO, "%d - Executing the test case %d", test_id_, test_case);
+    gpr_log(GPR_DEBUG, "%d - Executing the test case %d", test_id_, test_case);
     RunTest(test_case);
 
     num_calls_per_interval++;

test/cpp/interop/stress_test.cc

@@ -50,6 +50,10 @@
 #include "test/proto/metrics.grpc.pb.h"
 #include "test/proto/metrics.pb.h"
 
+extern "C" {
+extern void gpr_default_log(gpr_log_func_args* args);
+}
+
 DEFINE_int32(metrics_port, 8081, "The metrics server port.");
 
 DEFINE_int32(metrics_collection_interval_secs, 5,
@@ -94,6 +98,12 @@ DEFINE_string(test_cases, "",
               " 'large_unary', 10% of the time and 'empty_stream' the remaining"
               " 70% of the time");
 
+DEFINE_int32(log_level, GPR_LOG_SEVERITY_DEBUG,
+             "Severity level of messages that should be logged. Any messages "
+             "greater than or equal to the level set here will be logged. "
+             "The choices are: 0 (GPR_LOG_SEVERITY_DEBUG), 1 "
+             "(GPR_LOG_SEVERITY_INFO) and 2 (GPR_LOG_SEVERITY_ERROR.");
+
 using grpc::testing::kTestCaseList;
 using grpc::testing::MetricsService;
 using grpc::testing::MetricsServiceImpl;
@@ -102,6 +112,16 @@ using grpc::testing::TestCaseType;
 using grpc::testing::UNKNOWN_TEST;
 using grpc::testing::WeightedRandomTestSelector;
 
+static int log_level = GPR_LOG_SEVERITY_DEBUG;
+
+// A simple wrapper to grp_default_log() function. This only logs messages at or
+// above the current log level (set in 'log_level' variable)
+void TestLogFunction(gpr_log_func_args* args) {
+  if (args->severity >= log_level) {
+    gpr_default_log(args);
+  }
+}
+
 TestCaseType GetTestTypeFromName(const grpc::string& test_name) {
   TestCaseType test_case = UNKNOWN_TEST;
 
@@ -190,6 +210,18 @@ void LogParameterInfo(const std::vector<grpc::string>& addresses,
 int main(int argc, char** argv) {
   grpc::testing::InitTest(&argc, &argv, true);
 
+  if (FLAGS_log_level > GPR_LOG_SEVERITY_ERROR ||
+      FLAGS_log_level < GPR_LOG_SEVERITY_DEBUG) {
+    gpr_log(GPR_ERROR, "log_level should be an integer between %d and %d",
+            GPR_LOG_SEVERITY_DEBUG, GPR_LOG_SEVERITY_ERROR);
+    return 1;
+  }
+
+  // Change the default log function to TestLogFunction which respects the
+  // log_level setting.
+  log_level = FLAGS_log_level;
+  gpr_set_log_function(TestLogFunction);
+
   srand(time(NULL));
 
   // Parse the server addresses
@@ -198,7 +230,7 @@ int main(int argc, char** argv) {
 
   // Parse test cases and weights
   if (FLAGS_test_cases.length() == 0) {
-    gpr_log(GPR_INFO, "Not running tests. The 'test_cases' string is empty");
+    gpr_log(GPR_ERROR, "Not running tests. The 'test_cases' string is empty");
     return 1;
   }
 
@@ -270,6 +302,5 @@ int main(int argc, char** argv) {
     it->join();
   }
 
-  metrics_server->Wait();
   return 0;
 }