Forem: Alex Koutmos

How to Use gRPC in Elixir

Alex Koutmos — Thu, 02 Apr 2020 10:58:38 +0000

In today's post, we'll learn what gRPC is, when you should reach for such a tool, and some of the pros and cons of using it. After going over an introduction of gRPC, we'll dive right into a sample application where we'll build an Elixir backend API powered by gRPC.

Let's jump right in!

What Is gRPC and How Does It Work?

gRPC is a framework used to enable a remote procedure call (RPC) style of communication. RPC is a style of system communication where a client can directly invoke exposed methods on a server. From the client's perspective, it feels no different than making a call to a local function or method as long as you provide the applicable parameters and handle
the return type appropriately. gRPC facilitates this communication by providing 2 things for you:

A client library which can be used to invoke allowed procedures
A consistent data serialization standard via Protocol Buffers

Let's break these two items down so we can appreciate the inner workings of gRPC. When creating a gRPC server, you'll have to define what procedures are invokable from the client, what inputs they accept, and what outputs they return. This interface specification is what allows client libraries (generally called gRPC stubs) to be automatically generated for various languages and runtimes as the contract for the remote procedure call is explicitly defined. This gRPC client
library can then communicate with the server using Protocol Buffers. Protocol Buffers provide a mechanism for serializing and deserializing the payloads (both request and response) so that you can operate on the data with types native to your language. The diagram available in the gRPC documentation can help visualize this interaction:

One thing that we haven't discussed yet is how data is transmitted between the client and the server. For this, gRPC leans on the HTTP/2 protocol. By using HTTP/2 as the underlying protocol, gRPC is able to support features such as bi-directional data streaming and several other features that are not available in HTTP/1.1.

When Would You Use gRPC Over REST/GraphQL?

The obvious question that may come to mind is: "How does gRPC compare to REST/GraphQL, and when do I use one over the other?".

In general, if you plan to use gRPC for a frontend application, there are a couple of caveats that you need to keep in mind. In order to serialize and deserialize Protocol Buffer payloads from your Javascript application, you'll have to leverage grpc-web. In addition, you'll also need to run a proxy on the backend (the
default supported gRPC proxy is Envoy) since browsers cannot talk directly to gRPC servers. Depending on your resources and time constraints, this may be a show stopper; in which case, REST and GraphQL will do just fine.

If frontend application communication is not a requirement, and instead what you require is inter-microservice communication from within your service cluster, then the barrier to entry for gRPC gets lowered considerably. At the time of writing, gRPC currently has client libraries and tooling for most mainstream languages including Elixir, Python,
C++, Go, Ruby, to name a few. I would argue that the barrier to entry for consuming a RESTful API is still much lower than consuming a gRPC service given that all you need for the former is an HTTP client, which is baked into most languages and runtimes these days.

On the other hand, if you are willing to make the investment, you do get the added benefits of having your responses and requests checked against the Protocol Buffer specification that is used for code generation. This, in turn, provides you with some guarantees as to what you can expect on the client-side and the server-side. This guarantee and introspection is also something that you get with GraphQL when you define your server schemas. An added benefit of GraphQL over gRPC is that you are able to dynamically request embedded properties from within your schema depending on the query that you make to your backend server.

Like most things in the software engineering field, the technology you choose will largely depend on your application. Below are my personal TL;DR rules of thumb regarding the various technologies:

gRPC:

Use: When communicating between microservices in my service cluster or if performance is a requirement
Don't use: When I need to transmit data from the browser to the backend

GraphQL:

Use: When I need to aggregate data from multiple microservices for the purposes of streamlining frontend development, or if my frontend data requirements are dynamic
Don't use: When communicating between microservices in my service cluster, or if my API needs to be used by the lowest common denominator of consumers

REST:

Use: When I need to put something together quickly or if I need to cater to the lowest common denominator of consumers
Don't use: If I require any kind of type checking or if I want to reduce payload sizes over the wire

Experimenting With gRPC in Elixir

With all the theory out of the way, it's time to get our hands dirty and experiment with wiring up a gRPC server. In order to keep us focused on the gRPC experience, we'll opt for having a backend powered by an Agent versus an actual database, but all of the concepts should be easily transferable to an application backed by Postgres, for example. Our
gRPC application will be a simple user management service where we can create and fetch users. After creating our Elixir service, we'll interact with it via grpcurl, which is effectively cURL, but for gRPC. If at any point you get stuck or require the source code, feel free to check it out on GitHub. With all that being said, let's dive right in!

Before creating our Elixir project, there are a couple of things that we require on our machine in order to properly develop and test our application. We'll first need to install protoc so that .proto files can be compiled appropriately. If you are on an OSX machine, you can run brew install protobuf, otherwise, see instructions specific to your platform. Now, with protoc available on your machine, you'll also want to install grpcurl so that you can interact with the application. Once again, if you are on an OSX machine, you can run brew install grpcurl, otherwise, check for instructions specific to your platform.

Lastly, you'll want to run mix escript.install hex protobuf and ensure that protoc-gen-elixir script is available on your path (if you use ASDF as your runtime version manager, this requires running asdf reshim elixir). With all that boilerplate done, you can run mix new sample_app --sup to get a new application started.

Once inside your sample application directory, you'll want to update your mix.exs file to include our gRPC related dependencies. For this application, we will be leveraging https://github.com/elixir-grpc/grpc and
https://github.com/tony612/protobuf-elixir. In order to bring these two dependencies into your project, ensure that your deps/0 function looks like this:

defp deps do
  [
    {:grpc, "~> 0.5.0-beta"},
    {:cowlib, "~> 2.8.0", hex: :grpc_cowlib, override: true}
  ]
end

With that in place, run mix deps.get from the terminal to pull down all necessary project dependencies. Next, you'll want to create a configuration file at config/config.exs with the following content:

use Mix.Config

# Configures Elixir's Logger
config :logger, :console, format: "$time $metadata[$level] $message\n"

config :grpc, start_server: true

Next, we'll want to create the required Protocol Buffer definitions for our application. The Protocol Buffer specification is fairly large and we'll only be using a small subset of it to keep things simple. Create a file sample_app.proto at the root of your project with the following content:

syntax = "proto3";

package sample_app;

service User {
  rpc Create (CreateRequest) returns (UserReply) {}
  rpc Get (GetRequest) returns (UserReply) {}
}

message UserReply {
  int32 id = 1;
  string first_name = 2;
  string last_name = 3;
  int32 age = 4;
}

message CreateRequest {
  string first_name = 1;
  string last_name = 2;
  int32 age = 3;
}

message GetRequest {
  int32 id = 1;
}

As you can see, our Protocol Buffer definition is fairly straightforward and easy to read. We define a service that exposes two RPC methods—Create and Get. We also define the types that each of those RPC calls takes as input and returns as a result. With the sample_app.proto file in place, we'll want to open up a terminal and run the following:

$ protoc --elixir_out=plugins=grpc:./lib sample_app.proto

You'll notice that this command produces a file lib/sample_app.pb.ex with several modules within it. If you look carefully, you'll see that the code that was generated is the Elixir representation of the sample_app.proto file that we wrote. It contains all of the types that we defined along with the RPC method definitions. With our auto-generated
code in place, let's get to work on the actual RPC handlers.

As previously mentioned, we'll be using an Agent to persist state across gRPC calls instead of a database, for the sake of simplicity. Our agent will have the ability to look up users via their ID, and will also be able to create new users. Create a file lib/user_db.ex with the following code which provides that functionality:

defmodule UserDB do
  use Agent

  def start_link(_) do
    Agent.start_link(
      fn ->
        {%{}, 1}
      end,
      name: __MODULE__
    )
  end

  def add_user(user) do
    Agent.get_and_update(__MODULE__, fn {users_map, next_id} ->
      updated_users_map = Map.put(users_map, next_id, user)

      {Map.put(user, :id, next_id), {updated_users_map, next_id + 1}}
    end)
  end

  def get_user(id) do
    Agent.get(__MODULE__, fn {users_map, _next_id} ->
      Map.get(users_map, id)
    end)
  end
