Add support for OpenTelemetry tracing (#1612)

* Add support for OpenTelemetry tracing

* Use `open_telemetry_decorator`

* Disable tracing exporter during testing

* Add a custom OTel filtered sampler

Author: joshk

config/runtime.exs

@@ -387,6 +387,27 @@ config :sentry,
     ]
   ]
 
+config :opentelemetry, :resource, service: %{name: nerves_hub_app}
+
+if otlp_endpoint = System.get_env("OTLP_ENDPOINT") do
+  config :opentelemetry_exporter,
+    otlp_protocol: :http_protobuf,
+    otlp_endpoint: otlp_endpoint,
+    otlp_headers: [{System.get_env("OTLP_AUTH_HEADER"), System.get_env("OTLP_AUTH_HEADER_VALUE")}]
+
+  otlp_sampler_ratio =
+    if ratio = System.get_env("OTLP_SAMPLER_RATIO") do
+      String.to_float(ratio)
+    else
+      nil
+    end
+
+  config :opentelemetry,
+    sampler: {:parent_based, %{root: {NervesHub.Telemetry.FilteredSampler, otlp_sampler_ratio}}}
+else
+  config :opentelemetry, traces_exporter: :none
+end
+
 if host = System.get_env("STATSD_HOST") do
   config :nerves_hub, :statsd,
     host: System.get_env("STATSD_HOST"),

lib/nerves_hub/application.ex

@@ -12,6 +12,8 @@ defmodule NervesHub.Application do
         raise "fwup could not be found in the $PATH. This is a requirement of NervesHubWeb and cannot start otherwise"
     end
 
