[Logging] Fix flakiness in test (#33077)

Change - just make sure that before we verify the logging entries, we'll
wait for the expected number of entries to show up.

Logging_test has been recently reported as flaky. Sample failure -
https://source.cloud.google.com/results/invocations/ba581ad8-b652-4b9d-af56-07593f5d2deb/targets/%2F%2Ftest%2Fcpp%2Fext%2Ffilters%2Flogging:logging_test@poller%3Depoll1/tests

Failed to reproduce, but i have a suspicion that what's happening is
that the logging for the half-close events on the server side can happen
after the call has ended.. It is currently being logged after the
server-trailer (which has the status).

test/cpp/ext/filters/logging/library.h

@@ -24,6 +24,7 @@
 
 #include "absl/strings/escaping.h"
 #include "absl/strings/str_cat.h"
+#include "absl/time/time.h"
 #include "gmock/gmock.h"
 #include "google/protobuf/text_format.h"
 #include "gtest/gtest.h"
@@ -72,6 +73,7 @@ class TestLoggingSink : public grpc_core::LoggingSink {
     gpr_log(GPR_INFO, "is_sampled: %d", entry.is_sampled);
     grpc_core::MutexLock lock(&mu_);
     entries_.push_back(std::move(entry));
+    cv_.SignalAll();
   }
 
   void SetConfig(Config config) {
@@ -89,10 +91,20 @@ class TestLoggingSink : public grpc_core::LoggingSink {
     entries_.clear();
   }
 
+  // Waits for \a duration till we have \a num_entries in the log.
+  void WaitForNumEntries(size_t num_entries, absl::Duration duration) {
+    absl::Time deadline = absl::Now() + duration * grpc_test_slowdown_factor();
+    grpc_core::MutexLock lock(&mu_);
+    while (entries_.size() != num_entries) {
+      cv_.WaitWithDeadline(&mu_, deadline);
+    }
+  }
+
  private:
   grpc_core::Mutex mu_;
   std::vector<LoggingSink::Entry> entries_ ABSL_GUARDED_BY(mu_);
   Config config_ ABSL_GUARDED_BY(mu_);
+  grpc_core::CondVar cv_;  // Signalled each time an entry is logged
 };
 
 extern TestLoggingSink* g_test_logging_sink;

test/cpp/ext/filters/logging/logging_test.cc

@@ -60,6 +60,7 @@ TEST_F(LoggingTest, SimpleRpc) {
   context.AddMetadata("client-key", "client-value");
   grpc::Status status = stub_->Echo(&context, request, &response);
   EXPECT_TRUE(status.ok());
+  g_test_logging_sink->WaitForNumEntries(12, absl::Seconds(5));
   EXPECT_THAT(
       g_test_logging_sink->entries(),
       ::testing::UnorderedElementsAre(
@@ -230,6 +231,7 @@ TEST_F(LoggingTest, MetadataTruncated) {
   context.AddMetadata("client-key-2", "client-value-2");
   grpc::Status status = stub_->Echo(&context, request, &response);
   EXPECT_TRUE(status.ok());
+  g_test_logging_sink->WaitForNumEntries(12, absl::Seconds(5));
   EXPECT_THAT(
       g_test_logging_sink->entries(),
       ::testing::UnorderedElementsAre(
@@ -387,6 +389,7 @@ TEST_F(LoggingTest, PayloadTruncated) {
   context.AddMetadata("client-key", "client-value");
   grpc::Status status = stub_->Echo(&context, request, &response);
   EXPECT_TRUE(status.ok());
+  g_test_logging_sink->WaitForNumEntries(12, absl::Seconds(5));
   EXPECT_THAT(
       g_test_logging_sink->entries(),
       ::testing::UnorderedElementsAre(
@@ -593,6 +596,7 @@ TEST_F(LoggingTest, ServerCancelsRpc) {
   EXPECT_EQ(status.error_code(), 25);
   EXPECT_EQ(status.error_message(), "error message");
   EXPECT_EQ(status.error_details(), "binary error details");
+  g_test_logging_sink->WaitForNumEntries(10, absl::Seconds(5));
   EXPECT_THAT(
       g_test_logging_sink->entries(),
       ::testing::UnorderedElementsAre(