Merge pull request #6950 from cmrd-senya/consolidate-queues

Consolidate queues

Changelog.md

@@ -59,6 +59,24 @@ wiki](https://wiki.diasporafoundation.org/Integration/Chat#Vines_to_Prosody)
 for more information on how to migrate to Prosody if you've been using Vines
 before.
 
+## Sidekiq queue changes
+
+We've decreased the amount of sidekiq queues from 13 to 5 in PR [#6950](https://github.com/diaspora/diaspora/pull/6950).
+The new queues are organized according to priority for the jobs they will process. When upgrading please make sure to
+empty the sidekiq queues before shutting down the server for an update.
+
+If you run your sidekiq with a custom queue configuration, please make sure to update that for the new queues.
+
+The new queues are: `urgent, high, medium, low, default`.
+
+When you upgrade to the new version, some jobs may persist in the old queues. To ensure that jobs to be processed, launch
+job processing for old queues using command:
+```
+rake migrations:run_legacy_queues
+```
+
+The command will report queues that still have jobs and launch sidekiq process for that queues.
+
 ## Refactor
 * Improve bookmarklet [#5904](https://github.com/diaspora/diaspora/pull/5904)
 * Update listen configuration to listen on unix sockets by default [#5974](https://github.com/diaspora/diaspora/pull/5974)
@@ -118,6 +136,7 @@ before.
 * Extract relayable signatures into their own tables [#6932](https://github.com/diaspora/diaspora/pull/6932)
 * Remove outdated columns from posts table [#6940](https://github.com/diaspora/diaspora/pull/6940)
 * Remove some unused routes [#6781](https://github.com/diaspora/diaspora/pull/6781)
+* Consolidate sidekiq queues [#6950](https://github.com/diaspora/diaspora/pull/6950)
 
 ## Bug fixes
 * Destroy Participation when removing interactions with a post [#5852](https://github.com/diaspora/diaspora/pull/5852)

app/workers/clean_cached_files.rb

@@ -1,6 +1,6 @@
 module Workers
   class CleanCachedFiles < Base
-    sidekiq_options queue: :maintenance
+    sidekiq_options queue: :low
 
     def perform
       CarrierWave.clean_cached_files!

app/workers/deferred_dispatch.rb

@@ -4,7 +4,7 @@
 
 module Workers
   class DeferredDispatch < Base
-    sidekiq_options queue: :dispatch
+    sidekiq_options queue: :high
 
     def perform(user_id, object_class_name, object_id, opts)
       user = User.find(user_id)

app/workers/deferred_retraction.rb

@@ -4,7 +4,7 @@
 
 module Workers
   class DeferredRetraction < Base
-    sidekiq_options queue: :dispatch
+    sidekiq_options queue: :high
 
     def perform(user_id, retraction_data, recipient_ids, opts)
       user = User.find(user_id)

app/workers/delete_account.rb

@@ -5,7 +5,7 @@
 
 module Workers
   class DeleteAccount < Base
-    sidekiq_options queue: :delete_account
+    sidekiq_options queue: :low
     
     def perform(account_deletion_id)
       account_deletion = AccountDeletion.find(account_deletion_id)

app/workers/delete_post_from_service.rb

@@ -4,7 +4,7 @@
 #
 module Workers
   class DeletePostFromService < Base
-    sidekiq_options queue: :http_service
+    sidekiq_options queue: :high
 
     def perform(service_id, opts)
       service = Service.find_by_id(service_id)

app/workers/export_photos.rb

@@ -5,7 +5,7 @@
 
 module Workers
   class ExportPhotos < Base
-    sidekiq_options queue: :export
+    sidekiq_options queue: :low
 
     def perform(user_id)
       @user = User.find(user_id)

app/workers/export_user.rb

@@ -5,7 +5,7 @@
 
 module Workers
   class ExportUser < Base
-    sidekiq_options queue: :export
+    sidekiq_options queue: :low
 
     def perform(user_id)
       @user = User.find(user_id)

app/workers/fetch_profile_photo.rb

@@ -5,7 +5,7 @@
 
 module Workers
   class FetchProfilePhoto < Base
-    sidekiq_options queue: :photos
+    sidekiq_options queue: :medium
 
     def perform(user_id, service_id, fallback_image_url = nil)
       service = Service.find(service_id)

app/workers/fetch_public_posts.rb

@@ -4,7 +4,7 @@
 
 module Workers
   class FetchPublicPosts < Base
-    sidekiq_options queue: :http_service
+    sidekiq_options queue: :medium
 
     def perform(diaspora_id)
       Diaspora::Fetcher::Public.new.fetch!(diaspora_id)

app/workers/fetch_webfinger.rb

@@ -4,7 +4,7 @@
 
 module Workers
   class FetchWebfinger < Base
-    sidekiq_options queue: :socket_webfinger
+    sidekiq_options queue: :urgent
 
     def perform(account)
       person = Person.find_or_fetch_by_identifier(account)

app/workers/gather_o_embed_data.rb

@@ -5,7 +5,7 @@
 
 module Workers
   class GatherOEmbedData < Base
-    sidekiq_options queue: :http_service
+    sidekiq_options queue: :medium
 
     def perform(post_id, url, retry_count=1)
       post = Post.find(post_id)

app/workers/gather_open_graph_data.rb

@@ -5,7 +5,7 @@
 
 module Workers
   class GatherOpenGraphData < Base
-    sidekiq_options queue: :http_service
+    sidekiq_options queue: :medium
 
     def perform(post_id, url, retry_count=1)
       post = Post.find(post_id)

app/workers/mail/also_commented.rb

@@ -1,7 +1,7 @@
 module Workers
   module Mail
     class AlsoCommented < Base
-      sidekiq_options queue: :mail
+      sidekiq_options queue: :low
 
       def perform(recipient_id, sender_id, comment_id)
         if email = Notifier.also_commented(recipient_id, sender_id, comment_id)

app/workers/mail/comment_on_post.rb

@@ -1,7 +1,7 @@
 module Workers
   module Mail
     class CommentOnPost < Base
-      sidekiq_options queue: :mail
+      sidekiq_options queue: :low
 
       def perform(recipient_id, sender_id, comment_id)
         Notifier.comment_on_post(recipient_id, sender_id, comment_id).deliver_now

app/workers/mail/confirm_email.rb

@@ -1,7 +1,7 @@
 module Workers
   module Mail
     class ConfirmEmail < Base
-      sidekiq_options queue: :mail
+      sidekiq_options queue: :low
       
       def perform(user_id)
         Notifier.confirm_email(user_id).deliver_now

app/workers/mail/invite_email.rb

@@ -5,7 +5,7 @@
 module Workers
   module Mail
     class InviteEmail < Base
-      sidekiq_options queue: :mail
+      sidekiq_options queue: :low
 
       def perform(emails, inviter_id, options={})
         EmailInviter.new(emails, User.find(inviter_id), options).send!

app/workers/mail/liked.rb

@@ -1,7 +1,7 @@
 module Workers
   module Mail
     class Liked < Base
-      sidekiq_options queue: :mail
+      sidekiq_options queue: :low
 
       def perform(recipient_id, sender_id, like_id)
         Notifier.liked(recipient_id, sender_id, like_id).deliver_now

app/workers/mail/mentioned.rb

@@ -6,7 +6,7 @@
 module Workers
   module Mail
     class Mentioned < Base
-      sidekiq_options queue: :mail
+      sidekiq_options queue: :low
       
       def perform(recipient_id, actor_id, target_id)
         Notifier.mentioned( recipient_id, actor_id, target_id).deliver_now

app/workers/mail/private_message.rb

@@ -6,7 +6,7 @@
 module Workers
   module Mail
     class PrivateMessage < Base
-      sidekiq_options queue: :mail
+      sidekiq_options queue: :low
       
       def perform(recipient_id, actor_id, target_id)
         Notifier.private_message( recipient_id, actor_id, target_id).deliver_now

app/workers/mail/report_worker.rb

@@ -1,7 +1,7 @@
 module Workers
   module Mail
     class ReportWorker < Base
-      sidekiq_options queue: :mail
+      sidekiq_options queue: :low
 
       def perform(report_id)
         ReportMailer.new_report(report_id).each(&:deliver_now)

app/workers/mail/reshared.rb

@@ -1,7 +1,7 @@
 module Workers
   module Mail
     class Reshared < Base
-      sidekiq_options queue: :mail
+      sidekiq_options queue: :low
       
       def perform(recipient_id, sender_id, reshare_id)
         Notifier.reshared(recipient_id, sender_id, reshare_id).deliver_now

app/workers/mail/started_sharing.rb

@@ -6,7 +6,7 @@
 module Workers
   module Mail
     class StartedSharing < Base
-      sidekiq_options queue: :mail
+      sidekiq_options queue: :low
       
       def perform(recipient_id, sender_id, target_id)
         Notifier.started_sharing(recipient_id, sender_id).deliver_now

app/workers/post_to_service.rb

@@ -4,7 +4,7 @@
 #
 module Workers
   class PostToService < Base
-    sidekiq_options queue: :http_service
+    sidekiq_options queue: :medium
 
     def perform(service_id, post_id, url)
       service = Service.find_by_id(service_id)

app/workers/process_photo.rb

@@ -5,7 +5,7 @@
 
 module Workers
   class ProcessPhoto < Base
-    sidekiq_options queue: :photos
+    sidekiq_options queue: :low
 
     def perform(id)
       photo = Photo.find(id)

app/workers/publish_to_hub.rb

@@ -4,7 +4,7 @@
 
 module Workers
   class PublishToHub < Base
-    sidekiq_options queue: :http_service
+    sidekiq_options queue: :medium
 
     def perform(sender_atom_url)
       Pubsubhubbub.new(AppConfig.environment.pubsub_server.get).publish(sender_atom_url)

app/workers/queue_users_for_removal.rb

@@ -4,7 +4,7 @@
 
 module Workers
   class QueueUsersForRemoval < Base
-    sidekiq_options queue: :maintenance
+    sidekiq_options queue: :low
 
     def perform
       # Queue users for removal due to inactivity

app/workers/receive_base.rb

@@ -1,6 +1,6 @@
 module Workers
   class ReceiveBase < Base
-    sidekiq_options queue: :receive
+    sidekiq_options queue: :urgent
 
     include Diaspora::Logging
 

app/workers/receive_local.rb

@@ -1,6 +1,6 @@
 module Workers
   class ReceiveLocal < Base
-    sidekiq_options queue: :receive_local
+    sidekiq_options queue: :high
 
     def perform(object_class_string, object_id, recipient_user_ids)
       object = object_class_string.constantize.find(object_id)

app/workers/recurring_pod_check.rb

@@ -1,6 +1,6 @@
 module Workers
   class RecurringPodCheck < Base
-    sidekiq_options queue: :maintenance
+    sidekiq_options queue: :low
 
     def perform
       Pod.check_all!

app/workers/remove_old_user.rb

@@ -4,7 +4,7 @@
 
 module Workers
   class RemoveOldUser < Base
-    sidekiq_options queue: :maintenance
+    sidekiq_options queue: :low
     
     def safe_remove_after
       # extra safety time to compare in addition to remove_after
@@ -24,4 +24,4 @@ module Workers
       end
     end
   end 
-end
+end

app/workers/resend_invitation.rb

@@ -5,7 +5,7 @@
 
 module Workers
   class ResendInvitation < Base
-    sidekiq_options queue: :mail
+    sidekiq_options queue: :low
     
     def perform(invitation_id)
       inv = Invitation.find(invitation_id)

app/workers/reset_password.rb

@@ -1,6 +1,6 @@
 module Workers
   class ResetPassword < Base
-    sidekiq_options queue: :mail
+    sidekiq_options queue: :urgent
 
     def perform(user_id)
       User.find(user_id).send_reset_password_instructions!

app/workers/send_base.rb

@@ -1,6 +1,6 @@
 module Workers
   class SendBase < Base
-    sidekiq_options queue: :http, retry: 0
+    sidekiq_options queue: :medium, retry: 0
 
     MAX_RETRIES = AppConfig.environment.sidekiq.retry.get.to_i
 

config/sidekiq.yml

@@ -8,16 +8,8 @@
 :dead_max_jobs: <%= AppConfig.environment.sidekiq.dead_jobs_limit.to_i %>
 :dead_timeout_in_seconds: <%= AppConfig.environment.sidekiq.dead_jobs_timeout.to_i %>
 :queues:
-  - socket_webfinger
+  - urgent
-  - photos
+  - high
-  - http_service
+  - medium
-  - dispatch
+  - low
-  - mail
-  - delete_account
-  - receive_local
-  - receive
-  - receive_salmon
-  - http
-  - export
-  - maintenance
   - default

lib/tasks/migrations.rake

@@ -127,4 +127,20 @@ namespace :migrations do
       tag.destroy
     end
   end
+
+  LEGACY_QUEUES = %w(
+    maintenance dispatch delete_account http http_service export photos socket_webfinger mail receive_local receive
+  ).freeze
+
+  desc "Run sidekiq with old queues so it can finish deferred jobs"
+  task :run_legacy_queues do
+    queues_with_jobs = LEGACY_QUEUES.select {|queue| Sidekiq::Queue.new(queue).size > 0 }
+    if queues_with_jobs.empty?
+      puts "No jobs in legacy queues!"
+    else
+      puts "Launching sidekiq with queues: #{queues_with_jobs.join(', ')}"
+      queus_cli = queues_with_jobs.map {|queue| "-q #{queue}" }.join(" ")
+      system "bundle exec sidekiq #{queus_cli} -e #{Rails.env}"
+    end
+  end
 end