.BANCHO. Add check for MySQL connections pool saturation
This commit is contained in:
parent
6020f7cc47
commit
f912f6ea82
|
@ -1,14 +1,16 @@
|
|||
import MySQLdb
|
||||
import threading
|
||||
import glob
|
||||
from helpers import logHelper as log
|
||||
import threading
|
||||
|
||||
class mysqlWorker:
|
||||
"""
|
||||
Instance of a mysql worker
|
||||
Instance of a pettirosso meme
|
||||
"""
|
||||
def __init__(self, wid, host, username, password, database):
|
||||
"""
|
||||
Create a mysql worker
|
||||
Create a pettirosso meme (mysql worker)
|
||||
|
||||
wid -- worker id
|
||||
host -- hostname
|
||||
|
@ -26,6 +28,7 @@ class db:
|
|||
"""
|
||||
A MySQL db connection with multiple workers
|
||||
"""
|
||||
|
||||
def __init__(self, host, username, password, database, workers):
|
||||
"""
|
||||
Create MySQL workers aka pettirossi meme
|
||||
|
@ -39,10 +42,24 @@ class db:
|
|||
self.workers = []
|
||||
self.lastWorker = 0
|
||||
self.workersNumber = workers
|
||||
self.locked = 0
|
||||
for i in range(0,self.workersNumber):
|
||||
print(".", end="")
|
||||
self.workers.append(mysqlWorker(i, host, username, password, database))
|
||||
|
||||
def checkPoolSaturation(self):
|
||||
"""
|
||||
Check the number of busy connections in connections pool.
|
||||
If the pool is 100% busy, log a message to sentry
|
||||
"""
|
||||
if self.locked >= (self.workersNumber-1):
|
||||
msg = "MySQL connections pool is saturated!".format(self.locked, self.workersNumber)
|
||||
log.warning(msg)
|
||||
glob.application.sentry_client.captureMessage(msg, level="warning", extra={
|
||||
"workersBusy": self.locked,
|
||||
"workersTotal": self.workersNumber
|
||||
})
|
||||
|
||||
def getWorker(self):
|
||||
"""
|
||||
Return a worker object (round-robin way)
|
||||
|
@ -53,6 +70,10 @@ class db:
|
|||
self.lastWorker = 0
|
||||
else:
|
||||
self.lastWorker += 1
|
||||
|
||||
# Saturation check
|
||||
threading.Thread(target=self.checkPoolSaturation).start()
|
||||
self.locked += 1
|
||||
return self.workers[self.lastWorker]
|
||||
|
||||
def execute(self, query, params = ()):
|
||||
|
@ -77,8 +98,9 @@ class db:
|
|||
if cursor:
|
||||
cursor.close()
|
||||
worker.lock.release()
|
||||
self.locked -= 1
|
||||
|
||||
def fetch(self, query, params = (), all_ = False):
|
||||
def fetch(self, query, params = (), all = False):
|
||||
"""
|
||||
Fetch a single value from db that matches given query
|
||||
|
||||
|
@ -95,7 +117,7 @@ class db:
|
|||
# Create cursor, execute the query and fetch one/all result(s)
|
||||
cursor = worker.connection.cursor(MySQLdb.cursors.DictCursor)
|
||||
cursor.execute(query, params)
|
||||
if all_:
|
||||
if all == True:
|
||||
return cursor.fetchall()
|
||||
else:
|
||||
return cursor.fetchone()
|
||||
|
@ -104,6 +126,7 @@ class db:
|
|||
if cursor:
|
||||
cursor.close()
|
||||
worker.lock.release()
|
||||
self.locked -= 1
|
||||
|
||||
def fetchAll(self, query, params = ()):
|
||||
"""
|
||||
|
|
|
@ -3,6 +3,8 @@ import tornado.web
|
|||
import tornado.gen
|
||||
from tornado.ioloop import IOLoop
|
||||
from objects import glob
|
||||
import threading
|
||||
from helpers import logHelper as log
|
||||
|
||||
class asyncRequestHandler(tornado.web.RequestHandler):
|
||||
"""
|
||||
|
@ -54,8 +56,25 @@ def runBackground(data, callback):
|
|||
"""
|
||||
func, args, kwargs = data
|
||||
def _callback(result):
|
||||
#glob.busyThreads -= 1
|
||||
IOLoop.instance().add_callback(lambda: callback(result))
|
||||
glob.pool.apply_async(func, args, kwargs, _callback)
|
||||
#threading.Thread(target=checkPoolSaturation).start()
|
||||
#glob.busyThreads += 1
|
||||
|
||||
def checkPoolSaturation():
|
||||
"""
|
||||
Check the number of busy threads in connections pool.
|
||||
If the pool is 100% busy, log a message to sentry
|
||||
"""
|
||||
size = int(glob.conf.config["server"]["threads"])
|
||||
if glob.busyThreads >= size:
|
||||
msg = "Connections threads pool is saturated!"
|
||||
log.warning(msg)
|
||||
glob.application.sentry_client.captureMessage(msg, level="warning", extra={
|
||||
"workersBusy": glob.busyThreads,
|
||||
"workersTotal": size
|
||||
})
|
||||
|
||||
def checkArguments(arguments, requiredArguments):
|
||||
"""
|
||||
|
|
|
@ -15,6 +15,7 @@ try:
|
|||
except:
|
||||
VERSION = "¯\_(xd)_/¯"
|
||||
|
||||
application = None
|
||||
db = None
|
||||
conf = None
|
||||
banchoConf = None
|
||||
|
@ -29,6 +30,7 @@ cloudflare = False
|
|||
chatFilters = None
|
||||
userIDCache = {}
|
||||
pool = None
|
||||
busyThreads = 0
|
||||
|
||||
debug = False
|
||||
outputRequestTime = False
|
||||
|
|
6
pep.py
6
pep.py
|
@ -174,13 +174,13 @@ if __name__ == "__main__":
|
|||
consoleHelper.printColored("[!] Warning! Server running in debug mode!", bcolors.YELLOW)
|
||||
|
||||
# Make app
|
||||
application = make_app()
|
||||
glob.application = make_app()
|
||||
|
||||
# Set up sentry
|
||||
try:
|
||||
glob.sentry = generalFunctions.stringToBool(glob.conf.config["sentry"]["enable"])
|
||||
if glob.sentry:
|
||||
application.sentry_client = AsyncSentryClient(glob.conf.config["sentry"]["banchodns"], release=glob.VERSION)
|
||||
glob.application.sentry_client = AsyncSentryClient(glob.conf.config["sentry"]["banchodns"], release=glob.VERSION)
|
||||
else:
|
||||
consoleHelper.printColored("[!] Warning! Sentry logging is disabled!", bcolors.YELLOW)
|
||||
except:
|
||||
|
@ -214,7 +214,7 @@ if __name__ == "__main__":
|
|||
consoleHelper.printColored("> Tornado listening for HTTP(s) clients on 127.0.0.1:{}...".format(serverPort), bcolors.GREEN)
|
||||
|
||||
# Start tornado
|
||||
application.listen(serverPort)
|
||||
glob.application.listen(serverPort)
|
||||
tornado.ioloop.IOLoop.instance().start()
|
||||
finally:
|
||||
system.dispose()
|
Loading…
Reference in New Issue
Block a user