Persist tracing sessions

lib/voyager/actions/trace_entries.ex

@@ -0,0 +1,125 @@
+defmodule Voyager.Actions.TraceEntries do
+  @moduledoc """
+  Writes `Voyager.Services.Tracer` log entries of a saved session: `:trace` entries
+  as `Voyager.Schemas.TraceEvent` rows, the rest as `Voyager.Schemas.TraceSessionEvent` rows.
+  """
+
+  alias Voyager.Repo
+  alias Voyager.Schemas.TraceEvent
+  alias Voyager.Schemas.TraceSessionEvent
+  alias Voyager.StoredTerm
+
+  # Measured on SQLite: id, session_id, seq, `at` as ISO text, kind, flags and the seq index.
+  @row_overhead_bytes 64
+
+  @blank_session_event Map.new(TraceSessionEvent.__schema__(:fields) -- [:id], &{&1, nil})
+
+  @type prepared :: %{trace: [map()], session: [map()]}
+
+  @doc """
+  Inserts the entries. A trace's `search_text` has its terms inspected, with pids
+  in the traced node's `<0.X.Y>` form. Returns the bytes stored plus a fixed cost per row.
+  """
+  @spec insert_all(integer(), [map()]) :: non_neg_integer()
+  def insert_all(session_id, entries),
+    do: session_id |> prepare(entries) |> insert_prepared()
+
+  @doc "Encodes the entries into rows without touching the database, so it can run in parallel."
+  @spec prepare(integer(), [map()]) :: prepared()
+  def prepare(session_id, entries) do
+    {traces, session_events} = Enum.split_with(entries, &(&1.kind == :trace))
+
+    %{
+      trace: Enum.map(traces, &trace_row(session_id, &1)),
+      session: Enum.map(session_events, &session_event_row(session_id, &1))
+    }
+  end
+
+  @doc "Inserts rows from `prepare/2`; returns what `insert_all/2` does."
+  @spec insert_prepared(prepared()) :: non_neg_integer()
+  def insert_prepared(%{trace: traces, session: session_events}),
+    do: insert(TraceEvent, traces) + insert(TraceSessionEvent, session_events)
+
+  defp insert(_schema, []), do: 0
+
+  defp insert(schema, rows) do
+    Repo.insert_all(schema, rows)
+    rows |> Enum.map(&row_bytes/1) |> Enum.sum()
+  end
+
+  defp session_event_row(session_id, entry) do
+    @blank_session_event
+    |> Map.merge(base(session_id, entry))
+    |> Map.put(:kind, entry.kind)
+    |> Map.merge(session_event_fields(entry))
+  end
+
+  defp session_event_fields(%{kind: :configured} = entry) do
+    %{
+      origin: entry.origin,
+      node: Atom.to_string(entry.node),
+      config: Map.from_struct(entry.config)
+    }
+  end
+
+  defp session_event_fields(%{kind: :started} = entry), do: %{origin: entry.origin}
+
+  defp session_event_fields(%{kind: :stopped} = entry) do
+    %{
+      origin: entry.origin,
+      reason: StoredTerm.encode(entry.reason),
+      traces: entry.traces,
+      dropped: entry.dropped
+    }
+  end
+
+  defp session_event_fields(%{kind: :dropped} = entry), do: %{dropped: entry.count}
+
+  defp trace_row(session_id, %{mfa: {m, f, a}} = entry) do
+    Map.merge(base(session_id, entry), %{
+      type: entry.type,
+      pid: StoredTerm.encode(entry.pid),
+      process: StoredTerm.encode(entry.process),
+      module: Atom.to_string(m),
+      function: Atom.to_string(f),
+      arity: a,
+      args: if(Map.has_key?(entry, :args), do: StoredTerm.encode(entry.args)),
+      value: if(Map.has_key?(entry, :value), do: StoredTerm.encode(entry.value)),
+      truncated: entry[:truncated],
+      duration_us: entry[:duration_us],
+      search_text: search_text(entry)
+    })
+  end
+
+  defp search_text(%{mfa: {m, f, a}} = entry) do
+    terms = for key <- [:args, :value], Map.has_key?(entry, key), do: inspect_term(entry[key])
+
+    Enum.join(
+      [
+        inspect_term(entry.pid),
+        inspect_term(entry.process),
+        Exception.format_mfa(m, f, a) | terms
+      ],
+      " "
+    )
+  end
+
+  defp base(session_id, entry),
+    do: %{session_id: session_id, seq: entry.seq, at: DateTime.from_unix!(entry.at, :microsecond)}
+
+  defp inspect_term(term),
+    do: inspect(term, limit: :infinity, printable_limit: :infinity, inspect_fun: &inspect_pid/2)
+
+  defp inspect_pid(pid, _opts) when is_pid(pid), do: "#PID" <> Voyager.Pid.format(pid, :local)
+  defp inspect_pid(term, opts), do: Inspect.inspect(term, opts)
+
+  defp row_bytes(row) do
+    row
+    |> Map.values()
+    |> Enum.filter(&is_binary/1)
+    |> Enum.reduce(@row_overhead_bytes + config_bytes(row), &(byte_size(&1) + &2))
+  end
+
+  defp config_bytes(%{config: config}) when is_map(config), do: :erlang.external_size(config)
+  defp config_bytes(_row), do: 0
+end

