Add Telemetry/OpenTelemetry support

Author: JohnDoneth

lib/upload.ex

@@ -153,7 +153,7 @@ defmodule Upload do
       transform_fn,
       opts
     )
-    |> repo.transaction()
+    |> repo.transaction(Keyword.get(opts, :transaction_opts))
     |> case do
       {:ok, multi_result} ->
         {:ok, Map.values(multi_result)}

lib/upload/logger.ex

@@ -7,7 +7,6 @@ defmodule Upload.Multi do
   alias Ecto.Association.NotLoaded
   alias Ecto.Multi
   alias Upload.Blob
-  alias Upload.Logger
   alias Upload.Storage
 
   @doc """
@@ -37,11 +36,19 @@ defmodule Upload.Multi do
   defp do_upload_blob(%Blob{path: nil} = blob, _opts), do: {:ok, blob}
 
   defp do_upload_blob(%Blob{path: path, key: key} = blob, opts) when is_binary(key) do
-    Upload.Logger.info("Uploading #{key}")
-
-    with :ok <- Storage.upload(path, key),
-         :ok <- Upload.put_access_control_list(blob, opts[:canned_acl] || :private),
-         do: {:ok, blob}
+    metadata = %{key: key, path: path}
+
+    :telemetry.span(
+      [:upload, :storage_upload],
+      metadata,
+      fn ->
+        {with(
+           :ok <- Storage.upload(path, key),
+           :ok <- Upload.put_access_control_list(blob, opts[:canned_acl] || :private),
+           do: {:ok, blob}
+         ), metadata}
+      end
+    )
   end
 
   @doc """
@@ -219,20 +226,28 @@ defmodule Upload.Multi do
   defp do_delete(nil), do: {:ok, nil}
 
   defp do_delete(%Blob{key: key} = blob) when is_binary(key) do
-    Logger.info("Removing blob #{blob.key}")
-
     with :ok <- remove_variants(blob),
-         :ok <- Storage.delete(key),
+         :ok <- storage_delete_with_telemetry(key),
          do: {:ok, blob}
   end
 
+  defp storage_delete_with_telemetry(key) do
+    metadata = %{key: key}
+
+    :telemetry.span(
+      [:upload, :storage_delete],
+      metadata,
+      fn ->
+        {Storage.delete(key), metadata}
+      end
+    )
+  end
+
   defp remove_variants(blob) do
     repo = Upload.Config.repo()
     blob = repo.preload(blob, :variants)
 
     Enum.each(blob.variants, fn variant ->
-      Logger.info("Removing variant #{variant.variant} for key #{blob.key}")
-
       {:ok, _blob_variant} = do_delete(variant)
     end)
 
@@ -373,7 +388,7 @@ defmodule Upload.Multi do
       with {:ok, blob_path} <- create_random_file(),
            :ok <- download_file(original_blob.key, blob_path),
            {:ok, variant_path} <-
-             call_transform_fn(transform_fn, blob_path, variant, format),
+             call_transform_fn(original_blob.key, transform_fn, blob_path, variant, format),
            :ok <- cleanup(blob_path),
            {:ok, blob} <- insert_variant(repo, original_blob, variant, variant_path),
            {:ok, _} <- do_upload_blob(blob, opts),
@@ -385,19 +400,28 @@ defmodule Upload.Multi do
     end)
   end
 
-  defp call_transform_fn(transform_fn, blob_path, variant, format) do
-    case :timer.tc(transform_fn, [blob_path, variant, format]) do
-      {microseconds, {:ok, path}} ->
-        Upload.Logger.info(
-          "Processed image variant '#{variant}' with format '#{format}' in #{microseconds / 1_000}ms"
-        )
-
+  defp call_transform_fn(original_blob_key, transform_fn, blob_path, variant, format) do
+    metadata = %{
+      original_blob_key: original_blob_key,
+      blob_path: blob_path,
+      variant: variant,
+      format: format
+    }
+
+    case :telemetry.span(
+           [:upload, :transform],
+           metadata,
+           fn ->
+             {transform_fn.(blob_path, variant, format), metadata}
+           end
+         ) do
+      {:ok, path} ->
         {:ok, path}
 
-      {_time, {:error, error}} ->
+      {:error, error} ->
         {:error, error}
 
-      {_time, unexpected} ->
+      unexpected ->
         raise "Expected upload transform function to return {:ok, path} or {:error, error}, got: #{inspect(unexpected)}"
     end
   end
@@ -448,9 +472,23 @@ defmodule Upload.Multi do
   end
 
   defp download_file(key, path) do
