Module: Langfuse::Propagation

Defined in:
lib/langfuse/propagation.rb

Overview

Attribute propagation utilities for Langfuse OpenTelemetry integration.

This module provides the propagate_attributes method for setting trace-level attributes that automatically propagate to all child spans within the context.

rubocop:disable Metrics/ModuleLength

Examples:

Basic usage

Langfuse.observe("operation") do |span|
  Langfuse.propagate_attributes(user_id: "user_123", session_id: "session_abc") do
    # Current span has user_id and session_id
    span.start_observation("child") do |child|
      # Child span inherits user_id and session_id
    end
  end
end

Constant Summary collapse

BAGGAGE_PREFIX =

Baggage key prefix for cross-service propagation

"langfuse_"
LANGFUSE_TRACE_ID_BAGGAGE_KEY =

Baggage key that records which Langfuse trace already owns the application root

"#{BAGGAGE_PREFIX}trace_id".freeze
SPAN_KEY_MAP =

Map of propagated attribute keys to span attribute keys

{
  "user_id" => OtelAttributes::TRACE_USER_ID,
  "session_id" => OtelAttributes::TRACE_SESSION_ID,
  "version" => OtelAttributes::VERSION,
  "tags" => OtelAttributes::TRACE_TAGS,
  "metadata" => OtelAttributes::TRACE_METADATA,
  "trace_name" => OtelAttributes::TRACE_NAME,
  "release" => OtelAttributes::RELEASE,
  "environment" => OtelAttributes::ENVIRONMENT
}.freeze
CONTEXT_KEYS =

OpenTelemetry context keys for propagated attributes

SPAN_KEY_MAP.keys.to_h do |key|
  [key, OpenTelemetry::Context.create_key("#{BAGGAGE_PREFIX}#{key}")]
end.freeze

Class Method Summary collapse

Class Method Details

._drop_environment(reason) ⇒ Object 



368
369
370
371
372
373
 
# File 'lib/langfuse/propagation.rb', line 368

def self._drop_environment(reason)
  Langfuse.configuration.logger.warn(
    "Langfuse: Propagated attribute 'environment' #{reason}. Dropping value."
  )
  nil
end

._extract_baggage_attributes(context) ⇒ Hash<String, String, Array<String>>

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Extract propagated attributes from baggage



475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
 
# File 'lib/langfuse/propagation.rb', line 475

def self._extract_baggage_attributes(context)
  return {} unless baggage_available?

  baggage = OpenTelemetry::Baggage.values(context: context)
  return {} unless baggage.is_a?(Hash)

  attributes = {}
  baggage.each do |baggage_key, baggage_value|
    next unless baggage_key.to_s.start_with?(BAGGAGE_PREFIX)

    span_key = _get_span_key_from_baggage_key(baggage_key.to_s)
    next unless span_key

    attributes[span_key] = _parse_baggage_value(span_key, baggage_value)
  end
  attributes.compact
rescue StandardError => e
  Langfuse.configuration.logger.debug("Langfuse: Baggage extraction failed: #{e.message}")
  {}
end

._get_langfuse_trace_id_from_baggage(context) ⇒ String?

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Get the Langfuse trace claim from OpenTelemetry baggage.



440
441
442
443
444
445
446
447
 
# File 'lib/langfuse/propagation.rb', line 440

def self._get_langfuse_trace_id_from_baggage(context)
  return nil unless baggage_available?

  OpenTelemetry::Baggage.values(context: context)[LANGFUSE_TRACE_ID_BAGGAGE_KEY]&.to_s&.downcase
rescue StandardError => e
  Langfuse.configuration.logger.debug("Langfuse: Trace baggage read failed: #{e.message}")
  nil
end

._get_propagated_baggage_key(key) ⇒ String

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Get baggage key for a propagated attribute



402
403
404
 
# File 'lib/langfuse/propagation.rb', line 402

def self._get_propagated_baggage_key(key)
  "#{BAGGAGE_PREFIX}#{key}"
end

._get_propagated_context_key(key) ⇒ OpenTelemetry::Context::Key

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Get context key for a propagated attribute

Raises:

  • (ArgumentError) —

    if key is not a known propagated attribute



382
383
384
 
# File 'lib/langfuse/propagation.rb', line 382

def self._get_propagated_context_key(key)
  CONTEXT_KEYS[key] || raise(ArgumentError, "Unknown propagated attribute key: #{key}")
end

._get_propagated_span_key(key) ⇒ String

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Get span attribute key for a propagated attribute



392
393
394
 
# File 'lib/langfuse/propagation.rb', line 392