end

With that in place, we can create our RPC handlers for creating and getting users. Create a file lib/sample_app.ex with the following content:

defmodule SampleApp.Endpoint do
  use GRPC.Endpoint

  intercept GRPC.Logger.Server
  run SampleApp.User.Server
end

defmodule SampleApp.User.Server do
  use GRPC.Server, service: SampleApp.User.Service

  def create(request, _stream) do
    new_user =
      UserDB.add_user(%{
        first_name: request.first_name,
        last_name: request.last_name,
        age: request.age
      })

    SampleApp.UserReply.new(new_user)
  end

  def get(request, _stream) do
    user = UserDB.get_user(request.id)

    if user == nil do
      raise GRPC.RPCError, status: :not_found
    else
      SampleApp.UserReply.new(user)
    end
  end
end

Our file defines two modules. The SampleApp.Endpoint module defines the gRPC server and provides the handler module to service requests. The SampleApp.User.Server module contains the actual implementations of the two RPC calls that we defined. You'll notice that for each of the handlers, we provide the correct return type (as defined in our Protocol Buffer file). When we encounter an error (in this case, looking up a user that doesn't exist), we raise a GRPC.RPCError with the appropriate status code.

All that is left now is to start up our Agent and our gRPC server, and we're good to go. Open up lib/sample_app/application.ex and ensure that your process children list looks like this:

children = [
  UserDB,
  {GRPC.Server.Supervisor, {SampleApp.Endpoint, 50051}}
]

With that in place, you should be able to run mix grpc.server from the terminal to start your gRPC server. In another terminal session (and from within the project directory), you should be able to use grpcurl commands to interact with your application:

$ grpcurl -plaintext -proto sample_app.proto -d '{"first_name": "Bob", "last_name": "Smith", "age": 40}' localhost:50051 sample_app.User.Create
{
  "id": 1,
  "firstName": "Bob",
  "lastName": "Smith",
  "age": 40
}

$ grpcurl -plaintext -proto sample_app.proto -d '{"id": 1}' localhost:50051 sample_app.User.Get
{
  "firstName": "Bob",
  "lastName": "Smith",
  "age": 40
}

$ grpcurl -plaintext -proto sample_app.proto -d '{"id": 2}' localhost:50051 sample_app.User.Get
ERROR:
  Code: NotFound
  Message: Some requested entity (e.g., file or directory) was not found

Conclusion

Thanks for sticking with me to the end. Hopefully, you learned a thing or two about gRPC and how to go about using it within an Elixir application. If you would like to learn more about gRPC or any of the tools that I mentioned, I suggest going through the following resources:

Guest author Alex Koutmos is a Senior Software Engineer who writes backends in Elixir, frontends in VueJS and deploys his apps using Kubernetes. When he is not programming or blogging he is wrenching on his 1976 Datsun 280z.

P.S. If you'd like to read Elixir Alchemy posts as soon as they get off the press, subscribe to our Elixir Alchemy newsletter and never miss a single post!

Structured logging in Elixir using Loki

Alex Koutmos — Tue, 10 Mar 2020 20:16:50 +0000

Intro

In this blog post, we'll be talking about what exactly logging is, and how it differs from monitoring. We'll also cover some best practices that should be considered when logging from within your application. In addition, we'll cover the differences between structured and unstructured logs. And finally we'll set up a simple Phoenix application backed by Postgres and run Loki along side it to collect all of our application logs. Without further ado, let's dive right into things!

What is logging?

Logging is the practice of outputting and capturing information or events from your running applications. This information can be anything ranging from simple statements denoting which execution paths have been taken, to stack traces in the event of an error, to incoming request payloads. The goal is to capture information that can be leveraged when debugging application issues. In order to help categorize your various log messages, you can typically split up your log messages into four levels (these are the four levels available when using the Elixir Logger module [1]):

Debug: Debug log messages are primary used as a development-time tool. These messages can help you during development so you have a in depth look at what your application is doing and what code paths are being executed. You can generally be more verbose with your debug log messages as this log level will be pruned from production logs.
Info: Information level logs are typically used when you want to capture something that can be possibly used to help debug downstream errors. For example, capturing input parameters passed to an endpoint (with sensitive data scrubbed of course) can be useful if perhaps a 500 error occurs downstream. In short, informational messages are not used when something out of the ordinary happens, but can be instrumental in figuring out what caused something out of the ordinary to occur.
Warning: A warning level log should be used when something bad happens, but there is a possibility of recovering. For example, if you are making calls to a 3rd party API and leverage some sort of retry logic, you can emit a warning message whenever an API call attempt fails.
Error: An error level log should be used to denote that something bad has happened and there is no chance of recovering. This log message will generally be used to output some information that can help in debugging a customer impacting issue. To expand upon the previous example, you would use this log level when all retries have been exhausted and your 3rd party API is not responding.

Why do you need logging?

Logging is an important tool to have at your disposal as an engineer, as it allows you to root cause and resolve issues more quickly and with greater accuracy. By having a good logging solution at your disposal to search through logs and good logging coverage from within your application, you can reduce your MTTI and MTTR ("Mean Time to Identify" and "Mean Time to Repair" respectively [3]). Few things in business are as important as keeping your customers happy, and that means being able to identify issues and turn around a solution within a reasonable amount of time. This is exactly what an effective logging solution nets you. In fact, logging is so crucial to this process, that it is one of the Three Pillars of Observability 2. While I have not written anything about tracing (yet), I do have a series of posts related to monitoring and metrics collection in Elixir: Prometheus, PostGIS and Phoenix Part 1 and Prometheus, PostGIS and Phoenix Part 2. Feel free to check those out!

The obvious question that may come to mind is, how do all three of these differ and when should I use one over the other? In short:

Logging: Logging deals with individual events and contains textual information associated with each of those events. For example, route POST /users was invoked with payload {"user_name": "Bob", "password": "[FILTERED]"}
Metrics: Metrics aim to aggregate events and measure specific data points over time. For example, how many requests per second is our API servicing during peak hours and how many of these requests are resulting 200s, 400s, and 500s.
Tracing: Tracing is the process of aggregating and associating related events together. For example, if you are making an API call to a service, but that service needs to talk to 3 other services in order to fetch the information you need, you could leverage tracing to gain insight into the various service hops and their associated latencies all within the context of your original request. Tracing could also be localized to a single application and what are you tracing is the time across various function calls.

The next question that may come to mind is, do I need all of these things in place today in order to effectively debug issues? In an ideal world, your Engineering organization would have all three in place, and everything would be magical, and you would never actually need these tools since your code never has any bugs and there are never any network outages :). In the real world though, and from my experience, standing up these three pillars usually happens in phases. Logging being the first phase given that it has the lowest barrier to entry, and it is perhaps the most valuable (initially at least). By logging the various events occurring throughout the system along with relevant details, you can simulate or aggregate logs to produce metrics and tracing...but I would highly discourage against doing that given that it becomes very difficult to maintain that sort of implicit solution. Either way, logging will immediately prove beneficial given that you will be able to see errors, their frequency (via search), and some messages that may help in tracking down the root cause of problems.

Show me the code!

With all the theory and best practice out of the way, it is time to dive into the code and setup a Phoenix application alongside a self hosted logging solution. Our logging solution of choice for this tutorial will be Loki [4]. You could also use EFK, ELK, or Graylog to name a few. But I find those to be fairly heavy duty compared to Loki (not to turn this in a Vim versus Emacs style battle...they are all great tools and fit slightly different needs). In addition, we'll also be leveraging the Elixir library LoggerJSON [5] to create structured logs which will be captured by Loki. As a small tangent, logs can come in several different forms:

