Blame ev-server/pagure_stream_server.py

Pierre-Yves Chibon 1dd40f
#!/usr/bin/env python
Pierre-Yves Chibon 1dd40f
Pierre-Yves Chibon 1dd40f
"""
Pierre-Yves Chibon 0d7c61
 (c) 2015 - Copyright Red Hat Inc
Pierre-Yves Chibon 1dd40f
Pierre-Yves Chibon 1dd40f
 Authors:
Pierre-Yves Chibon 1dd40f
   Pierre-Yves Chibon <pingou@pingoured.fr></pingou@pingoured.fr>
Pierre-Yves Chibon 1dd40f
Pierre-Yves Chibon 1dd40f
Pierre-Yves Chibon 1dd40f
Streaming server for pagure's eventsource feature
Pierre-Yves Chibon 1dd40f
This server takes messages sent to redis and publish them at the specified
Pierre-Yves Chibon 1dd40f
endpoint
Pierre-Yves Chibon 1dd40f
Pierre-Yves Chibon 1dd40f
To test, run this script and in another terminal
Pierre-Yves Chibon 1dd40f
nc localhost 8080
Pierre-Yves Chibon 1dd40f
  HELLO
Pierre-Yves Chibon 1dd40f
Pierre-Yves Chibon 1dd40f
  GET /test/issue/26?foo=bar HTTP/1.1
Pierre-Yves Chibon 1dd40f
Pierre-Yves Chibon 1dd40f
"""
Pierre-Yves Chibon 1dd40f
Pierre-Yves Chibon 1dd40f
import datetime
Pierre-Yves Chibon 1dd40f
import logging
Pierre-Yves Chibon 1dd40f
import os
Pierre-Yves Chibon 1dd40f
import urlparse
Pierre-Yves Chibon 1dd40f
Pierre-Yves Chibon 1dd40f
import trollius
Pierre-Yves Chibon 1dd40f
import trollius_redis
Pierre-Yves Chibon 1dd40f
Pierre-Yves Chibon 1dd40f
log = logging.getLogger(__name__)
Pierre-Yves Chibon 1dd40f
Pierre-Yves Chibon 1dd40f
Pierre-Yves Chibon 1dd40f
if 'PAGURE_CONFIG' not in os.environ \
Pierre-Yves Chibon 1dd40f
        and os.path.exists('/etc/pagure/pagure.cfg'):
Pierre-Yves Chibon 1dd40f
    print 'Using configuration file `/etc/pagure/pagure.cfg`'
Pierre-Yves Chibon 1dd40f
    os.environ['PAGURE_CONFIG'] = '/etc/pagure/pagure.cfg'
Pierre-Yves Chibon 1dd40f
Pierre-Yves Chibon 1dd40f
Pierre-Yves Chibon 1dd40f
import pagure
Pierre-Yves Chibon 1dd40f
import pagure.lib
Pierre-Yves Chibon 116ab2
from pagure.exceptions import PagureEvException
Pierre-Yves Chibon 1dd40f
Pierre-Yves Chibon 39bf6b
SERVER = None
Pierre-Yves Chibon 1dd40f
Adam Williamson 01c1a4
Adam Williamson 01c1a4
def _get_issue(repo, objid):
Adam Williamson 01c1a4
    """Get a Ticket (issue) instance for a given repo (Project) and
Adam Williamson 01c1a4
    objid (issue number).
Pierre-Yves Chibon 116ab2
    """
Adam Williamson 01c1a4
    issue = None
Adam Williamson 01c1a4
    if not repo.settings.get('issue_tracker', True):
Adam Williamson 01c1a4
        raise PagureEvException("No issue tracker found for this project")
Pierre-Yves Chibon 116ab2
Adam Williamson 01c1a4
    issue = pagure.lib.search_issues(
Adam Williamson 01c1a4
        pagure.SESSION, repo, issueid=objid)
Pierre-Yves Chibon 116ab2
Adam Williamson 01c1a4
    if issue is None or issue.project != repo:
Adam Williamson 01c1a4
        raise PagureEvException("Issue '%s' not found" % objid)
Pierre-Yves Chibon 116ab2
Adam Williamson 01c1a4
    if issue.private:
Adam Williamson 01c1a4
        # TODO: find a way to do auth
Adam Williamson 01c1a4
        raise PagureEvException(
Adam Williamson 01c1a4
            "This issue is private and you are not allowed to view it")
Pierre-Yves Chibon 116ab2
Adam Williamson 01c1a4
    return issue
