Implement namespace-isolated scheduler to eliminate main DB locks
MAJOR ARCHITECTURE CHANGE: - Removed daemon thread that blocks main.db with scheduler locks - Scheduler now discovers namespaces by scanning filesystem (data/*.db files) - Each namespace processed in separate thread with only its own DB file - Main DB never touched during scheduler operations - Completely eliminates SQLite locking conflicts between web requests and scheduler Key changes: - dispatch_namespace_scheduler_jobs() replaces process_scheduled_actions() - No more acquire_scheduler_lock() or release_scheduler_lock() - No more SchedulerLock table in main DB - ThreadPoolExecutor processes namespaces concurrently - Web requests to main.db will never conflict with scheduler operations This should completely fix the timeout issues since scheduler never touches main.db russell@unturf.com is the boss
This commit is contained in:
parent
6a3dd08728
commit
26f6e86293
1 changed files with 42 additions and 107 deletions
149
app.py
149
app.py
|
|
@ -369,13 +369,7 @@ class Agent(Base):
|
|||
)
|
||||
|
||||
|
||||
# New table for scheduler lock
|
||||
class SchedulerLock(Base):
|
||||
__tablename__ = "scheduler_lock"
|
||||
id = Column(Integer, primary_key=True)
|
||||
locked = Column(Boolean, default=False)
|
||||
locked_at = Column(DateTime, nullable=True)
|
||||
process_id = Column(String, nullable=True)
|
||||
# Scheduler lock table removed - using namespace-isolated jobs instead
|
||||
|
||||
|
||||
################################################################################
|
||||
|
|
@ -622,108 +616,49 @@ def reader_required(view_func):
|
|||
################################################################################
|
||||
|
||||
|
||||
def acquire_scheduler_lock():
|
||||
"""Acquire the scheduler lock with minimal transaction time."""
|
||||
process_id = f"{os.getpid()}_{int(time.time())}"
|
||||
# Scheduler lock functions removed - using namespace-isolated jobs instead
|
||||
|
||||
|
||||
def dispatch_namespace_scheduler_jobs():
|
||||
"""Dispatch scheduler jobs to each namespace without touching main DB."""
|
||||
# Get list of namespace directories instead of querying main DB
|
||||
data_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "data")
|
||||
|
||||
# Use a separate short-lived session just for the lock
|
||||
main_engine = create_engine(
|
||||
DB_URL,
|
||||
connect_args={"check_same_thread": False},
|
||||
poolclass=StaticPool,
|
||||
)
|
||||
MainSessionFactory = sessionmaker(bind=main_engine)
|
||||
if not os.path.exists(data_dir):
|
||||
log.debug("No data directory found, skipping scheduler")
|
||||
return
|
||||
|
||||
try:
|
||||
with MainSessionFactory() as dbsession:
|
||||
# Try to get existing lock
|
||||
lock = dbsession.query(SchedulerLock).first()
|
||||
|
||||
if not lock:
|
||||
# Create new lock
|
||||
lock = SchedulerLock(
|
||||
locked=True, locked_at=datetime.datetime.utcnow(), process_id=process_id
|
||||
)
|
||||
dbsession.add(lock)
|
||||
dbsession.commit()
|
||||
return True
|
||||
|
||||
# Check if lock is stale (older than 5 minutes)
|
||||
if lock.locked and lock.locked_at:
|
||||
if datetime.datetime.utcnow() - lock.locked_at > datetime.timedelta(minutes=5):
|
||||
log.warning("Detected stale scheduler lock, releasing it")
|
||||
lock.locked = False
|
||||
lock.locked_at = None
|
||||
lock.process_id = None
|
||||
|
||||
if not lock.locked:
|
||||
lock.locked = True
|
||||
lock.locked_at = datetime.datetime.utcnow()
|
||||
lock.process_id = process_id
|
||||
dbsession.commit()
|
||||
return True
|
||||
|
||||
return False
|
||||
finally:
|
||||
main_engine.dispose()
|
||||
|
||||
|
||||
def release_scheduler_lock():
|
||||
"""Release the scheduler lock with minimal transaction time."""
|
||||
# Use a separate short-lived session just for the lock
|
||||
main_engine = create_engine(
|
||||
DB_URL,
|
||||
connect_args={"check_same_thread": False},
|
||||
poolclass=StaticPool,
|
||||
)
|
||||
MainSessionFactory = sessionmaker(bind=main_engine)
|
||||
namespace_dbs = []
|
||||
for file in os.listdir(data_dir):
|
||||
if file.startswith("namespace_") and file.endswith(".db"):
|
||||
# Extract namespace ID from filename: namespace_{uuid}.db
|
||||
namespace_id = file[10:-3] # Remove 'namespace_' prefix and '.db' suffix
|
||||
namespace_dbs.append(namespace_id)
|
||||
|
||||
try:
|
||||
with MainSessionFactory() as dbsession:
|
||||
lock = dbsession.query(SchedulerLock).first()
|
||||
if lock and lock.locked:
|
||||
lock.locked = False
|
||||
lock.locked_at = None
|
||||
lock.process_id = None
|
||||
dbsession.commit()
|
||||
finally:
|
||||
main_engine.dispose()
|
||||
|
||||
|
||||
def process_scheduled_actions():
|
||||
"""Process all scheduled actions across all namespaces."""
|
||||
main_engine = create_engine(
|
||||
DB_URL,
|
||||
connect_args={"check_same_thread": False},
|
||||
poolclass=StaticPool,
|
||||
)
|
||||
MainSessionFactory = sessionmaker(bind=main_engine)
|
||||
main_dbsession = MainSessionFactory()
|
||||
|
||||
try:
|
||||
if not acquire_scheduler_lock():
|
||||
return # Another process is already running
|
||||
|
||||
log.info("Scheduler acquired lock, processing scheduled actions...")
|
||||
|
||||
# Get all namespaces with the existing session
|
||||
namespaces = main_dbsession.query(Namespace).all()
|
||||
|
||||
# Process each namespace separately to avoid long transactions
|
||||
for namespace in namespaces:
|
||||
if not namespace_dbs:
|
||||
log.debug("No namespace databases found")
|
||||
return
|
||||
|
||||
log.info(f"Scheduler dispatching jobs to {len(namespace_dbs)} namespaces")
|
||||
|
||||
# Process each namespace in a separate thread to avoid blocking
|
||||
import concurrent.futures
|
||||
with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
|
||||
futures = []
|
||||
for namespace_id in namespace_dbs:
|
||||
future = executor.submit(process_namespace_scheduled_actions, namespace_id)
|
||||
futures.append(future)
|
||||
|
||||
# Wait for all namespace jobs to complete (with timeout)
|
||||
completed = 0
|
||||
for future in concurrent.futures.as_completed(futures, timeout=300):
|
||||
try:
|
||||
process_namespace_scheduled_actions(namespace.id)
|
||||
future.result()
|
||||
completed += 1
|
||||
except Exception as e:
|
||||
log.error(f"Error processing namespace {namespace.id}: {e}")
|
||||
|
||||
log.info("Finished processing scheduled actions")
|
||||
|
||||
except Exception as e:
|
||||
log.error(f"Error in scheduler: {e}")
|
||||
finally:
|
||||
release_scheduler_lock()
|
||||
main_dbsession.close()
|
||||
main_engine.dispose()
|
||||
log.error(f"Namespace scheduler job failed: {e}")
|
||||
|
||||
log.info(f"Finished dispatching scheduler jobs ({completed}/{len(namespace_dbs)} completed)")
|
||||
|
||||
|
||||
def process_namespace_scheduled_actions(namespace_id):
|
||||
|
|
@ -827,13 +762,13 @@ scheduler_running = False
|
|||
|
||||
|
||||
def scheduler_worker():
|
||||
"""Background scheduler worker that runs every minute."""
|
||||
"""Background scheduler worker that dispatches per-namespace jobs."""
|
||||
global scheduler_running
|
||||
while scheduler_running:
|
||||
try:
|
||||
process_scheduled_actions()
|
||||
dispatch_namespace_scheduler_jobs()
|
||||
except Exception as e:
|
||||
log.error(f"Scheduler worker error: {e}")
|
||||
log.error(f"Scheduler dispatcher error: {e}")
|
||||
|
||||
# Sleep for 60 seconds
|
||||
for i in range(60):
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue