Class: Fluent::Plugin::ForwardInput

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

Constant Summary collapse

LISTEN_PORT = 
24224
HEARTBEAT_UDP_PAYLOAD = 
"\0"

Constants included from Configurable

Configurable::CONFIG_TYPE_REGISTRY

Instance Attribute Summary

Attributes included from Fluent::PluginLoggerMixin

#log

Attributes inherited from Base

#under_plugin_development

Instance Method Summary collapse

Methods inherited from Input

#emit_records, #emit_size, #initialize, #metric_callback, #statistics

Methods included from Fluent::PluginHelper::Mixin

included

Methods included from Fluent::PluginLoggerMixin

included, #initialize, #terminate

Methods included from Fluent::PluginId

#initialize, #plugin_id, #plugin_id_configured?, #plugin_id_for_test?, #plugin_root_dir, #stop

Methods inherited from Base

#acquire_worker_lock, #after_shutdown, #after_shutdown?, #after_start, #after_started?, #before_shutdown, #before_shutdown?, #called_in_test?, #close, #closed?, #configured?, #context_router, #context_router=, #fluentd_worker_id, #get_lock_path, #has_router?, #initialize, #inspect, #plugin_root_dir, #reloadable_plugin?, #shutdown, #shutdown?, #started?, #stop, #stopped?, #string_safe_encoding, #terminate, #terminated?

Methods included from SystemConfig::Mixin

#system_config, #system_config_override

Methods included from Configurable

#config, #configure_proxy_generate, #configured_section_create, included, #initialize, lookup_type, register_type

Constructor Details

This class inherits a constructor from Fluent::Plugin::Input

Instance Method Details

#add_source_info(es, conn) ⇒ Object 



379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
 
# File 'lib/fluent/plugin/in_forward.rb', line 379

def add_source_info(es, conn)
  new_es = Fluent::MultiEventStream.new
  if @source_address_key && @source_hostname_key
    address = conn.remote_addr
    hostname = conn.remote_host
    es.each { |time, record|
      record[@source_address_key] = address
      record[@source_hostname_key] = hostname
      new_es.add(time, record)
    }
  elsif @source_address_key
    address = conn.remote_addr
    es.each { |time, record|
      record[@source_address_key] = address
      new_es.add(time, record)
    }
  elsif @source_hostname_key
    hostname = conn.remote_host
    es.each { |time, record|
      record[@source_hostname_key] = hostname
      new_es.add(time, record)
    }
  else
    raise "BUG: don't call this method in this case"
  end
  new_es
end

#check_and_skip_invalid_event(tag, es, remote_host) ⇒ Object 



367
368
369
370
371
372
373
374
375
376
377
 
# File 'lib/fluent/plugin/in_forward.rb', line 367

def check_and_skip_invalid_event(tag, es, remote_host)
  new_es = Fluent::MultiEventStream.new
  es.each { |time, record|
    if invalid_event?(tag, time, record)
      log.warn "skip invalid event:", host: remote_host, tag: tag, time: time, record: record
      next
    end
    new_es.add(time, record)
  }
  new_es
end

#check_ping(message, remote_addr, user_auth_salt, nonce) ⇒ Object 



425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
 
# File 'lib/fluent/plugin/in_forward.rb', line 425

def check_ping(message, remote_addr, user_auth_salt, nonce)
  log.debug "checking ping"
  # ['PING', self_hostname, shared_key_salt, sha512_hex(shared_key_salt + self_hostname + nonce + shared_key), username || '', sha512_hex(auth_salt + username + password) || '']
  unless message.size == 6 && message[0] == 'PING'
    return false, 'invalid ping message'
  end
  _ping, hostname, shared_key_salt, shared_key_hexdigest, username, password_digest = message

  node = @nodes.find{|n| n[:address].include?(remote_addr) rescue false }
  if !node && !@security.allow_anonymous_source
    log.warn "Anonymous client disallowed", address: remote_addr, hostname: hostname
    return false, "anonymous source host '#{remote_addr}' denied", nil
  end

  shared_key = node ? node[:shared_key] : @security.shared_key
  serverside = Digest::SHA512.new.update(shared_key_salt).update(hostname).update(nonce).update(shared_key).hexdigest
  if shared_key_hexdigest != serverside
    log.warn "Shared key mismatch", address: remote_addr, hostname: hostname
    return false, 'shared_key mismatch', nil
  end

  if @security.user_auth
    users = select_authenticate_users(node, username)
    success = false
    users.each do |user|
      passhash = Digest::SHA512.new.update(user_auth_salt).update(username).update(user[:password]).hexdigest
      success ||= (passhash == password_digest)
    end
    unless success
      log.warn "Authentication failed", address: remote_addr, hostname: hostname, username: username
      return false, 'username/password mismatch', nil
    end
  end

  return true, shared_key_salt, shared_key
