DEV Community: Lucas Pinheiro

Where is my SIGTERM, Docker?

Lucas Pinheiro — Sat, 21 Jan 2023 16:44:50 +0000

Originally published here.

This issue has robbed me of a good half hour this week. And this is not the first time that has happened. Previously, I just fixed it and moved on. This time, I'll write about it here, and hopefully, that'll make it stick in my brain.

The problem

I was running the Semaphore agent in a Kubernetes deployment. To do so, I had to put the agent in a Docker container. Here's what my Dockerfile looked like:

FROM ubuntu:20.04

# ... a lot of commands not relevant to the issue at hand ...
# Copy the agent binary to the container
COPY build/agent /app/agent

# Start the container using the agent binary
CMD /app/agent start --config-file /app/config/semaphore-agent.yml

When the Kubernetes deployment is created, the agent starts just fine.

In an ideal world, when the agent is interrupted, it should receive a SIGTERM signal and then disconnect from the API before completely shutting down. So, when deleting the Kubernetes deployment, that's what I was expecting to happen.

But that didn't happen. The agent didn't get any signals and continued its happy life without disruptions. That pod remained in the Terminating state for some time until the terminationGracePeriodSeconds was reached. Then, Kubernetes would send it a SIGKILL and move on.

The problem was that the agent needed that SIGTERM to clean up after itself properly. And it didn't. Where did that signal go?

Hunting a SIGTERM

I was pretty confident that the issue was not with the Semaphore agent. I looked at the Kubernetes configuration, and everything looked alright. I had run into this issue before, so I had a hunch that the issue was in the Dockerfile.

I looked at it but couldn't find anything wrong with it. Alright, let's go to the docs:

The CMD instruction has three forms:

CMD "executable","param1","param2"

CMD "param1","param2"

CMD command param1 param2 (shell form)

I laughed when I read that. I remembered going into the same docs to find the root cause of the same issue before. The sentence "this is the preferred form" was almost like a punch in my stupid face.

One last time, for the whole class

Docker has different forms for the CMD directive: a shell form and an exec form. Those forms are present in other directives too, like RUN and ENTRYPOINT.

The main difference between the two forms is that for shell form commands, the command runs in a shell (sh -c <command>). That means that if you are using the shell form to run another executable, the shell process will create a sub-process to run that executable. But the primary process in the container will still be the shell process.

If you look at the processes in the container, this is what you see:

USER  PID  COMMAND
root    1  /bin/sh -c /app/agent start --config-file /app/config/semaphore-agent.yml
root    6  /app/agent start --config-file /app/config/semaphore-agent.yml

Finding the SIGTERM

When Kubernetes sends a SIGTERM to the container, the shell is the process receiving it, not the executable. And the shell doesn't propagate that signal to the executable. To guarantee that the executable is the process receiving any signals coming from Kubernetes, the exec form needs to be used:

-CMD /app/agent start --config-file /app/config/semaphore-agent.yml
+CMD ["/app/agent" "start" "--config-file" "/app/config/semaphore-agent.yml"]

Now, if you look at the processes in the container, there's no additional sh -c process:

USER  PID  COMMAND
root    1  /app/agent start --config-file /app/config/semaphore-agent.yml

Friendly reminder

When you run into an issue, you will likely run into the same issue again in the future. If you just fix it and move on, it is also very likely that you will not remember how to fix it once you see it again.

This holds for programming, but I suspect it may hold for other parts of life as well. When finding the solution to a problem, it is always a good idea to document it. Your future self (and possibly other people) will thank you.

Cron jobs, everywhere

Lucas Pinheiro — Sat, 14 Jan 2023 20:02:30 +0000

I have used Kubernetes CronJobs for multiple purposes. I have used AWS EventBridge to schedule Lambda functions based on a schedule. I wrote several distinct Spring schedulers for various applications. And yet, I had never faced the need to use the father of them all: the UNIX cron job.

Well, there is a first time for everything. And just like Atahualpa and the Inca people before me, I had to create an actual cron job. Here are my notes on how to do it, on several different operating systems.

Ubuntu

Cron jobs are defined in crontab files. The location of these crontab files depends on its type: user or system.

user crontabs

To manage user crontab files, you use the crontab program directly:

crontab -e to create/edit it.
crontab -l to check its contents.
crontab -r to remove it.
Use the -u {{USER}} option to refer to a specific user's crontab.
All these files are located in /var/spool/cron/crontabs.

