Skip to content

Commit 596b9c6

Browse files
author
nadya02
committed
YT-22072: Add request dropping in http proxies when memory is overloaded
a26b0149d447c736dd6df3a186c4dc36cf58e98e
1 parent 05243a0 commit 596b9c6

File tree

4 files changed

+143
-89
lines changed

4 files changed

+143
-89
lines changed

‎yt/yt/core/http/server.cpp

Lines changed: 110 additions & 69 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,8 @@
1212
#include <yt/yt/core/concurrency/thread_pool_poller.h>
1313

1414
#include <yt/yt/core/misc/finally.h>
15+
#include <yt/yt/core/misc/memory_usage_tracker.h>
16+
#include <yt/yt/core/misc/public.h>
1517

1618
#include <yt/yt/core/ytree/convert.h>
1719

@@ -56,20 +58,22 @@ class TServer
5658
{
5759
public:
5860
TServer(
59-
const TServerConfigPtr& config,
60-
const IListenerPtr& listener,
61-
const IPollerPtr& poller,
62-
const IPollerPtr& acceptor,
63-
const IInvokerPtr& invoker,
64-
const IRequestPathMatcherPtr& requestPathMatcher,
61+
TServerConfigPtr config,
62+
IListenerPtr listener,
63+
IPollerPtr poller,
64+
IPollerPtr acceptor,
65+
IInvokerPtr invoker,
66+
IMemoryUsageTrackerPtr memoryUsageTracker,
67+
IRequestPathMatcherPtr requestPathMatcher,
6568
bool ownPoller = false)
66-
: Config_(config)
67-
, Listener_(listener)
68-
, Poller_(poller)
69-
, Acceptor_(acceptor)
70-
, Invoker_(invoker)
71-
, RequestPathMatcher_(requestPathMatcher)
69+
: Config_(std::move(config))
70+
, Listener_(std::move(listener))
71+
, Poller_(std::move(poller))
72+
, Acceptor_(std::move(acceptor))
73+
, Invoker_(std::move(invoker))
74+
, MemoryUsageTracker_(std::move(memoryUsageTracker))
7275
, OwnPoller_(ownPoller)
76+
, RequestPathMatcher_(std::move(requestPathMatcher))
7377
{ }
7478

7579
void AddHandler(const TString& path, const IHttpHandlerPtr& handler) override
@@ -122,9 +126,10 @@ class TServer
122126
const IPollerPtr Poller_;
123127
const IPollerPtr Acceptor_;
124128
const IInvokerPtr Invoker_;
125-
IRequestPathMatcherPtr RequestPathMatcher_;
129+
const IMemoryUsageTrackerPtr MemoryUsageTracker_;
126130
const bool OwnPoller_ = false;
127131

132+
IRequestPathMatcherPtr RequestPathMatcher_;
128133
bool Started_ = false;
129134
std::atomic<bool> Stopped_ = false;
130135

@@ -220,6 +225,14 @@ class TServer
220225

221226
SetRequestId(response, request->GetRequestId());
222227

228+
if (MemoryUsageTracker_ && MemoryUsageTracker_->IsExceeded()) {
229+
THROW_ERROR_EXCEPTION(
230+
EStatusCode::TooManyRequests,
231+
"Request is dropped due to high memory pressure")
232+
<< TErrorAttribute("total_memory_limit", MemoryUsageTracker_->GetLimit())
233+
<< TErrorAttribute("memory_usage", MemoryUsageTracker_->GetUsed());
234+
}
235+
223236
handler->HandleRequest(request, response);
224237

225238
NTracing::FlushCurrentTraceContextElapsedTime();
@@ -379,36 +392,46 @@ class TServer
379392
////////////////////////////////////////////////////////////////////////////////
380393

381394
IServerPtr CreateServer(
382-
const TServerConfigPtr& config,
383-
const IListenerPtr& listener,
384-
const IPollerPtr& poller,
385-
const IPollerPtr& acceptor,
386-
const IInvokerPtr& invoker,
395+
TServerConfigPtr config,
396+
IListenerPtr listener,
397+
IPollerPtr poller,
398+
IPollerPtr acceptor,
399+
IInvokerPtr invoker,
400+
IMemoryUsageTrackerPtr memoryUsageTracker,
387401
bool ownPoller)
388402
{
389403
auto handlers = New<TRequestPathMatcher>();
390404
return New<TServer>(
391-
config,
392-
listener,
393-
poller,
394-
acceptor,
395-
invoker,
396-
handlers,
405+
std::move(config),
406+
std::move(listener),
407+
std::move(poller),
408+
std::move(acceptor),
409+
std::move(invoker),
410+
std::move(memoryUsageTracker),
411+
std::move(handlers),
397412
ownPoller);
398413
}
399414

400415
IServerPtr CreateServer(
401-
const TServerConfigPtr& config,
402-
const IPollerPtr& poller,
403-
const IPollerPtr& acceptor,
404-
const IInvokerPtr& invoker,
416+
TServerConfigPtr config,
417+
IPollerPtr poller,
418+
IPollerPtr acceptor,
419+
IInvokerPtr invoker,
420+
IMemoryUsageTrackerPtr memoryUsageTracker,
405421
bool ownPoller)
406422
{
407423
auto address = TNetworkAddress::CreateIPv6Any(config->Port);
408424
for (int i = 0;; ++i) {
409425
try {
410426
auto listener = CreateListener(address, poller, acceptor, config->MaxBacklogSize);
411-
return CreateServer(config, listener, poller, acceptor, invoker, ownPoller);
427+
return CreateServer(
428+
std::move(config),
429+
std::move(listener),
430+
std::move(poller),
431+
std::move(acceptor),
432+
std::move(invoker),
433+
std::move(memoryUsageTracker),
434+
ownPoller);
412435
} catch (const std::exception& ex) {
413436
if (i + 1 == config->BindRetryCount) {
414437
throw;
@@ -425,80 +448,98 @@ IServerPtr CreateServer(
425448
////////////////////////////////////////////////////////////////////////////////
426449

427450
IServerPtr CreateServer(
428-
const TServerConfigPtr& config,
429-
const IListenerPtr& listener,
430-
const IPollerPtr& poller)
451+
TServerConfigPtr config,
452+
IListenerPtr listener,
453+
IPollerPtr poller)
431454
{
455+
auto acceptor = poller;
456+
auto invoker = poller->GetInvoker();
432457
return CreateServer(
433-
config,
434-
listener,
435-
poller,
436-
poller,
437-
poller->GetInvoker(),
458+
std::move(config),
459+
std::move(listener),
460+
std::move(poller),
461+
std::move(acceptor),
462+
std::move(invoker),
463+
/*memoryUsageTracker*/ GetNullMemoryUsageTracker(),
438464
/*ownPoller*/ false);
439465
}
440466

441467
IServerPtr CreateServer(
442-
const TServerConfigPtr& config,
443-
const IListenerPtr& listener,
444-
const IPollerPtr& poller,
445-
const IPollerPtr& acceptor)
468+
TServerConfigPtr config,
469+
IListenerPtr listener,
470+
IPollerPtr poller,
471+
IPollerPtr acceptor,
472+
IMemoryUsageTrackerPtr memoryUsageTracker)
446473
{
474+
auto invoker = poller->GetInvoker();
447475
return CreateServer(
448-
config,
449-
listener,
450-
poller,
451-
acceptor,
452-
poller->GetInvoker(),
476+
std::move(config),
477+
std::move(listener),
478+
std::move(poller),
479+
std::move(acceptor),
480+
std::move(invoker),
481+
std::move(memoryUsageTracker),
453482
/*ownPoller*/ false);
454483
}
455484

456-
IServerPtr CreateServer(const TServerConfigPtr& config, const IPollerPtr& poller, const IPollerPtr& acceptor)
485+
IServerPtr CreateServer(
486+
TServerConfigPtr config,
487+
IPollerPtr poller,
488+
IPollerPtr acceptor,
489+
IMemoryUsageTrackerPtr memoryUsageTracker)
457490
{
491+
auto invoker = poller->GetInvoker();
458492
return CreateServer(
459-
config,
460-
poller,
461-
acceptor,
462-
poller->GetInvoker(),
493+
std::move(config),
494+
std::move(poller),
495+
std::move(acceptor),
496+
std::move(invoker),
497+
std::move(memoryUsageTracker),
463498
/*ownPoller*/ false);
464499
}
465500

466-
IServerPtr CreateServer(const TServerConfigPtr& config, const IPollerPtr& poller)
501+
IServerPtr CreateServer(TServerConfigPtr config, IPollerPtr poller)
467502
{
503+
auto acceptor = poller;
468504
return CreateServer(
469-
config,
470-
poller,
471-
poller);
505+
std::move(config),
506+
std::move(poller),
507+
std::move(acceptor));
472508
}
473509

474-
IServerPtr CreateServer(int port, const IPollerPtr& poller)
510+
IServerPtr CreateServer(int port, IPollerPtr poller)
475511
{
476512
auto config = New<TServerConfig>();
477513
config->Port = port;
478-
return CreateServer(config, poller);
514+
return CreateServer(std::move(config), std::move(poller));
479515
}
480516

481-
IServerPtr CreateServer(const TServerConfigPtr& config, int pollerThreadCount)
517+
IServerPtr CreateServer(TServerConfigPtr config, int pollerThreadCount)
482518
{
483519
auto poller = CreateThreadPoolPoller(pollerThreadCount, config->ServerName);
520+
auto acceptor = poller;
521+
auto invoker = poller->GetInvoker();
484522
return CreateServer(
485-
config,
486-
poller,
487-
poller,
488-
poller->GetInvoker(),
523+
std::move(config),
524+
std::move(poller),
525+
std::move(acceptor),
526+
std::move(invoker),
527+
/*memoryUsageTracker*/ GetNullMemoryUsageTracker(),
489528
/*ownPoller*/ true);
490529
}
491530

492531
IServerPtr CreateServer(
493-
const TServerConfigPtr& config,
494-
const NConcurrency::IPollerPtr& poller,
495-
const IInvokerPtr& invoker)
532+
TServerConfigPtr config,
533+
NConcurrency::IPollerPtr poller,
534+
IInvokerPtr invoker)
496535
{
536+
auto acceptor = poller;
497537
return CreateServer(
498-
config,
499-
poller,
500-
poller,
501-
invoker,
538+
std::move(config),
539+
std::move(poller),
540+
std::move(acceptor),
541+
std::move(invoker),
542+
/*memoryUsageTracker*/ GetNullMemoryUsageTracker(),
502543
/*ownPoller*/ false);
503544
}
504545

‎yt/yt/core/http/server.h

Lines changed: 21 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,8 @@
99

1010
#include <yt/yt/core/actions/future.h>
1111

12+
#include <yt/yt/core/misc/public.h>
13+
1214
#include <library/cpp/yt/memory/ref.h>
1315

1416
namespace NYT::NHttp {
@@ -80,31 +82,33 @@ DEFINE_REFCOUNTED_TYPE(IServer)
8082
////////////////////////////////////////////////////////////////////////////////
8183

8284
IServerPtr CreateServer(
83-
const TServerConfigPtr& config,
84-
const NNet::IListenerPtr& listener,
85-
const NConcurrency::IPollerPtr& poller);
85+
TServerConfigPtr config,
86+
NNet::IListenerPtr listener,
87+
NConcurrency::IPollerPtr poller);
8688
IServerPtr CreateServer(
87-
const TServerConfigPtr& config,
88-
const NNet::IListenerPtr& listener,
89-
const NConcurrency::IPollerPtr& poller,
90-
const NConcurrency::IPollerPtr& acceptor);
89+
TServerConfigPtr config,
90+
NNet::IListenerPtr listener,
91+
NConcurrency::IPollerPtr poller,
92+
NConcurrency::IPollerPtr acceptor,
93+
IMemoryUsageTrackerPtr memoryTracker = GetNullMemoryUsageTracker());
9194
IServerPtr CreateServer(
92-
const TServerConfigPtr& config,
93-
const NConcurrency::IPollerPtr& poller);
95+
TServerConfigPtr config,
96+
NConcurrency::IPollerPtr poller);
9497
IServerPtr CreateServer(
95-
const TServerConfigPtr& config,
96-
const NConcurrency::IPollerPtr& poller,
97-
const NConcurrency::IPollerPtr& acceptor);
98+
TServerConfigPtr config,
99+
NConcurrency::IPollerPtr poller,
100+
NConcurrency::IPollerPtr acceptor,
101+
IMemoryUsageTrackerPtr memoryTracker = GetNullMemoryUsageTracker());
98102
IServerPtr CreateServer(
99103
int port,
100-
const NConcurrency::IPollerPtr& poller);
104+
NConcurrency::IPollerPtr poller);
101105
IServerPtr CreateServer(
102-
const TServerConfigPtr& config,
106+
TServerConfigPtr config,
103107
int pollerThreadCount = 1);
104108
IServerPtr CreateServer(
105-
const TServerConfigPtr& config,
106-
const NConcurrency::IPollerPtr& poller,
107-
const IInvokerPtr& invoker);
109+
TServerConfigPtr config,
110+
NConcurrency::IPollerPtr poller,
111+
IInvokerPtr invoker);
108112

