Use open_telemetry_decorator

Author: joshk

lib/nerves_hub/deployments/orchestrator.ex

@@ -9,11 +9,10 @@ defmodule NervesHub.Deployments.Orchestrator do
   """
 
   use GenServer
+  use OpenTelemetryDecorator
 
   require Logger
 
-  require OpenTelemetry.Tracer, as: Tracer
-
   alias NervesHub.Devices
   alias NervesHub.Devices.Device
   alias NervesHub.Repo
@@ -50,60 +49,59 @@ 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
-    Tracer.with_span "NervesHub.Deployments.Orchestrator.trigger_update" do
-      :telemetry.execute([:nerves_hub, :deployment, :trigger_update], %{count: 1})
-
-      match_conditions = [
-        {:and, {:==, {:map_get, :deployment_id, :"$1"}, deployment.id},
-         {:==, {:map_get, :updating, :"$1"}, false},
-         {:==, {:map_get, :updates_enabled, :"$1"}, true},
-         {:"/=", {:map_get, :firmware_uuid, :"$1"}, deployment.firmware.uuid}}
-      ]
-
-      match_return = %{
-        device_id: {:element, 1, :"$_"},
-        pid: {:element, 1, {:element, 2, :"$_"}},
-        firmware_uuid: {:map_get, :firmware_uuid, {:element, 2, {:element, 2, :"$_"}}}
-      }
-
-      devices =
-        Registry.select(NervesHub.Devices.Registry, [
-          {{:_, :_, :"$1"}, match_conditions, [match_return]}
-        ])
-
-      # Get a rough count of devices to update
-      count = deployment.concurrent_updates - Devices.count_inflight_updates_for(deployment)
-      # Just in case inflight goes higher than concurrent, limit it to 0
-      count = max(count, 0)
-
-      # use a reduce to bounce out early?
-      # limit the number of devices to 5 minutes / 500ms?
-
-      devices
-      |> Enum.take(count)
-      |> Enum.each(fn %{device_id: device_id, pid: pid} ->
-        :telemetry.execute([:nerves_hub, :deployment, :trigger_update, :device], %{count: 1})
-
-        device = %Device{id: device_id}
-
-        # Check again because other nodes are processing at the same time
-        if Devices.count_inflight_updates_for(deployment) < deployment.concurrent_updates do
-          case Devices.told_to_update(device, deployment) do
-            {:ok, inflight_update} ->
-              send(pid, {"deployments/update", inflight_update})
-
-            :error ->
-              Logger.error(
-                "An inflight update could not be created or found for the device #{device.identifier} (#{device.id})"
-              )
-          end
+    :telemetry.execute([:nerves_hub, :deployment, :trigger_update], %{count: 1})
+
+    match_conditions = [
+      {:and, {:==, {:map_get, :deployment_id, :"$1"}, deployment.id},
+       {:==, {:map_get, :updating, :"$1"}, false},
+       {:==, {:map_get, :updates_enabled, :"$1"}, true},
+       {:"/=", {:map_get, :firmware_uuid, :"$1"}, deployment.firmware.uuid}}
+    ]
+
+    match_return = %{
+      device_id: {:element, 1, :"$_"},
+      pid: {:element, 1, {:element, 2, :"$_"}},
+      firmware_uuid: {:map_get, :firmware_uuid, {:element, 2, {:element, 2, :"$_"}}}
+    }
+
+    devices =
+      Registry.select(NervesHub.Devices.Registry, [
+        {{:_, :_, :"$1"}, match_conditions, [match_return]}
+      ])
+
+    # Get a rough count of devices to update
+    count = deployment.concurrent_updates - Devices.count_inflight_updates_for(deployment)
+    # Just in case inflight goes higher than concurrent, limit it to 0
+    count = max(count, 0)
+
+    # use a reduce to bounce out early?
+    # limit the number of devices to 5 minutes / 500ms?
+
+    devices
+    |> Enum.take(count)
+    |> Enum.each(fn %{device_id: device_id, pid: pid} ->
+      :telemetry.execute([:nerves_hub, :deployment, :trigger_update, :device], %{count: 1})
+
+      device = %Device{id: device_id}
+
+      # Check again because other nodes are processing at the same time
+      if Devices.count_inflight_updates_for(deployment) < deployment.concurrent_updates do
+        case Devices.told_to_update(device, deployment) do
+          {:ok, inflight_update} ->
+            send(pid, {"deployments/update", inflight_update})
+
+          :error ->
+            Logger.error(
+              "An inflight update could not be created or found for the device #{device.identifier} (#{device.id})"
+            )
         end
+      end
 
-        # Slow the update a bit to allow for concurrent nodes
-        Process.sleep(500)
-      end)
-    end
+      # Slow the update a bit to allow for concurrent nodes
+      Process.sleep(500)
+    end)
   end
 
   def init(deployment) do

mix.exs

@@ -107,6 +107,7 @@ defmodule NervesHub.MixProject do
        branch: "update-obans-semantic-conventions",
        subdir: "instrumentation/opentelemetry_oban"},
       {:opentelemetry_bandit, "~> 0.2.0-rc.1"},
+      {:open_telemetry_decorator, "~> 1.5"},
       {:phoenix, "~> 1.7.0"},
       {:phoenix_ecto, "~> 4.0"},
       {:phoenix_html, "~> 3.3.1", override: true},

mix.lock

@@ -52,7 +52,9 @@
   "nimble_ownership": {:hex, :nimble_ownership, "1.0.0", "3f87744d42c21b2042a0aa1d48c83c77e6dd9dd357e425a038dd4b49ba8b79a1", [:mix], [], "hexpm", "7c16cc74f4e952464220a73055b557a273e8b1b7ace8489ec9d86e9ad56cb2cc"},
   "nimble_pool": {:hex, :nimble_pool, "1.1.0", "bf9c29fbdcba3564a8b800d1eeb5a3c58f36e1e11d7b7fb2e084a643f645f06b", [:mix], [], "hexpm", "af2e4e6b34197db81f7aad230c1118eac993acc0dae6bc83bac0126d4ae0813a"},
   "nimble_strftime": {:hex, :nimble_strftime, "0.1.1", "b988184d1bd945bc139b2c27dd00a6c0774ec94f6b0b580083abd62d5d07818b", [:mix], [], "hexpm", "89e599c9b8b4d1203b7bb5c79eb51ef7c6a28fbc6228230b312f8b796310d755"},
+  "o11y": {:hex, :o11y, "0.2.6", "9386b36f9cc65ea838733b4838dea02334c4fb8721e53b904d02d50396df5cdc", [:mix], [{:decimal, "~> 2.0", [hex: :decimal, repo: "hexpm", optional: true]}, {:ecto, "~> 3.0", [hex: :ecto, repo: "hexpm", optional: true]}, {:opentelemetry_api, "~> 1.3", [hex: :opentelemetry_api, repo: "hexpm", optional: false]}], "hexpm", "ea631c5ba03b155c1bc5849804da5d375538659446e4bb9388a0427959d122ef"},
   "oban": {:hex, :oban, "2.18.3", "1608c04f8856c108555c379f2f56bc0759149d35fa9d3b825cb8a6769f8ae926", [:mix], [{:ecto_sql, "~> 3.10", [hex: :ecto_sql, repo: "hexpm", optional: false]}, {:ecto_sqlite3, "~> 0.9", [hex: :ecto_sqlite3, repo: "hexpm", optional: true]}, {:jason, "~> 1.1", [hex: :jason, repo: "hexpm", optional: false]}, {:postgrex, "~> 0.16", [hex: :postgrex, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "36ca6ca84ef6518f9c2c759ea88efd438a3c81d667ba23b02b062a0aa785475e"},
+  "open_telemetry_decorator": {:hex, :open_telemetry_decorator, "1.5.7", "d0bb1942a3ea33e8fbf92ac44e084771a7dea076a559d974bf5b8ac11f76beaa", [:mix], [{:decorator, "~> 1.4", [hex: :decorator, repo: "hexpm", optional: false]}, {:o11y, "~> 0.2", [hex: :o11y, repo: "hexpm", optional: false]}, {:opentelemetry_api, "~> 1.4", [hex: :opentelemetry_api, repo: "hexpm", optional: false]}], "hexpm", "80566f68159c727c77986c617478a63c4e51256061baf52503d58d9e2316c6c4"},
   "opentelemetry": {:hex, :opentelemetry, "1.5.0", "7dda6551edfc3050ea4b0b40c0d2570423d6372b97e9c60793263ef62c53c3c2", [:rebar3], [{:opentelemetry_api, "~> 1.4", [hex: :opentelemetry_api, repo: "hexpm", optional: false]}], "hexpm", "cdf4f51d17b592fc592b9a75f86a6f808c23044ba7cf7b9534debbcc5c23b0ee"},
   "opentelemetry_api": {:hex, :opentelemetry_api, "1.4.0", "63ca1742f92f00059298f478048dfb826f4b20d49534493d6919a0db39b6db04", [:mix, :rebar3], [], "hexpm", "3dfbbfaa2c2ed3121c5c483162836c4f9027def469c41578af5ef32589fcfc58"},
   "opentelemetry_bandit": {:hex, :opentelemetry_bandit, "0.2.0-rc.1", "3b78e4a472c20102af2d6852a04a890a8780de26feda560a63283045bc9088f5", [:mix], [{:nimble_options, "~> 1.1", [hex: :nimble_options, repo: "hexpm", optional: false]}, {:opentelemetry_api, "~> 1.3", [hex: :opentelemetry_api, repo: "hexpm", optional: false]}, {:opentelemetry_semantic_conventions, "~> 1.27", [hex: :opentelemetry_semantic_conventions, repo: "hexpm", optional: false]}, {:otel_http, "~> 0.2", [hex: :otel_http, repo: "hexpm", optional: false]}, {:plug, ">= 1.15.0", [hex: :plug, repo: "hexpm", optional: false]}, {:telemetry, "~> 1.2", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "9755c38e8b952204299f568cc30b5462851340c8b4b50702614ca403275b573a"},