system crontabs

To manage system crontab files, you need to handle the files directly.

The main crontab file is located at /etc/crontab, but additional crontab files can be put in the /etc/cron.d directory.

You may also have a few extra directories available: cron.hourly, cron.daily, cron.weekly, cron.monthly. They do precisely what you are thinking. Put a script in one of these folders if you want it to execute with the folder's frequency.

Format

A crontab file is just a simple text file with a bunch of lines like this:

* * * * * {USER} {COMMAND}

Each line defines a cron job.

The {USER} column is only required for system crontab files.

Cron also keeps track of the modification time on its crontab files, so no reload is necessary when they change. But if you have trust issues, reload it with /etc/init.d/cron reload.

I never quite remember what each column means, so I always use https://crontab.guru/ when writing one.

Troubleshooting

By default, the cron service logs are stored in the /var/log/syslog global logging file. But you won't find the cron job's stdout and stderr there because the stdout/stderr for a cron job goes to the user's mailbox, in /var/mail/{USER}.

Inspecting the user's mailbox for a cron job is not ideal, so a usual pattern is to redirect the outputs of the cron job to a file, like:

* * * * * myuser /tmp/my-script.sh >> /tmp/cron-job.log

Also, if you want to ensure nothing goes to the mailbox, use MAILTO="" in your crontab file.

MacOS

Cron is also available in macOS. There is a launchd daemon located at /System/Library/LaunchDaemons/com.vix.cron.plist which manages the cron daemon. Initially, I was confused by the daemon's name (com.vix.cron). But apparently, the version of cron we use today is known as the Vixie cron, so that name makes sense, I guess.

Just put some cron job lines into a /etc/crontab file, and that launchd daemon will make sure your cron jobs run. You can't use the cron.d directory, though. You can also use the crontab CLI to manage the user's crontabs; those are located at /usr/lib/cron/tabs/{USER}.

Windows

Windows don't have cron. But Windows does offer support for scheduled tasks. If using PowerShell, you can use the following cmdlets to create scheduled tasks:

Here's an example:

# Our scheduled task runs a powershell script, so we use the powershell command.
$action = New-ScheduledTaskAction `
  -Execute 'powershell.exe' `
  -Argument '-NonInteractive -NoLogo -NoProfile -ExecutionPolicy bypass -File "C:\test.ps1"'

