Merge pull request #140 from parca-dev/rem-observer

Author: gnurizen

interpreter/instancestubs.go

@@ -12,34 +12,28 @@ import (
 	"go.opentelemetry.io/ebpf-profiler/tpbase"
 )
 
-// ObserverStubs provides empty implementations of Observer hooks that are
-// not mandatory for an Observer implementation.
-type ObserverStubs struct {
+// InstanceStubs provides empty implementations of Instance hooks that are
+// not mandatory for a Instance implementation.
+type InstanceStubs struct {
 }
 
-func (os *ObserverStubs) Detach(EbpfHandler, libpf.PID) error {
+func (is *InstanceStubs) Detach(EbpfHandler, libpf.PID) error {
 	return nil
 }
 
-func (os *ObserverStubs) SynchronizeMappings(EbpfHandler, reporter.SymbolReporter, process.Process,
+func (is *InstanceStubs) SynchronizeMappings(EbpfHandler, reporter.SymbolReporter, process.Process,
 	[]process.Mapping) error {
 	return nil
 }
 
-func (os *ObserverStubs) UpdateTSDInfo(EbpfHandler, libpf.PID, tpbase.TSDInfo) error {
+func (is *InstanceStubs) UpdateTSDInfo(EbpfHandler, libpf.PID, tpbase.TSDInfo) error {
 	return nil
 }
 
-func (os *ObserverStubs) GetAndResetMetrics() ([]metrics.Metric, error) {
+func (is *InstanceStubs) GetAndResetMetrics() ([]metrics.Metric, error) {
 	return []metrics.Metric{}, nil
 }
 
-// InstanceStubs provides empty implementations of Instance hooks that are
-// not mandatory for a Instance implementation.
-type InstanceStubs struct {
-	ObserverStubs
-}
-
 func (is *InstanceStubs) Symbolize(*host.Frame, *libpf.Frames) error {
 	return ErrMismatchInterpreterType
 }

interpreter/types.go

@@ -136,16 +136,14 @@ type Data interface {
 	Unload(ebpf EbpfHandler)
 }
 
-// Observer is the base interface for observing per-PID data without symbolization.
-// This interface is useful for components that need to observe processes without
-// providing frame symbolization capabilities.
-type Observer interface {
+// Instance is the interface to operate on per-PID data.
+type Instance interface {
 	// Detach removes any information from the ebpf maps. The pid is given as argument so
-	// simple observers can use the global Data also as the Observer implementation.
+	// simple interpreters can use the global Data also as the Instance implementation.
 	Detach(ebpf EbpfHandler, pid libpf.PID) error
 
 	// SynchronizeMappings is called when the processmanager has reread process memory
-	// mappings. Observers not needing to process these events can simply ignore them
+	// mappings. Interpreters not needing to process these events can simply ignore them
 	// by just returning a nil.
 	SynchronizeMappings(ebpf EbpfHandler, symbolReporter reporter.SymbolReporter,
 		pr process.Process, mappings []process.Mapping) error
@@ -154,16 +152,6 @@ type Observer interface {
 	// introspection data has been updated.
 	UpdateTSDInfo(ebpf EbpfHandler, pid libpf.PID, info tpbase.TSDInfo) error
 
-	// GetAndResetMetrics collects the metrics from the Observer and resets
-	// the counters to their initial value.
-	GetAndResetMetrics() ([]metrics.Metric, error)
-}
-
-// Instance is the interface to operate on per-PID data with symbolization support.
-// It extends Observer with the ability to symbolize frames.
-type Instance interface {
-	Observer
-
 	// Symbolize converts one ebpf frame to one or more (if inlining was expanded) libpf.Frame.
 	// The resulting libpf.Frame values are appended to frames.
 	Symbolize(ebpfFrame *host.Frame, frames *libpf.Frames) error

processmanager/execinfomanager/manager.go

@@ -66,9 +66,6 @@ type ExecutableInfo struct {
 	// instance belongs to was previously identified as an interpreter. Otherwise,
 	// this field is nil.
 	Data interpreter.Data
-	// Observers stores per-executable observer information. Multiple observers
-	// can be associated with a single executable.
-	Observers []interpreter.Data
 	// TSDInfo stores TSD information if the executable is libc, otherwise nil.
 	TSDInfo *tpbase.TSDInfo
 }
@@ -146,12 +143,7 @@ func NewExecutableInfoManager(
 	if collectCustomLabels {
 		interpreterLoaders = append(interpreterLoaders, customlabels.Loader)
 	}
-
-	// Initialize observer loaders
-	observerLoaders := []interpreter.Loader{
-		oomwatcher.Loader,
-		// Additional observers can be added here
-	}
+	interpreterLoaders = append(interpreterLoaders, oomwatcher.Loader)
 
 	deferredFileIDs, err := lru.NewSynced[host.FileID, libpf.Void](deferredFileIDSize,
 		func(id host.FileID) uint32 { return uint32(id) })
@@ -164,7 +156,6 @@ func NewExecutableInfoManager(
 		sdp: sdp,
 		state: xsync.NewRWMutex(executableInfoManagerState{
 			interpreterLoaders: interpreterLoaders,
-			observerLoaders:    observerLoaders,
 			executables:        map[host.FileID]*entry{},
 			unwindInfoIndex:    map[sdtypes.UnwindInfo]uint16{},
 			ebpf:               ebpf,
@@ -247,9 +238,8 @@ func (mgr *ExecutableInfoManager) AddOrIncRef(fileID host.FileID,
 	// Insert a corresponding record into our map.
 	info = &entry{
 		ExecutableInfo: ExecutableInfo{
-			Data:      state.detectAndLoadInterpData(loaderInfo),
-			Observers: state.detectAndLoadObservers(loaderInfo),
-			TSDInfo:   tsdInfo,
+			Data:    state.detectAndLoadInterpData(loaderInfo),
+			TSDInfo: tsdInfo,
 		},
 		mapRef: ref,
 		rc:     1,
@@ -321,11 +311,6 @@ type executableInfoManagerState struct {
 	// for loading the host agent support for a specific interpreter type.
 	interpreterLoaders []interpreter.Loader
 
-	// observerLoaders is a list of instances of an interface that provide functionality
-	// for loading observers for executables. Unlike interpreters, multiple observers
-	// can be associated with a single executable.
-	observerLoaders []interpreter.Loader
-
 	// ebpf provides the interface to manipulate eBPF maps.
 	ebpf pmebpf.EbpfHandler
 
@@ -390,38 +375,6 @@ func (state *executableInfoManagerState) detectAndLoadInterpData(
 	}
 }
 
-// detectAndLoadObservers attempts to detect observers for the given executable.
-// Unlike interpreters, multiple observers can be loaded for a single executable.
-func (state *executableInfoManagerState) detectAndLoadObservers(
-	loaderInfo *interpreter.LoaderInfo) []interpreter.Data {
-	var observers []interpreter.Data
-
-	// Ask all observer loaders to check this executable
-	for _, loader := range state.observerLoaders {
-		data, err := loader(state.ebpf, loaderInfo)
-		if err != nil {
-			if errors.Is(err, os.ErrNotExist) {
-				// Very common if the process exited when we tried to analyze it.
-				log.Debugf("Failed to load observer for %v (%#016x): file not found",
-					loaderInfo.FileName(), loaderInfo.FileID())
-			} else {
-				log.Errorf("Loader for %v (%#016x) failed: %v",
-					loaderInfo.FileName(), loaderInfo.FileID(), err)
-			}
-			continue
-		}
-
-		// All observers return a data instance (possibly no-op)
-		if data != nil {
-			log.Debugf("Observer data %v for %v (%#016x)",
-				data, loaderInfo.FileName(), loaderInfo.FileID())
-			observers = append(observers, data)
-		}
-	}
-
-	return observers
-}
-
 // loadDeltas converts the sdtypes.StackDelta to StackDeltaEBPF and passes that to
 // the ebpf interface to be loaded to kernel maps. While converting the deltas, it
 // also creates a list of all large gaps in the executable.

processmanager/manager.go

@@ -89,13 +89,11 @@ func New(ctx context.Context, includeTracers types.IncludedTracers, monitorInter
 	}
 
 	interpreters := make(map[libpf.PID]map[util.OnDiskFileIdentifier]interpreter.Instance)
-	observers := make(map[libpf.PID]map[util.OnDiskFileIdentifier][]interpreter.Observer)
 
 	pm := &ProcessManager{
 		interpreterTracerEnabled: em.NumInterpreterLoaders() > 0,
 		eim:                      em,
 		interpreters:             interpreters,
-		observers:                observers,
 		exitEvents:               make(map[libpf.PID]times.KTime),
 		pidToProcessInfo:         make(map[libpf.PID]*processInfo),
 		ebpf:                     ebpf,

processmanager/processinfo.go

@@ -283,60 +283,6 @@ func (pm *ProcessManager) handleNewInterpreter(pr process.Process, m *Mapping,
 	return nil
 }
 
-// handleNewObservers processes observers for a new executable mapping.
-// Unlike interpreters where only one can handle an executable, multiple observers
-// can be attached to the same executable.
-//
-// The caller is responsible to hold the ProcessManager lock to avoid race conditions.
-func (pm *ProcessManager) handleNewObservers(pr process.Process, m *Mapping,
-	ei *eim.ExecutableInfo) {
-	if len(ei.Observers) == 0 {
-		return
-	}
-
-	pid := pr.PID()
-	key := m.GetOnDiskFileIdentifier()
-
-	// Initialize observer map for this PID if needed
-	if _, ok := pm.observers[pid]; !ok {
-		pm.observers[pid] = make(map[util.OnDiskFileIdentifier][]interpreter.Observer)
-	}
-
-	// Check if observers are already attached for this key
-	if _, ok := pm.observers[pid][key]; ok {
-		// Already handled
-		return
-	}
-
-	// Attach all observers
-	var observers []interpreter.Observer //nolint:prealloc
-	for _, observerData := range ei.Observers {
-		observer, err := observerData.Attach(pm.ebpf, pid, libpf.Address(m.Bias),
-			pr.GetRemoteMemory())
-		if err != nil {
-			log.Errorf("Failed to attach observer %v to PID %v: %v", observerData, pid, err)
-			continue
-		}
-		if observer == nil {
-			continue
-		}
-
-		log.Debugf("Attached observer %v to PID %v", observerData, pid)
-		observers = append(observers, observer)
-
-		// Update TSD info if available
-		if tsdInfo := pm.getTSDInfo(pid); tsdInfo != nil {
-			if err := observer.UpdateTSDInfo(pm.ebpf, pid, *tsdInfo); err != nil {
-				log.Errorf("Failed to update observer TSDInfo for PID %v: %v", pid, err)
-			}
-		}
-	}
-
-	if len(observers) > 0 {
-		pm.observers[pid][key] = observers
-	}
-}
-
 // handleNewMapping processes new file backed mappings
 func (pm *ProcessManager) handleNewMapping(pr process.Process, m *Mapping,
 	elfRef *pfelf.Reference) error {
@@ -369,9 +315,6 @@ func (pm *ProcessManager) handleNewMapping(pr process.Process, m *Mapping,
 		}
 	}
 
-	// Handle observers if present
-	pm.handleNewObservers(pr, m, &ei)
-
 	return nil
 }
 
@@ -533,26 +476,6 @@ func (pm *ProcessManager) processRemovedMappings(pid libpf.PID, mappings []libpf
 		// remove the entry.
 		delete(pm.interpreters, pid)
 	}
-
-	// Clean up observers that are no longer valid
-	for key, observerList := range pm.observers[pid] {
-		if _, ok := interpretersValid[key]; ok {
-			continue
-		}
-		for _, observer := range observerList {
-			if err := observer.Detach(pm.ebpf, pid); err != nil {
-				log.Errorf("Failed to unload observer for PID %d: %v",
-					pid, err)
-			}
-		}
-		delete(pm.observers[pid], key)
-	}
-
-	if len(pm.observers[pid]) == 0 {
-		// There are no longer any mapped observers in the process, therefore we can
-		// remove the entry.
-		delete(pm.observers, pid)
-	}
 }
 
 // synchronizeMappings synchronizes executable mappings for the given PID.
@@ -857,18 +780,6 @@ func (pm *ProcessManager) ProcessedUntil(traceCaptureKTime times.KTime) {
 		}
 		delete(pm.interpreters, pid)
 
-		// Detach all observers for this PID
-		for _, observerList := range pm.observers[pid] {
-			for _, observer := range observerList {
-				if err2 := observer.Detach(pm.ebpf, pid); err2 != nil {
-					err = errors.Join(err,
-						fmt.Errorf("failed to handle observer process exit for PID %d: %v",
-							pid, err2))
-				}
-			}
-		}
-		delete(pm.observers, pid)
-
 		delete(pm.exitEvents, pid)
 		log.Debugf("PID %v exit latency %v ms", pid, (nowKTime-pidExitKTime)/1e6)
 	}

processmanager/types.go

@@ -50,13 +50,6 @@ type ProcessManager struct {
 	// the unique on-disk identifier of the interpreter DSO.
 	interpreters map[libpf.PID]map[util.OnDiskFileIdentifier]interpreter.Instance
 
-	// observers records the interpreter.Observer interface which contains hooks for
-	// process monitoring without symbolization. Unlike interpreters, multiple observers
-	// can be associated with each executable.
-	// The key of the first map is a process ID, while the key of the second map is
-	// the unique on-disk identifier of the observed DSO.
-	observers map[libpf.PID]map[util.OnDiskFileIdentifier][]interpreter.Observer
-
 	// pidToProcessInfo keeps track of the executable memory mappings.
 	pidToProcessInfo map[libpf.PID]*processInfo