Continious delivery

Ball Sort Puzzle это популярная мобильная игра на IOS/Android. Суть её заключается в перестановке шариков до тех пор, пока в колбах не будут шарики одного цвета. При этом шарик можно перетаскивать либо в пустую колбу, либо на такой же шарик.

Так случилось, что я в неё залип. Очнулся примерно через месяц, на 725 уровне. Он мне никак не давался насколько бы глубоко я не пытался продумать свою стратегию. В итоге с этим вопросом я вышел в интернет, и заодно выяснил несколько интересных особенностей головоломки.

Во-первых, игра бесконечна почти бесконечна. По крайней мере уже сейчас на YouTube есть прохождения всех уровней в плоть до 5350, а в телеграмме гуляют скриншоты 10к+ уровней. Вторая особенность, и вот это уже некрасиво, не у всех уровней есть решение.

Ну это ни в какие ворота против нас играет коварный ИИ. Нужно действовать соответственно!

Под катом мы:

• Придумаем алгоритм, решающий эту головоломку (Python)

• Научимся парсить скриншот игры, чтобы скармливать алгоритму задачки (OpenCV)

• Напишем телеграм бота, который будет принимать скриншоты и возвращать решения

• Выстроим CI/CD через GitHub Actions и задеплоим бота на Яндекс.Функции

Погнали!

Алгоритмическое решение задачи

Решать такую задачу было весьма занимательно. Поэтому предлагаю заинтересованному читателю попробовать решить её самостоятельно.

Я же в первую очередь решил побить проблему на сущности. Это сделает алгоритм чуть более элегантным, а так же поможет в будущем парсить скриншоты игры:

class Color:    def __init__(self, symbol, verbose_name, emoji):        self.symbol = symbol        self.verbose_name = verbose_name        self.emoji = emoji    def __repr__(self) -> str:        return f'Color({self})'    def __str__(self) -> str:        return self.emoji

Beta-редактор хабра ломается на рендеринге emoji :poop:

class Ball:    def __init__(self, color: Color):        self.color = color    def __eq__(self, other: 'Ball'):        return self.color is other.color    def __repr__(self):        return f'Ball({self.color.verbose_name})'    def __str__(self) -> str:        return str(self.color)

class Flask:    def __init__(self, column: List[Color], num: int, max_size: int):        self.num = num        self.balls = [Ball(color) for color in column]        self.max_size = max_size    @property    def is_full(self):        return len(self.balls) == self.max_size    @property    def is_empty(self) -> bool:        return not self.balls    def pop(self) -> Ball:        return self.balls.pop(-1)    def push(self, ball: Ball):        self.balls.append(ball)    def __iter__(self):        return iter(self.balls)    def __getitem__(self, item: int) -> Ball:        return self.balls[item]    def __len__(self) -> int:        return len(self.balls)    def __str__(self) -> str:        return str(self.balls)

class Move:    def __init__(self, i, j, i_color: Color):        self.i = i        self.j = j        self.emoji = i_color.emoji    def __eq__(self, other: 'Move') -> bool:        return (self.i, self.j) == (other.i, other.j)    def __repr__(self) -> str:        return f'Ball({self})'    def __str__(self) -> str:        return f'{self.i} -> {self.j}'

Для решения будем использовать метод поиска с возвратом (Backtracking).

Решение задачи методом поиска с возвратом сводится к последовательному расширению частичного решения. Если на очередном шаге такое расширение провести не удается, то возвращаются к более короткому частичному решению и продолжают поиск дальше. Данный алгоритм позволяет найти все решения поставленной задачи, если они существуют.

В случае с нашей игрой это метод применяется так: мы рекурсивно обходим все возможные перестановки шариков (move) до тех пор, пока

• Либо нас не выкинет наш критерий остановки решённый пазл

• Либо в нашем хранилище состояний (states) не будет всех возможных перестановок в таком случае решения нет

    def solve(self) -> bool:        if self.is_solved:            return True        for move in self.get_possible_moves():            new_state = self.commit_move(move)            if new_state in self.states:  # Cycle!                self.rollback_move(move)                continue            self.states.add(new_state)            if self.solve():                return True            self.rollback_move(move)        return False

Алгоритм достаточно прямолинейный и далеко не всегда выдаёт оптимальное решение. Тем не менее он справляется с решением большинства задачек из игры за 1 сек.

Проверим алгоритм на чём-нибудь попроще:

