cloud/filestore/libs/diagnostics/profile_log_ut.cpp - main - githubmirror/nbs

cloud/filestore/libs/diagnostics/profile_log_ut.cpp
591 строка18 KB
Darya Frolova
issue-6751: add metric for compressed profile log frame size (#6752)
11 авг 2026, 14:02
11 авг 2026, 14:02e1efdde
#include "profile_log.h"

#include <cloud/filestore/libs/diagnostics/events/profile_events.ev.pb.h>
#include <cloud/storage/core/libs/common/scheduler_test.h>
#include <cloud/storage/core/libs/common/timer.h>

#include <library/cpp/eventlog/dumper/evlogdump.h>
#include <library/cpp/eventlog/eventlog.h>
#include <library/cpp/monlib/dynamic_counters/counters.h>
#include <library/cpp/testing/unittest/registar.h>

#include <util/folder/tempdir.h>
#include <util/generic/algorithm.h>
#include <util/string/builder.h>
#include <util/system/fstat.h>

namespace NCloud::NFileStore {

namespace {

////////////////////////////////////////////////////////////////////////////////

using TRanges =
    google::protobuf::RepeatedPtrField<NProto::TProfileLogBlockRange>;

TString RangesToString(const TRanges& ranges)
{
    TStringBuilder sb;

    for (int i = 0; i < ranges.size(); ++i) {
        if (i) {
            sb << " ";
        }

        sb << ranges[i].GetNodeId()
            << "," << ranges[i].GetHandle()
            << "," << ranges[i].GetOffset()
            << "," << ranges[i].GetBytes();
    }

    return std::move(sb);
}

TString NodeInfoToString(const NProto::TProfileLogNodeInfo& nodeInfo)
{
    return TStringBuilder() << nodeInfo.GetParentNodeId()
        << "," << nodeInfo.GetNodeName()
        << "," << nodeInfo.GetNewParentNodeId()
        << "," << nodeInfo.GetNewNodeName()
        << "," << nodeInfo.GetFlags()
        << "," << nodeInfo.GetMode()
        << "," << nodeInfo.GetNodeId()
        << "," << nodeInfo.GetHandle()
        << "," << nodeInfo.GetSize()
        << "," << nodeInfo.GetType()
        << "," << nodeInfo.GetATime()
        << "," << nodeInfo.GetMTime()
        << "," << nodeInfo.GetCTime();
}

TString LockInfoToString(const NProto::TProfileLogLockInfo& lockInfo)
{
    return TStringBuilder() << lockInfo.GetNodeId()
        << "," << lockInfo.GetHandle()
        << "," << lockInfo.GetOwner()
        << "," << lockInfo.GetOffset()
        << "," << lockInfo.GetLength()
        << "," << lockInfo.GetType()
        << "," << lockInfo.GetConflictedOwner()
        << "," << lockInfo.GetConflictedOffset()
        << "," << lockInfo.GetConflictedLength();
}

TString ListNodesInfoToString(
    const NProto::TProfileLogListNodesInfo& listNodesInfo)
{
    return TStringBuilder() << listNodesInfo.GetNodeId()
        << "," << listNodesInfo.GetMaxBytes()
        << "," << listNodesInfo.GetRequestCookie()
        << "," << listNodesInfo.GetResponseCookie()
        << "," << listNodesInfo.GetNameCount();
}

////////////////////////////////////////////////////////////////////////////////

struct TEventProcessor
    : TProtobufEventProcessor
{
    TVector<TString> FlatMessages;
    TVector<ui32> MessageCountByRecord;
    ui64 DiscardedMessageCount = 0;

    void Clear()
    {
        FlatMessages.clear();
        MessageCountByRecord.clear();
        DiscardedMessageCount = 0;
    }

    void DoProcessEvent(const TEvent* ev, IOutputStream* out) override
    {
        Y_UNUSED(out);

        const auto* message =
            dynamic_cast<const NProto::TProfileLogRecord*>(ev->GetProto());
        if (message) {
            DiscardedMessageCount += message->GetDiscardedRequestCount();

            for (const auto& r: message->GetRequests()) {
                FlatMessages.push_back(
                    TStringBuilder() << message->GetFileSystemId()
                        << "\t" << r.GetTimestampMcs()
                        << "\t" << r.GetRequestType()
                        << "\t" << r.GetDurationMcs()
                        << "\t" << r.GetErrorCode()
                );
                if (r.HasNodeInfo()) {
                    FlatMessages.back() += TStringBuilder() <<
                        "\t" << NodeInfoToString(r.GetNodeInfo());
                }
                if (r.HasLockInfo()) {
                    FlatMessages.back() += TStringBuilder() <<
                        "\t" << LockInfoToString(r.GetLockInfo());
                }
                if (r.HasListNodesInfo()) {
                    FlatMessages.back() += TStringBuilder() <<
                        "\t" << ListNodesInfoToString(r.GetListNodesInfo());
                }
                if (r.RangesSize() > 0) {
                    FlatMessages.back() += TStringBuilder() <<
                        "\t" << RangesToString(r.GetRanges());
                }
            }

            MessageCountByRecord.push_back(message->GetRequests().size());
        }
    }
};

////////////////////////////////////////////////////////////////////////////////

using namespace NMonitoring;

struct TEnv
    : public NUnitTest::TBaseFixture
{
    const TTempDir TempDir;
    const TFsPath ProfilePath = TempDir.Path() / "profile.log";

    ITimerPtr Timer;
    std::shared_ptr<TTestScheduler> Scheduler;
    TProfileLogSettings Settings;
    IProfileLogPtr ProfileLog;
    TDynamicCountersPtr Counters;
    TEventProcessor EventProcessor;

    TEnv()
        : Timer(CreateWallClockTimer())
        , Scheduler(new TTestScheduler())
        , Settings{ProfilePath.c_str(), TDuration::Seconds(1), 0, 0}
        , ProfileLog(CreateProfileLog(Settings, Timer, Scheduler))
        , Counters(MakeIntrusive<TDynamicCounters>())
    {
        ProfileLog->RegisterCounters(*Counters);
    }

    TEnv(ui64 maxFlushRecords, ui64 maxFrameFlushRecords)
        : Timer(CreateWallClockTimer())
        , Scheduler(new TTestScheduler())
        , Settings(
              ProfilePath.c_str(),
              TDuration::Seconds(1),
              maxFlushRecords,
              maxFrameFlushRecords)
        , ProfileLog(CreateProfileLog(Settings, Timer, Scheduler))
        , Counters(MakeIntrusive<TDynamicCounters>())
    {
        ProfileLog->RegisterCounters(*Counters);
    }

    void SetUp(NUnitTest::TTestContext& /*context*/) override
    {
        ProfileLog->Start();
    }

    void TearDown(NUnitTest::TTestContext& /*context*/) override
    {
        ProfileLog->Stop();
        ProfilePath.DeleteIfExists();
    }

    void ProcessLog(bool runScheduler = true)
    {
        if (runScheduler) {
            Scheduler->RunAllScheduledTasks();
        }

        EventProcessor.Clear();
        const char* argv[] = {"foo", Settings.FilePath.c_str()};
        IterateEventLog(NEvClass::Factory(), &EventProcessor, 2, argv);
        Sort(EventProcessor.FlatMessages);
    }
};

struct TEnvWithLimits
    : public TEnv
{
    TEnvWithLimits(ui64 maxFlushRecords, ui64 maxFrameFlushRecords)
        : TEnv(maxFlushRecords, maxFrameFlushRecords)
    {
        ProfileLog->Start();
    }

    ~TEnvWithLimits() override
    {
        ProfileLog->Stop();
        ProfilePath.DeleteIfExists();
    }
};

////////////////////////////////////////////////////////////////////////////////

struct TRequestInfoBuilder
{
    NProto::TProfileLogRequestInfo R;

    auto& SetTimestamp(TInstant t)
    {
        R.SetTimestampMcs(t.MicroSeconds());
        return *this;
    }

    auto& SetDuration(TDuration d)
    {
        R.SetDurationMcs(d.MicroSeconds());
        return *this;
    }

    auto& SetRequestType(ui32 t)
    {
        R.SetRequestType(t);
        return *this;
    }

    auto& SetError(ui32 errorCode)
    {
        R.SetErrorCode(errorCode);
        return *this;
    }

    auto& AddRange(ui64 nodeId, ui64 handle, ui64 offset, ui32 size)
    {
        auto* range = R.AddRanges();
        range->SetNodeId(nodeId);
        range->SetHandle(handle);
        range->SetOffset(offset);
        range->SetBytes(size);
        return *this;
    }

    auto& AddNodeInfo(
        ui64 parentNodeId,
        TString nodeName,
        ui64 newParentNodeId,
        TString newNodeName,
        ui32 flags,
        ui32 mode,
        ui64 nodeId,
        ui64 handle,
        ui64 size,
        ui32 type)
    {
        auto nodeInfo = R.MutableNodeInfo();
        nodeInfo->SetParentNodeId(parentNodeId);
        nodeInfo->SetNodeName(std::move(nodeName));
        nodeInfo->SetNewParentNodeId(newParentNodeId);
        nodeInfo->SetNewNodeName(std::move(newNodeName));
        nodeInfo->SetFlags(flags);
        nodeInfo->SetMode(mode);
        nodeInfo->SetNodeId(nodeId);
        nodeInfo->SetHandle(handle);
        nodeInfo->SetSize(size);
        nodeInfo->SetType(type);
        return *this;
    }

    auto& AddLockInfo(
        ui64 nodeId,
        ui64 handle,
        ui64 owner,
        ui64 offset,
        ui64 length,
        ui32 type,
        ui64 conflictedOwner,
        ui64 conflictedOffset,
        ui64 conflictedLength)
    {
        auto lockInfo = R.MutableLockInfo();
        lockInfo->SetNodeId(nodeId);
        lockInfo->SetHandle(handle);
        lockInfo->SetOwner(owner);
        lockInfo->SetOffset(offset);
        lockInfo->SetLength(length);
        lockInfo->SetType(type);
        lockInfo->SetConflictedOwner(conflictedOwner);
        lockInfo->SetConflictedOffset(conflictedOffset);
        lockInfo->SetConflictedLength(conflictedLength);
        return *this;
    }

    auto& AddListNodesInfo(
        ui64 nodeId,
        ui32 maxBytes,
        TString requestCookie,
        TString responseCookie,
        ui64 nameCount)
    {
        auto listNodesInfo = R.MutableListNodesInfo();
        listNodesInfo->SetNodeId(nodeId);
        listNodesInfo->SetMaxBytes(maxBytes);
        listNodesInfo->SetRequestCookie(std::move(requestCookie));
        listNodesInfo->SetResponseCookie(std::move(responseCookie));
        listNodesInfo->SetNameCount(nameCount);
        return *this;
    }

    auto Build() const
    {
        return R;
    }
};

}   // namespace

////////////////////////////////////////////////////////////////////////////////

Y_UNIT_TEST_SUITE(TProfileLogTest)
{
    Y_UNIT_TEST_F(TestSmoke, TEnv)
    {
        for (ui32 i = 1; i <= 100'000; ++i) {
            ProfileLog->Write({
                TStringBuilder() << "fs" << (i % 10),
                TRequestInfoBuilder()
                    .SetTimestamp(TInstant::MilliSeconds(i))
                    .SetDuration(TDuration::MilliSeconds(i % 200))
                    .SetRequestType(i % 3)
                    .AddRange(i % 1'000, i % 10'000, i, i * 2)
                    .Build()
            });
        }

        ProcessLog();

        UNIT_ASSERT_VALUES_EQUAL(100'000, EventProcessor.FlatMessages.size());
    }

    Y_UNIT_TEST_F(TestDumpMessages, TEnv)
    {
        ProfileLog->Write({
            "fs1",
            TRequestInfoBuilder()
                .SetTimestamp(TInstant::Seconds(1))
                .SetDuration(TDuration::MilliSeconds(100))
                .SetRequestType(1)
                .SetError(0)
                .AddRange(111, 23, 200, 10)
                .AddRange(111, 23, 300, 5)
                .Build()
        });

        ProfileLog->Write({
            "fs1",
            TRequestInfoBuilder()
                .SetTimestamp(TInstant::Seconds(2))
                .SetDuration(TDuration::MilliSeconds(200))
                .SetRequestType(3)
                .SetError(1)
                .AddRange(6000, 42, 500, 20)
                .Build()
        });

        ProfileLog->Write({
            "fs2",
            TRequestInfoBuilder()
                .SetTimestamp(TInstant::Seconds(3))
                .SetDuration(TDuration::MilliSeconds(400))
                .SetRequestType(7)
                .SetError(0)
                .Build()
        });

        ProfileLog->Write({
            "fs2",
            TRequestInfoBuilder()
                .SetTimestamp(TInstant::Seconds(4))
                .SetDuration(TDuration::MilliSeconds(800))
                .SetRequestType(11)
                .SetError(1)
                .AddNodeInfo(1, "node", 2, "new_node", 3, 7, 12, 123, 32, 2)
                .Build()
        });

        ProfileLog->Write({
            "fs3",
            TRequestInfoBuilder()
                .SetTimestamp(TInstant::Seconds(5))
                .SetDuration(TDuration::MilliSeconds(1'600))
                .SetRequestType(11)
                .SetError(1)
                .AddLockInfo(1, 2, 3, 7, 12, 123, 32, 33, 34)
                .Build()
        });

        ProcessLog();

        UNIT_ASSERT_VALUES_EQUAL(5, EventProcessor.FlatMessages.size());
        UNIT_ASSERT_VALUES_EQUAL(
            "fs1\t1000000\t1\t100000\t0\t111,23,200,10 111,23,300,5",
            EventProcessor.FlatMessages[0]
        );
        UNIT_ASSERT_VALUES_EQUAL(
            "fs1\t2000000\t3\t200000\t1\t6000,42,500,20",
            EventProcessor.FlatMessages[1]
        );
        UNIT_ASSERT_VALUES_EQUAL(
            "fs2\t3000000\t7\t400000\t0",
            EventProcessor.FlatMessages[2]
        );
        UNIT_ASSERT_VALUES_EQUAL(
            "fs2\t4000000\t11\t800000\t1\t1,node,2,new_node,3,7,12,123,32,2,0,0,0",
            EventProcessor.FlatMessages[3]
        );
        UNIT_ASSERT_VALUES_EQUAL(
            "fs3\t5000000\t11\t1600000\t1\t1,2,3,7,12,123,32,33,34",
            EventProcessor.FlatMessages[4]
        );

        ProfileLog->Write({
            "fs3",
            TRequestInfoBuilder()
                .SetTimestamp(TInstant::Seconds(6))
                .SetDuration(TDuration::MilliSeconds(300))
                .SetRequestType(4)
                .SetError(0)
                .Build()
        });

        ProcessLog();

        UNIT_ASSERT_VALUES_EQUAL(6, EventProcessor.FlatMessages.size());
        UNIT_ASSERT_VALUES_EQUAL(
            "fs3\t6000000\t4\t300000\t0",
            EventProcessor.FlatMessages[5]
        );

        // Test flush on destruct
        ProfileLog->Write(
            {"fs3",
             TRequestInfoBuilder()
                 .SetTimestamp(TInstant::Seconds(9))
                 .SetDuration(TDuration::MilliSeconds(100))
                 .SetRequestType(4)
                 .SetError(0)
                 .Build()});

        ProfileLog = CreateProfileLogStub();
        ProcessLog(false);

        UNIT_ASSERT_VALUES_EQUAL(7, EventProcessor.FlatMessages.size());
        UNIT_ASSERT_VALUES_EQUAL(
            "fs3\t9000000\t4\t100000\t0",
            EventProcessor.FlatMessages[6]);

        // Testing counters
        auto pg = Counters->GetSubgroup("counters", "profile_log");
        auto requests = pg->FindCounter("Count");
        auto bytes = pg->FindCounter("RequestBytes");
        auto compressedFrameBytes = pg->FindCounter("CompressedFrameBytes");
        auto flushes = pg->FindCounter("FlushCount");
        auto discards = pg->FindCounter("DiscardCount");
        UNIT_ASSERT(requests);
        UNIT_ASSERT_VALUES_EQUAL(7, requests->Val());
        UNIT_ASSERT(bytes);
        UNIT_ASSERT_VALUES_EQUAL(177, bytes->Val());
        UNIT_ASSERT(compressedFrameBytes);
        UNIT_ASSERT_VALUES_EQUAL(
            GetFileLength(ProfilePath.c_str()),
            compressedFrameBytes->Val());
        UNIT_ASSERT(flushes);
        UNIT_ASSERT_VALUES_EQUAL(3, flushes->Val());
        UNIT_ASSERT(discards);
        UNIT_ASSERT_VALUES_EQUAL(0, discards->Val());
    }

    Y_UNIT_TEST_F(TestDumpListNodesInfo, TEnv)
    {
        ProfileLog->Write({
            "fs-list",
            TRequestInfoBuilder()
                .SetTimestamp(TInstant::Seconds(7))
                .SetDuration(TDuration::MilliSeconds(125))
                .SetRequestType(27)
                .SetError(0)
                .AddListNodesInfo(
                    42,
                    4096,
                    "request-cookie",
                    "response-cookie",
                    3)
                .Build()
        });

        ProcessLog();

        UNIT_ASSERT_VALUES_EQUAL(1, EventProcessor.FlatMessages.size());
        UNIT_ASSERT_VALUES_EQUAL(
            "fs-list\t7000000\t27\t125000\t0\t42,4096,request-cookie,response-"
            "cookie,3",
            EventProcessor.FlatMessages[0]);
    }

    Y_UNIT_TEST(TestRecordLimits)
    {
        ui64 maxFlushRecords = 10;
        ui64 maxFrameFlushRecords = 2;
        ui64 extraDiscardedRecords = 7;

        TEnvWithLimits env{maxFlushRecords, maxFrameFlushRecords};

        for (ui64 i = 0; i < maxFlushRecords + extraDiscardedRecords; i++) {
            env.ProfileLog->Write(
                {"fs1",
                 TRequestInfoBuilder()
                     .SetTimestamp(TInstant::Seconds(1))
                     .SetDuration(TDuration::MilliSeconds(100))
                     .SetRequestType(1)
                     .SetError(0)
                     .AddRange(111, 23, 200, 10)
                     .AddRange(111, 23, 300, 5)
                     .Build()});
        }

        env.ProcessLog();

        // check all messages flushed
        UNIT_ASSERT_VALUES_EQUAL(
            maxFlushRecords,
            env.EventProcessor.FlatMessages.size());

        // check maxFlushRecords / maxFrameFlushRecords records were written
        UNIT_ASSERT_VALUES_EQUAL(
            maxFlushRecords / maxFrameFlushRecords,
            env.EventProcessor.MessageCountByRecord.size());

        UNIT_ASSERT_VALUES_EQUAL(
            extraDiscardedRecords,
            env.EventProcessor.DiscardedMessageCount);

        // check each record contains maxFrameFlushRecords records
        for (ui64 i = 0; i < maxFlushRecords / maxFrameFlushRecords; ++i) {
            UNIT_ASSERT_VALUES_EQUAL(
                maxFrameFlushRecords,
                env.EventProcessor.MessageCountByRecord[i]);
        }

        // Testing counters
        auto pg = env.Counters->GetSubgroup("counters", "profile_log");
        auto requests = pg->FindCounter("Count");
        auto bytes = pg->FindCounter("RequestBytes");
        auto compressedFrameBytes = pg->FindCounter("CompressedFrameBytes");
        auto flushes = pg->FindCounter("FlushCount");
        auto discards = pg->FindCounter("DiscardCount");
        UNIT_ASSERT(requests);
        UNIT_ASSERT_VALUES_EQUAL(
            maxFlushRecords + extraDiscardedRecords,
            requests->Val());
        UNIT_ASSERT(bytes);
        UNIT_ASSERT_VALUES_EQUAL(578, bytes->Val());
        UNIT_ASSERT(compressedFrameBytes);
        UNIT_ASSERT_VALUES_EQUAL(
            GetFileLength(env.ProfilePath.c_str()),
            compressedFrameBytes->Val());
        UNIT_ASSERT(flushes);
        UNIT_ASSERT_VALUES_EQUAL(1, flushes->Val());
        UNIT_ASSERT(discards);
        UNIT_ASSERT_VALUES_EQUAL(extraDiscardedRecords, discards->Val());
    }
}

}   // namespace NCloud::NFileStore