import asyncio
import json
import logging
import traceback
from aiohttp import web
from aiolocals import context_middleware_factory
from sidecar.error import UnknownServiceError, ServiceUnavailableError
from sidecar.middleware import logging_middleware_factory
from sidecar.proxy_handler import ProxyHandler
from sidecar.status_handler import StatusHandler
from sidecar.web_decorator import unhandled_exception_500
log = logging.getLogger(__name__)
[docs]class SidecarApplication(web.Application):
def __init__(self, statsd, **kwargs):
self.statsd = statsd
super(SidecarApplication, self).__init__(**kwargs)
[docs] def run(self, host='localhost', port=12000):
loop = asyncio.get_event_loop()
create_server_task = loop.create_server(self.make_handler(), host, port)
server_instance = loop.run_until_complete(create_server_task)
host, port = server_instance.sockets[0].getsockname()
log.info("Sidecar running on {host}:{port}".format(host=host, port=port))
try:
loop.run_forever()
except KeyboardInterrupt:
# Press Ctrl+C to stop
pass
finally:
server_instance.close()
[docs]class Api:
def __init__(self, statsd, service_discovery):
self.statsd = statsd
self.service_discovery = service_discovery
self.app = SidecarApplication(statsd, middlewares=[context_middleware_factory,
logging_middleware_factory])
proxy_handler = ProxyHandler(statsd, service_discovery)
self.app.router.add_route('GET', r'/proxy/{service_id:\S+?}/{path:\S*}', proxy_handler.proxy_request_handler)
self.app.router.add_route('POST', r'/proxy/{service_id:\S+?}/{path:\S*}', proxy_handler.proxy_request_handler)
self.app.router.add_route('PUT', r'/proxy/{service_id:\S+?}/{path:\S*}', proxy_handler.proxy_request_handler)
self.app.router.add_route('DELETE', r'/proxy/{service_id:\S+?}/{path:\S*}', proxy_handler.proxy_request_handler)
host_handler = HostsHandler(service_discovery, self.statsd)
self.app.router.add_route('GET', '/host', host_handler.get_random_host)
status_handler = StatusHandler(proxy_handler, service_discovery)
self.app.router.add_route('GET', '/status', status_handler.report_status)
health_check_handler = HealthCheckHandler()
self.app.router.add_route('GET', '/health-check', health_check_handler.report_health)
[docs]class HealthCheckHandler:
@unhandled_exception_500
@asyncio.coroutine
def report_health(self, request):
# TODO: could do more to check that we're working before reporting liveness
return web.Response(text='UP', content_type='text/plain')
[docs]class HostsHandler:
def __init__(self, service_discovery, statsd):
self.service_discovery = service_discovery
self.statsd = statsd
@unhandled_exception_500
@asyncio.coroutine
def get_random_host(self, request):
log.debug("Called /host")
service_id = request.GET.get('sid', None)
if service_id is None:
return web.Response(text="Service Id ('sid' parameter) not provided", status=400)
lb = self.service_discovery.service_load_balancer(service_id)
try:
scheme, netloc = lb()
except UnknownServiceError:
self.statsd.increment("hosts.unknownServiceError.count")
log.error("Config for the service {!r} was not found when looking up hosts".format(service_id))
return web.Response(text="The requested service {!r} does not exist".format(service_id), status=502)
except ServiceUnavailableError:
log.error("No hosts available to answer /host query for service {!r}".format(service_id))
self.statsd.increment("hosts.serviceUnavailableError.count")
self.statsd.increment("hosts.serviceUnavailableError.{}.count".format(service_id))
return web.Response(text="No hosts avialable for service {!r}".format(service_id), status=503)
else:
return web.Response(text=json.dumps(dict(scheme=scheme, netloc=netloc)),
content_type="application/json")