DEV Community: Gábor Boros

Why GitHub Releases Suck for Binary Distribution

Gábor Boros — Wed, 03 Sep 2025 14:35:56 +0000

A lot of us ship binaries through GitHub Releases because it is right there in the repo. It looks easy, it is free, and for many open source projects it is the default.

But once you start relying on it for real distribution, things get messy. GitHub Releases is fine for attaching a zip or tarball, but it was never designed to be a proper release hosting platform.

Pain points with GitHub Releases

1. Poor user experience
Users have to click through the release page and manually pick files. Direct URLs exist but they are long and brittle. If you want to provide installation through a package manager, GitHub Releases gives you nothing.

2. No download insights
You cannot see how many times a binary was downloaded. You cannot compare which platforms or versions people care about. If you are building multiple artifacts, you are essentially guessing which ones are worth the effort.

3. Limited automation
GitHub Actions lets you attach artifacts to a release, but that is where it stops. There is no API for download stats. There is no support for APT or YUM repos. You end up gluing together scripts to do things that should be built-in.

What developers actually need

For projects that ship binaries, we really need:

Easy hosting and sharing of binaries
Real download analytics
Support for package managers like APT and YUM
CI/CD integration without friction
A service focused on binaries, not an afterthought to code hosting

That is why I started building ZipZen. It is lightweight release hosting made for developers. You get reliable binary hosting, zero-config APT/YUM package repositories, and download statistics (though detailed stats will come soon).

GitHub Releases will frequently be the default option for attaching a file, but if you care about your users and your workflow, the cracks start to show. If you are running into the same pain points, maybe ZipZen is for you.

Curious what others think: do you stick with GitHub Releases or use something else for binaries?

Why would you need an URL shortener?

Gábor Boros — Wed, 13 Aug 2025 15:53:12 +0000

Streamline Your Digital Sharing with Concise, Impactful Links.

Ever been mid-scroll on social media and paused at an unwieldy, endless URL that disrupts an otherwise delightful post? You’re not alone.

First impressions count—even online. A link shortener turns a sprawling web address into a concise, shareable snippet that not only looks cleaner but also enhances your message. Let’s explore what a URL shortener is, why it’s essential for modern online communication, and how you can quickly create your own short links.

What Is a URL Shortener?

At its essence, a URL shortener takes a long, clunky web address and transforms it into a neat, compact version. For example, a lengthy URL like:

https://www.example.com/articles/insightful-tips-on-effective-link-sharing?utm_source=expatgeekdiaries.com

can be converted into something as simple as:

https://exmpl.co/a1B2c

This process works by generating a unique key for the original link and storing it in a database. When a user clicks the shortened URL, the system retrieves the original address and redirects them seamlessly. It’s a straightforward yet powerful tool for decluttering your digital space.

Why Use a URL Shortener?

There are many practical reasons to use URL shorteners beyond mere aesthetics:

Clean and attractive links: A short URL looks far more polished on social media, emails, and even printed materials.
Shareability: On platforms with character limits—like Twitter—it leaves more room for your message.
Analytics: Many services, such as Ziplink, offer tracking features, providing valuable data such as click counts, geographic information, and device types.
Branding possibilities: Custom short links can include your brand’s name, reinforcing your identity every time a link is shared.
Improved experience: A tidy link reduces visual clutter and allows your audience to focus on your content.

How to Create Your Own Short Links

Using a URL shortener is a breeze. Here’s a quick guide to get you started:

Choose a shortener: Pick a service that fits your needs—many offer options like custom branding, QR code creation, and analytics. Ziplink, Short.io, and Bitly are all capable of that. If you’re after simplicity, TinyURL is a great option too.
Enter Your URL: Depending on the service, you may need to sign up first. Then simply input the long web address you want to shorten.
Customize (Optional): Some tools let you add a personalized alias or custom back-half to the short link. Just note that these must be unique—for instance, https://exmpl.co/a1B2c can't be duplicated.
Generate and share: Once created, your new short link is ready to be used across your digital platforms.

Final Thoughts

Simplicity can significantly enhance your communication. URL shorteners do more than just reduce character count—they refine your links and improve how you connect with your audience. Whether you’re crafting a social media post, sending out a newsletter, or printing promotional materials, consider the impact a clean, efficient link can have on your message.

Spread the word—hopefully we’ll see fewer clunky URLs in the future!

URL shorteners and why would you need one?

Gábor Boros — Tue, 25 Mar 2025 14:30:00 +0000

Ever been mid-scroll on social media and paused at an unwieldy, endless URL that disrupts an otherwise delighting post? You're not alone.

First impressions count---even online. A link shortener turns a sprawling web address into a concise, shareable snippet that not only looks cleaner but also enhances your message. Let's explore what a URL shortener is, why it's essential for modern online communication, and how you can quickly create your own short links.

What Is a URL Shortener?

At its essence, a URL shortener takes a long, clunky web address and transforms it into a neat, compact version. For example, a lengthy URL like:

https://www.example.com/articles/insightful-tips-on-effective-link-sharing?utm_source=devto.com

can be converted into something as simple as:

https://exmpl.co/a1B2c

This process works by generating a unique key for the original link and storing it in a database. When a user clicks the shortened URL, the system retrieves the original address and redirects them seamlessly. It's a straightforward yet powerful tool for decluttering your digital space.

The mechanics (in the case of this example) are simple. We throw a long URL at the link shortener, and in return, it outputs a short URL.

Why Use a URL Shortener?

There are many practical reasons to use URL shorteners beyond mere aesthetics:

Clean and attractive links: A short URL looks far more polished on social media, emails, and even printed materials.
Shareability: On platforms with character limits---like Twitter, khm...---a compact link leaves more room for your message.
Analytics: Many services, such as Ziplink, offer tracking features, providing valuable data such as click counts, geographic information, and device types.
Branding possibilities: Custom short links can include your brand's name, reinforcing your identity every time a link is shared.
Improved experience: A tidy link reduces visual clutter and allows your audience to focus on your content.

These handy tools aren't just for tech experts---they're practical in many everyday situations:

Social media campaigns: Marketers track post performance and engagement with every click.
Email marketing: Shortened links add a level of sophistication to your newsletters and can boost click-through rates.
Text and messaging: With limited space in SMS and chat apps, a compact URL can be the difference between your message being read or overlooked.
Printed media: Business cards, flyers, and posters become more effective when paired with memorable, short links.
Content embedding: In blog posts and articles, shortened links maintain a neat layout without overwhelming your text.

Consider your daily digital interactions. Whether you're sharing a recipe on Facebook, promoting an event via email, or even incorporating a QR code in your print materials, a short URL can make the experience smoother and more professional.

How to Create Your Own Short Links

Using a URL shortener is a breeze. Here's a quick guide to get you started:

Choose a shortener: Pick a link shortener that fits your needs---many offer options like custom branding, QR code creation, and analytics. Ziplink, Short.io, and Bitly are all capable of that. If you are looking for something really simple, you may want to check out TinyURL as well.
Enter Your URL: Depending on the shortener, you need to sign up beforehand, but in essential: input the long web address you want to shorten.
Customize (Optional): Some services let you add a personalized alias/back-half to the short link. This can be a great option, but be aware: the aliases must be unique. So in the case of https://exmpl.co/a1B2c, no two a1B2c alias can exist at the same time.
Generate and share: Once created, your new link is ready to be shared across your digital channels.

Final Thoughts

Simplicity can significantly enhance your communication. URL shorteners do more than just reduce character count---they refine your links and improve how you connect with your audience. Whether you're crafting a social media post, sending out a newsletter, or printing promotional materials, consider the impact a clean, efficient link can have on your message.

Spread the word, hopefully we see less clunky URLs in the future!

Visualizing IoT Data with MQTT, QuestDB, and Grafana

Gábor Boros — Mon, 10 Jul 2023 08:10:45 +0000

Monitoring Time-series IoT Device Data

Time-series data is crucial for IoT device monitoring and data visualization in industries such as agriculture, renewable energy, and meteorology. It enables trend analysis, anomaly detection, and predictive analytics, empowering
businesses to optimize performance and make data-driven decisions. Thanks to technological advancements and the accessibility of open-source tools, gathering and analyzing data from IoT devices has become easier than ever before.

In this tutorial, we will guide you through the process of setting up a monitoring system for IoT device data. We will use historical electricity consumption data from some European countries, captured at a 15-minute resolution. The data is sent to an MQTT-compatible message broker called Mosquitto, and then channeled to QuestDB through Telegraf, a highly efficient data collector. To visualize the results, we will connect Grafana to QuestDB using its Postgres interface.

Since we don't have our own sensors to collect the data and send it directly to the broker, we will create the scenario by having a script that gathers the electricity consumption data from Open Power System Data and sends it to Mosquitto.

Prerequisites

Before delving into the system setup, please make sure that you have installed the following:

Docker
The latest Go version

Setting up the prerequisites

To get started, clone the prepared GitHub repository:

git clone https://github.com/questdb/questdb-mock-power-sensor mock-sensor

This repository contains all the configuration files and mock sensor scripts we will need during the tutorial.

Furthermore, we are going to need a new Docker network named "tutorial" to enable communication between containers without installing additional software on the host system. To create the new network, execute the following command:

docker network create tutorial

Setting up an MQTT broker

To enable communication between IoT devices and the monitoring system, we need a message broker such as Eclipse Mosquitto, which implements the lightweight MQTT protocol. By using Eclipse Mosquitto, we establish a reliable and efficient communication channel for IoT devices.

To run Eclipse Mosquitto, we are going to use Docker. The default configuration for Mosquitto does not allow unauthenticated clients. However, no default credentials are set. To fix this, let's use the conf/mosquitto/mosquitto.conf to override the default settings:

allow_anonymous true
listener 1883 0.0.0.0

To start the Eclipse Mosquitto broker using Docker, you can execute the following command from the repository root:

docker run --rm -dit -p 1883:1883 -p 9001:9001 \
 -v "$(pwd)"/conf/mosquitto/mosquitto.conf:/mosquitto/config/mosquitto.conf \
 --network=tutorial --name=mosquitto eclipse-mosquitto

By running this command, the Eclipse Mosquitto broker will start within a Docker container, configured with the settings from the mosquitto.conf file. The specified ports will be exposed to enable MQTT communication and the container will be connected to the tutorial Docker network, allowing it to communicate with other containers in the network.

Tunneling data into Mosquitto

Now that the message broker is running, in a new terminal window, navigate to the repository root and start the script that will populate the data for us:

cd script
go get
go run ./main.go

Great! The script will continue running in the background until it is manually stopped or until it runs out of data to publish. Since we are using a historical dataset, the script will override the original timestamps with the current time when sending the records to the MQTT broker.

With a 1-second delay between records, the script will consistently publish data to the broker, ensuring a steady flow of simulated sensor data for your tutorial. This allows you to progress with the rest of the tutorial without worrying about starting or stopping the script manually.

Setting up QuestDB

To ensure that all incoming IoT data flowing through Mosquitto is stored in QuestDB by the end of this tutorial, let's start QuestDB in the background. This will allow us to connect to it later using Telegraf.

Execute the following Docker command to start QuestDB:

docker run --rm -dit -p 8812:8812 -p 9000:9000 \
    -p 9009:9009 -e QDB_PG_READONLY_USER_ENABLED=true \
    --network=tutorial --name=questdb questdb/questdb

Once the command is executed, QuestDB will start running in the background. To validate that QuestDB is up and running, you can visit http://localhost:9000 in your web browser.

By accessing the provided URL, you should be able to access the QuestDB web interface, indicating that QuestDB has been successfully started.

Connecting the dots and adding Telegraf

In this setup, Telegraf will play a key role in transferring data between Mosquitto and QuestDB. To enable this seamless data transfer, we will utilize QuestDB's ILP (Influx Line Protocol) interface, as ILP is capable of handling large volumes of data efficiently sent from Telegraf.

To configure Telegraf, we are using the conf/telegraf/telegraf.conf file, which has the following content:

# Configuration for Telegraf agent
[agent]
  ## Default data collection interval for all inputs
  interval = "1s"

[[inputs.mqtt_consumer]]
  servers = ["tcp://mosquitto:1883"]
  topics = ["sensor"]
  data_format = "influx"
  client_id = "telegraf"
  data_type = "string"

# Write results to QuestDB
[[outputs.socket_writer]]
  # Write metrics to a local QuestDB instance over TCP
  address = "tcp://questdb:9009"

Let’s start the Telegraf container by executing:

docker run --rm -it \
    -v "$(pwd)"/conf/telegraf/telegraf.conf:/etc/telegraf/telegraf.conf \
    --network=tutorial --name=telegraf telegraf

Once Telegraf is up and running, the sensor data is automatically tunneled into QuestDB, and a new table called "sensor" is created to store the incoming data. To verify that the data is successfully flowing into QuestDB, you can execute the following query on the QuestDB console, accessible at http://localhost:9000:

SELECT * FROM sensor;

In the next steps of the tutorial, we will set up Grafana, which is a powerful data visualization and monitoring tool. We will connect Grafana to QuestDB and create compelling visualizations based on the received sensor data.