Pierre-Yves Chibon 116ab2
Pierre-Yves Chibon 116ab2
Adam Williamson 01c1a4
def _get_pull_request(repo, objid):
Adam Williamson 01c1a4
    """Get a PullRequest instance for a given repo (Project) and objid
Adam Williamson 01c1a4
    (request number).
Adam Williamson 01c1a4
    """
Adam Williamson 01c1a4
    if not repo.settings.get('pull_requests', True):
Adam Williamson 01c1a4
        raise PagureEvException(
Adam Williamson 01c1a4
            "No pull-request tracker found for this project")
Adam Williamson 01c1a4
Adam Williamson 01c1a4
    request = pagure.lib.search_pull_requests(
Adam Williamson 01c1a4
        pagure.SESSION, project_id=repo.id, requestid=objid)
Adam Williamson 01c1a4
Adam Williamson 01c1a4
    if request is None or request.project != repo:
Adam Williamson 01c1a4
        raise PagureEvException("Pull-Request '%s' not found" % objid)
Adam Williamson 01c1a4
Adam Williamson 01c1a4
    return request
Adam Williamson 01c1a4
Adam Williamson 01c1a4
Adam Williamson 01c1a4
# Dict representing known object types that we handle requests for,
Adam Williamson 01c1a4
# and the bound functions for getting an object instance from the
Adam Williamson 01c1a4
# parsed path data. Has to come after the functions it binds
Adam Williamson 01c1a4
OBJECTS = {
Adam Williamson 01c1a4
    'issue': _get_issue,
Adam Williamson 01c1a4
    'pull-request': _get_pull_request
Adam Williamson 01c1a4
}
Adam Williamson 01c1a4
Adam Williamson 01c1a4
Adam Williamson 01c1a4
def _parse_path(path):
Adam Williamson 01c1a4
    """Get the repo name, object type, object ID, and (if present)
Adam Williamson 01c1a4
    username and/or namespace from a URL path component. Will only
Adam Williamson 01c1a4
    handle the known object types from the OBJECTS dict. Assumes:
Adam Williamson 01c1a4
    * Project name comes immediately before object type
Adam Williamson 01c1a4
    * Object ID comes immediately after object type
Adam Williamson 01c1a4
    * If a fork, path starts with /fork/(username)
Adam Williamson 01c1a4
    * Namespace, if present, comes after fork username (if present) or at start
Adam Williamson 01c1a4
    * No other components come before the project name
Adam Williamson 01c1a4
    * None of the parsed items can contain a /
Adam Williamson 01c1a4
    """
Adam Williamson 01c1a4
    username = None
Adam Williamson 01c1a4
    namespace = None
Adam Williamson 01c1a4
    # path always starts with / so split and throw away first item
Adam Williamson 01c1a4
    items = path.split('/')[1:]
Adam Williamson 01c1a4
    # find the *last* match for any object type
Adam Williamson 01c1a4
    try:
Adam Williamson 01c1a4
        objtype = [item for item in items if item in OBJECTS][-1]
Adam Williamson 01c1a4
    except IndexError:
Adam Williamson 01c1a4
        raise PagureEvException("No known object type found in path: %s" % path)
Adam Williamson 01c1a4
    try:
Adam Williamson 01c1a4
        # objid is the item after objtype, we need all items up to it
Adam Williamson 01c1a4
        items = items[:items.index(objtype) + 2]
Adam Williamson 01c1a4
        # now strip the repo, objtype and objid off the end
Adam Williamson 01c1a4
        (repo, objtype, objid) = items[-3:]
Adam Williamson 01c1a4
        items = items[:-3]
Adam Williamson 01c1a4
    except (IndexError, ValueError):
Adam Williamson 01c1a4
        raise PagureEvException("No project or object ID found in path: %s" % path)
Adam Williamson 01c1a4
    # now check for a fork
Adam Williamson 01c1a4
    if items and items[0] == 'fork':
Adam Williamson 01c1a4
        try:
Adam Williamson 01c1a4
            # get the username and strip it and 'fork'
Adam Williamson 01c1a4
            username = items[1]
Adam Williamson 01c1a4
            items = items[2:]
Adam Williamson 01c1a4
        except IndexError:
Adam Williamson 01c1a4
            raise PagureEvException("Path starts with /fork but no user found! Path: %s" % path)
Adam Williamson 01c1a4
    # if we still have an item left, it must be the namespace
Adam Williamson 01c1a4
    if items:
Adam Williamson 01c1a4
        namespace = items.pop(0)
