Add marina request telemetry events for 0.4.4

marina now emits two telemetry events at the request boundary:

  [marina, request, sent]  - shackle dispatch (query/batch/prepare/execute)
  [marina, request, error] - pool resolution failed

The sent event fires once per shackle:cast / shackle:call -- so a
reusable_query with a cache miss fires twice (prepare then execute),
giving accurate accounting of outgoing CQL operations rather than
API-level intent. Metadata is {operation, pool, async} so handlers
can break down by op kind and routing pool.

The error event fires at the four pool-resolution sites in marina.erl:
async_call/2, call/2, async_reusable_query/2, reusable_query/2.
Metadata is {operation, reason}.

Per-request shackle lifecycle (queue / send / receive) stays
observable via shackle's own telemetry hooks -- marina's events
surface the CQL-level intent without duplicating that work.

telemetry (1.4.2) promoted to a direct dep (was already transitive
via shackle). vsn switched from 'git' to an explicit '0.4.4' string,
matching the convention used by the rest of the family.

A new eunit subtest attaches a sent handler and asserts the event
fires on a real Scylla query. The error path is short and exercised
through a single shared helper -- I'm leaving its coverage to manual
review rather than wiring an app-stop dance into the parallel
integration suite.

Author: lpgauth

CHANGELOG.md

@@ -1,5 +1,34 @@
 # Changelog
 
+## 0.4.4
+
+### Added
+
+- Two telemetry events at the request boundary:
+
+  | Event | Measurements | Metadata |
+  |---|---|---|
+  | `[marina, request, sent]`  | `count => 1` | `operation, pool, async` |
+  | `[marina, request, error]` | `count => 1` | `operation, reason` |
+
+  `sent` fires at each shackle dispatch — once per query/batch/prepare/
+  execute. `reusable_query` with a cache miss fires twice (`prepare`
+  then `execute`), giving an accurate count of CQL ops. The `error`
+  event fires when `marina_pool:node/1` returns no pool (e.g.
+  `marina_pool_not_started`). Attach handlers via `telemetry:attach/4`.
+
+  Per-request shackle lifecycle (queue / send / receive) remains
+  observable via shackle's own telemetry — marina's events surface
+  the CQL-level intent without duplicating that work.
+
+- `telemetry` (1.4.2) is now a direct dependency (was already
+  transitively present via shackle).
+
+- `vsn` in `marina.app.src` is now an explicit string (`"0.4.4"`) —
+  was `git`, which only works when built from a checkout.
+
+No source or API changes beyond the instrumentation.
+
 ## 0.4.3
 
 ### Changed

README.md

@@ -120,6 +120,19 @@ Asynchronous (return a `shackle:request_id()`, consume via `marina:receive_respo
 - **Token-aware routing.** On boot, a balanced BST over token intervals is compiled to a runtime-generated `marina_ring_utils:lookup/1`. Routing keys are hashed with Murmur3 and traverse the tree in O(log N).
 - **Topology refresh.** `marina_control` holds one long-lived CQL connection subscribed to `TOPOLOGY_CHANGE`, `STATUS_CHANGE`, and `SCHEMA_CHANGE`. Any topology event — or any reconnect — re-queries `system.peers` and posts `{topology_full_sync, Nodes}` to the pool server, which diffs the node list, adds/removes pools, evicts prepared-statement cache for removed pools, and rebuilds the ring.
 
+## Telemetry
+
+marina emits two telemetry events at the request boundary. Attach handlers via `telemetry:attach/4`:
+
+| Event | Measurements | Metadata |
+|---|---|---|
+| `[marina, request, sent]`  | `count => 1` | `operation, pool, async` |
+| `[marina, request, error]` | `count => 1` | `operation, reason` |
+
+`sent` fires at each shackle dispatch — once per `query` / `batch` / `prepare` / `execute`. A `reusable_query` with a cache miss fires twice (`prepare` then `execute`), so the sum gives an accurate count of outgoing CQL operations. The `error` event fires when `marina_pool:node/1` returns no pool (e.g. `marina_pool_not_started`).
+
+Per-request shackle lifecycle (queue / send / receive) remains observable via shackle's own telemetry — marina's events surface the CQL-level intent without duplicating that work.
+
 ## Development
 
 ```sh

rebar.config

@@ -8,7 +8,8 @@
   {foil, "0.1.4"},
   {lz4_nif, "0.1.1"},
   {murmur_nif, "0.1.0"},
-  {shackle, "0.7.1"}
+  {shackle, "0.7.1"},
+  {telemetry, "1.4.2"}
 ]}.
 
 {dialyzer, [{plt_extra_apps, [public_key]}]}.

rebar.lock

@@ -5,7 +5,7 @@
  {<<"metal">>,{pkg,<<"metal">>,<<"0.1.2">>},1},
  {<<"murmur_nif">>,{pkg,<<"murmur_nif">>,<<"0.1.0">>},0},
  {<<"shackle">>,{pkg,<<"shackle">>,<<"0.7.1">>},0},
