(DynamicSelector) fixups for dynamic resource attributes (#2498)

Author: damemi

pkg/instrumenter/opts.go

@@ -21,7 +21,7 @@ type Option func(info *global.ContextInfo)
 // Root AddPIDs/RemovePIDs preserve the legacy behavior and apply to all supported signals.
 // Service name and resource attributes set via AddPID or SetPID are shared across all signals.
 // SetPID updates live FileInfo for instrumented PIDs; traces and app metrics read it at export time.
-// Network and stats metrics decorate flow records by pod IP when the PID is in that signal view.
+// Network and stats metrics decorate flow records with service name/namespace by pod IP.
 func WithDynamicPIDSelector(sel *discover.DynamicPIDSelector) Option {
 	return func(info *global.ContextInfo) {
 		info.DynamicPIDSelector = sel

pkg/internal/appolly/appolly.go

@@ -108,7 +108,7 @@ func New(ctx context.Context, ctxInfo *global.ContextInfo, config *obi.Config) (
 	// When sel is nil, finder gets nil: config target_pids are used as static criteria (FindingCriteria(cfg, false)).
 	if sel != nil {
 		sel.SetOnFileInfoUpdated(func(fi *exec.FileInfo) {
-			processEventsInput.Send(exec.ProcessEvent{Type: exec.ProcessEventCreated, File: fi})
+			processEventsInput.SendCtx(ctx, exec.ProcessEvent{Type: exec.ProcessEventCreated, File: fi})
 		})
 	}
 	instr := &Instrumenter{

pkg/selection/dynamic_app_ips.go

@@ -124,6 +124,8 @@ func (d *DynamicAppIPs) decrementIPsLocked(ips []string) {
 }
 
 // ResolveContainerIPs returns pod IPs for a PID when a Kubernetes store is available.
+// It registers the PID in store via AddProcess; callers that invoke it outside
+// DynamicAppIPs must call store.DeleteProcess when the PID is no longer needed.
 func ResolveContainerIPs(store *kube.Store, pid app.PID) []string {
 	if store == nil {
 		return nil

pkg/selection/dynamic_flow_attrs.go

@@ -16,27 +16,28 @@ import (
 type flowIPDecoration struct {
 	serviceName      string
 	serviceNamespace string
-	resourceAttrs    map[attr.Name]string
 }
 
-// DynamicFlowAttrs maps dynamically selected application IPs to service identity and resource
-// attributes for NetO11y and StatsO11y decoration.
+// DynamicFlowAttrs maps dynamically selected application IPs to service identity for NetO11y
+// and StatsO11y decoration.
 type DynamicFlowAttrs struct {
 	multiSel  MultiSignalPIDSelector
 	signalSel PIDSelector
 	store     *kube.Store
 
-	mu      sync.RWMutex
-	ipDecor map[string]flowIPDecoration
+	mu             sync.RWMutex
+	ipDecor        map[string]flowIPDecoration
+	registeredPIDs map[app.PID]struct{}
 }
 
 // NewDynamicFlowAttrs creates a tracker for the given signal view and optional Kubernetes store.
 func NewDynamicFlowAttrs(multiSel MultiSignalPIDSelector, signalSel PIDSelector, store *kube.Store) *DynamicFlowAttrs {
 	return &DynamicFlowAttrs{
-		multiSel:  multiSel,
-		signalSel: signalSel,
-		store:     store,
-		ipDecor:   map[string]flowIPDecoration{},
+		multiSel:       multiSel,
+		signalSel:      signalSel,
+		store:          store,
+		ipDecor:        map[string]flowIPDecoration{},
+		registeredPIDs: map[app.PID]struct{}{},
 	}
 }
 
@@ -83,31 +84,40 @@ func (d *DynamicFlowAttrs) loopAttrs(ctx context.Context) {
 
 func (d *DynamicFlowAttrs) rebuild() {
 	pids, ok := d.signalSel.GetPIDs()
-	if !ok {
-		d.mu.Lock()
-		d.ipDecor = map[string]flowIPDecoration{}
-		d.mu.Unlock()
-		return
-	}
 
 	next := map[string]flowIPDecoration{}
-	for _, pid := range pids {
-		entry, ok := d.multiSel.GetPID(uint32(pid))
-		if !ok {
-			continue
-		}
-		dec := decorationFromEntry(entry)
-		if dec.isEmpty() {
-			continue
-		}
-		for _, ip := range ResolveContainerIPs(d.store, pid) {
-			next[ip] = dec
+	storeRegisteredPIDs := map[app.PID]struct{}{}
+	if ok && d.store != nil {
+		for _, pid := range pids {
+			entry, found := d.multiSel.GetPID(uint32(pid))
+			if !found {
+				continue
+			}
+			dec := decorationFromEntry(entry)
+			if dec.isEmpty() {
+				continue
+			}
+			for _, ip := range ResolveContainerIPs(d.store, pid) {
+				next[ip] = dec
+			}
+			storeRegisteredPIDs[pid] = struct{}{}
 		}
 	}
 
 	d.mu.Lock()
+	defer d.mu.Unlock()
+	for pid := range d.registeredPIDs {
+		if _, still := storeRegisteredPIDs[pid]; !still {
+			if d.store != nil {
+				d.store.DeleteProcess(pid)
+			}
+			delete(d.registeredPIDs, pid)
+		}
+	}
+	for pid := range storeRegisteredPIDs {
+		d.registeredPIDs[pid] = struct{}{}
+	}
 	d.ipDecor = next
-	d.mu.Unlock()
 }
 
 func (d *DynamicFlowAttrs) Apply(a *pipe.CommonAttrs) {
@@ -133,21 +143,14 @@ func (d *DynamicFlowAttrs) Apply(a *pipe.CommonAttrs) {
 }
 
 func decorationFromEntry(entry DynamicPIDEntry) flowIPDecoration {
-	dec := flowIPDecoration{
+	return flowIPDecoration{
 		serviceName:      entry.ServiceName,
 		serviceNamespace: entry.ServiceNamespace,
 	}
-	if len(entry.ResourceAttributes) > 0 {
-		dec.resourceAttrs = make(map[attr.Name]string, len(entry.ResourceAttributes))
-		for k, v := range entry.ResourceAttributes {
-			dec.resourceAttrs[attr.Name(k)] = v
-		}
-	}
-	return dec
 }
 
 func (d flowIPDecoration) isEmpty() bool {
-	return d.serviceName == "" && d.serviceNamespace == "" && len(d.resourceAttrs) == 0
+	return d.serviceName == "" && d.serviceNamespace == ""
 }
 
 func applyFlowDecoration(dec flowIPDecoration, a *pipe.CommonAttrs, src bool) {
@@ -158,15 +161,12 @@ func applyFlowDecoration(dec flowIPDecoration, a *pipe.CommonAttrs, src bool) {
 		if dec.serviceNamespace != "" {
 			a.Metadata[attr.ServiceNamespace] = dec.serviceNamespace
 		}
-	} else {
-		if dec.serviceName != "" {
-			a.Metadata[attr.ServicePeerName] = dec.serviceName
-		}
-		if dec.serviceNamespace != "" {
-			a.Metadata[attr.ServicePeerNamespace] = dec.serviceNamespace
-		}
+		return
+	}
+	if dec.serviceName != "" {
+		a.Metadata[attr.ServicePeerName] = dec.serviceName
 	}
-	for k, v := range dec.resourceAttrs {
-		a.Metadata[k] = v
+	if dec.serviceNamespace != "" {
+		a.Metadata[attr.ServicePeerNamespace] = dec.serviceNamespace
 	}
 }

pkg/selection/dynamic_flow_attrs_test.go

@@ -62,9 +62,6 @@ func TestDynamicFlowAttrs_Apply_srcAndDst(t *testing.T) {
 				PID:              42,
 				ServiceName:      "payments",
 				ServiceNamespace: "prod",
-				ResourceAttributes: map[string]string{
-					"team": "checkout",
-				},
 			},
 		},
 	}
@@ -81,7 +78,6 @@ func TestDynamicFlowAttrs_Apply_srcAndDst(t *testing.T) {
 	require.NotNil(t, srcFlow.Metadata)
 	assert.Equal(t, "payments", srcFlow.Metadata[attr.ServiceName])
 	assert.Equal(t, "prod", srcFlow.Metadata[attr.ServiceNamespace])
-	assert.Equal(t, "checkout", srcFlow.Metadata[attr.Name("team")])
 
 	dstFlow := &pipe.CommonAttrs{
 		SrcAddr: pipe.IPAddr(net.ParseIP("10.0.0.9")),
@@ -94,15 +90,43 @@ func TestDynamicFlowAttrs_Apply_srcAndDst(t *testing.T) {
 }
 
 func TestDynamicFlowAttrs_rebuild_clearsWhenEmpty(t *testing.T) {
-	sel := &stubMultiPIDSelector{stubPIDSelector: stubPIDSelector{}}
+	sel := &stubMultiPIDSelector{
+		stubPIDSelector: stubPIDSelector{pids: []app.PID{1}},
+		entries: map[app.PID]DynamicPIDEntry{
+			1: {PID: 1, ServiceName: "a"},
+		},
+	}
 	tracker := NewDynamicFlowAttrs(sel, sel, nil)
 	tracker.mu.Lock()
 	tracker.ipDecor["10.0.0.1"] = flowIPDecoration{serviceName: "old"}
+	tracker.registeredPIDs[1] = struct{}{}
 	tracker.mu.Unlock()
 
+	sel.pids = nil
 	tracker.rebuild()
 
 	tracker.mu.RLock()
 	assert.Empty(t, tracker.ipDecor)
+	assert.Empty(t, tracker.registeredPIDs)
+	tracker.mu.RUnlock()
+}
+
+func TestDynamicFlowAttrs_rebuild_doesNotTrackPIDWithoutDecoration(t *testing.T) {
+	sel := &stubMultiPIDSelector{
+		stubPIDSelector: stubPIDSelector{pids: []app.PID{1, 2}},
+		entries: map[app.PID]DynamicPIDEntry{
+			1: {PID: 1, ServiceName: "a"},
+			2: {PID: 2},
+		},
+	}
+	tracker := NewDynamicFlowAttrs(sel, sel, nil)
+	tracker.mu.Lock()
+	tracker.registeredPIDs[2] = struct{}{}
+	tracker.mu.Unlock()
+
+	tracker.rebuild()
+
+	tracker.mu.RLock()
+	assert.Empty(t, tracker.registeredPIDs)
 	tracker.mu.RUnlock()
 }