Пишем простой buildserver на Python (Часть 1)

В этой статье я расскажу о том, как написать простой build server с использованием языка программирования Python.

Для начала хорошо было бы знать, что такое build server.
Это специальное ПО, предназначенное для обеспечения непрерывной интеграции.

Непрерывная интеграция (Continuous Integration, CI) - практика разработки ПО, при которой выполняются частые сборки проектов, что позволяет быстро выявить и решить различные проблемы.

Итак, билд сервер должен уметь следующее:

- Получить исходный код из репозитория.
- Собрать проект (установить сторонние библиотеки, используемые в нём, скомпилировать и т.п.).
- Выполнить тесты.
- Задеплоить готовый проект.

Так как я не собираюсь делать убийцу TeamCity или Travis CI, то остановимся на этом:

- Клонирование из git репозитория.
- Выполнение шагов, описанных в конфигурации проекта.
- Просмотр лога сборки.

Что нам понадобится:

- Python
- Buildout
- Flask
- Tornado
- psycopg2
- PyZMQ

Python, я думаю, в представлении не нуждается.

Buildout. Позволяет устанавливать сторонние библиотеки, не прибегая к virtualenv и не засоряя систему.
А ещё у него есть куча рецептов, которые позволяют вытворять всякие прикольные штуки
вроде генерации скриптов, конфигов из шаблонов, интеграции с различными библиотеками и т.п.
Можно написать какой-нибудь свой рецепт. Всё ограничено только полётом фантазии.

Flask -- это микро-фреймворк, предназначенный для написания веб-приложений.
Tornado -- неблокирующий веб-сервер и веб-фреймворк.
psycopg2 -- питоновские биндинги для PostgreSQL.
PyZMQ -- обеспечивает поддержку ZMQ.

ZMQ -- библиотека, позволяющая создать систему очереди сообщений.
Подробнее о ZMQ можно узнать например здесь http://habrahabr.ru/post/198578/

Подготовка проекта и установка зависимостей

Первым делом создаём директорий для проекта:

$ mkdir buildserver
$ cd buildserver

Создаём файл setup.py со следующим содержимым:

$ editor setup.py


Создаём файл buildout.cfg:

$ editor buildout.cfg


Создаём пакет buildserver:

$ mkdir src/buildserver -p
$ touch src/buildserver/__init__.py

Перед установкой зависимостей проекта, в систему придётся поставить некоторые пакеты,
которые требуются сборки этих зависимостей.
А именно:

python-dev
libzmq3-dev
postgresql-server-dev-9.3

Ну и сам постгрес надо будет поставить, если не стоит.

В убунте и прочих debian-based дистрибутивах достаточно выполнить:

$ sudo apt-get install python-dev libzmq3-dev postgresql-9.3 postgresql-server-dev-9.3

Ставим buildout и зависимости для проекта:

$ wget http://downloads.buildout.org/2/bootstrap.py
$ python bootstrap.py
$ bin/buildout

Настройка PostgreSQL

Устанавливаем пароль для пользователя postgres:

$ sudo passwd postgres

Эта команда запросит ваш пароль, новый пароль для postgres, и повтор нового пароля.

Логинимся:

$ su postgres

Создаём юзера:

$ createuser -sdrP kilte

Конечно имя пользователя может быть каким угодно, но для большего удобства лучше создать пользователя
с таким же именем, под каким вы залогинены. Это позволит не указывать имя пользователя каждый раз,
когда вы запускаете psql.

Выходим:
$ exit

Заходим в psql:

$ psql postgres

Создаём базу данных для нашего приложения:

create database buildserver owner kilte;

На этом настройку postgres можно считать оконченой.


Bower

Bower - это менеджер зависимостей для фронтенда.

http://bower.io/ - Оф. сайт
http://nano.sapegin.ru/all/bower - Инфа на русском.

Для того, чтобы установить bower, нужно поставить nodejs и npm.
Я недолюбливаю npm и всё, что с ним связано, потому что оно выкачивает десятки мегабайт непонятно чего.
Мы пойдём более простым путём.
Bower портирован на PHP и мы спокойно можем скачать один файл, и начать пользоваться.

$ wget http://bowerphp.org/bowerphp.phar

