accept complete http request instead of form to upload file

simao · simao · commit e7ffaf076e24 · 2019-01-17T10:37:10.000+01:00
diff --git a/src/main/scala/com/advancedtelematic/treehub/http/ObjectResource.scala b/src/main/scala/com/advancedtelematic/treehub/http/ObjectResource.scala
@@ -2,7 +2,7 @@ package com.advancedtelematic.treehub.http
 
 import akka.http.scaladsl.model.StatusCodes
 import akka.http.scaladsl.model.headers.Authorization
-import akka.http.scaladsl.server.{Directive0, Directive1, PathMatcher1}
+import akka.http.scaladsl.server.{Directive0, Directive1, PathMatcher1, ValidationRejection}
 import akka.stream.Materializer
 import com.advancedtelematic.data.DataType.ObjectId
 import com.advancedtelematic.libats.data.DataType.Namespace
@@ -11,6 +11,7 @@ import com.advancedtelematic.treehub.repo_metrics.UsageMetricsRouter.{UpdateBand
 import com.advancedtelematic.treehub.repo_metrics.UsageMetricsRouter
 import slick.jdbc.MySQLProfile.api._
 import cats.syntax.either._
+
 import scala.concurrent.ExecutionContext
 import scala.util.Success
 
@@ -56,11 +57,24 @@ class ObjectResource(namespace: Directive1[Namespace],
         complete(f)
       } ~
       (post & hintNamespaceStorage(ns)) {
+        // TODO: Use storeUploadedFile and change ObjectStore api to accept File or DataBytes, when using databytes, require size, stream up
+
+        headerValueByName("x-ats-accept-redirect") { _ =>
+          val f = objectStore.storeOutOfBand(ns, objectId, content).map(_ => StatusCodes.OK)
+          complete(f)
+        } ~
         fileUpload("file") { case (_, content) =>
           val f = objectStore.store(ns, objectId, content).map(_ => StatusCodes.OK)
           complete(f)
-        }
+        } ~
+          extractRequestEntity { entity =>
+            entity.contentLengthOption match {
+              case Some(size) => complete(objectStore.storeStream(ns, objectId, size, entity.dataBytes).map(_ => StatusCodes.OK))
+              case None => reject
+            }
+          }
       }
     }
   }
+
 }
diff --git a/src/main/scala/com/advancedtelematic/treehub/object_store/BlobStore.scala b/src/main/scala/com/advancedtelematic/treehub/object_store/BlobStore.scala
@@ -1,17 +1,28 @@
 package com.advancedtelematic.treehub.object_store
 
-import akka.http.scaladsl.model.{HttpEntity, HttpResponse, MediaTypes, StatusCodes}
+import akka.http.scaladsl.model._
 import akka.stream.scaladsl.Source
 import akka.util.ByteString
 import com.advancedtelematic.data.DataType.ObjectId
 import com.advancedtelematic.libats.data.DataType.Namespace
+import com.advancedtelematic.treehub.object_store.BlobStore.OutOfBandStoreResult
 
 import scala.concurrent.Future
 import scala.util.control.NoStackTrace
 
+object BlobStore {
+  sealed trait OutOfBandStoreResult
+  case class UploadAt(uri: Uri) extends OutOfBandStoreResult
+  case class Failure(cause: Throwable) extends OutOfBandStoreResult
+}
+
 trait BlobStore {
   def store(namespace: Namespace, id: ObjectId, blob: Source[ByteString, _]): Future[Long]
 
+//  def storeStream(namespace: Namespace, id: ObjectId, size: Long, blob: Source[ByteString, _]): Future[Long]
+
+  def storeOutOfBand(namespace: Namespace, id: ObjectId, blob: => Source[ByteString, _]): Future[OutOfBandStoreResult]
+
   def buildResponse(namespace: Namespace, id: ObjectId): Future[HttpResponse]
 
   def readFull(namespace: Namespace, id: ObjectId): Future[ByteString]
diff --git a/src/main/scala/com/advancedtelematic/treehub/object_store/LocalFsBlobStore.scala b/src/main/scala/com/advancedtelematic/treehub/object_store/LocalFsBlobStore.scala
@@ -49,6 +49,9 @@ class LocalFsBlobStore(root: Path)(implicit ec: ExecutionContext, mat: Materiali
     } yield res
   }
 