def test_3x3():    data_in = [        [color.RED, color.GREEN, color.RED],        [color.GREEN, color.RED, color.GREEN],        [],    ]    puzzle = BallSortPuzzle(data_in)    result = puzzle.solve()    assert result is True    play_moves(data_in, puzzle.moves)

Полная версия программы доступна на github.

Распознавание скриншотов игры

Мы будем работать с .jpg картинками двух видов

Каждый чётный раунд игры состоит из 11 колб и 36 шариков, а нечётный 14 колб и 48 шариков. Чётные и нечётные раунды отличаются расположением колб, но на счастье всё остальное у них одинаковое по 4 шарика в колбе, 2 колбы пустые, цвета используются одни и те же.

Первое, что хочется сделать это обрезать рабочую область, оставив только колбы с шариками. В противном случае наша программа за шарики может принимать элементы управления, фон или рекламу. Для этого отрежем по четверти сверху и снизу изображения.

class ImageParser:    def __init__(self, file_bytes: np.ndarray, debug=False):        self.image_orig = cv2.imdecode(file_bytes, cv2.IMREAD_COLOR)        self.image_cropped = self.get_cropped_image(self.image_orig)    @staticmethod    def get_cropped_image(image):        height, width, _ = image.shape        quarter = int(height / 4)        cropped_img = image[quarter : height - quarter]        return cropped_img

Теперь будем искать кружочки. В библиотеке OpenCV ровно для этих целей существует метод HoughCircles. Чтобы его использовать нужно перевести изображение в чёрно-белый вид, а также "эмпирически подобрать" параметры поиска. Чтобы найденные кружочки потом расфасовать по колбам, нормализуем и отсортируем их.

    @staticmethod    def normalize_circles(circles):        last_y = 0        for circle in circles:            if math.isclose(circle[1], last_y, abs_tol=3):                circle[1] = last_y            else:                last_y = circle[1]        return circles    def get_normalized_circles(self) -> List[Any]:        image_cropped_gray = cv2.cvtColor(self.image_cropped, cv2.COLOR_BGR2GRAY)        circles = cv2.HoughCircles(image_cropped_gray, cv2.HOUGH_GRADIENT, 2, 20, maxRadius=27)        if circles is None:            raise ImageParserError("No circles :shrug:")        circles = np.round(circles[0, :]).astype("int16")        ind = np.lexsort((circles[:, 0], circles[:, 1]))        circles = circles[ind]        circles = self.normalize_circles(circles)        ind = np.lexsort((circles[:, 0], circles[:, 1]))        circles = circles[ind]        return circles

Дальше будем определять цвет шарика.

Из-за того, что Telegram жмёт картинки мы не можем просто взять цвет центрального пикселя он может быть артефактом компрессии. Поэтому найдём доминирующий цвет, тот который в кружочке встречается чаще всего.

    @staticmethod    def get_dominant_color(circle) -> Color:        colors, count = np.unique(circle.reshape(-1, circle.shape[-1]), axis=0, return_counts=True)        dominant = colors[count.argmax()]        return dominant

Этот доминирующий цвет мы будем сравнивать с изначально заданными цветами, и искать ближайший. В данной задаче нам на помощь приходит Евклидово расстояние и теорема Пифагора: если представить цвет точкой в трёхмерном пространстве, то расстояние до любой другой точки этого пространства:

Посчитаем такое расстояние до каждого из изначально заданных цветов и найдём минимальное

RBG_TO_COLOR = {    (147, 42, 115): VIOLET,    (8, 74, 125): BROWN,    (229, 163, 85): L_BLUE,    (68, 140, 234): ORANGE,    (196, 46, 59): BLUE,    (51, 100, 18): GREEN,    (35, 43, 197): RED,    (87, 216, 241): YELLOW,    (125, 214, 97): L_GREEN,    (123, 94, 234): PINK,    (16, 150, 120): LIME,    (102, 100, 99): GRAY,}COLORS = np.array(list(RBG_TO_COLOR.keys()))def get_closest_color(color: np.ndarray) -> Color:    distances = np.sqrt(np.sum((COLORS - color) ** 2, axis=1))    index_of_smallest = np.where(distances == np.amin(distances))    smallest_distance = COLORS[index_of_smallest].flat    return RBG_TO_COLOR[tuple(smallest_distance)]  # type: ignore

Далее нам остаётся только распределить шарики по колбам. Итоговый class ImageParser доступен на github.

