Add parallel and sequential execution modes

.gitignore

@@ -243,3 +243,6 @@ distributed_performance_analyzer
 credo_sonarqube.json
 sobelow.json
 generic_test_execution_sonarqube.xml
+
+# Hooks
+.github/hooks

README.md

@@ -41,7 +41,7 @@ Open and edit config/performance.exs file to configure.
 ```elixir
 import Config
 
-config :perf_analyzer,
+config :distributed_performance_analyzer,
   url: "http://httpbin.org/get",
   request: %{
     method: "GET",
@@ -73,6 +73,7 @@ config :logger,
 |---------------|----------------------------------------------------------------------------------------------------------------------------|
 | url           | The url of the application you want to test. Make sure you have a network connection between two machines                  |
 | request       | Here you need to configure the HTTP verb, headers and the body of the request.                                             |
+| requests      | This option is an array of request structures, and it combines with the modes, which are parallel and sequential. In case you leave it as default ":normal", it randomly selects a request from the array.                                             |
 | steps         | The number of executions for the test. Each step adds the concurrency configured in the increment                          |
 | increment     | Increment in concurrency after each step                                                                                   |
 | duration      | Duration in milliseconds of each step                                                                                      |
@@ -81,6 +82,7 @@ config :logger,
 | separator     | Dataset separator (, ; :)                                                                                                  |
 | distributed   | Indicates if it should be run from a single node or in a distributed way                                                   |
 | jmeter report | Generates jmeter csv style report?                                                                                         |
+| mode | This option is not required as mandatory. By default, it is :normal. There are two more atoms, namely :sequential and :parallel.                                                                                       |
 
 In the example above will be executed a test of 5 steps with an increment of 50:
 

config/dev.exs

@@ -1,21 +1,25 @@
 import Config
 
 config :distributed_performance_analyzer,
-  url: "http://localhost:8080/wait/100",
-  request: %{
-    method: "GET",
-    headers: [{"Content-Type", "application/json"}],
-    body: fn _item ->
-      ~s/'{"data":  #{Enum.random(1..10)},"key": 1}}}'/
-    end
-  },
+  url: "http://localhost:8080",
+  requests: [
+    %{
+      url: "http://localhost:8080/wait/100",
+      method: "GET",
+      headers: [{"Content-Type", "application/json"}],
+      body: fn _item ->
+        ~s/'{"data":  #{Enum.random(1..10)},"key": 1}}}'/
+      end
+    }
+  ],
   execution: %{
     steps: 5,
     increment: 1,
     duration: 3000,
     constant_load: true,
     dataset: :none,
-    separator: ","
+    separator: ",",
+    mode: :normal
   },
   distributed: :none,
   jmeter_report: true

config/performance.exs

@@ -3,17 +3,33 @@ import Config
 config :distributed_performance_analyzer,
   url: "http://localhost:8080/wait/100",
   request: %{
+    url: "http://localhost:8080/wait/100",
     method: "GET",
     headers: [{"Content-Type", "application/json"}],
-    body: ""
+    body: "Test 0"
   },
+  requests: [
+    %{
+      url: "http://localhost:8080/wait/100",
+      method: "GET",
+      headers: [{"Content-Type", "application/json"}],
+      body: "Test 1"
+    },
+    %{
+      url: "http://localhost:8080/wait/100",
+      method: "GET",
+      headers: [{"Content-Type", "application/json"}],
+      body: "Test 2"
+    }
+  ],
   execution: %{
     steps: 5,
     increment: 1,
     duration: 3000,
     constant_load: false,
     dataset: :none,
-    separator: ","
+    separator: ",",
+    mode: :sequential
   },
   distributed: :none,
   jmeter_report: true

config/test.exs

@@ -1,34 +1,38 @@
 import Config
 
 config :distributed_performance_analyzer,
-  url: "http://localhost:8080/wait/100",
-  request: %{
-    method: "GET",
-    headers: [
-      {"Content-Type", "application/json"}
-    ],
-    body: ~s/{
-        "data": {
-            "customer": {
-                "identification": {
-                    "type": "{type}",
-                    "number": "{number}"
-                }
-            },
-            "pagination": {
-                "size": #{Enum.random(0..10)},
-                "key": 1
-            }
-        }
-    }/
-  },
+  url: "http://localhost:8080",
+  requests: [
+    %{
+      url: "http://localhost:8080/wait/100",
+      method: "GET",
+      headers: [
+        {"Content-Type", "application/json"}
+      ],
+      body: ~s/{
+          "data": {
+              "customer": {
+                  "identification": {
+                      "type": "{type}",
+                      "number": "{number}"
+                  }
+              },
+              "pagination": {
+                  "size": #{Enum.random(0..10)},
+                  "key": 1
+              }
+          }
+      }/
+    }
+  ],
   execution: %{
     steps: 2,
     increment: 1,
     duration: 1500,
     constant_load: false,
     dataset: :none,
-    separator: ","
+    separator: ",",
+    mode: :normal
   },
   distributed: :none,
   jmeter_report: true

lib/domain/model/execution_model.ex

@@ -5,16 +5,15 @@ defmodule DistributedPerformanceAnalyzer.Domain.Model.ExecutionModel do
   TODO Execution model
   """
 
-  alias DistributedPerformanceAnalyzer.Domain.Model.Request
-
   constructor do
-    field(:request, Request.t(), constructor: &Request.new/1)
+    field(:requests, :lists, constructor: &is_list/1)
     field(:steps, :integer, constructor: &is_integer/1)
     field(:increment, :integer, constructor: &is_integer/1)
     field(:duration, :integer, constructor: &is_integer/1)
     field(:dataset, :atomics | :string | :lists)
     field(:separator, :string, constructor: &is_string/1, default: ",")
     field(:constant_load, :boolean, constructor: &is_boolean/1, default: false)
+    field(:mode, :atomics, constructor: &is_atom/1, default: :normal)
   end
 
   @impl Constructor

lib/domain/model/load_process.ex

@@ -4,21 +4,23 @@ defmodule DistributedPerformanceAnalyzer.Domain.Model.LoadProcess do
   @moduledoc """
   TODO Steps orchestration
   """
-  alias DistributedPerformanceAnalyzer.Domain.Model.{Request, Step}
+  alias DistributedPerformanceAnalyzer.Domain.Model.{Step}
 
   constructor do
-    field(:request, Request.t(), constructor: &Request.new/1, enforce: true)
+    field(:requests, :lists, constructor: &is_list/1, enforce: true)
     field(:step_name, :string, constructor: &is_string/1, enforce: true)
     field(:end_time, :integer, constructor: &is_integer/1, enforce: true)
+    field(:mode, :atomics, constructor: &is_atom/1, default: :normal, enforce: false)
   end
 
   @impl Constructor
   def before_construct(input = %Step{execution_model: execution_model, name: name})
       when is_map(input) do
     {:ok,
      %{
-       request: execution_model.request,
+       requests: execution_model.requests,
        step_name: name,
+       mode: execution_model.mode,
        end_time: :erlang.system_time(:milli_seconds) + execution_model.duration
      }}
   end

lib/domain/use_cases/config/config_use_case.ex

@@ -12,23 +12,23 @@ defmodule DistributedPerformanceAnalyzer.Domain.UseCase.Config.ConfigUseCase do
 
     %{
       host: host,
-      path: path,
       scheme: scheme,
-      port: port,
-      query: query
+      port: port
     } = ConfigParser.parse(url)
 
     connection_conf = {scheme, host, port}
 
-    {:ok, request} =
-      application_envs[:request]
-      |> Map.put(:path, ConfigParser.path(path, query))
-      |> Map.put(:url, url)
-      |> Request.new()
+    requests = ConfigParser.parse_requests(application_envs[:requests], url)
+    request = ConfigParser.parse_requests(application_envs[:request], url)
+
+    requests =
+      (request ++ requests)
+      |> Enum.map(&Request.new/1)
+      |> Enum.map(fn {:ok, result} -> result end)
 
     {:ok, execution_conf} =
       application_envs[:execution]
-      |> Map.put(:request, request)
+      |> Map.put(:requests, requests)
       |> ExecutionModel.new()
 
     {:ok,

lib/domain/use_cases/load_generator_use_case.ex

@@ -16,29 +16,70 @@ defmodule DistributedPerformanceAnalyzer.Domain.UseCase.LoadGeneratorUseCase do
 
   ## TODO Add functions to business logic app
   def start(
-        %LoadProcess{request: request, step_name: step_name, end_time: end_time},
+        %LoadProcess{requests: requests, step_name: step_name, end_time: end_time, mode: mode},
         dataset,
         concurrency
       ) do
     Task.start(fn ->
       conn = ConnectionPoolUseCase.get_connection()
 
       try do
-        results = generate_load(request, dataset, [], end_time, conn, concurrency)
+        results = generate_load(requests, dataset, [], end_time, conn, concurrency, mode)
         MetricsCollectorUseCase.send_metrics(results, step_name, concurrency)
       after
         ConnectionPoolUseCase.return_connection(conn)
       end
     end)
   end
 
-  defp generate_load(conf, dataset, results, end_time, conn, concurrency) do
+  defp generate_load(requests, dataset, results, end_time, conn, concurrency, :sequential) do
+    result =
+      requests
+      |> Enum.map(fn x ->
+        item = DatasetUseCase.get_random_item(dataset)
+        request(x, item, conn, concurrency)
+      end)
+
+    if actual_time() < end_time do
+      results = result ++ results
+      generate_load(requests, dataset, results, end_time, conn, concurrency, :sequential)
+    else
+      results
+    end
+  end
+
+  defp generate_load(requests, dataset, results, end_time, conn, concurrency, :parallel) do
+    result =
+      requests
+      |> Task.async_stream(
+        fn request ->
+          new_conn = ConnectionPoolUseCase.get_connection()
+          item = DatasetUseCase.get_random_item(dataset)
+          data = request(request, item, new_conn, concurrency)
+          ConnectionPoolUseCase.return_connection(new_conn)
+          data
+        end,
+        max_concurrency: length(requests)
+      )
+      |> Enum.map(fn {:ok, result} -> result end)
+      |> Enum.to_list()
+
+    if actual_time() < end_time do
+      results = result ++ results
+      generate_load(requests, dataset, results, end_time, conn, concurrency, :parallel)
+    else
+      results
+    end
+  end
+
+  defp generate_load(requests, dataset, results, end_time, conn, concurrency, :normal) do
     item = DatasetUseCase.get_random_item(dataset)
-    result = request(conf, item, conn, concurrency)
+    request = Enum.random(requests)
+    result = request(request, item, conn, concurrency)
 
     if actual_time() < end_time do
       results = [result | results]
-      generate_load(conf, dataset, results, end_time, conn, concurrency)
+      generate_load(requests, dataset, results, end_time, conn, concurrency, :normal)
     else
       results
     end

lib/domain/use_cases/load_step_use_case.ex

@@ -50,7 +50,17 @@ defmodule DistributedPerformanceAnalyzer.Domain.UseCase.LoadStepUseCase do
         },
         concurrency
       ) do
-    ConnectionPoolUseCase.ensure_capacity(concurrency)
+    mode = execution_model.mode
+    requests = execution_model.requests
+    IO.puts("Starting step #{name} with #{mode} mode")
+
+    capacity =
+      if mode == :parallel,
+        do: concurrency * length(requests),
+        else: concurrency
+
+    ConnectionPoolUseCase.ensure_capacity(capacity)
+
     {:ok, launch_config} = LoadProcess.new(step_model)
 
     loads =

lib/utils/config_parser.ex

@@ -28,4 +28,30 @@ defmodule DistributedPerformanceAnalyzer.Utils.ConfigParser do
 
   defp default_port("http"), do: 80
   defp default_port("https"), do: 443
+
+  def parse_requests(data, url_base) do
+    case data do
+      [request | rest] when is_map(request) ->
+        [parse_request(request, url_base) | parse_requests(rest, url_base)]
+
+      request when is_map(request) ->
+        [parse_request(request, url_base)]
+
+      _ ->
+        []
+    end
+  end
+
+  defp parse_request(request, url_base) do
+    url = if Map.has_key?(request, :url), do: request.url, else: url_base
+
+    %{
+      path: path,
+      query: query
+    } = parse(url)
+
+    request
+    |> Map.put(:path, path(path, query))
+    |> Map.put(:url, url)
+  end
 end