Class: Legion::LLM::Router::HealthTracker

Inherits:
Object
  • Object
show all
Includes:
Legion::Logging::Helper
Defined in:
lib/legion/llm/router/health_tracker.rb

Constant Summary collapse

OPEN_PENALTY = 
-50
LATENCY_THRESHOLD_MS = 
5000
LATENCY_PENALTY_STEP = 
-10

Instance Method Summary collapse

Constructor Details

#initialize(window_seconds: 300, failure_threshold: 3, cooldown_seconds: 60, sweep_interval_seconds: 5) ⇒ HealthTracker

Returns a new instance of HealthTracker.



14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
 
# File 'lib/legion/llm/router/health_tracker.rb', line 14

def initialize(window_seconds: 300, failure_threshold: 3, cooldown_seconds: 60, sweep_interval_seconds: 5)
  @window_seconds         = window_seconds
  @failure_threshold      = failure_threshold
  @cooldown_seconds       = cooldown_seconds
  @sweep_interval_seconds = sweep_interval_seconds

  @circuits        = {}
  @latency_window  = {}
  @handlers        = {}
  @denied_models   = {}
  @last_sweep_at   = Time.now
  @mutex           = Monitor.new

  register_default_handlers
end

Instance Method Details

#adjustment(provider, instance: nil, offering_id: nil) ⇒ Object

Returns total priority adjustment for a provider. Combines circuit-breaker penalty and latency penalty. When instance: is given, returns that specific instance's adjustment. When nil, returns the average across all known instances so one bad node penalizes the provider proportionally instead of globally.



78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
 
# File 'lib/legion/llm/router/health_tracker.rb', line 78

def adjustment(provider, instance: nil, offering_id: nil)
  if instance
    key = instance_key(provider, instance)
    return circuit_adjustment(key) + latency_adjustment(key)
  end

  # Check for known instances — return average adjustment if any exist.
  instances = known_instances(provider)
  if instances.empty?
    # Backward compat: use provider-level or offering-level key
    key = health_key(provider, offering_id)
    key = provider if offering_id && !tracked?(key) && tracked?(provider)
    return circuit_adjustment(key) + latency_adjustment(key)
  end

  adjustments = instances.map { |k| circuit_adjustment(k) + latency_adjustment(k) }
  (adjustments.sum.to_f / adjustments.size).round
end

#clear_denied(provider: nil, instance: nil) ⇒ Object

Clear denied models for a provider (or all if no args).



135
136
137
138
139
140
141
142
143
144
 
# File 'lib/legion/llm/router/health_tracker.rb', line 135

def clear_denied(provider: nil, instance: nil)
  @mutex.synchronize do
    if provider
      key = instance ? instance_key(provider, instance) : provider.to_s
      @denied_models.delete(key)
    else
      @denied_models.clear
    end
  end
end

#denied_models ⇒ Object

List all denied models (for diagnostics).



130
131
132
 
# File 'lib/legion/llm/router/health_tracker.rb', line 130

def denied_models
  @mutex.synchronize { @denied_models.dup }
end

#deny_model(provider:, model:, instance: nil, reason: nil) ⇒ Object

Record that a model is denied for a provider+instance (e.g. AccessDenied). Excluded from routing until restart or explicit clear.



111
112
113
114
115
116
117
118
119
 
# File 'lib/legion/llm/router/health_tracker.rb', line 111

def deny_model(provider:, model:, instance: nil, reason: nil)
  key = instance ? instance_key(provider, instance) : provider.to_s
  @mutex.synchronize do
    @denied_models[key] ||= {}
    @denied_models[key][model.to_s] = { reason: reason, at: Time.now }
  end
  log.warn("[llm][health_tracker] action=model_denied provider=#{key} model=#{model} reason=#{reason}")
  write_deny_to_lane(key: key, model: model)
end

#model_denied?(provider:, model:, instance: nil) ⇒ Boolean

Check if a model is denied for a provider+instance.

Returns:

  • (Boolean) 


122
123
124
125
126
127
 
# File 'lib/legion/llm/router/health_tracker.rb', line 122

def model_denied?(provider:, model:, instance: nil)
  key = instance ? instance_key(provider, instance) : provider.to_s
  @mutex.synchronize do
    !@denied_models.dig(key, model.to_s).nil?
  end
end

#register_handler(signal, &block) ⇒ Object

Register a custom handler for a signal type.



31
32
33
 
# File 'lib/legion/llm/router/health_tracker.rb', line 31

