diff --git a/instructions.txt b/instructions.txt index e5e8824..9680ab6 100644 --- a/instructions.txt +++ b/instructions.txt @@ -15,4 +15,8 @@ pip freeze | grep psycopg deactivate -source /home/george/Desktop/PROJECT26/psqlDB/venv/bin/activate \ No newline at end of file +source /home/george/Desktop/PROJECT26/psqlDB/venv/bin/activate + + +INSTALL IN DB, IF NOT DONE!!! +CREATE EXTENSION IF NOT EXISTS pg_stat_statements; \ No newline at end of file diff --git a/main.py b/main.py index 9860dd4..98c43d6 100644 --- a/main.py +++ b/main.py @@ -791,6 +791,567 @@ async def database_performance_monitor(websocket: WebSocket): print("Database performance WebSocket disconnected") +#LOCKS +def get_database_locks(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 total_locks, + + COUNT(*) FILTER ( + WHERE l.granted = TRUE + ) AS granted_locks, + + COUNT(*) FILTER ( + WHERE l.granted = FALSE + ) AS waiting_locks + + FROM pg_locks l + + JOIN pg_stat_activity a + ON l.pid = a.pid + + WHERE a.datname = %s; + """, (data.dbname,)) + + row = cursor.fetchone() + + return { + "database_name": data.dbname, + "total_locks": row[0], + "granted_locks": row[1], + "waiting_locks": row[2] + } + + except psycopg2.Error as e: + + return { + "database_name": data.dbname, + "total_locks": None, + "granted_locks": None, + "waiting_locks": None, + "error": str(e) + } + + finally: + + if cursor: + cursor.close() + + if conn: + conn.close() + + +# DEADLOCK COUNT +def get_database_deadlocks(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 deadlocks + FROM pg_stat_database + WHERE datname = %s; + """, (data.dbname,)) + + row = cursor.fetchone() + + deadlocks = row[0] if row else 0 + + return { + "database_name": data.dbname, + "deadlocks": deadlocks + } + + except psycopg2.Error as e: + + return { + "database_name": data.dbname, + "deadlocks": None, + "error": str(e) + } + + finally: + + if cursor: + cursor.close() + + if conn: + conn.close() + +# LOCKS AND DEADLOCKS WEB SOCKET +@app.websocket("/ws/database/locks-deadlocks") +async def database_locks_deadlocks_monitor(websocket: WebSocket): + + await websocket.accept() + + try: + + raw_data = await websocket.receive_json() + + data = DatabaseRequest(**raw_data) + + while True: + + try: + + ( + locks_result, + deadlocks_result + ) = await asyncio.wait_for( + + asyncio.gather( + + asyncio.to_thread( + get_database_locks, + data + ), + + asyncio.to_thread( + get_database_deadlocks, + data + ) + + ), + + timeout=5 + ) + + result = { + "database_name": data.dbname, + + "total_locks": + locks_result["total_locks"], + + "granted_locks": + locks_result["granted_locks"], + + "waiting_locks": + locks_result["waiting_locks"], + + "deadlocks": + deadlocks_result["deadlocks"] + } + + except asyncio.TimeoutError: + + result = { + "database_name": data.dbname, + "total_locks": None, + "granted_locks": None, + "waiting_locks": None, + "deadlocks": None + } + + await websocket.send_json(result) + + await asyncio.sleep(2) + + except WebSocketDisconnect: + + print("Locks and deadlocks WebSocket disconnected") + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +#-========================================================================================================================================================================================= +#EXPERIMENTAL FUNTIONS +# QUERY LATENCY +def get_query_latency(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 + CASE + WHEN SUM(calls) = 0 THEN 0 + ELSE ROUND( + SUM(total_exec_time)::numeric / + SUM(calls), + 2 + ) + END + FROM pg_stat_statements + WHERE dbid = ( + SELECT oid + FROM pg_database + WHERE datname = %s + ); + """, (data.dbname,)) + + row = cursor.fetchone() + + query_latency_ms = ( + float(row[0]) + if row and row[0] is not None + else 0.0 + ) + + return { + "database_name": data.dbname, + "query_latency_ms": query_latency_ms + } + + except psycopg2.Error as e: + + return { + "database_name": data.dbname, + "query_latency_ms": None, + "error": str(e) + } + + finally: + + if cursor: + cursor.close() + + if conn: + conn.close() + +#SLOW RUNNING QUERIES + +def get_slow_queries( + data: DatabaseRequest, + threshold_seconds: int = 1 +): + + 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_factory=RealDictCursor + ) + + cursor.execute(""" + SELECT + pid, + usename, + application_name, + client_addr, + state, + query, + ROUND( + EXTRACT( + EPOCH FROM ( + CURRENT_TIMESTAMP - query_start + ) + )::numeric * 1000, + 2 + ) AS duration_ms, + query_start + FROM pg_stat_activity + WHERE datname = %s + AND state = 'active' + AND pid <> pg_backend_pid() + AND query_start IS NOT NULL + AND CURRENT_TIMESTAMP - query_start + > (%s * INTERVAL '1 second') + ORDER BY query_start ASC; + """, ( + data.dbname, + threshold_seconds + )) + + rows = cursor.fetchall() + + return { + "database_name": data.dbname, + "threshold_seconds": threshold_seconds, + "slow_query_count": len(rows), + "slow_queries": [ + { + "pid": row["pid"], + "user": row["usename"], + "application_name": row["application_name"], + "client_addr": ( + str(row["client_addr"]) + if row["client_addr"] + else None + ), + "state": row["state"], + "query": row["query"], + "duration_ms": float(row["duration_ms"]), + "query_start": row["query_start"].isoformat() + } + for row in rows + ] + } + + except psycopg2.Error as e: + + return { + "database_name": data.dbname, + "threshold_seconds": threshold_seconds, + "slow_query_count": None, + "slow_queries": [], + "error": str(e) + } + + finally: + + if cursor: + cursor.close() + + if conn: + conn.close() + + +#BLOCKED QUERIES + +def get_blocked_queries(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_factory=RealDictCursor + ) + + cursor.execute(""" + SELECT + blocked.pid AS blocked_pid, + blocked.usename AS blocked_user, + blocked.application_name AS blocked_application, + blocked.client_addr AS blocked_client_addr, + blocked.query AS blocked_query, + blocked.query_start AS blocked_query_start, + blocked.wait_event_type, + blocked.wait_event, + + ROUND( + EXTRACT( + EPOCH FROM ( + CURRENT_TIMESTAMP - blocked.query_start + ) + )::numeric * 1000, + 2 + ) AS blocked_duration_ms, + + pg_blocking_pids(blocked.pid) AS blocking_pids + + FROM pg_stat_activity blocked + + WHERE blocked.datname = %s + AND blocked.pid <> pg_backend_pid() + AND cardinality( + pg_blocking_pids(blocked.pid) + ) > 0 + + ORDER BY blocked.query_start ASC; + """, (data.dbname,)) + + rows = cursor.fetchall() + + return { + "database_name": data.dbname, + "blocked_query_count": len(rows), + + "blocked_queries": [ + { + "blocked_pid": row["blocked_pid"], + + "blocked_user": + row["blocked_user"], + + "blocked_application": + row["blocked_application"], + + "blocked_client_addr": + str(row["blocked_client_addr"]) + if row["blocked_client_addr"] + else None, + + "blocked_query": + row["blocked_query"], + + "blocked_query_start": + row["blocked_query_start"].isoformat() + if row["blocked_query_start"] + else None, + + "blocked_duration_ms": + float(row["blocked_duration_ms"]) + if row["blocked_duration_ms"] + is not None + else None, + + "wait_event_type": + row["wait_event_type"], + + "wait_event": + row["wait_event"], + + "blocking_pids": + row["blocking_pids"] + } + for row in rows + ] + } + + except psycopg2.Error as e: + + return { + "database_name": data.dbname, + "blocked_query_count": None, + "blocked_queries": [], + "error": str(e) + } + + finally: + + if cursor: + cursor.close() + + if conn: + conn.close() + +#======================================================================================================================================================================================================= + if __name__ == "__main__": host = os.getenv("HOST", "127.0.0.1") port = int(os.getenv("PORT", 9000))