FIX: actor SIGTERM signal by finish or error

Dockerfile

@@ -21,5 +21,6 @@ RUN mix release
 FROM base
 ENV MIX_ENV=performance
 COPY --from=builder $WORKDIR/_build/prod ./
+EXPOSE 8083
 VOLUME config
 ENTRYPOINT rel/$APP_NAME/bin/$APP_NAME start

Dockerfile.local

@@ -24,5 +24,6 @@ RUN mix release
 FROM base
 ENV MIX_ENV=performance
 COPY --from=builder $WORKDIR/_build/prod ./
+EXPOSE 8083
 VOLUME config
 ENTRYPOINT rel/$APP_NAME/bin/$APP_NAME start

lib/application.ex

@@ -1,4 +1,4 @@
 defmodule DistributedPerformanceAnalyzer.Application do
   alias DistributedPerformanceAnalyzer.Config.{AppConfig, AppRegistry, ConfigHolder}
   alias DistributedPerformanceAnalyzer.Utils.{CertificatesAdmin, CustomTelemetry}
 
@@ -11,6 +11,8 @@
     Dataset.DatasetUseCase
   }
 
+  alias DistributedPerformanceAnalyzer.Infrastructure.EntryPoints.{LogBuffer, HttpServer}
+
   use Application
   require Logger
 
@@ -88,7 +90,8 @@
        strategy: :one_for_one,
        max_restarts: 10_000,
        max_seconds: 1},
