pping: Only consider flow opened on reply

Wait with sending a flow open message until a reply has been seen for
the flow. Likewise, only emit a flow closing event if the flow has
first been opened (that is, a reply has been seen).

Concerns with this commit:
- Will no longer report any RTT for packets that close the flow (will
  delete the relevant flow_state before attempting to match the
  packet).
- Introduces potential concurrency issues for flow opening/closing
  messages (although I deem them unlikely). The README has been
  updated to describe these.

Signed-off-by: Simon Sundberg <simon.sundberg@kau.se>

Author: simosund

pping/README.md

@@ -253,6 +253,20 @@ estimate anyways (may never calculate RTT for packet with the true minimum
 RTT). And even without sampling there is some inherent sampling due to TCP
 timestamps only being updated at a limited rate (1000 Hz).
 
+#### Outputting flow opening/closing events
+A flow is not considered opened until a reply has been seen for it. The
+`flow_state` map keeps information about if the flow has been opened or not,
+which is checked and updated for each reply. The check and update of this
+information is not performed atomically, which may result in multiple replies
+thinking they are the first, emitting multiple flow-opened events, in case they
+are processed concurrently.
+
+Likewise, when flows are closed it checks if the flow has been opened to
+determine if a flow closing message should be sent. If multiple replies are
+processed concurrently, it's possible one of them will update the flow-open
+information and emit a flow opening message, but another reply closing the flow
+without thinking it's ever been opened, thus not sending a flow closing message.
+
 ## Similar projects
 Passively measuring the RTT for TCP traffic is not a novel concept, and there
 exists a number of other tools that can do so. A good overview of how passive

pping/pping.c

