Improve logging (#30427)

Author: FloatingCrowbar

ydb/core/persqueue/dread_cache_service/caching_service.cpp

@@ -70,7 +70,7 @@ class TPQDirectReadCacheService : public TActorBootstrapped<TPQDirectReadCacheSe
         auto sessionIter = ServerSessions.find(key);
         if (sessionIter.IsEnd()) {
             PQ_CPROXY_LOG_D("unknown session id '" << key.SessionId << "', close session");
-            CloseSession(ev->Sender, Ydb::PersQueue::ErrorCode::ErrorCode::BAD_REQUEST, "Unknown session");
+            CloseSession(ev->Sender, key.SessionId, Ydb::PersQueue::ErrorCode::ErrorCode::BAD_REQUEST, "Unknown session");
             return;
         }
 
@@ -338,7 +338,7 @@ class TPQDirectReadCacheService : public TActorBootstrapped<TPQDirectReadCacheSe
         if (nextData == sessionIter->second.Reads.end()) {
             return false;
         }
-        auto result = SendData(sessionIter->first.PartitionSessionId, client, nextData->first, nextData->second);
+        auto result = SendData(sessionIter->first.SessionId, sessionIter->first.PartitionSessionId, client, nextData->first, nextData->second);
         ChangeCounterValue("SendDataRate", 1, false, true);
         if (!result) {
             //ToDo: for discuss. Error in parsing partition response - shall we kill the entire session or just the partition session?
@@ -350,7 +350,7 @@ class TPQDirectReadCacheService : public TActorBootstrapped<TPQDirectReadCacheSe
     }
 
     [[nodiscard]] bool SendData(
-            ui64 partSessionId, TCacheClientContext& proxyClient, ui64 readId, const std::shared_ptr<NKikimrClient::TResponse>& response
+            const TString& sessionId, ui64 partSessionId, TCacheClientContext& proxyClient, ui64 readId, const std::shared_ptr<NKikimrClient::TResponse>& response
     ) {
         const auto& ctx = ActorContext();
         auto message = std::make_shared<StreamDirectReadMessage::FromServer>();
@@ -359,7 +359,7 @@ class TPQDirectReadCacheService : public TActorBootstrapped<TPQDirectReadCacheSe
         directReadMessage->set_partition_session_id(partSessionId);
         directReadMessage->set_bytes_size(response->GetPartitionResponse().GetCmdPrepareReadResult().GetBytesSizeEstimate());
 
-        auto ok = VaildatePartitionResponse(proxyClient, *response);
+        auto ok = VaildatePartitionResponse(sessionId, proxyClient, * response);
         if (!ok) {
             return false;
         }
@@ -368,20 +368,21 @@ class TPQDirectReadCacheService : public TActorBootstrapped<TPQDirectReadCacheSe
                         partSessionId);
         message->set_status(Ydb::StatusIds::SUCCESS);
 
-        PQ_CPROXY_LOG_D("send data to client. AssignId: " << partSessionId << ", readId: " << readId);
+        PQ_CPROXY_LOG_D("send data to client " << sessionId << ", assignId: " << partSessionId << ", readId: " << readId);
 
         ctx.Send(proxyClient.ProxyId, new TEvPQProxy::TEvDirectReadSendClientData(std::move(message)));
         return true;
     }
 
     void CloseSession(
             const TActorId& proxyId,
+            const TString& sessionId,
             Ydb::PersQueue::ErrorCode::ErrorCode code,
             const TString& reason
     ) {
         const auto& ctx = ActorContext();
         ctx.Send(proxyId, new TEvPQProxy::TEvDirectReadCloseSession(code, reason));
-        PQ_CPROXY_LOG_D("close session for proxy " << proxyId.ToString());
+        PQ_CPROXY_LOG_D("close session for proxy " << proxyId.ToString() << ", sessionId: " << sessionId);
     }
 
     bool DestroyPartitionSession(
@@ -395,7 +396,7 @@ class TPQDirectReadCacheService : public TActorBootstrapped<TPQDirectReadCacheSe
         ctx.Send(
                 sessionIter->second.Client->ProxyId, new TEvPQProxy::TEvDirectReadDestroyPartitionSession(sessionIter->first, code, reason)
         );
-        PQ_CPROXY_LOG_D("close session for proxy " << sessionIter->second.Client->ProxyId.ToString());
+        PQ_CPROXY_LOG_D("DestroyPartitionSession, sessionId: " << sessionIter->first.SessionId << ", proxy: " << sessionIter->second.Client->ProxyId.ToString());
         return true;
     }
 
