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

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

Overview

Publisher + claim/activation/readiness-commit for the vLLM discovery runner. Mixed into DiscoveryRefresh.

Instance Method Summary collapse

Instance Method Details

#claim_and_activate_instance(name:, instance_cfg:) ⇒ Object

The operator's CONFIG NAME is the instance identity (InstanceKey.instance_id); the derived host:port/ak string is the secondary physical_id (dedup/diagnostics, never identity).



33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
 
# File 'lib/legion/extensions/llm/vllm/runners/discovery_refresh/claim_activation.rb', line 33

def claim_and_activate_instance(name:, instance_cfg:)
  instance_id = name.to_s
  instance_key = Legion::Extensions::Llm::Inventory::Identity::InstanceKey.new(
    provider_family: :vllm, instance_id: instance_id,
    physical_id: derive_physical_id(instance_cfg: instance_cfg)
  )
  offerings = fetch_offerings(instance_cfg: instance_cfg, instance_key: instance_key)
  callable = Legion::Extensions::Llm::Vllm::VllmCallable.new(instance_cfg: instance_cfg, logger: log)
  probe_coordinator = Legion::Extensions::Llm::Inventory::ProbeCoordinator.new(
    instance_key: instance_key, enqueue: build_probe_enqueue(instance_id: instance_id)
  )
  publisher_token = publisher.claim_instance(
    instance_id: instance_id, physical_id: instance_key.physical_id,
    callable: callable, probe_request_handle: probe_coordinator
  )
  state = {
    name: instance_id, instance_key: instance_key, instance_cfg: instance_cfg,
    callable: callable, probe_coordinator: probe_coordinator,
    publisher_token: publisher_token, sequence: 0, offerings: offerings
  }
  Legion::Extensions::Llm::Inventory::WeightReconciler.track_initializing!(
    states: instance_states,
    state_key: instance_id,
    state: state,
    mutex: state_mutex
  )
  perform_readiness(instance_id: instance_id, state: state, offerings: offerings)
end

#perform_readiness(instance_id:, state:, offerings:) ⇒ Object

Run a readiness probe and commit the outcome. While the instance is still :initializing a passing probe re-activates it (D4); once activated it reports success/failure against the availability fact.



66
67
68
69
70
71
72
73
74
75
76
77
78
79
 
# File 'lib/legion/extensions/llm/vllm/runners/discovery_refresh/claim_activation.rb', line 66

def perform_readiness(instance_id:, state:, offerings:)
  reconcile_weight_snapshot(
    instance_id: instance_id, state: state, discovered_offerings: offerings
  )
  probe_token = publisher.readiness_probe_started(
    instance_id: instance_id, publisher_token: state[:publisher_token]
  )
  readiness = check_health(instance_cfg: state[:instance_cfg])
  report_probe_result(instance_id: instance_id, state: state, probe_token: probe_token,
                      readiness: readiness)
rescue StandardError => e
  handle_exception(e, level: :warn, operation: 'vllm.runner.discovery.readiness',
                      instance_id: instance_id)
end

#publisher ⇒ Object

D2: inject the legacy-coordinator compatibility bridge so the SSOT registry also projects into the old Legion::LLM::Inventory during the mixed-version window (no-op when it isn't loaded).



21
22
23
24
25
26
27
28
 
# File 'lib/legion/extensions/llm/vllm/runners/discovery_refresh/claim_activation.rb', line 21

def publisher
  @publisher ||= Legion::Extensions::Llm::Inventory::Publisher.new(
    provider_family: :vllm,
    compatibility_adapter: Legion::Extensions::Llm::Inventory::ScopedRefresher::LegacyCoordinatorAdapter.new(
      provider_family: :vllm
    )
  )
end

#report_probe_result(instance_id:, state:, probe_token:, readiness:) ⇒ Object 



81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
 
# File 'lib/legion/extensions/llm/vllm/runners/discovery_refresh/claim_activation.rb', line 81

def report_probe_result(instance_id:, state:, probe_token:, readiness:)
  committed = if readiness.ready? && tracked_unpublished?(instance_id: instance_id, state: state)
                activate_tracked_state(
                  instance_id: instance_id, state: state, probe_token: probe_token
                )
              else
                report_tracked_readiness(
                  instance_id: instance_id, state: state,
                  probe_token: probe_token, readiness: readiness
                )
              end
  write_instance_health(state) if committed
  committed
rescue StandardError => e
  handle_exception(e, level: :warn, operation: 'vllm.runner.discovery.report_probe',
                      instance_id: instance_id)
  false
end

#report_tracked_readiness(instance_id:, state:, probe_token:, readiness:) ⇒ Object 



106
107
108
109
110
111
112
113
114
115
116
117
118
119
 
# File 'lib/legion/extensions/llm/vllm/runners/discovery_refresh/claim_activation.rb', line 106

def report_tracked_readiness(instance_id:, state:, probe_token:, readiness:)
  state_mutex.synchronize do
    return false unless instance_states[instance_id].equal?(state)

    if readiness.ready?
      publisher.readiness_succeeded(instance_id: instance_id, probe_token: probe_token)
    else
      publisher.readiness_failed(
        instance_id: instance_id, probe_token: probe_token, reason: readiness.reason
      )
    end
    true
  end
end

#tracked_unpublished?(instance_id:, state:) ⇒ Boolean

Returns:

  • (Boolean) 


100
101
102
103
104
 
# File 'lib/legion/extensions/llm/vllm/runners/discovery_refresh/claim_activation.rb', line 100

def tracked_unpublished?(instance_id:, state:)
  state_mutex.synchronize do
    instance_states[instance_id].equal?(state) && !state.fetch(:published)
  end
end