42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
|
# File 'lib/fluent/plugin/out_sumologic.rb', line 42
def write(chunk)
messages_list = {}
case @format
when 'json'
chunk.msgpack_each do |tag, time, record|
if @include_tag_key
record.merge!(@tag_key => tag)
end
if @include_time_key
record.merge!(@time_key => @timef.format(time))
end
source_name = record[@source_name_key] || ''
record.delete(@source_name_key)
messages_list[source_name] = [] unless messages_list[source_name]
messages_list[source_name] << record.to_json
end
when 'text'
chunk.msgpack_each do |tag, time, record|
source_name = record[@source_name_key] || ''
messages_list[source_name] = [] unless messages_list[source_name]
messages_list[source_name] << record['message']
end
end
http = Net::HTTP.new(@host, @port.to_i)
proxy_string = if ENV['http_proxy'] then ENV['http_proxy'] else @proxy end
if proxy_string then
(proxy,proxy_port) = proxy_string.split(':')
http = Net::HTTP::Proxy(proxy,proxy_port).new(@host, @port.to_i)
end
http.use_ssl = true
http.verify_mode = @verify_ssl ? OpenSSL::SSL::VERIFY_PEER : OpenSSL::SSL::VERIFY_NONE
http.set_debug_output $stderr if @debug
messages_list.each do |source_name, messages|
request = Net::HTTP::Post.new(@path)
request['X-Sumo-Name'] = source_name unless source_name.empty?
request.body = messages.join("\n")
response = http.request(request)
unless response.is_a?(Net::HTTPSuccess)
raise "Failed to send data to #{@host}. #{response.code} #{response.message}"
end
end
end
|