Skip to content

Commit 9c466b2

Browse files
CyberROFLCyberROFL
authored
24-4-4: Limit inflight config updates (#17755)
1 parent 171b7a4 commit 9c466b2

File tree

5 files changed

+144
-41
lines changed

5 files changed

+144
-41
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
@@ -77,7 +77,7 @@ void TConfigsManager::Bootstrap(const TActorContext &ctx)
7777
ctx,
7878
false,
7979
NKikimrServices::CMS_CONFIGS);
80-
ConfigsProvider = ctx.Register(new TConfigsProvider(ctx.SelfID));
80+
ConfigsProvider = ctx.Register(new TConfigsProvider(ctx.SelfID, Counters));
8181

8282
ui32 item = (ui32)NKikimrConsole::TConfigItem::AllowEditYamlInUiItem;
8383
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
@@ -17,6 +17,8 @@
1717
#include <ydb/library/actors/core/hfunc.h>
1818
#include <ydb/library/actors/interconnect/interconnect.h>
1919

20+
#include <library/cpp/monlib/dynamic_counters/counters.h>
21+
2022
namespace NKikimr::NConsole {
2123

2224
using NTabletFlatExecutor::ITransaction;
@@ -229,8 +231,9 @@ class TConfigsManager : public TActorBootstrapped<TConfigsManager> {
229231
}
230232

231233
public:
232-
TConfigsManager(TConsole &self)
234+
TConfigsManager(TConsole &self, ::NMonitoring::TDynamicCounterPtr counters)
233235
: Self(self)
236+
, Counters(counters)
234237
{
235238
}
236239

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

250253
private:
251254
TConsole &Self;
255+
::NMonitoring::TDynamicCounterPtr Counters;
252256
TConfigsConfig Config;
253257
// All config items by id.
254258
TConfigIndex ConfigIndex;

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

Lines changed: 93 additions & 35 deletions
Original file line numberDiff line numberDiff line change
@@ -322,6 +322,7 @@ class TSubscriptionClientSender : public TActorBootstrapped<TSubscriptionClientS
322322
HFuncTraced(TEvConsole::TEvConfigSubscriptionNotification, Handle);
323323
HFuncTraced(TEvents::TEvPoisonPill, Handle);
324324
HFuncTraced(TEvents::TEvUndelivered, Handle);
325+
HFuncTraced(TEvents::TEvWakeup, Handle);
325326
HFuncTraced(TEvInterconnect::TEvNodeDisconnected, Handle);
326327
IgnoreFunc(TEvInterconnect::TEvNodeConnected);
327328

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

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

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

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

527563
void TConfigsProvider::CheckSubscription(TSubscription::TPtr subscription,
@@ -582,7 +618,7 @@ void TConfigsProvider::CheckSubscription(TSubscription::TPtr subscription,
582618
subscription->Worker = ctx.RegisterWithSameMailbox(worker);
583619
}
584620

585-
void TConfigsProvider::CheckSubscription(TInMemorySubscription::TPtr subscription,
621+
bool TConfigsProvider::CheckSubscription(TInMemorySubscription::TPtr subscription,
586622
const TActorContext &ctx)
587623
{
588624
LOG_TRACE_S(ctx, NKikimrServices::CMS_CONFIGS,
@@ -664,7 +700,7 @@ void TConfigsProvider::CheckSubscription(TInMemorySubscription::TPtr subscriptio
664700
LOG_TRACE_S(ctx, NKikimrServices::CMS_CONFIGS,
665701
"TConfigsProvider: no changes found for subscription"
666702
<< " " << subscription->Subscriber.ToString() << ":" << subscription->Generation);
667-
return;
703+
return false;
668704
}
669705

670706
LOG_TRACE_S(ctx, NKikimrServices::CMS_CONFIGS,
@@ -692,13 +728,6 @@ void TConfigsProvider::CheckSubscription(TInMemorySubscription::TPtr subscriptio
692728
for (auto &[id, hash] : VolatileYamlConfigHashes) {
693729
auto *volatileConfig = request->Record.AddVolatileConfigs();
694730
volatileConfig->SetId(id);
695-
auto hashes = subscription->VolatileYamlConfigHashes.size();
696-
Y_UNUSED(hashes);
697-
auto itt = subscription->VolatileYamlConfigHashes.find(id);
698-
if (itt != subscription->VolatileYamlConfigHashes.end()) {
699-
auto tmp = itt->second;
700-
Y_UNUSED(tmp);
701-
}
702731
if (auto it = subscription->VolatileYamlConfigHashes.find(id); it != subscription->VolatileYamlConfigHashes.end() && it->second == hash) {
703732
volatileConfig->SetNotChanged(true);
704733
} else {
@@ -709,8 +738,9 @@ void TConfigsProvider::CheckSubscription(TInMemorySubscription::TPtr subscriptio
709738
subscription->VolatileYamlConfigHashes = VolatileYamlConfigHashes;
710739

711740
ctx.Send(subscription->Worker, request.Release());
712-
713741
subscription->FirstUpdateSent = true;
742+
743+
return true;
714744
}
715745

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

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

825-
CheckSubscription(subscription, ctx);
855+
ScheduledUpdates[subscription->Subscriber] = EUpdate::All;
856+
ProcessScheduledUpdates(ctx);
826857
}
827858

828859
void TConfigsProvider::Handle(TEvConsole::TEvConfigSubscriptionCanceled::TPtr &ev, const TActorContext &ctx)
@@ -850,11 +881,27 @@ void TConfigsProvider::Handle(TEvPrivate::TEvWorkerDisconnected::TPtr &ev, const
850881
auto existing = InMemoryIndex.GetSubscription(subscription->Subscriber);
851882
if (existing == subscription) {
852883
InMemoryIndex.RemoveSubscription(subscription->Subscriber);
884+
ScheduledUpdates.erase(subscription->Subscriber);
885+
InflightUpdates.erase(subscription->Subscriber);
886+
853887
Send(subscription->Subscriber, new TEvConsole::TEvConfigSubscriptionCanceled(subscription->Generation));
854888

855889
LOG_DEBUG_S(ctx, NKikimrServices::CMS_CONFIGS, "TConfigsProvider removed subscription "
856890
<< subscription->Subscriber<< ":" << subscription->Generation << " (subscription worker died)");
857891
}
892+
893+
ProcessScheduledUpdates(ctx);
894+
}
895+
896+
void TConfigsProvider::Handle(TEvPrivate::TEvWorkerCoolDown::TPtr &ev, const TActorContext &ctx)
897+
{
898+
auto subscription = ev->Get()->Subscription;
899+
auto existing = InMemoryIndex.GetSubscription(subscription->Subscriber);
900+
if (existing == subscription) {
901+
InflightUpdates.erase(subscription->Subscriber);
902+
}
903+
904+
ProcessScheduledUpdates(ctx);
858905
}
859906

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

12391286
for (auto &[_, subscription] : InMemoryIndex.GetSubscriptions()) {
1240-
if (subscription->ServeYaml) {
1241-
auto request = MakeHolder<TEvConsole::TEvConfigSubscriptionNotification>(
1242-
subscription->Generation,
1243-
NKikimrConfig::TAppConfig{},
1244-
THashSet<ui32>{});
1245-
1246-
if (subscription->YamlConfigVersion != YamlConfigVersion) {
1247-
subscription->YamlConfigVersion = YamlConfigVersion;
1248-
request->Record.SetYamlConfig(YamlConfig);
1249-
} else {
1250-
request->Record.SetYamlConfigNotChanged(true);
1251-
}
1287+
ScheduledUpdates.emplace(subscription->Subscriber, EUpdate::Yaml);
1288+
}
12521289

1253-
for (auto &[id, hash] : VolatileYamlConfigHashes) {
1254-
auto *volatileConfig = request->Record.AddVolatileConfigs();
1255-
volatileConfig->SetId(id);
1256-
if (auto it = subscription->VolatileYamlConfigHashes.find(id); it != subscription->VolatileYamlConfigHashes.end() && it->second == hash) {
1257-
volatileConfig->SetNotChanged(true);
1258-
} else {
1259-
volatileConfig->SetConfig(VolatileYamlConfigs[id]);
1260-
}
1261-
}
1290+
ProcessScheduledUpdates(ctx);
1291+
}
12621292

1263-
subscription->VolatileYamlConfigHashes = VolatileYamlConfigHashes;
1293+
bool TConfigsProvider::UpdateConfig(TInMemorySubscription::TPtr subscription,
1294+
const TActorContext &ctx)
1295+
{
1296+
if (subscription->ServeYaml) {
1297+
auto request = MakeHolder<TEvConsole::TEvConfigSubscriptionNotification>(
1298+
subscription->Generation,
1299+
NKikimrConfig::TAppConfig{},
1300+
THashSet<ui32>{});
1301+
1302+
if (subscription->YamlConfigVersion != YamlConfigVersion) {
1303+
subscription->YamlConfigVersion = YamlConfigVersion;
1304+
request->Record.SetYamlConfig(YamlConfig);
1305+
} else {
1306+
request->Record.SetYamlConfigNotChanged(true);
1307+
}
12641308

1265-
ctx.Send(subscription->Worker, request.Release());
1309+
for (auto &[id, hash] : VolatileYamlConfigHashes) {
1310+
auto *volatileConfig = request->Record.AddVolatileConfigs();
1311+
volatileConfig->SetId(id);
1312+
if (auto it = subscription->VolatileYamlConfigHashes.find(id); it != subscription->VolatileYamlConfigHashes.end() && it->second == hash) {
1313+
volatileConfig->SetNotChanged(true);
1314+
} else {
1315+
volatileConfig->SetConfig(VolatileYamlConfigs[id]);
1316+
}
12661317
}
1318+
1319+
subscription->VolatileYamlConfigHashes = VolatileYamlConfigHashes;
1320+
1321+
ctx.Send(subscription->Worker, request.Release());
1322+
return true;
12671323
}
1324+
1325+
return false;
12681326
}
12691327

12701328
} // namespace NKikimr::NConsole

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

Lines changed: 44 additions & 3 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
class TConfigsProvider : public TActorBootstrapped<TConfigsProvider> {
@@ -26,6 +27,7 @@ class TConfigsProvider : public TActorBootstrapped<TConfigsProvider> {
2627
EvUpdateYamlConfig,
2728
EvUpdateSubscriptions,
2829
EvWorkerDisconnected,
30+
EvWorkerCoolDown,
2931

3032
EvEnd
3133
};
@@ -59,6 +61,15 @@ class TConfigsProvider : public TActorBootstrapped<TConfigsProvider> {
5961
TInMemorySubscription::TPtr Subscription;
6062
};
6163

64+
struct TEvWorkerCoolDown: public TEventLocal<TEvWorkerCoolDown, EvWorkerCoolDown> {
65+
explicit TEvWorkerCoolDown(TInMemorySubscription::TPtr subscription)
66+
: Subscription(subscription)
67+
{
68+
}
69+
70+
TInMemorySubscription::TPtr Subscription;
71+
};
72+
6273
struct TEvSetConfig : public TEventLocal<TEvSetConfig, EvSetConfig> {
6374
TEvSetConfig(const TConfigsConfig &config)
6475
: Config(config)
@@ -139,9 +150,14 @@ class TConfigsProvider : public TActorBootstrapped<TConfigsProvider> {
139150
const TActorContext &ctx);
140151
void CheckSubscription(TSubscription::TPtr subscriptions,
141152
const TActorContext &ctx);
142-
void CheckSubscription(TInMemorySubscription::TPtr subscriptions,
153+
bool CheckSubscription(TInMemorySubscription::TPtr subscriptions,
143154
const TActorContext &ctx);
144155

156+
bool UpdateConfig(TInMemorySubscription::TPtr subscription,
157+
const TActorContext &ctx);
158+
159+
void ProcessScheduledUpdates(const TActorContext &ctx);
160+
145161
void Handle(NMon::TEvHttpInfo::TPtr &ev);
146162
void Handle(TEvConsole::TEvConfigSubscriptionRequest::TPtr &ev, const TActorContext &ctx);
147163
void Handle(TEvConsole::TEvConfigSubscriptionCanceled::TPtr &ev, const TActorContext &ctx);
@@ -153,6 +169,7 @@ class TConfigsProvider : public TActorBootstrapped<TConfigsProvider> {
153169
void Handle(TEvConsole::TEvGetNodeConfigRequest::TPtr &ev, const TActorContext &ctx);
154170
void Handle(TEvConsole::TEvListConfigSubscriptionsRequest::TPtr &ev, const TActorContext &ctx);
155171
void Handle(TEvPrivate::TEvWorkerDisconnected::TPtr &ev, const TActorContext &ctx);
172+
void Handle(TEvPrivate::TEvWorkerCoolDown::TPtr &ev, const TActorContext &ctx);
156173
void Handle(TEvPrivate::TEvNotificationTimeout::TPtr &ev, const TActorContext &ctx);
157174
void Handle(TEvPrivate::TEvSenderDied::TPtr &ev, const TActorContext &ctx);
158175
void Handle(TEvPrivate::TEvSetConfig::TPtr &ev, const TActorContext &ctx);
@@ -183,6 +200,7 @@ class TConfigsProvider : public TActorBootstrapped<TConfigsProvider> {
183200
HFuncTraced(TEvConsole::TEvGetNodeConfigRequest, Handle);
184201
HFuncTraced(TEvConsole::TEvListConfigSubscriptionsRequest, Handle);
185202
HFuncTraced(TEvPrivate::TEvWorkerDisconnected, Handle);
203+
HFuncTraced(TEvPrivate::TEvWorkerCoolDown, Handle);
186204
HFuncTraced(TEvPrivate::TEvNotificationTimeout, Handle);
187205
HFuncTraced(TEvPrivate::TEvSenderDied, Handle);
188206
HFuncTraced(TEvPrivate::TEvSetConfig, Handle);
@@ -200,9 +218,22 @@ class TConfigsProvider : public TActorBootstrapped<TConfigsProvider> {
200218
}
201219
}
202220

221+
struct TCounters {
222+
using TCounterPtr = ::NMonitoring::TDynamicCounters::TCounterPtr;
223+
TCounterPtr ScheduledConfigUpdates;
224+
TCounterPtr InflightConfigUpdates;
225+
226+
explicit TCounters(::NMonitoring::TDynamicCounterPtr counters)
227+
: ScheduledConfigUpdates(counters->GetCounter("ScheduledConfigUpdates", false))
228+
, InflightConfigUpdates(counters->GetCounter("InflightConfigUpdates", false))
229+
{
230+
}
231+
};
232+
203233
public:
204-
TConfigsProvider(TActorId ownerId)
234+
TConfigsProvider(TActorId ownerId, ::NMonitoring::TDynamicCounterPtr counters)
205235
: ConfigsManager(ownerId)
236+
, Counters(counters)
206237
{
207238
}
208239

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

223254
private:
224255
TActorId ConfigsManager;
256+
TCounters Counters;
225257
TConfigsConfig Config;
226258
TConfigIndex ConfigIndex;
227259
TSubscriptionIndex SubscriptionIndex;
260+
261+
enum class EUpdate {
262+
All,
263+
Yaml,
264+
};
265+
228266
TInMemorySubscriptionIndex InMemoryIndex;
267+
THashMap<TActorId, EUpdate> ScheduledUpdates;
268+
THashSet<TActorId> InflightUpdates;
269+
static constexpr ui32 MAX_INFLIGHT_UPDATES = 50;
229270

230271
TString YamlConfig;
231272
TMap<ui64, TString> VolatileYamlConfigs;

0 commit comments

Comments
 (0)