Elixir Opentelemetry - Client interceptor

otel_propagator.ex

defmodule GRPC.Opentelemetry.OtelPropagator do
  @moduledoc """
  Propagates the traceparent across the network using HTTP the headers.
  """

  require OpenTelemetry.Tracer

  alias GRPC.Client.Stream

  @behaviour GRPC.ClientInterceptor

  @impl GRPC.ClientInterceptor
  def init(_options), do: %{}

  @impl GRPC.ClientInterceptor
  def call(%Stream{headers: headers} = stream, req, next, _options) do
    span_name = get_span_name(stream)

    OpenTelemetry.Tracer.with_span span_name, %{kind: :client} do
      new_headers = Map.merge(headers, build_headers())
      new_stream = GRPC.Client.Stream.put_headers(stream, new_headers)

      new_stream
      |> next.(req)
      |> set_span_attributes(stream)
    end
  end

  defp build_headers(), do: Enum.into(:otel_propagator_text_map.inject([]), %{})
  defp get_span_name(%Stream{path: path}), do: path

  defp set_span_attributes(result, stream) do
    attributes = %{
      :"rpc.system" => :grpc,
      :"rpc.service" => get_service_name(stream),
      :"rpc.method" => get_method_name(stream),
      :host => get_host(stream),
      :port => get_port(stream)
    }

    OpenTelemetry.Tracer.set_attributes(attributes)

    result
  end

  defp get_host(%Stream{channel: %GRPC.Channel{host: host}}), do: host
  defp get_port(%Stream{channel: %GRPC.Channel{port: port}}), do: port
  defp get_method_name(%Stream{method_name: method_name}), do: method_name
  defp get_service_name(%Stream{service_name: service_name}), do: service_name
end

otel_propagator_test.exs

defmodule GRPC.Opentelemetry.OtelPropagatorTest do
  @moduledoc """
  Tests inspired on https://github.com/open-telemetry/opentelemetry-erlang/blob/b8185bdb8d37783c4980925b40f86eda23e999df/test/otel_tests.exs
  """
  use ExUnit.Case

  alias GRPC.Opentelemetry.OtelPropagator

  defmodule HelloProto do
    defstruct [:message]
  end

  @stream %GRPC.Client.Stream{
    __interface__: %{
      recv: &GRPC.Stub.do_recv/2,
      send_request: &GRPC.Client.Stream.send_request/3
    },
    accepted_compressors: [],
    canceled: false,
    channel: %GRPC.Channel{
      accepted_compressors: [],
      adapter: GRPC.Client.Adapters.Gun,
      codec: GRPC.Codec.Proto,
      compressor: nil,
      cred: nil,
      headers: [],
      host: "localhost",
      interceptors: [
        {Telemetron.Instrumenter.GRPC, []},
        {GRPC.Opentelemetry.OtelPropagator, %{}}
      ],
      port: 8081,
      scheme: "http"
    },
    codec: GRPC.Codec.Proto,
    compressor: nil,
    grpc_type: :unary,
    headers: %{},
    method_name: "GetDriver",
    path: "/driver.GetDriver/GetDriver",
    payload: %{},
    request_mod: Driver.GetDriverRequest,
    response_mod: Driver.Driver,
    rpc: {"get_driver", {Driver.GetDriverRequest, false}, {Driver.Driver, false}},
    server_stream: false,
    service_name: "driver.GetDriver"
  }

  describe "call" do
    test "propagates the traceparent header properly when success" do
      :otel_batch_processor.set_exporter(:otel_exporter_pid, self())
      OpenTelemetry.get_tracer(:test_tracer, "0.1.0", :undefined)

      next = fn _stream, _req -> {:ok, %HelloProto{}} end
      OtelPropagator.call(@stream, nil, next, [])

      assert_receive {:span,
                      {:span, _, _, [], :undefined, "/driver.GetDriver/GetDriver", :client, _, _,
                       {:attributes, 128, :infinity, 0,
                        %{
                          :host => "localhost",
                          :"rpc.method" => "GetDriver",
                          :port => 8081,
                          :"rpc.system" => :grpc,
                          :"rpc.service" => "driver.GetDriver"
                        }}, _, _, _, _, _,
                       {:instrumentation_library, "pricing", "0.1.0", :undefined}}}
    end
  end
end