Class: Batchwatch::Client

Inherits:
Object
  • Object
show all
Defined in:
lib/batchwatch/client.rb

Overview

The client. Every submission is non-blocking and fails open.

Args:

token: API key. Falls back to $BATCHWATCH_TOKEN. Optional for
measurement, required to replay a spool.
base_url: Default $BATCHWATCH_URL or https://batchwatch.dev
timeout: Seconds per HTTP call. Default $BATCHWATCH_TIMEOUT or 2.0.
enabled: false turns every network call into a no-op.
spool: Path to the spool file, nil to turn it off. Default
$BATCHWATCH_SPOOL or a file in the temp directory. Spooling is inactive
without a token, because the replay route requires one.
logger: Optional logger. All swallowed errors go to logger.debug.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(token: nil, base_url: nil, timeout: nil, enabled: true, spool: UNSET, logger: nil) ⇒ Client

Returns a new instance of Client.



236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
 
# File 'lib/batchwatch/client.rb', line 236

def initialize(token: nil, base_url: nil, timeout: nil, enabled: true,
               spool: UNSET, logger: nil)
  @token = token || ENV["BATCHWATCH_TOKEN"]
  @base = (base_url || Batchwatch.default_url).sub(%r{/+\z}, "")
  @timeout = timeout.nil? ? Batchwatch.default_timeout : timeout
  @enabled = enabled
  @logger = logger
  @threads = []
  @threads_lock = Mutex.new

  path = spool.equal?(UNSET) ? Batchwatch.default_spool : spool
  # Without a key /v1/calls/complete cannot accept it, so a spool file could
  # never be sent. So we do not write it.
  @spool = (path && @token) ? Spool.new(path, logger: logger) : nil
  if path && !@token
    debug("batchwatch: no token - spool disabled " \
          "(/v1/calls/complete requires a key)")
  end
  @spool_last = 0.0
  @spool_lock = Mutex.new
  # Last advice per model (#101). When should_batch answers, we store here
  # what we recommended + the quoted percentiles, so the NEXT completion for
  # the same model can attach them. The correlation is a documented
  # approximation: latest-advice-per-model, not per-job.
  @advice = {}
  @advice_lock = Mutex.new
end

Instance Attribute Details

#base ⇒ Object (readonly)

Returns the value of attribute base.



231
232
233
 
# File 'lib/batchwatch/client.rb', line 231

def base
  @base
end

#enabled ⇒ Object (readonly)

Returns the value of attribute enabled.



231
232
233
 
# File 'lib/batchwatch/client.rb', line 231

def enabled
  @enabled
end

#spool ⇒ Object (readonly)

Returns the value of attribute spool.



231
232
233
 
# File 'lib/batchwatch/client.rb', line 231

def spool
  @spool
end

#timeout ⇒ Object (readonly)

Returns the value of attribute timeout.



231
232
233
 
# File 'lib/batchwatch/client.rb', line 231

def timeout
  @timeout
end

#token ⇒ Object (readonly)

Returns the value of attribute token.



231
232
233
 
# File 'lib/batchwatch/client.rb', line 231

def token
  @token
end

Instance Method Details

#advice(model, max_wait: nil, provider: "openai", input_tokens: nil, output_tokens: nil, max_tokens: nil, risk: "p90") ⇒ Object

The whole verdict. Returns nil if we cannot answer.

output_tokens is usually UNKNOWN at this point - the model decides them. The default is therefore nil, not zero. Sending zero would make the server compute the saving on zero output, and output costs five to six times as much as input: the answer would be systematically too low, without anyone seeing it. If you know a ceiling, pass max_tokens.



363
364
365
366
367
368
369
370
371
372
373
374
375
376
 
# File 'lib/batchwatch/client.rb', line 363

def advice(model, max_wait: nil, provider: "openai",
           input_tokens: nil, output_tokens: nil, max_tokens: nil,
           risk: "p90")
  q = { "provider" => provider, "model" => model, "risk" => risk }
  { "input_tokens" => input_tokens, "output_tokens" => output_tokens,
    "max_tokens" => max_tokens }.each do |name, value|
    q[name] = value.to_i unless value.nil?
  end
  q["max_wait"] = max_wait unless max_wait.nil?
  call("/v1/should-i-batch?#{URI.encode_www_form(q)}")
rescue StandardError => e
  debug("batchwatch advice failed: #{e}")
  nil
end

#call(path, method: "GET", body: nil, timeout: nil, idem: nil) ⇒ Object