end

#configure(conf) ⇒ Object

Raises:



101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
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
 
# File 'lib/fluent/plugin/in_forward.rb', line 101

def configure(conf)
  super

  if @source_hostname_key
    # TODO: add test
    if @resolve_hostname.nil?
      @resolve_hostname = true
    elsif !@resolve_hostname # user specifies "false" in config
      raise Fluent::ConfigError, "resolve_hostname must be true with source_hostname_key"
    end
  end
  @enable_field_injection = @source_address_key || @source_hostname_key

  raise Fluent::ConfigError, "'tag' parameter must not be empty" if @tag && @tag.empty?
  raise Fluent::ConfigError, "'add_tag_prefix' parameter must not be empty" if @add_tag_prefix && @add_tag_prefix.empty?

  if @security
    if @security.user_auth && @security.users.empty?
      raise Fluent::ConfigError, "<user> sections required if user_auth enabled"
    end
    if !@security.allow_anonymous_source && @security.clients.empty?
      raise Fluent::ConfigError, "<client> sections required if allow_anonymous_source disabled"
    end

    @nodes = []

    @security.clients.each do |client|
      if client.host && client.network
        raise Fluent::ConfigError, "both of 'host' and 'network' are specified for client"
      end
      if !client.host && !client.network
        raise Fluent::ConfigError, "Either of 'host' and 'network' must be specified for client"
      end
      source = nil
      if client.host
        begin
          source = IPSocket.getaddress(client.host)
        rescue SocketError
          raise Fluent::ConfigError, "host '#{client.host}' cannot be resolved"
        end
      end
      source_addr = begin
                      IPAddr.new(source || client.network)
                    rescue ArgumentError
                      raise Fluent::ConfigError, "network '#{client.network}' address format is invalid"
                    end
      @nodes.push({
          address: source_addr,
          shared_key: (client.shared_key || @security.shared_key),
          users: client.users
        })
    end
  end

  if @send_keepalive_packet && @deny_keepalive
    raise Fluent::ConfigError, "both 'send_keepalive_packet' and 'deny_keepalive' cannot be set to true"
  end
end

#generate_helo(nonce, user_auth_salt) ⇒ Object 



419
420
421
422
423
 
# File 'lib/fluent/plugin/in_forward.rb', line 419

def generate_helo(nonce, user_auth_salt)
  log.debug "generating helo"
  # ['HELO', options(hash)]
  ['HELO', {'nonce' => nonce, 'auth' => (@security ? user_auth_salt : ''), 'keepalive' => !@deny_keepalive}]
end

#generate_pong(auth_result, reason_or_salt, nonce, shared_key) ⇒ Object 



462
463
464
465
466
467
468
469
470
471
 
# File 'lib/fluent/plugin/in_forward.rb', line 462

def generate_pong(auth_result, reason_or_salt, nonce, shared_key)
  log.debug "generating pong"
  # ['PONG', bool(authentication result), 'reason if authentication failed', self_hostname, sha512_hex(salt + self_hostname + nonce + sharedkey)]
  unless auth_result
    return ['PONG', false, reason_or_salt, '', '']
  end

  shared_key_digest_hex = Digest::SHA512.new.update(reason_or_salt).update(@security.self_hostname).update(nonce).update(shared_key).hexdigest
  ['PONG', true, '', @security.self_hostname, shared_key_digest_hex]
end

