Measure latency on connections that were already established

README.md

@@ -58,6 +58,12 @@ The output was like below:
 
 Use the option `-t` would only show the request which the request latency was slower than threshold(in units of millisecond).
 
+tcpkit works out which side of a connection is the server from the TCP
+handshake when it captures one. For connections that were already established
+when tcpkit started -- the usual case with pooled clients -- it assumes the
+lower of the two ports is the service, since clients draw from the ephemeral
+range above it. A captured handshake always takes precedence.
+
 
 ## How to Use Lua Script
 

src/packet.c

@@ -326,12 +326,40 @@ void print_user_packet(struct sniffer *sniffer, struct user_packet *upacket) {
     }
 }
 
-void process_user_packet(struct sniffer *sniffer, struct user_packet *upacket) {
-    int src;
-    uint8_t syn_mask = 0x02, ack_mask = 0x10;
+/* Records `ip:port` as a server endpoint, which is what the latency and byte
+ * counters are keyed on. */
+static void register_endpoint(struct sniffer *sniffer, struct in_addr ip, uint16_t port) {
     char key[32];
     struct query_stats *stats;
 
+    snprintf(key, sizeof(key), "%u:%d", ip.s_addr, port);
+    sniffer_stats_lock(sniffer);
+    if (!hashtable_get(sniffer->syn_tab, key)) {
+        stats = calloc(1, sizeof(*stats));
+        if (stats) {
+            stats->ip = ip;
+            stats->port = port;
+            hashtable_add(sniffer->syn_tab, key, stats);
+        }
+    }
+    sniffer_stats_unlock(sniffer);
+}
+
+/* With no handshake to learn from, the server side has to be guessed. Service
+ * ports sit below the ephemeral range clients draw from, so the lower port is
+ * the server -- which holds whichever direction happens to be seen first. */
+static void infer_server_endpoint(struct sniffer *sniffer, struct user_packet *upacket) {
+    if (upacket->port_dst <= upacket->port_src) {
+        register_endpoint(sniffer, upacket->ip_dst, upacket->port_dst);
+    } else {
+        register_endpoint(sniffer, upacket->ip_src, upacket->port_src);
+    }
+}
+
+void process_user_packet(struct sniffer *sniffer, struct user_packet *upacket) {
+    int direction;
+    uint8_t syn_mask = 0x02, ack_mask = 0x10;
+
     // push to lua state if script exists 
     if (sniffer->lua_state) {
         push_packet_to_lua_state(sniffer->lua_state, upacket);
@@ -343,29 +371,25 @@ void process_user_packet(struct sniffer *sniffer, struct user_packet *upacket) {
     expire_stale_requests(sniffer, upacket->tv);
     if (upacket->payload_size == 0) {
         if ((upacket->flags & syn_mask) != 0) {
-            src = (upacket->flags & ack_mask) != 0; 
-            if (src) {
-                snprintf(key, sizeof(key), "%u:%d", upacket->ip_src.s_addr, upacket->port_src);
+            /* On a SYN the server is the destination, on its SYN+ACK the source. */
+            if ((upacket->flags & ack_mask) != 0) {
+                register_endpoint(sniffer, upacket->ip_src, upacket->port_src);
             } else {
-                snprintf(key, sizeof(key), "%u:%d", upacket->ip_dst.s_addr, upacket->port_dst);
-            }
-            sniffer_stats_lock(sniffer);
-            if (!hashtable_get(sniffer->syn_tab, key)) {
-                stats = calloc(1, sizeof(*stats));
-                if (stats) {
-                    stats->ip = src ? upacket->ip_src : upacket->ip_dst;
-                    stats->port = src ? upacket->port_src : upacket->port_dst;
-                    hashtable_add(sniffer->syn_tab, key, stats);
-                }
+                register_endpoint(sniffer, upacket->ip_dst, upacket->port_dst);
             }
-            sniffer_stats_unlock(sniffer);
-        }
-    } else {
-        switch(packet_direction(sniffer, upacket)) {
-            case 0:
-                return process_response_packet(sniffer, upacket);
-            case 1:
-                return process_request_packet(sniffer, upacket);
         }
+        return;
+    }
+
+    direction = packet_direction(sniffer, upacket);
+    if (direction == -1) {
+        /* The capture started after this connection was established. */
+        infer_server_endpoint(sniffer, upacket);
+        direction = packet_direction(sniffer, upacket);
+    }
+    if (direction == 0) {
+        process_response_packet(sniffer, upacket);
+    } else if (direction == 1) {
+        process_request_packet(sniffer, upacket);
     }
 }

tests/e2e/run.sh

@@ -127,6 +127,14 @@ expect_lines "each request stays on one line" "$out" 3
 expect_contains "a truncated resp array is summarised" "$out" '*2..$3..GET'
 expect_contains "a payload with no crlf is summarised" "$out" "GETNOCRLFATALL"
 
+echo "== e2e: established connections"
+replay established.pcap -p redis
+expect_status "a capture with no handshake exits cleanly" "$status" 0
+expect_lines "the answered request is reported" "$out" 1
+expect_contains "the server side is identified without a syn" "$out" \
+    "10.0.0.1:51137 => 10.0.0.2:6379"
+expect_contains "the latency is measured" "$out" "1.500 ms"
+
 echo "== e2e: stale requests"
 replay stale-request.pcap -p redis
 expect_status "a stale request exits cleanly" "$status" 0

tests/fixtures/gen_fixtures.py

@@ -184,6 +184,17 @@ def malformed_payload():
     ]
 
 
+def established():
+    """No handshake at all, and the first frame is a response, as when tcpkit
+    is pointed at a connection that a pooled client opened long ago."""
+    return [
+        # The request this answers was never captured.
+        full(1.000000, to_client(b"$1\r\nz\r\n", 5001, 1001, PSH | ACK)),
+        full(2.000000, to_server(resp("GET", "a"), 1001, 5008, PSH | ACK)),
+        full(2.001500, to_client(b"$1\r\nb\r\n", 5008, 1021, PSH | ACK)),
+    ]
+
+
 def stale_request():
     """A request whose response arrives long after any latency bucket, followed
     by a healthy exchange, so the sweep can be told apart from a broken path."""
@@ -224,3 +235,4 @@ def stress(connections=4000):
         write_pcap("truncated.pcap", truncated())
         write_pcap("malformed-payload.pcap", malformed_payload())
         write_pcap("stale-request.pcap", stale_request())
+        write_pcap("established.pcap", established())