opentelemetry_honeycomb

defmodule OpenTelemetry.Honeycomb.Exporter do
  @moduledoc """
  Exporter implementation.

  `:ot_batch_processor` calls `export/3`, which:

  * fetches the attributes of the named resource
  * loads the spans from the named ETS table
  * converts the spans to Honeycomb events
  * chunks them according to Honeycomb's [batch API limits][HClimits]

  [HClimits]: https://docs.honeycomb.io/api/events/#limits
  """

  alias OpenTelemetry.Honeycomb
  alias OpenTelemetry.Honeycomb.Attributes
  alias OpenTelemetry.Honeycomb.Config
  alias OpenTelemetry.Honeycomb.Event
  alias OpenTelemetry.Honeycomb.Http
  alias OpenTelemetry.Honeycomb.Http.WriteKeyMissingBackend
  alias OpenTelemetry.Honeycomb.Json

  require OpenTelemetry.Span

  @hc_event_limit 102_400
  @hc_batch_limit 5_242_880

  @behaviour :ot_exporter
  @impl :ot_exporter
  @spec init(config :: Config.t()) :: {:ok, Config.t()} | :ignore
  def init(config) do
    config = Keyword.merge(Config.default_config(), config)

    http_module = Keyword.fetch!(config, :http_module)
    json_module = Keyword.fetch!(config, :json_module)

    cond do
      not Code.ensure_loaded?(json_module) ->
        warn("json_module #{json_module} not loaded; disabling exporter...")
        :ignore

      # Ugly work-around to ensure the integration tests get a working exporter:
      http_module == MockedHttpBackend ->
        {:ok, config}

      not Code.ensure_loaded?(http_module) ->
        warn("http_module #{http_module} not loaded; disabling exporter...")
        :ignore

      Keyword.get(config, :write_key) == nil ->
        warn("write_key absent; setting http_module=WriteKeyMissingBackend")
        {:ok, Keyword.merge(config, http_module: WriteKeyMissingBackend)}

      true ->
        {:ok, config}
    end
  end

  @impl :ot_exporter
  def shutdown(_), do: :ok

  @impl :ot_exporter
  @spec export(
          tab :: :ets.tab(),
          resource :: :ot_resource.t(),
          Config.t()
        ) :: :ok | :success | :failed_not_retryable | :failed_retryable
  def export(tab, resource, config) do
    tab
    |> :ets.tab2list()
    |> export_loaded(resource, config)
  end

  @spec export_loaded(
          spans :: [:opentelemetry.span()],
          resource :: :ot_resource.t(),
          Config.t()
        ) :: :ok | :success | :failed_not_retryable | :failed_retryable

  defp export_loaded([], _, _), do: :ok

  defp export_loaded(spans, resource, config) do
    resource_attributes = :ot_resource.attributes(resource) |> Attributes.sort()
    attribute_map = config[:attribute_map]
    cook = fn ot_span -> Event.from_ot_span(ot_span, resource_attributes, attribute_map) end

    spans
    |> Enum.flat_map(cook)
    |> Enum.map(&Json.encode_to_iodata!(config, &1))
    |> chunk_events()
    |> Enum.map(&send_batch(&1, config))
    |> Enum.find_value(:ok, &find_hc_batch_errors/1)
  end

  defp find_hc_batch_errors(:ok), do: false
  defp find_hc_batch_errors(error), do: error

  defp send_batch(batch, config) do
    url = "#{config[:api_endpoint]}/1/batch/#{config[:dataset]}"

    headers = [
      {"Content-Type", "application/json"},
      {"User-Agent", "opentelemetry_honeycomb/#{Honeycomb.version()}"},
      {"X-Honeycomb-Team", config[:write_key]}
    ]

    config
    |> Http.request(:post, url, headers, batch, [])
    |> handle_hc_reply(config)
  end

  defp handle_hc_reply({:ok, 204, _, _}, _), do: :ok

  defp handle_hc_reply({:ok, 200, _, replies_payload}, config) do
    config
    |> Json.decode!(replies_payload)
    |> Enum.find_value(:ok, &find_hc_batch_item_errors/1)
  end

  defp handle_hc_reply({:ok, 401, _, _}, _),
    do: failed_not_retryable("write_key incorrect for api_endpoint; got 401")

  defp handle_hc_reply({:ok, status, _, _}, _) when status >= 500,
    do: failed_retryable("upstream reports #{status}; failing batch")

  defp handle_hc_reply({:ok, status, _, _}, _),
    do: failed_not_retryable("upstream reports #{status}; dropping batch")

  defp handle_hc_reply(_, _), do: :failed_retryable

  defp find_hc_batch_item_errors(%{"status" => 202}), do: false
  defp find_hc_batch_item_errors(%{"status" => 400}), do: :failed_not_retryable
  defp find_hc_batch_item_errors(_), do: :failed_retryable

  defp failed_not_retryable(message) do
    warn(message)
    :failed_not_retryable
  end

  defp failed_retryable(message) do
    warn(message)
    :failed_retryable
  end

  defp chunk_events(enum) do
    case Enum.reduce(enum, {[], [], 1}, &chunk_events/2) do
      {[], [], 1} -> []
      {finished, [], _} -> Enum.reverse(finished)
      {finished, unfinished, _} -> Enum.reverse([array_finish(unfinished) | finished])
    end
  end

  defp chunk_events(iodata, {finished, unfinished, unfinished_length}) do
    case IO.iodata_length(iodata) do
      length when length > @hc_event_limit ->
        warn("event exceeds #{@hc_event_limit} bytes; dropped.")
        {finished, unfinished, unfinished_length}

      length when length + unfinished_length + 1 > @hc_batch_limit ->
        {[array_finish(unfinished) | finished], array_append([], iodata), length + 1}

      length ->
        {finished, array_append(unfinished, iodata), unfinished_length + length + 1}
    end
  end

  defp array_append([], iodata), do: [?[, iodata]
  defp array_append(acc, iodata), do: [acc, ?,, iodata]

  defp array_finish([]), do: [?[, ?]]
  defp array_finish(iodata), do: [iodata, ?]]

  defp warn(message), do: IO.warn("opentelemetry_honeycomb: #{message}", [])
end