Improve zipkin plugins

fluent-plugin-zipkin/lib/fluent/plugin/in_zipkin.rb

@@ -72,6 +72,7 @@ class ZipkinInput < Input
 
     def configure(conf)
       compat_parameters_convert(conf, :parser)
+      @first_trace = true
 
       super
 
@@ -157,6 +158,12 @@ def on_request(path_info, params)
         end
 
         router.emit_stream(tag, mes)
+
+        if @first_trace
+          log.info "First trace received"
+          @first_trace = false
+        end
+
       rescue Exception => exception
         return ["500 Internal Server Error", {'Content-Type'=>'text/plain'}, "500 Internal Server Error\n#{$!}\n"]
       end

fluent-plugin-zipkin/lib/fluent/plugin/out_zipkin.rb

@@ -19,6 +19,7 @@
 require 'openssl'
 require 'fluent/plugin/output'
 require 'fluent/plugin_helper/socket'
+require 'fluent/event'
 require 'google/protobuf'
 require 'snappy'
 require 'json'
@@ -31,6 +32,8 @@ module Fluent::Plugin
   class ZipkinOutput < Output
     Fluent::Plugin.register_output('zipkin', self)
 
+    helpers :event_emitter
+
     class RetryableResponse < StandardError; end
 
     desc 'The endpoint for HTTP request, e.g. http://example.com/api'
@@ -93,6 +96,7 @@ def initialize
     def configure(conf)
       super
 
+      @first_trace = true
       if @content_type == 'application/json' then
         define_singleton_method(:get_spans, method(:get_spans_for_json))
       elsif @content_type != 'application/x-protobuf'
@@ -106,28 +110,23 @@ def multi_workers_ready?
       true
     end
 
-    def send_request_repeated(uri, req)
-      # TODO (sumo-drosiek, 2020.02.12): Rather emit records back to the fluent than resend same packet
-      begin
-        send_request(uri, req)
-        log.debug { "#{@http_method.capitalize} data to #{uri.to_s} with chunk(#{dump_unique_id_hex(chunk.unique_id)})" }
-      rescue Exception => exception
-        log.warn exception
-        log.info "Resend request due to #{exception}"
-        send_request_repeated(uri, req)
-        return
-      end
-    end
-
     def write(chunk)
       uri = parse_endpoint(chunk)
-      data = get_spans(chunk, size: 100)
+      data = split_chunk(chunk, size: 100)
 
-      data.each do |payload|
+      data.each do |spans|
         begin
           req = create_request(chunk, uri)
+          payload = get_spans(spans)
           req.body = payload
-          send_request_repeated(uri, req)
+          if not send_request(uri, req)
+            mes = Fluent::MultiEventStream.new
+            spans.each do |timestamp, record|
+              mes.add(timestamp, record)
+            end
+            # ToDo (sumo-drosiek, 2020.03.06): Use chunk tag. It's hardcoded because sometimes chunk.metadata.tag is nil
+            router.emit_stream("tracing.resend", mes)
+          end
         rescue Exception => exception
           log.warn exception
           return
@@ -215,21 +214,19 @@ def send_request(uri, req)
                 http.request(req)
               }
             end
-
       if res.is_a?(Net::HTTPSuccess)
         log.debug { "#{res.code} #{res.message.rstrip}#{res.body.lstrip}" }
-      else
-        msg = "#{res.code} #{res.message.rstrip} #{res.body.lstrip}"
 
-        if @retryable_response_codes.include?(res.code.to_i)
-          raise RetryableResponse, msg
+        if @first_trace
+          log.info "First trace successfully sent to the receiver"
+          @first_trace = false
         end
 
-        if @error_response_as_unrecoverable
-          raise Fluent::UnrecoverableError, msg
-        else
-          log.error "got error response from '#{@http_method.capitalize} #{uri.to_s}' : #{msg}"
-        end
+        return true
+      else
+        msg = "#{res.code} #{res.message.rstrip}"
+        log.error "Error during sending traces: #{msg}"
+        return false
       end
     end
 
