Merge pull request #32 from amplifiedai/telemetry

Add telemetry support

Author: stueccles

README.md

@@ -168,3 +168,28 @@ There are not many tests at the moment. if you want to run live tests on your ac
 ```
 SEGMENT_KEY=yourkey mix test
 ```
+
+## Telemetry
+
+This package wraps its Segment event sending in [`:telemetry.span/3`][telemetry-span-3]. You can attach to:
+
+- `[:segment, :send, :start]`
+- `[:segment, :send, :stop]`
+- `[:segment, :send, :exception]`
+- `[:segment, :batch, :start]`
+- `[:segment, :batch, :stop]`
+- `[:segment, :batch, :exception]`
+
+The measurements will include, in Erlang's `:native` time unit (likely `:nanosecond`):
+
+- `system_time` with `:start` events
+- `duration` with `:stop` and `:exception` events
+
+The metadata will include:
+
+- the original `event` or `events` with all `:send` and `:batch` events respectively
+- our `status` (`:ok` | `:error`) and Tesla's `result` with all `:stop` events
+- `error` matching `result` when it isn't `{:ok, env}`
+- `kind`, `reason`, and `stacktrace` with `:exception` events
+
+[telemetry-span-3]: https://hexdocs.pm/telemetry/telemetry.html#span-3

lib/segment/client/http.ex

@@ -85,7 +85,22 @@ defmodule Segment.Http do
   """
   @spec send(client(), Segment.segment_event()) :: :ok | :error
   def send(client, event) do
-    case make_request(client, event.type, prepare_events(event), Segment.Config.retry_attempts()) do
+    :telemetry.span([:segment, :send], %{event: event}, fn ->
+      tesla_result =
+        make_request(client, event.type, prepare_events(event), Segment.Config.retry_attempts())
+
+      case process_send_post_result(tesla_result) do
+        :ok ->
+          {:ok, %{event: event, status: :ok, result: tesla_result}}
+
+        :error ->
+          {:error, %{event: event, status: :error, error: tesla_result, result: tesla_result}}
+      end
+    end)
+  end
+
+  defp process_send_post_result(tesla_result) do
+    case tesla_result do
       {:ok, %{status: status}} when status == 200 ->
         :ok
 
@@ -114,12 +129,26 @@ defmodule Segment.Http do
   """
   @spec batch(client(), list(Segment.segment_event()), map() | nil, map() | nil) :: :ok | :error
   def batch(client, events, context \\ nil, integrations \\ nil) do
-    data =
-      %{batch: prepare_events(events)}
-      |> add_if(:context, context)
-      |> add_if(:integrations, integrations)
+    :telemetry.span([:segment, :batch], %{events: events}, fn ->
+      data =
+        %{batch: prepare_events(events)}
+        |> add_if(:context, context)
+        |> add_if(:integrations, integrations)
+
+      tesla_result = make_request(client, "batch", data, Segment.Config.retry_attempts())
+
+      case process_batch_post_result(tesla_result, events) do
+        :ok ->
+          {:ok, %{events: events, status: :ok, result: tesla_result}}
+
+        :error ->
+          {:error, %{events: events, status: :error, error: tesla_result, result: tesla_result}}
+      end
+    end)
+  end
 
-    case make_request(client, "batch", data, Segment.Config.retry_attempts()) do
+  defp process_batch_post_result(tesla_result, events) do
+    case tesla_result do
       {:ok, %{status: status}} when status == 200 ->
         :ok
 

mix.exs

@@ -18,7 +18,7 @@ defmodule AnalyticsElixir.Mixfile do
   end
 
   def application do
-    [applications: [:hackney, :logger, :retry, :tesla, :jason]]
+    [applications: [:hackney, :logger, :retry, :tesla, :jason, :telemetry]]
   end
 
   defp deps do
@@ -29,6 +29,7 @@ defmodule AnalyticsElixir.Mixfile do
       {:jason, ">= 1.0.0"},
       {:mox, "~> 0.5", only: :test},
       {:retry, "~> 0.13"},
+      {:telemetry, "~> 0.4.2"},
       {:tesla, "~> 1.2"}
     ]
   end

mix.lock

