[usm] reliability, preparation for pool of objects, Protocol.ReleaseStats()

cmd/system-probe/modules/network_tracer.go

@@ -172,6 +172,7 @@ func (nt *networkTracer) Register(httpMux *module.Router) error {
 		}
 
 		utils.WriteAsJSON(w, httpdebugging.HTTP(cs.HTTP, cs.DNS))
+		nt.tracer.ReleaseUSMStats()
 	})
 
 	httpMux.HandleFunc("/debug/kafka_monitoring", func(w http.ResponseWriter, req *http.Request) {

pkg/network/nettop/main.go

@@ -58,6 +58,7 @@ func main() {
 		for _, c := range cs.Conns {
 			fmt.Println(network.ConnectionSummary(&c, cs.DNS))
 		}
+		t.ReleaseUSMStats()
 	}
 
 	stopChan := make(chan struct{})

pkg/network/protocols/http/protocol.go

@@ -225,6 +225,11 @@ func (p *protocol) GetStats() *protocols.ProtocolStats {
 	}
 }
 
+// ReleaseStats releases stats objects.
+func (p *protocol) ReleaseStats() {
+	p.statkeeper.ReleaseStats()
+}
+
 // IsBuildModeSupported returns always true, as http module is supported by all modes.
 func (*protocol) IsBuildModeSupported(buildmode.Type) bool {
 	return true

pkg/network/protocols/http/statkeeper.go

@@ -21,6 +21,7 @@ import (
 type StatKeeper struct {
 	mux                  sync.Mutex
 	stats                map[Key]*RequestStats
+	prevStats            map[Key]*RequestStats
 	incomplete           IncompleteBuffer
 	maxEntries           int
 	quantizer            *URLQuantizer
@@ -87,7 +88,7 @@ func (h *StatKeeper) GetAndResetAllStats() (stats map[Key]*RequestStats) {
 		}
 
 		// Rotate stats
-		stats = h.stats
+		h.prevStats = h.stats
 		h.stats = make(map[Key]*RequestStats)
 
 		// Rotate ConnectionAggregator
@@ -100,8 +101,8 @@ func (h *StatKeeper) GetAndResetAllStats() (stats map[Key]*RequestStats) {
 		h.connectionAggregator = utils.NewConnectionAggregator()
 	}()
 
-	h.clearEphemeralPorts(previousAggregationState, stats)
-	return stats
+	h.clearEphemeralPorts(previousAggregationState, h.prevStats)
+	return h.prevStats
 }
 
 // Close closes the stat keeper.
@@ -217,3 +218,10 @@ func (h *StatKeeper) clearEphemeralPorts(aggregator *utils.ConnectionAggregator,
 		stats[key] = aggregation
 	}
 }
+
+// ReleaseStats release reported stats.
+func (h *StatKeeper) ReleaseStats() {
+	h.mux.Lock()
+	defer h.mux.Unlock()
+	h.prevStats = nil
+}

pkg/network/protocols/http/statkeeper_test.go

@@ -25,7 +25,9 @@ func TestProcessHTTPTransactions(t *testing.T) {
 	cfg.MaxHTTPStatsBuffered = 1000
 	tel := NewTelemetry("http")
 	sk := NewStatkeeper(cfg, tel, NewIncompleteBuffer(cfg, tel))
-
+	t.Cleanup(func() {
+		sk.ReleaseStats()
+	})
 	srcString := "1.1.1.1"
 	dstString := "2.2.2.2"
 	sourceIP := util.AddressFromString(srcString)

pkg/network/protocols/http2/protocol.go

@@ -431,6 +431,10 @@ func (p *Protocol) GetStats() *protocols.ProtocolStats {
 	}
 }
 
+// ReleaseStats releases stats objects.
+func (p *Protocol) ReleaseStats() {
+}
+
 // IsBuildModeSupported returns always true, as http2 module is supported by all modes.
 func (*Protocol) IsBuildModeSupported(buildmode.Type) bool {
 	return true

pkg/network/protocols/kafka/protocol.go

@@ -361,6 +361,10 @@ func (p *protocol) GetStats() *protocols.ProtocolStats {
 	}
 }
 
+// ReleaseStats releases stats objects.
+func (p *protocol) ReleaseStats() {
+}
+
 // IsBuildModeSupported returns always true, as kafka module is supported by all modes.
 func (*protocol) IsBuildModeSupported(buildmode.Type) bool {
 	return true

pkg/network/protocols/postgres/protocol.go

@@ -239,6 +239,10 @@ func (p *protocol) GetStats() *protocols.ProtocolStats {
 	}
 }
 
+// ReleaseStats releases stats objects.
+func (p *protocol) ReleaseStats() {
+}
+
 // IsBuildModeSupported returns always true, as postgres module is supported by all modes.
 func (*protocol) IsBuildModeSupported(buildmode.Type) bool {
 	return true

pkg/network/protocols/protocols.go

@@ -61,6 +61,9 @@ type Protocol interface {
 	// implementation.
 	GetStats() *ProtocolStats
 
+	// ReleaseStats releases stats related objects.
+	ReleaseStats()
+
 	// IsBuildModeSupported return true is the given build mode is supported by this protocol.
 	IsBuildModeSupported(buildmode.Type) bool
 }

pkg/network/protocols/redis/protocol.go

@@ -155,6 +155,10 @@ func (p *protocol) GetStats() *protocols.ProtocolStats {
 	}
 }
 
+// ReleaseStats releases stats objects.
+func (p *protocol) ReleaseStats() {
+}
+
 // IsBuildModeSupported returns always true, as Redis module is supported by all modes.
 func (*protocol) IsBuildModeSupported(buildmode.Type) bool {
 	return true

pkg/network/tracer/tracer.go

@@ -460,6 +460,11 @@ func (t *Tracer) GetActiveConnections(clientID string) (*network.Connections, er
 	return conns, nil
 }
 
+// ReleaseUSMStats releases usm stats objects.
+func (t *Tracer) ReleaseUSMStats() {
+	t.usmMonitor.ReleaseStats()
+}
+
 // RegisterClient registers a clientID with the tracer
 func (t *Tracer) RegisterClient(clientID string) error {
 	t.state.RegisterClient(clientID)

pkg/network/tracer/tracer_unsupported.go

@@ -34,6 +34,9 @@ func (t *Tracer) GetActiveConnections(_ string) (*network.Connections, error) {
 	return nil, ebpf.ErrNotImplemented
 }
 
+// ReleaseUSMStats is not implemented on this OS for Tracer
+func (t *Tracer) ReleaseUSMStats() {}
+
 // GetNetworkID is not implemented on this OS for Tracer
 func (t *Tracer) GetNetworkID(_ context.Context) (string, error) {
 	return "", ebpf.ErrNotImplemented

pkg/network/tracer/tracer_windows.go

@@ -251,6 +251,10 @@ func (t *Tracer) GetActiveConnections(clientID string) (*network.Connections, er
 	return conns, nil
 }
 
+// ReleaseUSMStats is not implemented on this OS for Tracer.
+func (t *Tracer) ReleaseUSMStats() {
+}
+
 // RegisterClient registers the client
 func (t *Tracer) RegisterClient(clientID string) error {
 	t.state.RegisterClient(clientID)

pkg/network/usm/ebpf_gotls.go

@@ -289,6 +289,10 @@ func (p *goTLSProgram) GetStats() *protocols.ProtocolStats {
 	return nil
 }
 
+// ReleaseStats is a no-op.
+func (p *goTLSProgram) ReleaseStats() {
+}
+
 // Stop terminates goTLS main goroutine.
 func (p *goTLSProgram) Stop(*manager.Manager) {
 	close(p.done)

pkg/network/usm/ebpf_main.go

@@ -544,6 +544,12 @@ func (e *ebpfProgram) getProtocolStats() map[protocols.ProtocolType]interface{}
 	return ret
 }
 
+func (e *ebpfProgram) releaseStats() {
+	for _, protocol := range e.enabledProtocols {
+		protocol.Instance.ReleaseStats()
+	}
+}
+
 // executePerProtocol runs the given callback (`cb`) for every protocol in the given list (`protocolList`).
 // If the callback failed, then we call the error callback (`errorCb`). Eventually returning a list of protocols which
 // successfully executed the callback.

pkg/network/usm/ebpf_ssl.go

@@ -569,6 +569,10 @@ func (o *sslProgram) GetStats() *protocols.ProtocolStats {
 	return nil
 }
 
+// ReleaseStats is a no-op.
+func (o *sslProgram) ReleaseStats() {
+}
+
 const (
 	// Defined in https://man7.org/linux/man-pages/man5/proc.5.html.
 	taskCommLen = 16

pkg/network/usm/monitor.go

@@ -196,6 +196,11 @@ func (m *Monitor) GetProtocolStats() map[protocols.ProtocolType]interface{} {
 	return m.ebpfProgram.getProtocolStats()
 }
 
+// ReleaseStats release stats related objects.
+func (m *Monitor) ReleaseStats() {
+	m.ebpfProgram.releaseStats()
+}
+
 // Stop HTTP monitoring
 func (m *Monitor) Stop() {
 	if m == nil {

pkg/network/usm/monitor_testutil.go

@@ -74,6 +74,7 @@ func (p *protocolMock) Stop(mgr *manager.Manager) {
 
 func (p *protocolMock) DumpMaps(io.Writer, string, *ebpf.Map) {}
 func (p *protocolMock) GetStats() *protocols.ProtocolStats    { return nil }
+func (p *protocolMock) ReleaseStats()                         {}
 
 // IsBuildModeSupported returns always true, as the mock is supported by all modes.
 func (*protocolMock) IsBuildModeSupported(buildmode.Type) bool { return true }

pkg/network/usm/tests/tracer_classification_test.go

@@ -367,6 +367,7 @@ func waitForConnectionsWithProtocol(t *testing.T, tr *tracer.Tracer, targetAddr,
 		if failed {
 			t.Log(conns)
 		}
+		tr.ReleaseUSMStats()
 		return !failed
 	}, 5*time.Second, 100*time.Millisecond, "could not find incoming or outgoing connections")
 	if failed {