diff --git a/CHANGELOG.md b/CHANGELOG.md index e5a2a4f..bddaedf 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] +### Fixed +- Mint a genuine parentless root span for `trace_id:` on `start_observation`/`observe`, instead of a synthetic never-exported parent, so traces are correctly recognized as roots by Langfuse ingestion, including the public Observations v2 API (same root cause as the still-open langfuse/langfuse#14868), while still preserving attributes set via `Langfuse.propagate_attributes` and baggage on the new root span + ## [0.10.1] - 2026-05-05 ### Changed diff --git a/lib/langfuse.rb b/lib/langfuse.rb index 5e5e28f..c84f594 100644 --- a/lib/langfuse.rb +++ b/lib/langfuse.rb @@ -361,13 +361,17 @@ def reset! # Creates a new observation (root or child) # # This is the module-level factory method that creates observations of any type. - # It can create root observations (when parent_span_context is nil) or child - # observations (when parent_span_context is provided). + # It creates a root observation whenever no `parent_span_context` is given — + # with a random trace ID, or (if `trace_id` is given) one pinned to a + # caller-chosen trace ID. It creates a child observation whenever + # `parent_span_context` is given explicitly, or implicitly by calling this + # from within another observation's block, which sets it as the ambient + # OTel parent. # # @param name [String] Descriptive name for the observation # @param attrs [Hash, Types::SpanAttributes, Types::GenerationAttributes, nil] Observation attributes # @param as_type [Symbol, String] Observation type (:span, :generation, :event, etc.) - # @param trace_id [String, nil] Optional 32-char lowercase hex trace ID to attach the observation to. + # @param trace_id [String, nil] Optional 32-char lowercase hex trace ID for a new root observation. # Mutually exclusive with `parent_span_context`. Use {Langfuse.create_trace_id} to generate one. # @param parent_span_context [OpenTelemetry::Trace::SpanContext, nil] Parent span context for child observations # @param start_time [Time, Integer, nil] Optional start time (Time object or Unix timestamp in nanoseconds) @@ -390,17 +394,22 @@ def reset! # rubocop:disable Metrics/ParameterLists def start_observation(name, attrs = {}, as_type: :span, trace_id: nil, parent_span_context: nil, start_time: nil, skip_validation: false) - parent_span_context = resolve_trace_context(trace_id, parent_span_context) + raise ArgumentError, "Cannot specify both trace_id and parent_span_context" if trace_id && parent_span_context + type_str = as_type.to_s validate_observation_type!(as_type, type_str) unless skip_validation otel_tracer = otel_tracer() - otel_span = create_otel_span( - name: name, - start_time: start_time, - parent_span_context: parent_span_context, - otel_tracer: otel_tracer - ) + + otel_span = + if parent_span_context + create_child_span(otel_tracer, name, parent_span_context, start_time) + elsif trace_id + create_root_span(otel_tracer, name, trace_id, start_time) + else + otel_tracer.start_span(name, start_timestamp: start_time) + end + apply_observation_attributes(otel_span, type_str, attrs) observation = wrap_otel_span(otel_span, type_str, otel_tracer) @@ -415,7 +424,7 @@ def start_observation(name, attrs = {}, as_type: :span, trace_id: nil, parent_sp # @param name [String] Descriptive name for the observation # @param attrs [Hash] Observation attributes (optional positional or keyword) # @param as_type [Symbol, String] Observation type (:span, :generation, :event, etc.) - # @param trace_id [String, nil] Optional 32-char lowercase hex trace ID to attach the observation to. + # @param trace_id [String, nil] Optional 32-char lowercase hex trace ID for a new root observation. # Use {Langfuse.create_trace_id} to generate one. Forwarded to {.start_observation}. # @param kwargs [Hash] Additional keyword arguments merged into observation attributes (e.g., input:, output:, metadata:) # @yield [observation] Optional block that receives the observation object @@ -457,14 +466,6 @@ def observe(name, attrs = {}, as_type: :span, trace_id: nil, **kwargs, &block) private - # @api private - def resolve_trace_context(trace_id, parent_span_context) - return parent_span_context unless trace_id - raise ArgumentError, "Cannot specify both trace_id and parent_span_context" if parent_span_context - - TraceId.send(:to_span_context, trace_id) - end - # @api private def validate_observation_type!(as_type, type_str) return if valid_observation_type?(as_type) @@ -513,28 +514,35 @@ def otel_tracer noop_tracer end - # Creates an OpenTelemetry span (root or child) + # Starts a child span under +parent_span_context+, via a non-recording span + # standing in for it as the active OTel context. # - # @param name [String] Span name - # @param start_time [Time, Integer, nil] Optional start time - # @param parent_span_context [OpenTelemetry::Trace::SpanContext, nil] Parent span context - # @param otel_tracer [OpenTelemetry::SDK::Trace::Tracer] The OTel tracer - # @return [OpenTelemetry::SDK::Trace::Span] The created span - def create_otel_span(name:, otel_tracer:, start_time: nil, parent_span_context: nil) - if parent_span_context - # Create child span with parent context - # Create a non-recording span from the parent context to set in context - parent_span = OpenTelemetry::Trace.non_recording_span(parent_span_context) - parent_context = OpenTelemetry::Trace.context_with_span(parent_span) - OpenTelemetry::Context.with_current(parent_context) do - otel_tracer.start_span(name, start_timestamp: start_time) - end - else - # Create root span + # @api private + def create_child_span(otel_tracer, name, parent_span_context, start_time) + parent_span = OpenTelemetry::Trace.non_recording_span(parent_span_context) + parent_context = OpenTelemetry::Trace.context_with_span(parent_span) + OpenTelemetry::Context.with_current(parent_context) do otel_tracer.start_span(name, start_timestamp: start_time) end end + # Starts a genuine root span (no OTel parent, trace ID pinned to +trace_id+) while + # keeping the rest of the ambient OTel context intact — unlike `Tracer#start_root_span`, + # which forces `Context.empty` and would silently drop propagated attributes + # (see {Propagation.propagate_attributes}) and baggage set by an enclosing block. + # Overriding only the "current span" slot (via the same `context_with_span` used by + # {#create_child_span}) is enough to make OTel treat this as parentless, since + # `TracerProvider#internal_start_span` derives root-ness from that slot alone. + # + # @api private + def create_root_span(otel_tracer, name, trace_id, start_time) + root_context = OpenTelemetry::Trace.context_with_span(OpenTelemetry::Trace::Span::INVALID) + + TraceId.pin_generation_to(trace_id) do + otel_tracer.start_span(name, with_parent: root_context, start_timestamp: start_time) + end + end + # Wraps an OpenTelemetry span in the appropriate observation class # # @param otel_span [OpenTelemetry::SDK::Trace::Span] The OTel span diff --git a/lib/langfuse/otel_setup.rb b/lib/langfuse/otel_setup.rb index e4bfaa1..21eff54 100644 --- a/lib/langfuse/otel_setup.rb +++ b/lib/langfuse/otel_setup.rb @@ -3,6 +3,7 @@ require "opentelemetry/sdk" require "opentelemetry/exporter/otlp" require "base64" +require_relative "trace_id" module Langfuse # OpenTelemetry initialization and setup for Langfuse tracing. @@ -128,7 +129,8 @@ def rollback_provider(provider) def build_tracer_provider(config) provider = OpenTelemetry::SDK::Trace::TracerProvider.new( - sampler: build_sampler(config.sample_rate) + sampler: build_sampler(config.sample_rate), + id_generator: TraceId ) provider.add_span_processor( SpanProcessor.new(config: config, exporter: build_exporter(config)) diff --git a/lib/langfuse/trace_id.rb b/lib/langfuse/trace_id.rb index f8e0e77..6c55f8a 100644 --- a/lib/langfuse/trace_id.rb +++ b/lib/langfuse/trace_id.rb @@ -11,6 +11,9 @@ module Langfuse # and score or reference traces later without having to persist the generated # Langfuse ID. # + # Also serves as the OpenTelemetry ID generator installed on Langfuse's + # TracerProvider (`id_generator: TraceId`) — see {.generate_trace_id}. + # # @example Deterministic from an external ID # trace_id = Langfuse::TraceId.create(seed: "order-12345") # Langfuse.observe("process-order", trace_id: trace_id) { |span| ... } @@ -21,8 +24,9 @@ module Langfuse module TraceId TRACE_ID_PATTERN = /\A[0-9a-f]{32}\z/ INVALID_TRACE_ID = ("0" * 32) + PINNED_TRACE_ID_KEY = :langfuse_pinned_trace_id - private_constant :TRACE_ID_PATTERN, :INVALID_TRACE_ID + private_constant :TRACE_ID_PATTERN, :INVALID_TRACE_ID, :PINNED_TRACE_ID_KEY class << self # Generate a W3C trace ID (32 lowercase hex chars). @@ -45,6 +49,56 @@ def create(seed: nil) Digest::SHA256.digest(validate_seed!(seed))[0, 16].unpack1("H*") end + # Runs the block with {.generate_trace_id} pinned to +trace_id+. + # + # Fiber-local (`Fiber[]`, Ruby 3.2+) — `Thread.current[]=` looks + # equivalent but doesn't inherit into a fiber spawned mid-call, which + # breaks under fiber-based schedulers (e.g. `async`, Falcon). Restores + # rather than clears the previous value, so nested calls on the same + # fiber stay correct. + # + # Validates before touching Fiber storage — validating inside the + # `ensure` would restore a captured-too-late `previous` of +nil+, + # clobbering an outer call's pinned trace ID. + # + # @param trace_id [String] 32-char lowercase hex trace ID + # @return [Object] the block's return value + # @raise [ArgumentError] if trace_id is invalid + # @api private + def pin_generation_to(trace_id) + raise ArgumentError, "Invalid trace_id: #{trace_id.inspect}" unless valid?(trace_id) + + previous_trace_id = Fiber[PINNED_TRACE_ID_KEY] + + # Convert a hex trace ID to the raw 16-byte form OpenTelemetry uses internally. + Fiber[PINNED_TRACE_ID_KEY] = [trace_id].pack("H*") + + begin + yield + ensure + Fiber[PINNED_TRACE_ID_KEY] = previous_trace_id + end + end + + # OpenTelemetry ID generator contract (see `TracerProvider.new(id_generator:)` + # in otel_setup.rb). OTel only ever consults this for spans with no valid + # parent (see `Langfuse.create_root_span`, which forces this by starting the span + # with a context whose "current span" slot is `Span::INVALID`); spans with a real + # or synthetic parent take their trace ID from that parent's context instead. Falls + # back to OpenTelemetry's own random generator whenever + # {.pin_generation_to} isn't active, so untouched root spans are + # unaffected. + # + # @api private + def generate_trace_id + Fiber[PINNED_TRACE_ID_KEY] || OpenTelemetry::Trace.generate_trace_id + end + + # @api private + def generate_span_id + OpenTelemetry::Trace.generate_span_id + end + private # @api private @@ -64,25 +118,6 @@ def valid?(trace_id) trace_id != INVALID_TRACE_ID end - - # Build a sampled OpenTelemetry SpanContext carrying the given hex trace ID. - # - # A random span_id is generated as a placeholder — only the trace_id is - # consumed by the child span that gets created. - # - # @api private - def to_span_context(trace_id) - raise ArgumentError, "Invalid trace_id: #{trace_id.inspect}" unless valid?(trace_id) - - OpenTelemetry::Trace::SpanContext.new( - trace_id: [trace_id].pack("H*"), - span_id: OpenTelemetry::Trace.generate_span_id, - trace_flags: OpenTelemetry::Trace::TraceFlags::SAMPLED, - # Cross-SDK parity: Python uses is_remote=False (_create_remote_parent_span). - # Changing this would alter ParentBased sampler behavior across SDKs. - remote: false - ) - end end end end diff --git a/spec/langfuse/trace_id_spec.rb b/spec/langfuse/trace_id_spec.rb index 117f9e8..ea4cbf1 100644 --- a/spec/langfuse/trace_id_spec.rb +++ b/spec/langfuse/trace_id_spec.rb @@ -77,35 +77,99 @@ end end - describe ".to_span_context" do - before do - OpenTelemetry.tracer_provider = OpenTelemetry::SDK::Trace::TracerProvider.new + describe ".pin_generation_to" do + it "makes .generate_trace_id return the raw 16-byte form of the given trace ID for the duration of the block" do + hex_trace_id = described_class.create(seed: "order-123") + + described_class.pin_generation_to(hex_trace_id) do + expect(described_class.generate_trace_id).to eq([hex_trace_id].pack("H*")) + end end - it "returns a SpanContext carrying the provided trace ID" do - hex_trace_id = described_class.create(seed: "order-123") - ctx = described_class.send(:to_span_context, hex_trace_id) + it "reverts to random trace ID generation after the block" do + pinned = described_class.create(seed: "x") + described_class.pin_generation_to(pinned) {} + + after = described_class.generate_trace_id + expect(after.bytesize).to eq(16) + expect(after).not_to eq([pinned].pack("H*")) + end + + it "restores the previous pinned trace ID after a nested call, even if it raises" do + outer = described_class.create(seed: "outer") + inner = described_class.create(seed: "inner") + + described_class.pin_generation_to(outer) do + expect do + described_class.pin_generation_to(inner) { raise "boom" } + end.to raise_error("boom") + + expect(described_class.generate_trace_id).to eq([outer].pack("H*")) + end + end + + it "isolates concurrent sibling fibers from each other's pinned trace ID" do + trace_id_a = described_class.create(seed: "fiber-a") + trace_id_b = described_class.create(seed: "fiber-b") + seen_by_a = nil + seen_by_b = nil + + fiber_a = Fiber.new do + described_class.pin_generation_to(trace_id_a) do + Fiber.yield + seen_by_a = described_class.generate_trace_id + end + end + fiber_b = Fiber.new do + described_class.pin_generation_to(trace_id_b) do + Fiber.yield + seen_by_b = described_class.generate_trace_id + end + end + + # Interleave: both fibers pin their trace ID, yield, then resume and + # read it back — a shared (non-fiber-local) store would leak b's + # value into a's read, or vice versa. + fiber_a.resume + fiber_b.resume + fiber_a.resume + fiber_b.resume - expect(ctx).to be_a(OpenTelemetry::Trace::SpanContext) - expect(ctx.trace_id.unpack1("H*")).to eq(hex_trace_id) + expect(seen_by_a).to eq([trace_id_a].pack("H*")) + expect(seen_by_b).to eq([trace_id_b].pack("H*")) end - it "sets the SAMPLED trace flag" do - ctx = described_class.send(:to_span_context, described_class.create(seed: "x")) - expect(ctx.trace_flags.sampled?).to be(true) + it "does not leak a child fiber's pinned trace ID back to its parent" do + parent_trace_id = described_class.create(seed: "parent-fiber") + child_trace_id = described_class.create(seed: "child-fiber") + + described_class.pin_generation_to(parent_trace_id) do + Fiber.new do + described_class.pin_generation_to(child_trace_id) {} + end.resume + + expect(described_class.generate_trace_id).to eq([parent_trace_id].pack("H*")) + end end - it "marks the span context as non-remote (cross-SDK parity lock)" do - ctx = described_class.send(:to_span_context, described_class.create(seed: "parity")) - expect(ctx.remote?).to be(false) + it "raises ArgumentError for an invalid trace ID without running the block" do + ran = false + expect do + described_class.pin_generation_to("not-valid") { ran = true } + end.to raise_error(ArgumentError, /Invalid trace_id/) + expect(ran).to be(false) end + end - it "raises ArgumentError for an invalid trace ID" do - expect { described_class.send(:to_span_context, "not-valid") }.to raise_error(ArgumentError, /Invalid trace_id/) + describe ".generate_trace_id" do + it "falls back to a random trace ID when none is pinned" do + expect(described_class.generate_trace_id.bytesize).to eq(16) end + end - it "raises ArgumentError for nil" do - expect { described_class.send(:to_span_context, nil) }.to raise_error(ArgumentError) + describe ".generate_span_id" do + it "returns a random 8-byte span ID" do + expect(described_class.generate_span_id.bytesize).to eq(8) end end end diff --git a/spec/langfuse_spec.rb b/spec/langfuse_spec.rb index 423de02..d25f3ef 100644 --- a/spec/langfuse_spec.rb +++ b/spec/langfuse_spec.rb @@ -412,12 +412,31 @@ expect(observation.trace_id).to eq(trace_id) end + it "creates a genuine root span with no parent, not a phantom-parented child" do + trace_id = described_class.create_trace_id(seed: "root-seed") + observation = described_class.start_observation("root", {}, trace_id: trace_id) + + expect(observation.otel_span.to_span_data.parent_span_id).to eq("\x00" * 8) + end + it "shares the trace ID with child observations created from the root" do trace_id = described_class.create_trace_id(seed: "shared-seed") root = described_class.start_observation("root", {}, trace_id: trace_id) child = root.start_observation("child") expect(child.trace_id).to eq(trace_id) + expect(child.otel_span.to_span_data.parent_span_id).to eq(root.otel_span.context.span_id) + end + + it "still applies attributes propagated via Langfuse.propagate_attributes" do + trace_id = described_class.create_trace_id(seed: "propagated-seed") + + described_class.propagate_attributes(user_id: "user_123") do + root = described_class.start_observation("root", {}, trace_id: trace_id) + + expect(root.otel_span.attributes["user.id"]).to eq("user_123") + expect(root.otel_span.to_span_data.parent_span_id).to eq("\x00" * 8) + end end it "raises when both trace_id and parent_span_context are provided" do