feat: add telemetry

lib/grpc/server.ex

@@ -182,6 +182,9 @@ defmodule GRPC.Server do
          %{server: server, endpoint: endpoint} = stream,
          req
        ) do
+    t0 = System.monotonic_time()
+    :ok = GRPC.Telemetry.server_rpc_start(server, endpoint, func_name, stream)
+
     last = fn r, s ->
       reply = apply(server, func_name, [r, s])
 
@@ -203,13 +206,50 @@ defmodule GRPC.Server do
       next.(req, stream)
     rescue
       e in GRPC.RPCError ->
+        duration = System.monotonic_time() - t0
+
+        :ok =
+          GRPC.Telemetry.server_rpc_exception(
+            server,
+            endpoint,
+            func_name,
+            stream,
+            :error,
+            e,
+            __STACKTRACE__,
+            duration
+          )
+
         {:error, e}
     catch
       kind, reason ->
-        stack = __STACKTRACE__
-        Logger.error(Exception.format(kind, reason, stack))
-        reason = Exception.normalize(kind, reason, stack)
-        {:error, %{kind: kind, reason: reason, stack: stack}}
+        stacktrace = __STACKTRACE__
+        Logger.error(Exception.format(kind, reason, stacktrace))
+        reason = Exception.normalize(kind, reason, stacktrace)
+
+        duration = System.monotonic_time() - t0
+
+        :ok =
+          GRPC.Telemetry.server_rpc_exception(
+            server,
+            endpoint,
+            func_name,
+            stream,
+            kind,
+            reason,
+            stacktrace,
+            duration
+          )
+
+        {:error, %{kind: kind, reason: reason, stack: stacktrace}}
+    else
+      result ->
+        duration = System.monotonic_time() - t0
+
+        :ok =
+          GRPC.Telemetry.server_rpc_stop(server, endpoint, func_name, stream, result, duration)
+
+        result
     end
   end
 

lib/grpc/stub.ex

@@ -265,7 +265,61 @@ defmodule GRPC.Stub do
         accepted_compressors: accepted_compressors
     }
 