Visualizing the dataset with Grafana

To proceed with creating a Grafana container and connecting it to QuestDB, execute the following command:

docker run --rm -dit -p 3000:3000 \
    --network=tutorial --name=grafana grafana/grafana

This command will start a Grafana container within the same tutorial network. To access the Grafana login page, navigate to http://localhost:3000 in your web browser. You will be prompted to log in using the default credentials:

Username: admin
Password: admin

To connect Grafana with QuestDB and establish a data source, follow these steps:

Open your web browser and go to http://localhost:3000/datasources/new.
On the "New data source" page, select "PostgreSQL" as the data source type from the list of available options.

Fill in the form with the following details:

Name: QuestDB
Host: questdb:8812
Database: qdb
User: user
Password: quest
TLS/SSL mode: disable

Once you have filled in the form with the correct information, click on the "Save & Test" button at the bottom of the page. Grafana will attempt to connect to QuestDB using the provided credentials and verify the connection. If everything is set up correctly, you should see a success message indicating that the data source was added successfully.

Now that Grafana is successfully connected to QuestDB, you can begin creating insightful dashboards to visualize the electricity consumption data. Follow these steps to create a new dashboard:

In your browser and navigate to the new dashboard page.
Once you're on the new dashboard page, click on the "Add visualization" button.

On the popup panel, select “QuestDB” data source.

Now you're ready to start building your dashboard and visualizing the electricity consumption data from QuestDB. Grafana provides a wide range of visualization options, including graphs, charts, tables, and more. You can customize the panel settings, apply different visualization types, and configure data queries to create meaningful visual representations of the data.

To visualize the electricity consumption data from QuestDB, we will use the "Time series" chart type, which is the default selected chart based on our data. The "Time series" chart is ideal for displaying data over time and is well-suited for analyzing trends and patterns.

In the query builder panel, switch to “Code” and paste the following SQL query:

SELECT
    date_trunc('minute', timestamp) AS minute,
    AVG(load_actual) AS avg_load_actual,
    AVG(load_forecast) AS avg_load_forecast
FROM sensor
WHERE country = 'NL'
GROUP BY minute
ORDER BY minute ASC

By executing this query, we will obtain a result set that provides minutely aggregated data, showing the average, actual, and forecasted energy load for each minute. Feel free to adjust the sampling or grouping interval if you wish. If you followed this tutorial successfully, you should see similar results.

To make the changes persistent while the container is running, rename the title of the graph on the right-hand side properties pane, and save it by clicking on the “Apply” button in the top right corner.

The time series chart we have created, with the average actual and forecasted energy consumption, provides valuable insights into the accuracy of the forecasting logic. By analyzing this chart, we can easily identify any discrepancies between the actual and forecasted values, helping you pinpoint areas where improvements can be made.

Cleaning up resources

Upon finishing the tutorial, you may want to clean up any dangling resources, such as the tutorial network we created. Removing a network requires the removal of existing resources attached to that first. To remove the containers, the network, and the images used during this tutorial, run the following commands:

docker ps -qa --filter="network=tutorial" | xargs -n1 docker kill
docker network rm tutorial
docker rmi telegraf eclipse-mosquitto grafana/grafana questdb/questdb

Summary

In this tutorial, we explored the setup and configuration of a monitoring system for IoT device data using MQTT, Telegraf, QuestDB, and Grafana. Through a series of steps, we established communication between IoT devices and the monitoring system using Eclipse-Mosquitto as the MQTT broker.

We simulated data collection using a script that gathered electricity consumption from Open Power System Data and sent it to the MQTT broker. We stored this data in QuestDB and visualized it using Grafana. By connecting Grafana to QuestDB, we created informative dashboards to gain insights.

The tutorial showcased the power of time-series data and its visualization, enabling us to compare actual and forecasted energy consumption.

Loading Pandas DataFrames into QuestDB

Gábor Boros — Mon, 13 Mar 2023 19:05:12 +0000

Learn how to improve your time series analysis capability by using the QuestDB Python package to ingest Pandas DataFrames.

Introduction

Pandas is an open-source data analysis and data manipulation library for Python that has become an essential tool for data scientists and analysts. It provides a simple and intuitive way to manipulate data, making it a popular choice for data analysis tasks. However, while Pandas is excellent for small to medium-sized datasets, it can struggle with large datasets that exceed the available memory of the machine it is running on. This is where QuestDB excels, designed specifically for high-performance operations in such scenarios, making it the go-to solution for demanding data analysis tasks. By loading Pandas DataFrames into QuestDB, we can leverage powerful data processing capabilities of the database, allowing you to scale your analysis and data manipulation operations to large datasets. We will learn how to load large Pandas dataframes into QuestDB. We use yellow and green taxi trip records published by NYC Taxi & Limousine Commission as our data source.

In this tutorial, we will learn how to load large Pandas dataframes into QuestDB. We use yellow and green taxi trip records published by NYC Taxi & Limousine Commission as our data source.

Prerequisites

For this tutorial, it is recommended to have a basic understanding of Python, and SQL. Also, you will need to have the following installed on your machine:

Docker

Getting the data to ingest

Before we begin loading the data into QuestDB, we need to obtain the data that we will be working with. As mentioned above, we will use the NYC TLC’s records of yellow and green taxi trips. Let’s download the data:

Create a new directory called pandas-to-questdb and a data directory inside that.
Edit and execute the following command in your terminal to download the parquet files:

curl -L -o ./data/yellow_tripdata_2022-<MONTH>.parquet https://d37ci6vzurychx.cloudfront.net/trip-data/yellow_tripdata_2022-<MONTH>.parquet

Make sure you replace <MONTH> with the zero-prefixed number of the month you wish to download (between 01 and 11, the 12th month is not available at the time of writing).

Now, we have the data to ingest. It is time to try loading it using Pandas.

Loading records into memory

You may already have noticed that the downloaded files are in Parquet format. Parquet is a columnar storage format commonly used for big data processing. They are optimized for use with modern big data processing engines and provide efficient storage and retrieval of data compared to traditional row-based storage formats like CSV and JSON.

Before being able to load any data, we are going to set up a simulation production environment in which we can easily test what happens if Pandas cannot load the Parquet files into memory. In production, we often meet situations where we have to deal with memory constraints, and this environment can reflect that.

Run the following command to create a new docker container running with 1GiB memory limit. If the container reaches that limit, either docker will kill it, or the OS will OOM kill the process we are running.

docker run -it -m 1g -v "$(pwd)":/tutorial -w /tutorial --net host python:3.11.1-slim-bullseye /bin/bash

Right, we have an Ubuntu-based Python 3.11 docker container. Let’s install our requirements. Create a requirements.txt file with the content below:

pandas>=1.5.3
psycopg[binary]>=3.1.8
pyarrow>=11.0.0
questdb>=1.1.0

Now, execute pip install -r requirements.txt within the container. Pip will install the python requirements.

At this point we have a test environment in which we can load the data. Create a new file, called data_loader.py, with the following content:

# data_loader.py

import pandas as pd

df = pd.read_parquet("./data/yellow_tripdata_2022-01.parquet")
print(df.head())

Now, execute it within the docker container by running python data_loader.py. The program runs successfully and we should see the following:

We just loaded the taxi trip records for 2022 January! Let’s try to load more data. Replace the content of data_loader.py by the code below to load all files from the data directory and execute the program again:

# data_loader.py

import os
import glob
import pandas as pd

records = glob.glob(os.path.join("data", "*.parquet"))

df = pd.concat((pd.read_parquet(r) for r in records), ignore_index=True)

print(df.head())

When executing the data_loader.py you should get an error message: “Killed”. As you may assume, OOM killer terminated the process. We were not able to load the dataset, therefore we cannot work with that. We need a different approach.

Ingesting to QuestDB

In a new terminal window, start a QuestDB container by executing:

docker run --rm -it -p 8812:8812 -p 9009:9009 --net host --name questdb questdb/questdb

The database is now ready to receive the data. Update the data_loader.py to ingest data into QuestDB using the questdb package that uses the InfluxDB Line Protocol (ILP) over TCP for maximum throughput.

To handle large datasets, we will read files one by one and transfer their contents to QuestDB. Then, we will use QuestDB to query the data and load the results back into Pandas DataFrames. Refactor the data loader based on the above:

# data_loader.py

import os
import glob
import pandas as pd
from questdb.ingress import Sender

def main():
   files = glob.glob(os.path.join("data", "*.parquet"))

   with Sender("127.0.0.1", 9009) as sender:
       for file in files:
           df = pd.read_parquet(file)
           print(f"ingesting {len(df.index)} rows from {file}")
           sender.dataframe(df, table_name="trips", at="tpep_pickup_datetime")

if __name__ == "__main__":
   main()

Let's start from the beginning. The first major change you'll notice is that we need to specify the hostname and port number in the script in order to run it.

Then we iterate over the parquet files and load them into memory using Pandas. After that, utilizing QuestDB’s Python client, we are ingesting to QuestDB directly from Pandas DataFrames.

In the Python container, run python data_loader.py. The script will ingest one parquet file at a time.

Working with trip data

So far, we have prepared the dataset and loaded it into QuestDB. It’s time to execute some queries and load the result into DataFrames. Using the whole dataset, we want to know what was the average total amount paid by passengers grouped by the passengers.

Create a new file, called query_amount.py with the following content:

# query_amount.py

import pandas as pd
import psycopg

QUERY = """
SELECT passenger_count, avg(total_amount)
   FROM 'trips'
   WHERE passenger_count > 0
   GROUP BY passenger_count
"""

if __name__ == "__main__":
   conn = psycopg.connect(
       dbname="questdb",
       host="127.0.0.1",
       user="admin",
       password="quest",
       port=8812,
   )

   df = pd.read_sql_query(QUERY, conn)

   print(df.head(10))

Similarly to the data loader script, this script requires the host and port too.

In the script above, we are using the Postgresql Python client and connecting to QuestDB using that. In the Python container, execute python query_amount.py:

When the script finishes, you should see the average total amount paid by passengers. Interestingly, there is a huge difference in the average between passenger count 6 and 7, the average almost 2.5x for 7 passengers compared to 6.

By further analyzing the data, it may turn out what was the possible root cause increase, but probably it is bound to human nature: we like to share the cost of rides if we are going on a longer trip.

Summary

In this tutorial, we have learned how to load large datasets into QuestDB using Pandas DataFrames. By transferring data from Pandas to QuestDB, we have taken advantage of the database's powerful data processing capabilities, enabling us to scale our analysis and data manipulation operations to handle large datasets.

The approach outlined in this tutorial is just one way to work with big data using Pandas and QuestDB. You can customize this method to suit your specific needs and continue to explore the possibilities of these powerful tools. The end goal is to make data analysis and manipulation easier and more efficient, regardless of the size of the dataset.

How to generate time-series data in QuestDB

Gábor Boros — Wed, 16 Mar 2022 17:55:33 +0000

As developers, we often have to generate sample data for numerous reasons: feeding integration tests, realistic staging environments or developing an application locally.

This process can be time-consuming, especially when we need data on a bigger scale. Fortunately, some databases are helping our work and trying to offload some weight from our shoulders by providing functions to generate data.

In this tutorial, we discover how to generate test data using QuestDB, the fastest time series database.

Types of time series data

Before diving deep into generating data using QuestDB, we need to take one step back to discuss what types of time series data we can generate, what "mock data" is, and what generator functions are available for use. Let's start with the different types of time series data.
We can classify the data based on the frequency we receive them into two categories: regular and irregular data.

Regular data is received per a pre-defined period, usually collected by a collector node or sent by an agent. Collecting metrics from a server fits into this category perfectly, as shown in the tutorial where we connected Telegraf with QuestDB. The Telegraf agent running on the server collects the metrics on a pre-defined basis and sends them to the QuestDB instance.

The opposite of regular data is the irregular data which is collected dynamically. The main point about irregular data is that we cannot predict how often the data will arrive. As of examples, automating ETL jobs could be a great example. The data comes in when automation or the user sends it to the pipeline, hence we cannot know beforehand the data arrives.

Of course, these categories can split into multiple groups, for instance, based on the field it used: IoT, financial, operations, industrial telemetry, and so on.
Regardless of how we categorize the data, the most important is knowing the data and use-case before generating it.

What is mocking and why is it useful?

Software development is about making things, but in some cases, testing what we make is challenging. There are multiple scenarios when we want to check for errors and failures, invalid input, we need lots of test data, or we want to operate a production-like environment without user data.

The function or application we test may have dependencies. To test the edge cases, we have to isolate and simulate the behavior of these dependencies. This kind of simulation can happen on multiple levels. Think about integration test suites; we do not want to imitate the behavior of the components we are testing, but we may want to load fake data in the system.

Simulation of the behavior of components or data is called mocking. Remember the ETL job mentioned above. If we would like to test hundreds of thousands of processing, we would need to craft the input data on our own. That would be time-consuming and ineffective. On the other hand, generating mock data help us focus on the processor job, not the data generation.

