feat:
--ADDED connections matrix websocket api
This commit is contained in:
1 parent
2ebd83c68d
commit
0313fbfee0
1 file changed
+186
@@ -397,11 +397,197 @@ async def database_status_monitor(websocket: WebSocket):
|
|||||||
print("Database status WebSocket disconnected")
|
print("Database status WebSocket disconnected")
|
||||||
|
|
||||||
|
|
||||||
|
#GET ACTIVE CONNECTIONS
|
||||||
|
|
||||||
|
def get_active_connections(
|
||||||
|
data: DatabaseRequest
|
||||||
|
):
|
||||||
|
|
||||||
|
conn = None
|
||||||
|
cursor = None
|
||||||
|
|
||||||
|
try:
|
||||||
|
conn = psycopg2.connect(
|
||||||
|
host=data.host,
|
||||||
|
port=data.port,
|
||||||
|
dbname=data.dbname,
|
||||||
|
user=data.user,
|
||||||
|
password=data.password,
|
||||||
|
connect_timeout=3
|
||||||
|
)
|
||||||
|
|
||||||
|
cursor = conn.cursor()
|
||||||
|
|
||||||
|
cursor.execute("""
|
||||||
|
SELECT COUNT(*)
|
||||||
|
FROM pg_stat_activity
|
||||||
|
WHERE state = 'active'
|
||||||
|
AND pid <> pg_backend_pid();
|
||||||
|
""")
|
||||||
|
|
||||||
|
active_connections = cursor.fetchone()[0]
|
||||||
|
|
||||||
|
return {
|
||||||
|
"database_name": data.dbname,
|
||||||
|
"active_connections": active_connections
|
||||||
|
}
|
||||||
|
|
||||||
|
except psycopg2.Error as e:
|
||||||
|
|
||||||
|
raise HTTPException(
|
||||||
|
status_code=500,
|
||||||
|
detail=f"Failed to fetch active connections: {str(e)}"
|
||||||
|
)
|
||||||
|
|
||||||
|
finally:
|
||||||
|
if cursor:
|
||||||
|
cursor.close()
|
||||||
|
|
||||||
|
if conn:
|
||||||
|
conn.close()
|
||||||
|
|
||||||
|
|
||||||
|
# HOW MANY CONNECTIONS ARE BEING USED
|
||||||
|
def get_connection_usage(data: DatabaseRequest):
|
||||||
|
|
||||||
|
conn = None
|
||||||
|
cursor = None
|
||||||
|
|
||||||
|
try:
|
||||||
|
|
||||||
|
conn = psycopg2.connect(
|
||||||
|
host=data.host,
|
||||||
|
port=data.port,
|
||||||
|
dbname=data.dbname,
|
||||||
|
user=data.user,
|
||||||
|
password=data.password,
|
||||||
|
connect_timeout=3
|
||||||
|
)
|
||||||
|
|
||||||
|
cursor = conn.cursor()
|
||||||
|
|
||||||
|
cursor.execute("""
|
||||||
|
SELECT
|
||||||
|
COUNT(*) AS current_connections,
|
||||||
|
current_setting('max_connections')::int AS max_connections
|
||||||
|
FROM pg_stat_activity
|
||||||
|
GROUP BY current_setting('max_connections');
|
||||||
|
""")
|
||||||
|
|
||||||
|
row = cursor.fetchone()
|
||||||
|
|
||||||
|
current_connections = row[0]
|
||||||
|
max_connections = row[1]
|
||||||
|
|
||||||
|
connection_usage_percent = round(
|
||||||
|
(current_connections / max_connections) * 100,
|
||||||
|
2
|
||||||
|
)
|
||||||
|
|
||||||
|
return {
|
||||||
|
"current_connections": current_connections,
|
||||||
|
"max_connections": max_connections,
|
||||||
|
"connection_usage_percent": connection_usage_percent
|
||||||
|
}
|
||||||
|
|
||||||
|
except psycopg2.Error as e:
|
||||||
|
|
||||||
|
return {
|
||||||
|
"current_connections": None,
|
||||||
|
"max_connections": None,
|
||||||
|
"connection_usage_percent": None,
|
||||||
|
"error": str(e)
|
||||||
|
}
|
||||||
|
|
||||||
|
finally:
|
||||||
|
|
||||||
|
if cursor:
|
||||||
|
cursor.close()
|
||||||
|
|
||||||
|
if conn:
|
||||||
|
conn.close()
|
||||||
|
|
||||||
|
# WEB SOCKETS FOR ACTIVE CONNECTIONS MONITORING
|
||||||
|
@app.websocket("/ws/database/active-connections")
|
||||||
|
async def database_active_connections_monitor(websocket: WebSocket):
|
||||||
|
|
||||||
|
await websocket.accept()
|
||||||
|
|
||||||
|
try:
|
||||||
|
|
||||||
|
raw_data = await websocket.receive_json()
|
||||||
|
|
||||||
|
data = DatabaseRequest(**raw_data)
|
||||||
|
|
||||||
|
while True:
|
||||||
|
|
||||||
|
try:
|
||||||
|
|
||||||
|
(
|
||||||
|
active_connections_result,
|
||||||
|
connection_usage_result
|
||||||
|
) = await asyncio.wait_for(
|
||||||
|
|
||||||
|
asyncio.gather(
|
||||||
|
|
||||||
|
asyncio.to_thread(
|
||||||
|
get_active_connections,
|
||||||
|
data
|
||||||
|
),
|
||||||
|
|
||||||
|
asyncio.to_thread(
|
||||||
|
get_connection_usage,
|
||||||
|
data
|
||||||
|
)
|
||||||
|
|
||||||
|
),
|
||||||
|
|
||||||
|
timeout=5
|
||||||
|
)
|
||||||
|
|
||||||
|
result = {
|
||||||
|
"database_name": data.dbname,
|
||||||
|
|
||||||
|
"active_connections":
|
||||||
|
active_connections_result[
|
||||||
|
"active_connections"
|
||||||
|
],
|
||||||
|
|
||||||
|
"current_connections":
|
||||||
|
connection_usage_result[
|
||||||
|
"current_connections"
|
||||||
|
],
|
||||||
|
|
||||||
|
"max_connections":
|
||||||
|
connection_usage_result[
|
||||||
|
"max_connections"
|
||||||
|
],
|
||||||
|
|
||||||
|
"connection_usage_percent":
|
||||||
|
connection_usage_result[
|
||||||
|
"connection_usage_percent"
|
||||||
|
]
|
||||||
|
}
|
||||||
|
|
||||||
|
except asyncio.TimeoutError:
|
||||||
|
|
||||||
|
result = {
|
||||||
|
"database_name": data.dbname,
|
||||||
|
"active_connections": None,
|
||||||
|
"current_connections": None,
|
||||||
|
"max_connections": None,
|
||||||
|
"connection_usage_percent": None
|
||||||
|
}
|
||||||
|
|
||||||
|
await websocket.send_json(result)
|
||||||
|
|
||||||
|
await asyncio.sleep(2)
|
||||||
|
|
||||||
|
except WebSocketDisconnect:
|
||||||
|
|
||||||
|
print(
|
||||||
|
"Database active connections WebSocket disconnected"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
Reference in new issue
Block a user