diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index a3ed857..fcd4e4b 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -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 diff --git a/Dockerfile b/Dockerfile index 6cfbd33..482285f 100644 --- a/Dockerfile +++ b/Dockerfile @@ -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 diff --git a/Gemfile b/Gemfile index 1077c90..1bcf326 100644 --- a/Gemfile +++ b/Gemfile @@ -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" diff --git a/Gemfile.lock b/Gemfile.lock index 1cd6c74..51fccd6 100644 --- a/Gemfile.lock +++ b/Gemfile.lock @@ -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 diff --git a/Procfile b/Procfile index 20e1430..4eab710 100644 --- a/Procfile +++ b/Procfile @@ -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 diff --git a/bin/bench b/bin/bench new file mode 100755 index 0000000..d848389 --- /dev/null +++ b/bin/bench @@ -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]}/" diff --git a/bin/bench-report b/bin/bench-report new file mode 100755 index 0000000..bff8d12 --- /dev/null +++ b/bin/bench-report @@ -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 } diff --git a/bin/jobs b/bin/jobs new file mode 100755 index 0000000..dcf59f3 --- /dev/null +++ b/bin/jobs @@ -0,0 +1,6 @@ +#!/usr/bin/env ruby + +require_relative "../config/environment" +require "solid_queue/cli" + +SolidQueue::Cli.start(ARGV) diff --git a/bin/setup b/bin/setup index e0f5bce..73492dd 100755 --- a/bin/setup +++ b/bin/setup @@ -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 diff --git a/config/cable.yml b/config/cable.yml index e95274f..0e2f3ca 100644 --- a/config/cable.yml +++ b/config/cable.yml @@ -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 diff --git a/config/cache.yml b/config/cache.yml new file mode 100644 index 0000000..d5b78d8 --- /dev/null +++ b/config/cache.yml @@ -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 diff --git a/config/database.yml b/config/database.yml index ec220a4..fd02a94 100644 --- a/config/database.yml +++ b/config/database.yml @@ -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 diff --git a/config/environments/development.rb b/config/environments/development.rb index 7994eb0..b1aaafd 100644 --- a/config/environments/development.rb +++ b/config/environments/development.rb @@ -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 diff --git a/config/environments/production.rb b/config/environments/production.rb index d58644a..16e9279 100644 --- a/config/environments/production.rb +++ b/config/environments/production.rb @@ -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 diff --git a/config/puma.rb b/config/puma.rb index 960ab6a..d82595e 100644 --- a/config/puma.rb +++ b/config/puma.rb @@ -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 diff --git a/config/queue.yml b/config/queue.yml new file mode 100644 index 0000000..72ca6da --- /dev/null +++ b/config/queue.yml @@ -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 diff --git a/config/recurring.yml b/config/recurring.yml new file mode 100644 index 0000000..b4207f9 --- /dev/null +++ b/config/recurring.yml @@ -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 diff --git a/config/redis.conf b/config/redis.conf deleted file mode 100644 index 804d865..0000000 --- a/config/redis.conf +++ /dev/null @@ -1,4 +0,0 @@ -port 6379 -daemonize no -appendonly no -save "" diff --git a/config/resque-pool.yml b/config/resque-pool.yml deleted file mode 100644 index 64f569b..0000000 --- a/config/resque-pool.yml +++ /dev/null @@ -1 +0,0 @@ -default: <%= (Concurrent.processor_count * 0.5).ceil %> diff --git a/db/cable_schema.rb b/db/cable_schema.rb new file mode 100644 index 0000000..2366660 --- /dev/null +++ b/db/cable_schema.rb @@ -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 diff --git a/db/cache_schema.rb b/db/cache_schema.rb new file mode 100644 index 0000000..81a410d --- /dev/null +++ b/db/cache_schema.rb @@ -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 diff --git a/db/queue_schema.rb b/db/queue_schema.rb new file mode 100644 index 0000000..85194b6 --- /dev/null +++ b/db/queue_schema.rb @@ -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 diff --git a/lib/tasks/resque.rake b/lib/tasks/resque.rake deleted file mode 100644 index 63ecc9d..0000000 --- a/lib/tasks/resque.rake +++ /dev/null @@ -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 diff --git a/test/performance/ramp.js b/test/performance/ramp.js new file mode 100644 index 0000000..38b1eb4 --- /dev/null +++ b/test/performance/ramp.js @@ -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(/= 400) { + console.log(`BENCH ERR ${Date.now()} post_status_${res.status}`); + } else { + console.log(`BENCH SENT ${Date.now()} ${seq}`); + } + sleep(sendIntervalS); + } +}