Yuri.lipnesh/usmon 765 http2 cont frames

pkg/network/ebpf/c/protocols/http2/decoding-defs.h

@@ -226,8 +226,8 @@ typedef struct {
 // literal_value_exceeds_frame          Count of times we couldn't retrieve the literal value due to reaching the end of the frame.
 // exceeding_max_interesting_frames		Count of times we reached the max number of frames per iteration.
 // exceeding_max_frames_to_filter		Count of times we have left with more frames to filter than the max number of frames to filter.
+// continuation_frames                  Count of occurrences where a frame of type CONTINUATION was found.
 // path_size_bucket                     Count of path sizes and divided into buckets.
-// frames_split_count                   Count of times we tried to read more data than the end of the data end.
 typedef struct {
     __u64 request_seen;
     __u64 response_seen;
@@ -236,6 +236,7 @@ typedef struct {
     __u64 literal_value_exceeds_frame;
     __u64 exceeding_max_interesting_frames;
     __u64 exceeding_max_frames_to_filter;
+    __u64 continuation_frames;
     __u64 path_size_bucket[HTTP2_TELEMETRY_PATH_BUCKETS+1];
 } http2_telemetry_t;
 

pkg/network/ebpf/c/protocols/http2/decoding.h

@@ -575,10 +575,14 @@ static __always_inline bool pktbuf_find_relevant_frames(pktbuf_t pkt, http2_tail
         // https://datatracker.ietf.org/doc/html/rfc7540#section-6.2 for headers frame.
         is_headers_or_rst_frame = current_frame.type == kHeadersFrame || current_frame.type == kRSTStreamFrame;
         is_data_end_of_stream = ((current_frame.flags & HTTP2_END_OF_STREAM) == HTTP2_END_OF_STREAM) && (current_frame.type == kDataFrame);
-        if (iteration_value->frames_count < HTTP2_MAX_FRAMES_ITERATIONS && (is_headers_or_rst_frame || is_data_end_of_stream)) {
-            iteration_value->frames_array[iteration_value->frames_count].frame = current_frame;
-            iteration_value->frames_array[iteration_value->frames_count].offset = pktbuf_data_offset(pkt);
-            iteration_value->frames_count++;
+        if (iteration_value->frames_count < HTTP2_MAX_FRAMES_ITERATIONS) {
+            if (is_headers_or_rst_frame || is_data_end_of_stream) {
+                iteration_value->frames_array[iteration_value->frames_count].frame = current_frame;
+                iteration_value->frames_array[iteration_value->frames_count].offset = pktbuf_data_offset(pkt);
+                iteration_value->frames_count++;
+            } else if (current_frame.type == kContinuationFrame) {
+                __sync_fetch_and_add(&http2_tel->continuation_frames, 1);
+            }
         }
 
         pktbuf_advance(pkt, current_frame.length);

pkg/network/protocols/http2/model_linux.go

@@ -445,6 +445,7 @@ HTTP2Telemetry{
 	"literal values exceed message count": %d,
 	"messages with more frames than we can filter": %d,
 	"messages with more interesting frames than we can process": %d,
+    "continuation frames" : %d,
 	"path headers length distribution": {
 		"in range [0, 120)": %d,
 		"in range [120, 130)": %d,
@@ -456,7 +457,7 @@ HTTP2Telemetry{
 		"in range [180, infinity)": %d
 	}
 }`, t.Request_seen, t.Response_seen, t.End_of_stream, t.End_of_stream_rst, t.Literal_value_exceeds_frame,
-		t.Exceeding_max_frames_to_filter, t.Exceeding_max_interesting_frames, t.Path_size_bucket[0], t.Path_size_bucket[1],
-		t.Path_size_bucket[2], t.Path_size_bucket[3], t.Path_size_bucket[4], t.Path_size_bucket[5], t.Path_size_bucket[6],
-		t.Path_size_bucket[7])
+		t.Exceeding_max_frames_to_filter, t.Exceeding_max_interesting_frames, t.Continuation_frames,
+		t.Path_size_bucket[0], t.Path_size_bucket[1], t.Path_size_bucket[2], t.Path_size_bucket[3],
+		t.Path_size_bucket[4], t.Path_size_bucket[5], t.Path_size_bucket[6], t.Path_size_bucket[7])
 }

pkg/network/protocols/http2/telemetry.go

@@ -34,6 +34,8 @@ type kernelTelemetry struct {
 	exceedingMaxInterestingFrames *libtelemetry.TLSAwareCounter
 	// exceedingMaxFramesToFilter Count of times we have left with more frames to filter than the max number of frames to filter.
 	exceedingMaxFramesToFilter *libtelemetry.TLSAwareCounter
+	// continuationFramesCount Count of occurrences where a frame of type CONTINUATION was found.
+	continuationFramesCount *libtelemetry.TLSAwareCounter
 	// fragmentedFrameCountRST Count of times we have seen a fragmented RST frame.
 	fragmentedFrameCountRST *libtelemetry.TLSAwareCounter
 	// fragmentedHeadersFrameEOSCount Count of times we have seen a fragmented headers frame with EOS.
@@ -58,6 +60,7 @@ func newHTTP2KernelTelemetry() *kernelTelemetry {
 		literalValueExceedsFrame:       libtelemetry.NewTLSAwareCounter(metricGroup, "literal_value_exceeds_frame"),
 		exceedingMaxInterestingFrames:  libtelemetry.NewTLSAwareCounter(metricGroup, "exceeding_max_interesting_frames"),
 		exceedingMaxFramesToFilter:     libtelemetry.NewTLSAwareCounter(metricGroup, "exceeding_max_frames_to_filter"),
+		continuationFramesCount:        libtelemetry.NewTLSAwareCounter(metricGroup, "continuation_frames"),
 		fragmentedDataFrameEOSCount:    libtelemetry.NewTLSAwareCounter(metricGroup, "exceeding_data_end_data_eos"),
 		fragmentedHeadersFrameCount:    libtelemetry.NewTLSAwareCounter(metricGroup, "exceeding_data_end_headers"),
 		fragmentedHeadersFrameEOSCount: libtelemetry.NewTLSAwareCounter(metricGroup, "exceeding_data_end_headers_eos"),
@@ -81,6 +84,7 @@ func (t *kernelTelemetry) update(tel *HTTP2Telemetry, isTLS bool) {
 	t.literalValueExceedsFrame.Add(int64(telemetryDelta.Literal_value_exceeds_frame), isTLS)
 	t.exceedingMaxInterestingFrames.Add(int64(telemetryDelta.Exceeding_max_interesting_frames), isTLS)
 	t.exceedingMaxFramesToFilter.Add(int64(telemetryDelta.Exceeding_max_frames_to_filter), isTLS)
+	t.continuationFramesCount.Add(int64(telemetryDelta.Continuation_frames), isTLS)
 	for bucketIndex := range t.pathSizeBucket {
 		t.pathSizeBucket[bucketIndex].Add(int64(telemetryDelta.Path_size_bucket[bucketIndex]), isTLS)
 	}
@@ -104,6 +108,7 @@ func (t *HTTP2Telemetry) Sub(other HTTP2Telemetry) *HTTP2Telemetry {
 		Literal_value_exceeds_frame:      t.Literal_value_exceeds_frame - other.Literal_value_exceeds_frame,
 		Exceeding_max_interesting_frames: t.Exceeding_max_interesting_frames - other.Exceeding_max_interesting_frames,
 		Exceeding_max_frames_to_filter:   t.Exceeding_max_frames_to_filter - other.Exceeding_max_frames_to_filter,
+		Continuation_frames:              t.Continuation_frames - other.Continuation_frames,
 		Path_size_bucket:                 computePathSizeBucketDifferences(t.Path_size_bucket, other.Path_size_bucket),
 	}
 }

pkg/network/protocols/http2/telemetry_test.go

@@ -46,6 +46,7 @@ func testKernelTelemetryUpdate(t *testing.T, isTLS bool) {
 		Literal_value_exceeds_frame:      20,
 		Exceeding_max_interesting_frames: 30,
 		Exceeding_max_frames_to_filter:   40,
+		Continuation_frames:              5,
 		Path_size_bucket:                 [8]uint64{1, 2, 3, 4, 5, 6, 7, 8},
 	}
 	kernelTelemetryGroup.update(http2Telemetry, isTLS)
@@ -61,6 +62,7 @@ func testKernelTelemetryUpdate(t *testing.T, isTLS bool) {
 	http2Telemetry.Literal_value_exceeds_frame = 26
 	http2Telemetry.Exceeding_max_interesting_frames = 32
 	http2Telemetry.Exceeding_max_frames_to_filter = 42
+	http2Telemetry.Continuation_frames = 6
 	http2Telemetry.Path_size_bucket = [8]uint64{2, 3, 4, 5, 6, 7, 8, 9}
 	kernelTelemetryGroup.update(http2Telemetry, isTLS)
 	assertTelemetryEquality(t, http2Telemetry, kernelTelemetryGroup, isTLS)
@@ -74,6 +76,8 @@ func assertTelemetryEquality(t *testing.T, http2Telemetry *HTTP2Telemetry, kerne
 	assert.Equal(t, http2Telemetry.Literal_value_exceeds_frame, uint64(kernelTelemetryGroup.literalValueExceedsFrame.Get(isTLS)))
 	assert.Equal(t, http2Telemetry.Exceeding_max_interesting_frames, uint64(kernelTelemetryGroup.exceedingMaxInterestingFrames.Get(isTLS)))
 	assert.Equal(t, http2Telemetry.Exceeding_max_frames_to_filter, uint64(kernelTelemetryGroup.exceedingMaxFramesToFilter.Get(isTLS)))
+	assert.Equal(t, http2Telemetry.Continuation_frames, uint64(kernelTelemetryGroup.continuationFramesCount.Get(isTLS)))
+
 	for i, bucket := range kernelTelemetryGroup.pathSizeBucket {
 		assert.Equal(t, http2Telemetry.Path_size_bucket[i], uint64(bucket.Get(isTLS)))
 	}

pkg/network/usm/usm_http2_monitor_test.go

@@ -1422,6 +1422,80 @@ func (s *usmHTTP2Suite) TestIncompleteFrameTable() {
 	}
 }
 
+// TestContinuationFrame tests CONTINUATION frame is captured by kernel telemetry.
+func (s *usmHTTP2Suite) TestContinuationFrame() {
+	t := s.T()
+	cfg := s.getCfg()
+
+	// Start local server and register its cleanup.
+	t.Cleanup(startH2CServer(t, authority, s.isTLS))
+
+	// Start the proxy server.
+	proxyProcess, cancel := proxy.NewExternalUnixTransparentProxyServer(t, unixPath, authority, s.isTLS)
+	t.Cleanup(cancel)
+	require.NoError(t, proxy.WaitForConnectionReady(unixPath))
+
+	tests := []struct {
+		name              string
+		messageBuilder    func() [][]byte
+		expectedTelemetry *usmhttp2.HTTP2Telemetry
+	}{
+		{
+			name: "CONTINUATION frame",
+			messageBuilder: func() [][]byte {
+				const headersFrameEndHeaders = false
+				fullHeaders := generateTestHeaderFields(headersGenerationOptions{})
+				prefixHeadersFrame, err := usmhttp2.NewHeadersFrameMessage(usmhttp2.HeadersFrameOptions{
+					Headers: fullHeaders[:2],
+				})
+				require.NoError(t, err, "could not create prefix headers frame")
+
+				suffixHeadersFrame, err := usmhttp2.NewHeadersFrameMessage(usmhttp2.HeadersFrameOptions{
+					Headers: fullHeaders[2:],
+				})
+				require.NoError(t, err, "could not create suffix headers frame")
+
+				framer := newFramer()
+				framer.
+					writeRawHeaders(t, 1, headersFrameEndHeaders, prefixHeadersFrame).
+					writeRawContinuation(t, 1, endHeaders, suffixHeadersFrame)
+				return [][]byte{framer.bytes()}
+			},
+			expectedTelemetry: &usmhttp2.HTTP2Telemetry{
+				Continuation_frames: 1,
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			monitor := setupUSMTLSMonitor(t, cfg, useExistingConsumer)
+			if s.isTLS {
+				utils.WaitForProgramsToBeTraced(t, consts.USMModuleName, GoTLSAttacherName, proxyProcess.Process.Pid, utils.ManualTracingFallbackEnabled)
+			}
+
+			c := dialHTTP2Server(t)
+
+			// Composing CONTINUATION frame.
+			require.NoError(t, writeInput(c, 500*time.Millisecond, tt.messageBuilder()...))
+
+			var telemetry *usmhttp2.HTTP2Telemetry
+			var err error
+			assert.Eventually(t, func() bool {
+
+				telemetry, err = getHTTP2KernelTelemetry(monitor, s.isTLS)
+				require.NoError(t, err)
+
+				return telemetry.Continuation_frames == tt.expectedTelemetry.Continuation_frames
+			}, time.Second*5, time.Millisecond*100)
+			if t.Failed() {
+				t.Logf("CONTINUATION frames count: %d != %d", tt.expectedTelemetry.Continuation_frames, telemetry.Continuation_frames)
+				ebpftest.DumpMapsTestHelper(t, monitor.DumpMaps, usmhttp2.InFlightMap, "http2_dynamic_table")
+				dumpTelemetry(t, monitor, s.isTLS)
+			}
+		})
+	}
+}
+
 func (s *usmHTTP2Suite) TestRawHuffmanEncoding() {
 	t := s.T()
 	cfg := s.getCfg()