out_f5_beacon.rb in fluent-plugin-f5-beacon-0.0.3

- old
+ new

@@ -34,11 +34,11 @@
   config_param :token, :string, desc: "The Beacon access token.  This is a required field."
   config_param :source_name, :string, desc: "The source name shown in Beacon.  This is a required field."
   config_param :measurement, :string, default: nil,
                desc: "The measurement name to insert event.  If not specified, fluentd's tag is used."
   config_param :time_key, :string, default: 'time',
-               desc: 'Use value of this tag if it exists in event instead of event timestamp.'
+               desc: 'Use value of this tag if it exists in event instead of event timestamp.  Values must be in Epoch time.'
   config_param :auto_tags, :bool, default: false,
                desc: "Enable/disable auto-tagging behavior which makes strings tags."
   config_param :tag_keys, :array, default: [],
                desc: "The names of the keys to use as tags."
   config_param :sequence_tag, :string, default: nil,
@@ -99,11 +99,11 @@
 
   def write(chunk)
     points = []
     tag = chunk.metadata.tag
     chunk.msgpack_each do |time, record|
-      timestamp = record.delete(@time_key) || time
+      timestamp = precision_time(record.delete(@time_key)) || time
       if tag_keys.empty? && !@auto_tags
         values = record
         tags = {}
       else
         values = {}
@@ -118,11 +118,10 @@
           else
             values[k] = v
           end
         end
       end
-
       if @sequence_tag
         if @prev_timestamp == timestamp
           @seq += 1
         else
           @seq = 0
@@ -177,10 +176,12 @@
       InfluxDB::PointValue.new(point).dump
     end.join("\n".freeze)
   end
 
   def precision_time(time)
+    return if time.nil?
+
     # nsec is supported from v0.14
     time * (10 ** 9) + (time.is_a?(Integer) ? 0 : time.nsec)
   end
 
   def handle_payload(payload)
@@ -200,10 +201,13 @@
   end
 
   def send_request(req, uri)
     res = nil
 
+    log.debug { "sending data to Beacon" }
+    log.trace { req.body }
+
     begin
       res = Net::HTTP.start(uri.host, uri.port, **http_opts(uri)) {|http| http.request(req) }
 
     rescue => e # rescue all StandardErrors
       # server didn't respond
@@ -231,5 +235,6 @@
     opts
   end
 
 end
 end
+