YT-28235: Improvements for multi-protocol RPC API

commit_hash:e19914386430972bb9973c195b62378c2cebfc99

13 files changed, 290 insertions, 1 deletions

diff --git a/yt/yt/core/rpc/backend.cpp b/yt/yt/core/rpc/backend.cpp
index b425b6ed6b2..b924d33b254 100644
--- a/yt/yt/core/rpc/backend.cpp
+++ b/yt/yt/core/rpc/backend.cpp
@@ -1,6 +1,7 @@
 #include "backend.h"
 
 #include <yt/yt/core/misc/collection_helpers.h>
+#include <yt/yt/core/misc/error.h>
 
 #include <library/cpp/yt/memory/leaky_singleton.h>
 
@@ -64,6 +65,15 @@ IBackend* TBackendRegistry::FindBackend(TStringBuf protocol)
     return TBackendRegistryImpl::Get()->FindBackend(protocol);
 }
 
+IBackend* TBackendRegistry::GetBackend(TStringBuf protocol)
+{
+    auto* backend = FindBackend(protocol);
+    if (!backend) {
+        THROW_ERROR_EXCEPTION("No RPC backend registered for protocol %Qv", protocol);
+    }
+    return backend;
+}
+
 void TBackendRegistry::RegisterBackend(IBackend* backend)
 {
     TBackendRegistryImpl::Get()->RegisterBackend(backend);
diff --git a/yt/yt/core/rpc/backend.h b/yt/yt/core/rpc/backend.h
index 911dfb5394b..5fa3fb7c620 100644
--- a/yt/yt/core/rpc/backend.h
+++ b/yt/yt/core/rpc/backend.h
@@ -17,6 +17,8 @@ struct IBackend
 
     virtual TStringBuf GetProtocol() = 0;
 
+    virtual std::string BuildLocalEndpointAddress(const std::any& config) = 0;
+
     virtual void RegisterClientConfigField(NYTree::TYsonStructRegistrar<TMultiProtocolClientConfig> registrar) = 0;
     virtual IChannelFactoryPtr CreateChannelFactory(const std::any& config) = 0;
 
@@ -31,6 +33,7 @@ class TBackendRegistry
 public:
     static std::vector<IBackend*> GetBackends();
     static IBackend* FindBackend(TStringBuf protocol);
+    static IBackend* GetBackend(TStringBuf protocol);
     static void RegisterBackend(IBackend* backend);
 };
 
