Learning Rust: HTTP via Unix Socket
Is TCP/IP the only way to carry HTTP traffic? I’m pretty sure you’ve heard about Unix sockets.
Today, I want to touch a bit on the HTTP protocol. It’s probably the most used protocol in the world. We browse the Internet every day and many our apps use it in the background to complete tasks for us. The biggest chunk of the protocol goes via the TCP/IP network stack. But did you know that Unix sockets can also carry it? I would bet your machine uses this path and some of you don’t know it yet.
In this story we are going to build a tiny Docker client capable of listing running containers, stopping and removing one of them. We will even try to spawn new one. It’s not a lot of functionality, but it may bring a lot of fun.
Imagine we can use hyper crate to set up a basic connection to the Docker Engine API:
struct DockerConnection {
sender: SendRequest<Full<Bytes>>,
connection: JoinHandle<Result<(), hyper::Error>>,
}
impl DockerConnection {
async fn open(socket: &str) -> DockerResult<Self> {
let stream: TokioIo<UnixStream> = match UnixStream::connect(Path::new(socket)).await {
Err(error) => return DockerError::raise_unix_socket_connect(socket, error),
Ok(stream) => TokioIo::new(stream),
};
let docker: DockerConnection = match handshake(stream).await {
Err(error) => return DockerError::raise_handshake_failed(socket, error),
Ok((sender, connection)) => Self {
sender: sender,
connection: spawn(async move { connection.await }),
},
};
Ok(docker)
}
}The connection can be opened with a socket /var/run/docker.sock. We perform a basic handshake to return a sender to send a single request and a connection object to be awaited in a background task.
We can extend the connection struct by adding a get function able to return a docker response. The response can only be converted to bytes and later deserialized from a JSON string into a custom struct or predefined error response.
#[derive(Deserialize)]
struct ErrorResponse {
message: String,
}
struct DockerResponse {
url: String,
inner: Response<Incoming>,
connection: JoinHandle<Result<(), hyper::Error>>,
}
impl DockerResponse {
fn new(url: &str, response: Response<Incoming>, connection: JoinHandle<Result<(), hyper::Error>>) -> Self {
Self {
url: url.to_owned(),
inner: response,
connection: connection,
}
}
async fn into_bytes(self) -> DockerResult<Bytes> {
let data: Bytes = match self.inner.collect().await {
Err(error) => return DockerError::raise_response_failed(&self.url, error),
Ok(value) => value.to_bytes(),
};
match self.connection.await {
Err(error) => return DockerError::raise_tokio_failed(&self.url, error),
Ok(Err(error)) => return DockerError::raise_connection_failed(&self.url, error),
_ => (),
}
Ok(data)
}
async fn into_json<T>(self) -> DockerResult<T>
where
T: serde::de::DeserializeOwned,
{
let status: StatusCode = self.inner.status();
let data: Bytes = self.into_bytes().await?;
match from_slice(data.as_ref()) {
Err(error) => DockerError::raise_deserialization_failed(Some(status), error, data),
Ok(value) => Ok(value),
}
}
async fn into_error(self) -> DockerResult<ErrorResponse> {
self.into_json().await
}
}
impl DockerConnection {
...
async fn execute(mut self, url: &str, request: Request<Full<Bytes>>) -> DockerResult<DockerResponse> {
let response: Response<Incoming> = match self.sender.send_request(request).await {
Err(error) => return DockerError::raise_request_failed(url, error),
Ok(value) => value,
};
let status: StatusCode = response.status();
let response: DockerResponse = DockerResponse::new(url, response, self.connection);
if !status.is_success() {
return DockerError::raise_status_failed(status, response);
}
Ok(response)
}
async fn get(self, url: &str) -> DockerResult<DockerResponse> {
let request = Request::builder()
.uri(url)
.method("GET")
.header("Host", "localhost")
.body(Full::new(Bytes::new()));
let request: Request<Full<Bytes>> = match request {
Err(error) => return DockerError::raise_builder_failed(url, error),
Ok(value) => value,
};
self.execute(url, request).await
}
}Do you wonder what can we do next? What about querying all containers? We can build a client based on the already working connection, right?
#[derive(Deserialize)]
struct ContainerListItem {
#[serde(rename = "Id")]
id: String,
#[serde(rename = "Created")]
created: u64,
#[serde(rename = "Image")]
image: String,
#[serde(rename = "Status")]
status: String,
}
enum ContainerList {
Succeeded(Vec<ContainerListItem>),
BadParameter(ErrorResponse),
ServerError(ErrorResponse),
}
struct DockerClient {
socket: String,
}
impl DockerClient {
fn open(socket: &str) -> Self {
Self {
socket: socket.to_owned(),
}
}
async fn containers_list(&self) -> DockerResult<ContainerList> {
let connection: DockerConnection = DockerConnection::open(&self.socket).await?;
match connection.get("/v1.42/containers/json?all=true").await {
Ok(response) => match response.into_json().await {
Ok(value) => Ok(ContainerList::Succeeded(value)),
Err(error) => Err(error),
},
Err(error) => match error {
DockerError::StatusFailed(url, status, response) => match status.as_u16() {
400 => Ok(ContainerList::BadParameter(response.into_error().await?)),
500 => Ok(ContainerList::ServerError(response.into_error().await?)),
_ => Err(DockerError::StatusFailed(url, status, response)),
},
error => Err(error),
},
}
}
}With very few lines we can open a client and list all containers.
#[tokio::main]
async fn main() {
let socket = "/var/run/docker.sock";
let engine = DockerClient::open(socket).await;
match engine.containers_list().await {
Err(error) => println!("{}", error),
Ok(ContainerList::BadParameter(value)) => println!("{:?}", value),
Ok(ContainerList::ServerError(value)) => println!("{:?}", value),
Ok(ContainerList::Succeeded(containers)) => {
for container in containers {
println!("{} | {:>32} | {}", &container.id[0..8], container.status, container.image)
}
}
}
}In my case, it generates something like the following output. It correctly detects one python:3.12 container along with some VSCode dev containers I used for various other projects.
0188b5e9 | Created | python:3.12
68cafdb9 | Up 10 hours | vsc-etl0-3eec192e9332b551e9e038180cfb64c2306031bdbd8b4d06da4a998604742be6
6aeee386 | Exited (0) 3 days ago | vsc-i13os-e7fe779d472afa64b6dc17ad8e968fc355779cbf32d849c619e817e846346699
d18221db | Exited (255) 5 days ago | vsc-learning-rust-ad0453e39e0a890e201da18cf2878f0034b69003bb2d8fee82ed70f48299e1a3
dd1b8b61 | Exited (0) 4 weeks ago | vsc-aws-local-b177b9bf1ec795e8a2a9f15954a3038c4d9cb78456cdf0d14a5fb97477f9a380After such an introduction to “hello world,” we can attempt to create a new container. Per the documentation, a POST request is required. Initially, the post function must be added in the DockerConnection structure
impl DockerConnection {
...
async fn post(self, url: &str, body: Option<Value>) -> DockerResult<DockerResponse> {
let request = Request::builder()
.uri(url)
.method("POST")
.header("Host", "localhost")
.header("Content-Type", "application/json");
let request = match body {
None => request.body(Full::new(Bytes::new())),
Some(value) => request.body(Full::new(Bytes::from(value.to_string()))),
};
let request: Request<Full<Bytes>> = match request {
Err(error) => return DockerError::raise_builder_failed(url, error),
Ok(value) => value,
};
self.execute(url, request).await
}
The code is similar to a GET request but can include a JSON payload. The client’s implementation is straightforward — almost like listing containers. It must prepare the POST request payload correctly.
struct ContainerCreateSpec<'a> {
image: &'a str,
command: Vec<&'a str>,
}
#[derive(Deserialize)]
struct ContainerCreateResponse {
#[serde(rename = "Id")]
id: String,
#[serde(rename = "Warnings")]
warnings: Vec<String>,
}
enum ContainerCreate {
Succeeded(ContainerCreateResponse),
BadParameter(ErrorResponse),
NoSuchImage(ErrorResponse),
Conflict(ErrorResponse),
ServerError(ErrorResponse),
}
impl DockerClient {
...
async fn containers_create(&self, spec: ContainerCreateSpec<'_>) -> DockerResult<ContainerCreate> {
let url: String = format!("/v1.42/containers/create");
let payload: Value = json!({"Image": spec.image, "Cmd": spec.command});
let connection: DockerConnection = DockerConnection::open(&self.socket).await?;
match connection.post(&url, Some(payload)).await {
Ok(response) => match response.into_json().await {
Ok(value) => Ok(ContainerCreate::Succeeded(value)),
Err(error) => Err(error),
},
Err(error) => match error {
DockerError::StatusFailed(url, status, response) => match status.as_u16() {
400 => Ok(ContainerCreate::BadParameter(response.into_error().await?)),
404 => Ok(ContainerCreate::NoSuchImage(response.into_error().await?)),
409 => Ok(ContainerCreate::Conflict(response.into_error().await?)),
500 => Ok(ContainerCreate::ServerError(response.into_error().await?)),
_ => Err(DockerError::StatusFailed(url, status, response)),
},
error => Err(error),
},
}
}
}Now we have a demo app capable of creating a container, but note that the container will not be running.
#[tokio::main]
async fn main() {
let socket = "/var/run/docker.sock";
let engine = DockerClient::open(socket);
let spec = ContainerCreateSpec {
image: "python:3.12",
command: vec!["pip", "install", "pandas"],
};
let container = match engine.containers_create(spec).await {
Err(error) => return println!("{:?}", error),
Ok(ContainerCreate::Succeeded(response)) => response,
Ok(value) => return println!("{:?}", value),
};
println!("Container ID: {}", response.id);
}If we only implemented the DELETE method, we would be able to remove any container. Let’s do it.
impl DockerConnection {
...
async fn delete(self, url: &str) -> DockerResult<DockerResponse> {
let request = Request::builder()
.uri(url)
.method("DELETE")
.header("Host", "localhost")
.body(Full::new(Bytes::new()));
let request: Request<Full<Bytes>> = match request {
Err(error) => return DockerError::raise_builder_failed(url, error),
Ok(value) => value,
};
self.execute(url, request).await
}
}And now, by implementing these functions, we can control the entire lifecycle of any container.
impl DockerClient {
...
async fn containers_start(&self, id: &str) -> DockerResult<ContainerStart> {
let url: String = format!("/v1.42/containers/{id}/start");
let connection: DockerConnection = DockerConnection::open(&self.socket).await?;
match connection.post(&url, None).await {
Ok(response) => match response.into_bytes().await {
Ok(_) => Ok(ContainerStart::Succeeded),
Err(error) => Err(error),
},
Err(error) => match error {
DockerError::StatusFailed(url, status, response) => match status.as_u16() {
304 => Ok(ContainerStart::AlreadyStarted),
404 => Ok(ContainerStart::NoSuchContainer(response.into_error().await?)),
500 => Ok(ContainerStart::ServerError(response.into_error().await?)),
_ => Err(DockerError::StatusFailed(url, status, response)),
},
error => Err(error),
},
}
}
async fn containers_stop(&self, id: &str) -> DockerResult<ContainerStop> {
let url: String = format!("/v1.42/containers/{id}/stop");
let connection: DockerConnection = DockerConnection::open(&self.socket).await?;
match connection.post(&url, None).await {
Ok(response) => match response.into_bytes().await {
Ok(_) => Ok(ContainerStop::Succeeded),
Err(error) => Err(error),
},
Err(error) => match error {
DockerError::StatusFailed(url, status, response) => match status.as_u16() {
304 => Ok(ContainerStop::AlreadyStopped),
404 => Ok(ContainerStop::NoSuchContainer(response.into_error().await?)),
500 => Ok(ContainerStop::ServerError(response.into_error().await?)),
_ => Err(DockerError::StatusFailed(url, status, response)),
},
error => Err(error),
},
}
}
async fn containers_wait(&self, id: &str) -> DockerResult<ContainerWait> {
let url: String = format!("/v1.42/containers/{id}/wait");
let connection: DockerConnection = DockerConnection::open(&self.socket).await?;
match connection.post(&url, None).await {
Ok(response) => match response.into_json().await {
Ok(value) => Ok(ContainerWait::Succeeded(value)),
Err(error) => Err(error),
},
Err(error) => match error {
DockerError::StatusFailed(url, status, response) => match status.as_u16() {
400 => Ok(ContainerWait::BadParameter(response.into_error().await?)),
404 => Ok(ContainerWait::NoSuchContainer(response.into_error().await?)),
500 => Ok(ContainerWait::ServerError(response.into_error().await?)),
_ => Err(DockerError::StatusFailed(url, status, response)),
},
error => Err(error),
},
}
}
async fn containers_remove(&self, id: &str) -> DockerResult<ContainerRemove> {
let url: String = format!("/v1.42/containers/{id}");
let connection: DockerConnection = DockerConnection::open(&self.socket).await?;
match connection.delete(&url).await {
Ok(response) => match response.into_bytes().await {
Ok(_) => Ok(ContainerRemove::Succeeded),
Err(error) => Err(error),
},
Err(error) => match error {
DockerError::StatusFailed(url, status, response) => match status.as_u16() {
400 => Ok(ContainerRemove::BadParameter(response.into_error().await?)),
404 => Ok(ContainerRemove::NoSuchContainer(response.into_error().await?)),
409 => Ok(ContainerRemove::Conflict(response.into_error().await?)),
500 => Ok(ContainerRemove::ServerError(response.into_error().await?)),
_ => Err(DockerError::StatusFailed(url, status, response)),
},
error => Err(error),
},
}
}
}In this short story, we basically built an HTTP client like reqwest using the low-level hyper crate. We didn’t use TCP/IP but Unix Sockets to establish a connection to the locally running Docker Engine API. It allowed us to execute basic operations to play with our containers.
I skipped a lot of code, especially the one to handle errors. If you are interested in a working example, look where I left it for you: https://github.com/amacal/etl0/tree/85d155b1cdf2f7962188cd8b8833442a1e6a1132/src/etl0/src/docker
