forked from BlueBrain/nexus
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
log slow blazegraph queries (BlueBrain#3823)
* WIP * add very basic store test * add BlazegraphSlowQueryLoggerTest * add config * refactors * scalafmt * replace view id and project with viewref * log subject as json, NOT NULL * move schema to schema.ddl * rename blazegraph_slow_queries to blazegraph_queries * log failed queries also * change config, add logging * dont log the query * rename logSlowQueries to apply * read record back from store test * dont use try.get, oops * add drop table for blazegraph_queries * make BlazegraphSlowQueryStore impl a private class * remove copied class, oops * split out slow query model class * change distage construction * move BlazegraphQueryContext * move BlazegraphSlowQuery * record whether outcome was successful * fix logging statement * add test for store failure * scalafmt * add deletion capability to store * remove boilerplate from tests * scalafmt * don't use arraylist * rename was_error to failed * make SlowQueriesConfig * make config match changes to scala config * add docs, remove Impl class and just use private class * scalafmt * pr feedback * some pr feedback * assert timestamp is epoch * more pr * scalafmt
- Loading branch information
1 parent
b7feed0
commit 898fa65
Showing
18 changed files
with
576 additions
and
9 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
27 changes: 27 additions & 0 deletions
27
...ain/scala/ch/epfl/bluebrain/nexus/delta/plugins/blazegraph/config/SlowQueriesConfig.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,27 @@ | ||
| package ch.epfl.bluebrain.nexus.delta.plugins.blazegraph.config | ||
|
|
||
| import pureconfig.ConfigReader | ||
| import pureconfig.generic.semiauto.deriveReader | ||
|
|
||
| import scala.concurrent.duration._ | ||
|
|
||
| /** | ||
| * Configuration for the Blazegraph slow queries logging. | ||
| * | ||
| * @param slowQueryThreshold | ||
| * how long a query takes before it is considered slow | ||
| * @param logTtl | ||
| * how long to keep logged slow queries | ||
| * @param deleteExpiredLogsEvery | ||
| * how often to delete expired logs | ||
| */ | ||
| final case class SlowQueriesConfig( | ||
| slowQueryThreshold: Duration, | ||
| logTtl: FiniteDuration, | ||
| deleteExpiredLogsEvery: FiniteDuration | ||
| ) | ||
|
|
||
| object SlowQueriesConfig { | ||
| implicit final val eventLogConfig: ConfigReader[SlowQueriesConfig] = | ||
| deriveReader[SlowQueriesConfig] | ||
| } |
41 changes: 41 additions & 0 deletions
41
...pfl/bluebrain/nexus/delta/plugins/blazegraph/slowqueries/BlazegraphSlowQueryDeleter.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,41 @@ | ||
| package ch.epfl.bluebrain.nexus.delta.plugins.blazegraph.slowqueries | ||
|
|
||
| import ch.epfl.bluebrain.nexus.delta.kernel.utils.IOUtils | ||
| import ch.epfl.bluebrain.nexus.delta.sourcing.stream.{CompiledProjection, ExecutionStrategy, ProjectionMetadata, Supervisor} | ||
| import fs2.Stream | ||
| import monix.bio.Task | ||
|
|
||
| import scala.concurrent.duration.FiniteDuration | ||
|
|
||
| class BlazegraphSlowQueryDeleter(store: BlazegraphSlowQueryStore, deletionThreshold: FiniteDuration) { | ||
| def deleteOldQueries: Task[Unit] = { | ||
| IOUtils.instant.flatMap { now => | ||
| store.removeQueriesOlderThan(now.minusMillis(deletionThreshold.toMillis)) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| object BlazegraphSlowQueryDeleter { | ||
| private val projectionMetadata: ProjectionMetadata = | ||
| ProjectionMetadata("system", "blazegraph-slow-query-log-deletion", None, None) | ||
| def start( | ||
| supervisor: Supervisor, | ||
| store: BlazegraphSlowQueryStore, | ||
| deletionThreshold: FiniteDuration, | ||
| deletionCheckInterval: FiniteDuration | ||
| ): Task[BlazegraphSlowQueryDeleter] = { | ||
| val runner = new BlazegraphSlowQueryDeleter(store, deletionThreshold) | ||
|
|
||
| val continuousStream = Stream | ||
| .fixedRate[Task](deletionCheckInterval) | ||
| .evalMap(_ => runner.deleteOldQueries) | ||
| .drain | ||
|
|
||
| val compiledProjection = | ||
| CompiledProjection.fromStream(projectionMetadata, ExecutionStrategy.TransientSingleNode, _ => continuousStream) | ||
|
|
||
| supervisor | ||
| .run(compiledProjection) | ||
| .map(_ => runner) | ||
| } | ||
| } |
62 changes: 62 additions & 0 deletions
62
...epfl/bluebrain/nexus/delta/plugins/blazegraph/slowqueries/BlazegraphSlowQueryLogger.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,62 @@ | ||
| package ch.epfl.bluebrain.nexus.delta.plugins.blazegraph.slowqueries | ||
|
|
||
| import cats.effect.Clock | ||
| import ch.epfl.bluebrain.nexus.delta.kernel.utils.IOUtils | ||
| import ch.epfl.bluebrain.nexus.delta.plugins.blazegraph.BlazegraphViewsQuery.BlazegraphQueryContext | ||
| import ch.epfl.bluebrain.nexus.delta.plugins.blazegraph.slowqueries.model.BlazegraphSlowQuery | ||
| import com.typesafe.scalalogging.Logger | ||
| import monix.bio.{IO, UIO} | ||
|
|
||
| import scala.concurrent.duration.{Duration, FiniteDuration} | ||
|
|
||
| /** | ||
| * Logs slow queries in order to help us determine problematic queries | ||
| */ | ||
| trait BlazegraphSlowQueryLogger { | ||
|
|
||
| /** | ||
| * When a query is slow, record this with context. | ||
| * | ||
| * @param context | ||
| * information about the query which can be used in logs | ||
| * @param query | ||
| * the query which should be timed, and logged if it is too slow | ||
| * @return | ||
| * the query | ||
| */ | ||
| def apply[E, A](context: BlazegraphQueryContext, query: IO[E, A]): IO[E, A] | ||
| } | ||
|
|
||
| object BlazegraphSlowQueryLogger { | ||
|
|
||
| private val logger = Logger[BlazegraphSlowQueryLogger] | ||
|
|
||
| def apply(sink: BlazegraphSlowQueryStore, longQueryThreshold: Duration)(implicit | ||
| clock: Clock[UIO] | ||
| ): BlazegraphSlowQueryLogger = new BlazegraphSlowQueryLogger { | ||
| def apply[E, A](context: BlazegraphQueryContext, query: IO[E, A]): IO[E, A] = { | ||
| query.attempt.timed | ||
| .flatMap { case (duration, outcome) => | ||
| UIO | ||
| .when(duration >= longQueryThreshold)(logSlowQuery(context, outcome.isLeft, duration)) | ||
| .flatMap(_ => IO.fromEither(outcome)) | ||
| } | ||
| } | ||
|
|
||
| private def logSlowQuery( | ||
| context: BlazegraphQueryContext, | ||
| isError: Boolean, | ||
| duration: FiniteDuration | ||
| ): UIO[Unit] = { | ||
| IOUtils.instant | ||
| .tapEval(_ => | ||
| UIO.delay(logger.warn(s"Slow blazegraph query recorded: duration '$duration', view '${context.view}'")) | ||
| ) | ||
| .flatMap { now => | ||
| sink | ||
| .save(BlazegraphSlowQuery(context.view, context.query, isError, duration, now, context.subject)) | ||
| .onErrorHandleWith(e => UIO.delay(logger.error("error logging blazegraph slow query", e))) | ||
| } | ||
| } | ||
| } | ||
| } |
55 changes: 55 additions & 0 deletions
55
.../epfl/bluebrain/nexus/delta/plugins/blazegraph/slowqueries/BlazegraphSlowQueryStore.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,55 @@ | ||
| package ch.epfl.bluebrain.nexus.delta.plugins.blazegraph.slowqueries | ||
|
|
||
| import ch.epfl.bluebrain.nexus.delta.kernel.database.Transactors | ||
| import ch.epfl.bluebrain.nexus.delta.plugins.blazegraph.slowqueries.model.BlazegraphSlowQuery | ||
| import ch.epfl.bluebrain.nexus.delta.sdk.views.ViewRef | ||
| import doobie.implicits._ | ||
| import doobie.postgres.implicits._ | ||
| import ch.epfl.bluebrain.nexus.delta.sourcing.implicits._ | ||
| import io.circe.syntax.EncoderOps | ||
| import monix.bio.Task | ||
| import ch.epfl.bluebrain.nexus.delta.sourcing.model.Identity.Database._ | ||
|
|
||
| import java.time.Instant | ||
|
|
||
| /** | ||
| * Persistence operations for slow query logs | ||
| */ | ||
| trait BlazegraphSlowQueryStore { | ||
| def save(query: BlazegraphSlowQuery): Task[Unit] | ||
| def removeQueriesOlderThan(instant: Instant): Task[Unit] | ||
| def listForTestingOnly(view: ViewRef): Task[List[BlazegraphSlowQuery]] | ||
| } | ||
|
|
||
| object BlazegraphSlowQueryStore { | ||
| def apply(xas: Transactors): BlazegraphSlowQueryStore = { | ||
| new BlazegraphSlowQueryStore { | ||
| override def save(query: BlazegraphSlowQuery): Task[Unit] = { | ||
| sql""" INSERT INTO blazegraph_queries(project, view_id, instant, duration, subject, query, failed) | ||
| | VALUES(${query.view.project}, ${query.view.viewId}, ${query.instant}, ${query.duration}, ${query.subject.asJson}, ${query.query.value}, ${query.failed}) | ||
| """.stripMargin.update.run | ||
| .transact(xas.write) | ||
| .void | ||
| } | ||
|
|
||
| override def listForTestingOnly(view: ViewRef): Task[List[BlazegraphSlowQuery]] = { | ||
| sql""" SELECT project, view_id, instant, duration, subject, query, failed FROM public.blazegraph_queries | ||
| |WHERE view_id = ${view.viewId} AND project = ${view.project} | ||
| """.stripMargin | ||
| .query[BlazegraphSlowQuery] | ||
| .stream | ||
| .transact(xas.read) | ||
| .compile | ||
| .toList | ||
| } | ||
|
|
||
| override def removeQueriesOlderThan(instant: Instant): Task[Unit] = { | ||
| sql""" DELETE FROM public.blazegraph_queries | ||
| |WHERE instant < $instant | ||
| """.stripMargin.update.run | ||
| .transact(xas.write) | ||
| .void | ||
| } | ||
| } | ||
| } | ||
| } |
Oops, something went wrong.