Skip to content

Commit e0b67f9

Browse files
BillOTeiBillOTei
authored
Merge pull request #474 from TheHive-Project/fix/DL-717/cortex-docker-client
[DL-717] Fix: cortex docker client
2 parents 423deff + 7103b71 commit e0b67f9

File tree

6 files changed

+261
-98
lines changed

6 files changed

+261
-98
lines changed

‎app/org/thp/cortex/services/DockerJobRunnerSrv.scala

+31-90
Original file line numberDiff line numberDiff line change
@@ -1,24 +1,21 @@
11
package org.thp.cortex.services
22

33
import akka.actor.ActorSystem
4-
import com.spotify.docker.client.DockerClient.LogsParam
5-
import com.spotify.docker.client.messages.HostConfig.Bind
6-
import com.spotify.docker.client.messages.{ContainerConfig, HostConfig}
7-
import com.spotify.docker.client.{DefaultDockerClient, DockerClient}
4+
import org.thp.cortex.util.docker.{DockerClient => DockerJavaClient}
85
import play.api.libs.json.Json
96
import play.api.{Configuration, Logger}
107

118
import java.nio.charset.StandardCharsets
129
import java.nio.file._
10+
import java.util.concurrent.TimeUnit
1311
import javax.inject.{Inject, Singleton}
1412
import scala.concurrent.ExecutionContext
1513
import scala.concurrent.duration.FiniteDuration
1614
import scala.util.Try
1715

