Remote address in audit log

ydb/core/client/server/msgbus_blobstorage_config.cpp

@@ -27,6 +27,7 @@ class TMessageBusBlobStorageConfig
     {
         SetSecurityToken(token);
         SetRequireAdminAccess(true);
+        SetPeerName(msg.GetPeerName());
     }
 
     void Handle(TEvBlobStorage::TEvControllerConfigResponse::TPtr &ev, const TActorContext &ctx) {

ydb/core/client/server/msgbus_server.cpp

@@ -16,6 +16,7 @@ class TBusMessageContext::TImpl : public TThrRefBase {
     virtual void SendReplyMove(NBus::TBusMessageAutoPtr response) = 0;
     virtual TVector<TStringBuf> FindClientCert() const = 0;
     virtual THolder<TMessageBusSessionIdentHolder::TImpl> CreateSessionIdentHolder() = 0;
+    virtual TString GetPeerName() const = 0;
 };
 
 class TBusMessageContext::TImplMessageBus
@@ -61,6 +62,13 @@ class TBusMessageContext::TImplMessageBus
         return {};
     }
 
+    TString GetPeerName() const override {
+        TStringBuilder ret;
+        if (IsConnectionAlive()) {
+            ret << GetPeerAddrNetAddr();
+        }
+        return std::move(ret);
+    }
 
     THolder<TMessageBusSessionIdentHolder::TImpl> CreateSessionIdentHolder() override;
 };
@@ -181,6 +189,10 @@ class TBusMessageContext::TImplGRpc
     };
 
     THolder<TMessageBusSessionIdentHolder::TImpl> CreateSessionIdentHolder() override;
+
+    TString GetPeerName() const override {
+        return RequestContext->GetPeer();
+    }
 };
 
 TBusMessageContext::TBusMessageContext()