#generate_salt ⇒ Object 



415
416
417
 
# File 'lib/fluent/plugin/in_forward.rb', line 415

def generate_salt
  ::SecureRandom.random_bytes(16)
end

#handle_connection(conn) ⇒ Object 



193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
 
# File 'lib/fluent/plugin/in_forward.rb', line 193

def handle_connection(conn)
  send_data = ->(serializer, data){ conn.write serializer.call(data) }

  log.trace "connected fluent socket", addr: conn.remote_addr, port: conn.remote_port
  state = :established
  nonce = nil
  user_auth_salt = nil

  if @security
    # security enabled session MUST use MessagePack as serialization format
    state = :helo
    nonce = generate_salt
    user_auth_salt = generate_salt
    send_data.call(:to_msgpack.to_proc, generate_helo(nonce, user_auth_salt))
    state = :pingpong
  end

  log.trace "accepted fluent socket", addr: conn.remote_addr, port: conn.remote_port

  read_messages(conn) do |msg, chunk_size, serializer|
    case state
    when :pingpong
      success, reason_or_salt, shared_key = check_ping(msg, conn.remote_addr, user_auth_salt, nonce)
      unless success
        conn.on(:write_complete) { |c| c.close_after_write_complete }
        send_data.call(serializer, generate_pong(false, reason_or_salt, nonce, shared_key))
        next
      end
      send_data.call(serializer, generate_pong(true, reason_or_salt, nonce, shared_key))

      log.debug "connection established", address: conn.remote_addr, port: conn.remote_port
      state = :established
    when :established
      options = on_message(msg, chunk_size, conn)
      if options && r = response(options)
        log.trace "sent response to fluent socket", address: conn.remote_addr, response: r
        conn.on(:write_complete) { |c| c.close } if @deny_keepalive
        send_data.call(serializer, r)
      else
        if @deny_keepalive
          conn.close
        end
      end
    else
      raise "BUG: unknown session state: #{state}"
    end
  end
end

#invalid_event?(tag, time, record) ⇒ Boolean

Returns:

  • (Boolean) 


363
364
365
 
# File 'lib/fluent/plugin/in_forward.rb', line 363

def invalid_event?(tag, time, record)
  !((time.is_a?(Integer) || time.is_a?(::Fluent::EventTime)) && record.is_a?(Hash) && tag.is_a?(String))
end

#multi_workers_ready? ⇒ Boolean

Returns:

  • (Boolean) 


160
161
162
 
# File 'lib/fluent/plugin/in_forward.rb', line 160

def multi_workers_ready?
  true
end

#on_message(msg, chunk_size, conn) ⇒ Object 



282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
 
# File 'lib/fluent/plugin/in_forward.rb', line 282

def on_message(msg, chunk_size, conn)
  if msg.nil?
    # for future TCP heartbeat_request
    return
  end

  # TODO: raise an exception if broken chunk is generated by recoverable situation
  unless msg.is_a?(Array)
    log.warn "incoming chunk is broken:", host: conn.remote_host, msg: msg
    return
  end

  tag = msg[0]
  entries = msg[1]

  if @chunk_size_limit && (chunk_size > @chunk_size_limit)
    log.warn "Input chunk size is larger than 'chunk_size_limit', dropped:", tag: tag, host: conn.remote_host, limit: @chunk_size_limit, size: chunk_size
    return
  elsif @chunk_size_warn_limit && (chunk_size > @chunk_size_warn_limit)
    log.warn "Input chunk size is larger than 'chunk_size_warn_limit':", tag: tag, host: conn.remote_host, limit: @chunk_size_warn_limit, size: chunk_size
  end

  tag = @tag.dup if @tag
  tag = "#{@add_tag_prefix}.#{tag}" if @add_tag_prefix

  case entries
  when String
    # PackedForward
    option = msg[2]
    size = (option && option['size']) || 0
    es_class = (option && option['compressed'] == 'gzip') ? Fluent::CompressedMessagePackEventStream : Fluent::MessagePackEventStream
    es = es_class.new(entries, nil, size.to_i)
    es = check_and_skip_invalid_event(tag, es, conn.remote_host) if @skip_invalid_event
    if @enable_field_injection
      es = add_source_info(es, conn)
    end
    router.emit_stream(tag, es)

  when Array
    # Forward
    es = if @skip_invalid_event
           check_and_skip_invalid_event(tag, entries, conn.remote_host)
         else
           es = Fluent::MultiEventStream.new
           entries.each { |e|
             record = e[1]
             next if record.nil?
             time = e[0]
             time = Fluent::EventTime.now if time.nil? || time.to_i == 0 # `to_i == 0` for empty EventTime
             es.add(time, record)
           }
           es
         end
    if @enable_field_injection
      es = add_source_info(es, conn)
    end
    router.emit_stream(tag, es)
    option = msg[2]

  else
    # Message
    time = msg[1]
    record = msg[2]
    if @skip_invalid_event && invalid_event?(tag, time, record)
      log.warn "got invalid event and drop it:", host: conn.remote_host, tag: tag, time: time, record: record
      return msg[3] # retry never succeeded so return ack and drop incoming event.
    end
    return if record.nil?
    time = Fluent::EventTime.now if time.to_i == 0
    if @enable_field_injection
      record[@source_address_key] = conn.remote_addr if @source_address_key
      record[@source_hostname_key] = conn.remote_host if @source_hostname_key
    end
    router.emit(tag, time, record)
    option = msg[3]
  end

  # return option for response
  option
