discovery.process and pyroscope.java

CHANGELOG.md

@@ -17,6 +17,12 @@ Main (unreleased)
   - [GO-2023-2412](https://github.com/advisories/GHSA-7ww5-4wqc-m92c)
   - [CVE-2023-49568](https://github.com/advisories/GHSA-mw99-9chc-xw7r)
 
+### Features
+
+- A new `discovery.process` component for discovering Linux OS processes on the current host. (@korniltsev)
+
+- A new `pyroscope.java` component for profiling Java processes using async-profiler. (@korniltsev)
+
 ### Enhancements
 
 - Add an option to the windows static mode installer for expanding environment vars in the yaml config. (@erikbaranowski)

component/all/all.go

@@ -25,6 +25,7 @@ import (
 	_ "github.com/grafana/agent/component/discovery/nomad"                          // Import discovery.nomad
 	_ "github.com/grafana/agent/component/discovery/openstack"                      // Import discovery.openstack
 	_ "github.com/grafana/agent/component/discovery/ovhcloud"                       // Import discovery.ovhcloud
+	_ "github.com/grafana/agent/component/discovery/process"                        // Import discovery.process
 	_ "github.com/grafana/agent/component/discovery/puppetdb"                       // Import discovery.puppetdb
 	_ "github.com/grafana/agent/component/discovery/relabel"                        // Import discovery.relabel
 	_ "github.com/grafana/agent/component/discovery/scaleway"                       // Import discovery.scaleway
@@ -127,6 +128,7 @@ import (
 	_ "github.com/grafana/agent/component/prometheus/remotewrite"                   // Import prometheus.remote_write
 	_ "github.com/grafana/agent/component/prometheus/scrape"                        // Import prometheus.scrape
 	_ "github.com/grafana/agent/component/pyroscope/ebpf"                           // Import pyroscope.ebpf
+	_ "github.com/grafana/agent/component/pyroscope/java"                           // Import pyroscope.java
 	_ "github.com/grafana/agent/component/pyroscope/scrape"                         // Import pyroscope.scrape
 	_ "github.com/grafana/agent/component/pyroscope/write"                          // Import pyroscope.write
 	_ "github.com/grafana/agent/component/remote/http"                              // Import remote.http

component/discovery/process/args.go

@@ -0,0 +1,37 @@
+package process
+
+import (
+	"time"
+
+	"github.com/grafana/agent/component/discovery"
+)
+
+type Arguments struct {
+	Join            []discovery.Target `river:"join,attr,optional"`
+	RefreshInterval time.Duration      `river:"refresh_interval,attr,optional"`
+	DiscoverConfig  DiscoverConfig     `river:"discover_config,block,optional"`
+}
+
+type DiscoverConfig struct {
+	Cwd         bool `river:"cwd,attr,optional"`
+	Exe         bool `river:"exe,attr,optional"`
+	Commandline bool `river:"commandline,attr,optional"`
+	Username    bool `river:"username,attr,optional"`
+	UID         bool `river:"uid,attr,optional"`
+	ContainerID bool `river:"container_id,attr,optional"`
+}
+
+var DefaultConfig = Arguments{
+	Join:            nil,
+	RefreshInterval: 60 * time.Second,
+	DiscoverConfig: DiscoverConfig{
+		Cwd:         true,
+		Exe:         true,
+		Commandline: true,
+		ContainerID: true,
+	},
+}
+
+func (args *Arguments) SetToDefault() {
+	*args = DefaultConfig
+}

component/discovery/process/container.go

@@ -0,0 +1,56 @@
+package process
+
+import (
+	"bufio"
+	"io"
+	"regexp"
+	"strings"
+
+	"github.com/grafana/agent/component/discovery"
+)
+
+var (
+	// cgroupContainerIDRe matches a container ID from a /proc/{pid}}/cgroup
+	cgroupContainerIDRe = regexp.MustCompile(`^.*/(?:.*-)?([0-9a-f]{64})(?:\.|\s*$)`)
+)
+
+func getContainerIDFromCGroup(cgroup io.Reader) string {
+	scanner := bufio.NewScanner(cgroup)
+	for scanner.Scan() {
+		line := scanner.Bytes()
+		matches := cgroupContainerIDRe.FindSubmatch(line)
+		if len(matches) <= 1 {
+			continue
+		}
+		return string(matches[1])
+	}
+	return ""
+}
+
+var knownContainerIDPrefixes = []string{"docker://", "containerd://", "cri-o://"}
+
+// get container id from __meta_kubernetes_pod_container_id label
+func getContainerIDFromK8S(k8sContainerID string) string {
+	for _, p := range knownContainerIDPrefixes {
+		if strings.HasPrefix(k8sContainerID, p) {
+			return strings.TrimPrefix(k8sContainerID, p)
+		}
+	}
+	return ""
+}
+
+func getContainerIDFromTarget(target discovery.Target) string {
+	cid, ok := target[labelProcessContainerID]
+	if ok && cid != "" {
+		return cid
+	}
+	cid, ok = target["__meta_kubernetes_pod_container_id"]
+	if ok && cid != "" {
+		return getContainerIDFromK8S(cid)
+	}
+	cid, ok = target["__meta_docker_container_id"]
+	if ok && cid != "" {
+		return cid
+	}
+	return ""
+}

component/discovery/process/container_test.go

@@ -0,0 +1,64 @@
+package process
+
+import (
+	"bytes"
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestCGroupMatching(t *testing.T) {
+	type testcase = struct {
+		containerID, cgroup, expectedID string
+	}
+	testcases := []testcase{
+		{
+			containerID: "containerd://a534eb629135e43beb13213976e37bb2ab95cba4c0d1d0b4e27c6bc4d8091b83",
+			cgroup: "12:cpuset:/kubepods.slice/kubepods-burstable.slice/" +
+				"kubepods-burstable-pod471203d1_984f_477e_9c35_db96487ffe5e.slice/" +
+				"cri-containerd-a534eb629135e43beb13213976e37bb2ab95cba4c0d1d0b4e27c6bc4d8091b83.scope",
+			expectedID: "a534eb629135e43beb13213976e37bb2ab95cba4c0d1d0b4e27c6bc4d8091b83",
+		},
+		{
+			containerID: "cri-o://0ecc7949cbaf17e883264ea1055f60b184a7cb264fd759c4a692e1155086fe2d",
+			cgroup: "0::/kubepods.slice/kubepods-burstable.slice/kubepods-burstable-podb57320a0_e7eb_4ac8_a791_4c4472796867.slice/" +
+				"crio-0ecc7949cbaf17e883264ea1055f60b184a7cb264fd759c4a692e1155086fe2d.scope",
+			expectedID: "0ecc7949cbaf17e883264ea1055f60b184a7cb264fd759c4a692e1155086fe2d",
+		},
+		{
+
+			containerID: "docker://656959d9ee87a0b131c601ce9d9f8f76b1dda60e8608c503b5979d849cbdc714",
+			cgroup: "0::/../../kubepods-besteffort-pod88f6f4e3_59c0_4ce8_9ecf_391c8b5a60ad.slice/" +
+				"docker-656959d9ee87a0b131c601ce9d9f8f76b1dda60e8608c503b5979d849cbdc714.scope",
+			expectedID: "656959d9ee87a0b131c601ce9d9f8f76b1dda60e8608c503b5979d849cbdc714",
+		},
+		{
+			containerID: "containerd://47e320f795efcec1ecf2001c3a09c95e3701ed87de8256837b70b10e23818251",
+			cgroup: "0::/kubepods.slice/kubepods-burstable.slice/" +
+				"kubepods-burstable-podf9a04ecc_1875_491b_926c_d2f64757704e.slice/" +
+				"cri-containerd-47e320f795efcec1ecf2001c3a09c95e3701ed87de8256837b70b10e23818251.scope",
+			expectedID: "47e320f795efcec1ecf2001c3a09c95e3701ed87de8256837b70b10e23818251",
+		},
+		{
+			containerID: "docker://7edda1de1e0d1d366351e478359cf5fa16bb8ab53063a99bb119e56971bfb7e2",
+			cgroup: "11:devices:/kubepods/besteffort/pod85adbef3-622f-4ef2-8f60-a8bdf3eb6c72/" +
+				"7edda1de1e0d1d366351e478359cf5fa16bb8ab53063a99bb119e56971bfb7e2",
+			expectedID: "7edda1de1e0d1d366351e478359cf5fa16bb8ab53063a99bb119e56971bfb7e2",
+		},
+		{
+			containerID: "",
+			cgroup:      "0::/../../user.slice/user-501.slice/session-3.scope",
+			expectedID:  "",
+		},
+	}
+	for i, tc := range testcases {
+		t.Run(fmt.Sprintf("testcase %d %s", i, tc.cgroup), func(t *testing.T) {
+			cid := getContainerIDFromCGroup(bytes.NewReader([]byte(tc.cgroup)))
+			expected := tc.expectedID
+			require.Equal(t, expected, cid)
+			cid = getContainerIDFromK8S(tc.containerID)
+			require.Equal(t, expected, cid)
+		})
+	}
+}

component/discovery/process/discover.go

@@ -0,0 +1,169 @@
+//go:build linux
+
+package process
+
+import (
+	"errors"
+	"fmt"
+	"os"
+	"path"
+	"runtime"
+
+	"github.com/go-kit/log"
+	"github.com/go-kit/log/level"
+	"github.com/grafana/agent/component/discovery"
+	gopsutil "github.com/shirou/gopsutil/v3/process"
+	"golang.org/x/sys/unix"
+)
+
+const (
+	labelProcessID          = "__process_pid__"
+	labelProcessExe         = "__meta_process_exe"
+	labelProcessCwd         = "__meta_process_cwd"
+	labelProcessCommandline = "__meta_process_commandline"
+	labelProcessUsername    = "__meta_process_username"
+	labelProcessUID         = "__meta_process_uid"
+	labelProcessContainerID = "__container_id__"
+)
+
+type process struct {
+	pid         string
+	exe         string
+	cwd         string
+	commandline string
+	containerID string
+	username    string
+	uid         string
+}
+
+func (p process) String() string {
+	return fmt.Sprintf("pid=%s exe=%s cwd=%s commandline=%s containerID=%s", p.pid, p.exe, p.cwd, p.commandline, p.containerID)
+}
+
+func convertProcesses(ps []process) []discovery.Target {
+	var res []discovery.Target
+	for _, p := range ps {
+		t := convertProcess(p)
+		res = append(res, t)
+	}
+	return res
+}
+
+func convertProcess(p process) discovery.Target {
+	t := make(discovery.Target, 5)
+	t[labelProcessID] = p.pid
+	if p.exe != "" {
+		t[labelProcessExe] = p.exe
+	}
+	if p.cwd != "" {
+		t[labelProcessCwd] = p.cwd
+	}
+	if p.commandline != "" {
+		t[labelProcessCommandline] = p.commandline
+	}
+	if p.containerID != "" {
+		t[labelProcessContainerID] = p.containerID
+	}
+	if p.username != "" {
+		t[labelProcessUsername] = p.username
+	}
+	if p.uid != "" {
+		t[labelProcessUID] = p.uid
+	}
+	return t
+}
+
+func discover(l log.Logger, cfg *DiscoverConfig) ([]process, error) {
+	processes, err := gopsutil.Processes()
+	if err != nil {
+		return nil, fmt.Errorf("failed to list processes: %w", err)
+	}
+	res := make([]process, 0, len(processes))
+	loge := func(pid int, e error) {
+		if errors.Is(e, unix.ESRCH) {
+			return
+		}
+		if errors.Is(e, os.ErrNotExist) {
+			return
+		}
+		_ = level.Error(l).Log("msg", "failed to get process info", "err", e, "pid", pid)
+	}
+	for _, p := range processes {
+		spid := fmt.Sprintf("%d", p.Pid)
+		var (
+			exe, cwd, commandline, containerID, username, uid string
+		)
+		if cfg.Exe {
+			exe, err = p.Exe()
+			if err != nil {
+				loge(int(p.Pid), err)
+				continue
+			}
+		}
+		if cfg.Cwd {
+			cwd, err = p.Cwd()
+			if err != nil {
+				loge(int(p.Pid), err)
+				continue
+			}
+		}
+		if cfg.Commandline {
+			commandline, err = p.Cmdline()
+			if err != nil {
+				loge(int(p.Pid), err)
+				continue
+			}
+		}
+		if cfg.Username {
+			username, err = p.Username()
+			if err != nil {
+				loge(int(p.Pid), err)
+				continue
+			}
+		}
+		if cfg.UID {
+			uids, err := p.Uids()
+			if err != nil {
+				loge(int(p.Pid), err)
+				continue
+			}
+			if len(uids) > 0 {
+				uid = fmt.Sprintf("%d", uids[0])
+			}
+		}
+
+		if cfg.ContainerID {
+			containerID, err = getLinuxProcessContainerID(l, spid)
+			if err != nil {
+				loge(int(p.Pid), err)
+				continue
+			}
+		}
+		res = append(res, process{
+			pid:         spid,
+			exe:         exe,
+			cwd:         cwd,
+			commandline: commandline,
+			containerID: containerID,
+			username:    username,
+			uid:         uid,
+		})
+	}
+
+	return res, nil
+}
+
+func getLinuxProcessContainerID(l log.Logger, pid string) (string, error) {
+	if runtime.GOOS == "linux" {
+		cgroup, err := os.Open(path.Join("/proc", pid, "cgroup"))
+		if err != nil {
+			return "", err
+		}
+		defer cgroup.Close()
+		cid := getContainerIDFromCGroup(cgroup)
+		if cid != "" {
+			return cid, nil
+		}
+	}
+	return "", nil
+}