Можете теперь сделать с ним всё, что угодно.
У меня же для подобных штук в домашнем директории есть директорий bin, куда я складываю все бинарники.
~/bin прописан в $PATH, что позволяет мне запускать все файлы из ~/bin, не набирая абсолютный путь до исполняемого файла.

Кому-то может такое и не понравится, потому можно сделать вот что:

$ sudo mv bowerphp.phar /usr/local/bin/bowerphp

Конечно нужно не забыть сделать этот файл исполняемым:

$ sudo chmod +x /usr/local/bin/bowerphp

Теперь пробуем выполнить:

$ bowerphp

Создаём в корне проекта два файла .bowerrc и bower.json

В .bowerrc пишем:

Здесь мы определили путь к директорию, куда будут установлены зависимости, описанные в bower.json

В bower.json:

Ну здесь всё понятно, а что не понятно, смотрите доку на офф сайте.

Выполняем:

$ bowerphp install

После чего получим в корне проекта директорий web/vendor, в котором лежат описанные в конфиге зависимости.


Настройка Nginx

Ещё не стоит? Почему? А ну быстро ставим. Хе-хе.

$ sudo apt-get install nginx

Пишем конфиг для него:

$ sudo editor /etc/nginx/sites-available/buildserver.conf

Содержимое конфига

Содержимое конфига

[копировать] [скачать]

1. server {
2.     listen 80;
3.     server_name buildserver;
4.     root /home/kilte/projects/buildserver/web;
5.     index index.html;
6.  
7.     location / {
8.         try_files $uri $uri/ @app;
9.     }
10.  
11.     # REST API
12.    location @app {
13.          proxy_pass http://127.0.0.1:5000;
14.     }
15.  
16.     # проксируем WebSockets
17.    location /broadcast/ {
18.         proxy_pass http://127.0.0.1:8888;
19.         proxy_http_version 1.1;
20.         proxy_set_header Upgrade $http_upgrade;
21.         proxy_set_header Connection "upgrade";
22.         proxy_set_header Host buildserver;  
23.         proxy_read_timeout 950s;
24.     }
25.  
26. }

Путь к проекту не забудьте заменить.

Врубаем хост:

$ sudo ln -s /etc/nginx/sites-available/buildserver.conf /etc/nginx/sites-enabled/
$ sudo service nginx restart

В /etc/hosts: 127.0.0.1 buildserver

Backend

Бэкэнд будет состоять из нескольких частей:

Worker -- отвечает за сборку проектов.
Web -- REST API, написанное на фреймворке Flask
Broadcast -- приложение, написанное на Tornado, которое позволит отображать ход сборки в режиме реального времени.

Все они будут связаны между собой с помощью ZeroMQ.
Из веб приложения отсылается команда воркеру на сборку проекта.
При просмотре билда tornado будет считывать лог и отправлять его клиенту через WebSockets.
По окончанию сборки воркер сообщает об этом tornado, а то в свою очередь отправляет сообщение клиенту.

Для начала давайте создадим таблицы в БД.

$ psql buildserver

Выполняем:

REFERENCES "projects" ("id") ON DELETE RESTRICT означает, что поле ссылается на поле id в таблице projects.

Если мы попытаемся создать билд, указав project_id, который отсутствует в таблице projects, то у нас ничего не получится.
При удалении проекта, нужно будет удалить сначала все сборки.
В ином случае postgres просто пошлёт нас куда подальше.
Подробности здесь: http://postgresql.ru.net/manua...l#DDL-CONSTRAINTS-FK

Большинство пакетов для работы с базой данных реализуют Database API Specification 2.0
https://www.python.org/dev/peps/pep-0249/

Вся работа сводится к созданию подключения, получению курсора, выполнению запроса и коммита транзакции.



Создаём файл src/buildserver/app/repositories.py
В нём будут располагаться классы, необходимые для работы с БД.

repositories.py

repositories.py

[копировать] [скачать]

