Moved "ydb_federated_topic: use settings from DbDriverState_ in OnDescribePartition" commit from ydb repo

Author: qyryq

src/client/federated_topic/impl/federated_read_session.cpp

@@ -98,7 +98,19 @@ void TFederatedReadSessionImpl::Start() {
 }
 
 void TFederatedReadSessionImpl::OpenSubSessionsImpl(const std::vector<std::shared_ptr<TDbInfo>>& dbInfos) {
-     for (const auto& db : dbInfos) {
+    {
+        TStringBuilder log(GetLogPrefix());
+        log << "Open read subsessions to databases: ";
+        bool first = true;
+        for (const auto& db : dbInfos) {
+            if (first) first = false; else log << ", ";
+            log << "{ name: " << db->name()
+                << ", endpoint: " << db->endpoint()
+                << ", path: " << db->path() << " }";
+        }
+        LOG_LAZY(Log, TLOG_INFO, log);
+    }
+    for (const auto& db : dbInfos) {
         NTopic::TTopicClientSettings settings = SubClientSettings;
         settings
             .Database(db->path())
@@ -113,7 +125,7 @@ void TFederatedReadSessionImpl::OpenSubSessionsImpl(const std::vector<std::share
 
 void TFederatedReadSessionImpl::OnFederatedStateUpdateImpl() {
     if (!FederationState->Status.IsSuccess()) {
-        LOG_LAZY(Log, TLOG_ERR, GetLogPrefix() << "Federated state update failed.");
+        LOG_LAZY(Log, TLOG_ERR, GetLogPrefix() << "Federated state update failed. FederationState: " << *FederationState);
         CloseImpl();
         return;
     }
@@ -151,6 +163,27 @@ void TFederatedReadSessionImpl::OnFederatedStateUpdateImpl() {
         // TODO: investigate here, why empty list?
         // Reason (and returned status) could be BAD_REQUEST or UNAVAILABLE.
         LOG_LAZY(Log, TLOG_ERR, GetLogPrefix() << "No available databases to read.");
+        auto issues = FederationState->Status.GetIssues();
+        TStringBuilder issue;
+        issue << "Requested databases {";
+        bool first = true;
+        for (auto const& dbFromSettings : Settings.GetDatabasesToReadFrom()) {
+            if (first) first = false; else issue << ",";
+            issue << " " << dbFromSettings;
+        }
+        issue << " } not found. Available databases {";
+        first = true;
+        for (auto const& db : FederationState->DbInfos) {
+            if (db->status() == Ydb::FederationDiscovery::DatabaseInfo_Status_AVAILABLE) {
+                if (first) first = false; else issue << ",";
+                issue << " { name: " << db->name()
+                      << ", endpoint: " << db->endpoint()
+                      << ", path: " << db->path() << " }";
+            }
+        }
+        issue << " }";
+        issues.AddIssue(issue);
+        FederationState->Status = TStatus(EStatus::BAD_REQUEST, std::move(issues));
         CloseImpl();
         return;
     }

src/client/federated_topic/impl/federated_write_session.cpp

@@ -3,6 +3,10 @@
 #include <src/client/persqueue_core/impl/log_lazy.h>
 #include <src/client/topic/impl/topic_impl.h>
 
+#define INCLUDE_YDB_INTERNAL_H
+#include <src/client/impl/ydb_internal/logger/log.h>
+#undef INCLUDE_YDB_INTERNAL_H
+
 #include <ydb-cpp-sdk/library/threading/future/future.h>
 
 #include <algorithm>
@@ -32,11 +36,16 @@ TFederatedWriteSession::TFederatedWriteSession(const TFederatedWriteSessionSetti
     , Observer(std::move(observer))
     , AsyncInit(Observer->WaitForFirstState())
     , FederationState(nullptr)
+    , Log(Connections->GetLog())
     , ClientEventsQueue(std::make_shared<NTopic::TWriteSessionEventsQueue>(Settings))
     , BufferFreeSpace(Settings.MaxMemoryUsage_)
 {
 }
 
+TStringBuilder TFederatedWriteSession::GetLogPrefix() const {
+    return TStringBuilder() << GetDatabaseLogPrefix(SubClientSetttings.Database_.value_or("")) << "[" << SessionId << "] ";
+}
+
 void TFederatedWriteSession::Start() {
     // TODO validate settings?
     Settings.EventHandlers_.HandlersExecutor_->Start();
@@ -165,6 +174,10 @@ void TFederatedWriteSession::OnFederatedStateUpdateImpl() {
     }
 
     if (!DatabasesAreSame(preferrableDb, CurrentDatabase)) {
+        LOG_LAZY(Log, TLOG_INFO, GetLogPrefix()
+            << "Start federated write session to database '" << preferrableDb->name()
+            << "' (previous was " << (CurrentDatabase ? CurrentDatabase->name() : "<empty>") << ")"
+            << " FederationState: " << *FederationState);
         OpenSubSessionImpl(preferrableDb);
     }
 

src/client/federated_topic/impl/federated_write_session.h

@@ -90,6 +90,8 @@ class TFederatedWriteSession : public NTopic::IWriteSession,
 
     void CloseImpl(EStatus statusCode, NYql::TIssues&& issues);
 
+    TStringBuilder GetLogPrefix() const;
+
 private:
     // For subsession creation
     const NTopic::TFederatedWriteSessionSettings Settings;
@@ -102,6 +104,8 @@ class TFederatedWriteSession : public NTopic::IWriteSession,
     std::shared_ptr<TFederatedDbState> FederationState;
     NYdbGrpc::IQueueClientContextPtr UpdateStateDelayContext;
 
+    TLog Log;
+
     std::shared_ptr<TDbInfo> CurrentDatabase;
 
     std::string SessionId;

src/client/federated_topic/impl/federation_observer.cpp

@@ -1,3 +1,4 @@
+#include <src/client/persqueue_core/impl/log_lazy.h>
 #include <src/api/grpc/ydb_federation_discovery_v1.grpc.pb.h>
 
 #include <src/client/federated_topic/impl/federation_observer.h>
@@ -131,14 +132,16 @@ void TFederatedDbObserverImpl::OnFederationDiscovery(TStatus&& status, Ydb::Fede
         //   1) The request was meant for a non-federated topic: fall back to single db mode.
         //   2) The database path in the request is simply wrong: the client should get the BAD_REQUEST status.
         if (status.GetStatus() == EStatus::CLIENT_CALL_UNIMPLEMENTED || status.GetStatus() == EStatus::BAD_REQUEST) {
-            // fall back to single db mode
+            LOG_LAZY(DbDriverState_->Log, TLOG_INFO, TStringBuilder()
+                << "OnFederationDiscovery fall back to single mode, database=" << DbDriverState_->Database);
             FederatedDbState->Status = TPlainStatus{};  // SUCCESS
+            FederatedDbState->ControlPlaneEndpoint = DbDriverState_->DiscoveryEndpoint;
             auto dbState = Connections_->GetDriverState(std::nullopt, std::nullopt, std::nullopt, std::nullopt, std::nullopt);
             FederatedDbState->ControlPlaneEndpoint = dbState->DiscoveryEndpoint;
             // FederatedDbState->SelfLocation = ???;
             auto db = std::make_shared<Ydb::FederationDiscovery::DatabaseInfo>();
-            db->set_path(dbState->Database);
-            db->set_endpoint(dbState->DiscoveryEndpoint);
+            db->set_path(DbDriverState_->Database);
+            db->set_endpoint(DbDriverState_->DiscoveryEndpoint);
             db->set_status(Ydb::FederationDiscovery::DatabaseInfo_Status_AVAILABLE);
             db->set_weight(100);
             FederatedDbState->DbInfos.emplace_back(std::move(db));
@@ -170,4 +173,25 @@ void TFederatedDbObserverImpl::OnFederationDiscovery(TStatus&& status, Ydb::Fede
     }
 }
 
+IOutputStream& operator<<(IOutputStream& out, TFederatedDbState const& state) {
+    out << "{ Status: " << state.Status.GetStatus();
+    if (auto const& issues = state.Status.GetIssues(); !issues.Empty()) {
+        out << ", Issues: { " << issues.ToOneLineString() << " }";
+    }
+    if (!state.DbInfos.empty()) {
+        out << ", DbInfos: { ";
+        bool first = true;
+        for (auto const& info : state.DbInfos) {
+            if (first) {
+                first = false;
+            } else {
+                out << ", ";
+            }
+            out << "{ " << info->ShortDebugString() << " }";
+        }
+        out << " }";
+    }
+    return out << " }";
+}
+
 } // namespace NYdb::NFederatedTopic

src/client/federated_topic/impl/federation_observer.h

@@ -51,6 +51,8 @@ struct TFederatedDbState {
         }
         return nullptr;
     }
+
+    friend IOutputStream& operator<<(IOutputStream& out, TFederatedDbState const& state);
 };