mirror of
https://github.com/basecamp/once-campfire.git
synced 2026-08-28 09:32:37 +09:00
Replace Redis with the Solid trifecta
This commit is contained in:
Stanko K.R.
@@ -71,12 +71,6 @@ jobs:
|
||||
runs-on: ubuntu-latest
|
||||
permissions:
|
||||
contents: read
|
||||
services:
|
||||
redis:
|
||||
image: redis # zizmor: ignore[unpinned-images] -- version tag is fine for service containers
|
||||
ports:
|
||||
- 6379:6379
|
||||
options: --health-cmd "redis-cli ping" --health-interval 10s --health-timeout 5s --health-retries 5
|
||||
steps:
|
||||
- name: Install packages
|
||||
run: sudo apt-get update && sudo apt-get install --no-install-recommends -y libsqlite3-0 libvips curl ffmpeg
|
||||
@@ -88,8 +82,6 @@ jobs:
|
||||
|
||||
- name: Set up Ruby
|
||||
uses: ruby/setup-ruby@319994f95fa847cf3fb3cd3dbe89f6dcde9f178f # v1.295.0
|
||||
env:
|
||||
REDIS_URL: redis://localhost:6379/0
|
||||
with:
|
||||
ruby-version: .ruby-version
|
||||
bundler-cache: true
|
||||
@@ -101,12 +93,6 @@ jobs:
|
||||
runs-on: ubuntu-latest
|
||||
permissions:
|
||||
contents: read
|
||||
services:
|
||||
redis:
|
||||
image: redis # zizmor: ignore[unpinned-images] -- version tag is fine for service containers
|
||||
ports:
|
||||
- 6379:6379
|
||||
options: --health-cmd "redis-cli ping" --health-interval 10s --health-timeout 5s --health-retries 5
|
||||
steps:
|
||||
- name: Install packages
|
||||
run: sudo apt-get update && sudo apt-get install --no-install-recommends -y libsqlite3-0 libvips curl ffmpeg
|
||||
@@ -118,8 +104,6 @@ jobs:
|
||||
|
||||
- name: Set up Ruby
|
||||
uses: ruby/setup-ruby@319994f95fa847cf3fb3cd3dbe89f6dcde9f178f # v1.295.0
|
||||
env:
|
||||
REDIS_URL: redis://localhost:6379/0
|
||||
with:
|
||||
ruby-version: .ruby-version
|
||||
bundler-cache: true
|
||||
|
||||
+1
-1
@@ -9,7 +9,7 @@ WORKDIR /rails
|
||||
|
||||
# Install base packages
|
||||
RUN apt-get update -qq && \
|
||||
apt-get install --no-install-recommends -y curl libsqlite3-0 libvips libjemalloc2 ffmpeg redis && \
|
||||
apt-get install --no-install-recommends -y curl libsqlite3-0 libvips libjemalloc2 ffmpeg && \
|
||||
ln -s /usr/lib/$(uname -m)-linux-gnu/libjemalloc.so.2 /usr/local/lib/libjemalloc.so && \
|
||||
rm -rf /var/lib/apt/lists /var/cache/apt/archive
|
||||
|
||||
|
||||
@@ -9,14 +9,14 @@ gem "benchmark"
|
||||
|
||||
# Drivers
|
||||
gem "sqlite3"
|
||||
gem "redis", "~> 5.4"
|
||||
|
||||
# Deployment
|
||||
gem "puma", "~> 7.2", ">= 7.2.1"
|
||||
|
||||
# Jobs
|
||||
gem "resque", "~> 2.7.0"
|
||||
gem "resque-pool", "~> 0.7.1"
|
||||
# Jobs, caching, WebSockets
|
||||
gem "solid_queue"
|
||||
gem "solid_cache"
|
||||
gem "solid_cable"
|
||||
|
||||
# Assets
|
||||
gem "propshaft", github: "rails/propshaft"
|
||||
@@ -41,7 +41,6 @@ gem "rails_autolink"
|
||||
gem "geared_pagination"
|
||||
gem "jbuilder"
|
||||
gem "net-http-persistent"
|
||||
gem "kredis"
|
||||
gem "platform_agent"
|
||||
gem "thruster"
|
||||
|
||||
|
||||
+25
-38
@@ -166,12 +166,17 @@ GEM
|
||||
drb (2.2.3)
|
||||
erb (6.0.4)
|
||||
erubi (1.13.1)
|
||||
et-orbi (1.4.0)
|
||||
tzinfo
|
||||
faker (3.5.2)
|
||||
i18n (>= 1.8.11, < 2)
|
||||
ffi (1.17.2-aarch64-linux-gnu)
|
||||
ffi (1.17.2-arm64-darwin)
|
||||
ffi (1.17.2-x86_64-darwin)
|
||||
ffi (1.17.2-x86_64-linux-gnu)
|
||||
fugit (1.13.0)
|
||||
et-orbi (~> 1.4)
|
||||
raabro (~> 1.4)
|
||||
geared_pagination (1.2.0)
|
||||
activesupport (>= 5.0)
|
||||
addressable (>= 2.5.0)
|
||||
@@ -194,10 +199,6 @@ GEM
|
||||
json (2.20.0)
|
||||
jwt (3.2.0)
|
||||
base64
|
||||
kredis (1.8.0)
|
||||
activemodel (>= 6.0.0)
|
||||
activesupport (>= 6.0.0)
|
||||
redis (>= 4.2, < 6)
|
||||
language_server-protocol (3.17.0.5)
|
||||
lint_roller (1.1.0)
|
||||
logger (1.7.0)
|
||||
@@ -218,10 +219,6 @@ GEM
|
||||
minitest (5.26.2)
|
||||
mocha (2.7.1)
|
||||
ruby2_keywords (>= 0.0.5)
|
||||
mono_logger (1.1.2)
|
||||
multi_json (1.17.0)
|
||||
mustermann (3.0.4)
|
||||
ruby2_keywords (~> 0.0.1)
|
||||
net-http-persistent (4.0.6)
|
||||
connection_pool (~> 2.2, >= 2.2.4)
|
||||
net-imap (0.6.4.1)
|
||||
@@ -261,12 +258,9 @@ GEM
|
||||
public_suffix (6.0.2)
|
||||
puma (7.2.1)
|
||||
nio4r (~> 2.0)
|
||||
raabro (1.4.0)
|
||||
racc (1.8.1)
|
||||
rack (3.2.6)
|
||||
rack-protection (4.2.1)
|
||||
base64 (>= 0.1.0)
|
||||
logger (>= 1.6.0)
|
||||
rack (>= 3.0.0, < 4)
|
||||
rack-session (2.1.2)
|
||||
base64 (>= 0.1.0)
|
||||
rack (>= 3.0.0)
|
||||
@@ -291,23 +285,9 @@ GEM
|
||||
erb
|
||||
psych (>= 4.0.0)
|
||||
tsort
|
||||
redis (5.4.1)
|
||||
redis-client (>= 0.22.0)
|
||||
redis-client (0.25.2)
|
||||
connection_pool
|
||||
redis-namespace (1.11.0)
|
||||
redis (>= 4)
|
||||
regexp_parser (2.11.2)
|
||||
reline (0.6.3)
|
||||
io-console (~> 0.5)
|
||||
resque (2.7.0)
|
||||
mono_logger (~> 1)
|
||||
multi_json (~> 1.0)
|
||||
redis-namespace (~> 1.6)
|
||||
sinatra (>= 0.9.2)
|
||||
resque-pool (0.7.1)
|
||||
rake (>= 10.0, < 14.0)
|
||||
resque (>= 1.22, < 3)
|
||||
rexml (3.4.4)
|
||||
rqrcode (3.2.0)
|
||||
chunky_png (~> 1.0)
|
||||
@@ -360,13 +340,22 @@ GEM
|
||||
sentry-ruby (5.26.0)
|
||||
bigdecimal
|
||||
concurrent-ruby (~> 1.0, >= 1.0.2)
|
||||
sinatra (4.2.1)
|
||||
logger (>= 1.6.0)
|
||||
mustermann (~> 3.0)
|
||||
rack (>= 3.0.0, < 4)
|
||||
rack-protection (= 4.2.1)
|
||||
rack-session (>= 2.0.0, < 3)
|
||||
tilt (~> 2.0)
|
||||
solid_cable (4.0.2)
|
||||
actioncable (>= 7.2)
|
||||
activejob (>= 7.2)
|
||||
activerecord (>= 7.2)
|
||||
railties (>= 7.2)
|
||||
solid_cache (1.0.10)
|
||||
activejob (>= 7.2)
|
||||
activerecord (>= 7.2)
|
||||
railties (>= 7.2)
|
||||
solid_queue (1.4.0)
|
||||
activejob (>= 7.1)
|
||||
activerecord (>= 7.1)
|
||||
concurrent-ruby (>= 1.3.1)
|
||||
fugit (~> 1.11)
|
||||
railties (>= 7.1)
|
||||
thor (>= 1.3.1)
|
||||
sqlite3 (2.9.5-aarch64-linux-gnu)
|
||||
sqlite3 (2.9.5-arm64-darwin)
|
||||
sqlite3 (2.9.5-x86_64-darwin)
|
||||
@@ -379,7 +368,6 @@ GEM
|
||||
thruster (0.1.15-arm64-darwin)
|
||||
thruster (0.1.15-x86_64-darwin)
|
||||
thruster (0.1.15-x86_64-linux)
|
||||
tilt (2.6.1)
|
||||
timeout (0.6.1)
|
||||
tsort (0.2.0)
|
||||
tzinfo (2.0.6)
|
||||
@@ -423,7 +411,6 @@ DEPENDENCIES
|
||||
image_processing (>= 1.2)
|
||||
importmap-rails!
|
||||
jbuilder
|
||||
kredis
|
||||
mocha
|
||||
net-http-persistent
|
||||
ostruct
|
||||
@@ -432,14 +419,14 @@ DEPENDENCIES
|
||||
puma (~> 7.2, >= 7.2.1)
|
||||
rails!
|
||||
rails_autolink
|
||||
redis (~> 5.4)
|
||||
resque (~> 2.7.0)
|
||||
resque-pool (~> 0.7.1)
|
||||
rqrcode
|
||||
rubocop-rails-omakase
|
||||
selenium-webdriver
|
||||
sentry-rails
|
||||
sentry-ruby
|
||||
solid_cable
|
||||
solid_cache
|
||||
solid_queue
|
||||
sqlite3
|
||||
stimulus-rails
|
||||
thruster
|
||||
|
||||
@@ -1,3 +1,2 @@
|
||||
web: bundle exec thrust bin/start-app
|
||||
redis: redis-server config/redis.conf
|
||||
workers: FORK_PER_JOB=false INTERVAL=0.1 bundle exec resque-pool
|
||||
workers: bundle exec bin/jobs
|
||||
|
||||
@@ -0,0 +1,186 @@
|
||||
#!/usr/bin/env ruby
|
||||
require "optparse"
|
||||
require "json"
|
||||
require "net/http"
|
||||
require "fileutils"
|
||||
require "etc"
|
||||
require "time"
|
||||
|
||||
options = {
|
||||
users: 3000, ramp: 480, hold: 120, cpus: 2, memory: "2g",
|
||||
send_interval: 1, sample: 25, build: true, port: 3000, image: "campfire-bench"
|
||||
}
|
||||
|
||||
OptionParser.new do |parser|
|
||||
parser.on("--label LABEL", "Run label (required), e.g. redis-2cpu-2g") { |v| options[:label] = v }
|
||||
parser.on("--users N", Integer, "Ramp target (default 3000, max 10000)") { |v| options[:users] = v }
|
||||
parser.on("--ramp SECONDS", Integer, "Ramp duration (default 480)") { |v| options[:ramp] = v }
|
||||
parser.on("--hold SECONDS", Integer, "Hold duration at target (default 120)") { |v| options[:hold] = v }
|
||||
parser.on("--cpus N", Integer, "App container CPU count via cpuset (default 2)") { |v| options[:cpus] = v }
|
||||
parser.on("--memory SIZE", "App container memory limit (default 2g)") { |v| options[:memory] = v }
|
||||
parser.on("--send-interval S", Float, "Seconds between messages (default 1)") { |v| options[:send_interval] = v }
|
||||
parser.on("--sample N", Integer, "1-in-N receivers log latency lines (default 25)") { |v| options[:sample] = v }
|
||||
parser.on("--port PORT", Integer, "Host port for the app (default 3000)") { |v| options[:port] = v }
|
||||
parser.on("--image TAG", "Image tag (default campfire-bench)") { |v| options[:image] = v }
|
||||
parser.on("--no-build", "Skip docker build") { options[:build] = false }
|
||||
end.parse!
|
||||
|
||||
abort "--label is required" unless options[:label]
|
||||
|
||||
class Bench
|
||||
APP_CONTAINER = "campfire-bench-app"
|
||||
K6_CONTAINER = "campfire-bench-k6"
|
||||
|
||||
attr_reader :opts, :results_dir
|
||||
|
||||
def initialize(opts)
|
||||
@opts = opts
|
||||
@root = File.expand_path("..", __dir__)
|
||||
@perf_dir = File.join(@root, "test/performance")
|
||||
@results_dir = File.join(@root, "tmp/bench", opts[:label])
|
||||
end
|
||||
|
||||
def run
|
||||
FileUtils.mkdir_p(results_dir)
|
||||
build if opts[:build]
|
||||
write_meta
|
||||
start_app
|
||||
wait_for_app
|
||||
sampler = start_stats_sampler
|
||||
run_k6
|
||||
ensure
|
||||
stop(sampler)
|
||||
capture_app_state
|
||||
cleanup
|
||||
end
|
||||
|
||||
private
|
||||
def build
|
||||
sh "docker", "build", "-t", opts[:image], @root
|
||||
end
|
||||
|
||||
def write_meta
|
||||
meta = opts.merge(
|
||||
git_revision: `git rev-parse HEAD`.strip,
|
||||
git_branch: `git rev-parse --abbrev-ref HEAD`.strip,
|
||||
host_cpus: Etc.nprocessors,
|
||||
started_at: Time.now.utc.iso8601
|
||||
)
|
||||
File.write(File.join(results_dir, "meta.json"), JSON.pretty_generate(meta))
|
||||
end
|
||||
|
||||
def start_app
|
||||
system "docker", "rm", "-f", APP_CONTAINER, err: File::NULL, out: File::NULL
|
||||
sh "docker", "run", "-d", "--name", APP_CONTAINER,
|
||||
"--cpuset-cpus", app_cpuset,
|
||||
"--memory", opts[:memory], "--memory-swap", opts[:memory],
|
||||
"-p", "#{opts[:port]}:80",
|
||||
"-e", "SECRET_KEY_BASE=dummy",
|
||||
"-e", "RAILS_ENV=performance",
|
||||
opts[:image]
|
||||
end
|
||||
|
||||
def app_cpuset
|
||||
(0...opts[:cpus]).to_a.join(",")
|
||||
end
|
||||
|
||||
def wait_for_app
|
||||
timeout_at = Time.now + 300
|
||||
uri = URI("http://127.0.0.1:#{opts[:port]}/")
|
||||
loop do
|
||||
raise "App not up after 300s" if Time.now > timeout_at
|
||||
begin
|
||||
break if Net::HTTP.get_response(uri).code.to_i < 400
|
||||
rescue Errno::ECONNREFUSED, Errno::ECONNRESET, EOFError
|
||||
end
|
||||
puts "Waiting for app (seeding 10k users on first boot)..."
|
||||
sleep 2
|
||||
end
|
||||
puts "App is up."
|
||||
end
|
||||
|
||||
def start_stats_sampler
|
||||
stats_path = File.join(results_dir, "stats.csv")
|
||||
fork do
|
||||
File.open(stats_path, "w") do |file|
|
||||
file.puts "epoch_ms,cpu_perc,mem_bytes"
|
||||
file.sync = true
|
||||
loop do
|
||||
line = `docker stats --no-stream --format "{{.CPUPerc}} {{.MemUsage}}" #{APP_CONTAINER} 2>/dev/null`.strip
|
||||
if line =~ /([\d.]+)%\s+([\d.]+)(\w+)/
|
||||
file.puts "#{(Time.now.to_f * 1000).to_i},#{$1},#{to_bytes($2.to_f, $3)}"
|
||||
end
|
||||
sleep 1
|
||||
end
|
||||
end
|
||||
end
|
||||
end
|
||||
|
||||
def to_bytes(value, unit)
|
||||
factors = { "B" => 1, "KiB" => 1024, "MiB" => 1024**2, "GiB" => 1024**3,
|
||||
"kB" => 1000, "MB" => 1000**2, "GB" => 1000**3 }
|
||||
(value * factors.fetch(unit, 1)).to_i
|
||||
end
|
||||
|
||||
def run_k6
|
||||
total = opts[:ramp] + opts[:hold]
|
||||
puts "Running k6: ramp to #{opts[:users]} users over #{opts[:ramp]}s, hold #{opts[:hold]}s (total #{total}s)..."
|
||||
system "docker", "rm", "-f", K6_CONTAINER, err: File::NULL, out: File::NULL
|
||||
k6_log = File.join(results_dir, "k6.log")
|
||||
summary = File.join(results_dir, "summary.json")
|
||||
finished = system(
|
||||
"docker", "run", "--rm", "--name", K6_CONTAINER,
|
||||
"--network", "host",
|
||||
"--cpuset-cpus", k6_cpuset,
|
||||
"--ulimit", "nofile=262144:262144",
|
||||
"-u", "#{Process.uid}:#{Process.gid}",
|
||||
"-v", "#{@perf_dir}:/src",
|
||||
"-v", "#{results_dir}:/results",
|
||||
"-e", "HOST=127.0.0.1", "-e", "PORT=#{opts[:port]}",
|
||||
"-e", "USERS=#{opts[:users]}",
|
||||
"-e", "RAMP_S=#{opts[:ramp]}", "-e", "HOLD_S=#{opts[:hold]}",
|
||||
"-e", "SEND_INTERVAL_S=#{opts[:send_interval]}",
|
||||
"-e", "SAMPLE=#{opts[:sample]}",
|
||||
"grafana/k6", "run",
|
||||
"--summary-export", "/results/summary.json",
|
||||
"--quiet",
|
||||
"/src/ramp.js",
|
||||
out: k6_log, err: [ k6_log, "a" ]
|
||||
)
|
||||
if finished
|
||||
puts "k6 finished."
|
||||
else
|
||||
puts "k6 exited non-zero — results still captured in #{k6_log}."
|
||||
end
|
||||
end
|
||||
|
||||
def k6_cpuset
|
||||
(opts[:cpus]...Etc.nprocessors).to_a.join(",")
|
||||
end
|
||||
|
||||
def stop(sampler)
|
||||
Process.kill("TERM", sampler)
|
||||
Process.wait(sampler)
|
||||
rescue Errno::ESRCH, Errno::ECHILD
|
||||
nil
|
||||
end
|
||||
|
||||
def capture_app_state
|
||||
oom = `docker inspect --format '{{.State.OOMKilled}}' #{APP_CONTAINER} 2>/dev/null`.strip
|
||||
File.write(File.join(results_dir, "oom.txt"), oom)
|
||||
system "docker logs --tail 200 #{APP_CONTAINER} > #{File.join(results_dir, 'app.log')} 2>&1"
|
||||
puts "App container was OOM-killed during the run." if oom == "true"
|
||||
end
|
||||
|
||||
def cleanup
|
||||
system "docker", "rm", "-f", APP_CONTAINER, err: File::NULL, out: File::NULL
|
||||
end
|
||||
|
||||
def sh(*cmd)
|
||||
puts "+ #{cmd.join(' ')}"
|
||||
system(*cmd) || abort("Command failed: #{cmd.join(' ')}")
|
||||
end
|
||||
end
|
||||
|
||||
Bench.new(options).run
|
||||
puts "Results in tmp/bench/#{options[:label]}/"
|
||||
Executable
+160
@@ -0,0 +1,160 @@
|
||||
#!/usr/bin/env ruby
|
||||
require "json"
|
||||
require "csv"
|
||||
|
||||
BIN_MS = 10_000
|
||||
SLO_P95_MS = 500
|
||||
|
||||
class Run
|
||||
attr_reader :label, :dir
|
||||
|
||||
def initialize(label)
|
||||
@label = label
|
||||
@dir = File.expand_path("../tmp/bench/#{label}", __dir__)
|
||||
abort "No results in tmp/bench/#{label}" unless File.exist?(File.join(dir, "k6.log"))
|
||||
end
|
||||
|
||||
def report
|
||||
write_series
|
||||
print_summary
|
||||
end
|
||||
|
||||
def series
|
||||
@series ||= build_series
|
||||
end
|
||||
|
||||
def meta
|
||||
@meta ||= JSON.parse(File.read(File.join(dir, "meta.json")))
|
||||
end
|
||||
|
||||
def summary
|
||||
@summary ||= JSON.parse(File.read(File.join(dir, "summary.json")))
|
||||
rescue Errno::ENOENT
|
||||
{}
|
||||
end
|
||||
|
||||
def capacity
|
||||
breach = first_sustained_breach
|
||||
if breach
|
||||
breach[:connected]
|
||||
else
|
||||
nil
|
||||
end
|
||||
end
|
||||
|
||||
private
|
||||
def build_series
|
||||
events = parse_k6_log
|
||||
stats = parse_stats
|
||||
first_ts = events.map { |e| e[:ts] }.min
|
||||
return [] unless first_ts
|
||||
|
||||
bins = Hash.new { |h, k| h[k] = { lats: [], conns: 0, sents: 0, errs: 0, cpu: [], mem: [] } }
|
||||
events.each do |event|
|
||||
bin = (event[:ts] - first_ts) / BIN_MS
|
||||
case event[:kind]
|
||||
when "CONN" then bins[bin][:conns] += 1
|
||||
when "LAT" then bins[bin][:lats] << event[:value]
|
||||
when "SENT" then bins[bin][:sents] += 1
|
||||
when "ERR" then bins[bin][:errs] += 1
|
||||
end
|
||||
end
|
||||
stats.each do |stat|
|
||||
bin = (stat[:ts] - first_ts) / BIN_MS
|
||||
next if bin < 0
|
||||
bins[bin][:cpu] << stat[:cpu]
|
||||
bins[bin][:mem] << stat[:mem]
|
||||
end
|
||||
|
||||
connected = 0
|
||||
(0..bins.keys.max).map do |bin|
|
||||
data = bins[bin]
|
||||
connected = [ connected + data[:conns], meta["users"] ].min
|
||||
{
|
||||
elapsed_s: bin * BIN_MS / 1000,
|
||||
connected: connected,
|
||||
sent: data[:sents],
|
||||
samples: data[:lats].size,
|
||||
lat_p50: percentile(data[:lats], 50),
|
||||
lat_p95: percentile(data[:lats], 95),
|
||||
lat_max: data[:lats].max,
|
||||
errors: data[:errs],
|
||||
cpu_perc: average(data[:cpu]),
|
||||
mem_mb: average(data[:mem])&.then { |bytes| (bytes / 1024.0 / 1024.0).round(1) }
|
||||
}
|
||||
end
|
||||
end
|
||||
|
||||
def parse_k6_log
|
||||
events = []
|
||||
File.foreach(File.join(dir, "k6.log")) do |line|
|
||||
if line =~ /BENCH (CONN|LAT|SENT|ERR) (\d+)(?: (\S+))?/
|
||||
events << { kind: $1, ts: $2.to_i, value: $3.to_i }
|
||||
end
|
||||
end
|
||||
events
|
||||
end
|
||||
|
||||
def parse_stats
|
||||
path = File.join(dir, "stats.csv")
|
||||
return [] unless File.exist?(path)
|
||||
CSV.read(path, headers: true).map do |row|
|
||||
{ ts: row["epoch_ms"].to_i, cpu: row["cpu_perc"].to_f, mem: row["mem_bytes"].to_i }
|
||||
end
|
||||
end
|
||||
|
||||
def percentile(values, pct)
|
||||
return nil if values.empty?
|
||||
sorted = values.sort
|
||||
sorted[((pct / 100.0) * (sorted.size - 1)).round]
|
||||
end
|
||||
|
||||
def average(values)
|
||||
return nil if values.empty?
|
||||
(values.sum / values.size.to_f).round(1)
|
||||
end
|
||||
|
||||
def write_series
|
||||
CSV.open(File.join(dir, "series.csv"), "w") do |csv|
|
||||
csv << series.first.keys
|
||||
series.each { |row| csv << row.values }
|
||||
end
|
||||
end
|
||||
|
||||
def first_sustained_breach
|
||||
series.each_cons(2) do |a, b|
|
||||
if breached?(a) && breached?(b)
|
||||
return a
|
||||
end
|
||||
end
|
||||
nil
|
||||
end
|
||||
|
||||
def breached?(bin)
|
||||
if bin[:samples] > 0 && bin[:lat_p95] && bin[:lat_p95] > SLO_P95_MS
|
||||
true
|
||||
else
|
||||
bin[:errors] > 0
|
||||
end
|
||||
end
|
||||
|
||||
def print_summary
|
||||
latency = summary.dig("metrics", "delivery_latency") || {}
|
||||
puts
|
||||
puts "== #{label} (#{meta['cpus']} CPU, #{meta['memory']} RAM, ramp to #{meta['users']}) =="
|
||||
puts " delivery latency: p50=#{latency['med']&.round(1)}ms p95=#{latency['p(95)']&.round(1)}ms max=#{latency['max']&.round(1)}ms"
|
||||
puts " messages received: #{summary.dig('metrics', 'bench_messages_received', 'count')}"
|
||||
puts " socket errors: #{summary.dig('metrics', 'bench_socket_errors', 'count') || 0}"
|
||||
peak = series.map { |bin| bin[:mem_mb] }.compact.max
|
||||
puts " peak container memory: #{peak} MB"
|
||||
if capacity
|
||||
puts " capacity at SLO (p95 <= #{SLO_P95_MS}ms, no errors): ~#{capacity} concurrent users"
|
||||
else
|
||||
puts " capacity at SLO: not reached (>= #{series.last[:connected]} concurrent users)"
|
||||
end
|
||||
puts " series: tmp/bench/#{label}/series.csv"
|
||||
end
|
||||
end
|
||||
|
||||
abort "Usage: bin/bench-report LABEL [LABEL...]" if ARGV.empty?
|
||||
ARGV.each { |label| Run.new(label).report }
|
||||
@@ -0,0 +1,6 @@
|
||||
#!/usr/bin/env ruby
|
||||
|
||||
require_relative "../config/environment"
|
||||
require "solid_queue/cli"
|
||||
|
||||
SolidQueue::Cli.start(ARGV)
|
||||
@@ -8,9 +8,6 @@ app_root="$(
|
||||
)"
|
||||
export PATH="$app_root/bin:$PATH"
|
||||
|
||||
REDIS_PORT=6379
|
||||
REDIS_HOST=localhost
|
||||
|
||||
if [ "$RAILS_ENV" = "production" ]; then
|
||||
echo "RAILS_ENV is production; bailing out"
|
||||
exit 1
|
||||
@@ -45,10 +42,6 @@ step() {
|
||||
return $exit_code
|
||||
}
|
||||
|
||||
redis_running() {
|
||||
nc -z "$REDIS_HOST" "$REDIS_PORT" 2>/dev/null
|
||||
}
|
||||
|
||||
echo
|
||||
gum style --foreground 214 " ) "
|
||||
gum style --foreground 208 " ) \\ campfire"
|
||||
@@ -88,21 +81,6 @@ if [[ $* == *--reset* ]]; then
|
||||
fi
|
||||
step "Preparing the database" rails db:prepare
|
||||
|
||||
# Start Redis if not running
|
||||
if ! redis_running; then
|
||||
if command -v docker &>/dev/null; then
|
||||
if docker ps -aq -f name=campfire-redis | grep -q .; then
|
||||
step "Starting Redis" docker start campfire-redis
|
||||
else
|
||||
step "Setting up Redis" docker run -d --name campfire-redis -p "$REDIS_PORT:$REDIS_PORT" redis:7
|
||||
fi
|
||||
else
|
||||
echo "Couldn't start Redis"
|
||||
echo "Install either docker or redis and then run this command again"
|
||||
exit 1
|
||||
fi
|
||||
fi
|
||||
|
||||
# Install GitHub Actions linting tools
|
||||
for tool in actionlint shellcheck zizmor; do
|
||||
if ! command -v "$tool" &> /dev/null; then
|
||||
|
||||
+17
-10
@@ -1,18 +1,25 @@
|
||||
default: &default
|
||||
adapter: redis
|
||||
url: redis://localhost:6379
|
||||
|
||||
# Async adapter only works within the same process, so for manually triggering cable updates from a console,
|
||||
# and seeing results in the browser, you must do so from the web console (running inside the dev process),
|
||||
# not a terminal started via bin/rails console! Add "console" to any action or any ERB template view
|
||||
# to make the web console appear.
|
||||
development:
|
||||
<<: *default
|
||||
channel_prefix: campfire_development
|
||||
adapter: async
|
||||
|
||||
test:
|
||||
adapter: test
|
||||
|
||||
performance:
|
||||
<<: *default
|
||||
channel_prefix: campfire_performance
|
||||
adapter: solid_cable
|
||||
connects_to:
|
||||
database:
|
||||
writing: cable
|
||||
polling_interval: 0.05.seconds
|
||||
message_retention: 1.day
|
||||
|
||||
production:
|
||||
<<: *default
|
||||
channel_prefix: campfire_production
|
||||
adapter: solid_cable
|
||||
connects_to:
|
||||
database:
|
||||
writing: cable
|
||||
polling_interval: 0.05.seconds
|
||||
message_retention: 1.day
|
||||
|
||||
@@ -0,0 +1,21 @@
|
||||
default: &default
|
||||
store_options:
|
||||
# Cap age of oldest cache entry to fulfill retention policies
|
||||
# max_age: <%= 60.days.to_i %>
|
||||
max_size: <%= 256.megabytes %>
|
||||
namespace: <%= Rails.env %>
|
||||
|
||||
development:
|
||||
database: cache
|
||||
<<: *default
|
||||
|
||||
test:
|
||||
<<: *default
|
||||
|
||||
performance:
|
||||
database: cache
|
||||
<<: *default
|
||||
|
||||
production:
|
||||
database: cache
|
||||
<<: *default
|
||||
@@ -14,6 +14,10 @@ development:
|
||||
primary:
|
||||
<<: *default
|
||||
database: storage/db/development.sqlite3
|
||||
cache:
|
||||
<<: *default
|
||||
database: storage/db/development-cache.sqlite3
|
||||
migrations_paths: db/cache_migrate
|
||||
|
||||
# Warning: The database defined as "test" will be erased and
|
||||
# re-generated from your development database when you run "rake".
|
||||
@@ -27,8 +31,32 @@ performance:
|
||||
primary:
|
||||
<<: *default
|
||||
database: storage/db/performance.sqlite3
|
||||
queue:
|
||||
<<: *default
|
||||
database: storage/db/performance-queue.sqlite3
|
||||
migrations_paths: db/queue_migrate
|
||||
cache:
|
||||
<<: *default
|
||||
database: storage/db/performance-cache.sqlite3
|
||||
migrations_paths: db/cache_migrate
|
||||
cable:
|
||||
<<: *default
|
||||
database: storage/db/performance-cable.sqlite3
|
||||
migrations_paths: db/cable_migrate
|
||||
|
||||
production:
|
||||
primary:
|
||||
<<: *default
|
||||
database: storage/db/production.sqlite3
|
||||
queue:
|
||||
<<: *default
|
||||
database: storage/db/production-queue.sqlite3
|
||||
migrations_paths: db/queue_migrate
|
||||
cache:
|
||||
<<: *default
|
||||
database: storage/db/production-cache.sqlite3
|
||||
migrations_paths: db/cache_migrate
|
||||
cable:
|
||||
<<: *default
|
||||
database: storage/db/production-cable.sqlite3
|
||||
migrations_paths: db/cable_migrate
|
||||
|
||||
@@ -21,7 +21,7 @@ Rails.application.configure do
|
||||
config.action_controller.perform_caching = true
|
||||
config.action_controller.enable_fragment_cache_logging = true
|
||||
|
||||
config.cache_store = :redis_cache_store
|
||||
config.cache_store = :solid_cache_store
|
||||
config.public_file_server.headers = {
|
||||
"cache-control" => "public, max-age=#{2.days.to_i}"
|
||||
}
|
||||
@@ -65,7 +65,7 @@ Rails.application.configure do
|
||||
# config.action_cable.disable_request_forgery_protection = true
|
||||
|
||||
# Uncomment to test with production class job queue
|
||||
# config.active_job.queue_adapter = :resque
|
||||
# config.active_job.queue_adapter = :solid_queue
|
||||
|
||||
# Highlight code that enqueued background job in logs.
|
||||
config.active_job.verbose_enqueue_logs = true
|
||||
|
||||
@@ -68,7 +68,7 @@ Rails.application.configure do
|
||||
config.active_support.report_deprecations = false
|
||||
|
||||
# Cache in memory for now
|
||||
config.cache_store = :redis_cache_store
|
||||
config.cache_store = :solid_cache_store
|
||||
|
||||
# Assets are cacheable
|
||||
config.public_file_server.headers = {
|
||||
@@ -92,5 +92,6 @@ Rails.application.configure do
|
||||
# Only use :id for inspections in production.
|
||||
config.active_record.attributes_for_inspect = [ :id ]
|
||||
|
||||
config.active_job.queue_adapter = :resque
|
||||
config.active_job.queue_adapter = :solid_queue
|
||||
config.solid_queue.connects_to = { database: { writing: :queue } }
|
||||
end
|
||||
|
||||
@@ -35,8 +35,6 @@ pidfile ENV.fetch("PIDFILE") { "tmp/pids/server.pid" }
|
||||
worker_count = (Concurrent.processor_count * 0.666).ceil
|
||||
workers ENV.fetch("WEB_CONCURRENCY") { worker_count }
|
||||
|
||||
ENV["JOB_CONCURRENCY"] ||= worker_count.to_s
|
||||
|
||||
# Use the `preload_app!` method when specifying a `workers` number.
|
||||
# This directive tells Puma to first boot the application and load code
|
||||
# before forking the application. This takes advantage of Copy On Write
|
||||
|
||||
@@ -0,0 +1,21 @@
|
||||
default: &default
|
||||
dispatchers:
|
||||
- polling_interval: 1
|
||||
batch_size: 500
|
||||
workers:
|
||||
- queues: "*"
|
||||
threads: 3
|
||||
processes: <%= ENV.fetch("JOB_CONCURRENCY") { (Concurrent.processor_count * 0.5).ceil } %>
|
||||
polling_interval: 0.1
|
||||
|
||||
development:
|
||||
<<: *default
|
||||
|
||||
test:
|
||||
<<: *default
|
||||
|
||||
performance:
|
||||
<<: *default
|
||||
|
||||
production:
|
||||
<<: *default
|
||||
@@ -0,0 +1,15 @@
|
||||
# examples:
|
||||
# periodic_cleanup:
|
||||
# class: CleanSoftDeletedRecordsJob
|
||||
# queue: background
|
||||
# args: [ 1000, { batch_size: 500 } ]
|
||||
# schedule: every hour
|
||||
# periodic_cleanup_with_command:
|
||||
# command: "SoftDeletedRecord.due.delete_all"
|
||||
# priority: 2
|
||||
# schedule: at 5am every day
|
||||
|
||||
production:
|
||||
clear_solid_queue_finished_jobs:
|
||||
command: "SolidQueue::Job.clear_finished_in_batches(sleep_between_batches: 0.3)"
|
||||
schedule: every hour at minute 12
|
||||
@@ -1,4 +0,0 @@
|
||||
port 6379
|
||||
daemonize no
|
||||
appendonly no
|
||||
save ""
|
||||
@@ -1 +0,0 @@
|
||||
default: <%= (Concurrent.processor_count * 0.5).ceil %>
|
||||
@@ -0,0 +1,11 @@
|
||||
ActiveRecord::Schema[7.1].define(version: 1) do
|
||||
create_table "solid_cable_messages", force: :cascade do |t|
|
||||
t.binary "channel", limit: 1024, null: false
|
||||
t.binary "payload", limit: 536870912, null: false
|
||||
t.datetime "created_at", null: false
|
||||
t.integer "channel_hash", limit: 8, null: false
|
||||
t.index ["channel"], name: "index_solid_cable_messages_on_channel"
|
||||
t.index ["channel_hash"], name: "index_solid_cable_messages_on_channel_hash"
|
||||
t.index ["created_at"], name: "index_solid_cable_messages_on_created_at"
|
||||
end
|
||||
end
|
||||
@@ -0,0 +1,12 @@
|
||||
ActiveRecord::Schema[7.2].define(version: 1) do
|
||||
create_table "solid_cache_entries", force: :cascade do |t|
|
||||
t.binary "key", limit: 1024, null: false
|
||||
t.binary "value", limit: 536870912, null: false
|
||||
t.datetime "created_at", null: false
|
||||
t.integer "key_hash", limit: 8, null: false
|
||||
t.integer "byte_size", limit: 4, null: false
|
||||
t.index ["byte_size"], name: "index_solid_cache_entries_on_byte_size"
|
||||
t.index ["key_hash", "byte_size"], name: "index_solid_cache_entries_on_key_hash_and_byte_size"
|
||||
t.index ["key_hash"], name: "index_solid_cache_entries_on_key_hash", unique: true
|
||||
end
|
||||
end
|
||||
@@ -0,0 +1,129 @@
|
||||
ActiveRecord::Schema[7.1].define(version: 1) do
|
||||
create_table "solid_queue_blocked_executions", force: :cascade do |t|
|
||||
t.bigint "job_id", null: false
|
||||
t.string "queue_name", null: false
|
||||
t.integer "priority", default: 0, null: false
|
||||
t.string "concurrency_key", null: false
|
||||
t.datetime "expires_at", null: false
|
||||
t.datetime "created_at", null: false
|
||||
t.index [ "concurrency_key", "priority", "job_id" ], name: "index_solid_queue_blocked_executions_for_release"
|
||||
t.index [ "expires_at", "concurrency_key" ], name: "index_solid_queue_blocked_executions_for_maintenance"
|
||||
t.index [ "job_id" ], name: "index_solid_queue_blocked_executions_on_job_id", unique: true
|
||||
end
|
||||
|
||||
create_table "solid_queue_claimed_executions", force: :cascade do |t|
|
||||
t.bigint "job_id", null: false
|
||||
t.bigint "process_id"
|
||||
t.datetime "created_at", null: false
|
||||
t.index [ "job_id" ], name: "index_solid_queue_claimed_executions_on_job_id", unique: true
|
||||
t.index [ "process_id", "job_id" ], name: "index_solid_queue_claimed_executions_on_process_id_and_job_id"
|
||||
end
|
||||
|
||||
create_table "solid_queue_failed_executions", force: :cascade do |t|
|
||||
t.bigint "job_id", null: false
|
||||
t.text "error"
|
||||
t.datetime "created_at", null: false
|
||||
t.index [ "job_id" ], name: "index_solid_queue_failed_executions_on_job_id", unique: true
|
||||
end
|
||||
|
||||
create_table "solid_queue_jobs", force: :cascade do |t|
|
||||
t.string "queue_name", null: false
|
||||
t.string "class_name", null: false
|
||||
t.text "arguments"
|
||||
t.integer "priority", default: 0, null: false
|
||||
t.string "active_job_id"
|
||||
t.datetime "scheduled_at"
|
||||
t.datetime "finished_at"
|
||||
t.string "concurrency_key"
|
||||
t.datetime "created_at", null: false
|
||||
t.datetime "updated_at", null: false
|
||||
t.index [ "active_job_id" ], name: "index_solid_queue_jobs_on_active_job_id"
|
||||
t.index [ "class_name" ], name: "index_solid_queue_jobs_on_class_name"
|
||||
t.index [ "finished_at" ], name: "index_solid_queue_jobs_on_finished_at"
|
||||
t.index [ "queue_name", "finished_at" ], name: "index_solid_queue_jobs_for_filtering"
|
||||
t.index [ "scheduled_at", "finished_at" ], name: "index_solid_queue_jobs_for_alerting"
|
||||
end
|
||||
|
||||
create_table "solid_queue_pauses", force: :cascade do |t|
|
||||
t.string "queue_name", null: false
|
||||
t.datetime "created_at", null: false
|
||||
t.index [ "queue_name" ], name: "index_solid_queue_pauses_on_queue_name", unique: true
|
||||
end
|
||||
|
||||
create_table "solid_queue_processes", force: :cascade do |t|
|
||||
t.string "kind", null: false
|
||||
t.datetime "last_heartbeat_at", null: false
|
||||
t.bigint "supervisor_id"
|
||||
t.integer "pid", null: false
|
||||
t.string "hostname"
|
||||
t.text "metadata"
|
||||
t.datetime "created_at", null: false
|
||||
t.string "name", null: false
|
||||
t.index [ "last_heartbeat_at" ], name: "index_solid_queue_processes_on_last_heartbeat_at"
|
||||
t.index [ "name", "supervisor_id" ], name: "index_solid_queue_processes_on_name_and_supervisor_id", unique: true
|
||||
t.index [ "supervisor_id" ], name: "index_solid_queue_processes_on_supervisor_id"
|
||||
end
|
||||
|
||||
create_table "solid_queue_ready_executions", force: :cascade do |t|
|
||||
t.bigint "job_id", null: false
|
||||
t.string "queue_name", null: false
|
||||
t.integer "priority", default: 0, null: false
|
||||
t.datetime "created_at", null: false
|
||||
t.index [ "job_id" ], name: "index_solid_queue_ready_executions_on_job_id", unique: true
|
||||
t.index [ "priority", "job_id" ], name: "index_solid_queue_poll_all"
|
||||
t.index [ "queue_name", "priority", "job_id" ], name: "index_solid_queue_poll_by_queue"
|
||||
end
|
||||
|
||||
create_table "solid_queue_recurring_executions", force: :cascade do |t|
|
||||
t.bigint "job_id", null: false
|
||||
t.string "task_key", null: false
|
||||
t.datetime "run_at", null: false
|
||||
t.datetime "created_at", null: false
|
||||
t.index [ "job_id" ], name: "index_solid_queue_recurring_executions_on_job_id", unique: true
|
||||
t.index [ "task_key", "run_at" ], name: "index_solid_queue_recurring_executions_on_task_key_and_run_at", unique: true
|
||||
end
|
||||
|
||||
create_table "solid_queue_recurring_tasks", force: :cascade do |t|
|
||||
t.string "key", null: false
|
||||
t.string "schedule", null: false
|
||||
t.string "command", limit: 2048
|
||||
t.string "class_name"
|
||||
t.text "arguments"
|
||||
t.string "queue_name"
|
||||
t.integer "priority", default: 0
|
||||
t.boolean "static", default: true, null: false
|
||||
t.text "description"
|
||||
t.datetime "created_at", null: false
|
||||
t.datetime "updated_at", null: false
|
||||
t.index [ "key" ], name: "index_solid_queue_recurring_tasks_on_key", unique: true
|
||||
t.index [ "static" ], name: "index_solid_queue_recurring_tasks_on_static"
|
||||
end
|
||||
|
||||
create_table "solid_queue_scheduled_executions", force: :cascade do |t|
|
||||
t.bigint "job_id", null: false
|
||||
t.string "queue_name", null: false
|
||||
t.integer "priority", default: 0, null: false
|
||||
t.datetime "scheduled_at", null: false
|
||||
t.datetime "created_at", null: false
|
||||
t.index [ "job_id" ], name: "index_solid_queue_scheduled_executions_on_job_id", unique: true
|
||||
t.index [ "scheduled_at", "priority", "job_id" ], name: "index_solid_queue_dispatch_all"
|
||||
end
|
||||
|
||||
create_table "solid_queue_semaphores", force: :cascade do |t|
|
||||
t.string "key", null: false
|
||||
t.integer "value", default: 1, null: false
|
||||
t.datetime "expires_at", null: false
|
||||
t.datetime "created_at", null: false
|
||||
t.datetime "updated_at", null: false
|
||||
t.index [ "expires_at" ], name: "index_solid_queue_semaphores_on_expires_at"
|
||||
t.index [ "key", "value" ], name: "index_solid_queue_semaphores_on_key_and_value"
|
||||
t.index [ "key" ], name: "index_solid_queue_semaphores_on_key", unique: true
|
||||
end
|
||||
|
||||
add_foreign_key "solid_queue_blocked_executions", "solid_queue_jobs", column: "job_id", on_delete: :cascade
|
||||
add_foreign_key "solid_queue_claimed_executions", "solid_queue_jobs", column: "job_id", on_delete: :cascade
|
||||
add_foreign_key "solid_queue_failed_executions", "solid_queue_jobs", column: "job_id", on_delete: :cascade
|
||||
add_foreign_key "solid_queue_ready_executions", "solid_queue_jobs", column: "job_id", on_delete: :cascade
|
||||
add_foreign_key "solid_queue_recurring_executions", "solid_queue_jobs", column: "job_id", on_delete: :cascade
|
||||
add_foreign_key "solid_queue_scheduled_executions", "solid_queue_jobs", column: "job_id", on_delete: :cascade
|
||||
end
|
||||
@@ -1,12 +0,0 @@
|
||||
task "resque:setup" do
|
||||
require_relative "../../config/environment"
|
||||
end
|
||||
|
||||
task "resque:pool:setup" do
|
||||
ActiveRecord::Base.connection.disconnect!
|
||||
|
||||
Resque::Pool.after_prefork do |job|
|
||||
ActiveRecord::Base.establish_connection
|
||||
Resque.redis.client.close
|
||||
end
|
||||
end
|
||||
@@ -0,0 +1,158 @@
|
||||
import ws from 'k6/ws';
|
||||
import http from 'k6/http';
|
||||
import { sleep } from 'k6';
|
||||
import { Trend, Counter } from 'k6/metrics';
|
||||
import papaparse from 'https://jslib.k6.io/papaparse/5.1.1/index.js';
|
||||
import { SharedArray } from 'k6/data';
|
||||
|
||||
// Stream names signed with the performance environment's SECRET_KEY_BASE=dummy:
|
||||
// the rooms list, room 1 messages, and user 1 rooms.
|
||||
const turboSignedStreamNames = [
|
||||
"InJvb21zIg==--54acd827f0a7db144c75316a9fc488c0a949f9635b1e47956ce1bd9d1cf2c41d",
|
||||
"IloybGtPaTh2WTJGdGNHWnBjbVV2VW05dmJYTTZPa05zYjNObFpDOHg6bWVzc2FnZXMi--84f0f3dde5d23eb0fdb410746c2fb76813a4ddff1e2798aac4be0c3d969702ba",
|
||||
"IloybGtPaTh2WTJGdGNHWnBjbVV2VlhObGNpOHg6cm9vbXMi--df547a679cd41f7531b53d9e48f9883c02481a4da0d862453106441d8546d084"
|
||||
];
|
||||
|
||||
const dummyCookies = new SharedArray('cookies', function () {
|
||||
return papaparse.parse(open('cookies.txt'), { header: false }).data;
|
||||
});
|
||||
|
||||
const host = __ENV.HOST || "127.0.0.1";
|
||||
let port = ":3000";
|
||||
if (__ENV.PORT) {
|
||||
port = `:${__ENV.PORT}`;
|
||||
}
|
||||
|
||||
const users = parseInt(__ENV.USERS || "3000");
|
||||
const rampS = parseInt(__ENV.RAMP_S || "480");
|
||||
const holdS = parseInt(__ENV.HOLD_S || "120");
|
||||
const sendIntervalS = parseFloat(__ENV.SEND_INTERVAL_S || "1");
|
||||
const sample = parseInt(__ENV.SAMPLE || "25");
|
||||
const totalS = rampS + holdS;
|
||||
|
||||
const deliveryLatency = new Trend('delivery_latency', true);
|
||||
const messagesReceived = new Counter('bench_messages_received');
|
||||
const socketErrors = new Counter('bench_socket_errors');
|
||||
|
||||
export const options = {
|
||||
discardResponseBodies: true,
|
||||
scenarios: {
|
||||
receivers: {
|
||||
executor: 'ramping-vus',
|
||||
exec: 'receiver',
|
||||
startVUs: 0,
|
||||
stages: [
|
||||
{ duration: `${rampS}s`, target: users },
|
||||
{ duration: `${holdS}s`, target: users },
|
||||
],
|
||||
gracefulRampDown: '0s',
|
||||
gracefulStop: '5s',
|
||||
},
|
||||
sender: {
|
||||
executor: 'constant-vus',
|
||||
exec: 'sender',
|
||||
vus: 1,
|
||||
duration: `${totalS}s`,
|
||||
gracefulStop: '10s',
|
||||
},
|
||||
},
|
||||
};
|
||||
|
||||
export function receiver() {
|
||||
const cookie = dummyCookies[(__VU - 1) % dummyCookies.length][0];
|
||||
const url = `ws://${host}${port}/cable`;
|
||||
const params = {
|
||||
headers: { 'Origin': 'http://localhost', 'Cookie': `session_token=${cookie}` }
|
||||
};
|
||||
const subscriptions = [
|
||||
'{"channel":"PresenceChannel", "room_id":1}',
|
||||
'{"channel":"UnreadRoomsChannel"}',
|
||||
'{"channel":"HeartbeatChannel"}',
|
||||
...turboSignedStreamNames.map((name) =>
|
||||
`{"channel":"Turbo::StreamsChannel", "signed_stream_name":"${name}"}`)
|
||||
];
|
||||
|
||||
const res = ws.connect(url, params, function (socket) {
|
||||
let confirmed = 0;
|
||||
|
||||
socket.on('open', function () {
|
||||
subscriptions.forEach((identifier) => {
|
||||
socket.send(JSON.stringify({ command: 'subscribe', identifier: identifier }));
|
||||
});
|
||||
});
|
||||
|
||||
socket.on('message', function (message) {
|
||||
if (message.includes('confirm_subscription')) {
|
||||
confirmed++;
|
||||
if (confirmed === subscriptions.length) {
|
||||
console.log(`BENCH CONN ${Date.now()} ${__VU}`);
|
||||
}
|
||||
} else if (message.includes('reject_subscription')) {
|
||||
socketErrors.add(1);
|
||||
console.log(`BENCH ERR ${Date.now()} reject_subscription`);
|
||||
} else if (message.includes('bench:')) {
|
||||
const match = message.match(/bench:(\d+):(\d+)/);
|
||||
if (match) {
|
||||
const latency = Date.now() - parseInt(match[1]);
|
||||
deliveryLatency.add(latency);
|
||||
messagesReceived.add(1);
|
||||
if (__VU % sample === 0) {
|
||||
console.log(`BENCH LAT ${Date.now()} ${latency}`);
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
socket.on('error', function (e) {
|
||||
if (e.error() != 'websocket: close sent') {
|
||||
socketErrors.add(1);
|
||||
console.log(`BENCH ERR ${Date.now()} ${e.error()}`);
|
||||
}
|
||||
});
|
||||
|
||||
socket.setTimeout(() => socket.close(), totalS * 1000);
|
||||
});
|
||||
|
||||
if (!res || res.status >= 400) {
|
||||
socketErrors.add(1);
|
||||
let status = "nil";
|
||||
if (res) {
|
||||
status = res.status;
|
||||
}
|
||||
console.log(`BENCH ERR ${Date.now()} connect_status_${status}`);
|
||||
}
|
||||
}
|
||||
|
||||
export function sender() {
|
||||
const cookie = `session_token=${dummyCookies[0][0]}`;
|
||||
|
||||
const page = http.get(`http://${host}${port}/rooms/1`, {
|
||||
headers: { "Cookie": cookie }, responseType: "text"
|
||||
});
|
||||
const csrfToken = page.body.match(/<meta name="csrf-token" content="([^"]*)"/i)[1];
|
||||
|
||||
const postHeaders = {
|
||||
"Cookie": cookie,
|
||||
"Accept": "text/vnd.turbo-stream.html, text/html, application/xhtml+xml"
|
||||
};
|
||||
|
||||
let seq = 0;
|
||||
const deadline = Date.now() + totalS * 1000;
|
||||
while (Date.now() < deadline) {
|
||||
seq++;
|
||||
const payload = {
|
||||
"message[body]": `bench:${Date.now()}:${seq}`,
|
||||
"message[client_message_id]": `bench-${seq}-${Math.random().toString(36)}`,
|
||||
"authenticity_token": csrfToken
|
||||
};
|
||||
const res = http.post(`http://${host}${port}/rooms/1/messages`, payload, {
|
||||
headers: postHeaders, responseType: "none", tags: { name: 'send_message' }
|
||||
});
|
||||
if (res.status >= 400) {
|
||||
console.log(`BENCH ERR ${Date.now()} post_status_${res.status}`);
|
||||
} else {
|
||||
console.log(`BENCH SENT ${Date.now()} ${seq}`);
|
||||
}
|
||||
sleep(sendIntervalS);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user