From 02e37e8a0c1b1f82324fce0fd9582a5eb49e49c0 Mon Sep 17 00:00:00 2001 From: Janusz Sobczak Date: Thu, 13 Jul 2023 02:34:20 -0700 Subject: [PATCH] Fix flakiness in analytics `session_was_logged_` could be set out of order if a new session started immediately after terminating the previous. Fixing that. Check the logged messages rather then the voltile analytics recorder state in the tests. PiperOrigin-RevId: 547732773 --- .../analytics/analytics_recorder.cc | 24 +++-- .../analytics/analytics_recorder.h | 9 +- .../implementation/client_proxy_test.cc | 94 ++++++++++++++----- 3 files changed, 95 insertions(+), 32 deletions(-) diff --git a/connections/implementation/analytics/analytics_recorder.cc b/connections/implementation/analytics/analytics_recorder.cc index f11f4e30..60b90a90 100644 --- a/connections/implementation/analytics/analytics_recorder.cc +++ b/connections/implementation/analytics/analytics_recorder.cc @@ -28,6 +28,7 @@ #include "absl/strings/string_view.h" #include "absl/time/time.h" #include "internal/analytics/event_logger.h" +#include "internal/platform/count_down_latch.h" #include "internal/platform/error_code_params.h" #include "internal/platform/implementation/system_clock.h" #include "internal/platform/logging.h" @@ -104,7 +105,8 @@ AnalyticsRecorder::AnalyticsRecorder(EventLogger *event_logger) AnalyticsRecorder::~AnalyticsRecorder() { serial_executor_.Shutdown(); - ResetClientSessionLoggingResouces(); + MutexLock lock(&mutex_); + ResetClientSessionLoggingResoucesLocked(); } bool AnalyticsRecorder::IsSessionLogged() { @@ -112,8 +114,7 @@ bool AnalyticsRecorder::IsSessionLogged() { return session_was_logged_; } -void AnalyticsRecorder::ResetClientSessionLoggingResouces() { - MutexLock lock(&mutex_); +void AnalyticsRecorder::ResetClientSessionLoggingResoucesLocked() { NEARBY_LOGS(INFO) << "Reset AnalyticsRecorder ctor event_logger_=" << event_logger_; @@ -724,7 +725,7 @@ void AnalyticsRecorder::LogSession() { FinishStrategySessionLocked(); client_session_->set_duration_millis(absl::ToInt64Milliseconds( SystemClock::ElapsedRealtime() - started_client_session_time_)); - LogClientSession(); + LogClientSessionLocked(); LogEvent(STOP_CLIENT_SESSION); session_was_logged_ = true; } @@ -769,12 +770,13 @@ bool AnalyticsRecorder::CanRecordAnalyticsLocked( return true; } -void AnalyticsRecorder::LogClientSession() { +void AnalyticsRecorder::LogClientSessionLocked() { serial_executor_.Execute( - "analytics-recorder", [this]() { + "analytics-recorder", + [this, client_session = std::move(client_session_)]() mutable { ConnectionsLog connections_log; connections_log.set_event_type(CLIENT_SESSION); - connections_log.set_allocated_client_session(client_session_.release()); + connections_log.set_allocated_client_session(client_session.release()); connections_log.set_version(kVersion); NEARBY_LOGS(VERBOSE) @@ -782,8 +784,8 @@ void AnalyticsRecorder::LogClientSession() { << connections_log.DebugString(); event_logger_->Log(connections_log); - ResetClientSessionLoggingResouces(); }); + ResetClientSessionLoggingResoucesLocked(); } void AnalyticsRecorder::LogEvent(EventType event_type) { @@ -1302,5 +1304,11 @@ AnalyticsRecorder::LogicalConnection::ResolvePendingPayloads( return completed_payloads; } +void AnalyticsRecorder::Sync() { + CountDownLatch latch(1); + serial_executor_.Execute([&]() { latch.CountDown(); }); + latch.Await(); +} + } // namespace analytics } // namespace nearby diff --git a/connections/implementation/analytics/analytics_recorder.h b/connections/implementation/analytics/analytics_recorder.h index ff425f34..d082e3e8 100644 --- a/connections/implementation/analytics/analytics_recorder.h +++ b/connections/implementation/analytics/analytics_recorder.h @@ -184,6 +184,10 @@ class AnalyticsRecorder { bool IsSessionLogged(); + // Waits until all logs are sent to the backend. + // For testing only. + void Sync(); + private: // Tracks the chunks and duration of a Payload on a particular medium. class PendingPayload { @@ -290,7 +294,7 @@ class AnalyticsRecorder { // Callbacks the ConnectionsLog proto byte array data to the EventLogger with // ClientSession sub-proto. - void LogClientSession(); + void LogClientSessionLocked() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); // Callbacks the ConnectionsLog proto byte array data to the EventLogger. void LogEvent(location::nearby::proto::connections::EventType event_type); @@ -334,7 +338,8 @@ class AnalyticsRecorder { // Reset the client cession's logging resources (e.g. current_strategy_, // current_advertising_phase_, current_discovery_phase_, etc) - void ResetClientSessionLoggingResouces(); + void ResetClientSessionLoggingResoucesLocked() + ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); location::nearby::proto::connections::ConnectionsStrategy StrategyToConnectionStrategy(connections::Strategy strategy); diff --git a/connections/implementation/client_proxy_test.cc b/connections/implementation/client_proxy_test.cc index 5b21988b..011cb8ec 100644 --- a/connections/implementation/client_proxy_test.cc +++ b/connections/implementation/client_proxy_test.cc @@ -19,6 +19,7 @@ #include #include #include +#include #include "gmock/gmock.h" #include "protobuf-matchers/protocol-buffer-matchers.h" @@ -39,12 +40,17 @@ #include "internal/platform/count_down_latch.h" #include "internal/platform/feature_flags.h" #include "internal/platform/medium_environment.h" +#include "proto/connections_enums.proto.h" namespace nearby { namespace connections { namespace { +using ::location::nearby::analytics::proto::ConnectionsLog; using ::location::nearby::connections::OsInfo; +using ::location::nearby::proto::connections::CLIENT_SESSION; +using ::location::nearby::proto::connections::START_CLIENT_SESSION; +using ::location::nearby::proto::connections::STOP_CLIENT_SESSION; using ::testing::MockFunction; using ::testing::StrictMock; @@ -61,7 +67,42 @@ class FakeEventLogger : public ::nearby::analytics::EventLogger { public: explicit FakeEventLogger() = default; - void Log(const ::google::protobuf::MessageLite& message) override {} + void Log(const ::google::protobuf::MessageLite& message) override { + ConnectionsLog log; + log.CheckTypeAndMergeFrom(message); + MutexLock lock(&mutex_); + logs_.push_back(std::move(log)); + } + + int GetCompleteClientSessionCount() { + MutexLock lock(&mutex_); + bool has_start_client_session = false; + bool has_client_session = false; + int session_count = 0; + // We expect series of START_CLIENT_SESSION, CLIENT_SESSION and + // STOP_CLIENT_SESSION events, possibly interleaved with other events. + for (const auto& log : logs_) { + if (log.event_type() == START_CLIENT_SESSION) { + EXPECT_FALSE(has_start_client_session); + EXPECT_FALSE(has_client_session); + has_start_client_session = true; + } else if (log.event_type() == CLIENT_SESSION) { + EXPECT_TRUE(has_start_client_session); + EXPECT_FALSE(has_client_session); + has_client_session = true; + } else if (log.event_type() == STOP_CLIENT_SESSION) { + EXPECT_TRUE(has_start_client_session); + EXPECT_TRUE(has_client_session); + has_start_client_session = false; + has_client_session = false; + ++session_count; + } + } + return session_count; + } + + Mutex mutex_; + std::vector logs_; }; class MockDeviceProvider : public nearby::NearbyDeviceProvider { @@ -802,11 +843,11 @@ TEST_F(ClientProxyTest, NotLogSessionForStoppedAdvertisingWithConnection) { advertising_endpoint.id)); // Connections are available EXPECT_FALSE(client1_.IsDiscovering()); // No Discovery EXPECT_TRUE(client1_.IsAdvertising()); // Advertising - EXPECT_FALSE(client1_.GetAnalyticsRecorder().IsSessionLogged()); // After StopAdvertising(&client1_); // No Advertising - EXPECT_FALSE(client1_.GetAnalyticsRecorder().IsSessionLogged()); + client1_.GetAnalyticsRecorder().Sync(); + EXPECT_EQ(event_logger1_.GetCompleteClientSessionCount(), 0); } TEST_F(ClientProxyTest, @@ -819,11 +860,13 @@ TEST_F(ClientProxyTest, advertising_endpoint.id)); // No Connections EXPECT_FALSE(client1_.IsDiscovering()); // No Discovery EXPECT_TRUE(client1_.IsAdvertising()); // Advertising - EXPECT_FALSE(client1_.GetAnalyticsRecorder().IsSessionLogged()); + client1_.GetAnalyticsRecorder().Sync(); + EXPECT_EQ(event_logger1_.GetCompleteClientSessionCount(), 0); // After StopAdvertising(&client1_); - EXPECT_TRUE(client1_.GetAnalyticsRecorder().IsSessionLogged()); + client1_.GetAnalyticsRecorder().Sync(); + EXPECT_GT(event_logger1_.GetCompleteClientSessionCount(), 0); } TEST_F(ClientProxyTest, NotLogSessionForStoppedDiscoveryWithConnection) { @@ -838,11 +881,11 @@ TEST_F(ClientProxyTest, NotLogSessionForStoppedDiscoveryWithConnection) { &client2_, advertising_endpoint); // Connections are available EXPECT_FALSE(client2_.IsAdvertising()); // No Advertising EXPECT_TRUE(client2_.IsDiscovering()); // Discovering - EXPECT_FALSE(client2_.GetAnalyticsRecorder().IsSessionLogged()); // After StopDiscovery(&client2_); - EXPECT_FALSE(client2_.GetAnalyticsRecorder().IsSessionLogged()); + client2_.GetAnalyticsRecorder().Sync(); + EXPECT_EQ(event_logger2_.GetCompleteClientSessionCount(), 0); } TEST_F(ClientProxyTest, @@ -860,7 +903,8 @@ TEST_F(ClientProxyTest, // After StopDiscovery(&client2_); - EXPECT_TRUE(client2_.GetAnalyticsRecorder().IsSessionLogged()); + client2_.GetAnalyticsRecorder().Sync(); + EXPECT_GT(event_logger2_.GetCompleteClientSessionCount(), 0); } TEST_F(ClientProxyTest, LogSessionOnDisconnectedWithOneConnection) { @@ -878,7 +922,8 @@ TEST_F(ClientProxyTest, LogSessionOnDisconnectedWithOneConnection) { // After OnDiscoveryConnectionDisconnected(&client2_, advertising_endpoint); - EXPECT_TRUE(client2_.GetAnalyticsRecorder().IsSessionLogged()); + client2_.GetAnalyticsRecorder().Sync(); + EXPECT_GT(event_logger2_.GetCompleteClientSessionCount(), 0); } TEST_F(ClientProxyTest, @@ -894,7 +939,8 @@ TEST_F(ClientProxyTest, // After client2_.OnDisconnected(advertising_endpoint.id, /*notify=*/false); - EXPECT_FALSE(client2_.GetAnalyticsRecorder().IsSessionLogged()); + client2_.GetAnalyticsRecorder().Sync(); + EXPECT_EQ(event_logger2_.GetCompleteClientSessionCount(), 0); } TEST_F(ClientProxyTest, NotLogSessionOnDisconnectedWhenMoreThanOneConnection) { @@ -921,7 +967,8 @@ TEST_F(ClientProxyTest, NotLogSessionOnDisconnectedWhenMoreThanOneConnection) { // After client2_.OnDisconnected(advertising_endpoint_1.id, /*notify=*/false); - EXPECT_FALSE(client2_.GetAnalyticsRecorder().IsSessionLogged()); + client2_.GetAnalyticsRecorder().Sync(); + EXPECT_EQ(event_logger2_.GetCompleteClientSessionCount(), 0); } TEST_F(ClientProxyTest, @@ -940,7 +987,8 @@ TEST_F(ClientProxyTest, // After OnDiscoveryConnectionDisconnected(&client2_, advertising_endpoint); - EXPECT_FALSE(client2_.GetAnalyticsRecorder().IsSessionLogged()); + client2_.GetAnalyticsRecorder().Sync(); + EXPECT_EQ(event_logger2_.GetCompleteClientSessionCount(), 0); } TEST_F(ClientProxyTest, LogSessionForResetClientProxy) { @@ -950,13 +998,18 @@ TEST_F(ClientProxyTest, LogSessionForResetClientProxy) { OnDiscoveryEndpointFound(&client2_, advertising_endpoint); OnDiscoveryConnectionInitiated(&client2_, advertising_endpoint); - EXPECT_FALSE(client1_.GetAnalyticsRecorder().IsSessionLogged()); + client1_.GetAnalyticsRecorder().Sync(); + EXPECT_EQ(event_logger1_.GetCompleteClientSessionCount(), 0); client1_.Reset(); - EXPECT_TRUE(client1_.GetAnalyticsRecorder().IsSessionLogged()); + client1_.GetAnalyticsRecorder().Sync(); + // TODO(b/290936886): Why are there more than one complete sessions? + EXPECT_GT(event_logger1_.GetCompleteClientSessionCount(), 0); - EXPECT_FALSE(client2_.GetAnalyticsRecorder().IsSessionLogged()); + client2_.GetAnalyticsRecorder().Sync(); + EXPECT_EQ(event_logger2_.GetCompleteClientSessionCount(), 0); client2_.Reset(); - EXPECT_TRUE(client2_.GetAnalyticsRecorder().IsSessionLogged()); + client2_.GetAnalyticsRecorder().Sync(); + EXPECT_GT(event_logger2_.GetCompleteClientSessionCount(), 0); } TEST_F(ClientProxyTest, GetLocalInfoCorrect) { @@ -1092,8 +1145,7 @@ TEST_F(ClientProxyTest, EnforceTopologyWhenRequestedAdvertising) { TEST_F(ClientProxyTest, EnforceTopologyWhenRequestedListeningWithStrategy) { EXPECT_FALSE(client1_.ShouldEnforceTopologyConstraints()); - StartListeningForIncomingConnections(&client1_, - {}, + StartListeningForIncomingConnections(&client1_, {}, {.strategy = Strategy::kP2pCluster, .enforce_topology_constraints = true}); EXPECT_TRUE(client1_.ShouldEnforceTopologyConstraints()); @@ -1101,8 +1153,7 @@ TEST_F(ClientProxyTest, EnforceTopologyWhenRequestedListeningWithStrategy) { TEST_F(ClientProxyTest, DontEnforceTopologyWhenRequestedWithNoStrategy) { EXPECT_FALSE(client1_.ShouldEnforceTopologyConstraints()); - StartListeningForIncomingConnections(&client1_, - {}, + StartListeningForIncomingConnections(&client1_, {}, {.strategy = Strategy::kNone}); EXPECT_TRUE(client1_.ShouldEnforceTopologyConstraints()); } @@ -1116,8 +1167,7 @@ TEST_F(ClientProxyTest, TestAutoBwuWhenAdvertisingWithAutoBwu) { TEST_F(ClientProxyTest, TestAutoBwuWhenListeningWithAutoBwu) { EXPECT_FALSE(client1_.AutoUpgradeBandwidth()); - StartListeningForIncomingConnections(&client1_, - {}, + StartListeningForIncomingConnections(&client1_, {}, {.auto_upgrade_bandwidth = true}); EXPECT_TRUE(client1_.AutoUpgradeBandwidth()); }