@@ -473,10 +473,11 @@ static bool packet_ts_timeout(void *key_ptr, void *val_ptr, __u64 now)
 static bool flow_timeout(void *key_ptr, void *val_ptr, __u64 now)
 {
 	struct flow_event fe;
-	__u64 ts = ((struct flow_state *)val_ptr)->last_timestamp;
+	struct flow_state *f_state = val_ptr;
 
-	if (now > ts && now - ts > FLOW_LIFETIME) {
-		if (print_event_func) {
+	if (now > f_state->last_timestamp &&
+	    now - f_state->last_timestamp > FLOW_LIFETIME) {
+		if (print_event_func && f_state->has_opened) {
 			fe.event_type = EVENT_TYPE_FLOW;
 			fe.timestamp = now;
 			reverse_flow(&fe.flow, key_ptr);

pping/pping.h

@@ -80,7 +80,9 @@ struct flow_state {
 	__u64 rec_pkts;
 	__u64 rec_bytes;
 	__u32 last_id;
-	__u32 reserved;
+	bool has_opened;
+	enum flow_event_reason opening_reason;
+	__u16 reserved;
 };
 
 struct packet_id {

pping/pping_kern.c

@@ -417,59 +417,105 @@ static void fill_flow_event(struct flow_event *fe, struct packet_info *p_info,
 }
 
 /*
- * Attempt to create a new flow-state and push flow-opening message
+ * Attempt to create a new flow-state.
  * Returns a pointer to the flow_state if successful, NULL otherwise
  */
-static struct flow_state *create_flow(struct network_tuple *flow, void *ctx,
-				      struct packet_info *p_info,
-				      struct flow_event *fe)
+static struct flow_state *create_flow(struct packet_info *p_info)
 {
 	struct flow_state new_state = { 0 };
-
 	new_state.last_timestamp = p_info->time;
+	new_state.opening_reason = p_info->event_type == FLOW_EVENT_OPENING ?
+					       p_info->event_reason :
+					       EVENT_REASON_FIRST_OBS_PCKT;
+
 	if (bpf_map_update_elem(&flow_state, &p_info->pid.flow, &new_state,
 				BPF_NOEXIST) != 0)
 		return NULL;
 
-	if (p_info->event_type != FLOW_EVENT_OPENING) {
-		p_info->event_type = FLOW_EVENT_OPENING;
-		p_info->event_reason = EVENT_REASON_FIRST_OBS_PCKT;
-	}
-	fill_flow_event(fe, p_info, true);
-	bpf_perf_event_output(ctx, &events, BPF_F_CURRENT_CPU, fe, sizeof(*fe));
-
 	return bpf_map_lookup_elem(&flow_state, &p_info->pid.flow);
 }
 
 static struct flow_state *update_flow(void *ctx, struct packet_info *p_info,
 				      struct flow_event *fe, bool *new_flow)
 {
 	struct flow_state *f_state;
+	bool has_opened;
 	*new_flow = false;
 
 	f_state = bpf_map_lookup_elem(&flow_state, &p_info->pid.flow);
+
+	// Flow is closing - attempt to delete state if it exists
+	if (p_info->event_type == FLOW_EVENT_CLOSING ||
+	    p_info->event_type == FLOW_EVENT_CLOSING_BOTH) {
+		if (!f_state)
+			return NULL;
+
+		has_opened = f_state->has_opened;
+		if (bpf_map_delete_elem(&flow_state, &p_info->pid.flow) == 0 &&
+		    has_opened) {
+			fill_flow_event(fe, p_info, true);
+			bpf_perf_event_output(ctx, &events, BPF_F_CURRENT_CPU,
+					      fe, sizeof(*fe));
+		}
+		return NULL;
+	}
+
+	// Attempt to create flow if it does not exist
 	if (!f_state && p_info->pid_valid) {
 		*new_flow = true;
-		f_state = create_flow(&p_info->pid.flow, ctx, p_info, fe);
+		f_state = create_flow(p_info);
 	}
 
 	if (!f_state)
 		return NULL;
 
+	// Update flow state
 	f_state->sent_pkts++;
 	f_state->sent_bytes += p_info->payload;
 
 	return f_state;
 }
 
-static struct flow_state *update_rev_flow(struct packet_info *p_info)
+static struct flow_state *update_rev_flow(void *ctx, struct packet_info *p_info,
+					  struct flow_event *fe)
 {
 	struct flow_state *f_state;
+	bool has_opened;
 
 	f_state = bpf_map_lookup_elem(&flow_state, &p_info->reply_pid.flow);
+
 	if (!f_state)
 		return NULL;
 
+	// Close reverse flow
+	if (p_info->event_type == FLOW_EVENT_CLOSING_BOTH) {
+
+		has_opened = f_state->has_opened;
+		if (bpf_map_delete_elem(&flow_state, &p_info->reply_pid.flow) ==
+			    0 && has_opened) {
+			fill_flow_event(fe, p_info, false);
+			bpf_perf_event_output(ctx, &events, BPF_F_CURRENT_CPU,
+					      fe, sizeof(*fe));
+		}
+		return NULL;
+	}
+
+	// Is a new flow, push opening flow message
+	if (!f_state->has_opened) {
+		f_state->has_opened = true;
+
+		fe->event_type = EVENT_TYPE_FLOW;
+		fe->flow = p_info->pid.flow;
+		fe->timestamp = f_state->last_timestamp;
+		fe->flow_event_type = FLOW_EVENT_OPENING;
+		fe->reason = f_state->opening_reason;
+		fe->source = EVENT_SOURCE_PKT_DEST;
+		fe->reserved = 0;
+		bpf_perf_event_output(ctx, &events, BPF_F_CURRENT_CPU, fe,
+				      sizeof(*fe));
+	}
+
+	// Update flow state
 	f_state->rec_pkts++;
 	f_state->rec_bytes += p_info->payload;
 
@@ -607,34 +653,11 @@ static void pping(void *ctx, struct parsing_context *pctx)
  	if (parse_packet_identifier(pctx, &p_info) < 0)
 		return;
 
-	if (p_info.event_type != FLOW_EVENT_CLOSING &&
-	    p_info.event_type != FLOW_EVENT_CLOSING_BOTH) {
-		f_state = update_flow(ctx, &p_info, &fe, &new_flow);
-		pping_timestamp_packet(f_state, ctx, pctx, &p_info, new_flow);
-	}
+	f_state = update_flow(ctx, &p_info, &fe, &new_flow);
+	pping_timestamp_packet(f_state, ctx, pctx, &p_info, new_flow);
 
-	f_state = update_rev_flow(&p_info);
+	f_state = update_rev_flow(ctx, &p_info, &fe);
 	pping_match_packet(f_state, ctx, pctx, &p_info);
-
-	// Flow closing - try to delete flow state and push closing-event
-	if (p_info.event_type == FLOW_EVENT_CLOSING ||
-	    p_info.event_type == FLOW_EVENT_CLOSING_BOTH) {
-		if (bpf_map_delete_elem(&flow_state, &p_info.pid.flow) == 0) {
-			fill_flow_event(&fe, &p_info, true);
-			bpf_perf_event_output(ctx, &events, BPF_F_CURRENT_CPU,
-					      &fe, sizeof(fe));
-		}
-	}
-
-	// Also close reverse flow
-	if (p_info.event_type == FLOW_EVENT_CLOSING_BOTH) {
-		if (bpf_map_delete_elem(&flow_state, &p_info.reply_pid.flow) ==
-		    0) {
-			fill_flow_event(&fe, &p_info, false);
-			bpf_perf_event_output(ctx, &events, BPF_F_CURRENT_CPU,
-					      &fe, sizeof(fe));
-		}
-	}
 }
 
 // Programs