Skip to content

24-4-4: Limit inflight config updates #17755

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

Merged
merged 1 commit into from
Apr 27, 2025
Merged

24-4-4: Limit inflight config updates #17755

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
2 changes: 1 addition & 1 deletion ydb/core/cms/console/console.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,7 +31,7 @@ void TConsole::OnActivateExecutor(const TActorContext &ctx)

TValidatorsRegistry::Instance()->LockValidators();

ConfigsManager = new TConfigsManager(*this);
ConfigsManager = new TConfigsManager(*this, Counters);
ctx.RegisterWithSameMailbox(ConfigsManager);

TenantsManager = new TTenantsManager(*this, domains->Domain,
Expand Down
2 changes: 1 addition & 1 deletion ydb/core/cms/console/console_configs_manager.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -77,7 +77,7 @@ void TConfigsManager::Bootstrap(const TActorContext &ctx)
ctx,
false,
NKikimrServices::CMS_CONFIGS);
ConfigsProvider = ctx.Register(new TConfigsProvider(ctx.SelfID));
ConfigsProvider = ctx.Register(new TConfigsProvider(ctx.SelfID, Counters));

ui32 item = (ui32)NKikimrConsole::TConfigItem::AllowEditYamlInUiItem;
ctx.Send(MakeConfigsDispatcherID(SelfId().NodeId()),
Expand Down
6 changes: 5 additions & 1 deletion ydb/core/cms/console/console_configs_manager.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,8 @@
#include <ydb/library/actors/core/hfunc.h>
#include <ydb/library/actors/interconnect/interconnect.h>

#include <library/cpp/monlib/dynamic_counters/counters.h>

namespace NKikimr::NConsole {

using NTabletFlatExecutor::ITransaction;
Expand Down Expand Up @@ -229,8 +231,9 @@ class TConfigsManager : public TActorBootstrapped<TConfigsManager> {
}

public:
TConfigsManager(TConsole &self)
TConfigsManager(TConsole &self, ::NMonitoring::TDynamicCounterPtr counters)
: Self(self)
, Counters(counters)
{
}

Expand All @@ -249,6 +252,7 @@ class TConfigsManager : public TActorBootstrapped<TConfigsManager> {

private:
TConsole &Self;
::NMonitoring::TDynamicCounterPtr Counters;
TConfigsConfig Config;
// All config items by id.
TConfigIndex ConfigIndex;
Expand Down
128 changes: 93 additions & 35 deletions ydb/core/cms/console/console_configs_provider.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -322,6 +322,7 @@ class TSubscriptionClientSender : public TActorBootstrapped<TSubscriptionClientS
HFuncTraced(TEvConsole::TEvConfigSubscriptionNotification, Handle);
HFuncTraced(TEvents::TEvPoisonPill, Handle);
HFuncTraced(TEvents::TEvUndelivered, Handle);
HFuncTraced(TEvents::TEvWakeup, Handle);
HFuncTraced(TEvInterconnect::TEvNodeDisconnected, Handle);
IgnoreFunc(TEvInterconnect::TEvNodeConnected);

Expand Down Expand Up @@ -350,6 +351,13 @@ class TSubscriptionClientSender : public TActorBootstrapped<TSubscriptionClientS
Die(ctx);
}

void Handle(TEvents::TEvWakeup::TPtr &/*ev*/, const TActorContext &ctx)
{
LOG_DEBUG_S(ctx, NKikimrServices::CMS_CONFIGS,
"TSubscriptionClientSender(" << Subscription->Subscriber.ToString() << ") received wake up");
Send(OwnerId, new TConfigsProvider::TEvPrivate::TEvWorkerCoolDown(Subscription));
}

void Handle(TEvInterconnect::TEvNodeDisconnected::TPtr &/*ev*/, const TActorContext &ctx)
{
LOG_DEBUG_S(ctx, NKikimrServices::CMS_CONFIGS,
Expand All @@ -369,6 +377,8 @@ class TSubscriptionClientSender : public TActorBootstrapped<TSubscriptionClientS
"TSubscriptionClientSender(" << Subscription->Subscriber.ToString() << ") send TEvConfigSubscriptionNotificationRequest: "
<< notification.Get()->Record.ShortDebugString());

const float mbytes = notification.Get()->GetCachedByteSize() / 1'000'000.f;
Schedule(TDuration::MilliSeconds(100) * mbytes, new TEvents::TEvWakeup());
Send(Subscription->Subscriber, notification.Release(), IEventHandle::FlagTrackDelivery | IEventHandle::FlagSubscribeOnSession);
}

Expand Down Expand Up @@ -521,7 +531,33 @@ void TConfigsProvider::CheckSubscriptions(const TInMemorySubscriptionSet &subscr
const TActorContext &ctx)
{
for (auto &subscription : subscriptions)
CheckSubscription(subscription, ctx);
ScheduledUpdates[subscription->Subscriber] = EUpdate::All;
ProcessScheduledUpdates(ctx);
}

void TConfigsProvider::ProcessScheduledUpdates(const TActorContext &ctx)
{
while (!ScheduledUpdates.empty() && InflightUpdates.size() < MAX_INFLIGHT_UPDATES) {
auto it = ScheduledUpdates.begin();
if (auto subscription = InMemoryIndex.GetSubscription(it->first)) {
switch (it->second) {
case EUpdate::All:
if (CheckSubscription(subscription, ctx)) {
InflightUpdates.insert(subscription->Subscriber);
}
break;
case EUpdate::Yaml:
if (UpdateConfig(subscription, ctx)) {
InflightUpdates.insert(subscription->Subscriber);
}
break;
}
}
ScheduledUpdates.erase(it);
}

*Counters.ScheduledConfigUpdates = ScheduledUpdates.size();
*Counters.InflightConfigUpdates = InflightUpdates.size();
}

void TConfigsProvider::CheckSubscription(TSubscription::TPtr subscription,
Expand Down Expand Up @@ -582,7 +618,7 @@ void TConfigsProvider::CheckSubscription(TSubscription::TPtr subscription,
subscription->Worker = ctx.RegisterWithSameMailbox(worker);
}

void TConfigsProvider::CheckSubscription(TInMemorySubscription::TPtr subscription,
bool TConfigsProvider::CheckSubscription(TInMemorySubscription::TPtr subscription,
const TActorContext &ctx)
{
LOG_TRACE_S(ctx, NKikimrServices::CMS_CONFIGS,
Expand Down Expand Up @@ -664,7 +700,7 @@ void TConfigsProvider::CheckSubscription(TInMemorySubscription::TPtr subscriptio
LOG_TRACE_S(ctx, NKikimrServices::CMS_CONFIGS,
"TConfigsProvider: no changes found for subscription"
<< " " << subscription->Subscriber.ToString() << ":" << subscription->Generation);
return;
return false;
}

LOG_TRACE_S(ctx, NKikimrServices::CMS_CONFIGS,
Expand Down Expand Up @@ -692,13 +728,6 @@ void TConfigsProvider::CheckSubscription(TInMemorySubscription::TPtr subscriptio
for (auto &[id, hash] : VolatileYamlConfigHashes) {
auto *volatileConfig = request->Record.AddVolatileConfigs();
volatileConfig->SetId(id);
auto hashes = subscription->VolatileYamlConfigHashes.size();
Y_UNUSED(hashes);
auto itt = subscription->VolatileYamlConfigHashes.find(id);
if (itt != subscription->VolatileYamlConfigHashes.end()) {
auto tmp = itt->second;
Y_UNUSED(tmp);
}
if (auto it = subscription->VolatileYamlConfigHashes.find(id); it != subscription->VolatileYamlConfigHashes.end() && it->second == hash) {
volatileConfig->SetNotChanged(true);
} else {
Expand All @@ -709,8 +738,9 @@ void TConfigsProvider::CheckSubscription(TInMemorySubscription::TPtr subscriptio
subscription->VolatileYamlConfigHashes = VolatileYamlConfigHashes;

ctx.Send(subscription->Worker, request.Release());

subscription->FirstUpdateSent = true;

return true;
}

void TConfigsProvider::DumpStateHTML(IOutputStream &os) const {
Expand Down Expand Up @@ -822,7 +852,8 @@ void TConfigsProvider::Handle(TEvConsole::TEvConfigSubscriptionRequest::TPtr &ev

subscription->Worker = RegisterWithSameMailbox(new TSubscriptionClientSender(subscription, SelfId()));

CheckSubscription(subscription, ctx);
ScheduledUpdates[subscription->Subscriber] = EUpdate::All;
ProcessScheduledUpdates(ctx);
}

void TConfigsProvider::Handle(TEvConsole::TEvConfigSubscriptionCanceled::TPtr &ev, const TActorContext &ctx)
Expand Down Expand Up @@ -850,11 +881,27 @@ void TConfigsProvider::Handle(TEvPrivate::TEvWorkerDisconnected::TPtr &ev, const
auto existing = InMemoryIndex.GetSubscription(subscription->Subscriber);
if (existing == subscription) {
InMemoryIndex.RemoveSubscription(subscription->Subscriber);
ScheduledUpdates.erase(subscription->Subscriber);
InflightUpdates.erase(subscription->Subscriber);

Send(subscription->Subscriber, new TEvConsole::TEvConfigSubscriptionCanceled(subscription->Generation));

LOG_DEBUG_S(ctx, NKikimrServices::CMS_CONFIGS, "TConfigsProvider removed subscription "
<< subscription->Subscriber<< ":" << subscription->Generation << " (subscription worker died)");
}

ProcessScheduledUpdates(ctx);
}

void TConfigsProvider::Handle(TEvPrivate::TEvWorkerCoolDown::TPtr &ev, const TActorContext &ctx)
{
auto subscription = ev->Get()->Subscription;
auto existing = InMemoryIndex.GetSubscription(subscription->Subscriber);
if (existing == subscription) {
InflightUpdates.erase(subscription->Subscriber);
}

ProcessScheduledUpdates(ctx);
}

void TConfigsProvider::Handle(TEvConsole::TEvCheckConfigUpdatesRequest::TPtr &ev, const TActorContext &ctx)
Expand Down Expand Up @@ -1237,34 +1284,45 @@ void TConfigsProvider::Handle(TEvPrivate::TEvUpdateYamlConfig::TPtr &ev, const T
}

for (auto &[_, subscription] : InMemoryIndex.GetSubscriptions()) {
if (subscription->ServeYaml) {
auto request = MakeHolder<TEvConsole::TEvConfigSubscriptionNotification>(
subscription->Generation,
NKikimrConfig::TAppConfig{},
THashSet<ui32>{});

if (subscription->YamlConfigVersion != YamlConfigVersion) {
subscription->YamlConfigVersion = YamlConfigVersion;
request->Record.SetYamlConfig(YamlConfig);
} else {
request->Record.SetYamlConfigNotChanged(true);
}
ScheduledUpdates.emplace(subscription->Subscriber, EUpdate::Yaml);
}

for (auto &[id, hash] : VolatileYamlConfigHashes) {
auto *volatileConfig = request->Record.AddVolatileConfigs();
volatileConfig->SetId(id);
if (auto it = subscription->VolatileYamlConfigHashes.find(id); it != subscription->VolatileYamlConfigHashes.end() && it->second == hash) {
volatileConfig->SetNotChanged(true);
} else {
volatileConfig->SetConfig(VolatileYamlConfigs[id]);
}
}
ProcessScheduledUpdates(ctx);
}

subscription->VolatileYamlConfigHashes = VolatileYamlConfigHashes;
bool TConfigsProvider::UpdateConfig(TInMemorySubscription::TPtr subscription,
const TActorContext &ctx)
{
if (subscription->ServeYaml) {
auto request = MakeHolder<TEvConsole::TEvConfigSubscriptionNotification>(
subscription->Generation,
NKikimrConfig::TAppConfig{},
THashSet<ui32>{});

if (subscription->YamlConfigVersion != YamlConfigVersion) {
subscription->YamlConfigVersion = YamlConfigVersion;
request->Record.SetYamlConfig(YamlConfig);
} else {
request->Record.SetYamlConfigNotChanged(true);
}

ctx.Send(subscription->Worker, request.Release());
for (auto &[id, hash] : VolatileYamlConfigHashes) {
auto *volatileConfig = request->Record.AddVolatileConfigs();
volatileConfig->SetId(id);
if (auto it = subscription->VolatileYamlConfigHashes.find(id); it != subscription->VolatileYamlConfigHashes.end() && it->second == hash) {
volatileConfig->SetNotChanged(true);
} else {
volatileConfig->SetConfig(VolatileYamlConfigs[id]);
}
}

subscription->VolatileYamlConfigHashes = VolatileYamlConfigHashes;

ctx.Send(subscription->Worker, request.Release());
return true;
}

return false;
}

} // namespace NKikimr::NConsole
47 changes: 44 additions & 3 deletions ydb/core/cms/console/console_configs_provider.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,9 +8,10 @@
#include <ydb/core/base/tablet_pipe.h>
#include <ydb/core/cms/console/util/config_index.h>
#include <ydb/core/tablet_flat/tablet_flat_executed.h>

#include <ydb/library/actors/core/hfunc.h>

#include <library/cpp/monlib/dynamic_counters/counters.h>

namespace NKikimr::NConsole {

class TConfigsProvider : public TActorBootstrapped<TConfigsProvider> {
Expand All @@ -26,6 +27,7 @@ class TConfigsProvider : public TActorBootstrapped<TConfigsProvider> {
EvUpdateYamlConfig,
EvUpdateSubscriptions,
EvWorkerDisconnected,
EvWorkerCoolDown,

EvEnd
};
Expand Down Expand Up @@ -59,6 +61,15 @@ class TConfigsProvider : public TActorBootstrapped<TConfigsProvider> {
TInMemorySubscription::TPtr Subscription;
};

struct TEvWorkerCoolDown: public TEventLocal<TEvWorkerCoolDown, EvWorkerCoolDown> {
explicit TEvWorkerCoolDown(TInMemorySubscription::TPtr subscription)
: Subscription(subscription)
{
}

TInMemorySubscription::TPtr Subscription;
};

struct TEvSetConfig : public TEventLocal<TEvSetConfig, EvSetConfig> {
TEvSetConfig(const TConfigsConfig &config)
: Config(config)
Expand Down Expand Up @@ -139,9 +150,14 @@ class TConfigsProvider : public TActorBootstrapped<TConfigsProvider> {
const TActorContext &ctx);
void CheckSubscription(TSubscription::TPtr subscriptions,
const TActorContext &ctx);
void CheckSubscription(TInMemorySubscription::TPtr subscriptions,
bool CheckSubscription(TInMemorySubscription::TPtr subscriptions,
const TActorContext &ctx);

bool UpdateConfig(TInMemorySubscription::TPtr subscription,
const TActorContext &ctx);

void ProcessScheduledUpdates(const TActorContext &ctx);

void Handle(NMon::TEvHttpInfo::TPtr &ev);
void Handle(TEvConsole::TEvConfigSubscriptionRequest::TPtr &ev, const TActorContext &ctx);
void Handle(TEvConsole::TEvConfigSubscriptionCanceled::TPtr &ev, const TActorContext &ctx);
Expand All @@ -153,6 +169,7 @@ class TConfigsProvider : public TActorBootstrapped<TConfigsProvider> {
void Handle(TEvConsole::TEvGetNodeConfigRequest::TPtr &ev, const TActorContext &ctx);
void Handle(TEvConsole::TEvListConfigSubscriptionsRequest::TPtr &ev, const TActorContext &ctx);
void Handle(TEvPrivate::TEvWorkerDisconnected::TPtr &ev, const TActorContext &ctx);
void Handle(TEvPrivate::TEvWorkerCoolDown::TPtr &ev, const TActorContext &ctx);
void Handle(TEvPrivate::TEvNotificationTimeout::TPtr &ev, const TActorContext &ctx);
void Handle(TEvPrivate::TEvSenderDied::TPtr &ev, const TActorContext &ctx);
void Handle(TEvPrivate::TEvSetConfig::TPtr &ev, const TActorContext &ctx);
Expand Down Expand Up @@ -183,6 +200,7 @@ class TConfigsProvider : public TActorBootstrapped<TConfigsProvider> {
HFuncTraced(TEvConsole::TEvGetNodeConfigRequest, Handle);
HFuncTraced(TEvConsole::TEvListConfigSubscriptionsRequest, Handle);
HFuncTraced(TEvPrivate::TEvWorkerDisconnected, Handle);
HFuncTraced(TEvPrivate::TEvWorkerCoolDown, Handle);
HFuncTraced(TEvPrivate::TEvNotificationTimeout, Handle);
HFuncTraced(TEvPrivate::TEvSenderDied, Handle);
HFuncTraced(TEvPrivate::TEvSetConfig, Handle);
Expand All @@ -200,9 +218,22 @@ class TConfigsProvider : public TActorBootstrapped<TConfigsProvider> {
}
}

struct TCounters {
using TCounterPtr = ::NMonitoring::TDynamicCounters::TCounterPtr;
TCounterPtr ScheduledConfigUpdates;
TCounterPtr InflightConfigUpdates;

explicit TCounters(::NMonitoring::TDynamicCounterPtr counters)
: ScheduledConfigUpdates(counters->GetCounter("ScheduledConfigUpdates", false))
, InflightConfigUpdates(counters->GetCounter("InflightConfigUpdates", false))
{
}
};

public:
TConfigsProvider(TActorId ownerId)
TConfigsProvider(TActorId ownerId, ::NMonitoring::TDynamicCounterPtr counters)
: ConfigsManager(ownerId)
, Counters(counters)
{
}

Expand All @@ -222,10 +253,20 @@ class TConfigsProvider : public TActorBootstrapped<TConfigsProvider> {

private:
TActorId ConfigsManager;
TCounters Counters;
TConfigsConfig Config;
TConfigIndex ConfigIndex;
TSubscriptionIndex SubscriptionIndex;

enum class EUpdate {
All,
Yaml,
};

TInMemorySubscriptionIndex InMemoryIndex;
THashMap<TActorId, EUpdate> ScheduledUpdates;
THashSet<TActorId> InflightUpdates;
static constexpr ui32 MAX_INFLIGHT_UPDATES = 50;

TString YamlConfig;
TMap<ui64, TString> VolatileYamlConfigs;
Expand Down
Loading