From 1eda3dfc9a2a0fe336b98abbb32ef1ad3eacb494 Mon Sep 17 00:00:00 2001 From: edwinwu Date: Fri, 3 Sep 2021 20:19:45 -0700 Subject: [PATCH] analytics: 3p NC: Implement BandwidthUpgradeAttempt. PiperOrigin-RevId: 394802558 --- cpp/analytics/analytics_recorder.cc | 116 +++++++++++++++++++---- cpp/analytics/analytics_recorder.h | 32 ++++++- cpp/analytics/analytics_recorder_test.cc | 74 ++++++++++++++- cpp/core/internal/bwu_manager.cc | 80 +++++++++++++++- cpp/core/internal/bwu_manager.h | 3 + 5 files changed, 279 insertions(+), 26 deletions(-) diff --git a/cpp/analytics/analytics_recorder.cc b/cpp/analytics/analytics_recorder.cc index 39b99925..10602bac 100644 --- a/cpp/analytics/analytics_recorder.cc +++ b/cpp/analytics/analytics_recorder.cc @@ -31,9 +31,12 @@ 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::BandwidthUpgradeErrorStage; +using ::location::nearby::proto::connections::BandwidthUpgradeResult; 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::ConnectionAttemptDirection; using ::location::nearby::proto::connections::ConnectionAttemptResult; using ::location::nearby::proto::connections::ConnectionAttemptType; using ::location::nearby::proto::connections::ConnectionRequestResponse; @@ -63,9 +66,13 @@ 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::UNFINISHED_ERROR; 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::UPGRADE_RESULT_SUCCESS; +using ::location::nearby::proto::connections::UPGRADE_SUCCESS; +using ::location::nearby::proto::connections::UPGRADE_UNFINISHED; using ::location::nearby::proto::connections::UPGRADED; // These definitions are necessary before C++17. @@ -86,6 +93,7 @@ AnalyticsRecorder::~AnalyticsRecorder() { incoming_connection_requests_.clear(); outgoing_connection_requests_.clear(); active_connections_.clear(); + bandwidth_upgrade_attempts_.clear(); serial_executor_.Shutdown(); } @@ -174,8 +182,8 @@ void AnalyticsRecorder::OnConnectionRequestReceived( return; } absl::Time current_time = SystemClock::ElapsedRealtime(); - auto connection_request( - absl::make_unique()); + auto connection_request = + absl::make_unique(); connection_request->set_duration_millis(absl::ToUnixMillis(current_time)); connection_request->set_request_delay_millis(absl::ToInt64Milliseconds( current_time - started_advertising_phase_time_)); @@ -190,8 +198,8 @@ void AnalyticsRecorder::OnConnectionRequestSent( return; } absl::Time current_time = SystemClock::ElapsedRealtime(); - auto connection_request( - absl::make_unique()); + auto connection_request = + absl::make_unique(); connection_request->set_duration_millis(absl::ToUnixMillis(current_time)); connection_request->set_request_delay_millis( absl::ToInt64Milliseconds(current_time - started_discovery_phase_time_)); @@ -303,7 +311,7 @@ void AnalyticsRecorder::OnConnectionEstablished( } auto it = active_connections_.find(endpoint_id); if (it != active_connections_.end()) { - std::unique_ptr &logical_connection = it->second; + const std::unique_ptr &logical_connection = it->second; logical_connection->PhysicalConnectionEstablished(medium, connection_token); } else { active_connections_.insert( @@ -323,7 +331,7 @@ void AnalyticsRecorder::OnConnectionClosed(const std::string &endpoint_id, if (it == active_connections_.end()) { return; } - std::unique_ptr &logical_connection = it->second; + const 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 @@ -348,7 +356,7 @@ void AnalyticsRecorder::OnIncomingPayloadStarted( if (it == active_connections_.end()) { return; } - std::unique_ptr &logical_connection = it->second; + const std::unique_ptr &logical_connection = it->second; logical_connection->IncomingPayloadStarted( payload_id, PayloadTypeToProtoPayloadType(type), total_size_bytes); } @@ -364,7 +372,7 @@ void AnalyticsRecorder::OnPayloadChunkReceived(const std::string &endpoint_id, if (it == active_connections_.end()) { return; } - std::unique_ptr &logical_connection = it->second; + const std::unique_ptr &logical_connection = it->second; logical_connection->ChunkReceived(payload_id, chunk_size_bytes); } @@ -379,7 +387,7 @@ void AnalyticsRecorder::OnIncomingPayloadDone(const std::string &endpoint_id, if (it == active_connections_.end()) { return; } - std::unique_ptr &logical_connection = it->second; + const std::unique_ptr &logical_connection = it->second; logical_connection->IncomingPayloadDone(payload_id, status); } @@ -395,7 +403,7 @@ void AnalyticsRecorder::OnOutgoingPayloadStarted( if (it == active_connections_.end()) { continue; } - std::unique_ptr &logical_connection = it->second; + const std::unique_ptr &logical_connection = it->second; logical_connection->OutgoingPayloadStarted( payload_id, PayloadTypeToProtoPayloadType(type), total_size_bytes); } @@ -412,7 +420,7 @@ void AnalyticsRecorder::OnPayloadChunkSent(const std::string &endpoint_id, if (it == active_connections_.end()) { return; } - std::unique_ptr &logical_connection = it->second; + const std::unique_ptr &logical_connection = it->second; logical_connection->ChunkSent(payload_id, chunk_size_bytes); } @@ -427,10 +435,49 @@ void AnalyticsRecorder::OnOutgoingPayloadDone(const std::string &endpoint_id, if (it == active_connections_.end()) { return; } - std::unique_ptr &logical_connection = it->second; + const std::unique_ptr &logical_connection = it->second; logical_connection->OutgoingPayloadDone(payload_id, status); } +void AnalyticsRecorder::OnBandwidthUpgradeStarted( + const std::string &endpoint_id, Medium from_medium, Medium to_medium, + ConnectionAttemptDirection direction, const std::string &connection_token) { + MutexLock lock(&mutex_); + if (!CanRecordAnalyticsLocked("OnBandwidthUpgradeStarted")) { + return; + } + auto bandwidth_upgrade_attempt = + absl::make_unique(); + bandwidth_upgrade_attempt->set_duration_millis( + absl::ToUnixMillis(SystemClock::ElapsedRealtime())); + bandwidth_upgrade_attempt->set_from_medium(from_medium); + bandwidth_upgrade_attempt->set_to_medium(to_medium); + bandwidth_upgrade_attempt->set_direction(direction); + bandwidth_upgrade_attempt->set_connection_token(connection_token); + bandwidth_upgrade_attempts_.insert( + {endpoint_id, std::move(bandwidth_upgrade_attempt)}); +} + +void AnalyticsRecorder::OnBandwidthUpgradeError( + const std::string &endpoint_id, BandwidthUpgradeResult result, + BandwidthUpgradeErrorStage error_stage) { + MutexLock lock(&mutex_); + if (!CanRecordAnalyticsLocked("OnBandwidthUpgradeError")) { + return; + } + FinishUpgradeAttemptLocked(endpoint_id, result, error_stage); +} + +void AnalyticsRecorder::OnBandwidthUpgradeSuccess( + const std::string &endpoint_id) { + MutexLock lock(&mutex_); + if (!CanRecordAnalyticsLocked("OnBandwidthUpgradeSuccess")) { + return; + } + FinishUpgradeAttemptLocked(endpoint_id, UPGRADE_RESULT_SUCCESS, + UPGRADE_SUCCESS); +} + void AnalyticsRecorder::LogSession() { MutexLock lock(&mutex_); if (!CanRecordAnalyticsLocked("LogSession")) { @@ -545,7 +592,8 @@ void AnalyticsRecorder::FinishAdvertisingPhaseLocked() { for (const auto &item : incoming_connection_requests_) { // ConnectionRequests still pending have been ignored by the local or // remote (or both) endpoints. - const auto &connection_request = item.second; + const std::unique_ptr + &connection_request = item.second; MarkConnectionRequestIgnoredLocked(connection_request.get()); UpdateAdvertiserConnectionRequestLocked(connection_request.get()); } @@ -573,7 +621,8 @@ void AnalyticsRecorder::FinishDiscoveryPhaseLocked() { for (const auto &item : outgoing_connection_requests_) { // ConnectionRequests still pending have been ignored by the local or // remote (or both) endpoints. - const auto &connection_request = item.second; + const std::unique_ptr + &connection_request = item.second; MarkConnectionRequestIgnoredLocked(connection_request.get()); UpdateDiscovererConnectionRequestLocked(connection_request.get()); } @@ -675,6 +724,30 @@ void AnalyticsRecorder::MarkConnectionRequestIgnoredLocked( } } +void AnalyticsRecorder::FinishUpgradeAttemptLocked( + const std::string &endpoint_id, BandwidthUpgradeResult result, + BandwidthUpgradeErrorStage error_stage, bool erase_item) { + if (current_strategy_session_ == nullptr) { + NEARBY_LOGS(INFO) << "Unable to record upgrade attempt due to null " + "current_strategy_session_"; + return; + } + // Add the BandwidthUpgradeAttempt in the current StrategySession. + auto it = bandwidth_upgrade_attempts_.find(endpoint_id); + if (it != bandwidth_upgrade_attempts_.end()) { + ConnectionsLog::BandwidthUpgradeAttempt *attempt = it->second.get(); + attempt->set_duration_millis( + absl::ToUnixMillis(SystemClock::ElapsedRealtime()) - + attempt->duration_millis()); + attempt->set_error_stage(error_stage); + attempt->set_upgrade_result(result); + *current_strategy_session_->add_upgrade_attempt() = *attempt; + if (erase_item) { + bandwidth_upgrade_attempts_.erase(it); + } + } +} + void AnalyticsRecorder::FinishStrategySessionLocked() { if (current_strategy_session_ != nullptr) { FinishAdvertisingPhaseLocked(); @@ -682,7 +755,8 @@ void AnalyticsRecorder::FinishStrategySessionLocked() { // Finish any unfinished LogicalConnections. for (const auto &item : active_connections_) { - auto &logical_connection = item.second; + const std::unique_ptr &logical_connection = + item.second; logical_connection->CloseAllPhysicalConnections(); absl::c_copy( logical_connection->GetEstablisedConnections(), @@ -691,6 +765,13 @@ void AnalyticsRecorder::FinishStrategySessionLocked() { } active_connections_.clear(); + // Finish any pending upgrade attempts. + for (const auto &item : bandwidth_upgrade_attempts_) { + FinishUpgradeAttemptLocked(item.first, UNFINISHED_ERROR, + UPGRADE_UNFINISHED, /*erase_item=*/false); + } + bandwidth_upgrade_attempts_.clear(); + // Add the StrategySession in ClientSession current_strategy_session_->set_duration_millis(absl::ToInt64Milliseconds( started_strategy_session_time_ - SystemClock::ElapsedRealtime())); @@ -810,7 +891,8 @@ void AnalyticsRecorder::LogicalConnection::PhysicalConnectionClosed( void AnalyticsRecorder::LogicalConnection::CloseAllPhysicalConnections() { for (const auto &physical_connection : physical_connections_) { - auto *established_connection = physical_connection.second.get(); + ConnectionsLog::EstablishedConnection *established_connection = + physical_connection.second.get(); if (!established_connection->has_disconnection_reason()) { FinishPhysicalConnection(established_connection, UNFINISHED); } @@ -938,7 +1020,7 @@ AnalyticsRecorder::LogicalConnection::ResolvePendingPayloads( PayloadStatus status = reason == UPGRADED ? MOVED_TO_NEW_MEDIUM : CONNECTION_CLOSED; for (const auto &item : pending_payloads) { - const auto &pending_payload = item.second; + const std::unique_ptr &pending_payload = item.second; ConnectionsLog::Payload proto_payload = pending_payload->GetProtoPayload(status); completed_payloads.push_back(proto_payload); diff --git a/cpp/analytics/analytics_recorder.h b/cpp/analytics/analytics_recorder.h index 9d51897e..8f8a6a16 100644 --- a/cpp/analytics/analytics_recorder.h +++ b/cpp/analytics/analytics_recorder.h @@ -106,7 +106,7 @@ class AnalyticsRecorder { ABSL_LOCKS_EXCLUDED(mutex_); void OnIncomingPayloadDone( const std::string &endpoint_id, std::int64_t payload_id, - ::location::nearby::proto::connections::PayloadStatus status) + location::nearby::proto::connections::PayloadStatus status) ABSL_LOCKS_EXCLUDED(mutex_); void OnOutgoingPayloadStarted(const std::vector &endpoint_ids, std::int64_t payload_id, @@ -119,7 +119,23 @@ class AnalyticsRecorder { ABSL_LOCKS_EXCLUDED(mutex_); void OnOutgoingPayloadDone( const std::string &endpoint_id, std::int64_t payload_id, - ::location::nearby::proto::connections::PayloadStatus status) + location::nearby::proto::connections::PayloadStatus status) + ABSL_LOCKS_EXCLUDED(mutex_); + + // BandwidthUpgrade + void OnBandwidthUpgradeStarted( + const std::string &endpoint_id, + location::nearby::proto::connections::Medium from_medium, + location::nearby::proto::connections::Medium to_medium, + location::nearby::proto::connections::ConnectionAttemptDirection + direction, + const std::string &connection_token) ABSL_LOCKS_EXCLUDED(mutex_); + void OnBandwidthUpgradeError( + const std::string &endpoint_id, + location::nearby::proto::connections::BandwidthUpgradeResult result, + location::nearby::proto::connections::BandwidthUpgradeErrorStage + error_stage) ABSL_LOCKS_EXCLUDED(mutex_); + void OnBandwidthUpgradeSuccess(const std::string &endpoint_id) ABSL_LOCKS_EXCLUDED(mutex_); // Invokes event_logger_.Log() at the end of life of client. Log action is @@ -264,11 +280,17 @@ class AnalyticsRecorder { void MarkConnectionRequestIgnoredLocked( proto::ConnectionsLog::ConnectionRequest *request) ABSL_SHARED_LOCKS_REQUIRED(mutex_); + void FinishUpgradeAttemptLocked( + const std::string &endpoint_id, + location::nearby::proto::connections::BandwidthUpgradeResult result, + location::nearby::proto::connections::BandwidthUpgradeErrorStage + error_stage, + bool erase_item = true) ABSL_SHARED_LOCKS_REQUIRED(mutex_); void FinishStrategySessionLocked() ABSL_EXCLUSIVE_LOCKS_REQUIRED(mutex_); location::nearby::proto::connections::ConnectionsStrategy StrategyToConnectionStrategy(connections::Strategy strategy); - ::location::nearby::proto::connections::PayloadType + location::nearby::proto::connections::PayloadType PayloadTypeToProtoPayloadType(connections::Payload::Type type); // Not owned by AnalyticsRecorder. Pointer must refer to a valid object @@ -310,6 +332,10 @@ class AnalyticsRecorder { outgoing_connection_requests_ ABSL_GUARDED_BY(mutex_); absl::btree_map> active_connections_ ABSL_GUARDED_BY(mutex_); + absl::btree_map< + std::string, + std::unique_ptr> + bandwidth_upgrade_attempts_ ABSL_GUARDED_BY(mutex_); }; } // namespace analytics diff --git a/cpp/analytics/analytics_recorder_test.cc b/cpp/analytics/analytics_recorder_test.cc index 16b9822a..70969da2 100644 --- a/cpp/analytics/analytics_recorder_test.cc +++ b/cpp/analytics/analytics_recorder_test.cc @@ -35,6 +35,7 @@ using ::location::nearby::proto::connections::BLE; using ::location::nearby::proto::connections::BLUETOOTH; using ::location::nearby::proto::connections::CLIENT_SESSION; using ::location::nearby::proto::connections::EventType; +using ::location::nearby::proto::connections::INCOMING; using ::location::nearby::proto::connections::INITIAL; using ::location::nearby::proto::connections::LOCAL_DISCONNECTION; using ::location::nearby::proto::connections::Medium; @@ -46,6 +47,9 @@ 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 ::location::nearby::proto::connections::WIFI_LAN_MEDIUM_ERROR; +using ::location::nearby::proto::connections::WIFI_LAN_SOCKET_CREATION; + using ::testing::Contains; using ::testing::EqualsProto; using ::testing::proto::Partially; @@ -239,7 +243,6 @@ TEST(AnalyticsRecorderTest, AdvertiserConnectionRequestsWorks) { medium: BLE medium: BLUETOOTH received_connection_request < - request_delay_millis: 0 local_response: ACCEPTED remote_response: ACCEPTED > @@ -536,9 +539,9 @@ TEST(AnalyticsRecorderTest, UnfinishedEstablishedConnectionsAddedAsUnfinished) { 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"); + 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); @@ -596,6 +599,69 @@ TEST(AnalyticsRecorderTest, OutgoingPayloadUpgraded) { >)pb"))); } +TEST(AnalyticsRecorderTest, UpgradeAttemptWorks) { + connections::Strategy strategy = connections::Strategy::kP2pStar; + std::vector mediums = {BLE, BLUETOOTH}; + std::string endpoint_id = "endpoint_id"; + std::string endpoint_id_1 = "endpoint_id_1"; + std::string endpoint_id_2 = "endpoint_id_2"; + 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.OnBandwidthUpgradeStarted(endpoint_id, BLE, WIFI_LAN, + INCOMING, connection_token); + + analytics_recorder.OnBandwidthUpgradeStarted( + endpoint_id_1, BLUETOOTH, WIFI_LAN, INCOMING, connection_token); + // Error to upgrade. + analytics_recorder.OnBandwidthUpgradeError(endpoint_id, WIFI_LAN_MEDIUM_ERROR, + WIFI_LAN_SOCKET_CREATION); + // Success to upgrade. + analytics_recorder.OnBandwidthUpgradeSuccess(endpoint_id_1); + // Upgrade is unfinished. + analytics_recorder.OnBandwidthUpgradeStarted( + endpoint_id_2, BLUETOOTH, WIFI_LAN, INCOMING, 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 > + upgrade_attempt < + direction: INCOMING + from_medium: BLE + to_medium: WIFI_LAN + upgrade_result: WIFI_LAN_MEDIUM_ERROR + error_stage: WIFI_LAN_SOCKET_CREATION + connection_token: "connection_token" + > + upgrade_attempt < + direction: INCOMING + from_medium: BLUETOOTH + to_medium: WIFI_LAN + upgrade_result: UPGRADE_RESULT_SUCCESS + error_stage: UPGRADE_SUCCESS + connection_token: "connection_token" + > + upgrade_attempt { + direction: INCOMING + from_medium: BLUETOOTH + to_medium: WIFI_LAN + upgrade_result: UNFINISHED_ERROR + error_stage: UPGRADE_UNFINISHED + connection_token: "connection_token" + } + >)pb"))); +} + } // namespace } // namespace analytics } // namespace nearby diff --git a/cpp/core/internal/bwu_manager.cc b/cpp/core/internal/bwu_manager.cc index a870fd1d..41d6d89c 100644 --- a/cpp/core/internal/bwu_manager.cc +++ b/cpp/core/internal/bwu_manager.cc @@ -27,7 +27,7 @@ #include "platform/base/byte_array.h" #include "platform/base/feature_flags.h" #include "platform/public/count_down_latch.h" -#include "proto/connections_enums.pb.h" +#include "proto/connections_enums.proto.h" namespace location { namespace nearby { @@ -171,11 +171,17 @@ void BwuManager::InitiateBwuForEndpoint(ClientProxy* client, CancelRetryUpgradeAlarm(endpoint_id); auto channel = channel_manager_->GetChannelForEndpoint(endpoint_id); + client->GetAnalyticsRecorder().OnBandwidthUpgradeStarted( + endpoint_id, channel->GetMedium(), medium_, + proto::connections::INCOMING, client->GetConnectionToken(endpoint_id)); if (channel == nullptr) { NEARBY_LOGS(INFO) << "BwuManager couldn't complete the upgrade for endpoint " << endpoint_id << " because it couldn't find an existing EndpointChannel for it."; + client->GetAnalyticsRecorder().OnBandwidthUpgradeError( + endpoint_id, proto::connections::CHANNEL_ERROR, + proto::connections::NETWORK_AVAILABLE); return; } @@ -212,6 +218,9 @@ void BwuManager::InitiateBwuForEndpoint(ClientProxy* client, info.set_medium(parser::MediumToUpgradePathInfoMedium(medium_)); ProcessUpgradeFailureEvent(client, endpoint_id, info); + client->GetAnalyticsRecorder().OnBandwidthUpgradeError( + endpoint_id, proto::connections::RESULT_IO_ERROR, + proto::connections::NETWORK_AVAILABLE); return; } if (!channel->Write(bytes).Ok()) { @@ -376,6 +385,9 @@ void BwuManager::OnIncomingConnection( "BwuManager failed to create new EndpointChannel for incoming " "socket."); connection->socket->Close(); + AttemptToRecordBandwidthUpgradeErrorForUnknownEndpoint( + proto::connections::MEDIUM_ERROR, + proto::connections::SOCKET_CREATION); return; } @@ -465,6 +477,9 @@ void BwuManager::RunUpgradeProtocol( << endpoint_id << " when registering the new EndpointChannel, short-circuiting the " "upgrade protocol."; + client->GetAnalyticsRecorder().OnBandwidthUpgradeError( + endpoint_id, proto::connections::CHANNEL_ERROR, + proto::connections::PRIOR_ENDPOINT_CHANNEL); return; } channel_manager_->ReplaceChannelForEndpoint(client, endpoint_id, @@ -479,6 +494,9 @@ void BwuManager::RunUpgradeProtocol( "BWU_NEGOTIATION.LAST_WRITE_TO_PRIOR_CHANNEL OfflineFrame to " "endpoint " << endpoint_id << ", short-circuiting the upgrade protocol."; + client->GetAnalyticsRecorder().OnBandwidthUpgradeError( + endpoint_id, proto::connections::RESULT_IO_ERROR, + proto::connections::LAST_WRITE_TO_PRIOR_CHANNEL); return; } NEARBY_LOGS(VERBOSE) << "BwuManager successfully wrote " @@ -546,6 +564,10 @@ void BwuManager::ProcessBwuPathAvailableEvent( return; } + client->GetAnalyticsRecorder().OnBandwidthUpgradeStarted( + endpoint_id, medium, medium_, proto::connections::OUTGOING, + client->GetConnectionToken(endpoint_id)); + absl::Time connection_attempt_start_time = SystemClock::ElapsedRealtime(); auto channel = ProcessBwuPathAvailableEventInternal(client, endpoint_id, upgrade_path_info); @@ -554,6 +576,9 @@ void BwuManager::ProcessBwuPathAvailableEvent( connection_attempt_result = proto::connections::RESULT_SUCCESS; } else if (client->GetCancellationFlag(endpoint_id)->Cancelled()) { connection_attempt_result = proto::connections::RESULT_CANCELLED; + client->GetAnalyticsRecorder().OnBandwidthUpgradeError( + endpoint_id, proto::connections::RESULT_REMOTE_ERROR, + proto::connections::UPGRADE_CANCEL); } else { connection_attempt_result = proto::connections::RESULT_ERROR; } @@ -589,6 +614,9 @@ BwuManager::ProcessBwuPathAvailableEventInternal( NEARBY_LOGS(ERROR) << "BwuManager failed to create an endpoint channel to endpoint" << endpoint_id << ", aborting upgrade."; + client->GetAnalyticsRecorder().OnBandwidthUpgradeError( + endpoint_id, proto::connections::RESULT_IO_ERROR, + proto::connections::SOCKET_CREATION); return nullptr; } @@ -604,7 +632,9 @@ BwuManager::ProcessBwuPathAvailableEventInternal( << "BwuManager failed to write BWU_NEGOTIATION.CLIENT_INTRODUCTION " "OfflineFrame to newly-created EndpointChannel " << channel->GetName() << ", aborting upgrade."; - + client->GetAnalyticsRecorder().OnBandwidthUpgradeError( + endpoint_id, proto::connections::RESULT_IO_ERROR, + proto::connections::CLIENT_INTRODUCTION); return {}; } @@ -651,6 +681,9 @@ void BwuManager::RunUpgradeFailedProtocol( << endpoint_id << " when sending an upgrade failure frame, short-circuiting the " "upgrade protocol."; + client->GetAnalyticsRecorder().OnBandwidthUpgradeError( + endpoint_id, proto::connections::CHANNEL_ERROR, + proto::connections::NETWORK_AVAILABLE); return; } @@ -662,6 +695,9 @@ void BwuManager::RunUpgradeFailedProtocol( << "BwuManager failed to write BWU_NEGOTIATION.UPGRADE_FAILURE " "OfflineFrame to endpoint " << endpoint_id << ", short-circuiting the upgrade protocol."; + client->GetAnalyticsRecorder().OnBandwidthUpgradeError( + endpoint_id, proto::connections::RESULT_IO_ERROR, + proto::connections::NETWORK_AVAILABLE); return; } @@ -799,6 +835,9 @@ void BwuManager::ProcessLastWriteToPriorChannelEvent( "OfflineFrame to endpoint " << endpoint_id << ", short-circuiting the upgrade protocol."; + client->GetAnalyticsRecorder().OnBandwidthUpgradeError( + endpoint_id, proto::connections::RESULT_IO_ERROR, + proto::connections::SAFE_TO_CLOSE_PRIOR_CHANNEL); return; } NEARBY_LOGS(VERBOSE) << "BwuManager successfully wrote " @@ -870,6 +909,8 @@ void BwuManager::ProcessSafeToClosePriorChannelEvent( // upgraded bandwidth connection... client->GetAnalyticsRecorder().OnConnectionEstablished( endpoint_id, medium_, client->GetConnectionToken(endpoint_id)); + // ...and the success of the upgrade itself. + client->GetAnalyticsRecorder().OnBandwidthUpgradeSuccess(endpoint_id); // Now that the old channel has been drained, we can unpause the new channel std::shared_ptr channel = @@ -914,6 +955,9 @@ void BwuManager::ProcessUpgradeFailureEvent( << endpoint_id << " because we have other connected endpoints and can't try a new " "upgrade medium."; + client->GetAnalyticsRecorder().OnBandwidthUpgradeError( + endpoint_id, proto::connections::CHANNEL_ERROR, + proto::connections::NETWORK_AVAILABLE); return; } @@ -1082,6 +1126,38 @@ void BwuManager::RetryUpgradesAfterDelay(ClientProxy* client, << absl::FormatDuration(delay); } +void BwuManager::AttemptToRecordBandwidthUpgradeErrorForUnknownEndpoint( + proto::connections::BandwidthUpgradeResult result, + proto::connections::BandwidthUpgradeErrorStage error_stage) { + if (in_progress_upgrades_.size() == 1) { + auto it = in_progress_upgrades_.begin(); + std::string endpoint_id = it->first; + ClientProxy* client = it->second; + // Note: Even though we know this is an error, we cannot clear state yet. + // We've sent the remote device the credentials they need and it's up to + // them if they want to repeatedly attempt to connect or if they want to + // give up and have us try a different medium. This isn't a decision we can + // make for them. + client->GetAnalyticsRecorder().OnBandwidthUpgradeError(endpoint_id, result, + error_stage); + NEARBY_LOGS(INFO) << "BwuManager got error " + << proto::connections::BandwidthUpgradeResult_Name(result) + << " at stage " + << proto::connections::BandwidthUpgradeErrorStage_Name( + error_stage) + << " when upgrading endpoint " << endpoint_id; + } + // Otherwise, we have no way of knowing which endpoint was trying to connect + // to us :( + NEARBY_LOGS(INFO) << "BwuManager got error " + << proto::connections::BandwidthUpgradeResult_Name(result) + << " at stage " + << proto::connections::BandwidthUpgradeErrorStage_Name( + error_stage) + << ", but we don't know which endpoint was trying to " + "connect to us, so skipping analytics for his error."; +} + absl::Duration BwuManager::CalculateNextRetryDelay( const std::string& endpoint_id) { auto item = retry_delays_.find(endpoint_id); diff --git a/cpp/core/internal/bwu_manager.h b/cpp/core/internal/bwu_manager.h index 05dc9c6a..b304a49a 100644 --- a/cpp/core/internal/bwu_manager.h +++ b/cpp/core/internal/bwu_manager.h @@ -160,6 +160,9 @@ class BwuManager : public EndpointManager::FrameProcessor { absl::Duration CalculateNextRetryDelay(const std::string& endpoint_id); void RetryUpgradesAfterDelay(ClientProxy* client, const std::string& endpoint_id); + void AttemptToRecordBandwidthUpgradeErrorForUnknownEndpoint( + proto::connections::BandwidthUpgradeResult result, + proto::connections::BandwidthUpgradeErrorStage error_stage); Config config_;