-      AppRegistry
+      AppRegistry,
+      LogBuffer
     ]
 
     master_children = [
@@ -104,6 +107,13 @@
         children
       end
 
+    children =
+      if Application.get_env(:distributed_performance_analyzer, :enable_server, false) do
+        children ++ [HttpServer]
+      else
+        children
+      end
+
     pid = Supervisor.start_link(children, strategy: :one_for_one)
 
     if execution_conf.steps > 0 && distributed == :none do

lib/domain/use_cases/connection_pool_use_case.ex

@@ -4,6 +4,7 @@ defmodule DistributedPerformanceAnalyzer.Domain.UseCase.ConnectionPoolUseCase do
   """
   alias DistributedPerformanceAnalyzer.Domain.UseCase.ConnectionProcessUseCase
   alias DistributedPerformanceAnalyzer.Config.AppRegistry
+  alias DistributedPerformanceAnalyzer.Utils.DpaEvent
 
   use GenServer
   require Logger
@@ -39,16 +40,33 @@ defmodule DistributedPerformanceAnalyzer.Domain.UseCase.ConnectionPoolUseCase do
   @impl true
   def handle_call({:ensure_capacity, capacity}, _from, {scheme, host, port, pool, total_cap}) do
     actual = Enum.count(pool)
-    to_create = capacity - actual
 
     if capacity > actual do
+      to_create = capacity - actual
       actual_from = total_cap + 1
       capacity_to = total_cap + 1 + to_create
 
-      created =
+      results =
         Enum.map(actual_from..capacity_to, fn id -> create_connection(scheme, host, port, id) end)
 
-      {:reply, {:ok, to_create}, {scheme, host, port, created ++ pool, total_cap + to_create + 1}}
+      {successes, failures} = Enum.split_with(results, &match?({:ok, _}, &1))
+
+      if failures != [] do
+        Logger.warning(
+          "Failed to create #{length(failures)} connections out of #{length(results)} requested"
+        )
+
+        DpaEvent.emit(%{
+          type: "connection_pool_error",
+          failed_count: length(failures),
+          total_requested: length(results)
+        })
+      end
+
+      names = Enum.map(successes, fn {:ok, name} -> name end)
+
+      {:reply, {:ok, length(names)},
+       {scheme, host, port, names ++ pool, total_cap + to_create + 1}}
     else
       {:reply, {:ok, 0}, {scheme, host, port, pool, total_cap}}
     end
@@ -72,12 +90,12 @@ defmodule DistributedPerformanceAnalyzer.Domain.UseCase.ConnectionPoolUseCase do
   defp create_connection(scheme, host, port, id) do
     name = AppRegistry.via_tuple(id)
 
-    {:ok, _pid} =
-      DynamicSupervisor.start_child(
-        DPA.ConnectionSupervisor,
-        {ConnectionProcessUseCase, {scheme, host, port, name}}
-      )
-
-    name
+    case DynamicSupervisor.start_child(
+           DPA.ConnectionSupervisor,
+           {ConnectionProcessUseCase, {scheme, host, port, name}}
+         ) do
+      {:ok, _pid} -> {:ok, name}
+      {:error, reason} -> {:error, reason}
+    end
   end
 end

lib/domain/use_cases/execution_use_case.ex

@@ -11,6 +11,7 @@
   }
 
   alias DistributedPerformanceAnalyzer.Config.ConfigHolder
+  alias DistributedPerformanceAnalyzer.Utils.DpaEvent
   use GenServer
   require Logger
 
@@ -41,25 +42,20 @@
             "Resuming from step #{completed + 1}."
         )
 
-        IO.puts(
-          "DPA_EVENT " <>
-            Jason.encode!(%{
-              type: "resume_detected",
-              completed_steps: completed,
-              resuming_from: completed + 1,
-              total_steps: total_steps
-            })
-        )
+        DpaEvent.emit(%{
+          type: "resume_detected",
+          completed_steps: completed,
+          resuming_from: completed + 1,
+          total_steps: total_steps
+        })
 
         Application.put_env(:distributed_performance_analyzer, :dpa_resume_step, completed)
         completed + 1
       else
+        DpaEvent.rotate_log()
         ReportUseCase.init_report_files()
 
-        IO.puts(
-          "DPA_EVENT " <>
-            Jason.encode!(%{type: "execution_start", total_steps: total_steps})
-        )
+        DpaEvent.emit(%{type: "execution_start", total_steps: total_steps})
 
         Application.put_env(:distributed_performance_analyzer, :dpa_resume_step, 0)
         1
@@ -77,7 +73,7 @@
   @impl true
   def handle_call(:launch_execution, _from, conf) do
     IO.warn("Performance test already running")
     IO.inspect(conf)
     {:reply, :error, conf}
   end
 
@@ -102,9 +98,31 @@
   defp start_step({:ok, step_conf}) do
     IO.puts("Initiating #{step_conf.name}, with #{step_conf.concurrency} actors")
 
-    Task.start_link(fn ->
-      LoadStepUseCase.start_step(step_conf)
-      GenServer.cast(__MODULE__, :continue_execution)
-    end)
+    {:ok, pid} =
+      Task.start(fn ->
+        LoadStepUseCase.start_step(step_conf)
+        GenServer.cast(__MODULE__, :continue_execution)
+      end)
+
+    Process.monitor(pid)
+  end
+
+  @impl true
+  def handle_info({:DOWN, _ref, :process, _pid, :normal}, state) do
+    {:noreply, state}
+  end
+
+  @impl true
+  def handle_info({:DOWN, _ref, :process, _pid, reason}, state) do
+    Logger.error("Step task crashed: #{inspect(reason)}")
+
+    DpaEvent.emit(%{
+      type: "step_task_crashed",
+      reason: inspect(reason),
+      actual_step: state.actual_step
+    })
+
+    GenServer.cast(self(), :continue_execution)
+    {:noreply, state}
   end
 end

lib/domain/use_cases/load_step_use_case.ex

@@ -3,13 +3,20 @@ defmodule DistributedPerformanceAnalyzer.Domain.UseCase.LoadStepUseCase do
   Load step use case
   """
 
+  require Logger
+
   alias DistributedPerformanceAnalyzer.Domain.Model.{LoadProcess, Step}
 
   alias DistributedPerformanceAnalyzer.Domain.UseCase.{
     ConnectionPoolUseCase,
     LoadGeneratorUseCase
   }
 
+  alias DistributedPerformanceAnalyzer.Utils.DpaEvent
+
+  @max_actor_retries 3
+  @retry_delay_ms 500
+
   def start_step(step_model = %Step{}) do
     # TODO: Agregar timeout y manejar errores remotos
     node_list = [Node.self() | Node.list()]
@@ -66,9 +73,40 @@ defmodule DistributedPerformanceAnalyzer.Domain.UseCase.LoadStepUseCase do
     )
   end
 
-  defp start_load(launch_config, dataset, concurrency) do
-    {:ok, pid} = LoadGeneratorUseCase.start(launch_config, dataset, concurrency)
-    Process.monitor(pid)
+  defp start_load(launch_config, dataset, concurrency, attempt \\ 1) do
+    case LoadGeneratorUseCase.start(launch_config, dataset, concurrency) do
+      {:ok, pid} ->
+        Process.monitor(pid)
+
+      {:error, reason} when attempt <= @max_actor_retries ->
+        Logger.warning(
+          "Actor failed to start (attempt #{attempt}/#{@max_actor_retries}): #{inspect(reason)}"
+        )
+
+        DpaEvent.emit(%{
+          type: "actor_retry",
+          attempt: attempt,
+          max_retries: @max_actor_retries,
+          reason: inspect(reason)
+        })
+
+        Process.sleep(@retry_delay_ms * attempt)
+        start_load(launch_config, dataset, concurrency, attempt + 1)
+
+      {:error, reason} ->
+        Logger.error(
+          "Actor permanently failed after #{@max_actor_retries} attempts: #{inspect(reason)}"
+        )
+
+        DpaEvent.emit(%{
+          type: "actor_failed",
+          reason: inspect(reason),
+          attempts: @max_actor_retries
+        })
+
+        {_pid, ref} = spawn_monitor(fn -> :ok end)
+        ref
+    end
   end
 
   defp wait_for(ref, timeout) do

