Class: Fluent::DatadogOutput
- Inherits:
-
BufferedOutput
- Object
- BufferedOutput
- Fluent::DatadogOutput
- Defined in:
- lib/fluent/plugin/out_datadog.rb
Defined Under Namespace
Classes: ConnectionFailure
Instance Method Summary collapse
- #client ⇒ Object
- #configure(conf) ⇒ Object
-
#format(tag, time, record) ⇒ Object
This method is called when an event reaches Fluentd.
-
#init_socket(socket) ⇒ Object
not used for now…
-
#initialize ⇒ DatadogOutput
constructor
A new instance of DatadogOutput.
- #send_to_datadog(data) ⇒ Object
- #shutdown ⇒ Object
- #start ⇒ Object
-
#write(chunk) ⇒ Object
NOTE! This method is called by internal thread, not Fluentd’s main thread.
Constructor Details
#initialize ⇒ DatadogOutput
Returns a new instance of DatadogOutput.
26 27 28 |
# File 'lib/fluent/plugin/out_datadog.rb', line 26 def initialize super end |
Instance Method Details
#client ⇒ Object
39 40 41 42 43 44 45 46 47 48 49 50 51 |
# File 'lib/fluent/plugin/out_datadog.rb', line 39 def client @_socket ||= 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 else socket = TCPSocket.new @host, @port end return @_socket end |
#configure(conf) ⇒ Object
35 36 37 |
# File 'lib/fluent/plugin/out_datadog.rb', line 35 def configure(conf) super end |
#format(tag, time, record) ⇒ Object
This method is called when an event reaches Fluentd.
95 96 97 |
# File 'lib/fluent/plugin/out_datadog.rb', line 95 def format(tag, time, record) return [tag, record].to_msgpack end |
#init_socket(socket) ⇒ Object
not used for now…
54 55 56 57 58 59 60 61 62 63 64 65 66 |
# File 'lib/fluent/plugin/out_datadog.rb', line 54 def init_socket(socket) socket.setsockopt(Socket::SOL_SOCKET, Socket::SO_KEEPALIVE, true) begin socket.setsockopt(Socket::SOL_TCP, Socket::TCP_KEEPINTVL, 3) socket.setsockopt(Socket::SOL_TCP, Socket::TCP_KEEPCNT, 3) socket.setsockopt(Socket::SOL_TCP, Socket::TCP_KEEPIDLE, 10) rescue log.info "DatadogOutput: Fallback on socket options during initialization" end return socket end |
#send_to_datadog(data) ⇒ Object
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 |
# File 'lib/fluent/plugin/out_datadog.rb', line 123 def send_to_datadog(data) @my_mutex.synchronize do retries = 0 begin log.trace "Send nb_event=#{data.size} events to Datadog" # Check the connectivity and write messages log.info "New attempt to Datadog attempt=#{retries}" if retries > 0 retries = retries + 1 data.each do |event| log.trace "Datadog plugin: about to send event=#{event}" client.write(event) end # Handle some failures rescue Errno::EHOSTUNREACH, Errno::ECONNREFUSED, Errno::ETIMEDOUT, Errno::EPIPE => e if retries < @max_retries || max_retries == -1 @_socket = nil a_couple_of_seconds = retries ** 2 a_couple_of_seconds = 30 unless a_couple_of_seconds < 30 retries += 1 log.warn "Could not push logs 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 logs to Datadog after #{retries} retries, #{e.}" end end end |
#shutdown ⇒ Object
86 87 88 89 90 91 92 |
# File 'lib/fluent/plugin/out_datadog.rb', line 86 def shutdown super @running = false if @_socket @_socket.close() end end |
#start ⇒ Object
68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 |
# File 'lib/fluent/plugin/out_datadog.rb', line 68 def start super @my_mutex = Mutex.new @running = true if @tcp_ping_rate > 0 @timer = Thread.new do while @running do = Array.new .push("fp\n") send_to_datadog() sleep(15) 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.
101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 |
# File 'lib/fluent/plugin/out_datadog.rb', line 101 def write(chunk) = Array.new log.trace "Datadog plugin: received chunck: #{chunk}" chunk.msgpack_each do |tag, record| next unless record.is_a? Hash log.trace "Datadog plugin: received record: #{record}" if @include_tag_key record[@tag_key] = tag end if @use_json .push "#{api_key} " + Yajl.dump(record) + "\n" else next unless record.has_key? "message" .push "#{api_key} " + record["message"].rstrip() + "\n" end end send_to_datadog() end |