Loki for voxelytics logs

CHANGELOG.unreleased.md

@@ -35,3 +35,4 @@ For upgrade instructions, please check the [migration guide](MIGRATIONS.released
 ### Removed
 
 ### Breaking Changes
+- Changes the storage backend for Voxelytics logging from Elasticsearch to Loki. [#6770](https://github.com/scalableminds/webknossos/pull/6770)

MIGRATIONS.unreleased.md

@@ -8,6 +8,8 @@ User-facing changes are documented in the [changelog](CHANGELOG.released.md).
 ## Unreleased
 [Commits](https://github.com/scalableminds/webknossos/compare/23.01.0...HEAD)
 
+- WEBKNOSSOS requires Loki instead of Elasticsearch for Voxelytics logging now. Please update the `application.conf`: Remove `voxelytics.elasticsearch.index`, rename `voxelytics.elasticsearch` to `voxelytics.loki`, and update `voxelytics.loki.uri`. [#6770](https://github.com/scalableminds/webknossos/pull/6770)
+
 ### Postgres Evolutions:
 
 - [094-pricing-plans.sql](conf/evolutions/reversions/094-pricing-plans.sql)

app/WebKnossosModule.scala

@@ -6,9 +6,9 @@ import models.binary.DataSetService
 import models.job.{JobService, WorkerLivenessService}
 import models.storage.UsedStorageService
 import models.task.TaskService
-import models.user.time.TimeSpanService
 import models.user._
-import models.voxelytics.ElasticsearchClient
+import models.user.time.TimeSpanService
+import models.voxelytics.LokiClient
 import oxalis.files.TempFileService
 import oxalis.mail.MailchimpTicker
 import oxalis.telemetry.SlackNotificationService
@@ -34,7 +34,7 @@ class WebKnossosModule extends AbstractModule {
     bind(classOf[SlackNotificationService]).asEagerSingleton()
     bind(classOf[AnalyticsSessionService]).asEagerSingleton()
     bind(classOf[WorkerLivenessService]).asEagerSingleton()
-    bind(classOf[ElasticsearchClient]).asEagerSingleton()
+    bind(classOf[LokiClient]).asEagerSingleton()
     bind(classOf[UsedStorageService]).asEagerSingleton()
   }
 }

app/controllers/VoxelyticsController.scala

@@ -2,6 +2,7 @@ package controllers
 
 import com.mohiva.play.silhouette.api.Silhouette
 import com.mohiva.play.silhouette.api.actions.SecuredRequest
+import com.scalableminds.util.time.Instant
 import com.scalableminds.util.tools.{Fox, FoxImplicits}
 import models.organization.OrganizationDAO
 import models.voxelytics._
@@ -18,7 +19,7 @@ class VoxelyticsController @Inject()(
     organizationDAO: OrganizationDAO,
     voxelyticsDAO: VoxelyticsDAO,
     voxelyticsService: VoxelyticsService,
-    elasticsearchClient: ElasticsearchClient,
+    lokiClient: LokiClient,
     wkConf: WkConf,
     sil: Silhouette[WkEnv])(implicit ec: ExecutionContext, bodyParsers: PlayBodyParsers)
     extends Controller
@@ -246,15 +247,17 @@ class VoxelyticsController @Inject()(
       for {
         _ <- bool2Fox(wkConf.Features.voxelyticsEnabled) ?~> "voxelytics.disabled"
         organization <- organizationDAO.findOne(request.identity._organization)
-        logEntries = request.body.map(
-          entry =>
-            entry ++ Json.obj("vx" -> ((entry \ "vx").as[JsObject] ++ Json.obj("wk_org" -> organization.name,
-                                                                               "wk_user" -> request.identity._id.id))))
-        _ <- elasticsearchClient.bulkInsert(logEntries)
+        logEntries = request.body
+        _ <- lokiClient.bulkInsert(logEntries, organization._id)
       } yield Ok
     }
 
-  def getLogs(runId: String, taskName: Option[String], minLevel: Option[String]): Action[AnyContent] =
+  def getLogs(runId: String,
+              taskName: Option[String],
+              minLevel: Option[String],
+              startTimestamp: Long,
+              endTimestamp: Long,
+              limit: Option[Long]): Action[AnyContent] =
     sil.SecuredAction.async { implicit request =>
       {
         for {
@@ -263,13 +266,16 @@ class VoxelyticsController @Inject()(
           runName <- voxelyticsDAO.getRunNameById(runIdValidated, request.identity._organization)
           _ <- voxelyticsService.checkAuth(runIdValidated, request.identity) ~> UNAUTHORIZED
           organization <- organizationDAO.findOne(request.identity._organization)
-          organizationName = organization.name
-          logEntries <- elasticsearchClient.queryLogs(
+          logEntries <- lokiClient.queryLogsBatched(
             runName,
-            organizationName,
+            organization._id,
             taskName,
-            minLevel.flatMap(VoxelyticsLogLevel.fromString).getOrElse(VoxelyticsLogLevel.INFO))
-        } yield JsonOk(logEntries)
+            minLevel.flatMap(VoxelyticsLogLevel.fromString).getOrElse(VoxelyticsLogLevel.INFO),
+            Instant(startTimestamp),
+            Instant(endTimestamp),
+            limit
+          )
+        } yield JsonOk(JsArray(logEntries))
       }
     }
 }