lib/voyager/actions/trace_sessions.ex

@@ -0,0 +1,62 @@
+defmodule Voyager.Actions.TraceSessions do
+  @moduledoc "Write actions for saved tracing sessions."
+
+  import Ecto.Query
+
+  alias Voyager.Queries.TraceSessions, as: TraceSessionQueries
+  alias Voyager.Repo
+  alias Voyager.Schemas.TraceSession
+  alias Voyager.Services.Tracer.Session
+
+  @spec create(map()) :: {:ok, TraceSession.t()} | {:error, Ecto.Changeset.t()}
+  def create(attrs), do: %TraceSession{} |> TraceSession.changeset(attrs) |> Repo.insert()
+
+  @doc "Writes the Tracer session's state to its row, adding `stored_bytes` to its `bytes`. Returns the new total."
+  @spec sync(Session.t(), non_neg_integer()) :: non_neg_integer()
+  def sync(%Session{} = session, stored_bytes) do
+    {1, [bytes]} =
+      TraceSession
+      |> where(id: ^session.id)
+      |> select([s], s.bytes)
+      |> Repo.update_all(
+        set: [
+          status: session.status,
+          node: session.node && Atom.to_string(session.node),
+          config: session.config && Map.from_struct(session.config),
+          stop_reason: session.stop_reason,
+          traces: session.traces,
+          dropped: session.dropped,
+          started_at: session.started_at,
+          stopped_at: session.stopped_at,
+          updated_at: NaiveDateTime.utc_now(:second)
+        ],
+        inc: [bytes: stored_bytes]
+      )
+
+    bytes
+  end
+
+  @doc "Deletes the session and its entries."
+  @spec delete(integer()) :: {:ok, TraceSession.t()} | {:error, :not_found}
+  def delete(id) do
+    case TraceSessionQueries.get(id) do
+      nil -> {:error, :not_found}
+      session -> Repo.delete(session)
+    end
+  end
+
+  @doc """
+  Run on boot: sessions left `running` are stopped with reason `interrupted`,
+  since the remote tracer stops when Voyager dies.
+  """
+  @spec interrupt_running() :: :ok
+  def interrupt_running do
+    TraceSession
+    |> where(status: :running)
+    |> Repo.update_all(
+      set: [status: :stopped, stop_reason: :interrupted, stopped_at: DateTime.utc_now()]
+    )
+
+    :ok
+  end
+end

lib/voyager/application.ex

@@ -19,7 +19,7 @@ defmodule Voyager.Application do
       {Task.Supervisor, name: Voyager.TaskSupervisor},
       Voyager.ProxyEpmd.TunnelRegistry,
       Voyager.NodeSession,