The way we generate mock data may differ case-by-case. Since it highly depends on the application being developed, let's see how can we mock different types of data with QuestDB easily.

Running QuestDB

To discover the capabilities of how QuestDB can generate data for us, we need to run it. We will use Docker and the latest questdb Docker image for convenience.
To start QuestDB via Docker, run the following:

docker run -p 9000:9000 \
  -p 9009:9009 \
  -p 8812:8812 \
  questdb/questdb

Alternatively, macOS users can use Homebrew:

brew install questdb
brew services start questdb

The QuestDB downloads page has downloads for binaries if you wish to run it as a standalone.

Mocking different types of data in QuestDB

After starting QuestDB, the web console is available on port 9000, so navigating to http://127.0.0.1:9000 should show the UI of a locally-running instance which looks like the following:

Mock data generation in QuestDB is surprisingly easy. The DB comes with numerous generator functions, all with a different purpose. If you ever used PostgreSQL's generate_series, you will find QuestDB's generator functions pleasant and relieving.
To generate rows, we use the long_sequence generator function that can create billions of rows depending on our needs. Its syntax looks like

SELECT <COLUMNS> FROM long_sequence(<NUMBER_OF_ROWS>[, <SEED1>, <SEED2>]]);

If we need deterministic row generation, we can optionally provide <SEED1> and <SEED2>, that are 64 bit integers, and anchor the start of the random generator to obtain the same sequence on every run. That could be useful when populating data for tests, CI/CD pipelines, or development environments. Let's generate some rows to see long_sequence in action.

For a first query, paste the following in the SQL editor:

SELECT x, x*x FROM long_sequence(5);

You will see five rows with two columns. The first column counts from one to five, while the second column contains the square of the values of column one.

x	column
1	1
2	4
3	9
4	16
5	25

Easy, right? It is that simple to get started on generating mock data with QuestDB.
We have basic example of generating numbers, and next, we'll look at generating booleans, characters, strings, timestamps, and many more types.

Generator functions in QuestDB

We have seen how to generate basic rows, and we mentioned random value generation too. It's time to see how we can generate random values with QuestDB's built-in random generator functions.

QuestDB supports random generator functions for several data types:

type	generator function
binary	rnd_bin
boolean	rnd_boolean
byte	rnd_byte
char	rnd_char
date	rnd_date
double	rnd_double
float	rnd_float
geohash	rnd_geohash
int	rnd_int
long256	rnd_long256
long	rnd_long
short	rnd_short
string	rnd_str
symbol	rnd_symbol
timestamp	rnd_timestamp

All these functions can be parameterized with seed values to narrow the function's range. This is an extremely useful feature, especially if we need to debug our application using the generated test data.

Although these functions can be handy in many scenarios, it is important to highlight that these functions should be used for populating test tables only. They do not hold values in memory and calculations should not be performed at the same time as the random numbers are generated.

The best practice when populating test tables is creating the table first, then populating and querying it. Consider the following example:

SELECT round(a,2), a FROM (SELECT rnd_double() a FROM long_sequence(10));

This would return inconsistent and ephemeral results while the approach below will return consistent results that are persisted as a normal database table.

-- creating the table first
CREATE TABLE test(val double);
-- populating the test table
INSERT INTO test SELECT * FROM (SELECT rnd_double() FROM long_sequence(10));
-- querying generated data
SELECT round(val,2) FROM test;

Now that we know what types of data can be generated and have seen how to generate random values, it is time to create more complex data.

Creating realistic generative data

To see data generation in action, we will replicate the schema used by the Time Series Benchmark Suite, a tool for comparing and evaluating databases for time series data.

In the benchmarking example, TSBS uses truck diagnostic data sent through the InfluxDB protocol to populate tables. We will take a different approach. Instead of sending the data, we will generate it directly using the web UI.

Following the approach described above, let's create the table first. Navigate to the web UI and create the diagnostics table that you can see below.

CREATE TABLE 'diagnostics' (
    name                     STRING,   -- identifier of the truck
    fleet                    STRING,   -- division where the truck belongs to
    driver                   STRING,   -- the name of the driver who drove the track on the given day
    model                    STRING,   -- the model of the truck
    device_version           STRING,   -- version number of the data collector device
    load_capacity            INT,      -- maximum load capacity of the truck at the beginning of the shift
    fuel_capacity            INT,      -- fuel capacity of the truck at the beginning of the shift
    nominal_fuel_consumption INT,      -- fuel consumption at the given time
    fuel_state               DOUBLE,   -- the current state of the fuel level: 1 is full, 0 is empty
    current_load             INT,      -- the weight of the current load
    status                   INT,      -- status of the truck
    ts                       TIMESTAMP -- timestamp the diagnostic entry got recorded
) timestamp(ts) PARTITION BY year;

By this point, we have the table without any records. For now, we don't set constraints on the data, like the capacity must be consistent or fuel capacity should remain the same over the days. Also, we assume that we get a report from a sensor per hour, no matter from which truck.

Keeping the above in mind, let's populate the table with some test data. For doing that, we are going to use four different random value generators, namely rnd_int, rnd_double, rnd_str, and rnd_timestamp. All these functions have different arguments and capabilities. To generate some rows, run the following query:

INSERT INTO 'diagnostics' SELECT * FROM (
    SELECT
        rnd_str('truck_1234', 'truck_5678', 'truck_9123', 'truck_3210'),
        rnd_str('North', 'East', 'South', 'West'),
        rnd_str('Alice', 'Bob', 'John', 'Mike', 'Robert'),
        rnd_str('G-0', 'H-2', 'I-1'),
        rnd_str('v1.2', 'v1.3'),
        rnd_int(1000, 1500, 0),
        rnd_int(150, 200, 0),
        rnd_int(0, 15, 0),
        rnd_double(),
        rnd_int(0, 1500, 0),
        rnd_int(0, 4, 0),
        rnd_timestamp(
            to_timestamp('2022-01-01T00:00:00', 'yyyy-mm-ddTHH:mm:ss'),
            to_timestamp('2022-02-28T23:00:00', 'yyyy-mm-ddTHH:mm:ss'),
            0
        )
    FROM long_sequence(28)
);

After the query ran successfully, ensure we have the 28 records we expect. Run SELECT * FROM 'diagnostics';

Congratulations, we successfully generated twenty eight rows with random data. Although the data is in place, the timestamps seem to be odd; we have multiple records for the same hour. The reason behind this is how rnd_timestamp works. It generates data within the given range, without considering any sequence.

To fix this issue, we need to use a timestamp sequence generator, called timestamp_sequence. This generator can be used as a timestamp generator to create data for testing. The timestamp is increased by providing a steps argument, which sets the step size in microseconds.

SELECT
    timestamp_sequence(
        to_timestamp('2022-02-01T00:00:00', 'yyyy-mm-ddTHH:mm:ss'),
        3600000000L
    )
FROM long_sequence(10);

Running the query above generates ten rows, where every row will follow the previous row's value increased by one hour.

Now that we know how we can generate sequentially increasing timestamps, refactor the query we used to generate test data as follows. Before running the query, execute truncate table 'diagnostics' to clean up the previously generated rows.

INSERT INTO 'diagnostics' SELECT * FROM (
    SELECT
        rnd_str('truck_1234', 'truck_5678', 'truck_9123', 'truck_3210'),
        rnd_str('North', 'East', 'South', 'West'),
        rnd_str('Alice', 'Bob', 'John', 'Mike', 'Robert'),
        rnd_str('G-0', 'H-2', 'I-1'),
        rnd_str('v1.2', 'v1.3'),
        rnd_int(1000, 1500, 0),
        rnd_int(150, 200, 0),
        rnd_int(0, 15, 0),
        rnd_double(),
        rnd_int(0, 1500, 0),
        rnd_int(0, 4, 0),
        timestamp_sequence(
            to_timestamp('2022-01-01T00:00:00', 'yyyy-mm-ddTHH:mm:ss'),
            3600000000L
        )
    FROM long_sequence(28)
);

