analytics: 3p NC: Implement Payload.

PiperOrigin-RevId: 394695112

cpp/analytics/analytics_recorder.cc

@@ -31,6 +31,7 @@ namespace analytics {
 using ::location::nearby::analytics::proto::ConnectionsLog;
 using ::location::nearby::proto::connections::ACCEPTED;
 using ::location::nearby::proto::connections::ADVERTISER;
+using ::location::nearby::proto::connections::BYTES;
 using ::location::nearby::proto::connections::CLIENT_SESSION;
 using ::location::nearby::proto::connections::CONNECTION_CLOSED;
 using ::location::nearby::proto::connections::ConnectionAttemptResult;
@@ -40,6 +41,7 @@ using ::location::nearby::proto::connections::ConnectionsStrategy;
 using ::location::nearby::proto::connections::DisconnectionReason;
 using ::location::nearby::proto::connections::DISCOVERER;
 using ::location::nearby::proto::connections::EventType;
+using ::location::nearby::proto::connections::FILE;
 using ::location::nearby::proto::connections::IGNORED;
 using ::location::nearby::proto::connections::INCOMING;
 using ::location::nearby::proto::connections::INITIAL;
@@ -59,8 +61,10 @@ 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::STREAM;
 using ::location::nearby::proto::connections::UNFINISHED;
 using ::location::nearby::proto::connections::UNKNOWN_MEDIUM;
+using ::location::nearby::proto::connections::UNKNOWN_PAYLOAD_TYPE;
 using ::location::nearby::proto::connections::UNKNOWN_STRATEGY;
 using ::location::nearby::proto::connections::UPGRADED;
 
@@ -333,6 +337,99 @@ void AnalyticsRecorder::OnConnectionClosed(const std::string &endpoint_id,
             current_strategy_session_->mutable_established_connection()));
   }
 }
