mirror of
https://github.com/galaxyproject/galaxy.git
synced 2026-09-24 16:30:27 +08:00
Merge pull request #16144 from jmchilton/caching_refactor
Refactor caching object stores ahead of adding task-based store.
This commit is contained in:
@@ -974,6 +974,23 @@ mapping:
|
||||
Configuration file for the object store
|
||||
If this is set and exists, it overrides any other objectstore settings.
|
||||
|
||||
object_store_cache_path:
|
||||
type: str
|
||||
default: object_store_cache
|
||||
path_resolves_to: cache_dir
|
||||
required: false
|
||||
desc: |
|
||||
Default cache path for caching object stores if cache not configured for
|
||||
that object store entry.
|
||||
|
||||
object_store_cache_size:
|
||||
type: int
|
||||
default: -1
|
||||
required: false
|
||||
desc: |
|
||||
Default cache size for caching object stores if cache not configured for
|
||||
that object store entry.
|
||||
|
||||
object_store_store_by:
|
||||
type: str
|
||||
required: false
|
||||
|
||||
@@ -30,6 +30,7 @@ from . import ConcreteObjectStore
|
||||
from .caching import (
|
||||
CacheTarget,
|
||||
InProcessCacheMonitor,
|
||||
parse_caching_config_dict_from_xml,
|
||||
)
|
||||
|
||||
NO_BLOBSERVICE_ERROR_MESSAGE = (
|
||||
@@ -51,9 +52,7 @@ def parse_config_xml(config_xml):
|
||||
container_name = container_xml.get("name")
|
||||
max_chunk_size = int(container_xml.get("max_chunk_size", 250)) # currently unused
|
||||
|
||||
c_xml = config_xml.findall("cache")[0]
|
||||
cache_size = float(c_xml.get("size", -1))
|
||||
staging_path = c_xml.get("path", None)
|
||||
cache_dict = parse_caching_config_dict_from_xml(config_xml)
|
||||
|
||||
tag, attrs = "extra_dir", ("type", "path")
|
||||
extra_dirs = config_xml.findall(tag)
|
||||
@@ -71,10 +70,7 @@ def parse_config_xml(config_xml):
|
||||
"name": container_name,
|
||||
"max_chunk_size": max_chunk_size,
|
||||
},
|
||||
"cache": {
|
||||
"size": cache_size,
|
||||
"path": staging_path,
|
||||
},
|
||||
"cache": cache_dict,
|
||||
"extra_dirs": extra_dirs,
|
||||
"private": ConcreteObjectStore.parse_private_from_config_xml(config_xml),
|
||||
}
|
||||
@@ -101,7 +97,7 @@ class AzureBlobObjectStore(ConcreteObjectStore):
|
||||
|
||||
auth_dict = config_dict["auth"]
|
||||
container_dict = config_dict["container"]
|
||||
cache_dict = config_dict["cache"]
|
||||
cache_dict = config_dict.get("cache") or {}
|
||||
self.enable_cache_monitor = config_dict.get("enable_cache_monitor", True)
|
||||
|
||||
self.account_name = auth_dict.get("account_name")
|
||||
@@ -110,7 +106,7 @@ class AzureBlobObjectStore(ConcreteObjectStore):
|
||||
self.container_name = container_dict.get("name")
|
||||
self.max_chunk_size = container_dict.get("max_chunk_size", 250) # currently unused
|
||||
|
||||
self.cache_size = cache_dict.get("size", -1)
|
||||
self.cache_size = cache_dict.get("size") or self.config.object_store_cache_size
|
||||
self.staging_path = cache_dict.get("path") or self.config.object_store_cache_path
|
||||
|
||||
self._initialize()
|
||||
|
||||
@@ -98,6 +98,23 @@ def _get_cache_size_files(cache_path) -> Tuple[int, FileListT]:
|
||||
return cache_size, file_list
|
||||
|
||||
|
||||
def parse_caching_config_dict_from_xml(config_xml):
|
||||
cache_els = config_xml.findall("cache")
|
||||
if len(cache_els) > 0:
|
||||
c_xml = config_xml.findall("cache")[0]
|
||||
cache_size = float(c_xml.get("size", -1))
|
||||
|
||||
staging_path = c_xml.get("path", None)
|
||||
|
||||
cache_dict = {
|
||||
"size": cache_size,
|
||||
"path": staging_path,
|
||||
}
|
||||
else:
|
||||
cache_dict = {}
|
||||
return cache_dict
|
||||
|
||||
|
||||
class InProcessCacheMonitor:
|
||||
def __init__(self, cache_target: CacheTarget, interval: int = 30, initial_sleep: Optional[int] = 2):
|
||||
# This Event object is initialized to False
|
||||
|
||||
@@ -86,7 +86,7 @@ class Cloud(ConcreteObjectStore, CloudConfigMixin):
|
||||
|
||||
bucket_dict = config_dict["bucket"]
|
||||
connection_dict = config_dict.get("connection", {})
|
||||
cache_dict = config_dict["cache"]
|
||||
cache_dict = config_dict.get("cache") or {}
|
||||
self.enable_cache_monitor = config_dict.get("enable_cache_monitor", True)
|
||||
|
||||
self.provider = config_dict["provider"]
|
||||
@@ -101,7 +101,7 @@ class Cloud(ConcreteObjectStore, CloudConfigMixin):
|
||||
self.is_secure = connection_dict.get("is_secure", True)
|
||||
self.conn_path = connection_dict.get("conn_path", "/")
|
||||
|
||||
self.cache_size = cache_dict.get("size", -1)
|
||||
self.cache_size = cache_dict.get("size") or self.config.object_store_cache_size
|
||||
self.staging_path = cache_dict.get("path") or self.config.object_store_cache_path
|
||||
|
||||
self._initialize()
|
||||
|
||||
@@ -202,10 +202,8 @@ class IRODSObjectStore(DiskObjectStore, CloudConfigMixin):
|
||||
if self.connection_pool_monitor_interval is None:
|
||||
_config_dict_error("connection->connection_pool_monitor_interval")
|
||||
|
||||
cache_dict = config_dict["cache"]
|
||||
if cache_dict is None:
|
||||
_config_dict_error("cache")
|
||||
self.cache_size = cache_dict.get("size", -1)
|
||||
cache_dict = config_dict.get("cache") or {}
|
||||
self.cache_size = cache_dict.get("size") or self.config.object_store_cache_path
|
||||
if self.cache_size is None:
|
||||
_config_dict_error("cache->size")
|
||||
self.staging_path = cache_dict.get("path") or self.config.object_store_cache_path
|
||||
|
||||
@@ -35,6 +35,7 @@ from . import ConcreteObjectStore
|
||||
from .caching import (
|
||||
CacheTarget,
|
||||
InProcessCacheMonitor,
|
||||
parse_caching_config_dict_from_xml,
|
||||
)
|
||||
from .s3_multipart_upload import multipart_upload
|
||||
|
||||
@@ -70,10 +71,7 @@ def parse_config_xml(config_xml):
|
||||
is_secure = string_as_bool(cn_xml.get("is_secure", "True"))
|
||||
conn_path = cn_xml.get("conn_path", "/")
|
||||
|
||||
c_xml = config_xml.findall("cache")[0]
|
||||
cache_size = float(c_xml.get("size", -1))
|
||||
|
||||
staging_path = c_xml.get("path", None)
|
||||
cache_dict = parse_caching_config_dict_from_xml(config_xml)
|
||||
|
||||
tag, attrs = "extra_dir", ("type", "path")
|
||||
extra_dirs = config_xml.findall(tag)
|
||||
@@ -100,10 +98,7 @@ def parse_config_xml(config_xml):
|
||||
"is_secure": is_secure,
|
||||
"conn_path": conn_path,
|
||||
},
|
||||
"cache": {
|
||||
"size": cache_size,
|
||||
"path": staging_path,
|
||||
},
|
||||
"cache": cache_dict,
|
||||
"extra_dirs": extra_dirs,
|
||||
"private": ConcreteObjectStore.parse_private_from_config_xml(config_xml),
|
||||
}
|
||||
@@ -158,7 +153,7 @@ class S3ObjectStore(ConcreteObjectStore, CloudConfigMixin):
|
||||
auth_dict = config_dict["auth"]
|
||||
bucket_dict = config_dict["bucket"]
|
||||
connection_dict = config_dict.get("connection", {})
|
||||
cache_dict = config_dict["cache"]
|
||||
cache_dict = config_dict.get("cache") or {}
|
||||
self.enable_cache_monitor = config_dict.get("enable_cache_monitor", True)
|
||||
|
||||
self.access_key = auth_dict.get("access_key")
|
||||
@@ -174,7 +169,7 @@ class S3ObjectStore(ConcreteObjectStore, CloudConfigMixin):
|
||||
self.is_secure = connection_dict.get("is_secure", True)
|
||||
self.conn_path = connection_dict.get("conn_path", "/")
|
||||
|
||||
self.cache_size = cache_dict.get("size", -1)
|
||||
self.cache_size = cache_dict.get("size") or self.config.object_store_cache_size
|
||||
self.staging_path = cache_dict.get("path") or self.config.object_store_cache_path
|
||||
|
||||
extra_dirs = {e["type"]: e["path"] for e in config_dict.get("extra_dirs", [])}
|
||||
|
||||
@@ -39,6 +39,7 @@ class Config:
|
||||
config_file = "store.yaml"
|
||||
self.write(config_str, config_file)
|
||||
config = MockConfig(self.temp_directory, config_file, store_by=store_by)
|
||||
self.global_config = config
|
||||
if clazz is None:
|
||||
self.object_store = objectstore.build_object_store_from_config(config)
|
||||
elif config_file == "store.xml":
|
||||
@@ -68,6 +69,7 @@ class MockConfig:
|
||||
self.file_path = temp_directory
|
||||
self.object_store_config_file = os.path.join(temp_directory, config_file)
|
||||
self.object_store_check_old_style = False
|
||||
self.object_store_cache_size = -1
|
||||
self.object_store_cache_path = os.path.join(temp_directory, "staging")
|
||||
self.object_store_store_by = store_by
|
||||
self.jobs_directory = temp_directory
|
||||
|
||||
Reference in New Issue
Block a user