Misc improvment in load test (#42)

* Use docker to run them
* Use `gun` instead of `httpc`. `gun` supports http2, `httpc` does not.
* Align some default value between load test and neurow to have green
load tests without setting env var

Author: bpaquet

.github/workflows/ci.yml

@@ -35,9 +35,17 @@ jobs:
     steps:
       - uses: actions/checkout@v4
       - run: |
-          cd neurow 
+          cd neurow
           docker build . -t neurow:${{ github.sha }}
           docker run -e JWT_CONFIG="{\"local_name\":\"neurow\",\"service_name\":\"neurow\",\"algorithm\":\"HS256\",\"clients\":{\"test\":\"secret\"}}" -d --name neurow_${{ github.sha }}_test neurow:${{ github.sha }}
           docker exec neurow_${{ github.sha }}_test curl -v --retry 5 --retry-connrefused --retry-max-time 30 --retry-delay 6 http://localhost:3000/ping 
           docker stop neurow_${{ github.sha }}_test
           docker rm neurow_${{ github.sha }}_test
+
+  docker_load_test:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - run: |
+          cd load_test
+          docker build . -t neurow:${{ github.sha }}

load_test/Dockerfile

@@ -0,0 +1,43 @@
+ARG BUILDER_IMAGE=elixir
+ARG VERSION=1.17-slim
+
+FROM ${BUILDER_IMAGE}:${VERSION} AS builder
+
+RUN apt-get update \
+  && apt-get install -y --no-install-recommends git build-essential ca-certificates \
+  && apt-get clean \
+  && rm -rf /var/lib/apt/lists/*
+
+RUN mkdir /app
+WORKDIR /app
+ENV MIX_ENV=prod
+
+RUN mix local.hex --force
+
+COPY mix.exs mix.lock /app/
+RUN mix deps.get \
+  && mix deps.compile
+
+COPY config /app/config/
+COPY lib /app/lib/
+
+ARG GIT_COMMIT_SHA1=no_commit
+RUN mix release
+
+FROM ${BUILDER_IMAGE}:${VERSION}
+
+RUN apt-get update \
+  && apt-get install -y --no-install-recommends curl dnsutils ca-certificates \
+  && apt-get clean \
+  && rm -rf /var/lib/apt/lists/*
+
+RUN mkdir /app
+WORKDIR /app
+
+COPY start.sh /
+COPY --from=builder /app/_build/prod/rel/load_test /app/
+
+ENV RELEASE_TMP=/tmp/
+ENV RELEASE_COOKIE=changme
+
+CMD [ "/app/bin/load_test", "start"]

load_test/config/runtime.exs

@@ -8,7 +8,7 @@ config :load_test, port: String.to_integer(System.get_env("PORT") || "2999")
 config :load_test, nb_user: String.to_integer(System.get_env("NB_USER") || "1")
 
 config :load_test, sse_user_agent: System.get_env("SSE_USER_AGENT") || "neurow_load_test/1.0"
-config :load_test, sse_timeout: String.to_integer(System.get_env("SSE_TIMEOUT") || "15000")
+config :load_test, sse_timeout: String.to_integer(System.get_env("SSE_TIMEOUT") || "900000")
 config :load_test, sse_url: System.get_env("SSE_URL") || "http://localhost:4000/v1/subscribe"
 config :load_test, sse_jwt_issuer: System.get_env("SSE_JWT_ISSUER") || "test_issuer1"
 

load_test/lib/load_test/main.ex

@@ -75,8 +75,9 @@ defmodule LoadTest.Main do
       number_of_messages_max: number_of_messages_max
     }
 
-    Logger.warning("SSE BASE URL: #{sse_url}")
-    Logger.warning("PUBLISH BASE URL: #{publish_url}")
+    Logger.warning("SSE base url: #{sse_url}")
+    Logger.warning("Publish base url: #{publish_url}")
+    Logger.warning("User agent: #{sse_user_agent}")
     Logger.warning("Starting load test with #{nb_user} users")
 
     Enum.map(1..nb_user, fn _ ->

load_test/lib/load_test/user/publisher.ex

@@ -12,7 +12,8 @@ defmodule LoadTest.User.Publisher do
       :publish_jwt_audience,
       :delay_between_messages_min,
       :delay_between_messages_max,
-      :start_time
+      :start_time,
+      :message_count
     ]
   end
 
@@ -35,7 +36,8 @@ defmodule LoadTest.User.Publisher do
       publish_jwt_audience: context.publish_jwt_audience,
       delay_between_messages_min: context.delay_between_messages_min,
       delay_between_messages_max: context.delay_between_messages_max,
-      start_time: start_time
+      start_time: start_time,
+      message_count: length(messages)
     }
 
     Logger.debug(fn ->
@@ -73,7 +75,7 @@ defmodule LoadTest.User.Publisher do
     duration = :os.system_time(:millisecond) - state.start_time
 
     Logger.info(fn ->
-      "publisher_#{state.user_name}: All messages published to #{state.publish_url}, duration: #{duration / 1000}"
+      "publisher_#{state.user_name}: All messages published (#{state.message_count}) to #{state.publish_url}, duration: #{duration / 1000}"
     end)
   end
 

load_test/lib/load_test/user/sse.ex

@@ -46,10 +46,23 @@ defmodule SseUser do
     end)
 
     headers = build_headers(context, topic)
-    http_request_opts = []
 
-    {:ok, request_id} =
-      :httpc.request(:get, {url, headers}, http_request_opts, [{:sync, false}, {:stream, :self}])
+    parsed_url = URI.parse(url)
+
+    opts = %{
+      tls_opts: [
+        {:customize_hostname_check,
+         [{:match_fun, :public_key.pkix_verify_hostname_match_fun(:https)}]}
+      ],
+      http_opts: %{
+        closing_timeout: :infinity
+      }
+    }
+
+    {:ok, conn_pid} = :gun.open(String.to_atom(parsed_url.host), parsed_url.port, opts)
+    {:ok, proto} = :gun.await_up(conn_pid)
+    Logger.debug(fn -> "Connection established with proto #{inspect(proto)}" end)
+    stream_ref = :gun.get(conn_pid, parsed_url.path, headers)
 
     state = %SseState{
       user_name: user_name,
@@ -63,65 +76,54 @@ defmodule SseUser do
       end
     }
 
-    wait_for_messages(state, request_id, expected_messages)
+    wait_for_messages(state, conn_pid, stream_ref, expected_messages)
   end
 
-  defp wait_for_messages(state, request_id, [first_message | remaining_messages]) do
+  defp wait_for_messages(state, conn_pid, stream_ref, [first_message | remaining_messages]) do
     Logger.debug(fn -> "#{header(state)} Waiting for message: #{first_message}" end)
 
-    receive do
-      {:http, {_, {:error, msg}}} ->
-        Logger.error("#{header(state)} Http error: #{inspect(msg)}")
-        :ok = :httpc.cancel_request(request_id)
+    result = :gun.await(conn_pid, stream_ref, state.sse_timeout)
+
+    case result do
+      {:response, _, code, _} when code == 200 ->
+        Logger.debug(
+          "#{header(state)} Connected, waiting: #{length(remaining_messages) + 1} messages, url #{state.url}"
+        )
+
+        state.start_publisher_callback.()
+        wait_for_messages(state, conn_pid, stream_ref, [first_message | remaining_messages])
+
+      {:response, _, code, _} ->
+        Logger.error("#{header(state)} Error: #{inspect(code)}")
+        :ok = :gun.close(conn_pid)
         Stats.inc_msg_received_http_error()
-        raise("#{header(state)} Http error")
+        raise("#{header(state)} Error")
 
-      {:http, {_, :stream, msg}} ->
+      {:data, _, msg} ->
         msg = String.trim(msg)
         Logger.debug(fn -> "#{header(state)} Received message: #{inspect(msg)}" end)
 
         if msg =~ "event: ping" do
-          wait_for_messages(state, request_id, [first_message | remaining_messages])
+          wait_for_messages(state, conn_pid, stream_ref, [first_message | remaining_messages])
         else
           if check_message(state, msg, first_message) == :error do
-            :ok = :httpc.cancel_request(request_id)
+            :ok = :gun.close(conn_pid)
             raise("#{header(state)} Message check error")
           end
 
           state = Map.put(state, :current_message, state.current_message + 1)
-          wait_for_messages(state, request_id, remaining_messages)
+          wait_for_messages(state, conn_pid, stream_ref, remaining_messages)
         end
 
-      {:http, {_, :stream_start, headers}} ->
-        {~c"x-sse-server", server} = List.keyfind(headers, ~c"x-sse-server", 0)
-
-        Logger.debug(fn ->
-          "#{header(state)} Connected, waiting: #{length(remaining_messages) + 1} messages, url #{state.url}, remote server: #{server}"
-        end)
-
-        state.start_publisher_callback.()
-
-        wait_for_messages(state, request_id, [first_message | remaining_messages])
-
       msg ->
         Logger.error("#{header(state)} Unexpected message #{inspect(msg)}")
-        :ok = :httpc.cancel_request(request_id)
+        :ok = :gun.close(conn_pid)
         raise("#{header(state)} Unexpected message")
-    after
-      state.sse_timeout ->
-        Logger.error(
-          "#{header(state)} Timeout waiting for message (timeout=#{state.sse_timeout}ms), remaining: #{length(remaining_messages) + 1} messages, url #{state.url}"
-        )
-
-        Stats.inc_msg_received_timeout()
-
-        :ok = :httpc.cancel_request(request_id)
-        raise("#{header(state)} Timeout waiting for message")
     end
   end
 
-  defp wait_for_messages(state, request_id, []) do
-    :ok = :httpc.cancel_request(request_id)
+  defp wait_for_messages(state, conn_pid, _, []) do
+    :ok = :gun.close(conn_pid)
     Logger.info("#{header(state)} All messages received, url #{state.url}")
   end
 

load_test/mix.exs

@@ -23,7 +23,7 @@ defmodule LoadTest.MixProject do
 
   def application do
     [
-      extra_applications: [:logger, :inets],
+      extra_applications: [:logger, :inets, :gun],
       mod: {LoadTest.Application, []}
     ]
   end
@@ -38,7 +38,8 @@ defmodule LoadTest.MixProject do
       {:finch, "~> 0.18"},
       {:jose, "~> 1.11"},
       {:jiffy, "~> 1.1"},
-      {:observer_cli, "~> 1.7"}
+      {:observer_cli, "~> 1.7"},
+      {:gun, "~> 2.1"}
     ]
   end
 end

load_test/mix.lock

@@ -5,6 +5,7 @@
   "cowboy_telemetry": {:hex, :cowboy_telemetry, "0.4.0", "f239f68b588efa7707abce16a84d0d2acf3a0f50571f8bb7f56a15865aae820c", [:rebar3], [{:cowboy, "~> 2.7", [hex: :cowboy, repo: "hexpm", optional: false]}, {:telemetry, "~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "7d98bac1ee4565d31b62d59f8823dfd8356a169e7fcbb83831b8a5397404c9de"},
   "cowlib": {:hex, :cowlib, "2.13.0", "db8f7505d8332d98ef50a3ef34b34c1afddec7506e4ee4dd4a3a266285d282ca", [:make, :rebar3], [], "hexpm", "e1e1284dc3fc030a64b1ad0d8382ae7e99da46c3246b815318a4b848873800a4"},
   "finch": {:hex, :finch, "0.18.0", "944ac7d34d0bd2ac8998f79f7a811b21d87d911e77a786bc5810adb75632ada4", [:mix], [{:castore, "~> 0.1 or ~> 1.0", [hex: :castore, repo: "hexpm", optional: false]}, {:mime, "~> 1.0 or ~> 2.0", [hex: :mime, repo: "hexpm", optional: false]}, {:mint, "~> 1.3", [hex: :mint, repo: "hexpm", optional: false]}, {:nimble_options, "~> 0.4 or ~> 1.0", [hex: :nimble_options, repo: "hexpm", optional: false]}, {:nimble_pool, "~> 0.2.6 or ~> 1.0", [hex: :nimble_pool, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "69f5045b042e531e53edc2574f15e25e735b522c37e2ddb766e15b979e03aa65"},
+  "gun": {:hex, :gun, "2.1.0", "b4e4cbbf3026d21981c447e9e7ca856766046eff693720ba43114d7f5de36e87", [:make, :rebar3], [{:cowlib, "2.13.0", [hex: :cowlib, repo: "hexpm", optional: false]}], "hexpm", "52fc7fc246bfc3b00e01aea1c2854c70a366348574ab50c57dfe796d24a0101d"},
   "hpax": {:hex, :hpax, "1.0.0", "28dcf54509fe2152a3d040e4e3df5b265dcb6cb532029ecbacf4ce52caea3fd2", [:mix], [], "hexpm", "7f1314731d711e2ca5fdc7fd361296593fc2542570b3105595bb0bc6d0fad601"},
   "jiffy": {:hex, :jiffy, "1.1.2", "a9b6c9a7ec268e7cf493d028f0a4c9144f59ccb878b1afe42841597800840a1b", [:rebar3], [], "hexpm", "bb61bc42a720bbd33cb09a410e48bb79a61012c74cb8b3e75f26d988485cf381"},
   "jose": {:hex, :jose, "1.11.10", "a903f5227417bd2a08c8a00a0cbcc458118be84480955e8d251297a425723f83", [:mix, :rebar3], [], "hexpm", "0d6cd36ff8ba174db29148fc112b5842186b68a90ce9fc2b3ec3afe76593e614"},

load_test/start.sh

@@ -0,0 +1,4 @@
+#!/bin/sh -e
+
+ulimit -n 1000000
+exec /app/bin/load_test start

load_test/terraform/datasource.tf

@@ -1,11 +1,11 @@
 data "aws_ami" "ami-x86" {
   filter {
     name   = "name"
-    values = ["amzn2-ami-kernel-*-hvm-*x86*"]
+    values = ["*/ubuntu-noble-24.04-amd64-server*"]
   }
 
   most_recent = true
-  owners      = ["137112412989"]
+  owners      = ["099720109477"]
 }
 
 data "aws_subnet" "first_public" {