def self._get_propagated_span_key(key)
  SPAN_KEY_MAP[key] || "#{OtelAttributes::TRACE_METADATA}.#{key}"
end

._get_span_key_from_baggage_key(baggage_key) ⇒ String?

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Get span key from baggage key



412
413
414
415
416
417
418
419
420
421
422
423
424
 
# File 'lib/langfuse/propagation.rb', line 412

def self._get_span_key_from_baggage_key(baggage_key)
  return nil unless baggage_key.start_with?(BAGGAGE_PREFIX)

  suffix = baggage_key[BAGGAGE_PREFIX.length..]

  # Handle metadata keys (format: langfuse_metadata_{key_name})
  if suffix.start_with?("metadata_")
    metadata_key = suffix[("metadata_".length)..]
    return "#{OtelAttributes::TRACE_METADATA}.#{metadata_key}"
  end

  SPAN_KEY_MAP[suffix]
end

._merge_metadata(context, context_key, new_metadata) ⇒ Hash<String, String>

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Merge metadata with existing context value



223
224
225
226
227
 
# File 'lib/langfuse/propagation.rb', line 223

def self._merge_metadata(context, context_key, new_metadata)
  existing = context.value(context_key) || {}
  existing = existing.to_h if existing.respond_to?(:to_h)
  existing.merge(new_metadata)
end

._merge_tags(context, context_key, new_tags) ⇒ Array<String>

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Merge tags with existing context value



237
238
239
240
241
 
# File 'lib/langfuse/propagation.rb', line 237

def self._merge_tags(context, context_key, new_tags)
  existing = context.value(context_key) || []
  existing = existing.to_a if existing.respond_to?(:to_a)
  (existing + new_tags).uniq.freeze
end

._parse_baggage_value(span_key, baggage_value) ⇒ String+

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Parse a baggage value into the appropriate format



503
504
505
506
507
508
509
510
511
 
# File 'lib/langfuse/propagation.rb', line 503

def self._parse_baggage_value(span_key, baggage_value)
  if span_key == OtelAttributes::ENVIRONMENT
    _validate_environment_value(baggage_value)
  elsif span_key == OtelAttributes::TRACE_TAGS && baggage_value.is_a?(String)
    baggage_value.split(",")
  else
    baggage_value.to_s
  end
end

._propagate_attributes(attributes, as_baggage:) ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Internal implementation of propagate_attributes



107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
 
# File 'lib/langfuse/propagation.rb', line 107

def self._propagate_attributes(attributes, as_baggage:, &)
  current_context = OpenTelemetry::Context.current
  current_span = OpenTelemetry::Trace.current_span

  attributes.each do |key, value|
    next if value.nil?
    next if key == "tags" && value.empty?

    validated_value = _validate_attribute_value(key, value)
    next unless validated_value

    current_context = _set_propagated_attribute(
      key: key,
      value: validated_value,
      context: current_context,
      span: current_span,
      as_baggage: as_baggage
    )
  end

  # Execute block in new context
  OpenTelemetry::Context.with_current(current_context, &)
end

._set_baggage_attribute(context:, key:, value:, baggage_key:) ⇒ OpenTelemetry::Context

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Set a propagated attribute in baggage

rubocop:disable Metrics/CyclomaticComplexity, Metrics/PerceivedComplexity



523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
 
# File 'lib/langfuse/propagation.rb', line 523

def self._set_baggage_attribute(context:, key:, value:, baggage_key:)
  return context unless baggage_available?

  if key == "metadata" && value.is_a?(Hash)
    value.each do |k, v|
      entry_key = "#{baggage_key}_#{k}"
      context = OpenTelemetry::Baggage.set_value(entry_key, v.to_s, context: context)
    end
  elsif key == "tags" && value.is_a?(Array)
    context = OpenTelemetry::Baggage.set_value(baggage_key, value.join(","), context: context)
  else
    context = OpenTelemetry::Baggage.set_value(baggage_key, value.to_s, context: context)
  end
  context
rescue StandardError => e
  Langfuse.configuration.logger.warn("Langfuse: Failed to set baggage: #{e.message}")
  context
end

._set_langfuse_trace_id_in_baggage(trace_id, context:) ⇒ OpenTelemetry::Context

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Set the Langfuse trace claim in OpenTelemetry baggage.



455
456
457
458
459
460
461
462
463
464
465
466
467
 
# File 'lib/langfuse/propagation.rb', line 455

