Add UseDetectedLocalDC balancing policy #564

Ilya-Repin · 2025-10-23T13:18:21Z

No description provided.

Gazizonoki · 2025-10-24T01:21:11Z

include/ydb-cpp-sdk/client/types/ydb.h

    //! location is a name of datacenter (VLA, MAN), if location is nullopt local datacenter is used
    static TBalancingPolicy UsePreferableLocation(const std::optional<std::string>& location = {});

+    //! Use detected local dc


Распиши чуть подробнее, чем этот вариант от предыдущего отличается. И какие предостережения к использованию (если есть)

Gazizonoki · 2025-10-24T01:43:33Z

src/client/impl/internal/local_dc_detector/local_dc_detector.cpp

+    std::vector<std::uint64_t> timings;
+    timings.reserve(PING_COUNT);
+    for (size_t i = 0; i < PING_COUNT; ++i) {
+        const auto& ep = endpoints[i % endpoints.size()].get();


Тут кажется проще каждый эндпоинт по PING_COUNT раз пинговать, читаемее будет

Gazizonoki · 2025-10-24T01:48:08Z

src/client/impl/internal/local_dc_detector/pinger.h

+
+static constexpr std::uint32_t PING_TIMEOUT_SECONDS = 5;
+
+TDuration PingEndpoint(const Ydb::Discovery::EndpointInfo& endpoint);


Лучше тут std::chrono::duration сделать, мы от TDuration и TInstant потихоньку избавляемся в SDK

Gazizonoki · 2025-10-24T01:48:52Z

src/client/impl/internal/local_dc_detector/pinger.h

+
+namespace NYdb::inline V3 {
+
+static constexpr std::uint32_t PING_TIMEOUT_SECONDS = 5;


Тут лучше временной тип использовать, а не в сырую хранить

Gazizonoki · 2025-10-24T01:49:56Z

src/client/impl/internal/local_dc_detector/pinger.h

+#include <src/client/impl/internal/internal_header.h>
+#include <src/api/protos/ydb_discovery.pb.h>
+
+#include <util/network/sock.h>


Это инклюд можно в .cpp унести

Gazizonoki · 2025-10-24T01:54:09Z

src/client/impl/internal/db_driver_state/endpoint_pool.cpp

        std::vector<std::string> removed;
        if (result.DiscoveryStatus.Status == EStatus::SUCCESS) {
+            if (BalancingPolicy_.PolicyType == TBalancingPolicy::TImpl::EPolicyType::UseDetectedLocalDC) {
+                LocalDCDetector_.DetectLocalDC(result.Result);


А это ведь прям тяжелая операция. А мы ее синхронно делаем. Это может discovery сильно замедлить, тут замеры не помешали бы. Лучше пинги в фоне делать, как по мне, но такое писать сложнее будет

Gazizonoki · 2025-10-24T01:55:31Z

src/client/impl/internal/local_dc_detector/local_dc_detector.h

+    std::string FindNearestLocation(const TEndpointsByLocation& endpointsByLocation);
+
+private:
+    static constexpr std::size_t MAX_ENDPOINTS_PER_LOCATION = 3;


С одной стороны, маловато нод, но тогда сильно тяжелее будет операция. Может действительно надо асинхронно

pnv1 · 2025-10-24T08:47:15Z

src/client/impl/internal/local_dc_detector/local_dc_detector.cpp

+    timings.reserve(PING_COUNT);
+    for (size_t i = 0; i < PING_COUNT; ++i) {
+        const auto& ep = endpoints[i % endpoints.size()].get();
+        timings.push_back(PingEndpoint_(ep).MicroSeconds());


А это же синхронные пинги из одного потока? На первый взгляд так нельзя делать:
Во-первых, 2*3 - это 6 пингов один за другим.
Во-вторых, 3 ендпоинта в ДЦ может быть слишком мало.
Мы можем параллельно пинги рассылать? Странно, если нет. А если да, то видимо надо пинговать асинхронно и бОльшее количество нод (может и по 1 разу) для точности

Gazizonoki · 2025-10-27T16:59:09Z

tests/unit/client/local_dc_detector/local_dc_detector_ut.cpp

+    return measures;
+}
+
+Y_UNIT_TEST_SUITE(LocalDCDetectionTest) {


Новые тесты мы пишем на googletest, от unittest мы отказываемся потихоньку

Copilot

Pull Request Overview

This PR adds a new UseDetectedLocalDC balancing policy that automatically detects the nearest datacenter by measuring endpoint latencies through network pings. The implementation includes:

A local DC detector that pings endpoints across different locations and identifies the nearest one based on median RTT measurements
Integration with the existing balancing policy framework
Unit tests covering basic detection, single location scenarios, unavailable DC handling, and offline DC scenarios

Reviewed Changes

Copilot reviewed 16 out of 16 changed files in this pull request and generated 15 comments.

Show a summary per file

File	Description
`include/ydb-cpp-sdk/client/types/ydb.h`	Added public API declaration for `UseDetectedLocalDC()`
`src/client/types/ydb.cpp`	Implemented the `UseDetectedLocalDC()` factory method
`src/client/impl/internal/common/balancing_policies.h`	Added `UseDetectedLocalDC` enum value to policy types
`src/client/impl/internal/common/balancing_policies.cpp`	Implemented internal policy initialization for detected local DC
`src/client/impl/internal/local_dc_detector/local_dc_detector.h`	Defined detector class interface with configurable pinger
`src/client/impl/internal/local_dc_detector/local_dc_detector.cpp`	Implemented DC detection logic using sampling and RTT measurement
`src/client/impl/internal/local_dc_detector/pinger.h`	Declared endpoint ping function
`src/client/impl/internal/local_dc_detector/pinger.cpp`	Implemented TCP connection-based ping with timeout handling
`src/client/impl/internal/db_driver_state/endpoint_pool.h`	Added `LocalDCDetector_` member to endpoint pool
`src/client/impl/internal/db_driver_state/endpoint_pool.cpp`	Integrated detector into endpoint preference evaluation
`src/client/table/impl/table_client.cpp`	Extended foreign location scanning to include detected local DC policy
`tests/unit/client/local_dc_detector/local_dc_detector_ut.cpp`	Added comprehensive unit tests with mocked ping functionality
CMake files	Added build configuration for the new local_dc_detector component

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

Copilot · 2025-10-27T16:59:10Z

tests/unit/client/local_dc_detector/local_dc_detector_ut.cpp

+    explicit TMockedPinger(std::unordered_map<std::string, std::vector<TDuration>> measuresByAdress) {
+        EndpointByAdress_.reserve(measuresByAdress.size());
+
+        for (auto& [adress, measures] : measuresByAdress) {
+            EndpointByAdress_.emplace(std::move(adress), std::move(measures));
+        }
+    }
+
+    TDuration operator()(const Ydb::Discovery::EndpointInfo& endpoint) const {
+        auto it = EndpointByAdress_.find(endpoint.address());
+        if (it == EndpointByAdress_.end() || Blacklist_.contains(endpoint.address())) {
+            return TDuration::Max();
+        }
+        return it->second.Ping();
+    }
+
+    void BanEndpoint(const std::string& adress) {
+        Blacklist_.insert(adress);
+    }
+
+    void UnbanEndpoint(const std::string& adress) {
+        Blacklist_.erase(adress);
+    }
+
+private:


Corrected spelling of 'measuresByAdress' to 'measuresByAddress'.

Suggested change

explicit TMockedPinger(std::unordered_map<std::string, std::vector<TDuration>> measuresByAdress) {

EndpointByAdress_.reserve(measuresByAdress.size());

for (auto& [adress, measures] : measuresByAdress) {

EndpointByAdress_.emplace(std::move(adress), std::move(measures));

}

}

TDuration operator()(const Ydb::Discovery::EndpointInfo& endpoint) const {

auto it = EndpointByAdress_.find(endpoint.address());

if (it == EndpointByAdress_.end() || Blacklist_.contains(endpoint.address())) {

return TDuration::Max();

}

return it->second.Ping();

}

void BanEndpoint(const std::string& adress) {

Blacklist_.insert(adress);

}

void UnbanEndpoint(const std::string& adress) {

Blacklist_.erase(adress);

}

private:

explicit TMockedPinger(std::unordered_map<std::string, std::vector<TDuration>> measuresByAddress) {

EndpointByAddress_.reserve(measuresByAddress.size());

for (auto& [address, measures] : measuresByAddress) {

EndpointByAddress_.emplace(std::move(address), std::move(measures));

}

}

TDuration operator()(const Ydb::Discovery::EndpointInfo& endpoint) const {

auto it = EndpointByAddress_.find(endpoint.address());

if (it == EndpointByAddress_.end() || Blacklist_.contains(endpoint.address())) {

return TDuration::Max();

}

return it->second.Ping();

}

void BanEndpoint(const std::string& address) {

Blacklist_.insert(address);

}

void UnbanEndpoint(const std::string& address) {

Blacklist_.erase(address);

}

private:

// Update the member variable name below from EndpointByAdress_ to EndpointByAddress_

Copilot · 2025-10-27T16:59:10Z

tests/unit/client/local_dc_detector/local_dc_detector_ut.cpp

+    explicit TMockedPinger(std::unordered_map<std::string, std::vector<TDuration>> measuresByAdress) {
+        EndpointByAdress_.reserve(measuresByAdress.size());
+
+        for (auto& [adress, measures] : measuresByAdress) {
+            EndpointByAdress_.emplace(std::move(adress), std::move(measures));
+        }
+    }
+
+    TDuration operator()(const Ydb::Discovery::EndpointInfo& endpoint) const {
+        auto it = EndpointByAdress_.find(endpoint.address());
+        if (it == EndpointByAdress_.end() || Blacklist_.contains(endpoint.address())) {
+            return TDuration::Max();
+        }
+        return it->second.Ping();
+    }
+
+    void BanEndpoint(const std::string& adress) {
+        Blacklist_.insert(adress);
+    }
+
+    void UnbanEndpoint(const std::string& adress) {
+        Blacklist_.erase(adress);


Corrected spelling of 'EndpointByAdress_' to 'EndpointByAddress_' and 'measuresByAdress' to 'measuresByAddress'.

Suggested change

explicit TMockedPinger(std::unordered_map<std::string, std::vector<TDuration>> measuresByAdress) {

EndpointByAdress_.reserve(measuresByAdress.size());

for (auto& [adress, measures] : measuresByAdress) {

EndpointByAdress_.emplace(std::move(adress), std::move(measures));

}

}

TDuration operator()(const Ydb::Discovery::EndpointInfo& endpoint) const {

auto it = EndpointByAdress_.find(endpoint.address());

if (it == EndpointByAdress_.end() || Blacklist_.contains(endpoint.address())) {

return TDuration::Max();

}

return it->second.Ping();

}

void BanEndpoint(const std::string& adress) {

Blacklist_.insert(adress);

}

void UnbanEndpoint(const std::string& adress) {

Blacklist_.erase(adress);

explicit TMockedPinger(std::unordered_map<std::string, std::vector<TDuration>> measuresByAddress) {

EndpointByAddress_.reserve(measuresByAddress.size());

for (auto& [address, measures] : measuresByAddress) {

EndpointByAddress_.emplace(std::move(address), std::move(measures));

}

}

TDuration operator()(const Ydb::Discovery::EndpointInfo& endpoint) const {

auto it = EndpointByAddress_.find(endpoint.address());

if (it == EndpointByAddress_.end() || Blacklist_.contains(endpoint.address())) {

return TDuration::Max();

}

return it->second.Ping();

}

void BanEndpoint(const std::string& address) {

Blacklist_.insert(address);

}

void UnbanEndpoint(const std::string& address) {

Blacklist_.erase(address);

Copilot · 2025-10-27T16:59:10Z

tests/unit/client/local_dc_detector/local_dc_detector_ut.cpp

+    explicit TMockedPinger(std::unordered_map<std::string, std::vector<TDuration>> measuresByAdress) {
+        EndpointByAdress_.reserve(measuresByAdress.size());
+
+        for (auto& [adress, measures] : measuresByAdress) {
+            EndpointByAdress_.emplace(std::move(adress), std::move(measures));
+        }
+    }
+
+    TDuration operator()(const Ydb::Discovery::EndpointInfo& endpoint) const {
+        auto it = EndpointByAdress_.find(endpoint.address());
+        if (it == EndpointByAdress_.end() || Blacklist_.contains(endpoint.address())) {
+            return TDuration::Max();
+        }
+        return it->second.Ping();
+    }
+
+    void BanEndpoint(const std::string& adress) {
+        Blacklist_.insert(adress);
+    }
+
+    void UnbanEndpoint(const std::string& adress) {
+        Blacklist_.erase(adress);
+    }
+
+private:
+    mutable std::unordered_map<std::string, TMockedEndpoint> EndpointByAdress_;


Corrected spelling of 'adress' to 'address' and 'measuresByAdress' to 'measuresByAddress'.

Suggested change

explicit TMockedPinger(std::unordered_map<std::string, std::vector<TDuration>> measuresByAdress) {

EndpointByAdress_.reserve(measuresByAdress.size());

for (auto& [adress, measures] : measuresByAdress) {

EndpointByAdress_.emplace(std::move(adress), std::move(measures));

}

}

TDuration operator()(const Ydb::Discovery::EndpointInfo& endpoint) const {

auto it = EndpointByAdress_.find(endpoint.address());

if (it == EndpointByAdress_.end() || Blacklist_.contains(endpoint.address())) {

return TDuration::Max();

}

return it->second.Ping();

}

void BanEndpoint(const std::string& adress) {

Blacklist_.insert(adress);

}

void UnbanEndpoint(const std::string& adress) {

Blacklist_.erase(adress);

}

private:

mutable std::unordered_map<std::string, TMockedEndpoint> EndpointByAdress_;

explicit TMockedPinger(std::unordered_map<std::string, std::vector<TDuration>> measuresByAddress) {

EndpointByAddress_.reserve(measuresByAddress.size());

for (auto& [address, measures] : measuresByAddress) {

EndpointByAddress_.emplace(std::move(address), std::move(measures));

}

}

TDuration operator()(const Ydb::Discovery::EndpointInfo& endpoint) const {

auto it = EndpointByAddress_.find(endpoint.address());

if (it == EndpointByAddress_.end() || Blacklist_.contains(endpoint.address())) {

return TDuration::Max();

}

return it->second.Ping();

}

void BanEndpoint(const std::string& address) {

Blacklist_.insert(address);

}

void UnbanEndpoint(const std::string& address) {

Blacklist_.erase(address);

}

private:

mutable std::unordered_map<std::string, TMockedEndpoint> EndpointByAddress_;

Copilot · 2025-10-27T16:59:11Z

tests/unit/client/local_dc_detector/local_dc_detector_ut.cpp

+    explicit TMockedPinger(std::unordered_map<std::string, std::vector<TDuration>> measuresByAdress) {
+        EndpointByAdress_.reserve(measuresByAdress.size());
+
+        for (auto& [adress, measures] : measuresByAdress) {
+            EndpointByAdress_.emplace(std::move(adress), std::move(measures));
+        }
+    }
+
+    TDuration operator()(const Ydb::Discovery::EndpointInfo& endpoint) const {
+        auto it = EndpointByAdress_.find(endpoint.address());
+        if (it == EndpointByAdress_.end() || Blacklist_.contains(endpoint.address())) {
+            return TDuration::Max();
+        }
+        return it->second.Ping();
+    }
+
+    void BanEndpoint(const std::string& adress) {
+        Blacklist_.insert(adress);
+    }
+
+    void UnbanEndpoint(const std::string& adress) {
+        Blacklist_.erase(adress);
+    }
+
+private:
+    mutable std::unordered_map<std::string, TMockedEndpoint> EndpointByAdress_;


Corrected spelling of 'EndpointByAdress_' to 'EndpointByAddress_' and 'adress' to 'address'.

Suggested change

explicit TMockedPinger(std::unordered_map<std::string, std::vector<TDuration>> measuresByAdress) {

EndpointByAdress_.reserve(measuresByAdress.size());

for (auto& [adress, measures] : measuresByAdress) {

EndpointByAdress_.emplace(std::move(adress), std::move(measures));

}

}

TDuration operator()(const Ydb::Discovery::EndpointInfo& endpoint) const {

auto it = EndpointByAdress_.find(endpoint.address());

if (it == EndpointByAdress_.end() || Blacklist_.contains(endpoint.address())) {

return TDuration::Max();

}

return it->second.Ping();

}

void BanEndpoint(const std::string& adress) {

Blacklist_.insert(adress);

}

void UnbanEndpoint(const std::string& adress) {

Blacklist_.erase(adress);

}

private:

mutable std::unordered_map<std::string, TMockedEndpoint> EndpointByAdress_;

explicit TMockedPinger(std::unordered_map<std::string, std::vector<TDuration>> measuresByAddress) {

EndpointByAddress_.reserve(measuresByAddress.size());

for (auto& [address, measures] : measuresByAddress) {

EndpointByAddress_.emplace(std::move(address), std::move(measures));

}

}

TDuration operator()(const Ydb::Discovery::EndpointInfo& endpoint) const {

auto it = EndpointByAddress_.find(endpoint.address());

if (it == EndpointByAddress_.end() || Blacklist_.contains(endpoint.address())) {

return TDuration::Max();

}

return it->second.Ping();

}

void BanEndpoint(const std::string& address) {

Blacklist_.insert(address);

}

void UnbanEndpoint(const std::string& address) {

Blacklist_.erase(address);

}

private:

mutable std::unordered_map<std::string, TMockedEndpoint> EndpointByAddress_;

Copilot · 2025-10-27T16:59:11Z

tests/unit/client/local_dc_detector/local_dc_detector_ut.cpp

+    }
+
+    TDuration operator()(const Ydb::Discovery::EndpointInfo& endpoint) const {
+        auto it = EndpointByAdress_.find(endpoint.address());


Corrected spelling of 'EndpointByAdress_' to 'EndpointByAddress_'.

Copilot · 2025-10-27T16:59:13Z

tests/unit/client/local_dc_detector/local_dc_detector_ut.cpp

+        const std::vector<std::string> endpointsB = {"B1", "B2", "B3"};
+        const std::vector<std::string> endpointsC = {"C1", "C2", "C3", "C4", "C5", "C6", "C7"};
+
+        const std::size_t epoches = 3;


Corrected spelling of 'epoches' to 'epochs'.

Copilot · 2025-10-27T16:59:13Z

tests/unit/client/local_dc_detector/local_dc_detector_ut.cpp

+        const std::vector<std::string> endpointsC = {"C1", "C2", "C3", "C4", "C5", "C6", "C7"};
+
+        const std::size_t epoches = 3;
+        const std::size_t measuresAmount = 10 * epoches;


Corrected spelling of 'epoches' to 'epochs'.

Copilot · 2025-10-27T16:59:13Z

tests/unit/client/local_dc_detector/local_dc_detector_ut.cpp

+        std::function<TDuration(const Ydb::Discovery::EndpointInfo& endpoint)> pinger = TMockedPinger(mockData);
+        TLocalDCDetector detector(pinger);
+
+        for (std::size_t i = 0; i < epoches; ++i) {


Corrected spelling of 'epoches' to 'epochs'.

Copilot · 2025-10-27T16:59:13Z

tests/unit/client/local_dc_detector/local_dc_detector_ut.cpp

+        const std::size_t epoches = 3;
+        const std::size_t measuresAmount = 10 * epoches;


Corrected spelling of 'epoches' to 'epochs'.

Suggested change

const std::size_t epoches = 3;

const std::size_t measuresAmount = 10 * epoches;

const std::size_t epochs = 3;

const std::size_t measuresAmount = 10 * epochs;

Copilot · 2025-10-27T16:59:14Z

tests/unit/client/local_dc_detector/local_dc_detector_ut.cpp

+        const std::size_t epoches = 3;
+        const std::size_t measuresAmount = 10 * epoches;


Corrected spelling of 'epoches' to 'epochs'.

Suggested change

const std::size_t epoches = 3;

const std::size_t measuresAmount = 10 * epoches;

const std::size_t epochs = 3;

const std::size_t measuresAmount = 10 * epochs;

Ilya-Repin added 5 commits October 23, 2025 07:17

Add UseDetectedLocalDC Policy

1286d20

Add tests

75939d5

Change CMakeLists.txt

08d7d37

Fix style

097d5d0

Remove TMeasureResult

4fcc1db

Gazizonoki requested changes Oct 24, 2025

View reviewed changes

pnv1 reviewed Oct 24, 2025

View reviewed changes

Gazizonoki linked an issue Oct 27, 2025 that may be closed by this pull request

C++: Support local DC detect based on smallest RTT #252

Open

Gazizonoki requested a review from Copilot October 27, 2025 16:55

Gazizonoki reviewed Oct 27, 2025

View reviewed changes

Copilot AI reviewed Oct 27, 2025

View reviewed changes


		static constexpr std::uint32_t PING_TIMEOUT_SECONDS = 5;

		TDuration PingEndpoint(const Ydb::Discovery::EndpointInfo& endpoint);


		namespace NYdb::inline V3 {

		static constexpr std::uint32_t PING_TIMEOUT_SECONDS = 5;

		const std::size_t epoches = 3;
		const std::size_t measuresAmount = 10 * epoches;

Add UseDetectedLocalDC balancing policy #564

Are you sure you want to change the base?

Add UseDetectedLocalDC balancing policy #564

Uh oh!

Conversation

Ilya-Repin commented Oct 23, 2025

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull Request Overview

Reviewed Changes

Uh oh!

Copilot AI Oct 27, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Oct 27, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Oct 27, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Oct 27, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Oct 27, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Oct 27, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Oct 27, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Oct 27, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Oct 27, 2025

Choose a reason for hiding this comment

Uh oh!

Copilot AI Oct 27, 2025

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants