Class: Datadog::Tracing::Contrib::Cloudwise::Propagation

Inherits:
Object
  • Object
show all
Defined in:
lib/datadog/tracing/contrib/cloudwise/propagation.rb

Overview

Cloudwise 分布式追踪传播器 用于在 Ruby 服务调用 Java 服务时添加 CLOUDWISE 请求头 以及处理外部服务调用 Ruby 时的 CLOUDWISE 请求头

Constant Summary collapse

TYPE_FROM = 
'RUBY'
HEADER_NAME = 
'CLOUDWISE'
HEADER_OTHER_NAME = 
'CLOUDWISE-OTHER'
SERVICE_TYPE_APPLICATION =

服务类型

'APPLICATION'
SERVICE_TYPE_TASK = 
'task'
FIELD_TYPE_FROM =

CLOUDWISE 字段索引

0
FIELD_SAMPLE = 
1
FIELD_HOST_ID = 
2
FIELD_APP_ID = 
3
FIELD_INSTANCE_ID = 
4
FIELD_TRACE_ID = 
5
FIELD_ASSUMED_APP_ID = 
6
FIELD_SPAN_ID = 
7
FIELD_SEGMENT_ID = 
8
FIELD_APP_NAME = 
9
FIELD_OTHER_SERVICE_TYPE_FROM =

CLOUDWISE-OTHER 字段索引

0
FIELD_OTHER_PARENT_SYS = 
1

Class Method Summary collapse

Class Method Details

.build_cloudwise_value(span:, trace:, service_name:, target_url:) ⇒ Object

构建 CLOUDWISE 值 格式：[type_from]:[sample]:[host_id]:[app_id]:[instance_id]:[trace_id]:[assumed_app_id]:[span_id]:[segment_id]:[app_name]



88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
 
# File 'lib/datadog/tracing/contrib/cloudwise/propagation.rb', line 88

def self.build_cloudwise_value(span:, trace:, service_name:, target_url:)
  type_from = TYPE_FROM
  sample = 0
  host_id = get_host_id
  app_id = generate_app_id(service_name)
  instance_id = get_instance_id
  # 如果上游传递了 trace_id_from（例如 Java1 -> Ruby -> Java2），
  # 则使用上游的 trace_id_from 作为透传的 trace_id，保持端到端追踪链路一致
  # 否则使用本服务的低 64 位 trace_id
  upstream_trace_id = trace.get_tag('trace_id_from')
  trace_id = if upstream_trace_id && !upstream_trace_id.empty? && upstream_trace_id != '-1'
    upstream_trace_id
  else
    Tracing::Utils::TraceId.to_low_order(trace.id).to_s
  end
  assumed_app_id = generate_assumed_app_id(target_url)
  span_id = span.id.to_s
  segment_id = get_segment_id(span)
  app_name = service_name

  "#{type_from}:#{sample}:#{host_id}:#{app_id}:#{instance_id}:#{trace_id}:#{assumed_app_id}:#{span_id}:#{segment_id}:#{app_name}"
end

.calculate_sample_rate ⇒ Object

计算采样率 0: 全采样, 1: 不采样



113
114
115
116
117
118
119
120
 
# File 'lib/datadog/tracing/contrib/cloudwise/propagation.rb', line 113

def self.calculate_sample_rate
  sample_rate = Datadog.configuration.tracing.sampling.default_rate
  # 如果采样率 >= 1.0，表示全采样，返回 0
  # 否则返回 1 表示不采样
  (sample_rate && sample_rate >= 1.0) ? 0 : 1
rescue
  1 # 默认不采样
end

.detect_service_type ⇒ String

检测当前服务类型

Returns:

  • 服务类型：'APPLICATION' 或 'task'



413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
 
# File 'lib/datadog/tracing/contrib/cloudwise/propagation.rb', line 413

