TySS-Dev/nearby
1
0
Fork 0
mirror of https://github.com/kidfromjupiter/nearby.git synced 2026-09-15 15:16:12 -04:00
Code Issues Packages Projects Releases Wiki Activity
Files
a2bcb2db271ef489c01deb7bcb6527ef344bc006
nearby/connections/implementation/analytics/analytics_recorder.cc
T
hai007 d999507674 update error code category 
PiperOrigin-RevId: 746238861
2025-04-10 17:43:50 -07:00

1726 lines
68 KiB
C++
Raw Blame History

// Copyright 2022-2023 Google LLC
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// https://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
#include "connections/implementation/analytics/analytics_recorder.h"
#include <algorithm>
#include <cstdint>
#include <iterator>
#include <memory>
#include <string>
#include <utility>
#include <vector>
#include "absl/algorithm/container.h"
#include "absl/container/btree_map.h"
#include "absl/strings/string_view.h"
#include "absl/time/time.h"
#include "connections/implementation/analytics/advertising_metadata_params.h"
#include "connections/implementation/analytics/connection_attempt_metadata_params.h"
#include "connections/implementation/analytics/discovery_metadata_params.h"
#include "connections/payload_type.h"
#include "connections/strategy.h"
#include "internal/analytics/event_logger.h"
#include "internal/platform/error_code_params.h"
#include "internal/platform/implementation/system_clock.h"
#include "internal/platform/logging.h"
#include "internal/platform/mutex_lock.h"
#include "internal/proto/analytics/connections_log.pb.h"
#include "proto/connections_enums.pb.h"
#include "google/protobuf/repeated_ptr_field.h"
namespace nearby {
namespace analytics {
namespace {
// const char kVersion_1_0_0[] = "v1.0.0";
const char kVersion [] = "v1.5.0";
constexpr absl::string_view kOnStartClientSession = "OnStartClientSession";
const absl::Duration kConnectionTokenMaxLife = absl::Hours(24);
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::ConnectionBand;
using ::location::nearby::proto::connections::ConnectionRequestResponse;
using ::location::nearby::proto::connections::ConnectionsStrategy;
using ::location::nearby::proto::connections::ConnectionTechnology;
using ::location::nearby::proto::connections::DisconnectionReason;
using ::location::nearby::proto::connections::DISCOVERER;
using ::location::nearby::proto::connections::ERROR_CODE;
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;
using ::location::nearby::proto::connections::Medium;
using ::location::nearby::proto::connections::MOVED_TO_NEW_MEDIUM;
using ::location::nearby::proto::connections::NOT_SENT;
using ::location::nearby::proto::connections::OperationResultCategory;
using ::location::nearby::proto::connections::OperationResultCode;
using ::location::nearby::proto::connections::OUTGOING;
using ::location::nearby::proto::connections::P2P_CLUSTER;
using ::location::nearby::proto::connections::P2P_POINT_TO_POINT;
using ::location::nearby::proto::connections::P2P_STAR;
using ::location::nearby::proto::connections::PayloadStatus;
using ::location::nearby::proto::connections::PayloadType;
using ::location::nearby::proto::connections::REJECTED;
using ::location::nearby::proto::connections::RESULT_SUCCESS;
using ::location::nearby::proto::connections::SessionRole;
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::StopAdvertisingReason;
using ::location::nearby::proto::connections::StopDiscoveringReason;
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;
using ::nearby::analytics::EventLogger;
using SafeDisconnectionResult = ::location::nearby::analytics::proto::
ConnectionsLog::EstablishedConnection::SafeDisconnectionResult;
OperationResultCategory ConvertToOperationResultCategory(
OperationResultCode result_code) {
if (result_code == OperationResultCode::DETAIL_SUCCESS) {
return OperationResultCategory::CATEGORY_SUCCESS;
}
// TODO(b/409865630): check later if we need to add back the dct error.
// Section of CATEGORY_DCT_ERROR, from 5000 to 5499 if (result_code
// >= OperationResultCode::DCT_ERROR_BLE_DISABLED) {
// return OperationResultCategory::CATEGORY_DCT_ERROR;
//}
// Section of CATEGORY_NEARBY_ERROR, starting from 4500 to 4999
if (result_code >=
OperationResultCode::NEARBY_BLE_ADVERTISEMENT_MAPPING_TO_MAC_ERROR) {
return OperationResultCategory::CATEGORY_NEARBY_ERROR;
}
// Section of CATEGORY_CONNECTIVITY_ERROR, starting from 3500 to 4499
if (result_code >=
OperationResultCode::CONNECTIVITY_WIFI_AWARE_ATTACH_FAILURE) {
return OperationResultCategory::CATEGORY_CONNECTIVITY_ERROR;
}
// Section of CATEGORY_IO_ERROR, from 3000 to 3499
if (result_code >= OperationResultCode::IO_FILE_OPENING_ERROR) {
return OperationResultCategory::CATEGORY_IO_ERROR;
}
// Section of CATEGORY_MISCELLANEOUS, from 2500 to 2999
if (result_code >=
OperationResultCode::MISCELLEANEOUS_BLUETOOTH_MAC_ADDRESS_NULL) {
return OperationResultCategory::CATEGORY_MISCELLANEOUS;
}
// Section of CATEGORY_CLIENT_ERROR, from 2000 to 2499
if (result_code >=
OperationResultCode::
CLIENT_WIFI_DIRECT_ALREADY_HOSTING_DIRECT_GROUP_FOR_THIS_CLIENT) {
return OperationResultCategory::CATEGORY_CLIENT_ERROR;
}
// Section of CATEGORY_MEDIUM_UNAVAILABLE, from 1500 to 1999
if (result_code >= OperationResultCode::
MEDIUM_UNAVAILABLE_WIFI_AWARE_RESOURCE_NOT_AVAILABLE) {
return OperationResultCategory::CATEGORY_MEDIUM_UNAVAILABLE;
}
// Section of CATEGORY_DEVICE_STATE_ERROR, from 1000 to 1499
if (result_code >=
OperationResultCode::DEVICE_STATE_ERROR_UNFINISHED_UPGRADE_ATTEMPTS) {
return OperationResultCategory::CATEGORY_DEVICE_STATE_ERROR;
}
// Section of CATEGORY_CLIENT_CANCELLATION, from 500 to 999
if (result_code >=
OperationResultCode::CLIENT_CANCELLATION_REMOTE_IN_CANCELED_STATE) {
return OperationResultCategory::CATEGORY_CLIENT_CANCELLATION;
}
// Clarify other non success cases as unknown
return OperationResultCategory::CATEGORY_UNKNOWN;
}
} // namespace
AnalyticsRecorder::AnalyticsRecorder(EventLogger *event_logger)
: event_logger_(event_logger) {
NEARBY_LOGS(INFO) << "Start AnalyticsRecorder ctor event_logger_="
<< event_logger_;
LogStartSession();
}
AnalyticsRecorder::AnalyticsRecorder(EventLogger *event_logger,
bool no_record_time_millis)
: event_logger_(event_logger),
no_record_time_millis_(no_record_time_millis) {
NEARBY_LOGS(INFO) << "Start AnalyticsRecorder ctor event_logger_="
<< event_logger_;
LogStartSession();
}
AnalyticsRecorder::~AnalyticsRecorder() {
}
bool AnalyticsRecorder::IsSessionLogged() {
MutexLock lock(&mutex_);
return session_was_logged_;
}
int AnalyticsRecorder::GetLatestUpdateIndexLocked(
const std::vector<ConnectionsLog::OperationResultWithMedium> &list) {
int latest_update_index = 0;
for (const auto &operation_result_with_medium : list) {
if (operation_result_with_medium.update_index() > latest_update_index) {
latest_update_index = operation_result_with_medium.update_index();
}
}
return latest_update_index;
}
void AnalyticsRecorder::OnStartAdvertising(
connections::Strategy strategy, const std::vector<Medium> &mediums,
AdvertisingMetadataParams *advertising_metadata_params) {
MutexLock lock(&mutex_);
if (!CanRecordAnalyticsLocked("OnStartAdvertising")) {
return;
}
if (!strategy.IsValid()) {
NEARBY_LOGS(INFO) << "AnalyticsRecorder OnStartAdvertising with unknown "
"strategy, bail out.";
return;
}
// Initialize/update a StrategySession.
UpdateStrategySessionLocked(strategy, ADVERTISER);
// Initialize and set a AdvertisingPhase.
started_advertising_phase_time_ = SystemClock::ElapsedRealtime();
current_advertising_phase_ =
std::make_unique<ConnectionsLog::AdvertisingPhase>();
absl::c_copy(mediums, RepeatedFieldBackInserter(
current_advertising_phase_->mutable_medium()));
// Set a AdvertisingMetadata.
AdvertisingMetadataParams default_params = {};
if (advertising_metadata_params == nullptr) {
advertising_metadata_params = &default_params;
}
if (!advertising_metadata_params->operation_result_with_mediums.empty()) {
absl::c_copy(advertising_metadata_params->operation_result_with_mediums,
RepeatedFieldBackInserter(
current_advertising_phase_->mutable_adv_dis_result()));
}
auto *advertising_metadata =
current_advertising_phase_->mutable_advertising_metadata();
advertising_metadata->set_supports_extended_ble_advertisements(
advertising_metadata_params->is_extended_advertisement_supported);
advertising_metadata->set_connected_ap_frequency(
advertising_metadata_params->connected_ap_frequency);
advertising_metadata->set_supports_nfc_technology(
advertising_metadata_params->is_nfc_available);
}
void AnalyticsRecorder::OnStopAdvertising() {
MutexLock lock(&mutex_);
if (!CanRecordAnalyticsLocked("OnStopAdvertising")) {
return;
}
RecordAdvertisingPhaseDurationAndReasonLocked(/* on_stop= */ true);
}
int AnalyticsRecorder::GetNextAdvertisingUpdateIndex() {
MutexLock lock(&mutex_);
if (current_advertising_phase_ == nullptr) {
return 0;
}
return GetLatestUpdateIndexLocked(
std::vector<ConnectionsLog::OperationResultWithMedium>(
current_advertising_phase_->adv_dis_result().begin(),
current_advertising_phase_->adv_dis_result().end())) +
1;
}
void AnalyticsRecorder::OnStartDiscovery(
connections::Strategy strategy, const std::vector<Medium> &mediums,
DiscoveryMetadataParams *discovery_metadata_params) {
MutexLock lock(&mutex_);
if (!CanRecordAnalyticsLocked("OnStartDiscovery")) {
return;
}
if (!strategy.IsValid()) {
NEARBY_LOGS(INFO) << "AnalyticsRecorder OnStartDiscovery unknown "
"strategy enter, bail out.";
return;
}
// Initialize/update a StrategySession.
UpdateStrategySessionLocked(strategy, DISCOVERER);
// Initialize and set a DiscoveryPhase.
started_discovery_phase_time_ = SystemClock::ElapsedRealtime();
current_discovery_phase_ = std::make_unique<ConnectionsLog::DiscoveryPhase>();
absl::c_copy(mediums, RepeatedFieldBackInserter(
current_discovery_phase_->mutable_medium()));
// Set a DiscoveryMetadata.
DiscoveryMetadataParams default_params = {};
if (discovery_metadata_params == nullptr) {
discovery_metadata_params = &default_params;
}
if (!discovery_metadata_params->operation_result_with_mediums.empty()) {
absl::c_copy(discovery_metadata_params->operation_result_with_mediums,
RepeatedFieldBackInserter(
current_discovery_phase_->mutable_adv_dis_result()));
}
auto *discovery_metadata =
current_discovery_phase_->mutable_discovery_metadata();
discovery_metadata->set_supports_extended_ble_advertisements(
discovery_metadata_params->is_extended_advertisement_supported);
discovery_metadata->set_connected_ap_frequency(
discovery_metadata_params->connected_ap_frequency);
discovery_metadata->set_supports_nfc_technology(
discovery_metadata_params->is_nfc_available);
}
void AnalyticsRecorder::OnStopDiscovery() {
MutexLock lock(&mutex_);
if (!CanRecordAnalyticsLocked("OnStopDiscovery")) {
return;
}
RecordDiscoveryPhaseDurationAndReasonLocked(/*on_stop=*/true);
}
int AnalyticsRecorder::GetNextDiscoveryUpdateIndex() {
MutexLock lock(&mutex_);
if (current_discovery_phase_ == nullptr) {
return 0;
}
return GetLatestUpdateIndexLocked(
std::vector<ConnectionsLog::OperationResultWithMedium>(
current_discovery_phase_->adv_dis_result().begin(),
current_discovery_phase_->adv_dis_result().end())) +
1;
}
void AnalyticsRecorder::OnStartedIncomingConnectionListening(
connections::Strategy strategy) {
MutexLock lock(&mutex_);
if (!CanRecordAnalyticsLocked("OnStartedIncomingConnectionListening")) {
return;
}
UpdateStrategySessionLocked(strategy, ADVERTISER);
if (started_advertising_phase_time_ == absl::InfinitePast()) {
started_advertising_phase_time_ = SystemClock::ElapsedRealtime();
}
}
void AnalyticsRecorder::OnStoppedIncomingConnectionListening() {
MutexLock lock(&mutex_);
if (!CanRecordAnalyticsLocked("OnStoppedIncomingConnectionListening")) {
return;
}
RecordAdvertisingPhaseDurationAndReasonLocked(/* on_stop= */ false);
}
void AnalyticsRecorder::OnEndpointFound(Medium medium) {
MutexLock lock(&mutex_);
if (!CanRecordAnalyticsLocked("OnEndpointFound")) {
return;
}
if (current_discovery_phase_ == nullptr) {
NEARBY_LOGS(INFO) << "Unable to record discovered endpoint due to null "
"current_discovery_phase_";
return;
}
ConnectionsLog::DiscoveredEndpoint *discovered_endpoint =
current_discovery_phase_->add_discovered_endpoint();
discovered_endpoint->set_medium(medium);
if (!no_record_time_millis_) {
discovered_endpoint->set_latency_millis(absl::ToInt64Milliseconds(
SystemClock::ElapsedRealtime() - started_discovery_phase_time_));
}
}
void AnalyticsRecorder::OnRequestConnection(
const connections::Strategy &strategy, const std::string &endpoint_id) {
MutexLock lock(&mutex_);
if (!CanRecordAnalyticsLocked("onRequestConnection")) {
return;
}
UpdateStrategySessionLocked(strategy, DISCOVERER);
if (started_discovery_phase_time_ == absl::InfinitePast()) {
started_discovery_phase_time_ = SystemClock::ElapsedRealtime();
}
}
void AnalyticsRecorder::OnConnectionRequestReceived(
const std::string &remote_endpoint_id) {
MutexLock lock(&mutex_);
if (!CanRecordAnalyticsLocked("OnConnectionRequestReceived")) {
return;
}
absl::Time current_time = SystemClock::ElapsedRealtime();
auto connection_request =
std::make_unique<ConnectionsLog::ConnectionRequest>();
if (!no_record_time_millis_) {
connection_request->set_duration_millis(absl::ToUnixMillis(current_time));
connection_request->set_request_delay_millis(absl::ToInt64Milliseconds(
current_time - started_advertising_phase_time_));
}
incoming_connection_requests_.insert(
{remote_endpoint_id, std::move(connection_request)});
}
void AnalyticsRecorder::OnConnectionRequestSent(
const std::string &remote_endpoint_id) {
MutexLock lock(&mutex_);
if (!CanRecordAnalyticsLocked("OnConnectionRequestSent")) {
return;
}
absl::Time current_time = SystemClock::ElapsedRealtime();
auto connection_request =
std::make_unique<ConnectionsLog::ConnectionRequest>();
if (!no_record_time_millis_) {
connection_request->set_duration_millis(absl::ToUnixMillis(current_time));
connection_request->set_request_delay_millis(absl::ToInt64Milliseconds(
current_time - started_discovery_phase_time_));
}
outgoing_connection_requests_.insert(
{remote_endpoint_id, std::move(connection_request)});
}
void AnalyticsRecorder::OnRemoteEndpointAccepted(
const std::string &remote_endpoint_id) {
MutexLock lock(&mutex_);
if (!CanRecordAnalyticsLocked("OnRemoteEndpointAccepted")) {
return;
}
RemoteEndpointRespondedLocked(remote_endpoint_id, ACCEPTED);
}
void AnalyticsRecorder::OnLocalEndpointAccepted(
const std::string &remote_endpoint_id) {
MutexLock lock(&mutex_);
if (!CanRecordAnalyticsLocked("OnLocalEndpointAccepted")) {
return;
}
LocalEndpointRespondedLocked(remote_endpoint_id, ACCEPTED);
}
void AnalyticsRecorder::OnRemoteEndpointRejected(
const std::string &remote_endpoint_id) {
MutexLock lock(&mutex_);
if (!CanRecordAnalyticsLocked("OnRemoteEndpointRejected")) {
return;
}
RemoteEndpointRespondedLocked(remote_endpoint_id, REJECTED);
}
void AnalyticsRecorder::OnLocalEndpointRejected(
const std::string &remote_endpoint_id) {
MutexLock lock(&mutex_);
if (!CanRecordAnalyticsLocked("OnLocalEndpointRejected")) {
return;
}
LocalEndpointRespondedLocked(remote_endpoint_id, REJECTED);
}
void AnalyticsRecorder::OnIncomingConnectionAttempt(
ConnectionAttemptType type, Medium medium, ConnectionAttemptResult result,
absl::Duration duration, const std::string &connection_token,
ConnectionAttemptMetadataParams *connection_attempt_metadata_params) {
MutexLock lock(&mutex_);
if (!CanRecordAnalyticsLocked("OnIncomingConnectionAttempt")) {
return;
}
if (current_strategy_session_ == nullptr) {
NEARBY_LOGS(INFO) << "Unable to record incoming connection attempt due to "
"null current_strategy_session_";
return;
}
ConnectionAttemptMetadataParams default_params = {};
if (connection_attempt_metadata_params == nullptr) {
connection_attempt_metadata_params = &default_params;
}
OnIncomingConnectionAttemptLocked(type, medium, result, duration,
connection_token,
connection_attempt_metadata_params);
}
void AnalyticsRecorder::OnIncomingConnectionAttemptLocked(
location::nearby::proto::connections::ConnectionAttemptType type,
location::nearby::proto::connections::Medium medium,
location::nearby::proto::connections::ConnectionAttemptResult result,
absl::Duration duration, const std::string &connection_token,
ConnectionAttemptMetadataParams *connection_attempt_metadata_params) {
auto *connection_attempt =
current_strategy_session_->add_connection_attempt();
if (!no_record_time_millis_) {
connection_attempt->set_duration_millis(
absl::ToInt64Milliseconds(duration));
}
connection_attempt->set_type(type);
connection_attempt->set_direction(INCOMING);
connection_attempt->set_medium(medium);
connection_attempt->set_attempt_result(result);
connection_attempt->set_connection_token(connection_token);
auto *connection_attempt_metadata =
connection_attempt->mutable_connection_attempt_metadata();
connection_attempt_metadata->set_technology(
connection_attempt_metadata_params->technology);
connection_attempt_metadata->set_band(
connection_attempt_metadata_params->band);
connection_attempt_metadata->set_frequency(
connection_attempt_metadata_params->frequency);
connection_attempt_metadata->set_network_operator(
connection_attempt_metadata_params->network_operator);
connection_attempt_metadata->set_country_code(
connection_attempt_metadata_params->country_code);
connection_attempt_metadata->set_frequency(
connection_attempt_metadata_params->frequency);
connection_attempt_metadata->set_is_tdls_used(
connection_attempt_metadata_params->is_tdls_used);
connection_attempt_metadata->set_wifi_hotspot_status(
connection_attempt_metadata_params->wifi_hotspot_enabled);
connection_attempt_metadata->set_try_counts(
connection_attempt_metadata_params->try_count);
connection_attempt_metadata->set_max_tx_speed(
connection_attempt_metadata_params->max_wifi_tx_speed);
connection_attempt_metadata->set_max_rx_speed(
connection_attempt_metadata_params->max_wifi_rx_speed);
connection_attempt_metadata->set_wifi_channel_width(
connection_attempt_metadata_params->channel_width);
auto operation_result_proto =
std::make_unique<ConnectionsLog::OperationResult>();
operation_result_proto->set_result_code(
connection_attempt_metadata_params->operation_result_code);
operation_result_proto->set_result_category(ConvertToOperationResultCategory(
connection_attempt_metadata_params->operation_result_code));
connection_attempt->set_allocated_operation_result(
operation_result_proto.release());
}
void AnalyticsRecorder::OnOutgoingConnectionAttempt(
const std::string &remote_endpoint_id, ConnectionAttemptType type,
Medium medium, ConnectionAttemptResult result, absl::Duration duration,
const std::string &connection_token,
ConnectionAttemptMetadataParams *connection_attempt_metadata_params) {
MutexLock lock(&mutex_);
if (!CanRecordAnalyticsLocked("OnOutgoingConnectionAttempt")) {
return;
}
if (current_strategy_session_ == nullptr) {
NEARBY_LOGS(INFO) << "Unable to record outgoing connection attempt due to "
"null current_strategy_session_";
return;
}
ConnectionAttemptMetadataParams default_params = {};
if (connection_attempt_metadata_params == nullptr) {
connection_attempt_metadata_params = &default_params;
}
// For the case of transfer a big file and the upgrades always failure, then
// there will have repeating upgrade attempt and cause many same attempt value
// be log. So add a method to skip.
if (ConnectionAttemptResultCodeExistedLocked(
medium, OUTGOING, connection_token, type,
connection_attempt_metadata_params->operation_result_code)) {
return;
}
OnOutgoingConnectionAttemptLocked(remote_endpoint_id, type, medium, result,
duration, connection_token,
connection_attempt_metadata_params);
}
void AnalyticsRecorder::OnOutgoingConnectionAttemptLocked(
const std::string &remote_endpoint_id, ConnectionAttemptType type,
Medium medium, ConnectionAttemptResult result, absl::Duration duration,
const std::string &connection_token,
ConnectionAttemptMetadataParams *connection_attempt_metadata_params) {
auto *connection_attempt =
current_strategy_session_->add_connection_attempt();
if (!no_record_time_millis_) {
connection_attempt->set_duration_millis(
absl::ToInt64Milliseconds(duration));
}
connection_attempt->set_type(type);
connection_attempt->set_direction(OUTGOING);
connection_attempt->set_medium(medium);
connection_attempt->set_attempt_result(result);
connection_attempt->set_connection_token(connection_token);
auto *connection_attempt_metadata =
connection_attempt->mutable_connection_attempt_metadata();
connection_attempt_metadata->set_technology(
connection_attempt_metadata_params->technology);
connection_attempt_metadata->set_band(
connection_attempt_metadata_params->band);
connection_attempt_metadata->set_frequency(
connection_attempt_metadata_params->frequency);
connection_attempt_metadata->set_network_operator(
connection_attempt_metadata_params->network_operator);
connection_attempt_metadata->set_country_code(
connection_attempt_metadata_params->country_code);
connection_attempt_metadata->set_frequency(
connection_attempt_metadata_params->frequency);
connection_attempt_metadata->set_is_tdls_used(
connection_attempt_metadata_params->is_tdls_used);
connection_attempt_metadata->set_wifi_hotspot_status(
connection_attempt_metadata_params->wifi_hotspot_enabled);
connection_attempt_metadata->set_try_counts(
connection_attempt_metadata_params->try_count);
connection_attempt_metadata->set_max_tx_speed(
connection_attempt_metadata_params->max_wifi_tx_speed);
connection_attempt_metadata->set_max_rx_speed(
connection_attempt_metadata_params->max_wifi_rx_speed);
connection_attempt_metadata->set_wifi_channel_width(
connection_attempt_metadata_params->channel_width);
auto operation_result_proto =
std::make_unique<ConnectionsLog::OperationResult>();
operation_result_proto->set_result_code(
connection_attempt_metadata_params->operation_result_code);
operation_result_proto->set_result_category(ConvertToOperationResultCategory(
connection_attempt_metadata_params->operation_result_code));
connection_attempt->set_allocated_operation_result(
operation_result_proto.release());
if (type == INITIAL && result != RESULT_SUCCESS) {
auto it = outgoing_connection_requests_.find(remote_endpoint_id);
if (it != outgoing_connection_requests_.end()) {
// An outgoing, initial ConnectionAttempt has a corresponding
// ConnectionRequest that, since the ConnectionAttempt has failed, will
// never be delivered to the advertiser.
auto pair = outgoing_connection_requests_.extract(it);
std::unique_ptr<ConnectionsLog::ConnectionRequest> &connection_request =
pair.mapped();
connection_request->set_local_response(NOT_SENT);
connection_request->set_remote_response(NOT_SENT);
UpdateDiscovererConnectionRequestLocked(connection_request.get());
}
}
}
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()) {
const std::unique_ptr<LogicalConnection> &logical_connection = it->second;
logical_connection->PhysicalConnectionEstablished(medium, connection_token);
} else {
active_connections_.insert(
{endpoint_id, std::make_unique<LogicalConnection>(
medium, connection_token, no_record_time_millis_)});
}
}
void AnalyticsRecorder::OnConnectionClosed(const std::string &endpoint_id,
Medium medium,
DisconnectionReason reason,
SafeDisconnectionResult result) {
MutexLock lock(&mutex_);
NEARBY_LOGS(INFO) << __func__
<< ": OnConnectionClosed is called with endpoint_id:"
<< endpoint_id << ", medium:" << Medium_Name(medium)
<< ", reason:" << DisconnectionReason_Name(reason)
<< ", result:" << result;
if (!CanRecordAnalyticsLocked("OnConnectionClosed")) {
return;
}
if (current_strategy_session_ == nullptr) {
NEARBY_VLOG(1) << "AnalyticsRecorder CanRecordAnalytics Unexpected call "
<< __func__
<< " since current_strategy_session_ is required.";
return;
}
auto it = active_connections_.find(endpoint_id);
if (it == active_connections_.end()) {
return;
}
const std::unique_ptr<LogicalConnection> &logical_connection = it->second;
logical_connection->PhysicalConnectionClosed(medium, reason, result);
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::OnIncomingPayloadStarted(
const std::string &endpoint_id, std::int64_t payload_id,
connections::PayloadType 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;
}
const 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;
}
const 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, OperationResultCode operation_result_code) {
MutexLock lock(&mutex_);
if (!CanRecordAnalyticsLocked("OnIncomingPayloadDone")) {
return;
}
auto it = active_connections_.find(endpoint_id);
if (it == active_connections_.end()) {
return;
}
const std::unique_ptr<LogicalConnection> &logical_connection = it->second;
logical_connection->IncomingPayloadDone(payload_id, status,
operation_result_code);
}
void AnalyticsRecorder::OnOutgoingPayloadStarted(
const std::vector<std::string> &endpoint_ids, std::int64_t payload_id,
connections::PayloadType 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;
}
const 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;
}
const 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, OperationResultCode operation_result_code) {
MutexLock lock(&mutex_);
if (!CanRecordAnalyticsLocked("OnOutgoingPayloadDone")) {
return;
}
auto it = active_connections_.find(endpoint_id);
if (it == active_connections_.end()) {
return;
}
const std::unique_ptr<LogicalConnection> &logical_connection = it->second;
logical_connection->OutgoingPayloadDone(payload_id, status,
operation_result_code);
}
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 =
std::make_unique<ConnectionsLog::BandwidthUpgradeAttempt>();
if (!no_record_time_millis_) {
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,
OperationResultCode operation_result_code) {
MutexLock lock(&mutex_);
if (!CanRecordAnalyticsLocked("OnBandwidthUpgradeError")) {
return;
}
// If the same records existed, drop this one.
if (EraseIfBandwidthUpgradeRecordExistedLocked(
endpoint_id, result, error_stage, operation_result_code)) {
return;
}
FinishUpgradeAttemptLocked(endpoint_id, result, error_stage,
operation_result_code);
}
void AnalyticsRecorder::OnBandwidthUpgradeSuccess(
const std::string &endpoint_id) {
MutexLock lock(&mutex_);
if (!CanRecordAnalyticsLocked("OnBandwidthUpgradeSuccess")) {
return;
}
FinishUpgradeAttemptLocked(endpoint_id, UPGRADE_RESULT_SUCCESS,
UPGRADE_SUCCESS,
OperationResultCode::DETAIL_SUCCESS);
}
void AnalyticsRecorder::OnErrorCode(const ErrorCodeParams &params) {
MutexLock lock(&mutex_);
if (!CanRecordAnalyticsLocked("OnErrorCode")) {
return;
}
auto error_code = std::make_unique<ConnectionsLog::ErrorCode>();
error_code->set_medium(params.medium);
error_code->set_event(params.event);
error_code->set_connection_token(params.connection_token);
error_code->set_description(params.description);
if (params.is_common_error) {
error_code->set_common_error(params.common_error);
} else {
switch (params.event) {
case location::nearby::errorcode::proto::START_ADVERTISING:
error_code->set_start_advertising_error(params.start_advertising_error);
break;
case location::nearby::errorcode::proto::STOP_ADVERTISING:
error_code->set_stop_advertising_error(params.stop_advertising_error);
break;
case location::nearby::errorcode::proto::
START_LISTENING_INCOMING_CONNECTION:
error_code->set_start_listening_incoming_connection_error(
params.start_listening_incoming_connection_error);
break;
case location::nearby::errorcode::proto::
STOP_LISTENING_INCOMING_CONNECTION:
error_code->set_stop_listening_incoming_connection_error(
params.stop_listening_incoming_connection_error);
break;
case location::nearby::errorcode::proto::START_DISCOVERING:
error_code->set_start_discovering_error(params.start_discovering_error);
break;
case location::nearby::errorcode::proto::STOP_DISCOVERING:
error_code->set_stop_discovering_error(params.stop_discovering_error);
break;
case location::nearby::errorcode::proto::CONNECT:
error_code->set_connect_error(params.connect_error);
break;
case location::nearby::errorcode::proto::DISCONNECT:
error_code->set_disconnect_error(params.disconnect_error);
break;
case location::nearby::errorcode::proto::UNKNOWN_EVENT:
default:
error_code->set_common_error(params.common_error);
break;
}
}
ConnectionsLog connections_log;
connections_log.set_event_type(ERROR_CODE);
connections_log.set_version(kVersion);
connections_log.set_allocated_error_code(error_code.release());
NEARBY_VLOG(1) << "AnalyticsRecorder LogErrorCode connections_log="
<< connections_log.DebugString(); // NOLINT
event_logger_->Log(connections_log);
}
void AnalyticsRecorder::LogStartSession() {
MutexLock lock(&mutex_);
if (start_client_session_was_logged_) {
NEARBY_LOGS(WARNING)
<< "AnalyticsRecorder CanRecordAnalytics Unexpected call "
<< kOnStartClientSession
<< " after start client session has already been logged.";
return;
}
session_was_logged_ = false;
if (CanRecordAnalyticsLocked(kOnStartClientSession)) {
client_session_ = std::make_unique<ConnectionsLog::ClientSession>();
started_client_session_time_ = SystemClock::ElapsedRealtime();
start_client_session_was_logged_ = true;
LogEvent(START_CLIENT_SESSION);
}
}
void AnalyticsRecorder::LogSession() {
MutexLock lock(&mutex_);
if (!CanRecordAnalyticsLocked("LogSession")) {
return;
}
FinishStrategySessionLocked();
if (!no_record_time_millis_) {
client_session_->set_duration_millis(absl::ToInt64Milliseconds(
SystemClock::ElapsedRealtime() - started_client_session_time_));
}
LogClientSessionLocked();
LogEvent(STOP_CLIENT_SESSION);
start_client_session_was_logged_ = false;
session_was_logged_ = true;
}
std::unique_ptr<AdvertisingMetadataParams>
AnalyticsRecorder::BuildAdvertisingMetadataParams(
bool is_extended_advertisement_supported, int connected_ap_frequency,
bool is_nfc_available,
const std::vector<ConnectionsLog::OperationResultWithMedium>
&operation_result_with_mediums) {
auto params = std::make_unique<AdvertisingMetadataParams>();
params->is_extended_advertisement_supported =
is_extended_advertisement_supported;
params->connected_ap_frequency = connected_ap_frequency;
params->is_nfc_available = is_nfc_available;
params->operation_result_with_mediums =
std::move(operation_result_with_mediums);
return params;
}
std::unique_ptr<DiscoveryMetadataParams>
AnalyticsRecorder::BuildDiscoveryMetadataParams(
bool is_extended_advertisement_supported, int connected_ap_frequency,
bool is_nfc_available,
const std::vector<ConnectionsLog::OperationResultWithMedium>
&operation_result_with_mediums) {
auto params = std::make_unique<DiscoveryMetadataParams>();
params->is_extended_advertisement_supported =
is_extended_advertisement_supported;
params->connected_ap_frequency = connected_ap_frequency;
params->is_nfc_available = is_nfc_available;
params->operation_result_with_mediums =
std::move(operation_result_with_mediums);
return params;
}
std::unique_ptr<ConnectionAttemptMetadataParams>
AnalyticsRecorder::BuildConnectionAttemptMetadataParams(
ConnectionTechnology technology, ConnectionBand band, int frequency,
int try_count, const std::string &network_operator,
const std::string &country_code, bool is_tdls_used,
bool wifi_hotspot_enabled, int max_wifi_tx_speed, int max_wifi_rx_speed,
int channel_width, OperationResultCode operation_result_code) {
auto params = std::make_unique<ConnectionAttemptMetadataParams>();
params->technology = technology;
params->band = band;
params->frequency = frequency;
params->try_count = try_count;
params->network_operator = network_operator;
params->country_code = country_code;
params->is_tdls_used = is_tdls_used;
params->wifi_hotspot_enabled = wifi_hotspot_enabled;
params->max_wifi_tx_speed = max_wifi_tx_speed;
params->max_wifi_rx_speed = max_wifi_rx_speed;
params->channel_width = channel_width;
params->operation_result_code = operation_result_code;
return params;
}
OperationResultCode AnalyticsRecorder::GetChannelIoErrorResultCodeFromMedium(
Medium medium) {
switch (medium) {
case Medium::BLUETOOTH:
return OperationResultCode::CONNECTIVITY_CHANNEL_IO_ERROR_ON_BT;
case Medium::WIFI_HOTSPOT:
return OperationResultCode::CONNECTIVITY_CHANNEL_IO_ERROR_ON_WIFI_HOTSPOT;
case Medium::BLE:
return OperationResultCode::CONNECTIVITY_CHANNEL_IO_ERROR_ON_BLE;
case Medium::BLE_L2CAP:
return OperationResultCode::CONNECTIVITY_CHANNEL_IO_ERROR_ON_BLE_L2CAP;
case Medium::WIFI_LAN:
return OperationResultCode::CONNECTIVITY_CHANNEL_IO_ERROR_ON_LAN;
case Medium::WIFI_AWARE:
return OperationResultCode::CONNECTIVITY_CHANNEL_IO_ERROR_ON_WIFI_AWARE;
case Medium::NFC:
return OperationResultCode::CONNECTIVITY_CHANNEL_IO_ERROR_ON_NFC;
case Medium::WIFI_DIRECT:
return OperationResultCode::CONNECTIVITY_CHANNEL_IO_ERROR_ON_WIFI_DIRECT;
case Medium::WEB_RTC:
return OperationResultCode::CONNECTIVITY_CHANNEL_IO_ERROR_ON_WEB_RTC;
default:
return OperationResultCode::
CONNECTIVITY_CHANNEL_IO_ERROR_ON_UNKNOWN_MEDIUM;
}
}
bool AnalyticsRecorder::CanRecordAnalyticsLocked(
absl::string_view method_name) {
NEARBY_VLOG(1) << "AnalyticsRecorder LogEvent " << method_name
<< " is calling.";
if (event_logger_ == nullptr) {
return false;
}
if (session_was_logged_) {
NEARBY_VLOG(1) << "AnalyticsRecorder CanRecordAnalytics Unexpected call "
<< method_name << " after session has already been logged.";
return false;
}
return true;
}
// TODO: b/391339677 - Investigate why we need to reset the resources. And
// verify in b/238375695 to see if we still meet the issue after removing the
// Reset function.
void AnalyticsRecorder::LogClientSessionLocked() {
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_VLOG(1) << "AnalyticsRecorder LogClientSession connections_log="
<< connections_log.DebugString(); // NOLINT
event_logger_->Log(connections_log);
client_session_ = nullptr;
}
void AnalyticsRecorder::LogEvent(EventType event_type) {
ConnectionsLog connections_log;
connections_log.set_event_type(event_type);
connections_log.set_version(kVersion);
NEARBY_VLOG(1) << "AnalyticsRecorder LogEvent connections_log="
<< connections_log.DebugString(); // NOLINT
event_logger_->Log(connections_log);
}
void AnalyticsRecorder::UpdateStrategySessionLocked(
connections::Strategy strategy, SessionRole role) {
// If we're not switching strategies, just update the current StrategySession
// with the new role.
if (strategy == current_strategy_ && current_strategy_session_ != nullptr) {
if (absl::c_linear_search(current_strategy_session_->role(), role)) {
// We've already acted as this role before, so make sure we've finished
// recording the previous round.
switch (role) {
case ADVERTISER:
FinishAdvertisingPhaseLocked();
break;
case DISCOVERER:
FinishDiscoveryPhaseLocked();
break;
default:
break;
}
} else {
current_strategy_session_->add_role(role);
}
} else {
// Otherwise, we're starting a new Strategy.
current_strategy_ = strategy;
FinishStrategySessionLocked();
LogEvent(START_STRATEGY_SESSION);
current_strategy_session_ =
std::make_unique<ConnectionsLog::StrategySession>();
started_strategy_session_time_ = SystemClock::ElapsedRealtime();
current_strategy_session_->set_strategy(
StrategyToConnectionStrategy(strategy));
current_strategy_session_->add_role(role);
}
}
void AnalyticsRecorder::RecordAdvertisingPhaseDurationAndReasonLocked(
bool on_stop) const {
if (current_advertising_phase_ == nullptr) {
NEARBY_LOGS(INFO) << "Unable to record advertising phase duration due to "
"null current_advertising_phase_";
return;
}
if (!current_advertising_phase_->has_duration_millis() &&
!no_record_time_millis_) {
current_advertising_phase_->set_duration_millis(absl::ToInt64Milliseconds(
SystemClock::ElapsedRealtime() - started_advertising_phase_time_));
}
if (!current_advertising_phase_->has_stop_reason()) {
current_advertising_phase_->set_stop_reason(
on_stop ? StopAdvertisingReason::CLIENT_STOP_ADVERTISING
: StopAdvertisingReason::FINISH_SESSION_STOP_ADVERTISING);
}
}
void AnalyticsRecorder::FinishAdvertisingPhaseLocked() {
if (current_advertising_phase_ != nullptr) {
for (const auto &item : incoming_connection_requests_) {
// ConnectionRequests still pending have been ignored by the local or
// remote (or both) endpoints.
const std::unique_ptr<ConnectionsLog::ConnectionRequest>
&connection_request = item.second;
MarkConnectionRequestIgnoredLocked(connection_request.get());
UpdateAdvertiserConnectionRequestLocked(connection_request.get());
}
RecordAdvertisingPhaseDurationAndReasonLocked(/* on_stop= */ false);
if (current_strategy_session_ != nullptr) {
*current_strategy_session_->add_advertising_phase() =
*std::move(current_advertising_phase_);
} else {
NEARBY_LOGS(INFO) << "Unable to record advertising phase due to null "
"current_strategy_session_";
}
}
incoming_connection_requests_.clear();
}
void AnalyticsRecorder::RecordDiscoveryPhaseDurationAndReasonLocked(
bool on_stop) const {
if (current_discovery_phase_ == nullptr) {
NEARBY_LOGS(INFO) << "Unable to record discovery phase duration due to "
"null current_discovery_phase_";
return;
}
if (!current_discovery_phase_->has_duration_millis() &&
!no_record_time_millis_) {
current_discovery_phase_->set_duration_millis(absl::ToInt64Milliseconds(
SystemClock::ElapsedRealtime() - started_discovery_phase_time_));
}
// If the stop reason haven't been set yet, then set it.
if (!current_discovery_phase_->has_stop_reason()) {
current_discovery_phase_->set_stop_reason(
on_stop ? StopDiscoveringReason::CLIENT_STOP_DISCOVERING
: StopDiscoveringReason::FINISH_SESSION_STOP_DISCOVERING);
}
}
void AnalyticsRecorder::FinishDiscoveryPhaseLocked() {
if (current_discovery_phase_ != nullptr) {
for (const auto &item : outgoing_connection_requests_) {
// ConnectionRequests still pending have been ignored by the local or
// remote (or both) endpoints.
const std::unique_ptr<ConnectionsLog::ConnectionRequest>
&connection_request = item.second;
MarkConnectionRequestIgnoredLocked(connection_request.get());
UpdateDiscovererConnectionRequestLocked(connection_request.get());
}
RecordDiscoveryPhaseDurationAndReasonLocked(/* on_stop=*/false);
if (current_strategy_session_ != nullptr) {
*current_strategy_session_->add_discovery_phase() =
*std::move(current_discovery_phase_);
} else {
NEARBY_LOGS(INFO) << "Unable to record discovery phase due to null "
"current_strategy_session_";
}
}
outgoing_connection_requests_.clear();
}
bool AnalyticsRecorder::UpdateAdvertiserConnectionRequestLocked(
ConnectionsLog::ConnectionRequest *request) {
if (current_advertising_phase_ == nullptr) {
NEARBY_LOGS(INFO)
<< "Unable to record advertiser connection request due to null "
"current_advertising_phase_";
return false;
}
if (BothEndpointsRespondedLocked(request)) {
if (!no_record_time_millis_) {
request->set_duration_millis(
absl::ToUnixMillis(SystemClock::ElapsedRealtime()) -
request->duration_millis());
}
*current_advertising_phase_->add_received_connection_request() = *request;
return true;
}
return false;
}
bool AnalyticsRecorder::UpdateDiscovererConnectionRequestLocked(
ConnectionsLog::ConnectionRequest *request) {
if (current_discovery_phase_ == nullptr) {
NEARBY_LOGS(INFO) << "Unable to record discoverer connection request due "
"to null current_discovery_phase_.";
return false;
}
if (BothEndpointsRespondedLocked(request) ||
request->local_response() == NOT_SENT) {
if (!no_record_time_millis_) {
request->set_duration_millis(
absl::ToUnixMillis(SystemClock::ElapsedRealtime()) -
request->duration_millis());
}
*current_discovery_phase_->add_sent_connection_request() = *request;
return true;
}
return false;
}
bool AnalyticsRecorder::BothEndpointsRespondedLocked(
ConnectionsLog::ConnectionRequest *request) {
return request->has_local_response() && request->has_remote_response();
}
void AnalyticsRecorder::LocalEndpointRespondedLocked(
const std::string &remote_endpoint_id, ConnectionRequestResponse response) {
auto out = outgoing_connection_requests_.find(remote_endpoint_id);
if (out != outgoing_connection_requests_.end()) {
ConnectionsLog::ConnectionRequest *connection_request = out->second.get();
connection_request->set_local_response(response);
if (UpdateDiscovererConnectionRequestLocked(connection_request)) {
outgoing_connection_requests_.erase(out);
}
}
auto in = incoming_connection_requests_.find(remote_endpoint_id);
if (in != incoming_connection_requests_.end()) {
ConnectionsLog::ConnectionRequest *connection_request = in->second.get();
connection_request->set_local_response(response);
if (UpdateAdvertiserConnectionRequestLocked(connection_request)) {
incoming_connection_requests_.erase(in);
}
}
}
void AnalyticsRecorder::RemoteEndpointRespondedLocked(
const std::string &remote_endpoint_id, ConnectionRequestResponse response) {
auto out = outgoing_connection_requests_.find(remote_endpoint_id);
if (out != outgoing_connection_requests_.end()) {
ConnectionsLog::ConnectionRequest *connection_request = out->second.get();
connection_request->set_remote_response(response);
if (UpdateDiscovererConnectionRequestLocked(connection_request)) {
outgoing_connection_requests_.erase(out);
}
}
auto in = incoming_connection_requests_.find(remote_endpoint_id);
if (in != incoming_connection_requests_.end()) {
ConnectionsLog::ConnectionRequest *connection_request = in->second.get();
connection_request->set_remote_response(response);
if (UpdateAdvertiserConnectionRequestLocked(connection_request)) {
incoming_connection_requests_.erase(in);
}
}
}
void AnalyticsRecorder::MarkConnectionRequestIgnoredLocked(
ConnectionsLog::ConnectionRequest *request) {
if (!request->has_local_response()) {
request->set_local_response(IGNORED);
}
if (!request->has_remote_response()) {
request->set_remote_response(IGNORED);
}
}
bool AnalyticsRecorder::ConnectionAttemptResultCodeExistedLocked(
Medium medium, ConnectionAttemptDirection direction,
const std::string &connection_token, ConnectionAttemptType type,
OperationResultCode operation_result_code) {
if (current_strategy_session_ == nullptr ||
current_strategy_session_->connection_attempt_size() == 0) {
return false;
}
for (auto &connection_attempt :
current_strategy_session_->connection_attempt()) {
if (connection_attempt.medium() == medium &&
connection_attempt.direction() == direction &&
connection_attempt.connection_token() == connection_token &&
connection_attempt.type() == type &&
connection_attempt.operation_result().result_code() ==
operation_result_code) {
return true;
}
}
return false;
}
// If bandwidth upgrade always failed on the same fromMedium, toMedium, result,
// stage and result code, we'll drop the duplicate logs for preventing the waste
// of log storage space
bool AnalyticsRecorder::EraseIfBandwidthUpgradeRecordExistedLocked(
const std::string &endpoint_id, BandwidthUpgradeResult result,
BandwidthUpgradeErrorStage error_stage,
OperationResultCode operation_result_code) {
if (current_strategy_session_ == nullptr) {
return false;
}
auto it = bandwidth_upgrade_attempts_.find(endpoint_id);
if (it != bandwidth_upgrade_attempts_.end()) {
ConnectionsLog::BandwidthUpgradeAttempt *attempt = it->second.get();
for (auto &existing_attempt :
current_strategy_session_->upgrade_attempt()) {
if (attempt->from_medium() == existing_attempt.from_medium() &&
attempt->to_medium() == existing_attempt.to_medium() &&
result == existing_attempt.upgrade_result() &&
error_stage == existing_attempt.error_stage() &&
operation_result_code ==
existing_attempt.operation_result().result_code()) {
bandwidth_upgrade_attempts_.erase(it);
return true;
}
}
}
return false;
}
void AnalyticsRecorder::FinishUpgradeAttemptLocked(
const std::string &endpoint_id, BandwidthUpgradeResult result,
BandwidthUpgradeErrorStage error_stage,
OperationResultCode operation_result_code, 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();
if (!no_record_time_millis_) {
attempt->set_duration_millis(
absl::ToUnixMillis(SystemClock::ElapsedRealtime()) -
attempt->duration_millis());
}
attempt->set_error_stage(error_stage);
attempt->set_upgrade_result(result);
auto operation_result_proto =
std::make_unique<ConnectionsLog::OperationResult>();
operation_result_proto->set_result_code(operation_result_code);
operation_result_proto->set_result_category(
ConvertToOperationResultCategory(operation_result_code));
attempt->set_allocated_operation_result(operation_result_proto.release());
*current_strategy_session_->add_upgrade_attempt() = *attempt;
if (erase_item) {
bandwidth_upgrade_attempts_.erase(it);
}
}
}
void AnalyticsRecorder::FinishStrategySessionLocked() {
if (current_strategy_session_ != nullptr) {
FinishAdvertisingPhaseLocked();
FinishDiscoveryPhaseLocked();
// Finish any unfinished LogicalConnections.
for (const auto &item : active_connections_) {
const std::unique_ptr<LogicalConnection> &logical_connection =
item.second;
logical_connection->CloseAllPhysicalConnections();
absl::c_copy(
logical_connection->GetEstablisedConnections(),
RepeatedFieldBackInserter(
current_strategy_session_->mutable_established_connection()));
}
active_connections_.clear();
// Finish any pending upgrade attempts.
for (const auto &item : bandwidth_upgrade_attempts_) {
FinishUpgradeAttemptLocked(
item.first, UNFINISHED_ERROR, UPGRADE_UNFINISHED,
OperationResultCode::DEVICE_STATE_ERROR_UNFINISHED_UPGRADE_ATTEMPTS,
/*erase_item=*/false);
}
bandwidth_upgrade_attempts_.clear();
// Add the StrategySession in ClientSession
if (current_strategy_session_ != nullptr) {
if (!no_record_time_millis_) {
current_strategy_session_->set_duration_millis(
absl::ToInt64Milliseconds(SystemClock::ElapsedRealtime() -
started_strategy_session_time_));
}
*client_session_->add_strategy_session() =
*std::move(current_strategy_session_);
}
current_strategy_session_ = nullptr;
current_strategy_ = connections::Strategy::kNone;
LogEvent(STOP_STRATEGY_SESSION);
}
}
ConnectionsStrategy AnalyticsRecorder::StrategyToConnectionStrategy(
connections::Strategy strategy) {
if (strategy == connections::Strategy::kP2pCluster) {
return P2P_CLUSTER;
}
if (strategy == connections::Strategy::kP2pStar) {
return P2P_STAR;
}
if (strategy == connections::Strategy::kP2pPointToPoint) {
return P2P_POINT_TO_POINT;
}
return UNKNOWN_STRATEGY;
}
PayloadType AnalyticsRecorder::PayloadTypeToProtoPayloadType(
connections::PayloadType type) {
switch (type) {
case connections::PayloadType::kBytes:
return BYTES;
case connections::PayloadType::kFile:
return FILE;
case connections::PayloadType::kStream:
return STREAM;
default:
return UNKNOWN_PAYLOAD_TYPE;
}
}
void AnalyticsRecorder::PendingPayload::AddChunk(
std::int64_t chunk_size_bytes) {
num_bytes_transferred_ += chunk_size_bytes;
num_chunks_++;
}
ConnectionsLog::Payload AnalyticsRecorder::PendingPayload::GetProtoPayload(
PayloadStatus status) {
ConnectionsLog::Payload payload;
if (!no_record_time_millis_) {
payload.set_duration_millis(absl::ToInt64Milliseconds(
SystemClock::ElapsedRealtime() - start_time_));
}
payload.set_type(type_);
payload.set_total_size_bytes(total_size_bytes_);
payload.set_num_bytes_transferred(num_bytes_transferred_);
payload.set_num_chunks(num_chunks_);
payload.set_status(status);
auto operation_result_proto =
std::make_unique<ConnectionsLog::OperationResult>();
operation_result_proto->set_result_code(operation_result_code_);
operation_result_proto->set_result_category(
ConvertToOperationResultCategory(operation_result_code_));
payload.set_allocated_operation_result(operation_result_proto.release());
return payload;
}
void AnalyticsRecorder::LogicalConnection::PhysicalConnectionEstablished(
Medium medium, const std::string &connection_token) {
if (current_medium_ != UNKNOWN_MEDIUM) {
NEARBY_LOGS(WARNING)
<< "Unexpected call to PhysicalConnectionEstablished while "
"AnalyticsRecorder still has an active current medium.";
}
auto established_connection =
std::make_unique<ConnectionsLog::EstablishedConnection>();
established_connection->set_medium(medium);
if (!no_record_time_millis_) {
established_connection->set_duration_millis(
absl::ToUnixMillis(SystemClock::ElapsedRealtime()));
}
established_connection->set_connection_token(connection_token);
auto operation_result_proto =
std::make_unique<ConnectionsLog::OperationResult>();
operation_result_proto->set_result_code(OperationResultCode::DETAIL_SUCCESS);
operation_result_proto->set_result_category(
OperationResultCategory::CATEGORY_SUCCESS);
established_connection->set_allocated_operation_result(
operation_result_proto.release());
physical_connections_.insert({medium, std::move(established_connection)});
current_medium_ = medium;
}
void AnalyticsRecorder::LogicalConnection::PhysicalConnectionClosed(
Medium medium, DisconnectionReason reason, SafeDisconnectionResult result) {
if (current_medium_ == UNKNOWN_MEDIUM) {
NEARBY_LOGS(WARNING)
<< "Unexpected call to PhysicalConnectionClosed() for medium "
<< Medium_Name(medium)
<< " while AnalyticsRecorder has no active current medium";
} else if (current_medium_ != medium) {
NEARBY_LOGS(WARNING)
<< "Unexpected call to PhysicalConnectionClosed() for medium "
<< Medium_Name(medium) << "while AnalyticsRecorder has active medium "
<< Medium_Name(current_medium_);
}
auto it = physical_connections_.find(medium);
if (it == physical_connections_.end()) {
NEARBY_LOGS(WARNING)
<< "Unexpected call to physicalConnectionClosed() for medium "
<< Medium_Name(medium)
<< " with no corresponding EstablishedConnection that was previously"
" opened.";
return;
}
ConnectionsLog::EstablishedConnection *established_connection =
it->second.get();
if (established_connection->has_disconnection_reason()) {
NEARBY_LOGS(WARNING)
<< "Unexpected call to physicalConnectionClosed() for medium "
<< Medium_Name(medium) << " which already has disconnection reason "
<< DisconnectionReason_Name(
established_connection->disconnection_reason());
return;
}
FinishPhysicalConnection(established_connection, reason, result);
if (medium == current_medium_) {
// If the EstablishedConnection we just closed was the one that we have
// marked as current, unset currentMedium.
current_medium_ = UNKNOWN_MEDIUM;
}
}
void AnalyticsRecorder::LogicalConnection::CloseAllPhysicalConnections() {
for (const auto &physical_connection : physical_connections_) {
ConnectionsLog::EstablishedConnection *established_connection =
physical_connection.second.get();
if (!established_connection->has_disconnection_reason()) {
FinishPhysicalConnection(
established_connection, UNFINISHED,
ConnectionsLog::EstablishedConnection::SAFE_DISCONNECTION);
}
}
current_medium_ = UNKNOWN_MEDIUM;
}
std::vector<ConnectionsLog::EstablishedConnection>
AnalyticsRecorder::LogicalConnection::GetEstablisedConnections() {
std::vector<ConnectionsLog::EstablishedConnection> established_connections;
if (current_medium_ != UNKNOWN_MEDIUM) {
NEARBY_LOGS(WARNING)
<< "AnalyticsRecorder expected no more active physical connections "
"before logging this endpoint connection.";
return established_connections;
}
std::transform(physical_connections_.begin(), physical_connections_.end(),
std::back_inserter(established_connections),
[](auto &kv) { return *kv.second; });
physical_connections_.clear();
for (auto &established_connection : established_connections) {
if (absl::Milliseconds(established_connection.duration_millis()) >=
kConnectionTokenMaxLife) {
NEARBY_LOGS(INFO) << "connection token exceed TTL, drop token.";
established_connection.set_connection_token("");
}
}
return established_connections;
}
void AnalyticsRecorder::LogicalConnection::IncomingPayloadStarted(
std::int64_t payload_id, PayloadType type, std::int64_t total_size_bytes) {
incoming_payloads_.insert(
{payload_id, std::make_unique<PendingPayload>(type, total_size_bytes,
no_record_time_millis_)});
}
void AnalyticsRecorder::LogicalConnection::ChunkReceived(
std::int64_t payload_id, std::int64_t size_bytes) {
auto it = incoming_payloads_.find(payload_id);
if (it == incoming_payloads_.end()) {
return;
}
PendingPayload *pending_payload = it->second.get();
pending_payload->AddChunk(size_bytes);
}
void AnalyticsRecorder::LogicalConnection::IncomingPayloadDone(
std::int64_t payload_id, PayloadStatus status,
OperationResultCode operation_result_code) {
if (current_medium_ == UNKNOWN_MEDIUM) {
NEARBY_LOGS(WARNING) << "Unexpected call to incomingPayloadDone() while "
"AnalyticsRecorder has no active current medium.";
return;
}
auto it = physical_connections_.find(current_medium_);
if (it != physical_connections_.end()) {
const std::unique_ptr<ConnectionsLog::EstablishedConnection>
&established_connection = it->second;
auto it = incoming_payloads_.find(payload_id);
if (it != incoming_payloads_.end()) {
it->second->SetOperationResultCode(operation_result_code);
*established_connection->add_received_payload() =
it->second->GetProtoPayload(status);
incoming_payloads_.erase(it);
}
}
}
void AnalyticsRecorder::LogicalConnection::OutgoingPayloadStarted(
std::int64_t payload_id, PayloadType type, std::int64_t total_size_bytes) {
outgoing_payloads_.insert(
{payload_id, std::make_unique<PendingPayload>(type, total_size_bytes,
no_record_time_millis_)});
}
void AnalyticsRecorder::LogicalConnection::ChunkSent(std::int64_t payload_id,
std::int64_t size_bytes) {
auto it = outgoing_payloads_.find(payload_id);
if (it == outgoing_payloads_.end()) {
return;
}
PendingPayload *payload = it->second.get();
payload->AddChunk(size_bytes);
}
void AnalyticsRecorder::LogicalConnection::OutgoingPayloadDone(
std::int64_t payload_id, PayloadStatus status,
OperationResultCode operation_result_code) {
if (current_medium_ == UNKNOWN_MEDIUM) {
NEARBY_LOGS(WARNING) << "Unexpected call to outgoingPayloadDone() while "
"AnalyticsRecorder has no active current medium.";
return;
}
auto it = physical_connections_.find(current_medium_);
if (it != physical_connections_.end()) {
const std::unique_ptr<ConnectionsLog::EstablishedConnection>
&established_connection = it->second;
auto it = outgoing_payloads_.find(payload_id);
if (it != outgoing_payloads_.end()) {
it->second->SetOperationResultCode(operation_result_code);
*established_connection->add_sent_payload() =
it->second->GetProtoPayload(status);
outgoing_payloads_.erase(it);
}
}
}
void AnalyticsRecorder::LogicalConnection::FinishPhysicalConnection(
ConnectionsLog::EstablishedConnection *established_connection,
DisconnectionReason reason, SafeDisconnectionResult result) {
established_connection->set_disconnection_reason(reason);
established_connection->set_safe_disconnection_result(result);
if (!no_record_time_millis_) {
established_connection->set_duration_millis(
absl::ToUnixMillis(SystemClock::ElapsedRealtime()) -
established_connection->duration_millis());
}
// Add any not-yet-finished payloads to this EstablishedConnection.
std::vector<ConnectionsLog::Payload> in_payloads =
ResolvePendingPayloads(incoming_payloads_, reason);
absl::c_move(in_payloads,
RepeatedFieldBackInserter(
established_connection->mutable_received_payload()));
std::vector<ConnectionsLog::Payload> out_payloads =
ResolvePendingPayloads(outgoing_payloads_, reason);
absl::c_move(out_payloads,
RepeatedFieldBackInserter(
established_connection->mutable_sent_payload()));
}
std::vector<ConnectionsLog::Payload>
AnalyticsRecorder::LogicalConnection::ResolvePendingPayloads(
absl::btree_map<std::int64_t, std::unique_ptr<PendingPayload>>
&pending_payloads,
DisconnectionReason reason) {
std::vector<ConnectionsLog::Payload> completed_payloads;
absl::btree_map<std::int64_t, std::unique_ptr<PendingPayload>>
upgraded_payloads;
PayloadStatus status =
reason == UPGRADED ? MOVED_TO_NEW_MEDIUM : CONNECTION_CLOSED;
OperationResultCode operation_result_code =
GetPendingPayloadResultCodeFromReason(reason);
for (const auto &item : pending_payloads) {
const std::unique_ptr<PendingPayload> &pending_payload = item.second;
pending_payload->SetOperationResultCode(operation_result_code);
ConnectionsLog::Payload proto_payload =
pending_payload->GetProtoPayload(status);
completed_payloads.push_back(proto_payload);
if (reason == UPGRADED) {
upgraded_payloads.insert(
{item.first,
std::make_unique<PendingPayload>(
pending_payload->type(), pending_payload->total_size_bytes(),
no_record_time_millis_, operation_result_code)});
}
}
pending_payloads.clear();
if (reason == UPGRADED) {
// Re-populate the map with a new PendingPayload for each pending payload,
// since we expect them to be completed on the next EstablishedConnection.
pending_payloads = std::move(upgraded_payloads);
}
// Return the list of completed payloads to be added to the current
// EstablishedConnection.
return completed_payloads;
}
OperationResultCode
AnalyticsRecorder::LogicalConnection::GetPendingPayloadResultCodeFromReason(
DisconnectionReason reason) {
switch (reason) {
case UPGRADED:
return OperationResultCode::MISCELLEANEOUS_MOVE_TO_NEW_MEDIUM;
case DisconnectionReason::LOCAL_DISCONNECTION:
return OperationResultCode::CLIENT_CANCELLATION_LOCAL_DISCONNECT;
case DisconnectionReason::REMOTE_DISCONNECTION:
return OperationResultCode::CLIENT_CANCELLATION_REMOTE_DISCONNECT;
default:
return OperationResultCode::NEARBY_GENERIC_CONNECTION_CLOSED;
}
}
OperationResultCategory AnalyticsRecorder::GetOperationResultCategory(
location::nearby::proto::connections::OperationResultCode result_code) {
return ConvertToOperationResultCategory(result_code);
}
} // namespace analytics
} // namespace nearby
Reference in New Issue View Git Blame Copy Permalink