kvasir_agent_server

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