fix races

Author: brycekahle

pkg/network/sender/event_consumer_linux.go

@@ -132,18 +132,21 @@ func (d *directSenderConsumer) HandleEvent(ev any) {
 		}
 		p.Cwd = cwd
 	}
-	if p.EventType == model.ForkEventType && p.PPid > 0 {
-		if parent, ok := d.processes[p.PPid]; ok && parent != nil {
-			p.Cmdline = parent.Cmdline
-		}
-	}
 	d.process(p)
+	d.proxyFilter.process(p)
+	d.extractor.process(p)
 }
 
 func (d *directSenderConsumer) process(p *process) {
 	d.mtx.Lock()
 	defer d.mtx.Unlock()
 
+	if p.EventType == model.ForkEventType && p.PPid > 0 {
+		if parent, ok := d.processes[p.PPid]; ok && parent != nil {
+			p.Cmdline = parent.Cmdline
+		}
+	}
+
 	if _, seen := d.processes[p.Pid]; seen {
 		if p.EventType == model.ExitEventType {
 			// mark process as dead so it will be removed after next set of connections are collected
@@ -154,8 +157,6 @@ func (d *directSenderConsumer) process(p *process) {
 	if p.EventType == model.ExecEventType || p.EventType == model.ForkEventType {
 		d.processes[p.Pid] = p
 	}
-	d.proxyFilter.process(p)
-	d.extractor.process(p)
 	eventConsumerTelemetry.processCount.Set(float64(len(d.processes)))
 }
 

pkg/network/sender/tags.go

@@ -99,6 +99,7 @@ func (d *directSender) getContainersForExplicitTagging(currentConnections *netwo
 }
 
 func (d *directSender) addContainerTags(c *model.Connection, containerIDForPID map[int32]string, containersForTagging map[string]types.EntityID, tagsEncoder model.TagEncoder) {
+	c.LocalContainerTagsIndex = -1
 	if c.Laddr.ContainerId == "" {
 		return
 	}

pkg/process/metadata/parser/service.go

@@ -11,6 +11,7 @@ import (
 	"runtime"
 	"slices"
 	"strings"
+	"sync"
 	"unicode"
 
 	"github.com/Masterminds/semver/v3"
@@ -64,6 +65,8 @@ type ServiceExtractor struct {
 	useWindowsServiceName bool
 	serviceByPID          map[int32]*serviceMetadata
 	scmReader             *scmReader
+
+	mtx sync.RWMutex
 }
 
 type serviceMetadata struct {
@@ -94,6 +97,8 @@ func (d *ServiceExtractor) Extract(processes map[int32]*procutil.Process) {
 	if !d.enabled {
 		return
 	}
+	d.mtx.Lock()
+	defer d.mtx.Unlock()
 
 	serviceByPID := make(map[int32]*serviceMetadata)
 
@@ -124,6 +129,8 @@ func (d *ServiceExtractor) ExtractSingle(proc *procutil.Process) {
 	if !d.enabled {
 		return
 	}
+	d.mtx.Lock()
+	defer d.mtx.Unlock()
 
 	if meta, seen := d.serviceByPID[proc.Pid]; seen {
 		// check the service metadata is for the same process
@@ -147,6 +154,8 @@ func (d *ServiceExtractor) Remove(pid int32) {
 	if !d.enabled {
 		return
 	}
+	d.mtx.Lock()
+	defer d.mtx.Unlock()
 
 	delete(d.serviceByPID, pid)
 }
@@ -156,6 +165,8 @@ func (d *ServiceExtractor) GetServiceContext(pid int32) []string {
 	if !d.enabled {
 		return nil
 	}
+	d.mtx.RLock()
+	defer d.mtx.RUnlock()
 
 	if runtime.GOOS == "windows" && d.useWindowsServiceName {
 		tags, err := d.getWindowsServiceTags(pid)