diff --git a/cpp/analytics/analytics_recorder.cc b/cpp/analytics/analytics_recorder.cc index 48ab0131..cb957c32 100644 --- a/cpp/analytics/analytics_recorder.cc +++ b/cpp/analytics/analytics_recorder.cc @@ -81,6 +81,7 @@ AnalyticsRecorder::~AnalyticsRecorder() { MutexLock lock(&mutex_); incoming_connection_requests_.clear(); outgoing_connection_requests_.clear(); + active_connections_.clear(); serial_executor_.Shutdown(); } @@ -101,7 +102,7 @@ void AnalyticsRecorder::OnStartAdvertising(connections::Strategy strategy, // Initialize and set a AdvertisingPhase. started_advertising_phase_time_ = SystemClock::ElapsedRealtime(); current_advertising_phase_ = - absl::make_unique(); + absl::make_unique(); absl::c_copy(mediums, RepeatedFieldBackInserter( current_advertising_phase_->mutable_medium())); } @@ -133,9 +134,8 @@ void AnalyticsRecorder::OnStartDiscovery(connections::Strategy strategy, started_discovery_phase_time_ = SystemClock::ElapsedRealtime(); current_discovery_phase_ = absl::make_unique(); - for (auto medium : mediums) { - current_discovery_phase_->add_medium(medium); - } + absl::c_copy(mediums, RepeatedFieldBackInserter( + current_discovery_phase_->mutable_medium())); } void AnalyticsRecorder::OnStopDiscovery() { @@ -290,6 +290,50 @@ void AnalyticsRecorder::OnOutgoingConnectionAttempt( } } +void AnalyticsRecorder::OnConnectionEstablished( + const std::string &endpoint_id, Medium medium, + const std::string &connection_token) { + MutexLock lock(&mutex_); + if (!CanRecordAnalyticsLocked("OnConnectionEstablished")) { + return; + } + auto it = active_connections_.find(endpoint_id); + if (it != active_connections_.end()) { + std::unique_ptr &logical_connection = it->second; + logical_connection->PhysicalConnectionEstablished(medium, connection_token); + } else { + active_connections_.insert( + {endpoint_id, + absl::make_unique(medium, connection_token)}); + } +} + +void AnalyticsRecorder::OnConnectionClosed(const std::string &endpoint_id, + Medium medium, + DisconnectionReason reason) { + MutexLock lock(&mutex_); + if (!CanRecordAnalyticsLocked("OnConnectionClosed")) { + return; + } + auto it = active_connections_.find(endpoint_id); + if (it == active_connections_.end()) { + return; + } + std::unique_ptr &logical_connection = it->second; + logical_connection->PhysicalConnectionClosed(medium, reason); + if (reason != UPGRADED) { + // Unless this is an upgraded connection, remove this from our active + // connections. Any future communication with an endpoint will need to be + // re-established with a new ConnectionRequest. + auto pair = active_connections_.extract(it); + std::unique_ptr &logical_connection = pair.mapped(); + absl::c_copy( + logical_connection->GetEstablisedConnections(), + RepeatedFieldBackInserter( + current_strategy_session_->mutable_established_connection())); + } +} + void AnalyticsRecorder::LogSession() { MutexLock lock(&mutex_); if (!CanRecordAnalyticsLocked("LogSession")) { @@ -537,6 +581,17 @@ void AnalyticsRecorder::FinishStrategySessionLocked() { FinishAdvertisingPhaseLocked(); FinishDiscoveryPhaseLocked(); + // Finish any unfinished LogicalConnections. + for (const auto &item : active_connections_) { + auto &logical_connection = item.second; + logical_connection->CloseAllPhysicalConnections(); + absl::c_copy( + logical_connection->GetEstablisedConnections(), + RepeatedFieldBackInserter( + current_strategy_session_->mutable_established_connection())); + } + active_connections_.clear(); + // Add the StrategySession in ClientSession current_strategy_session_->set_duration_millis(absl::ToInt64Milliseconds( started_strategy_session_time_ - SystemClock::ElapsedRealtime())); @@ -621,7 +676,8 @@ void AnalyticsRecorder::LogicalConnection::PhysicalConnectionClosed( " opened."; return; } - auto *established_connection = it->second.get(); + ConnectionsLog::EstablishedConnection *established_connection = + it->second.get(); if (established_connection->has_disconnection_reason()) { NEARBY_LOGS(WARNING) << "Unexpected call to physicalConnectionClosed() for medium " @@ -677,7 +733,7 @@ void AnalyticsRecorder::LogicalConnection::ChunkReceived( if (it == incoming_payloads_.end()) { return; } - auto *pending_payload = it->second.get(); + PendingPayload *pending_payload = it->second.get(); pending_payload->AddChunk(size_bytes); } @@ -690,7 +746,8 @@ void AnalyticsRecorder::LogicalConnection::IncomingPayloadDone( } auto it = physical_connections_.find(current_medium_); if (it != physical_connections_.end()) { - const auto &established_connection = it->second; + const std::unique_ptr + &established_connection = it->second; auto it = incoming_payloads_.find(payload_id); if (it != incoming_payloads_.end()) { *established_connection->add_received_payload() = @@ -712,7 +769,7 @@ void AnalyticsRecorder::LogicalConnection::ChunkSent(std::int64_t payload_id, if (it == outgoing_payloads_.end()) { return; } - auto *payload = it->second.get(); + PendingPayload *payload = it->second.get(); payload->AddChunk(size_bytes); } @@ -725,7 +782,8 @@ void AnalyticsRecorder::LogicalConnection::OutgoingPayloadDone( } auto it = physical_connections_.find(current_medium_); if (it != physical_connections_.end()) { - const auto &established_connection = it->second; + const std::unique_ptr + &established_connection = it->second; auto it = outgoing_payloads_.find(payload_id); if (it != outgoing_payloads_.end()) { *established_connection->add_sent_payload() = diff --git a/cpp/analytics/analytics_recorder.h b/cpp/analytics/analytics_recorder.h index b072c9e8..2471329b 100644 --- a/cpp/analytics/analytics_recorder.h +++ b/cpp/analytics/analytics_recorder.h @@ -83,6 +83,17 @@ class AnalyticsRecorder { absl::Duration duration, const std::string &connection_token) ABSL_LOCKS_EXCLUDED(mutex_); + // Connection established + void OnConnectionEstablished( + const std::string &endpoint_id, + location::nearby::proto::connections::Medium medium, + const std::string &connection_token) ABSL_LOCKS_EXCLUDED(mutex_); + void OnConnectionClosed( + const std::string &endpoint_id, + location::nearby::proto::connections::Medium medium, + location::nearby::proto::connections ::DisconnectionReason reason) + 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. @@ -130,9 +141,9 @@ class AnalyticsRecorder { LogicalConnection(const LogicalConnection &) = delete; LogicalConnection(LogicalConnection &&other) : current_medium_(std::move(other.current_medium_)), - physical_connections_{std::move(other.physical_connections_)}, - incoming_payloads_{std::move(other.incoming_payloads_)}, - outgoing_payloads_{std::move(other.outgoing_payloads_)} {} + physical_connections_(std::move(other.physical_connections_)), + incoming_payloads_(std::move(other.incoming_payloads_)), + outgoing_payloads_(std::move(other.outgoing_payloads_)) {} LogicalConnection &operator=(const LogicalConnection &) = delete; LogicalConnection &&operator=(LogicalConnection &&) = delete; ~LogicalConnection() = default; @@ -267,6 +278,8 @@ class AnalyticsRecorder { absl::btree_map> outgoing_connection_requests_ ABSL_GUARDED_BY(mutex_); + absl::btree_map> + active_connections_ ABSL_GUARDED_BY(mutex_); }; } // namespace analytics diff --git a/cpp/analytics/analytics_recorder_test.cc b/cpp/analytics/analytics_recorder_test.cc index cbb1075c..c55c28e1 100644 --- a/cpp/analytics/analytics_recorder_test.cc +++ b/cpp/analytics/analytics_recorder_test.cc @@ -30,6 +30,7 @@ namespace nearby { namespace analytics { namespace { +using ::location::nearby::analytics::proto::ConnectionsLog; using ::location::nearby::proto::connections::BLE; using ::location::nearby::proto::connections::BLUETOOTH; using ::location::nearby::proto::connections::CLIENT_SESSION; @@ -41,6 +42,8 @@ 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 ::location::nearby::proto::connections::UPGRADED; +using ::location::nearby::proto::connections::WIFI_LAN; using ::testing::Contains; using ::testing::EqualsProto; using ::testing::proto::Partially; @@ -52,7 +55,7 @@ class FakeEventLogger : public EventLogger { explicit FakeEventLogger(CountDownLatch& client_session_done_latch) : client_session_done_latch_(client_session_done_latch) {} - void Log(const proto::ConnectionsLog& connections_log) override { + void Log(const ConnectionsLog& connections_log) override { EventType event_type = connections_log.event_type(); logged_event_types_.push_back(event_type); if (event_type == CLIENT_SESSION) { @@ -68,7 +71,7 @@ class FakeEventLogger : public EventLogger { return logged_client_session_count_; } - const proto::ConnectionsLog::ClientSession& GetLoggedClientSession() { + const ConnectionsLog::ClientSession& GetLoggedClientSession() { return logged_client_session_; } @@ -77,7 +80,7 @@ class FakeEventLogger : public EventLogger { private: int logged_client_session_count_ = 0; CountDownLatch& client_session_done_latch_; - proto::ConnectionsLog::ClientSession logged_client_session_; + ConnectionsLog::ClientSession logged_client_session_; std::vector logged_event_types_; }; @@ -197,10 +200,10 @@ TEST(AnalyticsRecorderTest, TEST(AnalyticsRecorderTest, AdvertiserConnectionRequestsWorks) { 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"); - std::string endpoint_id_3("endpoint_id_3"); + std::string endpoint_id_0 = "endpoint_id_0"; + std::string endpoint_id_1 = "endpoint_id_1"; + std::string endpoint_id_2 = "endpoint_id_2"; + std::string endpoint_id_3 = "endpoint_id_3"; CountDownLatch client_session_done_latch(1); FakeEventLogger event_logger(client_session_done_latch); @@ -257,10 +260,10 @@ TEST(AnalyticsRecorderTest, AdvertiserConnectionRequestsWorks) { 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"); - std::string endpoint_id_3("endpoint_id_3"); + std::string endpoint_id_0 = "endpoint_id_0"; + std::string endpoint_id_1 = "endpoint_id_1"; + std::string endpoint_id_2 = "endpoint_id_2"; + std::string endpoint_id_3 = "endpoint_id_3"; CountDownLatch client_session_done_latch(1); FakeEventLogger event_logger(client_session_done_latch); @@ -318,9 +321,9 @@ TEST(AnalyticsRecorderTest, AdvertiserUnfinishedConnectionRequestsIncludedAsIgnored) { 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"); + std::string endpoint_id_0 = "endpoint_id_0"; + std::string endpoint_id_1 = "endpoint_id_1"; + std::string endpoint_id_2 = "endpoint_id_2"; CountDownLatch client_session_done_latch(1); FakeEventLogger event_logger(client_session_done_latch); @@ -368,9 +371,9 @@ TEST(AnalyticsRecorderTest, DiscovererUnfinishedConnectionRequestsIncludedAsIgnored) { 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"); + std::string endpoint_id_0 = "endpoint_id_0"; + std::string endpoint_id_1 = "endpoint_id_1"; + std::string endpoint_id_2 = "endpoint_id_2"; CountDownLatch client_session_done_latch(1); FakeEventLogger event_logger(client_session_done_latch); @@ -418,8 +421,8 @@ TEST(AnalyticsRecorderTest, TEST(AnalyticsRecorderTest, SuccessfulIncomingConnectionAttempt) { connections::Strategy strategy = connections::Strategy::kP2pStar; std::vector mediums = {BLE, BLUETOOTH}; - std::string endpoint_id("endpoint_id"); - std::string connection_token(""); + std::string endpoint_id = "endpoint_id"; + std::string connection_token = ""; CountDownLatch client_session_done_latch(1); FakeEventLogger event_logger(client_session_done_latch); @@ -452,8 +455,8 @@ TEST(AnalyticsRecorderTest, FailedConnectionAttemptUpdatesConnectionRequestNotSent) { connections::Strategy strategy = connections::Strategy::kP2pStar; std::vector mediums = {BLE, BLUETOOTH}; - std::string endpoint_id("endpoint_id"); - std::string connection_token(""); + std::string endpoint_id = "endpoint_id"; + std::string connection_token = ""; CountDownLatch client_session_done_latch(1); FakeEventLogger event_logger(client_session_done_latch); @@ -490,6 +493,44 @@ TEST(AnalyticsRecorderTest, >)pb"))); } +TEST(AnalyticsRecorderTest, UnfinishedEstablishedConnectionsAddedAsUnfinished) { + connections::Strategy strategy = connections::Strategy::kP2pStar; + std::vector mediums = {BLE, BLUETOOTH}; + std::string endpoint_id = "endpoint_id"; + std::string connection_token = "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.OnConnectionEstablished(endpoint_id, BLUETOOTH, + connection_token); + analytics_recorder.OnConnectionClosed(endpoint_id, BLUETOOTH, UPGRADED); + analytics_recorder.OnConnectionEstablished(endpoint_id, WIFI_LAN, + 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: ADVERTISER + advertising_phase < medium: BLE medium: BLUETOOTH > + established_connection < + medium: BLUETOOTH + disconnection_reason: UPGRADED + connection_token: "connection_token" + > + established_connection < + medium: WIFI_LAN + disconnection_reason: UNFINISHED + connection_token: "connection_token" + > + >)pb"))); +} + } // namespace } // namespace analytics } // namespace nearby diff --git a/cpp/core/internal/base_endpoint_channel.cc b/cpp/core/internal/base_endpoint_channel.cc index a6b276be..23e7e476 100644 --- a/cpp/core/internal/base_endpoint_channel.cc +++ b/cpp/core/internal/base_endpoint_channel.cc @@ -256,11 +256,22 @@ void BaseEndpointChannel::CloseIo() { } } +void BaseEndpointChannel::SetAnalyticsRecorder( + analytics::AnalyticsRecorder* analytics_recorder, + const std::string& endpoint_id) { + analytics_recorder_ = analytics_recorder; + endpoint_id_ = endpoint_id; +} + void BaseEndpointChannel::Close( proto::connections::DisconnectionReason reason) { NEARBY_LOGS(INFO) << __func__ << ": Closing endpoint channel, reason: " << reason; Close(); + + if (analytics_recorder_ != nullptr && !endpoint_id_.empty()) { + analytics_recorder_->OnConnectionClosed(endpoint_id_, GetMedium(), reason); + } } std::string BaseEndpointChannel::GetType() const { diff --git a/cpp/core/internal/base_endpoint_channel.h b/cpp/core/internal/base_endpoint_channel.h index 9def7378..aaab6760 100644 --- a/cpp/core/internal/base_endpoint_channel.h +++ b/cpp/core/internal/base_endpoint_channel.h @@ -21,6 +21,7 @@ #include "securegcm/d2d_connection_context_v1.h" #include "absl/base/thread_annotations.h" +#include "third_party/nearby_connections/cpp/analytics/analytics_recorder.h" #include "core/internal/endpoint_channel.h" #include "platform/base/byte_array.h" #include "platform/base/input_stream.h" @@ -88,6 +89,9 @@ class BaseEndpointChannel : public EndpointChannel { absl::Time GetLastReadTimestamp() const ABSL_LOCKS_EXCLUDED(last_read_mutex_) override; + void SetAnalyticsRecorder(analytics::AnalyticsRecorder* analytics_recorder, + const std::string& endpoint_id) override; + protected: virtual void CloseImpl() = 0; @@ -128,6 +132,9 @@ class BaseEndpointChannel : public EndpointChannel { ConditionVariable is_paused_cond_{&is_paused_mutex_}; // If true, writes should block until this has been set to false. bool is_paused_ ABSL_GUARDED_BY(is_paused_mutex_) = false; + + analytics::AnalyticsRecorder* analytics_recorder_ = nullptr; + std::string endpoint_id_ = ""; }; } // namespace connections diff --git a/cpp/core/internal/base_pcp_handler.cc b/cpp/core/internal/base_pcp_handler.cc index 44c1127f..031fbfd6 100644 --- a/cpp/core/internal/base_pcp_handler.cc +++ b/cpp/core/internal/base_pcp_handler.cc @@ -1377,6 +1377,11 @@ void BasePcpHandler::EvaluateConnectionResult(ClientProxy* client, channel_manager_->EncryptChannelForEndpoint(endpoint_id, std::move(context)); + + client->GetAnalyticsRecorder().OnConnectionEstablished( + endpoint_id, + channel_manager_->GetChannelForEndpoint(endpoint_id)->GetMedium(), + connection_info.connection_token); } else { NEARBY_LOGS(INFO) << "Pending connection rejected; endpoint_id=" << endpoint_id; diff --git a/cpp/core/internal/bwu_manager.cc b/cpp/core/internal/bwu_manager.cc index 27105c9a..a870fd1d 100644 --- a/cpp/core/internal/bwu_manager.cc +++ b/cpp/core/internal/bwu_manager.cc @@ -866,6 +866,11 @@ void BwuManager::ProcessSafeToClosePriorChannelEvent( << " EndpointChannel to conclude upgrade protocol for endpoint " << endpoint_id; + // Now the upgrade protocol has completed, record analytics for this new + // upgraded bandwidth connection... + client->GetAnalyticsRecorder().OnConnectionEstablished( + endpoint_id, medium_, client->GetConnectionToken(endpoint_id)); + // Now that the old channel has been drained, we can unpause the new channel std::shared_ptr channel = channel_manager_->GetChannelForEndpoint(endpoint_id); diff --git a/cpp/core/internal/encryption_runner_test.cc b/cpp/core/internal/encryption_runner_test.cc index fe7215a7..7945c0a5 100644 --- a/cpp/core/internal/encryption_runner_test.cc +++ b/cpp/core/internal/encryption_runner_test.cc @@ -61,6 +61,8 @@ class FakeEndpointChannel : public EndpointChannel { void Pause() override {} void Resume() override {} absl::Time GetLastReadTimestamp() const override { return read_timestamp_; } + void SetAnalyticsRecorder(analytics::AnalyticsRecorder* analytics_recorder, + const std::string& endpoint_id) override {} private: InputStream* in_ = nullptr; diff --git a/cpp/core/internal/endpoint_channel.h b/cpp/core/internal/endpoint_channel.h index 108fcab7..7ff8602a 100644 --- a/cpp/core/internal/endpoint_channel.h +++ b/cpp/core/internal/endpoint_channel.h @@ -20,6 +20,7 @@ #include "securegcm/d2d_connection_context_v1.h" #include "absl/time/clock.h" +#include "third_party/nearby_connections/cpp/analytics/analytics_recorder.h" #include "platform/base/byte_array.h" #include "platform/base/exception.h" #include "platform/public/mutex.h" @@ -79,6 +80,11 @@ class EndpointChannel { // Returns the timestamp of the last read from this endpoint, or -1 if no // reads have occurred. virtual absl::Time GetLastReadTimestamp() const = 0; + + // Sets the AnalyticsRecorder instance for analytics. + virtual void SetAnalyticsRecorder( + analytics::AnalyticsRecorder* analytics_recorder, + const std::string& endpoint_id) = 0; }; inline bool operator==(const EndpointChannel& lhs, const EndpointChannel& rhs) { diff --git a/cpp/core/internal/endpoint_channel_manager.cc b/cpp/core/internal/endpoint_channel_manager.cc index 9ea41dc5..55cead7e 100644 --- a/cpp/core/internal/endpoint_channel_manager.cc +++ b/cpp/core/internal/endpoint_channel_manager.cc @@ -98,6 +98,7 @@ void EndpointChannelManager::SetActiveEndpointChannel( std::unique_ptr channel) { // Update the channel first, then encrypt this new channel, if // crypto context is present. + channel->SetAnalyticsRecorder(&client->GetAnalyticsRecorder(), endpoint_id); channel_state_.UpdateChannelForEndpoint(endpoint_id, std::move(channel)); auto* endpoint = channel_state_.LookupEndpointData(endpoint_id); diff --git a/cpp/core/internal/endpoint_manager_test.cc b/cpp/core/internal/endpoint_manager_test.cc index f08468f9..04d2392e 100644 --- a/cpp/core/internal/endpoint_manager_test.cc +++ b/cpp/core/internal/endpoint_manager_test.cc @@ -64,6 +64,8 @@ class MockEndpointChannel : public EndpointChannel { MOCK_METHOD(void, Pause, (), (override)); MOCK_METHOD(void, Resume, (), (override)); MOCK_METHOD(absl::Time, GetLastReadTimestamp, (), (const override)); + MOCK_METHOD(void, SetAnalyticsRecorder, + (analytics::AnalyticsRecorder*, const std::string&), (override)); bool IsClosed() const { absl::MutexLock lock(&mutex_);