Commit a081616d authored by André Anjos's avatar André Anjos 💬

Merge branch '94_make_cache_access_more_specific' into 'master'

Make cache access more specific

See merge request !110
parents eedd5b6b 071e19ce
Pipeline #36775 failed with stages
in 1 minute and 16 seconds
[flake8]
max-line-length = 80
select = B,C,E,F,W,T4,B9,B950
ignore = E501, W503
ignore = E501, W503, E203
......@@ -400,7 +400,7 @@ class Host(object):
continue
image_info = _parse_image_info(image)
key = "{} {}".format(image_info["name"], image_info["version"])
key = "{} ({})".format(image_info["name"], image_info["version"])
image_name = image_info["image"]
if key in environments:
......
......@@ -49,6 +49,7 @@ import requests
import simplejson as json
from beat.backend.python.execution import MessageHandler
from beat.backend.python.data import getAllFilenames
from .. import stats
from .. import utils
......@@ -146,6 +147,9 @@ class DockerExecutor(RemoteExecutor):
"""
CONTAINER_PREFIX_PATH = "/beat/prefix"
CONTAINER_CACHE_PATH = "/beat/cache"
def __init__(
self,
host,
......@@ -218,8 +222,8 @@ class DockerExecutor(RemoteExecutor):
cmd = [
"databases_provider",
"0.0.0.0:%d" % database_port,
"/beat/prefix",
"/beat/cache",
self.CONTAINER_PREFIX_PATH,
self.CONTAINER_CACHE_PATH,
]
if configuration_name:
......@@ -235,8 +239,10 @@ class DockerExecutor(RemoteExecutor):
databases_info.set_name(databases_info_name)
# Specify the volumes to mount inside the container
databases_info.add_volume(databases_configuration_path, "/beat/prefix")
databases_info.add_volume(self.cache, "/beat/cache")
databases_info.add_volume(
databases_configuration_path, self.CONTAINER_PREFIX_PATH
)
databases_info.add_volume(self.cache, self.CONTAINER_CACHE_PATH)
for db_name, db_path in database_paths.items():
databases_info.add_volume(db_path, os.path.join("/databases", db_name))
......@@ -273,6 +279,54 @@ class DockerExecutor(RemoteExecutor):
)
return retval
def __setup_io_volumes(self, algorithm_container, configuration):
"""Setup all the volumes for input and output files.
Parameters:
algorithm_container: container that will execute an algorithm
configuration: json object containing the algorithm parameters
"""
for item in configuration["inputs"].values():
file_path = item["path"]
source_path = os.path.join(self.cache, file_path)
if os.path.isfile(source_path):
algorithm_container.add_volume(
source_path, os.path.join(self.CONTAINER_CACHE_PATH, file_path)
)
else:
all_files = getAllFilenames(source_path)
for file_list in all_files:
for file_ in file_list:
target_path = file_[len(self.cache) + 1 :]
cache_path = os.path.join(
self.CONTAINER_CACHE_PATH, target_path
)
algorithm_container.add_volume(file_, cache_path)
def __add_writable_volume(file_path):
output_folder = file_path[: file_path.rfind("/")]
source_folder = os.path.join(self.cache, output_folder)
if not os.path.exists(source_folder):
os.makedirs(source_folder)
algorithm_container.add_volume(
source_folder,
os.path.join(self.CONTAINER_CACHE_PATH, output_folder),
read_only=False,
)
for item in configuration.get("outputs", {}).values():
file_path = item["path"]
__add_writable_volume(file_path)
result = configuration.get("result")
if result:
file_path = result["path"]
__add_writable_volume(file_path)
def process(
self, virtual_memory_in_megabytes=0, max_cpu_percent=0, timeout_in_minutes=0
):
......@@ -384,8 +438,8 @@ class DockerExecutor(RemoteExecutor):
cmd = [
"loop_execute",
"0.0.0.0:{}".format(loop_algorithm_container_port),
"/beat/prefix",
"/beat/cache",
self.CONTAINER_PREFIX_PATH,
self.CONTAINER_CACHE_PATH,
]
if len(self.databases) > 0:
......@@ -406,10 +460,10 @@ class DockerExecutor(RemoteExecutor):
loop_algorithm_container.network_name = network_name
# Volumes
loop_algorithm_container.add_volume(configuration_path, "/beat/prefix")
loop_algorithm_container.add_volume(
self.cache, "/beat/cache", read_only=False
configuration_path, self.CONTAINER_PREFIX_PATH
)
self.__setup_io_volumes(loop_algorithm_container, self.data["loop"])
# Start the container
self.host.start(
......@@ -424,9 +478,9 @@ class DockerExecutor(RemoteExecutor):
# Command to execute
cmd = [
"execute",
"--cache=/beat/cache",
"--cache={}".format(self.CONTAINER_CACHE_PATH),
self.message_handler.address,
"/beat/prefix",
self.CONTAINER_PREFIX_PATH,
]
if len(self.databases) > 0:
......@@ -450,8 +504,8 @@ class DockerExecutor(RemoteExecutor):
algorithm_container.network_name = network_name
# Volumes
algorithm_container.add_volume(configuration_path, "/beat/prefix")
algorithm_container.add_volume(self.cache, "/beat/cache", read_only=False)
algorithm_container.add_volume(configuration_path, self.CONTAINER_PREFIX_PATH)
self.__setup_io_volumes(algorithm_container, self.data)
# Start the container
self.host.start(
......
{
"language": "python",
"splittable": false,
"groups": [
{
"name": "main",
"inputs": {
"in_data": {
"type": "user/single_integer/1"
}
},
"outputs": {
"out_data": {
"type": "user/single_integer/1"
}
}
}
]
}
#!/usr/bin/env python
# vim: set fileencoding=utf-8 :
###################################################################################
# #
# Copyright (c) 2020 Idiap Research Institute, http://www.idiap.ch/ #
# Contact: beat.support@idiap.ch #
# #
# Redistribution and use in source and binary forms, with or without #
# modification, are permitted provided that the following conditions are met: #
# #
# 1. Redistributions of source code must retain the above copyright notice, this #
# list of conditions and the following disclaimer. #
# #
# 2. Redistributions in binary form must reproduce the above copyright notice, #
# this list of conditions and the following disclaimer in the documentation #
# and/or other materials provided with the distribution. #
# #
# 3. Neither the name of the copyright holder nor the names of its contributors #
# may be used to endorse or promote products derived from this software without #
# specific prior written permission. #
# #
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND #
# ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED #
# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE #
# DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE #
# FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL #
# DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR #
# SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER #
# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, #
# OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE #
# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. #
# #
###################################################################################
class Algorithm:
def process(self, inputs, outputs):
with open("/beat/cache/truc.txt", "wt") as file:
file.write("Should not write there...")
# Should not arrive here
outputs["out_data"].write(inputs["in_data"].data)
return True
{
"analyzers": {
"analysis": {
"algorithm": "v1/integers_echo_analyzer/1",
"inputs": {
"in_data": "in"
}
}
},
"blocks": {
"echo": {
"algorithm": "errors/create_file/1",
"inputs": {
"in_data": "in"
},
"outputs": {
"out_data": "out"
}
}
},
"datasets": {
"set": {
"database": "simple/1",
"protocol": "protocol",
"set": "set"
}
},
"globals": {
"queue": "queue",
"environment": {
"name": "Python 2.7",
"version": "1.3.0"
}
}
}
......@@ -233,3 +233,10 @@ class TestDockerExecution(BaseExecutionMixIn):
nose.tools.assert_true(
"[sys] C++ algorithm can't be analyzers" in result["stderr"]
)
@slow
def test_read_only_error(self):
result = self.execute("errors/user/single/1/write_error", [{"out_data": 42}])
nose.tools.eq_(result["status"], 1)
nose.tools.assert_true("Permission denied" in result["user_error"])
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment