chore: change the resolution of reply latency to measurement to ns

Also, fix test_reply_count that did not really tested command due to missed await statements.

Author: romange

src/facade/reply_builder.cc

@@ -181,7 +181,7 @@ void SinkReplyBuilder::Send() {
 
   uint64_t after_ns = util::fb2::ProactorBase::GetMonotonicTimeNs();
   reply_stats.send_stats.count++;
-  reply_stats.send_stats.total_duration += (after_ns - pin.timestamp_ns) / 1'000;
+  reply_stats.send_stats.total_duration += (after_ns - pin.timestamp_ns);
   DVLOG(2) << "Finished writing " << total_size_ << " bytes";
 }
 

src/server/server_family.cc

@@ -1713,7 +1713,7 @@ void PrintPrometheusMetrics(uint64_t uptime, const Metrics& m, DflyCmd* dfly_cmd
                             MetricType::COUNTER, &resp->body());
   {
     AppendMetricWithoutLabels("reply_duration_seconds", "",
-                              m.facade_stats.reply_stats.send_stats.total_duration * 1e-6,
+                              m.facade_stats.reply_stats.send_stats.total_duration * 1e-9,
                               MetricType::COUNTER, &resp->body());
     AppendMetricWithoutLabels("reply_total", "", m.facade_stats.reply_stats.send_stats.count,
                               MetricType::COUNTER, &resp->body());
@@ -3015,8 +3015,6 @@ string ServerFamily::FormatInfoMetrics(const Metrics& m, std::string_view sectio
     append("defrag_attempt_total", m.shard_stats.defrag_attempt_total);
     append("defrag_realloc_total", m.shard_stats.defrag_realloc_total);
     append("defrag_task_invocation_total", m.shard_stats.defrag_task_invocation_total);
-    append("reply_count", reply_stats.send_stats.count);
-    append("reply_latency_usec", reply_stats.send_stats.total_duration);
 
     // Number of connections that are currently blocked on grabbing interpreter.
     append("blocked_on_interpreter", m.coordinator_stats.blocked_on_interpreter);

tests/dragonfly/connection_test.py

@@ -1,3 +1,4 @@
+import requests
 import random
 import logging
 import string
@@ -23,6 +24,8 @@
 from . import dfly_args
 from .instance import DflyInstance, DflyInstanceFactory
 
+from prometheus_client.parser import text_string_to_metric_families
+
 BASE_PORT = 1111
 
 
@@ -610,58 +613,68 @@ async def test_keyspace_events_config_set(async_client: aioredis.Redis):
 async def test_reply_count(async_client: aioredis.Redis):
     """Make sure reply aggregations reduce reply counts for common cases"""
 
-    async def get_reply_count():
-        return (await async_client.info("STATS"))["reply_count"]
+    port = int(async_client.get_connection_kwargs().get("port"))
+    METRICS_URL = f"http://localhost:{port}/metrics"
+    logging.getLogger("urllib3").setLevel(logging.WARNING)
+
+    def get_reply_count():
+        response = requests.get(METRICS_URL)
+        response.raise_for_status()
+        families = text_string_to_metric_families(response.text)
+        for metric in families:
+            if metric.name == "dragonfly_reply":
+                assert metric.type == "counter"
+                return int(metric.samples[0].value)
+        return None
 
     async def measure(aw):
-        before = await get_reply_count()
-        await aw
-        return await get_reply_count() - before - 1
+        before = get_reply_count()
+        aw
+        return get_reply_count() - before
 
     await async_client.config_resetstat()
-    base = await get_reply_count()
-    info_diff = await get_reply_count() - base
-    assert info_diff == 1
+    base = get_reply_count()
+    info_diff = get_reply_count() - base
+    assert info_diff == 0  # no commands yet
 
     # Warm client buffer up
     await async_client.lpush("warmup", *(i for i in range(500)))
     await async_client.lrange("warmup", 0, -1)
 
     # Integer list
     await async_client.lpush("list-1", *(i for i in range(100)))
-    assert await measure(async_client.lrange("list-1", 0, -1)) == 1
+    assert await measure(await async_client.lrange("list-1", 0, -1)) == 0
 
     # Integer set
     await async_client.sadd("set-1", *(i for i in range(100)))
-    assert await measure(async_client.smembers("set-1")) == 1
+    assert await measure(await async_client.smembers("set-1")) == 0
 
     # Sorted sets
     await async_client.zadd("zset-1", mapping={str(i): i for i in range(50)})
-    assert await measure(async_client.zrange("zset-1", 0, -1, withscores=True)) == 1
-
+    assert await measure(await async_client.zrange("zset-1", 0, -1, withscores=True)) == 0
     # Exec call
     e = async_client.pipeline(transaction=True)
     for _ in range(100):
         e.incr("num-1")
 
     # one - for MULTI-OK, one for the rest. Depends on the squashing efficiency,
     # can be either 1 or 2 replies.
-    assert await measure(e.execute()) <= 2
+    assert await measure(await e.execute()) <= 1
 
     # Just pipeline
     p = async_client.pipeline(transaction=False)
     for _ in range(100):
         p.incr("num-1")
-    assert await measure(p.execute()) <= 2
+    assert await measure(await p.execute()) <= 1
 
     # Script result
-    assert await measure(async_client.eval('return {1,2,{3,4},5,6,7,8,"nine"}', 0)) == 1
+    assert await measure(await async_client.eval('return {1,2,{3,4},5,6,7,8,"nine"}', 0)) == 0
 
     # Search results
     await async_client.execute_command("FT.CREATE i1 SCHEMA name text")
     for i in range(50):
         await async_client.hset(f"key-{i}", "name", f"name number {i}")
-    assert await measure(async_client.ft("i1").search("*")) == 1
+    assert await measure(await async_client.ft("i1").search("*")) == 0
 
 
 async def test_big_command(df_server, size=8 * 1024):