mirror of
https://github.com/galaxyproject/galaxy.git
synced 2026-09-24 16:30:27 +08:00
Implement new GIE proxy
This provides an alternative proxy to the original NodeJS proxy. The implementation is done in golang, which has the nice property of compiling to a single binary. While it's a negative in terms of number of devteam who write that language, I'd argue it's not that much worse than nodejs in terms of accessibility. Features include: - Simple rest-ish API instead of file-based IPC - Automatic culling of containers, no more monitor_traffic.sh hacks - Admin configurable, cross GIE timeout setting
This commit is contained in:
@@ -611,6 +611,10 @@ nglims_config_file = tool-data/nglims.yaml
|
||||
# launch the proxy if it is actually going to be used (e.g. for IPython).
|
||||
#dynamic_proxy_manage=True
|
||||
|
||||
# As of 16.04 Galaxy supports multiple proxy types. The original NodeJS
|
||||
# implementation, alongside a new Golang single-binary-no-dependencies
|
||||
# version. Valid values are (node, golang)
|
||||
#dynamic_proxy=node
|
||||
# Dynamic proxy can use an SQLite database or a JSON file for IPC, set that
|
||||
# here.
|
||||
#dynamic_proxy_session_map=database/session_map.sqlite
|
||||
@@ -633,6 +637,31 @@ nglims_config_file = tool-data/nglims.yaml
|
||||
# https://FQDN/galaxy-prefix/gie_proxy for proxying
|
||||
#dynamic_proxy_prefix=gie_proxy
|
||||
|
||||
# The Golang proxy has a large number of extra options which were not available
|
||||
# to the NodeJS proxy, as the Golang proxy takes over container management itself.
|
||||
# This should actually be a preferrable situation, as there will be less hackery
|
||||
# required in GIE configs.
|
||||
|
||||
# The length of time a container isn't accessed, before it is culled.
|
||||
# Previously, individual containers set their own time limit before killing
|
||||
# themselves. This made it harder for the administrator to centarlise configuration.
|
||||
# Since the golang proxy manages the containers itself, and tracks when the last packet
|
||||
# went through to a container, it has a much better idea of when the last connection was.
|
||||
dynamic_proxy_golang_noaccess = 60
|
||||
|
||||
# Given the noaccess time, the golang proxy checks regularly for containers
|
||||
# which need to be removed. Every $clean_interval seconds, the proxy will check through its route list
|
||||
# for all tracks which haven't been accessed in at least $noaccess seconds, and kill those.
|
||||
dynamic_proxy_golang_clean_interval = 10
|
||||
|
||||
# The golang proxy needs to know how to talk to your docker daemon
|
||||
dynamic_proxy_golang_docker_address = unix:///var/run/docker.sock
|
||||
|
||||
# The golang proxy uses a RESTful HTTP API for communication with Galaxy
|
||||
# instead of a JSON or SQLite file for IPC. If you do not specify this, it will
|
||||
# be set randomly for you.
|
||||
dynamic_proxy_golang_api_key = password
|
||||
|
||||
# -- Logging and Debugging
|
||||
|
||||
# If True, Galaxy will attempt to configure a simple root logger if a
|
||||
|
||||
@@ -463,6 +463,13 @@ class Configuration( object ):
|
||||
self.dynamic_proxy_external_proxy = string_as_bool( kwargs.get( "dynamic_proxy_external_proxy", "False" ) )
|
||||
self.dynamic_proxy_prefix = kwargs.get( "dynamic_proxy_prefix", "gie_proxy" )
|
||||
|
||||
self.dynamic_proxy = kwargs.get( "dynamic_proxy", "nodejs" )
|
||||
self.dynamic_proxy_golang_noaccess = kwargs.get( "dynamic_proxy_golang_noaccess", 60 )
|
||||
self.dynamic_proxy_golang_clean_interval = kwargs.get( "dynamic_proxy_golang_clean_interval", 10 )
|
||||
self.dynamic_proxy_golang_docker_address = kwargs.get( "dynamic_proxy_golang_docker_address", "unix:///var/run/docker.sock" )
|
||||
self.dynamic_proxy_golang_docker_address = kwargs.get( "dynamic_proxy_golang_docker_address", "unix:///var/run/docker.sock" )
|
||||
self.dynamic_proxy_golang_api_key = kwargs.get( "dynamic_proxy_golang_api_key", None )
|
||||
|
||||
# Default chunk size for chunkable datatypes -- 64k
|
||||
self.display_chunk_size = int( kwargs.get( 'display_chunk_size', 65536) )
|
||||
|
||||
|
||||
@@ -59,13 +59,11 @@ class InteractiveEnvironmentRequest(object):
|
||||
|
||||
# This duplicates the logic in the proxy manager
|
||||
if self.attr.galaxy_config.dynamic_proxy_external_proxy:
|
||||
slash = '/'
|
||||
if self.attr.galaxy_config.cookie_path.endswith('/'):
|
||||
slash = ''
|
||||
self.attr.proxy_prefix = '%s%s%s' % (
|
||||
self.attr.galaxy_config.cookie_path,
|
||||
slash,
|
||||
self.attr.galaxy_config.dynamic_proxy_prefix)
|
||||
self.attr.proxy_prefix = '/'.join(('',
|
||||
self.attr.galaxy_config.cookie_path.strip('/'),
|
||||
self.attr.galaxy_config.dynamic_proxy_prefix.strip('/'),
|
||||
self.attr.viz_id,
|
||||
))
|
||||
else:
|
||||
self.attr.proxy_prefix = ''
|
||||
|
||||
@@ -218,7 +216,7 @@ class InteractiveEnvironmentRequest(object):
|
||||
log.error( "%s\n%s" % (stdout, stderr) )
|
||||
return None
|
||||
else:
|
||||
container_id = stdout
|
||||
container_id = stdout.strip()
|
||||
log.debug( "Container id: %s" % container_id)
|
||||
inspect_data = self.inspect_container(container_id)
|
||||
port_mappings = self.get_container_port_mapping(inspect_data)
|
||||
@@ -239,6 +237,8 @@ class InteractiveEnvironmentRequest(object):
|
||||
host=self.attr.docker_hostname,
|
||||
port=host_port,
|
||||
proxy_prefix=self.attr.proxy_prefix,
|
||||
route_name=self.attr.viz_id,
|
||||
container_ids=[container_id],
|
||||
)
|
||||
# These variables then become available for use in templating URLs
|
||||
self.attr.proxy_url = self.attr.proxy_request[ 'proxy_url' ]
|
||||
|
||||
@@ -6,39 +6,63 @@ from .filelock import FileLock
|
||||
from galaxy.util import sockets
|
||||
from galaxy.util.lazy_process import LazyProcess, NoOpLazyProcess
|
||||
from galaxy.util import sqlite
|
||||
from galaxy.util import unique_id
|
||||
import urllib2
|
||||
import time
|
||||
|
||||
log = logging.getLogger( __name__ )
|
||||
|
||||
|
||||
DEFAULT_PROXY_TO_HOST = "localhost"
|
||||
SECURE_COOKIE = "galaxysession"
|
||||
# Randomly generate a password every launch
|
||||
|
||||
|
||||
class ProxyManager(object):
|
||||
|
||||
def __init__( self, config ):
|
||||
for option in ["manage_dynamic_proxy", "dynamic_proxy_bind_port",
|
||||
"dynamic_proxy_bind_ip", "dynamic_proxy_debug",
|
||||
"dynamic_proxy_external_proxy", "dynamic_proxy_prefix"]:
|
||||
"dynamic_proxy_bind_ip", "dynamic_proxy_debug",
|
||||
"dynamic_proxy_external_proxy", "dynamic_proxy_prefix",
|
||||
"proxy_session_map",
|
||||
"dynamic_proxy", "cookie_path",
|
||||
"dynamic_proxy_golang_noaccess",
|
||||
"dynamic_proxy_golang_clean_interval",
|
||||
"dynamic_proxy_golang_docker_address",
|
||||
"dynamic_proxy_golang_api_key"]:
|
||||
|
||||
setattr( self, option, getattr( config, option ) )
|
||||
self.launch_by = "node" # TODO: Support docker
|
||||
|
||||
if self.manage_dynamic_proxy:
|
||||
self.lazy_process = self.__setup_lazy_process( config )
|
||||
else:
|
||||
self.lazy_process = NoOpLazyProcess()
|
||||
|
||||
if self.dynamic_proxy_golang_api_key is None:
|
||||
self.dynamic_proxy_golang_api_key = unique_id()
|
||||
|
||||
self.proxy_ipc = proxy_ipc(config)
|
||||
|
||||
def shutdown( self ):
|
||||
self.lazy_process.shutdown()
|
||||
|
||||
def setup_proxy( self, trans, host=DEFAULT_PROXY_TO_HOST, port=None, proxy_prefix="" ):
|
||||
def setup_proxy( self, trans, host=DEFAULT_PROXY_TO_HOST, port=None, proxy_prefix="", route_name="", container_ids=None ):
|
||||
if self.manage_dynamic_proxy:
|
||||
log.info("Attempting to start dynamic proxy process")
|
||||
log.debug("Cmd: " + ' '.join(self.lazy_process.command_and_args))
|
||||
self.lazy_process.start_process()
|
||||
|
||||
if container_ids is None:
|
||||
container_ids = []
|
||||
|
||||
authentication = AuthenticationToken(trans)
|
||||
proxy_requests = ProxyRequests(host=host, port=port)
|
||||
self.proxy_ipc.handle_requests(authentication, proxy_requests)
|
||||
self.proxy_ipc.handle_requests(
|
||||
authentication,
|
||||
proxy_requests,
|
||||
'/%s' % route_name,
|
||||
container_ids
|
||||
)
|
||||
# TODO: These shouldn't need to be request.host and request.scheme -
|
||||
# though they are reasonable defaults.
|
||||
host = trans.request.host
|
||||
@@ -56,13 +80,17 @@ class ProxyManager(object):
|
||||
}
|
||||
|
||||
def __setup_lazy_process( self, config ):
|
||||
launcher = proxy_launcher(self)
|
||||
launcher = self.proxy_launcher()
|
||||
command = launcher.launch_proxy_command(config)
|
||||
return LazyProcess(command)
|
||||
|
||||
|
||||
def proxy_launcher(config):
|
||||
return NodeProxyLauncher()
|
||||
def proxy_launcher(self):
|
||||
if self.dynamic_proxy == "node":
|
||||
return NodeProxyLauncher()
|
||||
elif self.dynamic_proxy == "golang":
|
||||
return GolangProxyLauncher()
|
||||
else:
|
||||
raise Exception("Unknown proxy type")
|
||||
|
||||
|
||||
class ProxyLauncher(object):
|
||||
@@ -70,7 +98,6 @@ class ProxyLauncher(object):
|
||||
def launch_proxy_command(self, config):
|
||||
raise NotImplementedError()
|
||||
|
||||
|
||||
class NodeProxyLauncher(object):
|
||||
|
||||
def launch_proxy_command(self, config):
|
||||
@@ -87,6 +114,30 @@ class NodeProxyLauncher(object):
|
||||
command = [ path_to_application ] + args
|
||||
return command
|
||||
|
||||
class GolangProxyLauncher(object):
|
||||
|
||||
def launch_proxy_command(self, config):
|
||||
args = [
|
||||
"gxproxy", # Must be on path. TODO: wheel?
|
||||
"--listenAddr", '%s:%d' % (
|
||||
config.dynamic_proxy_bind_ip,
|
||||
config.dynamic_proxy_bind_port,
|
||||
),
|
||||
"--listenPath", "/".join((
|
||||
config.cookie_path,
|
||||
config.dynamic_proxy_prefix
|
||||
)),
|
||||
"--cookieName", "galaxysession",
|
||||
"--storage", config.proxy_session_map.replace('.sqlite', '.xml'), # just in case.
|
||||
"--apiKey", config.dynamic_proxy_golang_api_key,
|
||||
"--noAccess", config.dynamic_proxy_golang_noaccess,
|
||||
"--cleanInterval", config.dynamic_proxy_golang_clean_interval,
|
||||
"--dockerAddr", config.dynamic_proxy_golang_docker_address,
|
||||
]
|
||||
if config.dynamic_proxy_debug:
|
||||
args.append("--verbose")
|
||||
return args
|
||||
|
||||
|
||||
class AuthenticationToken(object):
|
||||
|
||||
@@ -109,15 +160,18 @@ class ProxyRequests(object):
|
||||
|
||||
def proxy_ipc(config):
|
||||
proxy_session_map = config.proxy_session_map
|
||||
if proxy_session_map.endswith(".sqlite"):
|
||||
return SqliteProxyIpc(proxy_session_map)
|
||||
else:
|
||||
return JsonFileProxyIpc(proxy_session_map)
|
||||
if config.dynamic_proxy == "nodejs":
|
||||
if proxy_session_map.endswith(".sqlite"):
|
||||
return SqliteProxyIpc(proxy_session_map)
|
||||
else:
|
||||
return JsonFileProxyIpc(proxy_session_map)
|
||||
elif config.dynamic_proxy == "golang":
|
||||
return RestGolangProxyIpc(config)
|
||||
|
||||
|
||||
class ProxyIpc(object):
|
||||
|
||||
def handle_requests(self, cookie, host, port):
|
||||
def handle_requests(self, authentication, proxy_requests, route_name, container_ids):
|
||||
raise NotImplementedError()
|
||||
|
||||
|
||||
@@ -126,7 +180,7 @@ class JsonFileProxyIpc(object):
|
||||
def __init__(self, proxy_session_map):
|
||||
self.proxy_session_map = proxy_session_map
|
||||
|
||||
def handle_requests(self, authentication, proxy_requests):
|
||||
def handle_requests(self, authentication, proxy_requests, route_name, container_ids):
|
||||
key = "%s:%s" % ( proxy_requests.host, proxy_requests.port )
|
||||
secure_id = authentication.cookie_value
|
||||
with FileLock( self.proxy_session_map ):
|
||||
@@ -150,7 +204,7 @@ class SqliteProxyIpc(object):
|
||||
def __init__(self, proxy_session_map):
|
||||
self.proxy_session_map = proxy_session_map
|
||||
|
||||
def handle_requests(self, authentication, proxy_requests):
|
||||
def handle_requests(self, authentication, proxy_requests, route_name, container_ids):
|
||||
key = "%s:%s" % ( proxy_requests.host, proxy_requests.port )
|
||||
secure_id = authentication.cookie_value
|
||||
with FileLock( self.proxy_session_map ):
|
||||
@@ -170,5 +224,38 @@ class SqliteProxyIpc(object):
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
# TODO: RESTful API driven proxy?
|
||||
class RestGolangProxyIpc(object):
|
||||
|
||||
def __init__(self, config):
|
||||
self.config = config
|
||||
self.api_url = 'http://127.0.0.1:%s/api?api_key=%s' % (self.config.dynamic_proxy_bind_port, self.config.dynamic_proxy_golang_api_key)
|
||||
|
||||
def handle_requests(self, authentication, proxy_requests, route_name, container_ids, sleep=1):
|
||||
"""Make a POST request to the GO proxy to register a route
|
||||
"""
|
||||
values = {
|
||||
'FrontendPath': route_name,
|
||||
'BackendAddr': "%s:%s" % ( proxy_requests.host, proxy_requests.port ),
|
||||
'AuthorizedCookie': authentication.cookie_value,
|
||||
'ContainerIds': container_ids,
|
||||
}
|
||||
|
||||
print values
|
||||
print self.api_url
|
||||
req = urllib2.Request(self.api_url)
|
||||
req.add_header('Content-Type', 'application/json')
|
||||
|
||||
# Sometimes it takes our poor little proxy a second or two to get
|
||||
# going, so if this fails, re-call ourselves with an increased timeout.
|
||||
try:
|
||||
urllib2.urlopen(req, json.dumps(values))
|
||||
except urllib2.URLError, err:
|
||||
log.debug(err)
|
||||
if sleep > 5:
|
||||
excp = "Could not contact proxy after %s seconds" % sum(range(sleep + 1))
|
||||
raise Exception(excp)
|
||||
time.sleep(sleep)
|
||||
self.handle_requests(authentication, proxy_requests, route_name, container_ids, sleep=sleep + 1)
|
||||
pass
|
||||
|
||||
# TODO: MQ diven proxy?
|
||||
|
||||
Reference in New Issue
Block a user