opensource/galaxy
3
0
Fork 0
mirror of https://github.com/galaxyproject/galaxy.git synced 2026-09-24 16:30:27 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #11989 from almahmoud/k8s_runner_its

Browse Source 
[21.05] More IT and k8s runner fixes
This commit is contained in:
John Chilton
2021-06-23 10:59:40 -04:00
committed by GitHub
parent a1300e3ef9 b6092e8c4c
commit 16d22d02f4
5 changed files with 168 additions and 127 deletions
Download Patch File Download Diff File Expand all files Collapse all files
doc/source/admin/galaxy_options.rst
+11
View File
@@ -1601,6 +1601,17 @@
:Type: str
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
``interactivetools_base_path``
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
:Description:
Base path for interactive tools running at a subpath without a
subdomain. Defaults to "/".
:Default: ``/``
:Type: str
~~~~~~~~~~~~~~~~~~~~~~~~
``interactivetools_map``
~~~~~~~~~~~~~~~~~~~~~~~~
lib/galaxy/config/sample/galaxy.yml.sample
+4
View File
@@ -877,6 +877,10 @@ galaxy:
# as Galaxy by default.
#interactivetools_proxy_host: null
# Base path for interactive tools running at a subpath without a
# subdomain. Defaults to "/".
#interactivetools_base_path: /
# Map for interactivetool proxy.
# The value of this option will be resolved with respect to
# <data_dir>.
lib/galaxy/jobs/runners/kubernetes.py
+136 -118
View File
@@ -74,11 +74,11 @@ class KubernetesJobRunner(AsynchronousJobRunner):
k8s_job_api_version=dict(map=str, default=DEFAULT_JOB_API_VERSION),
k8s_job_ttl_secs_after_finished=dict(map=int, valid=lambda x: x is None or int(x) >= 0, default=None),
k8s_job_metadata=dict(map=str, default=None),
k8s_supplemental_group_id=dict(map=str),
k8s_supplemental_group_id=dict(map=str, valid=lambda s: s == "$gid" or isinstance(s, int) or not s or s.isdigit(), default=None),
k8s_pull_policy=dict(map=str, default="Default"),
k8s_run_as_user_id=dict(map=str, valid=lambda s: s == "$uid" or isinstance(s, int) or s.isdigit()),
k8s_run_as_group_id=dict(map=str, valid=lambda s: s == "$gid" or isinstance(s, int) or s.isdigit()),
k8s_fs_group_id=dict(map=str, valid=lambda s: s == "$gid" or isinstance(s, int) or s.isdigit()),
k8s_run_as_user_id=dict(map=str, valid=lambda s: s == "$uid" or isinstance(s, int) or not s or s.isdigit(), default=None),
k8s_run_as_group_id=dict(map=str, valid=lambda s: s == "$gid" or isinstance(s, int) or not s or s.isdigit(), default=None),
k8s_fs_group_id=dict(map=str, valid=lambda s: s == "$gid" or isinstance(s, int) or not s or s.isdigit(), default=None),
k8s_cleanup_job=dict(map=str, valid=lambda s: s in {"onsuccess", "always", "never"}, default="always"),
k8s_pod_retries=dict(map=int, valid=lambda x: int(x) >= 0, default=3),
k8s_walltime_limit=dict(map=int, valid=lambda x: int(x) >= 0, default=172800),
@@ -151,10 +151,33 @@ class KubernetesJobRunner(AsynchronousJobRunner):
# Construction of the Kubernetes Job object follows: http://kubernetes.io/docs/user-guide/persistent-volumes/
k8s_job_prefix = self.__produce_k8s_job_prefix()
guest_ports = ajs.job_wrapper.guest_ports
ports_dict = {}
for guest_port in guest_ports:
ports_dict[str(guest_port)] = dict(host='manual', port=guest_port, protocol="https")
eps = None
if ajs.job_wrapper.guest_ports:
k8s_job_name = self.__get_k8s_job_name(k8s_job_prefix, ajs.job_wrapper)
log.debug(f'Configuring entry points and deploying service/ingress for job with ID {ajs.job_id}')
k8s_service_obj = service_object_dict(
self.runner_params,
k8s_job_name,
self.__get_k8s_service_spec(ajs)
)
eps = self.app.interactivetool_manager.configure_entry_points(ajs.job_wrapper.get_job(), ports_dict)
k8s_ingress_obj = ingress_object_dict(
self.runner_params,
k8s_job_name,
self.__get_k8s_ingress_spec(ajs, eps)
)
service = Service(self._pykube_api, k8s_service_obj)
service.create()
ingress = Ingress(self._pykube_api, k8s_ingress_obj)
ingress.create()
k8s_job_obj = job_object_dict(
self.runner_params,
k8s_job_prefix,
self.__get_k8s_job_spec(ajs)
self.__get_k8s_job_spec(ajs, eps)
)
job = Job(self._pykube_api, k8s_job_obj)
try:
@@ -178,39 +201,48 @@ class KubernetesJobRunner(AsynchronousJobRunner):
return pull_policy(self.runner_params)
def __get_run_as_user_id(self):
if "k8s_run_as_user_id" in self.runner_params:
if self.runner_params.get("k8s_run_as_user_id") or self.runner_params.get("k8s_run_as_user_id") == 0:
run_as_user = self.runner_params["k8s_run_as_user_id"]
if run_as_user == "$uid":
return os.getuid()
else:
return int(self.runner_params["k8s_run_as_user_id"])
try:
return int(self.runner_params["k8s_run_as_user_id"])
except Exception:
log.warning("User ID passed for Kubernetes runner needs to be an integer or \"$uid\", value "
+ self.runner_params["k8s_run_as_user_id"] + " passed is invalid")
return None
return None
def __get_run_as_group_id(self):
if "k8s_run_as_group_id" in self.runner_params:
if self.runner_params.get("k8s_run_as_group_id") or self.runner_params.get("k8s_run_as_group_id") == 0:
run_as_group = self.runner_params["k8s_run_as_group_id"]
if run_as_group == "$gid":
return self.app.config.gid
else:
return int(self.runner_params["k8s_run_as_group_id"])
try:
return int(self.runner_params["k8s_run_as_group_id"])
except Exception:
log.warning("Group ID passed for Kubernetes runner needs to be an integer or \"$gid\", value "
+ self.runner_params["k8s_run_as_group_id"] + " passed is invalid")
return None
def __get_supplemental_group(self):
if "k8s_supplemental_group_id" in self.runner_params:
if self.runner_params.get("k8s_supplemental_group_id") or self.runner_params.get("k8s_supplemental_group_id") == 0:
try:
return int(self.runner_params["k8s_supplemental_group_id"])
except Exception:
log.warning("Supplemental group passed for Kubernetes runner needs to be an integer, value "
log.warning("Supplemental group passed for Kubernetes runner needs to be an integer or \"$gid\", value "
+ self.runner_params["k8s_supplemental_group_id"] + " passed is invalid")
return None
return None
def __get_fs_group(self):
if "k8s_fs_group_id" in self.runner_params:
if self.runner_params.get("k8s_fs_group_id") or self.runner_params.get("k8s_fs_group_id") == 0:
try:
return int(self.runner_params["k8s_fs_group_id"])
except Exception:
log.warning("FS group passed for Kubernetes runner needs to be an integer, value "
log.warning("FS group passed for Kubernetes runner needs to be an integer or \"$gid\", value "
+ self.runner_params["k8s_fs_group_id"] + " passed is invalid")
return None
return None
@@ -223,10 +255,10 @@ class KubernetesJobRunner(AsynchronousJobRunner):
instance_id = self._galaxy_instance_id or ''
return produce_k8s_job_prefix(app_prefix='gxy', instance_id=instance_id)
def __get_k8s_job_spec(self, ajs):
def __get_k8s_job_spec(self, ajs, eps=None):
"""Creates the k8s Job spec. For a Job spec, the only requirement is to have a .spec.template.
If the job hangs around unlimited it will be ended after k8s wall time limit, which sets activeDeadlineSeconds"""
k8s_job_spec = {"template": self.__get_k8s_job_spec_template(ajs),
k8s_job_spec = {"template": self.__get_k8s_job_spec_template(ajs, eps),
"activeDeadlineSeconds": int(self.runner_params['k8s_walltime_limit'])}
job_ttl = self.runner_params["k8s_job_ttl_secs_after_finished"]
if self.runner_params["k8s_cleanup_job"] != "never" and job_ttl is not None:
@@ -247,7 +279,7 @@ class KubernetesJobRunner(AsynchronousJobRunner):
label_val += 'x'
return label_val
def __get_k8s_job_spec_template(self, ajs):
def __get_k8s_job_spec_template(self, ajs, eps=None):
"""The k8s spec template is nothing but a Pod spec, except that it is nested and does not have an apiversion
nor kind. In addition to required fields for a Pod, a pod template in a job must specify appropriate labels
(see pod selector) and an appropriate restart policy."""
@@ -272,7 +304,7 @@ class KubernetesJobRunner(AsynchronousJobRunner):
"spec": {
"volumes": self.runner_params['k8s_mountable_volumes'],
"restartPolicy": self.__get_k8s_restart_policy(ajs.job_wrapper),
"containers": self.__get_k8s_containers(ajs),
"containers": self.__get_k8s_containers(ajs, eps),
"priorityClassName": self.runner_params['k8s_pod_priority_class'],
"tolerations": yaml.safe_load(self.runner_params['k8s_tolerations'] or "[]"),
"affinity": yaml.safe_load(self.__get_overridable_params(ajs.job_wrapper,
@@ -295,100 +327,90 @@ class KubernetesJobRunner(AsynchronousJobRunner):
"""The k8s spec template is nothing but a Service spec, except that it is nested and does not have an apiversion
nor kind."""
guest_ports = ajs.job_wrapper.guest_ports
if not guest_ports:
return None
else:
k8s_spec_template = {
"metadata": {
"labels": {
"app.galaxyproject.org/handler": self.__force_label_conformity(self.app.config.server_name),
"app.galaxyproject.org/destination": self.__force_label_conformity(
str(ajs.job_wrapper.job_destination.id))
},
"annotations": {
"app.galaxyproject.org/tool_id": ajs.job_wrapper.tool.id
}
k8s_spec_template = {
"metadata": {
"labels": {
"app.galaxyproject.org/handler": self.__force_label_conformity(self.app.config.server_name),
"app.galaxyproject.org/destination": self.__force_label_conformity(
str(ajs.job_wrapper.job_destination.id))
},
"spec": {
"ports": [{"name": "job-{}-{}".format(self.__force_label_conformity(ajs.job_wrapper.get_id_tag()), p),
"port": int(p),
"protocol": "TCP",
"targetPort": int(p)} for p in guest_ports],
"selector": {
"app.kubernetes.io/name": self.__force_label_conformity(ajs.job_wrapper.tool.old_id),
"app.kubernetes.io/component": "tool",
"app.galaxyproject.org/job_id": self.__force_label_conformity(ajs.job_wrapper.get_id_tag())
},
"type": "ClusterIP"
"annotations": {
"app.galaxyproject.org/tool_id": ajs.job_wrapper.tool.id
}
},
"spec": {
"ports": [{"name": "job-{}-{}".format(self.__force_label_conformity(ajs.job_wrapper.get_id_tag()), p),
"port": int(p),
"protocol": "TCP",
"targetPort": int(p)} for p in guest_ports],
"selector": {
"app.kubernetes.io/name": self.__force_label_conformity(ajs.job_wrapper.tool.old_id),
"app.kubernetes.io/component": "tool",
"app.galaxyproject.org/job_id": self.__force_label_conformity(ajs.job_wrapper.get_id_tag())
},
"type": "ClusterIP"
}
return k8s_spec_template
}
return k8s_spec_template
def __get_k8s_ingress_spec(self, ajs):
def __get_k8s_ingress_spec(self, ajs, eps=None):
"""The k8s spec template is nothing but a Ingress spec, except that it is nested and does not have an apiversion
nor kind."""
guest_ports = ajs.job_wrapper.guest_ports
if not guest_ports:
return None
else:
if len(guest_ports) > 0:
ports_dict = {}
for guest_port in guest_ports:
ports_dict[str(guest_port)] = dict(host='manual', port=guest_port, protocol="https")
eps = self.app.interactivetool_manager.configure_entry_points(ajs.job_wrapper.get_job(), ports_dict)
entry_points = []
for entry_point in eps.get('configured', []):
# sending in self.app as `trans` since it's only used for `.security` so seems to work
entry_point_path = self.app.interactivetool_manager.get_entry_point_path(self.app, entry_point)
if '?' in entry_point_path:
# Removing all the parameters from the ingress path, but they will still be in the database
# so the link that the user clicks on will still have them
log.warn("IT urls including parameters (eg: /myit?mykey=myvalue) are only experimentally supported on K8S")
entry_point_path = entry_point_path.split('?')[0]
entry_point_domain = f'{self.app.config.interactivetools_proxy_host}'
if entry_point.requires_domain:
entry_point_subdomain = self.app.interactivetool_manager.get_entry_point_subdomain(self.app, entry_point)
entry_point_domain = f'{entry_point_subdomain}.{entry_point_domain}'
entry_point_path = '/'
entry_points.append({"tool_port": entry_point.tool_port, "domain": entry_point_domain, "entry_path": entry_point_path})
k8s_spec_template = {
"metadata": {
"labels": {
"app.galaxyproject.org/handler": self.__force_label_conformity(self.app.config.server_name),
"app.galaxyproject.org/destination": self.__force_label_conformity(
str(ajs.job_wrapper.job_destination.id))
},
"annotations": {
"app.galaxyproject.org/tool_id": ajs.job_wrapper.tool.id
}
if len(guest_ports) > 0:
entry_points = []
for entry_point in eps.get('configured', []):
# sending in self.app as `trans` since it's only used for `.security` so seems to work
entry_point_path = self.app.interactivetool_manager.get_entry_point_path(self.app, entry_point)
if '?' in entry_point_path:
# Removing all the parameters from the ingress path, but they will still be in the database
# so the link that the user clicks on will still have them
log.warn("IT urls including parameters (eg: /myit?mykey=myvalue) are only experimentally supported on K8S")
entry_point_path = entry_point_path.split('?')[0]
entry_point_domain = f'{self.app.config.interactivetools_proxy_host}'
if entry_point.requires_domain:
entry_point_subdomain = self.app.interactivetool_manager.get_entry_point_subdomain(self.app, entry_point)
entry_point_domain = f'{entry_point_subdomain}.{entry_point_domain}'
entry_point_path = '/'
entry_points.append({"tool_port": entry_point.tool_port, "domain": entry_point_domain, "entry_path": entry_point_path})
k8s_spec_template = {
"metadata": {
"labels": {
"app.galaxyproject.org/handler": self.__force_label_conformity(self.app.config.server_name),
"app.galaxyproject.org/destination": self.__force_label_conformity(
str(ajs.job_wrapper.job_destination.id))
},
"spec": {
"rules": [{"host": ep["domain"],
"http": {
"paths": [{
"backend": {
"serviceName": self.__get_k8s_job_name(self.__produce_k8s_job_prefix(), ajs.job_wrapper),
"servicePort": int(ep["tool_port"])
},
"path": ep.get("entry_path", '/'),
"pathType": "Prefix"
}]}} for ep in entry_points]
"annotations": {
"app.galaxyproject.org/tool_id": ajs.job_wrapper.tool.id
}
},
"spec": {
"rules": [{"host": ep["domain"],
"http": {
"paths": [{
"backend": {
"serviceName": self.__get_k8s_job_name(self.__produce_k8s_job_prefix(), ajs.job_wrapper),
"servicePort": int(ep["tool_port"])
},
"path": ep.get("entry_path", '/'),
"pathType": "Prefix"
}]}} for ep in entry_points]
}
if self.runner_params.get("k8s_interactivetools_use_ssl"):
domains = list(set([e["domain"] for e in entry_points]))
k8s_spec_template["spec"]["tls"] = [{"hosts": [domain],
"secretName": re.sub("[^a-z0-9-]", "-", domain)} for domain in domains]
if self.runner_params.get("k8s_interactivetools_ingress_annotations"):
new_ann = yaml.safe_load(self.runner_params.get("k8s_interactivetools_ingress_annotations"))
k8s_spec_template["metadata"]["annotations"].update(new_ann)
return k8s_spec_template
}
if self.runner_params.get("k8s_interactivetools_use_ssl"):
domains = list(set([e["domain"] for e in entry_points]))
k8s_spec_template["spec"]["tls"] = [{"hosts": [domain],
"secretName": re.sub("[^a-z0-9-]", "-", domain)} for domain in domains]
if self.runner_params.get("k8s_interactivetools_ingress_annotations"):
new_ann = yaml.safe_load(self.runner_params.get("k8s_interactivetools_ingress_annotations"))
k8s_spec_template["metadata"]["annotations"].update(new_ann)
return k8s_spec_template
def __get_k8s_security_context(self):
security_context = {}
if self._run_as_user_id:
if self._run_as_user_id or self._run_as_user_id == 0:
security_context["runAsUser"] = self._run_as_user_id
if self._run_as_group_id:
if self._run_as_group_id or self._run_as_group_id == 0:
security_context["runAsGroup"] = self._run_as_group_id
if self._supplemental_group and self._supplemental_group > 0:
security_context["supplementalGroups"] = [self._supplemental_group]
@@ -400,7 +422,7 @@ class KubernetesJobRunner(AsynchronousJobRunner):
"""The default Kubernetes restart policy for Jobs"""
return "Never"
def __get_k8s_containers(self, ajs):
def __get_k8s_containers(self, ajs, eps=None):
"""Fills in all required for setting up the docker containers to be used, including setting a pull policy if
this has been set.
$GALAXY_VIRTUAL_ENV is set to None to avoid the galaxy virtualenv inside the tool container.
@@ -448,6 +470,22 @@ class KubernetesJobRunner(AsynchronousJobRunner):
extra_envs = yaml.safe_load(self.__get_overridable_params(ajs.job_wrapper, 'k8s_extra_job_envs') or "{}")
for key in extra_envs:
envs.append({'name': key, 'value': extra_envs[key]})
if eps:
for entry_point in eps.get('configured', []):
# sending in self.app as `trans` since it's only used for `.security` so seems to work
entry_point_path = self.app.interactivetool_manager.get_entry_point_path(self.app, entry_point)
if '?' in entry_point_path:
# Removing all the parameters from the ingress path, but they will still be in the database
# so the link that the user clicks on will still have them
log.warn("IT urls including parameters (eg: /myit?mykey=myvalue) are only experimentally supported on K8S")
entry_point_path = entry_point_path.split('?')[0]
entry_point_domain = f'{self.app.config.interactivetools_proxy_host}'
if entry_point.requires_domain:
entry_point_subdomain = self.app.interactivetool_manager.get_entry_point_subdomain(self.app, entry_point)
entry_point_domain = f'{entry_point_subdomain}.{entry_point_domain}'
envs.append({'name': 'INTERACTIVETOOL_PORT', 'value': str(entry_point.tool_port)})
envs.append({'name': 'INTERACTIVETOOL_DOMAIN', 'value': str(entry_point_domain)})
envs.append({'name': 'INTERACTIVETOOL_PATH', 'value': str(entry_point_path)})
k8s_container['resources'] = resources
k8s_container['env'] = envs
@@ -606,26 +644,6 @@ class KubernetesJobRunner(AsynchronousJobRunner):
return None
elif active > 0 and failed <= max_pod_retries:
if not job_state.running:
job_state.running = True
job_state.job_wrapper.change_state(model.Job.states.RUNNING)
if job_state.job_wrapper.guest_ports:
k8s_job_prefix = self.__produce_k8s_job_prefix()
k8s_job_name = self.__get_k8s_job_name(k8s_job_prefix, job_state.job_wrapper)
log.debug(f'Configuring entry points and deploying service/ingress for job with ID {job_state.job_id}')
k8s_service_obj = service_object_dict(
self.runner_params,
k8s_job_name,
self.__get_k8s_service_spec(job_state)
)
k8s_ingress_obj = ingress_object_dict(
self.runner_params,
k8s_job_name,
self.__get_k8s_ingress_spec(job_state)
)
service = Service(self._pykube_api, k8s_service_obj)
service.create()
ingress = Ingress(self._pykube_api, k8s_ingress_obj)
ingress.create()
if self.__job_pending_due_to_unschedulable_pod(job_state):
creation_time_str = job.obj['metadata'].get('creationTimestamp')
creation_time = datetime.strptime(creation_time_str, '%Y-%m-%dT%H:%M:%SZ')
lib/galaxy/managers/interactivetool.py
+9 -9
View File
@@ -1,7 +1,6 @@
import logging
import sqlite3
from sqlalchemy import or_
@@ -255,15 +254,15 @@ class InteractiveToolManager:
if entry_point.active and not entry_point.deleted:
request_host = trans.request.host
protocol = trans.request.host_url.split('//', 1)[0]
entry_point_encoded_id = trans.security.encode_id(entry_point.id)
entry_point_class = entry_point.__class__.__name__.lower()
entry_point_prefix = self.app.config.interactivetools_prefix
if entry_point.requires_domain:
rval = f'{protocol}//{self.get_entry_point_subdomain(trans, entry_point)}.{request_host}/'
if entry_point.entry_url:
rval = '{}/{}'.format(rval.rstrip('/'), entry_point.entry_url.lstrip('/'))
else:
rval = self.app.url_for(f'/{entry_point_prefix}/access/{entry_point_class}/{entry_point_encoded_id}/{entry_point.token}/')
if entry_point.entry_url:
rval = '{}/{}'.format(rval.rstrip('/'), entry_point.entry_url.lstrip('/'))
rval = self.get_entry_point_path(trans, entry_point)
# Add trailing slash to the URL
if rval[-1] != "/":
rval = f'{rval}/'
return rval
def get_entry_point_subdomain(self, trans, entry_point):
@@ -281,10 +280,11 @@ class InteractiveToolManager:
entry_point_prefix = self.app.config.interactivetools_prefix
rval = "/"
if not entry_point.requires_domain:
rval = str(self.app.config.interactivetools_base_path).rstrip("/").lstrip("/")
if self.app.config.interactivetools_shorten_url:
rval = self.app.url_for(f'/{entry_point_prefix}/{entry_point_encoded_id}/{entry_point.token[:10]}/')
rval = f'/{rval}/{entry_point_prefix}/{entry_point_encoded_id}/{entry_point.token[:10]}/'
else:
rval = self.app.url_for(f'/{entry_point_prefix}/access/{entry_point_class}/{entry_point_encoded_id}/{entry_point.token}/')
rval = f'/{rval}/{entry_point_prefix}/access/{entry_point_class}/{entry_point_encoded_id}/{entry_point.token}/'
if entry_point.entry_url:
rval = '{}/{}'.format(rval.rstrip('/'), entry_point.entry_url.lstrip('/'))
if rval[0] != "/":
lib/galaxy/webapps/galaxy/config_schema.yml
+8
View File
@@ -1152,10 +1152,18 @@ mapping:
interactivetools_proxy_host:
type: str
required: false
desc: |
Proxy host - assumed to just be hosted on the same hostname and port as
Galaxy by default.
interactivetools_base_path:
type: str
required: false
default: "/"
desc: |
Base path for interactive tools running at a subpath without a subdomain. Defaults to "/".
interactivetools_map:
type: str
default: interactivetools_map.sqlite