import errno import eventlet import logging import os import re import socket import ssl import time from eventlet import greenio as eventlet_greenio from eventlet import wsgi as eventlet_wsgi from synergy.exception import SynergyError from sys import exc_info from traceback import format_tb __author__ = "Lisa Zangrando" __email__ = "lisa.zangrando[AT]pd.infn.it" __copyright__ = """Copyright (c) 2015 INFN - INDIGO-DataCloud All Rights Reserved Licensed under the Apache License, Version 2.0; you may not use this file except in compliance with the License. You may obtain a copy of the License at: http://www.apache.org/licenses/LICENSE-2.0 Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.""" LOG = logging.getLogger(__name__) class Dispatcher(object): """Dispatcher The main WSGI application. Dispatch the current request to the functions from above and store the regular expression captures in the WSGI environment as `myapp.url_args` so that the functions from above can access the url placeholders. If nothing matches call the `not_found` function. """ def __init__(self): self.actions = {} def register(self, action, callback): self.actions[action] = callback def unregister(self, action): del self.actions[action] def __call__(self, environ, start_response): """Call the application can catch exceptions.""" appiter = None # just call the application and send the output back unchanged # but catch exceptions path = environ.get('PATH_INFO', '').lstrip('/') application = None for regex, callback in self.actions.items(): match = re.search(regex, path) if match is not None: environ['myapp.url_args'] = match.groups() application = callback break if application is not None: try: self.appiter = callback(environ, start_response) for item in self.appiter: yield item # if an exception occours we get the exception information and # prepare a traceback we can render except Exception: e_type, e_value, tb = exc_info() traceback = ['Traceback (most recent call last):'] traceback += format_tb(tb) traceback.append('%s: %s' % (e_type.__name__, e_value)) # we might have not a stated response by now. # Try to start one with the status # code 500 or ignore an raised exception if the application # already started one. try: start_response("500 INTERNAL SERVER ERROR", [('Content-Type', 'text/plain')]) except Exception: pass yield '\n'.join(traceback) # wsgi applications might have a close function. # If it exists it *must* be called. if hasattr(appiter, 'close'): self.appiter.close() else: """Called if no applations matches.""" try: start_response("404 NOT FOUND", [('Content-Type', 'text/plain')]) except Exception: pass yield "Not Found" class WSGILog(object): """A thin wrapper that responds to `write` and logs.""" def __init__(self, logger, level=20): self.logger = logger self.level = level def write(self, msg): self.logger.log(self.level, msg.rstrip()) class Server(object): """Server class to manage multiple WSGI sockets and applications.""" def __init__(self, name, host_name, host_port=8051, threads=1000, application=None, use_ssl=False, ssl_ca_file=None, ssl_cert_file=None, ssl_key_file=None, max_header_line=16384, retry_until_window=30, tcp_keepidle=600, backlog=4096): """Parameters name: the server's name host_name: the host's name host_port: application: backlog: number of backlog requests to configure the socket with tcp_keepidle: sets the value of TCP_KEEPIDLE in seconds for each server socket. Not supported on OS X retry_until_window: number of seconds to keep retrying to listen max_header_line: max header line to accommodate large tokens use_ssl: enable SSL on the API server ssl_ca_file: CA certificate file to use to verify connecting clients ssl_cert_file: the certificate file ssl_key_file: the private key file """ # Raise the default from 8192 to accommodate large tokens eventlet_wsgi.MAX_HEADER_LINE = max_header_line self.name = name self.host_name = host_name self.host_port = host_port self.application = application self.threads = threads self.socket = None self.use_ssl = use_ssl self.tcp_keepidle = tcp_keepidle self.backlog = backlog self.retry_until_window = retry_until_window self.running = False self.dispatcher = Dispatcher() if not application: self.application = self.dispatcher if use_ssl: if not os.path.exists(ssl_cert_file): raise RuntimeError("Unable to find ssl_cert_file: %s" % ssl_cert_file) if not os.path.exists(ssl_key_file): raise RuntimeError("Unable to find ssl_key_file : %s" % ssl_key_file) # ssl_ca_file is optional if ssl_ca_file and not os.path.exists(ssl_ca_file): raise RuntimeError("Unable to find ssl_ca_file: %s" % ssl_ca_file) self.ssl_kwargs = { 'server_side': True, 'certfile': ssl_cert_file, 'keyfile': ssl_key_file, 'cert_reqs': ssl.CERT_NONE, } if ssl_ca_file: self.ssl_kwargs['ca_certs'] = ssl_ca_file self.ssl_kwargs['cert_reqs'] = ssl.CERT_REQUIRED def register(self, action, callback): self.dispatcher.register(action, callback) def unregister(self, action): self.dispatcher.unregister(action) def start(self): """Run a WSGI server with the given application. :param application: The application to be run in the WSGI server :param port: Port to bind to if none is specified in conf """ pgid = os.getpid() try: # NOTE(flaper87): Make sure this process # runs in its own process group. os.setpgid(pgid, pgid) except OSError: pgid = 0 try: info = socket.getaddrinfo(self.host_name, self.host_port, socket.AF_UNSPEC, socket.SOCK_STREAM)[0] family = info[0] bind_addr = info[-1] except Exception as ex: raise SynergyError("Unable to listen on %s:%s: %s" % (self.host_name, self.host_port, ex)) retry_until = time.time() + self.retry_until_window exception = None while not self.socket and time.time() < retry_until: try: self.socket = eventlet.listen(bind_addr, backlog=self.backlog, family=family) if self.use_ssl: self.socket = ssl.wrap_socket(self.socket, **self.ssl_kwargs) if self.use_ssl: ssl.wrap_socket(self.sock, **self.ssl_kwarg) except socket.error as ex: exception = ex LOG.error("Unable to listen on %s:%s: %s" % (self.host_name, self.host_port, ex)) if ex.errno == errno.EADDRINUSE: retry_until = 0 eventlet.sleep(0.1) break if exception is not None: raise exception if not self.socket: raise RuntimeError("Could not bind to %s:%s after trying for %d s" % (self.host_name, self.host_port, self.retry_until_window)) self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) # sockets can hang around forever without keepalive self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1) # This option isn't available in the OS X version of eventlet if hasattr(socket, 'TCP_KEEPIDLE'): self.socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPIDLE, self.tcp_keepidle) os.umask(0o27) # ensure files are created with the correct privileges self.pool = eventlet.GreenPool(self.threads) self.pool.spawn_n(self._single_run, self.application, self.socket) self.running = True def isRunning(self): return self.running def stop(self): LOG.info("shutting down: requests left: %s", self.pool.running()) self.running = False self.pool.resize(0) # self.pool.waitall() if self.socket: eventlet_greenio.shutdown_safe(self.socket) self.socket.close() self.running = False def wait(self): """Wait until all servers have completed running""" try: self.pool.waitall() except KeyboardInterrupt: pass def _single_run(self, application, sock): """Start a WSGI server in a new green thread.""" LOG.info("Starting single process server") eventlet_wsgi.server(sock, application, custom_pool=self.pool, log=WSGILog(LOG), debug=False)