Class: Bitfab::HttpClient

Inherits:
Object
  • Object
show all
Defined in:
lib/bitfab/http_client.rb

Constant Summary collapse

OTLP_TRACES_ENDPOINT = 
"/api/sdk/otel/v1/traces"

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(api_key:, service_url: nil, timeout: 120) ⇒ HttpClient

Returns a new instance of HttpClient.



29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
 
# File 'lib/bitfab/http_client.rb', line 29

def initialize(api_key:, service_url: nil, timeout: 120)
  @api_key = api_key
  @service_url = (service_url || DEFAULT_SERVICE_URL).chomp("/")
  @timeout = timeout
  @transport_mutex = Mutex.new
  # Eager for the same reason the transport's mutex is: `||=` racing between
  # the export threads recording acks would hand each a different mutex.
  @delivery_mutex = Mutex.new
  @trace_deliveries = {}
  @delivery_pid = Process.pid
  @carrier_seq_mutex = Mutex.new
  @carrier_seq = 0
  @transport = nil
  @closed = false
end

Instance Attribute Details

#service_url ⇒ Object (readonly)

Returns the value of attribute service_url.



27
28
29
 
# File 'lib/bitfab/http_client.rb', line 27

def service_url
  @service_url
end

Instance Method Details

#close(timeout: 30) ⇒ Object

Flush and permanently close this client's tracing transport. Returns true when everything it queued was delivered within the deadline.



47
48
49
50
51
52
53
54
55
56
57
 
# File 'lib/bitfab/http_client.rb', line 47

def close(timeout: 30)
  transport = @transport_mutex.synchronize do
    current = @transport
    @transport = nil
    @closed = true
    current
  end
  return true if transport.nil?

  transport.shutdown(timeout)
end

#closed_deliveries?(trace_ids) ⇒ Boolean

Whether any tracked trace has had its closing carrier submitted.

Returns:

  • (Boolean) 


472
473
474
475
476
 
# File 'lib/bitfab/http_client.rb', line 472

def closed_deliveries?(trace_ids)
  @delivery_mutex.synchronize do
    trace_ids.any? { |trace_id| trace_deliveries[trace_id]&.closed }
  end
end

#complete_replay(test_run_id) ⇒ Object

Mark a replay test run as completed. Blocking call.



253
254
255
 
# File 'lib/bitfab/http_client.rb', line 253

def complete_replay(test_run_id)
  request("/api/sdk/replay/complete", {"testRunId" => test_run_id}, timeout: 30)
end

#flush(timeout: 30) ⇒ Object

Wait for the spans and traces this client queued to be delivered.



60
61
62
63
64
65
 
# File 'lib/bitfab/http_client.rb', line 60

def flush(timeout: 30)
  transport = @transport_mutex.synchronize { @transport }
  return true if transport.nil?

  transport.flush(timeout)
end

#get(endpoint, timeout: nil) ⇒ Object

Make a GET request to the Bitfab API. Returns parsed JSON response hash.



135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
 
# File 'lib/bitfab/http_client.rb', line 135

def get(endpoint, timeout: nil)
  uri = URI("#{@service_url}#{endpoint}")
  request_timeout = timeout || @timeout

  http = Net::HTTP.new(uri.host, uri.port)
  http.use_ssl = uri.scheme == "https"
  http.open_timeout = request_timeout
  http.read_timeout = request_timeout

  # request_uri (not path) so any query string on the endpoint survives.
  req = Net::HTTP::Get.new(uri.request_uri, headers)
  response = http.request(req)

  unless response.is_a?(Net::HTTPSuccess)
    raise Net::HTTPError.new("HTTP #{response.code}: #{response.body}", response)
  end

  result = JSON.parse(response.body)

  if result["error"]
    msg = result["error"]
    msg = "#{msg} Configure it at: #{@service_url}#{result["url"]}" if result["url"]
    raise StandardError, msg
  end

  result
end

#get_external_span(span_id, replay_view: false) ⇒ Object

