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
- 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
-
#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.
108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 |
# File 'lib/logstash/inputs/tcp.rb', line 108 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
151 152 153 154 155 156 157 |
# File 'lib/logstash/inputs/tcp.rb', line 151 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 end |
#register ⇒ Object
124 125 126 127 128 129 130 131 132 |
# File 'lib/logstash/inputs/tcp.rb', line 124 def register fix_streaming_codecs # note that since we are opening a socket in register, we must also make sure we close it # in the close method even if we also close it in the stop method since we could have # a situation where register is called but not run & stop. self.server_socket = new_server_socket if server? end |
#run(output_queue) ⇒ Object
134 135 136 137 138 139 140 |
# File 'lib/logstash/inputs/tcp.rb', line 134 def run(output_queue) if server? run_server(output_queue) else run_client(output_queue) end end |
#stop ⇒ Object
142 143 144 145 146 147 148 149 |
# File 'lib/logstash/inputs/tcp.rb', line 142 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 client_socket.close rescue nil connection_sockets.each{|socket| socket.close rescue nil} end |