Преобразуем программу в Telegram Bot

Узнать про то, как сделать телеграм бота на Python можно сразу из нескольких статей на хабре. Я лишь опишу пару нюансов, с которыми столкнулся.

Так как наш бот хоститься на Яндекс.Функции триггером к его запуску будет запрос на заданный нами webhook.

Whenever there is an update for the bot, we will send an HTTPS POST request to the specified url, containing a JSON-serializedUpdate.

Если в сообщении есть массив photo, то можно увеличить вероятность распознавания шариков выбрав фотографию с максимальным весом:

if photos := message.get('photo'):    # here photos is an array with same photo of different sizes    # get one with the highest resolution    hd_photo = max(photos, key=lambda x: x['file_size'])

Чтобы скачать картинку, придётся сделать 2 запроса к Telegram API

# Получение данных о файле, нас интересует ключ ответа file_pathGET https://api.telegram.org/bot{BOT_TOKEN}/getFile?file_id={file_id}# Получение самого файлаGET https://api.telegram.org/file/bot{BOT_TOKEN}/{file_path}

В остальном же всё просто получаем картинку, скармливаем её парсеру и затем алгоритму-решателю.

def handler(event: Optional[dict], context: Optional[dict]):    body = json.loads(event['body'])  # type: ignore    print(body)    message = body['message']    chat_id = message['chat']['id']    if photos := message.get('photo'):        # here photos is an array with same photo of different sizes        hd_photo = max(photos, key=lambda x: x['file_size'])  # get one with the highest resolution        try:            file = telegram_client.download_file(hd_photo['file_id'])        except TelegramClientError:            text = "Cant download the image from TG :("        else:            file_bytes = np.asarray(bytearray(file.read()), dtype=np.uint8)            try:                image_parser = ImageParser(file_bytes)                colors = image_parser.to_colors()            except ImageParserError as exp:                text = f"Cant parse image: {exp}"            else:                puzzle = BallSortPuzzle(colors)  # type: ignore                solved = puzzle.solve()                if solved:                    text = get_telegram_repr(puzzle)                else:                    text = "This lvl don't have a solution"    else:        return {            'statusCode': 200,            'headers': {'Content-Type': 'application/json'},            'body': '',            'isBase64Encoded': False,        }    msg = {        'method': 'sendMessage',        'chat_id': chat_id,        'text': text,        'parse_mode': 'Markdown',        'reply_to_message_id': message['message_id'],    }    return {        'statusCode': 200,        'headers': {'Content-Type': 'application/json'},        'body': json.dumps(msg, ensure_ascii=False),        'isBase64Encoded': False,    }

Отмечу ещё один нюанс: телеграм очень строго следует политике экранирования спецсимволов. Для Markdown это:

To escape characters '_', '*', '`', '[' outside of an entity, prepend the characters '\' before them.

Любой такой неэкранированный символ и вы не увидите ответа в телеграм-чате. И останется только гадать является ли это ошибка интеграции или вот такой коварный баг. Будьте осторожны.

Деплой бота в Яндекс.Функцию

Про создание Я.Функции также есть отличная статья от @mzaharov. Там подробно описан процесс заведения функции, а также установки вебхука для телеграмм бота.

Я расскажу как сделал Continuous Delivery при помощи GitHub Actions. Каждая сборка мастера увенчивается деплоем новой версии функции. Такой подход заставляет придерживаться модели разработки GithubFlow с его главным манифестом

Anything in themasterbranch is always deployable.

Каждая сборка мастера состоит из 3ёх этапов

• lint (black, flake8, isort, mypy) проверка кода на соответствие всем стандартам Python 2020

• test тестируем программу с помощью pytest, поддерживая качество и покрытие кода

• deploy непосредственно заливаем новую версию приложения в облако

Деплоить будем с помощью Yandex-Serveless-Action уже готового Action для использования в своих пайплайнах

  deploy:    name: deploy    needs: pytest    runs-on: ubuntu-latest    if: github.ref == 'refs/heads/master'    steps:      - uses: actions/checkout@master      - uses: goodsmileduck/yandex-serverless-action@v1        with:          token: ${{ secrets.YC_TOKEN }}          function_id: ${{ secrets.YC_FUNCTION_ID }}          runtime: 'python38'          memory: '256'          execution_timeout: "120"          entrypoint: 'main.handler'          environment: "\            TELEGRAM_BOT_TOKEN=${{ secrets.TELEGRAM_BOT_TOKEN }}"          source: 'app'

