Skip to content

Commit f930039

Browse files
CyberROFLCyberROFL
authored
Limit inflight config updates (#17725)
1 parent bcdb29a commit f930039

File tree

5 files changed

+116
-22
lines changed

5 files changed

+116
-22
lines changed

‎ydb/core/cms/console/console.cpp‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,7 @@ void TConsole::OnActivateExecutor(const TActorContext &ctx)
3131

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

34-
ConfigsManager = new TConfigsManager(*this);
34+
ConfigsManager = new TConfigsManager(*this, Counters);
3535
ctx.RegisterWithSameMailbox(ConfigsManager);
3636

3737
TenantsManager = new TTenantsManager(*this, domains->Domain,

‎ydb/core/cms/console/console_configs_manager.cpp‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -222,7 +222,7 @@ void TConfigsManager::Bootstrap(const TActorContext &ctx)
222222
ctx,
223223
false,
224224
NKikimrServices::CMS_CONFIGS);
225-
ConfigsProvider = ctx.Register(new TConfigsProvider(ctx.SelfID));
225+
ConfigsProvider = ctx.Register(new TConfigsProvider(ctx.SelfID, Counters));
226226

227227
ui32 item = (ui32)NKikimrConsole::TConfigItem::AllowEditYamlInUiItem;
228228
ctx.Send(MakeConfigsDispatcherID(SelfId().NodeId()),

‎ydb/core/cms/console/console_configs_manager.h‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,8 @@
1919
#include <ydb/library/actors/core/hfunc.h>
2020
#include <ydb/library/actors/interconnect/interconnect.h>
2121

22+
#include <library/cpp/monlib/dynamic_counters/counters.h>
23+
2224
namespace NKikimr::NConsole {
2325

2426
using NTabletFlatExecutor::ITransaction;
@@ -301,8 +303,9 @@ class TConfigsManager : public TActorBootstrapped<TConfigsManager> {
301303
}
302304

303305
public:
304-
TConfigsManager(TConsole &self)
306+
TConfigsManager(TConsole &self, ::NMonitoring::TDynamicCounterPtr counters)
305307
: Self(self)
308+
, Counters(counters)
306309
{
307310
}
308311

@@ -321,6 +324,7 @@ class TConfigsManager : public TActorBootstrapped<TConfigsManager> {
321324

322325
private:
323326
TConsole &Self;
327+
::NMonitoring::TDynamicCounterPtr Counters;
324328
TConfigsConfig Config;
325329
TString DomainName;
326330
// All config items by id.

‎ydb/core/cms/console/console_configs_provider.cpp‎

Lines changed: 67 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -323,6 +323,7 @@ class TSubscriptionClientSender : public TActorBootstrapped<TSubscriptionClientS
323323
HFuncTraced(TEvConsole::TEvConfigSubscriptionNotification, Handle);
324324
HFuncTraced(TEvents::TEvPoisonPill, Handle);
325325
HFuncTraced(TEvents::TEvUndelivered, Handle);
326+
HFuncTraced(TEvents::TEvWakeup, Handle);
326327
HFuncTraced(TEvInterconnect::TEvNodeDisconnected, Handle);
327328
IgnoreFunc(TEvInterconnect::TEvNodeConnected);
328329

@@ -351,6 +352,13 @@ class TSubscriptionClientSender : public TActorBootstrapped<TSubscriptionClientS
351352
Die(ctx);
352353
}
353354

355+
void Handle(TEvents::TEvWakeup::TPtr &/*ev*/, const TActorContext &ctx)
356+
{
357+
LOG_DEBUG_S(ctx, NKikimrServices::CMS_CONFIGS,
358+
"TSubscriptionClientSender(" << Subscription->Subscriber.ToString() << ") received wake up");
359+
Send(OwnerId, new TConfigsProvider::TEvPrivate::TEvWorkerCoolDown(Subscription));
360+
}
361+
354362
void Handle(TEvInterconnect::TEvNodeDisconnected::TPtr &/*ev*/, const TActorContext &ctx)
355363
{
356364
LOG_DEBUG_S(ctx, NKikimrServices::CMS_CONFIGS,
@@ -370,6 +378,8 @@ class TSubscriptionClientSender : public TActorBootstrapped<TSubscriptionClientS
370378
"TSubscriptionClientSender(" << Subscription->Subscriber.ToString() << ") send TEvConfigSubscriptionNotificationRequest: "
371379
<< notification.Get()->Record.ShortDebugString());
372380

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

@@ -522,7 +532,33 @@ void TConfigsProvider::CheckSubscriptions(const TInMemorySubscriptionSet &subscr
522532
const TActorContext &ctx)
523533
{
524534
for (auto &subscription : subscriptions)
525-
CheckSubscription(subscription, ctx);
535+
ScheduledUpdates[subscription->Subscriber] = EUpdate::All;
536+
ProcessScheduledUpdates(ctx);
537+
}
538+
539+
void TConfigsProvider::ProcessScheduledUpdates(const TActorContext &ctx)
540+
{
541+
while (!ScheduledUpdates.empty() && InflightUpdates.size() < MAX_INFLIGHT_UPDATES) {
542+
auto it = ScheduledUpdates.begin();
543+
if (auto subscription = InMemoryIndex.GetSubscription(it->first)) {
544+
switch (it->second) {
545+
case EUpdate::All:
546+
if (CheckSubscription(subscription, ctx)) {
547+
InflightUpdates.insert(subscription->Subscriber);
548+
}
549+
break;
550+
case EUpdate::Yaml:
551+
if (UpdateConfig(subscription, ctx)) {
552+
InflightUpdates.insert(subscription->Subscriber);
553+
}
554+
break;
555+
}
556+
}
557+
ScheduledUpdates.erase(it);
558+
}
559+
560+
*Counters.ScheduledConfigUpdates = ScheduledUpdates.size();
561+
*Counters.InflightConfigUpdates = InflightUpdates.size();
526562
}
527563

528564
void TConfigsProvider::CheckSubscription(TSubscription::TPtr subscription,
@@ -583,7 +619,7 @@ void TConfigsProvider::CheckSubscription(TSubscription::TPtr subscription,
583619
subscription->Worker = ctx.RegisterWithSameMailbox(worker);
584620
}
585621

586-
void TConfigsProvider::CheckSubscription(TInMemorySubscription::TPtr subscription,
622+
bool TConfigsProvider::CheckSubscription(TInMemorySubscription::TPtr subscription,
587623
const TActorContext &ctx)
588624
{
589625
LOG_TRACE_S(ctx, NKikimrServices::CMS_CONFIGS,
@@ -665,7 +701,7 @@ void TConfigsProvider::CheckSubscription(TInMemorySubscription::TPtr subscriptio
665701
LOG_TRACE_S(ctx, NKikimrServices::CMS_CONFIGS,
666702
"TConfigsProvider: no changes found for subscription"
667703
<< " " << subscription->Subscriber.ToString() << ":" << subscription->Generation);
668-
return;
704+
return false;
669705
}
670706

671707
LOG_TRACE_S(ctx, NKikimrServices::CMS_CONFIGS,
@@ -693,13 +729,6 @@ void TConfigsProvider::CheckSubscription(TInMemorySubscription::TPtr subscriptio
693729
for (auto &[id, hash] : VolatileYamlConfigHashes) {
694730
auto *volatileConfig = request->Record.AddVolatileConfigs();
695731
volatileConfig->SetId(id);
696-
auto hashes = subscription->VolatileYamlConfigHashes.size();
697-
Y_UNUSED(hashes);
698-
auto itt = subscription->VolatileYamlConfigHashes.find(id);
699-
if (itt != subscription->VolatileYamlConfigHashes.end()) {
700-
auto tmp = itt->second;
701-
Y_UNUSED(tmp);
702-
}
703732
if (auto it = subscription->VolatileYamlConfigHashes.find(id); it != subscription->VolatileYamlConfigHashes.end() && it->second == hash) {
704733
volatileConfig->SetNotChanged(true);
705734
} else {
@@ -720,8 +749,9 @@ void TConfigsProvider::CheckSubscription(TInMemorySubscription::TPtr subscriptio
720749
}
721750

722751
ctx.Send(subscription->Worker, request.Release());
723-
724752
subscription->FirstUpdateSent = true;
753+
754+
return true;
725755
}
726756

727757
void TConfigsProvider::DumpStateHTML(IOutputStream &os) const {
@@ -837,7 +867,8 @@ void TConfigsProvider::Handle(TEvConsole::TEvConfigSubscriptionRequest::TPtr &ev
837867

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

840-
CheckSubscription(subscription, ctx);
870+
ScheduledUpdates[subscription->Subscriber] = EUpdate::All;
871+
ProcessScheduledUpdates(ctx);
841872
}
842873

843874
void TConfigsProvider::Handle(TEvConsole::TEvConfigSubscriptionCanceled::TPtr &ev, const TActorContext &ctx)
@@ -865,11 +896,27 @@ void TConfigsProvider::Handle(TEvPrivate::TEvWorkerDisconnected::TPtr &ev, const
865896
auto existing = InMemoryIndex.GetSubscription(subscription->Subscriber);
866897
if (existing == subscription) {
867898
InMemoryIndex.RemoveSubscription(subscription->Subscriber);
899+
ScheduledUpdates.erase(subscription->Subscriber);
900+
InflightUpdates.erase(subscription->Subscriber);
901+
868902
Send(subscription->Subscriber, new TEvConsole::TEvConfigSubscriptionCanceled(subscription->Generation));
869903

870904
LOG_DEBUG_S(ctx, NKikimrServices::CMS_CONFIGS, "TConfigsProvider removed subscription "
871905
<< subscription->Subscriber<< ":" << subscription->Generation << " (subscription worker died)");
872906
}
907+
908+
ProcessScheduledUpdates(ctx);
909+
}
910+
911+
void TConfigsProvider::Handle(TEvPrivate::TEvWorkerCoolDown::TPtr &ev, const TActorContext &ctx)
912+
{
913+
auto subscription = ev->Get()->Subscription;
914+
auto existing = InMemoryIndex.GetSubscription(subscription->Subscriber);
915+
if (existing == subscription) {
916+
InflightUpdates.erase(subscription->Subscriber);
917+
}
918+
919+
ProcessScheduledUpdates(ctx);
873920
}
874921

875922
void TConfigsProvider::Handle(TEvConsole::TEvCheckConfigUpdatesRequest::TPtr &ev, const TActorContext &ctx)
@@ -1259,19 +1306,21 @@ void TConfigsProvider::Handle(TEvPrivate::TEvUpdateYamlConfig::TPtr &ev, const T
12591306
}
12601307

12611308
for (auto &[_, subscription] : InMemoryIndex.GetSubscriptions()) {
1262-
UpdateConfig(subscription, ctx);
1309+
ScheduledUpdates.emplace(subscription->Subscriber, EUpdate::Yaml);
12631310
}
12641311
} else {
12651312
const auto* subs = InMemoryIndex.GetSubscriptions(ev->Get()->ChangedDatabase);
12661313
if (subs) {
12671314
for (auto &subscription : *subs) {
1268-
UpdateConfig(subscription, ctx);
1315+
ScheduledUpdates.emplace(subscription->Subscriber, EUpdate::Yaml);
12691316
}
12701317
}
12711318
}
1319+
1320+
ProcessScheduledUpdates(ctx);
12721321
}
12731322

1274-
void TConfigsProvider::UpdateConfig(TInMemorySubscription::TPtr subscription,
1323+
bool TConfigsProvider::UpdateConfig(TInMemorySubscription::TPtr subscription,
12751324
const TActorContext &ctx)
12761325
{
12771326
if (subscription->ServeYaml) {
@@ -1309,7 +1358,10 @@ void TConfigsProvider::UpdateConfig(TInMemorySubscription::TPtr subscription,
13091358
}
13101359

13111360
ctx.Send(subscription->Worker, request.Release());
1361+
return true;
13121362
}
1363+
1364+
return false;
13131365
}
13141366

13151367
} // namespace NKikimr::NConsole

‎ydb/core/cms/console/console_configs_provider.h‎

Lines changed: 42 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -8,9 +8,10 @@
88
#include <ydb/core/base/tablet_pipe.h>
99
#include <ydb/core/cms/console/util/config_index.h>
1010
#include <ydb/core/tablet_flat/tablet_flat_executed.h>
11-
1211
#include <ydb/library/actors/core/hfunc.h>
1312

13+
#include <library/cpp/monlib/dynamic_counters/counters.h>
14+
1415
namespace NKikimr::NConsole {
1516

1617
struct TDatabaseYamlConfig {
@@ -31,6 +32,7 @@ class TConfigsProvider : public TActorBootstrapped<TConfigsProvider> {
3132
EvUpdateYamlConfig,
3233
EvUpdateSubscriptions,
3334
EvWorkerDisconnected,
35+
EvWorkerCoolDown,
3436

3537
EvEnd
3638
};
@@ -64,6 +66,15 @@ class TConfigsProvider : public TActorBootstrapped<TConfigsProvider> {
6466
TInMemorySubscription::TPtr Subscription;
6567
};
6668

69+
struct TEvWorkerCoolDown: public TEventLocal<TEvWorkerCoolDown, EvWorkerCoolDown> {
70+
explicit TEvWorkerCoolDown(TInMemorySubscription::TPtr subscription)
71+
: Subscription(subscription)
72+
{
73+
}
74+
75+
TInMemorySubscription::TPtr Subscription;
76+
};
77+
6778
struct TEvSetConfig : public TEventLocal<TEvSetConfig, EvSetConfig> {
6879
TEvSetConfig(const TConfigsConfig &config)
6980
: Config(config)
@@ -178,12 +189,14 @@ class TConfigsProvider : public TActorBootstrapped<TConfigsProvider> {
178189
const TActorContext &ctx);
179190
void CheckSubscription(TSubscription::TPtr subscriptions,
180191
const TActorContext &ctx);
181-
void CheckSubscription(TInMemorySubscription::TPtr subscriptions,
192+
bool CheckSubscription(TInMemorySubscription::TPtr subscriptions,
182193
const TActorContext &ctx);
183194

184-
void UpdateConfig(TInMemorySubscription::TPtr subscription,
195+
bool UpdateConfig(TInMemorySubscription::TPtr subscription,
185196
const TActorContext &ctx);
186197

198+
void ProcessScheduledUpdates(const TActorContext &ctx);
199+
187200
void Handle(NMon::TEvHttpInfo::TPtr &ev);
188201
void Handle(TEvConsole::TEvConfigSubscriptionRequest::TPtr &ev, const TActorContext &ctx);
189202
void Handle(TEvConsole::TEvConfigSubscriptionCanceled::TPtr &ev, const TActorContext &ctx);
@@ -195,6 +208,7 @@ class TConfigsProvider : public TActorBootstrapped<TConfigsProvider> {
195208
void Handle(TEvConsole::TEvGetNodeConfigRequest::TPtr &ev, const TActorContext &ctx);
196209
void Handle(TEvConsole::TEvListConfigSubscriptionsRequest::TPtr &ev, const TActorContext &ctx);
197210
void Handle(TEvPrivate::TEvWorkerDisconnected::TPtr &ev, const TActorContext &ctx);
211+
void Handle(TEvPrivate::TEvWorkerCoolDown::TPtr &ev, const TActorContext &ctx);
198212
void Handle(TEvPrivate::TEvNotificationTimeout::TPtr &ev, const TActorContext &ctx);
199213
void Handle(TEvPrivate::TEvSenderDied::TPtr &ev, const TActorContext &ctx);
200214
void Handle(TEvPrivate::TEvSetConfig::TPtr &ev, const TActorContext &ctx);
@@ -225,6 +239,7 @@ class TConfigsProvider : public TActorBootstrapped<TConfigsProvider> {
225239
HFuncTraced(TEvConsole::TEvGetNodeConfigRequest, Handle);
226240
HFuncTraced(TEvConsole::TEvListConfigSubscriptionsRequest, Handle);
227241
HFuncTraced(TEvPrivate::TEvWorkerDisconnected, Handle);
242+
HFuncTraced(TEvPrivate::TEvWorkerCoolDown, Handle);
228243
HFuncTraced(TEvPrivate::TEvNotificationTimeout, Handle);
229244
HFuncTraced(TEvPrivate::TEvSenderDied, Handle);
230245
HFuncTraced(TEvPrivate::TEvSetConfig, Handle);
@@ -242,9 +257,22 @@ class TConfigsProvider : public TActorBootstrapped<TConfigsProvider> {
242257
}
243258
}
244259

260+
struct TCounters {
261+
using TCounterPtr = ::NMonitoring::TDynamicCounters::TCounterPtr;
262+
TCounterPtr ScheduledConfigUpdates;
263+
TCounterPtr InflightConfigUpdates;
264+
265+
explicit TCounters(::NMonitoring::TDynamicCounterPtr counters)
266+
: ScheduledConfigUpdates(counters->GetCounter("ScheduledConfigUpdates", false))
267+
, InflightConfigUpdates(counters->GetCounter("InflightConfigUpdates", false))
268+
{
269+
}
270+
};
271+
245272
public:
246-
TConfigsProvider(TActorId ownerId)
273+
TConfigsProvider(TActorId ownerId, ::NMonitoring::TDynamicCounterPtr counters)
247274
: ConfigsManager(ownerId)
275+
, Counters(counters)
248276
{
249277
}
250278

@@ -264,10 +292,20 @@ class TConfigsProvider : public TActorBootstrapped<TConfigsProvider> {
264292

265293
private:
266294
TActorId ConfigsManager;
295+
TCounters Counters;
267296
TConfigsConfig Config;
268297
TConfigIndex ConfigIndex;
269298
TSubscriptionIndex SubscriptionIndex;
299+
300+
enum class EUpdate {
301+
All,
302+
Yaml,
303+
};
304+
270305
TInMemorySubscriptionIndex InMemoryIndex;
306+
THashMap<TActorId, EUpdate> ScheduledUpdates;
307+
THashSet<TActorId> InflightUpdates;
308+
static constexpr ui32 MAX_INFLIGHT_UPDATES = 50;
271309

272310
TString MainYamlConfig;
273311
TMap<ui64, TString> VolatileYamlConfigs;

0 commit comments

Comments
 (0)