@@ -260,37 +257,35 @@ def string_to_ipv4(value, length)
       return IPAddr.new_ntoh(value).to_s
     end
 
-    def get_spans(chunk, size: 100)
-      # Using array is more efficient than inserting to Proto3 element by element
+    def split_chunk(chunk, size: 100)
       return_value = []
       spans = []
 
-      chunk.each do |time, record|
-        if !record['trace_id'] then
+      chunk.each do |time, span|
+        if !span['trace_id'] then
           next
         end
         if spans.length >= size
-          return_value.push(
-            Zipkin::Proto3::ListOfSpans.encode(
-              Zipkin::Proto3::ListOfSpans::new({
-                'spans' => spans
-                })))
+          return_value.push(spans)
           spans = []
         end
-        spans.push(Zipkin::Proto3::Span.new(record))
+        spans.push([time, span])
       end
 
       if spans.length > 0
-        return_value.push(
-          Zipkin::Proto3::ListOfSpans.encode(
-            Zipkin::Proto3::ListOfSpans::new({
-              'spans' => spans
-              })))
+        return_value.push(spans)
       end
 
       return return_value
     end
 
+    def get_spans(spans)
+      return Zipkin::Proto3::ListOfSpans.encode(
+            Zipkin::Proto3::ListOfSpans::new({
+              'spans' => spans.map { |time, span| span }
+              }))
+    end
+
     def get_str_or_nil(value)
       if !value || value == ''
         return nil
@@ -299,29 +294,8 @@ def get_str_or_nil(value)
       return value
     end
 
-    def get_spans_for_json(chunk, size: 100)
-      return_value = []
-      spans = []
-
-      chunk.each do |time, record|
-        if !record['trace_id'] then
-          next
-        end
-
-        if spans.length >= size
-          return_value.push(Oj.dump(spans))
-          spans = []
-        end
-
-        spans.push(get_span_for_json(record))
-      end
-
-      if spans.length > 0
-        return_value.push(Oj.dump(spans))
-        spans = []
-      end
-
-      return return_value
+    def get_spans_for_json(spans)
+      return Oj.dump(spans.map { |time, span| get_span_for_json(span) })
     end
 
     def get_span_for_json(record)

fluent-plugin-zipkin/test/plugin/test_out_zipkin.rb

@@ -83,9 +83,9 @@ def test_protobuf
             'tag2' => 'test_second_tag'
         }
     }]]
-    spans = @plugin.instance.send(:get_spans, chunk)
+    spans = @plugin.instance.send(:get_spans, @plugin.instance.send(:split_chunk, chunk)[0])
 
-    assert_equal(spans[0], EXPECTED_PROTOBUF.force_encoding("ASCII-8BIT"))
+    assert_equal(spans, EXPECTED_PROTOBUF.force_encoding("ASCII-8BIT"))
   end
 
   def test_json
@@ -127,9 +127,9 @@ def test_json
         }
     }]]
 
-    spans = @plugin.instance.send(:get_spans_for_json, SPANS_CHUNK)
+    spans = @plugin.instance.send(:get_spans_for_json, @plugin.instance.send(:split_chunk, SPANS_CHUNK)[0])
 
-    assert_equal(Oj.load(spans[0], mode: :compat), Oj.load(EXPECTED_JSON, mode: :compat))
+    assert_equal(Oj.load(spans, mode: :compat), Oj.load(EXPECTED_JSON, mode: :compat))
   end
 
   def test_json_empty_parent_id
@@ -172,8 +172,8 @@ def test_json_empty_parent_id
         }
     }]]
 
-    spans = @plugin.instance.send(:get_spans_for_json, chunk)
+    spans = @plugin.instance.send(:get_spans_for_json, @plugin.instance.send(:split_chunk, chunk)[0])
 
-    assert_equal(Oj.load(spans[0], mode: :compat), Oj.load(expected_json, mode: :compat))
+    assert_equal(Oj.load(spans, mode: :compat), Oj.load(expected_json, mode: :compat))
   end
 end