diff --git a/doc/source/admin/galaxy_options.rst b/doc/source/admin/galaxy_options.rst index 895ad01274b..0eacbacbada 100644 --- a/doc/source/admin/galaxy_options.rst +++ b/doc/source/admin/galaxy_options.rst @@ -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`` ~~~~~~~~~~~~~~~~~~~~~~~~ diff --git a/lib/galaxy/config/sample/galaxy.yml.sample b/lib/galaxy/config/sample/galaxy.yml.sample index ae2da998746..de816cd969f 100644 --- a/lib/galaxy/config/sample/galaxy.yml.sample +++ b/lib/galaxy/config/sample/galaxy.yml.sample @@ -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 # . diff --git a/lib/galaxy/jobs/runners/kubernetes.py b/lib/galaxy/jobs/runners/kubernetes.py index 6942acfc9a7..cce7183fb83 100644 --- a/lib/galaxy/jobs/runners/kubernetes.py +++ b/lib/galaxy/jobs/runners/kubernetes.py @@ -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') diff --git a/lib/galaxy/managers/interactivetool.py b/lib/galaxy/managers/interactivetool.py index 207f802cb35..6a72a05e859 100644 --- a/lib/galaxy/managers/interactivetool.py +++ b/lib/galaxy/managers/interactivetool.py @@ -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] != "/": diff --git a/lib/galaxy/webapps/galaxy/config_schema.yml b/lib/galaxy/webapps/galaxy/config_schema.yml index c8540a33011..ca3f3bcd6ac 100644 --- a/lib/galaxy/webapps/galaxy/config_schema.yml +++ b/lib/galaxy/webapps/galaxy/config_schema.yml @@ -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