1. from psycopg2.extras import RealDictCursor
2. from time import time
3.  
4.  
5. class Base(object):
6.     __tablename__ = None
7.     __primary_key__ = None
8.  
9.     def __init__(self, connection):
10.         if self.__tablename__ is None:
11.             raise ValueError('Table name is not defined')
12.         self.connection = connection
13.  
14.     def _cursor(self):
15.         return self.connection.cursor(cursor_factory=RealDictCursor)
16.  
17.     def _insert(self, columns, values):
18.         # Осуществляет INSERT запрос
19.         cursor = self._cursor()
20.         placeholders = ('%s, ' * len(columns)).strip(', ')
21.         columns = '", "'.join(columns)
22.         columns = '"{}"'.format(columns)
23.         statement = 'INSERT INTO {table} ({cols}) VALUES({places})'.format(
24.             table=self.__tablename__,
25.             cols=columns,
26.             places=placeholders
27.         )
28.         cursor.execute(statement, values)
29.         self.connection.commit()
30.         # Получаем последний вставленный ID
31.         if self.__primary_key__:
32.             cursor.execute('SELECT "last_value" FROM "{}_{}_seq"'.format(self.__tablename__, self.__primary_key__))
33.             return cursor.fetchone()['last_value']
34.  
35.     def _update(self, columns, values, key, value):
36.         # Осуществляет UPDATE запрос
37.         self._cursor().execute(
38.             'UPDATE {table} SET "{values}" = %s WHERE {key} = %s'.format(
39.                 table=self.__tablename__,
40.                 values='" = %s, "'.join(columns),
41.                 key=key
42.             ),
43.             list(values) + [value]
44.         )
45.         self.connection.commit()
46.  
47.     def _delete(self, key, value):
48.         # Осуществляет DELETE запрос
49.         self._cursor().execute('DELETE FROM {} WHERE "{}" = %s'.format(self.__tablename__, key), [value])
50.         self.connection.commit()
51.  
52.     def _find(self, key, value):
53.         # Находит запись в таблице
54.         cursor = self._cursor()
55.         cursor.execute('SELECT * FROM {} WHERE "{}" = %s'.format(self.__tablename__, key), (value, ))
56.         return cursor.fetchone()
57.  
58.  
59. class Builds(Base):
60.     __tablename__ = 'builds'
61.     __primary_key__ = 'id'
62.  
63.     def all(self, pid):
64.         # Список всех сборок для проекта
65.         cursor = self._cursor()
66.         cursor.execute(
67.             'SELECT * FROM {} WHERE "project_id" = %s ORDER BY "start_date" DESC'.format(self.__tablename__),
68.             [pid]
69.         )
70.         return cursor.fetchall()
71.  
72.     def start(self, project_id):
73.         # Создаёт билд
74.         columns = ('project_id', 'start_date', 'finish_date', 'state')
75.         values = (project_id, int(time()), 0, 'running')
76.         return self._insert(columns, values)
77.  
78.     def finish(self, bid, state):
79.         # Отмечает билд завершённым
80.         if state not in ['success', 'failed']:
81.             raise ValueError('Unexpected state given')
82.         self._update(('finish_date', 'state'), (int(time()), state), 'id', bid)
83.  
84.     def remove(self, pid):
85.         # Удаляет все сборки проекта
86.         self._delete('project_id', pid)
87.  
88.     def find(self, bid):
89.         # Возвращает билд
90.         return self._find('id', bid)
91.  
92.  
93. class Projects(Base):
94.     __tablename__ = 'projects'
95.     __primary_key__ = 'id'
96.  
97.     def all(self):
98.         # Возвращает список всех проектов
99.         cursor = self._cursor()
100.         cursor.execute('SELECT * FROM "{}" ORDER BY "name" ASC'.format(self.__tablename__))
101.         return cursor.fetchall()
102.  
103.     def save(self, name, description, url, pid=None):
104.         # Создаёт/обновляет проект
105.         columns = ('name', 'description', 'url')
106.         values = (name, description, url)
107.         if pid is not None:
108.             self._update(columns, values, 'id', pid)
109.         else:
110.             pid = self._insert(columns, values)
111.         return pid
112.  
113.     def remove(self, pid):
114.         # Удаляет проект
115.         self._delete('id', pid)
116.  
117.     def find(self, pid):
118.         # Возвращает проект
119.         return self._find('id', pid)

Чтобы выполнять команды в шелле из питона, воспользуемся модулем subprocess.
Почитать о нём на русском языке можно здесь.

src/buildserver/app/cmd.py

src/buildserver/app/cmd.py

[копировать] [скачать]