-    case Upload.Storage.download(key, path) do
-      :ok -> :ok
-      {:error, reason} -> {:error, %Upload.DownloadError{reason: reason, key: key, path: path}}
-    end
+    metadata = %{key: key, path: path}
+
+    :telemetry.span(
+      [:upload, :storage_download],
+      metadata,
+      fn ->
+        result =
+          case Upload.Storage.download(key, path) do
+            :ok ->
+              :ok
+
+            {:error, reason} ->
+              {:error, %Upload.DownloadError{reason: reason, key: key, path: path}}
+          end
+
+        {result, metadata}
+      end
+    )
   end
 end

lib/upload/multi.ex

@@ -0,0 +1,161 @@
+defmodule Upload.OpenTelemetry do
+  require Logger
+
+  @tracer_id __MODULE__
+
+  def attach do
+    :telemetry.attach_many(
+      "upload-opentelemetry",
+      [
+        [:upload, :transform, :start],
+        [:upload, :transform, :stop],
+        [:upload, :transform, :exception],
+        [:upload, :storage_upload, :start],
+        [:upload, :storage_upload, :stop],
+        [:upload, :storage_upload, :exception],
+        [:upload, :storage_download, :start],
+        [:upload, :storage_download, :stop],
+        [:upload, :storage_download, :exception],
+        [:upload, :analyze, :start],
+        [:upload, :analyze, :stop],
+        [:upload, :analyze, :exception],
+        [:upload, :stat, :start],
+        [:upload, :stat, :stop],
+        [:upload, :stat, :exception]
+      ],
+      &__MODULE__.handle_event/4,
+      %{}
+    )
+  end
+
+  def handle_event(
+        [_, _, :start] = event,
+        _measurements,
+        metadata,
+        %{}
+      ) do
+    OpentelemetryTelemetry.start_telemetry_span(
+      @tracer_id,
+      span_name(event),
+      metadata,
+      %{}
+    )
+
+    add_start_attributes(event, metadata)
+
+    :ok
+  end
+
+  def handle_event(
+        [_, _, :stop] = event,
+        _measurements,
+        metadata,
+        %{}
+      ) do
+    add_stop_attributes(event, metadata)
+
+    OpentelemetryTelemetry.set_current_telemetry_span(@tracer_id, metadata)
+    OpentelemetryTelemetry.end_telemetry_span(@tracer_id, metadata)
+
+    :ok
+  end
+
+  def handle_event(
+        [_, _, :exception],
+        %{duration: _duration},
+        %{kind: kind, reason: reason, stacktrace: stacktrace} = metadata,
+        _
+      ) do
+    ctx = OpentelemetryTelemetry.set_current_telemetry_span(@tracer_id, metadata)
+    status = OpenTelemetry.status(:error, to_string(reason))
+    exception = Exception.normalize(kind, reason, stacktrace)
+
+    OpenTelemetry.Span.record_exception(ctx, exception, stacktrace, [])
+    OpenTelemetry.Tracer.set_status(status)
+    OpentelemetryTelemetry.end_telemetry_span(@tracer_id, metadata)
+    :ok
+  end
+
+  defp span_name([:upload, :transform, :start]), do: "Upload.transform"
+  defp span_name([:upload, :storage_upload, :start]), do: "Upload.storage_upload"
+  defp span_name([:upload, :storage_download, :start]), do: "Upload.storage_download"
+  defp span_name([:upload, :storage_delete, :start]), do: "Upload.storage_delete"
+  defp span_name([:upload, :analyze, :start]), do: "Upload.analyze"
+  defp span_name([:upload, :stat, :start]), do: "Upload.stat"
+
+  defp add_start_attributes([:upload, :transform, :start], %{
+         original_blob_key: original_blob_key,
+         blob_path: blob_path,
+         variant: variant,
+         format: format
+       }) do
+    OpenTelemetry.Span.set_attributes(OpenTelemetry.Tracer.current_span_ctx(),
+      "upload.transform.original_blob_key": original_blob_key,
+      "upload.transform.blob_path": blob_path,
+      "upload.transform.variant": variant,
+      "upload.transform.format": format
+    )
+  end
+
+  defp add_start_attributes([:upload, :storage_upload, :start], %{key: key, path: path}) do
+    OpenTelemetry.Span.set_attributes(OpenTelemetry.Tracer.current_span_ctx(),
+      "upload.storage_upload.key": key,
+      "upload.storage_upload.path": path
+    )
+  end
+
+  defp add_start_attributes([:upload, :storage_download, :start], %{key: key, path: path}) do
+    OpenTelemetry.Span.set_attributes(OpenTelemetry.Tracer.current_span_ctx(),
+      "upload.storage_download.key": key,
+      "upload.storage_download.path": path
+    )
+  end
+
+  defp add_start_attributes([:upload, :storage_delete, :start], %{key: key}) do
+    OpenTelemetry.Span.set_attributes(OpenTelemetry.Tracer.current_span_ctx(),
+      "upload.storage_delete.key": key
+    )
+  end
+
+  defp add_start_attributes([:upload, :analyze, :start], %{
+         analyzer: analyzer,
+         path: path,
+         content_type: content_type
+       }) do
+    OpenTelemetry.Span.set_attributes(OpenTelemetry.Tracer.current_span_ctx(),
+      "upload.analyze.analyzer": analyzer,
+      "upload.analyze.path": path,
+      "upload.analyze.content_type": content_type
+    )
+  end
+
+  defp add_start_attributes(_, _), do: :ok
+
+  defp add_stop_attributes([:upload, :stat, :stop], %{
+         stat: %{
+           path: path,
+           filename: filename,
+           checksum: checksum,
+           byte_size: byte_size,
+           content_type: content_type,
+           metadata: metadata
+         }
+       }) do
+    OpenTelemetry.Span.set_attributes(OpenTelemetry.Tracer.current_span_ctx(),
+      "upload.stat.path": path,
+      "upload.stat.filename": filename,
+      "upload.stat.checksum": checksum,
+      "upload.stat.byte_size": byte_size,
+      "upload.stat.content_type": content_type
+    )
+
+    OpenTelemetry.Span.set_attributes(
+      OpenTelemetry.Tracer.current_span_ctx(),
+      Enum.map(metadata, fn {key, value} ->
+        {"upload.stat.metadata." <> to_string(key), to_string(value)}
+      end)
+    )
+  end
+
+  defp add_stop_attributes(_, _), do: :ok
+end

