Fixing up edge cases with influxdb's new line protocol

lib/logstash/outputs/influxdb.rb

@@ -217,14 +217,13 @@ def close
   # cpu_load_short,host=server01,region=us-west value=0.64 cpu_load_short,host=server02,region=us-west value=0.55 1422568543702900257 cpu_load_short,direction=in,host=server01,region=us-west value=23422.0 1422568543702900257  
   def events_to_request_body(events)
     events.map do |event|
-      result = event["measurement"].dup
-      result << "," << event["tags"].map { |tag,value| "#{tag}=#{value}" }.join(',') if event.has_key?("tags")
-      result << " " << event["fields"].map { |field,value| "#{field}=#{quoted(value)}" }.join(',')
+      result = escaped_measurement(event["measurement"].dup)
+      result << "," << event["tags"].map { |tag,value| "#{escaped(tag)}=#{escaped(value)}" }.join(',') if event.has_key?("tags")
+      result << " " << event["fields"].map { |field,value| "#{escaped(field)}=#{quoted(value)}" }.join(',')
       result << " #{event["time"]}"
     end.join("\n") #each measurement should be on a separate line
   end
 
-
   # Create a data point from an event. If @use_event_fields_for_data_points is
   # true, convert the event to a hash. Otherwise, use @data_points. Each key and 
   # value will be run through event#sprintf with the exception of a non-String
@@ -337,4 +336,18 @@ def read_body?( response )
   def quoted(value)
     Numeric === value ? value : %Q|"#{value.gsub('"','\"')}"|
   end
+
+
+  # Escape tag key, tag value, or field key
+  def escaped(value)
+    value.gsub(/[ ,=]/, ' ' => '\ ', ',' => '\,', '=' => '\=')
+  end
+
+
+  # Escape measurements note they don't need to worry about the '=' case
+  def escaped_measurement(value)
+    value.gsub(/[ ,]/, ' ' => '\ ', ',' => '\,')
+  end
+
+
 end # class LogStash::Outputs::InfluxDB

logstash-output-influxdb.gemspec

@@ -1,7 +1,7 @@
 Gem::Specification.new do |s|
 
   s.name            = 'logstash-output-influxdb'
-  s.version         = '2.0.2'
+  s.version         = '2.0.3'
   s.licenses        = ['Apache License (2.0)']
   s.summary         = "This output lets you output Metrics to InfluxDB"
   s.description     = "This gem is a logstash plugin required to be installed on top of the Logstash core pipeline using $LS_HOME/bin/plugin install gemname. This gem is not a stand-alone program"

spec/outputs/influxdb_spec.rb

@@ -115,6 +115,175 @@
     end
   end
 