Fetch an external span by ID. Blocking GET request. The replay view keeps only the input/output serialization fields used by replay.



214
215
216
217
 
# File 'lib/bitfab/http_client.rb', line 214

def get_external_span(span_id, replay_view: false)
  query = replay_view ? "?view=replay" : ""
  get("/api/sdk/externalSpans/#{span_id}#{query}", timeout: 30)
end

#get_replay_status(test_run_id, expected_span_counts) ⇒ Object

Read the replay traces the server has fully persisted so far.



288
289
290
291
292
293
294
 
# File 'lib/bitfab/http_client.rb', line 288

def get_replay_status(test_run_id, expected_span_counts)
  request(
    "/api/sdk/replay/status",
    {"testRunId" => test_run_id, "expectedSpanCounts" => expected_span_counts},
    timeout: 30
  )
end

#get_span_tree(external_span_id, include_outputs: true, include_root_output: true) ⇒ Object

Fetch the span tree rooted at an external span. Blocking GET request. Used by replay when a mock strategy is active so child spans can be matched against their historical outputs.

Returns a hash shaped { "root" => SpanTreeNode } where each node has sourceSpanId, externalSpanId, traceFunctionKey, spanName, type, and children. When include_outputs is true each node also carries its recorded output (and optional outputMeta); when false the server omits the output payloads so only the spans actually mocked are fetched later (by externalSpanId), avoiding dragging down every span's output when only a few are mocked.



243
244
245
246
247
248
249
250
 
# File 'lib/bitfab/http_client.rb', line 243

def get_span_tree(external_span_id, include_outputs: true, include_root_output: true)
  endpoint = "/api/sdk/replay/spanTree/#{external_span_id}"
  query = []
  query << "includeOutputs=false" unless include_outputs
  query << "includeRootOutput=false" unless include_root_output
  endpoint += "?#{query.join("&")}" unless query.empty?
  get(endpoint, timeout: 30)
end

#get_trace_span(trace_id, id: nil, name: nil, occurrence: "last") ⇒ Object 



219
220
221
222
223
224
225
226
227
228
229
230
 
# File 'lib/bitfab/http_client.rb', line 219

def get_trace_span(trace_id, id: nil, name: nil, occurrence: "last")
  query = if id
    {id:}
  else
    {name:, occurrence: occurrence.to_s}
  end
  encoded_trace_id = URI.encode_www_form_component(trace_id)
  get(
    "/api/sdk/traces/#{encoded_trace_id}/span?#{URI.encode_www_form(query)}",
    timeout: 30
  )["span"]
end

#release_db_branch_lease(neon_branch_id) ⇒ Object

Release a previously-resolved DB branch by deleting its Neon branch. Blocking call. Idempotent server-side (a missing branch is treated as already released).



260
261
262
 
# File 'lib/bitfab/http_client.rb', line 260

def release_db_branch_lease(neon_branch_id)
  request("/api/sdk/replay/releaseDbBranchLease", {"neonBranchId" => neon_branch_id}, timeout: 30)
end

#request(endpoint, payload, timeout: nil, max_retries: 1, retry_delay: 0.1) ⇒ Object

Make a POST request to the Bitfab API. Returns parsed JSON response hash.



69
70
71
 
# File 'lib/bitfab/http_client.rb', line 69

def request(endpoint, payload, timeout: nil, max_retries: 1, retry_delay: 0.1)
  send_encoded(endpoint, Serialize.safe_generate(payload), timeout:, max_retries:, retry_delay:)
end

#resolve_db_branch_lease(test_run_id, trace_id, db_branch_settings = nil) ⇒ Object 



264
265
266
267
268
269
270
271
272
 
# File 'lib/bitfab/http_client.rb', line 264

def resolve_db_branch_lease(test_run_id, trace_id, db_branch_settings = nil)
  payload = {"testRunId" => test_run_id, "traceId" => trace_id}
  payload["dbBranchSettings"] = db_branch_settings unless db_branch_settings.nil?
  request(
    "/api/sdk/replay/resolveDbBranchLease",
    payload,
    timeout: REPLAY_DB_BRANCH_REQUEST_TIMEOUT_SECONDS
  )