lib/upload/opentelemetry.ex

@@ -22,23 +22,33 @@ defmodule Upload.Stat do
   @doc false
   @spec stat(Path.t()) :: {:ok, t()} | {:error, error()}
   def stat(path) do
-    with {:ok, byte_size} <- get_byte_size(path),
-         {:ok, checksum} <- compute_checksum(path),
-         {:ok, detected_type} <- detect_type(path),
-         {:ok, metadata} <- analyze(path, detected_type) do
-      content_type = get_content_type(path, detected_type)
-
-      stat = %__MODULE__{
-        path: path,
-        byte_size: byte_size,
-        checksum: checksum,
-        metadata: metadata,
-        content_type: content_type,
-        filename: Path.basename(path)
-      }
-
-      {:ok, stat}
-    end
+    metadata = %{path: path}
+
+    :telemetry.span(
+      [:upload, :stat],
+      metadata,
+      fn ->
+        with {:ok, byte_size} <- get_byte_size(path),
+             {:ok, checksum} <- compute_checksum(path),
+             {:ok, detected_type} <- detect_type(path),
+             {:ok, metadata} <- analyze(path, detected_type) do
+          content_type = get_content_type(path, detected_type)
+
+          stat = %__MODULE__{
+            path: path,
+            byte_size: byte_size,
+            checksum: checksum,
+            metadata: metadata,
+            content_type: content_type,
+            filename: Path.basename(path)
+          }
+
+          {{:ok, stat}, Map.put(metadata, :stat, stat)}
+        else
+          {:error, error} -> {{:error, error}, metadata}
+        end
+      end
+    )
   end
 
   @doc false
@@ -97,11 +107,23 @@ defmodule Upload.Stat do
     analyzers = Upload.Config.analyzers()
 
     Enum.find_value(analyzers, {:ok, nil}, fn analyzer ->
-      case analyzer.stat(path, content_type) do
+      case run_analyzer_with_telemetry(analyzer, path, content_type) do
         {:ok, nil} -> nil
         {:ok, metadata} -> {:ok, metadata}
         {:error, reason} when is_struct(reason) -> {:error, reason}
       end
     end)
   end
+
+  defp run_analyzer_with_telemetry(analyzer, path, content_type) do
+    metadata = %{analyzer: analyzer, path: path, content_type: content_type}
+
+    :telemetry.span(
+      [:upload, :analyze],
+      metadata,
+      fn ->
+        {analyzer.stat(path, content_type), metadata}
+      end
+    )
+  end
 end