External-Mirrors/awx
2
0
Fork 0
mirror of https://github.com/ansible/awx.git synced 2026-05-08 01:47:35 -02:30
Code Issues Packages Projects Releases Wiki Activity

further optimize callback receiver buffering for certain situations

Browse Source 
see: https://github.com/ansible/awx/issues/9085
This commit is contained in:
Ryan Petrello
2021-01-14 14:43:54 -05:00
parent d88ed19edf
commit b744c4ebb7
2 changed files with 6 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
awx/main/dispatch/worker/callback.py
View File

@@ -38,6 +38,7 @@ class CallbackBrokerWorker(BaseWorker):
MAX_RETRIES = 2 MAX_RETRIES = 2
last_stats = time.time() last_stats = time.time()
last_flush = time.time()
total = 0 total = 0
last_event = '' last_event = ''
prof = None prof = None
@@ -52,7 +53,7 @@ class CallbackBrokerWorker(BaseWorker):
def read(self, queue): def read(self, queue):
try: try:
res = self.redis.blpop(settings.CALLBACK_QUEUE, timeout=settings.JOB_EVENT_BUFFER_SECONDS) res = self.redis.blpop(settings.CALLBACK_QUEUE, timeout=1)
if res is None: if res is None:
return {'event': 'FLUSH'} return {'event': 'FLUSH'}
self.total += 1 self.total += 1
@@ -102,6 +103,7 @@ class CallbackBrokerWorker(BaseWorker):
now = tz_now() now = tz_now()
if ( if (
force or force or
(time.time() - self.last_flush) > settings.JOB_EVENT_BUFFER_SECONDS or
any([len(events) >= 1000 for events in self.buff.values()]) any([len(events) >= 1000 for events in self.buff.values()])
): ):
for cls, events in self.buff.items(): for cls, events in self.buff.items():
@@ -124,6 +126,7 @@ class CallbackBrokerWorker(BaseWorker):
for e in events: for e in events:
emit_event_detail(e) emit_event_detail(e)
self.buff = {} self.buff = {}
self.last_flush = time.time()
def perform_work(self, body): def perform_work(self, body):
try: try:

4
awx/settings/defaults.py
View File

@@ -196,9 +196,9 @@ LOCAL_STDOUT_EXPIRE_TIME = 2592000
# events into the database # events into the database
JOB_EVENT_WORKERS = 4 JOB_EVENT_WORKERS = 4
# The number of seconds (must be an integer) to buffer callback receiver bulk # The number of seconds to buffer callback receiver bulk
# writes in memory before flushing via JobEvent.objects.bulk_create() # writes in memory before flushing via JobEvent.objects.bulk_create()
JOB_EVENT_BUFFER_SECONDS = 1 JOB_EVENT_BUFFER_SECONDS = .1
# The interval at which callback receiver statistics should be # The interval at which callback receiver statistics should be
# recorded # recorded