The new query will create 28 records, where all timestamps increase by one hour. Although this is close to what we would like to see, it is still not 28 days. From now on, we need to do some basic math. To get 28 days of records where every record is one hour away from the previous one, we need to generate 28 * 24 records (that's 672 records).

Changing FROM long_sequence(28) to FROM long_sequence(672) and running the query will generate the expected number of records with the expected timestamp differences.

So far, we have learned how to generate simple and more complex test data, also we experienced the difference between generating random timestamps and sequentially increasing timestamps. You may now wonder, "How could we generate ten years of data?". The answer is simple, and you probably assume it: let's calculate a bit.

If we would like to generate test data for one year, we would need to generate 365 * 24 records.
Based on this logic, to produce the desired ten years of random data we generate 10 * 365 * 24 records, which is 87600 rows.
By simply changing the long_sequence parameters to 87600 and running the query one more time, we can generate data for the next ten years in less than a second.

Summary

In this tutorial, we discussed what mock data is useful, what kinds of types are available in QuestDB, how to generate random data, and create tables with test data for 1, 3, or 10 years. This tutorial demonstrated well how to combine the generator functions to create complex generative data. Thank you for your attention!

If you like the content and want to learn more about QuestDB, check out the GitHub repository.

Real-time stock price dashboard using QuestDB, Python and Plotly

Gábor Boros — Thu, 04 Nov 2021 14:40:40 +0000

Why Plotly and Dash are useful for real-time applications

If you're working with large amounts of data, efficiently storing raw information will be your first obstacle. The next challenge is to make sense of the data utilizing analytics. One of the fastest ways to convey the state of
data is through charts and graphs.

In this tutorial, we will create a real-time streaming dashboard using QuestDB, Celery, Redis, Plotly, and Dash. It will be a fun project with excellent charts to quickly understand the state of a system with beautiful data visualizations.

Plotly defines itself as "the front end for ML and data science models", which describes it really well. Plotly has an "app framework" called Dash which we can use to create web applications quickly and efficiently. Dash abstracts away the boilerplate needed to set up a web server and several handlers for it.

Project overview

The project will be built from two main components:

a backend that periodically fetches user-defined stock data from Finnhub, and
a front-end that utilizes Plotly and Dash to visualize the gathered data on interactive charts

For this tutorial, you will need some experience in Python and basic SQL knowledge. We will use Celery backed by Redis as the message broker and QuestDB as storage to periodically fetch data.

Let's see the prerequisites and jump right in!

Prerequisites

Python 3.8
Docker & Docker Compose
Finnhub account and sandbox API key
Basic SQL skills

The source code for this tutorial is available at the corresponding GitHub repository.

Environment setup

Create a new project

First of all, we are going to create empty directories for our project root and the Python module:

mkdir -p streaming-dashboard/app
# streaming-dashboard
# └── app

Installing QuestDB & Redis

To install the services required for our project, we are using Docker and Docker Compose to avoid polluting our host machine. Within the project root, let's create a file, called docker-compose.yml. This file describes all the necessary requirements the project will use; later on we will extend this file with other services too.

# docker-compose.yml

version: "3"

volumes:
  questdb_data: {}

services:
  redis:
    image: "redis:latest"
    ports:
      - "6379:6379"

  questdb:
    image: "questdb/questdb:latest"
    volumes:
      - questdb_data:/root/.questdb/db
    ports:
      - "9000:9000"
      - "8812:8812"

Here we go! When you run docker-compose up, QuestDB and Redis will fire up. After starting the services, we can access QuestDB's interactive console on http://127.0.0.1:9000.

Create the database table

We could create the database table later, but we will take this opportunity and create the table now since we have already started QuestDB. Connect to QuestDB's interactive console, and run the following SQL statement:

CREATE TABLE
      quotes(stock_symbol SYMBOL CAPACITY 5 CACHE INDEX, -- we are in fact just checking 3
             current_price DOUBLE,
             high_price DOUBLE,
             low_price DOUBLE,
             open_price DOUBLE,
             percent_change DOUBLE,
             tradets TIMESTAMP, -- timestamp of the trade
             ts TIMESTAMP)      -- time of insert in our table
      timestamp(ts)
PARTITION BY DAY;

After executing the command, we will see a success message in the bottom left corner, confirming that the table creation was successful and the table appears on the right-hand side's table list view.

Voilà! The table is ready for use.

Creating workers using Celery

Define Python dependencies

As mentioned, our project will have two parts. For now, let's focus on the routine jobs that will fetch the data from Finnhub. As is the case of every standard Python project, we are using requirements.txt to define the dependencies the project will use. Place the requirements.txt in your project root with the content below:

finnhub-python==2.4.14  # The official Finnhub Python client
pydantic[dotenv]==1.9.2 # We will use Pydantic to create data models
celery[redis]==5.2.7    # Celery will be the periodic task executor
psycopg[c,pool]==3.1.1  # We are using QuestDB's PostgreSQL connector
dash==2.6.1             # Dash is used for building data apps
pandas==1.4.3           # Pandas will handle the data frames from QuestDB
plotly==5.10.0          # Plotly will help us with beautiful charts

We can split the requirements into two logical groups:

requirements for fetching the data, and
requirements needed to visualize this data

For the sake of simplicity, we did not create two separate requirements files, though in a production environment we would do.

In order to let the application communicate with QuestDB utilizing the psycopg
client library, we need to install libpq-dev package on our system. To install
it, use your package manager; on Windows, you may need to install PostgreSQL on
your system.

Then, create a virtualenv and install the dependencies:

$ virtualenv -p python3.8 virtualenv
$ source virtualenv/bin/activate
$ pip install -r requirements.txt

Setting up the DB connection

Since the periodic tasks would need to store the fetched quotes, we need to connect to QuestDB. Therefore, we create a new file in the app package, called db.py. This file contains the PostgreSQL connection pool that will serve as the
base for our connections.

# app/db.py

from psycopg_pool import ConnectionPool
from app.settings import settings

pool = ConnectionPool(
    settings.database_url,
    min_size=1,
    max_size=settings.database_pool_size,
)

Define the worker settings

Before we jump right into the implementation, we must configure Celery. To create a configuration used by both the workers and the dashboard, create a settings.py file in the app package. We will use pydantic's BaseSettings to define the configuration. This helps us to read the settings from a .env file, environment variable, and prefix them if needed.

Ensuring that we do not overwrite any other environment variables, we will set the prefix to SMD that stands for "stock market dashboard", our application. Below you can see the settings file:

# app/settings.py

from typing import List

from pydantic import BaseSettings


class Settings(BaseSettings):
    """
    Settings of the application, used by workers and dashboard.
    """

    # Celery settings
    celery_broker: str = "redis://127.0.0.1:6379/0"

    # Database settings
    database_url: str = "postgresql://admin:quest@127.0.0.1:8812/qdb"
    database_pool_size: int = 3

    # Finnhub settings
    api_key: str = ""
    frequency: int = 5  # default stock data fetch frequency in seconds
    symbols: List[str] = list()

    # Dash/Plotly
    debug: bool = True
    graph_interval: int = 10

    class Config:
        """
        Meta configuration of the settings parser.
        """

        env_file = ".env"
        # Prefix the environment variable not to mix up with other variables
        # used by the OS or other software.
        env_prefix = "SMD_"  # SMD stands for Stock Market Dashboard


settings = Settings()

In the settings, you can notice we already defined the celery_broker and database_url settings with unusual default values.

Some bits are missing at the moment. We still have to define the correct settings and run the worker in a Docker container. Get started with the settings!

To keep our environment separated, we will use a .env file. One of pydantic based settings' most significant advantage is that it can read environment variables from .env files.

Let's create a .env file in the project root, next to docker-compose.yml, so
your project structure should look like this:

├── app
│   ├── __init__.py
│   ├── db.py
│   ├── settings.py
│   └── worker.py
├── .env
├── docker-compose.yml

Add the following content to the .env file:

SMD_API_KEY = "<YOUR SANDBOX API KEY>"
SMD_FREQUENCY = 10
SMD_SYMBOLS = ["AAPL","DOCN","EBAY"]

As you may assume, you will need to get your API key for the sandbox environment at this step. To retrieve the key, the only thing you have to do is sign up to Finnhub, and your API key will appear on the dashboard after login.

Create the periodic task

# app/worker.py
import finnhub
from celery import Celery
from app.db import pool
from app.settings import settings
client = finnhub.Client(api_key=settings.api_key)
celery_app = Celery(broker=settings.celery_broker)
@celery_app.on_after_configure.connect
def setup_periodic_tasks(sender, **kwargs):
    """
    Set up a periodic task for every symbol defined in the settings.
    """
    for symbol in settings.symbols:
        sender.add_periodic_task(settings.frequency, fetch.s(symbol))
    quote: dict = client.quote(symbol)
    # https://finnhub.io/docs/api/quote
    # quote = {'c': 148.96, 'd': -0.84, 'dp': -0.5607, 'h': 149.7, 'l': 147.8, 'o': 148.985, 'pc': 149.8, 't': 1635796803}
    # c: Current price
    # d: Change
    # dp: Percent change
    # h: High price of the day
    # l: Low price of the day
    # o: Open price of the day
    # pc: Previous close price
    # t: when it was traded
    query = f"""
    INSERT INTO quotes(stock_symbol, current_price, high_price, low_price, open_price, percent_change, tradets, ts)
    VALUES(
        '{symbol}',
        {quote["c"]},
        {quote["h"]},
        {quote["l"]},
        {quote["o"]},
        {quote["pc"]},
        {quote["t"]} * 1000000,
        systimestamp()
    );
    """
    with pool.connection() as conn:
        conn.execute(query)

Going through the code above:

import finnhub
from celery import Celery

from app.db import pool
from app.settings import settings

# [...]

In the first few lines, we import the requirements that are needed to fetch and store the data.

After importing the requirements, we configure the Finnhub client and Celery to use the Redis broker we defined in the application settings.

# [...]

client = finnhub.Client(api_key=settings.api_key)
celery_app = Celery(broker=settings.celery_broker)

# [...]

To fetch the data periodically per stock symbol, we need to programmatically
create a periodic task for every symbol we defined in the settings.

# [...]

@celery_app.on_after_configure.connect
def setup_periodic_tasks(sender, **kwargs):
    """
    Set up a periodic task for every symbol defined in the settings.
    """
    for symbol in settings.symbols:
        sender.add_periodic_task(settings.frequency, fetch.s(symbol))

# [...]

The snippet above will register a new periodic per stock symbol after Celery is connected to the broker.

The last step is to define the fetch task that does the majority of the work.

# [...]

@celery_app.task
def fetch(symbol: str):
    """
    Fetch the stock info for a given symbol from Finnhub and load it into QuestDB.
    """

    quote: dict = client.quote(symbol)
    # https://finnhub.io/docs/api/quote
    # quote = {'c': 148.96, 'd': -0.84, 'dp': -0.5607, 'h': 149.7, 'l': 147.8, 'o': 148.985, 'pc': 149.8, 't': 1635796803}
    # c: Current price
    # d: Change
    # dp: Percent change
    # h: High price of the day
    # l: Low price of the day
    # o: Open price of the day
    # pc: Previous close price
    # t: when it was traded

    query = f"""
    INSERT INTO quotes(stock_symbol, current_price, high_price, low_price, open_price, percent_change, tradets, ts)
    VALUES(
        '{symbol}',
        {quote["c"]},
        {quote["h"]},
        {quote["l"]},
        {quote["o"]},
        {quote["pc"]},
        {quote["t"]} * 1000000,
        systimestamp()
    );
    """

    with pool.connection() as conn:
        conn.execute(query)

Using the Finnhub client, we get a quote for the given symbol. After the quote is retrieved successfully, we prepare a SQL query to insert the quote into the database. At the end of the function, as the last step, we open a connection to QuestDB and insert the new quote.

Congratulations! The worker is ready for use; let's try it out!

Execute the command below in a new terminal window within the virtualenv, and wait some seconds to let Celery kick in:

python -m celery --app app.worker.celery_app worker --beat -l info -c 1

Soon, you will see that the tasks are scheduled, and the database is slowly
filling.

Checking in on what we've built so far

Before proceeding to the visualization steps, let's have a look at what we have built so far:

we created the project root
a docker-compose.yml file to manage related services
app/settings.py that handles our application configuration
app/db.py configuring the database pool, and
last but not least, app/worker.py that handles the hard work, fetches, and stores the data.

At this point, we should have the following project structure:

├── app
│   ├── __init__.py
│   ├── db.py
│   ├── settings.py
│   └── worker.py
├── .env
└── docker-compose.yml

Visualize the data with Plotly and Dash

Getting static assets

This tutorial is not about writing the necessary style sheets or collecting static assets, so you only need to copy-paste some code. As the first step, create an assets directory next to the app package with the structure below:

├── app
│   ├── __init__.py
│   ├── db.py
│   ├── settings.py
│   └── worker.py
├── assets
├── .env
├── docker-compose.yml

The style.css will define the styling for our application. As mentioned above, Dash will save us from boilerplate code, so the assets directory will be used by default in conjunction with the stylesheet in it.

Download the style.css file to the assets directory, this can be done using curl:

curl -s -Lo ./assets/style.css https://raw.githubusercontent.com/gabor-boros/questdb-stock-market-dashboard/main/assets/style.css

Setting up the application

This is the most interesting part of the tutorial. We are going to visualize the data we collect. Create a main.py file in the app package, and let's begin with the imports:

# app/main.py

from datetime import datetime, timedelta

import dash
import pandas
from dash import dcc, html
from dash.dependencies import Input, Output
from plotly import graph_objects

from app.db import pool
from app.settings import settings

# [...]

After having the imports in place, we are defining some helper functions and constants.

# [...]

GRAPH_INTERVAL = settings.graph_interval * 1000

TIME_DELTA = 5  # last T hours of data are looked into as per insert time

COLORS = [
    "#1e88e5",
    "#7cb342",
    "#fbc02d",
    "#ab47bc",
    "#26a69a",
    "#5d8aa8",
]


def now() -> datetime:
    return datetime.utcnow()


def get_stock_data(start: datetime, end: datetime, stock_symbol: str):
    def format_date(dt: datetime) -> str:
        return dt.isoformat(timespec="microseconds") + "Z"

    query = f"quotes WHERE ts BETWEEN '{format_date(start)}' AND '{format_date(end)}'"

    if stock_symbol:
        query += f" AND stock_symbol = '{stock_symbol}' "

    with pool.connection() as conn:
        return pandas.read_sql_query(query, conn)

# [...]

In the first few lines, we define constants for setting a graph update frequency (GRAPH_INTERVAL) and colors that will be used for coloring the graph (COLORS).

After that, we define two helper functions, now and get_stock_data. While now is responsible only for getting the current time in UTC (as Finnhub returns the date in UTC too), the get_stock_data does more. It is the core of
our front-end application, it fetches the stock data from QuestDB that workers inserted.

Define the initial data frame and the application:

# [...]

df = get_stock_data(now() - timedelta(hours=TIME_DELTA), now(), "")

app = dash.Dash(
    __name__,
    title="Real-time stock market changes",
    assets_folder="../assets",
    meta_tags=[{"name": "viewport", "content": "width=device-width, initial-scale=1"}],
)

# [...]

As you can see above, the initial data frame (df) will contain the latest 5 hours of data we have. This is needed to pre-populate the application with some data we have. The application definition app describes the application's title, asset folder, and some HTML meta tags used during rendering.

Create the application layout that will be rendered as HTML. We won't write HTML, we will use Dash's helpers for that:

# [...]

app.layout = html.Div(
    [
        html.Div(
            [
                html.Div(
                    [
                        html.H4("Stock market changes", className="app__header__title"),
                        html.P(
                            "Continually query QuestDB and display live changes of the specified stocks.",
                            className="app__header__subtitle",
                        ),
                    ],
                    className="app__header__desc",
                ),
            ],
            className="app__header",
        ),
        html.Div(
            [
                html.P("Select a stock symbol"),
                dcc.Dropdown(
                    id="stock-symbol",
                    searchable=True,
                    options=[
                        {"label": symbol, "value": symbol}
                        for symbol in df["stock_symbol"].unique()
                    ],
                ),
            ],
            className="app__selector",
        ),
        html.Div(
            [
                html.Div(
                    [
                        html.Div(
                            [html.H6("Current price changes", className="graph__title")]
                        ),
                        dcc.Graph(id="stock-graph"),
                    ],
                    className="one-half column",
                ),
                html.Div(
                    [
                        html.Div(
                            [html.H6("Percent changes", className="graph__title")]
                        ),
                        dcc.Graph(id="stock-graph-percent-change"),
                    ],
                    className="one-half column",
                ),
            ],
            className="app__content",
        ),
        dcc.Interval(
            id="stock-graph-update",
            interval=int(GRAPH_INTERVAL),
            n_intervals=0,
        ),
    ],
    className="app__container",
)

# [...]

This snippet is a bit longer, though it has only one interesting part, dcc.Interval. The interval is used to set up periodic graph refresh.

We are nearly finished with our application, but the last steps are to define two callbacks that will listen to input changes and the interval discussed above. The first callback is for generating the graph data and rendering the
lines per stock symbol.

# [...]

@app.callback(
    Output("stock-graph", "figure"),
    [Input("stock-symbol", "value"), Input("stock-graph-update", "n_intervals")],
)
def generate_stock_graph(selected_symbol, _):
    data = []
    filtered_df = get_stock_data(now() - timedelta(hours=TIME_DELTA), now(), selected_symbol)
    groups = filtered_df.groupby(by="stock_symbol")

    for group, data_frame in groups:
        data_frame = data_frame.sort_values(by=["ts"])
        trace = graph_objects.Scatter(
            x=data_frame.ts.tolist(),
            y=data_frame.current_price.tolist(),
            marker=dict(color=COLORS[len(data)]),
            name=group,
        )
        data.append(trace)

    layout = graph_objects.Layout(
        xaxis={"title": "Time"},
        yaxis={"title": "Price"},
        margin={"l": 70, "b": 70, "t": 70, "r": 70},
        hovermode="closest",
        plot_bgcolor="#282a36",
        paper_bgcolor="#282a36",
        font={"color": "#aaa"},
    )

    figure = graph_objects.Figure(data=data, layout=layout)
    return figure

# [...]

The other callback is very similar to the previous one; it will be responsible for updating the percentage change representation of the stocks or a given
stock.

# [...]

@app.callback(
    Output("stock-graph-percent-change", "figure"),
    [
        Input("stock-symbol", "value"),
        Input("stock-graph-update", "n_intervals"),
    ],
)
def generate_stock_graph_percentage(selected_symbol, _):
    data = []
    filtered_df = get_stock_data(now() - timedelta(hours=TIME_DELTA), now(), selected_symbol)
    groups = filtered_df.groupby(by="stock_symbol")

    for group, data_frame in groups:
        data_frame = data_frame.sort_values(by=["ts"])
        trace = graph_objects.Scatter(
            x=data_frame.ts.tolist(),
            y=data_frame.percent_change.tolist(),
            marker=dict(color=COLORS[len(data)]),
            name=group,
        )
        data.append(trace)

    layout = graph_objects.Layout(
        xaxis={"title": "Time"},
        yaxis={"title": "Percent change"},
        margin={"l": 70, "b": 70, "t": 70, "r": 70},
        hovermode="closest",
        plot_bgcolor="#282a36",
        paper_bgcolor="#282a36",
        font={"color": "#aaa"},
    )

    figure = graph_objects.Figure(data=data, layout=layout)
    return figure

# [...]

The last step is to call run_server on the app object when the script is called from the CLI.

# [...]

if __name__ == "__main__":
    app.run_server(host="0.0.0.0", debug=settings.debug)

We are now ready to try our application with actual data. Make sure that the Docker containers are started and execute PYTHONPATH=. python app/main.py from the project root:

$ PYTHONPATH=. python app/main.py

Dash is running on http://0.0.0.0:8050/

 * Tip: There are `.env` or `.flaskenv` files present. Do "pip install python-dotenv" to use them.
 * Serving Flask app 'main' (lazy loading)
 * Environment: production
   WARNING: This is a development server. Do not use it in a production deployment.
   Use a production WSGI server instead.
 * Debug mode: off
 * Running on all addresses.
   WARNING: This is a development server. Do not use it in a production deployment.
 * Running on http://192.168.0.14:8050/ (Press CTRL+C to quit)

Navigate to http://127.0.0.1:8050/, to see the application in action.

To select only one stock, in the dropdown field choose the desired stock symbol and let the application refresh.

Summary

In this tutorial, we've learned how to schedule tasks in Python, store data in QuestDB, and create beautiful dashboards using Plotly and Dash. Although we won't start trading just right now; this tutorial demonstrated well how to combine these separately powerful tools and software to create something bigger and more useful. Thank you for your attention!

The source code is available at
https://github.com/gabor-boros/questdb-stock-market-dashboard.

Using Telegraf and QuestDB to store metrics in a time series database

Gábor Boros — Mon, 12 Jul 2021 18:44:46 +0000

Telegraf is a plugin-driven server agent for collecting, processing, aggregating, and writing metrics. With more than 200 plugins, it can collect almost any kind of data about the server it is running on, application data or even filesystem changes.

Although Telegraf can collect an exceptional amount and variety of data, we need to store and visualize this information at some point. Considering that we collect the metrics over time, a convenient way to store time series data is using a time series database. We'll use QuestDB for ingestion and perform some basic visualization for this tutorial.

Multiple telegraf clients and out-of-order data

When you use multiple clients, it can happen that data coming from various sources simultaneously can arrive out-of-order by time. QuestDB used to have the downside of dropping this kind of out-of-order data. The QuestDB team solved this as of the 6.0 release, meaning there is no need to apply any workarounds like sorting data ourselves before inserting.

This tutorial will set up multiple virtual machines, install Telegraf, QuestDB and experiment with how we can visualize the incoming data about server status (load, CPU, swap, and memory usage) over time.

Prerequisites

Celebrating the recent public market debut of DigitalOcean and QuestDB's marketplace offering, we are going to join the celebration. Therefore, we will need the following resources for the tutorial:

A DigitalOcean account (get $100 credit for free by signing up using this link)
Basic shell knowledge
Basic knowledge of vi/vim/nano or any terminal-based text editors

Enough talking, let's jump right in and create our Droplets!

Setting up DigitalOcean droplets

The resources we will create are:

1 x QuestDB Droplet for storing and visualizing metrics.
2 x Droplets running the Telegraf agent collecting system metrics.

Create a QuestDB Droplet

Let's get started with the database Droplet. DigitalOcean has an excellent marketplace offering preinstalled, so-called, 1-Click Apps reviewed by its staff. QuestDB is available on the marketplace; therefore its setup is less than 30 seconds:

Navigate to the marketplace listing
Click on "Create QuestDB Droplet"
Select the basic plan for your Droplet and the desired resources (use at least 4GB RAM to avoid slow queries)
Choose the region of your choice that is the closest to you
At the "Authentication" section, select your SSH key or set a password for the Droplet's root account
Set the hostname to telegraf-questdb-tutorial
Leave all other settings with their defaults, and click "Create Droplet" at the bottom of the page

In about 30 seconds, QuestDB is ready to use. To validate that we set everything up successfully, copy the Droplet's IP address by clicking on it and navigate to http://<IP ADDRESS>:9000/ where <IP ADDRESS> is the IP address you just copied. The interactive console should load and we can start querying the database and inserting data!

Create DigitalOcean Droplets with Telegraf agent

We don't have any data in the database to query yet, so the next steps are to send some metrics to QuestDB for inspection. Scripts that create dummy data are always good to get started, but in this case, let's use some actual data collected on demo machines, so we have proper metrics to play with instead of synthetic data.

QuestDB exposes a reader for InfluxDB line protocol which allows using QuestDB as a drop-in replacement for InfluxDB and other systems which implement this protocol. – questdb.io

We will utilize InfluxDB line protocol to send data via Telegraf to QuestDB directly. The next step is to create some Droplets, start Telegraf agents, and point them to QuestDB. Create the Droplets following these steps:

Navigate to the Droplets dashboard
In the top-right section of the page, click "Create" and select "Droplets"
At the "Choose an image" section, select Ubuntu (20.04 LTS x64 at the time of writing)
Select the basic plan for your Droplet and the minimum resource type
Choose a region of your choice that is the closest to you
At the "Authentication" section, select your SSH key or set a password for the Droplet's root account
Set the number of Droplets to 2
Set the hostname to telegraf-agent-1 and telegraf-agent-2
Leave all other settings with their defaults, and click "Create Droplet" at the bottom of the page

Compared to the previous Droplet creation, DigitalOcean will create two Droplets instead of one. In a few seconds, the Droplets are ready to start up the Telegraf agent.

Configuring Telegraf to send metrics to QuestDB

In this section, we will install the Telegraf agent on all three Droplets. To install Telegraf, we will follow the official installation method.

Installing Telegraf on the QuestDB Droplet

First of all, login to telegraf-questdb-tutorial Droplet by executing ssh root@<IP ADDRESS> where <IP ADDRESS> is the Droplet's IP address. Then, on the server, run the following to make the Telegraf client available for installation.

# Download the signing keys from influxdata.com
curl -s https://repos.influxdata.com/influxdb.key | apt-key add -

# Source release information
source /etc/lsb-release

# Add influxdata.com APT repository to the APT repository list
echo "deb https://repos.influxdata.com/${DISTRIB_ID,,} ${DISTRIB_CODENAME} stable" | tee /etc/apt/sources.list.d/influxdb.list

# Fetch available repositories and read package lists
apt-get update

We are running the above commands to add the APT repository to our list of available repositories. Now, we can install the agent as we would do with any packages by executing apt-get install -y telegraf.

The agent is installed but not configured yet. To configure it, let's create a new configuration file at /etc/telegraf/telegraf.d/questdb.conf with the following content:

# Configuration for Telegraf agent
[agent]
  ## Default data collection interval for all inputs
  interval = "5s"

# Write results to QuestDB
[[outputs.socket_writer]]
  # Write metrics to a local QuestDB instance over TCP
  address = "tcp://127.0.0.1:9009"

# Read metrics about CPU usage
[[inputs.cpu]]

# Read metrics about memory usage
[[inputs.mem]]

# Read system statistics, like load on the server
[[inputs.system]]

After saving the configuration file, we have one thing left to do: restart Telegraf by running systemctl restart telegraf. In 5 seconds, the agent will start reporting to QuestDB.

Installing Telegraf on demo Droplets

Lastly, install Telegraf on the remaining Droplets. As you may expect, we have to perform the same process as the QuestDB droplet. SSH into both Droplets, telegraf-agent-1 and telegraf-agent-2.

Add the necessary signing keys and prepare the local APT repository list:

# Download the signing keys from influxdata.com
curl -s https://repos.influxdata.com/influxdb.key | apt-key add -

# Source release information
source /etc/lsb-release

# Add influxdata.com APT repository to the APT repository list
echo "deb https://repos.influxdata.com/${DISTRIB_ID,,} ${DISTRIB_CODENAME} stable" | tee /etc/apt/sources.list.d/influxdb.list

# Fetch available repositories and read package lists
apt-get update

Install Telegraf by executing apt-get install -y telegraf and edit the configuration file /etc/telegraf/telegraf.d/reporter.conf as the following:

Note that below, we set the socket_writer address in the configuration to <QUESTDB IP ADDRESS>, which is the IP address of the QuestDB Droplet.

# Configuration for Telegraf agent
[agent]
  ## Default data collection interval for all inputs
  interval = "5s"

# Write results to QuestDB
[[outputs.socket_writer]]
  # Write metrics to a local QuestDB instance over TCP
  address = "tcp://<QUESTDB IP ADDRESS>:9009"

# Read metrics about CPU usage
[[inputs.cpu]]

# Read metrics about memory usage
[[inputs.mem]]

# Read system statistics, like load on the server
[[inputs.system]]

Restart the Telegraf agents with systemctl restart telegraf just like the QuestDB Droplet; in a few seconds, the agents will start reporting to our database.

Run SQL queries on Telegraf metrics

At this point, we have every component set up and running to visualize some incoming data. Navigate to http://<QUESTDB IP ADDRESS>:9000 where the <QUESTDB IP ADDRESS> is the IP address of your QuestDB droplet, and write the following SQL statement in the SQL editor:

SELECT * FROM cpu

This will return all data in the table for CPU metrics sent by Telegraf. We can easily create some aggregates like the average CPU usage per machine:

SELECT 
    host,
    avg(usage_system) cpu_average,
    timestamp
FROM cpu

If we want to perform some more complex queries, we can perform JOINs across the three tables:

SELECT 
    cpu.host,
    avg(mem.used_percent) mem_usage_average,
    avg(cpu.usage_system) cpu_average,
    avg(system.load1) load1_average,
    cpu.timestamp as timestamp 
FROM cpu
INNER JOIN mem ON mem.host = cpu.host
INNER JOIN system ON system.host = cpu.host
SAMPLE BY 5m
ORDER BY timestamp DESC

And to visualize this data, query memory usage using the following SQL:

SELECT 
    host,
    avg(mem.used_percent) usage_average,
    timestamp 
FROM mem
WHERE host = 'telegraf-agent-1'
SAMPLE BY 30s
ORDER BY timestamp DESC

The basic in-built charting functionality that QuestDB has can be used like so:

Click the Chart tab
Set Chart type to line
Set Labels to timestamp
Plot usage_average as a Series and click Draw

Summary

We've installed QuestDB on DigitalOcean using the new Marketplace offering for QuestDB, set up multiple Droplets to report actual metrics to QuestDB via Telegraf, and visualized these metrics on the interactive console. This tutorial shows how easy it is to send real system metrics as time series data to a database like QuestDB for reports and visualization.

For next steps, we can experiment with some of the additional integrations that Telegraf supports to grab insights from other applications and use the Grafana integration that QuestDB offers to set up more detailed dashboards for visualization or even alerting and notifications.

Thank you for your attention!

Automating ETL jobs on time series data with QuestDB on Google Cloud Platform

Gábor Boros — Mon, 29 Mar 2021 17:26:47 +0000

In the world of big data, software developers and data analysts often have to write scripts or complex software collections to process data before sending it to a data store for further analysis. This process is commonly called ETL, which stands for Extract, Transform and Load.

What could we use ETL jobs for?

Let's consider the following example: a medium-sized webshop with a few thousand orders per day exports order information hourly. After a while, we would like to visualize purchase trends, and we might want to share the results between departments or even publicly. Since the exported data contains personally identifiable information (PII), we should anonymize it before using or exposing it to the public.

For the example above, we can use an ETL job to extract the incoming data, remove any PII and load the transformed data into a database used as the data visualization backend later.

Prerequisites

During this tutorial, we will use Python to write the cloud functions, so basic python knowledge is essential. Aside from these skills, you will need the following resources:

A Google Cloud Platform (GCP) account and a GCP Project.
Enable the Cloud Build API - when enabling APIs ensure that the correct GCP project is selected.

Creating an ETL job

As an intermediate datastore where the webshop exports the data, we will use Google Storage and use Google Cloud Functions to transform it before loading it into QuestDB.

We won't be building a webshop or a data exporter for an existing webshop, but we will use a script to simulate the export to Google Storage.

In the following sections, we will set up the necessary components on GCP. Ensure the required APIs mentioned in the prerequisites are enabled, and that you have selected the GCP project in which you would like to create the tutorial resources.

Create a Compute Engine instance for QuestDB

First things first, we start with installing QuestDB on a virtual machine. To get started, navigate to the Compute Engine console. Visiting this page for the first time will take a few moments to initialize. After the loading indicator has gone, start a new virtual machine:

Click on "create" and give the instance the name questdb-vm
Select a region close to you
Select the first generation "N1" series
Choose the f1-micro machine type - in a production environment you would choose a more performant instance, but for tutorial purposes, this is enough
In the "Container section, check "Deploy a container image to this VM instance", and type "questdb/questdb:latest" for the "Container image"
In the "Firewall" section, click on "Management, security, disks, networking, sole tenancy"
In the newly appeared panel, select "Networking" and add questdb as a "Network tag"
Leave all other settings with their defaults, and click on "create"

Make sure you note the "External IP" of the instance as we will need that later.

After a short time, the new instance will be up and running. As soon as the instance is provisioned, we can initiate a remote session to install QuestDB by clicking ssh in the VM panel.

Allow networking on the instance

If we try to open the web console by opening the http://<EXTERNAL_IP>:9000 (where <EXTERNAL_IP> is the external IP of your virtual machine) it won't load and we will face a timeout. The reason behind this is that the firewall is not opened for port 9000 yet.

To allow port 9000 used by QuestDB, we must allow the port by adding a new firewall rule on the firewall console:

Click on "create firewall rule" at the top of the page
Give the rule the name "QuestDBPorts"
In the "Target tags" field, write the same tag used for instance creation (questdb)
For the "Source IP ranges" field, set 0.0.0.0/0
In the "Protocols and ports" section, select tcp and set port to 9000,8812
Click on "create"

Some seconds later, the rule will be applied on every instance with the matching questdb tag and port 9000 will be open. You may ask what port 8812 is for; this port will be used by the Cloud Function later to connect to the database.

If you try to open the interactive console again, you should see the QuestDB Web Console and start writing queries.

As our first query, create the table in which the Cloud Function will write the anonymized data. To create the table run the following SQL statement:

CREATE TABLE
    purchases(buyer STRING, item_id INT, quantity INT, price INT, purchase_date TIMESTAMP)
    timestamp(purchase_date);

The query above uses timestamp(purchase_date) to set a designated timestamp on the table so we can easily perform time series analysis in QuestDB.
For more information on designated timestamps, see the official QuestDB documentation for timestamp.

Create a Storage bucket

Now, we create the bucket where we will store the simulated webshop data. Storage buckets are in a single global namespace in GCP, which means that the bucket's name must be unique across all GCP customers. You can read more about Storage and buckets on Google's documentation site.

To create a new bucket:

Navigate to the cloud storage console
Select your project if not selected yet
Click on "create bucket" and choose a unique name
Select the same region as the instance above
Leave other settings on default and click on "continue" to create the bucket

If you successfully created the bucket, it should show up in the storage browser as you can see below.

At this point, we don't set any permissions, ACLs, or visibility settings on the bucket, but we will come back to that later.

Create a Cloud Function

We have the bucket to upload the data, but we have nothing to process the data yet, and for this, we will use Cloud Functions to remove the PII.

Cloud Functions are functions as a service (FaaS) solution within GCP, similar to AWS Lambda. The functions are triggered by an event that can come from various sources. Our scenario Cloud Functions are convenient since we don't need to pay for a server to run all day, which is mostly idle; the function will be executed when the trigger event is fired, and we only pay for the execution time the number of function calls.

To create a Cloud Function:

Navigate to cloud functions console
Click on "create function" and give it the name remove-pii
Select the region we are using for other resources
For "Trigger" select "Cloud Storage" from the dropdown list
Set the event type to "Finalise/Create"
Choose the bucket created above and click "variables, networking, and advanced settings"
Select "environment variables" on the tabbed panel
Click on "add variable" right below the "Runtime environment variables" section
Add a new variable called DATABASE_URL with the value postgresql://admin:quest@<EXTERNAL_IP>:8812/qdb, where <EXTERNAL_IP> is the external IP of your virtual machine
Click "save" then "next"

The next step is to select the runtime our function will use and provide the code. On this page, we can choose between numerous runtimes, including multiple versions of Python, NodeJS, Go, Ruby, and even Java.

Since this tutorial uses Python, select Python 3.8 as it is the latest non-beta version at the time of writing. Leave the rest of the settings as default, and write the function in the next section. Click "deploy" at the bottom of the page. Some seconds later, you will see that the deployment of the function is in progress.

The deployment may take a while, so we can move on to the next section of the tutorial.

Generating and processing data

Before moving on, here's a quick recap on what we did so far:

Set up a new Google Storage bucket
Created the Cloud Function which will process the data later on
Connected the bucket with the function to trigger on a new object is created on the bucket

Now for the fun part of the tutorial: writing the data processing script and loading the data in the database. Let's write the function to remove PII, but first, talk a bit about the data's structure.

Inspect the data structure

ETL jobs, by their nature, heavily depend on the structure of incoming data. A job may process multiple data sources, and data structure can vary per source.
The data structure we will use is simple. We have a CSV file with the following information:

purchase date
email address
purchased item's ID
quantity
the price per item

As you see, there is no currency column since we will assume every price is in one currency.

To generate random data, you can use the pre-made script written for this tutorial.

Create the data transformer function

By now, we have everything to write the data transformer function, connect the dots and try out the PII removal.

We will work in the "inline editor" of the cloud function, so as a first step, open the edit the cloud function created above by navigating to the cloud functions console and clicking on the function's name. That will open the details of the function. At the top, click on "edit", then at the bottom, click on "next" to open the editor.

Let's start with the requirements. On the left-hand side, click on the requirements.txt and paste the following:

google-cloud-storage==1.36.2
psycopg2==2.8.6
sqlalchemy==1.4.2

Here we add the required packages to connect to Google Storage and QuestDB. Next, click on main.py, remove its whole content and start adding the following:

import csv
import hashlib
import json
import logging
import os
from dataclasses import dataclass
from datetime import datetime
from typing import List

from google.cloud import storage
from sqlalchemy.sql import text
from sqlalchemy.engine import Connection, create_engine

logger = logging.getLogger(__name__)

# Create a database engine
engine = create_engine(os.getenv("DATABASE_URL"))

# ...

As you may expect, we start with the imports, but we added two extra lines: one for the logger and one for configuring the database engine. We will need the logger to log warnings and exceptions during the execution, while we will use the engine later to insert anonymized data into the database.

To make our job easier, we are going to add a data class, called Record. This data class will be used to store the parsed and anonymized CSV data for a line of the uploaded file.

# ...

@dataclass
class Record:
    buyer: str
    item_id: int
    quantity: int
    price: int
    purchase_date: datetime

# ...

As we discussed, ETL jobs are validating the data that they receive as input. In our case, we will trigger the function if an object is created on the storage. This means any object, like a CSV, PDF, TXT, PNG file, or even a directory, is created, though we only want to execute CSV files' transformation. To validate the incoming data, we write two simple validator functions:

# ...

def is_event_valid(event: dict) -> bool:
    """
    Validate that the event has all the necessary attributes required for the
    execution.
    """

    attributes = event.keys()
    required_parameters = ["bucket", "contentType", "name", "size"]

    return all(parameter in attributes for parameter in required_parameters)


def is_object_valid(event: dict) -> bool:
    """
    Validate that the finalized/created object is a CSV file and its size is
    greater than zero.
    """

    has_content = int(event["size"]) > 0
    is_csv = event["contentType"] == "text/csv"

    return has_content and is_csv

# ...

The first function will validate that event has all the necessary parameters, while the second function checks that the object created and triggered the event is a CSV and has any content.
The next function we create is used to get an object from the storage which, in our case, the file triggered the event:

# ...

def get_content(bucket: storage.Bucket, file_path: str) -> str:
    """
    Get the blob from the bucket and return its content as a string.
    """

    blob = bucket.get_blob(file_path)
    return blob.download_as_string().decode("utf-8")

# ...

Anonymizing the data, in this scenario, is relatively easy, though we need to ensure we can build statistics and visualizations later based on this data, so the anonymized parts should be consistent for a user. To achieve this, we will hash the buyer's email address, so nobody may track it back to the person owning the email, but we can use it for visualization:

# ...

def anonymize_pii(row: List[str]) -> Record:
    """
    Unpack and anonymize data.
    """

    email, item_id, quantity, price, purchase_date = row

    # Anonymize email address
    hashed_email = hashlib.sha1(email.encode()).hexdigest()

    return Record(
        buyer=hashed_email,
        item_id=int(item_id),
        quantity=int(quantity),
        price=int(price),
        purchase_date=purchase_date,
    )

# ...

So far, we have functions to validate the data, get the file's content which triggered the Cloud Function, and anonymize the data. The next thing we need to be able to do is to load the data into our database. Up to this point, every function we have wrote was simple, and this one is no exception:

# ...

def write_to_db(conn: Connection, record: Record):
    """
    Write the records into the database.
    """

    query = """
    INSERT INTO purchases(buyer, item_id, quantity, price, purchase_date)
    VALUES(:buyer, :item_id, :quantity, :price, to_timestamp(:purchase_date, 'yyyy-MM-ddTHH:mm:ss'));
    """

    try:
        conn.execute(text(query), **record.__dict__)
    except Exception as exc:
        # If an error occures, log the exception and continue
        logger.exception("cannot write record", exc_info=exc)

# ...

As you see, writing to the database is easy. We get the connection and the record we need to write into the database, prepare the query and execute it. In case of an exception, we don't want to block the whole processing, so we catch the exception, log it and let the script go on. If an exception occurred, we can check it later and fix the script or load the data manually.

The last bit is the glue code, which brings together these functions. Let's have a look at that:

# ...

def entrypoint(event: dict, context):
    """
    Triggered by a creation on a Cloud Storage bucket.
    """

    # Check if the event has all the necessary parameters. In case any of the
    # required parameters are missing, return early not to waste execution time.
    if not is_event_valid(event):
        logger.error("invalid event: %s", json.dumps(event))
        return

    file_path = event["name"]

    # Check if the created object is valid or not. In case the object is invalid
    # return early not to waste execution time.
    if not is_object_valid(event):
        logger.warning("invalid object: %s", file_path)
        return

    storage_client = storage.Client()
    bucket = storage_client.get_bucket(event["bucket"])

    data = get_content(bucket, file_path)
    reader = csv.reader(data.splitlines())

    # Anonymize PII and filter out invalid records
    records: List[Record] = filter(lambda r: r, [anonymize_pii(row) for row in reader])

    # Write the anonymized data to database
    with engine.connect() as conn:
        for record in records:
            write_to_db(conn, record)

In the example above, we call the two validators to ensure it worth processing the data, and we get the file path from the event. After that we initialize the client used to connect to Google Storage, then we get the object's content, parse the CSV file, and anonymize the content of it.

Last but not least, we connect to the database - defined by the DATABASE_URL configured for the engine and write all records to the database one by one.

As you see, the entrypoint of the function has been changed as well. In the text box called "Entrypoint" set the entrypoint as a function name to call. The entrypoint is the function that will be called by Cloud Functions when an event is triggered.

Connecting the services

We are close to finishing this tutorial, so it's time to test our Cloud Function.

To test the Cloud Function:

Download the pre-made script and run it to generate random data.
Navigate to the bucket you created
Above the list of objects in the bucket (which should be empty) click on "upload files"
Select and upload the random generated data *1
After the file is uploaded, go to Cloud Functions console
Click on the actions button and select "view logs"
Calidate that the script did not encounter any issues
Navigate to http://<EXTERNAL_IP>:9000, where <EXTERNAL_IP> is the external IP of your virtual machine

We can now execute the following SQL query:

SELECT * FROM purchases ORDER BY purchase_date;

As you can see, the data is loaded and we have no PII there. By creating a simple chart, we can even observe trends in the generated data, how our imaginary buyers purchased items on the webshop.

*1 QuestDB at the time of writing does not support "out of order" writes. This means you need to upload data with delay to let the previous function finish processing the data. Also, the uploaded purchase data must be in time order and increasing across the uploads. Example: We are uploading data1.csv and data2.csv the last generated purchase data in data1.csv is 2021-03-21T11:59:49, therefor data2.csv's first purchase order must be greater than or equal to 2021-03-21T11:59:49.

Summary

We've installed QuestDB on Google Cloud Platform, set up a Google Storage bucket to store the simulated purchase data exports, built an ETL job that anonymized our buyers' data, and loaded it into a time series database, QuestDB. Data analysts could write more jobs as Cloud Functions in multiple languages and set up multiple sources. Furthermore, this data could be loaded into a business intelligence (BI) dashboard like Power BI to have a more comprehensive overview of the data as it does not contains PII anymore.

Thank you for your attention!

The source code is available at https://github.com/gabor-boros/questdb-etl-jobs.

Monitoring the uptime of an application with Python, Nuxt.js and QuestDB

Gábor Boros — Fri, 15 Jan 2021 21:52:44 +0000

Highly available services that serve millions of requests rely on the visibility of the system status for customers and internal teams.
This tutorial shows how a lightweight and performant time-series database coupled with queued status checks and a simple UI are key ingredients for robust application monitoring.

Why build a status page for an application?

Even if we design the most reliable systems, incidents will occur for hard-to-predict reasons.
It's critical to provide as much information as possible to users, customers, and service teams.
The most convenient way to display this is through a status page.

Although the page's responsibility is to provide information, it can reduce the support team's load and eliminate duplicate support tickets.
Status pages are a crucial part of incident management, and usually, other teams enjoy benefits like client and service owners when they need to refer to SLAs.
In this tutorial, I'll show you how to build a simple yet powerful status page that scores well on performance and design.

What we will build

Overview

As mentioned above, we will build a simple status page made of two parts: the backend monitors our service, and a frontend shows our services' status on an hourly scale.

You will need some experience in Python, JavaScript, and basic SQL knowledge. To build our service, we will use FastAPI, an ultra-fast Python web framework, Celery for scheduling monitoring tasks, QuestDB, the fastest open-source time-series database, to store monitoring results, and NuxtJs to display them.

There's a lot to learn, so let's jump right in!

The containerized source code is available at https://github.com/gabor-boros/questdb-statuspage.

Prerequisites

You will need to have the following installed on your machine:

Python 3.8
NodeJS 14.x
Docker
Docker Compose

Setting up the environment

Create a new project

First things first, we create a directory, called status-page, this is our project root.
We also need to create another directory called app which will contain the backend code.
After following these steps, you should have a project structure like this.

status-page (project root)
└── app (backend service directory)

Installing QuestDB & Redis

Now, we will install QuestDB and Redis. QuestDB is used to store the HTTP status and the service state of our application over time, and Redis is used as a message broker between the backend application and the workers who will do the scheduled monitoring.

To install these services, we will use Docker and Docker Compose. We are going to create a docker-compose.yml file within the project root with the following content:

version: '3'

volumes:
  questdb_data: {}

services:
  redis:
    image: 'redis:latest'
    ports:
      - '6379:6379'

  questdb:
    image: 'questdb/questdb:latest'
    volumes:
      # Map QuestDB's data directory to the host
      - 'questdb_data:/root/.questdb/db'
    ports:
      - '9000:9000'
      - '8812:8812'

Voila! When we run docker-compose up, QuestDB and Redis start, and we can access QuestDB's interactive console on http://127.0.0.1:9000.

Install backend dependencies

Now, we have the project structure, and we can run the required services, so we need to set up our backend service to collect data about the website or service we would like to monitor. We will use poetry to manage Python dependencies during this tutorial, so let's start by installing that.

$ pip install poetry

To define the project requirements, create a pyproject.toml file with the following content:

[tool.poetry]
name = "status-page"
version = "0.1.0"
description = "QuestDB tutorial for creating a simple status page."
authors = ["Your name <your.email@example.com>"]
license = "MIT"

[tool.poetry.dependencies]
python = "^3.8"

[build-system]
requires = ["poetry-core>=1.0.0"]
build-backend = "poetry.core.masonry.api"

In order to let the application communicate with QuestDB utilizing the psycopg client library, we need to install libpq-dev package on our system. To install it, use your package manager; on Windows, you may need to install PostgreSQL on your system.

Then, install the project dependencies by executing the following:

poetry add python fastapi pydantic uvicorn requests \
 "psycopg[c,pool]" "celery[redis]"

As you may assume by checking the requirements, we will use QuestDB's Postgres interface to connect.
When poetry finishes its job, it will add the dependencies to pyproject.toml and we can now start to implement the backend service.

Create a simple API

The time has come, let's create the backend service, but step-by-step.

Within the app directory, create an __init__.py and main.py. The first one is responsible for making the app directory to a package, while the latter will define the APIs our service exposes. Open main.py for edit and add the following:

# main.py

from fastapi import FastAPI

app = FastAPI(
    title="Status Page",
    description="This service gives back the status of the configured URL.",
    version="0.1.0",
)

Congratulations! You just created the backend service. You can go and try it out by executing:

$ poetry run uvicorn app.main:app --host 127.0.0.1 --port 8000 --reload
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
...
INFO:     Application startup complete.

Although the service does nothing yet, it works and listens for any code change. Add a new endpoint and watch it reload:

# main.py

# ...

@app.get(path="/signals", tags=["Monitoring"])
async def get_signals():
  return {}

We have now created an API endpoint that will serve the system status data of the monitored URL.
If you open http://127.0.0.1:8000/redoc, you can see the generated documentation for the endpoint, or you can check it working at http://127.0.0.1:8000/signals, though it won't return any data yet.

It is time to have fun, we are going to integrate QuestDB with our shiny new backend service.

Integrate QuestDB with FastAPI

Integrating QuestDB with FastAPI is easier than you think. Thanks to QuestDB's Postgres compatibility, you can use any standard or popular third-party libraries of any programming language which implements Postgres wire protocol.

Set up the table

The very first step is to create the table in QuestDB. As said before, our approach is simple, so that the table is simple, too. QuestDB is running from our docker compose script so, we open the interactive console at http://127.0.0.1:9000 and create a new table by running the following query:

CREATE TABLE
    signals(url STRING, http_status INT, received TIMESTAMP, available BOOLEAN)
    timestamp(received);

The query executes, and after refreshing the table list on the left, you can see the table we created.

Connect QuestDB and FastAPI

As we have the table in the database, it is time to connect to QuestDB and query some data to return through the API. To connect, we will use the Postgres interface of QuestDB and Postgres using connection pooling to connect to it.

To be able to reuse the pool later on, create a new file in the app package which is responsible for defining how to connect and name it db.py:

# db.py

from psycopg_pool import ConnectionPool

pool = ConnectionPool(
    "postgresql://admin:quest@127.0.0.1:8812/qdb", # Use a the default credentials
    min_size=1, # Set pool size minimum to 1
    max_size=5, # Set pool size greater than 1 to not block async requests
)

To set up a schema that represents the table in the database, create a models.py in the app package, containing the schema definition:

# models.py

from datetime import datetime
from pydantic import BaseModel, Field

class Signal(BaseModel):
    """
    Signal model stands for the results of monitoring requests
    """
    url: str = Field(..., description="The monitored URL")
    http_status: int = Field(..., description="HTTP status code returned by upstream")
    available: bool = Field(..., description="Represents the service availability")
    received: datetime = Field(..., description="Timestamp when the signal received")

Let's stop here for a moment and talk through what we did in the last steps:

set up the API which will serve the requests coming from the frontend
created a table in QuestDB for our status records and provided connection credentials for Postgres wire
implemented the schema which is used to serialize the results returned by the database

The next step is to initiate a connection and return the results from the database.
First, import the pool and Signal schema and then extend the function which serves the /signals endpoint:

# main.py

# Other imports ...
from app.db import pool
from app.models import Signal

from typing import List
from pydantic import BaseModel

class SignalResponse(BaseModel):
    url: str
    records: List[Signal]

# app and endpoint definitions are below...

After adding the defaultdict import, the implementation of the /signals endpoint should look like this:

# main.py

# Other imports ...
from collections import defaultdict

# ...

@app.get(path="/signals", response_model=List[SignalResponse], tags=["Monitoring"])
async def get_signals(limit: int = 60):

    # A simple query to return every record belongs to the website we will monitor
    query = f"""
    SELECT url, http_status, available, received FROM signals
    WHERE url = 'https://questdb.io' ORDER BY received DESC LIMIT {limit};
    """

    signals = defaultdict(list)

    with pool.connection() as conn: # connect to the database
        # parse the results results returned by QuestDB
        signal = Signal(
            url=result[0],
            http_status=result[1],
            available=result[2],
            received=result[3],
        )
    # Return the response which is validated against the `response_model` schema
    return [
        SignalResponse(url=url, records=list(reversed(records)))
        for url, records in signals.items()
    ]

Let's recap on our code above, starting from the top:

we added defaultdict import (we'll explain that later)
extended the function decorator to use response_model=List[SignalResponse], the response model we defined already
changed the function signature to include a limit parameter and set its default value to 60 since we will monitor HTTP status every minute
select the records from the database and prepare a dictionary for the parsed Signals.

You may ask why to group the returned records per URL. Although we will monitor only one URL for the sake of simplicity, I challenge you to change the implementation later and explore QuestDB to handle the monitoring of multiple URLs.

In the following lines, we are connecting to the database, executing the query, and populates the dictionary, which we will use in the last four lines to construct the SignalResponse.
Our version of main.py at this point looks like the following:

# main.py

from collections import defaultdict
from typing import List

from fastapi import FastAPI
from pydantic import BaseModel

from app.db import pool
from app.models import Signal


# Add a response model to indicate the structure of the signals API response.
class SignalResponse(BaseModel):
    url: str
    records: List[Signal]


app = FastAPI(
    title="Status Page",
    description="This service gives back the status of the configured URL.",
    version="0.1.0",
)

@app.get(path="/signals", response_model=List[SignalResponse], tags=["Monitoring"])
async def get_signals(limit: int = 60):

    # A simple query to return every record belongs to the website we will monitor
    query = f"""
    SELECT * FROM signals
    WHERE url = 'https://questdb.io' ORDER BY received DESC LIMIT {limit};
    """

    signals = defaultdict(list)

    with pool.connection() as conn: # connect to the database
        # parse the results results returned by QuestDB
        signal = Signal(
            url=result[0],
            http_status=result[1],
            available=result[2],
            received=result[3],
        )

    # Return the response which is validated against the `response_model` schema
    return [
        SignalResponse(url=url, records=list(reversed(records)))
        for url, records in signals.items()
    ]

Schedule monitoring tasks

For scheduling the monitoring task, we will use Celery Beat, the built-in periodic task scheduler implementation of Celery.

Scheduling with Celery

Before we schedule any task, we need to configure Celery. In the app package, create a new celery.py which will contain the Celery and beat schedule configuration. Import Celery for creating tasks, and crontab for constructing Unix-like crontabs for our tasks.
The task is the dotted path representation of the function which is executed by Celery (app.tasks.monitor) and sent to queues handled by Redis.

The only thing left is to configure the beat schedule, which is a simple dictionary.
We give a name for the schedule, define the dotted path pointing to the task (function), and specify the schedule itself:

# celery.py

from celery import Celery
from celery.schedules import crontab

MONITORING_TASK = "app.tasks.monitor"

celery_app = Celery("tasks", broker="redis://localhost:6379/0")

# Set a queue for task routes
celery_app.conf.task_routes = {
  MONITORING_TASK: "main-queue"
}

# Schedule the monitoring task
celery_app.conf.beat_schedule = {
    "monitor": { # Name of the schedule
        "task": MONITORING_TASK, # Register the monitoring task
        "schedule": crontab(
            minute=f"*/1" # Run the task every minute
        ),
    }
}

Create a monitoring task

And the last part: creating the monitoring task. In the previous section, we talked about the "monitoring task" multiple times, but we didn't see the concrete implementation.

In this final backend related section, you will implement the task which will check the availability of the desired website or service and saves the results as records in QuestDB. The monitoring task is a simple HTTP HEAD request and saving the response to the database. We see the implementation in pieces of the app/tasks.py referenced in celery as the dotted path before.

First, we start with imports:

# tasks.py

from datetime import datetime

import requests

from app.celery import celery_app
from app.db import pool
from app.models import Signal

# ...

We import celery_app which represents the Celery application, an engine to save the results in the database, and finally Signal to construct the record we will save. As the necessary imports are in place, we can define the monitor task.

# tasks.py

# ...

@celery_app.task # register the function as a Celery task
def monitor():
    try:
        response = requests.head("https://questdb.io")
    except Exception as exc: # handle any exception which may occur due to connection errors
        query = f"""
            INSERT INTO signals(received,url,http_status,available)
            VALUES(systimestamp(), 'https://questdb.io', -1, False);
            """

        # Open a connection and execute the query
        with pool.connection() as conn:
            conn.execute(query)

        # Re-raise the exception to not hide issues
        raise exc

# ...

As you can see, we send a request to the desired website and store the response for later use. In case the website is down and unreachable, an exception will be raised by requests or any underlying packages. As we need to log that the request does not finish, we catch the exception, save a record in the database, and re-raise the exception to not hide anything. Next, we construct a signal to save.

# ...

@celery_app.task
def monitor():
    # ...

        signal = Signal(
        url="https://questdb.io",
        http_status=response.status_code,
        received=datetime.now(),
        available=response.status_code >= 200 and response.status_code < 400,
    )

    # ...

We don't do anything special here, though the following step is more interesting: inserting the result in the database. Finally, we prepare and execute the query based on the signal.

# ...

@celery_app.task
def monitor():
    # ...

    query = f"""
    INSERT INTO signals(received,url,http_status,available)
    VALUES(systimestamp(), '{signal.url}', {signal.http_status}, {signal.available});
    """

    with engine.connect() as conn:
        conn.execute(query)

The tasks.py should look like this now:

# tasks.py

from datetime import datetime
import requests

from app.celery import celery_app
from app.db import pool
from app.models import Signal

@celery_app.task # register the function as a Celery task
def monitor():
    try:
        response = requests.head("https://questdb.io")
    except Exception as exc: # handle any exception which may occur due to connection errors
        query = f"""
            INSERT INTO signals(received,url,http_status,available)
            VALUES(systimestamp(), 'https://questdb.io', -1, False);
            """
        # Open a connection and execute the query
        with pool.connection() as conn:
            conn.execute(query)
        # Re-raise the exception to not hide issues
        raise exc
    signal = Signal(
        url="https://questdb.io",
        http_status=response.status_code,
        received=datetime.now(),
        available=response.status_code >= 200 and response.status_code < 400,
    )
    query = f"""
    INSERT INTO signals(received,url,http_status,available)
    VALUES(systimestamp(), '{signal.url}', {signal.http_status}, {signal.available});
    """
    with pool.connection() as conn:
        conn.execute(query)

Congratulations! You just arrived at the last part of the backend service implementation. We did many things and built a service that can periodically check the website's status, save it in the database, and expose the results through an API.

The very last thing we need to address is to allow connections initiated by the frontend later on. As it will run on localhost:3000 and we don't use domain names, the port is different hence all requests will be rejected with errors related to Cross-Origin Resource Sharing.

For the sake of simplicity, we are going to allow all origins. We highly recommended to avoid this in production for security reason but for the tutorial, it will be sufficient for us:

# main.py

# Other imports ...
from fastapi.middleware.cors import CORSMiddleware

# app = FastAPI ...

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# rest of the code ...

Implement the frontend

Setting up frontend

To build the frontend, we will use Nuxt.js. We will use yarn to set up the starter project by running yarn and selecting the answers detailed below.

$ yarn create nuxt-app frontend

[...]

? Project name: frontend
? Programming language: JavaScript
? Package manager: Yarn
? UI framework: Tailwind CSS
? Nuxt.js modules: Axios
? Linting tools: (Press <space> to select, <a> to toggle all, <i> to invert selection)
? Testing framework: None
? Rendering mode: Single Page App
? Deployment target: Static (Static/JAMStack hosting)
? Development tools: (Press <space> to select, <a> to toggle all, <i> to invert selection)
? What is your GitHub username? gabor-boros
? Version control system: Git

The project root now looks like this:

status-page
├── app/
├── frontend/
├── docker-compose.yml
├── poetry.lock
└── pyproject.toml

Cleaning up generated project

Since we don't need any styling delivered by the project generation, we need to get rid of them. Open frontend/layouts/default.vue and replace its content with

<!-- frontend/layouts/default.vue -->

<template>
  <div>
    <Nuxt />
  </div>
</template>

Now, we will change frontend/pages/index.vue and call the backend service. Let's begin with <scripts>.

<!-- frontend/pages/index.vue -->

<!-- template ... -->

<script>
const LIMIT = 60;

async function fetchSignals($axios, limit) {
  const signals = await $axios.$get(`http://localhost:8000/signals?limit=${limit}`)
  return signals
}

export default { 
  data() {
    return {
      signals: []
    }
  },

  // Handle initial call
  async asyncData({ $axios }) {
    const signals = await fetchSignals($axios, LIMIT)
    return { signals }
  },

  methods: {
    // Calculate uptime based on the signals belongs to a URL
    uptime: (records) => {
      let availableRecords = records.filter(record => record.available).length;
      return ((availableRecords / records.length) * 100).toFixed(2)
    }
  },

  // Set up periodic calls when the component is mounted
  mounted() {
    let that = this
    const axios = this.$axios;

    setInterval(() => {
      Promise.resolve(fetchSignals(axios, LIMIT)).then((signals) => {
        that.signals = signals
      });
    }, 1000 * LIMIT);
  }
}
</script>

At the first sight, it might look a lot, but if we check the most important parts in pieces everything will be crystal clear.

We define fetchSignals to reduce code duplication later on. Then, we set up initial signals data, where we will store the periodically fetched responses returned by the backed. After that, as part of asyncData, we initiate an async call towards the backend to get the initial signals to show.

The last part is to define a periodic call to the backend when the component is mounted. Right, we have the logic which will call backend and keep the data up to date. Now we have to display the results.

<!-- frontend/pages/index.vue -->

<template>
  <div class="p-8 h-screen text-center">
    <h1 class="text-2xl font-light">QuestDB website status</h1>
    <h2 class="text-lg font-thin">service uptime in the past 60 minutes</h2>

    <div class="h-8 mt-12 flex justify-center" v-if="signals.length > 0">

      <!-- Iterate over the signals belongs to records -->
      <div class="w-1/4" v-for="(signal, s) in signals" :key="s">
        <div class="flex mb-1 text-sm">
          <p class="flex-1 text-left font-normal">{{ signal.url }}</p>
          <p class="flex-1 text-right font-thin">{{ uptime(signal.records) }}% uptime</p>
        </div>
        <div class="grid grid-flow-col auto-cols-max gap-x-1">

          <!-- Draw a green or yellow bar depending on service availability -->
          <div 
            v-for="(signal, r) in signal.records"
            :key="r"
            :class="`w-1 bg-${signal.available ? 'green' : 'yellow'}-700`"
          >&nbsp;</div>
        </div>
      </div>
    </div>

    <!-- In case we have no records yet, show an informative message -->
    <div v-else>
      <p>No signals found</p>
    </div>
  </div>
</template>

<!-- scripts ... -->

Run the project

We reached the end of the tutorial. We have both the backend and the frontend. It is time to try everything out.
Run the following commands in different shells from the project root:

# Shell 1 - Start the application
$ docker-compose up -d
$ poetry run uvicorn app.main:app --host 127.0.0.1 --port 8000 --reload

# Shell 2 - Start the worker process
$ poetry run celery --app=app.tasks worker --beat -l info -Q main-queue -c 1

# Shell 3 - Start the frontend
$ cd frontend
$ yarn dev

Navigate to http://localhost:3000 to see the backend reporting the status of the monitored URL.
The first task to check the system status is executed when the scheduler and worker starts and
the status of the website over time can be seen after a few minutes on the page or when you check back at a later stage:

Summary

We've successfully built a pretty status page that can be publicly-visible to users or used for internal teams to monitor an application's uptime. We've learned how to queue and schedule tasks and store the responses in a time-series database and make use of low-latency queries. Engineers can modify this demo to monitor a website's HTTP response code or multiple endpoints or services for a robust overview of an entire system's status.

Thank you for your attention!

The containerized source code is available at https://github.com/gabor-boros/questdb-statuspage.

Effortless mass change management with Hammurabi

Gábor Boros — Mon, 15 Jun 2020 00:00:00 +0000

If you have ever developed microservices, you already know some of the challenges around it. If it’s done right, development can speed up, the cognitive complexity of the services can be decreased, and there are many many more advantages over monoliths. On the other hand, you have to solve different problems, like keeping a level of uniformity among services developed independently by even geographically distributed teams to reduce the maintenance costs. When we list the pros and cons of microservices, we always forget the most crucial question: How will we keep the services uniform?

The Challenge

Let’s imagine the following situation: You’ve created 5 microservices. One day you get the task to change the gunicorn configuration for all the services. Not a hard job. It’s around 10 minutes with quality git commit messages. Some months later, it turns out that you need to fine-tune the configuration, so you spend that 10 minutes again.

Now, what if you have over 70 Python services and 6 different project layouts generated from templates — which evolved during the years — where gunicorn configuration is at different places? What if you have to deal with conditions like only do modifications if the service is running on Python 3 or Scala? I think you get it.

The Solution

Since we required something for various stacks across the company, we created a proof of concept, which was able to handle simple automation tasks. After serving us for a year, we replaced the proof of concept with its open-source variant, Hammurabi.

Hammurabi is an extensible CLI tool designed for enforcing user-defined rules on a git repository. It integrates well with both git and GitHub to make sure that the execution happens on a separate branch and the committed changes are pushed to the target repository in the form of a pull request.

Its rule execution engine allows us to write complex rule sets and implement changes more efficiently. Thanks to its modularity, the default rules can be extended by open-source and private rules as well. This trait, and the execution chain, makes Hammurabi powerful, especially when it is integrated into CI/CD workflows.

Take the previous gunicorn configuration as an example. In the case where we have to write a rule for changing the configuration we could write the following.

from pathlib import Path
from hammurabi import Pillar, Law, LineReplaced

gunicorn_config_path = Path("./gunicorn.conf.py")

gunicorn_workers_set = Law(
    name='Gunicorn worker settings',
    description="Make sure gunicorn workers are configured",
    rules=(
        LineReplaced(
            name='Worker class is set to gevent',
            path=gunicorn_config_path,
            target=r'^(\s+)?worker_class\ =\ .*',
            text='worker_class = "gevent"',
        ),
    )
)

pillar = Pillar()
pillar.register(gunicorn_workers_set)

Let’s check what we did so far. We created a Law (line 6–17) which is the logical collection of rules. Laws are responsible for executing the rules assigned to them. On the other hand, rules are the units which are doing the work. Every rule has an input and output depending on its type, this ensures that the rules can be chained like pipes or in a fan-out pattern — called children in Hammurabi.

In the example above we created a Rule (line 10–15) which states the following: Make sure that any line matching the regex ^(\s+)?worker_class\ =\ .* is replaced by worker_class = "gevent" in the file gunicorn.conf.py.

We did not talk about lines 19–20 yet, so let’s take a closer look at that. As you may remember, Hammurabi — the sixth king in the Babylonian dynasty — had a finger-shaped black stone stele (pillar), where all his rules were registered. This is exactly what’s done here. We register all the laws that we would like to execute later on. The Pillar is responsible for the execution process and handling possible exceptions.

When we execute the registered laws, we will see that the target is replaced successfully by worker_class = "gevent". Although the rule will run perfectly, we missed something. The regex will match even for the text which would replace our target. To fix this issue, we can use Preconditions. Preconditions are another building block which can prevent the execution of a rule or law. Now let see how we can use preconditions to prevent the replacement if the text equals the target.

from pathlib import Path
from hammurabi import Pillar, Law, LineReplaced, IsLineNotExists

gunicorn_config_path = Path("./gunicorn.conf.py")

gunicorn_workers_set = Law(
    name='Gunicorn worker settings',
    description="Make sure gunicorn workers are configured",
    rules=(
        LineReplaced(
            name='Worker class is set to gevent',
            path=gunicorn_config_path,
            target=r'^(\s+)?worker_class\ =\ .*',
            text='worker_class = "gevent"',
            preconditions=[
              IsLineNotExists(
                path=gunicorn_config_path,
                criteria=r'^(\s+)?worker_class\ =\ "gevent"'
              )
            ]
        ),
    )
)

pillar = Pillar()
pillar.register(gunicorn_workers_set)

In line 15–20 we defined our precondition IsLineNotExists. The rule will be executed in case all of the preconditions passed, so we made sure we will not replace the line if not needed.

Hammurabi is shipped with some batteries included. Let’s imagine the case when you have a service descriptor file, which describes what’s the name of the service, how much memory is needed by the application, or even the stack of the service. At Prezi, we have a descriptor file like this, so we need a custom Precondition to make sure we are not running Python-related rules and laws on Scala services. For these cases, it is possible to define one’s own Rules or Preconditions.

Since custom Preconditions should not be available for anyone else, we implemented that in our own package, which extends Hammurabi’s default rules and preconditions.

import yaml

from pathlib import Path
from typing import Any, Dict, Iterable

from hammurabi.preconditions.base import Precondition


class ServiceYamlMixin:
    service_config: Dict[str, Any] = dict()

    def _load_service_yaml(self):
        try:
            self.service_config = yaml.safe_load(Path("service.yml").read_text())
        except FileNotFoundError:
            self.service_config = {}


class HasStack(Precondition, ServiceYamlMixin):
    def __init__(self, stacks: Iterable[str] = (), **kwargs):
        stacks = self.validate(stacks, required=True)
        super().__init__(None, stacks, **kwargs)

    def pre_task_hook(self):
        self._load_service_yaml()

    def task(self) -> bool:
        return self.service_config.get("stack") in self.param

As you see, thanks to the building blocks Hammurabi has, we can create highly customizable and powerful automation pipelines for our repositories.

We believe in open-source and care about our communities. I truly hope this tool can help developers around the world to keep their services uniform.

Thank you for reading!

Credits

Cover photo by Emilio Guzman on Unsplash.