+void AnalyticsRecorder::OnIncomingPayloadStarted(
+    const std::string &endpoint_id, std::int64_t payload_id,
+    connections::Payload::Type type, std::int64_t total_size_bytes) {
+  MutexLock lock(&mutex_);
+  if (!CanRecordAnalyticsLocked("OnIncomingPayloadStarted")) {
+    return;
+  }
+  auto it = active_connections_.find(endpoint_id);
+  if (it == active_connections_.end()) {
+    return;
+  }
+  std::unique_ptr<LogicalConnection> &logical_connection = it->second;
+  logical_connection->IncomingPayloadStarted(
+      payload_id, PayloadTypeToProtoPayloadType(type), total_size_bytes);
+}
+
+void AnalyticsRecorder::OnPayloadChunkReceived(const std::string &endpoint_id,
+                                               std::int64_t payload_id,
+                                               std::int64_t chunk_size_bytes) {
+  MutexLock lock(&mutex_);
+  if (!CanRecordAnalyticsLocked("OnPayloadChunkReceived")) {
+    return;
+  }
+  auto it = active_connections_.find(endpoint_id);
+  if (it == active_connections_.end()) {
+    return;
+  }
+  std::unique_ptr<LogicalConnection> &logical_connection = it->second;
+  logical_connection->ChunkReceived(payload_id, chunk_size_bytes);
+}
+
+void AnalyticsRecorder::OnIncomingPayloadDone(const std::string &endpoint_id,
+                                              std::int64_t payload_id,
+                                              PayloadStatus status) {
+  MutexLock lock(&mutex_);
+  if (!CanRecordAnalyticsLocked("OnIncomingPayloadDone")) {
+    return;
+  }
+  auto it = active_connections_.find(endpoint_id);
+  if (it == active_connections_.end()) {
+    return;
+  }
+  std::unique_ptr<LogicalConnection> &logical_connection = it->second;
+  logical_connection->IncomingPayloadDone(payload_id, status);
+}
+
+void AnalyticsRecorder::OnOutgoingPayloadStarted(
+    const std::vector<std::string> &endpoint_ids, std::int64_t payload_id,
+    connections::Payload::Type type, std::int64_t total_size_bytes) {
+  MutexLock lock(&mutex_);
+  if (!CanRecordAnalyticsLocked("OnOutgoingPayloadStarted")) {
+    return;
+  }
+  for (const auto &endpoint_id : endpoint_ids) {
+    auto it = active_connections_.find(endpoint_id);
+    if (it == active_connections_.end()) {
+      continue;
+    }
+    std::unique_ptr<LogicalConnection> &logical_connection = it->second;
+    logical_connection->OutgoingPayloadStarted(
+        payload_id, PayloadTypeToProtoPayloadType(type), total_size_bytes);
+  }
+}
+
+void AnalyticsRecorder::OnPayloadChunkSent(const std::string &endpoint_id,
+                                           std::int64_t payload_id,
+                                           std::int64_t chunk_size_bytes) {
+  MutexLock lock(&mutex_);
+  if (!CanRecordAnalyticsLocked("OnPayloadChunkSent")) {
+    return;
+  }
+  auto it = active_connections_.find(endpoint_id);
+  if (it == active_connections_.end()) {
+    return;
+  }
+  std::unique_ptr<LogicalConnection> &logical_connection = it->second;
+  logical_connection->ChunkSent(payload_id, chunk_size_bytes);
+}
+
+void AnalyticsRecorder::OnOutgoingPayloadDone(const std::string &endpoint_id,
+                                              std::int64_t payload_id,
+                                              PayloadStatus status) {
+  MutexLock lock(&mutex_);
+  if (!CanRecordAnalyticsLocked("OnOutgoingPayloadDone")) {
+    return;
+  }
+  auto it = active_connections_.find(endpoint_id);
+  if (it == active_connections_.end()) {
+    return;
+  }
+  std::unique_ptr<LogicalConnection> &logical_connection = it->second;
+  logical_connection->OutgoingPayloadDone(payload_id, status);
+}
 
 void AnalyticsRecorder::LogSession() {
   MutexLock lock(&mutex_);
@@ -368,17 +465,19 @@ bool AnalyticsRecorder::CanRecordAnalyticsLocked(
 }
 
 void AnalyticsRecorder::LogClientSession() {
-  serial_executor_.Execute("analytics-recorder", [this]() {
-    ConnectionsLog connections_log;
-    connections_log.set_event_type(CLIENT_SESSION);
-    connections_log.set_allocated_client_session(client_session_.release());
-    connections_log.set_version(kVersion);
+  serial_executor_.Execute(
+      "analytics-recorder", [this]() {
+        ConnectionsLog connections_log;
+        connections_log.set_event_type(CLIENT_SESSION);
+        connections_log.set_allocated_client_session(client_session_.release());
+        connections_log.set_version(kVersion);
 
-    NEARBY_LOGS(INFO) << "AnalyticsRecorder LogClientSession connections_log="
-                      << connections_log.DebugString();
+        NEARBY_LOGS(VERBOSE)
+            << "AnalyticsRecorder LogClientSession connections_log="
+            << connections_log.DebugString();
 
-    event_logger_->Log(connections_log);
-  });
+        event_logger_->Log(connections_log);
+      });
 }
 
 void AnalyticsRecorder::LogEvent(EventType event_type) {
@@ -615,6 +714,20 @@ ConnectionsStrategy AnalyticsRecorder::StrategyToConnectionStrategy(
   return UNKNOWN_STRATEGY;
 }
 
+PayloadType AnalyticsRecorder::PayloadTypeToProtoPayloadType(
+    connections::Payload::Type type) {
+  switch (type) {
+    case connections::Payload::Type::kBytes:
+      return BYTES;
+    case connections::Payload::Type::kFile:
+      return FILE;
+    case connections::Payload::Type::kStream:
+      return STREAM;
+    default:
+      return UNKNOWN_PAYLOAD_TYPE;
+  }
+}
+
 void AnalyticsRecorder::PendingPayload::AddChunk(
     std::int64_t chunk_size_bytes) {
   num_bytes_transferred_ += chunk_size_bytes;

cpp/analytics/analytics_recorder.h

@@ -94,6 +94,34 @@ class AnalyticsRecorder {
       location::nearby::proto::connections ::DisconnectionReason reason)
       ABSL_LOCKS_EXCLUDED(mutex_);
 
+  // Payload
+  void OnIncomingPayloadStarted(const std::string &endpoint_id,
+                                std::int64_t payload_id,
+                                connections::Payload::Type type,
+                                std::int64_t total_size_bytes)
+      ABSL_LOCKS_EXCLUDED(mutex_);
+  void OnPayloadChunkReceived(const std::string &endpoint_id,
+                              std::int64_t payload_id,
+                              std::int64_t chunk_size_bytes)
+      ABSL_LOCKS_EXCLUDED(mutex_);
+  void OnIncomingPayloadDone(
+      const std::string &endpoint_id, std::int64_t payload_id,
+      ::location::nearby::proto::connections::PayloadStatus status)
+      ABSL_LOCKS_EXCLUDED(mutex_);
+  void OnOutgoingPayloadStarted(const std::vector<std::string> &endpoint_ids,
+                                std::int64_t payload_id,
+                                connections::Payload::Type type,
+                                std::int64_t total_size_bytes)
+      ABSL_LOCKS_EXCLUDED(mutex_);
+  void OnPayloadChunkSent(const std::string &endpoint_id,
+                          std::int64_t payload_id,
+                          std::int64_t chunk_size_bytes)
+      ABSL_LOCKS_EXCLUDED(mutex_);
+  void OnOutgoingPayloadDone(
+      const std::string &endpoint_id, std::int64_t payload_id,
+      ::location::nearby::proto::connections::PayloadStatus status)
+      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.
@@ -240,6 +268,8 @@ class AnalyticsRecorder {
 
   location::nearby::proto::connections::ConnectionsStrategy
   StrategyToConnectionStrategy(connections::Strategy strategy);
+  ::location::nearby::proto::connections::PayloadType
+  PayloadTypeToProtoPayloadType(connections::Payload::Type type);
 
   // Not owned by AnalyticsRecorder. Pointer must refer to a valid object
   // that outlives the one constructed.

cpp/analytics/analytics_recorder_test.cc

@@ -36,12 +36,14 @@ using ::location::nearby::proto::connections::BLUETOOTH;
 using ::location::nearby::proto::connections::CLIENT_SESSION;
 using ::location::nearby::proto::connections::EventType;
 using ::location::nearby::proto::connections::INITIAL;
+using ::location::nearby::proto::connections::LOCAL_DISCONNECTION;
 using ::location::nearby::proto::connections::Medium;
 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 ::location::nearby::proto::connections::SUCCESS;
 using ::location::nearby::proto::connections::UPGRADED;
 using ::location::nearby::proto::connections::WIFI_LAN;
 using ::testing::Contains;
@@ -531,6 +533,69 @@ TEST(AnalyticsRecorderTest, UnfinishedEstablishedConnectionsAddedAsUnfinished) {
                 >)pb")));
 }
 
+TEST(AnalyticsRecorderTest, OutgoingPayloadUpgraded) {
+  connections::Strategy strategy = connections::Strategy::kP2pStar;
+  std::vector<Medium> mediums = {BLE, BLUETOOTH};
+  std::string endpoint_id("endpoint_id");
+  std::int64_t payload_id(123456789);
+  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.OnOutgoingPayloadStarted(
+      {endpoint_id}, payload_id, connections::Payload::Type::kFile, 50);
+  analytics_recorder.OnPayloadChunkSent(endpoint_id, payload_id, 10);
+  analytics_recorder.OnPayloadChunkSent(endpoint_id, payload_id, 10);
+  analytics_recorder.OnConnectionClosed(endpoint_id, BLUETOOTH, UPGRADED);
+  analytics_recorder.OnConnectionEstablished(endpoint_id, WIFI_LAN,
+                                             connection_token);
+  analytics_recorder.OnPayloadChunkSent(endpoint_id, payload_id, 10);
+  analytics_recorder.OnPayloadChunkSent(endpoint_id, payload_id, 10);
+  analytics_recorder.OnPayloadChunkSent(endpoint_id, payload_id, 10);
+  analytics_recorder.OnOutgoingPayloadDone(endpoint_id, payload_id, SUCCESS);
+  analytics_recorder.OnConnectionClosed(endpoint_id, WIFI_LAN,
+                                        LOCAL_DISCONNECTION);
+
+  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
+                    sent_payload <
+                      type: FILE
+                      total_size_bytes: 50
+                      num_bytes_transferred: 20
+                      num_chunks: 2
+                      status: MOVED_TO_NEW_MEDIUM
+                    >
+                    disconnection_reason: UPGRADED
+                    connection_token: "connection_token"
+                  >
+                  established_connection <
+                    medium: WIFI_LAN
+                    sent_payload <
+                      type: FILE
+                      total_size_bytes: 50
+                      num_bytes_transferred: 30
+                      num_chunks: 3
+                      status: SUCCESS
+                    >
+                    disconnection_reason: LOCAL_DISCONNECTION
+                    connection_token: "connection_token"
+                  >
+                >)pb")));
+}
+
 }  // namespace
 }  // namespace analytics
 }  // namespace nearby