One HTTP call. Raises on error - only public methods swallow.

idem is an optional Idempotency-Key. On the write routes (/v1/calls and /v1/calls/complete) it lets the server dedupe a resend: the same body + the same key writes the measurement once, no matter how many times a die-and-reflush sends it (#30). The key is derived from the measurement, so the replay carries the same key as the first attempt. nil = no header.

Raises:



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
303
304
305
 
# File 'lib/batchwatch/client.rb', line 273

def call(path, method: "GET", body: nil, timeout: nil, idem: nil)
  uri = URI.parse(@base + path)
  deadline = timeout || @timeout

  req_class = case method
              when "GET" then Net::HTTP::Get
              when "POST" then Net::HTTP::Post
              when "PATCH" then Net::HTTP::Patch
              else raise ArgumentError, "unknown method #{method}"
              end
  req = req_class.new(uri)
  req["user-agent"] = "batchwatch-ruby/#{VERSION}"
  unless body.nil?
    req["content-type"] = "application/json"
    req.body = JSON.generate(body)
  end
  req["authorization"] = "Bearer #{@token}" if @token
  req["idempotency-key"] = idem if idem && !idem.empty?

  http = Net::HTTP.new(uri.host, uri.port)
  http.use_ssl = (uri.scheme == "https")
  # Both the connect and the read timeout, so a server that accepts but
  # never answers cannot hold us longer than the deadline.
  http.open_timeout = deadline
  http.read_timeout = deadline

  res = http.request(req)
  code = res.code.to_i
  raw = res.body || ""
  raise HTTPError.new(code, raw) if code < 200 || code >= 300

  raw.strip.empty? ? nil : JSON.parse(raw)
end

#flush(timeout: 5.0) ⇒ Object

Wait for outstanding submissions. Call it before the process exits.



331
332
333
334
335
336
337
338
339
340
 
# File 'lib/batchwatch/client.rb', line 331

def flush(timeout: 5.0)
  deadline = monotonic + timeout
  snapshot = @threads_lock.synchronize { @threads.dup }
  snapshot.each do |t|
    rest = deadline - monotonic
    t.join(rest > 0 ? rest : 0)
  end
  @threads_lock.synchronize { @threads.reject! { |t| !t.alive? } }
  @threads_lock.synchronize { @threads.empty? }
end

#flush_spool(timeout: nil) ⇒ Object

Send everything waiting on disk. Returns the number accepted.

Synchronous and safe to call from a shutdown hook. Never raises. Records the server rejects as invalid are dropped - they will never become valid

  • and the count is logged.


457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
 
# File 'lib/batchwatch/client.rb', line 457

def flush_spool(timeout: nil)
  return 0 if !@spool || !@enabled

  records = @spool.take
  return 0 if records.empty?

  sent = 0
  rest = records.dup
  until rest.empty?
    group = rest.shift(MAX_BATCH)
    begin
      # The key is derived from the group itself: a replay after a crash
      # chunks the records identically (the spool preserves the order), so
      # the key is the same and the server dedupes the double write (#30).
      r = call("/v1/calls/complete", method: "POST", body: group,
                                     timeout: timeout || [@timeout, 10.0].max,
                                     idem: Batchwatch.idem_complete(group))
    rescue StandardError => e
      debug("batchwatch: spool could not be sent: #{e}")
      # The group that failed stays put along with the rest.
      @spool.keep(group + rest)
      return sent
    end
    sent += (r || {}).fetch("accepted", 0)
    rejected = (r || {}).fetch("rejected", 0)
    if rejected && rejected > 0
      debug("batchwatch: #{rejected} spooled measurements were rejected and dropped")
    end
  end
  @spool.keep([])
  debug("batchwatch: #{sent} spooled measurements sent")
  sent
end

#get_advice(model) ⇒ Object

The latest advice for a model, or nil. The approximation is latest-advice-per-model: the newest should_batch attaches to the next completion of the same model.



416
417
418
 
# File 'lib/batchwatch/client.rb', line 416

def get_advice(model)
  @advice_lock.synchronize { @advice[model] }
end

#in_background(&block) ⇒ Object

Run on a background thread. Errors are swallowed - this must never take the caller's own call down with it.



309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
 
# File 'lib/batchwatch/client.rb', line 309

def in_background(&block)
  return nil unless @enabled

  t = Thread.new do
    begin
      block.call
    rescue HTTPError => e
      debug("batchwatch #{e.status}: #{e.body}")
    rescue StandardError => e
      debug("batchwatch unavailable: #{e}")
    end
  end
  # Clean up finished threads, so the list does not grow without bound in a
  # long-running process.
  @threads_lock.synchronize do
    @threads.reject! { |x| !x.alive? }
    @threads << t
  end
  t
end

#maybe_flush_spool ⇒ Object

Called from the background thread AFTER a successful call - so we know the network is up right now, and we avoid hammering a server that is not answering anyway.



494
495
496
497
498
499
500
501
502
503
504
505
506
 
# File 'lib/batchwatch/client.rb', line 494

def maybe_flush_spool
  return unless @spool

  now = monotonic
  @spool_lock.synchronize do
    return if now - @spool_last < SPOOL_INTERVAL_S

    @spool_last = now
  end
  flush_spool
rescue StandardError => e
  debug("batchwatch: spool drain failed: #{e}")
end

#remember_advice(model, acted_verdict, max_wait, answer) ⇒ Object

Store the advice we just gave for THIS model (#101). Only numbers and our own decision end up here - no user data. The quoted percentiles are taken from the server's response; if one of them is missing, we store nil and simply do not attach it (rule #30: nothing invented as 0).



402
403
404
405
406
407
408
409
410
411
 
# File 'lib/batchwatch/client.rb', line 402

def remember_advice(model, acted_verdict, max_wait, answer)
  num = ->(v) { v.is_a?(Numeric) ? v.to_f : nil }
  advice = {
    "acted_verdict" => acted_verdict,
    "deadline_s" => Batchwatch.seconds(max_wait),
    "quoted_p50_s" => num.call(answer["p50_s"]),
    "quoted_p90_s" => num.call(answer["p90_s"])
  }
  @advice_lock.synchronize { @advice[model] = advice }
end

#should_batch(model, max_wait: nil, default: false, **kw) ⇒ Object

true/false. On any doubt you get your default back.

We never guess on the caller's behalf: if we cannot answer, the caller gets their own predetermined value. The default is false - "run it synchronously" - which is the safe way to be wrong, because a synchronous call just costs more, while an unexpected eight-hour queue can take a product down.



385
386
387
388
389
390
391
392
393
394
395
396
 
# File 'lib/batchwatch/client.rb', line 385

def should_batch(model, max_wait: nil, default: false, **kw)
  r = advice(model, max_wait: max_wait, **kw)
  return default unless r

  answer = case r["verdict"]
           when "run_batch" then true
           when "run_sync", "batch_at" then false
           else default
           end
  remember_advice(model, answer, max_wait, r)
  answer
end

#spool_measurement(body) ⇒ Object

Store a COMPLETED measurement that could not be delivered.



509
510
511
512
513
514
515
 
# File 'lib/batchwatch/client.rb', line 509

def spool_measurement(body)
  if @spool
    @spool.append(body)
  else
    debug("batchwatch: the measurement was lost (no spool)")
  end
end

#track(model, provider: "openai", mode: "batch", requests: 1, input_tokens: nil, endpoint: nil) ⇒ Object

Measure one call. The submission happens in the background.

No exception from batchwatch ever reaches the caller. An exception from the caller's own block is recorded as "failed" and re-raised untouched.

With a block it acts like the Python context manager: it closes the measurement for you, including on the error path. Without a block it returns the tracking handle and you call done() yourself.



430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
 
# File 'lib/batchwatch/client.rb', line 430

def track(model, provider: "openai", mode: "batch", requests: 1,
          input_tokens: nil, endpoint: nil)
  t = Tracking.new(self, provider, model, mode, requests, input_tokens,
                   endpoint)
  t.start
  return t unless block_given?

  begin
    yield t
  rescue Exception # rubocop:disable Lint/RescueException
    # Everything - including Interrupt/SignalException - is recorded as
    # failed and re-raised untouched. We swallow our own errors, never the
    # caller's.
    t.done(status: "failed")
    raise
  else
    t.done unless t.finished?
  end
end

#wait_now(model, provider: "openai", mode: "batch") ⇒ Object

What the queue is doing right now, or nil if we cannot say.



345
346
347
348
349
350
351
352
353
354
 
# File 'lib/batchwatch/client.rb', line 345

def wait_now(model, provider: "openai", mode: "batch")
  q = URI.encode_www_form(provider: provider, model: model, mode: mode)
  r = call("/v1/wait?#{q}")
  return nil if !r || r["verdict"] == "insufficient_data"

  r
rescue StandardError => e
  debug("batchwatch wait failed: #{e}")
  nil
end