51 lines
1.7 KiB
Python
51 lines
1.7 KiB
Python
import requests
|
|
|
|
def get_node_status(node_address, port=20202):
|
|
"""
|
|
Queries the LiteFS HTTP API on a specific node for its status.
|
|
Tries /status first, then falls back to /debug/vars.
|
|
"""
|
|
# Try /status first
|
|
url = f"http://{node_address}:{port}/status"
|
|
try:
|
|
response = requests.get(url, timeout=3)
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
status = {
|
|
"is_primary": data.get("primary", False),
|
|
"uptime": data.get("uptime", 0),
|
|
"advertise_url": data.get("advertiseURL", "")
|
|
}
|
|
if "replicationLag" in data:
|
|
status["replication_lag"] = data["replicationLag"]
|
|
if "primaryURL" in data:
|
|
status["primary_url"] = data["primaryURL"]
|
|
return status
|
|
except Exception:
|
|
pass
|
|
|
|
# Fallback to /debug/vars
|
|
url = f"http://{node_address}:{port}/debug/vars"
|
|
try:
|
|
response = requests.get(url, timeout=3)
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
store = data.get("store", {})
|
|
|
|
status = {
|
|
"is_primary": store.get("isPrimary", False),
|
|
"uptime": "N/A", # Not available in /debug/vars
|
|
"advertise_url": f"http://{node_address}:{port}" # Best guess
|
|
}
|
|
|
|
# Look for lag in dbs or store if it exists in other versions
|
|
if "replicationLag" in store:
|
|
status["replication_lag"] = store["replicationLag"]
|
|
|
|
return status
|
|
except Exception as e:
|
|
return {
|
|
"error": str(e),
|
|
"is_primary": False,
|
|
"uptime": "N/A"
|
|
} |