From 24539d170f6ab7acaa225018244691e6db6c3af4 Mon Sep 17 00:00:00 2001 From: edwinwu Date: Fri, 3 Sep 2021 09:06:35 -0700 Subject: [PATCH] analytics: 3p NC: Implement Payload. PiperOrigin-RevId: 394695112 --- cpp/analytics/analytics_recorder.cc | 131 +++++++++++++++++++++-- cpp/analytics/analytics_recorder.h | 30 ++++++ cpp/analytics/analytics_recorder_test.cc | 65 +++++++++++ cpp/core/internal/payload_manager.cc | 110 ++++++++++++++++++- cpp/core/internal/payload_manager.h | 16 +++ 5 files changed, 341 insertions(+), 11 deletions(-) diff --git a/cpp/analytics/analytics_recorder.cc b/cpp/analytics/analytics_recorder.cc index cb957c32..39b99925 100644 --- a/cpp/analytics/analytics_recorder.cc +++ b/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 &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 &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 &logical_connection = it->second; + logical_connection->IncomingPayloadDone(payload_id, status); +} + +void AnalyticsRecorder::OnOutgoingPayloadStarted( + const std::vector &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 &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 &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 &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; diff --git a/cpp/analytics/analytics_recorder.h b/cpp/analytics/analytics_recorder.h index 2471329b..9d51897e 100644 --- a/cpp/analytics/analytics_recorder.h +++ b/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 &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. diff --git a/cpp/analytics/analytics_recorder_test.cc b/cpp/analytics/analytics_recorder_test.cc index c55c28e1..16b9822a 100644 --- a/cpp/analytics/analytics_recorder_test.cc +++ b/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 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 diff --git a/cpp/core/internal/payload_manager.cc b/cpp/core/internal/payload_manager.cc index fc3728b1..4c877edd 100644 --- a/cpp/core/internal/payload_manager.cc +++ b/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 diff --git a/cpp/core/internal/payload_manager.h b/cpp/core/internal/payload_manager.h index cc1d5926..4e605ab8 100644 --- a/cpp/core/internal/payload_manager.h +++ b/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 shutdown_barrier_;