Cherry-pick #20325 to 7.8: system/socket: Allow running multiple instances by grouping kprobes by PID

CHANGELOG.next.asciidoc

@@ -100,6 +100,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - system/package: Fix an error that can occur while trying to persist package metadata. {issue}18536[18536] {pull}18887[18887]
 - system/socket: Fix dataset using 100% CPU and becoming unresponsive in some scenarios. {pull}19033[19033] {pull}19764[19764]
 - system/socket: Fixed tracking of long-running connections. {pull}19033[19033]
+- system/socket: Fix kprobe grouping to allow running more than one instance. {pull}20325[20325]
 
 *Filebeat*
 

x-pack/auditbeat/module/system/socket/socket_linux.go

@@ -9,8 +9,11 @@ package socket
 import (
 	"context"
 	"fmt"
+	"os"
+	"path/filepath"
 	"sort"
 	"strconv"
+	"strings"
 	"sync/atomic"
 	"syscall"
 	"time"
@@ -36,17 +39,18 @@ import (
 )
 
 const (
-	moduleName     = "system"
-	metricsetName  = "socket"
-	fullName       = moduleName + "/" + metricsetName
-	namespace      = "system.audit.socket"
-	detailSelector = metricsetName + "detailed"
-	auditbeatGroup = "auditbeat"
+	moduleName      = "system"
+	metricsetName   = "socket"
+	fullName        = moduleName + "/" + metricsetName
+	namespace       = "system.audit.socket"
+	detailSelector  = metricsetName + "detailed"
+	groupNamePrefix = "auditbeat_"
 	// Magic value to detect clock-sync events generated by the metricset.
 	clockSyncMagic uint64 = 0x42DEADBEEFABCDEF
 )
 
 var (
+	groupName     = fmt.Sprintf("%s%d", groupNamePrefix, os.Getpid())
 	kernelVersion string
 	eventCount    uint64
 )
@@ -290,7 +294,7 @@ func (m *MetricSet) Setup() (err error) {
 		extra = WithFilterPort(22)
 	}
 	m.installer = newProbeInstaller(traceFS,
-		WithGroup(auditbeatGroup),
+		WithGroup(groupName),
 		WithTemplates(m.templateVars),
 		extra)
 	defer func() {
@@ -300,10 +304,18 @@ func (m *MetricSet) Setup() (err error) {
 	}()
 
 	//
-	// remove existing KProbes from Auditbeat
+	// remove dangling KProbes from terminated Auditbeat processes.
+	// Not a fatal error if they can't be removed.
 	//
-	if err = m.installer.UninstallIf(isOwnProbe); err != nil {
-		return errors.Wrap(err, "unable to delete existing KProbes. Is Auditbeat already running?")
+	if err = m.installer.UninstallIf(isDeadAuditbeat); err != nil {
+		m.log.Debugf("Removing existing probes from terminated instances: %+v", err)
+	}
+
+	//
+	// remove existing Auditbeat KProbes that match the current PID.
+	//
+	if err = m.installer.UninstallIf(isThisAuditbeat); err != nil {
+		return errors.Wrapf(err, "unable to delete existing KProbes for group %s", groupName)
 	}
 
 	//
@@ -409,7 +421,7 @@ func (m *MetricSet) Cleanup() {
 		}
 	}
 	if m.installer != nil {
-		if err := m.installer.UninstallIf(isOwnProbe); err != nil {
+		if err := m.installer.UninstallIf(isThisAuditbeat); err != nil {
 			m.log.Warnf("Failed to remove KProbes on exit: %v", err)
 		}
 	}
@@ -468,8 +480,28 @@ func triggerClockSync() {
 	unix.Uname(&buf)
 }
 
-func isOwnProbe(probe tracing.Probe) bool {
-	return probe.Group == auditbeatGroup
+func isRunningAuditbeat(pid int) bool {
+	path := fmt.Sprintf("/proc/%d/exe", pid)
+	exePath, err := os.Readlink(path)
+	if err != nil {
+		// Not a running process
+		return false
+	}
+	exeName := filepath.Base(exePath)
+	return strings.HasPrefix(exeName, "auditbeat")
+}
+
+func isDeadAuditbeat(probe tracing.Probe) bool {
+	if strings.HasPrefix(probe.Group, groupNamePrefix) && probe.Group != groupName {
+		if pid, err := strconv.Atoi(probe.Group[len(groupNamePrefix):]); err == nil && !isRunningAuditbeat(pid) {
+			return true
+		}
+	}
+	return false
+}
+
+func isThisAuditbeat(probe tracing.Probe) bool {
+	return probe.Group == groupName
 }
 
 type mountPoint struct {