diff --git a/cloud/filestore/apps/client/lib/execute_action.cpp b/cloud/filestore/apps/client/lib/execute_action.cpp index 1732b880d7f..9e72ace2340 100644 --- a/cloud/filestore/apps/client/lib/execute_action.cpp +++ b/cloud/filestore/apps/client/lib/execute_action.cpp @@ -1,8 +1,9 @@ #include "command.h" -#include "util/stream/file.h" #include +#include + namespace NCloud::NFileStore::NClient { namespace { diff --git a/cloud/filestore/apps/client/lib/factory.cpp b/cloud/filestore/apps/client/lib/factory.cpp index 0eb295564fe..8a43685955d 100644 --- a/cloud/filestore/apps/client/lib/factory.cpp +++ b/cloud/filestore/apps/client/lib/factory.cpp @@ -35,6 +35,7 @@ TCommandPtr NewDestroySessionCommand(); TCommandPtr NewStatCommand(); TCommandPtr NewSetNodeAttrCommand(); TCommandPtr NewFindGarbageCommand(); +TCommandPtr NewForcedCompactionCommand(); //////////////////////////////////////////////////////////////////////////////// @@ -50,6 +51,7 @@ static const TMap Commands = { { "destroysession", NewDestroySessionCommand }, { "executeaction", NewExecuteActionCommand }, { "findgarbage", NewFindGarbageCommand }, + { "forcedcompaction", NewForcedCompactionCommand }, { "kickendpoint", NewKickEndpointCommand }, { "listclusternodes", NewListClusterNodesCommand }, { "listendpoints", NewListEndpointsCommand }, diff --git a/cloud/filestore/apps/client/lib/forced_compaction.cpp b/cloud/filestore/apps/client/lib/forced_compaction.cpp new file mode 100644 index 00000000000..fd95b6fe2c9 --- /dev/null +++ b/cloud/filestore/apps/client/lib/forced_compaction.cpp @@ -0,0 +1,123 @@ +#include "command.h" + +#include + +#include + +#include + +namespace NCloud::NFileStore::NClient { + +namespace { + +//////////////////////////////////////////////////////////////////////////////// + +class TForcedCompactionCommand final + : public TFileStoreCommand +{ +private: + ui32 MinRangeId = 0; + +private: + template + void ExecuteAction( + const TString& action, + const TRequest& requestProto, + TResponse* responseProto) + { + auto callContext = PrepareCallContext(); + + TString input; + google::protobuf::util::MessageToJsonString(requestProto, &input); + + STORAGE_DEBUG("Reading ExecuteAction request"); + auto request = std::make_shared(); + request->SetAction(action); + request->SetInput(std::move(input)); + + STORAGE_DEBUG("Sending ExecuteAction request"); + const auto requestId = GetRequestId(*request); + auto result = WaitFor(Client->ExecuteAction( + MakeIntrusive(requestId), + std::move(request))); + + STORAGE_DEBUG("Received ExecuteAction response"); + + if (HasError(result)) { + responseProto->MutableError()->CopyFrom(result.GetError()); + return; + } + + auto parsed = google::protobuf::util::JsonStringToMessage( + result.GetOutput(), + responseProto).ok(); + + if (!parsed) { + responseProto->MutableError()->CopyFrom(MakeError( + E_FAIL, + TStringBuilder() << "failed to parse response json: " + << result.GetOutput())); + } + } + +public: + TForcedCompactionCommand() + { + Opts.AddLongOption("min-range-id", "initial compaction range id") + .RequiredArgument("NUM") + .StoreResult(&MinRangeId); + } + +public: + bool Execute() override + { + auto callContext = PrepareCallContext(); + + NProtoPrivate::TForcedOperationRequest request; + request.SetFileSystemId(FileSystemId); + request.SetOpType(NProtoPrivate::TForcedOperationRequest::E_COMPACTION); + request.SetMinRangeId(MinRangeId); + NProtoPrivate::TForcedOperationResponse response; + ExecuteAction("forcedoperation", request, &response); + CheckResponse(response); + + while (true) { + NProtoPrivate::TForcedOperationStatusRequest statusRequest; + statusRequest.SetFileSystemId(FileSystemId); + statusRequest.SetOperationId(response.GetOperationId()); + NProtoPrivate::TForcedOperationStatusResponse statusResponse; + ExecuteAction( + "forcedoperationstatus", + statusRequest, + &statusResponse); + + if (statusResponse.GetError().GetCode() == E_NOT_FOUND) { + // TODO: distinguish between finished operations and tablet + // restarts + Cout << "finished" << Endl; + break; + } + + CheckResponse(statusResponse); + + Cout << "progress: " << statusResponse.GetProcessedRangeCount() + << "/" << statusResponse.GetRangeCount() << ", last=" + << statusResponse.GetLastProcessedRangeId() << Endl; + + Sleep(TDuration::Seconds(1)); + } + + return true; + } +}; + +} // namespace + +//////////////////////////////////////////////////////////////////////////////// + +TCommandPtr NewForcedCompactionCommand() +{ + return std::make_shared(); +} + +} // namespace NCloud::NFileStore::NClient diff --git a/cloud/filestore/apps/client/lib/ya.make b/cloud/filestore/apps/client/lib/ya.make index 0d53310c66d..1f61bb6a5a9 100644 --- a/cloud/filestore/apps/client/lib/ya.make +++ b/cloud/filestore/apps/client/lib/ya.make @@ -11,6 +11,7 @@ SRCS( execute_action.cpp factory.cpp find_garbage.cpp + forced_compaction.cpp kick_endpoint.cpp list_cluster_nodes.cpp list_endpoints.cpp diff --git a/cloud/filestore/tests/client/canondata/result.json b/cloud/filestore/tests/client/canondata/result.json index 1bb59bd5bb0..882c6ab7045 100644 --- a/cloud/filestore/tests/client/canondata/result.json +++ b/cloud/filestore/tests/client/canondata/result.json @@ -11,6 +11,9 @@ "test.test_describe_sessions": { "uri": "file://test.test_describe_sessions/results.txt" }, + "test.test_forced_compaction": { + "uri": "file://test.test_forced_compaction/results.txt" + }, "test.test_large_file": { "uri": "file://test.test_large_file/results.txt" }, diff --git a/cloud/filestore/tests/client/canondata/test.test_forced_compaction/results.txt b/cloud/filestore/tests/client/canondata/test.test_forced_compaction/results.txt new file mode 100644 index 00000000000..979270fd76b --- /dev/null +++ b/cloud/filestore/tests/client/canondata/test.test_forced_compaction/results.txt @@ -0,0 +1,2 @@ +progress: 0/64, last=1177944064 +finished diff --git a/cloud/filestore/tests/client/test.py b/cloud/filestore/tests/client/test.py index 8046957bcbd..16d11ffa628 100644 --- a/cloud/filestore/tests/client/test.py +++ b/cloud/filestore/tests/client/test.py @@ -498,3 +498,34 @@ def test_large_file(): ret = common.canonical_file(results_path, local=True) return ret + + +def test_forced_compaction(): + data_file = os.path.join(common.output_path(), "data.txt") + chunk_size = 128 * 1024 + chunk = [] + for i in range(chunk_size): + chunk.append("a") + chunk_str = "".join(chunk) + with open(data_file, "w") as f: + f.write(chunk_str) + + client, results_path = __init_test() + client.create("fs0", "test_cloud", "test_folder") + + for i in range(128): + client.write( + "fs0", + "/aaa", + "--data", data_file, + "--offset", str(i * chunk_size)) + + result = client.forced_compaction("fs0").decode("utf8") + + client.destroy("fs0") + + with open(results_path, "w") as results_file: + results_file.write(result) + + ret = common.canonical_file(results_path, local=True) + return ret diff --git a/cloud/filestore/tests/python/lib/client.py b/cloud/filestore/tests/python/lib/client.py index 8bf390071fb..0d225443fc2 100644 --- a/cloud/filestore/tests/python/lib/client.py +++ b/cloud/filestore/tests/python/lib/client.py @@ -232,6 +232,14 @@ def set_node_attr(self, fs, node_id, *argv): return common.execute(cmd, env=self.__env, check_exit_code=self.__check_exit_code).stdout + def forced_compaction(self, fs): + cmd = [ + self.__binary_path, "forcedcompaction", + "--filesystem", fs, + ] + self.__cmd_opts() + + return common.execute(cmd, env=self.__env, check_exit_code=self.__check_exit_code).stdout + def execute_action(self, action, request): request_file = tempfile.NamedTemporaryFile(mode="w", delete=False) json.dump(request, request_file)