From e65a4114497cb81e9005fb647bcb89fdfbf2bfc7 Mon Sep 17 00:00:00 2001
From: Blake Kostner <git@btkostner.io>
Date: Tue, 21 Jun 2022 20:18:35 -0600
Subject: [PATCH] feat: setup telemetry events

---
 README.md                   | 50 +++++++++++++++++++++++++++++++++++++
 lib/elsa/consumer/worker.ex | 15 ++++++++++-
 mix.exs                     |  5 ++--
 mix.lock                    |  1 +
 4 files changed, 68 insertions(+), 3 deletions(-)

diff --git a/README.md b/README.md
index d80e06b..44166ba 100644
--- a/README.md
+++ b/README.md
@@ -74,6 +74,56 @@ within your Elsa configuration.
 
 You can find an example of configuring and using Elsa [here](https://github.com/jdenen/let_it_go).
 
+## Telemetry
+
+Starting with Elsa `1.0.0-rc.4` [`:telemetry`](https://github.com/beam-telemetry/telemetry) events are available.
+
+- `[:elsa, :consumer, :handle_messages, :start]` - dispatched by `Elsa.Consumer.Worker` before handling a list of messages.
+
+    - Measurement: `%{system_time: integer}`
+    - Metadata:
+      ```elixir
+      %{
+        topic: Elsa.topic(),
+        partition: Elsa.partition(),
+        generation_id: non_neg_integer,
+        offset: non_neg_integer,
+        messages: [Elsa.Message()]
+      }
+      ```
+
+- `[:elsa, :consumer, :handle_messages, :end]` - dispatched by `Elsa.Consumer.Worker` after handling a list of messages.
+
+    - Measurement: `%{duration: native_time}`
+    - Metadata:
+      ```elixir
+      %{
+        topic: Elsa.topic(),
+        partition: Elsa.partition(),
+        generation_id: non_neg_integer,
+        offset: non_neg_integer,
+        messages: [Elsa.Message()],
+        result: any
+      }
+      ```
+
+- `[:elsa, :consumer, :handle_messages, :exception]` - dispatched by `Elsa.Consumer.Worker` if an exception occures processing messages.
+
+    - Measurement: `%{duration: native_time}`
+    - Metadata:
+      ```elixir
+      %{
+        topic: Elsa.topic(),
+        partition: Elsa.partition(),
+        generation_id: non_neg_integer,
+        offset: non_neg_integer,
+        messages: [Elsa.Message()],
+        kind: kind,
+        reason: reason,
+        stacktrace: stacktrace,
+      }
+      ```
+
 ## Testing
 
 Elsa uses the standard ExUnit testing library for unit testing. For integration testing interactions with Kafka, it uses the [`divo`](https://github.com/smartcitiesdata/divo) library. Run tests with the command `mix test.integration`.
diff --git a/lib/elsa/consumer/worker.ex b/lib/elsa/consumer/worker.ex
index 504b986..37d43e1 100644
--- a/lib/elsa/consumer/worker.ex
+++ b/lib/elsa/consumer/worker.ex
@@ -143,7 +143,20 @@ defmodule Elsa.Consumer.Worker do
   end
 
   defp send_messages_to_handler(messages, state) do
-    state.handler.handle_messages(messages, state.handler_state)
+    :telemetry.span(
+      [:elsa, :consumer, :handle_messages],
+      %{
+        topic: state.topic,
+        partition: state.partition,
+        generation_id: state.generation_id,
+        offset: state.offset,
+        messages: messages
+      },
+      fn ->
+        result = state.handler.handle_messages(messages, state.handler_state)
+        {result, %{result: result}}
+      end
+    )
   end
 
   defp ack_messages(_topic, _partition, offset, %{generation_id: nil} = state) do
diff --git a/mix.exs b/mix.exs
index bf1fafa..a5d3835 100644
--- a/mix.exs
+++ b/mix.exs
@@ -1,7 +1,7 @@
 defmodule Elsa.MixProject do
   use Mix.Project
 
-  @version "1.0.0-rc.3"
+  @version "1.0.0-rc.4"
   @github "https://github.com/bbalser/elsa"
 
   def project do
@@ -37,7 +37,8 @@ defmodule Elsa.MixProject do
       {:placebo, "~> 2.0", only: [:dev, :test]},
       {:checkov, "~> 1.0", only: [:test, :integration]},
       {:ex_doc, "~> 0.25.3", only: [:dev]},
-      {:dialyxir, "~> 1.1.0", only: [:dev], runtime: false}
+      {:dialyxir, "~> 1.1.0", only: [:dev], runtime: false},
+      {:telemetry, "~> 1.0"}
     ]
   end
 
diff --git a/mix.lock b/mix.lock
index 60879ee..2c9c61e 100644
--- a/mix.lock
+++ b/mix.lock
@@ -19,4 +19,5 @@
   "placebo": {:hex, :placebo, "2.0.0", "c0e773dec77e941bcbcc14d10b759f2d66775aff9b75051f3e41939b64300e81", [:mix], [{:meck, "~> 0.9", [hex: :meck, repo: "hexpm", optional: false]}], "hexpm", "e0872cec8848d7e59ba96396f45ee1ad34662c689c86ba6190694d38b4289844"},
   "snappyer": {:hex, :snappyer, "1.2.8", "201ce9067a33c71a6a5087c0c3a49a010b17112d461e6df696c722dcb6d0934a", [:rebar3], [], "hexpm", "35518e79a28548b56d8fd6aee2f565f12f51c2d3d053f9cfa817c83be88c4f3d"},
   "supervisor3": {:hex, :supervisor3, "1.1.11", "d81cdec31d102fde407423e1d05b569572850deebed86b951d5233c387cba80b", [:rebar3], [], "hexpm", "e6c2dedbcabcba24995a218aca12db5e208b80d3252692b22ef0f1a266104b50"},
+  "telemetry": {:hex, :telemetry, "1.1.0", "a589817034a27eab11144ad24d5c0f9fab1f58173274b1e9bae7074af9cbee51", [:rebar3], [], "hexpm", "b727b2a1f75614774cff2d7565b64d0dfa5bd52ba517f16543e6fc7efcc0df48"},
 }