Class: Fluent::Papertrail
- Inherits:
-
BufferedOutput
- Object
- BufferedOutput
- Fluent::Papertrail
show all
- Defined in:
- lib/fluent/plugin/out_papertrail.rb
Defined Under Namespace
Classes: SocketFailureError
Instance Attribute Summary collapse
Instance Method Summary
collapse
Instance Attribute Details
#sockets ⇒ Object
Returns the value of attribute sockets.
6
7
8
|
# File 'lib/fluent/plugin/out_papertrail.rb', line 6
def sockets
@sockets
end
|
Instance Method Details
19
20
21
22
23
24
25
26
27
|
# File 'lib/fluent/plugin/out_papertrail.rb', line 19
def configure(conf)
super
@sockets = {}
socket_key = "#{@papertrail_host}:#{@papertrail_port}"
@sockets[socket_key] = create_socket(@papertrail_host, @papertrail_port)
@default_hostname = ENV['FLUENT_HOSTNAME'] || @default_hostname
end
|
#create_packet(tag, time, record) ⇒ Object
58
59
60
61
62
63
64
65
66
67
68
|
# File 'lib/fluent/plugin/out_papertrail.rb', line 58
def create_packet(tag, time, record)
packet = SyslogProtocol::Packet.new
packet.hostname = record['hostname'] || @default_hostname
packet.facility = record['facility'] || 'local0'
packet.severity = record['severity'] || 'info'
packet.tag = record['program'] || tag
packet.content = record['message']
packet.time = time ? Time.at(time) : Time.now
packet
end
|
#create_socket(host, port) ⇒ Object
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
|
# File 'lib/fluent/plugin/out_papertrail.rb', line 41
def create_socket(host, port)
log.info "initializing tcp socket for #{host}:#{port}"
begin
socket = TCPSocket.new(host, port)
log.debug "enabling ssl for socket #{host}:#{port}"
ssl = OpenSSL::SSL::SSLSocket.new(socket)
ssl.sync_close = true
ssl.connect
rescue => e
log.warn "failed to create tcp socket #{host}:#{port}: #{e}"
ssl = nil
end
ssl
end
|
29
30
31
|
# File 'lib/fluent/plugin/out_papertrail.rb', line 29
def format(tag, time, record)
[tag, time, record].to_msgpack
end
|
#pick_socket(record) ⇒ Object
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
|
# File 'lib/fluent/plugin/out_papertrail.rb', line 70
def pick_socket(record)
if record.dig('kubernetes', 'annotations', 'solarwinds_io/papertrail_host') && \
record.dig('kubernetes', 'annotations', 'solarwinds_io/papertrail_port')
host = record['kubernetes']['annotations']['solarwinds_io/papertrail_host']
port = record['kubernetes']['annotations']['solarwinds_io/papertrail_port']
elsif record.dig('kubernetes', 'namespace_annotations', 'solarwinds_io/papertrail_host') && \
record.dig('kubernetes', 'namespace_annotations', 'solarwinds_io/papertrail_port')
host = record['kubernetes']['namespace_annotations']['solarwinds_io/papertrail_host']
port = record['kubernetes']['namespace_annotations']['solarwinds_io/papertrail_port']
else
host = @papertrail_host
port = @papertrail_port
end
socket_key = "#{host}:#{port}"
@sockets[socket_key] ||= create_socket(host, port)
socket_key
end
|
#send_to_papertrail(packet, socket_key) ⇒ Object
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
|
# File 'lib/fluent/plugin/out_papertrail.rb', line 92
def send_to_papertrail(packet, socket_key)
if @sockets[socket_key].nil?
err_msg = "Unable to create socket with #{socket_key}"
log.error err_msg
raise SocketFailureError, err_msg
else
begin
@sockets[socket_key].puts packet.assemble
rescue => e
err_msg = "Error writing to #{socket_key}: #{e}"
log.error err_msg
@sockets[socket_key] = nil
raise SocketFailureError, err_msg, e.backtrace
end
end
end
|
#write(chunk) ⇒ Object
33
34
35
36
37
38
39
|
# File 'lib/fluent/plugin/out_papertrail.rb', line 33
def write(chunk)
chunk.msgpack_each {|(tag, time, record)|
socket_key = pick_socket(record)
packet = create_packet(tag, time, record)
send_to_papertrail(packet, socket_key)
}
end
|