TySS-Dev/nearby
1
0
Fork 0
mirror of https://github.com/kidfromjupiter/nearby.git synced 2026-09-14 14:46:12 -04:00
Code Issues Packages Projects Releases Wiki Activity

Fix flakiness in analytics

Browse Source 
`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
This commit is contained in:
Janusz Sobczak
2023-07-13 02:35:21 -07:00
committed by Copybara-Service
parent ddaf447212
commit 02e37e8a0c
3 changed files with 95 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
connections/implementation/analytics/analytics_recorder.cc
+16 -8
View File
@@ -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
connections/implementation/analytics/analytics_recorder.h
+7 -2
View File
@@ -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);
connections/implementation/client_proxy_test.cc
+72 -22
View File
@@ -19,6 +19,7 @@
#include <optional>
#include <string>
#include <utility>
#include <vector>
#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<ConnectionsLog> 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());
}