diff --git a/cpp/analytics/analytics_recorder.cc b/cpp/analytics/analytics_recorder.cc index c3a0a261..8542a00c 100644 --- a/cpp/analytics/analytics_recorder.cc +++ b/cpp/analytics/analytics_recorder.cc @@ -28,20 +28,45 @@ namespace location { namespace nearby { namespace analytics { -using ConnectionsLog = ::location::nearby::analytics::proto::ConnectionsLog; -using ClientSession = ConnectionsLog::ClientSession; -using StrategySession = ConnectionsLog::StrategySession; -using AdvertisingPhase = ConnectionsLog::AdvertisingPhase; -using DiscoveryPhase = ConnectionsLog::DiscoveryPhase; -using DiscoveredEndpoint = ConnectionsLog::DiscoveredEndpoint; -using ConnectionRequest = ConnectionsLog::ConnectionRequest; -using ConnectionRequestResponse = - ::location::nearby::proto::connections::ConnectionRequestResponse; -using ConnectionStrategy = - ::location::nearby::proto::connections::ConnectionsStrategy; -using EventType = ::location::nearby::proto::connections::EventType; -using Medium = ::location::nearby::proto::connections::Medium; -using SessionRole = ::location::nearby::proto::connections::SessionRole; +using ::location::nearby::analytics::proto::ConnectionsLog; +using AdvertisingPhase = + ::location::nearby::analytics::proto::ConnectionsLog::AdvertisingPhase; +using ClientSession = + ::location::nearby::analytics::proto::ConnectionsLog::ClientSession; +using ConnectionRequest = + ::location::nearby::analytics::proto::ConnectionsLog::ConnectionRequest; +using DiscoveredEndpoint = + ::location::nearby::analytics::proto::ConnectionsLog::DiscoveredEndpoint; +using DiscoveryPhase = + ::location::nearby::analytics::proto::ConnectionsLog::DiscoveryPhase; +using StrategySession = + ::location::nearby::analytics::proto::ConnectionsLog::StrategySession; +using ::location::nearby::proto::connections::ACCEPTED; +using ::location::nearby::proto::connections::ADVERTISER; +using ::location::nearby::proto::connections::CLIENT_SESSION; +using ::location::nearby::proto::connections::ConnectionAttemptResult; +using ::location::nearby::proto::connections::ConnectionAttemptType; +using ::location::nearby::proto::connections::ConnectionRequestResponse; +using ::location::nearby::proto::connections::ConnectionsStrategy; +using ::location::nearby::proto::connections::DISCOVERER; +using ::location::nearby::proto::connections::EventType; +using ::location::nearby::proto::connections::IGNORED; +using ::location::nearby::proto::connections::INCOMING; +using ::location::nearby::proto::connections::INITIAL; +using ::location::nearby::proto::connections::Medium; +using ::location::nearby::proto::connections::NOT_SENT; +using ::location::nearby::proto::connections::OUTGOING; +using ::location::nearby::proto::connections::P2P_CLUSTER; +using ::location::nearby::proto::connections::P2P_POINT_TO_POINT; +using ::location::nearby::proto::connections::P2P_STAR; +using ::location::nearby::proto::connections::REJECTED; +using ::location::nearby::proto::connections::RESULT_SUCCESS; +using ::location::nearby::proto::connections::SessionRole; +using ::location::nearby::proto::connections::START_CLIENT_SESSION; +using ::location::nearby::proto::connections::START_STRATEGY_SESSION; +using ::location::nearby::proto::connections::STOP_CLIENT_SESSION; +using ::location::nearby::proto::connections::STOP_STRATEGY_SESSION; +using ::location::nearby::proto::connections::UNKNOWN_STRATEGY; // These definitions are necessary before C++17. constexpr absl::string_view AnalyticsRecorder::kVersion; @@ -52,7 +77,7 @@ AnalyticsRecorder::AnalyticsRecorder(EventLogger *event_logger) NEARBY_LOGS(INFO) << "AnalyticsRecorder ctor event_logger_=" << event_logger_; MutexLock lock(&mutex_); if (CanRecordAnalyticsLocked("OnStartClientSession")) { - LogEvent(::location::nearby::proto::connections::START_CLIENT_SESSION); + LogEvent(START_CLIENT_SESSION); } } @@ -75,8 +100,7 @@ void AnalyticsRecorder::OnStartAdvertising(connections::Strategy strategy, return; } // Initialize/update a StrategySession. - UpdateStrategySessionLocked( - strategy, ::location::nearby::proto::connections::ADVERTISER); + UpdateStrategySessionLocked(strategy, ADVERTISER); // Initialize and set a AdvertisingPhase. started_advertising_phase_time_ = SystemClock::ElapsedRealtime(); @@ -92,6 +116,7 @@ void AnalyticsRecorder::OnStopAdvertising() { } RecordAdvertisingPhaseDurationLocked(); } + void AnalyticsRecorder::OnStartDiscovery(connections::Strategy strategy, const std::vector &mediums) { MutexLock lock(&mutex_); @@ -105,8 +130,7 @@ void AnalyticsRecorder::OnStartDiscovery(connections::Strategy strategy, } // Initialize/update a StrategySession. - UpdateStrategySessionLocked( - strategy, ::location::nearby::proto::connections::DISCOVERER); + UpdateStrategySessionLocked(strategy, DISCOVERER); // Initialize and set a DiscoveryPhase. started_discovery_phase_time_ = SystemClock::ElapsedRealtime(); @@ -177,8 +201,7 @@ void AnalyticsRecorder::OnRemoteEndpointAccepted( if (!CanRecordAnalyticsLocked("OnRemoteEndpointAccepted")) { return; } - RemoteEndpointRespondedLocked( - remote_endpoint_id, ::location::nearby::proto::connections::ACCEPTED); + RemoteEndpointRespondedLocked(remote_endpoint_id, ACCEPTED); } void AnalyticsRecorder::OnLocalEndpointAccepted( @@ -187,8 +210,7 @@ void AnalyticsRecorder::OnLocalEndpointAccepted( if (!CanRecordAnalyticsLocked("OnLocalEndpointAccepted")) { return; } - LocalEndpointRespondedLocked( - remote_endpoint_id, ::location::nearby::proto::connections::ACCEPTED); + LocalEndpointRespondedLocked(remote_endpoint_id, ACCEPTED); } void AnalyticsRecorder::OnRemoteEndpointRejected( @@ -197,8 +219,7 @@ void AnalyticsRecorder::OnRemoteEndpointRejected( if (!CanRecordAnalyticsLocked("OnRemoteEndpointRejected")) { return; } - RemoteEndpointRespondedLocked( - remote_endpoint_id, ::location::nearby::proto::connections::REJECTED); + RemoteEndpointRespondedLocked(remote_endpoint_id, REJECTED); } void AnalyticsRecorder::OnLocalEndpointRejected( @@ -207,8 +228,65 @@ void AnalyticsRecorder::OnLocalEndpointRejected( if (!CanRecordAnalyticsLocked("OnLocalEndpointRejected")) { return; } - LocalEndpointRespondedLocked( - remote_endpoint_id, ::location::nearby::proto::connections::REJECTED); + LocalEndpointRespondedLocked(remote_endpoint_id, REJECTED); +} + +void AnalyticsRecorder::OnIncomingConnectionAttempt( + ConnectionAttemptType type, Medium medium, ConnectionAttemptResult result, + absl::Duration duration, const std::string &connection_token) { + MutexLock lock(&mutex_); + if (!CanRecordAnalyticsLocked("OnIncomingConnectionAttempt")) { + return; + } + if (current_strategy_session_ == nullptr) { + NEARBY_LOGS(INFO) << "Unable to record incoming connection attempt due to " + "null current_strategy_session_"; + return; + } + auto *connection_attempt = + current_strategy_session_->add_connection_attempt(); + connection_attempt->set_duration_millis(absl::ToInt64Milliseconds(duration)); + connection_attempt->set_type(type); + connection_attempt->set_direction(INCOMING); + connection_attempt->set_medium(medium); + connection_attempt->set_attempt_result(result); + connection_attempt->set_connection_token(connection_token); +} + +void AnalyticsRecorder::OnOutgoingConnectionAttempt( + const std::string &remote_endpoint_id, ConnectionAttemptType type, + Medium medium, ConnectionAttemptResult result, absl::Duration duration, + const std::string &connection_token) { + MutexLock lock(&mutex_); + if (!CanRecordAnalyticsLocked("OnOutgoingConnectionAttempt")) { + return; + } + if (current_strategy_session_ == nullptr) { + NEARBY_LOGS(INFO) << "Unable to record outgoing connection attempt due to " + "null current_strategy_session_"; + return; + } + auto *connection_attempt = + current_strategy_session_->add_connection_attempt(); + connection_attempt->set_duration_millis(absl::ToInt64Milliseconds(duration)); + connection_attempt->set_type(type); + connection_attempt->set_direction(OUTGOING); + connection_attempt->set_medium(medium); + connection_attempt->set_attempt_result(result); + connection_attempt->set_connection_token(connection_token); + if (type == INITIAL && result != RESULT_SUCCESS) { + auto it = outgoing_connection_requests_.find(remote_endpoint_id); + if (it != outgoing_connection_requests_.end()) { + // An outgoing, initial ConnectionAttempt has a corresponding + // ConnectionRequest that, since the ConnectionAttempt has failed, will + // never be delivered to the advertiser. + auto pair = outgoing_connection_requests_.extract(it); + std::unique_ptr &connection_request = pair.mapped(); + connection_request->set_local_response(NOT_SENT); + connection_request->set_remote_response(NOT_SENT); + UpdateDiscovererConnectionRequestLocked(connection_request.release()); + } + } } void AnalyticsRecorder::LogSession() { @@ -220,7 +298,7 @@ void AnalyticsRecorder::LogSession() { client_session_->set_duration_millis(absl::ToInt64Milliseconds( SystemClock::ElapsedRealtime() - started_client_session_time_)); LogClientSession(); - LogEvent(::location::nearby::proto::connections::STOP_CLIENT_SESSION); + LogEvent(STOP_CLIENT_SESSION); session_was_logged_ = true; } @@ -248,12 +326,11 @@ void AnalyticsRecorder::LogClientSession() { serial_executor_.Execute( "analytics-recorder", [this]() { ConnectionsLog connections_log; - connections_log.set_event_type( - ::location::nearby::proto::connections::CLIENT_SESSION); + connections_log.set_event_type(CLIENT_SESSION); connections_log.set_allocated_client_session(client_session_.release()); connections_log.set_version(kVersion); - NEARBY_LOGS(VERBOSE) + NEARBY_LOGS(INFO) << "AnalyticsRecorder LogClientSession connections_log=" << connections_log.DebugString(); @@ -283,10 +360,10 @@ void AnalyticsRecorder::UpdateStrategySessionLocked( // We've already acted as this role before, so make sure we've finished // recording the previous round. switch (role) { - case ::location::nearby::proto::connections::ADVERTISER: + case ADVERTISER: FinishAdvertisingPhaseLocked(); break; - case ::location::nearby::proto::connections::DISCOVERER: + case DISCOVERER: FinishDiscoveryPhaseLocked(); break; default: @@ -299,7 +376,7 @@ void AnalyticsRecorder::UpdateStrategySessionLocked( // Otherwise, we're starting a new Strategy. current_strategy_ = strategy; FinishStrategySessionLocked(); - LogEvent(::location::nearby::proto::connections::START_STRATEGY_SESSION); + LogEvent(START_STRATEGY_SESSION); current_strategy_session_ = std::make_unique(); started_strategy_session_time_ = SystemClock::ElapsedRealtime(); current_strategy_session_->set_strategy( @@ -391,8 +468,7 @@ bool AnalyticsRecorder::UpdateDiscovererConnectionRequestLocked( return false; } if (BothEndpointsRespondedLocked(request) || - request->local_response() == - ::location::nearby::proto::connections::NOT_SENT) { + request->local_response() == NOT_SENT) { request->set_duration_millis( absl::ToUnixMillis(SystemClock::ElapsedRealtime()) - request->duration_millis()); @@ -451,12 +527,10 @@ void AnalyticsRecorder::RemoteEndpointRespondedLocked( void AnalyticsRecorder::MarkConnectionRequestIgnoredLocked( ConnectionRequest *request) { if (!request->has_local_response()) { - request->set_local_response( - ::location::nearby::proto::connections::IGNORED); + request->set_local_response(IGNORED); } if (!request->has_remote_response()) { - request->set_remote_response( - ::location::nearby::proto::connections::IGNORED); + request->set_remote_response(IGNORED); } } @@ -470,22 +544,22 @@ void AnalyticsRecorder::FinishStrategySessionLocked() { SystemClock::ElapsedRealtime() - SystemClock::ElapsedRealtime())); *client_session_->add_strategy_session() = *std::move(current_strategy_session_); - LogEvent(::location::nearby::proto::connections::STOP_STRATEGY_SESSION); + LogEvent(STOP_STRATEGY_SESSION); } } -ConnectionStrategy AnalyticsRecorder::StrategyToConnectionStrategy( +ConnectionsStrategy AnalyticsRecorder::StrategyToConnectionStrategy( connections::Strategy strategy) { if (strategy == connections::Strategy::kP2pCluster) { - return ::location::nearby::proto::connections::P2P_CLUSTER; + return P2P_CLUSTER; } if (strategy == connections::Strategy::kP2pStar) { - return ::location::nearby::proto::connections::P2P_STAR; + return P2P_STAR; } if (strategy == connections::Strategy::kP2pPointToPoint) { - return ::location::nearby::proto::connections::P2P_POINT_TO_POINT; + return P2P_POINT_TO_POINT; } - return ::location::nearby::proto::connections::UNKNOWN_STRATEGY; + return UNKNOWN_STRATEGY; } } // namespace analytics diff --git a/cpp/analytics/analytics_recorder.h b/cpp/analytics/analytics_recorder.h index b06f3ab1..930a9e2e 100644 --- a/cpp/analytics/analytics_recorder.h +++ b/cpp/analytics/analytics_recorder.h @@ -67,6 +67,21 @@ class AnalyticsRecorder { void OnLocalEndpointRejected(const std::string &remote_endpoint_id) ABSL_LOCKS_EXCLUDED(mutex_); + // Connection attempt + void OnIncomingConnectionAttempt( + ::location::nearby::proto::connections::ConnectionAttemptType type, + ::location::nearby::proto::connections::Medium medium, + ::location::nearby::proto::connections::ConnectionAttemptResult result, + absl::Duration duration, const std::string &connection_token) + ABSL_LOCKS_EXCLUDED(mutex_); + void OnOutgoingConnectionAttempt( + const std::string &remote_endpoint_id, + ::location::nearby::proto::connections::ConnectionAttemptType type, + ::location::nearby::proto::connections::Medium medium, + ::location::nearby::proto::connections::ConnectionAttemptResult result, + absl::Duration duration, const std::string &connection_token) + ABSL_LOCKS_EXCLUDED(mutex_); + // Invokes event_logger_.Log() at the end of life of client. Log action is // called in a separate thread to allow synchronous potentially lengthy // execution. diff --git a/cpp/analytics/analytics_recorder_test.cc b/cpp/analytics/analytics_recorder_test.cc index dc16d19f..1b088e7d 100644 --- a/cpp/analytics/analytics_recorder_test.cc +++ b/cpp/analytics/analytics_recorder_test.cc @@ -30,21 +30,20 @@ namespace nearby { namespace analytics { namespace { -using ::location::nearby::proto::connections::ACCEPTED; using ::location::nearby::proto::connections::BLE; using ::location::nearby::proto::connections::BLUETOOTH; using ::location::nearby::proto::connections::CLIENT_SESSION; -using ::location::nearby::proto::connections::ConnectionRequestResponse; using ::location::nearby::proto::connections::EventType; -using ::location::nearby::proto::connections::IGNORED; +using ::location::nearby::proto::connections::INITIAL; using ::location::nearby::proto::connections::Medium; -using ::location::nearby::proto::connections::REJECTED; +using ::location::nearby::proto::connections::RESULT_ERROR; +using ::location::nearby::proto::connections::RESULT_SUCCESS; using ::location::nearby::proto::connections::START_STRATEGY_SESSION; using ::location::nearby::proto::connections::STOP_CLIENT_SESSION; using ::location::nearby::proto::connections::STOP_STRATEGY_SESSION; using ::testing::Contains; -using ::testing::ElementsAre; -using ::testing::UnorderedElementsAreArray; +using ::testing::EqualsProto; +using ::testing::proto::Partially; using ::location::nearby::analytics::proto::ConnectionsLog; using ClientSession = @@ -105,7 +104,7 @@ TEST(AnalyticsRecorderTest, SessionOnlyLoggedOnceWorks) { analytics_recorder.LogSession(); analytics_recorder.LogSession(); analytics_recorder.LogSession(); - EXPECT_TRUE(client_session_done_latch.Await(kDefaultTimeout).result()); + ASSERT_TRUE(client_session_done_latch.Await(kDefaultTimeout).result()); // Only called once. EXPECT_EQ(event_logger.GetLoggedClientSessionCount(), 1); @@ -124,15 +123,20 @@ TEST(AnalyticsRecorderTest, SetFieldsCorrectlyForNestedAdvertisingCalls) { analytics_recorder.OnStartAdvertising(strategy, {BLUETOOTH}); analytics_recorder.LogSession(); - EXPECT_TRUE(client_session_done_latch.Await(kDefaultTimeout).result()); + ASSERT_TRUE(client_session_done_latch.Await(kDefaultTimeout).result()); + + constexpr char kExpected[] = + R"pb( + strategy_session < + strategy: P2P_STAR + role: ADVERTISER + advertising_phase < medium: BLE medium: BLUETOOTH > + advertising_phase < medium: BLUETOOTH > + >)pb"; const ClientSession& client_session = event_logger.GetLoggedClientSession(); - ASSERT_EQ(client_session.strategy_session_size(), 1); - ASSERT_EQ(client_session.strategy_session(0).advertising_phase_size(), 2); - EXPECT_THAT(client_session.strategy_session(0).advertising_phase(0).medium(), - UnorderedElementsAreArray(mediums)); - EXPECT_THAT(client_session.strategy_session(0).advertising_phase(1).medium(), - ElementsAre(BLUETOOTH)); + + EXPECT_THAT(client_session, Partially(EqualsProto(kExpected))); } TEST(AnalyticsRecorderTest, SetFieldsCorrectlyForNestedDiscoveryCalls) { @@ -150,29 +154,20 @@ TEST(AnalyticsRecorderTest, SetFieldsCorrectlyForNestedDiscoveryCalls) { analytics_recorder.OnStartDiscovery(strategy, {BLUETOOTH}); analytics_recorder.LogSession(); - EXPECT_TRUE(client_session_done_latch.Await(kDefaultTimeout).result()); + ASSERT_TRUE(client_session_done_latch.Await(kDefaultTimeout).result()); - const ClientSession& client_session = event_logger.GetLoggedClientSession(); - EXPECT_EQ(client_session.strategy_session_size(), 1); - EXPECT_EQ(client_session.strategy_session(0).discovery_phase_size(), 2); - EXPECT_THAT(client_session.strategy_session(0).discovery_phase(0).medium(), - UnorderedElementsAreArray(mediums)); - EXPECT_EQ(client_session.strategy_session(0) - .discovery_phase(0) - .discovered_endpoint_size(), - 2); - EXPECT_EQ(client_session.strategy_session(0) - .discovery_phase(0) - .discovered_endpoint(0) - .medium(), - BLUETOOTH); - EXPECT_EQ(client_session.strategy_session(0) - .discovery_phase(0) - .discovered_endpoint(1) - .medium(), - BLE); - EXPECT_THAT(client_session.strategy_session(0).discovery_phase(1).medium(), - ElementsAre(BLUETOOTH)); + EXPECT_THAT(event_logger.GetLoggedClientSession(), Partially(EqualsProto(R"pb( + strategy_session < + strategy: P2P_STAR + role: DISCOVERER + discovery_phase < + medium: BLE + medium: BLUETOOTH + discovered_endpoint < medium: BLUETOOTH > + discovered_endpoint < medium: BLE > + > + discovery_phase < medium: BLUETOOTH > + >)pb"))); } TEST(AnalyticsRecorderTest, @@ -198,15 +193,24 @@ TEST(AnalyticsRecorderTest, analytics_recorder.OnStopAdvertising(); analytics_recorder.LogSession(); - EXPECT_TRUE(client_session_done_latch.Await(kDefaultTimeout).result()); + ASSERT_TRUE(client_session_done_latch.Await(kDefaultTimeout).result()); std::vector event_types = event_logger.GetLoggedEventTypes(); EXPECT_THAT(event_types, Contains(START_STRATEGY_SESSION).Times(1)); EXPECT_THAT(event_types, Contains(STOP_STRATEGY_SESSION).Times(1)); - const ClientSession& client_session = event_logger.GetLoggedClientSession(); - EXPECT_EQ(client_session.strategy_session_size(), 1); - EXPECT_EQ(client_session.strategy_session(0).advertising_phase_size(), 3); - EXPECT_EQ(client_session.strategy_session(0).discovery_phase_size(), 3); + + EXPECT_THAT(event_logger.GetLoggedClientSession(), Partially(EqualsProto(R"pb( + strategy_session < + strategy: P2P_STAR + role: ADVERTISER + role: DISCOVERER + discovery_phase < medium: BLE medium: BLUETOOTH > + discovery_phase < medium: BLE medium: BLUETOOTH > + discovery_phase < medium: BLE medium: BLUETOOTH > + advertising_phase < medium: BLE medium: BLUETOOTH > + advertising_phase < medium: BLE medium: BLUETOOTH > + advertising_phase < medium: BLE medium: BLUETOOTH > + >)pb"))); } TEST(AnalyticsRecorderTest, AdvertiserConnectionRequestsWorks) { @@ -239,52 +243,39 @@ TEST(AnalyticsRecorderTest, AdvertiserConnectionRequestsWorks) { analytics_recorder.OnRemoteEndpointRejected(endpoint_id_3); analytics_recorder.LogSession(); - EXPECT_TRUE(client_session_done_latch.Await(kDefaultTimeout).result()); + ASSERT_TRUE(client_session_done_latch.Await(kDefaultTimeout).result()); - const ClientSession& client_session = event_logger.GetLoggedClientSession(); - StrategySession strategy_session = client_session.strategy_session(0); - EXPECT_EQ(strategy_session.advertising_phase_size(), 1); - EXPECT_EQ( - strategy_session.advertising_phase(0).received_connection_request_size(), - 4); - EXPECT_EQ(strategy_session.advertising_phase(0) - .received_connection_request(0) - .local_response(), - ACCEPTED); - EXPECT_EQ(strategy_session.advertising_phase(0) - .received_connection_request(0) - .remote_response(), - ACCEPTED); - EXPECT_EQ(strategy_session.advertising_phase(0) - .received_connection_request(1) - .local_response(), - ACCEPTED); - EXPECT_EQ(strategy_session.advertising_phase(0) - .received_connection_request(1) - .remote_response(), - REJECTED); - EXPECT_EQ(strategy_session.advertising_phase(0) - .received_connection_request(2) - .local_response(), - REJECTED); - EXPECT_EQ(strategy_session.advertising_phase(0) - .received_connection_request(2) - .remote_response(), - ACCEPTED); - EXPECT_EQ(strategy_session.advertising_phase(0) - .received_connection_request(3) - .local_response(), - REJECTED); - EXPECT_EQ(strategy_session.advertising_phase(0) - .received_connection_request(3) - .remote_response(), - REJECTED); + EXPECT_THAT(event_logger.GetLoggedClientSession(), Partially(EqualsProto(R"pb( + strategy_session < + strategy: P2P_STAR + role: ADVERTISER + advertising_phase < + medium: BLE + medium: BLUETOOTH + received_connection_request < + request_delay_millis: 0 + local_response: ACCEPTED + remote_response: ACCEPTED + > + received_connection_request < + local_response: ACCEPTED + remote_response: REJECTED + > + received_connection_request < + local_response: REJECTED + remote_response: ACCEPTED + > + received_connection_request < + local_response: REJECTED + remote_response: REJECTED + > + > + >)pb"))); } TEST(AnalyticsRecorderTest, DiscoveryConnectionRequestsWorks) { connections::Strategy strategy = connections::Strategy::kP2pStar; std::vector mediums = {BLE, BLUETOOTH}; - std::string endpoint_id_0("endpoint_id_0"); std::string endpoint_id_1("endpoint_id_1"); std::string endpoint_id_2("endpoint_id_2"); @@ -313,29 +304,33 @@ TEST(AnalyticsRecorderTest, DiscoveryConnectionRequestsWorks) { analytics_recorder.OnRemoteEndpointRejected(endpoint_id_3); analytics_recorder.LogSession(); - EXPECT_TRUE(client_session_done_latch.Await(kDefaultTimeout).result()); + ASSERT_TRUE(client_session_done_latch.Await(kDefaultTimeout).result()); - const ClientSession& client_session = event_logger.GetLoggedClientSession(); - StrategySession strategy_session = client_session.strategy_session(0); - EXPECT_EQ(strategy_session.discovery_phase_size(), 1); - EXPECT_EQ(strategy_session.discovery_phase(0).sent_connection_request_size(), - 4); - auto& sent_connection_request_0 = - strategy_session.discovery_phase(0).sent_connection_request(0); - EXPECT_EQ(sent_connection_request_0.local_response(), ACCEPTED); - EXPECT_EQ(sent_connection_request_0.remote_response(), ACCEPTED); - auto& sent_connection_request_1 = - strategy_session.discovery_phase(0).sent_connection_request(1); - EXPECT_EQ(sent_connection_request_1.local_response(), ACCEPTED); - EXPECT_EQ(sent_connection_request_1.remote_response(), REJECTED); - auto& sent_connection_request_2 = - strategy_session.discovery_phase(0).sent_connection_request(2); - EXPECT_EQ(sent_connection_request_2.local_response(), REJECTED); - EXPECT_EQ(sent_connection_request_2.remote_response(), ACCEPTED); - auto& sent_connection_request_3 = - strategy_session.discovery_phase(0).sent_connection_request(3); - EXPECT_EQ(sent_connection_request_3.local_response(), REJECTED); - EXPECT_EQ(sent_connection_request_3.remote_response(), REJECTED); + EXPECT_THAT(event_logger.GetLoggedClientSession(), Partially(EqualsProto(R"pb( + strategy_session < + strategy: P2P_STAR + role: DISCOVERER + discovery_phase < + medium: BLE + medium: BLUETOOTH + sent_connection_request < + local_response: ACCEPTED + remote_response: ACCEPTED + > + sent_connection_request < + local_response: ACCEPTED + remote_response: REJECTED + > + sent_connection_request < + local_response: REJECTED + remote_response: ACCEPTED + > + sent_connection_request < + local_response: REJECTED + remote_response: REJECTED + > + > + >)pb"))); } TEST(AnalyticsRecorderTest, @@ -363,39 +358,29 @@ TEST(AnalyticsRecorderTest, analytics_recorder.OnConnectionRequestReceived(endpoint_id_2); analytics_recorder.LogSession(); - EXPECT_TRUE(client_session_done_latch.Await(kDefaultTimeout).result()); + ASSERT_TRUE(client_session_done_latch.Await(kDefaultTimeout).result()); - const ClientSession& client_session = event_logger.GetLoggedClientSession(); - StrategySession strategy_session = client_session.strategy_session(0); - EXPECT_EQ(strategy_session.advertising_phase_size(), 1); - EXPECT_EQ( - strategy_session.advertising_phase(0).received_connection_request_size(), - 3); - - EXPECT_EQ(strategy_session.advertising_phase(0) - .received_connection_request(0) - .local_response(), - IGNORED); - EXPECT_EQ(strategy_session.advertising_phase(0) - .received_connection_request(0) - .remote_response(), - ACCEPTED); - EXPECT_EQ(strategy_session.advertising_phase(0) - .received_connection_request(1) - .local_response(), - ACCEPTED); - EXPECT_EQ(strategy_session.advertising_phase(0) - .received_connection_request(1) - .remote_response(), - IGNORED); - EXPECT_EQ(strategy_session.advertising_phase(0) - .received_connection_request(2) - .local_response(), - IGNORED); - EXPECT_EQ(strategy_session.advertising_phase(0) - .received_connection_request(2) - .remote_response(), - IGNORED); + EXPECT_THAT(event_logger.GetLoggedClientSession(), Partially(EqualsProto(R"pb( + strategy_session < + strategy: P2P_STAR + role: ADVERTISER + advertising_phase < + medium: BLE + medium: BLUETOOTH + received_connection_request < + local_response: IGNORED + remote_response: ACCEPTED + > + received_connection_request < + local_response: ACCEPTED + remote_response: IGNORED + > + received_connection_request < + local_response: IGNORED + remote_response: IGNORED + > + > + >)pb"))); } TEST(AnalyticsRecorderTest, @@ -424,38 +409,104 @@ TEST(AnalyticsRecorderTest, analytics_recorder.OnConnectionRequestSent(endpoint_id_2); analytics_recorder.LogSession(); - EXPECT_TRUE(client_session_done_latch.Await(kDefaultTimeout).result()); + ASSERT_TRUE(client_session_done_latch.Await(kDefaultTimeout).result()); - const ClientSession& client_session = event_logger.GetLoggedClientSession(); - StrategySession strategy_session = client_session.strategy_session(0); - EXPECT_EQ(strategy_session.discovery_phase_size(), 1); - EXPECT_EQ(strategy_session.discovery_phase(0).sent_connection_request_size(), - 3); + EXPECT_THAT(event_logger.GetLoggedClientSession(), Partially(EqualsProto(R"pb( + strategy_session < + strategy: P2P_STAR + role: DISCOVERER + discovery_phase < + medium: BLE + medium: BLUETOOTH + sent_connection_request < + local_response: IGNORED + remote_response: ACCEPTED + > + sent_connection_request < + local_response: ACCEPTED + remote_response: IGNORED + > + sent_connection_request < + local_response: IGNORED + remote_response: IGNORED + > + > + >)pb"))); +} - EXPECT_EQ(strategy_session.discovery_phase(0) - .sent_connection_request(0) - .local_response(), - IGNORED); - EXPECT_EQ(strategy_session.discovery_phase(0) - .sent_connection_request(0) - .remote_response(), - ACCEPTED); - EXPECT_EQ(strategy_session.discovery_phase(0) - .sent_connection_request(1) - .local_response(), - ACCEPTED); - EXPECT_EQ(strategy_session.discovery_phase(0) - .sent_connection_request(1) - .remote_response(), - IGNORED); - EXPECT_EQ(strategy_session.discovery_phase(0) - .sent_connection_request(2) - .local_response(), - IGNORED); - EXPECT_EQ(strategy_session.discovery_phase(0) - .sent_connection_request(2) - .remote_response(), - IGNORED); +TEST(AnalyticsRecorderTest, SuccessfulIncomingConnectionAttempt) { + connections::Strategy strategy = connections::Strategy::kP2pStar; + std::vector mediums = {BLE, BLUETOOTH}; + std::string endpoint_id("endpoint_id"); + std::string connection_token(""); + + CountDownLatch client_session_done_latch(1); + FakeEventLogger event_logger(client_session_done_latch); + AnalyticsRecorder analytics_recorder{&event_logger}; + + analytics_recorder.OnStartAdvertising(strategy, mediums); + analytics_recorder.OnIncomingConnectionAttempt( + INITIAL, BLUETOOTH, RESULT_SUCCESS, absl::Duration{}, connection_token); + analytics_recorder.OnStopAdvertising(); + + analytics_recorder.LogSession(); + ASSERT_TRUE(client_session_done_latch.Await(kDefaultTimeout).result()); + + EXPECT_THAT(event_logger.GetLoggedClientSession(), Partially(EqualsProto(R"pb( + strategy_session < + strategy: P2P_STAR + role: ADVERTISER + advertising_phase < medium: BLE medium: BLUETOOTH > + connection_attempt < + type: INITIAL + direction: INCOMING + medium: BLUETOOTH + attempt_result: RESULT_SUCCESS + connection_token: "" + > + >)pb"))); +} + +TEST(AnalyticsRecorderTest, + FailedConnectionAttemptUpdatesConnectionRequestNotSent) { + connections::Strategy strategy = connections::Strategy::kP2pStar; + std::vector mediums = {BLE, BLUETOOTH}; + std::string endpoint_id("endpoint_id"); + std::string connection_token(""); + + CountDownLatch client_session_done_latch(1); + FakeEventLogger event_logger(client_session_done_latch); + AnalyticsRecorder analytics_recorder{&event_logger}; + + analytics_recorder.OnStartDiscovery(strategy, mediums); + analytics_recorder.OnConnectionRequestSent(endpoint_id); + analytics_recorder.OnOutgoingConnectionAttempt( + endpoint_id, INITIAL, BLUETOOTH, RESULT_ERROR, absl::Duration{}, + connection_token); + + analytics_recorder.LogSession(); + ASSERT_TRUE(client_session_done_latch.Await(kDefaultTimeout).result()); + + EXPECT_THAT(event_logger.GetLoggedClientSession(), Partially(EqualsProto(R"pb( + strategy_session < + strategy: P2P_STAR + role: DISCOVERER + discovery_phase < + medium: BLE + medium: BLUETOOTH + sent_connection_request < + local_response: NOT_SENT + remote_response: NOT_SENT + > + > + connection_attempt < + type: INITIAL + direction: OUTGOING + medium: BLUETOOTH + attempt_result: RESULT_ERROR + connection_token: "" + > + >)pb"))); } } // namespace diff --git a/cpp/core/internal/base_pcp_handler.cc b/cpp/core/internal/base_pcp_handler.cc index 6b3813d9..c94400d6 100644 --- a/cpp/core/internal/base_pcp_handler.cc +++ b/cpp/core/internal/base_pcp_handler.cc @@ -338,7 +338,9 @@ void BasePcpHandler::OnEncryptionSuccessRunnable( if (!ukey2) { // Fail early, if there is no crypto context. ProcessPreConnectionInitiationFailure( - endpoint_id, connection_info.channel.get(), {Status::kEndpointIoError}, + connection_info.client, connection_info.channel->GetMedium(), + endpoint_id, connection_info.channel.get(), connection_info.is_incoming, + connection_info.start_time, {Status::kEndpointIoError}, connection_info.result.lock().get()); return; } @@ -388,6 +390,21 @@ void BasePcpHandler::OnEncryptionSuccessRunnable( std::move(connection_info.channel), connection_info.listener, connection_info.connection_token); + if (connection_info.is_incoming) { + connection_info.client->GetAnalyticsRecorder().OnIncomingConnectionAttempt( + proto::connections::INITIAL, connection_info.channel->GetMedium(), + proto::connections::RESULT_SUCCESS, + SystemClock::ElapsedRealtime() - connection_info.start_time, + connection_info.connection_token); + } else { + connection_info.client->GetAnalyticsRecorder().OnOutgoingConnectionAttempt( + endpoint_id, proto::connections::INITIAL, + connection_info.channel->GetMedium(), + proto::connections::RESULT_SUCCESS, + SystemClock::ElapsedRealtime() - connection_info.start_time, + connection_info.connection_token); + } + if (auto future_status = connection_info.result.lock()) { NEARBY_LOGS(INFO) << "Connection established; Finalising future OK."; future_status->Set({Status::kSuccess}); @@ -420,9 +437,10 @@ void BasePcpHandler::OnEncryptionFailureRunnable( return; } - ProcessPreConnectionInitiationFailure(endpoint_id, info.channel.get(), - {Status::kEndpointIoError}, - info.result.lock().get()); + ProcessPreConnectionInitiationFailure( + info.client, info.channel->GetMedium(), endpoint_id, info.channel.get(), + info.is_incoming, info.start_time, {Status::kEndpointIoError}, + info.result.lock().get()); } Status BasePcpHandler::RequestConnection(ClientProxy* client, @@ -486,6 +504,7 @@ Status BasePcpHandler::RequestConnection(ClientProxy* client, ConnectImplResult connect_impl_result; for (auto connect_endpoint : discovered_endpoints) { + absl::Time connect_start_time = SystemClock::ElapsedRealtime(); if (!MediumSupportedByClientOptions(connect_endpoint->medium, options)) continue; @@ -493,15 +512,20 @@ Status BasePcpHandler::RequestConnection(ClientProxy* client, if (connect_impl_result.status.Ok()) { channel = std::move(connect_impl_result.endpoint_channel); break; + } else { + LogConnectionAttempt(client, connect_endpoint->medium, + connect_endpoint->endpoint_id, + /* is_incoming = */ false, connect_start_time); } } if (channel == nullptr) { NEARBY_LOGS(INFO) << "Endpoint channel not available: endpoint_id=" << endpoint_id; - ProcessPreConnectionInitiationFailure(endpoint_id, channel.get(), - connect_impl_result.status, - result.get()); + ProcessPreConnectionInitiationFailure( + client, channel->GetMedium(), endpoint_id, channel.get(), + /* is_incoming = */ false, start_time, connect_impl_result.status, + result.get()); return; } @@ -522,9 +546,10 @@ Status BasePcpHandler::RequestConnection(ClientProxy* client, if (!write_exception.Ok()) { NEARBY_LOGS(INFO) << "Failed to send connection request: endpoint_id=" << endpoint_id; - ProcessPreConnectionInitiationFailure(endpoint_id, channel.get(), - {Status::kEndpointIoError}, - result.get()); + ProcessPreConnectionInitiationFailure( + client, channel->GetMedium(), endpoint_id, channel.get(), + /* is_incoming = */ false, start_time, {Status::kEndpointIoError}, + result.get()); return; } @@ -682,8 +707,9 @@ Exception BasePcpHandler::WriteConnectionRequestFrame( } void BasePcpHandler::ProcessPreConnectionInitiationFailure( - const std::string& endpoint_id, EndpointChannel* channel, Status status, - Future* result) { + ClientProxy* client, Medium medium, const std::string& endpoint_id, + EndpointChannel* channel, bool is_incoming, absl::Time start_time, + Status status, Future* result) { if (channel != nullptr) { channel->Close(); } @@ -693,6 +719,7 @@ void BasePcpHandler::ProcessPreConnectionInitiationFailure( result->Set(status); } + LogConnectionAttempt(client, medium, endpoint_id, is_incoming, start_time); // result is hold inside a swapper, and saved in PendingConnectionInfo. // PendingConnectionInfo destructor will clear the memory of SettableFuture // shared_ptr for result. @@ -1051,7 +1078,9 @@ Exception BasePcpHandler::OnIncomingConnection( << "Failed to parse incoming connection request; client=" << client->GetClientId() << "; device=" << absl::BytesToHexString(remote_endpoint_info.data()); - ProcessPreConnectionInitiationFailure("", channel.get(), {Status::kError}, + ProcessPreConnectionInitiationFailure(client, medium, "", channel.get(), + /* is_incoming= */ false, + start_time, {Status::kError}, nullptr); return {Exception::kSuccess}; } @@ -1217,9 +1246,10 @@ bool BasePcpHandler::BreakTie(ClientProxy* client, void BasePcpHandler::ProcessTieBreakLoss( ClientProxy* client, const std::string& endpoint_id, BasePcpHandler::PendingConnectionInfo* info) { - ProcessPreConnectionInitiationFailure(endpoint_id, info->channel.get(), - {Status::kEndpointIoError}, - info->result.lock().get()); + ProcessPreConnectionInitiationFailure( + client, info->channel->GetMedium(), endpoint_id, info->channel.get(), + info->is_incoming, info->start_time, {Status::kEndpointIoError}, + info->result.lock().get()); ProcessPreConnectionResultFailure(client, endpoint_id); } @@ -1430,6 +1460,35 @@ std::string BasePcpHandler::GetHashedConnectionToken( .substr(0, kConnectionTokenLength); } +void BasePcpHandler::LogConnectionAttempt(ClientProxy* client, Medium medium, + const std::string& endpoint_id, + bool is_incoming, + absl::Time start_time) { + proto::connections::ConnectionAttemptResult result = + Cancelled(client, endpoint_id) ? proto::connections::RESULT_CANCELLED + : proto::connections::RESULT_ERROR; + if (is_incoming) { + client->GetAnalyticsRecorder().OnIncomingConnectionAttempt( + proto::connections::INITIAL, medium, result, + SystemClock::ElapsedRealtime() - start_time, + /* connection_token= */ ""); + } else { + client->GetAnalyticsRecorder().OnOutgoingConnectionAttempt( + endpoint_id, proto::connections::INITIAL, medium, result, + SystemClock::ElapsedRealtime() - start_time, + /* connection_token= */ ""); + } +} + +bool BasePcpHandler::Cancelled(ClientProxy* client, + const std::string& endpoint_id) { + if (endpoint_id.empty()) { + return false; + } + + return client->GetCancellationFlag(endpoint_id)->Cancelled(); +} + ///////////////////// BasePcpHandler::PendingConnectionInfo /////////////////// void BasePcpHandler::PendingConnectionInfo::SetCryptoContext( diff --git a/cpp/core/internal/base_pcp_handler.h b/cpp/core/internal/base_pcp_handler.h index 63015f6d..8d3da9c5 100644 --- a/cpp/core/internal/base_pcp_handler.h +++ b/cpp/core/internal/base_pcp_handler.h @@ -427,10 +427,10 @@ class BasePcpHandler : public PcpHandler, bool AppendWebRTCEndpoint(const std::string& endpoint_id, const ConnectionOptions& local_discovery_options); - void ProcessPreConnectionInitiationFailure(const std::string& endpoint_id, - EndpointChannel* channel, - Status status, - Future* result); + void ProcessPreConnectionInitiationFailure( + ClientProxy* client, Medium medium, const std::string& endpoint_id, + EndpointChannel* channel, bool is_incoming, absl::Time start_time, + Status status, Future* result); void ProcessPreConnectionResultFailure(ClientProxy* client, const std::string& endpoint_id); @@ -454,6 +454,14 @@ class BasePcpHandler : public PcpHandler, // array. std::string GetHashedConnectionToken(const ByteArray& token_bytes); + static void LogConnectionAttempt(ClientProxy* client, Medium medium, + const std::string& endpoint_id, + bool is_incoming, absl::Time start_time); + + // Returns true if the client cancels the operation in progress through the + // endpoint id. This is done by CancellationFlag. + static bool Cancelled(ClientProxy* client, const std::string& endpoint_id); + void WaitForLatch(const std::string& method_name, CountDownLatch* latch); Status WaitForResult(const std::string& method_name, std::int64_t client_id, Future* future); diff --git a/cpp/core/internal/bwu_manager.cc b/cpp/core/internal/bwu_manager.cc index 85196a06..7573d226 100644 --- a/cpp/core/internal/bwu_manager.cc +++ b/cpp/core/internal/bwu_manager.cc @@ -367,6 +367,8 @@ void BwuManager::OnIncomingConnection( mutable_connection.release()); RunOnBwuManagerThread( "bwu-on-incoming-connection", [this, client, connection]() { + absl::Time connection_attempt_start_time = + SystemClock::ElapsedRealtime(); EndpointChannel* channel = connection->channel.get(); if (channel == nullptr) { NEARBY_LOG( @@ -422,6 +424,13 @@ void BwuManager::OnIncomingConnection( CHECK(client == mapped_client); + // The ConnectionAttempt has now succeeded, so record it as such. + client->GetAnalyticsRecorder().OnIncomingConnectionAttempt( + proto::connections::UPGRADE, channel->GetMedium(), + proto::connections::RESULT_SUCCESS, + SystemClock::ElapsedRealtime() - connection_attempt_start_time, + client->GetConnectionToken(endpoint_id)); + // Use the introductory client information sent over to run the upgrade // protocol. RunUpgradeProtocol(mapped_client, endpoint_id, @@ -537,8 +546,24 @@ void BwuManager::ProcessBwuPathAvailableEvent( return; } + absl::Time connection_attempt_start_time = SystemClock::ElapsedRealtime(); auto channel = ProcessBwuPathAvailableEventInternal(client, endpoint_id, upgrade_path_info); + proto::connections::ConnectionAttemptResult connection_attempt_result; + if (channel != nullptr) { + connection_attempt_result = proto::connections::RESULT_SUCCESS; + } else if (client->GetCancellationFlag(endpoint_id)->Cancelled()) { + connection_attempt_result = proto::connections::RESULT_CANCELLED; + } else { + connection_attempt_result = proto::connections::RESULT_ERROR; + } + + client->GetAnalyticsRecorder().OnOutgoingConnectionAttempt( + endpoint_id, proto::connections::UPGRADE, channel->GetMedium(), + connection_attempt_result, + SystemClock::ElapsedRealtime() - connection_attempt_start_time, + client->GetConnectionToken(endpoint_id)); + if (channel == nullptr) { NEARBY_LOG(INFO, "Failed to get new channel."); RunUpgradeFailedProtocol(client, endpoint_id, upgrade_path_info); diff --git a/cpp/core/internal/client_proxy.h b/cpp/core/internal/client_proxy.h index e6a580c5..539b5e35 100644 --- a/cpp/core/internal/client_proxy.h +++ b/cpp/core/internal/client_proxy.h @@ -57,6 +57,10 @@ class ClientProxy final { std::string GetLocalEndpointId(); + analytics::AnalyticsRecorder& GetAnalyticsRecorder() const { + return *analytics_recorder_; + } + std::string GetConnectionToken(const std::string& endpoint_id); // Clears all the runtime state of this client.