+  context "Escapeing space characters" do
+    let(:config) do <<-CONFIG
+        input {
+           generator {
+             message => "foo=1 bar=2 baz=3 time=4"
+             count => 1
+             type => "generator"
+           }
+         }
+
+         filter {
+           kv { 
+	      add_field => {
+		"test1" => "yellow cat"
+		"test space" => "making life hard"
+		"feild space" => "pink dog"
+	      }
+	   }
+         }
+
+         output {
+           influxdb {
+             host => "localhost"
+             measurement => "my series"
+             allow_time_override => true
+             use_event_fields_for_data_points => true
+             exclude_fields => ["@version", "@timestamp", "sequence", "message", "type", "host"]
+             send_as_tags => ["bar", "baz", "test1", "test space"]
+           }
+         }
+      CONFIG
+    end
+
+    let(:expected_url)  { 'http://localhost:8086/write?db=statistics&rp=default&precision=ms&u=&p='}
+    let(:expected_body) { 'my\ series,bar=2,baz=3,test1=yellow\ cat,test\ space=making\ life\ hard foo="1",feild\ space="pink dog" 4' }
+
+    it "should send the specified fields as tags" do
+      expect_any_instance_of(Manticore::Client).to receive(:post!).with(expected_url, body: expected_body)
+      pipeline.run
+    end
+  end
+
+  context "Escapeing comma characters" do
+    let(:config) do <<-CONFIG
+        input {
+           generator {
+             message => "foo=1 bar=2 baz=3 time=4"
+             count => 1
+             type => "generator"
+           }
+         }
+
+         filter {
+           kv {
+              add_field => {
+                "test1" => "yellow, cat"
+                "test, space" => "making, life, hard"
+                "feild, space" => "pink, dog"
+              }
+           }
+         }
+
+         output {
+           influxdb {
+             host => "localhost"
+             measurement => "my, series"
+             allow_time_override => true
+             use_event_fields_for_data_points => true
+             exclude_fields => ["@version", "@timestamp", "sequence", "message", "type", "host"]
+             send_as_tags => ["bar", "baz", "test1", "test, space"]
+           }
+         }
+      CONFIG
+    end
+
+    let(:expected_url)  { 'http://localhost:8086/write?db=statistics&rp=default&precision=ms&u=&p='}
+    let(:expected_body) { 'my\,\ series,bar=2,baz=3,test1=yellow\,\ cat,test\,\ space=making\,\ life\,\ hard foo="1",feild\,\ space="pink, dog" 4' }
+
+    it "should send the specified fields as tags" do
+      expect_any_instance_of(Manticore::Client).to receive(:post!).with(expected_url, body: expected_body)
+      pipeline.run
+    end
+  end
+
+  context "Escapeing equal characters" do
+    let(:config) do <<-CONFIG
+        input {
+           generator {
+             message => "foo=1 bar=2 baz=3 time=4"
+             count => 1
+             type => "generator"
+           }
+         }
+
+         filter {
+           kv {
+              add_field => {
+                "test1" => "yellow=cat"
+                "test=space" => "making= life=hard"
+                "feild= space" => "pink= dog"
+              }
+           }
+         }
+
+         output {
+           influxdb {
+             host => "localhost"
+             measurement => "my=series"
+             allow_time_override => true
+             use_event_fields_for_data_points => true
+             exclude_fields => ["@version", "@timestamp", "sequence", "message", "type", "host"]
+             send_as_tags => ["bar", "baz", "test1", "test=space"]
+           }
+         }
+      CONFIG
+    end
+
+    let(:expected_url)  { 'http://localhost:8086/write?db=statistics&rp=default&precision=ms&u=&p='}
+    let(:expected_body) { 'my=series,bar=2,baz=3,test1=yellow\=cat,test\=space=making\=\ life\=hard foo="1",feild\=\ space="pink= dog" 4' }
+
+    it "should send the specified fields as tags" do
+      expect_any_instance_of(Manticore::Client).to receive(:post!).with(expected_url, body: expected_body)
+      pipeline.run
+    end
+  end
+
+  context "testing backslash characters" do
+    let(:config) do <<-CONFIG
+        input {
+           generator {
+             message => 'foo\\=1 bar=2 baz=3 time=4'
+             count => 1
+             type => "generator"
+           }
+         }
+
+         filter {
+           kv {
+              add_field => {
+                "test1" => "yellow=cat"
+                "test=space" => "making=, life=hard"
+                "feildspace" => 'C:\\Griffo'
+              }
+           }
+         }
+
+         output {
+           influxdb {
+             host => "localhost"
+             measurement => 'my\\series'
+             allow_time_override => true
+             use_event_fields_for_data_points => true
+             exclude_fields => ["@version", "@timestamp", "sequence", "message", "type", "host"]
+             send_as_tags => ['bar', "baz", "test1", "test=space"]
+           }
+         }
+      CONFIG
+    end
+
+    let(:expected_url)  { 'http://localhost:8086/write?db=statistics&rp=default&precision=ms&u=&p='}
+    let(:expected_body) { 'my\series,bar=2,baz=3,test1=yellow\=cat,test\=space=making\=\,\ life\=hard foo\="1",feildspace="C:\Griffo" 4' }
+
+    it "should send the specified fields as tags" do
+      expect_any_instance_of(Manticore::Client).to receive(:post!).with(expected_url, body: expected_body)
+      pipeline.run
+    end
+  end
+
+
   context "when fields data contains a list of tags" do
     let(:config) do <<-CONFIG
         input {