github-mirrors/logstash
1
0
Fork 0
mirror of https://github.com/elastic/logstash.git synced 2025-04-24 06:37:19 -04:00
Code Issues Projects Releases Packages Wiki Activity

Make tcp input and output act as either server or client.

Browse source
This commit is contained in:
Dan Peterson 2011-07-12 22:59:58 -03:00
parent fc1ba8baaf
commit ac492053f0
2 changed files with 130 additions and 58 deletions
Download patch file Download diff file Expand all files Collapse all files

112
lib/logstash/inputs/tcp.rb
View file

@ -12,61 +12,89 @@ class LogStash::Inputs::Tcp < LogStash::Inputs::Base
# The address to listen on
config :host, :validate => :string, :default => "0.0.0.0"
# the port to listen on
config :port, :validate => :number, :required => true
# Read timeout in seconds. If a particular tcp connection is
# idle for more than this timeout period, we will assume
# idle for more than this timeout period, we will assume
# it is dead and close it.
# If you never want to timeout, use -1.
config :data_timeout, :validate => :number, :default => 5
# Enable server.
config :server, :validate => :boolean, :default => true
module SocketPeer
def peer
"#{peeraddr[3]}:#{peeraddr[1]}"
end
end
public
def register
@logger.info("Starting tcp listener on #{@host}:#{@port}")
@server = TCPServer.new(@host, @port)
if @server
@logger.info("Starting tcp listener on #{@host}:#{@port}")
@server = TCPServer.new(@host, @port)
else
@socket = nil
end
end # def register
def handle_socket(socket, output_queue, event_source)
begin
loop do
buf = nil
# NOTE(petef): the timeout only hits after the line is read
# or socket dies
# TODO(sissel): Why do we have a timeout here? What's the point?
if @data_timeout == -1
buf = socket.readline
else
Timeout::timeout(@data_timeout) do
buf = socket.readline
end
end
e = self.to_event(buf, event_source)
if e
output_queue << e
end
end # loop do
rescue => e
@logger.debug(["Closing connection with #{socket.peer}", $!])
@logger.debug(["Backtrace", e.backtrace])
rescue Timeout::Error
@logger.debug("Closing connection with #{socket.peer} after read timeout")
end # begin
begin
socket.close
rescue IOError
pass
end # begin
end
public
def run(output_queue)
loop do
# Start a new thread for each connection.
Thread.start(@server.accept) do |s|
# TODO(sissel): put this block in its own method.
peer = "#{s.peeraddr[3]}:#{s.peeraddr[1]}"
@logger.debug("Accepted connection from #{peer} on #{@host}:#{@port}")
begin
loop do
buf = nil
# NOTE(petef): the timeout only hits after the line is read
# or socket dies
# TODO(sissel): Why do we have a timeout here? What's the point?
if @data_timeout == -1
buf = s.readline
else
Timeout::timeout(@data_timeout) do
buf = s.readline
end
end
e = self.to_event(buf, "tcp://#{@host}:#{@port}/client/#{peer}")
if e
output_queue << e
end
end # loop do
rescue => e
@logger.debug(["Closing connection with #{peer}", $!])
@logger.debug(["Backtrace", e.backtrace])
rescue Timeout::Error
@logger.debug("Closing connection with #{peer} after read timeout")
end # begin
begin
s.close
rescue IOError
pass
end # begin
end # Thread.start
end # loop (outer)
if @server
loop do
# Start a new thread for each connection.
Thread.start(@server.accept) do |s|
# TODO(sissel): put this block in its own method.
s.instance_eval { class << self; include SocketPeer end }
@logger.debug("Accepted connection from #{s.peer} on #{@host}:#{@port}")
handle_socket(s, output_queue, "tcp://#{@host}:#{@port}/client/#{s.peer}")
end # Thread.start
end # loop (outer)
else
Thread.start do
loop do
socket = TCPSocket.new(@host, @port)
socket.instance_eval { class << self; include SocketPeer end }
@logger.debug("Opened connection to #{socket.peer}")
handle_socket(socket, output_queue, "tcp://#{socket.peer}/server")
end
end
end
end # def run
end # class LogStash::Inputs::Tcp

76
lib/logstash/outputs/tcp.rb
View file

@ -1,7 +1,8 @@
require "logstash/outputs/base"
require "logstash/namespace"
require "thread"
# This output writes each event in json format to
# This output writes each event in json format to
# the specified host:port over tcp.
#
# Each event json is separated by a newline.
@ -9,20 +10,53 @@ class LogStash::Outputs::Tcp < LogStash::Outputs::Base
config_name "tcp"
# The host to connect to
# The host to connect or bind to
config :host, :validate => :string, :required => true
# The port to connect to
# The port to connect or bind to
config :port, :validate => :number, :required => true
public
def initialize(params)
super
end # def initialize
# Enable server.
config :server, :validate => :boolean
class Client < Thread
def initialize(socket)
@socket = socket
@queue = Queue.new
super do
loop do
begin
@socket.write(@queue.pop)
rescue => e
@logger.warn(["tcp output exception", @socket, $!])
@logger.debug(["backtrace", e.backtrace])
break
end
end
end
end
def write(msg)
@queue.push(msg)
end
end
public
def register
@socket = nil
if @server
@logger.info("Starting tcp output listener on #{@host}:#{@port}")
@server_socket = TCPServer.new(@host, @port)
@client_threads = []
@accept_thread = Thread.new(@server_socket) do |server|
loop do
@client_threads << Client.new(server.accept)
end
end
else
@socket = nil
end
end # def register
private
@ -32,14 +66,24 @@ class LogStash::Outputs::Tcp < LogStash::Outputs::Base
public
def receive(event)
begin
connect unless @socket
@socket.write(event.to_hash.to_json)
@socket.write("\n")
rescue => e
@logger.warn(["tcp output exception", @host, @port, $!])
@logger.debug(["backtrace", e.backtrace])
@socket = nil
wire_event = event.to_hash.to_json + "\n"
if @server
@client_threads.each do |client_thread|
client_thread.write(wire_event)
end
@client_threads.reject! {|t| !t.alive? }
else
begin
connect unless @socket
@socket.write(event.to_hash.to_json)
@socket.write("\n")
rescue => e
@logger.warn(["tcp output exception", @host, @port, $!])
@logger.debug(["backtrace", e.backtrace])
@socket = nil
end
end
end # def receive
end # class LogStash::Outputs::Tcp