+  override def storeStream(namespace: Namespace, id: ObjectId, size: Long, blob: Source[ByteString, _]): Future[Long] =
+    store(namespace, id, blob)
+
   override def buildResponse(ns: Namespace, id: ObjectId): Future[HttpResponse] = {
     exists(ns, id).flatMap {
       case true =>
diff --git a/src/main/scala/com/advancedtelematic/treehub/object_store/ObjectStore.scala b/src/main/scala/com/advancedtelematic/treehub/object_store/ObjectStore.scala
@@ -16,16 +16,39 @@ class ObjectStore(blobStore: BlobStore)(implicit ec: ExecutionContext, db: Datab
 
   import scala.async.Async._
 
-  def store(namespace: Namespace, id: ObjectId, blob: Source[ByteString, _]): Future[TObject] = {
-    // TODO: Do we already know size somehow? Sort of...
+  def storeOutOfBand(namespace: Namespace, id: ObjectId, size: Long, blob: Source[ByteString, _]): Future[TObject] = {
+    blobStore.storeStream()
+  }
+
+  // TODO:SM Almost the same as `store`
+  def storeStream(namespace: Namespace, id: ObjectId, size: Long, blob: Source[ByteString, _]): Future[TObject] = {
+    val obj = TObject(namespace, id, size, ObjectStatus.SERVER_UPLOADING)
+    lazy val createF = objectRepository.create(obj)
+
+    lazy val uploadF = async {
+      val _size = await(blobStore.storeStream(namespace, id, size, blob))
+      val newObj = obj.copy(byteSize = _size, status = ObjectStatus.UPLOADED)
+      await(objectRepository.update(namespace, id, _size, ObjectStatus.UPLOADED))
+      newObj
+    }.recoverWith {
+      case e =>
+        objectRepository.delete(namespace, id)
+          .flatMap(_ => FastFuture.failed(e))
+          .recoverWith { case _ => FastFuture.failed(e) }
+    }
 
+    createF.flatMap(_ => uploadF)
+  }
+
+
+  def store(namespace: Namespace, id: ObjectId, blob: Source[ByteString, _]): Future[TObject] = {
     val obj = TObject(namespace, id, -1, ObjectStatus.SERVER_UPLOADING)
     lazy val createF = objectRepository.create(obj)
 
     lazy val uploadF = async {
-      val size = await(blobStore.store(namespace, id, blob))
-      val newObj = obj.copy(byteSize = size, status = ObjectStatus.UPLOADED)
-      await(objectRepository.update(namespace, id, size, ObjectStatus.UPLOADED))
+      val _size = await(blobStore.store(namespace, id, blob))
+      val newObj = obj.copy(byteSize = _size, status = ObjectStatus.UPLOADED)
+      await(objectRepository.update(namespace, id, _size, ObjectStatus.UPLOADED))
       newObj
     }.recoverWith {
       case e =>
diff --git a/src/main/scala/com/advancedtelematic/treehub/object_store/S3BlobStore.scala b/src/main/scala/com/advancedtelematic/treehub/object_store/S3BlobStore.scala
@@ -1,22 +1,23 @@
 package com.advancedtelematic.treehub.object_store
 
-import java.io.File
+import java.io.{File, InputStream}
 import java.nio.file.Paths
 import java.time.{Duration, Instant}
 
 import akka.http.scaladsl.model.headers.Location
 import akka.http.scaladsl.model.{HttpResponse, StatusCodes, Uri}
 import akka.stream.Materializer
-import akka.stream.scaladsl.{FileIO, Source, StreamConverters}
+import akka.stream.scaladsl.{FileIO, Keep, Source, StreamConverters}
 import akka.util.ByteString
+import cats.instances.long
 import cats.syntax.either._
 import com.advancedtelematic.common.DigestCalculator
 import com.advancedtelematic.data.DataType.ObjectId
 import com.advancedtelematic.libats.data.DataType.Namespace
 import com.amazonaws.auth.{AWSCredentials, AWSCredentialsProvider}
 import com.amazonaws.regions.Regions
 import com.amazonaws.services.s3.AmazonS3ClientBuilder
-import com.amazonaws.services.s3.model.{CannedAccessControlList, PutObjectRequest}
+import com.amazonaws.services.s3.model.{CannedAccessControlList, GetObjectRequest, ObjectMetadata, PutObjectRequest}
 import org.slf4j.LoggerFactory
 
 import scala.async.Async._
@@ -36,6 +37,27 @@ class S3BlobStore(s3Credentials: S3Credentials, allowRedirects: Boolean)
       .withRegion(s3Credentials.region)
       .build()
 
+  override def storeStream(namespace: Namespace, id: ObjectId, size: Long, blob: Source[ByteString, _]): Future[Long] = {
+    val filename = objectFilename(namespace, id)
+
+    val sink =  StreamConverters.asInputStream().mapMaterializedValue { is =>
+      val meta = new ObjectMetadata()
+      meta.setContentLength(size)
+      val request = new PutObjectRequest(s3Credentials.blobBucketId, filename, is, meta).withCannedAcl(CannedAccessControlList.AuthenticatedRead)
+
+      log.info(s"Uploading $filename to amazon s3")
+
+      async {
+        await(Future { blocking { s3client.putObject(request) } })
+        log.info(s"$filename with size $size uploaded to s3")
+        size
+      }
+    }
+
+    blob.runWith(sink)
+  }
+
+
   override def store(namespace: Namespace, id: ObjectId, blob: Source[ByteString, _]): Future[Long] = {
     val filename = objectFilename(namespace, id)
     val tempFile = File.createTempFile(filename, ".tmp")
@@ -44,7 +66,7 @@ class S3BlobStore(s3Credentials: S3Credentials, allowRedirects: Boolean)
     // so we always need to cache the file into the filesystem before uploading
     val sink = FileIO.toPath(tempFile.toPath).mapMaterializedValue {
       _.flatMap { result =>
-        if(result.wasSuccessful) {
+        if (result.wasSuccessful) {
           upload(tempFile, filename).andThen { case _ => Try(tempFile.delete()) }
         } else
           Future.failed(result.getError)
diff --git a/src/test/resources/blobs/myfile.bin b/src/test/resources/blobs/myfile.bin
diff --git a/src/test/scala/com/advancedtelematic/treehub/http/ObjectResourceSpec.scala b/src/test/scala/com/advancedtelematic/treehub/http/ObjectResourceSpec.scala
@@ -6,6 +6,8 @@ package com.advancedtelematic.treehub.http
 
 import akka.http.scaladsl.model._
 import akka.pattern.ask
+import akka.stream.scaladsl.Source
+import akka.util.ByteString
 import com.advancedtelematic.treehub.db.ObjectRepositorySupport
 import com.advancedtelematic.util.FakeUsageUpdate.{CurrentBandwith, CurrentStorage}
 import com.advancedtelematic.util.ResourceSpec.ClientTObject
@@ -15,7 +17,7 @@ import scala.concurrent.duration._
 
 class ObjectResourceSpec extends TreeHubSpec with ResourceSpec with ObjectRepositorySupport {
 
-  test("POST creates a new blob") {
+  test("POST creates a new blob when uploading form with `file` field") {
     val obj = new ClientTObject()
 
     Post(apiUri(s"objects/${obj.prefixedObjectId}"), obj.form) ~> routes ~> check {
@@ -29,6 +31,32 @@ class ObjectResourceSpec extends TreeHubSpec with ResourceSpec with ObjectReposi
     }
   }
 
+  // This is the same as using `curl -H "Content-Type: application/octet-stream" --data-binary @file`
+  test("POST creates a new blob when uploading application/octet-stream directory") {
+    val obj = new ClientTObject()
+
+    Post(apiUri(s"objects/${obj.prefixedObjectId}"), obj.blob) ~> routes ~> check {
+      status shouldBe StatusCodes.OK
+    }
+
+    Get(apiUri(s"objects/${obj.prefixedObjectId}")) ~> routes ~> check {
+      status shouldBe StatusCodes.OK
+      responseAs[Array[Byte]] shouldBe obj.blob
+    }
+  }
+
+  test("POST redirects client to redirect endpoint if client supports it") {
+    val obj = new ClientTObject()
+
+    val accept = HttpHeader
+
+
+    Post(apiUri(s"objects/${obj.prefixedObjectId}"), obj.blob) ~> routes ~> check {
+      status shouldBe StatusCodes.OK
+    }
+
+  }
+
   test("POST hints updater to update current storage") {
     val obj = new ClientTObject()
 
diff --git a/src/test/scala/com/advancedtelematic/treehub/object_store/S3BlobStoreIntegrationSpec.scala b/src/test/scala/com/advancedtelematic/treehub/object_store/S3BlobStoreIntegrationSpec.scala
@@ -1,9 +1,11 @@
 package com.advancedtelematic.treehub.object_store
 
+import java.io.File
+
 import akka.actor.ActorSystem
 import akka.http.scaladsl.model.StatusCodes
 import akka.stream.ActorMaterializer
-import akka.stream.scaladsl.Source
+import akka.stream.scaladsl.{FileIO, Source}
 import akka.util.ByteString
 import com.advancedtelematic.data.DataType.{ObjectId, ObjectIdOps, ObjectStatus, TObject}
 import com.advancedtelematic.util.TreeHubSpec
@@ -22,22 +24,24 @@ class S3BlobStoreIntegrationSpec extends TreeHubSpec {
 
   val s3BlobStore = new S3BlobStore(s3Credentials, true)
 
-  override implicit def patienceConfig = PatienceConfig().copy(timeout = Span(3, Seconds))
+  override implicit def patienceConfig = PatienceConfig().copy(timeout = Span(15, Seconds))
 
   test("can store object")  {
     val tobj = TObject(defaultNs, ObjectId.parse("ce720e82a727efa4b30a6ab73cefe31a8d4ec6c0d197d721f07605913d2a279a.commit").toOption.get, 0L, ObjectStatus.UPLOADED)
+    val blob = ByteString("this is byte.")
 
-    val source = Source.single(ByteString("this is byte."))
+    val source = Source.single(blob)
 
     val size = s3BlobStore.store(defaultNs, tobj.id, source).futureValue
 
     size shouldBe 13
   }
 
-  test("can retrieve an object") {
+  test("can retrieve big object") {
     val tobj = TObject(defaultNs, ObjectId.parse("ce720e82a727efa4b30a6ab73cefe31a8d4ec6c0d197d721f07605913d2a279a.commit").toOption.get, 0L, ObjectStatus.UPLOADED)
+    val blob = ByteString("this is byte. Call me. maybe.")
 
-    val source = Source.single(ByteString("this is byte. Call me. maybe."))
+    val source = Source.single(blob)
 
     val f = async {
       await(s3BlobStore.store(defaultNs, tobj.id, source))
@@ -48,10 +52,28 @@ class S3BlobStoreIntegrationSpec extends TreeHubSpec {
     s3BlobStore.exists(tobj.namespace, tobj.id).futureValue shouldBe true
   }
 
+
+  test("XXX can retrieve an object") {
+    val file = new File(this.getClass.getResource(s"/blobs/myfile.bin").getFile)
+    val source = FileIO.fromPath(file.toPath)
+    val tobj = TObject(defaultNs, ObjectId.parse("625e61876cbe98fbf164c8ce5975c6d69a4ba0e9fa57c729ea06d02fd966a9cc.file").toOption.get, 0L, ObjectStatus.UPLOADED)
+
+    val f = async {
+      await(s3BlobStore.store(defaultNs, tobj.id, source))
+
+      println("upload finished")
+
+      await(s3BlobStore.readFull(tobj.namespace, tobj.id))
+    }
+
+    f.futureValue.size shouldBe file.length()
+    s3BlobStore.exists(tobj.namespace, tobj.id).futureValue shouldBe true
+  }
+
   test("build response builds a redirect") {
     val tobj = TObject(defaultNs, ObjectId.parse("ce720e82a727efa4b30a6ab73cefe31a8d4ec6c0d197d721f07605913d2a279a.commit").toOption.get, 0L, ObjectStatus.UPLOADED)
-
-    val source = Source.single(ByteString("this is byte. Call me. maybe."))
+    val blob = ByteString("this is byte. Call me. maybe.")
+    val source = Source.single(blob)
 
     val response = async {
       await(s3BlobStore.store(defaultNs, tobj.id, source))
@@ -63,9 +85,11 @@ class S3BlobStoreIntegrationSpec extends TreeHubSpec {
   }
 
   test("build response a response containing the object content") {
-    val tobj = TObject(defaultNs, ObjectId.parse("ce720e82a727efa4b30a6ab73cefe31a8d4ec6c0d197d721f07605913d2a279a.commit").toOption.get, 0L, ObjectStatus.UPLOADED)
+    pending
 
-    val source = Source.single(ByteString("this is byte. Call me. maybe."))
+    val tobj = TObject(defaultNs, ObjectId.parse("ce720e82a727efa4b30a6ab73cefe31a8d4ec6c0d197d721f07605913d2a279a.commit").toOption.get, 0L, ObjectStatus.UPLOADED)
+    val blob = ByteString("this is byte. Call me. maybe.")
+    val source = Source.single(blob)
 
     val response = async {
       await(s3BlobStore.store(defaultNs, tobj.id, source))
@@ -75,5 +99,4 @@ class S3BlobStoreIntegrationSpec extends TreeHubSpec {
     response.status shouldBe StatusCodes.OK
     response.entity.dataBytes.runFold(ByteString.empty)(_ ++ _).futureValue.utf8String shouldBe "this is byte. Call me. maybe."
   }
-
 }

Original file line number	Diff line number	Diff line change
`@@ -49,6 +49,9 @@ class LocalFsBlobStore(root: Path)(implicit ec: ExecutionContext, mat: Materiali`
`49`	`49`	`} yield res`
`50`	`50`	`}`
`51`	`51`
	`52`	`+ override def storeStream(namespace: Namespace, id: ObjectId, size: Long, blob: Source[ByteString, _]): Future[Long] =`
	`53`	`+ store(namespace, id, blob)`
	`54`	`+`
`52`	`55`	`override def buildResponse(ns: Namespace, id: ObjectId): Future[HttpResponse] = {`
`53`	`56`	`exists(ns, id).flatMap {`
`54`	`57`	`case true =>`