diff --git a/Dockerfile b/Dockerfile index 9dd84d9..d7fdce8 100644 --- a/Dockerfile +++ b/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 \ No newline at end of file diff --git a/Dockerfile.local b/Dockerfile.local index 5be9922..edf297e 100644 --- a/Dockerfile.local +++ b/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 diff --git a/lib/application.ex b/lib/application.ex index 30f22e4..1d3cfef 100644 --- a/lib/application.ex +++ b/lib/application.ex @@ -11,6 +11,8 @@ defmodule DistributedPerformanceAnalyzer.Application do Dataset.DatasetUseCase } + alias DistributedPerformanceAnalyzer.Infrastructure.EntryPoints.{LogBuffer, HttpServer} + use Application require Logger @@ -88,7 +90,8 @@ defmodule DistributedPerformanceAnalyzer.Application do strategy: :one_for_one, max_restarts: 10_000, max_seconds: 1}, - AppRegistry + AppRegistry, + LogBuffer ] master_children = [ @@ -104,6 +107,13 @@ defmodule DistributedPerformanceAnalyzer.Application do 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 diff --git a/lib/domain/use_cases/connection_pool_use_case.ex b/lib/domain/use_cases/connection_pool_use_case.ex index 938e43f..307326a 100644 --- a/lib/domain/use_cases/connection_pool_use_case.ex +++ b/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 diff --git a/lib/domain/use_cases/execution_use_case.ex b/lib/domain/use_cases/execution_use_case.ex index 8ba8450..10893d7 100644 --- a/lib/domain/use_cases/execution_use_case.ex +++ b/lib/domain/use_cases/execution_use_case.ex @@ -11,6 +11,7 @@ defmodule DistributedPerformanceAnalyzer.Domain.UseCase.ExecutionUseCase do } alias DistributedPerformanceAnalyzer.Config.ConfigHolder + alias DistributedPerformanceAnalyzer.Utils.DpaEvent use GenServer require Logger @@ -41,25 +42,20 @@ defmodule DistributedPerformanceAnalyzer.Domain.UseCase.ExecutionUseCase do "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 @@ -102,9 +98,31 @@ defmodule DistributedPerformanceAnalyzer.Domain.UseCase.ExecutionUseCase do 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 diff --git a/lib/domain/use_cases/load_step_use_case.ex b/lib/domain/use_cases/load_step_use_case.ex index c595fee..63999d2 100644 --- a/lib/domain/use_cases/load_step_use_case.ex +++ b/lib/domain/use_cases/load_step_use_case.ex @@ -3,6 +3,8 @@ defmodule DistributedPerformanceAnalyzer.Domain.UseCase.LoadStepUseCase do Load step use case """ + require Logger + alias DistributedPerformanceAnalyzer.Domain.Model.{LoadProcess, Step} alias DistributedPerformanceAnalyzer.Domain.UseCase.{ @@ -10,6 +12,11 @@ defmodule DistributedPerformanceAnalyzer.Domain.UseCase.LoadStepUseCase do 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 diff --git a/lib/domain/use_cases/partial_result_use_case.ex b/lib/domain/use_cases/partial_result_use_case.ex index ad70e89..83c013a 100644 --- a/lib/domain/use_cases/partial_result_use_case.ex +++ b/lib/domain/use_cases/partial_result_use_case.ex @@ -4,6 +4,7 @@ defmodule DistributedPerformanceAnalyzer.Domain.UseCase.PartialResultUseCase do """ alias DistributedPerformanceAnalyzer.Domain.Model.PartialResult alias DistributedPerformanceAnalyzer.Utils.{Statistics, DataTypeUtils} + alias DistributedPerformanceAnalyzer.Utils.DpaEvent require Logger @@ -116,27 +117,24 @@ defmodule DistributedPerformanceAnalyzer.Domain.UseCase.PartialResultUseCase do "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 diff --git a/lib/infrastructure/entry_points/dpa_router.ex b/lib/infrastructure/entry_points/dpa_router.ex new file mode 100644 index 0000000..1c9615d --- /dev/null +++ b/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= — 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 diff --git a/lib/infrastructure/entry_points/http_server.ex b/lib/infrastructure/entry_points/http_server.ex new file mode 100644 index 0000000..0c64f0e --- /dev/null +++ b/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 diff --git a/lib/infrastructure/entry_points/log_buffer.ex b/lib/infrastructure/entry_points/log_buffer.ex new file mode 100644 index 0000000..4cc41b1 --- /dev/null +++ b/lib/infrastructure/entry_points/log_buffer.ex @@ -0,0 +1,50 @@ +defmodule DistributedPerformanceAnalyzer.Infrastructure.EntryPoints.LogBuffer do + @max_size 500 + + @moduledoc """ + In-memory ring-buffer of the last #{@max_size} DPA events. + Used by DpaRouter to serve HTTP polling requests from the extension. + """ + + use GenServer + + def start_link(_), do: GenServer.start_link(__MODULE__, [], name: __MODULE__) + + @spec push(map()) :: :ok + def push(event), do: GenServer.cast(__MODULE__, {:push, event}) + + @spec get_since(integer()) :: [map()] + def get_since(since_ms) when is_integer(since_ms), + do: GenServer.call(__MODULE__, {:get_since, since_ms}) + + @spec get_all() :: [map()] + def get_all, do: GenServer.call(__MODULE__, :get_all) + + @impl true + def init(_), do: {:ok, []} + + @impl true + def handle_cast({:push, event}, buffer) do + {:noreply, [event | buffer] |> Enum.take(@max_size)} + end + + @impl true + def handle_call({:get_since, since_ms}, _from, buffer) do + events = + buffer + |> Enum.filter(fn event -> + case DateTime.from_iso8601(Map.get(event, :timestamp, "")) do + {:ok, dt, _} -> DateTime.to_unix(dt, :millisecond) >= since_ms + _ -> false + end + end) + |> Enum.reverse() + + {:reply, events, buffer} + end + + @impl true + def handle_call(:get_all, _from, buffer) do + {:reply, Enum.reverse(buffer), buffer} + end +end diff --git a/lib/utils/dpa_event.ex b/lib/utils/dpa_event.ex new file mode 100644 index 0000000..1e544c6 --- /dev/null +++ b/lib/utils/dpa_event.ex @@ -0,0 +1,61 @@ +defmodule DistributedPerformanceAnalyzer.Utils.DpaEvent do + @moduledoc """ + Centralized DPA event emitter. + + Every call to `emit/1` does three things in order: + 1. Writes "DPA_EVENT " to stdout (preserves existing behavior). + 2. Appends the JSON line to `config/dpa_events.log` (NDJSON, readable from host via bind-mount). + 3. Pushes the event into the in-memory LogBuffer for HTTP polling on :8083. + + Call `rotate_log/0` at the start of each fresh execution (not resumes) to + truncate the file so each run has its own clean log. + """ + + require Logger + + alias DistributedPerformanceAnalyzer.Infrastructure.EntryPoints.LogBuffer + + @log_path "config/dpa_events.log" + + @spec emit(map()) :: :ok + def emit(payload) when is_map(payload) do + timestamped = Map.put(payload, :timestamp, iso8601_now()) + json = Jason.encode!(timestamped) + IO.puts("DPA_EVENT " <> json) + append_to_file(json) + push_to_buffer(timestamped) + :ok + end + + @spec rotate_log() :: :ok + def rotate_log do + case File.write(@log_path, "") do + :ok -> + :ok + + {:error, reason} -> + Logger.warning("DpaEvent: could not rotate log file: #{inspect(reason)}") + end + end + + defp append_to_file(json) do + case File.write(@log_path, json <> "\n", [:append]) do + :ok -> + :ok + + {:error, reason} -> + Logger.warning("DpaEvent: failed to write log file: #{inspect(reason)}") + end + end + + defp push_to_buffer(event) do + case Process.whereis(LogBuffer) do + nil -> :ok + _pid -> LogBuffer.push(event) + end + end + + defp iso8601_now do + DateTime.utc_now() |> DateTime.to_iso8601() + end +end diff --git a/mix.exs b/mix.exs index b9dc365..ecd2f2e 100644 --- a/mix.exs +++ b/mix.exs @@ -53,11 +53,10 @@ defmodule DistributedPerformanceAnalyzer.MixProject do {:sobelow, "~> 0.13", only: :dev}, {:credo_sonarqube, "~> 0.1"}, {:finch, "~> 0.13"}, - {:opentelemetry_plug, - git: "https://github.com/juancgalvis/opentelemetry_plug.git", tag: "master"}, - {:opentelemetry_api, "~> 1.2"}, + {:opentelemetry_plug, git: "https://github.com/opentelemetry-beam/opentelemetry_plug.git", tag: "master"}, + {:opentelemetry_api, "~> 1.2", override: true}, {:opentelemetry_exporter, "~> 1.6"}, - {:telemetry, "~> 1.0"}, + {:telemetry, "~> 0.4 or ~> 1.0", override: true}, {:telemetry_poller, "~> 1.0"}, {:telemetry_metrics_prometheus, "~> 1.0"}, {:castore, "~> 1.0"}, diff --git a/mix.lock b/mix.lock index 19ee69b..269bbe4 100644 --- a/mix.lock +++ b/mix.lock @@ -47,7 +47,6 @@ "opentelemetry": {:hex, :opentelemetry, "1.3.1", "f0a342a74379e3540a634e7047967733da4bc8b873ec9026e224b2bd7369b1fc", [:rebar3], [{:opentelemetry_api, "~> 1.2.2", [hex: :opentelemetry_api, repo: "hexpm", optional: false]}, {:opentelemetry_semantic_conventions, "~> 0.2", [hex: :opentelemetry_semantic_conventions, repo: "hexpm", optional: false]}], "hexpm", "de476b2ac4faad3e3fe3d6e18b35dec9cb338c3b9910c2ce9317836dacad3483"}, "opentelemetry_api": {:hex, :opentelemetry_api, "1.2.2", "693f47b0d8c76da2095fe858204cfd6350c27fe85d00e4b763deecc9588cf27a", [:mix, :rebar3], [{:opentelemetry_semantic_conventions, "~> 0.2", [hex: :opentelemetry_semantic_conventions, repo: "hexpm", optional: false]}], "hexpm", "dc77b9a00f137a858e60a852f14007bb66eda1ffbeb6c05d5fe6c9e678b05e9d"}, "opentelemetry_exporter": {:hex, :opentelemetry_exporter, "1.6.0", "f4fbf69aa9f1541b253813221b82b48a9863bc1570d8ecc517bc510c0d1d3d8c", [:rebar3], [{:grpcbox, ">= 0.0.0", [hex: :grpcbox, repo: "hexpm", optional: false]}, {:opentelemetry, "~> 1.3", [hex: :opentelemetry, repo: "hexpm", optional: false]}, {:opentelemetry_api, "~> 1.2", [hex: :opentelemetry_api, repo: "hexpm", optional: false]}, {:tls_certificate_check, "~> 1.18", [hex: :tls_certificate_check, repo: "hexpm", optional: false]}], "hexpm", "1802d1dca297e46f21e5832ecf843c451121e875f73f04db87355a6cb2ba1710"}, - "opentelemetry_plug": {:git, "https://github.com/juancgalvis/opentelemetry_plug.git", "28c6a4b11001bdb74ba488c33d5558012cfa7425", [tag: "master"]}, "opentelemetry_semantic_conventions": {:hex, :opentelemetry_semantic_conventions, "0.2.0", "b67fe459c2938fcab341cb0951c44860c62347c005ace1b50f8402576f241435", [:mix, :rebar3], [], "hexpm", "d61fa1f5639ee8668d74b527e6806e0503efc55a42db7b5f39939d84c07d6895"}, "parse_trans": {:hex, :parse_trans, "3.3.1", "16328ab840cc09919bd10dab29e431da3af9e9e7e7e6f0089dd5a2d2820011d8", [:rebar3], [], "hexpm", "07cd9577885f56362d414e8c4c4e6bdf10d43a8767abb92d24cbe8b24c54888b"}, "plug": {:hex, :plug, "1.14.2", "cff7d4ec45b4ae176a227acd94a7ab536d9b37b942c8e8fa6dfc0fff98ff4d80", [:mix], [{:mime, "~> 1.0 or ~> 2.0", [hex: :mime, repo: "hexpm", optional: false]}, {:plug_crypto, "~> 1.1.1 or ~> 1.2", [hex: :plug_crypto, repo: "hexpm", optional: false]}, {:telemetry, "~> 0.4.3 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "842fc50187e13cf4ac3b253d47d9474ed6c296a8732752835ce4a86acdf68d13"},