[ydb-sdk] add federated topic client GetAllClusters method (#16070)

Author: yumkam

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

@@ -15,6 +15,8 @@ using TDbInfo = Ydb::FederationDiscovery::DatabaseInfo;
 
 using TSessionClosedEvent = NTopic::TSessionClosedEvent;
 
+using TAsyncDescribeTopicResult = NTopic::TAsyncDescribeTopicResult;
+
 //! Federated partition session.
 struct TFederatedPartitionSession : public TThrRefBase, public TPrintable<TFederatedPartitionSession> {
     using TPtr = TIntrusivePtr<TFederatedPartitionSession>;
@@ -516,6 +518,32 @@ class TFederatedTopicClient {
     // std::shared_ptr<NTopic::ISimpleBlockingWriteSession> CreateSimpleBlockingWriteSession(const TFederatedWriteSessionSettings& settings);
     std::shared_ptr<NTopic::IWriteSession> CreateWriteSession(const TFederatedWriteSessionSettings& settings);
 
+    struct TClusterInfo {
+        enum class EStatus : int {
+            STATUS_UNSPECIFIED,
+            AVAILABLE,
+            READ_ONLY,
+            UNAVAILABLE,
+        };
+        std::string Name;
+        std::string Endpoint;
+        std::string Path;
+        EStatus Status;
+        // TODO: Id, Weight, ...?
+        //! Replaces Endpoint and Database for federated clusters
+        void AdjustTopicClientSettings(NTopic::TTopicClientSettings& settings) const;
+        //! Prepend Database for federated clusters
+        void AdjustTopicPath(std::string& path) const;
+        //! Usable for at least read operations
+        bool IsAvailableForRead() const;
+        bool IsAvailableForWrite() const;
+    };
+
+    //! Discover all clusters for federated topic.
+    // Will return single cluster with empty name for non-federated clusters.
+    // May return empty list if FederatedTopicClient was destroyed when future fired.
+    NThreading::TFuture<std::vector<TClusterInfo>> GetAllClusterInfo();
+
 protected:
     void OverrideCodec(NTopic::ECodec codecId, std::unique_ptr<NTopic::ICodec>&& codecImpl);
 

src/client/federated_topic/impl/federated_topic.cpp

@@ -86,4 +86,33 @@ void TFederatedTopicClient::OverrideCodec(NTopic::ECodec codecId, std::unique_pt
     return Impl_->OverrideCodec(codecId, std::move(codecImpl));
 }
 
+NThreading::TFuture<std::vector<TFederatedTopicClient::TClusterInfo>> TFederatedTopicClient::GetAllClusterInfo() {
+    return Impl_->GetAllClusterInfo();
+}
+
+void TFederatedTopicClient::TClusterInfo::AdjustTopicClientSettings(NTopic::TTopicClientSettings& settings) const {
+    if (Name.empty()) {
+        return;
+    }
+    settings.DiscoveryEndpoint(Endpoint);
+    settings.Database(Path);
+}
+
+void TFederatedTopicClient::TClusterInfo::AdjustTopicPath(std::string& path) const {
+    if (Name.empty()) {
+        return;
+    }
+    if (path.empty() || path[0] != '/') {
+        path = Path + '/' + path;
+    }
+}
+
+bool TFederatedTopicClient::TClusterInfo::IsAvailableForRead() const {
+    return Status == TClusterInfo::EStatus::AVAILABLE || Status == TClusterInfo::EStatus::READ_ONLY;
+}
+
+bool TFederatedTopicClient::TClusterInfo::IsAvailableForWrite() const {
+    return Status == TClusterInfo::EStatus::AVAILABLE;
+}
+
 } // namespace NYdb::NFederatedTopic

src/client/federated_topic/impl/federated_topic_impl.cpp

@@ -49,6 +49,40 @@ void TFederatedTopicClient::TImpl::InitObserver() {
     }
 }
 
+NThreading::TFuture<std::vector<TFederatedTopicClient::TClusterInfo>> TFederatedTopicClient::TImpl::GetAllClusterInfo() {
+    InitObserver();
+    return Observer->WaitForFirstState().Apply(
+            [weakObserver = std::weak_ptr(Observer)] (const auto& ) {
+                std::vector<TClusterInfo> result;
+                auto observer = weakObserver.lock();
+                if (!observer) {
+                    return result;
+                }
+                auto state = observer->GetState();
+                result.reserve(state->DbInfos.size());
+                for (const auto& db: state->DbInfos) {
+                    auto& dbinfo = result.emplace_back();
+                    switch (db->status()) {
+#define TRANSLATE_STATUS(NAME) \
+                    case TDbInfo::Status::DatabaseInfo_Status_##NAME: \
+                        dbinfo.Status = TClusterInfo::EStatus::NAME; \
+                        break
+                    TRANSLATE_STATUS(STATUS_UNSPECIFIED);
+                    TRANSLATE_STATUS(AVAILABLE);
+                    TRANSLATE_STATUS(READ_ONLY);
+                    TRANSLATE_STATUS(UNAVAILABLE);
+                    default:
+                        Y_ENSURE(false /* impossible status */);
+                    }
+#undef TRANSLATE_STATUS
+                    dbinfo.Name = db->name();
+                    dbinfo.Endpoint = db->endpoint();
+                    dbinfo.Path = db->path();
+                }
+                return result;
+            });
+}
+
 auto TFederatedTopicClient::TImpl::GetSubsessionHandlersExecutor() -> NTopic::IExecutor::TPtr {
     with_lock (Lock) {
         if (!SubsessionHandlersExecutor) {

src/client/federated_topic/impl/federated_topic_impl.h

@@ -65,6 +65,8 @@ class TFederatedTopicClient::TImpl {
     std::shared_ptr<NTopic::ISimpleBlockingWriteSession> CreateSimpleBlockingWriteSession(const TFederatedWriteSessionSettings& settings);
     std::shared_ptr<NTopic::IWriteSession> CreateWriteSession(const TFederatedWriteSessionSettings& settings);
 
+    NThreading::TFuture<std::vector<TFederatedTopicClient::TClusterInfo>> GetAllClusterInfo();
+
     std::shared_ptr<TFederatedDbObserver> GetObserver() {
         std::lock_guard guard(Lock);
         return Observer;

src/client/federated_topic/impl/federation_observer.cpp

@@ -197,6 +197,9 @@ IOutputStream& operator<<(IOutputStream& out, TFederatedDbState const& state) {
         }
         out << " ]";
     }
+    if (!state.ControlPlaneEndpoint.empty()) {
+        out << " ControlPlaneEndpoint: " << state.ControlPlaneEndpoint;
+    }
     return out << " }";
 }