xuang7 commented on code in PR #6447:
URL: https://github.com/apache/texera/pull/6447#discussion_r3626225070
##########
file-service/src/main/scala/org/apache/texera/service/resource/DatasetResource.scala:
##########
@@ -1361,6 +1305,187 @@ class DatasetResource extends LazyLogging {
}
}
+ @POST
+ @RolesAllowed(Array("REGULAR", "ADMIN"))
+ @Path("/{did}/drive-export")
+ @Consumes(Array(MediaType.APPLICATION_JSON))
+ def exportDatasetToDrive(
+ @PathParam("did") did: Integer,
+ @QueryParam("dvid") dvid: Integer,
+ @QueryParam("latest") latest: java.lang.Boolean,
+ request: DriveExportRequest,
+ @Auth user: SessionUser
+ ): Response = {
+ if
(!request.sessionUri.startsWith("https://www.googleapis.com/upload/drive/";)) {
+ throw new BadRequestException("Invalid session URI")
+ }
+ withTransaction(context) { ctx =>
+ if ((dvid != null && latest != null) || (dvid == null && latest ==
null)) {
+ throw new BadRequestException("Specify exactly one: dvid=<ID> OR
latest=true")
+ }
+
+ val uid = user.getUid
+ if (!userHasReadAccess(ctx, did, uid)) {
+ throw new ForbiddenException(ERR_USER_HAS_NO_ACCESS_TO_DATASET_MESSAGE)
+ }
+
+ val dataset = getDatasetByID(ctx, did)
+ if (!userOwnDataset(ctx, did, uid) && !dataset.getIsDownloadable) {
+ throw new ForbiddenException("Dataset download is not allowed")
+ }
+
+ val datasetVersion = if (dvid != null) {
+ getDatasetVersionByID(ctx, dvid)
+ } else {
+ getLatestDatasetVersion(ctx, did).getOrElse(
+ throw new NotFoundException(ERR_DATASET_VERSION_NOT_FOUND_MESSAGE)
+ )
+ }
+
+ val repositoryName = dataset.getRepositoryName
+ val versionHash = datasetVersion.getVersionHash
+ val objects = withLakeFSErrorHandling(
+ s"listing files of version '$versionHash' of dataset
'${dataset.getName}'"
+ ) {
+ LakeFSStorageClient.retrieveObjectsOfVersion(repositoryName,
versionHash)
+ }
+
+ if (objects.isEmpty) {
+ throw new NotFoundException(s"No objects found in version
$versionHash")
+ }
+
+ val totalBytes = objects.map(_.getSizeBytes.longValue()).sum
+ // TODO: remove this limit once chunked resumable upload to Drive is
implemented
+ if (totalBytes > DRIVE_EXPORT_MAX_DATASET_BYTES) {
+ throw new BadRequestException("Dataset version exceeds the 500 MB
export limit.")
+ }
+ if (totalBytes > DRIVE_EXPORT_MAX_FILE_BYTES) {
+ throw new ForbiddenException("Dataset version exceeds the 5 TB Google
Drive export limit.")
+ }
+
+ // Build the zip in memory
+ val baos = new ByteArrayOutputStream()
+ val zipOut = new java.util.zip.ZipOutputStream(baos)
+ try {
+ objects.foreach { obj =>
+ val filePath = obj.getPath
+ val file = withLakeFSErrorHandling(s"downloading file '$filePath'
for Drive export") {
+ LakeFSStorageClient.getFileFromRepo(repositoryName, versionHash,
filePath)
+ }
+ zipOut.putNextEntry(new java.util.zip.ZipEntry(filePath))
+ Files.copy(Paths.get(file.toURI), zipOut)
+ zipOut.closeEntry()
+ }
+ } finally {
+ zipOut.close()
+ }
+
+ val zipBytes = baos.toByteArray
+ val sessionUri = request.sessionUri
+
+ // PUT the zip to the Google Drive session URI
+ val conn = new
URL(sessionUri).openConnection().asInstanceOf[HttpURLConnection]
+ try {
+ conn.setDoOutput(true)
+ conn.setRequestMethod("PUT")
+ conn.setRequestProperty("Content-Type", "application/zip")
+ conn.setFixedLengthStreamingMode(zipBytes.length)
+ val out = conn.getOutputStream
+ out.write(zipBytes)
+ out.close()
+
+ val code = conn.getResponseCode
+ if (code < 200 || code >= 300) {
+ val body =
+ Option(conn.getErrorStream).map(s => new
String(s.readAllBytes())).getOrElse("")
+ if (body.contains("storageQuotaExceeded")) {
+ throw new ForbiddenException("Google Drive storage quota
exceeded.")
+ }
+ throw new WebApplicationException(
+ s"Google Drive upload failed with HTTP $code",
+ Response.Status.BAD_GATEWAY
+ )
+ }
+ } finally {
+ conn.disconnect()
+ }
+
+ Response.ok(Map("message" -> "Dataset exported to Google Drive")).build()
+ }
+ }
+
+ @POST
+ @RolesAllowed(Array("REGULAR", "ADMIN"))
+ @Path("/drive-export/file")
+ @Consumes(Array(MediaType.APPLICATION_JSON))
+ def exportFileToDrive(
+ request: DriveFileExportRequest,
+ @Auth user: SessionUser
+ ): Response = {
+ if
(!request.sessionUri.startsWith("https://www.googleapis.com/upload/drive/";)) {
+ throw new BadRequestException("Invalid session URI")
+ }
+ val uid = user.getUid
+ resolveDatasetAndPath(request.filePath, null, null, uid) match {
Review Comment:
resolveDatasetAndPath only checks read access ("download restrictions
handled per endpoint" per its comment), and this endpoint never applies the
owner-or-downloadable check that exportDatasetToDrive does at line 1333. Any
user with read access to a non-downloadable dataset can export its files to
their own Drive.
##########
file-service/src/main/scala/org/apache/texera/service/resource/DatasetResource.scala:
##########
@@ -1361,6 +1305,187 @@ class DatasetResource extends LazyLogging {
}
}
+ @POST
+ @RolesAllowed(Array("REGULAR", "ADMIN"))
+ @Path("/{did}/drive-export")
+ @Consumes(Array(MediaType.APPLICATION_JSON))
+ def exportDatasetToDrive(
+ @PathParam("did") did: Integer,
+ @QueryParam("dvid") dvid: Integer,
+ @QueryParam("latest") latest: java.lang.Boolean,
+ request: DriveExportRequest,
+ @Auth user: SessionUser
+ ): Response = {
+ if
(!request.sessionUri.startsWith("https://www.googleapis.com/upload/drive/";)) {
+ throw new BadRequestException("Invalid session URI")
+ }
+ withTransaction(context) { ctx =>
Review Comment:
The LakeFS downloads, in-memory zip, and Drive PUT all run inside
withTransaction, holding one of the pool's 10 DB connections for the whole
transfer (unbounded, since no read timeout is set). Note getDatasetVersionZip
avoids this: its downloads run via StreamingOutput after the transaction
commits.
##########
common/config/src/main/resources/user-system.conf:
##########
@@ -27,6 +27,12 @@ user-sys {
clientId = ""
clientId = ${?USER_SYS_GOOGLE_CLIENT_ID}
+ clientSecret = ""
+ clientSecret = ${?USER_SYS_GOOGLE_CLIENT_SECRET}
Review Comment:
ClientSecret is added but referenced nowhere in this PR. If the reworked
frontend PR no longer does a server-side exchange, please drop this entry.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
