Compare commits

...

2 commits

Author SHA1 Message Date
sanj
e76a059f60 Auto-update: Mon Aug 12 21:44:51 PDT 2024 2024-08-12 21:44:51 -07:00
sanj
6c6c3a7b65 Auto-update: Mon Aug 12 21:28:09 PDT 2024 2024-08-12 21:28:09 -07:00
3 changed files with 80 additions and 133 deletions

View file

@ -1,5 +1,3 @@
# database.py
import json
import yaml
import time
@ -20,8 +18,6 @@ from zoneinfo import ZoneInfo
from srtm import get_data
import os
import sys
from sqlalchemy.dialects.postgresql import UUID
import uuid
from loguru import logger
from sqlalchemy import text, select, func, and_
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
@ -45,18 +41,15 @@ ENV_PATH = CONFIG_DIR / ".env"
load_dotenv(ENV_PATH)
TS_ID = os.environ.get('TS_ID')
class QueryTracking(Base):
__tablename__ = 'query_tracking'
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
id = Column(Integer, primary_key=True)
ts_id = Column(String, nullable=False)
query = Column(Text, nullable=False)
args = Column(JSON)
args = Column(JSONB)
executed_at = Column(DateTime(timezone=True), server_default=func.now())
completed_by = Column(JSON, default={})
result_checksum = Column(String(32)) # MD5 checksum
completed_by = Column(JSONB, default={})
class Database:
@classmethod
@ -109,9 +102,11 @@ class Database:
async with engine.connect() as conn:
await conn.execute(text("SELECT 1"))
online_servers.append(ts_id)
l.debug(f"Server {ts_id} is online")
except OperationalError:
pass
l.warning(f"Server {ts_id} is offline")
self.online_servers = set(online_servers)
l.info(f"Online servers: {', '.join(online_servers)}")
return online_servers
async def read(self, query: str, **kwargs):
@ -144,15 +139,8 @@ class Database:
result = await session.execute(text(query), serialized_kwargs)
await session.commit()
# Calculate result checksum
result_str = str(result.fetchall())
result_checksum = hashlib.md5(result_str.encode()).hexdigest()
# Add the write query to the query_tracking table
await self.add_query_to_tracking(query, serialized_kwargs, result_checksum)
# Initiate async operations
asyncio.create_task(self._async_sync_operations())
asyncio.create_task(self._async_sync_operations(query, kwargs))
# Return the result
return result
@ -165,53 +153,74 @@ class Database:
l.error(f"Traceback: {traceback.format_exc()}")
return None
async def _async_sync_operations(self):
async def _async_sync_operations(self, query: str, kwargs: dict):
try:
# Add the write query to the query_tracking table
await self.add_query_to_tracking(query, kwargs)
# Call /db/sync on all online servers
await self.call_db_sync_on_servers()
except Exception as e:
l.error(f"Error in async sync operations: {str(e)}")
l.error(f"Traceback: {traceback.format_exc()}")
async def add_query_to_tracking(self, query: str, kwargs: dict, result_checksum: str):
async def add_query_to_tracking(self, query: str, kwargs: dict):
async with self.sessions[self.local_ts_id]() as session:
new_query = QueryTracking(
ts_id=self.local_ts_id,
query=query,
args=json_dumps(kwargs),
completed_by={self.local_ts_id: True},
result_checksum=result_checksum
completed_by={self.local_ts_id: True}
)
session.add(new_query)
await session.commit()
l.info(f"Added query to tracking: {query[:50]}...")
async def pull_query_tracking_from_primary(self):
primary_ts_id = await self.get_primary_server()
if not primary_ts_id:
l.error("Failed to get primary server")
async def sync_db(self):
current_time = time.time()
if current_time - self.last_sync_time < 30:
l.info("Skipping sync, last sync was less than 30 seconds ago")
return
primary_server = next((s for s in self.config['POOL'] if s['ts_id'] == primary_ts_id), None)
if not primary_server:
l.error(f"Primary server {primary_ts_id} not found in config")
return
try:
l.info("Starting database synchronization")
await self.pull_query_tracking_from_all_servers()
await self.execute_unexecuted_queries()
self.last_sync_time = current_time
l.info("Database synchronization completed successfully")
except Exception as e:
l.error(f"Error during database sync: {str(e)}")
l.error(f"Traceback: {traceback.format_exc()}")
async with self.sessions[primary_ts_id]() as session:
queries = await session.execute(select(QueryTracking))
queries = queries.fetchall()
async with self.sessions[self.local_ts_id]() as local_session:
for query in queries:
existing = await local_session.get(QueryTracking, query.id)
if existing:
existing.completed_by = {**existing.completed_by, **query.completed_by}
else:
local_session.add(query)
await local_session.commit()
async def pull_query_tracking_from_all_servers(self):
online_servers = await self.get_online_servers()
l.info(f"Pulling query tracking from {len(online_servers)} online servers")
for server_id in online_servers:
if server_id == self.local_ts_id:
continue # Skip local server
l.info(f"Pulling queries from server: {server_id}")
async with self.sessions[server_id]() as remote_session:
queries = await remote_session.execute(select(QueryTracking))
queries = queries.fetchall()
l.info(f"Retrieved {len(queries)} queries from server {server_id}")
async with self.sessions[self.local_ts_id]() as local_session:
for query in queries:
existing = await local_session.execute(
select(QueryTracking).where(QueryTracking.id == query.id)
)
existing = existing.scalar_one_or_none()
if existing:
existing.completed_by = {**existing.completed_by, **query.completed_by}
l.debug(f"Updated existing query: {query.id}")
else:
local_session.add(query)
l.debug(f"Added new query: {query.id}")
await local_session.commit()
l.info("Finished pulling queries from all servers")
async def execute_unexecuted_queries(self):
async with self.sessions[self.local_ts_id]() as session:
@ -220,35 +229,30 @@ class Database:
)
unexecuted_queries = unexecuted_queries.fetchall()
l.info(f"Executing {len(unexecuted_queries)} unexecuted queries")
for query in unexecuted_queries:
try:
params = json.loads(query.args)
result = await session.execute(text(query.query), params)
# Validate result checksum
result_str = str(result.fetchall())
result_checksum = hashlib.md5(result_str.encode()).hexdigest()
if result_checksum == query.result_checksum:
query.completed_by[self.local_ts_id] = True
await session.commit()
l.info(f"Successfully executed query ID {query.id}")
else:
l.error(f"Checksum mismatch for query ID {query.id}")
await session.rollback()
await session.execute(text(query.query), params)
query.completed_by[self.local_ts_id] = True
await session.commit()
l.info(f"Successfully executed query ID {query.id}")
except Exception as e:
l.error(f"Failed to execute query ID {query.id}: {str(e)}")
await session.rollback()
l.info("Finished executing unexecuted queries")
async def call_db_sync_on_servers(self):
"""Call /db/sync on all online servers."""
online_servers = await self.get_online_servers()
l.info(f"Calling /db/sync on {len(online_servers)} online servers")
for server in self.config['POOL']:
if server['ts_id'] in online_servers and server['ts_id'] != self.local_ts_id:
try:
await self.call_db_sync(server)
except Exception as e:
l.error(f"Failed to call /db/sync on {server['ts_id']}: {str(e)}")
l.info("Finished calling /db/sync on all servers")
async def call_db_sync(self, server):
url = f"http://{server['ts_ip']}:{server['app_port']}/db/sync"
@ -267,20 +271,6 @@ class Database:
except Exception as e:
l.error(f"Error calling /db/sync on {url}: {str(e)}")
async def sync_db(self):
current_time = time.time()
if current_time - self.last_sync_time < 30:
l.info("Skipping sync, last sync was less than 30 seconds ago")
return
try:
await self.pull_query_tracking_from_all_servers()
await self.execute_unexecuted_queries()
self.last_sync_time = current_time
except Exception as e:
l.error(f"Error during database sync: {str(e)}")
l.error(f"Traceback: {traceback.format_exc()}")
async def ensure_query_tracking_table(self):
for ts_id, engine in self.engines.items():
try:
@ -289,33 +279,8 @@ class Database:
l.info(f"Ensured query_tracking table exists for {ts_id}")
except Exception as e:
l.error(f"Failed to create query_tracking table for {ts_id}: {str(e)}")
async def pull_query_tracking_from_all_servers(self):
online_servers = await self.get_online_servers()
for server_id in online_servers:
if server_id == self.local_ts_id:
continue # Skip local server
async with self.sessions[server_id]() as remote_session:
queries = await remote_session.execute(select(QueryTracking))
queries = queries.fetchall()
async with self.sessions[self.local_ts_id]() as local_session:
for query in queries:
existing = await local_session.execute(
select(QueryTracking).where(QueryTracking.id == query.id)
)
existing = existing.scalar_one_or_none()
if existing:
existing.completed_by = {**existing.completed_by, **query.completed_by}
else:
local_session.add(query)
await local_session.commit()
async def close(self):
for engine in self.engines.values():
await engine.dispose()
l.info("Closed all database connections")