1. import os
2. import subprocess
3.  
4.  
5. class Cmd(object):
6.     def __init__(self, cwd, logger):
7.         self.log_tpl = '[{cwd}] {msg}'
8.         self.cwd = cwd
9.         self.logger = logger
10.  
11.     def run(self, *args):
12.         process = subprocess.Popen(
13.             args,
14.             stdout=subprocess.PIPE,
15.             stderr=subprocess.PIPE,
16.             cwd=self.cwd,
17.             shell=False,
18.             env=self.env
19.         )
20.         self.info('Executing: {}'.format(' '.join(args)))
21.         out, err = process.communicate()
22.         if out:
23.             self.info(out)
24.         if err:
25.             self.warning(err)
26.         return process.returncode
27.  
28.     @property
29.     def env(self):
30.         return os.environ.copy()
31.  
32.     def log(self, msg, level):
33.         try:
34.             getattr(self.logger, level)(self.log_tpl.format(cwd=os.path.basename(self.cwd), msg=msg))
35.         except UnicodeDecodeError:
36.             pass
37.  
38.     def info(self, msg):
39.         self.log(msg, 'info')
40.  
41.     def warning(self, msg):
42.         self.log(msg, 'warning')

Если вы читали материал, приведённый по ссылке выше, то никаких вопросов возникнуть не должно.

Настройки приложения будут располагаться в src/buildserver/app/settings.py:


Создаём директории logs и builds в корне проекта.

Было бы неплохо иметь централизованный доступ к логам сборок.

src/buildserver/app/log.py

src/buildserver/app/log.py

[копировать] [скачать]

1. import os
2. import logging
3. import shutil
4.  
5. from .settings import LOGS_PATH
6.  
7.  
8. def get_logger(name, path=None):
9.     # Возвращает сконфигурированный логгер
10.     logger = logging.getLogger(name)
11.     formatter = logging.Formatter('%(levelname)s [%(name)s/%(asctime)s]: %(message)s')  # Формат сообщения
12.     if not path:
13.         path = os.path.join(LOGS_PATH, '{}.log'.format(name))
14.     handler = logging.FileHandler(path)  # Обработчик, который будет писать сообщения в файл
15.     logger.addHandler(handler)
16.     handler.setFormatter(formatter)
17.     logger.setLevel(logging.INFO)  # Задаём уровень логгирования
18.     return logger
19.  
20.  
21. def get_project(pid):
22.     # Возвращает путь к директорию с логами для конкретного проекта
23.     return os.path.join(LOGS_PATH, 'builds', str(pid))
24.  
25.  
26. def create_project(pid):
27.     # Создаёт директорий для логов проекта
28.     path = get_project(pid)
29.     if not os.path.exists(path):
30.         os.makedirs(path)
31.     return path
32.  
33.  
34. def remove_project(pid):
35.     # Удаляет директорий для логов проекта
36.     shutil.rmtree(get_project(pid), True)
37.  
38.  
39. def get_build(pid, bid, open_=False):
40.     # Считывает лог сборки. Если значение аргумента open_ является True, вернёт объект файла
41.     path = os.path.join(LOGS_PATH, 'builds', str(pid), '{}.log'.format(bid))
42.     if os.path.exists(path):
43.         if open_:
44.             return open(path)
45.         with open(path) as f:
46.             log = f.read()
47.     else:
48.         if open_:
49.             raise OSError('File {} does not exists'.format(path))
50.         log = 'Unable to find log for this build {}'.format(path)
51.     return log

Теперь, чтобы можно было импортировать только что созданные модули, необходимо сказать питону, что src/buildserver/app является пакетом.
Для этого просто создаём пустой файл с именем __init__.py в этом директории.

REST API

Читаем про REST https://ru.wikipedia.org/wiki/REST
Еще можно здесь почитать: http://eax.me/rest/
Ну и это: http://habrahabr.ru/post/181988/

API приложения будет выглядеть примерно следующим образом:

HTTP Метод | URL | Описание

GET /api/v1/projects - Получить список проектов
POST /api/v1/projects - Создать проект
GET /api/v1/projects/<pid> - Получить данные конктретного проекта
PUT /api/v1/projects/<pid> - Обновить проект
DELETE /api/v1/projects/<pid> - Удалить проект

POST /api/v1/projects/<pid>/build - Начать сборку проекта
GET /api/v1/projects/<pid>/builds/<bid> - Получить информацию о сборке

Переходим к реализации.

src/buildserver/web.py

src/buildserver/web.py

[копировать] [скачать]

