Merge pull request #27034 from nextcloud/enh/s3-multipart-upload-api

Use MultipartUpload for uploading chunks to s3

6 files changed, 266 insertions, 3 deletions

diff --git a/lib/composer/composer/autoload_classmap.php b/lib/composer/composer/autoload_classmap.php
index 57a828dbefb..ee6117f9b73 100644
--- a/lib/composer/composer/autoload_classmap.php
+++ b/lib/composer/composer/autoload_classmap.php
@@ -331,6 +331,7 @@ return array(
     'OCP\\Files\\Notify\\INotifyHandler' => $baseDir . '/lib/public/Files/Notify/INotifyHandler.php',
     'OCP\\Files\\Notify\\IRenameChange' => $baseDir . '/lib/public/Files/Notify/IRenameChange.php',
     'OCP\\Files\\ObjectStore\\IObjectStore' => $baseDir . '/lib/public/Files/ObjectStore/IObjectStore.php',
+    'OCP\\Files\\ObjectStore\\IObjectStoreMultiPartUpload' => $baseDir . '/lib/public/Files/ObjectStore/IObjectStoreMultiPartUpload.php',
     'OCP\\Files\\ReservedWordException' => $baseDir . '/lib/public/Files/ReservedWordException.php',
     'OCP\\Files\\Search\\ISearchBinaryOperator' => $baseDir . '/lib/public/Files/Search/ISearchBinaryOperator.php',
     'OCP\\Files\\Search\\ISearchComparison' => $baseDir . '/lib/public/Files/Search/ISearchComparison.php',
@@ -348,6 +349,7 @@ return array(
     'OCP\\Files\\StorageInvalidException' => $baseDir . '/lib/public/Files/StorageInvalidException.php',
     'OCP\\Files\\StorageNotAvailableException' => $baseDir . '/lib/public/Files/StorageNotAvailableException.php',
     'OCP\\Files\\StorageTimeoutException' => $baseDir . '/lib/public/Files/StorageTimeoutException.php',
+    'OCP\\Files\\Storage\\IChunkedFileWrite' => $baseDir . '/lib/public/Files/Storage/IChunkedFileWrite.php',
     'OCP\\Files\\Storage\\IDisableEncryptionStorage' => $baseDir . '/lib/public/Files/Storage/IDisableEncryptionStorage.php',
     'OCP\\Files\\Storage\\ILockingStorage' => $baseDir . '/lib/public/Files/Storage/ILockingStorage.php',
     'OCP\\Files\\Storage\\INotifyStorage' => $baseDir . '/lib/public/Files/Storage/INotifyStorage.php',
diff --git a/lib/composer/composer/autoload_static.php b/lib/composer/composer/autoload_static.php
index e9d1ba50024..253b9ddbada 100644
--- a/lib/composer/composer/autoload_static.php
+++ b/lib/composer/composer/autoload_static.php
@@ -364,6 +364,7 @@ class ComposerStaticInit749170dad3f5e7f9ca158f5a9f04f6a2
         'OCP\\Files\\Notify\\INotifyHandler' => __DIR__ . '/../../..' . '/lib/public/Files/Notify/INotifyHandler.php',
         'OCP\\Files\\Notify\\IRenameChange' => __DIR__ . '/../../..' . '/lib/public/Files/Notify/IRenameChange.php',
         'OCP\\Files\\ObjectStore\\IObjectStore' => __DIR__ . '/../../..' . '/lib/public/Files/ObjectStore/IObjectStore.php',
+        'OCP\\Files\\ObjectStore\\IObjectStoreMultiPartUpload' => __DIR__ . '/../../..' . '/lib/public/Files/ObjectStore/IObjectStoreMultiPartUpload.php',
         'OCP\\Files\\ReservedWordException' => __DIR__ . '/../../..' . '/lib/public/Files/ReservedWordException.php',
         'OCP\\Files\\Search\\ISearchBinaryOperator' => __DIR__ . '/../../..' . '/lib/public/Files/Search/ISearchBinaryOperator.php',
         'OCP\\Files\\Search\\ISearchComparison' => __DIR__ . '/../../..' . '/lib/public/Files/Search/ISearchComparison.php',
@@ -381,6 +382,7 @@ class ComposerStaticInit749170dad3f5e7f9ca158f5a9f04f6a2
         'OCP\\Files\\StorageInvalidException' => __DIR__ . '/../../..' . '/lib/public/Files/StorageInvalidException.php',
         'OCP\\Files\\StorageNotAvailableException' => __DIR__ . '/../../..' . '/lib/public/Files/StorageNotAvailableException.php',
         'OCP\\Files\\StorageTimeoutException' => __DIR__ . '/../../..' . '/lib/public/Files/StorageTimeoutException.php',
+        'OCP\\Files\\Storage\\IChunkedFileWrite' => __DIR__ . '/../../..' . '/lib/public/Files/Storage/IChunkedFileWrite.php',
         'OCP\\Files\\Storage\\IDisableEncryptionStorage' => __DIR__ . '/../../..' . '/lib/public/Files/Storage/IDisableEncryptionStorage.php',
         'OCP\\Files\\Storage\\ILockingStorage' => __DIR__ . '/../../..' . '/lib/public/Files/Storage/ILockingStorage.php',
         'OCP\\Files\\Storage\\INotifyStorage' => __DIR__ . '/../../..' . '/lib/public/Files/Storage/INotifyStorage.php',
diff --git a/lib/private/Files/ObjectStore/ObjectStoreStorage.php b/lib/private/Files/ObjectStore/ObjectStoreStorage.php
index d0c5bd14b38..4ca00cf6a16 100644
--- a/lib/private/Files/ObjectStore/ObjectStoreStorage.php
+++ b/lib/private/Files/ObjectStore/ObjectStoreStorage.php
@@ -29,6 +29,8 @@
  */
 namespace OC\Files\ObjectStore;
 
+use Aws\S3\Exception\S3Exception;
+use Aws\S3\Exception\S3MultipartUploadException;
 use Icewind\Streams\CallbackWrapper;
 use Icewind\Streams\CountWrapper;
 use Icewind\Streams\IteratorDirectory;
@@ -37,11 +39,14 @@ use OC\Files\Cache\CacheEntry;
 use OC\Files\Storage\PolyFill\CopyDirectory;
 use OCP\Files\Cache\ICacheEntry;
 use OCP\Files\FileInfo;
+use OCP\Files\GenericFileException;
 use OCP\Files\NotFoundException;
 use OCP\Files\ObjectStore\IObjectStore;
+use OCP\Files\ObjectStore\IObjectStoreMultiPartUpload;
+use OCP\Files\Storage\IChunkedFileWrite;
 use OCP\Files\Storage\IStorage;
 
-class ObjectStoreStorage extends \OC\Files\Storage\Common {
+class ObjectStoreStorage extends \OC\Files\Storage\Common implements IChunkedFileWrite {
 	use CopyDirectory;
 
 	/**
@@ -91,7 +96,6 @@ class ObjectStoreStorage extends \OC\Files\Storage\Common {
 
 	public function mkdir($path) {
 		$path = $this->normalizePath($path);
-
 		if ($this->file_exists($path)) {
 			return false;
 		}
@@ -627,4 +631,72 @@ class ObjectStoreStorage extends \OC\Files\Storage\Common {
 			throw $e;
 		}
 	}
+
+	public function startChunkedWrite(string $targetPath): string {
+		if (!$this->objectStore instanceof IObjectStoreMultiPartUpload) {
+			throw new GenericFileException('Object store does not support multipart upload');
+		}
+		$cacheEntry = $this->getCache()->get($targetPath);
+		$urn = $this->getURN($cacheEntry->getId());
+		return $this->objectStore->initiateMultipartUpload($urn);
+	}
+
+	/**
+	 *
+	 * @throws GenericFileException
+	 */
+	public function putChunkedWritePart(string $targetPath, string $writeToken, string $chunkId, $data, $size = null): ?array {
+		if (!$this->objectStore instanceof IObjectStoreMultiPartUpload) {
+			throw new GenericFileException('Object store does not support multipart upload');
+		}
+		$cacheEntry = $this->getCache()->get($targetPath);
+		$urn = $this->getURN($cacheEntry->getId());
+
+		$result = $this->objectStore->uploadMultipartPart($urn, $writeToken, (int)$chunkId, $data, $size);
+
+		$parts[$chunkId] = [
+			'PartNumber' => $chunkId,
+			'ETag' => trim($result->get('ETag'), '"')
+		];
+		return $parts[$chunkId];
+	}
+
+	public function completeChunkedWrite(string $targetPath, string $writeToken): int {
+		if (!$this->objectStore instanceof IObjectStoreMultiPartUpload) {
+			throw new GenericFileException('Object store does not support multipart upload');
+		}
+		$cacheEntry = $this->getCache()->get($targetPath);
+		$urn = $this->getURN($cacheEntry->getId());
+		$parts = $this->objectStore->getMultipartUploads($urn, $writeToken);
+		$sortedParts = array_values($parts);
+		sort($sortedParts);
+		try {
+			$size = $this->objectStore->completeMultipartUpload($urn, $writeToken, $sortedParts);
+			$stat = $this->stat($targetPath);
+			$mtime = time();
+			if (is_array($stat)) {
+				$stat['size'] = $size;
+				$stat['mtime'] = $mtime;
+				$stat['mimetype'] = $this->getMimeType($targetPath);
+				$this->getCache()->update($stat['fileid'], $stat);
+			}
+		} catch (S3MultipartUploadException | S3Exception $e) {
+			$this->objectStore->abortMultipartUpload($urn, $writeToken);
+			$this->logger->logException($e, [
+				'app' => 'objectstore',
+				'message' => 'Could not compete multipart upload ' . $urn. ' with uploadId ' . $writeToken
+			]);
+			throw new GenericFileException('Could not write chunked file');
+		}
+		return $size;
+	}
+
+	public function cancelChunkedWrite(string $targetPath, string $writeToken): void {
+		if (!$this->objectStore instanceof IObjectStoreMultiPartUpload) {
+			throw new GenericFileException('Object store does not support multipart upload');
+		}
+		$cacheEntry = $this->getCache()->get($targetPath);
+		$urn = $this->getURN($cacheEntry->getId());
+		$this->objectStore->abortMultipartUpload($urn, $writeToken);
+	}
 }
diff --git a/lib/private/Files/ObjectStore/S3.php b/lib/private/Files/ObjectStore/S3.php
index 6492145fb63..ebc8886f12d 100644
--- a/lib/private/Files/ObjectStore/S3.php
+++ b/lib/private/Files/ObjectStore/S3.php
@@ -23,9 +23,12 @@
  */
 namespace OC\Files\ObjectStore;
 
+use Aws\Result;
+use Exception;
 use OCP\Files\ObjectStore\IObjectStore;
+use OCP\Files\ObjectStore\IObjectStoreMultiPartUpload;
 
-class S3 implements IObjectStore {
+class S3 implements IObjectStore, IObjectStoreMultiPartUpload {
 	use S3ConnectionTrait;
 	use S3ObjectTrait;
 
@@ -41,4 +44,59 @@ class S3 implements IObjectStore {
 	public function getStorageId() {
 		return $this->id;
 	}
+
+	public function initiateMultipartUpload(string $urn): string {
+		$upload = $this->getConnection()->createMultipartUpload([
+			'Bucket' => $this->bucket,
+			'Key' => $urn,
+		]);
+		$uploadId = $upload->get('UploadId');
+		if ($uploadId === null) {
+			throw new Exception('No upload id returned');
+		}
+		return (string)$uploadId;
+	}
+
+	public function uploadMultipartPart(string $urn, string $uploadId, int $partId, $stream, $size): Result {
+		return $this->getConnection()->uploadPart([
+			'Body' => $stream,
+			'Bucket' => $this->bucket,
+			'Key' => $urn,
+			'ContentLength' => $size,
+			'PartNumber' => $partId,
+			'UploadId' => $uploadId,
+		]);
+	}
+
+	public function getMultipartUploads(string $urn, string $uploadId): array {
+		$parts = $this->getConnection()->listParts([
+			'Bucket' => $this->bucket,
+			'Key' => $urn,
+			'UploadId' => $uploadId,
+			'MaxParts' => 10000
+		]);
+		return $parts->get('Parts') ?? [];
+	}
+
+	public function completeMultipartUpload(string $urn, string $uploadId, array $result): int {
+		$this->getConnection()->completeMultipartUpload([
+			'Bucket' => $this->bucket,
+			'Key' => $urn,
+			'UploadId' => $uploadId,
+			'MultipartUpload' => ['Parts' => $result],
+		]);
+		$stat = $this->getConnection()->headObject([
+			'Bucket' => $this->bucket,
+			'Key' => $urn,
+		]);
+		return (int)$stat->get('ContentLength');
+	}
+
+	public function abortMultipartUpload($urn, $uploadId): void {
+		$this->getConnection()->abortMultipartUpload([
+			'Bucket' => $this->bucket,
+			'Key' => $urn,
+			'UploadId' => $uploadId
+		]);
+	}
 }
diff --git a/lib/public/Files/ObjectStore/IObjectStoreMultiPartUpload.php b/lib/public/Files/ObjectStore/IObjectStoreMultiPartUpload.php
new file mode 100644
index 00000000000..f46982f3112
--- /dev/null
+++ b/lib/public/Files/ObjectStore/IObjectStoreMultiPartUpload.php
@@ -0,0 +1,59 @@
+<?php
+/*
+ * @copyright Copyright (c) 2021 Julius Härtl <jus@bitgrid.net>
+ *
+ * @author Julius Härtl <jus@bitgrid.net>
+ *
+ * @license GNU AGPL version 3 or any later version
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ *
+ */
+
+declare(strict_types=1);
+
+
+namespace OCP\Files\ObjectStore;
+
+use Aws\Result;
+
+/**
+ * @since 26.0.0
+ */
+interface IObjectStoreMultiPartUpload {
+	/**
+	 * @since 26.0.0
+	 */
+	public function initiateMultipartUpload(string $urn): string;
+
+	/**
+	 * @since 26.0.0
+	 */
+	public function uploadMultipartPart(string $urn, string $uploadId, int $partId, $stream, $size): Result;
+
+	/**
+	 * @since 26.0.0
+	 */
+	public function completeMultipartUpload(string $urn, string $uploadId, array $result): int;
+
+	/**
+	 * @since 26.0.0
+	 */
+	public function abortMultipartUpload(string $urn, string $uploadId): void;
+
+	/**
+	 * @since 26.0.0
+	 */
+	public function getMultipartUploads(string $urn, string $uploadId): array;
+}
diff --git a/lib/public/Files/Storage/IChunkedFileWrite.php b/lib/public/Files/Storage/IChunkedFileWrite.php
new file mode 100644
index 00000000000..01f5cbbb20a
--- /dev/null
+++ b/lib/public/Files/Storage/IChunkedFileWrite.php
@@ -0,0 +1,70 @@
+<?php
+/*
+ * @copyright Copyright (c) 2021 Julius Härtl <jus@bitgrid.net>
+ *
+ * @author Julius Härtl <jus@bitgrid.net>
+ *
+ * @license GNU AGPL version 3 or any later version
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program. If not, see <http://www.gnu.org/licenses/>.
+ *
+ */
+
+declare(strict_types=1);
+
+
+namespace OCP\Files\Storage;
+
+use OCP\Files\GenericFileException;
+
+/**
+ * @since 26.0.0
+ */
+interface IChunkedFileWrite extends IStorage {
+	/**
+	 * @param string $targetPath Relative target path in the storage
+	 * @return string writeToken to be used with the other methods to uniquely identify the file write operation
+	 * @throws GenericFileException
+	 * @since 26.0.0
+	 */
+	public function startChunkedWrite(string $targetPath): string;
+
+	/**
+	 * @param string $targetPath
+	 * @param string $writeToken
+	 * @param string $chunkId
+	 * @param resource $data
+	 * @param int|null $size
+	 * @throws GenericFileException
+	 * @since 26.0.0
+	 */
+	public function putChunkedWritePart(string $targetPath, string $writeToken, string $chunkId, $data, int $size = null): ?array;
+
+	/**
+	 * @param string $targetPath
+	 * @param string $writeToken
+	 * @return int
+	 * @throws GenericFileException
+	 * @since 26.0.0
+	 */
+	public function completeChunkedWrite(string $targetPath, string $writeToken): int;
+
+	/**
+	 * @param string $targetPath
+	 * @param string $writeToken
+	 * @throws GenericFileException
+	 * @since 26.0.0
+	 */
+	public function cancelChunkedWrite(string $targetPath, string $writeToken): void;
+}