• Blog:

  • Home
  • Ably News
  • Ably Engineering
  • Developers
  • Industry Trends
  • Hardest Aspects of Realtime Engineering
  • Scalable event streaming with Redis and Golang

    By: Thomas Camp 17 min read

    This blog has been updated as of the 15th July 2021 to include changes to the Ably Go client library.


    A common problem that our customers have is event streaming spikey traffic to their clients. For example, imagine a server which needs to stream details on cryptocurrency trades. The quantity of trades happening each second is likely to fluctuate, sometimes nothing could happen, other times perhaps thousands.

    In order to ensure reliable communication, it’s important that rate limits are in place on how many messages are to be sent to subscribed clients. If the number of messages being generated spikes beyond acceptable levels for a period of time, how do we ensure that we are able to handle it and provide these messages to clients in a timely manner?

    Event streaming is inherently intended to be an implementation of Pub/Sub which has a unidirectional flow of data. That is, data flows from Publishers into Consumers without the consumers ever having to publish anything themselves. This means that we have a lot more control when it comes to rate limiting all potential publishers, as in most event streaming scenarios you’ll have the publishers be servers directly within your control.

    With a combination of Redis and Ably, written in Go, we’ll look to create the following reliable event streaming structure.

    • We’ll have our trading server which will have bursts of messages to send to our clients.
    • We’ll have a Redis Container which will hold messages from our trading server
    • We’ll have a publishing server, which keeps track of the current rate of messages sent and consumes from Redis whenever it can publish to Ably
    • Clients are then subscribed to the Ably Channel that the Publisher publishes to and displays the information.

    We’ll have our core functionality (server, redis, publisher) all contained within a Kubernetes cluster.

    The stack

    Before we do that though, let’s go over what each of the tools we’ll be using are, and why we’re using them.

    Golang

    Go is an efficient, reliable programming language which is at its core based on the premise of scalability and concurrency. Go effectively implements Pub/Sub communication with its own channels to communicate between goroutines, with multiple goroutines able to both publish and subscribe to these channels. This core design to Go makes it one of the most well-fitted languages for extending this Pub/Sub functionality to networked Pub/Sub communication.

    Want to learn more about Golang and realtime? We've got you covered:

    - Guide to Pub/Sub in Golang
    - Building realtime apps with Golang and WebSockets: client-side considerations

    Redis

    Redis is one of the most popular and versatile in-memory data structure stores that exist. Due to being in-memory it’s extremely fast to retrieve messages from, and it is available in most programming languages.

    In this tutorial we’ll be using it as a Message Queue, holding messages from our Trading Server until our Publisher Server is ready to handle the messages.

    Ably

    Ably is a pub/sub messaging platform which handles the entire distribution process of messages, scaling up to billions of users. The main protocol used by the Ably’s client libraries is WebSockets, however Ably is inherently Protocol Agnostic, allowing for any realtime protocols such as SSE or MQTT to seamlessly work as well.

    We’ll be using Ably to distribute messages that are held by our Redis instance to as many clients as we want.

    Docker and Kubernetes

    Docker is an OS-level virtualization tool which allows one to create Images, which are representations of virtual machines in specific states which are set up with a Dockerfile. This is incredibly powerful as it allows for a Docker Image to be made of a specific virtual machine, and anyone who now uses this Docker Image will start with the exact same state. An instance of a Docker Image is called a Docker Container.

    Kubernetes takes Docker and makes it even more powerful by allowing for Docker Containers to be grouped together, scaled dynamically, and in general makes managing a collection of Docker Containers considerably easier.

    In this tutorial we’ll have a Docker Container for the Trading Server, for the Redis Instance, and three Containers for the Ably Publisher. We’ll manage these containers with Kubernetes, and demonstrate how to easily scale up and down the Ably Publishers as we need.

    Architecture diagram of the event streaming app built with Golang, Redis, Kubernetes, and Ably.
    Diagram representing what we will be building

    Starting off

    To start, let’s create a logical file structure to hold the various servers, container configuration files and more.

    Create a folder on your machine which will contain this project. Within it, create two folders: ably-publishing-server, and trading-server. We’ll be placing our code for both types of server in their respective folders.

    my-redis-go-app
    - ably-publishing-server
    - trading-server

    We’ll start by making our trading server. Within the trading-server folder create a go.mod file with the following contents:

    module go-redis-trading-server
    
    go 1.16
    
    require (
    github.com/go-redis/redis v6.15.9+incompatible // indirect
    )
    

    Here we’re specifying that to run our trading server we will need to import the redis module.

    Next, create a file called trading-server.go. This will contain our core code for our trading server. What we want from our server is:

    • Generate data on trades we want to be sent to all our clients. In this tutorial we’ll be simulating this data, however it should be easy to substitute in a real source
    • Send this data to Redis, which will hold the data until our publishing servers are ready to send them to Ably to be streamed to our clients

    Within the trading-server.go file, let’s add our basic structure:

    package main
    
    import (
    "log"
    "os"
    "fmt"
    "sync"
    "time"
    "math/rand"
    "strconv"
    
    "github.com/go-redis/redis"
    )
    
    var wg = &sync.WaitGroup{}
    
    func main() {
    
    }
    

    In addition to the usual imports and main function, we’re specifying a WaitGroup. We’ll be using this to stop the script from stopping until we explicitly call for it to.

    We’ll be using environment variables to make it easy for us to configure our server whenever we deploy it, so to make setup simpler let’s also create a function which will get an environment variable for us, or make use of a default value if it’s not been specified:

    func getEnv(envName, valueDefault string) string {
    value := os.Getenv(envName)
    if value == "" {
    return valueDefault
    }
    return value
    }
    

    Within the main function, let’s connect to Redis using the go-redis library. Create a function, getRedis, which will handle this for us:

    func getRedis() *redis.Client {
    // Create Redis Client
    var (
    host     = getEnv("REDIS_HOST", "localhost")
    port     = string(getEnv("REDIS_PORT", "6379"))
    password = getEnv("REDIS_PASSWORD", "")
    )
    
    client := redis.NewClient(&redis.Options{
    Addr:     host + ":" + port,
    Password: password,
    DB:       0,
    })
    
    // Check we have correctly connected to our Redis server
    _, err := client.Ping().Result()
    if err != nil {
    log.Fatal(err)
    }
    
    return client
    }
    

    Once we’ve got our Redis client set up, we’ll need a function to send our fake data through to Redis. Add the following:

    func publishingLoop(redisClient *redis.Client) {
    	queueKey := getEnv("QUEUE_KEY", "myJobQueue")
    	publishRate, _ := strconv.Atoi(getEnv("PUBLISH_RATE", "200"))
    	baseTradeValue := float64(200)
    
    	// Send a burst of messages to Redis every 5 seconds
    	ticker := time.NewTicker(5 * time.Second)
    	quit := make(chan struct{})
    	for {
    		select {
    		case <- ticker.C:
    			for i := 0; i < publishRate; i++ {
    				// Random test value varying +- 5 around the baseTradeValue
    				tradeValue := baseTradeValue + (rand.Float64() * 10 - 5)
    				redisClient.RPush(queueKey, fmt.Sprintf("%f", tradeValue))
    			}
    		case <- quit:
    			// If you want a way to cleanly stop the server, call quit <- true
    			// So this runs
    			ticker.Stop()
    
    			defer wg.Done()
    			return
    		}
    	}
    }
    

    All we’ve got here is a ticker which will cause us to send a large number of messages to Redis every 5 seconds. We’re using a burst of messages here so we can simulate a scenario where we have too many messages to handle as they’re generated without also resulting in a queue which continues to grow.

    Now let’s add the initialization of the Redis Client and the publishing functions to our main function, as well as our publishing function.

    func main() {
        client := getRedis()
    
        go publishingLoop(client)
    
        // Add an item to the wait group so the server keeps running
        wg.Add(1)
        wg.Wait()
    }
    

    With that we have our trading server, ready to start feeding data into Redis!

    Ably Publisher setup

    Next, let’s set up a server which will be responsible for taking messages from our Redis queue and streaming them to our clients via Ably. In order to ensure reliable service to subscribing clients Ably enforces a rate limit per channel and per connection on published messages, which for Free Accounts is 50. As part of our publisher, we will need to ensure that we maintain a publish rate of at most 50 messages per second.

    So that we can scale up the number of Ably Publishers in the future, the easiest method of enforcing a rate limit here is a Sliding Log. Effectively each time one of the Ably Publishers attempts to get a message to publish from Redis, it will check a log of previous messages sent in the last 1 second period. If the number of messages returned is greater than our limit, we wait a moment and try again.

    To get started on the Ably Publisher, within the ably-publisher folder let’s create a go.mod file with the following contents:

    module ably-go-publisher
    
    go 1.16
    
    require (
        github.com/ably/ably-go v1.2.0
        github.com/go-redis/redis v6.15.9+incompatible // indirect
    )
    

    Here we’re just including the go-redis module so we can interact with the Redis instance, and the ably-go library to interact with Ably.

    Next, create a file again in ably-publisher called publisher-server.go. Inside, let’s add our imports and the basic main function. In addition, we’ll include the getEnv and getRedis functions we used for the trading-server as we’ll be needing them:

    import (
    	"log"
    	"os"
    	"time"
    	"fmt"
    	"context"
    	"sync"
    	"strconv"
    
    	"github.com/go-redis/redis"
    	"github.com/ably/ably-go/ably"
    )
    
    var ctx = context.Background()
    
    var wg = &sync.WaitGroup{}
    
    func main() {
    
    }
    
    func getRedis() *redis.Client {
    	// Create Redis Client
    	var (
    		host     = getEnv("REDIS_HOST", "localhost")
    		port     = string(getEnv("REDIS_PORT", "6379"))
    		password = getEnv("REDIS_PASSWORD", "")
    	)
    
    	client := redis.NewClient(&redis.Options{
    		Addr:     host + ":" + port,
    		Password: password,
    		DB:       0,
    	})
    
    	_, err := client.Ping().Result()
    	if err != nil {
    		log.Fatal(err)
    	}
    
    	return client
    }
    
    func getEnv(envName, valueDefault string) string {
    	value := os.Getenv(envName)
    	if value == "" {
    		return valueDefault
    	}
    	return value
    }
    

    Next, to interact with Ably to send messages to clients, we’ll need to instantiate the Ably library. As part of connecting and using Ably, you will need to sign up for a free Ably Account. Once you have an account, you will need to get your API key from one of your account’s apps.

    Once you have your API key, add the following function to your code:

    func getAblyChannel() *ably.RealtimeChannel {
    	// Connect to Ably using the API key and ClientID specified above
    	ablyClient, err := ably.NewRealtime(
    		// If you have an Ably account, you can find
    		// your API key at https://www.ably.io/accounts/any/apps/any/app_keys
    		ably.WithKey(getEnv("ABLY_KEY", "No key specified")),
    		// ably.WithEchoMessages(false) // // Uncomment to stop messages you send from being sent back
    	)
    	if err != nil {
    		panic(err)
    	}
    
    	// Connect to the Ably Channel with name 'trades'
    	return ablyClient.Channels.Get(getEnv("CHANNEL_NAME", "trades"))
    }
    

    All we’re doing here is instantiating a connection to our Ably App, and then returning an instance of a channel from that app called ‘trades’. We’ll be using this channel to publish our trade updates to.

    Now we have functions to set up our connections to both Redis and Ably, we need to write some code to connect the two together with considerations for rate limits.

    For rate limit enforcement, we’ll be needing the following Redis commands, which will all be performed as part of a Redis Transaction (ensures that the commands occur sequentially and are atomic):

    • ZRemRangeByScore(redisLogName, startScore, endScore) - This will remove any values from our logs which fall within the startScore and endScore value. Our ‘score’ in this context will be timestamps of past publishes, so if we set the startScore to “0” and the “endScore” to 1 second ago, we’ll be removing all logs from before 1 second ago
    • ZCard(redisLogName) - ZCard will get the number of messages within a set, in this case the number of logs. If we’ve executed ZRemRangeByScore, this will be the number of messages published in the last second
    • ZAdd(redisLogName, {score, member}) - ZAdd adds a new score and member to the specified Redis key. In our case, when we are going to publish a new message to Ably we will add the current timestamp to our log

    In order to actually consume from the message queue we’ve created from our trading-server, we’ll need to make use of BLPop(timeout redisQueueName). This will remove the first item in our Redis queue and return it to us. Notably it is blocking, which means that if there are no messages on the queue, it will wait until there is a message to consume before continuing.

    After popping a message from the queue, we will then send it to our Ably Channel with a simple channel.Publish(‘trade’, messageData).

    Combining these all within a single transaction, we end up with the following function:

    func transactionWithRedis(client *redis.Client, channel *ably.RealtimeChannel) error {
    	// Redis key where messages from the trading server are stored
    	redisQueueName := getEnv("QUEUE_KEY", "myJobQueue")
    
    	// Values to be used for checking our Redis log key for the rate limit
    	redisLogName := redisQueueName + ":log"
    	now := time.Now().UnixNano()
    	windowSize := int64(time.Second)
    	clearBefore := now - windowSize
    	rateLimit, _ := strconv.ParseInt(getEnv("RATE_LIMIT", "50"), 10, 64)
    
    	err := client.Watch(func(tx *redis.Tx) error {
    		tx.ZRemRangeByScore(redisLogName, "0", strconv.FormatInt(clearBefore, 10))
    
    		// Get the number of messages sent this second
    		messagesThisSecond, err := tx.ZCard(redisLogName).Result()
    		if err != nil && err != redis.Nil {
    			return err
    		}
    
    		// If under rate limit, indicate that we'll be publishing another message
    		// And publish it to Ably
    		if messagesThisSecond < rateLimit {
    			err = tx.ZAdd(redisLogName, redis.Z{
    				Score:  float64(now),
    				Member: now,
    			}).Err()
    			if err != nil && err != redis.Nil {
    				return err
    			}
    
    			messageToPublish, err := tx.BLPop(0*time.Second, redisQueueName).Result()
    			if err != nil && err != redis.Nil {
    				return err
    			}
    
    			err = channel.Publish(ctx, "trade", messageToPublish[1])
    			if err != nil {
    				fmt.Println(err)
    			}
    		}
    
    		return err
    	}, redisLogName)
    
    	return err
    }
    

    With that done, we have all the functions we need to connect to both Redis and Ably, and send messages from our Redis Queue through Ably to clients! Add the following to the main function to make use of them:

    func main() {
        client := getRedis()
        channel := getAblyChannel()
    
        go func() {
            for {
                transactionWithRedis(client, channel)
            }
        }()
    
        wg.Add(1)
        wg.Wait()
    }
    

    Dockerizing and scaling with Kubernetes

    We now have the code we need for both parts of our system, so now we need to work out a logical way to deploy them both together, along with a Redis instance. To do this we’ll be using Docker to wrap each of our services along with a Redis instance, and then use Kubernetes to group this logic up and allow for easy scaling of them.

    Docker

    Starting this is very simple, we’ll just need to create a Dockerfile for both the trades server as well as the Ably publishing server which defines how each container should be constructed. Within the ably-publishing-server folder, create a file called Dockerfile with the following contents:

    FROM golang:latest as builder
    
    WORKDIR /app
    
    COPY go.mod go.sum ./
    
    # Download dependencies
    RUN go mod download
    
    # Copy from the current directory to the WORKDIR in the image
    COPY . .
    
    # Build the server
    RUN go build -o publisher publisher-server.go
    
    # Command to run the executable
    CMD ["./publisher"]
    

    Effectively all this is doing is building the publisher code we wrote, and then setting the Docker Image’s ‘run’ command to run the executable generated.

    We’ll do the same sort of script as a Dockerfile in the trading-server folder:

    FROM golang:latest as builder
    
    WORKDIR /app
    
    COPY go.mod go.sum ./
    
    # Download dependencies
    RUN go mod download
    
    # Copy from the current directory to the WORKDIR in the image
    COPY . .
    
    # Build the server
    RUN go build -o trading trading-server.go
    
    # Command to run the executable
    CMD ["./trading"]
    

    That’s all we need to build our Docker Images from our code! Download Docker to your machine, and then from inside the trading-server folder, run the following to build it:

    > docker build -t go-redis-trading-server:latest .
    

    You should now have a Docker Image containing your trading server! You can check that the image was created by running docker images to see what images exist on your machine.

    Let’s do the same for the Ably publishing server. In the ably-publishing-server folder, run the following:

    > docker build -t go-redis-publishing-server:latest .
    

    Kubernetes

    Now that we’ve made our Docker containers for our service, we can now use Kubernetes to orchestrate our system.

    What we’ll need to do is create a Deployment for a Redis Container, our Trades Container, and our Ably Publishing Container. Within each deployment, we’ll be able to optionally scale the number of Pods and thus the number of containers running of each type.

    Additionally, we’ll need to create a Service, which will define access to the Redis Pod from our other Pods.

    Let’s first create a file to contain our Redis Container and Service. Create a new folder in our project’s base folder called deployments, and in it add redis.yaml. Add the following to it:

    ---
    apiVersion: apps/v1
    kind: Deployment
    metadata:
      name: redis-deployment # Unique name for the deployment
      labels:
        app: redis # Label used to select this deployment
    spec:
      selector:
        matchLabels:
          app: redis
      replicas: 1 # Number of pods to run in this deployment
      template:
        metadata:
          labels:
            app: redis
        spec: # Spec for the Docker Container inside of the Pod
          containers:
          - name: redis-container
            image: redis # Use the generic Redis Docker Image
    ---        
    apiVersion: v1
    kind: Service
    metadata:
      name: redis-service
      labels:
        app: redis
    spec:
      ports:
      - port: 6379
        targetPort: 6379
      selector: # Pods with the 'app: redis' label will use this service
        app: redis
    

    Here we have two sections defined: a Redis Deployment, and a Redis Service. The deployment effectively defines that we will have a single instance of the official Docker Redis Container, pulled from the Docker Hub. We tag this container and its Pods with  ‘app: redis’ so the Service can identify them and be applied correctly.

    The Service defined below the Deployment specification simply states that the Redis Pod should be made available on the port 6379.

    Next, let’s create Deployment specifications for our trades and ably publishing containers. In order for these specs to correctly use the Docker Images you’ve generated you’ll need to have the Image hosted somewhere for the Deployment to pull from eventually. The easiest place to do this is the Docker Hub. If you’d prefer to skip pushing your Images to a repository you can also use the Images of this code which have already been made and are available as ‘tomably/trading-server:latest’ and ‘tomably/go-redis-ably-publisher:latest’ on the Docker Hub.

    Let’s create the trading server deployment first. Create a new file in the deployment` folder called trading-deployment.yml with the following contents:

    ---
    apiVersion: apps/v1
    kind: Deployment
    metadata:
      name: trading-server
    spec:
      replicas: 1 # Number of pods to have within this deployment
      selector:
        matchLabels:
          app: trading-server
      template: # Template for Pods within this deployment
        metadata:
          labels: # The label used for pods within this deployment
            app: trading-server
        spec: # Details on the Docker Container to use in each Pod
          containers:
          - name: trading-server
            image: tomably/trading-server:latest
            imagePullPolicy: Always
            env: # Environment variables passed to the container
              - name: REDIS_HOST
                value: redis-service
              - name: REDIS_PORT
                value: "6379"    
    

    Here we’re again specifying that we want to have a single instance (replicas) of the Image we’re pulling. We’re also specifying some environment variables that’ll be used by the container.

    Create another file in the same folder called ably-publisher-deployment.yml. We’ll use the exact same structure, except we’ll set the replicas to 3. This should mean that when we eventually use this configuration file, we’ll have three instances of our publisher running. If our rate limiter is working correctly, they should respect the shared rate limit between themselves when publishing.

    As part of the environment variables we’re also adding our Ably API key. Make sure to replace the placeholder INSERT_API_KEY_HERE text with the API key you obtained earlier in the tutorial.

    ---
    apiVersion: apps/v1
    kind: Deployment
    metadata:
      name: ably-publisher
    spec:
      replicas: 3 # Number of pods to have within this deployment
      selector:
        matchLabels:
          app: ably-publisher
      template: # Template for Pods within this deployment
        metadata:
          labels: # The label used for pods within this deployment
            app: ably-publisher
        spec:
          containers:
          - name: ably-publisher
            image: tomably/go-redis-ably-publisher:latest
            imagePullPolicy: Always
            env: # Environment variables passed to the container
              - name: REDIS_HOST
                value: redis-service
              - name: REDIS_PORT
                value: "6379"
              - name: ABLY_KEY
                value: "INSERT_API_KEY_HERE"
              - name: RATE_LIMIT
                value: "50"
    

    With that done we should have everything we need to run our project! In order to run our Kubernetes Cluster, we’ll need two things: Kubectl to run Kubernetes commands, and Minikube to run the cluster on.

    You’ll need to download kubectl as described on the Kubernetes website, and then ensure you’ve done so correctly by running kubectl version --client.

    Next, Download Minikube, and then run in your terminal minikube start to start up an instance of it.

    Once you have that, we’ll use kubectl to deploy our deployments and service as described in our configuration files! Run the following commands to have kubectl read and use our configuration files to construct our cluster:

    > kubectl apply -f redis.yml
    > kubectl apply -f ably-publisher-deployment.yml
    > kubectl apply -f trading-deployment.yml
    

    If everything has gone according to plan, if you run kubectl get pods, you should see something like the following returned:

    NAME                               READY   STATUS    RESTARTS   AGE
    ably-publisher-6b6cb8c796-8px6w    1/1     Running   0          10s
    ably-publisher-6b6cb8c796-mmh7p    1/1     Running   0          10s
    ably-publisher-6b6cb8c796-nrxqp    1/1     Running   0          10s
    redis-deployment-6f6964f4c-8gwsm   1/1     Running   0          10s
    trading-server-7c5d445747-mrn54    1/1     Running   0          10s
    

    If any of the pods returned aren’t Running, and instead are in a state of Error or CrashLoopBackOff, you can try checking what’s wrong by running kubectl logs POD_NAME, replacing POD_NAME with the name of the pod in question.

    If you go to your Ably App’s Dev Console and connect to the ‘trades’ channel (or whichever channel you’ve renamed it to), you should see the trades coming through.

    If you wish to delete any of the deployments, you can run the following command, making sure to change DEPLOYMENT_NAME to the relevant deployment name (trading-server, ably-publisher, or redis-deployment in this tutorial).

    > kubectl delete deployment DEPLOYMENT_NAME
    

    Subscribing to the data

    Now that we have our data going into Ably, let’s create a very basic webpage to consume from it. Create a new file called client.html, and add the following code to it, making sure to replace the INSERT_API_KEY_HERE text with your Ably API key:

    <html>
    <body>
    <h1>Trades</h1>
    <p>This page is subscribed to the channel 'trades' in Ably, and should add these updates under this text.</p>
    <ul class="row" id="messages"></ul>
    
    </body>
    
    <!-- Include the latest Ably Library  -->
    <script src="https://cdn.ably.io/lib/ably.min-1.js"></script>
    
    <!-- Instance the Ably library  -->
    <script type="text/javascript">
    let realtime = new Ably.Realtime("INSERT_API_KEY_HERE"); /* ADD YOUR API KEY HERE */
    
    /* Subscribe to messages on the trades channel */
    let channel = realtime.channels.get("trades");
    channel.subscribe(function(msg) {
    show(msg.data);
    });
    
    function show(msg) {
    let list = document.createElement("LI");
    let text = document.createTextNode(msg);
    list.appendChild(text);
    document.getElementById("messages").appendChild(list);
    }
    </script>
    </html>
    

    If you open this it’ll be a very bare web page, which is subscribed to the Ably channel ‘trades’. Assuming you’re publishing data into the channel from our cluster, you should see the trades start coming in.

    Webpage subscribed to the Ably channel ‘trades’

    Conclusion

    With that we’re done! We now have an easily scalable infrastructure for streaming large amounts of traffic easily to clients. This is really just scraping the surface of possibilities, but should act as a good starting point for building good tooling.

    Event streaming can benefit from many other features, such as checking the presence of clients on channels before publishing or processing data for them, enabling history on channels so clients can check for previous events, or even forwarding events on to Webhooks and Cloud Functions for further processing.

    If you’d like to try this with existing real data, you can find real cryptocurrency pricing for free in the Ably Hub.

    All of the code for this project can be found on GitHub.