@@ -227,6 +239,8 @@ void TBusMessageContext::Swap(TBusMessageContext &msg) {
 
 TVector<TStringBuf> TBusMessageContext::FindClientCert() const { return Impl->FindClientCert(); }
 
+TString TBusMessageContext::GetPeerName() const { return Impl->GetPeerName(); }
+
 THolder<TMessageBusSessionIdentHolder::TImpl> TBusMessageContext::CreateSessionIdentHolder() {
     Y_ABORT_UNLESS(Impl);
     return Impl->CreateSessionIdentHolder();

ydb/core/client/server/msgbus_server.h

@@ -87,6 +87,7 @@ class TBusMessageContext {
     void SendReplyMove(NBus::TBusMessageAutoPtr response);
     void Swap(TBusMessageContext& msg);
     TVector<TStringBuf> FindClientCert() const;
+    TString GetPeerName() const;
 
 private:
     friend class TMessageBusSessionIdentHolder;

ydb/core/client/server/msgbus_server_cms.cpp

@@ -31,6 +31,7 @@ class TCmsRequestActor : public TMessageBusSecureRequest<TMessageBusServerReques
         , Request(request)
     {
         TBase::SetSecurityToken(request.GetSecurityToken());
+        TBase::SetPeerName(msg.GetPeerName());
     }
 
     void Bootstrap(const TActorContext &ctx)

ydb/core/client/server/msgbus_server_console.cpp

@@ -44,6 +44,7 @@ class TConsoleRequestActor : public TMessageBusSecureRequest<TMessageBusServerRe
             TBase::SetRequireAdminAccess(true);
         }
 
+        TBase::SetPeerName(msg.GetPeerName());
     }
 
     void Bootstrap(const TActorContext &ctx)

ydb/core/client/server/msgbus_server_drain_node.cpp

@@ -20,6 +20,7 @@ class TMessageBusDrainNode : public TMessageBusSecureRequest<TMessageBusTabletRe
     {
         SetSecurityToken(Request->Record.GetSecurityToken());
         SetRequireAdminAccess(true);
+        SetPeerName(msg.GetPeerName());
     }
 
     std::pair<ui64, TAutoPtr<IEventBase>> MakeReqPair(const TActorContext& ctx) {

ydb/core/client/server/msgbus_server_fill_node.cpp

@@ -20,6 +20,7 @@ class TMessageBusFillNode : public TMessageBusSecureRequest<TMessageBusTabletReq
     {
         SetSecurityToken(Request->Record.GetSecurityToken());
         SetRequireAdminAccess(true);
+        SetPeerName(msg.GetPeerName());
     }
 
     std::pair<ui64, TAutoPtr<IEventBase>> MakeReqPair(const TActorContext& ctx) {

ydb/core/client/server/msgbus_server_local_minikql.cpp

@@ -17,6 +17,7 @@ class TMessageBusLocalMKQL : public TMessageBusSecureRequest<TMessageBusSimpleTa
     {
         SetSecurityToken(static_cast<TBusTabletLocalMKQL*>(msg.GetMessage())->Record.GetSecurityToken());
         SetRequireAdminAccess(true);
+        SetPeerName(msg.GetPeerName());
     }
 
     void Handle(TEvTablet::TEvLocalMKQLResponse::TPtr &ev, const TActorContext &ctx) {

ydb/core/client/server/msgbus_server_local_scheme_tx.cpp

@@ -21,6 +21,7 @@ class TMessageBusLocalSchemeTx : public TMessageBusSecureRequest<TMessageBusSimp
         Request.Swap(&request);
         TBase::SetSecurityToken(Request.GetSecurityToken());
         TBase::SetRequireAdminAccess(true);
+        TBase::SetPeerName(msg.GetPeerName());
     }
 
     void Handle(TEvTablet::TEvLocalSchemeTxResponse::TPtr &ev, const TActorContext &ctx) {

ydb/core/client/server/msgbus_server_node_registration.cpp

@@ -40,6 +40,7 @@ class TNodeRegistrationActor : public TMessageBusSecureRequest<TMessageBusServer
         } else {
             TBase::SetSecurityToken(BUILTIN_ACL_ROOT); // NBS compatibility
         }
+        TBase::SetPeerName(msg.GetPeerName());
     }
 
     void Bootstrap(const TActorContext &ctx)

ydb/core/client/server/msgbus_server_proxy.cpp

@@ -67,6 +67,7 @@ class TMessageBusServerFlatDescribeRequest : public TMessageBusSecureRequest<TMe
         , Request(static_cast<TBusSchemeDescribe*>(msg->MsgContext.ReleaseMessage()))
     {
         TBase::SetSecurityToken(Request->Record.GetSecurityToken());
+        TBase::SetPeerName(msg->MsgContext.GetPeerName());
     }
 
     //STFUNC(StateWork)

ydb/core/client/server/msgbus_server_request.cpp

@@ -55,6 +55,7 @@ class TMessageBusServerRequest : public TMessageBusSecureRequest<TMessageBusServ
     {
         TBase::SetSecurityToken(Request->Record.GetSecurityToken());
         TBase::SetRequireAdminAccess(true); // MiniKQL and ReadTable execution required administative access
+        TBase::SetPeerName(msg->MsgContext.GetPeerName());
     }
 
     //STFUNC(StateWork)
@@ -72,6 +73,7 @@ class TMessageBusServerRequest : public TMessageBusSecureRequest<TMessageBusServ
         ProposalStatus.Reset(new NKikimrTxUserProxy::TEvProposeTransactionStatus());
         Proposal.Reset(new TEvTxUserProxy::TEvProposeTransaction());
         NKikimrTxUserProxy::TEvProposeTransaction &record = Proposal->Record;
+        record.SetPeerName(GetPeerName());
 
         // Transaction protobuf structure might be very heavy (if it has a batch of parameters)
         // so we don't want to copy it, just move its contents

ydb/core/client/server/msgbus_server_scheme_initroot.cpp

@@ -54,6 +54,7 @@ class TMessageBusSchemeInitRoot : public TMessageBusSecureRequest<TMessageBusSer
     {
         TBase::SetSecurityToken(Request->Record.GetSecurityToken());
         TBase::SetRequireAdminAccess(true);
+        TBase::SetPeerName(msg->MsgContext.GetPeerName());
     }
 
     void Bootstrap(const TActorContext &ctx) {

ydb/core/client/server/msgbus_server_scheme_request.cpp

@@ -74,6 +74,7 @@ class TMessageBusServerSchemeRequest : public TMessageBusSecureRequest<TMessageB
     {
         TBase::SetSecurityToken(Request->Record.GetSecurityToken());
         TBase::SetRequireAdminAccess(true);
+        TBase::SetPeerName(msg->MsgContext.GetPeerName());
     }
 
     //STFUNC(StateWork)
@@ -95,6 +96,7 @@ template <>
 void TMessageBusServerSchemeRequest<TBusPersQueue>::SendProposeRequest(const TActorContext &ctx) {
     TAutoPtr<TEvTxUserProxy::TEvProposeTransaction> req(new TEvTxUserProxy::TEvProposeTransaction());
     NKikimrTxUserProxy::TEvProposeTransaction &record = req->Record;
+    record.SetPeerName(GetPeerName());
 
     if (Request->Record.HasMetaRequest() && Request->Record.GetMetaRequest().HasCmdCreateTopic()) {
         const auto& cmd = Request->Record.GetMetaRequest().GetCmdCreateTopic();
@@ -157,6 +159,7 @@ template <>
 void TMessageBusServerSchemeRequest<TBusSchemeOperation>::SendProposeRequest(const TActorContext &ctx) {
     TAutoPtr<TEvTxUserProxy::TEvProposeTransaction> req(new TEvTxUserProxy::TEvProposeTransaction());
     NKikimrTxUserProxy::TEvProposeTransaction &record = req->Record;
+    record.SetPeerName(GetPeerName());
 
     if (!Request->Record.HasTransaction()) {
         return HandleError(MSTATUS_ERROR, TEvTxUserProxy::TResultStatus::Unknown, "Malformed request: no modify scheme transaction provided", ctx);

ydb/core/grpc_services/query/rpc_execute_script.cpp

@@ -137,6 +137,7 @@ class TExecuteScriptRPC : public TRpcRequestActor<TExecuteScriptRPC, TEvExecuteS
         SetAuthToken(ev, *Request);
         SetDatabase(ev, *Request);
         SetRlPath(ev, *Request);
+        ev->Record.MutableRequest()->SetClientAddress(Request->GetPeerName());
 
         if (traceId) {
             ev->Record.SetTraceId(traceId.GetRef());

ydb/core/grpc_services/query/rpc_kqp_tx.cpp

@@ -68,6 +68,7 @@ class TBeginTransactionRPC : public  TActorBootstrapped<TBeginTransactionRPC> {
         auto ev = MakeHolder<NKqp::TEvKqp::TEvQueryRequest>();
         SetAuthToken(ev, *Request);
         SetDatabase(ev, *Request);
+        ev->Record.MutableRequest()->SetClientAddress(Request->GetPeerName());
 
         if (CheckSession(req->session_id(), Request.get())) {
             ev->Record.MutableRequest()->SetSessionId(req->session_id());
@@ -189,6 +190,7 @@ class TFinishTransactionRPC : public  TActorBootstrapped<TFinishTransactionRPC>
         auto ev = MakeHolder<NKqp::TEvKqp::TEvQueryRequest>();
         SetAuthToken(ev, *Request);
         SetDatabase(ev, *Request);
+        ev->Record.MutableRequest()->SetClientAddress(Request->GetPeerName());
 
         const auto& [sessionId, txId] = GetReqData();
 

ydb/core/grpc_services/rpc_begin_transaction.cpp

@@ -56,6 +56,7 @@ class TBeginTransactionRPC : public TRpcKqpRequestActor<TBeginTransactionRPC, TE
         auto ev = MakeHolder<NKqp::TEvKqp::TEvQueryRequest>();
         SetAuthToken(ev, *Request_);
         SetDatabase(ev, *Request_);
+        ev->Record.MutableRequest()->SetClientAddress(Request_->GetPeerName());
 
         if (CheckSession(req->session_id(), Request_.get())) {
             ev->Record.MutableRequest()->SetSessionId(req->session_id());

ydb/core/grpc_services/rpc_commit_transaction.cpp

@@ -54,6 +54,7 @@ class TCommitTransactionRPC : public TRpcKqpRequestActor<TCommitTransactionRPC,
         auto ev = MakeHolder<NKqp::TEvKqp::TEvQueryRequest>();
         SetAuthToken(ev, *Request_);
         SetDatabase(ev, *Request_);
+        ev->Record.MutableRequest()->SetClientAddress(Request_->GetPeerName());
 
         if (CheckSession(req->session_id(), Request_.get())) {
             ev->Record.MutableRequest()->SetSessionId(req->session_id());

ydb/core/grpc_services/rpc_dynamic_config.cpp

@@ -205,6 +205,7 @@ class TDynamicConfigRPC : public TRpcOperationRequestActor<TDynamicConfigRPC<TRe
         auto request = MakeHolder<TConsoleRequest>();
         request->Record.MutableRequest()->CopyFrom(*this->GetProtoRequest());
         request->Record.SetUserToken(this->Request_->GetSerializedToken());
+        request->Record.SetPeerName(this->Request_->GetPeerName());
         NTabletPipe::SendData(IActor::SelfId(), ConsolePipe, request.Release());
     }
 };

ydb/core/grpc_services/rpc_execute_scheme_query.cpp

@@ -50,6 +50,7 @@ class TExecuteSchemeQueryRPC : public TRpcKqpRequestActor<TExecuteSchemeQueryRPC
         auto ev = MakeHolder<NKqp::TEvKqp::TEvQueryRequest>();
         SetAuthToken(ev, *Request_);
         SetDatabase(ev, *Request_);
+        ev->Record.MutableRequest()->SetClientAddress(Request_->GetPeerName());
 
         if (CheckSession(req->session_id(), Request_.get())) {
             ev->Record.MutableRequest()->SetSessionId(req->session_id());

ydb/core/grpc_services/rpc_explain_data_query.cpp

@@ -50,6 +50,7 @@ class TExplainDataQueryRPC : public TRpcKqpRequestActor<TExplainDataQueryRPC, TE
         auto ev = MakeHolder<NKqp::TEvKqp::TEvQueryRequest>();
         SetAuthToken(ev, *Request_);
         SetDatabase(ev, *Request_);
+        ev->Record.MutableRequest()->SetClientAddress(Request_->GetPeerName());
 
         if (traceId) {
             ev->Record.SetTraceId(traceId.GetRef());

ydb/core/grpc_services/rpc_explain_yql_script.cpp

@@ -42,6 +42,7 @@ class TExplainYqlScriptRPC : public TRpcKqpRequestActor<TExplainYqlScriptRPC, TE
         auto ev = MakeHolder<NKqp::TEvKqp::TEvQueryRequest>();
         SetAuthToken(ev, *Request_);
         SetDatabase(ev, *Request_);
+        ev->Record.MutableRequest()->SetClientAddress(Request_->GetPeerName());
 
         if (traceId) {
             ev->Record.SetTraceId(traceId.GetRef());

ydb/core/grpc_services/rpc_prepare_data_query.cpp

@@ -54,6 +54,7 @@ class TPrepareDataQueryRPC : public TRpcKqpRequestActor<TPrepareDataQueryRPC, TE
         auto ev = MakeHolder<NKqp::TEvKqp::TEvQueryRequest>();
         SetAuthToken(ev, *Request_);
         SetDatabase(ev, *Request_);
+        ev->Record.MutableRequest()->SetClientAddress(Request_->GetPeerName());
 
         AuditContextAppend(Request_.get(), *req);
 

ydb/core/grpc_services/rpc_rollback_transaction.cpp

@@ -53,6 +53,7 @@ class TRollbackTransactionRPC : public TRpcKqpRequestActor<TRollbackTransactionR
         auto ev = MakeHolder<NKqp::TEvKqp::TEvQueryRequest>();
         SetAuthToken(ev, *Request_);
         SetDatabase(ev, *Request_);
+        ev->Record.MutableRequest()->SetClientAddress(Request_->GetPeerName());
 
         if (CheckSession(req->session_id(), Request_.get())) {
             ev->Record.MutableRequest()->SetSessionId(req->session_id());

ydb/core/kqp/common/compilation/events.h

@@ -14,13 +14,14 @@
 namespace NKikimr::NKqp::NPrivateEvents {
 
 struct TEvCompileRequest: public TEventLocal<TEvCompileRequest, TKqpEvents::EvCompileRequest> {
-    TEvCompileRequest(const TIntrusiveConstPtr<NACLib::TUserToken>& userToken, const TMaybe<TString>& uid,
+    TEvCompileRequest(const TIntrusiveConstPtr<NACLib::TUserToken>& userToken, const TString& clientAddress, const TMaybe<TString>& uid,
         TMaybe<TKqpQueryId>&& query, bool keepInCache, bool isQueryActionPrepare, bool perStatementResult, TInstant deadline,
         TKqpDbCountersPtr dbCounters, const TGUCSettings::TPtr& gUCSettings, const TMaybe<TString>& applicationName,
         std::shared_ptr<std::atomic<bool>> intrestedInResult, const TIntrusivePtr<TUserRequestContext>& userRequestContext, NLWTrace::TOrbit orbit = {},
         TKqpTempTablesState::TConstPtr tempTablesState = nullptr, bool collectDiagnostics = false, TMaybe<TQueryAst> queryAst = Nothing(),
         bool split = false, NYql::TExprContext* splitCtx = nullptr, NYql::TExprNode::TPtr splitExpr = nullptr)
         : UserToken(userToken)
+        , ClientAddress(clientAddress)
         , Uid(uid)
         , Query(std::move(query))
         , KeepInCache(keepInCache)
@@ -44,6 +45,7 @@ struct TEvCompileRequest: public TEventLocal<TEvCompileRequest, TKqpEvents::EvCo
     }
 
     TIntrusiveConstPtr<NACLib::TUserToken> UserToken;
+    TString ClientAddress;
     TMaybe<TString> Uid;
     TMaybe<TKqpQueryId> Query;
     bool KeepInCache = false;
@@ -72,13 +74,14 @@ struct TEvCompileRequest: public TEventLocal<TEvCompileRequest, TKqpEvents::EvCo
 };
 
 struct TEvRecompileRequest: public TEventLocal<TEvRecompileRequest, TKqpEvents::EvRecompileRequest> {
-    TEvRecompileRequest(const TIntrusiveConstPtr<NACLib::TUserToken>& userToken, const TString& uid,
+    TEvRecompileRequest(const TIntrusiveConstPtr<NACLib::TUserToken>& userToken, const TString& clientAddress, const TString& uid,
         const TMaybe<TKqpQueryId>& query, bool isQueryActionPrepare, TInstant deadline,
         TKqpDbCountersPtr dbCounters, const TGUCSettings::TPtr& gUCSettings, const TMaybe<TString>& applicationName,
         std::shared_ptr<std::atomic<bool>> intrestedInResult, const TIntrusivePtr<TUserRequestContext>& userRequestContext,
         NLWTrace::TOrbit orbit = {}, TKqpTempTablesState::TConstPtr tempTablesState = nullptr, TMaybe<TQueryAst> queryAst = Nothing(),
         bool split = false, NYql::TExprContext* splitCtx = nullptr, NYql::TExprNode::TPtr splitExpr = nullptr)
         : UserToken(userToken)
+        , ClientAddress(clientAddress)
         , Uid(uid)
         , Query(query)
         , IsQueryActionPrepare(isQueryActionPrepare)
@@ -98,6 +101,7 @@ struct TEvRecompileRequest: public TEventLocal<TEvRecompileRequest, TKqpEvents::
     }
 
     TIntrusiveConstPtr<NACLib::TUserToken> UserToken;
+    TString ClientAddress;
     TString Uid;
     TMaybe<TKqpQueryId> Query;
     bool IsQueryActionPrepare = false;

ydb/core/kqp/common/events/query.h

@@ -228,6 +228,14 @@ struct TEvQueryRequest: public NActors::TEventLocal<TEvQueryRequest, TKqpEvents:
         return Token_;
     }
 
+    TString GetClientAddress() const {
+        if (RequestCtx) {
+            return RequestCtx->GetPeerName();
+        }
+
+        return Record.GetRequest().GetClientAddress();
+    }
+
     const ::google::protobuf::Map<TProtoStringType, ::Ydb::TypedValue>& GetYdbParameters() const {
         if (YdbParameters) {
             return *YdbParameters;

ydb/core/kqp/common/kqp_event_impl.cpp

@@ -49,6 +49,7 @@ void TEvKqp::TEvQueryRequest::PrepareRemote() const {
         if (RequestCtx->GetSerializedToken()) {
             Record.SetUserToken(RequestCtx->GetSerializedToken());
         }
+        Record.MutableRequest()->SetClientAddress(RequestCtx->GetPeerName());
 
         Record.MutableRequest()->SetDatabase(Database);
         ActorIdToProto(RequestActorId, Record.MutableCancelationActor());

ydb/core/kqp/compile_service/kqp_compile_actor.cpp

@@ -48,7 +48,7 @@ class TKqpCompileActor : public TActorBootstrapped<TKqpCompileActor> {
         TIntrusivePtr<TModuleResolverState> moduleResolverState, TIntrusivePtr<TKqpCounters> counters,
         const TGUCSettings::TPtr& gUCSettings,
         const TMaybe<TString>& applicationName, const TString& uid, const TKqpQueryId& queryId,
-        const TIntrusiveConstPtr<NACLib::TUserToken>& userToken,
+        const TIntrusiveConstPtr<NACLib::TUserToken>& userToken, const TString& clientAddress,
         TKqpDbCountersPtr dbCounters, std::optional<TKqpFederatedQuerySetup> federatedQuerySetup,
         const TIntrusivePtr<TUserRequestContext>& userRequestContext,
         NWilson::TTraceId traceId, TKqpTempTablesState::TConstPtr tempTablesState, bool collectFullDiagnostics,
@@ -66,6 +66,7 @@ class TKqpCompileActor : public TActorBootstrapped<TKqpCompileActor> {
         , QueryId(queryId)
         , QueryRef(QueryId.Text, QueryId.QueryParameterTypes, queryAst)
         , UserToken(userToken)
+        , ClientAddress(clientAddress)
         , DbCounters(dbCounters)
         , Config(MakeIntrusive<TKikimrConfiguration>())
         , QueryServiceConfig(queryServiceConfig)
@@ -269,6 +270,7 @@ class TKqpCompileActor : public TActorBootstrapped<TKqpCompileActor> {
         Gateway = CreateKikimrIcGateway(QueryId.Cluster, QueryId.Settings.QueryType, QueryId.Database, QueryId.DatabaseId, std::move(loader),
             ctx.ExecutorThread.ActorSystem, ctx.SelfID.NodeId(), counters, QueryServiceConfig);
         Gateway->SetToken(QueryId.Cluster, UserToken);
+        Gateway->SetClientAddress(ClientAddress);
 
         Config->FeatureFlags = AppData(ctx)->FeatureFlags;
 
@@ -544,6 +546,7 @@ class TKqpCompileActor : public TActorBootstrapped<TKqpCompileActor> {
     TKqpQueryId QueryId;
     TKqpQueryRef QueryRef;
     TIntrusiveConstPtr<NACLib::TUserToken> UserToken;
+    TString ClientAddress;
     TKqpDbCountersPtr DbCounters;
     TKikimrConfiguration::TPtr Config;
     TQueryServiceConfig QueryServiceConfig;
@@ -614,7 +617,7 @@ IActor* CreateKqpCompileActor(const TActorId& owner, const TKqpSettings::TConstP
     const TTableServiceConfig& tableServiceConfig,
     const TQueryServiceConfig& queryServiceConfig,
     TIntrusivePtr<TModuleResolverState> moduleResolverState, TIntrusivePtr<TKqpCounters> counters,
-    const TString& uid, const TKqpQueryId& query, const TIntrusiveConstPtr<NACLib::TUserToken>& userToken,
+    const TString& uid, const TKqpQueryId& query, const TIntrusiveConstPtr<NACLib::TUserToken>& userToken, const TString& clientAddress,
     std::optional<TKqpFederatedQuerySetup> federatedQuerySetup, TKqpDbCountersPtr dbCounters, const TGUCSettings::TPtr& gUCSettings,
     const TMaybe<TString>& applicationName, const TIntrusivePtr<TUserRequestContext>& userRequestContext,
     NWilson::TTraceId traceId, TKqpTempTablesState::TConstPtr tempTablesState,
@@ -623,7 +626,7 @@ IActor* CreateKqpCompileActor(const TActorId& owner, const TKqpSettings::TConstP
 {
     return new TKqpCompileActor(owner, kqpSettings, tableServiceConfig, queryServiceConfig,
                                 moduleResolverState, counters, gUCSettings, applicationName,
-                                uid, query, userToken, dbCounters,
+                                uid, query, userToken, clientAddress, dbCounters,
                                 federatedQuerySetup, userRequestContext,
                                 std::move(traceId), std::move(tempTablesState), collectFullDiagnostics,
                                 perStatementResult, compileAction, std::move(queryAst),