end

#send_encoded(endpoint, body, timeout: nil, max_retries: 1, retry_delay: 0.1) ⇒ Object

POST an already-encoded body. The span transport encodes its own batches, so routing them back through #request would encode the same data twice.



75
76
77
78
79
80
81
82
83
 
# File 'lib/bitfab/http_client.rb', line 75

def send_encoded(endpoint, body, timeout: nil, max_retries: 1, retry_delay: 0.1)
  send_prepared(
    endpoint,
    Compress.prepare_request_body(body),
    timeout:,
    max_retries:,
    retry_delay:
  )
end

#send_external_span(payload) ⇒ Object

Queue an external span on this client's trace transport (fire-and-forget).



125
126
127
128
129
130
131
 
# File 'lib/bitfab/http_client.rb', line 125

def send_external_span(payload)
  trace_transport&.submit(
    "external_span",
    payload.merge("sdkVersion" => VERSION),
    recorded_meta("external_span", payload, carrier_ref(payload))
  )
end

#send_external_trace(payload) ⇒ Object

Queue an external trace on this client's trace transport (fire-and-forget).



275
276
277
278
279
280
281
282
283
284
285
 
# File 'lib/bitfab/http_client.rb', line 275

def send_external_trace(payload)
  trace_transport&.submit(
    "external_trace",
    payload.merge("sdkVersion" => VERSION),
    recorded_meta(
      "external_trace",
      payload,
      (payload["completed"] == true) ? carrier_ref(payload) : nil
    )
  )
end

#send_prepared(endpoint, prepared, timeout: nil, max_retries: 1, retry_delay: 0.1) ⇒ Object 



85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
 
# File 'lib/bitfab/http_client.rb', line 85

def send_prepared(endpoint, prepared, timeout: nil, max_retries: 1, retry_delay: 0.1)
  uri = URI("#{@service_url}#{endpoint}")
  request_timeout = timeout || @timeout

  last_error = nil

  max_retries.times do |attempt|
    http = Net::HTTP.new(uri.host, uri.port)
    http.use_ssl = uri.scheme == "https"
    http.open_timeout = request_timeout
    http.read_timeout = request_timeout

    req = Net::HTTP::Post.new(uri.path, headers)
    req["Content-Encoding"] = prepared.content_encoding if prepared.content_encoding
    req.body = prepared.body

    response = http.request(req)

    unless response.is_a?(Net::HTTPSuccess)
      raise Net::HTTPError.new("HTTP #{response.code}: #{response.body}", response)
    end

    result = JSON.parse(response.body)

    if result["error"]
      msg = result["error"]
      msg = "#{msg} Configure it at: #{@service_url}#{result["url"]}" if result["url"]
      raise StandardError, msg
    end

    return result
  rescue => e
    last_error = e
    sleep(retry_delay) if attempt < max_retries - 1
  end

  raise last_error
end

#start_replay(trace_function_key, limit, trace_ids: nil, code_change_description: nil, code_change_files: nil, experiment_group_id: nil, name: nil, include_db_branch_lease: false, dataset_id: nil, grader_ids: nil, db_branch_settings: nil) ⇒ Object

Start a replay session by fetching historical traces. Blocking call. Returns hash with testRunId, testRunUrl, and items array.

Parameters:

  • code_change_description (String, nil) (defaults to: nil) —

    optional rationale for the code change being tested in this replay

  • code_change_files (Array<Hash>, nil) (defaults to: nil) —

    optional list of edited files, each as { path:, before:, after: } (use "" for new/deleted files)

  • experiment_group_id (String, nil) (defaults to: nil) —

    optional UUID grouping multiple replay runs into a single experiment batch

  • name (String, nil) (defaults to: nil) —

    optional display name for the resulting experiment/test run

  • dataset_id (String, nil) (defaults to: nil) —

    optional UUID of the dataset this replay runs against, stored on the resulting experiment for durable attribution

  • grader_ids (Array<String>, nil) (defaults to: nil) —

    optional UUIDs of graders attached directly to this experiment, graded as the union with the dataset's runnable graders at completion; each must be an active/live grader in the same org and trace function or the server rejects the replay