cpp/core/internal/payload_manager.cc

@@ -351,11 +351,29 @@ void PayloadManager::SendPayload(ClientProxy* client,
   if (shutdown_.Get()) return;
   NEARBY_LOG(INFO, "SendPayload: endpoint_ids={%s}",
              ToString(endpoint_ids).c_str());
+  // Before transfer to internal payload, retrieves the Payload size for
+  // analytics.
+  std::int64_t payload_total_size;
+  switch (payload.GetType()) {
+    case connections::Payload::Type::kBytes:
+      payload_total_size = payload.AsBytes().size();
+      break;
+    case connections::Payload::Type::kFile:
+      payload_total_size = payload.AsFile()->GetTotalSize();
+      break;
+    case connections::Payload::Type::kStream:
+    case connections::Payload::Type::kUnknown:
+      payload_total_size = -1;
+      break;
+  }
   auto executor = GetOutgoingPayloadExecutor(payload.GetType());
   // The |executor| will be null if the payload is of a type we cannot work
   // with. This should never be reached since the ServiceControllerRouter has
   // already checked whether or not we can work with this Payload type.
   if (!executor) {
+    RecordInvalidPayloadAnalytics(client, endpoint_ids, payload.GetId(),
+                                  payload.GetType(), payload.GetOffset(),
+                                  payload_total_size);
     NEARBY_LOGS(INFO)
         << "PayloadManager failed to determine the right executor for "
            "outgoing payload_id="
@@ -376,11 +394,14 @@ void PayloadManager::SendPayload(ClientProxy* client,
   Payload::Id payload_id =
       CreateOutgoingPayload(std::move(payload), endpoint_ids);
   executor->Execute(
-      "send-payload",
-      [this, client, endpoint_ids, payload_id, payload_type, resume_offset]() {
+      "send-payload", [this, client, endpoint_ids, payload_id, payload_type,
+                       resume_offset, payload_total_size]() {
         if (shutdown_.Get()) return;
         PendingPayload* pending_payload = GetPayload(payload_id);
         if (!pending_payload) {
+          RecordInvalidPayloadAnalytics(client, endpoint_ids, payload_id,
+                                        payload_type, resume_offset,
+                                        payload_total_size);
           NEARBY_LOGS(INFO)
               << "PayloadManager failed to create InternalPayload for outgoing "
                  "payload_id="
@@ -390,6 +411,11 @@ void PayloadManager::SendPayload(ClientProxy* client,
         }
         auto* internal_payload = pending_payload->GetInternalPayload();
         if (!internal_payload) return;
+
+        RecordPayloadStartedAnalytics(client, endpoint_ids, payload_id,
+                                      payload_type, resume_offset,
+                                      internal_payload->GetTotalSize());
+
         PayloadTransferFrame::PayloadHeader payload_header{
             CreatePayloadHeader(*internal_payload, resume_offset)};
         bool should_continue = true;
@@ -501,6 +527,16 @@ void PayloadManager::OnEndpointDisconnect(ClientProxy* client,
 
               // Send a client notification of a payload transfer failure.
               client->OnPayloadProgress(endpoint_id, update);
+
+              if (pending_payload->IsIncoming()) {
+                client->GetAnalyticsRecorder().OnIncomingPayloadDone(
+                    endpoint_id, pending_payload->GetId(),
+                    proto::connections::ENDPOINT_IO_ERROR);
+              } else {
+                client->GetAnalyticsRecorder().OnOutgoingPayloadDone(
+                    endpoint_id, pending_payload->GetId(),
+                    proto::connections::ENDPOINT_IO_ERROR);
+              }
             }
 
             barrier.CountDown();
@@ -650,6 +686,10 @@ void PayloadManager::SendClientCallbacksForFinishedOutgoingPayload(
 
           // Notify the client.
           client->OnPayloadProgress(endpoint_id, update);
+
+          // Mark this payload as done for analytics.
+          client->GetAnalyticsRecorder().OnOutgoingPayloadDone(
+              endpoint_id, payload_header.id(), status);
         }
 
         // Remove these endpoints from our tracking list for this payload.
@@ -684,6 +724,10 @@ void PayloadManager::SendClientCallbacksForFinishedIncomingPayload(
             payload_header.total_size(), offset_bytes};
         NotifyClientOfIncomingPayloadProgressInfo(client, endpoint_id, update);
         DestroyPendingPayload(payload_header.id());
+
+        // Analyze
+        client->GetAnalyticsRecorder().OnIncomingPayloadDone(
+            endpoint_id, payload_header.id(), status);
       });
 }
 
@@ -806,6 +850,9 @@ void PayloadManager::HandleSuccessfulOutgoingChunk(
         client->OnPayloadProgress(endpoint_id, update);
 
         if (is_last_chunk) {
+          client->GetAnalyticsRecorder().OnOutgoingPayloadDone(
+              endpoint_id, payload_header.id(), proto::connections::SUCCESS);
+
           // Stop tracking this endpoint.
           pending_payload->RemoveEndpoints({endpoint_id});
 
@@ -813,6 +860,9 @@ void PayloadManager::HandleSuccessfulOutgoingChunk(
           if (pending_payload->GetEndpoints().empty()) {
             pending_payload->Close();
           }
+        } else {
+          client->GetAnalyticsRecorder().OnPayloadChunkSent(
+              endpoint_id, payload_header.id(), payload_chunk_body_size);
         }
       });
 }
@@ -864,6 +914,15 @@ void PayloadManager::HandleSuccessfulIncomingChunk(
 
         // Notify the client of this update.
         NotifyClientOfIncomingPayloadProgressInfo(client, endpoint_id, update);
+
+        // Analyze the success.
+        if (is_last_chunk) {
+          client->GetAnalyticsRecorder().OnIncomingPayloadDone(
+              endpoint_id, payload_header.id(), proto::connections::SUCCESS);
+        } else {
+          client->GetAnalyticsRecorder().OnPayloadChunkReceived(
+              endpoint_id, payload_header.id(), payload_chunk_body_size);
+        }
       });
 }
 
@@ -882,6 +941,17 @@ void PayloadManager::ProcessDataPacket(
 
   PendingPayload* pending_payload;
   if (payload_chunk.offset() == 0) {
+    RunOnStatusUpdateThread(
+        "process-data-packet", [to_client, from_endpoint_id, payload_header,
+                                this]() RUN_ON_PAYLOAD_STATUS_UPDATE_THREAD() {
+          // This is the first chunk of a new incoming
+          // payload. Start the analysis.
+          to_client->GetAnalyticsRecorder().OnIncomingPayloadStarted(
+              from_endpoint_id, payload_header.id(),
+              FramePayloadTypeToPayloadType(payload_header.type()),
+              payload_header.total_size());
+        });
+
     pending_payload =
         CreateIncomingPayload(payload_transfer_frame, from_endpoint_id);
     if (!pending_payload) {
@@ -1026,6 +1096,42 @@ void PayloadManager::NotifyClientOfIncomingPayloadProgressInfo(
   client->OnPayloadProgress(endpoint_id, payload_transfer_update);
 }
 
+void PayloadManager::RecordPayloadStartedAnalytics(
+    ClientProxy* client, const EndpointIds& endpoint_ids,
+    std::int64_t payload_id, Payload::Type payload_type, std::int64_t offset,
+    std::int64_t total_size) {
+  client->GetAnalyticsRecorder().OnOutgoingPayloadStarted(
+      endpoint_ids, payload_id, payload_type,
+      total_size == -1 ? -1 : total_size - offset);
+}
+
+void PayloadManager::RecordInvalidPayloadAnalytics(
+    ClientProxy* client, const EndpointIds& endpoint_ids,
+    std::int64_t payload_id, Payload::Type payload_type, std::int64_t offset,
+    std::int64_t total_size) {
+  RecordPayloadStartedAnalytics(client, endpoint_ids, payload_id, payload_type,
+                                offset, total_size);
+
+  for (const auto& endpoint_id : endpoint_ids) {
+    client->GetAnalyticsRecorder().OnOutgoingPayloadDone(
+        endpoint_id, payload_id, proto::connections::LOCAL_ERROR);
+  }
+}
+
+Payload::Type PayloadManager::FramePayloadTypeToPayloadType(
+    PayloadTransferFrame::PayloadHeader::PayloadType type) {
+  switch (type) {
+    case PayloadTransferFrame_PayloadHeader_PayloadType_BYTES:
+      return connections::Payload::Type::kBytes;
+    case PayloadTransferFrame_PayloadHeader_PayloadType_FILE:
+      return connections::Payload::Type::kFile;
+    case PayloadTransferFrame_PayloadHeader_PayloadType_STREAM:
+      return connections::Payload::Type::kStream;
+    default:
+      return connections::Payload::Type::kUnknown;
+  }
+}
+
 ///////////////////////////////// EndpointInfo /////////////////////////////////
 
 PayloadManager::EndpointInfo::Status

cpp/core/internal/payload_manager.h

@@ -286,6 +286,22 @@ class PayloadManager : public EndpointManager::FrameProcessor {
       ABSL_LOCKS_EXCLUDED(mutex_);
   void CancelAllPayloads() ABSL_LOCKS_EXCLUDED(mutex_);
 
+  void RecordPayloadStartedAnalytics(ClientProxy* client,
+                                     const EndpointIds& endpoint_ids,
+                                     std::int64_t payload_id,
+                                     Payload::Type payload_type,
+                                     std::int64_t offset,
+                                     std::int64_t total_size);
+  void RecordInvalidPayloadAnalytics(ClientProxy* client,
+                                     const EndpointIds& endpoint_ids,
+                                     std::int64_t payload_id,
+                                     Payload::Type payload_type,
+                                     std::int64_t offset,
+                                     std::int64_t total_size);
+
+  Payload::Type FramePayloadTypeToPayloadType(
+      PayloadTransferFrame::PayloadHeader::PayloadType type);
+
   mutable Mutex mutex_;
   AtomicBoolean shutdown_{false};
   std::unique_ptr<CountDownLatch> shutdown_barrier_;