chore: features for 2.2.4 (#912)

* chore: events throughput on the Event Log dashboard * feat: sent_events_count metrics collecting sent events counts (#909) * fix: authorizer to work with security records with mixed visibility (#908) * refactor: GitLab client in KG service (#906) * chore: CLI upgraded to 1.1.0 * fix: transformation flow to deduplicate different members with the same id (#903) * fix: to awaiting deletion status change not to fail if the event is gone * fix: missing project on cloning error as MalformedRepository * fix: auth related problems on cloning as recoverable failures * feat: migrations to restore failed events (#900) * fix: triples generation not to fail when repo is dirty after checkout (#898) * fix: event for repos that cannot be cloned to be non-recoverable (#897) * fix: sparql encoder to escape ' (#895) * fix: event persistence to work in case of race condition (#893) * fix: reduce noise in EL logs (#891) * chore: MemberSyncEvent to be sent less frequently in the first hour * fix: triples removal process not to timeout after 45s * fix: token association not to fail on race condition (#889) * fix: status change event not to fail on 0 updated (#888) * fix: TG processes not to log malformed repo issues (#887) * fix: added retry mechanism when storing and fetching the access token (#886) * fix: auth related issues on cloning to be classified as AuthRecoverableError (#885) * chore: sentry-sdk for renku CLI upgraded to 1.5.5 * fix: project clean-up process not to log error for 401 on webhook removal (#884) * fix: global commit sync finder not to throw error when project is gone (#882) * fix: GET /events to render links in the response correctly * fix: global commit sync to traverse commits from the oldest (#881) * fix: commit sync flows to call GitLab without token on 403 (#878)
SwissDataScienceCenter · Mar 9, 2022 · 8c958f6 · 8c958f6
1 parent a4f29ef
commit 8c958f6
Show file tree

Hide file tree

Showing 372 changed files with 5,426 additions and 3,909 deletions.
diff --git a/acceptance-tests/src/test/scala/io/renku/graph/acceptancetests/knowledgegraph/package.scala b/acceptance-tests/src/test/scala/io/renku/graph/acceptancetests/knowledgegraph/package.scala
@@ -1,3 +1,21 @@
+/*
+ * Copyright 2022 Swiss Data Science Center (SDSC)
+ * A partnership between École Polytechnique Fédérale de Lausanne (EPFL) and
+ * Eidgenössische Technische Hochschule Zürich (ETHZ).
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
 package io.renku.graph.acceptancetests
 
 package object knowledgegraph {

diff --git a/acceptance-tests/src/test/scala/io/renku/graph/acceptancetests/stubs/GitLab.scala b/acceptance-tests/src/test/scala/io/renku/graph/acceptancetests/stubs/GitLab.scala
@@ -190,7 +190,7 @@ trait GitLab {
         .withAccessTokenInHeader
     }
     stubFor {
-      get(s"/api/v4/projects/$projectId/repository/commits?page=1&per_page=50")
+      get(s"/api/v4/projects/$projectId/repository/commits?page=1&per_page=50&order=topo")
         .willReturn(okJson(commitIds.map(commitAsJson(_, theMostRecentEventDate)).asJson.noSpaces))
         .withAccessTokenInHeader
     }

diff --git a/commit-event-service/src/main/scala/io/renku/commiteventservice/Microservice.scala b/commit-event-service/src/main/scala/io/renku/commiteventservice/Microservice.scala
@@ -41,30 +41,32 @@ object Microservice extends IOMicroservice {
   val ServicePort:             Int Refined Positive = 9006
   private implicit val logger: Logger[IO]           = ApplicationLogger
 
-  override def run(args: List[String]): IO[ExitCode] = for {
-    certificateLoader     <- CertificateLoader[IO]
-    sentryInitializer     <- SentryInitializer[IO]
-    gitLabRateLimit       <- RateLimit.fromConfig[IO, GitLab]("services.gitlab.rate-limit")
-    gitLabThrottler       <- Throttler[IO, GitLab](gitLabRateLimit)
-    gitLabClient          <- GitLabClient[IO](gitLabThrottler)
-    executionTimeRecorder <- ExecutionTimeRecorder[IO]()
-    commitSyncCategory <-
-      events.categories.commitsync.SubscriptionFactory(gitLabClient, executionTimeRecorder)
-    globalCommitSyncCategory <-
-      events.categories.globalcommitsync.SubscriptionFactory(gitLabClient, executionTimeRecorder)
-    eventConsumersRegistry  <- consumers.EventConsumersRegistry(commitSyncCategory, globalCommitSyncCategory)
-    metricsRegistry         <- MetricsRegistry[IO]()
-    serviceReadinessChecker <- ServiceReadinessChecker[IO](ServicePort)
-    microserviceRoutes      <- MicroserviceRoutes(eventConsumersRegistry, new RoutesMetrics[IO](metricsRegistry))
-    exitcode <- microserviceRoutes.routes.use { routes =>
-                  new MicroserviceRunner[IO](serviceReadinessChecker,
-                                             certificateLoader,
-                                             sentryInitializer,
-                                             eventConsumersRegistry,
-                                             HttpServer[IO](serverPort = ServicePort.value, routes)
-                  ).run()
-                }
-  } yield exitcode
+  override def run(args: List[String]): IO[ExitCode] =
+    MetricsRegistry[IO]().flatMap { implicit metricsRegistry =>
+      for {
+        certificateLoader     <- CertificateLoader[IO]
+        sentryInitializer     <- SentryInitializer[IO]
+        gitLabRateLimit       <- RateLimit.fromConfig[IO, GitLab]("services.gitlab.rate-limit")
+        gitLabThrottler       <- Throttler[IO, GitLab](gitLabRateLimit)
+        gitLabClient          <- GitLabClient[IO](gitLabThrottler)
+        executionTimeRecorder <- ExecutionTimeRecorder[IO]()
+        commitSyncCategory <-
+          events.categories.commitsync.SubscriptionFactory(gitLabClient, executionTimeRecorder)
+        globalCommitSyncCategory <-
+          events.categories.globalcommitsync.SubscriptionFactory(gitLabClient, executionTimeRecorder)
+        eventConsumersRegistry  <- consumers.EventConsumersRegistry(commitSyncCategory, globalCommitSyncCategory)
+        serviceReadinessChecker <- ServiceReadinessChecker[IO](ServicePort)
+        microserviceRoutes      <- MicroserviceRoutes(eventConsumersRegistry, new RoutesMetrics[IO])
+        exitcode <- microserviceRoutes.routes.use { routes =>
+                      new MicroserviceRunner[IO](serviceReadinessChecker,
+                                                 certificateLoader,
+                                                 sentryInitializer,
+                                                 eventConsumersRegistry,
+                                                 HttpServer[IO](serverPort = ServicePort.value, routes)
+                      ).run()
+                    }
+      } yield exitcode
+    }
 }
 
 class MicroserviceRunner[F[_]: Spawn: Logger](serviceReadinessChecker: ServiceReadinessChecker[F],

diff --git a/...rc/main/scala/io/renku/commiteventservice/events/categories/commitsync/EventHandler.scala b/...rc/main/scala/io/renku/commiteventservice/events/categories/commitsync/EventHandler.scala
@@ -21,17 +21,17 @@ package io.renku.commiteventservice.events.categories.commitsync
 import cats.MonadThrow
 import cats.data.EitherT
 import cats.data.EitherT.fromEither
-import cats.effect.kernel.Temporal
 import cats.effect.{Async, Concurrent, Spawn}
 import cats.syntax.all._
 import io.circe.Decoder
 import io.renku.commiteventservice.events.categories.commitsync.eventgeneration.CommitsSynchronizer
 import io.renku.events.consumers.EventSchedulingResult.{Accepted, BadRequest}
 import io.renku.events.consumers._
-import io.renku.events.{EventRequestContent, consumers}
-import io.renku.graph.model.events.{CategoryName, CommitId, LastSyncedDate}
+import io.renku.events.{CategoryName, EventRequestContent, consumers}
+import io.renku.graph.model.events.{CommitId, LastSyncedDate}
 import io.renku.http.client.GitLabClient
 import io.renku.logging.ExecutionTimeRecorder
+import io.renku.metrics.MetricsRegistry
 import org.typelevel.log4cats.Logger
 
 import scala.util.control.NonFatal
@@ -85,7 +85,7 @@ private[events] class EventHandler[F[_]: MonadThrow: Spawn: Concurrent: Logger](
 }
 
 private[events] object EventHandler {
-  def apply[F[_]: Async: Spawn: Concurrent: Temporal: Logger](
+  def apply[F[_]: Async: Logger: MetricsRegistry](
       gitLabClient:          GitLabClient[F],
       executionTimeRecorder: ExecutionTimeRecorder[F]
   ): F[EventHandler[F]] = for {

diff --git a/.../scala/io/renku/commiteventservice/events/categories/commitsync/SubscriptionFactory.scala b/.../scala/io/renku/commiteventservice/events/categories/commitsync/SubscriptionFactory.scala
@@ -19,18 +19,18 @@
 package io.renku.commiteventservice.events.categories.commitsync
 
 import cats.effect.Async
-import cats.effect.kernel.{Concurrent, Temporal}
 import cats.syntax.all._
 import io.renku.commiteventservice.Microservice
 import io.renku.events.consumers.subscriptions.SubscriptionMechanism
 import io.renku.events.consumers.subscriptions.SubscriptionPayloadComposer.categoryAndUrlPayloadsComposerFactory
 import io.renku.http.client.GitLabClient
 import io.renku.logging.ExecutionTimeRecorder
+import io.renku.metrics.MetricsRegistry
 import org.typelevel.log4cats.Logger
 
 object SubscriptionFactory {
 
-  def apply[F[_]: Async: Concurrent: Temporal: Logger](
+  def apply[F[_]: Async: Logger: MetricsRegistry](
       gitLabClient:          GitLabClient[F],
       executionTimeRecorder: ExecutionTimeRecorder[F]
   ): F[(EventHandler[F], SubscriptionMechanism[F])] = for {

diff --git a/...commiteventservice/events/categories/commitsync/eventgeneration/CommitsSynchronizer.scala b/...commiteventservice/events/categories/commitsync/eventgeneration/CommitsSynchronizer.scala
@@ -21,24 +21,25 @@ package io.renku.commiteventservice.events.categories.commitsync.eventgeneration
 import cats.MonadThrow
 import cats.data.StateT
 import cats.effect.Async
-import cats.effect.kernel.Temporal
 import cats.syntax.all._
+import io.circe.literal._
 import io.renku.commiteventservice.events.categories.commitsync._
 import io.renku.commiteventservice.events.categories.common.SynchronizationSummary._
 import io.renku.commiteventservice.events.categories.common.UpdateResult._
 import io.renku.commiteventservice.events.categories.common._
 import io.renku.events.consumers.Project
+import io.renku.events.producers.EventSender
+import io.renku.events.{CategoryName, EventRequestContent}
 import io.renku.graph.model.events.{BatchDate, CommitId}
 import io.renku.graph.tokenrepository.AccessTokenFinder
 import io.renku.graph.tokenrepository.AccessTokenFinder._
 import io.renku.http.client.{AccessToken, GitLabClient}
 import io.renku.logging.ExecutionTimeRecorder
 import io.renku.logging.ExecutionTimeRecorder.ElapsedTime
+import io.renku.metrics.MetricsRegistry
 import org.typelevel.log4cats.Logger
-import io.circe.literal._
+
 import scala.util.control.NonFatal
-import io.renku.events.producers.EventSender
-import io.renku.events.EventRequestContent
 
 private[commitsync] trait CommitsSynchronizer[F[_]] {
   def synchronizeEvents(event: CommitSyncEvent): F[Unit]
@@ -103,9 +104,14 @@ private[commitsync] class CommitsSynchronizerImpl[F[_]: MonadThrow: Logger](
   private def triggerGlobalCommitSync(event: CommitSyncEvent): F[Unit] =
     eventSender.sendEvent(
       EventRequestContent.NoPayload(
-        json"""{ "categoryName": "GLOBAL_COMMIT_SYNC_REQUEST", "project":{ "id": ${event.project.id.value}, "path": ${event.project.path.value} }}"""
+        json"""{
+          "categoryName": "GLOBAL_COMMIT_SYNC_REQUEST", 
+          "project":{ "id": ${event.project.id.value}, "path": ${event.project.path.value}}
+        }"""
       ),
-      s"$categoryName - Triggering Global Commit Sync Failed"
+      EventSender.EventContext(CategoryName("GLOBAL_COMMIT_SYNC_REQUEST"),
+                               s"$categoryName - Triggering Global Commit Sync Failed"
+      )
     )
 
   private val DontCareCommitId = CommitId("0000000000000000000000000000000000000000")
@@ -168,8 +174,7 @@ private[commitsync] class CommitsSynchronizerImpl[F[_]: MonadThrow: Logger](
   private def loggingErrorAndIgnoreError(event:        CommitSyncEvent,
                                          errorMessage: String
   ): PartialFunction[Throwable, F[Unit]] = { case NonFatal(exception) =>
-    Logger[F]
-      .error(exception)(s"${logMessageCommon(event)} -> $errorMessage")
+    Logger[F].error(exception)(s"${logMessageCommon(event)} -> $errorMessage")
   }
 
   private def loggingError(event: CommitSyncEvent, errorMessage: String): PartialFunction[Throwable, F[Unit]] = {
@@ -223,7 +228,7 @@ private[commitsync] class CommitsSynchronizerImpl[F[_]: MonadThrow: Logger](
 }
 
 private[commitsync] object CommitsSynchronizer {
-  def apply[F[_]: Async: Temporal: Logger](
+  def apply[F[_]: Async: Logger: MetricsRegistry](
       gitLabClient:          GitLabClient[F],
       executionTimeRecorder: ExecutionTimeRecorder[F]
   ) = for {

diff --git a/.../commiteventservice/events/categories/commitsync/eventgeneration/EventDetailsFinder.scala b/.../commiteventservice/events/categories/commitsync/eventgeneration/EventDetailsFinder.scala
@@ -35,7 +35,6 @@ import org.http4s.circe.jsonOf
 import org.typelevel.log4cats.Logger
 
 private trait EventDetailsFinder[F[_]] {
-  def checkIfExists(projectId:   projects.Id, commitId: CommitId): F[Boolean]
   def getEventDetails(projectId: projects.Id, commitId: CommitId): F[Option[CommitWithParents]]
 }
 
@@ -46,27 +45,12 @@ private class EventDetailsFinderImpl[F[_]: Async: Temporal: Logger](
 
   import org.http4s.Method.GET
 
-  override def checkIfExists(projectId: projects.Id, commitId: CommitId): F[Boolean] =
-    fetchEventDetails(projectId, commitId)(mapResponseToBoolean)
-
   override def getEventDetails(projectId: projects.Id, commitId: CommitId): F[Option[CommitWithParents]] =
-    fetchEventDetails(projectId, commitId)(mapResponseCommitDetails)
-
-  private def fetchEventDetails[ResultType](projectId: projects.Id, commitId: CommitId)(
-      mapResponse: PartialFunction[(Status, Request[F], Response[F]), F[
-        ResultType
-      ]]
-  ) =
-    validateUri(s"$eventLogUrl/events/$commitId/$projectId") >>= (uri => send(request(GET, uri))(mapResponse))
+    validateUri(s"$eventLogUrl/events/$commitId/$projectId") >>=
+      (uri => send(request(GET, uri))(mapResponseCommitDetails))
 
-  private lazy val mapResponseToBoolean: PartialFunction[(Status, Request[F], Response[F]), F[Boolean]] = {
-    case (Ok, _, _)       => true.pure[F]
-    case (NotFound, _, _) => false.pure[F]
-  }
-
-  private lazy val mapResponseCommitDetails: PartialFunction[(Status, Request[F], Response[F]), F[
-    Option[CommitWithParents]
-  ]] = {
+  private lazy val mapResponseCommitDetails
+      : PartialFunction[(Status, Request[F], Response[F]), F[Option[CommitWithParents]]] = {
     case (Ok, _, response) => response.as[CommitWithParents].map(_.some)
     case (NotFound, _, _)  => Option.empty[CommitWithParents].pure[F]
   }
@@ -86,7 +70,6 @@ private class EventDetailsFinderImpl[F[_]: Async: Temporal: Logger](
       } yield CommitWithParents(id, projectId, parents)
     jsonOf[F, CommitWithParents]
   }
-
 }
 
 private object EventDetailsFinder {

diff --git a/.../commiteventservice/events/categories/commitsync/eventgeneration/LatestCommitFinder.scala b/.../commiteventservice/events/categories/commitsync/eventgeneration/LatestCommitFinder.scala
@@ -19,7 +19,6 @@
 package io.renku.commiteventservice.events.categories.commitsync.eventgeneration
 
 import cats.effect.Async
-import cats.effect.kernel.Temporal
 import cats.syntax.all._
 import eu.timepit.refined.auto._
 import io.circe.Decoder
@@ -36,7 +35,7 @@ private trait LatestCommitFinder[F[_]] {
   def findLatestCommit(projectId: Id)(implicit maybeAccessToken: Option[AccessToken]): F[Option[CommitInfo]]
 }
 
-private class LatestCommitFinderImpl[F[_]: Async: Temporal: Logger](
+private class LatestCommitFinderImpl[F[_]: Async: Logger](
     gitLabClient: GitLabClient[F]
 ) extends LatestCommitFinder[F] {
 
@@ -52,9 +51,9 @@ private class LatestCommitFinderImpl[F[_]: Async: Temporal: Logger](
     )(mapResponse(projectId))
 
   private def mapResponse(projectId: Id): PartialFunction[(Status, Request[F], Response[F]), F[Option[CommitInfo]]] = {
-    case (Ok, _, response)    => response.as[List[CommitInfo]] map (_.headOption)
-    case (NotFound, _, _)     => Option.empty[CommitInfo].pure[F]
-    case (Unauthorized, _, _) => findLatestCommit(projectId)(maybeAccessToken = None)
+    case (Ok, _, response)                => response.as[List[CommitInfo]] map (_.headOption)
+    case (NotFound, _, _)                 => Option.empty[CommitInfo].pure[F]
+    case (Unauthorized | Forbidden, _, _) => findLatestCommit(projectId)(maybeAccessToken = None)
   }
 
   private implicit val commitInfosEntityDecoder: EntityDecoder[F, List[CommitInfo]] = {

diff --git a/...u/commiteventservice/events/categories/commitsync/eventgeneration/ProjectInfoFinder.scala b/...u/commiteventservice/events/categories/commitsync/eventgeneration/ProjectInfoFinder.scala
diff --git a/...ice/src/main/scala/io/renku/commiteventservice/events/categories/commitsync/package.scala b/...ice/src/main/scala/io/renku/commiteventservice/events/categories/commitsync/package.scala
@@ -18,7 +18,7 @@
 
 package io.renku.commiteventservice.events.categories
 
-import io.renku.graph.model.events.CategoryName
+import io.renku.events.CategoryName
 
 package object commitsync {
   val categoryName: CategoryName = CategoryName("COMMIT_SYNC")
@@ -29,5 +29,4 @@ package object commitsync {
     case MinimalCommitSyncEvent(project) =>
       s"$categoryName: projectId = ${project.id}, projectPath = ${project.path}"
   }
-
 }