Unstructured: With unstructured logs, the event messages appear as free-form text and are not structured in any particular way. For example, your log message could be something like: /api/some-resource returned a 201
Structured: With structured logging, your event messages are encapsulated within a standardized format. For example, your messages can be contained within a JSON blob along with some additional metadata {"time": "2020-02-14 02:03:48.578284", "status": 500, "message": "Something bad happened!"}.
Binary: Binary log messages are not human readable like unstructured and structured logs. Instead, binary log messages are meant to be consumed by an application. For example in MySQL, binary logs can be used to keep track of all of the changes that have occurred to a database. This can later be used to update a database replica.

The application we will be working on for this tutorial is a basic used car search engine. We'll leverage LoggerJSON to generate structured logs from Ecto queries and Phoenix requests along with our own custom log statements. For the sake of simplicity, we'll seed our database with all of our dummy data, and will only have a single endpoint to power our search functionality. With that all being said, let's jump right into it!

Step 1: Create a new Phoenix project with required dependencies

To begin, let's start off by making sure we have the latest Phoenix project generator installed. Do that by running the following in the terminal:

$ mix archive.install hex phx_new 1.4.15

For this tutorial we'll also require Docker to be installed along with the Loki Logging driver. You can install Docker via https://www.docker.com/ or your package manager of choice. With Docker installed and running on your machine, run the following command to install the Loki Logging driver:

$ docker plugin install grafana/loki-docker-driver:latest --alias loki --grant-all-permissions

With this Docker logging driver installed, you will be able to ship logs directly from your Docker container to Loki without the use of Promtail [6] or Fluentd [7]. While you would not do this in production, this greatly simplifies the tutorial and let's us focus on the application logging aspect versus operationalizing this setup. If you would like to disable this plugin after completing the tutorial, simply run:

$ docker plugin disable loki

With our system dependencies satisfied, let's turn our focus to our Phoenix application. Since this is strictly a RESTful backend API, we don't require Webpack or any of the HTML related boilerplate. As a result, we'll create a new Phoenix project with the following command + flags and leave out the unused bits:

$ mix phx.new auto_finder --no-html --no-webpack --no-gettext

Let's also go ahead and add all of the Elixir dependencies that we will need for this tutorial. Open up your mix.exs file and make sure that you dependencies list looks like so:

defp deps do
  [
    {:phoenix, "~> 1.4.15"},
    {:phoenix_pubsub, "~> 1.1"},
    {:phoenix_ecto, "~> 4.0"},
    {:ecto_sql, "~> 3.3.4"},
    {:postgrex, ">= 0.0.0"},
    {:jason, "~> 1.0"},
    {:plug_cowboy, "~> 2.0"},
    {:logger_json, "~> 3.3.0"}
  ]
end