1. import psycopg2
2. import zmq
3. from datetime import datetime
4. from functools import wraps
5. from werkzeug.exceptions import default_exceptions
6. from werkzeug.exceptions import HTTPException
7. from flask import Flask, g, jsonify, request, abort
8. from .app import log, settings
9. from .app.repositories import Builds, Projects
10.  
11.  
12. def make_app(import_name, **kwargs):
13.     def error_handler(ex): # Позволяет отдавать сообщения об ошибках в json формате.
14.         response = jsonify(message=str(ex))
15.         response.status_code = (ex.code if isinstance(ex, HTTPException) else 500)
16.         return response
17.  
18.     application = Flask(import_name, **kwargs)
19.  
20.     for code in default_exceptions.iterkeys():
21.         application.error_handler_spec[None][code] = error_handler
22.  
23.     return application
24.  
25.  
26. app = make_app(__name__)
27. app.debug = settings.DEBUG
28. ZMQContext = zmq.Context()
29.  
30.  
31. def run():
32.     app.run()
33.  
34.  
35. @app.before_request
36. def setup_request():
37.     # Соединяемся с базой и создаём экземпляры репозиториев при каждом запросе
38.     g.db = psycopg2.connect(settings.PG_DSN)
39.     g.projects = Projects(g.db)
40.     g.builds = Builds(g.db)
41.  
42.  
43. @app.teardown_request
44. def teardown_request(exception):
45.     # Закрываем соединение с базой
46.     db = getattr(g, 'db', None)
47.     if db is not None:
48.         db.close()
49.  
50.  
51. @app.route('/api/v1/projects', methods=['GET', 'POST'])
52. def projects():
53.     if request.method == 'GET':
54.         # Возвращаем список проектов
55.         return jsonify(items=g.projects.all())
56.     elif request.method == 'POST':
57.         # Создаём проект
58.         validate_project()
59.         return jsonify(id=g.projects.save(**request.json))
60.  
61.  
62. # Декоратор, позволяющий проверить наличие проекта в базе
63. def requires_project(func):
64.     @wraps(func)
65.     def wrapper(*args, **kwargs):
66.         if 'project' in kwargs:
67.             project = g.projects.find(kwargs['project'])
68.             if not project:
69.                 abort(404)
70.             kwargs['project'] = project
71.         # Вызываем обёрнутый контроллёр
72.         return func(*args, **kwargs)
73.     return wrapper
74.  
75.  
76. # Конвертирует временную метоку в человекочитаемую дату
77. def timestamp_to_date(timestamp, fmt='%Y-%m-%d / %H:%M:%S'):
78.     return datetime.fromtimestamp(timestamp).strftime(fmt)
79.  
80.  
81. # Форматирует сборку для отдачи её клиенту
82. def format_build(build_data):
83.     build_data['start_date'] = timestamp_to_date(build_data['start_date'])
84.     if build_data['finish_date']:
85.         build_data['finish_date'] = timestamp_to_date(build_data['finish_date'])
86.     else:
87.         build_data['finish_date'] = ''
88.     return build_data
89.  
90.  
91. @app.route('/api/v1/projects/<project>', methods=['GET', 'PUT', "DELETE"])
92. @requires_project
93. def manage_project(project):
94.     if request.method == 'GET':
95.         # Возвращает данные проекта
96.         return jsonify(project=project, builds=map(format_build, g.builds.all(project['id'])))
97.     elif request.method == 'PUT':
98.         # Обновляет данные проекта
99.         validate_project()
100.         g.projects.save(pid=project['id'], **request.json)
101.         return jsonify(id=project['id'])
102.     elif request.method == 'DELETE':
103.         # Удаляет проект
104.         g.builds.remove(project['id'])
105.         g.projects.remove(project['id'])
106.         log.remove_project(project['id'])
107.         return jsonify(id=project['id'])
108.  
109.  
110. def validate_project():
111.     # Валидация данных проекта для создания/обновления
112.     if not request.json:
113.         abort(400)
114.     keys = sorted(('name', 'description', 'url'))
115.     if sorted(request.json.keys()) != keys:
116.         abort(400)
117.     for key in keys:
118.         if not request.json[key]:
119.             abort(400)
120.  
121.  
122. @app.route('/api/v1/projects/<project>/build', methods=['POST'])
123. @requires_project
124. def builds(project):
125.     # Добавляет проект в очередь на сборку
126.     socket = ZMQContext.socket(getattr(zmq, 'PUSH'))
127.     socket.bind(settings.TASK_NEW_PUBLISHER)
128.     socket.send_json({'action': 'build', 'project_id': project['id']})
129.     return jsonify(msg='Build was added to queue')
130.  
131.  
132. @app.route('/api/v1/projects/<project>/builds/<bid>', methods=['GET'])
133. @requires_project
134. def get_build(project, bid):
135.     # Возвращает информацию о проекте и сборке
136.     build = g.builds.find(bid)
137.     if not build:
138.         abort(404)
139.     build = format_build(build)
140.     if build['state'] != 'running':
141.         build['log'] = log.get_build(project['id'], build['id'])
142.     return jsonify(project=project, build=build)

