analytics: 3p NC: Implement EstablishedConnections.

PiperOrigin-RevId: 394582833

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<proto::ConnectionsLog::AdvertisingPhase>();
+      absl::make_unique<ConnectionsLog::AdvertisingPhase>();
   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<ConnectionsLog::DiscoveryPhase>();
-  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<LogicalConnection> &logical_connection = it->second;
+    logical_connection->PhysicalConnectionEstablished(medium, connection_token);
+  } else {
+    active_connections_.insert(
+        {endpoint_id,
+         absl::make_unique<LogicalConnection>(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<LogicalConnection> &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<LogicalConnection> &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<ConnectionsLog::EstablishedConnection>
+        &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<ConnectionsLog::EstablishedConnection>
+        &established_connection = it->second;
     auto it = outgoing_payloads_.find(payload_id);
     if (it != outgoing_payloads_.end()) {
       *established_connection->add_sent_payload() =

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<std::string,
                   std::unique_ptr<proto::ConnectionsLog::ConnectionRequest>>
       outgoing_connection_requests_ ABSL_GUARDED_BY(mutex_);
+  absl::btree_map<std::string, std::unique_ptr<LogicalConnection>>
+      active_connections_ ABSL_GUARDED_BY(mutex_);
 };
 
 }  // namespace analytics

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<EventType> logged_event_types_;
 };
 
@@ -197,10 +200,10 @@ TEST(AnalyticsRecorderTest,
 TEST(AnalyticsRecorderTest, AdvertiserConnectionRequestsWorks) {
   connections::Strategy strategy = connections::Strategy::kP2pStar;
   std::vector<Medium> 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<Medium> 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<Medium> 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<Medium> 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<Medium> 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<Medium> 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<Medium> 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

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 {

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

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;

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<EndpointChannel> channel =
       channel_manager_->GetChannelForEndpoint(endpoint_id);

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;

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) {

cpp/core/internal/endpoint_channel_manager.cc

@@ -98,6 +98,7 @@ void EndpointChannelManager::SetActiveEndpointChannel(
     std::unique_ptr<EndpointChannel> 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);

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_);