Class: Insika::Soak::Runner

Inherits:
Object
  • Object
show all
Defined in:
lib/insika/soak/runner.rb

Overview

The arrival-rate runner: sustain a declared load against a live deploy for N hours, poll vitals hourly, append every observation to the results file as it happens. It computes NO verdict (C5 does, offline) and never retries a turn — a failed turn is evidence, not a problem to smooth over. Threads, not fibers: the tool that measures the reactor should not share its failure modes.

Defined Under Namespace

Classes: Failure, Http, PreflightError

Constant Summary collapse

USAGE = 
<<~TXT
  insika soak — the 72h soak runner

  Usage: insika soak --run | --verify FILE | --preflight [options]

  Modes:
    --run                  fire the soak (mutually exclusive with --verify)
    --verify FILE          re-read an archived run; NO traffic, no network
    --preflight            run every precondition check and exit
    --dry-run              print the plan + one sample request, send nothing

  Options:
    --envelope PATH        the frozen envelope (required)
    --agent ID             the soak agent (default: from envelope)
    --out DIR              where the results file lands (default: soak-out)
    --resume FILE          continue an interrupted run, recording the gap

  Environment:
    INSIKA_URL              base URL of the engine (default: http://localhost:9292)
    OPENCLAW_GATEWAY_TOKEN  Bearer; falls back to ADMIN_TOKEN, then "local-demo"
TXT

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(envelope:, out: "soak-out", agent: nil, env: ENV, http: nil, seed: nil, clock: nil, sleeper: nil, stdout: $stdout, stderr: $stderr) ⇒ Runner

Returns a new instance of Runner.



162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
 
# File 'lib/insika/soak/runner.rb', line 162

def initialize(envelope:, out: "soak-out", agent: nil, env: ENV, http: nil,
               seed: nil, clock: nil, sleeper: nil, stdout: $stdout, stderr: $stderr)
  @envelope = envelope
  @out = out
  @agent = agent || envelope[:agent]
  @env = env
  @http = http || Http.new(token: resolve_token(env))
  @seed = seed || Time.now.to_i
  @clock = clock || -> { Process.clock_gettime(Process::CLOCK_MONOTONIC) }
  @sleeper = sleeper || ->(s) { sleep s }
  @stdout = stdout
  @stderr = stderr
  @stop_reason = nil
  @lanes = []
  @in_flight = 0
  @mutex = Mutex.new
  @write_mutex = Mutex.new
  @cond = ConditionVariable.new
  @vitals_failures = 0
  @vitals_degraded_noted = false
end

Instance Attribute Details

#envelope ⇒ Object (readonly)

Returns the value of attribute envelope.



160
161
162
 
# File 'lib/insika/soak/runner.rb', line 160

def envelope
  @envelope
end

Class Method Details

.arrivals(seed:, rate_per_hour:, count:) ⇒ Object

Poisson inter-arrival seconds: -mean * Math.log(1.0 - rand). Seeded, so a re-run with the same seed is the same arrival sequence. The rate is expressed in the ARRIVAL unit (sessions per hour when session_turns > 1); the envelope's turns_per_hour is divided by session_turns before this.



60
61
62
63
64
 
# File 'lib/insika/soak/runner.rb', line 60

def self.arrivals(seed:, rate_per_hour:, count:)
  rng = Random.new(seed)
  mean = 3600.0 / rate_per_hour
  Array.new(count) { -mean * Math.log(1.0 - rng.rand) }
end

.main(argv, stdout: $stdout, stderr: $stderr, env: ENV, http: nil) ⇒ Object

CLI entry (bin/insika delegates here). -> exit status.



67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
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
123
124
125
126
127
128
 
# File 'lib/insika/soak/runner.rb', line 67

def self.main(argv, stdout: $stdout, stderr: $stderr, env: ENV, http: nil)
  opts = { out: "soak-out" }
  modes = []
  OptionParser.new do |o|
    o.banner = "Usage: insika soak --run | --verify FILE | --preflight [options]"
    o.on("-h", "--help", "show this help") { stdout.puts USAGE; return 0 }
    o.on("--run", "fire the soak") { modes << :run }
    o.on("--verify FILE", "re-read an archived run") { |v| modes << [:verify, v] }
    o.on("--preflight", "run every precondition check") { modes << :preflight }
    o.on("--dry-run", "print the plan, send nothing") { modes << :dry_run }
    o.on("--envelope PATH") { |v| opts[:envelope] = v }
    o.on("--agent ID") { |v| opts[:agent] = v }
    o.on("--out DIR") { |v| opts[:out] = v }
    o.on("--resume FILE") { |v| opts[:resume] = v }
  end.parse!(argv)

  if modes.length != 1
    stderr.puts "insika soak: choose exactly one of --run, --verify, --preflight, --dry-run\n\n#{USAGE}"
    return 2
  end
  if opts[:envelope].nil?
    stderr.puts "insika soak: --envelope PATH is required (the deployment's frozen envelope)\n\n#{USAGE}"
    return 2
  end

  mode = modes.first
  if mode.is_a?(Array) && mode.first == :verify
    return verify(mode[1], envelope_path: opts[:envelope], stdout: stdout, stderr: stderr)
  end

  envelope = Envelope.load(opts[:envelope])
  runner = new(envelope: envelope, agent: opts[:agent], out: opts[:out],
               env: env, http: http, stdout: stdout, stderr: stderr)

  case mode
  when :dry_run
    stdout.puts runner.dry_run_plan
    0
  when :preflight
    failures = runner.preflight
    if failures.empty?
      stdout.puts "preflight OK — every precondition holds"
      0
    else
      failures.each { |f| stderr.puts "preflight #{f.check}: #{f.message}" }
      2
    end
  when :run
    status = runner.run(resume: opts[:resume], traps: true)
    status == :complete ? 0 : 1
  end
rescue OptionParser::ParseError => e
  stderr.puts "insika soak: #{e.message}\n\n#{USAGE}"
  2
rescue PreflightError => e
  e.failures.each { |f| stderr.puts "preflight #{f.check}: #{f.message}" }
  stderr.puts "insika soak: refusing to start"
  2
rescue Insika::ConfigError => e
  stderr.puts "insika soak: #{e.message}"
  2
end

.read_records(path) ⇒ Object

Lazy line reader: .jsonl while running, .jsonl.gz once archived. Lines that fail to parse travel through RAW so the fold can count them (a truncated file must not read as a clean one).



147
148
149
150
151
152
153
154
155
156
157
158
 
# File 'lib/insika/soak/runner.rb', line 147

def self.read_records(path)
  lines = if path.end_with?(".gz")
            Zlib::GzipReader.open(path) { |gz| gz.each_line.to_a }
          else
            File.foreach(path).to_a
          end
  lines.map do |line|
    JSON.parse(line)
  rescue JSON::ParserError
    line
  end
end

.verify(path, envelope_path: nil, stdout: $stdout, stderr: $stderr) ⇒ Object

Pure fold over the archived file — no traffic, no network, no clock.



131
132
133
134
135
136
137
138
139
140
141
142
 
# File 'lib/insika/soak/runner.rb', line 131

def self.verify(path, envelope_path: nil, stdout: $stdout, stderr: $stderr)
  raise Insika::ConfigError, "soak --verify needs --envelope PATH (the frozen envelope the run was declared with)" if envelope_path.nil?

  envelope = Envelope.load(envelope_path)
  records = read_records(path)
  result = Report.fold(records, envelope: envelope)
  stdout.puts result.to_s
  result.pass? ? 0 : 1
rescue Insika::ConfigError => e
  stderr.puts "insika soak: #{e.message}"
  2
end

Instance Method Details

#corpus ⇒ Object 



188
189
190
191
192
193
194
195
196
197
198
 
# File 'lib/insika/soak/runner.rb', line 188

def corpus
  path = @envelope[:corpus]
  raise Insika::ConfigError, "soak envelope declares no corpus" if path.to_s.empty?

  lines = File.readlines(path, chomp: true).map(&:strip).reject(&:empty?)
  raise Insika::ConfigError, "soak corpus #{path} has no usable lines" if lines.empty?

  lines
rescue Errno::ENOENT
  raise Insika::ConfigError, "soak corpus not found: #{path}"
end

#dry_run_plan ⇒ Object 



240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
 
# File 'lib/insika/soak/runner.rb', line 240

def dry_run_plan
  lines = [
    "soak dry-run — no requests sent",
    "  envelope: #{@envelope[:target]} (sha #{@envelope.sha})",
    "  target:   #{target_url}",
    "  agent:    #{@agent}",
    "  shape:    #{@envelope[:turns_per_hour]} turns/h poisson, #{@envelope[:session_turns]}-turn sessions, " \
    "cap #{@envelope[:concurrency_cap]}, #{@envelope[:duration_hours]}h (#{@envelope[:warmup_hours]}h warmup)",
    "  sample:   POST #{URI.join(target_url + '/', 'v1/responses')} model=openclaw:#{@agent} user=soak-1",
    "  out:      #{@out}/"
  ]
  unless @envelope.calibrated?
    lines << "  NOTE: not calibrated — a #{@envelope[:duration_hours]}h run will refuse to start (E1 first)"
  end
  lines.join("\n")
end

#preflight ⇒ Object

The five preconditions (techspec §7.2). Each is a REFUSAL TO START, not a runtime behaviour: a soak must not measure its own limiter, a wrong deploy, or a box that hides what it runs on.



203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
 
# File 'lib/insika/soak/runner.rb', line 203

def preflight
  failures = []
  vitals = @http.get_vitals(target_url)

  if vitals[:status] != 200 || !vitals.dig(:body, "rss_bytes") || vitals.dig(:body, "boot_id").to_s.empty?
    failures << Failure.new("P1", "GET /v1/vitals must return 200 with a non-null rss_bytes and a non-empty boot_id")
  end

  probe = @http.post_turn(target_url, @agent, user: "soak-preflight", message: "ok",
                          timeout: (@envelope[:request_timeout_s] || 120))
  unless probe.dig(:timing).is_a?(Hash) && probe[:timing].any?
    failures << Failure.new("P2", "the probe turn carries no timing block (INSIKA_TURN_TIMING off on the target)")
  end
  unless probe.dig(:usage, "total_tokens").to_i.positive?
    failures << Failure.new("P3", "the probe turn called no model (no usage) — the run would measure the edge limiter")
  end

  pids = [vitals.dig(:body, "pid")]
  2.times do
    @sleeper.call(5)
    pids << @http.get_vitals(target_url).dig(:body, "pid")
  end
  if pids.compact.uniq.length != 1
    failures << Failure.new("P4", "vitals.pid is not stable across three polls — the RSS series would be a random worker per hour")
  end
  if @envelope[:web_concurrency] != 1
    failures << Failure.new("P4", "the envelope declares web_concurrency #{@envelope[:web_concurrency].inspect}, not 1")
  end
  unless URI(target_url).host == @envelope[:target_url_host]
    failures << Failure.new("P5", "target host #{URI(target_url).host.inspect} does not match the envelope's #{@envelope[:target_url_host].inspect}")
  end
  if @envelope[:agent] && @agent != @envelope[:agent]
    failures << Failure.new("P5", "agent #{@agent.inspect} does not match the envelope's #{@envelope[:agent].inspect}")
  end
  failures
end

#run(resume: nil, traps: false) ⇒ Object

Fires the soak. -> :complete | :interrupted | :aborted.

Raises:



258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
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
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
 
# File 'lib/insika/soak/runner.rb', line 258

def run(resume: nil, traps: false)
  unless @envelope.calibrated? || @envelope.dry_run?
    raise Insika::ConfigError, "the #{@envelope[:duration_hours]}h run refuses to start: " \
                               "the envelope is not calibrated (run E1 first, then write the three ceilings)"
  end

  failures = preflight
  raise PreflightError, failures unless failures.empty?

  install_traps if traps

  header, lane_id = resume ? resume_state(resume) : [new_header, 0]
  path = resume || File.join(@out, "#{header['run_id']}.jsonl")
  FileUtils.mkdir_p(File.dirname(path)) unless resume
  @file = File.open(path, "a")
  if resume
    append_gap_record(resume)
  else
    append(header)
  end

  run_id = header["run_id"]
  @stdout.puts "soak -> target=#{target_url} agent=#{@agent} duration=#{@envelope[:duration_hours]}h " \
               "turns=#{@envelope[:turns_per_hour]}/h seed=#{@seed} out=#{path}"
  @stdout.puts "deploys are frozen for the window: any restart (boot_id or pid change) fails the run"

  duration_hours = @envelope[:duration_hours]
  offset_hours = resume ? elapsed_hours(header["started_at"]) : 0
  start = @clock.call
  deadline = start + (duration_hours - offset_hours) * 3600.0
  # The envelope declares TURNS per hour; each arrival is a SESSION of
  # session_turns turns, so arrivals arrive at turns_per_hour/session_turns
  # per hour — otherwise a 60/h x 7-turn envelope fires 420 turns/h.
  sessions_per_hour = @envelope[:turns_per_hour].to_f / @envelope[:session_turns]
  arrivals = self.class.arrivals(seed: @seed, rate_per_hour: sessions_per_hour,
                                 count: (sessions_per_hour * duration_hours).ceil)
  messages = corpus
  next_turn = start
  snapshot_index = 1
  turn_index = 0

  loop do
    break if @stop_reason

    now = @clock.call
    # Hour 1 is due at start+3600 — the LAST snapshot (hour 72) is due
    # exactly at the deadline, so the snapshot fires before the loop
    # ends. Freshly computed from the index, never accumulated (float
    # drift must not eat the final hour).
    snapshot_due = start + snapshot_index * 3600.0
    if now >= snapshot_due - 1e-9 && snapshot_due <= deadline + 1e-9
      append_snapshot(offset_hours + snapshot_index)
      snapshot_index += 1
      next
    end
    break if now >= deadline

    # All scheduled arrivals consumed: nothing turn-shaped is ever due
    # again, so neither the spawn branch nor the wait math may keep
    # waking on next_turn.
    next_turn = Float::INFINITY if turn_index >= arrivals.length

    if now >= next_turn && next_turn <= deadline
      spawn_lane(lane_id, turn_index, messages)
      @lanes.reject! { |t| !t.alive? }
      lane_id += 1
      next_turn += arrivals[turn_index]
      turn_index += 1
      next
    end

    wait = [[snapshot_due, next_turn].min, deadline].min - now
    break if wait <= 0

    @sleeper.call(wait)
  end

  @lanes.each(&:join)
  reason = @stop_reason || "complete"
  append({ "t" => "end", "at" => Time.now.utc.iso8601, "reason" => reason })
  @file.close
  archive_path = gzip!(path)

  result = Report.fold(self.class.read_records(archive_path), envelope: @envelope)
  @stdout.puts result.to_s
  reason == "complete" ? :complete : :interrupted
end

#target_url ⇒ Object 



184
185
186
 
# File 'lib/insika/soak/runner.rb', line 184

def target_url
  (@env["INSIKA_URL"] || @env["HARNESS_URL"] || "http://localhost:9292").sub(%r{/$}, "")
end