View file

@ -67,37 +67,9 @@ async def get_tailscale_ip():
else:
return "No devices found"
async def sync_process():
async with Db.sessions[TS_ID]() as session:
# Find unexecuted queries
unexecuted_queries = await session.execute(
select(QueryTracking).where(~QueryTracking.completed_by.has_key(TS_ID)).order_by(QueryTracking.id)
)
for query in unexecuted_queries:
try:
params = json_loads(query.args)
await session.execute(text(query.query), params)
actual_checksum = await Db._local_compute_checksum(query.query, params)
if actual_checksum != query.result_checksum:
l.error(f"Checksum mismatch for query ID {query.id}")
continue
# Update the completed_by field
query.completed_by[TS_ID] = True
await session.commit()
l.info(f"Successfully executed and verified query ID {query.id}")
except Exception as e:
l.error(f"Failed to execute query ID {query.id} during sync: {str(e)}")
await session.rollback()
l.info(f"Sync process completed. Executed {unexecuted_queries.rowcount} queries.")
# After executing all queries, perform combinatorial sync
await Db.sync_query_tracking()
@sys.post("/db/sync")
async def db_sync(background_tasks: BackgroundTasks):
background_tasks.add_task(sync_process)
return {"message": "Sync process initiated"}
l.info(f"Received request to /db/sync")
background_tasks.add_task(Db.sync_db)
return {"message": "Sync process initiated"}

