feat: Custom Metrics Reporter Support for Zipkin (#1382)

Long overdue follow up task to make it possible to add a custom metrics reporter via the SDK Configurator.

This will make initializing Zipkin exporter and the OTLP Exporter symmetric and reduce complexity in #1110

Author: arielvalentin

exporter/zipkin/lib/opentelemetry/exporter/zipkin/exporter.rb

@@ -29,7 +29,8 @@ class Exporter # rubocop:disable Metrics/ClassLength
 
         def initialize(endpoint: OpenTelemetry::Common::Utilities.config_opt('OTEL_EXPORTER_ZIPKIN_ENDPOINT', default: 'http://localhost:9411/api/v2/spans'),
                        headers: OpenTelemetry::Common::Utilities.config_opt('OTEL_EXPORTER_ZIPKIN_TRACES_HEADERS', 'OTEL_EXPORTER_ZIPKIN_HEADERS'),
-                       timeout: OpenTelemetry::Common::Utilities.config_opt('OTEL_EXPORTER_ZIPKIN_TRACES_TIMEOUT', 'OTEL_EXPORTER_ZIPKIN_TIMEOUT', default: 10))
+                       timeout: OpenTelemetry::Common::Utilities.config_opt('OTEL_EXPORTER_ZIPKIN_TRACES_TIMEOUT', 'OTEL_EXPORTER_ZIPKIN_TIMEOUT', default: 10),
+                       metrics_reporter: nil)
           raise ArgumentError, "invalid url for Zipkin::Exporter #{endpoint}" unless OpenTelemetry::Common::Utilities.valid_url?(endpoint)
           raise ArgumentError, 'headers must be comma-separated k=v pairs or a Hash' unless valid_headers?(headers)
 