1816
@Singleton
1917
class DockerJobRunnerSrv(
20-
client: DockerClient,
21-
config: Configuration,
18+
javaClient: DockerJavaClient,
2219
autoUpdate: Boolean,
2320
jobBaseDirectory: Path,
2421
dockerJobBaseDirectory: Path,
@@ -28,17 +25,7 @@ class DockerJobRunnerSrv(
2825
@Inject()
2926
def this(config: Configuration, system: ActorSystem) =
3027
this(
31-
new DefaultDockerClient.Builder()
32-
.apiVersion(config.getOptional[String]("docker.version").orNull)
33-
.connectionPoolSize(config.getOptional[Int]("docker.connectionPoolSize").getOrElse(100))
34-
.connectTimeoutMillis(config.getOptional[Long]("docker.connectTimeoutMillis").getOrElse(5000))
35-
//.dockerCertificates()
36-
.readTimeoutMillis(config.getOptional[Long]("docker.readTimeoutMillis").getOrElse(30000))
37-
//.registryAuthSupplier()
38-
.uri(config.getOptional[String]("docker.uri").getOrElse("unix:///var/run/docker.sock"))
39-
.useProxy(config.getOptional[Boolean]("docker.useProxy").getOrElse(false))
40-
.build(),
41-
config,
28+
new DockerJavaClient(config),
4229
config.getOptional[Boolean]("docker.autoUpdate").getOrElse(true),
4330
Paths.get(config.get[String]("job.directory")),
4431
Paths.get(config.get[String]("job.dockerDirectory")),
@@ -50,89 +37,43 @@ class DockerJobRunnerSrv(
5037
lazy val isAvailable: Boolean =
5138
Try {
5239
logger.debug(s"Retrieve docker information ...")
53-
logger.info(s"Docker is available:\n${client.info()}")
40+
logger.info(s"Docker is available:\n${javaClient.info}")
5441
true
5542
}.recover {
5643
case error =>
5744
logger.info(s"Docker is not available", error)
5845
false
5946
}.get
6047

61-
def run(jobDirectory: Path, dockerImage: String, timeout: Option[FiniteDuration])(implicit
62-
ec: ExecutionContext
63-
): Try[Unit] = {
64-
import scala.collection.JavaConverters._
65-
if (autoUpdate) Try(client.pull(dockerImage))
66-
// ContainerConfig.builder().addVolume()
67-
val hostConfigBuilder = HostConfig.builder()
68-
config.getOptional[Seq[String]]("docker.container.capAdd").map(_.asJava).foreach(hostConfigBuilder.capAdd)
69-
config.getOptional[Seq[String]]("docker.container.capDrop").map(_.asJava).foreach(hostConfigBuilder.capDrop)
70-
config.getOptional[String]("docker.container.cgroupParent").foreach(hostConfigBuilder.cgroupParent)
71-
config.getOptional[Long]("docker.container.cpuPeriod").foreach(hostConfigBuilder.cpuPeriod(_))
72-
config.getOptional[Long]("docker.container.cpuQuota").foreach(hostConfigBuilder.cpuQuota(_))
73-
config.getOptional[Seq[String]]("docker.container.dns").map(_.asJava).foreach(hostConfigBuilder.dns)
74-
config.getOptional[Seq[String]]("docker.container.dnsSearch").map(_.asJava).foreach(hostConfigBuilder.dnsSearch)
75-
config.getOptional[Seq[String]]("docker.container.extraHosts").map(_.asJava).foreach(hostConfigBuilder.extraHosts)
76-
config.getOptional[Long]("docker.container.kernelMemory").foreach(hostConfigBuilder.kernelMemory(_))
77-
config.getOptional[Long]("docker.container.memoryReservation").foreach(hostConfigBuilder.memoryReservation(_))
78-
config.getOptional[Long]("docker.container.memory").foreach(hostConfigBuilder.memory(_))
79-
config.getOptional[Long]("docker.container.memorySwap").foreach(hostConfigBuilder.memorySwap(_))
80-
config.getOptional[Int]("docker.container.memorySwappiness").foreach(hostConfigBuilder.memorySwappiness(_))
81-
config.getOptional[String]("docker.container.networkMode").foreach(hostConfigBuilder.networkMode)
82-
config.getOptional[Boolean]("docker.container.privileged").foreach(hostConfigBuilder.privileged(_))
83-
hostConfigBuilder.appendBinds(
84-
Bind
85-
.from(dockerJobBaseDirectory.resolve(jobBaseDirectory.relativize(jobDirectory)).toAbsolutePath.toString)
86-
.to("/job")
87-
.readOnly(false)
88-
.build()
89-
)
90-
val cacertsFile = jobDirectory.resolve("input").resolve("cacerts")
91-
val containerConfigBuilder = ContainerConfig
92-
.builder()
93-
.hostConfig(hostConfigBuilder.build())
94-
.image(dockerImage)
95-
.cmd("/job")
48+
private def generateErrorOutput(containerId: String, f: Path) = {
49+
logger.warn(s"the runner didn't generate any output file $f")
50+
for {
51+
output <- javaClient.getLogs(containerId)
52+
report = Json.obj("success" -> false, "errorMessage" -> output)
53+
_ <- Try(Files.write(f, report.toString.getBytes(StandardCharsets.UTF_8)))
54+
} yield report
55+
}
9656

97-
val containerConfig =
98-
if (Files.exists(cacertsFile)) containerConfigBuilder.env(s"REQUESTS_CA_BUNDLE=/job/input/cacerts").build()
99-
else containerConfigBuilder.build()
100-
val containerCreation = client.createContainer(containerConfig)
101-
// Option(containerCreation.warnings()).flatMap(_.asScala).foreach(logger.warn)
57+
def run(jobDirectory: Path, dockerImage: String, timeout: Option[FiniteDuration])(implicit executionContext: ExecutionContext): Try[Unit] = {
58+
val to = timeout.getOrElse(FiniteDuration(5000, TimeUnit.SECONDS))
10259

103-
logger.debug(s"Container configuration: $containerConfig")
104-
logger.info(
105-
s"Execute container ${containerCreation.id()}\n" +
106-
s" timeout: ${timeout.fold("none")(_.toString)}\n" +
107-
s" image : $dockerImage\n" +
108-
s" volume : ${jobDirectory.toAbsolutePath}:/job" +
109-
Option(containerConfig.env()).fold("")(_.asScala.map("\n env : " + _).mkString)
110-
)
111-
112-
val timeoutSched = timeout.map(to =>
113-
system.scheduler.scheduleOnce(to) {
114-
logger.info("Timeout reached, stopping the container")
115-
client.removeContainer(containerCreation.id(), DockerClient.RemoveContainerParam.forceKill())
116-
}
117-
)
118-
val execution = Try {
119-
client.startContainer(containerCreation.id())
120-
client.waitContainer(containerCreation.id())
121-
()
122-
}
123-
timeoutSched.foreach(_.cancel())
124-
val outputFile = jobDirectory.resolve("output").resolve("output.json")
125-
if (!Files.exists(outputFile) || Files.size(outputFile) == 0) {
126-
logger.warn(s"The worker didn't generate output file.")
127-
val output = Try(client.logs(containerCreation.id(), LogsParam.stdout(), LogsParam.stderr()).readFully())
128-
.fold(e => s"Container logs can't be read (${e.getMessage})", identity)
129-
val message = execution.fold(e => s"Docker creation error: ${e.getMessage}\n$output", _ => output)
60+
if (autoUpdate) Try(javaClient.pullImage(dockerImage))
13061

131-
val report = Json.obj("success" -> false, "errorMessage" -> message)
132-
Files.write(outputFile, report.toString.getBytes(StandardCharsets.UTF_8))
133-
}
134-
client.removeContainer(containerCreation.id(), DockerClient.RemoveContainerParam.forceKill())
135-
execution
62+
for {
63+
containerId <- javaClient.prepare(dockerImage, jobDirectory, jobBaseDirectory, dockerJobBaseDirectory, to)
64+
timeoutScheduled = timeout.map(to =>
65+
system.scheduler.scheduleOnce(to) {
66+
logger.info("Timeout reached, stopping the container")
67+
javaClient.clean(containerId)
68+
}
69+
)
70+
_ <- javaClient.execute(containerId)
71+
_ = timeoutScheduled.foreach(_.cancel())
72+
outputFile <- Try(jobDirectory.resolve("output").resolve("output.json"))
73+
isError = Files.notExists(outputFile) || Files.size(outputFile) == 0 || Files.isDirectory(outputFile)
74+
_ = if (isError) generateErrorOutput(containerId, outputFile).toOption else None
75+
_ <- javaClient.clean(containerId)
76+
} yield ()
13677
}
13778

13879
}

‎app/org/thp/cortex/util/docker/DockerClient.scala

+167
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,167 @@
1+
package org.thp.cortex.util.docker
2+
3+
import com.github.dockerjava.api.model._
4+
import com.github.dockerjava.core.{DefaultDockerClientConfig, DockerClientConfig, DockerClientImpl}
5+
import com.github.dockerjava.transport.DockerHttpClient
6+
import com.github.dockerjava.zerodep.ZerodepDockerHttpClient
7+
import play.api.{Configuration, Logger}
8+
9+
import java.nio.file.{Files, Path}
10+
import java.time.Duration
11+
import java.util.concurrent.{Executors, TimeUnit}
12+
import scala.concurrent.blocking
13+
import scala.concurrent.duration.FiniteDuration
14+
import scala.jdk.CollectionConverters._
15+
import scala.util.Try
16+
17+
class DockerClient(config: Configuration) {
18+
private lazy val logger: Logger = Logger(getClass.getName)
19+
private lazy val (dockerConf, httpClient) = getHttpClient
20+
private lazy val underlyingClient = DockerClientImpl.getInstance(dockerConf, httpClient)
21+
22+
def execute(containerId: String): Try[Int] =
23+
Try {
24+
val startContainerCmd = underlyingClient.startContainerCmd(containerId)
25+
startContainerCmd.exec()
26+
val waitResult = underlyingClient
27+
.waitContainerCmd(containerId)
28+
.start()
29+
.awaitStatusCode()
30+
logger.info(s"container $containerId started and awaited with code: $waitResult")
31+
32+
waitResult
33+
}
34+
35+
def prepare(image: String, jobDirectory: Path, jobBaseDirectory: Path, dockerJobBaseDirectory: Path, timeout: FiniteDuration): Try[String] = Try {
36+
logger.info(s"image $image pull result: ${pullImage(image)}")
37+
val containerCmd = underlyingClient
38+
.createContainerCmd(image)
39+
.withHostConfig(configure(jobDirectory, jobBaseDirectory, dockerJobBaseDirectory))
40+
if (Files.exists(jobDirectory.resolve("input").resolve("cacerts")))
41+
containerCmd.withEnv(s"REQUESTS_CA_BUNDLE=/job/input/cacerts")
42+
val containerResponse = containerCmd.exec()
43+
logger.info(
44+
s"about to start container ${containerResponse.getId}\n" +
45+
s" timeout: ${timeout.toString}\n" +
46+
s" image : $image\n" +
47+
s" volumes : ${jobDirectory.toAbsolutePath}"
48+
)
49+
if (containerResponse.getWarnings.nonEmpty) logger.warn(s"${containerResponse.getWarnings.mkString(", ")}")
50+
scheduleContainerTimeout(containerResponse.getId, timeout)
51+
52+
containerResponse.getId
53+
}
54+
55+
private def configure(jobDirectory: Path, jobBaseDirectory: Path, dockerJobBaseDirectory: Path): HostConfig = {
56+
val hostConfigMut = HostConfig
57+
.newHostConfig()
58+
.withBinds(
59+
Seq(
60+
new Bind(
61+
dockerJobBaseDirectory.resolve(jobBaseDirectory.relativize(jobDirectory)).toAbsolutePath.toString,
62+
new Volume(s"/job"),
63+
AccessMode.rw
64+
)
65+
): _*
66+
)
67+
68+
config.getOptional[Seq[String]]("docker.container.capAdd").map(_.map(Capability.valueOf)).foreach(hostConfigMut.withCapAdd(_: _*))
69+
config.getOptional[Seq[String]]("docker.container.capDrop").map(_.map(Capability.valueOf)).foreach(hostConfigMut.withCapDrop(_: _*))
70+
config.getOptional[String]("docker.container.cgroupParent").foreach(hostConfigMut.withCgroupParent)
71+
config.getOptional[Long]("docker.container.cpuPeriod").foreach(hostConfigMut.withCpuPeriod(_))
72+
config.getOptional[Long]("docker.container.cpuQuota").foreach(hostConfigMut.withCpuQuota(_))
73+
config.getOptional[Seq[String]]("docker.container.dns").map(_.asJava).foreach(hostConfigMut.withDns)
74+
config.getOptional[Seq[String]]("docker.container.dnsSearch").map(_.asJava).foreach(hostConfigMut.withDnsSearch)
75+
config.getOptional[Seq[String]]("docker.container.extraHosts").foreach(l => hostConfigMut.withExtraHosts(l: _*))
76+
config.getOptional[Long]("docker.container.kernelMemory").foreach(hostConfigMut.withKernelMemory(_))
77+
config.getOptional[Long]("docker.container.memoryReservation").foreach(hostConfigMut.withMemoryReservation(_))
78+
config.getOptional[Long]("docker.container.memory").foreach(hostConfigMut.withMemory(_))
79+
config.getOptional[Long]("docker.container.memorySwap").foreach(hostConfigMut.withMemorySwap(_))
80+
config.getOptional[Long]("docker.container.memorySwappiness").foreach(hostConfigMut.withMemorySwappiness(_))
81+
config.getOptional[String]("docker.container.networkMode").foreach(hostConfigMut.withNetworkMode)
82+
config.getOptional[Boolean]("docker.container.privileged").foreach(hostConfigMut.withPrivileged(_))
83+
84+
hostConfigMut
85+
}
86+
87+
def info: Info = underlyingClient.infoCmd().exec()
88+
def pullImage(image: String): Boolean = blocking {
89+
val pullImageResultCbk = underlyingClient // Blocking
90+
.pullImageCmd(image)
91+
.start()
92+
.awaitCompletion()
93+
val timeout = config.get[FiniteDuration]("docker.pullImageTimeout")
94+
95+
pullImageResultCbk.awaitCompletion(timeout.toMillis, TimeUnit.MILLISECONDS)
96+
}
97+
98+
def clean(containerId: String): Try[Unit] = Try {
99+
underlyingClient
100+
.killContainerCmd(containerId)
101+
.exec()
102+
underlyingClient
103+
.removeContainerCmd(containerId)
104+
.withForce(true)
105+
.exec()
106+
logger.info(s"removed container $containerId")
107+
}
108+
109+
def getLogs(containerId: String): Try[String] = Try {
110+
val stringBuilder = new StringBuilder()
111+
val callback = new DockerLogsStringBuilder(stringBuilder)
112+
underlyingClient
113+
.logContainerCmd(containerId)
114+
.withStdErr(true)
115+
.withStdOut(true)
116+
.withFollowStream(true)
117+
.withTailAll()
118+
.exec(callback)
119+
.awaitCompletion()
120+
121+
callback.builder.toString
122+
}
123+
124+
private def scheduleContainerTimeout(containerId: String, timeout: FiniteDuration) =
125+
Executors
126+
.newSingleThreadScheduledExecutor()
127+
.schedule(
128+
() => {
129+
logger.info(s"timeout $timeout reached, stopping container $containerId}")
130+
underlyingClient.removeContainerCmd(containerId).withForce(true).exec()
131+
},
132+
timeout.length,
133+
timeout.unit
134+
)
135+
136+
private def getHttpClient: (DockerClientConfig, DockerHttpClient) = {
137+
val dockerConf = getBaseConfig
138+
val dockerClient = new ZerodepDockerHttpClient.Builder()
139+
.dockerHost(dockerConf.getDockerHost)
140+
.sslConfig(dockerConf.getSSLConfig)
141+
.maxConnections(if (config.has("docker.httpClient.maxConnections")) config.get[Int]("docker.httpClient.maxConnections") else 100)
142+
.connectionTimeout(
143+
if (config.has("docker.httpClient.connectionTimeout")) Duration.ofMillis(config.get[Long]("docker.httpClient.connectionTimeout"))
144+
else Duration.ofSeconds(30)
145+
)
146+
.responseTimeout(
147+
if (config.has("docker.httpClient.responseTimeout")) Duration.ofMillis(config.get[Long]("docker.httpClient.responseTimeout"))
148+
else Duration.ofSeconds(45)
149+
)
150+
.build()
151+
152+
(dockerConf, dockerClient)
153+
}
154+
155+
private def getBaseConfig: DockerClientConfig = {
156+
val confBuilder = DefaultDockerClientConfig.createDefaultConfigBuilder()
157+
config.getOptional[String]("docker.host").foreach(confBuilder.withDockerHost)
158+
config.getOptional[Boolean]("docker.tlsVerify").foreach(confBuilder.withDockerTlsVerify(_))
159+
config.getOptional[String]("docker.certPath").foreach(confBuilder.withDockerCertPath)
160+
config.getOptional[String]("docker.registry.user").foreach(confBuilder.withRegistryUsername)
161+
config.getOptional[String]("docker.registry.password").foreach(confBuilder.withRegistryPassword)
162+
config.getOptional[String]("docker.registry.email").foreach(confBuilder.withRegistryEmail)
163+
config.getOptional[String]("docker.registry.url").foreach(confBuilder.withRegistryUrl)
164+
165+
confBuilder.build()
166+
}
167+
}

‎app/org/thp/cortex/util/docker/DockerLogsStringBuilder.scala

+11
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,11 @@
1+
package org.thp.cortex.util.docker
2+
3+
import com.github.dockerjava.api.async.ResultCallback
4+
import com.github.dockerjava.api.model.Frame
5+
6+
class DockerLogsStringBuilder(var builder: StringBuilder) extends ResultCallback.Adapter[Frame] {
7+
override def onNext(item: Frame): Unit = {
8+
builder.append(new String(item.getPayload))
9+
super.onNext(item)
10+
}
11+
}

‎build.sbt

+2
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,8 @@ lazy val cortex = (project in file("."))
2929
Dependencies.reflections,
3030
Dependencies.zip4j,
3131
Dependencies.dockerClient,
32+
Dependencies.dockerJavaClient,
33+
Dependencies.dockerJavaTransport,
3234
Dependencies.akkaCluster,
3335
Dependencies.akkaClusterTyped
3436
),

‎conf/application.sample

+39
Original file line numberDiff line numberDiff line change
@@ -219,4 +219,43 @@ responder {
219219
# port = 3128
220220
# }
221221

222+
# Docker
223+
docker {
224+
host = "tcp://docker.somewhere.tld:2376"
225+
tlsVerify = false
226+
certPath = "/home/user/.docker"
227+
registry {
228+
user = "username"
229+
password = "pwdReg"
230+
email = "user@docker.com"
231+
url = "https://www.docker-registry.com"
232+
}
233+
httpClient {
234+
maxConnections = 100
235+
connectionTimeout = 30000 # millis
236+
responseTimeout = 45000
237+
}
238+
container {
239+
capAdd = ["ALL"]
240+
capDrop = ["NET_ADMIN", "SYS_ADMIN"]
241+
cgroupParent = "m-executor-abcd"
242+
privileged = false
243+
244+
dns = ["8.8.8.8", "9.9.9.9"]
245+
dnsSearch = ["dc1.example.com", "dc2.example.com"]
246+
extraHosts = ["somehost=162.242.195.82", "otherhost=50.31.209.229", "myhostv6=::1"]
247+
networkMode = "host"
248+
249+
cpuPeriod = 100000
250+
cpuQuota = 50000
251+
kernelMemory = 2147483648
252+
memoryReservation = 1024
253+
memory = 4294967296
254+
memorySwap = 1073741824
255+
memorySwappiness = 0
256+
}
257+
autoUpdate = false
258+
pullImageTimeout = 10 minutes
259+
}
260+
222261
# It's the end my friend. Happy hunting!

0 commit comments

Comments
 (0)