@@ -412,11 +413,12 @@ class TPQDirectReadCacheService : public TActorBootstrapped<TPQDirectReadCacheSe
     }
 
     bool VaildatePartitionResponse(
-            TCacheClientContext& proxyClient, NKikimrClient::TResponse& response
+            const TString& sessionId, TCacheClientContext& proxyClient, NKikimrClient::TResponse& response
     ) {
         if (response.HasErrorCode() && response.GetErrorCode() != NPersQueue::NErrorCode::OK) {
             CloseSession(
                     proxyClient.ProxyId,
+                    sessionId,
                     NGRpcProxy::V1::ConvertOldCode(response.GetErrorCode()),
                     "Status is not ok: " + response.GetErrorReason()
             );
@@ -426,6 +428,7 @@ class TPQDirectReadCacheService : public TActorBootstrapped<TPQDirectReadCacheSe
         if (response.GetStatus() != NKikimr::NMsgBusProxy::MSTATUS_OK) { //this is incorrect answer, die
             CloseSession(
                     proxyClient.ProxyId,
+                    sessionId,
                     Ydb::PersQueue::ErrorCode::ERROR,
                     "Status is not ok: " + response.GetErrorReason()
             );
@@ -434,6 +437,7 @@ class TPQDirectReadCacheService : public TActorBootstrapped<TPQDirectReadCacheSe
         if (!response.HasPartitionResponse()) { //this is incorrect answer, die
             CloseSession(
                     proxyClient.ProxyId,
+                    sessionId,
                     Ydb::PersQueue::ErrorCode::ERROR,
                     "Direct read cache got empty partition response"
             );
@@ -444,6 +448,7 @@ class TPQDirectReadCacheService : public TActorBootstrapped<TPQDirectReadCacheSe
         if (!partResponse.HasCmdReadResult()) { //this is incorrect answer, die
             CloseSession(
                     proxyClient.ProxyId,
+                    sessionId,
                     Ydb::PersQueue::ErrorCode::ERROR,
                     "Malformed response from partition"
             );

ydb/services/persqueue_v1/actors/direct_read_actor.cpp

@@ -485,6 +485,9 @@ void TDirectReadSessionActor::RunAuthActor(const TActorContext& ctx) {
 }
 
 void TDirectReadSessionActor::HandleDestroyPartitionSession(TEvPQProxy::TEvDirectReadDestroyPartitionSession::TPtr& ev) {
+    const auto& ctx = ActorContext();
+    LOG_DEBUG_S(ctx, NKikimrServices::PQ_READ_PROXY, PQ_LOG_PREFIX << " got EvDirectReadDestroyPartitionSession, assignId: " << ev->Get()->ReadKey.PartitionSessionId);
+
     TServerMessage result;
     result.set_status(Ydb::StatusIds::SUCCESS);
     auto* stop = result.mutable_stop_direct_read_partition_session();
@@ -500,6 +503,8 @@ void TDirectReadSessionActor::HandleSessionKilled(TEvPQProxy::TEvDirectReadClose
 }
 
 void TDirectReadSessionActor::HandleGotData(TEvPQProxy::TEvDirectReadSendClientData::TPtr& ev) {
+    const auto& ctx = ActorContext();
+    LOG_DEBUG_S(ctx, NKikimrServices::PQ_READ_PROXY, PQ_LOG_PREFIX << " got direct read data, message size: " << ev->Get()->Message->ByteSizeLong());
     auto formedResponse = MakeIntrusive<TFormedDirectReadResponse>();
     formedResponse->Response = std::move(ev->Get()->Message);
     ProcessAnswer(formedResponse, ActorContext());