Skip to content

Add stats for grouped limiter and caches #20738

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 9 commits into
base: main
Choose a base branch
from
+233 −125
Open

Add stats for grouped limiter and caches #20738

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
1957ac0
Add consumer and stats for grouped limiter
Vladilen Jul 7, 2025
320a636
Blob and General cache WIP
Vladilen Jul 7, 2025
30facd3
WIP tests with no general cache double registration
Vladilen Jul 8, 2025
f989e0a
Fixed missed counters
Vladilen Jul 8, 2025
94162e1
Some cleaning
Vladilen Jul 8, 2025
d8fbd40
Fix review notes
Vladilen Jul 8, 2025
91f1db7
Fix review notes 2
Vladilen Jul 8, 2025
9f51600
Fix review notes 3
Vladilen Jul 8, 2025
3e2ba33
Fix GML consumption
Vladilen Jul 8, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 9 additions & 3 deletions ydb/core/base/memory_controller_iface.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,10 @@ namespace NKikimr::NMemory {
enum class EMemoryConsumerKind {
SharedCache,
MemTable,
ScanGroupedMemoryLimiter,
CompGroupedMemoryLimiter,
BlobCache,
DataAccessorCache
};

struct IMemoryConsumer : public TThrRefBase {
Expand All @@ -17,7 +21,7 @@ enum EEvMemory {
EvConsumerRegister = EventSpaceBegin(TKikimrEvents::ES_MEMORY),
EvConsumerRegistered,
EvConsumerLimit,

EvMemTableRegister,
EvMemTableRegistered,
EvMemTableCompact,
Expand Down Expand Up @@ -47,10 +51,12 @@ struct TEvConsumerRegistered : public TEventLocal<TEvConsumerRegistered, EvConsu

struct TEvConsumerLimit : public TEventLocal<TEvConsumerLimit, EvConsumerLimit> {
ui64 LimitBytes;
std::optional<ui64> HardLimitBytes;

TEvConsumerLimit(ui64 limitBytes)
TEvConsumerLimit(ui64 limitBytes, std::optional<ui64> hardLimitBytes = std::nullopt)
: LimitBytes(limitBytes)
{}
, HardLimitBytes(std::move(hardLimitBytes)) {
}
};

struct TEvMemTableRegister : public TEventLocal<TEvMemTableRegister, EvMemTableRegister> {
Expand Down
28 changes: 23 additions & 5 deletions ydb/core/base/ut/memory_stats_ut.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,7 +69,7 @@ Y_UNIT_TEST_SUITE (TMemoryStatsAggregator) {
aggregator.Add(stats1, "host1");
aggregator.Add(stats2, "host2");
aggregator.Add(stats3, "host3");

TMemoryStats aggregated = aggregator.Aggregate();

Cerr << aggregated.ShortDebugString() << Endl;
Expand All @@ -96,7 +96,7 @@ Y_UNIT_TEST_SUITE (TMemoryStatsAggregator) {
aggregator.Add(stats1, "host1");
aggregator.Add(stats2, "host2");
aggregator.Add(stats3, "host3");

TMemoryStats aggregated = aggregator.Aggregate();

Cerr << aggregated.ShortDebugString() << Endl;
Expand All @@ -119,7 +119,7 @@ Y_UNIT_TEST_SUITE (TMemoryStatsAggregator) {
aggregator.Add(stats1, "host");
aggregator.Add(stats2, "host");
aggregator.Add(stats3, "host");

TMemoryStats aggregated = aggregator.Aggregate();

Cerr << aggregated.ShortDebugString() << Endl;
Expand All @@ -146,7 +146,7 @@ Y_UNIT_TEST_SUITE (TMemoryStatsAggregator) {
aggregator.Add(stats1, "host");
aggregator.Add(stats2, "host");
aggregator.Add(stats3, "host");

TMemoryStats aggregated = aggregator.Aggregate();

Cerr << aggregated.ShortDebugString() << Endl;
Expand All @@ -169,14 +169,32 @@ Y_UNIT_TEST_SUITE (TMemoryStatsAggregator) {
aggregator.Add(stats1, "host1");
aggregator.Add(stats2, "host1");
aggregator.Add(stats3, "host2");

TMemoryStats aggregated = aggregator.Aggregate();

Cerr << aggregated.ShortDebugString() << Endl;

UNIT_ASSERT_VALUES_EQUAL(aggregated.ShortDebugString(),
"AnonRss: 36 CGroupLimit: 66 MemTotal: 65 MemAvailable: 85 AllocatedMemory: 156 AllocatorCachesMemory: 186 HardLimit: 145 SoftLimit: 165 TargetUtilization: 185 ExternalConsumption: 194 SharedCacheConsumption: 336 SharedCacheLimit: 366 MemTableConsumption: 396 MemTableLimit: 426 QueryExecutionConsumption: 456 QueryExecutionLimit: 486");
}

Y_UNIT_TEST(ColumnShard_Single) {
TMemoryStatsAggregator aggregator;

TMemoryStats stats;
stats.SetColumnTablesReadExecutionConsumption(1);
stats.SetColumnTablesReadExecutionLimit(2);
stats.SetColumnTablesCompactionConsumption(3);
stats.SetColumnTablesCompactionLimit(4);
stats.SetColumnTablesCacheConsumption(5);
stats.SetColumnTablesCacheLimit(6);

aggregator.Add(stats, "host");

TMemoryStats aggregated = aggregator.Aggregate();

UNIT_ASSERT_VALUES_EQUAL(aggregated.ShortDebugString(), stats.ShortDebugString());
}
}

}
116 changes: 64 additions & 52 deletions ydb/core/memory_controller/memory_controller.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,8 +16,6 @@
#include <ydb/core/tx/columnshard/blob_cache.h>
#include <ydb/core/tx/columnshard/common/limits.h>
#include <ydb/core/tx/columnshard/data_accessor/cache_policy/policy.h>
#include <ydb/core/tx/limiter/grouped_memory/usage/events.h>
#include <ydb/core/tx/limiter/grouped_memory/usage/service.h>
#include <ydb/library/actors/core/actor_bootstrapped.h>
#include <ydb/library/actors/core/log.h>
#include <ydb/library/actors/core/process_stats.h>
Expand Down Expand Up @@ -86,6 +84,7 @@ struct TConsumerState {
ui64 MinBytes = 0;
ui64 MaxBytes = 0;
bool CanZeroLimit = false;
std::optional<ui64> ExactLimit;

TConsumerState(const TMemoryConsumer& consumer)
: Kind(consumer.Kind)
Expand Down Expand Up @@ -266,21 +265,31 @@ class TMemoryController : public TActorBootstrapped<TMemoryController> {

ui64 consumersLimitBytes = 0;
for (const auto& consumer : consumers) {
ui64 limitBytes = consumer.GetLimit(coefficient);
if (resultingConsumersConsumption + otherConsumption + externalConsumption > softLimitBytes && consumer.CanZeroLimit) {
limitBytes = SafeDiff(limitBytes, resultingConsumersConsumption + otherConsumption + externalConsumption - softLimitBytes);
const bool isExactLimitConsumer = consumer.ExactLimit.has_value();
ui64 limitBytes;
if (isExactLimitConsumer) {
limitBytes = consumer.ExactLimit.value();
} else {
limitBytes = consumer.GetLimit(coefficient);
if (resultingConsumersConsumption + otherConsumption + externalConsumption > softLimitBytes && consumer.CanZeroLimit) {
limitBytes = SafeDiff(limitBytes, resultingConsumersConsumption + otherConsumption + externalConsumption - softLimitBytes);
}
}
consumersLimitBytes += limitBytes;

LOG_INFO_S(ctx, NKikimrServices::MEMORY_CONTROLLER, "Consumer " << consumer.Kind << " state:"
<< " Consumption: " << HumanReadableBytes(consumer.Consumption) << " Limit: " << HumanReadableBytes(limitBytes)
<< " Min: " << HumanReadableBytes(consumer.MinBytes) << " Max: " << HumanReadableBytes(consumer.MaxBytes));
auto& counters = GetConsumerCounters(consumer.Kind);
auto& counters = GetConsumerCounters(consumer.Kind, !isExactLimitConsumer);
counters.Consumption->Set(consumer.Consumption);
counters.Reservation->Set(SafeDiff(limitBytes, consumer.Consumption));
counters.LimitBytes->Set(limitBytes);
counters.LimitMinBytes->Set(consumer.MinBytes);
counters.LimitMaxBytes->Set(consumer.MaxBytes);
if (counters.LimitMinBytes) {
counters.LimitMinBytes->Set(consumer.MinBytes);
}
if (counters.LimitMaxBytes) {
counters.LimitMaxBytes->Set(consumer.MaxBytes);
}
SetMemoryStats(consumer, memoryStats, limitBytes);

ApplyLimit(consumer, limitBytes);
Expand All @@ -289,8 +298,6 @@ class TMemoryController : public TActorBootstrapped<TMemoryController> {
Counters->GetCounter("Stats/ConsumersLimit")->Set(consumersLimitBytes);

ProcessResourceBrokerConfig(ctx, memoryStats, hardLimitBytes, activitiesLimitBytes);
ProcessGroupedMemoryLimiterConfig(ctx, memoryStats, hardLimitBytes);
ProcessCacheConfig(ctx, memoryStats, hardLimitBytes);

Send(NNodeWhiteboard::MakeNodeWhiteboardServiceId(SelfId().NodeId()), memoryStatsUpdate);

Expand Down Expand Up @@ -361,10 +368,15 @@ class TMemoryController : public TActorBootstrapped<TMemoryController> {
case EMemoryConsumerKind::MemTable:
ApplyMemTableLimit(limitBytes);
break;
default:
case EMemoryConsumerKind::SharedCache:
case EMemoryConsumerKind::BlobCache:
case EMemoryConsumerKind::DataAccessorCache:
Send(consumer.ActorId, new TEvConsumerLimit(limitBytes));
break;

case EMemoryConsumerKind::ScanGroupedMemoryLimiter:
case EMemoryConsumerKind::CompGroupedMemoryLimiter:
Send(consumer.ActorId, new TEvConsumerLimit(limitBytes * NKikimr::NOlap::TGlobalLimits::GroupedMemoryLimiterSoftLimitCoefficient, limitBytes));
break;
}
}

Expand All @@ -377,43 +389,6 @@ class TMemoryController : public TActorBootstrapped<TMemoryController> {
}
}

void ProcessGroupedMemoryLimiterConfig(const TActorContext& /*ctx*/, NKikimrMemory::TMemoryStats& /*memoryStats*/, ui64 hardLimitBytes) {
ui64 columnTablesScanLimitBytes = GetColumnTablesReadExecutionLimitBytes(Config, hardLimitBytes);
ui64 columnTablesCompactionLimitBytes = GetColumnTablesCompactionLimitBytes(Config, hardLimitBytes) *
NKikimr::NOlap::TGlobalLimits::GroupedMemoryLimiterCompactionLimitCoefficient;

ApplyGroupedMemoryLimiterConfig(columnTablesScanLimitBytes, columnTablesCompactionLimitBytes);
}

void ApplyGroupedMemoryLimiterConfig(const ui64 scanHardLimitBytes, const ui64 compactionHardLimitBytes) {
namespace NGroupedMemoryManager = ::NKikimr::NOlap::NGroupedMemoryManager;
using UpdateMemoryLimitsEv = NGroupedMemoryManager::NEvents::TEvExternal::TEvUpdateMemoryLimits;

Send(NGroupedMemoryManager::TScanMemoryLimiterOperator::MakeServiceId(SelfId().NodeId()),
new UpdateMemoryLimitsEv(scanHardLimitBytes * NKikimr::NOlap::TGlobalLimits::GroupedMemoryLimiterSoftLimitCoefficient,
scanHardLimitBytes));

Send(NGroupedMemoryManager::TCompMemoryLimiterOperator::MakeServiceId(SelfId().NodeId()),
new UpdateMemoryLimitsEv(compactionHardLimitBytes * NKikimr::NOlap::TGlobalLimits::GroupedMemoryLimiterSoftLimitCoefficient,
compactionHardLimitBytes));
}

void ProcessCacheConfig(const TActorContext& /*ctx*/, NKikimrMemory::TMemoryStats& /*memoryStats*/, ui64 hardLimitBytes) {
ui64 columnTablesBlobCacheLimitBytes = GetColumnTablesCacheLimitBytes(Config, hardLimitBytes);

ApplyCacheConfig(columnTablesBlobCacheLimitBytes);
}

void ApplyCacheConfig(const ui64 cacheLimitBytes) {
using TUpdateMaxBlobCacheDataSizeEv = NBlobCache::TEvBlobCache::TEvUpdateMaxCacheDataSize;
using TGeneralCache = NKikimr::NOlap::NDataAccessorControl::TGeneralCache;
using TGlobalLimits = NKikimr::NOlap::TGlobalLimits;

Send(NKikimr::NBlobCache::MakeBlobCacheServiceId(), new TUpdateMaxBlobCacheDataSizeEv(cacheLimitBytes * TGlobalLimits::BlobCacheCoefficient));

TGeneralCache::UpdateMaxCacheSize(cacheLimitBytes * TGlobalLimits::DataAccessorCoefficient);
}

void ProcessResourceBrokerConfig(const TActorContext& ctx, NKikimrMemory::TMemoryStats& memoryStats, ui64 hardLimitBytes,
ui64 activitiesLimitBytes) {
ui64 queryExecutionConsumption = TAlignedPagePool::GetGlobalPagePoolSize();
Expand Down Expand Up @@ -470,18 +445,21 @@ class TMemoryController : public TActorBootstrapped<TMemoryController> {
queue->MutableLimit()->SetMemory(limitBytes);
}

TConsumerCounters& GetConsumerCounters(EMemoryConsumerKind consumer) {
TConsumerCounters& GetConsumerCounters(EMemoryConsumerKind consumer, const bool minMaxRequired) {
auto it = ConsumerCounters.FindPtr(consumer);
if (it) {
return *it;
}

TCounterPtr limitMinBytes = minMaxRequired ? Counters->GetCounter(TStringBuilder() << "Consumer/" << consumer << "/LimitMin") : nullptr;
TCounterPtr limitMaxBytes = minMaxRequired ? Counters->GetCounter(TStringBuilder() << "Consumer/" << consumer << "/LimitMax") : nullptr;

return ConsumerCounters.emplace(consumer, TConsumerCounters{
Counters->GetCounter(TStringBuilder() << "Consumer/" << consumer << "/Consumption"),
Counters->GetCounter(TStringBuilder() << "Consumer/" << consumer << "/Reservation"),
Counters->GetCounter(TStringBuilder() << "Consumer/" << consumer << "/Limit"),
Counters->GetCounter(TStringBuilder() << "Consumer/" << consumer << "/LimitMin"),
Counters->GetCounter(TStringBuilder() << "Consumer/" << consumer << "/LimitMax"),
limitMinBytes,
limitMaxBytes,
}).first->second;
}

Expand All @@ -497,6 +475,23 @@ class TMemoryController : public TActorBootstrapped<TMemoryController> {
stats.SetSharedCacheLimit(limitBytes);
break;
}
case EMemoryConsumerKind::ScanGroupedMemoryLimiter: {
stats.SetColumnTablesReadExecutionConsumption(consumer.Consumption);
stats.SetColumnTablesReadExecutionLimit(limitBytes);
break;
}
case EMemoryConsumerKind::CompGroupedMemoryLimiter: {
stats.SetColumnTablesCompactionConsumption(consumer.Consumption);
stats.SetColumnTablesCompactionLimit(limitBytes);
break;
}
case EMemoryConsumerKind::DataAccessorCache:
case EMemoryConsumerKind::BlobCache: {
stats.SetColumnTablesCacheConsumption(
(stats.HasColumnTablesCacheConsumption() ? stats.GetColumnTablesCacheConsumption() : 0) + consumer.Consumption);
stats.SetColumnTablesCacheLimit((stats.HasColumnTablesCacheLimit() ? stats.GetColumnTablesCacheLimit() : 0) + limitBytes);
break;
}
default:
Y_ABORT("Unhandled consumer");
}
Expand All @@ -517,6 +512,23 @@ class TMemoryController : public TActorBootstrapped<TMemoryController> {
result.CanZeroLimit = true;
break;
}
case EMemoryConsumerKind::ScanGroupedMemoryLimiter: {
result.ExactLimit = GetColumnTablesReadExecutionLimitBytes(Config, hardLimitBytes);
break;
}
case EMemoryConsumerKind::CompGroupedMemoryLimiter: {
result.ExactLimit = GetColumnTablesCompactionLimitBytes(Config, hardLimitBytes) *
NKikimr::NOlap::TGlobalLimits::GroupedMemoryLimiterCompactionLimitCoefficient;
break;
}
case EMemoryConsumerKind::BlobCache: {
result.ExactLimit = GetColumnTablesCacheLimitBytes(Config, hardLimitBytes) * NKikimr::NOlap::TGlobalLimits::BlobCacheCoefficient;
break;
}
case EMemoryConsumerKind::DataAccessorCache: {
result.ExactLimit = GetColumnTablesCacheLimitBytes(Config, hardLimitBytes) * NKikimr::NOlap::TGlobalLimits::DataAccessorCoefficient;
break;
}
default:
Y_ABORT("Unhandled consumer");
}
Expand Down
8 changes: 8 additions & 0 deletions ydb/core/protos/memory_stats.proto
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,4 +27,12 @@ message TMemoryStats {

optional uint64 QueryExecutionConsumption = 18;
optional uint64 QueryExecutionLimit = 19;

optional uint64 ColumnTablesReadExecutionConsumption = 20;
optional uint64 ColumnTablesReadExecutionLimit = 21;
optional uint64 ColumnTablesCompactionConsumption = 22;
optional uint64 ColumnTablesCompactionLimit = 23;
optional uint64 ColumnTablesCacheConsumption = 24;
optional uint64 ColumnTablesCacheLimit = 25;

}
5 changes: 0 additions & 5 deletions ydb/core/testlib/test_client.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1207,11 +1207,6 @@ namespace Tests {
const auto aid = Runtime->Register(actor, nodeIdx, appData.UserPoolId, TMailboxType::Revolving, 0);
Runtime->RegisterService(NConveyorComposite::TServiceOperator::MakeServiceId(Runtime->GetNodeId(nodeIdx)), aid, nodeIdx);
}
{
auto* actor = NOlap::NDataAccessorControl::TGeneralCache::CreateService(NGeneralCache::NPublic::TConfig::BuildDefault(), new ::NMonitoring::TDynamicCounters());
const auto aid = Runtime->Register(actor, nodeIdx, appData.UserPoolId, TMailboxType::Revolving, 0);
Runtime->RegisterService(NOlap::NDataAccessorControl::TGeneralCache::MakeServiceId(Runtime->GetNodeId(nodeIdx)), aid, nodeIdx);
}
Runtime->Register(CreateLabelsMaintainer({}), nodeIdx, appData.SystemPoolId, TMailboxType::Revolving, 0);

auto sysViewService = NSysView::CreateSysViewServiceForTests();
Expand Down
Loading