[Filestore] Implement a tool which replays filestore-vhost's profile log - GRPC player

cloud/filestore/tools/testing/loadtest/lib/request.h

@@ -79,4 +79,11 @@ IRequestGeneratorPtr CreateReplayRequestGeneratorFs(
     TString filesystemId,
     NProto::THeaders headers);
 
+IRequestGeneratorPtr CreateReplayRequestGeneratorGRPC(
+    NProto::TReplaySpec spec,
+    ILoggingServicePtr logging,
+    NClient::ISessionPtr session,
+    TString filesystemId,
+    NProto::THeaders headers);
+
 }   // namespace NCloud::NFileStore::NLoadTest

cloud/filestore/tools/testing/loadtest/lib/request_replay.cpp

@@ -18,9 +18,10 @@ IReplayRequestGenerator::IReplayRequestGenerator(
         NProto::TReplaySpec spec,
         ILoggingServicePtr logging,
         NClient::ISessionPtr session,
-        TString /*filesystemId*/,
+        TString filesystemId,
         NProto::THeaders headers)
     : Spec(std::move(spec))
+    , TargetFilesystemId(std::move(filesystemId))
     , Headers(std::move(headers))
     , Session(std::move(session))
 {
@@ -29,8 +30,10 @@ IReplayRequestGenerator::IReplayRequestGenerator(
     NEventLog::TOptions options;
     options.FileName = Spec.GetFileName();
 
-    // Sort eventlog items by timestamp
-    options.SetForceStrongOrdering(true);
+    if (const auto sleep = Spec.GetMaxSleepMcs()) {
+        MaxSleepMcs = sleep;
+    }
+
     CurrentEvent = CreateIterator(options);
 }
 
@@ -54,9 +57,8 @@ bool IReplayRequestGenerator::ShouldFailOnError()
 
 void IReplayRequestGenerator::Advance()
 {
-    for (EventPtr = CurrentEvent->Next(); EventPtr;
-         EventPtr = CurrentEvent->Next())
-    {
+    while (EventPtr = CurrentEvent->Next()) {
+        ++EventsProcessed;
         MessagePtr = dynamic_cast<const NProto::TProfileLogRecord*>(
             EventPtr->GetProto());
 
@@ -68,12 +70,23 @@ void IReplayRequestGenerator::Advance()
         if (!Spec.GetFileSystemIdFilter().empty() &&
             fileSystemId != Spec.GetFileSystemIdFilter())
         {
+            ++EventsSkipped;
             STORAGE_DEBUG(
-                "Skipped event with FileSystemId=%s",
-                fileSystemId.c_str());
+                "Skipped events=%zu FileSystemId=%s Filter=%s",
+                EventsSkipped,
+                fileSystemId.c_str(),
+                Spec.GetFileSystemIdFilter().c_str());
             continue;
         }
 
+
+        if (TargetFilesystemId.empty() && !fileSystemId.empty()) {
+            TargetFilesystemId = fileSystemId;
+            STORAGE_INFO(
+                "Using FileSystemId from profile log %s",
+                TargetFilesystemId.c_str());
+        }
+
         EventMessageNumber = MessagePtr->GetRequests().size();
         return;
     }
@@ -87,7 +100,7 @@ TFuture<TCompletedRequest> IReplayRequestGenerator::ProcessRequest(
         case EFileStoreRequest::ReadData:
             return DoReadData(request);
         case EFileStoreRequest::WriteData:
-            return DoWrite(request);
+            return DoWriteData(request);
         case EFileStoreRequest::CreateNode:
             return DoCreateNode(request);
         case EFileStoreRequest::RenameNode:
@@ -114,6 +127,7 @@ TFuture<TCompletedRequest> IReplayRequestGenerator::ProcessRequest(
         case EFileStoreRequest::GenerateBlobIds:
         case EFileStoreRequest::PingSession:
         case EFileStoreRequest::Ping:
+        case EFileStoreRequest::DescribeData:
             return {};
 
         default:
@@ -128,7 +142,7 @@ TFuture<TCompletedRequest> IReplayRequestGenerator::ProcessRequest(
             break;
     }
 
-    STORAGE_INFO(
+    STORAGE_DEBUG(
         "Uninmplemented action=%u %s",
         action,
         RequestName(request.GetRequestType()).c_str());
@@ -148,37 +162,59 @@ IReplayRequestGenerator::ExecuteNextRequest()
             continue;
         }
 
-        for (; EventMessageNumber > 0;) {
+        while (EventMessageNumber > 0) {
             NProto::TProfileLogRequestInfo request =
                 MessagePtr->GetRequests()[--EventMessageNumber];
             {
-                auto timediff = (request.GetTimestampMcs() - TimestampMcs) *
+                ++MessagesProcessed;
+                ui64 timediff = (request.GetTimestampMcs() - TimestampMcs) *
                                 Spec.GetTimeScale();
                 TimestampMcs = request.GetTimestampMcs();
                 if (timediff > MaxSleepMcs) {
+                    STORAGE_DEBUG(
+                        "Ignore too long timediff=%lu MaxSleepMcs=%lu ",
+                        timediff,
+                        MaxSleepMcs);
+
                     timediff = 0;
                 }
 
                 const auto current = TInstant::Now();
+
+                if (NextStatusAt <= current) {
+                    NextStatusAt = current + StatusEverySeconds;
+                    STORAGE_INFO(
+                        "Current event=%zu skipped=%zu Msg=%zd TotalMsg=%zu "
+                        "Sleeps=%f",
+                        EventsProcessed,
+                        EventsSkipped,
+                        EventMessageNumber,
+                        MessagesProcessed,
+                        Sleeps)
+                }
+
                 auto diff = current - Started;
 
                 if (timediff > diff.MicroSeconds()) {
                     auto sleep =
                         TDuration::MicroSeconds(timediff - diff.MicroSeconds());
                     STORAGE_DEBUG(
-                        "Sleep=%lu timediff=%f diff=%lu",
+                        "Sleep=%lu timediff=%lu diff=%lu",
                         sleep.MicroSeconds(),
                         timediff,
                         diff.MicroSeconds());
 
                     Sleep(sleep);
+                    Sleeps += sleep.SecondsFloat();
                 }
 
                 Started = current;
             }
 
             STORAGE_DEBUG(
-                "Processing message n=%d typename=%s type=%d name=%s data=%s",
+                "Event=%zu Msg=%zd: Processing typename=%s type=%d name=%s "
+                "data=%s",
+                EventsProcessed,
                 EventMessageNumber,
                 request.GetTypeName().c_str(),
                 request.GetRequestType(),
@@ -193,7 +229,7 @@ IReplayRequestGenerator::ExecuteNextRequest()
     }
 
     STORAGE_INFO(
-        "Profile log finished n=%d hasPtr=%d",
+        "Profile log finished n=%zd hasPtr=%d",
         EventMessageNumber,
         !!EventPtr);
 

cloud/filestore/tools/testing/loadtest/lib/request_replay.h

@@ -19,21 +19,28 @@ class IReplayRequestGenerator: public IRequestGenerator
 protected:
     const ::NCloud::NFileStore::NProto::TReplaySpec Spec;
     TLog Log;
+    TString TargetFilesystemId;   // Only for GRPC
     TString FileSystemIdFilter;
     const ::NCloud::NFileStore::NProto::THeaders Headers;
     NClient::ISessionPtr Session;
-
+    ssize_t EventMessageNumber = 0;
     ui64 TimestampMcs = 0;
     TInstant Started;
 
     // Do not sleep too much if timestamps in log are broken
-    constexpr static auto MaxSleepMcs = 1000000;
+    ui64 MaxSleepMcs = 1000000;
 
 private:
     THolder<NEventLog::IIterator> CurrentEvent;
     TConstEventPtr EventPtr;
-    int EventMessageNumber = 0;
     const NProto::TProfileLogRecord* MessagePtr{};
+    size_t EventsProcessed = 0;
+    size_t EventsSkipped = 0;
+    double Sleeps = 0;
+    size_t MessagesProcessed = 0;
+    static constexpr TDuration StatusEverySeconds = TDuration::Seconds(10);
+    TInstant NextStatusAt = TInstant::Now() + StatusEverySeconds;
+
     TFuture<TCompletedRequest> ProcessRequest(
         const NProto::TProfileLogRequestInfo& request);
 
@@ -58,7 +65,7 @@ class IReplayRequestGenerator: public IRequestGenerator
     virtual TFuture<TCompletedRequest> DoReadData(
         const NCloud::NFileStore::NProto::
             TProfileLogRequestInfo& /*unused*/) = 0;
-    virtual TFuture<TCompletedRequest> DoWrite(
+    virtual TFuture<TCompletedRequest> DoWriteData(
         const NCloud::NFileStore::NProto::
             TProfileLogRequestInfo& /*unused*/) = 0;
     virtual TFuture<TCompletedRequest> DoCreateNode(