Module: Legion::Extensions::Llm::Vllm::Runners::DiscoveryRefresh::Probing

Included in:
Legion::Extensions::Llm::Vllm::Runners::DiscoveryRefresh
Defined in:
lib/legion/extensions/llm/vllm/runners/discovery_refresh/probing.rb

Overview

Tick refresh + cadence/reactive readiness probing for the vLLM discovery runner. Mixed into DiscoveryRefresh.

Instance Method Summary collapse

Instance Method Details

#build_probe_enqueue(instance_id:) ⇒ Object 



80
81
82
83
84
85
86
87
88
89
 
# File 'lib/legion/extensions/llm/vllm/runners/discovery_refresh/probing.rb', line 80

def build_probe_enqueue(instance_id:)
  proc do |request:|
    handle_reactive_probe(instance_id: instance_id, request: request)
    true
  rescue StandardError => e
    handle_exception(e, level: :warn, operation: 'vllm.runner.discovery.probe_enqueue',
                        instance_id: instance_id)
    false
  end
end

#handle_reactive_probe(instance_id:, request:) ⇒ Object 



55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
 
# File 'lib/legion/extensions/llm/vllm/runners/discovery_refresh/probing.rb', line 55

def handle_reactive_probe(instance_id:, request:)
  state = state_mutex.synchronize { instance_states[instance_id] }
  return unless state

  coordinator = state[:probe_coordinator]
  return unless coordinator.begin_probe(request: request)

  probe_token = publisher.readiness_probe_started(
    instance_id: instance_id, publisher_token: state[:publisher_token]
  )
  readiness = check_health(instance_cfg: state[:instance_cfg])
  coordinator.finish_probe(request: request)
  report_probe_result(instance_id: instance_id, state: state, probe_token: probe_token,
                      readiness: readiness)
rescue StandardError => e
  begin
    coordinator.finish_probe(request: request)
  rescue StandardError => finish_e
    handle_exception(finish_e, level: :warn, operation: 'vllm.runner.discovery.reactive_probe.finish',
                               instance_id: instance_id)
  end
  handle_exception(e, level: :warn, operation: 'vllm.runner.discovery.reactive_probe',
                      instance_id: instance_id)
end

#refresh_instance(instance_id:, instance_cfg:) ⇒ Object 



12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
 
# File 'lib/legion/extensions/llm/vllm/runners/discovery_refresh/probing.rb', line 12

def refresh_instance(instance_id:, instance_cfg:)
  state = state_mutex.synchronize { instance_states[instance_id] }
  return unless state

  status = publisher.snapshot.publication_status(instance_key: state[:instance_key])
  if status.state == :initializing
    # D4: an instance that failed readiness at boot stays
    # :initializing until a probe passes. readiness_succeeded is
    # invalid on :initializing, so re-activation is the only path.
    offerings = fetch_offerings(instance_cfg: instance_cfg, instance_key: state[:instance_key])
    perform_readiness(instance_id: instance_id, state: state, offerings: offerings)
    return
  end

  replace_if_changed(instance_id: instance_id, state: state, instance_cfg: instance_cfg)
  run_cadence_probe(instance_id: instance_id, state: state)
end

#replace_if_changed(instance_id:, state:, instance_cfg:) ⇒ Object 



30
31
32
33
34
35
36
 
# File 'lib/legion/extensions/llm/vllm/runners/discovery_refresh/probing.rb', line 30

def replace_if_changed(instance_id:, state:, instance_cfg:)
  new_offerings = fetch_offerings(instance_cfg: instance_cfg, instance_key: state[:instance_key])
  changed = reconcile_weight_snapshot(
    instance_id: instance_id, state: state, discovered_offerings: new_offerings
  )
  write_instance_health(state) if changed
end

#run_cadence_probe(instance_id:, state:) ⇒ Object 



38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
 
# File 'lib/legion/extensions/llm/vllm/runners/discovery_refresh/probing.rb', line 38

def run_cadence_probe(instance_id:, state:)
  coordinator = state[:probe_coordinator]
  return unless coordinator.begin_probe

  probe_token = publisher.readiness_probe_started(
    instance_id: instance_id, publisher_token: state[:publisher_token]
  )
  readiness = check_health(instance_cfg: state[:instance_cfg])
  coordinator.finish_probe
  report_probe_result(instance_id: instance_id, state: state, probe_token: probe_token,
                      readiness: readiness)
rescue StandardError => e
  finish_probe_safely(coordinator)
  handle_exception(e, level: :warn, operation: 'vllm.runner.discovery.cadence_probe',
                      instance_id: instance_id)
end