Class: Fluent::ScribeInput::FluentScribeHandler

Inherits:
Object
  • Object
show all
Defined in:
lib/fluent/plugin/in_scribe.rb

Instance Attribute Summary collapse

Instance Method Summary collapse

Instance Attribute Details

#add_prefix ⇒ Object

Returns the value of attribute add_prefix.



127
128
129
 
# File 'lib/fluent/plugin/in_scribe.rb', line 127

def add_prefix
  @add_prefix
end

#ignore_invalid_record ⇒ Object

Returns the value of attribute ignore_invalid_record.



130
131
132
 
# File 'lib/fluent/plugin/in_scribe.rb', line 130

def ignore_invalid_record
  @ignore_invalid_record
end

#logger ⇒ Object

Use logger instead of log to avoid confusion with Log method



131
132
133
 
# File 'lib/fluent/plugin/in_scribe.rb', line 131

def logger
  @logger
end

#msg_format ⇒ Object

Returns the value of attribute msg_format.



129
130
131
 
# File 'lib/fluent/plugin/in_scribe.rb', line 129

def msg_format
  @msg_format
end

#remove_newline ⇒ Object

Returns the value of attribute remove_newline.



128
129
130
 
# File 'lib/fluent/plugin/in_scribe.rb', line 128

def remove_newline
  @remove_newline
end

#router ⇒ Object

Returns the value of attribute router.



132
133
134
 
# File 'lib/fluent/plugin/in_scribe.rb', line 132

def router
  @router
end

Instance Method Details

#Log(msgs) ⇒ Object 



134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
 
# File 'lib/fluent/plugin/in_scribe.rb', line 134

def Log(msgs)
  bucket = {} # tag -> events(array of [time,record])
  time_now = Engine.now
  begin
    msgs.each do |msg|
      begin
        record = create_record(msg)
      rescue => e
        if @ignore_invalid_record
          # This warning can be disabled by 'log_level error'
          logger.warn "got invalid record", message: msg, error_class: e.class, error: e
          next
        end

        raise
      end
      tag = @add_prefix ? @add_prefix + '.' + msg.category : msg.category
      bucket[tag] ||= []
      bucket[tag].push([time_now,record])
    end
  rescue => e
    logger.error "unexpected error", error_class: e.class, error: e
    logger.error_backtrace
    return ResultCode::TRY_LATER
  end

  begin
    bucket.each do |tag,events|
      router.emit_array(tag, events)
    end
    return ResultCode::OK
  rescue => e
    logger.error "unexpected error", error_class: e.class, error: e
    logger.error_backtrace
    return ResultCode::TRY_LATER
  end
end