diff --git a/README.md b/README.md index 562c83e..c54265f 100644 --- a/README.md +++ b/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 diff --git a/src/packet.c b/src/packet.c index 223df48..9bfb6a0 100644 --- a/src/packet.c +++ b/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); } } diff --git a/tests/e2e/run.sh b/tests/e2e/run.sh index 3678e5a..24d898f 100755 --- a/tests/e2e/run.sh +++ b/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 diff --git a/tests/fixtures/established.pcap b/tests/fixtures/established.pcap new file mode 100644 index 0000000..54c66ec Binary files /dev/null and b/tests/fixtures/established.pcap differ diff --git a/tests/fixtures/gen_fixtures.py b/tests/fixtures/gen_fixtures.py index 84e4e5d..cdbc758 100644 --- a/tests/fixtures/gen_fixtures.py +++ b/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())