Class: WatchmonkeyCli::Requeue

Inherits:
Object
  • Object
show all
Defined in:
lib/watchmonkey_cli/hooks/requeue.rb

Class Method Summary collapse

Class Method Details

.hook!(app) ⇒ Object 



3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
 
# File 'lib/watchmonkey_cli/hooks/requeue.rb', line 3

def self.hook!(app)
  app.instance_eval do
    @requeue = []

    # app options
    @opts[:loop_forever] = true
    @opts[:logfile] = logger_filename # enable logging

    # scheduler options
    @opts[:requeue_scheduler_hibernation] = 1 # tickrate of schedule in seconds

    # module options
    @opts[:default_requeue]                   = 60
    # @opts[:default_requeue_ftp_availability]  = 60
    @opts[:default_requeue_mysql_replication] = 30
    @opts[:default_requeue_ssl_expiration]    = 1.hour
    @opts[:default_requeue_ts3_license]       = 1.hour
    @opts[:default_requeue_unix_defaults]     = false
    # @opts[:default_requeue_unix_df]           = 60
    # @opts[:default_requeue_unix_file_exists]  = 60
    # @opts[:default_requeue_unix_load]         = 60
    @opts[:default_requeue_unix_mdadm]        = 5.minutes
    # @opts[:default_requeue_unix_memory]       = 60
    @opts[:default_requeue_www_availability]  = 30


    # =================
    # = Status thread =
    # =================
    @requeue_status_thread = Thread.new do
      Thread.current.abort_on_exception = true
      while STDIN.gets
        sync do
          puts "==========  STATUS  =========="
          puts "     Queue: #{@queue.length}"
          puts "   Requeue: #{@requeue.length}"
          puts "   Workers: #{@threads.select{|t| t[:working] }.length}/#{@threads.length} working (#{@threads.select(&:alive?).length} alive)"
          puts "   Threads: #{filtered_threads.length}"
          # puts "            #{@threads.select(&:alive?).length} alive"
          # puts "            #{@threads.select{|t| t.status == "run" }.length} running"
          # puts "            #{@threads.select{|t| t.status == "sleep" }.length} sleeping"
          puts " Processed: #{@processed}"
          puts "  Promises: #{@telegram_bot_egress_promises.length}" if @telegram_bot_egress_promises
          puts "========== //STATUS =========="
        end
      end
    end


    # =================
    # = Scheduler thread =
    # =================
    @requeue_scheduler_thread = Thread.new do
      Thread.current.abort_on_exception = true
      loop do
        break if $wm_runtime_exiting
        sync do
          @requeue.each_with_index do |(run_at, callback), index|
            next if run_at > Time.now
            callback.call()
            @requeue.delete_at(index)
          end
        end
        sleep @opts[:requeue_scheduler_hibernation]
      end
    end


    # =========
    # = Hooks =
    # =========
    hook :dequeue do |checker, args|
      opts = args.extract_options!
      retry_in = opts[:every] if opts[:every].is_a?(Numeric)
      retry_in = @opts[:"default_requeue_#{checker.class.checker_name}"] if retry_in.nil?
      retry_in = @opts[:default_requeue] if retry_in.nil?
      if retry_in
        debug "Requeuing #{checker} in #{retry_in} seconds"
        requeue checker, args + [opts], retry_in
      end
    end

    hook :wm_shutdown do
      sync do
        @requeue_scheduler_thread.try(:join)
        debug "[ReQ] Clearing #{@requeue.length} items in requeue..."
        @requeue_status_thread.try(:kill).try(:join)
      end
    end


    # ===========
    # = Methods =
    # ===========
    def requeue checker, args, delay = 10
      return if $wm_runtime_exiting
      sync do
        @requeue << [Time.now + delay, ->{
          checker.enqueue(*args)
        }]
      end
    end

    def requeue_runall
      return if $wm_runtime_exiting
      sync do
        debug "Running all queued tasks immediately!"
        @requeue.each_with_index do |(run_at, callback), index|
          @requeue[index][0] = Time.now
        end
      end
    end
  end
end