Class: Fluent::Papertrail
- Inherits:
-
BufferedOutput
- Object
- BufferedOutput
- Fluent::Papertrail
show all
- Defined in:
- lib/fluent/plugin/out_papertrail.rb
Defined Under Namespace
Classes: SocketFailureError
Constant Summary
collapse
- DISCARD_STRING =
declare const string for nullifying token if we decide to discard records
'DISCARD'
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
24
25
26
27
28
29
30
31
32
|
# File 'lib/fluent/plugin/out_papertrail.rb', line 24
def configure(conf)
super
@sockets = {}
socket_key = form_socket_key(@papertrail_host, @papertrail_port)
@sockets[socket_key] = create_socket(socket_key)
@default_hostname = ENV['FLUENT_HOSTNAME'] || @default_hostname
end
|
#create_packet(tag, time, record) ⇒ Object
77
78
79
80
81
82
83
84
85
86
87
|
# File 'lib/fluent/plugin/out_papertrail.rb', line 77
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(socket_key) ⇒ Object
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
|
# File 'lib/fluent/plugin/out_papertrail.rb', line 59
def create_socket(socket_key)
log.info "initializing tcp socket for #{socket_key}"
begin
host, port = split_socket_key(socket_key)
socket = TCPSocket.new(host, port)
log.debug "enabling ssl for socket #{socket_key}"
ssl = OpenSSL::SSL::SSLSocket.new(socket)
ssl.sync_close = true
ssl.connect
rescue => e
log.warn "failed to create tcp socket #{socket_key}: #{e}"
ssl = nil
end
ssl
end
|
50
51
52
|
# File 'lib/fluent/plugin/out_papertrail.rb', line 50
def form_socket_key(host, port)
"#{host}:#{port}"
end
|
34
35
36
|
# File 'lib/fluent/plugin/out_papertrail.rb', line 34
def format(tag, time, record)
[tag, time, record].to_msgpack
end
|
#pick_socket(record) ⇒ Object
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
|
# File 'lib/fluent/plugin/out_papertrail.rb', line 89
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']
elsif @discard_unannotated_pod_logs && record.dig('kubernetes')
host = DISCARD_STRING
port = DISCARD_STRING
else
host = @papertrail_host
port = @papertrail_port
end
form_socket_key(host, port)
end
|
#send_to_papertrail(packet, socket_key) ⇒ Object
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
|
# File 'lib/fluent/plugin/out_papertrail.rb', line 112
def send_to_papertrail(packet, socket_key)
if @sockets[socket_key].nil?
err_msg = "Unable to create socket with #{socket_key}"
raise SocketFailureError, err_msg
else
begin
@sockets[socket_key].puts packet.assemble(max_size=@maximum_syslog_packet_size)
rescue => e
err_msg = "Error writing to #{socket_key}: #{e}"
@sockets[socket_key] = nil
raise SocketFailureError, err_msg, e.backtrace
end
end
end
|
#split_socket_key(socket_key) ⇒ Object
54
55
56
57
|
# File 'lib/fluent/plugin/out_papertrail.rb', line 54
def split_socket_key(socket_key)
socket_key_arr = socket_key.split(':')
return socket_key_arr[0], socket_key_arr[1]
end
|
#write(chunk) ⇒ Object
38
39
40
41
42
43
44
45
46
47
48
|
# File 'lib/fluent/plugin/out_papertrail.rb', line 38
def write(chunk)
chunk.msgpack_each {|(tag, time, record)|
socket_key = pick_socket(record)
unless socket_key.eql? form_socket_key(DISCARD_STRING, DISCARD_STRING)
@sockets[socket_key] ||= create_socket(socket_key)
packet = create_packet(tag, time, record)
send_to_papertrail(packet, socket_key)
end
}
end
|