- {<<"telemetry">>,{pkg,<<"telemetry">>,<<"1.4.2">>},1}]}.
+ {<<"telemetry">>,{pkg,<<"telemetry">>,<<"1.4.2">>},0}]}.
 [
 {pkg_hash,[
  {<<"foil">>, <<"1B786A2CC3E81496F98F799DF23AA24065253A94032AB79E5D60AED5423850AE">>},

src/marina.app.src

@@ -1,6 +1,6 @@
 {application, marina, [
   {description, "High-Performance Erlang Cassandra / Scylla CQL Client"},
-  {vsn, git},
+  {vsn, "0.4.4"},
   {registered, []},
   {applications, [
     kernel,
@@ -11,7 +11,8 @@
     lz4_nif,
     metal,
     syntax_tools,
-    compiler
+    compiler,
+    telemetry
   ]},
   {mod, {marina_app, []}},
   {env, []},

src/marina.erl

@@ -34,6 +34,7 @@ async_reusable_query(Query, QueryOpts) ->
         {ok, Pool} ->
             async_reusable_query(Pool, Query, QueryOpts);
         {error, Reason} ->
+            telemetry_error(reusable_query, Reason),
             {error, Reason}
     end.
 
@@ -72,6 +73,7 @@ reusable_query(Query, QueryOpts) ->
         {ok, Pool} ->
             reusable_query(Pool, Query, QueryOpts);
         {error, Reason} ->
+            telemetry_error(reusable_query, Reason),
             {error, Reason}
     end.
 
@@ -82,13 +84,18 @@ async_call(Msg, QueryOpts) ->
         {ok, Pool} ->
             async_call(Pool, Msg, QueryOpts);
         {error, Reason} ->
+            telemetry_error(element(1, Msg), Reason),
             {error, Reason}
     end.
 
 
 async_call(Pool, Msg, QueryOpts) ->
     Pid = marina_utils:query_opts(pid, QueryOpts),
     Timeout = marina_utils:query_opts(timeout, QueryOpts),
+    telemetry:execute([marina, request, sent],
+                      #{count => 1},
+                      #{operation => element(1, Msg), pool => Pool,
+                        async => true}),
     shackle:cast(Pool, {Msg, QueryOpts}, Pid, Timeout).
 
 async_reusable_query(Pool, Query, QueryOpts) ->
@@ -111,13 +118,23 @@ call(Msg, QueryOpts) ->
         {ok, Pool} ->
             call(Pool, Msg, QueryOpts);
         {error, Reason} ->
+            telemetry_error(element(1, Msg), Reason),
             {error, Reason}
     end.
 
 call(Pool, Msg, QueryOpts) ->
     Timeout = marina_utils:query_opts(timeout, QueryOpts),
+    telemetry:execute([marina, request, sent],
+                      #{count => 1},
+                      #{operation => element(1, Msg), pool => Pool,
+                        async => false}),
     response(shackle:call(Pool, {Msg, QueryOpts}, Timeout)).
 
+telemetry_error(Operation, Reason) ->
+    telemetry:execute([marina, request, error],
+                      #{count => 1},
+                      #{operation => Operation, reason => Reason}).
+
 reusable_query(Pool, Query, QueryOpts) ->
     Timeout = marina_utils:query_opts(timeout, QueryOpts),
     Timestamp = os:timestamp(),

test/marina_tests.erl

@@ -20,6 +20,7 @@ marina_test_() ->
         fun reusable_query/0,
         fun reusable_query_invalid/0,
         fun schema_change_udt/0,
+        fun telemetry_sent/0,
         fun tuples/0
     ]).
 
@@ -226,6 +227,26 @@ reusable_query_invalid() ->
     ?assertMatch({error, {8704, _}},
         marina:reusable_query(<<"SELECT * FROM user LIMIT 1;">>, #{})).
 
+telemetry_sent() ->
+    Self = self(),
+    HandlerId = <<"marina-test-sent">>,
+    ok = telemetry:attach(HandlerId, [marina, request, sent],
+        fun (Event, Measurements, Metadata, _) ->
+            Self ! {telemetry, Event, Measurements, Metadata}
+        end, undefined),
+    try
+        ?QUERY1_RESULT = q(?QUERY1),
+        receive
+            {telemetry, [marina, request, sent],
+             #{count := 1},
+             #{operation := query, async := false}} -> ok
+        after 1000 ->
+            erlang:error(timeout_waiting_for_sent_event)
+        end
+    after
+        telemetry:detach(HandlerId)
+    end.
+
 schema_change_udt() ->
     q(<<"DROP KEYSPACE IF EXISTS test2;">>),
     q(<<"CREATE KEYSPACE test2 WITH REPLICATION ="