diff --git a/CHANGELOG.md b/CHANGELOG.md index e5a2a4f..4453473 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added +- `metrics_reporter` configuration option to expose `BatchSpanProcessor` operational metrics (dropped spans, export success/failure, buffer utilization). Defaults to `nil`, which preserves the OTel SDK's existing no-op behaviour — zero change for current users. + ## [0.10.1] - 2026-05-05 ### Changed diff --git a/docs/CONFIGURATION.md b/docs/CONFIGURATION.md index 4d06e41..2d673de 100644 --- a/docs/CONFIGURATION.md +++ b/docs/CONFIGURATION.md @@ -393,6 +393,48 @@ config.mask = lambda { |data:| See [TRACING.md](TRACING.md#masking) for usage patterns and behavior details. +#### `metrics_reporter` + +- **Type:** Object responding to `add_to_counter` and `observe_value`, or `nil` +- **Default:** `nil` (defers to the OTel SDK's built-in no-op reporter) +- **Description:** Pluggable reporter for `BatchSpanProcessor` operational metrics. Forwarded directly to the underlying BSP. Useful for tracking back-pressure and queue health in production. + +```ruby +module MyMetricsReporter + # Only forward counters that indicate problems — skip success/throughput noise. + SUCCESS_METRICS = %w[otel.bsp.export.success otel.bsp.exported_spans].freeze + + # Warn when the buffer is more than 75% full after a drain — a rising trend + # here means producers are outpacing the export thread and drops are imminent. + BUFFER_UTILIZATION_WARN_THRESHOLD = 0.75 + + def self.add_to_counter(metric, increment: 1, labels: {}) + return if SUCCESS_METRICS.include?(metric) + + StatsD.increment(metric, increment, labels, no_prefix: true) + end + + def self.observe_value(metric, value:, labels: {}) + return if metric == "otel.bsp.buffer_utilization" && value < BUFFER_UTILIZATION_WARN_THRESHOLD + + StatsD.gauge(metric, value, labels, no_prefix: true) + end +end + +config.metrics_reporter = MyMetricsReporter +``` + +Key metrics emitted by the `BatchSpanProcessor`: + +| Metric | Method | Description | +|---|---|---| +| `otel.bsp.dropped_spans` | `add_to_counter` | Spans evicted from the buffer; `reason` label is `buffer-full`, `export-failure`, or `terminating` | +| `otel.bsp.exported_spans` | `add_to_counter` | Spans successfully handed to the exporter | +| `otel.bsp.export.success` | `add_to_counter` | Successful batch export calls | +| `otel.bsp.export.failure` | `add_to_counter` | Failed batch export calls | +| `otel.bsp.error` | `add_to_counter` | Internal errors (export exception or thread-creation failure); `reason` label is the exception class or `ThreadError` | +| `otel.bsp.buffer_utilization` | `observe_value` | Queue depth as a ratio of `max_queue_size`, sampled after each batch dequeue | + ## Tracing Behavior and OpenTelemetry Ownership There are three states worth documenting. @@ -623,6 +665,7 @@ Validation rules: - `prompt_cache_observer` must respond to `#call` (if set) - `should_export_span` must respond to `#call` (if set) - `mask` must respond to `#call` (if set) +- `metrics_reporter` must respond to `add_to_counter` and `observe_value` (if set) ## Accessing Current Configuration diff --git a/lib/langfuse/config.rb b/lib/langfuse/config.rb index 3d98c15..ef9c558 100644 --- a/lib/langfuse/config.rb +++ b/lib/langfuse/config.rb @@ -88,6 +88,12 @@ class Config # Receives `data:` keyword argument. nil disables masking. attr_accessor :mask + # @return [#add_to_counter, #observe_value, nil] Reporter for span processor + # operational metrics (dropped spans, buffer utilization, export success/failure). + # Must respond to `add_to_counter` and `observe_value`. nil defers to + # the OTel SDK's built-in no-op reporter — zero behavior change for current users. + attr_accessor :metrics_reporter + # @return [String] Default Langfuse API base URL DEFAULT_BASE_URL = "https://cloud.langfuse.com" @@ -197,6 +203,7 @@ def validate! validate_sample_rate! validate_should_export_span! validate_mask! + validate_metrics_reporter! end # rubocop:enable Metrics/AbcSize, Metrics/CyclomaticComplexity, Metrics/PerceivedComplexity @@ -242,6 +249,7 @@ def initialize_tracing_defaults self.sample_rate = env_value("LANGFUSE_SAMPLE_RATE") || DEFAULT_SAMPLE_RATE @should_export_span = nil @mask = nil + @metrics_reporter = nil end def validate_cache_backend! @@ -308,6 +316,18 @@ def validate_should_export_span! raise ConfigurationError, "should_export_span must respond to #call" end + def validate_metrics_reporter! + return if metrics_reporter.nil? + + unless metrics_reporter.respond_to?(:add_to_counter) + raise ConfigurationError, "metrics_reporter must respond to #add_to_counter" + end + + return if metrics_reporter.respond_to?(:observe_value) + + raise ConfigurationError, "metrics_reporter must respond to #observe_value" + end + def detect_release_from_ci_env COMMON_RELEASE_ENV_KEYS.each do |key| value = env_value(key) diff --git a/lib/langfuse/otel_setup.rb b/lib/langfuse/otel_setup.rb index e4bfaa1..1d83ce3 100644 --- a/lib/langfuse/otel_setup.rb +++ b/lib/langfuse/otel_setup.rb @@ -16,6 +16,7 @@ module OtelSetup release sample_rate should_export_span + metrics_reporter tracing_async batch_size flush_interval @@ -153,9 +154,22 @@ def validate_tracing_config!(config) raise ConfigurationError, "public_key is required" if blank?(config.public_key) raise ConfigurationError, "secret_key is required" if blank?(config.secret_key) raise ConfigurationError, "base_url cannot be empty" if blank?(config.base_url) - return if config.should_export_span.nil? || config.should_export_span.respond_to?(:call) + raise ConfigurationError, "should_export_span must respond to #call" unless + config.should_export_span.nil? || config.should_export_span.respond_to?(:call) - raise ConfigurationError, "should_export_span must respond to #call" + validate_tracing_metrics_reporter!(config.metrics_reporter) + end + + def validate_tracing_metrics_reporter!(metrics_reporter) + return if metrics_reporter.nil? + + unless metrics_reporter.respond_to?(:add_to_counter) + raise ConfigurationError, "metrics_reporter must respond to #add_to_counter" + end + + return if metrics_reporter.respond_to?(:observe_value) + + raise ConfigurationError, "metrics_reporter must respond to #observe_value" end def tracing_config_snapshot(config) diff --git a/lib/langfuse/span_processor.rb b/lib/langfuse/span_processor.rb index 88cc955..cfdd09c 100644 --- a/lib/langfuse/span_processor.rb +++ b/lib/langfuse/span_processor.rb @@ -18,7 +18,8 @@ def initialize(config:, exporter:) exporter, max_queue_size: config.batch_size * 2, schedule_delay: schedule_delay_for(config), - max_export_batch_size: config.batch_size + max_export_batch_size: config.batch_size, + metrics_reporter: config.metrics_reporter ) end diff --git a/spec/langfuse/config_spec.rb b/spec/langfuse/config_spec.rb index 40b89cf..5747c4b 100644 --- a/spec/langfuse/config_spec.rb +++ b/spec/langfuse/config_spec.rb @@ -546,6 +546,46 @@ end end + describe "metrics_reporter validation" do + let(:config) do + described_class.new do |c| + c.public_key = "pk_test" + c.secret_key = "sk_test" + end + end + + def valid_reporter + Object.new.tap do |r| + def r.add_to_counter(*) = nil + def r.observe_value(*) = nil + end + end + + it "defaults to nil" do + expect(config.metrics_reporter).to be_nil + end + + it "passes validation when nil" do + config.metrics_reporter = nil + expect { config.validate! }.not_to raise_error + end + + it "passes validation when responding to both required methods" do + config.metrics_reporter = valid_reporter + expect { config.validate! }.not_to raise_error + end + + it "raises ConfigurationError when at least one interface method is missing" do + reporter = Object.new + def reporter.observe_value(*) = nil + config.metrics_reporter = reporter + expect { config.validate! }.to raise_error( + Langfuse::ConfigurationError, + /#add_to_counter/ + ) + end + end + describe "mask validation" do let(:config) do described_class.new do |c| diff --git a/spec/langfuse/otel_setup_spec.rb b/spec/langfuse/otel_setup_spec.rb index 7f76798..8998700 100644 --- a/spec/langfuse/otel_setup_spec.rb +++ b/spec/langfuse/otel_setup_spec.rb @@ -88,6 +88,22 @@ ) end + it "validates metrics_reporter in setup" do + config.metrics_reporter = "bad" + + expect { described_class.setup(config) }.to raise_error( + Langfuse::ConfigurationError, + /metrics_reporter must respond to/ + ) + end + + it "accepts a valid metrics_reporter in setup" do + reporter = double("reporter", add_to_counter: nil, observe_value: nil) + config.metrics_reporter = reporter + + expect { described_class.setup(config) }.not_to raise_error + end + context "with sample_rate below 1.0" do before do config.sample_rate = 0.1 diff --git a/spec/langfuse/span_processor_spec.rb b/spec/langfuse/span_processor_spec.rb index f9fd0a7..4f1edf7 100644 --- a/spec/langfuse/span_processor_spec.rb +++ b/spec/langfuse/span_processor_spec.rb @@ -116,4 +116,13 @@ def exported_span_names expect { processor.force_flush(timeout: 1) }.not_to raise_error end end + + describe "metrics_reporter" do + it "forwards a configured metrics_reporter to the underlying BatchSpanProcessor" do + reporter = double("reporter", add_to_counter: nil, observe_value: nil) + config.metrics_reporter = reporter + proc = described_class.new(config: config, exporter: exporter) + expect(proc.instance_variable_get(:@metrics_reporter)).to eq(reporter) + end + end end