Class: Pazuzu::Worker

Inherits:
Object
  • Object
show all
Includes:
Utility::Runnable
Defined in:
lib/pazuzu/worker.rb

Instance Attribute Summary collapse

Attributes included from Utility::Runnable

#started_at, #stopped_at

Instance Method Summary collapse

Methods included from Utility::Runnable

#run_state, #start!, #stop!, #wait_for_state_change!

Constructor Details

#initialize(application, name) ⇒ Worker

Returns a new instance of Worker.



7
8
9
10
11
12
13
14
15
16
 
# File 'lib/pazuzu/worker.rb', line 7

def initialize(application, name)
  @application = application
  @name = name
  @instances = Utility::RunnablePool.new
  @target_instance_count = 1
  @dynamic_instance_count = nil
  @logger = Utility::AnnotatedLogger.new(
    @application.supervisor.logger, qname)
  super()
end

Instance Attribute Details

#application ⇒ Object (readonly)

Returns the value of attribute application.



127
128
129
 
# File 'lib/pazuzu/worker.rb', line 127

def application
  @application
end

#command_line ⇒ Object (readonly)

Returns the value of attribute command_line.



126
127
128
 
# File 'lib/pazuzu/worker.rb', line 126

def command_line
  @command_line
end

#name ⇒ Object (readonly)

Returns the value of attribute name.



125
126
127
 
# File 'lib/pazuzu/worker.rb', line 125

def name
  @name
end

Instance Method Details

#add_instance_count!(value) ⇒ Object 



64
65
66
67
68
 
# File 'lib/pazuzu/worker.rb', line 64

def add_instance_count!(value)
  @dynamic_instance_count ||= @target_instance_count
  @dynamic_instance_count = [0, @dynamic_instance_count + value].max
  adjust_instances!
end

#adjust_instances! ⇒ Object 



70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
 
# File 'lib/pazuzu/worker.rb', line 70

def adjust_instances!
  target_count = @dynamic_instance_count
  target_count ||= @target_instance_count
  if @instances.length != target_count
    if run_state == :running
      @logger.info "Adjusting from #{@instances.length} to #{target_count} instances"
    end
    while @instances.length < target_count
      instance = Instance.new(self, @instances.length + 1, @root_path, @command_line)
      @instances.register(instance)
    end
    if @instances.length > target_count
      children_by_index = @instances.children.sort_by { |c| c.index }
      (children_by_index[target_count..-1] || []).each do |instance|
        @instances.unregister(instance)
      end
    end
  end
end

#configure!(root_path, command_line, configuration) ⇒ Object 



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
 
# File 'lib/pazuzu/worker.rb', line 18

def configure!(root_path, command_line, configuration)
  restart_required = false

  if [:starting, :running].include?(run_state) and root_path != @root_path
    restart_required = true
  end
  @root_path = root_path

  new_command_line = configuration['command_line']
  new_command_line ||= command_line
  new_command_line ||= @command_line
  if new_command_line != @command_line
    if [:starting, :running].include?(run_state)
      restart_required = true
    end
    @command_line = new_command_line
  end
  
  if restart_required
    @logger.warn "Root path or command line has changed, instances must be restarted/added manually"
  end

  new_target_instance_count = configuration['num_instances'].try(:to_i) || 1
  if new_target_instance_count != @target_instance_count
    @target_instance_count = new_target_instance_count
  end

  if [:starting, :running].include?(run_state)
    adjust_instances!
  end
end

#instances ⇒ Object 



114
115
116
 
# File 'lib/pazuzu/worker.rb', line 114

def instances
  @instances.children
end

#log_entries ⇒ Object 



118
119
120
121
122
123
 
# File 'lib/pazuzu/worker.rb', line 118

def log_entries
  entries = @instances.children.map(&:log_entries)
  entries.flatten!(1)
  entries.sort_by! { |(source, time, message)| time }
  entries
end

#qname ⇒ Object 



50
51
52
 
# File 'lib/pazuzu/worker.rb', line 50

def qname
  [@application.name, @name].join('.')
end

#recover_orphaned_instances! ⇒ Object

Go through available cgroups and find instances that are orphaned.



91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
 
# File 'lib/pazuzu/worker.rb', line 91

def recover_orphaned_instances!
  index = @instances.length + 1
  loop do
    instance = Instance.new(self, index, @root_path, @command_line)
    cgroup = @application.supervisor.cgroup_for_instance(instance)
    if cgroup.mounted?
      if instance.attachable?
        @logger.info "Discovered orphaned instance #{index}, attaching it"
        @instances.register(instance)
      end
      index += 1
    else
      break
    end
  end
end

#revert_dynamic_instance_count! ⇒ Object 



54
55
56
57
 
# File 'lib/pazuzu/worker.rb', line 54

def revert_dynamic_instance_count!
  @dynamic_instance_count = nil
  adjust_instances!
end

#set_instance_count!(value) ⇒ Object 



59
60
61
62
 
# File 'lib/pazuzu/worker.rb', line 59

def set_instance_count!(value)
  @dynamic_instance_count = [0, value].max
  adjust_instances!
end

#setup_spawned_process! ⇒ Object

Call to set up a forked process according to the application’s configuration.



110
111
112
 
# File 'lib/pazuzu/worker.rb', line 110

def setup_spawned_process!
  @application.setup_spawned_process!
end