def self._set_langfuse_trace_id_in_baggage(trace_id, context:)
  return context unless baggage_available?

  normalized_trace_id = trace_id.downcase
  return context if _get_langfuse_trace_id_from_baggage(context) == normalized_trace_id

  OpenTelemetry::Baggage.set_value(
    LANGFUSE_TRACE_ID_BAGGAGE_KEY, normalized_trace_id, context: context
  )
rescue StandardError => e
  Langfuse.configuration.logger.debug("Langfuse: Trace baggage write failed: #{e.message}")
  context
end

._set_propagated_attribute(key:, value:, context:, span:, as_baggage:) ⇒ OpenTelemetry::Context

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Set a propagated attribute in context and on current span

rubocop:disable Metrics/AbcSize, Metrics/CyclomaticComplexity, Metrics/MethodLength, Metrics/PerceivedComplexity



254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
 
# File 'lib/langfuse/propagation.rb', line 254

def self._set_propagated_attribute(key:, value:, context:, span:, as_baggage:)
  context_key = _get_propagated_context_key(key)
  span_key = _get_propagated_span_key(key)
  baggage_key = _get_propagated_baggage_key(key)

  # Merge metadata/tags with existing context values
  merged = if key == "metadata" && value.is_a?(Hash)
             _merge_metadata(context, context_key, value)
           elsif key == "tags" && value.is_a?(Array)
             _merge_tags(context, context_key, value)
           else
             value
           end

  context = context.set_value(context_key, merged)

  # Set on current span (if recording)
  if span&.recording?
    if key == "metadata" && merged.is_a?(Hash)
      merged.each do |k, v|
        metadata_key = "#{OtelAttributes::TRACE_METADATA}.#{k}"
        span.set_attribute(metadata_key, v.to_s)
      end
    elsif key == "tags" && merged.is_a?(Array)
      span.set_attribute(span_key, merged) unless merged.empty?
    else
      span.set_attribute(span_key, merged.to_s)
    end
  end

  # Set in baggage (if requested and available)
  if as_baggage
    unless baggage_available?
      Langfuse.configuration.logger.warn(
        "Langfuse: Baggage propagation requested but opentelemetry-baggage gem not available. " \
        "Install opentelemetry-baggage for cross-service propagation."
      )
    end

    context = _set_baggage_attribute(
      context: context,
      key: key,
      value: merged,
      baggage_key: baggage_key
    )
  end

  context
end

._validate_attribute_value(key, value) ⇒ Object?

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Validate an attribute value based on its type

rubocop:disable Metrics/CyclomaticComplexity



140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
 
# File 'lib/langfuse/propagation.rb', line 140

def self._validate_attribute_value(key, value)
  case key
  when "tags"
    validated_tags = value.filter_map { |tag| _validate_propagated_value(tag, "tag") }
    validated_tags.any? ? validated_tags : nil
  when "metadata"
    validated_metadata = {}
    value.each do |k, v|
      validated_metadata[k.to_s] = v.to_s if _validate_string_value(v, "metadata.#{k}")
    end
    validated_metadata.any? ? validated_metadata : nil
  when "environment"
    _validate_environment_value(value)
  else
    _validate_propagated_value(value, key)
  end
end

._validate_environment_value(value) ⇒ String?

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Validate a propagated environment value against the cross-SDK contract.



357
358
359
360
361
362
363
364
365
366
 
# File 'lib/langfuse/propagation.rb', line 357

def self._validate_environment_value(value)
  return _drop_environment("value is not a string") unless value.is_a?(String)
  return _drop_environment("value is over 40 characters (#{value.length} chars)") if value.length > 40

  return value if ENVIRONMENT_VALUE_PATTERN.match?(value)

  _drop_environment(
    "must use lowercase letters, numbers, hyphens, or underscores and must not start with 'langfuse'"
  )
end

._validate_propagated_value(value, key) ⇒ String, ...

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Validate a propagated value (string or array of strings)



312
313
314
315
316
317
318
319
320
321
322
 
# File 'lib/langfuse/propagation.rb', line 312

def self._validate_propagated_value(value, key)
  if value.is_a?(Array)
    validated = value.filter_map { |v| _validate_string_value(v, key) ? v : nil }
    return validated.any? ? validated : nil
  end

  # Validate string value (will log warning if not a string)
  return nil unless _validate_string_value(value, key)

  value
end

._validate_string_value(value, key) ⇒ Boolean

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Validate a string value

rubocop:disable Naming/PredicateMethod



332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
 
# File 'lib/langfuse/propagation.rb', line 332

