issue-2137: filestore-client forcedcompaction command (#2383)

* issue-2137: filestore-client forcedcompaction command

* issue-2137: filestore-client forcedcompaction command - forgot canondata

cloud/filestore/apps/client/lib/execute_action.cpp

@@ -1,8 +1,9 @@
 #include "command.h"
-#include "util/stream/file.h"
 
 #include <library/cpp/json/json_reader.h>
 
+#include <util/stream/file.h>
+
 namespace NCloud::NFileStore::NClient {
 
 namespace {

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<TString, TFactoryFunc> Commands = {
     { "destroysession", NewDestroySessionCommand },
     { "executeaction", NewExecuteActionCommand },
     { "findgarbage", NewFindGarbageCommand },
+    { "forcedcompaction", NewForcedCompactionCommand },
     { "kickendpoint", NewKickEndpointCommand },
     { "listclusternodes", NewListClusterNodesCommand },
     { "listendpoints", NewListEndpointsCommand },

cloud/filestore/apps/client/lib/forced_compaction.cpp

@@ -0,0 +1,123 @@
+#include "command.h"
+
+#include <cloud/filestore/private/api/protos/tablet.pb.h>
+
+#include <library/cpp/json/json_reader.h>
+
+#include <google/protobuf/util/json_util.h>
+
+namespace NCloud::NFileStore::NClient {
+
+namespace {
+
+////////////////////////////////////////////////////////////////////////////////
+
+class TForcedCompactionCommand final
+    : public TFileStoreCommand
+{
+private:
+    ui32 MinRangeId = 0;
+
+private:
+    template <typename TRequest, typename TResponse>
+    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<NProto::TExecuteActionRequest>();
+        request->SetAction(action);
+        request->SetInput(std::move(input));
+
+        STORAGE_DEBUG("Sending ExecuteAction request");
+        const auto requestId = GetRequestId(*request);
+        auto result = WaitFor(Client->ExecuteAction(
+            MakeIntrusive<TCallContext>(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<TForcedCompactionCommand>();
+}
+
+}   // namespace NCloud::NFileStore::NClient

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

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"
     },

cloud/filestore/tests/client/canondata/test.test_forced_compaction/results.txt

@@ -0,0 +1,2 @@
+progress: 0/64, last=1177944064
+finished

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

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)