From df12fb5188893359bd46b1dc274351d4285841e3 Mon Sep 17 00:00:00 2001 From: George Date: Mon, 7 Sep 2026 16:47:19 +0530 Subject: [PATCH] feat: --Added sproket status (experimental) --- instructions.txt | 51 +++++++++++++++------- main.py | 110 +++++++++++++++++++++++++++++++++++++++++++++-- 2 files changed, 142 insertions(+), 19 deletions(-) diff --git a/instructions.txt b/instructions.txt index 9fb6085..e1a4f1d 100644 --- a/instructions.txt +++ b/instructions.txt @@ -251,20 +251,20 @@ psql "postgresql://postgres_exporter:postgres@172.16.10.178:5444/postgres" |FEATURES COMPLETED TODAY| |========================| -| Requirement | `postgres_exporter` | What you need | -| ----------------------- | ------------------: | --------------------------------- | -| Connections | ✅ | postgres_exporter | -| Transactions | ✅ | postgres_exporter | -| Locks | ✅/⚠️ | PostgreSQL metrics/custom query | -| Blocking sessions | ⚠️ | Custom PostgreSQL query | -| Deadlocks | ✅ | postgres_exporter | -| Long-running queries | ⚠️ | `pg_stat_activity` / custom query | -| WAL | ✅ | postgres_exporter | -| Database health | ✅ | postgres_exporter + Prometheus | -| PostgreSQL CPU usage | ❌ | cAdvisor / process monitoring | -| PostgreSQL memory usage | ❌ | cAdvisor / process monitoring | -| Storage/disk usage | ❌ | node_exporter / cAdvisor | -| Network usage | ❌ | node_exporter / cAdvisor | +| Requirement | `postgres_exporter` | What you need | +| ----------------------- | ------------------: | --------------------------------- | +| Connections | ✅ | postgres_exporter | +| Transactions | ✅ | postgres_exporter | +| Locks | ✅/⚠️ | PostgreSQL metrics/custom query | +| Blocking sessions | ⚠️ | Custom PostgreSQL query | +| Deadlocks | ✅ | postgres_exporter | +| Long-running queries | ⚠️ | `pg_stat_activity` / custom query | +| WAL | ✅ | postgres_exporter | +| Database health | ✅ | postgres_exporter + Prometheus | +| PostgreSQL CPU usage | ❌ | cAdvisor / process monitoring | +| PostgreSQL memory usage | ❌ | cAdvisor / process monitoring | +| Storage/disk usage | ❌ | node_exporter / cAdvisor | +| Network usage | ❌ | node_exporter / cAdvisor | |==================| @@ -326,7 +326,7 @@ postgres=# SELECT * FROM spock.replication_set; |===============================| -|CREATE A TABLE NOT USING SPROKE| +|CREATE A TABLE USING SPROKE| |===============================| SELECT spock.replicate_ddl( $$ @@ -339,7 +339,22 @@ CREATE TABLE public.vm2_ddl_test ( $$ ); +|===============================| + |OR NOT USING SPROKE| +|===============================| +ALTER SYSTEM SET spock.enable_ddl_replication = on; +ALTER SYSTEM SET spock.include_ddl_repset = on; + +SELECT pg_reload_conf(); +--------- +VERIFY IT +--------- +SHOW spock.enable_ddl_replication; +SHOW spock.include_ddl_repset; + +|===============================| WTF IS THIS!!! +|===============================| To check if provider is sending data postgres=# SELECT slot_name, @@ -349,4 +364,10 @@ postgres=# SELECT confirmed_flush_lsn FROM pg_replication_slots; +|===========| +| CHECK ACE | +|===========| +sudo docker compose run --rm ace \ + table-diff cluster1 public.test_replication \ + --nodes=node1,node2,node3 \ No newline at end of file diff --git a/main.py b/main.py index 60c51df..ff0b7d0 100644 --- a/main.py +++ b/main.py @@ -1,7 +1,8 @@ -from fastapi import FastAPI, HTTPException +from fastapi import FastAPI, HTTPException,WebSocket, WebSocketDisconnect from pydantic import BaseModel import psycopg2 - +from psycopg2.extras import RealDictCursor +import asyncio app = FastAPI() @@ -195,7 +196,7 @@ def link_nodes(data: LinkNodeRequest): 'default_insert_only', 'ddl_sql' ], - synchronize_structure := false, # CHANGE THIS TO TRUE IF YOU WANT TO SYNCHRONIZE DATA + synchronize_structure := false, synchronize_data := false ); """, ( @@ -231,4 +232,105 @@ def link_nodes(data: LinkNodeRequest): cur.close() if conn: - conn.close() \ No newline at end of file + conn.close() + + +DB_CONFIG = { + "host": "172.16.10.123", + "port": 5444, + "dbname": "postgres", + "user": "postgres", + "password": "postgres" +} + + + + +def get_spock_status(): + conn = psycopg2.connect(**DB_CONFIG) + + cursor = conn.cursor(cursor_factory=RealDictCursor) + + # cursor.execute(""" + # SELECT * + # FROM spock.sub_show_status(); + # """) + + cursor.execute(""" + SELECT + subscription_name, + status, + provider_node + FROM spock.sub_show_status(); + """) + + rows = cursor.fetchall() + + cursor.close() + conn.close() + + return rows + + +@app.websocket("/ws/spock-status") +async def websocket_spock_status(websocket: WebSocket): + await websocket.accept() + + try: + while True: + + try: + status = await asyncio.to_thread(get_spock_status) + + await websocket.send_json({ + "database_reachable": True, + "subscriptions": status + }) + + except Exception as e: + + await websocket.send_json({ + "database_reachable": False, + "subscriptions": [], + "error": str(e) + }) + return False + + await asyncio.sleep(2) + + except WebSocketDisconnect: + print("WebSocket client disconnected") + return True + + + +@app.get("/spock/subscriptions") +def get_spock_subscription_status(): + try: + conn = psycopg2.connect(**DB_CONFIG) + + cursor = conn.cursor(cursor_factory=RealDictCursor) + + cursor.execute(""" + SELECT * + FROM spock.sub_show_status(); + """) + + rows = cursor.fetchall() + + cursor.close() + conn.close() + + return { + "success": True, + "count": len(rows), + "subscriptions": rows + } + + except Exception as e: + raise HTTPException( + status_code=500, + detail=str(e) + ) + +