def self._validate_string_value(value, key)
  unless value.is_a?(String)
    Langfuse.configuration.logger.warn(
      "Langfuse: Propagated attribute '#{key}' value is not a string. Dropping value."
    )
    return false
  end

  if value.length > 200
    Langfuse.configuration.logger.warn(
      "Langfuse: Propagated attribute '#{key}' value is over 200 characters " \
      "(#{value.length} chars). Dropping value."
    )
    return false
  end

  true
end

._with_experiment_attributes(attributes) { ... } ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Apply SDK-owned experiment attributes to the current span and future children.

Yields:

  • Block within which experiment attributes propagate



204
205
206
207
208
209
210
211
212
213
 
# File 'lib/langfuse/propagation.rb', line 204

def self._with_experiment_attributes(attributes, &)
  return yield if attributes.nil? || attributes.empty?

  frozen_attributes = attributes.dup.freeze
  current_span = OpenTelemetry::Trace.current_span
  frozen_attributes.each { |key, value| current_span.set_attribute(key, value) } if current_span.recording?

  context = OpenTelemetry::Context.current.set_value(EXPERIMENT_ATTRIBUTES_CONTEXT_KEY, frozen_attributes)
  OpenTelemetry::Context.with_current(context, &)
end

.baggage_available? ⇒ Boolean

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Check if baggage API is available



431
432
433
 
# File 'lib/langfuse/propagation.rb', line 431

def self.baggage_available?
  defined?(OpenTelemetry::Baggage)
end

.get_propagated_attributes_from_context(context) ⇒ Hash<String, String, Array<String>>

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Get propagated attributes from context for span processor

rubocop:disable Metrics/AbcSize, Metrics/CyclomaticComplexity, Metrics/PerceivedComplexity



166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
 
# File 'lib/langfuse/propagation.rb', line 166

def self.get_propagated_attributes_from_context(context)
  propagated_attributes = _extract_baggage_attributes(context)

  # Handle OTEL context values
  SPAN_KEY_MAP.each_key do |key|
    context_key = _get_propagated_context_key(key)
    value = context.value(context_key)

    next if value.nil?

    span_key = _get_propagated_span_key(key)

    if key == "environment"
      validated_environment = _validate_environment_value(value)
      propagated_attributes[span_key] = validated_environment if validated_environment
    elsif key == "metadata" && value.is_a?(Hash)
      value.each do |k, v|
        metadata_key = "#{OtelAttributes::TRACE_METADATA}.#{k}"
        propagated_attributes[metadata_key] = v.to_s
      end
    elsif key == "tags" && value.is_a?(Array)
      propagated_attributes[span_key] = value unless value.empty?
    else
      propagated_attributes[span_key] = value.to_s
    end
  end
  # rubocop:enable Metrics/AbcSize, Metrics/CyclomaticComplexity, Metrics/PerceivedComplexity

  experiment_attributes = context.value(EXPERIMENT_ATTRIBUTES_CONTEXT_KEY) || {}
  propagated_attributes.merge(experiment_attributes)
end

.propagate_attributes(user_id: nil, session_id: nil, metadata: nil, version: nil, tags: nil, trace_name: nil, release: nil, environment: nil, as_baggage: false) { ... } ⇒ Object

Propagate trace-level attributes to all spans created within this context.

This method sets attributes on the currently active span AND automatically propagates them to all new child spans created within the block. This is the recommended way to set trace-level attributes like user_id, session_id, and metadata dimensions that should be consistently applied across all observations in a trace.

rubocop:disable Metrics/ParameterLists

Examples:

Basic usage

Langfuse.propagate_attributes(user_id: "user_123", session_id: "session_abc") do
  # All spans created here inherit attributes
end

With metadata and tags

Langfuse.propagate_attributes(
  user_id: "user_123",
  metadata: { environment: "production", region: "us-east" },
  tags: ["api", "v2"]
) do
  # All spans inherit these attributes
end

Yields:

  • Block within which attributes are propagated

Raises:

  • (ArgumentError) —

    if no block is given



90
91
92
93
94
95
96
97
98
99
100
 
# File 'lib/langfuse/propagation.rb', line 90

def self.propagate_attributes(user_id: nil, session_id: nil, metadata: nil, version: nil, tags: nil,
                              trace_name: nil, release: nil, environment: nil, as_baggage: false, &block)
  raise ArgumentError, "Block required" unless block

  attributes = {
    "user_id" => user_id, "session_id" => session_id, "metadata" => metadata,
    "version" => version, "tags" => tags, "trace_name" => trace_name,
    "release" => release, "environment" => environment
  }
  _propagate_attributes(attributes, as_baggage: as_baggage, &block)
end