109113
////////////////////////////////////////////////////////////////////////////////
110114

‎yt/yt/core/https/server.cpp

Lines changed: 8 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -112,7 +112,8 @@ IServerPtr CreateServer(
112112
const TServerConfigPtr& config,
113113
const IPollerPtr& poller,
114114
const IPollerPtr& acceptor,
115-
const IInvokerPtr& controlInvoker)
115+
const IInvokerPtr& controlInvoker,
116+
const IMemoryUsageTrackerPtr& memoryTracker)
116117
{
117118
auto sslContext = New<TSslContext>();
118119
ApplySslConfig(sslContext, config->Credentials);
@@ -160,7 +161,12 @@ IServerPtr CreateServer(
160161

161162
auto configCopy = CloneYsonStruct(config);
162163
configCopy->IsHttps = true;
163-
auto httpServer = NHttp::CreateServer(configCopy, tlsListener, poller, acceptor);
164+
auto httpServer = NHttp::CreateServer(
165+
configCopy,
166+
tlsListener,
167+
poller,
168+
acceptor,
169+
memoryTracker);
164170

165171
return New<TServer>(std::move(httpServer), std::move(certificateUpdater));
166172
}

‎yt/yt/core/https/server.h

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,8 @@
88

99
#include <yt/yt/core/http/public.h>
1010

11+
#include <yt/yt/core/misc/memory_usage_tracker.h>
12+
1113
namespace NYT::NHttps {
1214

1315
////////////////////////////////////////////////////////////////////////////////
@@ -26,7 +28,8 @@ NHttp::IServerPtr CreateServer(
2628
const TServerConfigPtr& config,
2729
const NConcurrency::IPollerPtr& poller,
2830
const NConcurrency::IPollerPtr& acceptor,
29-
const IInvokerPtr& controlInvoker);
31+
const IInvokerPtr& controlInvoker,
32+
const IMemoryUsageTrackerPtr& memoryTracker = GetNullMemoryUsageTracker());
3033

3134
////////////////////////////////////////////////////////////////////////////////
3235

0 commit comments

Comments
 (0)