def self.detect_service_type
  # 优先检查环境变量显式设置
  service_type_env = DATADOG_ENV['CLOUDWISE_SERVICE_TYPE']
  if service_type_env
    return SERVICE_TYPE_TASK if service_type_env.downcase == 'task'
    return SERVICE_TYPE_APPLICATION if service_type_env.downcase == 'application'
  end

  # 检查是否在 Worker 进程中运行（更精确的判断）
  # Sidekiq Worker 进程（检查 Sidekiq 是否作为服务器运行）
  if defined?(::Sidekiq) && ::Sidekiq.respond_to?(:server?) && ::Sidekiq.server?
    return SERVICE_TYPE_TASK
  end

  # Resque Worker 进程
  return SERVICE_TYPE_TASK if defined?(::Resque) && DATADOG_ENV['QUEUE']

  # DelayedJob Worker 进程
  return SERVICE_TYPE_TASK if defined?(::Delayed::Worker) && DATADOG_ENV['DELAYED_JOB']

  # 显式的 Worker 模式
  return SERVICE_TYPE_TASK if DATADOG_ENV['WORKER_MODE'] == 'true'

  # 默认为 Web 应用
  SERVICE_TYPE_APPLICATION
rescue => e
  # 如果检测出错，默认为 Web 应用
  Datadog.logger.debug { "Error detecting service type: #{e.message}, defaulting to APPLICATION" } if defined?(Datadog.logger)
  SERVICE_TYPE_APPLICATION
end

.extract_and_tag_from_header!(span, cloudwise_header) ⇒ Object

从入站请求中提取 CLOUDWISE 头并设置到 span

Parameters:

  • 当前 span

  • CLOUDWISE 请求头的值



280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
 
# File 'lib/datadog/tracing/contrib/cloudwise/propagation.rb', line 280

def self.extract_and_tag_from_header!(span, cloudwise_header)
  return unless span && cloudwise_header

  # 解析 CLOUDWISE 头
  fields = parse_cloudwise_header(cloudwise_header)
  return unless fields

  # 为 span 添加标签（除了 type_from，其他字段都加 _from 后缀）
  tags = {}
  tags['type_from'] = fields[:type_from] if fields[:type_from]
  tags['sample_from'] = fields[:sample] if fields[:sample]
  tags['host_id_from'] = fields[:host_id] if fields[:host_id]
  tags['app_id_from'] = fields[:app_id] if fields[:app_id]
  tags['instance_id_from'] = fields[:instance_id] if fields[:instance_id]
  tags['trace_id_from'] = fields[:trace_id] if fields[:trace_id]
  tags['assumed_app_id_from'] = fields[:assumed_app_id] if fields[:assumed_app_id]
  tags['span_id_from'] = fields[:span_id] if fields[:span_id]
  tags['segment_id_from'] = fields[:segment_id] if fields[:segment_id]
  tags['app_name_from'] = fields[:app_name] if fields[:app_name]

  current_app_id = get_current_app_id
  tags['app_id'] = current_app_id if current_app_id

  span.set_tags(tags) unless tags.empty?

  if defined?(Datadog.logger)
    Datadog.logger.debug do
      "Extracted CLOUDWISE header : cloudwise_header=#{cloudwise_header}"
    end
  end
rescue => e
  if defined?(Datadog.logger)
    Datadog.logger.error do
      "Error extracting CLOUDWISE header: #{e.message}"
    end
  end
end

.extract_other_from_request!(span, request_headers) ⇒ Object

从请求头中提取 CLOUDWISE-OTHER 并添加到根 span

Parameters:

  • 根 span

  • 请求头



472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
 
# File 'lib/datadog/tracing/contrib/cloudwise/propagation.rb', line 472