+    setup_open_telemetry()
+
     _ =
       :logger.add_handler(:my_sentry_handler, Sentry.LoggerHandler, %{
         config: %{metadata: [:file, :line]}
@@ -47,6 +49,25 @@ defmodule NervesHub.Application do
     Supervisor.start_link(children, opts)
   end
 
+  defp setup_open_telemetry() do
+    if System.get_env("ECTO_IPV6") do
+      :httpc.set_option(:ipfamily, :inet6fb4)
+    end
+
+    :ok = NervesHub.Telemetry.Customizations.setup()
+
+    :ok = OpentelemetryBandit.setup()
+    :ok = OpentelemetryPhoenix.setup(adapter: :bandit)
+    :ok = OpentelemetryOban.setup(trace: [:jobs])
+
+    :ok =
+      NervesHub.Repo.config()
+      |> Keyword.fetch!(:telemetry_prefix)
+      |> OpentelemetryEcto.setup(db_statement: :enabled)
+
+    :ok
+  end
+
   def config_change(changed, _new, removed) do
     NervesHubWeb.Endpoint.config_change(changed, removed)
     :ok

lib/nerves_hub/deployments/orchestrator.ex

@@ -9,6 +9,7 @@ defmodule NervesHub.Deployments.Orchestrator do
   """
 
   use GenServer
+  use OpenTelemetryDecorator
 
   require Logger
 
@@ -48,6 +49,7 @@ defmodule NervesHub.Deployments.Orchestrator do
   As devices update and reconnect, the new orchestrator is told that the update
   was successful, and the process is repeated.
   """
+  @decorate with_span("Deployments.Orchestrator.trigger_update")
   def trigger_update(deployment) do
     :telemetry.execute([:nerves_hub, :deployment, :trigger_update], %{count: 1})
 
@@ -106,6 +108,7 @@ defmodule NervesHub.Deployments.Orchestrator do
     {:ok, deployment, {:continue, :boot}}
   end
 
+  @decorate with_span("Deployments.Orchestrator.boot")
   def handle_continue(:boot, deployment) do
     _ = PubSub.subscribe(NervesHub.PubSub, "deployment:#{deployment.id}")
 
@@ -126,6 +129,7 @@ defmodule NervesHub.Deployments.Orchestrator do
     {:noreply, deployment}
   end
 
+  @decorate with_span("Deployments.Orchestrator.handle_info:deployments/update")
   def handle_info(%Broadcast{event: "deployments/update"}, deployment) do
     deployment =
       deployment

lib/nerves_hub/telemetry/customizations.ex

@@ -0,0 +1,23 @@
+defmodule NervesHub.Telemetry.Customizations do
+  alias OpenTelemetry.Tracer
+  require OpenTelemetry.Tracer
+
+  def setup() do
+    :telemetry.attach_many(
+      {__MODULE__, :bandit_customizations},
+      [
+        [:bandit, :request, :stop]
+      ],
+      &__MODULE__.handle_request/4,
+      nil
+    )
+  end
+
+  def handle_request([:bandit, :request, :stop], _measurements, %{conn: conn}, _config) do
+    if conn.request_path =~ ~r/\/websocket$/ do
+      Tracer.update_name("WEBSOCKET #{conn.request_path}")
+    end
+
+    :ok
+  end
+end

lib/nerves_hub/telemetry/filtered_sampler.ex

@@ -0,0 +1,88 @@
+defmodule NervesHub.Telemetry.FilteredSampler do
+  # Inspired by https://arathunku.com/b/2024/notes-on-adding-opentelemetry-to-an-elixir-app/
+
+  # TODO: Add ratio sampling support
+
+  require OpenTelemetry.Tracer, as: Tracer
+  require Logger
+
+  @behaviour :otel_sampler
+
+  @ignored_static_paths ~r/^\/(assets|fonts|images|css)\/.*/
+
+  @ignored_url_paths [
+    "/status/alive",
+    "/phoenix/live_reload/socket/websocket",
+    "/live/websocket",
+    "/favicon.ico",
+    "/"
+  ]
+
+  @ignored_span_names [
+    "Channels.DeviceSocket.heartbeat",
+    "nerves_hub.repo.query:schema_migrations"
+  ]
+
+  @impl :otel_sampler
+  def setup(probability \\ nil) do
+    if probability do
+      [ratio_sampler_config: :otel_sampler_trace_id_ratio_based.setup(probability)]
+    else
+      []
+    end
+  end
+
+  @impl :otel_sampler
+  def description(_sampler_config), do: "NervesHub.Sampler"
+
+  @impl :otel_sampler
+  def should_sample(
+        ctx,
+        trace_id,
+        links,
+        span_name,
+        span_kind,
+        attributes,
+        sampler_config
+      ) do
+    result = drop_trace?(span_name, attributes)
+
+    tracestate = Tracer.current_span_ctx(ctx) |> OpenTelemetry.Span.tracestate()
+
+    case result do
+      true ->
+        {:drop, [], tracestate}
+
+      false ->
+        if config = sampler_config[:ratio_sampler_config] do
+          :otel_sampler_trace_id_ratio_based.should_sample(
+            ctx,
+            trace_id,
+            links,
+            span_name,
+            span_kind,
+            attributes,
+            config
+          )
+        else
+          {:record_and_sample, [], tracestate}
+        end
+    end
+  end
+
+  def drop_trace?(span_name, attributes) do
+    cond do
+      Enum.member?(@ignored_span_names, span_name) ->
+        true
+
+      span_name == :GET && Enum.member?(@ignored_url_paths, attributes[:"url.path"]) ->
+        true
+
+      span_name == :GET && (attributes[:"url.path"] || "") =~ @ignored_static_paths ->
+        true
+
+      true ->
+        false
+    end
+  end
+end

lib/nerves_hub_web/channels/device_channel.ex

@@ -6,6 +6,7 @@ defmodule NervesHubWeb.DeviceChannel do
   """
 
   use Phoenix.Channel
+  use OpenTelemetryDecorator
 
   require Logger
 
@@ -19,6 +20,7 @@ defmodule NervesHubWeb.DeviceChannel do
   alias NervesHub.Repo
   alias Phoenix.Socket.Broadcast
 
+  @decorate with_span("Channels.DeviceChannel.join")
   def join("device", params, %{assigns: %{device: device}} = socket) do
     with {:ok, device} <- update_metadata(device, params) do
       send(self(), {:after_join, params})
@@ -31,6 +33,7 @@ defmodule NervesHubWeb.DeviceChannel do
     end
   end
 
+  @decorate with_span("Channels.DeviceChannel.handle_info:after_join")
   def handle_info({:after_join, params}, %{assigns: %{device: device}} = socket) do
     device = maybe_update_deployment(device)
 
@@ -77,6 +80,7 @@ defmodule NervesHubWeb.DeviceChannel do
     {:stop, :shutdown, socket}
   end
 
+  @decorate with_span("Channels.DeviceChannel.handle_info:device_registration")
   def handle_info({:device_registation, attempt}, socket) do
     %{assigns: %{device: device}} = socket
 
@@ -98,6 +102,7 @@ defmodule NervesHubWeb.DeviceChannel do
 
   # We can save a fairly expensive query by checking the incoming deployment's payload
   # If it matches, we can set the deployment directly and only do 3 queries (update, two preloads)
+  @decorate with_span("Channels.DeviceChannel.handle_info:deployments/changed,deployment:none")
   def handle_info(
         %Broadcast{event: "deployments/changed", topic: "deployment:none", payload: payload},
         %{assigns: %{device: device}} = socket
@@ -121,6 +126,7 @@ defmodule NervesHubWeb.DeviceChannel do
     {:noreply, assign_deployment(socket, payload)}
   end
 
+  @decorate with_span("Channels.DeviceChannel.handle_info:deployments/changed")
   def handle_info(
         %Broadcast{event: "deployments/changed", payload: payload},
         %{assigns: %{device: device}} = socket
@@ -138,6 +144,7 @@ defmodule NervesHubWeb.DeviceChannel do
     end
   end
 
+  @decorate with_span("Channels.DeviceChannel.handle_info:resolve_changed_deployment")
   def handle_info(:resolve_changed_deployment, %{assigns: %{device: device}} = socket) do
     :telemetry.execute([:nerves_hub, :devices, :deployment, :changed], %{count: 1})
 
@@ -171,6 +178,7 @@ defmodule NervesHubWeb.DeviceChannel do
     {:noreply, socket}
   end
 
+  @decorate with_span("Channels.DeviceChannel.handle_info:deployments/update")
   def handle_info({"deployments/update", inflight_update}, %{assigns: %{device: device}} = socket) do
     device = deployment_preload(device)
 
@@ -215,6 +223,7 @@ defmodule NervesHubWeb.DeviceChannel do
   end
 
   # Update local state and tell the various servers of the new information
+  @decorate with_span("Channels.DeviceChannel.handle_info:devices-updated")
   def handle_info(%Broadcast{event: "devices/updated"}, %{assigns: %{device: device}} = socket) do
     device = Repo.reload(device)
 
@@ -355,6 +364,7 @@ defmodule NervesHubWeb.DeviceChannel do
     end
   end
 
+  @decorate with_span("Channels.DeviceChannel.handle_in:location:update")
   def handle_in("location:update", location, %{assigns: %{device: device}} = socket) do
     metadata = Map.put(device.connection_metadata, "location", location)
 
@@ -405,6 +415,7 @@ defmodule NervesHubWeb.DeviceChannel do
     {:noreply, socket}
   end
 
+  @decorate with_span("Channels.DeviceChannel.handle_in:health_check_report")
   def handle_in("health_check_report", %{"value" => device_status}, socket) do
     device_meta =
       for {key, val} <- Map.from_struct(socket.assigns.device.firmware_metadata),
@@ -474,6 +485,7 @@ defmodule NervesHubWeb.DeviceChannel do
     :ok
   end
 
+  @decorate with_span("Channels.DeviceChannel.maybe_update_deployment")
   defp maybe_update_deployment(device) do
     device
     |> Deployments.preload_with_firmware_and_archive()

lib/nerves_hub_web/channels/device_socket.ex

@@ -1,5 +1,6 @@
 defmodule NervesHubWeb.DeviceSocket do
   use Phoenix.Socket
+  use OpenTelemetryDecorator
 
   require Logger
 
@@ -28,6 +29,7 @@ defmodule NervesHubWeb.DeviceSocket do
   end
 
   @impl Phoenix.Socket.Transport
+  @decorate with_span("Channels.DeviceSocket.terminate")
   def terminate(reason, {_channels_info, socket} = state) do
     on_disconnect(reason, socket)
     super(reason, state)
@@ -42,6 +44,7 @@ defmodule NervesHubWeb.DeviceSocket do
     super(msg, {state, socket})
   end
 
+  @decorate with_span("Channels.DeviceSocket.heartbeat")
   defp heartbeat(
          %Phoenix.Socket.Message{topic: "phoenix", event: "heartbeat"},
          %{
@@ -84,6 +87,7 @@ defmodule NervesHubWeb.DeviceSocket do
 
   # Used by Devices connecting with SSL certificates
   @impl Phoenix.Socket
+  @decorate with_span("Channels.DeviceSocket.connect")
   def connect(_params, socket, %{peer_data: %{ssl_cert: ssl_cert}})
       when not is_nil(ssl_cert) do
     X509.Certificate.from_der!(ssl_cert)
@@ -103,6 +107,7 @@ defmodule NervesHubWeb.DeviceSocket do
   end
 
   # Used by Devices connecting with HMAC Shared Secrets
+  @decorate with_span("Channels.DeviceSocket.connect")
   def connect(_params, socket, %{x_headers: x_headers})
       when is_list(x_headers) and length(x_headers) > 0 do
     headers = Map.new(x_headers)
@@ -210,12 +215,14 @@ defmodule NervesHubWeb.DeviceSocket do
     {:ok, socket}
   end
 
+  @decorate with_span("Channels.DeviceSocket.on_connect#registered")
   defp on_connect(%{assigns: %{device: %{status: :registered} = device}} = socket) do
     socket
     |> assign(device: Devices.set_as_provisioned!(device))
     |> on_connect()
   end
 
+  @decorate with_span("Channels.DeviceSocket.on_connect#provisioned")
   defp on_connect(%{assigns: %{device: device}} = socket) do
     # Report connection and use connection id as reference
     {:ok, %DeviceConnection{id: connection_id}} =
@@ -235,6 +242,9 @@ defmodule NervesHubWeb.DeviceSocket do
     |> assign(:reference_id, connection_id)
   end
 
+  @decorate with_span("Channels.DeviceSocket.on_disconnect")
+  defp on_disconnect(exit_reason, socket)
+
   defp on_disconnect({:error, reason}, %{
          assigns: %{
            device: device,
@@ -262,6 +272,7 @@ defmodule NervesHubWeb.DeviceSocket do
     shutdown(device, reference_id)
   end
 
+  @decorate with_span("Channels.DeviceSocket.shutdown")
   defp shutdown(device, reference_id) do
     :telemetry.execute([:nerves_hub, :devices, :disconnect], %{count: 1}, %{
       ref_id: reference_id,