REST API готово. Переходим к воркеру.
Но для начала давайте немного поиграемся с pyzmq.
Создадим в корне проекта два файла producer.py и publisher.py

producer:


publisher:


Запускаем:

$ bin/py publisher.py
$ bin/py producer.py

На выходе получим:

Круто, не правда ли?
Удаляем publisher.py и producer.py, они нам больше не пригодятся.

Создаём воркер.

src/buildserver/worker.py

src/buildserver/worker.py

[копировать] [скачать]

1. import os
2. import psycopg2
3. import shutil
4. import traceback
5. import uuid
6. import zmq
7.  
8. from datetime import datetime
9. from .app import log, settings
10. from .app.cmd import Cmd
11. from .app.repositories import Builds, Projects
12.  
13.  
14. class Worker(object):
15.  
16.     def __init__(self, projects, builds, zmq_context, logger):
17.         self.projects = projects  # Репозиторий для работы с проектами
18.         self.builds = builds  # Репозиторий для работы со сборками
19.         self.logger = logger  # Логгер для воркера
20.  
21.         # ZMQ сокет, от которого будем узнавать, что нужно собрать проект
22.         self.listener = zmq_context.socket(getattr(zmq, 'PULL'))
23.         self.listener.connect(settings.TASK_NEW_PUBLISHER)
24.         self.logger.info('Connected to {}'.format(settings.TASK_NEW_PUBLISHER))
25.  
26.         # ZMQ сокет, который будет сообщать о том, что сборка проекта завершена
27.         self.publisher = zmq_context.socket(getattr(zmq, 'PUB'))
28.         self.publisher.bind(settings.TASK_COMPLETE_PUBLISHER)
29.         self.logger.info('Bind to {}'.format(settings.TASK_COMPLETE_PUBLISHER))
30.  
31.     def validate_message(self, message):
32.         # Валидация сообщения, пришедшего от self.listener
33.         if not isinstance(message, dict):
34.             raise ValueError('Wrong message format')
35.         if 'action' not in message or 'project_id' not in message:
36.             raise ValueError('Wrong message format')
37.         project = self.projects.find(message['project_id'])
38.         if not project:
39.             raise RuntimeError('Project {} does not exists'.format(message['project_id']))
40.         return message['action'], project
41.  
42.     def action_build(self, project):
43.         # Осуществляет сборку проекта
44.  
45.         # Создаём билд в базе
46.         build_id = self.builds.start(project['id'])
47.         # Получаем путь к логам проекта
48.         logs_path = log.create_project(project['id'])
49.         # Создаём логгер для текущей сборки
50.         l = log.get_logger('#{}-{}'.format(project['id'], build_id), os.path.join(logs_path, str(build_id) + '.log'))
51.         l.info('Started build for project {}'.format(build_id, project['id']))
52.  
53.         try:
54.             l.info('Cloning repository from {}...'.format(project['url']))
55.             # Создаём директорий, куда будет клонироваться проект
56.             directory = os.path.join(settings.BUILDS_PATH, str(uuid.uuid4()).replace('-', ''))
57.             os.makedirs(directory)
58.             try:
59.                 cmd = Cmd(directory, l)
60.                 # Клонируем проект
61.                 code = cmd.run('git', 'clone', '--recursive', '--quiet', project['url'], '.')
62.                 if code != 0:
63.                     raise RuntimeError('Cloning failed')
64.                 # Читаем конфиг
65.                 with open(os.path.join(directory, '.buildserver')) as config:
66.                     commands = config.read().strip().split('\n')
67.                 # Выполняем команды из конфига
68.                 for command in commands:
69.                     try:
70.                         cmd.run(*command.split(' '))
71.                     except Exception as exc:
72.                         raise RuntimeError('Exception has occurred while running command "{}": {}'.format(
73.                             command,
74.                             str(exc)
75.                         ))
76.             finally:
77.                 # Удаляем директорий, куда был склонирован проект
78.                 shutil.rmtree(directory)
79.             l.info('Build finished!')
80.             # Обновляем билд в базе
81.             self.builds.finish(build_id, 'success')
82.             # Отправляем сообщение по zmq о том, что сборка завершена успешно
83.             self.publisher.send_json({'build_id': build_id, 'state': 'success'})
84.         except Exception as exc:
85.             # Произошла какая-то ошибка, пишем её в лог.
86.             l.error('{}\n{}'.format(str(exc), traceback.format_exc()))
87.             l.error('Build failed!')
88.             # Обновляем билд в базе
89.             self.builds.finish(build_id, 'failed')
90.             # Отправляем сообщение по zmq о том, что сборка провалилась
91.             self.publisher.send_json({'build_id': build_id, 'state': 'failed'})
92.  
93. # Дальше всё и без комментариев должно быть понятно
94.  
95.     def run(self):
96.         while True:
97.             try:
98.                 msg = self.listener.recv_json()
99.                 self.logger.info('Received message: {}'.format(msg))
100.                 action, project = self.validate_message(msg)
101.                 getattr(self, 'action_{}'.format(action))(project)
102.             except Exception as e:
103.                 self.logger.error('{}:\n{}'.format(str(e), traceback.format_exc()))
104.  
105.  
106. def run():
107.     connection = psycopg2.connect(settings.PG_DSN)
108.     Worker(
109.         Projects(connection),
110.         Builds(connection),
111.         zmq.Context(),
112.         log.get_logger('worker-{}-{}'.format(
113.             datetime.now().strftime('%Y-%m-%d'),
114.             str(uuid.uuid4()).replace('-', '')[0:10])
115.         )
116.     ).run()

