add instrum.body_size metric

Author: e-n-0

instrumentation/appsec/proxy/message_processor.go

@@ -10,6 +10,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"strconv"
 	"sync"
 
 	"github.com/DataDog/dd-trace-go/v2/appsec"
@@ -19,6 +20,7 @@ import (
 	"github.com/DataDog/dd-trace-go/v2/instrumentation/httptrace"
 	"github.com/DataDog/dd-trace-go/v2/internal/appsec/body"
 	"github.com/DataDog/dd-trace-go/v2/internal/appsec/body/json"
+	"github.com/DataDog/dd-trace-go/v2/internal/telemetry"
 )
 
 // Processor is a state machine that handles incoming HTTP request and response is a streaming manner
@@ -100,7 +102,6 @@ func (mp *Processor) OnRequestHeaders(ctx context.Context, req RequestHeaders) (
 
 	if !req.GetEndOfStream() && mp.isBodySupported(httpRequest.Header.Get("Content-Type")) {
 		reqState.State = MessageTypeRequestBody
-		// Todo: Set telemetry body size (using content-length)
 	}
 
 	if err := mp.ContinueMessageFunc(reqState.Context, ContinueActionOptions{
@@ -134,7 +135,7 @@ func (mp *Processor) OnRequestBody(req HTTPBody, reqState *RequestState) error {
 		return mp.ContinueMessageFunc(reqState.Context, ContinueActionOptions{MessageType: MessageTypeRequestBody})
 	}
 
-	blocked := processBody(reqState.Context, reqState.requestBuffer, req.GetBody(), req.GetEndOfStream(), appsec.MonitorParsedHTTPBody)
+	blocked := processBody(reqState.Context, reqState.requestBuffer, req.GetBody(), req.GetEndOfStream(), appsec.MonitorParsedHTTPBody, "request")
 	if blocked != nil && !mp.BlockingUnavailable {
 		mp.instr.Logger().Debug("external_processing: request blocked, end the stream")
 		actionOpts := reqState.BlockAction()
@@ -178,7 +179,6 @@ func (mp *Processor) OnResponseHeaders(res ResponseHeaders, reqState *RequestSta
 		}
 	}
 
-	// TODO: Set telemetry body size (using content-length)
 	reqState.State = MessageTypeResponseBody
 
 	// Run the waf on the response headers only when we are sure to not receive a response body
@@ -217,7 +217,7 @@ func (mp *Processor) OnResponseBody(resp HTTPBody, reqState *RequestState) error
 		return io.EOF
 	}
 
-	blocked := processBody(reqState.Context, reqState.responseBuffer, resp.GetBody(), resp.GetEndOfStream(), appsec.MonitorHTTPResponseBody)
+	blocked := processBody(reqState.Context, reqState.responseBuffer, resp.GetBody(), resp.GetEndOfStream(), appsec.MonitorHTTPResponseBody, "response")
 	if reqState.responseBuffer.analyzed {
 		reqState.Close() // Call Close to ensure the response headers are analyzed
 
@@ -251,14 +251,19 @@ func (mp *Processor) OnResponseTrailers(reqState *RequestState) error {
 	return mp.ContinueMessageFunc(reqState.Context, ContinueActionOptions{MessageType: MessageTypeResponseTrailers})
 }
 
-func processBody(ctx context.Context, bodyBuffer *bodyBuffer, body []byte, eos bool, analyzeBody func(ctx context.Context, encodable any) error) error {
+func processBody(ctx context.Context, bodyBuffer *bodyBuffer, body []byte, eos bool, analyzeBody func(ctx context.Context, encodable any) error, direction string) error {
 	if bodyBuffer.analyzed {
 		return nil
 	}
 
 	bodyBuffer.append(body)
 
 	if eos || bodyBuffer.truncated {
+		telemetry.Distribution(telemetry.NamespaceAppSec, "instrum.body_size", []string{
+			"direction:" + direction,
+			"truncated:" + strconv.FormatBool(bodyBuffer.truncated),
+		}).Submit(float64(len(bodyBuffer.buffer)))
+
 		bodyBuffer.analyzed = true
 		return analyzeBody(ctx, json.NewEncodableFromData(bodyBuffer.buffer, bodyBuffer.truncated))
 	}