Forem: Roman Kudryashov

Monitoring Rust web application with Prometheus and Grafana

Roman Kudryashov — Thu, 06 Jan 2022 18:14:03 +0000

Overview

In this article, I’ll show you how to set up monitoring of a Rust web application. The application will expose Prometheus metrics that will be visualized using Grafana. The monitored application is mongodb-redis demo described in detail here. Finally, the following architecture will be obtained:

The monitoring system includes:

Prometheus — monitoring platform which collects real-time metrics and records them in a time series database
Grafana — observability and data visualization platform
AlertManager — an application that handles alerts sent by the Prometheus server (for example, when something goes wrong in your application) and notifies an end user through email, Slack, Telegram, or other channels
cAdvisor — a platform providing container users an understanding of the resource usage and performance characteristics of their running containers. (Actually, it gathers information on all Docker containers on the scheme)

To start all the tools at once you can use the following:

Docker Compose file

version: '3.8'
services:

  prometheus:
    image: prom/prometheus:latest
    container_name: prometheus
    restart: always
    ports:
      - '9090:9090'
    volumes:
      - ./monitoring/prometheus:/etc/prometheus
    command:
      - '--config.file=/etc/prometheus/prometheus.yml'
      - '--web.external-url=http://localhost:9090'

  grafana:
    image: grafana/grafana:latest
    container_name: grafana
    restart: always
    ports:
      - '3000:3000'
    volumes:
      - ./monitoring/grafana/data:/var/lib/grafana
      - ./monitoring/grafana/provisioning:/etc/grafana/provisioning
    environment:
      GF_SECURITY_ADMIN_USER: admin
      GF_SECURITY_ADMIN_PASSWORD: admin

  alertmanager:
    image: prom/alertmanager:latest
    container_name: alertmanager
    ports:
      - '9093:9093'
    volumes:
      - ./monitoring/alertmanager:/etc/alertmanager
    command:
      - '--config.file=/etc/alertmanager/alertmanager.yml'
      - '--web.external-url=http://localhost:9093'

  cadvisor:
    image: gcr.io/cadvisor/cadvisor:latest
    container_name: cadvisor
    restart: always
    ports:
      - '8080:8080'
    volumes:
      - /:/rootfs:ro
      - /var/run:/var/run:rw
      - /sys:/sys:ro
      - /var/lib/docker/:/var/lib/docker:ro

Exposing Prometheus metrics in a Rust application

Metrics exposition is implemented with prometheus crate.

There are four core types of Prometheus metrics: counter, gauge, histogram, and summary. The usage of first three of them will be described in the article (the crate currently doesn’t support summary type).

Metrics definition

Metrics can be created and registered in this way:

Metrics creation and registration

lazy_static! {
    pub static ref HTTP_REQUESTS_TOTAL: IntCounterVec = register_int_counter_vec!(
        opts!("http_requests_total", "HTTP requests total"),
        &["method", "path"]
    )
    .expect("Can't create a metric");
    pub static ref HTTP_CONNECTED_SSE_CLIENTS: IntGauge =
        register_int_gauge!(opts!("http_connected_sse_clients", "Connected SSE clients"))
            .expect("Can't create a metric");
    pub static ref HTTP_RESPONSE_TIME_SECONDS: HistogramVec = register_histogram_vec!(
        "http_response_time_seconds",
        "HTTP response times",
        &["method", "path"],
        HTTP_RESPONSE_TIME_CUSTOM_BUCKETS.to_vec()
    )
    .expect("Can't create a metric");
}

In the code snippet above, custom metrics are added to the default registry; it is also possible to register them in a custom registry (see example).

Counter

If we want to count all incoming HTTP requests, it is possible to use IntCounter type. But it is more useful to see not just the total number of requests, but also some other dimensions, such as a path and a method of a request. That can be done with IntCounterVec; HTTP_REQUESTS_TOTAL of the type is used in custom Actix middleware as follows:

Usage of HTTP_REQUESTS_TOTAL metric

let request_path = req.path();
let is_registered_resource = req.resource_map().has_resource(request_path);
// this check prevents possible DoS attacks that can be done by flooding the application
// using requests to different unregistered paths. That can cause high memory consumption
// of the application and Prometheus server and also overflow Prometheus's TSDB
if is_registered_resource {
    let request_method = req.method().to_string();
    metrics::HTTP_REQUESTS_TOTAL
        .with_label_values(&[&request_method, request_path])
        .inc();
}

After making some requests to API, something like the following will be produced:

Output of HTTP_REQUESTS_TOTAL metric

# HELP http_requests_total HTTP requests total
# TYPE http_requests_total counter
http_requests_total{method="GET",path="/"} 1
http_requests_total{method="GET",path="/events"} 1
http_requests_total{method="GET",path="/metrics"} 22
http_requests_total{method="GET",path="/planets"} 20634

Each sample of the metric has method and path labels (the metric’s attributes), so Prometheus server can differentiate the samples.

As is shown in the previous snippet, requests to GET /metrics (endpoint from which Prometheus server collects metrics of an application) are also observed.

Gauge

Gauge differs from the counter by the fact that its value can go down. The example of the gauge shows how many clients at present are connected using SSE. The gauge is used as follows:

Usage of HTTP_CONNECTED_SSE_CLIENTS metric

crate::metrics::HTTP_CONNECTED_SSE_CLIENTS.inc();

crate::metrics::HTTP_CONNECTED_SSE_CLIENTS.set(broadcaster_mutex.clients.len() as i64)

If you navigate to http://localhost:9000 in your browser, connection over SSE will be established and the metric will be incremented. After that, its output will look like this:

Output of HTTP_CONNECTED_SSE_CLIENTS metric

# HELP http_connected_sse_clients Connected SSE clients
# TYPE http_connected_sse_clients gauge
http_connected_sse_clients 1

Broadcaster

To implement SSE clients gauge it was needed to refactor the code of the application and implement broadcaster. It stores all connected (using sse function) clients in a vector and periodically pings each of them (in remove_stale_clients function) to make sure that the connection is still alive otherwise removing disconnected clients from the vector. The broadcaster allows opening just one Redis Pub/Sub connection; messages from it are sent (broadcasted) to all the clients.

Histogram

In this guide, histogram is used to collect data on response times. As in the case of request counter, tracking is done in Actix middleware; it is implemented using the following code:

Response time observation

...

histogram_timer = Some(
    metrics::HTTP_RESPONSE_TIME_SECONDS
        .with_label_values(&[&request_method, request_path])
        .start_timer(),
);

...

if let Some(histogram_timer) = histogram_timer {
    histogram_timer.observe_duration();
};

I think this approach is not precise (the question is how less is this time than the actual response time) but nevertheless the observation data will be useful as an example of a histogram and for its further visualization in Grafana.

A histogram samples observations and counts them in configurable buckets (there are default buckets but most likely you will be required to define buckets customized to your use case); to configure them it is nice to know approximate spread of a metric’s values. In this application response time is quite small so the following configuration is used:

Response time buckets

const HTTP_RESPONSE_TIME_CUSTOM_BUCKETS: &[f64; 14] = &[
    0.0005, 0.0008, 0.00085, 0.0009, 0.00095, 0.001, 0.00105, 0.0011, 0.00115, 0.0012, 0.0015,
    0.002, 0.003, 1.0,
];

The output will look like this (only a part of data is shown to save space):

Output of HTTP_RESPONSE_TIME_SECONDS metric

# HELP http_response_time_seconds HTTP response times
# TYPE http_response_time_seconds histogram
http_response_time_seconds_bucket{method="GET",path="/planets",le="0.0005"} 0
http_response_time_seconds_bucket{method="GET",path="/planets",le="0.0008"} 6
http_response_time_seconds_bucket{method="GET",path="/planets",le="0.00085"} 1307
http_response_time_seconds_bucket{method="GET",path="/planets",le="0.0009"} 10848
http_response_time_seconds_bucket{method="GET",path="/planets",le="0.00095"} 22334
http_response_time_seconds_bucket{method="GET",path="/planets",le="0.001"} 31698
http_response_time_seconds_bucket{method="GET",path="/planets",le="0.00105"} 38973
http_response_time_seconds_bucket{method="GET",path="/planets",le="0.0011"} 44619
http_response_time_seconds_bucket{method="GET",path="/planets",le="0.00115"} 48707
http_response_time_seconds_bucket{method="GET",path="/planets",le="0.0012"} 51495
http_response_time_seconds_bucket{method="GET",path="/planets",le="0.0015"} 57066
http_response_time_seconds_bucket{method="GET",path="/planets",le="0.002"} 59542
http_response_time_seconds_bucket{method="GET",path="/planets",le="0.003"} 60532
http_response_time_seconds_bucket{method="GET",path="/planets",le="1"} 60901
http_response_time_seconds_bucket{method="GET",path="/planets",le="+Inf"} 60901
http_response_time_seconds_sum{method="GET",path="/planets"} 66.43133770000004
http_response_time_seconds_count{method="GET",path="/planets"} 60901

The data show numbers of observations falling into particular buckets of observation values. The output also provides information on total count and sum of observations.

System metrics

process feature enables exporting of process metrics such as CPU or memory usage. All you need is just to specify the feature in Cargo.toml. After that, you will get something like this:

Output of process metrics

# HELP process_cpu_seconds_total Total user and system CPU time spent in seconds.
# TYPE process_cpu_seconds_total counter
process_cpu_seconds_total 134.49
# HELP process_max_fds Maximum number of open file descriptors.
# TYPE process_max_fds gauge
process_max_fds 1048576
# HELP process_open_fds Number of open file descriptors.
# TYPE process_open_fds gauge
process_open_fds 37
# HELP process_resident_memory_bytes Resident memory size in bytes.
# TYPE process_resident_memory_bytes gauge
process_resident_memory_bytes 15601664
# HELP process_start_time_seconds Start time of the process since unix epoch in seconds.
# TYPE process_start_time_seconds gauge
process_start_time_seconds 1636309802.38
# HELP process_threads Number of OS threads in the process.
# TYPE process_threads gauge
process_threads 6
# HELP process_virtual_memory_bytes Virtual memory size in bytes.
# TYPE process_virtual_memory_bytes gauge
process_virtual_memory_bytes 439435264

Note that prometheus crate supports exporting process metrics of an application running on Linux (for example in such a Docker container).