Переменные окружения программы и сборки спрячем в GitHub Secrets на уровне репозитория.

Результат

Бота можно найти в telegram по позывному @ballsortpuzzlebot.

Все исходники на Github.

Присоединяйтесь к маленькому community любителей этой игры в telegram. Бот был добавлен в группу и внимательно следит за всеми отправленными картинками.

Lvl 4091Lvl 6071

Мой алгоритм умывает руки говорит что перебрал все возможные комбинации и решения нет. Возможно это баг алгоритма.. или QA-отдел мобильной игры просто забил на эти уровни, так как не предполагал, что кто-то так далеко зайдёт)

Заключение

Для меня это был интересный опыт скрещивания технологий (Telegram API + Python + OpenCV + Lambda). Надеюсь он окажется полезен кому-нибудь ещё.

Найденные баги, предложения по оптимизации алгоритма, или даже PR в репозиторий крайне приветствуются

С новым годом!

На собеседованиях на позицию, предполагающую понимание DevOps, я люблю задавать кандидатам такой вопрос (а иногда его еще задают и мне):

Каким, по вашему мнению, должен быть идеальный пайплайн от коммита до продашкена?/Опишите идеальный CI/CD / etc

Сегодня я хочу рассказать про своё видение идеального пайплайна. Материал ориентирован на людей, имеющих опыт в построении CI/CD или стремящихся его получить.

Почему это важно?

1. Вопрос об идеальном пайплайне хорош тем, что он не содержит точного ответа.

2. Кандидат начинает рассуждать, а в крутых специалистах ценится именно умение думать.

3. Когда в вопрос добавляется такое абсолютное прилагательное, как "идеальный", то мы сразу развязываем кандидатам руки в просторе для творчества и фантазий. У соискателей появляется возможность показать, какие улучшения они видят (или не видят) в текущей работе, и что хотели бы добавить сами. Также мы можем узнать, есть ли у нашего предполагаемого будущего коллеги мотивация к улучшениям процессов, ведь концепция "работает не трогай" не про динамичный мир DevOps.

4. Организационная проверка. Позволяет узнать, насколько широка картина мира у соискателя. Условно: от создания задачи в Jira до настроек ноды в production. Сюда же можно добавить понимание стратегий gitflow, gitlabFlow, githubFlow.

Итак, прежде чем перейти к построению какого-либо процесса CI, необходимо определиться, а какие шаги нам доступны?

Что можно делать в CI?

• сканить код;

• билдить код;

• тестить код;

• деплоить приложение;

• тестить приложение;

• делать Merge;

• просить других людей подтверждать MR через code review.

Рассмотрим подробнее каждый пункт.

Code scanning

На этой стадии основная мысль никому нельзя верить.

Даже если Вася Senior/Lead Backend Developer. Несмотря на то, что Вася хороший человек/друг/товарищ и кум. Человеческий фактор, это все еще человеческий фактор.

Необходимо просканировать код на:

• соотвествие общему гайдлайну;

• уязвимости;

• качество.

Задачи на этой стадии следует выполнять параллельно.

И триггерить только если меняются исходные файлы, или только если было событие git push.

Пример для gitlab-ci

stages:  - code-scanning.code-scanning: only: [pushes] stage: code-scanning 

Linters

Линтеры это прекрасная вещь! Про них уже написано много статей. Подробнее можно почитать в материале "Холиварный рассказ про линтеры".

Самая важная задача линтеров приводить код к единообразию.

После внедрения этой штучки разработчики начнут вас любить. Потому что они наконец-то начнут понимать друг друга. Или ненавидеть, решив, что вы вставляете им палки в колеса линтеры в CI. Это уже зависит от ваших soft skills, культуры и обмена знаниями.

Инструменты

Code Quality

code quality этими инструментами могут быть как продвинутые линтеры, так и совмещающие в себе всякие ML-модели на поиск слабых мест в коде: утечек памяти, небезопасных методов, уязвимостей зависимостей и т.д, перетягивая на себя еще code security компетенции.

Инструменты

Code Security

Но существуют также и отдельные инструменты, заточенные только для code security. Они призваны:

1. Бороться с утечкой паролей/ключей/сертификатов.

2. Cканировать на известные уязвимости.