# Run a task now, and repeat after 1 minute, forever.
$trigger = New-ScheduledTaskTrigger `
  -Once `
  -At (Get-Date) `
  -RepetitionInterval (New-TimeSpan -Minutes 1)

# The default one is good enough for most cases.
$settings = New-ScheduledTaskSettingsSet

# Create the scheduled task object and register it.
$task = New-ScheduledTask -Action $action -Trigger $trigger -Settings $settings
Register-ScheduledTask `
  -TaskName 'My useless scheduled task' `
  -InputObject $task `
  -User $UserName `
  -Password $Password

Alternatively, the schtasks command can be used.

Wrapping up

I had to figure out how to run these scheduled tasks on different types of machines to implement a custom health check for EC2 instances. My cron job was supposed to run every minute to guarantee that a specific binary was running on the EC2 instance and mark the instance as unhealthy if not. This pull request has the final solution.

Where is my PATH, launchD?

Lucas Pinheiro — Thu, 12 Jan 2023 12:07:54 +0000

Originally published here.

While working on the macOS support for agent-aws-stack, I ran into a small issue with launchD. My scenario was: a Go binary executed as a launchD daemon. That daemon, upon termination, executes (in a non-login shell) a bash script:

token=$(curl -X PUT -H "X-aws-ec2-metadata-token-ttl-seconds: 60" --fail --silent --show-error --location "http://169.254.169.254/latest/api/token")
instance_id=$(curl -H "X-aws-ec2-metadata-token: $token" --fail --silent --show-error --location "http://169.254.169.254/latest/meta-data/instance-id")
region=$(curl -H "X-aws-ec2-metadata-token: $token" --fail --silent --show-error --location "http://169.254.169.254/latest/meta-data/placement/region")

aws autoscaling terminate-instance-in-auto-scaling-group \
    --region "$region" \
    --instance-id "$instance_id" \
    --no-should-decrement-desired-capacity

All the script does is terminate the EC2 instance and remove it from the auto-scaling group.

In a Linux machine, as a systemD unit, the script executes successfully. However, in macOS, as a launchD daemon, it doesn't. The script doesn't find the AWS CLI, exits with exit code 127, and displays that error message we all know and love:

aws: command not found

The AWS CLI is located at /usr/local/bin/aws, but our PATH when the script is executed is PATH=/usr/bin:/bin:/usr/sbin:/sbin. For some reason, launchD is not including /usr/local/bin in the PATH it exposes to its daemons, as systemD does.

The reason it works on systemD

You can see the environment systemD exposes to its system and user units using the show-environment command:

$ systemctl show-environment
LANG=C.UTF-8
PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin:/snap/bin

$ systemctl --user show-environment
HOME=/home/ubuntu
LANG=C.UTF-8
LOGNAME=ubuntu
PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin:/usr/games:/usr/local/games:/snap/bin:/snap/bin
SHELL=/bin/bash
SYSTEMD_EXEC_PID=1260
USER=ubuntu
XDG_RUNTIME_DIR=/run/user/1000
XDG_DATA_DIRS=/usr/local/share/:/usr/share/:/var/lib/snapd/desktop
DBUS_SESSION_BUS_ADDRESS=unix:path=/run/user/1000/bus

Where does that environment come from? From environment generators. SystemD comes with a default one which loads the environment from some special places.

What about launchD?

I was expecting launchD to have at least some environment setup for its daemons, but it doesn't. At least, not a default one, like systemD. There are a few different ways to configure it, though.

1. In the .plist daemon configuration

In our launchD daemon configuration (located at /Library/LaunchDaemons), we can explicitly add the PATH variable we want:

<!-- rest of plist -->
<key>EnvironmentVariables</key>
<dict>
  <key>PATH</key>
  <string>/usr/local/bin:/usr/bin:/bin:/usr/sbin:/sbin</string>
</dict>
<!-- rest of plist -->

2. Using launchctl setenv

From launchctl's man page:

setenv key value
    Specify an environment variable to be set on all future processes launched by launchd in the caller's context.

This would require restarting our daemon.

3. Using launchctl config

From launchctl's man page:

config system | user parameter value
      Sets persistent configuration information for launchd(8) domains. Only the system domain and user domains may be configured. The location of the persistent storage is an implementation detail, and changes to that storage should only be made through this subcommand. A reboot is required for changes made through this subcommand to take effect.

      Supported configuration parameters are:

      umask    Sets the umask(2) for services within the target domain to the value specified by value. Note that this value is parsed by
               strtoul(3) as an octal-encoded number, so there is no need to prefix it with a leading '0'.

      path     Sets the PATH environment variable for all services within the target domain to the string value. The string value should
               conform to the format outlined for the PATH environment variable in environ(7). Note that if a service specifies its own PATH,
               the service-specific environment variable will take precedence.
               NOTE: This facility cannot be used to set general environment variables for all services within the domain. It is intentionally scoped to the PATH environment variable and nothing else for security reasons.

This would require a reboot, though.

Note: previously (on macOS versions < 10.10), you could use /etc/launchd.conf, but that doesn’t work on newer versions.

Conclusion

As the only thing I needed was to use the AWS CLI, I temporarily added the /usr/local/bin to the PATH in the script itself and called it a day.

In hindsight, I could've just done that in the first place and seen if the script would work. But it's nice to understand what is going on before choosing a simpler fix.

Other resources

Read the documentation first

Lucas Pinheiro — Sat, 19 Mar 2022 20:34:16 +0000

Originally published at: https://lucaspin.github.io/powershell/encoding/utf-8/2022/03/19/read-the-docs-first.html

The following story contains high quantities of sadness (or humor, I guess). It displays my stubbornness to read a piece of documentation while looking for a quick solution to my problem. I decided to document it here to remind myself of this. And since we all learn better with a story, here's mine.

The problem

I have been working with Windows and PowerShell for a few weeks now. It was part of a need for one of our projects to support Windows. I had never touched Windows before, so a few challenges arose from that little journey, the last one (so far) being the one that surprisingly reverberated with me the most, and it had nothing to do with Windows and PowerShell at all.

Here is what I needed to do:

Write PowerShell commands in a file
Execute that file with PowerShell, using Go's exec.Command

It is straightforward. Except for one issue: I needed to support UTF-8 encoded characters, and that was not working. For example, if the file had this command to be executed:

Write-Host "特定の伝説に拠る物語の由来については諸説存在し。特定の伝説に拠る物"

The result was this:

ç‰¹å®šã®ä¼èª¬ã«æ‹ ã‚‹ç‰©èªžã®ç ±æ¥ã«ã¤ãã¦ã¯è«¸èª¬å˜åœ¨ã—ã€‚ç‰¹å®šã®ä¼èª¬ã«æ‹ ã‚‹ç‰©

Show me the code

This is the function I was using:

func createFile(filePath, content string) error {
  file, err := os.OpenFile(filePath, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0644)
  if err != nil {
    return err
  }

  defer file.Close()

  _, err = file.Write([]byte(content))
  if err != nil {
    return err
  }

  return nil
}

It creates a file with some content. If you ever wrote or read Go code, you have seen this code thousands of times. Here's an example of how that function was being used:

package main

import (
  "fmt"
  "os"
  "os/exec"
  "path/filepath"
)

func main() {
  fmt.Println(executeCommand())
}

func executeCommand() string {
  // create file with unicode characters
  path := filepath.Join(os.TempDir(), "myscript.ps1")
  _ = createFile(path, `Write-Host "特定の伝説に拠る物語の由来については諸説存在し。特定の伝説に拠る物"`)

  // execute file using powershell
  cmd := exec.Command("powershell", "-NoProfile", "-NonInteractive", "-File", path)
  output, err := cmd.CombinedOutput()
  if err != nil {
    fmt.Printf("Error executing script: %v\n", err)
  }

  return string(output)
}

Investigation

I went straight to Google and started looking for things that could lead me to a solution. A few of those searches led to Stack Overflow answers which tangentially touched the issue. I started following those paths. After a few hours, none of them gave me a solution.

I was getting frustrated. Ok, stop. Let's rewind and start over. What's in the file?

user@windows-machine > type .\script.ps1
Write-Host "ç‰¹å®šã®ä¼èª¬ã«æ‹ ã‚‹ç‰©èªžã®ç”±æ¥ã«ã¤ã„ã¦ã¯è«¸èª¬å˜åœ¨ã—ã€‚ç‰¹å®šã®ä¼èª¬ã«æ‹ ã‚‹ç‰©"

What?! That's not what I wrote and that's not what I see in my MacOS machine (Windows is running in a vagrant box):

user@macos-machine > type script.ps1
Write-Host "特定の伝説に拠る物語の由来については諸説存在し。特定の伝説に拠る物"

Ok, MacOS understands the file one way, and Windows understands it another. What if I execute the same PowerShell command but without a file?

user@windows-machine > powershell -NoProfile -NonInteractive -Command 'Write-Host "特定の伝説に拠る物語の由来については諸説存在し。特定の伝説に拠る物"'
特定の伝説に拠る物語の由来については諸説存在し。特定の伝説に拠る物

We're onto something, but this doesn't make much sense. Why does PowerShell interpret the same file differently?

Reading the documentation

I went to the PowerShell documentation, searched for "encoding" and ended up at the character encoding page. I discovered something called the byte-order-mark, also known as the BOM, which is a signature that takes a few bytes at the start of the file. Its purpose is to let a reader know of a few essential things: the UTF format in use (UTF-8, UTF-16, or UTF-32) and the endianness, in the case of the latter two.

I also encountered this:

Creating PowerShell scripts on a Unix-like platform or using a cross-platform editor on Windows, such as Visual Studio Code, results in a file encoded using UTF8NoBOM. These files work fine in PowerShell, but may break in Windows PowerShell if the file contains non-Ascii characters.

That sentence precisely describes what I was doing. It also explains why the issue I was seeing was happening:

If you need to use non-Ascii characters in your scripts, save them as UTF-8 with BOM. Without the BOM, Windows PowerShell misinterprets your script as being encoded in the legacy "ANSI" codepage. Conversely, files that do have the UTF-8 BOM can be problematic on Unix-like platforms. Many Unix tools such as cat, sed, awk, and some editors such as gedit don't know how to treat the BOM.

This was my :face-palm: moment.

Fixing the problem

After screaming in despair for a few moments, I fixed the problem. Since I was only concerned about UTF-8, I included the UTF-8 BOM in the file being executed, turning my createFile() function into this:

func createFile(filePath, command string) error {
  file, err := os.OpenFile(filePath, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0644)
  if err != nil {
    return err
  }

  defer file.Close()

  if runtime.GOOS == "windows" {
    file.Write([]byte{0xEF, 0xBB, 0xBF})
  }

  _, err = file.Write([]byte(command))
  if err != nil {
    return err
  }

  return nil
}

Issue is fixed. Everybody is happy. End of story.

Conclusion

While this post documents a PowerShell behavior, which might be helpful again for that purpose in the future, I am writing this to remind myself (and whoever is reading) that reading the documentation should be the first thing you do when you encounter an issue.

Looking for answers in stack overflow can be helpful, but it can sometimes be a problem. It can be distractive and lead us into paths we shouldn't be pursuing at all. If I had gone to the documentation page first, I would've saved myself a few good hours.

Documentation is critical. Good documentation has all the context you need. Also, good documentation has search capabilities, so you don't need to read the whole thing or search it yourself for what you want.

What if the documentation is terrible? Well, then I guess you can search for other sources of information (or cry in the bathroom).

Atomicity in Redis operations

Lucas Pinheiro — Wed, 21 Jul 2021 22:19:59 +0000

Originally published at: https://lucaspin.github.io/redis/databases/2021/07/21/atomicity-in-redis-operations.html

When working with databases, atomicity is always something to have in mind. I recently started working with Redis again and came across a little problem I found worth sharing over here.

You need to store elements in a list in Redis. More specifically, you need to append elements to the end of a list using an index. if the list length is above the index, you trim it before appending. if not, you just append to it.

There's not a built-in command to perform that operation available, so let's define an Append() function. I'll be using Golang and this redis library to do it:

func Append(client *redis.Client, key string, elements []string, index int64) error {
  length, _ := client.LLen(context.TODO(), key).Result()
  if length > index {
    _, err := client.LTrim(context.TODO(), key, 0, index-1).Result()
    if err != nil {
      return err
    }
  }

  _, err := client.RPush(context.TODO(), key, elements).Result()
  if err != nil {
    return err
  }

  return nil
}

Note that the Append() operation has 3 sub-operations in it: LLen(), LTrim(), RPush(). Each sub-operation is atomic in itself and contains a whole trip from the application running the code to the Redis cluster. But there is a problem with this approach: the Append() operation isn't atomic, even though its sub-operations are, which can lead to a few synchronization issues. For example, consider this code:

var INITIAL_VALUES = []string{"1", "2", "3", "4", "5"}
var ELEMENTS_TO_APPEND = []string{"4", "5"}

// create client
ctx := context.TODO()
client := redis.NewClient(&redis.Options{
  Addr:     "0.0.0.0:6379",
  Username: "",
  Password: "",
  DB:       0,
})

// initialize key
key := uuid.NewString()
log.Printf("Initializing %s", key)
client.RPush(ctx, key, INITIAL_VALUES).Result()

// call Append() twice right about the same time
go Append(client, key, ELEMENTS_TO_APPEND, 3) // #1
go Append(client, key, ELEMENTS_TO_APPEND, 3) // #2

// wait for both Append() calls to finish and check result
time.Sleep(time.Second)
elements := client.LRange(ctx, key, 0, -1).Val()
log.Printf("Final elements: %v", elements)

If we run that a few times, we can see what's wrong. To be honest, I thought it would take a decent amount of runs, but it looks like the race condition gods like me:

» go run main.go
Initializing 256ceca4-180f-4746-a568-2b0d384038cf
Final elements: [1 2 3 4 5]
» go run main.go
Initializing 34dacaa7-127d-45ca-8b47-78fd74ca01ee
Final elements: [1 2 3 4 5 4 5]

[1 2 3 4 5 4 5]? That's not correct. How did we end up with that? Let's break the operations down:

Making it atomic

The usual way of solving the problem of accessing/modifying a shared resource is using locks or transactions. In Redis, if we were to follow that path, there are a few things to look at:

Using SETNX: this is the historical way of doing it, but this pattern is well-explained here and here though, so I won't go into it here.
Using Redlock: to be quite honest, I started reading about Redlock but ended up leaving it, since I found another simpler approach. But here is their docs on it, Martin Kleppmann's review of the algorithm and also the author's response to the review. It's quite extensive, but hey, you might be looking for something to read.

These solutions are mostly fine, but I wanted something simpler and didn't really want to rely on locks to solve this problem. And here is where the hero of this story emerges.

Lua scripts

Redis allows you to execute lua scripts and guarantees atomicity when executing them. Putting in other words, a lua script is a way of wrapping multiple instructions into an atomic operation in Redis, without having to deal with locks or transactions.

Let's create a little lua script to do what we want and use EVAL to execute it:

var APPEND_SCRIPT = `
local length = redis.call('LLEN', KEYS[1])
local upto = tonumber(ARGV[1])
if length > upto then
  redis.call('LTRIM', KEYS[1], 0, (upto - 1))
