Class: Lumberjack::Beats::Connection

Inherits:
Object
  • Object
show all
Defined in:
lib/lumberjack/beats/server.rb

Overview

class Parser

Defined Under Namespace

Classes: ConnectionClosed

Constant Summary collapse

READ_SIZE = 
16384
PEER_INFORMATION_NOT_AVAILABLE = 
"<PEER INFORMATION NOT AVAILABLE>"
RESCUED_CONNECTION_EXCEPTIONS = 
[
  EOFError,
  OpenSSL::SSL::SSLError,
  IOError,
  Errno::ECONNRESET,
  Errno::EPIPE,
  Lumberjack::Beats::Parser::UnsupportedProtocol
]

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(fd, server) ⇒ Connection

Returns a new instance of Connection.



422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
 
# File 'lib/lumberjack/beats/server.rb', line 422

def initialize(fd, server)
  @parser = Parser.new
  @fd = fd

  @server = server
  @ack_handler = nil

  # Fetch the details of the host before reading anything from the socket
  # se we can use that information when debugging connection issues with
  # remote hosts.
  begin
    @peer = "#{@fd.peeraddr[3]}:#{@fd.peeraddr[1]}"
  rescue IOError
    # This can happen if the connection is drop or close before
    # fetching the host details, lets return a generic string.
    @peer = PEER_INFORMATION_NOT_AVAILABLE
  end
end

Instance Attribute Details

#peer ⇒ Object (readonly)

Returns the value of attribute peer.



420
421
422
 
# File 'lib/lumberjack/beats/server.rb', line 420

def peer
  @peer
end

#server ⇒ Object

Returns the value of attribute server.



419
420
421
 
# File 'lib/lumberjack/beats/server.rb', line 419

def server
  @server
end

Instance Method Details

#ack_if_needed(sequence, &block) ⇒ Object 



502
503
504
505
 
# File 'lib/lumberjack/beats/server.rb', line 502

def ack_if_needed(sequence, &block)
  block.call
  send_ack(sequence) if @ack_handler.ack?(sequence)
end

#close ⇒ Object 



507
508
509
 
# File 'lib/lumberjack/beats/server.rb', line 507

def close
  @fd.close unless @fd.closed?
end

#data(map, &block) ⇒ Object 



511
512
513
 
# File 'lib/lumberjack/beats/server.rb', line 511

def data(map, &block)
  block.call(map, identity_stream(map)) if block_given?
end

#identity_stream(map) ⇒ Object 



528
529
530
531
532
533
534
535
536
537
538
539
 
# File 'lib/lumberjack/beats/server.rb', line 528

def identity_stream(map)
  id = map.fetch("beat", {})["id"]

  if id && map["resource_id"]
    identity_values = [id, map["resource_id"]]
  else
    identity_values = [map.fetch("beat", {})["name"],
                      map["source"]]
  end

  identity_values.compact.join("-")
end

#normalize_v1_metadata_encoding(map) ⇒ Object 



489
490
491
492
493
494
495
496
 
# File 'lib/lumberjack/beats/server.rb', line 489

def normalize_v1_metadata_encoding(map)
  # lets normalize the metadata of the v1 frame to make
  # sure everything is in utf-8 format, because LSF don't enforce the encoding when he send
  # the data to the server. Path, offset can be in another encoding, when the data is assigned to the event.
  # the event will validate it and crash when the encoding is in the wrong format.
  map.each { |k, v| map[k].force_encoding(Encoding::UTF_8) unless k == Lumberjack::Beats::LSF_LOG_LINE_FIELD }
  map
end

#read_socket(&block) ⇒ Object

def run



457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
 
# File 'lib/lumberjack/beats/server.rb', line 457

def read_socket(&block)
  # TODO(sissel): Ack on idle.
  # X: - if any unacked, IO.select
  # X:   - on timeout, ack all.
  # X: Doing so will prevent slow streams from retransmitting
  # X: too many events after errors.
  @parser.feed(@fd.sysread(READ_SIZE)) do |event, *args|
    case event
    when :version
      version(*args)
    when :window_size
      reset_next_ack(*args)
    when :data
      sequence, map = args
      ack_if_needed(sequence) { data(normalize_v1_metadata_encoding(map), &block) }
    when :json
      # If the payload is an array of items we will emit multiple events
      # this behavior was moved from the plugin to the library.
      # see this commit: https://github.com/logstash-plugins/logstash-input-lumberjack/pull/57/files#diff-1b9590423b15f04f215635164e7376ecR158
      sequence, map = args

      ack_if_needed(sequence) do
        if map.is_a?(Array)
          map.each { |e| data(e, &block) }
        else
          data(map, &block)
        end
      end
    end
  end
end

#reset_next_ack(window_size) ⇒ Object 



515
516
517
518
 
# File 'lib/lumberjack/beats/server.rb', line 515

def reset_next_ack(window_size)
  klass = version_1? ? AckingProtocolV1 : AckingProtocolV2
  @ack_handler = klass.new(window_size)
end

#run(&block) ⇒ Object 



441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
 
# File 'lib/lumberjack/beats/server.rb', line 441

def run(&block)
  while !server.closed?
    read_socket(&block)
  end
rescue *RESCUED_CONNECTION_EXCEPTIONS => e
  # EOF or other read errors, only action is to shutdown which we'll do in
  # 'ensure'
  raise ConnectionClosed.new(e)
rescue
  # when the server is shutting down we can safely ignore any exceptions
  # On windows, we can get a `SystemCallErr`
  raise unless server.closed?
ensure
  close rescue 'Already closed stream'
end

#send_ack(sequence) ⇒ Object 



520
521
522
 
# File 'lib/lumberjack/beats/server.rb', line 520

def send_ack(sequence)
  @fd.syswrite(@ack_handler.ack_frame(sequence))
end

#version(version) ⇒ Object 



498
499
500
 
# File 'lib/lumberjack/beats/server.rb', line 498

def version(version)
  @version = version
end

#version_1? ⇒ Boolean

Returns:

  • (Boolean) 


524
525
526
 
# File 'lib/lumberjack/beats/server.rb', line 524

def version_1?
  @version == Parser::PROTOCOL_VERSION_1
end