Class: Fluent::PingMessageCheckerOutput

Inherits:
Output
  • Object
show all
Includes:
Mixin::ConfigPlaceholders
Defined in:
lib/fluent/plugin/out_ping_message_checker.rb

Instance Method Summary collapse

Instance Method Details

#check_and_flush ⇒ Object 



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
 
# File 'lib/fluent/plugin/out_ping_message_checker.rb', line 61

def check_and_flush
  notifications = []

  @mutex.synchronize do
    @checks.keys.each do |key|
      if @checks[key] == 0
        @checks[key] = -1

      elsif @checks[key] < 0
        notifications.push(key)
        @checks[key] = 1

      else # @checks[key] > 0
        if @checks[key] < @notification_times
          notifications.push(key)
          @checks[key] += 1
        else
          @checks.delete(key)
        end
      end
    end
  end
  
  if @notifications
    notifications.each do |data|
      Fluent::Engine.emit(@tag, Fluent::Engine.now, {@data_field => data})
    end
  end

  notifications
end

#configure(conf) ⇒ Object 



25
26
27
28
 
# File 'lib/fluent/plugin/out_ping_message_checker.rb', line 25

def configure(conf)
  super
  @exclude_regex = @exclude_pattern ? Regexp.compile(@exclude_pattern) : nil
end

#emit(tag, es, chain) ⇒ Object 



108
109
110
111
112
113
114
115
116
117
118
119
 
# File 'lib/fluent/plugin/out_ping_message_checker.rb', line 108

def emit(tag, es, chain)
  datalist = []
  es.each do |time,record|
    datalist.push record[@data_field] if @exclude_regex.nil? or not @exclude_regex.match(record[@data_field])
  end
  datalist.uniq!
  update_state(datalist)

  chain.next
rescue => e
  log.warn "out_ping_message_checker: #{e.message} #{e.class} #{e.backtrace.first}"
end

#shutdown ⇒ Object 



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

def shutdown
  super
  @watcher.terminate
  @watcher.join
end

#start ⇒ Object 



30
31
32
33
34
35
36
37
38
39
 
# File 'lib/fluent/plugin/out_ping_message_checker.rb', line 30

def start
  super
  @checks = {}
  # 'data' => notification_counts
  # -1: checked in previous term, but not in this term
  #  0: checked in this term
  # 1,2,...: counts of ping missing notifications
  @mutex = Mutex.new
  start_watch
end

#start_watch ⇒ Object 



47
48
49
 
# File 'lib/fluent/plugin/out_ping_message_checker.rb', line 47

def start_watch
  @watcher = Thread.new(&method(:watch))
end

#update_state(list) ⇒ Object 



51
52
53
54
55
56
57
58
59
 
# File 'lib/fluent/plugin/out_ping_message_checker.rb', line 51

def update_state(list)
  @mutex.synchronize do
    list.each do |data|
      if not @checks.has_key?(data) or @checks[data] != 0
        @checks[data] = 0
      end
    end
  end
end

#watch ⇒ Object 



93
94
95
96
97
98
99
100
101
102
103
104
105
106
 
# File 'lib/fluent/plugin/out_ping_message_checker.rb', line 93

def watch
  @last_checked = Fluent::Engine.now
  loop do
    sleep 1
    begin
      if Fluent::Engine.now - @last_checked >= @check_interval
        check_and_flush()
        @last_checked = Fluent::Engine.now
      end
    rescue => e
      log.warn "out_ping_message_checker: #{e.class} #{e.message} #{e.backtrace.first}"
    end
  end
end