diff --git a/ydb/core/control/lib/generated/codegen/main.cpp b/ydb/core/control/lib/generated/codegen/main.cpp index b6403aab18d18..bca55ed96b8be 100644 --- a/ydb/core/control/lib/generated/codegen/main.cpp +++ b/ydb/core/control/lib/generated/codegen/main.cpp @@ -172,6 +172,7 @@ void CodeGenRequestConfigsInner(TCodeGenContext& context); std::vector GetRequestConfigsServices() { return { "CoordinationService_Session", + "CoordinationService_ListSemaphores", "ClickhouseInternal_Scan", "ClickhouseInternal_GetShardLocations", "ClickhouseInternal_DescribeTable", diff --git a/ydb/core/kesus/proxy/proxy_actor.cpp b/ydb/core/kesus/proxy/proxy_actor.cpp index 7efc77244e0ed..ebf1901c062a1 100644 --- a/ydb/core/kesus/proxy/proxy_actor.cpp +++ b/ydb/core/kesus/proxy/proxy_actor.cpp @@ -530,6 +530,16 @@ class TKesusProxyActor : public TActorBootstrapped { } } + void Handle(TEvKesus::TEvListSemaphores::TPtr& ev) { + auto msg = ev->Release(); + HandleDirectRequest(ev->Sender, ev->Cookie, std::move(msg)); + } + + void Handle(TEvKesus::TEvListSemaphoresResult::TPtr& ev) { + auto msg = ev->Release(); + HandleDirectResponse(ev->Cookie, std::move(msg)); + } + void Handle(TEvKesus::TEvAttachSession::TPtr& ev) { KPROXY_LOG_TRACE_S("Received TEvAttachSession from " << ev->Sender); Y_ABORT_UNLESS(ev->Sender); @@ -823,6 +833,8 @@ class TKesusProxyActor : public TActorBootstrapped { hFunc(TEvKesus::TEvUpdateSemaphoreResult, Handle); hFunc(TEvKesus::TEvDeleteSemaphore, Handle); hFunc(TEvKesus::TEvDeleteSemaphoreResult, Handle); + hFunc(TEvKesus::TEvListSemaphores, Handle); + hFunc(TEvKesus::TEvListSemaphoresResult, Handle); hFunc(TEvKesus::TEvAttachSession, Handle); hFunc(TEvKesus::TEvAttachSessionResult, Handle); hFunc(TEvKesus::TEvProxyExpired, Handle); diff --git a/ydb/core/kesus/tablet/events.h b/ydb/core/kesus/tablet/events.h index 8b5b2de401076..07c5dde88fd25 100644 --- a/ydb/core/kesus/tablet/events.h +++ b/ydb/core/kesus/tablet/events.h @@ -58,6 +58,9 @@ namespace TEvKesus { EvReleaseSemaphore, EvReleaseSemaphoreResult, + EvListSemaphores, + EvListSemaphoresResult, + // Notifications EvProxyExpired = EvBegin + 512, EvSessionExpired, @@ -249,6 +252,30 @@ namespace TEvKesus { using TResultBase::TResultBase; }; + struct TEvListSemaphores : public TEventPB { + TEvListSemaphores() = default; + + TEvListSemaphores(const TString& kesusPath, bool includeDetails) { + Record.SetKesusPath(kesusPath); + Record.SetProxyGeneration(0); + Record.SetIncludeDetails(includeDetails); + } + }; + + struct TEvListSemaphoresResult : public TEventPB { + TEvListSemaphoresResult() = default; + + explicit TEvListSemaphoresResult(ui64 generation) { + Record.SetProxyGeneration(generation); + Record.MutableError()->SetStatus(Ydb::StatusIds::SUCCESS); + } + + TEvListSemaphoresResult(ui64 generation, Ydb::StatusIds::StatusCode status, const TString& reason) { + Record.SetProxyGeneration(generation); + FillError(Record.MutableError(), status, reason); + } + }; + struct TEvRegisterProxy : public TEventPB { TEvRegisterProxy() = default; diff --git a/ydb/core/kesus/tablet/tablet_impl.cpp b/ydb/core/kesus/tablet/tablet_impl.cpp index e53aaaf24a3c9..16c1a2968e66d 100644 --- a/ydb/core/kesus/tablet/tablet_impl.cpp +++ b/ydb/core/kesus/tablet/tablet_impl.cpp @@ -279,6 +279,7 @@ STFUNC(TKesusTablet::StateWork) { hFunc(TEvKesus::TEvAcquireSemaphore, Handle); hFunc(TEvKesus::TEvCreateSemaphore, Handle); hFunc(TEvKesus::TEvDescribeSemaphore, Handle); + hFunc(TEvKesus::TEvListSemaphores, Handle); hFunc(TEvKesus::TEvDeleteSemaphore, Handle); hFunc(TEvKesus::TEvReleaseSemaphore, Handle); hFunc(TEvKesus::TEvUpdateSemaphore, Handle); diff --git a/ydb/core/kesus/tablet/tablet_impl.h b/ydb/core/kesus/tablet/tablet_impl.h index 6d40117715d8d..68fa765f55f5b 100644 --- a/ydb/core/kesus/tablet/tablet_impl.h +++ b/ydb/core/kesus/tablet/tablet_impl.h @@ -50,6 +50,8 @@ class TKesusTablet : public TActor, public NTabletFlatExecutor::TT struct TTxSemaphoreTimeout; struct TTxSemaphoreUpdate; + struct TTxSemaphoreList; + struct TTxQuoterResourceAdd; struct TTxQuoterResourceUpdate; struct TTxQuoterResourceDelete; @@ -393,6 +395,7 @@ class TKesusTablet : public TActor, public NTabletFlatExecutor::TT void Handle(TEvKesus::TEvSetConfig::TPtr& ev); void Handle(TEvKesus::TEvGetConfig::TPtr& ev); void Handle(TEvKesus::TEvDescribeSemaphore::TPtr& ev); + void Handle(TEvKesus::TEvListSemaphores::TPtr& ev); void Handle(TEvKesus::TEvDescribeProxies::TPtr& ev); void Handle(TEvKesus::TEvDescribeSessions::TPtr& ev); void Handle(TEvKesus::TEvRegisterProxy::TPtr& ev); diff --git a/ydb/core/kesus/tablet/tablet_ut.cpp b/ydb/core/kesus/tablet/tablet_ut.cpp index bb3ed2a32f439..19727ceea2752 100644 --- a/ydb/core/kesus/tablet/tablet_ut.cpp +++ b/ydb/core/kesus/tablet/tablet_ut.cpp @@ -80,6 +80,54 @@ Y_UNIT_TEST_SUITE(TKesusTest) { ctx.SetConfig(12345, MakeConfig("/foo/bar/baz"), 41, Ydb::StatusIds::PRECONDITION_FAILED); } + Y_UNIT_TEST(TestListSemaphoresFromTablet) { + TTestContext ctx; + ctx.Setup(); + + // Locks are acquired via TEvAcquireSemaphore and appear in Semaphores; keep this test focused on + // explicit CreateSemaphore entries only (no proxy/session lock setup needed for ListSemaphores). + + UNIT_ASSERT_VALUES_EQUAL(ctx.ListSemaphoresFromTablet(false).GetSemaphoreDescriptions().size(), 0u); + + ctx.CreateSemaphore("Alpha", 1); + ctx.CreateSemaphore("Beta", 1); + ctx.UpdateSemaphore("Alpha", "meta-alpha"); + + { + const auto r = ctx.ListSemaphoresFromTablet(false); + UNIT_ASSERT_VALUES_EQUAL(r.GetSemaphoreDescriptions().size(), 2u); + ui32 alphaCount = 0; + ui32 betaCount = 0; + for (const auto& d : r.GetSemaphoreDescriptions()) { + UNIT_ASSERT_VALUES_EQUAL(d.owners_size(), 0); + UNIT_ASSERT_VALUES_EQUAL(d.waiters_size(), 0); + if (d.name() == "Alpha") { + ++alphaCount; + } else if (d.name() == "Beta") { + ++betaCount; + } + } + UNIT_ASSERT_VALUES_EQUAL(alphaCount, 1u); + UNIT_ASSERT_VALUES_EQUAL(betaCount, 1u); + } + + { + const auto r = ctx.ListSemaphoresFromTablet(true); + UNIT_ASSERT_VALUES_EQUAL(r.GetSemaphoreDescriptions().size(), 2u); + for (const auto& d : r.GetSemaphoreDescriptions()) { + UNIT_ASSERT_VALUES_EQUAL(d.owners_size(), 0u); + UNIT_ASSERT_VALUES_EQUAL(d.waiters_size(), 0u); + if (d.name() == "Alpha") { + UNIT_ASSERT_VALUES_EQUAL(d.data(), "meta-alpha"); + } else if (d.name() == "Beta") { + UNIT_ASSERT_VALUES_EQUAL(d.data(), ""); + } else { + UNIT_FAIL("unexpected semaphore"); + } + } + } + } + Y_UNIT_TEST(TestRegisterProxy) { TTestContext ctx; ctx.Setup(); diff --git a/ydb/core/kesus/tablet/tx_semaphore_list.cpp b/ydb/core/kesus/tablet/tx_semaphore_list.cpp new file mode 100644 index 0000000000000..8e9f10e7796e2 --- /dev/null +++ b/ydb/core/kesus/tablet/tx_semaphore_list.cpp @@ -0,0 +1,100 @@ +#include "tablet_impl.h" + +namespace NKikimr { +namespace NKesus { + +struct TKesusTablet::TTxSemaphoreList : public TTxBase { + const TActorId Sender; + const ui64 Cookie; + const NKikimrKesus::TEvListSemaphores Record; + + THolder Reply; + + TTxSemaphoreList(TSelf* self, const TActorId& sender, ui64 cookie, const NKikimrKesus::TEvListSemaphores& record) + : TTxBase(self) + , Sender(sender) + , Cookie(cookie) + , Record(record) + {} + + TTxType GetTxType() const override { return TXTYPE_SEMAPHORE_LIST; } + + bool Execute(TTransactionContext& txc, const TActorContext& ctx) override { + LOG_DEBUG_S(ctx, NKikimrServices::KESUS_TABLET, + "[" << Self->TabletID() << "] TTxSemaphoreList::Execute (sender=" << Sender + << ", cookie=" << Cookie << ")"); + + NIceDb::TNiceDb db(txc.DB); + + if (Record.GetProxyGeneration() != 0) { + Reply.Reset(new TEvKesus::TEvListSemaphoresResult( + Record.GetProxyGeneration(), + Ydb::StatusIds::BAD_REQUEST, + "Only direct proxy generation 0 is supported for listing semaphores")); + return true; + } + + if (Self->UseStrictRead()) { + Self->PersistStrictMarker(db); + } + + Reply.Reset(new TEvKesus::TEvListSemaphoresResult(0)); + + TVector sorted; + sorted.reserve(Self->Semaphores.size()); + for (auto& kv : Self->Semaphores) { + sorted.push_back(&kv.second); + } + Sort(sorted.begin(), sorted.end(), [](const TSemaphoreInfo* a, const TSemaphoreInfo* b) { + return a->Name < b->Name; + }); + + const bool includeDetails = Record.GetIncludeDetails(); + for (TSemaphoreInfo* semaphore : sorted) { + auto* desc = Reply->Record.AddSemaphoreDescriptions(); + desc->set_name(semaphore->Name); + desc->set_limit(semaphore->Limit); + desc->set_ephemeral(semaphore->Ephemeral); + desc->set_count(semaphore->Count); + if (includeDetails) { + desc->set_data(semaphore->Data); + for (const auto* owner : semaphore->Owners) { + auto* p = desc->add_owners(); + p->set_order_id(owner->OrderId); + p->set_session_id(owner->SessionId); + p->set_count(owner->Count); + p->set_data(owner->Data); + } + for (const auto& kv : semaphore->Waiters) { + auto* waiter = kv.second; + auto* p = desc->add_waiters(); + p->set_order_id(waiter->OrderId); + p->set_session_id(waiter->SessionId); + p->set_timeout_millis(waiter->TimeoutMillis); + p->set_count(waiter->Count); + p->set_data(waiter->Data); + } + } + } + + return true; + } + + void Complete(const TActorContext& ctx) override { + LOG_DEBUG_S(ctx, NKikimrServices::KESUS_TABLET, + "[" << Self->TabletID() << "] TTxSemaphoreList::Complete (sender=" << Sender + << ", cookie=" << Cookie << ")"); + Y_ABORT_UNLESS(Reply); + ctx.Send(Sender, Reply.Release(), 0, Cookie); + } +}; + +void TKesusTablet::Handle(TEvKesus::TEvListSemaphores::TPtr& ev) { + const auto& record = ev->Get()->Record; + VerifyKesusPath(record.GetKesusPath()); + + Execute(new TTxSemaphoreList(this, ev->Sender, ev->Cookie, record), TActivationContext::AsActorContext()); +} + +} +} diff --git a/ydb/core/kesus/tablet/ut_helpers.cpp b/ydb/core/kesus/tablet/ut_helpers.cpp index 5c9af3ee5b7d9..2ec3ffa9d7977 100644 --- a/ydb/core/kesus/tablet/ut_helpers.cpp +++ b/ydb/core/kesus/tablet/ut_helpers.cpp @@ -170,6 +170,15 @@ NKikimrKesus::TEvGetConfigResult TTestContext::GetConfig() { return result->Record; } +NKikimrKesus::TEvListSemaphoresResult TTestContext::ListSemaphoresFromTablet(bool includeDetails) { + const ui64 cookie = RandomNumber(); + const auto edge = Runtime->AllocateEdgeActor(); + SendFromEdge(edge, new TEvKesus::TEvListSemaphores("", includeDetails), cookie); + + auto result = ExpectEdgeEvent(edge, cookie); + return result->Record; +} + NKikimrKesus::TEvSetConfigResult TTestContext::SetConfig(ui64 txId, const Ydb::Coordination::Config& config, ui64 version, Ydb::StatusIds::StatusCode status) { const ui64 cookie = RandomNumber(); const auto edge = Runtime->AllocateEdgeActor(); diff --git a/ydb/core/kesus/tablet/ut_helpers.h b/ydb/core/kesus/tablet/ut_helpers.h index 52632d5c9a584..b03771e8489f4 100644 --- a/ydb/core/kesus/tablet/ut_helpers.h +++ b/ydb/core/kesus/tablet/ut_helpers.h @@ -76,6 +76,7 @@ struct TTestContext { // set/get config requests NKikimrKesus::TEvGetConfigResult GetConfig(); + NKikimrKesus::TEvListSemaphoresResult ListSemaphoresFromTablet(bool includeDetails = false); NKikimrKesus::TEvSetConfigResult SetConfig(ui64 txId, const Ydb::Coordination::Config& config, ui64 version, Ydb::StatusIds::StatusCode status = Ydb::StatusIds::SUCCESS); // Makes a dummy request using this proxy/generation pair diff --git a/ydb/core/kesus/tablet/ya.make b/ydb/core/kesus/tablet/ya.make index ce4b14c6818a9..4be030f50c0ac 100644 --- a/ydb/core/kesus/tablet/ya.make +++ b/ydb/core/kesus/tablet/ya.make @@ -25,6 +25,7 @@ SRCS( tx_semaphore_create.cpp tx_semaphore_delete.cpp tx_semaphore_describe.cpp + tx_semaphore_list.cpp tx_semaphore_release.cpp tx_semaphore_timeout.cpp tx_semaphore_update.cpp diff --git a/ydb/core/protos/counters_kesus.proto b/ydb/core/protos/counters_kesus.proto index 33df5de6266fc..3b5590c16f9f9 100644 --- a/ydb/core/protos/counters_kesus.proto +++ b/ydb/core/protos/counters_kesus.proto @@ -75,4 +75,5 @@ enum ETxTypes { TXTYPE_QUOTER_RESOURCE_ADD = 19 [(TxTypeOpts) = {Name: "TxQouterResourceAdd"}]; TXTYPE_QUOTER_RESOURCE_UPDATE = 20 [(TxTypeOpts) = {Name: "TxQouterResourceUpdate"}]; TXTYPE_QUOTER_RESOURCE_DELETE = 21 [(TxTypeOpts) = {Name: "TxQouterResourceDelete"}]; + TXTYPE_SEMAPHORE_LIST = 22 [(TxTypeOpts) = {Name: "TxSemaphoreList"}]; } diff --git a/ydb/core/protos/kesus.proto b/ydb/core/protos/kesus.proto index 2bf2fb000d37d..425c2aea965d0 100644 --- a/ydb/core/protos/kesus.proto +++ b/ydb/core/protos/kesus.proto @@ -139,6 +139,18 @@ message TEvDeleteSemaphoreResult { TKesusError Error = 2; } +message TEvListSemaphores { + string KesusPath = 1; + uint64 ProxyGeneration = 2; + bool IncludeDetails = 3; +} + +message TEvListSemaphoresResult { + uint64 ProxyGeneration = 1; + TKesusError Error = 2; + repeated Ydb.Coordination.SemaphoreDescription SemaphoreDescriptions = 3; +} + message TEvRegisterProxy { string KesusPath = 1; uint64 ProxyGeneration = 2; diff --git a/ydb/public/api/grpc/ydb_coordination_v1.proto b/ydb/public/api/grpc/ydb_coordination_v1.proto index c4120a2cd0240..10571722e2d40 100644 --- a/ydb/public/api/grpc/ydb_coordination_v1.proto +++ b/ydb/public/api/grpc/ydb_coordination_v1.proto @@ -31,4 +31,7 @@ service CoordinationService { // Describes a coordination node rpc DescribeNode(Coordination.DescribeNodeRequest) returns (Coordination.DescribeNodeResponse); + + // Lists semaphores in a coordination node (stream of SemaphoreDescription) + rpc ListSemaphores(Coordination.ListSemaphoresRequest) returns (stream Coordination.SemaphoreDescription); } diff --git a/ydb/public/api/protos/ydb_coordination.proto b/ydb/public/api/protos/ydb_coordination.proto index 378045e78e83a..1b41268efe5ed 100644 --- a/ydb/public/api/protos/ydb_coordination.proto +++ b/ydb/public/api/protos/ydb_coordination.proto @@ -494,3 +494,14 @@ message DescribeNodeResult { Ydb.Scheme.Entry self = 1; Config config = 2; } + +/** + * Lists semaphores stored in a coordination node (server-streaming RPC). + * With include_details = false only compact fields are filled (name, limit, ephemeral, count). + * With include_details = true the response matches DescribeSemaphore with owners and waiters included. + */ +message ListSemaphoresRequest { + string path = 1; + bool include_details = 2; + Ydb.Operations.OperationParams operation_params = 3; +} diff --git a/ydb/public/sdk/cpp/include/ydb-cpp-sdk/client/coordination/coordination.h b/ydb/public/sdk/cpp/include/ydb-cpp-sdk/client/coordination/coordination.h index bb03bc4e4c3e8..847d2f407fe5c 100644 --- a/ydb/public/sdk/cpp/include/ydb-cpp-sdk/client/coordination/coordination.h +++ b/ydb/public/sdk/cpp/include/ydb-cpp-sdk/client/coordination/coordination.h @@ -1,12 +1,14 @@ #pragma once #include +#include namespace Ydb { namespace Coordination { class Config; class CreateNodeRequest; class DescribeNodeResult; + class ListSemaphoresRequest; class SemaphoreDescription; class SemaphoreSession; } @@ -205,6 +207,40 @@ struct TDescribeNodeSettings : public TOperationRequestSettings { + FLUENT_SETTING_DEFAULT(bool, IncludeDetails, false); +}; + +class TListSemaphoresPart : public TStreamPartStatus { +public: + bool HasSemaphoreDescription() const { + return Description_.has_value(); + } + + const TSemaphoreDescription& GetSemaphoreDescription() const { + return *Description_; + } + + explicit TListSemaphoresPart(TStatus&& status) + : TStreamPartStatus(std::move(status)) + {} + + TListSemaphoresPart(TStatus&& status, TSemaphoreDescription&& description) + : TStreamPartStatus(std::move(status)) + , Description_(std::move(description)) + {} + +private: + std::optional Description_; +}; + +using TAsyncListSemaphoresPart = NThreading::TFuture; + +class TListSemaphoresIterator; +using TAsyncListSemaphoresIterator = NThreading::TFuture; + +//////////////////////////////////////////////////////////////////////////////// + class TSession; using TSessionResult = TResult; @@ -293,6 +329,8 @@ struct TDescribeSemaphoreSettings { class TClient { public: + class TImpl; + TClient(const TDriver& driver, const TCommonClientSettings& settings = TCommonClientSettings()); ~TClient(); @@ -312,11 +350,27 @@ class TClient { TAsyncDescribeNodeResult DescribeNode(const std::string& path, const TDescribeNodeSettings& settings = TDescribeNodeSettings()); + TAsyncListSemaphoresIterator ListSemaphores(const std::string& path, + const TListSemaphoresSettings& settings = TListSemaphoresSettings()); + private: - class TImpl; std::shared_ptr Impl_; }; +class TListSemaphoresIterator : public TStatus { + friend class TClient::TImpl; + +public: + TAsyncListSemaphoresPart ReadNext(); + +private: + struct TReaderImpl; + + TListSemaphoresIterator(std::shared_ptr impl, TPlainStatus&& status); + + std::shared_ptr ReaderImpl_; +}; + //////////////////////////////////////////////////////////////////////////////// class TSessionContext; diff --git a/ydb/public/sdk/cpp/src/client/coordination/coordination.cpp b/ydb/public/sdk/cpp/src/client/coordination/coordination.cpp index 71d3853bd6adc..a13ebf6885061 100644 --- a/ydb/public/sdk/cpp/src/client/coordination/coordination.cpp +++ b/ydb/public/sdk/cpp/src/client/coordination/coordination.cpp @@ -7,6 +7,7 @@ #include #include +#include #include #include @@ -123,6 +124,67 @@ void TNodeDescription::SerializeTo(Ydb::Coordination::CreateNodeRequest& creatio //////////////////////////////////////////////////////////////////////////////// +struct TListSemaphoresIterator::TReaderImpl { +public: + using TStreamProcessorPtr = NYdbGrpc::IStreamRequestReadProcessor::TPtr; + using TGRpcStatus = NYdbGrpc::TGrpcStatus; + + TReaderImpl(TStreamProcessorPtr streamProcessor, std::string endpoint) + : StreamProcessor_(std::move(streamProcessor)) + , Endpoint_(std::move(endpoint)) + {} + + ~TReaderImpl() { + StreamProcessor_->Cancel(); + } + + bool IsFinished() const { + return Finished_; + } + + TAsyncListSemaphoresPart ReadNext(std::shared_ptr self) { + auto promise = NewPromise(); + auto readCb = [self, promise](TGRpcStatus&& grpcStatus) mutable { + if (!grpcStatus.Ok()) { + self->Finished_ = true; + if (grpcStatus.GRpcStatusCode == grpc::StatusCode::OUT_OF_RANGE) { + promise.SetValue(TListSemaphoresPart(TStatus(TPlainStatus( + EStatus::SUCCESS, NYdb::NIssue::TIssues{}, self->Endpoint_, {})))); + } else { + promise.SetValue(TListSemaphoresPart(TStatus(TPlainStatus( + std::move(grpcStatus), self->Endpoint_, {})))); + } + } else { + promise.SetValue(TListSemaphoresPart( + TStatus(TPlainStatus(EStatus::SUCCESS, NYdb::NIssue::TIssues{}, self->Endpoint_, {})), + TSemaphoreDescription(self->Response_))); + } + }; + StreamProcessor_->Read(&Response_, readCb); + return promise.GetFuture(); + } + +private: + TStreamProcessorPtr StreamProcessor_; + Ydb::Coordination::SemaphoreDescription Response_; + bool Finished_ = false; + std::string Endpoint_; +}; + +TListSemaphoresIterator::TListSemaphoresIterator(std::shared_ptr impl, TPlainStatus&& status) + : TStatus(std::move(status)) + , ReaderImpl_(std::move(impl)) +{} + +TAsyncListSemaphoresPart TListSemaphoresIterator::ReadNext() { + if (!ReaderImpl_ || ReaderImpl_->IsFinished()) { + RaiseError("Attempt to perform read on invalid or finished stream"); + } + return ReaderImpl_->ReadNext(ReaderImpl_); +} + +//////////////////////////////////////////////////////////////////////////////// + TSemaphoreSession::TSemaphoreSession() { OrderId_ = 0; SessionId_ = 0; @@ -1928,6 +1990,38 @@ class TClient::TImpl : public TClientImplCommon { return promise.GetFuture(); } + + TAsyncListSemaphoresIterator ListSemaphores( + const std::string& path, + const TListSemaphoresSettings& settings) + { + auto request = MakeOperationRequest(settings); + request.set_path(TStringType{path}); + request.set_include_details(settings.IncludeDetails_); + + auto promise = NewPromise(); + + Connections_->StartReadStream( + request, + [promise](TPlainStatus status, + NYdbGrpc::IStreamRequestReadProcessor::TPtr processor) mutable { + if (!status.Ok()) { + promise.SetValue(TListSemaphoresIterator(nullptr, std::move(status))); + return; + } + auto impl = std::make_shared( + std::move(processor), + status.Endpoint); + promise.SetValue(TListSemaphoresIterator(impl, std::move(status))); + }, + &Ydb::Coordination::V1::CoordinationService::Stub::AsyncListSemaphores, + DbDriverState_, + TRpcRequestSettings::Make(settings)); + + return promise.GetFuture(); + } }; TClient::TClient(const TDriver& driver, const TCommonClientSettings& settings) @@ -1983,6 +2077,13 @@ TAsyncDescribeNodeResult TClient::DescribeNode( return Impl_->DescribeNode(std::move(request), settings); } +TAsyncListSemaphoresIterator TClient::ListSemaphores( + const std::string& path, + const TListSemaphoresSettings& settings) +{ + return Impl_->ListSemaphores(path, settings); +} + //////////////////////////////////////////////////////////////////////////////// class TSession::TImpl { diff --git a/ydb/services/kesus/grpc_list_semaphores.cpp b/ydb/services/kesus/grpc_list_semaphores.cpp new file mode 100644 index 0000000000000..9b86859484640 --- /dev/null +++ b/ydb/services/kesus/grpc_list_semaphores.cpp @@ -0,0 +1,169 @@ +#include "grpc_list_semaphores.h" + +#include +#include +#include + +#include +#include +#include +#include +#include + +#include + +#include + +namespace NKikimr { +namespace NKesus { + +namespace { + +grpc::StatusCode YdbStatusToGrpcStatus(Ydb::StatusIds::StatusCode status) { + switch (status) { + case Ydb::StatusIds::NOT_FOUND: + return grpc::NOT_FOUND; + case Ydb::StatusIds::BAD_REQUEST: + return grpc::INVALID_ARGUMENT; + case Ydb::StatusIds::UNAUTHORIZED: + return grpc::PERMISSION_DENIED; + case Ydb::StatusIds::UNAVAILABLE: + return grpc::UNAVAILABLE; + default: + return grpc::INTERNAL; + } +} + +class TGRpcListSemaphoresActor : public TActorBootstrapped { +public: + explicit TGRpcListSemaphoresActor(TIntrusivePtr grpcRequest) + : GrpcRequest_(std::move(grpcRequest)) + {} + + void Bootstrap(const TActorContext& ctx) { + Y_UNUSED(ctx); + + const auto* protoReq = dynamic_cast(GrpcRequest_->GetRequest()); + if (!protoReq || protoReq->path().empty()) { + GrpcRequest_->ReplyError(grpc::INVALID_ARGUMENT, "Invalid ListSemaphores request"); + return PassAway(); + } + + KesusPath_ = protoReq->path(); + IncludeDetails_ = protoReq->include_details(); + + auto dbVals = GrpcRequest_->GetPeerMetaValues(TStringBuf("x-ydb-database")); + if (!dbVals.empty()) { + Database_ = TString{dbVals[0]}; + } + + auto tokenVals = GrpcRequest_->GetPeerMetaValues(TStringBuf("x-ydb-auth-ticket")); + if (!tokenVals.empty() && !tokenVals[0].empty()) { + UserToken_ = std::make_unique(TString{tokenVals[0]}); + } + + if (!Send(MakeKesusProxyServiceId(), new TEvKesusProxy::TEvResolveKesusProxy(Database_, KesusPath_))) { + GrpcRequest_->ReplyError(grpc::UNIMPLEMENTED, "Coordination service not implemented on this server"); + return PassAway(); + } + + Become(&TThis::StateResolve); + } + +private: + bool CheckAccess(ui32 access) const { + if (UserToken_) { + if (!SecurityObject_) { + return true; + } + return SecurityObject_->CheckAccess(access, *UserToken_); + } + return true; + } + + void FailWithKesusError(const NKikimrKesus::TKesusError& err) { + TString msg = err.IssuesSize() ? TString{err.GetIssues(0).Getmessage()} : TString{"Coordination error"}; + GrpcRequest_->ReplyError(YdbStatusToGrpcStatus(err.GetStatus()), msg); + PassAway(); + } + + void HandleResolve(const TEvKesusProxy::TEvAttachProxyActor::TPtr& ev) { + const TActorContext& ctx = ActorContext(); + ProxyActor_ = ev->Get()->ProxyActor; + SecurityObject_ = ev->Get()->SecurityObject; + + const bool readAllowed = CheckAccess(NACLib::EAccessRights::SelectRow); + if (!readAllowed) { + GrpcRequest_->ReplyError(grpc::PERMISSION_DENIED, "Read permission denied"); + return PassAway(); + } + + ctx.Send(ProxyActor_, + new TEvKesus::TEvListSemaphores(KesusPath_, IncludeDetails_), + 0, + RequestCookie_); + + Become(&TThis::StateWaitTablet); + } + + void HandleProxyError(const TEvKesusProxy::TEvProxyError::TPtr& ev) { + FailWithKesusError(ev->Get()->Error); + } + + void HandleListResult(const TEvKesus::TEvListSemaphoresResult::TPtr& ev) { + const auto& record = ev->Get()->Record; + if (record.GetError().GetStatus() != Ydb::StatusIds::SUCCESS) { + FailWithKesusError(record.GetError()); + return; + } + + const auto& items = record.GetSemaphoreDescriptions(); + for (const auto& src : items) { + auto* msg = google::protobuf::Arena::CreateMessage(GrpcRequest_->GetArena()); + msg->CopyFrom(src); + GrpcRequest_->Reply(msg, 0); + } + + GrpcRequest_->FinishStreamingOk(); + PassAway(); + } + + STFUNC(StateResolve) { + switch (ev->GetTypeRewrite()) { + hFunc(TEvKesusProxy::TEvProxyError, HandleProxyError); + hFunc(TEvKesusProxy::TEvAttachProxyActor, HandleResolve); + default: + Y_ABORT("Unexpected event 0x%x for TGRpcListSemaphoresActor::StateResolve", ev->GetTypeRewrite()); + } + } + + STFUNC(StateWaitTablet) { + switch (ev->GetTypeRewrite()) { + hFunc(TEvKesusProxy::TEvProxyError, HandleProxyError); + hFunc(TEvKesus::TEvListSemaphoresResult, HandleListResult); + default: + Y_ABORT("Unexpected event 0x%x for TGRpcListSemaphoresActor::StateWaitTablet", ev->GetTypeRewrite()); + } + } + +private: + TIntrusivePtr GrpcRequest_; + TString Database_; + TString KesusPath_; + bool IncludeDetails_ = false; + std::unique_ptr UserToken_; + TActorId ProxyActor_; + TIntrusivePtr SecurityObject_; + static constexpr ui64 RequestCookie_ = 1; +}; + +} // namespace + +void StartGRpcListSemaphores(NActors::TActorSystem* actorSystem, NYdbGrpc::IRequestContextBase* grpcRequest) { + Y_ABORT_UNLESS(actorSystem); + Y_ABORT_UNLESS(grpcRequest); + actorSystem->Register(new TGRpcListSemaphoresActor(TIntrusivePtr(grpcRequest))); +} + +} +} diff --git a/ydb/services/kesus/grpc_list_semaphores.h b/ydb/services/kesus/grpc_list_semaphores.h new file mode 100644 index 0000000000000..6676ba8e74d27 --- /dev/null +++ b/ydb/services/kesus/grpc_list_semaphores.h @@ -0,0 +1,13 @@ +#pragma once + +#include + +namespace NYdbGrpc { +class IRequestContextBase; +} + +namespace NKikimr::NKesus { + +void StartGRpcListSemaphores(NActors::TActorSystem* actorSystem, NYdbGrpc::IRequestContextBase* grpcRequest); + +} diff --git a/ydb/services/kesus/grpc_service.cpp b/ydb/services/kesus/grpc_service.cpp index 5eed4a2a57b9e..c261d25ad8d0e 100644 --- a/ydb/services/kesus/grpc_service.cpp +++ b/ydb/services/kesus/grpc_service.cpp @@ -1,4 +1,5 @@ #include "grpc_service.h" +#include "grpc_list_semaphores.h" #include #include @@ -646,6 +647,10 @@ void TKesusGRpcService::SetupIncomingRequests(NYdbGrpc::TLoggerPtr logger) { #error SETUP_KESUS_STREAM_METHOD macro already defined #endif +#ifdef SETUP_KESUS_LIST_SEMAPHORES_STREAM +#error SETUP_KESUS_LIST_SEMAPHORES_STREAM macro already defined +#endif + #define SETUP_KESUS_METHOD(methodName, methodCallback, rlMode, requestType, auditMode) \ for (auto* cq : CQS) { \ SETUP_RUNTIME_EVENT_METHOD(methodName, \ @@ -668,6 +673,32 @@ void TKesusGRpcService::SetupIncomingRequests(NYdbGrpc::TLoggerPtr logger) { #define GET_LIMITER_BY_PATH(ICB_PATH) \ getLimiter(#ICB_PATH, icb.ICB_PATH, DEFAULT_MAX_SESSIONS_INFLIGHT) +// Server-streaming read (unary ListSemaphoresRequest, stream SemaphoreDescription). Uses TGRpcRequest's async-writer +// path like Session, but runs in-process on the Kesus GRpc service instead of TEvCoordinationSessionRequest. +#define SETUP_KESUS_LIST_SEMAPHORES_STREAM() \ + for (auto* cq : CQS) { \ + MakeIntrusive<::NKikimr::NGRpcService::TGRpcRequest< \ + Ydb::Coordination::ListSemaphoresRequest, \ + Ydb::Coordination::SemaphoreDescription, \ + TKesusGRpcService>>( \ + this, \ + &Service_, \ + cq, \ + [this](NYdbGrpc::IRequestContextBase* reqCtx) { \ + ::NKikimr::NGRpcService::ReportGrpcReqToMon( \ + *ActorSystem_, \ + reqCtx->GetPeer(), \ + GetSdkBuildInfoIfNeeded(reqCtx)); \ + StartGRpcListSemaphores(ActorSystem_, reqCtx); \ + }, \ + &Ydb::Coordination::V1::CoordinationService::AsyncService::RequestListSemaphores, \ + "ListSemaphores", \ + logger, \ + getCounterBlock("coordination", "ListSemaphores", true), \ + GET_LIMITER_BY_PATH(GRpcControls.RequestConfigs.CoordinationService_ListSemaphores.MaxInFlight) \ + )->Run(); \ + } + #define SETUP_KESUS_STREAM_METHOD(methodName, rlMode, requestType, auditMode, operationCallClass) \ for (auto* cq : CQS) { \ SETUP_RUNTIME_EVENT_STREAM_METHOD(methodName, \ @@ -689,11 +720,13 @@ void TKesusGRpcService::SetupIncomingRequests(NYdbGrpc::TLoggerPtr logger) { SETUP_KESUS_METHOD(AlterNode, DoAlterCoordinationNode, RLSWITCH(Rps), UNSPECIFIED, TAuditMode::Modifying(TAuditMode::TLogClassConfig::Ddl)); SETUP_KESUS_METHOD(DropNode, DoDropCoordinationNode, RLSWITCH(Rps), UNSPECIFIED, TAuditMode::Modifying(TAuditMode::TLogClassConfig::Ddl)); SETUP_KESUS_METHOD(DescribeNode, DoDescribeCoordinationNode, RLSWITCH(Rps), UNSPECIFIED, TAuditMode::NonModifying()); + SETUP_KESUS_LIST_SEMAPHORES_STREAM(); SETUP_KESUS_STREAM_METHOD(Session, RLMODE(Off), UNSPECIFIED, TAuditMode::NonModifying(), NGRpcService::TEvCoordinationSessionRequest); #undef GET_LIMITER_BY_PATH #undef SETUP_KESUS_METHOD #undef SETUP_KESUS_STREAM_METHOD +#undef SETUP_KESUS_LIST_SEMAPHORES_STREAM } } // namespace NKesus diff --git a/ydb/services/kesus/ya.make b/ydb/services/kesus/ya.make index 6a32f51a99a20..ecd7f6cdc58de 100644 --- a/ydb/services/kesus/ya.make +++ b/ydb/services/kesus/ya.make @@ -1,10 +1,12 @@ LIBRARY() SRCS( + grpc_list_semaphores.cpp grpc_service.cpp ) PEERDIR( + ydb/library/aclib ydb/library/grpc/server ydb/core/base ydb/core/grpc_services diff --git a/ydb/services/ydb/ydb_coordination_ut.cpp b/ydb/services/ydb/ydb_coordination_ut.cpp index 8f9dd0e1643cb..af53b55658f04 100644 --- a/ydb/services/ydb/ydb_coordination_ut.cpp +++ b/ydb/services/ydb/ydb_coordination_ut.cpp @@ -9,6 +9,9 @@ #include #include +#include +#include + namespace NKikimr { using namespace Tests; @@ -64,6 +67,36 @@ struct TClientContext { } }; +void AssertSemaphoreSessionsEqual( + const NYdb::NCoordination::TSemaphoreSession& a, + const NYdb::NCoordination::TSemaphoreSession& b) +{ + UNIT_ASSERT_VALUES_EQUAL(a.GetOrderId(), b.GetOrderId()); + UNIT_ASSERT_VALUES_EQUAL(a.GetSessionId(), b.GetSessionId()); + UNIT_ASSERT_VALUES_EQUAL(a.GetCount(), b.GetCount()); + UNIT_ASSERT_VALUES_EQUAL(a.GetData(), b.GetData()); + UNIT_ASSERT_VALUES_EQUAL(a.GetTimeout(), b.GetTimeout()); +} + +void AssertSemaphoreDescriptionMatchesDescribe( + const NYdb::NCoordination::TSemaphoreDescription& listDesc, + const NYdb::NCoordination::TSemaphoreDescription& describeDesc) +{ + UNIT_ASSERT_VALUES_EQUAL(listDesc.GetName(), describeDesc.GetName()); + UNIT_ASSERT_VALUES_EQUAL(listDesc.GetData(), describeDesc.GetData()); + UNIT_ASSERT_VALUES_EQUAL(listDesc.GetCount(), describeDesc.GetCount()); + UNIT_ASSERT_VALUES_EQUAL(listDesc.GetLimit(), describeDesc.GetLimit()); + UNIT_ASSERT_VALUES_EQUAL(listDesc.IsEphemeral(), describeDesc.IsEphemeral()); + UNIT_ASSERT_VALUES_EQUAL(listDesc.GetOwners().size(), describeDesc.GetOwners().size()); + for (size_t i = 0; i < listDesc.GetOwners().size(); ++i) { + AssertSemaphoreSessionsEqual(listDesc.GetOwners()[i], describeDesc.GetOwners()[i]); + } + UNIT_ASSERT_VALUES_EQUAL(listDesc.GetWaiters().size(), describeDesc.GetWaiters().size()); + for (size_t i = 0; i < listDesc.GetWaiters().size(); ++i) { + AssertSemaphoreSessionsEqual(listDesc.GetWaiters()[i], describeDesc.GetWaiters()[i]); + } +} + template class TSimpleQueue : public TThrRefBase { private: @@ -275,6 +308,98 @@ Y_UNIT_TEST_SUITE(TGRpcNewCoordinationClient) { EStatus::NOT_FOUND); } + Y_UNIT_TEST(ListSemaphoresStream) { + TKikimrWithGrpcAndRootSchema server; + TClientContext context(server); + + ExpectSuccess(context.Client.CreateNode("/Root/node1")); + + auto session1 = ExpectSuccess( + context.Client.StartSession("/Root/node1", + TSessionSettings().Timeout(TDuration::Seconds(30)))); + ExpectSuccess(session1.CreateSemaphore("SemA", 3)); + ExpectSuccess(session1.CreateSemaphore("SemB", 2)); + ExpectSuccess(session1.UpdateSemaphore("SemA", "meta-a")); + UNIT_ASSERT(ExpectSuccess( + session1.AcquireSemaphore("SemA", + TAcquireSemaphoreSettings() + .Count(2) + .Data("owner-a")))); + + auto session2 = ExpectSuccess( + context.Client.StartSession("/Root/node1", + TSessionSettings().Timeout(TDuration::Seconds(30)))); + TPromise waiterAccepted = NewPromise(); + auto waiterAcceptedLambda = [=]() mutable { + waiterAccepted.SetValue(); + }; + auto waitAcquire = session2.AcquireSemaphore("SemA", + TAcquireSemaphoreSettings() + .Count(2) + .Timeout(TDuration::Max()) + .OnAccepted(waiterAcceptedLambda)); + waiterAccepted.GetFuture().GetValueSync(); + + auto describeSemA = ExpectSuccess( + session1.DescribeSemaphore("SemA", + TDescribeSemaphoreSettings() + .IncludeOwners() + .IncludeWaiters())); + auto describeSemB = ExpectSuccess( + session1.DescribeSemaphore("SemB", + TDescribeSemaphoreSettings() + .IncludeOwners() + .IncludeWaiters())); + + { + auto itFuture = context.Client.ListSemaphores( + "/Root/node1", + NYdb::NCoordination::TListSemaphoresSettings().IncludeDetails(false)); + auto it = itFuture.ExtractValueSync(); + UNIT_ASSERT_C(it.IsSuccess(), TStatusDescription(it)); + + std::unordered_set names; + while (true) { + auto part = it.ReadNext().ExtractValueSync(); + UNIT_ASSERT_C(part.IsSuccess(), TStatusDescription(part)); + if (!part.HasSemaphoreDescription()) { + break; + } + UNIT_ASSERT_VALUES_EQUAL(part.GetSemaphoreDescription().GetOwners().size(), 0u); + UNIT_ASSERT_VALUES_EQUAL(part.GetSemaphoreDescription().GetWaiters().size(), 0u); + UNIT_ASSERT_VALUES_EQUAL(part.GetSemaphoreDescription().GetData(), ""); + names.insert(part.GetSemaphoreDescription().GetName()); + } + UNIT_ASSERT(names.count("SemA")); + UNIT_ASSERT(names.count("SemB")); + } + + { + auto itFuture = context.Client.ListSemaphores( + "/Root/node1", + NYdb::NCoordination::TListSemaphoresSettings().IncludeDetails(true)); + auto it = itFuture.ExtractValueSync(); + UNIT_ASSERT_C(it.IsSuccess(), TStatusDescription(it)); + + std::unordered_map byName; + while (true) { + auto part = it.ReadNext().ExtractValueSync(); + UNIT_ASSERT_C(part.IsSuccess(), TStatusDescription(part)); + if (!part.HasSemaphoreDescription()) { + break; + } + const auto& d = part.GetSemaphoreDescription(); + byName[d.GetName()] = d; + } + UNIT_ASSERT_VALUES_EQUAL(byName.size(), 2u); + AssertSemaphoreDescriptionMatchesDescribe(byName.at("SemA"), describeSemA); + AssertSemaphoreDescriptionMatchesDescribe(byName.at("SemB"), describeSemB); + } + + UNIT_ASSERT(ExpectSuccess(session1.ReleaseSemaphore("SemA"))); + UNIT_ASSERT(ExpectSuccess(std::move(waitAcquire))); + UNIT_ASSERT(ExpectSuccess(session2.ReleaseSemaphore("SemA"))); + } Y_UNIT_TEST(SessionMethods) { TKikimrWithGrpcAndRootSchema server;