end
local elements = {unpack(ARGV)}
table.remove(elements, 1)
return redis.call('RPUSH', KEYS[1], unpack(elements))
`

func AppendWithScript(client *redis.Client, key string, elements []string, index int64) error {
  keys := []string{key}
  arguments := buildEvalArguments(elements, index)
  _, err := client.Eval(context.TODO(), APPEND_SCRIPT, keys, arguments).Result()
  if err != nil {
    return err
  }

  return nil
}

func buildEvalArguments(logs []string, index int64) []interface{} {
  arguments := make([]interface{}, len(logs)+1)
  arguments[0] = index
  for i, v := range logs {
    arguments[i+1] = v
  }
  return arguments
}

What about performance?

Ok, this seems great and all, but what about my performance? Well, let's write some benchmarks for our functions:

func Benchmark__Append(b *testing.B) {
  for i := 0; i < b.N; i++ {
    Append(client, key, ELEMENTS_TO_APPEND, 3)
  }
}

func Benchmark__AppendWithScript(b *testing.B) {
  for i := 0; i < b.N; i++ {
    AppendWithScript(client, key, ELEMENTS_TO_APPEND, 3)
  }
}

And this is the result we get if we run them against a local redis instance running inside a docker container:

» go test -bench=.
goos: darwin
goarch: amd64
pkg: redis-playground
cpu: Intel(R) Core(TM) i7-6700HQ CPU @ 2.60GHz
Benchmark__Append-8                      282       4323202 ns/op
Benchmark__AppendWithScript-8            712       1556933 ns/op
PASS
ok      redis-playground    3.124s

Append() has an average speed of ~4.32ms while AppendWithScript() executes at ~1.55ms, ~278% times faster.

Pre-loading lua scripts

Even though it is fast enough for our case, there's one thing that might hurt your performance when using lua scripts. Note that are sending the whole script every time to Redis and Redis is recompiling it every time. But the script is always the same.

Considering that, another thing you can do is preload the script before executing it in Redis using SCRIPT LOAD and EVAL SHA:

func LoadScript(client *redis.Client) (string, error) {
    scriptSha, err := client.ScriptLoad(context.TODO(), APPEND_SCRIPT).Result()
    if err != nil {
        return "", err
    }

    return scriptSha, nil
}

func AppendWithLoadedScript(client *redis.Client, scriptSha string, key string, elements []string, index int64) error {
    keys := []string{key}
    arguments := buildEvalArguments(elements, index)
    _, err := client.EvalSha(context.TODO(), scriptSha, keys, arguments).Result()
    if err != nil {
        return err
    }

    return nil
}

Running the tests again:

» go test -bench=.
goos: darwin
goarch: amd64
pkg: redis-playground
cpu: Intel(R) Core(TM) i7-6700HQ CPU @ 2.60GHz
Benchmark__Append-8                              268       4288026 ns/op
Benchmark__AppendWithScript-8                    710       1555603 ns/op
Benchmark__AppendWithLoadedScript-8              746       1532615 ns/op
PASS
ok      redis-playground    4.381s

Ok, so not a whole lot faster.

It's important to note here though that I am running these tests against a local redis instance running inside a docker container in my own machine, so there's no network latency in place. Also, the script we're using is not that big and doesn't consume a lot of bandwidth, but if you have a large script, pre-loading it might really help.

And that's what I had for today. Thanks for reading!

Building an RTP proxy using Spring

Lucas Pinheiro — Fri, 09 Jul 2021 17:39:18 +0000

Originally published here.

I have been really into understanding more about how audio/video streamings work. One piece of that world is RTP. It is defined in RFC 3550, which I read but I don’t actually understand things until I see them working for myself, so let’s do that. This is what we are building:

ffmpeg generates an RTP stream from an audio file
The RTP receiver consumes that stream, creating a websocket stream from it and sending it to the websocket server
The websocket server consumes the stream and writes it to a file

The code for this experiment can be found at https://github.com/lucaspin/spring-replication-proxy.

The RTP header structure

Here’s the RTP header structure, from the RFC:

+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
|V=2|P|X|  CC   |M|     PT      |       sequence number         |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
|                           timestamp                           |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
|           synchronization source (SSRC) identifier            |
+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+=+
|            contributing source (CSRC) identifiers             |
|                             ....                              |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+

A few things to note there:

version (V): the first two bytes in the packet. This is always 2, at least until someone comes up with a newer version of RTP. Food for thought: as just 2 bits are used for this, would it only be possible for a version 3 to exist?
contributing sources count (CC): works together with the CSRC identifiers part of the header, defining how many of those we’ll have in the packet. As only 4 bits are used for this, the maximum number of CSRC identifiers we can have is 15.
payload type (PT): the format that the media in this packet uses. This only makes sense in the context of a profile. RFC 3551 defines a good example of one. The profile defines which values map to which formats. The payload type can be static or dynamic. For a static payload type, you just need to know which profile the sender is using and you’re good. For a dynamic payload type, you need the profile and something else to know the format. Usually, an SDP description is that “something else”.
sequence number: the sender puts sequence numbers on all packets before sending them, so that the receiver can recompose the media packets in the correct order. Later on, you’ll be able to hear why that is important.
timestamp: Why would you need a sequence number and a timestamp?
SSRC identifier: synchronization source; a unique id used to distinguish media streams coming from different sources.
CSRC identifiers: contributing sources; a way of RTP translators to keep the original sources that were mixed into the same media stream.

The P (padding), X (extension) and M (marker) fields are flags to indicate changes in how the packet is structured, and are not really important for us right now. So, the RTP header can be defined as this:

public class RTPPacket {
    private final int version;
    private final boolean padding;
    private final boolean extension;
    private final int contributingSourcesCount;
    private final boolean marker;
    private final int payloadType;
    private final int sequenceNumber;
    private final int timestamp;
    private final int synchronizationSourceId;
    private final byte[] payload;
}

Parsing the header

If you’re like me and doesn’t speak binary fluently, you’ll need some little reminders on how to manipulate bits to grab the information you want.

For instance, we want to grab the version of the packet. We know that field takes the first two bits of the first byte of the packet. But the first byte also has three other fields. How do we get just the version then? Using the AND and SHIFT bitwise operations.

Let’s say that the first byte is 10010010. In order to grab just the first two leftmost bits, we need to “erase” the other six bits. In order to do that, we can use an AND operation where the bits in the same position as the ones we want to erase are 0:

         10010010
     AND 11000000
     ------------
         10000000

Remember: ANDing anything with 0 gives you 0. Now we zeroed them out, but the bits we want are still in the leftmost position. Now, we shift them right 6 times:

         10000000
         --------
    (1x) 01000000
    (2x) 00100000
    (3x) 00010000
    (4x) 00001000
    (5x) 00000100
    (6x) 00000010

Now, we have 00000010, or 2 in the fantastic decimal system. With that idea in mind, let’s create a parsePacket() method to transform an array of bytes into an RTPPacket object:

public static RTPPacket parsePacket(byte[] packet) {
    return RTPPacket.builder()
        .version((packet[0] & 0b11000000) >>> 6)
        .padding(((packet[0] & 0b00100000) >> 5) == 1)
        .extension(((packet[0] & 0b00010000) >> 4) == 1)
        .contributingSourcesCount(packet[0] & 0b00001111)
        .marker(((packet[1] & 0b10000000) >> 7) == 1)
        .payloadType(packet[1] & 0b01111111)
        .sequenceNumber(ByteBuffer.wrap(packet, 2, 2).getShort())
        .timestamp(ByteBuffer.wrap(packet, 4, 4).getInt())
        .synchronizationSourceId(ByteBuffer.wrap(packet, 8, 4).getInt())
        .payload(Arrays.copyOfRange(packet, 12, packet.length))
        .build();
}

The UDP inbound receiver

As we’re only interested in understanding RTP, let’s let Spring take care of the UDP complexity for us. Spring has support for TCP and UDP and here is how you create a UDP inbound receiver on port 11111 with it:

IntegrationFlows.from(new UnicastReceivingChannelAdapter(11111))
    .handle(new RTPMessageHandler(rtpManager))
    .get();

The RTPMessageHandler class extends Spring’s AbstractMessageHandler and does just two things:

Parses the RTP packet
Forwards it to our RTPManager

public class RTPMessageHandler extends AbstractMessageHandler {
    private final RTPManager rtpManager;

    @Override
    protected void handleMessageInternal(Message<?> message) {
        RTPPacket packet = parsePacket((byte[]) message.getPayload());
        rtpManager.onPacketReceived(packet);
    }
}

The RTPManager is the one responsible for a very fundamental part of an RTP receiver using UDP: packet reordering.

Packet reordering

RTP may be used over TCP (less common) or UDP (more common). As it doesn’t rely on the delivery order correctness that might be provided by the transport protocol, it needs a way to sequence packets. And that’s why the RTP header has a sequence number on it. The RTP receiver that will play the media stream needs to reorder the packets received, otherwise the media will not sound/look very nice. In our example, we won’t be playing it, but the consumer of the WebSocket will, so we need to reorder the packets for them.

The RTPManager will be responsible for two things:

Reordering packets before sending them through the websocket
Grouping the packets that come in by SSRC id

public class RTPManager {
    private final Map<Integer, SyncSourceStatus> syncSources = new HashMap<>();

    public synchronized void onPacketReceived(RTPPacket packet) {
        if (syncSources.containsKey(packet.getSynchronizationSourceId())) {
            SyncSourceStatus status = syncSources.get(packet.getSynchronizationSourceId());
            synchronized (status.getLock()) {
                status.addPacket(packet);
                if (status.getPackets().size() > MAX_PACKETS_BEFORE_FLUSHING) {
                    status.flush();
                }
            }
        } else {
            syncSources.put(packet.getSynchronizationSourceId(), SyncSourceStatus.builder()
                    .syncSourceId(packet.getSynchronizationSourceId())
                    .packets(new ArrayList<>(List.of(packet)))
                    .webSocket(initializeSocket())
                    .lock(new Object())
                    .build());
        }
    }
}

Here, I use another class called SyncSourceStatus, to help me with those things:

static class SyncSourceStatus {
    private int syncSourceId;
    private List<RTPPacket> packets;
    private Socket webSocket;
    private final Object lock;

    public void addPacket(RTPPacket packet) {
        packets.add(packet);
    }

    public void flush() {
        packets = packets.stream().sorted().collect(Collectors.toList());
        packets.forEach(packet -> webSocket.send(packet.getPayload()));
        packets = new ArrayList<>();
    }
}

Note that the SyncSourceStatus.flush() sorts the packets before sending them through the websocket, so we need to make our RTPPacket implement Comparable:

public class RTPPacket implements Comparable<RTPPacket> {
    // Fields and getters

    @Override
    public int compareTo(RTPPacket o) {
        return Integer.compare(getSequenceNumber(), o.getSequenceNumber());
    }
}

Bringing everything up

For our WebSocket server and client, we’ll use socket.io. Our server just grabs everything that comes from the websocket stream and saves it into a file:

const http = require('http').Server();
const fs = require('fs');
const io = require('socket.io')(http);

io.on('connection', (socket) => {
  const wstream = fs.createWriteStream('/tmp/audio-from-ffmpeg');

  socket.on('disconnect', () => {
    wstream.end();
  });

  socket.on('message', msg => {
    wstream.write(Buffer.from(msg));
  });
});

http.listen(4010, () => {
  console.log('listening on *:4010');
});

And we’ll use ffmpeg to generate the RTP stream that’ll be consumed by our RTP receiver:

ffmpeg \
    -re \
    -i media/pcm_s16le-44100hz-s16-10s.wav \
    -c:a copy \
    -f rtp \
    "rtp://127.0.0.1:11111"

Once the ffmpeg command exits, we can check the audio written to /tmp/audio-from-ffmpeg by the server and make sure it plays well.

Reorder or not to reorder

To illustrate why packet reordering is important:

Media stream replicated with packet reordering: https://lucaspin.github.io/public/media/reordering-yes.ogg
Media stream replicated without packet reordering: https://lucaspin.github.io/public/media/reordering-no.ogg

Big difference, huh? That’s why you need the sequence number field, because the RTP receiver needs a way to put the packets that arrive in the same order in which they were sent.

That’s all for today, thanks for reading!