180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
 
# File 'lib/bitfab/http_client.rb', line 180

def start_replay(trace_function_key, limit, trace_ids: nil, code_change_description: nil,
  code_change_files: nil, experiment_group_id: nil, name: nil, include_db_branch_lease: false, dataset_id: nil,
  grader_ids: nil, db_branch_settings: nil)
  payload = {
    "traceFunctionKey" => trace_function_key
  }
  # limit is only meaningful without trace_ids (an explicit ID list
  # already determines the count), so it's omitted when nil.
  payload["limit"] = limit unless limit.nil?
  payload["traceIds"] = trace_ids if trace_ids
  payload["name"] = name unless name.nil?
  payload["codeChangeDescription"] = code_change_description unless code_change_description.nil?
  payload["codeChangeFiles"] = normalize_code_change_files(code_change_files) unless code_change_files.nil?
  payload["experimentGroupId"] = experiment_group_id unless experiment_group_id.nil?
  payload["includeDbBranchLease"] = true if include_db_branch_lease
  payload["lazyDbBranchLease"] = true if include_db_branch_lease
  payload["datasetId"] = dataset_id unless dataset_id.nil?
  payload["graderIds"] = grader_ids unless grader_ids.nil?
  payload["dbBranchSettings"] = db_branch_settings unless db_branch_settings.nil?

  # When DB branching is on, the server resolves a Neon preview branch per
  # item (snapshot + restore + poll), which can run several seconds each,
  # and runs any warm-up SQL against each branch on a 240s budget of its
  # own. The server gives up at 280s and answers, so this is a backstop for
  # a reply that never comes rather than the thing that normally fires; it
  # sits above the server's own ceiling so the server's error is the one
  # callers see. Net::HTTP would otherwise default to 60s here, which the
  # branch creation alone can outlast.
  timeout = include_db_branch_lease ? REPLAY_DB_BRANCH_REQUEST_TIMEOUT_SECONDS : 30
  request("/api/sdk/replay/start", payload, timeout:)
end

#take_trace_deliveries(trace_ids) ⇒ Object

Report what each tracked trace submitted and whether the server confirmed it, and stop tracking them. Every id passed is freed, so a caller cannot leak a record for a trace that never closed.

delivered is only meaningful once a flush has settled: acks land before an export returns, so a flush that reported success has already collected every ack it is going to collect.



485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
 
# File 'lib/bitfab/http_client.rb', line 485

def take_trace_deliveries(trace_ids)
  @delivery_mutex.synchronize do
    trace_ids.each_with_object({}) do |trace_id, acc|
      delivery = trace_deliveries.delete(trace_id)
      next if delivery.nil?

      acc[trace_id] = DeliveryReport.new(
        span_count: delivery.submitted_span_ids.size,
        closed: delivery.closed,
        delivered: delivery.closing_acked &&
          delivery.submitted_span_ids.subset?(delivery.acked_span_ids)
      )
    end
  end
end

#track_trace_deliveries(trace_ids) ⇒ Object

Start tracking delivery for trace_ids. Nothing is recorded for a trace that was never tracked, so ordinary tracing costs no bookkeeping at all.



460
461
462
463
464
465
466
467
468
469
 
# File 'lib/bitfab/http_client.rb', line 460

def track_trace_deliveries(trace_ids)
  @delivery_mutex.synchronize do
    trace_ids.each do |trace_id|
      trace_deliveries[trace_id] ||= TraceDelivery.new(
        submitted_span_ids: Set.new, acked_span_ids: Set.new,
        closed: false, closing_acked: false
      )
    end
  end
end