@@ -50,6 +51,7 @@ def initialize(endpoint: OpenTelemetry::Common::Utilities.config_opt('OTEL_EXPOR
                      when Hash then headers
                      end
 
+          @metrics_reporter = metrics_reporter || OpenTelemetry::SDK::Trace::Export::MetricsReporter
           @shutdown = false
         end
 
@@ -66,6 +68,7 @@ def export(span_data, timeout: nil)
           zipkin_spans = encode_spans(span_data)
           send_spans(zipkin_spans, timeout: timeout)
         rescue StandardError => e
+          @metrics_reporter.add_to_counter('otel.zipkin_exporter.failure', labels: { 'reason' => e.class.to_s })
           OpenTelemetry.handle_error(exception: e, message: 'unexpected error in Zipkin::Exporter#export')
           FAILURE
         end
@@ -130,7 +133,7 @@ def send_spans(zipkin_spans, timeout: nil) # rubocop:disable Metrics/CyclomaticC
             @http.write_timeout = remaining_timeout if WRITE_TIMEOUT_SUPPORTED
             @http.start unless @http.started?
 
-            response = @http.request(request)
+            response = measure_request_duration { @http.request(request) }
             response.body # Read and discard body
             # in opentelemetry-js 200-399 is succcess, in opentelemetry-collector zipkin exporter,200-299 is a success
             # zipkin api docs list 202 as default success code
@@ -197,6 +200,19 @@ def backoff?(retry_after: nil, retry_count:, reason:)
           sleep(sleep_interval)
           true
         end
+
+        def measure_request_duration
+          start = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+          begin
+            response = yield
+          ensure
+            stop = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+            duration_ms = 1000.0 * (stop - start)
+            @metrics_reporter.record_value('otel.zipkin_exporter.request_duration',
+                                           value: duration_ms,
+                                           labels: { 'status' => response&.code || 'unknown' })
+          end
+        end
       end
     end
   end

exporter/zipkin/test/opentelemetry/exporters/zipkin/exporter_test.rb

@@ -66,10 +66,18 @@
       _(http.address).must_equal 'localhost'
       _(http.port).must_equal 4321
     end
+
+    it 'accepts a custom metrics reporter' do
+      custom_metrics_reporter = Object.new
+      exp = OpenTelemetry::Exporter::Zipkin::Exporter.new(metrics_reporter: custom_metrics_reporter)
+      _(exp.instance_variable_get(:@metrics_reporter)).must_equal(custom_metrics_reporter)
+    end
   end
 
   describe '#export' do
-    let(:exporter) { OpenTelemetry::Exporter::Zipkin::Exporter.new }
+    let(:metrics_reporter) { nil }
+
+    let(:exporter) { OpenTelemetry::Exporter::Zipkin::Exporter.new(metrics_reporter: metrics_reporter) }
 
     before do
       OpenTelemetry.tracer_provider = OpenTelemetry::SDK::Trace::TracerProvider.new
@@ -142,6 +150,45 @@
       OpenTelemetry.tracer_provider.shutdown
       assert_requested(stub_post)
     end
+
+    describe 'given a custom metrics reporter' do
+      let(:metrics_reporter) { InMemoryMetricsReporter.new }
+
+      describe 'on successful exports' do
+        it 'provides the metric reporter with the HTTP request duration for an export' do
+          stub_request(:post, DEFAULT_ZIPKIN_COLLECTOR_ENDPOINT).to_return(status: 202)
+          span_data = create_resource_span_data
+          result = exporter.export([span_data], timeout: nil)
+          _(result).must_equal(OpenTelemetry::SDK::Trace::Export::SUCCESS)
+
+          request_duration = metrics_reporter.records.first
+          _(request_duration.fetch(:metric)).must_equal('otel.zipkin_exporter.request_duration')
+          _(request_duration.fetch(:value)).must_be :>, 0
+          _(request_duration.fetch(:labels)).must_equal('status' => '202')
+        end
+      end
+
+      describe 'when an error occurs' do
+        it 'notifies the metric reporter when an export fails' do
+          stub_request(:post, DEFAULT_ZIPKIN_COLLECTOR_ENDPOINT).to_raise(StandardError)
+          span_data = create_resource_span_data
+          result = exporter.export([span_data], timeout: nil)
+          _(result).must_equal(OpenTelemetry::SDK::Trace::Export::FAILURE)
+
+          expect(metrics_reporter).wont_be :empty?
+
+          failed_count = metrics_reporter.counters.first
+          _(failed_count.fetch(:metric)).must_equal('otel.zipkin_exporter.failure')
+          _(failed_count.fetch(:value)).must_equal(1)
+          _(failed_count.fetch(:labels)).must_equal('reason' => 'StandardError')
+
+          request_duration = metrics_reporter.records.first
+          _(request_duration.fetch(:metric)).must_equal('otel.zipkin_exporter.request_duration')
+          _(request_duration.fetch(:value)).must_be :>, 0
+          _(request_duration.fetch(:labels)).must_equal('status' => 'unknown')
+        end
+      end
+    end
   end
 end
 

exporter/zipkin/test/test_helper.rb

@@ -12,6 +12,8 @@
 require 'minitest/autorun'
 require 'webmock/minitest'
 
+OpenTelemetry.logger = Logger.new(File::NULL)
+
 def create_span_data(status: nil, kind: nil, attributes: nil, total_recorded_attributes: 0, events: nil, total_recorded_events: 0, links: nil, total_recorded_links: 0, instrumentation_scope: OpenTelemetry::SDK::InstrumentationScope.new('vendorlib', '0.0.0'), trace_id: OpenTelemetry::Trace.generate_trace_id, trace_flags: OpenTelemetry::Trace::TraceFlags::DEFAULT, tracestate: nil)
   OpenTelemetry::SDK::Trace::SpanData.new(
     '',
@@ -34,3 +36,31 @@ def create_span_data(status: nil, kind: nil, attributes: nil, total_recorded_att
     tracestate
   )
 end
+
+# Test helper for Zipkin Exporter
+class InMemoryMetricsReporter
+  include OpenTelemetry::SDK::Trace::Export::MetricsReporter
+  attr_reader :counters, :records, :observes
+
+  def initialize
+    @counters = []
+    @records = []
+    @observes = []
+  end
+
+  def record_value(metric, value:, labels: {})
+    @records << { metric: metric, value: value, labels: labels }
+  end
+
+  def observe_value(metric, value:, labels: {})
+    @observes << { metric: metric, value: value, labels: labels }
+  end
+
+  def add_to_counter(metric, increment: 1, labels: {})
+    @counters << { metric: metric, value: increment, labels: labels }
+  end
+
+  def empty?
+    @counters.empty? && @records.empty? && @observes.empty?
+  end
+end