lib/domain/use_cases/partial_result_use_case.ex

@@ -4,6 +4,7 @@
   """
   alias DistributedPerformanceAnalyzer.Domain.Model.PartialResult
   alias DistributedPerformanceAnalyzer.Utils.{Statistics, DataTypeUtils}
+  alias DistributedPerformanceAnalyzer.Utils.DpaEvent
 
   require Logger
 
@@ -30,7 +31,7 @@
     partial
   end
 
   def consolidate(
         %PartialResult{
           success_times: success_times,
           times: times,
@@ -116,27 +117,24 @@
       "Concurrency -> users: #{concurrency} - tps: #{throughput} | Latency -> min: #{min}ms - avg: #{avg}ms - max: #{max}ms - p90: #{p90}ms | Requests -> 2xx: #{status_200} - 4xx: #{status_400} - 5xx: #{status_500} | others_errors: #{nil_conn_errors + invocation_errors + protocol_errors + conn_errors} | total_errors: #{errors} - total_request: #{total}"
     )
 
-    IO.puts(
-      "DPA_EVENT " <>
-        Jason.encode!(%{
-          type: "step_complete",
-          concurrency: concurrency,
-          throughput: throughput,
-          min_latency: min,
-          avg_latency: avg,
-          max_latency: max,
-          p90_latency: p90,
-          success_count: status_200,
-          bad_request_count: status_400,
-          server_error_count: status_500,
-          nil_conn_errors: nil_conn_errors,
-          invocation_errors: invocation_errors,
-          protocol_errors: protocol_errors,
-          conn_errors: conn_errors,
-          error_count: errors,
-          total_count: total
-        })
-    )
+    DpaEvent.emit(%{
+      type: "step_complete",
+      concurrency: concurrency,
+      throughput: throughput,
+      min_latency: min,
+      avg_latency: avg,
+      max_latency: max,
+      p90_latency: p90,
+      success_count: status_200,
+      bad_request_count: status_400,
+      server_error_count: status_500,
+      nil_conn_errors: nil_conn_errors,
+      invocation_errors: invocation_errors,
+      protocol_errors: protocol_errors,
+      conn_errors: conn_errors,
+      error_count: errors,
+      total_count: total
+    })
   end
 
   def calculate(result_list, opts) do

lib/infrastructure/entry_points/dpa_router.ex

@@ -0,0 +1,53 @@
+defmodule DistributedPerformanceAnalyzer.Infrastructure.EntryPoints.DpaRouter do
+  @moduledoc """
+  HTTP polling endpoint for DPA events.
+
+  Routes:
+    GET /health            — liveness probe: {"status":"ok"}
+    GET /events            — returns all buffered events
+    GET /events?since=<ms> — returns events with timestamp >= since_ms (Unix ms)
+  """
+
+  use Plug.Router
+
+  alias DistributedPerformanceAnalyzer.Infrastructure.EntryPoints.LogBuffer
+
+  plug CORSPlug
+  plug :match
+  plug :dispatch
+
+  get "/health" do
+    send_json(conn, 200, %{status: "ok"})
+  end
+
+  get "/events" do
+    conn = Plug.Conn.fetch_query_params(conn)
+    since_ms = parse_since(conn.query_params["since"])
+
+    events =
+      if since_ms == 0,
+        do: LogBuffer.get_all(),
+        else: LogBuffer.get_since(since_ms)
+
+    send_json(conn, 200, %{events: events})
+  end
+
+  match _ do
+    send_resp(conn, 404, "not found")
+  end
+
+  defp send_json(conn, status, body) do
+    conn
+    |> put_resp_content_type("application/json")
+    |> send_resp(status, Jason.encode!(body))
+  end
+
+  defp parse_since(nil), do: 0
+
+  defp parse_since(val) do
+    case Integer.parse(val) do
+      {n, _} -> max(n, 0)
+      :error -> 0
+    end
+  end
+end

lib/infrastructure/entry_points/http_server.ex

@@ -0,0 +1,16 @@
+defmodule DistributedPerformanceAnalyzer.Infrastructure.EntryPoints.HttpServer do
+  @moduledoc """
+  Starts a Cowboy HTTP server on the port configured by :http_port (default 8083).
+  Only added to the supervision tree when :enable_server is true in app config.
+  """
+
+  require Logger
+
+  alias DistributedPerformanceAnalyzer.Infrastructure.EntryPoints.DpaRouter
+
+  def child_spec(_opts) do
+    port = Application.get_env(:distributed_performance_analyzer, :http_port, 8083)
+    Logger.info("Starting DPA HTTP server on port #{port}")
+    Plug.Cowboy.child_spec(scheme: :http, plug: DpaRouter, options: [port: port])
+  end
+end