-
Notifications
You must be signed in to change notification settings - Fork 74
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add an endpoint to return the history of a resource (#5116)
* Add an endpoint to return the history of a resource * Fix test --------- Co-authored-by: Simon Dumas <[email protected]>
- Loading branch information
Showing
27 changed files
with
777 additions
and
207 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -8,5 +8,10 @@ | |
| } | ||
| } | ||
| } | ||
| ] | ||
| ], | ||
| "properties": { | ||
| "rev": { | ||
| "type": "integer" | ||
| } | ||
| } | ||
| } | ||
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
60 changes: 60 additions & 0 deletions
60
...scala/ch/epfl/bluebrain/nexus/delta/plugins/elasticsearch/metrics/EventMetricsQuery.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,60 @@ | ||
| package ch.epfl.bluebrain.nexus.delta.plugins.elasticsearch.metrics | ||
|
|
||
| import akka.http.scaladsl.model.Uri | ||
| import cats.effect.IO | ||
| import ch.epfl.bluebrain.nexus.delta.plugins.elasticsearch.client.{ElasticSearchClient, QueryBuilder} | ||
| import ch.epfl.bluebrain.nexus.delta.rdf.IriOrBNode.Iri | ||
| import ch.epfl.bluebrain.nexus.delta.sdk.model.search.SearchResults | ||
| import ch.epfl.bluebrain.nexus.delta.sourcing.model.ProjectRef | ||
| import io.circe.JsonObject | ||
| import io.circe.literal.JsonStringContext | ||
| import io.circe.syntax.EncoderOps | ||
|
|
||
| trait EventMetricsQuery { | ||
|
|
||
| def history(project: ProjectRef, id: Iri): IO[SearchResults[JsonObject]] | ||
|
|
||
| } | ||
|
|
||
| object EventMetricsQuery { | ||
|
|
||
| def apply(client: ElasticSearchClient, prefix: String): EventMetricsQuery = new EventMetricsQuery { | ||
|
|
||
| val index = eventMetricsIndex(prefix) | ||
|
|
||
| private def searchQuery(project: ProjectRef, id: Iri) = | ||
| json"""{ | ||
| "query": { | ||
| "bool": { | ||
| "must": [ | ||
| { | ||
| "term": { | ||
| "project": ${project.asJson} | ||
| } | ||
| }, | ||
| { | ||
| "term": { | ||
| "@id": ${id.asJson} | ||
| } | ||
| } | ||
| ] | ||
| } | ||
| }, | ||
| "size": 2000, | ||
| "from": 0, | ||
| "sort": [ | ||
| { "rev": { "order" : "asc" } } | ||
| ] | ||
| } | ||
| """.asObject.toRight(new IllegalStateException("Should not happen, an es query is an object")) | ||
|
|
||
| override def history(project: ProjectRef, id: Iri): IO[SearchResults[JsonObject]] = { | ||
| for { | ||
| jsonQuery <- IO.fromEither(searchQuery(project, id)) | ||
| queryBuilder = QueryBuilder.unsafe(jsonQuery) | ||
| results <- client.search(queryBuilder, Set(index.value), Uri.Query.Empty) | ||
| } yield results | ||
| } | ||
| } | ||
|
|
||
| } |
8 changes: 8 additions & 0 deletions
8
.../src/main/scala/ch/epfl/bluebrain/nexus/delta/plugins/elasticsearch/metrics/package.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,8 @@ | ||
| package ch.epfl.bluebrain.nexus.delta.plugins.elasticsearch | ||
|
|
||
| import ch.epfl.bluebrain.nexus.delta.plugins.elasticsearch.client.IndexLabel | ||
|
|
||
| package object metrics { | ||
|
|
||
| val eventMetricsIndex: String => IndexLabel = prefix => IndexLabel.unsafe(s"${prefix}_project_metrics") | ||
| } |
42 changes: 42 additions & 0 deletions
42
.../epfl/bluebrain/nexus/delta/plugins/elasticsearch/routes/ElasticSearchHistoryRoutes.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,42 @@ | ||
| package ch.epfl.bluebrain.nexus.delta.plugins.elasticsearch.routes | ||
|
|
||
| import akka.http.scaladsl.server.Route | ||
| import cats.syntax.all._ | ||
| import ch.epfl.bluebrain.nexus.delta.plugins.elasticsearch.metrics.EventMetricsQuery | ||
| import ch.epfl.bluebrain.nexus.delta.plugins.elasticsearch.query.ElasticSearchQueryError | ||
| import ch.epfl.bluebrain.nexus.delta.rdf.jsonld.context.RemoteContextResolution | ||
| import ch.epfl.bluebrain.nexus.delta.rdf.utils.JsonKeyOrdering | ||
| import ch.epfl.bluebrain.nexus.delta.sdk.acls.AclCheck | ||
| import ch.epfl.bluebrain.nexus.delta.sdk.directives.AuthDirectives | ||
| import ch.epfl.bluebrain.nexus.delta.sdk.directives.DeltaDirectives.{emit, projectRef} | ||
| import ch.epfl.bluebrain.nexus.delta.sdk.directives.UriDirectives.iriSegment | ||
| import ch.epfl.bluebrain.nexus.delta.sdk.identities.Identities | ||
| import ch.epfl.bluebrain.nexus.delta.sdk.marshalling.RdfMarshalling | ||
| import ch.epfl.bluebrain.nexus.delta.sdk.model.search.SearchResults | ||
| import ch.epfl.bluebrain.nexus.delta.sdk.model.search.SearchResults.searchResultsEncoder | ||
| import ch.epfl.bluebrain.nexus.delta.sdk.permissions.Permissions.resources.{read => Read} | ||
| import io.circe.syntax.EncoderOps | ||
| import io.circe.{Encoder, JsonObject} | ||
|
|
||
| class ElasticSearchHistoryRoutes(identities: Identities, aclCheck: AclCheck, metricsQuery: EventMetricsQuery)(implicit | ||
| cr: RemoteContextResolution, | ||
| ordering: JsonKeyOrdering | ||
| ) extends AuthDirectives(identities, aclCheck) | ||
| with RdfMarshalling { | ||
| implicit private val searchEncoder: Encoder.AsObject[SearchResults[JsonObject]] = searchResultsEncoder(_ => None) | ||
|
|
||
| def routes: Route = | ||
| pathPrefix("history") { | ||
| pathPrefix("resources") { | ||
| extractCaller { implicit caller => | ||
| projectRef.apply { project => | ||
| authorizeFor(project, Read).apply { | ||
| (get & iriSegment & pathEndOrSingleSlash) { id => | ||
| emit(metricsQuery.history(project, id).map(_.asJson).attemptNarrow[ElasticSearchQueryError]) | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } |
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
89 changes: 65 additions & 24 deletions
89
...pfl/bluebrain/nexus/delta/plugins/elasticsearch/metrics/EventMetricsProjectionSuite.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 |
|---|---|---|
| @@ -1,49 +1,90 @@ | ||
| package ch.epfl.bluebrain.nexus.delta.plugins.elasticsearch.metrics | ||
|
|
||
| import cats.effect.IO | ||
| import ch.epfl.bluebrain.nexus.delta.plugins.elasticsearch.metrics.MetricsStream._ | ||
| import ch.epfl.bluebrain.nexus.delta.plugins.elasticsearch.{EventMetricsProjection, Fixtures} | ||
| import ch.epfl.bluebrain.nexus.delta.plugins.elasticsearch.client.ElasticSearchClient.Refresh | ||
| import ch.epfl.bluebrain.nexus.delta.plugins.elasticsearch.indexing.ElasticSearchSink | ||
| import ch.epfl.bluebrain.nexus.delta.plugins.elasticsearch.metrics.EventMetricsProjectionSuite.{metric1, metric2} | ||
| import ch.epfl.bluebrain.nexus.delta.plugins.elasticsearch.{ElasticSearchClientSetup, Fixtures} | ||
| import ch.epfl.bluebrain.nexus.delta.rdf.Vocabulary.nxv | ||
| import ch.epfl.bluebrain.nexus.delta.rdf.syntax.iriStringContextSyntax | ||
| import ch.epfl.bluebrain.nexus.delta.sdk.metrics.ProjectScopedMetricStream | ||
| import ch.epfl.bluebrain.nexus.delta.sdk.model.metrics.EventMetric.{Created, ProjectScopedMetric, Updated} | ||
| import ch.epfl.bluebrain.nexus.delta.sourcing.model.Identity.Anonymous | ||
| import ch.epfl.bluebrain.nexus.delta.sourcing.model.{EntityType, Label, ProjectRef} | ||
| import ch.epfl.bluebrain.nexus.delta.sourcing.offset.Offset | ||
| import ch.epfl.bluebrain.nexus.delta.sourcing.stream.{CacheSink, ProjectionProgress, SupervisorSetup} | ||
| import ch.epfl.bluebrain.nexus.delta.sourcing.stream.{ProjectionProgress, SupervisorSetup} | ||
| import ch.epfl.bluebrain.nexus.testkit.mu.NexusSuite | ||
| import ch.epfl.bluebrain.nexus.testkit.mu.ce.PatienceConfig | ||
| import io.circe.Json | ||
| import io.circe.syntax.EncoderOps | ||
| import io.circe.syntax.{EncoderOps, KeyOps} | ||
| import io.circe.{Json, JsonObject} | ||
| import munit.AnyFixture | ||
|
|
||
| import java.time.Instant | ||
| import scala.concurrent.duration.DurationInt | ||
|
|
||
| class EventMetricsProjectionSuite extends NexusSuite with SupervisorSetup.Fixture with Fixtures { | ||
| class EventMetricsProjectionSuite | ||
| extends NexusSuite | ||
| with SupervisorSetup.Fixture | ||
| with ElasticSearchClientSetup.Fixture | ||
| with Fixtures { | ||
|
|
||
| override def munitFixtures: Seq[AnyFixture[_]] = List(supervisor) | ||
| override def munitFixtures: Seq[AnyFixture[_]] = List(supervisor, esClient) | ||
|
|
||
| implicit private val patienceConfig: PatienceConfig = PatienceConfig(2.seconds, 10.millis) | ||
|
|
||
| private lazy val sv = supervisor().supervisor | ||
| private val sink = CacheSink.events[Json] | ||
| private val index = eventMetricsIndex("nexus") | ||
|
|
||
| test("Start the metrics projection") { | ||
| private lazy val sv = supervisor().supervisor | ||
| private lazy val client = esClient() | ||
| private lazy val sink = ElasticSearchSink.events(client, 2, 50.millis, index, Refresh.True) | ||
|
|
||
| test("Start the metrics projection and index metrics") { | ||
| def createIndex = client | ||
| .createIndex(index, Some(metricsMapping.value), Some(metricsSettings.value)) | ||
| .assertEquals(true) | ||
| for { | ||
| _ <- EventMetricsProjection( | ||
| sink, | ||
| sv, | ||
| _ => metricsStream.take(2), | ||
| IO.unit | ||
| ) | ||
| _ <- EventMetricsProjection(sink, sv, _ => EventMetricsProjectionSuite.stream, createIndex) | ||
| _ <- sv.describe(EventMetricsProjection.projectionMetadata.name) | ||
| .map(_.map(_.progress)) | ||
| .assertEquals(Some(ProjectionProgress(Offset.at(2L), Instant.EPOCH, 2, 0, 0))) | ||
| .eventually | ||
| _ <- client.count(index.value).assertEquals(2L) | ||
| // Asserting the sources | ||
| _ <- client.getSource[Json](index, metric1.eventId).assertEquals(metric1.asJson) | ||
| _ <- client.getSource[Json](index, metric2.eventId).assertEquals(metric2.asJson) | ||
| } yield () | ||
| } | ||
| } | ||
|
|
||
| test("Sink has the correct metrics") { | ||
| assertEquals(sink.successes.size, 2) | ||
| assert(sink.dropped.isEmpty) | ||
| assert(sink.failed.isEmpty) | ||
| assert(sink.successes.values.toSet.contains(metric1.asJson)) | ||
| assert(sink.successes.values.toSet.contains(metric2.asJson)) | ||
| } | ||
| object EventMetricsProjectionSuite { | ||
| private val org = Label.unsafe("org") | ||
| private val proj1 = Label.unsafe("proj1") | ||
| private val projectRef1: ProjectRef = ProjectRef(org, proj1) | ||
|
|
||
| private val metric1: ProjectScopedMetric = ProjectScopedMetric( | ||
| Instant.EPOCH, | ||
| Anonymous, | ||
| 1, | ||
| Set(Created), | ||
| projectRef1, | ||
| org, | ||
| iri"http://bbp.epfl.ch/file1", | ||
| Set(nxv + "Resource1", nxv + "Resource2"), | ||
| JsonObject("extraField" := "extraValue") | ||
| ) | ||
| private val metric2: ProjectScopedMetric = ProjectScopedMetric( | ||
| Instant.EPOCH, | ||
| Anonymous, | ||
| 2, | ||
| Set(Updated), | ||
| projectRef1, | ||
| org, | ||
| iri"http://bbp.epfl.ch/file1", | ||
| Set(nxv + "Resource1", nxv + "Resource3"), | ||
| JsonObject( | ||
| "extraField" := "extraValue", | ||
| "extraField2" := 42 | ||
| ) | ||
| ) | ||
|
|
||
| private val stream = ProjectScopedMetricStream(EntityType("entity"), metric1, metric2) | ||
| } |
Oops, something went wrong.