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.
34
35
36
|
# File 'lib/fluent/plugin/out_datadog.rb', line 34
def initialize
super
end
|
Instance Method Details
43
44
45
|
# File 'lib/fluent/plugin/out_datadog.rb', line 43
def configure(conf)
super
end
|
This method is called when an event reaches Fluentd.
89
90
91
|
# File 'lib/fluent/plugin/out_datadog.rb', line 89
def format(tag, time, record)
return [tag, record].to_msgpack
end
|
172
173
174
175
176
177
|
# File 'lib/fluent/plugin/out_datadog.rb', line 172
def get_container_tags(record)
return [
get_kubernetes_tags(record),
get_docker_tags(record)
].compact.join(",")
end
|
192
193
194
195
196
197
198
199
200
|
# File 'lib/fluent/plugin/out_datadog.rb', line 192
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
|
179
180
181
182
183
184
185
186
187
188
189
190
|
# File 'lib/fluent/plugin/out_datadog.rb', line 179
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
47
48
49
|
# File 'lib/fluent/plugin/out_datadog.rb', line 47
def multi_workers_ready?
true
end
|
#new_client ⇒ Object
51
52
53
54
55
56
57
58
59
60
61
|
# File 'lib/fluent/plugin/out_datadog.rb', line 51
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
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
|
# File 'lib/fluent/plugin/out_datadog.rb', line 135
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
80
81
82
83
84
85
86
|
# File 'lib/fluent/plugin/out_datadog.rb', line 80
def shutdown
super
@running = false
if @client
@client.close
end
end
|
#start ⇒ Object
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
|
# File 'lib/fluent/plugin/out_datadog.rb', line 63
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.
95
96
97
98
99
100
101
102
103
104
105
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
|
# File 'lib/fluent/plugin/out_datadog.rb', line 95
def write(chunk)
messages = Array.new
chunk.msgpack_each do |tag, 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 @include_tag_key
record[@tag_key] = tag
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
|