diff --git a/yt/yt/core/rpc/backend_detail-inl.h b/yt/yt/core/rpc/backend_detail-inl.h
index f21a7b0f63d..cd396471df9 100644
--- a/yt/yt/core/rpc/backend_detail-inl.h
+++ b/yt/yt/core/rpc/backend_detail-inl.h
@@ -9,6 +9,14 @@ namespace NYT::NRpc {
 ////////////////////////////////////////////////////////////////////////////////
 
 template <class TClientConfig, class TServerConfig>
+std::string TBackendBase<TClientConfig, TServerConfig>::BuildLocalEndpointAddress(const std::any& config)
+{
+    auto typedConfig = std::any_cast<TIntrusivePtr<TServerConfig>>(config);
+    YT_VERIFY(typedConfig);
+    return DoBuildLocalEndpointAddress(typedConfig);
+}
+
+template <class TClientConfig, class TServerConfig>
 void TBackendBase<TClientConfig, TServerConfig>::RegisterClientConfigField(NYTree::TYsonStructRegistrar<TMultiProtocolClientConfig> registrar)
 {
     registrar
diff --git a/yt/yt/core/rpc/backend_detail.h b/yt/yt/core/rpc/backend_detail.h
index a5f75d88aaa..ae200ccdc94 100644
--- a/yt/yt/core/rpc/backend_detail.h
+++ b/yt/yt/core/rpc/backend_detail.h
@@ -8,6 +8,8 @@
 
 #include <library/cpp/yt/memory/leaky_singleton.h>
 
+#include <library/cpp/yt/assert/assert.h>
+
 #include <any>
 
 namespace NYT::NRpc {
@@ -19,6 +21,8 @@ class TBackendBase
     : public IBackend
 {
 public:
+    std::string BuildLocalEndpointAddress(const std::any& config) final;
+
     void RegisterClientConfigField(NYTree::TYsonStructRegistrar<TMultiProtocolClientConfig> registrar) final;
     IChannelFactoryPtr CreateChannelFactory(const std::any& config) final;
 
@@ -26,6 +30,7 @@ public:
     IServerPtr CreateServer(const std::any& config) final;
 
 protected:
+    virtual std::string DoBuildLocalEndpointAddress(const TIntrusivePtr<TServerConfig>& config) = 0;
     virtual IChannelFactoryPtr DoCreateChannelFactory(const TIntrusivePtr<TClientConfig>& config) = 0;
     virtual IServerPtr DoCreateServer(const TIntrusivePtr<TServerConfig>& config) = 0;
 };
diff --git a/yt/yt/core/rpc/bus/tcp_backend.cpp b/yt/yt/core/rpc/bus/tcp_backend.cpp
index c5a4b4822d0..5244b1c290f 100644
--- a/yt/yt/core/rpc/bus/tcp_backend.cpp
+++ b/yt/yt/core/rpc/bus/tcp_backend.cpp
@@ -2,10 +2,14 @@
 #include "server.h"
 
 #include <yt/yt/core/rpc/backend_detail.h>
+#include <yt/yt/core/rpc/endpoint_address.h>
 
 #include <yt/yt/core/bus/tcp/config.h>
 #include <yt/yt/core/bus/tcp/server.h>
 
+#include <yt/yt/core/net/address.h>
+#include <yt/yt/core/net/local_address.h>
+
 namespace NYT::NRpc::NBus {
 namespace {
 
@@ -27,6 +31,18 @@ public:
     }
 
 protected:
+    std::string DoBuildLocalEndpointAddress(const TServerConfigPtr& config) final
+    {
+        if (!config->Port) {
+            THROW_ERROR_EXCEPTION("RPC backend %Qv is not bound to a port",
+                GetProtocol());
+        }
+        return FormatEndpointAddress({
+            .Protocol = GetProtocol(),
+            .Address = NNet::BuildServiceAddress(NNet::GetLocalHostName(), *config->Port),
+        });
+    }
+
     IChannelFactoryPtr DoCreateChannelFactory(const TClientConfigPtr& config) final
     {
         return CreateTcpBusChannelFactory(config);
diff --git a/yt/yt/core/rpc/bus/uds_backend.cpp b/yt/yt/core/rpc/bus/uds_backend.cpp
index e2f42efd480..106f9083752 100644
--- a/yt/yt/core/rpc/bus/uds_backend.cpp
+++ b/yt/yt/core/rpc/bus/uds_backend.cpp
@@ -2,6 +2,7 @@
 #include "server.h"
 
 #include <yt/yt/core/rpc/backend_detail.h>
+#include <yt/yt/core/rpc/endpoint_address.h>
 
 #include <yt/yt/core/bus/tcp/config.h>
 #include <yt/yt/core/bus/tcp/server.h>
@@ -27,6 +28,18 @@ public:
     }
 
 protected:
+    std::string DoBuildLocalEndpointAddress(const TServerConfigPtr& config) final
+    {
+        if (!config->UnixDomainSocketPath) {
+            THROW_ERROR_EXCEPTION("RPC backend %Qv has no unix domain socket path",
+                GetProtocol());
+        }
+        return FormatEndpointAddress({
+            .Protocol = GetProtocol(),
+            .Address = *config->UnixDomainSocketPath,
+        });
+    }
+
     IChannelFactoryPtr DoCreateChannelFactory(const TClientConfigPtr& config) final
     {
         return CreateUdsBusChannelFactory(config);
diff --git a/yt/yt/core/rpc/config-inl.h b/yt/yt/core/rpc/config-inl.h
index 03f1ca407b7..0779ba76e16 100644
--- a/yt/yt/core/rpc/config-inl.h
+++ b/yt/yt/core/rpc/config-inl.h
@@ -25,6 +25,33 @@ TIntrusivePtr<TConfig>* TProtocolMapConfigBase::MutableTypedConfig(TStringBuf pr
     return &std::any_cast<TIntrusivePtr<TConfig>&>(it->second.CurrentConfig);
 }
 
+template <class TConfig>
+void TProtocolMapConfigBase::SetTypedConfig(TStringBuf protocol, TIntrusivePtr<TConfig> config)
+{
+    *MutableTypedConfig<TConfig>(protocol) = std::move(config);
+}
+
+template <class TConfig>
+TIntrusivePtr<TConfig> TProtocolMapConfigBase::FindTypedConfig(TStringBuf protocol)
+{
+    auto config = FindUntypedConfig(protocol);
+    if (!config.has_value()) {
+        return nullptr;
+    }
+    return std::any_cast<TIntrusivePtr<TConfig>>(std::move(config));
+}
+
+template <class TConfig>
+TIntrusivePtr<TConfig> TProtocolMapConfigBase::GetTypedConfigOrThrow(TStringBuf protocol)
+{
+    auto config = FindTypedConfig<TConfig>(protocol);
+    if (!config) {
+        THROW_ERROR_EXCEPTION("RPC protocol %Qv is not configured",
+            protocol);
+    }
+    return config;
+}
+
 ////////////////////////////////////////////////////////////////////////////////
 
 } // namespace NYT::NRpc
diff --git a/yt/yt/core/rpc/config.cpp b/yt/yt/core/rpc/config.cpp
index d2772dc5129..1417ed1f699 100644
--- a/yt/yt/core/rpc/config.cpp
+++ b/yt/yt/core/rpc/config.cpp
@@ -454,6 +454,17 @@ void TOverloadControllerConfig::Register(TRegistrar registrar)
 
 ////////////////////////////////////////////////////////////////////////////////
 
+std::vector<std::string> TProtocolMapConfigBase::GetConfiguredProtocols() const
+{
+    std::vector<std::string> result;
+    for (const auto& [protocol, entry] : ProtocolToEntry_) {
+        if (!entry.IsNull(entry.CurrentConfig)) {
+            result.push_back(protocol);
+        }
+    }
+    return result;
+}
+
 std::any TProtocolMapConfigBase::GetUntypedConfig(TStringBuf protocol)
 {
     return GetOrCrash(ProtocolToEntry_, protocol).CurrentConfig;
diff --git a/yt/yt/core/rpc/config.h b/yt/yt/core/rpc/config.h
index 72c90f474ed..337c0214ea9 100644
--- a/yt/yt/core/rpc/config.h
+++ b/yt/yt/core/rpc/config.h
@@ -586,13 +586,34 @@ class TProtocolMapConfigBase
     : public NYTree::TYsonStruct
 {
 public:
+    //! Returns the protocols that are actually configured (i.e. present with a
+    //! non-null config), skipping registered-but-unconfigured ones.
+    std::vector<std::string> GetConfiguredProtocols() const;
+
+    //! Returns a pointer to the typed config for #protocol, creating an
+    //! empty (null) entry if needed.
+    template <class TConfig>
+    TIntrusivePtr<TConfig>* MutableTypedConfig(TStringBuf protocol);
+
+    //! Returns the type-erased config for #protocol; crashes if it is absent.
     std::any GetUntypedConfig(TStringBuf protocol);
+
     //! Like #GetUntypedConfig, but returns an empty std::any for an unconfigured protocol
     //! (i.e. one that is absent or whose config is null).
     std::any FindUntypedConfig(TStringBuf protocol);
 
+    //! Sets the typed config for #protocol.
     template <class TConfig>
-    TIntrusivePtr<TConfig>* MutableTypedConfig(TStringBuf protocol);
+    void SetTypedConfig(TStringBuf protocol, TIntrusivePtr<TConfig> config);
+
+    //! Returns the typed config for #protocol, or null if it is unconfigured
+    //! (i.e. absent or holding a null pointer).
+    template <class TConfig>
+    TIntrusivePtr<TConfig> FindTypedConfig(TStringBuf protocol);
+
+    //! Like #FindTypedConfig, but throws if #protocol is unconfigured.
+    template <class TConfig>
+    TIntrusivePtr<TConfig> GetTypedConfigOrThrow(TStringBuf protocol);
 
 private:
     struct TProtocolEntry
diff --git a/yt/yt/core/rpc/grpc/backend.cpp b/yt/yt/core/rpc/grpc/backend.cpp
index 3b3cbe2cc3a..08f2add1786 100644
--- a/yt/yt/core/rpc/grpc/backend.cpp
+++ b/yt/yt/core/rpc/grpc/backend.cpp
@@ -24,6 +24,14 @@ public:
     }
 
 protected:
+    std::string DoBuildLocalEndpointAddress(const TServerConfigPtr& /*config*/) final
+    {
+        // gRPC binds a list of (often wildcard) addresses with no single routable
+        // local host, so there is no meaningful local endpoint to advertise.
+        THROW_ERROR_EXCEPTION("RPC backend %Qv does not support local endpoint address resolution",
+            GetProtocol());
+    }
+
     IChannelFactoryPtr DoCreateChannelFactory(const TClientConfigPtr& config) final
     {
         return CreateGrpcChannelFactory(config);
diff --git a/yt/yt/core/rpc/http/backend.cpp b/yt/yt/core/rpc/http/backend.cpp
index f61e3dad3b9..e68d9d0b0bc 100644
--- a/yt/yt/core/rpc/http/backend.cpp
+++ b/yt/yt/core/rpc/http/backend.cpp
@@ -3,6 +3,10 @@
 #include "server.h"
 
 #include <yt/yt/core/rpc/backend_detail.h>
+#include <yt/yt/core/rpc/endpoint_address.h>
+
+#include <yt/yt/core/net/address.h>
+#include <yt/yt/core/net/local_address.h>
 
 namespace NYT::NRpc::NHttp {
 namespace {
@@ -22,6 +26,14 @@ public:
     }
 
 protected:
+    std::string DoBuildLocalEndpointAddress(const TServerConfigPtr& config) final
+    {
+        return FormatEndpointAddress({
+            .Protocol = GetProtocol(),
+            .Address = NNet::BuildServiceAddress(NNet::GetLocalHostName(), config->Port),
+        });
+    }
+
     IChannelFactoryPtr DoCreateChannelFactory(const TClientConfigPtr& config) final
     {
         return CreateHttpChannelFactory(config);
diff --git a/yt/yt/core/rpc/unittests/main/ya.make b/yt/yt/core/rpc/unittests/main/ya.make
index 476bf100020..db7e0a28fad 100644
--- a/yt/yt/core/rpc/unittests/main/ya.make
+++ b/yt/yt/core/rpc/unittests/main/ya.make
@@ -9,6 +9,7 @@ SRCS(
     yt/yt/core/rpc/unittests/endpoint_address_ut.cpp
     yt/yt/core/rpc/unittests/overload_controller_ut.cpp
     yt/yt/core/rpc/unittests/handle_channel_failure_ut.cpp
+    yt/yt/core/rpc/unittests/multi_protocol_config_ut.cpp
     yt/yt/core/rpc/unittests/roaming_channel_ut.cpp
     yt/yt/core/rpc/unittests/rpc_ut.cpp
     yt/yt/core/rpc/unittests/viable_peer_registry_ut.cpp
diff --git a/yt/yt/core/rpc/unittests/multi_protocol_config_ut.cpp b/yt/yt/core/rpc/unittests/multi_protocol_config_ut.cpp
new file mode 100644
index 00000000000..4179c541c02
--- /dev/null
+++ b/yt/yt/core/rpc/unittests/multi_protocol_config_ut.cpp
@@ -0,0 +1,154 @@
+#include <yt/yt/core/test_framework/framework.h>
+
+#include <yt/yt/core/rpc/backend.h>
+#include <yt/yt/core/rpc/config.h>
+#include <yt/yt/core/rpc/endpoint_address.h>
+
+#include <yt/yt/core/bus/tcp/config.h>
+
+#include <yt/yt/core/ytree/convert.h>
+#include <yt/yt/core/ytree/fluent.h>
+
+#include <algorithm>
+
+namespace NYT::NRpc {
+namespace {
+
+using namespace NYT::NBus::NTcp;
+
+////////////////////////////////////////////////////////////////////////////////
+
+constexpr auto UnregisteredProtocol = "no-such-protocol"_sb;
+
+////////////////////////////////////////////////////////////////////////////////
+
+// Exercises the typed/untyped config accessors of TProtocolMapConfigBase through
+// TMultiProtocolServerConfig over the always-registered "yt-tcp" backend.
+TBusServerConfigPtr MakeBusServerConfig(int port)
+{
+    auto config = New<TBusServerConfig>();
+    config->Port = port;
+    return config;
+}
+
+TMultiProtocolServerConfigPtr MakeServerConfig(int port)
+{
+    return ConvertTo<TMultiProtocolServerConfigPtr>(NYTree::BuildYsonNodeFluently()
+        .BeginMap()
+            .Item(DefaultProtocolName).BeginMap()
+                .Item("port").Value(port)
+            .EndMap()
+        .EndMap());
+}
+
+////////////////////////////////////////////////////////////////////////////////
+
+TEST(TMultiProtocolConfigTest, FindConfiguredProtocol)
+{
+    auto config = MakeServerConfig(123);
+
+    auto typedConfig = config->FindTypedConfig<TBusServerConfig>(DefaultProtocolName);
+    ASSERT_TRUE(typedConfig);
+    EXPECT_EQ(123, typedConfig->Port);
+
+    auto orThrow = config->GetTypedConfigOrThrow<TBusServerConfig>(DefaultProtocolName);
+    EXPECT_EQ(typedConfig, orThrow);
+
+    auto untyped = config->FindUntypedConfig(DefaultProtocolName);
+    ASSERT_TRUE(untyped.has_value());
+    EXPECT_EQ(typedConfig, std::any_cast<TBusServerConfigPtr>(untyped));
+
+    EXPECT_EQ(typedConfig, std::any_cast<TBusServerConfigPtr>(config->GetUntypedConfig(DefaultProtocolName)));
+}
+
+TEST(TMultiProtocolConfigTest, RegisteredButUnconfiguredProtocol)
+{
+    // A freshly created config has the "yt-tcp" entry registered, but null.
+    auto config = New<TMultiProtocolServerConfig>();
+
+    EXPECT_FALSE(config->FindTypedConfig<TBusServerConfig>(DefaultProtocolName));
+
+    // FindUntypedConfig treats a null entry as unconfigured.
+    EXPECT_FALSE(config->FindUntypedConfig(DefaultProtocolName).has_value());
+
+    EXPECT_THROW_WITH_SUBSTRING(
+        config->GetTypedConfigOrThrow<TBusServerConfig>(DefaultProtocolName),
+        "is not configured");
+
+    // GetUntypedConfig still returns the (null) entry without crashing.
+    EXPECT_FALSE(std::any_cast<TBusServerConfigPtr>(config->GetUntypedConfig(DefaultProtocolName)));
+}
+
+TEST(TMultiProtocolConfigTest, UnregisteredProtocol)
+{
+    auto config = New<TMultiProtocolServerConfig>();
+
+    EXPECT_FALSE(config->FindTypedConfig<TBusServerConfig>(UnregisteredProtocol));
+    EXPECT_FALSE(config->FindUntypedConfig(UnregisteredProtocol).has_value());
+    EXPECT_THROW_WITH_SUBSTRING(
+        config->GetTypedConfigOrThrow<TBusServerConfig>(UnregisteredProtocol),
+        "is not configured");
+}
+
+TEST(TMultiProtocolConfigTest, SetTypedConfig)
+{
+    auto config = New<TMultiProtocolServerConfig>();
+
+    auto busConfig = MakeBusServerConfig(777);
+    config->SetTypedConfig(DefaultProtocolName, busConfig);
+
+    auto typedConfig = config->FindTypedConfig<TBusServerConfig>(DefaultProtocolName);
+    EXPECT_EQ(busConfig, typedConfig);
+    EXPECT_EQ(777, typedConfig->Port);
+}
+
+TEST(TMultiProtocolConfigTest, MutableTypedConfig)
+{
+    auto config = New<TMultiProtocolServerConfig>();
+
+    // Creates the entry and lets the caller install a config in place.
+    *config->MutableTypedConfig<TBusServerConfig>(DefaultProtocolName) = MakeBusServerConfig(555);
+    EXPECT_EQ(555, config->FindTypedConfig<TBusServerConfig>(DefaultProtocolName)->Port);
+
+    // The returned pointer aliases the stored config.
+    (*config->MutableTypedConfig<TBusServerConfig>(DefaultProtocolName))->Port = 556;
+    EXPECT_EQ(556, config->FindTypedConfig<TBusServerConfig>(DefaultProtocolName)->Port);
+}
+
+TEST(TMultiProtocolConfigTest, GetConfiguredProtocols)
+{
+    EXPECT_TRUE(
+        New<TMultiProtocolServerConfig>()->GetConfiguredProtocols().empty());
+
+    EXPECT_EQ(
+        std::vector{std::string(DefaultProtocolName)},
+        MakeServerConfig(123)->GetConfiguredProtocols());
+}
+
+////////////////////////////////////////////////////////////////////////////////
+
+TEST(TBackendTest, BuildLocalEndpointAddress)
+{
+    auto* backend = TBackendRegistry::FindBackend(DefaultProtocolName);
+    ASSERT_TRUE(backend);
+
+    auto address = backend->BuildLocalEndpointAddress(std::any(MakeBusServerConfig(9000)));
+    auto parsed = ParseEndpointAddress(address);
+    EXPECT_EQ(DefaultProtocolName, parsed.Protocol);
+    EXPECT_TRUE(parsed.Address.EndsWith(":9000")) << address;
+}
+
+TEST(TBackendTest, BuildLocalEndpointAddressThrowsWithoutPort)
+{
+    auto* backend = TBackendRegistry::FindBackend(DefaultProtocolName);
+    ASSERT_TRUE(backend);
+
+    EXPECT_THROW_WITH_SUBSTRING(
+        backend->BuildLocalEndpointAddress(std::any(New<TBusServerConfig>())),
+        "not bound to a port");
+}
+
+////////////////////////////////////////////////////////////////////////////////
+
+} // namespace
+} // namespace NYT::NRpc