def self.extract_other_from_request!(span, request_headers)
  return unless span

  # 从请求头中获取 CLOUDWISE-OTHER
  other_header = request_headers[HEADER_OTHER_NAME] ||
    request_headers['HTTP_' + HEADER_OTHER_NAME.upcase.tr('-', '_')]

  # ✅ 优化：使用 set_tags 批量设置，减少锁竞争
  tags = {}

  if other_header
    # 解析 CLOUDWISE-OTHER：service_type_from:parent_sys
    parts = other_header.split(':')

    service_type_from = parts[FIELD_OTHER_SERVICE_TYPE_FROM].to_s.strip
    parent_sys = parts[FIELD_OTHER_PARENT_SYS].to_s.strip

    # 只添加上游相关的字段到 span tags
    # sys 和 service_instance_id 将在 tag_cloudwise_metadata! 中设置
    tags['service_type_from'] = service_type_from
    tags['parent_sys'] = parent_sys

    if defined?(Datadog.logger)
      Datadog.logger.debug do
        "Extracted CLOUDWISE-OTHER from request: service_type_from=#{service_type_from}, parent_sys=#{parent_sys}"
      end
    end
  else
    # 如果没有上游请求头，设置为空字符串
    tags['service_type_from'] = ''
    tags['parent_sys'] = ''

    if defined?(Datadog.logger)
      Datadog.logger.debug do
        "No CLOUDWISE-OTHER header in request, set service_type_from and parent_sys to empty"
      end
    end
  end

  span.set_tags(tags) unless tags.empty?
rescue => e
  if defined?(Datadog.logger)
    Datadog.logger.error do
      "Error extracting CLOUDWISE-OTHER: #{e.message}"
    end
  end
end

.extract_target_url(request) ⇒ Object

从请求对象中提取目标 URL



240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
 
# File 'lib/datadog/tracing/contrib/cloudwise/propagation.rb', line 240

def self.extract_target_url(request)
  return nil unless request

  # 尝试从不同类型的请求对象中提取 URL
  if request.respond_to?(:uri) && request.uri
    request.uri.to_s
  elsif request.respond_to?(:path)
    # Net::HTTP::Request
    host = request['Host'] || 'unknown'
    scheme = request.instance_variable_get(:@is_secure) ? 'https' : 'http'
    "#{scheme}://#{host}#{request.path}"
  end
rescue => e
  Datadog.logger.debug { "Error extracting target URL: #{e.message}" } if defined?(Datadog.logger)
  nil
end

.generate_app_id(service_name) ⇒ Object

生成 app_id（基于服务名称的 MD5）



164
165
166
167
168
169
170
171
172
173
 
# File 'lib/datadog/tracing/contrib/cloudwise/propagation.rb', line 164

def self.generate_app_id(service_name)
  @app_id_cache ||= {}
  @app_id_cache[service_name] ||= begin
    # 计算 MD5 并转换为数字
    md5_hex = Digest::MD5.hexdigest(service_name)
    # 取前15位转为整数（避免超过 Java Long.MAX_VALUE）
    # 15位十六进制 = 60 bits，最大值 2^60-1 = 1152921504606846975（19位数字）
    md5_hex[0..14].to_i(16).to_s
  end
end

.generate_assumed_app_id(url) ⇒ Object

生成 assumed_app_id（基于目标 URL 的域名/IP:端口）



217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
 
# File 'lib/datadog/tracing/contrib/cloudwise/propagation.rb', line 217

def self.generate_assumed_app_id(url)
  return '-1' unless url

  uri = URI.parse(url)
  target = "#{uri.host}:#{uri.port}"

  # 使用与 app_id 相同的生成逻辑
  md5_hex = Digest::MD5.hexdigest(target)
  # 取前15位转为整数（避免超过 Java Long.MAX_VALUE）
  # 15位十六进制 = 60 bits，最大值 2^60-1 = 1152921504606846975（19位数字）
  md5_hex[0..14].to_i(16).to_s
rescue => e
  Datadog.logger.debug { "Error generating assumed_app_id: #{e.message}" } if defined?(Datadog.logger)
  '-1'
end

.generate_service_instance_id ⇒ String

生成 service_instance_id（基于 IP + 进程路径 + PID）

Returns:

  • service_instance_id



392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
 
# File 'lib/datadog/tracing/contrib/cloudwise/propagation.rb', line 392

