Measure used disk storage

CHANGELOG.unreleased.md

@@ -14,6 +14,7 @@ For upgrade instructions, please check the [migration guide](MIGRATIONS.released
 - Added sign in via OIDC. [#6534](https://github.com/scalableminds/webknossos/pull/6534)
 - Added a new datasets tab to the dashboard which supports managing datasets in folders. Folders can be organized hierarchically and datasets can be moved into these folders. Selecting a dataset will show dataset details in a sidebar. [#6591](https://github.com/scalableminds/webknossos/pull/6591)
 - Added the option to search a specific folder in the new datasets tab. [#6677](https://github.com/scalableminds/webknossos/pull/6677)
+- The storage used by an organization’s datasets can now be measured. [#6685](https://github.com/scalableminds/webknossos/pull/6685)
 
 ### Changed
 - webKnossos is now able to recover from a lost webGL context. [#6663](https://github.com/scalableminds/webknossos/pull/6663)

MIGRATIONS.unreleased.md

@@ -14,3 +14,4 @@ User-facing changes are documented in the [changelog](CHANGELOG.released.md).
 
 - [091-folders.sql](conf/evolutions/091-folders.sql)
 - [092-oidc.sql](conf/evolutions/092-oidc.sql)
+- [093-storage.sql](conf/evolutions/093-storage.sql)

app/WebKnossosModule.scala

@@ -4,6 +4,7 @@ import models.analytics.AnalyticsSessionService
 import models.annotation.AnnotationStore
 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._
@@ -35,5 +36,6 @@ class WebKnossosModule extends AbstractModule {
     bind(classOf[AnalyticsSessionService]).asEagerSingleton()
     bind(classOf[WorkerLivenessService]).asEagerSingleton()
     bind(classOf[ElasticsearchClient]).asEagerSingleton()
+    bind(classOf[UsedStorageService]).asEagerSingleton()
   }
 }

app/controllers/DataSetController.scala

@@ -99,6 +99,7 @@ class DataSetController @Inject()(userService: UserService,
               .clientFor(dataSet)(GlobalAccessContext)
               .flatMap(
                 _.requestDataLayerThumbnail(organizationName,
+                                            dataSet,
                                             dataLayerName,
                                             width,
                                             height,
@@ -320,7 +321,7 @@ class DataSetController @Inject()(userService: UserService,
         datalayer <- usableDataSource.dataLayers.headOption.toFox ?~> "dataSet.noLayers"
         _ <- dataSetService
           .clientFor(dataSet)(GlobalAccessContext)
-          .flatMap(_.findPositionWithData(organizationName, datalayer.name).flatMap(posWithData =>
+          .flatMap(_.findPositionWithData(organizationName, dataSet, datalayer.name).flatMap(posWithData =>
             bool2Fox(posWithData.value("position") != JsNull))) ?~> "dataSet.loadingDataFailed"
       } yield {
         Ok("Ok")

app/controllers/WKRemoteDataStoreController.scala

@@ -11,11 +11,11 @@ import com.scalableminds.webknossos.datastore.services.{
   ReserveUploadInformation
 }
 import com.typesafe.scalalogging.LazyLogging
-import javax.inject.Inject
 import models.analytics.{AnalyticsService, UploadDatasetEvent}
 import models.binary._
 import models.job.JobDAO
 import models.organization.OrganizationDAO
+import models.storage.UsedStorageService
 import models.user.{User, UserDAO, UserService}
 import net.liftweb.common.Full
 import oxalis.mail.{MailchimpClient, MailchimpTag}
@@ -25,6 +25,7 @@ import play.api.libs.json.{JsError, JsSuccess, JsValue, Json}
 import play.api.mvc.{Action, AnyContent, PlayBodyParsers}
 import utils.ObjectId
 
+import javax.inject.Inject
 import scala.concurrent.{ExecutionContext, Future}
 
 class WKRemoteDataStoreController @Inject()(
@@ -34,6 +35,7 @@ class WKRemoteDataStoreController @Inject()(
     analyticsService: AnalyticsService,
     userService: UserService,
     organizationDAO: OrganizationDAO,
+    usedStorageService: UsedStorageService,
     dataSetDAO: DataSetDAO,
     userDAO: UserDAO,
     jobDAO: JobDAO,
@@ -90,6 +92,7 @@ class WKRemoteDataStoreController @Inject()(
           dataSet <- dataSetDAO.findOneByNameAndOrganization(dataSetName, user._organization)(GlobalAccessContext) ?~> Messages(
             "dataSet.notFound",
             dataSetName) ~> NOT_FOUND
+          _ <- usedStorageService.refreshStorageReportForDataset(dataSet)
           _ = analyticsService.track(UploadDatasetEvent(user, dataSet, dataStore, dataSetSizeBytes))
           _ = mailchimpClient.tagUser(user, MailchimpTag.HasUploadedOwnDataset)
         } yield Ok
@@ -101,7 +104,11 @@ class WKRemoteDataStoreController @Inject()(
       request.body.validate[DataStoreStatus] match {
         case JsSuccess(status, _) =>
           logger.debug(s"Status update from data store '$name'. Status: " + status.ok)
-          dataStoreDAO.updateUrlByName(name, status.url).map(_ => Ok)
+          for {
+            _ <- dataStoreDAO.updateUrlByName(name, status.url)
+            _ <- dataStoreDAO.updateReportUsedStorageEnabledByName(name,
+                                                                   status.reportUsedStorageEnabled.getOrElse(false))
+          } yield Ok
         case e: JsError =>
           logger.error("Data store '$name' sent invalid update. Error: " + e)
           Future.successful(JsonBadRequest(JsError.toJson(e)))
@@ -154,8 +161,12 @@ class WKRemoteDataStoreController @Inject()(
           .findOneByNameAndOrganizationName(datasourceId.name, datasourceId.team)(GlobalAccessContext)
           .futureBox
         _ <- existingDataset.flatMap {
-          case Full(dataset) => dataSetDAO.deleteDataset(dataset._id)
-          case _             => Fox.successful(())
+          case Full(dataset) => {
+            dataSetDAO
+              .deleteDataset(dataset._id)
+              .flatMap(_ => usedStorageService.refreshStorageReportForDataset(dataset))
+          }
+          case _ => Fox.successful(())
         }
       } yield Ok
     }

app/models/binary/DataSetService.scala

@@ -256,7 +256,7 @@ class DataSetService @Inject()(organizationDAO: OrganizationDAO,
   def clientFor(dataSet: DataSet)(implicit ctx: DBAccessContext): Fox[WKRemoteDataStoreClient] =
     for {
       dataStore <- dataStoreFor(dataSet)
-    } yield new WKRemoteDataStoreClient(dataStore, dataSet, rpc)
+    } yield new WKRemoteDataStoreClient(dataStore, rpc)
 
   def lastUsedTimeFor(_dataSet: ObjectId, userOpt: Option[User]): Fox[Instant] =
     userOpt match {

app/models/binary/DataStore.scala

@@ -22,6 +22,7 @@ case class DataStore(
     isDeleted: Boolean = false,
     isConnector: Boolean = false,
     allowsUpload: Boolean = true,
+    reportUsedStorageEnabled: Boolean = false,
     onlyAllowedOrganization: Option[ObjectId] = None
 )
 
@@ -44,6 +45,7 @@ object DataStore {
       isDeleted = false,
       isConnector.getOrElse(false),
       allowsUpload.getOrElse(true),
+      reportUsedStorageEnabled = false,
       None
     )
 
@@ -101,6 +103,7 @@ class DataStoreDAO @Inject()(sqlClient: SQLClient)(implicit ec: ExecutionContext
         r.isdeleted,
         r.isconnector,
         r.allowsupload,
+        r.reportusedstorageenabled,
         r.onlyallowedorganization.map(ObjectId(_))
       ))
 
@@ -125,16 +128,28 @@ class DataStoreDAO @Inject()(sqlClient: SQLClient)(implicit ec: ExecutionContext
       parsed <- parseAll(r)
     } yield parsed
 
+  def findAllWithStorageReporting: Fox[List[DataStore]] =
+    for {
+      r <- run(sql"select #$columns from webknossos.datastores_ where reportUsedStorageEnabled".as[DatastoresRow])
+      parsed <- parseAll(r)
+    } yield parsed
+
   def updateUrlByName(name: String, url: String): Fox[Unit] = {
     val q = for { row <- Datastores if notdel(row) && row.name === name } yield row.url
     for { _ <- run(q.update(url)) } yield ()
   }
 
+  def updateReportUsedStorageEnabledByName(name: String, reportUsedStorageEnabled: Boolean): Fox[Unit] =
+    for {
+      _ <- run(
+        sqlu"UPDATE webknossos.dataStores SET reportUsedStorageEnabled = $reportUsedStorageEnabled WHERE name = $name")
+    } yield ()
+
   def insertOne(d: DataStore): Fox[Unit] =
     for {
       _ <- run(
-        sqlu"""insert into webknossos.dataStores(name, url, publicUrl, key, isScratch, isDeleted, isConnector, allowsUpload)
-                             values(${d.name}, ${d.url}, ${d.publicUrl},  ${d.key}, ${d.isScratch}, ${d.isDeleted}, ${d.isConnector}, ${d.allowsUpload})""")
+        sqlu"""insert into webknossos.dataStores(name, url, publicUrl, key, isScratch, isDeleted, isConnector, allowsUpload, reportUsedStorageEnabled)
+                             values(${d.name}, ${d.url}, ${d.publicUrl},  ${d.key}, ${d.isScratch}, ${d.isDeleted}, ${d.isConnector}, ${d.allowsUpload}, ${d.reportUsedStorageEnabled})""")
     } yield ()
 
   def deleteOneByName(name: String): Fox[Unit] =

app/models/binary/WKRemoteDataStoreClient.scala

@@ -3,16 +3,16 @@ package models.binary
 import com.scalableminds.util.geometry.Vec3Int
 import com.scalableminds.util.tools.Fox
 import com.scalableminds.webknossos.datastore.rpc.RPC
+import com.scalableminds.webknossos.datastore.services.DirectoryStorageReport
 import com.typesafe.scalalogging.LazyLogging
 import controllers.RpcTokenHolder
 import play.api.libs.json.JsObject
 import play.utils.UriEncoding
 
-class WKRemoteDataStoreClient(dataStore: DataStore, dataSet: DataSet, rpc: RPC) extends LazyLogging {
-
-  def baseInfo = s"Dataset: ${dataSet.name} Datastore: ${dataStore.url}"
+class WKRemoteDataStoreClient(dataStore: DataStore, rpc: RPC) extends LazyLogging {
 
   def requestDataLayerThumbnail(organizationName: String,
+                                dataSet: DataSet,
                                 dataLayerName: String,
                                 width: Int,
                                 height: Int,
@@ -29,12 +29,19 @@ class WKRemoteDataStoreClient(dataStore: DataStore, dataSet: DataSet, rpc: RPC)
       .getWithBytesResponse
   }
 
-  def findPositionWithData(organizationName: String, dataLayerName: String): Fox[JsObject] =
+  def findPositionWithData(organizationName: String, dataSet: DataSet, dataLayerName: String): Fox[JsObject] =
     rpc(
       s"${dataStore.url}/data/datasets/${urlEncode(organizationName)}/${dataSet.urlEncodedName}/layers/$dataLayerName/findData")
       .addQueryString("token" -> RpcTokenHolder.webKnossosToken)
       .getWithJsonResponse[JsObject]
 
   private def urlEncode(text: String) = UriEncoding.encodePathSegment(text, "UTF-8")
 
+  def fetchStorageReport(organizationName: String, datasetName: Option[String]): Fox[List[DirectoryStorageReport]] =
+    rpc(s"${dataStore.url}/data/datasets/measureUsedStorage/${urlEncode(organizationName)}")
+      .addQueryString("token" -> RpcTokenHolder.webKnossosToken)
+      .addQueryStringOptional("dataSetName", datasetName)
+      .silent
+      .getWithJsonResponse[List[DirectoryStorageReport]]
+
 }

app/models/organization/Organization.scala

@@ -3,16 +3,17 @@ package models.organization
 import com.scalableminds.util.accesscontext.DBAccessContext
 import com.scalableminds.util.time.Instant
 import com.scalableminds.util.tools.Fox
+import com.scalableminds.webknossos.datastore.services.DirectoryStorageReport
 import com.scalableminds.webknossos.schema.Tables._
-
-import javax.inject.Inject
 import models.team.PricingPlan
 import models.team.PricingPlan.PricingPlan
 import slick.jdbc.PostgresProfile.api._
 import slick.lifted.Rep
 import utils.{ObjectId, SQLClient, SQLDAO}
 
+import javax.inject.Inject
 import scala.concurrent.ExecutionContext
+import scala.concurrent.duration.FiniteDuration
 
 case class Organization(
     _id: ObjectId,
@@ -116,4 +117,66 @@ class OrganizationDAO @Inject()(sqlClient: SQLClient)(implicit ec: ExecutionCont
                       where _id = $organizationId""")
     } yield ()
 
+  def deleteUsedStorage(organizationId: ObjectId): Fox[Unit] =
+    for {
+      _ <- run(sqlu"DELETE FROM webknossos.organization_usedStorage WHERE _organization = $organizationId")
+    } yield ()
+
+  def deleteUsedStorageForDataset(datasetId: ObjectId): Fox[Unit] =
+    for {
+      _ <- run(sqlu"DELETE FROM webknossos.organization_usedStorage WHERE _dataSet = $datasetId")
+    } yield ()
+
+  def updateLastStorageScanTime(organizationId: ObjectId, time: Instant): Fox[Unit] =
+    for {
+      _ <- run(sqlu"UPDATE webknossos.organizations SET lastStorageScanTime = $time WHERE _id = $organizationId")
+    } yield ()
+
+  def upsertUsedStorage(organizationId: ObjectId,
+                        dataStoreName: String,
+                        usedStorageEntries: List[DirectoryStorageReport]): Fox[Unit] = {
+    val queries = usedStorageEntries.map(entry => sqlu"""
+               WITH ds AS (
+                 SELECT _id
+                 FROM webknossos.datasets_
+                 WHERE _organization = $organizationId
+                 AND name = ${entry.dataSetName}
+                 LIMIT 1
+               )
+               INSERT INTO webknossos.organization_usedStorage(
+                  _organization, _dataStore, _dataSet, layerName,
+                  magOrDirectoryName, usedStorageBytes, lastUpdated)
+               SELECT
+                $organizationId, $dataStoreName, ds._id, ${entry.layerName},
+                ${entry.magOrDirectoryName}, ${entry.usedStorageBytes}, NOW()
+               FROM ds
+               ON CONFLICT (_organization, _dataStore, _dataSet, layerName, magOrDirectoryName)
+               DO UPDATE
+                 SET usedStorageBytes = ${entry.usedStorageBytes}, lastUpdated = NOW()
+               """)
+    for {
+      _ <- Fox.serialCombined(queries)(q => run(q))
+    } yield ()
+  }
+
+  def getUsedStorage(organizationId: ObjectId): Fox[Long] =
+    for {
+      rows <- run(
+        sql"SELECT SUM(usedStorageBytes) FROM webknossos.organization_usedStorage WHERE _organization = $organizationId"
+          .as[Long])
+      firstRow <- rows.headOption
+    } yield firstRow
+
+  def findNotRecentlyScanned(scanInterval: FiniteDuration, limit: Int): Fox[List[Organization]] =
+    for {
+      rows <- run(sql"""
+                  SELECT #$columns
+                  FROM #$existingCollectionName
+                  WHERE lastStorageScanTime < ${Instant.now - scanInterval}
+                  ORDER BY lastStorageScanTime
+                  LIMIT $limit
+                  """.as[OrganizationsRow])
+      parsed <- parseAll(rows)
+    } yield parsed
+
 }

app/models/organization/OrganizationService.scala

@@ -34,15 +34,17 @@ class OrganizationService @Inject()(organizationDAO: OrganizationDAO,
         "pricingPlan" -> organization.pricingPlan
       )
     } else Json.obj()
-    Fox.successful(
+    for {
+      usedStorageBytes <- organizationDAO.getUsedStorage(organization._id)
+    } yield
       Json.obj(
         "id" -> organization._id.toString,
         "name" -> organization.name,
         "additionalInformation" -> organization.additionalInformation,
         "enableAutoVerify" -> organization.enableAutoVerify,
-        "displayName" -> organization.displayName
+        "displayName" -> organization.displayName,
+        "usedStorageBytes" -> usedStorageBytes
       ) ++ adminOnlyInfo
-    )
   }
 
   def findOneByInviteByNameOrDefault(inviteOpt: Option[Invite], organizationNameOpt: Option[String])(