Access Google Cloud Storage via NIO (#6775)

* [WIP] Access Google Cloud Storage via NIO * info request * storage options * use credentials, try without gzip * WIP: streamline credential passing * wip: create gs file systems on demand * compile * bucket name * handling of path vs uri for gcs case * remove unused code * cleanup file system instantiation * allow legacy credentials, rename things * cache credentials, use alfu cache for file systems * re-add gunzip * rename credential type enum values * Add create route for google service account credentials * Add create route for google service account credentials * fix enum names * hide Optional hint in add-zarr view; also reset original input url when clicking Reset * format backend * pr feedback part 2: messages * changelog * Update CHANGELOG.unreleased.md Co-authored-by: Daniel <[email protected]> * catch parsing of invalid json * lint * fix typo * adapt docs * update and mention supported protocols in UI --------- Co-authored-by: Philipp Otto <[email protected]> Co-authored-by: Philipp Otto <[email protected]> Co-authored-by: Daniel <[email protected]>
scalableminds · Feb 7, 2023 · a185e65 · a185e65
1 parent 79d21cb
commit a185e65
Show file tree

Hide file tree

Showing 44 changed files with 650 additions and 426 deletions.
diff --git a/CHANGELOG.unreleased.md b/CHANGELOG.unreleased.md
@@ -11,6 +11,7 @@ For upgrade instructions, please check the [migration guide](MIGRATIONS.released
 [Commits](https://github.com/scalableminds/webknossos/compare/23.02.0...HEAD)
 
 ### Added
+- Remote datasets can now also be streamed from Google Cloud Storage URIs (`gs://`). [#6775](https://github.com/scalableminds/webknossos/pull/6775)
 
 ### Changed
 - Limit paid team sharing features to respective organization plans. [#6767](https://github.com/scalableminds/webknossos/pull/6776)

diff --git a/app/controllers/CredentialController.scala b/app/controllers/CredentialController.scala
@@ -2,28 +2,39 @@ package controllers
 
 import com.mohiva.play.silhouette.api.Silhouette
 import com.scalableminds.util.tools.FoxImplicits
-import com.scalableminds.webknossos.datastore.storage.{HttpBasicAuthCredential, S3AccessKeyCredential}
+import com.scalableminds.webknossos.datastore.storage.{
+  GoogleServiceAccountCredential,
+  HttpBasicAuthCredential,
+  S3AccessKeyCredential
+}
 import models.binary.credential.CredentialDAO
 import oxalis.security.WkEnv
-import play.api.libs.json.{Json, OFormat}
+import play.api.libs.json.{JsValue, Json, OFormat}
 import play.api.mvc.{Action, PlayBodyParsers}
 import utils.ObjectId
 
 import javax.inject.Inject
 import scala.concurrent.ExecutionContext
 
-case class HttpBasicAuthCredentialParameters(name: String, username: String, password: String, domain: Option[String])
+case class HttpBasicAuthCredentialParameters(name: String, username: String, password: String)
 
 object HttpBasicAuthCredentialParameters {
   implicit val jsonFormat: OFormat[HttpBasicAuthCredentialParameters] = Json.format[HttpBasicAuthCredentialParameters]
 }
 
-case class S3AccessKeyCredentialParameters(name: String, keyId: String, key: String, bucket: Option[String])
+case class S3AccessKeyCredentialParameters(name: String, accessKeyId: String, secretAccessKey: String)
 
 object S3AccessKeyCredentialParameters {
   implicit val jsonFormat: OFormat[S3AccessKeyCredentialParameters] = Json.format[S3AccessKeyCredentialParameters]
 }
 
+case class GoogleServiceAccountCredentialParameters(name: String, secretJson: JsValue)
+
+object GoogleServiceAccountCredentialParameters {
+  implicit val jsonFormat: OFormat[GoogleServiceAccountCredentialParameters] =
+    Json.format[GoogleServiceAccountCredentialParameters]
+}
+
 class CredentialController @Inject()(credentialDAO: CredentialDAO, sil: Silhouette[WkEnv])(
     implicit ec: ExecutionContext,
     val bodyParsers: PlayBodyParsers)
@@ -54,12 +65,27 @@ class CredentialController @Inject()(credentialDAO: CredentialDAO, sil: Silhouet
         _ <- credentialDAO.insertOne(
           _id,
           S3AccessKeyCredential(request.body.name,
-                                request.body.keyId,
-                                request.body.key,
+                                request.body.accessKeyId,
+                                request.body.secretAccessKey,
                                 request.identity._id.toString,
                                 request.identity._organization.toString)
         ) ?~> "create.failed"
       } yield Ok(Json.toJson(_id))
     }
 
+  def createGoogleServiceAccountCredential: Action[GoogleServiceAccountCredentialParameters] =
+    sil.SecuredAction.async(validateJson[GoogleServiceAccountCredentialParameters]) { implicit request =>
+      val _id = ObjectId.generate
+      for {
+        _ <- bool2Fox(request.identity.isAdmin) ?~> "notAllowed" ~> FORBIDDEN
+        _ <- credentialDAO.insertOne(
+          _id,
+          GoogleServiceAccountCredential(request.body.name,
+                                         request.body.secretJson,
+                                         request.identity._id.toString,
+                                         request.identity._organization.toString)
+        ) ?~> "create.failed"
+      } yield Ok(Json.toJson(_id))
+    }
+
 }
diff --git a/app/models/binary/credential/CredentialDAO.scala b/app/models/binary/credential/CredentialDAO.scala
@@ -1,8 +1,15 @@
 package models.binary.credential
 
 import com.scalableminds.util.tools.Fox
-import com.scalableminds.webknossos.datastore.storage.{AnyCredential, HttpBasicAuthCredential, S3AccessKeyCredential}
+import com.scalableminds.webknossos.datastore.storage.{
+  FileSystemCredential,
+  GoogleServiceAccountCredential,
+  HttpBasicAuthCredential,
+  S3AccessKeyCredential
+}
 import com.scalableminds.webknossos.schema.Tables.{Credentials, CredentialsRow}
+import net.liftweb.util.Helpers.tryo
+import play.api.libs.json.Json
 import utils.sql.{SecuredSQLDAO, SqlClient, SqlToken}
 import utils.ObjectId
 
@@ -42,28 +49,51 @@ class CredentialDAO @Inject()(sqlClient: SqlClient)(implicit ec: ExecutionContex
         r._Organization
       )
 
+  private def parseAsGoogleServiceAccountCredential(r: CredentialsRow): Fox[GoogleServiceAccountCredential] =
+    for {
+      secret <- r.secret.toFox
+      secretJson <- tryo(Json.parse(secret)).toFox
+    } yield
+      GoogleServiceAccountCredential(
+        r.name,
+        secretJson,
+        r._User,
+        r._Organization
+      )
+
   def insertOne(_id: ObjectId, credential: HttpBasicAuthCredential): Fox[Unit] =
     for {
       _ <- run(q"""insert into webknossos.credentials(_id, type, name, identifier, secret, _user, _organization)
-                     values(${_id}, ${CredentialType.HTTP_Basic_Auth}, ${credential.name}, ${credential.username}, ${credential.password}, ${credential.user}, ${credential.organization})""".asUpdate)
+                     values(${_id}, ${CredentialType.HttpBasicAuth}, ${credential.name}, ${credential.username}, ${credential.password}, ${credential.user}, ${credential.organization})""".asUpdate)
     } yield ()
 
   def insertOne(_id: ObjectId, credential: S3AccessKeyCredential): Fox[Unit] =
     for {
       _ <- run(q"""insert into webknossos.credentials(_id, type, name, identifier, secret, _user, _organization)
-                     values(${_id}, ${CredentialType.S3_Access_Key}, ${credential.name}, ${credential.keyId}, ${credential.key}, ${credential.user}, ${credential.organization})""".asUpdate)
+                       values(${_id}, ${CredentialType.S3AccessKey}, ${credential.name}, ${credential.accessKeyId}, ${credential.secretAccessKey}, ${credential.user}, ${credential.organization})""".asUpdate)
+    } yield ()
+
+  def insertOne(_id: ObjectId, credential: GoogleServiceAccountCredential): Fox[Unit] =
+    for {
+      _ <- run(q"""insert into webknossos.credentials(_id, type, name, secret, _user, _organization)
+                       values(${_id}, ${CredentialType.GoogleServiceAccount}, ${credential.name}, ${credential.secretJson.toString}, ${credential.user}, ${credential.organization})""".asUpdate)
     } yield ()
 
-  def findOne(id: ObjectId): Fox[AnyCredential] =
+  def findOne(id: ObjectId): Fox[FileSystemCredential] =
     for {
       r <- run(q"select $columns from webknossos.credentials_ where _id = $id".as[CredentialsRow])
       firstRow <- r.headOption.toFox
       parsed <- parseAnyCredential(firstRow)
     } yield parsed
 
-  private def parseAnyCredential(r: CredentialsRow): Fox[AnyCredential] =
-    r.`type` match {
-      case "HTTP_Basic_Auth" => parseAsHttpBasicAuthCredential(r)
-      case "S3_Access_Key"   => parseAsS3AccessKeyCredential(r)
-    }
+  private def parseAnyCredential(r: CredentialsRow): Fox[FileSystemCredential] =
+    for {
+      typeParsed <- CredentialType.fromString(r.`type`).toFox
+      parsed <- typeParsed match {
+        case CredentialType.HttpBasicAuth        => parseAsHttpBasicAuthCredential(r)
+        case CredentialType.S3AccessKey          => parseAsS3AccessKeyCredential(r)
+        case CredentialType.GoogleServiceAccount => parseAsGoogleServiceAccountCredential(r)
+        case _                                   => Fox.failure(s"Unknown credential type: ${r.`type`}")
+      }
+    } yield parsed
 }
diff --git a/app/models/binary/credential/CredentialService.scala b/app/models/binary/credential/CredentialService.scala
@@ -2,52 +2,59 @@ package models.binary.credential
 
 import com.scalableminds.util.tools.Fox
 import com.scalableminds.webknossos.datastore.storage.{
+  FileSystemCredential,
   FileSystemsHolder,
+  GoogleServiceAccountCredential,
   HttpBasicAuthCredential,
   S3AccessKeyCredential
 }
+import net.liftweb.util.Helpers.tryo
+import play.api.libs.json.Json
 import utils.ObjectId
 
 import java.net.URI
 import javax.inject.Inject
 import scala.concurrent.ExecutionContext
 
-class CredentialService @Inject()(credentialDao: CredentialDAO) {
+class CredentialService @Inject()(credentialDAO: CredentialDAO) {
 
-  def createCredential(uri: URI,
-                       username: Option[String],
-                       password: Option[String],
-                       user: String,
-                       organization: String)(implicit ec: ExecutionContext): Fox[Option[ObjectId]] = {
-    val scheme = uri.getScheme
-    scheme match {
+  def createCredentialOpt(uri: URI,
+                          credentialIdentifier: Option[String],
+                          credentialSecret: Option[String],
+                          userId: ObjectId,
+                          organizationId: ObjectId): Option[FileSystemCredential] =
+    uri.getScheme match {
       case FileSystemsHolder.schemeHttps =>
-        username match {
-          case Some(u) =>
-            val _id = ObjectId.generate
-            for {
-              _ <- credentialDao.insertOne(
-                _id,
-                HttpBasicAuthCredential(uri.toString, u, password.getOrElse(""), user, organization))
-            } yield Some(_id)
-          case None => Fox.successful(None)
-        }
+        credentialIdentifier.map(
+          username =>
+            HttpBasicAuthCredential(uri.toString,
+                                    username,
+                                    credentialSecret.getOrElse(""),
+                                    userId.toString,
+                                    organizationId.toString))
       case FileSystemsHolder.schemeS3 =>
-        username match {
-          case Some(keyId) =>
-            password match {
-              case Some(secretKey) =>
-                val _id = ObjectId.generate
-                for {
-                  _ <- credentialDao.insertOne(
-                    _id,
-                    S3AccessKeyCredential(uri.toString, keyId, secretKey, user, organization))
-                } yield Some(_id)
-              case None => Fox.successful(None)
-            }
-          case None => Fox.successful(None)
+        (credentialIdentifier, credentialSecret) match {
+          case (Some(keyId), Some(secretKey)) =>
+            Some(S3AccessKeyCredential(uri.toString, keyId, secretKey, userId.toString, organizationId.toString))
+          case _ => None
         }
+      case FileSystemsHolder.schemeGS =>
+        for {
+          secret <- credentialSecret
+          secretJson <- tryo(Json.parse(secret)).toOption
+        } yield GoogleServiceAccountCredential(uri.toString, secretJson, userId.toString, organizationId.toString)
     }
+
+  def insertOne(credential: FileSystemCredential)(implicit ec: ExecutionContext): Fox[ObjectId] = {
+    val _id = ObjectId.generate
+    for {
+      _ <- credential match {
+        case c: HttpBasicAuthCredential        => credentialDAO.insertOne(_id, c)
+        case c: S3AccessKeyCredential          => credentialDAO.insertOne(_id, c)
+        case c: GoogleServiceAccountCredential => credentialDAO.insertOne(_id, c)
+        case _                                 => Fox.failure("Unknown credential type")
+      }
+    } yield _id
   }
 
 }
diff --git a/app/models/binary/credential/CredentialType.scala b/app/models/binary/credential/CredentialType.scala
@@ -5,5 +5,5 @@ import com.scalableminds.util.enumeration.ExtendedEnumeration
 object CredentialType extends ExtendedEnumeration {
   type CredentialType = Value
 
-  val HTTP_Basic_Auth, S3_Access_Key, HTTP_Token, GCS = Value
+  val HttpBasicAuth, HttpToken, S3AccessKey, GoogleServiceAccount = Value
 }
diff --git a/app/models/binary/explore/ExploreRemoteLayerService.scala b/app/models/binary/explore/ExploreRemoteLayerService.scala
@@ -7,7 +7,7 @@ import com.scalableminds.webknossos.datastore.dataformats.zarr._
 import com.scalableminds.webknossos.datastore.datareaders.n5.N5Header
 import com.scalableminds.webknossos.datastore.datareaders.zarr._
 import com.scalableminds.webknossos.datastore.models.datasource._
-import com.scalableminds.webknossos.datastore.storage.FileSystemsHolder
+import com.scalableminds.webknossos.datastore.storage.{FileSystemsHolder, RemoteSourceDescriptor}
 import com.typesafe.scalalogging.LazyLogging
 import models.binary.credential.CredentialService
 import models.user.User
@@ -23,7 +23,9 @@ import scala.collection.mutable.ListBuffer
 import scala.concurrent.ExecutionContext
 import scala.util.Try
 
-case class ExploreRemoteDatasetParameters(remoteUri: String, user: Option[String], password: Option[String])
+case class ExploreRemoteDatasetParameters(remoteUri: String,
+                                          credentialIdentifier: Option[String],
+                                          credentialSecret: Option[String])
 
 object ExploreRemoteDatasetParameters {
   implicit val jsonFormat: OFormat[ExploreRemoteDatasetParameters] = Json.format[ExploreRemoteDatasetParameters]
@@ -39,8 +41,8 @@ class ExploreRemoteLayerService @Inject()(credentialService: CredentialService)
       exploredLayersNested <- Fox.serialCombined(urisWithCredentials)(
         parameters =>
           exploreRemoteLayersForUri(parameters.remoteUri,
-                                    parameters.user,
-                                    parameters.password,
+                                    parameters.credentialIdentifier,
+                                    parameters.credentialSecret,
                                     reportMutable,
                                     requestIdentity))
       layersWithVoxelSizes = exploredLayersNested.flatten
@@ -138,25 +140,27 @@ class ExploreRemoteLayerService @Inject()(credentialService: CredentialService)
 
   private def exploreRemoteLayersForUri(
       layerUri: String,
-      user: Option[String],
-      password: Option[String],
+      credentialIdentifier: Option[String],
+      credentialSecret: Option[String],
       reportMutable: ListBuffer[String],
       requestingUser: User)(implicit ec: ExecutionContext): Fox[List[(DataLayer, Vec3Double)]] =
     for {
-      remoteSource <- tryo(RemoteSourceDescriptor(new URI(normalizeUri(layerUri)), user, password)).toFox ?~> s"Received invalid URI: $layerUri"
-      credentialId <- credentialService.createCredential(
-        new URI(normalizeUri(layerUri)),
-        user,
-        password,
-        requestingUser._id.toString,
-        requestingUser._organization.toString) ?~> "Failed to set up remote file system credentaial"
-      fileSystem <- FileSystemsHolder.getOrCreate(remoteSource).toFox ?~> "Failed to set up remote file system"
-      remotePath <- tryo(fileSystem.getPath(remoteSource.remotePath)) ?~> "Failed to get remote path"
+      uri <- tryo(new URI(normalizeUri(layerUri))) ?~> s"Received invalid URI: $layerUri"
+      credentialOpt = credentialService.createCredentialOpt(uri,
+                                                            credentialIdentifier,
+                                                            credentialSecret,
+                                                            requestingUser._id,
+                                                            requestingUser._organization)
+      remoteSource = RemoteSourceDescriptor(uri, credentialOpt)
+      credentialId <- Fox.runOptional(credentialOpt)(c => credentialService.insertOne(c)) ?~> "remoteFileSystem.credential.insert.failed"
+      fileSystem <- FileSystemsHolder.getOrCreate(remoteSource) ?~> "remoteFileSystem.setup.failed"
+      remotePath <- tryo(fileSystem.getPath(FileSystemsHolder.pathFromUri(remoteSource.uri))) ?~> "remoteFileSystem.getPath.failed"
       layersWithVoxelSizes <- exploreRemoteLayersForRemotePath(
         remotePath,
         credentialId.map(_.toString),
         reportMutable,
-        List(new ZarrArrayExplorer, new NgffExplorer, new N5ArrayExplorer, new N5MultiscalesExplorer))
+        List(new ZarrArrayExplorer, new NgffExplorer, new N5ArrayExplorer, new N5MultiscalesExplorer)
+      )
     } yield layersWithVoxelSizes
 
   private def normalizeUri(uri: String): String =

diff --git a/app/models/binary/explore/N5ArrayExplorer.scala b/app/models/binary/explore/N5ArrayExplorer.scala
@@ -22,7 +22,12 @@ class N5ArrayExplorer extends RemoteLayerExplorer {
       elementClass <- n5Header.elementClass ?~> "failed to read element class from n5 header"
       guessedAxisOrder = AxisOrder.asZyxFromRank(n5Header.rank)
       boundingBox <- n5Header.boundingBox(guessedAxisOrder) ?~> "failed to read bounding box from zarr header. Make sure data is in (T/C)ZYX format"
-      magLocator = MagLocator(Vec3Int.ones, Some(remotePath.toString), None, Some(guessedAxisOrder), None, credentialId)
+      magLocator = MagLocator(Vec3Int.ones,
+                              Some(remotePath.toUri.toString),
+                              None,
+                              Some(guessedAxisOrder),
+                              None,
+                              credentialId)
       layer: N5Layer = if (looksLikeSegmentationLayer(name, elementClass)) {
         N5SegmentationLayer(name, boundingBox, elementClass, List(magLocator), largestSegmentId = None)
       } else N5DataLayer(name, Category.color, boundingBox, elementClass, List(magLocator))

diff --git a/app/models/binary/explore/N5MultiscalesExplorer.scala b/app/models/binary/explore/N5MultiscalesExplorer.scala
@@ -108,7 +108,7 @@ class N5MultiscalesExplorer extends RemoteLayerExplorer with FoxImplicits {
       elementClass <- n5Header.elementClass ?~> s"failed to read element class from n5 header at $headerPath"
       boundingBox <- n5Header.boundingBox(axisOrder) ?~> s"failed to read bounding box from n5 header at $headerPath"
     } yield
-      MagWithAttributes(MagLocator(mag, Some(magPath.toString), None, Some(axisOrder), None, credentialId),
+      MagWithAttributes(MagLocator(mag, Some(magPath.toUri.toString), None, Some(axisOrder), None, credentialId),
                         magPath,
                         elementClass,
                         boundingBox)

diff --git a/app/models/binary/explore/NgffExplorer.scala b/app/models/binary/explore/NgffExplorer.scala
@@ -126,10 +126,11 @@ class NgffExplorer extends RemoteLayerExplorer {
       elementClass <- zarrHeader.elementClass ?~> s"failed to read element class from zarr header at $zarrayPath"
       boundingBox <- zarrHeader.boundingBox(axisOrder) ?~> s"failed to read bounding box from zarr header at $zarrayPath"
     } yield
-      MagWithAttributes(MagLocator(mag, Some(magPath.toString), None, Some(axisOrder), channelIndex, credentialId),
-                        magPath,
-                        elementClass,
-                        boundingBox)
+      MagWithAttributes(
+        MagLocator(mag, Some(magPath.toUri.toString), None, Some(axisOrder), channelIndex, credentialId),
+        magPath,
+        elementClass,
+        boundingBox)
 
   private def extractAxisOrder(axes: List[NgffAxis]): Fox[AxisOrder] = {
     def axisMatches(axis: NgffAxis, name: String) = axis.name.toLowerCase == name && axis.`type` == "space"

diff --git a/app/models/binary/explore/RemoteLayerExplorer.scala b/app/models/binary/explore/RemoteLayerExplorer.scala
@@ -1,6 +1,7 @@
 package models.binary.explore
 
 import com.scalableminds.util.geometry.{BoundingBox, Vec3Double}
+import com.scalableminds.util.io.ZipIO
 import com.scalableminds.util.tools.{Fox, FoxImplicits, JsonHelper}
 import com.scalableminds.webknossos.datastore.dataformats.MagLocator
 import com.scalableminds.webknossos.datastore.models.datasource.{DataLayer, ElementClass}
@@ -24,8 +25,9 @@ trait RemoteLayerExplorer extends FoxImplicits {
 
   protected def parseJsonFromPath[T: Reads](path: Path): Fox[T] =
     for {
-      fileAsString <- tryo(new String(Files.readAllBytes(path), StandardCharsets.UTF_8)).toFox ?~> "Failed to read remote file"
-      parsed <- JsonHelper.parseAndValidateJson[T](fileAsString) ?~> "Failed to parse or validate json against data schema"
+      fileBytes <- tryo(ZipIO.tryGunzip(Files.readAllBytes(path))) ?~> "dataSet.explore.failed.readFile"
+      fileAsString <- tryo(new String(fileBytes, StandardCharsets.UTF_8)).toFox ?~> "dataSet.explore.failed.readFile"
+      parsed <- JsonHelper.parseAndValidateJson[T](fileAsString)
     } yield parsed
 
   protected def looksLikeSegmentationLayer(layerName: String, elementClass: ElementClass.Value): Boolean =