-      Voyager.Services.Tracer,
+      Voyager.Services.Tracer.Supervisor,
       {ElixirKit.PubSub, connect: elixirkit_pubsub || :ignore, on_exit: fn -> System.stop() end},
       {Voyager.Services.RateLimiter,
        config: Application.get_env(:voyager, :rate_limiter_config, %{})},

lib/voyager/queries/trace_entries.ex

@@ -0,0 +1,122 @@
+defmodule Voyager.Queries.TraceEntries do
+  @moduledoc """
+  Reads the log of a saved tracing session from its `Voyager.Schemas.TraceSessionEvent`
+  and `Voyager.Schemas.TraceEvent` rows, merged by their shared `seq`. Entries come back
+  in the `Voyager.Services.Tracer` entry shape, with `at` in microseconds.
+  """
+
+  import Ecto.Query
+  alias Voyager.Repo
+  alias Voyager.Schemas.TraceEvent
+  alias Voyager.Schemas.TraceSessionEvent
+  alias Voyager.Services.Tracer.Config
+  alias Voyager.StoredTerm
+
+  @doc """
+  Up to `:limit` entries with `seq` in `[:since, :before)`, `:order` (default `:desc`) by `seq`.
+  """
+  @spec page(integer(), keyword()) :: [map()]
+  def page(session_id, opts) do
+    merge(window(TraceSessionEvent, session_id, opts), window(TraceEvent, session_id, opts), opts)
+  end
+
+  @doc "Every configured, started and stopped entry, newest first."
+  @spec controls(integer()) :: [map()]
+  def controls(session_id) do
+    TraceSessionEvent
+    |> where([e], e.kind != :dropped)
+    |> window(session_id, [])
+  end
+
+  @doc "The newest `limit` trace and dropped entries, newest first."
+  @spec traces(integer(), pos_integer()) :: [map()]
+  def traces(session_id, limit) do
+    opts = [limit: limit]
+    dropped = TraceSessionEvent |> where(kind: :dropped) |> window(session_id, opts)
+    merge(dropped, window(TraceEvent, session_id, opts), opts)
+  end
+
+  @doc "The `seq` after the session's last entry."
+  @spec next_seq(integer()) :: non_neg_integer()
+  def next_seq(session_id) do
+    [TraceSessionEvent, TraceEvent]
+    |> Enum.map(&Repo.one(from e in &1, where: e.session_id == ^session_id, select: max(e.seq)))
+    |> Enum.reject(&is_nil/1)
+    |> Enum.max(fn -> -1 end)
+    |> Kernel.+(1)
+  end
+
+  defp window(query, session_id, opts) do
+    query
+    |> where([e], e.session_id == ^session_id and e.seq >= ^Keyword.get(opts, :since, 0))
+    |> where_before(opts[:before])
+    |> order_by([e], [{^order(opts), e.seq}])
+    |> limit_to(opts[:limit])
+    |> Repo.all()
+    |> Enum.map(&to_entry/1)
+  end
+
+  # Each window already holds its table's first `limit` rows, so the merged first `limit` is exact.
+  defp merge(session_events, events, opts) do
+    (session_events ++ events)
+    |> Enum.sort_by(& &1.seq, order(opts))
+    |> Enum.take(Keyword.fetch!(opts, :limit))
+  end
+
+  defp order(opts), do: Keyword.get(opts, :order, :desc)
+
+  defp where_before(query, nil), do: query
+  defp where_before(query, before), do: where(query, [e], e.seq < ^before)
+
+  defp limit_to(query, nil), do: query
+  defp limit_to(query, limit), do: limit(query, ^limit)
+
+  defp to_entry(%TraceEvent{} = row) do
+    terms =
+      for field <- [:args, :value],
+          bin = Map.fetch!(row, field),
+          into: %{},
+          do: {field, StoredTerm.decode(bin)}
+
+    row
+    |> Map.take([:truncated, :duration_us])
+    |> Map.reject(fn {_field, value} -> is_nil(value) end)
+    |> Map.merge(terms)
+    |> Map.merge(%{
+      seq: row.seq,
+      kind: :trace,
+      at: DateTime.to_unix(row.at, :microsecond),
+      type: row.type,
+      pid: StoredTerm.decode(row.pid),
+      process: StoredTerm.decode(row.process),
+      mfa: {String.to_atom(row.module), String.to_atom(row.function), row.arity}
+    })
+  end
+
+  defp to_entry(%TraceSessionEvent{} = row) do
+    Map.merge(
+      %{seq: row.seq, kind: row.kind, at: DateTime.to_unix(row.at, :microsecond)},
+      fields(row)
+    )
+  end
+
+  defp fields(%{kind: :configured} = row) do
+    %{
+      origin: row.origin,
+      node: String.to_atom(row.node),
+      config: Ecto.embedded_load(Config, row.config, :json)
+    }
+  end
+
+  defp fields(%{kind: :started} = row), do: %{origin: row.origin}
+
+  defp fields(%{kind: :stopped} = row),
+    do: %{
+      origin: row.origin,
+      reason: StoredTerm.decode(row.reason),
+      traces: row.traces,
+      dropped: row.dropped
+    }
+
+  defp fields(%{kind: :dropped} = row), do: %{origin: row.origin, count: row.dropped}
+end

