diff --git a/service-t/native/libs/libnetworkstats/NetworkTraceHandler.cpp b/service-t/native/libs/libnetworkstats/NetworkTraceHandler.cpp index 696a29a49e..3d22aac014 100644 --- a/service-t/native/libs/libnetworkstats/NetworkTraceHandler.cpp +++ b/service-t/native/libs/libnetworkstats/NetworkTraceHandler.cpp @@ -179,24 +179,25 @@ void NetworkTraceHandler::Write(const std::vector& packets, bundle.bytes += pkt.length; } - // If state was cleared, emit a separate packet to indicate it. This uses the - // overall minTs so it is sorted before any packets that follow. NetworkTraceState* incr_state = ctx.GetIncrementalState(); - if (!bundles.empty() && mInternLimit && incr_state->cleared) { - auto clear = ctx.NewTracePacket(); - clear->set_sequence_flags(TracePacket::SEQ_INCREMENTAL_STATE_CLEARED); - clear->set_timestamp(minTs); - incr_state->cleared = false; - } - for (const auto& kv : bundles) { const BundleKey& key = kv.first; const BundleDetails& details = kv.second; auto dst = ctx.NewTracePacket(); - dst->set_sequence_flags(TracePacket::SEQ_NEEDS_INCREMENTAL_STATE); dst->set_timestamp(details.minTs); + // Incremental state is only used when interning. Set the flag based on + // whether state was cleared. Leave the flag empty in non-intern configs. + if (mInternLimit > 0) { + if (incr_state->cleared) { + dst->set_sequence_flags(TracePacket::SEQ_INCREMENTAL_STATE_CLEARED); + incr_state->cleared = false; + } else { + dst->set_sequence_flags(TracePacket::SEQ_NEEDS_INCREMENTAL_STATE); + } + } + auto* event = FillWithInterning(incr_state, key, dst.get()); int count = details.time_and_len.size(); diff --git a/service-t/native/libs/libnetworkstats/NetworkTraceHandlerTest.cpp b/service-t/native/libs/libnetworkstats/NetworkTraceHandlerTest.cpp index c9eb183046..27dce759f0 100644 --- a/service-t/native/libs/libnetworkstats/NetworkTraceHandlerTest.cpp +++ b/service-t/native/libs/libnetworkstats/NetworkTraceHandlerTest.cpp @@ -136,6 +136,7 @@ TEST_F(NetworkTraceHandlerTest, WriteBasicFields) { EXPECT_THAT(events[0].network_packet().ip_proto(), 6); EXPECT_THAT(events[0].network_packet().tcp_flags(), 1); EXPECT_THAT(events[0].network_packet().length(), 100); + EXPECT_THAT(events[0].has_sequence_flags(), false); } TEST_F(NetworkTraceHandlerTest, WriteDirectionAndPorts) { @@ -374,20 +375,28 @@ TEST_F(NetworkTraceHandlerTest, Interning) { ASSERT_EQ(events[0].interned_data().packet_context().size(), 1); EXPECT_EQ(events[0].interned_data().packet_context(0).iid(), 1); EXPECT_EQ(events[0].interned_data().packet_context(0).ctx().uid(), 123); + EXPECT_EQ(events[0].sequence_flags(), + TracePacket::SEQ_INCREMENTAL_STATE_CLEARED); // First time seen, emit new interned data, bundle uses iid instead of ctx. EXPECT_EQ(events[1].network_packet_bundle().iid(), 2); ASSERT_EQ(events[1].interned_data().packet_context().size(), 1); EXPECT_EQ(events[1].interned_data().packet_context(0).iid(), 2); EXPECT_EQ(events[1].interned_data().packet_context(0).ctx().uid(), 456); + EXPECT_EQ(events[1].sequence_flags(), + TracePacket::SEQ_NEEDS_INCREMENTAL_STATE); // Not enough room in intern table (limit 2), inline the context. EXPECT_EQ(events[2].network_packet_bundle().ctx().uid(), 789); EXPECT_EQ(events[2].interned_data().packet_context().size(), 0); + EXPECT_EQ(events[2].sequence_flags(), + TracePacket::SEQ_NEEDS_INCREMENTAL_STATE); // Second time seen, no need to re-emit interned data, only record iid. EXPECT_EQ(events[3].network_packet_bundle().iid(), 1); EXPECT_EQ(events[3].interned_data().packet_context().size(), 0); + EXPECT_EQ(events[3].sequence_flags(), + TracePacket::SEQ_NEEDS_INCREMENTAL_STATE); } } // namespace bpf diff --git a/service-t/native/libs/libnetworkstats/include/netdbpf/NetworkTraceHandler.h b/service-t/native/libs/libnetworkstats/include/netdbpf/NetworkTraceHandler.h index 80871c6313..9bc777baeb 100644 --- a/service-t/native/libs/libnetworkstats/include/netdbpf/NetworkTraceHandler.h +++ b/service-t/native/libs/libnetworkstats/include/netdbpf/NetworkTraceHandler.h @@ -49,7 +49,7 @@ struct BundleEq { // loss). When state is cleared, the state object is replaced with a new default // constructed instance. struct NetworkTraceState { - bool cleared; + bool cleared = true; std::unordered_map iids; };