From d79d69672b7e366be97f89887507ab06038f9449 Mon Sep 17 00:00:00 2001 From: Katarzyna Ziomek-Zdanowicz <24753130+ziomecka@users.noreply.github.com> Date: Wed, 2 Sep 2026 19:35:23 +0200 Subject: [PATCH] feat(server): publish [:grpc, :server, :rpc, :abort] telemetry MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An RPC the adapter stops before it returns — expired deadline, client cancellation, a dropped connection — is stopped with an exit signal that does not unwind its process. `after` blocks never run, so the `:telemetry.span/3` around the call publishes neither `:stop` nor `:exception`, and such calls leave no telemetry at all. Publish `:abort` from the cowboy handler process, which outlives the RPC process, carrying the stream, request path, RPC pid and exit reason. Both abort paths can run for one call and the exit signal is asynchronous, so the event is guarded to one per call. --- grpc_core/lib/grpc/telemetry.ex | 13 ++ grpc_server/CHANGELOG.md | 6 + .../grpc/server/adapters/cowboy/handler.ex | 60 ++++++++- .../server/adapters/cowboy/handler_test.exs | 125 ++++++++++++++++++ grpc_server/test/support/data_case.ex | 21 +++ 5 files changed, 220 insertions(+), 5 deletions(-) diff --git a/grpc_core/lib/grpc/telemetry.ex b/grpc_core/lib/grpc/telemetry.ex index 8764493dc..d1e84d480 100644 --- a/grpc_core/lib/grpc/telemetry.ex +++ b/grpc_core/lib/grpc/telemetry.ex @@ -47,12 +47,19 @@ defmodule GRPC.Telemetry do * `[: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. + * `[:grpc, :server, :rpc, :abort]` - Published when the adapter stops an in-flight + RPC: expired deadline, client cancellation or a dropped connection. The exit + signal doesn't unwind the RPC process, so `:stop` and `:exception` are not + published for such a call; this event comes from the adapter's process. + * `:duration` - the duration as measured through `System.monotonic_time()` + from the arrival of the request until the abort. | 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` | + | `[:rpc, :abort]` | `:duration` | `:stream`, `:server`, `:endpoint`, `:path`, `:pid`, `:reason` | ### Metadata @@ -62,6 +69,12 @@ defmodule GRPC.Telemetry do * `:endpoint` - the endpoint module name. * `:result` - the result returned from the interceptor pipeline. + `:abort` events also include `:path` (the request path), `:pid` (the RPC + process being stopped) and `:reason` (the exit reason, e.g. `:timeout`). + They carry no `:function_name`, which is resolved in the RPC process; for + the same reason `:stream` holds only what the adapter built on arrival, so + read `:server` and `:endpoint` from the metadata rather than from it. + `:exception` events also include some error metadata: * `:reason` is the error value in case of `catch` or the actual exception in case of `rescue`. diff --git a/grpc_server/CHANGELOG.md b/grpc_server/CHANGELOG.md index 686e0b26c..08e7ba7f7 100644 --- a/grpc_server/CHANGELOG.md +++ b/grpc_server/CHANGELOG.md @@ -1,5 +1,11 @@ # Changelog +## Unreleased + +### Enhancements + + * `[:grpc, :server, :rpc, :abort]` is now published when the adapter stops an in-flight RPC — an expired deadline, a client cancellation, a dropped connection. The exit signal that stops the RPC process does not unwind it, so `:stop` and `:exception` cannot be published for such a call; the new event carries the stream, server, endpoint, request path, RPC pid and exit reason. + ## v1.0.4 (2026-0-15) ### Bug Fixes diff --git a/grpc_server/lib/grpc/server/adapters/cowboy/handler.ex b/grpc_server/lib/grpc/server/adapters/cowboy/handler.ex index c2a9fccb0..944815e17 100644 --- a/grpc_server/lib/grpc/server/adapters/cowboy/handler.ex +++ b/grpc_server/lib/grpc/server/adapters/cowboy/handler.ex @@ -14,6 +14,8 @@ defmodule GRPC.Server.Adapters.Cowboy.Handler do @default_trailers HTTP2.server_trailers() @trailers_flag 0b1000_0000 + @abort_published :"$grpc_abort_published" + # 4 MB – matches gRPC-Go's default max receive message size. # Override per-server with the :max_body_size option (bytes). @default_max_body_size 4 * 1024 * 1024 @@ -35,6 +37,7 @@ defmodule GRPC.Server.Adapters.Cowboy.Handler do {:ok, codec} <- find_codec(sub_type, content_type, server), {:ok, compressor} <- find_compressor(req, server) do stream_pid = self() + started_at = System.monotonic_time() http_transcode = access_mode == :http_transcoding request_headers = :cowboy_req.headers(req) @@ -81,6 +84,10 @@ defmodule GRPC.Server.Adapters.Cowboy.Handler do req, %{ pid: server_rpc_pid, + stream: stream, + endpoint: endpoint, + route: route, + started_at: started_at, handling_timer: timer_ref, pending_reader: nil, access_mode: access_mode, @@ -525,8 +532,8 @@ defmodule GRPC.Server.Adapters.Cowboy.Handler do {:stop, req, state} end - def terminate(reason, _req, %{pid: pid}) do - exit_handler(pid, reason) + def terminate(reason, _req, state = %{pid: _pid}) do + abort_rpc(state, reason) :ok end @@ -644,6 +651,51 @@ defmodule GRPC.Server.Adapters.Cowboy.Handler do end end + # Stops an RPC that is still running, and publishes it. The exit signal does + # not unwind the RPC process, so `:telemetry.span/3` around the call publishes + # neither `:stop` nor `:exception`; this process outlives it and publishes + # `:abort` before signalling. + # + # Both abort paths can run for one call — `send_error/4` signals the RPC + # process and cowboy then calls `terminate/3` — and the signal is + # asynchronous, so the process may still be alive for the second. The flag + # keeps the event to one per call; it lives in the cowboy request process, + # which handles this call only. + defp abort_rpc(state, reason) do + case Map.get(state, :pid) do + pid when is_pid(pid) -> + if Process.alive?(pid) do + unless Process.put(@abort_published, true) do + publish_abort(state, pid, reason) + end + + exit_handler(pid, reason) + end + + _no_rpc_process -> + :ok + end + + :ok + end + + defp publish_abort(state, pid, reason) do + %{stream: stream, endpoint: endpoint, route: route, started_at: started_at} = state + + :telemetry.execute( + GRPC.Telemetry.server_rpc_prefix() ++ [:abort], + %{duration: System.monotonic_time() - started_at}, + %{ + stream: stream, + server: stream.server, + endpoint: endpoint, + path: route, + pid: pid, + reason: reason + } + ) + end + defp timeout_left_opt(timer, opts \\ %{}) do case timer do nil -> @@ -709,9 +761,7 @@ defmodule GRPC.Server.Adapters.Cowboy.Handler do do: GRPC.Status.http_code(error.status), else: 200 - if pid = Map.get(state, :pid) do - exit_handler(pid, reason) - end + abort_rpc(state, reason) send_error_trailers(req, status, trailers, state) end diff --git a/grpc_server/test/grpc/server/adapters/cowboy/handler_test.exs b/grpc_server/test/grpc/server/adapters/cowboy/handler_test.exs index a7971478b..702ee6c93 100644 --- a/grpc_server/test/grpc/server/adapters/cowboy/handler_test.exs +++ b/grpc_server/test/grpc/server/adapters/cowboy/handler_test.exs @@ -2,6 +2,7 @@ defmodule GRPC.Server.Adapters.Cowboy.HandlerTest do use ExUnit.Case, async: false import ExUnit.CaptureLog + import GRPC.DataCase, only: [attach_telemetry: 1] # -------------------------------------------------------------------------- # Minimal server used across all tests @@ -15,6 +16,25 @@ defmodule GRPC.Server.Adapters.Cowboy.HandlerTest do end end + defmodule SlowServer do + use GRPC.Server, service: Helloworld.Greeter.Service + + def say_hello(req, _stream) do + Process.sleep(5_000) + %Helloworld.HelloReply{message: "Hello, #{req.name}"} + end + end + + defmodule TrappingServer do + use GRPC.Server, service: Helloworld.Greeter.Service + + def say_hello(req, _stream) do + Process.flag(:trap_exit, true) + Process.sleep(3_000) + %Helloworld.HelloReply{message: "Hello, #{req.name}"} + end + end + # -------------------------------------------------------------------------- # Helpers # -------------------------------------------------------------------------- @@ -155,6 +175,111 @@ defmodule GRPC.Server.Adapters.Cowboy.HandlerTest do end end + # -------------------------------------------------------------------------- + # Tests: telemetry for RPCs the adapter stops before they return + # -------------------------------------------------------------------------- + + describe "aborted RPCs" do + test "an expired deadline publishes :abort and no :stop" do + attach_telemetry([:grpc, :server, :rpc, :abort]) + attach_telemetry([:grpc, :server, :rpc, :stop]) + attach_telemetry([:grpc, :server, :rpc, :exception]) + + capture_log(fn -> + run_server_with_opts([SlowServer], [], fn port -> + headers = [{"grpc-timeout", "50m"} | grpc_request_headers()] + body = grpc_frame(Protobuf.encode(%Helloworld.HelloRequest{name: "slow"})) + + conn = open_h2(port) + ref = :gun.post(conn, "/helloworld.Greeter/SayHello", headers, body) + + assert collect_grpc_status(conn, ref) == "4" + + :gun.close(conn) + end) + end) + + assert_receive {:telemetry, [:grpc, :server, :rpc, :abort], measurements, metadata} + assert metadata.reason == :timeout + assert metadata.path == "/helloworld.Greeter/SayHello" + assert metadata.server == SlowServer + assert metadata.endpoint == nil + assert is_pid(metadata.pid) + assert metadata.stream.http_request_headers["grpc-timeout"] == "50m" + assert metadata.stream.deadline + assert measurements.duration > 0 + + # The RPC process is stopped by a signal it cannot unwind, so the span + # around the call publishes nothing: without `:abort` the call would be + # absent from telemetry entirely. + refute_receive {:telemetry, [:grpc, :server, :rpc, :stop], _, _}, 200 + refute_receive {:telemetry, [:grpc, :server, :rpc, :exception], _, _}, 10 + end + + test "publishes :abort once when both abort paths run for one call" do + attach_telemetry([:grpc, :server, :rpc, :abort]) + + capture_log(fn -> + run_server_with_opts([TrappingServer], [], fn port -> + headers = [{"grpc-timeout", "50m"} | grpc_request_headers()] + body = grpc_frame(Protobuf.encode(%Helloworld.HelloRequest{name: "trap"})) + + conn = open_h2(port) + ref = :gun.post(conn, "/helloworld.Greeter/SayHello", headers, body) + + assert collect_grpc_status(conn, ref) == "4" + + :gun.close(conn) + end) + end) + + # `send_error/4` publishes and signals; cowboy then calls `terminate/3`, + # by which point this RPC process has not died. + assert_receive {:telemetry, [:grpc, :server, :rpc, :abort], _, %{reason: :timeout}} + refute_receive {:telemetry, [:grpc, :server, :rpc, :abort], _, _}, 500 + end + + test "a dropped connection publishes :abort" do + attach_telemetry([:grpc, :server, :rpc, :abort]) + + capture_log(fn -> + run_server_with_opts([SlowServer], [], fn port -> + body = grpc_frame(Protobuf.encode(%Helloworld.HelloRequest{name: "slow"})) + + conn = open_h2(port) + _ref = :gun.post(conn, "/helloworld.Greeter/SayHello", grpc_request_headers(), body) + + # No deadline: the RPC is still running when the client goes away. + Process.sleep(100) + :gun.close(conn) + + assert_receive {:telemetry, [:grpc, :server, :rpc, :abort], _, metadata}, 1_000 + assert metadata.path == "/helloworld.Greeter/SayHello" + refute metadata.reason == :timeout + end) + end) + end + + test "a call that returns publishes :stop and no :abort" do + attach_telemetry([:grpc, :server, :rpc, :abort]) + attach_telemetry([:grpc, :server, :rpc, :stop]) + + run_server_with_opts([HelloServer], [], fn port -> + body = grpc_frame(Protobuf.encode(%Helloworld.HelloRequest{name: "hi"})) + + conn = open_h2(port) + ref = :gun.post(conn, "/helloworld.Greeter/SayHello", grpc_request_headers(), body) + + assert collect_grpc_status(conn, ref) == "0" + + :gun.close(conn) + end) + + assert_receive {:telemetry, [:grpc, :server, :rpc, :stop], _measurements, _metadata} + refute_receive {:telemetry, [:grpc, :server, :rpc, :abort], _, _}, 200 + end + end + # -------------------------------------------------------------------------- # Private helper: start a server with specific opts and run a test function # -------------------------------------------------------------------------- diff --git a/grpc_server/test/support/data_case.ex b/grpc_server/test/support/data_case.ex index 8e0de59d3..11f99788f 100644 --- a/grpc_server/test/support/data_case.ex +++ b/grpc_server/test/support/data_case.ex @@ -6,4 +6,25 @@ defmodule GRPC.DataCase do import GRPC.Factory end end + + @doc """ + Attaches a telemetry handler for `event` that forwards emissions to the + test process as `{:telemetry, event, measurements, metadata}` messages, + for use with `assert_receive`. The handler is detached on test exit. + """ + def attach_telemetry(event) do + handler_id = {__MODULE__, self(), System.unique_integer()} + test_pid = self() + + :telemetry.attach( + handler_id, + event, + fn event, measurements, metadata, _config -> + send(test_pid, {:telemetry, event, measurements, metadata}) + end, + nil + ) + + ExUnit.Callbacks.on_exit(fn -> :telemetry.detach(handler_id) end) + end end