CHAOSSEARCH Utilities
Written in Scala.
- chaossearch.utils.io.UnixDomain.UDSClientTransport - A ClientTransport pattern for AKKA HTTP to allow HTTP transport over a Unix Domain Socket.
./gradlew :jar
This will render the jar
file:
build/libs/chaossearch-utils.jar
At ChaosSearch we use this Akka-Http
ClientTransport
pattern to access the Docker Engine
Unix Domain socket for our swarm manager nodes.
The following snippets of code should give you a flavor of how to use this ClientTransport
pattern to create a connection pool to the Docker Engine
.
Typically the Docker Engine
socket is created at /var/run/docker.sock
. So first we need to create some transport settings for the connection pool.
def transportSettings = ConnectionPoolSettings(system)
.withConnectionSettings(ClientConnectionSettings(system))
.withTransport(DomainSocket.createClientTransport("/var/run/docker.sock"))
.withIdleTimeout(Duration.Inf)
.withMaxConnections(1)
.withMaxOpenRequests(1)
.withPipeliningLimit(1)
In this use case we are assuming an Actor
will be able to ensure ordely and atomic access to the socket. So we can confidently set the MaxConnections
, MaxOpenRequests
and PipeliningLimit
to 1
.
We can now use the transportSettings
to create the connection pool.
val maxQueueDepth = 1024
val queue = Source.queue[(HttpRequest,Promise[HttpResponse])](maxQueueDepth, OverflowStrategy.backpressure)
.via(
Http().newHostConnectionPool[Promise[HttpResponse]](
host = "chaossearch.domain.socket",
port = 80,
settings = transportSettings
)
)
.to(Sink.foreach {
case (res,p) => p.complete(res)
}).run()
Some things to note here, the host
and port
are needed by the API but will not be used by the underlying clientTransport
. In fact if you dig on to the code you will see we set the local and remote addresses to nonsense values. There is no TCP involved here, and therefore ignored.
Now we can create an HttpRequest
.
The Docker Engine
has a test interface in the REST API to verify in the engine is operational.
In this example we are going to use it.
val pingUri = Uri(
scheme = "http",
authority = Uri.Authority(host=Uri.Host("chaossearch.domain.socket"), port=80, userinfo=""),
path = Uri.Path("/v1.35/_ping")
)
val request = HttpRequest(uri = pingUri)
Now we can queue the request
to the connection pool queue
. We create a Promise
to return an HttpResponse
val p = Promise[HttpResponse]
Now we can offer up the request
.
val response = queue.offer((request,p))
.flatMap { _ => p.future }
.flatMap { case response =>
response.entity.toStrict(5.seconds).map { case entity => response.withEntity(entity) }
}
The response is a flattened Future
of an HttpResponse
. You can then map this. For example:
response.map {
case HttpResponse(StatusCodes.OK, _, entity, _) => {
// If we got here then the Docker Engine is functional.
entity.discardBytes()
StatusCodes.OK
}
case HttpResponse(code, _, entity, _) => ... // Handle the error case
}