[YDB_LOG] Migrate ydb/core/tx/replication/ydb_proxy (#44670)

7 files changed, 87 insertions, 49 deletions

diff --git a/ydb/core/tx/replication/ydb_proxy/local_proxy/local_partition_actor.cpp b/ydb/core/tx/replication/ydb_proxy/local_proxy/local_partition_actor.cpp
index bafe47aeaca..2da9da71ff4 100644
--- a/ydb/core/tx/replication/ydb_proxy/local_proxy/local_partition_actor.cpp
+++ b/ydb/core/tx/replication/ydb_proxy/local_proxy/local_partition_actor.cpp
@@ -1,5 +1,7 @@
 #include "local_partition_actor.h"
-#include "logging.h"
+#include <ydb/library/actors/core/log.h>
+
+#define YDB_LOG_THIS_FILE_COMPONENT NKikimrServices::LOCAL_YDB_PROXY
 
 namespace NKikimr::NReplication {
 
@@ -17,6 +19,7 @@ TBaseLocalTopicPartitionActor::TBaseLocalTopicPartitionActor(
 
 void TBaseLocalTopicPartitionActor::Bootstrap() {
     LogPrefix = MakeLogPrefix();
+    YDB_LOG_CREATE_CONTEXT(LogPrefix);
     DoDescribe(TopicPath);
 }
 
@@ -34,7 +37,8 @@ TString TBaseLocalTopicPartitionActor::MakeAbsolutePath(TString path) const {
 
 void TBaseLocalTopicPartitionActor::DoDescribe(const TString& topicPath) {
     auto path = MakeAbsolutePath(topicPath);
-    LOG_D("Describe topic '" << path << "'");
+    YDB_LOG_DEBUG("Describe topic",
+        {"path", path});
 
     auto request = MakeHolder<TNavigate>();
     request->DatabaseName = Database;
@@ -45,7 +49,8 @@ void TBaseLocalTopicPartitionActor::DoDescribe(const TString& topicPath) {
 }
 
 void TBaseLocalTopicPartitionActor::Handle(TEvNavigateResult::TPtr& ev) {
-    LOG_T("Handle " << ev->Get()->ToString());
+    YDB_LOG_TRACE("Handle",
+        {"ev", ev->Get()->ToString()});
 
     auto& result = ev->Get()->Request;
     static const TString errorMarker = "LocalYdbProxy";
@@ -108,6 +113,8 @@ TSchemeCacheHelpers::TCheckFailFunc TBaseLocalTopicPartitionActor::LeaveOnError(
 }
 
 STATEFN(TBaseLocalTopicPartitionActor::StateDescribe) {
+    YDB_LOG_CREATE_CONTEXT(LogPrefix,
+        {"actorState", "StateDescribe"});
     switch (ev->GetTypeRewrite()) {
         hFunc(TEvNavigateResult, Handle);
         hFunc(TEvents::TEvWakeup, HandleOnDescribe);
@@ -119,7 +126,8 @@ STATEFN(TBaseLocalTopicPartitionActor::StateDescribe) {
 }
 
 void TBaseLocalTopicPartitionActor::DoCreatePipe() {
-    LOG_T("Create pipe to " << PartitionTabletId);
+    YDB_LOG_TRACE("Create pipe",
+        {"partitionTabletId", PartitionTabletId});
 
     Attempt = 0;
     CreatePipe();
@@ -133,7 +141,8 @@ void TBaseLocalTopicPartitionActor::CreatePipe() {
 }
 
 void TBaseLocalTopicPartitionActor::Handle(TEvTabletPipe::TEvClientConnected::TPtr& ev) {
-    LOG_T("Handle " << ev->Get()->ToString());
+    YDB_LOG_TRACE("Handle",
+        {"ev", ev->Get()->ToString()});
 
     auto& msg = *ev->Get();
     if (msg.Status != NKikimrProto::OK) {
@@ -143,17 +152,20 @@ void TBaseLocalTopicPartitionActor::Handle(TEvTabletPipe::TEvClientConnected::TP
         return CreatePipe();
     }
 
-    LOG_T("Pipe has been connected");
+    YDB_LOG_TRACE("Pipe has been connected");
 
     OnDescribeFinished();
 }
 
 void TBaseLocalTopicPartitionActor::Handle(TEvTabletPipe::TEvClientDestroyed::TPtr& ev) {
-    LOG_T("Handle " << ev->Get()->ToString());
+    YDB_LOG_TRACE("Handle",
+        {"ev", ev->Get()->ToString()});
     OnError("Pipe destroyed");
 }
 
 STATEFN(TBaseLocalTopicPartitionActor::StateCreatePipe) {
+    YDB_LOG_CREATE_CONTEXT(LogPrefix,
+        {"actorState", "StateCreatePipe"});
     switch (ev->GetTypeRewrite()) {
         hFunc(TEvTabletPipe::TEvClientConnected, Handle);
         hFunc(TEvTabletPipe::TEvClientDestroyed, Handle);
diff --git a/ydb/core/tx/replication/ydb_proxy/local_proxy/local_partition_actor.h b/ydb/core/tx/replication/ydb_proxy/local_proxy/local_partition_actor.h
index f7cf7721a8c..111a37ccee4 100644
--- a/ydb/core/tx/replication/ydb_proxy/local_proxy/local_partition_actor.h
+++ b/ydb/core/tx/replication/ydb_proxy/local_proxy/local_partition_actor.h
@@ -21,7 +21,7 @@ protected:
     virtual void OnError(const TString& error) = 0;
     virtual void OnFatalError(const TString& error) = 0;
     virtual STATEFN(OnInitEvent) = 0;
-    virtual TString MakeLogPrefix() = 0;
+    virtual NActors::NStructuredLog::TStructuredMessage MakeLogPrefix() = 0;
 };
 
 class TBaseLocalTopicPartitionActor
@@ -71,7 +71,7 @@ protected:
     TActorId PartitionPipeClient;
 
     size_t Attempt = 0;
-    TString LogPrefix;
+    NActors::NStructuredLog::TStructuredMessage LogPrefix;
 };
 
 }
diff --git a/ydb/core/tx/replication/ydb_proxy/local_proxy/local_partition_committer.cpp b/ydb/core/tx/replication/ydb_proxy/local_proxy/local_partition_committer.cpp
index c0981576f55..f8c47e393ad 100644
--- a/ydb/core/tx/replication/ydb_proxy/local_proxy/local_partition_committer.cpp
+++ b/ydb/core/tx/replication/ydb_proxy/local_proxy/local_partition_committer.cpp
@@ -1,11 +1,14 @@
 #include "local_partition_actor.h"
 #include "local_proxy.h"
-#include "logging.h"
 
 #include <ydb/core/persqueue/events/global.h>
 #include <ydb/core/persqueue/writer/common.h>
 #include <ydb/core/tx/replication/ydb_proxy/ydb_proxy.h>
 
+#include <ydb/library/actors/core/log.h>
+
+#define YDB_LOG_THIS_FILE_COMPONENT NKikimrServices::LOCAL_YDB_PROXY
+
 namespace NKikimr::NReplication {
 
 class TLocalTopicPartitionCommitActor: public TBaseLocalTopicPartitionActor {
@@ -41,8 +44,13 @@ protected:
         PassAway();
     }
 
-    TString MakeLogPrefix() override {
-        return TStringBuilder() << "Committer[" << SelfId() << ":/" << Database << TopicPath <<" ] ";
+    NActors::NStructuredLog::TStructuredMessage MakeLogPrefix() override {
+        return YDB_LOG_CREATE_MESSAGE(
+            {"actorClassName", "LocalTopicPartitionCommitActor"},
+            {"selfId", SelfId()},
+            {"database", Database},
+            {"topicPath", TopicPath},
+        );
     }
 
     static std::unique_ptr<TEvYdbProxy::TEvCommitOffsetResponse> MakeResponse(NYdb::EStatus status, const TString& error) {
@@ -60,14 +68,15 @@ protected:
 
 private:
     void DoCommitOffset() {
-        LOG_T("DoCommit");
+        YDB_LOG_TRACE("DoCommit");
 
         NTabletPipe::SendData(SelfId(), PartitionPipeClient, MakeCommitRequest().release());
         Become(&TLocalTopicPartitionCommitActor::StateCommitOffset);
     }
 
     void Handle(TEvPersQueue::TEvResponse::TPtr& ev) {
-        LOG_T("Handle " << ev->Get()->ToString());
+        YDB_LOG_TRACE("Handle",
+            {"ev", ev->Get()->ToString()});
 
         const auto& record = ev->Get()->Record;
 
@@ -94,6 +103,8 @@ private:
     }
 
     STATEFN(StateCommitOffset) {
+        YDB_LOG_CREATE_CONTEXT(LogPrefix,
+            {"actorState", "StateCommitOffset"});
         switch (ev->GetTypeRewrite()) {
             hFunc(TEvPersQueue::TEvResponse, Handle);
 
@@ -112,7 +123,8 @@ private:
 }; // TLocalTopicPartitionCommitActor
 
 void TLocalProxyActor::Handle(TEvYdbProxy::TEvCommitOffsetRequest::TPtr& ev) {
-    LOG_T("Handle " << ev->Get()->ToString());
+    YDB_LOG_TRACE("Handle",
+        {"ev", ev->Get()->ToString()});
 
     auto [topicName, partitionId, consumerName, offset, settings] = std::move(ev->Get()->GetArgs());
     RegisterWithSameMailbox(new TLocalTopicPartitionCommitActor(
diff --git a/ydb/core/tx/replication/ydb_proxy/local_proxy/local_partition_reader.cpp b/ydb/core/tx/replication/ydb_proxy/local_proxy/local_partition_reader.cpp
index 5bbd1f42f98..6037640f61f 100644
--- a/ydb/core/tx/replication/ydb_proxy/local_proxy/local_partition_reader.cpp
+++ b/ydb/core/tx/replication/ydb_proxy/local_proxy/local_partition_reader.cpp
@@ -1,12 +1,15 @@
 #include "local_partition_actor.h"
 #include "local_proxy.h"
-#include "logging.h"
 
 #include <ydb/core/persqueue/writer/common.h>
 #include <ydb/core/persqueue/events/global.h>
 #include <ydb/core/protos/grpc_pq_old.pb.h>
 #include <ydb/core/tx/replication/ydb_proxy/ydb_proxy.h>
 
+#include <ydb/library/actors/core/log.h>
+
+#define YDB_LOG_THIS_FILE_COMPONENT NKikimrServices::LOCAL_YDB_PROXY
+
 namespace NKikimr::NReplication {
 
 class TLocalTopicPartitionReaderActor: public TBaseLocalTopicPartitionActor {
@@ -52,8 +55,12 @@ protected:
         PassAway();
     }
 
-    TString MakeLogPrefix() override {
-        return TStringBuilder() << "Reader[" << SelfId() << ":/" << Database << TopicPath <<" ] ";
+    NActors::NStructuredLog::TStructuredMessage MakeLogPrefix() override {
+        return YDB_LOG_CREATE_MESSAGE(
+            {"actorClassName", "LocalTopicPartitionReaderActor"},
+            {"selfId", SelfId()},
+            {"database", Database},
+            {"topicPath", TopicPath});
     }
 
     static std::unique_ptr<TEvYdbProxy::TEvTopicReaderGone> MakeError(NYdb::EStatus status, const TString& error) {
@@ -63,6 +70,8 @@ protected:
     }
 
     STATEFN(OnInitEvent) override {
+        YDB_LOG_CREATE_CONTEXT(LogPrefix,
+            {"actorState", "OnInitEvent"});
         switch (ev->GetTypeRewrite()) {
             hFunc(TEvYdbProxy::TEvReadTopicRequest, HandleInit);
             hFunc(TEvYdbProxy::TEvCommitOffsetRequest, Handle);
@@ -79,12 +88,14 @@ private:
     }
 
     void HandleInit(TEvYdbProxy::TEvReadTopicRequest::TPtr& ev) {
-        LOG_T("Handle on init " << ev->Get()->ToString());
+        YDB_LOG_TRACE("Handle on init",
+            {"ev", ev->Get()->ToString()});
         RequestsQueue.emplace_back(ev->Sender, ev->Cookie, GetSkipCommit(ev));
     }
 
     void Handle(TEvYdbProxy::TEvCommitOffsetRequest::TPtr& ev) {
-        LOG_T("Handle " << ev->Get()->ToString());
+        YDB_LOG_TRACE("Handle",
+            {"ev", ev->Get()->ToString()});
     }
 
 private:
@@ -94,7 +105,8 @@ private:
     }
 
     void HandleOnInitOffset(TEvPersQueue::TEvResponse::TPtr& ev) {
-        LOG_T("Handle " << ev->Get()->ToString());
+        YDB_LOG_TRACE("Handle",
+            {"ev", ev->Get()->ToString()});
 
         auto& record = ev->Get()->Record;
         if (record.GetErrorCode() == NPersQueue::NErrorCode::INITIALIZING) {
@@ -137,6 +149,8 @@ private:
     }
 
     STATEFN(StateInitOffset) {
+        YDB_LOG_CREATE_CONTEXT(LogPrefix,
+            {"actorState", "StateInitOffset"});
         switch (ev->GetTypeRewrite()) {
             hFunc(TEvPersQueue::TEvResponse, HandleOnInitOffset);
             hFunc(TEvents::TEvWakeup, HandleOnInitOffset);
@@ -158,7 +172,8 @@ private:
     }
 
     void Handle(TEvYdbProxy::TEvReadTopicRequest::TPtr& ev) {
-        LOG_T("Handle " << ev->Get()->ToString());
+        YDB_LOG_TRACE("Handle",
+            {"ev", ev->Get()->ToString()});
 
         HandleInit(ev);
         Handle(RequestsQueue.front());
@@ -204,6 +219,8 @@ private:
     }
 
     STATEFN(StateWork) {
+        YDB_LOG_CREATE_CONTEXT(LogPrefix,
+            {"actorState", "StateWork"});
         switch (ev->GetTypeRewrite()) {
             hFunc(TEvYdbProxy::TEvReadTopicRequest, Handle);
             hFunc(TEvYdbProxy::TEvCommitOffsetRequest, Handle);
@@ -226,7 +243,8 @@ private:
     }
 
     void HandleOnWaitData(TEvPersQueue::TEvResponse::TPtr& ev) {
-        LOG_T("Handle " << ev->Get()->ToString());
+        YDB_LOG_TRACE("Handle",
+            {"ev", ev->Get()->ToString()});
 
         const auto& record = ev->Get()->Record;
 
@@ -316,6 +334,8 @@ private:
     }
 
     STATEFN(StateWaitData) {
+        YDB_LOG_CREATE_CONTEXT(LogPrefix,
+            {"actorState", "StateWaitData"});
         switch (ev->GetTypeRewrite()) {
             hFunc(TEvPersQueue::TEvResponse, HandleOnWaitData);
 
@@ -339,7 +359,8 @@ private:
 }; // TLocalTopicPartitionReaderActor
 
 void TLocalProxyActor::Handle(TEvYdbProxy::TEvCreateTopicReaderRequest::TPtr& ev) {
-    LOG_T("Handle " << ev->Get()->ToString());
+        YDB_LOG_TRACE("Handle",
+            {"ev", ev->Get()->ToString()});
 
     auto args = std::move(ev->Get()->GetArgs());
     auto& settings = std::get<TEvYdbProxy::TTopicReaderSettings>(args);
diff --git a/ydb/core/tx/replication/ydb_proxy/local_proxy/local_proxy.cpp b/ydb/core/tx/replication/ydb_proxy/local_proxy/local_proxy.cpp
index a087fb7cb92..2bbe23da69c 100644
--- a/ydb/core/tx/replication/ydb_proxy/local_proxy/local_proxy.cpp
+++ b/ydb/core/tx/replication/ydb_proxy/local_proxy/local_proxy.cpp
@@ -1,12 +1,14 @@
 #include "local_proxy.h"
 #include "local_proxy_request.h"
-#include "logging.h"
 
 #include <ydb/core/grpc_services/service_scheme.h>
 #include <ydb/core/grpc_services/service_topic.h>
+#include <ydb/library/actors/core/log.h>
 #include <ydb/library/yverify_stream/yverify_stream.h>
 #include <ydb/public/api/protos/ydb_table.pb.h>
 
+#define YDB_LOG_THIS_FILE_COMPONENT NKikimrServices::LOCAL_YDB_PROXY
+
 namespace NKikimr::NReplication {
 
 TLocalProxyActor::TLocalProxyActor(const TString& database)
@@ -15,7 +17,10 @@ TLocalProxyActor::TLocalProxyActor(const TString& database)
 }
 
 void TLocalProxyActor::Bootstrap() {
-    LogPrefix = TStringBuilder() << "[" << SelfId() << ":" << Database << "] ";
+    LogPrefix = YDB_LOG_CREATE_MESSAGE(
+        {"actorClassName", "LocalTopicPartitionReaderActor"},
+        {"selfId", SelfId()},
+        {"database", Database});
     Become(&TLocalProxyActor::StateWork);
 }
 
@@ -65,7 +70,8 @@ auto CreateCallback(std::shared_ptr<T>&& ctx) {
 }
 
 void TLocalProxyActor::Handle(TEvYdbProxy::TEvAlterTopicRequest::TPtr& ev) {
-    LOG_T("Handle " << ev->Get()->ToString());
+    YDB_LOG_TRACE("Handle",
+        {"ev", ev->Get()->ToString()});
 
     auto [path, settings] = std::move(ev->Get()->GetArgs());
     path = MakeLocalPath(path);
@@ -97,7 +103,8 @@ void TLocalProxyActor::Handle(TEvYdbProxy::TEvAlterTopicRequest::TPtr& ev) {
 }
 
 void TLocalProxyActor::Handle(TEvYdbProxy::TEvDescribeTopicRequest::TPtr& ev) {
-    LOG_T("Handle " << ev->Get()->ToString());
+    YDB_LOG_TRACE("Handle",
+        {"ev", ev->Get()->ToString()});
 
     auto [path, _] = std::move(ev->Get()->GetArgs());
     path = MakeLocalPath(path);
@@ -133,7 +140,8 @@ void TLocalProxyActor::Handle(TEvYdbProxy::TEvDescribeTopicRequest::TPtr& ev) {
 }
 
 void TLocalProxyActor::Handle(TEvYdbProxy::TEvDescribePathRequest::TPtr& ev) {
-    LOG_T("Handle " << ev->Get()->ToString());
+    YDB_LOG_TRACE("Handle",
+        {"ev", ev->Get()->ToString()});
 
     auto [path, _] = std::move(ev->Get()->GetArgs());
     path = MakeLocalPath(path);
@@ -169,7 +177,8 @@ void TLocalProxyActor::Handle(TEvYdbProxy::TEvDescribePathRequest::TPtr& ev) {
 }
 
 void TLocalProxyActor::Handle(TEvYdbProxy::TEvDescribeTableRequest::TPtr& ev) {
-    LOG_E("Handle " << ev->Get()->ToString());
+    YDB_LOG_ERROR("Handle",
+        {"ev", ev->Get()->ToString()});
 
     auto [path, settings] = std::move(ev->Get()->GetArgs());
 
@@ -182,6 +191,8 @@ void TLocalProxyActor::Handle(TEvYdbProxy::TEvDescribeTableRequest::TPtr& ev) {
 }
 
 STATEFN(TLocalProxyActor::StateWork) {
+    YDB_LOG_CREATE_CONTEXT(LogPrefix,
+        {"actorState", "StateWaitData"});
     switch (ev->GetTypeRewrite()) {
         hFunc(TEvYdbProxy::TEvCreateTopicReaderRequest, Handle);
         hFunc(TEvYdbProxy::TEvAlterTopicRequest, Handle);
diff --git a/ydb/core/tx/replication/ydb_proxy/local_proxy/local_proxy.h b/ydb/core/tx/replication/ydb_proxy/local_proxy/local_proxy.h
index 0781b99bbe6..b0c4089aaef 100644
--- a/ydb/core/tx/replication/ydb_proxy/local_proxy/local_proxy.h
+++ b/ydb/core/tx/replication/ydb_proxy/local_proxy/local_proxy.h
@@ -33,7 +33,7 @@ private:
 
 private:
     const TString Database;
-    TString LogPrefix;
+    NActors::NStructuredLog::TStructuredMessage LogPrefix;
 };
 
 }
diff --git a/ydb/core/tx/replication/ydb_proxy/local_proxy/logging.h b/ydb/core/tx/replication/ydb_proxy/local_proxy/logging.h
deleted file mode 100644
index 339a6597c23..00000000000
--- a/ydb/core/tx/replication/ydb_proxy/local_proxy/logging.h
+++ /dev/null
@@ -1,18 +0,0 @@
-#pragma once
-
-#include <ydb/library/actors/core/log.h>
-#include <ydb/library/services/services.pb.h>
-
-#define CLOG_T(ctx, stream) LOG_TRACE_S (ctx, NKikimrServices::LOCAL_YDB_PROXY, LogPrefix << stream)
-#define CLOG_D(ctx, stream) LOG_DEBUG_S (ctx, NKikimrServices::LOCAL_YDB_PROXY, LogPrefix << stream)
-#define CLOG_I(ctx, stream) LOG_INFO_S  (ctx, NKikimrServices::LOCAL_YDB_PROXY, LogPrefix << stream)
-#define CLOG_N(ctx, stream) LOG_NOTICE_S(ctx, NKikimrServices::LOCAL_YDB_PROXY, LogPrefix << stream)
-#define CLOG_W(ctx, stream) LOG_WARN_S  (ctx, NKikimrServices::LOCAL_YDB_PROXY, LogPrefix << stream)
-#define CLOG_E(ctx, stream) LOG_ERROR_S (ctx, NKikimrServices::LOCAL_YDB_PROXY, LogPrefix << stream)
-
-#define LOG_T(stream) CLOG_T(*TlsActivationContext, stream)
-#define LOG_D(stream) CLOG_D(*TlsActivationContext, stream)
-#define LOG_I(stream) CLOG_I(*TlsActivationContext, stream)
-#define LOG_N(stream) CLOG_N(*TlsActivationContext, stream)
-#define LOG_W(stream) CLOG_W(*TlsActivationContext, stream)
-#define LOG_E(stream) CLOG_E(*TlsActivationContext, stream)