Adam Williamson 01c1a4
    # if we have any items left at this point, we've no idea
Adam Williamson 01c1a4
    if items:
Adam Williamson 01c1a4
        raise PagureEvException("More path components than expected! Path: %s" % path)
Adam Williamson 01c1a4
Adam Williamson 01c1a4
    return (username, namespace, repo, objtype, objid)
Pierre-Yves Chibon 116ab2
Pierre-Yves Chibon 116ab2
Adam Williamson 01c1a4
def get_obj_from_path(path):
Adam Williamson 01c1a4
    """ Return the Ticket or Request object based on the path provided.
Adam Williamson 01c1a4
    """
Adam Williamson 01c1a4
    (username, namespace, reponame, objtype, objid) = _parse_path(path)
Adam Williamson 01c1a4
    repo = pagure.lib.get_project(pagure.SESSION, reponame, user=username, namespace=namespace)
Adam Williamson 01c1a4
    if repo is None:
Adam Williamson 01c1a4
        raise PagureEvException("Project '%s' not found" % reponame)
Pierre-Yves Chibon 116ab2
Adam Williamson 01c1a4
    # find the appropriate object getter function from OBJECTS
Adam Williamson 01c1a4
    try:
Adam Williamson 01c1a4
        getfunc = OBJECTS[objtype]
Adam Williamson 01c1a4
    except KeyError:
Adam Williamson 01c1a4
        raise PagureEvException("Invalid object provided: '%s'" % objtype)
Pierre-Yves Chibon e07ca8
Adam Williamson 01c1a4
    return getfunc(repo, objid)
Pierre-Yves Chibon 116ab2
Pierre-Yves Chibon 116ab2
Pierre-Yves Chibon 1dd40f
@trollius.coroutine
Pierre-Yves Chibon 1dd40f
def handle_client(client_reader, client_writer):
Pierre-Yves Chibon ac3518
    data = None
Pierre-Yves Chibon ac3518
    while True:
Pierre-Yves Chibon ac3518
        # give client a chance to respond, timeout after 10 seconds
Pierre-Yves Chibon ac3518
        line = yield trollius.From(trollius.wait_for(
Pierre-Yves Chibon ac3518
            client_reader.readline(),
Pierre-Yves Chibon ac3518
            timeout=10.0))
Pierre-Yves Chibon ac3518
        if not line.decode().strip():
Pierre-Yves Chibon ac3518
            break
Pierre-Yves Chibon ac3518
        line = line.decode().rstrip()
Pierre-Yves Chibon ac3518
        if data is None:
Pierre-Yves Chibon ac3518
            data = line
Pierre-Yves Chibon 1dd40f
Pierre-Yves Chibon 1dd40f
    if data is None:
Pierre-Yves Chibon 1dd40f
        log.warning("Expected ticket uid, received None")
Pierre-Yves Chibon 1dd40f
        return
Pierre-Yves Chibon 1dd40f
Pierre-Yves Chibon 1dd40f
    data = data.decode().rstrip().split()
Pierre-Yves Chibon 1dd40f
    log.info("Received %s", data)
Pierre-Yves Chibon 1dd40f
    if not data:
Pierre-Yves Chibon 1dd40f
        log.warning("No URL provided: %s" % data)
Pierre-Yves Chibon 1dd40f
        return
Pierre-Yves Chibon 1dd40f
Pierre-Yves Chibon 1dd40f
    if not '/' in data[1]:
Pierre-Yves Chibon 1dd40f
        log.warning("Invalid URL provided: %s" % data[1])
Pierre-Yves Chibon 1dd40f
        return
Pierre-Yves Chibon 1dd40f
Pierre-Yves Chibon 1dd40f
    url = urlparse.urlsplit(data[1])
Pierre-Yves Chibon 1dd40f
Pierre-Yves Chibon 3d3084
    try:
Pierre-Yves Chibon 3d3084
        obj = get_obj_from_path(url.path)
Pierre-Yves Chibon 3d3084
    except PagureEvException as err:
Pierre-Yves Chibon 3d3084
        log.warning(err.message)
Pierre-Yves Chibon 3d3084
        return
Pierre-Yves Chibon 3d3084
Pierre-Yves Chibon 734609
    origin = pagure.APP.config.get('APP_URL')
Pierre-Yves Chibon 734609
    if origin.endswith('/'):
Pierre-Yves Chibon 734609
        origin = origin[:-1]
