Class: Fluent::DatadogOutput
- Inherits:
-
BufferedOutput
- Object
- BufferedOutput
- Fluent::DatadogOutput
show all
- Defined in:
- lib/fluent/plugin/out_datadog.rb
Defined Under Namespace
Classes: ConnectionFailure
Instance Method Summary
collapse
Constructor Details
Returns a new instance of DatadogOutput.
38
39
40
|
# File 'lib/fluent/plugin/out_datadog.rb', line 38
def initialize
super
end
|
Instance Method Details
47
48
49
50
51
52
53
|
# File 'lib/fluent/plugin/out_datadog.rb', line 47
def configure(conf)
super
return if @dd_hostname
@dd_hostname = %x[hostname -f 2> /dev/null].strip
@dd_hostname = Socket.gethostname if @dd_hostname.empty?
end
|
This method is called when an event reaches Fluentd.
97
98
99
100
101
102
|
# File 'lib/fluent/plugin/out_datadog.rb', line 97
def format(tag, time, record)
return [tag, time.to_f, record].to_msgpack
end
|
193
194
195
196
197
198
|
# File 'lib/fluent/plugin/out_datadog.rb', line 193
def get_container_tags(record)
return [
get_kubernetes_tags(record),
get_docker_tags(record)
].compact.join(",")
end
|
213
214
215
216
217
218
219
220
221
|
# File 'lib/fluent/plugin/out_datadog.rb', line 213
def get_docker_tags(record)
if record.key?('docker') and not record.fetch('docker').nil?
docker = record['docker']
tags = Array.new
tags.push("container_id:" + docker['container_id']) unless docker['container_id'].nil?
return tags.join(",")
end
return nil
end
|
200
201
202
203
204
205
206
207
208
209
210
211
|
# File 'lib/fluent/plugin/out_datadog.rb', line 200
def get_kubernetes_tags(record)
if record.key?('kubernetes') and not record.fetch('kubernetes').nil?
kubernetes = record['kubernetes']
tags = Array.new
tags.push("image_name:" + kubernetes['container_image']) unless kubernetes['container_image'].nil?
tags.push("container_name:" + kubernetes['container_name']) unless kubernetes['container_name'].nil?
tags.push("kube_namespace:" + kubernetes['namespace_name']) unless kubernetes['namespace_name'].nil?
tags.push("pod_name:" + kubernetes['pod_name']) unless kubernetes['pod_name'].nil?
return tags.join(",")
end
return nil
end
|
#multi_workers_ready? ⇒ Boolean
55
56
57
|
# File 'lib/fluent/plugin/out_datadog.rb', line 55
def multi_workers_ready?
true
end
|
#new_client ⇒ Object
59
60
61
62
63
64
65
66
67
68
69
|
# File 'lib/fluent/plugin/out_datadog.rb', line 59
def new_client
if @use_ssl
context = OpenSSL::SSL::SSLContext.new
socket = TCPSocket.new @host, @ssl_port
ssl_client = OpenSSL::SSL::SSLSocket.new socket, context
ssl_client.connect
return ssl_client
else
return TCPSocket.new @host, @port
end
end
|
#send_to_datadog(events) ⇒ Object
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
|
# File 'lib/fluent/plugin/out_datadog.rb', line 156
def send_to_datadog(events)
@my_mutex.synchronize do
events.each do |event|
log.trace "Datadog plugin: about to send event=#{event}"
retries = 0
begin
log.info "New attempt to Datadog attempt=#{retries}" if retries > 1
@client ||= new_client
@client.write(event)
rescue => e
@client.close rescue nil
@client = nil
if retries == 0
retries += 1
retry
end
if retries < @max_retries || @max_retries == -1
a_couple_of_seconds = retries ** 2
a_couple_of_seconds = 30 unless a_couple_of_seconds < 30
retries += 1
log.warn "Could not push event to Datadog, attempt=#{retries} max_attempts=#{max_retries} wait=#{a_couple_of_seconds}s error=#{e}"
sleep a_couple_of_seconds
retry
end
raise ConnectionFailure, "Could not push event to Datadog after #{retries} retries, #{e}"
end
end
end
end
|
#shutdown ⇒ Object
88
89
90
91
92
93
94
|
# File 'lib/fluent/plugin/out_datadog.rb', line 88
def shutdown
super
@running = false
if @client
@client.close
end
end
|
#start ⇒ Object
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
|
# File 'lib/fluent/plugin/out_datadog.rb', line 71
def start
super
@my_mutex = Mutex.new
@running = true
if @tcp_ping_rate > 0
@timer = Thread.new do
while @running do
messages = Array.new
messages.push("fp\n")
send_to_datadog(messages)
sleep(@tcp_ping_rate)
end
end
end
end
|
#write(chunk) ⇒ Object
NOTE! This method is called by internal thread, not Fluentd’s main thread. ‘chunk’ is a buffer chunk that includes multiple formatted events.
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
|
# File 'lib/fluent/plugin/out_datadog.rb', line 106
def write(chunk)
messages = Array.new
chunk.msgpack_each do |tag, time, record|
next unless record.is_a? Hash
next if record.empty?
if @dd_sourcecategory
record["ddsourcecategory"] = @dd_sourcecategory
end
if @dd_source
record["ddsource"] = @dd_source
end
if @dd_tags
record["ddtags"] = @dd_tags
end
if @service
record["service"] = @service
end
if @dd_hostname
record["hostname"] = @dd_hostname
end
if @include_tag_key
record[@tag_key] = tag
end
if @timestamp_key and record[@timestamp_key].nil? and time
record[@timestamp_key] = Time.at(time).utc.iso8601(3)
end
container_tags = get_container_tags(record)
if not container_tags.empty?
if record["ddtags"].nil? || record["ddtags"].empty?
record["ddtags"] = container_tags
else
record["ddtags"] = record["ddtags"] + "," + container_tags
end
end
if @use_json
messages.push "#{api_key} " + Yajl.dump(record) + "\n"
else
next unless record.has_key? "message"
messages.push "#{api_key} " + record["message"].strip + "\n"
end
end
send_to_datadog(messages)
end
|