defmodule Kvasir.AgentServer.SystemMonitor do alias Kvasir.AgentServer.Metrics use GenServer @interval 1 @refresh 60_000 * @interval @agent_data ~w(id agent partition)a @sup Kvasir.AgentServer.Client.ConnectionManager require Logger @spec register(String.t() | charlist, Kvasir.AgentServer.agent()) :: :ok def register(path, agent) do if pid = Process.whereis(__MODULE__), do: send(pid, {:register_agent, path, Map.take(agent, @agent_data)}) :ok end @spec start :: DynamicSupervisor.on_start_child() def start do Logger.info("Kvasir AgentServer: Starting system monitor.") Application.put_all_env( [ os_mon: [ disk_space_check_interval: @interval, disk_almost_full_threshold: 0.85, start_cpu_sup: false, start_disksup: true, start_memsup: false, start_os_sup: false ] ], persistent: true ) :application.ensure_all_started(:os_mon) DynamicSupervisor.start_child(@sup, __MODULE__) end def start_link(opts \\ []) do GenServer.start_link(__MODULE__, opts, name: __MODULE__) end @impl GenServer def init(_) do Logger.info("Kvasir AgentServer: Initializing system monitor.") send(self(), :check_disks) send(self(), :agent_counts) paths = :disksup.get_disk_data() |> Enum.map(&elem(&1, 0)) |> Enum.sort_by(&(-Enum.count(&1))) Logger.info(fn -> "Kvasir AgentServer: System Monitor Detected Disks: #{inspect(:disksup.get_disk_data())}" end) metrics = system_metrics(System.get_env("STATSD_URL")) {:ok, %{agents: %{}, paths: paths, metrics: metrics}} end @impl GenServer def handle_info(:check_disks, state = %{agents: agents, metrics: metrics}) do Enum.map(:disksup.get_disk_data(), fn {path, _size, used} -> if a = agents[path], do: Enum.each(a, &metrics.("disks", &1, used, ["disk:#{path}"])) end) Process.send_after(self(), :check_disks, @refresh) {:noreply, state} end def handle_info(:agent_counts, state = %{agents: agents, metrics: metrics}) do agents |> Map.values() |> List.flatten() |> Enum.each(&agent_metrics(&1, metrics)) Process.send_after(self(), :agent_counts, @refresh) {:noreply, state} end def handle_info({:register_agent, path, agent}, state = %{agents: agents, paths: paths}) do full = Path.expand(path) index = Enum.find(paths, &String.starts_with?(full, to_string(&1))) updated = Map.update(agents, index, [agent], &[agent | &1]) Logger.info("Kvasir AgentServer: Add #{inspect(agent)} to #{inspect(index)} monitor.") {:noreply, %{state | agents: updated}} end def handle_info({:inet_reply, _, _}, state), do: {:noreply, state} @impl GenServer def terminate(reason, _reason) do Logger.info("Kvasir AgentServer: Shutting down system monitor. #{inspect(reason)}") :normal end @spec agent_metrics(map, fun) :: :ok defp agent_metrics(agent, metrics) defp agent_metrics(agent = %{agent: a}, metrics) do metrics.("agents", agent, a.count(), []) rescue _ -> :ok end @spec system_metrics(String.t() | nil) :: fun() defp system_metrics(url) defp system_metrics(nil), do: fn _, _, _ -> :ok end defp system_metrics(url) do {:ok, socket, header} = Metrics.open(url) fqdn = :net_adm.localhost() |> :net_adm.dns_hostname() |> elem(1) |> to_string() |> String.trim() |> String.downcase() tags = [ "host:#{fqdn}", System.get_env("STATSD_TAGS") ] |> Enum.reject(&is_nil/1) |> Enum.join(",") static = "kvasir.agent_server." prefix = IO.iodata_to_binary([header, static]) midfix = "|g|##{tags}," fn metric, %{agent: agent, id: id, partition: partition}, value, tags -> add = tags |> Kernel.++([ "agent:#{String.downcase(inspect(agent))}", "topic:#{id}", unless partition == "*" do m = partition |> String.split(" ") |> List.last() |> String.trim("\"") "partition:#{m}" end ]) |> Enum.reject(&is_nil/1) |> Enum.join(",") Port.command(socket, [prefix, metric, ?:, to_string(value), midfix, add]) end end end