end

#read_messages(conn, &block) ⇒ Object 



242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
 
# File 'lib/fluent/plugin/in_forward.rb', line 242

def read_messages(conn, &block)
  feeder = nil
  serializer = nil
  bytes = 0
  conn.data do |data|
    # only for first call of callback
    unless feeder
      first = data[0]
      if first == '{' || first == '[' # json
        parser = Yajl::Parser.new
        parser.on_parse_complete = ->(obj){
          block.call(obj, bytes, serializer)
          bytes = 0
        }
        serializer = :to_json.to_proc
        feeder = ->(d){ parser << d }
      else # msgpack
        parser = Fluent::MessagePackFactory.msgpack_unpacker
        serializer = :to_msgpack.to_proc
        feeder = ->(d){
          parser.feed_each(d){|obj|
            block.call(obj, bytes, serializer)
            bytes = 0
          }
        }
      end
    end

    bytes += data.bytesize
    feeder.call(data)
  end
end

#response(option) ⇒ Object 



275
276
277
278
279
280
 
# File 'lib/fluent/plugin/in_forward.rb', line 275

def response(option)
  if option && option['chunk']
    return { 'ack' => option['chunk'] }
  end
  nil
end

#select_authenticate_users(node, username) ⇒ Object 



407
408
409
410
411
412
413
 
# File 'lib/fluent/plugin/in_forward.rb', line 407

def select_authenticate_users(node, username)
  if node.nil? || node[:users].empty?
    @security.users.select{|u| u.username == username}
  else
    @security.users.select{|u| node[:users].include?(u.username) && u.username == username}
  end
end

#start ⇒ Object 



166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
 
# File 'lib/fluent/plugin/in_forward.rb', line 166

def start
  super

  shared_socket = system_config.workers > 1

  log.info "listening port", port: @port, bind: @bind
  server_create_connection(
    :in_forward_server, @port,
    bind: @bind,
    shared: shared_socket,
    resolve_name: @resolve_hostname,
    linger_timeout: @linger_timeout,
    send_keepalive_packet: @send_keepalive_packet,
    backlog: @backlog,
    &method(:handle_connection)
  )

  server_create(:in_forward_server_udp_heartbeat, @port, shared: shared_socket, proto: :udp, bind: @bind, resolve_name: @resolve_hostname, max_bytes: 128) do |data, sock|
    log.trace "heartbeat udp data arrived", host: sock.remote_host, port: sock.remote_port, data: data
    begin
      sock.write HEARTBEAT_UDP_PAYLOAD
    rescue Errno::EAGAIN, Errno::EWOULDBLOCK, Errno::EINTR
      log.trace "error while heartbeat response", host: sock.remote_host, error: e
    end
  end
end