lib/voyager/queries/trace_sessions.ex

@@ -0,0 +1,21 @@
+defmodule Voyager.Queries.TraceSessions do
+  @moduledoc "Read queries for tracing sessions."
+
+  import Ecto.Query
+  alias Voyager.Repo
+  alias Voyager.Schemas.TraceSession
+
+  @spec all() :: [TraceSession.t()]
+  def all do
+    TraceSession
+    |> order_by([s], desc: s.id)
+    |> Repo.all()
+  end
+
+  @spec get(integer()) :: TraceSession.t() | nil
+  def get(id), do: Repo.get(TraceSession, id)
+
+  @spec bytes_except(integer()) :: non_neg_integer()
+  def bytes_except(id),
+    do: Repo.one(from s in TraceSession, where: s.id != ^id, select: coalesce(sum(s.bytes), 0))
+end

lib/voyager/schemas/trace_event.ex

@@ -0,0 +1,46 @@
+defmodule Voyager.Schemas.TraceEvent do
+  @moduledoc """
+  Ecto schema for a traced call, return or exception of a `Voyager.Schemas.TraceSession`.
+  `pid`, `process`, `args` and `value` are `Voyager.StoredTerm.encode/1` binaries, so
+  they load back as the traced terms; `search_text` holds them and the MFA inspected, for search.
+  Shares `seq` with `Voyager.Schemas.TraceSessionEvent`.
+  """
+
+  use Ecto.Schema
+
+  @type t :: %__MODULE__{
+          __meta__: Ecto.Schema.Metadata.t(),
+          id: pos_integer() | nil,
+          session_id: pos_integer() | nil,
+          seq: non_neg_integer() | nil,
+          at: DateTime.t() | nil,
+          type: :call | :return | :exception | nil,
+          pid: binary() | nil,
+          process: binary() | nil,
+          module: String.t() | nil,
+          function: String.t() | nil,
+          arity: non_neg_integer() | nil,
+          args: binary() | nil,
+          value: binary() | nil,
+          truncated: boolean() | nil,
+          duration_us: non_neg_integer() | nil,
+          search_text: String.t() | nil
+        }
+
+  schema "trace_events" do
+    belongs_to :session, Voyager.Schemas.TraceSession
+    field :seq, :integer
+    field :at, :utc_datetime_usec
+    field :type, Ecto.Enum, values: [:call, :return, :exception]
+    field :pid, :binary
+    field :process, :binary
+    field :module, :string
+    field :function, :string
+    field :arity, :integer
+    field :args, :binary
+    field :value, :binary
+    field :truncated, :boolean
+    field :duration_us, :integer
+    field :search_text, :string
+  end
+end