@@ -19,6 +19,7 @@
   "poison": {:hex, :poison, "1.5.2", "560bdfb7449e3ddd23a096929fb9fc2122f709bcc758b2d5d5a5c7d0ea848910", [:mix], []},
   "retry": {:hex, :retry, "0.14.0", "751c0f6db0b5127acf346ea6f6c363ec4588320db785c62aa51776b4d280bf07", [:mix], [], "hexpm", "5c158bccf5e4de2a13d044b9930ceb7e7499c29e3fccd7c96f131a6b83a1cff3"},
   "ssl_verify_fun": {:hex, :ssl_verify_fun, "1.1.5", "6eaf7ad16cb568bb01753dbbd7a95ff8b91c7979482b95f38443fe2c8852a79b", [:make, :mix, :rebar3], [], "hexpm", "13104d7897e38ed7f044c4de953a6c28597d1c952075eb2e328bc6d6f2bfc496"},
+  "telemetry": {:hex, :telemetry, "0.4.2", "2808c992455e08d6177322f14d3bdb6b625fbcfd233a73505870d8738a2f4599", [:rebar3], [], "hexpm", "2d1419bd9dda6a206d7b5852179511722e2b18812310d304620c7bd92a13fcef"},
   "tesla": {:hex, :tesla, "1.3.3", "26ae98627af5c406584aa6755ab5fc96315d70d69a24dd7f8369cfcb75094a45", [:mix], [{:castore, "~> 0.1", [hex: :castore, repo: "hexpm", optional: true]}, {:exjsx, ">= 3.0.0", [hex: :exjsx, repo: "hexpm", optional: true]}, {:fuse, "~> 2.4", [hex: :fuse, repo: "hexpm", optional: true]}, {:gun, "~> 1.3", [hex: :gun, repo: "hexpm", optional: true]}, {:hackney, "~> 1.6", [hex: :hackney, repo: "hexpm", optional: true]}, {:ibrowse, "~> 4.4.0", [hex: :ibrowse, repo: "hexpm", optional: true]}, {:jason, ">= 1.0.0", [hex: :jason, repo: "hexpm", optional: true]}, {:mime, "~> 1.0", [hex: :mime, repo: "hexpm", optional: false]}, {:mint, "~> 1.0", [hex: :mint, repo: "hexpm", optional: true]}, {:poison, ">= 1.0.0", [hex: :poison, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4", [hex: :telemetry, repo: "hexpm", optional: true]}], "hexpm", "2648f1c276102f9250299e0b7b57f3071c67827349d9173f34c281756a1b124c"},
   "unicode_util_compat": {:hex, :unicode_util_compat, "0.4.1", "d869e4c68901dd9531385bb0c8c40444ebf624e60b6962d95952775cac5e90cd", [:rebar3], [], "hexpm", "1d1848c40487cdb0b30e8ed975e34e025860c02e419cb615d255849f3427439d"},
 }

test/segment_test.exs

@@ -2,6 +2,15 @@ defmodule SegmentTest do
   use ExUnit.Case
 
   test "track debugging" do
+    test_began = :erlang.system_time()
+
+    :telemetry.attach_many(
+      self(),
+      [[:segment, :batch, :start], [:segment, :batch, :stop]],
+      fn n, m10s, m6a, pid -> send(pid, {:telemetry, n, m10s, m6a}) end,
+      self()
+    )
+
     Segment.start_link(System.get_env("SEGMENT_KEY"))
 
     Segment.Analytics.track("user1", "track debugging #{elem(:os.timestamp(), 2)}")
@@ -31,6 +40,20 @@ defmodule SegmentTest do
     Segment.Analytics.page("user1", "page debugging #{elem(:os.timestamp(), 2)}")
 
     Segment.Analytics.Batcher.flush()
+
+    test_ended = :erlang.system_time()
+
+    assert_received {:telemetry, [:segment, :batch, :start], %{system_time: system_time},
+                     %{events: events}}
+
+    assert_received {:telemetry, [:segment, :batch, :stop], %{duration: duration},
+                     %{events: ^events, status: :ok, result: {:ok, env}}}
+
+    assert system_time > test_began
+    assert system_time <= test_ended
+    assert length(events) == 6
+    assert duration <= test_ended - test_began
+    assert env.status == 200
   end
 
   defp wait_random(n \\ 1000), do: Process.sleep(:rand.uniform(n))