1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
|
package io.crashbox.ci
import java.io.{File, OutputStream}
import scala.collection.JavaConverters._
import scala.concurrent.Future
import scala.concurrent.duration._
import com.spotify.docker.client.DefaultDockerClient
import com.spotify.docker.client.DockerClient.{
AttachParameter,
ListContainersParam
}
import com.spotify.docker.client.LogStream
import com.spotify.docker.client.exceptions.ContainerNotFoundException
import com.spotify.docker.client.messages.{ContainerConfig, HostConfig}
import com.spotify.docker.client.messages.HostConfig.Bind
trait Executors { core: Core =>
val dockerClient =
DefaultDockerClient.builder().uri("unix:///run/docker.sock").build()
core.system.registerOnTermination {
dockerClient.close()
}
def containerUser = "crashbox"
def containerWorkDirectory = "/home/crashbox"
def containerKillTimeout = 10.seconds
case class ExecutionId(containerId: String) {
override def toString = containerId
}
def startExecution(
image: String,
script: String,
dir: File,
out: OutputStream
): Future[ExecutionId] =
Future {
val volume = Bind
.builder()
.from(dir.getAbsolutePath)
.to(containerWorkDirectory)
.build()
val hostConfig = HostConfig.builder().binds(volume).build()
val containerConfig = ContainerConfig
.builder()
.labels(Map("crashbox" -> "build").asJava)
.hostConfig(hostConfig)
.tty(true) // combine stdout and stderr into stdout
.image(image)
.user(containerUser)
.workingDir(containerWorkDirectory)
.entrypoint("/bin/sh", "-c")
.cmd(script)
.build()
val container = dockerClient.createContainer(containerConfig).id
log.debug(s"Starting container $container")
dockerClient.startContainer(container)
log.debug(s"Attaching log stream of container $container")
blockingDispatcher execute new Runnable {
override def run() = {
var stream: LogStream = null
try {
stream = dockerClient.attachContainer(
container,
AttachParameter.LOGS,
AttachParameter.STDOUT,
AttachParameter.STREAM
)
stream.attach(out, null, true)
} finally {
if (stream != null) stream.close()
}
}
}
ExecutionId(container)
}(blockingDispatcher)
def waitExecution(id: ExecutionId): Future[Int] =
Future {
log.debug(s"Waiting for container $id to exit")
val res: Int = dockerClient.waitContainer(id.containerId).statusCode()
cancelExecution(id)
res
}(blockingDispatcher)
def cancelExecution(id: ExecutionId): Unit = {
try {
log.debug(s"Stopping container $id")
dockerClient.stopContainer(id.containerId,
containerKillTimeout.toUnit(SECONDS).toInt)
log.debug(s"Removing container $id")
dockerClient.removeContainer(id.containerId)
} catch {
case _: ContainerNotFoundException => // build already cancelled
}
}
def reapDeadBuilds(): Unit = {
val stale = dockerClient
.listContainers(
ListContainersParam.withLabel("crashbox"),
ListContainersParam.withStatusExited()
)
.asScala
stale.foreach { container =>
log.warning(s"Removing stale container ${container.id}")
dockerClient.removeContainer(container.id)
}
}
}
|
