Added list bucket method to S3 library. #253

Scalahansolo · 2017-04-06T23:07:52Z

Adds the ability to get a source of all keys under a given prefix.

…he Query type

…s they are already retried on the underlying layer with max-retries

Scalahansolo · 2017-04-07T16:42:55Z

When running sbt test:compile locally, I get successful runs. Im not sure what exactly is going wrong in the Travis build.

cheleb · 2017-04-07T16:51:22Z

IMHO there it is an "external" issue, I have *sometime* the same error ! Le ven. 7 avr. 2017 à 18:42, Sean Callahan <[email protected]> a écrit : When running sbt test:compile locally, I get successful runs. Im not sure what exactly is going wrong in the Travis build. — You are receiving this because you are subscribed to this thread. Reply to this email directly, view it on GitHub <#253 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AAXGq-CQrAbVrjGntrHzy2ptLLmtZawvks5rtmeQgaJpZM4M2RmP> .

jrudolph · 2017-04-20T15:01:46Z

I tried to resolve the conflicts (hopefully correctly...).

Scalahansolo · 2017-04-20T15:11:00Z

All still looks good at first glance.

jrudolph

I added a few comments. Two things should be fixed:

the recursion in fileSourceFromFuture
and the potential double encoding listBucket

Have you tested that it works?

jrudolph · 2017-04-20T15:03:08Z

s3/src/main/scala/akka/stream/alpakka/s3/impl/HttpRequests.scala

-        case None => uri.withScheme("https")
-        case Some(proxy) => uri.withPort(proxy.port).withScheme(proxy.scheme)
-      }
+  private[this] def requestHost(bucket: String, region: String)(implicit conf: S3Settings): Uri.Host =


Good idea, to put extract that method.

jrudolph · 2017-04-20T15:04:24Z

s3/src/main/scala/akka/stream/alpakka/s3/impl/S3Stream.scala

@@ -43,6 +41,8 @@ final case class FailedUpload(reasons: Seq[Throwable]) extends Exception

 final case class CompleteMultipartUploadResult(location: Uri, bucket: String, key: String, etag: String)

+final case class ListBucketResult(is_truncated: Boolean, continuation_token: Option[String], keys: Seq[String])


Maybe we should keep the camel case style for parameters?

jrudolph · 2017-04-20T15:08:23Z

s3/src/main/scala/akka/stream/alpakka/s3/impl/HttpRequests.scala

