mirror of
https://github.com/galaxyproject/galaxy.git
synced 2026-09-24 16:30:27 +08:00
Add periodic history audit prune task
that runs when not using celery.
This commit is contained in:
@@ -61,6 +61,7 @@ from galaxy.util import (
|
||||
heartbeat,
|
||||
StructuredExecutionTimer,
|
||||
)
|
||||
from galaxy.util.task import IntervalTask
|
||||
from galaxy.visualization.data_providers.registry import DataProviderRegistry
|
||||
from galaxy.visualization.genomes import Genomes
|
||||
from galaxy.visualization.plugins.registry import VisualizationsRegistry
|
||||
@@ -304,6 +305,16 @@ class UniverseApplication(StructuredApp, GalaxyManagerApplication):
|
||||
self.authnz_manager = managers.AuthnzManager(self,
|
||||
self.config.oidc_config_file,
|
||||
self.config.oidc_backends_config_file)
|
||||
|
||||
if not self.config.enable_celery_tasks and self.config.history_audit_table_prune_interval > 0:
|
||||
self.prune_history_audit_task = IntervalTask(
|
||||
func=lambda: galaxy.model.HistoryAudit.prune(self.model.session),
|
||||
name="HistoryAuditTablePruneTask",
|
||||
interval=self.config.history_audit_table_prune_interval,
|
||||
immediate_start=False,
|
||||
time_execution=True)
|
||||
self.application_stack.register_postfork_function(self.prune_history_audit_task.start)
|
||||
self.haltables.append(("HistoryAuditTablePruneTask", self.prune_history_audit_task.shutdown))
|
||||
# Start the job manager
|
||||
self.application_stack.register_postfork_function(self.job_manager.start)
|
||||
self.proxy_manager = ProxyManager(self.config)
|
||||
|
||||
@@ -0,0 +1,51 @@
|
||||
import logging
|
||||
from threading import (
|
||||
Event,
|
||||
Thread,
|
||||
)
|
||||
|
||||
from galaxy.util import ExecutionTimer
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class IntervalTask:
|
||||
|
||||
def __init__(self, func, name="Periodic task", interval=3600, immediate_start=False, time_execution=False):
|
||||
"""
|
||||
Run an arbitrary function `func` every `interval` seconds.
|
||||
|
||||
Set `immediate_start` to True to run `func` when task is started.
|
||||
"""
|
||||
self.func = func
|
||||
self.name = name
|
||||
self.interval = interval
|
||||
self.time_execution = time_execution
|
||||
self.immediate_start = immediate_start
|
||||
self.event = Event()
|
||||
self.thread = Thread(target=self.run, name=self.name, daemon=True)
|
||||
self.running = False
|
||||
|
||||
def start(self):
|
||||
self.running = True
|
||||
self.thread.start()
|
||||
|
||||
def _exec(self):
|
||||
if self.time_execution:
|
||||
timer = ExecutionTimer()
|
||||
self.func()
|
||||
if self.time_execution:
|
||||
log.debug(f"Executed periodic task {self.name} {timer}")
|
||||
|
||||
def run(self):
|
||||
if self.immediate_start:
|
||||
self._exec()
|
||||
while not self.event.isSet():
|
||||
self.event.wait(self.interval)
|
||||
if self.running:
|
||||
self._exec()
|
||||
|
||||
def shutdown(self):
|
||||
self.running = False
|
||||
self.event.set()
|
||||
self.thread.join(5)
|
||||
@@ -0,0 +1,28 @@
|
||||
import time
|
||||
|
||||
from galaxy.util.task import IntervalTask
|
||||
|
||||
|
||||
def test_interval_task_immediate_start():
|
||||
results = []
|
||||
task = IntervalTask(lambda: results.append(1), name="test_task", interval=0.2, immediate_start=True)
|
||||
task.start()
|
||||
task.shutdown()
|
||||
assert len(results) == 1
|
||||
|
||||
|
||||
def test_interval_task_delayed_start():
|
||||
results = []
|
||||
task = IntervalTask(lambda: results.append(1), name="test_task", interval=0.2, immediate_start=False)
|
||||
task.start()
|
||||
task.shutdown()
|
||||
assert len(results) == 0
|
||||
|
||||
|
||||
def test_interval_task_delayed_start_run_once():
|
||||
results = []
|
||||
task = IntervalTask(lambda: results.append(1), name="test_task", interval=0.2, immediate_start=False)
|
||||
task.start()
|
||||
time.sleep(0.25)
|
||||
task.shutdown()
|
||||
assert len(results) == 1
|
||||
Reference in New Issue
Block a user