Pierre-Yves Chibon 734609
Pierre-Yves Chibon 1dd40f
    client_writer.write((
Pierre-Yves Chibon 1dd40f
        "HTTP/1.0 200 OK\n"
Pierre-Yves Chibon 1dd40f
        "Content-Type: text/event-stream\n"
Pierre-Yves Chibon 1dd40f
        "Cache: nocache\n"
Pierre-Yves Chibon 1dd40f
        "Connection: keep-alive\n"
Pierre-Yves Chibon 734609
        "Access-Control-Allow-Origin: %s\n\n" % origin
Pierre-Yves Chibon 1dd40f
    ).encode())
Pierre-Yves Chibon 1dd40f
Pierre-Yves Chibon 1f4ebf
    connection = yield trollius.From(trollius_redis.Connection.create(
Pierre-Yves Chibon 1f4ebf
        host=pagure.APP.config['REDIS_HOST'],
Pierre-Yves Chibon 1f4ebf
        port=pagure.APP.config['REDIS_PORT'],
Pierre-Yves Chibon 1f4ebf
        db=pagure.APP.config['REDIS_DB']))
Pierre-Yves Chibon 1f4ebf
Pierre-Yves Chibon 116ab2
    try:
Pierre-Yves Chibon 1dd40f
Pierre-Yves Chibon 1dd40f
        # Create subscriber.
Pierre-Yves Chibon 1dd40f
        subscriber = yield trollius.From(connection.start_subscribe())
Pierre-Yves Chibon 1dd40f
Pierre-Yves Chibon 1dd40f
        # Subscribe to channel.
Pierre-Yves Chibon 8e41fb
        yield trollius.From(subscriber.subscribe(['pagure.%s' % obj.uid]))
Pierre-Yves Chibon 1dd40f
Pierre-Yves Chibon 1dd40f
        # Inside a while loop, wait for incoming events.
Pierre-Yves Chibon 1dd40f
        while True:
Pierre-Yves Chibon 1dd40f
            reply = yield trollius.From(subscriber.next_published())
Pierre-Yves Chibon 1dd40f
            #print(u'Received: ', repr(reply.value), u'on channel', reply.channel)
Pierre-Yves Chibon 1dd40f
            log.info(reply)
Pierre-Yves Chibon 1dd40f
            log.info("Sending %s", reply.value)
Pierre-Yves Chibon 1dd40f
            client_writer.write(('data: %s\n\n' % reply.value).encode())
Pierre-Yves Chibon 1dd40f
            yield trollius.From(client_writer.drain())
Pierre-Yves Chibon 1dd40f
Pierre-Yves Chibon 1f4ebf
    except trollius.ConnectionResetError as err:
Pierre-Yves Chibon e89a19
        log.exception("ERROR: ConnectionResetError in handle_client")
Pierre-Yves Chibon 1f4ebf
    except Exception as err:
Pierre-Yves Chibon e89a19
        log.exception("ERROR: Exception in handle_client")
Pierre-Yves Chibon 1dd40f
    finally:
Pierre-Yves Chibon 1dd40f
        # Wathever happens, close the connection.
Pierre-Yves Chibon 1dd40f
        connection.close()
Pierre-Yves Chibon 1dd40f
        client_writer.close()
Pierre-Yves Chibon 1dd40f
Pierre-Yves Chibon 1dd40f
Pierre-Yves Chibon 39bf6b
@trollius.coroutine
Pierre-Yves Chibon 39bf6b
def stats(client_reader, client_writer):
Pierre-Yves Chibon 39bf6b
Pierre-Yves Chibon 39bf6b
    try:
Pierre-Yves Chibon 39bf6b
        log.info('Clients: %s', SERVER.active_count)
Pierre-Yves Chibon 39bf6b
        client_writer.write((
Pierre-Yves Chibon 39bf6b
            "HTTP/1.0 200 OK\n"
Pierre-Yves Chibon 39bf6b
            "Cache: nocache\n\n"
Pierre-Yves Chibon 39bf6b
        ).encode())
Pierre-Yves Chibon 39bf6b
        client_writer.write(('data: %s\n\n' % SERVER.active_count).encode())
Pierre-Yves Chibon 39bf6b
        yield trollius.From(client_writer.drain())
Pierre-Yves Chibon 39bf6b
Pierre-Yves Chibon fa97f7
    except trollius.ConnectionResetError as err:
Pierre-Yves Chibon af9aab
        log.info(err)
Pierre-Yves Chibon 39bf6b
        pass
Pierre-Yves Chibon 39bf6b
    finally:
