From bdfe2f7c8dbc7fd6d9144d25bb8c80435d543675 Mon Sep 17 00:00:00 2001
From: Stefano Alberto Russo <stefano.russo@gmail.com>
Date: Mon, 8 Nov 2021 15:46:39 +0100
Subject: [PATCH] Added the Storage model.
---
.../prestartup_slurmclustermaster.sh | 6 +-
.../webapp/code/rosetta/core_app/admin.py | 3 +-
services/webapp/code/rosetta/core_app/api.py | 315 ++++++++++--------
.../rosetta/core_app/computing_managers.py | 116 ++++---
.../management/commands/core_app_populate.py | 54 ++-
.../core_app/migrations/0011_storage.py | 37 ++
.../webapp/code/rosetta/core_app/models.py | 52 +++
7 files changed, 390 insertions(+), 193 deletions(-)
create mode 100644 services/webapp/code/rosetta/core_app/migrations/0011_storage.py
diff --git a/services/slurmclustermaster/prestartup_slurmclustermaster.sh b/services/slurmclustermaster/prestartup_slurmclustermaster.sh
index 9769460..c27ae01 100644
--- a/services/slurmclustermaster/prestartup_slurmclustermaster.sh
+++ b/services/slurmclustermaster/prestartup_slurmclustermaster.sh
@@ -7,8 +7,12 @@ mkdir -p /shared/rosetta && chown rosetta:rosetta /shared/rosetta
# Shared home for slurmtestuser to simulate a shared home folders filesystem
cp -a /home_slurmtestuser_vanilla /shared/home_slurmtestuser
-# Create shared "data" and "scratch" directories
+# Create shared data directories
mkdir -p /shared/scratch
chmod 777 /shared/scratch
+
+mkdir -p /shared/data/shared
+chmod 777 /shared/data/shared
+
mkdir -p /shared/data/users/slurmtestuser
chown slurmtestuser:slurmtestuser /shared/data/users/slurmtestuser
diff --git a/services/webapp/code/rosetta/core_app/admin.py b/services/webapp/code/rosetta/core_app/admin.py
index db79005..0edd07f 100644
--- a/services/webapp/code/rosetta/core_app/admin.py
+++ b/services/webapp/code/rosetta/core_app/admin.py
@@ -1,6 +1,6 @@
from django.contrib import admin
-from .models import Profile, LoginToken, Task, Container, Computing, ComputingSysConf, ComputingUserConf, KeyPair, Text
+from .models import Profile, LoginToken, Task, Container, Computing, ComputingSysConf, ComputingUserConf, Storage, KeyPair, Text
admin.site.register(Profile)
admin.site.register(LoginToken)
@@ -9,5 +9,6 @@ admin.site.register(Container)
admin.site.register(Computing)
admin.site.register(ComputingSysConf)
admin.site.register(ComputingUserConf)
+admin.site.register(Storage)
admin.site.register(KeyPair)
admin.site.register(Text)
diff --git a/services/webapp/code/rosetta/core_app/api.py b/services/webapp/code/rosetta/core_app/api.py
index 025a701..4633c79 100644
--- a/services/webapp/code/rosetta/core_app/api.py
+++ b/services/webapp/code/rosetta/core_app/api.py
@@ -12,7 +12,8 @@ from rest_framework.response import Response
from rest_framework import status, serializers, viewsets
from rest_framework.views import APIView
from .utils import format_exception, send_email, os_shell, now_t
-from .models import Profile, Task, TaskStatuses, Computing, KeyPair
+from .models import Profile, Task, TaskStatuses, Computing, Storage, KeyPair
+from .exceptions import ConsistencyException
import json
# Setup logging
@@ -372,6 +373,7 @@ class FileManagerAPI(PrivateGETAPI, PrivatePOSTAPI):
user_keys = KeyPair.objects.get(user=user, default=True)
# Get computing host
+ computing.attach_user_conf(user)
computing_host = computing.conf.get('host')
computing_user = computing.conf.get('user')
@@ -399,6 +401,7 @@ class FileManagerAPI(PrivateGETAPI, PrivatePOSTAPI):
user_keys = KeyPair.objects.get(user=user, default=True)
# Get computing host
+ computing.attach_user_conf(user)
computing_host = computing.conf.get('host')
computing_user = computing.conf.get('user')
@@ -426,43 +429,74 @@ class FileManagerAPI(PrivateGETAPI, PrivatePOSTAPI):
path = re.sub(cleaner,r"\\",path)
return path
- def get_computing(self, path, request):
- # Get the computing based on the folder name # TODO: this is very weak..
- computing_resource_name = path.split('/')[1]
-
- # First try to get platform-level computing resource
- computing = Computing.objects.filter(name=computing_resource_name, user=None)
-
- # If not, fallback on the user computing name
- if not computing:
- computing = Computing.objects.filter(name=computing_resource_name, user=request.user)
-
- if not computing:
- raise Exception('Cannot find any computing resource named "{}"'.format(computing_resource_name+'1'))
-
- # Check that we had no more than one computing resource
- if len(computing) > 1:
- raise Exception('Found more than one computign resource named "{}", cannot continue!'.format(computing_resource_name))
+ @staticmethod
+ def sanitize_and_prepare_shell_path(path, storage, user):
+ path = path.replace(' ', '\ ')
+ cleaner = re.compile('(?:\\\)+')
+ path = re.sub(cleaner,r"\\",path)
+
+ # Prepare the base path (expand it with variables substitution)
+ base_path_expanded = storage.base_path
+ if '$SSH_USER' in base_path_expanded:
+ if storage.access_through_computing:
+ computing = storage.computing
+ computing.attach_user_conf(user)
+ base_path_expanded = base_path_expanded.replace('$SSH_USER', computing.conf.get('user'))
+ else:
+ raise NotImplementedError('Accessing a storage with ssh+cli without going through its computing resource is not implemented')
+ if '$USER' in base_path_expanded:
+ base_path_expanded = base_path_expanded.replace('$USER', user.name)
- computing = computing[0]
+ # If the path is not starting with the base path, do it
+ if not path.startswith(base_path_expanded):
+ path = base_path_expanded+'/'+path
+
+ return path
- # Attach user conf in any
- computing.attach_user_conf(request.user)
+ def get_storage_from_path(self, path, request):
+ # Get the storage based on the "root" folder name
+ # TODO: this is extremely weak..
+ storage_id = path.split('/')[1]
+ storage_name = storage_id.split('@')[0]
+ try:
+ computing_name = storage_id.split('@')[1]
+ except IndexError:
+ computing_name = None
+
+ # Get all the storages for this name:
+ storages = Storage.objects.filter(name=storage_name, user=None)
- return computing
+ # Filter by computing resource name
+ if computing_name:
+ unfiltered_storages = storages
+ storages = []
+ for storage in unfiltered_storages:
+ if storage.computing.name == computing_name:
+ storages.append(storage)
+
+ # Check that we had at least and no more than one storage in the end
+ if len(storages) == 0:
+ raise Exception('Found no storage for id "{}", cannot continue!'.format(storage_id))
+ if len(storages) > 1:
+ raise Exception('Found more than one storage for id "{}", cannot continue!'.format(storage_id))
+
+ # Assign the storage
+ storage = storages[0]
+
+ return storage
- def ls(self, path, user, computing, binds=[]):
+ def ls(self, path, user, storage):
# Data container
data = []
- path = self.sanitize_shell_path(path)
+ shell_path = self.sanitize_and_prepare_shell_path(path, storage, user)
# Prepare command
# https://askubuntu.com/questions/1116634/ls-command-show-time-only-in-iso-format
# https://www.howtogeek.com/451022/how-to-use-the-stat-command-on-linux/
- command = self.ssh_command('cd {} && stat --printf=\'%F/%s/%Y/%n\\n\' * .*'.format(path), user, computing)
+ command = self.ssh_command('cd {} && stat --printf=\'%F/%s/%Y/%n\\n\' * .*'.format(shell_path), user, storage.computing)
# Execute_command
out = os_shell(command, capture=True)
@@ -490,63 +524,48 @@ class FileManagerAPI(PrivateGETAPI, PrivatePOSTAPI):
timestamp = line_pieces[2]
name = line_pieces[3]
- # Check against binds if set
- if binds:
- if not path == '/':
- full_path = path + '/' + name
- else:
- full_path = '/' + name
-
- show = False
- for bind in binds:
- if bind.startswith(full_path) or full_path.startswith(bind):
- show = True
- break
-
- if not binds or (binds and show):
-
- # Define and clean listing path:
- listing_path = '/{}/{}/{}/'.format(computing.name, path, name)
- listing_path = self.clean_path(listing_path)
-
- # File or directory?
- if type == 'directory':
- if name not in ['.', '..']:
- data.append({
- 'id': listing_path,
- 'type': 'folder',
- 'attributes':{
- 'modified': timestamp,
- 'name': name,
- 'readable': 1,
- 'writable': 1,
- 'path': listing_path
- }
- })
- else:
+ # Define and clean listing path:
+ listing_path = '/{}/{}/{}/'.format(storage.id, path, name)
+ listing_path = self.clean_path(listing_path)
+
+ # File or directory?
+ if type == 'directory':
+ if name not in ['.', '..']:
data.append({
- 'id': listing_path[:-1], # Remove trailing slash
- 'type': 'file',
+ 'id': listing_path,
+ 'type': 'folder',
'attributes':{
'modified': timestamp,
'name': name,
'readable': 1,
'writable': 1,
- "size": size,
- 'path': listing_path[:-1] # Remove trailing slash
+ 'path': listing_path
}
- })
+ })
+ else:
+ data.append({
+ 'id': listing_path[:-1], # Remove trailing slash
+ 'type': 'file',
+ 'attributes':{
+ 'modified': timestamp,
+ 'name': name,
+ 'readable': 1,
+ 'writable': 1,
+ "size": size,
+ 'path': listing_path[:-1] # Remove trailing slash
+ }
+ })
return data
- def stat(self, path, user, computing):
+ def stat(self, path, user, storage):
- path = self.sanitize_shell_path(path)
+ path = self.sanitize_and_prepare_shell_path(path, storage, user)
# Prepare command. See the ls function above for some more info
- command = self.ssh_command('stat --printf=\'%F/%s/%Y/%n\\n\' {}'.format(path), user, computing)
+ command = self.ssh_command('stat --printf=\'%F/%s/%Y/%n\\n\' {}'.format(path), user, storage.computing)
# Execute_command
out = os_shell(command, capture=True)
@@ -585,12 +604,12 @@ class FileManagerAPI(PrivateGETAPI, PrivatePOSTAPI):
- def delete(self, path, user, computing):
+ def delete(self, path, user, storage):
- path = self.sanitize_shell_path(path)
+ path = self.sanitize_and_prepare_shell_path(path, storage, user)
# Prepare command
- command = self.ssh_command('rm -rf {}'.format(path), user, computing)
+ command = self.ssh_command('rm -rf {}'.format(path), user, storage.computing)
# Execute_command
out = os_shell(command, capture=True)
@@ -599,12 +618,12 @@ class FileManagerAPI(PrivateGETAPI, PrivatePOSTAPI):
return out.stdout
- def mkdir(self, path, user, computing):
+ def mkdir(self, path, user, storage):
- path = self.sanitize_shell_path(path)
+ path = self.sanitize_and_prepare_shell_path(path, storage, user)
# Prepare command
- command = self.ssh_command('mkdir {}'.format(path), user, computing)
+ command = self.ssh_command('mkdir {}'.format(path), user, storage.computing)
# Execute_command
out = os_shell(command, capture=True)
@@ -613,12 +632,12 @@ class FileManagerAPI(PrivateGETAPI, PrivatePOSTAPI):
return out.stdout
- def cat(self, path, user, computing):
+ def cat(self, path, user, storage):
- path = self.sanitize_shell_path(path)
+ path = self.sanitize_and_prepare_shell_path(path, storage, user)
# Prepare command
- command = self.ssh_command('cat {}'.format(path), user, computing)
+ command = self.ssh_command('cat {}'.format(path), user, storage.computing)
# Execute_command
out = os_shell(command, capture=True)
@@ -627,13 +646,13 @@ class FileManagerAPI(PrivateGETAPI, PrivatePOSTAPI):
return out.stdout
- def rename(self, old, new, user, computing):
+ def rename(self, old, new, user, storage):
- old = self.sanitize_shell_path(old)
- new = self.sanitize_shell_path(new)
+ old = self.sanitize_and_prepare_shell_path(old, storage, user)
+ new = self.sanitize_and_prepare_shell_path(new, storage, user)
# Prepare command
- command = self.ssh_command('mv {} {}'.format(old, new), user, computing)
+ command = self.ssh_command('mv {} {}'.format(old, new), user, storage.computing)
logger.critical(command)
@@ -644,13 +663,13 @@ class FileManagerAPI(PrivateGETAPI, PrivatePOSTAPI):
return out.stdout
- def copy(self, source, target, user, computing):
+ def copy(self, source, target, user, storage):
- source = self.sanitize_shell_path(source)
- target = self.sanitize_shell_path(target)
+ source = self.sanitize_and_prepare_shell_path(source, storage, user)
+ target = self.sanitize_and_prepare_shell_path(target, storage, user)
# Prepare command
- command = self.ssh_command('cp -a {} {}'.format(source, target), user, computing)
+ command = self.ssh_command('cp -a {} {}'.format(source, target), user, storage.computing)
# Execute_command
out = os_shell(command, capture=True)
@@ -659,13 +678,13 @@ class FileManagerAPI(PrivateGETAPI, PrivatePOSTAPI):
return out.stdout
- def scp(self, source, target, user, computing, mode='get'):
+ def scp_from(self, source, target, user, storage, mode='get'):
- source = self.sanitize_shell_path(source)
- target = self.sanitize_shell_path(target)
+ source = self.sanitize_and_prepare_shell_path(source, storage, user)
+ target = self.sanitize_shell_path(target) # This is a folder on Rosetta (/tmp)
# Prepare command
- command = self.scp_command(source, target, user, computing, mode)
+ command = self.scp_command(source, target, user, storage.computing, mode)
# Execute_command
out = os_shell(command, capture=True)
@@ -673,6 +692,23 @@ class FileManagerAPI(PrivateGETAPI, PrivatePOSTAPI):
raise Exception(out.stderr)
+ def scp_to(self, source, target, user, storage, mode='get'):
+
+ source = self.sanitize_shell_path(source) # This is a folder on Rosetta (/tmp)
+ target = self.sanitize_and_prepare_shell_path(target, storage, user)
+
+ # Prepare command
+ command = self.scp_command(source, target, user, storage.computing, mode)
+
+ # Execute_command
+ out = os_shell(command, capture=True)
+ if out.exit_code != 0:
+ raise Exception(out.stderr)
+
+
+ #============================
+ # API GET
+ #============================
def _get(self, request):
mode = request.GET.get('mode', None)
@@ -692,54 +728,37 @@ class FileManagerAPI(PrivateGETAPI, PrivatePOSTAPI):
# Base folder (computing resource-level)
if path == '/':
-
+
# Data container
data = {'data':[]}
- # Get computing resources
- computings = list(Computing.objects.filter(user=None)) + list(Computing.objects.filter(user=request.user))
-
- for computing in computings:
+ # Get storages
+ storages = list(Storage.objects.filter(user=None)) + list(Storage.objects.filter(user=request.user))
+
+ for storage in storages:
- # For now, we only support SSH-based computing resources
- if not 'ssh' in computing.access_mode:
+ # For now, we only support generic posix, SSH-based storages
+ if not storage.type=='generic_posix' and storage.access_mode=='ssh+cli':
continue
-
+
data['data'].append({
- 'id': '/{}/'.format(computing.name),
+ 'id': '/{}/'.format(storage.id),
'type': 'folder',
'attributes':{
- 'name': computing.name,
+ 'name': storage.id,
'readable': 1,
'writable': 1,
- 'path': '/{}/'.format(computing.name)
+ 'path': '/{}/'.format(storage.id)
}
})
-
+
else:
- computing = self.get_computing(path, request)
+ storage = self.get_storage_from_path(path, request)
- # If we just "entered" a computing resource, filter for its bindings
- # TODO: we can remove this and just always filter agains bind probably...
- if len(path.split('/')) == 3:
- if computing.user != request.user:
- binds = computing.sys_conf.get('binds')
- else:
- binds = computing.conf.get('binds')
-
- if binds:
- binds = binds.split(',')
- binds = [bind.split(':')[0] for bind in binds]
-
- # Ok, get directoris and files for this folder (always filtering by binds)
- ls_path = '/'+'/'.join(path.split('/')[2:])
- data = {'data': self.ls(ls_path, request.user, computing, binds)}
-
- else:
- # Ok, get directoris and files for this folder:
- ls_path = '/'+'/'.join(path.split('/')[2:])
- data = {'data': self.ls(ls_path, request.user, computing)}
+ # Get base directoris and files for this storage:
+ ls_path = '/'+'/'.join(path.split('/')[2:])
+ data = {'data': self.ls(ls_path, request.user, storage)}
elif mode in ['download', 'getimage']:
@@ -755,12 +774,12 @@ class FileManagerAPI(PrivateGETAPI, PrivatePOSTAPI):
# See here: https://github.com/psolom/RichFilemanager/wiki/API
# Set support vars
- computing = self.get_computing(path, request)
+ storage = self.get_storage_from_path(path, request)
file_path = '/'+'/'.join(path.split('/')[2:])
target_path = '/tmp/{}'.format(uuid.uuid4())
# Get the file
- self.scp(file_path, target_path, request.user, computing, mode='get')
+ self.scp_from(file_path, target_path, request.user, storage, mode='get')
# Detect content type
try:
@@ -784,11 +803,11 @@ class FileManagerAPI(PrivateGETAPI, PrivatePOSTAPI):
logger.debug('Reading "{}"'.format(path))
# Set support vars
- computing = self.get_computing(path, request)
+ storage = self.get_storage_from_path(path, request)
file_path = '/'+'/'.join(path.split('/')[2:])
# Get file contents
- data = self.cat(file_path, request.user, computing)
+ data = self.cat(file_path, request.user, storage)
# Return file contents
return HttpResponse(data, status=status.HTTP_200_OK)
@@ -798,7 +817,7 @@ class FileManagerAPI(PrivateGETAPI, PrivatePOSTAPI):
logger.debug('Deleting "{}"'.format(path))
# Set support vars
- computing = self.get_computing(path, request)
+ storage = self.get_storage_from_path(path, request)
path = '/'+'/'.join(path.split('/')[2:])
# Is it a folder?
@@ -808,17 +827,17 @@ class FileManagerAPI(PrivateGETAPI, PrivatePOSTAPI):
is_folder=False
# Get file contents
- data = self.delete(path, request.user, computing)
+ data = self.delete(path, request.user, storage)
# Response data
data = { 'data': {
- 'id': '/{}{}'.format(computing.name, path),
+ 'id': '/{}{}'.format(storage.id, path),
'type': 'folder' if is_folder else 'file',
'attributes':{
'name': path,
'readable': 1,
'writable': 1,
- 'path': '/{}{}'.format(computing.name, path)
+ 'path': '/{}{}'.format(storage.id, path)
}
}
}
@@ -836,22 +855,22 @@ class FileManagerAPI(PrivateGETAPI, PrivatePOSTAPI):
raise ValueError('No folder name set')
# Set support vars
- computing = self.get_computing(path, request)
+ storage = self.get_storage_from_path(path, request)
path = '/'+'/'.join(path.split('/')[2:]) + name
# Get file contents
- data = self.mkdir(path, request.user, computing)
+ data = self.mkdir(path, request.user, storage)
# Response data
data = { 'data': {
- 'id': '/{}{}'.format(computing.name, path),
+ 'id': '/{}{}'.format(storage.id, path),
'type': 'folder',
'attributes':{
'modified': now_t(), # This is an approximation!
'name': name,
'readable': 1,
'writable': 1,
- 'path': '/{}{}'.format(computing.name, path)
+ 'path': '/{}{}'.format(storage.id, path)
}
}
}
@@ -870,7 +889,7 @@ class FileManagerAPI(PrivateGETAPI, PrivatePOSTAPI):
raise Exception('Missing old name')
# Set support vars
- computing = self.get_computing(old_name_with_path, request)
+ storage = self.get_storage_from_path(old_name_with_path, request)
old_name_with_path = '/'+'/'.join(old_name_with_path.split('/')[2:])
# Is it a folder?
@@ -891,25 +910,25 @@ class FileManagerAPI(PrivateGETAPI, PrivatePOSTAPI):
new_name_with_path = '/'.join(old_name_with_path.split('/')[:-1]) + '/' + new_name
# Rename
- self.rename(old_name_with_path, new_name_with_path, request.user, computing)
+ self.rename(old_name_with_path, new_name_with_path, request.user, storage)
# Add trailing slash for listing
if is_folder:
new_name_with_path = new_name_with_path+'/'
# Get new info
- stat = self.stat(new_name_with_path, request.user, computing)
+ stat = self.stat(new_name_with_path, request.user, storage)
# Response data
data = { 'data': {
- 'id': '/{}{}'.format(computing.name, new_name_with_path),
+ 'id': '/{}{}'.format(storage.id, new_name_with_path),
'type': 'folder' if is_folder else 'file',
'attributes':{
'modified': stat['timestamp'],
'name': new_name,
'readable': 1,
'writable': 1,
- 'path': '/{}{}'.format(computing.name, new_name_with_path)
+ 'path': '/{}{}'.format(storage.id, new_name_with_path)
}
}
}
@@ -942,7 +961,7 @@ class FileManagerAPI(PrivateGETAPI, PrivatePOSTAPI):
# Set support vars
- computing = self.get_computing(source_name_with_path, request)
+ storage = self.get_storage_from_path(source_name_with_path, request)
if is_folder:
source_name_with_path = '/'+'/'.join(source_name_with_path.split('/')[2:])[:-1]
@@ -960,25 +979,25 @@ class FileManagerAPI(PrivateGETAPI, PrivatePOSTAPI):
#logger.debug('Copy target: "{}"'.format(target_name_with_path))
# Rename
- self.copy(source_name_with_path, target_name_with_path, request.user, computing)
+ self.copy(source_name_with_path, target_name_with_path, request.user, storage)
# Add trailing slash for listing
if is_folder:
target_name_with_path = target_name_with_path + '/'
# Get new info
- stat = self.stat(target_name_with_path, request.user, computing)
+ stat = self.stat(target_name_with_path, request.user, storage)
# Response data
data = { 'data': {
- 'id': '/{}{}'.format(computing.name, target_name_with_path),
+ 'id': '/{}{}'.format(storage.id, target_name_with_path),
'type': 'folder' if is_folder else 'file',
'attributes':{
'modified': stat['timestamp'],
'name': target_name_with_path.split('/')[-2] if is_folder else target_name_with_path.split('/')[-1],
'readable': 1,
'writable': 1,
- 'path': '/{}{}'.format(computing.name, target_name_with_path)
+ 'path': '/{}{}'.format(storage.id, target_name_with_path)
}
}
}
@@ -998,7 +1017,7 @@ class FileManagerAPI(PrivateGETAPI, PrivatePOSTAPI):
#============================
- # POST
+ # API POST
#============================
def _post(self, request):
@@ -1014,7 +1033,7 @@ class FileManagerAPI(PrivateGETAPI, PrivatePOSTAPI):
elif mode == 'upload':
# Set support vars
- computing = self.get_computing(path, request)
+ storage = self.get_storage_from_path(path, request)
path = '/'+'/'.join(path.split('/')[2:])
# Get the file upload
@@ -1029,11 +1048,11 @@ class FileManagerAPI(PrivateGETAPI, PrivatePOSTAPI):
logger.debug('Wrote "/tmp/{}" for "{}"'.format(file_uuid, file_upload.name))
# Now copy with scp
- self.scp('/tmp/{}'.format(file_uuid), path + file_upload.name , request.user, computing, mode='put')
+ self.scp_to('/tmp/{}'.format(file_uuid), path + file_upload.name , request.user, storage, mode='put')
# Response data
data = { 'data': [{
- 'id': '/{}{}{}'.format(computing.name, path, file_upload.name),
+ 'id': '/{}{}{}'.format(storage.id, path, file_upload.name),
'type': 'file',
'attributes':{
'modified': now_t(), # This is an approximation!
@@ -1041,7 +1060,7 @@ class FileManagerAPI(PrivateGETAPI, PrivatePOSTAPI):
'readable': 1,
'size': os.path.getsize('/tmp/{}'.format(file_uuid)), # This is kind of an approximation!
'writable': 1,
- 'path': '/{}{}{}'.format(computing.name, path, file_upload.name)
+ 'path': '/{}{}{}'.format(storage.id, path, file_upload.name)
}
}]
}
diff --git a/services/webapp/code/rosetta/core_app/computing_managers.py b/services/webapp/code/rosetta/core_app/computing_managers.py
index 792e5a7..a5a2116 100644
--- a/services/webapp/code/rosetta/core_app/computing_managers.py
+++ b/services/webapp/code/rosetta/core_app/computing_managers.py
@@ -1,4 +1,4 @@
-from .models import TaskStatuses, KeyPair, Task
+from .models import TaskStatuses, KeyPair, Task, Storage
from .utils import os_shell
from .exceptions import ErrorMessage, ConsistencyException
from django.conf import settings
@@ -180,9 +180,9 @@ class SSHSingleNodeComputingManager(SingleNodeComputingManager, SSHComputingMana
def _start_task(self, task, **kwargs):
logger.debug('Starting a remote task "{}"'.format(self.computing))
- # Get computing host
- host = self.computing.conf.get('host')
- user = self.computing.conf.get('user')
+ # Get computing user and host
+ computing_user = self.computing.conf.get('user')
+ computing_host = self.computing.conf.get('host')
# Get user keys
if self.computing.requires_user_keys:
@@ -195,7 +195,7 @@ class SSHSingleNodeComputingManager(SingleNodeComputingManager, SSHComputingMana
webapp_conn_string = get_webapp_conn_string()
# Handle container runtime
- if task.computing.default_container_runtime == 'singularity':
+ if self.computing.default_container_runtime == 'singularity':
#if not task.container.supports_custom_interface_port:
# raise Exception('This task does not support dynamic port allocation and is therefore not supported using singularity on Slurm')
@@ -205,24 +205,40 @@ class SSHSingleNodeComputingManager(SingleNodeComputingManager, SSHComputingMana
if not task.requires_proxy_auth and task.password:
authstring = ' export SINGULARITYENV_AUTH_PASS={} && '.format(task.password)
- # Set binds, only from sys config if the resource is not owned by the user
- if self.computing.user != task.user:
- binds = self.computing.sys_conf.get('binds')
- else:
- binds = self.computing.conf.get('binds')
- if not binds:
- binds = ''
- else:
- binds = '-B {}'.format(binds)
-
- # Manage task extra binds
- if task.extra_binds:
- if not binds:
- binds = '-B {}'.format(task.extra_binds)
- else:
- binds += ',{}'.format(task.extra_binds)
+ # Handle storages (binds)
+ binds = ''
+ storages = Storage.objects.filter(computing=self.computing)
+ for storage in storages:
+ if storage.type == 'generic_posix' and storage.bind_path:
+
+ # Expand the base path
+ expanded_base_path = storage.base_path
+ if '$SSH_USER' in expanded_base_path:
+ if storage.access_through_computing:
+ self.computing.attach_user_conf(self.computing.user)
+ expanded_base_path = expanded_base_path.replace('$SSH_USER', computing_user)
+ else:
+ raise NotImplementedError('Accessing a storage with ssh+cli without going through its computing resource is not implemented')
+ if '$USER' in expanded_base_path:
+ expanded_base_path = expanded_base_path.replace('$USER', self.task.user.name)
+
+ # Expand the bind_path
+ expanded_bind_path = storage.bind_path
+ if '$SSH_USER' in expanded_bind_path:
+ if storage.access_through_computing:
+ expanded_bind_path = expanded_bind_path.replace('$SSH_USER', computing_user)
+ else:
+ raise NotImplementedError('Accessing a storage with ssh+cli without going through its computing resource is not implemented')
+ if '$USER' in expanded_bind_path:
+ expanded_bind_path = expanded_bind_path.replace('$USER', self.task.user.name)
+
+ # Add the bind
+ if not binds:
+ binds = '-B {}:{}'.format(expanded_base_path, expanded_bind_path)
+ else:
+ binds += ',{}:{}'.format(expanded_base_path, expanded_bind_path)
- run_command = 'ssh -o LogLevel=ERROR -i {} -4 -o StrictHostKeyChecking=no {}@{} '.format(user_keys.private_key_file, user, host)
+ run_command = 'ssh -o LogLevel=ERROR -i {} -4 -o StrictHostKeyChecking=no {}@{} '.format(user_keys.private_key_file, computing_user, computing_host)
run_command += '/bin/bash -c \'"rm -rf /tmp/{}_data && mkdir -p /tmp/{}_data/tmp && mkdir -p /tmp/{}_data/home && chmod 700 /tmp/{}_data && '.format(task.uuid, task.uuid, task.uuid, task.uuid)
run_command += 'wget {}/api/v1/base/agent/?task_uuid={} -O /tmp/{}_data/agent.py &> /dev/null && export BASE_PORT=\$(python /tmp/{}_data/agent.py 2> /tmp/{}_data/task.log) && '.format(webapp_conn_string, task.uuid, task.uuid, task.uuid, task.uuid)
run_command += 'export SINGULARITY_NOHTTPS=true && export SINGULARITYENV_BASE_PORT=\$BASE_PORT && {} '.format(authstring)
@@ -306,8 +322,8 @@ class SlurmSSHClusterComputingManager(ClusterComputingManager, SSHComputingManag
logger.debug('Starting a remote task "{}"'.format(self.computing))
# Get computing host
- host = self.computing.conf.get('host')
- user = self.computing.conf.get('user')
+ computing_host = self.computing.conf.get('host')
+ computing_user = self.computing.conf.get('user')
# Get user keys
if self.computing.requires_user_keys:
@@ -350,24 +366,40 @@ class SlurmSSHClusterComputingManager(ClusterComputingManager, SSHComputingManag
if not task.requires_proxy_auth and task.password:
authstring = ' export SINGULARITYENV_AUTH_PASS={} && '.format(task.password)
- # Set binds, only from sys config if the resource is not owned by the user
- if self.computing.user != task.user:
- binds = self.computing.sys_conf.get('binds')
- else:
- binds = self.computing.conf.get('binds')
- if not binds:
- binds = ''
- else:
- binds = '-B {}'.format(binds)
-
- # Manage task extra binds
- if task.extra_binds:
- if not binds:
- binds = '-B {}'.format(task.extra_binds)
- else:
- binds += ',{}'.format(task.extra_binds)
-
- run_command = 'ssh -o LogLevel=ERROR -i {} -4 -o StrictHostKeyChecking=no {}@{} '.format(user_keys.private_key_file, user, host)
+ # Handle storages (binds)
+ binds = ''
+ storages = Storage.objects.filter(computing=self.computing)
+ for storage in storages:
+ if storage.type == 'generic_posix' and storage.bind_path:
+
+ # Expand the base path
+ expanded_base_path = storage.base_path
+ if '$SSH_USER' in expanded_base_path:
+ if storage.access_through_computing:
+ self.computing.attach_user_conf(self.computing.user)
+ expanded_base_path = expanded_base_path.replace('$SSH_USER', computing_user)
+ else:
+ raise NotImplementedError('Accessing a storage with ssh+cli without going through its computing resource is not implemented')
+ if '$USER' in expanded_base_path:
+ expanded_base_path = expanded_base_path.replace('$USER', self.task.user.name)
+
+ # Expand the bind_path
+ expanded_bind_path = storage.bind_path
+ if '$SSH_USER' in expanded_bind_path:
+ if storage.access_through_computing:
+ expanded_bind_path = expanded_bind_path.replace('$SSH_USER', computing_user)
+ else:
+ raise NotImplementedError('Accessing a storage with ssh+cli without going through its computing resource is not implemented')
+ if '$USER' in expanded_bind_path:
+ expanded_bind_path = expanded_bind_path.replace('$USER', self.task.user.name)
+
+ # Add the bind
+ if not binds:
+ binds = '-B {}:{}'.format(expanded_base_path, expanded_bind_path)
+ else:
+ binds += ',{}:{}'.format(expanded_base_path, expanded_bind_path)
+
+ run_command = 'ssh -o LogLevel=ERROR -i {} -4 -o StrictHostKeyChecking=no {}@{} '.format(user_keys.private_key_file, computing_user, computing_host)
run_command += '\'bash -c "echo \\"#!/bin/bash\nwget {}/api/v1/base/agent/?task_uuid={} -O \$HOME/agent_{}.py &> \$HOME/{}.log && export BASE_PORT=\\\\\\$(python \$HOME/agent_{}.py 2> \$HOME/{}.log) && '.format(webapp_conn_string, task.uuid, task.uuid, task.uuid, task.uuid, task.uuid)
run_command += 'export SINGULARITY_NOHTTPS=true && export SINGULARITYENV_BASE_PORT=\\\\\\$BASE_PORT && {} '.format(authstring)
run_command += 'rm -rf /tmp/{}_data && mkdir -p /tmp/{}_data/tmp &>> \$HOME/{}.log && mkdir -p /tmp/{}_data/home &>> \$HOME/{}.log && chmod 700 /tmp/{}_data && '.format(task.uuid, task.uuid, task.uuid, task.uuid, task.uuid, task.uuid)
diff --git a/services/webapp/code/rosetta/core_app/management/commands/core_app_populate.py b/services/webapp/code/rosetta/core_app/management/commands/core_app_populate.py
index 6b83541..46f32b8 100644
--- a/services/webapp/code/rosetta/core_app/management/commands/core_app_populate.py
+++ b/services/webapp/code/rosetta/core_app/management/commands/core_app_populate.py
@@ -1,6 +1,6 @@
from django.core.management.base import BaseCommand
from django.contrib.auth.models import User
-from ...models import Profile, Container, Computing, ComputingSysConf, ComputingUserConf, KeyPair, Text
+from ...models import Profile, Container, Computing, ComputingSysConf, ComputingUserConf, Storage, KeyPair, Text
class Command(BaseCommand):
help = 'Adds the admin superuser with \'a\' password.'
@@ -233,3 +233,55 @@ class Command(BaseCommand):
data = {'user': 'slurmtestuser'})
+
+ #=====================
+ # Storages
+ #=====================
+ storages = Storage.objects.all()
+ if storages:
+ print('Not creating demo storages as they already exist')
+ else:
+ print('Creating demo storages...')
+
+ # Get demo computing resources
+ demo_computing_resources = []
+ try:
+ demo_slurm_computing = Computing.objects.get(name='Demo Cluster')
+ demo_computing_resources.append(demo_slurm_computing)
+ except:
+ pass
+ try:
+ demo_standalone_computing = Computing.objects.get(name='Demo Standalone')
+ demo_computing_resources.append(demo_standalone_computing)
+ except:
+ pass
+
+
+ for computing in demo_computing_resources:
+ # Demo shared computing plus conf
+ Storage.objects.create(user = None,
+ computing = computing,
+ access_through_computing = True,
+ name = 'Shared',
+ type = 'generic_posix',
+ access_mode = 'ssh+cli',
+ auth_mode = 'user_keys',
+ base_path = '/shared/data/shared',
+ bind_path = '/storages/shared')
+
+ # Demo shared computing plus conf
+ Storage.objects.create(user = None,
+ computing = computing,
+ access_through_computing = True,
+ name = 'Personal',
+ type = 'generic_posix',
+ access_mode = 'ssh+cli',
+ auth_mode = 'user_keys',
+ base_path = '/shared/data/users/$SSH_USER',
+ bind_path = '/storages/personal')
+
+
+
+
+
+
diff --git a/services/webapp/code/rosetta/core_app/migrations/0011_storage.py b/services/webapp/code/rosetta/core_app/migrations/0011_storage.py
new file mode 100644
index 0000000..78c1666
--- /dev/null
+++ b/services/webapp/code/rosetta/core_app/migrations/0011_storage.py
@@ -0,0 +1,37 @@
+# Generated by Django 2.2.1 on 2021-11-08 14:45
+
+from django.conf import settings
+import django.contrib.postgres.fields.jsonb
+from django.db import migrations, models
+import django.db.models.deletion
+import uuid
+
+
+class Migration(migrations.Migration):
+
+ dependencies = [
+ migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+ ('core_app', '0010_profile_is_power_user'),
+ ]
+
+ operations = [
+ migrations.CreateModel(
+ name='Storage',
+ fields=[
+ ('uuid', models.UUIDField(default=uuid.uuid4, editable=False, primary_key=True, serialize=False)),
+ ('name', models.CharField(max_length=255, verbose_name='Name')),
+ ('type', models.CharField(max_length=255, verbose_name='Type')),
+ ('access_mode', models.CharField(max_length=36, verbose_name='Access (control) mode')),
+ ('auth_mode', models.CharField(max_length=36, verbose_name='Auth mode')),
+ ('base_path', models.CharField(max_length=4096, verbose_name='Base path')),
+ ('bind_path', models.CharField(max_length=4096, verbose_name='Bind path')),
+ ('access_through_computing', models.BooleanField(default=False, verbose_name='Access through linked computing resource?')),
+ ('config', django.contrib.postgres.fields.jsonb.JSONField(blank=True, null=True)),
+ ('computing', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.CASCADE, related_name='storages', to='core_app.Computing')),
+ ('user', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.CASCADE, related_name='+', to=settings.AUTH_USER_MODEL)),
+ ],
+ options={
+ 'ordering': ['name'],
+ },
+ ),
+ ]
diff --git a/services/webapp/code/rosetta/core_app/models.py b/services/webapp/code/rosetta/core_app/models.py
index d4cea0d..a5c5caa 100644
--- a/services/webapp/code/rosetta/core_app/models.py
+++ b/services/webapp/code/rosetta/core_app/models.py
@@ -381,6 +381,58 @@ class Task(models.Model):
return get_task_tunnel_host()
+
+
+
+#=========================
+# Storages
+#=========================
+
+class Storage(models.Model):
+
+ uuid = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False)
+ user = models.ForeignKey(User, related_name='+', on_delete=models.CASCADE, blank=True, null=True)
+
+ name = models.CharField('Name', max_length=255, blank=False, null=False)
+ #description = models.TextField('Description', blank=True, null=True)
+
+ # Storage type
+ type = models.CharField('Type', max_length=255, blank=False, null=False)
+
+ # Access and auth mode
+ access_mode = models.CharField('Access (control) mode', max_length=36, blank=False, null=False)
+ auth_mode = models.CharField('Auth mode', max_length=36, blank=False, null=False)
+
+ # Paths
+ base_path = models.CharField('Base path', max_length=4096, blank=False, null=False)
+ bind_path = models.CharField('Bind path', max_length=4096, blank=False, null=False)
+
+ # Link with a computing resource
+ computing = models.ForeignKey(Computing, related_name='storages', on_delete=models.CASCADE, blank=True, null=True) # Make optional?
+ access_through_computing = models.BooleanField('Access through linked computing resource?', default=False)
+ # If the above is linked, some configuration can be taken from the linked computing resource (i.e. the hostname)
+
+ # Configuration
+ config = JSONField(blank=True, null=True)
+
+
+ class Meta:
+ ordering = ['name']
+
+ def __str__(self):
+ if self.user:
+ return str('Storage "{}" of user "{}"'.format(self.id, self.user))
+ else:
+ return str('Storage "{}"'.format(self.id))
+
+ @property
+ def id(self):
+ return (self.name if not self.computing else '{}@{}'.format(self.name,self.computing.name))
+
+
+
+
+
#=========================
# KeyPair
#=========================
--
GitLab