View file

@ -205,7 +205,12 @@ async def store_weather_to_db(date_time: dt_datetime, weather_data: dict):
if daily_weather_result is None:
raise ValueError("Failed to insert daily weather data: no result returned")
daily_weather_id = daily_weather_result.fetchone()[0]
daily_weather_row = daily_weather_result.fetchone()
if daily_weather_row is None:
raise ValueError("Failed to retrieve inserted daily weather ID: fetchone() returned None")
daily_weather_id = daily_weather_row[0]
l.debug(f"Inserted daily weather data with id: {daily_weather_id}")
# Hourly weather insertion
@ -262,8 +267,12 @@ async def store_weather_to_db(date_time: dt_datetime, weather_data: dict):
if hourly_result is None:
l.warning(f"Failed to insert hourly weather data for {hour_data.get('datetimeEpoch')}")
else:
hourly_id = hourly_result.fetchone()[0]
l.debug(f"Inserted hourly weather data with id: {hourly_id}")
hourly_row = hourly_result.fetchone()
if hourly_row is None:
l.warning(f"Failed to retrieve inserted hourly weather ID for {hour_data.get('datetimeEpoch')}")
else:
hourly_id = hourly_row[0]
l.debug(f"Inserted hourly weather data with id: {hourly_id}")
return "SUCCESS"
except Exception as e:
@ -271,6 +280,7 @@ async def store_weather_to_db(date_time: dt_datetime, weather_data: dict):
l.error(f"Traceback: {traceback.format_exc()}")
return "FAILURE"
async def get_weather_from_db(date_time: dt_datetime, latitude: float, longitude: float):
l.debug(f"Using {date_time.strftime('%Y-%m-%d %H:%M:%S')} as our datetime in get_weather_from_db.")
query_date = date_time.date()