def self.generate_service_instance_id
  host_ip = get_host_ip
  process_path = $PROGRAM_NAME || ''
  process_pid = Process.pid.to_s

  # 组合：IP + 进程路径 + PID
  combined = "#{host_ip}#{process_path}#{process_pid}"

  # 使用 MD5 生成唯一 ID
  require 'digest/md5'
  md5_hex = Digest::MD5.hexdigest(combined)
  # 取前15位转为整数（避免超过 Java Long.MAX_VALUE）
  md5_hex[0..14].to_i(16).to_s
rescue => e
  Datadog.logger.debug { "Error generating service_instance_id: #{e.message}" } if defined?(Datadog.logger)
  # 降级方案：使用 PID
  Process.pid.to_s
end

.get_account_id ⇒ Object

获取账户 ID TODO: 实际应该通过接口获取，当前暂定默认值 110



147
148
149
 
# File 'lib/datadog/tracing/contrib/cloudwise/propagation.rb', line 147

def self.
  @account_id ||= DATADOG_ENV['CLOUDWISE_ACCOUNT_ID'] || '110'
end

.get_current_app_id ⇒ String?

获取当前应用的 app_id 用于 Consumer 端设置 app_id_to 标签，表示当前消费消息的应用

Returns:

  • 当前应用的 app_id



178
179
180
181
182
 
# File 'lib/datadog/tracing/contrib/cloudwise/propagation.rb', line 178

def self.get_current_app_id
  service_name = Datadog.configuration.service
  return nil unless service_name
  generate_app_id(service_name)
end

.get_host_id ⇒ Object

获取主机 ID（基于 account_id + host_ip 的 MD5） account_id: 账户ID，通过接口获取，暂定默认 110 host_ip: 本机 IP 地址 返回格式: MD5(account_id + host_ip) 的十进制字符串



126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
 
# File 'lib/datadog/tracing/contrib/cloudwise/propagation.rb', line 126

def self.get_host_id
  @host_id ||= begin
     = 
    host_ip = get_host_ip

    # 生成 host_id: MD5(account_id + host_ip)
    combined = "#{account_id}#{host_ip}"
    md5_hex = Digest::MD5.hexdigest(combined)
    # 取前15位转为整数（避免超过 Java Long.MAX_VALUE）
    # 15位十六进制 = 60 bits，最大值 2^60-1 = 1152921504606846975（19位数字）
    md5_hex[0..14].to_i(16).to_s
  rescue => e
    Datadog.logger.debug { "Error generating host_id: #{e.message}" } if defined?(Datadog.logger)
    # 降级方案：使用 hostname 的 MD5
    md5_hex = Digest::MD5.hexdigest(Socket.gethostname)
    md5_hex[0..14].to_i(16).to_s
  end
end

.get_host_ip ⇒ Object

获取本机 IP 地址



152
153
154
155
156
157
158
159
160
161
 
# File 'lib/datadog/tracing/contrib/cloudwise/propagation.rb', line 152

def self.get_host_ip
  @host_ip ||= begin
    # 尝试获取本机 IP（优先非回环的 IPv4 地址）
    ip = Socket.ip_address_list.find { |addr| addr.ipv4? && !addr.ipv4_loopback? }
    ip ? ip.ip_address : '127.0.0.1'
  rescue => e
    Datadog.logger.debug { "Error getting host IP: #{e.message}" } if defined?(Datadog.logger)
    '127.0.0.1'
  end
end

.get_instance_id ⇒ Object

获取实例 ID（基于 host_ip + port 的 MD5） host_ip: 本机 IP 地址 port: 应用端口 返回格式: MD5(host_ip + port) 的十进制字符串



188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
 
# File 'lib/datadog/tracing/contrib/cloudwise/propagation.rb', line 188

