Merge pull request #514 from warpc/510-resque_migrate

[refs #510] migrate to resque for heavy task

.gitignore

@@ -14,3 +14,4 @@ public/downloads/*
 *.swp
 *.tmproj
 .sass-cache/
+dump.rdb

Gemfile

@@ -15,6 +15,7 @@ gem 'cancan', '~> 1.6.7'
 gem 'ancestry', '~> 1.2.5'
 gem 'paperclip', '~> 3.0.2'
 gem 'delayed_job_active_record', '~> 0.3.2'
+gem 'resque'
 gem 'russian', '~> 0.6.0'
 gem 'highline', '~> 1.6.11'
 

Gemfile.lock

@@ -235,6 +235,14 @@ GEM
     rdoc (3.12)
       json (~> 1.4)
     redcarpet (1.17.2)
+    redis (2.2.2)
+    redis-namespace (1.0.3)
+      redis (< 3.0.0)
+    resque (1.20.0)
+      multi_json (~> 1.0)
+      redis-namespace (~> 1.0.2)
+      sinatra (>= 0.9.2)
+      vegas (~> 0.1.2)
     rr (1.0.4)
     rspec (2.9.0)
       rspec-core (~> 2.9.0)
@@ -303,6 +311,8 @@ GEM
       kgio (~> 2.6)
       rack
       raindrops (~> 0.7)
+    vegas (0.1.11)
+      rack (>= 1.0.0)
     warden (1.1.1)
       rack (>= 1.0)
     whenever (0.7.3)
@@ -356,6 +366,7 @@ DEPENDENCIES
   rdiscount
   redcarpet (= 1.17.2)
   redhillonrails_core!
+  resque
   rr (~> 1.0.4)
   rspec-rails (~> 2.9.0)
   ruby-haml-js (~> 0.0.3)

Rakefile

@@ -3,5 +3,11 @@
 
 require File.expand_path('../config/application', __FILE__)
 require 'rake'
+require 'resque/tasks'
+
+# This fixes connection fail with Postgres server on new fork:
+task "resque:setup" => :environment do
+  Resque.before_fork = Proc.new { ActiveRecord::Base.establish_connection }
+end
 
 Rosa::Application.load_tasks

app/models/ability.rb

@@ -27,6 +27,8 @@ class Ability
     else # Registered user rights
       if user.admin?
         can :manage, :all
+        # Resque authorize
+        can :manage, Resque
         # Protection
         cannot :approve, RegisterRequest, :approved => true
         cannot :reject, RegisterRequest, :rejected => true

app/models/project.rb

@@ -41,12 +41,12 @@ class Project < ActiveRecord::Base
 
   after_create :attach_to_personal_repository
   after_create :create_git_repo
-  after_create {|p| p.delay(:queue => 'fork', :priority => 20).fork_git_repo unless is_root?}
+  after_create {|p| p.async(:fork_git_repo) unless is_root?}
   after_save :create_wiki
 
   after_destroy :destroy_git_repo
   after_destroy :destroy_wiki
-  after_save {|p| p.delay(:queue => 'import', :priority => 10).import_attached_srpm if p.srpm?} # should be after create_git_repo
+  after_save {|p| p.async(:import_attached_srpm) if p.srpm?} # should be after create_git_repo
   # after_rollback lambda { destroy_git_repo rescue true if new_record? }
 
   has_ancestry
@@ -55,6 +55,20 @@ class Project < ActiveRecord::Base
 
   include Modules::Models::Owner
 
+  @queue = :fork_and_import
+
+  # This will be called by a worker when a job needs to be processed
+  def self.perform(id, method, *args)
+    find(id).send(method, *args)
+  end
+
+  # We can pass this any Repository instance method that we want to
+  # run later.
+  def async(method, *args)
+    Resque.enqueue(Project, id, method, *args)
+  end
+
+
   def to_param
     name
   end
@@ -252,7 +266,7 @@ class Project < ActiveRecord::Base
   def create_git_repo
     if is_root?
       Grit::Repo.init_bare(path)
-      write_hook.delay(:queue => 'fork', :priority => 15)
+      async(:write_hook)
     end
   end
 

config/deploy.rb

@@ -37,6 +37,7 @@ require './lib/recipes/nginx'
 require './lib/recipes/unicorn'
 require './lib/recipes/bluepill'
 require './lib/recipes/delayed_job'
+require './lib/recipes/resque'
 
 namespace :deploy do
   task :stub_xml_rpc do
@@ -89,6 +90,11 @@ after "deploy:stop",    "delayed_job:stop"
 after "deploy:start",   "delayed_job:start"
 after "deploy:restart", "delayed_job:restart"
 
+# Resque
+after "deploy:stop",    "resque:stop"
+after "deploy:start",   "resque:start"
+after "deploy:restart", "resque:restart"
+
 after "deploy:restart", "deploy:cleanup"
 
 namespace :rake_tasks do

config/initializers/admin.rb

@@ -0,0 +1,8 @@
+# config/initializers/admin.rb
+class CanAccessResque
+  def self.matches?(request)
+    current_user = request.env['warden'].user
+    return false if current_user.blank?
+    Ability.new(current_user).can? :manage, Resque
+  end
+end

config/production.pill

@@ -4,7 +4,7 @@ app_name = ENV['APP_NAME'] || 'rosa_build'
 Bluepill.application(app_name) do |app|
   app.uid = app.gid = 'rosa'
   app.working_dir = "/srv/#{app_name}/current"
-  %w(fork import hook default).each do |queue|
+  %w(hook default).each do |queue|
     app.process("delayed_job_#{queue}_queue") do |process|
       process.start_grace_time    = 10.seconds
       process.stop_grace_time     = 10.seconds

config/routes.rb

@@ -1,5 +1,7 @@
 # -*- encoding : utf-8 -*-
 Rosa::Application.routes.draw do
+  require 'resque/server'
+
   devise_scope :users do
     get '/users/auth/:provider' => 'users/omniauth_callbacks#passthru'
   end
@@ -22,6 +24,10 @@ Rosa::Application.routes.draw do
   end
 
   namespace :admin do
+    constraints CanAccessResque do
+      mount Resque::Server, at: 'resque'
+    end
+
     resources :users do
       get :list, :on => :collection
     end

lib/recipes/delayed_job.rb

@@ -2,7 +2,7 @@
 Capistrano::Configuration.instance(:must_exist).load do
   namespace :delayed_job do
     def dj_queues
-      %w(fork import hook default)
+      %w(hook default)
     end
 
     def rails_env

lib/recipes/resque.rb

@@ -0,0 +1,37 @@
+# -*- encoding : utf-8 -*-
+Capistrano::Configuration.instance(:must_exist).load do
+  namespace :resque do
+    task :start do
+      start_workers
+    end
+
+    task :stop do
+      stop_workers
+    end
+
+    task :restart do
+      stop_workers
+      start_workers
+    end
+
+    def rails_env
+      fetch(:rails_env, false) ? "RAILS_ENV=#{fetch(:rails_env)}" : ''
+    end
+
+    def stop_workers
+      pids = Array.new
+
+      Resque.workers.each do |worker|
+        pids << worker.to_s.split(/:/).second
+      end
+
+      if pids.size > 0
+        system("kill -QUIT #{pids.join(' ')}")
+      end
+    end
+
+    def start_workers
+      run "cd #{fetch :release_path} && QUEUE=* bundle exec rake resque:work"
+    end
+  end
+end