After updating the dependencies list, go ahead and run mix deps.get to fetch all the dependencies and to also update your mix.lock file (we'll need the lock file up to date for when we leverage it in our docker-compose.yml file). With all that done, let's turn our focus to our docker-compose.yml file and get our development stack up and running!

Step 2: Set up Docker Compose stack

In order to run our development stack, we'll require a few services to be running. Those services include:

Loki: In short, Loki is a log aggregation database. Log messages are forwarded to Loki via your log collector of choice (Fluentd, Promtail, Logstash, etc) and are then stored for future querying. Loki operates in the same space as Elasticsearch, but makes some different design decisions. Where as Elasticsearch indexes everything that comes into it (roughly speaking), Loki instead leverages labels (much like Prometheus) to group similar log streams. When it comes to searching, Loki will then effectively Grep the relevant log streams using your labels to get you applicable results. While this may not seem as robust as Elasticsearch, in practice it means that Loki is far cheaper to run from an Operations perspective, and it is still more than quick enough (from my experience) given you are able to segment your search by time range and labels. Whether this tradeoff is applicable to your organization or project is something you'll have to investigate. Like most things in Engineering...it depends :D.
Grafana: Grafana is an opensource tool which can be used to visualize your application/system metrics (these metrics are stored Prometheus for example). In addition to visualizing metrics, it also provides a fontend to view logs that are stored in your log aggregation database (in our case Loki is that database). I have an in depth tutorial series on setting up an Elixir application with Prometheus and Grafana if you are interested: Prometheus, PostGIS and Phoenix Part 1.
Postgres: Postgres is an SQL database and honestly one of my favorite tools to leverage when developing backends. I rarely find myself needing another database unless I have specific needs like timeseries, full text search, graph data structures (and even a lot of these are often doable in Postgres these days via extensions and the rich standard library).

With our technical decisions explained, let's put together our docker-compose.yml file to bring up all of our applications:

version: '3.7'

services:
  auto_finder_app:
    image: elixir:1.10.2
    command: >
      /bin/sh -c 'apt-get update && apt-get install -y inotify-tools &&
      mix local.hex --force &&
      mix local.rebar --force &&
      mix deps.get &&
      mix ecto.setup &&
      mix phx.server'
    ports:
      - '4000:4000'
    depends_on:
      - loki
      - postgres
    working_dir: /app
    logging:
      driver: loki
      options:
        loki-url: 'http://loki:loki@localhost:3100/loki/api/v1/push'
    volumes:
      - ./config:/app/config:ro
      - ./lib:/app/lib:ro
      - ./priv:/app/priv:ro
      - ./mix.exs:/app/mix.exs:ro
      - ./mix.lock:/app/mix.lock:rw
      - elixir-deps:/app/deps/
      - elixir-build:/app/_build/

  postgres:
    image: postgres:12.2
    ports:
      - '5432:5432'
    volumes:
      - postgres-data:/var/lib/postgresql/data
    environment:
      POSTGRES_PASSWORD: postgres
      POSTGRES_USER: postgres

  loki:
    image: grafana/loki:master-f106c09
    volumes:
      - ./docker/loki/loki-config.yml:/etc/loki/loki-config.yml
      - loki-data-index:/loki-index
      - loki-data-chunks:/loki-chunks
    expose:
      - 3100
    ports:
      - '3100:3100'
    command: '-config.file=/etc/loki/loki-config.yml'

  grafana:
    image: grafana/grafana:6.6.2
    ports:
      - '3000:3000'
    volumes:
      - grafana-data:/var/lib/grafana
      - ./docker/grafana/:/etc/grafana/provisioning/

volumes:
  elixir-deps: {}
  elixir-build: {}
  postgres-data: {}
  loki-data-index: {}
  loki-data-chunks: {}
  grafana-data: {}

You'll notice that we reference the Loki logging driver in the auto_finder_app definition and also set the Loki URL to http://loki:loki@localhost:3100/loki/api/v1/push. The reason that we reference the Loki instance via localhost is that we are accessing it from the perspective of the host OS instead of as an adjacent container. With that said, the rest of the docker-compose.yml file is pretty standard and should be familiar if you are accustomed to Docker and Docker Compose. Let's also add the Grafana and Loki configuration files that are volume mounted. Create the file docker/grafana/datasources/datasource.yml with the following contents (this will automatically provision Loki as a datasource so that it is available in Grafana upon initialization):

# config file version
apiVersion: 1

datasources:
  - name: loki
    type: loki
    access: proxy
    url: http://loki:3100
    editable: false

With Grafana taken care of, let's create the file docker/loki/loki-config.yml which will be used to configure Loki. For detailed documentation on the Loki configuration options, see [8]

auth_enabled: false

server:
  http_listen_port: 3100

ingester:
  lifecycler:
    address: 127.0.0.1
    ring:
      kvstore:
        store: inmemory
      replication_factor: 1
    final_sleep: 0s
  chunk_idle_period: 5m
  chunk_retain_period: 30s

schema_config:
  configs:
    - from: 2018-04-15
      store: boltdb
      object_store: filesystem
      schema: v9
      index:
        prefix: index_
        period: 168h

storage_config:
  boltdb:
    directory: /loki-index

  filesystem:
    directory: /loki-chunks

limits_config:
  enforce_metric_name: false
  reject_old_samples: true
  reject_old_samples_max_age: 168h

chunk_store_config:
  max_look_back_period: 0

table_manager:
  chunk_tables_provisioning:
    inactive_read_throughput: 0
    inactive_write_throughput: 0
    provisioned_read_throughput: 0
    provisioned_write_throughput: 0
  index_tables_provisioning:
    inactive_read_throughput: 0
    inactive_write_throughput: 0
    provisioned_read_throughput: 0
    provisioned_write_throughput: 0
  retention_deletes_enabled: false
  retention_period: 0

As it stands now, we cannot yet start our application stack as our Phoenix app is not configured properly. We need to set the appropriate configurations to have the Phoenix app connect to Postgres and to also log via LoggerJSON. Let's tackle these things next!

Step 3: Configure application logging in Phoenix app

In order to have Elixir's Logger generate structured logs, we'll need to configure Logger to leverage the LoggerJSON library that we specified in our mix.exs file. We'll also want to disable the default Phoenix logger so that all of our logs are in the JSON format. Open up your config/config.exs file and make sure that it looks like the following:

use Mix.Config

config :auto_finder,
  ecto_repos: [AutoFinder.Repo]

# Configures the endpoint
config :auto_finder, AutoFinderWeb.Endpoint,
  url: [host: "localhost"],
  secret_key_base: "<YOU_SECRET_KEY_BASE_GOES_HERE>",
  render_errors: [view: AutoFinderWeb.ErrorView, accepts: ~w(json)],
  pubsub: [name: AutoFinder.PubSub, adapter: Phoenix.PubSub.PG2],
  live_view: [signing_salt: "60QXfDxg"]

# Configures Elixir's Logger
config :logger, backends: [LoggerJSON]
config :auto_finder, AutoFinder.Repo, loggers: [{LoggerJSON.Ecto, :log, [:info]}]

config :logger_json, :backend,
  metadata: [:file, :line, :function, :module, :application, :httpRequest, :query],
  formatter: AutoFinder.LoggerFormatter

# Use Jason for JSON parsing in Phoenix
config :phoenix, :json_library, Jason
config :phoenix, :logger, false

# Import environment specific config. This must remain at the bottom
# of this file so it overrides the configuration defined above.
import_config "#{Mix.env()}.exs"

We'll also want to update our config/dev.exs file so that we are able to access our instance of Postgres running from within Docker. Update the AutoFinder.Repo configuration section so that it looks like the following:

# Configure your database
config :auto_finder, AutoFinder.Repo,
  username: "postgres",
  password: "postgres",
  database: "auto_finder_dev",
  hostname: "postgres",
  show_sensitive_data_on_connection_error: true,
  pool_size: 10

You'll notice that we reference a formatter within the AutoFinder namespace when configuring :logger_json in our config/config.exs file. By default, the LoggerJSON library comes with a GCP (Google Cloud Platform) specific formatter, so we will create our own formatter to keep our logs more generic. Our log formatter is almost an identical clone of the default LoggerJSON formatter [9], but with all the GCP specific bits removed. Below is the formatter which you should put in lib/log_formatter.ex:

defmodule AutoFinder.LoggerFormatter do
  @moduledoc """
  Custom Formatter
  """
  import Jason.Helpers, only: [json_map: 1]

  @behaviour LoggerJSON.Formatter

  @processed_metadata_keys ~w[pid file line function module application]a

  @severity_levels [
    debug: "DEBUG",
    info: "INFO",
    warning: "WARNING",
    warn: "WARNING",
    error: "ERROR"
  ]

  for {level, gcp_level} <- @severity_levels do
    def format_event(unquote(level), msg, ts, md, md_keys) do
      Map.merge(
        %{
          time: format_timestamp(ts),
          severity: unquote(gcp_level),
          message: IO.iodata_to_binary(msg)
        },
        format_metadata(md, md_keys)
      )
    end
  end

  def format_event(_level, msg, ts, md, md_keys) do
    Map.merge(
      %{
        time: format_timestamp(ts),
        severity: "DEFAULT",
        message: IO.iodata_to_binary(msg)
      },
      format_metadata(md, md_keys)
    )
  end

  defp format_metadata(md, md_keys) do
    LoggerJSON.take_metadata(md, md_keys, @processed_metadata_keys)
    |> maybe_put(:error, format_process_crash(md))
  end

  defp maybe_put(map, _key, nil), do: map
  defp maybe_put(map, key, value), do: Map.put(map, key, value)

  defp format_process_crash(md) do
    if crash_reason = Keyword.get(md, :crash_reason) do
      initial_call = Keyword.get(md, :initial_call)

      json_map(
        initial_call: format_initial_call(initial_call),
        reason: format_crash_reason(crash_reason)
      )
    end
  end

  defp format_initial_call(nil), do: nil

  defp format_initial_call({module, function, arity}),
    do: "#{module}.#{function}/#{arity}"

  defp format_crash_reason({:throw, reason}) do
    Exception.format(:throw, reason)
  end

  defp format_crash_reason({:exit, reason}) do
    Exception.format(:exit, reason)
  end

  defp format_crash_reason({%{} = exception, stacktrace}) do
    Exception.format(:error, exception, stacktrace)
  end

  defp format_crash_reason(other) do
    inspect(other)
  end

  # RFC3339 UTC "Zulu" format
  defp format_timestamp({date, time}) do
    [format_date(date), ?T, format_time(time), ?Z]
    |> IO.iodata_to_binary()
  end

  defp format_time({hh, mi, ss, ms}) do
    [pad2(hh), ?:, pad2(mi), ?:, pad2(ss), ?., pad3(ms)]
  end

  defp format_date({yy, mm, dd}) do
    [Integer.to_string(yy), ?-, pad2(mm), ?-, pad2(dd)]
  end

  defp pad3(int) when int < 10, do: [?0, ?0, Integer.to_string(int)]
  defp pad3(int) when int < 100, do: [?0, Integer.to_string(int)]
  defp pad3(int), do: Integer.to_string(int)

  defp pad2(int) when int < 10, do: [?0, Integer.to_string(int)]
  defp pad2(int), do: Integer.to_string(int)
end

Before getting to our business logic related code, we have two additional changes that we need to make. One of those being that we need to add a Telemtry handler for Ecto events so that we can log information when queries are executed against Postgres. The other being that we need to edit the plug pipeline in our endpoint.ex file so that we can trigger the LoggerJSON plug when requests are serviced. Let's first open up our application.ex file and add the following snippet within our start/2 callback.

def start(_type, _args) do
  children = [
    ...
  ]

  # Attach Telemetry handler for Ecto events
  :ok =
    :telemetry.attach(
      "logger-json-ecto",
      [:auto_finder, :repo, :query],
      &LoggerJSON.Ecto.telemetry_logging_handler/4,
      :info
    )

  ...
end

Afterwards, open up the lib/auto_finder_web/endpoint.ex file and add a call to the LoggerJSON plug by adding plug LoggerJSON.Plug in the plug chain. I usually add the logger plug before the plug Plug.Static call.

With all that in place, our foundation work is complete and it is now time to turn our attention to our business logic.

Step 4: Implementing our business logic

The first thing we'll need to do is create a migration for our used cars table. Run the following command to create an Ecto migration:

$ mix ecto.gen.migration used_cars

After running that command, open up the generated file and add the following contents (if you feel ambitious, you can add indices for the various searchable fields :)):

defmodule AutoFinder.Repo.Migrations.UsedCars do
  use Ecto.Migration

  def change do
    create table(:used_cars) do
      add :make, :string
      add :model, :string
      add :year, :integer
      add :mileage, :integer
      add :price, :integer

      timestamps()
    end
  end
end

With our migration in place, let's also update our priv/repo/seeds.exs script so that we can seed our database with dummy data:

alias AutoFinder.{Repo, UsedCars.UsedCar}