def register_handler(signal, &block)
  @handlers[signal.to_sym] = block
end

#report(provider:, signal:, value:, instance: nil, metadata: {}, offering_id: nil) ⇒ Object

Thread-safe signal intake. Dispatches to the registered handler if one exists. When instance: is given, tracks under "provider/instance". When instance: is nil, tracks under "provider" (backward compat) or broadcasts to all known instances of that provider.



39
40
41
42
43
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
 
# File 'lib/legion/llm/router/health_tracker.rb', line 39

def report(provider:, signal:, value:, instance: nil, metadata: {}, offering_id: nil)
  sym     = signal.to_sym
  handler = @handlers[sym]
  return nil unless handler

  log.debug "[llm][health_tracker] action=signal_received provider=#{provider} instance=#{instance || 'all'} signal=#{sym} value=#{value}"

  if instance
    payload = build_payload(provider: provider, instance: instance,
                            key: instance_key(provider, instance),
                            offering_id: offering_id, signal: sym,
                            value: value, metadata: metadata)
    @mutex.synchronize { handler.call(payload) }
  else
    instances = known_instances(provider)
    if instances.empty?
      payload = build_payload(provider: provider, instance: nil,
                              key: health_key(provider, offering_id),
                              offering_id: offering_id, signal: sym,
                              value: value, metadata: metadata)
      @mutex.synchronize { handler.call(payload) }
    else
      @mutex.synchronize do
        instances.each do |inst_key|
          payload = build_payload(provider: provider, instance: nil,
                                  key: inst_key, offering_id: offering_id,
                                  signal: sym, value: value, metadata: metadata)
          handler.call(payload)
        end
      end
    end
  end
end

#reset(provider, instance: nil, offering_id: nil) ⇒ Object

Clears circuit and latency data for a single provider.



147
148
149
150
151
152
153
 
# File 'lib/legion/llm/router/health_tracker.rb', line 147

def reset(provider, instance: nil, offering_id: nil)
  key = instance ? instance_key(provider, instance) : health_key(provider, offering_id)
  @mutex.synchronize do
    @circuits.delete(key)
    @latency_window.delete(key)
  end
end

#reset_all ⇒ Object

Clears all state.



156
157
158
159
160
161
162
 
# File 'lib/legion/llm/router/health_tracker.rb', line 156

def reset_all
  @mutex.synchronize do
    @circuits.clear
    @latency_window.clear
    @denied_models.clear
  end
end

#sweep_circuits! ⇒ Object

Advance any :open circuits past their cooldown to :half_open and write the updated health to Inventory lanes. This decouples the open→half_open transition from inbound traffic — without it, an excluded lane never receives reports so circuit_state_for_key never fires, and the circuit stays permanently :open (half-open probe starvation).

Called by Router.request_lane on every selection so eligible circuits are already advanced before the soft filter runs. Throttled by sweep_interval_seconds to avoid per-request overhead under high load.



173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
 
# File 'lib/legion/llm/router/health_tracker.rb', line 173

def sweep_circuits!
  now = Time.now
  return if (now - @last_sweep_at) < @sweep_interval_seconds

  @last_sweep_at = now
  @mutex.synchronize do
    @circuits.each do |key, circuit|
      next unless circuit[:state] == :open
      next unless circuit[:opened_at]
      next unless (now - circuit[:opened_at]) >= @cooldown_seconds

      do_transition_circuit!(key: key, to_state: :half_open)
      log.info("[llm][health_tracker] action=sweep_half_open provider=#{key} cooldown_elapsed_s=#{(now - circuit[:opened_at]).round}")
    end
  end
rescue StandardError => e
  handle_exception(e, level: :warn, handled: true,
                      operation: 'health_tracker.sweep_circuits!')
end

#trip_circuit(provider:, instance: nil, reason: nil) ⇒ Object

Open a circuit immediately, bypassing the failure threshold. Used when a single observation is conclusive (e.g., discovery unreachable).



99
100
101
102
103
104
105
106
107
 
# File 'lib/legion/llm/router/health_tracker.rb', line 99

def trip_circuit(provider:, instance: nil, reason: nil)
  key = instance ? instance_key(provider, instance) : provider.to_s
  @mutex.synchronize do
    ensure_circuit(key)
    @circuits[key][:failures] = @failure_threshold.to_f
    do_transition_circuit!(key: key, to_state: :open)
  end
  log.warn("[llm][health_tracker] action=circuit_tripped provider=#{key} reason=#{reason}")
end