+      continuation_token: Option[String] = None
+  )(implicit conf: S3Settings): HttpRequest = {
+
+    val listTypeQuery: (String, String) = "list-type" -> "2"


I guess this implementation is fine as is.

It could probably be simplified by collecting all parameters as a Seq[(String, Option[String]) and then converting that to a Map, filtering out the None values.

Something like

Seq("list-type" -> Some("2"), "prefix" -> prefix, "continuation-token" -> continuation_token.map(_.replace...) ).collect { case (k, Some(v)) => k -> v }.toMap

Ya, this is a little nicer. Still leveling up on my use of the collections in scala and using them properly.

jrudolph · 2017-04-20T15:11:50Z

s3/src/main/scala/akka/stream/alpakka/s3/impl/HttpRequests.scala

+    val listTypeQuery: (String, String) = "list-type" -> "2"
+    val prefixQuery: Option[(String, String)] = prefix.map("prefix" -> _)
+    val continuationTokenQuery: Option[(String, String)] =
+      continuation_token.map("continuation-token" -> _.replaceAll("=", "%3D"))


Why does only this character need to be replaced? Are you use that's correct at all, because Uri.query will also encode characters, so that the percent character may then be double encoded?

jrudolph · 2017-04-20T15:13:13Z

s3/src/main/scala/akka/stream/alpakka/s3/impl/HttpRequests.scala

+      bucket: String,
+      region: String,
+      prefix: Option[String] = None,
+      continuation_token: Option[String] = None


Let's use camel case there as well.

jrudolph · 2017-04-20T15:35:40Z

s3/src/main/scala/akka/stream/alpakka/s3/impl/S3Stream.scala

+    def fileSourceFromFuture(f: Future[ListBucketResult]): Source[String, NotUsed] =
+      Source
+        .fromFuture(f)
+        .flatMapConcat((res: ListBucketResult) => {


The recursive calls here look like a nice pattern but it doesn't run in constant memory as expected for streams and may lead to OOM eventually. See the comment akka/akka#22352 (comment) here for a potential workaround.

Interesting. Ill dig in and try to get something working.

After digging a little here, Im not quite sure how you use this technique to page through the results properly. I dont how statefulMapConcat gives you access to the previous value than continues to page until completed.

In the meantime I figured that using Source.unfoldAsync would probably even simpler. It allows you to thread state through between each call. So, it should basically look similar to this:

def getNextBucket(continuationToken: Option[String]): Future[Option[(String, Seq[...])]] // needs to return a future of a tuple `Some(next_continuation_token, sequence of results)` (or `None` if complete). Source.unfoldAsync(None)(getNextBucket) .mapConcat(identity) // flatten sequence of results

Can you try that?

jrudolph · 2017-04-20T15:37:03Z

s3/src/main/scala/akka/stream/alpakka/s3/impl/HttpRequests.scala

 import scala.collection.immutable.Seq
 import scala.concurrent.{ExecutionContext, Future}

 private[alpakka] object HttpRequests {

+  def listBucket(


I wonder if it wouldn't make sense to use separate methods for the initial call and the continuation. It seems that you either would want to add a prefix or a token both never both?

jrudolph · 2017-04-20T15:38:38Z

s3/src/main/scala/akka/stream/alpakka/s3/impl/S3Stream.scala

+    def listBucketCall(continuation_token: Option[String] = None): Future[ListBucketResult] =
+      signAndGetAs[ListBucketResult](HttpRequests.listBucket(bucket, region, prefix, continuation_token))
+
+    def fileSourceFromFuture(f: Future[ListBucketResult]): Source[String, NotUsed] =


Why is it called fileSourceFromFuture? ;)

Not the best name now that you mention it. Hah. Got any better ideas?

Scalahansolo · 2017-04-20T16:31:32Z

Also, for what its worth, I have been using a forked version of the repo for a while now using this method with success.

jrudolph · 2017-05-08T11:00:20Z

The increased memory usage will probably only show for particular use cases. @Scalahansolo, could you try unfoldAsync? It might even simplify the code. Otherwise, I think this only needs the smallish fixups I commented last time.

…get keys and used unfoldAsync to get all keys to run in constant memory.

…s3-listBucket

Scalahansolo · 2017-05-08T16:06:09Z

The unfoldAsync ended up working out. Thats a really cool API method that I was unaware of. The type on the listBucket call ended up being pretty brutal, but the logic seems sound.

jrudolph · 2017-05-08T16:28:49Z

s3/src/main/scala/akka/stream/alpakka/s3/impl/S3Stream.scala

+     *             and the first call Boolean is false, we know that we are done getting keys.
+     * @return
+     */
+    def listBucketCall(args: (Option[String], Boolean)): Future[Option[((Option[String], Boolean), Seq[String])]] =


Ah, I see, you need to distinguish "Start", "Running", and "End" states. Maybe it would make sense to create a three state ADT instead? That would read a bit cleaner:

sealed trait State case object Starting extends State case class Running(continuationTokenL String) extends State case object Finished extends State

WDYT?

Ya, that would be a little cleaner for sure. Ill see what I can come up. Will update this soon.

…f the brutal type signature from earlier.

Scalahansolo · 2017-05-08T16:56:11Z

I think that you'll like this a bit better. I definitely feel this is cleaner and much easier to read.

jrudolph · 2017-05-09T12:34:53Z

s3/src/main/scala/akka/stream/alpakka/s3/impl/S3Stream.scala

+        }
+
+    Source
+      .unfoldAsync[ListBucketState, Seq[String]](Starting) {


Yes, this is great now.

jrudolph · 2017-05-09T12:36:00Z

s3/src/main/scala/akka/stream/alpakka/s3/impl/S3Stream.scala

@@ -43,6 +41,13 @@ final case class FailedUpload(reasons: Seq[Throwable]) extends Exception

 final case class CompleteMultipartUploadResult(location: Uri, bucket: String, key: String, etag: String)

+final case class ListBucketResult(isTruncated: Boolean, continuationToken: Option[String], keys: Seq[String])
+
+sealed trait ListBucketState


Can you move those into a private place so that it doesn't clutter the public namespace? Would be ok from my side to but the whole hierarchy into the listBucket method if it's only used there.

jrudolph

Almost done. Can be merged from my side after the internal state hierarchy classes are moved into a more private place.

…he public namespace.

jrudolph

LGTM

jrudolph · 2017-05-10T08:41:42Z

Great one, thanks a lot @Scalahansolo.

@johanandren

- Set default settings as default parameter - Improve tests Implement java dsl [FTP] Critical fix for infinite loop of traversing "." and ".." directories Upgrade to aws-java-sdk-dynamodb 1.11.106 Allow to pass a `SSLSocketFactory` to `MqttConnectionSettings` FTP - attribute enrichment of FTPFile akka#153 Add KairosDB connector = akka#135 Limit parallelism for the SqsSource (akka#163) * Create a custom thread pool for the SqsSource and limit concurrency with the buffer size * Provide AWSCredentials for all SqsSource tests * Provide better java api * Use ArrayDeque as FIFO queue Messages for asserts in SqsSourceSettings * Remove AmazonSQSAsyncClient from factory method - Correct the javadoc * Add implicit AmazonSQSAsync to factory method - It now conforms again with the SqsSink - Passing the clients from extern seems the preferred way (awslambda module) * Update documentation to reflect thread pool usage * Improvements from review by ktoso - Replace IntStream with Source.range - Use Sink.head instead of Sink.seq =pro update akka http to 10.0.5 (akka#230) Make Travis fail build on code format differences =sqs fix typo in require in SqsSourceSettings (akka#228) FTP - toPath sink akka#182 Add GCE Pubsub with publish and subscribe. improve naming consistency of private vars. PR feedback. Improve java api, java examples, better json marshalling. use mapAsyncUnordered s3: provide access to the returned response s3: clean up test log s3: add javadsl for request() Time goes by, next try Update connectors.md (akka#237) Make external libs more visible (reactive kafka) (akka#229) * Update TOC depth to 3, to show Reactive Kafka Now people don't notice Kafka is in here since it's "external", expanding the TOC one more level makes it more visible. WDYT? * Update index.md FTP: make lastModified test more robust (fixes akka#236) Add SqsAckSink (akka#129) Add SqsAckSink * update elasticmq version * update dependencies Added possibility configure Sftp connection using private key akka#197. - Added SftpIdentity case class to allow configuring private/public key, - Added option to configure known_hosts file and test to check its usage. - Added spec that should fail password based authentication and revert to private key one, - Added docs paragraph to describe this option. Upgrade to scalafmt 0.6.6 Remove deperecated binPack.callSite scalafmt setting Format with updated scalafmt and fixed settings S3 - add documentation akka#103 Fix alphabetical ordering in the docs Separate out release docs S3 path style access akka#64 Make SqsSourceTest less likely to fail - Reduce amount of sent messages to 1 (multiple batch streaming is tested in the SqsSourceSpec) - Increase timeout Introduced "secure" boolean property for S3 which controls whether HTTPS is used akka#247 README: add scaladex, travis badges And make docs links less scary to click on :) Add CSV data transformation module (akka#213) * Alpakka Issue akka#66: CSV component * Alpakka Issue akka#66: revised CSV parser * Alpakka Issue akka#60: CSV parsing stage * wait for line end before issuing line As the byte string may not contain a whole line the parser needs to read until a line end is reached. * Add Java API and JUnit test; add a bit of documentation * Introduce CsvToMap stage; more documentation * Parse line even without line end at upstream finish * Add Java API for CsvToMap; more documentation * More restricted API, incorporated comments by @johanandren * Format sequence as CSV in ByteString * Add Scala CSV formatting stage * Add Java API for CSV formatting; more docs * Separate enums for Java and Scala DSLs * Use Flow.fromGraph to construct flow * Rename CsvFraming to CsvParsing * Check for Byte Order Mark and ignore it for UTF-8 * Emit Byte Order Marks in formatting; CsvFormatting is just a map * Byte Order Mark for Java API * Add line number to error messages; sample files exported from third party software * Use Charset directly instead of name * csv: autoformatted files * simplified dependency declaration Fixes akka#60. SQS flows + Embedded ElasticMQ akka#255 * Add a flow stage and use ElasticMQ * Use flow-based stage for ACKs * Use AmazonSQSAsync instead of AmazonSQSAsyncClient * Using embedded ElasticMQ for tests add SNS connector with publish sink akka#204 Await futures before performing assertion (fixes akka#235) When an assertion fails after the test has already succeeded it will be ignored, so Await the future before continuing with the check. Document 'docker-compose' for running tests Fail MqttSourceStage mat. value on connection loss And increase the timeout. Might help with akka#189, or otherwise help generate a better error message when it does happen again. Ref akka#2 add IronMq integration Refs akka#2 add at-least-one semantic to IronMq connector Improve documentation and test coverage for IronMq ref akka#2 - Document the IronMq domain classes - Document IronMq client - Test the at-least-once producer/consumer mechanism - Improve the IronMQ connector documentation Ref akka#2 Preserve newline in reference.conf Ref akka#2 Make seure the actor system is fully terminated after each test Ref akka#2 Reformat code Refs akka#2 define a different Committable and CommittableMessage for Java and Scala DSL Refs akka#2 Fix typos in IronMQ documentations Refs akka#2 Remove non needed Environment variables from TravisCI config file Refs akka#2 Add a simple Java test and refactor the Java DSL to looks better in Java FTP: Attempt to fix flaky test on Travis Link to scaladex (akka#266) s3: support for encryption, storage class, custom headers akka#109 s3: Added support for partial file download from S3 akka#264 (akka#265) Add version info and links in index page (akka#273) FTP - append mode for toPath sink + improved upstream failure handling akka#207 Fix broken recovery of EventSource (sse) Replace scala.binaryVersion with scalaBinaryVersion (see akka#278) Fix minor typo in alpakka MQTT Connector doc Add Flow to support RabbitMQ RPC workflow akka#160 Changes Amqp sinks to materialize to Future[Done]. As currently it was very difficult to determine when/if a sink failed due to a amqp error. AMQP: add more options to configuration of the ConnectionFactory, akka#191 Directory sources akka#272 sse: Upgrade to Akka SSE 3 and make test more robust CSV: Fixes ignored second double quote S3: add listBucket method to S3 library (akka#253) * Added recursive listBucket call to get all keys under a specific prefix. * Properly using the request URI method and constructing queries with the Query type * Added tests around query parsing * Fixed formatting and removed recoverWithRetries on listbucket calls as they are already retried on the underlying layer with max-retries * Using signAndGetAs instead of signAndGet as to not duplicate logic. * Implemented quick fixes based on comments. Removed recursive call to get keys and used unfoldAsync to get all keys to run in constant memory. * Added execution context. Fixed broken test * Fixed formatting error. * Cleaned up lisBucket call by added a ListBucketState object instead of the brutal type signature from earlier. * Moved trait for listBucket into the def itself as to remove it from the public namespace. azure-storage-queue connector akka#280 Add attribute parameters to sqs source settings akka#302 Formatting fix for akka#302 Streaming XML parser and utilities. Prepare XML parser to join Alpakka family Remove duplicated region argument in client methods akka#297 Build with Akka 2.5 as well Add Azure Storage Queue documentation to TOC Stub documentation for S3.listBucket S3: fix formatting Run the deployment only against Akka 2.4 PubSub: Add support for emulator host variables Initial commit for apache geode connector CSV: Emit all lines on completion akka#315 XML: make code in tests more consistent Add whitesource plugin Merge branch 'master' into add-kairosdb-connector Add copyright header update docker-compose Make execution context optional in java api Make execution context optional in scala api remove ec from sink spec

Sean Callahan added 4 commits April 6, 2017 16:10

Added recursive listBucket call to get all keys under a specific prefix.

0802ce2

Properly using the request URI method and constructing queries with t…

eca8fd8

…he Query type

Added tests around query parsing

eafcbc1

Fixed formatting and removed recoverWithRetries on listbucket calls a…

0e91955

…s they are already retried on the underlying layer with max-retries

Using signAndGetAs instead of signAndGet as to not duplicate logic.

61a27bf

patriknw requested a review from jrudolph April 18, 2017 15:12

patriknw assigned jrudolph Apr 18, 2017

Merge branch 'master' into s3-listBucket

a337d3c

jrudolph reviewed Apr 20, 2017

View reviewed changes

Sean Callahan added 3 commits May 8, 2017 09:47

Implemented quick fixes based on comments. Removed recursive call to …

4d4a92e

…get keys and used unfoldAsync to get all keys to run in constant memory.

Merge branch 's3-listBucket' of github.com:Scalahansolo/alpakka into …

9b3d4e6

…s3-listBucket

Added execution context. Fixed broken test

5d50d73

Fixed formatting error.

414faee

jrudolph reviewed May 8, 2017

View reviewed changes

Cleaned up lisBucket call by added a ListBucketState object instead o…

515ee41

…f the brutal type signature from earlier.

jrudolph reviewed May 9, 2017

View reviewed changes

Moved trait for listBucket into the def itself as to remove it from t…

a634d2b

…he public namespace.

jrudolph approved these changes May 10, 2017

View reviewed changes

jrudolph merged commit 7f270bb into akka:master May 10, 2017

jrudolph added this to the 0.9 milestone May 10, 2017

jotsif mentioned this pull request May 22, 2017

Limit memory use jotsif/akkastreamgcs#2

Open

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Added list bucket method to S3 library. #253

Added list bucket method to S3 library. #253

Scalahansolo commented Apr 6, 2017

Scalahansolo commented Apr 7, 2017

cheleb commented Apr 7, 2017 via email

jrudolph commented Apr 20, 2017

Scalahansolo commented Apr 20, 2017

jrudolph left a comment

jrudolph Apr 20, 2017

jrudolph Apr 20, 2017

jrudolph Apr 20, 2017

Scalahansolo Apr 20, 2017

jrudolph Apr 20, 2017

jrudolph Apr 20, 2017

Scalahansolo Apr 20, 2017

jrudolph Apr 20, 2017

Scalahansolo Apr 20, 2017

Scalahansolo Apr 24, 2017

jrudolph May 8, 2017

jrudolph Apr 20, 2017

jrudolph Apr 20, 2017

Scalahansolo Apr 20, 2017

Scalahansolo commented Apr 20, 2017

jrudolph commented May 8, 2017

Scalahansolo commented May 8, 2017

jrudolph May 8, 2017

Scalahansolo May 8, 2017

Scalahansolo commented May 8, 2017

jrudolph May 9, 2017

jrudolph May 9, 2017

Scalahansolo May 9, 2017

jrudolph left a comment

jrudolph left a comment

jrudolph commented May 10, 2017

		@@ -43,6 +41,8 @@ final case class FailedUpload(reasons: Seq[Throwable]) extends Exception

		final case class CompleteMultipartUploadResult(location: Uri, bucket: String, key: String, etag: String)

		final case class ListBucketResult(is_truncated: Boolean, continuation_token: Option[String], keys: Seq[String])

Added list bucket method to S3 library. #253

Added list bucket method to S3 library. #253

Conversation

Scalahansolo commented Apr 6, 2017

Scalahansolo commented Apr 7, 2017

cheleb commented Apr 7, 2017 via email

jrudolph commented Apr 20, 2017

Scalahansolo commented Apr 20, 2017

jrudolph left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Scalahansolo commented Apr 20, 2017

jrudolph commented May 8, 2017

Scalahansolo commented May 8, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Scalahansolo commented May 8, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jrudolph left a comment

Choose a reason for hiding this comment

jrudolph left a comment

Choose a reason for hiding this comment

jrudolph commented May 10, 2017