OPER-5336 Update AWS SDK library #55
Conversation
aaa486a to
df9d9be
Compare
79b6439 to
89530e0
Compare
lib/chore/unit_of_work.rb
Outdated
| @@ -19,7 +20,7 @@ def initialize(*) #:nodoc: | |||
|
|
|||
| # The current attempt number for the worker processing this message. | |||
| def current_attempt | |||
| previous_attempts + 1 | |||
| previous_attempts.to_i + 1 | |||
There was a problem hiding this comment.
I came across this and it seemed like a subtle bug. Can anyone think of a reason not to do this?
There was a problem hiding this comment.
Under what circumstances is this not an integer? As far as I can tell, both the filesystem and sqs consumer pass an integer through here. If the specs are passing a string, then I'd say the specs are wrong.
There was a problem hiding this comment.
In reality we never pass in anything but an integer as far as I can tell. I noticed this because by looking at UnitOfWork it appears that :previous_attempts could at the very least be nil. But yeah, specs should cover that.
There was a problem hiding this comment.
I think in that case we should remove this change as I see that as both a Ruby language design problem and a failed contract for the consumer. I think better documentation could make it clearer what is expected to be nil vs. not nil.
There was a problem hiding this comment.
nod I will back out the change as part of addressing forthcoming feedback :)
| # TODO: This seems like it's no longer necessary. If it is, fix and reenable it, otherwise delete | ||
| xit 'should look up the queue based on the queue url' do | ||
| expect(sqs).to receive(:get_queue_url).with(:queue_name=>queue_name).and_return(queue_url) | ||
| expect(queue).to receive(:[]).with(queue_url).and_return(queue) | ||
| consume | ||
| end |
There was a problem hiding this comment.
You've replaced this with a call to Aws::SQS::Queue.new(url: @queue_url), so I think you can get rid of this unless you want to replace it with a spec that validates Aws::SQS::Queue is built properly and passed the correct options.
| # TODO: AThe reset_connection! method is setup so every call resets it, so this spec shouldn't even exist right? | ||
| # Also, the test itself seems like basic identity (i.e. not even a real test) | ||
| xit 'should not reset the connection between calls' do | ||
| sqs = publisher.send(:queue, queue_name) | ||
| expect(sqs).to be publisher.send(:queue, queue_name) |
There was a problem hiding this comment.
I think this is still a spec we'll want
|
Submitted to ops team for a first pass review. Will expand out to more people when this has passed through that filter. |
3de07db to
6aaa5e7
Compare
README.md
Outdated
|
|
||
| For producers, you must do all of your Chore configuration in an intializer. | ||
| **IS THIS TRUE?** => (the consumer is also able to produce messages if need be, but is running as its own isolated instance of your application). |
There was a problem hiding this comment.
This doesn't seem correct to me, should I remove it?
There was a problem hiding this comment.
I assume this is just saying one could call the perform of another job from within the consumer which seems plausible?
There was a problem hiding this comment.
A worker can do that, but I don't see any logic in the consumer that would actually publish a message. @obrie do you happen to be able to figure out what was meant by this language?
There was a problem hiding this comment.
I think it's just talking about the consuming application, not the consumer queue implementation.
There was a problem hiding this comment.
That makes a lot more sense, thank you. I will update this language.
|
@alanbrent Could you provide a more comprehensive overview of what changes were made and why they were made? It would be helpful to know what were QOL changes, what were bug fixes, what changes were made to explicitly support the AWS upgrade, etc. etc. It seems like there's a lot of stuff in this PR that's not specifically for the AWS upgrade, but I'm not certain. As a side note, is it possible to split out the changes into logical commits if they are unrelated to each other? |
6418591 to
140010c
Compare
4c238c9 to
bafe85b
Compare
lib/chore.rb
Outdated
| @@ -63,6 +63,24 @@ def self.logger | |||
| end | |||
| end | |||
|
|
|||
| def self.log_level_to_sym | |||
| return self.config[:log_level] if self.config[:log_level].kind_of?Symbol | |||
There was a problem hiding this comment.
Style nit -- missing parens with kind_of? (also I typically see is_a? be used, but they're synonymous either way)
lib/chore/consumer.rb
Outdated
| raise NotImplementedError | ||
| end | ||
|
|
||
| # Perform any shutdown behavior and stop consuming messages | ||
| # | ||
| # @return [FalseClass] |
There was a problem hiding this comment.
While this method does return false, I don't think that's a behavior we want to document or an interface we would want to guarantee. It just happens to be the last line executed in this method.
lib/chore/queues/sqs.rb
Outdated
| Chore.prefixed_queue_names | ||
| end | ||
|
|
||
| # Collect a list of queues that already exist | ||
| # | ||
| # @return [TrueClass, FalseClass] |
There was a problem hiding this comment.
This method returns an array of strings (the prefixed_queue_names that exist)
lib/chore/queues/sqs/consumer.rb
Outdated
| def self.reset_connection! | ||
| @@reset_at = Time.now | ||
| Aws.empty_connection_pools! |
There was a problem hiding this comment.
This logic shouldn't be in here -- this is just responsible for setting the timestamp
lib/chore/queues/sqs/consumer.rb
Outdated
| AWS::Core::Http::ConnectionPool.pools.each do |p| | ||
| p.empty! | ||
| end | ||
| self.class.reset_connection! |
There was a problem hiding this comment.
Careful here -- we'll never get here because reset_connection! is the one that sets @@reset_at. We're safer leaving the previous logic as-is and simply switching the AWS::Core::Http::ConnectionPool.pools.each do |p| block to call Aws.empty_connection_pools!
There was a problem hiding this comment.
nod I plan to do the first part (revert to previous flow), but the call defining that loop is no longer a thing. AFAICT the Aws.reset_connection_pools! does the same conceptual work.
[1] pry(main)> show-source Aws.empty_connection_pools!
From: /Users/brent.stephens/Documents/development/src/github.com/tapjoy/chore/vendor/bundle/ruby/2.3.0/gems/aws-sdk-core-3.105.0/lib/aws-sdk-core.rb:153:
Owner: #<Class:Aws>
Visibility: public
Signature: empty_connection_pools!()
Number of lines: 5
def empty_connection_pools!
Seahorse::Client::NetHttp::ConnectionPool.pools.each do |pool|
pool.empty!
end
end
lib/chore/queues/sqs/consumer.rb
Outdated
| Chore.logger.debug "Completing (deleting): #{id}" | ||
| queue.batch_delete([id]) | ||
| raise Chore::Consumer::CouldNotComplete, 'Required param "receipt_handle" missing!' unless opts[:receipt_handle] |
There was a problem hiding this comment.
If this is just to enforce an interface rather than an edge case error condition, I think we'd be better off defining it in the method, e.g.
def complete(message_id, receipt_handle:)
...There was a problem hiding this comment.
On second thought, maybe we should just make it a standard second parameter. That's probably simpler and doesn't give the impression that you can just easily pass through any generic options through here. It also reflects the fact that this data is always passed in.
There was a problem hiding this comment.
Should I document anything around the signature being different than the one defined in the interface, or is that a common thing that won't surprise developers?
There was a problem hiding this comment.
What do you think about just updating the interface? Implementations can then decide whether they need/respect the argument.
lib/chore/queues/sqs/publisher.rb
Outdated
| def self.reset_connection! | ||
| @@reset_next = true | ||
| Aws.empty_connection_pools! |
There was a problem hiding this comment.
Just noting that this is also impacted by the same comment on where the logic should live
| # single queue which can be kept up with at a relatively low volume. If you have more than a single queue | ||
| # configured, it will raise an exception. | ||
| # | ||
| # @return [TrueClass, FalseClass] |
There was a problem hiding this comment.
No return doc needed here (I'll stop repeating myself -- just need to apply this across the doc changes you've made)
| Chore.run_hooks_for(:consumed_from_source, work) | ||
| @queue.push(work) if running? | ||
| Chore.run_hooks_for(:added_to_queue, work) | ||
| end | ||
| end | ||
| end # ThrottledConsumerStrategyyeah | ||
| end # ThrottledConsumerStrategy |
There was a problem hiding this comment.
😆 I do like the idea of a Lil' Jon language where the classes all end in yeah
F438
| @@ -75,7 +75,7 @@ def ipc_help | |||
|
|
|||
| private | |||
|
|
|||
| # TODO: do we need this as a optional param | |||
| # TODO: Decide if we should make this customizable via an optional param | |||
There was a problem hiding this comment.
I think the answer is probably "no" since we haven't needed to do it in years 😆
CHANGELOG.md
Outdated
| @@ -1,15 +1,22 @@ | |||
| # Change Log | |||
|
|
|||
| ## [master](https://github.com/Tapjoy/chore/tree/master) | |||
| ## [v4.0.0](https://github.com/Tapjoy/chore/tree/v4.0.0) | |||
There was a problem hiding this comment.
Can you create a new [master] section as part of this?
docs/Delayed Jobs.md
Outdated
|
|
||
| All of this should be taken into consideration when using this feature. In general, any queue requiring this will likely | ||
| need to run as its own Facet and not share workers with other queues. | ||
| All of this should be taken into consideration when using this feature. In general, any queue requiring this will likely need to run as its own Facet and not share workers with other queues. |
There was a problem hiding this comment.
Could you go back to limiting these lines to 80 / 100 / 120-ish characters? It becomes pretty difficult to read these files when I have to horizontally scroll each line. I'm sure you can set up the editor to do this for you, but I'd prefer to not assume what editor settings people have, particularly for open-source projects. Wrapped lines in Markdown is fairly common because Markdown will render it the same either way.
Examples:
lib/chore/queues/sqs/consumer.rb
Outdated
| Chore.logger.debug "Completing (deleting): #{id}" | ||
| queue.batch_delete([id]) | ||
| raise Chore::Consumer::CouldNotComplete, 'Required param "receipt_handle" missing!' unless opts[:receipt_handle] |
There was a problem hiding this comment.
On second thought, maybe we should just make it a standard second parameter. That's probably simpler and doesn't give the impression that you can just easily pass through any generic options through here. It also reflects the fact that this data is always passed in.
spec/chore/queues/spec_helper.rb
Outdated
| let(:queue_name) { 'test_queue' } | ||
| let(:queue_url) { "http://amazon.sqs.url/queues/#{queue_name}" } | ||
|
|
||
| let(:queue) do double(Aws::SQS::Queue, |
There was a problem hiding this comment.
Missing a newline between do and double
| consumer.consume | ||
| require_relative '../spec_helper' | ||
|
|
||
| module Chore |
There was a problem hiding this comment.
This is a bit unusual. I don't think I've ever seen modules used for specs. Curious what led to this change?
There was a problem hiding this comment.
The goal was to share AWS API object doubles in an effort to reduce the cognitive overhead of reasoning about all the SQS-related specs together. I just cargo-culted module Chore because that's the context the specs were all running in. I will finish reading through your interrelated comments and either make updates or ask more questions :)
There was a problem hiding this comment.
I'm not sure if I follow here. What's the benefit in putting specs in a module? As far as I'm aware, we don't do this in any of our codebases and I'm not sure I've ever seen that as a practice in rspec docs. The describe blocks themselves create anonymous modules, so there isn't typically value in surrounding them with another module.
There was a problem hiding this comment.
The same-functionality-but-different-implementation fake objects between the publisher and consumer was confusing to me, and it seemed to make sense to unify the objects to reduce that confusion as well as prevent drift. As far as the enclosing module declaration in the specs themselves, that was pre-existing. Did you want me to remove them?
There was a problem hiding this comment.
Update: it was pre-existing in the publisher, and I guess I updated the other spec to match.
| let(:consumer) { Chore::Queues::SQS::Consumer.new(queue_name) } | ||
| let(:backoff_func) { Proc.new { 2 + 2 } } | ||
| let(:dupe_detector) { double(Chore::DuplicateDetector) } | ||
| let(:empty_message_collection) { double(Aws::SQS::Message::Collection.new([])) } |
There was a problem hiding this comment.
Neither of the above 2 seem to be used
spec/chore/queues/spec_helper.rb
Outdated
| require 'spec_helper' | ||
|
|
||
| module Chore | ||
| RSpec.shared_context 'fake objects' do |
There was a problem hiding this comment.
I think this is going to cause problems in the long-run. The helpers here are specific to s3 and the shared context name will get clobbered when you have non-s3 queue implementations being spec'd. The pattern I typically see is to put this as something like spec/support/contexts/s3_queue_context.rb.
It then just gets loaded automatically in spec_helper, e.g.
Dir[Rails.root.join("spec/support/**/*.rb")].each {|f| require f}This also avoids having to use require_relative and confusion with which spec_helper is being referred to.
spec/chore/queues/spec_helper.rb
Outdated
|
|
||
| module Chore | ||
| RSpec.shared_context 'fake objects' do | ||
| let(:job) { {'class' => 'TestJob', 'args'=>[1,2,'3']}} |
There was a problem hiding this comment.
This seems like an outlier that doesn't need to be in the shared context. I can see an argument for the more complex s3 class creation being shared because you're effectively creating a fake_sqs, though.
| body: job.to_json, | ||
| data: job, | ||
| queue: queue, | ||
| queue_url: sqs.get_queue_url.queue_url, |
There was a problem hiding this comment.
This seems like we're taking advantage of the fact that the get_queue_url method doesn't actually require any arguments which I think isn't actually the case and is a bit confusing. I think it'd be clearer to reference the queue_url variable here.
lib/chore/queues/sqs/consumer.rb
Outdated
| @queue ||= sqs.queues[@queue_url] | ||
|
|
||
| @queue_url ||= sqs.get_queue_url(queue_name: @queue_name).queue_url | ||
| @queue ||= Aws::SQS::Queue.new(url: @queue_url) |
There was a problem hiding this comment.
Looks like we should be passing in client: sqs here: https://github.com/aws/aws-sdk-ruby/blob/877d729fa4176891e68c77143b80a3ae6612b9e7/gems/aws-sdk-sqs/lib/aws-sdk-sqs/queue.rb#L16-L26
Example of how the Resource class builds queues: https://github.com/aws/aws-sdk-ruby/blob/877d729fa4176891e68c77143b80a3ae6612b9e7/gems/aws-sdk-sqs/lib/aws-sdk-sqs/resource.rb#L260-L269
| # TODO: This seems like it's no longer necessary. If it is, fix and reenable it, otherwise delete | ||
| xit 'should look up the queue based on the queue url' do | ||
| expect(sqs).to receive(:get_queue_url).with(:queue_name=>queue_name).and_return(queue_url) | ||
| expect(queue).to receive(:[]).with(queue_url).and_return(queue) | ||
| consume | ||
| end |
There was a problem hiding this comment.
You've replaced this with a call to Aws::SQS::Queue.new(url: @queue_url), so I think you can get rid of this unless you want to replace it with a spec that validates Aws::SQS::Queue is built properly and passed the correct options.
| context "should receive a message from the queue" do | ||
| before do | ||
| allow(consumer).to receive(:queue).and_return(queue) | ||
| end |
There was a problem hiding this comment.
Is this required if you've previously stubbed it with allow(Aws::SQS::Queue).to receive(:new).and_return(queue)?
| it 'should create a new SQS client before every publish' do | ||
| expect(Aws::SQS::Client).to receive(:new).twice | ||
| publisher.send(:queue, queue_name) | ||
| publisher.send(:queue, queue_name) |
There was a problem hiding this comment.
Can be 2.times { publisher.send(:queue, queue_name) }
| # TODO: AThe reset_connection! method is setup so every call resets it, so this spec shouldn't even exist right? | ||
| # Also, the test itself seems like basic identity (i.e. not even a real test) | ||
| xit 'should not reset the connection between calls' do | ||
| sqs = publisher.send(:queue, queue_name) | ||
| expect(sqs).to be publisher.send(:queue, queue_name) |
There was a problem hiding this comment.
I think this is still a spec we'll want
57990c7 to
5215c51
Compare
|
Alright I think we're ready for the next round. Thanks for all the feedback :) |
lib/chore/queues/sqs/publisher.rb
Outdated
| end | ||
|
|
||
| # Sets a flag that instructs the publisher to reset the connection the next time it's used | ||
| # | ||
| # @return [Array<Seahorse::Client::NetHttp::ConnectionPool>] |
There was a problem hiding this comment.
Out-of-date doc (No need to document a return value here)
| consumer.consume | ||
| require_relative '../spec_helper' | ||
|
|
||
| module Chore |
There was a problem hiding this comment.
I'm not sure if I follow here. What's the benefit in putting specs in a module? As far as I'm aware, we don't do this in any of our codebases and I'm not sure I've ever seen that as a practice in rspec docs. The describe blocks themselves create anonymous modules, so there isn't typically value in surrounding them with another module.
c775f79 to
c077a12
Compare
c077a12 to
aab5bb7
Compare
aab5bb7 to
6440e9f
Compare
|
Thanks @ehealy for catching a bug in the consumer whose fix is captured in the diff below. I have updated the PR with a fix as well as an additional spec to more fully test the diff --git a/lib/chore/queues/sqs/consumer.rb b/lib/chore/queues/sqs/consumer.rb
index 7a76d49..4ca8ed5 100644
--- a/lib/chore/queues/sqs/consumer.rb
+++ b/lib/chore/queues/sqs/consumer.rb
@@ -58,8 +58,8 @@ module Chore
# @param [String] message_id Unique ID of the SQS message
# @param [Hash] receipt_handle Receipt handle (unique per consume request) of the SQS message
def complete(message_id, receipt_handle)
- Chore.logger.debug "Completing (deleting): #{id}"
- queue.delete_messages(entries: [{ id: id, receipt_handle: receipt_handle }])
+ Chore.logger.debug "Completing (deleting): #{message_id}"
+ queue.delete_messages(entries: [{ id: message_id, receipt_handle: receipt_handle }])
end |
6440e9f to
be77edd
Compare
There was a problem hiding this comment.
Using https://github.com/Tapjoy/opsbot/pull/84 as a testbed, I have validated the following:
- SQS Producer
- Initializes without issue
- Successfully submits messages to SQS
- SQS Consumer utilizing the Threaded Consumer Strategy
- Initializes without issue
- Successfully polls SQS for batches of messages for the Chore jobs specified
- Assigns messages to workers
- Workers complete Chore jobs
- Chore jobs are successfully removed form the appropriate SQS queues.
Any feedback/bugfixes were addressed via facewords and re-validated during testing cycle. Minor non-blocking YARD documentation suggestion.
Outstanding work @alanbrent.
2f80ef7 to
03bd447
Compare
03bd447 to
4bdde71
Compare
This PR updates the AWS SDK to a modern version with support for authentication via Web Federated Identity. While is designed to be a drop-in replacement for applications that already use Chore, there are various API changes in the AWS SDK, some of which have necessitated API changes in
Chore::Queues. Hence I have incremented the major version of the gem.Along the way I have made a bunch of QOL updates to the documentation. Hopefully people will find them useful :)
Chore API Changes
Minor Changes
Chore::Consumer#consumesignature has been unified between interface and implementationsChore::Consumer#handle_jobsrenamed toChore::Consumer::#handle_messagesto unify names between publisher and consumer interfacesChore::Consumer#handle_messagesprivate method used in all consumers has been added to the interface definitionChore::Publisher#reset_connection!added to interface definitionChore::Consumer-s andChore::Publisher-s unified to usemessage_idargument name rather thanidwhere applicableMajor Changes
UnitOfWorknow takes:receipt_handleargument to accommodate AWS SDK requirementsnilin the filesystem handler, which doesn't need itChore::Queues::SQSmodule rather than being duplicated in the publisher and consumer classesChore::Queues::Consumer#completesignature now takes an options hash to accommodate changes needed by the SQS consumerChore::Queues::SQS::Consumer#completenow requires the SQS message receipt handle to be passed in the options hashSpec Changes
double-ing the actual AWS API objects rather than generically named doublesexpect(thing).to receive(:action).and_return(expected_value)has been fixedand_returnis actually a mock being set rather than an expected value to be compared againstQuality Of Life Changes