Skip to content

Commit 986ee2e

Browse files
BillOTeiBillOTei
committed
release 4.1.0
1 parent fad1616 commit 986ee2e

40 files changed

Lines changed: 792 additions & 287 deletions

‎.gitignore‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -21,3 +21,5 @@ conf/application.conf
2121

2222
sbt-launch.jar
2323
.vscode
24+
.claude
25+
.site

‎CHANGELOG.md‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,15 @@
11
# Change Log
22

3+
## Unreleased
4+
5+
- [DL-5871] fix: run analyzer/responder jobs on dedicated thread pools to keep the HTTP API responsive under heavy job load
6+
7+
**Upgrade note:** the `analyzer` and `responder` thread pools now use a `thread-pool-executor`
8+
(`fixed-pool-size`) instead of a `fork-join-executor`. Any custom
9+
`analyzer.fork-join-executor` / `responder.fork-join-executor` tuning in `application.conf` is
10+
no longer applied — switch to `analyzer.thread-pool-executor.fixed-pool-size`
11+
(see `conf/application.sample`).
12+
313
## 3.2.0 (2025-06-02)
414

515
- [DL-1231] Add support of Kubernetes

‎app/org/thp/cortex/controllers/AnalyzerConfigCtrl.scala‎

Lines changed: 11 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -1,28 +1,29 @@
11
package org.thp.cortex.controllers
22

3-
import javax.inject.{Inject, Singleton}
4-
import scala.concurrent.{ExecutionContext, Future}
5-
3+
import org.elastic4play.BadRequestError
4+
import org.elastic4play.controllers.{Authenticated, Fields, FieldsBodyParser, Renderer}
5+
import org.thp.cortex.models.{BaseConfig, Roles}
6+
import org.thp.cortex.services.AnalyzerConfigSrv
7+
import play.api.Logger
68
import play.api.libs.json.JsObject
79
import play.api.mvc.{AbstractController, Action, AnyContent, ControllerComponents}
810

9-
import org.thp.cortex.models.{BaseConfig, Roles}
10-
import org.thp.cortex.services.{AnalyzerConfigSrv, UserSrv}
11-
12-
import org.elastic4play.BadRequestError
13-
import org.elastic4play.controllers.{Authenticated, Fields, FieldsBodyParser, Renderer}
11+
import javax.inject.{Inject, Singleton}
12+
import scala.concurrent.{ExecutionContext, Future}
13+
import scala.util.chaining.scalaUtilChainingOps
1414