car_selection = [
  {"Acura", ~w(ILX TLX RLX RDX MDX NSX), 15_000..35_000},
  {"Honda", ~w(Accord Civic CR-V Odyssey Passport), 10_000..25_000},
  {"Nissan", ~w(GT-R 370Z Titan Leaf Sentra), 25_000..50_000},
  {"Mazda", ~w(MX-5 CX-3 CX5 CX-9), 15_000..25_000},
  {"Chevrolet", ~w(Camaro Corvette Colorado Silverado), 25_000..50_000},
  {"Ford", ~w(Escape Explorer Mustang Focus), 15_000..25_000},
  {"Audi", ~w(A4 Q3 A6 Q7 R8 S3 S4 RS5), 20_000..50_000},
  {"BMW", ~w(M2 M3 M5 X4 X7), 20_000..50_000},
  {"Subaru", ~w(Impreza Legacy Forester BRZ WRX), 15_000..25_000},
  {"Porsche", ~w(Taycan Panamera MAcan Cayenne Carrera Cayman), 40_000..70_000},
  {"Ferrari", ~w(812 F8 488 GTC4 Portofino), 150_000..250_000}
]

1..1_000
|> Enum.each(fn _ ->
  {make, models, price_range} = Enum.random(car_selection)
  model = Enum.random(models)
  price = Enum.random(price_range)
  year = Enum.random(2015..2020)
  mileage = Enum.random(10_000..60_000)

  %UsedCar{}
  |> UsedCar.changeset(%{make: make, model: model, price: price, year: year, mileage: mileage})
  |> Repo.insert!()
end)

You probably noticed that we made a call to a changeset and referenced a struct that we have not yet implemented. Let's go ahead and add those now. Create a file lib/auto_finder/used_cars/used_car.ex with the following contents:

defmodule AutoFinder.UsedCars.UsedCar do
  use Ecto.Schema

  import Ecto.Changeset

  @fields ~w(make model year mileage price)a

  @derive {Jason.Encoder, only: @fields}

  schema "used_cars" do
    field :make, :string
    field :model, :string
    field :year, :integer
    field :mileage, :integer
    field :price, :integer

    timestamps()
  end

  def changeset(used_car, attrs \\ %{}) do
    used_car
    |> cast(attrs, @fields)
    |> validate_required(@fields)
  end
end

We'll also want to create a module that will act as the entry point into our "used car" context. Let's create a file lib/auto_finder/used_cars/used_cars.ex with the following contents:

defmodule AutoFinder.UsedCars do
  import Ecto.Query

  alias AutoFinder.{Repo, UsedCars.UsedCar}

  def get_used_cars(query_params) do
    base_query = from(used_car in UsedCar)

    query_params
    |> Enum.reduce(base_query, &handle_query_param/2)
    |> Repo.all()
  end

  defp handle_query_param({"make", make}, acc_query) do
    from used_car in acc_query, where: ilike(used_car.make, ^make)
  end

  defp handle_query_param({"model", model}, acc_query) do
    from used_car in acc_query, where: ilike(used_car.model, ^model)
  end

  defp handle_query_param({"min_year", min_year}, acc_query) do
    from used_car in acc_query, where: used_car.year >= ^min_year
  end

  defp handle_query_param({"max_price", max_price}, acc_query) do
    from used_car in acc_query, where: used_car.price <= ^max_price
  end

  defp handle_query_param({"max_mileage", max_mileage}, acc_query) do
    from used_car in acc_query, where: used_car.mileage <= ^max_mileage
  end
end

Our get_used_cars/1 function will be called from our controller and will be used to dynamically build the user's query. The function reduces on all of the user's search options, appending to the query additional where clauses. Once the query is built, a call to Repo.all/1 is made to fetch all the used cars that match the users search terms. As a side note, I find this pattern of dynamically building queries very flexible, easy to test, and clean from a readers perspective :).

With our context function in place, let's create a new controller at lib/auto_finder_web/controllers/used_car_controller.ex and add the following contents (for the purposes of this tutorial we'll omit input validation...but you should always do that in production):

defmodule AutoFinderWeb.UsedCarController do
  use AutoFinderWeb, :controller

  alias AutoFinder.UsedCars

  require Logger

  def index(conn, params) do
    results = UsedCars.get_used_cars(params)

    if results == [] do
      Logger.info(
        "Search #{inspect(params)} yielded no results...perhaps we should suggest something similar"
      )
    else
      Logger.info("Search #{inspect(params)} yielded some results...customer should be happy")
    end

    json(conn, results)
  end
end

You'll notice that we added some log statements to our controller to get a sense for how people are using our API and to have something that simulates the real world show up in Grafana once we make some requests against our API. With our controller in place, all that is left is to update our router.ex file so that we can hit our controller. Open up lib/auto_finder_web/router.ex and ensure that it looks like the following:

defmodule AutoFinderWeb.Router do
  use AutoFinderWeb, :router

  pipeline :api do
    plug :accepts, ["json"]
  end

  scope "/api", AutoFinderWeb do
    pipe_through :api

    get "/used_cars", UsedCarController, :index
  end
end

Step 5: Taking our application stack for a test drive

With all this in place, we are ready to fire up our Docker Compose stack and see it all in action. To recap a bit, we have configured LoggerJSON to capture logs from Ecto via Telemetry, Phoenix, and our own custom log statements. We have a Phoenix app that exposes a single endpoint that is used to search used cars in our database that is seeded with sample entries. And we have all of our stdout/stderr logs being forwarded to Loki via the Docker logging driver. Those logs will be surfaced via Grafana which is also up and running within our development stack. That being said, let's fire up the stack by running:

$ docker-compose up

In order to exercise the API a bit, you can run the following cURL commands and get some sample data as you can tell...I'm in the market for a Ferrari :D (I wish):

