Fix encoding error in fluentd client

This change fixes a bug that is caused by incorrectly trying to encode
any values other than string.
This bug was a regression introduced with #5107 and caused
`NoMethodError` errors.

Fixes #5099

Signed-off-by: Christian Haudum <christian.haudum@gmail.com>

CHANGELOG.md

@@ -1,5 +1,6 @@
 ## Main
 
+* [5163](https://github.com/grafana/loki/pull/5163) **chaudum** Fix regression in fluentd plugin introduced with #5107 that caused `NoMethodError` when parsing non-string values of log lines.
 * [5144](https://github.com/grafana/loki/pull/5144) **dannykopping** Ruler: fix remote write basic auth credentials.
 * [5107](https://github.com/grafana/loki/pull/5107) **chaudum** Fix bug in fluentd plugin that caused log lines containing non UTF-8 characters to be dropped.
 * [5091](https://github.com/grafana/loki/pull/5091) **owen-d**: Changes `ingester.concurrent-flushes` default to 32

clients/cmd/fluentd/lib/fluent/plugin/out_loki.rb

@@ -279,7 +279,9 @@ def record_to_line(record)
             formatted_labels = []
             record.each do |k, v|
               # Remove non UTF-8 characters by force-encoding the string
-              v = v.encode('utf-8', invalid: :replace)
+              if v.is_a?(String)
+                v = v.encode('utf-8', invalid: :replace)
+              end
               # Escape double quotes and backslashes by prefixing them with a backslash
               v = v.to_s.gsub(%r{(["\\])}, '\\\\\1')
               if v.include?(' ') || v.include?('=')

clients/cmd/fluentd/spec/gems/fluent/plugin/loki_output_spec.rb

@@ -117,13 +117,29 @@
     CONF
     driver = Fluent::Test::Driver::Output.new(described_class)
     driver.configure(config)
-    content = File.readlines('spec/gems/fluent/plugin/data/non_utf8.log')
-    chunk = [Time.at(1_546_270_458), {'message'=>content[0], 'stream'=>'stdout'}]
+    content = File.readlines('spec/gems/fluent/plugin/data/non_utf8.log')[0]
+    chunk = [Time.at(1_546_270_458), {'message'=>content, 'number': 1.2345, 'stream'=>'stdout'}]
     payload = driver.instance.generic_to_loki([chunk])
     expect(payload[0]['stream'].empty?).to eq true
     expect(payload[0]['values'].count).to eq 1
-    expect(payload[0]['values'][0][0]).to eq '1546270458000000000'
-    expect(payload[0]['values'][0][1]).to eq 'message="� rest of line" stream=stdout'
+    expect(payload[0]['values'][0][0]).to eq "1546270458000000000"
+    expect(payload[0]['values'][0][1]).to eq "message=\"� rest of line\" number=1.2345 stream=stdout"
+  end
+
+  it 'handle non utf-8 characters from log lines in json format' do
+    config = <<-CONF
+      url         https://logs-us-west1.grafana.net
+      line_format json
+    CONF
+    driver = Fluent::Test::Driver::Output.new(described_class)
+    driver.configure(config)
+    content = File.readlines('spec/gems/fluent/plugin/data/non_utf8.log')[0]
+    chunk = [Time.at(1_546_270_458), {'message'=>content, 'number': 1.2345, 'stream'=>'stdout'}]
+    payload = driver.instance.generic_to_loki([chunk])
+    expect(payload[0]['stream'].empty?).to eq true
+    expect(payload[0]['values'].count).to eq 1
+    expect(payload[0]['values'][0][0]).to eq "1546270458000000000"
+    expect(payload[0]['values'][0][1]).to eq "{\"message\":\"\xC1 rest of line\",\"number\":1.2345,\"stream\":\"stdout\"}"
   end
 
   it 'formats record hash as key_value' do