-    do_call(req_stream, stream, request, opts)
+    t0 = System.monotonic_time()
+
+    try do
+      :ok = GRPC.Telemetry.client_rpc_start(stream)
+      do_call(req_stream, stream, request, opts)
+    rescue
+      e ->
+        stacktrace = __STACKTRACE__
+        Logger.error(Exception.format(:error, e, stacktrace))
+        duration = System.monotonic_time() - t0
+        normalized_reason = Exception.normalize(:error, e, stacktrace)
+
+        :ok =
+          GRPC.Telemetry.client_rpc_exception(
+            stream,
+            :error,
+            normalized_reason,
+            stacktrace,
+            duration
+          )
+
+        reraise e, __STACKTRACE__
+    catch
+      kind, reason ->
+        stacktrace = __STACKTRACE__
+        Logger.error(Exception.format(kind, reason, stacktrace))
+        duration = System.monotonic_time() - t0
+        normalized_reason = Exception.normalize(kind, reason, stacktrace)
+
+        :ok =
+          GRPC.Telemetry.client_rpc_exception(
+            stream,
+            kind,
+            normalized_reason,
+            stacktrace,
+            duration
+          )
+
+        # re-raise accordingly
+        case kind do
+          :exit ->
+            exit(reason)
+
+          :throw ->
+            throw(reason)
+
+          :error ->
+            :erlang.error(reason)
+        end
+    else
+      result ->
+        duration = System.monotonic_time() - t0
+        :ok = GRPC.Telemetry.client_rpc_stop(stream, duration)
+        result
+    end
   end
 
   defp do_call(

lib/grpc/telemetry.ex

@@ -0,0 +1,171 @@
+defmodule GRPC.Telemetry do
+  @moduledoc """
+  Events published by GRPC
+
+  These can be divided in client-side events and server-side events.
+
+  ## Client-side Events
+
+    * `[:grpc, :client, :rpc, :start]` - Published before all interceptors are executed.
+    * `[:grpc, :client, :rpc, :stop]` - Published after all interceptors executed successfully.
+      * `:duration` - the duration as measured through `System.monotonic_time()`
+        for the whole interceptor pipeline.
+    * `[:grpc, :client, :rpc, :exception]` - Published if any exception occurs while receiving a message.
+      * `:duration` - the duration as measured through `System.monotonic_time()`
+        for the execution since the start of the pipeline until the exception happened.
+
+  | event        | measurements | metadata |
+  |--------------|--------------|----------|
+  | `[:rpc, :start]`  | `:count`     | `:stream` |
+  | `[:rpc, :stop]`  | `:duration`  | `:stream` |
+  | `[:rpc, :exception]` | `:duration`  | `:stream`, `:kind`, `:reason`, `:stacktrace` |
+
+  ### Metadata
+
+    * `:stream` - the `%GRPC.Server.Stream{}` for the request
+    * `:function_name` - the name of the function called
+    * `:server` - the server module name
+    * `:endpoint` - the endpoint module name
+
+  `:exception` events also include some error metadata:
+
+    * `:reason` is the error value in case of `catch` or the actual exception in case of `rescue`.
+    * `:kind` can be one of:
+      * `:error` — from an `{:error, error}` return value. Some Erlang functions may also throw an
+        `:error` tuple, which will be reported as `:error`.
+      * `:exit` — from a caught process exit.
+      * `:throw` — from a caught value, this doesn't necessarily mean that an error occurred.
+
+  ## Server-side Events
+
+    * `[:grpc, :server, :rpc, :start]` - Published before all interceptors are executed.
+    * `[:grpc, :server, :rpc, :stop]` - Published after all interceptors executed successfully.
+      * `:duration` - the duration as measured through `System.monotonic_time()`
+        for the whole interceptor pipeline.
+    * `[:grpc, :server, :rpc, :exception]` - Published if any exception occurs while receiving a message.
+      * `:duration` - the duration as measured through `System.monotonic_time()`
+        for the execution since the start of the pipeline until the exception happened.
+
+  | event        | measurements | metadata |
+  |--------------|--------------|----------|
+  | `[:rpc, :start]`  | `:count`     | `:stream`, `:server`, `:endpoint`, `:function_name` |
+  | `[:rpc, :stop]`  | `:duration`  | `:stream`, `:server`, `:endpoint`, `:function_name` , `:result` |
+  | `[:rpc, :exception]` | `:duration`  | `:stream`, `:server`, `:endpoint`, `:function_name`, `:kind`, `:reason`, `:stacktrace` |
+
+  ### Metadata
+
+    * `:stream` - the `%GRPC.Server.Stream{}` for the request.
+    * `:function_name` - the name of the function called.
+    * `:server` - the server module name.
+    * `:endpoint` - the endpoint module name.
+    * `:result` - the result returned from the interceptor pipeline.
+
+  `:exception` events also include some error metadata:
+
+    * `:reason` is the error value in case of `catch` or the actual exception in case of `rescue`.
+    * `:kind` can be one of:
+      * `:error` — from an `{:error, error}` return value. Some Erlang functions may also throw an
+        `:error` tuple, which will be reported as `:error`.
+      * `:exit` — from a caught process exit.
+      * `:throw` — from a caught value, this doesn't necessarily mean that an error occurred.
+  """
+
+  @server_rpc [:grpc, :server, :rpc]
+  @client_rpc [:grpc, :client, :rpc]
+
+  @server_rpc_start_name @server_rpc ++ [:start]
+
+  @doc false
+  def server_rpc_start_name, do: @server_rpc_start_name
+
+  @doc false
+  def server_rpc_start(server, endpoint, func_name, stream) do
+    :telemetry.execute(@server_rpc_start_name, %{count: 1}, %{
+      server: server,
+      endpoint: endpoint,
+      function_name: func_name,
+      stream: stream
+    })
+  end
+
+  @server_rpc_stop_name @server_rpc ++ [:stop]
+
+  @doc false
+  def server_rpc_stop_name, do: @server_rpc_stop_name
+
+  @doc false
+  def server_rpc_stop(server, endpoint, func_name, stream, result, duration) do
+    :telemetry.execute(@server_rpc_stop_name, %{duration: duration}, %{
+      server: server,
+      endpoint: endpoint,
+      function_name: func_name,
+      stream: stream,
+      result: result
+    })
+  end
+
+  @server_rpc_exception_name @server_rpc ++ [:exception]
+
+  @doc false
+  def server_rpc_exception_name, do: @server_rpc_exception_name
+
+  @doc false
+  def server_rpc_exception(
+        server,
+        endpoint,
+        func_name,
+        stream,
+        kind,
+        reason,
+        stacktrace,
+        duration
+      ) do
+    :telemetry.execute(@server_rpc_exception_name, %{duration: duration}, %{
+      server: server,
+      endpoint: endpoint,
+      function_name: func_name,
+      stream: stream,
+      kind: kind,
+      reason: reason,
+      stacktrace: stacktrace
+    })
+  end
+
+  @client_rpc_start_name @client_rpc ++ [:start]
+  @doc false
+  def client_rpc_start_name, do: @client_rpc_start_name
+
+  @doc false
+  def client_rpc_start(stream) do
+    :telemetry.execute(@client_rpc_start_name, %{count: 1}, %{stream: stream})
+  end
+
+  @client_rpc_stop_name @client_rpc ++ [:stop]
+  @doc false
+  def client_rpc_stop_name, do: @client_rpc_stop_name
+
+  @doc false
+  def client_rpc_stop(stream, duration) do
+    :telemetry.execute(@client_rpc_stop_name, %{duration: duration}, %{stream: stream})
+  end
+
+  @client_rpc_exception_name @client_rpc ++ [:exception]
+  @doc false
+  def client_rpc_exception_name, do: @client_rpc_exception_name
+
+  @doc false
+  def client_rpc_exception(
+        stream,
+        kind,
+        reason,
+        stacktrace,
+        duration
+      ) do
+    :telemetry.execute(@client_rpc_exception_name, %{duration: duration}, %{
+      stream: stream,
+      kind: kind,
+      reason: reason,
+      stacktrace: stacktrace
+    })
+  end
+end