Неважно, насколько большая компания, люди в ней работают одинаковые. Если разработчик "ходит" в production через сертификат, то для своего удобства разработчик добавит его в git. Поэтому придется потратить время, чтобы объяснить, что сертификат должен храниться в vault, а не в git

Инструменты

Сканер уязвимостей необходимо запускать регулярно, так как новые уязвимости имеют свойство со временем ВНЕЗАПНО обнаруживаться.

Code Coverage

Ну и конечно, после тестирования, нужно узнать code coverage.

Процент исходного кода программы, который был выполнен в процессе тестирования.

Инструменты

Unit test

Модульные тесты имеют тенденцию перетекать в инструменты code quality, которые умеют в юнит тесты.

Инструменты

Build

Этап для сборки artifacts/packages/images и т.д. Здесь уже можно задуматься о том, каким будет стратегия версионирования всего приложения.

За модель версионирования вы можете выбрать:

• semVer (пример с gitflow);

• romVer;

• номер cборки;

• datetime, timestamp;

• etc

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

Инструменты для сборки образов

Итак, сборка прошла успешно идем дальше.

Scan package

Пакет/образ собрали. Теперь нужно просканировать его на уязвимости. Современные registry уже содержат инструментарий для этого.

Инструменты

Deploy

Стадия для развертывания приложения в различных окружениях.

Не все окружения хорошо сочетаются со стратегиями развертывания.

• rolling классика;

• recreate все что угодно, но не production;

• blue/green в 90% процентов случаев этот способ применим только к production окружениям;

• canary в 99% процентов случаев этот способ применим только к production окружениям.

Stateful

Нужно еще помнить, что даже имея одинаковый код в stage и production, production может развалиться именно из-за того, что stateful у них разный. Миграции могут отлично пройти на пустой базе, но появившись на проде, сломать зеленые кружочки/галочки в пайплайне. Поэтому для stage/pre-production следует предоставлять обезличенный бэкап основной базы.

И не забудьте придумать способ откатывания ваших релизов на последний/конкретный релиз.

Инструменты

На правах рекламы скажу что helmwav'у очень не хватает ваших звезд на GitHub. Первая публикация про helmwave.

Integration testing

Приложение задеплоили. Оно где-то живет в отдельном контуре.Наступает этап интеграционного тестирования. Тестирование может быть как ручным, так и автоматизированным. Автоматизированные тесты можно встроить в пайплайн.

Инструменты

Performance testing (load/stress testing)

Данный вид тестирования имеет смысл проводить на stage/pre-production окружениях. С тем условием, что ресурсные мощности на нем такие же, как в production.

Инструменты, чтобы дать нагрузку

Инструменты, чтобы оценить работу сервиса

Code Review / Approved

Одним из важнейших этапов являются Merge Request. Именно в них могут производиться отдельные действия в pipeline перед слиянием, а также назначаться группы лиц, требующих одобрения перед cлиянием.

Список команд/ролей:

• QA;

• Security;

• Tech leads;

• Release managers;

• Maintainers;

• DevOps;

• etc.

Очевидно, что созывать весь консилиум перед каждым MR не нужно, каждая команда должна появится в свой определённый момент MR:

• вызывать безопасников имеет смысл только перед сливанием в production;

• QA перед release ветками;

• DevOps'ов беспокоить, только если затрагиваются их компетенции: изменения в helm-charts / pipeline / конфигурации сервера / etc.

Developing flow

Чаще всего каждая компания, а то и каждый проект в компании, решает изобрести свой велосипед-флоу. Что после нескольких итераций приходит к чему-то, что может напоминать gitflow, gitlabFlow, githubFlow или все сразу.

Это и не хорошо, и не плохо это специфика проекта. Есть мнения, что gitflow не торт. GithubFlow для относительно маленьких команд. А про gitlabFlow мне нечего добавить, но есть наблюдение, что его не очень любят продакты - за то, что нельзя отслеживать feature-ветки.

Если вкратце, то:

• Gitflow: feature -> develop -> release-vX.X.X -> master (aka main) -> tag;

• GitHubFlow: branch -> master (aka main);

• GitLabFlow: environmental branches.

TL;DR

Общий концепт

_

Feature-ветка

Pre-Production -> Production

P.S.

Если я где-то опечатался, упустил важную деталь или, по вашему мнению, пайплайн недостаточно идеальный, напишите об этом мне сделаю update.

Разработчик создал ветку и запушил в нее код. Что дальше?

Оставляйте варианты ваших сценариев в комментариях.