Background Tasks on Faust, Part I: Introduction

How did I end up like this?

Not long ago, I had to work on the backend of a highly loaded project that required organizing the regular execution of a large number of background tasks with complex calculations and requests to third-party services. The project is asynchronous, and before I arrived, there was a simple cron-based task starting mechanism: a loop checking the current time and launching groups of coroutines through gather — this approach worked well until the number of such coroutines reached hundreds and then thousands. However, when their count exceeded two thousand, I had to think about organizing a proper task queue with a broker, several workers, and other components.

Initially, I decided to try Celery, which I had used before. Due to the project's asynchronous nature, I delved into the issue and saw the article, as well as project, created by the article's author.

I'll say this, the project is very interesting and works quite successfully in other applications of our team, and the author mentions that he managed to launch it in production, utilizing an asynchronous pool. But unfortunately, it didn't suit me very well, as I discovered issue issues with group task launches (see group). At the time of writing the article issue it had already been closed, but work was ongoing for a month. In any case, good luck and all the best to the author, as there are already functional things in the library… in general, the issue lies with me, and the tool turned out to be a bit raw for my needs. Moreover, some tasks had 2-3 HTTP requests to different services, thus even with task optimization, we were creating 4,000 TCP connections roughly every 2 hours — not ideal… I would like to establish a session for one type of task when launching workers. A bit more detail about the large number of requests through aiohttp here.

In connection with this, I started looking for alternatives and I found one! Created by the authors of Celery, specifically, as I understood Ask Solem, they created Faust, originally for the Robinhood. Faust is inspired by Kafka Streams and works with Kafka as a broker; it also uses RocksDB to store results from agent operations, and most importantly — this library is asynchronous.

You can also check a brief comparison Celery and Faust from the creators of the latest: their differences, broker differences, and the implementation of a basic task. It's all quite simple, but in Faust, a nice feature catches the eye — typed data for passing to the topic.

What shall we do?

So, in a small series of articles, I will show how to collect data in background tasks using Faust. The source for our example project will be, as the name suggests, alphavantage.co.I will demonstrate how to write agents (sink, topics, partitions), how to perform regular (cron) executions, the convenient CLI commands of Faust (a wrapper over click), simple clustering, and in the end, we will integrate Datadog (which works out of the box) and try to see something. We will use MongoDB and Motor for connecting to store collected data.

P.S. Given the confidence with which the monitoring point is written, I think that by the end of the last article the reader will look something like this:

Project Requirements

Since I've already made some promises, let's compile a short list of what the service should be able to do:

1. Export securities and an overview of them (including profits and losses, balance, cash flow — for the last year) — regularly
2. Export historical data (find the extremes of closing prices for each trading year) — regularly
3. Export the latest trading data — regularly
4. Export a configured list of indicators for each security — regularly

As it should be, we choose a name for the project out of thin air: horton

Preparing Infrastructure

The title is indeed strong; however, all that needs to be done is to write a small config for Docker Compose with Kafka (and Zookeeper — in one container), Kafdrop (if we want to look at messages in topics), MongoDB. We get [docker-compose.yml](https://github.com/Egnod/horton/blob/562fa5ec14df952cd74760acf76e141707d2ef58/docker-compose.yml) of the following form:

version: '3'

services:
  db:
    container_name: horton-mongodb-local
    image: mongo:4.2-bionic
    command: mongod --port 20017
    restart: always
    ports:
      - 20017:20017
    environment:
      - MONGO_INITDB_DATABASE=horton
      - MONGO_INITDB_ROOT_USERNAME=admin
      - MONGO_INITDB_ROOT_PASSWORD=admin_password

  kafka-service:
    container_name: horton-kafka-local
    image: obsidiandynamics/kafka
    restart: always
    ports:
      - "2181:2181"
      - "9092:9092"
    environment:
      KAFKA_LISTENERS: "INTERNAL://:29092,EXTERNAL://:9092"
      KAFKA_ADVERTISED_LISTENERS: "INTERNAL://kafka-service:29092,EXTERNAL://localhost:9092"
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: "INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT"
      KAFKA_INTER_BROKER_LISTENER_NAME: "INTERNAL"
      KAFKA_ZOOKEEPER_SESSION_TIMEOUT: "6000"
      KAFKA_RESTART_ATTEMPTS: "10"
      KAFKA_RESTART_DELAY: "5"
      ZOOKEEPER_AUTOPURGE_PURGE_INTERVAL: "0"

  kafdrop:
    container_name: horton-kafdrop-local
    image: 'obsidiandynamics/kafdrop:latest'
    restart: always
    ports:
      - '9000:9000'
    environment:
      KAFKA_BROKERCONNECT: kafka-service:29092
    depends_on:
      - kafka-service

There's nothing complicated here. For Kafka, two listeners were declared: one (internal) for use within the composite network, and the second (external) for requests from outside, hence it's exposed. 2181 is the Zookeeper port. I think the rest is clear.

Preparing the project skeleton

In the basic variant, the structure of our project should look like this:

horton
├── docker-compose.yml
└── horton
    ├── agents.py *
    ├── alphavantage.py *
    ├── app.py *
    ├── config.py
    ├── database
    │   ├── connect.py
    │   ├── cruds
    │   │   ├── base.py
    │   │   ├── __init__.py
    │   │   └── security.py *
    │   └── __init__.py
    ├── __init__.py
    ├── records.py *
    └── tasks.py *

*Everything I've marked we will not touch for now, and simply create empty files.**

We've created the structure. Now let's add the necessary dependencies, write the config, and connect to MongoDB. I won't provide the full text of the files in the article to avoid dragging it out, but I'll link to the necessary versions.

Let's start with dependencies and project metadata — pyproject.toml

Next, we initiate the installation of dependencies and creating a virtualenv (or, you can create a venv folder by yourself and activate the environment):

pip3 install poetry (if it's not installed yet)
poetry install

Now let's create config.yml — for credentials and connection details. You can also place the data for Alphavantage there. And now we proceed to config.py — extracting data for the application from our config. Yes, I admit, I used my library — sitri.

Connection to Mongo is quite simple. We've declared a client class for the connection and a base class for CRUD operations, to simplify requests to the collections.

What will happen next?

The article turned out to be quite short, as here I only talk about motivation and preparation, so I apologize — I promise that in the next part there will be some action and graphics.

So, in this very next part we will:

1. Write a small client for alphavantage using aiohttp with requests to the endpoints we need.
2. Create an agent that will collect data on securities and their historical prices.

Project Code

The code for this part

Source: habr.com