[POA-3519] Report rate limited requests & configured rate limit (#108)

Count the number of rate limited requests. Report the configured rate
limit during the initial telemetry report and with each stats capture
report.

Author: cesher

apidump/apidump.go

@@ -244,6 +244,7 @@ func (a *apidump) SendInitialTelemetry() {
 		CLITargetArch:             architecture.GetCanonicalArch(),
 		AkitaDockerRelease:        env.InDocker(),
 		DockerDesktop:             env.HasDockerInternalHostAddress(),
+		AgentRateLimit:            a.WitnessesPerMinute,
 	}
 
 	if pod, present := a.Args.Tags[tags.XAkitaKubernetesPod]; present {
@@ -285,6 +286,7 @@ func (a *apidump) SendPacketTelemetry(observedDuration int) {
 	req := &kgxapi.PostClientPacketCaptureStatsRequest{
 		AgentResourceUsage:        usage.Get(),
 		ObservedDurationInSeconds: observedDuration,
+		AgentRateLimit:            a.WitnessesPerMinute,
 	}
 	if a.dumpSummary != nil {
 		req.PacketCountSummary = a.dumpSummary.FilterSummary.Summary(topNForSummary)
@@ -766,7 +768,7 @@ func (a *apidump) Run() error {
 			// Subsampling.
 			collector = trace.NewSamplingCollector(args.SampleRate, collector)
 			if rateLimit != nil {
-				collector = rateLimit.NewCollector(collector)
+				collector = rateLimit.NewCollector(collector, summary)
 			}
 
 			// Path and host filters.

go.mod

@@ -7,7 +7,7 @@ require (
 	github.com/OneOfOne/xxhash v1.2.8
 	github.com/Pallinder/go-randomdata v1.2.0
 	github.com/akitasoftware/akita-ir v0.0.0-20241213050034-057d7b6097e8
-	github.com/akitasoftware/akita-libs v0.0.0-20250314233547-6ff5afce3bf6
+	github.com/akitasoftware/akita-libs v0.0.0-20250428180153-cb2e977a2ee3
 	github.com/akitasoftware/go-utils v0.0.0-20240213133309-b95d4ace8803
 	github.com/andybalholm/brotli v1.0.1
 	github.com/aws/aws-sdk-go-v2 v1.17.1

go.sum

@@ -32,6 +32,8 @@ github.com/akitasoftware/akita-libs v0.0.0-20250207124702-a568277a6ab6 h1:HPvFcp
 github.com/akitasoftware/akita-libs v0.0.0-20250207124702-a568277a6ab6/go.mod h1:Fg14kX6+N7we3KdP1c11W/SzbKsgapV1hP5d4Z/Hqwc=
 github.com/akitasoftware/akita-libs v0.0.0-20250314233547-6ff5afce3bf6 h1:kuJ8kv8oG3h+B7MZF3x5sj4z/WTmJaNlC5FxM41fh9c=
 github.com/akitasoftware/akita-libs v0.0.0-20250314233547-6ff5afce3bf6/go.mod h1:Fg14kX6+N7we3KdP1c11W/SzbKsgapV1hP5d4Z/Hqwc=
+github.com/akitasoftware/akita-libs v0.0.0-20250428180153-cb2e977a2ee3 h1:ANLqfVeDdlJoLf1jCjWlxtKaj65N8bLJ9OLmp1h8w/o=
+github.com/akitasoftware/akita-libs v0.0.0-20250428180153-cb2e977a2ee3/go.mod h1:Fg14kX6+N7we3KdP1c11W/SzbKsgapV1hP5d4Z/Hqwc=
 github.com/akitasoftware/go-utils v0.0.0-20240213133309-b95d4ace8803 h1:ebIh/EFuaP8GczzMe8EwVID/blSv5Tej6S8NE4xyarQ=
 github.com/akitasoftware/go-utils v0.0.0-20240213133309-b95d4ace8803/go.mod h1:+IOXf7l/QCAQECJzjJwhTp1sBkRoJ6WciZwJezUwBa4=
 github.com/akitasoftware/gopacket v1.1.18-0.20240820200020-7289ae956f70 h1:VnU7QLDBwRujpQoHwShs5yu0Ahv1fSalNJa4UijwlmY=

trace/rate_limit.go

@@ -6,6 +6,7 @@ import (
 	"time"
 
 	"github.com/akitasoftware/akita-libs/akinet"
+	"github.com/akitasoftware/akita-libs/client_telemetry"
 	"github.com/postmanlabs/postman-insights-agent/printer"
 	"github.com/spf13/viper"
 )
@@ -90,8 +91,7 @@ func (r *SharedRateLimit) endInterval(end time.Time) {
 		r.FirstEstimate = false
 	} else {
 		alpha := viper.GetFloat64(RateLimitExponentialAlpha)
-		exponentialMovingAverage :=
-			(1-alpha)*float64(r.EstimatedSampleInterval) + alpha*float64(intervalLength)
+		exponentialMovingAverage := (1-alpha)*float64(r.EstimatedSampleInterval) + alpha*float64(intervalLength)
 		printer.Debugln("New estimate:", exponentialMovingAverage)
 		r.EstimatedSampleInterval = time.Duration(uint64(exponentialMovingAverage))
 	}
@@ -206,14 +206,18 @@ type rateLimitCollector struct {
 
 	// Channel from RateLimit for epoch starts
 	epochCh chan time.Time
+
+	// Packet counter
+	packetCount PacketCountConsumer
 }
 
-func (r *SharedRateLimit) NewCollector(next Collector) Collector {
+func (r *SharedRateLimit) NewCollector(next Collector, packetCounts PacketCountConsumer) Collector {
 	c := &rateLimitCollector{
 		RateLimit:           r,
 		NextCollector:       next,
 		RequestArrivalTimes: make(map[requestKey]time.Time),
 		epochCh:             make(chan time.Time, 1),
+		packetCount:         packetCounts,
 	}
 	r.lock.Lock()
 	defer r.lock.Unlock()
@@ -240,6 +244,14 @@ func (r *rateLimitCollector) Process(pnt akinet.ParsedNetworkTraffic) error {
 			r.NextCollector.Process(pnt)
 			key := requestKey{c.StreamID.String(), c.Seq}
 			r.RequestArrivalTimes[key] = pnt.ObservationTime
+		} else {
+			r.packetCount.Update(client_telemetry.PacketCounts{
+				Interface:               pnt.Interface,
+				DstHost:                 c.Host,
+				SrcPort:                 pnt.SrcPort,
+				DstPort:                 pnt.DstPort,
+				HTTPRequestsRateLimited: 1,
+			})
 		}
 	case akinet.HTTPResponse:
 		// Collect iff the request is in our map. (This means responses to calls

trace/rate_limit_test.go

@@ -44,7 +44,8 @@ func TestRateLimit_FirstSample(t *testing.T) {
 	start := time.Now()
 	cc := &countingCollector{}
 	rl := NewRateLimit(1.0)
-	c := rl.NewCollector(cc).(*rateLimitCollector)
+	pc := NewPacketCounter()
+	c := rl.NewCollector(cc, pc).(*rateLimitCollector)
 
 	// Sample packet from another test
 	streamID := uuid.New()
@@ -95,6 +96,9 @@ func TestRateLimit_FirstSample(t *testing.T) {
 	if cc.GetNumPackets() != 5 {
 		t.Errorf("Expected 5 packets in collector, got %v", cc.GetNumPackets())
 	}
+	if pc.total.HTTPRequestsRateLimited != 5 {
+		t.Errorf("Expected 5 rate limited request, got %v", pc.total.HTTPRequestsRateLimited)
+	}
 	if rl.FirstEstimate {
 		t.Errorf("Expected FirstEstimate to be false")
 	}