add s3collection_marker_each() helper for s3 consumers

[ender672]

The motivation behind this helper is to leverage boto3 collections,
which allow chaining.

This makes it possible to write consumers that control the s3 connection
details. It is also an experiment in creating an alternate API for
consumers.

Usage example would be:

s3 = boto3.resource('s3')
collection = s3.Bucket(BUCKET).objects.filter(Prefix=PREFIX)
for obj in s3collection_marker_each(collection, MARKER_PATH):
print obj.get()['Body'].read()

The above is essentially the same as:

def process(obj):
print obj

c = S3Consumer(ROUTING_KEY, process, NAME)
c.consume()

Which is equivalent to:

@consumer(ROUTING_KEY)
def process(obj):
print obj

send_email_to_interested_client.consumer.consume()

[sirsgriffin]

Probably good to be able to set a property via the S3Consumer class + decorator that allows use to elect to feed the consumer a string (default ... cause everything is treating this as a string already) or json. Would be nice to write new consumers using json.

[sirsgriffin]

As long as we're working on Muskrat it would be nice to feed optional metadata into the consumer. Namely, the cursor is not available to the consumer except by inspecting the consumer property that is added to the function. Even then it only corresponds to the last cursor that was written. Would definitely be nice to have this in a few cases. The main one is, of course, logging.

[ender672]

Thanks, you got me thinking about how to make cursor behavior extensible.

I'm experimenting with directly using S3Cursor in the consumer.

I like the S3Cursor class, and now I'm thinking this functionality that I had as standalone methods, e.g. s3collection_marker_each() should really be instance methods of S3Cursor:

class S3Cursor():
    def filter_collection(self, collection):
        marker = self.get()
        if marker:
            collection = collection.filter(Marker=marker)
        return collection

    def each(self, collection):
        for obj in filter_collection(collection):
            yield obj
            self.update(obj.key)

This would highlight S3Cursor as the place for consumers to customize cursor behavior.

Here is a sample vanilla consumer:

for obj in S3Cursor('MyName').each(bucket_objects):
    my_handler(obj)

Here is a consumer that wants to log cursor errors:

cursor = S3Cursor('MyName')
for obj in cursor.filter_collection(bucket_objects):
    my_handler(obj)

    try:
        cursor.update(obj.key)
    except IOError as e:
        log_cursor_error(e)

[ender672]

Hi Scott, can you take a look at the latest version? I reduced the scope of this PR to the following two additions to the Muskrat API:

S3Cursor.at_path()
This is an alternative initializer for S3Cursor which allows you to specify the path of the marker file.

s3cursor.each(collection)
This reads the marker file and returns a new collection pointing at the read marker, and after each iteration it will update the persistent marker file.

[sirsgriffin]

What is the behavior of this if the obj/key is not message. If this key points to something that is considered a namespace (something like a 'sub directory'. Not sure what to call it in S3) then what happens?

[ender672]

I just created a test for this. Just to be sure that I have it right:

Given that you are monitoring the prefix "FOO/BAR/".
And the bucket is empty.

When you add object A with key "FOO/BAR/2016-09-21T13:53:23.594894"
And you add object B with key "FOO/BAR/BAZ/2016-09-21T13:54:37.164853"
And the muskrat consumer is invoked

Then we should process object A
And we should not process object B

Is this the right test? If so, the implementation in this PR fails -- it processes both A and B.

[ender672]

This is fixed in 698660e

[ender672]

Is satisfying this test the behavior that we want?

When a consumer is monitoring "FOO/BAR" do we want to ignore "FOO/BAR/BAZ/2016-09-21T13:54:37.164853" ?

[ender672]

One more thing to point out -- this PR doesn't use the "delimiter" parameter when listing S3 entries, so the collection shouldn't yield prefix objects.

[ender672]

The test I mentioned is here:
ender672/muskrat@cca4b32