#include "service.h" #include "service_private.h" #include "service_ut_helpers.h" #include "service_ut_sharding.h" #include #include #include #include #include #include #include #include #include #include #include #include #include namespace NCloud::NFileStore::NStorage { using namespace NActors; using namespace NKikimr; namespace { //////////////////////////////////////////////////////////////////////////////// NProtoPrivate::TGetFileSystemTopologyResponse GetFileSystemTopology( TServiceClient& service, const TString& fsId) { NProtoPrivate::TGetFileSystemTopologyRequest request; request.SetFileSystemId(fsId); TString buf; google::protobuf::util::MessageToJsonString(request, &buf); const auto actionResponse = service.ExecuteAction("getfilesystemtopology", buf); NProtoPrivate::TGetFileSystemTopologyResponse response; auto status = google::protobuf::util::JsonStringToMessage( actionResponse->Record.GetOutput(), &response); return response; } void CheckShardsSize( const TShardedFileSystemConfig& fsConfig, TServiceClient& service, const ui64 blocksCount) { const auto mainStats = GetStorageStats(service, fsConfig.FsId); const auto shard1Stats = GetStorageStats( service, fsConfig.Shard1Id, 0 /* cacheTTL */, NProtoPrivate::STATS_REQUEST_MODE_FORCE_FETCH_SHARDS); const auto shard2Stats = GetStorageStats( service, fsConfig.Shard2Id, 0 /* cacheTTL */, NProtoPrivate::STATS_REQUEST_MODE_FORCE_FETCH_SHARDS); UNIT_ASSERT_EQUAL(blocksCount, mainStats.GetStats().GetTotalBlocksCount()); UNIT_ASSERT_EQUAL( blocksCount, shard1Stats.GetStats().GetTotalBlocksCount()); UNIT_ASSERT_EQUAL( blocksCount, shard2Stats.GetStats().GetTotalBlocksCount()); } NProtoPrivate::TUnsafeChangeTabletStateResponse SetCompressNodeRef( const TString& fsId, TServiceClient& service, const bool compressNodeRef) { NProtoPrivate::TUnsafeChangeTabletStateRequest request; request.SetFileSystemId(fsId); request.SetCompressNodeRef(compressNodeRef); TString buf; google::protobuf::util::MessageToJsonString(request, &buf); auto jsonResponse = service.SendAndRecvExecuteAction("UnsafeChangeTabletState", buf); UNIT_ASSERT_C( SUCCEEDED(jsonResponse->GetStatus()), jsonResponse->GetErrorReason()); NProtoPrivate::TUnsafeChangeTabletStateResponse response; UNIT_ASSERT( google::protobuf::util::JsonStringToMessage( jsonResponse->Record.GetOutput(), &response) .ok()); return response; } auto GetFileSystemCounters(TTestEnv& env, const TString& fsId) { const auto counters = env.GetRuntime().GetAppData().Counters; auto subgroup = counters->FindSubgroup("counters", "filestore"); UNIT_ASSERT(subgroup); subgroup = subgroup->FindSubgroup("component", "storage_fs"); UNIT_ASSERT(subgroup); subgroup = subgroup->FindSubgroup("host", "cluster"); UNIT_ASSERT(subgroup); subgroup = subgroup->FindSubgroup("filesystem", fsId); UNIT_ASSERT(subgroup); subgroup = subgroup->FindSubgroup("cloud", "test_cloud"); UNIT_ASSERT(subgroup); subgroup = subgroup->FindSubgroup("folder", "test_folder"); UNIT_ASSERT(subgroup); return subgroup; } void SetQuota( TServiceClient& service, const TString& fsId, ui32 quotaId, ui64 maxBytes, ui64 maxNodes) { NProtoPrivate::TSetQuotaRequest request; request.SetFileSystemId(fsId); request.SetQuotaId(quotaId); request.SetMaxBytes(maxBytes); request.SetMaxNodes(maxNodes); TString buf; google::protobuf::util::MessageToJsonString(request, &buf); service.ExecuteAction("setquota", buf); } } // namespace //////////////////////////////////////////////////////////////////////////////// #define SERVICE_TEST_DECL(name) \ void TestImpl##name(NProto::TStorageConfig config) \ // SERVICE_TEST_DECL #define SERVICE_TEST(name) \ SERVICE_TEST_DECL(name); \ Y_UNIT_TEST(name) \ { \ TestImpl##name(MakeStorageConfig()); \ } \ SERVICE_TEST_DECL(name) \ // SERVICE_TEST_SIMPLE #define SERVICE_TEST_DIR_CREATION_IN_SHARDS(name) \ SERVICE_TEST_DECL(name); \ Y_UNIT_TEST(name) \ { \ TestImpl##name(MakeStorageConfig()); \ } \ Y_UNIT_TEST(name##WithDirectoryCreationInShards) \ { \ TestImpl##name(MakeStorageConfigWithDirectoryCreationInShards()); \ } \ SERVICE_TEST_DECL(name) // SERVICE_TEST_DIR_CREATION_IN_SHARDS //////////////////////////////////////////////////////////////////////////////// Y_UNIT_TEST_SUITE(TStorageServiceShardingTest) { const auto StartupEventType = TEvIndexTabletPrivate::EvLoadCompactionMapChunkRequest; void CatchActorIds(TServiceClient& service, TVector& ids) { service.AccessRuntime().SetEventFilter( [&] (auto& runtime, TAutoPtr& event) { Y_UNUSED(runtime); switch (event->GetTypeRewrite()) { case StartupEventType: { if (Find(ids, event->Sender) == ids.end()) { ids.push_back(event->Sender); } break; } } return false; }); } #define CREATE_ENV_AND_SHARDED_FILESYSTEM() \ /* Enable automatic shard creation */ \ config.SetAutomaticShardCreationEnabled(true); \ config.SetAutomaticallyCreatedShardSize(fsConfig.ShardBlockCount * 4_KB); \ config.SetShardAllocationUnit(fsConfig.ShardBlockCount * 4_KB); \ \ TTestEnv env({}, config); \ \ ui32 nodeIdx = env.AddDynamicNode(); \ \ TServiceClient service(env.GetRuntime(), nodeIdx); \ auto fsInfo = CreateFileSystem(service, fsConfig); \ Y_UNUSED(fsInfo); \ // CREATE_ENV_AND_SHARDED_FILESYSTEM SERVICE_TEST(ShouldCreateSessionInShards) { TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); TVector responses; env.GetRuntime().SetEventFilter( [&] (auto& runtime, TAutoPtr& e) { Y_UNUSED(runtime); if (e->GetTypeRewrite() == TEvIndexTablet::EvCreateSessionResponse) { const auto* msg = e->Get(); responses.push_back(msg->Record); } return false; }); auto headers = service.InitSession(fsConfig.FsId, "client"); auto headers1 = headers; headers1.FileSystemId = fsConfig.Shard1Id; auto headers2 = headers; headers2.FileSystemId = fsConfig.Shard2Id; UNIT_ASSERT_C(3 <= responses.size(), responses.size()); for (ui32 i = 0; i < responses.size() - 1; ++i) { UNIT_ASSERT_C( !responses[i].HasFileStore(), responses[i].ShortUtf8DebugString()); } UNIT_ASSERT_VALUES_EQUAL_C( fsConfig.FsId, responses.back().GetFileStore().GetFileSystemId(), responses.back().ShortUtf8DebugString()); ui64 nodeId1 = service .CreateNode(headers1, TCreateNodeArgs::File(RootNodeId, "file")) ->Record.GetNode() .GetId(); UNIT_ASSERT_VALUES_EQUAL((1LU << 56U) + 2, nodeId1); ui64 handle1 = service .CreateHandle( headers1, headers1.FileSystemId, nodeId1, "", TCreateHandleArgs::RDWR) ->Record.GetHandle(); UNIT_ASSERT_C( handle1 >= (1LU << 56U) && handle1 < (2LU << 56U), handle1); service.WriteData( headers1, headers1.FileSystemId, nodeId1, handle1, 0, TString(1_MB, 'a')); ui64 nodeId2 = service .CreateNode(headers2, TCreateNodeArgs::File(RootNodeId, "file")) ->Record.GetNode() .GetId(); UNIT_ASSERT_VALUES_EQUAL((2LU << 56U) + 2, nodeId2); ui64 handle2 = service .CreateHandle( headers2, headers2.FileSystemId, nodeId2, "", TCreateHandleArgs::RDWR) ->Record.GetHandle(); UNIT_ASSERT_C( handle2 >= (2LU << 56U) && handle2 < (3LU << 56U), handle2); service.WriteData( headers2, headers2.FileSystemId, nodeId2, handle2, 0, TString(1_MB, 'a')); for (const auto& shardId: fsConfig.ShardIds()) { NProtoPrivate::TDescribeSessionsRequest request; request.SetFileSystemId(shardId); TString buf; google::protobuf::util::MessageToJsonString(request, &buf); auto jsonResponse = service.ExecuteAction("describesessions", buf); NProtoPrivate::TDescribeSessionsResponse response; UNIT_ASSERT(google::protobuf::util::JsonStringToMessage( jsonResponse->Record.GetOutput(), &response).ok()); const auto& sessions = response.GetSessions(); UNIT_ASSERT_VALUES_EQUAL(1, sessions.size()); UNIT_ASSERT_VALUES_EQUAL( headers.SessionId, sessions[0].GetSessionId()); UNIT_ASSERT_VALUES_EQUAL( headers.ClientId, sessions[0].GetClientId()); UNIT_ASSERT_VALUES_EQUAL("", sessions[0].GetSessionState()); } const TString sessionState = "some_state"; service.ResetSession(headers, sessionState); for (const auto& shardId: fsConfig.ShardIds()) { NProtoPrivate::TDescribeSessionsRequest request; request.SetFileSystemId(shardId); TString buf; google::protobuf::util::MessageToJsonString(request, &buf); auto jsonResponse = service.ExecuteAction("describesessions", buf); NProtoPrivate::TDescribeSessionsResponse response; UNIT_ASSERT(google::protobuf::util::JsonStringToMessage( jsonResponse->Record.GetOutput(), &response).ok()); const auto& sessions = response.GetSessions(); UNIT_ASSERT_VALUES_EQUAL(1, sessions.size()); UNIT_ASSERT_VALUES_EQUAL( headers.SessionId, sessions[0].GetSessionId()); UNIT_ASSERT_VALUES_EQUAL( headers.ClientId, sessions[0].GetClientId()); UNIT_ASSERT_VALUES_EQUAL( sessionState, sessions[0].GetSessionState()); } service.DestroySession(headers); for (const auto& shardId: fsConfig.ShardIds()) { NProtoPrivate::TDescribeSessionsRequest request; request.SetFileSystemId(shardId); TString buf; google::protobuf::util::MessageToJsonString(request, &buf); auto jsonResponse = service.ExecuteAction("describesessions", buf); NProtoPrivate::TDescribeSessionsResponse response; UNIT_ASSERT(google::protobuf::util::JsonStringToMessage( jsonResponse->Record.GetOutput(), &response).ok()); const auto& sessions = response.GetSessions(); UNIT_ASSERT_VALUES_EQUAL(0, sessions.size()); } } SERVICE_TEST( ShouldRemoveStaleSessionFromShardDuringSyncAfterDestroyPushFails) { constexpr TDuration IdleSessionTimeout = TDuration::Seconds(5); config.SetIdleSessionTimeout(IdleSessionTimeout.MilliSeconds()); TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); // When the main tablet destroys the session, it also sends a // DestroySessionRequest straight to each shard. Drop the first one // going to Shard1Id, so only the sync process can remove it there - // this is what would happen if the main tablet died before that // direct request reached the shard. Only the first one is dropped, // so a later retry sent by sync still gets through. bool droppedDestroyPushToShard1 = false; bool sawSyncTick = false; // Set when a second (non-dropped) DestroySessionRequest reaches // Shard1Id - this is the sync actually deleting the stale // session there. bool sawStaleSessionDeleteRequest = false; env.GetRuntime().SetEventFilter( [&](auto& runtime, TAutoPtr& event) { Y_UNUSED(runtime); switch (event->GetTypeRewrite()) { case TEvIndexTablet::EvDestroySessionRequest: { const auto* msg = event->Get< TEvIndexTablet::TEvDestroySessionRequest>(); if (msg->Record.GetFileSystemId() == fsConfig.Shard1Id) { if (!droppedDestroyPushToShard1) { droppedDestroyPushToShard1 = true; return true; } sawStaleSessionDeleteRequest = true; } break; } case TEvIndexTablet::EvDescribeSessionsRequest: { const auto* msg = event->Get< TEvIndexTablet::TEvDescribeSessionsRequest>(); if (msg->Record.GetFileSystemId() == fsConfig.Shard1Id) { sawSyncTick = true; } break; } } return false; }); auto describeShardSessions = [&](const TString& fsId) { NProtoPrivate::TDescribeSessionsRequest request; request.SetFileSystemId(fsId); TString buf; google::protobuf::util::MessageToJsonString(request, &buf); auto jsonResponse = service.ExecuteAction("describesessions", buf); NProtoPrivate::TDescribeSessionsResponse response; UNIT_ASSERT( google::protobuf::util::JsonStringToMessage( jsonResponse->Record.GetOutput(), &response) .ok()); return response; }; TIndexTabletClient tablet( env.GetRuntime(), nodeIdx, fsInfo.MainTabletId); tablet.InitSession("client", "session"); // Check: the session already exists in the shard, and it is not // orphan yet. { auto response = describeShardSessions(fsConfig.Shard1Id); UNIT_ASSERT_VALUES_EQUAL(1, response.GetSessions().size()); UNIT_ASSERT(!response.GetSessions(0).GetIsOrphan()); } // Rebooting reloads sessions from the DB through LoadSessions, and // LoadSessions always marks a session as orphan with a // IdleSessionTimeout deadline tablet.RebootTablet(); // Force a sync tick while the session is still inside its grace // period on main and confirm it actually ran. The session must // still exist in the shard. sawSyncTick = false; env.GetRuntime().AdvanceCurrentTime( IdleSessionTimeout / 3 + TDuration::MilliSeconds(500)); tablet.SendRequest( std::make_unique()); { TDispatchOptions options; options.CustomFinalCondition = [&] { return sawSyncTick; }; env.GetRuntime().DispatchEvents(options); } UNIT_ASSERT_C(sawSyncTick, "sync did not run"); { auto response = describeShardSessions(fsConfig.Shard1Id); UNIT_ASSERT_VALUES_EQUAL_C( 1, response.GetSessions().size(), "session removed from shard too early"); UNIT_ASSERT(!response.GetSessions(0).GetIsOrphan()); } // Advance past the orphan grace period, force the cleanup tick // (which destroys the now-expired session on main) immediately // followed by the next sync tick, and stop as soon as sync // actually deletes the stale session from the shard. env.GetRuntime().AdvanceCurrentTime(IdleSessionTimeout); tablet.SendRequest( std::make_unique< TEvIndexTabletPrivate::TEvCleanupSessionsRequest>()); tablet.SendRequest( std::make_unique()); { TDispatchOptions options; options.CustomFinalCondition = [&] { return sawStaleSessionDeleteRequest; }; env.GetRuntime().DispatchEvents(options); } UNIT_ASSERT_C( sawStaleSessionDeleteRequest, "sync delete session did not run"); { auto response = describeShardSessions(fsConfig.Shard1Id); UNIT_ASSERT_VALUES_EQUAL_C( 0, response.GetSessions().size(), "stale session was not removed from the shard during sync"); } } SERVICE_TEST(ShouldPropagateQuotasToShards) { TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); auto listQuotas = [&](const TString& fileSystemId) { NProtoPrivate::TListQuotasRequest request; request.SetFileSystemId(fileSystemId); TString buf; google::protobuf::util::MessageToJsonString(request, &buf); auto jsonResponse = service.ExecuteAction("listquotas", buf); NProtoPrivate::TListQuotasResponse response; UNIT_ASSERT(google::protobuf::util::JsonStringToMessage( jsonResponse->Record.GetOutput(), &response).ok()); return response.GetQuotas(); }; { NProtoPrivate::TSetQuotaRequest request; request.SetFileSystemId(fsConfig.FsId); request.SetQuotaId(42); request.SetMaxBytes(1_GB); request.SetMaxNodes(100); TString buf; google::protobuf::util::MessageToJsonString(request, &buf); service.ExecuteAction("setquota", buf); } for (const auto& id: fsConfig.MainAndShardIds()) { const auto quotas = listQuotas(id); UNIT_ASSERT_VALUES_EQUAL_C(1, quotas.size(), id); UNIT_ASSERT_VALUES_EQUAL_C(42u, quotas[0].GetQuotaId(), id); UNIT_ASSERT_VALUES_EQUAL_C(1_GB, quotas[0].GetMaxBytes(), id); UNIT_ASSERT_VALUES_EQUAL_C(100u, quotas[0].GetMaxNodes(), id); } { NProtoPrivate::TDeleteQuotaRequest request; request.SetFileSystemId(fsConfig.FsId); request.SetQuotaId(42); TString buf; google::protobuf::util::MessageToJsonString(request, &buf); service.ExecuteAction("deletequota", buf); } for (const auto& id: fsConfig.MainAndShardIds()) { UNIT_ASSERT_VALUES_EQUAL_C(0, listQuotas(id).size(), id); } } SERVICE_TEST(ShouldAggregateQuotaUsageAcrossShards) { TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); { NProtoPrivate::TSetQuotaRequest request; request.SetFileSystemId(fsConfig.FsId); request.SetQuotaId(42); request.SetMaxBytes(1_GB); request.SetMaxNodes(100); TString buf; google::protobuf::util::MessageToJsonString(request, &buf); service.ExecuteAction("setquota", buf); } auto headers = service.InitSession(fsConfig.FsId, "client"); // attach the quota to a directory on main - files are regular nodes, // so main auto-routes their creation into shards (round-robin) and // forwards the inherited QuotaId along with each one; the directory // itself stays on main and its own attach bumps main's own local // usage by 1 auto dirId = service .CreateNode( headers, TCreateNodeArgs::Directory(RootNodeId, "dir")) ->Record.GetNode() .GetId(); service.SetNodeAttr( headers, fsConfig.FsId, TSetNodeAttrArgs(dirId).SetQuotaId(42)); constexpr ui32 fileCount = 4; constexpr ui64 fileSize = 100; for (ui32 i = 0; i < fileCount; ++i) { auto fileId = service .CreateNode( headers, TCreateNodeArgs::File(dirId, "file" + ToString(i))) ->Record.GetNode() .GetId(); service.SetNodeAttr( headers, fsConfig.FsId, TSetNodeAttrArgs(fileId).SetSize(fileSize)); } // a second, distinct quota that only ever lands on one shard (a // single file) { NProtoPrivate::TSetQuotaRequest request; request.SetFileSystemId(fsConfig.FsId); request.SetQuotaId(43); request.SetMaxBytes(1_GB); request.SetMaxNodes(100); TString buf; google::protobuf::util::MessageToJsonString(request, &buf); service.ExecuteAction("setquota", buf); } constexpr ui64 otherQuotaFileSize = 50; auto otherQuotaDirId = service .CreateNode( headers, TCreateNodeArgs::Directory(RootNodeId, "other_quota_dir")) ->Record.GetNode() .GetId(); service.SetNodeAttr( headers, fsConfig.FsId, TSetNodeAttrArgs(otherQuotaDirId).SetQuotaId(43)); auto otherQuotaFileId = service .CreateNode( headers, TCreateNodeArgs::File(otherQuotaDirId, "file")) ->Record.GetNode() .GetId(); service.SetNodeAttr( headers, fsConfig.FsId, TSetNodeAttrArgs(otherQuotaFileId).SetSize(otherQuotaFileSize)); for (const auto& shardId: fsConfig.ShardIds()) { const auto stats = GetStorageStats(service, shardId); const auto& usages = stats.GetStats().GetQuotaUsages(); UNIT_ASSERT_C(usages.contains(42), shardId); const auto& usage = usages.at(42); UNIT_ASSERT_C(usage.GetUsedNodes() > 0, shardId); UNIT_ASSERT_C(usage.GetUsedBytes() > 0, shardId); } // forcing main to fan out and aggregate should sum main's own local // usage (the directory attach points) and both shards' contributions const auto aggregate = GetStorageStats( service, fsConfig.FsId, 0 /* cacheTTL */, NProtoPrivate::STATS_REQUEST_MODE_FORCE_FETCH_SHARDS); const auto& usages = aggregate.GetStats().GetQuotaUsages(); UNIT_ASSERT_VALUES_EQUAL(2, usages.size()); UNIT_ASSERT(usages.contains(42)); const auto& usage42 = usages.at(42); UNIT_ASSERT_VALUES_EQUAL(1 + fileCount, usage42.GetUsedNodes()); UNIT_ASSERT_VALUES_EQUAL(fileCount * fileSize, usage42.GetUsedBytes()); UNIT_ASSERT(usages.contains(43)); const auto& usage43 = usages.at(43); UNIT_ASSERT_VALUES_EQUAL(2, usage43.GetUsedNodes()); UNIT_ASSERT_VALUES_EQUAL(otherQuotaFileSize, usage43.GetUsedBytes()); } SERVICE_TEST(ShouldAggregateQuotaUsageInBackground) { TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); { NProtoPrivate::TSetQuotaRequest request; request.SetFileSystemId(fsConfig.FsId); request.SetQuotaId(42); request.SetMaxBytes(1_GB); request.SetMaxNodes(100); TString buf; google::protobuf::util::MessageToJsonString(request, &buf); service.ExecuteAction("setquota", buf); } auto headers = service.InitSession(fsConfig.FsId, "client"); auto dirId = service .CreateNode( headers, TCreateNodeArgs::Directory(RootNodeId, "dir")) ->Record.GetNode() .GetId(); service.SetNodeAttr( headers, fsConfig.FsId, TSetNodeAttrArgs(dirId).SetQuotaId(42)); auto fileId = service.CreateNode(headers, TCreateNodeArgs::File(dirId, "file")) ->Record.GetNode() .GetId(); ui64 handle = service .CreateHandle( headers, fsConfig.FsId, fileId, "", TCreateHandleArgs::RDWR) ->Record.GetHandle(); auto data = GenerateValidateData(256_KB); service.WriteData(headers, fsConfig.FsId, fileId, handle, 0, data); // triggering background shard stats collection env.GetRuntime().AdvanceCurrentTime(TDuration::Seconds(15)); { using TRequest = TEvIndexTabletPrivate::TEvUpdateCounters; env.GetRuntime().Send( new IEventHandle( fsInfo.MainTabletActorId, // recipient TActorId(), // sender new TRequest(), 0, // flags 0), nodeIdx); } TDispatchOptions options; options.FinalEvents = {TDispatchOptions::TFinalEventCondition( TEvIndexTabletPrivate::EvAggregateStatsCompleted)}; service.AccessRuntime().DispatchEvents(options); const auto response = GetStorageStats(service, fsConfig.FsId, Max() /* cacheTTL */); const auto& usages = response.GetStats().GetQuotaUsages(); UNIT_ASSERT_VALUES_EQUAL(1, usages.size()); UNIT_ASSERT(usages.contains(42)); const auto& usage = usages.at(42); UNIT_ASSERT_VALUES_EQUAL(2, usage.GetUsedNodes()); UNIT_ASSERT_VALUES_EQUAL(data.size(), usage.GetUsedBytes()); } SERVICE_TEST(ShouldCheckForShardsInAdapterModeUponSessionCreationInShards) { // // Enabling some features which shouldn't work for adapter-mode shards. // TShardedFileSystemConfig fsConfig; config.SetTwoStageReadEnabled(true); config.SetThreeStageWriteEnabled(true); CREATE_ENV_AND_SHARDED_FILESYSTEM(); // // Configuring one of the shards to use the adapter mode. // { NProtoPrivate::TConfigureAsShardRequest request; request.SetFileSystemId(fsConfig.Shard1Id); request.SetIsFastShard(true); request.MutableFastShardConfig()->MutableMemConfig(); request.SetShardNo(1); TString buf; google::protobuf::util::MessageToJsonString(request, &buf); auto jsonResponse = service.ExecuteAction("configureasshard", buf); NProtoPrivate::TConfigureAsShardResponse response; UNIT_ASSERT(google::protobuf::util::JsonStringToMessage( jsonResponse->Record.GetOutput(), &response).ok()); UNIT_ASSERT_VALUES_EQUAL_C( S_OK, response.GetError().GetCode(), response.GetError().GetMessage()); } THeaders headers = { .FileSystemId = fsConfig.FsId, .ClientId = "client", .SessionId = "", .SessionSeqNo = 0}; auto response = service.CreateSession( headers, "" /* checkpointId */, false /* restoreClientSession */, 0 /* sessionSeqNo */, false /* readOnly */); // // Checking that the unsupported features are disabled. // const auto& f = response->Record.GetFileStore().GetFeatures(); UNIT_ASSERT(!f.GetTwoStageReadEnabled()); UNIT_ASSERT(!f.GetThreeStageWriteEnabled()); } SERVICE_TEST(ShouldRestoreSessionInShardAfterShardRestart) { const auto idleSessionTimeout = TDuration::Minutes(2); config.SetIdleSessionTimeout(idleSessionTimeout.MilliSeconds()); TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); auto headers = service.InitSession(fsConfig.FsId, "client"); auto headers1 = headers; headers1.FileSystemId = fsConfig.Shard1Id; auto headers2 = headers; headers2.FileSystemId = fsConfig.Shard2Id; // creating nodes and handles in both shards ui64 nodeId1 = service .CreateNode(headers1, TCreateNodeArgs::File(RootNodeId, "file")) ->Record.GetNode() .GetId(); UNIT_ASSERT_VALUES_EQUAL((1LU << 56U) + 2, nodeId1); ui64 handle1 = service .CreateHandle( headers1, headers1.FileSystemId, nodeId1, "", TCreateHandleArgs::RDWR) ->Record.GetHandle(); UNIT_ASSERT_C( handle1 >= (1LU << 56U) && handle1 < (2LU << 56U), handle1); ui64 nodeId2 = service .CreateNode(headers2, TCreateNodeArgs::File(RootNodeId, "file")) ->Record.GetNode() .GetId(); UNIT_ASSERT_VALUES_EQUAL((2LU << 56U) + 2, nodeId2); ui64 handle2 = service .CreateHandle( headers2, headers2.FileSystemId, nodeId2, "", TCreateHandleArgs::RDWR) ->Record.GetHandle(); UNIT_ASSERT_C( handle2 >= (2LU << 56U) && handle2 < (3LU << 56U), handle2); // rebooting shards TVector fsActorIds; CatchActorIds(service, fsActorIds); TIndexTabletClient shard1( env.GetRuntime(), nodeIdx, fsInfo.Shard1TabletId, {}, // config false // updateConfig ); shard1.RebootTablet(); TIndexTabletClient shard2( env.GetRuntime(), nodeIdx, fsInfo.Shard2TabletId, {}, // config false // updateConfig ); shard2.RebootTablet(); // triggering shard sessions sync // sending the event manually since registration observers which enable // scheduling for actors are reset upon tablet reboot env.GetRuntime().AdvanceCurrentTime(idleSessionTimeout / 2); { using TRequest = TEvIndexTabletPrivate::TEvSyncSessionsRequest; env.GetRuntime().Send( new IEventHandle( fsInfo.MainTabletActorId, // recipient TActorId(), // sender new TRequest(), 0, // flags 0), nodeIdx); } env.GetRuntime().DispatchEvents({}, TDuration::MilliSeconds(100)); // waiting for idle session expiration // sending the event manually since registration observers which enable // scheduling for actors are reset upon tablet reboot env.GetRuntime().AdvanceCurrentTime(idleSessionTimeout / 2); service.PingSession(headers); for (const auto& actorId: fsActorIds) { using TRequest = TEvIndexTabletPrivate::TEvCleanupSessionsRequest; env.GetRuntime().Send( new IEventHandle( actorId, // recipient TActorId(), // sender new TRequest(), 0, // flags 0), nodeIdx); } // need to pass deadline instead of timeout here since otherwise the // adjusted time gets added to the timeout env.GetRuntime().DispatchEvents( {}, TInstant::Now() + TDuration::MilliSeconds(100)); // shard sessions should exist for (const auto& id: fsConfig.MainAndShardIds()) { NProtoPrivate::TDescribeSessionsRequest request; request.SetFileSystemId(id); TString buf; google::protobuf::util::MessageToJsonString(request, &buf); auto jsonResponse = service.ExecuteAction("describesessions", buf); NProtoPrivate::TDescribeSessionsResponse response; UNIT_ASSERT(google::protobuf::util::JsonStringToMessage( jsonResponse->Record.GetOutput(), &response).ok()); const auto& sessions = response.GetSessions(); UNIT_ASSERT_VALUES_EQUAL(1, sessions.size()); UNIT_ASSERT_VALUES_EQUAL( headers.SessionId, sessions[0].GetSessionId()); UNIT_ASSERT_VALUES_EQUAL( headers.ClientId, sessions[0].GetClientId()); } // handles should be alive service.WriteData( headers1, headers1.FileSystemId, nodeId1, handle1, 0, TString(1_MB, 'a')); service.WriteData( headers2, headers2.FileSystemId, nodeId2, handle2, 0, TString(1_MB, 'a')); } SERVICE_TEST(ShouldCreateNodeInShardViaLeader) { TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); auto headers = service.InitSession(fsConfig.FsId, "client"); headers.DisableMultiTabletForwarding = true; ui64 nodeId1 = service .CreateNode(headers, TCreateNodeArgs::File( RootNodeId, "file1", 0, // mode fsConfig.Shard1Id)) ->Record.GetNode() .GetId(); UNIT_ASSERT_VALUES_EQUAL((1LU << 56U) + 2, nodeId1); auto createHandleResponse = service.CreateHandle( headers, fsConfig.FsId, RootNodeId, "file1", TCreateHandleArgs::RDWR)->Record; UNIT_ASSERT_VALUES_EQUAL( fsConfig.Shard1Id, createHandleResponse.GetShardFileSystemId()); UNIT_ASSERT_VALUES_UNEQUAL( "", createHandleResponse.GetShardNodeName()); auto headers1 = headers; headers1.FileSystemId = fsConfig.Shard1Id; createHandleResponse = service.CreateHandle( headers1, headers1.FileSystemId, RootNodeId, createHandleResponse.GetShardNodeName(), TCreateHandleArgs::RDWR)->Record; auto handle1 = createHandleResponse.GetHandle(); UNIT_ASSERT_C( handle1 >= (1LU << 56U) && handle1 < (2LU << 56U), handle1); UNIT_ASSERT_VALUES_EQUAL( nodeId1, createHandleResponse.GetNodeAttr().GetId()); service.WriteData( headers1, headers1.FileSystemId, nodeId1, handle1, 0, TString(1_MB, 'a')); } void DoShouldCheckAttrForNodeCreatedInShardViaLeader( NProto::TStorageConfig & config, std::function newArgsConstructor, bool shouldTriggerCriticalEvent) { TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); auto headers = service.InitSession(fsConfig.FsId, "client"); TAutoPtr createNodeInShard; TString name; bool intercept = true; env.GetRuntime().SetEventFilter( [&] (auto& runtime, TAutoPtr& e) { Y_UNUSED(runtime); if (e->GetTypeRewrite() == TEvService::EvCreateNodeRequest) { const auto* msg = e->Get(); if (intercept && msg->Record.GetFileSystemId() == fsConfig.Shard1Id) { name = msg->Record.GetName(); createNodeInShard = e; return true; } } return false; }); service.SendCreateNodeRequest( headers, TCreateNodeArgs::File( RootNodeId, "file1", 0, // mode fsConfig.Shard1Id)); ui32 iterations = 0; while (!createNodeInShard && iterations++ < 100) { env.GetRuntime().DispatchEvents({}, TDuration::MilliSeconds(50)); } UNIT_ASSERT(createNodeInShard); intercept = false; auto headers1 = headers; headers1.FileSystemId = fsConfig.Shard1Id; headers1.DisableMultiTabletForwarding = true; const auto nodeId = service.CreateNode( headers1, TCreateNodeArgs::File(RootNodeId, name))->Record.GetNode().GetId(); UNIT_ASSERT_VALUES_EQUAL(1, ExtractShardNo(nodeId)); TSetNodeAttrArgs newArgs = newArgsConstructor(); newArgs.Node = nodeId; service.SetNodeAttr(headers1, fsConfig.Shard1Id, newArgs); const auto counters = env.GetCounters()->FindSubgroup("component", "service"); UNIT_ASSERT(counters); const auto counter = counters->GetCounter( "AppCriticalEvents/CreateNodeRequestResponseMismatchInShard"); UNIT_ASSERT_VALUES_EQUAL(0, counter->GetAtomic()); env.GetRuntime().Send(createNodeInShard.Release(), nodeIdx); auto response = service.RecvCreateNodeResponse(); UNIT_ASSERT_VALUES_EQUAL_C( S_OK, response->GetStatus(), response->GetErrorReason()); UNIT_ASSERT_VALUES_EQUAL(nodeId, response->Record.GetNode().GetId()); UNIT_ASSERT_VALUES_EQUAL( shouldTriggerCriticalEvent, counter->GetAtomic()); } SERVICE_TEST(ShouldCheckAttrForNodeCreatedInShardViaLeader) { // If the size is changed with mtime update, this is an acceptable // situation and should not trigger the critical event DoShouldCheckAttrForNodeCreatedInShardViaLeader( config, []() { return TSetNodeAttrArgs(InvalidNodeId) .SetSize(1_MB) .SetMTime( (TInstant::Now() + TDuration::Hours(1)).MicroSeconds()); }, false); // Even if the MTime is within a reasonable range, it should not // trigger the critical event DoShouldCheckAttrForNodeCreatedInShardViaLeader( config, []() { return TSetNodeAttrArgs(InvalidNodeId) .SetSize(1_MB) .SetMTime((TInstant::Now() - TDuration::Seconds(10)) .MicroSeconds()); }, false); // If the MTime is too long ago, it should trigger the critical // event, since it is likely that the creation has overwritten // the node created earlier in the shard DoShouldCheckAttrForNodeCreatedInShardViaLeader( config, []() { return TSetNodeAttrArgs(InvalidNodeId) .SetSize(1_MB) .SetMTime((TInstant::Now() - TDuration::Minutes(10)) .MicroSeconds()); }, true); } SERVICE_TEST(ShouldCreateNodeInShardByCreateHandleViaLeader) { TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); auto headers = service.InitSession(fsConfig.FsId, "client"); headers.DisableMultiTabletForwarding = true; auto createHandleResponse = service.CreateHandle( headers, fsConfig.FsId, RootNodeId, "file1", TCreateHandleArgs::CREATE, fsConfig.Shard1Id)->Record; UNIT_ASSERT_VALUES_EQUAL( fsConfig.Shard1Id, createHandleResponse.GetShardFileSystemId()); const auto shardNodeName = createHandleResponse.GetShardNodeName(); UNIT_ASSERT_VALUES_UNEQUAL("", shardNodeName); const auto nodeId1 = createHandleResponse.GetNodeAttr().GetId(); UNIT_ASSERT_VALUES_EQUAL((1LU << 56U) + 2, nodeId1); auto headers1 = headers; headers1.FileSystemId = fsConfig.Shard1Id; createHandleResponse = service.CreateHandle( headers1, headers1.FileSystemId, RootNodeId, shardNodeName, TCreateHandleArgs::RDWR)->Record; auto handle1 = createHandleResponse.GetHandle(); UNIT_ASSERT_C( handle1 >= (1LU << 56U) && handle1 < (2LU << 56U), handle1); UNIT_ASSERT_VALUES_EQUAL( nodeId1, createHandleResponse.GetNodeAttr().GetId()); service.WriteData( headers1, headers1.FileSystemId, nodeId1, handle1, 0, TString(1_MB, 'a')); auto getAttrResponse = service.GetNodeAttr( headers, fsConfig.FsId, RootNodeId, "file1")->Record; UNIT_ASSERT_VALUES_EQUAL( fsConfig.Shard1Id, getAttrResponse.GetNode().GetShardFileSystemId()); UNIT_ASSERT_VALUES_EQUAL( shardNodeName, getAttrResponse.GetNode().GetShardNodeName()); getAttrResponse = service.GetNodeAttr( headers1, fsConfig.Shard1Id, RootNodeId, shardNodeName)->Record; UNIT_ASSERT_VALUES_EQUAL(nodeId1, getAttrResponse.GetNode().GetId()); UNIT_ASSERT_VALUES_EQUAL(1_MB, getAttrResponse.GetNode().GetSize()); auto listNodesResponse = service.ListNodes( headers, fsConfig.FsId, RootNodeId)->Record; UNIT_ASSERT_VALUES_EQUAL(1, listNodesResponse.NamesSize()); UNIT_ASSERT_VALUES_EQUAL("file1", listNodesResponse.GetNames(0)); UNIT_ASSERT_VALUES_EQUAL(1, listNodesResponse.NodesSize()); UNIT_ASSERT_VALUES_EQUAL( fsConfig.Shard1Id, listNodesResponse.GetNodes(0).GetShardFileSystemId()); UNIT_ASSERT_VALUES_EQUAL( shardNodeName, listNodesResponse.GetNodes(0).GetShardNodeName()); } SERVICE_TEST(ShouldReturnShardInfoFromUnsafeListNodes) { TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); auto headers = service.InitSession(fsConfig.FsId, "client"); service.CreateNode( headers, TCreateNodeArgs::File(RootNodeId, "file1")); auto noForwardHeaders = headers; noForwardHeaders.DisableMultiTabletForwarding = true; auto getAttrResponse = service.GetNodeAttr( noForwardHeaders, fsConfig.FsId, RootNodeId, "file1")->Record; const auto expectedShardId = getAttrResponse.GetNode().GetShardFileSystemId(); const auto expectedShardNodeName = getAttrResponse.GetNode().GetShardNodeName(); UNIT_ASSERT_VALUES_UNEQUAL("", expectedShardId); UNIT_ASSERT_VALUES_UNEQUAL("", expectedShardNodeName); UNIT_ASSERT(expectedShardId == fsConfig.Shard1Id); auto listNodesResponse = service.ListNodes( headers, fsConfig.FsId, RootNodeId, true)->Record; UNIT_ASSERT_VALUES_EQUAL(1, listNodesResponse.NamesSize()); UNIT_ASSERT_VALUES_EQUAL("file1", listNodesResponse.GetNames(0)); UNIT_ASSERT_VALUES_EQUAL(1, listNodesResponse.NodesSize()); UNIT_ASSERT_VALUES_EQUAL( expectedShardId, listNodesResponse.GetNodes(0).GetShardFileSystemId()); UNIT_ASSERT_VALUES_EQUAL( expectedShardNodeName, listNodesResponse.GetNodes(0).GetShardNodeName()); } SERVICE_TEST(ShouldForwardRequestsToShard) { config.SetLazyXAttrsEnabled(false); TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); auto headers = service.InitSession(fsConfig.FsId, "client"); auto createHandleResponse = service.CreateHandle( headers, fsConfig.FsId, RootNodeId, "file1", TCreateHandleArgs::CREATE_EXL)->Record; UNIT_ASSERT_VALUES_EQUAL( "", createHandleResponse.GetShardFileSystemId()); UNIT_ASSERT_VALUES_EQUAL( "", createHandleResponse.GetShardNodeName()); const auto nodeId1 = createHandleResponse.GetNodeAttr().GetId(); UNIT_ASSERT_VALUES_EQUAL((1LU << 56U) + 2, nodeId1); const auto handle1 = createHandleResponse.GetHandle(); UNIT_ASSERT_C( handle1 >= (1LU << 56U) && handle1 < (2LU << 56U), handle1); auto accessNodeResponse = service.AccessNode( headers, fsConfig.FsId, nodeId1)->Record; Y_UNUSED(accessNodeResponse); auto setNodeAttrResponse = service.SetNodeAttr( headers, fsConfig.FsId, nodeId1, 1_MB)->Record; UNIT_ASSERT_VALUES_EQUAL( nodeId1, setNodeAttrResponse.GetNode().GetId()); UNIT_ASSERT_VALUES_EQUAL(1_MB, setNodeAttrResponse.GetNode().GetSize()); auto allocateDataResponse = service.AllocateData( headers, fsConfig.FsId, nodeId1, handle1, 0, 2_MB)->Record; Y_UNUSED(allocateDataResponse); auto data = GenerateValidateData(256_KB); service.WriteData(headers, fsConfig.FsId, nodeId1, handle1, 0, data); auto readDataResponse = service.ReadData( headers, fsConfig.FsId, nodeId1, handle1, 0, data.size())->Record; UNIT_ASSERT_VALUES_EQUAL(data, readDataResponse.GetBuffer()); auto destroyHandleResponse = service.DestroyHandle( headers, fsConfig.FsId, nodeId1, handle1)->Record; Y_UNUSED(destroyHandleResponse); const auto createNodeResponse = service.CreateNode( headers, TCreateNodeArgs::File(RootNodeId, "file2"))->Record; const auto nodeId2 = createNodeResponse.GetNode().GetId(); UNIT_ASSERT_VALUES_EQUAL((2LU << 56U) + 2, nodeId2); auto getNodeAttrResponse = service.GetNodeAttr( headers, fsConfig.FsId, RootNodeId, "file1")->Record; UNIT_ASSERT_VALUES_EQUAL( nodeId1, getNodeAttrResponse.GetNode().GetId()); UNIT_ASSERT_VALUES_EQUAL( 2_MB, getNodeAttrResponse.GetNode().GetSize()); getNodeAttrResponse = service.GetNodeAttr( headers, fsConfig.FsId, nodeId1, "")->Record; UNIT_ASSERT_VALUES_EQUAL( nodeId1, getNodeAttrResponse.GetNode().GetId()); UNIT_ASSERT_VALUES_EQUAL( 2_MB, getNodeAttrResponse.GetNode().GetSize()); auto listNodesResponse = service.ListNodes( headers, fsConfig.FsId, RootNodeId)->Record; UNIT_ASSERT_VALUES_EQUAL(2, listNodesResponse.NamesSize()); UNIT_ASSERT_VALUES_EQUAL("file1", listNodesResponse.GetNames(0)); UNIT_ASSERT_VALUES_EQUAL("file2", listNodesResponse.GetNames(1)); UNIT_ASSERT_VALUES_EQUAL(2, listNodesResponse.NodesSize()); UNIT_ASSERT_VALUES_EQUAL( nodeId1, listNodesResponse.GetNodes(0).GetId()); UNIT_ASSERT_VALUES_EQUAL( 2_MB, listNodesResponse.GetNodes(0).GetSize()); UNIT_ASSERT_VALUES_EQUAL( nodeId2, listNodesResponse.GetNodes(1).GetId()); UNIT_ASSERT_VALUES_EQUAL( 0, listNodesResponse.GetNodes(1).GetSize()); service.UnlinkNode(headers, RootNodeId, "file1"); listNodesResponse = service.ListNodes( headers, fsConfig.FsId, RootNodeId)->Record; UNIT_ASSERT_VALUES_EQUAL(1, listNodesResponse.NamesSize()); UNIT_ASSERT_VALUES_EQUAL("file2", listNodesResponse.GetNames(0)); UNIT_ASSERT_VALUES_EQUAL(1, listNodesResponse.NodesSize()); UNIT_ASSERT_VALUES_EQUAL( nodeId2, listNodesResponse.GetNodes(0).GetId()); UNIT_ASSERT_VALUES_EQUAL( 0, listNodesResponse.GetNodes(0).GetSize()); auto headers1 = headers; headers1.FileSystemId = fsConfig.Shard1Id; listNodesResponse = service.ListNodes( headers1, fsConfig.Shard1Id, RootNodeId)->Record; UNIT_ASSERT_VALUES_EQUAL(0, listNodesResponse.NamesSize()); UNIT_ASSERT_VALUES_EQUAL(0, listNodesResponse.NodesSize()); auto headers2 = headers; headers2.FileSystemId = fsConfig.Shard2Id; listNodesResponse = service.ListNodes( headers2, fsConfig.Shard2Id, RootNodeId)->Record; UNIT_ASSERT_VALUES_EQUAL(1, listNodesResponse.NamesSize()); UNIT_ASSERT_VALUES_EQUAL(1, listNodesResponse.NodesSize()); auto setXAttrResponse = service.SetNodeXAttr( headers, fsConfig.FsId, nodeId2, "user.some_attr", "some_value")->Record; UNIT_ASSERT_VALUES_EQUAL(1, setXAttrResponse.GetVersion()); auto getXAttrResponse = service.GetNodeXAttr( headers, fsConfig.FsId, nodeId2, "user.some_attr")->Record; UNIT_ASSERT_VALUES_EQUAL(1, getXAttrResponse.GetVersion()); UNIT_ASSERT_VALUES_EQUAL("some_value", getXAttrResponse.GetValue()); service.SendSetNodeXAttrRequest( headers, fsConfig.FsId, nodeId1, "user.some_attr", "some_value"); auto setNodeXAttrResponseEvent = service.RecvSetNodeXAttrResponse(); UNIT_ASSERT_VALUES_EQUAL_C( E_FS_NOENT, setNodeXAttrResponseEvent->GetStatus(), setNodeXAttrResponseEvent->GetErrorReason()); } SERVICE_TEST(ShouldSetHasXAttrsFlagOnFirstSetNodeXAttr) { for (bool lazyXAttrsEnabled: {false, true}) { config.SetLazyXAttrsEnabled(lazyXAttrsEnabled); TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); auto counters = env.GetCounters()->FindSubgroup("component", "service"); auto getCount = [&counters](const char* name) { return counters->FindSubgroup("request", name) ->GetCounter("Count") ->GetAtomic(); }; auto getErrorCount = [&counters](const char* name) { return counters->FindSubgroup("request", name) ->GetCounter("Errors") ->GetAtomic(); }; // Create initial session THeaders headers; auto session = service.InitSession(headers, fsConfig.FsId, "client"); if (lazyXAttrsEnabled) { // There should be no XAttrs at the begining if // lazyXAttrsEnabled is true UNIT_ASSERT(!session->Record.GetFileStore() .GetFeatures() .GetHasXAttrs()); } else { // Otherwise it's always true UNIT_ASSERT(session->Record.GetFileStore() .GetFeatures() .GetHasXAttrs()); } // Create two files: 'file1', 'file2' const auto createNodeResponse1 = service .CreateNode( headers, TCreateNodeArgs::File(RootNodeId, "file1")) ->Record; const auto nodeId1 = createNodeResponse1.GetNode().GetId(); const auto createNodeResponse2 = service .CreateNode( headers, TCreateNodeArgs::File(RootNodeId, "file2")) ->Record; const auto nodeId2 = createNodeResponse2.GetNode().GetId(); // Check that GetNodeXAttr returns error 'E_FS_NOXATTR' in case // 'HasXAttrs' is false auto getXAttrResponse = service .AssertGetNodeXAttrFailed( headers, fsConfig.FsId, nodeId1, "user.some_attr") ->Record; UNIT_ASSERT_VALUES_EQUAL( ui32(NProto::E_FS_NOXATTR), STATUS_FROM_CODE(getXAttrResponse.GetError().GetCode())); UNIT_ASSERT_VALUES_EQUAL(1, getErrorCount("GetNodeXAttr")); auto listXAttrResponse = service.ListNodeXAttr(headers, fsConfig.FsId, nodeId1)->Record; UNIT_ASSERT(listXAttrResponse.GetNames().empty()); UNIT_ASSERT_VALUES_EQUAL(1, getCount("ListNodeXAttr")); if (lazyXAttrsEnabled) { // If lazyXAttrsEnabled is true the first attempt should fail // and cause the main tablet to restart service.AssertSetNodeXAttrFailed( headers, fsConfig.FsId, nodeId1, "user.some_attr1", "some_value1"); WaitForTabletStart(service); session = service.InitSession(headers, fsConfig.FsId, "client"); // After the first XAttr is set the flag HasXAttrs should be // true UNIT_ASSERT(session->Record.GetFileStore() .GetFeatures() .GetHasXAttrs()); } service.SetNodeXAttr( headers, fsConfig.FsId, nodeId1, "user.some_attr1", "some_value1"); UNIT_ASSERT_VALUES_EQUAL(1, getCount("SetNodeXAttr")); // Check that XAttr was seccessfully set auto getXAttrResponse1 = service .GetNodeXAttr( headers, fsConfig.FsId, nodeId1, "user.some_attr1") ->Record; UNIT_ASSERT_VALUES_EQUAL( "some_value1", getXAttrResponse1.GetValue()); // The same check for 'file2'. // As this time the index tablet should not restart, we don't need // to create a new session service.SetNodeXAttr( headers, fsConfig.FsId, nodeId2, "user.some_attr2", "some_value2"); UNIT_ASSERT_VALUES_EQUAL(2, getCount("SetNodeXAttr")); auto getXAttrResponse2 = service .GetNodeXAttr( headers, fsConfig.FsId, nodeId2, "user.some_attr2") ->Record; UNIT_ASSERT_VALUES_EQUAL( "some_value2", getXAttrResponse2.GetValue()); UNIT_ASSERT_VALUES_EQUAL(2, getCount("GetNodeXAttr")); service.RemoveNodeXAttr( headers, fsConfig.FsId, nodeId1, "user.some_attr1"); UNIT_ASSERT_VALUES_EQUAL(1, getCount("RemoveNodeXAttr")); listXAttrResponse = service.ListNodeXAttr(headers, fsConfig.FsId, nodeId2)->Record; const auto& names = listXAttrResponse.GetNames(); UNIT_ASSERT_VALUES_EQUAL(1, names.size()); UNIT_ASSERT_VALUES_EQUAL("user.some_attr2", names[0]); listXAttrResponse = service.ListNodeXAttr(headers, fsConfig.FsId, nodeId1)->Record; UNIT_ASSERT(names.empty()); UNIT_ASSERT_VALUES_EQUAL(3, getCount("ListNodeXAttr")); } } SERVICE_TEST(ShouldCreateDirectoryStructureInLeader) { TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); auto headers = service.InitSession(fsConfig.FsId, "client"); auto createNodeResponse = service.CreateNode( headers, TCreateNodeArgs::Directory(RootNodeId, "dir1"))->Record; const auto dir1Id = createNodeResponse.GetNode().GetId(); UNIT_ASSERT_VALUES_EQUAL(0, ExtractShardNo(dir1Id)); createNodeResponse = service.CreateNode( headers, TCreateNodeArgs::Directory(dir1Id, "dir1_1"))->Record; const auto dir1_1Id = createNodeResponse.GetNode().GetId(); UNIT_ASSERT_VALUES_EQUAL(0, ExtractShardNo(dir1_1Id)); createNodeResponse = service.CreateNode( headers, TCreateNodeArgs::Directory(dir1Id, "dir1_2"))->Record; const auto dir1_2Id = createNodeResponse.GetNode().GetId(); UNIT_ASSERT_VALUES_EQUAL(0, ExtractShardNo(dir1_2Id)); auto createHandleResponse = service.CreateHandle( headers, fsConfig.FsId, dir1_1Id, "file1", TCreateHandleArgs::CREATE)->Record; const auto nodeId1 = createHandleResponse.GetNodeAttr().GetId(); UNIT_ASSERT_VALUES_EQUAL(ShardedId(2, 1), nodeId1); const auto handle1 = createHandleResponse.GetHandle(); UNIT_ASSERT_C(ExtractShardNo(handle1) == 1, handle1); createHandleResponse = service.CreateHandle( headers, fsConfig.FsId, dir1_2Id, "file1", TCreateHandleArgs::CREATE)->Record; const auto nodeId2 = createHandleResponse.GetNodeAttr().GetId(); UNIT_ASSERT_VALUES_EQUAL(ShardedId(2, 2), nodeId2); const auto handle2 = createHandleResponse.GetHandle(); UNIT_ASSERT_C(ExtractShardNo(handle2) == 2, handle2); auto data = GenerateValidateData(256_KB); service.WriteData(headers, fsConfig.FsId, nodeId1, handle1, 0, data); auto readDataResponse = service.ReadData( headers, fsConfig.FsId, nodeId1, handle1, 0, data.size())->Record; UNIT_ASSERT_VALUES_EQUAL(data, readDataResponse.GetBuffer()); data = GenerateValidateData(1_MB); service.WriteData(headers, fsConfig.FsId, nodeId2, handle2, 0, data); readDataResponse = service.ReadData( headers, fsConfig.FsId, nodeId2, handle2, 0, data.size())->Record; UNIT_ASSERT_VALUES_EQUAL(data, readDataResponse.GetBuffer()); service.DestroyHandle(headers, fsConfig.FsId, nodeId1, handle1); service.DestroyHandle(headers, fsConfig.FsId, nodeId2, handle2); } SERVICE_TEST(ShouldReturnRetriableErrorForUnknownShardNo) { TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); auto headers = service.InitSession(fsConfig.FsId, "client"); auto createHandleResponse = service.CreateHandle( headers, fsConfig.FsId, RootNodeId, "file1", TCreateHandleArgs::CREATE_EXL)->Record; auto data = GenerateValidateData(256_KB); const ui64 nodeId1 = createHandleResponse.GetNodeAttr().GetId(); const ui64 handle1 = createHandleResponse.GetHandle(); service.SendWriteDataRequest( headers, fsConfig.FsId, nodeId1, ShardedId(handle1, 111), 0, data); auto writeDataResponse = service.RecvWriteDataResponse(); UNIT_ASSERT_VALUES_EQUAL_C( E_REJECTED, writeDataResponse->GetStatus(), writeDataResponse->GetErrorReason()); } SERVICE_TEST(ShouldNotReturnInvalidShardNoErrorForDirectShardRequests) { TTestEnv env({}, config); ui32 nodeIdx = env.AddDynamicNode(); const TString fsId = "test"; TServiceClient service(env.GetRuntime(), nodeIdx); service.CreateFileStore(fsId, 1'000); // forcing fs to set high bits for handle ids { NProtoPrivate::TConfigureAsShardRequest request; request.SetFileSystemId(fsId); request.SetShardNo(111); TString buf; google::protobuf::util::MessageToJsonString(request, &buf); auto jsonResponse = service.ExecuteAction("configureasshard", buf); NProtoPrivate::TConfigureAsShardResponse response; UNIT_ASSERT(google::protobuf::util::JsonStringToMessage( jsonResponse->Record.GetOutput(), &response).ok()); } auto headers = service.InitSession(fsId, "client"); auto createHandleResponse = service.CreateHandle( headers, fsId, RootNodeId, "file1", TCreateHandleArgs::CREATE_EXL)->Record; const auto nodeId1 = createHandleResponse.GetNodeAttr().GetId(); UNIT_ASSERT_VALUES_EQUAL(111, ExtractShardNo(nodeId1)); const auto handle1 = createHandleResponse.GetHandle(); UNIT_ASSERT_VALUES_EQUAL(111, ExtractShardNo(handle1)); auto data = GenerateValidateData(256_KB); service.SendWriteDataRequest(headers, fsId, nodeId1, handle1, 0, data); auto writeDataResponse = service.RecvWriteDataResponse(); UNIT_ASSERT_VALUES_EQUAL_C( S_OK, writeDataResponse->GetStatus(), writeDataResponse->GetErrorReason()); service.SendReadDataRequest( headers, fsId, nodeId1, handle1, 0, data.size()); auto readDataResponse = service.RecvReadDataResponse(); UNIT_ASSERT_VALUES_EQUAL_C( S_OK, readDataResponse->GetStatus(), readDataResponse->GetErrorReason()); UNIT_ASSERT_VALUES_EQUAL(data, readDataResponse->Record.GetBuffer()); } SERVICE_TEST(ShouldHandleCreateNodeErrorFromShardUponCreateHandleViaLeader) { TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); auto headers = service.InitSession(fsConfig.FsId, "client"); auto error = MakeError(E_FAIL); env.GetRuntime().SetEventFilter( [&] (TTestActorRuntimeBase&, TAutoPtr& event) { switch (event->GetTypeRewrite()) { case TEvService::EvCreateNodeResponse: { auto* msg = event->Get(); if (error.GetCode()) { msg->Record.MutableError()->CopyFrom(error); } break; } } return false; }); const ui64 requestId = 111; service.SendCreateHandleRequest( headers, fsConfig.FsId, RootNodeId, "file1", TCreateHandleArgs::CREATE_EXL, "", requestId); auto response = service.RecvCreateHandleResponse(); UNIT_ASSERT_VALUES_EQUAL_C( error.GetCode(), response->GetError().GetCode(), FormatError(response->GetError())); error = {}; service.SendCreateHandleRequest( headers, fsConfig.FsId, RootNodeId, "file1", TCreateHandleArgs::CREATE_EXL, "", requestId); response = service.RecvCreateHandleResponse(); UNIT_ASSERT_VALUES_EQUAL_C( error.GetCode(), response->GetError().GetCode(), FormatError(response->GetError())); } SERVICE_TEST(ShouldNotFailListNodesUponGetAttrENOENT) { TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); auto headers = service.InitSession(fsConfig.FsId, "client"); service.CreateNode( headers, TCreateNodeArgs::File(RootNodeId, "file1")); service.CreateNode( headers, TCreateNodeArgs::File(RootNodeId, "file2")); service.CreateNode( headers, TCreateNodeArgs::File(RootNodeId, "file3")); service.CreateNode( headers, TCreateNodeArgs::File(RootNodeId, "file4")); auto listNodesResponse = service.ListNodes( headers, fsConfig.FsId, RootNodeId)->Record; UNIT_ASSERT_VALUES_EQUAL(4, listNodesResponse.NamesSize()); UNIT_ASSERT_VALUES_EQUAL("file1", listNodesResponse.GetNames(0)); UNIT_ASSERT_VALUES_EQUAL("file2", listNodesResponse.GetNames(1)); UNIT_ASSERT_VALUES_EQUAL("file3", listNodesResponse.GetNames(2)); UNIT_ASSERT_VALUES_EQUAL("file4", listNodesResponse.GetNames(3)); TVector> nodes(4); for (ui32 i = 0; i < 4; ++i) { nodes[i] = { listNodesResponse.GetNodes(i).GetId(), listNodesResponse.GetNames(i)}; UNIT_ASSERT_VALUES_UNEQUAL(0, nodes[i].first); } auto headers1 = headers; headers1.FileSystemId = fsConfig.Shard1Id; listNodesResponse = service.ListNodes( headers1, fsConfig.Shard1Id, RootNodeId)->Record; UNIT_ASSERT_VALUES_EQUAL(2, listNodesResponse.NamesSize()); const auto shard1NodeName1 = listNodesResponse.GetNames(0); const auto shard1NodeId1 = listNodesResponse.GetNodes(0).GetId(); const auto shard1NodeName2 = listNodesResponse.GetNames(1); auto headers2 = headers; headers2.FileSystemId = fsConfig.Shard2Id; listNodesResponse = service.ListNodes( headers2, fsConfig.Shard2Id, RootNodeId)->Record; UNIT_ASSERT_VALUES_EQUAL(2, listNodesResponse.NamesSize()); const auto shard2NodeName1 = listNodesResponse.GetNames(0); const auto shard2NodeName2 = listNodesResponse.GetNames(1); // "breaking" one node - deleting it directly from the shard service.UnlinkNode(headers1, RootNodeId, shard1NodeName1); EraseIf(nodes, [=] (const auto& node) { return node.first == shard1NodeId1; }); // ListNodes should still succeed listNodesResponse = service.ListNodes( headers, fsConfig.FsId, RootNodeId)->Record; // unresolved nodes should be removed from the response UNIT_ASSERT_VALUES_EQUAL(3, listNodesResponse.NamesSize()); UNIT_ASSERT_VALUES_EQUAL( nodes[0].second, listNodesResponse.GetNames(0)); UNIT_ASSERT_VALUES_EQUAL( nodes[1].second, listNodesResponse.GetNames(1)); UNIT_ASSERT_VALUES_EQUAL( nodes[2].second, listNodesResponse.GetNames(2)); UNIT_ASSERT_VALUES_EQUAL(3, listNodesResponse.NodesSize()); UNIT_ASSERT_VALUES_EQUAL( nodes[0].first, listNodesResponse.GetNodes(0).GetId()); UNIT_ASSERT_VALUES_EQUAL( nodes[1].first, listNodesResponse.GetNodes(1).GetId()); UNIT_ASSERT_VALUES_EQUAL( nodes[2].first, listNodesResponse.GetNodes(2).GetId()); const auto counters = env.GetCounters()->FindSubgroup("component", "service"); UNIT_ASSERT(counters); const auto counter = counters->GetCounter("AppCriticalEvents/NodeNotFoundInShard"); UNIT_ASSERT_EQUAL(1, counter->GetAtomic()); auto& runtime = env.GetRuntime(); // However, ListNodes should not report a critical event in case of // a retriable error from the leader { runtime.SetEventFilter( [&](TTestActorRuntimeBase&, TAutoPtr& event) { switch (event->GetTypeRewrite()) { case TEvService::EvGetNodeAttrResponse: { auto* msg = event ->Get(); msg->Record.MutableError()->CopyFrom( MakeError(E_REJECTED, "error")); break; } } return false; }); auto response = service.SendAndRecvListNodes( headers, fsConfig.FsId, RootNodeId); UNIT_ASSERT_VALUES_EQUAL_C( E_REJECTED, response->GetError().GetCode(), response->GetErrorReason()); UNIT_ASSERT_VALUES_EQUAL(1, counter->GetAtomic()); runtime.SetEventFilter(TTestActorRuntimeBase::DefaultFilterFunc); } // "breaking" all nodes - ListNodes should fail with E_IO after this service.UnlinkNode(headers1, RootNodeId, shard1NodeName2); service.UnlinkNode(headers2, RootNodeId, shard2NodeName1); service.UnlinkNode(headers2, RootNodeId, shard2NodeName2); service.SendListNodesRequest(headers, fsConfig.FsId, RootNodeId); auto response = service.RecvListNodesResponse(); UNIT_ASSERT_VALUES_EQUAL_C( E_IO, response->GetError().GetCode(), response->GetErrorReason()); } SERVICE_TEST(ShouldListMultipleNodesWithGetNodeAttrBatch) { TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); auto headers = service.InitSession(fsConfig.FsId, "client"); auto headers1 = headers; headers1.FileSystemId = fsConfig.Shard1Id; TVector names; TVector ids; for (ui32 i = 10; i < 50; ++i) { const auto name = Sprintf("file%u", i); const auto id = service.CreateNode( headers, TCreateNodeArgs::File(RootNodeId, name) )->Record.GetNode().GetId(); names.push_back(name); ids.push_back(id); } auto listNodesResponse = service.ListNodes( headers1, fsConfig.Shard1Id, RootNodeId)->Record; UNIT_ASSERT_VALUES_EQUAL( names.size() / 2, listNodesResponse.NamesSize()); const ui32 idxToUnlink1 = 13; const ui32 idxToUnlink2 = 17; const auto shard1NodeName1 = listNodesResponse.GetNames(idxToUnlink1); const ui64 unlinkedId1 = listNodesResponse.GetNodes(idxToUnlink1).GetId(); const auto shard1NodeName2 = listNodesResponse.GetNames(idxToUnlink2); const ui64 unlinkedId2 = listNodesResponse.GetNodes(idxToUnlink2).GetId(); listNodesResponse = service.ListNodes( headers, fsConfig.FsId, RootNodeId)->Record; UNIT_ASSERT_VALUES_EQUAL(names.size(), listNodesResponse.NamesSize()); for (ui32 i = 0; i < names.size(); ++i) { UNIT_ASSERT_VALUES_EQUAL(names[i], listNodesResponse.GetNames(i)); UNIT_ASSERT_VALUES_EQUAL( ids[i], listNodesResponse.GetNodes(i).GetId()); } service.UnlinkNode(headers1, RootNodeId, shard1NodeName1); service.UnlinkNode(headers1, RootNodeId, shard1NodeName2); listNodesResponse = service.ListNodes( headers, fsConfig.FsId, RootNodeId)->Record; for (auto id: {unlinkedId1, unlinkedId2}) { auto idx = Find(ids, id) - ids.begin(); ids.erase(ids.begin() + idx); names.erase(names.begin() + idx); } UNIT_ASSERT_VALUES_EQUAL(names.size(), listNodesResponse.NamesSize()); for (ui32 i = 0; i < names.size(); ++i) { UNIT_ASSERT_VALUES_EQUAL(names[i], listNodesResponse.GetNames(i)); UNIT_ASSERT_VALUES_EQUAL( ids[i], listNodesResponse.GetNodes(i).GetId()); } } SERVICE_TEST(ShouldValidateRequestsWithShardId) { TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); auto headers = service.InitSession(fsConfig.FsId, "client"); const auto createNodeResponse = service.CreateNode( headers, TCreateNodeArgs::File(RootNodeId, "file1"))->Record; const auto nodeId1 = createNodeResponse.GetNode().GetId(); auto headers1 = headers; headers1.FileSystemId = fsConfig.Shard1Id; // a request with ShardId and without Name service.SendCreateHandleRequest( headers, fsConfig.FsId, nodeId1, "", TCreateHandleArgs::RDWR, fsConfig.Shard1Id); auto createHandleResponseEvent = service.RecvCreateHandleResponse(); UNIT_ASSERT_VALUES_EQUAL_C( E_ARGUMENT, createHandleResponseEvent->GetError().GetCode(), createHandleResponseEvent->GetErrorReason()); } SERVICE_TEST(ShouldValidateShardConfiguration) { TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); const auto shard3Id = fsConfig.FsId + "_s3"; service.CreateFileStore(shard3Id, fsConfig.ShardBlockCount - 1); // ShardNo change not allowed { NProtoPrivate::TConfigureAsShardRequest request; request.SetFileSystemId(fsConfig.Shard1Id); request.SetShardNo(2); TString buf; google::protobuf::util::MessageToJsonString(request, &buf); service.SendExecuteActionRequest("configureasshard", buf); auto response = service.RecvExecuteActionResponse(); UNIT_ASSERT_VALUES_EQUAL_C( E_ARGUMENT, response->GetError().GetCode(), response->GetErrorReason()); } // Shard deletion not allowed { NProtoPrivate::TConfigureShardsRequest request; request.SetFileSystemId(fsConfig.FsId); *request.AddShardFileSystemIds() = fsConfig.Shard1Id; TString buf; google::protobuf::util::MessageToJsonString(request, &buf); service.SendExecuteActionRequest("configureshards", buf); auto response = service.RecvExecuteActionResponse(); UNIT_ASSERT_VALUES_EQUAL_C( E_ARGUMENT, response->GetError().GetCode(), response->GetErrorReason()); } // Shard reordering not allowed { NProtoPrivate::TConfigureShardsRequest request; request.SetFileSystemId(fsConfig.FsId); *request.AddShardFileSystemIds() = fsConfig.Shard2Id; *request.AddShardFileSystemIds() = fsConfig.Shard1Id; TString buf; google::protobuf::util::MessageToJsonString(request, &buf); service.SendExecuteActionRequest("configureshards", buf); auto response = service.RecvExecuteActionResponse(); UNIT_ASSERT_VALUES_EQUAL_C( E_ARGUMENT, response->GetError().GetCode(), response->GetErrorReason()); } // Shard addition IS allowed { NProtoPrivate::TConfigureAsShardRequest request; request.SetFileSystemId(shard3Id); request.SetShardNo(3); TString buf; google::protobuf::util::MessageToJsonString(request, &buf); auto jsonResponse = service.ExecuteAction("configureasshard", buf); NProtoPrivate::TConfigureAsShardResponse response; UNIT_ASSERT(google::protobuf::util::JsonStringToMessage( jsonResponse->Record.GetOutput(), &response).ok()); } { NProtoPrivate::TConfigureShardsRequest request; request.SetFileSystemId(fsConfig.FsId); *request.AddShardFileSystemIds() = fsConfig.Shard1Id; *request.AddShardFileSystemIds() = fsConfig.Shard2Id; *request.AddShardFileSystemIds() = shard3Id; TString buf; google::protobuf::util::MessageToJsonString(request, &buf); auto jsonResponse = service.ExecuteAction("configureshards", buf); NProtoPrivate::TConfigureShardsResponse response; UNIT_ASSERT(google::protobuf::util::JsonStringToMessage( jsonResponse->Record.GetOutput(), &response).ok()); } // TODO(#1350): leader should check that shards' ShardNos correspond // to the shard order in leader's config } SERVICE_TEST(ShouldRenameExternalNodes) { TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); auto headers = service.InitSession(fsConfig.FsId, "client"); // creating 2 files auto createNodeResponse = service.CreateNode( headers, TCreateNodeArgs::File(RootNodeId, "file1"))->Record; const auto nodeId1 = createNodeResponse.GetNode().GetId(); UNIT_ASSERT_VALUES_EQUAL(1, ExtractShardNo(nodeId1)); createNodeResponse = service.CreateNode( headers, TCreateNodeArgs::File(RootNodeId, "file2"))->Record; const auto nodeId2 = createNodeResponse.GetNode().GetId(); UNIT_ASSERT_VALUES_EQUAL(2, ExtractShardNo(nodeId2)); ui64 handle1 = service.CreateHandle( headers, fsConfig.FsId, nodeId1, "", TCreateHandleArgs::RDWR)->Record.GetHandle(); // writing some data to file1 then moving file1 to file3 auto data1 = GenerateValidateData(256_KB); service.WriteData(headers, fsConfig.FsId, nodeId1, handle1, 0, data1); auto renameNodeResponse = service.RenameNode( headers, RootNodeId, "file1", RootNodeId, "file3", 0); // opening file3 for reading auto createHandleResponse = service.CreateHandle( headers, fsConfig.FsId, RootNodeId, "file3", TCreateHandleArgs::RDNLY)->Record; // checking that file3 refers to the same node as formerly file1 UNIT_ASSERT_VALUES_EQUAL( nodeId1, createHandleResponse.GetNodeAttr().GetId()); auto handle1r = createHandleResponse.GetHandle(); // checking that we can read the data that we wrote to file1 auto readDataResponse = service.ReadData( headers, fsConfig.FsId, nodeId1, handle1r, 0, data1.size())->Record; UNIT_ASSERT_VALUES_EQUAL(data1, readDataResponse.GetBuffer()); // checking that node listing shows file3 and file2 auto listNodesResponse = service.ListNodes( headers, fsConfig.FsId, RootNodeId)->Record; UNIT_ASSERT_VALUES_EQUAL(2, listNodesResponse.NamesSize()); UNIT_ASSERT_VALUES_EQUAL("file2", listNodesResponse.GetNames(0)); UNIT_ASSERT_VALUES_EQUAL("file3", listNodesResponse.GetNames(1)); UNIT_ASSERT_VALUES_EQUAL(2, listNodesResponse.NodesSize()); UNIT_ASSERT_VALUES_EQUAL( nodeId2, listNodesResponse.GetNodes(0).GetId()); UNIT_ASSERT_VALUES_EQUAL( nodeId1, listNodesResponse.GetNodes(1).GetId()); // checking that move into an existing file (file2) works renameNodeResponse = service.RenameNode( headers, RootNodeId, "file3", RootNodeId, "file2", 0); // checking that we can still read the same data createHandleResponse = service.CreateHandle( headers, fsConfig.FsId, RootNodeId, "file2", TCreateHandleArgs::RDNLY)->Record; // nodeId should be kept intact UNIT_ASSERT_VALUES_EQUAL( nodeId1, createHandleResponse.GetNodeAttr().GetId()); handle1r = createHandleResponse.GetHandle(); readDataResponse = service.ReadData( headers, fsConfig.FsId, nodeId1, handle1r, 0, data1.size())->Record; UNIT_ASSERT_VALUES_EQUAL(data1, readDataResponse.GetBuffer()); // listing should show only file2 now listNodesResponse = service.ListNodes( headers, fsConfig.FsId, RootNodeId)->Record; UNIT_ASSERT_VALUES_EQUAL(1, listNodesResponse.NamesSize()); UNIT_ASSERT_VALUES_EQUAL("file2", listNodesResponse.GetNames(0)); UNIT_ASSERT_VALUES_EQUAL(1, listNodesResponse.NodesSize()); UNIT_ASSERT_VALUES_EQUAL( nodeId1, listNodesResponse.GetNodes(0).GetId()); // listing in shard2 should show nothing auto headers2 = headers; headers2.FileSystemId = fsConfig.Shard2Id; listNodesResponse = service.ListNodes( headers2, fsConfig.Shard2Id, RootNodeId)->Record; UNIT_ASSERT_VALUES_EQUAL(0, listNodesResponse.NamesSize()); UNIT_ASSERT_VALUES_EQUAL(0, listNodesResponse.NodesSize()); // now try to move to another subdirectory auto subdirId = service .CreateNode( headers, TCreateNodeArgs::Directory(RootNodeId, "subdir")) ->Record.GetNode() .GetId(); renameNodeResponse = service.RenameNode( headers, RootNodeId, "file2", subdirId, "file2", 0); // listing should show only subdir with file2 in it listNodesResponse = service.ListNodes( headers, fsConfig.FsId, RootNodeId)->Record; UNIT_ASSERT_VALUES_EQUAL(1, listNodesResponse.NamesSize()); UNIT_ASSERT_VALUES_EQUAL("subdir", listNodesResponse.GetNames(0)); listNodesResponse = service.ListNodes(headers, fsConfig.FsId, subdirId)->Record; UNIT_ASSERT_VALUES_EQUAL(1, listNodesResponse.NamesSize()); UNIT_ASSERT_VALUES_EQUAL("file2", listNodesResponse.GetNames(0)); // create 2 files in the same shard ui64 nodeId4 = service .CreateNode(headers, TCreateNodeArgs::File(RootNodeId, "file4")) ->Record.GetNode() .GetId(); // round robin is used for 2 clusters, so we just skip one shard by // creating a file in it service.CreateNode(headers, TCreateNodeArgs::File(RootNodeId, "file5")); ui64 nodeId6 = service .CreateNode(headers, TCreateNodeArgs::File(RootNodeId, "file6")) ->Record.GetNode() .GetId(); UNIT_ASSERT_VALUES_EQUAL( ExtractShardNo(nodeId4), ExtractShardNo(nodeId6)); // now move to the same shard renameNodeResponse = service.RenameNode( headers, RootNodeId, "file4", RootNodeId, "file6", 0); // file4 should not be present in the listing service.SendGetNodeAttrRequest( headers, fsConfig.FsId, RootNodeId, "file4"); auto getNodeAttrResponse = service.RecvGetNodeAttrResponse(); UNIT_ASSERT(getNodeAttrResponse); UNIT_ASSERT_C( FAILED(getNodeAttrResponse->GetStatus()), getNodeAttrResponse->GetErrorReason().c_str()); } SERVICE_TEST(ShouldPerformLocksForExternalNodes) { TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); auto headers = service.InitSession(fsConfig.FsId, "client"); ui64 nodeId = service .CreateNode(headers, TCreateNodeArgs::File(RootNodeId, "file")) ->Record.GetNode() .GetId(); ui64 handle = service.CreateHandle( headers, fsConfig.FsId, nodeId, "", TCreateHandleArgs::RDWR)->Record.GetHandle(); service.AcquireLock(headers, fsConfig.FsId, handle, 1, 0, 4_KB); auto response = service.AssertAcquireLockFailed( headers, fsConfig.FsId, handle, 2, 0, 4_KB); UNIT_ASSERT_VALUES_EQUAL(response->GetError().GetCode(), E_FS_WOULDBLOCK); service.ReleaseLock(headers, fsConfig.FsId, handle, 2, 0, 4_KB); service.AcquireLock(headers, fsConfig.FsId, handle, 1, 0, 0); response = service.AssertAcquireLockFailed( headers, fsConfig.FsId, handle, 2, 0, 4_KB); UNIT_ASSERT_VALUES_EQUAL(response->GetError().GetCode(), E_FS_WOULDBLOCK); service.ReleaseLock(headers, fsConfig.FsId, handle, 1, 0, 0); // ShouldTrackSharedLocks service.AcquireLock( headers, fsConfig.FsId, handle, 1, 0, 4_KB, DefaultPid, NProto::E_SHARED); service.AcquireLock( headers, fsConfig.FsId, handle, 2, 0, 4_KB, DefaultPid, NProto::E_SHARED); response = service.AssertAcquireLockFailed( headers, fsConfig.FsId, handle, 1, 0, 0); UNIT_ASSERT_VALUES_EQUAL( response->GetError().GetCode(), E_FS_WOULDBLOCK); response = service.AssertAcquireLockFailed( headers, fsConfig.FsId, handle, 1, 0, 0); UNIT_ASSERT_VALUES_EQUAL(response->GetError().GetCode(), E_FS_WOULDBLOCK); service.AcquireLock( headers, fsConfig.FsId, handle, 3, 0, 4_KB, DefaultPid, NProto::E_SHARED); service.DestroyHandle(headers, fsConfig.FsId, nodeId, handle); handle = service.CreateHandle( headers, fsConfig.FsId, nodeId, "", TCreateHandleArgs::WRNLY)->Record.GetHandle(); service.AssertTestLockFailed( headers, fsConfig.FsId, handle, 1, 0, 4_KB, DefaultPid, NProto::E_SHARED); service.AssertAcquireLockFailed( headers, fsConfig.FsId, handle, 1, 0, 4_KB, DefaultPid, NProto::E_SHARED); } SERVICE_TEST(ShouldLinkExternalNodes) { TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); auto headers = service.InitSession(fsConfig.FsId, "client"); auto headers1 = headers; headers1.FileSystemId = fsConfig.Shard1Id; const auto nodeId1 = service .CreateNode(headers, TCreateNodeArgs::File(RootNodeId, "file1")) ->Record.GetNode() .GetId(); UNIT_ASSERT_VALUES_EQUAL(1, ExtractShardNo(nodeId1)); auto node = service.GetNodeAttr( headers, fsConfig.FsId, RootNodeId, "file1")->Record.GetNode(); UNIT_ASSERT_VALUES_EQUAL(1, node.GetLinks()); auto linkNodeResponse = service.CreateNode( headers, TCreateNodeArgs::Link(RootNodeId, "file2", nodeId1)); UNIT_ASSERT(linkNodeResponse); UNIT_ASSERT_C( SUCCEEDED(linkNodeResponse->GetStatus()), linkNodeResponse->GetErrorReason().c_str()); // attributes of the linked node should be updated UNIT_ASSERT_VALUES_EQUAL( 2, linkNodeResponse->Record.GetNode().GetLinks()); UNIT_ASSERT_VALUES_UNEQUAL( 0, linkNodeResponse->Record.GetNode().GetCTime()); // validate that the links field is incremented auto links = service.GetNodeAttr( headers, fsConfig.FsId, RootNodeId, "file1")->Record.GetNode().GetLinks(); UNIT_ASSERT_VALUES_EQUAL(2, links); // get node attr should also work with the link links = service.GetNodeAttr(headers, fsConfig.FsId, RootNodeId, "file1") ->Record.GetNode().GetLinks(); UNIT_ASSERT_VALUES_EQUAL(2, links); // validate that reading from hardlinked file works auto data = GenerateValidateData(256_KB); ui64 handle = service.CreateHandle( headers, fsConfig.FsId, RootNodeId, "file1", TCreateHandleArgs::RDWR)->Record.GetHandle(); service.WriteData(headers, fsConfig.FsId, nodeId1, handle, 0, data); ui64 handle2 = service.CreateHandle( headers, fsConfig.FsId, RootNodeId, "file2", TCreateHandleArgs::RDWR)->Record.GetHandle(); auto data2 = service.ReadData( headers, fsConfig.FsId, linkNodeResponse->Record.GetNode().GetId(), handle2, 0, data.size())->Record.GetBuffer(); UNIT_ASSERT_VALUES_EQUAL(data, data2); // Removal of both the file and a hardlink should remove file from both // the shard and a leader service.DestroyHandle(headers, fsConfig.FsId, nodeId1, handle); service.DestroyHandle( headers, fsConfig.FsId, linkNodeResponse->Record.GetNode().GetId(), handle2); service.UnlinkNode(headers, RootNodeId, "file1"); service.UnlinkNode(headers, RootNodeId, "file2"); // Now listing of the root should show no files auto listNodesResponse = service.ListNodes( headers, fsConfig.FsId, RootNodeId)->Record; UNIT_ASSERT_VALUES_EQUAL(0, listNodesResponse.NamesSize()); UNIT_ASSERT_VALUES_EQUAL(0, listNodesResponse.NodesSize()); listNodesResponse = service.ListNodes( headers1, fsConfig.Shard1Id, RootNodeId)->Record; UNIT_ASSERT_VALUES_EQUAL(0, listNodesResponse.NamesSize()); UNIT_ASSERT_VALUES_EQUAL(0, listNodesResponse.NodesSize()); // GetNodeAttr should fail as well service.AssertGetNodeAttrFailed( headers, fsConfig.FsId, RootNodeId, "file1"); service.AssertGetNodeAttrFailed( headers, fsConfig.FsId, RootNodeId, "file2"); // Creating hardlinks to non-existing files should fail. It is // reasonable to assume that nodeId + 100 is not a valid node id. linkNodeResponse = service.AssertCreateNodeFailed( headers, TCreateNodeArgs::Link(RootNodeId, "file3", nodeId1 + 100)); } SERVICE_TEST(ShouldRejectCrossShardHardLinkAcrossQuotaDomains) { TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); auto headers = service.InitSession(fsConfig.FsId, "client"); SetQuota(service, fsConfig.FsId, 1, 1_GB, 100); SetQuota(service, fsConfig.FsId, 2, 1_GB, 100); const auto dirAId = service .CreateNode(headers, TCreateNodeArgs::Directory(RootNodeId, "dirA")) ->Record.GetNode() .GetId(); service.SetNodeAttr( headers, fsConfig.FsId, TSetNodeAttrArgs(dirAId).SetQuotaId(1)); const auto dirBId = service .CreateNode(headers, TCreateNodeArgs::Directory(RootNodeId, "dirB")) ->Record.GetNode() .GetId(); service.SetNodeAttr( headers, fsConfig.FsId, TSetNodeAttrArgs(dirBId).SetQuotaId(2)); // target inherits QuotaId 1 from dirA and lands in a shard, // external to both dirA and dirB const auto targetId = service .CreateNode(headers, TCreateNodeArgs::File(dirAId, "target")) ->Record.GetNode() .GetId(); UNIT_ASSERT_VALUES_UNEQUAL(0, ExtractShardNo(targetId)); UNIT_ASSERT_VALUES_EQUAL( 1u, service.GetNodeAttr(headers, fsConfig.FsId, targetId, "") ->Record.GetNode() .GetLinks()); // dirB is a different quota domain - the cross-shard hard link is // rejected outright service.AssertCreateNodeFailed( headers, TCreateNodeArgs::Link(dirBId, "linked", targetId)); // the shard's link count must have been reverted, not left at 2 // with nothing to show for it UNIT_ASSERT_VALUES_EQUAL( 1u, service.GetNodeAttr(headers, fsConfig.FsId, targetId, "") ->Record.GetNode() .GetLinks()); // no ref should have been left behind under dirB either service.AssertGetNodeAttrFailed( headers, fsConfig.FsId, dirBId, "linked"); // a same-domain cross-shard hard link still works const auto dirCId = service .CreateNode(headers, TCreateNodeArgs::Directory(RootNodeId, "dirC")) ->Record.GetNode() .GetId(); service.SetNodeAttr( headers, fsConfig.FsId, TSetNodeAttrArgs(dirCId).SetQuotaId(1)); auto linkResponse = service.CreateNode( headers, TCreateNodeArgs::Link(dirCId, "linked", targetId)); UNIT_ASSERT_C( SUCCEEDED(linkResponse->GetStatus()), linkResponse->GetErrorReason()); UNIT_ASSERT_VALUES_EQUAL(2u, linkResponse->Record.GetNode().GetLinks()); } SERVICE_TEST(ShouldNotLeaveDanglingLinksForInvalidHardLinks) { TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); auto headers = service.InitSession(fsConfig.FsId, "client"); const auto targetId = service .CreateNode(headers, TCreateNodeArgs::File(RootNodeId, "target")) ->Record.GetNode() .GetId(); UNIT_ASSERT_VALUES_UNEQUAL(0, ExtractShardNo(targetId)); UNIT_ASSERT_VALUES_EQUAL( 1u, service.GetNodeAttr(headers, fsConfig.FsId, targetId, "") ->Record.GetNode() .GetLinks()); // a local file already occupies the destination name on the leader const auto existingId = service .CreateNode(headers, TCreateNodeArgs::File(RootNodeId, "existing")) ->Record.GetNode() .GetId(); // the cross-shard hard link is rejected with E_FS_EXIST at the // leader, after the shard has already created its side of the link service.AssertCreateNodeFailed( headers, TCreateNodeArgs::Link(RootNodeId, "existing", targetId)); // TLinkActor's compensating undo must have reverted the shard's // link count - not left it at 2 with nothing to show for it UNIT_ASSERT_VALUES_EQUAL( 1u, service.GetNodeAttr(headers, fsConfig.FsId, targetId, "") ->Record.GetNode() .GetLinks()); // the pre-existing "existing" entry must be untouched UNIT_ASSERT_VALUES_EQUAL( existingId, service.GetNodeAttr(headers, fsConfig.FsId, RootNodeId, "existing") ->Record.GetNode() .GetId()); // now make the undo itself fail non-retriably and check that this // is reported as a critical event rather than swallowed silently const auto target2Id = service .CreateNode(headers, TCreateNodeArgs::File(RootNodeId, "target2")) ->Record.GetNode() .GetId(); UNIT_ASSERT_VALUES_UNEQUAL(0, ExtractShardNo(target2Id)); const auto counters = env.GetCounters()->FindSubgroup("component", "service"); UNIT_ASSERT(counters); const auto counter = counters->GetCounter("AppCriticalEvents/HardLinkUndoFailed"); UNIT_ASSERT_VALUES_EQUAL(0, counter->GetAtomic()); env.GetRuntime().SetEventFilter( [&](auto& runtime, TAutoPtr& event) { Y_UNUSED(runtime); if (event->GetTypeRewrite() == TEvService::EvUnlinkNodeResponse) { using TResponse = TEvService::TEvUnlinkNodeResponse; auto* msg = event->Get(); *msg->Record.MutableError() = MakeError(E_ARGUMENT, "injected undo failure"); } return false; }); // same collision as above, once again forcing the leader to reject // the link after the shard already created its side of it service.AssertCreateNodeFailed( headers, TCreateNodeArgs::Link(RootNodeId, "existing", target2Id)); UNIT_ASSERT_VALUES_EQUAL(1, counter->GetAtomic()); // E_FS_NOENT on the undo is acceptable const auto target3Id = service .CreateNode( headers, TCreateNodeArgs::File(RootNodeId, "target3")) ->Record.GetNode() .GetId(); UNIT_ASSERT_VALUES_UNEQUAL(0, ExtractShardNo(target3Id)); ui32 unlinkResponseCount = 0; env.GetRuntime().SetEventFilter( [&](auto& runtime, TAutoPtr& event) { Y_UNUSED(runtime); if (event->GetTypeRewrite() == TEvService::EvUnlinkNodeResponse) { using TResponse = TEvService::TEvUnlinkNodeResponse; auto* msg = event->Get(); if (!msg->Record.GetError().GetMessage().Contains( "injected")) { *msg->Record.MutableError() = ++unlinkResponseCount == 1 ? MakeError(E_REJECTED, "injected retriable") : MakeError(E_FS_NOENT, "injected no ent"); } } return false; }); service.AssertCreateNodeFailed( headers, TCreateNodeArgs::Link(RootNodeId, "existing", target3Id)); UNIT_ASSERT_VALUES_EQUAL(2u, unlinkResponseCount); // no new critical event - the counter must stay exactly where the // previous (genuine failure) scenario left it UNIT_ASSERT_VALUES_EQUAL(1, counter->GetAtomic()); // the real unlink went through regardless of the injected responses UNIT_ASSERT_VALUES_EQUAL( 1u, service.GetNodeAttr(headers, fsConfig.FsId, target3Id, "") ->Record.GetNode() .GetLinks()); // a retriable error from the leader is ambiguous - its CreateNode // may have actually committed despite it (e.g. a pipe reset can // surface as E_REJECTED for an already-committed request). Undoing // here would risk deleting a live shard-side link, so it must be // left alone and the error simply propagated instead. const auto target4Id = service .CreateNode( headers, TCreateNodeArgs::File(RootNodeId, "target4")) ->Record.GetNode() .GetId(); UNIT_ASSERT_VALUES_UNEQUAL(0, ExtractShardNo(target4Id)); // intercept the leader-bound hop specifically (identified by // ShardNodeAttr - only set on the 2nd, leader-bound request) and // answer it with a synthetic retriable error before it ever reaches // the leader tablet, mirroring ShouldRetryNodeCreationInShard's // interception style TAutoPtr leaderCreateResponse; bool intercept = true; env.GetRuntime().SetEventFilter( [&](auto& runtime, TAutoPtr& event) { Y_UNUSED(runtime); if (event->GetTypeRewrite() == TEvService::EvCreateNodeRequest) { const auto* msg = event->Get(); if (intercept && msg->Record.HasShardNodeAttr()) { auto response = std::make_unique< TEvService::TEvCreateNodeResponse>( MakeError( E_REJECTED, "injected retriable leader error")); leaderCreateResponse = new IEventHandle( event->Sender, event->Recipient, response.release(), 0, // flags event->Cookie); return true; } } return false; }); service.SendCreateNodeRequest( headers, TCreateNodeArgs::Link(RootNodeId, "target4link", target4Id)); ui32 iterations = 0; while (!leaderCreateResponse && iterations++ < 100) { env.GetRuntime().DispatchEvents({}, TDuration::MilliSeconds(50)); } UNIT_ASSERT(leaderCreateResponse); intercept = false; env.GetRuntime().Send(leaderCreateResponse.Release(), nodeIdx); auto createResponse = service.RecvCreateNodeResponse(); UNIT_ASSERT_C( FAILED(createResponse->GetStatus()), "CreateNode has not failed as expected"); // never actually reached the leader tablet - no nodeRef was created service.AssertGetNodeAttrFailed( headers, fsConfig.FsId, RootNodeId, "target4link"); // the shard-side link count must be left alone - undoing it would // have deleted a link that, for all TLinkActor knows, a leader // nodeRef might still point to UNIT_ASSERT_VALUES_EQUAL( 2u, service.GetNodeAttr(headers, fsConfig.FsId, target4Id, "") ->Record.GetNode() .GetLinks()); // no critical event either - undo was never attempted UNIT_ASSERT_VALUES_EQUAL(1, counter->GetAtomic()); } SERVICE_TEST(ShouldAggregateFileSystemMetrics) { TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); auto headers = service.InitSession(fsConfig.FsId, "client"); // creating 2 files auto createNodeResponse = service.CreateNode( headers, TCreateNodeArgs::File(RootNodeId, "file1"))->Record; const auto nodeId1 = createNodeResponse.GetNode().GetId(); UNIT_ASSERT_VALUES_EQUAL(1, ExtractShardNo(nodeId1)); createNodeResponse = service.CreateNode( headers, TCreateNodeArgs::File(RootNodeId, "file2"))->Record; const auto nodeId2 = createNodeResponse.GetNode().GetId(); UNIT_ASSERT_VALUES_EQUAL(2, ExtractShardNo(nodeId2)); ui64 handle1 = service.CreateHandle( headers, fsConfig.FsId, nodeId1, "", TCreateHandleArgs::RDWR)->Record.GetHandle(); auto data1 = GenerateValidateData(256_KB); service.WriteData(headers, fsConfig.FsId, nodeId1, handle1, 0, data1); ui64 handle2 = service.CreateHandle( headers, fsConfig.FsId, nodeId2, "", TCreateHandleArgs::RDWR)->Record.GetHandle(); auto data2 = GenerateValidateData(512_KB); service.WriteData(headers, fsConfig.FsId, nodeId2, handle2, 0, data2); // triggering stats collection from shards const auto fsStat = service.StatFileStore(headers, fsConfig.FsId)->Record; const auto& fileStore = fsStat.GetFileStore(); UNIT_ASSERT_VALUES_EQUAL(fsConfig.FsId, fileStore.GetFileSystemId()); UNIT_ASSERT_VALUES_EQUAL(2'000, fileStore.GetBlocksCount()); UNIT_ASSERT_VALUES_EQUAL(4_KB, fileStore.GetBlockSize()); const auto& fileStoreStats = fsStat.GetStats(); UNIT_ASSERT_VALUES_EQUAL(2, fileStoreStats.GetUsedNodesCount()); UNIT_ASSERT_VALUES_EQUAL( 768_KB / 4_KB, fileStoreStats.GetUsedBlocksCount()); { const auto response = GetStorageStats( service, fsConfig.FsId, Max() /* cacheTTL */); const auto& stats = response.GetStats(); UNIT_ASSERT_VALUES_EQUAL( (data1.size() + data2.size()) / 4_KB, stats.GetUsedBlocksCount()); UNIT_ASSERT_VALUES_EQUAL( (data1.size() + data2.size()) / 4_KB, stats.GetMixedBlocksCount()); } service.WriteData(headers, fsConfig.FsId, nodeId2, handle2, 0, data2); // triggering stats collection from shards service.StatFileStore(headers, fsConfig.FsId); { const auto response = GetStorageStats( service, fsConfig.FsId, Max() /* cacheTTL */); const auto& stats = response.GetStats(); UNIT_ASSERT_VALUES_EQUAL( (data1.size() + data2.size()) / 4_KB, stats.GetUsedBlocksCount()); UNIT_ASSERT_VALUES_EQUAL( (data1.size() + 2 * data2.size()) / 4_KB, stats.GetMixedBlocksCount()); } } SERVICE_TEST( ShouldGetMultishardedSystemTopologyWithStrictFileSystemSizeEnforcement) { config.SetStrictFileSystemSizeEnforcementEnabled(true); TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); auto headers = service.InitSession(fsConfig.FsId, "client"); // create 2 files auto createNodeResponse = service.CreateNode( headers, TCreateNodeArgs::File(RootNodeId, "file1"))->Record; const auto nodeId1 = createNodeResponse.GetNode().GetId(); UNIT_ASSERT_VALUES_EQUAL(1, ExtractShardNo(nodeId1)); createNodeResponse = service.CreateNode( headers, TCreateNodeArgs::File(RootNodeId, "file2"))->Record; const auto nodeId2 = createNodeResponse.GetNode().GetId(); UNIT_ASSERT_VALUES_EQUAL(2, ExtractShardNo(nodeId2)); ui64 handle1 = service.CreateHandle( headers, fsConfig.FsId, nodeId1, "", TCreateHandleArgs::RDWR)->Record.GetHandle(); auto data1 = GenerateValidateData(256_KB); service.WriteData(headers, fsConfig.FsId, nodeId1, handle1, 0, data1); ui64 handle2 = service.CreateHandle( headers, fsConfig.FsId, nodeId2, "", TCreateHandleArgs::RDWR)->Record.GetHandle(); auto data2 = GenerateValidateData(512_KB); service.WriteData(headers, fsConfig.FsId, nodeId2, handle2, 0, data2); auto testFileSystemTopology = [&] (const ui32 shardNo, const TString& shardId, const TString& mainFileSystemId) { const auto response = GetFileSystemTopology(service, shardId); const auto& shardIds = response.GetShardFileSystemIds(); UNIT_ASSERT_EQUAL(shardNo, response.GetShardNo()); UNIT_ASSERT_EQUAL( mainFileSystemId, response.GetMainFileSystemId()); UNIT_ASSERT_EQUAL(2, shardIds.size()); UNIT_ASSERT_EQUAL(fsConfig.Shard1Id, shardIds[0]); UNIT_ASSERT_EQUAL(fsConfig.Shard2Id, shardIds[1]); UNIT_ASSERT(response.GetStrictFileSystemSizeEnforcementEnabled()); }; testFileSystemTopology(0, fsConfig.FsId, fsConfig.FsId); testFileSystemTopology(1, fsConfig.Shard1Id, fsConfig.FsId); testFileSystemTopology(2, fsConfig.Shard2Id, fsConfig.FsId); } SERVICE_TEST(ShouldGetStorageStatsInDifferentModes) { config.SetStrictFileSystemSizeEnforcementEnabled(true); TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); auto headers = service.InitSession(fsConfig.FsId, "client"); // create 2 files auto createNodeResponse = service.CreateNode( headers, TCreateNodeArgs::File(RootNodeId, "file1"))->Record; const auto nodeId1 = createNodeResponse.GetNode().GetId(); UNIT_ASSERT_VALUES_EQUAL(1, ExtractShardNo(nodeId1)); createNodeResponse = service.CreateNode( headers, TCreateNodeArgs::File(RootNodeId, "file2"))->Record; const auto nodeId2 = createNodeResponse.GetNode().GetId(); UNIT_ASSERT_VALUES_EQUAL(2, ExtractShardNo(nodeId2)); ui64 handle1 = service.CreateHandle( headers, fsConfig.FsId, nodeId1, "", TCreateHandleArgs::RDWR)->Record.GetHandle(); auto data1 = GenerateValidateData(256_KB); service.WriteData(headers, fsConfig.FsId, nodeId1, handle1, 0, data1); ui64 handle2 = service.CreateHandle( headers, fsConfig.FsId, nodeId2, "", TCreateHandleArgs::RDWR)->Record.GetHandle(); auto data2 = GenerateValidateData(512_KB); service.WriteData(headers, fsConfig.FsId, nodeId2, handle2, 0, data2); { const auto response = GetStorageStats(service, fsConfig.FsId); const auto& stats = response.GetStats(); // In default mode a request to the main filesystem fetches // statistics form shards UNIT_ASSERT_VALUES_EQUAL( (data1.size() + data2.size()) / 4_KB, stats.GetUsedBlocksCount()); } { const auto response = GetStorageStats( service, fsConfig.FsId, 0, /* cacheTTL */ NProtoPrivate::STATS_REQUEST_MODE_GET_ONLY_SELF); const auto& stats = response.GetStats(); // No blocks are used by the main filesystem itself, all blocks are // used by shards. UNIT_ASSERT_VALUES_EQUAL(0, stats.GetUsedBlocksCount()); } // Shards in default mode report only their own statistics. { const auto response = GetStorageStats(service, fsConfig.Shard1Id); const auto& stats = response.GetStats(); UNIT_ASSERT_VALUES_EQUAL( data1.size() / 4_KB, stats.GetUsedBlocksCount()); } { const auto response = GetStorageStats( service, fsConfig.Shard2Id); const auto& stats = response.GetStats(); UNIT_ASSERT_VALUES_EQUAL( data2.size() / 4_KB, stats.GetUsedBlocksCount()); } // Shards in FORCE_FETCH_SHARDS mode fetch statistics from all the // shards. { const auto response = GetStorageStats( service, fsConfig.Shard1Id, 0 /* cacheTTL */, NProtoPrivate::STATS_REQUEST_MODE_FORCE_FETCH_SHARDS); const auto& stats = response.GetStats(); UNIT_ASSERT_VALUES_EQUAL( (data1.size() + data2.size()) / 4_KB, stats.GetUsedBlocksCount()); } { const auto response = GetStorageStats( service, fsConfig.Shard2Id, 0 /* cacheTTL */, NProtoPrivate::STATS_REQUEST_MODE_FORCE_FETCH_SHARDS); const auto& stats = response.GetStats(); UNIT_ASSERT_VALUES_EQUAL( (data1.size() + data2.size()) / 4_KB, stats.GetUsedBlocksCount()); } } SERVICE_TEST( ShouldEnableStrictFileSystemSizeEnforcementAndDirectoryCreationInShards) { // Create file system with two shards 1000 * 4 * 1024 bytes each config.SetStrictFileSystemSizeEnforcementEnabled(false); TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); auto headers = service.InitSession(fsConfig.FsId, "client"); const ui64 blockSize = 4_KB; const ui64 fsSize = fsConfig.ShardIds().size() * fsConfig.ShardBlockCount * blockSize; const ui64 filesCount = 8; const ui64 fileSize = fsSize / (filesCount * 2); auto createFile = [&](const TString& fileName) -> auto { auto createNodeResponse = service .CreateNode( headers, TCreateNodeArgs::File(RootNodeId, fileName)) ->Record; const auto nodeId = createNodeResponse.GetNode().GetId(); const ui64 handle = service .CreateHandle( headers, fsConfig.FsId, nodeId, "", TCreateHandleArgs::RDWR) ->Record.GetHandle(); return std::pair(nodeId, handle); }; // Write some data to some files for (ui64 i = 0; i < filesCount; ++i) { const auto file = createFile(TStringBuilder() << "file" << i); service.WriteData( headers, fsConfig.FsId, file.first, file.second, 0, GenerateValidateData(fileSize, i)); } { // check topology before turning strict mode on const auto mainFsTopology = GetFileSystemTopology(service, fsConfig.FsId); UNIT_ASSERT( !mainFsTopology.GetStrictFileSystemSizeEnforcementEnabled()); UNIT_ASSERT_EQUAL(2, mainFsTopology.ShardFileSystemIdsSize()); const auto shard1Topology = GetFileSystemTopology(service, fsConfig.Shard1Id); UNIT_ASSERT( !shard1Topology.GetStrictFileSystemSizeEnforcementEnabled()); UNIT_ASSERT_EQUAL(0, shard1Topology.ShardFileSystemIdsSize()); const auto shard2Topology = GetFileSystemTopology(service, fsConfig.Shard2Id); UNIT_ASSERT( !shard2Topology.GetStrictFileSystemSizeEnforcementEnabled()); UNIT_ASSERT_EQUAL(0, shard2Topology.ShardFileSystemIdsSize()); } // Resize the filesystem to the same size turinng on // StrictFileSystemSizeEnforcementEnabled service.ResizeFileStore( fsConfig.FsId, fsConfig.MainFsBlockCount, false /* force */, 0 /* shardCount */, true /* enableStrictSizeMode */, true /* directoryCreationInShards */); { // check topology after turning strict mode on const auto mainFsTopology = GetFileSystemTopology(service, fsConfig.FsId); UNIT_ASSERT( mainFsTopology.GetStrictFileSystemSizeEnforcementEnabled()); UNIT_ASSERT(mainFsTopology.GetDirectoryCreationInShardsEnabled()); UNIT_ASSERT_EQUAL(2, mainFsTopology.ShardFileSystemIdsSize()); const auto shard1Topology = GetFileSystemTopology(service, fsConfig.Shard1Id); UNIT_ASSERT( shard1Topology.GetStrictFileSystemSizeEnforcementEnabled()); UNIT_ASSERT(shard1Topology.GetDirectoryCreationInShardsEnabled()); UNIT_ASSERT_EQUAL(2, shard1Topology.ShardFileSystemIdsSize()); const auto shard2Topology = GetFileSystemTopology(service, fsConfig.Shard2Id); UNIT_ASSERT( shard2Topology.GetStrictFileSystemSizeEnforcementEnabled()); UNIT_ASSERT(shard2Topology.GetDirectoryCreationInShardsEnabled()); UNIT_ASSERT_EQUAL(2, shard2Topology.ShardFileSystemIdsSize()); } // After resizing and turning on StrictFileSystemSizeEnforcement every // shard should have the same TotalBytesCount equal to that of the main // filesystem CheckShardsSize(fsConfig, service, fsConfig.MainFsBlockCount); // Downsize the filesystem by force const auto newBlocksCount = fsConfig.MainFsBlockCount / 2; service.ResizeFileStore( fsConfig.FsId, newBlocksCount, true /* force */); CheckShardsSize(fsConfig, service, newBlocksCount); // Explicitly restore session after tablet restart headers = service.InitSession( fsConfig.FsId, "client", {}, /* checkpoint */ true /* restore */); // Attempt to write one more file should fail const auto file = createFile("notEnoughSpace"); service.AssertWriteDataFailed( headers, fsConfig.FsId, file.first, file.second, 0, GenerateValidateData(fileSize, 0xBADF00D)); } SERVICE_TEST( ShouldResizeShardsInStrictWithStrictFileSystemSizeEnforcement) { // Create file system with two shards 1000 * 4 * 1024 bytes each config.SetStrictFileSystemSizeEnforcementEnabled(true); TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); // Downsize the filesystem by force to a size that less than // ShardAllocationUnit const auto newBlocksCount = config.GetShardAllocationUnit() / 2; service.ResizeFileStore( fsConfig.FsId, newBlocksCount, true /* force */); CheckShardsSize(fsConfig, service, newBlocksCount); // It is prohibited to resize an individual shard if // StrictFileSystemSizeEnforcementEnabled service.AssertResizeFileStoreFailed( fsConfig.Shard1Id, newBlocksCount * 2); } SERVICE_TEST(ShouldRetryFileSystemResize) { // Create a filsystem with two shards. config.SetStrictFileSystemSizeEnforcementEnabled(true); TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); auto headers = service.InitSession(fsConfig.FsId, "client"); { const auto response = GetStorageStats(service, fsConfig.FsId); const auto& stats = response.GetStats(); UNIT_ASSERT_EQUAL(2, stats.GetShardStats().size()); } ui32 alterEventCount = 0; env.GetRuntime().SetEventFilter( [&](auto& runtime, TAutoPtr& event) { Y_UNUSED(runtime); if (event->GetTypeRewrite() == TEvSSProxy::EvAlterFileStoreResponse) { if (alterEventCount == 0) { auto* msg = event->Get(); const_cast(msg->Error) .set_code(E_REJECTED); } alterEventCount++; } return false; }); // After we replied with error to the main filesystem resize: // 1. Main filesystem is actually resized. // 2. No shards are resized, and no configure requests were sent. const ui64 newBlocksCount = fsConfig.MainFsBlockCount * 23 / 20; service.AssertResizeFileStoreFailed(fsConfig.FsId, newBlocksCount); { const auto response = GetStorageStats(service, fsConfig.FsId); const auto& stats = response.GetStats(); UNIT_ASSERT_EQUAL(2, stats.GetShardStats().size()); UNIT_ASSERT_EQUAL(newBlocksCount, stats.GetTotalBlocksCount()); UNIT_ASSERT_EQUAL( fsConfig.MainFsBlockCount, stats.GetShardStats()[0].GetTotalBlocksCount()); UNIT_ASSERT_EQUAL( fsConfig.MainFsBlockCount, stats.GetShardStats()[1].GetTotalBlocksCount()); } // The next step retries ResizeFileStore. It should complete resize and // bring the file system into consistent state. env.GetRuntime().SetEventFilter( &TTestActorRuntimeBase::DefaultFilterFunc); service.ResizeFileStore(fsConfig.FsId, newBlocksCount); { const auto response = GetStorageStats(service, fsConfig.FsId); const auto& stats = response.GetStats(); UNIT_ASSERT_EQUAL(3, stats.GetShardStats().size()); UNIT_ASSERT_EQUAL(newBlocksCount, stats.GetTotalBlocksCount()); UNIT_ASSERT_EQUAL( newBlocksCount, stats.GetShardStats()[0].GetTotalBlocksCount()); UNIT_ASSERT_EQUAL( newBlocksCount, stats.GetShardStats()[1].GetTotalBlocksCount()); UNIT_ASSERT_EQUAL( newBlocksCount, stats.GetShardStats()[2].GetTotalBlocksCount()); } } SERVICE_TEST(ShouldSwitchOldFilesystemToStrictMode) { // This test starts the storage service with // StrictFileSystemSizeEnforcementEnabled == flase, then restarts it // with Strict... == true and switches a filesystem created with // Strict... == false to Strct... == true. const ui64 blockSize = 4_KB; config.SetAutomaticShardCreationEnabled(true); TStorageConfig storageConfig(config); TTestEnv env({}, config); ui32 nodeIdx = env.AddDynamicNode(); TServiceClient service(env.GetRuntime(), nodeIdx); const TString fsId = "test"; const ui64 fsSize = 1000; const ui64 shardsCount = 2; { TShardRequestCounter counters(env.GetRuntime(), fsId); service.CreateFileStore(fsId, fsSize); UNIT_ASSERT_VALUES_EQUAL_C( 1, counters.CreateMaxInFlight, "#" << counters.CreateRequests << ", " << counters.CreateResponses); } { TShardRequestCounter counters(env.GetRuntime(), fsId); service.ResizeFileStore(fsId, fsSize, false /* force */, shardsCount); UNIT_ASSERT_VALUES_EQUAL_C( shardsCount, counters.CreateMaxInFlight, "#" << counters.CreateRequests << ", " << counters.CreateResponses); } WaitForTabletStart(service); { const auto response = GetStorageStats(service, fsId); const auto& stats = response.GetStats(); for (const auto& stat: stats.GetShardStats()) { UNIT_ASSERT_EQUAL( storageConfig.GetAutomaticallyCreatedShardSize() / blockSize, stat.GetTotalBlocksCount()); } } config.SetStrictFileSystemSizeEnforcementEnabled(true); env.UpdateStorageConfig(config); env.CreateAndRegisterStorageService(nodeIdx); service.ResizeFileStore( fsId, fsSize, false /* force */, shardsCount, true /* enableStrictSizeMode */); WaitForTabletStart(service); { const auto response = GetStorageStats(service, fsId); const auto& stats = response.GetStats(); for (const auto& stat: stats.GetShardStats()) { UNIT_ASSERT_EQUAL(fsSize, stat.GetTotalBlocksCount()); } } } SERVICE_TEST(ShouldUseOldFileSystemInStrictMode) { const ui64 blockSize = 4_KB; config.SetAutomaticShardCreationEnabled(true); config.SetShardAllocationUnit(1_GB); config.SetAutomaticallyCreatedShardSize(1_GB); const ui64 shardBlockCount = 1_GB / blockSize; const ui64 fsSize = (1_GB / blockSize) * 3 / 2; TStorageConfig storageConfig(config); TTestEnv env({}, config); ui32 nodeIdx = env.AddDynamicNode(); TServiceClient service(env.GetRuntime(), nodeIdx); service.CreateFileStore("test", fsSize); { const auto response = GetStorageStats(service, "test"); const auto& stats = response.GetStats(); UNIT_ASSERT_EQUAL(2, stats.GetShardStats().size()); for (const auto& stat: stats.GetShardStats()) { UNIT_ASSERT_EQUAL(shardBlockCount, stat.GetTotalBlocksCount()); } } config.SetStrictFileSystemSizeEnforcementEnabled(true); env.UpdateStorageConfig(config); env.CreateAndRegisterStorageService(nodeIdx); // The filestore has Strict... == false. After resize the all shards and // a newly created shard should have the same size equal to // storageConfig.GetAutomaticallyCreatedShardSize(). const ui64 newFsSize = fsSize * 3 / 2; service.ResizeFileStore("test", newFsSize); { const auto response = GetStorageStats(service, "test"); const auto& stats = response.GetStats(); UNIT_ASSERT_EQUAL(newFsSize, stats.GetTotalBlocksCount()); UNIT_ASSERT_EQUAL(3, stats.GetShardStats().size()); for (const auto& stat: stats.GetShardStats()) { UNIT_ASSERT_EQUAL(shardBlockCount, stat.GetTotalBlocksCount()); } } } SERVICE_TEST(ShouldAggregateFileSystemMetricsInBackground) { TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); auto headers = service.InitSession(fsConfig.FsId, "client"); // creating 2 files auto createNodeResponse = service.CreateNode( headers, TCreateNodeArgs::File(RootNodeId, "file1"))->Record; const auto nodeId1 = createNodeResponse.GetNode().GetId(); UNIT_ASSERT_VALUES_EQUAL(1, ExtractShardNo(nodeId1)); createNodeResponse = service.CreateNode( headers, TCreateNodeArgs::File(RootNodeId, "file2"))->Record; const auto nodeId2 = createNodeResponse.GetNode().GetId(); UNIT_ASSERT_VALUES_EQUAL(2, ExtractShardNo(nodeId2)); ui64 handle1 = service.CreateHandle( headers, fsConfig.FsId, nodeId1, "", TCreateHandleArgs::RDWR)->Record.GetHandle(); auto data1 = GenerateValidateData(256_KB); service.WriteData(headers, fsConfig.FsId, nodeId1, handle1, 0, data1); ui64 handle2 = service.CreateHandle( headers, fsConfig.FsId, nodeId2, "", TCreateHandleArgs::RDWR)->Record.GetHandle(); auto data2 = GenerateValidateData(512_KB); service.WriteData(headers, fsConfig.FsId, nodeId2, handle2, 0, data2); // triggering background shard stats collection env.GetRuntime().AdvanceCurrentTime(TDuration::Seconds(15)); { using TRequest = TEvIndexTabletPrivate::TEvUpdateCounters; env.GetRuntime().Send( new IEventHandle( fsInfo.MainTabletActorId, // recipient TActorId(), // sender new TRequest(), 0, // flags 0), nodeIdx); } TDispatchOptions options; options.FinalEvents = { TDispatchOptions::TFinalEventCondition( TEvIndexTabletPrivate::EvAggregateStatsCompleted)}; service.AccessRuntime().DispatchEvents(options); { const auto response = GetStorageStats( service, fsConfig.FsId, Max() /* cacheTTL */); const auto& stats = response.GetStats(); UNIT_ASSERT_VALUES_EQUAL( (data1.size() + data2.size()) / 4_KB, stats.GetUsedBlocksCount()); UNIT_ASSERT_VALUES_EQUAL( (data1.size() + data2.size()) / 4_KB, stats.GetMixedBlocksCount()); } // writing some more data to change stats service.WriteData(headers, fsConfig.FsId, nodeId2, handle2, 0, data2); // configuring one time request drop to break a single iteration of // background stats collection bool dropped = false; env.GetRuntime().SetEventFilter( [&] (auto& runtime, TAutoPtr& event) { Y_UNUSED(runtime); switch (event->GetTypeRewrite()) { case TEvIndexTablet::EvGetStorageStatsRequest: { if (!dropped && event->Recipient != fsInfo.MainTabletActorId && event->Recipient != MakeIndexTabletProxyServiceId()) { dropped = true; return true; } break; } } return false; }); // triggering background stats collection env.GetRuntime().AdvanceCurrentTime(TDuration::Seconds(15)); { using TRequest = TEvIndexTabletPrivate::TEvUpdateCounters; env.GetRuntime().Send( new IEventHandle( fsInfo.MainTabletActorId, // recipient TActorId(), // sender new TRequest(), 0, // flags 0), nodeIdx); } options.FinalEvents = { TDispatchOptions::TFinalEventCondition( TEvIndexTablet::EvGetStorageStatsRequest)}; service.AccessRuntime().DispatchEvents(options); const auto counters = env.GetCounters()->FindSubgroup("component", "service"); UNIT_ASSERT(counters); const auto counter = counters->GetCounter( "AppCriticalEvents/ShardStatsRetrievalTimeout"); UNIT_ASSERT_EQUAL(0, counter->GetAtomic()); env.GetRuntime().AdvanceCurrentTime(TDuration::Minutes(15)); // stats not updated yet { const auto response = GetStorageStats( service, fsConfig.FsId, Max() /* cacheTTL */); const auto& stats = response.GetStats(); UNIT_ASSERT_VALUES_EQUAL( (data1.size() + data2.size()) / 4_KB, stats.GetUsedBlocksCount()); UNIT_ASSERT_VALUES_EQUAL( (data1.size() + data2.size()) / 4_KB, stats.GetMixedBlocksCount()); } // triggering background stats collection once more - main tablet // should notice that previous background request timed out and // should send the request again { using TRequest = TEvIndexTabletPrivate::TEvUpdateCounters; env.GetRuntime().Send( new IEventHandle( fsInfo.MainTabletActorId, // recipient TActorId(), // sender new TRequest(), 0, // flags 0), nodeIdx); } options.FinalEvents = { TDispatchOptions::TFinalEventCondition( TEvIndexTabletPrivate::EvAggregateStatsCompleted)}; service.AccessRuntime().DispatchEvents(options); // crit event should be raised UNIT_ASSERT_EQUAL(1, counter->GetAtomic()); // stats should be up to date now { const auto response = GetStorageStats( service, fsConfig.FsId, Max() /* cacheTTL */); const auto& stats = response.GetStats(); UNIT_ASSERT_VALUES_EQUAL( (data1.size() + data2.size()) / 4_KB, stats.GetUsedBlocksCount()); UNIT_ASSERT_VALUES_EQUAL( (data1.size() + 2 * data2.size()) / 4_KB, stats.GetMixedBlocksCount()); } // smoke test for GetStorageStats.TimeSumUs metric - it should not show // unnaturally large values env.GetRegistry()->Update(env.GetRuntime().GetCurrentTime()); auto tabletCounters = env.GetRuntime().GetAppData().Counters ->FindSubgroup("counters", "filestore") ->FindSubgroup("component", "storage") ->FindSubgroup("type", "hdd"); UNIT_ASSERT_LT( tabletCounters->FindSubgroup("request", "GetStorageStats") ->GetCounter("TimeSumUs") ->GetAtomic(), 10'000'000); } SERVICE_TEST(ShouldRetryUnlinkingInShard) { TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); auto headers = service.InitSession(fsConfig.FsId, "client"); const auto createNodeResponse = service.CreateNode( headers, TCreateNodeArgs::File(RootNodeId, "file1"))->Record; const auto nodeId1 = createNodeResponse.GetNode().GetId(); UNIT_ASSERT_VALUES_EQUAL((1LU << 56U) + 2, nodeId1); TAutoPtr shardUnlinkResponse; bool intercept = true; env.GetRuntime().SetEventFilter( [&] (auto& runtime, TAutoPtr& event) { Y_UNUSED(runtime); if (event->GetTypeRewrite() == TEvService::EvUnlinkNodeRequest) { const auto* msg = event->Get(); if (intercept && msg->Record.GetFileSystemId() == fsConfig.Shard1Id) { auto response = std::make_unique< TEvService::TEvUnlinkNodeResponse>( MakeError(E_REJECTED, "error")); shardUnlinkResponse = new IEventHandle( event->Sender, event->Recipient, response.release(), 0, // flags event->Cookie); return true; } } return false; }); const ui64 requestId = 111; service.SendUnlinkNodeRequest( headers, RootNodeId, "file1", false, // unlinkDirectory requestId); ui32 iterations = 0; while (!shardUnlinkResponse && iterations++ < 100) { env.GetRuntime().DispatchEvents({}, TDuration::MilliSeconds(50)); } UNIT_ASSERT(shardUnlinkResponse); intercept = false; env.GetRuntime().Send(shardUnlinkResponse.Release(), nodeIdx); auto unlinkResponse = service.RecvUnlinkNodeResponse(); UNIT_ASSERT_VALUES_EQUAL_C( S_OK, unlinkResponse->GetError().GetCode(), unlinkResponse->GetError().GetMessage()); auto listNodesResponse = service.ListNodes( headers, fsConfig.FsId, RootNodeId)->Record; UNIT_ASSERT_VALUES_EQUAL(0, listNodesResponse.NamesSize()); UNIT_ASSERT_VALUES_EQUAL(0, listNodesResponse.NodesSize()); auto headers1 = headers; headers1.FileSystemId = fsConfig.Shard1Id; listNodesResponse = service.ListNodes( headers1, fsConfig.Shard1Id, RootNodeId)->Record; UNIT_ASSERT_VALUES_EQUAL(0, listNodesResponse.NamesSize()); UNIT_ASSERT_VALUES_EQUAL(0, listNodesResponse.NodesSize()); // checking DupCache logic - just in case service.SendUnlinkNodeRequest( headers, RootNodeId, "file1", false, // unlinkDirectory requestId); unlinkResponse = service.RecvUnlinkNodeResponse(); UNIT_ASSERT_VALUES_EQUAL_C( S_OK, unlinkResponse->GetError().GetCode(), unlinkResponse->GetError().GetMessage()); } SERVICE_TEST(ShouldRetryUnlinkingInShardUponLeaderRestart) { TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); auto headers = service.InitSession(fsConfig.FsId, "client"); const auto createNodeResponse = service.CreateNode( headers, TCreateNodeArgs::File(RootNodeId, "file1"))->Record; const auto nodeId1 = createNodeResponse.GetNode().GetId(); UNIT_ASSERT_VALUES_EQUAL((1LU << 56U) + 2, nodeId1); bool intercept = true; bool intercepted = false; env.GetRuntime().SetEventFilter( [&] (auto& runtime, TAutoPtr& event) { Y_UNUSED(runtime); if (event->GetTypeRewrite() == TEvService::EvUnlinkNodeRequest) { const auto* msg = event->Get(); if (intercept && msg->Record.GetFileSystemId() == fsConfig.Shard1Id) { intercepted = true; return true; } } return false; }); service.SendUnlinkNodeRequest(headers, RootNodeId, "file1"); ui32 iterations = 0; while (!intercepted && iterations++ < 100) { env.GetRuntime().DispatchEvents({}, TDuration::MilliSeconds(50)); } UNIT_ASSERT(intercepted); intercept = false; auto headers1 = headers; headers1.FileSystemId = fsConfig.Shard1Id; auto listNodesResponse = service.ListNodes( headers1, fsConfig.Shard1Id, RootNodeId)->Record; UNIT_ASSERT_VALUES_EQUAL(1, listNodesResponse.NamesSize()); UNIT_ASSERT_VALUES_EQUAL(1, listNodesResponse.NodesSize()); TIndexTabletClient tablet( env.GetRuntime(), nodeIdx, fsInfo.MainTabletId); tablet.RebootTablet(); auto unlinkResponse = service.RecvUnlinkNodeResponse(); UNIT_ASSERT_VALUES_EQUAL_C( E_REJECTED, unlinkResponse->GetError().GetCode(), unlinkResponse->GetError().GetMessage()); // remaking session since CreateSessionActor doesn't do it by itself // because EvWakeup never arrives because Scheduling doesn't work by // default and RegistrationObservers get reset after RebootTablet headers = service.InitSession(fsConfig.FsId, "client"); listNodesResponse = service.ListNodes( headers, fsConfig.FsId, RootNodeId)->Record; UNIT_ASSERT_VALUES_EQUAL(0, listNodesResponse.NamesSize()); UNIT_ASSERT_VALUES_EQUAL(0, listNodesResponse.NodesSize()); listNodesResponse = service.ListNodes( headers1, fsConfig.Shard1Id, RootNodeId)->Record; UNIT_ASSERT_VALUES_EQUAL(0, listNodesResponse.NamesSize()); UNIT_ASSERT_VALUES_EQUAL(0, listNodesResponse.NodesSize()); } SERVICE_TEST( ShouldUnlockNodeRefUponEnablingDirectoryCreationInShardsDuringUnlink) { config.SetDirectoryCreationInShardsEnabled(false); TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); auto headers = service.InitSession(fsConfig.FsId, "client"); const auto createNodeResponse = service.CreateNode( headers, TCreateNodeArgs::File(RootNodeId, "file1"))->Record; UNIT_ASSERT_VALUES_EQUAL( 1, ExtractShardNo(createNodeResponse.GetNode().GetId())); bool intercepted = false; auto prevFilter = env.GetRuntime().SetEventFilter( [&](auto& runtime, TAutoPtr& event) { Y_UNUSED(runtime); if (event->GetTypeRewrite() == TEvService::EvUnlinkNodeRequest) { const auto* msg = event->Get(); if (msg->Record.GetFileSystemId() == fsConfig.Shard1Id) { intercepted = true; return true; } } return false; }); service.SendUnlinkNodeRequest(headers, RootNodeId, "file1"); ui32 iterations = 0; while (!intercepted && iterations++ < 100) { env.GetRuntime().DispatchEvents({}, TDuration::MilliSeconds(50)); } UNIT_ASSERT(intercepted); env.GetRuntime().SetEventFilter(prevFilter); // The NodeRef has already been removed from the leader, but the OpLog // entry is still present because the shard request was intercepted. // Enabling directory creation in shards restarts the leader and makes // it replay the entry using the newly enabled mode. service.ResizeFileStore( fsConfig.FsId, fsConfig.MainFsBlockCount, false /* force */, 0 /* shardCount */, true /* enableStrictSizeMode */, true /* directoryCreationInShards */, true /* forceDirectoryCreationInShards */); const auto unlinkResponse = service.RecvUnlinkNodeResponse(); UNIT_ASSERT_VALUES_EQUAL_C( E_REJECTED, unlinkResponse->GetStatus(), FormatError(unlinkResponse->GetError())); WaitForTabletStart(service); headers = service.InitSession(fsConfig.FsId, "client"); const auto counters = env.GetCounters()->FindSubgroup("component", "service"); UNIT_ASSERT(counters); const auto counter = counters->GetCounter( "AppCriticalEvents/InvalidNodeRefUponCompleteUnlinkNode"); iterations = 0; while (!counter->GetAtomic() && iterations++ < 100) { env.GetRuntime().DispatchEvents({}, TDuration::MilliSeconds(50)); } UNIT_ASSERT_VALUES_EQUAL(0, counter->GetAtomic()); // The replayed unlink must release the in-memory NodeRef lock even // though the reference was removed before the mode switch. const auto createHandleResponse = service.CreateHandle( headers, fsConfig.FsId, RootNodeId, "file1", TCreateHandleArgs::CREATE); UNIT_ASSERT_VALUES_EQUAL_C( S_OK, createHandleResponse->GetStatus(), FormatError(createHandleResponse->GetError())); } SERVICE_TEST_DIR_CREATION_IN_SHARDS( ShouldReplayLegacyUnlinkNodeInShardOpLogEntry) { const bool directoryCreationInShardsEnabled = config.GetDirectoryCreationInShardsEnabled(); TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); auto headers = service.InitSession(fsConfig.FsId, "client"); const auto createNodeResponse = service.CreateNode( headers, TCreateNodeArgs::File(RootNodeId, "file1"))->Record; UNIT_ASSERT_VALUES_EQUAL( 1, ExtractShardNo(createNodeResponse.GetNode().GetId())); bool intercept = true; bool intercepted = false; env.GetRuntime().SetEventFilter( [&](auto& runtime, TAutoPtr& event) { Y_UNUSED(runtime); if (event->GetTypeRewrite() == TEvService::EvUnlinkNodeRequest) { const auto* msg = event->Get(); if (intercept && msg->Record.GetFileSystemId() == fsConfig.Shard1Id) { intercepted = true; return true; } } return false; }); service.SendUnlinkNodeRequest(headers, RootNodeId, "file1"); ui32 iterations = 0; while (!intercepted && iterations++ < 100) { env.GetRuntime().DispatchEvents({}, TDuration::MilliSeconds(50)); } UNIT_ASSERT(intercepted); TIndexTabletClient tablet( env.GetRuntime(), nodeIdx, fsInfo.MainTabletId); auto listResponse = tablet.ListOpLogEntries(); UNIT_ASSERT_VALUES_EQUAL(1, listResponse->OpLogEntries.size()); auto legacyEntry = std::move(listResponse->OpLogEntries[0]); UNIT_ASSERT(legacyEntry.HasUnlinkNodeInShardRequest()); auto* shardRequest = legacyEntry.MutableUnlinkNodeInShardRequest(); UNIT_ASSERT(shardRequest->HasShouldUnlockUponCompletion()); UNIT_ASSERT_VALUES_EQUAL( directoryCreationInShardsEnabled, shardRequest->GetShouldUnlockUponCompletion()); // Simulate an OpLog entry written by a version which did not persist // ShouldUnlockUponCompletion. shardRequest->ClearShouldUnlockUponCompletion(); UNIT_ASSERT(!shardRequest->HasShouldUnlockUponCompletion()); tablet.DeleteOpLogEntry(legacyEntry.GetEntryId()); tablet.WriteOpLogEntry(std::move(legacyEntry)); listResponse = tablet.ListOpLogEntries(); UNIT_ASSERT_VALUES_EQUAL(1, listResponse->OpLogEntries.size()); UNIT_ASSERT( !listResponse->OpLogEntries[0] .GetUnlinkNodeInShardRequest() .HasShouldUnlockUponCompletion()); intercept = false; tablet.RebootTablet(); const auto unlinkResponse = service.RecvUnlinkNodeResponse(); UNIT_ASSERT_VALUES_EQUAL_C( E_REJECTED, unlinkResponse->GetStatus(), FormatError(unlinkResponse->GetError())); tablet.ReconnectPipe(); tablet.WaitReady(); headers = service.InitSession(fsConfig.FsId, "client"); listResponse = tablet.ListOpLogEntries(); UNIT_ASSERT_VALUES_EQUAL(0, listResponse->OpLogEntries.size()); const auto createHandleResponse = service.CreateHandle( headers, fsConfig.FsId, RootNodeId, "file1", TCreateHandleArgs::CREATE); UNIT_ASSERT_VALUES_EQUAL_C( S_OK, createHandleResponse->GetStatus(), FormatError(createHandleResponse->GetError())); } SERVICE_TEST(ShouldRetryUnlinkingInShardUponLeaderRestartForRenameNode) { TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); auto headers = service.InitSession(fsConfig.FsId, "client"); const auto createNodeResponse1 = service.CreateNode( headers, TCreateNodeArgs::File(RootNodeId, "file1"))->Record; const auto nodeId1 = createNodeResponse1.GetNode().GetId(); UNIT_ASSERT_VALUES_EQUAL((1LU << 56U) + 2, nodeId1); const auto createNodeResponse2 = service.CreateNode( headers, TCreateNodeArgs::File(RootNodeId, "file2"))->Record; const auto nodeId2 = createNodeResponse2.GetNode().GetId(); UNIT_ASSERT_VALUES_EQUAL((2LU << 56U) + 2, nodeId2); bool intercept = true; bool intercepted = false; env.GetRuntime().SetEventFilter( [&] (auto& runtime, TAutoPtr& event) { Y_UNUSED(runtime); if (event->GetTypeRewrite() == TEvService::EvUnlinkNodeRequest) { const auto* msg = event->Get(); if (intercept && msg->Record.GetFileSystemId() == fsConfig.Shard2Id) { intercepted = true; return true; } } return false; }); service.SendRenameNodeRequest( headers, RootNodeId, "file1", RootNodeId, "file2", 0); ui32 iterations = 0; while (!intercepted && iterations++ < 100) { env.GetRuntime().DispatchEvents({}, TDuration::MilliSeconds(50)); } UNIT_ASSERT(intercepted); intercept = false; auto headers2 = headers; headers2.FileSystemId = fsConfig.Shard2Id; auto listNodesResponse = service.ListNodes( headers2, fsConfig.Shard2Id, RootNodeId)->Record; UNIT_ASSERT_VALUES_EQUAL(1, listNodesResponse.NamesSize()); UNIT_ASSERT_VALUES_EQUAL(1, listNodesResponse.NodesSize()); TIndexTabletClient tablet( env.GetRuntime(), nodeIdx, fsInfo.MainTabletId); tablet.RebootTablet(); auto renameResponse = service.RecvRenameNodeResponse(); UNIT_ASSERT_VALUES_EQUAL_C( E_REJECTED, renameResponse->GetError().GetCode(), renameResponse->GetError().GetMessage()); // remaking session since CreateSessionActor doesn't do it by itself // because EvWakeup never arrives because Scheduling doesn't work by // default and RegistrationObservers get reset after RebootTablet headers = service.InitSession(fsConfig.FsId, "client"); listNodesResponse = service.ListNodes( headers, fsConfig.FsId, RootNodeId)->Record; UNIT_ASSERT_VALUES_EQUAL(1, listNodesResponse.NamesSize()); UNIT_ASSERT_VALUES_EQUAL(1, listNodesResponse.NodesSize()); UNIT_ASSERT_VALUES_EQUAL("file2", listNodesResponse.GetNames(0)); UNIT_ASSERT_VALUES_EQUAL( nodeId1, listNodesResponse.GetNodes(0).GetId()); listNodesResponse = service.ListNodes( headers2, fsConfig.Shard2Id, RootNodeId)->Record; UNIT_ASSERT_VALUES_EQUAL(0, listNodesResponse.NamesSize()); UNIT_ASSERT_VALUES_EQUAL(0, listNodesResponse.NodesSize()); } SERVICE_TEST(ShouldRetryNodeCreationInShard) { TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); auto headers = service.InitSession(fsConfig.FsId, "client"); TAutoPtr shardCreateResponse; bool intercept = true; env.GetRuntime().SetEventFilter( [&] (auto& runtime, TAutoPtr& event) { Y_UNUSED(runtime); if (event->GetTypeRewrite() == TEvService::EvCreateNodeRequest) { const auto* msg = event->Get(); if (intercept && msg->Record.GetFileSystemId() == fsConfig.Shard1Id) { auto response = std::make_unique< TEvService::TEvCreateNodeResponse>( MakeError(E_REJECTED, "error")); shardCreateResponse = new IEventHandle( event->Sender, event->Recipient, response.release(), 0, // flags event->Cookie); return true; } } return false; }); const ui64 requestId = 111; service.SendCreateNodeRequest( headers, TCreateNodeArgs::File(RootNodeId, "file1"), requestId); ui32 iterations = 0; while (!shardCreateResponse && iterations++ < 100) { env.GetRuntime().DispatchEvents({}, TDuration::MilliSeconds(50)); } UNIT_ASSERT(shardCreateResponse); intercept = false; env.GetRuntime().Send(shardCreateResponse.Release(), nodeIdx); auto createResponse = service.RecvCreateNodeResponse(); UNIT_ASSERT_VALUES_EQUAL_C( S_OK, createResponse->GetError().GetCode(), createResponse->GetError().GetMessage()); const auto nodeId1 = createResponse->Record.GetNode().GetId(); UNIT_ASSERT_VALUES_EQUAL((1LU << 56U) + 2, nodeId1); auto listNodesResponse = service.ListNodes( headers, fsConfig.FsId, RootNodeId)->Record; UNIT_ASSERT_VALUES_EQUAL(1, listNodesResponse.NamesSize()); UNIT_ASSERT_VALUES_EQUAL(1, listNodesResponse.NodesSize()); UNIT_ASSERT_VALUES_EQUAL("file1", listNodesResponse.GetNames(0)); UNIT_ASSERT_VALUES_EQUAL( nodeId1, listNodesResponse.GetNodes(0).GetId()); // checking DupCache logic service.SendCreateNodeRequest( headers, TCreateNodeArgs::File(RootNodeId, "file1"), requestId); createResponse = service.RecvCreateNodeResponse(); UNIT_ASSERT_VALUES_EQUAL_C( S_OK, createResponse->GetError().GetCode(), createResponse->GetError().GetMessage()); UNIT_ASSERT_VALUES_EQUAL( nodeId1, createResponse->Record.GetNode().GetId()); } SERVICE_TEST(ShouldRetryNodeCreationInShardUponLeaderRestart) { TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); auto headers = service.InitSession(fsConfig.FsId, "client"); bool intercept = true; bool intercepted = false; env.GetRuntime().SetEventFilter( [&] (auto& runtime, TAutoPtr& event) { Y_UNUSED(runtime); if (event->GetTypeRewrite() == TEvService::EvCreateNodeRequest) { const auto* msg = event->Get(); if (intercept && msg->Record.GetFileSystemId() == fsConfig.Shard1Id) { intercepted = true; return true; } } return false; }); const ui64 requestId = 111; service.SendCreateNodeRequest( headers, TCreateNodeArgs::File(RootNodeId, "file1"), requestId); ui32 iterations = 0; while (!intercepted && iterations++ < 100) { env.GetRuntime().DispatchEvents({}, TDuration::MilliSeconds(50)); } UNIT_ASSERT(intercepted); intercept = false; // TODO listNodes in leader? auto headers1 = headers; headers1.FileSystemId = fsConfig.Shard1Id; auto listNodesResponse = service.ListNodes( headers1, fsConfig.Shard1Id, RootNodeId)->Record; UNIT_ASSERT_VALUES_EQUAL(0, listNodesResponse.NamesSize()); UNIT_ASSERT_VALUES_EQUAL(0, listNodesResponse.NodesSize()); TIndexTabletClient tablet( env.GetRuntime(), nodeIdx, fsInfo.MainTabletId); tablet.RebootTablet(); auto createResponse = service.RecvCreateNodeResponse(); UNIT_ASSERT_VALUES_EQUAL_C( E_REJECTED, createResponse->GetError().GetCode(), createResponse->GetError().GetMessage()); // remaking session since CreateSessionActor doesn't do it by itself // because EvWakeup never arrives because Scheduling doesn't work by // default and RegistrationObservers get reset after RebootTablet // restoreClientSession = true headers = service.InitSession(fsConfig.FsId, "client", {}, true); listNodesResponse = service.ListNodes( headers, fsConfig.FsId, RootNodeId)->Record; const ui64 nodeId1 = (1LU << 56U) + 2; UNIT_ASSERT_VALUES_EQUAL(1, listNodesResponse.NamesSize()); UNIT_ASSERT_VALUES_EQUAL(1, listNodesResponse.NodesSize()); UNIT_ASSERT_VALUES_EQUAL("file1", listNodesResponse.GetNames(0)); UNIT_ASSERT_VALUES_EQUAL( nodeId1, listNodesResponse.GetNodes(0).GetId()); listNodesResponse = service.ListNodes( headers1, fsConfig.Shard1Id, RootNodeId)->Record; UNIT_ASSERT_VALUES_EQUAL(1, listNodesResponse.NamesSize()); UNIT_ASSERT_VALUES_EQUAL(1, listNodesResponse.NodesSize()); // checking DupCache logic service.SendCreateNodeRequest( headers, TCreateNodeArgs::File(RootNodeId, "file1"), requestId); createResponse = service.RecvCreateNodeResponse(); UNIT_ASSERT_VALUES_EQUAL_C( S_OK, createResponse->GetError().GetCode(), createResponse->GetError().GetMessage()); UNIT_ASSERT_VALUES_EQUAL( nodeId1, createResponse->Record.GetNode().GetId()); } SERVICE_TEST(ShouldRetryNodeCreationInShardUponCreateHandle) { TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); auto headers = service.InitSession(fsConfig.FsId, "client"); TAutoPtr shardCreateResponse; bool intercept = true; env.GetRuntime().SetEventFilter( [&] (auto& runtime, TAutoPtr& event) { Y_UNUSED(runtime); if (event->GetTypeRewrite() == TEvService::EvCreateNodeRequest) { const auto* msg = event->Get(); if (intercept && msg->Record.GetFileSystemId() == fsConfig.Shard1Id) { auto response = std::make_unique< TEvService::TEvCreateNodeResponse>( MakeError(E_REJECTED, "error")); shardCreateResponse = new IEventHandle( event->Sender, event->Recipient, response.release(), 0, // flags event->Cookie); return true; } } return false; }); const ui64 requestId = 111; service.SendCreateHandleRequest( headers, fsConfig.FsId, RootNodeId, "file1", TCreateHandleArgs::CREATE, "", // shardId requestId); ui32 iterations = 0; while (!shardCreateResponse && iterations++ < 100) { env.GetRuntime().DispatchEvents({}, TDuration::MilliSeconds(50)); } UNIT_ASSERT(shardCreateResponse); intercept = false; env.GetRuntime().Send(shardCreateResponse.Release(), nodeIdx); auto createHandleResponse = service.RecvCreateHandleResponse(); UNIT_ASSERT_VALUES_EQUAL_C( S_OK, createHandleResponse->GetError().GetCode(), createHandleResponse->GetError().GetMessage()); const auto nodeId1 = createHandleResponse->Record.GetNodeAttr().GetId(); UNIT_ASSERT_VALUES_EQUAL((1LU << 56U) + 2, nodeId1); auto listNodesResponse = service.ListNodes( headers, fsConfig.FsId, RootNodeId)->Record; UNIT_ASSERT_VALUES_EQUAL(1, listNodesResponse.NamesSize()); UNIT_ASSERT_VALUES_EQUAL(1, listNodesResponse.NodesSize()); UNIT_ASSERT_VALUES_EQUAL("file1", listNodesResponse.GetNames(0)); UNIT_ASSERT_VALUES_EQUAL( nodeId1, listNodesResponse.GetNodes(0).GetId()); // checking DupCache logic service.SendCreateHandleRequest( headers, fsConfig.FsId, RootNodeId, "file1", TCreateHandleArgs::CREATE, "", // shardId requestId); createHandleResponse = service.RecvCreateHandleResponse(); UNIT_ASSERT_VALUES_EQUAL_C( S_OK, createHandleResponse->GetError().GetCode(), createHandleResponse->GetError().GetMessage()); UNIT_ASSERT_VALUES_EQUAL( nodeId1, createHandleResponse->Record.GetNodeAttr().GetId()); TTestRegistryVisitor visitor; auto registry = env.GetRegistry(); registry->Visit(TInstant::Zero(), visitor); visitor.ValidateExpectedCounters( {{{{"sensor", "CreateNodeInShardRetryCount"}, {"filesystem", "test"}}, 1}, {{{"sensor", "CreateNodeInShardRetryCount"}, {"filesystem", "test_s1"}}, 0}, {{{"sensor", "CreateNodeInShardRetryCount"}, {"filesystem", "test_s2"}}, 0}}); } SERVICE_TEST( ShouldRetryNodeCreationInShardUponCreateHandleUponLeaderRestart) { TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); auto headers = service.InitSession(fsConfig.FsId, "client"); bool intercept = true; bool intercepted = false; env.GetRuntime().SetEventFilter( [&] (auto& runtime, TAutoPtr& event) { Y_UNUSED(runtime); if (event->GetTypeRewrite() == TEvService::EvCreateNodeRequest) { const auto* msg = event->Get(); if (intercept && msg->Record.GetFileSystemId() == fsConfig.Shard1Id) { intercepted = true; return true; } } return false; }); const ui64 requestId = 111; service.SendCreateHandleRequest( headers, fsConfig.FsId, RootNodeId, "file1", TCreateHandleArgs::CREATE, "", // shardId requestId); ui32 iterations = 0; while (!intercepted && iterations++ < 100) { env.GetRuntime().DispatchEvents({}, TDuration::MilliSeconds(50)); } UNIT_ASSERT(intercepted); intercept = false; TIndexTabletClient tablet( env.GetRuntime(), nodeIdx, fsInfo.MainTabletId); tablet.RebootTablet(); auto createHandleResponse = service.RecvCreateHandleResponse(); UNIT_ASSERT_VALUES_EQUAL_C( E_REJECTED, createHandleResponse->GetError().GetCode(), createHandleResponse->GetError().GetMessage()); const auto nodeId1 = (1LU << 56U) + 2; // remaking session since CreateSessionActor doesn't do it by itself // because EvWakeup never arrives because Scheduling doesn't work by // default and RegistrationObservers get reset after RebootTablet // restoreClientSession = true headers = service.InitSession(fsConfig.FsId, "client", {}, true); auto listNodesResponse = service.ListNodes( headers, fsConfig.FsId, RootNodeId)->Record; UNIT_ASSERT_VALUES_EQUAL(1, listNodesResponse.NamesSize()); UNIT_ASSERT_VALUES_EQUAL(1, listNodesResponse.NodesSize()); UNIT_ASSERT_VALUES_EQUAL("file1", listNodesResponse.GetNames(0)); UNIT_ASSERT_VALUES_EQUAL( nodeId1, listNodesResponse.GetNodes(0).GetId()); // checking DupCache logic service.SendCreateHandleRequest( headers, fsConfig.FsId, RootNodeId, "file1", TCreateHandleArgs::CREATE, "", // shardId requestId); createHandleResponse = service.RecvCreateHandleResponse(); UNIT_ASSERT_VALUES_EQUAL_C( S_OK, createHandleResponse->GetError().GetCode(), createHandleResponse->GetError().GetMessage()); UNIT_ASSERT_VALUES_EQUAL( nodeId1, createHandleResponse->Record.GetNodeAttr().GetId()); } void DoTestShardedFileSystemConfigured( const TString& fsId, TServiceClient& service, TVector fsList) { auto listing = service.ListFileStores(); const auto& fsIds = listing->Record.GetFileStores(); TVector ids(fsIds.begin(), fsIds.end()); Sort(ids); UNIT_ASSERT_VALUES_EQUAL(fsList, ids); auto headers = service.InitSession(fsId, "client"); for (const auto& id: ids) { NProtoPrivate::TDescribeSessionsRequest request; request.SetFileSystemId(id); TString buf; google::protobuf::util::MessageToJsonString(request, &buf); auto jsonResponse = service.ExecuteAction("describesessions", buf); NProtoPrivate::TDescribeSessionsResponse response; UNIT_ASSERT(google::protobuf::util::JsonStringToMessage( jsonResponse->Record.GetOutput(), &response).ok()); const auto& sessions = response.GetSessions(); UNIT_ASSERT_VALUES_EQUAL(1, sessions.size()); UNIT_ASSERT_VALUES_EQUAL( headers.SessionId, sessions[0].GetSessionId()); UNIT_ASSERT_VALUES_EQUAL( headers.ClientId, sessions[0].GetClientId()); UNIT_ASSERT_VALUES_EQUAL("", sessions[0].GetSessionState()); } const TString sessionState = "some_state"; service.ResetSession(headers, sessionState); for (const auto& id: ids) { NProtoPrivate::TDescribeSessionsRequest request; request.SetFileSystemId(id); TString buf; google::protobuf::util::MessageToJsonString(request, &buf); auto jsonResponse = service.ExecuteAction("describesessions", buf); NProtoPrivate::TDescribeSessionsResponse response; UNIT_ASSERT(google::protobuf::util::JsonStringToMessage( jsonResponse->Record.GetOutput(), &response).ok()); const auto& sessions = response.GetSessions(); UNIT_ASSERT_VALUES_EQUAL(1, sessions.size()); UNIT_ASSERT_VALUES_EQUAL( headers.SessionId, sessions[0].GetSessionId()); UNIT_ASSERT_VALUES_EQUAL( headers.ClientId, sessions[0].GetClientId()); UNIT_ASSERT_VALUES_EQUAL( sessionState, sessions[0].GetSessionState()); } service.DestroySession(headers); for (const auto& id: ids) { NProtoPrivate::TDescribeSessionsRequest request; request.SetFileSystemId(id); TString buf; google::protobuf::util::MessageToJsonString(request, &buf); auto jsonResponse = service.ExecuteAction("describesessions", buf); NProtoPrivate::TDescribeSessionsResponse response; UNIT_ASSERT(google::protobuf::util::JsonStringToMessage( jsonResponse->Record.GetOutput(), &response).ok()); const auto& sessions = response.GetSessions(); UNIT_ASSERT_VALUES_EQUAL(0, sessions.size()); } } SERVICE_TEST(ShouldConfigureShardsAutomatically) { config.SetAutomaticShardCreationEnabled(true); config.SetShardAllocationUnit(1_GB); config.SetAutomaticallyCreatedShardSize(2_GB); TTestEnv env({}, config); ui32 nodeIdx = env.AddDynamicNode(); const TString fsId = "test"; TServiceClient service(env.GetRuntime(), nodeIdx); service.CreateFileStore(fsId, 2_GB / 4_KB); // waiting for IndexTablet start after the restart triggered by // configureshards WaitForTabletStart(service); DoTestShardedFileSystemConfigured( fsId, service, {fsId, fsId + "_s1", fsId + "_s2"}); } SERVICE_TEST(ShouldNotConfigureShardsAutomaticallyForSmallFileSystems) { config.SetAutomaticShardCreationEnabled(true); config.SetShardAllocationUnit(1_GB); config.SetAutomaticallyCreatedShardSize(2_GB); TTestEnv env({}, config); ui32 nodeIdx = env.AddDynamicNode(); const TString fsId = "test"; TServiceClient service(env.GetRuntime(), nodeIdx); service.CreateFileStore(fsId, (1_GB - 4_KB) / 4_KB); DoTestShardedFileSystemConfigured(fsId, service, {fsId}); } SERVICE_TEST(ShouldHandleErrorsDuringShardedFileSystemCreation) { config.SetAutomaticShardCreationEnabled(true); config.SetShardAllocationUnit(1_GB); config.SetAutomaticallyCreatedShardSize(2_GB); TTestEnv env({}, config); ui32 nodeIdx = env.AddDynamicNode(); const TString fsId = "test"; const auto blockCount = 2_GB / 4_KB; TServiceClient service(env.GetRuntime(), nodeIdx); TVector expected = {fsId, fsId + "_s1", fsId + "_s2"}; NProto::TError createShardError; NProto::TError configureShardError; NProto::TError configureShardsError; TAutoPtr toSend; env.GetRuntime().SetEventFilter( [&] (TTestActorRuntimeBase&, TAutoPtr& event) { switch (event->GetTypeRewrite()) { case TEvSSProxy::EvCreateFileStoreRequest: { using TRequest = TEvSSProxy::TEvCreateFileStoreRequest; using TResponse = TEvSSProxy::TEvCreateFileStoreResponse; const auto* msg = event->Get(); if (msg->Config.GetFileSystemId() != expected[1]) { break; } if (!HasError(createShardError)) { break; } auto response = std::make_unique( createShardError); toSend = new IEventHandle( event->Sender, event->Recipient, response.release(), 0, // flags event->Cookie); return true; } case TEvIndexTablet::EvConfigureAsShardRequest: { using TRequest = TEvIndexTablet::TEvConfigureAsShardRequest; using TResponse = TEvIndexTablet::TEvConfigureAsShardResponse; const auto* msg = event->Get(); if (msg->Record.GetFileSystemId() != expected[1]) { break; } if (!HasError(configureShardError)) { break; } auto response = std::make_unique( configureShardError); toSend = new IEventHandle( event->Sender, event->Recipient, response.release(), 0, // flags event->Cookie); return true; } case TEvIndexTablet::EvConfigureShardsRequest: { using TResponse = TEvIndexTablet::TEvConfigureShardsResponse; if (!HasError(configureShardsError)) { break; } auto response = std::make_unique( configureShardsError); toSend = new IEventHandle( event->Sender, event->Recipient, response.release(), 0, // flags event->Cookie); return true; } } return false; }); createShardError = MakeError(E_REJECTED, "failed to create shard"); service.SendCreateFileStoreRequest(fsId, blockCount); env.GetRuntime().DispatchEvents({}, TDuration::MilliSeconds(100)); UNIT_ASSERT(toSend); env.GetRuntime().Send(toSend, nodeIdx); { auto response = service.RecvCreateFileStoreResponse(); UNIT_ASSERT_VALUES_EQUAL( FormatError(createShardError), FormatError(response->GetError())); } createShardError = {}; configureShardError = MakeError(E_REJECTED, "failed to configure shard"); service.SendCreateFileStoreRequest(fsId, blockCount); env.GetRuntime().DispatchEvents({}, TDuration::MilliSeconds(100)); UNIT_ASSERT(toSend); env.GetRuntime().Send(toSend, nodeIdx); { auto response = service.RecvCreateFileStoreResponse(); UNIT_ASSERT_VALUES_EQUAL( FormatError(configureShardError), FormatError(response->GetError())); } configureShardError = {}; configureShardsError = MakeError(E_REJECTED, "failed to configure shards"); service.SendCreateFileStoreRequest(fsId, blockCount); env.GetRuntime().DispatchEvents({}, TDuration::MilliSeconds(100)); UNIT_ASSERT(toSend); env.GetRuntime().Send(toSend, nodeIdx); { auto response = service.RecvCreateFileStoreResponse(); UNIT_ASSERT_VALUES_EQUAL( FormatError(configureShardsError), FormatError(response->GetError())); } configureShardsError = {}; service.SendCreateFileStoreRequest(fsId, blockCount); { auto response = service.RecvCreateFileStoreResponse(); UNIT_ASSERT_VALUES_EQUAL( FormatError(MakeError(S_OK)), FormatError(response->GetError())); } // waiting for IndexTablet start after the restart triggered by // configureshards WaitForTabletStart(service); DoTestShardedFileSystemConfigured( fsId, service, {fsId, fsId + "_s1", fsId + "_s2"}); } SERVICE_TEST(ShouldDeleteShardsBeThrottled) { config.SetAutomaticShardCreationEnabled(true); config.SetShardAllocationUnit(1_GB); config.SetAutomaticallyCreatedShardSize(2_GB); config.SetMaxShardManagementRequestsInFlight(2); TTestEnv env({}, config); ui32 nodeIdx = env.AddDynamicNode(); const TString fsId1 = "test1"; const TString fsId2 = "test2"; TServiceClient service(env.GetRuntime(), nodeIdx); service.CreateFileStore(fsId1, 2_GB / 4_KB); service.CreateFileStore(fsId2, 4_GB / 4_KB); TVector expected = { fsId1, fsId1 + "_s1", fsId1 + "_s2", fsId2, fsId2 + "_s1", fsId2 + "_s2", fsId2 + "_s3", fsId2 + "_s4", }; auto listing = service.ListFileStores(); auto fsIds = listing->Record.GetFileStores(); TVector ids(fsIds.begin(), fsIds.end()); Sort(ids); UNIT_ASSERT_VALUES_EQUAL(expected, ids); service.DestroyFileStore(fsId1); expected = { fsId2, fsId2 + "_s1", fsId2 + "_s2", fsId2 + "_s3", fsId2 + "_s4", }; listing = service.ListFileStores(); fsIds = listing->Record.GetFileStores(); ids = TVector(fsIds.begin(), fsIds.end()); Sort(ids); UNIT_ASSERT_VALUES_EQUAL(expected, ids); { TShardRequestCounter counters(env.GetRuntime(), fsId2); service.DestroyFileStore(fsId2); UNIT_ASSERT_VALUES_EQUAL(5, counters.DestroyRequests); UNIT_ASSERT_VALUES_EQUAL(5, counters.DestroyResponses); UNIT_ASSERT_VALUES_EQUAL(2, counters.DestroyMaxInFlight); } expected = {}; listing = service.ListFileStores(); fsIds = listing->Record.GetFileStores(); ids = TVector(fsIds.begin(), fsIds.end()); Sort(ids); UNIT_ASSERT_VALUES_EQUAL(expected, ids); } SERVICE_TEST(ShouldDeleteShardsAutomatically) { config.SetAutomaticShardCreationEnabled(true); config.SetShardAllocationUnit(1_GB); config.SetAutomaticallyCreatedShardSize(2_GB); config.SetMaxShardManagementRequestsInFlight(0); TTestEnv env({}, config); ui32 nodeIdx = env.AddDynamicNode(); const TString fsId1 = "test1"; const TString fsId2 = "test2"; TServiceClient service(env.GetRuntime(), nodeIdx); service.CreateFileStore(fsId1, 2_GB / 4_KB); service.CreateFileStore(fsId2, 4_GB / 4_KB); TVector expected = { fsId1, fsId1 + "_s1", fsId1 + "_s2", fsId2, fsId2 + "_s1", fsId2 + "_s2", fsId2 + "_s3", fsId2 + "_s4", }; auto listing = service.ListFileStores(); auto fsIds = listing->Record.GetFileStores(); TVector ids(fsIds.begin(), fsIds.end()); Sort(ids); UNIT_ASSERT_VALUES_EQUAL(expected, ids); service.DestroyFileStore(fsId1); expected = { fsId2, fsId2 + "_s1", fsId2 + "_s2", fsId2 + "_s3", fsId2 + "_s4", }; listing = service.ListFileStores(); fsIds = listing->Record.GetFileStores(); ids = TVector(fsIds.begin(), fsIds.end()); Sort(ids); UNIT_ASSERT_VALUES_EQUAL(expected, ids); { TShardRequestCounter counters(env.GetRuntime(), fsId2); service.DestroyFileStore(fsId2); UNIT_ASSERT_VALUES_EQUAL(4, counters.DestroyMaxInFlight); } expected = {}; listing = service.ListFileStores(); fsIds = listing->Record.GetFileStores(); ids = TVector(fsIds.begin(), fsIds.end()); Sort(ids); UNIT_ASSERT_VALUES_EQUAL(expected, ids); } SERVICE_TEST(ShouldHandleErrorsDuringShardedFileSystemDestruction) { config.SetAutomaticShardCreationEnabled(true); config.SetShardAllocationUnit(1_GB); config.SetAutomaticallyCreatedShardSize(2_GB); TTestEnv env({}, config); ui32 nodeIdx = env.AddDynamicNode(); const TString fsId = "test"; TServiceClient service(env.GetRuntime(), nodeIdx); service.CreateFileStore(fsId, 2_GB / 4_KB); // waiting for IndexTablet start after the restart triggered by // configureshards WaitForTabletStart(service); TVector expected = {fsId, fsId + "_s1", fsId + "_s2"}; auto listing = service.ListFileStores(); auto fsIds = listing->Record.GetFileStores(); TVector ids(fsIds.begin(), fsIds.end()); Sort(ids); UNIT_ASSERT_VALUES_EQUAL(expected, ids); NProto::TError getTopologyError; NProto::TError destroyShardError; TAutoPtr toSend; env.GetRuntime().SetEventFilter( [&] (TTestActorRuntimeBase&, TAutoPtr& event) { switch (event->GetTypeRewrite()) { case TEvIndexTablet::EvGetFileSystemTopologyRequest: { using TResponse = TEvIndexTablet::TEvGetFileSystemTopologyResponse; if (!HasError(getTopologyError)) { break; } auto response = std::make_unique( getTopologyError); toSend = new IEventHandle( event->Sender, event->Recipient, response.release(), 0, // flags event->Cookie); return true; } case TEvSSProxy::EvDestroyFileStoreRequest: { using TRequest = TEvSSProxy::TEvDestroyFileStoreRequest; using TResponse = TEvSSProxy::TEvDestroyFileStoreResponse; const auto* msg = event->Get(); if (msg->FileSystemId != expected[1]) { break; } if (!HasError(destroyShardError)) { break; } auto response = std::make_unique( destroyShardError); toSend = new IEventHandle( event->Sender, event->Recipient, response.release(), 0, // flags event->Cookie); return true; } } return false; }); getTopologyError = MakeError(E_REJECTED, "failed to get topology"); service.SendDestroyFileStoreRequest(fsId); env.GetRuntime().DispatchEvents({}, TDuration::MilliSeconds(100)); UNIT_ASSERT(toSend); env.GetRuntime().Send(toSend, nodeIdx); { auto response = service.RecvDestroyFileStoreResponse(); UNIT_ASSERT_VALUES_EQUAL( FormatError(getTopologyError), FormatError(response->GetError())); } listing = service.ListFileStores(); fsIds = listing->Record.GetFileStores(); ids = TVector(fsIds.begin(), fsIds.end()); Sort(ids); UNIT_ASSERT_VALUES_EQUAL(expected, ids); getTopologyError = {}; destroyShardError = MakeError(E_REJECTED, "failed to destroy shard"); service.SendDestroyFileStoreRequest(fsId); env.GetRuntime().DispatchEvents({}, TDuration::MilliSeconds(100)); UNIT_ASSERT(toSend); env.GetRuntime().Send(toSend, nodeIdx); { auto response = service.RecvDestroyFileStoreResponse(); UNIT_ASSERT_VALUES_EQUAL( FormatError(destroyShardError), FormatError(response->GetError())); } expected = {fsId, fsId + "_s1"}; listing = service.ListFileStores(); fsIds = listing->Record.GetFileStores(); ids = TVector(fsIds.begin(), fsIds.end()); Sort(ids); UNIT_ASSERT_VALUES_EQUAL(expected, ids); destroyShardError = {}; service.SendDestroyFileStoreRequest(fsId); { auto response = service.RecvDestroyFileStoreResponse(); UNIT_ASSERT_VALUES_EQUAL( FormatError(MakeError(S_OK)), FormatError(response->GetError())); } expected = {}; listing = service.ListFileStores(); fsIds = listing->Record.GetFileStores(); ids = TVector(fsIds.begin(), fsIds.end()); Sort(ids); UNIT_ASSERT_VALUES_EQUAL(expected, ids); } SERVICE_TEST(ShouldAddShardsAutomaticallyUponResizeThrottled) { config.SetAutomaticShardCreationEnabled(true); config.SetShardAllocationUnit(1_GB); config.SetAutomaticallyCreatedShardSize(2_GB); config.SetMaxShardManagementRequestsInFlight(1); config.SetStrictFileSystemSizeEnforcementEnabled(true); TTestEnv env({}, config); ui32 nodeIdx = env.AddDynamicNode(); const TString fsId = "test"; TServiceClient service(env.GetRuntime(), nodeIdx); { TShardRequestCounter counters(env.GetRuntime(), fsId); service.CreateFileStore(fsId, 2_GB / 4_KB); UNIT_ASSERT_VALUES_EQUAL(3, counters.CreateRequests); UNIT_ASSERT_VALUES_EQUAL(3, counters.CreateResponses); UNIT_ASSERT_VALUES_EQUAL(1, counters.CreateMaxInFlight); UNIT_ASSERT_VALUES_EQUAL(2, counters.ConfigureRequests); UNIT_ASSERT_VALUES_EQUAL(2, counters.ConfigureResponses); UNIT_ASSERT_VALUES_EQUAL(1, counters.ConfigureMaxInFlight); UNIT_ASSERT_VALUES_EQUAL(0, counters.AlterRequests); } TVector expected = {fsId, fsId + "_s1", fsId + "_s2"}; auto listing = service.ListFileStores(); auto fsIds = listing->Record.GetFileStores(); TVector ids(fsIds.begin(), fsIds.end()); Sort(ids); UNIT_ASSERT_VALUES_EQUAL(expected, ids); { TShardRequestCounter counters(env.GetRuntime(), fsId); service.ResizeFileStore(fsId, 6_GB / 4_KB); UNIT_ASSERT_VALUES_EQUAL(6, counters.DescribeRequests); UNIT_ASSERT_VALUES_EQUAL(6, counters.DescribeResponses); UNIT_ASSERT_VALUES_EQUAL(1, counters.DescribeMaxInFlight); UNIT_ASSERT_VALUES_EQUAL(4, counters.CreateRequests); UNIT_ASSERT_VALUES_EQUAL(4, counters.CreateResponses); UNIT_ASSERT_VALUES_EQUAL(1, counters.CreateMaxInFlight); UNIT_ASSERT_VALUES_EQUAL(6, counters.ConfigureRequests); UNIT_ASSERT_VALUES_EQUAL(6, counters.ConfigureResponses); UNIT_ASSERT_VALUES_EQUAL(1, counters.ConfigureMaxInFlight); UNIT_ASSERT_VALUES_EQUAL(2, counters.AlterRequests); UNIT_ASSERT_VALUES_EQUAL(2, counters.AlterResponses); UNIT_ASSERT_VALUES_EQUAL(1, counters.AlterMaxInFlight); } expected = TVector{ fsId, fsId + "_s1", fsId + "_s2", fsId + "_s3", fsId + "_s4", fsId + "_s5", fsId + "_s6", }; listing = service.ListFileStores(); fsIds = listing->Record.GetFileStores(); ids = TVector(fsIds.begin(), fsIds.end()); Sort(ids); UNIT_ASSERT_VALUES_EQUAL(expected, ids); } SERVICE_TEST(ShouldAddShardsAutomaticallyUponResize) { config.SetAutomaticShardCreationEnabled(true); config.SetShardAllocationUnit(1_GB); config.SetAutomaticallyCreatedShardSize(2_GB); config.SetMaxShardManagementRequestsInFlight(0); TTestEnv env({}, config); ui32 nodeIdx = env.AddDynamicNode(); const TString fsId = "test"; TServiceClient service(env.GetRuntime(), nodeIdx); service.CreateFileStore(fsId, 1_GB / 4_KB); TVector expected = {fsId, fsId + "_s1"}; auto listing = service.ListFileStores(); auto fsIds = listing->Record.GetFileStores(); TVector ids(fsIds.begin(), fsIds.end()); Sort(ids); UNIT_ASSERT_VALUES_EQUAL(expected, ids); service.ResizeFileStore(fsId, 3_GB / 4_KB); expected = TVector{ fsId, fsId + "_s1", fsId + "_s2", fsId + "_s3", }; listing = service.ListFileStores(); fsIds = listing->Record.GetFileStores(); ids = TVector(fsIds.begin(), fsIds.end()); Sort(ids); UNIT_ASSERT_VALUES_EQUAL(expected, ids); service.ResizeFileStore(fsId, (4_GB - 4_KB) / 4_KB); expected = TVector{ fsId, fsId + "_s1", fsId + "_s2", fsId + "_s3", fsId + "_s4" }; listing = service.ListFileStores(); fsIds = listing->Record.GetFileStores(); ids = TVector(fsIds.begin(), fsIds.end()); Sort(ids); UNIT_ASSERT_VALUES_EQUAL(expected, ids); service.ResizeFileStore(fsId, 4_GB / 4_KB); listing = service.ListFileStores(); fsIds = listing->Record.GetFileStores(); ids = TVector(fsIds.begin(), fsIds.end()); Sort(ids); UNIT_ASSERT_VALUES_EQUAL(expected, ids); } SERVICE_TEST(ShouldHandleErrorsDuringShardedFileSystemResize) { config.SetAutomaticShardCreationEnabled(true); config.SetShardAllocationUnit(1_GB); config.SetAutomaticallyCreatedShardSize(2_GB); TTestEnv env({}, config); ui32 nodeIdx = env.AddDynamicNode(); const TString fsId = "test"; const auto newBlockCount = 4_GB / 4_KB; TServiceClient service(env.GetRuntime(), nodeIdx); TVector expected = {fsId, fsId + "_s1", fsId + "_s2"}; service.CreateFileStore(fsId, 2_GB / 4_KB); WaitForTabletStart(service); auto listing = service.ListFileStores(); auto fsIds = listing->Record.GetFileStores(); TVector ids(fsIds.begin(), fsIds.end()); Sort(ids); UNIT_ASSERT_VALUES_EQUAL(expected, ids); expected = { fsId, fsId + "_s1", fsId + "_s2", fsId + "_s3", fsId + "_s4", }; NProto::TError getTopologyError; NProto::TError createShardError; NProto::TError configureShardError; NProto::TError configureShardsError; TAutoPtr toSend; env.GetRuntime().SetEventFilter( [&] (TTestActorRuntimeBase&, TAutoPtr& event) { switch (event->GetTypeRewrite()) { case TEvIndexTablet::EvGetFileSystemTopologyRequest: { using TResponse = TEvIndexTablet::TEvGetFileSystemTopologyResponse; if (!HasError(getTopologyError)) { break; } auto response = std::make_unique( getTopologyError); toSend = new IEventHandle( event->Sender, event->Recipient, response.release(), 0, // flags event->Cookie); return true; } case TEvSSProxy::EvCreateFileStoreRequest: { using TRequest = TEvSSProxy::TEvCreateFileStoreRequest; using TResponse = TEvSSProxy::TEvCreateFileStoreResponse; const auto* msg = event->Get(); if (msg->Config.GetFileSystemId() != expected[3]) { break; } if (!HasError(createShardError)) { break; } auto response = std::make_unique( createShardError); toSend = new IEventHandle( event->Sender, event->Recipient, response.release(), 0, // flags event->Cookie); return true; } case TEvIndexTablet::EvConfigureAsShardRequest: { using TRequest = TEvIndexTablet::TEvConfigureAsShardRequest; using TResponse = TEvIndexTablet::TEvConfigureAsShardResponse; const auto* msg = event->Get(); if (msg->Record.GetFileSystemId() != expected[3]) { break; } if (!HasError(configureShardError)) { break; } auto response = std::make_unique( configureShardError); toSend = new IEventHandle( event->Sender, event->Recipient, response.release(), 0, // flags event->Cookie); return true; } case TEvIndexTablet::EvConfigureShardsRequest: { using TResponse = TEvIndexTablet::TEvConfigureShardsResponse; if (!HasError(configureShardsError)) { break; } auto response = std::make_unique( configureShardsError); toSend = new IEventHandle( event->Sender, event->Recipient, response.release(), 0, // flags event->Cookie); return true; } } return false; }); getTopologyError = MakeError(E_REJECTED, "failed to get topology"); service.SendResizeFileStoreRequest(fsId, newBlockCount); env.GetRuntime().DispatchEvents({}, TDuration::MilliSeconds(100)); UNIT_ASSERT(toSend); env.GetRuntime().Send(toSend, nodeIdx); { auto response = service.RecvResizeFileStoreResponse(); UNIT_ASSERT_VALUES_EQUAL( FormatError(getTopologyError), FormatError(response->GetError())); } getTopologyError = {}; createShardError = MakeError(E_REJECTED, "failed to create shard"); service.SendResizeFileStoreRequest(fsId, newBlockCount); env.GetRuntime().DispatchEvents({}, TDuration::MilliSeconds(100)); UNIT_ASSERT(toSend); env.GetRuntime().Send(toSend, nodeIdx); { auto response = service.RecvResizeFileStoreResponse(); UNIT_ASSERT_VALUES_EQUAL( FormatError(createShardError), FormatError(response->GetError())); } createShardError = {}; configureShardError = MakeError(E_REJECTED, "failed to configure shard"); service.SendResizeFileStoreRequest(fsId, newBlockCount); env.GetRuntime().DispatchEvents({}, TDuration::MilliSeconds(100)); UNIT_ASSERT(toSend); env.GetRuntime().Send(toSend, nodeIdx); { auto response = service.RecvResizeFileStoreResponse(); UNIT_ASSERT_VALUES_EQUAL( FormatError(configureShardError), FormatError(response->GetError())); } configureShardError = {}; configureShardsError = MakeError(E_REJECTED, "failed to configure shards"); service.SendResizeFileStoreRequest(fsId, newBlockCount); env.GetRuntime().DispatchEvents({}, TDuration::MilliSeconds(100)); UNIT_ASSERT(toSend); env.GetRuntime().Send(toSend, nodeIdx); { auto response = service.RecvResizeFileStoreResponse(); UNIT_ASSERT_VALUES_EQUAL( FormatError(configureShardsError), FormatError(response->GetError())); } configureShardsError = {}; service.SendResizeFileStoreRequest(fsId, newBlockCount); { auto response = service.RecvResizeFileStoreResponse(); UNIT_ASSERT_VALUES_EQUAL( FormatError(MakeError(S_OK)), FormatError(response->GetError())); } // waiting for IndexTablet start after the restart triggered by // configureshards WaitForTabletStart(service); DoTestShardedFileSystemConfigured(fsId, service, expected); } SERVICE_TEST(ShouldValidateShardList) { TTestEnv env({}, config); ui32 nodeIdx = env.AddDynamicNode(); const TString fsId = "test"; const auto shard1Id = fsId + "-f1"; const auto shard2Id = fsId + "-f2"; TServiceClient service(env.GetRuntime(), nodeIdx); service.CreateFileStore(fsId, 1'000); service.CreateFileStore(shard1Id, 1'000); service.CreateFileStore(shard2Id, 1'000); ui32 shardNo = 0; for (const auto& shardId: {shard1Id, shard2Id}) { NProtoPrivate::TConfigureAsShardRequest request; request.SetFileSystemId(shardId); request.SetShardNo(++shardNo); *request.AddShardFileSystemIds() = shard1Id; *request.AddShardFileSystemIds() = shard2Id; TString buf; google::protobuf::util::MessageToJsonString(request, &buf); auto jsonResponse = service.ExecuteAction("configureasshard", buf); NProtoPrivate::TConfigureAsShardResponse response; UNIT_ASSERT(google::protobuf::util::JsonStringToMessage( jsonResponse->Record.GetOutput(), &response).ok()); } { NProtoPrivate::TConfigureShardsRequest request; request.SetFileSystemId(fsId); *request.AddShardFileSystemIds() = shard1Id; *request.AddShardFileSystemIds() = shard2Id; TString buf; google::protobuf::util::MessageToJsonString(request, &buf); auto jsonResponse = service.ExecuteAction("configureshards", buf); NProtoPrivate::TConfigureShardsResponse response; UNIT_ASSERT(google::protobuf::util::JsonStringToMessage( jsonResponse->Record.GetOutput(), &response).ok()); } // waiting for IndexTablet start after the restart triggered by // configureshards WaitForTabletStart(service); { auto response = service.CreateSession(THeaders{fsId, "client", ""}); const auto& fileStore = response->Record.GetFileStore(); UNIT_ASSERT_VALUES_EQUAL(2, fileStore.ShardFileSystemIdsSize()); UNIT_ASSERT_VALUES_EQUAL( shard1Id, fileStore.GetShardFileSystemIds(0)); UNIT_ASSERT_VALUES_EQUAL( shard2Id, fileStore.GetShardFileSystemIds(1)); UNIT_ASSERT_VALUES_EQUAL(0, fileStore.GetShardNo()); } // shard order change not allowed { NProtoPrivate::TConfigureShardsRequest request; request.SetFileSystemId(fsId); *request.AddShardFileSystemIds() = shard2Id; *request.AddShardFileSystemIds() = shard1Id; TString buf; google::protobuf::util::MessageToJsonString(request, &buf); service.SendExecuteActionRequest("configureshards", buf); auto response = service.RecvExecuteActionResponse(); UNIT_ASSERT_VALUES_EQUAL_C( E_ARGUMENT, response->GetStatus(), response->GetErrorReason()); } // shard list cropping not allowed { NProtoPrivate::TConfigureShardsRequest request; request.SetFileSystemId(fsId); *request.AddShardFileSystemIds() = shard1Id; TString buf; google::protobuf::util::MessageToJsonString(request, &buf); service.SendExecuteActionRequest("configureshards", buf); auto response = service.RecvExecuteActionResponse(); UNIT_ASSERT_VALUES_EQUAL_C( E_ARGUMENT, response->GetStatus(), response->GetErrorReason()); } // force flag should override checks { NProtoPrivate::TConfigureShardsRequest request; request.SetFileSystemId(fsId); request.SetForce(true); TString buf; google::protobuf::util::MessageToJsonString(request, &buf); service.SendExecuteActionRequest("configureshards", buf); auto response = service.RecvExecuteActionResponse(); UNIT_ASSERT_VALUES_EQUAL_C( S_OK, response->GetStatus(), response->GetErrorReason()); } // waiting for IndexTablet start after the restart triggered by // configureshards WaitForTabletStart(service); { auto response = service.CreateSession(THeaders{fsId, "client", ""}); const auto& fileStore = response->Record.GetFileStore(); UNIT_ASSERT_VALUES_EQUAL(0, fileStore.ShardFileSystemIdsSize()); } // configureshards for a shard should fail { NProtoPrivate::TConfigureShardsRequest request; request.SetFileSystemId(shard1Id); *request.AddShardFileSystemIds() = shard2Id; TString buf; google::protobuf::util::MessageToJsonString(request, &buf); service.SendExecuteActionRequest("configureshards", buf); auto response = service.RecvExecuteActionResponse(); UNIT_ASSERT_VALUES_EQUAL_C( E_INVALID_STATE, response->GetStatus(), response->GetErrorReason()); // configureshards for a shard should work with force flag request.SetForce(true); buf.clear(); google::protobuf::util::MessageToJsonString(request, &buf); service.SendExecuteActionRequest("configureshards", buf); response = service.RecvExecuteActionResponse(); UNIT_ASSERT_VALUES_EQUAL_C( S_OK, response->GetStatus(), response->GetErrorReason()); } // reconfigure shard 1 { NProtoPrivate::TConfigureAsShardRequest request; request.SetFileSystemId(shard1Id); request.SetShardNo(1); request.SetForce(true); *request.AddShardFileSystemIds() = shard1Id; *request.AddShardFileSystemIds() = shard2Id; TString buf; google::protobuf::util::MessageToJsonString(request, &buf); service.SendExecuteActionRequest("configureasshard", buf); auto response = service.RecvExecuteActionResponse(); UNIT_ASSERT_VALUES_EQUAL_C( S_OK, response->GetStatus(), response->GetErrorReason()); } // shard order change not allowed { NProtoPrivate::TConfigureAsShardRequest request; request.SetFileSystemId(shard1Id); request.SetShardNo(1); *request.AddShardFileSystemIds() = shard2Id; *request.AddShardFileSystemIds() = shard1Id; TString buf; google::protobuf::util::MessageToJsonString(request, &buf); service.SendExecuteActionRequest("configureasshard", buf); auto response = service.RecvExecuteActionResponse(); UNIT_ASSERT_VALUES_EQUAL_C( E_ARGUMENT, response->GetStatus(), response->GetErrorReason()); } // shard list cropping not allowed { NProtoPrivate::TConfigureAsShardRequest request; request.SetFileSystemId(shard1Id); request.SetShardNo(1); *request.AddShardFileSystemIds() = shard1Id; TString buf; google::protobuf::util::MessageToJsonString(request, &buf); service.SendExecuteActionRequest("configureasshard", buf); auto response = service.RecvExecuteActionResponse(); UNIT_ASSERT_VALUES_EQUAL_C( E_ARGUMENT, response->GetStatus(), response->GetErrorReason()); } // force flag should override checks { NProtoPrivate::TConfigureAsShardRequest request; request.SetFileSystemId(shard1Id); request.SetShardNo(1); request.SetForce(true); TString buf; google::protobuf::util::MessageToJsonString(request, &buf); service.SendExecuteActionRequest("configureasshard", buf); auto response = service.RecvExecuteActionResponse(); UNIT_ASSERT_VALUES_EQUAL_C( S_OK, response->GetStatus(), response->GetErrorReason()); } // reconfigure shard 1 { NProtoPrivate::TConfigureAsShardRequest request; request.SetFileSystemId(shard1Id); request.SetShardNo(1); *request.AddShardFileSystemIds() = shard1Id; *request.AddShardFileSystemIds() = shard2Id; TString buf; google::protobuf::util::MessageToJsonString(request, &buf); service.SendExecuteActionRequest("configureasshard", buf); auto response = service.RecvExecuteActionResponse(); UNIT_ASSERT_VALUES_EQUAL_C( S_OK, response->GetStatus(), response->GetErrorReason()); } // file shard list should be a strict subset of the shard list { NProtoPrivate::TConfigureAsShardRequest request; request.SetFileSystemId(shard1Id); request.SetShardNo(1); *request.AddShardFileSystemIds() = shard1Id; *request.AddShardFileSystemIds() = shard2Id; *request.AddFileShardFileSystemIds() = shard1Id; *request.AddFileShardFileSystemIds() = "shard3"; TString buf; google::protobuf::util::MessageToJsonString(request, &buf); service.SendExecuteActionRequest("configureasshard", buf); auto response = service.RecvExecuteActionResponse(); UNIT_ASSERT_VALUES_EQUAL_C( E_ARGUMENT, response->GetStatus(), response->GetErrorReason()); } { NProtoPrivate::TConfigureAsShardRequest request; request.SetFileSystemId(shard1Id); request.SetShardNo(1); *request.AddShardFileSystemIds() = shard1Id; *request.AddShardFileSystemIds() = shard2Id; *request.AddFileShardFileSystemIds() = shard1Id; *request.AddFileShardFileSystemIds() = shard2Id; TString buf; google::protobuf::util::MessageToJsonString(request, &buf); service.SendExecuteActionRequest("configureasshard", buf); auto response = service.RecvExecuteActionResponse(); UNIT_ASSERT_VALUES_EQUAL_C( E_ARGUMENT, response->GetStatus(), response->GetErrorReason()); } { NProtoPrivate::TConfigureAsShardRequest request; request.SetFileSystemId(shard1Id); request.SetShardNo(1); *request.AddShardFileSystemIds() = shard1Id; *request.AddShardFileSystemIds() = shard2Id; *request.AddFileShardFileSystemIds() = shard1Id; TString buf; google::protobuf::util::MessageToJsonString(request, &buf); service.SendExecuteActionRequest("configureasshard", buf); auto response = service.RecvExecuteActionResponse(); UNIT_ASSERT_VALUES_EQUAL_C( S_OK, response->GetStatus(), response->GetErrorReason()); } } SERVICE_TEST(ShouldCreateSessionDirectlyInShard) { TTestEnv env({}, config); ui32 nodeIdx = env.AddDynamicNode(); const TString fsId = "test"; const auto shard1Id = fsId + "-f1"; const auto shard2Id = fsId + "-f2"; TServiceClient service(env.GetRuntime(), nodeIdx); service.CreateFileStore(fsId, 1'000); service.CreateFileStore(shard1Id, 1'000); service.CreateFileStore(shard2Id, 1'000); ui32 shardNo = 0; for (const auto& shardId: {shard1Id, shard2Id}) { NProtoPrivate::TConfigureAsShardRequest request; request.SetFileSystemId(shardId); request.SetShardNo(++shardNo); TString buf; google::protobuf::util::MessageToJsonString(request, &buf); auto jsonResponse = service.ExecuteAction("configureasshard", buf); NProtoPrivate::TConfigureAsShardResponse response; UNIT_ASSERT(google::protobuf::util::JsonStringToMessage( jsonResponse->Record.GetOutput(), &response).ok()); auto createSessionResponse = service.CreateSession(THeaders{shardId, "client", ""}); const auto& fileStore = createSessionResponse->Record.GetFileStore(); UNIT_ASSERT_VALUES_EQUAL(0, fileStore.ShardFileSystemIdsSize()); UNIT_ASSERT_VALUES_EQUAL(shardNo, fileStore.GetShardNo()); } } SERVICE_TEST(ShouldNotAutomaticallyAddShardsToShards) { config.SetAutomaticShardCreationEnabled(true); config.SetShardAllocationUnit(1_GB); config.SetAutomaticallyCreatedShardSize(2_GB); TTestEnv env({}, config); ui32 nodeIdx = env.AddDynamicNode(); const TString fsId = "test"; const TString shard1Id = fsId + "_s1"; TServiceClient service(env.GetRuntime(), nodeIdx); service.CreateFileStore(fsId, 1_GB / 4_KB); TVector expected = {fsId, shard1Id}; auto listing = service.ListFileStores(); auto fsIds = listing->Record.GetFileStores(); TVector ids(fsIds.begin(), fsIds.end()); Sort(ids); UNIT_ASSERT_VALUES_EQUAL(expected, ids); service.ResizeFileStore(shard1Id, 4_GB / 4_KB); // subshards shouldn't have been created for shard1 expected = TVector{fsId, shard1Id}; listing = service.ListFileStores(); fsIds = listing->Record.GetFileStores(); ids = TVector(fsIds.begin(), fsIds.end()); Sort(ids); UNIT_ASSERT_VALUES_EQUAL(expected, ids); } void CheckPendingCreateNodeInShards(bool withCreateHandle, bool withTabletReboot) { NProto::TStorageConfig config; TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); auto headers = service.InitSession(fsConfig.FsId, "client"); bool createNodeObserved = false; bool createHandleObserved = false; TAutoPtr createNodeOnShard; auto& runtime = env.GetRuntime(); runtime.SetEventFilter( [&](auto& runtime, TAutoPtr& event) { Y_UNUSED(runtime); switch (event->GetTypeRewrite()) { case TEvService::EvCreateHandleRequest: createHandleObserved = true; break; case TEvService::EvCreateNodeRequest: { const auto* msg = event->Get(); if (msg->Record.GetShardFileSystemId().empty()) { createNodeOnShard = event.Release(); return true; } createNodeObserved = true; break; } } return false; }); auto checkListNodes = [&](auto expectedNames) { auto listNodesRsp = service.ListNodes(headers, fsConfig.FsId, RootNodeId)->Record; UNIT_ASSERT_VALUES_EQUAL(expectedNames.size(), listNodesRsp.NamesSize()); UNIT_ASSERT_VALUES_EQUAL(expectedNames.size(), listNodesRsp.NodesSize()); const auto& names = listNodesRsp.GetNames(); TVector listedNames(names.begin(), names.end()); Sort(listedNames); UNIT_ASSERT_VALUES_EQUAL(expectedNames, listedNames); }; // send create node/handle, delay response in shard and make sure node is not // listed if (!withCreateHandle) { service.SendCreateNodeRequest( headers, TCreateNodeArgs::File( RootNodeId, "file1", 0, // mode fsConfig.Shard1Id)); } else { service.SendCreateHandleRequest( headers, fsConfig.FsId, RootNodeId, "file1", TCreateHandleArgs::CREATE, fsConfig.Shard1Id); } runtime.DispatchEvents(TDispatchOptions(), TDuration::Seconds(1)); if (withCreateHandle) { UNIT_ASSERT(!createNodeObserved); UNIT_ASSERT(createHandleObserved); } else { UNIT_ASSERT(createNodeObserved); UNIT_ASSERT(!createHandleObserved); } UNIT_ASSERT(createNodeOnShard); if (withTabletReboot) { TIndexTabletClient tablet( env.GetRuntime(), nodeIdx, fsInfo.MainTabletId); tablet.RebootTablet(); headers = service.InitSession(fsConfig.FsId, "client", {}, true); } checkListNodes(TVector{}); // send delayed response in shard and make sure node is listed runtime.SetEventFilter(TTestActorRuntimeBase::DefaultFilterFunc); runtime.Send(createNodeOnShard.Release(), nodeIdx); runtime.DispatchEvents(TDispatchOptions(), TDuration::Seconds(1)); checkListNodes(TVector{"file1"}); } Y_UNIT_TEST(ShouldNotListPendingCreateNodeInShards) { CheckPendingCreateNodeInShards( false, // don't create handle false // don't reboot tablet ); CheckPendingCreateNodeInShards( false, // don't create handle true // reboot tablet ); } Y_UNIT_TEST(ShouldNotListPendingCreateHandleInShards) { CheckPendingCreateNodeInShards( true, // create handle false // don't reboot tablet ); CheckPendingCreateNodeInShards( true, // create handle true // reboot tablet ); } Y_UNIT_TEST(ShouldBalanceShardsByFreeSpace) { NProto::TStorageConfig config; config.SetAutomaticShardCreationEnabled(true); config.SetShardAllocationUnit(4_MB); config.SetAutomaticallyCreatedShardSize(4_MB); config.SetShardBalancerMinFreeSpaceReserve(4_KB); config.SetShardBalancerDesiredFreeSpaceReserve(1_MB); config.SetShardBalancerPrecisionBytes(4_KB); TTestEnv env({}, config); ui32 nodeIdx = env.AddDynamicNode(); const TString fsId = "test"; TServiceClient service(env.GetRuntime(), nodeIdx); service.CreateFileStore(fsId, 20_MB / 4_KB); // waiting for IndexTablet start after the restart triggered by // configureshards WaitForTabletStart(service); TVector expected = { fsId, fsId + "_s1", fsId + "_s2", fsId + "_s3", fsId + "_s4", fsId + "_s5", }; auto listing = service.ListFileStores(); auto fsIds = listing->Record.GetFileStores(); TVector ids(fsIds.begin(), fsIds.end()); Sort(ids); UNIT_ASSERT_VALUES_EQUAL(expected, ids); auto headers = service.InitSession(fsId, "client"); // creating 5 files - we should have round-robin balancing at this // point since all shards have enough free space - more than the // DesiredFreeSpaceReserve threshold TVector handles; TVector nodes; TSet shards; for (ui32 i = 0; i < 5; ++i) { auto createHandleResponse = service.CreateHandle( headers, fsId, RootNodeId, Sprintf("file%u", i), TCreateHandleArgs::CREATE)->Record; const auto nodeId = createHandleResponse.GetNodeAttr().GetId(); shards.insert(ExtractShardNo(nodeId)); nodes.push_back(nodeId); const auto handleId = createHandleResponse.GetHandle(); handles.push_back(handleId); } // checking that we indeed created those 5 files in 5 shards UNIT_ASSERT_VALUES_EQUAL(5, shards.size()); // writing some data to 3 of the 5 files to make 3 of the 5 shards have // less than DesiredFreeSpaceReserve free space service.WriteData( headers, fsId, nodes[0], handles[0], 0, TString(3_MB + 4_KB, 'a')); service.WriteData( headers, fsId, nodes[2], handles[2], 0, TString(3_MB + 4_KB, 'a')); service.WriteData( headers, fsId, nodes[4], handles[4], 0, TString(3_MB + 4_KB, 'a')); // triggering stats collection from shards service.StatFileStore(headers, fsId); TSet emptyShards; emptyShards.insert(ExtractShardNo(nodes[1])); emptyShards.insert(ExtractShardNo(nodes[3])); handles.clear(); nodes.clear(); shards.clear(); // creating 5 new files - these files should be balanced among the // 2 remaining shards which still have more than DesiredFreeSpaceReserve // free space for (ui32 i = 0; i < 5; ++i) { auto createHandleResponse = service.CreateHandle( headers, fsId, RootNodeId, Sprintf("file%u", 5 + i), TCreateHandleArgs::CREATE)->Record; const auto nodeId = createHandleResponse.GetNodeAttr().GetId(); shards.insert(ExtractShardNo(nodeId)); nodes.push_back(nodeId); const auto handleId = createHandleResponse.GetHandle(); handles.push_back(handleId); } // checking that the new 5 files were indeed created in those 2 shards UNIT_ASSERT_VALUES_EQUAL(2, shards.size()); auto l = emptyShards.begin(); auto r = shards.begin(); while (l != emptyShards.end()) { UNIT_ASSERT_VALUES_EQUAL(*l, *r); ++l; ++r; } } Y_UNIT_TEST( ShouldIgnoreDirectoryRestrictionWhenDirectoryCreationInShardsDisabled) { constexpr ui64 blockSize = 4_KB; constexpr ui64 shardCount = 8; constexpr ui64 fsSize = 4_MB + 100_KB; const ui64 shardAllocationUnit = fsSize / shardCount; NProto::TStorageConfig config; config.SetAutomaticShardCreationEnabled(true); config.SetShardAllocationUnit(shardAllocationUnit); config.SetStrictFileSystemSizeEnforcementEnabled(true); config.SetShardBalancerPolicy(NProto::SBP_WEIGHTED_DETERMINISTIC); config.SetDirectoryCreationInShardsEnabled(false); config.SetShardsPerDirectoryCount(4); TTestEnv env({}, config); const ui32 nodeIdx = env.AddDynamicNode(); const TString fsId = "test"; TServiceClient service(env.GetRuntime(), nodeIdx); service.CreateFileStore(fsId, fsSize / blockSize); WaitForTabletStart(service); const auto headers = service.InitSession(fsId, "client"); const auto directory = service.CreateNode( headers, TCreateNodeArgs::Directory(RootNodeId, "dir")); const ui64 directoryId = directory->Record.GetNode().GetId(); UNIT_ASSERT_VALUES_EQUAL(0, ExtractShardNo(directoryId)); TSet usedShards; for (ui32 i = 0; i < shardCount; ++i) { const auto response = service.CreateHandle( headers, fsId, directoryId, TStringBuilder() << "file" << i, TCreateHandleArgs::CREATE); usedShards.insert( ExtractShardNo(response->Record.GetNodeAttr().GetId())); } const TSet expectedShards = {1, 2, 3, 4, 5, 6, 7, 8}; UNIT_ASSERT_VALUES_EQUAL(expectedShards, usedShards); } SERVICE_TEST(ShouldAddExplicitShardCountAutomaticallyUponResize) { config.SetAutomaticShardCreationEnabled(true); config.SetShardAllocationUnit(4_TB); config.SetAutomaticallyCreatedShardSize(5_TB); TTestEnv env({}, config); ui32 nodeIdx = env.AddDynamicNode(); const TString fsId = "test"; TServiceClient service(env.GetRuntime(), nodeIdx); service.CreateFileStore(fsId, 1_GB / 4_KB); TVector expected = {fsId}; auto listing = service.ListFileStores(); auto fsIds = listing->Record.GetFileStores(); TVector ids(fsIds.begin(), fsIds.end()); Sort(ids); UNIT_ASSERT_VALUES_EQUAL(expected, ids); // explicitly specifying shard count, other params are left intact service.ResizeFileStore(fsId, 1_GB / 4_KB, false, 10); expected = TVector{ fsId, fsId + "_s1", fsId + "_s10", fsId + "_s2", fsId + "_s3", fsId + "_s4", fsId + "_s5", fsId + "_s6", fsId + "_s7", fsId + "_s8", fsId + "_s9", }; listing = service.ListFileStores(); fsIds = listing->Record.GetFileStores(); ids = TVector(fsIds.begin(), fsIds.end()); Sort(ids); UNIT_ASSERT_VALUES_EQUAL(expected, ids); } SERVICE_TEST(ShouldCreateShardsUponExplicitShardCountInCreateRequest) { // AutomaticShardCreationEnabled is left disabled - shards should be // created only because of the explicit ShardCount in the request UNIT_ASSERT(!config.GetAutomaticShardCreationEnabled()); TTestEnv env({}, config); ui32 nodeIdx = env.AddDynamicNode(); TServiceClient service(env.GetRuntime(), nodeIdx); const TString fsId = "test"; auto request = service.CreateCreateFileStoreRequest(fsId, 1_GB / 4_KB); request->Record.SetShardCount(3); service.SendCreateFileStoreRequest(std::move(request)); auto response = service.RecvCreateFileStoreResponse(); UNIT_ASSERT_C( SUCCEEDED(response->GetStatus()), response->GetErrorReason()); auto listIds = [&] { auto listing = service.ListFileStores(); auto fsIds = listing->Record.GetFileStores(); TVector ids(fsIds.begin(), fsIds.end()); Sort(ids); return ids; }; TVector expected = { fsId, fsId + "_s1", fsId + "_s2", fsId + "_s3", }; UNIT_ASSERT_VALUES_EQUAL(expected, listIds()); // without an explicit ShardCount no shards should be created const TString fsId2 = "test2"; service.CreateFileStore(fsId2, 1_GB / 4_KB); expected.push_back(fsId2); Sort(expected); UNIT_ASSERT_VALUES_EQUAL(expected, listIds()); // resizing the sharded filesystem without an explicit ShardCount // should keep the shard layout intact service.ResizeFileStore(fsId, 2_GB / 4_KB); UNIT_ASSERT_VALUES_EQUAL(expected, listIds()); // resizing with a larger explicit ShardCount should add shards service.ResizeFileStore(fsId, 2_GB / 4_KB, false, 4); expected.push_back(fsId + "_s4"); Sort(expected); UNIT_ASSERT_VALUES_EQUAL(expected, listIds()); // decreasing the shard count is prohibited service.AssertResizeFileStoreFailed(fsId, 2_GB / 4_KB, false, 2); UNIT_ASSERT_VALUES_EQUAL(expected, listIds()); } SERVICE_TEST(ShouldCreateDirectoryStructureInShards) { config.SetDirectoryCreationInShardsEnabled(true); TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); auto headers = service.InitSession(fsConfig.FsId, "client"); auto createNodeResponse = service.CreateNode( headers, TCreateNodeArgs::Directory(RootNodeId, "dir1"))->Record; const auto dir1Id = createNodeResponse.GetNode().GetId(); UNIT_ASSERT_VALUES_EQUAL(1, ExtractShardNo(dir1Id)); createNodeResponse = service.CreateNode( headers, TCreateNodeArgs::Directory(dir1Id, "dir1_1"))->Record; const auto dir1_1Id = createNodeResponse.GetNode().GetId(); UNIT_ASSERT_VALUES_EQUAL(1, ExtractShardNo(dir1_1Id)); createNodeResponse = service.CreateNode( headers, TCreateNodeArgs::Directory(dir1Id, "dir1_2"))->Record; const auto dir1_2Id = createNodeResponse.GetNode().GetId(); UNIT_ASSERT_VALUES_EQUAL(2, ExtractShardNo(dir1_2Id)); auto createHandleResponse = service.CreateHandle( headers, fsConfig.FsId, dir1_1Id, "file1", TCreateHandleArgs::CREATE)->Record; const auto nodeId1 = createHandleResponse.GetNodeAttr().GetId(); UNIT_ASSERT_VALUES_EQUAL(1, ExtractShardNo(nodeId1)); const auto handle1 = createHandleResponse.GetHandle(); UNIT_ASSERT_VALUES_EQUAL_C(1, ExtractShardNo(handle1), handle1); createHandleResponse = service.CreateHandle( headers, fsConfig.FsId, dir1_2Id, "file1", TCreateHandleArgs::CREATE)->Record; const auto nodeId2 = createHandleResponse.GetNodeAttr().GetId(); UNIT_ASSERT_VALUES_EQUAL(1, ExtractShardNo(nodeId2)); const auto handle2 = createHandleResponse.GetHandle(); UNIT_ASSERT_VALUES_EQUAL_C(1, ExtractShardNo(handle2), handle2); createHandleResponse = service.CreateHandle( headers, fsConfig.FsId, dir1_2Id, "file2", TCreateHandleArgs::CREATE)->Record; const auto nodeId3 = createHandleResponse.GetNodeAttr().GetId(); UNIT_ASSERT_VALUES_EQUAL(2, ExtractShardNo(nodeId3)); const auto handle3 = createHandleResponse.GetHandle(); UNIT_ASSERT_VALUES_EQUAL_C(2, ExtractShardNo(handle3), handle3); auto data1 = GenerateValidateData(256_KB, 1); service.WriteData(headers, fsConfig.FsId, nodeId1, handle1, 0, data1); auto data2 = GenerateValidateData(1_MB, 2); service.WriteData(headers, fsConfig.FsId, nodeId2, handle2, 0, data2); auto data3 = GenerateValidateData(512_KB, 3); service.WriteData(headers, fsConfig.FsId, nodeId3, handle3, 0, data3); auto readDataResponse = service.ReadData( headers, fsConfig.FsId, nodeId1, handle1, 0, data1.size())->Record; UNIT_ASSERT_VALUES_EQUAL(data1, readDataResponse.GetBuffer()); readDataResponse = service.ReadData( headers, fsConfig.FsId, nodeId2, handle2, 0, data2.size())->Record; UNIT_ASSERT_VALUES_EQUAL(data2, readDataResponse.GetBuffer()); readDataResponse = service.ReadData( headers, fsConfig.FsId, nodeId3, handle3, 0, data3.size())->Record; UNIT_ASSERT_VALUES_EQUAL(data3, readDataResponse.GetBuffer()); service.DestroyHandle(headers, fsConfig.FsId, nodeId1, handle1); service.DestroyHandle(headers, fsConfig.FsId, nodeId2, handle2); service.DestroyHandle(headers, fsConfig.FsId, nodeId3, handle3); service.GetNodeAttr(headers, fsConfig.FsId, dir1_1Id, "file1"); service.UnlinkNode(headers, dir1_1Id, "file1", false); env.GetRegistry()->Update(env.GetRuntime().GetCurrentTime()); auto counters = env.GetRuntime().GetAppData().Counters ->FindSubgroup("counters", "filestore") ->FindSubgroup("component", "storage") ->FindSubgroup("type", "hdd"); // counters->OutputPlainText(Cerr); UNIT_ASSERT_VALUES_EQUAL( 3, counters->FindSubgroup("request", "CreateHandle") ->GetCounter("Count") ->GetAtomic()); UNIT_ASSERT_VALUES_EQUAL( 3, counters->FindSubgroup("request", "CreateHandleInShard") ->GetCounter("Count") ->GetAtomic()); UNIT_ASSERT_VALUES_EQUAL( 3, counters->FindSubgroup("request", "CreateNode") ->GetCounter("Count") ->GetAtomic()); UNIT_ASSERT_VALUES_EQUAL( 6, counters->FindSubgroup("request", "CreateNodeInShard") ->GetCounter("Count") ->GetAtomic()); UNIT_ASSERT_VALUES_EQUAL( 1, counters->FindSubgroup("request", "GetNodeAttr") ->GetCounter("Count") ->GetAtomic()); UNIT_ASSERT_VALUES_EQUAL( 1, counters->FindSubgroup("request", "GetNodeAttrInShard") ->GetCounter("Count") ->GetAtomic()); UNIT_ASSERT_VALUES_EQUAL( 1, counters->FindSubgroup("request", "UnlinkNode") ->GetCounter("Count") ->GetAtomic()); UNIT_ASSERT_VALUES_EQUAL( 1, counters->FindSubgroup("request", "UnlinkNodeInShard") ->GetCounter("Count") ->GetAtomic()); } SERVICE_TEST(ShouldCreateFilesInFileShards) { config.SetMultiTabletForwardingEnabled(true); config.SetDirectoryCreationInShardsEnabled(true); TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); // // Configuring file shard list. // { NProtoPrivate::TConfigureAsShardRequest request; request.SetFileSystemId(fsConfig.Shard1Id); request.SetMainFileSystemId(fsConfig.FsId); request.SetShardNo(1); request.AddShardFileSystemIds(fsConfig.Shard1Id); request.AddShardFileSystemIds(fsConfig.Shard2Id); request.AddFileShardFileSystemIds(fsConfig.Shard1Id); request.SetDirectoryCreationInShardsEnabled(true); TString buf; google::protobuf::util::MessageToJsonString(request, &buf); auto jsonResponse = service.ExecuteAction("configureasshard", buf); NProtoPrivate::TConfigureAsShardResponse response; UNIT_ASSERT(google::protobuf::util::JsonStringToMessage( jsonResponse->Record.GetOutput(), &response).ok()); request.SetFileSystemId(fsConfig.Shard2Id); request.SetShardNo(2); buf.clear(); google::protobuf::util::MessageToJsonString(request, &buf); jsonResponse = service.ExecuteAction("configureasshard", buf); response.Clear(); UNIT_ASSERT(google::protobuf::util::JsonStringToMessage( jsonResponse->Record.GetOutput(), &response).ok()); } { NProtoPrivate::TConfigureShardsRequest request; request.SetFileSystemId(fsConfig.FsId); request.AddShardFileSystemIds(fsConfig.Shard1Id); request.AddShardFileSystemIds(fsConfig.Shard2Id); request.AddFileShardFileSystemIds(fsConfig.Shard1Id); request.SetDirectoryCreationInShardsEnabled(true); TString buf; google::protobuf::util::MessageToJsonString(request, &buf); auto jsonResponse = service.ExecuteAction("configureshards", buf); NProtoPrivate::TConfigureShardsResponse response; UNIT_ASSERT(google::protobuf::util::JsonStringToMessage( jsonResponse->Record.GetOutput(), &response).ok()); } // // Waiting for IndexTablet start after the restart triggered by // configureshards // WaitForTabletStart(service); auto headers = service.InitSession(fsConfig.FsId, "client"); auto createNodeResponse = service.CreateNode( headers, TCreateNodeArgs::Directory(RootNodeId, "dir1"))->Record; const auto dir1Id = createNodeResponse.GetNode().GetId(); UNIT_ASSERT_VALUES_EQUAL(2, ExtractShardNo(dir1Id)); createNodeResponse = service.CreateNode( headers, TCreateNodeArgs::Directory(dir1Id, "dir1_1"))->Record; const auto dir1_1Id = createNodeResponse.GetNode().GetId(); UNIT_ASSERT_VALUES_EQUAL(2, ExtractShardNo(dir1_1Id)); createNodeResponse = service.CreateNode( headers, TCreateNodeArgs::Directory(dir1Id, "dir1_2"))->Record; const auto dir1_2Id = createNodeResponse.GetNode().GetId(); UNIT_ASSERT_VALUES_EQUAL(2, ExtractShardNo(dir1_2Id)); auto createHandleResponse = service.CreateHandle( headers, fsConfig.FsId, dir1_1Id, "file1", TCreateHandleArgs::CREATE)->Record; const auto nodeId1 = createHandleResponse.GetNodeAttr().GetId(); UNIT_ASSERT_VALUES_EQUAL(1, ExtractShardNo(nodeId1)); const auto handle1 = createHandleResponse.GetHandle(); UNIT_ASSERT_VALUES_EQUAL_C(1, ExtractShardNo(handle1), handle1); createHandleResponse = service.CreateHandle( headers, fsConfig.FsId, dir1_2Id, "file1", TCreateHandleArgs::CREATE)->Record; const auto nodeId2 = createHandleResponse.GetNodeAttr().GetId(); UNIT_ASSERT_VALUES_EQUAL(1, ExtractShardNo(nodeId2)); const auto handle2 = createHandleResponse.GetHandle(); UNIT_ASSERT_VALUES_EQUAL_C(1, ExtractShardNo(handle2), handle2); createHandleResponse = service.CreateHandle( headers, fsConfig.FsId, dir1_2Id, "file2", TCreateHandleArgs::CREATE)->Record; const auto nodeId3 = createHandleResponse.GetNodeAttr().GetId(); UNIT_ASSERT_VALUES_EQUAL(1, ExtractShardNo(nodeId3)); const auto handle3 = createHandleResponse.GetHandle(); UNIT_ASSERT_VALUES_EQUAL_C(1, ExtractShardNo(handle3), handle3); auto data1 = GenerateValidateData(256_KB, 1); service.WriteData(headers, fsConfig.FsId, nodeId1, handle1, 0, data1); auto data2 = GenerateValidateData(1_MB, 2); service.WriteData(headers, fsConfig.FsId, nodeId2, handle2, 0, data2); auto data3 = GenerateValidateData(512_KB, 3); service.WriteData(headers, fsConfig.FsId, nodeId3, handle3, 0, data3); auto readDataResponse = service.ReadData( headers, fsConfig.FsId, nodeId1, handle1, 0, data1.size())->Record; UNIT_ASSERT_VALUES_EQUAL(data1, readDataResponse.GetBuffer()); readDataResponse = service.ReadData( headers, fsConfig.FsId, nodeId2, handle2, 0, data2.size())->Record; UNIT_ASSERT_VALUES_EQUAL(data2, readDataResponse.GetBuffer()); readDataResponse = service.ReadData( headers, fsConfig.FsId, nodeId3, handle3, 0, data3.size())->Record; UNIT_ASSERT_VALUES_EQUAL(data3, readDataResponse.GetBuffer()); service.DestroyHandle(headers, fsConfig.FsId, nodeId1, handle1); service.DestroyHandle(headers, fsConfig.FsId, nodeId2, handle2); service.DestroyHandle(headers, fsConfig.FsId, nodeId3, handle3); service.GetNodeAttr(headers, fsConfig.FsId, dir1_1Id, "file1"); service.UnlinkNode(headers, dir1_1Id, "file1", false); } SERVICE_TEST(ShouldResizeFileSystemWithDirectoryCreationInShardsEnabled) { config.SetDirectoryCreationInShardsEnabled(true); config.SetAutomaticShardCreationEnabled(true); config.SetShardAllocationUnit(1_GB); config.SetStrictFileSystemSizeEnforcementEnabled(true); TTestEnv env({}, config); ui32 nodeIdx = env.AddDynamicNode(); const TString fsId = "test"; const TString shard1Id = fsId + "_s1"; const TString shard2Id = fsId + "_s2"; TServiceClient service(env.GetRuntime(), nodeIdx); const ui32 initialBlockCount = 1_GB / 4_KB; service.CreateFileStore(fsId, initialBlockCount); { const auto response = service.GetFileStoreInfo(fsId)->Record.GetFileStore(); UNIT_ASSERT_VALUES_EQUAL(fsId, response.GetFileSystemId()); UNIT_ASSERT_VALUES_EQUAL( initialBlockCount, response.GetBlocksCount()); UNIT_ASSERT_VALUES_EQUAL(1, response.GetConfigVersion()); } { const auto response = GetFileSystemTopology(service, fsId); const auto& shardIds = response.GetShardFileSystemIds(); UNIT_ASSERT_VALUES_EQUAL(1, shardIds.size()); UNIT_ASSERT_VALUES_EQUAL(shard1Id, shardIds[0]); UNIT_ASSERT(response.GetDirectoryCreationInShardsEnabled()); } { const auto response = service.GetFileStoreInfo(shard1Id)->Record.GetFileStore(); UNIT_ASSERT_VALUES_EQUAL(shard1Id, response.GetFileSystemId()); UNIT_ASSERT_VALUES_EQUAL( initialBlockCount, response.GetBlocksCount()); UNIT_ASSERT_VALUES_EQUAL(1, response.GetConfigVersion()); } { const auto response = GetFileSystemTopology(service, shard1Id); const auto& shardIds = response.GetShardFileSystemIds(); UNIT_ASSERT_VALUES_EQUAL(1, shardIds.size()); UNIT_ASSERT_VALUES_EQUAL(shard1Id, shardIds[0]); UNIT_ASSERT(response.GetDirectoryCreationInShardsEnabled()); } const ui32 newBlockCount = 2_GB / 4_KB; service.ResizeFileStore(fsId, newBlockCount); { const auto response = service.GetFileStoreInfo(fsId)->Record.GetFileStore(); UNIT_ASSERT_VALUES_EQUAL(fsId, response.GetFileSystemId()); UNIT_ASSERT_VALUES_EQUAL(newBlockCount, response.GetBlocksCount()); UNIT_ASSERT_VALUES_EQUAL(2, response.GetConfigVersion()); } { const auto response = GetFileSystemTopology(service, fsId); const auto& shardIds = response.GetShardFileSystemIds(); UNIT_ASSERT_VALUES_EQUAL(2, shardIds.size()); UNIT_ASSERT_VALUES_EQUAL(shard1Id, shardIds[0]); UNIT_ASSERT_VALUES_EQUAL(shard2Id, shardIds[1]); UNIT_ASSERT(response.GetDirectoryCreationInShardsEnabled()); } { const auto response = service.GetFileStoreInfo(shard1Id)->Record.GetFileStore(); UNIT_ASSERT_VALUES_EQUAL(shard1Id, response.GetFileSystemId()); UNIT_ASSERT_VALUES_EQUAL(newBlockCount, response.GetBlocksCount()); UNIT_ASSERT_VALUES_EQUAL(2, response.GetConfigVersion()); } { const auto response = service.GetFileStoreInfo(shard2Id)->Record.GetFileStore(); UNIT_ASSERT_VALUES_EQUAL(shard2Id, response.GetFileSystemId()); UNIT_ASSERT_VALUES_EQUAL(newBlockCount, response.GetBlocksCount()); UNIT_ASSERT_VALUES_EQUAL(1, response.GetConfigVersion()); } { const auto response = GetFileSystemTopology(service, shard1Id); const auto& shardIds = response.GetShardFileSystemIds(); UNIT_ASSERT_VALUES_EQUAL(2, shardIds.size()); UNIT_ASSERT_VALUES_EQUAL(shard1Id, shardIds[0]); UNIT_ASSERT_VALUES_EQUAL(shard2Id, shardIds[1]); UNIT_ASSERT(response.GetDirectoryCreationInShardsEnabled()); UNIT_ASSERT_VALUES_EQUAL(1, response.GetShardNo()); } { const auto response = GetFileSystemTopology(service, shard2Id); const auto& shardIds = response.GetShardFileSystemIds(); UNIT_ASSERT_VALUES_EQUAL(2, shardIds.size()); UNIT_ASSERT_VALUES_EQUAL(shard1Id, shardIds[0]); UNIT_ASSERT_VALUES_EQUAL(shard2Id, shardIds[1]); UNIT_ASSERT(response.GetDirectoryCreationInShardsEnabled()); UNIT_ASSERT_VALUES_EQUAL(2, response.GetShardNo()); } } SERVICE_TEST( ShouldNotResizeShardWithStrictModeAndDirectoryCreationInShardsEnabled) { config.SetDirectoryCreationInShardsEnabled(true); config.SetAutomaticShardCreationEnabled(true); config.SetShardAllocationUnit(1_GB); config.SetStrictFileSystemSizeEnforcementEnabled(true); TTestEnv env({}, config); ui32 nodeIdx = env.AddDynamicNode(); const TString fsId = "test"; const TString shard1Id = fsId + "_s1"; const TString shard2Id = fsId + "_s2"; TServiceClient service(env.GetRuntime(), nodeIdx); const ui32 initialBlockCount = 2_GB / 4_KB; service.CreateFileStore(fsId, initialBlockCount); { const auto response = service.GetFileStoreInfo(fsId)->Record.GetFileStore(); UNIT_ASSERT_VALUES_EQUAL(fsId, response.GetFileSystemId()); UNIT_ASSERT_VALUES_EQUAL( initialBlockCount, response.GetBlocksCount()); UNIT_ASSERT_VALUES_EQUAL(1, response.GetConfigVersion()); } { const auto response = GetFileSystemTopology(service, fsId); const auto& shardIds = response.GetShardFileSystemIds(); UNIT_ASSERT_VALUES_EQUAL(2, shardIds.size()); UNIT_ASSERT_VALUES_EQUAL(shard1Id, shardIds[0]); UNIT_ASSERT_VALUES_EQUAL(shard2Id, shardIds[1]); UNIT_ASSERT(response.GetDirectoryCreationInShardsEnabled()); } { const auto response = service.GetFileStoreInfo(shard1Id)->Record.GetFileStore(); UNIT_ASSERT_VALUES_EQUAL(shard1Id, response.GetFileSystemId()); UNIT_ASSERT_VALUES_EQUAL( initialBlockCount, response.GetBlocksCount()); UNIT_ASSERT_VALUES_EQUAL(1, response.GetConfigVersion()); } { const auto response = GetFileSystemTopology(service, shard1Id); const auto& shardIds = response.GetShardFileSystemIds(); UNIT_ASSERT_VALUES_EQUAL(2, shardIds.size()); UNIT_ASSERT_VALUES_EQUAL(shard1Id, shardIds[0]); UNIT_ASSERT_VALUES_EQUAL(shard2Id, shardIds[1]); UNIT_ASSERT(response.GetDirectoryCreationInShardsEnabled()); } const ui32 newShardBlockCount = 4_GB / 4_KB; { service.SendResizeFileStoreRequest(shard1Id, newShardBlockCount); auto response = service.RecvResizeFileStoreResponse(); UNIT_ASSERT_EQUAL_C( E_ARGUMENT, response->GetStatus(), response->GetErrorReason()); } } SERVICE_TEST(ShouldResizeShardWithDirectoryCreationInShardsEnabled) { config.SetDirectoryCreationInShardsEnabled(true); config.SetAutomaticShardCreationEnabled(true); config.SetShardAllocationUnit(1_GB); config.SetAutomaticallyCreatedShardSize(2_GB); // just being explicit about the fact that this test assumes no Strict // mode config.SetStrictFileSystemSizeEnforcementEnabled(false); TTestEnv env({}, config); ui32 nodeIdx = env.AddDynamicNode(); const TString fsId = "test"; const TString shard1Id = fsId + "_s1"; const TString shard2Id = fsId + "_s2"; TServiceClient service(env.GetRuntime(), nodeIdx); const ui32 initialBlockCount = 2_GB / 4_KB; service.CreateFileStore(fsId, initialBlockCount); { const auto response = service.GetFileStoreInfo(fsId)->Record.GetFileStore(); UNIT_ASSERT_VALUES_EQUAL(fsId, response.GetFileSystemId()); UNIT_ASSERT_VALUES_EQUAL( initialBlockCount, response.GetBlocksCount()); UNIT_ASSERT_VALUES_EQUAL(1, response.GetConfigVersion()); } { const auto response = GetFileSystemTopology(service, fsId); const auto& shardIds = response.GetShardFileSystemIds(); UNIT_ASSERT_VALUES_EQUAL(2, shardIds.size()); UNIT_ASSERT_VALUES_EQUAL(shard1Id, shardIds[0]); UNIT_ASSERT_VALUES_EQUAL(shard2Id, shardIds[1]); UNIT_ASSERT(response.GetDirectoryCreationInShardsEnabled()); } { const auto response = service.GetFileStoreInfo(shard1Id)->Record.GetFileStore(); UNIT_ASSERT_VALUES_EQUAL(shard1Id, response.GetFileSystemId()); UNIT_ASSERT_VALUES_EQUAL( initialBlockCount, response.GetBlocksCount()); UNIT_ASSERT_VALUES_EQUAL(1, response.GetConfigVersion()); } { const auto response = GetFileSystemTopology(service, shard1Id); const auto& shardIds = response.GetShardFileSystemIds(); UNIT_ASSERT_VALUES_EQUAL(2, shardIds.size()); UNIT_ASSERT_VALUES_EQUAL(shard1Id, shardIds[0]); UNIT_ASSERT_VALUES_EQUAL(shard2Id, shardIds[1]); UNIT_ASSERT(response.GetDirectoryCreationInShardsEnabled()); } const ui32 newShardBlockCount = 4_GB / 4_KB; { service.SendResizeFileStoreRequest(shard1Id, newShardBlockCount); auto response = service.RecvResizeFileStoreResponse(); UNIT_ASSERT_EQUAL_C( S_OK, response->GetStatus(), response->GetErrorReason()); } { const auto response = service.GetFileStoreInfo(fsId)->Record.GetFileStore(); UNIT_ASSERT_VALUES_EQUAL(fsId, response.GetFileSystemId()); UNIT_ASSERT_VALUES_EQUAL(initialBlockCount, response.GetBlocksCount()); UNIT_ASSERT_VALUES_EQUAL(1, response.GetConfigVersion()); } { const auto response = GetFileSystemTopology(service, fsId); const auto& shardIds = response.GetShardFileSystemIds(); UNIT_ASSERT_VALUES_EQUAL(2, shardIds.size()); UNIT_ASSERT_VALUES_EQUAL(shard1Id, shardIds[0]); UNIT_ASSERT_VALUES_EQUAL(shard2Id, shardIds[1]); UNIT_ASSERT(response.GetDirectoryCreationInShardsEnabled()); } { const auto response = service.GetFileStoreInfo(shard1Id)->Record.GetFileStore(); UNIT_ASSERT_VALUES_EQUAL(shard1Id, response.GetFileSystemId()); UNIT_ASSERT_VALUES_EQUAL(newShardBlockCount, response.GetBlocksCount()); UNIT_ASSERT_VALUES_EQUAL(2, response.GetConfigVersion()); } { const auto response = service.GetFileStoreInfo(shard2Id)->Record.GetFileStore(); UNIT_ASSERT_VALUES_EQUAL(shard2Id, response.GetFileSystemId()); UNIT_ASSERT_VALUES_EQUAL(initialBlockCount, response.GetBlocksCount()); UNIT_ASSERT_VALUES_EQUAL(1, response.GetConfigVersion()); } { const auto response = GetFileSystemTopology(service, shard1Id); const auto& shardIds = response.GetShardFileSystemIds(); UNIT_ASSERT_VALUES_EQUAL(2, shardIds.size()); UNIT_ASSERT_VALUES_EQUAL(shard1Id, shardIds[0]); UNIT_ASSERT_VALUES_EQUAL(shard2Id, shardIds[1]); UNIT_ASSERT(response.GetDirectoryCreationInShardsEnabled()); UNIT_ASSERT_VALUES_EQUAL(1, response.GetShardNo()); } { const auto response = GetFileSystemTopology(service, shard2Id); const auto& shardIds = response.GetShardFileSystemIds(); UNIT_ASSERT_VALUES_EQUAL(2, shardIds.size()); UNIT_ASSERT_VALUES_EQUAL(shard1Id, shardIds[0]); UNIT_ASSERT_VALUES_EQUAL(shard2Id, shardIds[1]); UNIT_ASSERT(response.GetDirectoryCreationInShardsEnabled()); UNIT_ASSERT_VALUES_EQUAL(2, response.GetShardNo()); } } SERVICE_TEST(ShouldCreateHardLinks) { config.SetDirectoryCreationInShardsEnabled(true); TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); auto headers = service.InitSession(fsConfig.FsId, "client"); const auto dirId = service .CreateNode( headers, TCreateNodeArgs::Directory(RootNodeId, "dir")) ->Record.GetNode() .GetId(); const auto file1Id = service.CreateNode( headers, TCreateNodeArgs::File(dirId, "file1"))->Record.GetNode().GetId(); service.CreateNode( headers, TCreateNodeArgs::Link(dirId, "link1", file1Id)); } // Regression test for the DupCache bug with cross-shard hard links. // // When a hard link's target node lives in a shard, TLinkActor (service // side) creates the shard node first and then sends a nodeRef-creation // request to the leader. The leader used to write a DupCache entry with // node id = 0 for such requests. On retry (same requestId) the zero-id // guard in HandleCreateNode would fire and return E_REJECTED ("node not // yet created in shard") even though the hard link was fully committed. // // After the fix TLinkActor forwards the shard node attrs in the leader // request (ShardNodeAttr field). The leader populates the DupCache entry // with a real node id so retry returns S_OK with the original node attrs. SERVICE_TEST(ShouldNotReturnSpuriousRejectedOnCrossShardHardLinkRetry) { config.SetDirectoryCreationInShardsEnabled(true); TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); auto headers = service.InitSession(fsConfig.FsId, "client"); const auto dirId = service .CreateNode( headers, TCreateNodeArgs::Directory(RootNodeId, "dir")) ->Record.GetNode() .GetId(); UNIT_ASSERT_VALUES_UNEQUAL(0, ExtractShardNo(dirId)); const auto fileId = service .CreateNode(headers, TCreateNodeArgs::File(dirId, "file")) ->Record.GetNode() .GetId(); UNIT_ASSERT_VALUES_UNEQUAL(0, ExtractShardNo(fileId)); // First attempt — must succeed. const ui64 requestId = 111; service.SendCreateNodeRequest( headers, TCreateNodeArgs::Link(dirId, "link", fileId), requestId); auto response = service.RecvCreateNodeResponse(); UNIT_ASSERT_VALUES_EQUAL_C( S_OK, response->GetError().GetCode(), response->GetError().GetMessage()); const auto firstNodeId = response->Record.GetNode().GetId(); UNIT_ASSERT_VALUES_UNEQUAL(0, firstNodeId); // Retry with the same requestId. Before the fix the DupCache zero-id // entry caused E_REJECTED here. After the fix the DupCache entry holds // the real node attrs so the retry returns S_OK with the same node id. service.SendCreateNodeRequest( headers, TCreateNodeArgs::Link(dirId, "link", fileId), requestId); response = service.RecvCreateNodeResponse(); UNIT_ASSERT_VALUES_EQUAL_C( S_OK, response->GetError().GetCode(), response->GetError().GetMessage()); UNIT_ASSERT_VALUES_EQUAL(firstNodeId, response->Record.GetNode().GetId()); } // See #5826 for more details SERVICE_TEST(ShouldRejectHardLinkFromShardDirToMainTabletNode) { config.SetAutomaticShardCreationEnabled(true); const TString fsId = "test"; const ui64 blockCount = 1'000; TTestEnv env({}, config); ui32 nodeIdx = env.AddDynamicNode(); TServiceClient service(env.GetRuntime(), nodeIdx); // Create a single-tablet filesystem (no shards yet; default // AutomaticallyCreatedShardSize is 5TB, so none are created). service.CreateFileStore(fsId, blockCount); auto headers = service.InitSession(fsId, "client"); // File created before shards exist — lives in the main tablet, // shardNo == 0 even after shards are added later. const auto fileId = service .CreateNode(headers, TCreateNodeArgs::File(RootNodeId, "file")) ->Record.GetNode() .GetId(); UNIT_ASSERT_VALUES_EQUAL(0, ExtractShardNo(fileId)); // Add two shards and enable DirectoryCreationInShards. service.ResizeFileStore( fsId, blockCount, false /* force */, 2 /* shardCount */, false /* enableStrictSizeMode */, true /* directoryCreationInShards */, true /* forceDirectoryCreationInShards */); WaitForTabletStart(service); headers = service.InitSession(fsId, "client"); const auto dirId = service .CreateNode( headers, TCreateNodeArgs::Directory(RootNodeId, "dir")) ->Record.GetNode() .GetId(); UNIT_ASSERT_VALUES_UNEQUAL(0, ExtractShardNo(dirId)); const auto counters = env.GetCounters()->FindSubgroup("component", "service"); UNIT_ASSERT(counters); const auto counter = counters->GetCounter( "AppCriticalEvents/HardLinkFromShardDirToMainTabletNode"); UNIT_ASSERT_VALUES_EQUAL(0, counter->GetAtomic()); // Linking a main-tablet node into a shard directory is not yet // supported and must return E_FS_NOTSUPP. auto response = service.SendAndRecvCreateNode( headers, TCreateNodeArgs::Link(dirId, "link", fileId)); UNIT_ASSERT_VALUES_EQUAL_C( E_FS_NOTSUPP, response->GetError().GetCode(), response->GetError().GetMessage()); UNIT_ASSERT_VALUES_EQUAL(1, counter->GetAtomic()); } SERVICE_TEST(ShouldReturnXDevRegardlessOfForceFlag) { config.SetAutomaticShardCreationEnabled(true); const TString fsId = "test"; const ui64 blockCount = 1'000; TTestEnv env({}, config); ui32 nodeIdx = env.AddDynamicNode(); TServiceClient service(env.GetRuntime(), nodeIdx); // // Creating a single-tablet filesystem. No shards yet - default // AutomaticallyCreatedShardSize is 5TB, so none are created. // service.CreateFileStore(fsId, blockCount); auto headers = service.InitSession(fsId, "client"); // // Dir and file are created before shards exist — they live in the main // tablet, shardNo == 0 even after shards are added later. // const auto dir1Id = service.CreateNode( headers, TCreateNodeArgs::Directory(RootNodeId, "dir1") )->Record.GetNode().GetId(); UNIT_ASSERT_VALUES_EQUAL(0, ExtractShardNo(dir1Id)); const auto file1Id = service.CreateNode( headers, TCreateNodeArgs::File(dir1Id, "file1") )->Record.GetNode().GetId(); UNIT_ASSERT_VALUES_EQUAL(0, ExtractShardNo(file1Id)); // // Adding a shard and enabling DirectoryCreationInShards. // service.ResizeFileStore( fsId, blockCount, false /* force */, 1 /* shardCount */, false /* enableStrictSizeMode */, true /* directoryCreationInShards */, false /* forceDirectoryCreationInShards */); WaitForTabletStart(service); headers = service.InitSession(fsId, "client"); // // Creating a dir directly in shard - it's not possible to create it // via the API of the main filesystem because it checks whether there // are any nodes created in the main tablet. // // But situations like the one tested here were possible before some of // the recent changes - that's why we have this test. // const auto shardId = fsId + "_s1"; auto shardHeaders = service.InitSession(shardId, "client"); const auto uuid = CreateGuidAsString(); const auto dir2Id =service.CreateNode( shardHeaders, TCreateNodeArgs::Directory(RootNodeId, uuid) )->Record.GetNode().GetId(); UNIT_ASSERT_VALUES_EQUAL(1, ExtractShardNo(dir2Id)); // // This node ref is not going to be used directly but let's still create // it to make the directory structure similar to what we would have in // real clusters. // { NProtoPrivate::TUnsafeCreateNodeRefRequest request; request.SetFileSystemId(fsId); request.SetParentId(RootNodeId); request.SetName("dir2"); request.SetShardId(shardId); request.SetShardNodeName(uuid); TString buf; google::protobuf::util::MessageToJsonString(request, &buf); service.ExecuteAction("UnsafeCreateNodeRef", buf); } // // Creating a file in the new dir. // const auto fileId2 = service.CreateNode( headers, TCreateNodeArgs::File(dir2Id, "file2") )->Record.GetNode().GetId(); UNIT_ASSERT_VALUES_EQUAL(1, ExtractShardNo(fileId2)); // // Trying to move file1 into dir2. Can't be done because dir1/file1 // NodeRef is not external. // { auto response = service.SendAndRecvRenameNode( headers, dir1Id, "file1", dir2Id, "file1", 0 /* flags */); UNIT_ASSERT_VALUES_EQUAL_C( E_FS_XDEV, response->GetError().GetCode(), response->GetError().GetMessage()); } // // Trying to move file2 into dir1. Should be possible. // { auto response = service.SendAndRecvRenameNode( headers, dir2Id, "file2", dir1Id, "file2", 0 /* flags */); UNIT_ASSERT_VALUES_EQUAL_C( S_OK, response->GetError().GetCode(), response->GetError().GetMessage()); } // // file1 can be renamed in dir1. // { auto response = service.SendAndRecvRenameNode( headers, dir1Id, "file1", dir1Id, "file1.new", 0 /* flags */); UNIT_ASSERT_VALUES_EQUAL_C( S_OK, response->GetError().GetCode(), response->GetError().GetMessage()); } // // Creating a file in the new dir. // const auto fileId3 = service.CreateNode( headers, TCreateNodeArgs::File(dir2Id, "file3") )->Record.GetNode().GetId(); UNIT_ASSERT_VALUES_EQUAL(1, ExtractShardNo(fileId3)); // // Can't move file3 on top of file1.new because dir1/file1.new is not an // external NodeRef. Technically such an op can be implemented but right // now it's not. // { auto response = service.SendAndRecvRenameNode( headers, dir2Id, "file3", dir1Id, "file1.new", 0 /* flags */); UNIT_ASSERT_VALUES_EQUAL_C( E_FS_XDEV, response->GetError().GetCode(), response->GetError().GetMessage()); } } SERVICE_TEST(ShouldListNodesAndGetNodeAttrInDirectoryInShard) { config.SetDirectoryCreationInShardsEnabled(true); TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); auto headers = service.InitSession(fsConfig.FsId, "client"); auto createNodeResponse = service.CreateNode( headers, TCreateNodeArgs::Directory(RootNodeId, "dir1"))->Record; const auto dir1Id = createNodeResponse.GetNode().GetId(); UNIT_ASSERT_VALUES_EQUAL(1, ExtractShardNo(dir1Id)); service.CreateNode( headers, TCreateNodeArgs::File(dir1Id, "file1")); service.CreateNode( headers, TCreateNodeArgs::File(dir1Id, "file2")); service.CreateNode( headers, TCreateNodeArgs::File(dir1Id, "file3")); service.CreateNode( headers, TCreateNodeArgs::File(dir1Id, "file4")); auto listNodesResponse = service.ListNodes( headers, fsConfig.FsId, dir1Id)->Record; UNIT_ASSERT_VALUES_EQUAL(4, listNodesResponse.NamesSize()); UNIT_ASSERT_VALUES_EQUAL("file1", listNodesResponse.GetNames(0)); UNIT_ASSERT_VALUES_EQUAL("file2", listNodesResponse.GetNames(1)); UNIT_ASSERT_VALUES_EQUAL("file3", listNodesResponse.GetNames(2)); UNIT_ASSERT_VALUES_EQUAL("file4", listNodesResponse.GetNames(3)); TVector> nodes(4); for (ui32 i = 0; i < 4; ++i) { nodes[i] = { listNodesResponse.GetNodes(i).GetId(), listNodesResponse.GetNames(i)}; UNIT_ASSERT_VALUES_UNEQUAL(0, nodes[i].first); } UNIT_ASSERT_VALUES_EQUAL(1, ExtractShardNo(nodes[0].first)); UNIT_ASSERT_VALUES_EQUAL(2, ExtractShardNo(nodes[1].first)); UNIT_ASSERT_VALUES_EQUAL(1, ExtractShardNo(nodes[2].first)); UNIT_ASSERT_VALUES_EQUAL(2, ExtractShardNo(nodes[3].first)); auto getAttrResponse = service.GetNodeAttr( headers, fsConfig.FsId, RootNodeId, "dir1")->Record; UNIT_ASSERT_VALUES_EQUAL(dir1Id, getAttrResponse.GetNode().GetId()); UNIT_ASSERT_VALUES_EQUAL( static_cast(NProto::E_DIRECTORY_NODE), getAttrResponse.GetNode().GetType()); getAttrResponse = service.GetNodeAttr( headers, fsConfig.FsId, dir1Id, "file1")->Record; UNIT_ASSERT_VALUES_EQUAL( nodes[0].first, getAttrResponse.GetNode().GetId()); UNIT_ASSERT_VALUES_EQUAL( static_cast(NProto::E_REGULAR_NODE), getAttrResponse.GetNode().GetType()); getAttrResponse = service.GetNodeAttr( headers, fsConfig.FsId, dir1Id, "file2")->Record; UNIT_ASSERT_VALUES_EQUAL( nodes[1].first, getAttrResponse.GetNode().GetId()); UNIT_ASSERT_VALUES_EQUAL( static_cast(NProto::E_REGULAR_NODE), getAttrResponse.GetNode().GetType()); getAttrResponse = service.GetNodeAttr( headers, fsConfig.FsId, dir1Id, "file3")->Record; UNIT_ASSERT_VALUES_EQUAL( nodes[2].first, getAttrResponse.GetNode().GetId()); UNIT_ASSERT_VALUES_EQUAL( static_cast(NProto::E_REGULAR_NODE), getAttrResponse.GetNode().GetType()); getAttrResponse = service.GetNodeAttr( headers, fsConfig.FsId, dir1Id, "file4")->Record; UNIT_ASSERT_VALUES_EQUAL( nodes[3].first, getAttrResponse.GetNode().GetId()); UNIT_ASSERT_VALUES_EQUAL( static_cast(NProto::E_REGULAR_NODE), getAttrResponse.GetNode().GetType()); } SERVICE_TEST( ShouldAggregateFileSystemMetricsInBackgroundWithDirectoriesInShards) { config.SetDirectoryCreationInShardsEnabled(true); TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); auto headers = service.InitSession(fsConfig.FsId, "client"); // creating 2 files auto createNodeResponse = service.CreateNode( headers, TCreateNodeArgs::File(RootNodeId, "file1"))->Record; const auto nodeId1 = createNodeResponse.GetNode().GetId(); UNIT_ASSERT_VALUES_EQUAL(1, ExtractShardNo(nodeId1)); createNodeResponse = service.CreateNode( headers, TCreateNodeArgs::File(RootNodeId, "file2"))->Record; const auto nodeId2 = createNodeResponse.GetNode().GetId(); UNIT_ASSERT_VALUES_EQUAL(2, ExtractShardNo(nodeId2)); ui64 handle1 = service.CreateHandle( headers, fsConfig.FsId, nodeId1, "", TCreateHandleArgs::RDWR)->Record.GetHandle(); auto data1 = GenerateValidateData(256_KB); service.WriteData(headers, fsConfig.FsId, nodeId1, handle1, 0, data1); ui64 handle2 = service.CreateHandle( headers, fsConfig.FsId, nodeId2, "", TCreateHandleArgs::RDWR)->Record.GetHandle(); auto data2 = GenerateValidateData(512_KB); service.WriteData(headers, fsConfig.FsId, nodeId2, handle2, 0, data2); // triggering background shard stats collection env.GetRuntime().AdvanceCurrentTime(TDuration::Seconds(15)); env.GetRuntime().DispatchEvents( TDispatchOptions{ .FinalEvents = {TDispatchOptions::TFinalEventCondition( TEvIndexTablet::EvGetStorageStatsResponse, 2)}}); { using TRequest = TEvIndexTabletPrivate::TEvUpdateCounters; env.GetRuntime().Send( new IEventHandle( fsInfo.Shard1ActorId, // recipient TActorId(), // sender new TRequest(), 0, // flags 0), nodeIdx); } TDispatchOptions options; options.FinalEvents = { TDispatchOptions::TFinalEventCondition( TEvIndexTabletPrivate::EvAggregateStatsCompleted)}; service.AccessRuntime().DispatchEvents(options); { auto response = GetStorageStats(service, fsConfig.Shard1Id); auto stats = response.GetStats(); UNIT_ASSERT_VALUES_EQUAL(0, stats.ShardStatsSize()); UNIT_ASSERT_VALUES_EQUAL( data1.size() / 4_KB, stats.GetUsedBlocksCount()); UNIT_ASSERT_VALUES_EQUAL( data1.size() / 4_KB, stats.GetMixedBlocksCount()); response = GetStorageStats( service, fsConfig.Shard1Id, Max() /* cacheTTL */, NProtoPrivate::STATS_REQUEST_MODE_FORCE_FETCH_SHARDS); stats = response.GetStats(); UNIT_ASSERT_VALUES_EQUAL(2, stats.ShardStatsSize()); UNIT_ASSERT_VALUES_EQUAL( 64, stats.GetShardStats(0).GetUsedBlocksCount()); UNIT_ASSERT_VALUES_EQUAL( 1000, stats.GetShardStats(0).GetTotalBlocksCount()); UNIT_ASSERT_VALUES_EQUAL( 128, stats.GetShardStats(1).GetUsedBlocksCount()); UNIT_ASSERT_VALUES_EQUAL( 1000, stats.GetShardStats(1).GetTotalBlocksCount()); UNIT_ASSERT_VALUES_EQUAL( (data1.size() + data2.size()) / 4_KB, stats.GetUsedBlocksCount()); UNIT_ASSERT_VALUES_EQUAL( (data1.size() + data2.size()) / 4_KB, stats.GetMixedBlocksCount()); } } struct TEventHelper { TTestEnv& Env; TAutoPtr Event; bool ShouldIntercept = true; void CatchEvent(ui32 eventType) { Env.GetRuntime().SetEventFilter( [&, eventType] (auto& runtime, TAutoPtr& event) { Y_UNUSED(runtime); if (ShouldIntercept && event->GetTypeRewrite() == eventType) { Event = event.Release(); return true; } return false; }); } }; SERVICE_TEST(ShouldUnlinkNodeWithDirectoriesInShards) { config.SetDirectoryCreationInShardsEnabled(true); TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); auto headers = service.InitSession(fsConfig.FsId, "client"); // / // └── dir (shard 1) // ├── file (shard 1) // ├── file-link (shard 1, same as file) // ├── file-link2 (shard 1, same as file) // └── subdir (shard 2) // └── file (shard 1) auto dirId = service .CreateNode( headers, TCreateNodeArgs::Directory(RootNodeId, "dir")) ->Record.GetNode() .GetId(); Cerr << "dirId: " << dirId << Endl; auto dirFileId = service.CreateNode( headers, TCreateNodeArgs::File(dirId, "file")) ->Record.GetNode().GetId(); Cerr << "dirFileId: " << dirFileId << Endl; auto subdirId = service .CreateNode( headers, TCreateNodeArgs::Directory(dirId, "subdir")) ->Record.GetNode() .GetId(); Cerr << "subdirId: " << subdirId << Endl; auto dirFileLinkId = service.CreateNode( headers, TCreateNodeArgs::Link( dirId, "file-link", dirFileId)) ->Record.GetNode().GetId(); Cerr << "dirFileLinkId: " << dirFileLinkId << Endl; auto dirFileLinkId2 = service.CreateNode( headers, TCreateNodeArgs::Link( dirId, "file-link2", dirFileId)) ->Record.GetNode().GetId(); Cerr << "dirFileLinkId2: " << dirFileLinkId2 << Endl; UNIT_ASSERT(ExtractShardNo(dirId) != ExtractShardNo(subdirId)); auto fileId = service .CreateNode(headers, TCreateNodeArgs::File(subdirId, "file")) ->Record.GetNode() .GetId(); Cerr << "fileId: " << fileId << Endl; auto unlinkResponse = service.SendAndRecvUnlinkNode(headers, dirId, "subdir", true); UNIT_ASSERT_VALUES_EQUAL_C( E_FS_NOTEMPTY, unlinkResponse->GetError().GetCode(), unlinkResponse->GetError().GetMessage()); service.GetNodeAttr(headers, fsConfig.FsId, dirId, "subdir"); // unlinking the file in subdir service.UnlinkNode(headers, subdirId, "file", false); // unlinking subdir should fail if UnlinkDirectory==false unlinkResponse = service.SendAndRecvUnlinkNode( headers, dirId, "subdir", false); UNIT_ASSERT_VALUES_EQUAL_C( E_FS_ISDIR, unlinkResponse->GetError().GetCode(), unlinkResponse->GetError().GetMessage()); // unlinking subdir should now succeed service.UnlinkNode(headers, dirId, "subdir", true); // creating the subdir again subdirId = service .CreateNode( headers, TCreateNodeArgs::Directory(dirId, "subdir")) ->Record.GetNode() .GetId(); // unlinking the subdir should now succeed service.UnlinkNode(headers, dirId, "subdir", true); // unlinking the file but keeping the hardlinks to check that they are // listed correctly service.UnlinkNode(headers, dirId, "file", false); // checking that listing works auto listing = service.ListNodes(headers, fsConfig.FsId, dirId)->Record; UNIT_ASSERT_VALUES_EQUAL(2, listing.NodesSize()); UNIT_ASSERT_VALUES_EQUAL(2, listing.NamesSize()); // dirFileLinkId and dirFileLinkId2 should be equal to each other and // to dirFileId but let's explicitly compare the ids in the listing to // dirFileLinkId and dirFileLinkId2 to make the test a bit more flexible // if we decide to change this internal API layer to return the ids of // the link nodes (in sharded filesystems we have separate hidden link // nodes) UNIT_ASSERT_VALUES_EQUAL(dirFileLinkId, listing.GetNodes(0).GetId()); UNIT_ASSERT_VALUES_EQUAL("file-link", listing.GetNames(0)); UNIT_ASSERT_VALUES_EQUAL(dirFileLinkId2, listing.GetNodes(1).GetId()); UNIT_ASSERT_VALUES_EQUAL("file-link2", listing.GetNames(1)); } SERVICE_TEST(ShouldHandleUnlinkedNodeUponListingWithDirectoriesInShards) { config.SetDirectoryCreationInShardsEnabled(true); TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); auto headers = service.InitSession(fsConfig.FsId, "client"); auto dirId = service.CreateNode( headers, TCreateNodeArgs::Directory(RootNodeId, "dir") )->Record.GetNode().GetId(); auto subdir1Id = service.CreateNode( headers, TCreateNodeArgs::Directory(dirId, "subdir1") )->Record.GetNode().GetId(); Y_UNUSED(subdir1Id); auto subdir2Id = service.CreateNode( headers, TCreateNodeArgs::Directory(dirId, "subdir2") )->Record.GetNode().GetId(); IEventHandlePtr unlinkNodeResponse; bool shouldIntercept = true; env.GetRuntime().SetEventFilter( [&](auto& runtime, TAutoPtr& event) { Y_UNUSED(runtime); if (event->GetTypeRewrite() == TEvService::EvUnlinkNodeResponse && shouldIntercept) { unlinkNodeResponse.reset(event.Release()); return true; } return false; }); service.SendUnlinkNodeRequest(headers, dirId, "subdir1", true); env.GetRuntime().DispatchEvents({}, TDuration::MilliSeconds(100)); UNIT_ASSERT(unlinkNodeResponse); auto listing = service.ListNodes(headers, fsConfig.FsId, dirId)->Record; UNIT_ASSERT_VALUES_EQUAL(1, listing.NodesSize()); UNIT_ASSERT_VALUES_EQUAL(1, listing.NamesSize()); UNIT_ASSERT_VALUES_EQUAL(subdir2Id, listing.GetNodes(0).GetId()); UNIT_ASSERT_VALUES_EQUAL("subdir2", listing.GetNames(0)); shouldIntercept = false; env.GetRuntime().Send(unlinkNodeResponse.release()); auto unlinkResponse = service.RecvUnlinkNodeResponse(); UNIT_ASSERT_VALUES_EQUAL_C( S_OK, unlinkResponse->GetStatus(), FormatError(unlinkResponse->GetError())); shouldIntercept = true; service.SendUnlinkNodeRequest(headers, dirId, "subdir2", true); env.GetRuntime().DispatchEvents({}, TDuration::MilliSeconds(100)); UNIT_ASSERT(unlinkNodeResponse); service.SendListNodesRequest(headers, fsConfig.FsId, dirId); auto listingResponse = service.RecvListNodesResponse(); // all the listed node refs point to the nodes that are being unlinked UNIT_ASSERT_VALUES_EQUAL_C( E_REJECTED, listingResponse->GetStatus(), FormatError(listingResponse->GetError())); shouldIntercept = false; env.GetRuntime().Send(unlinkNodeResponse.release()); unlinkResponse = service.RecvUnlinkNodeResponse(); UNIT_ASSERT_VALUES_EQUAL_C( S_OK, unlinkResponse->GetStatus(), FormatError(unlinkResponse->GetError())); listing = service.ListNodes(headers, fsConfig.FsId, dirId)->Record; UNIT_ASSERT_VALUES_EQUAL(0, listing.NodesSize()); UNIT_ASSERT_VALUES_EQUAL(0, listing.NamesSize()); const auto counters = env.GetCounters()->FindSubgroup("component", "service"); UNIT_ASSERT(counters); const auto counter = counters->GetCounter( "AppCriticalEvents/NodeNotFoundInShard"); UNIT_ASSERT_VALUES_EQUAL(0, counter->GetAtomic()); } SERVICE_TEST(ShouldHandleGetNodeAttrErrorsFromShardInListNodesRaceDetection) { config.SetDirectoryCreationInShardsEnabled(true); TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); auto headers = service.InitSession(fsConfig.FsId, "client"); auto dirId = service.CreateNode( headers, TCreateNodeArgs::Directory(RootNodeId, "dir") )->Record.GetNode().GetId(); auto subdir1Id = service.CreateNode( headers, TCreateNodeArgs::Directory(dirId, "subdir1") )->Record.GetNode().GetId(); Y_UNUSED(subdir1Id); auto subdir2Id = service.CreateNode( headers, TCreateNodeArgs::Directory(dirId, "subdir2") )->Record.GetNode().GetId(); IEventHandlePtr unlinkNodeResponse; bool shouldIntercept = true; NProto::TError getAttrError; env.GetRuntime().SetEventFilter( [&](auto& runtime, TAutoPtr& event) { Y_UNUSED(runtime); if (event->GetTypeRewrite() == TEvService::EvUnlinkNodeResponse && shouldIntercept) { unlinkNodeResponse.reset(event.Release()); return true; } if (event->GetTypeRewrite() == TEvService::EvGetNodeAttrResponse) { using TResponse = TEvService::TEvGetNodeAttrResponse; auto* msg = event->Get(); *msg->Record.MutableError() = getAttrError; } return false; }); service.SendUnlinkNodeRequest(headers, dirId, "subdir1", true); env.GetRuntime().DispatchEvents({}, TDuration::MilliSeconds(100)); UNIT_ASSERT(unlinkNodeResponse); const auto counters = env.GetCounters()->FindSubgroup("component", "service"); UNIT_ASSERT(counters); const auto counter = counters->GetCounter( "AppCriticalEvents/NodeNotFoundInShard"); { auto response = service.SendAndRecvListNodes(headers, fsConfig.FsId, dirId); UNIT_ASSERT_VALUES_EQUAL_C( S_OK, response->GetStatus(), response->GetErrorReason()); const auto& listing = response->Record; UNIT_ASSERT_VALUES_EQUAL(1, listing.NodesSize()); UNIT_ASSERT_VALUES_EQUAL(1, listing.NamesSize()); UNIT_ASSERT_VALUES_EQUAL(subdir2Id, listing.GetNodes(0).GetId()); UNIT_ASSERT_VALUES_EQUAL("subdir2", listing.GetNames(0)); UNIT_ASSERT_VALUES_EQUAL(0, counter->GetAtomic()); } getAttrError = MakeError(E_REJECTED); { auto response = service.SendAndRecvListNodes(headers, fsConfig.FsId, dirId); UNIT_ASSERT_VALUES_EQUAL_C( getAttrError.GetCode(), response->GetStatus(), response->GetErrorReason()); UNIT_ASSERT_VALUES_EQUAL(0, counter->GetAtomic()); } getAttrError = MakeError(E_FS_INVALID_SESSION); { auto response = service.SendAndRecvListNodes(headers, fsConfig.FsId, dirId); UNIT_ASSERT_VALUES_EQUAL_C( getAttrError.GetCode(), response->GetStatus(), response->GetErrorReason()); UNIT_ASSERT_VALUES_EQUAL(0, counter->GetAtomic()); } getAttrError = MakeError(E_IO); { auto response = service.SendAndRecvListNodes(headers, fsConfig.FsId, dirId); UNIT_ASSERT_VALUES_EQUAL_C( S_OK, response->GetStatus(), response->GetErrorReason()); const auto& listing = response->Record; UNIT_ASSERT_VALUES_EQUAL(1, listing.NodesSize()); UNIT_ASSERT_VALUES_EQUAL(1, listing.NamesSize()); UNIT_ASSERT_VALUES_EQUAL(subdir2Id, listing.GetNodes(0).GetId()); UNIT_ASSERT_VALUES_EQUAL("subdir2", listing.GetNames(0)); UNIT_ASSERT_VALUES_EQUAL(1, counter->GetAtomic()); } } SERVICE_TEST( ShouldNotRouteCreateNodeToNotYetConfiguredShardDuringExpansion) { // See https://github.com/ydb-platform/nbs/issues/7054 config.SetDirectoryCreationInShardsEnabled(true); // the test relies on round-robin shard selection (the default) config.SetShardBalancerPolicy(NProto::SBP_ROUND_ROBIN); TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); const TString shard3Id = fsConfig.FsId + "_s3"; auto headers = service.InitSession(fsConfig.FsId, "client"); // a directory owned by an existing shard - CreateNode under it is // handled by that shard, which picks the target shard for the new node const ui64 dirId = service.CreateNode( headers, TCreateNodeArgs::Directory(RootNodeId, "dir")) ->Record.GetNode().GetId(); UNIT_ASSERT_VALUES_UNEQUAL(0, ExtractShardNo(dirId)); // hold back ConfigureAsShard for the new shard s3 (it stays at // ShardNo == 0); count how many times s1/s2 get reconfigured TVector> delayedShard3Config; bool delayShard3Config = true; ui32 existingShardConfigureCount = 0; env.GetRuntime().SetEventFilter( [&](auto& runtime, TAutoPtr& ev) { Y_UNUSED(runtime); if (ev->GetTypeRewrite() == TEvIndexTablet::EvConfigureAsShardRequest) { const auto* msg = ev->Get(); const auto& fsId = msg->Record.GetFileSystemId(); if (delayShard3Config && fsId == shard3Id) { delayedShard3Config.emplace_back(ev.Release()); return true; } if (fsId == fsConfig.Shard1Id || fsId == fsConfig.Shard2Id) { ++existingShardConfigureCount; } } return false; }); // 2 -> 3 shard expansion service.SendResizeFileStoreRequest( fsConfig.FsId, fsConfig.MainFsBlockCount, false /* force */, 3 /* shardCount */); // wait until s3 is created and its ConfigureAsShard is intercepted for (ui32 i = 0; i < 200 && delayedShard3Config.empty(); ++i) { env.GetRuntime().DispatchEvents({}, TDuration::MilliSeconds(10)); } UNIT_ASSERT(!delayedShard3Config.empty()); env.GetRuntime().DispatchEvents({}, TDuration::MilliSeconds(100)); // s1/s2 must not be reconfigured (learn about s3) while s3 is held back UNIT_ASSERT_VALUES_EQUAL(0, existingShardConfigureCount); const auto critCounter = env.GetCounters() ->FindSubgroup("component", "service") ->GetCounter("AppCriticalEvents/ReceivedNodeOpErrorFromShard"); UNIT_ASSERT_VALUES_EQUAL(0, critCounter->GetAtomic()); // files created under the directory while the expansion is stuck must // all succeed and none may be routed to the unconfigured shard s3 for (ui32 i = 0; i < 6; ++i) { auto response = service.SendAndRecvCreateNode( headers, TCreateNodeArgs::File( dirId, TStringBuilder() << "file" << i)); UNIT_ASSERT_VALUES_EQUAL_C( S_OK, response->GetError().GetCode(), FormatError(response->GetError())); UNIT_ASSERT_VALUES_UNEQUAL( 3, ExtractShardNo(response->Record.GetNode().GetId())); } UNIT_ASSERT_VALUES_EQUAL(0, critCounter->GetAtomic()); // release the held config and let the expansion finish delayShard3Config = false; for (auto& ev: delayedShard3Config) { env.GetRuntime().Send(ev.Release(), nodeIdx); } delayedShard3Config.clear(); { auto response = service.RecvResizeFileStoreResponse(); UNIT_ASSERT_VALUES_EQUAL_C( S_OK, response->GetError().GetCode(), FormatError(response->GetError())); } // now s1/s2 get reconfigured with the 3-shard list UNIT_ASSERT_VALUES_UNEQUAL(0, existingShardConfigureCount); // main tablet suicides after ConfigureShards WaitForTabletStart(service); headers = service.InitSession(fsConfig.FsId, "client"); // s3 is a normal shard now and gets used without errors bool sawShard3 = false; for (ui32 i = 0; i < 12; ++i) { auto response = service.SendAndRecvCreateNode( headers, TCreateNodeArgs::File( dirId, TStringBuilder() << "file_after_" << i)); UNIT_ASSERT_VALUES_EQUAL_C( S_OK, response->GetError().GetCode(), FormatError(response->GetError())); if (ExtractShardNo(response->Record.GetNode().GetId()) == 3) { sawShard3 = true; } } UNIT_ASSERT(sawShard3); UNIT_ASSERT_VALUES_EQUAL(0, critCounter->GetAtomic()); } SERVICE_TEST(ShouldRejectRequestsToNewShardWhileSessionHasStaleShardList) { // See https://github.com/ydb-platform/nbs/issues/7061 config.SetDirectoryCreationInShardsEnabled(true); // the test relies on round-robin shard selection (the default) config.SetShardBalancerPolicy(NProto::SBP_ROUND_ROBIN); TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); // the session caches the 2-shard filesystem config in the service auto headers = service.InitSession(fsConfig.FsId, "client"); // a directory owned by an existing shard - that shard picks the target // shard for nodes created under it const ui64 dirId = service.CreateNode( headers, TCreateNodeArgs::Directory(RootNodeId, "dir")) ->Record.GetNode().GetId(); UNIT_ASSERT_VALUES_UNEQUAL(0, ExtractShardNo(dirId)); // hold back the final ConfigureShards for the main tablet: s3 is // configured and s1/s2 already know the 3-shard list, but the main // tablet has not been reconfigured (and restarted) yet, so the session // keeps the 2-shard config TVector> delayedMainConfig; env.GetRuntime().SetEventFilter( [&](auto& runtime, TAutoPtr& ev) { Y_UNUSED(runtime); if (ev->GetTypeRewrite() == TEvIndexTablet::EvConfigureShardsRequest) { const auto* msg = ev->Get(); if (msg->Record.GetFileSystemId() == fsConfig.FsId) { delayedMainConfig.emplace_back(ev.Release()); return true; } } return false; }); // 2 -> 3 shard expansion service.SendResizeFileStoreRequest( fsConfig.FsId, fsConfig.MainFsBlockCount, false /* force */, 3 /* shardCount */); for (ui32 i = 0; i < 200 && delayedMainConfig.empty(); ++i) { env.GetRuntime().DispatchEvents({}, TDuration::MilliSeconds(10)); } UNIT_ASSERT(!delayedMainConfig.empty()); // create files under the directory until one lands on the new shard ui64 nodeOnShard3 = 0; for (ui32 i = 0; i < 12 && !nodeOnShard3; ++i) { const ui64 nodeId = service.CreateNode( headers, TCreateNodeArgs::File(dirId, TStringBuilder() << "file" << i)) ->Record.GetNode().GetId(); if (ExtractShardNo(nodeId) == 3) { nodeOnShard3 = nodeId; } } UNIT_ASSERT_VALUES_UNEQUAL(0, nodeOnShard3); // a by-id request for that node is routed via the stale 2-shard list // and must be rejected with a retriable error { auto response = service.SendAndRecvGetNodeAttr( headers, fsConfig.FsId, nodeOnShard3, ""); UNIT_ASSERT_VALUES_EQUAL_C( E_REJECTED, response->GetError().GetCode(), FormatError(response->GetError())); } // let the expansion finish: the main tablet restarts after // ConfigureShards and the service recreates the session by itself, // refreshing the cached shard list without the client's involvement env.GetRuntime().SetEventFilter( TTestActorRuntimeBase::DefaultFilterFunc); for (auto& ev: delayedMainConfig) { env.GetRuntime().Send(ev.Release(), nodeIdx); } UNIT_ASSERT_VALUES_EQUAL_C( S_OK, service.RecvResizeFileStoreResponse()->GetError().GetCode(), "resize failed"); WaitForTabletStart(service); // the durable client keeps retrying with the same session - emulate it NProto::TError error; for (ui32 i = 0; i < 100; ++i) { auto response = service.SendAndRecvGetNodeAttr( headers, fsConfig.FsId, nodeOnShard3, ""); error = response->GetError(); if (!HasError(error)) { UNIT_ASSERT_VALUES_EQUAL( nodeOnShard3, response->Record.GetNode().GetId()); break; } env.GetRuntime().DispatchEvents({}, TDuration::MilliSeconds(100)); } UNIT_ASSERT_VALUES_EQUAL_C(S_OK, error.GetCode(), FormatError(error)); } SERVICE_TEST(ShouldHandleRenameNodeInDestinationError) { config.SetDirectoryCreationInShardsEnabled(true); TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); auto headers = service.InitSession(fsConfig.FsId, "client"); // / // └── dir1 (shard 1) // └── dir1_1 (shard 1) // └── file (shard 1) // └── dir2 (shard 2) // └── dir2_1 (shard 1) // └── file (shard 1) auto dir1 = service.CreateNode( headers, TCreateNodeArgs::Directory(RootNodeId, "dir1"))->Record.GetNode(); auto dir1_1 = service.CreateNode( headers, TCreateNodeArgs::Directory(dir1.GetId(), "dir1_1")) ->Record.GetNode(); auto dir2 = service.CreateNode( headers, TCreateNodeArgs::Directory(RootNodeId, "dir2"))->Record.GetNode(); auto dir2_1 = service.CreateNode( headers, TCreateNodeArgs::Directory(dir2.GetId(), "dir2_1")) ->Record.GetNode(); auto file1 = service.CreateNode( headers, TCreateNodeArgs::File(dir1_1.GetId(), "file"))->Record.GetNode(); auto file2 = service.CreateNode( headers, TCreateNodeArgs::File(dir2_1.GetId(), "file"))->Record.GetNode(); // // This condition ensures cross-shard rename data path. // UNIT_ASSERT_VALUES_UNEQUAL( ExtractShardNo(dir1.GetId()), ExtractShardNo(dir2.GetId())); auto renameResponse = service.SendAndRecvRenameNode( headers, dir1.GetId(), "dir1_1", dir2.GetId(), "dir2_1", 0 /* flags */); UNIT_ASSERT_VALUES_EQUAL_C( E_FS_NOTEMPTY, renameResponse->GetError().GetCode(), renameResponse->GetError().GetMessage()); service.UnlinkNode(headers, dir2_1.GetId(), "file", false /* unlinkDirectory */); renameResponse = service.SendAndRecvRenameNode( headers, dir1_1.GetId(), "file", dir2.GetId(), "dir2_1", 0 /* flags */); UNIT_ASSERT_VALUES_EQUAL_C( E_FS_ISDIR, renameResponse->GetError().GetCode(), renameResponse->GetError().GetMessage()); renameResponse = service.SendAndRecvRenameNode( headers, dir1_1.GetId(), "file", dir2.GetId() + 1111, // nonexistent parent in shard2 "dir2_1", 0 /* flags */); UNIT_ASSERT_VALUES_EQUAL_C( E_FS_NOENT, renameResponse->GetError().GetCode(), renameResponse->GetError().GetMessage()); renameResponse = service.SendAndRecvRenameNode( headers, dir1_1.GetId(), "file", dir2.GetId(), "dir2_1", ProtoFlag(NProto::TRenameNodeRequest::F_NOREPLACE)); UNIT_ASSERT_VALUES_EQUAL_C( E_FS_EXIST, renameResponse->GetError().GetCode(), renameResponse->GetError().GetMessage()); renameResponse = service.SendAndRecvRenameNode( headers, dir1.GetId(), "dir1_1", dir2.GetId(), "dir2_1", 0 /* flags */); UNIT_ASSERT_VALUES_EQUAL_C( S_OK, renameResponse->GetError().GetCode(), renameResponse->GetError().GetMessage()); auto attr = service.GetNodeAttr( headers, fsConfig.FsId, dir2.GetId(), "dir2_1")->Record.GetNode(); UNIT_ASSERT_VALUES_EQUAL(dir1_1.GetId(), attr.GetId()); const auto counters = env.GetCounters()->FindSubgroup("component", "service"); UNIT_ASSERT(counters); const auto counter = counters->GetCounter( "AppCriticalEvents/ReceivedNodeOpErrorFromShard"); // // E_FS_NOTEMPTY and E_FS_ISDIR errors received from shard shouldn't // trigger this critical event. // UNIT_ASSERT_VALUES_EQUAL(0, counter->GetAtomic()); } SERVICE_TEST( ShouldRetryRenameNodeInShardUponLeaderRestartWithDirectoriesInShards) { config.SetDirectoryCreationInShardsEnabled(true); TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); auto headers = service.InitSession(fsConfig.FsId, "client"); // creating 2 dirs - we'll move a file from one dir to another ui64 dir1Id = service.CreateNode( headers, TCreateNodeArgs::Directory(RootNodeId, "dir1") )->Record.GetNode().GetId(); ui64 dir2Id = service.CreateNode( headers, TCreateNodeArgs::Directory(RootNodeId, "dir2") )->Record.GetNode().GetId(); UNIT_ASSERT_VALUES_UNEQUAL(0, ExtractShardNo(dir1Id)); UNIT_ASSERT_VALUES_UNEQUAL(0, ExtractShardNo(dir2Id)); UNIT_ASSERT_VALUES_UNEQUAL( ExtractShardNo(dir1Id), ExtractShardNo(dir2Id)); // creating a file which we will move ui64 node1Id = service.CreateNode( headers, TCreateNodeArgs::File(dir1Id, "file1") )->Record.GetNode().GetId(); // configuring an interceptor to make RenameNode get stuck in the shard // in charge of dir1 TEventHelper evHelper(env); evHelper.CatchEvent(TEvIndexTablet::EvRenameNodeInDestinationRequest); service.SendRenameNodeRequest( headers, dir1Id, "file1", dir2Id, "file2", 0); ui32 iterations = 0; while (!evHelper.Event && iterations++ < 100) { env.GetRuntime().DispatchEvents({}, TDuration::MilliSeconds(50)); } UNIT_ASSERT(evHelper.Event); evHelper.ShouldIntercept = false; // listing the nodes in dir1 and dir2 - no changes should be visible // at this point { auto listNodesResponse = service.ListNodes( headers, fsConfig.FsId, dir1Id)->Record; UNIT_ASSERT_VALUES_EQUAL(1, listNodesResponse.NamesSize()); UNIT_ASSERT_VALUES_EQUAL(1, listNodesResponse.NodesSize()); } { auto listNodesResponse = service.ListNodes( headers, fsConfig.FsId, dir2Id)->Record; UNIT_ASSERT_VALUES_EQUAL(0, listNodesResponse.NamesSize()); UNIT_ASSERT_VALUES_EQUAL(0, listNodesResponse.NodesSize()); } // rebooting the shard in charge of dir1 to trigger OpLog replay which // should rerun the part of the RenameNode operation that got stuck const ui64 tabletId = ExtractShardNo(dir1Id) == 1 ? fsInfo.Shard1TabletId : fsInfo.Shard2TabletId; TIndexTabletClient tablet(env.GetRuntime(), nodeIdx, tabletId); tablet.RebootTablet(); { auto renameResponse = service.RecvRenameNodeResponse(); UNIT_ASSERT_VALUES_EQUAL_C( E_REJECTED, renameResponse->GetError().GetCode(), renameResponse->GetError().GetMessage()); } // remaking session since CreateSessionActor doesn't do it by itself // because EvWakeup never arrives because Scheduling doesn't work by // default and RegistrationObservers get reset after RebootTablet headers = service.InitSession(fsConfig.FsId, "client"); // after shard reboot our RenameNode operation should have completed // checking it via directory listing { auto listNodesResponse = service.ListNodes( headers, fsConfig.FsId, dir1Id)->Record; UNIT_ASSERT_VALUES_EQUAL(0, listNodesResponse.NamesSize()); UNIT_ASSERT_VALUES_EQUAL(0, listNodesResponse.NodesSize()); } { auto listNodesResponse = service.ListNodes( headers, fsConfig.FsId, dir2Id)->Record; UNIT_ASSERT_VALUES_EQUAL(1, listNodesResponse.NamesSize()); UNIT_ASSERT_VALUES_EQUAL(1, listNodesResponse.NodesSize()); UNIT_ASSERT_VALUES_EQUAL("file2", listNodesResponse.GetNames(0)); UNIT_ASSERT_VALUES_EQUAL( node1Id, listNodesResponse.GetNodes(0).GetId()); } } SERVICE_TEST(ShouldLockSourceNodeRefsUponRenameNodeWithDirectoriesInShards) { config.SetDirectoryCreationInShardsEnabled(true); TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); auto headers = service.InitSession(fsConfig.FsId, "client"); // creating 2 dirs - we'll move a file from one dir to another ui64 dir1Id = service.CreateNode( headers, TCreateNodeArgs::Directory(RootNodeId, "dir1") )->Record.GetNode().GetId(); ui64 dir2Id = service.CreateNode( headers, TCreateNodeArgs::Directory(RootNodeId, "dir2") )->Record.GetNode().GetId(); UNIT_ASSERT_VALUES_UNEQUAL(0, ExtractShardNo(dir1Id)); UNIT_ASSERT_VALUES_UNEQUAL(0, ExtractShardNo(dir2Id)); UNIT_ASSERT_VALUES_UNEQUAL( ExtractShardNo(dir1Id), ExtractShardNo(dir2Id)); // creating a file which we will move ui64 node1Id = service.CreateNode( headers, TCreateNodeArgs::File(dir1Id, "file1") )->Record.GetNode().GetId(); // configuring an interceptor to make RenameNode get stuck in the shard // in charge of dir1 TEventHelper evHelper(env); evHelper.CatchEvent(TEvIndexTablet::EvRenameNodeInDestinationRequest); service.SendRenameNodeRequest( headers, dir1Id, "file1", dir2Id, "file2", 0); ui32 iterations = 0; while (!evHelper.Event && iterations++ < 100) { env.GetRuntime().DispatchEvents({}, TDuration::MilliSeconds(50)); } UNIT_ASSERT(evHelper.Event); evHelper.ShouldIntercept = false; // listing dir1 - at this point our file should still be seen there { auto listNodesResponse = service.ListNodes( headers, fsConfig.FsId, dir1Id)->Record; UNIT_ASSERT_VALUES_EQUAL(1, listNodesResponse.NamesSize()); UNIT_ASSERT_VALUES_EQUAL(1, listNodesResponse.NodesSize()); UNIT_ASSERT_VALUES_EQUAL("file1", listNodesResponse.GetNames(0)); } // unlinking of that file should be unallowed since we have an inflight // RenameNode operation service.SendUnlinkNodeRequest(headers, dir1Id, "file1"); { auto unlinkResponse = service.RecvUnlinkNodeResponse(); UNIT_ASSERT_VALUES_EQUAL_C( E_REJECTED, unlinkResponse->GetError().GetCode(), unlinkResponse->GetError().GetMessage()); } // renaming should not be allowed as well service.SendRenameNodeRequest( headers, dir1Id, "file1", dir1Id, "file1_moved", 0); { auto renameResponse = service.RecvRenameNodeResponse(); UNIT_ASSERT_VALUES_EQUAL_C( E_REJECTED, renameResponse->GetError().GetCode(), renameResponse->GetError().GetMessage()); } // listing dir1 - our file should still be there { auto listNodesResponse = service.ListNodes( headers, fsConfig.FsId, dir1Id)->Record; UNIT_ASSERT_VALUES_EQUAL(1, listNodesResponse.NamesSize()); UNIT_ASSERT_VALUES_EQUAL(1, listNodesResponse.NodesSize()); UNIT_ASSERT_VALUES_EQUAL("file1", listNodesResponse.GetNames(0)); } // unblocking the inflight RenameNode op env.GetRuntime().Send(evHelper.Event.Release(), nodeIdx); { auto renameResponse = service.RecvRenameNodeResponse(); UNIT_ASSERT_VALUES_EQUAL_C( S_OK, renameResponse->GetError().GetCode(), renameResponse->GetError().GetMessage()); } // now the file should finally be moved { auto listNodesResponse = service.ListNodes( headers, fsConfig.FsId, dir1Id)->Record; UNIT_ASSERT_VALUES_EQUAL(0, listNodesResponse.NamesSize()); UNIT_ASSERT_VALUES_EQUAL(0, listNodesResponse.NodesSize()); } { auto listNodesResponse = service.ListNodes( headers, fsConfig.FsId, dir2Id)->Record; UNIT_ASSERT_VALUES_EQUAL(1, listNodesResponse.NamesSize()); UNIT_ASSERT_VALUES_EQUAL(1, listNodesResponse.NodesSize()); UNIT_ASSERT_VALUES_EQUAL("file2", listNodesResponse.GetNames(0)); UNIT_ASSERT_VALUES_EQUAL( node1Id, listNodesResponse.GetNodes(0).GetId()); } // RenameNode and UnlinkNode ops should work again ui64 node2Id = service.CreateNode( headers, TCreateNodeArgs::File(dir1Id, "file1") )->Record.GetNode().GetId(); UNIT_ASSERT_VALUES_UNEQUAL(node1Id, node2Id); service.SendRenameNodeRequest( headers, dir1Id, "file1", dir1Id, "file1_moved", 0); { auto renameResponse = service.RecvRenameNodeResponse(); UNIT_ASSERT_VALUES_EQUAL_C( S_OK, renameResponse->GetError().GetCode(), renameResponse->GetError().GetMessage()); } ui64 node3Id = service.CreateNode( headers, TCreateNodeArgs::File(dir1Id, "file1") )->Record.GetNode().GetId(); UNIT_ASSERT_VALUES_UNEQUAL(node1Id, node3Id); UNIT_ASSERT_VALUES_UNEQUAL(node2Id, node3Id); service.SendUnlinkNodeRequest(headers, dir1Id, "file1"); { auto unlinkResponse = service.RecvUnlinkNodeResponse(); UNIT_ASSERT_VALUES_EQUAL_C( S_OK, unlinkResponse->GetError().GetCode(), unlinkResponse->GetError().GetMessage()); } { auto listNodesResponse = service.ListNodes( headers, fsConfig.FsId, dir1Id)->Record; UNIT_ASSERT_VALUES_EQUAL(1, listNodesResponse.NamesSize()); UNIT_ASSERT_VALUES_EQUAL(1, listNodesResponse.NodesSize()); UNIT_ASSERT_VALUES_EQUAL("file1_moved", listNodesResponse.GetNames(0)); UNIT_ASSERT_VALUES_EQUAL( node2Id, listNodesResponse.GetNodes(0).GetId()); } } SERVICE_TEST( ShouldRestoreNodeRefLocksUponLeaderRestartWithDirectoriesInShards) { config.SetDirectoryCreationInShardsEnabled(true); TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); auto headers = service.InitSession(fsConfig.FsId, "client"); // creating 2 dirs - we'll move a file from one dir to another ui64 dir1Id = service.CreateNode( headers, TCreateNodeArgs::Directory(RootNodeId, "dir1") )->Record.GetNode().GetId(); ui64 dir2Id = service.CreateNode( headers, TCreateNodeArgs::Directory(RootNodeId, "dir2") )->Record.GetNode().GetId(); UNIT_ASSERT_VALUES_UNEQUAL(0, ExtractShardNo(dir1Id)); UNIT_ASSERT_VALUES_UNEQUAL(0, ExtractShardNo(dir2Id)); UNIT_ASSERT_VALUES_UNEQUAL( ExtractShardNo(dir1Id), ExtractShardNo(dir2Id)); // creating a file which we will move service.CreateNode(headers, TCreateNodeArgs::File(dir1Id, "file1")); // configuring an interceptor to make RenameNode get stuck in the shard // in charge of dir1 TEventHelper evHelper(env); evHelper.CatchEvent(TEvIndexTablet::EvRenameNodeInDestinationRequest); service.SendRenameNodeRequest( headers, dir1Id, "file1", dir2Id, "file2", 0); ui32 iterations = 0; while (!evHelper.Event && iterations++ < 100) { env.GetRuntime().DispatchEvents({}, TDuration::MilliSeconds(50)); } UNIT_ASSERT(evHelper.Event); // rebooting the shard in charge of dir1 to trigger OpLog replay which // should lock the affected NodeRef const ui64 tabletId = ExtractShardNo(dir1Id) == 1 ? fsInfo.Shard1TabletId : fsInfo.Shard2TabletId; TIndexTabletClient tablet(env.GetRuntime(), nodeIdx, tabletId); tablet.RebootTablet(); { auto renameResponse = service.RecvRenameNodeResponse(); UNIT_ASSERT_VALUES_EQUAL_C( E_REJECTED, renameResponse->GetError().GetCode(), renameResponse->GetError().GetMessage()); } // remaking session since CreateSessionActor doesn't do it by itself // because EvWakeup never arrives because Scheduling doesn't work by // default and RegistrationObservers get reset after RebootTablet headers = service.InitSession(fsConfig.FsId, "client"); // unlinking of that file should be unallowed since we have an inflight // RenameNode operation service.SendUnlinkNodeRequest(headers, dir1Id, "file1"); { auto unlinkResponse = service.RecvUnlinkNodeResponse(); UNIT_ASSERT_VALUES_EQUAL_C( E_REJECTED, unlinkResponse->GetError().GetCode(), unlinkResponse->GetError().GetMessage()); } // renaming should not be allowed as well service.SendRenameNodeRequest( headers, dir1Id, "file1", dir1Id, "file1_moved", 0); { auto renameResponse = service.RecvRenameNodeResponse(); UNIT_ASSERT_VALUES_EQUAL_C( E_REJECTED, renameResponse->GetError().GetCode(), renameResponse->GetError().GetMessage()); } } SERVICE_TEST( ShouldProcessRenameNodeInDestinationErrorWithDirectoriesInShards) { config.SetDirectoryCreationInShardsEnabled(true); TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); auto headers = service.InitSession(fsConfig.FsId, "client"); // creating 2 dirs - we'll move a file from one dir to another ui64 dir1Id = service.CreateNode( headers, TCreateNodeArgs::Directory(RootNodeId, "dir1") )->Record.GetNode().GetId(); ui64 dir2Id = service.CreateNode( headers, TCreateNodeArgs::Directory(RootNodeId, "dir2") )->Record.GetNode().GetId(); UNIT_ASSERT_VALUES_UNEQUAL(0, ExtractShardNo(dir1Id)); UNIT_ASSERT_VALUES_UNEQUAL(0, ExtractShardNo(dir2Id)); UNIT_ASSERT_VALUES_UNEQUAL( ExtractShardNo(dir1Id), ExtractShardNo(dir2Id)); // creating a file which we will try move ui64 node1Id = service.CreateNode( headers, TCreateNodeArgs::File(dir1Id, "file1") )->Record.GetNode().GetId(); // creating a file which will prevent the move from succeeding ui64 node2Id = service.CreateNode( headers, TCreateNodeArgs::File(dir2Id, "file2") )->Record.GetNode().GetId(); UNIT_ASSERT_VALUES_UNEQUAL(node1Id, node2Id); service.SendRenameNodeRequest( headers, dir1Id, "file1", dir2Id, "file2", ProtoFlag(NProto::TRenameNodeRequest::F_NOREPLACE)); { auto renameResponse = service.RecvRenameNodeResponse(); UNIT_ASSERT_VALUES_EQUAL_C( E_FS_EXIST, renameResponse->GetError().GetCode(), renameResponse->GetError().GetMessage()); } // listing the nodes in dir1 and dir2 - no changes should be done { auto listNodesResponse = service.ListNodes( headers, fsConfig.FsId, dir1Id)->Record; UNIT_ASSERT_VALUES_EQUAL(1, listNodesResponse.NamesSize()); UNIT_ASSERT_VALUES_EQUAL(1, listNodesResponse.NodesSize()); UNIT_ASSERT_VALUES_EQUAL( node1Id, listNodesResponse.GetNodes(0).GetId()); UNIT_ASSERT_VALUES_EQUAL("file1", listNodesResponse.GetNames(0)); } { auto listNodesResponse = service.ListNodes( headers, fsConfig.FsId, dir2Id)->Record; UNIT_ASSERT_VALUES_EQUAL(1, listNodesResponse.NamesSize()); UNIT_ASSERT_VALUES_EQUAL(1, listNodesResponse.NodesSize()); UNIT_ASSERT_VALUES_EQUAL( node2Id, listNodesResponse.GetNodes(0).GetId()); UNIT_ASSERT_VALUES_EQUAL("file2", listNodesResponse.GetNames(0)); } } SERVICE_TEST(ShouldRestoreSessionAfterShardRestartViaGetNodeAttr) { TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); auto headers = service.InitSession(fsConfig.FsId, "client"); TAutoPtr getNodeAttrEvent; bool responseFromShardObserved = false; TActorId shardActorId; env.GetRuntime().SetEventFilter( [&](auto& runtime, TAutoPtr& event) { Y_UNUSED(runtime); // Intercept CreateNodeResponse from shard if (event->GetTypeRewrite() == TEvService::EvCreateNodeResponse) { auto* msg = event ->template Get(); if (msg->Record.HasNode() && ExtractShardNo(msg->Record.GetNode().GetId()) == 1) { msg->Record.MutableError()->SetCode(E_FS_EXIST); if (!shardActorId) { shardActorId = event->Sender; } responseFromShardObserved = true; } } else if ( event->GetTypeRewrite() == TEvService::EvGetNodeAttrRequest) { // Intercept GetNodeAttrRequest getNodeAttrEvent = event.Release(); return true; } return false; }); // Create a file - this will go to main tablet, then to shard service.SendCreateNodeRequest( headers, TCreateNodeArgs::File(RootNodeId, "file1")); TDispatchOptions options; options.FinalEvents = {TDispatchOptions::TFinalEventCondition( TEvService::EvCreateNodeResponse, 1)}; env.GetRuntime().DispatchEvents(options); UNIT_ASSERT(responseFromShardObserved); // Now the main filesystem will send a GetNodeAttr request to the shard // to check the created node. We will intercept that request and in the // meantime reboot the shard. for (ui32 attempt = 0; attempt < 100 && !getNodeAttrEvent; ++attempt) { env.GetRuntime().DispatchEvents({}, TDuration::MilliSeconds(50)); } UNIT_ASSERT(getNodeAttrEvent); env.GetRuntime().SetEventFilter( TTestActorRuntimeBase::DefaultFilterFunc); // Restart the shard by sending poison pill to the shard actor { env.GetRuntime().Send( new IEventHandle( shardActorId, // recipient TActorId(), // sender new TEvents::TEvPoisonPill(), 0, // flags 0), nodeIdx); WaitForTabletStart(service); } // Resend the GetNodeAttr request env.GetRuntime().Send(getNodeAttrEvent.Release(), nodeIdx); for (ui32 attempt = 0; attempt < 100; ++attempt) { env.GetRuntime().DispatchEvents({}, TDuration::MilliSeconds(50)); } // Recieve the original CreateNode response auto createResponse = service.RecvCreateNodeResponse(); UNIT_ASSERT_VALUES_EQUAL_C( S_OK, createResponse->GetError().GetCode(), createResponse->GetError().GetMessage()); UNIT_ASSERT(createResponse->Record.HasNode()); const auto counters = env.GetCounters()->FindSubgroup("component", "service"); UNIT_ASSERT(counters); const auto counter = counters->GetCounter( "AppCriticalEvents/ReceivedNodeOpErrorFromShard"); UNIT_ASSERT_VALUES_EQUAL(0, counter->GetAtomic()); } SERVICE_TEST_DIR_CREATION_IN_SHARDS( ShouldNotRequireActiveSessionForGetNodeAttrBatchToShard) { TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); auto headers = service.InitSession(fsConfig.FsId, "client"); const auto nodeId = service .CreateNode(headers, TCreateNodeArgs::File(RootNodeId, "file1")) ->Record.GetNode() .GetId(); const ui64 shardTabletId = ExtractShardNo(nodeId) == 1 ? fsInfo.Shard1TabletId : fsInfo.Shard2TabletId; TAutoPtr getNodeAttrBatchEvent; env.GetRuntime().SetEventFilter( [&](auto& runtime, TAutoPtr& event) { Y_UNUSED(runtime); if (event->GetTypeRewrite() == TEvIndexTablet::EvGetNodeAttrBatchRequest) { getNodeAttrBatchEvent = event.Release(); return true; } return false; }); service.SendListNodesRequest(headers, RootNodeId); for (ui32 attempt = 0; attempt < 100 && !getNodeAttrBatchEvent; ++attempt) { env.GetRuntime().DispatchEvents({}, TDuration::MilliSeconds(50)); } UNIT_ASSERT(getNodeAttrBatchEvent); env.GetRuntime().SetEventFilter( TTestActorRuntimeBase::DefaultFilterFunc); TIndexTabletClient shard(env.GetRuntime(), nodeIdx, shardTabletId); shard.RebootTablet(); env.GetRuntime().Send(getNodeAttrBatchEvent.Release(), nodeIdx); for (ui32 attempt = 0; attempt < 100; ++attempt) { env.GetRuntime().DispatchEvents({}, TDuration::MilliSeconds(50)); } auto listResponse = service.RecvListNodesResponse(); UNIT_ASSERT_VALUES_EQUAL_C( S_OK, listResponse->GetError().GetCode(), listResponse->GetError().GetMessage()); UNIT_ASSERT_VALUES_EQUAL(1, listResponse->Record.NodesSize()); } // TODO(2566) get rid of this test after migration SERVICE_TEST(ShouldReportSevenBytesHandlesCount) { config.SetStrictFileSystemSizeEnforcementEnabled(true); TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); auto headers = service.InitSession(fsConfig.FsId, "client"); auto registry = env.GetRegistry(); // Create some number of nodes and handles TVector handles; TVector nodes; const ui64 filesCount = 128; for (ui64 i = 0; i < filesCount; ++i) { auto createNodeResponse = service.CreateNode( headers, TCreateNodeArgs::File( RootNodeId, TStringBuilder() << "file" << i)) ->Record; const ui64 nodeId = createNodeResponse.GetNode().GetId(); nodes.push_back(nodeId); const ui64 shardNo = i % 2 + 1; UNIT_ASSERT(!IsSeventhByteUsed(nodeId)); UNIT_ASSERT_VALUES_EQUAL(shardNo, ExtractShardNo(nodeId)); const ui64 handle = service.CreateHandle( headers, fsConfig.FsId, nodeId, "", TCreateHandleArgs::RDWR)->Record.GetHandle(); handles.push_back(handle); UNIT_ASSERT(!IsSeventhByteUsed(handle)); UNIT_ASSERT_VALUES_EQUAL(shardNo, ExtractShardNo(handle)); } env.GetRuntime().AdvanceCurrentTime(TDuration::Seconds(15)); NActors::TDispatchOptions options; // As the main tablet is restarted after ConfigureShards we need to // dispatch shard count + 2 EvUpdateCounters events. options.FinalEvents.emplace_back( TDispatchOptions::TFinalEventCondition( TEvIndexTabletPrivate::EvUpdateCounters, fsConfig.ShardIds().size() + 2)); env.GetRuntime().DispatchEvents(options); TTestRegistryVisitor visitor; registry->Visit(TInstant::Zero(), visitor); // We should not create new handles that use seventh byte visitor.ValidateExpectedCounters({ {{{"sensor", "SevenBytesHandlesCount"}, {"filesystem", "test"}}, 0}, {{{"sensor", "SevenBytesHandlesCount"}, {"filesystem", "test_s1"}}, 0}, {{{"sensor", "SevenBytesHandlesCount"}, {"filesystem", "test_s2"}}, 0}, {{{"sensor", "UsedHandlesCount"}, {"filesystem", "test"}}, 0}, {{{"sensor", "UsedHandlesCount"}, {"filesystem", "test_s1"}}, filesCount / 2}, {{{"sensor", "UsedHandlesCount"}, {"filesystem", "test_s2"}}, filesCount / 2} }); for (ui64 i = 0; i < handles.size(); i++) { service.DestroyHandle(headers, fsConfig.FsId, nodes[i], handles[i]); } } void DoShouldReportStrictFileSystemSizeEnforcementMetrics( NProto::TStorageConfig& config, bool strictFileSystemSizeEnforcementEnabled, bool directoryCreationInShardsEnabled) { config.SetStrictFileSystemSizeEnforcementEnabled( strictFileSystemSizeEnforcementEnabled); config.SetDirectoryCreationInShardsEnabled( directoryCreationInShardsEnabled); TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); auto headers = service.InitSession(fsConfig.FsId, "client"); auto registry = env.GetRegistry(); const ui32 restartsCount = (directoryCreationInShardsEnabled || strictFileSystemSizeEnforcementEnabled) ? 1 : 0; env.GetRuntime().AdvanceCurrentTime(TDuration::Seconds(16)); NActors::TDispatchOptions options; options.FinalEvents.emplace_back( TDispatchOptions::TFinalEventCondition( TEvIndexTabletPrivate::EvUpdateCounters, fsConfig.ShardIds().size() + 1 + restartsCount)); env.GetRuntime().DispatchEvents(options); TTestRegistryVisitor visitor; registry->Visit(TInstant::Zero(), visitor); const auto directoryCreationInShardsEnabledMetrics = directoryCreationInShardsEnabled ? 1 : 0; const auto strictFileSystemSizeEnforcementEnabledMetrics = strictFileSystemSizeEnforcementEnabled ? 1 : 0; visitor.ValidateExpectedCounters({ {{{"sensor", "StrictFileSystemSizeEnforcementEnabled"}, {"filesystem", "test"}}, strictFileSystemSizeEnforcementEnabledMetrics}, {{{"sensor", "StrictFileSystemSizeEnforcementEnabled"}, {"filesystem", "test_s1"}}, strictFileSystemSizeEnforcementEnabledMetrics}, {{{"sensor", "StrictFileSystemSizeEnforcementEnabled"}, {"filesystem", "test_s2"}}, strictFileSystemSizeEnforcementEnabledMetrics}, {{{"sensor", "DirectoryCreationInShardsEnabled"}, {"filesystem", "test"}}, directoryCreationInShardsEnabledMetrics}, {{{"sensor", "DirectoryCreationInShardsEnabled"}, {"filesystem", "test_s1"}}, directoryCreationInShardsEnabledMetrics}, {{{"sensor", "DirectoryCreationInShardsEnabled"}, {"filesystem", "test_s2"}}, directoryCreationInShardsEnabledMetrics}, }); } SERVICE_TEST(ShouldReportStrictFileSystemSizeEnforcementMetrics) { DoShouldReportStrictFileSystemSizeEnforcementMetrics( config, true, false); DoShouldReportStrictFileSystemSizeEnforcementMetrics( config, false, true); } SERVICE_TEST( ShouldReportWeightedDeterministicBalancerWithoutStrictEnforcement) { config.SetShardBalancerPolicy( NProto::SBP_WEIGHTED_DETERMINISTIC); config.SetStrictFileSystemSizeEnforcementEnabled(false); TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); const auto counters = env.GetCounters()->FindSubgroup("component", "service"); UNIT_ASSERT(counters); const auto counter = counters->GetCounter( "AppCriticalEvents/" "IncompatibleFeatures"); UNIT_ASSERT(counter->GetAtomic() > 0); } void DoShouldHaveCorrectAggregateCountersWithFanoutDisabled( NProto::TStorageConfig config, const bool fanoutStatsCollectionInShardsDisabled) { constexpr ui64 blockSize = 4_KB; constexpr ui64 shardBlockCount = 1024; constexpr ui64 shardAllocationUnit = shardBlockCount * blockSize; constexpr ui64 shardCount = 16; constexpr ui64 filesCount = 32; constexpr ui64 fsSize = shardBlockCount * (shardCount - 1) + shardBlockCount / 2; config.SetStrictFileSystemSizeEnforcementEnabled(true); config.SetFanoutStatsCollectionInShardsDisabled( fanoutStatsCollectionInShardsDisabled); config.SetAutomaticShardCreationEnabled(true); config.SetShardAllocationUnit(shardAllocationUnit); const TString fsId = "test"; TTestEnv env({}, config); const ui32 nodeIdx = env.AddDynamicNode(); TServiceClient service(env.GetRuntime(), nodeIdx); service.CreateFileStore(fsId, fsSize); WaitForTabletStart(service); const auto headers = service.InitSession(fsId, "client"); const auto data = GenerateValidateData(blockSize); for (ui64 i = 0; i < filesCount; ++i) { const auto response = service.CreateNode( headers, TCreateNodeArgs::File( RootNodeId, TStringBuilder() << "file" << i)); const ui64 nodeId = response->Record.GetNode().GetId(); const ui64 handle = service.CreateHandle( headers, fsId, nodeId, "", TCreateHandleArgs::RDWR)->Record.GetHandle(); service.WriteData(headers, fsId, nodeId, handle, 0, data); } env.GetRuntime().AdvanceCurrentTime(TDuration::Seconds(15)); // To make the following counter updates less dependent on timing, // GetStorageStats is called to refresh the cache. GetStorageStats(service, fsId); // Update counters in the main tablet and all the shards. TDispatchOptions options; options.FinalEvents.emplace_back( TDispatchOptions::TFinalEventCondition( TEvIndexTabletPrivate::EvUpdateCounters, shardCount + 2)); env.GetRuntime().DispatchEvents(options); env.GetRegistry()->Update(env.GetRuntime().GetCurrentTime()); auto tabletCounters = env.GetRuntime().GetAppData().Counters ->FindSubgroup("counters", "filestore") ->FindSubgroup("component", "storage") ->FindSubgroup("type", "hdd"); const auto storageStatsRequestsPerShard = tabletCounters->FindSubgroup("request", "GetStorageStats") ->GetCounter("Count") ->GetAtomic() / shardCount; if (fanoutStatsCollectionInShardsDisabled) { // Linear complexity by shardCount. UNIT_ASSERT_LE(storageStatsRequestsPerShard, 4); } else { // Quadratic complexity by shardCount. UNIT_ASSERT_GE(storageStatsRequestsPerShard, shardCount); } auto mainFsCounters = GetFileSystemCounters(env, fsId); UNIT_ASSERT_VALUES_EQUAL( filesCount, mainFsCounters->GetCounter("AggregateUsedNodesCount")->GetAtomic()); UNIT_ASSERT_VALUES_EQUAL( filesCount * blockSize, mainFsCounters->GetCounter("AggregateUsedBytesCount")->GetAtomic()); for (ui64 shardNo = 1; shardNo <= shardCount; ++shardNo) { const TString shardId = TStringBuilder() << fsId << ShardNumPrefix << shardNo; auto shardCounters = GetFileSystemCounters(env, shardId); UNIT_ASSERT_VALUES_EQUAL( filesCount, shardCounters->GetCounter("AggregateUsedNodesCount") ->GetAtomic()); UNIT_ASSERT_VALUES_EQUAL( filesCount * blockSize, shardCounters->GetCounter("AggregateUsedBytesCount") ->GetAtomic()); } } SERVICE_TEST(ShouldHaveCorrectAggregateCountersWithFanoutDisabled) { DoShouldHaveCorrectAggregateCountersWithFanoutDisabled(config, false); DoShouldHaveCorrectAggregateCountersWithFanoutDisabled(config, true); } SERVICE_TEST(ShouldUseOldHandles) { // This test verifies that when TStorageConfig::maxShardCount is defualt // (254) and we create a handle that uses the 7th byte, // TStorageServiceActor::SelectShard selects a correct shard. const ui64 blockSize = 4_KB; const ui64 shardBlockCount = 1024; const ui64 shardAllocationUnit = shardBlockCount * blockSize; const ui64 shardCount = 16; const ui64 fsSize = shardBlockCount * (shardCount - 1) + shardBlockCount / 2; config.SetAutomaticShardCreationEnabled(true); config.SetShardAllocationUnit(shardAllocationUnit); const TString fsId = "test"; TTestEnv env({}, config); ui32 nodeIdx = env.AddDynamicNode(); TServiceClient service(env.GetRuntime(), nodeIdx); service.CreateFileStore(fsId, fsSize); WaitForTabletStart(service); auto headers = service.InitSession(fsId, "client"); auto createNodeResponse = service .CreateNode(headers, TCreateNodeArgs::File(RootNodeId, "file")) ->Record; const ui64 nodeId = createNodeResponse.GetNode().GetId(); TString fsIdFromRequest; env.GetRuntime().SetEventFilter( [&](auto& runtime, TAutoPtr& e) { Y_UNUSED(runtime); if (e->GetTypeRewrite() == TEvService::EvReadDataRequest) { const auto* msg = e->Get(); fsIdFromRequest = msg->Record.GetFileSystemId(); } return false; }); // Test handle that contains non-zero shard service.AssertReadDataFailed( headers, fsId, nodeId, 0x0101000000000001, 0, 1); UNIT_ASSERT_VALUES_EQUAL(fsId + "_s1", fsIdFromRequest); // Test handle that contains zero shard service.AssertReadDataFailed( headers, fsId, nodeId, 0x0001000000000001, 0, 1); UNIT_ASSERT_VALUES_EQUAL(fsId, fsIdFromRequest); } SERVICE_TEST(ShouldUnsafeCreateHandlesInShards) { const ui64 blockSize = 4_KB; const ui64 shardBlockCount = 1024; const ui64 shardAllocationUnit = shardBlockCount * blockSize; const ui64 shardCount = 4; const ui64 fsSize = shardBlockCount * (shardCount - 1) + shardBlockCount / 2; config.SetAutomaticShardCreationEnabled(true); config.SetShardAllocationUnit(shardAllocationUnit); config.SetMaxShardCount(1024); const TString fsId = "test"; TTestEnv env({}, config); ui32 nodeIdx = env.AddDynamicNode(); TServiceClient service(env.GetRuntime(), nodeIdx); service.CreateFileStore(fsId, fsSize); WaitForTabletStart(service); THeaders headers = service.InitSession(fsId, "client"); TVector nodeIds; nodeIds.push_back( service .CreateNode( headers, TCreateNodeArgs::Directory(RootNodeId, "dir")) ->Record.GetNode() .GetId()); for (ui64 i = 0; i < shardCount; ++i) { nodeIds.push_back( service .CreateNode( headers, TCreateNodeArgs::File( RootNodeId, TStringBuilder() << "file_" << i)) ->Record.GetNode() .GetId()); } NProtoPrivate::TUnsafeCreateHandleRequest request; for (ui64 i = 0; i < nodeIds.size(); i++) { const ui32 shardNo = ExtractShardNo(nodeIds[i]); UNIT_ASSERT_EQUAL(i, shardNo); const ui64 handle = (ShardedId(i + 1, shardNo) | (16ULL << 48)); request.SetFileSystemId( shardNo == 0 ? fsId : TStringBuilder() << fsId << "_s" << shardNo); request.SetSessionId(headers.SessionId); request.SetHandle(handle); request.SetNodeId(nodeIds[i]); request.SetCommitId(2048 + i); request.SetFlags(ProtoFlag(NProto::TCreateHandleRequest::E_READ)); TString buf; google::protobuf::util::MessageToJsonString(request, &buf); const auto actionResponse = service.ExecuteAction("unsafecreatehandle", buf); NProtoPrivate::TUnsafeCreateHandleResponse response; auto status = google::protobuf::util::JsonStringToMessage( actionResponse->Record.GetOutput(), &response); UNIT_ASSERT(status.ok()); } env.GetRuntime().AdvanceCurrentTime(TDuration::Seconds(15)); NActors::TDispatchOptions options; // As the main tablet is restarted after ConfigureShards we need to // dispatch tablet count + 1 (== shard count + 2) EvUpdateCounters // events options.FinalEvents.emplace_back(TDispatchOptions::TFinalEventCondition( TEvIndexTabletPrivate::EvUpdateCounters, shardCount + 2)); env.GetRuntime().DispatchEvents(options); TTestRegistryVisitor visitor; env.GetRegistry()->Visit(TInstant::Zero(), visitor); // We should not create new handles that use seventh byte visitor.ValidateExpectedCounters( {{{{"sensor", "SevenBytesHandlesCount"}, {"filesystem", "test"}}, 1}, {{{"sensor", "SevenBytesHandlesCount"}, {"filesystem", "test_s1"}}, 1}, {{{"sensor", "SevenBytesHandlesCount"}, {"filesystem", "test_s2"}}, 1}, {{{"sensor", "SevenBytesHandlesCount"}, {"filesystem", "test_s3"}}, 1}, {{{"sensor", "SevenBytesHandlesCount"}, {"filesystem", "test_s4"}}, 1}}); // Try to resize to more than MaxOneByteShardCount shards. const ui64 newShardCount = MaxOneByteShardCount + 8; const ui64 newFsSize = shardBlockCount * (newShardCount - 1) + shardBlockCount / 2; service.ResizeFileStore(fsId, newFsSize); WaitForTabletStart(service); // The filesystem should be resized to have 255 (MaxOneByteShardCount) // shards and a critical event should be raised. const auto counters = env.GetCounters()->FindSubgroup("component", "service"); UNIT_ASSERT(counters); const auto counter = counters->GetCounter( "AppCriticalEvents/TryingToUseTwoBytesShardNoWithObsoleteHandles"); UNIT_ASSERT_EQUAL(1, counter->GetAtomic()); const auto response = GetStorageStats(service, fsId); const auto& stats = response.GetStats(); UNIT_ASSERT_EQUAL(MaxOneByteShardCount, stats.GetShardStats().size()); } SERVICE_TEST(ShouldReturnListNodesMissingFromShardsWithUnsafeFlag) { TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); auto headers = service.InitSession(fsConfig.FsId, "client"); ui64 dirNodeId = service.CreateNode( headers, TCreateNodeArgs::Directory(RootNodeId, "dir") )->Record.GetNode().GetId(); TString lastNodeName = "file2"; auto node1Id = service.CreateNode( headers, TCreateNodeArgs::File( dirNodeId, "file1"))->Record.GetNode().GetId(); // we assume that file2 will be created in shard 1 since shards are distributed in round-robin service.CreateNode( headers, TCreateNodeArgs::File( dirNodeId, lastNodeName))->Record.GetNode().GetId(); auto listNodesResponse = service.ListNodes( headers, fsConfig.FsId, dirNodeId)->Record; UNIT_ASSERT_VALUES_EQUAL(2, listNodesResponse.NodesSize()); const auto shardId = fsConfig.Shard2Id; const auto shardHeaders = service.InitSession(shardId, "client1"); const auto nodeNameInShard = service.ListNodes( shardHeaders, shardId, RootNodeId)->Record.GetNames(0); // remove node from shard directly service.UnlinkNode( shardHeaders, RootNodeId, nodeNameInShard); // with the flag, the missing in the shard node should be returned with id=0 auto listNodesResponseWithFlag = service.ListNodes( headers, fsConfig.FsId, dirNodeId, true)->Record; UNIT_ASSERT_VALUES_EQUAL(2, listNodesResponseWithFlag.NodesSize()); TSet expectedNodeIds = {node1Id, InvalidNodeId}; TSet foundNodeIds; for (const auto & node: listNodesResponseWithFlag.GetNodes()) { foundNodeIds.insert(node.GetId()); } UNIT_ASSERT_VALUES_EQUAL(expectedNodeIds, foundNodeIds); // without the flag, the node missing in the shard should be skipped auto listNodesResponseRegular = service.ListNodes( headers, fsConfig.FsId, dirNodeId, false)->Record; UNIT_ASSERT_VALUES_EQUAL(1, listNodesResponseRegular.NodesSize()); } void DoShouldShardedFileSystemHitNodesCountLimit( NProto::TStorageConfig config, const bool strictFileSystemSizeEnforcementEnabled, const bool fanoutStatsCollectionInShardsDisabled) { const ui64 blockSize = 4_KB; const ui64 shardBlockCount = 512; const ui64 shardAllocationUnit = shardBlockCount * blockSize; const ui64 shardCount = 2; const ui64 fsSize = shardBlockCount * (shardCount - 1) + shardBlockCount / 2; const ui32 sizeToNodeRatio = 64_KB; // Strict and non-strict filesystems enforce the allowed number of nodes // differently. Strict systems use the total filesystem size, while in // non-strict systems the limit is enforced per shard, based on the size // of each individual shard. const ui64 filesPerFs = strictFileSystemSizeEnforcementEnabled ? (fsSize * blockSize) / sizeToNodeRatio : (2 * shardAllocationUnit) / sizeToNodeRatio; config.SetAutomaticShardCreationEnabled(true); config.SetStrictFileSystemSizeEnforcementEnabled( strictFileSystemSizeEnforcementEnabled); config.SetShardAllocationUnit(shardAllocationUnit); config.SetAutomaticallyCreatedShardSize(shardAllocationUnit); config.SetSizeToNodesRatio(sizeToNodeRatio); config.SetDefaultNodesLimit(filesPerFs / 2); config.SetFanoutStatsCollectionInShardsDisabled( fanoutStatsCollectionInShardsDisabled); TShardedFileSystemConfig fsConfig = { .ShardBlockCount = shardBlockCount, .MainFsBlockCount = fsSize }; TTestEnv env({}, config); ui32 nodeIdx = env.AddDynamicNode(); TServiceClient service(env.GetRuntime(), nodeIdx); TFileSystemInfo fsInfo = CreateFileSystem(service, fsConfig); THeaders headers = service.InitSession(fsConfig.FsId, "client"); // Create maximum allowed number of files. for (ui64 i = 0; i < filesPerFs; ++i) { service.CreateNode( headers, TCreateNodeArgs::File( RootNodeId, TStringBuilder() << "file_" << i)); } // Update statistics to have correct AggregateNodesCount // in the main FS and all the shards. if (strictFileSystemSizeEnforcementEnabled) { auto updateCounters = [&](TActorId actorId) { using TRequest = TEvIndexTabletPrivate::TEvUpdateCounters; env.GetRuntime().Send( new IEventHandle( actorId, // recipient TActorId(), // sender new TRequest(), 0, // flags 0), nodeIdx); }; updateCounters(fsInfo.MainTabletActorId); updateCounters(fsInfo.Shard1ActorId); updateCounters(fsInfo.Shard2ActorId); NActors::TDispatchOptions options; options.FinalEvents.emplace_back( TDispatchOptions::TFinalEventCondition( TEvIndexTabletPrivate::EvUpdateCounters, shardCount + 1)); env.GetRuntime().DispatchEvents(options); } const ui32 response = strictFileSystemSizeEnforcementEnabled ? E_FS_NOSPC : S_OK; service.SendCreateNodeRequest( headers, TCreateNodeArgs::File(RootNodeId, "too_many_files")); UNIT_ASSERT_EQUAL( response, service.RecvCreateNodeResponse()->GetStatus()); UNIT_ASSERT_EQUAL( 0, env.GetCounters() ->FindSubgroup("component", "service") ->GetCounter("AppCriticalEvents/ReceivedNodeOpErrorFromShard") ->GetAtomic()); service.SendCreateHandleRequest( headers, fsConfig.FsId, RootNodeId, "too_many_files", TCreateHandleArgs::CREATE); UNIT_ASSERT_EQUAL( response, service.RecvCreateHandleResponse()->GetStatus()); UNIT_ASSERT_EQUAL( 0, env.GetCounters() ->FindSubgroup("component", "service") ->GetCounter("AppCriticalEvents/ReceivedNodeOpErrorFromShard") ->GetAtomic()); } SERVICE_TEST(ShouldShardedFileSystemHitNodesCountLimit) { DoShouldShardedFileSystemHitNodesCountLimit(config, false, false); DoShouldShardedFileSystemHitNodesCountLimit(config, true, false); DoShouldShardedFileSystemHitNodesCountLimit(config, true, true); } SERVICE_TEST(ShouldNotUpdateShardBalancerInTabletZombieState) { // This test verifies the fix for the following crash: // 1. TEvAggregateStatsCompleted was sent. // 2. The filesystem was resized, // TIndexTabletActor::ExecuteTx_ConfigureShards was called, // and ShardBalancer was recreated with a new number of shards. // 3. TEvAggregateStatsCompleted was handled, and UpdateShardBalancer // was called with the old number of shards, // which caused Y_ABORT_UNLESS(stats.size() == Metas.size()) TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); // filter out TEvents::TEvPoisonPill in order not to kill the main // tablet bool poisonPillDropped = false; bool shardsConfigured = false; env.GetRuntime().SetEventFilter( [&](auto& runtime, TAutoPtr& event) { Y_UNUSED(runtime); switch (event->GetTypeRewrite()) { case TEvIndexTablet::EvConfigureShardsRequest: { shardsConfigured = true; break; } case TEvents::TSystem::PoisonPill: { if (shardsConfigured) { poisonPillDropped = true; return true; } } } return false; }); service.ResizeFileStore(fsConfig.FsId, fsConfig.MainFsBlockCount * 2); env.GetRuntime().SetEventFilter( &TTestActorRuntimeBase::DefaultFilterFunc); using TCompletion = TEvIndexTabletPrivate::TEvAggregateStatsCompleted; NProtoPrivate::TStorageStats statsForTablet; statsForTablet.AddShardStats()->SetShardId(fsConfig.Shard1Id); statsForTablet.AddShardStats()->SetShardId(fsConfig.Shard2Id); auto completion = std::make_unique( NProto::TError(), std::move(statsForTablet), TInstant()); env.GetRuntime().Send( new IEventHandle( fsInfo.MainTabletActorId, // recipient TActorId(), // sender completion.release(), 0, // flags 0), nodeIdx); } SERVICE_TEST(ShouldSwitchOnDirectoryCreationInShardsForExistingFilesystem) { config.SetDirectoryCreationInShardsEnabled(false); TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); auto headers = service.InitSession(fsConfig.FsId, "client"); // // dir0_1 and dir0_2 - managed by the main tablet. // ui64 dir0_1Id = service.CreateNode( headers, TCreateNodeArgs::Directory(RootNodeId, "dir0_1") )->Record.GetNode().GetId(); UNIT_ASSERT_VALUES_EQUAL(0, ExtractShardNo(dir0_1Id)); ui64 dir0_2Id = service.CreateNode( headers, TCreateNodeArgs::Directory(RootNodeId, "dir0_2") )->Record.GetNode().GetId(); UNIT_ASSERT_VALUES_EQUAL(0, ExtractShardNo(dir0_2Id)); // // file1 - managed by a shard. // subdir1 - managed by the main tablet. // auto node1Id = service.CreateNode( headers, TCreateNodeArgs::File(dir0_1Id, "file1") )->Record.GetNode().GetId(); UNIT_ASSERT_VALUES_UNEQUAL(0, ExtractShardNo(node1Id)); auto subdir1Id = service.CreateNode( headers, TCreateNodeArgs::Directory(dir0_1Id, "subdir1") )->Record.GetNode().GetId(); UNIT_ASSERT_VALUES_EQUAL(0, ExtractShardNo(subdir1Id)); // // Forcing directory creation in shards. // service.ResizeFileStore( fsConfig.FsId, fsConfig.MainFsBlockCount, false /* force */, 0 /* shardCount */, true /* enableStrictSizeMode */, true /* directoryCreationInShards */, true /* forceDirectoryCreationInShards */); // ResizeFileStore triggers tablet suicide (shard reconfiguration), // need to re-establish the session after restart WaitForTabletStart(service); headers = service.InitSession(fsConfig.FsId, "client"); // // dir2, dir3, subdir2 - managed by shards. // ui64 dir2Id = service.CreateNode( headers, TCreateNodeArgs::Directory(RootNodeId, "dir2") )->Record.GetNode().GetId(); UNIT_ASSERT_VALUES_UNEQUAL(0, ExtractShardNo(dir2Id)); ui64 dir3Id = service.CreateNode( headers, TCreateNodeArgs::Directory(dir2Id, "dir3") )->Record.GetNode().GetId(); UNIT_ASSERT_VALUES_UNEQUAL(0, ExtractShardNo(dir3Id)); auto subdir2Id = service.CreateNode( headers, TCreateNodeArgs::Directory(dir0_1Id, "subdir2") )->Record.GetNode().GetId(); UNIT_ASSERT_VALUES_UNEQUAL(0, ExtractShardNo(subdir2Id)); // // Trying to move dir0_1 into dir2 - should fail because the // RootNodeId/dir0_1 NodeRef is not external. // service.SendRenameNodeRequest( headers, RootNodeId, "dir0_1", dir2Id, "dir0_1", 0 /* flags */); { auto renameResponse = service.RecvRenameNodeResponse(); UNIT_ASSERT_VALUES_EQUAL_C( E_FS_XDEV, renameResponse->GetError().GetCode(), renameResponse->GetError().GetMessage()); } // // Trying to rename subdir1 to subdir2 - should succeed. // service.SendRenameNodeRequest( headers, dir0_1Id, "subdir1", dir0_1Id, "subdir2", 0 /* flags */); { auto renameResponse = service.RecvRenameNodeResponse(); UNIT_ASSERT_VALUES_EQUAL_C( S_OK, renameResponse->GetError().GetCode(), renameResponse->GetError().GetMessage()); } // // Moving file1 to dir2 should be possible. // service.SendRenameNodeRequest( headers, dir0_1Id, "file1", dir2Id, "file1", 0 /* flags */); { auto renameResponse = service.RecvRenameNodeResponse(); UNIT_ASSERT_VALUES_EQUAL_C( S_OK, renameResponse->GetError().GetCode(), renameResponse->GetError().GetMessage()); } // // Moving dir3 into dir0_1 should be possible as well. // service.SendRenameNodeRequest( headers, dir2Id, "dir3", dir0_1Id, "dir3", 0 /* flags */); { auto renameResponse = service.RecvRenameNodeResponse(); UNIT_ASSERT_VALUES_EQUAL_C( S_OK, renameResponse->GetError().GetCode(), renameResponse->GetError().GetMessage()); } // // Moving dir0_1 into dir0_2 should be possible as well because even // though their node refs are local, both of them are managed by the // main tablet. // service.SendRenameNodeRequest( headers, RootNodeId, "dir0_1", dir0_2Id, "dir0_1", 0 /* flags */); { auto renameResponse = service.RecvRenameNodeResponse(); UNIT_ASSERT_VALUES_EQUAL_C( S_OK, renameResponse->GetError().GetCode(), renameResponse->GetError().GetMessage()); } // // Verifying that the resulting directory structure is correct. // auto listNodesResponse = service.ListNodes( headers, fsConfig.FsId, RootNodeId)->Record; UNIT_ASSERT_VALUES_EQUAL(2, listNodesResponse.NamesSize()); UNIT_ASSERT_VALUES_EQUAL(2, listNodesResponse.NodesSize()); UNIT_ASSERT_VALUES_EQUAL("dir0_2", listNodesResponse.GetNames(0)); UNIT_ASSERT_VALUES_EQUAL( dir0_2Id, listNodesResponse.GetNodes(0).GetId()); UNIT_ASSERT_VALUES_EQUAL("dir2", listNodesResponse.GetNames(1)); UNIT_ASSERT_VALUES_EQUAL( dir2Id, listNodesResponse.GetNodes(1).GetId()); listNodesResponse = service.ListNodes( headers, fsConfig.FsId, dir0_2Id)->Record; UNIT_ASSERT_VALUES_EQUAL(1, listNodesResponse.NamesSize()); UNIT_ASSERT_VALUES_EQUAL(1, listNodesResponse.NodesSize()); UNIT_ASSERT_VALUES_EQUAL("dir0_1", listNodesResponse.GetNames(0)); UNIT_ASSERT_VALUES_EQUAL( dir0_1Id, listNodesResponse.GetNodes(0).GetId()); listNodesResponse = service.ListNodes( headers, fsConfig.FsId, dir0_1Id)->Record; UNIT_ASSERT_VALUES_EQUAL(2, listNodesResponse.NamesSize()); UNIT_ASSERT_VALUES_EQUAL(2, listNodesResponse.NodesSize()); UNIT_ASSERT_VALUES_EQUAL("dir3", listNodesResponse.GetNames(0)); UNIT_ASSERT_VALUES_EQUAL( dir3Id, listNodesResponse.GetNodes(0).GetId()); UNIT_ASSERT_VALUES_EQUAL("subdir2", listNodesResponse.GetNames(1)); UNIT_ASSERT_VALUES_EQUAL( subdir1Id, listNodesResponse.GetNodes(1).GetId()); listNodesResponse = service.ListNodes( headers, fsConfig.FsId, dir3Id)->Record; UNIT_ASSERT_VALUES_EQUAL(0, listNodesResponse.NamesSize()); UNIT_ASSERT_VALUES_EQUAL(0, listNodesResponse.NodesSize()); listNodesResponse = service.ListNodes( headers, fsConfig.FsId, dir2Id)->Record; UNIT_ASSERT_VALUES_EQUAL(1, listNodesResponse.NamesSize()); UNIT_ASSERT_VALUES_EQUAL(1, listNodesResponse.NodesSize()); UNIT_ASSERT_VALUES_EQUAL("file1", listNodesResponse.GetNames(0)); UNIT_ASSERT_VALUES_EQUAL( node1Id, listNodesResponse.GetNodes(0).GetId()); } SERVICE_TEST(ShouldLockNodeRefWhileCreatingHandle) { config.SetMultiTabletForwardingEnabled(true); config.SetStrictFileSystemSizeEnforcementEnabled(true); config.SetShardBalancerPolicy(NProto::SBP_ROUND_ROBIN); TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); THeaders headers = service.InitSession(fsConfig.FsId, "client"); TAutoPtr createNodeInShard; TString name; bool intercept = true; env.GetRuntime().SetEventFilter( [&] (auto& runtime, TAutoPtr& e) { Y_UNUSED(runtime); if (e->GetTypeRewrite() == TEvService::EvCreateNodeRequest) { const auto* msg = e->Get(); if (intercept && msg->Record.GetFileSystemId() == fsConfig.Shard1Id) { name = msg->Record.GetName(); createNodeInShard = e; return true; } } return false; }); ui64 requestId = 111; service.SendCreateHandleRequest( headers, fsConfig.FsId, RootNodeId, "file1", TCreateHandleArgs::CREATE, "", requestId); ui32 iterations = 0; while (!createNodeInShard && iterations++ < 100) { env.GetRuntime().DispatchEvents({}, TDuration::MilliSeconds(50)); } UNIT_ASSERT(createNodeInShard); // At this point a NodeRef is created in the main filesystem but a node // is not created in the shard. The NodeRef in the main filesystem is // locked. intercept = false; service.SendCreateHandleRequest( headers, fsConfig.FsId, RootNodeId, "file1", TCreateHandleArgs::CREATE, "", ++requestId); // An attempt to create a handle or to unlink the node should fail as // the NodeRef is locked. const auto createHandleResponse = service.RecvCreateHandleResponse(); UNIT_ASSERT_VALUES_EQUAL_C( E_REJECTED, createHandleResponse->GetError().GetCode(), createHandleResponse->GetError().GetMessage()); service.SendUnlinkNodeRequest(headers, RootNodeId, "file1"); const auto unlinkNodeResponse = service.RecvUnlinkNodeResponse(); UNIT_ASSERT_VALUES_EQUAL_C( E_REJECTED, unlinkNodeResponse->GetError().GetCode(), unlinkNodeResponse->GetError().GetMessage()); // Sending stolen createNodeInShard event should create a node in the // shard and unlock the NodeRef. env.GetRuntime().Send(createNodeInShard.Release(), nodeIdx); auto createHandleResponse1 = service.RecvCreateHandleResponse(); UNIT_ASSERT_VALUES_EQUAL_C( S_OK, createHandleResponse1->GetStatus(), createHandleResponse1->GetErrorReason()); const auto nodeId1 = createHandleResponse1->Record.GetNodeAttr().GetId(); UNIT_ASSERT_VALUES_EQUAL(1, ExtractShardNo(nodeId1)); // As the NodeRef is unlocked now we should be able to create a handle auto createHandleResponse2 = service.CreateHandle( headers, fsConfig.FsId, RootNodeId, "file1", TCreateHandleArgs::RDWR, "", ++requestId); const auto nodeId2 = createHandleResponse2->Record.GetNodeAttr().GetId(); UNIT_ASSERT_VALUES_EQUAL(1, ExtractShardNo(nodeId2)); TTestRegistryVisitor visitor; auto registry = env.GetRegistry(); registry->Visit(TInstant::Zero(), visitor); visitor.ValidateExpectedCounters( {{{{"sensor", "NodeExistsWhileCreatingInShardCount"}, {"filesystem", "test"}}, 0}, {{{"sensor", "NodeExistsWhileCreatingInShardCount"}, {"filesystem", "test_s1"}}, 0}, {{{"sensor", "NodeExistsWhileCreatingInShardCount"}, {"filesystem", "test_s2"}}, 0}}); } SERVICE_TEST(ShouldLockNodeRefWhileCreatingNode) { config.SetMultiTabletForwardingEnabled(true); config.SetStrictFileSystemSizeEnforcementEnabled(true); config.SetShardBalancerPolicy(NProto::SBP_ROUND_ROBIN); TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); THeaders headers = service.InitSession(fsConfig.FsId, "client"); TAutoPtr createNodeInShard; TString name; bool intercept = true; env.GetRuntime().SetEventFilter( [&] (auto& runtime, TAutoPtr& e) { Y_UNUSED(runtime); if (e->GetTypeRewrite() == TEvService::EvCreateNodeRequest) { const auto* msg = e->Get(); if (intercept && msg->Record.GetFileSystemId() == fsConfig.Shard1Id) { name = msg->Record.GetName(); createNodeInShard = e; return true; } } return false; }); service.SendCreateNodeRequest( headers, TCreateNodeArgs::File(RootNodeId, "file1")); ui32 iterations = 0; while (!createNodeInShard && iterations++ < 100) { env.GetRuntime().DispatchEvents({}, TDuration::MilliSeconds(50)); } UNIT_ASSERT(createNodeInShard); // At this point a NodeRef is created in the main filesystem but a node // is not created in the shard. ui64 requestId = 111; intercept = false; service.SendCreateHandleRequest( headers, fsConfig.FsId, RootNodeId, "file1", TCreateHandleArgs::CREATE, "", ++requestId); // An attempt to create a handle or to unlink the node should fail as // the NodeRef is locked. const auto createHandleResponse = service.RecvCreateHandleResponse(); UNIT_ASSERT_VALUES_EQUAL_C( E_REJECTED, createHandleResponse->GetError().GetCode(), createHandleResponse->GetError().GetMessage()); service.SendUnlinkNodeRequest(headers, RootNodeId, "file1"); const auto unlinkNodeResponse = service.RecvUnlinkNodeResponse(); UNIT_ASSERT_VALUES_EQUAL_C( E_REJECTED, unlinkNodeResponse->GetError().GetCode(), unlinkNodeResponse->GetError().GetMessage()); // Sending stolen createNodeInShard event should create a node in the // shard and unlock the NodeRef. env.GetRuntime().Send(createNodeInShard.Release(), nodeIdx); const auto createNodeResponse = service.RecvCreateNodeResponse(); UNIT_ASSERT_VALUES_EQUAL_C( S_OK, createNodeResponse->GetStatus(), createNodeResponse->GetErrorReason()); const auto nodeId = createNodeResponse->Record.GetNode().GetId(); UNIT_ASSERT_VALUES_EQUAL(1, ExtractShardNo(nodeId)); // The NodeRef is unlocked now and we should be able to create a handle. const auto createHandleResponse1 = service.CreateHandle( headers, fsConfig.FsId, RootNodeId, "file1", TCreateHandleArgs::RDWR, "", ++requestId); UNIT_ASSERT_VALUES_EQUAL_C( S_OK, createHandleResponse1->GetStatus(), createHandleResponse1->GetErrorReason()); const auto nodeId1 = createHandleResponse1->Record.GetNodeAttr().GetId(); UNIT_ASSERT_VALUES_EQUAL(1, ExtractShardNo(nodeId1)); TTestRegistryVisitor visitor; auto registry = env.GetRegistry(); registry->Visit(TInstant::Zero(), visitor); visitor.ValidateExpectedCounters( {{{{"sensor", "NodeExistsWhileCreatingInShardCount"}, {"filesystem", "test"}}, 0}, {{{"sensor", "NodeExistsWhileCreatingInShardCount"}, {"filesystem", "test_s1"}}, 0}, {{{"sensor", "NodeExistsWhileCreatingInShardCount"}, {"filesystem", "test_s2"}}, 0}}); } SERVICE_TEST(ShouldReplayCreateNodeInShardFromOpLog) { config.SetMultiTabletForwardingEnabled(true); config.SetStrictFileSystemSizeEnforcementEnabled(true); TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); THeaders headers = service.InitSession(fsConfig.FsId, "client"); TAutoPtr createNodeInShard; TString name; bool intercept = true; env.GetRuntime().SetEventFilter( [&] (auto& runtime, TAutoPtr& e) { Y_UNUSED(runtime); if (e->GetTypeRewrite() == TEvService::EvCreateNodeRequest) { const auto* msg = e->Get(); if (intercept && msg->Record.GetFileSystemId() == fsConfig.Shard1Id) { name = msg->Record.GetName(); createNodeInShard = e; return true; } } return false; }); ui64 requestId = 111; service.SendCreateHandleRequest( headers, fsConfig.FsId, RootNodeId, "file1", TCreateHandleArgs::CREATE, "", requestId); ui32 iterations = 0; while (!createNodeInShard && iterations++ < 100) { env.GetRuntime().DispatchEvents({}, TDuration::MilliSeconds(50)); } UNIT_ASSERT(createNodeInShard); intercept = false; TIndexTabletClient tablet( env.GetRuntime(), nodeIdx, fsInfo.MainTabletId); tablet.RebootTablet(); const auto createHandleResponse = service.RecvCreateHandleResponse(); UNIT_ASSERT_VALUES_EQUAL_C( E_REJECTED, createHandleResponse->GetError().GetCode(), createHandleResponse->GetError().GetMessage()); headers = service.InitSession(fsConfig.FsId, "client"); // After the tablet is restarted "file1" creation is completed and // CreateHandle should be successful. const auto createHandleResponse1 = service.CreateHandle( headers, fsConfig.FsId, RootNodeId, "file1", TCreateHandleArgs::RDWR, "", ++requestId); UNIT_ASSERT_VALUES_EQUAL_C( S_OK, createHandleResponse1->GetError().GetCode(), createHandleResponse1->GetError().GetMessage()); TTestRegistryVisitor visitor; auto registry = env.GetRegistry(); registry->Visit(TInstant::Zero(), visitor); visitor.ValidateExpectedCounters( {{{{"sensor", "ReplayedCreateNodeInShardRequestsCount"}, {"filesystem", "test"}}, 1}, {{{"sensor", "ReplayedCreateNodeInShardRequestsCount"}, {"filesystem", "test_s1"}}, 0}, {{{"sensor", "ReplayedCreateNodeInShardRequestsCount"}, {"filesystem", "test_s2"}}, 0}, {{{"sensor", "NodeExistsWhileCreatingInShardCount"}, {"filesystem", "test"}}, 0}, {{{"sensor", "NodeExistsWhileCreatingInShardCount"}, {"filesystem", "test_s1"}}, 0}, {{{"sensor", "NodeExistsWhileCreatingInShardCount"}, {"filesystem", "test_s2"}}, 0}}); } void DoShouldReadAndWriteCompressedNodeRef( NProto::TStorageConfig & config, bool directoryCreationInShards) { const ui64 blockSize = 4_KB; const ui64 shardBlockCount = 1024; const ui64 shardAllocationUnit = shardBlockCount * blockSize; const ui64 shardCount = 4; const ui64 fsSize = shardBlockCount * (shardCount - 1) + shardBlockCount / 2; config.SetMultiTabletForwardingEnabled(true); config.SetStrictFileSystemSizeEnforcementEnabled(true); config.SetAutomaticShardCreationEnabled(true); config.SetShardAllocationUnit(shardAllocationUnit); config.SetDirectoryCreationInShardsEnabled(directoryCreationInShards); const TString fsId = "testfilesystem-filesystem1234id"; TTestEnv env({}, config); ui32 nodeIdx = env.AddDynamicNode(); TServiceClient service(env.GetRuntime(), nodeIdx); service.CreateFileStore(fsId, fsSize); WaitForTabletStart(service); auto headers = service.InitSession(fsId, "client"); constexpr ui32 filesCount = 8; ui32 fileNum = 0; ui32 dirNum = 0; ui32 unsafeNodeNum = 0; TVector dirs; dirs.push_back(RootNodeId); auto setCompress = [&](const bool compress) { // Change config in the main filesystem SetCompressNodeRef(fsId, service, compress); // Change config in shards if (directoryCreationInShards) { for (ui64 shardNo = 1; shardNo < shardCount + 1; ++shardNo) { SetCompressNodeRef( TStringBuilder() << fsId << "_s" << shardNo, service, compress); } } }; auto unsafeCreateNodeRef = [&](const NProtoPrivate::TUnsafeCreateNodeRefRequest& request, const bool isOk = true) { TString buf; google::protobuf::util::MessageToJsonString(request, &buf); const auto actionResponse = service.SendAndRecvExecuteAction("unsafecreatenoderef", buf); bool succeded = SUCCEEDED(actionResponse->GetStatus()); UNIT_ASSERT_EQUAL_C( isOk, succeded, actionResponse->GetErrorReason()); NProtoPrivate::TUnsafeCreateNodeRefResponse response; auto status = google::protobuf::util::JsonStringToMessage( actionResponse->Record.GetOutput(), &response); UNIT_ASSERT(status.ok()); }; auto unsafeGetNodeRef = [&](const NProtoPrivate::TUnsafeGetNodeRefRequest& request, const bool isOk = true) { TString buf; google::protobuf::util::MessageToJsonString(request, &buf); const auto actionResponse = service.SendAndRecvExecuteAction("unsafegetnoderef", buf); bool succeded = SUCCEEDED(actionResponse->GetStatus()); UNIT_ASSERT_EQUAL_C( isOk, succeded, actionResponse->GetErrorReason()); NProtoPrivate::TUnsafeGetNodeRefResponse response; auto status = google::protobuf::util::JsonStringToMessage( actionResponse->Record.GetOutput(), &response); UNIT_ASSERT(status.ok()); return response; }; // For each mode in compressionModes, we create several files. Then, for // every mode greater than or equal to the one in which the files were // created, we create handles to all of those files. for (bool writeCompression: {false, true}) { setCompress(writeCompression); // Create a directory for the the next batch of files const TString dirName = TStringBuilder() << "dir" << dirNum++; const ui64 dirId = service.CreateNode( headers, TCreateNodeArgs::Directory(RootNodeId, dirName)) ->Record.GetNode().GetId(); dirs.push_back(dirId); ui32 shardNo = directoryCreationInShards ? ((dirs.size() - 2) % shardCount) + 1 : 0; UNIT_ASSERT_VALUES_EQUAL(shardNo, ExtractShardNo(dirs.back())); // Create a batch of files TVector> files; for (ui32 i = 0; i < filesCount; ++i) { const TString fileName = TStringBuilder() << "file_" << fileNum++; files.push_back({(i % 2) ? RootNodeId : dirId, fileName}); service.CreateNode( headers, TCreateNodeArgs::File( files.back().first, files.back().second)); } // Unsafe create nodeRef in the shard _s1 that references the main // filesystem. NProtoPrivate::TUnsafeCreateNodeRefRequest createNodeRefRequest; createNodeRefRequest.SetFileSystemId(fsId + "_s1"); createNodeRefRequest.SetParentId(2); createNodeRefRequest.SetName( TStringBuilder() << "UnsafeNodeName" << unsafeNodeNum); createNodeRefRequest.SetChildId(unsafeNodeNum++); createNodeRefRequest.SetShardId(fsId); createNodeRefRequest.SetShardNodeName(CreateGuidAsString()); unsafeCreateNodeRef(createNodeRefRequest); for (bool readCompression: {false, true}) { setCompress(readCompression); // Check that we can resolve all the node in each dir for (const auto& dir: dirs) { service.ListNodes(headers, fsId, dir); } // Check that we can resolve ther node ref by creating handles for (const auto& file: files) { service.CreateHandle( headers, fsId, file.first, file.second, TCreateHandleArgs::RDWR); } // Check that we can read the refernce to the main FS NProtoPrivate::TUnsafeGetNodeRefRequest getNodeRefRequest; getNodeRefRequest.SetFileSystemId( createNodeRefRequest.GetFileSystemId()); getNodeRefRequest.SetParentId( createNodeRefRequest.GetParentId()); getNodeRefRequest.SetName(createNodeRefRequest.GetName()); const auto getNodeRefResponse = unsafeGetNodeRef(getNodeRefRequest); UNIT_ASSERT_EQUAL( createNodeRefRequest.GetShardId(), getNodeRefResponse.GetShardId()); UNIT_ASSERT_EQUAL( createNodeRefRequest.GetShardNodeName(), getNodeRefResponse.GetShardNodeName()); } } // Check that attempts to write and read malformed ShardId/ShardNodeName // raise critical events. setCompress(false); NProtoPrivate::TUnsafeCreateNodeRefRequest createNodeRefRequest; createNodeRefRequest.SetFileSystemId(fsId); createNodeRefRequest.SetParentId(12123123); createNodeRefRequest.SetName( TStringBuilder() << "UnsafeNodeName" << unsafeNodeNum); createNodeRefRequest.SetChildId(unsafeNodeNum++); createNodeRefRequest.SetShardId("\x03\x01\xFF"); createNodeRefRequest.SetShardNodeName("123456789ABCDEF"); unsafeCreateNodeRef(createNodeRefRequest); NProtoPrivate::TUnsafeGetNodeRefRequest getNodeRefRequest; getNodeRefRequest.SetFileSystemId( createNodeRefRequest.GetFileSystemId()); getNodeRefRequest.SetParentId( createNodeRefRequest.GetParentId()); getNodeRefRequest.SetName(createNodeRefRequest.GetName()); const auto getNodeRefResponse = unsafeGetNodeRef(getNodeRefRequest, false); setCompress(true); createNodeRefRequest.SetFileSystemId(fsId); createNodeRefRequest.SetParentId(12123123); createNodeRefRequest.SetName( TStringBuilder() << "UnsafeNodeName" << unsafeNodeNum); createNodeRefRequest.SetChildId(unsafeNodeNum++); createNodeRefRequest.SetShardId(fsId + "_sabcd"); createNodeRefRequest.SetShardNodeName("123456789ABCDEF"); unsafeCreateNodeRef(createNodeRefRequest); UNIT_ASSERT_EQUAL(1, env.GetCounters() ->FindSubgroup("component", "service") ->GetCounter("AppCriticalEvents/MalformedShardNodeRef") ->GetAtomic()); UNIT_ASSERT_EQUAL(1, env.GetCounters() ->FindSubgroup("component", "service") ->GetCounter("AppCriticalEvents/MalformedEncodedShardNodeRef") ->GetAtomic()); } SERVICE_TEST(ShouldReadAndWriteCompressedNodeRef) { DoShouldReadAndWriteCompressedNodeRef(config, false); DoShouldReadAndWriteCompressedNodeRef(config, true); } SERVICE_TEST(ShouldValidateFilesystemIdDuringCreation) { TTestEnv env({}, config); ui32 nodeIdx = env.AddDynamicNode(); TServiceClient service(env.GetRuntime(), nodeIdx); for (const auto& badFsId: {"\rwq2e123123", ""}) { service.SendCreateFileStoreRequest(badFsId, 1024); auto response = service.RecvCreateFileStoreResponse(); UNIT_ASSERT_VALUES_EQUAL_C( E_ARGUMENT, response->GetError().GetCode(), response->GetErrorReason()); } for (const char* goodFsId: {"nfs_shared", "_s", "fs_s1234_s", "testfs-f_s98723rhfhnkJKid", "fs_s1234", "abcd\x03gefg"}) { service.SendCreateFileStoreRequest(goodFsId, 1024); auto response = service.RecvCreateFileStoreResponse(); UNIT_ASSERT_VALUES_EQUAL_C( S_OK, response->GetError().GetCode(), response->GetErrorReason()); } } SERVICE_TEST(ShouldSetCompressNodeRefFlag) { config.SetEnableNodeRefCompression(true); { // Create a filesystem with 2 shards. TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); // Check that the CompressNodeRef flag is set for the main // filesystem and all shards. const auto mainFsTopology = GetFileSystemTopology(service, fsConfig.FsId); UNIT_ASSERT(mainFsTopology.GetCompressNodeRef()); const auto shard1Topology = GetFileSystemTopology(service, fsConfig.Shard1Id); UNIT_ASSERT(shard1Topology.GetCompressNodeRef()); const auto shard2Topology = GetFileSystemTopology(service, fsConfig.Shard2Id); UNIT_ASSERT(shard2Topology.GetCompressNodeRef()); // Resize the filesystem to have 3 shards, check that the new shard // has CompressNodeRef. service.ResizeFileStore( fsConfig.FsId, fsConfig.MainFsBlockCount * 3 / 2); WaitForTabletStart(service); const auto shard3Topology = GetFileSystemTopology(service, fsConfig.FsId + "_s3"); UNIT_ASSERT(shard3Topology.GetCompressNodeRef()); } { // Create a filesystem without shards. TTestEnv env({}, config); ui32 nodeIdx = env.AddDynamicNode(); TServiceClient service(env.GetRuntime(), nodeIdx); const ui64 fsSize = 1000; const ui64 shardsCount = 2; const TString fsId = "test"; service.CreateFileStore(fsId, fsSize); const auto mainFsTopology = GetFileSystemTopology(service, fsId); UNIT_ASSERT(mainFsTopology.GetCompressNodeRef()); // Resize the filesystem to have 2 shards, check the new shards have // CompressNodeRef. service .ResizeFileStore(fsId, fsSize, false /* force */, shardsCount); WaitForTabletStart(service); const auto shard1Topology = GetFileSystemTopology(service, fsId + "_s1"); UNIT_ASSERT(shard1Topology.GetCompressNodeRef()); const auto shard2Topology = GetFileSystemTopology(service, fsId + "_s2"); UNIT_ASSERT(shard2Topology.GetCompressNodeRef()); } } SERVICE_TEST(ShouldUseCacheInStatFileStore) { constexpr ui64 cacheTTLMs = 1000; config.SetStatFileStoreCacheTTL(cacheTTLMs); // Create a filesystem with 2 shards. TShardedFileSystemConfig fsConfig; CREATE_ENV_AND_SHARDED_FILESYSTEM(); auto headers = service.InitSession(fsConfig.FsId, "client"); auto writeToFile = [&](const TString& name, const TString& data) { auto createNodeResponse = service .CreateNode( headers, TCreateNodeArgs::File(RootNodeId, name)) ->Record; const auto nodeId = createNodeResponse.GetNode().GetId(); const ui64 handle = service .CreateHandle( headers, fsConfig.FsId, nodeId, "", TCreateHandleArgs::RDWR) ->Record.GetHandle(); service.WriteData( headers, headers.FileSystemId, nodeId, handle, 0, data); }; constexpr ui64 data1Size = 256_KB; constexpr ui64 data2Size = 128_KB; writeToFile("file1", GenerateValidateData(data1Size)); writeToFile("file2", GenerateValidateData(data2Size)); env.GetRuntime().AdvanceCurrentTime( TDuration::MilliSeconds(cacheTTLMs * 2)); // Get the latest stats as cache is older than the cache TTL. auto stats = service.StatFileStore(headers, fsConfig.FsId)->Record.GetStats(); UNIT_ASSERT_EQUAL(2, stats.GetUsedNodesCount()); UNIT_ASSERT_EQUAL( (data1Size + data2Size) / 4_KB, stats.GetUsedBlocksCount()); writeToFile("file3", GenerateValidateData(data1Size)); writeToFile("file4", GenerateValidateData(data2Size)); // Get stats again, it should be the same as the cache TTL = 1s. stats = service.StatFileStore(headers, fsConfig.FsId)->Record.GetStats(); UNIT_ASSERT_EQUAL(2, stats.GetUsedNodesCount()); UNIT_ASSERT_EQUAL( (data1Size + data2Size) / 4_KB, stats.GetUsedBlocksCount()); env.GetRuntime().AdvanceCurrentTime( TDuration::MilliSeconds(cacheTTLMs * 2)); // Finally get updated stats. stats = service.StatFileStore(headers, fsConfig.FsId)->Record.GetStats(); UNIT_ASSERT_EQUAL(4, stats.GetUsedNodesCount()); UNIT_ASSERT_EQUAL( 2 * (data1Size + data2Size) / 4_KB, stats.GetUsedBlocksCount()); } } } // namespace NCloud::NFileStore::NStorage