Endpoint for metrics exposition

Actix is configured to process GET /metrics requests using this handler:

Metrics handler

pub async fn metrics() -> Result<HttpResponse, CustomError> {
    let encoder = TextEncoder::new();
    let mut buffer = vec![];
    encoder
        .encode(&prometheus::gather(), &mut buffer)
        .expect("Failed to encode metrics");

    let response = String::from_utf8(buffer.clone()).expect("Failed to convert bytes to string");
    buffer.clear();

    Ok(HttpResponse::Ok()
        .insert_header(header::ContentType(mime::TEXT_PLAIN))
        .body(response))
}

Now, after the successful configuration of the application, you can get all previously described metrics by executing GET http://localhost:9000/metrics request. The endpoint is used by the Prometheus server to collect metrics of the application.

Metrics are exposed in a simple text-based format.

Prometheus setup for metrics gathering

Prometheus collects (scrapes) metrics using the following config:

Prometheus config for metrics gathering

scrape_configs:

  - job_name: mongodb_redis_web_app
    scrape_interval: 5s
    static_configs:
      - targets: ['host.docker.internal:9000']

  - job_name: cadvisor
    scrape_interval: 5s
    static_configs:
      - targets: ['cadvisor:8080']

In the config, two jobs are specified. The former job collects previously described metrics of the application, the latter collects resource usage and performance metrics of running containers (that will be described in detail in the section about cAdvisor). scrape_interval specifies how frequently to scrape a target. metrics_path parameter is not specified so Prometheus expects metrics to be available on targets on a path of /metrics.

Expression browser and graphing interface

To use Prometheus’s built-in expression browser, navigate to http://localhost:9090/graph and try to query any of the previously described metrics, for example, http_requests_total. Use the "Graph" tab to visualize the data.

PromQL allows you to make more complex queries; let’s consider a couple of examples.

return all time series with the metric http_requests_total and the given job:

http_requests_total{job="mongodb_redis_web_app"}

job and instance labels are attached automatically to the scraped time series by the Prometheus server.

return per-second rate of incoming requests, as measured over the last 5 minutes:

rate(http_requests_total[5m])

You can find more examples here.

Grafana setup for metrics visualization

In this project, Grafana is configured with the following:

data sources (from where Grafana will request data)

Data sources config of Grafana

``` yaml
apiVersion: 1

datasources:
  - name: Prometheus
    type: prometheus
    access: proxy
    url: prometheus:9090
    isDefault: true
```

dashboards provider (from where Grafana will load dashboards)

Dashboards config of Grafana

``` yaml
apiVersion: 1

providers:
  - name: 'default'
    folder: 'default'
    type: file
    allowUiUpdates: true
    updateIntervalSeconds: 30
    options:
      path: /etc/grafana/provisioning/dashboards
      foldersFromFilesStructure: true
```

After launching the project as was specified in Docker Compose file, you can go to http://localhost:3000/, log in with admin/admin credentials, and find webapp_metrics dashboard. A little bit later, it will look like this:

The dashboard displays the state of the application under easy load test. (If you will launch some load test, for better visibility of the graphs (particularly the histogram) you will need to disable somehow MAX_REQUESTS_PER_MINUTE restriction, for example, by a sharp increase of that number.)

To visualize the data, the dashboard uses PromQL queries that utilize previously described metrics, for example:

rate(http_response_time_seconds_sum[5m]) / rate(http_response_time_seconds_count[5m])

Shows average response time for the last five minutes

sum(increase(http_response_time_seconds_bucket{path="/planets"}[30s])) by (le)

Is used to visualize response time distribution in the form of a heatmap. A heatmap is like a histogram, but over time, where each time slice represents its own histogram

rate(process_cpu_seconds_total{job="mongodb_redis_web_app"}[1m]), sum(rate(container_cpu_usage_seconds_total{name='mongodb-redis'}[1m])) by (name)

Shows CPU usage for the last five minutes. Requested data are coming from two different sources and show CPU usage of the process and the container respectively. The two graphs are almost identical. (sum is used because container_cpu_usage_seconds_total provides information on the usage of each core.)

Note: "Memory usage" graph shows memory used by:

the process (process_resident_memory_bytes{job="mongodb_redis_web_app"} / 1024 / 1024)
the container (container_memory_usage_bytes{name="mongodb-redis"} / 1024 / 1024)

For an unknown reason, the graph shows that the process consumes much more memory than the whole container. Also, I created an issue on this question. Please, let me know if there’s something wrong with that comparison or if you know how this can be explained.

Monitoring metrics of application container using cAdvisor

In addition to system metrics of a process (was shown previously), system metrics of the Docker container can also be exported. This can be done using cAdvisor.

Web UI of cAdvisor is accessible via http://localhost:8080/. All running Docker containers are shown at http://localhost:8080/docker/:

You can get information on resource usage of any container:

Metrics are gathered by the Prometheus server from http://localhost:8080/metrics.

Metrics exported by cAdvisor are listed here.

System metrics of machines can be exported with Node exporter or Windows exporter.

Setup of alert notifications using rules and AlertManager

In this project, the following part of Prometheus config is responsible for alerting:

Prometheus config for alerting

rule_files:
  - 'rules.yml'

alerting:
  alertmanagers:
    - static_configs:
        - targets: ['alertmanager:9093']

The alerting section defines the AlertManager instance Prometheus server communicates with.

Alerting rules allow you to define some conditions based on PromQL expressions:

Example of an alerting rule in rules.yml

groups:
- name: default
  rules:
  - alert: SseClients
    expr:  http_connected_sse_clients > 0
    for: 1m
    labels:
      severity: high
    annotations:
      summary: Too many SSE clients

alert – a name of the alert
expr – the actual rule definition in the form of Prometheus expression
for – how long the rule needs to be broken before raising an alert. In our case, if the number of SSE clients remains greater than 0 for 1 minute, an alert will be raised
labels – extra information that can be attached to the alert, e.g. severity
annotations – extra descriptions that can be attached to the alert, e.g. summary

This rule of SSE clients number greater than 0 isn’t something you would typically setup for your application. It is used as an example as it will be easy for us to break the rule by making just one request.

If the rule is broken, the Prometheus server will send an alert to the AlertManager instance. It provides numerous features, such as alerts deduplication, grouping, silencing, and dispatching notifications to an end user. We will consider only routing ability: the alert will be dispatched to an email.

AlertManager is configured as follows:

AlertManager config

route:
  receiver: gmail

receivers:
- name: gmail
  email_configs:
  - to: recipient@gmail.com
    from: email_id@gmail.com
    smarthost: smtp.gmail.com:587
    auth_username: email_id@gmail.com
    auth_identity: email_id@gmail.com
    auth_password: password

In this project, AlertManager is configured with a Gmail account. To generate an app password, you can use this guide.

To trigger SseClients alerting rule, you just need to navigate to http://localhost:9000 in your browser. This will increase http_connected_sse_clients metric by 1. You can track the status of the SseClients alert on http://localhost:9090/alerts. After the triggering, the alert will go to the Pending status. After the for interval defined in rules.yml has passed (in our case 1 minute), the alert will go to the Firing status.

That will cause the Prometheus server sending the alert to AlertManager. It will decide what to do with the alert, in our case sending an email:

Monitoring third-party systems using Prometheus exporters

For third-party tools, such as MongoDB, Redis, and many others, it is possible to set up monitoring using Prometheus exporters.

Launch

docker compose up --build

Conclusion

In this article, I showed how to setup metrics exposition in a Rust web application, their gathering with Prometheus, and data visualization with Grafana. Also, it was shown how to start working with cAdvisor to gather containers' metrics and with alerting using AlertManager. Feel free to contact me if you have found any mistakes in the article or the source code. Thanks for reading!

Useful links

Getting started with MongoDB and Redis in Rust

Roman Kudryashov — Mon, 28 Jun 2021 08:02:25 +0000

Overview

This guide will show you how to create a Rust web backend that uses MongoDB, a document-based database, to store data and Redis for caching, rate limiting, and notifications. For visibility purposes, created demo application will also be a web application that exposes REST API. Finally, the following architecture will be obtained:

MongoDB acts as storage, whereas Redis is used for the following use cases:

cache (including images)
rate limiting for HTTP requests
notifications using publish-subscribe pattern

Please note that the specified use cases don’t mean that for a similar use case you should use an approach given in this article. First of all, the examples are intended to familiarize you with MongoDB and Redis.

The project is implemented with MongoDB Rust driver and redis-rs crate.

You will also be able to test REST API of the application: it is deployed on Google Cloud Platform.

The domain model includes data about planets in the Solar System and their satellites.

Launch MongoDB and Redis

This section doesn’t require Rust skills and can be used regardless of a programming language used in an application.

Both tools can be launched as Docker containers:

docker-compose.yml

version: '3.8'

services:

  ...

  mongodb:
    image: mongo
    container_name: mongodb
    restart: always
    environment:
      MONGO_INITDB_ROOT_USERNAME: $MONGODB_USERNAME
      MONGO_INITDB_ROOT_PASSWORD: $MONGODB_PASSWORD
      MONGO_INITDB_DATABASE: solar_system_info
    ports:
      - 27017:27017

  mongodb-seed:
    image: mongo
    container_name: mongodb-seed
    depends_on:
      - mongodb
    volumes:
      - ./mongodb-init:/mongodb-init
    links:
      - mongodb
    command:
      mongoimport --host mongodb --db solar_system_info --collection planets --authenticationDatabase admin --username $MONGODB_USERNAME --password $MONGODB_PASSWORD --drop --jsonArray --file /mongodb-init/init.json

  redis:
    image: redis:alpine
    container_name: redis
    ports:
      - 6379:6379

The purpose of mongodb-seed container will be described further.

You can access mongo shell by running the following command:

docker exec -it mongodb mongo --username admin --password password

(where mongodb is the name of the Docker container, mongo is the shell)

Next, you can start executing commands, for example:

list all databases using show dbs
list all inserted data
- use solar_system_info
- show collections
- db.planets.find()

Redis CLI can be accessed by using the following command:

docker exec -it redis redis-cli

The simplest example of a command looks like this:

Example of a Redis command

> set mykey somevalue
OK
> get mykey
"somevalue"

To see all keys use keys * command.

