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