mirror of
https://github.com/Novartis/cellxgene-gateway.git
synced 2026-10-09 23:40:56 +08:00
added support for gateway ttl and manual pruning
This commit is contained in:
@@ -44,17 +44,17 @@ class CacheEntry:
|
|||||||
@classmethod
|
@classmethod
|
||||||
def for_dataset(cls, dataset, file_path, port):
|
def for_dataset(cls, dataset, file_path, port):
|
||||||
return cls(
|
return cls(
|
||||||
"",
|
None,
|
||||||
dataset,
|
dataset,
|
||||||
file_path,
|
file_path,
|
||||||
port,
|
port,
|
||||||
current_time_stamp(),
|
current_time_stamp(),
|
||||||
current_time_stamp(),
|
current_time_stamp(),
|
||||||
"loading",
|
"loading",
|
||||||
"",
|
None,
|
||||||
"",
|
None,
|
||||||
"",
|
None,
|
||||||
"",
|
None,
|
||||||
)
|
)
|
||||||
|
|
||||||
def set_loaded(self, pid):
|
def set_loaded(self, pid):
|
||||||
@@ -67,6 +67,12 @@ class CacheEntry:
|
|||||||
self.http_status = http_status
|
self.http_status = http_status
|
||||||
self.status = "error"
|
self.status = "error"
|
||||||
|
|
||||||
|
def append_output(self, output):
|
||||||
|
if self.all_output == None:
|
||||||
|
self.all_output = output
|
||||||
|
else:
|
||||||
|
self.all_output += output
|
||||||
|
|
||||||
def serve_content(self, path):
|
def serve_content(self, path):
|
||||||
dataset = self.dataset
|
dataset = self.dataset
|
||||||
|
|
||||||
|
|||||||
@@ -15,6 +15,7 @@ gateway_host = os.environ.get("GATEWAY_HOST")
|
|||||||
gateway_protocol = os.environ.get("GATEWAY_PROTOCOL")
|
gateway_protocol = os.environ.get("GATEWAY_PROTOCOL")
|
||||||
ip = os.environ.get("GATEWAY_IP")
|
ip = os.environ.get("GATEWAY_IP")
|
||||||
extra_scripts = os.environ.get("GATEWAY_EXTRA_SCRIPTS")
|
extra_scripts = os.environ.get("GATEWAY_EXTRA_SCRIPTS")
|
||||||
|
ttl = os.environ.get("GATEWAY_TTL")
|
||||||
|
|
||||||
env_vars = {
|
env_vars = {
|
||||||
"CELLXGENE_LOCATION": cellxgene_location,
|
"CELLXGENE_LOCATION": cellxgene_location,
|
||||||
|
|||||||
@@ -187,9 +187,21 @@ def do_GET_status():
|
|||||||
"cache_status.html", entry_list=cache.entry_list
|
"cache_status.html", entry_list=cache.entry_list
|
||||||
)
|
)
|
||||||
|
|
||||||
|
pruner = PruneProcessCache(cache)
|
||||||
|
|
||||||
|
@app.route("/prune/<path:path>", methods=["DELETE"])
|
||||||
|
def do_prune(path):
|
||||||
|
dataset = get_dataset(path)
|
||||||
|
match = cache.check_entry(dataset)
|
||||||
|
if match is None:
|
||||||
|
return "not found"
|
||||||
|
else:
|
||||||
|
pruner.prune(match)
|
||||||
|
return "success"
|
||||||
|
|
||||||
def main():
|
def main():
|
||||||
env.validate()
|
env.validate()
|
||||||
background_thread = Thread(target=PruneProcessCache(cache))
|
background_thread = Thread(target=pruner)
|
||||||
background_thread.start()
|
background_thread.start()
|
||||||
|
|
||||||
app.run(host="0.0.0.0", port=5005, debug=False)
|
app.run(host="0.0.0.0", port=5005, debug=False)
|
||||||
|
|||||||
@@ -12,12 +12,22 @@ import time
|
|||||||
import psutil
|
import psutil
|
||||||
|
|
||||||
from cellxgene_gateway.util import current_time_stamp
|
from cellxgene_gateway.util import current_time_stamp
|
||||||
|
from cellxgene_gateway.env import ttl
|
||||||
|
|
||||||
|
|
||||||
class PruneProcessCache:
|
class PruneProcessCache:
|
||||||
def __init__(self, cache):
|
def __init__(self, cache):
|
||||||
self.cache = cache
|
self.cache = cache
|
||||||
|
|
||||||
|
def prune(self, process):
|
||||||
|
self.cache.entry_list.remove(process)
|
||||||
|
pid = process.pid
|
||||||
|
if pid != None:
|
||||||
|
p = psutil.Process(pid)
|
||||||
|
p.terminate()
|
||||||
|
p = psutil.Process(pid + 2)
|
||||||
|
p.terminate()
|
||||||
|
|
||||||
def __call__(self):
|
def __call__(self):
|
||||||
while True:
|
while True:
|
||||||
time.sleep(60)
|
time.sleep(60)
|
||||||
@@ -26,14 +36,13 @@ class PruneProcessCache:
|
|||||||
|
|
||||||
processes_to_delete = []
|
processes_to_delete = []
|
||||||
for p in self.cache.entry_list:
|
for p in self.cache.entry_list:
|
||||||
if timestamp - p.timestamp > 3600:
|
if timestamp - p.timestamp > (3600 if ttl is None else ttl):
|
||||||
processes_to_delete.append(p)
|
processes_to_delete.append(p)
|
||||||
processes_to_delete
|
processes_to_delete
|
||||||
|
|
||||||
for process in processes_to_delete:
|
for process in processes_to_delete:
|
||||||
self.cache.entry_list.remove(process)
|
try:
|
||||||
pid = process.pid
|
self.prune(process)
|
||||||
p = psutil.Process(pid)
|
except Exception as detail:
|
||||||
p.terminate()
|
print('failed to prune process:', detail)
|
||||||
p = psutil.Process(pid + 2)
|
|
||||||
p.terminate()
|
|
||||||
|
|||||||
@@ -64,7 +64,7 @@ class SubprocessBackend:
|
|||||||
|
|
||||||
raise ProcessException.from_pid_object(cache_entry)
|
raise ProcessException.from_pid_object(cache_entry)
|
||||||
else:
|
else:
|
||||||
cache_entry.all_output += output
|
cache_entry.append_output(output)
|
||||||
|
|
||||||
cache_entry.set_loaded(process.pid)
|
cache_entry.set_loaded(process.pid)
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user