You can find more examples of Redis CLI commands in this guide.

Data initialization

MongoDB is initialized with JSON data using mongodb-seed container and mongoimport command running inside it:

docker-compose.yml

mongodb-seed:
  image: mongo
  container_name: mongodb-seed
  depends_on:
    - mongodb
  volumes:
    - ./mongodb-init:/mongodb-init
  links:
    - mongodb
  command:
    mongoimport --host mongodb --db solar_system_info --collection planets --authenticationDatabase admin --username $MONGODB_USERNAME --password $MONGODB_PASSWORD --drop --jsonArray --file /mongodb-init/init.json

Also, the database initialization can be performed using JavaScript file.

The application also serves images of planets. My first thought was to store them in MongoDB; that can be performed using the following command:

mongofiles --host mongodb --db solar_system_info --authenticationDatabase admin --username $MONGODB_USERNAME --password $MONGODB_PASSWORD put /mongodb-init/images/*.jpg

But then it turned out that the images can’t be read from the database due to the lack of GridFS support in MongoDB Rust Driver (open ticket). So for simplicity purposes, I decided to use rust_embed crate which allows including images in the application binary at compile time (at the development time they are loaded from a file system). (It is also possible to store images separately from the application; images folder should be mounted as a volume in the Docker Compose service definition)

Next, you will see how MongoDB and Redis can be used in a Rust application.

Implementation of the application

Dependencies

The application is implemented using:

Cargo.toml

[package]
name = "mongodb-redis"
version = "0.1.0"
edition = "2018"

[dependencies]
mongodb = "2.0.0-beta.1"
redis = { version = "0.20.2", features = ["tokio-comp", "connection-manager"] }
actix-web = "4.0.0-beta.7"
tokio = "1.7.1"
tokio-stream = "0.1.6"
chrono = { version = "0.4.19", features = ["serde"] }
serde = "1.0.126"
serde_json = "1.0.64"
dotenv = "0.15.0"
derive_more = "0.99.14"
log = "0.4.14"
env_logger = "0.8.4"
rust-embed = "5.9.0"
mime = "0.3.16"

Project structure

Project structure

├───images
│
├───mongodb-init
│       init.json
│
└───src
        db.rs
        dto.rs
        errors.rs
        handlers.rs
        index.html
        main.rs
        model.rs
        redis.rs
        services.rs

`main` function

main function

#[actix_web::main]
async fn main() -> std::io::Result<()> {
    dotenv::from_filename(".env.local").ok();
    env_logger::init();

    info!("Starting MongoDB & Redis demo server");

    let mongodb_uri = env::var("MONGODB_URI").expect("MONGODB_URI env var should be specified");
    let mongodb_client = MongoDbClient::new(mongodb_uri).await;

    let redis_uri = env::var("REDIS_URI").expect("REDIS_URI env var should be specified");
    let redis_client = redis::create_client(redis_uri)
        .await
        .expect("Can't create Redis client");
    let redis_connection_manager = redis_client
        .get_tokio_connection_manager()
        .await
        .expect("Can't create Redis connection manager");

    let planet_service = Arc::new(PlanetService::new(
        mongodb_client,
        redis_client,
        redis_connection_manager.clone(),
    ));
    let rate_limiting_service = Arc::new(RateLimitingService::new(redis_connection_manager));

    ...
}

Here custom MongoDbClient, Redis client, and Redis connection manager are created.

Working with MongoDB

Let’s start with the function that returns all planets stored in the database and uses async API:

Function returning list of planets

const DB_NAME: &str = "solar_system_info";
const COLLECTION_NAME: &str = "planets";

pub async fn get_planets(
    &self,
    planet_type: Option<PlanetType>,
) -> Result<Vec<Planet>, CustomError> {
    let filter = planet_type.map(|pt| {
        doc! { "type": pt.to_string() }
    });

    let mut planets = self.get_planets_collection().find(filter, None).await?;

    let mut result: Vec<Planet> = Vec::new();
    while let Some(planet) = planets.next().await {
        result.push(planet?);
    }

    Ok(result)
}

fn get_planets_collection(&self) -> Collection<Planet> {
    self.client
        .database(DB_NAME)
        .collection::<Planet>(COLLECTION_NAME)
}

get_planets also contains the example on how to filter MongoDB documents by some criteria.

Data models are defined as follows:

Data models

#[derive(Serialize, Deserialize, Debug)]
pub struct Planet {
    #[serde(rename = "_id", skip_serializing_if = "Option::is_none")]
    pub id: Option<ObjectId>,
    pub name: String,
    pub r#type: PlanetType,
    pub mean_radius: f32,
    pub satellites: Option<Vec<Satellite>>,
}

#[derive(Copy, Clone, Eq, PartialEq, Serialize, Deserialize, Debug)]
pub enum PlanetType {
    TerrestrialPlanet,
    GasGiant,
    IceGiant,
    DwarfPlanet,
}

#[derive(Serialize, Deserialize, Debug)]
pub struct Satellite {
    pub name: String,
    pub first_spacecraft_landing_date: Option<mongodb::bson::DateTime>,
}

The structures contain fields of some common types (string, f32), as well as of:

ObjectId (Planet.id)
list (Planet.satellites)
date/timestamp (Satellite.first_spacecraft_landing_date)
enum (Planet.type)
nullable fields (Planet.id, Planet.satellites)

The project also contains examples on how to get, create, update, and delete MongoDB documents. I won’t stop in detail on these functions due to the obvious code of their implementation. You can test the functions using REST API methods:

get all planets

GET http://localhost:9000/planets

Example with filtering:

GET http://localhost:9000/planets?type=IceGiant

create planet

POST http://localhost:9000/planets

Body:

{
    "name": "Pluto",
    "type": "DwarfPlanet",
    "mean_radius": 1188,
    "satellites": null
}

get planet by id

GET http://localhost:9000/{planet_id}

update planet

PUT http://localhost:9000/{planet_id}

Body:

{
    "name": "Mercury",
    "type": "TerrestrialPlanet",
    "mean_radius": 2439.7,
    "satellites": null
}

delete planet

DELETE http://localhost:9000/{planet_id}
get image of a planet

GET http://localhost:9000/planets/{planet_id}/image

Use it to test caching with Redis

MongoDB documents are stored in BSON format.

Working with Redis

Redis client is created as follows:

Redis client creation

pub async fn create_client(redis_uri: String) -> Result<Client, RedisError> {
    Ok(Client::open(redis_uri)?)
}

Redis connection manager can be created like this:

Obtaining Redis connection manager

let redis_client = redis::create_client(redis_uri)
    .await
    .expect("Can't create Redis client");
let redis_connection_manager = redis_client
    .get_tokio_connection_manager()
    .await
    .expect("Can't create Redis connection manager");

Cache

Let’s consider the service layer’s function that is used to retrieve a planet by id:

Getting a planet by id

pub async fn get_planet(&self, planet_id: &str) -> Result<Planet, CustomError> {
    let cache_key = self.get_planet_cache_key(planet_id);
    let mut con = self.redis_client.get_async_connection().await?;

    let cached_planet = con.get(&cache_key).await?;
    match cached_planet {
        Value::Nil => {
            debug!("Use database to retrieve a planet by id: {}", &planet_id);
            let result: Planet = self
                .mongodb_client
                .get_planet(ObjectId::from_str(planet_id)?)
                .await?;

            let _: () = redis::pipe()
                .atomic()
                .set(&cache_key, &result)
                .expire(&cache_key, 60)
                .query_async(&mut con)
                .await?;

            Ok(result)
        }
        Value::Data(val) => {
            debug!("Use cache to retrieve a planet by id: {}", planet_id);
            Ok(serde_json::from_slice(&val)?)
        }
        _ => Err(RedisError {
            message: "Unexpected response from Redis".to_string(),
        }),
    }
}

If a key is not presented in the cache (Nil arm), key-value pair is put to Redis using set function (with auto expiration (its purpose will be described later)); in the second arm of match expression cached value is converted to the target structure.

To put a value to the cache, you need to implement ToRedisArgs trait for a structure:

Implementation of ToRedisArgs trait

impl ToRedisArgs for &Planet {
    fn write_redis_args<W>(&self, out: &mut W)
    where
        W: ?Sized + RedisWrite,
    {
        out.write_arg_fmt(serde_json::to_string(self).expect("Can't serialize Planet as string"))
    }
}

In get_planet function, Redis async connection is used. The following code snippet demonstrates another approach, ConnectionManager, on the example of clearing cache using del function:

Clearing cache example

pub async fn update_planet(
    &self,
    planet_id: &str,
    planet: Planet,
) -> Result<Planet, CustomError> {
    let updated_planet = self
        .mongodb_client
        .update_planet(ObjectId::from_str(planet_id)?, planet)
        .await?;

    let cache_key = self.get_planet_cache_key(planet_id);
    self.redis_connection_manager.clone().del(cache_key).await?;

    Ok(updated_planet)
}

Something can go wrong after successful entity updating (or removal); for example, a problem with the network is possible which will result in an error when calling Redis, or application shutdown/restart so that del function won't be called. That can lead to incorrect data in cache; consequences of that can be reduced by auto expiration of cached entries described earlier.

ConnectionManager can be cloned. It is also used in all the remaining Redis examples instead of Redis client.

Images cache can be implemented like a cache of other data types (using set/get functions):

Images caching

pub async fn get_image_of_planet(&self, planet_id: &str) -> Result<Vec<u8>, CustomError> {
    let cache_key = self.get_image_cache_key(planet_id);
    let mut redis_connection_manager = self.redis_connection_manager.clone();

    let cached_image = redis_connection_manager.get(&cache_key).await?;
    match cached_image {
        Value::Nil => {
            debug!(
                "Use database to retrieve an image of a planet by id: {}",
                &planet_id
            );
            let planet = self
                .mongodb_client
                .get_planet(ObjectId::from_str(planet_id)?)
                .await?;
            let result = crate::db::get_image_of_planet(&planet.name).await;

            let _: () = redis::pipe()
                .atomic()
                .set(&cache_key, result.clone())
                .expire(&cache_key, 60)
                .query_async(&mut redis_connection_manager)
                .await?;

            Ok(result)
        }
        Value::Data(val) => {
            debug!(
                "Use cache to retrieve an image of a planet by id: {}",
                &planet_id
            );
            Ok(val)
        }
        _ => Err(RedisError {
            message: "Unexpected response from Redis".to_string(),
        }),
    }
}

Caching can be tested using REST API described above.

Rate limiting

Rate limiting is implemented accordingly with the official guide as follows:

Rate limiter implementation

#[derive(Clone)]
pub struct RateLimitingService {
    redis_connection_manager: ConnectionManager,
}

impl RateLimitingService {
    pub fn new(redis_connection_manager: ConnectionManager) -> Self {
        RateLimitingService {
            redis_connection_manager,
        }
    }

    pub async fn assert_rate_limit_not_exceeded(&self, ip_addr: String) -> Result<(), CustomError> {
        let current_minute = Utc::now().minute();
        let rate_limit_key = format!("{}:{}:{}", RATE_LIMIT_KEY_PREFIX, ip_addr, current_minute);

        let (count,): (u64,) = redis::pipe()
            .atomic()
            .incr(&rate_limit_key, 1)
            .expire(&rate_limit_key, 60)
            .ignore()
            .query_async(&mut self.redis_connection_manager.clone())
            .await?;

        if count > MAX_REQUESTS_PER_MINUTE {
            Err(TooManyRequests {
                actual_count: count,
                permitted_count: MAX_REQUESTS_PER_MINUTE,
            })
        } else {
            Ok(())
        }
    }
}

For every minute per IP address, Redis key is created. After each call of assert_rate_limit_not_exceeded function, the value is incremented by 1. To make sure we don’t fill up our entire database with junk, the key is expired after one minute.

Rate limiter can be used in Actix handler like this:

Rate limiter usage

pub async fn get_planets(
    req: HttpRequest,
    web::Query(query_params): web::Query<GetPlanetsQueryParams>,
    rate_limit_service: web::Data<Arc<RateLimitingService>>,
    planet_service: web::Data<Arc<PlanetService>>,
) -> Result<HttpResponse, CustomError> {
    rate_limit_service
        .assert_rate_limit_not_exceeded(get_ip_addr(&req)?)
        .await?;

    let planets = planet_service.get_planets(query_params.r#type).await?;
    Ok(HttpResponse::Ok().json(planets.into_iter().map(PlanetDto::from).collect::<Vec<_>>()))
}

If you try to call the method of getting planets too often, you’ll get the following error:

Notifications

In this project, notifications are implemented using Redis' Pub/Sub and Server-Sent Events to deliver events to a user.

An event is published when a planet is created:

Redis event publishing

pub async fn create_planet(&self, planet: Planet) -> Result<Planet, CustomError> {
    let planet = self.mongodb_client.create_planet(planet).await?;
    self.redis_connection_manager
        .clone()
        .publish(
            NEW_PLANETS_CHANNEL_NAME,
            serde_json::to_string(&PlanetMessage::from(&planet))?,
        )
        .await?;
    Ok(planet)
}

Subscription is implemented in this way:

Redis subscription example

pub async fn get_new_planets_stream(
    &self,
) -> Result<Receiver<Result<Bytes, CustomError>>, CustomError> {
    let (tx, rx) = mpsc::channel::<Result<Bytes, CustomError>>(100);

    tx.send(Ok(Bytes::from("data: Connected\n\n")))
        .await
        .expect("Can't send a message to the stream");

    let mut pubsub_con = self
        .redis_client
        .get_async_connection()
        .await?
        .into_pubsub();
    pubsub_con.subscribe(NEW_PLANETS_CHANNEL_NAME).await?;

    tokio::spawn(async move {
        while let Some(msg) = pubsub_con.on_message().next().await {
            let payload = msg.get_payload().expect("Can't get payload of message");
            let payload: String = FromRedisValue::from_redis_value(&payload)
                .expect("Can't convert from Redis value");
            let msg = Bytes::from(format!("data: Planet created: {:?}\n\n", payload));
            tx.send(Ok(msg))
                .await
                .expect("Can't send a message to the stream");
        }
    });

    Ok(rx)
}

The subscription is used in Actix handler in this way:

SSE handler example

pub async fn sse(
    planet_service: web::Data<Arc<PlanetService>>,
) -> Result<HttpResponse, CustomError> {
    let new_planets_stream = planet_service.get_new_planets_stream().await?;
    let response_stream = tokio_stream::wrappers::ReceiverStream::new(new_planets_stream);

    Ok(HttpResponse::build(StatusCode::OK)
        .insert_header(header::ContentType(mime::TEXT_EVENT_STREAM))
        .streaming(response_stream))
}

To test notifications, you need to subscribe to notifications and generate an event. I’ll show you two approaches for this; in both the event is generated using cURL:

subscription from a browser

Navigate to http://localhost:9000/ where HTML page is accessible:
subscription from a command line using cURL

Use curl -X GET localhost:9000/events:

To generate the event the following cURL request was used:

Request to test notifications

curl -X POST -H 'Content-Type: application/json' -d '{
    \"name\": \"Pluto\",
    \"type\": \"DwarfPlanet\",
    \"mean_radius\": 1188,
    \"satellites\": null
}' localhost:9000/planets

Web application

Some parts of this topic were included in the previous sections, so now only uncovered stuff will be explained.

REST API handlers

REST API handlers are defined in this way:

REST API handlers definition

#[actix_web::main]
async fn main() -> std::io::Result<()> {
    ...

    let enable_write_handlers = env::var("ENABLE_WRITE_HANDLERS")
        .expect("ENABLE_WRITE_HANDLERS env var should be specified")
        .parse::<bool>()
        .expect("Can't parse ENABLE_WRITE_HANDLERS");

    HttpServer::new(move || {
        let mut app = App::new()
            .route("/planets", web::get().to(handlers::get_planets))
            .route("/planets/{planet_id}", web::get().to(handlers::get_planet))
            .route(
                "/planets/{planet_id}/image",
                web::get().to(handlers::get_image_of_planet),
            )
            .route("/events", web::get().to(handlers::sse))
            .route("/", web::get().to(handlers::index))
            .data(Arc::clone(&planet_service))
            .data(Arc::clone(&rate_limiting_service));

        if enable_write_handlers {
            app = app
                .route("/planets", web::post().to(handlers::create_planet))
                .route(
                    "/planets/{planet_id}",
                    web::put().to(handlers::update_planet),
                )
                .route(
                    "/planets/{planet_id}",
                    web::delete().to(handlers::delete_planet),
                );
        }

        app
    })
    .bind("0.0.0.0:9000")?
    .run()
    .await
}

Error handling

Error handling is implemented accordingly with the documentation.

Launch and testing

For launching the project locally you have two options:

using Docker Compose (docker-compose.yml):

docker compose up (or docker-compose up in older Docker versions)
without Docker

Start the application using cargo run (in this case, mongodb-redis service should be disabled in the docker-compose.yml)

CI/CD

CI/CD is configured using GitHub Actions workflow that builds Docker image of the application and deploys it on Google Cloud Platform.

To access REST API of the deployed application, you can use one of the available GET endpoints, for example:

GET http://demo.romankudryashov.com:9000/planets

Writing methods of REST API are disabled on the production environment.

Conclusion

In this article, I showed how to start with MongoDB and Redis and the basics of how to use them in a Rust application. Feel free to contact me if you have found any mistakes in the article or the source code. Thanks for reading!

Useful links

Introduction to gRPC in Rust

Roman Kudryashov — Tue, 04 May 2021 04:09:31 +0000

Overview

In this article, you will see how to create gRPC server and client in Rust. For visibility purposes, the client will also be a Telegram bot. Finally, the following architecture will be obtained:

This article is not a complete tutorial on gRPC in Rust, but rather a practical guide demonstrating the basics and how to create a gRPC-based application.

The domain model includes data about planets in the Solar System and their satellites.

Implementation

There are several gRPC implementations in Rust. For this project, tonic is used.

The project contains the following modules:

gRPC server
gRPC client (it is also a Telegram bot)
shared rpc module

The latter module contains gRPC service definition and is responsible for the generation of gRPC code needed for both server and client.

Service definition and code generation

Service definition is written in proto3 version of Protocol Buffers language and is located in the .proto file:

solar-system-info.proto

syntax = "proto3";

package solar_system_info;

import "google/protobuf/timestamp.proto";
import "google/protobuf/empty.proto";

service SolarSystemInfo {
  rpc GetPlanetsList (google.protobuf.Empty) returns (PlanetsListResponse);
  rpc GetPlanet (PlanetRequest) returns (PlanetResponse);
  rpc GetPlanets (google.protobuf.Empty) returns (stream PlanetResponse);
}

message PlanetsListResponse {
  repeated string list = 1;
}

message PlanetRequest {
  string name = 1;
}

message PlanetResponse {
  Planet planet = 1;
}

message Planet {
  uint64 id = 1;
  string name = 2;
  Type type = 3;
  float meanRadius = 4;
  float mass = 5;
  repeated Satellite satellites = 6;
  bytes image = 7;
}

enum Type {
  TERRESTRIAL_PLANET = 0;
  GAS_GIANT = 1;
  ICE_GIANT = 2;
  DWARF_PLANET = 3;
}

message Satellite {
  uint64 id = 1;
  string name = 2;
  google.protobuf.Timestamp first_spacecraft_landing_date = 3;
}

Here simple RPCs (GetPlanetsList and GetPlanet), server-side streaming RPC (GetPlanets), and structures for passing the required data are defined. The structures contain fields of some common types (uint64, string, etc.), as well as of:

enum (Planet.type)
list (Planet.satellites)
binary data (Planet.image)
date/timestamp type (Satellite.first_spacecraft_landing_date)

To set up the generation of server and client gRPC code, first let’s add the required dependencies:

Cargo.toml

[package]
name = "solar-system-info-rpc"
version = "0.1.0"
edition = "2018"

[dependencies]
tonic = "0.4.2" # Rust gRPC implementation
prost = "0.7.0" # Rust Protocol Buffers implementation
prost-types = "0.7.0" # Contains definitions of Protocol Buffers well-known types

[build-dependencies]
tonic-build = "0.4.2"

prost-types crate allows us to use some of well-known Protobuf types, such as Empty and Timestamp.

At the root of the module the following should be located:

build.rs

fn main() -> Result<(), Box<dyn std::error::Error>> {
    tonic_build::compile_protos("proto/solar-system-info/solar-system-info.proto")?;
    Ok(())
}

Now let’s create a module that will contain the generated code and will be used by both server and client applications:

lib.rs

pub mod solar_system_info {
    tonic::include_proto!("solar_system_info");
}

After you run server or client, you will find all the generated code in /target/debug/build/solar-system-info-rpc/out/solar_system_info.rs file. For example, to implement the server you will need to implement generated SolarSystemInfo trait:

Generated SolarSystemInfo trait

#[doc = r" Generated server implementations."]
pub mod solar_system_info_server {
    #![allow(unused_variables, dead_code, missing_docs)]
    use tonic::codegen::*;
    #[doc = "Generated trait containing gRPC methods that should be implemented for use with SolarSystemInfoServer."]
    #[async_trait]
    pub trait SolarSystemInfo: Send + Sync + 'static {
        async fn get_planets_list(
            &self,
            request: tonic::Request<()>,
        ) -> Result<tonic::Response<super::PlanetsListResponse>, tonic::Status>;
        async fn get_planet(
            &self,
            request: tonic::Request<super::PlanetRequest>,
        ) -> Result<tonic::Response<super::PlanetResponse>, tonic::Status>;
        #[doc = "Server streaming response type for the GetPlanets method."]
        type GetPlanetsStream: futures_core::Stream<Item = Result<super::PlanetResponse, tonic::Status>>
            + Send
            + Sync
            + 'static;
        async fn get_planets(
            &self,
            request: tonic::Request<()>,
        ) -> Result<tonic::Response<Self::GetPlanetsStream>, tonic::Status>;
    }
    #[derive(Debug)]
    pub struct SolarSystemInfoServer<T: SolarSystemInfo> {
        inner: _Inner<T>,
    }
}

Generated structures used by get_planet function look like:

Generated structures for get_planet function

#[derive(Clone, PartialEq, ::prost::Message)]
pub struct PlanetRequest {
    #[prost(string, tag = "1")]
    pub name: ::prost::alloc::string::String,
}
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct PlanetResponse {
    #[prost(message, optional, tag = "1")]
    pub planet: ::core::option::Option<Planet>,
}
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct Planet {
    #[prost(uint64, tag = "1")]
    pub id: u64,
    #[prost(string, tag = "2")]
    pub name: ::prost::alloc::string::String,
    #[prost(enumeration = "Type", tag = "3")]
    pub r#type: i32,
    #[prost(float, tag = "4")]
    pub mean_radius: f32,
    #[prost(float, tag = "5")]
    pub mass: f32,
    #[prost(message, repeated, tag = "6")]
    pub satellites: ::prost::alloc::vec::Vec<Satellite>,
    #[prost(bytes = "vec", tag = "7")]
    pub image: ::prost::alloc::vec::Vec<u8>,
}
#[derive(Clone, PartialEq, ::prost::Message)]
pub struct Satellite {
    #[prost(uint64, tag = "1")]
    pub id: u64,
    #[prost(string, tag = "2")]
    pub name: ::prost::alloc::string::String,
    #[prost(message, optional, tag = "3")]
    pub first_spacecraft_landing_date: ::core::option::Option<::prost_types::Timestamp>,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)]
#[repr(i32)]
pub enum Type {
    TerrestrialPlanet = 0,
    GasGiant = 1,
    IceGiant = 2,
    DwarfPlanet = 3,
}

gRPC server

main function of the server looks like this:

main function

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    dotenv().ok();
    env_logger::init();

    info!("Starting Solar System info server");

    let addr = std::env::var("GRPC_SERVER_ADDRESS")?.parse()?;

    let pool = create_connection_pool();
    run_migrations(&pool);

    let solar_system_info = SolarSystemInfoService { pool };
    let svc = SolarSystemInfoServer::new(solar_system_info);

    Server::builder().add_service(svc).serve(addr).await?;

    Ok(())
}

An implementation of SolarSystemInfo trait (it was shown in the previous section) looks like:

gRPC server implementation

struct SolarSystemInfoService {
    pool: PgPool,
}

#[tonic::async_trait]
impl SolarSystemInfo for SolarSystemInfoService {
    type GetPlanetsStream =
        Pin<Box<dyn Stream<Item = Result<PlanetResponse, Status>> + Send + Sync + 'static>>;

    async fn get_planets_list(
        &self,
        request: Request<()>,
    ) -> Result<Response<PlanetsListResponse>, Status> {
        debug!("Got a request: {:?}", request);

        let names_of_planets = persistence::repository::get_names(&get_connection(&self.pool))
            .expect("Can't get names of the planets");

        let reply = PlanetsListResponse {
            list: names_of_planets,
        };

        Ok(Response::new(reply))
    }

    async fn get_planets(
        &self,
        request: Request<()>,
    ) -> Result<Response<Self::GetPlanetsStream>, Status> {
        debug!("Got a request: {:?}", request);

        let (tx, rx) = mpsc::channel(4);

        let planets: Vec<Planet> = persistence::repository::get_all(&get_connection(&self.pool))
            .expect("Can't load planets")
            .into_iter()
            .map(|p| {
                PlanetWrapper {
                    planet: p.0,
                    satellites: p.1,
                }
                .into()
            })
            .collect();

        tokio::spawn(async move {
            let mut stream = tokio_stream::iter(&planets);

            while let Some(planet) = stream.next().await {
                tx.send(Ok(PlanetResponse {
                    planet: Some(planet.clone()),
                }))
                .await
                .unwrap();
            }
        });

        Ok(Response::new(Box::pin(
            tokio_stream::wrappers::ReceiverStream::new(rx),
        )))
    }

    async fn get_planet(
        &self,
        request: Request<PlanetRequest>,
    ) -> Result<Response<PlanetResponse>, Status> {
        debug!("Got a request: {:?}", request);

        let planet_name = request.into_inner().name;

        let planet =
            persistence::repository::get_by_name(&planet_name, &get_connection(&self.pool));

        match planet {
            Ok(planet) => {
                let planet = PlanetWrapper {
                    planet: planet.0,
                    satellites: planet.1,
                }
                .into();

                let reply = PlanetResponse {
                    planet: Some(planet),
                };

                Ok(Response::new(reply))
            }
            Err(e) => {
                error!(
                    "There was an error while getting a planet {}: {}",
                    &planet_name, e
                );
                match e {
                    Error::NotFound => Err(Status::not_found(format!(
                        "Planet with name {} not found",
                        &planet_name
                    ))),
                    _ => Err(Status::unknown(format!(
                        "There was an error while getting a planet {}: {}",
                        &planet_name, e
                    ))),
                }
            }
        }
    }
}

Here SolarSystemInfoService custom struct is defined which has access to the database by using Diesel ORM.

Recall that get_planets_list and get_planet are examples of unary RPC, and get_planets is the example of server-side streaming RPC.

Images of planets are included in the application binary at compile time using rust_embed crate (at the development time they are loaded from a file system).

gRPC client

gRPC client in the bot module is created as follows:

gRPC client creation

async fn create_grpc_client() -> SolarSystemInfoClient<tonic::transport::Channel> {
    let channel = tonic::transport::Channel::from_static(&GRPC_SERVER_ADDRESS)
        .connect()
        .await
        .expect("Can't create a channel");

    SolarSystemInfoClient::new(channel)
}

It then can be used like this:

Using of gRPC client

let response = get_planets_list(grpc_client).await?;

Telegram bot

As was said earlier, for visibility purposes, gRPC client is also a Telegram bot. To implement the bot, teloxide library was used.

We’ll go right away to the main.rs:

main.rs

#[tokio::main]
async fn main() {
    dotenv().ok();
    teloxide::enable_logging!();
    log::info!("Starting Solar System info bot");

    let api_url = std::env::var("TELEGRAM_API_URL").expect("Can't get Telegram API URL");
    let api_url = Url::parse(&api_url).expect("Can't parse Telegram API URL");

    let bot = Bot::from_env()
        .set_api_url(api_url)
        .parse_mode(Html)
        .auto_send();

    let bot = Arc::new(bot);

    let grpc_client = create_grpc_client().await;

    teloxide::commands_repl(bot, "solar-system-info-bot", move |cx, command| {
        answer(cx, command, grpc_client.clone())
    })
    .await;
}

To simplify SSL/TLS setup, I included nginx module in the project. It acts as a forward proxy that receives HTTP requests from the bot and redirects them to Telegram API servers.

Launch and testing

For launching the project locally you have two options:

using Docker Compose (docker-compose.yml):

docker-compose up
without Docker

Start both gRPC server and client by using cargo run

To perform requests to the server you can use some gRPC client (for example, BloomRPC):

or do it indirectly by using running Telegram bot:

Commands of the bot are mapped to RPCs as follows:

/list → GetPlanetsList
/planets → GetPlanets
/planet → GetPlanet

To test the application using the bot, you need a Telegram account and your own bot (here is an introduction on this topic). Depending on the chosen launch option, a token of the bot should be specified here or here.

CI/CD

CI/CD is configured using GitHub Actions (workflow) that builds Docker images of gRPC server and client (that is Telegram bot) and deploys them on Google Cloud Platform.

The bot can be accessed here.

Conclusion

In this article, I showed how to create gRPC server and client in Rust and also how to use the client as a data source for a Telegram bot. Feel free to contact me if you have found any mistakes in the article or the source code. Thanks for reading!

Useful links

GraphQL in Rust

Roman Kudryashov — Thu, 24 Dec 2020 10:25:38 +0000

In today’s article, I’ll describe how to create a GraphQL backend using Rust and its ecosystem. The article provides examples of an implementation of the most common tasks that can be encountered while creating GraphQL API. Finally, three microservices will be combined into a single endpoint using Apollo Server and Apollo Federation. This allows clients to fetch data from any number of sources simultaneously, without needing to know which data comes from which source.

Introduction

Overview

In terms of functionality described project is quite similar to one described in my previous article, but now it is written using Rust’s stack. The project’s architecture looks like this:

Each component of the architecture answers several questions that may arise when implementing GraphQL API. The domain model includes data about planets in the Solar System and their satellites. The project has a multi-module structure and consists of the following modules:

There are two main libraries for creating a GraphQL backend in Rust: Juniper and Async-graphql, but only the latter supports Apollo Federation, so I chose it for the project (there is also an open issue for Federation support in Juniper). Both libraries follow the code-first approach.

Also, PostgreSQL is used for persistence layer implementation, JWT — for authentication, and Kafka — for asynchronous messaging.

Stack of technologies

The following table summarizes a stack of main technologies used in the project:

Type	Name	Site	GitHub
Language	Rust	link	link
GraphQL library	Async-graphql	link	link
Single GraphQL endpoint	Apollo Server	link	link
Web framework	actix-web	link	link
Database	PostgreSQL	link	link
Event streaming platform	Apache Kafka	link	link
Container orchestration tool	Docker Compose	link	link

Also, some other used Rust libraries:

Type	Name	Site	GitHub
ORM	Diesel	link	link
Kafka client	rust-rdkafka	link	link
Password hashing library	argonautica	link	link
JWT library	jsonwebtoken	link	link
Testing library	Testcontainers-rs	link	link

Prerequisites

To launch the project locally you only need Docker Compose. Without Docker, you might need to install the following:

Rust
Diesel CLI (to install run cargo install diesel_cli --no-default-features --features postgres)
LLVM (this is needed for the argonautica crate to work)
CMake (this is needed for the rust-rdkafka crate to work)
PostgreSQL
Apache Kafka
npm

Implementation

Root Cargo.toml specifies three applications and one library:

Root Cargo.toml

[workspace]
members = [
    "auth-service",
    "planets-service",
    "satellites-service",
    "common-utils",
]

Let’s start with planets-service.

Dependencies

Cargo.toml looks like this:

Cargo.toml

[package]
name = "planets-service"
version = "0.1.0"
edition = "2018"

[dependencies]
common-utils = { path = "../common-utils" }
async-graphql = "2.4.3"
async-graphql-actix-web = "2.4.3"
actix-web = "3.3.2"
actix-rt = "1.1.1"
actix-web-actors = "3.0.0"
futures = "0.3.8"
async-trait = "0.1.42"
bigdecimal = { version = "0.1.2", features = ["serde"] }
serde = { version = "1.0.118", features = ["derive"] }
serde_json = "1.0.60"
diesel = { version = "1.4.5", features = ["postgres", "r2d2", "numeric"] }
diesel_migrations = "1.4.0"
dotenv = "0.15.0"
strum = "0.20.0"
strum_macros = "0.20.1"
rdkafka = { version = "0.24.0", features = ["cmake-build"] }
async-stream = "0.3.0"
lazy_static = "1.4.0"

[dev-dependencies]
jsonpath_lib = "0.2.6"
testcontainers = "0.9.1"

async-graphql is a GraphQL library, actix-web is a web framework, and async-graphql-actix-web provides integration between them.

Core functions

We’ll go right away to the main.rs:

main.rs

#[actix_rt::main]
async fn main() -> std::io::Result<()> {
    dotenv().ok();
    let pool = create_connection_pool();
    run_migrations(&pool);

    let schema = create_schema_with_context(pool);

    HttpServer::new(move || App::new()
        .configure(configure_service)
        .data(schema.clone())
    )
        .bind("0.0.0.0:8001")?
        .run()
        .await
}

Here an environment and HTTP server are configured using functions defined in the lib.rs:

lib.rs

pub fn configure_service(cfg: &mut web::ServiceConfig) {
    cfg
        .service(web::resource("/")
            .route(web::post().to(index))
            .route(web::get().guard(guard::Header("upgrade", "websocket")).to(index_ws))
            .route(web::get().to(index_playground))
        );
}

async fn index(schema: web::Data<AppSchema>, http_req: HttpRequest, req: Request) -> Response {
    let mut query = req.into_inner();

    let maybe_role = common_utils::get_role(http_req);
    if let Some(role) = maybe_role {
        query = query.data(role);
    }

    schema.execute(query).await.into()
}

async fn index_ws(schema: web::Data<AppSchema>, req: HttpRequest, payload: web::Payload) -> Result<HttpResponse> {
    WSSubscription::start(Schema::clone(&*schema), &req, payload)
}

async fn index_playground() -> HttpResponse {
    HttpResponse::Ok()
        .content_type("text/html; charset=utf-8")
        .body(playground_source(GraphQLPlaygroundConfig::new("/").subscription_endpoint("/")))
}

pub fn create_schema_with_context(pool: PgPool) -> Schema<Query, Mutation, Subscription> {
    let arc_pool = Arc::new(pool);
    let cloned_pool = Arc::clone(&arc_pool);
    let details_batch_loader = Loader::new(DetailsBatchLoader {
        pool: cloned_pool
    }).with_max_batch_size(10);

    let kafka_consumer_counter = Mutex::new(0);

    Schema::build(Query, Mutation, Subscription)
        .data(arc_pool)
        .data(details_batch_loader)
        .data(kafka::create_producer())
        .data(kafka_consumer_counter)
        .finish()
}

These functions do the following:

index — processes GraphQL queries and mutations
index_ws — processes GraphQL subscriptions
index_playground — provides Playground GraphQL IDE
create_schema_with_context — creates GraphQL schema with global contextual data that are accessible at runtime, for example, a pool of database connections

Query and type definition

Let’s consider how to define a query:

Query definition

#[Object]
impl Query {
    async fn get_planets(&self, ctx: &Context<'_>) -> Vec<Planet> {
        repository::get_all(&get_conn_from_ctx(ctx)).expect("Can't get planets")
            .iter()
            .map(|p| { Planet::from(p) })
            .collect()
    }

    async fn get_planet(&self, ctx: &Context<'_>, id: ID) -> Option<Planet> {
        find_planet_by_id_internal(ctx, id)
    }

    #[graphql(entity)]
    async fn find_planet_by_id(&self, ctx: &Context<'_>, id: ID) -> Option<Planet> {
        find_planet_by_id_internal(ctx, id)
    }
}

fn find_planet_by_id_internal(ctx: &Context<'_>, id: ID) -> Option<Planet> {
    let id = id.to_string().parse::<i32>().expect("Can't get id from String");
    repository::get(id, &get_conn_from_ctx(ctx)).ok()
        .map(|p| { Planet::from(&p) })
}

The queries are getting data from a database using a repository. Obtained entities are converted to GraphQL DTO (that allows us to preserve the single-responsibility principle for each struct). get_planets and get_planet queries can be accessed from any GraphQL IDE, for example like this:

Query example

{
  getPlanets {
    name
    type
  }
}

Planet struct is defined as follows:

GraphQL type definition

#[derive(Serialize, Deserialize)]
struct Planet {
    id: ID,
    name: String,
    planet_type: PlanetType,
}

#[Object]
impl Planet {
    async fn id(&self) -> &ID {
        &self.id
    }

    async fn name(&self) -> &String {
        &self.name
    }

    /// From an astronomical point of view
    #[graphql(name = "type")]
    async fn planet_type(&self) -> &PlanetType {
        &self.planet_type
    }

    #[graphql(deprecation = "Now it is not in doubt. Do not use this field")]
    async fn is_rotating_around_sun(&self) -> bool {
        true
    }

    async fn details(&self, ctx: &Context<'_>) -> Details {
        let loader = ctx.data::<Loader<i32, Details, DetailsBatchLoader>>().expect("Can't get loader");
        let planet_id = self.id.to_string().parse::<i32>().expect("Can't convert id");
        loader.load(planet_id).await
    }
}

Here we define a resolver for each field. Also, for some fields, description (as Rust’s comment) and deprecation reason are specified. These will be shown in a GraphQL IDE.

N+1 problem

If the definition of details function of Planet above was naive, this would lead to N+1 problem, that is, if you made such a request:

Example of possible resource-consuming GraphQL request

{
  getPlanets {
    name
    details {
      meanRadius
    }
  }
}

then separate SQL call would be executed for details field of each planet because Details is a separate entity from Planet and is stored in its own table.

But with help of Async-graphql’s implementation of DataLoader details resolver can be defined as follows:

Resolver definition

async fn details(&self, ctx: &Context<'_>) -> Result<Details> {
    let data_loader = ctx.data::<DataLoader<DetailsLoader>>().expect("Can't get data loader");
    let planet_id = self.id.to_string().parse::<i32>().expect("Can't convert id");
    let details = data_loader.load_one(planet_id).await?;
    details.ok_or_else(|| "Not found".into())
}

data_loader is an application-scoped object defined in this way:

DataLoader definition

let details_data_loader = DataLoader::new(DetailsLoader {
    pool: cloned_pool
}).max_batch_size(10);

DetailsLoader is implemented like this:

DetailsLoader definition

pub struct DetailsLoader {
    pub pool: Arc<PgPool>
}

#[async_trait::async_trait]
impl Loader<i32> for DetailsLoader {
    type Value = Details;
    type Error = Error;

    async fn load(&self, keys: &[i32]) -> Result<HashMap<i32, Self::Value>, Self::Error> {
        let conn = self.pool.get().expect("Can't get DB connection");
        let details = repository::get_details(keys, &conn).expect("Can't get planets' details");

        Ok(details.iter()
            .map(|details_entity| (details_entity.planet_id, Details::from(details_entity)))
            .collect::<HashMap<_, _>>())
    }
}

This method helps us to prevent N+1 problem because each DetailsLoader.load call executes just one SQL call returning a bunch of DetailsEntity.

Interface definition

GraphQL interface and its implementations can be defined in this way:

GraphQL interface definition

#[derive(Interface, Clone)]
#[graphql(
    field(name = "mean_radius", type = "&CustomBigDecimal"),
    field(name = "mass", type = "&CustomBigInt"),
)]
pub enum Details {
    InhabitedPlanetDetails(InhabitedPlanetDetails),
    UninhabitedPlanetDetails(UninhabitedPlanetDetails),
}

#[derive(SimpleObject, Clone)]
pub struct InhabitedPlanetDetails {
    mean_radius: CustomBigDecimal,
    mass: CustomBigInt,
    /// In billions
    population: CustomBigDecimal,
}

#[derive(SimpleObject, Clone)]
pub struct UninhabitedPlanetDetails {
    mean_radius: CustomBigDecimal,
    mass: CustomBigInt,
}

Here you can also see that if the structure doesn’t have a single field with a complex resolver, it can be implemented using SimpleObject attribute.

Custom scalar definition

Custom scalars allow defining how to represent a value of some type and how to parse it. The project contains two examples of custom scalar definition; both are wrappers for numeric structures (because you can’t implement an external trait on an external type due to the orphan rule). The wrappers are implemented as follows:

Custom scalar: wrapper for BigInt

#[derive(Clone)]
pub struct CustomBigInt(BigDecimal);

#[Scalar(name = "BigInt")]
impl ScalarType for CustomBigInt {
    fn parse(value: Value) -> InputValueResult<Self> {
        match value {
            Value::String(s) => {
                let parsed_value = BigDecimal::from_str(&s)?;
                Ok(CustomBigInt(parsed_value))
            }
            _ => Err(InputValueError::expected_type(value)),
        }
    }

    fn to_value(&self) -> Value {
        Value::String(format!("{:e}", &self))
    }
}

impl LowerExp for CustomBigInt {
    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
        let val = &self.0.to_f64().expect("Can't convert BigDecimal");
        LowerExp::fmt(val, f)
    }
}

Custom scalar: wrapper for BigDecimal

#[derive(Clone)]
pub struct CustomBigDecimal(BigDecimal);

#[Scalar(name = "BigDecimal")]
impl ScalarType for CustomBigDecimal {
    fn parse(value: Value) -> InputValueResult<Self> {
        match value {
            Value::String(s) => {
                let parsed_value = BigDecimal::from_str(&s)?;
                Ok(CustomBigDecimal(parsed_value))
            }
            _ => Err(InputValueError::expected_type(value)),
        }
    }

    fn to_value(&self) -> Value {
        Value::String(self.0.to_string())
    }
}

The former example also shows how to represent a large number using exponential (scientific) notation.

Mutation definition

A mutation can be defined as the following:

Mutation definition

pub struct Mutation;

#[Object]
impl Mutation {
    #[graphql(guard(RoleGuard(role = "Role::Admin")))]
    async fn create_planet(&self, ctx: &Context<'_>, planet: PlanetInput) -> Result<Planet, Error> {
        let new_planet = NewPlanetEntity {
            name: planet.name,
            planet_type: planet.planet_type.to_string(),
        };

        let details = planet.details;
        let new_planet_details = NewDetailsEntity {
            mean_radius: details.mean_radius.0,
            mass: BigDecimal::from_str(&details.mass.0.to_string()).expect("Can't get BigDecimal from string"),
            population: details.population.map(|wrapper| { wrapper.0 }),
            planet_id: 0,
        };

        let created_planet_entity = repository::create(new_planet, new_planet_details, &get_conn_from_ctx(ctx))?;

        let producer = ctx.data::<FutureProducer>().expect("Can't get Kafka producer");
        let message = serde_json::to_string(&Planet::from(&created_planet_entity)).expect("Can't serialize a planet");
        kafka::send_message(producer, message).await;

        Ok(Planet::from(&created_planet_entity))
    }
}

To use an object as input parameter of a mutation you need to define a structure in this way:

Input type definition

#[derive(InputObject)]
struct PlanetInput {
    name: String,
    #[graphql(name = "type")]
    planet_type: PlanetType,
    details: DetailsInput,
}

The mutation is protected by RoleGuard which ensures that only users with Admin role can access it. So to execute the mutation, for example, like this:

Example of mutation usage

mutation {
  createPlanet(
    planet: {
      name: "test_planet"
      type: TERRESTRIAL_PLANET
      details: { meanRadius: "10.5", mass: "8.8e24", population: "0.5" }
    }
  ) {
    id
  }
}

you need to specify Authorization header with a JWT obtained from auth-service (this will be described later).

Subscription definition

In the mutation definition above you could see that on a planet creation a message is sent:

Sending a message to Kafka

let producer = ctx.data::<FutureProducer>().expect("Can't get Kafka producer");
let message = serde_json::to_string(&Planet::from(&created_planet_entity)).expect("Can't serialize a planet");
kafka::send_message(producer, message).await;

The API client can be notified of the event by using a subscription that listens to Kafka consumer:

Subscription definition

pub struct Subscription;

#[Subscription]
impl Subscription {
    async fn latest_planet<'ctx>(&self, ctx: &'ctx Context<'_>) -> impl Stream<Item=Planet> + 'ctx {
        let kafka_consumer_counter = ctx.data::<Mutex<i32>>().expect("Can't get Kafka consumer counter");
        let consumer_group_id = kafka::get_kafka_consumer_group_id(kafka_consumer_counter);
        let consumer = kafka::create_consumer(consumer_group_id);

        async_stream::stream! {
            let mut stream = consumer.start();

            while let Some(value) = stream.next().await {
                yield match value {
                    Ok(message) => {
                        let payload = message.payload().expect("Kafka message should contain payload");
                        let message = String::from_utf8_lossy(payload).to_string();
                        serde_json::from_str(&message).expect("Can't deserialize a planet")
                    }
                    Err(e) => panic!("Error while Kafka message processing: {}", e)
                };
            }
        }
    }
}

A subscription can be used like queries and mutations:

Example of subscription usage

subscription {
  latestPlanet {
    id
    name
    type
    details {
      meanRadius
    }
  }
}

Subscription URL is ws://localhost:8001.

Integration tests

Tests for queries and mutations can be written like this:

Query test

#[actix_rt::test]
async fn test_get_planets() {
    let docker = Cli::default();
    let (_pg_container, pool) = common::setup(&docker);

    let mut service = test::init_service(App::new()
        .configure(configure_service)
        .data(create_schema_with_context(pool))
    ).await;

    let query = "
        {
            getPlanets {
                id
                name
                type
                details {
                    meanRadius
                    mass
                    ... on InhabitedPlanetDetails {
                        population
                    }
                }
            }
        }
        ".to_string();

    let request_body = GraphQLCustomRequest {
        query,
        variables: Map::new(),
    };

    let request = test::TestRequest::post().uri("/").set_json(&request_body).to_request();

    let response: GraphQLCustomResponse = test::read_response_json(&mut service, request).await;

    fn get_planet_as_json(all_planets: &serde_json::Value, index: i32) -> &serde_json::Value {
        jsonpath::select(all_planets, &format!("$.getPlanets[{}]", index)).expect("Can't get planet by JSON path")[0]
    }

    let mercury_json = get_planet_as_json(&response.data, 0);
    common::check_planet(mercury_json, 1, "Mercury", "TERRESTRIAL_PLANET", "2439.7");

    let earth_json = get_planet_as_json(&response.data, 2);
    common::check_planet(earth_json, 3, "Earth", "TERRESTRIAL_PLANET", "6371.0");

    let neptune_json = get_planet_as_json(&response.data, 7);
    common::check_planet(neptune_json, 8, "Neptune", "ICE_GIANT", "24622.0");
}

If a part of a query can be reused in another query, you can use fragments:

Query test using fragment

const PLANET_FRAGMENT: &str = "
    fragment planetFragment on Planet {
        id
        name
        type
        details {
            meanRadius
            mass
            ... on InhabitedPlanetDetails {
                population
            }
        }
    }
";

#[actix_rt::test]
async fn test_get_planet_by_id() {
    ...

    let query = "
        {
            getPlanet(id: 3) {
                ... planetFragment
            }
        }
        ".to_string() + PLANET_FRAGMENT;

    let request_body = GraphQLCustomRequest {
        query,
        variables: Map::new(),
    };

    ...
}

To use variables, you can write test in this way:

Query test using a fragment and variables

#[actix_rt::test]
async fn test_get_planet_by_id_with_variable() {
    ...

    let query = "
        query testPlanetById($planetId: String!) {
            getPlanet(id: $planetId) {
                ... planetFragment
            }
        }".to_string() + PLANET_FRAGMENT;

    let jupiter_id = 5;
    let mut variables = Map::new();
    variables.insert("planetId".to_string(), jupiter_id.into());

    let request_body = GraphQLCustomRequest {
        query,
        variables,
    };

    ...
}

In this project, Testcontainers-rs library is used to prepare test environment, that is, to create a temporary PostgreSQL database.

GraphQL API client

You can use code snippets from the previous section to create a client to an external GraphQL API. Also, there are special libraries for this purpose, for example, graphql-client, but I haven’t used them yet.

API security

There are different security threats to your GraphQL API (see this checklist to learn more); let’s consider some of these aspects.

Query depth and complexity limit

If the Satellite structure contained planet field, the following request would be possible:

Example of expensive query

{
  getPlanet(id: "1") {
    satellites {
      planet {
        satellites {
          planet {
            satellites {
              ... # more deep nesting!
            }
          }
        }
      }
    }
  }
}

To make such a request invalid we can specify:

Example of query depth and complexity limit

pub fn create_schema_with_context(pool: PgPool) -> Schema<Query, Mutation, Subscription> {
    ...

    Schema::build(Query, Mutation, Subscription)
        .limit_depth(3)
        .limit_complexity(15)

    ...
}

Note that if you specify depth or complexity limit, then documentation of a service may stop showing in your GraphQL IDE. This is because the IDE tries to perform an introspection query that has considerable depth and complexity.

Authentication

This functionality is implemented in auth-service using argonautica and jsonwebtoken crates. The former crate is responsible for a user’s password hashing using Argon2 algorithm. Authentication and authorization functionality is only for demo; please do more research for production use.

Let’s consider how signing in is implemented:

Signing in implementation

pub struct Mutation;

#[Object]
impl Mutation {

    async fn sign_in(&self, ctx: &Context<'_>, input: SignInInput) -> Result<String, Error> {
        let maybe_user = repository::get_user(&input.username, &get_conn_from_ctx(ctx)).ok();

        if let Some(user) = maybe_user {
            if let Ok(matching) = verify_password(&user.hash, &input.password) {
                if matching {
                    let role = AuthRole::from_str(user.role.as_str()).expect("Can't convert &str to AuthRole");
                    return Ok(common_utils::create_token(user.username, role));
                }
            }
        }

        Err(Error::new("Can't authenticate a user"))
    }
}

#[derive(InputObject)]
struct SignInInput {
    username: String,
    password: String,
}

You can check the implementation of verify_password function in utils module and create_token in common_utils module. As you may expect, sign_in function issues JWT that further can be used for authorization in other services.

To get a JWT you need to perform the following mutation:

Getting JWT

mutation {
  signIn(input: { username: "john_doe", password: "password" })
}

For that, you can use john_doe/password credentials. Including obtained JWT to further requests allows access to protected resources (see next section).

Authorization

To request protected data you need to add a header to an HTTP request in the format Authorization: Bearer $JWT. index function will extract a user’s role from the request and will add it to a query data:

Role extraction

async fn index(schema: web::Data<AppSchema>, http_req: HttpRequest, req: Request) -> Response {
    let mut query = req.into_inner();

    let maybe_role = common_utils::get_role(http_req);
    if let Some(role) = maybe_role {
        query = query.data(role);
    }

    schema.execute(query).await.into()
}

The following attribute was applied to create_planet mutation defined earlier:

Guard usage

#[graphql(guard(RoleGuard(role = "Role::Admin")))]

Guard itself is implemented as follows:

Guard implementation

struct RoleGuard {
    role: Role,
}

#[async_trait::async_trait]
impl Guard for RoleGuard {
    async fn check(&self, ctx: &Context<'_>) -> Result<()> {
        if ctx.data_opt::<Role>() == Some(&self.role) {
            Ok(())
        } else {
            Err("Forbidden".into())
        }
    }
}

So if you don’t specify a token, the server will respond with a message containing "Forbidden".

Enum definition

GraphQL enum can be defined in this way:

Enum definition

#[derive(SimpleObject)]
struct Satellite {
    ...
    life_exists: LifeExists,
}

#[derive(Copy, Clone, Eq, PartialEq, Debug, Enum, EnumString)]
#[strum(serialize_all = "SCREAMING_SNAKE_CASE")]
pub enum LifeExists {
    Yes,
    OpenQuestion,
    NoData,
}

Work with dates

Async-graphql supports date/time types from chrono library, so you can define such fields as usual:

Date field definition

#[derive(SimpleObject)]
struct Satellite {
    ...
    first_spacecraft_landing_date: Option<NaiveDate>,
}

Apollo Federation support

One of the purposes of satellites-service is to demonstrate how the distributed GraphQL entity (Planet) can be resolved in two (or more) services and then accessed through Apollo Server.

Planet type was earlier defined in the planets-service in this way:

Definition of Planet type in planets-service

#[derive(Serialize, Deserialize)]
struct Planet {
    id: ID,
    name: String,
    planet_type: PlanetType,
}

Also, in planets-service Planet type is an entity:

Planet entity definition

#[Object]
impl Query {
    #[graphql(entity)]
    async fn find_planet_by_id(&self, ctx: &Context<'_>, id: ID) -> Option<Planet> {
        find_planet_by_id_internal(ctx, id)
    }
}

satellites-service extends Planet entity by adding the satellites field to it:

Extension of Planet type in satellites-service

struct Planet {
    id: ID
}

#[Object(extends)]
impl Planet {
    #[graphql(external)]
    async fn id(&self) -> &ID {
        &self.id
    }

    async fn satellites(&self, ctx: &Context<'_>) -> Vec<Satellite> {
        let id = self.id.to_string().parse::<i32>().expect("Can't get id from String");
        repository::get_by_planet_id(id, &get_conn_from_ctx(ctx)).expect("Can't get satellites of planet")
            .iter()
            .map(|e| { Satellite::from(e) })
            .collect()
    }
}

You should also provide a lookup function for an extended type (here it just creates a new instance of Planet):

Lookup function for Planet type

#[Object]
impl Query {

    #[graphql(entity)]
    async fn get_planet_by_id(&self, id: ID) -> Planet {
        Planet { id }
    }
}

Async-graphql generates two additional queries (_service and _entities) that will be used by Apollo Server. These queries are internal, i.e., they won’t be exposed by Apollo Server. Of course, a service with Apollo Federation support can still work independently.

Apollo Server

Apollo Server and Apollo Federation allow achieving 2 main goals:

create a single endpoint to access GraphQL APIs provided by multiple services
create a single data graph from distributed entities

That is, even if you don’t use federated entities, it is more convenient for frontend developers to use a single endpoint rather than multiple endpoints.

There is another way for creating a single GraphQL schema, schema stitching, but I haven’t used this approach.

This module includes the following sources:

Meta information and dependencies

{
  "name": "api-gateway",
  "main": "gateway.js",
  "scripts": {
    "start-gateway": "nodemon gateway.js"
  },
  "devDependencies": {
    "concurrently": "5.3.0",
    "nodemon": "2.0.6"
  },
  "dependencies": {
    "@apollo/gateway": "0.21.3",
    "apollo-server": "2.19.0",
    "graphql": "15.4.0"
  }
}

Apollo Server definition

const {ApolloServer} = require("apollo-server");
const {ApolloGateway, RemoteGraphQLDataSource} = require("@apollo/gateway");

class AuthenticatedDataSource extends RemoteGraphQLDataSource {
    willSendRequest({request, context}) {
        if (context.authHeaderValue) {
            request.http.headers.set('Authorization', context.authHeaderValue);
        }
    }
}

let node_env = process.env.NODE_ENV;

function get_service_url(service_name, port) {
    let host;
    switch (node_env) {
        case 'docker':
            host = service_name;
            break;
        case 'local': {
            host = 'localhost';
            break
        }
    }

    return "http://" + host + ":" + port;
}

const gateway = new ApolloGateway({
    serviceList: [
        {name: "planets-service", url: get_service_url("planets-service", 8001)},
        {name: "satellites-service", url: get_service_url("satellites-service", 8002)},
        {name: "auth-service", url: get_service_url("auth-service", 8003)},
    ],
    buildService({name, url}) {
        return new AuthenticatedDataSource({url});
    },
});

const server = new ApolloServer({
    gateway, subscriptions: false, context: ({req}) => ({
        authHeaderValue: req.headers.authorization
    })
});

server.listen({host: "0.0.0.0", port: 4000}).then(({url}) => {
    console.log(`🚀 Server ready at ${url}`);
});

If the source above can be simplified, feel free to contact me for change.

Authorization in apollo-service works as was described earlier for Rust services (you just need to specify Authorization header and its value).

An application written in any language or framework can be added as a downstream service to Apollo Server if it implements Federation specification; a list of libraries that offer such support is available in the documentation.

While implementing this module, I encountered some limitations:

Apollo Gateway doesn’t support subscriptions (but they still work in a standalone Rust GraphQL application)
a service trying to extend a GraphQL interface requires knowledge of concrete implementations

Interaction with a database

The persistence layer is implemented using PostgreSQL and Diesel. If you don’t use Docker locally, you should run diesel setup being in each service’s folder. This will create an empty database to which then will be applied migrations creating tables and inserting data.

Project launch and API testing

As was mentioned before, for launching the project locally you have two options:

using Docker Compose (docker-compose.yml)

In its turn, here you also have two options:
- development mode (using locally built images)
  
  docker-compose up --build
- production mode (using released images)
  
  docker-compose -f docker-compose.yml up
without Docker

Start each Rust service using cargo run, then start Apollo Server:
- cd to the apollo-server folder
- define NODE_ENV variable, for example, set NODE_ENV=local (for Windows)
- npm install
- npm run start-gateway

A successful launch of apollo-server should look like this:

Apollo Server startup log

[nodemon] 2.0.6
[nodemon] to restart at any time, enter `rs`
[nodemon] watching path(s): *.*
[nodemon] watching extensions: js,mjs,json
[nodemon] starting `node gateway.js`
Server ready at http://0.0.0.0:4000/

You can navigate to http://localhost:4000 in your browser and use a built-in Playground IDE:

Here you can execute queries, mutations, and subscriptions defined in downstream services. Also, each of those services has its own Playground IDE.

Subscription test

To test whether the subscription works open two tabs of any GraphQL IDE; in the first subscribe as follows:

Request for subscription

subscription {
  latestPlanet {
    name
    type
  }
}

In the second specify Authorization header as was described above and perform mutation like this:

Request for mutation

mutation {
  createPlanet(
    planet: {
      name: "Pluto"
      type: DWARF_PLANET
      details: { meanRadius: "1188", mass: "1.303e22" }
    }
  ) {
    id
  }
}

The subscribed client will be notified of a planet creation:

CI/CD

CI/CD is configured using GitHub Actions (workflow) that runs applications' tests, builds their Docker images, and deploys them on Google Cloud Platform.

You can take a look at the described API here.

Note: On the "production" environment password is different from the one specified earlier to prevent changing initial data.

Conclusion

In this article, I considered how to solve the most common problems that can arise while developing GraphQL API in Rust. Also, I showed how to combine APIs of Rust GraphQL microservices to provide a unified GraphQL interface; in such an architecture an entity can be distributed among several microservices. It is achieved by using Apollo Server, Apollo Federation, and Async-graphql library. The source code of the considered project is on GitHub. Feel free to contact me if you have found any mistakes in the article or the source code. Thanks for reading!

Forem: Roman Kudryashov

Monitoring Rust web application with Prometheus and Grafana

Overview

Exposing Prometheus metrics in a Rust application

Metrics definition

Counter

Gauge

Broadcaster

Histogram

System metrics

Endpoint for metrics exposition

Prometheus setup for metrics gathering

Expression browser and graphing interface

Grafana setup for metrics visualization

Monitoring metrics of application container using cAdvisor

Setup of alert notifications using rules and AlertManager

Monitoring third-party systems using Prometheus exporters

Launch

Conclusion

Useful links

Getting started with MongoDB and Redis in Rust

Overview

Launch MongoDB and Redis

Data initialization

Implementation of the application

Dependencies

Project structure

main function

Working with MongoDB

Working with Redis

Cache

Rate limiting

Notifications

Web application

REST API handlers

Error handling

Launch and testing

CI/CD

Conclusion

Useful links

Introduction to gRPC in Rust

Overview

Implementation

Service definition and code generation

gRPC server

gRPC client

Telegram bot

Launch and testing

CI/CD

Conclusion

Useful links

GraphQL in Rust

Introduction

Overview

Stack of technologies

Prerequisites

Implementation

Dependencies

Core functions

Query and type definition

N+1 problem

Interface definition

Custom scalar definition

Mutation definition

Subscription definition

Integration tests

GraphQL API client

API security

Query depth and complexity limit

Authentication

Authorization

Enum definition

Work with dates

Apollo Federation support

Apollo Server

Interaction with a database

Project launch and API testing

Subscription test

CI/CD

Conclusion

`main` function