Pierre-Yves Chibon 39bf6b
        client_writer.close()
Pierre-Yves Chibon 39bf6b
    return
Pierre-Yves Chibon 39bf6b
Pierre-Yves Chibon 39bf6b
Pierre-Yves Chibon 1dd40f
def main():
Pierre-Yves Chibon 39bf6b
    global SERVER
Pierre-Yves Chibon 1dd40f
Pierre-Yves Chibon 1dd40f
    try:
Pierre-Yves Chibon 1dd40f
        loop = trollius.get_event_loop()
Pierre-Yves Chibon 1dd40f
        coro = trollius.start_server(
Pierre-Yves Chibon 8d3302
            handle_client,
Pierre-Yves Chibon 8d3302
            host=None,
Pierre-Yves Chibon 8d3302
            port=pagure.APP.config['EVENTSOURCE_PORT'],
Pierre-Yves Chibon 8d3302
            loop=loop)
Pierre-Yves Chibon 39bf6b
        SERVER = loop.run_until_complete(coro)
Pierre-Yves Chibon d8a38d
        log.info('Serving server at {}'.format(SERVER.sockets[0].getsockname()))
Pierre-Yves Chibon 39bf6b
        if pagure.APP.config.get('EV_STATS_PORT'):
Pierre-Yves Chibon 39bf6b
            stats_coro = trollius.start_server(
Pierre-Yves Chibon 39bf6b
                stats,
Pierre-Yves Chibon 39bf6b
                host=None,
Pierre-Yves Chibon 39bf6b
                port=pagure.APP.config.get('EV_STATS_PORT'),
Pierre-Yves Chibon 39bf6b
                loop=loop)
Pierre-Yves Chibon 39bf6b
            stats_server = loop.run_until_complete(stats_coro)
Pierre-Yves Chibon d8a38d
            log.info('Serving stats  at {}'.format(
Pierre-Yves Chibon d8a38d
                stats_server.sockets[0].getsockname()))
Pierre-Yves Chibon 1dd40f
        loop.run_forever()
Pierre-Yves Chibon 1dd40f
    except KeyboardInterrupt:
Pierre-Yves Chibon 1dd40f
        pass
Pierre-Yves Chibon 069455
    except trollius.ConnectionResetError as err:
Pierre-Yves Chibon e89a19
        log.exception("ERROR: ConnectionResetError in main")
Pierre-Yves Chibon 069455
    except Exception as err:
Pierre-Yves Chibon e89a19
        log.exception("ERROR: Exception in main")
Pierre-Yves Chibon 069455
    finally:
Pierre-Yves Chibon 069455
        # Close the server
Pierre-Yves Chibon 069455
        SERVER.close()
Pierre-Yves Chibon 069455
        if pagure.APP.config.get('EV_STATS_PORT'):
Pierre-Yves Chibon 069455
            stats_server.close()
Pierre-Yves Chibon 069455
        log.info("End Connection")
Pierre-Yves Chibon 069455
        loop.run_until_complete(SERVER.wait_closed())
Pierre-Yves Chibon 069455
        loop.close()
Pierre-Yves Chibon 069455
        log.info("End")
Pierre-Yves Chibon 1dd40f
Pierre-Yves Chibon 1dd40f
Pierre-Yves Chibon 1dd40f
if __name__ == '__main__':
Pierre-Yves Chibon 1dd40f
    log = logging.getLogger("")
Pierre-Yves Chibon 1dd40f
    formatter = logging.Formatter(
Pierre-Yves Chibon 1dd40f
        "%(asctime)s %(levelname)s [%(module)s:%(lineno)d] %(message)s")
Pierre-Yves Chibon 1dd40f
Pierre-Yves Chibon 1dd40f
    # setup console logging
Pierre-Yves Chibon 1dd40f
    log.setLevel(logging.DEBUG)
Pierre-Yves Chibon 1dd40f
    ch = logging.StreamHandler()
Pierre-Yves Chibon 1dd40f
    ch.setLevel(logging.DEBUG)
Pierre-Yves Chibon 1dd40f
Pierre-Yves Chibon 1dd40f
    aslog = logging.getLogger("asyncio")
Pierre-Yves Chibon 1dd40f
    aslog.setLevel(logging.DEBUG)
Pierre-Yves Chibon 1dd40f
Pierre-Yves Chibon 1dd40f
    ch.setFormatter(formatter)
Pierre-Yves Chibon 1dd40f
    log.addHandler(ch)
Pierre-Yves Chibon 1dd40f
    main()