Class: LogStash::Inputs::Tcp
- Inherits:
-
Base
- Object
- Base
- LogStash::Inputs::Tcp
- Defined in:
- lib/logstash/inputs/tcp.rb
Overview
Read events over a TCP socket.
Like stdin and file inputs, each event is assumed to be one line of text.
Can either accept connections from clients or connect to a server, depending on ‘mode`.
#### Accepting log4j2 logs
Log4j2 can send JSON over a socket, and we can use that combined with our tcp input to accept the logs.
First, we need to configure your application to send logs in JSON over a socket. The following log4j2.xml accomplishes this task.
Note, you will want to change the ‘host` and `port` settings in this configuration to match your needs.
<Configuration>
<Appenders>
<Socket name="Socket" host="localhost" port="12345">
<JsonLayout compact="true" eventEol="true" />
</Socket>
</Appenders>
<Loggers>
<Root level="info">
<AppenderRef ref="Socket"/>
</Root>
</Loggers>
</Configuration>
To accept this in Logstash, you will want tcp input and a date filter:
input {
tcp {
port => 12345
codec => json
}
}
and add a date filter to take log4j2’s ‘timeMillis` field and use it as the event timestamp
filter {
date {
match => [ "timeMillis", "UNIX_MS" ]
}
}
Constant Summary collapse
- HOST_FIELD =
"host".freeze
- HOST_IP_FIELD =
"[@metadata][ip_address]".freeze
- PORT_FIELD =
"port".freeze
- PROXY_HOST_FIELD =
"proxy_host".freeze
- PROXY_PORT_FIELD =
"proxy_port".freeze
- SSLSUBJECT_FIELD =
"sslsubject".freeze
Instance Method Summary collapse
- #close ⇒ Object
- #decode_buffer(client_ip_address, client_address, client_port, codec, proxy_address, proxy_port, tbuf, socket) ⇒ Object
- #dns_reverse_lookup_enabled? ⇒ Boolean
- #flush_codec(codec, client_ip_address, client_address, client_port, socket) ⇒ Object
-
#initialize(*args) ⇒ Tcp
constructor
A new instance of Tcp.
- #register ⇒ Object
- #run(output_queue) ⇒ Object
- #stop ⇒ Object
Constructor Details
#initialize(*args) ⇒ Tcp
Returns a new instance of Tcp.
122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 |
# File 'lib/logstash/inputs/tcp.rb', line 122 def initialize(*args) super(*args) # monkey patch TCPSocket and SSLSocket to include socket peer TCPSocket.module_eval{include ::LogStash::Util::SocketPeer} OpenSSL::SSL::SSLSocket.module_eval{include ::LogStash::Util::SocketPeer} # threadsafe socket bookkeeping @server_socket = nil @client_socket = nil @connection_sockets = {} @socket_mutex = Mutex.new @ssl_context = nil end |
Instance Method Details
#close ⇒ Object
169 170 171 172 173 174 175 176 |
# File 'lib/logstash/inputs/tcp.rb', line 169 def close # see related comment in register: we must make sure to close the server socket here # because it is created in the register method and we could be in the context of having # register called but never run & stop, only close. # catch all rescue nil on close to discard any close errors or invalid socket server_socket.close rescue nil @loop.close rescue nil end |
#decode_buffer(client_ip_address, client_address, client_port, codec, proxy_address, proxy_port, tbuf, socket) ⇒ Object
178 179 180 181 182 183 184 185 186 187 |
# File 'lib/logstash/inputs/tcp.rb', line 178 def decode_buffer(client_ip_address, client_address, client_port, codec, proxy_address, proxy_port, tbuf, socket) codec.decode(tbuf) do |event| if @proxy_protocol event.set(PROXY_HOST_FIELD, proxy_address) unless event.get(PROXY_HOST_FIELD) event.set(PROXY_PORT_FIELD, proxy_port) unless event.get(PROXY_PORT_FIELD) end enqueue_decorated(event, client_ip_address, client_address, client_port, socket) end end |
#dns_reverse_lookup_enabled? ⇒ Boolean
195 196 197 |
# File 'lib/logstash/inputs/tcp.rb', line 195 def dns_reverse_lookup_enabled? @dns_reverse_lookup_enabled end |
#flush_codec(codec, client_ip_address, client_address, client_port, socket) ⇒ Object
189 190 191 192 193 |
# File 'lib/logstash/inputs/tcp.rb', line 189 def flush_codec(codec, client_ip_address, client_address, client_port, socket) codec.flush do |event| enqueue_decorated(event, client_ip_address, client_address, client_port, socket) end end |
#register ⇒ Object
138 139 140 141 142 143 144 145 146 147 |
# File 'lib/logstash/inputs/tcp.rb', line 138 def register fix_streaming_codecs if server? ssl_context = get_ssl_context(SslOptions) @loop = InputLoop.new(@host, @port, DecoderImpl.new(@codec, self), @tcp_keep_alive, ssl_context, @logger.to_java(org.apache.logging.log4j.Logger)) end end |
#run(output_queue) ⇒ Object
149 150 151 152 153 154 155 156 157 |
# File 'lib/logstash/inputs/tcp.rb', line 149 def run(output_queue) @output_queue = output_queue if server? @logger.info("Starting tcp input listener", :address => "#{@host}:#{@port}", :ssl_enable => "#{@ssl_enable}") @loop.run else run_client() end end |
#stop ⇒ Object
159 160 161 162 163 164 165 166 167 |
# File 'lib/logstash/inputs/tcp.rb', line 159 def stop # force close all sockets which will escape any blocking read with a IO exception # and any thread using them will exit. # catch all rescue nil on close to discard any close errors or invalid socket server_socket.close rescue nil @loop.close rescue nil client_socket.close rescue nil connection_sockets.each{|socket| socket.close rescue nil} end |