Class: Fluent::ProcCountInput

Inherits:
Input
  • Object
show all
Includes:
Sys
Defined in:
lib/fluent/plugin/in_proc_count.rb

Instance Method Summary collapse

Constructor Details

#initialize ⇒ ProcCountInput

Returns a new instance of ProcCountInput.



10
11
12
 
# File 'lib/fluent/plugin/in_proc_count.rb', line 10

def initialize
  super
end

Instance Method Details

#configure(conf) ⇒ Object 



23
24
25
26
27
28
29
30
31
32
33
34
 
# File 'lib/fluent/plugin/in_proc_count.rb', line 23

def configure(conf)
  super
  @processes = @processes.map { |process|
    s = OpenStruct.new
    s.regexp = Regexp.new(process.regexp)
    s.proc_count = process.proc_count
    s.tag = process.tag
    s.operator_name = process.operator_name
    s.operator = select_operator(process.operator_name)
    s
  }
end

#correct_process?(process_spec, expect_proc_count) ⇒ Boolean

Returns:

  • (Boolean) 


93
94
95
 
# File 'lib/fluent/plugin/in_proc_count.rb', line 93

def correct_process?(process_spec, expect_proc_count)
  expect_proc_count.public_send(process_spec.operator, process_spec.proc_count.to_i)
end

#emit_proc_count ⇒ Object 



53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
 
# File 'lib/fluent/plugin/in_proc_count.rb', line 53

def emit_proc_count
  @processes.each do |process_spec|
    begin
      record_size = get_processes(process_spec).size
      if !correct_process?(process_spec, record_size)
        router.emit(
          process_spec.tag,
          Fluent::Engine.now,
          {
            regexp: process_spec.regexp.source,
            proc_count: record_size,
            expect_proc_count: process_spec.proc_count,
            operator_name: process_spec.operator_name,
            hostname: Socket.gethostname
          }
        )
      end
    rescue => e
      log.error e
    end
  end
end

#get_processes(process_spec) ⇒ Object 



97
98
99
100
101
102
103
104
105
 
# File 'lib/fluent/plugin/in_proc_count.rb', line 97

def get_processes(process_spec)
  processes = []
  ProcTable.ps do |process|
    if process_spec.regexp.match(process.cmdline)
      processes << process
    end
  end
  processes
end

#run ⇒ Object 



46
47
48
49
50
51
 
# File 'lib/fluent/plugin/in_proc_count.rb', line 46

def run
  loop do
    Thread.new(&method(:emit_proc_count))
    sleep @interval
  end
end

#select_operator(operator_name) ⇒ Object 



76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
 
# File 'lib/fluent/plugin/in_proc_count.rb', line 76

def select_operator(operator_name)
  case operator_name.to_sym
  when :equal
    "=="
  when :gather_than
    ">"
  when :gather_equal
    ">="
  when :less_than
    "<"
  when :less_equal
    "<="
  else
    raise Fluent::ConfigError, "proc_count operator allows equal/gather_than/gather_equal/less_than/less_equal"
  end
end

#shutdown ⇒ Object 



41
42
43
44
 
# File 'lib/fluent/plugin/in_proc_count.rb', line 41

def shutdown
  super
  Thread.kill(@thread)
end

#start ⇒ Object 



36
37
38
39
 
# File 'lib/fluent/plugin/in_proc_count.rb', line 36

def start
  super
  @thread = Thread.new(&method(:run))
end