1515
@Singleton
1616
class AnalyzerConfigCtrl @Inject() (
1717
analyzerConfigSrv: AnalyzerConfigSrv,
18-
userSrv: UserSrv,
1918
authenticated: Authenticated,
2019
fieldsBodyParser: FieldsBodyParser,
2120
renderer: Renderer,
2221
components: ControllerComponents,
2322
implicit val ec: ExecutionContext
2423
) extends AbstractController(components) {
2524

25+
private lazy val logger: Logger = Logger(getClass.getName)
26+
2627
def get(analyzerConfigName: String): Action[AnyContent] = authenticated(Roles.orgAdmin).async { request =>
2728
analyzerConfigSrv
2829
.getForUser(request.userId, analyzerConfigName)
@@ -50,6 +51,7 @@ class AnalyzerConfigCtrl @Inject() (
5051
analyzerConfigSrv
5152
.updateOrCreate(request.userId, analyzerConfigName, config)
5253
.map(renderer.toOutput(OK, _))
54+
.tap(_ => logger.info(s"Analyzer $analyzerConfigName updated with $config by user id ${request.userId}"))
5355
case None => Future.failed(BadRequestError("attribute config has invalid format"))
5456
}
5557
}

‎app/org/thp/cortex/controllers/AssetCtrl.scala‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -13,7 +13,9 @@ trait AssetCtrl {
1313
}
1414

1515
@Singleton
16-
class AssetCtrlProd @Inject() (errorHandler: HttpErrorHandler, meta: AssetsMetadata, env: Environment) extends Assets(errorHandler, meta, env) with AssetCtrl {
16+
class AssetCtrlProd @Inject() (errorHandler: HttpErrorHandler, meta: AssetsMetadata, env: Environment)
17+
extends Assets(errorHandler, meta, env)
18+
with AssetCtrl {
1719
def get(file: String): Action[AnyContent] = at("/www", file)
1820
}
1921

‎app/org/thp/cortex/controllers/OrganizationCtrl.scala‎

Lines changed: 23 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -1,26 +1,22 @@
11
package org.thp.cortex.controllers
22

3-
import javax.inject.{Inject, Singleton}
4-
5-
import scala.concurrent.{ExecutionContext, Future}
6-
7-
import play.api.Logger
8-
import play.api.http.Status
9-
import play.api.mvc._
10-
11-
import org.thp.cortex.models.Roles
12-
import org.thp.cortex.services.{OrganizationSrv, UserSrv}
13-
14-
import org.elastic4play.{BadRequestError, NotFoundError}
153
import org.elastic4play.controllers.{Authenticated, Fields, FieldsBodyParser, Renderer}
164
import org.elastic4play.models.JsonFormat.baseModelEntityWrites
175
import org.elastic4play.services.JsonFormat.{aggReads, queryReads}
186
import org.elastic4play.services.{UserSrv => _, _}
7+
import org.elastic4play.{BadRequestError, NotFoundError}
8+
import org.thp.cortex.models.Roles
9+
import org.thp.cortex.services.{OrganizationSrv, UserSrv}
10+
import play.api.Logger
11+
import play.api.http.Status
12+
import play.api.mvc._
13+
14+
import javax.inject.{Inject, Singleton}
15+
import scala.concurrent.{ExecutionContext, Future}
1916

2017
@Singleton
2118
class OrganizationCtrl @Inject() (
2219
organizationSrv: OrganizationSrv,
23-
authSrv: AuthSrv,
2420
auxSrv: AuxSrv,
2521
userSrv: UserSrv,
2622
authenticated: Authenticated,
@@ -36,7 +32,10 @@ class OrganizationCtrl @Inject() (
3632
def create: Action[Fields] = authenticated(Roles.superAdmin).async(fieldsBodyParser) { implicit request =>
3733
organizationSrv
3834
.create(request.body)
39-
.map(organization => renderer.toOutput(CREATED, organization))
35+
.map { organization =>
36+
logger.info(s"Organization ${organization.id} created by user ${request.userId}")
37+
renderer.toOutput(CREATED, organization)
38+
}
4039
}
4140

4241
def get(organizationId: String): Action[Fields] = authenticated(Roles.superAdmin, Roles.orgAdmin).async(fieldsBodyParser) { implicit request =>
@@ -55,9 +54,12 @@ class OrganizationCtrl @Inject() (
5554
if (organizationId == "cortex")
5655
Future.failed(BadRequestError("Cortex organization can't be updated"))
5756
else
58-
organizationSrv.update(organizationId, request.body).map { organization =>
59-
renderer.toOutput(OK, organization)
60-
}
57+
organizationSrv
58+
.update(organizationId, request.body)
59+
.map { organization =>
60+
logger.info(s"Organization ${organization.id} updated by user ${request.userId}")
61+
renderer.toOutput(OK, organization)
62+
}
6163
}
6264

6365
def delete(organizationId: String): Action[AnyContent] = authenticated(Roles.superAdmin).async { implicit request =>
@@ -66,7 +68,10 @@ class OrganizationCtrl @Inject() (
6668
else
6769
organizationSrv
6870
.delete(organizationId)
69-
.map(_ => NoContent)
71+
.map { organization =>
72+
logger.info(s"Organization ${organization.id} deleted by user ${request.userId}")
73+
NoContent
74+
}
7075
}
7176

7277
def find: Action[Fields] = authenticated(Roles.superAdmin).async(fieldsBodyParser) { implicit request =>

‎app/org/thp/cortex/controllers/ResponderConfigCtrl.scala‎

Lines changed: 14 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -1,31 +1,32 @@
11
package org.thp.cortex.controllers
22

3-
import scala.concurrent.{ExecutionContext, Future}
4-
3+
import org.elastic4play.BadRequestError
4+
import org.elastic4play.controllers.{Authenticated, Fields, FieldsBodyParser, Renderer}
5+
import org.thp.cortex.models.{BaseConfig, Roles}
6+
import org.thp.cortex.services.ResponderConfigSrv
7+
import play.api.Logger
58
import play.api.libs.json.JsObject
69
import play.api.mvc.{AbstractController, Action, AnyContent, ControllerComponents}
710

811
import javax.inject.{Inject, Singleton}
9-
import org.thp.cortex.models.{BaseConfig, Roles}
10-
import org.thp.cortex.services.{ResponderConfigSrv, UserSrv}
11-
12-
import org.elastic4play.BadRequestError
13-
import org.elastic4play.controllers.{Authenticated, Fields, FieldsBodyParser, Renderer}
12+
import scala.concurrent.{ExecutionContext, Future}
13+
import scala.util.chaining.scalaUtilChainingOps
1414

1515
@Singleton
1616
class ResponderConfigCtrl @Inject() (
1717
responderConfigSrv: ResponderConfigSrv,
18-
userSrv: UserSrv,
1918
authenticated: Authenticated,
2019
fieldsBodyParser: FieldsBodyParser,
2120
renderer: Renderer,
2221
components: ControllerComponents,
2322
implicit val ec: ExecutionContext
2423
) extends AbstractController(components) {
2524

26-
def get(analyzerConfigName: String): Action[AnyContent] = authenticated(Roles.orgAdmin).async { request =>
25+
private lazy val logger: Logger = Logger(getClass.getName)
26+
27+
def get(responderConfigName: String): Action[AnyContent] = authenticated(Roles.orgAdmin).async { request =>
2728
responderConfigSrv
28-
.getForUser(request.userId, analyzerConfigName)
29+
.getForUser(request.userId, responderConfigName)
2930
.map(renderer.toOutput(OK, _))
3031
}
3132

@@ -44,12 +45,13 @@ class ResponderConfigCtrl @Inject() (
4445
}
4546
}
4647

47-
def update(analyzerConfigName: String): Action[Fields] = authenticated(Roles.orgAdmin).async(fieldsBodyParser) { implicit request =>
48+
def update(responderConfigName: String): Action[Fields] = authenticated(Roles.orgAdmin).async(fieldsBodyParser) { implicit request =>
4849
request.body.getValue("config").flatMap(_.asOpt[JsObject]) match {
4950
case Some(config) =>
5051
responderConfigSrv
51-
.updateOrCreate(request.userId, analyzerConfigName, config)
52+
.updateOrCreate(request.userId, responderConfigName, config)
5253
.map(renderer.toOutput(OK, _))
54+
.tap(_ => logger.info(s"Responder $responderConfigName updated with $config by user id ${request.userId}"))
5355
case None => Future.failed(BadRequestError("attribute config has invalid format"))
5456
}
5557
}

‎app/org/thp/cortex/controllers/StreamCtrl.scala‎

Lines changed: 56 additions & 34 deletions
Original file line numberDiff line numberDiff line change
@@ -1,27 +1,24 @@
11
package org.thp.cortex.controllers
22

3-
import javax.inject.{Inject, Singleton}
4-
5-
import scala.collection.immutable
6-
import scala.concurrent.{ExecutionContext, Future}
7-
import scala.concurrent.duration.{DurationLong, FiniteDuration}
8-
import scala.util.Random
9-
10-
import play.api.{Configuration, Logger}
11-
import play.api.http.Status
12-
import play.api.libs.json.Json
13-
import play.api.mvc.{AbstractController, Action, AnyContent, ControllerComponents}
14-
153
import org.apache.pekko.actor.{ActorSystem, Props}
16-
import org.apache.pekko.util.Timeout
174
import org.apache.pekko.pattern.ask
5+
import org.apache.pekko.util.Timeout
6+
import org.elastic4play.Timed
7+
import org.elastic4play.controllers._
8+
import org.elastic4play.services.{AuxSrv, EventSrv, MigrationSrv}
189
import org.thp.cortex.models.Roles
19-
import org.thp.cortex.services.StreamActor
2010
import org.thp.cortex.services.StreamActor.StreamMessages
11+
import org.thp.cortex.services.{StreamActor, UserSrv}
12+
import play.api.http.Status
13+
import play.api.libs.json.Json
14+
import play.api.mvc.{AbstractController, Action, AnyContent, ControllerComponents}
15+
import play.api.{Configuration, Logger}
2116

22-
import org.elastic4play.Timed
23-
import org.elastic4play.controllers._
24-
import org.elastic4play.services.{AuxSrv, EventSrv, MigrationSrv, UserSrv}
17+
import java.security.SecureRandom
18+
import javax.inject.{Inject, Singleton}
19+
import scala.collection.immutable
20+
import scala.concurrent.duration.{DurationLong, FiniteDuration}
21+
import scala.concurrent.{ExecutionContext, Future}
2522

2623
@Singleton
2724
class StreamCtrl(
@@ -70,17 +67,32 @@ class StreamCtrl(
7067
)
7168
private[StreamCtrl] lazy val logger = Logger(getClass)
7269

70+
// The bootstrap user has no User document in ES yet (initial setup / migration), so there is no
71+
// organization to look up. Such a stream is left unbound (None) instead of being tied to an org.
72+
private val initialUserId = "init"
73+
74+
private def organizationId(userId: String): Future[Option[String]] =
75+
if (userId == initialUserId) Future.successful(None)
76+
else userSrv.getOrganizationId(userId).map(Some(_))
77+
7378
/** Create a new stream entry with the event head
7479
*/
7580
@Timed("controllers.StreamCtrl.create")
76-
def create: Action[AnyContent] = authenticated(Roles.read) {
77-
val id = generateStreamId()
78-
system.actorOf(Props(classOf[StreamActor], cacheExpiration, refresh, nextItemMaxWait, globalMaxWait, eventSrv, auxSrv), s"stream-$id")
79-
Ok(id)
81+
def create: Action[AnyContent] = authenticated(Roles.read).async { request =>
82+
// the stream is bound to the creator's organization
83+
organizationId(request.userId).map { organizationId =>
84+
val id = generateStreamId()
85+
system.actorOf(
86+
Props(classOf[StreamActor], cacheExpiration, refresh, nextItemMaxWait, globalMaxWait, eventSrv, auxSrv, userSrv, organizationId),
87+
s"stream-$id"
88+
)
89+
Ok(id)
90+
}
8091
}
8192

8293
val alphanumeric: immutable.IndexedSeq[Char] = ('a' to 'z') ++ ('A' to 'Z') ++ ('0' to '9')
83-
private[controllers] def generateStreamId() = Seq.fill(10)(alphanumeric(Random.nextInt(alphanumeric.size))).mkString
94+
private val random = new SecureRandom()
95+
private[controllers] def generateStreamId() = Seq.fill(10)(alphanumeric(random.nextInt(alphanumeric.size))).mkString
8496
private[controllers] def isValidStreamId(streamId: String): Boolean =
8597
streamId.length == 10 && streamId.forall(alphanumeric.contains)
8698

@@ -94,18 +106,28 @@ class StreamCtrl(
94106
if (!isValidStreamId(id)) {
95107
Future.successful(BadRequest("Invalid stream id"))
96108
} else {
97-
val futureStatus = authenticated.expirationStatus(request) match {
98-
case ExpirationError if !migrationSrv.isMigrating =>
99-
userSrv.getInitialUser(request).recoverWith { case _ => authenticated.getFromApiKey(request) }.map(_ => OK)
100-
case _: ExpirationWarning => Future.successful(220)
101-
case _ => Future.successful(OK)
102-
}
109+
val futureOrganizationAndStatus: Future[(Option[String], Status)] =
110+
if (migrationSrv.isMigrating)
111+
Future.successful((None, Ok))
112+
else
113+
authenticated.expirationStatus(request) match {
114+
case ExpirationError =>
115+
userSrv
116+
.getInitialUser(request)
117+
.recoverWith { case _ => authenticated.getFromApiKey(request) }
118+
.flatMap(authContext => organizationId(authContext.userId).map(_ -> Ok))
119+
case _: ExpirationWarning =>
120+
authenticated.getContext(request).flatMap(authContext => organizationId(authContext.userId).map(_ -> new Status(220)))
121+
case _ =>
122+
authenticated.getContext(request).flatMap(authContext => organizationId(authContext.userId).map(_ -> Ok))
123+
}
103124

104-
futureStatus.flatMap { status =>
105-
(system.actorSelection(s"/user/stream-$id") ? StreamActor.GetOperations) map {
106-
case StreamMessages(operations) => renderer.toOutput(status, operations)
107-
case m => InternalServerError(s"Unexpected message : $m (${m.getClass})")
108-
}
125+
futureOrganizationAndStatus.flatMap {
126+
case (organizationId, status) =>
127+
(system.actorSelection(s"/user/stream-$id") ? StreamActor.GetOperations(organizationId)) map {
128+
case StreamMessages(operations) => renderer.toOutput(status.header.status, operations)
129+
case m => InternalServerError(s"Unexpected message : $m (${m.getClass})")
130+
}
109131
}
110132
}
111133
}
@@ -114,7 +136,7 @@ class StreamCtrl(
114136
def status: Action[AnyContent] = Action { implicit request =>
115137
val status = authenticated.expirationStatus(request) match {
116138
case ExpirationWarning(duration) => Json.obj("remaining" -> duration.toSeconds, "warning" -> true)
117-
case ExpirationError => Json.obj("remaining" -> 0, "warning" -> true)
139+
case ExpirationError => Json.obj("remaining" -> 0, "warning" -> true)
118140
case ExpirationOk(duration) => Json.obj("remaining" -> duration.toSeconds, "warning" -> false)
119141
}
120142
Ok(status)

0 commit comments

Comments
 (0)