Теперь необходимо сделать отображение лога сборки в режиме реального времени.

src/buildserver/broadcast.py

src/buildserver/broadcast.py

[копировать] [скачать]

1. import json
2. import signal
3. import time
4. import tornado.ioloop
5. import tornado.httpserver
6. import uuid
7. import zmq
8.  
9. from psycopg2.pool import SimpleConnectionPool
10. from tornado.ioloop import PeriodicCallback
11. from tornado.web import Application as WebApplication
12. from tornado.websocket import WebSocketHandler as BaseWebSocketHandler
13. from zmq.eventloop import ioloop as zmq_ioloop
14. from zmq.eventloop.zmqstream import ZMQStream
15. from .app import log, settings
16. from .app.repositories import Builds
17.  
18.  
19. # Пул соединений с БД. Не знаю, на сколько это решение верное, но для примера вполне норм.
20. pool = SimpleConnectionPool(5, 20, settings.PG_DSN)
21.  
22.  
23. # Запускает HTTPServer и позволяет остановить его с помощью UNIX сигналов SIGTERM и SIGINT
24. # https://ru.wikipedia.org/wiki/%D0%A1%D0%B8%D0%B3%D0%BD%D0%B0%D0%BB%D1%8B_%28UNIX%29
25. class Application(object):
26.     def __init__(self, app, address, port):
27.         self.http_server = tornado.httpserver.HTTPServer(app)
28.         self.address = address
29.         self.port = port
30.  
31.     def sig_handler(*args):
32.         tornado.ioloop.IOLoop.instance().add_callback(args[0].shutdown)
33.  
34.     def shutdown(self):
35.         self.http_server.stop()
36.         io_loop = tornado.ioloop.IOLoop.instance()
37.         io_loop.add_timeout(time.time() + 2, io_loop.stop)
38.  
39.     def run(self):
40.         self.http_server.listen(self.port, self.address)
41.         signal.signal(signal.SIGTERM, self.sig_handler)
42.         signal.signal(signal.SIGINT, self.sig_handler)
43.         tornado.ioloop.IOLoop.instance().start()
44.  
45.  
46. # Обработчик WebSocket соединений
47. class WebSocketHandler(BaseWebSocketHandler):
48.     connections = {}  # Контейнер для соединений
49.  
50.     def __init__(self, *args, **kwargs):
51.         super(WebSocketHandler, self).__init__(*args, **kwargs)
52.         self.sess_id = str(uuid.uuid4())  # Идентификатор соединения
53.         self.db = pool.getconn()  # Получаем подключение к БД из пула
54.         self.builds = Builds(self.db)  # Репозиторий для работы со сборками
55.         self.connections[self.sess_id] = {'handler': self, 'watchers': {}}  # Добавляем текущее соединение в контейнер
56.  
57.     def on_message(self, message):
58.         # Валидация сообщения
59.         # Сообщение должно быть вида {'action': 'some_action', 'params': {}}
60.         # Опция params не является обязательной
61.         try:
62.             if not message:
63.                 raise ValueError('Empty message')
64.             message = json.loads(message)
65.             if 'action' not in message:
66.                 raise ValueError('Action is not specified')
67.             params = message.get('params', {})
68.             if not isinstance(params, dict):
69.                 raise ValueError('Wrong params')
70.         except ValueError as e:
71.             self.write_message({'error': str(e)})
72.             return
73.  
74.         if message['action'] == 'subscribe':
75.             # Подписываемся на сборку
76.             if 'build_id' not in params:
77.                 raise ValueError('Wrong params')
78.             build = self.builds.find(params['build_id'])
79.             if not build:
80.                 self.write_message({'error': 'Build #{} does not exists'.format(params['build_id'])})
81.                 return
82.             if build['state'] != 'running':
83.                 return
84.             # Добавляем наблюдателя, который будет слать лог клиенту
85.             self.connections[self.sess_id]['watchers'][build['id']] = Watcher(self, build)
86.         elif message['action'] == 'unsubscribe':
87.             # Отписываемся от всех сборок
88.             for build in self.connections[self.sess_id]['watchers'].keys():
89.                 self.connections[self.sess_id]['watchers'].pop(build).stop()
90.         else:
91.             self.write_message({'error': 'Unknown action'})
92.  
93.     def on_close(self):
94.         # Незабываем вернуть соединение в пул при закрытии WebSocket соединения
95.         pool.putconn(self.db)
96.         self.db = None
97.         self.builds = None
98.         del self.connections[self.sess_id]
99.  
100.     @classmethod
101.     def on_zmq_message(cls, msg):
102.         # Слушаем сообщения от воркера
103.         msg = json.loads(msg[0])
104.         for connection in cls.connections.values():
105.             for build in connection['watchers'].keys():
106.                 if build == msg['build_id']:
107.                     connection['watchers'].pop(build).stop()
108.                     connection['handler'].write_message({'action': 'build_finished', 'params': msg})
109.  
110.  
111. class Watcher(object):
112.     def __init__(self, handler, build):
113.         self.handler = handler  # WebSocket соединение
114.         self.log_file = log.get_build(build['project_id'], build['id'], True)  # Лог сборки
115.         self.timer = PeriodicCallback(self.callback, 500)  # Таймер, который выполняет коллбэк с интервалом в 500 миллисекунд
116.         self.timer.start()
117.  
118.     def stop(self):
119.         self.timer.stop()
120.         self.log_file.close()
121.  
122.     def callback(self):
123.         msg = self.log_file.read()
124.         if msg:
125.             # Шлём лог клиенту
126.             self.handler.write_message({'action': 'build_log', 'params': {'line': msg}})
127.  
128.  
129. def run():
130.     # Связываем tornado ioloop с zmq
131.     zmq_ioloop.install()
132.  
133.     # Подписываемся на сообщения от воркера
134.     context = zmq.Context()
135.     socket = context.socket(getattr(zmq, 'SUB'))
136.     socket.setsockopt(getattr(zmq, 'SUBSCRIBE'), '')
137.     socket.connect(settings.TASK_COMPLETE_PUBLISHER)
138.     stream = ZMQStream(socket)
139.     stream.on_recv(WebSocketHandler.on_zmq_message)
140.  
141.     app = Application(
142.         WebApplication([(r"/broadcast/", WebSocketHandler)]),
143.         settings.BROADCAST['address'],
144.         settings.BROADCAST['port']
145.     )
146.     app.run()

Ну вот, как-то так.
Остаётся сделать фронтенд и научиться запускать всё это дело.