$ curl 'http://localhost:4000/api/used_cars?make=ferrari'
[{"make":"Ferrari","model":"812","year":2017,"mileage":46945,"price":244424}, ...

$ curl 'http://localhost:4000/api/used_cars?make=ferrari&model=F8&max_price=155000'
[{"make":"Ferrari","model":"F8","year":2016,"mileage":56021,"price":152403}, ...

$ curl 'http://localhost:4000/api/used_cars?make=ferrari&model=Portofino&min_year=2020'
[{"make":"Ferrari","model":"Portofino","year":2020,"mileage":21333,"price":169790}, ...

$ curl 'http://localhost:4000/api/used_cars?make=invalid_car_make'
[]

With some queries having been made to our backend API, we should be able to open up Grafana and see some logs regarding these queries. Navigate to http://localhost:3000/explore (the default username and password for Grafana are both admin). From the Log labels drop down list select compose_service > auto_finder_app. If everything has been set up properly, you should see something like the following:

Feel free to play around with the Explore section of Grafana. You can live tail logs, search [10], and expand log entries to see all of the details regarding the message. Luckily, Grafana and Loki are smart enough to break up the JSON messages so that it is easier to consume to structured log entries as seen below (one of the benefits of structured logging ;)):

As we can see, it if very easy to extract the detailed information regarding the SQL query that Ecto generated including the time it took to fetch the data from the database. As a bonus for making it through the entire tutorial, below is a screen capture of Grafana while tailing the logs from the container:

Closing thoughts

Well done and thanks for sticking with me to the end! We covered quite a lot of ground and hopefully you picked up a couple of cool tips and tricks along the way. To recap, we created a simple Phoenix backend API backed by Postgres for the purposes of searching for used cars. Our API makes use of LoggerJSON to create Ecto, Phoenix, and application structured logs. The JSON structured logs are forwarded to Loki via Loki's Docker logging driver. Our logs that are stored in Loki are then searchable via Grafana. With all this in place, we have one of our pillars of observability set up and can gain valuable insight into how our application is behaving.

While this worked out great for a local development environment, I would suggest looking into Promtail [6] or Fluentd [7] when running Loki (or any kind of log aggregation tool for that matter) in order to ship your logs out of your applications in a production setting. In regards to the Loki logging driver, If you would like to play around with the settings for it, or uninstall it reference the documentation [11].

Feel free to leave comments or feedback or even what you would like to see in the next tutorial. Till next time!

Additional Resources

Below are some additional resources if you would like to deep dive into any of the topics covered in the post.

How to Get Your Elixir Application Ready for CI/CD

Alex Koutmos — Thu, 06 Feb 2020 13:30:00 +0000

In today's post, we'll go over what continuous integration and continuous delivery are, the benefits that come along with employing CI/CD, and some best practices that you should follow. We'll also explore a wide array of Elixir ecosystem tools that can help you create top-notch CI pipelines. In order to experiment with a handful of the tools that we will be discussing, we'll use a Git hooks Elixir library to execute our CI/CD validation steps, but on our local machine.

Let's jump right in!

What is CI/CD and When Do You Use It?

Continuous integration is the process by which new features and bug fixes are frequently merged into mainline branches. For every change to the code base, the project’s validation suite is executed to ensure it's up to team standards and doesn't introduce any regressions.

Continuous delivery is the process by which validated code, once merged, generates a new build artifact. The build artifact with the new code is then automatically deployed to your non-production environments. Note that continuous delivery deploys to non-production environments. The act of deploying to production directly is called, confusingly enough, continuous deployment.

CI and CD are critical tools to have at your disposal as they allow you and your team to move faster and with a higher degree of confidence. They enable you to catch bugs early on and ensure that they don't impact customers.

How Does CI/CD Influence Application Design?

With the vocabulary explained, let's move on to discussing the 12 Factor App. This is a collection of techniques and guidelines that can be used to create sustainable and scalable web services. As the name implies, there are 12 concepts that make up the manifesto. For the purposes of CI/CD, we'll focus on "Config" and "Codebase", but I highly recommend reading up on the 12 Factor App if you are unfamiliar with it.

The Codebase section states that a single code repository should contain only a single application. If your entire system is made up of multiple applications, then each of those applications should be contained within their own repositories. This is particularly relevant to CI/CD since, by following this convention, it is very easy to ensure that only the necessary applications are built, tested, and deployed. If multiple applications are contained within the same repository, it becomes a bit more difficult to ascertain which application should be built, tested, and deployed.

The Config section states that any kind of configuration data, credentials and secrets need to be kept out of the code. Instead, these bits of data should be set in the application's environment via environment variables. This is important from a CI/CD perspective since you will be able to spin up supporting services for testing, and can easily point your application to those services via runtime configuration. You can do this with a wide array of services such as Postgres, Redis and RabbitMQ. Not only will it make your application portable between higher up environments, but it will also make it easy to test.

CI/CD Best Practices

Before jumping into some Elixir ecosystem tooling, let's review some best practices that we should adhere to when designing our CI/CD pipelines (for clarification, a CI/CD pipeline is defined as a series of steps that are performed in order to take code from commit to deployment):

Ensure that your tests are deterministic, as having an unreliable suite of tests can have terrible consequences for the velocity of a team and the reliability of the software that is being validated.
Run cheap validations first and expensive validations last so that the CI/CD pipeline fails as early as possible. You don't want to take up CI/CD server resources running tests suites over and over again when you have other cheaper issues that need to be sorted out.
Promote the same build artifact to higher-level environments as it gives you the greatest amount of confidence that what you tested is what you are deploying. By leveraging the Config philosophies from the 12 Factor App, it should be possible to promote the same build artifact by only changing some environment variables.

Elixir Tools for Continuous Integration

Luckily, there are many tools at our disposal that come out of the box with Elixir. Some of these tools include:

mix compile --warnings-as-errors — Running this will return a non-zero exist status if your code contains any warnings.
mix xref unreachable --abort-if-any — Running this will return a non-zero exit status if your code makes any references to functions/modules that do no exist. If you are using Elixir 1.10+ this Mix task has been deprecated and its functionality has been rolled into mix compile --warnings-as-errors.
mix xref deprecated --abort-if-any — Running this will return a non-zero exit status if your code leverages any functions that have been marked as deprecated. If you are using Elixir 1.10+ this Mix task has been deprecated and its functionality has been rolled into mix compile --warnings-as-errors.
mix format --check-formatted — Running this will return a non-zero exist status if any of your source files do not adhere to the format configuration specified in your .formatter.exs. This helps ensure that the codebase has a uniform look and feel for all team members.
mix test — Elixir has an amazing built-in testing framework called ExUnit. By running the preceding command you can execute all of your project's tests and the exit status will be non-zero if any tests failed.

After you've incorporated the aforementioned items into your CI flow, it is time to reach for some community tools. Luckily, the Elixir ecosystem is packed full of great tools! Below is a list of a few of the tools that I use day in and day out:

Credo is a configurable static analysis tool that checks your code for design, readability, and consistency issues. In addition, it also enforces its own style guide, which can help a codebase stay uniform even if there are many developers in and out of the code.
Dialyxir is an amazing static analysis tool that we inherit from Erlang. In order to streamline its usage in Elixir, the Dialyxir library provides a nice wrapper around Dialyzer, which makes it easier to consume from an Elixir perspective. If you make heavy usage of typespecs then this is a great tool for you to ensure that you aren't introducing any type related errors into your code.
Doctor is a static analysis tool that focuses on scanning your project's documentation. It can be used to enforce the presence of typespecs, function docs, and module docs. In addition, it can be configured to return a non-zero exit status if documentation coverage falls below a certain threshold.
Bypass is a useful library for when you want to mock out external HTTP services for the purposes of your tests. Your mock server can be brought up during your tests and can return pre-canned responses.
Faker is an excellent library for when you need to generate fake data like names, address, phone numbers, etc.
ExMachina — If you leverage Ecto in your application and want a streamlined way of generating test data along with associations, then ExMachina is what you are looking for. When paired with Faker, ExMachina can make a great test data creation tool.

Elixir Tools for Continuous Delivery

Once your application has been tested and statically analyzed, you'll want to deploy it somehow. Below are a few options that are available to you for doing so:

Mix Releases — As of Elixir 1.9, the ability to create self-contained application releases has been brought into Elixir core (historically, you would need to use separate tools like Distillery). Elixir is interesting from a deployment standpoint given that you have the ability to package your application and the Erlang virtual machine (the BEAM) all within a build artifact. You need to make sure that the machine you build on is the same as the machine that you deploy to, but seeing as the application+runtime are all bundled, you will not need Elixir or Erlang installed on the target machine.
Docker — Like many programming languages available today, Elixir plays nicely with Docker and it is a relatively straightforward task to create a Docker image with your application. You can leverage the base Elixir Docker image and run your application via Mix inside the container for development purposes, or you can leverage Mix Releases and multistage builds to create lightweight images with only your application and the bundled Erlang virtual machine.
eDeliver is a tool that can be used to deploy Elixir applications to remote hosts using hot-code upgrades. By using hot-code upgrades, you can update your application with zero downtime on one or more hosts.

Performing CI Validations on Your Local Machine

In order to play around with some of the ideas presented here without trying to learn a new CI/CD system, we'll instead experiment with validating a sample Elixir project using Git hooks. Leveraging Git hooks is a good habit to get into as it will help you validate your code locally before pushing it to your team's repository. It makes fixing any errors easier, given that you don't need to dig through logs to figure out why builds failed.

To begin, start by cloning a sample repository that I put together on GitHub:

$ git clone https://github.com/akoutmos/sample_math.git

Once we have our project cloned locally, we'll want to open up our mix.exs file and add the following dependency:

defp deps do
  [
    ...
    {:git_hooks, "~> 0.4.0", only: [:test, :dev], runtime: false}
  ]
end

With that in place, we can now open up our config/config.exs file and add the following (feel free to omit the mix xref tasks if you are using Elixir 1.10+):

...

if Mix.env() != :prod do
  config :git_hooks,
    verbose: true,
    hooks: [
      pre_commit: [
        tasks: [
          "mix clean",
          "mix compile --warnings-as-errors",
          "mix xref deprecated --abort-if-any",
          "mix xref unreachable --abort-if-any",
          "mix format --check-formatted",
          "mix credo --strict",
          "mix doctor --summary",
          "mix test"
        ]
      ]
    ]
end

With that in place, let's fetch our new dependency and attempt to commit our code via the terminal:

$ mix deps.get
$ mix git_hooks.install
$ git add .
$ git commit -m "Added git hooks to sample math project"
↗ Running hooks for :pre_commit
✔ `mix clean` was successful
Compiling 1 file (.ex)
warning: variable "num_2" is unused (if the variable is not meant to be used, prefix it with an underscore)
  lib/sample_math.ex:10: SampleMath.sum/2

Compilation failed due to warnings while using the --warnings-as-errors option
× pre_commit failed on `mix`
** (exit) 1
    lib/mix/tasks/git_hooks/run.ex:175: Mix.Tasks.GitHooks.Run.error_exit/1
    lib/mix/tasks/git_hooks/run.ex:128: Mix.Tasks.GitHooks.Run.run_task/3
    (elixir) lib/enum.ex:783: Enum."-each/2-lists^foreach/1-0-"/2
    (elixir) lib/enum.ex:783: Enum.each/2
    lib/mix/tasks/git_hooks/run.ex:63: Mix.Tasks.GitHooks.Run.run/1
    (mix) lib/mix/task.ex:331: Mix.Task.run_task/3
    (mix) lib/mix/cli.ex:79: Mix.CLI.run_task/2

As you may have guessed, there is an issue with our project that requires fixing! Luckily, our commit will not go through until we are able to rectify the issue. I'll leave the fixing of issues in your capable hands, but as we can see there is a problem with the compile step because of the presence of an unused variable. Our validation steps worked just as expected!

While we only performed the validation steps on our local machine, taking these same validation steps to an actual CI/CD pipeline is a relatively simple task. The Git hooks library that we leveraged allows us to run all of our configured steps via mix git_hooks.run all. In other words, we can run this command in our CI/CD solution of choice and validate that our code changes pass team standards. The benefit of this is that our CI/CD validation steps can be easily run locally for quick and easy debugging.

Summary

Thanks for sticking with me to the end! Hopefully, you learned a thing or two related to CI/CD and how to go about doing it with an Elixir application.

P.S. If you'd like to read Elixir Alchemy posts as soon as they get off the press, subscribe to our Elixir Alchemy newsletter and never miss a single post!

How to Use Broadway in Your Elixir Application

Alex Koutmos — Fri, 20 Dec 2019 16:16:41 +0000

In today's post, we will be covering the Elixir library named Broadway. This library is maintained by the kind folks at Plataformatec and allows us to create highly concurrent data processing pipelines with relative ease.

After an overview of how Broadway works and when to use it, we'll dive into a sample project where we'll leverage Broadway to fetch temperature data from https://openweathermap.org/ in order to find the coldest city on earth. Our Broadway pipeline will loop over a list of all the cities in the world from a GenStage producer that we write.

Let's jump right in!

What Is Broadway and When Should You Use It in Your Elixir App?

Broadway allows us to create highly concurrent data processing pipelines largely due to how it is built on top of GenStage. GenStage is another Elixir library that is used to construct an event/message exchange between processes. Specifically, GenStage provides the features necessary to create coordinated producer and consumer processes and ensure that the event pipeline is never flooded. Consumers subscribe to upstream producers and demand messages when they are free to do work. With this model, we can scale the number of consumer processes as is needed for the task at hand to achieve the performance we desire.

Issues with Using GenStage in Production

A major problem with using GenStage directly for a production-grade application is that the onus is on the developer to create the proper supervision tree to ensure that failures are handled properly. This is exactly where Broadway comes into play. Broadway provides the necessary abstractions on top of GenStage that you would leverage in a production context such as:

Rate limiting
Batching
Ordering and partitioning
Automatic restarts
Graceful shutdowns
Automatic message acknowledgment

Additional Supporting Libraries

In addition to the aforementioned features, Broadway also has several supporting libraries that allow you to use a message queue service such as a Broadway producer (some are maintained by Plataformatec and some are community maintained). At the time of writing this post, there are official Broadway Producers available for Amazon SQS, Google Pub/Sub and RabbitMQ. Support for Kafka is currently underway, but a release has not yet been cut for the project. The benefit of using Broadway Producers is that it abstracts away the problems that come along with managing a persistent and valid connection to the data source and provides a convenient way of using these data sources as the entry point into your data processing pipeline.

All this is well and good, but when should you reach for a tool like Broadway?

Broadway is a useful tool when the task at hand is embarrassingly parallel and spawning more processes yields positive results. It is also useful when you have an ETL (extract, transform and load) pipeline and want to break up the problem into discrete components and scale them independently. One example would be processing user image uploads. Increasing the number of workers will allow you to process more image uploads, and you can disconnect the image upload from the HTTP request/response cycle by leveraging a queuing system like RabbitMQ, SQS, etc.

How Does Broadway Work Internally?

As previously mentioned, Broadway leverages GenStage in order to orchestrate the event pipelines. If you look at the Broadway.Producer module, you'll notice the line use GenStage at the top. If you read further down into the module, you'll find implementation for the handle_subscribe/4, handle_demand/2, handle_cancel/3, and handle_call/3, callbacks (to name a few) that are defined in the GenStage behavior. The magic of Broadway comes into play in how it automatically creates a supervision tree that ensures that your pipeline is fault-tolerant and reliable. Below is the supervision tree from the world temperature application that we will be writing shortly:

Let's break the supervision tree down step by step. At the top of our tree, we have our project supervisor—WorldTemp.Supervisor—which is defined in our application.ex file. From there, we turn our focus to WorldTemp.TempProcessor as that is the Broadway module that we'll be creating (WorldTemp.CityProducer and WorldTemp.TempTracker are supporting GenServers). Further down the supervision tree, we get to WorldTemp.TempProcessor.Broadway.Supervisor. This supervisor is responsible for monitoring all the various components of Broadway and restarting them if any errors occur. The processes that WorldTemp.TempProcessor.Broadway.Supervisor supervises are listed below along with their purpose:

WorldTemp.TempProcessor.Broadway.ProducerSupervisor: This supervisor is responsible for monitoring the data producer processes. This supervision tree has a strategy of :one_for_one as it only needs to restart the particular data producer that is experiencing issues. All other producers can keep running if everything is okay.
WorldTemp.TempProcessor.Broadway.ProcessSupervisor: This supervisor is responsible for monitoring the worker processes that consume data from the producers. This supervision tree has a strategy of :one_for_all. The reason that it's not :one_for_one is that the processing callback functions that we write should be stateless and any errors that occur can be handled without crashing the process. If an error does occur to crash the process, it is likely that some internal bookkeeping related to Broadway has gone awry and all the consumers need to be restarted.
WorldTemp.TempProcessor.Broadway.Terminator: This process is responsible for the proper stoppage of your Broadway pipeline. It will notify all the consumer processes that they should not resubscribe to producers once they terminate and it also notifies all the producers to flush all of their current events and ignore any subsequent data requests.
WorldTemp.TempProcessor.RateLimiter: This process is optionally started if your pipeline requires a rate limiter. It is effectively a token bucket rate limiter and throttles how much work your processors can perform within a configurable time interval.

It is also important to note that the WorldTemp.TempProcessor.Broadway.Supervisor has a supervision policy of :rest_for_one. The reason for this is that if the producer supervision tree crashes, the parent supervisor can restart all the subsequent supervision trees and restore the pipeline back to a working fresh state.

Hands-on with Broadway

As previously mentioned, we will be creating a very simple Broadway based application which will read weather data from an API and then keep a running record of the coldest city on earth. We will also be using a new feature of Broadway (rate limiting) to ensure that we don't go over our free plan limit on https://openweathermap.org.

With that being said, let's jump right in. Start by creating a new project using mix new world_temp --sup. You'll want to use the --sup flag for convenience to set up the root application supervision tree. With your new project created, you want to change into the project directory and open up the mix.exs file. Update the dependencies to look like the following (we'll use my fork of Broadway as there is a pending issue with the rate limiter that I fixed in my fork):

defp deps do
  [
    {:httpoison, "~> 1.6"},
    {:jason, "~> 1.1"},
    {:broadway, github: "plataformatec/broadway", tag: "08497708e10867935f2e92351e4cde9e4a57135e"}
  ]
end

To fetch your dependencies, run mix deps.get in your project directory. From there, we'll want to create the file lib/city_producer.ex which will act as the data producer for our Broadway pipeline. For the purposes of this sample project, this data producer will be a GenStage based module, but you can use the other Broadway producers if that better serves your needs (Kafka, RabbitMQ, SQS, etc). In the lib/city_producer file, add the following:

defmodule WorldTemp.CityProducer do
  use GenStage

  require Logger

  def start_link(_args) do
    GenStage.start_link(__MODULE__, [], name: __MODULE__)
  end

  def init(_args) do
    {:producer, city_list()}
  end

  # If the demand is greater than the state of the GenState process,
  # readd the city list and reprocess
  def handle_demand(demand, state) when demand > length(state) do
    handle_demand(demand, state ++ city_list())
  end

  # Enough data is available in the GenStage's state, serve that to
  # the consumer
  def handle_demand(demand, state) do
    {to_dispatch, remaining} = Enum.split(state, demand)

    {:noreply, to_dispatch, remaining}
  end

  # List of cities for which to get weather data
  defp city_list do
    [
      {"Abu Dhabi", "United Arab Emirates"},
      {"Abuja", "Nigeria"},
      {"Accra", "Ghana"},
      {"Adamstown", "Pitcairn Islands"},
      {"Addis Ababa", "Ethiopia"},
      {"Algiers", "Algeria"},
      {"Alofi", "Niue"},
      ...
      # For full contents of this function go to Github project
      # https://github.com/akoutmos/world_temp
    ]
  end
end

Next, we'll want to create a module that fetches weather data from the OpenWeatherMap API. You'll want to go to https://openweathermap.org/ and create a free account to get an API key. Go ahead and create lib/temp_fetcher.ex with the following content:

defmodule WorldTemp.TempFetcher do
  require Logger

  @api_key "YOU_API_KEY_GOES_HERE"

  def fetch_data(city, country) do
    city
    |> generate_url(country)
    |> HTTPoison.get()
    |> handle_response()
  end

  defp handle_response({:ok, %HTTPoison.Response{status_code: 200, body: body}}) do
    body
    |> Jason.decode!()
    |> get_in(["main", "temp"])
  end

  defp handle_response(resp) do
    Logger.warn("Failed to fetch temperature data: #{inspect(resp)}")

    :error
  end

  defp generate_url(city, country) do
    "http://api.openweathermap.org/data/2.5/weather?q=#{city},#{country}&appid=#{@api_key}"
  end
end

With that in place, we need to create our Broadway consumer module to leverage this HTTP API wrapper. Create lib/temp_processor.ex with the following content. We'll go over what it does shortly:

defmodule WorldTemp.TempProcessor do
  use Broadway

  alias Broadway.Message

  def start_link(_opts) do
    Broadway.start_link(__MODULE__,
      name: __MODULE__,
      producer: [
        module: {WorldTemp.CityProducer, []},
        transformer: {__MODULE__, :transform, []},
        rate_limiting: [
          allowed_messages: 60,
          interval: 60_000
        ]
      ],
      processors: [
        default: [concurrency: 5]
      ]
    )
  end

  @impl true
  def handle_message(:default, message, _context) do
    message
    |> Message.update_data(fn {city, country} ->
      city_data = {city, country, WorldTemp.TempFetcher.fetch_data(city, country)}
      WorldTemp.TempTracker.update_coldest_city(city_data)
    end)
  end

  def transform(event, _opts) do
    %Message{
      data: event,
      acknowledger: {__MODULE__, :ack_id, :ack_data}
    }
  end

  def ack(:ack_id, _successful, _failed) do
    :ok
  end
end

The start_link/1 function defines the various options required by Broadway to orchestrate the data pipeline. You'll notice that the GenStage producer that we previously created is referenced in the producer keyword list section and we also defined a rate limiter to ensure that we don't go over our free tier usage on OpenWeatherMap. The transformer entry invokes the transform/2 function at the bottom of the module which is required boilerplate to format our incoming messages to a %Broadway.Message{} struct. An important thing to note here is that we specify 5 concurrent processors (if you recall from the supervision tree image earlier, there were 5 consumer processes). By changing that one value, you can determine how much concurrency/throughput you would like from your Broadway pipeline. The meat of our logic is in handle_message/3 where we retrieve our message (a city+country tuple), make an API call, and then update our rolling record of the coldest city.

Now let's create that WorldTemp.TempTracker module in lib/temp_tracker.ex with the following content:

defmodule WorldTemp.TempTracker do
  use Agent

  def start_link(_) do
    Agent.start_link(fn -> nil end, name: __MODULE__)
  end

  def get_coldest_city do
    Agent.get(__MODULE__, fn {city, country, temp} ->
      "The coldest city on earth is currently #{city}, #{country} with a temperature of #{
        kelvin_to_c(temp)
      }°C"
    end)
  end

  def update_coldest_city(:error), do: nil

  def update_coldest_city({_, _, new_temp} = new_data) do
    Agent.update(__MODULE__, fn
      {_, _, orig_temp} = orig_data ->
        if new_temp < orig_temp, do: new_data, else: orig_data

      nil ->
        new_data
    end)
  end

  defp kelvin_to_c(kelvin), do: kelvin - 273.15
end

This module is a relatively straight forward Agent-based module that allows us to retrieve the current coldest city, or update it if the provided value is lower than the currently set value.

Lastly, we need to update our lib/world_temp/application.ex file and add a couple of items to our supervision tree. If you named your modules the same as I did, your application.ex file should look like this:

defmodule WorldTemp.Application do
  @moduledoc false

  use Application

  alias WorldTemp.{CityProducer, TempProcessor, TempTracker}

  def start(_type, _args) do
    children = [
      TempTracker,
      CityProducer,
      TempProcessor
    ]

    opts = [strategy: :one_for_one, name: WorldTemp.Supervisor]
    Supervisor.start_link(children, opts)
  end
end

With all that in place, we can run iex -S mix from the command line, and be able to interact with our application:

iex(1) ▶ WorldTemp.TempTracker.get_coldest_city()
"The coldest city on earth is currently Bern, Switzerland with a temperature of 1.8100000000000023°C"
iex(2) ▶ WorldTemp.TempTracker.get_coldest_city()
"The coldest city on earth is currently Copenhagen, Denmark with a temperature of -0.37000000000000455°C"
iex(3) ▶ WorldTemp.TempTracker.get_coldest_city()
"The coldest city on earth is currently Copenhagen, Denmark with a temperature of -0.37000000000000455°C"
iex(4) ▶ WorldTemp.TempTracker.get_coldest_city()
"The coldest city on earth is currently Helsinki, Finland with a temperature of -1.8899999999999864°C"
iex(5) ▶ WorldTemp.TempTracker.get_coldest_city()
"The coldest city on earth is currently Roseau, Dominica with a temperature of -18.99999999999997°C"

By calling WorldTemp.TempTracker.get_coldest_city() periodically, we can see that as the Broadway processors work through the city+country list, the city with the coldest temperature changes. It may take a few minutes to run through the whole list given that we are only processing 60 cities a minute and our list has a length of 216 elements.

Summary

As we can see from the sample application that we've written, with relative ease, we were able to create a data processing pipeline for our Elixir application by using Broadway. With all of the abstractions given to us via Broadway, we can rest easy knowing that our pipeline will operate as intended and will recover from any issues if they arise. In addition, we have all the levers necessary to adjust the performance characteristics of our pipeline via some configuration.

Thanks for sticking with me to the end and if you would like to learn more about Broadway, I suggest going through the following resources:

P.S. If you'd like to read Elixir Alchemy posts as soon as they get off the press, subscribe to our Elixir Alchemy newsletter and never miss a single post!