def self.get_instance_id
  @instance_id ||= begin
    host_ip = get_host_ip
    port = get_process_port

    # 生成 instance_id: MD5(host_ip + port)
    combined = "#{host_ip}#{port}"
    md5_hex = Digest::MD5.hexdigest(combined)
    # 取前15位转为整数（避免超过 Java Long.MAX_VALUE）
    # 15位十六进制 = 60 bits，最大值 2^60-1 = 1152921504606846975（19位数字）
    md5_hex[0..14].to_i(16).to_s
  rescue => e
    Datadog.logger.debug { "Error generating instance_id: #{e.message}" } if defined?(Datadog.logger)
    # 降级方案：使用进程 PID
    Process.pid.to_s
  end
end

.get_process_port ⇒ Object

获取进程端口（尝试从环境变量获取）



207
208
209
210
211
212
213
214
 
# File 'lib/datadog/tracing/contrib/cloudwise/propagation.rb', line 207

def self.get_process_port
  # 尝试从常见的环境变量中获取端口
  port = DATADOG_ENV['PORT'] || DATADOG_ENV['APP_PORT']
  return port.to_i if port && !port.empty?

  # 默认返回 -1 表示未知端口
  -1
end

.get_segment_id(span) ⇒ Object

获取 segment_id（请求唯一标识） 如果 span 有自定义的 segment_id，使用它；否则返回 -1



235
236
237
 
# File 'lib/datadog/tracing/contrib/cloudwise/propagation.rb', line 235

def self.get_segment_id(span)
  span.get_tag('segment_id') || '-1'
end

.get_sys ⇒ Object

获取当前服务的 sys 值 优先级: Datadog.configuration.cloudwise.sys > CW_SYS 环境变量 > 默认值 'default'



377
378
379
380
381
382
383
384
385
386
387
388
 
# File 'lib/datadog/tracing/contrib/cloudwise/propagation.rb', line 377

def self.get_sys
  # 优先从配置读取
  if defined?(Datadog.configuration) &&
      Datadog.configuration.respond_to?(:cloudwise) &&
      Datadog.configuration.cloudwise.respond_to?(:sys)
    sys = Datadog.configuration.cloudwise.sys
    return sys if sys && !sys.empty? && sys != 'default'
  end

  # 其次从环境变量读取
  DATADOG_ENV['CW_SYS'] || 'default'
end

.inject!(span, trace, request, service_name = nil) ⇒ Object

注入 CLOUDWISE 请求头

Parameters:

  • 当前 span

  • 当前 trace

  • HTTP 请求对象

  • (defaults to: nil)

    服务名称



44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
 
# File 'lib/datadog/tracing/contrib/cloudwise/propagation.rb', line 44

def self.inject!(span, trace, request, service_name = nil)
  return unless span && trace

  service_name ||= Datadog.configuration.service
  return unless service_name

  # 获取目标 URL
  target_url = extract_target_url(request)

  # 生成 assumed_app_id
  assumed_app_id = generate_assumed_app_id(target_url)

  # 构建 CLOUDWISE header 值
  cloudwise_value = build_cloudwise_value(
    span: span,
    trace: trace,
    service_name: service_name,
    target_url: target_url
  )

  # 添加到请求头
  request[HEADER_NAME] = cloudwise_value if cloudwise_value

  # 注入 CLOUDWISE-OTHER 请求头
  inject_other_header!(request)

  # 将 assumed_app_id 添加到 span 的 tags 中
  span.set_tag('assumed_app_id', assumed_app_id) if assumed_app_id && assumed_app_id != '-1'

  if defined?(Datadog.logger)
    Datadog.logger.debug do
      "Injected CLOUDWISE header: #{cloudwise_value}"
    end
  end
rescue => e
  if defined?(Datadog.logger)
    Datadog.logger.error do
      "Error injecting CLOUDWISE header: #{e.message}\n#{e.backtrace.join("\n")}"
    end
  end
end

.inject_other_header!(request) ⇒ Object

注入 CLOUDWISE-OTHER 请求头到下游服务

Parameters:

  • HTTP 请求对象



446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
 
# File 'lib/datadog/tracing/contrib/cloudwise/propagation.rb', line 446

def self.inject_other_header!(request)
  return unless request

  service_type = detect_service_type
  sys = get_sys

  # 构建 CLOUDWISE-OTHER 值：service_type_from:sys
  other_value = "#{service_type}:#{sys}"
  request[HEADER_OTHER_NAME] = other_value

  if defined?(Datadog.logger)
    Datadog.logger.debug do
      "Injected CLOUDWISE-OTHER header: #{other_value}"
    end
  end
rescue => e
  if defined?(Datadog.logger)
    Datadog.logger.error do
      "Error injecting CLOUDWISE-OTHER header: #{e.message}"
    end
  end
end

.normalize_field_value(value) ⇒ String

标准化字段值：去除空格，将 "null" 转换为空字符串

Parameters:

  • 原始字段值

Returns:

  • 标准化后的字段值



367
368
369
370
371
372
373
 
# File 'lib/datadog/tracing/contrib/cloudwise/propagation.rb', line 367

def self.normalize_field_value(value)
  return '-1' if value.nil?

  normalized = value.strip
  # 将字符串 "null" 转换为空字符串
  (normalized == 'null') ? '-1' : normalized
end

.parse_cloudwise_header(header_value) ⇒ Hash?

解析 CLOUDWISE 请求头 格式：[type_from]:[sample]:[host_id]:[app_id]:[instance_id]:[trace_id]:[assumed_app_id]:[span_id]:[segment_id]:[app_name]

Parameters:

  • CLOUDWISE 请求头的值

Returns:

  • 解析后的字段哈希，如果解析失败返回 nil



322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
 
# File 'lib/datadog/tracing/contrib/cloudwise/propagation.rb', line 322

def self.parse_cloudwise_header(header_value)
  return nil unless header_value.is_a?(String)

  # 去除首尾空格
  header_value = header_value.strip
  return nil if header_value.empty?

  # 以冒号分隔字段
  parts = header_value.split(':')

  # 至少需要 10 个字段
  if parts.length < 10
    if defined?(Datadog.logger)
      Datadog.logger.debug do
        "CLOUDWISE header has insufficient fields: expected 10, got #{parts.length}"
      end
    end
    return nil
  end

  # 构建字段哈希（将 "null" 字符串转换为空字符串）
  {
    type_from: parts[FIELD_TYPE_FROM]&.strip,
    sample: normalize_field_value(parts[FIELD_SAMPLE]),
    host_id: normalize_field_value(parts[FIELD_HOST_ID]),
    app_id: parts[FIELD_APP_ID]&.strip,
    instance_id: normalize_field_value(parts[FIELD_INSTANCE_ID]),
    trace_id: parts[FIELD_TRACE_ID]&.strip,
    assumed_app_id: parts[FIELD_ASSUMED_APP_ID]&.strip,
    span_id: parts[FIELD_SPAN_ID]&.strip,
    segment_id: normalize_field_value(parts[FIELD_SEGMENT_ID]),
    app_name: parts[FIELD_APP_NAME]&.strip
  }
rescue => e
  if defined?(Datadog.logger)
    Datadog.logger.error do
      "Error parsing CLOUDWISE header: #{e.message}"
    end
  end
  nil
end

.tag_span_with_app_id(span, service_name) ⇒ Object

为 span 添加 app_id 标签



258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
 
# File 'lib/datadog/tracing/contrib/cloudwise/propagation.rb', line 258

def self.tag_span_with_app_id(span, service_name)
  return unless span && service_name

  app_id = generate_app_id(service_name)
  span.set_tag('app_id', app_id)

  if defined?(Datadog.logger)
    Datadog.logger.debug do
      "Set app_id tag on span: #{app_id} (service: #{service_name})"
    end
  end
rescue => e
  if defined?(Datadog.logger)
    Datadog.logger.error do
      "Error setting app_id tag: #{e.message}"
    end
  end
end